fips-core 0.3.74

//! Time-series history of node-level and per-peer statistics.
//!
//! Maintains a fast ring (1s × 3600 = 1h) and a slow ring (1m × 1440 = 24h)
//! per metric, in daemon memory. Used by the control socket
//! `show_stats_history` family and rendered as sparklines / braille plots
//! by `fipsctl` and `fipstop`. Lost on restart.
//!
//! Storage is split between node-level metrics (one ring per metric) and
//! per-peer metrics (one map `NodeAddr -> PeerStatsRings`, each holding
//! one ring per per-peer metric). Per-peer rings are back-filled with
//! NaN on first sight so every peer shares the same time axis with the
//! node-level rings. When a peer is absent from a tick, NaN is appended
//! to keep alignment. Peers are evicted once they have been absent from
//! every tick in the full 24h slow-ring window.
//!
//! Gap representation: `f64::NAN` for any sample where data is not
//! available (new peer back-fill, disconnected peer, MMP not yet
//! established, counter reset on link reconnect). NaN is serialized as
//! JSON `null` via a custom serializer.

use crate::identity::NodeAddr;
use serde::{Serialize, Serializer};
use std::collections::{HashMap, HashSet, VecDeque};
use std::str::FromStr;
use std::time::{Duration, Instant};

/// Fast-ring capacity: 3600 seconds = 1 hour at 1s resolution.
pub const FAST_RING_CAPACITY: usize = 3600;

/// Slow-ring capacity: 1440 minutes = 24 hours at 1m resolution.
pub const SLOW_RING_CAPACITY: usize = 1440;

/// Downsample window: how many fast samples fold into one slow sample.
pub const DOWNSAMPLE_FACTOR: usize = 60;

/// Evict peers that have been silent for at least this long.
pub const PEER_EVICTION_SECS: u64 = 24 * 3600;

/// Node-level metrics tracked in the history. Keep this list in sync
/// with `ALL_METRICS` and with the snapshot construction in
/// [`StatsHistory::tick`].
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum Metric {
    MeshSize,
    TreeDepth,
    PeerCount,
    ParentSwitches,
    BytesIn,
    BytesOut,
    PacketsIn,
    PacketsOut,
    LossRate,
    ActiveSessions,
}

/// Every node-level metric tracked, in a stable order (for enumeration
/// via `stats list` and for Graphs-tab cycling).
pub const ALL_METRICS: &[Metric] = &[
    Metric::MeshSize,
    Metric::TreeDepth,
    Metric::PeerCount,
    Metric::ParentSwitches,
    Metric::BytesIn,
    Metric::BytesOut,
    Metric::PacketsIn,
    Metric::PacketsOut,
    Metric::LossRate,
    Metric::ActiveSessions,
];

/// Per-peer metrics tracked in the history (one ring per metric, per peer).
/// Names collide with some `Metric` variants because the two live in
/// separate namespaces on the wire — a query is per-peer iff `peer` is
/// specified in the request.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum PeerMetric {
    SrttMs,
    LossRate,
    BytesIn,
    BytesOut,
    PacketsIn,
    PacketsOut,
    EcnCe,
}

pub const ALL_PEER_METRICS: &[PeerMetric] = &[
    PeerMetric::SrttMs,
    PeerMetric::LossRate,
    PeerMetric::BytesIn,
    PeerMetric::BytesOut,
    PeerMetric::PacketsIn,
    PeerMetric::PacketsOut,
    PeerMetric::EcnCe,
];

/// How a metric reduces a window of fast samples into one slow sample.
/// NaN samples are excluded from all reductions; a window of entirely
/// NaN samples produces NaN.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Aggregation {
    /// Keep the last non-NaN value.
    Last,
    /// Sum non-NaN values.
    Sum,
    /// Mean of non-NaN values.
    Mean,
}

impl Metric {
    pub fn name(self) -> &'static str {
        match self {
            Metric::MeshSize => "mesh_size",
            Metric::TreeDepth => "tree_depth",
            Metric::PeerCount => "peer_count",
            Metric::ParentSwitches => "parent_switches",
            Metric::BytesIn => "bytes_in",
            Metric::BytesOut => "bytes_out",
            Metric::PacketsIn => "packets_in",
            Metric::PacketsOut => "packets_out",
            Metric::LossRate => "loss_rate",
            Metric::ActiveSessions => "active_sessions",
        }
    }

    pub fn unit(self) -> &'static str {
        match self {
            Metric::MeshSize => "nodes",
            Metric::TreeDepth => "hops",
            Metric::PeerCount => "peers",
            Metric::ParentSwitches => "events/s",
            Metric::BytesIn | Metric::BytesOut => "bytes/s",
            Metric::PacketsIn | Metric::PacketsOut => "packets/s",
            Metric::LossRate => "fraction",
            Metric::ActiveSessions => "sessions",
        }
    }

    pub fn aggregation(self) -> Aggregation {
        match self {
            Metric::MeshSize | Metric::TreeDepth | Metric::PeerCount | Metric::ActiveSessions => {
                Aggregation::Last
            }
            Metric::ParentSwitches => Aggregation::Sum,
            Metric::BytesIn
            | Metric::BytesOut
            | Metric::PacketsIn
            | Metric::PacketsOut
            | Metric::LossRate => Aggregation::Mean,
        }
    }
}

impl FromStr for Metric {
    type Err = String;
    fn from_str(s: &str) -> Result<Self, Self::Err> {
        for m in ALL_METRICS {
            if m.name() == s {
                return Ok(*m);
            }
        }
        Err(format!("unknown metric: {s}"))
    }
}

impl PeerMetric {
    pub fn name(self) -> &'static str {
        match self {
            PeerMetric::SrttMs => "srtt_ms",
            PeerMetric::LossRate => "loss_rate",
            PeerMetric::BytesIn => "bytes_in",
            PeerMetric::BytesOut => "bytes_out",
            PeerMetric::PacketsIn => "packets_in",
            PeerMetric::PacketsOut => "packets_out",
            PeerMetric::EcnCe => "ecn_ce",
        }
    }

    pub fn unit(self) -> &'static str {
        match self {
            PeerMetric::SrttMs => "ms",
            PeerMetric::LossRate => "fraction",
            PeerMetric::BytesIn | PeerMetric::BytesOut => "bytes/s",
            PeerMetric::PacketsIn | PeerMetric::PacketsOut => "packets/s",
            PeerMetric::EcnCe => "events/s",
        }
    }

    pub fn aggregation(self) -> Aggregation {
        match self {
            PeerMetric::SrttMs => Aggregation::Mean,
            PeerMetric::LossRate => Aggregation::Mean,
            PeerMetric::BytesIn
            | PeerMetric::BytesOut
            | PeerMetric::PacketsIn
            | PeerMetric::PacketsOut => Aggregation::Mean,
            PeerMetric::EcnCe => Aggregation::Sum,
        }
    }

    /// Whether this metric is derived from a monotonic counter (sample =
    /// delta per tick, reset to NaN if the counter decreases).
    pub fn is_counter(self) -> bool {
        matches!(
            self,
            PeerMetric::BytesIn
                | PeerMetric::BytesOut
                | PeerMetric::PacketsIn
                | PeerMetric::PacketsOut
                | PeerMetric::EcnCe
        )
    }
}

impl FromStr for PeerMetric {
    type Err = String;
    fn from_str(s: &str) -> Result<Self, Self::Err> {
        for m in ALL_PEER_METRICS {
            if m.name() == s {
                return Ok(*m);
            }
        }
        Err(format!("unknown peer metric: {s}"))
    }
}

/// Snapshot of raw node-level counter state used to derive per-tick
/// samples. Produced by `Node` and passed into [`StatsHistory::tick`].
#[derive(Clone, Copy, Debug)]
pub struct Snapshot {
    pub mesh_size: Option<u64>,
    pub tree_depth: u32,
    pub peer_count: u64,
    pub parent_switches_total: u64,
    pub bytes_in_total: u64,
    pub bytes_out_total: u64,
    pub packets_in_total: u64,
    pub packets_out_total: u64,
    pub loss_rate: f64,
    pub active_sessions: u64,
}

/// Snapshot of one peer's state at the current tick. An entry missing
/// from the `peers` slice of [`StatsHistory::tick`] is treated as "peer
/// absent this tick" and backs NaN into each of its rings.
#[derive(Clone, Debug)]
pub struct PeerSnapshot {
    pub node_addr: NodeAddr,
    pub last_seen: Instant,
    /// MMP SRTT; `None` when no MMP measurement exists yet.
    pub srtt_ms: Option<f64>,
    /// MMP loss rate; `None` when the peer has no MMP session yet.
    pub loss_rate: Option<f64>,
    /// Monotonic counters. May decrease when the peer reconnects on a
    /// new link (fresh LinkStats); that's detected per-ring and emits
    /// NaN for the affected tick.
    pub bytes_in_total: u64,
    pub bytes_out_total: u64,
    pub packets_in_total: u64,
    pub packets_out_total: u64,
    pub ecn_ce_total: u64,
}

/// Which ring a query reads from.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Granularity {
    /// 1-second samples from the fast ring.
    Fast,
    /// 1-minute samples from the slow ring.
    Slow,
}

impl Granularity {
    pub fn seconds(self) -> u64 {
        match self {
            Granularity::Fast => 1,
            Granularity::Slow => 60,
        }
    }
}

impl FromStr for Granularity {
    type Err = String;
    fn from_str(s: &str) -> Result<Self, Self::Err> {
        match s {
            "1s" => Ok(Granularity::Fast),
            "1m" => Ok(Granularity::Slow),
            other => Err(format!("unknown granularity: {other} (expected 1s or 1m)")),
        }
    }
}

/// One metric's dual-tier ring.
struct Ring {
    fast: VecDeque<f64>,
    slow: VecDeque<f64>,
    /// Accumulator used for downsampling fast → slow on minute boundaries.
    accum: DownsampleAccum,
    aggregation: Aggregation,
    /// Value from the previous tick, used to derive deltas for counter
    /// metrics. `None` means "first sample upcoming" and produces NaN.
    prev_total: Option<u64>,
}

/// Running accumulator over up to `DOWNSAMPLE_FACTOR` fast samples.
/// NaN samples are skipped from all statistics; `total` still tracks
/// them so we know whether ANY sample arrived this window.
struct DownsampleAccum {
    sum: f64,
    /// Count of non-NaN samples.
    count: u32,
    /// Most recent non-NaN sample, or NaN if none.
    last: f64,
    /// Total samples observed (including NaN).
    total: u32,
}

impl DownsampleAccum {
    fn new() -> Self {
        Self {
            sum: 0.0,
            count: 0,
            last: f64::NAN,
            total: 0,
        }
    }

    fn push(&mut self, v: f64) {
        self.total += 1;
        if !v.is_nan() {
            self.sum += v;
            self.count += 1;
            self.last = v;
        }
    }

    fn reduce(&self, agg: Aggregation) -> Option<f64> {
        if self.total == 0 {
            return None;
        }
        if self.count == 0 {
            return Some(f64::NAN);
        }
        Some(match agg {
            Aggregation::Last => self.last,
            Aggregation::Sum => self.sum,
            Aggregation::Mean => self.sum / self.count as f64,
        })
    }

    fn reset(&mut self) {
        *self = Self::new();
    }
}

impl Ring {
    fn new(aggregation: Aggregation) -> Self {
        Self {
            fast: VecDeque::with_capacity(FAST_RING_CAPACITY),
            slow: VecDeque::with_capacity(SLOW_RING_CAPACITY),
            accum: DownsampleAccum::new(),
            aggregation,
            prev_total: None,
        }
    }

    fn push_fast(&mut self, value: f64) {
        if self.fast.len() == FAST_RING_CAPACITY {
            self.fast.pop_front();
        }
        self.fast.push_back(value);
        self.accum.push(value);
    }

    fn flush_slow(&mut self) {
        if let Some(v) = self.accum.reduce(self.aggregation) {
            if self.slow.len() == SLOW_RING_CAPACITY {
                self.slow.pop_front();
            }
            self.slow.push_back(v);
        }
        self.accum.reset();
    }
}

/// Helper: convert a monotonic counter into a per-tick delta. Returns
/// NaN when no previous sample exists (first observation) or when the
/// counter decreased (new link). Updates `prev_total` on every call so
/// the next tick's baseline is the current value.
fn delta_or_nan(ring: &mut Ring, total: u64) -> f64 {
    let prev = ring.prev_total;
    ring.prev_total = Some(total);
    match prev {
        None => f64::NAN,
        Some(p) if total < p => f64::NAN,
        Some(p) => (total - p) as f64,
    }
}

/// Custom serializer: NaN / infinity → JSON `null`; finite values pass
/// through as numbers.
fn serialize_nan_as_null<S: Serializer>(values: &[f64], s: S) -> Result<S::Ok, S::Error> {
    use serde::ser::SerializeSeq;
    let mut seq = s.serialize_seq(Some(values.len()))?;
    for &v in values {
        if v.is_finite() {
            seq.serialize_element(&v)?;
        } else {
            seq.serialize_element(&Option::<f64>::None)?;
        }
    }
    seq.end()
}

/// Query result — a contiguous series of samples newest-last.
/// Gap samples are NaN in memory and `null` in JSON.
#[derive(Clone, Debug, Serialize)]
pub struct Series {
    pub metric: &'static str,
    pub unit: &'static str,
    pub granularity_seconds: u64,
    #[serde(serialize_with = "serialize_nan_as_null")]
    pub values: Vec<f64>,
}

/// One peer's per-metric rings plus lifecycle metadata.
pub struct PeerStatsRings {
    rings: Vec<Ring>,
    first_seen: Instant,
    last_contact: Instant,
}

impl PeerStatsRings {
    fn new(now: Instant, fast_pushes_so_far: u64) -> Self {
        let mut rings: Vec<Ring> = ALL_PEER_METRICS
            .iter()
            .map(|m| Ring::new(m.aggregation()))
            .collect();

        // Back-fill NaN so this peer's rings share a time axis with the
        // node-level rings that have been collecting since start. We
        // fill up to (but not including) the slot this tick will take.
        let n_fast = (fast_pushes_so_far as usize).min(FAST_RING_CAPACITY);
        let n_slow = ((fast_pushes_so_far as usize) / DOWNSAMPLE_FACTOR).min(SLOW_RING_CAPACITY);
        for ring in &mut rings {
            for _ in 0..n_fast {
                ring.fast.push_back(f64::NAN);
            }
            for _ in 0..n_slow {
                ring.slow.push_back(f64::NAN);
            }
        }

        Self {
            rings,
            first_seen: now,
            last_contact: now,
        }
    }

    fn ring(&self, metric: PeerMetric) -> &Ring {
        let idx = ALL_PEER_METRICS.iter().position(|m| *m == metric).unwrap();
        &self.rings[idx]
    }

    fn ring_mut(&mut self, metric: PeerMetric) -> &mut Ring {
        let idx = ALL_PEER_METRICS.iter().position(|m| *m == metric).unwrap();
        &mut self.rings[idx]
    }

    fn push_sample(&mut self, snap: &PeerSnapshot, now: Instant) {
        self.last_contact = now;
        for &metric in ALL_PEER_METRICS {
            let value = match metric {
                PeerMetric::SrttMs => snap.srtt_ms.unwrap_or(f64::NAN),
                PeerMetric::LossRate => snap.loss_rate.unwrap_or(f64::NAN),
                PeerMetric::BytesIn => delta_or_nan(self.ring_mut(metric), snap.bytes_in_total),
                PeerMetric::BytesOut => delta_or_nan(self.ring_mut(metric), snap.bytes_out_total),
                PeerMetric::PacketsIn => delta_or_nan(self.ring_mut(metric), snap.packets_in_total),
                PeerMetric::PacketsOut => {
                    delta_or_nan(self.ring_mut(metric), snap.packets_out_total)
                }
                PeerMetric::EcnCe => delta_or_nan(self.ring_mut(metric), snap.ecn_ce_total),
            };
            self.ring_mut(metric).push_fast(value);
        }
    }

    /// Push NaN for every ring (peer was absent this tick). Also clears
    /// the counter baseline so the next real sample produces NaN rather
    /// than an inflated delta accumulated over the silence.
    fn push_nan(&mut self) {
        for (i, ring) in self.rings.iter_mut().enumerate() {
            ring.push_fast(f64::NAN);
            if ALL_PEER_METRICS[i].is_counter() {
                ring.prev_total = None;
            }
        }
    }

    fn flush_slow(&mut self) {
        for ring in &mut self.rings {
            ring.flush_slow();
        }
    }

    pub fn first_seen(&self) -> Instant {
        self.first_seen
    }

    pub fn last_contact(&self) -> Instant {
        self.last_contact
    }
}

/// Per-metric ring storage for node-level metrics plus a map of
/// per-peer rings keyed by `NodeAddr`.
pub struct StatsHistory {
    rings: Vec<Ring>,
    peers: HashMap<NodeAddr, PeerStatsRings>,
    /// Wall-clock anchor for 1-minute downsample boundaries. Set on the
    /// first tick; downsample fires when elapsed since the anchor crosses
    /// a multiple of 60s (coarsely — we just count fast pushes).
    fast_pushes: u64,
    /// Monotonic timestamp of the most recent tick, used by readers that
    /// want to label the series in wall-clock terms.
    last_tick: Option<Instant>,
}

impl StatsHistory {
    pub fn new() -> Self {
        let rings = ALL_METRICS
            .iter()
            .map(|m| Ring::new(m.aggregation()))
            .collect();
        Self {
            rings,
            peers: HashMap::new(),
            fast_pushes: 0,
            last_tick: None,
        }
    }

    fn ring_mut(&mut self, metric: Metric) -> &mut Ring {
        let idx = ALL_METRICS.iter().position(|m| *m == metric).unwrap();
        &mut self.rings[idx]
    }

    fn ring(&self, metric: Metric) -> &Ring {
        let idx = ALL_METRICS.iter().position(|m| *m == metric).unwrap();
        &self.rings[idx]
    }

    /// Record one tick. Should be invoked once per second from the node
    /// event loop, passing the latest snapshot and the set of peers
    /// observed this tick.
    ///
    /// Derives per-second rates from delta on counter totals; gauges
    /// are sampled directly. Every 60 pushes, the accumulator is
    /// flushed to the slow ring. Peers that have been absent for the
    /// full eviction window are dropped from the map.
    pub fn tick(&mut self, now: Instant, snapshot: &Snapshot, peers: &[PeerSnapshot]) {
        // Node-level metrics.
        for &metric in ALL_METRICS {
            let value = match metric {
                Metric::MeshSize => snapshot.mesh_size.unwrap_or(0) as f64,
                Metric::TreeDepth => snapshot.tree_depth as f64,
                Metric::PeerCount => snapshot.peer_count as f64,
                Metric::ParentSwitches => {
                    Self::node_delta(self.ring_mut(metric), snapshot.parent_switches_total)
                }
                Metric::BytesIn => Self::node_delta(self.ring_mut(metric), snapshot.bytes_in_total),
                Metric::BytesOut => {
                    Self::node_delta(self.ring_mut(metric), snapshot.bytes_out_total)
                }
                Metric::PacketsIn => {
                    Self::node_delta(self.ring_mut(metric), snapshot.packets_in_total)
                }
                Metric::PacketsOut => {
                    Self::node_delta(self.ring_mut(metric), snapshot.packets_out_total)
                }
                Metric::LossRate => snapshot.loss_rate,
                Metric::ActiveSessions => snapshot.active_sessions as f64,
            };
            self.ring_mut(metric).push_fast(value);
        }

        // Per-peer metrics.
        let mut seen: HashSet<NodeAddr> = HashSet::with_capacity(peers.len());
        for ps in peers {
            seen.insert(ps.node_addr);
            let entry = self
                .peers
                .entry(ps.node_addr)
                .or_insert_with(|| PeerStatsRings::new(now, self.fast_pushes));
            entry.push_sample(ps, now);
        }
        for (addr, rings) in self.peers.iter_mut() {
            if !seen.contains(addr) {
                rings.push_nan();
            }
        }

        self.fast_pushes += 1;
        if self.fast_pushes.is_multiple_of(DOWNSAMPLE_FACTOR as u64) {
            for ring in &mut self.rings {
                ring.flush_slow();
            }
            for rings in self.peers.values_mut() {
                rings.flush_slow();
            }
        }

        // Evict peers silent for at least PEER_EVICTION_SECS.
        let threshold = Duration::from_secs(PEER_EVICTION_SECS);
        self.peers
            .retain(|_, rings| now.duration_since(rings.last_contact) < threshold);

        self.last_tick = Some(now);
    }

    /// Helper: node-level monotonic counter → per-tick delta. Uses
    /// `saturating_sub` because node totals never reset; the defensive
    /// saturation matches the pre-per-peer behavior.
    fn node_delta(ring: &mut Ring, total: u64) -> f64 {
        let prev = ring.prev_total;
        ring.prev_total = Some(total);
        match prev {
            None => 0.0,
            Some(p) => total.saturating_sub(p) as f64,
        }
    }

    /// Answer a query for a single node-level metric across a given
    /// window and granularity. The returned series always has the full
    /// window width (clipped only to ring capacity); any samples older
    /// than the ring has seen are front-padded with NaN so each window
    /// renders at its chosen density.
    pub fn query(&self, metric: Metric, window: Duration, granularity: Granularity) -> Series {
        let ring = self.ring(metric);
        Self::build_series(ring, metric.name(), metric.unit(), window, granularity)
    }

    /// Answer a query for one peer's metric. Returns `None` if the peer
    /// is not tracked.
    pub fn peer_query(
        &self,
        addr: &NodeAddr,
        metric: PeerMetric,
        window: Duration,
        granularity: Granularity,
    ) -> Option<Series> {
        let rings = self.peers.get(addr)?;
        Some(Self::build_series(
            rings.ring(metric),
            metric.name(),
            metric.unit(),
            window,
            granularity,
        ))
    }

    fn build_series(
        ring: &Ring,
        name: &'static str,
        unit: &'static str,
        window: Duration,
        granularity: Granularity,
    ) -> Series {
        let (source, capacity): (&VecDeque<f64>, usize) = match granularity {
            Granularity::Fast => (&ring.fast, FAST_RING_CAPACITY),
            Granularity::Slow => (&ring.slow, SLOW_RING_CAPACITY),
        };

        let want = (window.as_secs() / granularity.seconds()) as usize;
        let want = want.min(capacity);
        let take = source.len().min(want);
        let tail: Vec<f64> = source.iter().rev().take(take).rev().copied().collect();
        let values = if tail.len() < want {
            let pad = want - tail.len();
            let mut out = Vec::with_capacity(want);
            out.resize(pad, f64::NAN);
            out.extend(tail);
            out
        } else {
            tail
        };

        Series {
            metric: name,
            unit,
            granularity_seconds: granularity.seconds(),
            values,
        }
    }

    /// Most recent node-level value for a metric, reading from the fast
    /// ring.
    pub fn latest(&self, metric: Metric) -> Option<f64> {
        self.ring(metric).fast.back().copied()
    }

    /// Return the last `n` node-level samples from the fast ring,
    /// oldest-first.
    pub fn recent(&self, metric: Metric, n: usize) -> Vec<f64> {
        let ring = self.ring(metric);
        let n = n.min(ring.fast.len());
        ring.fast.iter().rev().take(n).rev().copied().collect()
    }

    /// Iterate tracked peer addresses.
    pub fn peer_addrs(&self) -> impl Iterator<Item = &NodeAddr> {
        self.peers.keys()
    }

    /// Iterate tracked peers with their ring metadata.
    pub fn peers(&self) -> impl Iterator<Item = (&NodeAddr, &PeerStatsRings)> {
        self.peers.iter()
    }

    /// Number of tracked peers (includes recently-disconnected within
    /// the 24h retention window).
    pub fn tracked_peer_count(&self) -> usize {
        self.peers.len()
    }

    /// Whether this peer is currently in the tracking map (has been
    /// seen at some point and not yet evicted).
    pub fn has_peer(&self, addr: &NodeAddr) -> bool {
        self.peers.contains_key(addr)
    }

    /// Whether tick() has ever been called.
    pub fn has_data(&self) -> bool {
        self.last_tick.is_some()
    }
}

impl Default for StatsHistory {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(test)]
mod tests;