From b67ee4d3d59434f5a911b4eb0ee815ac9bcad3fe Mon Sep 17 00:00:00 2001
From: tv <tv@krebsco.de>
Date: Thu, 5 Nov 2026 01:27:07 +0100
Subject: predecessors go along

---
 src/document/sync.rs | 63 +++++++++++++++++++++++++++++++++++++++++++++++++---
 1 file changed, 60 insertions(+), 3 deletions(-)

(limited to 'src/document')

diff --git a/src/document/sync.rs b/src/document/sync.rs
index befb8e5..82b37f5 100644
--- a/src/document/sync.rs
+++ b/src/document/sync.rs
@@ -120,6 +120,10 @@ pub(crate) struct SyncSession {
     /// peer never learns our heads (a joining node then never finishes its initial
     /// sync).
     before_generate: (Vec<crate::document::ChangeHash>, bool),
+    /// Changes counted as sent because another parent brings them (`known_to_peer`),
+    /// and when: still in flight perhaps, so they go along with a change of ours that
+    /// builds on them (`with_predecessors`).
+    known_elsewhere: std::collections::HashMap<crate::document::ChangeHash, Instant>,
 
     /// The answer we expect to our last sync message(s), if any.
     pub(crate) reply: Option<SyncReplyWait>,
@@ -127,6 +131,10 @@ pub(crate) struct SyncSession {
     pub(crate) reply_backoff: Backoff,
 }
 
+/// How long a change counted as sent because another parent brings it may still be in
+/// flight to the peer (`SyncSession::with_predecessors`): a slow round trip, and more.
+const KNOWN_ELSEWHERE_FOR: Duration = Duration::from_secs(2);
+
 /// Sync messages we sent and expect an answer to. Automerge assumes reliable delivery:
 /// a lost message leaves both sides waiting (`in_flight`), and changes it carried are
 /// never sent again (`sent_hashes`). We resend after a timeout instead.
@@ -156,6 +164,7 @@ impl SyncSession {
             last_offer: None,
             before_receive: None,
             before_generate: (Vec::new(), false),
+            known_elsewhere: std::collections::HashMap::new(),
             reply: None,
             reply_backoff,
         }
@@ -479,8 +488,36 @@ impl SyncSession {
     /// The peer has `changes`, or will without us (another parent in their writer's
     /// tree brings them, `swarm::tree::forward`): counted as sent, they don't go to it,
     /// unless it asks for them (`resend_if_needed`).
-    pub(crate) fn known_to_peer(&mut self, changes: &[crate::document::ChangeHash]) {
-        self.state.sent_hashes.extend(changes.iter().copied());
+    pub(crate) fn known_to_peer(&mut self, changes: &[crate::document::ChangeHash], now: Instant) {
+        self.known_elsewhere
+            .retain(|_, at| now.saturating_duration_since(*at) < KNOWN_ELSEWHERE_FOR);
+        for hash in changes {
+            if self.state.sent_hashes.insert(*hash) {
+                self.known_elsewhere.insert(*hash, now);
+            }
+        }
+    }
+
+    /// We pass `changes` on to the peer: of their predecessors, those counted as sent
+    /// lately only because another parent brings them (`known_to_peer`) go along. That
+    /// one may not have arrived yet (with writers in turn, each change builds on the
+    /// last writer's, which comes along another tree): without them, the peer couldn't
+    /// apply ours, and would ask for them, a round trip later.
+    pub(crate) fn with_predecessors(
+        &mut self,
+        doc: &Document,
+        changes: &[crate::document::ChangeHash],
+    ) {
+        for hash in changes {
+            let Some(change) = doc.get_change_by_hash(hash) else {
+                continue;
+            };
+            for dep in change.deps() {
+                if self.known_elsewhere.remove(dep).is_some() {
+                    self.state.sent_hashes.remove(dep);
+                }
+            }
+        }
     }
 
     /// What the sync state knows of the peer (`PeerView`).
@@ -1257,7 +1294,7 @@ mod tests {
         let (now, bytes) = (Instant::now(), SyncBytes::default());
 
         change(&mut ours, |doc| doc.put(ROOT, "elsewhere", 2).unwrap());
-        a.known_to_peer(&ours.get_heads());
+        a.known_to_peer(&ours.get_heads(), Instant::now());
         let message = a.generate(&ours, now, bytes).expect("our heads moved");
         a.send_succeeded(a.id());
         let (carried, _) = decode(&message).unwrap().sizes();
@@ -1274,6 +1311,26 @@ mod tests {
         assert_eq!(theirs.get_heads(), ours.get_heads());
     }
 
+    /// A change counted as known to the peer lately goes along with one of ours that
+    /// builds on it: the peer applies both at once, without asking.
+    #[test]
+    fn predecessors_known_elsewhere_go_along() {
+        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());
+
+        change(&mut ours, |doc| doc.put(ROOT, "elsewhere", 2).unwrap());
+        a.known_to_peer(&ours.get_heads(), now);
+        change(&mut ours, |doc| doc.put(ROOT, "along", 3).unwrap());
+        a.with_predecessors(&ours, &ours.get_heads());
+        let message = a.generate(&ours, now, bytes).expect("our heads moved");
+        b.receive(&mut theirs, decode(&message).unwrap()).unwrap();
+        assert_eq!(theirs.get_heads(), ours.get_heads(), "both in one message");
+    }
+
     /// Our transport refused a message taken as delivered: the peer's heads are as
     /// before, so the next message carries the change again.
     #[test]
-- 
cgit v1.3.1