1 files changed, 109 insertions, 2 deletions
diff --git a/src/swarm/source.rs b/src/swarm/source.rs
index 378c40a..9eac445 100644
--- a/src/swarm/source.rs
+++ b/src/swarm/source.rs
@@ -15,12 +15,22 @@
 //! 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
+//! `Timing::delivery_limit`, after which Mycelium has delivered or dropped anything it
+//! sent before. It isn't chosen again for another `delivery_limit`, so another peer
+//! takes over rather than the same one stalling again; if it is the only candidate,
+//! catching up waits that long before trying it again. 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 +45,12 @@ 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 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
@@ -42,6 +58,13 @@ fn lock_preference(state: &AppState) -> std::sync::MutexGuard<'_, Option<Instant
         .unwrap_or_else(|poisoned| poisoned.into_inner())
 }
 
+fn lock_avoided(state: &AppState) -> std::sync::MutexGuard<'_, Option<(String, Instant)>> {
+    state
+        .catch_up_avoided
+        .lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner())
+}
+
 fn catching_up(state: &AppState) -> bool {
     !state.initial_sync_done.load(Ordering::Acquire)
 }
@@ -60,12 +83,20 @@ pub(crate) fn defer(state: &AppState, peer: &str, peer_state: &PeerState) -> Opt
         return None;
     }
     let stall = state.timing.catch_up_probe * UNANSWERED_PROBES;
+    let now = Instant::now();
+    let avoided_until = lock_avoided(state)
+        .as_ref()
+        .filter(|(avoided, until)| avoided == peer && now < *until)
+        .map(|(_, until)| *until);
+    if let Some(until) = avoided_until {
+        // Given up for not answering; the timer brings it back once it may be chosen.
+        return Some(until - now);
+    }
     if peer_state.reports_ready == Some(false) || peer_state.syncs_with_us == Some(false) {
         // Not a candidate; its next message re-evaluates this, the timer is a fallback.
         return Some(stall);
     }
 
-    let now = Instant::now();
     let mut source = lock(state);
     let quiet = match &*source {
         Some(current) if current.peer == peer => return None,
@@ -115,6 +146,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 +162,78 @@ 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 >= state.timing.delivery_limit;
+        let timeouts = current.reply_timeouts;
+        if give_up {
+            *source = None;
+            *lock_avoided(state) = Some((
+                peer.to_owned(),
+                Instant::now() + state.timing.delivery_limit,
+            ));
+        }
+        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>) {
@@ -188,6 +294,7 @@ pub(crate) fn probe_due(state: &AppState, peer: &str, now: Instant) -> bool {
 pub(crate) fn finished(state: &AppState, from: &str) {
     *lock(state) = None;
     *lock_preference(state) = None;
+    *lock_avoided(state) = None;
     let state = state.clone();
     let from = from.to_owned();
     tokio::spawn(async move {