diff options
| -rw-r--r-- | src/document/sync.rs | 42 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 21 |
2 files changed, 38 insertions, 25 deletions
diff --git a/src/document/sync.rs b/src/document/sync.rs index a46e059..4ae41ba 100644 --- a/src/document/sync.rs +++ b/src/document/sync.rs @@ -59,19 +59,27 @@ impl Received { /// unknown, they give way to the heads we share (`known_last_sync`), as for an /// overtaken message. /// - /// No heads stand for the tips of the changes (`SyncSession::push`). Changes that - /// don't decode leave the heads out; applying them fails as it would have anyway. - pub(crate) fn pushed(heads: Vec<crate::document::ChangeHash>, changes: Vec<Vec<u8>>) -> Self { + /// No heads stand for the tips of the changes (`SyncSession::push`). None without + /// changes, or without heads where a change doesn't decode: no heads would say the + /// sender lost all it had, and we'd send it everything again. + pub(crate) fn pushed( + heads: Vec<crate::document::ChangeHash>, + changes: Vec<Vec<u8>>, + ) -> Option<Self> { + if changes.is_empty() { + return None; + } let heads = if heads.is_empty() { - let decoded: Vec<automerge::Change> = changes + let decoded = changes .iter() - .filter_map(|bytes| automerge::Change::try_from(bytes.as_slice()).ok()) - .collect(); + .map(|bytes| automerge::Change::try_from(bytes.as_slice())) + .collect::<Result<Vec<_>, _>>() + .ok()?; tips(decoded.iter().map(|change| (change.hash(), change.deps()))) } else { heads }; - Self(sync::Message { + Some(Self(sync::Message { heads: heads.clone(), need: Vec::new(), have: vec![sync::Have { @@ -81,7 +89,7 @@ impl Received { changes: changes.into(), flags: None, version: sync::MessageVersion::V2, - }) + })) } /// The sender's heads, as it says. @@ -1499,7 +1507,7 @@ mod tests { assert!(a.push(&ours, 1 << 20).is_none(), "sent already"); let before = theirs.get_heads(); - let message = Received::pushed(heads, changes); + let message = Received::pushed(heads, changes).expect("decodes"); assert_eq!(message.heads(), ours.get_heads(), "found in the changes"); b.receive(&mut theirs, message).unwrap(); b.came_from_peer(&theirs, &before); @@ -1530,8 +1538,11 @@ mod tests { a.send_succeeded(a.id()); let before = theirs.get_heads(); - b.receive(&mut theirs, Received::pushed(heads, changes)) - .unwrap(); + b.receive( + &mut theirs, + Received::pushed(heads, changes).expect("decodes"), + ) + .unwrap(); b.came_from_peer(&theirs, &before); assert!(!b.answered_as_delivered(&theirs, &theirs.get_heads())); let answer = b.generate(&theirs, now, bytes).expect("an answer"); @@ -1542,6 +1553,15 @@ mod tests { assert_eq!(ours.get_heads(), theirs.get_heads()); } + /// A push without changes, or without heads and changes that don't decode, is no + /// message: no heads would say the sender lost all it had. + #[test] + fn broken_pushes_are_refused() { + let hash = crate::document::ChangeHash([1; 32]); + assert!(Received::pushed(vec![hash], Vec::new()).is_none()); + assert!(Received::pushed(Vec::new(), vec![vec![1, 2, 3]]).is_none()); + } + /// Tips are the changes none of the others depend on, sorted as heads are. #[test] fn tips_are_what_nothing_depends_on() { diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index 0f607e7..aea6a0b 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -471,8 +471,7 @@ pub(crate) async fn handle_sync( let writer = envelope.writer; let delivered = envelope.delivered; // Changes pushed along a tree (`SyncBody::Changes`): their heads and bytes, made - // into the Automerge message that would have brought them once the document's lock - // is held (`Received::pushed`). + // into the Automerge message that would have brought them (`Received::pushed`). let (payload, deflated, pushed) = match envelope.body { transport::SyncBody::Message(payload) => (payload, false, None), transport::SyncBody::Deflated(payload) => (payload, true, None), @@ -537,12 +536,13 @@ pub(crate) async fn handle_sync( } else { message }; - let (payload_len, decoded) = match &pushed { - Some((_, _, len)) => (*len, None), - None => ( - message.0.len(), - Some(crate::document::sync::decode(&message)?), + let (payload_len, message) = match pushed { + Some((heads, changes, len)) => ( + len, + crate::document::sync::Received::pushed(heads, changes) + .context("Pushed changes that don't decode")?, ), + None => (message.0.len(), crate::document::sync::decode(&message)?), }; stats.sync_received(payload_len); let mut replica = doc.blocking_write(); @@ -552,13 +552,6 @@ pub(crate) async fn handle_sync( if replica.epoch != epoch { return Ok(None); } - let message = match decoded { - Some(message) => message, - None => { - let (heads, changes, _) = pushed.expect("pushed or decoded"); - crate::document::sync::Received::pushed(heads, changes) - } - }; let carried = message.carries_changes(); let sizes = if carried { message.sizes() } else { (0, 0) }; if carried { |