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/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");

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<arc_swap::ArcSwap<crate::control::snapshot::StatsSnapshot>>,
+
     // === 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 ===

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>,