Merge branch 'master' into next

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,

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<Response>);
 async fn handle_connection_generic<S>(
     stream: S,
     control_tx: mpsc::Sender<ControlMessage>,
+    read_handle: ControlReadHandle,
 ) -> Result<(), Box<dyn std::error::Error>>
 where
     S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
@@ -73,15 +77,23 @@ where
             // Parse the request
             match serde_json::from_str::<Request>(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<ControlMessage>) {
+        pub(crate) async fn accept_loop(
+            self,
+            control_tx: mpsc::Sender<ControlMessage>,
+            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<ControlMessage>) {
+        pub(crate) async fn accept_loop(
+            self,
+            control_tx: mpsc::Sender<ControlMessage>,
+            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");
                     }
                 });

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();
@@ -977,6 +1029,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<NodeAddr, String>,
+    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<Value> = 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.
@@ -1125,7 +1337,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();
 
@@ -1152,6 +1382,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 {
@@ -1547,4 +1797,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::<Vec<_>>()
+        );
+        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<Value>| 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"
+        );
+    }
 }

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<StatsSnapshot>`: stats_history dual-ring + the
+//!   scalar gauges `show_status` needs, published from the tick.
+//! - `routing` (R3) — `ArcSwap<RoutingSnapshot>`: tree / bloom / coord /
+//!   identity, published from their announce / discovery mutators.
+//! - `entities` (R4) — `ArcSwap<EntitySnapshot>`: peers / sessions / links /
+//!   connections / transports, published per-entity with `Vec<Arc<Row>>`
+//!   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<NodeContext>,
+    /// Metrics registry (counters / gauges) for `show_stats_*`.
+    metrics: Arc<MetricsRegistry>,
+    /// stats_history dual-ring read copy + the scalar gauges/counts
+    /// `show_status` needs, published from the tick (R2, Q1-b).
+    stats: Arc<ArcSwap<StatsSnapshot>>,
+}
+
+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<NodeContext>,
+        metrics: Arc<MetricsRegistry>,
+        stats: Arc<ArcSwap<StatsSnapshot>>,
+    ) -> 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<Arc<StatsSnapshot>> {
+        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<Response> {
+    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,
+    }
+}

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<StatsHistory>,
+    /// Cached estimated mesh size, or `None` when no estimate is available.
+    pub estimated_mesh_size: Option<u64>,
+    /// Node operational state.
+    pub state: NodeState,
+    /// TUN device state.
+    pub tun_state: TunState,
+    /// TUN interface name, if active.
+    pub tun_name: Option<String>,
+    /// 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<HashMap<NodeAddr, String>>,
+}
+
+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()),
+        }
+    }
+}

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

src/node/handlers/rx_loop.rs

@@ -78,10 +78,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");

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;
@@ -406,6 +406,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<arc_swap::ArcSwap<crate::control::snapshot::StatsSnapshot>>,
+
     // === TUN Interface ===
     /// TUN device state.
     tun_state: TunState,
@@ -676,6 +682,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,
@@ -830,6 +839,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,
@@ -1413,6 +1425,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
@@ -1475,6 +1500,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 ===

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<f64>,
     slow: VecDeque<f64>,
@@ -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<Ring>,
     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<Ring>,
     peers: HashMap<NodeAddr, PeerStatsRings>,