unanswered members not reached

1 files changed, 25 insertions, 7 deletions

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<()> {