From 31e6e6a2801a19515daa7adda09b0d24d420f18e Mon Sep 17 00:00:00 2001 From: tv Date: Sun, 25 Oct 2026 20:40:00 +0100 Subject: require catch-up source to make progress --- src/state.rs | 6 ++++ src/swarm/peer.rs | 3 ++ src/swarm/source.rs | 90 ++++++++++++++++++++++++++++++++++++++++++++++++++++- 3 files changed, 98 insertions(+), 1 deletion(-) (limited to 'src') diff --git a/src/state.rs b/src/state.rs index d14c1ea..af01cf0 100644 --- a/src/state.rs +++ b/src/state.rs @@ -649,6 +649,12 @@ pub(crate) struct CatchUpSource { pub(crate) peer: String, /// When it was chosen or last sent us anything. pub(crate) last_heard: Instant, + /// When it was chosen or last sent us a sync message: progress, not just liveness. + pub(crate) last_sync: Instant, + /// Our sync messages to it in a row whose answer didn't come in time. + pub(crate) reply_timeouts: u32, + /// When we last warned that catching up from it stalls. + pub(crate) last_warned: Option, /// Our last probe PING to it. pub(crate) last_probe: Option, /// Its keepalives in a row saying it wouldn't sync with us. diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs index 8c1e7eb..238194c 100644 --- a/src/swarm/peer.rs +++ b/src/swarm/peer.rs @@ -468,6 +468,7 @@ pub(crate) async fn run_peer_worker(state: AppState) { if let Some(changes) = unanswered { debug!(peer = %peer, changes, "No answer to our sync message; sending again"); + crate::swarm::source::reply_timed_out(&state, peer); let _ = state .sync_events .send(SyncEvent::PeerNeedsSync(peer.clone())); @@ -500,6 +501,7 @@ pub(crate) async fn run_peer_worker(state: AppState) { crate::swarm::members::ensure_registered(&state).await; crate::swarm::stability::tick(&state).await; crate::swarm::stats::measure_saved_size(&state); + crate::swarm::source::warn_if_stalled(&state); crate::swarm::state_file::update(&state, false).await; { let known = state.peer_states.read().await; @@ -847,6 +849,7 @@ pub(crate) async fn handle_message( debug!(peer = %peer, epoch = %envelope.epoch, "Dropping sync message from another epoch"); return Ok(()); } + crate::swarm::source::synced_from(state, &peer); handle_sync(state, &peer, envelope).await?; } transport::ParsedMessage::Ping(remote) => { diff --git a/src/swarm/source.rs b/src/swarm/source.rs index 378c40a..a8866a6 100644 --- a/src/swarm/source.rs +++ b/src/swarm/source.rs @@ -15,12 +15,20 @@ //! is quiet for `Timing::catch_up_probe`, we PING it; after three unanswered probes, or //! when it says it isn't ready, or twice in a row that it wouldn't sync with us (the //! first time may be a handshake still under way), the next peer takes over. +//! +//! Alive isn't enough: a source that answers keepalives but not our sync messages (a +//! big answer lost on the way, a stuck sync state) would keep us catching up forever. +//! So it is also given up after three of our sync messages in a row went unanswered, +//! once nothing from it can still be on the way: no sync message from it for +//! `MYCELIUM_DELIVERY_LIMIT`, after which Mycelium has delivered or dropped anything it +//! sent before. Meanwhile, every minute without a sync message from it is logged as a +//! warning. use std::sync::atomic::Ordering; use std::time::Duration; use tokio::time::Instant; -use tracing::{debug, info}; +use tracing::{debug, info, warn}; use crate::state::{AppState, CatchUpSource, PeerState}; use crate::swarm::sync::SyncEvent; @@ -35,6 +43,16 @@ const UNANSWERED_PROBES: u32 = 3; /// Keepalives in a row saying "I wouldn't sync with you" before the source is given up. const REFUSALS: u32 = 2; +/// Unanswered sync messages in a row before the source may be given up. +const REPLY_TIMEOUTS: u32 = 3; + +/// How long Mycelium tries to deliver a message before dropping it: a source silent +/// (in sync) for longer has nothing on the way to us any more. +const MYCELIUM_DELIVERY_LIMIT: Duration = Duration::from_secs(300); + +/// How often a stalled catch-up is logged. +const STALL_WARNING: Duration = Duration::from_secs(60); + fn lock_preference(state: &AppState) -> std::sync::MutexGuard<'_, Option> { state .catch_up_preference @@ -115,6 +133,9 @@ pub(crate) fn defer(state: &AppState, peer: &str, peer_state: &PeerState) -> Opt *source = Some(CatchUpSource { peer: peer.to_owned(), last_heard: now, + last_sync: now, + reply_timeouts: 0, + last_warned: None, last_probe: None, refusals: 0, }); @@ -128,6 +149,73 @@ pub(crate) fn heard_from(state: &AppState, peer: &str) { } } +/// A sync message from `peer` arrived: progress, if it is the source. +pub(crate) fn synced_from(state: &AppState, peer: &str) { + if let Some(current) = lock(state).as_mut().filter(|current| current.peer == peer) { + current.last_sync = Instant::now(); + current.reply_timeouts = 0; + } +} + +/// Our sync message to `peer` went unanswered (`SyncSession` reply deadline). Gives up +/// the source after `REPLY_TIMEOUTS` in a row, once nothing from it can be on the way. +pub(crate) fn reply_timed_out(state: &AppState, peer: &str) { + if !catching_up(state) { + return; + } + let given_up = { + let mut source = lock(state); + let Some(current) = source.as_mut().filter(|current| current.peer == peer) else { + return; + }; + current.reply_timeouts += 1; + let silent = current.last_sync.elapsed(); + let give_up = current.reply_timeouts >= REPLY_TIMEOUTS && silent >= MYCELIUM_DELIVERY_LIMIT; + let timeouts = current.reply_timeouts; + if give_up { + *source = None; + } + give_up.then_some((timeouts, silent)) + }; + if let Some((timeouts, silent)) = given_up { + warn!( + peer = %peer, + reply_timeouts = timeouts, + silent_secs = silent.as_secs(), + "Catch-up source doesn't answer our sync messages; catching up from another peer" + ); + resync_everyone(state); + } +} + +/// Warns, at most every `STALL_WARNING`, while the source sends no sync messages. Called +/// by the peer worker on its maintenance tick. +pub(crate) fn warn_if_stalled(state: &AppState) { + if !catching_up(state) { + return; + } + let mut source = lock(state); + let Some(current) = source.as_mut() else { + return; + }; + let silent = current.last_sync.elapsed(); + if silent < STALL_WARNING + || current + .last_warned + .is_some_and(|at| at.elapsed() < STALL_WARNING) + { + return; + } + current.last_warned = Some(Instant::now()); + warn!( + peer = %current.peer, + silent_secs = silent.as_secs(), + reply_timeouts = current.reply_timeouts, + heard_secs_ago = current.last_heard.elapsed().as_secs(), + "Catching up stalls: no sync message from the catch-up source" + ); +} + /// What `peer` said about itself, in a sync message (`syncs_with_us: None`) or a /// keepalive. Gives up the source if that rules it out. pub(crate) fn reported(state: &AppState, peer: &str, ready: bool, syncs_with_us: Option) { -- cgit v1.3.1