diff options
| -rw-r--r-- | src/main.rs | 1 | ||||
| -rw-r--r-- | src/state.rs | 2 | ||||
| -rw-r--r-- | src/swarm/members.rs | 37 | ||||
| -rw-r--r-- | src/swarm/mod.rs | 1 | ||||
| -rw-r--r-- | src/swarm/peer.rs | 31 | ||||
| -rw-r--r-- | src/swarm/stability.rs | 263 | ||||
| -rw-r--r-- | src/transport/mod.rs | 3 | ||||
| -rw-r--r-- | src/transport/mycelium.rs | 3 |
8 files changed, 330 insertions, 11 deletions
diff --git a/src/main.rs b/src/main.rs index 15f5c0e..972c325 100644 --- a/src/main.rs +++ b/src/main.rs @@ -195,6 +195,7 @@ async fn main() -> Result<()> { neighbours: Default::default(), pulling: Default::default(), rtt: Default::default(), + stability: Default::default(), stale: Default::default(), catch_up_source: Default::default(), catch_up_preference: Default::default(), diff --git a/src/state.rs b/src/state.rs index 9d567a8..769dbad 100644 --- a/src/state.rs +++ b/src/state.rs @@ -542,6 +542,8 @@ pub(crate) struct AppState { /// Target number of neighbours (`--neighbours`); up to this many + 1 members the /// swarm is a full mesh. pub(crate) neighbours_k: usize, + /// Our own connectivity record and stability tier (`swarm::stability`). + pub(crate) stability: Arc<std::sync::Mutex<crate::swarm::stability::Stability>>, /// Round-trip times to peers (`swarm::rtt`). pub(crate) rtt: Arc<std::sync::Mutex<crate::swarm::rtt::RttTable>>, /// The neighbour we are pulling changes from, and since when (`swarm::relay`). diff --git a/src/swarm/members.rs b/src/swarm/members.rs index 45c8aeb..e64dd20 100644 --- a/src/swarm/members.rs +++ b/src/swarm/members.rs @@ -9,7 +9,7 @@ //! cancels the record and registers again. use automerge::AutoCommit; -use serde_json::json; +use serde_json::{Value as JsonValue, json}; use tracing::{info, warn}; use crate::document::store::{self, StoreError}; @@ -34,6 +34,41 @@ pub(crate) fn members(doc: &AutoCommit) -> Vec<String> { members } +/// A member's stability tier, judged by itself (`swarm::stability`). +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub(crate) enum Tier { + Stable, + #[default] + Transient, +} + +impl Tier { + pub(crate) fn as_str(self) -> &'static str { + match self { + Self::Stable => "stable", + Self::Transient => "transient", + } + } +} + +fn tier_path(pk: &str) -> Vec<String> { + let mut path = member_path(pk); + path.push("tier".to_owned()); + path +} + +/// `pk`'s tier; transient unless its entry says stable. +pub(crate) fn tier(doc: &AutoCommit, pk: &str) -> Tier { + match store::read_path(doc, &tier_path(pk)) { + Ok(Some(JsonValue::String(tier))) if tier == "stable" => Tier::Stable, + _ => Tier::Transient, + } +} + +pub(crate) fn set_tier(doc: &mut AutoCommit, pk: &str, tier: Tier) -> Result<(), StoreError> { + store::write_path(doc, &tier_path(pk), &json!(tier.as_str())) +} + /// Registers this node if it is ready and its entry is missing. pub(crate) async fn ensure_registered(state: &AppState) { if !state.is_ready().await { diff --git a/src/swarm/mod.rs b/src/swarm/mod.rs index 6de7bf3..ee0de20 100644 --- a/src/swarm/mod.rs +++ b/src/swarm/mod.rs @@ -9,5 +9,6 @@ pub(crate) mod peer; pub(crate) mod relay; pub(crate) mod rtt; pub(crate) mod source; +pub(crate) mod stability; pub(crate) mod sync; pub(crate) mod topology; diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs index 55d0022..a936fa2 100644 --- a/src/swarm/peer.rs +++ b/src/swarm/peer.rs @@ -315,8 +315,14 @@ async fn ensure_handshake(state: &AppState, peer: &str) { } } -/// Answers a HELLO that carried `run_id`. -async fn send_hello_reply(state: &AppState, peer: &str, run_id: transport::RunId) -> Result<()> { +/// Answers a HELLO that carried `run_id`; `away` is how long before it we had last heard +/// from the peer. +async fn send_hello_reply( + state: &AppState, + peer: &str, + run_id: transport::RunId, + away: Option<Duration>, +) -> Result<()> { let swarm_id = state .swarm_id() .await @@ -355,6 +361,7 @@ async fn send_hello_reply(state: &AppState, peer: &str, run_id: transport::RunId auth, ready, run_id, + away_ms: away.map(|away| away.as_millis() as u64), }; debug!(to = %peer, ready, "Sending HELLO_REPLY"); @@ -484,6 +491,7 @@ pub(crate) async fn run_peer_worker(state: AppState) { if last_forget_check.is_none_or(|t| t.elapsed() >= FORGET_CHECK_INTERVAL) { apply_forget_records(&state).await; crate::swarm::members::ensure_registered(&state).await; + crate::swarm::stability::tick(&state).await; if let Some(change) = topology::refresh(&state).await { apply_neighbour_change(&state, change).await; } @@ -724,13 +732,14 @@ pub(crate) async fn handle_message( remember_peer(state, &peer).await; - // Any inbound message proves liveness. - let reactivated = state - .get_or_create_peer_state(&peer) - .await - .lock() - .await - .mark_seen(Instant::now()); + // Any inbound message proves liveness. When we heard from it before tells a + // restarted peer how long it was away (in our HELLO_REPLY). + let (reactivated, previously_seen) = { + let peer_state_mutex = state.get_or_create_peer_state(&peer).await; + let mut peer_state = peer_state_mutex.lock().await; + let previously_seen = peer_state.last_seen; + (peer_state.mark_seen(Instant::now()), previously_seen) + }; crate::swarm::source::heard_from(state, &peer); if reactivated { @@ -757,7 +766,8 @@ pub(crate) async fn handle_message( } establish_peer(state, &peer, swarm_id, true, hello.ready).await?; - if let Err(error) = send_hello_reply(state, &peer, hello.run_id).await { + let away = previously_seen.map(|seen| seen.elapsed()); + if let Err(error) = send_hello_reply(state, &peer, hello.run_id, away).await { // The peer retries its HELLO. warn!(peer = %peer, error = %format!("{error:#}"), "HELLO_REPLY failed"); } @@ -768,6 +778,7 @@ pub(crate) async fn handle_message( } transport::ParsedMessage::HelloReply(reply) => { state.lock_rtt().hello_reply_received(&peer, Instant::now()); + crate::swarm::stability::note_away(state, reply.away_ms.map(Duration::from_millis)); establish_peer(state, &peer, reply.swarm_id.clone(), false, reply.ready).await?; discover_peers(state, &peer, &reply.peers).await; diff --git a/src/swarm/stability.rs b/src/swarm/stability.rs new file mode 100644 index 0000000..31ba4b8 --- /dev/null +++ b/src/swarm/stability.rs @@ -0,0 +1,263 @@ +//! 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/<pk>/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<Bucket>, + last_tick: Option<Instant>, + /// The shortest absence any peer reported to us in this run. + away: Option<Duration>, + ready_since: Option<Instant>, + /// 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<Duration>) { + 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); + } +} diff --git a/src/transport/mod.rs b/src/transport/mod.rs index e8d654d..19c4126 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -139,6 +139,9 @@ pub struct HelloReply { /// The `run_id` of the HELLO answered. A reply to a HELLO from an earlier run of the /// recipient (delivered late, or queued while it was down) is ignored. pub run_id: RunId, + /// How long before this HELLO the sender had last heard from the recipient, if ever + /// (in milliseconds): tells a restarted node how long it was away. + pub away_ms: Option<u64>, } #[derive(Debug, Clone)] diff --git a/src/transport/mycelium.rs b/src/transport/mycelium.rs index 51bbfab..9075ae8 100644 --- a/src/transport/mycelium.rs +++ b/src/transport/mycelium.rs @@ -892,6 +892,7 @@ pub struct HelloReplyShadow<'a> { auth: ByteBuf, ready: bool, run_id: ByteBuf, + away_ms: Option<u64>, } #[derive(Serialize, Deserialize)] @@ -972,6 +973,7 @@ pub fn hello_reply_to_bytes(msg: &HelloReply) -> Result<Vec<u8>, rmp_serde::enco auth: ByteBuf::from(msg.auth.clone()), ready: msg.ready, run_id: ByteBuf::from(msg.run_id.to_vec()), + away_ms: msg.away_ms, }; rmp_serde::to_vec(&shadow) } @@ -984,6 +986,7 @@ pub fn bytes_to_hello_reply(bytes: &[u8]) -> Result<HelloReply, TransportError> auth: shadow.auth.into_vec(), ready: shadow.ready, run_id: run_id_from_shadow(&shadow.run_id)?, + away_ms: shadow.away_ms, }) } |