path: root/src/swarm/stability.rs

//! 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())
}

/// The fraction of the observed time (up to a day) we had a live peer, and how much
/// time was observed.
pub(crate) fn measured(state: &AppState) -> (f64, Duration) {
    let stability = lock(state);
    (stability.fraction(), stability.observed())
}

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