//! What a node knows about each peer: standing, liveness, handshake, incidents. use std::time::Duration; use tokio::time::Instant; use super::{SyncSession, Timing}; use crate::backoff::Backoff; use crate::document::replica::Epoch; use crate::transport; /// How far a peer got into the swarm, as far as we're concerned. #[derive(Debug, Clone, Default, PartialEq, Eq)] pub(crate) enum Standing { /// Nothing proven yet. With a swarm key, everything but a handshake from it is /// ignored (`swarm::peer::admit`). #[default] Stranger, /// Proved the swarm key in a HELLO; no HELLO_REPLY yet. Only reached with a key. Admitted, /// Its HELLO_REPLY named our swarm (and, with a key, proved it): we sync with it. Member { swarm_id: String }, } impl Standing { /// The swarm a member belongs to; None before its HELLO_REPLY. pub(crate) fn swarm_id(&self) -> Option<&str> { match self { Self::Member { swarm_id } => Some(swarm_id), Self::Stranger | Self::Admitted => None, } } /// Proved the swarm key (meaningful only with one). pub(crate) fn is_admitted(&self) -> bool { !matches!(self, Self::Stranger) } } #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub(crate) enum Liveness { Active, /// No inbound message for Timing::dormant_after. Kept indefinitely, probed slowly. Dormant { since: Instant, }, } #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] pub(crate) enum HelloState { #[default] NotSent, /// A send is in flight. Sending, /// HELLO was sent (or the send failed); no HELLO_REPLY yet. Retried at /// `next_attempt` (`PeerState::hello_backoff`, or `dormant_probe` while dormant). AwaitingReply { next_attempt: Instant }, /// HELLO_REPLY received. No more retries. Established, } /// First wait for a HELLO_REPLY before sending the HELLO again; doubles while none /// comes. const HELLO_RETRY: Duration = Duration::from_secs(1); 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 PINGs. Fixed for /// the pair, so computed once. pub(crate) leads: bool, pub(crate) standing: Standing, pub(crate) hello_state: HelloState, /// Wait between unanswered HELLOs; reset whenever a new handshake starts. pub(crate) hello_backoff: Backoff, /// Everything about the current Automerge sync with this peer; a reset replaces it. pub(crate) sync: SyncSession, pub(crate) liveness: Liveness, pub(crate) added_at: Instant, pub(crate) last_seen: Option, pub(crate) last_ping_at: Option, /// Last time a sync message from this peer was applied (survives resets: the /// keepalive heal looks at it). pub(crate) last_sync_rx: Option, /// Last self-healing re-handshake (cooldown). pub(crate) last_resync_at: Option, /// Since when keepalives show the peer's heads differ from ours with no sync message /// from it in between (None while they match). pub(crate) diverged_since: Option, /// Epoch the peer advertised in its most recent message (late ones aside: see /// `epoch_seq`). pub(crate) epoch: Option, /// The peer's `epoch_seq` with `epoch`: how many times it changed epoch in its run. /// An epoch with a lower one is older than what we have, delivered late. pub(crate) epoch_seq: Option, /// The peer's run, from its HELLO or HELLO_REPLY: a new one starts `epoch_seq` over. pub(crate) run_id: Option, /// Whether the peer was ready according to its most recent sync message or keepalive. pub(crate) reports_ready: Option, /// Whether the peer said, in its most recent keepalive, that it would sync with us. pub(crate) syncs_with_us: Option, /// How many heads its most recent keepalive listed (None if it sent a digest). pub(crate) reported_heads: Option, /// HELLOs sent in the current handshake attempt without hearing anything back. pub(crate) hello_attempts: u32, /// Since when we consider the peer unreachable (three unanswered HELLOs, or /// dormant). Cleared by any message from it; ignored after `dormant_probe`, so the /// peer becomes a neighbour candidate again and gets tried once more. pub(crate) unreachable_since: Option, /// What went wrong with this peer so far (counted only, for /peers and /admin/stats). pub(crate) incidents: Incidents, } /// Trouble with a peer, counted since it was first listed. Lost keepalives and refused /// sends are in the traffic counters (`transport::traffic`): every PONG answers a PING. #[derive(Debug, Default, Clone, Copy, serde::Serialize, utoipa::ToSchema)] #[serde(rename_all = "camelCase")] pub(crate) struct Incidents { /// Our sync messages left unanswered until the reply timeout (then sent again). pub(crate) reply_timeouts: u64, /// Re-handshakes because our documents differed while sync was idle. pub(crate) heals: u64, /// Times the peer went dormant. pub(crate) dormancies: u64, /// Time spent dormant, in milliseconds, up to its last return (the current /// dormancy, if any, isn't included). pub(crate) dormant_ms: u64, } impl Incidents { pub(crate) fn add(&mut self, other: &Incidents) { self.reply_timeouts += other.reply_timeouts; self.heals += other.heals; self.dormancies += other.dormancies; self.dormant_ms += other.dormant_ms; } } /// Unanswered HELLOs after which a peer counts as unreachable (1 + 2 + 4 s apart). const UNREACHABLE_AFTER_HELLOS: u32 = 3; impl PeerState { pub(crate) fn new(leads: bool, timing: &Timing) -> Self { Self { leads, standing: Standing::default(), hello_state: HelloState::default(), hello_backoff: Backoff::new(HELLO_RETRY, HELLO_RETRY_MAX), sync: SyncSession::new(timing.sync_reply_backoff()), liveness: Liveness::Active, added_at: Instant::now(), last_seen: None, last_ping_at: None, last_sync_rx: None, last_resync_at: None, diverged_since: None, epoch: None, epoch_seq: None, run_id: None, reports_ready: None, syncs_with_us: None, reported_heads: None, hello_attempts: 0, unreachable_since: None, incidents: Incidents::default(), } } /// Start Automerge sync with this peer from scratch. A message still being handed to /// Mycelium was generated against the old state; its outcome is then ignored. pub(crate) fn reset_sync(&mut self) { // The same waits for answers, from the start. let mut reply_backoff = self.sync.reply_backoff; reply_backoff.reset(); self.sync = SyncSession::new(reply_backoff); } /// 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 { return None; } let overdue = self.sync.take_overdue_reply(now); if overdue.is_some() { self.incidents.reply_timeouts += 1; } overdue } fn quiet_for(&self, now: Instant) -> Duration { now.saturating_duration_since(self.last_seen.unwrap_or(self.added_at)) } /// Records inbound activity. Returns true if the peer was dormant. pub(crate) fn mark_seen(&mut self, now: Instant) -> bool { self.last_seen = Some(now); self.hello_attempts = 0; self.unreachable_since = None; if let Liveness::Dormant { since } = self.liveness { self.incidents.dormant_ms += now.saturating_duration_since(since).as_millis() as u64; self.liveness = Liveness::Active; // Force a fresh handshake; its reply resets sync on both sides. if self.hello_state != HelloState::Sending { self.hello_state = HelloState::NotSent; } return true; } false } /// A HELLO went out (or failed to). After the third unanswered one, the peer counts /// as unreachable. pub(crate) fn hello_sent(&mut self, now: Instant) { self.hello_attempts += 1; if self.hello_attempts >= UNREACHABLE_AFTER_HELLOS { self.unreachable_since.get_or_insert(now); } } /// Whether to leave the peer out when choosing neighbours (`swarm::topology`). pub(crate) fn is_unreachable(&self, timing: &Timing, now: Instant) -> bool { self.unreachable_since .is_some_and(|since| now.saturating_duration_since(since) < timing.dormant_probe) } /// No more HELLO retries (the peer isn't our neighbour any more). A HELLO in flight /// finishes; a later handshake starts over. pub(crate) fn stop_hello_retries(&mut self) { if matches!(self.hello_state, HelloState::AwaitingReply { .. }) { self.hello_state = HelloState::NotSent; } } /// 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 { return false; } self.liveness = Liveness::Dormant { since: now }; self.incidents.dormancies += 1; self.unreachable_since.get_or_insert(now); self.reset_sync(); if self.hello_state != HelloState::Sending { self.hello_state = HelloState::AwaitingReply { next_attempt: now + timing.dormant_probe, }; } true } /// True (and records the ping) if an established, quiet peer needs a keepalive. /// /// One PING and its PONG tell both sides that the other is alive, so only one side of /// a pair needs to ping: the one leading the pair (`leads`). Each of its PINGs /// makes the other side's peer look recently heard from, so the other side never /// pings while it does. If the pings stop, the other side pings 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 /// PINGs would often cross. pub(crate) fn take_keepalive_due(&mut self, timing: &Timing, now: Instant) -> bool { let quiet_enough = if self.leads { timing.keepalive } else { timing.keepalive + timing.keepalive / 2 }; let due = self.liveness == Liveness::Active && self.hello_state == HelloState::Established && now.saturating_duration_since(self.added_at) >= timing.keepalive && self.quiet_for(now) >= quiet_enough && self .last_ping_at .is_none_or(|t| now.saturating_duration_since(t) >= timing.keepalive); if due { self.last_ping_at = Some(now); } due } /// Self-healing check on PING/PONG. Returns true if a re-handshake should start. pub(crate) fn needs_heal( &mut self, local: &transport::HeadsSummary, remote: &transport::HeadsSummary, timing: &Timing, now: Instant, ) -> bool { if local == remote { self.diverged_since = None; return false; } // How long the heads have differed without sync making progress. Measured from // when we noticed, not from the last sync message: in a quiet swarm that one is // long past, and the first keepalive after a lost message would heal at once, // before the sender's reply timeout could resend it. Any sync message restarts // the clock. let noticed = match self.diverged_since { Some(since) if self.last_sync_rx.is_none_or(|rx| rx <= since) => since, _ => *self.diverged_since.insert(now), }; // 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 || self.hello_state != HelloState::Established || self.sync.reply.is_some() { return false; } let persisted = now.saturating_duration_since(noticed) >= timing.resync_after; let cooled_down = self .last_resync_at .is_none_or(|t| now.saturating_duration_since(t) >= timing.resync_after); if !(persisted && cooled_down) { return false; } self.diverged_since = None; self.incidents.heals += 1; self.last_resync_at = Some(now); self.hello_state = HelloState::NotSent; // establish_peer resets sync on the reply true } /// 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 .last_seen .is_some_and(|t| now.saturating_duration_since(t) < timing.dormant_after) } }