1 files changed, 82 insertions, 3 deletions

diff --git a/src/state/peer.rs b/src/state/peer.rs
index ca721b5..a1d40ab 100644
--- a/src/state/peer.rs
+++ b/src/state/peer.rs
@@ -108,6 +108,13 @@ pub(crate) struct PeerState {
     pub(crate) keepalive_received_at: Option<Instant>,
     /// The last STATUS between us, either way (`status_refresh`).
     pub(crate) last_status_at: Option<Instant>,
+    /// Since when we wait for an answer from it (to a keepalive, or to a sync message
+    /// whose answer is overdue) with nothing heard from it since, and whether we asked
+    /// again: halfway to `Timing::unreached_after` a PING (`take_ask_again_due`; a sync
+    /// message whose answer is overdue goes again anyway), so one lost message or answer
+    /// doesn't do it; after it, we don't count it as reached (`swarm::reach`). Cleared by
+    /// any message from it (`mark_seen`).
+    awaiting_answer: Option<(Instant, bool)>,
 
     /// Last time a sync message from this peer was applied (survives resets: the
     /// keepalive heal looks at it).
@@ -230,6 +237,7 @@ impl PeerState {
             last_keepalive_at: None,
             keepalive_received_at: None,
             last_status_at: None,
+            awaiting_answer: None,
             last_sync_rx: None,
             last_resync_at: None,
             diverged_since: None,
@@ -270,9 +278,14 @@ impl PeerState {
             return None;
         }
         let largest = self.sync.reply.as_ref().map(|wait| wait.largest);
+        let started = self.sync.reply.as_ref().map(|wait| wait.started.0);
         let overdue = self.sync.take_overdue_reply(now);
         if overdue.is_some() {
             self.incidents.reply_timeouts += 1;
+            if let Some(started) = started {
+                // It goes again now: that is asking again.
+                self.awaiting_answer.get_or_insert((started, true));
+            }
             if let Some(largest) = largest {
                 self.whole_failed(largest, now);
             }
@@ -303,6 +316,7 @@ impl PeerState {
     /// said farewell (`left`), only that it was heard from.
     pub(crate) fn mark_seen(&mut self, now: Instant) -> bool {
         self.last_seen = Some(now);
+        self.awaiting_answer = None;
         if self.left {
             return false;
         }
@@ -342,6 +356,32 @@ impl PeerState {
             && self.epoch == Some(epoch)
     }
 
+    /// Whether what we sent it went unanswered, with nothing else heard from it, for
+    /// `Timing::unreached_after` (`awaiting_answer`): then we don't count it as reached
+    /// (`swarm::reach`), long before it would be dormant.
+    pub(crate) fn answer_overdue(&self, timing: &Timing, now: Instant) -> bool {
+        self.awaiting_answer.is_some_and(|(since, _)| {
+            now.saturating_duration_since(since) >= timing.unreached_after
+        })
+    }
+
+    /// Whether to ask an established, active peer again, with a PING, as nothing answered
+    /// us for half of `Timing::unreached_after` (`awaiting_answer`); records it.
+    pub(crate) fn take_ask_again_due(&mut self, timing: &Timing, now: Instant) -> bool {
+        if self.liveness.is_dormant() || self.hello_state != HelloState::Established {
+            return false;
+        }
+        match &mut self.awaiting_answer {
+            Some((since, asked_again))
+                if !*asked_again && now >= *since + timing.unreached_after / 2 =>
+            {
+                *asked_again = true;
+                true
+            }
+            _ => false,
+        }
+    }
+
     /// Whether to leave the peer out when choosing neighbours (`swarm::topology`).
     pub(crate) fn is_unreachable(&self) -> bool {
         self.unreachable_since.is_some()
@@ -452,6 +492,7 @@ impl PeerState {
             return None;
         }
         self.last_keepalive_at = Some(now);
+        self.awaiting_answer.get_or_insert((now, false));
         let status = self
             .last_status_at
             .is_none_or(|t| now.saturating_duration_since(t) >= timing.status_refresh);
@@ -493,8 +534,9 @@ impl PeerState {
 
     /// When the peer worker next has something to do for this peer
     /// (`swarm::peer::run_peer_worker`): a keepalive (to a `neighbour`), dormancy (of a
-    /// neighbour, once `ready`), a HELLO retry, an overdue sync answer. None if nothing
-    /// is due until something changes.
+    /// neighbour, once `ready`), a HELLO retry, an overdue sync answer, asking again for
+    /// an answer and its being overdue (`take_ask_again_due`, `answer_overdue`). None if nothing is due until something
+    /// changes.
     pub(crate) fn next_due(
         &self,
         timing: &Timing,
@@ -515,7 +557,18 @@ impl PeerState {
         let reply = (!self.liveness.is_dormant() && self.hello_state == HelloState::Established)
             .then(|| self.sync.reply.as_ref().map(|wait| wait.deadline))
             .flatten();
-        [dormancy, keepalive, hello, reply]
+        let unreached = self
+            .awaiting_answer
+            .map(|(since, asked_again)| {
+                if asked_again {
+                    since + timing.unreached_after
+                } else {
+                    since + timing.unreached_after / 2
+                }
+            })
+            // Once past, nothing more is due for it.
+            .filter(|at| !self.liveness.is_dormant() && *at > Instant::now());
+        [dormancy, keepalive, hello, reply, unreached]
             .into_iter()
             .flatten()
             .min()
@@ -674,6 +727,32 @@ mod tests {
         assert!(b.take_keepalive_due(&timing, next).is_none());
     }
 
+    /// A keepalive nothing answers: asked again halfway, after `unreached_after` not
+    /// reached; any message from the peer, reached again. One lost answer doesn't do it.
+    #[test]
+    fn unanswered_keepalives_stop_counting_as_reached() {
+        let timing = Timing::from_keepalive(Duration::from_secs(60), Duration::from_secs(300));
+        let mut peer = PeerState::new(true, &timing);
+        peer.hello_state = HelloState::Established;
+        let sent = peer.added_at + timing.keepalive;
+        assert!(peer.take_keepalive_due(&timing, sent).is_some());
+        let half = sent + timing.unreached_after / 2;
+        assert!(!peer.take_ask_again_due(&timing, half - Duration::from_millis(1)));
+        assert!(peer.take_ask_again_due(&timing, half));
+        assert!(!peer.take_ask_again_due(&timing, half), "asked again once");
+        assert!(!peer.answer_overdue(&timing, half));
+        // The second ask was answered: reached all along.
+        peer.mark_seen(half + Duration::from_millis(50));
+        assert!(!peer.answer_overdue(&timing, sent + timing.unreached_after));
+        // The next keepalive, and its second ask, go unanswered.
+        let sent = half + timing.keepalive * 2;
+        assert!(peer.take_keepalive_due(&timing, sent).is_some());
+        assert!(peer.take_ask_again_due(&timing, sent + timing.unreached_after / 2));
+        assert!(peer.answer_overdue(&timing, sent + timing.unreached_after));
+        peer.mark_seen(sent + timing.unreached_after * 2);
+        assert!(!peer.answer_overdue(&timing, sent + timing.unreached_after * 2));
+    }
+
     /// Keepalives are PINGs, every `status_refresh` a STATUS; only when quiet.
     #[test]
     fn keepalives_are_pings_and_now_and_then_a_status() {