changes from a peer never go back

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);
             }