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

proto/fmp: sans-IO connection-lifecycle state machine

Browse Source
This commit is contained in:
Johnathan Corgan
2026-07-07 06:20:02 +00:00
parent 9ea57b483a
commit 4802792e38
32 changed files with 4279 additions and 1392 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/node/handlers/dispatch.rs
+1 -1
View File
@@ -73,7 +73,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
+257 -260
View File
@@ -1,15 +1,62 @@
//! Handshake handlers and connection promotion.
use crate::NodeAddr;
use crate::PeerIdentity;
use crate::node::acl::PeerAclContext;
use crate::node::reject::{HandshakeReject, RejectReason};
use crate::node::wire::{Msg1Header, Msg2Header, build_msg2};
use crate::node::{Node, NodeError};
use crate::peer::{ActivePeer, PeerConnection, PromotionResult, cross_connection_winner};
use crate::peer::{ActivePeer, PeerConnection, PromotionResult};
use crate::proto::fmp::{
ConnAction, EstablishSnapshot, EstablishView, InboundDecision, InboundReject, OutboundDecision,
OutboundSnapshot, WireOutcome, cross_connection_winner,
};
use crate::transport::{Link, LinkDirection, LinkId, ReceivedPacket};
use std::time::Duration;
use tracing::{debug, info, warn};
impl EstablishView for Node {
fn establish_snapshot(&self, peer_addr: &NodeAddr) -> EstablishSnapshot {
let existing = self.peers.get(peer_addr);
let max_peers = self.max_peers();
EstablishSnapshot {
has_existing_peer: existing.is_some(),
existing_peer_epoch: existing.and_then(|p| p.remote_epoch()),
existing_session_age_secs: existing
.map(|p| p.session_established_at().elapsed().as_secs())
.unwrap_or(0),
has_session: existing.map(|p| p.has_session()).unwrap_or(false),
is_healthy: existing.map(|p| p.is_healthy()).unwrap_or(false),
pending_new_session: existing
.map(|p| p.pending_new_session().is_some())
.unwrap_or(false),
rekey_in_progress: existing.map(|p| p.rekey_in_progress()).unwrap_or(false),
existing_msg2: existing.and_then(|p| p.handshake_msg2().map(|m| m.to_vec())),
at_max_peers: max_peers > 0 && self.peers.len() >= max_peers,
has_pending_outbound_to_peer: self.connections.values().any(|conn| {
conn.expected_identity()
.map(|id| id.node_addr() == peer_addr)
.unwrap_or(false)
}),
rekey_enabled: self.config().node.rekey.enabled,
our_node_addr: *self.identity().node_addr(),
}
}
fn outbound_snapshot(&self, peer_addr: &NodeAddr) -> OutboundSnapshot {
OutboundSnapshot {
has_existing_peer: self.peers.contains_key(peer_addr),
// Tie-break for THIS outbound connection (`is_outbound = true`),
// pre-evaluated here so the core stays free of the peer helper.
our_outbound_wins: cross_connection_winner(
self.identity().node_addr(),
peer_addr,
true,
),
}
}
}
impl Node {
/// Returns true if an inbound msg1 should be admitted past the
/// `accept_connections` gate.
@@ -96,32 +143,23 @@ impl Node {
}
};
// Check for existing connection from this address.
//
// If we already have an *inbound* link from this address, this could be:
// 1. A duplicate msg1 (our msg2 was lost) — resend msg2
// 2. A restarted peer (different epoch) — tear down and reprocess
//
// If we have an *outbound* link to this address (we initiated to them
// AND they initiated to us), this is a cross-connection — allow it.
//
// Epoch-based restart detection: if the sender already has an inbound
// link AND is an active peer in self.peers, fall through to decrypt
// the msg1 and check the epoch. Otherwise, treat as duplicate.
// Pre-crypto duplicate short-circuit. An *inbound* link from this
// address that is not (yet) a promoted peer means our earlier msg2 was
// lost: resend the stored msg2 without paying the crypto cost and
// return. An inbound link that DOES belong to an active peer (a possible
// restart/rekey) or an *outbound* link (a cross-connection) falls
// through to the wire step and the structured classification below —
// the pre-refactor `possible_restart` flag is no longer needed because
// that classification now gates on `has_existing_peer` (identity), which
// subsumes it.
let addr_key = (packet.transport_id, packet.remote_addr.clone());
let mut possible_restart = false;
if let Some(&existing_link_id) = self.addr_to_link.get(&addr_key)
&& let Some(link) = self.links.get(&existing_link_id)
{
if link.direction() == LinkDirection::Inbound {
// Check if this link belongs to an already-promoted active peer
let is_active_peer = self.peers.values().any(|p| p.link_id() == existing_link_id);
if is_active_peer {
// Possible restart — fall through to decrypt and check epoch
possible_restart = true;
} else {
// Genuinely pending handshake — resend msg2
if !is_active_peer {
// Genuinely pending handshake — resend msg2.
let msg2_bytes = self.find_stored_msg2(existing_link_id);
if let Some(msg2) = msg2_bytes {
if let Some(transport) = self.transports.get(&packet.transport_id) {
@@ -150,14 +188,11 @@ impl Node {
return;
}
} else {
// Outbound link to this address. If it belongs to an active
// peer, this may be a rekey msg1 (same epoch) or a
// restart (different epoch). Set possible_restart to enable
// the epoch/rekey check below.
// Outbound link to this address with no active peer yet: a
// cross-connection. Just log; it is classified as a net-new
// inbound below.
let is_active_peer = self.peers.values().any(|p| p.link_id() == existing_link_id);
if is_active_peer {
possible_restart = true;
} else {
if !is_active_peer {
debug!(
transport_id = %packet.transport_id,
remote_addr = %packet.remote_addr,
@@ -212,247 +247,184 @@ impl Node {
let peer_node_addr = *peer_identity.node_addr();
// Identity-based restart/rekey detection: if the peer is already
// active but addr_to_link didn't match (different source address, e.g.,
// TCP from a different port), we still need to check for restart/rekey.
if !possible_restart && self.peers.contains_key(&peer_node_addr) {
possible_restart = true;
}
// === PHASE B result ===
// Bundle the Noise wire-step outputs (identity, remote epoch, sender
// index, opaque msg2 payload). The wire step touched no `Node` registry
// state; from here the decision reads only `wire` and the snapshot.
let wire = WireOutcome {
peer_identity,
remote_epoch: conn.remote_epoch(),
their_index: header.sender_idx,
msg2_payload: msg2_response,
};
// Early cap check: at max_peers and this is a net-new identity?
// Bypass for known peers (reconnect / cross-connection) — admitting
// them doesn't grow peers.len(). This silent-drops the Msg1 before
// the Msg2 build/send and index allocation, avoiding wasted wire
// bytes and giving the peer cleaner semantics (no fake-completed
// handshake whose data frames subsequently fail decryption here).
// The late cap check inside promote_connection() is intentionally
// left in place as defense-in-depth.
if self.max_peers() > 0 && self.peers.len() >= self.max_peers() {
let is_known_active = self.peers.contains_key(&peer_node_addr);
let is_pending_outbound = self.connections.iter().any(|(_, conn)| {
conn.expected_identity()
.map(|id| *id.node_addr() == peer_node_addr)
.unwrap_or(false)
});
if !is_known_active && !is_pending_outbound {
debug!(
peer = %self.peer_display_name(&peer_node_addr),
max = self.max_peers(),
"Silent-dropping Msg1 at max_peers cap (early gate; no Msg2 sent)"
);
// `link_id` was allocated above but `conn` is still a local
// (not yet inserted into self.connections / self.links /
// self.addr_to_link), so the local drop suffices.
// === PHASE C input ===
// Snapshot the registry state the inbound classification reads about
// this peer identity (existing epoch/session/rekey state with the
// session age resolved here, the max-peers cap, our own address for the
// tie-break). Taken before this connection is inserted into the
// registry, matching the pre-refactor read points.
let est = self.establish_snapshot(&peer_node_addr);
// === PHASE C: structured classification (pure core) ===
// The decision reads only the snapshot + wire outcome; the shell below
// drives the effects. `Promote`/`RestartThenPromote` fall through to the
// shared authorize → allocate → send-msg2 → promote tail; the other
// variants complete the rate-limiter and return here.
match self.fmp.establish_inbound(&est, &wire) {
InboundDecision::Reject { reason } => {
match reason {
InboundReject::AtMaxPeers => debug!(
peer = %self.peer_display_name(&peer_node_addr),
max = self.max_peers(),
"Silent-dropping Msg1 at max_peers cap (early gate; no Msg2 sent)"
),
InboundReject::PendingSession => debug!(
peer = %self.peer_display_name(&peer_node_addr),
"Rekey msg1 received but already have pending session, dropping"
),
InboundReject::DualRekeyWon => debug!(
peer = %self.peer_display_name(&peer_node_addr),
"Dual rekey initiation: we win (smaller addr), dropping their msg1"
),
}
// `conn`/`link_id` were never inserted into the registry, so the
// local drop suffices — no cleanup needed.
self.msg1_rate_limiter.complete_handshake();
self.stats_mut()
.record_reject(RejectReason::Handshake(HandshakeReject::BadState));
return;
}
}
// Epoch-based restart detection and duplicate msg1 handling.
//
// If we fell through from the addr_to_link check above with
// possible_restart=true, we now have the decrypted epoch from msg1.
// Compare it against the stored epoch for this peer.
if possible_restart && let Some(existing_peer) = self.peers.get(&peer_node_addr) {
let new_epoch = conn.remote_epoch();
let existing_epoch = existing_peer.remote_epoch();
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
InboundDecision::ResendMsg2 { msg2 } => {
if let Some(msg2) = msg2.as_deref()
&& 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 msg1 (same epoch)"
),
Err(e) => debug!(
peer = %self.peer_display_name(&peer_node_addr),
error = %e,
"Failed to resend msg2"
),
}
}
_ => {
// Same epoch (or no epoch stored).
// If the peer has an active session and rekey is enabled,
// this is a rekey msg1 (not a duplicate initial msg1).
// Guard: the session must be at least 30s old to avoid
// misidentifying a cross-connection msg1 as a rekey.
// During simultaneous connection, both sides promote
// within the same tick and the peer's msg1 arrives
// 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
&& existing_peer.has_session()
&& existing_peer.is_healthy()
&& session_age_secs >= 30
self.msg1_rate_limiter.complete_handshake();
return;
}
InboundDecision::RekeyRespond {
peer,
abandon_first,
} => {
if abandon_first {
// Dual-initiation loser: abandon our own in-flight rekey and
// free its index before responding as the rekey responder.
debug!(
peer = %self.peer_display_name(&peer),
"Dual rekey initiation: we lose (larger addr), abandoning ours"
);
if let Some(existing) = self.peers.get_mut(&peer)
&& let Some(idx) = existing.abandon_rekey()
{
// Guard: already have a pending session from a completed
// rekey (waiting for K-bit cutover). Don't overwrite it
// with a new handshake — drop this msg1.
if existing_peer.pending_new_session().is_some() {
if let Some(tid) = existing.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 = conn.take_session();
let our_new_index = match self.index_allocator.allocate() {
Ok(idx) => idx,
Err(e) => {
warn!(error = %e, "Failed to allocate index for rekey");
self.msg1_rate_limiter.complete_handshake();
self.stats_mut()
.record_reject(RejectReason::Handshake(HandshakeReject::BadState));
return;
}
};
let noise_session = match noise_session {
Some(s) => s,
None => {
warn!("Rekey msg1: no session from handshake");
let _ = self.index_allocator.free(our_new_index);
self.msg1_rate_limiter.complete_handshake();
self.stats_mut()
.record_reject(RejectReason::Handshake(HandshakeReject::BadState));
return;
}
};
// Send msg2 response using the new handshake.
let wire_msg2 = build_msg2(our_new_index, wire.their_index, &wire.msg2_payload);
if let Some(transport) = self.transports.get(&packet.transport_id) {
match transport.send(&packet.remote_addr, &wire_msg2).await {
Ok(_) => {
debug!(
peer = %self.peer_display_name(&peer_node_addr),
"Rekey msg1 received but already have pending session, dropping"
peer = %self.peer_display_name(&peer),
new_our_index = %our_new_index,
"Sent rekey msg2 response"
);
self.connections.remove(&link_id);
self.links.remove(&link_id);
}
Err(e) => {
warn!(
peer = %self.peer_display_name(&peer),
error = %e,
"Failed to send rekey msg2"
);
let _ = self.index_allocator.free(our_new_index);
self.msg1_rate_limiter.complete_handshake();
self.stats_mut()
.record_reject(RejectReason::Handshake(HandshakeReject::BadState));
return;
}
// Dual-initiation detection: both sides sent msg1
// 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();
if our_addr < &peer_node_addr {
// We win as initiator — drop their msg1.
// Our msg2 will arrive at peer, who completes
// as our responder.
debug!(
peer = %self.peer_display_name(&peer_node_addr),
"Dual rekey initiation: we win (smaller addr), dropping their msg1"
);
self.connections.remove(&link_id);
self.links.remove(&link_id);
self.msg1_rate_limiter.complete_handshake();
self.stats_mut().record_reject(RejectReason::Handshake(
HandshakeReject::BadState,
));
return;
}
// We lose — abandon our rekey, become responder below.
debug!(
peer = %self.peer_display_name(&peer_node_addr),
"Dual rekey initiation: we lose (larger addr), abandoning 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
}
// Rekey: process as responder, store new session as pending
let noise_session = conn.take_session();
let our_new_index = match self.index_allocator.allocate() {
Ok(idx) => idx,
Err(e) => {
warn!(error = %e, "Failed to allocate index for rekey");
self.msg1_rate_limiter.complete_handshake();
self.stats_mut().record_reject(RejectReason::Handshake(
HandshakeReject::BadState,
));
return;
}
};
let noise_session = match noise_session {
Some(s) => s,
None => {
warn!("Rekey msg1: no session from handshake");
let _ = self.index_allocator.free(our_new_index);
self.msg1_rate_limiter.complete_handshake();
self.stats_mut().record_reject(RejectReason::Handshake(
HandshakeReject::BadState,
));
return;
}
};
// Send msg2 response using the new handshake
let wire_msg2 =
build_msg2(our_new_index, header.sender_idx, &msg2_response);
if let Some(transport) = self.transports.get(&packet.transport_id) {
match transport.send(&packet.remote_addr, &wire_msg2).await {
Ok(_) => {
debug!(
peer = %self.peer_display_name(&peer_node_addr),
new_our_index = %our_new_index,
"Sent rekey msg2 response"
);
}
Err(e) => {
warn!(
peer = %self.peer_display_name(&peer_node_addr),
error = %e,
"Failed to send rekey msg2"
);
let _ = self.index_allocator.free(our_new_index);
self.msg1_rate_limiter.complete_handshake();
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();
}
// 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 we created.
// Do NOT remove addr_to_link — the entry must remain pointing
// to the original link so future msg1s from this address are
// recognized as rekeys (not new connections).
self.connections.remove(&link_id);
self.links.remove(&link_id);
self.msg1_rate_limiter.complete_handshake();
return;
}
// Not a rekey — duplicate msg1. 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 msg1 (same epoch)"
),
Err(e) => debug!(
peer = %self.peer_display_name(&peer_node_addr),
error = %e,
"Failed to resend msg2"
),
}
}
self.msg1_rate_limiter.complete_handshake();
return;
}
// Store pending session on the existing peer.
if let Some(existing) = self.peers.get_mut(&peer) {
existing.set_pending_session(noise_session, our_new_index, wire.their_index);
existing.record_peer_rekey();
}
// Register new index in peers_by_index.
self.peers_by_index
.insert((packet.transport_id, our_new_index.as_u32()), peer);
// Do NOT touch addr_to_link — the entry must keep pointing at the
// original link so future msg1s from this address are recognized
// as rekeys (not new connections). The temporary `conn`/`link_id`
// were never inserted into the registry, so no cleanup is needed.
self.msg1_rate_limiter.complete_handshake();
return;
}
InboundDecision::RestartThenPromote { peer } => {
// Epoch mismatch — peer restarted. Tear down the stale session
// and schedule a reconnect, then fall through to promote the
// fresh handshake as a new connection.
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);
}
InboundDecision::Promote => {}
}
// If possible_restart was true but peer is no longer in self.peers
// (removed by another path), fall through to process as new connection.
if self
.authorize_peer(
&peer_identity,
&wire.peer_identity,
PeerAclContext::InboundHandshake,
packet.transport_id,
&packet.remote_addr,
@@ -481,7 +453,7 @@ impl Node {
};
conn.set_our_index(our_index);
conn.set_their_index(header.sender_idx);
conn.set_their_index(wire.their_index);
// Create link
let link = Link::connectionless(
@@ -497,7 +469,7 @@ impl Node {
self.connections.insert(link_id, conn);
// Build and send msg2 response, storing for potential resend
let wire_msg2 = build_msg2(our_index, header.sender_idx, &msg2_response);
let wire_msg2 = build_msg2(our_index, wire.their_index, &wire.msg2_payload);
if let Some(conn) = self.connections.get_mut(&link_id) {
conn.set_handshake_msg2(wire_msg2.clone());
}
@@ -508,7 +480,7 @@ impl Node {
debug!(
link_id = %link_id,
our_index = %our_index,
their_index = %header.sender_idx,
their_index = %wire.their_index,
bytes,
"Sent msg2 response"
);
@@ -536,7 +508,8 @@ impl Node {
// Responder handshake is complete after receive_handshake_init (Noise IK
// pattern: responder processes msg1 and generates msg2 in one step).
// Promote the connection to active peer now.
match self.promote_connection(link_id, peer_identity, packet.timestamp_ms) {
let promote = ConnAction::PromoteToActive { link: link_id };
match self.drive_promote_to_active(promote, wire.peer_identity, packet.timestamp_ms) {
Ok(result) => {
match result {
PromotionResult::Promoted(node_addr) => {
@@ -841,13 +814,12 @@ impl Node {
//
// This ensures both nodes use the same Noise handshake (the winner's
// outbound = the loser's inbound).
if self.peers.contains_key(&peer_node_addr) {
let our_outbound_wins = cross_connection_winner(
self.identity().node_addr(),
&peer_node_addr,
true, // this IS our outbound
);
// Structured classification (pure core): cross-connection swap/keep, or
// a net-new promote. The tie-break is pre-evaluated in the snapshot; the
// effect bodies below are unchanged.
let out_snap = self.outbound_snapshot(&peer_node_addr);
let out_decision = self.fmp.establish_outbound(&out_snap);
if out_decision != OutboundDecision::Promote {
// Extract the outbound connection
let mut conn = match self.connections.remove(&link_id) {
Some(c) => c,
@@ -859,7 +831,7 @@ impl Node {
}
};
if our_outbound_wins {
if out_decision == OutboundDecision::CrossConnectionSwap {
// We're the smaller node. Swap to outbound session + indices.
// The peer will keep their inbound session (complement of ours).
let outbound_our_index = conn.our_index();
@@ -961,7 +933,8 @@ impl Node {
}
// Normal path: promote to active peer
match self.promote_connection(link_id, peer_identity, packet.timestamp_ms) {
let promote = ConnAction::PromoteToActive { link: link_id };
match self.drive_promote_to_active(promote, peer_identity, packet.timestamp_ms) {
Ok(result) => {
// Clean up pending_outbound
self.pending_outbound.remove(&key);
@@ -1041,6 +1014,30 @@ impl Node {
}
}
/// Execute a [`ConnAction::PromoteToActive`] from the establish machine.
///
/// The decision to promote is made by the establish handlers (and, from the
/// establish-core stage on, the pure decision in `proto::fmp`); this is the
/// executor half of the seam. It runs the promotion through
/// [`Self::promote_connection`], resolving the verified identity and
/// promotion timestamp from the ambient wire context, and returns the
/// [`PromotionResult`] so the caller can drive the site-specific
/// post-promotion tail (TreeAnnounce, bloom mark, discovery-backoff reset,
/// loser-link cleanup).
fn drive_promote_to_active(
&mut self,
action: ConnAction,
verified_identity: PeerIdentity,
current_time_ms: u64,
) -> Result<PromotionResult, NodeError> {
match action {
ConnAction::PromoteToActive { link } => {
self.promote_connection(link, verified_identity, current_time_ms)
}
_ => unreachable!("drive_promote_to_active requires a PromoteToActive action"),
}
}
/// Promote a connection to active peer after successful authentication.
///
/// Handles cross-connection detection and resolution using tie-breaker rules.
src/node/handlers/rekey.rs
+165 -159
View File
@@ -9,6 +9,7 @@ use crate::NodeAddr;
use crate::node::Node;
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, trace, warn};
@@ -41,121 +42,112 @@ 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.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).
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"
);
debug!(
peer = %self.peer_display_name(&node_addr),
"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"
);
debug!(
peer = %self.peer_display_name(&node_addr),
"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),
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.
@@ -260,60 +252,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) {
to_resend.push((*node_addr, peer.rekey_msg1().unwrap().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 FSP rekey msg3 until the responder is confirmed on the
src/node/handlers/timeout.rs
+103 -59
View File
@@ -3,14 +3,69 @@
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()
})
.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 +74,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);
}
}
@@ -97,26 +145,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.
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()
})
.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,
@@ -126,11 +172,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"
);
@@ -141,13 +187,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
+6
View File
@@ -45,6 +45,7 @@ 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::routing::{self, Router, RoutingErrorRateLimiter};
#[cfg(unix)]
use crate::transport::ethernet::EthernetTransport;
@@ -416,6 +417,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,
@@ -658,6 +662,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,
),
@@ -817,6 +822,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.
///
@@ -295,7 +295,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/establish_chartests.rs
+998
View File
@@ -0,0 +1,998 @@
//! Characterization tests for the inbound-handshake (`handle_msg1`) establish
//! branches.
//!
//! These lock in the *current* observable behavior of the undertested inbound
//! classification paths by driving a REAL framed msg1 into `handle_msg1`
//! (constructing a genuine Noise IK msg1 and delivering it as a
//! `ReceivedPacket`), rather than poking rekey state directly the way the
//! `arm_rekey` helper does. They are an oracle for a later behavior-neutral
//! refactor: assertions capture what happens today, surprising or not.
//!
//! Coverage map (branch → test):
//! * epoch-restart → `chartest_msg1_epoch_restart_replaces_active_peer`
//! * duplicate (pre-crypto) → `chartest_msg1_duplicate_pending_resends_stored_msg2`
//! * duplicate (post-crypto) → `chartest_msg1_duplicate_active_same_epoch_resends_stored_msg2`
//! * cross-connection precedence→ `chartest_msg1_inbound_promote_defers_pending_outbound_to_same_identity`
//! * max-peers cap (bypass) → `chartest_msg1_at_cap_with_pending_outbound_bypasses_early_gate`
//! * tie-break (winner+loser) → `chartest_cross_connection_tiebreak_winner_and_loser`
//! * rekey-responder → `chartest_msg1_rekey_responder_stores_pending_session`
//! * rekey dual-init (we win) → `chartest_msg1_rekey_dual_init_we_win_drops_their_msg1`
//! * rekey dual-init (we lose) → `chartest_msg1_rekey_dual_init_we_lose_becomes_responder`
//!
//! The three rekey branches sit behind the hardcoded
//! `existing_session_age_secs >= 30` guard in `handle_msg1`, resolved from
//! `ActivePeer::session_established_at()` (a monotonic `std::time::Instant`
//! with no natural test seam — the field is private and `tokio::time` cannot
//! advance a std `Instant`). They are unblocked by the sole `#[cfg(test)]`
//! production seam `ActivePeer::test_backdate_session_established(age)`, which
//! only shifts that private timestamp — it changes no decision logic and no
//! threshold, and is compiled out of release builds.
use super::*;
use crate::config::UdpConfig;
use crate::noise::HandshakeState;
use crate::peer::ActivePeer;
use crate::transport::udp::UdpTransport;
use crate::transport::{TransportHandle, packet_channel};
use tokio::time::timeout;
/// Build a genuine wire-format Noise IK msg1 addressed to `node`, carrying a
/// chosen startup `epoch` and `sender_index`, from `sender`'s identity. Returns
/// the opaque wire bytes ready to place in a `ReceivedPacket`.
fn craft_msg1_wire(
node: &Node,
sender: &Identity,
epoch: [u8; 8],
sender_index: SessionIndex,
ts: u64,
) -> Vec<u8> {
use crate::node::wire::build_msg1;
let peer_b_identity = PeerIdentity::from_pubkey_full(node.identity().pubkey_full());
let link_id = LinkId::new(0x0BAD_C0DE);
let mut conn = PeerConnection::outbound(link_id, peer_b_identity, ts);
let noise_msg1 = conn
.start_handshake(sender.keypair(), epoch, ts)
.expect("start_handshake produces noise msg1");
build_msg1(sender_index, &noise_msg1)
}
/// Register a real UDP transport on `node` and return an independent socket
/// (plus its addr) that plays the peer: the node's msg2 responses are sent to
/// this addr, so a test can observe wire-level output.
async fn register_udp_with_peer_socket(
node: &mut Node,
transport_id: TransportId,
) -> (tokio::net::UdpSocket, TransportAddr) {
let peer_sock = tokio::net::UdpSocket::bind("127.0.0.1:0")
.await
.expect("bind peer socket");
let peer_addr = TransportAddr::from_string(&peer_sock.local_addr().unwrap().to_string());
let cfg = UdpConfig {
bind_addr: Some("127.0.0.1:0".to_string()),
mtu: Some(1280),
..Default::default()
};
let (tx, _rx) = packet_channel(64);
let mut transport = UdpTransport::new(transport_id, None, cfg, tx);
transport.start_async().await.unwrap();
node.transports
.insert(transport_id, TransportHandle::Udp(transport));
(peer_sock, peer_addr)
}
/// Local re-impl of the `unit.rs` dummy-peer injector (that one is private to
/// its module). Fills the peer table with distinct identities so cap tests can
/// reach saturation.
fn inject_dummy_peers(node: &mut Node, count: usize) {
for i in 0..count {
let identity = make_peer_identity();
let addr = *identity.node_addr();
let peer = ActivePeer::new(identity, LinkId::new((i + 1) as u64), 0);
node.peers.insert(addr, peer);
}
}
/// Epoch-restart: an inbound msg1 from an already-active peer that carries a
/// DIFFERENT startup epoch is treated as a peer restart. The stale peer is torn
/// down and the fresh handshake is promoted in its place.
///
/// Oracle: after the push, the identity is still present but is a NEW peer —
/// it now holds a live Noise session (the stale one had none), its stored
/// remote epoch has advanced to the restart value, and it occupies a fresh link
/// and session index. `schedule_reconnect` is a no-op under a bare test config
/// (no auto-connect peer configured), so `retry_pending` stays empty.
#[tokio::test]
async fn chartest_msg1_epoch_restart_replaces_active_peer() {
let mut node = make_node();
let transport_id = TransportId::new(1);
let (peer_sock, peer_addr) = register_udp_with_peer_socket(&mut node, transport_id).await;
let sender = Identity::generate();
let sender_pid = PeerIdentity::from_pubkey_full(sender.pubkey_full());
let sender_addr = *sender_pid.node_addr();
let old_epoch = [1u8; 8];
let new_epoch = [2u8; 8];
let old_link = LinkId::new(4242);
// Pre-existing active peer at the OLD epoch, sessionless. `current_addr`
// makes `should_admit_msg1` recognize the source as an established peer.
let mut old_peer = ActivePeer::new(sender_pid, old_link, 1000);
old_peer.set_remote_epoch(Some(old_epoch));
old_peer.set_current_addr(transport_id, peer_addr.clone());
node.peers.insert(sender_addr, old_peer);
assert!(!node.get_peer(&sender_addr).unwrap().has_session());
assert_eq!(
node.get_peer(&sender_addr).unwrap().remote_epoch(),
Some(old_epoch)
);
// Real msg1 carrying the NEW (restart) epoch.
let data = craft_msg1_wire(&node, &sender, new_epoch, SessionIndex::new(0x77), 2000);
let packet = ReceivedPacket {
transport_id,
remote_addr: peer_addr.clone(),
data,
timestamp_ms: 2000,
};
node.handle_msg1(packet).await;
let peer = node
.get_peer(&sender_addr)
.expect("restarted peer must remain present (replaced, not dropped)");
assert!(
peer.has_session(),
"restart promotes a fresh handshake, so the new peer holds a session"
);
assert_eq!(
peer.remote_epoch(),
Some(new_epoch),
"stored remote epoch advances to the restart value"
);
assert_ne!(
peer.link_id(),
old_link,
"restart replaces the link with the freshly allocated one"
);
let our_index = peer.our_index().expect("promoted peer has our_index");
assert!(
node.peers_by_index
.contains_key(&(transport_id, our_index.as_u32())),
"fresh session index registered in peers_by_index"
);
assert_eq!(node.peer_count(), 1, "old peer removed, new peer added");
assert!(
node.retry_pending.is_empty(),
"schedule_reconnect is a no-op with no auto-connect config"
);
// A msg2 response was emitted to the restarting peer.
let mut buf = [0u8; 2048];
let got = timeout(Duration::from_millis(500), peer_sock.recv_from(&mut buf)).await;
assert!(
got.is_ok(),
"restart path must emit a msg2 response to the peer"
);
}
/// Duplicate msg1, pre-crypto short-circuit: a second msg1 from an address that
/// already has a genuinely-pending (not yet promoted) inbound link resends the
/// stored msg2 without paying the crypto cost or touching registry state.
///
/// Oracle: the exact stored msg2 bytes are resent verbatim, nothing is
/// promoted, the pending connection is left intact, and the msg1 rate limiter
/// rebalances (start then complete) to baseline.
#[tokio::test]
async fn chartest_msg1_duplicate_pending_resends_stored_msg2() {
let mut node = make_node();
let transport_id = TransportId::new(1);
let (peer_sock, peer_addr) = register_udp_with_peer_socket(&mut node, transport_id).await;
// A pending inbound connection with a stored msg2, keyed in addr_to_link,
// NOT promoted to an active peer.
let link_id = node.allocate_link_id();
let mut conn =
PeerConnection::inbound_with_transport(link_id, transport_id, peer_addr.clone(), 1000);
let stored_msg2 = vec![0xC1, 0xC2, 0xC3, 0xC4, 0xC5];
conn.set_handshake_msg2(stored_msg2.clone());
let link = Link::connectionless(
link_id,
transport_id,
peer_addr.clone(),
LinkDirection::Inbound,
Duration::from_millis(100),
);
node.links.insert(link_id, link);
node.addr_to_link
.insert((transport_id, peer_addr.clone()), link_id);
node.connections.insert(link_id, conn);
assert_eq!(node.peer_count(), 0);
let before_pending = node.msg1_rate_limiter.pending_count();
// Duplicate msg1 from the same address. Content is irrelevant past a valid
// header — the pre-crypto branch fires before any decrypt.
let sender = Identity::generate();
let data = craft_msg1_wire(&node, &sender, [9u8; 8], SessionIndex::new(5), 2000);
let packet = ReceivedPacket {
transport_id,
remote_addr: peer_addr.clone(),
data,
timestamp_ms: 2000,
};
node.handle_msg1(packet).await;
let mut buf = [0u8; 2048];
let (n, _) = timeout(Duration::from_millis(500), peer_sock.recv_from(&mut buf))
.await
.expect("stored msg2 must be resent")
.expect("recv_from");
assert_eq!(
&buf[..n],
&stored_msg2[..],
"the exact stored msg2 is resent for a duplicate msg1"
);
assert_eq!(node.peer_count(), 0, "duplicate msg1 promotes nothing");
assert!(
node.connections.contains_key(&link_id),
"pending connection is left intact"
);
assert_eq!(
node.msg1_rate_limiter.pending_count(),
before_pending,
"rate limiter rebalances to baseline"
);
}
/// Duplicate msg1, post-crypto same-epoch path: an inbound msg1 from an active
/// peer at the SAME epoch, on a session too young (< 30s) to be a rekey, is a
/// duplicate. The peer's stored msg2 is resent.
///
/// Oracle: the active peer's stored msg2 is resent, no new peer or session
/// index is allocated, and the existing peer is untouched.
#[tokio::test]
async fn chartest_msg1_duplicate_active_same_epoch_resends_stored_msg2() {
let mut node = make_node();
let transport_id = TransportId::new(1);
let (peer_sock, peer_addr) = register_udp_with_peer_socket(&mut node, transport_id).await;
let sender = Identity::generate();
let sender_pid = PeerIdentity::from_pubkey_full(sender.pubkey_full());
let sender_addr = *sender_pid.node_addr();
let epoch = [7u8; 8];
let stored_msg2 = vec![0xD0, 0xD1, 0xD2, 0xD3];
let link_id = LinkId::new(555);
let mut peer = ActivePeer::new(sender_pid, link_id, 1000);
peer.set_remote_epoch(Some(epoch));
peer.set_current_addr(transport_id, peer_addr.clone());
peer.set_handshake_msg2(stored_msg2.clone());
node.peers.insert(sender_addr, peer);
// Session age is ~0 (< 30s) → the rekey gate is false → a same-epoch msg1
// classifies as a duplicate, not a rekey initiation.
assert!(!node.get_peer(&sender_addr).unwrap().has_session());
let data = craft_msg1_wire(&node, &sender, epoch, SessionIndex::new(0x33), 2000);
let packet = ReceivedPacket {
transport_id,
remote_addr: peer_addr.clone(),
data,
timestamp_ms: 2000,
};
node.handle_msg1(packet).await;
let mut buf = [0u8; 2048];
let (n, _) = timeout(Duration::from_millis(500), peer_sock.recv_from(&mut buf))
.await
.expect("stored msg2 must be resent")
.expect("recv_from");
assert_eq!(&buf[..n], &stored_msg2[..]);
assert_eq!(node.peer_count(), 1);
assert_eq!(
node.get_peer(&sender_addr).unwrap().link_id(),
link_id,
"existing peer untouched by a duplicate msg1"
);
assert!(
node.peers_by_index.is_empty(),
"no new session index allocated on the duplicate path"
);
}
/// Cross-connection precedence: an inbound establish that promotes a peer while
/// a concurrent PENDING OUTBOUND connection to the SAME identity exists must NOT
/// tear that outbound down — it is deferred (kept alive so its later msg2 can
/// update `their_index` on the promoted peer).
///
/// Oracle: the inbound msg1 promotes the peer (with a live session), and both
/// the pending outbound connection and its `pending_outbound` index entry are
/// preserved.
#[tokio::test]
async fn chartest_msg1_inbound_promote_defers_pending_outbound_to_same_identity() {
let mut node = make_node();
let transport_id = TransportId::new(1);
let (_peer_sock, inbound_addr) = register_udp_with_peer_socket(&mut node, transport_id).await;
let sender = Identity::generate();
let sender_pid = PeerIdentity::from_pubkey_full(sender.pubkey_full());
let sender_addr = *sender_pid.node_addr();
// A concurrent pending OUTBOUND connection to the same identity, at a
// different source address.
let out_link = node.allocate_link_id();
let out_addr = TransportAddr::from_string("10.0.0.9:2121");
let mut out_conn = PeerConnection::outbound(out_link, sender_pid, 1000);
let our_keypair = node.identity().keypair();
let _ = out_conn
.start_handshake(our_keypair, node.startup_epoch(), 1000)
.unwrap();
let out_index = node.index_allocator.allocate().unwrap();
out_conn.set_our_index(out_index);
out_conn.set_transport_id(transport_id);
out_conn.set_source_addr(out_addr.clone());
let out_l = Link::connectionless(
out_link,
transport_id,
out_addr.clone(),
LinkDirection::Outbound,
Duration::from_millis(100),
);
node.links.insert(out_link, out_l);
node.addr_to_link
.insert((transport_id, out_addr.clone()), out_link);
node.connections.insert(out_link, out_conn);
node.pending_outbound
.insert((transport_id, out_index.as_u32()), out_link);
assert_eq!(node.peer_count(), 0);
// Inbound msg1 from the same identity, different source addr.
let data = craft_msg1_wire(&node, &sender, [3u8; 8], SessionIndex::new(0x22), 2000);
let packet = ReceivedPacket {
transport_id,
remote_addr: inbound_addr.clone(),
data,
timestamp_ms: 2000,
};
node.handle_msg1(packet).await;
let peer = node
.get_peer(&sender_addr)
.expect("inbound establish must promote the peer");
assert!(peer.has_session());
assert_eq!(node.peer_count(), 1);
assert!(
node.connections.contains_key(&out_link),
"pending outbound to the same identity must be preserved (deferred cleanup)"
);
assert!(
node.pending_outbound
.contains_key(&(transport_id, out_index.as_u32())),
"the outbound pending_outbound entry is preserved for msg2 index-learning"
);
}
/// Max-peers cap, pending-outbound bypass: at saturation, a msg1 from a NEW
/// identity that already has a pending outbound to it is NOT silent-dropped by
/// the early cap gate — it proceeds far enough to emit a msg2, then the late gate
/// inside `promote_connection` rejects it (peer table is full).
///
/// Oracle discriminator vs. the plain new-peer silent-drop: a msg2 IS observed
/// on the wire (the early gate was bypassed), yet the peer is NOT promoted (the
/// late gate rejects). This locks in the asymmetry between the two cap gates.
#[tokio::test]
async fn chartest_msg1_at_cap_with_pending_outbound_bypasses_early_gate() {
let mut node = make_node_with_max_peers(2);
let transport_id = TransportId::new(1);
let (peer_sock, peer_addr) = register_udp_with_peer_socket(&mut node, transport_id).await;
inject_dummy_peers(&mut node, 2);
assert_eq!(node.peer_count(), 2, "precondition: at cap");
let sender = Identity::generate();
let sender_pid = PeerIdentity::from_pubkey_full(sender.pubkey_full());
let sender_addr = *sender_pid.node_addr();
// A pending outbound to the (new) sender identity — this sets
// `has_pending_outbound_to_peer`, which turns off the early silent-drop.
let out_link = node.allocate_link_id();
let out_addr = TransportAddr::from_string("10.0.0.9:2121");
let mut out_conn = PeerConnection::outbound(out_link, sender_pid, 1000);
let our_keypair = node.identity().keypair();
let _ = out_conn
.start_handshake(our_keypair, node.startup_epoch(), 1000)
.unwrap();
let out_index = node.index_allocator.allocate().unwrap();
out_conn.set_our_index(out_index);
out_conn.set_transport_id(transport_id);
out_conn.set_source_addr(out_addr.clone());
node.connections.insert(out_link, out_conn);
node.pending_outbound
.insert((transport_id, out_index.as_u32()), out_link);
let data = craft_msg1_wire(&node, &sender, [4u8; 8], SessionIndex::new(0x44), 2000);
let packet = ReceivedPacket {
transport_id,
remote_addr: peer_addr.clone(),
data,
timestamp_ms: 2000,
};
node.handle_msg1(packet).await;
// Late gate rejects: still at cap, sender not promoted.
assert_eq!(
node.peer_count(),
2,
"late cap gate rejects the new identity"
);
assert!(
!node.peers.contains_key(&sender_addr),
"new identity is not adopted at capacity"
);
// But the early gate was bypassed: a msg2 WAS put on the wire before the
// late-gate rejection (the discriminator against the plain silent-drop).
let mut buf = [0u8; 2048];
let got = timeout(Duration::from_millis(500), peer_sock.recv_from(&mut buf)).await;
assert!(
got.is_ok() && got.unwrap().is_ok(),
"pending-outbound identity bypasses the early silent-drop, so a msg2 \
is emitted before the late cap gate rejects"
);
}
/// Cross-connection tie-break, winner AND loser in one deterministic run: both
/// nodes initiate to each other (simultaneous cross-connection). The rule is
/// "the smaller node_addr's OUTBOUND wins" (`cross_connection_winner`). After
/// both sides exchange msg1 (promote inbound) and msg2 (resolve), the winner has
/// swapped to its outbound session index while the loser keeps the inbound index
/// it assigned during its own msg1 handling.
///
/// Oracle: the smaller-addr node's peer.our_index equals the OUTBOUND index it
/// allocated at setup; the larger-addr node's peer.our_index equals the INBOUND
/// index it assigned while promoting the peer's msg1.
#[tokio::test]
async fn chartest_cross_connection_tiebreak_winner_and_loser() {
use crate::node::wire::build_msg1;
let mut node_a = make_node();
let mut node_b = make_node();
let transport_id_a = TransportId::new(1);
let transport_id_b = TransportId::new(1);
let udp_config = UdpConfig {
bind_addr: Some("127.0.0.1:0".to_string()),
mtu: Some(1280),
..Default::default()
};
let (packet_tx_a, mut packet_rx_a) = packet_channel(64);
let (packet_tx_b, mut packet_rx_b) = packet_channel(64);
let mut transport_a = UdpTransport::new(transport_id_a, None, udp_config.clone(), packet_tx_a);
let mut transport_b = UdpTransport::new(transport_id_b, None, udp_config, packet_tx_b);
transport_a.start_async().await.unwrap();
transport_b.start_async().await.unwrap();
let addr_a = transport_a.local_addr().unwrap();
let addr_b = transport_b.local_addr().unwrap();
let remote_addr_b = TransportAddr::from_string(&addr_b.to_string());
let remote_addr_a = TransportAddr::from_string(&addr_a.to_string());
node_a
.transports
.insert(transport_id_a, TransportHandle::Udp(transport_a));
node_b
.transports
.insert(transport_id_b, TransportHandle::Udp(transport_b));
let peer_b_identity = PeerIdentity::from_pubkey_full(node_b.identity().pubkey_full());
let peer_a_identity = PeerIdentity::from_pubkey_full(node_a.identity().pubkey_full());
let node_a_addr = *node_a.node_addr();
let node_b_addr = *node_b.node_addr();
// A initiates to B.
let link_a_out = node_a.allocate_link_id();
let mut conn_a = PeerConnection::outbound(link_a_out, peer_b_identity, 1000);
let out_index_a = node_a.index_allocator.allocate().unwrap();
let noise_msg1_a = conn_a
.start_handshake(node_a.identity().keypair(), node_a.startup_epoch(), 1000)
.unwrap();
conn_a.set_our_index(out_index_a);
conn_a.set_transport_id(transport_id_a);
conn_a.set_source_addr(remote_addr_b.clone());
let wire_msg1_a = build_msg1(out_index_a, &noise_msg1_a);
node_a.links.insert(
link_a_out,
Link::connectionless(
link_a_out,
transport_id_a,
remote_addr_b.clone(),
LinkDirection::Outbound,
Duration::from_millis(100),
),
);
node_a
.addr_to_link
.insert((transport_id_a, remote_addr_b.clone()), link_a_out);
node_a.connections.insert(link_a_out, conn_a);
node_a
.pending_outbound
.insert((transport_id_a, out_index_a.as_u32()), link_a_out);
// B initiates to A.
let link_b_out = node_b.allocate_link_id();
let mut conn_b = PeerConnection::outbound(link_b_out, peer_a_identity, 1000);
let out_index_b = node_b.index_allocator.allocate().unwrap();
let noise_msg1_b = conn_b
.start_handshake(node_b.identity().keypair(), node_b.startup_epoch(), 1000)
.unwrap();
conn_b.set_our_index(out_index_b);
conn_b.set_transport_id(transport_id_b);
conn_b.set_source_addr(remote_addr_a.clone());
let wire_msg1_b = build_msg1(out_index_b, &noise_msg1_b);
node_b.links.insert(
link_b_out,
Link::connectionless(
link_b_out,
transport_id_b,
remote_addr_a.clone(),
LinkDirection::Outbound,
Duration::from_millis(100),
),
);
node_b
.addr_to_link
.insert((transport_id_b, remote_addr_a.clone()), link_b_out);
node_b.connections.insert(link_b_out, conn_b);
node_b
.pending_outbound
.insert((transport_id_b, out_index_b.as_u32()), link_b_out);
// Both put msg1 on the wire.
node_a
.transports
.get(&transport_id_a)
.unwrap()
.send(&remote_addr_b, &wire_msg1_a)
.await
.unwrap();
node_b
.transports
.get(&transport_id_b)
.unwrap()
.send(&remote_addr_a, &wire_msg1_b)
.await
.unwrap();
// Each processes the other's msg1 (promotes inbound, assigns an inbound index).
let pkt_at_b = timeout(Duration::from_secs(1), packet_rx_b.recv())
.await
.unwrap()
.unwrap();
node_b.handle_msg1(pkt_at_b).await;
let pkt_at_a = timeout(Duration::from_secs(1), packet_rx_a.recv())
.await
.unwrap()
.unwrap();
node_a.handle_msg1(pkt_at_a).await;
// Inbound indices assigned during promotion (before resolution).
let inbound_index_a = node_a.get_peer(&node_b_addr).unwrap().our_index().unwrap();
let inbound_index_b = node_b.get_peer(&node_a_addr).unwrap().our_index().unwrap();
// Each processes the other's msg2 (cross-connection resolution).
let msg2_at_a = timeout(Duration::from_secs(1), packet_rx_a.recv())
.await
.unwrap()
.unwrap();
node_a.handle_msg2(msg2_at_a).await;
let msg2_at_b = timeout(Duration::from_secs(1), packet_rx_b.recv())
.await
.unwrap()
.unwrap();
node_b.handle_msg2(msg2_at_b).await;
// Rule: smaller node_addr's OUTBOUND wins → that node swaps to its outbound
// index; the larger node keeps the inbound index from its own msg1 handling.
let a_is_winner = node_a_addr < node_b_addr;
let final_our_index_a = node_a.get_peer(&node_b_addr).unwrap().our_index().unwrap();
let final_our_index_b = node_b.get_peer(&node_a_addr).unwrap().our_index().unwrap();
if a_is_winner {
assert_eq!(
final_our_index_a, out_index_a,
"winner (smaller addr) swaps to its outbound session index"
);
assert_eq!(
final_our_index_b, inbound_index_b,
"loser (larger addr) keeps the inbound index from its msg1 handling"
);
} else {
assert_eq!(
final_our_index_b, out_index_b,
"winner (smaller addr) swaps to its outbound session index"
);
assert_eq!(
final_our_index_a, inbound_index_a,
"loser (larger addr) keeps the inbound index from its msg1 handling"
);
}
// Both remain single, sendable peers after resolution.
assert_eq!(node_a.peer_count(), 1);
assert_eq!(node_b.peer_count(), 1);
assert!(node_a.get_peer(&node_b_addr).unwrap().can_send());
assert!(node_b.get_peer(&node_a_addr).unwrap().can_send());
for (_, t) in node_a.transports.iter_mut() {
t.stop().await.ok();
}
for (_, t) in node_b.transports.iter_mut() {
t.stop().await.ok();
}
}
// ===========================================================================
// Rekey establish branches (unblocked by the `#[cfg(test)]`
// `ActivePeer::test_backdate_session_established` seam that lets a test age a
// real session past the hardcoded 30s rekey gate in `handle_msg1`).
// ===========================================================================
/// Drive a real inbound msg1 through `handle_msg1` so `node` promotes an active
/// peer for `sender` at startup `epoch`, draining the msg2 the promotion emits.
/// Returns the sender's NodeAddr.
async fn establish_active_peer_via_msg1(
node: &mut Node,
sender: &Identity,
epoch: [u8; 8],
transport_id: TransportId,
peer_addr: &TransportAddr,
peer_sock: &tokio::net::UdpSocket,
ts: u64,
) -> NodeAddr {
let sender_addr = *PeerIdentity::from_pubkey_full(sender.pubkey_full()).node_addr();
let data = craft_msg1_wire(node, sender, epoch, SessionIndex::new(0x01), ts);
let packet = ReceivedPacket {
transport_id,
remote_addr: peer_addr.clone(),
data,
timestamp_ms: ts,
};
node.handle_msg1(packet).await;
// Promotion emits a msg2 AND an initial TreeAnnounce; drain every queued
// datagram so a later recv observes only the rekey response (or its
// absence), never a leftover from establishment.
let mut buf = [0u8; 2048];
while timeout(Duration::from_millis(150), peer_sock.recv_from(&mut buf))
.await
.is_ok()
{}
sender_addr
}
/// Draw a fresh sender identity whose NodeAddr is greater-than (`want_greater`)
/// or less-than the node's own NodeAddr, so the dual-init tie-break outcome is
/// deterministic. The comparison invariant is enforced, so the test outcome is
/// deterministic even though the identity draw is random.
fn sender_with_addr_relation(node: &Node, want_greater: bool) -> Identity {
let node_addr = *node.node_addr();
loop {
let s = Identity::generate();
let a = *PeerIdentity::from_pubkey_full(s.pubkey_full()).node_addr();
if a != node_addr && (a > node_addr) == want_greater {
return s;
}
}
}
/// Arm a local in-flight (initiator) rekey on `node`'s peer for `sender`, with a
/// real allocated index registered in `peers_by_index`/`pending_outbound` (as a
/// genuine in-flight rekey would be). Returns the armed rekey index.
fn arm_local_rekey(
node: &mut Node,
sender: &Identity,
sender_addr: &NodeAddr,
transport_id: TransportId,
) -> SessionIndex {
let rekey_index = node.index_allocator.allocate().unwrap();
node.peers_by_index
.insert((transport_id, rekey_index.as_u32()), *sender_addr);
node.pending_outbound
.insert((transport_id, rekey_index.as_u32()), LinkId::new(0xF00D));
let local = Identity::generate();
let hs = HandshakeState::new_initiator(local.keypair(), sender.pubkey_full());
node.get_peer_mut(sender_addr)
.unwrap()
.set_rekey_state(hs, rekey_index, vec![0xAB; 64], 0);
rekey_index
}
/// Rekey-responder: a genuine rekey msg1 (a fresh IK handshake at the SAME
/// epoch) arriving for an active peer whose session is past the 30s gate is
/// processed as a rekey. The responder extracts the new session and holds it as
/// PENDING (awaiting K-bit cutover) without disturbing the live session.
///
/// Oracle: the new session lands in `pending_new_session` with a freshly
/// allocated `pending_our_index` and `pending_their_index` == the rekey msg1
/// sender index; the current session/index stay live and registered; the new
/// index is additionally registered in `peers_by_index`; and a rekey msg2 is
/// emitted. The peer is neither replaced nor left `rekey_in_progress`.
#[tokio::test]
async fn chartest_msg1_rekey_responder_stores_pending_session() {
let mut node = make_node();
let transport_id = TransportId::new(1);
let (peer_sock, peer_addr) = register_udp_with_peer_socket(&mut node, transport_id).await;
let sender = Identity::generate();
let epoch = [5u8; 8];
let sender_addr = establish_active_peer_via_msg1(
&mut node,
&sender,
epoch,
transport_id,
&peer_addr,
&peer_sock,
1000,
)
.await;
let old_index = {
let p = node.get_peer(&sender_addr).expect("peer established");
assert!(p.has_session());
assert!(p.is_healthy());
assert_eq!(p.remote_epoch(), Some(epoch));
assert!(!p.rekey_in_progress());
assert!(p.pending_new_session().is_none());
p.our_index().unwrap()
};
assert!(
node.peers_by_index
.contains_key(&(transport_id, old_index.as_u32()))
);
let index_count_before = node.index_allocator.count();
// Age the live session past the 30s rekey gate (test-only seam).
node.get_peer_mut(&sender_addr)
.unwrap()
.test_backdate_session_established(Duration::from_secs(31));
// A genuine rekey msg1 (fresh IK handshake, SAME epoch) arrives.
let rekey_sender_index = SessionIndex::new(0xBEEF);
let data = craft_msg1_wire(&node, &sender, epoch, rekey_sender_index, 2000);
let packet = ReceivedPacket {
transport_id,
remote_addr: peer_addr.clone(),
data,
timestamp_ms: 2000,
};
node.handle_msg1(packet).await;
let p = node
.get_peer(&sender_addr)
.expect("peer still present (not replaced)");
assert_eq!(
node.peer_count(),
1,
"rekey neither adds nor replaces the peer"
);
assert!(
p.pending_new_session().is_some(),
"new session held as pending"
);
let new_index = p.pending_our_index().expect("pending our_index allocated");
assert_eq!(
p.pending_their_index(),
Some(rekey_sender_index),
"pending their_index = the rekey msg1 sender index"
);
assert!(
!p.rekey_in_progress(),
"set_pending_session clears rekey_in_progress"
);
assert_eq!(
p.our_index(),
Some(old_index),
"current session index stays live until cutover"
);
assert!(p.has_session(), "current session remains live");
assert!(
node.peers_by_index
.contains_key(&(transport_id, old_index.as_u32())),
"current index still registered"
);
assert!(
node.peers_by_index
.contains_key(&(transport_id, new_index.as_u32())),
"new pending index registered"
);
assert_eq!(
node.index_allocator.count(),
index_count_before + 1,
"exactly one extra index allocated for the pending session"
);
let mut buf = [0u8; 2048];
let got = timeout(Duration::from_millis(500), peer_sock.recv_from(&mut buf)).await;
assert!(
got.is_ok() && got.unwrap().is_ok(),
"rekey responder emits a rekey msg2"
);
}
/// Rekey dual-init, WE WIN: with a local rekey in flight, a simultaneous rekey
/// msg1 arrives from a peer whose NodeAddr is larger than ours. The tie-break
/// ("smaller NodeAddr wins as initiator") makes us the winner, so we DROP their
/// msg1 and keep driving our own rekey.
///
/// Oracle: our in-flight rekey is untouched (`rekey_in_progress` stays true, our
/// rekey index retained and still registered), no responder session is stored,
/// no responder index is allocated, and no rekey msg2 is emitted.
#[tokio::test]
async fn chartest_msg1_rekey_dual_init_we_win_drops_their_msg1() {
let mut node = make_node();
let transport_id = TransportId::new(1);
let (peer_sock, peer_addr) = register_udp_with_peer_socket(&mut node, transport_id).await;
// We win when our node_addr < peer's → pick a sender greater than us.
let sender = sender_with_addr_relation(&node, true);
let epoch = [6u8; 8];
let sender_addr = establish_active_peer_via_msg1(
&mut node,
&sender,
epoch,
transport_id,
&peer_addr,
&peer_sock,
1000,
)
.await;
assert!(
*node.node_addr() < sender_addr,
"precondition: node wins the tie-break"
);
node.get_peer_mut(&sender_addr)
.unwrap()
.test_backdate_session_established(Duration::from_secs(31));
let rekey_index = arm_local_rekey(&mut node, &sender, &sender_addr, transport_id);
assert!(node.get_peer(&sender_addr).unwrap().rekey_in_progress());
let index_count_before = node.index_allocator.count();
// Their simultaneous rekey msg1 arrives.
let data = craft_msg1_wire(&node, &sender, epoch, SessionIndex::new(0xAAAA), 2000);
let packet = ReceivedPacket {
transport_id,
remote_addr: peer_addr.clone(),
data,
timestamp_ms: 2000,
};
node.handle_msg1(packet).await;
let p = node.get_peer(&sender_addr).expect("peer present");
assert!(
p.rekey_in_progress(),
"our rekey survives; the winner does not abandon it"
);
assert!(
p.pending_new_session().is_none(),
"no responder session stored on the winner path"
);
assert_eq!(
p.rekey_our_index(),
Some(rekey_index),
"our rekey index retained"
);
assert!(
node.peers_by_index
.contains_key(&(transport_id, rekey_index.as_u32())),
"our rekey index still registered in peers_by_index"
);
assert!(
node.pending_outbound
.contains_key(&(transport_id, rekey_index.as_u32())),
"our rekey pending_outbound entry retained"
);
assert_eq!(
node.index_allocator.count(),
index_count_before,
"no responder index allocated on the winner path"
);
let mut buf = [0u8; 2048];
let got = timeout(Duration::from_millis(300), peer_sock.recv_from(&mut buf)).await;
assert!(
got.is_err(),
"winner emits no msg2 in response to the dropped rekey msg1"
);
}
/// Rekey dual-init, WE LOSE: with a local rekey in flight, a simultaneous rekey
/// msg1 arrives from a peer whose NodeAddr is smaller than ours. The tie-break
/// makes us the loser, so we ABANDON our own rekey and respond as the rekey
/// responder.
///
/// Oracle: our in-flight rekey is abandoned (`rekey_in_progress` cleared, the
/// abandoned index freed and unregistered from `pending_outbound`), the new
/// session is stored as pending with `pending_their_index` == the rekey msg1
/// sender index, the pending index is registered, and a rekey msg2 is emitted.
#[tokio::test]
async fn chartest_msg1_rekey_dual_init_we_lose_becomes_responder() {
let mut node = make_node();
let transport_id = TransportId::new(1);
let (peer_sock, peer_addr) = register_udp_with_peer_socket(&mut node, transport_id).await;
// We lose when our node_addr > peer's → pick a sender smaller than us.
let sender = sender_with_addr_relation(&node, false);
let epoch = [6u8; 8];
let sender_addr = establish_active_peer_via_msg1(
&mut node,
&sender,
epoch,
transport_id,
&peer_addr,
&peer_sock,
1000,
)
.await;
assert!(
*node.node_addr() > sender_addr,
"precondition: node loses the tie-break"
);
node.get_peer_mut(&sender_addr)
.unwrap()
.test_backdate_session_established(Duration::from_secs(31));
let rekey_index = arm_local_rekey(&mut node, &sender, &sender_addr, transport_id);
assert!(node.get_peer(&sender_addr).unwrap().rekey_in_progress());
// Their simultaneous rekey msg1 arrives.
let rekey_sender_index = SessionIndex::new(0xCCCC);
let data = craft_msg1_wire(&node, &sender, epoch, rekey_sender_index, 2000);
let packet = ReceivedPacket {
transport_id,
remote_addr: peer_addr.clone(),
data,
timestamp_ms: 2000,
};
node.handle_msg1(packet).await;
let p = node.get_peer(&sender_addr).expect("peer present");
assert!(
!p.rekey_in_progress(),
"we abandoned our rekey and became responder"
);
assert!(
p.pending_new_session().is_some(),
"responder stores the new session as pending"
);
assert_eq!(
p.pending_their_index(),
Some(rekey_sender_index),
"pending their_index = the rekey msg1 sender index"
);
let new_index = p
.pending_our_index()
.expect("responder allocated a pending index");
assert!(
!node
.pending_outbound
.contains_key(&(transport_id, rekey_index.as_u32())),
"abandoned rekey pending_outbound entry removed"
);
assert!(
node.peers_by_index
.contains_key(&(transport_id, new_index.as_u32())),
"new pending index registered"
);
let mut buf = [0u8; 2048];
let got = timeout(Duration::from_millis(500), peer_sock.recv_from(&mut buf)).await;
assert!(
got.is_ok() && got.unwrap().is_ok(),
"loser (now responder) emits a rekey msg2"
);
}
src/node/tests/mod.rs
+1
View File
@@ -13,6 +13,7 @@ mod bootstrap;
mod decrypt_failure;
mod disconnect;
mod discovery;
mod establish_chartests;
#[cfg(target_os = "linux")]
mod ethernet;
mod forwarding;
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/unit.rs
+3 -3
View File
@@ -1260,7 +1260,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,
@@ -1296,7 +1296,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);
}
@@ -1308,7 +1308,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();