overdue sync asked after from then

3 files changed, 92 insertions, 49 deletions

diff --git a/src/state/peer.rs b/src/state/peer.rs
index 434f5fd..0d666e6 100644
--- a/src/state/peer.rs
+++ b/src/state/peer.rs
@@ -108,12 +108,10 @@ 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`), 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)>,
+    /// What we wait for from it with nothing heard from it since: an answer to a
+    /// keepalive, or to a sync message whose answer is overdue (`AwaitingAnswer`).
+    /// Cleared by any message from it (`mark_seen`).
+    awaiting_answer: Option<AwaitingAnswer>,
 
     /// Last time a sync message from this peer was applied (survives resets: the
     /// keepalive heal looks at it).
@@ -180,6 +178,39 @@ pub(crate) struct PeerState {
     whole_failed: Option<(usize, Instant)>,
 }
 
+/// An answer we wait for (`PeerState::awaiting_answer`): unanswered for half of
+/// `Timing::unreached_after`, the peer is asked again with a PING
+/// (`take_ask_again_due`), so one lost message or answer doesn't do it; unanswered for
+/// all of it, we don't count it as reached (`swarm::reach`). A sync message whose answer
+/// is overdue is asked after at once: by then it waited that half or longer (the waits
+/// for sync answers grow while none comes), and a peer with nothing to say answers no
+/// sync message, however often it goes.
+#[derive(Clone, Copy, Debug)]
+struct AwaitingAnswer {
+    /// When to ask again; None once asked.
+    ask_again_at: Option<Instant>,
+    /// When it no longer counts as reached.
+    unreached_at: Instant,
+}
+
+impl AwaitingAnswer {
+    /// Waiting from `now` on, for what was just sent.
+    fn sent(timing: &Timing, now: Instant) -> Self {
+        Self {
+            ask_again_at: Some(now + timing.unreached_after / 2),
+            unreached_at: now + timing.unreached_after,
+        }
+    }
+
+    /// Waiting for an overdue sync answer: asked again now.
+    fn overdue(timing: &Timing, now: Instant) -> Self {
+        Self {
+            ask_again_at: Some(now),
+            unreached_at: now + timing.unreached_after / 2,
+        }
+    }
+}
+
 /// How long a sync message that didn't make it keeps others that large from going whole.
 const WHOLE_FAILED_KEPT: Duration = Duration::from_secs(3600);
 
@@ -272,20 +303,20 @@ impl PeerState {
 
     /// An overdue sync answer from an established, active peer (see
     /// `SyncSession::take_overdue_reply`).
-    pub(crate) fn take_overdue_sync_reply(&mut self, now: Instant) -> Option<usize> {
+    pub(crate) fn take_overdue_sync_reply(
+        &mut self,
+        timing: &Timing,
+        now: Instant,
+    ) -> Option<usize> {
         if self.liveness.is_dormant() || self.hello_state != HelloState::Established {
             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 {
-                // Asked again with a PING too: a peer that has nothing to say answers no
-                // sync message, however often it goes.
-                self.awaiting_answer.get_or_insert((started, false));
-            }
+            self.awaiting_answer
+                .get_or_insert(AwaitingAnswer::overdue(timing, now));
             if let Some(largest) = largest {
                 self.whole_failed(largest, now);
             }
@@ -356,26 +387,23 @@ 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
+    /// Whether what we sent it went unanswered, with nothing else heard from it, even
+    /// once asked again (`AwaitingAnswer`): 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
-        })
+    pub(crate) fn answer_overdue(&self, now: Instant) -> bool {
+        self.awaiting_answer
+            .is_some_and(|awaiting| now >= awaiting.unreached_at)
     }
 
     /// 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 {
+    /// us for a while (`AwaitingAnswer`); records it.
+    pub(crate) fn take_ask_again_due(&mut self, 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;
+            Some(awaiting) if awaiting.ask_again_at.is_some_and(|at| now >= at) => {
+                awaiting.ask_again_at = None;
                 true
             }
             _ => false,
@@ -492,7 +520,8 @@ impl PeerState {
             return None;
         }
         self.last_keepalive_at = Some(now);
-        self.awaiting_answer.get_or_insert((now, false));
+        self.awaiting_answer
+            .get_or_insert(AwaitingAnswer::sent(timing, now));
         let status = self
             .last_status_at
             .is_none_or(|t| now.saturating_duration_since(t) >= timing.status_refresh);
@@ -559,13 +588,7 @@ impl PeerState {
             .flatten();
         let unreached = self
             .awaiting_answer
-            .map(|(since, asked_again)| {
-                if asked_again {
-                    since + timing.unreached_after
-                } else {
-                    since + timing.unreached_after / 2
-                }
-            })
+            .map(|awaiting| awaiting.ask_again_at.unwrap_or(awaiting.unreached_at))
             // Once past, nothing more is due for it.
             .filter(|at| !self.liveness.is_dormant() && *at > Instant::now());
         [dormancy, keepalive, hello, reply, unreached]
@@ -737,25 +760,26 @@ mod tests {
         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));
+        assert!(!peer.take_ask_again_due(half - Duration::from_millis(1)));
+        assert!(peer.take_ask_again_due(half));
+        assert!(!peer.take_ask_again_due(half), "asked again once");
+        assert!(!peer.answer_overdue(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));
+        assert!(!peer.answer_overdue(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));
+        assert!(peer.take_ask_again_due(sent + timing.unreached_after / 2));
+        assert!(peer.answer_overdue(sent + timing.unreached_after));
         peer.mark_seen(sent + timing.unreached_after * 2);
-        assert!(!peer.answer_overdue(&timing, sent + timing.unreached_after * 2));
+        assert!(!peer.answer_overdue(sent + timing.unreached_after * 2));
     }
 
     /// A sync message whose answer is overdue is asked after with a PING too: a peer with
     /// nothing to say answers none, however often it goes, and still counts as reached
-    /// once it answers the PING.
+    /// once it answers the PING. Also when the wait for the sync answer grew longer than
+    /// `unreached_after`: the PING gets its own time.
     #[test]
     fn unanswered_sync_messages_are_asked_after() {
         let timing = Timing::from_keepalive(Duration::from_secs(60), Duration::from_secs(300));
@@ -769,10 +793,29 @@ mod tests {
             100,
         );
         let overdue = sent + timing.sync_reply_timeout;
-        assert!(peer.take_overdue_sync_reply(overdue).is_some());
-        assert!(peer.take_ask_again_due(&timing, overdue));
+        assert!(peer.take_overdue_sync_reply(&timing, overdue).is_some());
+        assert!(peer.take_ask_again_due(overdue));
         peer.mark_seen(overdue + Duration::from_millis(50));
-        assert!(!peer.answer_overdue(&timing, sent + timing.unreached_after));
+        assert!(!peer.answer_overdue(sent + timing.unreached_after));
+        // The next one waits twice as long for its answer, all of `unreached_after`.
+        let sent = overdue + Duration::from_secs(1);
+        peer.sync.expect_reply(
+            Vec::new(),
+            sent,
+            crate::swarm::throughput::SyncBytes::default(),
+            100,
+        );
+        let overdue = sent + timing.sync_reply_timeout * 2;
+        assert!(peer.take_overdue_sync_reply(&timing, overdue).is_some());
+        assert!(
+            !peer.answer_overdue(overdue),
+            "not before the PING had its time"
+        );
+        assert!(peer.take_ask_again_due(overdue));
+        assert!(
+            !peer.answer_overdue(overdue + timing.unreached_after / 2 - Duration::from_millis(1))
+        );
+        assert!(peer.answer_overdue(overdue + timing.unreached_after / 2));
     }
 
     /// Keepalives are PINGs, every `status_refresh` a STATUS; only when quiet.
diff --git a/src/swarm/peer/worker.rs b/src/swarm/peer/worker.rs
index dad0870..67213e7 100644
--- a/src/swarm/peer/worker.rs
+++ b/src/swarm/peer/worker.rs
@@ -104,9 +104,9 @@ pub(crate) async fn run_peer_worker(state: AppState) {
                 if !neighbour {
                     peer_state.stop_hello_retries(topology::is_listed(&state, peer));
                 }
-                let unanswered = peer_state.take_overdue_sync_reply(now);
-                let ask_again = peer_state.take_ask_again_due(&state.timing, now);
-                reach_changed |= peer_state.answer_overdue(&state.timing, now)
+                let unanswered = peer_state.take_overdue_sync_reply(&state.timing, now);
+                let ask_again = peer_state.take_ask_again_due(now);
+                reach_changed |= peer_state.answer_overdue(now)
                     && crate::swarm::reach::reached_stops_answering(&state, peer);
                 (became_dormant, keepalive, unanswered, ask_again)
             };
diff --git a/src/swarm/reach.rs b/src/swarm/reach.rs
index 5c89d04..8c22ef6 100644
--- a/src/swarm/reach.rs
+++ b/src/swarm/reach.rs
@@ -50,7 +50,7 @@ async fn measure(state: &AppState) -> Vec<u64> {
         if peer_state.hello_state == HelloState::Established
             && !peer_state.liveness.is_dormant()
             && !peer_state.is_unreachable()
-            && !peer_state.answer_overdue(&state.timing, now)
+            && !peer_state.answer_overdue(now)
             && matches!(peer_state.standing, Standing::Member { .. })
         {
             reached.push(topology::pair_key(&peer));