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

Merge refactor-sans-io: FMP sans-IO connection-lifecycle on the next line

Browse Source
This commit is contained in:
Johnathan Corgan
2026-07-07 07:32:02 +00:00
parent 78eadbcd65 4802792e38
commit 3b46bbe17d
40 changed files with 3877 additions and 2165 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/node/bloom.rs
+3 -3
View File
@@ -24,7 +24,7 @@ impl Node {
let mut filters = HashMap::new();
for (addr, peer) in &self.peers {
if self.is_tree_peer(addr)
&& peer.peer_profile() == crate::protocol::NodeProfile::Full
&& peer.peer_profile() == crate::proto::fmp::NodeProfile::Full
&& let Some(filter) = peer.inbound_filter()
{
filters.insert(*addr, filter.clone());
@@ -179,7 +179,7 @@ impl Node {
/// Non-routing nodes do not send filters (they receive only).
pub(super) async fn send_pending_filter_announces(&mut self) {
// Non-routing and leaf nodes don't send bloom filters
if self.node_profile() != crate::protocol::NodeProfile::Full {
if self.node_profile() != crate::proto::fmp::NodeProfile::Full {
return;
}
@@ -402,7 +402,7 @@ impl Node {
/// and marks all peers for update.
fn check_adaptive_sizing(&mut self) {
// Only Full nodes participate in filter sizing
if self.node_profile() != crate::protocol::NodeProfile::Full {
if self.node_profile() != crate::proto::fmp::NodeProfile::Full {
return;
}
src/node/context.rs
+1 -1
View File
@@ -15,7 +15,7 @@
use std::sync::Arc;
use crate::protocol::NodeProfile;
use crate::proto::fmp::NodeProfile;
use crate::{Config, Identity};
/// Effectively-immutable `Node` state, shared via `Arc<NodeContext>`.
src/node/handlers/discovery.rs
+2 -2
View File
@@ -39,13 +39,13 @@ impl crate::proto::discovery::RoutingView for NodeRoutingView<'_> {
.collect()
}
fn node_is_leaf(&self) -> bool {
self.node.node_profile() == crate::protocol::NodeProfile::Leaf
self.node.node_profile() == crate::proto::fmp::NodeProfile::Leaf
}
fn peer_is_full(&self, addr: &NodeAddr) -> bool {
self.node
.peers
.get(addr)
.is_some_and(|peer| peer.peer_profile() == crate::protocol::NodeProfile::Full)
.is_some_and(|peer| peer.peer_profile() == crate::proto::fmp::NodeProfile::Full)
}
fn peer_meets_mtu(&self, addr: &NodeAddr, min_mtu: u16) -> bool {
self.node
src/node/handlers/dispatch.rs
+1 -1
View File
@@ -77,7 +77,7 @@ impl Node {
/// entries — other removal paths (link-dead, decrypt failure, peer
/// restart) all schedule reconnect.
pub(in crate::node) fn handle_disconnect(&mut self, from: &NodeAddr, payload: &[u8]) {
let disconnect = match crate::protocol::Disconnect::decode(payload) {
let disconnect = match crate::proto::fmp::Disconnect::decode(payload) {
Ok(msg) => msg,
Err(e) => {
debug!(from = %self.peer_display_name(from), error = %e, "Malformed disconnect message");
src/node/handlers/handshake.rs
+275 -376
View File
@@ -10,8 +10,11 @@ use crate::node::acl::PeerAclContext;
use crate::node::reject::{HandshakeReject, RejectReason};
use crate::node::wire::{Msg1Header, Msg2Header, Msg3Header, build_msg2, build_msg3};
use crate::node::{Node, NodeError};
use crate::peer::{ActivePeer, PeerConnection, PromotionResult, cross_connection_winner};
use crate::protocol::{Disconnect, DisconnectReason, NegotiationPayload};
use crate::peer::{ActivePeer, PeerConnection, PromotionResult};
use crate::proto::fmp::{
Disconnect, DisconnectReason, EstablishSnapshot, InboundDecision, InboundReject,
NegotiationPayload, WireOutcome, cross_connection_winner, decide_fmp_negotiation,
};
use crate::transport::{Link, LinkDirection, LinkId, ReceivedPacket};
use std::time::Duration;
use tracing::{debug, info, warn};
@@ -886,11 +889,15 @@ impl Node {
let link_id = match self.pending_inbound.remove(&key) {
Some(id) => id,
None => {
// Check if this is a rekey msg3 for an active peer.
// handle_rekey_msg3 records its own UnknownConnection or
// BadState classification depending on whether a matching
// rekey-responder slot is found.
self.handle_rekey_msg3(&packet, &header).await;
// No pending inbound handshake matches this msg3. The live
// rekey-responder path completes via pending_inbound above, so
// a miss here is an unknown connection.
debug!(
receiver_idx = %header.receiver_idx,
"No pending inbound or rekey state for msg3"
);
self.stats_mut()
.record_reject(RejectReason::Handshake(HandshakeReject::UnknownConnection));
return;
}
};
@@ -1039,310 +1046,275 @@ impl Node {
// debug rather than warn (expected policy rejection, not a fault).
let our_index = our_index.unwrap_or(header.receiver_idx);
// Identity-based restart/rekey detection.
// Identity-based restart/rekey/cross-connection classification.
//
// Now that we know the initiator's identity from msg3, perform the
// same checks that the old handle_msg1 used to do after decrypting msg1.
if let Some(existing_peer) = self.peers.get(&peer_node_addr) {
let new_epoch = remote_epoch;
let existing_epoch = existing_peer.remote_epoch();
// Now that we know the initiator's identity from msg3, classify this
// inbound handshake against any existing active peer. The classification
// tree is the pure `Fmp::establish_inbound` decision; each effect it
// selects is driven shell-side below, preserving the pre-refactor per-
// branch ordering and cleanup exactly. The snapshot resolves the one
// clock read (session age) and the config-derived rekey floor up front.
//
// The rekey age floor sits BELOW the minimum possible rekey interval, or
// jittered rekeys are wrongly rejected. It bounds both the
// cross-connection branch (`< floor` -> initial cross-connection) and the
// rekey-responder branch (`>= floor` -> rekey), so the two partition
// cleanly; see the pre-refactor commentary retained on the decision.
let our_node_addr = *self.identity().node_addr();
let rekey_enabled = self.config().node.rekey.enabled;
let rekey_age_floor_secs = {
let min_interval = self
.config()
.node
.rekey
.after_secs
.saturating_sub(crate::node::REKEY_JITTER_SECS.unsigned_abs());
min_interval.saturating_sub(5).max(5)
};
let wire = WireOutcome {
peer_node_addr,
remote_epoch,
};
let snap = match self.peers.get(&peer_node_addr) {
Some(existing_peer) => EstablishSnapshot {
has_existing_peer: true,
existing_peer_epoch: existing_peer.remote_epoch(),
existing_session_age_secs: existing_peer
.session_established_at()
.elapsed()
.as_secs(),
has_session: existing_peer.has_session(),
is_healthy: existing_peer.is_healthy(),
pending_new_session: existing_peer.pending_new_session().is_some(),
rekey_in_progress: existing_peer.rekey_in_progress(),
existing_msg2: existing_peer.handshake_msg2().map(|m| m.to_vec()),
different_link: existing_peer.link_id() != link_id,
rekey_enabled,
rekey_age_floor_secs,
our_node_addr,
},
None => EstablishSnapshot {
has_existing_peer: false,
existing_peer_epoch: None,
existing_session_age_secs: 0,
has_session: false,
is_healthy: false,
pending_new_session: false,
rekey_in_progress: false,
existing_msg2: None,
different_link: false,
rekey_enabled,
rekey_age_floor_secs,
our_node_addr,
},
};
match (existing_epoch, new_epoch) {
(Some(existing), Some(new)) if existing != new => {
// Epoch mismatch — peer restarted. Tear down stale session.
debug!(
peer = %self.peer_display_name(&peer_node_addr),
"Peer restart detected (epoch mismatch), removing stale session"
);
self.remove_active_peer(&peer_node_addr);
let now_ms = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as u64)
.unwrap_or(0);
self.schedule_reconnect(peer_node_addr, now_ms);
// Fall through to process as new connection
}
_ => {
// Same epoch (or no epoch stored).
let session_age_secs =
existing_peer.session_established_at().elapsed().as_secs();
// The minimum plausible age for an inbound XX handshake to
// be a scheduled REKEY rather than an initial-handshake
// cross-connection. Derived from the configured interval
// and jitter so it tracks the real minimum rekey spacing
// (see the rekey-responder gate below, which uses the same
// value). A session at least this old that receives an
// inbound msg3 on a different link is a rekey, NOT an
// initial cross-connection.
let rekey_age_floor_secs = {
let min_interval = self
.config()
.node
.rekey
.after_secs
.saturating_sub(crate::node::REKEY_JITTER_SECS.unsigned_abs());
min_interval.saturating_sub(5).max(5)
};
// Simultaneous-init cross-connection (msg2-then-msg3 ordering).
//
// When both sides initiate XX in parallel (typical in
// bootstrap-handoff after Nostr UDP punch), each side runs
// two handshakes concurrently — its own outbound paired with
// the peer's inbound, and the peer's outbound paired with our
// inbound. If our outbound's msg2 arrives before the peer's
// outbound's msg3, handle_msg2 promoted our outbound under
// the "Normal path" (peers_contains_key was false). Now
// msg3 arrives for the unrelated inbound link with the peer
// already promoted at the same epoch — apply the same
// tie-breaker handle_msg2 uses for the inverse ordering, so
// both sides converge on a single Noise session pair.
//
// CRITICAL (jitter × XX rekey): the upper age bound MUST sit
// below the rekey floor. An initial cross-connection always
// resolves within ~1 RTT of promotion (both handshakes race
// in the same sub-second burst); a session old enough to be
// rekeying that receives a concurrent rekey msg3 must NOT be
// routed here. The old fixed `< 30` bound overlapped the
// jittered rekey floor (as low as 15s): under jitter a recent
// cutover resets `session_established_at`, so a concurrent
// rekey msg3 (always on a different temp link_id) landed in
// this branch and, on the "our outbound wins" side, the
// peer's rekey session was DISCARDED (index freed, no pending
// slot) — yet the peer cut over to it regardless, leaving the
// discarding node unable to decrypt the peer until the 30s
// dead-timer fired (Phase-5 link death, green crypto). Gating
// on the rekey floor makes any rekey-aged msg3 fall through to
// the rekey-responder path below, which converges both sides
// (dual-init tie-break) AND installs a `pending` slot.
if existing_peer.link_id() != link_id && session_age_secs < rekey_age_floor_secs
{
let our_inbound_wins = cross_connection_winner(
self.identity().node_addr(),
&peer_node_addr,
false, // this connection is inbound
);
if our_inbound_wins {
// Larger node side: swap to the inbound session so
// it pairs with the peer's kept outbound session.
let inbound_session = match self
.connections
.get_mut(&link_id)
.and_then(|c| c.take_session())
{
Some(s) => s,
None => {
self.connections.remove(&link_id);
self.remove_link(&link_id);
self.stats_mut().record_reject(RejectReason::Handshake(
HandshakeReject::BadState,
));
return;
}
};
if let Some(peer) = self.peers.get_mut(&peer_node_addr) {
let old_our_index = peer.replace_session(
inbound_session,
our_index,
header.sender_idx,
);
let Some(transport_id) = peer.transport_id() else {
self.connections.remove(&link_id);
self.remove_link(&link_id);
self.stats_mut().record_reject(RejectReason::Handshake(
HandshakeReject::BadState,
));
return;
};
if let Some(old_idx) = old_our_index {
self.peers_by_index
.remove(&(transport_id, old_idx.as_u32()));
let _ = self.index_allocator.free(old_idx);
}
self.peers_by_index
.insert((transport_id, our_index.as_u32()), peer_node_addr);
debug!(
peer = %self.peer_display_name(&peer_node_addr),
new_our_index = %our_index,
new_their_index = %header.sender_idx,
"Simultaneous-init (msg3): swapped to inbound session (our inbound wins)"
);
}
} else {
// Smaller node side: keep the existing outbound
// session, drop the inbound's allocated index.
let _ = self.index_allocator.free(our_index);
debug!(
peer = %self.peer_display_name(&peer_node_addr),
"Simultaneous-init (msg3): keeping outbound session (our outbound wins)"
);
}
self.connections.remove(&link_id);
self.remove_link(&link_id);
return;
match self.fmp.establish_inbound(&snap, &wire) {
InboundDecision::Reject {
reason: InboundReject::DualRekeyWon,
} => {
// Dual-init rekey tie-break: we win (smaller addr), drop their msg3.
info!(
peer = %self.peer_display_name(&peer_node_addr),
our_addr = %our_node_addr,
their_addr = %peer_node_addr,
rekey_in_progress = snap.rekey_in_progress,
pending_new_session = snap.pending_new_session,
"rekey-msg3 tie-break: we win (smaller addr), drop their msg3"
);
self.connections.remove(&link_id);
self.links.remove(&link_id);
self.stats_mut()
.record_reject(RejectReason::Handshake(HandshakeReject::BadState));
return;
}
InboundDecision::ResendMsg2 { msg2 } => {
// Not a rekey — duplicate handshake from same epoch. Resend
// stored msg2, leaving the active peer untouched.
if let Some(msg2) = msg2
&& let Some(transport) = self.transports.get(&packet.transport_id)
{
match transport.send(&packet.remote_addr, &msg2).await {
Ok(_) => debug!(
peer = %self.peer_display_name(&peer_node_addr),
"Resent msg2 for duplicate handshake (same epoch)"
),
Err(e) => debug!(
peer = %self.peer_display_name(&peer_node_addr),
error = %e,
"Failed to resend msg2"
),
}
// Check for rekey: session must be old enough that an
// inbound XX handshake is plausibly a scheduled rekey
// rather than a fresh duplicate/restart.
//
// The floor (`rekey_age_floor_secs`, computed above) sits
// BELOW the minimum possible rekey interval, or jittered
// rekeys are wrongly rejected. The initiator's effective
// interval is `after_secs - REKEY_JITTER_SECS` at its lowest
// (20s for 35s ± 15s). A fixed 30s floor exceeds that 20s
// minimum, so under jitter the responder rejects a legitimate
// rekey as a "duplicate handshake" (resends msg2), the
// initiator cuts over anyway, and the endpoints diverge by
// one epoch → receiver starves → 30s link-dead. The same
// floor also bounds the cross-connection branch above, so the
// two paths partition cleanly: `< floor` → initial
// cross-connection, `>= floor` → rekey responder.
if self.config().node.rekey.enabled
&& existing_peer.has_session()
&& existing_peer.is_healthy()
&& session_age_secs >= rekey_age_floor_secs
}
self.connections.remove(&link_id);
self.links.remove(&link_id);
return;
}
InboundDecision::CrossConnect {
peer,
our_inbound_wins,
} => {
debug_assert_eq!(peer, peer_node_addr);
// Simultaneous-init cross-connection (msg2-then-msg3 ordering):
// apply the same tie-breaker handle_msg2 uses for the inverse
// ordering so both sides converge on a single Noise session pair.
if our_inbound_wins {
// Larger node side: swap to the inbound session so it pairs
// with the peer's kept outbound session.
let inbound_session = match self
.connections
.get_mut(&link_id)
.and_then(|c| c.take_session())
{
// Dual-initiation detection: both sides initiated rekey
// simultaneously. Two states can reach this point:
// - rekey_in_progress=true: both sides still mid-handshake
// - pending_new_session=Some && !rekey_in_progress: both
// sides already completed their initiator path
// (set_pending_session cleared rekey_in_progress)
// The IK fix only caught the first state; the XX three-message
// handshake widens the window so the second state is reached
// when both sides' set_pending_session runs before either's
// msg3 lands at the peer. Apply the smaller-NodeAddr
// tie-breaker uniformly in both states so both sides converge
// on a single Noise session post-cutover.
if existing_peer.rekey_in_progress()
|| existing_peer.pending_new_session().is_some()
{
let our_addr = self.identity().node_addr();
if our_addr < &peer_node_addr {
// We win — keep our session, drop their msg3.
info!(
peer = %self.peer_display_name(&peer_node_addr),
our_addr = %our_addr,
their_addr = %peer_node_addr,
rekey_in_progress = existing_peer.rekey_in_progress(),
pending_new_session = existing_peer.pending_new_session().is_some(),
"rekey-msg3 tie-break: we win (smaller addr), drop their msg3"
);
self.connections.remove(&link_id);
self.links.remove(&link_id);
self.stats_mut().record_reject(RejectReason::Handshake(
HandshakeReject::BadState,
));
return;
}
// We lose — abandon our rekey/pending, fall through as responder.
// abandon_rekey clears both rekey_in_progress and any pending
// session state, returning whichever index needs freeing.
info!(
peer = %self.peer_display_name(&peer_node_addr),
our_addr = %our_addr,
their_addr = %peer_node_addr,
rekey_in_progress = existing_peer.rekey_in_progress(),
pending_new_session = existing_peer.pending_new_session().is_some(),
"rekey-msg3 tie-break: we lose (larger addr), abandon ours"
);
if let Some(peer) = self.peers.get_mut(&peer_node_addr)
&& let Some(idx) = peer.abandon_rekey()
{
if let Some(tid) = peer.transport_id() {
self.peers_by_index.remove(&(tid, idx.as_u32()));
self.pending_outbound.remove(&(tid, idx.as_u32()));
}
let _ = self.index_allocator.free(idx);
}
// Fall through to respond as responder
Some(s) => s,
None => {
self.connections.remove(&link_id);
self.remove_link(&link_id);
self.stats_mut()
.record_reject(RejectReason::Handshake(HandshakeReject::BadState));
return;
}
// Rekey: process as responder, store new session as pending
let noise_session = {
let Some(conn) = self.connections.get_mut(&link_id) else {
warn!(link_id = %link_id, "Connection removed during rekey msg3 processing");
self.links.remove(&link_id);
self.stats_mut().record_reject(RejectReason::Handshake(
HandshakeReject::UnknownConnection,
));
return;
};
conn.take_session()
};
if let Some(peer) = self.peers.get_mut(&peer_node_addr) {
let old_our_index =
peer.replace_session(inbound_session, our_index, header.sender_idx);
let Some(transport_id) = peer.transport_id() else {
self.connections.remove(&link_id);
self.remove_link(&link_id);
self.stats_mut()
.record_reject(RejectReason::Handshake(HandshakeReject::BadState));
return;
};
let our_new_index = our_index;
let noise_session = match noise_session {
Some(s) => s,
None => {
warn!("Rekey msg3: no session from handshake");
self.connections.remove(&link_id);
self.links.remove(&link_id);
self.stats_mut().record_reject(RejectReason::Handshake(
HandshakeReject::BadState,
));
return;
}
};
// Store pending session on the existing peer
if let Some(peer) = self.peers.get_mut(&peer_node_addr) {
peer.set_pending_session(
noise_session,
our_new_index,
header.sender_idx,
);
peer.record_peer_rekey();
if let Some(old_idx) = old_our_index {
self.peers_by_index
.remove(&(transport_id, old_idx.as_u32()));
let _ = self.index_allocator.free(old_idx);
}
// Register new index in peers_by_index
self.peers_by_index.insert(
(packet.transport_id, our_new_index.as_u32()),
peer_node_addr,
);
// Clean up: remove the temporary connection/link.
// Do NOT remove addr_to_link — the entry must remain pointing
// to the original link.
self.connections.remove(&link_id);
self.links.remove(&link_id);
self.peers_by_index
.insert((transport_id, our_index.as_u32()), peer_node_addr);
debug!(
peer = %self.peer_display_name(&peer_node_addr),
our_addr = %self.identity().node_addr(),
new_our_index = %our_new_index,
new_our_index = %our_index,
new_their_index = %header.sender_idx,
"rekey-msg3 responder: pending session set, awaiting K-bit cutover"
"Simultaneous-init (msg3): swapped to inbound session (our inbound wins)"
);
}
} else {
// Smaller node side: keep the existing outbound session, drop
// the inbound's allocated index.
let _ = self.index_allocator.free(our_index);
debug!(
peer = %self.peer_display_name(&peer_node_addr),
"Simultaneous-init (msg3): keeping outbound session (our outbound wins)"
);
}
self.connections.remove(&link_id);
self.remove_link(&link_id);
return;
}
InboundDecision::RekeyRespond {
peer,
abandon_first,
} => {
debug_assert_eq!(peer, peer_node_addr);
if abandon_first {
// We lose — abandon our rekey/pending, fall through as
// responder. abandon_rekey clears both rekey_in_progress and
// any pending session state, returning whichever index needs
// freeing.
info!(
peer = %self.peer_display_name(&peer_node_addr),
our_addr = %our_node_addr,
their_addr = %peer_node_addr,
rekey_in_progress = snap.rekey_in_progress,
pending_new_session = snap.pending_new_session,
"rekey-msg3 tie-break: we lose (larger addr), abandon ours"
);
if let Some(peer) = self.peers.get_mut(&peer_node_addr)
&& let Some(idx) = peer.abandon_rekey()
{
if let Some(tid) = peer.transport_id() {
self.peers_by_index.remove(&(tid, idx.as_u32()));
self.pending_outbound.remove(&(tid, idx.as_u32()));
}
let _ = self.index_allocator.free(idx);
}
}
// Rekey: process as responder, store new session as pending.
let noise_session = {
let Some(conn) = self.connections.get_mut(&link_id) else {
warn!(link_id = %link_id, "Connection removed during rekey msg3 processing");
self.links.remove(&link_id);
self.stats_mut().record_reject(RejectReason::Handshake(
HandshakeReject::UnknownConnection,
));
return;
};
conn.take_session()
};
let our_new_index = our_index;
let noise_session = match noise_session {
Some(s) => s,
None => {
warn!("Rekey msg3: no session from handshake");
self.connections.remove(&link_id);
self.links.remove(&link_id);
self.stats_mut()
.record_reject(RejectReason::Handshake(HandshakeReject::BadState));
return;
}
};
// Not a rekey — duplicate handshake from same epoch.
// Resend stored msg2.
if let Some(msg2) = existing_peer.handshake_msg2().map(|m| m.to_vec())
&& let Some(transport) = self.transports.get(&packet.transport_id)
{
match transport.send(&packet.remote_addr, &msg2).await {
Ok(_) => debug!(
peer = %self.peer_display_name(&peer_node_addr),
"Resent msg2 for duplicate handshake (same epoch)"
),
Err(e) => debug!(
peer = %self.peer_display_name(&peer_node_addr),
error = %e,
"Failed to resend msg2"
),
}
}
self.connections.remove(&link_id);
self.links.remove(&link_id);
return;
// Store pending session on the existing peer
if let Some(peer) = self.peers.get_mut(&peer_node_addr) {
peer.set_pending_session(noise_session, our_new_index, header.sender_idx);
peer.record_peer_rekey();
}
// Register new index in peers_by_index
self.peers_by_index.insert(
(packet.transport_id, our_new_index.as_u32()),
peer_node_addr,
);
// Clean up: remove the temporary connection/link.
// Do NOT remove addr_to_link — the entry must remain pointing
// to the original link.
self.connections.remove(&link_id);
self.links.remove(&link_id);
debug!(
peer = %self.peer_display_name(&peer_node_addr),
our_addr = %self.identity().node_addr(),
new_our_index = %our_new_index,
new_their_index = %header.sender_idx,
"rekey-msg3 responder: pending session set, awaiting K-bit cutover"
);
return;
}
InboundDecision::RestartThenPromote { peer } => {
// Epoch mismatch — peer restarted. Tear down stale session, then
// fall through to promote the fresh connection in its place.
debug!(
peer = %self.peer_display_name(&peer),
"Peer restart detected (epoch mismatch), removing stale session"
);
self.remove_active_peer(&peer);
let now_ms = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as u64)
.unwrap_or(0);
self.schedule_reconnect(peer, now_ms);
// Fall through to process as new connection.
}
InboundDecision::Promote => {
// Net-new inbound (or the post-restart re-promote): fall through
// to promote_connection, whose late max-peers cap and
// cross-connection won/lost handling stay shell-side.
}
}
@@ -1460,77 +1432,6 @@ impl Node {
}
}
/// Handle a rekey msg3 for an already-active peer.
///
/// When a rekey is in progress (responder side), the ActivePeer holds the
/// handshake state. This processes msg3 to complete the rekey responder
/// handshake.
async fn handle_rekey_msg3(&mut self, packet: &ReceivedPacket, header: &Msg3Header) {
// Look for a peer expecting a rekey msg3 as responder.
// The responder's rekey handshake state is stored after processing
// the initiator's rekey msg1+msg2 exchange via the existing link.
let peer_addr = self.peers.iter().find_map(|(addr, peer)| {
if peer.has_rekey_responder_handshake()
&& peer.rekey_responder_our_index() == Some(header.receiver_idx)
{
Some(*addr)
} else {
None
}
});
let peer_node_addr = match peer_addr {
Some(addr) => addr,
None => {
debug!(
receiver_idx = %header.receiver_idx,
"No pending inbound or rekey state for msg3"
);
self.stats_mut()
.record_reject(RejectReason::Handshake(HandshakeReject::UnknownConnection));
return;
}
};
let display_name = self.peer_display_name(&peer_node_addr);
let noise_msg3 = &packet.data[header.noise_msg3_offset..];
if let Some(peer) = self.peers.get_mut(&peer_node_addr) {
match peer.complete_rekey_msg3(noise_msg3) {
Ok(session) => {
let our_index = peer
.rekey_responder_our_index()
.unwrap_or(header.receiver_idx);
peer.set_pending_session(session, our_index, header.sender_idx);
peer.record_peer_rekey();
if let Some(transport_id) = peer.transport_id() {
self.peers_by_index
.insert((transport_id, our_index.as_u32()), peer_node_addr);
}
debug!(
peer = %display_name,
our_addr = %self.identity().node_addr(),
new_our_index = %our_index,
new_their_index = %header.sender_idx,
"rekey-msg3 responder (existing link): pending session set, awaiting K-bit cutover"
);
}
Err(e) => {
warn!(
peer = %display_name,
error = %e,
"Rekey msg3 processing failed"
);
peer.clear_rekey_responder();
self.stats_mut()
.record_reject(RejectReason::Handshake(HandshakeReject::BadState));
}
}
}
}
/// Promote a connection to active peer after successful authentication.
///
/// Handles cross-connection detection and resolution using tie-breaker rules.
@@ -1543,7 +1444,7 @@ impl Node {
) -> Result<PromotionResult, NodeError> {
// Leaf nodes: reject if we already have a peer (single-peer enforcement)
let peer_node_addr_check = *verified_identity.node_addr();
if self.node_profile() == crate::protocol::NodeProfile::Leaf
if self.node_profile() == crate::proto::fmp::NodeProfile::Leaf
&& !self.peers.is_empty()
&& !self.peers.contains_key(&peer_node_addr_check)
{
@@ -1606,7 +1507,7 @@ impl Node {
let remote_epoch = connection.remote_epoch();
let peer_profile = connection
.peer_profile()
.unwrap_or(crate::protocol::NodeProfile::Full);
.unwrap_or(crate::proto::fmp::NodeProfile::Full);
let peer_node_addr = *verified_identity.node_addr();
let is_outbound = connection.is_outbound();
@@ -1700,7 +1601,7 @@ impl Node {
// Non-routing peers don't send filters; include them as
// dependents so our bloom filter advertises their identity.
if peer_profile != crate::protocol::NodeProfile::Full {
if peer_profile != crate::proto::fmp::NodeProfile::Full {
self.bloom_state.add_leaf_dependent(peer_node_addr);
}
@@ -1814,7 +1715,7 @@ impl Node {
// Non-routing peers don't send filters; include them as
// dependents so our bloom filter advertises their identity.
if peer_profile != crate::protocol::NodeProfile::Full {
if peer_profile != crate::proto::fmp::NodeProfile::Full {
self.bloom_state.add_leaf_dependent(peer_node_addr);
}
@@ -1844,15 +1745,13 @@ impl Node {
/// Decodes the payload, validates profile pairing, and stores the
/// results on the PeerConnection.
fn process_fmp_negotiation(
our_profile: crate::protocol::NodeProfile,
our_profile: crate::proto::fmp::NodeProfile,
conn: &mut PeerConnection,
neg_bytes: &[u8],
) -> Result<(), crate::protocol::ProtocolError> {
let their_payload = NegotiationPayload::decode(neg_bytes)?;
// Validate profile pairing (at least one Full)
let their_profile = their_payload.node_profile()?;
NegotiationPayload::validate_profiles(our_profile, their_profile)?;
// The decode -> validate -> profile decision is the pure core split; the
// shell records the result on the connection and logs.
let their_profile = decide_fmp_negotiation(our_profile, neg_bytes)?;
conn.set_negotiation_results(their_profile);
src/node/handlers/rekey.rs
+172 -180
View File
@@ -10,6 +10,7 @@ use crate::node::Node;
use crate::node::reject::{HandshakeReject, RejectReason};
use crate::node::wire::build_msg1;
use crate::noise::HandshakeState;
use crate::proto::fmp::{ConnAction, LifecycleView, PeerSnapshot, RekeyCfg, RekeyResendSnapshot};
use crate::protocol::{SessionDatagram, SessionSetup};
use tracing::{debug, info, trace, warn};
@@ -42,140 +43,119 @@ impl Node {
return;
}
let rekey_after_secs = self.config().node.rekey.after_secs;
let rekey_after_messages = self.config().node.rekey.after_messages;
let cfg = RekeyCfg {
after_secs: self.config().node.rekey.after_secs,
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();
let mut peers_to_drain: Vec<NodeAddr> = Vec::new();
let mut peers_to_rekey: Vec<NodeAddr> = Vec::new();
for (node_addr, peer) in &self.peers {
if !peer.has_session() || !peer.is_healthy() {
continue;
}
// 1. Initiator-side cutover: we completed a rekey and have
// a pending session ready. Cut over on the next tick.
if peer.pending_new_session().is_some() && !peer.rekey_in_progress() {
peers_to_cutover.push(*node_addr);
continue;
}
// 2. Drain window expiry
if peer.is_draining() && peer.drain_expired(DRAIN_WINDOW_SECS) {
peers_to_drain.push(*node_addr);
}
// 3. Rekey trigger
if peer.rekey_in_progress() {
continue;
}
if peer.pending_new_session().is_some() {
// Completed rekey awaiting cutover; don't stack another.
continue;
}
if peer.rekey_msg3_payload().is_some() {
// Initiator already cut over on its timer but is still
// retransmitting msg3 to a responder not yet confirmed on
// the new epoch. Don't start another rekey (which would
// overwrite the retained payload) until this cycle's msg3
// is delivered or its budget exhausted. Mirrors FSP
// check_session_rekey.
continue;
}
if peer.is_rekey_dampened(REKEY_DAMPENING_SECS) {
continue;
}
let elapsed = peer.session_established_at().elapsed().as_secs();
let counter = peer
.noise_session()
.map(|s| s.current_send_counter())
.unwrap_or(0);
// Apply per-session symmetric jitter to desynchronize
// dual-initiation in symmetric-start meshes.
let effective_after_secs =
rekey_after_secs.saturating_add_signed(peer.rekey_jitter_secs());
if elapsed >= effective_after_secs || counter >= rekey_after_messages {
peers_to_rekey.push(*node_addr);
}
}
// Execute cutover for initiator side
for node_addr in peers_to_cutover {
let did_cutover = if let Some(peer) = self.peers.get_mut(&node_addr) {
if let Some(_old_our_index) = peer.cutover_to_new_session() {
// New index was pre-registered in peers_by_index during
// msg2 handling (handshake.rs). Verify, don't duplicate.
debug_assert!(
peer.transport_id().is_some()
&& peer.our_index().is_some()
&& self.peers_by_index.contains_key(&(
peer.transport_id().unwrap(),
peer.our_index().unwrap().as_u32()
)),
"peers_by_index should contain pre-registered new index after cutover"
);
let our_index = peer.our_index();
let their_index = peer.their_index();
info!(
peer = %self.peer_display_name(&node_addr),
our_addr = %self.identity().node_addr(),
their_addr = %node_addr,
our_index = ?our_index,
their_index = ?their_index,
"Rekey cutover complete (initiator), K-bit flipped"
);
true
} else {
false
}
} else {
false
};
// Re-register the new session with the decrypt worker — the
// cache_key (transport_id, our_index) just changed, so the
// old worker entry is stale and every packet on the new
// session would miss the worker's HashMap lookup.
#[cfg(unix)]
if did_cutover {
self.register_decrypt_worker_session(&node_addr);
}
#[cfg(not(unix))]
let _ = did_cutover;
}
// Execute drain completion
for node_addr in peers_to_drain {
// Extract the old index and transport_id under the peer
// borrow, then drop the borrow so the cache_key cleanup
// below can take &mut self for unregister_decrypt_worker_session.
let drained = self
.peers
.get_mut(&node_addr)
.and_then(|peer| peer.complete_drain().map(|idx| (idx, peer.transport_id())));
if let Some((old_our_index, transport_id)) = drained {
if let Some(tid) = transport_id {
let cache_key = (tid, old_our_index.as_u32());
self.peers_by_index.remove(&cache_key);
// The shell snapshots each healthy peer's rekey ages/flags (every clock
// read resolved here); the core decides cutover/drain/trigger with no
// clock, phase-grouped to preserve the pre-refactor execution order.
let snapshots = self.rekey_peers();
for action in self.fmp.poll_rekey(snapshots, &cfg) {
match action {
// Execute cutover for initiator side.
ConnAction::Cutover { peer: node_addr } => {
let did_cutover = if let Some(peer) = self.peers.get_mut(&node_addr) {
if let Some(_old_our_index) = peer.cutover_to_new_session() {
// New index was pre-registered in peers_by_index
// during msg2 handling (handshake.rs).
debug_assert!(
peer.transport_id().is_some()
&& peer.our_index().is_some()
&& self.peers_by_index.contains_key(&(
peer.transport_id().unwrap(),
peer.our_index().unwrap().as_u32()
)),
"peers_by_index should contain pre-registered new index after cutover"
);
let our_index = peer.our_index();
let their_index = peer.their_index();
info!(
peer = %self.peer_display_name(&node_addr),
our_addr = %self.identity().node_addr(),
their_addr = %node_addr,
our_index = ?our_index,
their_index = ?their_index,
"Rekey cutover complete (initiator), K-bit flipped"
);
true
} else {
false
}
} else {
false
};
// Re-register the new session with the decrypt worker — the
// cache_key (transport_id, our_index) just changed, so the
// old worker entry is stale and every packet on the new
// session would miss the worker's HashMap lookup.
#[cfg(unix)]
self.unregister_decrypt_worker_session(cache_key);
if did_cutover {
self.register_decrypt_worker_session(&node_addr);
}
#[cfg(not(unix))]
let _ = did_cutover;
}
let _ = self.index_allocator.free(old_our_index);
trace!(
peer = %self.peer_display_name(&node_addr),
old_index = %old_our_index,
"Drain complete, previous session erased"
);
// Execute drain completion.
ConnAction::Drain { peer: node_addr } => {
// Extract the old index and transport_id under the peer
// borrow, then drop the borrow so the cache_key cleanup
// below can take &mut self for unregister_decrypt_worker_session.
let drained = self.peers.get_mut(&node_addr).and_then(|peer| {
peer.complete_drain().map(|idx| (idx, peer.transport_id()))
});
if let Some((old_our_index, transport_id)) = drained {
if let Some(tid) = transport_id {
let cache_key = (tid, old_our_index.as_u32());
self.peers_by_index.remove(&cache_key);
#[cfg(unix)]
self.unregister_decrypt_worker_session(cache_key);
}
let _ = self.index_allocator.free(old_our_index);
trace!(
peer = %self.peer_display_name(&node_addr),
old_index = %old_our_index,
"Drain complete, previous session erased"
);
}
}
// Initiate a new rekey.
ConnAction::InitiateRekey { peer: node_addr } => {
self.initiate_rekey(&node_addr).await;
}
#[allow(unreachable_patterns)]
_ => {}
}
}
}
// Initiate new rekeys
for node_addr in peers_to_rekey {
self.initiate_rekey(&node_addr).await;
}
/// Snapshot every healthy peer with a session for the rekey decision,
/// pre-computing its monotonic ages and timer predicates so the pure core
/// applies the thresholds without reading a clock (see [`PeerSnapshot`]).
///
/// Lives here, beside the drain/dampening constants and the FSP analog, so
/// the forward-merge onto `next` reconciles rekey timing in one place.
pub(in crate::node) fn rekey_peer_snapshots(&self) -> Vec<PeerSnapshot> {
self.peers
.iter()
.filter(|(_, peer)| peer.has_session() && peer.is_healthy())
.map(|(node_addr, peer)| PeerSnapshot {
addr: *node_addr,
has_pending: peer.pending_new_session().is_some(),
rekey_in_progress: peer.rekey_in_progress(),
is_draining: peer.is_draining(),
drain_expired: peer.drain_expired(DRAIN_WINDOW_SECS),
is_dampened: peer.is_rekey_dampened(REKEY_DAMPENING_SECS),
rekey_msg3_pending: peer.rekey_msg3_payload().is_some(),
elapsed_secs: peer.session_established_at().elapsed().as_secs(),
counter: peer
.noise_session()
.map(|s| s.current_send_counter())
.unwrap_or(0),
jitter_secs: peer.rekey_jitter_secs(),
})
.collect()
}
/// Initiate an outbound rekey to a peer.
@@ -285,62 +265,74 @@ impl Node {
let backoff = self.config().node.rate_limit.handshake_resend_backoff;
let max_resends = self.config().node.rate_limit.handshake_max_resends;
// Collect peers needing action
let mut to_resend: Vec<(NodeAddr, Vec<u8>)> = Vec::new();
let mut to_abandon: Vec<NodeAddr> = Vec::new();
// The shell snapshots each in-flight rekey (resend-due predicate
// resolved here); the core classifies abandon-vs-resend and computes
// the backoff, abandons first.
let candidates = self.rekey_resend_candidates(now_ms);
for action in
self.fmp
.poll_rekey_resends(candidates, now_ms, interval_ms, backoff, max_resends)
{
match action {
// Abandon rekey cycles that exhausted their retransmission budget.
ConnAction::AbandonRekey { peer: node_addr } => {
if let Some(peer) = self.peers.get_mut(&node_addr) {
peer.abandon_rekey();
}
debug!(
peer = %self.peer_display_name(&node_addr),
"FMP rekey aborted: msg1 unconfirmed after max retransmissions, abandoning cycle"
);
}
ConnAction::ResendRekeyMsg1 {
peer: node_addr,
bytes,
next_resend_at_ms,
} => {
let (transport_id, remote_addr) = match self.peers.get(&node_addr) {
Some(p) => match (p.transport_id(), p.current_addr()) {
(Some(tid), Some(addr)) => (tid, addr.clone()),
_ => continue,
},
None => continue,
};
for (node_addr, peer) in &self.peers {
if !peer.rekey_in_progress() || peer.rekey_msg1().is_none() {
continue;
}
if peer.rekey_msg1_resend_count() >= max_resends {
to_abandon.push(*node_addr);
continue;
}
if peer.needs_msg1_resend(now_ms)
&& let Some(msg1) = peer.rekey_msg1()
{
to_resend.push((*node_addr, msg1.to_vec()));
let sent = if let Some(transport) = self.transports.get(&transport_id) {
transport.send(&remote_addr, &bytes).await.is_ok()
} else {
false
};
if sent && let Some(peer) = self.peers.get_mut(&node_addr) {
peer.record_rekey_msg1_resend(next_resend_at_ms);
let count = peer.rekey_msg1_resend_count();
trace!(
peer = %self.peer_display_name(&node_addr),
resend = count,
"Resent rekey msg1"
);
}
}
#[allow(unreachable_patterns)]
_ => {}
}
}
}
// Abandon rekey cycles that exhausted their retransmission budget.
for node_addr in to_abandon {
if let Some(peer) = self.peers.get_mut(&node_addr) {
peer.abandon_rekey();
}
debug!(
peer = %self.peer_display_name(&node_addr),
"FMP rekey aborted: msg1 unconfirmed after max retransmissions, abandoning cycle"
);
}
for (node_addr, msg1_bytes) in to_resend {
let (transport_id, remote_addr) = match self.peers.get(&node_addr) {
Some(p) => match (p.transport_id(), p.current_addr()) {
(Some(tid), Some(addr)) => (tid, addr.clone()),
_ => continue,
},
None => continue,
};
let sent = if let Some(transport) = self.transports.get(&transport_id) {
transport.send(&remote_addr, &msg1_bytes).await.is_ok()
} else {
false
};
if sent && let Some(peer) = self.peers.get_mut(&node_addr) {
let count = peer.rekey_msg1_resend_count() + 1;
let next = now_ms + (interval_ms as f64 * backoff.powi(count as i32)) as u64;
peer.record_rekey_msg1_resend(next);
trace!(
peer = %self.peer_display_name(&node_addr),
resend = count,
"Resent rekey msg1"
);
}
}
/// Snapshot every peer with a rekey handshake in flight (and a stored
/// msg1) for the retransmission decision, pre-evaluating the resend-due
/// predicate against `now_ms` so the core reads no clock.
pub(in crate::node) fn rekey_resend_snapshots(&self, now_ms: u64) -> Vec<RekeyResendSnapshot> {
self.peers
.iter()
.filter(|(_, peer)| peer.rekey_in_progress() && peer.rekey_msg1().is_some())
.map(|(node_addr, peer)| RekeyResendSnapshot {
peer: *node_addr,
resend_count: peer.rekey_msg1_resend_count(),
needs_resend: peer.needs_msg1_resend(now_ms),
msg1: peer.rekey_msg1().unwrap().to_vec(),
})
.collect()
}
/// Retransmit FMP rekey msg3 until the responder is confirmed on the
src/node/handlers/session.rs
+1 -1
View File
@@ -20,10 +20,10 @@ use crate::node::session_wire::{
use crate::node::wire::{ESTABLISHED_HEADER_SIZE, FLAG_KEY_EPOCH, build_established_header};
use crate::node::{Node, NodeError};
use crate::noise::{HANDSHAKE_MSG1_SIZE, HANDSHAKE_MSG2_SIZE, HANDSHAKE_MSG3_SIZE, HandshakeState};
use crate::proto::fmp::NegotiationPayload;
use crate::proto::routing::{CoordsRequired, MtuExceeded, PathBroken};
#[cfg(unix)]
use crate::protocol::LinkMessageType;
use crate::protocol::NegotiationPayload;
#[cfg(unix)]
use crate::protocol::SESSION_DATAGRAM_HEADER_SIZE;
use crate::protocol::{
src/node/handlers/timeout.rs
+111 -66
View File
@@ -3,14 +3,77 @@
use crate::node::Node;
use crate::peer::HandshakeState;
use crate::proto::fmp::{
ConnAction, ConnSnapshot, LifecycleView, PeerSnapshot, RekeyResendSnapshot,
};
use crate::transport::LinkId;
use tracing::{debug, info};
impl LifecycleView for Node {
fn stale_connections(&self, now_ms: u64, timeout_ms: u64) -> Vec<ConnSnapshot> {
self.connections
.iter()
.filter(|(_, conn)| conn.is_timed_out(now_ms, timeout_ms) || conn.is_failed())
.map(|(link_id, conn)| ConnSnapshot {
link: *link_id,
is_outbound: conn.is_outbound(),
retry_addr: conn.expected_identity().map(|id| *id.node_addr()),
resend_count: 0,
msg1: Vec::new(),
})
.collect()
}
fn resend_candidates(&self, now_ms: u64, max_resends: u32) -> Vec<ConnSnapshot> {
self.connections
.iter()
.filter(|(_, conn)| {
conn.is_outbound()
&& conn.handshake_state() == HandshakeState::SentMsg1
&& conn.resend_count() < max_resends
&& conn.next_resend_at_ms() > 0
&& now_ms >= conn.next_resend_at_ms()
// Skip resend if the target peer is already promoted — a
// cross-connection was resolved via the inbound path and
// resending msg1 would start a new handshake on the peer,
// creating a session mismatch.
&& !conn
.expected_identity()
.map(|id| self.peers.contains_key(id.node_addr()))
.unwrap_or(false)
})
.filter_map(|(link_id, conn)| {
conn.handshake_msg1().map(|msg1| ConnSnapshot {
link: *link_id,
is_outbound: true,
retry_addr: None,
resend_count: conn.resend_count(),
msg1: msg1.to_vec(),
})
})
.collect()
}
fn rekey_peers(&self) -> Vec<PeerSnapshot> {
// The snapshot builder lives in `rekey` beside its drain/dampening
// constants; the read-seam unifies here.
self.rekey_peer_snapshots()
}
fn rekey_resend_candidates(&self, now_ms: u64) -> Vec<RekeyResendSnapshot> {
self.rekey_resend_snapshots(now_ms)
}
}
impl Node {
/// Check for timed-out handshake connections and clean them up.
///
/// Called periodically by the RX event loop. Removes connections that have
/// been idle longer than the configured handshake timeout or are in Failed state.
///
/// The stale/failed predicate and every registry mutation stay shell-side;
/// the retry-then-teardown choreography is the pure
/// [`Fmp::poll_timeouts`](crate::proto::fmp::Fmp::poll_timeouts) decision.
pub(in crate::node) fn check_timeouts(&mut self) {
if self.connections.is_empty() {
return;
@@ -19,41 +82,34 @@ impl Node {
let now_ms = Self::now_ms();
let timeout_ms = self.config().node.rate_limit.handshake_timeout_secs * 1000;
let stale: Vec<LinkId> = self
.connections
.iter()
.filter(|(_, conn)| conn.is_timed_out(now_ms, timeout_ms) || conn.is_failed())
.map(|(link_id, _)| *link_id)
.collect();
for link_id in stale {
// Log and schedule retry before cleanup (need connection state)
if let Some(conn) = self.connections.get(&link_id) {
let direction = conn.direction();
let idle_ms = conn.idle_time(now_ms);
if conn.is_failed() {
debug!(
link_id = %link_id,
direction = %direction,
"Failed handshake connection cleaned up"
);
} else {
debug!(
link_id = %link_id,
direction = %direction,
idle_secs = idle_ms / 1000,
"Stale handshake connection timed out"
);
}
// Schedule retry for failed outbound auto-connect peers
if conn.is_outbound()
&& let Some(identity) = conn.expected_identity()
{
self.schedule_retry(*identity.node_addr(), now_ms);
let stale = self.stale_connections(now_ms, timeout_ms);
for action in self.fmp.poll_timeouts(stale) {
match action {
ConnAction::ScheduleRetry { peer } => self.schedule_retry(peer, now_ms),
ConnAction::Teardown { link } => {
// Log before cleanup (needs live connection state).
if let Some(conn) = self.connections.get(&link) {
let direction = conn.direction();
if conn.is_failed() {
debug!(
link_id = %link,
direction = %direction,
"Failed handshake connection cleaned up"
);
} else {
debug!(
link_id = %link,
direction = %direction,
idle_secs = conn.idle_time(now_ms) / 1000,
"Stale handshake connection timed out"
);
}
}
self.cleanup_stale_connection(link, now_ms);
}
#[allow(unreachable_patterns)]
_ => {}
}
self.cleanup_stale_connection(link_id, now_ms);
}
}
@@ -98,33 +154,24 @@ impl Node {
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.
// Skip resend if the target peer is already promoted — a cross-connection
// was resolved via the inbound path and resending msg1 would start a new
// handshake on the peer, creating a session mismatch.
let candidates: Vec<(LinkId, Vec<u8>)> = self
.connections
.iter()
.filter(|(_, conn)| {
conn.is_outbound()
&& conn.handshake_state() == HandshakeState::SentMsg1
&& conn.resend_count() < max_resends
&& conn.next_resend_at_ms() > 0
&& now_ms >= conn.next_resend_at_ms()
&& !conn
.expected_identity()
.map(|id| self.peers.contains_key(id.node_addr()))
.unwrap_or(false)
})
.filter_map(|(link_id, conn)| {
conn.handshake_msg1().map(|msg1| (*link_id, msg1.to_vec()))
})
.collect();
// The shell resolves the resend-candidate predicate and copies the
// opaque msg1 bytes; the core computes the backoff schedule.
let candidates = self.resend_candidates(now_ms, max_resends);
for action in self
.fmp
.poll_resends(candidates, now_ms, interval_ms, backoff)
{
let ConnAction::ResendMsg1 {
link,
bytes,
next_resend_at_ms,
} = action
else {
continue;
};
for (link_id, msg1_bytes) in candidates {
// Get transport and address info from the connection
let (transport_id, remote_addr) = match self.connections.get(&link_id) {
let (transport_id, remote_addr) = match self.connections.get(&link) {
Some(conn) => match (conn.transport_id(), conn.source_addr()) {
(Some(tid), Some(addr)) => (tid, addr.clone()),
_ => continue,
@@ -134,11 +181,11 @@ impl Node {
// Send the stored msg1
let sent = if let Some(transport) = self.transports.get(&transport_id) {
match transport.send(&remote_addr, &msg1_bytes).await {
match transport.send(&remote_addr, &bytes).await {
Ok(_) => true,
Err(e) => {
debug!(
link_id = %link_id,
link_id = %link,
error = %e,
"Handshake msg1 resend failed"
);
@@ -149,13 +196,11 @@ impl Node {
false
};
if sent && let Some(conn) = self.connections.get_mut(&link_id) {
let count = conn.resend_count() + 1;
let next = now_ms + (interval_ms as f64 * backoff.powi(count as i32)) as u64;
conn.record_resend(next);
if sent && let Some(conn) = self.connections.get_mut(&link) {
conn.record_resend(next_resend_at_ms);
debug!(
link_id = %link_id,
resend = count,
link_id = %link,
resend = conn.resend_count(),
"Resent handshake msg1"
);
}
src/node/lifecycle.rs
+1 -1
View File
@@ -10,7 +10,7 @@ use crate::discovery::{BootstrapHandoffResult, EstablishedTraversal};
use crate::node::acl::PeerAclContext;
use crate::node::wire::build_msg1;
use crate::peer::PeerConnection;
use crate::protocol::{Disconnect, DisconnectReason};
use crate::proto::fmp::{Disconnect, DisconnectReason};
use crate::transport::{Link, LinkDirection, LinkId, TransportAddr, TransportId, packet_channel};
use crate::upper::tun::{TunDevice, TunState, run_tun_reader, shutdown_tun_interface};
use crate::{NodeAddr, PeerIdentity};
src/node/mod.rs
+7 -1
View File
@@ -58,8 +58,9 @@ use crate::cache::CoordCache;
use crate::node::session::SessionEntry;
use crate::peer::{ActivePeer, PeerConnection};
use crate::proto::discovery::{Discovery, DiscoveryBackoff, DiscoveryForwardRateLimiter};
use crate::proto::fmp::Fmp;
use crate::proto::fmp::NodeProfile;
use crate::proto::routing::{self, Router, RoutingErrorRateLimiter};
use crate::protocol::NodeProfile;
#[cfg(unix)]
use crate::transport::ethernet::EthernetTransport;
use crate::transport::nym::NymTransport;
@@ -436,6 +437,9 @@ pub struct Node {
icmp_rate_limiter: IcmpRateLimiter,
/// Routing-subsystem state (routing error-signal rate limiter).
routing: Router,
/// FMP connection-lifecycle decision anchor (stateless; drives the
/// tick-poll maintain/teardown decisions).
fmp: Fmp,
/// Rate limiter for source-side CoordsRequired/PathBroken responses.
coords_response_rate_limiter: RoutingErrorRateLimiter,
@@ -681,6 +685,7 @@ impl Node {
msg1_rate_limiter,
icmp_rate_limiter: IcmpRateLimiter::new(),
routing: Router::new(),
fmp: Fmp::new(),
coords_response_rate_limiter: RoutingErrorRateLimiter::with_interval_ms(
coords_response_interval_ms,
),
@@ -842,6 +847,7 @@ impl Node {
msg1_rate_limiter,
icmp_rate_limiter: IcmpRateLimiter::new(),
routing: Router::new(),
fmp: Fmp::new(),
coords_response_rate_limiter: RoutingErrorRateLimiter::with_interval_ms(
coords_response_interval_ms,
),
src/node/retry.rs
+5 -87
View File
@@ -8,6 +8,7 @@ use super::{Node, NodeError};
use crate::PeerIdentity;
use crate::config::PeerConfig;
use crate::identity::NodeAddr;
use crate::proto::fmp::backoff_ms;
use tracing::{debug, info, warn};
// MAX_BACKOFF_MS is now derived from config: node.retry.max_backoff_secs * 1000
@@ -45,17 +46,6 @@ impl RetryState {
expires_at_ms: None,
}
}
/// Calculate the backoff delay in milliseconds for the current retry count.
///
/// Uses exponential backoff: `base_interval_ms * 2^retry_count`,
/// capped at `MAX_BACKOFF_MS`.
pub fn backoff_ms(&self, base_interval_ms: u64, max_backoff_ms: u64) -> u64 {
let multiplier = 1u64.checked_shl(self.retry_count).unwrap_or(u64::MAX);
base_interval_ms
.saturating_mul(multiplier)
.min(max_backoff_ms)
}
}
impl Node {
@@ -93,7 +83,7 @@ impl Node {
self.retry_pending.remove(&node_addr);
return;
}
let delay = state.backoff_ms(base_interval_ms, max_backoff_ms);
let delay = backoff_ms(state.retry_count, base_interval_ms, max_backoff_ms);
state.retry_after_ms = now_ms + delay;
debug!(
peer = %peer_name,
@@ -118,7 +108,7 @@ impl Node {
let mut state = RetryState::new(pc);
state.retry_count = 1;
state.reconnect = true;
let delay = state.backoff_ms(base_interval_ms, max_backoff_ms);
let delay = backoff_ms(state.retry_count, base_interval_ms, max_backoff_ms);
state.retry_after_ms = now_ms + delay;
debug!(
peer = %self.peer_display_name(&node_addr),
@@ -175,7 +165,7 @@ impl Node {
if let Some(state) = self.retry_pending.get_mut(&node_addr) {
state.reconnect = true;
state.retry_count += 1;
let delay = state.backoff_ms(base_interval_ms, max_backoff_ms);
let delay = backoff_ms(state.retry_count, base_interval_ms, max_backoff_ms);
state.retry_after_ms = now_ms + delay;
debug!(
peer = %peer_name,
@@ -188,7 +178,7 @@ impl Node {
let mut state = RetryState::new(pc);
state.reconnect = true;
let delay = state.backoff_ms(base_interval_ms, max_backoff_ms);
let delay = backoff_ms(state.retry_count, base_interval_ms, max_backoff_ms);
state.retry_after_ms = now_ms + delay;
debug!(
@@ -343,75 +333,3 @@ impl Node {
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::config::PeerConfig;
const TEST_MAX_BACKOFF_MS: u64 = 300_000;
#[test]
fn test_backoff_exponential() {
let state = RetryState {
peer_config: PeerConfig::default(),
retry_count: 0,
retry_after_ms: 0,
reconnect: false,
expires_at_ms: None,
};
// base = 5000ms
assert_eq!(state.backoff_ms(5000, TEST_MAX_BACKOFF_MS), 5000); // 5s * 2^0
let state = RetryState {
retry_count: 1,
..state
};
assert_eq!(state.backoff_ms(5000, TEST_MAX_BACKOFF_MS), 10_000); // 5s * 2^1
let state = RetryState {
retry_count: 2,
..state
};
assert_eq!(state.backoff_ms(5000, TEST_MAX_BACKOFF_MS), 20_000); // 5s * 2^2
let state = RetryState {
retry_count: 3,
..state
};
assert_eq!(state.backoff_ms(5000, TEST_MAX_BACKOFF_MS), 40_000); // 5s * 2^3
let state = RetryState {
retry_count: 4,
..state
};
assert_eq!(state.backoff_ms(5000, TEST_MAX_BACKOFF_MS), 80_000); // 5s * 2^4
}
#[test]
fn test_backoff_cap() {
let state = RetryState {
peer_config: PeerConfig::default(),
retry_count: 20, // 2^20 * 5000 would be huge
retry_after_ms: 0,
reconnect: false,
expires_at_ms: None,
};
assert_eq!(
state.backoff_ms(5000, TEST_MAX_BACKOFF_MS),
TEST_MAX_BACKOFF_MS
);
}
#[test]
fn test_backoff_zero_base() {
let state = RetryState {
peer_config: PeerConfig::default(),
retry_count: 3,
retry_after_ms: 0,
reconnect: false,
expires_at_ms: None,
};
assert_eq!(state.backoff_ms(0, TEST_MAX_BACKOFF_MS), 0);
}
}
src/node/tests/disconnect.rs
+2 -2
View File
@@ -6,7 +6,7 @@
use super::spanning_tree::*;
use super::*;
use crate::protocol::{Disconnect, DisconnectReason};
use crate::proto::fmp::{Disconnect, DisconnectReason};
/// 3-node chain: middle node disconnects one peer.
///
@@ -296,7 +296,7 @@ async fn test_disconnect_clears_session() {
);
// Node 0 sends Disconnect to node 1.
let disconnect = crate::protocol::Disconnect::new(DisconnectReason::Shutdown);
let disconnect = crate::proto::fmp::Disconnect::new(DisconnectReason::Shutdown);
nodes[0]
.node
.send_encrypted_link_message(&node1_addr, &disconnect.encode())
src/node/tests/handshake.rs
+8 -8
View File
@@ -998,8 +998,8 @@ async fn test_duplicate_msg2_dropped() {
/// Helper: create two test nodes, set their profiles, attempt a handshake,
/// and return whether they successfully peered.
async fn attempt_profile_handshake(
profile_a: crate::protocol::NodeProfile,
profile_b: crate::protocol::NodeProfile,
profile_a: crate::proto::fmp::NodeProfile,
profile_b: crate::proto::fmp::NodeProfile,
) -> (usize, usize) {
let mut nodes = vec![
make_test_node_with_profile(profile_a).await,
@@ -1016,7 +1016,7 @@ async fn attempt_profile_handshake(
#[tokio::test]
async fn test_nonrouting_nonrouting_rejected() {
use crate::protocol::NodeProfile;
use crate::proto::fmp::NodeProfile;
let (a, b) = attempt_profile_handshake(NodeProfile::NonRouting, NodeProfile::NonRouting).await;
assert_eq!(a, 0, "NonRouting↔NonRouting should reject: node A");
assert_eq!(b, 0, "NonRouting↔NonRouting should reject: node B");
@@ -1024,7 +1024,7 @@ async fn test_nonrouting_nonrouting_rejected() {
#[tokio::test]
async fn test_leaf_leaf_rejected() {
use crate::protocol::NodeProfile;
use crate::proto::fmp::NodeProfile;
let (a, b) = attempt_profile_handshake(NodeProfile::Leaf, NodeProfile::Leaf).await;
assert_eq!(a, 0, "Leaf↔Leaf should reject: node A");
assert_eq!(b, 0, "Leaf↔Leaf should reject: node B");
@@ -1032,7 +1032,7 @@ async fn test_leaf_leaf_rejected() {
#[tokio::test]
async fn test_nonrouting_leaf_rejected() {
use crate::protocol::NodeProfile;
use crate::proto::fmp::NodeProfile;
let (a, b) = attempt_profile_handshake(NodeProfile::NonRouting, NodeProfile::Leaf).await;
assert_eq!(a, 0, "NonRouting↔Leaf should reject: node A");
assert_eq!(b, 0, "NonRouting↔Leaf should reject: node B");
@@ -1040,7 +1040,7 @@ async fn test_nonrouting_leaf_rejected() {
#[tokio::test]
async fn test_leaf_nonrouting_rejected() {
use crate::protocol::NodeProfile;
use crate::proto::fmp::NodeProfile;
let (a, b) = attempt_profile_handshake(NodeProfile::Leaf, NodeProfile::NonRouting).await;
assert_eq!(a, 0, "Leaf↔NonRouting should reject: node A");
assert_eq!(b, 0, "Leaf↔NonRouting should reject: node B");
@@ -1048,7 +1048,7 @@ async fn test_leaf_nonrouting_rejected() {
#[tokio::test]
async fn test_full_nonrouting_accepted() {
use crate::protocol::NodeProfile;
use crate::proto::fmp::NodeProfile;
let (a, b) = attempt_profile_handshake(NodeProfile::Full, NodeProfile::NonRouting).await;
assert_eq!(a, 1, "Full↔NonRouting should accept: node A");
assert_eq!(b, 1, "Full↔NonRouting should accept: node B");
@@ -1056,7 +1056,7 @@ async fn test_full_nonrouting_accepted() {
#[tokio::test]
async fn test_full_leaf_accepted() {
use crate::protocol::NodeProfile;
use crate::proto::fmp::NodeProfile;
let (a, b) = attempt_profile_handshake(NodeProfile::Full, NodeProfile::Leaf).await;
assert_eq!(a, 1, "Full↔Leaf should accept: node A");
assert_eq!(b, 1, "Full↔Leaf should accept: node B");
src/node/tests/routing.rs
+1 -1
View File
@@ -809,7 +809,7 @@ async fn test_routing_reachability_100_nodes() {
/// Node 0 should no longer be able to route to node 3.
#[tokio::test]
async fn test_routing_stops_after_peer_removal() {
use crate::protocol::{Disconnect, DisconnectReason};
use crate::proto::fmp::{Disconnect, DisconnectReason};
let edges = vec![(0, 1), (1, 2), (2, 3)];
let mut nodes = run_tree_test(4, &edges, false).await;
src/node/tests/spanning_tree.rs
+4 -2
View File
@@ -84,8 +84,10 @@ pub(super) async fn make_test_node_with_mtu(mtu: u16) -> TestNode {
/// Create a test node with a specific routing profile. Profile is immutable
/// (lives in the shared context), so it is set via the `Config` flags that
/// `Config::node_profile()` reads rather than poked post-construction.
pub(super) async fn make_test_node_with_profile(profile: crate::protocol::NodeProfile) -> TestNode {
use crate::protocol::NodeProfile;
pub(super) async fn make_test_node_with_profile(
profile: crate::proto::fmp::NodeProfile,
) -> TestNode {
use crate::proto::fmp::NodeProfile;
let mut config = Config::new();
match profile {
NodeProfile::Leaf => config.node.leaf_only = true,
src/node/tests/unit.rs
+3 -3
View File
@@ -1263,7 +1263,7 @@ fn test_schedule_reconnect_preserves_backoff() {
// With count=3, backoff should be 5s * 2^3 = 40s.
let base_ms = node.config().node.retry.base_interval_secs * 1000;
let max_ms = node.config().node.retry.max_backoff_secs * 1000;
let expected_delay = state.backoff_ms(base_ms, max_ms);
let expected_delay = crate::proto::fmp::backoff_ms(state.retry_count, base_ms, max_ms);
assert_eq!(
state.retry_after_ms,
31_000 + expected_delay,
@@ -1299,7 +1299,7 @@ fn test_schedule_reconnect_fresh_state() {
// Base delay: 5s * 2^0 = 5s
let base_ms = node.config().node.retry.base_interval_secs * 1000;
let max_ms = node.config().node.retry.max_backoff_secs * 1000;
let expected_delay = state.backoff_ms(base_ms, max_ms);
let expected_delay = crate::proto::fmp::backoff_ms(state.retry_count, base_ms, max_ms);
assert_eq!(state.retry_after_ms, 1_000 + expected_delay);
}
@@ -1311,7 +1311,7 @@ fn test_schedule_reconnect_fresh_state() {
/// decrypt failure, peer restart) all schedule reconnect.
#[test]
fn test_disconnect_schedules_reconnect() {
use crate::protocol::{Disconnect, DisconnectReason};
use crate::proto::fmp::{Disconnect, DisconnectReason};
let peer_identity = Identity::generate();
let peer_npub = peer_identity.npub();
src/node/tree.rs
+1 -1
View File
@@ -37,7 +37,7 @@ impl Node {
&mut self,
peer_addr: &NodeAddr,
) -> Result<(), NodeError> {
if self.node_profile() == crate::protocol::NodeProfile::Leaf {
if self.node_profile() == crate::proto::fmp::NodeProfile::Leaf {
return Ok(());
}
let now_ms = std::time::SystemTime::now()