assess swarm stability

1 files changed, 263 insertions, 0 deletions

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);
+    }
+}