From 2a85e21ab81d4497dcc5680b4348bbf802866944 Mon Sep 17 00:00:00 2001 From: tv Date: Tue, 3 Nov 2026 08:34:09 +0100 Subject: unanswered members not reached --- src/state/peer.rs | 85 +++++++++++++++++++++++++++++++++++++++++++++++++++-- src/state/timing.rs | 6 ++++ 2 files changed, 88 insertions(+), 3 deletions(-) (limited to 'src/state') 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, /// 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`; 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() { diff --git a/src/state/timing.rs b/src/state/timing.rs index af5c0c0..fed48e9 100644 --- a/src/state/timing.rs +++ b/src/state/timing.rs @@ -35,6 +35,11 @@ pub(crate) struct Timing { /// First wait for an answer to a sync message before sending again; doubles while /// answers stay out (`SyncSession::reply_backoff`). pub(crate) sync_reply_timeout: Duration, + /// How long what we sent a peer may go unanswered, with nothing else heard from it, + /// before we don't count it as reached (`swarm::reach`, `PeerState::answer_overdue`): + /// twice the first sync reply timeout, so one lost message (or its answer) doesn't + /// do it. + pub(crate) unreached_after: Duration, /// How long the transport tries to deliver a message before dropping it /// (`Transport::delivery_limit`): a peer silent for longer has nothing on the way to /// us any more. @@ -93,6 +98,7 @@ impl Timing { // Only the source gets these, and only while catching up. catch_up_probe: (keepalive / 30).clamp(Duration::from_secs(1), Duration::from_secs(5)), sync_reply_timeout, + unreached_after: sync_reply_timeout * 2, delivery_limit, } } -- cgit v1.3.1