fix catch-up / compaction in general

4 files changed, 51 insertions, 17 deletions

diff --git a/src/compaction/switch.rs b/src/compaction/switch.rs
index 9f17df2..4093e7d 100644
--- a/src/compaction/switch.rs
+++ b/src/compaction/switch.rs
@@ -141,6 +141,11 @@ async fn switch<H: Host>(host: &H, id: &str, record: &Record) -> Result<ReplaySt
     let previous = request.previous;
     let base = request.base_heads.clone();
     let retention = record.proposal.policy.archive_retention.clone();
+    // Only the coordinator carries the record into the new epoch: one writer, so the
+    // nodes don't all write the same path concurrently right after switching (a merge
+    // conflict, plus one change per node to sync). The epoch metadata in the snapshot
+    // keeps the essentials on every node anyway.
+    let carry_record = record.proposal.coordinator == host.node_id();
     let record_path = records::record_path(id, "");
 
     info!(compaction = %id, from = %previous, to = %expected, "Switching epoch");
@@ -172,8 +177,10 @@ async fn switch<H: Host>(host: &H, id: &str, record: &Record) -> Result<ReplaySt
             let mut doc = snapshot.doc.with_actor(ActorId::random());
 
             // Keep this compaction's record visible in the new epoch, for its history.
-            if let Err(error) = store::copy_path(&copy, &mut doc, &record_path) {
-                warn!(error = %error, "Could not carry the compaction record into the new epoch");
+            if carry_record {
+                if let Err(error) = store::copy_path(&copy, &mut doc, &record_path) {
+                    warn!(error = %error, "Could not carry the compaction record into the new epoch");
+                }
             }
 
             let heads = copy.get_heads();
diff --git a/src/peer.rs b/src/peer.rs
index 015b5ff..50460b0 100644
--- a/src/peer.rs
+++ b/src/peer.rs
@@ -108,7 +108,7 @@ fn heads_digest(doc: &mut AutoCommit) -> transport::HeadsDigest {
     hasher.finalize().into()
 }
 
-async fn local_keepalive(state: &AppState) -> transport::Keepalive {
+pub(crate) async fn local_keepalive(state: &AppState) -> transport::Keepalive {
     // AutoCommit::get_heads takes &mut self (it closes the pending transaction).
     // Digest and epoch come from the same locked replica, so they always match.
     let mut replica = state.doc.write().await;
@@ -517,8 +517,11 @@ async fn observe_peer_epoch(state: &AppState, peer: &str, remote: Epoch) -> bool
         let mut peer_state = peer_state_mutex.lock().await;
         let previous = peer_state.epoch.replace(remote);
         if previous.is_some_and(|previous| previous != remote) {
-            // The peer switched documents; our sync state describes its old one.
+            // The peer switched documents; our sync state describes its old one. Count
+            // the restart as sync activity, so the keepalive heal doesn't re-handshake
+            // (resetting both sides again) while this fresh sync is under way.
             peer_state.reset_sync();
+            peer_state.last_sync_rx = Some(Instant::now());
         }
         previous
     };
diff --git a/src/state.rs b/src/state.rs
index 4c7282f..f9da5e0 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -531,6 +531,20 @@ impl AppState {
         // Sync states were reset under the document lock by the swap itself.
         self.publish_change();
 
+        // Tell peers about the new epoch now instead of with the next keepalive (up to a
+        // keepalive interval later): sync with peers already on it resumes right away,
+        // and peers that never saw the commit start catching up.
+        let keepalive = crate::peer::local_keepalive(self).await;
+        for (id, _) in &peers {
+            if let Err(error) = self
+                .transport
+                .send_message(id, &transport::ParsedMessage::Ping(keepalive.clone()))
+                .await
+            {
+                debug!(peer = %id, error = %format!("{error:#}"), "Epoch announcement failed");
+            }
+        }
+
         for (id, _) in peers {
             let _ = self.sync_events.send(SyncEvent::PeerNeedsSync(id));
         }
@@ -669,8 +683,12 @@ impl Host for AppState {
                     // 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.)
+                    let now = Instant::now();
                     for peer in state.peer_states.blocking_read().values() {
-                        peer.blocking_lock().reset_sync();
+                        let mut peer = peer.blocking_lock();
+                        peer.reset_sync();
+                        // A fresh start, not an idle sync: no heal re-handshake for it.
+                        peer.last_sync_rx = Some(now);
                     }
                     Ok(old_doc)
                 })
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!(