diff --git a/CHANGELOG.md b/CHANGELOG.md index c83ed8f..bf33382 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,6 +25,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 direct counters at the bloom / discovery / forwarding sites are retained alongside the new dispatch while the rollout is in progress; a later change collapses the duplicate increment. +- Internal atomic metric registry (`Arc`) that shadows + the plain-`u64` `NodeStats` counters, written alongside them and + validated by a whole-struct debug-build parity check. Covers the + forwarding receive counters, the full discovery counter family, and the + tree, bloom, congestion, and error-signal counter families so far, with + the hottest counters cache-line padded. Behavior-neutral: + `NodeStats` remains the serving path. Groundwork for sampling metrics + without contending the receive loop. - `Node::update_peers` for runtime peer-list refresh, returning an `UpdatePeersOutcome` summarizing added, removed, and retained peers. Re-derives active peer connections from a new peer configuration @@ -179,6 +187,23 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Six discovery counters (`req_decode_error`, `req_duplicate`, + `req_ttl_exhausted`, `resp_decode_error`, `resp_identity_miss`, + `resp_proof_failed`) no longer double-count. Each was incremented both + by a direct bump and again through the typed reject dispatch; the + redundant direct increment is removed, so each counts once per event. +- Six bloom counters (`decode_error`, `invalid`, `non_v1`, + `unknown_peer`, `stale`, `fill_exceeded`) no longer double-count. Each + was incremented both by a direct bump and again through the typed + reject dispatch; the redundant direct increment is removed, so each + counts once per event. +- Five forwarding reject packet counters (`decode_error_packets`, + `ttl_exhausted_packets`, `drop_no_route_packets`, + `drop_mtu_exceeded_packets`, `drop_send_error_packets`) no longer + double-count. Each was incremented both by the byte-aware outcome + recorder and again through the typed reject dispatch; the two calls are + collapsed into a single byte-aware reject entry point, so packets and + bytes each count once per event. - A stale FSP (session-layer) session is now cleared when a peer restart is detected during FMP rekey or cross-connection promotion. Previously the old session could linger after the peer came back diff --git a/src/control/queries.rs b/src/control/queries.rs index 2b0b699..abf7b82 100644 --- a/src/control/queries.rs +++ b/src/control/queries.rs @@ -47,7 +47,7 @@ pub fn show_status(node: &Node) -> Value { .map(|p| p.display().to_string()) .unwrap_or_else(|_| "-".into()); let uptime_secs = node.uptime().as_secs(); - let fwd = node.stats().snapshot().forwarding; + let fwd = node.metrics().forwarding.snapshot(); // Inline last-N-second sparklines for dashboard rendering. Kept // short so the status payload stays compact; longer windows use @@ -336,7 +336,7 @@ pub fn show_tree(node: &Node) -> Value { let parent_hex = hex::encode(parent_addr.as_bytes()); let parent_display = node.peer_display_name(parent_addr); - let tree_stats = node.stats().snapshot().tree; + let tree_stats = node.metrics().tree.snapshot(); json!({ "my_node_addr": hex::encode(tree.my_node_addr().as_bytes()), @@ -469,7 +469,7 @@ pub fn show_bloom(node: &Node) -> Value { }) .collect(); - let bloom_stats = node.stats().snapshot().bloom; + let bloom_stats = node.metrics().bloom.snapshot(); json!({ "own_node_addr": hex::encode(node.node_addr().as_bytes()), @@ -692,7 +692,7 @@ pub fn show_routing(node: &Node) -> Value { let cache = node.coord_cache(); let now = now_ms(); let cache_stats = cache.stats(now); - let node_stats = node.stats().snapshot(); + let metrics = node.metrics(); // Pending discovery lookups (individual targets) let lookups: Vec = node @@ -731,10 +731,10 @@ pub fn show_routing(node: &Node) -> Value { "pending_tun_packets": node.pending_tun_total_packets(), "recent_requests": node.recent_request_count(), "retries": retries, - "forwarding": serde_json::to_value(&node_stats.forwarding).unwrap_or_default(), - "discovery": serde_json::to_value(&node_stats.discovery).unwrap_or_default(), - "error_signals": serde_json::to_value(&node_stats.errors).unwrap_or_default(), - "congestion": serde_json::to_value(&node_stats.congestion).unwrap_or_default(), + "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(), }) } diff --git a/src/identity/local.rs b/src/identity/local.rs index 064e0aa..f79e7cd 100644 --- a/src/identity/local.rs +++ b/src/identity/local.rs @@ -11,6 +11,7 @@ use super::{FipsAddress, IdentityError, NodeAddr, sha256}; /// /// The identity holds the secp256k1 keypair and provides methods for signing /// and verifying protocol messages. +#[derive(Clone)] pub struct Identity { keypair: Keypair, node_addr: NodeAddr, diff --git a/src/node/bloom.rs b/src/node/bloom.rs index 285988d..f16c243 100644 --- a/src/node/bloom.rs +++ b/src/node/bloom.rs @@ -7,7 +7,7 @@ use crate::NodeAddr; use crate::bloom::BloomFilter; use crate::protocol::FilterAnnounce; -use super::reject::{BloomReject, RejectReason}; +use super::reject::BloomReject; use super::{Node, NodeError}; use std::collections::HashMap; use tracing::{debug, warn}; @@ -58,7 +58,7 @@ impl Node { // Check debounce if !self.bloom_state.should_send_update(peer_addr, now_ms) { - self.stats_mut().bloom.debounce_suppressed += 1; + self.metrics().bloom.debounce_suppressed.inc(); // Either not pending or rate-limited; will retry on tick return Ok(()); } @@ -73,11 +73,11 @@ impl Node { // Send if let Err(e) = self.send_encrypted_link_message(peer_addr, &encoded).await { - self.stats_mut().bloom.send_failed += 1; + self.metrics().bloom.send_failed.inc(); return Err(e); } - self.stats_mut().bloom.sent += 1; + self.metrics().bloom.sent.inc(); // Self-plausibility check: WARN if our own outgoing filter is // above the antipoison cap. Independent detection signal if @@ -85,7 +85,7 @@ impl Node { // despite M1. Rate-limited to once per 60s globally — outgoing // cadence can be per-tick during churn, and we want the // operator to see one clear message, not spam. - let max_fpr = self.config.node.bloom.max_inbound_fpr; + let max_fpr = self.config().node.bloom.max_inbound_fpr; let out_fill = sent_filter.fill_ratio(); let out_fpr = out_fill.powi(sent_filter.hash_count() as i32); if out_fpr > max_fpr { @@ -160,14 +160,12 @@ impl Node { /// 3. Store the filter on the peer /// 4. Mark other peers for outgoing filter update pub(super) async fn handle_filter_announce(&mut self, from: &NodeAddr, payload: &[u8]) { - self.stats_mut().bloom.received += 1; + self.metrics().bloom.received.inc(); let announce = match FilterAnnounce::decode(payload) { Ok(a) => a, Err(e) => { - self.stats_mut().bloom.decode_error += 1; - self.stats_mut() - .record_reject(RejectReason::Bloom(BloomReject::DecodeError)); + self.metrics().bloom.record_reject(BloomReject::DecodeError); debug!(from = %self.peer_display_name(from), error = %e, "Malformed FilterAnnounce"); return; } @@ -175,16 +173,12 @@ impl Node { // Validate if !announce.is_valid() { - self.stats_mut().bloom.invalid += 1; - self.stats_mut() - .record_reject(RejectReason::Bloom(BloomReject::Invalid)); + self.metrics().bloom.record_reject(BloomReject::Invalid); debug!(from = %self.peer_display_name(from), "FilterAnnounce filter/size_class mismatch"); return; } if !announce.is_v1_compliant() { - self.stats_mut().bloom.non_v1 += 1; - self.stats_mut() - .record_reject(RejectReason::Bloom(BloomReject::NonV1)); + self.metrics().bloom.record_reject(BloomReject::NonV1); debug!(from = %self.peer_display_name(from), size_class = announce.size_class, "Non-v1 FilterAnnounce rejected"); return; } @@ -193,9 +187,7 @@ impl Node { let current_seq = match self.peers.get(from) { Some(peer) => peer.filter_sequence(), None => { - self.stats_mut().bloom.unknown_peer += 1; - self.stats_mut() - .record_reject(RejectReason::Bloom(BloomReject::UnknownPeer)); + self.metrics().bloom.record_reject(BloomReject::UnknownPeer); debug!(from = %self.peer_display_name(from), "FilterAnnounce from unknown peer"); return; } @@ -203,9 +195,7 @@ impl Node { // Reject stale/replay if announce.sequence <= current_seq { - self.stats_mut().bloom.stale += 1; - self.stats_mut() - .record_reject(RejectReason::Bloom(BloomReject::Stale)); + self.metrics().bloom.record_reject(BloomReject::Stale); debug!( from = %self.peer_display_name(from), received_seq = announce.sequence, @@ -221,13 +211,13 @@ impl Node { // untouched so the peer is not permanently silenced and an // on-path attacker cannot weaponize a single corrupted frame // to wipe a victim's contribution to aggregation. - let max_fpr = self.config.node.bloom.max_inbound_fpr; + let max_fpr = self.config().node.bloom.max_inbound_fpr; let fill = announce.filter.fill_ratio(); let fpr = fill.powi(announce.filter.hash_count() as i32); if fpr > max_fpr { - self.stats_mut().bloom.fill_exceeded += 1; - self.stats_mut() - .record_reject(RejectReason::Bloom(BloomReject::FillExceeded)); + self.metrics() + .bloom + .record_reject(BloomReject::FillExceeded); warn!( from = %self.peer_display_name(from), seq = announce.sequence, @@ -239,7 +229,7 @@ impl Node { return; } - self.stats_mut().bloom.accepted += 1; + self.metrics().bloom.accepted.inc(); let now_ms = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) diff --git a/src/node/context.rs b/src/node/context.rs new file mode 100644 index 0000000..38538bf --- /dev/null +++ b/src/node/context.rs @@ -0,0 +1,80 @@ +//! Shared immutable context bundle. +//! +//! [`NodeContext`] groups the [`Node`](super::Node)'s effectively-immutable +//! fields behind a single `Arc` so that handlers can eventually borrow a +//! cheap `&NodeContext` clone instead of `&self`. +//! +//! During the migration it is a *parallel, authoritative* copy of the +//! corresponding `Node` fields: both are kept in lockstep at the only three +//! mutation points — the constructor, [`update_peers`](super::Node::update_peers), +//! and the test-only `set_max_*` setters — via +//! [`Node::rebuild_context`](super::Node::rebuild_context). Readers migrate +//! onto the bundle incrementally; the duplicated `Node` fields are removed +//! once the last reader has moved over. + +use std::sync::Arc; + +use crate::{Config, Identity}; + +/// Effectively-immutable `Node` state, shared via `Arc`. +#[derive(Clone)] +pub(crate) struct NodeContext { + /// Loaded configuration. A static snapshot: replaced wholesale (never + /// interior-mutated) when `update_peers` rebuilds the runtime peer list. + pub config: Arc, + + /// This node's cryptographic identity. + pub identity: Identity, + + /// Random epoch generated at startup for peer restart detection. + // Consumed by readers migrating in a later sub-PR. + #[allow(dead_code)] + pub startup_epoch: [u8; 8], + + /// Instant when the node was created, for uptime reporting. + pub started_at: std::time::Instant, + + /// Whether this is a leaf-only node. + pub is_leaf_only: bool, + + /// Maximum connections (0 = unlimited). + // Consumed by readers migrating in a later sub-PR. + #[allow(dead_code)] + pub max_connections: usize, + + /// Maximum peers (0 = unlimited). + // Consumed by readers migrating in a later sub-PR. + #[allow(dead_code)] + pub max_peers: usize, + + /// Maximum links (0 = unlimited). + // Consumed by readers migrating in a later sub-PR. + #[allow(dead_code)] + pub max_links: usize, +} + +impl NodeContext { + /// Build a context bundle from the individual values. + #[allow(clippy::too_many_arguments)] + pub fn new( + config: Arc, + identity: Identity, + startup_epoch: [u8; 8], + started_at: std::time::Instant, + is_leaf_only: bool, + max_connections: usize, + max_peers: usize, + max_links: usize, + ) -> Self { + Self { + config, + identity, + startup_epoch, + started_at, + is_leaf_only, + max_connections, + max_peers, + max_links, + } + } +} diff --git a/src/node/handlers/discovery.rs b/src/node/handlers/discovery.rs index 872a99b..e3d8c7b 100644 --- a/src/node/handlers/discovery.rs +++ b/src/node/handlers/discovery.rs @@ -5,7 +5,7 @@ //! bloom filter contains the target. TTL and request_id dedup provide //! safety bounds. -use crate::node::reject::{DiscoveryReject, RejectReason}; +use crate::node::reject::DiscoveryReject; use crate::node::{Node, RecentRequest}; use crate::protocol::{LookupRequest, LookupResponse}; use crate::transport::{TransportAddr, TransportId}; @@ -25,14 +25,14 @@ impl Node { /// 5. If we're the target, generate and send response /// 6. If TTL > 0, forward to tree peers whose bloom filter matches pub(in crate::node) async fn handle_lookup_request(&mut self, from: &NodeAddr, payload: &[u8]) { - self.stats_mut().discovery.req_received += 1; + self.metrics().discovery.req_received.inc(); let request = match LookupRequest::decode(payload) { Ok(req) => req, Err(e) => { - self.stats_mut().discovery.req_decode_error += 1; - self.stats_mut() - .record_reject(RejectReason::Discovery(DiscoveryReject::ReqDecodeError)); + self.metrics() + .discovery + .record_reject(DiscoveryReject::ReqDecodeError); debug!(from = %self.peer_display_name(from), error = %e, "Malformed LookupRequest"); return; } @@ -45,9 +45,9 @@ impl Node { // Also serves as loop protection — tree routing is loop-free, // but request_id dedup catches edge cases during tree restructuring. if self.recent_requests.contains_key(&request.request_id) { - self.stats_mut().discovery.req_duplicate += 1; - self.stats_mut() - .record_reject(RejectReason::Discovery(DiscoveryReject::ReqDuplicate)); + self.metrics() + .discovery + .record_reject(DiscoveryReject::ReqDuplicate); debug!( request_id = request.request_id, from = %self.peer_display_name(from), @@ -57,8 +57,9 @@ impl Node { } if self.recent_requests.len() >= MAX_RECENT_DISCOVERY_REQUESTS { - self.stats_mut() - .record_reject(RejectReason::Discovery(DiscoveryReject::ReqDedupCacheFull)); + self.metrics() + .discovery + .record_reject(DiscoveryReject::ReqDedupCacheFull); debug!( request_id = request.request_id, from = %self.peer_display_name(from), @@ -75,7 +76,7 @@ impl Node { // Are we the target? if request.target == *self.node_addr() { - self.stats_mut().discovery.req_target_is_us += 1; + self.metrics().discovery.req_target_is_us.inc(); debug!( request_id = request.request_id, origin = %self.peer_display_name(&request.origin), @@ -93,7 +94,7 @@ impl Node { .discovery_forward_limiter .should_forward(&request.target) { - self.stats_mut().discovery.req_forward_rate_limited += 1; + self.metrics().discovery.req_forward_rate_limited.inc(); debug!( request_id = request.request_id, target = %self.peer_display_name(&request.target), @@ -101,12 +102,12 @@ impl Node { ); return; } - self.stats_mut().discovery.req_forwarded += 1; + self.metrics().discovery.req_forwarded.inc(); self.forward_lookup_request(request).await; } else { - self.stats_mut().discovery.req_ttl_exhausted += 1; - self.stats_mut() - .record_reject(RejectReason::Discovery(DiscoveryReject::ReqTtlExhausted)); + self.metrics() + .discovery + .record_reject(DiscoveryReject::ReqTtlExhausted); debug!( request_id = request.request_id, target = %self.peer_display_name(&request.target), @@ -127,14 +128,14 @@ impl Node { from: &NodeAddr, payload: &[u8], ) { - self.stats_mut().discovery.resp_received += 1; + self.metrics().discovery.resp_received.inc(); let mut response = match LookupResponse::decode(payload) { Ok(resp) => resp, Err(e) => { - self.stats_mut().discovery.resp_decode_error += 1; - self.stats_mut() - .record_reject(RejectReason::Discovery(DiscoveryReject::RespDecodeError)); + self.metrics() + .discovery + .record_reject(DiscoveryReject::RespDecodeError); debug!(from = %self.peer_display_name(from), error = %e, "Malformed LookupResponse"); return; } @@ -158,7 +159,7 @@ impl Node { // Transit node: reverse-path forward let from_peer = recent.from_peer; - self.stats_mut().discovery.resp_forwarded += 1; + self.metrics().discovery.resp_forwarded.inc(); // Apply path_mtu min() from the outgoing link's transport MTU self.apply_outgoing_link_mtu_to_response(&mut response, &from_peer); @@ -190,9 +191,9 @@ impl Node { let target_pubkey = match self.lookup_by_fips_prefix(&prefix) { Some((_addr, pubkey)) => pubkey, None => { - self.stats_mut().discovery.resp_identity_miss += 1; - self.stats_mut() - .record_reject(RejectReason::Discovery(DiscoveryReject::RespIdentityMiss)); + self.metrics() + .discovery + .record_reject(DiscoveryReject::RespIdentityMiss); warn!( request_id = response.request_id, target = %self.peer_display_name(&target), @@ -208,9 +209,9 @@ impl Node { let proof_data = LookupResponse::proof_bytes(response.request_id, &target, &response.target_coords); if !peer_id.verify(&proof_data, &response.proof) { - self.stats_mut().discovery.resp_proof_failed += 1; - self.stats_mut() - .record_reject(RejectReason::Discovery(DiscoveryReject::RespProofFailed)); + self.metrics() + .discovery + .record_reject(DiscoveryReject::RespProofFailed); warn!( request_id = response.request_id, target = %self.peer_display_name(&target), @@ -219,7 +220,7 @@ impl Node { return; } - self.stats_mut().discovery.resp_accepted += 1; + self.metrics().discovery.resp_accepted.inc(); // Clear backoff on success — target is reachable self.discovery_backoff.record_success(&target); @@ -265,10 +266,10 @@ impl Node { self.pending_lookups.remove(&target); // If an established session exists, reset the warmup counter. + let n = self.config().node.session.coords_warmup_packets; if let Some(entry) = self.sessions.get_mut(&target) && entry.is_established() { - let n = self.config.node.session.coords_warmup_packets; entry.set_coords_warmup_remaining(n); debug!( dest = %self.peer_display_name(&target), @@ -315,8 +316,9 @@ impl Node { origin = %self.peer_display_name(&request.origin), "Cannot route LookupResponse: no reverse path or tree route to origin" ); - self.stats_mut() - .record_reject(RejectReason::Discovery(DiscoveryReject::RespNoRoute)); + self.metrics() + .discovery + .record_reject(DiscoveryReject::RespNoRoute); return; } } @@ -380,7 +382,7 @@ impl Node { .map(|(addr, _)| *addr) .collect(); if fallback.is_empty() { - self.stats_mut().discovery.req_no_tree_peer += 1; + self.metrics().discovery.req_no_tree_peer.inc(); trace!( request_id = request.request_id, "No eligible peers to forward LookupRequest" @@ -393,7 +395,7 @@ impl Node { }; if used_fallback { - self.stats_mut().discovery.req_fallback_forwarded += 1; + self.metrics().discovery.req_fallback_forwarded.inc(); debug!( request_id = request.request_id, target = %self.peer_display_name(&request.target), @@ -431,7 +433,7 @@ impl Node { /// The originator does NOT record the request_id in recent_requests, /// so when the response arrives, it's recognized as "our request". pub(in crate::node) async fn initiate_lookup(&mut self, target: &NodeAddr, ttl: u8) -> usize { - self.stats_mut().discovery.req_initiated += 1; + self.metrics().discovery.req_initiated.inc(); let origin = *self.node_addr(); let origin_coords = self.tree_state().my_coords().clone(); @@ -487,7 +489,7 @@ impl Node { // Dedup: any pending lookup means we are already trying. if self.pending_lookups.contains_key(dest) { - self.stats_mut().discovery.req_deduplicated += 1; + self.metrics().discovery.req_deduplicated.inc(); debug!( target_node = %self.peer_display_name(dest), "Discovery lookup deduplicated, already pending" @@ -498,7 +500,7 @@ impl Node { // Optional post-failure suppression. Defaults are 0/0 (inert); // operators can opt in by setting `node.discovery.backoff_*_secs`. if self.discovery_backoff.is_suppressed(dest) { - self.stats_mut().discovery.req_backoff_suppressed += 1; + self.metrics().discovery.req_backoff_suppressed.inc(); debug!( target_node = %self.peer_display_name(dest), failures = self.discovery_backoff.failure_count(dest), @@ -511,7 +513,7 @@ impl Node { // it's not in the mesh — skip the lookup and record as failure. let reachable = self.peers.values().any(|peer| peer.may_reach(dest)); if !reachable { - self.stats_mut().discovery.req_bloom_miss += 1; + self.metrics().discovery.req_bloom_miss.inc(); self.discovery_backoff.record_failure(dest); debug!( target_node = %self.peer_display_name(dest), @@ -522,7 +524,7 @@ impl Node { self.pending_lookups .insert(*dest, PendingLookup::new(now_ms)); - let ttl = self.config.node.discovery.ttl; + let ttl = self.config().node.discovery.ttl; let sent = self.initiate_lookup(dest, ttl).await; // If no tree peers had the target, fail immediately @@ -547,7 +549,7 @@ impl Node { /// - Otherwise: declare the destination unreachable, drop queued packets, /// and emit ICMPv6 destination-unreachable for each. pub(in crate::node) async fn check_pending_lookups(&mut self, now_ms: u64) { - let timeouts = self.config.node.discovery.attempt_timeouts_secs.clone(); + let timeouts = self.config().node.discovery.attempt_timeouts_secs.clone(); let max_attempts = timeouts.len() as u8; // Collect targets needing action @@ -573,7 +575,7 @@ impl Node { entry.last_sent_ms = now_ms; let attempt = entry.attempt; - let ttl = self.config.node.discovery.ttl; + let ttl = self.config().node.discovery.ttl; let sent = self.initiate_lookup(&target, ttl).await; if sent > 0 { debug!( @@ -587,7 +589,7 @@ impl Node { // Process timeouts for addr in to_timeout { - self.stats_mut().discovery.resp_timed_out += 1; + self.metrics().discovery.resp_timed_out.inc(); self.pending_lookups.remove(&addr); // Record failure for optional backoff @@ -623,7 +625,7 @@ impl Node { /// Remove expired entries from the recent_requests cache. fn purge_expired_requests(&mut self, current_time_ms: u64) { - let expiry_ms = self.config.node.discovery.recent_expiry_secs * 1000; + let expiry_ms = self.config().node.discovery.recent_expiry_secs * 1000; self.recent_requests .retain(|_, entry| !entry.is_expired(current_time_ms, expiry_ms)); } diff --git a/src/node/handlers/forwarding.rs b/src/node/handlers/forwarding.rs index f89fe8a..0bb2a62 100644 --- a/src/node/handlers/forwarding.rs +++ b/src/node/handlers/forwarding.rs @@ -6,7 +6,7 @@ //! locally, and generates error signals on routing failure. use crate::NodeAddr; -use crate::node::reject::{ForwardingReject, RejectReason}; +use crate::node::reject::ForwardingReject; use crate::node::session_wire::{ FSP_COMMON_PREFIX_SIZE, FSP_HEADER_SIZE, FSP_PHASE_ESTABLISHED, FSP_PHASE_MSG1, FSP_PHASE_MSG2, FspCommonPrefix, parse_encrypted_coords, @@ -30,16 +30,14 @@ impl Node { payload: &[u8], incoming_ce: bool, ) { - self.stats_mut().forwarding.record_received(payload.len()); + self.metrics().forwarding.record_received(payload.len()); let datagram_ref = match SessionDatagramRef::decode(payload) { Ok(dg) => dg, Err(e) => { - self.stats_mut() + self.metrics() .forwarding - .record_decode_error(payload.len()); - self.stats_mut() - .record_reject(RejectReason::Forwarding(ForwardingReject::DecodeError)); + .record_reject_bytes(ForwardingReject::DecodeError, payload.len()); debug!(error = %e, "Malformed SessionDatagram"); return; } @@ -48,11 +46,9 @@ impl Node { // TTL enforcement: decrement for forwarding and drop only if the // received datagram was already exhausted. if datagram_ref.ttl == 0 { - self.stats_mut() + self.metrics() .forwarding - .record_ttl_exhausted(payload.len()); - self.stats_mut() - .record_reject(RejectReason::Forwarding(ForwardingReject::TtlExhausted)); + .record_reject_bytes(ForwardingReject::TtlExhausted, payload.len()); debug!( src = %datagram_ref.src_addr, dest = %datagram_ref.dest_addr, @@ -68,7 +64,7 @@ impl Node { // Local delivery: dispatch to session layer handlers without // materializing an owned SessionDatagram payload Vec. if datagram_ref.dest_addr == *self.node_addr() { - self.stats_mut().forwarding.record_delivered(payload.len()); + self.metrics().forwarding.record_delivered(payload.len()); self.handle_session_payload( &datagram_ref.src_addr, datagram_ref.payload, @@ -86,11 +82,9 @@ impl Node { let next_hop_addr = match self.find_next_hop(&datagram.dest_addr) { Some(peer) => *peer.node_addr(), None => { - self.stats_mut() + self.metrics() .forwarding - .record_drop_no_route(payload.len()); - self.stats_mut() - .record_reject(RejectReason::Forwarding(ForwardingReject::NoRoute)); + .record_reject_bytes(ForwardingReject::NoRoute, payload.len()); debug!( src = %self.peer_display_name(&datagram.src_addr), dest = %self.peer_display_name(&datagram.dest_addr), @@ -118,7 +112,7 @@ impl Node { let local_congestion = self.detect_congestion(&next_hop_addr); let outgoing_ce = incoming_ce || local_congestion; if local_congestion { - self.stats_mut().congestion.record_congestion_detected(); + self.metrics().congestion.congestion_detected.inc(); let now = Instant::now(); let should_log = self .last_congestion_log @@ -138,19 +132,15 @@ impl Node { { match e { NodeError::MtuExceeded { mtu, .. } => { - self.stats_mut() + self.metrics() .forwarding - .record_drop_mtu_exceeded(payload.len()); - self.stats_mut() - .record_reject(RejectReason::Forwarding(ForwardingReject::MtuExceeded)); + .record_reject_bytes(ForwardingReject::MtuExceeded, payload.len()); self.send_mtu_exceeded_error(&datagram, mtu).await; } _ => { - self.stats_mut() + self.metrics() .forwarding - .record_drop_send_error(payload.len()); - self.stats_mut() - .record_reject(RejectReason::Forwarding(ForwardingReject::SendError)); + .record_reject_bytes(ForwardingReject::SendError, payload.len()); debug!( next_hop = %next_hop_addr, dest = %datagram.dest_addr, @@ -160,9 +150,9 @@ impl Node { } } } else { - self.stats_mut().forwarding.record_forwarded(encoded.len()); + self.metrics().forwarding.record_forwarded(encoded.len()); if outgoing_ce { - self.stats_mut().congestion.record_ce_forwarded(); + self.metrics().congestion.ce_forwarded.inc(); } } } @@ -291,7 +281,7 @@ impl Node { }; let error_dg = SessionDatagram::new(my_addr, original.src_addr, error_payload) - .with_ttl(self.config.node.session.default_ttl); + .with_ttl(self.config().node.session.default_ttl); let next_hop_addr = match self.find_next_hop(&original.src_addr) { Some(peer) => *peer.node_addr(), @@ -343,7 +333,7 @@ impl Node { let error_payload = MtuExceeded::new(original.dest_addr, my_addr, bottleneck_mtu).encode(); let error_dg = SessionDatagram::new(my_addr, original.src_addr, error_payload) - .with_ttl(self.config.node.session.default_ttl); + .with_ttl(self.config().node.session.default_ttl); let next_hop_addr = match self.find_next_hop(&original.src_addr) { Some(peer) => *peer.node_addr(), @@ -385,7 +375,7 @@ impl Node { /// /// Returns `true` if any signal indicates congestion. pub(in crate::node) fn detect_congestion(&self, next_hop: &NodeAddr) -> bool { - if !self.config.node.ecn.enabled { + if !self.config().node.ecn.enabled { return false; } // Outgoing link MMP metrics @@ -393,8 +383,8 @@ impl Node { && let Some(mmp) = peer.mmp() { let metrics = &mmp.metrics; - if metrics.loss_rate() >= self.config.node.ecn.loss_threshold - || metrics.etx >= self.config.node.ecn.etx_threshold + if metrics.loss_rate() >= self.config().node.ecn.loss_threshold + || metrics.etx >= self.config().node.ecn.etx_threshold { return true; } @@ -423,7 +413,7 @@ impl Node { } } for tid in new_drop_events { - self.stats_mut().congestion.record_kernel_drop_event(); + self.metrics().congestion.kernel_drop_events.inc(); warn!( transport_id = tid.as_u32(), "Kernel recv drops first observed on transport" diff --git a/src/node/handlers/handshake.rs b/src/node/handlers/handshake.rs index 91f0603..e7c0083 100644 --- a/src/node/handlers/handshake.rs +++ b/src/node/handlers/handshake.rs @@ -177,7 +177,7 @@ impl Node { packet.timestamp_ms, ); - let our_keypair = self.identity.keypair(); + let our_keypair = self.identity().keypair(); let noise_msg1 = &packet.data[header.noise_msg1_offset..]; let msg2_response = match conn.receive_handshake_init( our_keypair, @@ -285,7 +285,7 @@ impl Node { // immediately — a genuine rekey can't fire that fast. let session_age_secs = existing_peer.session_established_at().elapsed().as_secs(); - if self.config.node.rekey.enabled + if self.config().node.rekey.enabled && existing_peer.has_session() && existing_peer.is_healthy() && session_age_secs >= 30 @@ -310,7 +310,7 @@ impl Node { // simultaneously. Apply tie-breaker — smaller NodeAddr // wins as initiator (same as cross-connection resolution). if existing_peer.rekey_in_progress() { - let our_addr = self.identity.node_addr(); + let our_addr = self.identity().node_addr(); if our_addr < &peer_node_addr { // We win as initiator — drop their msg1. // Our msg2 will arrive at peer, who completes @@ -489,7 +489,7 @@ impl Node { packet.transport_id, packet.remote_addr.clone(), LinkDirection::Inbound, - Duration::from_millis(self.config.node.base_rtt_ms), + Duration::from_millis(self.config().node.base_rtt_ms), ); self.links.insert(link_id, link); @@ -846,7 +846,7 @@ impl Node { // outbound = the loser's inbound). if self.peers.contains_key(&peer_node_addr) { let our_outbound_wins = cross_connection_winner( - self.identity.node_addr(), + self.identity().node_addr(), &peer_node_addr, true, // this IS our outbound ); @@ -1111,7 +1111,11 @@ impl Node { // authenticated connection must replace them regardless of the // tie-breaker direction. let this_wins = remote_epoch_changed - || cross_connection_winner(self.identity.node_addr(), &peer_node_addr, is_outbound); + || cross_connection_winner( + self.identity().node_addr(), + &peer_node_addr, + is_outbound, + ); if this_wins { // This connection wins, replace the existing peer @@ -1165,11 +1169,11 @@ impl Node { current_addr, link_stats, is_outbound, - &self.config.node.mmp, + &self.config().node.mmp, remote_epoch, ); new_peer.set_tree_announce_min_interval_ms( - self.config.node.tree.announce_min_interval_ms, + self.config().node.tree.announce_min_interval_ms, ); self.peers.insert(peer_node_addr, new_peer); @@ -1268,11 +1272,12 @@ impl Node { current_addr, link_stats, is_outbound, - &self.config.node.mmp, + &self.config().node.mmp, remote_epoch, ); - new_peer - .set_tree_announce_min_interval_ms(self.config.node.tree.announce_min_interval_ms); + new_peer.set_tree_announce_min_interval_ms( + self.config().node.tree.announce_min_interval_ms, + ); if let Some(ts) = old_announce_ts { new_peer.set_last_tree_announce_sent_ms(ts); } diff --git a/src/node/handlers/mmp.rs b/src/node/handlers/mmp.rs index b2fd15a..c0647cd 100644 --- a/src/node/handlers/mmp.rs +++ b/src/node/handlers/mmp.rs @@ -158,18 +158,23 @@ impl Node { .unwrap_or(0); let flap_dampened = self.tree_state.set_parent(new_parent, new_seq, timestamp); self.tree_state.recompute_coords(); - if let Err(e) = self.tree_state.sign_declaration(&self.identity) { + // Clone identity once: sign_declaration borrows &mut tree_state while + // the identity() accessor borrows all of &self, so an owned copy avoids + // the split-borrow conflict on this infrequent parent-switch path. + let our_identity = self.identity().clone(); + if let Err(e) = self.tree_state.sign_declaration(&our_identity) { warn!(error = %e, "Failed to sign declaration after first-RTT parent eval"); - self.stats_mut() - .record_reject(RejectReason::Tree(TreeReject::OutboundSignFailed)); + self.metrics() + .tree + .record_reject(TreeReject::OutboundSignFailed); return; } // Surgical invalidation — see CoordCache::invalidate_via_node doc. self.coord_cache - .invalidate_via_node(self.identity.node_addr()); + .invalidate_via_node(our_identity.node_addr()); self.reset_discovery_backoff(); - self.stats_mut().tree.parent_switched += 1; - self.stats_mut().tree.parent_switches += 1; + self.metrics().tree.parent_switched.inc(); + self.metrics().tree.parent_switches.inc(); info!( new_parent = %self.peer_display_name(&new_parent), new_seq = new_seq, @@ -179,7 +184,7 @@ impl Node { "Parent switched after first RTT measurement" ); if flap_dampened { - self.stats_mut().tree.flap_dampened += 1; + self.metrics().tree.flap_dampened.inc(); warn!("Flap dampening engaged: excessive parent switches detected"); } self.send_tree_announce_to_all().await; @@ -187,18 +192,21 @@ impl Node { self.bloom_state.mark_all_updates_needed(all_peers); } else if !self.tree_state.is_root() && self.tree_state.should_be_root() { self.tree_state.become_root(); - if let Err(e) = self.tree_state.sign_declaration(&self.identity) { + // Clone identity once (see the parent-switch branch above for why). + let our_identity = self.identity().clone(); + if let Err(e) = self.tree_state.sign_declaration(&our_identity) { warn!(error = %e, "Failed to sign self-root declaration after first-RTT"); - self.stats_mut() - .record_reject(RejectReason::Tree(TreeReject::OutboundSignFailed)); + self.metrics() + .tree + .record_reject(TreeReject::OutboundSignFailed); return; } // Surgical invalidation — see CoordCache::invalidate_other_roots doc. self.coord_cache - .invalidate_other_roots(self.identity.node_addr()); + .invalidate_other_roots(our_identity.node_addr()); self.reset_discovery_backoff(); - self.stats_mut().tree.parent_switched += 1; - self.stats_mut().tree.parent_switches += 1; + self.metrics().tree.parent_switched.inc(); + self.metrics().tree.parent_switches.inc(); info!( new_root = %self.tree_state.root(), trigger = "first-rtt", @@ -532,8 +540,8 @@ impl Node { /// hasn't sent us a frame within the link dead timeout. pub(in crate::node) async fn check_link_heartbeats(&mut self) { let now = Instant::now(); - let heartbeat_interval = Duration::from_secs(self.config.node.heartbeat_interval_secs); - let dead_timeout = Duration::from_secs(self.config.node.link_dead_timeout_secs); + let heartbeat_interval = Duration::from_secs(self.config().node.heartbeat_interval_secs); + let dead_timeout = Duration::from_secs(self.config().node.link_dead_timeout_secs); let heartbeat_msg = [LinkMessageType::Heartbeat.to_byte()]; // Collect heartbeats to send and dead peers to remove @@ -576,7 +584,7 @@ impl Node { for addr in &dead_peers { warn!( peer = %self.peer_display_name(addr), - timeout_secs = self.config.node.link_dead_timeout_secs, + timeout_secs = self.config().node.link_dead_timeout_secs, "Removing peer: link dead timeout" ); self.remove_active_peer(addr); diff --git a/src/node/handlers/rekey.rs b/src/node/handlers/rekey.rs index 7b9d8de..2dfb6ce 100644 --- a/src/node/handlers/rekey.rs +++ b/src/node/handlers/rekey.rs @@ -37,12 +37,12 @@ impl Node { /// - If the drain window has expired, clean up the previous session /// - If the rekey timer/counter fires, initiate a new handshake pub(in crate::node) async fn check_rekey(&mut self) { - if !self.config.node.rekey.enabled { + if !self.config().node.rekey.enabled { return; } - let rekey_after_secs = self.config.node.rekey.after_secs; - let rekey_after_messages = self.config.node.rekey.after_messages; + let rekey_after_secs = self.config().node.rekey.after_secs; + let rekey_after_messages = self.config().node.rekey.after_messages; // Collect peers that need action (to avoid borrow conflicts) let mut peers_to_cutover: Vec = Vec::new(); @@ -194,7 +194,7 @@ impl Node { }; // Create IK initiator handshake directly (no PeerConnection) - let our_keypair = self.identity.keypair(); + let our_keypair = self.identity().keypair(); let mut hs = HandshakeState::new_initiator(our_keypair, peer_pubkey); hs.set_local_epoch(self.startup_epoch); @@ -236,7 +236,7 @@ impl Node { } // Store handshake state on the ActivePeer (not a separate PeerConnection) - let resend_interval = self.config.node.rate_limit.handshake_resend_interval_ms; + let resend_interval = self.config().node.rate_limit.handshake_resend_interval_ms; let now_ms = Self::now_ms(); if let Some(peer) = self.peers.get_mut(node_addr) { peer.set_rekey_state(hs, our_index, wire_msg1, now_ms + resend_interval); @@ -252,11 +252,11 @@ impl Node { /// Called from the tick loop. Uses the same resend interval and max /// resend count as initial handshakes. pub(in crate::node) async fn resend_pending_rekeys(&mut self, now_ms: u64) { - if !self.config.node.rekey.enabled { + if !self.config().node.rekey.enabled { return; } - let interval_ms = self.config.node.rate_limit.handshake_resend_interval_ms; + let interval_ms = self.config().node.rate_limit.handshake_resend_interval_ms; // Collect peers needing action let mut to_resend: Vec<(NodeAddr, Vec)> = Vec::new(); @@ -323,14 +323,14 @@ impl Node { /// is safe — an abandoned cycle never leaves a divergent unsafe /// state. pub(in crate::node) async fn resend_pending_session_msg3(&mut self, now_ms: u64) { - if !self.config.node.rekey.enabled || self.sessions.is_empty() { + if !self.config().node.rekey.enabled || self.sessions.is_empty() { return; } - let interval_ms = self.config.node.rate_limit.handshake_resend_interval_ms; - let backoff = self.config.node.rate_limit.handshake_resend_backoff; - let max_resends = self.config.node.rate_limit.handshake_max_resends; - let ttl = self.config.node.session.default_ttl; + let interval_ms = self.config().node.rate_limit.handshake_resend_interval_ms; + let backoff = self.config().node.rate_limit.handshake_resend_backoff; + let max_resends = self.config().node.rate_limit.handshake_max_resends; + let ttl = self.config().node.session.default_ttl; let my_addr = *self.node_addr(); // Collect rekey initiators whose msg3 retransmission is due. @@ -406,12 +406,12 @@ impl Node { /// `resend_pending_session_msg3`; its lifetime is tied to the /// responder receiving msg3, not to this initiator's cutover. pub(in crate::node) async fn check_session_rekey(&mut self) { - if !self.config.node.rekey.enabled { + if !self.config().node.rekey.enabled { return; } - let rekey_after_secs = self.config.node.rekey.after_secs; - let rekey_after_messages = self.config.node.rekey.after_messages; + let rekey_after_secs = self.config().node.rekey.after_secs; + let rekey_after_messages = self.config().node.rekey.after_messages; let now_ms = Self::now_ms(); let drain_ms = DRAIN_WINDOW_SECS * 1000; let dampening_ms = REKEY_DAMPENING_SECS * 1000; @@ -527,7 +527,7 @@ impl Node { let dest_pubkey = *entry.remote_pubkey(); // Create Noise XK initiator handshake - let our_keypair = self.identity.keypair(); + let our_keypair = self.identity().keypair(); let mut handshake = HandshakeState::new_xk_initiator(our_keypair, dest_pubkey); handshake.set_local_epoch(self.startup_epoch); @@ -552,7 +552,7 @@ impl Node { // Send through the mesh let my_addr = *self.node_addr(); let mut datagram = SessionDatagram::new(my_addr, *dest_addr, setup_payload) - .with_ttl(self.config.node.session.default_ttl); + .with_ttl(self.config().node.session.default_ttl); if let Err(e) = self.send_session_datagram(&mut datagram).await { debug!( diff --git a/src/node/handlers/rx_loop.rs b/src/node/handlers/rx_loop.rs index 941c687..ae88002 100644 --- a/src/node/handlers/rx_loop.rs +++ b/src/node/handlers/rx_loop.rs @@ -67,14 +67,14 @@ impl Node { }; let mut tick = - tokio::time::interval(Duration::from_secs(self.config.node.tick_interval_secs)); + tokio::time::interval(Duration::from_secs(self.config().node.tick_interval_secs)); // Set up control socket channel let (control_tx, mut control_rx) = tokio::sync::mpsc::channel::(32); - if self.config.node.control.enabled { - let config = self.config.node.control.clone(); + if self.config().node.control.enabled { + let config = self.config().node.control.clone(); let tx = control_tx.clone(); tokio::spawn(async move { match ControlSocket::bind(&config) { diff --git a/src/node/handlers/session.rs b/src/node/handlers/session.rs index a6f6c41..481c319 100644 --- a/src/node/handlers/session.rs +++ b/src/node/handlers/session.rs @@ -353,7 +353,7 @@ impl Node { Some(mut packet) => { if ce_flag { mark_ipv6_ecn_ce(&mut packet); - self.stats_mut().congestion.record_ce_received(); + self.metrics().congestion.ce_received.inc(); } if let Some(tun_tx) = &self.tun_tx { if let Err(e) = tun_tx.send(packet) { @@ -444,7 +444,7 @@ impl Node { if let Some(existing) = self.sessions.get(src_addr) { if existing.is_initiating() { // Simultaneous initiation: smaller NodeAddr wins as initiator - if self.identity.node_addr() < src_addr { + if self.identity().node_addr() < src_addr { // We win — drop their setup, they'll process ours debug!( src = %self.peer_display_name(src_addr), @@ -463,7 +463,7 @@ impl Node { debug!(src = %self.peer_display_name(src_addr), "Duplicate SessionSetup, resending SessionAck"); let my_addr = *self.node_addr(); let mut datagram = SessionDatagram::new(my_addr, *src_addr, payload.to_vec()) - .with_ttl(self.config.node.session.default_ttl); + .with_ttl(self.config().node.session.default_ttl); if let Err(e) = self.send_session_datagram(&mut datagram).await { debug!(error = %e, dest = %self.peer_display_name(src_addr), "Failed to resend SessionAck"); } @@ -474,7 +474,7 @@ impl Node { } else if existing.is_established() { // Rekey: if rekey enabled, treat as rekey for key rotation. // The existing established session remains active for traffic. - if self.config.node.rekey.enabled { + if self.config().node.rekey.enabled { let rekey_in_progress = existing.has_rekey_in_progress(); let has_pending = existing.pending_new_session().is_some(); @@ -482,7 +482,7 @@ impl Node { // simultaneously. Apply tie-breaker — smaller NodeAddr // wins as initiator (same as initial session setup). if rekey_in_progress { - if self.identity.node_addr() < src_addr { + if self.identity().node_addr() < src_addr { // We win as initiator — drop their msg1. debug!( src = %self.peer_display_name(src_addr), @@ -505,7 +505,7 @@ impl Node { ); return; } - let our_keypair = self.identity.keypair(); + let our_keypair = self.identity().keypair(); let mut handshake = HandshakeState::new_xk_responder(our_keypair); handshake.set_local_epoch(self.startup_epoch); @@ -529,7 +529,7 @@ impl Node { let ack_payload = ack.encode(); let my_addr = *self.node_addr(); let mut datagram = SessionDatagram::new(my_addr, *src_addr, ack_payload) - .with_ttl(self.config.node.session.default_ttl); + .with_ttl(self.config().node.session.default_ttl); if let Err(e) = self.send_session_datagram(&mut datagram).await { debug!(error = %e, dest = %self.peer_display_name(src_addr), "Failed to send rekey SessionAck"); @@ -555,7 +555,7 @@ impl Node { } // Create XK responder handshake and process msg1 - let our_keypair = self.identity.keypair(); + let our_keypair = self.identity().keypair(); let mut handshake = HandshakeState::new_xk_responder(our_keypair); handshake.set_local_epoch(self.startup_epoch); @@ -583,7 +583,7 @@ impl Node { let ack_payload = ack.encode(); let my_addr = *self.node_addr(); let mut datagram = SessionDatagram::new(my_addr, *src_addr, ack_payload.clone()) - .with_ttl(self.config.node.session.default_ttl); + .with_ttl(self.config().node.session.default_ttl); // Route the ack back to the initiator if let Err(e) = self.send_session_datagram(&mut datagram).await { @@ -594,9 +594,9 @@ impl Node { // Store session entry in AwaitingMsg3 state with ack payload for potential resend. // Use a dummy pubkey since we don't know the initiator's identity yet. // We use our own pubkey as placeholder; it will be replaced in handle_session_msg3. - let placeholder_pubkey = self.identity.keypair().public_key(); + let placeholder_pubkey = self.identity().keypair().public_key(); let now_ms = Self::now_ms(); - let resend_interval = self.config.node.rate_limit.handshake_resend_interval_ms; + let resend_interval = self.config().node.rate_limit.handshake_resend_interval_ms; let mut entry = SessionEntry::new( *src_addr, placeholder_pubkey, @@ -676,7 +676,7 @@ impl Node { let msg3_payload = msg3_wire.encode(); let my_addr = *self.node_addr(); let mut datagram = SessionDatagram::new(my_addr, *src_addr, msg3_payload.clone()) - .with_ttl(self.config.node.session.default_ttl); + .with_ttl(self.config().node.session.default_ttl); if let Err(e) = self.send_session_datagram(&mut datagram).await { debug!(error = %e, dest = %self.peer_display_name(src_addr), "Failed to send rekey SessionMsg3"); @@ -705,7 +705,7 @@ impl Node { // before the responder receives msg3; overlapping-epoch // decrypt keeps both directions safe meanwhile. let now_ms = Self::now_ms(); - let resend_interval = self.config.node.rate_limit.handshake_resend_interval_ms; + let resend_interval = self.config().node.rate_limit.handshake_resend_interval_ms; entry.set_pending_session(session); entry.set_rekey_completed_ms(now_ms); entry.set_rekey_msg3_payload(msg3_payload, now_ms + resend_interval); @@ -751,7 +751,7 @@ impl Node { let msg3_payload = msg3_wire.encode(); let my_addr = *self.node_addr(); let mut datagram = SessionDatagram::new(my_addr, *src_addr, msg3_payload) - .with_ttl(self.config.node.session.default_ttl); + .with_ttl(self.config().node.session.default_ttl); if let Err(e) = self.send_session_datagram(&mut datagram).await { debug!(error = %e, dest = %self.peer_display_name(src_addr), "Failed to send SessionMsg3"); @@ -769,9 +769,9 @@ impl Node { let now_ms = Self::now_ms(); entry.set_state(EndToEndState::Established(session)); - entry.set_coords_warmup_remaining(self.config.node.session.coords_warmup_packets); + entry.set_coords_warmup_remaining(self.config().node.session.coords_warmup_packets); entry.mark_established(now_ms); - entry.init_mmp(&self.config.node.session_mmp); + entry.init_mmp(&self.config().node.session_mmp); entry.clear_handshake_payload(); entry.touch(now_ms); self.sessions.insert(*src_addr, entry); @@ -905,9 +905,9 @@ impl Node { now_ms, false, ); - new_entry.set_coords_warmup_remaining(self.config.node.session.coords_warmup_packets); + new_entry.set_coords_warmup_remaining(self.config().node.session.coords_warmup_packets); new_entry.mark_established(now_ms); - new_entry.init_mmp(&self.config.node.session_mmp); + new_entry.init_mmp(&self.config().node.session_mmp); new_entry.touch(now_ms); self.sessions.insert(*src_addr, new_entry); @@ -1106,7 +1106,7 @@ impl Node { /// immediately (rate-limited), trigger discovery, and reset the /// warmup counter for subsequent data packets. async fn handle_coords_required(&mut self, inner: &[u8]) { - self.stats_mut().errors.coords_required += 1; + self.metrics().errors.coords_required.inc(); let msg = match CoordsRequired::decode(inner) { Ok(m) => m, @@ -1150,8 +1150,8 @@ impl Node { // Reset coords warmup counter so the next N packets also include // COORDS_PRESENT, re-warming transit caches along the path. + let n = self.config().node.session.coords_warmup_packets; if let Some(entry) = self.sessions.get_mut(&msg.dest_addr) { - let n = self.config.node.session.coords_warmup_packets; entry.set_coords_warmup_remaining(n); debug!( dest = %msg.dest_addr, @@ -1167,7 +1167,7 @@ impl Node { /// Send a standalone CoordsWarmup immediately (rate-limited), invalidate /// cached coordinates, trigger re-discovery, and reset the warmup counter. async fn handle_path_broken(&mut self, inner: &[u8]) { - self.stats_mut().errors.path_broken += 1; + self.metrics().errors.path_broken.inc(); let msg = match PathBroken::decode(inner) { Ok(m) => m, @@ -1216,8 +1216,8 @@ impl Node { // Reset coords warmup counter so the next N packets include // COORDS_PRESENT, re-warming transit caches along the new path. + let n = self.config().node.session.coords_warmup_packets; if let Some(entry) = self.sessions.get_mut(&msg.dest_addr) { - let n = self.config.node.session.coords_warmup_packets; entry.set_coords_warmup_remaining(n); debug!( dest = %msg.dest_addr, @@ -1233,7 +1233,7 @@ impl Node { /// next-hop transport MTU. Apply the reported bottleneck MTU to our /// PathMtuState for the affected session, causing an immediate decrease. pub(in crate::node) async fn handle_mtu_exceeded(&mut self, inner: &[u8]) { - self.stats_mut().errors.mtu_exceeded += 1; + self.metrics().errors.mtu_exceeded.inc(); let msg = match MtuExceeded::decode(inner) { Ok(m) => m, @@ -1331,7 +1331,7 @@ impl Node { } // Create Noise XK initiator handshake - let our_keypair = self.identity.keypair(); + let our_keypair = self.identity().keypair(); let mut handshake = HandshakeState::new_xk_initiator(our_keypair, dest_pubkey); handshake.set_local_epoch(self.startup_epoch); let msg1 = handshake @@ -1350,7 +1350,7 @@ impl Node { // Wrap in SessionDatagram let my_addr = *self.node_addr(); let mut datagram = SessionDatagram::new(my_addr, dest_addr, setup_payload.clone()) - .with_ttl(self.config.node.session.default_ttl); + .with_ttl(self.config().node.session.default_ttl); // Route toward destination self.send_session_datagram(&mut datagram).await?; @@ -1360,7 +1360,7 @@ impl Node { // Store session entry with handshake payload for potential resend let now_ms = Self::now_ms(); - let resend_interval = self.config.node.rate_limit.handshake_resend_interval_ms; + let resend_interval = self.config().node.rate_limit.handshake_resend_interval_ms; let mut entry = SessionEntry::new( dest_addr, dest_pubkey, @@ -1523,7 +1523,7 @@ impl Node { let my_addr = *self.node_addr(); let mut datagram = SessionDatagram::new(my_addr, *dest_addr, fsp_payload) - .with_ttl(self.config.node.session.default_ttl); + .with_ttl(self.config().node.session.default_ttl); self.send_session_datagram(&mut datagram).await?; @@ -1752,7 +1752,7 @@ impl Node { wire_buf.extend_from_slice(×tamp_ms.to_le_bytes()); // SessionDatagram-encoded layout (matches `SessionDatagram::encode`): wire_buf.push(LinkMessageType::SessionDatagram.to_byte()); - wire_buf.push(self.config.node.session.default_ttl); + wire_buf.push(self.config().node.session.default_ttl); wire_buf.extend_from_slice(&path_mtu.to_le_bytes()); wire_buf.extend_from_slice(self.node_addr().as_bytes()); wire_buf.extend_from_slice(dest_addr.as_bytes()); @@ -1776,7 +1776,7 @@ impl Node { .record_sent(fmp_counter, timestamp_ms, predicted_bytes); } } - self.stats_mut() + self.metrics() .forwarding .record_originated(link_plaintext_len + crate::noise::TAG_SIZE); @@ -1914,7 +1914,7 @@ impl Node { let my_addr = *self.node_addr(); let mut datagram = SessionDatagram::new(my_addr, *dest_addr, fsp_payload) - .with_ttl(self.config.node.session.default_ttl); + .with_ttl(self.config().node.session.default_ttl); self.send_session_datagram(&mut datagram).await?; @@ -1999,7 +1999,7 @@ impl Node { let my_addr = *self.node_addr(); let mut datagram = SessionDatagram::new(my_addr, *dest_addr, fsp_payload) - .with_ttl(self.config.node.session.default_ttl); + .with_ttl(self.config().node.session.default_ttl); self.send_session_datagram(&mut datagram).await?; @@ -2056,7 +2056,7 @@ impl Node { let encoded = datagram.encode(); self.send_encrypted_link_message(&next_hop_addr, &encoded) .await?; - self.stats_mut().forwarding.record_originated(encoded.len()); + self.metrics().forwarding.record_originated(encoded.len()); Ok(()) } @@ -2225,15 +2225,16 @@ impl Node { /// Queue a packet while waiting for session establishment. fn queue_pending_packet(&mut self, dest_addr: NodeAddr, packet: Vec) { // Reject if we already have too many pending destinations - let max_dests = self.config.node.session.pending_max_destinations; + let max_dests = self.config().node.session.pending_max_destinations; if !self.pending_tun_packets.contains_key(&dest_addr) && self.pending_tun_packets.len() >= max_dests { return; } + let per_dest = self.config().node.session.pending_packets_per_dest; let queue = self.pending_tun_packets.entry(dest_addr).or_default(); - if queue.len() >= self.config.node.session.pending_packets_per_dest { + if queue.len() >= per_dest { queue.pop_front(); // Drop oldest } queue.push_back(packet); diff --git a/src/node/handlers/timeout.rs b/src/node/handlers/timeout.rs index e87a3db..e59c4e8 100644 --- a/src/node/handlers/timeout.rs +++ b/src/node/handlers/timeout.rs @@ -17,7 +17,7 @@ impl Node { } let now_ms = Self::now_ms(); - let timeout_ms = self.config.node.rate_limit.handshake_timeout_secs * 1000; + let timeout_ms = self.config().node.rate_limit.handshake_timeout_secs * 1000; let stale: Vec = self .connections @@ -93,9 +93,9 @@ impl Node { return; } - let max_resends = self.config.node.rate_limit.handshake_max_resends; - let interval_ms = self.config.node.rate_limit.handshake_resend_interval_ms; - let backoff = self.config.node.rate_limit.handshake_resend_backoff; + let max_resends = self.config().node.rate_limit.handshake_max_resends; + let interval_ms = self.config().node.rate_limit.handshake_resend_interval_ms; + let backoff = self.config().node.rate_limit.handshake_resend_backoff; // Collect resend candidates: outbound, in SentMsg1, with stored msg1, // under max resends, and past the scheduled time. @@ -165,11 +165,11 @@ impl Node { return; } - let timeout_ms = self.config.node.rate_limit.handshake_timeout_secs * 1000; - let max_resends = self.config.node.rate_limit.handshake_max_resends; - let interval_ms = self.config.node.rate_limit.handshake_resend_interval_ms; - let backoff = self.config.node.rate_limit.handshake_resend_backoff; - let ttl = self.config.node.session.default_ttl; + let timeout_ms = self.config().node.rate_limit.handshake_timeout_secs * 1000; + let max_resends = self.config().node.rate_limit.handshake_max_resends; + let interval_ms = self.config().node.rate_limit.handshake_resend_interval_ms; + let backoff = self.config().node.rate_limit.handshake_resend_backoff; + let ttl = self.config().node.session.default_ttl; // First pass: find timed-out sessions to remove let timed_out: Vec = self @@ -237,7 +237,7 @@ impl Node { /// Only targets sessions in the Established state. Initiating/AwaitingMsg3 /// sessions are handled by the handshake timeout. pub(in crate::node) fn purge_idle_sessions(&mut self, now_ms: u64) { - let timeout_ms = self.config.node.session.idle_timeout_secs * 1000; + let timeout_ms = self.config().node.session.idle_timeout_secs * 1000; if timeout_ms == 0 { return; // disabled } diff --git a/src/node/lifecycle.rs b/src/node/lifecycle.rs index 771cb32..a96273d 100644 --- a/src/node/lifecycle.rs +++ b/src/node/lifecycle.rs @@ -55,6 +55,10 @@ impl Node { new_by_addr.insert(*identity.node_addr(), peer); } + // Read the current peer set directly from the field: update_peers is the + // config-source mutation owner (it writes self.config.peers below and then + // rebuilds the context), so it manages config.peers directly rather than + // through the context accessor — same rationale as the write at line ~124. let current_by_addr: HashMap = self .config .peers() @@ -121,6 +125,7 @@ impl Node { .collect(); self.config.peers = new_by_addr.into_values().collect(); + self.rebuild_context(); for peer_config in added_configs { outcome.added += 1; @@ -171,11 +176,12 @@ impl Node { } else { match self.initiate_peer_connection(&peer_config).await { Ok(()) => { + let handshake_timeout_secs = + self.config().node.rate_limit.handshake_timeout_secs; if let Some(state) = self.retry_pending.get_mut(&node_addr) { state.peer_config = peer_config; - state.retry_after_ms = Self::now_ms().saturating_add( - self.config.node.rate_limit.handshake_timeout_secs * 1000, - ); + state.retry_after_ms = + Self::now_ms().saturating_add(handshake_timeout_secs * 1000); } } Err(err) => { @@ -204,7 +210,7 @@ impl Node { // initiation) immediately on startup — without waiting for the link-layer // handshake to complete first. let peer_identities: Vec<(PeerIdentity, Option)> = self - .config + .config() .peers() .iter() .filter_map(|pc| { @@ -224,7 +230,7 @@ impl Node { } // Collect peer configs to avoid borrow conflicts - let peer_configs: Vec<_> = self.config.auto_connect_peers().cloned().collect(); + let peer_configs: Vec<_> = self.config().auto_connect_peers().cloned().collect(); if peer_configs.is_empty() { debug!("No static peers configured"); @@ -397,7 +403,7 @@ impl Node { transport_id, remote_addr.clone(), LinkDirection::Outbound, - Duration::from_millis(self.config.node.base_rtt_ms), + Duration::from_millis(self.config().node.base_rtt_ms), ) } else { Link::connectionless( @@ -405,7 +411,7 @@ impl Node { transport_id, remote_addr.clone(), LinkDirection::Outbound, - Duration::from_millis(self.config.node.base_rtt_ms), + Duration::from_millis(self.config().node.base_rtt_ms), ) }; @@ -479,7 +485,7 @@ impl Node { }; // Start the Noise handshake and get message 1 - let our_keypair = self.identity.keypair(); + let our_keypair = self.identity().keypair(); let noise_msg1 = match connection.start_handshake(our_keypair, self.startup_epoch, current_time_ms) { Ok(msg) => msg, @@ -510,7 +516,7 @@ impl Node { ); // Store msg1 for resend and schedule first resend - let resend_interval = self.config.node.rate_limit.handshake_resend_interval_ms; + let resend_interval = self.config().node.rate_limit.handshake_resend_interval_ms; connection.set_handshake_msg1(wire_msg1.clone(), current_time_ms + resend_interval); // Track in pending_outbound for msg2 dispatch @@ -581,7 +587,7 @@ impl Node { let node_addr = *identity.node_addr(); // Skip self - if node_addr == *self.identity.node_addr() { + if node_addr == *self.identity().node_addr() { continue; } @@ -720,7 +726,7 @@ impl Node { // semantics, and its existing cross-connection // logic in handle_msg1 reconciles when the winner's // fresh msg1 arrives over the adopted socket. - let our_addr = self.identity.node_addr(); + let our_addr = self.identity().node_addr(); if our_addr >= &peer_addr { debug!( peer_npub = %peer_npub, @@ -868,14 +874,14 @@ impl Node { /// extracts a scope from the Nostr app tag used by default scoped /// discovery. pub(super) fn lan_discovery_scope(&self) -> Option { - if let Some(scope) = self.config.node.discovery.lan.scope.as_deref() { + if let Some(scope) = self.config().node.discovery.lan.scope.as_deref() { let scope = scope.trim(); if !scope.is_empty() { return Some(scope.to_string()); } } - let app = self.config.node.discovery.nostr.app.trim(); + let app = self.config().node.discovery.nostr.app.trim(); if app.is_empty() { return None; } @@ -1051,7 +1057,7 @@ impl Node { self.state = NodeState::Starting; // Create packet channel for transport -> Node communication - let packet_buffer_size = self.config.node.buffers.packet_channel; + let packet_buffer_size = self.config().node.buffers.packet_channel; let (packet_tx, packet_rx) = packet_channel(packet_buffer_size); self.packet_tx = Some(packet_tx.clone()); self.packet_rx = Some(packet_rx); @@ -1128,8 +1134,8 @@ impl Node { } } - if self.config.node.discovery.nostr.enabled { - match NostrDiscovery::start(&self.identity, self.config.node.discovery.nostr.clone()) + if self.config().node.discovery.nostr.enabled { + match NostrDiscovery::start(self.identity(), self.config().node.discovery.nostr.clone()) .await { Ok(runtime) => { @@ -1149,7 +1155,7 @@ impl Node { // mDNS / DNS-SD LAN discovery. Independent of Nostr — runs even // when Nostr is disabled, since it gives us sub-second pairing // on the same link without any relay or NAT-traversal roundtrip. - if self.config.node.discovery.lan.enabled { + if self.config().node.discovery.lan.enabled { // Advertise the port of a non-bootstrap operational UDP transport. // Bootstrap transports must be excluded (they are not the node's // listening data-plane socket), and a stable selector (lowest @@ -1170,10 +1176,10 @@ impl Node { .unwrap_or(0); let scope = self.lan_discovery_scope(); match crate::discovery::lan::LanDiscovery::start( - &self.identity, + self.identity(), scope, advertised_udp_port, - self.config.node.discovery.lan.clone(), + self.config().node.discovery.lan.clone(), ) .await { @@ -1192,9 +1198,9 @@ impl Node { self.initiate_peer_connections().await; // Initialize TUN interface last, after transports and peers are ready - if self.config.tun.enabled { - let address = *self.identity.address(); - match TunDevice::create(&self.config.tun, address).await { + if self.config().tun.enabled { + let address = *self.identity().address(); + match TunDevice::create(&self.config().tun, address).await { Ok(device) => { let mtu = device.mtu(); let name = device.name().to_string(); @@ -1241,7 +1247,7 @@ impl Node { let reader_tun_tx = tun_tx.clone(); // Create outbound channel for TUN reader → Node - let tun_channel_size = self.config.node.buffers.tun_channel; + let tun_channel_size = self.config().node.buffers.tun_channel; let (outbound_tx, outbound_rx) = tokio::sync::mpsc::channel(tun_channel_size); // Spawn reader thread @@ -1307,19 +1313,19 @@ impl Node { // `IPV6_V6ONLY=0` is set explicitly so IPv4 clients on // 127.0.0.1 still reach us regardless of kernel sysctl // defaults — but only when bind is on a wildcard / IPv6 path. - if self.config.dns.enabled { - let addr_str = self.config.dns.bind_addr(); + if self.config().dns.enabled { + let addr_str = self.config().dns.bind_addr(); match addr_str.parse::() { Ok(ip) => { - let bind = std::net::SocketAddr::new(ip, self.config.dns.port()); + let bind = std::net::SocketAddr::new(ip, self.config().dns.port()); match Self::bind_dns_socket(bind) { Ok(socket) => { - let dns_channel_size = self.config.node.buffers.dns_channel; + let dns_channel_size = self.config().node.buffers.dns_channel; let (identity_tx, identity_rx) = tokio::sync::mpsc::channel(dns_channel_size); - let dns_ttl = self.config.dns.ttl(); + let dns_ttl = self.config().dns.ttl(); let base_hosts = crate::upper::hosts::HostMap::from_peer_configs( - self.config.peers(), + self.config().peers(), ); let hosts_path = std::path::PathBuf::from(crate::upper::hosts::DEFAULT_HOSTS_PATH); @@ -1332,7 +1338,7 @@ impl Node { // When TUN isn't enabled or the name can't be // resolved, `None` disables the filter (there // is no mesh surface to defend anyway). - let mesh_ifindex = Self::lookup_mesh_ifindex(self.config.tun.name()); + let mesh_ifindex = Self::lookup_mesh_ifindex(self.config().tun.name()); info!( bind = %bind, hosts = reloader.hosts().len(), @@ -1599,9 +1605,9 @@ impl Node { peer_config: &PeerConfig, existing: &[PeerAddress], ) -> Vec { - if !self.config.node.discovery.nostr.enabled + if !self.config().node.discovery.nostr.enabled || !peer_config.via_nostr - || self.config.node.discovery.nostr.policy + || self.config().node.discovery.nostr.policy == crate::config::NostrDiscoveryPolicy::Disabled { return Vec::new(); @@ -1825,14 +1831,15 @@ impl Node { max_age_secs: Option, caller: &'static str, ) { - if !self.config.node.discovery.nostr.enabled - || self.config.node.discovery.nostr.policy != crate::config::NostrDiscoveryPolicy::Open + if !self.config().node.discovery.nostr.enabled + || self.config().node.discovery.nostr.policy + != crate::config::NostrDiscoveryPolicy::Open { return; } let configured_npubs = self - .config + .config() .peers() .iter() .map(|peer| peer.npub.clone()) @@ -1901,7 +1908,7 @@ impl Node { } }; let node_addr = *peer_identity.node_addr(); - if node_addr == *self.identity.node_addr() { + if node_addr == *self.identity().node_addr() { skipped_self = skipped_self.saturating_add(1); continue; } @@ -2023,8 +2030,9 @@ impl Node { if self.startup_open_discovery_sweep_done { return; } - if !self.config.node.discovery.nostr.enabled - || self.config.node.discovery.nostr.policy != crate::config::NostrDiscoveryPolicy::Open + if !self.config().node.discovery.nostr.enabled + || self.config().node.discovery.nostr.policy + != crate::config::NostrDiscoveryPolicy::Open { // Mark done so we don't keep re-checking on every tick. self.startup_open_discovery_sweep_done = true; @@ -2035,7 +2043,7 @@ impl Node { }; let now_ms = Self::now_ms(); let delay_ms = self - .config + .config() .node .discovery .nostr @@ -2045,7 +2053,12 @@ impl Node { return; } - let max_age_secs = self.config.node.discovery.nostr.startup_sweep_max_age_secs; + let max_age_secs = self + .config() + .node + .discovery + .nostr + .startup_sweep_max_age_secs; self.run_open_discovery_sweep(bootstrap, Some(max_age_secs), "startup") .await; self.startup_open_discovery_sweep_done = true; @@ -2134,7 +2147,7 @@ impl Node { .count(); let cap_remaining = self - .config + .config() .node .discovery .nostr @@ -2146,7 +2159,7 @@ impl Node { fn open_discovery_retry_expires_at_ms(&self, now_ms: u64) -> u64 { now_ms.saturating_add( - self.config + self.config() .node .discovery .nostr @@ -2160,7 +2173,7 @@ impl Node { &self, bootstrap: &std::sync::Arc, ) -> Option { - if !self.config.node.discovery.nostr.enabled { + if !self.config().node.discovery.nostr.enabled { return None; } @@ -2300,9 +2313,10 @@ impl Node { identifier: ADVERT_IDENTIFIER.to_string(), version: ADVERT_VERSION, endpoints, - signal_relays: has_udp_nat.then(|| self.config.node.discovery.nostr.dm_relays.clone()), + signal_relays: has_udp_nat + .then(|| self.config().node.discovery.nostr.dm_relays.clone()), stun_servers: has_udp_nat - .then(|| self.config.node.discovery.nostr.stun_servers.clone()), + .then(|| self.config().node.discovery.nostr.stun_servers.clone()), }) } @@ -2315,7 +2329,7 @@ impl Node { } fn lookup_udp_config(&self, transport_name: Option<&str>) -> Option<&crate::config::UdpConfig> { - match (&self.config.transports.udp, transport_name) { + match (&self.config().transports.udp, transport_name) { (crate::config::TransportInstances::Single(cfg), None) => Some(cfg), (crate::config::TransportInstances::Named(configs), Some(name)) => configs.get(name), _ => None, @@ -2323,7 +2337,7 @@ impl Node { } fn lookup_tcp_config(&self, transport_name: Option<&str>) -> Option<&crate::config::TcpConfig> { - match (&self.config.transports.tcp, transport_name) { + match (&self.config().transports.tcp, transport_name) { (crate::config::TransportInstances::Single(cfg), None) => Some(cfg), (crate::config::TransportInstances::Named(configs), Some(name)) => configs.get(name), _ => None, @@ -2331,7 +2345,7 @@ impl Node { } fn lookup_tor_config(&self, transport_name: Option<&str>) -> Option<&crate::config::TorConfig> { - match (&self.config.transports.tor, transport_name) { + match (&self.config().transports.tor, transport_name) { (crate::config::TransportInstances::Single(cfg), None) => Some(cfg), (crate::config::TransportInstances::Named(configs), Some(name)) => configs.get(name), _ => None, @@ -2465,7 +2479,7 @@ impl Node { return false; }; let stale_after_ms = self - .config + .config() .node .heartbeat_interval_secs .saturating_mul(1000) diff --git a/src/node/metrics.rs b/src/node/metrics.rs new file mode 100644 index 0000000..b58de38 --- /dev/null +++ b/src/node/metrics.rs @@ -0,0 +1,476 @@ +//! Lock-free metric counters backed by atomics. +//! +//! Mirrors the `NodeStats` counter surface (`stats.rs`) but stores each +//! counter in an `AtomicU64`, so it can be bumped through `&self` and, in +//! a later step, sampled without dispatching through the rx_loop task. The +//! hottest counters are cache-line padded to avoid false sharing once +//! reads move off-thread. +//! +//! The forwarding, discovery, tree, bloom, congestion, and error families +//! live here exclusively and are both written and served from the registry. +//! The remaining families (session, handshake, mmp, transport) stay on +//! `NodeStats`. + +use std::sync::atomic::{AtomicU64, Ordering}; + +use crate::node::reject::{BloomReject, DiscoveryReject, ForwardingReject, TreeReject}; +use crate::node::stats::{ + BloomStatsSnapshot, CongestionStatsSnapshot, DiscoveryStatsSnapshot, ErrorSignalStatsSnapshot, + ForwardingStatsSnapshot, TreeStatsSnapshot, +}; + +/// An atomic counter. +#[derive(Default)] +pub struct Counter(AtomicU64); + +impl Counter { + #[inline] + pub fn inc(&self) { + self.0.fetch_add(1, Ordering::Relaxed); + } + + #[inline] + pub fn add(&self, n: u64) { + self.0.fetch_add(n, Ordering::Relaxed); + } + + #[inline] + pub fn get(&self) -> u64 { + self.0.load(Ordering::Relaxed) + } +} + +/// Cache-line padding wrapper for the hottest counters. +/// +/// Padding keeps a hot counter off shared cache lines so that concurrent +/// reads (introduced when metric sampling moves off the rx_loop task) do +/// not false-share with the writer. With a single writer today the padding +/// is forward-looking insurance. Derefs to the inner counter so the call +/// sites are identical to an unpadded one. +#[repr(align(64))] +#[derive(Default)] +pub struct Padded(pub T); + +impl std::ops::Deref for Padded { + type Target = T; + + #[inline] + fn deref(&self) -> &T { + &self.0 + } +} + +/// Forwarding metric counters. +#[derive(Default)] +pub struct ForwardingMetrics { + pub received_packets: Padded, + pub received_bytes: Counter, + pub decode_error_packets: Counter, + pub decode_error_bytes: Counter, + pub ttl_exhausted_packets: Counter, + pub ttl_exhausted_bytes: Counter, + pub delivered_packets: Counter, + pub delivered_bytes: Counter, + pub forwarded_packets: Counter, + pub forwarded_bytes: Counter, + pub drop_no_route_packets: Counter, + pub drop_no_route_bytes: Counter, + pub drop_mtu_exceeded_packets: Counter, + pub drop_mtu_exceeded_bytes: Counter, + pub drop_send_error_packets: Counter, + pub drop_send_error_bytes: Counter, + pub originated_packets: Counter, + pub originated_bytes: Counter, +} + +impl ForwardingMetrics { + /// Record a received packet of `bytes` payload (packets and bytes). + #[inline] + pub fn record_received(&self, bytes: usize) { + self.received_packets.inc(); + self.received_bytes.add(bytes as u64); + } + + /// Record a locally-delivered packet of `bytes` payload. + #[inline] + pub fn record_delivered(&self, bytes: usize) { + self.delivered_packets.inc(); + self.delivered_bytes.add(bytes as u64); + } + + /// Record a forwarded (transit) packet of `bytes` payload. + #[inline] + pub fn record_forwarded(&self, bytes: usize) { + self.forwarded_packets.inc(); + self.forwarded_bytes.add(bytes as u64); + } + + /// Record a locally-originated packet of `bytes` payload. + #[inline] + pub fn record_originated(&self, bytes: usize) { + self.originated_packets.inc(); + self.originated_bytes.add(bytes as u64); + } + + /// Mirror of `ForwardingStats::record_reject_bytes`: route a typed + /// forwarding rejection of `bytes` payload to its packet and byte + /// counters. + #[inline] + pub fn record_reject_bytes(&self, reason: ForwardingReject, bytes: usize) { + let bytes = bytes as u64; + match reason { + ForwardingReject::DecodeError => { + self.decode_error_packets.inc(); + self.decode_error_bytes.add(bytes); + } + ForwardingReject::TtlExhausted => { + self.ttl_exhausted_packets.inc(); + self.ttl_exhausted_bytes.add(bytes); + } + ForwardingReject::NoRoute => { + self.drop_no_route_packets.inc(); + self.drop_no_route_bytes.add(bytes); + } + ForwardingReject::MtuExceeded => { + self.drop_mtu_exceeded_packets.inc(); + self.drop_mtu_exceeded_bytes.add(bytes); + } + ForwardingReject::SendError => { + self.drop_send_error_packets.inc(); + self.drop_send_error_bytes.add(bytes); + } + } + } + + /// Sample every counter into a serializable snapshot. + pub fn snapshot(&self) -> ForwardingStatsSnapshot { + ForwardingStatsSnapshot { + received_packets: self.received_packets.get(), + received_bytes: self.received_bytes.get(), + decode_error_packets: self.decode_error_packets.get(), + decode_error_bytes: self.decode_error_bytes.get(), + ttl_exhausted_packets: self.ttl_exhausted_packets.get(), + ttl_exhausted_bytes: self.ttl_exhausted_bytes.get(), + delivered_packets: self.delivered_packets.get(), + delivered_bytes: self.delivered_bytes.get(), + forwarded_packets: self.forwarded_packets.get(), + forwarded_bytes: self.forwarded_bytes.get(), + drop_no_route_packets: self.drop_no_route_packets.get(), + drop_no_route_bytes: self.drop_no_route_bytes.get(), + drop_mtu_exceeded_packets: self.drop_mtu_exceeded_packets.get(), + drop_mtu_exceeded_bytes: self.drop_mtu_exceeded_bytes.get(), + drop_send_error_packets: self.drop_send_error_packets.get(), + drop_send_error_bytes: self.drop_send_error_bytes.get(), + originated_packets: self.originated_packets.get(), + originated_bytes: self.originated_bytes.get(), + } + } +} + +/// Discovery metric counters. +#[derive(Default)] +pub struct DiscoveryMetrics { + pub req_received: Padded, + pub req_decode_error: Counter, + pub req_duplicate: Counter, + pub req_dedup_cache_full: Counter, + pub req_target_is_us: Counter, + pub req_forwarded: Counter, + pub req_ttl_exhausted: Counter, + pub req_initiated: Counter, + pub req_deduplicated: Counter, + pub req_backoff_suppressed: Counter, + pub req_forward_rate_limited: Counter, + pub req_bloom_miss: Counter, + pub req_no_tree_peer: Counter, + pub req_fallback_forwarded: Counter, + pub resp_received: Counter, + pub resp_decode_error: Counter, + pub resp_forwarded: Counter, + pub resp_identity_miss: Counter, + pub resp_proof_failed: Counter, + pub resp_no_route: Counter, + pub resp_accepted: Counter, + pub resp_timed_out: Counter, +} + +impl DiscoveryMetrics { + /// Mirror of `DiscoveryStats::record_reject`: route a typed discovery + /// rejection to its counter. + #[inline] + pub fn record_reject(&self, reason: DiscoveryReject) { + match reason { + DiscoveryReject::ReqDecodeError => self.req_decode_error.inc(), + DiscoveryReject::ReqDuplicate => self.req_duplicate.inc(), + DiscoveryReject::ReqDedupCacheFull => self.req_dedup_cache_full.inc(), + DiscoveryReject::ReqTtlExhausted => self.req_ttl_exhausted.inc(), + DiscoveryReject::RespDecodeError => self.resp_decode_error.inc(), + DiscoveryReject::RespIdentityMiss => self.resp_identity_miss.inc(), + DiscoveryReject::RespProofFailed => self.resp_proof_failed.inc(), + DiscoveryReject::RespNoRoute => self.resp_no_route.inc(), + } + } + + /// Sample every counter into a serializable snapshot. + pub fn snapshot(&self) -> DiscoveryStatsSnapshot { + DiscoveryStatsSnapshot { + req_received: self.req_received.get(), + req_decode_error: self.req_decode_error.get(), + req_duplicate: self.req_duplicate.get(), + req_dedup_cache_full: self.req_dedup_cache_full.get(), + req_target_is_us: self.req_target_is_us.get(), + req_forwarded: self.req_forwarded.get(), + req_ttl_exhausted: self.req_ttl_exhausted.get(), + req_initiated: self.req_initiated.get(), + req_deduplicated: self.req_deduplicated.get(), + req_backoff_suppressed: self.req_backoff_suppressed.get(), + req_forward_rate_limited: self.req_forward_rate_limited.get(), + req_bloom_miss: self.req_bloom_miss.get(), + req_no_tree_peer: self.req_no_tree_peer.get(), + req_fallback_forwarded: self.req_fallback_forwarded.get(), + resp_received: self.resp_received.get(), + resp_decode_error: self.resp_decode_error.get(), + resp_forwarded: self.resp_forwarded.get(), + resp_identity_miss: self.resp_identity_miss.get(), + resp_proof_failed: self.resp_proof_failed.get(), + resp_no_route: self.resp_no_route.get(), + resp_accepted: self.resp_accepted.get(), + resp_timed_out: self.resp_timed_out.get(), + } + } +} + +/// Spanning-tree metric counters. +#[derive(Default)] +pub struct TreeMetrics { + pub received: Counter, + pub decode_error: Counter, + pub unknown_peer: Counter, + pub addr_mismatch: Counter, + pub sig_failed: Counter, + pub stale: Counter, + pub ancestry_invalid: Counter, + pub accepted: Counter, + pub parent_switched: Counter, + pub loop_detected: Counter, + pub ancestry_changed: Counter, + pub sent: Counter, + pub rate_limited: Counter, + pub send_failed: Counter, + pub outbound_sign_failed: Counter, + pub parent_switches: Counter, + pub parent_losses: Counter, + pub flap_dampened: Counter, +} + +impl TreeMetrics { + /// Mirror of `TreeStats::record_reject`: route a typed tree + /// rejection to its counter. + #[inline] + pub fn record_reject(&self, reason: TreeReject) { + match reason { + TreeReject::AncestryInvalid => self.ancestry_invalid.inc(), + TreeReject::OutboundSignFailed => self.outbound_sign_failed.inc(), + } + } + + /// Sample every counter into a serializable snapshot. + pub fn snapshot(&self) -> TreeStatsSnapshot { + TreeStatsSnapshot { + received: self.received.get(), + decode_error: self.decode_error.get(), + unknown_peer: self.unknown_peer.get(), + addr_mismatch: self.addr_mismatch.get(), + sig_failed: self.sig_failed.get(), + stale: self.stale.get(), + ancestry_invalid: self.ancestry_invalid.get(), + accepted: self.accepted.get(), + parent_switched: self.parent_switched.get(), + loop_detected: self.loop_detected.get(), + ancestry_changed: self.ancestry_changed.get(), + sent: self.sent.get(), + rate_limited: self.rate_limited.get(), + send_failed: self.send_failed.get(), + outbound_sign_failed: self.outbound_sign_failed.get(), + parent_switches: self.parent_switches.get(), + parent_losses: self.parent_losses.get(), + flap_dampened: self.flap_dampened.get(), + } + } +} + +/// Bloom-filter metric counters. +#[derive(Default)] +pub struct BloomMetrics { + pub received: Counter, + pub decode_error: Counter, + pub invalid: Counter, + pub non_v1: Counter, + pub unknown_peer: Counter, + pub stale: Counter, + pub fill_exceeded: Counter, + pub accepted: Counter, + pub sent: Counter, + pub debounce_suppressed: Counter, + pub send_failed: Counter, +} + +impl BloomMetrics { + /// Mirror of `BloomStats::record_reject`: route a typed bloom + /// rejection to its counter. + #[inline] + pub fn record_reject(&self, reason: BloomReject) { + match reason { + BloomReject::DecodeError => self.decode_error.inc(), + BloomReject::Invalid => self.invalid.inc(), + BloomReject::NonV1 => self.non_v1.inc(), + BloomReject::UnknownPeer => self.unknown_peer.inc(), + BloomReject::Stale => self.stale.inc(), + BloomReject::FillExceeded => self.fill_exceeded.inc(), + } + } + + /// Sample every counter into a serializable snapshot. + pub fn snapshot(&self) -> BloomStatsSnapshot { + BloomStatsSnapshot { + received: self.received.get(), + decode_error: self.decode_error.get(), + invalid: self.invalid.get(), + non_v1: self.non_v1.get(), + unknown_peer: self.unknown_peer.get(), + stale: self.stale.get(), + fill_exceeded: self.fill_exceeded.get(), + accepted: self.accepted.get(), + sent: self.sent.get(), + debounce_suppressed: self.debounce_suppressed.get(), + send_failed: self.send_failed.get(), + } + } +} + +/// Congestion metric counters. +#[derive(Default)] +pub struct CongestionMetrics { + pub ce_forwarded: Counter, + pub ce_received: Counter, + pub congestion_detected: Counter, + pub kernel_drop_events: Counter, +} + +impl CongestionMetrics { + /// Sample every counter into a serializable snapshot. + pub fn snapshot(&self) -> CongestionStatsSnapshot { + CongestionStatsSnapshot { + ce_forwarded: self.ce_forwarded.get(), + ce_received: self.ce_received.get(), + congestion_detected: self.congestion_detected.get(), + kernel_drop_events: self.kernel_drop_events.get(), + } + } +} + +/// Error-signal metric counters. +#[derive(Default)] +pub struct ErrorMetrics { + pub coords_required: Counter, + pub path_broken: Counter, + pub mtu_exceeded: Counter, +} + +impl ErrorMetrics { + /// Sample every counter into a serializable snapshot. + pub fn snapshot(&self) -> ErrorSignalStatsSnapshot { + ErrorSignalStatsSnapshot { + coords_required: self.coords_required.get(), + path_broken: self.path_broken.get(), + mtu_exceeded: self.mtu_exceeded.get(), + } + } +} + +/// Atomic counter registry shared across the node via `Arc`. +/// +/// Sole storage for the forwarding, discovery, tree, bloom, congestion, +/// and error counter families; these were migrated off `NodeStats`, which +/// now holds only the session, handshake, mmp, and transport families. +#[derive(Default)] +pub struct MetricsRegistry { + pub forwarding: ForwardingMetrics, + pub discovery: DiscoveryMetrics, + pub tree: TreeMetrics, + pub bloom: BloomMetrics, + pub congestion: CongestionMetrics, + pub errors: ErrorMetrics, +} + +impl MetricsRegistry { + pub fn new() -> Self { + Self::default() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn forwarding_received_tracks_packets_and_bytes() { + let m = ForwardingMetrics::default(); + m.record_received(100); + m.record_received(40); + assert_eq!(m.received_packets.get(), 2); + assert_eq!(m.received_bytes.get(), 140); + } + + #[test] + fn discovery_record_reject_routes_to_field() { + let m = DiscoveryMetrics::default(); + m.record_reject(DiscoveryReject::ReqDuplicate); + m.record_reject(DiscoveryReject::ReqDuplicate); + m.record_reject(DiscoveryReject::RespNoRoute); + assert_eq!(m.req_duplicate.get(), 2); + assert_eq!(m.resp_no_route.get(), 1); + assert_eq!(m.req_decode_error.get(), 0); + } + + #[test] + fn discovery_direct_counters_increment() { + let m = DiscoveryMetrics::default(); + m.req_received.inc(); + m.req_forwarded.inc(); + m.req_forwarded.inc(); + assert_eq!(m.req_received.get(), 1); + assert_eq!(m.req_forwarded.get(), 2); + } + + #[test] + fn tree_record_reject_routes_to_field() { + let m = TreeMetrics::default(); + m.record_reject(TreeReject::OutboundSignFailed); + m.record_reject(TreeReject::OutboundSignFailed); + m.record_reject(TreeReject::AncestryInvalid); + assert_eq!(m.outbound_sign_failed.get(), 2); + assert_eq!(m.ancestry_invalid.get(), 1); + } + + #[test] + fn bloom_record_reject_routes_to_field() { + let m = BloomMetrics::default(); + m.record_reject(BloomReject::Stale); + m.record_reject(BloomReject::Stale); + m.record_reject(BloomReject::DecodeError); + assert_eq!(m.stale.get(), 2); + assert_eq!(m.decode_error.get(), 1); + assert_eq!(m.invalid.get(), 0); + } + + #[test] + fn registry_subcounters_are_independent() { + let r = MetricsRegistry::new(); + r.forwarding.record_received(10); + r.discovery.req_received.inc(); + assert_eq!(r.forwarding.received_packets.get(), 1); + assert_eq!(r.forwarding.received_bytes.get(), 10); + assert_eq!(r.discovery.req_received.get(), 1); + } +} diff --git a/src/node/mod.rs b/src/node/mod.rs index 5776b14..baba892 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -6,6 +6,7 @@ mod acl; mod bloom; +mod context; #[cfg(unix)] pub(crate) mod decrypt_worker; mod discovery_rate_limit; @@ -13,6 +14,7 @@ mod discovery_rate_limit; pub(crate) mod encrypt_worker; mod handlers; mod lifecycle; +pub(crate) mod metrics; mod rate_limit; pub(crate) mod reject; mod reloadable; @@ -308,6 +310,13 @@ pub struct Node { /// Loaded configuration. config: Config, + /// Shared immutable context bundle. A parallel, authoritative copy of the + /// effectively-immutable fields (config/identity/startup_epoch/started_at/ + /// is_leaf_only/max_*), kept in lockstep with the `Node` fields via + /// `rebuild_context`. Readers migrate onto it in later sub-PRs; the + /// duplicated `Node` fields are removed once the last reader has moved. + context: Arc, + // === State === /// Node operational state. state: NodeState, @@ -399,6 +408,10 @@ pub struct Node { /// Routing, forwarding, discovery, and error signal counters. stats: stats::NodeStats, + /// Lock-free atomic metric counters. Shadows `stats` during the + /// counter migration; bumped alongside it with a parity check. + metrics: std::sync::Arc, + /// Time-series history of node-level metrics (1s/1m rings). stats_history: stats_history::StatsHistory, @@ -630,11 +643,24 @@ impl Node { let (decrypt_fallback_tx, decrypt_fallback_rx) = tokio::sync::mpsc::unbounded_channel::(); + let started_at = std::time::Instant::now(); + let context = Arc::new(context::NodeContext::new( + Arc::new(config.clone()), + identity.clone(), + startup_epoch, + started_at, + is_leaf_only, + max_connections, + max_peers, + max_links, + )); + Ok(Self { identity, startup_epoch, - started_at: std::time::Instant::now(), + started_at, config, + context, state: NodeState::Created, is_leaf_only, tree_state, @@ -659,6 +685,7 @@ impl Node { next_link_id: 1, next_transport_id: 1, stats: stats::NodeStats::new(), + metrics: std::sync::Arc::new(metrics::MetricsRegistry::new()), stats_history: stats_history::StatsHistory::new(), tun_state, tun_name: None, @@ -776,11 +803,24 @@ impl Node { let (decrypt_fallback_tx, decrypt_fallback_rx) = tokio::sync::mpsc::unbounded_channel::(); + let started_at = std::time::Instant::now(); + let context = Arc::new(context::NodeContext::new( + Arc::new(config.clone()), + identity.clone(), + startup_epoch, + started_at, + false, + max_connections, + max_peers, + max_links, + )); + Ok(Self { identity, startup_epoch, - started_at: std::time::Instant::now(), + started_at, config, + context, state: NodeState::Created, is_leaf_only: false, tree_state, @@ -805,6 +845,7 @@ impl Node { next_link_id: 1, next_transport_id: 1, stats: stats::NodeStats::new(), + metrics: std::sync::Arc::new(metrics::MetricsRegistry::new()), stats_history: stats_history::StatsHistory::new(), tun_state, tun_name: None, @@ -862,6 +903,7 @@ impl Node { let mut node = Self::new(config)?; node.is_leaf_only = true; node.bloom_state = BloomState::leaf_only(*node.identity.node_addr()); + node.rebuild_context(); Ok(node) } @@ -1079,7 +1121,7 @@ impl Node { /// Get this node's identity. pub fn identity(&self) -> &Identity { - &self.identity + &self.context.identity } /// Get this node's NodeAddr. @@ -1129,7 +1171,25 @@ impl Node { /// Get the configuration. pub fn config(&self) -> &Config { - &self.config + self.context.config.as_ref() + } + + /// Rebuild the shared [`context::NodeContext`] from the current `Node` + /// fields. Called after any mutation of a bundled field (`update_peers`, + /// the `set_max_*` setters) so `self.context` stays equal to the `Node` + /// fields it mirrors. Cheap — the only deep copy is the (rare) `Config` + /// clone. + fn rebuild_context(&mut self) { + self.context = Arc::new(context::NodeContext::new( + Arc::new(self.config.clone()), + self.identity.clone(), + self.startup_epoch, + self.started_at, + self.is_leaf_only, + self.max_connections, + self.max_peers, + self.max_links, + )); } /// Calculate the effective IPv6 MTU that can be sent over FIPS. @@ -1183,7 +1243,7 @@ impl Node { /// Get the node uptime. pub fn uptime(&self) -> std::time::Duration { - self.started_at.elapsed() + self.context.started_at.elapsed() } /// Check if node is operational. @@ -1193,7 +1253,7 @@ impl Node { /// Check if this is a leaf-only node. pub fn is_leaf_only(&self) -> bool { - self.is_leaf_only + self.context.is_leaf_only } // === Tree State === @@ -1339,6 +1399,11 @@ impl Node { &mut self.stats } + /// Get the atomic metric registry. + pub(crate) fn metrics(&self) -> &metrics::MetricsRegistry { + &self.metrics + } + /// Get the stats history collector. pub fn stats_history(&self) -> &stats_history::StatsHistory { &self.stats_history @@ -1347,7 +1412,7 @@ impl Node { /// Sample the current node state into the stats history ring. /// Called once per tick from the RX loop. pub(crate) fn record_stats_history(&mut self) { - let fwd = &self.stats.forwarding; + let fwd = &self.metrics.forwarding; let peers_with_mmp: Vec = self .peers .values() @@ -1363,11 +1428,11 @@ impl Node { mesh_size: self.estimated_mesh_size, tree_depth: self.tree_state.my_coords().depth() as u32, peer_count: self.peers.len() as u64, - parent_switches_total: self.stats.tree.parent_switches, - bytes_in_total: fwd.received_bytes, - bytes_out_total: fwd.forwarded_bytes + fwd.originated_bytes, - packets_in_total: fwd.received_packets, - packets_out_total: fwd.forwarded_packets + fwd.originated_packets, + parent_switches_total: self.metrics.tree.parent_switches.get(), + bytes_in_total: fwd.received_bytes.get(), + bytes_out_total: fwd.forwarded_bytes.get() + fwd.originated_bytes.get(), + packets_in_total: fwd.received_packets.get(), + packets_out_total: fwd.forwarded_packets.get() + fwd.originated_packets.get(), loss_rate, active_sessions: self.sessions.len() as u64, }; @@ -1420,11 +1485,13 @@ impl Node { /// Set the maximum number of connections (handshake phase). pub fn set_max_connections(&mut self, max: usize) { self.max_connections = max; + self.rebuild_context(); } /// Set the maximum number of peers (authenticated). pub fn set_max_peers(&mut self, max: usize) { self.max_peers = max; + self.rebuild_context(); } /// Returns false when we are at or above the configured `max_peers` @@ -1442,6 +1509,7 @@ impl Node { /// Set the maximum number of links. pub fn set_max_links(&mut self, max: usize) { self.max_links = max; + self.rebuild_context(); } // === Counts === diff --git a/src/node/retry.rs b/src/node/retry.rs index 91f2f08..db5bf0d 100644 --- a/src/node/retry.rs +++ b/src/node/retry.rs @@ -66,7 +66,7 @@ impl Node { /// indefinitely). Does nothing if the peer is already connected or has /// a connection in progress. pub(super) fn schedule_retry(&mut self, node_addr: NodeAddr, now_ms: u64) { - let retry_cfg = &self.config.node.retry; + let retry_cfg = &self.config().node.retry; let max_retries = retry_cfg.max_retries; if max_retries == 0 { return; @@ -105,7 +105,7 @@ impl Node { } else { // First failure — find the matching PeerConfig let peer_config = self - .config + .config() .auto_connect_peers() .find(|pc| { PeerIdentity::from_npub(&pc.npub) @@ -144,7 +144,7 @@ impl Node { pub(super) fn schedule_reconnect(&mut self, node_addr: NodeAddr, now_ms: u64) { // Find peer in auto-connect config let peer_config = self - .config + .config() .auto_connect_peers() .find(|pc| { PeerIdentity::from_npub(&pc.npub) @@ -165,8 +165,8 @@ impl Node { return; } - let base_interval_ms = self.config.node.retry.base_interval_secs * 1000; - let max_backoff_ms = self.config.node.retry.max_backoff_secs * 1000; + let base_interval_ms = self.config().node.retry.base_interval_secs * 1000; + let max_backoff_ms = self.config().node.retry.max_backoff_secs * 1000; let peer_name = self.peer_display_name(&node_addr); // If we already have accumulated backoff from previous failed attempts, @@ -305,14 +305,14 @@ impl Node { // succeeds, promote_connection() clears retry_pending. If // it times out, check_timeouts() calls schedule_retry() // which bumps the counter and applies proper backoff. - let hs_timeout_ms = self.config.node.rate_limit.handshake_timeout_secs * 1000; + let hs_timeout_ms = self.config().node.rate_limit.handshake_timeout_secs * 1000; if let Some(state) = self.retry_pending.get_mut(&node_addr) { state.retry_after_ms = now_ms + hs_timeout_ms; } debug!( peer = %self.peer_display_name(&node_addr), "Retry connection initiated, suppressing re-fire for {}s", - self.config.node.rate_limit.handshake_timeout_secs, + self.config().node.rate_limit.handshake_timeout_secs, ); } Err(e) => { diff --git a/src/node/stats.rs b/src/node/stats.rs index 3e96033..f6db420 100644 --- a/src/node/stats.rs +++ b/src/node/stats.rs @@ -1,4 +1,7 @@ -//! Node-level statistics for routing, forwarding, and discovery operations. +//! Node-level statistics for the session, handshake, mmp, and transport +//! families. The forwarding, discovery, tree, bloom, congestion, and error +//! families have migrated to the atomic +//! [`MetricsRegistry`](crate::node::metrics::MetricsRegistry). //! //! Unlike `EthernetStats` (which uses `AtomicU64` + `Arc` for cross-task //! sharing), these counters use plain `u64` because `Node` handlers run @@ -8,297 +11,9 @@ use serde::Serialize; use crate::node::reject::{ - BloomReject, DiscoveryReject, ForwardingReject, HandshakeReject, MmpReject, RejectReason, - SessionReject, TransportReject, TreeReject, + HandshakeReject, MmpReject, RejectReason, SessionReject, TransportReject, }; -/// Forwarding statistics — packets and bytes for each outcome. -#[derive(Default)] -pub struct ForwardingStats { - pub received_packets: u64, - pub received_bytes: u64, - pub decode_error_packets: u64, - pub decode_error_bytes: u64, - pub ttl_exhausted_packets: u64, - pub ttl_exhausted_bytes: u64, - pub delivered_packets: u64, - pub delivered_bytes: u64, - pub forwarded_packets: u64, - pub forwarded_bytes: u64, - pub drop_no_route_packets: u64, - pub drop_no_route_bytes: u64, - pub drop_mtu_exceeded_packets: u64, - pub drop_mtu_exceeded_bytes: u64, - pub drop_send_error_packets: u64, - pub drop_send_error_bytes: u64, - pub originated_packets: u64, - pub originated_bytes: u64, -} - -impl ForwardingStats { - pub fn record_received(&mut self, bytes: usize) { - self.received_packets += 1; - self.received_bytes += bytes as u64; - } - - pub fn record_decode_error(&mut self, bytes: usize) { - self.decode_error_packets += 1; - self.decode_error_bytes += bytes as u64; - } - - pub fn record_ttl_exhausted(&mut self, bytes: usize) { - self.ttl_exhausted_packets += 1; - self.ttl_exhausted_bytes += bytes as u64; - } - - pub fn record_delivered(&mut self, bytes: usize) { - self.delivered_packets += 1; - self.delivered_bytes += bytes as u64; - } - - pub fn record_forwarded(&mut self, bytes: usize) { - self.forwarded_packets += 1; - self.forwarded_bytes += bytes as u64; - } - - pub fn record_drop_no_route(&mut self, bytes: usize) { - self.drop_no_route_packets += 1; - self.drop_no_route_bytes += bytes as u64; - } - - pub fn record_drop_mtu_exceeded(&mut self, bytes: usize) { - self.drop_mtu_exceeded_packets += 1; - self.drop_mtu_exceeded_bytes += bytes as u64; - } - - pub fn record_drop_send_error(&mut self, bytes: usize) { - self.drop_send_error_packets += 1; - self.drop_send_error_bytes += bytes as u64; - } - - pub fn record_originated(&mut self, bytes: usize) { - self.originated_packets += 1; - self.originated_bytes += bytes as u64; - } - - /// Dispatch a typed forwarding rejection to its packet counter. - /// - /// The byte-counted side of each outcome is recorded by the - /// existing `record_*` methods at the call site (which know the - /// payload size); `record_reject` only bumps the packet count and - /// is paired with the byte-aware call at the call site while the - /// typed-rejection rollout is in progress. A later change may - /// collapse the two calls into a single typed entry point. - pub(super) fn record_reject(&mut self, reason: ForwardingReject) { - match reason { - ForwardingReject::DecodeError => self.decode_error_packets += 1, - ForwardingReject::TtlExhausted => self.ttl_exhausted_packets += 1, - ForwardingReject::NoRoute => self.drop_no_route_packets += 1, - ForwardingReject::MtuExceeded => self.drop_mtu_exceeded_packets += 1, - ForwardingReject::SendError => self.drop_send_error_packets += 1, - } - } - - pub fn snapshot(&self) -> ForwardingStatsSnapshot { - ForwardingStatsSnapshot { - received_packets: self.received_packets, - received_bytes: self.received_bytes, - decode_error_packets: self.decode_error_packets, - decode_error_bytes: self.decode_error_bytes, - ttl_exhausted_packets: self.ttl_exhausted_packets, - ttl_exhausted_bytes: self.ttl_exhausted_bytes, - delivered_packets: self.delivered_packets, - delivered_bytes: self.delivered_bytes, - forwarded_packets: self.forwarded_packets, - forwarded_bytes: self.forwarded_bytes, - drop_no_route_packets: self.drop_no_route_packets, - drop_no_route_bytes: self.drop_no_route_bytes, - drop_mtu_exceeded_packets: self.drop_mtu_exceeded_packets, - drop_mtu_exceeded_bytes: self.drop_mtu_exceeded_bytes, - drop_send_error_packets: self.drop_send_error_packets, - drop_send_error_bytes: self.drop_send_error_bytes, - originated_packets: self.originated_packets, - originated_bytes: self.originated_bytes, - } - } -} - -/// Discovery statistics — packet counts for request and response handling. -#[derive(Default)] -pub struct DiscoveryStats { - // Request counters - pub req_received: u64, - pub req_decode_error: u64, - pub req_duplicate: u64, - pub req_dedup_cache_full: u64, - pub req_target_is_us: u64, - pub req_forwarded: u64, - pub req_ttl_exhausted: u64, - pub req_initiated: u64, - pub req_deduplicated: u64, - pub req_backoff_suppressed: u64, - pub req_forward_rate_limited: u64, - pub req_bloom_miss: u64, - pub req_no_tree_peer: u64, - pub req_fallback_forwarded: u64, - // Response counters - pub resp_received: u64, - pub resp_decode_error: u64, - pub resp_forwarded: u64, - pub resp_identity_miss: u64, - pub resp_proof_failed: u64, - pub resp_no_route: u64, - pub resp_accepted: u64, - pub resp_timed_out: u64, -} - -impl DiscoveryStats { - pub(super) fn record_reject(&mut self, reason: DiscoveryReject) { - match reason { - DiscoveryReject::ReqDecodeError => self.req_decode_error += 1, - DiscoveryReject::ReqDuplicate => self.req_duplicate += 1, - DiscoveryReject::ReqDedupCacheFull => self.req_dedup_cache_full += 1, - DiscoveryReject::ReqTtlExhausted => self.req_ttl_exhausted += 1, - DiscoveryReject::RespDecodeError => self.resp_decode_error += 1, - DiscoveryReject::RespIdentityMiss => self.resp_identity_miss += 1, - DiscoveryReject::RespProofFailed => self.resp_proof_failed += 1, - DiscoveryReject::RespNoRoute => self.resp_no_route += 1, - } - } - - pub fn snapshot(&self) -> DiscoveryStatsSnapshot { - DiscoveryStatsSnapshot { - req_received: self.req_received, - req_decode_error: self.req_decode_error, - req_duplicate: self.req_duplicate, - req_dedup_cache_full: self.req_dedup_cache_full, - req_target_is_us: self.req_target_is_us, - req_forwarded: self.req_forwarded, - req_ttl_exhausted: self.req_ttl_exhausted, - req_initiated: self.req_initiated, - req_deduplicated: self.req_deduplicated, - req_backoff_suppressed: self.req_backoff_suppressed, - req_forward_rate_limited: self.req_forward_rate_limited, - req_bloom_miss: self.req_bloom_miss, - req_no_tree_peer: self.req_no_tree_peer, - req_fallback_forwarded: self.req_fallback_forwarded, - resp_received: self.resp_received, - resp_decode_error: self.resp_decode_error, - resp_forwarded: self.resp_forwarded, - resp_identity_miss: self.resp_identity_miss, - resp_proof_failed: self.resp_proof_failed, - resp_no_route: self.resp_no_route, - resp_accepted: self.resp_accepted, - resp_timed_out: self.resp_timed_out, - } - } -} - -/// Spanning tree statistics — announce handling and parent tracking. -#[derive(Default)] -pub struct TreeStats { - // Inbound announce handling - pub received: u64, - pub decode_error: u64, - pub unknown_peer: u64, - pub addr_mismatch: u64, - pub sig_failed: u64, - pub stale: u64, - pub ancestry_invalid: u64, - pub accepted: u64, - pub parent_switched: u64, - pub loop_detected: u64, - pub ancestry_changed: u64, - // Outbound announce sending - pub sent: u64, - pub rate_limited: u64, - pub send_failed: u64, - pub outbound_sign_failed: u64, - // Cumulative events - pub parent_switches: u64, - pub parent_losses: u64, - pub flap_dampened: u64, -} - -impl TreeStats { - pub fn snapshot(&self) -> TreeStatsSnapshot { - TreeStatsSnapshot { - received: self.received, - decode_error: self.decode_error, - unknown_peer: self.unknown_peer, - addr_mismatch: self.addr_mismatch, - sig_failed: self.sig_failed, - stale: self.stale, - ancestry_invalid: self.ancestry_invalid, - accepted: self.accepted, - parent_switched: self.parent_switched, - loop_detected: self.loop_detected, - ancestry_changed: self.ancestry_changed, - sent: self.sent, - rate_limited: self.rate_limited, - send_failed: self.send_failed, - outbound_sign_failed: self.outbound_sign_failed, - parent_switches: self.parent_switches, - parent_losses: self.parent_losses, - flap_dampened: self.flap_dampened, - } - } - - pub(super) fn record_reject(&mut self, reason: TreeReject) { - match reason { - TreeReject::AncestryInvalid => self.ancestry_invalid += 1, - TreeReject::OutboundSignFailed => self.outbound_sign_failed += 1, - } - } -} - -/// Bloom filter statistics — filter announce handling. -#[derive(Default)] -pub struct BloomStats { - // Inbound announce handling - pub received: u64, - pub decode_error: u64, - pub invalid: u64, - pub non_v1: u64, - pub unknown_peer: u64, - pub stale: u64, - pub fill_exceeded: u64, - pub accepted: u64, - // Outbound announce sending - pub sent: u64, - pub debounce_suppressed: u64, - pub send_failed: u64, -} - -impl BloomStats { - pub(super) fn record_reject(&mut self, reason: BloomReject) { - match reason { - BloomReject::DecodeError => self.decode_error += 1, - BloomReject::Invalid => self.invalid += 1, - BloomReject::NonV1 => self.non_v1 += 1, - BloomReject::UnknownPeer => self.unknown_peer += 1, - BloomReject::Stale => self.stale += 1, - BloomReject::FillExceeded => self.fill_exceeded += 1, - } - } - - pub fn snapshot(&self) -> BloomStatsSnapshot { - BloomStatsSnapshot { - received: self.received, - decode_error: self.decode_error, - invalid: self.invalid, - non_v1: self.non_v1, - unknown_peer: self.unknown_peer, - stale: self.stale, - fill_exceeded: self.fill_exceeded, - accepted: self.accepted, - sent: self.sent, - debounce_suppressed: self.debounce_suppressed, - send_failed: self.send_failed, - } - } -} - /// FSP session statistics — receive-path silent-rejection counters. /// /// Covers the unknown-session and state-machine-mismatch rejection @@ -446,77 +161,19 @@ impl TransportStats { } } -/// Error signal statistics — counts of each error signal type received. -#[derive(Default)] -pub struct ErrorSignalStats { - pub coords_required: u64, - pub path_broken: u64, - pub mtu_exceeded: u64, -} - -impl ErrorSignalStats { - pub fn snapshot(&self) -> ErrorSignalStatsSnapshot { - ErrorSignalStatsSnapshot { - coords_required: self.coords_required, - path_broken: self.path_broken, - mtu_exceeded: self.mtu_exceeded, - } - } -} - -/// Congestion event statistics — ECN CE tracking and detection triggers. -#[derive(Default)] -pub struct CongestionStats { - /// Packets forwarded with the CE flag set (incoming CE or locally detected). - pub ce_forwarded: u64, - /// CE-flagged packets received at this node as final destination. - pub ce_received: u64, - /// Number of times detect_congestion() returned true. - pub congestion_detected: u64, - /// Rising-edge transport kernel drop events (not-dropping → dropping). - pub kernel_drop_events: u64, -} - -impl CongestionStats { - pub fn record_ce_forwarded(&mut self) { - self.ce_forwarded += 1; - } - - pub fn record_ce_received(&mut self) { - self.ce_received += 1; - } - - pub fn record_congestion_detected(&mut self) { - self.congestion_detected += 1; - } - - pub fn record_kernel_drop_event(&mut self) { - self.kernel_drop_events += 1; - } - - pub fn snapshot(&self) -> CongestionStatsSnapshot { - CongestionStatsSnapshot { - ce_forwarded: self.ce_forwarded, - ce_received: self.ce_received, - congestion_detected: self.congestion_detected, - kernel_drop_events: self.kernel_drop_events, - } - } -} - /// Aggregate node statistics. +/// +/// Holds only the families that have not migrated to the atomic +/// [`MetricsRegistry`](crate::node::metrics::MetricsRegistry): session, +/// handshake, mmp, and transport. The forwarding, discovery, tree, +/// bloom, congestion, and error families are served exclusively from +/// the registry. #[derive(Default)] pub struct NodeStats { - pub forwarding: ForwardingStats, - pub discovery: DiscoveryStats, - pub tree: TreeStats, - pub bloom: BloomStats, pub session: SessionStats, pub handshake: HandshakeStats, pub mmp: MmpStats, pub transport: TransportStats, - pub errors: ErrorSignalStats, - pub congestion: CongestionStats, } impl NodeStats { @@ -524,37 +181,28 @@ impl NodeStats { Self::default() } - pub fn snapshot(&self) -> NodeStatsSnapshot { - NodeStatsSnapshot { - forwarding: self.forwarding.snapshot(), - discovery: self.discovery.snapshot(), - tree: self.tree.snapshot(), - bloom: self.bloom.snapshot(), - session: self.session.snapshot(), - handshake: self.handshake.snapshot(), - mmp: self.mmp.snapshot(), - transport: self.transport.snapshot(), - errors: self.errors.snapshot(), - congestion: self.congestion.snapshot(), - } - } - /// Record a typed rejection from a silent-rejection site. /// /// Dispatches to the appropriate sub-stats `record_reject` based on - /// the [`RejectReason`] top-level variant. Sub-enums that have not - /// yet had any variants populated still use `match r {}` to keep - /// the dispatch arm exhaustive without dead-code complaints. + /// the [`RejectReason`] top-level variant. Only the families still + /// stored on `NodeStats` (session, handshake, mmp, transport) are + /// routed here; the migrated families are recorded directly on the + /// [`MetricsRegistry`](crate::node::metrics::MetricsRegistry). pub fn record_reject(&mut self, reason: RejectReason) { match reason { - RejectReason::Tree(r) => self.tree.record_reject(r), - RejectReason::Bloom(r) => self.bloom.record_reject(r), - RejectReason::Discovery(r) => self.discovery.record_reject(r), RejectReason::Session(r) => self.session.record_reject(r), RejectReason::Handshake(r) => self.handshake.record_reject(r), - RejectReason::Forwarding(r) => self.forwarding.record_reject(r), RejectReason::Transport(r) => self.transport.record_reject(r), RejectReason::Mmp(r) => self.mmp.record_reject(r), + // The forwarding, discovery, tree, and bloom families are + // recorded directly on the MetricsRegistry and never reach + // this NodeStats dispatch. + RejectReason::Forwarding(_) + | RejectReason::Discovery(_) + | RejectReason::Tree(_) + | RejectReason::Bloom(_) => { + debug_assert!(false, "migrated reject family must use MetricsRegistry"); + } } } } @@ -684,50 +332,10 @@ pub struct CongestionStatsSnapshot { pub kernel_drop_events: u64, } -#[derive(Clone, Debug, Default, Serialize)] -pub struct NodeStatsSnapshot { - pub forwarding: ForwardingStatsSnapshot, - pub discovery: DiscoveryStatsSnapshot, - pub tree: TreeStatsSnapshot, - pub bloom: BloomStatsSnapshot, - pub session: SessionStatsSnapshot, - pub handshake: HandshakeStatsSnapshot, - pub mmp: MmpStatsSnapshot, - pub transport: TransportStatsSnapshot, - pub errors: ErrorSignalStatsSnapshot, - pub congestion: CongestionStatsSnapshot, -} - #[cfg(test)] mod tests { use super::*; - #[test] - fn tree_stats_record_reject_ancestry_invalid() { - let mut stats = TreeStats::default(); - stats.record_reject(TreeReject::AncestryInvalid); - stats.record_reject(TreeReject::AncestryInvalid); - assert_eq!(stats.ancestry_invalid, 2); - assert_eq!(stats.outbound_sign_failed, 0); - } - - #[test] - fn tree_stats_record_reject_outbound_sign_failed() { - let mut stats = TreeStats::default(); - stats.record_reject(TreeReject::OutboundSignFailed); - stats.record_reject(TreeReject::OutboundSignFailed); - assert_eq!(stats.outbound_sign_failed, 2); - assert_eq!(stats.ancestry_invalid, 0); - } - - #[test] - fn node_stats_record_reject_dispatches_to_tree() { - let mut stats = NodeStats::new(); - stats.record_reject(RejectReason::Tree(TreeReject::OutboundSignFailed)); - assert_eq!(stats.tree.outbound_sign_failed, 1); - assert_eq!(stats.tree.ancestry_invalid, 0); - } - #[test] fn session_stats_record_reject_unknown_session() { let mut stats = SessionStats::default(); @@ -753,7 +361,6 @@ mod tests { stats.record_reject(RejectReason::Session(SessionReject::BadState)); assert_eq!(stats.session.unknown_session, 1); assert_eq!(stats.session.bad_state, 1); - assert_eq!(stats.tree.ancestry_invalid, 0); } #[test] @@ -784,165 +391,6 @@ mod tests { assert_eq!(stats.handshake.bad_state, 2); assert_eq!(stats.handshake.unknown_connection, 1); assert_eq!(stats.session.unknown_session, 0); - assert_eq!(stats.tree.ancestry_invalid, 0); - } - - #[test] - fn bloom_stats_record_reject_decode_error() { - let mut s = BloomStats::default(); - s.record_reject(BloomReject::DecodeError); - s.record_reject(BloomReject::DecodeError); - assert_eq!(s.decode_error, 2); - assert_eq!(s.invalid, 0); - } - - #[test] - fn bloom_stats_record_reject_invalid() { - let mut s = BloomStats::default(); - s.record_reject(BloomReject::Invalid); - assert_eq!(s.invalid, 1); - } - - #[test] - fn bloom_stats_record_reject_non_v1() { - let mut s = BloomStats::default(); - s.record_reject(BloomReject::NonV1); - assert_eq!(s.non_v1, 1); - } - - #[test] - fn bloom_stats_record_reject_unknown_peer() { - let mut s = BloomStats::default(); - s.record_reject(BloomReject::UnknownPeer); - assert_eq!(s.unknown_peer, 1); - } - - #[test] - fn bloom_stats_record_reject_stale() { - let mut s = BloomStats::default(); - s.record_reject(BloomReject::Stale); - assert_eq!(s.stale, 1); - } - - #[test] - fn bloom_stats_record_reject_fill_exceeded() { - let mut s = BloomStats::default(); - s.record_reject(BloomReject::FillExceeded); - assert_eq!(s.fill_exceeded, 1); - } - - #[test] - fn node_stats_record_reject_dispatches_to_bloom() { - let mut stats = NodeStats::new(); - stats.record_reject(RejectReason::Bloom(BloomReject::DecodeError)); - stats.record_reject(RejectReason::Bloom(BloomReject::Stale)); - assert_eq!(stats.bloom.decode_error, 1); - assert_eq!(stats.bloom.stale, 1); - assert_eq!(stats.tree.ancestry_invalid, 0); - } - - #[test] - fn discovery_stats_record_reject_req_decode_error() { - let mut s = DiscoveryStats::default(); - s.record_reject(DiscoveryReject::ReqDecodeError); - assert_eq!(s.req_decode_error, 1); - } - - #[test] - fn discovery_stats_record_reject_req_duplicate() { - let mut s = DiscoveryStats::default(); - s.record_reject(DiscoveryReject::ReqDuplicate); - assert_eq!(s.req_duplicate, 1); - } - - #[test] - fn discovery_stats_record_reject_req_dedup_cache_full() { - let mut s = DiscoveryStats::default(); - s.record_reject(DiscoveryReject::ReqDedupCacheFull); - assert_eq!(s.req_dedup_cache_full, 1); - } - - #[test] - fn discovery_stats_record_reject_req_ttl_exhausted() { - let mut s = DiscoveryStats::default(); - s.record_reject(DiscoveryReject::ReqTtlExhausted); - assert_eq!(s.req_ttl_exhausted, 1); - } - - #[test] - fn discovery_stats_record_reject_resp_decode_error() { - let mut s = DiscoveryStats::default(); - s.record_reject(DiscoveryReject::RespDecodeError); - assert_eq!(s.resp_decode_error, 1); - } - - #[test] - fn discovery_stats_record_reject_resp_identity_miss() { - let mut s = DiscoveryStats::default(); - s.record_reject(DiscoveryReject::RespIdentityMiss); - assert_eq!(s.resp_identity_miss, 1); - } - - #[test] - fn discovery_stats_record_reject_resp_proof_failed() { - let mut s = DiscoveryStats::default(); - s.record_reject(DiscoveryReject::RespProofFailed); - assert_eq!(s.resp_proof_failed, 1); - } - - #[test] - fn node_stats_record_reject_dispatches_to_discovery() { - let mut stats = NodeStats::new(); - stats.record_reject(RejectReason::Discovery(DiscoveryReject::ReqDecodeError)); - stats.record_reject(RejectReason::Discovery(DiscoveryReject::RespProofFailed)); - assert_eq!(stats.discovery.req_decode_error, 1); - assert_eq!(stats.discovery.resp_proof_failed, 1); - assert_eq!(stats.tree.ancestry_invalid, 0); - } - - #[test] - fn forwarding_stats_record_reject_decode_error() { - let mut s = ForwardingStats::default(); - s.record_reject(ForwardingReject::DecodeError); - assert_eq!(s.decode_error_packets, 1); - } - - #[test] - fn forwarding_stats_record_reject_ttl_exhausted() { - let mut s = ForwardingStats::default(); - s.record_reject(ForwardingReject::TtlExhausted); - assert_eq!(s.ttl_exhausted_packets, 1); - } - - #[test] - fn forwarding_stats_record_reject_no_route() { - let mut s = ForwardingStats::default(); - s.record_reject(ForwardingReject::NoRoute); - assert_eq!(s.drop_no_route_packets, 1); - } - - #[test] - fn forwarding_stats_record_reject_mtu_exceeded() { - let mut s = ForwardingStats::default(); - s.record_reject(ForwardingReject::MtuExceeded); - assert_eq!(s.drop_mtu_exceeded_packets, 1); - } - - #[test] - fn forwarding_stats_record_reject_send_error() { - let mut s = ForwardingStats::default(); - s.record_reject(ForwardingReject::SendError); - assert_eq!(s.drop_send_error_packets, 1); - } - - #[test] - fn node_stats_record_reject_dispatches_to_forwarding() { - let mut stats = NodeStats::new(); - stats.record_reject(RejectReason::Forwarding(ForwardingReject::NoRoute)); - stats.record_reject(RejectReason::Forwarding(ForwardingReject::MtuExceeded)); - assert_eq!(stats.forwarding.drop_no_route_packets, 1); - assert_eq!(stats.forwarding.drop_mtu_exceeded_packets, 1); - assert_eq!(stats.tree.ancestry_invalid, 0); } #[test] @@ -969,7 +417,6 @@ mod tests { stats.record_reject(RejectReason::Mmp(MmpReject::UnknownPeer)); assert_eq!(stats.mmp.decode_error, 1); assert_eq!(stats.mmp.unknown_peer, 1); - assert_eq!(stats.tree.ancestry_invalid, 0); } #[test] @@ -985,6 +432,5 @@ mod tests { let mut stats = NodeStats::new(); stats.record_reject(RejectReason::Transport(TransportReject::InboundCapExceeded)); assert_eq!(stats.transport.inbound_cap_exceeded, 1); - assert_eq!(stats.tree.ancestry_invalid, 0); } } diff --git a/src/node/tests/bloom_poison.rs b/src/node/tests/bloom_poison.rs index b49dde7..e09437b 100644 --- a/src/node/tests/bloom_poison.rs +++ b/src/node/tests/bloom_poison.rs @@ -43,24 +43,22 @@ async fn test_m1_rejects_all_ones_filter_announce() { let announce = FilterAnnounce::new(all_ones, 1); let payload = encode_payload(&announce); - let before_fill_exceeded = node.stats().bloom.fill_exceeded; - let before_accepted = node.stats().bloom.accepted; + let before_fill_exceeded = node.metrics().bloom.fill_exceeded.get(); + let before_accepted = node.metrics().bloom.accepted.get(); node.handle_filter_announce(&peer_addr, &payload).await; - let after = &node.stats().bloom; - // While the typed-rejection rollout is in progress the call site - // bumps the counter directly AND dispatches through record_reject, - // which hits the same counter. A later change will collapse this to - // a single increment by removing the legacy direct bump; for now - // the rejection-path event yields a +2 delta. + // The rejection path bumps the counter once, through the typed + // record_reject dispatch. The legacy direct bump has been removed, + // so the rejection-path event yields a +1 delta. assert_eq!( - after.fill_exceeded, - before_fill_exceeded + 2, + node.metrics().bloom.fill_exceeded.get(), + before_fill_exceeded + 1, "fill_exceeded counter must increment on all-ones rejection" ); assert_eq!( - after.accepted, before_accepted, + node.metrics().bloom.accepted.get(), + before_accepted, "accepted counter must NOT increment on rejection" ); @@ -93,18 +91,18 @@ async fn test_m1_accepts_sub_cap_filter() { let announce = FilterAnnounce::new(filter, 1); let payload = encode_payload(&announce); - let before_fill_exceeded = node.stats().bloom.fill_exceeded; - let before_accepted = node.stats().bloom.accepted; + let before_fill_exceeded = node.metrics().bloom.fill_exceeded.get(); + let before_accepted = node.metrics().bloom.accepted.get(); node.handle_filter_announce(&peer_addr, &payload).await; - let after = &node.stats().bloom; assert_eq!( - after.fill_exceeded, before_fill_exceeded, + node.metrics().bloom.fill_exceeded.get(), + before_fill_exceeded, "fill_exceeded must NOT increment on legitimate sub-cap filter" ); assert_eq!( - after.accepted, + node.metrics().bloom.accepted.get(), before_accepted + 1, "accepted must increment on legitimate filter" ); @@ -164,9 +162,8 @@ async fn test_m1_sequence_not_advanced_allows_recovery() { "compliant announce at same seq must be accepted after rejection" ); assert_eq!(peer.filter_sequence(), 1); - // Direct bump + record_reject dispatch both increment the same - // counter while the typed-rejection rollout is in progress. A later - // change collapses these back to a single increment. - assert_eq!(node.stats().bloom.fill_exceeded, 2); - assert_eq!(node.stats().bloom.accepted, 1); + // The typed record_reject dispatch increments the counter once; the + // legacy direct bump has been removed. + assert_eq!(node.metrics().bloom.fill_exceeded.get(), 1); + assert_eq!(node.metrics().bloom.accepted.get(), 1); } diff --git a/src/node/tests/bootstrap.rs b/src/node/tests/bootstrap.rs index 4e59cda..9486259 100644 --- a/src/node/tests/bootstrap.rs +++ b/src/node/tests/bootstrap.rs @@ -292,11 +292,12 @@ async fn test_third_peer_can_handshake_via_adopted_transport_socket() { #[tokio::test] async fn test_adopted_udp_inherits_mtu_from_single_primary_config() { - let mut node = make_node(); - node.config.transports.udp = TransportInstances::Single(UdpConfig { + let mut config = Config::new(); + config.transports.udp = TransportInstances::Single(UdpConfig { mtu: Some(1500), ..Default::default() }); + let mut node = make_node_with(config); let (packet_tx, packet_rx) = packet_channel(64); node.packet_tx = Some(packet_tx); @@ -329,7 +330,6 @@ async fn test_adopted_udp_inherits_mtu_from_single_primary_config() { #[tokio::test] async fn test_adopted_udp_inherits_mtu_from_named_primary_config() { - let mut node = make_node(); let mut named = HashMap::new(); named.insert( "primary".to_string(), @@ -345,7 +345,9 @@ async fn test_adopted_udp_inherits_mtu_from_named_primary_config() { ..Default::default() }, ); - node.config.transports.udp = TransportInstances::Named(named); + let mut config = Config::new(); + config.transports.udp = TransportInstances::Named(named); + let mut node = make_node_with(config); let (packet_tx, packet_rx) = packet_channel(64); node.packet_tx = Some(packet_tx); diff --git a/src/node/tests/discovery.rs b/src/node/tests/discovery.rs index 458201d..0139046 100644 --- a/src/node/tests/discovery.rs +++ b/src/node/tests/discovery.rs @@ -1110,8 +1110,8 @@ async fn test_check_pending_lookups_default_sequence_unreachable() { node.pending_lookups .insert(target_addr, PendingLookup::new(0)); - let baseline_initiated = node.stats().discovery.req_initiated; - let baseline_timed_out = node.stats().discovery.resp_timed_out; + let baseline_initiated = node.metrics().discovery.req_initiated.get(); + let baseline_timed_out = node.metrics().discovery.resp_timed_out.get(); // --- t = 1100ms: first retry deadline (1*1000) --- node.check_pending_lookups(1100).await; @@ -1124,7 +1124,7 @@ async fn test_check_pending_lookups_default_sequence_unreachable() { assert_eq!(entry.last_sent_ms, 1100); } assert_eq!( - node.stats().discovery.req_initiated, + node.metrics().discovery.req_initiated.get(), baseline_initiated + 1, "retry #1 must invoke initiate_lookup exactly once" ); @@ -1140,7 +1140,7 @@ async fn test_check_pending_lookups_default_sequence_unreachable() { assert_eq!(entry.last_sent_ms, 3100); } assert_eq!( - node.stats().discovery.req_initiated, + node.metrics().discovery.req_initiated.get(), baseline_initiated + 2, "retry #2 must invoke initiate_lookup exactly once more" ); @@ -1156,7 +1156,7 @@ async fn test_check_pending_lookups_default_sequence_unreachable() { assert_eq!(entry.last_sent_ms, 7100); } assert_eq!( - node.stats().discovery.req_initiated, + node.metrics().discovery.req_initiated.get(), baseline_initiated + 3, "retry #3 must invoke initiate_lookup exactly once more" ); @@ -1168,12 +1168,12 @@ async fn test_check_pending_lookups_default_sequence_unreachable() { "8s window not yet expired: pending_lookup must persist" ); assert_eq!( - node.stats().discovery.req_initiated, + node.metrics().discovery.req_initiated.get(), baseline_initiated + 3, "no new attempt before final deadline" ); assert_eq!( - node.stats().discovery.resp_timed_out, + node.metrics().discovery.resp_timed_out.get(), baseline_timed_out, "no timeout before final deadline" ); @@ -1192,13 +1192,13 @@ async fn test_check_pending_lookups_default_sequence_unreachable() { ); // resp_timed_out counter ticked. assert_eq!( - node.stats().discovery.resp_timed_out, + node.metrics().discovery.resp_timed_out.get(), baseline_timed_out + 1, "final timeout must increment discovery.resp_timed_out" ); // No additional initiate_lookup on the timeout step. assert_eq!( - node.stats().discovery.req_initiated, + node.metrics().discovery.req_initiated.get(), baseline_initiated + 3, "the final-timeout step must NOT call initiate_lookup" ); diff --git a/src/node/tests/forwarding.rs b/src/node/tests/forwarding.rs index 603f028..a90f620 100644 --- a/src/node/tests/forwarding.rs +++ b/src/node/tests/forwarding.rs @@ -757,8 +757,9 @@ fn test_detect_congestion_with_transport_drops() { #[test] fn test_detect_congestion_disabled_ecn() { - let mut node = make_node(); - node.config.node.ecn.enabled = false; + let mut config = Config::new(); + config.node.ecn.enabled = false; + let mut node = Node::new(config).unwrap(); // Even with transport drops, disabled ECN should return false let tid = TransportId::new(1); diff --git a/src/node/tests/mod.rs b/src/node/tests/mod.rs index dbd1892..031191d 100644 --- a/src/node/tests/mod.rs +++ b/src/node/tests/mod.rs @@ -24,7 +24,14 @@ mod tcp; mod unit; pub(super) fn make_node() -> Node { - let config = Config::new(); + make_node_with(Config::new()) +} + +/// Build a test node from an explicit `Config`. Prefer this over poking +/// `node.config.*` after construction: immutable fields are mirrored into the +/// shared `NodeContext` at build time, so a post-construction field poke is +/// invisible to any reader that has migrated onto the `config()` accessor. +pub(super) fn make_node_with(config: Config) -> Node { Node::new(config).unwrap() } diff --git a/src/node/tests/session.rs b/src/node/tests/session.rs index 014a7d2..bdab289 100644 --- a/src/node/tests/session.rs +++ b/src/node/tests/session.rs @@ -1470,8 +1470,9 @@ fn test_purge_idle_sessions_cleans_pending_packets() { #[test] fn test_purge_idle_sessions_disabled_when_zero() { - let mut node = make_node(); - node.config.node.session.idle_timeout_secs = 0; + let mut config = Config::new(); + config.node.session.idle_timeout_secs = 0; + let mut node = make_node_with(config); let remote = Identity::generate(); let remote_addr = *remote.node_addr(); diff --git a/src/node/tests/spanning_tree.rs b/src/node/tests/spanning_tree.rs index f24221d..75accaa 100644 --- a/src/node/tests/spanning_tree.rs +++ b/src/node/tests/spanning_tree.rs @@ -830,7 +830,7 @@ async fn test_rejects_tree_announce_with_inconsistent_root() { .coords() .unwrap() .clone(); - let accepted_before = nodes[1].node.stats().tree.accepted; + let accepted_before = nodes[1].node.metrics().tree.accepted.get(); // Use two fixed synthetic ancestors so the forged path is explicit: // - fake_parent = 00000000000000000000000000000000 @@ -875,7 +875,7 @@ async fn test_rejects_tree_announce_with_inconsistent_root() { nodes[1].node.tree_state().my_coords().depth(), current_depth ); - assert_eq!(nodes[1].node.stats().tree.accepted, accepted_before); + assert_eq!(nodes[1].node.metrics().tree.accepted.get(), accepted_before); assert_eq!( nodes[1].node.get_peer(&a_addr).unwrap().coords().unwrap(), &peer_coords_before diff --git a/src/node/tests/tcp.rs b/src/node/tests/tcp.rs index 18fe25c..32076f4 100644 --- a/src/node/tests/tcp.rs +++ b/src/node/tests/tcp.rs @@ -6,7 +6,7 @@ //! All tests use 127.0.0.1:0 (ephemeral ports) and need no privileges. use super::*; -use crate::config::TcpConfig; +use crate::config::{Config, TcpConfig}; use crate::transport::tcp::TcpTransport; use crate::transport::{TransportAddr, TransportHandle, TransportId, packet_channel}; use spanning_tree::{ @@ -20,7 +20,14 @@ use std::time::Duration; /// TcpTransport instead of UDP. Binds to 127.0.0.1:0 for an /// ephemeral port. async fn make_test_node_tcp() -> TestNode { - let mut node = make_node(); + make_test_node_tcp_with(Config::new()).await +} + +/// Like `make_test_node_tcp` but builds the node from an explicit `Config`, +/// so immutable fields (e.g. heartbeat/link-dead timeouts) are set before the +/// `NodeContext` is built rather than poked afterward. +async fn make_test_node_tcp_with(config: Config) -> TestNode { + let mut node = make_node_with(config); let transport_id = TransportId::new(1); let config = TcpConfig { @@ -166,13 +173,14 @@ async fn test_tcp_mixed_transport_coexistence() { /// link-dead timeout fires. #[tokio::test] async fn test_tcp_connection_loss_detection() { - let mut nodes = vec![make_test_node_tcp().await, make_test_node_tcp().await]; - // Short heartbeat/link-dead timeouts for faster test execution - for tn in nodes.iter_mut() { - tn.node.config.node.heartbeat_interval_secs = 1; - tn.node.config.node.link_dead_timeout_secs = 3; - } + let mut config = Config::new(); + config.node.heartbeat_interval_secs = 1; + config.node.link_dead_timeout_secs = 3; + let mut nodes = vec![ + make_test_node_tcp_with(config.clone()).await, + make_test_node_tcp_with(config).await, + ]; // Establish peering initiate_handshake(&mut nodes, 0, 1).await; @@ -215,13 +223,14 @@ async fn test_tcp_connection_loss_detection() { /// Verifies that bidirectional peering is restored. #[tokio::test] async fn test_tcp_reconnection_after_link_death() { - let mut nodes = vec![make_test_node_tcp().await, make_test_node_tcp().await]; - // Short timeouts - for tn in nodes.iter_mut() { - tn.node.config.node.heartbeat_interval_secs = 1; - tn.node.config.node.link_dead_timeout_secs = 3; - } + let mut config = Config::new(); + config.node.heartbeat_interval_secs = 1; + config.node.link_dead_timeout_secs = 3; + let mut nodes = vec![ + make_test_node_tcp_with(config.clone()).await, + make_test_node_tcp_with(config).await, + ]; // Establish initial peering initiate_handshake(&mut nodes, 0, 1).await; diff --git a/src/node/tests/unit.rs b/src/node/tests/unit.rs index acbb403..fff71fd 100644 --- a/src/node/tests/unit.rs +++ b/src/node/tests/unit.rs @@ -1003,6 +1003,38 @@ fn active_peer_same_path_discovery_refreshes_stale_peer() { )); } +#[tokio::test] +async fn node_context_mirrors_config_and_immutable_facades() { + let mut node = make_node(); + + // The immutable facades read the shared NodeContext. + let expected_addr = *node.identity().node_addr(); + assert_eq!(node.node_addr(), &expected_addr); + assert!(!node.is_leaf_only()); + let _ = node.uptime(); + assert_eq!(node.config().peers().len(), 0); + + // update_peers must rebuild the context so config() — which now reads the + // context — reflects the new peer list. Guards the copy-on-write sync. + let peer = Identity::generate(); + let new_peer = crate::config::PeerConfig { + npub: peer.npub(), + alias: None, + addresses: vec![], + connect_policy: crate::config::ConnectPolicy::OnDemand, + auto_reconnect: false, + via_nostr: false, + }; + node.update_peers(vec![new_peer]).await.unwrap(); + + assert_eq!( + node.config().peers().len(), + 1, + "config() must reflect update_peers through the rebuilt context" + ); + assert_eq!(node.config().peers()[0].npub, peer.npub()); +} + #[tokio::test] async fn update_peers_races_new_alternative_without_dropping_active_peer() { let mut node = make_node(); diff --git a/src/node/tree.rs b/src/node/tree.rs index e6ca1cd..9378da1 100644 --- a/src/node/tree.rs +++ b/src/node/tree.rs @@ -8,7 +8,7 @@ use std::collections::HashMap; use crate::NodeAddr; use crate::protocol::TreeAnnounce; -use super::reject::{RejectReason, TreeReject}; +use super::reject::TreeReject; use super::{Node, NodeError}; use tracing::{debug, info, trace, warn}; @@ -20,7 +20,7 @@ impl Node { if !decl.is_signed() { return Err(NodeError::SendFailed { - node_addr: *self.identity.node_addr(), + node_addr: *self.identity().node_addr(), reason: "declaration not signed".into(), }); } @@ -49,7 +49,7 @@ impl Node { if !peer.can_send_tree_announce(now_ms) { peer.mark_tree_announce_pending(); - self.stats_mut().tree.rate_limited += 1; + self.metrics().tree.rate_limited.inc(); debug!( peer = %self.peer_display_name(peer_addr), "TreeAnnounce rate-limited, marking pending" @@ -66,11 +66,11 @@ impl Node { // Send if let Err(e) = self.send_encrypted_link_message(peer_addr, &encoded).await { - self.stats_mut().tree.send_failed += 1; + self.metrics().tree.send_failed.inc(); return Err(e); } - self.stats_mut().tree.sent += 1; + self.metrics().tree.sent.inc(); // Record send time if let Some(peer) = self.peers.get_mut(peer_addr) { @@ -131,12 +131,12 @@ impl Node { /// 4. Re-evaluate parent selection /// 5. If parent changed: increment seq, sign, recompute coords, announce to all pub(super) async fn handle_tree_announce(&mut self, from: &NodeAddr, payload: &[u8]) { - self.stats_mut().tree.received += 1; + self.metrics().tree.received.inc(); let announce = match TreeAnnounce::decode(payload) { Ok(a) => a, Err(e) => { - self.stats_mut().tree.decode_error += 1; + self.metrics().tree.decode_error.inc(); debug!(from = %self.peer_display_name(from), error = %e, "Malformed TreeAnnounce"); return; } @@ -146,7 +146,7 @@ impl Node { let pubkey = match self.peers.get(from) { Some(peer) => peer.pubkey(), None => { - self.stats_mut().tree.unknown_peer += 1; + self.metrics().tree.unknown_peer.inc(); debug!(from = %self.peer_display_name(from), "TreeAnnounce from unknown peer"); return; } @@ -154,7 +154,7 @@ impl Node { // The declaring node_addr in the announce should match the sender if announce.declaration.node_addr() != from { - self.stats_mut().tree.addr_mismatch += 1; + self.metrics().tree.addr_mismatch.inc(); debug!( from = %self.peer_display_name(from), declared = %announce.declaration.node_addr(), @@ -164,7 +164,7 @@ impl Node { } if let Err(e) = announce.declaration.verify(&pubkey) { - self.stats_mut().tree.sig_failed += 1; + self.metrics().tree.sig_failed.inc(); warn!( from = %self.peer_display_name(from), error = %e, @@ -174,8 +174,9 @@ impl Node { } if let Err(e) = announce.validate_semantics() { - self.stats_mut() - .record_reject(RejectReason::Tree(TreeReject::AncestryInvalid)); + self.metrics() + .tree + .record_reject(TreeReject::AncestryInvalid); warn!( from = %self.peer_display_name(from), error = %e, @@ -204,12 +205,12 @@ impl Node { .update_peer(announce.declaration.clone(), announce.ancestry.clone()); if !updated { - self.stats_mut().tree.stale += 1; + self.metrics().tree.stale.inc(); debug!(from = %self.peer_display_name(from), "TreeAnnounce not fresher than existing, ignored"); return; } - self.stats_mut().tree.accepted += 1; + self.metrics().tree.accepted.inc(); debug!( from = %self.peer_display_name(from), @@ -244,24 +245,28 @@ impl Node { .map(|d| d.as_secs()) .unwrap_or(0); + // Clone identity up front to avoid a split borrow against the + // &mut self.tree_state / &mut self.coord_cache calls below (cold path). + let our_identity = self.identity().clone(); let flap_dampened = self.tree_state.set_parent(new_parent, new_seq, timestamp); // recompute_coords may demote to self_root if the new path would be // invalid; sign AFTER recompute so the signature covers the final // declaration. self.tree_state.recompute_coords(); - if let Err(e) = self.tree_state.sign_declaration(&self.identity) { + if let Err(e) = self.tree_state.sign_declaration(&our_identity) { warn!(error = %e, "Failed to sign declaration after parent switch"); - self.stats_mut() - .record_reject(RejectReason::Tree(TreeReject::OutboundSignFailed)); + self.metrics() + .tree + .record_reject(TreeReject::OutboundSignFailed); return; } // Surgical invalidation — see CoordCache::invalidate_via_node doc. self.coord_cache - .invalidate_via_node(self.identity.node_addr()); + .invalidate_via_node(our_identity.node_addr()); self.reset_discovery_backoff(); - self.stats_mut().tree.parent_switched += 1; - self.stats_mut().tree.parent_switches += 1; + self.metrics().tree.parent_switched.inc(); + self.metrics().tree.parent_switches.inc(); info!( new_parent = %self.peer_display_name(&new_parent), @@ -271,7 +276,7 @@ impl Node { "Parent switched, invalidated downstream coord cache entries, announcing to all peers" ); if flap_dampened { - self.stats_mut().tree.flap_dampened += 1; + self.metrics().tree.flap_dampened.inc(); warn!("Flap dampening engaged: excessive parent switches detected"); } @@ -283,19 +288,23 @@ impl Node { } else if !self.tree_state.is_root() && self.tree_state.should_be_root() { // Self is the smallest visible NodeAddr — promote to root rather // than continuing to advertise a stale ancestry rooted elsewhere. + // Clone identity up front to avoid a split borrow against the + // &mut self.tree_state / &mut self.coord_cache calls below (cold path). + let our_identity = self.identity().clone(); self.tree_state.become_root(); - if let Err(e) = self.tree_state.sign_declaration(&self.identity) { + if let Err(e) = self.tree_state.sign_declaration(&our_identity) { warn!(error = %e, "Failed to sign self-root declaration"); - self.stats_mut() - .record_reject(RejectReason::Tree(TreeReject::OutboundSignFailed)); + self.metrics() + .tree + .record_reject(TreeReject::OutboundSignFailed); return; } // Surgical invalidation — see CoordCache::invalidate_other_roots doc. self.coord_cache - .invalidate_other_roots(self.identity.node_addr()); + .invalidate_other_roots(our_identity.node_addr()); self.reset_discovery_backoff(); - self.stats_mut().tree.parent_switched += 1; - self.stats_mut().tree.parent_switches += 1; + self.metrics().tree.parent_switched.inc(); + self.metrics().tree.parent_switches.inc(); info!( new_root = %self.tree_state.root(), "Self-promoted to root: smallest visible NodeAddr" @@ -308,9 +317,9 @@ impl Node { { // Check for loop: if parent's ancestry now contains us, drop parent if let Some(parent_coords) = self.tree_state.peer_coords(from) - && parent_coords.contains(self.identity.node_addr()) + && parent_coords.contains(self.identity().node_addr()) { - self.stats_mut().tree.loop_detected += 1; + self.metrics().tree.loop_detected.inc(); warn!( parent = %self.peer_display_name(from), "Parent ancestry contains us — loop detected, dropping parent" @@ -322,16 +331,20 @@ impl Node { .map(|(addr, peer)| (*addr, peer.link_cost())) .collect(); if self.tree_state.handle_parent_lost(&peer_costs) { - if let Err(e) = self.tree_state.sign_declaration(&self.identity) { + // Clone identity up front to avoid a split borrow against the + // &mut self.tree_state / &mut self.coord_cache calls below (cold path). + let our_identity = self.identity().clone(); + if let Err(e) = self.tree_state.sign_declaration(&our_identity) { warn!(error = %e, "Failed to sign declaration after loop detection"); - self.stats_mut() - .record_reject(RejectReason::Tree(TreeReject::OutboundSignFailed)); + self.metrics() + .tree + .record_reject(TreeReject::OutboundSignFailed); return; } // handle_parent_lost may promote to root OR find new parent; // cover both invalidation classes. self.coord_cache - .invalidate_via_node(self.identity.node_addr()); + .invalidate_via_node(our_identity.node_addr()); self.coord_cache .invalidate_other_roots(self.tree_state.root()); self.reset_discovery_backoff(); @@ -362,24 +375,28 @@ impl Node { .map(|d| d.as_secs()) .unwrap_or(0); + // Clone identity up front to avoid a split borrow against the + // &mut self.tree_state / &mut self.coord_cache calls below (cold path). + let our_identity = self.identity().clone(); self.tree_state.set_parent(*from, new_seq, timestamp); self.tree_state.recompute_coords(); - if let Err(e) = self.tree_state.sign_declaration(&self.identity) { + if let Err(e) = self.tree_state.sign_declaration(&our_identity) { warn!(error = %e, "Failed to sign declaration after parent update"); - self.stats_mut() - .record_reject(RejectReason::Tree(TreeReject::OutboundSignFailed)); + self.metrics() + .tree + .record_reject(TreeReject::OutboundSignFailed); return; } // Surgical invalidation — see CoordCache::invalidate_via_node doc. self.coord_cache - .invalidate_via_node(self.identity.node_addr()); + .invalidate_via_node(our_identity.node_addr()); self.reset_discovery_backoff(); let new_addrs: Vec = self.tree_state.my_coords().node_addrs().copied().collect(); if old_addrs != new_addrs { - self.stats_mut().tree.ancestry_changed += 1; + self.metrics().tree.ancestry_changed.inc(); info!( parent = %self.peer_display_name(from), old_root = %old_root, @@ -420,7 +437,7 @@ impl Node { /// `reeval_interval_secs`. When a better parent is found, follows /// the same switch flow as TreeAnnounce-triggered switches. async fn check_periodic_parent_reeval(&mut self) { - let interval_secs = self.config.node.tree.reeval_interval_secs; + let interval_secs = self.config().node.tree.reeval_interval_secs; if interval_secs == 0 { return; } @@ -455,21 +472,25 @@ impl Node { .map(|d| d.as_secs()) .unwrap_or(0); + // Clone identity up front to avoid a split borrow against the + // &mut self.tree_state / &mut self.coord_cache calls below (cold path). + let our_identity = self.identity().clone(); let flap_dampened = self.tree_state.set_parent(new_parent, new_seq, timestamp); self.tree_state.recompute_coords(); - if let Err(e) = self.tree_state.sign_declaration(&self.identity) { + if let Err(e) = self.tree_state.sign_declaration(&our_identity) { warn!(error = %e, "Failed to sign declaration after periodic parent re-eval"); - self.stats_mut() - .record_reject(RejectReason::Tree(TreeReject::OutboundSignFailed)); + self.metrics() + .tree + .record_reject(TreeReject::OutboundSignFailed); return; } // Surgical invalidation — see CoordCache::invalidate_via_node doc. self.coord_cache - .invalidate_via_node(self.identity.node_addr()); + .invalidate_via_node(our_identity.node_addr()); self.reset_discovery_backoff(); - self.stats_mut().tree.parent_switched += 1; - self.stats_mut().tree.parent_switches += 1; + self.metrics().tree.parent_switched.inc(); + self.metrics().tree.parent_switches.inc(); info!( new_parent = %self.peer_display_name(&new_parent), @@ -480,7 +501,7 @@ impl Node { "Parent switched via periodic cost re-evaluation" ); if flap_dampened { - self.stats_mut().tree.flap_dampened += 1; + self.metrics().tree.flap_dampened.inc(); warn!("Flap dampening engaged: excessive parent switches detected"); } @@ -489,19 +510,23 @@ impl Node { let all_peers: Vec = self.peers.keys().copied().collect(); self.bloom_state.mark_all_updates_needed(all_peers); } else if !self.tree_state.is_root() && self.tree_state.should_be_root() { + // Clone identity up front to avoid a split borrow against the + // &mut self.tree_state / &mut self.coord_cache calls below (cold path). + let our_identity = self.identity().clone(); self.tree_state.become_root(); - if let Err(e) = self.tree_state.sign_declaration(&self.identity) { + if let Err(e) = self.tree_state.sign_declaration(&our_identity) { warn!(error = %e, "Failed to sign self-root declaration in periodic reeval"); - self.stats_mut() - .record_reject(RejectReason::Tree(TreeReject::OutboundSignFailed)); + self.metrics() + .tree + .record_reject(TreeReject::OutboundSignFailed); return; } // Surgical invalidation — see CoordCache::invalidate_other_roots doc. self.coord_cache - .invalidate_other_roots(self.identity.node_addr()); + .invalidate_other_roots(our_identity.node_addr()); self.reset_discovery_backoff(); - self.stats_mut().tree.parent_switched += 1; - self.stats_mut().tree.parent_switches += 1; + self.metrics().tree.parent_switched.inc(); + self.metrics().tree.parent_switches.inc(); info!( new_root = %self.tree_state.root(), trigger = "periodic", @@ -542,7 +567,7 @@ impl Node { self.tree_state.remove_peer(node_addr); if was_parent { - self.stats_mut().tree.parent_losses += 1; + self.metrics().tree.parent_losses.inc(); let peer_costs: HashMap = self .peers .iter() @@ -551,11 +576,14 @@ impl Node { .collect(); let changed = self.tree_state.handle_parent_lost(&peer_costs); if changed { - // Re-sign the new declaration - if let Err(e) = self.tree_state.sign_declaration(&self.identity) { + // Re-sign the new declaration. Clone identity to avoid a split + // borrow against the &mut self.tree_state receiver (cold path). + let our_identity = self.identity().clone(); + if let Err(e) = self.tree_state.sign_declaration(&our_identity) { warn!(error = %e, "Failed to sign declaration after parent loss"); - self.stats_mut() - .record_reject(RejectReason::Tree(TreeReject::OutboundSignFailed)); + self.metrics() + .tree + .record_reject(TreeReject::OutboundSignFailed); } info!( new_root = %self.tree_state.root(),