unanswered members not reached

1 files changed, 18 insertions, 7 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;