From c0f53c8d163339c675fa0336e3588b7cfa4ac87e Mon Sep 17 00:00:00 2001
From: tv <tv@krebsco.de>
Date: Wed, 28 Oct 2026 15:12:25 +0100
Subject: batch-send missing changes during sync

---
 src/state/session.rs | 84 ++++++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 84 insertions(+)

(limited to 'src')

diff --git a/src/state/session.rs b/src/state/session.rs
index f09ae41..8885699 100644
--- a/src/state/session.rs
+++ b/src/state/session.rs
@@ -220,6 +220,13 @@ impl SyncSession {
             );
             self.state.sent_hashes.clear();
         }
+        if answer_for_everything_since_shared(&mut self.state) {
+            debug!(
+                peer = %peer,
+                need = self.state.their_need.as_ref().map_or(0, Vec::len),
+                "Peer lacks the changes ours build on; sending everything since what we share"
+            );
+        }
     }
 
     /// For the simulator's dump: the peer's heads as Automerge knows them, and whether
@@ -329,10 +336,87 @@ impl SyncSession {
     }
 }
 
+/// A peer that got a change of ours without the changes it builds on (they were lost,
+/// a later one arrived) asks for those it lacks, and while it does, Automerge leaves
+/// out what it has (its bloom filter: `have` waits until all it needs is in our heads).
+/// Without a `have`, Automerge sends exactly what was asked for, and the peer, finding
+/// each change's own predecessors missing in turn, asks again: one round trip per
+/// missing change, hundreds after a short outage. So answer as if it had said it has
+/// nothing since the heads we share: everything since then goes in one message (a
+/// delta when large: `swarm::sync`), and what it had already only twice. Returns
+/// whether it did. Call after receiving.
+fn answer_for_everything_since_shared(state: &mut sync::State) -> bool {
+    let asks_without_saying = state.their_have.as_ref().is_some_and(Vec::is_empty)
+        && state
+            .their_need
+            .as_ref()
+            .is_some_and(|need| !need.is_empty());
+    if !asks_without_saying {
+        return false;
+    }
+    state.their_have = Some(vec![sync::Have {
+        last_sync: state.shared_heads.clone(),
+        bloom: sync::BloomFilter::default(),
+    }]);
+    // What we sent since may be what was lost.
+    state.sent_hashes.clear();
+    true
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
 
+    /// After a loss, the changes the peer lacks go in one message, not one per round
+    /// trip (`answer_for_everything_since_shared`).
+    #[test]
+    fn missing_predecessors_go_at_once() {
+        use automerge::transaction::Transactable;
+        let mut a = AutoCommit::new();
+        let mut b = AutoCommit::new();
+        let (mut at_a, mut at_b) = (sync::State::new(), sync::State::new());
+        a.put(automerge::ROOT, "early", 1).unwrap();
+        // Until in step.
+        for _ in 0..10 {
+            let to_b = a.sync().generate_sync_message(&mut at_a);
+            let to_a = b.sync().generate_sync_message(&mut at_b);
+            if to_b.is_none() && to_a.is_none() {
+                break;
+            }
+            if let Some(m) = to_b {
+                b.sync().receive_sync_message(&mut at_b, m).unwrap();
+            }
+            if let Some(m) = to_a {
+                a.sync().receive_sync_message(&mut at_a, m).unwrap();
+            }
+        }
+        assert_eq!(a.get_heads(), b.get_heads());
+        // Five changes, whose message is lost; then a sixth, which arrives alone.
+        for i in 0..5 {
+            a.put(automerge::ROOT, format!("lost{i}"), i).unwrap();
+            a.commit();
+        }
+        let _lost = a.sync().generate_sync_message(&mut at_a).unwrap();
+        a.put(automerge::ROOT, "last", 6).unwrap();
+        a.commit();
+        at_a.in_flight = false; // As after our reply timeout.
+        let last = a.sync().generate_sync_message(&mut at_a).unwrap();
+        assert_eq!(last.changes.len(), 1);
+        b.sync().receive_sync_message(&mut at_b, last).unwrap();
+        // b asks for the fifth, saying nothing of what it has.
+        let ask = b.sync().generate_sync_message(&mut at_b).unwrap();
+        assert!(ask.have.is_empty() && ask.need.len() == 1);
+        a.sync().receive_sync_message(&mut at_a, ask).unwrap();
+        assert!(answer_for_everything_since_shared(&mut at_a));
+        // All since the shared heads, in one message (here the whole document: more
+        // than a third of it is missing).
+        let answer = a.sync().generate_sync_message(&mut at_a).unwrap();
+        b.sync().receive_sync_message(&mut at_b, answer).unwrap();
+        assert_eq!(b.get_heads(), a.get_heads());
+        // A message with what it has is left alone.
+        assert!(!answer_for_everything_since_shared(&mut at_b));
+    }
+
     fn session() -> SyncSession {
         SyncSession::new(Backoff::new(
             Duration::from_secs(1),
-- 
cgit v1.3.1