fix catch-up / compaction in general

1 files changed, 18 insertions, 12 deletions

diff --git a/src/sync.rs b/src/sync.rs
index 407e88e..7def081 100644
--- a/src/sync.rs
+++ b/src/sync.rs
@@ -336,6 +336,7 @@ pub(crate) async fn handle_sync(
 
 /// Exchanges in a row that changed nothing on either side before `sent_hashes` is
 /// cleared, and before a full two-sided reset (re-handshake).
+const STALL_RESEND_NEEDED: u32 = 2;
 const STALL_CLEAR_SENT: u32 = 8;
 const STALL_RESET: u32 = 32;
 
@@ -347,27 +348,32 @@ const STALL_RESET: u32 = 32;
 /// with differing heads and no changes, as fast as the transport allows. The keepalive
 /// heal never fires then, because sync traffic keeps flowing.
 ///
-/// * A change the peer explicitly needs is never withheld as "already sent".
+/// * A change the peer keeps asking for is sent again.
 /// * After a few unproductive exchanges, forget what we sent: the next message carries
 ///   everything the peer's bloom filter lacks.
 /// * If that doesn't help either, reset both sides through a fresh handshake.
 fn break_sync_stall(peer: &str, peer_state: &mut PeerState, unproductive: bool) {
-    let sync_state = &mut peer_state.sync_state;
-    if let Some(need) = &sync_state.their_need {
-        let before = sync_state.sent_hashes.len();
-        sync_state.sent_hashes.retain(|hash| !need.contains(hash));
-        let resend = before - sync_state.sent_hashes.len();
-        if resend > 0 {
-            debug!(peer = %peer, resend, "Peer needs changes we had sent; sending them again");
-        }
-    }
-
     if !unproductive {
         peer_state.stalled_exchanges = 0;
         return;
     }
-
     peer_state.stalled_exchanges += 1;
+
+    let sync_state = &mut peer_state.sync_state;
+
+    // Not on the first unproductive exchange: when messages cross, the peer asks for a
+    // change that is merely still in flight, and resending it would only duplicate it.
+    if peer_state.stalled_exchanges >= STALL_RESEND_NEEDED {
+        if let Some(need) = &sync_state.their_need {
+            let before = sync_state.sent_hashes.len();
+            sync_state.sent_hashes.retain(|hash| !need.contains(hash));
+            let resend = before - sync_state.sent_hashes.len();
+            if resend > 0 {
+                debug!(peer = %peer, resend, "Peer still needs changes we had sent; sending them again");
+            }
+        }
+    }
+
     match peer_state.stalled_exchanges {
         STALL_CLEAR_SENT => {
             warn!(