From becffeb2c78b0e2e17eabf5060de2f2056710ff1 Mon Sep 17 00:00:00 2001 From: tv Date: Wed, 28 Oct 2026 03:07:30 +0100 Subject: keepalive takes turns --- src/state/peer.rs | 96 +++++++++++++++++++++++++++++++++++++++------ src/swarm/pair.rs | 4 +- src/swarm/peer/inbound.rs | 3 +- src/swarm/peer/keepalive.rs | 15 +++++++ 4 files changed, 104 insertions(+), 14 deletions(-) diff --git a/src/state/peer.rs b/src/state/peer.rs index 1f08767..b723ccd 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 keepalives. Fixed for the - /// pair, so computed once. + /// Whether we lead the pair (`swarm::pair`): we send the first keepalive. Fixed for + /// the pair, so computed once. pub(crate) leads: bool, pub(crate) standing: Standing, @@ -87,7 +87,9 @@ pub(crate) struct PeerState { 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`). + /// Its last keepalive to us (a PING or a STATUS): after it, the next is ours. + pub(crate) keepalive_received_at: Option, + /// The last STATUS between us, either way (`status_refresh`). pub(crate) last_status_at: Option, /// Last time a sync message from this peer was applied (survives resets: the @@ -186,6 +188,7 @@ impl PeerState { added_at: Instant::now(), last_seen: None, last_keepalive_at: None, + keepalive_received_at: None, last_status_at: None, last_sync_rx: None, last_resync_at: None, @@ -302,21 +305,37 @@ impl PeerState { } /// 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`. + /// or a STATUS if none went between us for `status_refresh`. /// /// 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. + /// other is alive, so one per interval does for the pair. The two sides take turns: + /// after a keepalive from the peer, the next is ours (the one leading the pair, + /// `leads`, sends the first), so both measure the round trip and each sends every + /// other interval. On our turn we send once the pair was quiet for an interval; on its + /// turn only after half an interval more, in case it doesn't (a keepalive or its + /// answer was lost). Without turns, 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. + /// + /// When they do cross anyway (both sent before seeing the other's: the peer's came + /// within half an interval after ours), each side would take the next turn for its + /// own and they would keep crossing; then the next turn is the leader's. pub(crate) fn take_keepalive_due( &mut self, timing: &Timing, now: Instant, ) -> Option { - let quiet_enough = if self.leads { + let our_turn = match (self.last_keepalive_at, self.keepalive_received_at) { + (None, None) => self.leads, + (None, Some(_)) => true, + (Some(_), None) => false, + (Some(sent), Some(received)) if received > sent => { + let crossed = received.saturating_duration_since(sent) < timing.keepalive / 2; + !crossed || self.leads + } + (Some(_), Some(_)) => false, + }; + let quiet_enough = if our_turn { timing.keepalive } else { timing.keepalive + timing.keepalive / 2 @@ -405,6 +424,61 @@ impl PeerState { mod tests { use super::*; + /// The two sides of a pair take turns: each sends every other interval. + #[test] + fn keepalives_take_turns() { + let timing = Timing::from_keepalive(Duration::from_secs(60), Duration::from_secs(300)); + let mut a = PeerState::new(true, &timing); + let mut b = PeerState::new(false, &timing); + a.hello_state = HelloState::Established; + b.hello_state = HelloState::Established; + b.added_at = a.added_at; + let mut senders = Vec::new(); + let mut now = a.added_at; + for _ in 0..40 { + now += Duration::from_secs(5); + // Whoever is due sends; the other one gets it at once, and both hear from + // each other (the keepalive, its answer). + if a.take_keepalive_due(&timing, now).is_some() { + senders.push("a"); + b.keepalive_received_at = Some(now); + } else if b.take_keepalive_due(&timing, now).is_some() { + senders.push("b"); + a.keepalive_received_at = Some(now); + } else { + continue; + } + a.mark_seen(now); + b.mark_seen(now); + } + // 200 s: one keepalive a minute, a first, then taking turns. + assert_eq!(senders, ["a", "b", "a"]); + } + + /// Keepalives that crossed (both sides sent before seeing the other's) leave the next + /// turn to the leader, rather than both taking it and crossing again. + #[test] + fn crossed_keepalives_go_back_to_taking_turns() { + let timing = Timing::from_keepalive(Duration::from_secs(60), Duration::from_secs(300)); + let mut a = PeerState::new(true, &timing); + let mut b = PeerState::new(false, &timing); + let start = a.added_at; + b.added_at = start; + for peer in [&mut a, &mut b] { + peer.hello_state = HelloState::Established; + } + // Both sent at 100 s and got the other's a moment later. + let sent = start + Duration::from_secs(100); + for peer in [&mut a, &mut b] { + peer.last_keepalive_at = Some(sent); + peer.keepalive_received_at = Some(sent + Duration::from_millis(50)); + peer.mark_seen(sent + Duration::from_millis(50)); + } + let next = sent + timing.keepalive + Duration::from_secs(1); + assert!(a.take_keepalive_due(&timing, next).is_some()); // The leader's turn. + assert!(b.take_keepalive_due(&timing, next).is_none()); + } + /// 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/swarm/pair.rs b/src/swarm/pair.rs index 41fc219..0d45b32 100644 --- a/src/swarm/pair.rs +++ b/src/swarm/pair.rs @@ -1,5 +1,5 @@ -//! Which side of a pair of nodes leads, for duties only one side needs to do (sending -//! keepalive STATUSes). +//! Which side of a pair of nodes leads, for what one side needs to start (the first +//! keepalive; then the two take turns, `PeerState::take_keepalive_due`). //! //! Both sides decide on their own, from the two keys alone, and agree. The leader is //! chosen per pair by one bit of a hash of both keys, not by comparing the keys: a diff --git a/src/swarm/peer/inbound.rs b/src/swarm/peer/inbound.rs index 0eaefb9..2e89739 100644 --- a/src/swarm/peer/inbound.rs +++ b/src/swarm/peer/inbound.rs @@ -11,7 +11,7 @@ use super::epoch::observe_peer_epoch; use super::handshake::{ discover_peers, ensure_handshake, establish_peer, peer_swarm, remember_peer, send_hello_reply, }; -use super::keepalive::{heal_if_diverged, local_status, note_alive, note_status}; +use super::keepalive::{heal_if_diverged, local_status, note_alive, note_status, status_received}; use super::lanes::SyncLanes; use crate::state::AppState; use crate::swarm::sync::SyncEvent; @@ -149,6 +149,7 @@ pub(crate) async fn handle_message( crate::swarm::blob::handle(state, &peer, blob).await; } transport::ParsedMessage::Status(remote) => { + status_received(state, &peer).await; let local = on_status(state, &peer, &remote, reactivated).await; if let Err(error) = state .transport diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs index 460b71e..efb8a65 100644 --- a/src/swarm/peer/keepalive.rs +++ b/src/swarm/peer/keepalive.rs @@ -63,6 +63,10 @@ pub(super) async fn note_alive(state: &AppState, peer: &str, message: &ParsedMes }; let reactivated = { let mut peer_state = peer_state.lock().await; + if matches!(message, ParsedMessage::Ping(_)) { + // Its keepalive (or a path probe): the next is ours. + peer_state.keepalive_received_at = Some(now); + } if peer_state.unreachable_since.is_some() { // Reachable again (`mark_seen` clears it). state.topology_changed.notify_one(); @@ -147,6 +151,17 @@ pub(crate) async fn local_status(state: &AppState, peer: &str) -> transport::Sta } } +/// A STATUS (not a reply) from `peer`: its keepalive, and a STATUS between us +/// (`Timing::status_refresh`). +pub(super) async fn status_received(state: &AppState, peer: &str) { + if let Some(peer_state) = state.get_peer_state(peer).await { + let mut peer_state = peer_state.lock().await; + let now = Instant::now(); + peer_state.keepalive_received_at = Some(now); + peer_state.last_status_at = Some(now); + } +} + /// What a peer's status says about it. pub(super) async fn note_status(state: &AppState, peer: &str, remote: &transport::Status) { if let Some(peer_state) = state.get_peer_state(peer).await { -- cgit v1.3.1