From e13d027567f781c9729a518913b21ec4ee273039 Mon Sep 17 00:00:00 2001 From: tv Date: Wed, 28 Oct 2026 11:57:54 +0100 Subject: peers from member list start as unknown --- src/state/host.rs | 6 +++--- src/state/peer.rs | 39 +++++++++++++++++++++++++++++++++------ 2 files changed, 36 insertions(+), 9 deletions(-) (limited to 'src/state') diff --git a/src/state/host.rs b/src/state/host.rs index 6eab803..b55239d 100644 --- a/src/state/host.rs +++ b/src/state/host.rs @@ -11,7 +11,7 @@ use tokio::sync::broadcast; use tokio::time::Instant; use tracing::debug; -use super::{AppState, HelloState, Liveness, NodeLifecycle, StaleView, Standing}; +use super::{AppState, HelloState, NodeLifecycle, StaleView, Standing}; use crate::document::replica::Epoch; use crate::locks::MutexExt; use crate::remake::{Host, PeerSummary, RemakeState}; @@ -42,7 +42,7 @@ impl AppState { let mut peer_state = peer_state_mutex.lock().await; // What it said about syncing with us was about our old epoch. peer_state.syncs_with_us = None; - let gets_our_opening = peer_state.liveness == Liveness::Active + let gets_our_opening = !peer_state.liveness.is_dormant() && matches!(peer_state.standing, Standing::Member { .. }) && peer_state.epoch == Some(local) && peer_state.reports_ready == Some(true); @@ -131,7 +131,7 @@ impl Host for AppState { let peer = peer.lock().await; summaries.push(PeerSummary { id, - active: peer.liveness == Liveness::Active, + active: !peer.liveness.is_dormant(), member: matches!(peer.standing, Standing::Member { .. }) && peer.hello_state == HelloState::Established, ready: peer.reports_ready, diff --git a/src/state/peer.rs b/src/state/peer.rs index b723ccd..9f2da0f 100644 --- a/src/state/peer.rs +++ b/src/state/peer.rs @@ -39,6 +39,11 @@ impl Standing { #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub(crate) enum Liveness { + /// Nothing heard from it since this node started: known only from the member list + /// or another peer. It counts as active (sync, keepalives and dormancy treat it so) + /// until its first message makes it active, or `Timing::dormant_after` of silence + /// dormant. + Unknown, Active, /// No inbound message for Timing::dormant_after. Kept indefinitely, probed slowly. Dormant { @@ -46,6 +51,12 @@ pub(crate) enum Liveness { }, } +impl Liveness { + pub(crate) fn is_dormant(self) -> bool { + matches!(self, Self::Dormant { .. }) + } +} + #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] pub(crate) enum HelloState { #[default] @@ -184,7 +195,7 @@ impl PeerState { hello_backoff: Backoff::new(HELLO_RETRY, HELLO_RETRY_MAX), probe_backoff: timing.probe_backoff(), sync: SyncSession::new(timing.sync_reply_backoff()), - liveness: Liveness::Active, + liveness: Liveness::Unknown, added_at: Instant::now(), last_seen: None, last_keepalive_at: None, @@ -219,7 +230,7 @@ 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 { - if self.liveness != Liveness::Active || self.hello_state != HelloState::Established { + if self.liveness.is_dormant() || self.hello_state != HelloState::Established { return None; } let overdue = self.sync.take_overdue_reply(now); @@ -240,6 +251,9 @@ impl PeerState { self.unreachable_since = None; self.probe_backoff.reset(); + if self.liveness == Liveness::Unknown { + self.liveness = Liveness::Active; + } if let Liveness::Dormant { since } = self.liveness { self.incidents.dormant_ms += now.saturating_duration_since(since).as_millis() as u64; self.liveness = Liveness::Active; @@ -287,7 +301,7 @@ impl PeerState { /// Moves a silent peer to Dormant. Returns true on the transition. pub(crate) fn check_dormancy(&mut self, timing: &Timing, now: Instant) -> bool { - if self.liveness != Liveness::Active || self.quiet_for(now) < timing.dormant_after { + if self.liveness.is_dormant() || self.quiet_for(now) < timing.dormant_after { return false; } @@ -340,7 +354,7 @@ impl PeerState { } else { timing.keepalive + timing.keepalive / 2 }; - let due = self.liveness == Liveness::Active + let due = !self.liveness.is_dormant() && self.hello_state == HelloState::Established && now.saturating_duration_since(self.added_at) >= timing.keepalive && self.quiet_for(now) >= quiet_enough @@ -387,7 +401,7 @@ impl PeerState { // While an answer to our last sync message is due, a lost message is the reply // timeout's to resend; healing is for when neither side has anything outstanding. - if self.liveness != Liveness::Active + if self.liveness.is_dormant() || self.hello_state != HelloState::Established || self.sync.reply.is_some() { @@ -413,7 +427,7 @@ impl PeerState { /// Heard from recently. A returning node counts as alive from its first message, /// before the handshake finishes, so a swarm-wide forget can't make it flap. pub(crate) fn is_alive(&self, timing: &Timing, now: Instant) -> bool { - self.liveness == Liveness::Active + !self.liveness.is_dormant() && self .last_seen .is_some_and(|t| now.saturating_duration_since(t) < timing.dormant_after) @@ -424,6 +438,19 @@ impl PeerState { mod tests { use super::*; + /// Unknown until heard from; silent long enough, dormant like an active peer. + #[test] + fn unknown_until_heard_from() { + let timing = Timing::from_keepalive(Duration::from_secs(60), Duration::from_secs(300)); + let mut heard = PeerState::new(true, &timing); + assert_eq!(heard.liveness, Liveness::Unknown); + assert!(!heard.mark_seen(heard.added_at), "no re-handshake: none yet"); + assert_eq!(heard.liveness, Liveness::Active); + let mut silent = PeerState::new(true, &timing); + assert!(silent.check_dormancy(&timing, silent.added_at + timing.dormant_after)); + assert!(silent.liveness.is_dormant()); + } + /// The two sides of a pair take turns: each sends every other interval. #[test] fn keepalives_take_turns() { -- cgit v1.3.1