//! Whether this node is a stable member of the swarm, judged by itself. //! //! Long-running, well-connected nodes make a good backbone; laptops that sleep and //! phones that lose signal don't. Each node measures what fraction of the last 24 hours //! it had at least one live peer, and publishes only a coarse tier in its registry entry //! (`_continuum/members//tier`): `stable` or `transient`, written only when it //! changes, with hysteresis. So the swarm learns about stability from the replicated //! document alone, without anyone measuring anyone else, and the tier changes rarely. //! //! * transient → stable: at least 24 h observed, connected at least 95% of it; //! * stable → transient: connected less than 80% (over at least an hour observed). //! //! After a restart a node has no history. If its entry says `stable` and a peer reports //! (in its HELLO_REPLY) that it last heard from us less than an hour ago, the node was //! only briefly away and keeps its tier, starting with a full day's credit. Otherwise it //! starts over as transient. Time the process didn't run (a laptop asleep) counts as //! disconnected. use std::collections::VecDeque; use std::time::Duration; use tokio::time::Instant; use tracing::{info, warn}; use crate::state::{AppState, HelloState}; use crate::swarm::members::{self, Tier}; const WINDOW: Duration = Duration::from_secs(24 * 3600); const BUCKET: Duration = Duration::from_secs(5 * 60); const BECOME_STABLE: f64 = 0.95; const BECOME_TRANSIENT: f64 = 0.80; /// A day observed (the oldest bucket may be partly outside the window). const FULL_DAY: Duration = WINDOW.saturating_sub(BUCKET); /// Less than this observed is too little to demote a stable node. const MIN_TO_DEMOTE: Duration = Duration::from_secs(3600); /// An absence shorter than this keeps a stable node stable across a restart. const BRIEF_ABSENCE: Duration = Duration::from_secs(3600); /// How long after becoming ready we wait for a peer to tell us how long we were away. const AWAY_REPORT_GRACE: Duration = Duration::from_secs(30); /// A gap between two measurements longer than this means we didn't run (asleep). const MAX_TICK_GAP: Duration = Duration::from_secs(30); #[derive(Debug, Default)] pub(crate) struct Stability { buckets: VecDeque, last_tick: Option, /// The shortest absence any peer reported to us in this run. away: Option, ready_since: Option, /// The decision about the tier we found after a restart is made. settled: bool, } #[derive(Debug, Clone, Copy)] struct Bucket { start: Instant, observed: Duration, connected: Duration, } impl Stability { fn record(&mut self, now: Instant, connected: bool) { let Some(last) = self.last_tick.replace(now) else { return; }; let elapsed = now.saturating_duration_since(last).min(WINDOW); let connected = connected && elapsed <= MAX_TICK_GAP; if self .buckets .back() .is_none_or(|bucket| now.saturating_duration_since(bucket.start) >= BUCKET) { self.buckets.push_back(Bucket { start: now, observed: Duration::ZERO, connected: Duration::ZERO, }); } let bucket = self.buckets.back_mut().expect("just ensured"); bucket.observed += elapsed; if connected { bucket.connected += elapsed; } while self .buckets .front() .is_some_and(|bucket| now.saturating_duration_since(bucket.start) > WINDOW) { self.buckets.pop_front(); } } /// A full day of connectivity, for a stable node that was only briefly away. fn credit(&mut self, now: Instant) { self.buckets.push_front(Bucket { start: now, observed: WINDOW, connected: WINDOW, }); } fn observed(&self) -> Duration { self.buckets.iter().map(|bucket| bucket.observed).sum() } fn fraction(&self) -> f64 { let observed = self.observed().as_secs_f64(); if observed == 0.0 { return 0.0; } let connected: f64 = self.buckets.iter().map(|b| b.connected.as_secs_f64()).sum(); connected / observed } /// The tier we should have, given the one we have now. fn assess(&self, current: Tier) -> Tier { match current { Tier::Transient if self.observed() >= FULL_DAY && self.fraction() >= BECOME_STABLE => { Tier::Stable } Tier::Stable if self.observed() >= MIN_TO_DEMOTE && self.fraction() < BECOME_TRANSIENT => { Tier::Transient } tier => tier, } } } fn lock(state: &AppState) -> std::sync::MutexGuard<'_, Stability> { state .stability .lock() .unwrap_or_else(|poisoned| poisoned.into_inner()) } /// A HELLO_REPLY told us how long ago that peer last heard from us before this run. pub(crate) fn note_away(state: &AppState, away: Option) { if let Some(away) = away { let mut stability = lock(state); stability.away = Some(stability.away.map_or(away, |known| known.min(away))); } } /// Measures, and updates our tier in the registry when it should change. Called every /// few seconds by the peer worker. pub(crate) async fn tick(state: &AppState) { let now = Instant::now(); let mut connected = false; let peers: Vec<_> = state.peer_states.read().await.values().cloned().collect(); for peer_state in peers { let peer_state = peer_state.lock().await; if peer_state.hello_state == HelloState::Established && peer_state.is_alive(&state.timing, now) { connected = true; break; } } lock(state).record(now, connected); if !state.is_ready().await { return; } let local = state.local_id().to_owned(); let current = members::tier(&state.doc.read().await.doc, &local); let desired = { let mut stability = lock(state); let ready_since = *stability.ready_since.get_or_insert(now); if stability.settled { stability.assess(current) } else { let brief = stability.away.is_some_and(|away| away < BRIEF_ABSENCE); if current == Tier::Stable && !brief && now < ready_since + AWAY_REPORT_GRACE { return; // a peer may still tell us we were only briefly away } stability.settled = true; match current { Tier::Stable if brief => { stability.credit(now); info!("Back after a brief absence; staying a stable member"); Tier::Stable } // Restarted without a report that we were only briefly away. Tier::Stable => Tier::Transient, Tier::Transient => Tier::Transient, } } }; if desired == current { return; } let fraction = lock(state).fraction(); match state .mutate_doc(move |doc| members::set_tier(doc, &local, desired)) .await { Ok(()) => info!( tier = desired.as_str(), connected_percent = (fraction * 100.0).round(), "Stability tier changed" ), Err(error) => warn!(error = %error, "Failed to record our stability tier"), } } #[cfg(test)] mod tests { use super::*; fn run(stability: &mut Stability, start: Instant, hours: u64, connected: impl Fn(u64) -> bool) { let step = Duration::from_secs(5); let steps = hours * 3600 / 5; for i in 0..=steps { stability.record(start + step * i as u32, connected(i)); } } #[test] fn becomes_stable_after_a_well_connected_day() { let mut stability = Stability::default(); let start = Instant::now(); run(&mut stability, start, 12, |_| true); assert_eq!(stability.assess(Tier::Transient), Tier::Transient); run( &mut stability, start + Duration::from_secs(12 * 3600), 13, |_| true, ); assert_eq!(stability.assess(Tier::Transient), Tier::Stable); } #[test] fn hysteresis_between_the_thresholds() { let mut stability = Stability::default(); // Connected 9 of every 10 measurements: 90%. run(&mut stability, Instant::now(), 25, |i| i % 10 != 0); assert_eq!(stability.assess(Tier::Transient), Tier::Transient); assert_eq!(stability.assess(Tier::Stable), Tier::Stable); } #[test] fn falls_back_to_transient_when_often_disconnected() { let mut stability = Stability::default(); // Connected half the time. run(&mut stability, Instant::now(), 2, |i| i % 2 == 0); assert_eq!(stability.assess(Tier::Stable), Tier::Transient); } #[test] fn time_asleep_counts_as_disconnected() { let mut stability = Stability::default(); let start = Instant::now(); run(&mut stability, start, 1, |_| true); // The process didn't run for two hours, then one more measurement. stability.record(start + Duration::from_secs(3 * 3600), true); assert!(stability.fraction() < 0.5); } }