avoid stalled source

A source given up for not answering our sync messages could become the
source again straight away.  Whoever asks first after the reset takes the
role, and that's often the same peer, which would then stall for another
delivery limit.

Now, a stalled source isn't chosen again for one more delivery_limit.
If it's the only candidate, catching up waits that long and then tries
it again.  Before, a retry with that peer couldn't have helped any
sooner anyway: a source is only given up after a whole delivery limit
without a sync message, and that clock restarts when it is chosen again.

3 files changed, 28 insertions, 2 deletions

diff --git a/src/node.rs b/src/node.rs
index 8a49c56..57d961c 100644
--- a/src/node.rs
+++ b/src/node.rs
@@ -113,6 +113,7 @@ impl Node {
             stale: Default::default(),
             catch_up_source: Default::default(),
             catch_up_preference: Default::default(),
+            catch_up_avoided: Default::default(),
         };
         for actor in config.earlier_actors {
             state.remake.add_own_actor(actor);
diff --git a/src/state.rs b/src/state.rs
index 5ec1c5a..78b058c 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -673,6 +673,9 @@ pub(crate) struct AppState {
     pub(crate) catch_up_source: Arc<std::sync::Mutex<Option<CatchUpSource>>>,
     /// Since when a closer peer has had a head start to become the source.
     pub(crate) catch_up_preference: Arc<std::sync::Mutex<Option<Instant>>>,
+    /// A catch-up source given up for not answering our sync messages, and until when it
+    /// isn't chosen again (`swarm::source`).
+    pub(crate) catch_up_avoided: Arc<std::sync::Mutex<Option<(String, Instant)>>>,
 }
 
 /// The peer a catching-up node syncs with first (`swarm::source`).
diff --git a/src/swarm/source.rs b/src/swarm/source.rs
index d5b572a..9eac445 100644
--- a/src/swarm/source.rs
+++ b/src/swarm/source.rs
@@ -21,7 +21,9 @@
 //! 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. Meanwhile, every minute without a sync message from it is logged as a
+//! 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;
@@ -56,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)
 }
@@ -74,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,
@@ -171,6 +188,10 @@ pub(crate) fn reply_timed_out(state: &AppState, peer: &str) {
         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))
     };
@@ -273,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 {