1 files changed, 65 insertions, 1 deletions

diff --git a/src/sync.rs b/src/sync.rs
index 07ed1a0..407e88e 100644
--- a/src/sync.rs
+++ b/src/sync.rs
@@ -13,7 +13,7 @@ use tokio::time::Instant;
 use tracing::{debug, info, instrument, trace, warn};
 
 use crate::peer::{peer_cluster, send_hello_once};
-use crate::state::{AppState, Liveness, PeerMembership, PendingOutbound};
+use crate::state::{AppState, Liveness, PeerMembership, PeerState, PendingOutbound};
 use crate::transport::{self, ParsedMessage};
 
 const INITIAL_SYNC_RETRY: Duration = Duration::from_millis(500);
@@ -268,6 +268,7 @@ pub(crate) async fn handle_sync(
         let check_caught_up = !state.initial_sync_done.load(Ordering::Acquire);
         let epoch = envelope.epoch;
         let message = envelope.message;
+        let peer = peer.to_owned();
 
         // Some(caught_up) when applied, None when the message belongs to another epoch.
         tokio::task::spawn_blocking(move || -> Result<Option<bool>> {
@@ -281,6 +282,9 @@ pub(crate) async fn handle_sync(
 
             let mut peer_state = peer_state_mutex.blocking_lock();
 
+            let heads_before = replica.doc.get_heads();
+            let their_heads_before = peer_state.sync_state.their_heads.clone();
+
             replica
                 .doc
                 .sync()
@@ -288,6 +292,12 @@ pub(crate) async fn handle_sync(
                 .context("Automerge synchronization failed")?;
             peer_state.last_sync_rx = Some(Instant::now());
 
+            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);
+
             Ok(Some(
                 check_caught_up
                     && peer_state
@@ -324,6 +334,60 @@ 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_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 explicitly needs is never withheld as "already sent".
+/// * 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;
+    match peer_state.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");
+            peer_state.stalled_exchanges = 0;
+            peer_state.request_resync();
+        }
+        _ => {}
+    }
+}
+
 pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedReceiver<SyncEvent>) {
     let mut dirty_peers = HashSet::<String>::new();
     let mut retry_peers = HashSet::<String>::new();