1 files changed, 52 insertions, 11 deletions

diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index bc4b079..f1715a6 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -46,12 +46,23 @@ const CATCH_UP_STALL: Duration = Duration::from_secs(30);
 /// initial sync is done, and then only exchange heads. If the source sends nothing for
 /// `CATCH_UP_STALL`, the next peer takes over.
 ///
+/// Only a peer that is ready can be the source; one that said it isn't (another node
+/// still catching up) waits like the rest, until it says otherwise.
+///
 /// Returns how long to hold back sync with `peer`, if at all. Messages from waiting
 /// peers are still applied; we just don't answer them yet.
-fn defer_for_catch_up(state: &AppState, peer: &str) -> Option<Duration> {
+fn defer_for_catch_up(
+    state: &AppState,
+    peer: &str,
+    reports_ready: Option<bool>,
+) -> Option<Duration> {
     if state.initial_sync_done.load(Ordering::Acquire) {
         return None;
     }
+    if reports_ready == Some(false) {
+        // Its next message re-evaluates this; the timer is only a fallback.
+        return Some(CATCH_UP_STALL);
+    }
     let now = Instant::now();
     let mut source = state
         .catch_up_source
@@ -81,14 +92,37 @@ fn defer_for_catch_up(state: &AppState, peer: &str) -> Option<Duration> {
     None
 }
 
-/// A message from the catch-up source arrived: it is still making progress.
-fn note_catch_up_progress(state: &AppState, peer: &str) {
-    let mut source = state
-        .catch_up_source
-        .lock()
-        .unwrap_or_else(|poisoned| poisoned.into_inner());
-    if let Some(current) = source.as_mut().filter(|current| current.peer == peer) {
-        current.last_progress = Instant::now();
+/// A sync message from `peer` was applied. From the catch-up source, that's progress,
+/// unless the source turns out not to be ready: then it's dropped, and every peer is
+/// enqueued so the next ready one takes over at once.
+fn note_catch_up_progress(state: &AppState, peer: &str, ready: bool) {
+    let dropped = {
+        let mut source = state
+            .catch_up_source
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner());
+        match source.as_mut().filter(|current| current.peer == peer) {
+            Some(current) if ready => {
+                current.last_progress = Instant::now();
+                false
+            }
+            Some(_) => {
+                *source = None;
+                true
+            }
+            None => false,
+        }
+    };
+    if dropped {
+        debug!(peer = %peer, "Catch-up source isn't ready itself; catching up from another peer");
+        let tx = state.sync_events.clone();
+        let state = state.clone();
+        tokio::spawn(async move {
+            let peers: Vec<String> = state.peer_states.read().await.keys().cloned().collect();
+            for other in peers {
+                let _ = tx.send(SyncEvent::PeerNeedsSync(other));
+            }
+        });
     }
 }
 
@@ -121,7 +155,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool
         };
 
         if !retry_delivery {
-            if let Some(wait) = defer_for_catch_up(state, peer) {
+            if let Some(wait) = defer_for_catch_up(state, peer, peer_state.reports_ready) {
                 trace!(
                     wait_ms = wait.as_millis(),
                     "Catching up from another peer first; not syncing with this one yet"
@@ -199,6 +233,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool
             let psm = Arc::clone(&peer_state_mutex);
             let transport = Arc::clone(&state.transport);
             let peer_owned = peer.to_owned();
+            let ready = state.is_ready().await;
 
             let generated = tokio::task::spawn_blocking(
                 move || -> Result<Option<transport::EncodedMessage>> {
@@ -240,6 +275,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool
                         swarm_id,
                         epoch,
                         message,
+                        ready,
                     };
                     transport
                         .encode(&peer_owned, &ParsedMessage::Sync(envelope))
@@ -382,6 +418,7 @@ pub(crate) async fn handle_sync(
     }
 
     let epoch_label = envelope.epoch;
+    let from_ready = envelope.ready;
 
     let applied = {
         let doc = Arc::clone(&state.doc);
@@ -402,6 +439,7 @@ 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();
@@ -421,7 +459,10 @@ pub(crate) async fn handle_sync(
             break_sync_stall(&peer, &mut peer_state, unproductive);
 
             Ok(Some(
+                // Only from a ready peer: a peer still catching up has empty heads, which
+                // every document "contains".
                 check_caught_up
+                    && from_ready
                     && peer_state
                         .sync
                         .state
@@ -443,7 +484,7 @@ pub(crate) async fn handle_sync(
         return Ok(());
     };
 
-    note_catch_up_progress(state, peer);
+    note_catch_up_progress(state, peer, from_ready);
 
     if caught_up && !state.initial_sync_done.swap(true, Ordering::AcqRel) {
         info!(peer = %peer, "Initial sync complete");