drop stall breaker; resend needed changes

The stall breaker was superseded by regular re-handshaking Re-handshaking also
gets triggered, when a node needs some of the heads its peer believes it has
already sent.  In that case the peer now resends what's needed.

1 files changed, 21 insertions, 65 deletions

diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index c918fe8..be4b1e2 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -12,7 +12,7 @@ use tokio::sync::mpsc;
 use tokio::time::Instant;
 use tracing::{debug, info, trace, warn};
 
-use crate::state::{AppState, Liveness, PeerState, PendingOutbound, Standing};
+use crate::state::{AppState, Liveness, PendingOutbound, Standing};
 use crate::swarm::peer::{peer_swarm, send_hello_once};
 use crate::swarm::source;
 use crate::transport::{self, ParsedMessage};
@@ -326,9 +326,6 @@ pub(crate) async fn handle_sync(
             let mut peer_state = peer_state_mutex.blocking_lock();
             peer_state.reports_ready = Some(from_ready);
 
-            let heads_before = replica.doc.get_heads();
-            let their_heads_before = peer_state.sync.state.their_heads.clone();
-
             replica
                 .doc
                 .sync()
@@ -337,11 +334,7 @@ pub(crate) async fn handle_sync(
             peer_state.last_sync_rx = Some(Instant::now());
             peer_state.sync.reply_settled();
 
-            let heads_after = replica.doc.get_heads();
-            let unproductive = heads_before == heads_after
-                && peer_state.sync.state.their_heads == their_heads_before
-                && peer_state.sync.state.their_heads.as_ref() != Some(&heads_after);
-            break_sync_stall(&peer, &mut peer_state, unproductive);
+            resend_if_needed(&peer, &mut peer_state.sync.state);
 
             Ok(Some(
                 // Only from a ready peer: a peer still catching up has empty heads, which
@@ -386,64 +379,27 @@ pub(crate) async fn handle_sync(
     Ok(())
 }
 
-/// 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;
-
-/// Breaks sync livelocks.
-///
 /// Automerge never resends a change it believes the peer has received (`sent_hashes`).
-/// If that change was lost on the way (a dropped or undeliverable message, an epoch
-/// check, a reset of only one side's state), the two sides keep exchanging messages
-/// 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 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 session = &mut peer_state.sync;
-    if !unproductive {
-        session.stalled_exchanges = 0;
+/// If the message carrying it was lost while later ones arrive, the 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.)
+fn resend_if_needed(peer: &str, sync_state: &mut automerge::sync::State) {
+    let Some(need) = &sync_state.their_need else {
         return;
-    }
-    session.stalled_exchanges += 1;
-
-    let sync_state = &mut session.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 session.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 session.stalled_exchanges {
-        STALL_CLEAR_SENT => {
-            warn!(
-                peer = %peer,
-                sent = sync_state.sent_hashes.len(),
-                their_heads = ?sync_state.their_heads,
-                shared_heads = ?sync_state.shared_heads,
-                "Sync is stalled; forgetting which changes were sent"
-            );
-            sync_state.sent_hashes.clear();
-        }
-        STALL_RESET => {
-            warn!(peer = %peer, "Sync is still stalled; resetting it on both sides");
-            session.stalled_exchanges = 0;
-            peer_state.request_resync();
-        }
-        _ => {}
+    };
+    if need
+        .iter()
+        .any(|hash| sync_state.sent_hashes.contains(hash))
+    {
+        debug!(
+            peer = %peer,
+            sent = sync_state.sent_hashes.len(),
+            "Peer needs changes we had sent; forgetting what was sent"
+        );
+        sync_state.sent_hashes.clear();
     }
 }