laantungir/fips
1
0
Fork 0
mirror of https://github.com/jmcorgan/fips.git synced 2026-08-11 17:17:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

node: extract immutable state into a shared context and atomic metric registry

Browse Source 
Store node counters in an atomic metric registry read through &self, and
introduce a shared NodeContext bundle holding the effectively-immutable
fields (config, identity, startup epoch, capability limits). Source the
immutable config and identity reads across the receive hot path, the
handshake/session/mmp/encrypted state machines, and the discovery, tree,
bloom, retry, and lifecycle modules through the context accessors rather
than direct field reads. The Node fields and the context are rebuilt in
lockstep at every mutation site.
This commit is contained in:
Johnathan Corgan
2026-06-02 13:05:03 +00:00
parent 2d0e8de8c8
commit 08b8b3908e
28 changed files with 1140 additions and 957 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/node/handlers/discovery.rs
+44 -42
View File
@@ -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));
}
src/node/handlers/forwarding.rs
+22 -32
View File
@@ -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"
src/node/handlers/handshake.rs
+16 -11
View File
@@ -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);
}
src/node/handlers/mmp.rs
+24 -16
View File
@@ -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);
src/node/handlers/rekey.rs
+17 -17
View File
@@ -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<NodeAddr> = 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<u8>)> = 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!(
src/node/handlers/rx_loop.rs
+3 -3
View File
@@ -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::<crate::control::ControlMessage>(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) {
src/node/handlers/session.rs
+35 -34
View File
@@ -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(&timestamp_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<u8>) {
// 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);
src/node/handlers/timeout.rs
+10 -10
View File
@@ -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<LinkId> = 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<crate::NodeAddr> = 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
}