From 5abae0859e54a350e36f07a240f7867b75ff533c Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Tue, 14 Apr 2026 10:24:16 +0100 Subject: [PATCH] Add historical node and per-peer statistics with btop-style graphs (#64) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit In-memory time-series history on the daemon: fast ring (1s × 3600) plus slow ring (1m × 1440) per metric, covering node-level gauges (mesh size, tree depth, peer count, active sessions), counters (parent switches, aggregate bytes/packets in/out), loss rate, and seven per-peer metrics keyed by NodeAddr (srtt_ms, loss_rate, bytes_in/out, packets_in/out, ecn_ce). The slow ring is produced by downsampling the fast ring on minute boundaries with Last / Sum / Mean aggregation chosen per metric type. Missing data is first-class. New peers back-fill NaN so every ring shares a time axis with the node rings; peers absent from a tick sample NaN (keeps alignment, shows as a visible gap); counter metrics emit NaN on decrease (new link_stats baseline after reconnect) so deltas aren't polluted. Peers are evicted 24h after last contact. Downsampling is NaN-aware: mean skips NaN, all-NaN slow windows stay NaN. Each history window always returns its full span at the chosen density (1m / 10m / 1h / 24h), front-padded with NaN when the ring hasn't yet accumulated enough samples, so switching between windows feels like zooming in or out rather than clipping to whatever has arrived. NaN serializes to JSON null via a custom serializer. Control socket queries: - show_stats_list enumerates registered metrics plus scope field and peer_retention_seconds. - show_stats_history returns one metric's series for a given window and granularity; accepts optional peer (npub) for per-peer metrics. - show_stats_all_history returns every metric in a single round trip; accepts optional peer to fetch all seven per-peer metrics. - show_stats_peers enumerates tracked peers with lifecycle metadata. - show_stats_history_all_peers returns one metric across all peers for grid rendering. - show_status carries short sparkline windows for the dashboard so the client can render without extra fetches. fipsctl gains `stats list`, `stats peers`, and `stats history ` with `--peer` (hostname or npub) and `--plot` for a Unicode block sparkline. Plot header reports sample count, granularity, window, and gap count; NaN renders as a blank cell. fipstop dashboard grows inline sparklines (peer count, mesh size, aggregate bytes in/out). A new Graphs tab stacks every metric as an independent mini plot with its own autoscaled range; each plot uses btop's braille 2×4 filled-area algorithm (25-entry lookup table packing two samples per character, per-row gradient coloring for the characteristic btop vertical-band look, rounded borders with embedded titles). Three modes are cycled with `m`: Node (node-level stack), MetricByPeer (small-multiples grid, 1 / 2 / 3 columns by terminal width), PeerByMetric (existing stack scoped to one peer). `n` / `N` cycles the mode-specific selector (metric or peer), a selector row shows the current choice, and Graphs-tab refreshes re-fetch show_stats_peers so selectors track peer churn. Up / Down scrolls the stack, Left / Right cycles the window, `g` jumps to the tab. Implements IDEA-0084 (TASK-2026-0062). --- src/bin/fipsctl.rs | 174 +++++ src/bin/fipstop/app.rs | 167 ++++- src/bin/fipstop/client.rs | 11 +- src/bin/fipstop/main.rs | 133 +++- src/bin/fipstop/ui/dashboard.rs | 27 +- src/bin/fipstop/ui/graphs.rs | 634 ++++++++++++++++ src/bin/fipstop/ui/helpers.rs | 39 + src/bin/fipstop/ui/mod.rs | 2 + src/control/mod.rs | 1 + src/control/queries.rs | 368 +++++++++- src/node/handlers/rx_loop.rs | 3 +- src/node/mod.rs | 70 ++ src/node/stats_history.rs | 1218 +++++++++++++++++++++++++++++++ 13 files changed, 2834 insertions(+), 13 deletions(-) create mode 100644 src/bin/fipstop/ui/graphs.rs create mode 100644 src/node/stats_history.rs diff --git a/src/bin/fipsctl.rs b/src/bin/fipsctl.rs index 799d336..3dc2376 100644 --- a/src/bin/fipsctl.rs +++ b/src/bin/fipsctl.rs @@ -66,6 +66,38 @@ enum Commands { /// Peer identifier: npub (bech32) or hostname from /etc/fips/hosts peer: String, }, + /// Query historical node statistics + Stats { + #[command(subcommand)] + what: StatsCommands, + }, +} + +#[derive(Subcommand, Debug)] +enum StatsCommands { + /// List available history metrics + List, + /// List peers tracked in the stats history + Peers, + /// Fetch a time-series window for a metric + History { + /// Metric name (see `fipsctl stats list`). Node-level metrics + /// need no `--peer`; per-peer metrics require it. + metric: String, + /// Peer npub (bech32) or hostname from /etc/fips/hosts for + /// per-peer metrics + #[arg(long)] + peer: Option, + /// Window duration — `s`, `m`, `h` + #[arg(long, default_value = "10m")] + window: String, + /// Sample resolution — `1s` (fast ring) or `1m` (slow ring) + #[arg(long, default_value = "1s")] + granularity: String, + /// Render a Unicode block sparkline instead of JSON + #[arg(long)] + plot: bool, + }, } #[derive(Subcommand, Debug)] @@ -394,9 +426,49 @@ fn main() { let npub = resolve_peer(peer); build_command("disconnect", serde_json::json!({"npub": npub})) } + Commands::Stats { what } => match what { + StatsCommands::List => build_query("show_stats_list"), + StatsCommands::Peers => build_query("show_stats_peers"), + StatsCommands::History { + metric, + peer, + window, + granularity, + .. + } => { + let mut params = serde_json::json!({ + "metric": metric, + "window": window, + "granularity": granularity, + }); + if let Some(p) = peer { + let resolved = resolve_peer(p); + params["peer"] = serde_json::json!(resolved); + } + build_command("show_stats_history", params) + } + }, Commands::Keygen { .. } => unreachable!(), }; + // For plot output we need to post-process the JSON response rather + // than pretty-print it. + if let Commands::Stats { + what: StatsCommands::History { + plot: true, metric, .. + }, + } = &cli.command + { + match send_request(&socket_path, &request) { + Ok(value) => print_plot(&value, metric), + Err(e) => { + eprintln!("error: {e}"); + std::process::exit(1); + } + } + return; + } + match send_request(&socket_path, &request) { Ok(value) => print_response(&value), Err(e) => { @@ -406,6 +478,108 @@ fn main() { } } +/// Render the response as a Unicode block sparkline plot. +fn print_plot(value: &serde_json::Value, metric: &str) { + let status = value + .get("status") + .and_then(|v| v.as_str()) + .unwrap_or("unknown"); + if status == "error" { + let msg = value + .get("message") + .and_then(|v| v.as_str()) + .unwrap_or("unknown error"); + eprintln!("error: {msg}"); + std::process::exit(1); + } + + let data = match value.get("data") { + Some(d) => d, + None => { + eprintln!("error: no data in response"); + std::process::exit(1); + } + }; + + let values: Vec = data + .get("values") + .and_then(|v| v.as_array()) + .map(|arr| arr.iter().map(|v| v.as_f64().unwrap_or(f64::NAN)).collect()) + .unwrap_or_default(); + let unit = data.get("unit").and_then(|v| v.as_str()).unwrap_or(""); + let granularity_seconds = data + .get("granularity_seconds") + .and_then(|v| v.as_u64()) + .unwrap_or(1); + + if values.is_empty() { + println!("{metric}: no data yet"); + return; + } + + let (min, max) = values + .iter() + .filter(|v| !v.is_nan()) + .fold((f64::INFINITY, f64::NEG_INFINITY), |(lo, hi), &v| { + (lo.min(v), hi.max(v)) + }); + let (min, max) = if min.is_finite() { + (min, max) + } else { + (0.0, 0.0) + }; + let last = values + .iter() + .rev() + .find(|v| !v.is_nan()) + .copied() + .unwrap_or(f64::NAN); + let width_secs = (values.len() as u64) * granularity_seconds; + let gap_count = values.iter().filter(|v| v.is_nan()).count(); + + println!( + "{metric} ({unit}) — {n} samples @ {g}s = {w}s window{gap}", + n = values.len(), + g = granularity_seconds, + w = width_secs, + gap = if gap_count > 0 { + format!(" ({gap_count} gaps)") + } else { + String::new() + }, + ); + let last_str = if last.is_nan() { + "-".to_string() + } else { + format!("{last:.3}") + }; + println!(" min={min:.3} max={max:.3} last={last_str}"); + println!(" {}", sparkline(&values, min, max)); +} + +/// Render a slice of values as Unicode block characters. +/// +/// Uses eight discrete levels: `▁▂▃▄▅▆▇█`. Constant series and empty +/// inputs render as a single-level line (`▄`). +fn sparkline(values: &[f64], min: f64, max: f64) -> String { + const BLOCKS: [char; 8] = ['▁', '▂', '▃', '▄', '▅', '▆', '▇', '█']; + let range = max - min; + values + .iter() + .map(|&v| { + if v.is_nan() { + ' ' + } else if !range.is_finite() || range <= 0.0 { + BLOCKS[3] + } else { + let norm = ((v - min) / range).clamp(0.0, 1.0); + let idx = (norm * (BLOCKS.len() as f64 - 1.0)).round() as usize; + BLOCKS[idx.min(BLOCKS.len() - 1)] + } + }) + .collect() +} + #[cfg(test)] mod tests { use super::*; diff --git a/src/bin/fipstop/app.rs b/src/bin/fipstop/app.rs index 93dd763..7d0f906 100644 --- a/src/bin/fipstop/app.rs +++ b/src/bin/fipstop/app.rs @@ -15,10 +15,11 @@ pub enum Tab { Transports, Routing, Gateway, + Graphs, } impl Tab { - pub const ALL: [Tab; 9] = [ + pub const ALL: [Tab; 10] = [ Tab::Node, Tab::Peers, Tab::Transports, @@ -27,6 +28,7 @@ impl Tab { Tab::Bloom, Tab::Mmp, Tab::Routing, + Tab::Graphs, Tab::Gateway, ]; @@ -36,6 +38,7 @@ impl Tab { Tab::Node => 0, Tab::Peers | Tab::Transports => 1, Tab::Gateway => 3, + Tab::Graphs => 2, _ => 2, } } @@ -53,6 +56,7 @@ impl Tab { Tab::Transports => "Transports", Tab::Routing => "Routing", Tab::Gateway => "Gateway", + Tab::Graphs => "Graphs", } } @@ -69,6 +73,10 @@ impl Tab { Tab::Transports => "show_transports", Tab::Routing => "show_routing", Tab::Gateway => "show_gateway", + // Graphs uses show_stats_history with params; fetched via a + // dedicated path in main.rs rather than the generic command() + // dispatcher. + Tab::Graphs => "show_stats_history", } } @@ -124,6 +132,67 @@ pub enum SelectedTreeItem { Link, } +/// Options for the Graphs tab window selector. +pub const GRAPHS_WINDOWS: &[(&str, &str)] = + &[("1m", "1s"), ("10m", "1s"), ("1h", "1s"), ("24h", "1m")]; + +/// Node-level metric display order for Graphs tab Node mode. Must +/// match names returned by `show_stats_all_history` (no `peer` param). +pub const GRAPHS_METRICS: &[&str] = &[ + "mesh_size", + "tree_depth", + "peer_count", + "parent_switches", + "bytes_in", + "bytes_out", + "packets_in", + "packets_out", + "loss_rate", + "active_sessions", +]; + +/// Per-peer metric display order for Graphs tab PeerByMetric mode and +/// MetricByPeer selector. Must match names returned by +/// `show_stats_all_history` with a `peer` param. +pub const PEER_GRAPHS_METRICS: &[&str] = &[ + "srtt_ms", + "loss_rate", + "bytes_in", + "bytes_out", + "packets_in", + "packets_out", + "ecn_ce", +]; + +/// Which variety of plot the Graphs tab shows. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum GraphsMode { + /// Stacked node-level metrics (the original view). + Node, + /// Grid: one metric, small-multiples across peers. + MetricByPeer, + /// Stacked per-peer metrics for one selected peer. + PeerByMetric, +} + +impl GraphsMode { + pub fn next(self) -> Self { + match self { + GraphsMode::Node => GraphsMode::MetricByPeer, + GraphsMode::MetricByPeer => GraphsMode::PeerByMetric, + GraphsMode::PeerByMetric => GraphsMode::Node, + } + } +} + +/// Cached peer summary for Graphs-tab selector (peer-list population +/// is independent of the per-tick metric data). +#[derive(Clone, Debug)] +pub struct GraphsPeer { + pub npub: String, + pub display_name: String, +} + pub struct App { pub active_tab: Tab, pub should_quit: bool, @@ -141,6 +210,20 @@ pub struct App { pub gateway_running: bool, /// Mappings data fetched from the gateway (separate from summary). pub gateway_mappings: Option, + /// Scroll offset (rows) for the stacked Graphs tab. + pub graphs_scroll: u16, + /// Selected (window, granularity) index for the Graphs tab. + pub graphs_window_idx: usize, + /// Current Graphs-tab view mode. + pub graphs_mode: GraphsMode, + /// Selected metric index for MetricByPeer mode (into + /// `PEER_GRAPHS_METRICS`). + pub graphs_peer_metric_idx: usize, + /// Selected peer index for PeerByMetric mode (into `graphs_peers`). + pub graphs_peer_idx: usize, + /// Cached peer list from `show_stats_peers`, populated when the + /// Graphs tab is active in a non-Node mode. + pub graphs_peers: Vec, } impl App { @@ -160,9 +243,91 @@ impl App { selected_tree_item: SelectedTreeItem::None, gateway_running: false, gateway_mappings: None, + graphs_scroll: 0, + graphs_window_idx: 1, // default 10m + graphs_mode: GraphsMode::Node, + graphs_peer_metric_idx: 0, + graphs_peer_idx: 0, + graphs_peers: Vec::new(), } } + /// Cycle the Graphs-tab view mode. + pub fn graphs_next_mode(&mut self) { + self.graphs_mode = self.graphs_mode.next(); + self.graphs_scroll = 0; + } + + /// Advance the mode-specific selector (metric or peer). + pub fn graphs_next_selector(&mut self) { + match self.graphs_mode { + GraphsMode::Node => {} + GraphsMode::MetricByPeer => { + let n = PEER_GRAPHS_METRICS.len(); + self.graphs_peer_metric_idx = (self.graphs_peer_metric_idx + 1) % n; + } + GraphsMode::PeerByMetric => { + let n = self.graphs_peers.len(); + if n > 0 { + self.graphs_peer_idx = (self.graphs_peer_idx + 1) % n; + } + } + } + } + + /// Reverse the mode-specific selector. + pub fn graphs_prev_selector(&mut self) { + match self.graphs_mode { + GraphsMode::Node => {} + GraphsMode::MetricByPeer => { + let n = PEER_GRAPHS_METRICS.len(); + self.graphs_peer_metric_idx = (self.graphs_peer_metric_idx + n - 1) % n; + } + GraphsMode::PeerByMetric => { + let n = self.graphs_peers.len(); + if n > 0 { + self.graphs_peer_idx = (self.graphs_peer_idx + n - 1) % n; + } + } + } + } + + /// Current per-peer metric name for MetricByPeer mode. + pub fn graphs_selected_peer_metric(&self) -> &'static str { + PEER_GRAPHS_METRICS[self.graphs_peer_metric_idx % PEER_GRAPHS_METRICS.len()] + } + + /// Current selected peer for PeerByMetric mode, if any. + pub fn graphs_selected_peer(&self) -> Option<&GraphsPeer> { + if self.graphs_peers.is_empty() { + return None; + } + let idx = self.graphs_peer_idx % self.graphs_peers.len(); + Some(&self.graphs_peers[idx]) + } + + /// Current Graphs-tab (window, granularity) pair. + pub fn graphs_window(&self) -> (&'static str, &'static str) { + GRAPHS_WINDOWS[self.graphs_window_idx % GRAPHS_WINDOWS.len()] + } + + pub fn graphs_scroll_up(&mut self) { + self.graphs_scroll = self.graphs_scroll.saturating_sub(1); + } + + pub fn graphs_scroll_down(&mut self) { + self.graphs_scroll = self.graphs_scroll.saturating_add(1); + } + + pub fn graphs_next_window(&mut self) { + self.graphs_window_idx = (self.graphs_window_idx + 1) % GRAPHS_WINDOWS.len(); + } + + pub fn graphs_prev_window(&mut self) { + self.graphs_window_idx = + (self.graphs_window_idx + GRAPHS_WINDOWS.len() - 1) % GRAPHS_WINDOWS.len(); + } + /// Number of rows in the active tab's data array. pub fn row_count(&self) -> usize { if self.active_tab == Tab::Transports { diff --git a/src/bin/fipstop/client.rs b/src/bin/fipstop/client.rs index fa5b6cb..f241874 100644 --- a/src/bin/fipstop/client.rs +++ b/src/bin/fipstop/client.rs @@ -18,11 +18,20 @@ impl ControlClient { } pub async fn query(&self, command: &str) -> Result { + self.send(&format!("{{\"command\":\"{command}\"}}\n")).await + } + + pub async fn query_with_params(&self, command: &str, params: Value) -> Result { + let req = serde_json::json!({"command": command, "params": params}); + let line = format!("{}\n", serde_json::to_string(&req).unwrap()); + self.send(&line).await + } + + async fn send(&self, request: &str) -> Result { let stream = self.connect().await?; let (reader, mut writer) = tokio::io::split(stream); - let request = format!("{{\"command\":\"{command}\"}}\n"); timeout(IO_TIMEOUT, writer.write_all(request.as_bytes())) .await .map_err(|_| "write timed out".to_string())? diff --git a/src/bin/fipstop/main.rs b/src/bin/fipstop/main.rs index a599526..434cd43 100644 --- a/src/bin/fipstop/main.rs +++ b/src/bin/fipstop/main.rs @@ -95,12 +95,97 @@ fn fetch_data( // Fetch active tab data (if not Dashboard, which we already fetched) if app.active_tab != Tab::Node { - match rt.block_on(client.query(app.active_tab.command())) { - Ok(data) => { - app.data.insert(app.active_tab, data); + // Graphs tab pulls all metrics in one round trip via + // show_stats_all_history; all other tabs use the generic + // command() path. + if app.active_tab == Tab::Graphs { + let (window, granularity) = app.graphs_window(); + + // Always refresh the peer list on Graphs-tab fetches so + // selector indices stay current across peer churn. + if let Ok(data) = rt.block_on(client.query("show_stats_peers")) { + app.graphs_peers = data + .get("peers") + .and_then(|v| v.as_array()) + .map(|arr| { + arr.iter() + .map(|p| crate::app::GraphsPeer { + npub: p + .get("npub") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(), + display_name: p + .get("display_name") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(), + }) + .collect() + }) + .unwrap_or_default(); + if app.graphs_peer_idx >= app.graphs_peers.len().max(1) { + app.graphs_peer_idx = 0; + } } - Err(e) => { - app.last_error = Some((std::time::Instant::now(), e)); + + let (command, params) = match app.graphs_mode { + crate::app::GraphsMode::Node => ( + "show_stats_all_history", + serde_json::json!({ + "window": window, + "granularity": granularity, + }), + ), + crate::app::GraphsMode::MetricByPeer => ( + "show_stats_history_all_peers", + serde_json::json!({ + "metric": app.graphs_selected_peer_metric(), + "window": window, + "granularity": granularity, + }), + ), + crate::app::GraphsMode::PeerByMetric => { + let npub = app + .graphs_selected_peer() + .map(|p| p.npub.clone()) + .unwrap_or_default(); + ( + "show_stats_all_history", + serde_json::json!({ + "peer": npub, + "window": window, + "granularity": granularity, + }), + ) + } + }; + + // Only issue the per-peer query if there is a peer to query. + let should_query = match app.graphs_mode { + crate::app::GraphsMode::PeerByMetric => !app.graphs_peers.is_empty(), + _ => true, + }; + if should_query { + match rt.block_on(client.query_with_params(command, params)) { + Ok(data) => { + app.data.insert(Tab::Graphs, data); + } + Err(e) => { + app.last_error = Some((std::time::Instant::now(), e)); + } + } + } else { + app.data.insert(Tab::Graphs, serde_json::json!({})); + } + } else { + match rt.block_on(client.query(app.active_tab.command())) { + Ok(data) => { + app.data.insert(app.active_tab, data); + } + Err(e) => { + app.last_error = Some((std::time::Instant::now(), e)); + } } } } @@ -193,6 +278,8 @@ fn main() { (KeyCode::Down, _) => { if app.detail_view.is_some() { app.scroll_detail_down(); + } else if app.active_tab == Tab::Graphs { + app.graphs_scroll_down(); } else if app.active_tab.has_table() { app.select_next(); } @@ -200,6 +287,8 @@ fn main() { (KeyCode::Up, _) => { if app.detail_view.is_some() { app.scroll_detail_up(); + } else if app.active_tab == Tab::Graphs { + app.graphs_scroll_up(); } else if app.active_tab.has_table() { app.select_prev(); } @@ -210,7 +299,10 @@ fn main() { } } (KeyCode::Char(' '), _) | (KeyCode::Right, _) => { - if app.active_tab == Tab::Transports + if app.active_tab == Tab::Graphs && app.detail_view.is_none() { + app.graphs_next_window(); + fetch_data(&rt, &client, &gateway_client, &mut app); + } else if app.active_tab == Tab::Transports && app.detail_view.is_none() && let SelectedTreeItem::Transport(tid) = app.selected_tree_item { @@ -221,8 +313,29 @@ fn main() { } } } + (KeyCode::Char('m'), KeyModifiers::NONE) + if app.active_tab == Tab::Graphs && app.detail_view.is_none() => + { + app.graphs_next_mode(); + fetch_data(&rt, &client, &gateway_client, &mut app); + } + (KeyCode::Char('n'), KeyModifiers::NONE) + if app.active_tab == Tab::Graphs && app.detail_view.is_none() => + { + app.graphs_next_selector(); + fetch_data(&rt, &client, &gateway_client, &mut app); + } + (KeyCode::Char('N'), KeyModifiers::SHIFT) + if app.active_tab == Tab::Graphs && app.detail_view.is_none() => + { + app.graphs_prev_selector(); + fetch_data(&rt, &client, &gateway_client, &mut app); + } (KeyCode::Left, _) => { - if app.active_tab == Tab::Transports + if app.active_tab == Tab::Graphs && app.detail_view.is_none() { + app.graphs_prev_window(); + fetch_data(&rt, &client, &gateway_client, &mut app); + } else if app.active_tab == Tab::Transports && app.detail_view.is_none() && let SelectedTreeItem::Transport(tid) = app.selected_tree_item { @@ -251,6 +364,12 @@ fn main() { app.expanded_transports.clear(); } } + (KeyCode::Char('g'), KeyModifiers::NONE) => { + app.close_detail(); + app.active_tab = Tab::Graphs; + app.graphs_scroll = 0; + fetch_data(&rt, &client, &gateway_client, &mut app); + } _ => {} } } diff --git a/src/bin/fipstop/ui/dashboard.rs b/src/bin/fipstop/ui/dashboard.rs index a0af958..b723d35 100644 --- a/src/bin/fipstop/ui/dashboard.rs +++ b/src/bin/fipstop/ui/dashboard.rs @@ -22,7 +22,7 @@ pub fn draw(frame: &mut Frame, app: &App, area: Rect) { let chunks = Layout::vertical([ Constraint::Length(7), // Runtime Constraint::Length(7), // Identity - Constraint::Length(5), // State + Constraint::Length(6), // State (sparkline row adds one line) Constraint::Length(9), // Traffic Constraint::Min(0), // remaining ]) @@ -133,6 +133,12 @@ fn draw_state(frame: &mut Frame, data: &serde_json::Value, area: Rect) { .map(|n| format!("~{n}")) .unwrap_or_else(|| "-".into()); + let mesh_spark = + helpers::sparkline(&helpers::nested_f64_array(data, "sparklines", "mesh_size")); + let peer_spark = + helpers::sparkline(&helpers::nested_f64_array(data, "sparklines", "peer_count")); + let spark_style = Style::default().fg(Color::DarkGray); + let lines = vec![ Line::from(vec![ Span::styled(" state: ", label), @@ -158,6 +164,12 @@ fn draw_state(frame: &mut Frame, data: &serde_json::Value, area: Rect) { Span::styled(" mesh: ", label), Span::styled(mesh_size, count), ]), + Line::from(vec![ + Span::styled(" peers: ", label), + Span::styled(peer_spark, spark_style), + Span::styled(" mesh: ", label), + Span::styled(mesh_spark, spark_style), + ]), ]; frame.render_widget(Paragraph::new(lines), inner); @@ -168,6 +180,13 @@ fn draw_node_stats(frame: &mut Frame, data: &serde_json::Value, area: Rect) { let inner = block.inner(area); frame.render_widget(block, area); + let spark_style = Style::default().fg(Color::DarkGray); + let label = Style::default().fg(Color::DarkGray); + let bytes_in_spark = + helpers::sparkline(&helpers::nested_f64_array(data, "sparklines", "bytes_in")); + let bytes_out_spark = + helpers::sparkline(&helpers::nested_f64_array(data, "sparklines", "bytes_out")); + let lines = vec![ helpers::section_header("TUN (IPv6)"), fwd_line(data, "Tx", "delivered_packets", "delivered_bytes"), @@ -175,6 +194,12 @@ fn draw_node_stats(frame: &mut Frame, data: &serde_json::Value, area: Rect) { Line::from(""), helpers::section_header("Forwarded (transit)"), fwd_line(data, "Packets", "forwarded_packets", "forwarded_bytes"), + Line::from(vec![ + Span::styled(" rate in: ", label), + Span::styled(bytes_in_spark, spark_style), + Span::styled(" rate out: ", label), + Span::styled(bytes_out_spark, spark_style), + ]), ]; frame.render_widget(Paragraph::new(lines), inner); diff --git a/src/bin/fipstop/ui/graphs.rs b/src/bin/fipstop/ui/graphs.rs new file mode 100644 index 0000000..f7b90b1 --- /dev/null +++ b/src/bin/fipstop/ui/graphs.rs @@ -0,0 +1,634 @@ +//! Graphs tab — stack of per-metric mini plots, btop-style. +//! +//! All node-level history metrics render as independent mini-graphs +//! stacked vertically: each has its own title line, its own +//! autoscaled min/max, and its own braille 2×4 filled-area plot +//! colored per-row on a green → yellow → red gradient. Content +//! scrolls with Up/Down; Left/Right cycles the (window, granularity) +//! pair which applies to the whole stack. +//! +//! Rendering follows btop's algorithm: 25-entry braille lookup table +//! `BRAILLE[left_level][right_level]` where each level is 0..=4, two +//! time samples per character cell, row-position-keyed gradient +//! coloring for the characteristic btop "vertical bands" look. + +use ratatui::Frame; +use ratatui::layout::{Constraint, Layout, Rect}; +use ratatui::style::{Color, Modifier, Style}; +use ratatui::text::{Line, Span}; +use ratatui::widgets::{Block, BorderType, Borders, Paragraph}; + +use crate::app::{App, GRAPHS_METRICS, GraphsMode, PEER_GRAPHS_METRICS, Tab}; + +/// 5×5 braille lookup table indexed by (left fill 0..=4, right fill +/// 0..=4). Direct transcription of btop's `braille_up` glyph set. +const BRAILLE: [[char; 5]; 5] = [ + [' ', '⢀', '⢠', '⢰', '⢸'], + ['⡀', '⣀', '⣠', '⣰', '⣸'], + ['⡄', '⣄', '⣤', '⣴', '⣼'], + ['⡆', '⣆', '⣦', '⣶', '⣾'], + ['⡇', '⣇', '⣧', '⣷', '⣿'], +]; + +/// Height in rows of each per-metric mini block (title row plus plot). +const METRIC_TITLE_ROWS: u16 = 1; +const METRIC_PLOT_ROWS: u16 = 4; +const METRIC_SEPARATOR_ROWS: u16 = 1; +const METRIC_BLOCK_ROWS: u16 = METRIC_TITLE_ROWS + METRIC_PLOT_ROWS + METRIC_SEPARATOR_ROWS; + +pub fn draw(frame: &mut Frame, app: &mut App, area: Rect) { + let chunks = Layout::vertical([ + Constraint::Length(2), // selector strip + Constraint::Min(5), // scrollable stack + ]) + .split(area); + + draw_selector(frame, app, chunks[0]); + draw_stack(frame, app, chunks[1]); +} + +fn draw_selector(frame: &mut Frame, app: &App, area: Rect) { + let label = Style::default().fg(Color::DarkGray); + let dim = Style::default().fg(Color::White); + let emph = Style::default().fg(Color::Cyan); + + let (window, granularity) = app.graphs_window(); + + let mode_str = match app.graphs_mode { + GraphsMode::Node => "node".to_string(), + GraphsMode::MetricByPeer => { + format!("metric-by-peer [{}]", app.graphs_selected_peer_metric()) + } + GraphsMode::PeerByMetric => { + let name = app + .graphs_selected_peer() + .map(|p| p.display_name.clone()) + .unwrap_or_else(|| "(no peers)".into()); + format!("peer-by-metric [{}]", name) + } + }; + + let line1 = Line::from(vec![ + Span::styled(" mode: ", label), + Span::styled(mode_str, emph), + Span::styled(" window: ", label), + Span::styled(window.to_string(), dim), + Span::styled(" granularity: ", label), + Span::styled(granularity.to_string(), dim), + Span::styled(" scroll: ", label), + Span::styled(format!("{}", app.graphs_scroll), dim), + ]); + let line2 = Line::from(Span::styled( + " [↑/↓] scroll [←/→] window [m] mode [n/N] cycle [g] graphs [q] quit", + label, + )); + + frame.render_widget(Paragraph::new(vec![line1, line2]), area); +} + +fn draw_stack(frame: &mut Frame, app: &mut App, area: Rect) { + let title = match app.graphs_mode { + GraphsMode::Node => " Graphs — Node ".to_string(), + GraphsMode::MetricByPeer => { + format!(" Graphs — {} by peer ", app.graphs_selected_peer_metric()) + } + GraphsMode::PeerByMetric => { + let name = app + .graphs_selected_peer() + .map(|p| p.display_name.clone()) + .unwrap_or_else(|| "(no peers)".into()); + format!(" Graphs — peer {} ", name) + } + }; + + let block = Block::default() + .borders(Borders::ALL) + .border_type(BorderType::Rounded) + .border_style(Style::default().fg(Color::DarkGray)) + .title(Span::styled( + title, + Style::default() + .fg(Color::White) + .add_modifier(Modifier::BOLD), + )); + let inner = block.inner(area); + frame.render_widget(block, area); + + match app.graphs_mode { + GraphsMode::Node | GraphsMode::PeerByMetric => draw_stacked(frame, app, inner), + GraphsMode::MetricByPeer => draw_metric_by_peer(frame, app, inner), + } +} + +/// Parse `data.series` into `[(name, values)]` for stacked views. +fn series_from_data(data: &serde_json::Value) -> Vec<(String, Vec)> { + data.get("series") + .and_then(|v| v.as_array()) + .map(|arr| { + arr.iter() + .map(|s| { + let name = s + .get("metric") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(); + let values: Vec = s + .get("values") + .and_then(|v| v.as_array()) + .map(|a| a.iter().map(|v| v.as_f64().unwrap_or(f64::NAN)).collect()) + .unwrap_or_default(); + (name, values) + }) + .collect() + }) + .unwrap_or_default() +} + +fn draw_stacked(frame: &mut Frame, app: &mut App, inner: Rect) { + let data = app.data.get(&Tab::Graphs); + let all_series = match data { + Some(d) => series_from_data(d), + None => Vec::new(), + }; + + if all_series.is_empty() { + frame.render_widget( + Paragraph::new(" Waiting for data...").style(Style::default().fg(Color::DarkGray)), + inner, + ); + return; + } + + // Pick metric list by mode. + let metrics: &[&str] = match app.graphs_mode { + GraphsMode::Node => GRAPHS_METRICS, + GraphsMode::PeerByMetric => PEER_GRAPHS_METRICS, + _ => unreachable!("draw_stacked only renders Node or PeerByMetric modes"), + }; + + let mut lines: Vec> = Vec::new(); + for metric_name in metrics { + let series = all_series + .iter() + .find(|(n, _)| n == metric_name) + .map(|(_, v)| v.as_slice()) + .unwrap_or(&[]); + lines.extend(render_metric_block(metric_name, series, inner.width)); + } + + let total_rows = lines.len() as u16; + let visible = inner.height; + let max_scroll = total_rows.saturating_sub(visible); + if app.graphs_scroll > max_scroll { + app.graphs_scroll = max_scroll; + } + + let paragraph = Paragraph::new(lines).scroll((app.graphs_scroll, 0)); + frame.render_widget(paragraph, inner); +} + +fn draw_metric_by_peer(frame: &mut Frame, app: &mut App, inner: Rect) { + let data = match app.data.get(&Tab::Graphs) { + Some(d) => d, + None => { + frame.render_widget( + Paragraph::new(" Waiting for data...").style(Style::default().fg(Color::DarkGray)), + inner, + ); + return; + } + }; + + let metric_name = app.graphs_selected_peer_metric(); + let peers = data.get("peers").and_then(|v| v.as_array()); + let peer_series: Vec<(String, Vec)> = peers + .map(|arr| { + arr.iter() + .map(|p| { + let name = p + .get("display_name") + .and_then(|v| v.as_str()) + .unwrap_or("?") + .to_string(); + let values: Vec = p + .get("values") + .and_then(|v| v.as_array()) + .map(|a| a.iter().map(|v| v.as_f64().unwrap_or(f64::NAN)).collect()) + .unwrap_or_default(); + (name, values) + }) + .collect() + }) + .unwrap_or_default(); + + if peer_series.is_empty() { + frame.render_widget( + Paragraph::new(" No peers tracked yet in stats history.") + .style(Style::default().fg(Color::DarkGray)), + inner, + ); + return; + } + + // Pick a column count that keeps each cell wide enough for a + // readable braille plot. Each cell needs ~30 columns minimum. + let cols = if inner.width < 40 { + 1 + } else if inner.width < 100 { + 2 + } else { + 3 + }; + let rows = peer_series.len().div_ceil(cols); + + // Stack of cell-rows; each row is a horizontal split of cell cells. + let row_constraints: Vec = (0..rows) + .map(|_| Constraint::Length(METRIC_BLOCK_ROWS)) + .collect(); + let row_areas = Layout::vertical(row_constraints).split(inner); + + for row_idx in 0..rows { + let col_constraints: Vec = (0..cols) + .map(|_| Constraint::Ratio(1, cols as u32)) + .collect(); + let col_areas = Layout::horizontal(col_constraints).split(row_areas[row_idx]); + + for col_idx in 0..cols { + let peer_idx = row_idx * cols + col_idx; + if peer_idx >= peer_series.len() { + break; + } + let (peer_name, values) = &peer_series[peer_idx]; + let cell_lines = render_metric_block_labeled( + metric_name, + peer_name, + values, + col_areas[col_idx].width, + ); + frame.render_widget(Paragraph::new(cell_lines), col_areas[col_idx]); + } + } +} + +/// Variant of `render_metric_block` that labels the block with the +/// peer name in addition to the metric. Used by the metric-by-peer grid. +fn render_metric_block_labeled( + metric: &str, + peer_name: &str, + values: &[f64], + width: u16, +) -> Vec> { + let unit = metric_unit(metric); + let mut out: Vec> = Vec::with_capacity(METRIC_BLOCK_ROWS as usize); + + let (min, max, last, n) = summarize(values); + let title_style = Style::default() + .fg(Color::White) + .add_modifier(Modifier::BOLD); + let label = Style::default().fg(Color::DarkGray); + let title = Line::from(vec![ + Span::styled(format!(" {peer_name}"), title_style), + Span::styled(format!(" [{unit}]"), label), + Span::styled(" max ", label), + Span::raw(format_value(max)), + Span::styled(" last ", label), + Span::raw(format_value(last)), + Span::styled(" n=", label), + Span::raw(format!("{n}")), + ]); + out.push(title); + + let gutter = 2u16; + let plot_cols = width.saturating_sub(gutter) as usize; + + if plot_cols == 0 || values.is_empty() { + for _ in 0..METRIC_PLOT_ROWS { + out.push(Line::from(Span::styled( + " (no samples)", + Style::default().fg(Color::DarkGray), + ))); + } + out.push(Line::from("")); + return out; + } + + let sampled = resample(values, plot_cols * 2); + let rows = METRIC_PLOT_ROWS as usize; + let plot_lines = render_btop_graph(&sampled, rows, min, max, gutter as usize); + out.extend(plot_lines); + out.push(Line::from("")); + out +} + +/// Render a single metric's mini block: one title row, four plot rows, +/// one separator row. The plot is autoscaled against its own min/max +/// so every metric uses the full vertical resolution regardless of +/// its unit. +fn render_metric_block(metric: &str, values: &[f64], width: u16) -> Vec> { + let unit = metric_unit(metric); + let mut out: Vec> = Vec::with_capacity(METRIC_BLOCK_ROWS as usize); + + // Title row. + let (min, max, last, n) = summarize(values); + let title_style = Style::default() + .fg(Color::White) + .add_modifier(Modifier::BOLD); + let label = Style::default().fg(Color::DarkGray); + let title = Line::from(vec![ + Span::styled(format!(" {metric}"), title_style), + Span::styled(format!(" [{unit}]"), label), + Span::styled(" max ", label), + Span::raw(format_value(max)), + Span::styled(" last ", label), + Span::raw(format_value(last)), + Span::styled(" samples ", label), + Span::raw(format!("{n}")), + ]); + out.push(title); + + // Plot rows. Width budget: leave the first two columns for a tiny + // left gutter so titles and plots align consistently. + let gutter = 2u16; + let plot_cols = width.saturating_sub(gutter) as usize; + + if plot_cols == 0 || values.is_empty() { + for _ in 0..METRIC_PLOT_ROWS { + out.push(Line::from(Span::styled( + " (no samples)", + Style::default().fg(Color::DarkGray), + ))); + } + out.push(Line::from("")); + return out; + } + + let sampled = resample(values, plot_cols * 2); + let rows = METRIC_PLOT_ROWS as usize; + let plot_lines = render_btop_graph(&sampled, rows, min, max, gutter as usize); + out.extend(plot_lines); + + // Blank separator. + out.push(Line::from("")); + out +} + +fn summarize(values: &[f64]) -> (f64, f64, f64, usize) { + if values.is_empty() { + return (0.0, 0.0, 0.0, 0); + } + let (min, max) = values + .iter() + .filter(|v| !v.is_nan()) + .fold((f64::INFINITY, f64::NEG_INFINITY), |(lo, hi), &v| { + (lo.min(v), hi.max(v)) + }); + let (min, max) = if min.is_finite() { + (min, max) + } else { + (0.0, 0.0) + }; + let last = values + .iter() + .rev() + .find(|v| !v.is_nan()) + .copied() + .unwrap_or(f64::NAN); + (min, max, last, values.len()) +} + +/// Down- or up-sample to exactly `target` points using linear +/// interpolation across the source index space. Returns empty on +/// empty input. +fn resample(values: &[f64], target: usize) -> Vec { + if values.is_empty() || target == 0 { + return Vec::new(); + } + if values.len() == target { + return values.to_vec(); + } + if values.len() == 1 { + return vec![values[0]; target]; + } + let mut out = Vec::with_capacity(target); + let last_src = (values.len() - 1) as f64; + let last_dst = (target - 1).max(1) as f64; + for i in 0..target { + let src = i as f64 * last_src / last_dst; + let lo = src.floor() as usize; + let hi = (src.ceil() as usize).min(values.len() - 1); + let frac = src - lo as f64; + out.push(values[lo] * (1.0 - frac) + values[hi] * frac); + } + out +} + +/// Render a filled-area graph using btop's braille algorithm. +/// +/// - Normalize values to 0..=100 against the visible min/max. +/// - For each row `horizon` (0 = top), compute the percent band it +/// covers; each sample's fill level is 0..=4 based on where the +/// sample falls within that band (+0.1 modulus bias so small +/// non-zero values still draw). +/// - Two samples per character cell via the 25-entry BRAILLE table. +/// - Per-row gradient color keyed by row position: top rows hot, bottom +/// rows cool. This matches btop's "vertical color bands" look. +fn render_btop_graph( + values: &[f64], + rows: usize, + min: f64, + max: f64, + gutter: usize, +) -> Vec> { + if rows == 0 || values.is_empty() { + return Vec::new(); + } + + let range = max - min; + // NaN samples pass through normalize as NaN so the cell loop below + // can blank them. Non-NaN samples are clamped into 0..=100. + let normalized: Vec = values + .iter() + .map(|&v| { + if v.is_nan() { + f64::NAN + } else if !range.is_finite() || range <= 0.0 { + 50.0 + } else { + ((v - min) / range * 100.0).clamp(0.0, 100.0) + } + }) + .collect(); + + let clamp_min = 0i32; + let mod_bias = 0.1f64; + let gutter_str: String = " ".repeat(gutter); + + let mut lines: Vec> = Vec::with_capacity(rows); + + for horizon in 0..rows { + let cur_high = ((rows - horizon) as f64 * 100.0 / rows as f64).round() as i32; + let cur_low = ((rows - horizon - 1) as f64 * 100.0 / rows as f64).round() as i32; + let color = row_gradient(horizon, rows); + let style = Style::default().fg(color); + + let mut chars = String::with_capacity(values.len() / 2 + 1); + let mut i = 0; + while i + 1 < normalized.len() { + let l = normalized[i]; + let r = normalized[i + 1]; + let c = if l.is_nan() || r.is_nan() { + ' ' + } else { + let level_l = sample_level(l, cur_low, cur_high, clamp_min, mod_bias); + let level_r = sample_level(r, cur_low, cur_high, clamp_min, mod_bias); + BRAILLE[level_l as usize][level_r as usize] + }; + chars.push(c); + i += 2; + } + if i < normalized.len() { + let l = normalized[i]; + let c = if l.is_nan() { + ' ' + } else { + let level_l = sample_level(l, cur_low, cur_high, clamp_min, mod_bias); + BRAILLE[level_l as usize][0] + }; + chars.push(c); + } + + lines.push(Line::from(vec![ + Span::raw(gutter_str.clone()), + Span::styled(chars, style), + ])); + } + + lines +} + +fn sample_level(value: f64, cur_low: i32, cur_high: i32, clamp_min: i32, mod_bias: f64) -> i32 { + let v = value.round() as i32; + if v >= cur_high { + 4 + } else if v <= cur_low { + clamp_min + } else { + let span = (cur_high - cur_low).max(1) as f64; + let scaled = ((value - cur_low as f64) * 4.0 / span + mod_bias).round() as i32; + scaled.clamp(clamp_min, 4) + } +} + +fn row_gradient(horizon: usize, rows: usize) -> Color { + let t = 1.0 - (horizon as f64 / rows.max(1) as f64); + gradient_rgb(t) +} + +fn gradient_rgb(t: f64) -> Color { + let t = t.clamp(0.0, 1.0); + let (sr, sg, sb) = (0.0, 200.0, 0.0); + let (mr, mg, mb) = (240.0, 200.0, 0.0); + let (er, eg, eb) = (240.0, 40.0, 40.0); + let (r, g, b) = if t < 0.5 { + let k = t * 2.0; + (lerp(sr, mr, k), lerp(sg, mg, k), lerp(sb, mb, k)) + } else { + let k = (t - 0.5) * 2.0; + (lerp(mr, er, k), lerp(mg, eg, k), lerp(mb, eb, k)) + }; + Color::Rgb(r as u8, g as u8, b as u8) +} + +fn lerp(a: f64, b: f64, t: f64) -> f64 { + a + (b - a) * t +} + +fn format_value(v: f64) -> String { + if v.is_nan() { + return "-".to_string(); + } + if v.abs() < 10.0 { + format!("{:.2}", v) + } else if v.abs() < 1000.0 { + format!("{:.1}", v) + } else if v.abs() < 1_000_000.0 { + format!("{:.1}K", v / 1000.0) + } else { + format!("{:.1}M", v / 1_000_000.0) + } +} + +fn metric_unit(name: &str) -> &'static str { + match name { + "mesh_size" => "nodes", + "tree_depth" => "hops", + "peer_count" => "peers", + "parent_switches" => "events/s", + "bytes_in" | "bytes_out" => "bytes/s", + "packets_in" | "packets_out" => "packets/s", + "loss_rate" => "fraction", + "active_sessions" => "sessions", + "srtt_ms" => "ms", + "ecn_ce" => "events/s", + _ => "", + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn resample_identity_when_length_matches() { + let v = vec![1.0, 2.0, 3.0]; + assert_eq!(resample(&v, 3), v); + } + + #[test] + fn resample_returns_requested_length() { + let v = vec![1.0, 2.0, 3.0, 4.0, 5.0]; + let out = resample(&v, 10); + assert_eq!(out.len(), 10); + } + + #[test] + fn braille_table_corners() { + assert_eq!(BRAILLE[0][0], ' '); + assert_eq!(BRAILLE[4][4], '⣿'); + assert_eq!(BRAILLE[4][0], '⡇'); + assert_eq!(BRAILLE[0][4], '⢸'); + } + + #[test] + fn sample_level_boundaries() { + assert_eq!(sample_level(0.0, 0, 25, 0, 0.1), 0); + assert_eq!(sample_level(25.0, 0, 25, 0, 0.1), 4); + } + + #[test] + fn metric_block_has_expected_rows() { + let v = vec![1.0, 2.0, 3.0, 4.0, 5.0]; + let lines = render_metric_block("mesh_size", &v, 40); + assert_eq!(lines.len(), METRIC_BLOCK_ROWS as usize); + } + + #[test] + fn metric_block_handles_empty() { + let lines = render_metric_block("mesh_size", &[], 40); + assert_eq!(lines.len(), METRIC_BLOCK_ROWS as usize); + } + + #[test] + fn gradient_spans_stops() { + if let Color::Rgb(r, g, _) = gradient_rgb(0.0) { + assert!(g > r, "bottom of gradient is greener than it is red"); + } else { + panic!("expected RGB"); + } + if let Color::Rgb(r, g, _) = gradient_rgb(1.0) { + assert!(r > g, "top of gradient is redder than it is green"); + } else { + panic!("expected RGB"); + } + } +} diff --git a/src/bin/fipstop/ui/helpers.rs b/src/bin/fipstop/ui/helpers.rs index 8a472e9..4dbff67 100644 --- a/src/bin/fipstop/ui/helpers.rs +++ b/src/bin/fipstop/ui/helpers.rs @@ -171,3 +171,42 @@ pub fn kv_line(key: &str, value: &str) -> Line<'static> { Span::raw(value.to_string()), ]) } + +/// Render a sequence of values as Unicode block characters. +/// +/// Returns an empty string for empty input. Constant series render as a +/// mid-level row. Used inline beside metric values in the dashboard and +/// as the per-column renderer for the Graphs tab. +pub fn sparkline(values: &[f64]) -> String { + const BLOCKS: [char; 8] = ['▁', '▂', '▃', '▄', '▅', '▆', '▇', '█']; + if values.is_empty() { + return String::new(); + } + let (min, max) = values + .iter() + .fold((f64::INFINITY, f64::NEG_INFINITY), |(lo, hi), &v| { + (lo.min(v), hi.max(v)) + }); + let range = max - min; + values + .iter() + .map(|&v| { + if !range.is_finite() || range <= 0.0 { + BLOCKS[3] + } else { + let norm = ((v - min) / range).clamp(0.0, 1.0); + let idx = (norm * (BLOCKS.len() as f64 - 1.0)).round() as usize; + BLOCKS[idx.min(BLOCKS.len() - 1)] + } + }) + .collect() +} + +/// Extract a `Vec` from a nested JSON array (e.g., `sparklines.mesh_size`). +pub fn nested_f64_array(data: &Value, outer: &str, inner: &str) -> Vec { + data.get(outer) + .and_then(|o| o.get(inner)) + .and_then(|v| v.as_array()) + .map(|arr| arr.iter().filter_map(|v| v.as_f64()).collect()) + .unwrap_or_default() +} diff --git a/src/bin/fipstop/ui/mod.rs b/src/bin/fipstop/ui/mod.rs index 5d2d5c8..1b5bf53 100644 --- a/src/bin/fipstop/ui/mod.rs +++ b/src/bin/fipstop/ui/mod.rs @@ -1,6 +1,7 @@ mod bloom; mod dashboard; mod gateway; +mod graphs; mod helpers; mod mmp; mod peers; @@ -77,6 +78,7 @@ fn draw_content(frame: &mut Frame, app: &mut App, area: Rect) { Tab::Routing => routing::draw(frame, app, area), Tab::Links => {} // not a navigable tab; data stored for cross-references Tab::Gateway => gateway::draw(frame, app, area), + Tab::Graphs => graphs::draw(frame, app, area), } } diff --git a/src/control/mod.rs b/src/control/mod.rs index f0c003b..9a052e7 100644 --- a/src/control/mod.rs +++ b/src/control/mod.rs @@ -374,6 +374,7 @@ pub use windows_impl::ControlSocket; #[cfg(test)] mod tests { + #[cfg(windows)] use super::*; #[cfg(windows)] diff --git a/src/control/queries.rs b/src/control/queries.rs index 4d39697..908dd15 100644 --- a/src/control/queries.rs +++ b/src/control/queries.rs @@ -3,9 +3,19 @@ //! Each function takes `&Node` and returns a `serde_json::Value`. //! Query logic is kept separate from socket handling. -use crate::identity::encode_npub; +use crate::identity::{NodeAddr, PeerIdentity, encode_npub}; use crate::node::Node; +use crate::node::stats_history::{ALL_METRICS, ALL_PEER_METRICS, Granularity, Metric, PeerMetric}; use serde_json::{Value, json}; +use std::str::FromStr; +use std::time::Duration; + +/// Resolve an `npub1...` string to the corresponding `NodeAddr`. +fn parse_peer_npub(s: &str) -> Result { + PeerIdentity::from_npub(s) + .map(|p| *p.node_addr()) + .map_err(|e| format!("invalid peer npub: {e}")) +} /// Helper: get current Unix time in milliseconds. fn now_ms() -> u64 { @@ -39,6 +49,20 @@ pub fn show_status(node: &Node) -> Value { let uptime_secs = node.uptime().as_secs(); let fwd = node.stats().snapshot().forwarding; + // Inline last-N-second sparklines for dashboard rendering. Kept + // short so the status payload stays compact; longer windows use + // `show_stats_history`. + const SPARK_N: usize = 30; + let hist = node.stats_history(); + let sparklines = json!({ + "mesh_size": hist.recent(Metric::MeshSize, SPARK_N), + "tree_depth": hist.recent(Metric::TreeDepth, SPARK_N), + "peer_count": hist.recent(Metric::PeerCount, SPARK_N), + "bytes_in": hist.recent(Metric::BytesIn, SPARK_N), + "bytes_out": hist.recent(Metric::BytesOut, SPARK_N), + "loss_rate": hist.recent(Metric::LossRate, SPARK_N), + }); + json!({ "version": crate::version::short_version(), "npub": node.npub(), @@ -60,6 +84,7 @@ pub fn show_status(node: &Node) -> Value { "uptime_secs": uptime_secs, "estimated_mesh_size": node.estimated_mesh_size(), "forwarding": serde_json::to_value(&fwd).unwrap_or_default(), + "sparklines": sparklines, }) } @@ -688,8 +713,342 @@ pub fn show_identity_cache(node: &Node) -> Value { }) } +/// `show_stats_list` — Enumerate available history metrics and their units. +pub fn show_stats_list() -> Value { + let metrics: Vec = ALL_METRICS + .iter() + .map(|m| { + json!({ + "name": m.name(), + "unit": m.unit(), + "scope": "node", + }) + }) + .chain(ALL_PEER_METRICS.iter().map(|m| { + json!({ + "name": m.name(), + "unit": m.unit(), + "scope": "peer", + }) + })) + .collect(); + json!({ + "metrics": metrics, + "fast_ring_seconds": crate::node::stats_history::FAST_RING_CAPACITY, + "slow_ring_minutes": crate::node::stats_history::SLOW_RING_CAPACITY, + "peer_retention_seconds": crate::node::stats_history::PEER_EVICTION_SECS, + }) +} + +/// `show_stats_history` — Time-series samples for one metric. +/// +/// Params: +/// - `metric` (required): metric name. Node-level metrics (e.g. +/// `mesh_size`) are resolved against `Metric`; per-peer metrics (e.g. +/// `srtt_ms`, `ecn_ce`) require the `peer` param and resolve against +/// `PeerMetric`. +/// - `peer` (optional): `npub1...` of the peer; required for per-peer +/// metrics. +/// - `window` (default `10m`): duration `s`, `m`, or `h`. +/// - `granularity` (default `1s`): `1s` or `1m`. +pub fn show_stats_history(node: &Node, params: Option<&Value>) -> super::protocol::Response { + use super::protocol::Response; + let Some(params) = params else { + return Response::error("missing params for show_stats_history"); + }; + + let metric_name = match params.get("metric").and_then(|v| v.as_str()) { + Some(v) => v, + None => return Response::error("missing 'metric' parameter"), + }; + + let window_str = params + .get("window") + .and_then(|v| v.as_str()) + .unwrap_or("10m"); + let window = match parse_duration(window_str) { + Ok(d) => d, + Err(e) => return Response::error(e), + }; + + let granularity_str = params + .get("granularity") + .and_then(|v| v.as_str()) + .unwrap_or("1s"); + let granularity = match Granularity::from_str(granularity_str) { + Ok(g) => g, + Err(e) => return Response::error(e), + }; + + let peer_npub = params.get("peer").and_then(|v| v.as_str()); + let hist = node.stats_history(); + + if let Some(npub) = peer_npub { + let addr = match parse_peer_npub(npub) { + Ok(a) => a, + Err(e) => return Response::error(e), + }; + let peer_metric = match PeerMetric::from_str(metric_name) { + Ok(m) => m, + Err(e) => return Response::error(e), + }; + match hist.peer_query(&addr, peer_metric, window, granularity) { + Some(series) => Response::ok(serde_json::to_value(&series).unwrap_or(Value::Null)), + None => Response::error(format!( + "peer not tracked in stats history: {}", + node.peer_display_name(&addr) + )), + } + } else { + let metric = match Metric::from_str(metric_name) { + Ok(m) => m, + Err(e) => return Response::error(e), + }; + let series = hist.query(metric, window, granularity); + Response::ok(serde_json::to_value(&series).unwrap_or(Value::Null)) + } +} + +/// Parse a duration of the form `s`, `m`, or `h` into a `Duration`. +fn parse_duration(s: &str) -> Result { + if s.is_empty() { + return Err("empty duration".to_string()); + } + let (num_part, unit) = s.split_at(s.len() - 1); + let n: u64 = num_part + .parse() + .map_err(|_| format!("invalid duration: {s}"))?; + let secs = match unit { + "s" => n, + "m" => n * 60, + "h" => n * 3600, + _ => return Err(format!("unknown duration unit: {unit} (expected s, m, h)")), + }; + Ok(Duration::from_secs(secs)) +} + +/// `show_stats_all_history` — Return a series for every tracked metric +/// in one round trip. Intended for the fipstop Graphs tab. +/// +/// Without `peer`: returns the 10 node-level metrics. +/// With `peer` (npub): returns the 7 per-peer metrics for that peer. +/// +/// Params: `{"peer": ""?, "window": "", "granularity": "<1s|1m>"}`. +pub fn show_stats_all_history(node: &Node, params: Option<&Value>) -> super::protocol::Response { + use super::protocol::Response; + let params = params.cloned().unwrap_or_else(|| json!({})); + + let window_str = params + .get("window") + .and_then(|v| v.as_str()) + .unwrap_or("10m"); + let window = match parse_duration(window_str) { + Ok(d) => d, + Err(e) => return Response::error(e), + }; + + let granularity_str = params + .get("granularity") + .and_then(|v| v.as_str()) + .unwrap_or("1s"); + let granularity = match Granularity::from_str(granularity_str) { + Ok(g) => g, + Err(e) => return Response::error(e), + }; + + let peer_npub = params.get("peer").and_then(|v| v.as_str()); + let hist = node.stats_history(); + + let series: Vec = if let Some(npub) = peer_npub { + let addr = match parse_peer_npub(npub) { + Ok(a) => a, + Err(e) => return Response::error(e), + }; + if !hist.has_peer(&addr) { + return Response::error(format!( + "peer not tracked in stats history: {}", + node.peer_display_name(&addr) + )); + } + ALL_PEER_METRICS + .iter() + .map(|m| { + let s = hist + .peer_query(&addr, *m, window, granularity) + .unwrap_or_else(|| { + // Unreachable: has_peer checked above, but degrade + // gracefully rather than panic. + crate::node::stats_history::Series { + metric: m.name(), + unit: m.unit(), + granularity_seconds: granularity.seconds(), + values: Vec::new(), + } + }); + serde_json::to_value(&s).unwrap_or(Value::Null) + }) + .collect() + } else { + ALL_METRICS + .iter() + .map(|m| { + let s = hist.query(*m, window, granularity); + serde_json::to_value(&s).unwrap_or(Value::Null) + }) + .collect() + }; + + Response::ok(json!({ + "granularity_seconds": granularity.seconds(), + "window_seconds": window.as_secs(), + "peer": peer_npub, + "series": series, + })) +} + +/// `show_stats_peers` — Enumerate peers tracked in the stats history +/// with their lifecycle metadata. Used by operator tools to populate +/// peer selectors and to confirm a peer is in the retention window. +pub fn show_stats_peers(node: &Node) -> Value { + let hist = node.stats_history(); + let now = std::time::Instant::now(); + + let mut peers: Vec = hist + .peers() + .map(|(addr, rings)| { + let last_contact_secs = now.duration_since(rings.last_contact()).as_secs(); + let first_seen_secs = now.duration_since(rings.first_seen()).as_secs(); + let is_active = node.peers().any(|p| p.node_addr() == addr); + let npub = node + .peers() + .find(|p| p.node_addr() == addr) + .map(|p| p.npub()) + .unwrap_or_else(|| hex::encode(addr.as_bytes())); + json!({ + "npub": npub, + "node_addr": hex::encode(addr.as_bytes()), + "display_name": node.peer_display_name(addr), + "is_active": is_active, + "first_seen_secs_ago": first_seen_secs, + "last_contact_secs_ago": last_contact_secs, + }) + }) + .collect(); + + // Stable display order: active peers first, then by display name. + peers.sort_by(|a, b| { + let a_active = a + .get("is_active") + .and_then(|v| v.as_bool()) + .unwrap_or(false); + let b_active = b + .get("is_active") + .and_then(|v| v.as_bool()) + .unwrap_or(false); + match (b_active, a_active) { + (true, false) => std::cmp::Ordering::Greater, + (false, true) => std::cmp::Ordering::Less, + _ => a + .get("display_name") + .and_then(|v| v.as_str()) + .unwrap_or("") + .cmp(b.get("display_name").and_then(|v| v.as_str()).unwrap_or("")), + } + }); + + json!({ "peers": peers, "count": peers.len() }) +} + +/// `show_stats_history_all_peers` — One metric across every tracked +/// peer in one round trip. Backs the fipstop MetricByPeer grid view. +/// +/// Params: `{"metric": "", "window": "", "granularity": "<1s|1m>"}`. +/// `metric` must be a per-peer metric name (see `PeerMetric`). +pub fn show_stats_history_all_peers( + node: &Node, + params: Option<&Value>, +) -> super::protocol::Response { + use super::protocol::Response; + let Some(params) = params else { + return Response::error("missing params for show_stats_history_all_peers"); + }; + + let metric_name = match params.get("metric").and_then(|v| v.as_str()) { + Some(v) => v, + None => return Response::error("missing 'metric' parameter"), + }; + let metric = match PeerMetric::from_str(metric_name) { + Ok(m) => m, + Err(e) => return Response::error(e), + }; + + let window_str = params + .get("window") + .and_then(|v| v.as_str()) + .unwrap_or("10m"); + let window = match parse_duration(window_str) { + Ok(d) => d, + Err(e) => return Response::error(e), + }; + + let granularity_str = params + .get("granularity") + .and_then(|v| v.as_str()) + .unwrap_or("1s"); + let granularity = match Granularity::from_str(granularity_str) { + Ok(g) => g, + Err(e) => return Response::error(e), + }; + + let hist = node.stats_history(); + let peer_addrs: Vec = hist.peer_addrs().copied().collect(); + + let mut peers: Vec = peer_addrs + .iter() + .filter_map(|addr| { + let s = hist.peer_query(addr, metric, window, granularity)?; + let is_active = node.peers().any(|p| p.node_addr() == addr); + Some(json!({ + "node_addr": hex::encode(addr.as_bytes()), + "display_name": node.peer_display_name(addr), + "is_active": is_active, + "values": serde_json::to_value(&s.values).unwrap_or(Value::Null), + })) + }) + .collect(); + + // Active peers first, then by display name. + peers.sort_by(|a, b| { + let a_active = a + .get("is_active") + .and_then(|v| v.as_bool()) + .unwrap_or(false); + let b_active = b + .get("is_active") + .and_then(|v| v.as_bool()) + .unwrap_or(false); + match (b_active, a_active) { + (true, false) => std::cmp::Ordering::Greater, + (false, true) => std::cmp::Ordering::Less, + _ => a + .get("display_name") + .and_then(|v| v.as_str()) + .unwrap_or("") + .cmp(b.get("display_name").and_then(|v| v.as_str()).unwrap_or("")), + } + }); + + Response::ok(json!({ + "metric": metric.name(), + "unit": metric.unit(), + "granularity_seconds": granularity.seconds(), + "window_seconds": window.as_secs(), + "peers": peers, + })) +} + /// Dispatch a command string to the appropriate query function. -pub fn dispatch(node: &Node, command: &str) -> super::protocol::Response { +pub fn dispatch(node: &Node, command: &str, params: Option<&Value>) -> super::protocol::Response { match command { "show_status" => super::protocol::Response::ok(show_status(node)), "show_peers" => super::protocol::Response::ok(show_peers(node)), @@ -703,6 +1062,11 @@ pub fn dispatch(node: &Node, command: &str) -> super::protocol::Response { "show_transports" => super::protocol::Response::ok(show_transports(node)), "show_routing" => super::protocol::Response::ok(show_routing(node)), "show_identity_cache" => super::protocol::Response::ok(show_identity_cache(node)), + "show_stats_list" => super::protocol::Response::ok(show_stats_list()), + "show_stats_history" => show_stats_history(node, params), + "show_stats_all_history" => show_stats_all_history(node, params), + "show_stats_peers" => super::protocol::Response::ok(show_stats_peers(node)), + "show_stats_history_all_peers" => show_stats_history_all_peers(node, params), _ => super::protocol::Response::error(format!("unknown command: {}", command)), } } diff --git a/src/node/handlers/rx_loop.rs b/src/node/handlers/rx_loop.rs index 225bb72..d165624 100644 --- a/src/node/handlers/rx_loop.rs +++ b/src/node/handlers/rx_loop.rs @@ -100,7 +100,7 @@ impl Node { } Some((request, response_tx)) = control_rx.recv() => { let response = if request.command.starts_with("show_") { - queries::dispatch(self, &request.command) + queries::dispatch(self, &request.command, request.params.as_ref()) } else { commands::dispatch( self, @@ -125,6 +125,7 @@ impl Node { self.check_tree_state().await; self.check_bloom_state().await; self.compute_mesh_size(); + self.record_stats_history(); self.check_mmp_reports().await; self.check_session_mmp_reports().await; self.check_link_heartbeats().await; diff --git a/src/node/mod.rs b/src/node/mod.rs index a9d8c97..297bc59 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -14,6 +14,7 @@ mod routing_error_rate_limit; pub(crate) mod session; pub(crate) mod session_wire; pub(crate) mod stats; +pub(crate) mod stats_history; #[cfg(test)] mod tests; mod tree; @@ -358,6 +359,9 @@ pub struct Node { /// Routing, forwarding, discovery, and error signal counters. stats: stats::NodeStats, + /// Time-series history of node-level metrics (1s/1m rings). + stats_history: stats_history::StatsHistory, + // === TUN Interface === /// TUN device state. tun_state: TunState, @@ -535,6 +539,7 @@ impl Node { next_link_id: 1, next_transport_id: 1, stats: stats::NodeStats::new(), + stats_history: stats_history::StatsHistory::new(), tun_state, tun_name: None, tun_tx: None, @@ -644,6 +649,7 @@ impl Node { next_link_id: 1, next_transport_id: 1, stats: stats::NodeStats::new(), + stats_history: stats_history::StatsHistory::new(), tun_state, tun_name: None, tun_tx: None, @@ -1115,6 +1121,70 @@ impl Node { &mut self.stats } + /// Get the stats history collector. + pub fn stats_history(&self) -> &stats_history::StatsHistory { + &self.stats_history + } + + /// Sample the current node state into the stats history ring. + /// Called once per tick from the RX loop. + pub(crate) fn record_stats_history(&mut self) { + let fwd = &self.stats.forwarding; + let peers_with_mmp: Vec = self + .peers + .values() + .filter_map(|p| p.mmp().map(|m| m.metrics.loss_rate())) + .collect(); + let loss_rate = if peers_with_mmp.is_empty() { + 0.0 + } else { + peers_with_mmp.iter().sum::() / peers_with_mmp.len() as f64 + }; + + let snap = stats_history::Snapshot { + mesh_size: self.estimated_mesh_size, + tree_depth: self.tree_state.my_coords().depth() as u32, + peer_count: self.peers.len() as u64, + parent_switches_total: self.stats.tree.parent_switches, + bytes_in_total: fwd.received_bytes, + bytes_out_total: fwd.forwarded_bytes + fwd.originated_bytes, + packets_in_total: fwd.received_packets, + packets_out_total: fwd.forwarded_packets + fwd.originated_packets, + loss_rate, + active_sessions: self.sessions.len() as u64, + }; + + let now = std::time::Instant::now(); + let peer_snaps: Vec = self + .peers + .values() + .map(|p| { + let stats = p.link_stats(); + let (srtt_ms, loss_rate, ecn_ce) = match p.mmp() { + Some(m) => ( + m.metrics.srtt_ms(), + Some(m.metrics.loss_rate()), + m.receiver.ecn_ce_count() as u64, + ), + None => (None, None, 0), + }; + stats_history::PeerSnapshot { + node_addr: *p.node_addr(), + last_seen: now, + srtt_ms, + loss_rate, + bytes_in_total: stats.bytes_recv, + bytes_out_total: stats.bytes_sent, + packets_in_total: stats.packets_recv, + packets_out_total: stats.packets_sent, + ecn_ce_total: ecn_ce, + } + }) + .collect(); + + self.stats_history.tick(now, &snap, &peer_snaps); + } + // === TUN Interface === /// Get the TUN state. diff --git a/src/node/stats_history.rs b/src/node/stats_history.rs new file mode 100644 index 0000000..6f594da --- /dev/null +++ b/src/node/stats_history.rs @@ -0,0 +1,1218 @@ +//! 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 { + 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 { + 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, + 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, + /// MMP loss rate; `None` when the peer has no MMP session yet. + pub loss_rate: Option, + /// 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 { + 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, + slow: VecDeque, + /// 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, +} + +/// 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 { + 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(values: &[f64], s: S) -> Result { + 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::::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, +} + +/// One peer's per-metric rings plus lifecycle metadata. +pub struct PeerStatsRings { + rings: Vec, + first_seen: Instant, + last_contact: Instant, +} + +impl PeerStatsRings { + fn new(now: Instant, fast_pushes_so_far: u64) -> Self { + let mut rings: Vec = 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, + peers: HashMap, + /// 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, +} + +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 = 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 { + 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, 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 = 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 { + 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 { + 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 { + self.peers.keys() + } + + /// Iterate tracked peers with their ring metadata. + pub fn peers(&self) -> impl Iterator { + 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 { + use super::*; + + fn make_snap(t: u64) -> Snapshot { + Snapshot { + mesh_size: Some(10 + t), + tree_depth: 2, + peer_count: 3, + parent_switches_total: t, + bytes_in_total: 100 * t, + bytes_out_total: 200 * t, + packets_in_total: t, + packets_out_total: 2 * t, + loss_rate: 0.01 * t as f64, + active_sessions: t, + } + } + + fn make_addr(tag: u8) -> NodeAddr { + NodeAddr::from_bytes([tag; 16]) + } + + fn make_peer_snap(tag: u8, now: Instant, t: u64) -> PeerSnapshot { + PeerSnapshot { + node_addr: make_addr(tag), + last_seen: now, + srtt_ms: Some(10.0 + t as f64), + loss_rate: Some(0.01 * t as f64), + bytes_in_total: 50 * t, + bytes_out_total: 75 * t, + packets_in_total: t, + packets_out_total: 2 * t, + ecn_ce_total: 0, + } + } + + #[test] + fn push_and_query_fast_ring() { + let mut h = StatsHistory::new(); + let t0 = Instant::now(); + for i in 0..10 { + h.tick(t0 + Duration::from_secs(i), &make_snap(i), &[]); + } + let s = h.query(Metric::MeshSize, Duration::from_secs(5), Granularity::Fast); + assert_eq!(s.values.len(), 5); + assert_eq!(s.values, vec![15.0, 16.0, 17.0, 18.0, 19.0]); + assert_eq!(s.granularity_seconds, 1); + } + + #[test] + fn fast_ring_wraps_at_capacity() { + let mut h = StatsHistory::new(); + let t0 = Instant::now(); + for i in 0..3610u64 { + h.tick(t0 + Duration::from_secs(i), &make_snap(i), &[]); + } + let s = h.query( + Metric::MeshSize, + Duration::from_secs(FAST_RING_CAPACITY as u64 * 2), + Granularity::Fast, + ); + assert_eq!(s.values.len(), FAST_RING_CAPACITY); + assert_eq!(s.values[0], 20.0); + assert_eq!(*s.values.last().unwrap(), 3619.0); + } + + #[test] + fn delta_for_counter_metric() { + let mut h = StatsHistory::new(); + let t0 = Instant::now(); + let totals = [0, 0, 2, 5]; + for (i, &v) in totals.iter().enumerate() { + let mut s = make_snap(i as u64); + s.parent_switches_total = v; + h.tick(t0 + Duration::from_secs(i as u64), &s, &[]); + } + let s = h.query( + Metric::ParentSwitches, + Duration::from_secs(10), + Granularity::Fast, + ); + assert_eq!(s.values.len(), 10); + assert!(s.values[..6].iter().all(|v| v.is_nan())); + assert_eq!(s.values[6..], [0.0, 0.0, 2.0, 3.0]); + } + + #[test] + fn downsample_last_aggregation() { + let mut h = StatsHistory::new(); + let t0 = Instant::now(); + for i in 0..60u64 { + h.tick(t0 + Duration::from_secs(i), &make_snap(i), &[]); + } + let s = h.query( + Metric::MeshSize, + Duration::from_secs(60 * 5), + Granularity::Slow, + ); + assert_eq!(s.values.len(), 5); + assert!(s.values[..4].iter().all(|v| v.is_nan())); + assert_eq!(s.values[4], 69.0); + assert_eq!(s.granularity_seconds, 60); + } + + #[test] + fn downsample_mean_aggregation() { + let mut h = StatsHistory::new(); + let t0 = Instant::now(); + for i in 0..60u64 { + h.tick(t0 + Duration::from_secs(i), &make_snap(i), &[]); + } + let s = h.query(Metric::LossRate, Duration::from_secs(60), Granularity::Slow); + assert_eq!(s.values.len(), 1); + assert!((s.values[0] - 0.295).abs() < 1e-9); + } + + #[test] + fn downsample_sum_aggregation() { + let mut h = StatsHistory::new(); + let t0 = Instant::now(); + for i in 0..60u64 { + let mut s = make_snap(0); + s.parent_switches_total = i; + h.tick(t0 + Duration::from_secs(i), &s, &[]); + } + let s = h.query( + Metric::ParentSwitches, + Duration::from_secs(60), + Granularity::Slow, + ); + assert_eq!(s.values.len(), 1); + assert_eq!(s.values[0], 59.0); + } + + #[test] + fn query_pads_front_with_nan_when_ring_is_short() { + let mut h = StatsHistory::new(); + let t0 = Instant::now(); + for i in 0..3u64 { + h.tick(t0 + Duration::from_secs(i), &make_snap(i), &[]); + } + let s = h.query(Metric::MeshSize, Duration::from_secs(10), Granularity::Fast); + assert_eq!(s.values.len(), 10); + assert!(s.values[..7].iter().all(|v| v.is_nan())); + assert_eq!(s.values[7..], [10.0, 11.0, 12.0]); + } + + #[test] + fn fast_query_young_ring_returns_full_hour_with_leading_nan() { + let mut h = StatsHistory::new(); + let t0 = Instant::now(); + // 5 minutes of data. + for i in 0..300u64 { + h.tick(t0 + Duration::from_secs(i), &make_snap(i), &[]); + } + let s = h.query( + Metric::MeshSize, + Duration::from_secs(3600), + Granularity::Fast, + ); + assert_eq!(s.values.len(), 3600); + assert!(s.values[..3300].iter().all(|v| v.is_nan())); + assert_eq!(s.values[3300], 10.0); + assert_eq!(*s.values.last().unwrap(), 309.0); + } + + #[test] + fn slow_query_young_ring_returns_full_day_with_leading_nan() { + let mut h = StatsHistory::new(); + let t0 = Instant::now(); + // 30 minutes of data → 30 slow samples flushed. + for i in 0u64..1800 { + h.tick(t0 + Duration::from_secs(i), &make_snap(i), &[]); + } + let s = h.query( + Metric::MeshSize, + Duration::from_secs(24 * 3600), + Granularity::Slow, + ); + assert_eq!(s.values.len(), 1440); + assert!(s.values[..1410].iter().all(|v| v.is_nan())); + assert!(s.values[1410..].iter().all(|v| !v.is_nan())); + } + + #[test] + fn metric_parse_roundtrip() { + for m in ALL_METRICS { + assert_eq!(Metric::from_str(m.name()).unwrap(), *m); + } + assert!(Metric::from_str("bogus").is_err()); + } + + #[test] + fn peer_metric_parse_roundtrip() { + for m in ALL_PEER_METRICS { + assert_eq!(PeerMetric::from_str(m.name()).unwrap(), *m); + } + assert!(PeerMetric::from_str("bogus").is_err()); + } + + #[test] + fn granularity_parse() { + assert_eq!(Granularity::from_str("1s").unwrap(), Granularity::Fast); + assert_eq!(Granularity::from_str("1m").unwrap(), Granularity::Slow); + assert!(Granularity::from_str("1h").is_err()); + } + + #[test] + fn latest_and_recent() { + let mut h = StatsHistory::new(); + let t0 = Instant::now(); + for i in 0..5u64 { + h.tick(t0 + Duration::from_secs(i), &make_snap(i), &[]); + } + assert_eq!(h.latest(Metric::MeshSize), Some(14.0)); + let r = h.recent(Metric::MeshSize, 3); + assert_eq!(r, vec![12.0, 13.0, 14.0]); + let r2 = h.recent(Metric::MeshSize, 100); + assert_eq!(r2.len(), 5); + } + + #[test] + fn active_sessions_is_sampled_as_gauge() { + let mut h = StatsHistory::new(); + let t0 = Instant::now(); + for i in 0..3u64 { + let mut s = make_snap(i); + s.active_sessions = 10 + i; + h.tick(t0 + Duration::from_secs(i), &s, &[]); + } + let s = h.query( + Metric::ActiveSessions, + Duration::from_secs(5), + Granularity::Fast, + ); + assert_eq!(s.values.len(), 5); + assert!(s.values[..2].iter().all(|v| v.is_nan())); + assert_eq!(s.values[2..], [10.0, 11.0, 12.0]); + } + + #[test] + fn new_peer_backfills_nan_to_align_with_node_rings() { + let mut h = StatsHistory::new(); + let t0 = Instant::now(); + // Tick 5 times with no peers (node rings fill up). + for i in 0..5u64 { + h.tick(t0 + Duration::from_secs(i), &make_snap(i), &[]); + } + // Peer A joins on tick 6. + let a = make_addr(1); + h.tick( + t0 + Duration::from_secs(5), + &make_snap(5), + &[make_peer_snap(1, t0 + Duration::from_secs(5), 5)], + ); + // A's srtt ring has 5 NaN backfill + 1 real = 6 samples. A 60s + // window front-pads with 54 more NaN so the real value lands at + // the tail. + let s = h + .peer_query( + &a, + PeerMetric::SrttMs, + Duration::from_secs(60), + Granularity::Fast, + ) + .unwrap(); + assert_eq!(s.values.len(), 60); + assert!(s.values[..59].iter().all(|v| v.is_nan())); + assert_eq!(s.values[59], 15.0); + } + + #[test] + fn absent_peer_gets_nan_sample() { + let mut h = StatsHistory::new(); + let t0 = Instant::now(); + let a = make_addr(1); + // Tick 3 times with A present. + for i in 0..3u64 { + h.tick( + t0 + Duration::from_secs(i), + &make_snap(i), + &[make_peer_snap(1, t0 + Duration::from_secs(i), i)], + ); + } + // A disappears for 2 ticks. + for i in 3..5u64 { + h.tick(t0 + Duration::from_secs(i), &make_snap(i), &[]); + } + let s = h + .peer_query( + &a, + PeerMetric::SrttMs, + Duration::from_secs(60), + Granularity::Fast, + ) + .unwrap(); + assert_eq!(s.values.len(), 60); + // 55 NaN front-pad, then 3 real, then 2 NaN (A gone). + assert!(s.values[..55].iter().all(|v| v.is_nan())); + assert_eq!(s.values[55], 10.0); + assert_eq!(s.values[57], 12.0); + assert!(s.values[58].is_nan()); + assert!(s.values[59].is_nan()); + } + + #[test] + fn counter_decrease_emits_nan_and_rebaselines() { + let mut h = StatsHistory::new(); + let t0 = Instant::now(); + let a = make_addr(1); + // Three ticks with bytes_in increasing. + for (i, total) in [(0u64, 100u64), (1, 200), (2, 300)].iter().copied() { + let mut ps = make_peer_snap(1, t0 + Duration::from_secs(i), i); + ps.bytes_in_total = total; + h.tick(t0 + Duration::from_secs(i), &make_snap(i), &[ps]); + } + // Fourth tick: bytes_in drops to 50 (link reconnected). + let mut ps = make_peer_snap(1, t0 + Duration::from_secs(3), 3); + ps.bytes_in_total = 50; + h.tick(t0 + Duration::from_secs(3), &make_snap(3), &[ps]); + // Fifth tick: bytes_in grows to 80. + let mut ps = make_peer_snap(1, t0 + Duration::from_secs(4), 4); + ps.bytes_in_total = 80; + h.tick(t0 + Duration::from_secs(4), &make_snap(4), &[ps]); + + let s = h + .peer_query( + &a, + PeerMetric::BytesIn, + Duration::from_secs(60), + Granularity::Fast, + ) + .unwrap(); + assert_eq!(s.values.len(), 60); + // 55 NaN front-pad, then the 5 per-tick samples at the tail. + assert!(s.values[..55].iter().all(|v| v.is_nan())); + // First real tick has no prev → NaN. + assert!(s.values[55].is_nan()); + assert_eq!(s.values[56], 100.0); + assert_eq!(s.values[57], 100.0); + // Decrease → NaN, rebaseline to 50. + assert!(s.values[58].is_nan()); + // Next delta from new baseline. + assert_eq!(s.values[59], 30.0); + } + + #[test] + fn peer_eviction_fires_after_24h_of_silence() { + let mut h = StatsHistory::new(); + let t0 = Instant::now(); + let a = make_addr(1); + // One real sample for A at t=0. + h.tick(t0, &make_snap(0), &[make_peer_snap(1, t0, 0)]); + assert!(h.has_peer(&a)); + // Keep ticking every minute without A for 24 hours + 1 minute. + // (we tick at 60s intervals to avoid building a 24h fast ring) + let eviction = Duration::from_secs(PEER_EVICTION_SECS); + let mut i = 1u64; + loop { + let t = t0 + Duration::from_secs(i * 60); + h.tick(t, &make_snap(i), &[]); + if t.duration_since(t0) >= eviction { + break; + } + i += 1; + } + assert!(!h.has_peer(&a)); + } + + #[test] + fn nan_mean_downsample_skips_nan_samples() { + let mut h = StatsHistory::new(); + let t0 = Instant::now(); + let a = make_addr(1); + // 60 ticks alternating present / absent — 30 real SRTT samples + // at values 10, 12, 14, ..., 68, mean = 39. + for i in 0..60u64 { + if i.is_multiple_of(2) { + h.tick( + t0 + Duration::from_secs(i), + &make_snap(i), + &[make_peer_snap(1, t0 + Duration::from_secs(i), i)], + ); + } else { + h.tick(t0 + Duration::from_secs(i), &make_snap(i), &[]); + } + } + let s = h + .peer_query( + &a, + PeerMetric::SrttMs, + Duration::from_secs(60), + Granularity::Slow, + ) + .unwrap(); + assert_eq!(s.values.len(), 1); + let expected: f64 = (0..60u64) + .filter(|i| i.is_multiple_of(2)) + .map(|i| 10.0 + i as f64) + .sum::() + / 30.0; + assert!((s.values[0] - expected).abs() < 1e-9); + } + + #[test] + fn all_nan_window_downsamples_to_nan() { + let mut h = StatsHistory::new(); + let t0 = Instant::now(); + let a = make_addr(1); + // Introduce A, then silence it for 60+ ticks so one full slow + // sample accumulates entirely of NaN. + h.tick(t0, &make_snap(0), &[make_peer_snap(1, t0, 0)]); + for i in 1..=60u64 { + h.tick(t0 + Duration::from_secs(i), &make_snap(i), &[]); + } + let s = h + .peer_query( + &a, + PeerMetric::SrttMs, + Duration::from_secs(60 * 5), + Granularity::Slow, + ) + .unwrap(); + // We got one slow sample after 60 fast ticks. First 60 samples + // in the fast ring were 1 real + 59 NaN → Last = 10.0. But the + // boundary lands at fast_pushes == 60, AFTER pushing tick 59 + // (index 59). So the slow window covers fast indices 0..59, i.e. + // tick 0 (real) + ticks 1..59 (NaN) → Last = 10.0. Not all-NaN. + // + // Window is 300s / 60s = 5 slots; ring has 1 slow sample, so 4 + // leading NaN from the front-pad and the real value at the tail. + // + // Let's instead assert that the NEXT slow flush (after another + // 60 all-NaN ticks) is NaN. + assert_eq!(s.values.len(), 5); + assert!(s.values[..4].iter().all(|v| v.is_nan())); + assert_eq!(s.values[4], 10.0); + + for i in 61..=120u64 { + h.tick(t0 + Duration::from_secs(i), &make_snap(i), &[]); + } + let s = h + .peer_query( + &a, + PeerMetric::SrttMs, + Duration::from_secs(60 * 5), + Granularity::Slow, + ) + .unwrap(); + // 3 leading NaN from front-pad, then 2 real slow samples: the + // first Last=10.0, the second a fully-NaN slow window → NaN. + assert_eq!(s.values.len(), 5); + assert!(s.values[..3].iter().all(|v| v.is_nan())); + assert_eq!(s.values[3], 10.0); + assert!(s.values[4].is_nan()); + } + + #[test] + fn nan_serializes_to_json_null() { + let series = Series { + metric: "srtt_ms", + unit: "ms", + granularity_seconds: 1, + values: vec![1.0, f64::NAN, 3.0], + }; + let json = serde_json::to_value(&series).unwrap(); + let values = json.get("values").unwrap().as_array().unwrap(); + assert!(values[0].is_f64()); + assert!(values[1].is_null()); + assert!(values[2].is_f64()); + } +}