From 81cd10d5db6bb5004e5fd23e3fd831b9cf4b601b Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Wed, 10 Jun 2026 17:45:49 +0000 Subject: [PATCH] control: serve the full show_* read surface off the rx_loop via a read-snapshot plane Complete the control-plane read-isolation work: every pure-read show_* query now renders in the control accept task from published read snapshots, so none round-trips the data-plane receive loop. Only the mutating connect/disconnect commands still reach that loop. Three subsystem snapshots are published via ArcSwap and served through the read handle's snapshot_dispatch: - A routing read view (spanning tree, bloom filters, coordinate cache, identity cache, and the discovery F-queue summary scalars), published from the tick, serving show_tree/show_bloom/show_cache/show_routing/ show_identity_cache. - A per-entity read view (peers, sessions, links, connections, transports, and the MMP link/session views) as Vec> tables reconciled against the prior snapshot so a republish reuses unchanged rows by pointer and re-allocates only changed or new rows, keeping the per-tick publish cost bounded as the peer/session count grows. Serves show_peers/show_sessions/show_links/show_connections/show_transports/ show_mmp. - The stats snapshot is extended with the peer-ACL status and a per-peer metadata map (is_active, npub, display name), resolved at publish time, serving show_acl and the two per-peer stats queries. Display names and other cross-subsystem fields are resolved at publish time; time-relative fields are derived at render time from captured absolute timestamps, so rendered output is byte-identical to the prior on-loop handlers, which are retained as the equality oracle. With every read query served off-loop, the show_* branch is removed from the rx_loop control handler and the now-dead on-loop dispatcher deleted. The snapshot projections are forward-compatible with the later structural extraction of the derived-state and session tables: they become thin views over the extracted types without changing the read-handle interface. --- src/control/queries.rs | 1325 +++++++++++++++++++++++++++++++--- src/control/read_handle.rs | 68 +- src/control/snapshot.rs | 636 ++++++++++++++++ src/node/handlers/rx_loop.rs | 22 +- src/node/mod.rs | 591 ++++++++++++++- 5 files changed, 2543 insertions(+), 99 deletions(-) diff --git a/src/control/queries.rs b/src/control/queries.rs index d875e78..b71dbb1 100644 --- a/src/control/queries.rs +++ b/src/control/queries.rs @@ -26,7 +26,7 @@ fn now_ms() -> u64 { } /// Classify a DualEwma trend as "rising", "falling", or "stable". -fn trend_label(short: f64, long: f64) -> &'static str { +pub(crate) fn trend_label(short: f64, long: f64) -> &'static str { if !short.is_finite() || !long.is_finite() || long == 0.0 { return "stable"; } @@ -159,6 +159,29 @@ pub fn show_acl(node: &Node) -> Value { }) } +/// Off-loop variant of [`show_acl`]: renders from the tick-published +/// [`StatsSnapshot`](super::snapshot::StatsSnapshot) ACL status. The ACL is an +/// `arc_swap::ArcSwap` reloaded only on the tick, and the status is a +/// cheap projection of it captured at the same tick, so this renders entirely +/// off the rx_loop. Output is byte-identical to [`show_acl`]. +pub(crate) fn show_acl_from_handle(handle: &super::read_handle::ControlReadHandle) -> Value { + let status = &handle.stats().acl_status; + + json!({ + "allow_file": status.allow_file, + "deny_file": status.deny_file, + "enforcement_active": status.enforcement_active, + "effective_mode": status.effective_mode, + "default_decision": status.default_decision, + "allow_all": status.allow_all, + "deny_all": status.deny_all, + "allow_file_entries": status.allow_file_entries, + "deny_file_entries": status.deny_file_entries, + "allow_entries": status.allow_entries, + "deny_entries": status.deny_entries, + }) +} + /// `show_peers` — Authenticated peers. pub fn show_peers(node: &Node) -> Value { let tree = node.tree_state(); @@ -320,6 +343,159 @@ pub fn show_peers(node: &Node) -> Value { json!({ "peers": peers }) } +/// Render a snapshot [`EntityMmp`](super::snapshot::EntityMmp) into the inline +/// MMP JSON block, with the quality-index key named `quality_key` (`lqi` for +/// peers, `sqi` for sessions). Reproduces the on-loop key insertion order +/// exactly. `path_mtu` is emitted (inside the leading literal) only when +/// present (session-layer); for peers it is `None` and omitted. +fn render_entity_mmp(mmp: &super::snapshot::EntityMmp, quality_key: &str) -> Value { + // The on-loop `show_sessions` block places loss_rate/etx/goodput_bps/ + // delivery ratios/path_mtu in the leading json! literal, while `show_peers` + // emits mode first then loss_rate/etx/goodput_bps/delivery ratios after the + // conditional srtt_ms. Both orderings are reproduced by branching on whether + // a session path_mtu is present. + if let Some(path_mtu) = mmp.path_mtu { + // Session-layer ordering (show_sessions). + let mut j = json!({ + "mode": mmp.mode, + "loss_rate": mmp.loss_rate, + "etx": mmp.etx, + "goodput_bps": mmp.goodput_bps, + "delivery_ratio_forward": mmp.delivery_ratio_forward, + "delivery_ratio_reverse": mmp.delivery_ratio_reverse, + "path_mtu": path_mtu, + }); + if let Some(srtt) = mmp.srtt_ms { + j["srtt_ms"] = json!(srtt); + } + if let Some(smoothed_loss) = mmp.smoothed_loss { + j["smoothed_loss"] = json!(smoothed_loss); + } + if let Some(smoothed_etx) = mmp.smoothed_etx { + j["smoothed_etx"] = json!(smoothed_etx); + } + if let Some(qi) = mmp.quality_index { + j[quality_key] = json!(qi); + } + j + } else { + // Link-layer ordering (show_peers). + let mut j = json!({ + "mode": mmp.mode, + }); + if let Some(srtt) = mmp.srtt_ms { + j["srtt_ms"] = json!(srtt); + } + j["loss_rate"] = json!(mmp.loss_rate); + j["etx"] = json!(mmp.etx); + j["goodput_bps"] = json!(mmp.goodput_bps); + j["delivery_ratio_forward"] = json!(mmp.delivery_ratio_forward); + j["delivery_ratio_reverse"] = json!(mmp.delivery_ratio_reverse); + if let Some(smoothed_loss) = mmp.smoothed_loss { + j["smoothed_loss"] = json!(smoothed_loss); + } + if let Some(smoothed_etx) = mmp.smoothed_etx { + j["smoothed_etx"] = json!(smoothed_etx); + } + if let Some(qi) = mmp.quality_index { + j[quality_key] = json!(qi); + } + j + } +} + +/// Off-loop variant of [`show_peers`]: renders from the tick-published +/// [`EntitySnapshot`](super::snapshot::EntitySnapshot) peer table (display +/// names, tree-relationship flags, Nostr-traversal state resolved at publish +/// time). Output is byte-identical to [`show_peers`]. +pub(crate) fn show_peers_from_handle(handle: &super::read_handle::ControlReadHandle) -> Value { + let entities = handle.entities(); + let peers: Vec = entities + .peers + .iter() + .map(|peer| { + let mut peer_json = json!({ + "node_addr": hex::encode(peer.node_addr.as_bytes()), + "npub": peer.npub, + "display_name": peer.display_name, + "ipv6_addr": peer.ipv6_addr, + "connectivity": peer.connectivity, + "link_id": peer.link_id, + "authenticated_at_ms": peer.authenticated_at_ms, + "last_seen_ms": peer.last_seen_ms, + "has_tree_position": peer.has_tree_position, + "has_bloom_filter": peer.has_bloom_filter, + "filter_sequence": peer.filter_sequence, + "is_parent": peer.is_parent, + "is_child": peer.is_child, + }); + + if let Some(addr) = &peer.transport_addr { + peer_json["transport_addr"] = json!(addr); + } + + if let Some(link) = &peer.link_info { + peer_json["direction"] = json!(link.direction); + if let Some(tt) = &link.transport_type { + peer_json["transport_type"] = json!(tt); + } + } + + if let Some(depth) = peer.tree_depth { + peer_json["tree_depth"] = json!(depth); + } + + peer_json["stats"] = json!({ + "packets_sent": peer.stats.packets_sent, + "packets_recv": peer.stats.packets_recv, + "bytes_sent": peer.stats.bytes_sent, + "bytes_recv": peer.stats.bytes_recv, + }); + + peer_json["replay_suppressed"] = json!(peer.replay_suppressed); + peer_json["consecutive_decrypt_failures"] = json!(peer.consecutive_decrypt_failures); + + let nostr = &peer.nostr_traversal; + peer_json["nostr_traversal"] = json!({ + "consecutive_failures": nostr.consecutive_failures, + "in_cooldown": nostr.cooldown_until_ms.is_some(), + "cooldown_until_ms": nostr.cooldown_until_ms.map(|t| json!(t)).unwrap_or(Value::Null), + "last_observed_skew_ms": nostr + .last_observed_skew_ms + .map(|s| json!(s)) + .unwrap_or(Value::Null), + }); + + if let Some(noise) = &peer.noise { + peer_json["noise"] = json!({ + "send_counter": noise.send_counter, + "highest_recv_counter": noise.highest_recv_counter, + }); + } + + if let Some(idx) = peer.our_session_index { + peer_json["our_session_index"] = json!(format!("{:08x}", idx)); + } + + if peer.rekey_in_progress { + peer_json["rekey_in_progress"] = json!(true); + } + if peer.rekey_draining { + peer_json["rekey_draining"] = json!(true); + } + peer_json["current_k_bit"] = json!(peer.current_k_bit); + + if let Some(mmp) = &peer.mmp { + peer_json["mmp"] = render_entity_mmp(mmp, "lqi"); + } + + peer_json + }) + .collect(); + + json!({ "peers": peers }) +} + /// `show_links` — Active links. pub fn show_links(node: &Node) -> Value { let links: Vec = node @@ -347,6 +523,36 @@ pub fn show_links(node: &Node) -> Value { json!({ "links": links }) } +/// Off-loop variant of [`show_links`]: renders from the tick-published +/// [`EntitySnapshot`](super::snapshot::EntitySnapshot) link table. Output is +/// byte-identical to [`show_links`]. +pub(crate) fn show_links_from_handle(handle: &super::read_handle::ControlReadHandle) -> Value { + let entities = handle.entities(); + let links: Vec = entities + .links + .iter() + .map(|link| { + json!({ + "link_id": link.link_id, + "transport_id": link.transport_id, + "remote_addr": link.remote_addr, + "direction": link.direction, + "state": link.state, + "created_at_ms": link.created_at_ms, + "stats": { + "packets_sent": link.stats.packets_sent, + "packets_recv": link.stats.packets_recv, + "bytes_sent": link.stats.bytes_sent, + "bytes_recv": link.stats.bytes_recv, + "last_recv_ms": link.stats.last_recv_ms, + }, + }) + }) + .collect(); + + json!({ "links": links }) +} + /// `show_tree` — Spanning tree state. pub fn show_tree(node: &Node) -> Value { let tree = node.tree_state(); @@ -406,6 +612,61 @@ pub fn show_tree(node: &Node) -> Value { }) } +/// Off-loop variant of [`show_tree`]: renders from the tick-published +/// [`RoutingSnapshot`](super::snapshot::RoutingSnapshot) tree view (display +/// names resolved at publish time) plus the `tree` counter family from the +/// `MetricsRegistry`. Output is byte-identical to [`show_tree`]. +pub(crate) fn show_tree_from_handle(handle: &super::read_handle::ControlReadHandle) -> Value { + let routing = handle.routing(); + let tree = &routing.tree; + + let coords: Vec = tree + .my_coords + .iter() + .map(|a| hex::encode(a.as_bytes())) + .collect(); + + let peers: Vec = tree + .peers + .iter() + .map(|peer| { + let mut peer_json = json!({ + "node_addr": hex::encode(peer.node_addr.as_bytes()), + "display_name": peer.display_name, + }); + if let Some(coords) = &peer.coords { + let coord_path: Vec = coords + .coord_path + .iter() + .map(|a| hex::encode(a.as_bytes())) + .collect(); + peer_json["depth"] = json!(coords.depth); + peer_json["root"] = json!(hex::encode(coords.root.as_bytes())); + peer_json["coords"] = json!(coord_path); + peer_json["distance_to_us"] = json!(coords.distance_to_us); + } + peer_json + }) + .collect(); + + let tree_stats = handle.metrics().tree.snapshot(); + + json!({ + "my_node_addr": hex::encode(tree.my_node_addr.as_bytes()), + "root": hex::encode(tree.root.as_bytes()), + "is_root": tree.is_root, + "depth": tree.depth, + "my_coords": coords, + "parent": hex::encode(tree.parent.as_bytes()), + "parent_display_name": tree.parent_display_name, + "declaration_sequence": tree.declaration_sequence, + "declaration_signed": tree.declaration_signed, + "peer_tree_count": tree.peer_tree_count, + "peers": peers, + "stats": serde_json::to_value(&tree_stats).unwrap_or_default(), + }) +} + /// `show_sessions` — End-to-end sessions. pub fn show_sessions(node: &Node) -> Value { let sessions: Vec = node @@ -490,6 +751,55 @@ pub fn show_sessions(node: &Node) -> Value { json!({ "sessions": sessions }) } +/// Off-loop variant of [`show_sessions`]: renders from the tick-published +/// [`EntitySnapshot`](super::snapshot::EntitySnapshot) session table (display +/// name, npub, established/handshake state resolved at publish time). Output is +/// byte-identical to [`show_sessions`]. +pub(crate) fn show_sessions_from_handle(handle: &super::read_handle::ControlReadHandle) -> Value { + let entities = handle.entities(); + let sessions: Vec = entities + .sessions + .iter() + .map(|session| { + let mut session_json = json!({ + "remote_addr": hex::encode(session.remote_addr.as_bytes()), + "display_name": session.display_name, + "state": session.state, + "is_initiator": session.is_initiator, + "last_activity_ms": session.last_activity_ms, + }); + + session_json["npub"] = json!(session.npub); + + session_json["stats"] = json!({ + "packets_sent": session.stats.packets_sent, + "packets_recv": session.stats.packets_recv, + "bytes_sent": session.stats.bytes_sent, + "bytes_recv": session.stats.bytes_recv, + }); + + if let Some(resend) = session.resend_count { + session_json["resend_count"] = json!(resend); + } + + if let Some(est) = &session.established { + session_json["session_start_ms"] = json!(est.session_start_ms); + session_json["current_k_bit"] = json!(est.current_k_bit); + session_json["coords_warmup_remaining"] = json!(est.coords_warmup_remaining); + session_json["is_draining"] = json!(est.is_draining); + } + + if let Some(mmp) = &session.mmp { + session_json["mmp"] = render_entity_mmp(mmp, "sqi"); + } + + session_json + }) + .collect(); + + json!({ "sessions": sessions }) +} + /// `show_bloom` — Bloom filter state. pub fn show_bloom(node: &Node) -> Value { let bloom = node.bloom_state(); @@ -534,6 +844,52 @@ pub fn show_bloom(node: &Node) -> Value { }) } +/// Off-loop variant of [`show_bloom`]: renders from the tick-published +/// [`RoutingSnapshot`](super::snapshot::RoutingSnapshot) bloom view (display +/// names resolved at publish time) plus the `bloom` counter family from the +/// `MetricsRegistry`. Output is byte-identical to [`show_bloom`]. +pub(crate) fn show_bloom_from_handle(handle: &super::read_handle::ControlReadHandle) -> Value { + let routing = handle.routing(); + let bloom = &routing.bloom; + + let leaf_deps: Vec = bloom + .leaf_dependents + .iter() + .map(|addr| hex::encode(addr.as_bytes())) + .collect(); + + let peer_filters: Vec = bloom + .peer_filters + .iter() + .map(|peer| { + let mut pf = json!({ + "peer": hex::encode(peer.peer.as_bytes()), + "display_name": peer.display_name, + "has_filter": peer.has_filter, + "filter_sequence": peer.filter_sequence, + }); + if let Some(filter) = &peer.filter { + pf["estimated_count"] = json!(filter.estimated_count); + pf["set_bits"] = json!(filter.set_bits); + pf["fill_ratio"] = json!(filter.fill_ratio); + } + pf + }) + .collect(); + + let bloom_stats = handle.metrics().bloom.snapshot(); + + json!({ + "own_node_addr": hex::encode(bloom.own_node_addr.as_bytes()), + "is_leaf_only": bloom.is_leaf_only, + "sequence": bloom.sequence, + "leaf_dependent_count": bloom.leaf_dependents.len(), + "leaf_dependents": leaf_deps, + "peer_filters": peer_filters, + "stats": serde_json::to_value(&bloom_stats).unwrap_or_default(), + }) +} + /// `show_mmp` — MMP metrics summary. pub fn show_mmp(node: &Node) -> Value { // Link-layer MMP per peer @@ -629,6 +985,101 @@ pub fn show_mmp(node: &Node) -> Value { }) } +/// Off-loop variant of [`show_mmp`]: renders from the tick-published +/// [`EntitySnapshot`](super::snapshot::EntitySnapshot) mmp tables (display +/// names and trend labels resolved at publish time). Output is byte-identical +/// to [`show_mmp`]. +pub(crate) fn show_mmp_from_handle(handle: &super::read_handle::ControlReadHandle) -> Value { + let entities = handle.entities(); + + let peers: Vec = entities + .mmp_peers + .iter() + .map(|peer| { + let mut link_layer = json!({ + "loss_rate": peer.loss_rate, + "etx": peer.etx, + "goodput_bps": peer.goodput_bps, + "spin_bit_role": if peer.spin_bit_initiator { "initiator" } else { "responder" }, + }); + + if let Some(smoothed_loss) = peer.smoothed_loss { + link_layer["smoothed_loss"] = json!(smoothed_loss); + } + if let Some(smoothed_etx) = peer.smoothed_etx { + link_layer["smoothed_etx"] = json!(smoothed_etx); + } + if let Some(srtt) = peer.srtt_ms { + link_layer["srtt_ms"] = json!(srtt); + if let Some(lqi) = peer.lqi { + link_layer["lqi"] = json!(lqi); + } + } + + if let Some(t) = peer.trends.rtt_trend { + link_layer["rtt_trend"] = json!(t); + } + if let Some(t) = peer.trends.loss_trend { + link_layer["loss_trend"] = json!(t); + } + if let Some(t) = peer.trends.goodput_trend { + link_layer["goodput_trend"] = json!(t); + } + if let Some(t) = peer.trends.jitter_trend { + link_layer["jitter_trend"] = json!(t); + } + + link_layer["delivery_ratio_forward"] = json!(peer.delivery_ratio_forward); + link_layer["delivery_ratio_reverse"] = json!(peer.delivery_ratio_reverse); + link_layer["ecn_ce_count"] = json!(peer.ecn_ce_count); + + json!({ + "peer": hex::encode(peer.peer.as_bytes()), + "display_name": peer.display_name, + "mode": peer.mode, + "link_layer": link_layer, + }) + }) + .collect(); + + let sessions: Vec = entities + .mmp_sessions + .iter() + .map(|session| { + let mut session_layer = json!({ + "loss_rate": session.loss_rate, + "etx": session.etx, + "path_mtu": session.path_mtu, + }); + + if let Some(smoothed_loss) = session.smoothed_loss { + session_layer["smoothed_loss"] = json!(smoothed_loss); + } + if let Some(smoothed_etx) = session.smoothed_etx { + session_layer["smoothed_etx"] = json!(smoothed_etx); + } + if let Some(srtt) = session.srtt_ms { + session_layer["srtt_ms"] = json!(srtt); + if let Some(sqi) = session.sqi { + session_layer["sqi"] = json!(sqi); + } + } + + json!({ + "remote": hex::encode(session.remote.as_bytes()), + "display_name": session.display_name, + "mode": session.mode, + "session_layer": session_layer, + }) + }) + .collect(); + + json!({ + "peers": peers, + "sessions": sessions, + }) +} + /// `show_cache` — Coordinate cache stats and entries. pub fn show_cache(node: &Node) -> Value { let cache = node.coord_cache(); @@ -673,6 +1124,54 @@ pub fn show_cache(node: &Node) -> Value { }) } +/// Off-loop variant of [`show_cache`]: renders from the tick-published +/// [`RoutingSnapshot`](super::snapshot::RoutingSnapshot) coord-cache view +/// (display names resolved at publish time). `age_ms` is derived at render +/// time from the captured `created_at`, exactly as [`show_cache`] computed it, +/// so the rendered age stays fresh relative to the read. Output is +/// byte-identical to [`show_cache`]. +pub(crate) fn show_cache_from_handle(handle: &super::read_handle::ControlReadHandle) -> Value { + let routing = handle.routing(); + let cache = &routing.cache; + let now = now_ms(); + + let entries: Vec = cache + .entries + .iter() + .map(|entry| { + let fips_addr = crate::identity::FipsAddress::from_node_addr(&entry.node_addr); + let coord_path: Vec = entry + .coord_path + .iter() + .map(|a| hex::encode(a.as_bytes())) + .collect(); + let mut entry_json = json!({ + "node_addr": hex::encode(entry.node_addr.as_bytes()), + "display_name": entry.display_name, + "ipv6_addr": format!("{}", fips_addr), + "depth": entry.depth, + "coords": coord_path, + "age_ms": now.saturating_sub(entry.created_at), + "last_used_ms": entry.last_used_ms, + }); + if let Some(mtu) = entry.path_mtu { + entry_json["path_mtu"] = json!(mtu); + } + entry_json + }) + .collect(); + + json!({ + "count": cache.count, + "max_entries": cache.max_entries, + "fill_ratio": cache.fill_ratio, + "default_ttl_ms": cache.default_ttl_ms, + "expired": cache.expired, + "avg_age_ms": cache.avg_age_ms, + "entries": entries, + }) +} + /// `show_connections` — Pending handshakes. pub fn show_connections(node: &Node) -> Value { let now = now_ms(); @@ -699,6 +1198,40 @@ pub fn show_connections(node: &Node) -> Value { json!({ "connections": connections }) } +/// Off-loop variant of [`show_connections`]: renders from the tick-published +/// [`EntitySnapshot`](super::snapshot::EntitySnapshot) connection table. +/// `idle_ms` is derived at render time from the captured `last_activity_ms`, +/// exactly as [`show_connections`] computed it. Output is byte-identical to +/// [`show_connections`]. +pub(crate) fn show_connections_from_handle( + handle: &super::read_handle::ControlReadHandle, +) -> Value { + let entities = handle.entities(); + let now = now_ms(); + let connections: Vec = entities + .connections + .iter() + .map(|conn| { + let mut conn_json = json!({ + "link_id": conn.link_id, + "direction": conn.direction, + "handshake_state": conn.handshake_state, + "started_at_ms": conn.started_at_ms, + "idle_ms": now.saturating_sub(conn.last_activity_ms), + "resend_count": conn.resend_count, + }); + + if let Some(expected) = &conn.expected_peer { + conn_json["expected_peer"] = json!(expected); + } + + conn_json + }) + .collect(); + + json!({ "connections": connections }) +} + /// `show_transports` — Transport instances. pub fn show_transports(node: &Node) -> Value { let transports: Vec = node @@ -739,6 +1272,50 @@ pub fn show_transports(node: &Node) -> Value { json!({ "transports": transports }) } +/// Off-loop variant of [`show_transports`]: renders from the tick-published +/// [`EntitySnapshot`](super::snapshot::EntitySnapshot) transport table. The +/// `stats` and `tor_monitoring` blocks are already-projected `serde_json::Value` +/// data captured at publish time. Output is byte-identical to +/// [`show_transports`]. +pub(crate) fn show_transports_from_handle(handle: &super::read_handle::ControlReadHandle) -> Value { + let entities = handle.entities(); + let transports: Vec = entities + .transports + .iter() + .map(|t| { + let mut t_json = json!({ + "transport_id": t.transport_id, + "type": t.transport_type, + "state": t.state, + "mtu": t.mtu, + }); + + if let Some(name) = &t.name { + t_json["name"] = json!(name); + } + if let Some(addr) = &t.local_addr { + t_json["local_addr"] = json!(addr); + } + + if let Some(mode) = &t.tor_mode { + t_json["tor_mode"] = json!(mode); + } + if let Some(onion) = &t.onion_address { + t_json["onion_address"] = json!(onion); + } + if let Some(monitoring) = &t.tor_monitoring { + t_json["tor_monitoring"] = monitoring.clone(); + } + + t_json["stats"] = t.stats.clone(); + + t_json + }) + .collect(); + + json!({ "transports": transports }) +} + /// `show_routing` — Routing table summary and node statistics. pub fn show_routing(node: &Node) -> Value { let cache = node.coord_cache(); @@ -790,6 +1367,62 @@ pub fn show_routing(node: &Node) -> Value { }) } +/// Off-loop variant of [`show_routing`]: renders from the tick-published +/// [`RoutingSnapshot`](super::snapshot::RoutingSnapshot) (coord-cache count, +/// identity-cache count, F-queue scalars, pending lookups, retries — display +/// names resolved at publish time) plus the forwarding / discovery / error / +/// congestion counter families from the `MetricsRegistry`. `age_ms` is derived +/// at render time. Output is byte-identical to [`show_routing`]. +pub(crate) fn show_routing_from_handle(handle: &super::read_handle::ControlReadHandle) -> Value { + let routing = handle.routing(); + let view = &routing.routing; + let now = now_ms(); + let metrics = handle.metrics(); + + let lookups: Vec = view + .pending_lookups + .iter() + .map(|lookup| { + json!({ + "target": hex::encode(lookup.target.as_bytes()), + "display_name": lookup.display_name, + "initiated_ms": lookup.initiated_ms, + "last_sent_ms": lookup.last_sent_ms, + "attempt": lookup.attempt, + "age_ms": now.saturating_sub(lookup.initiated_ms), + }) + }) + .collect(); + + let retries: Vec = view + .retries + .iter() + .map(|state| { + json!({ + "node_addr": hex::encode(state.node_addr.as_bytes()), + "display_name": state.display_name, + "retry_count": state.retry_count, + "retry_after_ms": state.retry_after_ms, + "auto_reconnect": state.auto_reconnect, + }) + }) + .collect(); + + json!({ + "coord_cache_entries": routing.cache.count, + "identity_cache_entries": routing.identity.entries.len(), + "pending_lookups": lookups, + "pending_tun_destinations": view.pending_tun_destinations, + "pending_tun_packets": view.pending_tun_packets, + "recent_requests": view.recent_requests, + "retries": retries, + "forwarding": serde_json::to_value(metrics.forwarding.snapshot()).unwrap_or_default(), + "discovery": serde_json::to_value(metrics.discovery.snapshot()).unwrap_or_default(), + "error_signals": serde_json::to_value(metrics.errors.snapshot()).unwrap_or_default(), + "congestion": serde_json::to_value(metrics.congestion.snapshot()).unwrap_or_default(), + }) +} + /// `show_identity_cache` — Known node identities. /// /// Lists every node whose public key has been cached by this daemon. @@ -822,6 +1455,41 @@ pub fn show_identity_cache(node: &Node) -> Value { }) } +/// Off-loop variant of [`show_identity_cache`]: renders from the tick-published +/// [`RoutingSnapshot`](super::snapshot::RoutingSnapshot) identity view (npub / +/// ipv6 / display name resolved at publish time). `age_ms` is derived at render +/// time from the captured `last_seen_ms`, exactly as [`show_identity_cache`] +/// computed it. Output is byte-identical to [`show_identity_cache`]. +pub(crate) fn show_identity_cache_from_handle( + handle: &super::read_handle::ControlReadHandle, +) -> Value { + let routing = handle.routing(); + let identity = &routing.identity; + let now = now_ms(); + + let entries: Vec = identity + .entries + .iter() + .map(|entry| { + json!({ + "node_addr": hex::encode(entry.node_addr.as_bytes()), + "npub": entry.npub, + "display_name": entry.display_name, + "ipv6_addr": entry.ipv6_addr, + "last_seen_ms": entry.last_seen_ms, + "age_ms": now.saturating_sub(entry.last_seen_ms), + }) + }) + .collect(); + let count = entries.len(); + + json!({ + "entries": entries, + "count": count, + "max_entries": identity.max_entries, + }) +} + /// `show_stats_list` — Enumerate available history metrics and their units. pub fn show_stats_list() -> Value { let metrics: Vec = ALL_METRICS @@ -1228,6 +1896,66 @@ pub fn show_stats_peers(node: &Node) -> Value { json!({ "peers": peers, "count": peers.len() }) } +/// Off-loop variant of [`show_stats_peers`]: renders from the tick-published +/// [`StatsSnapshot`](super::snapshot::StatsSnapshot). The lifecycle timestamps +/// (`first_seen` / `last_contact`) and the tracked-peer set come from the +/// snapshot's `history` rings; the cross-subsystem fields (`is_active`, npub, +/// display name) come from the per-peer `peer_meta` resolved at publish time. +/// The `*_secs_ago` deltas are derived at render time from the captured +/// `Instant`s, exactly as [`show_stats_peers`] computed them. Output is +/// byte-identical to [`show_stats_peers`]. +pub(crate) fn show_stats_peers_from_handle( + handle: &super::read_handle::ControlReadHandle, +) -> Value { + let stats = handle.stats(); + let hist = &stats.history; + let meta = &stats.peer_meta; + let now = std::time::Instant::now(); + + let mut peers: Vec = hist + .peers() + .map(|(addr, rings)| { + let last_contact_secs = now.duration_since(rings.last_contact()).as_secs(); + let first_seen_secs = now.duration_since(rings.first_seen()).as_secs(); + let (is_active, npub, display_name) = match meta.get(addr) { + Some(m) => (m.is_active, m.npub.clone(), m.display_name.clone()), + None => (false, hex::encode(addr.as_bytes()), addr.short_hex()), + }; + json!({ + "npub": npub, + "node_addr": hex::encode(addr.as_bytes()), + "display_name": display_name, + "is_active": is_active, + "first_seen_secs_ago": first_seen_secs, + "last_contact_secs_ago": last_contact_secs, + }) + }) + .collect(); + + // Stable display order: active peers first, then by display name. + peers.sort_by(|a, b| { + let a_active = a + .get("is_active") + .and_then(|v| v.as_bool()) + .unwrap_or(false); + let b_active = b + .get("is_active") + .and_then(|v| v.as_bool()) + .unwrap_or(false); + match (b_active, a_active) { + (true, false) => std::cmp::Ordering::Greater, + (false, true) => std::cmp::Ordering::Less, + _ => a + .get("display_name") + .and_then(|v| v.as_str()) + .unwrap_or("") + .cmp(b.get("display_name").and_then(|v| v.as_str()).unwrap_or("")), + } + }); + + json!({ "peers": peers, "count": peers.len() }) +} + /// `show_stats_history_all_peers` — One metric across every tracked /// peer in one round trip. Backs the fipstop MetricByPeer grid view. /// @@ -1316,6 +2044,100 @@ pub fn show_stats_history_all_peers( })) } +/// Off-loop variant of [`show_stats_history_all_peers`]: serves one per-peer +/// metric across every tracked peer from the tick-published +/// [`StatsSnapshot`](super::snapshot::StatsSnapshot). The per-peer rings and +/// tracked-peer set come from the snapshot's `history`; `is_active` and the +/// display name come from the per-peer `peer_meta` resolved at publish time. +/// Output is byte-identical to [`show_stats_history_all_peers`]. +pub(crate) fn show_stats_history_all_peers_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_all_peers"); + }; + + let metric_name = match params.get("metric").and_then(|v| v.as_str()) { + Some(v) => v, + None => return Response::error("missing 'metric' parameter"), + }; + let metric = match PeerMetric::from_str(metric_name) { + Ok(m) => m, + Err(e) => return Response::error(e), + }; + + let window_str = params + .get("window") + .and_then(|v| v.as_str()) + .unwrap_or("10m"); + let window = match parse_duration(window_str) { + Ok(d) => d, + Err(e) => return Response::error(e), + }; + + let granularity_str = params + .get("granularity") + .and_then(|v| v.as_str()) + .unwrap_or("1s"); + let granularity = match Granularity::from_str(granularity_str) { + Ok(g) => g, + Err(e) => return Response::error(e), + }; + + let stats = handle.stats(); + let hist = &stats.history; + let meta = &stats.peer_meta; + let peer_addrs: Vec = hist.peer_addrs().copied().collect(); + + let mut peers: Vec = peer_addrs + .iter() + .filter_map(|addr| { + let s = hist.peer_query(addr, metric, window, granularity)?; + let (is_active, display_name) = match meta.get(addr) { + Some(m) => (m.is_active, m.display_name.clone()), + None => (false, addr.short_hex()), + }; + Some(json!({ + "node_addr": hex::encode(addr.as_bytes()), + "display_name": display_name, + "is_active": is_active, + "values": serde_json::to_value(&s.values).unwrap_or(Value::Null), + })) + }) + .collect(); + + // Active peers first, then by display name. + peers.sort_by(|a, b| { + let a_active = a + .get("is_active") + .and_then(|v| v.as_bool()) + .unwrap_or(false); + let b_active = b + .get("is_active") + .and_then(|v| v.as_bool()) + .unwrap_or(false); + match (b_active, a_active) { + (true, false) => std::cmp::Ordering::Greater, + (false, true) => std::cmp::Ordering::Less, + _ => a + .get("display_name") + .and_then(|v| v.as_str()) + .unwrap_or("") + .cmp(b.get("display_name").and_then(|v| v.as_str()).unwrap_or("")), + } + }); + + Response::ok(json!({ + "metric": metric.name(), + "unit": metric.unit(), + "granularity_seconds": granularity.seconds(), + "window_seconds": window.as_secs(), + "peers": peers, + })) +} + /// `show_listening_sockets` — IPv6 listeners reachable from fips0, /// each annotated with its current `inet fips` filter classification. /// @@ -1388,31 +2210,14 @@ pub(crate) fn show_metrics_from_handle(handle: &super::read_handle::ControlReadH }) } -/// Dispatch a command string to the appropriate query function. -pub fn dispatch(node: &Node, command: &str, params: Option<&Value>) -> super::protocol::Response { - match command { - "show_acl" => super::protocol::Response::ok(show_acl(node)), - "show_status" => super::protocol::Response::ok(show_status(node)), - "show_peers" => super::protocol::Response::ok(show_peers(node)), - "show_links" => super::protocol::Response::ok(show_links(node)), - "show_tree" => super::protocol::Response::ok(show_tree(node)), - "show_sessions" => super::protocol::Response::ok(show_sessions(node)), - "show_bloom" => super::protocol::Response::ok(show_bloom(node)), - "show_mmp" => super::protocol::Response::ok(show_mmp(node)), - "show_cache" => super::protocol::Response::ok(show_cache(node)), - "show_connections" => super::protocol::Response::ok(show_connections(node)), - "show_transports" => super::protocol::Response::ok(show_transports(node)), - "show_routing" => super::protocol::Response::ok(show_routing(node)), - "show_identity_cache" => super::protocol::Response::ok(show_identity_cache(node)), - "show_listening_sockets" => super::protocol::Response::ok(show_listening_sockets(node)), - "show_stats_list" => super::protocol::Response::ok(show_stats_list()), - "show_stats_history" => show_stats_history(node, params), - "show_stats_all_history" => show_stats_all_history(node, params), - "show_stats_peers" => super::protocol::Response::ok(show_stats_peers(node)), - "show_stats_history_all_peers" => show_stats_history_all_peers(node, params), - _ => super::protocol::Response::error(format!("unknown command: {}", command)), - } -} +// No on-loop `show_*` dispatcher remains: every pure-read query is served +// off-loop from the read handle via +// [`snapshot_dispatch`](super::read_handle::snapshot_dispatch) in the control +// accept task, and the rx_loop's control path now carries only the mutating +// COMMAND handlers (`connect` / `disconnect`) in +// [`crate::control::commands::dispatch`]. The on-loop `show_*` functions above +// are retained as the byte-identity oracle the snapshot tests compare each +// off-loop renderer against. #[cfg(test)] mod tests { @@ -1734,56 +2539,118 @@ mod tests { assert_snapshot("show_stats_history_all_peers", &render_response(resp)); } - /// Sanity check: every handler advertised in `dispatch` is also - /// covered by a snapshot test above. If a new handler is added - /// without a matching snapshot, this test fails. + /// The five Category-D queries cut over to off-loop serving in R3. Served + /// via `snapshot_dispatch`; coverage asserted in + /// `snapshot_dispatch_serves_category_d_queries` below. + const OFF_LOOP_CATEGORY_D: &[&str] = &[ + "show_tree", + "show_bloom", + "show_cache", + "show_routing", + "show_identity_cache", + ]; + + /// The six Category-E queries cut over to off-loop serving in R4. Served via + /// `snapshot_dispatch`; coverage asserted in + /// `snapshot_dispatch_serves_category_e_queries`. + const OFF_LOOP_CATEGORY_E: &[&str] = &[ + "show_peers", + "show_sessions", + "show_links", + "show_connections", + "show_transports", + "show_mmp", + ]; + + /// Milestone-completion contract: every pure-read `show_*` query is served + /// off-loop via `snapshot_dispatch`, and the rx_loop control path carries no + /// `show_*` arm at all — only the mutating COMMAND handlers (`connect` / + /// `disconnect`) reach it. This test enumerates the full read surface and + /// asserts each renders off-loop, then asserts the mutations do NOT. #[test] - fn dispatch_covers_all_snapshotted_handlers() { - let expected = [ - "show_status", - "show_acl", - "show_peers", - "show_links", - "show_tree", - "show_sessions", - "show_bloom", - "show_mmp", - "show_cache", - "show_connections", - "show_transports", - "show_routing", - "show_identity_cache", - "show_listening_sockets", - "show_stats_list", - "show_stats_history", - "show_stats_all_history", - "show_stats_peers", - "show_stats_history_all_peers", + fn snapshot_dispatch_serves_every_read_query() { + use super::super::protocol::Request; + use super::super::read_handle::snapshot_dispatch; + + let mut node = build_test_node(); + // Publish every snapshot from the tick before reading. + node.record_stats_history(); + let handle = node.control_read_handle(); + + // The complete pure-read `show_*` surface, with params where required. + let read_queries: &[(&str, Option)] = &[ + ("show_status", None), + ("show_acl", None), + ("show_listening_sockets", None), + ("show_metrics", None), + ("show_stats_list", None), + ( + "show_stats_history", + Some(json!({ "metric": "mesh_size", "window": "10s", "granularity": "1s" })), + ), + ( + "show_stats_all_history", + Some(json!({ "window": "10s", "granularity": "1s" })), + ), + ("show_stats_peers", None), + ( + "show_stats_history_all_peers", + Some(json!({ "metric": "srtt_ms", "window": "10s", "granularity": "1s" })), + ), + ("show_tree", None), + ("show_bloom", None), + ("show_cache", None), + ("show_routing", None), + ("show_identity_cache", None), + ("show_peers", None), + ("show_sessions", None), + ("show_links", None), + ("show_connections", None), + ("show_transports", None), + ("show_mmp", None), ]; - assert_eq!(expected.len(), 19, "expected exactly 19 query handlers"); - let node = build_test_node(); - for cmd in expected { - // Each must dispatch successfully (status == "ok") with - // minimal params. Handlers requiring params get them. - let params = match cmd { - "show_stats_history" => Some(json!({ - "metric": "mesh_size", "window": "10s", "granularity": "1s" - })), - "show_stats_all_history" => Some(json!({ "window": "10s", "granularity": "1s" })), - "show_stats_history_all_peers" => Some(json!({ - "metric": "srtt_ms", "window": "10s", "granularity": "1s" - })), - _ => None, + for (cmd, params) in read_queries { + let req = Request { + command: cmd.to_string(), + params: params.clone(), }; - let resp = dispatch(&node, cmd, params.as_ref()); - assert_eq!( - resp.status, "ok", - "dispatch({cmd}) returned status={} message={:?}", - resp.status, resp.message + let resp = snapshot_dispatch(&req, &handle) + .unwrap_or_else(|| panic!("{cmd} must be served off-loop")); + assert_eq!(resp.status, "ok", "{cmd} off-loop response not ok"); + } + + // Mutations are NOT served off-loop; they take the rx_loop COMMAND path. + for cmd in ["connect", "disconnect"] { + let req = Request { + command: cmd.to_string(), + params: None, + }; + assert!( + snapshot_dispatch(&req, &handle).is_none(), + "{cmd} must fall through to the rx_loop command path" ); } } + /// Structural confirmation that the rx_loop no longer dispatches `show_*`: + /// the rx_loop source carries no `queries::dispatch` call and no + /// `starts_with("show_")` routing branch. Reads the committed source of + /// `src/node/handlers/rx_loop.rs` and asserts both markers are absent. This + /// is the milestone's "remove `show_*` from the data-plane dispatch path" + /// invariant, guarded against regression. + #[test] + fn rx_loop_has_no_show_dispatch() { + let src = include_str!("../node/handlers/rx_loop.rs"); + assert!( + !src.contains("queries::dispatch"), + "rx_loop must not call queries::dispatch (show_* served off-loop)" + ); + assert!( + !src.contains("starts_with(\"show_\")"), + "rx_loop must not branch on a show_* command prefix" + ); + } + // ---- off-loop (snapshot_dispatch) coverage --------------------------- /// `show_metrics` is counter-only and served off the rx_loop. Raw @@ -1833,11 +2700,13 @@ mod tests { } } - /// 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. + /// The R1/R2 scalar-and-series queries are served off-loop via + /// `snapshot_dispatch`; mutations return `None` and take the rx_loop COMMAND + /// path. (`show_stats_peers` / `show_stats_history_all_peers`, formerly + /// asserted on-loop here, were cut over in R5 — see + /// `snapshot_dispatch_serves_every_read_query` for the full read surface.) #[test] - fn snapshot_dispatch_serves_only_cutover_queries() { + fn snapshot_dispatch_serves_scalar_and_series_queries() { use super::super::protocol::Request; use super::super::read_handle::snapshot_dispatch; @@ -1869,6 +2738,12 @@ mod tests { "show_stats_all_history", Some(json!({ "window": "10s", "granularity": "1s" })), ), + // R3 Category-D cutover. + ("show_tree", None), + ("show_bloom", None), + ("show_cache", None), + ("show_routing", None), + ("show_identity_cache", None), ]; for (cmd, params) in off_loop { let resp = snapshot_dispatch(&req_params(cmd, params), &handle) @@ -1876,22 +2751,69 @@ mod tests { 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", - ] { + // Mutations take the rx_loop COMMAND path. + for cmd in ["connect", "disconnect"] { assert!( snapshot_dispatch(&req(cmd), &handle).is_none(), - "{cmd} must fall through to the rx_loop path" + "{cmd} must fall through to the rx_loop command path" ); } } + /// R5 cutover + byte-identity: after a `record_stats_history()` tick the + /// off-loop `show_acl` / `show_stats_peers` / `show_stats_history_all_peers` + /// renders each equal their on-loop oracle byte-for-byte, and all three are + /// served off-loop via `snapshot_dispatch`. + #[test] + fn snapshot_dispatch_serves_r5_queries() { + use super::super::protocol::Request; + use super::super::read_handle::snapshot_dispatch; + + let mut node = build_test_node(); + node.record_stats_history(); + let handle = node.control_read_handle(); + + // All three served off-loop (return Some, status ok). + let cases: &[(&str, Option)] = &[ + ("show_acl", None), + ("show_stats_peers", None), + ( + "show_stats_history_all_peers", + Some(json!({ "metric": "srtt_ms", "window": "10s", "granularity": "1s" })), + ), + ]; + for (cmd, params) in cases { + let req = Request { + command: cmd.to_string(), + params: params.clone(), + }; + let resp = snapshot_dispatch(&req, &handle) + .unwrap_or_else(|| panic!("{cmd} must be served off-loop")); + assert_eq!(resp.status, "ok", "{cmd} off-loop response not ok"); + } + + // Byte-identity vs the on-loop oracle. + assert_eq!( + render(show_acl(&node)), + render(show_acl_from_handle(&handle)), + "off-loop show_acl must match on-loop output" + ); + assert_eq!( + render(show_stats_peers(&node)), + render(show_stats_peers_from_handle(&handle)), + "off-loop show_stats_peers must match on-loop output" + ); + let params = json!({ "metric": "srtt_ms", "window": "10s", "granularity": "1s" }); + assert_eq!( + render_response(show_stats_history_all_peers(&node, Some(¶ms))), + render_response(show_stats_history_all_peers_from_handle( + &handle, + Some(¶ms) + )), + "off-loop show_stats_history_all_peers must match on-loop output" + ); + } + /// 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 @@ -1924,6 +2846,8 @@ mod tests { 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()); + // R5: the ACL status projection matches the node's live ACL status. + assert_eq!(snap.acl_status, node.peer_acl_status()); // Off-loop render must equal the on-loop render byte-for-byte. let on_loop = render(show_status(&node)); @@ -1933,4 +2857,239 @@ mod tests { "off-loop show_status must match on-loop output" ); } + + /// The five Category-D queries are served off-loop via `snapshot_dispatch` + /// (return `Some` with status ok); everything not cut over stays on the + /// rx_loop path (`None`). + #[test] + fn snapshot_dispatch_serves_category_d_queries() { + use super::super::protocol::Request; + use super::super::read_handle::snapshot_dispatch; + + let mut node = build_test_node(); + // Publish the routing snapshot from its tick before reading it. + node.record_stats_history(); + let handle = node.control_read_handle(); + + let req = |command: &str| Request { + command: command.to_string(), + params: None, + }; + + for cmd in OFF_LOOP_CATEGORY_D { + let resp = snapshot_dispatch(&req(cmd), &handle) + .unwrap_or_else(|| panic!("{cmd} must be served off-loop")); + assert_eq!(resp.status, "ok", "{cmd} off-loop response not ok"); + } + + // Mutations take the rx_loop COMMAND path. (Every read query, including + // the per-peer stats-series queries, is served off-loop as of R5.) + for cmd in ["connect", "disconnect"] { + assert!( + snapshot_dispatch(&req(cmd), &handle).is_none(), + "{cmd} must fall through to the rx_loop command path" + ); + } + } + + /// The tick-published `RoutingSnapshot` reflects node state, and each + /// off-loop Category-D render equals its on-loop render byte-for-byte + /// (modulo the volatile-key redaction the wire-schema tests already apply). + #[test] + fn routing_snapshot_matches_on_loop_after_tick() { + let mut node = build_test_node(); + + // Before any tick the seeded routing snapshot is empty. + let handle = node.control_read_handle(); + assert!( + handle.routing().tree.peers.is_empty(), + "seed routing snapshot has no tree peers before first tick" + ); + + // Advance one tick (the publisher site). + node.record_stats_history(); + let handle = node.control_read_handle(); + + // Snapshot reflects the node's own tree identity. + let routing = handle.routing(); + assert_eq!( + routing.tree.my_node_addr, + *node.node_addr(), + "routing snapshot tree identity reflects the node" + ); + assert_eq!( + routing.identity.max_entries, + node.identity_cache_max(), + "routing snapshot carries the identity-cache capacity" + ); + drop(routing); + + // Each off-loop render must equal the on-loop render byte-for-byte. + assert_eq!( + render(show_tree(&node)), + render(show_tree_from_handle(&handle)), + "off-loop show_tree must match on-loop output" + ); + assert_eq!( + render(show_bloom(&node)), + render(show_bloom_from_handle(&handle)), + "off-loop show_bloom must match on-loop output" + ); + assert_eq!( + render(show_cache(&node)), + render(show_cache_from_handle(&handle)), + "off-loop show_cache must match on-loop output" + ); + assert_eq!( + render(show_routing(&node)), + render(show_routing_from_handle(&handle)), + "off-loop show_routing must match on-loop output" + ); + assert_eq!( + render(show_identity_cache(&node)), + render(show_identity_cache_from_handle(&handle)), + "off-loop show_identity_cache must match on-loop output" + ); + } + + // ---- R4 Category-E coverage ------------------------------------------ + + /// The six Category-E queries are served off-loop via `snapshot_dispatch` + /// (return `Some` with status ok); mutations take the rx_loop COMMAND path. + #[test] + fn snapshot_dispatch_serves_category_e_queries() { + use super::super::protocol::Request; + use super::super::read_handle::snapshot_dispatch; + + let mut node = build_test_node(); + // Publish the entity snapshot from its tick before reading it. + node.record_stats_history(); + let handle = node.control_read_handle(); + + let req = |command: &str| Request { + command: command.to_string(), + params: None, + }; + + for cmd in OFF_LOOP_CATEGORY_E { + let resp = snapshot_dispatch(&req(cmd), &handle) + .unwrap_or_else(|| panic!("{cmd} must be served off-loop")); + assert_eq!(resp.status, "ok", "{cmd} off-loop response not ok"); + } + + // Mutations take the rx_loop COMMAND path. (Every read query is served + // off-loop as of R5.) + for cmd in ["connect", "disconnect"] { + assert!( + snapshot_dispatch(&req(cmd), &handle).is_none(), + "{cmd} must fall through to the rx_loop command path" + ); + } + } + + /// Freshness + fidelity: after a `record_stats_history()` tick (the entity + /// publisher site) each off-loop Category-E render equals its on-loop render + /// byte-for-byte, and the seeded snapshot is empty before the first tick. + #[test] + fn entity_snapshot_matches_on_loop_after_tick() { + let mut node = build_test_node(); + + // Before any tick the seeded entity snapshot is empty. + let handle = node.control_read_handle(); + assert!( + handle.entities().peers.is_empty(), + "seed entity snapshot has no peers before first tick" + ); + + // Advance one tick (the publisher site). + node.record_stats_history(); + let handle = node.control_read_handle(); + + // Each off-loop render must equal the on-loop render byte-for-byte. + assert_eq!( + render(show_peers(&node)), + render(show_peers_from_handle(&handle)), + "off-loop show_peers must match on-loop output" + ); + assert_eq!( + render(show_sessions(&node)), + render(show_sessions_from_handle(&handle)), + "off-loop show_sessions must match on-loop output" + ); + assert_eq!( + render(show_links(&node)), + render(show_links_from_handle(&handle)), + "off-loop show_links must match on-loop output" + ); + assert_eq!( + render(show_connections(&node)), + render(show_connections_from_handle(&handle)), + "off-loop show_connections must match on-loop output" + ); + assert_eq!( + render(show_transports(&node)), + render(show_transports_from_handle(&handle)), + "off-loop show_transports must match on-loop output" + ); + assert_eq!( + render(show_mmp(&node)), + render(show_mmp_from_handle(&handle)), + "off-loop show_mmp must match on-loop output" + ); + } + + /// Structural sharing (the R4 umbrella mandate): a republish in which only + /// one row changed re-allocates only that one `Arc` — every unchanged + /// row is reused by pointer (`Arc::ptr_eq`). Exercises + /// [`reconcile_rows`](super::super::snapshot::reconcile_rows), the + /// `Vec>` reconciliation the per-tick publisher uses for every + /// entity table. + #[test] + fn entity_snapshot_structural_sharing() { + use super::super::snapshot::{LinkRow, LinkStats, reconcile_rows}; + + let mk = |link_id: u64, bytes_recv: u64| LinkRow { + link_id, + transport_id: 1, + remote_addr: "0.0.0.0:0".to_string(), + direction: "outbound".to_string(), + state: "up".to_string(), + created_at_ms: 1000, + stats: LinkStats { + packets_sent: 0, + packets_recv: 0, + bytes_sent: 0, + bytes_recv, + last_recv_ms: 0, + }, + }; + + // First publish: two distinct rows, both freshly allocated. + let prev = reconcile_rows::(&[], vec![mk(1, 10), mk(2, 20)], |r| r.link_id); + assert_eq!(prev.len(), 2); + + // Second publish: row 1 unchanged, row 2 changed (bytes_recv differs). + let next = reconcile_rows(&prev, vec![mk(1, 10), mk(2, 999)], |r| r.link_id); + assert_eq!(next.len(), 2); + + // The unchanged row is the SAME Arc (reused by pointer). + assert!( + std::sync::Arc::ptr_eq(&prev[0], &next[0]), + "unchanged row must be reused by pointer (structural sharing)" + ); + // The changed row is a fresh allocation. + assert!( + !std::sync::Arc::ptr_eq(&prev[1], &next[1]), + "changed row must be re-allocated" + ); + assert_eq!(next[1].stats.bytes_recv, 999); + + // A republish with no changes reuses every row. + let again = reconcile_rows(&next, vec![mk(1, 10), mk(2, 999)], |r| r.link_id); + assert!( + std::sync::Arc::ptr_eq(&next[0], &again[0]) + && std::sync::Arc::ptr_eq(&next[1], &again[1]), + "an unchanged republish must reuse every row Arc" + ); + } } diff --git a/src/control/read_handle.rs b/src/control/read_handle.rs index 38167f6..22660f1 100644 --- a/src/control/read_handle.rs +++ b/src/control/read_handle.rs @@ -32,7 +32,7 @@ use crate::node::context::NodeContext; use crate::node::metrics::MetricsRegistry; use super::protocol::{Request, Response}; -use super::snapshot::StatsSnapshot; +use super::snapshot::{EntitySnapshot, RoutingSnapshot, StatsSnapshot}; /// Cloneable read-only view of node state for off-loop control serving. /// @@ -49,6 +49,13 @@ pub(crate) struct ControlReadHandle { /// stats_history dual-ring read copy + the scalar gauges/counts /// `show_status` needs, published from the tick (R2, Q1-b). stats: Arc>, + /// Category-D derived/routing/cache read view (tree / bloom / coord / + /// identity + F-queue scalars), published from the tick (R3). + routing: Arc>, + /// Category-E per-entity table read view (peers / sessions / links / + /// connections / transports + mmp), published from the tick with + /// `Vec>` structural sharing (R4). + entities: Arc>, } impl ControlReadHandle { @@ -60,11 +67,15 @@ impl ControlReadHandle { context: Arc, metrics: Arc, stats: Arc>, + routing: Arc>, + entities: Arc>, ) -> Self { Self { context, metrics, stats, + routing, + entities, } } @@ -83,6 +94,18 @@ impl ControlReadHandle { pub(crate) fn stats(&self) -> arc_swap::Guard> { self.stats.load() } + + /// Load the latest published Category-D routing snapshot (freshest + /// available by construction; no staleness gate, per Q1-e). + pub(crate) fn routing(&self) -> arc_swap::Guard> { + self.routing.load() + } + + /// Load the latest published Category-E entity snapshot (freshest available + /// by construction; no staleness gate, per Q1-e). + pub(crate) fn entities(&self) -> arc_swap::Guard> { + self.entities.load() + } } /// Attempt to serve a request entirely from the read handle, off the rx_loop. @@ -104,15 +127,22 @@ pub(crate) fn snapshot_dispatch(request: &Request, handle: &ControlReadHandle) - )), "show_stats_list" => Some(Response::ok(queries::show_stats_list())), "show_metrics" => Some(Response::ok(queries::show_metrics_from_handle(handle))), + // R5: peer-ACL status, served from the tick-published `StatsSnapshot`. + // The ACL is an `arc_swap::ArcSwap` reloaded only on the tick; + // its status projection is captured at the same tick. + "show_acl" => Some(Response::ok(queries::show_acl_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. + // satisfiable). + // + // R5 closes out the per-peer stats queries: `show_stats_peers` and + // `show_stats_history_all_peers` now read the snapshot's per-peer + // `peer_meta` (live `is_active`, resolved npub / display name, captured + // at publish time) joined against the `history` rings, so they no longer + // need live `&Node` and render off-loop too. "show_status" => Some(Response::ok(queries::show_status_from_handle(handle))), "show_stats_history" => Some(queries::show_stats_history_from_handle( handle, @@ -122,6 +152,34 @@ pub(crate) fn snapshot_dispatch(request: &Request, handle: &ControlReadHandle) - handle, request.params.as_ref(), )), + "show_stats_peers" => Some(Response::ok(queries::show_stats_peers_from_handle(handle))), + "show_stats_history_all_peers" => Some(queries::show_stats_history_all_peers_from_handle( + handle, + request.params.as_ref(), + )), + // R3: served from the tick-published `RoutingSnapshot` (tree / bloom / + // coord cache / identity cache + F-queue scalars). Display names are + // resolved at publish time, so these render entirely off-loop. The + // counter-family `stats` blocks come from the `MetricsRegistry` (also + // in the handle). All five are parameterless. + "show_tree" => Some(Response::ok(queries::show_tree_from_handle(handle))), + "show_bloom" => Some(Response::ok(queries::show_bloom_from_handle(handle))), + "show_cache" => Some(Response::ok(queries::show_cache_from_handle(handle))), + "show_routing" => Some(Response::ok(queries::show_routing_from_handle(handle))), + "show_identity_cache" => Some(Response::ok(queries::show_identity_cache_from_handle( + handle, + ))), + // R4: served from the tick-published `EntitySnapshot` (per-entity + // `Vec>` tables with structural sharing). Display names, + // tree-relationship flags, and Nostr-traversal state are resolved at + // publish time, so these render entirely off-loop. All six are + // parameterless. + "show_peers" => Some(Response::ok(queries::show_peers_from_handle(handle))), + "show_sessions" => Some(Response::ok(queries::show_sessions_from_handle(handle))), + "show_links" => Some(Response::ok(queries::show_links_from_handle(handle))), + "show_connections" => Some(Response::ok(queries::show_connections_from_handle(handle))), + "show_transports" => Some(Response::ok(queries::show_transports_from_handle(handle))), + "show_mmp" => Some(Response::ok(queries::show_mmp_from_handle(handle))), _ => None, } } diff --git a/src/control/snapshot.rs b/src/control/snapshot.rs index 9d22890..c6aeb9f 100644 --- a/src/control/snapshot.rs +++ b/src/control/snapshot.rs @@ -23,6 +23,7 @@ use std::sync::Arc; use crate::identity::NodeAddr; use crate::node::NodeState; +use crate::node::acl::PeerAclStatus; use crate::node::stats_history::StatsHistory; use crate::upper::tun::TunState; @@ -55,6 +56,32 @@ pub(crate) struct StatsSnapshot { /// Configured peer aliases, keyed by `NodeAddr`. Effectively immutable /// after construction; shared to avoid a per-tick map clone. pub peer_aliases: Arc>, + /// Loaded peer-ACL status (`show_acl`). The ACL itself is an + /// `arc_swap::ArcSwap` mutated only by the tick's `reload_peer_acl`; + /// the human-readable status is a cheap projection of it (R5). + pub acl_status: PeerAclStatus, + /// Per-stats-history-peer metadata resolved against the live peer/session + /// tables and host map at publish time (`show_stats_peers` / + /// `show_stats_history_all_peers`), keyed by `NodeAddr`. The lifecycle + /// timestamps and per-peer metric rings stay in `history`; this map carries + /// only the cross-subsystem fields a renderer can't derive from the rings + /// alone (`is_active`, resolved `npub`, resolved `display_name`) (R5). + pub peer_meta: Arc>, +} + +/// Cross-subsystem metadata for one peer tracked in the stats-history rings, +/// resolved at publish time. Joined against `StatsSnapshot::history`'s rings +/// (lifecycle timestamps + metric series) by the off-loop `show_stats_peers` / +/// `show_stats_history_all_peers` renderers. +#[derive(Clone)] +pub(crate) struct StatsPeerMeta { + /// Whether this peer is currently in the live authenticated-peer table. + pub is_active: bool, + /// Resolved npub (live peer npub, or `node_addr` hex when not a live peer), + /// matching the on-loop `show_stats_peers` fallback. + pub npub: String, + /// Display name resolved via `Node::peer_display_name` at publish time. + pub display_name: String, } impl StatsSnapshot { @@ -74,6 +101,615 @@ impl StatsSnapshot { transport_count: 0, session_count: 0, peer_aliases: Arc::new(HashMap::new()), + acl_status: empty_acl_status(), + peer_meta: Arc::new(HashMap::new()), } } } + +/// An empty/default [`PeerAclStatus`] for seeding the snapshot before the first +/// tick publishes the real ACL status. `PeerAclStatus` does not derive +/// `Default`, so this spells out the inert "no ACL loaded" shape. +fn empty_acl_status() -> PeerAclStatus { + PeerAclStatus { + allow_file: String::new(), + deny_file: String::new(), + enforcement_active: false, + effective_mode: String::new(), + default_decision: String::new(), + allow_all: false, + deny_all: false, + allow_file_entries: Vec::new(), + deny_file_entries: Vec::new(), + allow_entries: Vec::new(), + deny_entries: Vec::new(), + } +} + +// ===================================================================== +// RoutingSnapshot (R3 — Category-D derived/routing/cache read view) +// ===================================================================== + +/// Read-only snapshot of the Category-D derived/routing/cache subsystems that +/// the pure-snapshot `show_tree` / `show_bloom` / `show_cache` / `show_routing` +/// / `show_identity_cache` queries render. Published via `ArcSwap`. +/// +/// The R0 stub (`design/fast-path-refactoring-r0-read-handle.md`) names a +/// single combined `ArcSwap` for R3. This is that cell: one +/// cohesive routing view holding the four subsystems (tree / bloom / coord +/// cache / identity cache) plus the F-queue summary scalars. +/// +/// **Publisher placement (Q1).** The four subsystems mutate at many scattered +/// handler sites (28 `coord_cache_mut` call sites, 16 `tree_state_mut`, ~32 +/// identity-cache touches), and every projected row needs a *display name* +/// resolved against the live peer/session tables and host map — Category-E +/// state reachable only with `&Node`. Wiring an on-change `publish_*` at each +/// mutation site would be large, error-prone surgery, and each call would still +/// need `&Node` to resolve names across subsystem boundaries. So this snapshot +/// is published from the **tick** (Q1-b acceptable-at-mutator / the documented +/// interim the spec permits, mirroring R2's stats publish): the tick is the one +/// site with coherent `&Node` access to resolve all display names together. A +/// single combined cell is the natural shape because there is exactly one +/// publisher — the multi-mutator "rebuild the whole snapshot N times" hazard +/// that Q1-c warns against does not arise. +/// +/// The snapshot holds *data* (typed rows + scalars), not rendered `Response` +/// envelopes (Q1-d); rendering happens off the rx_loop in the control task. The +/// counter-family `stats` blocks the queries also emit come from the +/// `MetricsRegistry` (already `Arc`-shared in the handle) at render time, not +/// from this snapshot. +/// +/// Time-relative fields (`age_ms`, `idle_ms`) are derived at render time from +/// the captured absolute timestamps, so the rendered age stays fresh relative +/// to the read, exactly as the on-loop queries computed it. +/// +/// Forward-compat: when step 5 structurally extracts the Category-D subsystems +/// into typed types, these projections become thin views over them without +/// changing the read-handle interface or this publisher placement. +#[derive(Clone)] +pub(crate) struct RoutingSnapshot { + /// Spanning-tree read view (`show_tree`). + pub tree: TreeView, + /// Bloom-filter read view (`show_bloom`). + pub bloom: BloomView, + /// Coordinate-cache read view (`show_cache`, `show_routing`). + pub cache: CacheView, + /// F-queue / discovery routing scalars + rows (`show_routing`). + pub routing: RoutingView, + /// Identity-cache read view (`show_identity_cache`, `show_routing`). + pub identity: IdentityView, +} + +impl RoutingSnapshot { + /// 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 { + tree: TreeView::default(), + bloom: BloomView::default(), + cache: CacheView::default(), + routing: RoutingView::default(), + identity: IdentityView::default(), + } + } +} + +/// Zero `NodeAddr` for empty/seed views (all-zero 16 bytes). +fn zero_addr() -> NodeAddr { + NodeAddr::from_bytes([0u8; 16]) +} + +/// Spanning-tree read view for `show_tree`. +#[derive(Clone)] +pub(crate) struct TreeView { + pub my_node_addr: NodeAddr, + pub root: NodeAddr, + pub is_root: bool, + pub depth: usize, + /// `my_coords` entries as `NodeAddr`s (rendered as hex). + pub my_coords: Vec, + pub parent: NodeAddr, + pub parent_display_name: String, + pub declaration_sequence: u64, + pub declaration_signed: bool, + pub peer_tree_count: usize, + pub peers: Vec, +} + +impl Default for TreeView { + fn default() -> Self { + Self { + my_node_addr: zero_addr(), + root: zero_addr(), + is_root: false, + depth: 0, + my_coords: Vec::new(), + parent: zero_addr(), + parent_display_name: String::new(), + declaration_sequence: 0, + declaration_signed: false, + peer_tree_count: 0, + peers: Vec::new(), + } + } +} + +/// One peer's tree position in `show_tree`. +#[derive(Clone)] +pub(crate) struct TreePeerRow { + pub node_addr: NodeAddr, + pub display_name: String, + /// Present only when the peer's coordinates are known. + pub coords: Option, +} + +/// Coordinate detail for a tree peer (present only when known). +#[derive(Clone)] +pub(crate) struct TreePeerCoords { + pub depth: usize, + pub root: NodeAddr, + pub coord_path: Vec, + pub distance_to_us: usize, +} + +/// Bloom-filter read view for `show_bloom`. +#[derive(Clone)] +pub(crate) struct BloomView { + pub own_node_addr: NodeAddr, + pub is_leaf_only: bool, + pub sequence: u64, + pub leaf_dependents: Vec, + pub peer_filters: Vec, +} + +impl Default for BloomView { + fn default() -> Self { + Self { + own_node_addr: zero_addr(), + is_leaf_only: false, + sequence: 0, + leaf_dependents: Vec::new(), + peer_filters: Vec::new(), + } + } +} + +/// One peer's bloom-filter state in `show_bloom`. +#[derive(Clone)] +pub(crate) struct BloomPeerRow { + pub peer: NodeAddr, + pub display_name: String, + pub has_filter: bool, + pub filter_sequence: u64, + /// Present only when the peer has supplied an inbound filter. + pub filter: Option, +} + +/// Inbound-filter statistics for a bloom peer (present only when known). +#[derive(Clone)] +pub(crate) struct BloomPeerFilter { + /// Estimated cardinality (`None` when undefined for the saturation), + /// matching `BloomFilter::estimated_count`'s `Option`. + pub estimated_count: Option, + pub set_bits: usize, + pub fill_ratio: f64, +} + +/// Coordinate-cache read view for `show_cache` (and the cache scalars in +/// `show_routing`). +#[derive(Clone, Default)] +pub(crate) struct CacheView { + pub count: usize, + pub max_entries: usize, + pub fill_ratio: f64, + pub default_ttl_ms: u64, + pub expired: usize, + pub avg_age_ms: u64, + pub entries: Vec, +} + +/// One coordinate-cache entry in `show_cache`. +#[derive(Clone)] +pub(crate) struct CacheEntryRow { + pub node_addr: NodeAddr, + pub display_name: String, + pub depth: usize, + pub coord_path: Vec, + /// Absolute creation time (Unix ms); `age_ms` derived at render time. + pub created_at: u64, + pub last_used_ms: u64, + pub path_mtu: Option, +} + +/// F-queue / discovery routing read view for `show_routing`. +#[derive(Clone, Default)] +pub(crate) struct RoutingView { + pub pending_lookups: Vec, + pub pending_tun_destinations: usize, + pub pending_tun_packets: usize, + pub recent_requests: usize, + pub retries: Vec, +} + +/// One in-flight discovery lookup in `show_routing`. +#[derive(Clone)] +pub(crate) struct PendingLookupRow { + pub target: NodeAddr, + pub display_name: String, + /// Absolute initiation time (Unix ms); `age_ms` derived at render time. + pub initiated_ms: u64, + pub last_sent_ms: u64, + pub attempt: u8, +} + +/// One connection-retry entry in `show_routing`. +#[derive(Clone)] +pub(crate) struct RetryRow { + pub node_addr: NodeAddr, + pub display_name: String, + pub retry_count: u32, + pub retry_after_ms: u64, + pub auto_reconnect: bool, +} + +/// Identity-cache read view for `show_identity_cache` (and the +/// `identity_cache_entries` scalar in `show_routing`). +#[derive(Clone, Default)] +pub(crate) struct IdentityView { + pub entries: Vec, + pub max_entries: usize, +} + +/// One identity-cache entry in `show_identity_cache`. +#[derive(Clone)] +pub(crate) struct IdentityRow { + pub node_addr: NodeAddr, + pub npub: String, + pub display_name: String, + pub ipv6_addr: String, + pub last_seen_ms: u64, +} + +// ===================================================================== +// EntitySnapshot (R4 — Category-E per-entity table read views) +// ===================================================================== + +/// Read-only snapshot of the Category-E per-entity tables that the +/// pure-snapshot `show_peers` / `show_sessions` / `show_links` / +/// `show_connections` / `show_transports` / `show_mmp` queries render. +/// Published via `ArcSwap`. +/// +/// The R0 stub (`design/fast-path-refactoring-r0-read-handle.md`) pre-scopes +/// R4 as `entities — ArcSwap: peers / sessions / links / +/// connections / transports, published per-entity with `Vec>` +/// structural sharing`. This is that cell. +/// +/// **Structural sharing (the umbrella mandate).** Every entity table is a +/// `Vec>`, so a republish in which only one row changed re-allocates +/// only that one `Arc` — the unchanged rows are reused by pointer from the +/// previous snapshot (`Arc::ptr_eq`-stable). The publisher diffs each freshly +/// projected row against the prior published row by value (`PartialEq`) and +/// keeps the old `Arc` when they are equal. A clone of the snapshot for each +/// accepted control connection is then a vector of cheap pointer clones, not a +/// deep table copy. This is what keeps the per-tick publish cost off the hot +/// path at scale, as the umbrella requires for R4. +/// +/// **Publisher placement (Q1).** Like R3, this is published from the **tick**, +/// not per-mutator. Two reasons, both stronger than for R3: +/// +/// 1. Every projected row needs a *display name* resolved against the live +/// peer/session tables and host map (`&Node`), and `show_peers` additionally +/// needs the live tree state to derive `is_parent` / `is_child` and the +/// Nostr-discovery failure-state map — cross-subsystem reads available only +/// with `&Node`. +/// 2. Most of the projected fields (link/session traffic counters, MMP +/// metrics, `last_seen`, noise counters, replay/decrypt counters) are +/// mutated continuously on the **data plane / rx_loop**, not at the discrete +/// peer/session/link lifecycle mutators. Per-lifecycle-mutator publication +/// (Q1-a) would therefore not even capture freshness for those fields; the +/// tick is the natural cadence at which this read view advances. +/// +/// The diff-and-reuse therefore satisfies the structural-sharing goal the +/// umbrella mandates (only changed rows re-allocate) while keeping a single +/// coherent `&Node` publisher — the "no monolithic per-tick *re-allocation* of +/// every row" warning is honored because unchanged rows are reused, not rebuilt. +/// This is the documented acceptable interim (the spec's tick-publish-with- +/// Arc-reuse fallback), consistent with R3. +/// +/// The snapshot holds typed rows (Q1-d data, not rendered `Response` +/// envelopes). Time-relative fields (`idle_ms`) are derived at render time from +/// captured absolute timestamps, so the rendered age stays fresh relative to +/// the read, exactly as the on-loop queries computed it. +/// +/// Forward-compat: step 10 later extracts the session table into a typed +/// `(transport_id, our_index)`-indexed type; these projections then become thin +/// views over it without changing the read-handle interface or this publisher +/// placement. +#[derive(Clone)] +pub(crate) struct EntitySnapshot { + /// `show_peers` rows. + pub peers: Vec>, + /// `show_sessions` rows. + pub sessions: Vec>, + /// `show_links` rows. + pub links: Vec>, + /// `show_connections` rows. + pub connections: Vec>, + /// `show_transports` rows. + pub transports: Vec>, + /// `show_mmp` link-layer rows (peers with an MMP instance). + pub mmp_peers: Vec>, + /// `show_mmp` session-layer rows (sessions with an MMP instance). + pub mmp_sessions: Vec>, +} + +impl EntitySnapshot { + /// 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 { + peers: Vec::new(), + sessions: Vec::new(), + links: Vec::new(), + connections: Vec::new(), + transports: Vec::new(), + mmp_peers: Vec::new(), + mmp_sessions: Vec::new(), + } + } +} + +/// Per-peer link/transport/connectivity fields for `show_peers` derived from a +/// peer's resolved link (present only when the link is found). +#[derive(Clone, PartialEq)] +pub(crate) struct PeerLinkInfo { + pub direction: String, + /// Transport type name, present only when the transport handle is found. + pub transport_type: Option, +} + +/// Nostr-traversal failure-state for a peer's npub in `show_peers`. Always +/// emitted (the on-loop query emits a default object even when absent); the +/// `present` flag distinguishes "seen by Nostr discovery" from the default. +#[derive(Clone, PartialEq)] +pub(crate) struct PeerNostrState { + pub consecutive_failures: u32, + pub cooldown_until_ms: Option, + pub last_observed_skew_ms: Option, +} + +/// Noise session counters surfaced in `show_peers` (present when the peer has a +/// Noise session). +#[derive(Clone, PartialEq)] +pub(crate) struct PeerNoiseCounters { + pub send_counter: u64, + pub highest_recv_counter: u64, +} + +/// Link/session MMP metrics surfaced inline in `show_peers` (and the +/// per-session block in `show_sessions`). Fields mirror the on-loop projection; +/// `Option` fields are emitted only when present. +#[derive(Clone, PartialEq)] +pub(crate) struct EntityMmp { + pub mode: String, + pub srtt_ms: Option, + pub loss_rate: f64, + pub etx: f64, + pub goodput_bps: f64, + pub delivery_ratio_forward: f64, + pub delivery_ratio_reverse: f64, + pub smoothed_loss: Option, + pub smoothed_etx: Option, + /// `lqi` (peers) / `sqi` (sessions): present only when both `srtt_ms` and + /// `smoothed_etx` are present. Precomputed so the render is a plain emit. + pub quality_index: Option, + /// Session-only: path MTU (`show_sessions`). `None` for peer rows. + pub path_mtu: Option, +} + +/// Link-layer stat counters for a peer in `show_peers`. +#[derive(Clone, PartialEq)] +pub(crate) struct PeerLinkStats { + pub packets_sent: u64, + pub packets_recv: u64, + pub bytes_sent: u64, + pub bytes_recv: u64, +} + +/// One authenticated peer in `show_peers`. Holds every field the on-loop +/// `show_peers` emits; `Option` fields gate the conditionally-emitted keys. +#[derive(Clone, PartialEq)] +pub(crate) struct PeerRow { + pub node_addr: NodeAddr, + pub npub: String, + pub display_name: String, + pub ipv6_addr: String, + pub connectivity: String, + pub link_id: u64, + pub authenticated_at_ms: u64, + pub last_seen_ms: u64, + pub has_tree_position: bool, + pub has_bloom_filter: bool, + pub filter_sequence: u64, + pub is_parent: bool, + pub is_child: bool, + pub transport_addr: Option, + pub link_info: Option, + pub tree_depth: Option, + pub stats: PeerLinkStats, + pub replay_suppressed: u32, + pub consecutive_decrypt_failures: u32, + pub nostr_traversal: PeerNostrState, + pub noise: Option, + pub our_session_index: Option, + pub rekey_in_progress: bool, + pub rekey_draining: bool, + pub current_k_bit: bool, + pub mmp: Option, +} + +/// Traffic counters for a session in `show_sessions`. +#[derive(Clone, PartialEq)] +pub(crate) struct SessionStats { + pub packets_sent: u64, + pub packets_recv: u64, + pub bytes_sent: u64, + pub bytes_recv: u64, +} + +/// One end-to-end session in `show_sessions`. +#[derive(Clone, PartialEq)] +pub(crate) struct SessionRow { + pub remote_addr: NodeAddr, + pub display_name: String, + pub state: &'static str, + pub is_initiator: bool, + pub last_activity_ms: u64, + pub npub: String, + pub stats: SessionStats, + /// Handshake resend count, emitted only while not established. + pub resend_count: Option, + /// Established-only health block (session_start_ms, current_k_bit, + /// coords_warmup_remaining, is_draining). `None` while handshaking. + pub established: Option, + pub mmp: Option, +} + +/// Established-session health fields in `show_sessions` (emitted only when the +/// session is established). +#[derive(Clone, PartialEq)] +pub(crate) struct SessionEstablished { + pub session_start_ms: u64, + pub current_k_bit: bool, + pub coords_warmup_remaining: u8, + pub is_draining: bool, +} + +/// Stat counters for a link in `show_links`. +#[derive(Clone, PartialEq)] +pub(crate) struct LinkStats { + pub packets_sent: u64, + pub packets_recv: u64, + pub bytes_sent: u64, + pub bytes_recv: u64, + pub last_recv_ms: u64, +} + +/// One active link in `show_links`. +#[derive(Clone, PartialEq)] +pub(crate) struct LinkRow { + pub link_id: u64, + pub transport_id: u32, + pub remote_addr: String, + pub direction: String, + pub state: String, + pub created_at_ms: u64, + pub stats: LinkStats, +} + +/// One pending handshake in `show_connections`. `idle_ms` is derived at render +/// time from the captured `last_activity_ms`. +#[derive(Clone, PartialEq)] +pub(crate) struct ConnectionRow { + pub link_id: u64, + pub direction: String, + pub handshake_state: String, + pub started_at_ms: u64, + /// Absolute last-activity time (Unix ms); `idle_ms` derived at render time. + pub last_activity_ms: u64, + pub resend_count: u32, + /// Expected peer npub, emitted only when the connection has an expected + /// identity. + pub expected_peer: Option, +} + +/// One transport instance in `show_transports`. The `stats` and +/// `tor_monitoring` fields are stored as already-projected `serde_json::Value` +/// (data, produced by the transport handle), not as rendered `Response` +/// envelopes. +#[derive(Clone, PartialEq)] +pub(crate) struct TransportRow { + pub transport_id: u32, + pub transport_type: String, + pub state: String, + pub mtu: u16, + pub name: Option, + pub local_addr: Option, + pub tor_mode: Option, + pub onion_address: Option, + pub tor_monitoring: Option, + pub stats: serde_json::Value, +} + +/// MMP trend labels for a peer's link-layer block in `show_mmp` (each present +/// only when the corresponding trend is initialized). +#[derive(Clone, PartialEq)] +pub(crate) struct MmpTrends { + pub rtt_trend: Option<&'static str>, + pub loss_trend: Option<&'static str>, + pub goodput_trend: Option<&'static str>, + pub jitter_trend: Option<&'static str>, +} + +/// One peer's link-layer MMP block in `show_mmp`. +#[derive(Clone, PartialEq)] +pub(crate) struct MmpPeerRow { + pub peer: NodeAddr, + pub display_name: String, + pub mode: String, + pub loss_rate: f64, + pub etx: f64, + pub goodput_bps: f64, + pub spin_bit_initiator: bool, + pub smoothed_loss: Option, + pub smoothed_etx: Option, + pub srtt_ms: Option, + /// `lqi`: present only when both `srtt_ms` and `smoothed_etx` are present. + pub lqi: Option, + pub trends: MmpTrends, + pub delivery_ratio_forward: f64, + pub delivery_ratio_reverse: f64, + pub ecn_ce_count: u32, +} + +/// One session's session-layer MMP block in `show_mmp`. +#[derive(Clone, PartialEq)] +pub(crate) struct MmpSessionRow { + pub remote: NodeAddr, + pub display_name: String, + pub mode: String, + pub loss_rate: f64, + pub etx: f64, + pub path_mtu: u16, + pub smoothed_loss: Option, + pub smoothed_etx: Option, + pub srtt_ms: Option, + /// `sqi`: present only when both `srtt_ms` and `smoothed_etx` are present. + pub sqi: Option, +} + +/// Reconcile a freshly-projected entity table against the previously published +/// one, preserving structural sharing: an `Arc` from `prev` is reused +/// (kept by pointer) whenever a new row matches an old row by identity `key` +/// **and** compares equal by value, so only changed/new rows allocate a fresh +/// `Arc`. This is the `Vec>` discipline the R4 umbrella mandates — a +/// single-row change re-allocates one row, not the whole table, keeping the +/// per-tick publish cost off the hot path at scale. +/// +/// `key` extracts a stable, hashable identity (e.g. `node_addr`, `link_id`) so +/// matching is order-independent across the source table's iteration order. +pub(crate) fn reconcile_rows(prev: &[Arc], new_rows: Vec, key: F) -> Vec> +where + R: PartialEq, + K: std::hash::Hash + Eq, + F: Fn(&R) -> K, +{ + let index: HashMap> = prev.iter().map(|arc| (key(arc), arc)).collect(); + new_rows + .into_iter() + .map(|row| match index.get(&key(&row)) { + Some(old) if ***old == row => Arc::clone(old), + _ => Arc::new(row), + }) + .collect() +} diff --git a/src/node/handlers/rx_loop.rs b/src/node/handlers/rx_loop.rs index e46e02e..525550f 100644 --- a/src/node/handlers/rx_loop.rs +++ b/src/node/handlers/rx_loop.rs @@ -1,6 +1,5 @@ //! RX event loop and packet dispatch. -use crate::control::queries; use crate::control::{ControlSocket, commands}; use crate::node::wire::{ COMMON_PREFIX_SIZE, CommonPrefix, FMP_VERSION, PHASE_ESTABLISHED, PHASE_MSG1, PHASE_MSG2, @@ -236,15 +235,18 @@ impl Node { self.register_identity(identity.node_addr, identity.pubkey); } Some((request, response_tx)) = control_rx.recv() => { - let response = if request.command.starts_with("show_") { - queries::dispatch(self, &request.command, request.params.as_ref()) - } else { - commands::dispatch( - self, - &request.command, - request.params.as_ref(), - ).await - }; + // Only mutating COMMAND requests (`connect` / `disconnect`) + // reach the rx_loop now. Every pure-read `show_*` query is + // served off-loop from the read handle in the control accept + // task (`snapshot_dispatch`), so it never round-trips here — + // the data-plane dispatch path carries no `show_*` arm. A + // `show_*` that somehow arrives (none does) falls through to + // `commands::dispatch`, which returns "unknown command". + let response = commands::dispatch( + self, + &request.command, + request.params.as_ref(), + ).await; let _ = response_tx.send(response); } _ = tick.tick() => { diff --git a/src/node/mod.rs b/src/node/mod.rs index 42adfcf..c24a60c 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -4,7 +4,7 @@ //! holds all state required for mesh routing: identity, tree state, //! Bloom filters, coordinate caches, transports, links, and peers. -mod acl; +pub(crate) mod acl; mod bloom; pub(crate) mod context; #[cfg(unix)] @@ -396,6 +396,21 @@ pub struct Node { /// live mutable `stats_history` above stays on the tick. stats_snapshot: std::sync::Arc>, + /// Read-side snapshot of the Category-D derived/routing/cache subsystems + /// (tree / bloom / coord cache / identity cache + F-queue scalars) that the + /// `show_tree` / `show_bloom` / `show_cache` / `show_routing` / + /// `show_identity_cache` queries render off the rx_loop. Published from the + /// tick (see [`Self::publish_routing_snapshot`] for the Q1 rationale). + routing_snapshot: std::sync::Arc>, + + /// Read-side snapshot of the Category-E per-entity tables (peers / sessions + /// / links / connections / transports + mmp) that the `show_peers` / + /// `show_sessions` / `show_links` / `show_connections` / `show_transports` + /// / `show_mmp` queries render off the rx_loop. Published from the tick with + /// `Vec>` structural sharing (unchanged rows reused by pointer); + /// see [`Self::publish_entities_snapshot`] for the Q1 rationale. + entities_snapshot: std::sync::Arc>, + // === TUN Interface === /// TUN device state. tun_state: TunState, @@ -663,6 +678,12 @@ impl Node { stats_snapshot: std::sync::Arc::new(arc_swap::ArcSwap::from_pointee( crate::control::snapshot::StatsSnapshot::empty(), )), + routing_snapshot: std::sync::Arc::new(arc_swap::ArcSwap::from_pointee( + crate::control::snapshot::RoutingSnapshot::empty(), + )), + entities_snapshot: std::sync::Arc::new(arc_swap::ArcSwap::from_pointee( + crate::control::snapshot::EntitySnapshot::empty(), + )), tun_state, tun_name: None, tun_tx: None, @@ -818,6 +839,12 @@ impl Node { stats_snapshot: std::sync::Arc::new(arc_swap::ArcSwap::from_pointee( crate::control::snapshot::StatsSnapshot::empty(), )), + routing_snapshot: std::sync::Arc::new(arc_swap::ArcSwap::from_pointee( + crate::control::snapshot::RoutingSnapshot::empty(), + )), + entities_snapshot: std::sync::Arc::new(arc_swap::ArcSwap::from_pointee( + crate::control::snapshot::EntitySnapshot::empty(), + )), tun_state, tun_name: None, tun_tx: None, @@ -1395,6 +1422,8 @@ impl Node { self.context.clone(), self.metrics.clone(), self.stats_snapshot.clone(), + self.routing_snapshot.clone(), + self.entities_snapshot.clone(), ) } @@ -1468,6 +1497,30 @@ impl Node { // 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. + // Per-stats-history-peer metadata (R5). `show_stats_peers` / + // `show_stats_history_all_peers` need each tracked peer's live + // membership (`is_active`), resolved npub, and display name — all + // cross-subsystem reads against the live peer table and host map, + // available only here with `&self`. The lifecycle timestamps and + // metric rings the renderers also read live in `history` (the dual-ring + // read copy above), so this map carries only the resolved fields. + let peer_meta: HashMap = self + .stats_history + .peer_addrs() + .copied() + .map(|addr| { + let live = self.peers.get(&addr); + let meta = crate::control::snapshot::StatsPeerMeta { + is_active: live.is_some(), + npub: live + .map(|p| p.npub()) + .unwrap_or_else(|| hex::encode(addr.as_bytes())), + display_name: self.peer_display_name(&addr), + }; + (addr, meta) + }) + .collect(); + let snapshot = crate::control::snapshot::StatsSnapshot { history: std::sync::Arc::new(self.stats_history.clone()), estimated_mesh_size: self.estimated_mesh_size, @@ -1481,8 +1534,512 @@ impl Node { transport_count: self.transports.len(), session_count: self.sessions.len(), peer_aliases: std::sync::Arc::new(self.peer_aliases.clone()), + acl_status: self.peer_acl_status(), + peer_meta: std::sync::Arc::new(peer_meta), }; self.stats_snapshot.store(std::sync::Arc::new(snapshot)); + + // Publish the Category-D routing read view alongside the stats + // snapshot, from the same tick. + self.publish_routing_snapshot(); + + // Publish the Category-E per-entity read view from the same tick, with + // `Vec>` structural sharing against the previous snapshot. + self.publish_entities_snapshot(); + } + + /// Project the Category-D derived/routing/cache state into a + /// [`RoutingSnapshot`](crate::control::snapshot::RoutingSnapshot) and + /// publish it via `ArcSwap`, so `show_tree` / `show_bloom` / `show_cache` + /// / `show_routing` / `show_identity_cache` render off the rx_loop. + /// + /// **Q1 publisher placement.** The four projected subsystems (tree / bloom + /// / coord cache / identity cache) mutate at dozens of scattered handler + /// sites, and every projected row carries a *display name* resolved against + /// the live peer/session tables and host map — state reachable only with + /// `&Node`. Per-mutator on-change publication (Q1-a) would therefore be + /// large, error-prone surgery, and each call would still need `&Node` to + /// resolve names across subsystem boundaries. So this projection is + /// published from the tick — the documented acceptable interim (the spec's + /// "publish from the tick" allowance, mirroring R2's stats publish). The + /// tick is the one site with coherent `&Node` access to resolve every + /// display name together. A single combined cell is the natural shape + /// because there is exactly one publisher, so the multi-mutator + /// whole-snapshot-rebuild hazard Q1-c warns against does not arise. + /// + /// The snapshot holds typed rows + scalars (Q1-d data, not rendered + /// responses); the counter-family `stats` blocks the queries also emit are + /// served from the `MetricsRegistry` (already `Arc`-shared) at render time. + fn publish_routing_snapshot(&self) { + use crate::control::snapshot as snap; + + let now = Self::now_ms(); + + // --- tree (show_tree) --- + let tree = self.tree_state(); + let my_coords = tree.my_coords(); + let tree_peers: Vec = tree + .peer_ids() + .map(|peer_id| { + let coords = tree + .peer_coords(peer_id) + .map(|coords| snap::TreePeerCoords { + depth: coords.depth(), + root: *coords.root_id(), + coord_path: coords.entries().iter().map(|e| e.node_addr).collect(), + distance_to_us: my_coords.distance_to(coords), + }); + snap::TreePeerRow { + node_addr: *peer_id, + display_name: self.peer_display_name(peer_id), + coords, + } + }) + .collect(); + let parent_addr = my_coords.parent_id(); + let tree_view = snap::TreeView { + my_node_addr: *tree.my_node_addr(), + root: *tree.root(), + is_root: tree.is_root(), + depth: my_coords.depth(), + my_coords: my_coords.entries().iter().map(|e| e.node_addr).collect(), + parent: *parent_addr, + parent_display_name: self.peer_display_name(parent_addr), + declaration_sequence: tree.my_declaration().sequence(), + declaration_signed: tree.my_declaration().is_signed(), + peer_tree_count: tree.peer_count(), + peers: tree_peers, + }; + + // --- bloom (show_bloom) --- + let bloom = self.bloom_state(); + let max_inbound_fpr = self.config().node.bloom.max_inbound_fpr; + let bloom_peers: Vec = self + .peers() + .map(|peer| { + let addr = *peer.node_addr(); + let filter = peer.inbound_filter().map(|f| snap::BloomPeerFilter { + estimated_count: f.estimated_count(max_inbound_fpr), + set_bits: f.count_ones(), + fill_ratio: f.fill_ratio(), + }); + snap::BloomPeerRow { + peer: addr, + display_name: self.peer_display_name(&addr), + has_filter: peer.filter_sequence() > 0, + filter_sequence: peer.filter_sequence(), + filter, + } + }) + .collect(); + let bloom_view = snap::BloomView { + own_node_addr: *self.node_addr(), + is_leaf_only: self.is_leaf_only(), + sequence: bloom.sequence(), + leaf_dependents: bloom.leaf_dependents().iter().copied().collect(), + peer_filters: bloom_peers, + }; + + // --- coord cache (show_cache, show_routing) --- + let cache = self.coord_cache(); + let cache_stats = cache.stats(now); + let cache_entries: Vec = cache + .iter(now) + .map(|(addr, entry)| snap::CacheEntryRow { + node_addr: *addr, + display_name: self.peer_display_name(addr), + depth: entry.coords().depth(), + coord_path: entry + .coords() + .entries() + .iter() + .map(|e| e.node_addr) + .collect(), + created_at: entry.created_at(), + last_used_ms: entry.last_used(), + path_mtu: entry.path_mtu(), + }) + .collect(); + let cache_view = snap::CacheView { + count: cache_stats.entries, + max_entries: cache_stats.max_entries, + fill_ratio: cache_stats.fill_ratio(), + default_ttl_ms: cache.default_ttl_ms(), + expired: cache_stats.expired, + avg_age_ms: cache_stats.avg_age_ms, + entries: cache_entries, + }; + + // --- F-queue / discovery routing scalars (show_routing) --- + let pending_lookups: Vec = self + .pending_lookups_iter() + .map(|(addr, lookup)| snap::PendingLookupRow { + target: *addr, + display_name: self.peer_display_name(addr), + initiated_ms: lookup.initiated_ms, + last_sent_ms: lookup.last_sent_ms, + attempt: lookup.attempt, + }) + .collect(); + let retries: Vec = self + .retry_state_iter() + .map(|(addr, state)| snap::RetryRow { + node_addr: *addr, + display_name: self.peer_display_name(addr), + retry_count: state.retry_count, + retry_after_ms: state.retry_after_ms, + auto_reconnect: state.reconnect, + }) + .collect(); + let routing_view = snap::RoutingView { + pending_lookups, + pending_tun_destinations: self.pending_tun_destinations(), + pending_tun_packets: self.pending_tun_total_packets(), + recent_requests: self.recent_request_count(), + retries, + }; + + // --- identity cache (show_identity_cache, show_routing) --- + let identity_entries: Vec = self + .identity_cache_iter() + .map(|(node_addr, pubkey, last_seen_ms)| { + let (xonly, _parity) = pubkey.x_only_public_key(); + let fips_addr = crate::identity::FipsAddress::from_node_addr(node_addr); + snap::IdentityRow { + node_addr: *node_addr, + npub: crate::identity::encode_npub(&xonly), + display_name: self.peer_display_name(node_addr), + ipv6_addr: format!("{}", fips_addr), + last_seen_ms, + } + }) + .collect(); + let identity_view = snap::IdentityView { + entries: identity_entries, + max_entries: self.identity_cache_max(), + }; + + let snapshot = snap::RoutingSnapshot { + tree: tree_view, + bloom: bloom_view, + cache: cache_view, + routing: routing_view, + identity: identity_view, + }; + self.routing_snapshot.store(std::sync::Arc::new(snapshot)); + } + + /// Project the Category-E per-entity tables (peers / sessions / links / + /// connections / transports + mmp) into an + /// [`EntitySnapshot`](crate::control::snapshot::EntitySnapshot) and publish + /// it via `ArcSwap`, so `show_peers` / `show_sessions` / `show_links` / + /// `show_connections` / `show_transports` / `show_mmp` render off the + /// rx_loop. + /// + /// **Q1 publisher placement (tick, like R3).** Every projected row needs a + /// display name resolved against the live peer/session tables and host map + /// (`&Node`); `show_peers` additionally needs the live tree state to derive + /// `is_parent` / `is_child` plus the Nostr-discovery failure-state map — + /// cross-subsystem reads available only with `&Node`. And most fields + /// (link/session traffic counters, MMP metrics, `last_seen`, noise counters) + /// mutate continuously on the data plane, not at the discrete entity + /// lifecycle mutators, so per-lifecycle-mutator publication (Q1-a) would not + /// capture their freshness anyway. The tick is the natural cadence with + /// coherent `&Node` access. + /// + /// **Structural sharing (the R4 umbrella mandate).** Each table is a + /// `Vec>`. The freshly-projected rows are reconciled against the + /// previously published snapshot via + /// [`reconcile_rows`](crate::control::snapshot::reconcile_rows): a row's + /// `Arc` is reused (kept by pointer) whenever it matches the prior row by + /// identity and compares equal by value, so a tick in which only one + /// peer/session changed re-allocates only that one row, not the whole table. + /// This is what keeps the publish cost off the hot path at scale (the exact + /// thing the umbrella warns a naive per-tick rebuild would violate). + fn publish_entities_snapshot(&self) { + use crate::control::snapshot as snap; + + let prev = self.entities_snapshot.load(); + + // --- peers (show_peers) --- + let tree = self.tree_state(); + let my_addr = *tree.my_node_addr(); + let parent_id = *tree.my_declaration().parent_id(); + let is_root = tree.is_root(); + + // Per-npub Nostr-traversal failure-state, indexed by npub for O(1) + // per-peer lookup (empty when Nostr discovery is disabled). + let nostr_state: std::collections::HashMap = self + .nostr_discovery_handle() + .map(|d| { + d.failure_state_snapshot() + .into_iter() + .map(|view| (view.npub.clone(), view)) + .collect() + }) + .unwrap_or_default(); + + let peer_rows: Vec = self + .peers() + .map(|peer| { + let node_addr = *peer.node_addr(); + let is_parent = !is_root && node_addr == parent_id; + let is_child = tree + .peer_declaration(&node_addr) + .is_some_and(|decl| *decl.parent_id() == my_addr); + + let link_info = self.get_link(&peer.link_id()).map(|link| { + let transport_type = self + .get_transport(&link.transport_id()) + .map(|h| h.transport_type().name.to_string()); + snap::PeerLinkInfo { + direction: format!("{}", link.direction()), + transport_type, + } + }); + + let stats = peer.link_stats(); + let nostr = nostr_state.get(&peer.npub()); + let nostr_traversal = snap::PeerNostrState { + consecutive_failures: nostr.map(|s| s.consecutive_failures).unwrap_or(0), + cooldown_until_ms: nostr.and_then(|s| s.cooldown_until_ms), + last_observed_skew_ms: nostr.and_then(|s| s.last_observed_skew_ms), + }; + + let noise = peer.noise_session().map(|session| snap::PeerNoiseCounters { + send_counter: session.current_send_counter(), + highest_recv_counter: session.highest_received_counter(), + }); + + let mmp = peer + .mmp() + .map(|mmp| project_entity_mmp(&mmp.metrics, format!("{}", mmp.mode()), None)); + + snap::PeerRow { + node_addr, + npub: peer.npub(), + display_name: self.peer_display_name(&node_addr), + ipv6_addr: format!("{}", peer.address()), + connectivity: format!("{}", peer.connectivity()), + link_id: peer.link_id().as_u64(), + authenticated_at_ms: peer.authenticated_at(), + last_seen_ms: peer.last_seen(), + has_tree_position: peer.has_tree_position(), + has_bloom_filter: peer.filter_sequence() > 0, + filter_sequence: peer.filter_sequence(), + is_parent, + is_child, + transport_addr: peer.current_addr().map(|a| format!("{}", a)), + link_info, + tree_depth: peer.coords().map(|c| c.depth()), + stats: snap::PeerLinkStats { + packets_sent: stats.packets_sent, + packets_recv: stats.packets_recv, + bytes_sent: stats.bytes_sent, + bytes_recv: stats.bytes_recv, + }, + replay_suppressed: peer.replay_suppressed_count(), + consecutive_decrypt_failures: peer.consecutive_decrypt_failures(), + nostr_traversal, + noise, + our_session_index: peer.our_index().map(|idx| idx.as_u32()), + rekey_in_progress: peer.rekey_in_progress(), + rekey_draining: peer.is_draining(), + current_k_bit: peer.current_k_bit(), + mmp, + } + }) + .collect(); + + // --- sessions (show_sessions) --- + let session_rows: Vec = self + .session_entries() + .map(|(addr, entry)| { + let state = if entry.is_established() { + "established" + } else if entry.is_initiating() { + "initiating" + } else if entry.is_awaiting_msg3() { + "awaiting_msg3" + } else { + "unknown" + }; + let (xonly, _parity) = entry.remote_pubkey().x_only_public_key(); + let (pkts_tx, pkts_rx, bytes_tx, bytes_rx) = entry.traffic_counters(); + + let resend_count = (!entry.is_established()).then(|| entry.resend_count()); + let established = entry.is_established().then(|| snap::SessionEstablished { + session_start_ms: entry.session_start_ms(), + current_k_bit: entry.current_k_bit(), + coords_warmup_remaining: entry.coords_warmup_remaining(), + is_draining: entry.is_draining(), + }); + let mmp = entry.mmp().map(|mmp| { + project_entity_mmp( + &mmp.metrics, + format!("{}", mmp.mode()), + Some(mmp.path_mtu.current_mtu()), + ) + }); + + snap::SessionRow { + remote_addr: *addr, + display_name: self.peer_display_name(addr), + state, + is_initiator: entry.is_initiator(), + last_activity_ms: entry.last_activity(), + npub: crate::identity::encode_npub(&xonly), + stats: snap::SessionStats { + packets_sent: pkts_tx, + packets_recv: pkts_rx, + bytes_sent: bytes_tx, + bytes_recv: bytes_rx, + }, + resend_count, + established, + mmp, + } + }) + .collect(); + + // --- links (show_links) --- + let link_rows: Vec = self + .links() + .map(|link| { + let stats = link.stats(); + snap::LinkRow { + link_id: link.link_id().as_u64(), + transport_id: link.transport_id().as_u32(), + remote_addr: format!("{}", link.remote_addr()), + direction: format!("{}", link.direction()), + state: format!("{}", link.state()), + created_at_ms: link.created_at(), + stats: snap::LinkStats { + packets_sent: stats.packets_sent, + packets_recv: stats.packets_recv, + bytes_sent: stats.bytes_sent, + bytes_recv: stats.bytes_recv, + last_recv_ms: stats.last_recv_ms, + }, + } + }) + .collect(); + + // --- connections (show_connections) --- + let connection_rows: Vec = self + .connections() + .map(|conn| snap::ConnectionRow { + link_id: conn.link_id().as_u64(), + direction: format!("{}", conn.direction()), + handshake_state: format!("{}", conn.handshake_state()), + started_at_ms: conn.started_at(), + last_activity_ms: conn.last_activity(), + resend_count: conn.resend_count(), + expected_peer: conn.expected_identity().map(|id| id.npub()), + }) + .collect(); + + // --- transports (show_transports) --- + let transport_rows: Vec = self + .transport_ids() + .map(|id| { + let handle = self.get_transport(id).unwrap(); + snap::TransportRow { + transport_id: id.as_u32(), + transport_type: handle.transport_type().name.to_string(), + state: format!("{}", handle.state()), + mtu: handle.mtu(), + name: handle.name().map(|s| s.to_string()), + local_addr: handle.local_addr().map(|a| format!("{}", a)), + tor_mode: handle.tor_mode().map(|s| s.to_string()), + onion_address: handle.onion_address().map(|s| s.to_string()), + tor_monitoring: handle + .tor_monitoring() + .map(|m| serde_json::to_value(&m).unwrap_or_default()), + stats: handle.transport_stats(), + } + }) + .collect(); + + // --- mmp peers (show_mmp link-layer) --- + let mmp_peer_rows: Vec = self + .peers() + .filter_map(|peer| { + let mmp = peer.mmp()?; + let addr = *peer.node_addr(); + let metrics = &mmp.metrics; + let srtt_ms = metrics.srtt_ms(); + let smoothed_etx = metrics.smoothed_etx(); + let lqi = match (srtt_ms, smoothed_etx) { + (Some(srtt), Some(setx)) => Some(setx * (1.0 + srtt / 100.0)), + _ => None, + }; + let trend = |dual: &crate::mmp::algorithms::DualEwma| { + dual.initialized() + .then(|| crate::control::queries::trend_label(dual.short(), dual.long())) + }; + Some(snap::MmpPeerRow { + peer: addr, + display_name: self.peer_display_name(&addr), + mode: format!("{}", mmp.mode()), + loss_rate: metrics.loss_rate(), + etx: metrics.etx, + goodput_bps: metrics.goodput_bps, + spin_bit_initiator: mmp.spin_bit.is_initiator(), + smoothed_loss: metrics.smoothed_loss(), + smoothed_etx, + srtt_ms, + lqi, + trends: snap::MmpTrends { + rtt_trend: trend(&metrics.rtt_trend), + loss_trend: trend(&metrics.loss_trend), + goodput_trend: trend(&metrics.goodput_trend), + jitter_trend: trend(&metrics.jitter_trend), + }, + delivery_ratio_forward: metrics.delivery_ratio_forward, + delivery_ratio_reverse: metrics.delivery_ratio_reverse, + ecn_ce_count: metrics.last_ecn_ce_count(), + }) + }) + .collect(); + + // --- mmp sessions (show_mmp session-layer) --- + let mmp_session_rows: Vec = self + .session_entries() + .filter_map(|(addr, entry)| { + let mmp = entry.mmp()?; + let metrics = &mmp.metrics; + let srtt_ms = metrics.srtt_ms(); + let smoothed_etx = metrics.smoothed_etx(); + let sqi = match (srtt_ms, smoothed_etx) { + (Some(srtt), Some(setx)) => Some(setx * (1.0 + srtt / 100.0)), + _ => None, + }; + Some(snap::MmpSessionRow { + remote: *addr, + display_name: self.peer_display_name(addr), + mode: format!("{}", mmp.mode()), + loss_rate: metrics.loss_rate(), + etx: metrics.etx, + path_mtu: mmp.path_mtu.current_mtu(), + smoothed_loss: metrics.smoothed_loss(), + smoothed_etx, + srtt_ms, + sqi, + }) + }) + .collect(); + + let snapshot = snap::EntitySnapshot { + peers: snap::reconcile_rows(&prev.peers, peer_rows, |r| r.node_addr), + sessions: snap::reconcile_rows(&prev.sessions, session_rows, |r| r.remote_addr), + links: snap::reconcile_rows(&prev.links, link_rows, |r| r.link_id), + connections: snap::reconcile_rows(&prev.connections, connection_rows, |r| r.link_id), + transports: snap::reconcile_rows(&prev.transports, transport_rows, |r| r.transport_id), + mmp_peers: snap::reconcile_rows(&prev.mmp_peers, mmp_peer_rows, |r| r.peer), + mmp_sessions: snap::reconcile_rows(&prev.mmp_sessions, mmp_session_rows, |r| r.remote), + }; + self.entities_snapshot.store(std::sync::Arc::new(snapshot)); } // === TUN Interface === @@ -2297,6 +2854,38 @@ impl Node { } } +/// Project an MMP metrics block into the snapshot +/// [`EntityMmp`](crate::control::snapshot::EntityMmp) shared by `show_peers` +/// (link-layer, `path_mtu = None`) and `show_sessions` (session-layer, +/// `path_mtu = Some`). `quality_index` (`lqi` for peers / `sqi` for sessions) +/// is precomputed here exactly as the on-loop queries do, so the render is a +/// plain field emit. +fn project_entity_mmp( + metrics: &crate::mmp::metrics::MmpMetrics, + mode: String, + path_mtu: Option, +) -> crate::control::snapshot::EntityMmp { + let srtt_ms = metrics.srtt_ms(); + let smoothed_etx = metrics.smoothed_etx(); + let quality_index = match (srtt_ms, smoothed_etx) { + (Some(srtt), Some(setx)) => Some(setx * (1.0 + srtt / 100.0)), + _ => None, + }; + crate::control::snapshot::EntityMmp { + mode, + srtt_ms, + loss_rate: metrics.loss_rate(), + etx: metrics.etx, + goodput_bps: metrics.goodput_bps, + delivery_ratio_forward: metrics.delivery_ratio_forward, + delivery_ratio_reverse: metrics.delivery_ratio_reverse, + smoothed_loss: metrics.smoothed_loss(), + smoothed_etx, + quality_index, + path_mtu, + } +} + impl fmt::Debug for Node { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("Node")