unreachability doesn't expire

5 files changed, 70 insertions, 19 deletions

diff --git a/src/simtest/mod.rs b/src/simtest/mod.rs
index e7fc719..3202ae5 100644
--- a/src/simtest/mod.rs
+++ b/src/simtest/mod.rs
@@ -218,7 +218,7 @@ impl Sim {
                     p.epoch,
                     ago(p.last_seen),
                     ago(p.last_sync_rx),
-                    p.is_unreachable(&node.timing, now),
+                    p.is_unreachable(),
                     p.sync.automerge_view().0.map(|heads| heads
                         .iter()
                         .map(|head| head.to_string()[..8].to_owned())
diff --git a/src/state/peer.rs b/src/state/peer.rs
index 2ac6b1a..944b886 100644
--- a/src/state/peer.rs
+++ b/src/state/peer.rs
@@ -53,7 +53,8 @@ pub(crate) enum HelloState {
     /// A send is in flight.
     Sending,
     /// HELLO was sent (or the send failed); no HELLO_REPLY yet. Retried at
-    /// `next_attempt` (`PeerState::hello_backoff`, or `dormant_probe` while dormant).
+    /// `next_attempt` (`PeerState::hello_backoff`, or `PeerState::probe_backoff` while
+    /// unreachable).
     AwaitingReply { next_attempt: Instant },
     /// HELLO_REPLY received. No more retries.
     Established,
@@ -74,6 +75,9 @@ pub(crate) struct PeerState {
     pub(crate) hello_state: HelloState,
     /// Wait between unanswered HELLOs; reset whenever a new handshake starts.
     pub(crate) hello_backoff: Backoff,
+    /// Wait between HELLOs to an unreachable peer (probes): `dormant_probe`, doubling
+    /// while none is answered. Reset by any message from it.
+    pub(crate) probe_backoff: Backoff,
 
     /// Everything about the current Automerge sync with this peer; a reset replaces it.
     pub(crate) sync: SyncSession,
@@ -120,8 +124,8 @@ pub(crate) struct PeerState {
     /// HELLOs sent in the current handshake attempt without hearing anything back.
     pub(crate) hello_attempts: u32,
     /// Since when we consider the peer unreachable (three unanswered HELLOs, or
-    /// dormant). Cleared by any message from it; ignored after `dormant_probe`, so the
-    /// peer becomes a neighbour candidate again and gets tried once more.
+    /// dormant). Only a message from it clears it (`mark_seen`): until then it is no
+    /// neighbour candidate, and a member gets probes (`probe_backoff`) instead.
     pub(crate) unreachable_since: Option<Instant>,
 
     /// What went wrong with this peer so far (counted only, for /peers and /admin/stats).
@@ -163,6 +167,7 @@ impl PeerState {
             standing: Standing::default(),
             hello_state: HelloState::default(),
             hello_backoff: Backoff::new(HELLO_RETRY, HELLO_RETRY_MAX),
+            probe_backoff: timing.probe_backoff(),
             sync: SyncSession::new(timing.sync_reply_backoff()),
             liveness: Liveness::Active,
             added_at: Instant::now(),
@@ -216,6 +221,7 @@ impl PeerState {
         self.last_seen = Some(now);
         self.hello_attempts = 0;
         self.unreachable_since = None;
+        self.probe_backoff.reset();
 
         if let Liveness::Dormant { since } = self.liveness {
             self.incidents.dormant_ms += now.saturating_duration_since(since).as_millis() as u64;
@@ -239,15 +245,25 @@ impl PeerState {
     }
 
     /// Whether to leave the peer out when choosing neighbours (`swarm::topology`).
-    pub(crate) fn is_unreachable(&self, timing: &Timing, now: Instant) -> bool {
-        self.unreachable_since
-            .is_some_and(|since| now.saturating_duration_since(since) < timing.dormant_probe)
+    pub(crate) fn is_unreachable(&self) -> bool {
+        self.unreachable_since.is_some()
+    }
+
+    /// The wait before the next HELLO: a probe's if unreachable.
+    pub(crate) fn next_hello_wait(&mut self) -> Duration {
+        if self.is_unreachable() {
+            self.probe_backoff.next()
+        } else {
+            self.hello_backoff.next()
+        }
     }
 
     /// No more HELLO retries (the peer isn't our neighbour any more). A HELLO in flight
-    /// finishes; a later handshake starts over.
+    /// finishes; a later handshake starts over. An unreachable member keeps its probes:
+    /// it is no neighbour because it is unreachable, and only its answer can change that.
     pub(crate) fn stop_hello_retries(&mut self) {
-        if matches!(self.hello_state, HelloState::AwaitingReply { .. }) {
+        let probed = self.is_unreachable() && matches!(self.standing, Standing::Member { .. });
+        if !probed && matches!(self.hello_state, HelloState::AwaitingReply { .. }) {
             self.hello_state = HelloState::NotSent;
         }
     }
@@ -265,7 +281,7 @@ impl PeerState {
 
         if self.hello_state != HelloState::Sending {
             self.hello_state = HelloState::AwaitingReply {
-                next_attempt: now + timing.dormant_probe,
+                next_attempt: now + self.probe_backoff.next(),
             };
         }
         true
@@ -357,3 +373,34 @@ impl PeerState {
                 .is_some_and(|t| now.saturating_duration_since(t) < timing.dormant_after)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// R41: unreachability lasts until the peer is heard from, not for a set time, and
+    /// probes back off meanwhile.
+    #[test]
+    fn unreachable_until_heard_from() {
+        let timing = Timing::from_keepalive(Duration::from_secs(60));
+        let mut peer = PeerState::new(true, &timing);
+        let start = Instant::now();
+        for _ in 0..UNREACHABLE_AFTER_HELLOS {
+            peer.hello_sent(start);
+        }
+        assert!(peer.is_unreachable());
+
+        let probes: Vec<Duration> = (0..4).map(|_| peer.next_hello_wait()).collect();
+        let probe = timing.dormant_probe;
+        assert_eq!(probes, [probe, probe * 2, probe * 4, probe * 6]);
+        peer.hello_sent(start + probe * 20);
+        assert!(peer.is_unreachable());
+
+        peer.mark_seen(start + probe * 21);
+        assert!(!peer.is_unreachable());
+        peer.hello_sent(start + probe * 22);
+        peer.hello_sent(start + probe * 22);
+        peer.hello_sent(start + probe * 22);
+        assert_eq!(peer.next_hello_wait(), probe);
+    }
+}
diff --git a/src/state/timing.rs b/src/state/timing.rs
index 406e207..c2993d7 100644
--- a/src/state/timing.rs
+++ b/src/state/timing.rs
@@ -10,6 +10,10 @@ const SYNC_REPLY_TIMEOUT: Duration = Duration::from_secs(10);
 /// Upper bound, as a multiple of the first wait: 5 minutes (Mycelium gives up on a
 /// message after five minutes).
 const SYNC_REPLY_TIMEOUT_MAX_FACTOR: u32 = 30;
+/// Upper bound on the wait between probes of an unreachable peer, as a multiple of
+/// `dormant_probe`: an hour at the default keepalive (1m). A returning node HELLOs its own
+/// neighbours, so this only bounds how long one we never hear from stays out.
+const PROBE_MAX_FACTOR: u32 = 6;
 /// How long Mycelium tries to deliver a message before dropping it.
 const MYCELIUM_DELIVERY_LIMIT: Duration = Duration::from_secs(300);
 
@@ -65,6 +69,12 @@ impl Timing {
         }
     }
 
+    /// Waits between probes of an unreachable peer: `dormant_probe`, doubling up to
+    /// `PROBE_MAX_FACTOR` times as much.
+    pub(crate) fn probe_backoff(&self) -> Backoff {
+        Backoff::new(self.dormant_probe, self.dormant_probe * PROBE_MAX_FACTOR)
+    }
+
     /// Waits for sync answers: `sync_reply_timeout`, doubling up to 30 times as much.
     pub(crate) fn sync_reply_backoff(&self) -> Backoff {
         Backoff::new(
diff --git a/src/swarm/peer/handshake.rs b/src/swarm/peer/handshake.rs
index f89e371..f08929a 100644
--- a/src/swarm/peer/handshake.rs
+++ b/src/swarm/peer/handshake.rs
@@ -150,10 +150,7 @@ async fn transmit_hello(state: &AppState, peer: &str) -> Result<()> {
             if !was_unreachable && peer_state.unreachable_since.is_some() {
                 state.topology_changed.notify_one();
             }
-            let wait = match peer_state.liveness {
-                Liveness::Dormant { .. } => state.timing.dormant_probe,
-                Liveness::Active => peer_state.hello_backoff.next(),
-            };
+            let wait = peer_state.next_hello_wait();
             peer_state.hello_state = HelloState::AwaitingReply {
                 next_attempt: Instant::now() + wait,
             };
diff --git a/src/swarm/topology.rs b/src/swarm/topology.rs
index e396a9d..84ae1cd 100644
--- a/src/swarm/topology.rs
+++ b/src/swarm/topology.rs
@@ -20,7 +20,6 @@
 use std::collections::{BTreeMap, BTreeSet};
 
 use sha2::{Digest, Sha256};
-use tokio::time::Instant;
 use tracing::info;
 
 use crate::locks::RwLockExt;
@@ -111,8 +110,7 @@ pub(crate) struct NeighbourChange {
 /// Recomputes the neighbours whenever something they depend on may have changed
 /// (`AppState::topology_changed`): the member list (`swarm::sync::run_change_publisher`),
 /// a peer becoming unreachable or reachable again (`swarm::peer`), and every 5 s as a
-/// fallback, which also covers unreachability expiring after `dormant_probe`. A burst of
-/// notifications coalesces into one recomputation.
+/// fallback. A burst of notifications coalesces into one recomputation.
 pub(crate) async fn run_worker(state: AppState) {
     loop {
         state.topology_changed.notified().await;
@@ -145,11 +143,10 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
         (current, stable)
     };
     let k = state.neighbours_k;
-    let now = Instant::now();
     let mut unreachable = BTreeSet::new();
     for member in &current {
         if let Some(peer_state) = state.get_peer_state(member).await {
-            if peer_state.lock().await.is_unreachable(&state.timing, now) {
+            if peer_state.lock().await.is_unreachable() {
                 unreachable.insert(member.clone());
             }
         }