From 1f457d84f985efede323b3378654e37e4b90f5f8 Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Wed, 10 Jun 2026 02:17:46 +0000 Subject: [PATCH 1/2] gateway: pin virtual-IP mapping while data-plane traffic flows The pool's TTL clock (VirtualIpMapping.last_referenced) advanced only on DNS re-query, never on traffic, and the mapping-TTL is wired equal to the DNS TTL, so an in-use mapping was forced to drain at TTL and reclaimed at the first zero-conntrack tick (a stale drain_start gave no grace effective protection), breaking long-lived, bursty, or DNS-cached clients. In tick(), refresh last_referenced whenever conntrack reports sessions > 0 so an actively used mapping never ages out, and recover a Draining mapping to Active (clearing drain_start) when traffic resumes, so a later drain gets a fresh grace window instead of a stale one. The Active arm now only drains an idle mapping. The DNS-TTL / idle-reclaim-TTL wiring is unchanged. Adds regression tests for continuous-traffic-survives-past-TTL, bursty drain-then-recover, and fresh-grace-on-redrain. --- src/gateway/pool.rs | 167 ++++++++++++++++++++++++++++++++++++++------ 1 file changed, 147 insertions(+), 20 deletions(-) diff --git a/src/gateway/pool.rs b/src/gateway/pool.rs index 90d1948..4b88bdf 100644 --- a/src/gateway/pool.rs +++ b/src/gateway/pool.rs @@ -226,6 +226,13 @@ impl VirtualIpPool { let sessions = conntrack.active_sessions(mapping.virtual_ip).unwrap_or(0); mapping.session_count = sessions; + // Live data-plane traffic pins the mapping: refresh the TTL + // clock whenever conntrack reports active sessions, so an + // in-use mapping never ages out from under the client. + if sessions > 0 { + mapping.last_referenced = now; + } + match mapping.state { MappingState::Allocated => { if sessions > 0 { @@ -249,26 +256,29 @@ impl VirtualIpPool { } } MappingState::Active => { + // The traffic refresh above keeps last_referenced == now + // while sessions > 0, so the TTL can only trip once the + // mapping is idle (no conntrack sessions). An actively used + // mapping never drains; an idle one enters the grace period. if now.duration_since(mapping.last_referenced) > ttl { - if sessions > 0 { - mapping.state = MappingState::Draining; - mapping.drain_start = Some(now); - debug!( - virtual_ip = %mapping.virtual_ip, - sessions, - "Mapping draining (TTL expired, sessions active)" - ); - } else { - // TTL expired and no sessions — start grace - mapping.state = MappingState::Draining; - mapping.drain_start = Some(now); - mapping.session_count = 0; - } + mapping.state = MappingState::Draining; + mapping.drain_start = Some(now); } } MappingState::Draining => { - if sessions == 0 - && let Some(drain_start) = mapping.drain_start + if sessions > 0 { + // Traffic resumed before reclamation: recover to + // Active and clear drain_start so the next drain + // gets a fresh grace window rather than reusing a + // stale one. + mapping.state = MappingState::Active; + mapping.drain_start = None; + debug!( + virtual_ip = %mapping.virtual_ip, + sessions, + "Draining mapping recovered to active (traffic resumed)" + ); + } else if let Some(drain_start) = mapping.drain_start && now.duration_since(drain_start) > grace { to_free.push(*node_addr); @@ -512,15 +522,14 @@ mod tests { assert!(events.is_empty()); assert_eq!(pool.mappings[&node].state, MappingState::Active); - // TTL expires but sessions remain → Draining + // TTL expires after sessions drop to 0 → Draining let later = now + std::time::Duration::from_secs(2); - ct.set(vip, 1); + ct.set(vip, 0); let events = pool.tick(later, &ct); assert!(events.is_empty()); assert_eq!(pool.mappings[&node].state, MappingState::Draining); - // Sessions drop to 0 but grace period not elapsed - ct.set(vip, 0); + // Still draining, grace period not elapsed let events = pool.tick(later, &ct); assert!(events.is_empty()); assert_eq!(pool.mappings[&node].state, MappingState::Draining); @@ -533,6 +542,124 @@ mod tests { assert_eq!(pool.mappings.len(), 0); } + #[test] + fn test_active_traffic_never_reclaimed() { + // A mapping with continuous sessions > 0 across many ticks + // spanning well past the TTL must never be reclaimed and must + // stay Active: live traffic refreshes last_referenced each tick. + let mut pool = VirtualIpPool::new("fd01::/120", 1, 1).unwrap(); + let mut ct = MockConntrack::new(); + let node = make_node_addr(1); + let mesh = make_mesh_addr(1); + + let (vip, _) = pool.allocate(node, mesh, "test.fips").unwrap(); + ct.set(vip, 2); + + let mut t = Instant::now(); + // First tick activates the mapping. + let events = pool.tick(t, &ct); + assert!(events.is_empty()); + assert_eq!(pool.mappings[&node].state, MappingState::Active); + + // Advance many TTL-spans with continuous traffic. + for _ in 0..10 { + t += std::time::Duration::from_secs(5); // 5x the 1s TTL + let events = pool.tick(t, &ct); + assert!(events.is_empty(), "mapping must not be reclaimed"); + assert_eq!( + pool.mappings[&node].state, + MappingState::Active, + "mapping must stay Active while traffic flows" + ); + } + assert_eq!(pool.mappings.len(), 1); + } + + #[test] + fn test_bursty_draining_recovers_to_active() { + // Active -> drains when sessions hit 0 -> regains sessions before + // grace elapses -> recovers to Active and is not freed. + let mut pool = VirtualIpPool::new("fd01::/120", 1, 5).unwrap(); + let mut ct = MockConntrack::new(); + let node = make_node_addr(1); + let mesh = make_mesh_addr(1); + + let (vip, _) = pool.allocate(node, mesh, "test.fips").unwrap(); + + // Activate with traffic. + ct.set(vip, 1); + let now = Instant::now(); + let events = pool.tick(now, &ct); + assert!(events.is_empty()); + assert_eq!(pool.mappings[&node].state, MappingState::Active); + + // TTL passes with sessions dropping to 0 -> Draining. + let drained = now + std::time::Duration::from_secs(2); + ct.set(vip, 0); + let events = pool.tick(drained, &ct); + assert!(events.is_empty()); + assert_eq!(pool.mappings[&node].state, MappingState::Draining); + + // Traffic resumes before grace (5s) elapses -> recover to Active. + let resumed = drained + std::time::Duration::from_secs(2); + ct.set(vip, 3); + let events = pool.tick(resumed, &ct); + assert!(events.is_empty()); + assert_eq!(pool.mappings[&node].state, MappingState::Active); + assert!(pool.mappings[&node].drain_start.is_none()); + assert_eq!(pool.mappings.len(), 1); + } + + #[test] + fn test_redrain_honors_fresh_grace_window() { + // After recovering from Draining, a subsequent drain must get a + // fresh drain_start so the full grace window is honored again, + // not reclaimed immediately off a stale drain_start. + let mut pool = VirtualIpPool::new("fd01::/120", 1, 5).unwrap(); + let mut ct = MockConntrack::new(); + let node = make_node_addr(1); + let mesh = make_mesh_addr(1); + + let (vip, _) = pool.allocate(node, mesh, "test.fips").unwrap(); + + // Activate. + ct.set(vip, 1); + let now = Instant::now(); + pool.tick(now, &ct); + assert_eq!(pool.mappings[&node].state, MappingState::Active); + + // First drain. + let first_drain = now + std::time::Duration::from_secs(2); + ct.set(vip, 0); + pool.tick(first_drain, &ct); + assert_eq!(pool.mappings[&node].state, MappingState::Draining); + + // Recover. + let recover = first_drain + std::time::Duration::from_secs(2); + ct.set(vip, 2); + pool.tick(recover, &ct); + assert_eq!(pool.mappings[&node].state, MappingState::Active); + + // Second drain begins; drain_start must be re-stamped fresh. + let second_drain = recover + std::time::Duration::from_secs(2); + ct.set(vip, 0); + pool.tick(second_drain, &ct); + assert_eq!(pool.mappings[&node].state, MappingState::Draining); + + // Just before the fresh grace window expires (5s): not reclaimed. + let before_grace = second_drain + std::time::Duration::from_secs(4); + let events = pool.tick(before_grace, &ct); + assert!(events.is_empty(), "fresh grace window must be honored"); + assert_eq!(pool.mappings.len(), 1); + + // After the fresh grace window: reclaimed. + let after_grace = second_drain + std::time::Duration::from_secs(6); + let events = pool.tick(after_grace, &ct); + assert_eq!(events.len(), 1); + assert!(matches!(events[0], PoolEvent::MappingRemoved { .. })); + assert_eq!(pool.mappings.len(), 0); + } + #[test] fn test_pool_status() { let mut pool = VirtualIpPool::new("fd01::/120", 60, 60).unwrap(); From c77e564462c8749a23518b9ad84ef847f3d72d2d Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Wed, 10 Jun 2026 02:31:54 +0000 Subject: [PATCH 2/2] control: serve high-traffic show_* queries off the rx_loop hot path Introduce a read-snapshot plane so pure-snapshot control queries render in the control-socket task instead of round-tripping the rx_loop, removing the head-of-line coupling that let a busy or slow rx_loop time out fipsctl and fipstop observability. - ControlReadHandle: a cloneable bundle the control accept loop holds, over the node's already-shared NodeContext and MetricsRegistry plus an ArcSwap-published StatsSnapshot. A snapshot_dispatch seam serves cut-over commands off-loop and falls through to the rx_loop for the rest, keeping the rx_loop's ownership of Node intact. - StatsSnapshot is published from the tick (the natural and sole mutator of stats_history), carrying the history rings plus the scalar gauges and counts show_status reports. Readers serve the latest snapshot unconditionally, with staleness bounded by the tick interval and no IO_TIMEOUT-coupled fallback. - Off-loop now: show_status, show_stats_history, show_stats_all_history, show_listening_sockets, show_stats_list, and a new counter-only show_metrics (exposed as fipsctl "stats metrics", the enabler for a Prometheus scraper at no hot-path cost). Queries that need live per-entity state (peers, links, sessions, routing, and the per-peer stats variants) stay on the rx_loop path pending later phases. Quartet green; forward-merge to next verified clean. --- src/bin/fipsctl.rs | 3 + src/control/mod.rs | 48 +++-- src/control/queries.rs | 402 ++++++++++++++++++++++++++++++++++- src/control/read_handle.rs | 127 +++++++++++ src/control/snapshot.rs | 79 +++++++ src/node/handlers/rx_loop.rs | 3 +- src/node/mod.rs | 50 ++++- src/node/stats_history.rs | 4 + 8 files changed, 700 insertions(+), 16 deletions(-) create mode 100644 src/control/read_handle.rs create mode 100644 src/control/snapshot.rs diff --git a/src/bin/fipsctl.rs b/src/bin/fipsctl.rs index bf621e6..3adf155 100644 --- a/src/bin/fipsctl.rs +++ b/src/bin/fipsctl.rs @@ -82,6 +82,8 @@ enum Commands { enum StatsCommands { /// List available history metrics List, + /// Dump current counter values for every protocol metric family + Metrics, /// List peers tracked in the stats history Peers, /// Fetch a time-series window for a metric @@ -448,6 +450,7 @@ fn main() { } Commands::Stats { what } => match what { StatsCommands::List => build_query("show_stats_list"), + StatsCommands::Metrics => build_query("show_metrics"), StatsCommands::Peers => build_query("show_stats_peers"), StatsCommands::History { metric, diff --git a/src/control/mod.rs b/src/control/mod.rs index 4e7a1e9..5e770df 100644 --- a/src/control/mod.rs +++ b/src/control/mod.rs @@ -13,9 +13,12 @@ pub mod firewall_state; pub mod listening; pub mod protocol; pub mod queries; +pub mod read_handle; +pub mod snapshot; use crate::config::ControlConfig; use protocol::{Request, Response}; +use read_handle::{ControlReadHandle, snapshot_dispatch}; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::sync::{mpsc, oneshot}; use tracing::{debug, info, warn}; @@ -36,6 +39,7 @@ pub type ControlMessage = (Request, oneshot::Sender); async fn handle_connection_generic( stream: S, control_tx: mpsc::Sender, + read_handle: ControlReadHandle, ) -> Result<(), Box> where S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin, @@ -73,15 +77,23 @@ where // Parse the request match serde_json::from_str::(line.trim()) { Ok(request) => { - // Send to main loop and wait for response - let (resp_tx, resp_rx) = oneshot::channel(); - if control_tx.send((request, resp_tx)).await.is_err() { - Response::error("node shutting down") - } else { - match tokio::time::timeout(IO_TIMEOUT, resp_rx).await { - Ok(Ok(resp)) => resp, - Ok(Err(_)) => Response::error("response channel closed"), - Err(_) => Response::error("query timeout"), + // First try to serve the request entirely off-loop from the + // read handle. In R0 this always returns None (no query is + // cut over yet); R1+ adds the per-command snapshot branches. + match snapshot_dispatch(&request, &read_handle) { + Some(resp) => resp, + None => { + // Send to main loop and wait for response + let (resp_tx, resp_rx) = oneshot::channel(); + if control_tx.send((request, resp_tx)).await.is_err() { + Response::error("node shutting down") + } else { + match tokio::time::timeout(IO_TIMEOUT, resp_rx).await { + Ok(Ok(resp)) => resp, + Ok(Err(_)) => Response::error("response channel closed"), + Err(_) => Response::error("query timeout"), + } + } } } } @@ -229,7 +241,11 @@ mod unix_impl { /// 3. Wait for the response via oneshot /// 4. Write the response as one line of JSON /// 5. Close the connection - pub async fn accept_loop(self, control_tx: mpsc::Sender) { + pub(crate) async fn accept_loop( + self, + control_tx: mpsc::Sender, + read_handle: ControlReadHandle, + ) { loop { let (stream, _addr) = match self.listener.accept().await { Ok(conn) => conn, @@ -240,8 +256,9 @@ mod unix_impl { }; let tx = control_tx.clone(); + let handle = read_handle.clone(); tokio::spawn(async move { - if let Err(e) = handle_connection_generic(stream, tx).await { + if let Err(e) = handle_connection_generic(stream, tx, handle).await { debug!(error = %e, "Control connection error"); } }); @@ -341,7 +358,11 @@ mod windows_impl { /// /// Each accepted connection is handled in a spawned task using the /// shared `handle_connection_generic` protocol handler. - pub async fn accept_loop(self, control_tx: mpsc::Sender) { + pub(crate) async fn accept_loop( + self, + control_tx: mpsc::Sender, + read_handle: ControlReadHandle, + ) { loop { let (stream, addr) = match self.listener.accept().await { Ok(conn) => conn, @@ -358,8 +379,9 @@ mod windows_impl { } let tx = control_tx.clone(); + let handle = read_handle.clone(); tokio::spawn(async move { - if let Err(e) = handle_connection_generic(stream, tx).await { + if let Err(e) = handle_connection_generic(stream, tx, handle).await { debug!(error = %e, "Control connection error"); } }); diff --git a/src/control/queries.rs b/src/control/queries.rs index abf7b82..d875e78 100644 --- a/src/control/queries.rs +++ b/src/control/queries.rs @@ -88,6 +88,58 @@ pub fn show_status(node: &Node) -> Value { }) } +/// Off-loop variant of [`show_status`]: renders from the +/// [`ControlReadHandle`](super::read_handle::ControlReadHandle) in the control +/// task. Reads the effectively-immutable `NodeContext`, the `MetricsRegistry` +/// counters, and the tick-published [`StatsSnapshot`](super::snapshot::StatsSnapshot) +/// (rings + scalar gauges/counts), with no `Node` state, so it never +/// round-trips the rx_loop. Output is byte-identical to [`show_status`]. +pub(crate) fn show_status_from_handle(handle: &super::read_handle::ControlReadHandle) -> Value { + let ctx = handle.context(); + let stats = handle.stats(); + let pid = std::process::id(); + let exe_path = std::env::current_exe() + .map(|p| p.display().to_string()) + .unwrap_or_else(|_| "-".into()); + let uptime_secs = ctx.started_at.elapsed().as_secs(); + let fwd = handle.metrics().forwarding.snapshot(); + + const SPARK_N: usize = 30; + let hist = &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": ctx.identity.npub(), + "node_addr": hex::encode(ctx.identity.node_addr().as_bytes()), + "ipv6_addr": format!("{}", ctx.identity.address()), + "state": format!("{}", stats.state), + "is_leaf_only": ctx.is_leaf_only, + "peer_count": stats.peer_count, + "session_count": stats.session_count, + "link_count": stats.link_count, + "transport_count": stats.transport_count, + "connection_count": stats.connection_count, + "tun_state": format!("{}", stats.tun_state), + "tun_name": stats.tun_name.as_deref().unwrap_or("-"), + "effective_ipv6_mtu": stats.effective_ipv6_mtu, + "control_socket": &ctx.config.node.control.socket_path, + "pid": pid, + "exe_path": exe_path, + "uptime_secs": uptime_secs, + "estimated_mesh_size": stats.estimated_mesh_size, + "forwarding": serde_json::to_value(&fwd).unwrap_or_default(), + "sparklines": sparklines, + }) +} + /// `show_acl` — Loaded peer ACL state. pub fn show_acl(node: &Node) -> Value { let status = node.peer_acl_status(); @@ -963,6 +1015,166 @@ pub fn show_stats_all_history(node: &Node, params: Option<&Value>) -> super::pro })) } +/// Off-loop display name for the stats-history error paths. The full +/// [`Node::peer_display_name`](crate::node::Node) lookup also consults the +/// host map and live peer/session tables, which are not in the snapshot; +/// off-loop we resolve a configured alias, else fall back to truncated hex. +/// Only reached on the "peer not tracked" error branch, never on the golden +/// happy path. +fn snapshot_display_name( + aliases: &std::collections::HashMap, + addr: &NodeAddr, +) -> String { + match aliases.get(addr) { + Some(name) => name.clone(), + None => addr.short_hex(), + } +} + +/// Off-loop variant of [`show_stats_history`]: serves one metric's series from +/// the tick-published [`StatsSnapshot`](super::snapshot::StatsSnapshot) rings +/// (node-level or per-peer) in the control task, off the rx_loop. Output is +/// byte-identical to [`show_stats_history`] for the series; the "peer not +/// tracked" error message uses [`snapshot_display_name`] (alias-or-hex). +pub(crate) fn show_stats_history_from_handle( + handle: &super::read_handle::ControlReadHandle, + 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 stats = handle.stats(); + let hist = &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: {}", + snapshot_display_name(&stats.peer_aliases, &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)) + } +} + +/// Off-loop variant of [`show_stats_all_history`]: serves every node-level +/// (or per-peer) series from the tick-published +/// [`StatsSnapshot`](super::snapshot::StatsSnapshot) rings, off the rx_loop. +/// Output is byte-identical to [`show_stats_all_history`]; the "peer not +/// tracked" error message uses [`snapshot_display_name`] (alias-or-hex). +pub(crate) fn show_stats_all_history_from_handle( + handle: &super::read_handle::ControlReadHandle, + 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 stats = handle.stats(); + let hist = &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: {}", + snapshot_display_name(&stats.peer_aliases, &addr) + )); + } + ALL_PEER_METRICS + .iter() + .map(|m| { + let s = hist + .peer_query(&addr, *m, window, granularity) + .unwrap_or_else(|| 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. @@ -1111,7 +1323,25 @@ pub fn show_stats_history_all_peers( /// [`crate::control::listening`] and [`crate::control::firewall_state`] /// for the per-half implementations. pub fn show_listening_sockets(node: &Node) -> Value { - let fips0 = crate::FipsAddress::from_node_addr(node.identity().node_addr()).to_ipv6(); + render_listening_sockets(node.identity().node_addr()) +} + +/// Off-loop variant of [`show_listening_sockets`]: renders from the +/// [`ControlReadHandle`](super::read_handle::ControlReadHandle) in the control +/// task. Reads only the node identity (from `NodeContext`) plus host-OS facts +/// (`/proc` socket enumeration, nftables firewall classification) — no `Node` +/// state — so it never round-trips the rx_loop. +pub(crate) fn show_listening_sockets_from_handle( + handle: &super::read_handle::ControlReadHandle, +) -> Value { + render_listening_sockets(handle.context().identity.node_addr()) +} + +/// Shared renderer for the listening-sockets panel. Given the node's +/// `NodeAddr` it derives the fips0 address, enumerates listening sockets, and +/// classifies each against the shipped firewall baseline. +fn render_listening_sockets(node_addr: &NodeAddr) -> Value { + let fips0 = crate::FipsAddress::from_node_addr(node_addr).to_ipv6(); let sockets = super::listening::enumerate(fips0); let classifier = super::firewall_state::FilterClassifier::query(); @@ -1138,6 +1368,26 @@ pub fn show_listening_sockets(node: &Node) -> Value { }) } +/// `show_metrics` — Counter-family snapshot served off the rx_loop. +/// +/// Renders every counter family in the [`MetricsRegistry`] as a flat JSON +/// object keyed by family name. Each family's value is its +/// `*StatsSnapshot` (a `u64`-per-counter struct). Counter-only by design: +/// gauges and histograms that need live `Node` state are out of scope and +/// stay on the rx_loop path. This is the Prometheus-exporter enabler — an +/// automated scraper reads this without ever touching the hot path. +pub(crate) fn show_metrics_from_handle(handle: &super::read_handle::ControlReadHandle) -> Value { + let m = handle.metrics(); + json!({ + "forwarding": m.forwarding.snapshot(), + "discovery": m.discovery.snapshot(), + "tree": m.tree.snapshot(), + "bloom": m.bloom.snapshot(), + "congestion": m.congestion.snapshot(), + "errors": m.errors.snapshot(), + }) +} + /// Dispatch a command string to the appropriate query function. pub fn dispatch(node: &Node, command: &str, params: Option<&Value>) -> super::protocol::Response { match command { @@ -1533,4 +1783,154 @@ mod tests { ); } } + + // ---- off-loop (snapshot_dispatch) coverage --------------------------- + + /// `show_metrics` is counter-only and served off the rx_loop. Raw + /// counter values are runtime-varying, so instead of a value-exact + /// golden fixture this pins the *shape*: every counter family appears + /// as a key, each maps to an object, and a representative counter key + /// is present in each family. This catches family renames / drops + /// without flaking on live values. + #[test] + fn show_metrics_shape_covers_all_families() { + let node = build_test_node(); + let handle = node.control_read_handle(); + let value = show_metrics_from_handle(&handle); + let obj = value.as_object().expect("show_metrics renders an object"); + + let expected_families = [ + ("forwarding", "received_packets"), + ("discovery", "req_received"), + ("tree", "accepted"), + ("bloom", "accepted"), + ("congestion", "ce_forwarded"), + ("errors", "coords_required"), + ]; + assert_eq!( + obj.len(), + expected_families.len(), + "show_metrics has exactly {} counter families, got keys {:?}", + expected_families.len(), + obj.keys().collect::>() + ); + for (family, sample_key) in expected_families { + let fam = obj + .get(family) + .unwrap_or_else(|| panic!("missing counter family {family}")) + .as_object() + .unwrap_or_else(|| panic!("family {family} is not an object")); + assert!( + fam.contains_key(sample_key), + "family {family} missing expected counter {sample_key}" + ); + // On a fresh node every counter is zero. + assert_eq!( + fam.get(sample_key).and_then(Value::as_u64), + Some(0), + "fresh-node counter {family}.{sample_key} should be 0" + ); + } + } + + /// The three R1 cutover queries are served off-loop via + /// `snapshot_dispatch`; everything else (state-bearing queries, + /// mutations) returns `None` and falls through to the rx_loop path. + #[test] + fn snapshot_dispatch_serves_only_cutover_queries() { + use super::super::protocol::Request; + use super::super::read_handle::snapshot_dispatch; + + let node = build_test_node(); + let handle = node.control_read_handle(); + + let req = |command: &str| Request { + command: command.to_string(), + params: None, + }; + + // Cut over to off-loop serving. The parameterized stats-series + // queries need their params to render; everything else is + // parameterless. + let req_params = |command: &str, params: Option| Request { + command: command.to_string(), + params, + }; + let off_loop = [ + ("show_listening_sockets", None), + ("show_stats_list", None), + ("show_metrics", None), + ("show_status", None), + ( + "show_stats_history", + Some(json!({ "metric": "mesh_size", "window": "10s", "granularity": "1s" })), + ), + ( + "show_stats_all_history", + Some(json!({ "window": "10s", "granularity": "1s" })), + ), + ]; + for (cmd, params) in off_loop { + let resp = snapshot_dispatch(&req_params(cmd, params), &handle) + .unwrap_or_else(|| panic!("{cmd} must be served off-loop")); + assert_eq!(resp.status, "ok", "{cmd} off-loop response not ok"); + } + + // Still on the rx_loop path: state-bearing queries that need live + // peer membership / npub, and all mutations. + for cmd in [ + "show_peers", + "show_stats_peers", + "show_stats_history_all_peers", + "connect", + "disconnect", + ] { + assert!( + snapshot_dispatch(&req(cmd), &handle).is_none(), + "{cmd} must fall through to the rx_loop path" + ); + } + } + + /// The tick-published `StatsSnapshot` reflects node state: after a + /// simulated `record_stats_history()` tick, the snapshot's counts and + /// scalar gauges match the node, and the off-loop `show_status` render + /// equals the on-loop `show_status` render byte-for-byte. + #[test] + fn stats_snapshot_reflects_state_after_tick() { + let mut node = build_test_node(); + + // Before any tick the seeded snapshot is empty. + let handle = node.control_read_handle(); + assert!( + !handle.stats().history.has_data(), + "seed snapshot has no history before first tick" + ); + + // Advance one tick (the natural publisher site). + node.record_stats_history(); + + let handle = node.control_read_handle(); + let snap = handle.stats(); + assert!( + snap.history.has_data(), + "snapshot history reflects the tick" + ); + // Scalar gauges / counts match the node's live accessors. + assert_eq!(snap.peer_count, node.peer_count()); + assert_eq!(snap.session_count, node.session_count()); + assert_eq!(snap.link_count, node.link_count()); + assert_eq!(snap.transport_count, node.transport_count()); + assert_eq!(snap.connection_count, node.connection_count()); + assert_eq!(snap.estimated_mesh_size, node.estimated_mesh_size()); + assert_eq!(snap.effective_ipv6_mtu, node.effective_ipv6_mtu()); + + // Off-loop render must equal the on-loop render byte-for-byte. + let on_loop = render(show_status(&node)); + let off_loop = render(show_status_from_handle(&handle)); + assert_eq!( + on_loop, off_loop, + "off-loop show_status must match on-loop output" + ); + } } diff --git a/src/control/read_handle.rs b/src/control/read_handle.rs new file mode 100644 index 0000000..38167f6 --- /dev/null +++ b/src/control/read_handle.rs @@ -0,0 +1,127 @@ +//! Read-only handle the control accept loop holds, so pure-snapshot `show_*` +//! queries can render off the rx_loop hot path instead of round-tripping the +//! mpsc → rx_loop oneshot. +//! +//! This is the stable seam of the control read-isolation milestone +//! (TASK-2026-0152, phase R0). The handle bundles the state that is already +//! independently shareable, and grows one `ArcSwap` snapshot cell per phase as +//! each subsystem's read state is published from its natural mutator: +//! +//! - `context` / `metrics` — already `Arc`-shared (refactor steps B/C). +//! - `stats` (R2) — `ArcSwap`: stats_history dual-ring + the +//! scalar gauges `show_status` needs, published from the tick. +//! - `routing` (R3) — `ArcSwap`: tree / bloom / coord / +//! identity, published from their announce / discovery mutators. +//! - `entities` (R4) — `ArcSwap`: peers / sessions / links / +//! connections / transports, published per-entity with `Vec>` +//! structural sharing. +//! +//! Publisher placement follows the Q1 rules in +//! `design/fast-path-refactoring-r0-read-handle.md`: every snapshot is +//! published at its state's natural mutation site (on-change), never by the +//! contended rx_loop task it is meant to bypass. +//! +//! R0 ships only the type and the dispatch seam ([`snapshot_dispatch`]); no +//! query reads the handle yet. Cutover begins in R1. + +use std::sync::Arc; + +use arc_swap::ArcSwap; + +use crate::node::context::NodeContext; +use crate::node::metrics::MetricsRegistry; + +use super::protocol::{Request, Response}; +use super::snapshot::StatsSnapshot; + +/// Cloneable read-only view of node state for off-loop control serving. +/// +/// All fields are `Arc` / `ArcSwap` handles, so cloning is cheap and a clone +/// can be held by every accepted control connection. Fields are consumed +/// starting R1 as `show_*` queries cut over to off-loop rendering; until then +/// they are wired but unread. +#[derive(Clone)] +pub(crate) struct ControlReadHandle { + /// Effectively-immutable node context (config, identity, limits). + context: Arc, + /// Metrics registry (counters / gauges) for `show_stats_*`. + metrics: Arc, + /// stats_history dual-ring read copy + the scalar gauges/counts + /// `show_status` needs, published from the tick (R2, Q1-b). + stats: Arc>, +} + +impl ControlReadHandle { + /// Build the handle from the node's already-shared state. Called once at + /// control-socket spawn time; the result is cloned per connection. The + /// `stats` cell is the same `Arc` the tick publishes into, so every clone + /// observes fresh snapshots. + pub(crate) fn new( + context: Arc, + metrics: Arc, + stats: Arc>, + ) -> Self { + Self { + context, + metrics, + stats, + } + } + + /// Borrow the effectively-immutable node context. + pub(crate) fn context(&self) -> &NodeContext { + &self.context + } + + /// Borrow the metrics registry. + pub(crate) fn metrics(&self) -> &MetricsRegistry { + &self.metrics + } + + /// Load the latest published stats snapshot (the freshest available by + /// construction; no IO_TIMEOUT staleness gate, per Q1-e). + pub(crate) fn stats(&self) -> arc_swap::Guard> { + self.stats.load() + } +} + +/// Attempt to serve a request entirely from the read handle, off the rx_loop. +/// +/// Returns `Some(response)` when the command is a pure-snapshot query that has +/// been cut over to off-loop rendering, or `None` when it must take the +/// mpsc → rx_loop path (parameterized queries, mutations, and any query not +/// yet cut over). +/// +/// Cutover queries (R1) read only `NodeContext` / `MetricsRegistry` (the state +/// the read handle already bundles) plus host-OS facts (`/proc`, nftables), so +/// they render entirely in the control task without touching `Node`. +pub(crate) fn snapshot_dispatch(request: &Request, handle: &ControlReadHandle) -> Option { + use crate::control::queries; + + match request.command.as_str() { + "show_listening_sockets" => Some(Response::ok( + queries::show_listening_sockets_from_handle(handle), + )), + "show_stats_list" => Some(Response::ok(queries::show_stats_list())), + "show_metrics" => Some(Response::ok(queries::show_metrics_from_handle(handle))), + // R2: served from the tick-published `StatsSnapshot` (rings + scalar + // gauges/counts). `show_status` and the two node-level/per-peer series + // queries carry enough data in the snapshot to render faithfully + // off-loop, including the parameterized series selectors (the snapshot + // holds the full rings, so any metric / window / granularity is + // satisfiable). `show_stats_peers` and `show_stats_history_all_peers` + // stay on the rx_loop path: they need live peer membership + // (`is_active`) and per-peer npub, which are Category-E state not yet + // in the snapshot. + "show_status" => Some(Response::ok(queries::show_status_from_handle(handle))), + "show_stats_history" => Some(queries::show_stats_history_from_handle( + handle, + request.params.as_ref(), + )), + "show_stats_all_history" => Some(queries::show_stats_all_history_from_handle( + handle, + request.params.as_ref(), + )), + _ => None, + } +} diff --git a/src/control/snapshot.rs b/src/control/snapshot.rs new file mode 100644 index 0000000..9d22890 --- /dev/null +++ b/src/control/snapshot.rs @@ -0,0 +1,79 @@ +//! Read-side state snapshots published from the node's natural mutators so +//! pure-snapshot `show_*` queries render off the rx_loop hot path. +//! +//! [`StatsSnapshot`] is the R2 reference implementation of the canonical +//! snapshot pattern (see `design/fast-path-refactoring-r0-read-handle.md`): a +//! read-only data bundle published via `ArcSwap` from the tick after +//! `StatsHistory::tick()`. It carries +//! +//! - the `stats_history` read-side rings (the "dual-ring": the live mutable +//! ring stays on the tick; this is the cloned read copy), and +//! - the cheap scalar gauges `show_status` needs (`estimated_mesh_size`, +//! node `state`, `tun_state`, `tun_name`, `effective_ipv6_mtu`, and the +//! peer / session / link / connection / transport counts), plus +//! `peer_aliases` (effectively immutable after construction). +//! +//! The snapshot holds *data*, not rendered `Response` envelopes (Q1-d): +//! rendering happens in the control task off the rx_loop. Staleness is bounded +//! by the tick interval and is never staler than the underlying data, which +//! also advances only on the tick (Q1-b). + +use std::collections::HashMap; +use std::sync::Arc; + +use crate::identity::NodeAddr; +use crate::node::NodeState; +use crate::node::stats_history::StatsHistory; +use crate::upper::tun::TunState; + +/// Read-only snapshot of the stats-history rings plus the scalar gauges and +/// counts `show_status` reports. Published from the tick (Q1-b). +#[derive(Clone)] +pub(crate) struct StatsSnapshot { + /// Cloned read copy of the history rings (the dual-ring read side). + pub history: Arc, + /// Cached estimated mesh size, or `None` when no estimate is available. + pub estimated_mesh_size: Option, + /// Node operational state. + pub state: NodeState, + /// TUN device state. + pub tun_state: TunState, + /// TUN interface name, if active. + pub tun_name: Option, + /// Effective IPv6 MTU over the mesh. + pub effective_ipv6_mtu: u16, + /// Number of pending connections (handshake in progress). + pub connection_count: usize, + /// Number of authenticated peers. + pub peer_count: usize, + /// Number of active links. + pub link_count: usize, + /// Number of active transports. + pub transport_count: usize, + /// Number of active sessions. + pub session_count: usize, + /// Configured peer aliases, keyed by `NodeAddr`. Effectively immutable + /// after construction; shared to avoid a per-tick map clone. + pub peer_aliases: Arc>, +} + +impl StatsSnapshot { + /// Build an empty snapshot for seeding the `ArcSwap` cell at construction, + /// before the first tick has published real state. + pub(crate) fn empty() -> Self { + Self { + history: Arc::new(StatsHistory::new()), + estimated_mesh_size: None, + state: NodeState::Created, + tun_state: TunState::Disabled, + tun_name: None, + effective_ipv6_mtu: 0, + connection_count: 0, + peer_count: 0, + link_count: 0, + transport_count: 0, + session_count: 0, + peer_aliases: Arc::new(HashMap::new()), + } + } +} diff --git a/src/node/handlers/rx_loop.rs b/src/node/handlers/rx_loop.rs index ae88002..e46e02e 100644 --- a/src/node/handlers/rx_loop.rs +++ b/src/node/handlers/rx_loop.rs @@ -76,10 +76,11 @@ impl Node { if self.config().node.control.enabled { let config = self.config().node.control.clone(); let tx = control_tx.clone(); + let read_handle = self.control_read_handle(); tokio::spawn(async move { match ControlSocket::bind(&config) { Ok(socket) => { - socket.accept_loop(tx).await; + socket.accept_loop(tx, read_handle).await; } Err(e) => { warn!(error = %e, "Failed to bind control socket"); diff --git a/src/node/mod.rs b/src/node/mod.rs index 1b71a0e..42adfcf 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -6,7 +6,7 @@ mod acl; mod bloom; -mod context; +pub(crate) mod context; #[cfg(unix)] pub(crate) mod decrypt_worker; mod discovery_rate_limit; @@ -390,6 +390,12 @@ pub struct Node { /// Time-series history of node-level metrics (1s/1m rings). stats_history: stats_history::StatsHistory, + /// Read-side snapshot of `stats_history` plus the scalar gauges/counts + /// `show_status` reports, published from the tick (the natural mutator) + /// so those queries serve off the rx_loop. The dual-ring read copy: the + /// live mutable `stats_history` above stays on the tick. + stats_snapshot: std::sync::Arc>, + // === TUN Interface === /// TUN device state. tun_state: TunState, @@ -654,6 +660,9 @@ impl Node { stats: stats::NodeStats::new(), metrics: std::sync::Arc::new(metrics::MetricsRegistry::new()), stats_history: stats_history::StatsHistory::new(), + stats_snapshot: std::sync::Arc::new(arc_swap::ArcSwap::from_pointee( + crate::control::snapshot::StatsSnapshot::empty(), + )), tun_state, tun_name: None, tun_tx: None, @@ -806,6 +815,9 @@ impl Node { stats: stats::NodeStats::new(), metrics: std::sync::Arc::new(metrics::MetricsRegistry::new()), stats_history: stats_history::StatsHistory::new(), + stats_snapshot: std::sync::Arc::new(arc_swap::ArcSwap::from_pointee( + crate::control::snapshot::StatsSnapshot::empty(), + )), tun_state, tun_name: None, tun_tx: None, @@ -1373,6 +1385,19 @@ impl Node { &self.metrics } + /// Build a [`ControlReadHandle`](crate::control::read_handle::ControlReadHandle) + /// over this node's already-shared `NodeContext` and `MetricsRegistry`. + /// + /// Used at control-socket spawn time so pure-snapshot `show_*` queries + /// render off the rx_loop. Cloneable; cheap (all `Arc` clones). + pub(crate) fn control_read_handle(&self) -> crate::control::read_handle::ControlReadHandle { + crate::control::read_handle::ControlReadHandle::new( + self.context.clone(), + self.metrics.clone(), + self.stats_snapshot.clone(), + ) + } + /// Get the stats history collector. pub fn stats_history(&self) -> &stats_history::StatsHistory { &self.stats_history @@ -1435,6 +1460,29 @@ impl Node { .collect(); self.stats_history.tick(now, &snap, &peer_snaps); + + // Publish the read-side snapshot (R2 dual-ring, Q1-b). The tick is the + // natural and sole mutator of `stats_history`, so publishing here can + // never produce false staleness: the snapshot and the underlying data + // advance together. This is data, not a rendered response (Q1-d), and + // it is published only here, not in a monolithic per-tick rebuild of + // every query (Q1-c). It also is not gated behind any slow I/O on the + // tick the way the abandoned 2edc8a1 republish was. + let snapshot = crate::control::snapshot::StatsSnapshot { + history: std::sync::Arc::new(self.stats_history.clone()), + estimated_mesh_size: self.estimated_mesh_size, + state: self.state, + tun_state: self.tun_state, + tun_name: self.tun_name.clone(), + effective_ipv6_mtu: self.effective_ipv6_mtu(), + connection_count: self.connections.len(), + peer_count: self.peers.len(), + link_count: self.links.len(), + transport_count: self.transports.len(), + session_count: self.sessions.len(), + peer_aliases: std::sync::Arc::new(self.peer_aliases.clone()), + }; + self.stats_snapshot.store(std::sync::Arc::new(snapshot)); } // === TUN Interface === diff --git a/src/node/stats_history.rs b/src/node/stats_history.rs index 6f594da..53e202b 100644 --- a/src/node/stats_history.rs +++ b/src/node/stats_history.rs @@ -292,6 +292,7 @@ impl FromStr for Granularity { } /// One metric's dual-tier ring. +#[derive(Clone)] struct Ring { fast: VecDeque, slow: VecDeque, @@ -306,6 +307,7 @@ struct Ring { /// 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. +#[derive(Clone)] struct DownsampleAccum { sum: f64, /// Count of non-NaN samples. @@ -425,6 +427,7 @@ pub struct Series { } /// One peer's per-metric rings plus lifecycle metadata. +#[derive(Clone)] pub struct PeerStatsRings { rings: Vec, first_seen: Instant, @@ -516,6 +519,7 @@ impl PeerStatsRings { /// Per-metric ring storage for node-level metrics plus a map of /// per-peer rings keyed by `NodeAddr`. +#[derive(Clone)] pub struct StatsHistory { rings: Vec, peers: HashMap,