unanswered members not reached

2 files changed, 43 insertions, 14 deletions

diff --git a/src/swarm/peer/worker.rs b/src/swarm/peer/worker.rs
index 45e8d8f..dad0870 100644
--- a/src/swarm/peer/worker.rs
+++ b/src/swarm/peer/worker.rs
@@ -1,8 +1,9 @@
-//! The peer worker: keepalives due, dormancy, HELLO retries and overdue sync replies,
-//! when the earliest of them is due (or something may have made one due sooner,
-//! `Due`); every `FORGET_CHECK_INTERVAL` (and at once when the node becomes ready),
-//! forget records, the member registry, stability, measurements and the state file;
-//! forget records also at once when they changed.
+//! The peer worker: keepalives due, dormancy, HELLO retries, overdue sync replies and
+//! answers (`PeerState::answer_overdue`, `swarm::reach`), when the earliest of them is
+//! due (or something may have made one due sooner, `Due`); every
+//! `FORGET_CHECK_INTERVAL` (and at once when the node becomes ready), forget records,
+//! the member registry, stability, measurements and the state file; forget records
+//! also at once when they changed.
 
 use std::time::Duration;
 
@@ -80,13 +81,15 @@ pub(crate) async fn run_peer_worker(state: AppState) {
         was_ready = ready;
 
         let peers = state.peers_snapshot().await;
+        // A member we said we reach stopped answering: tell our neighbours now.
+        let mut reach_changed = false;
 
         for (peer, peer_state_mutex) in &peers {
             let now = Instant::now();
             // Only neighbours are kept alive and judged; others we only answer.
             let neighbour = topology::is_neighbour(&state, peer);
 
-            let (became_dormant, keepalive, unanswered) = {
+            let (became_dormant, keepalive, unanswered, ask_again) = {
                 let mut peer_state = peer_state_mutex.lock().await;
                 let became_dormant = ready
                     && neighbour
@@ -102,7 +105,10 @@ pub(crate) async fn run_peer_worker(state: AppState) {
                     peer_state.stop_hello_retries(topology::is_listed(&state, peer));
                 }
                 let unanswered = peer_state.take_overdue_sync_reply(now);
-                (became_dormant, keepalive, unanswered)
+                let ask_again = peer_state.take_ask_again_due(&state.timing, now);
+                reach_changed |= peer_state.answer_overdue(&state.timing, now)
+                    && crate::swarm::reach::reached_stops_answering(&state, peer);
+                (became_dormant, keepalive, unanswered, ask_again)
             };
 
             if let Some(changes) = unanswered {
@@ -128,6 +134,8 @@ pub(crate) async fn run_peer_worker(state: AppState) {
                 }
                 keepalive => keepalive,
             };
+            // Nothing answered us for a while: once more before it counts as not reached.
+            let keepalive = keepalive.or(ask_again.then_some(Keepalive::Ping));
             let sent = match keepalive {
                 Some(Keepalive::Ping) => send_ping(&state, peer).await,
                 Some(Keepalive::Status) => send_status(&state, peer).await,
@@ -160,6 +168,9 @@ pub(crate) async fn run_peer_worker(state: AppState) {
         if records_changed || periodic {
             apply_forget_records(&state, &mut forgets_applied).await;
         }
+        if reach_changed && !periodic {
+            crate::swarm::reach::check(&state).await;
+        }
         if periodic {
             forget_strangers(&state).await;
             crate::swarm::members::ensure_registered(&state).await;
diff --git a/src/swarm/reach.rs b/src/swarm/reach.rs
index 901b47c..7bd0f73 100644
--- a/src/swarm/reach.rs
+++ b/src/swarm/reach.rs
@@ -2,13 +2,16 @@
 //! member list says.
 //!
 //! A node knows its own links, and from them whom it reaches: the members it has an
-//! active session with (handshake done, neither dormant nor unreachable). It tells its
+//! active session with (handshake done, neither dormant nor unreachable) that answer
+//! (`PeerState::answer_overdue`: none left a keepalive or sync message unanswered for
+//! `Timing::unreached_after`, asked again halfway, with nothing else heard from it; a
+//! member gone shows within seconds, not after `dormant_after`). It tells its
 //! neighbours so in its STATUS (`Status::reaches`, each member by its pair key,
 //! `topology::pair_key`, 8 bytes instead of a node ID): in every keepalive STATUS and
-//! STATUS_REPLY, and soon (the peer worker looks every 5 s) to every neighbour when it
-//! changed,
-//! but not in announcements of heads, which go often. So every node knows whom its
-//! neighbours reach, as of their latest STATUS, for as long as the session lasts (a
+//! STATUS_REPLY, and soon to every neighbour when it changed (the peer worker looks
+//! every 5 s, and at once when a member we reach stops answering), but not in
+//! announcements of heads, which go often. So every node knows whom its neighbours
+//! reach, as of their latest STATUS, for as long as the session lasts (a
 //! neighbour that goes dormant, or starts anew, says it again). Not for a set time:
 //! while a pair exchanges changes it sends no keepalives, and so no STATUS, for as
 //! long as the writes go on. Nothing goes into the document, and nobody knows more than
@@ -37,14 +40,17 @@ pub(crate) struct Reach {
     published: Option<Vec<u64>>,
 }
 
-/// The pair keys of the members we reach now, sorted: those with an active session.
+/// The pair keys of the members we reach now, sorted: those with an active session
+/// that answer.
 async fn measure(state: &AppState) -> Vec<u64> {
+    let now = Instant::now();
     let mut reached = Vec::new();
     for (peer, peer_state) in state.peers_snapshot().await {
         let peer_state = peer_state.lock().await;
         if peer_state.hello_state == HelloState::Established
             && !peer_state.liveness.is_dormant()
             && !peer_state.is_unreachable()
+            && !peer_state.answer_overdue(&state.timing, now)
             && matches!(peer_state.standing, Standing::Member { .. })
         {
             reached.push(topology::pair_key(&peer));
@@ -64,7 +70,8 @@ pub(crate) fn published(state: &AppState) -> Vec<u64> {
         .unwrap_or_default()
 }
 
-/// With the peer worker's periodic pass (every 5 s): whom we reach now; if it changed,
+/// With the peer worker's periodic pass (every 5 s), and at once when a member we said
+/// we reach stops answering (`reached_stops_answering`): whom we reach now; if it changed,
 /// our neighbours are told at once (a STATUS each, unanswered if the heads are the
 /// same). A change that is undone within a pass isn't told.
 pub(crate) async fn check(state: &AppState) {
@@ -91,6 +98,17 @@ pub(crate) async fn check(state: &AppState) {
     }
 }
 
+/// Whether `peer`, whose answer is overdue (`PeerState::answer_overdue`), is one we
+/// said we reach: then the peer worker checks at once (`check`).
+pub(crate) fn reached_stops_answering(state: &AppState, peer: &str) -> bool {
+    state
+        .reach
+        .lock_or_recover()
+        .published
+        .as_ref()
+        .is_some_and(|keys| keys.contains(&topology::pair_key(peer)))
+}
+
 /// A STATUS to `peer` with whom we reach: as an announcement (unanswered if the heads
 /// are the same), but with `reaches`.
 async fn tell(state: &AppState, peer: &str) -> anyhow::Result<()> {