From f9731343f71beced8ec8c817d19022fdfebc4545 Mon Sep 17 00:00:00 2001 From: tv Date: Fri, 23 Oct 2026 21:15:39 +0200 Subject: catch up with a single source only --- src/main.rs | 1 + src/state.rs | 15 ++++++++++ src/swarm/sync.rs | 87 ++++++++++++++++++++++++++++++++++++++++++++++++++++++- 3 files changed, 102 insertions(+), 1 deletion(-) diff --git a/src/main.rs b/src/main.rs index 5abc0e2..eac3c4c 100644 --- a/src/main.rs +++ b/src/main.rs @@ -185,6 +185,7 @@ async fn main() -> Result<()> { remake: Arc::new(remake::RemakeState::default()), swarm_key, stale: Default::default(), + catch_up_source: Default::default(), }; trace!("Application state initialized"); diff --git a/src/state.rs b/src/state.rs index 3ee9c6c..4fbc05b 100644 --- a/src/state.rs +++ b/src/state.rs @@ -528,6 +528,17 @@ pub(crate) struct AppState { /// While rejoining a newer epoch, the previous document, served read-only (and /// marked stale) instead of answering 503. pub(crate) stale: Arc>>, + + /// While the initial sync isn't done: the one peer we catch up from. + pub(crate) catch_up_source: Arc>>, +} + +/// The peer a catching-up node syncs with first (`swarm::sync::defer_for_catch_up`). +#[derive(Clone, Debug)] +pub(crate) struct CatchUpSource { + pub(crate) peer: String, + /// When it was chosen or last sent us something. + pub(crate) last_progress: Instant, } impl AppState { @@ -986,6 +997,10 @@ impl Host for AppState { fn reset_initial_sync(&self) { self.initial_sync_done .store(false, std::sync::atomic::Ordering::Release); + *self + .catch_up_source + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) = None; } fn subscribe_changes(&self) -> broadcast::Receiver> { 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>, } +/// 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 { + 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 = 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(); -- cgit v1.3.1