From 2311d0cc57a2e0d5776b355983ca2e3f82f9980f Mon Sep 17 00:00:00 2001 From: tv Date: Fri, 23 Oct 2026 21:25:34 +0200 Subject: use only ready peers as sync source Sync messages now carries a ready flag, telling whether the sending peer itself is joined or is itself still (re-)joining. In the latter case, the peer won't be used as sync source. --- src/state.rs | 4 +++ src/swarm/sync.rs | 63 ++++++++++++++++++++++++++++++++++++++--------- src/transport/mod.rs | 4 +++ src/transport/mycelium.rs | 5 ++++ 4 files changed, 65 insertions(+), 11 deletions(-) diff --git a/src/state.rs b/src/state.rs index 4fbc05b..23892a9 100644 --- a/src/state.rs +++ b/src/state.rs @@ -120,6 +120,9 @@ pub(crate) struct PeerState { /// Epoch the peer advertised in its most recent message. pub(crate) epoch: Option, + + /// Whether the peer was ready according to its most recent sync message. + pub(crate) reports_ready: Option, } /// One Automerge sync with a peer, from a reset to the next. Everything a reset must @@ -322,6 +325,7 @@ impl PeerState { last_sync_rx: None, last_resync_at: None, epoch: None, + reports_ready: None, } } 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 { +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 @@ -81,14 +92,37 @@ fn defer_for_catch_up(state: &AppState, peer: &str) -> Option { 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 = 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> { @@ -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"); diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 69d9e70..eeba47f 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -62,6 +62,10 @@ pub struct SyncEnvelope { /// Epoch of the document the message was generated from. pub epoch: Epoch, pub message: Message, + /// The sender was ready when it generated the message: its document is the swarm's, + /// not one it is still filling (joining, rejoining). Only such a peer can tell a + /// catching-up node that it has everything; an empty document says nothing else. + pub ready: bool, } /// A message serialized for the wire. Cloning is cheap (shared buffer). diff --git a/src/transport/mycelium.rs b/src/transport/mycelium.rs index 84a34ea..193beaa 100644 --- a/src/transport/mycelium.rs +++ b/src/transport/mycelium.rs @@ -818,6 +818,9 @@ pub struct SyncEnvelopeShadow<'a> { pub message: automerge::sync::Message, #[serde(default)] epoch: EpochShadow, + /// Absent from nodes that predate it: never counts as ready. + #[serde(default)] + ready: bool, } // ===================================================================== @@ -890,6 +893,7 @@ pub fn sync_envelope_to_bytes(msg: &SyncEnvelope) -> Result, rmp_serde:: swarm_id: Cow::Borrowed(&msg.swarm_id), message: msg.message.clone(), // Clone required by automerge_msg_codec ownership rules epoch: epoch_to_shadow(&msg.epoch), + ready: msg.ready, }; rmp_serde::to_vec(&shadow) } @@ -899,6 +903,7 @@ pub fn bytes_to_sync_envelope(bytes: &[u8]) -> Result