diff options
| -rw-r--r-- | src/document/sync.rs | 48 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 4 |
2 files changed, 52 insertions, 0 deletions
diff --git a/src/document/sync.rs b/src/document/sync.rs index f1582a7..f45a316 100644 --- a/src/document/sync.rs +++ b/src/document/sync.rs @@ -281,6 +281,24 @@ impl SyncSession { self.reply = None; } + /// The changes since `before` in `doc` came from the peer (we just applied its + /// message): it has them, so they never go back to it. Automerge goes by what it + /// last heard the peer has, and a message that overtook another makes that older + /// than what the peer sent since: the answer would carry the peer's own changes + /// back. Counted as sent (`sent_hashes`), they don't; Automerge forgets them as sent + /// once the peer's heads cover them. (Changes kept until their predecessors came, + /// from another peer, count too: should this one lack them, it asks, and gets them, + /// `resend_if_needed`.) + pub(crate) fn came_from_peer( + &mut self, + doc: &Document, + before: &[crate::document::ChangeHash], + ) { + for change in doc.get_changes_meta(before) { + self.state.sent_hashes.insert(change.hash); + } + } + /// What the sync state knows of the peer (`PeerView`). fn peer_view(&self) -> PeerView { PeerView { @@ -888,6 +906,36 @@ mod tests { assert_eq!(carried, 1, "only the news"); } + /// Three messages arrive in reverse order: once the first is in, all three apply, + /// and whatever goes back to the peer carries none of its own changes. + #[test] + fn changes_from_the_peer_never_go_back() { + use crate::document::testing::change; + use automerge::{ROOT, transaction::Transactable}; + let (mut ours, mut theirs) = (Document::new(), Document::new()); + change(&mut ours, |doc| doc.put(ROOT, "early", 1).unwrap()); + let (mut a, mut b) = in_step(&mut ours, &mut theirs); + let (now, bytes) = (Instant::now(), SyncBytes::default()); + let mut messages = Vec::new(); + for i in 0..3 { + change(&mut ours, |doc| doc.put(ROOT, format!("k{i}"), i).unwrap()); + messages.push(a.generate(&ours, now, bytes).unwrap()); + a.assume_delivered(&ours.get_heads()); + a.send_succeeded(a.id()); + } + for message in messages.iter().rev() { + let before = theirs.get_heads(); + b.receive(&mut theirs, decode(message).unwrap()).unwrap(); + b.came_from_peer(&theirs, &before); + b.answered_as_delivered(&theirs, &theirs.get_heads()); + } + assert_eq!(theirs.get_heads(), ours.get_heads()); + if let Some(answer) = b.generate(&theirs, now, bytes) { + let carried = sync::Message::decode(&answer.0).unwrap().changes.len(); + assert_eq!(carried, 0, "a's own changes went back to it"); + } + } + /// Our transport refused a message taken as delivered: the peer's heads are as /// before, so the next message carries the change again. #[test] diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index 00ba1ad..07c990c 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -442,6 +442,10 @@ pub(crate) async fn handle_sync( } stats.heads_seen(heads_after.len()); let changed = heads_after != heads_before; + if changed { + // They came from it: none goes back (`SyncSession::came_from_peer`). + peer_state.sync.came_from_peer(&replica.doc, &heads_before); + } if carried && !changed { stats.nothing_new(payload_len, writer.is_some(), delivered); } |