From b88be3ea1b475b1a656298b98708026388eaf2af Mon Sep 17 00:00:00 2001 From: tv Date: Wed, 28 Oct 2026 02:54:55 +0100 Subject: move keepalive and firewall to paths --- src/state/host.rs | 26 +++++++++++----- src/state/mod.rs | 2 +- src/state/peer.rs | 90 ++++++++++++++++++++++++++++++++++++++++++----------- src/state/timing.rs | 14 +++++++-- 4 files changed, 102 insertions(+), 30 deletions(-) (limited to 'src/state') diff --git a/src/state/host.rs b/src/state/host.rs index 441139e..6eab803 100644 --- a/src/state/host.rs +++ b/src/state/host.rs @@ -161,14 +161,24 @@ impl Host for AppState { fn resume(&self) -> impl Future + Send { let state = self.clone(); async move { - let mut lifecycle = state.lifecycle.write().await; - if let NodeLifecycle::Joining { - discovered_swarm_id: Some(swarm_id), - } = &*lifecycle - { - let swarm_id = swarm_id.clone(); - debug!(swarm_id = %swarm_id, "Node resumed"); - *lifecycle = NodeLifecycle::Ready { swarm_id }; + let resumed = { + let mut lifecycle = state.lifecycle.write().await; + let swarm_id = match &*lifecycle { + NodeLifecycle::Joining { + discovered_swarm_id: Some(swarm_id), + } => Some(swarm_id.clone()), + _ => None, + }; + if let Some(swarm_id) = &swarm_id { + debug!(swarm_id = %swarm_id, "Node resumed"); + *lifecycle = NodeLifecycle::Ready { + swarm_id: swarm_id.clone(), + }; + } + swarm_id.is_some() + }; + if resumed { + crate::swarm::peer::announce(&state); } } } diff --git a/src/state/mod.rs b/src/state/mod.rs index 20f92c5..d8291a7 100644 --- a/src/state/mod.rs +++ b/src/state/mod.rs @@ -24,7 +24,7 @@ mod read; mod session; mod timing; -pub(crate) use peer::{HelloState, Incidents, Liveness, PeerState, Standing}; +pub(crate) use peer::{HelloState, Incidents, Keepalive, Liveness, PeerState, Standing}; pub(crate) use read::{ReadSource, StaleView}; pub(crate) use session::SyncSession; pub(crate) use timing::Timing; diff --git a/src/state/peer.rs b/src/state/peer.rs index c9356ea..1f08767 100644 --- a/src/state/peer.rs +++ b/src/state/peer.rs @@ -67,8 +67,8 @@ const HELLO_RETRY_MAX: Duration = Duration::from_secs(30); #[derive(Clone)] pub(crate) struct PeerState { - /// Whether we lead the pair (`swarm::pair`): we send the keepalive STATUSes. Fixed for - /// the pair, so computed once. + /// Whether we lead the pair (`swarm::pair`): we send the keepalives. Fixed for the + /// pair, so computed once. pub(crate) leads: bool, pub(crate) standing: Standing, @@ -85,6 +85,9 @@ pub(crate) struct PeerState { pub(crate) liveness: Liveness, pub(crate) added_at: Instant, pub(crate) last_seen: Option, + /// Our last keepalive to it (a PING or a STATUS). + pub(crate) last_keepalive_at: Option, + /// Our last STATUS to it, for whatever reason (`status_refresh`). pub(crate) last_status_at: Option, /// Last time a sync message from this peer was applied (survives resets: the @@ -133,7 +136,8 @@ pub(crate) struct PeerState { } /// Trouble with a peer, counted since it was first listed. Lost keepalives and refused -/// sends are in the traffic counters (`transport::traffic`): every STATUS_REPLY answers a STATUS. +/// sends are in the traffic counters (`transport::traffic`): every PONG answers a PING, +/// every STATUS_REPLY a STATUS. #[derive(Debug, Default, Clone, Copy, serde::Serialize)] #[serde(rename_all = "camelCase")] pub(crate) struct Incidents { @@ -157,6 +161,15 @@ impl Incidents { } } +/// What a keepalive due is (`PeerState::take_keepalive_due`). +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum Keepalive { + /// A PING: the peer's transport answers it with a PONG. + Ping, + /// A STATUS: our state, answered with the peer's. + Status, +} + /// Unanswered HELLOs after which a peer counts as unreachable (1 + 2 + 4 s apart). const UNREACHABLE_AFTER_HELLOS: u32 = 3; @@ -172,6 +185,7 @@ impl PeerState { liveness: Liveness::Active, added_at: Instant::now(), last_seen: None, + last_keepalive_at: None, last_status_at: None, last_sync_rx: None, last_resync_at: None, @@ -287,16 +301,21 @@ impl PeerState { true } - /// True (and records the time) if an established, quiet peer needs a keepalive. + /// The keepalive an established, quiet peer needs, if any (and records it): a PING, + /// or a STATUS if we sent it none for `status_refresh`. /// - /// One STATUS and its STATUS_REPLY tell both sides that the other is alive, so only one side of - /// a pair needs to send them: the one leading the pair (`leads`). Each of its STATUSes - /// makes the other side's peer look recently heard from, so the other side never - /// sends while it does. If they stop, the other side sends once it has heard - /// nothing for half an interval more. Without that, both sides would get due at about - /// the same time (each one's quiet time starts with the same exchange) and their - /// STATUSes would often cross. - pub(crate) fn take_keepalive_due(&mut self, timing: &Timing, now: Instant) -> bool { + /// A PING and its PONG (or a STATUS and its STATUS_REPLY) tell both sides that the + /// other is alive, so only one side of a pair needs to send them: the one leading the + /// pair (`leads`). Each of its keepalives makes the other side's peer look recently + /// heard from, so the other side never sends while it does. If they stop, the other + /// side sends once it has heard nothing for half an interval more. Without that, both + /// sides would get due at about the same time (each one's quiet time starts with the + /// same exchange) and their keepalives would often cross. + pub(crate) fn take_keepalive_due( + &mut self, + timing: &Timing, + now: Instant, + ) -> Option { let quiet_enough = if self.leads { timing.keepalive } else { @@ -307,16 +326,24 @@ impl PeerState { && now.saturating_duration_since(self.added_at) >= timing.keepalive && self.quiet_for(now) >= quiet_enough && self - .last_status_at + .last_keepalive_at .is_none_or(|t| now.saturating_duration_since(t) >= timing.keepalive); - - if due { - self.last_status_at = Some(now); + if !due { + return None; } - due + self.last_keepalive_at = Some(now); + let status = self + .last_status_at + .is_none_or(|t| now.saturating_duration_since(t) >= timing.status_refresh); + Some(if status { + Keepalive::Status + } else { + Keepalive::Ping + }) } - /// Self-healing check on STATUS/STATUS_REPLY. Returns true if a re-handshake should start. + /// Self-healing check on a STATUS or STATUS_REPLY. Returns true if a re-handshake + /// should start. pub(crate) fn needs_heal( &mut self, local: &transport::HeadsSummary, @@ -378,6 +405,33 @@ impl PeerState { mod tests { use super::*; + /// Keepalives are PINGs, every `status_refresh` a STATUS; only when quiet. + #[test] + fn keepalives_are_pings_and_now_and_then_a_status() { + 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 start = peer.added_at; + assert_eq!(peer.take_keepalive_due(&timing, start), None); // Too new. + let mut kinds = Vec::new(); + for interval in 1..=6 { + let now = start + timing.keepalive * interval; + let due = peer.take_keepalive_due(&timing, now); + if due == Some(Keepalive::Status) { + peer.last_status_at = Some(now); + } + kinds.extend(due); + // Not again within the interval. + assert_eq!(peer.take_keepalive_due(&timing, now), None); + } + use Keepalive::{Ping, Status}; + assert_eq!(kinds, [Status, Ping, Ping, Status, Ping, Ping]); + // Heard from just now: nothing due. + let now = start + timing.keepalive * 7; + peer.mark_seen(now); + assert_eq!(peer.take_keepalive_due(&timing, now), None); + } + /// R41: unreachability lasts until the peer is heard from, not for a set time, and /// probes back off meanwhile. #[test] diff --git a/src/state/timing.rs b/src/state/timing.rs index 6ce5bef..9055ed6 100644 --- a/src/state/timing.rs +++ b/src/state/timing.rs @@ -18,6 +18,10 @@ const PROBE_MAX_FACTOR: u32 = 6; #[derive(Clone, Copy, Debug)] pub(crate) struct Timing { pub(crate) keepalive: Duration, + /// How often a quiet pair's keepalive is a STATUS rather than a PING + /// (`PeerState::take_keepalive_due`): its state (heads, readiness, epoch) for the heal + /// and against lost announcements. Changes go out as they happen anyway. + pub(crate) status_refresh: Duration, pub(crate) dormant_after: Duration, pub(crate) resync_after: Duration, pub(crate) dormant_probe: Duration, @@ -43,8 +47,9 @@ impl Timing { /// /// Each node keeps about `k` neighbours (`--neighbours`) whatever the swarm's size, /// and only one side of a pair sends them, so idle, a node sends about `k` messages and - /// receives about `k` (STATUSes and STATUS_REPLYs) per keepalive interval (more on a stable - /// member others prefer). The interval trades that against how soon a silent peer is + /// receives about `k` per keepalive interval (more on a stable member others prefer): + /// PINGs and PONGs (a few bytes), every third a STATUS and STATUS_REPLY (heads, + /// readiness, epoch) instead. The interval trades that against how soon a silent peer is /// noticed (`dormant_after`, 4 intervals) and a gap closes (within an interval). /// Recommended: /// @@ -68,7 +73,10 @@ impl Timing { ) -> Self { Self { keepalive, - dormant_after: keepalive * 4, // dormant_after >= 4* keepalive to tolerate three lost STATUSes in a row + // Every third keepalive; with `resync_after` at two keepalives, a heal needs + // two STATUSes that differ: within about two refreshes. + status_refresh: keepalive * 3, + dormant_after: keepalive * 4, // dormant_after >= 4* keepalive to tolerate three lost keepalives in a row // >= 2 keepalives, so a second exchange confirms the difference; and >= twice // the first sync reply timeout, so a lost message's sender resends it first. resync_after: (keepalive * 2).max(sync_reply_timeout * 2), -- cgit v1.3.1