1 files changed, 86 insertions, 1 deletions

diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index dc2e665..bc4b079 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, CatchUpSource, Liveness, PeerState, PendingOutbound, Standing};
 use crate::swarm::peer::{peer_swarm, send_hello_once};
 use crate::transport::{self, ParsedMessage};
 
@@ -34,6 +34,64 @@ pub(crate) struct ChangeSet {
     pub(crate) paths: Vec<Vec<String>>,
 }
 
+/// A catch-up source that sent nothing for this long is given up for the next peer.
+const CATCH_UP_STALL: Duration = Duration::from_secs(30);
+
+/// While catching up (joining, or rejoining a newer epoch), sync with one peer only.
+///
+/// Every peer would otherwise send us what we lack, all at once: a node with an empty
+/// document gets the whole document from each of its peers, and a node that is merely
+/// behind gets the missing changes from each. Automerge can't deduplicate across peers.
+/// So the first peer we would sync with becomes the source; the others wait until the
+/// initial sync is done, and then only exchange heads. If the source sends nothing for
+/// `CATCH_UP_STALL`, the next peer takes over.
+///
+/// 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> {
+    if state.initial_sync_done.load(Ordering::Acquire) {
+        return None;
+    }
+    let now = Instant::now();
+    let mut source = state
+        .catch_up_source
+        .lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner());
+    let stalled = match &*source {
+        Some(current) if current.peer == peer => return None,
+        Some(current) if now < current.last_progress + CATCH_UP_STALL => {
+            return Some(current.last_progress + CATCH_UP_STALL - now);
+        }
+        Some(current) => Some(current.peer.clone()),
+        None => None,
+    };
+    match stalled {
+        Some(previous) => info!(
+            peer = %peer,
+            previous = %previous,
+            stalled_secs = CATCH_UP_STALL.as_secs(),
+            "Catch-up source sent nothing; catching up from another peer"
+        ),
+        None => debug!(peer = %peer, "Catching up from this peer first"),
+    }
+    *source = Some(CatchUpSource {
+        peer: peer.to_owned(),
+        last_progress: now,
+    });
+    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();
+    }
+}
+
 async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool) -> Result<()> {
     let Some(peer_state_mutex) = state.get_peer_state(peer).await else {
         trace!("Peer was forgotten; skipping synchronization");
@@ -63,6 +121,22 @@ 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) {
+                trace!(
+                    wait_ms = wait.as_millis(),
+                    "Catching up from another peer first; not syncing with this one yet"
+                );
+                // Done earlier when the initial sync completes (every peer is enqueued
+                // then); this only matters if the source stalls.
+                let tx = state.sync_events.clone();
+                let peer = peer.to_owned();
+                tokio::spawn(async move {
+                    tokio::time::sleep(wait).await;
+                    let _ = tx.send(SyncEvent::PeerNeedsSync(peer));
+                });
+                return Ok(());
+            }
+
             let now = Instant::now();
             if let Some(wait) = peer_state.sync.opening_wait(now) {
                 trace!(
@@ -369,8 +443,19 @@ pub(crate) async fn handle_sync(
         return Ok(());
     };
 
+    note_catch_up_progress(state, peer);
+
     if caught_up && !state.initial_sync_done.swap(true, Ordering::AcqRel) {
         info!(peer = %peer, "Initial sync complete");
+        *state
+            .catch_up_source
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner()) = None;
+        // The peers we held back: now they only need to learn our heads.
+        let peers: Vec<String> = state.peer_states.read().await.keys().cloned().collect();
+        for other in peers.into_iter().filter(|other| other != peer) {
+            let _ = state.sync_events.send(SyncEvent::PeerNeedsSync(other));
+        }
     }
 
     state.notify_store_changed();