probe sync source during catch-up

This allows changing sources without the fixed 30 s stall duration.

1 files changed, 6 insertions, 105 deletions

diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index f1715a6..04a174c 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -12,8 +12,9 @@ use tokio::sync::mpsc;
 use tokio::time::Instant;
 use tracing::{debug, info, trace, warn};
 
-use crate::state::{AppState, CatchUpSource, Liveness, PeerState, PendingOutbound, Standing};
+use crate::state::{AppState, Liveness, PeerState, PendingOutbound, Standing};
 use crate::swarm::peer::{peer_swarm, send_hello_once};
+use crate::swarm::source;
 use crate::transport::{self, ParsedMessage};
 
 const INITIAL_SYNC_RETRY: Duration = Duration::from_millis(500);
@@ -34,98 +35,6 @@ 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.
-///
-/// 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,
-    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
-        .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 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));
-            }
-        });
-    }
-}
-
 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");
@@ -155,13 +64,13 @@ 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, peer_state.reports_ready) {
+            if let Some(wait) = source::defer(state, peer, &peer_state) {
                 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.
+                // then); this only matters if the source goes quiet.
                 let tx = state.sync_events.clone();
                 let peer = peer.to_owned();
                 tokio::spawn(async move {
@@ -484,19 +393,11 @@ pub(crate) async fn handle_sync(
         return Ok(());
     };
 
-    note_catch_up_progress(state, peer, from_ready);
+    source::reported(state, peer, from_ready, None);
 
     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));
-        }
+        source::finished(state, peer);
     }
 
     state.notify_store_changed();