resend only what is needed

2 files changed, 62 insertions, 27 deletions

diff --git a/src/document/sync.rs b/src/document/sync.rs
index fda8889..628fd4f 100644
--- a/src/document/sync.rs
+++ b/src/document/sync.rs
@@ -310,25 +310,21 @@ impl SyncSession {
     /// peer keeps asking for it (`need`, from the heads we advertise) and we keep not
     /// sending it: a livelock the reply timeout can't see (answers keep coming) and the
     /// keepalive heal can't either (sync never goes idle). So when the peer needs a
-    /// change we think it has, what we think it has is wrong: forget all of it, and the
-    /// next message carries everything its bloom filter lacks. (A reordered message can
-    /// make that send a change twice; cheap.) Call after `receive`.
-    pub(crate) fn resend_if_needed(&mut self, peer: &str) {
-        let Some(need) = &self.state.their_need else {
-            return;
-        };
-        if need
-            .iter()
-            .any(|hash| self.state.sent_hashes.contains(hash))
-        {
+    /// change we think it has, what we think of it is wrong: forget having sent it, and
+    /// what it builds on that we sent too (`forget_sent_for_need`), and the next message
+    /// carries them. Not what we sent after it: messages overtake each other (we don't
+    /// wait for answers to those taken as delivered), so a peer often needs what is only
+    /// late, while the messages after it are still on their way. Call after `receive`.
+    pub(crate) fn resend_if_needed(&mut self, doc: &Document, peer: &str) {
+        let forgotten = forget_sent_for_need(&mut self.state, doc);
+        if forgotten > 0 {
             debug!(
                 peer = %peer,
-                sent = self.state.sent_hashes.len(),
-                "Peer needs changes we had sent; forgetting what was sent"
+                forgotten,
+                "Peer needs changes we had sent; sending them again"
             );
-            self.state.sent_hashes.clear();
         }
-        if answer_for_everything_since_shared(&mut self.state) {
+        if answer_for_everything_since_shared(&mut self.state, doc) {
             debug!(
                 peer = %peer,
                 need = self.state.their_need.as_ref().map_or(0, Vec::len),
@@ -462,9 +458,11 @@ impl SyncSession {
 /// 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 {
+/// delta when large: `swarm::sync`), and what it had already only twice. But what we
+/// sent and it doesn't need may be on its way still: that stays sent, but for what it
+/// needs and what that builds on (`forget_sent_for_need`). Returns whether it did. Call
+/// after receiving.
+fn answer_for_everything_since_shared(state: &mut sync::State, doc: &Document) -> bool {
     let asks_without_saying = state.their_have.as_ref().is_some_and(Vec::is_empty)
         && state
             .their_need
@@ -477,11 +475,43 @@ fn answer_for_everything_since_shared(state: &mut sync::State) -> bool {
         last_sync: state.shared_heads.clone(),
         bloom: sync::BloomFilter::default(),
     }]);
-    // What we sent since may be what was lost.
-    state.sent_hashes.clear();
+    // What it needs, and what that builds on, may be what was lost.
+    forget_sent_for_need(state, doc);
     true
 }
 
+/// Forgets having sent what the peer needs (`their_need`) and the changes those build
+/// on that we sent too, following their predecessors as long as we sent them: so all of
+/// a run of lost messages goes again at once. Changes sent after them stay sent (they
+/// may only be late). Returns how many were forgotten.
+fn forget_sent_for_need(state: &mut sync::State, doc: &Document) -> usize {
+    let Some(need) = &state.their_need else {
+        return 0;
+    };
+    let mut forgotten = 0;
+    let mut next: Vec<crate::document::ChangeHash> = need
+        .iter()
+        .filter(|hash| state.sent_hashes.contains(hash))
+        .copied()
+        .collect();
+    while let Some(hash) = next.pop() {
+        if !state.sent_hashes.remove(&hash) {
+            continue;
+        }
+        forgotten += 1;
+        if let Some(change) = doc.get_change_by_hash(&hash) {
+            next.extend(
+                change
+                    .deps()
+                    .iter()
+                    .filter(|dep| state.sent_hashes.contains(dep))
+                    .copied(),
+            );
+        }
+    }
+    forgotten
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -527,14 +557,14 @@ mod tests {
         let ask = b.generate_sync_message(&mut at_b).unwrap();
         assert!(ask.have.is_empty() && ask.need.len() == 1);
         a.receive_sync_message(&mut at_a, ask).unwrap();
-        assert!(answer_for_everything_since_shared(&mut at_a));
+        assert!(answer_for_everything_since_shared(&mut at_a, &a));
         // All since the shared heads, in one message (here the whole document: more
         // than a third of it is missing).
         let answer = a.generate_sync_message(&mut at_a).unwrap();
         b.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));
+        assert!(!answer_for_everything_since_shared(&mut at_b, &b));
     }
 
     fn session() -> SyncSession {
@@ -728,7 +758,7 @@ mod tests {
             .expect("b asks for what it lacks");
         b.send_succeeded(b.id());
         a.receive(&mut ours, decode(&ask).unwrap()).unwrap();
-        a.resend_if_needed("b");
+        a.resend_if_needed(&ours, "b");
         let answer = a.generate(&ours, now, bytes).expect("a sends the lost one");
         a.send_succeeded(a.id());
         b.receive(&mut theirs, decode(&answer).unwrap()).unwrap();
@@ -736,8 +766,8 @@ mod tests {
     }
 
     /// Two messages taken as delivered arrive in reverse order: the peer asks for the
-    /// first, which arrives meanwhile, and what is sent again is taken as delivered too.
-    /// Nobody waits for an answer that won't come.
+    /// first, which arrives meanwhile; only the first is sent again (the second was on
+    /// its way), taken as delivered too. Nobody waits for an answer that won't come.
     #[test]
     fn reordered_delivered_messages_leave_nothing_awaited() {
         use crate::document::testing::change;
@@ -773,8 +803,13 @@ mod tests {
         b.send_succeeded(b.id());
 
         a.receive(&mut ours, decode(&ask).unwrap()).unwrap();
-        a.resend_if_needed("b");
+        a.resend_if_needed(&ours, "b");
         let again = send(&mut a, &ours);
+        let carried = sync::Message::decode(&again.0).unwrap().changes.len();
+        assert_eq!(
+            carried, 1,
+            "only the first again, not the second, on its way"
+        );
         if let Some(level) = level {
             a.receive(&mut ours, decode(&level).unwrap()).unwrap();
         }
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 700764f..3737f85 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -430,7 +430,7 @@ pub(crate) async fn handle_sync(
             }
             peer_state.sync.reply_settled();
 
-            peer_state.sync.resend_if_needed(&peer);
+            peer_state.sync.resend_if_needed(&replica.doc, &peer);
             replica.refresh_heads();
             let heads_after = replica.heads().to_vec();
             if delivered {