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