From d8f0933b5c88634ee7e83b435241084baa6abf5b Mon Sep 17 00:00:00 2001 From: tv Date: Fri, 23 Oct 2026 22:21:15 +0200 Subject: probe sync source during catch-up This allows changing sources without the fixed 30 s stall duration. --- src/swarm/sync.rs | 111 +++--------------------------------------------------- 1 file changed, 6 insertions(+), 105 deletions(-) (limited to 'src/swarm/sync.rs') 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>, } -/// 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, -) -> Option { - 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 = 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 = 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(); -- cgit v1.3.1