require catch-up source to make progress

2 files changed, 92 insertions, 1 deletions

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<Instant>> {
     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<bool>) {