From 0c0066780fefb96b49c1b51efb63b6cfa37aee12 Mon Sep 17 00:00:00 2001 From: tv Date: Wed, 4 Nov 2026 05:34:55 +0100 Subject: overdue sync asked after from then --- src/state/peer.rs | 133 +++++++++++++++++++++++++++++++---------------- src/swarm/peer/worker.rs | 6 +-- src/swarm/reach.rs | 2 +- 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, /// The last STATUS between us, either way (`status_refresh`). pub(crate) last_status_at: Option, - /// 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, /// 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, + /// 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 { + pub(crate) fn take_overdue_sync_reply( + &mut self, + timing: &Timing, + now: Instant, + ) -> Option { 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 { 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)); -- cgit v1.3.1