fix catch-up

1 files changed, 24 insertions, 4 deletions

diff --git a/src/state.rs b/src/state.rs
index b3a496c..4c7282f 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -102,6 +102,9 @@ pub(crate) struct PeerState {
 
     /// Epoch the peer advertised in its most recent message.
     pub(crate) epoch: Option<Epoch>,
+
+    /// Consecutive sync exchanges that changed nothing (livelock detection).
+    pub(crate) stalled_exchanges: u32,
 }
 
 impl Default for PeerState {
@@ -120,6 +123,7 @@ impl Default for PeerState {
             last_ping_at: None,
             last_sync_rx: None,
             last_resync_at: None,
+            stalled_exchanges: 0,
             epoch: None,
         }
     }
@@ -186,6 +190,19 @@ impl PeerState {
         due
     }
 
+    /// Schedules a fresh handshake, which resets sync on both sides: the peer resets on
+    /// our HELLO, we reset on its reply. The peer worker sends the HELLO on its next tick.
+    /// Usable without awaiting (e.g. under a blocking lock).
+    pub(crate) fn request_resync(&mut self) {
+        if matches!(self.hello_state, HelloState::Sending { .. }) {
+            return; // a HELLO is already on its way
+        }
+        self.hello_state = HelloState::AwaitingReply {
+            next_attempt: Instant::now(),
+            next_delay: Duration::from_millis(500),
+        };
+    }
+
     /// Self-healing check on PING/PONG. Returns true if a re-handshake should start.
     pub(crate) fn needs_heal(
         &mut self,
@@ -511,10 +528,7 @@ impl AppState {
             .map(|(id, peer)| (id.clone(), Arc::clone(peer)))
             .collect();
 
-        for (_, peer) in &peers {
-            peer.lock().await.reset_sync();
-        }
-
+        // Sync states were reset under the document lock by the swap itself.
         self.publish_change();
 
         for (id, _) in peers {
@@ -652,6 +666,12 @@ impl Host for AppState {
                     // Updated under the same lock, so the cached epoch never disagrees
                     // with the document for anyone holding the lock.
                     state.epoch_tx.send_replace(epoch);
+                    // Reset every peer's sync state before the lock is released, so no
+                    // message is generated from or applied to a state that describes the
+                    // old document. (Lock order doc -> peer, as in sync generation.)
+                    for peer in state.peer_states.blocking_read().values() {
+                        peer.blocking_lock().reset_sync();
+                    }
                     Ok(old_doc)
                 })
                 .await