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

node: establish dataplane/ and session/ concept homes (behavior-neutral)

Browse Source 
Reorganize the node module tree by concept rather than by
message-handling verb, as the first step of the node runtime
decomposition. Pure relocation: no wire, config, metric, or log
semantics change; the lib test count is unchanged (1577 passed).

Moves (git mv, 100% rename similarity):
- handlers/{forwarding,rx_loop,connected_udp,dispatch,encrypted}.rs
  -> node/dataplane/ — the whole RX hot path (the select! run loop,
  transit/local forwarding, the link-message router, the RX decrypt
  path with responder K-bit cutover + roam writes, and connected-UDP
  fast-path activation) now lives in one home.
- node/session.rs -> node/session/mod.rs — establishes the session
  concept home for the data/state types. The message-behavior file
  handlers/session.rs stays put for now (folds in with the later FSP
  session step).

The IK/XX-divergent establishment files (handlers/{handshake,rekey,
timeout}.rs) and the deferred-home files (handlers/{mmp,lookup}.rs)
deliberately stay in handlers/, to move once rather than twice.

Every module is reached through impl Node methods, so no call site or
re-export shim was needed. Updated in lockstep with the moves: the
module_path!-derived tracing targets in the two mesh-lab compose-trace
overlays, a structural test's include_str! source path, doc-comments
in proto/routing and the mesh-lab docs, and the stale source-location
citations (node/handlers/{forwarding,rx_loop,encrypted}.rs and
node/session.rs) in doc-comments and the discovery design doc.
This commit is contained in:
Johnathan Corgan
2026-07-12 19:22:08 +00:00
parent 26d70ebb59
commit 434b9726aa
21 changed files with 34 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/node/handlers/connected_udp.rs
-236
View File
@@ -1,236 +0,0 @@
//! Lifecycle for per-peer connected UDP sockets.
//!
//! Tick-driven, idempotent, **on by default** for established UDP peers on
//! Linux and macOS:
//!
//! - **Tick-driven:** every node tick, scan established UDP peers
//! that don't yet have a connected socket installed and try to
//! open one. No need to thread an activation call through every
//! handshake-completion code path.
//! - **Idempotent:** if `peer.connected_udp()` is already `Some`,
//! skip. Replaces stale sockets lazily by clearing them on
//! address change / rekey from elsewhere (see
//! `deregister_session_index` and the rekey handler).
//!
//! Implementation note: only the **listen socket → wildcard** demux
//! path delivers the very first packets of a session (handshakes).
//! Once the peer's session is established, Linux/macOS install the connected
//! socket; from that moment on the kernel routes that peer's traffic
//! to it (most-specific 5-tuple match wins under `SO_REUSEPORT`), and
//! the drain thread feeds the existing `packet_tx` just like the
//! wildcard listen socket does. The rx_loop dispatch sees no
//! difference.
//!
//! macOS originally defaulted to the wildcard UDP socket because early
//! Darwin tests found liveness regressions under load. Later testing
//! showed the problem was mismatched listener/peer `SO_REUSE*` state:
//! with the live listener and connected sibling in the same reuse group,
//! the connected `send(2)` path improves the MacBook Wi-Fi sender case
//! and is now the default. Operators can still disable it with
//! `FIPS_MACOS_CONNECTED_UDP=0` or `FIPS_CONNECTED_UDP=0` for A/B tests.
use crate::NodeAddr;
use crate::node::Node;
#[cfg(any(target_os = "linux", target_os = "macos"))]
use crate::transport::TransportHandle;
#[cfg(any(target_os = "linux", target_os = "macos"))]
use std::sync::atomic::{AtomicU64, Ordering::Relaxed};
#[cfg(any(target_os = "linux", target_os = "macos"))]
use tracing::{debug, warn};
impl Node {
/// Tick-driven activation of per-peer connected UDP sockets.
/// Scans established UDP peers that don't yet have a connected
/// socket and opens one. No-op when there are no eligible peers
/// (e.g. only non-UDP transports). Enabled on Linux and macOS:
/// both kernels route a matching peer 5-tuple to the connected
/// socket when it shares the wildcard listen port via SO_REUSEPORT.
pub(in crate::node) async fn activate_connected_udp_sessions(&mut self) {
#[cfg(not(any(target_os = "linux", target_os = "macos")))]
{
// No-op on platforms without the connected-UDP fast path.
}
#[cfg(any(target_os = "linux", target_os = "macos"))]
{
if !connected_udp_enabled() {
return;
}
// Collect candidate NodeAddrs first so we can iterate
// without holding the &mut on self.peers across awaits.
let candidates: Vec<NodeAddr> = self
.peers
.iter()
.filter_map(|(addr, peer)| {
let has_session = peer.noise_session().is_some();
let has_transport = peer.transport_id().is_some();
let has_addr = peer.current_addr().is_some();
let already_active = peer.connected_udp().is_some();
if has_session && has_transport && has_addr && !already_active {
Some(*addr)
} else {
None
}
})
.collect();
for addr in candidates {
if let Err(e) = self.activate_connected_udp_for_peer(&addr).await {
static FAILURES: AtomicU64 = AtomicU64::new(0);
crate::perf_profile::record_event(
crate::perf_profile::Event::ConnectedUdpActivationFailed,
);
let n = FAILURES.fetch_add(1, Relaxed);
if n < 8 || n.is_multiple_of(1000) {
warn!(peer = %addr, error = %e, failures = n + 1, "connected UDP activation deferred");
} else {
debug!(peer = %addr, error = %e, "connected UDP activation deferred");
}
}
}
}
}
/// Open the connected UDP socket + spawn its drain thread for
/// one peer. Idempotent — re-checks the eligibility conditions
/// inside the &mut so a race with peer drop doesn't install on a
/// freshly-removed peer. Returns `Ok(())` on success or if the
/// peer is no longer eligible (treated as benign).
#[cfg(any(target_os = "linux", target_os = "macos"))]
async fn activate_connected_udp_for_peer(
&mut self,
node_addr: &NodeAddr,
) -> Result<(), String> {
// Read-only pass: figure out which transport + remote addr we need.
let (transport_id, peer_transport_addr) = {
let Some(peer) = self.peers.get(node_addr) else {
return Ok(());
};
if peer.connected_udp().is_some() {
return Ok(()); // already activated
}
let Some(tid) = peer.transport_id() else {
return Ok(());
};
let Some(addr) = peer.current_addr().cloned() else {
return Ok(());
};
(tid, addr)
};
// Resolve the peer's TransportAddr → kernel SocketAddr via
// the UDP transport's DNS cache. This may await on a DNS
// lookup the very first time we see a hostname; subsequent
// calls hit the cache.
let (peer_socket_addr, local_addr, recv_buf, send_buf, packet_tx) = {
let Some(transport) = self.transports.get(&transport_id) else {
return Ok(());
};
let udp = match transport {
TransportHandle::Udp(u) => u,
_ => return Ok(()), // not a UDP transport — feature N/A
};
let peer_sa = udp
.resolve_for_off_task(&peer_transport_addr)
.await
.map_err(|e| format!("address resolve: {e}"))?;
let local = udp
.local_addr()
.ok_or_else(|| "udp transport not started".to_string())?;
let recv_buf = udp.recv_buf_size();
let send_buf = udp.send_buf_size();
let tx = udp.clone_packet_tx();
(peer_sa, local, recv_buf, send_buf, tx)
};
// Open the connected socket on the kernel side, then adopt the
// fd into the owning handle.
let owned = crate::transport::udp::open_connected_fd(
local_addr,
peer_socket_addr,
recv_buf,
send_buf,
)
.map_err(|e| format!("open_connected_fd: {e}"))?;
let socket = std::sync::Arc::new(crate::peer::connected_udp::ConnectedPeerSocket::from_fd(
owned,
peer_socket_addr,
local_addr,
));
// Spawn the drain thread. It feeds `packet_tx` exactly like
// the wildcard listen socket — rx_loop dispatches identically.
let drain = crate::peer::connected_udp::PeerRecvDrain::spawn(
socket.clone(),
transport_id,
peer_socket_addr,
packet_tx,
)
.map_err(|e| format!("PeerRecvDrain::spawn: {e}"))?;
// Install on the peer, idempotent re-check.
if let Some(peer) = self.peers.get_mut(node_addr) {
if peer.connected_udp().is_some() {
// Lost the race — somebody else activated us first.
// Drop the new socket + drain so we don't leak.
drop(drain);
drop(socket);
return Ok(());
}
peer.set_connected_udp(socket, drain);
crate::perf_profile::record_event(crate::perf_profile::Event::ConnectedUdpInstalled);
debug!(
peer = %self.peer_display_name(node_addr),
peer_addr = %peer_socket_addr,
"connected UDP socket installed"
);
} else {
// Peer disappeared between read-only pass and now.
drop(drain);
drop(socket);
}
Ok(())
}
/// Clear the per-peer connected UDP socket + drain for a peer.
/// Called on peer disconnect / removal. The drain thread exits
/// via self-pipe; the kernel fd closes when the last `Arc`
/// drops.
#[cfg(any(target_os = "linux", target_os = "macos"))]
#[allow(dead_code)] // wired by session-deregister + rekey teardown follow-up
pub(in crate::node) fn clear_connected_udp_for_peer(&mut self, node_addr: &NodeAddr) {
if let Some(peer) = self.peers.get_mut(node_addr)
&& peer.connected_udp().is_some()
{
peer.clear_connected_udp();
debug!(peer = %self.peer_display_name(node_addr), "connected UDP socket cleared");
}
}
/// No-op shim for non-Linux builds so the rx_loop tick site can
/// call us unconditionally.
#[cfg(not(any(target_os = "linux", target_os = "macos")))]
#[allow(dead_code)] // wired by session-deregister + rekey teardown follow-up
pub(in crate::node) fn clear_connected_udp_for_peer(&mut self, _node_addr: &NodeAddr) {}
}
#[cfg(target_os = "linux")]
fn connected_udp_enabled() -> bool {
env_flag("FIPS_CONNECTED_UDP").unwrap_or(true)
}
#[cfg(target_os = "macos")]
fn connected_udp_enabled() -> bool {
env_flag("FIPS_MACOS_CONNECTED_UDP")
.or_else(|| env_flag("FIPS_CONNECTED_UDP"))
.unwrap_or(true)
}
#[cfg(any(target_os = "linux", target_os = "macos"))]
fn env_flag(name: &str) -> Option<bool> {
let value = std::env::var(name).ok()?;
match value.trim().to_ascii_lowercase().as_str() {
"1" | "true" | "yes" | "on" => Some(true),
"0" | "false" | "no" | "off" => Some(false),
_ => None,
}
}
src/node/handlers/dispatch.rs
-216
View File
@@ -1,216 +0,0 @@
//! Link message dispatch and peer removal.
use crate::NodeAddr;
use crate::node::Node;
use tracing::{debug, info, trace};
impl Node {
/// Dispatch a decrypted link message to the appropriate handler.
///
/// Link messages are protocol messages exchanged between authenticated peers.
pub(in crate::node) async fn dispatch_link_message(
&mut self,
from: &NodeAddr,
plaintext: &[u8],
ce_flag: bool,
) {
if plaintext.is_empty() {
return;
}
let msg_type = plaintext[0];
let payload = &plaintext[1..];
match msg_type {
0x00 => {
// SessionDatagram
self.handle_session_datagram(from, payload, ce_flag).await;
}
0x01 => {
// SenderReport
self.handle_sender_report(from, payload);
}
0x02 => {
// ReceiverReport
self.handle_receiver_report(from, payload).await;
}
0x10 => {
// TreeAnnounce
self.handle_tree_announce(from, payload).await;
}
0x20 => {
// FilterAnnounce
self.handle_filter_announce(from, payload).await;
}
0x30 => {
// LookupRequest
self.handle_lookup_request(from, payload).await;
}
0x31 => {
// LookupResponse
self.handle_lookup_response(from, payload).await;
}
0x50 => {
// Disconnect
self.handle_disconnect(from, payload);
}
0x51 => {
// Heartbeat — no-op, last_recv_time already updated by record_recv()
trace!(peer = %self.peer_display_name(from), "Received heartbeat");
}
_ => {
debug!(msg_type = msg_type, "Unknown link message type");
}
}
}
/// Handle a Disconnect notification from a peer.
///
/// The peer is signaling an orderly departure. We immediately remove
/// them from all state rather than waiting for timeout detection, and
/// schedule a reconnect if the peer is configured as auto-connect.
/// Without this, a graceful upstream shutdown orphans auto-connect
/// 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::proto::fmp::Disconnect::decode(payload) {
Ok(msg) => msg,
Err(e) => {
debug!(from = %self.peer_display_name(from), error = %e, "Malformed disconnect message");
return;
}
};
info!(
peer = %self.peer_display_name(from),
reason = %disconnect.reason,
"Peer sent disconnect notification"
);
let addr = *from;
self.remove_active_peer(from);
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(addr, now_ms);
}
/// Remove an active peer and clean up all associated state.
///
/// Frees session index, removes link and address mappings. Used for
/// both graceful disconnect and timeout-based eviction.
///
/// Also handles tree state cleanup: if the removed peer was our parent,
/// selects an alternative or becomes root, and marks remaining peers
/// for pending tree announce (delivered on next tick).
pub(in crate::node) fn remove_active_peer(&mut self, node_addr: &NodeAddr) {
let peer = match self.peers.remove(node_addr) {
Some(p) => p,
None => {
debug!(peer = %self.peer_display_name(node_addr), "Peer already removed");
return;
}
};
// Log suppressed replay detection summary before teardown
let suppressed = peer.replay_suppressed_count();
if suppressed > 0 {
debug!(
peer = %self.peer_display_name(node_addr),
count = suppressed,
"Suppressed replay detections during link transition"
);
}
// MMP teardown log (before we drop the peer)
let peer_name = self
.peer_aliases
.get(node_addr)
.cloned()
.unwrap_or_else(|| peer.identity().short_npub());
if let Some(mmp) = peer.mmp() {
Self::log_mmp_teardown(&peer_name, mmp);
}
// Remove any end-to-end session associated with this peer.
//
// Sessions are tracked separately from peers (self.sessions vs self.peers).
// Leaving a stale session alive after removing the peer causes:
// 1. check_session_mmp_reports() keeps logging stale "MMP session metrics"
// with frozen counters until purge_idle_sessions() eventually fires.
// 2. initiate_session() finds is_established() == true on the stale entry
// and silently returns Ok(()), preventing a new session from being
// established even after the link layer reconnects successfully.
if let Some(session_entry) = self.sessions.remove(node_addr)
&& let Some(mmp) = session_entry.mmp()
{
Self::log_session_mmp_teardown(&peer_name, mmp);
}
self.pending_tun_packets.remove(node_addr);
let link_id = peer.link_id();
let transport_id = peer.transport_id();
// Free session indices (current, rekey, pending, previous)
if let Some(tid) = transport_id {
if let Some(idx) = peer.our_index() {
let cache_key = (tid, idx.as_u32());
self.peers_by_index.remove(&cache_key);
#[cfg(unix)]
self.unregister_decrypt_worker_session(cache_key);
let _ = self.index_allocator.free(idx);
}
if let Some(idx) = peer.rekey_our_index() {
let cache_key = (tid, idx.as_u32());
self.pending_outbound.remove(&cache_key);
self.peers_by_index.remove(&cache_key);
#[cfg(unix)]
self.unregister_decrypt_worker_session(cache_key);
let _ = self.index_allocator.free(idx);
}
if let Some(idx) = peer.pending_our_index() {
let cache_key = (tid, idx.as_u32());
self.peers_by_index.remove(&cache_key);
#[cfg(unix)]
self.unregister_decrypt_worker_session(cache_key);
let _ = self.index_allocator.free(idx);
}
if let Some(idx) = peer.previous_our_index() {
let cache_key = (tid, idx.as_u32());
self.peers_by_index.remove(&cache_key);
#[cfg(unix)]
self.unregister_decrypt_worker_session(cache_key);
let _ = self.index_allocator.free(idx);
}
}
// Remove link and address mapping
self.remove_link(&link_id);
if let Some(transport_id) = transport_id {
self.cleanup_bootstrap_transport_if_unused(transport_id);
}
// Tree state cleanup
let tree_changed = self.handle_peer_removal_tree_cleanup(node_addr);
if tree_changed {
// Mark all remaining peers for pending tree announce.
// These will be sent on the next tick via check_tree_state().
for peer in self.peers.values_mut() {
peer.mark_tree_announce_pending();
}
}
// Bloom filter cleanup: clear state for removed peer, mark all remaining peers
self.bloom_state.remove_peer_state(node_addr);
let remaining_peers: Vec<NodeAddr> = self.peers.keys().copied().collect();
self.bloom_state.mark_all_updates_needed(remaining_peers);
debug!(
peer = %self.peer_display_name(node_addr),
link_id = %link_id,
tree_changed = tree_changed,
"Peer removed and state cleaned up"
);
}
}
src/node/handlers/encrypted.rs
-541
View File
@@ -1,541 +0,0 @@
//! Encrypted frame handling (hot path).
use crate::node::Node;
use crate::noise::NoiseError;
use crate::proto::fmp::wire::{
EncryptedHeader, FLAG_CE, FLAG_KEY_EPOCH, FLAG_SP, strip_inner_header,
};
use crate::transport::ReceivedPacket;
use tracing::{debug, trace, warn};
/// Force-remove a peer after this many consecutive decryption failures.
const DECRYPT_FAILURE_THRESHOLD: u32 = 20;
impl Node {
/// Handle an encrypted frame (phase 0x0).
///
/// This is the hot path for established sessions. We use O(1)
/// index-based lookup to find the session, then decrypt.
///
/// K-bit handling: when the peer flips the K-bit after a rekey,
/// we promote the pending new session to current and demote the old
/// session to previous for a drain window. During drain, we try the
/// current session first, then fall back to the previous session.
pub(in crate::node) async fn handle_encrypted_frame(&mut self, packet: ReceivedPacket) {
// Parse header (fail fast)
let header = match EncryptedHeader::parse(&packet.data) {
Some(h) => h,
None => return, // Malformed, drop silently
};
// O(1) session lookup by our receiver index
let key = (packet.transport_id, header.receiver_idx.as_u32());
let node_addr = match self.peers_by_index.get(&key) {
Some(id) => *id,
None => {
trace!(
receiver_idx = %header.receiver_idx,
transport_id = %packet.transport_id,
"Unknown session index, dropping"
);
return;
}
};
if !self.peers.contains_key(&node_addr) {
self.peers_by_index.remove(&key);
return;
}
// Extract K-bit from flags
let received_k_bit = header.flags & FLAG_KEY_EPOCH != 0;
// K-bit flip detection: peer has cut over to the new session.
//
// The header K-bit is NOT a sufficient gating event on its own.
// Under jitter the FMP rekey interval shrinks and the two
// directions' rekeys interleave, so a node can hold a `pending`
// session from rekey N while the peer's observed K-bit flip
// actually belongs to rekey N+1. Promoting on the bare bit then
// installs the WRONG Noise session as current — the two endpoints
// diverge, every subsequent frame fails AEAD on the far side, the
// receiver starves, and the link is declared dead at the heartbeat
// timeout (routing failure, green crypto). This mirrors the FSP fix
// (node/session.rs / node/handlers/session.rs): the authenticated
// decrypt, not the header bit, is the cutover signal. Trial-decrypt
// the frame against `pending` first; only promote if it
// authenticates. On success the same frame is delivered via
// `process_authentic_fmp_plaintext` and we return — it must not
// fall through to a second decrypt, which would be rejected as a
// replay (the trial-decrypt already advanced `pending`'s window).
{
let Some(peer) = self.peers.get(&node_addr) else {
return;
};
let k_bit_flipped =
received_k_bit != peer.current_k_bit() && peer.pending_new_session().is_some();
if k_bit_flipped {
let ciphertext = &packet.data[header.ciphertext_offset()..];
let display_name = self.peer_display_name(&node_addr);
let Some(peer) = self.peers.get_mut(&node_addr) else {
return;
};
// Authenticate the frame against the pending session.
// Trial-decrypt mutates `pending`'s replay window only on
// success, so a failed trial leaves it untouched.
let pending_plaintext = peer.pending_new_session_mut().and_then(|pending| {
pending
.decrypt_with_replay_check_and_aad(
ciphertext,
header.counter,
&header.header_bytes,
)
.ok()
});
if let Some(plaintext) = pending_plaintext {
debug!(
peer = %display_name,
"Peer new-epoch frame authenticated, K-bit flip promoting new session"
);
// The trial-decrypt already advanced the pending
// session's replay window; `handle_peer_kbit_flip`
// moves that same session object to `current`, so no
// re-decrypt.
let did_flip = peer.handle_peer_kbit_flip().is_some();
if did_flip {
// New index was pre-registered in peers_by_index
// during msg1 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 K-bit flip"
);
}
// Re-register the (now-promoted) session with the
// decrypt worker: cache_key = (transport_id, our_index)
// changed at the flip, so the old worker entry is
// stranded and every packet on the new session would
// miss the worker's HashMap lookup. Without this,
// throughput drops back to the inline-decrypt path
// after each rekey.
#[cfg(unix)]
if did_flip {
self.register_decrypt_worker_session(&node_addr);
}
// Deliver the frame we just authenticated via the
// canonical post-decrypt path, then return — it must
// not fall through to a second decrypt attempt.
let ce_flag = header.flags & FLAG_CE != 0;
let sp_flag = header.flags & FLAG_SP != 0;
self.process_authentic_fmp_plaintext(
&node_addr,
packet.transport_id,
&packet.remote_addr,
packet.timestamp_ms,
packet.data.len(),
header.counter,
ce_flag,
sp_flag,
&plaintext,
)
.await;
return;
}
// Pending did NOT authenticate this frame: the flip belongs
// to a different rekey epoch (stale pending). Do not
// promote. Fall through to the normal current/previous
// decrypt; the genuine cutover is recognized when a frame
// that authenticates against `pending` arrives.
}
}
// ── Decrypt-worker fast path (unix) ─────────────────────────
// Once the session has been registered with a decrypt shard
// (at FMP-establishment in `promote_connection`), the worker
// owns the FMP recv cipher + replay window. Dispatch the
// packet and return; the worker will run AEAD off-task and
// bounce the plaintext back via `decrypt_fallback_tx` for
// rx_loop to do the post-decrypt side-effects.
//
// The in-line decrypt below is the **synchronous test-mode
// path** for unit tests that construct `Node` without
// `lifecycle::start_async`; in production every established
// session is dispatched to the worker.
#[cfg(unix)]
{
let cache_key = (packet.transport_id, header.receiver_idx.as_u32());
if let Some(workers) = self.decrypt_workers.as_ref().cloned()
&& self.decrypt_registered_sessions.contains(&cache_key)
{
let job = crate::node::decrypt_worker::DecryptJob {
packet_data: packet.data,
cache_key,
_transport_id: packet.transport_id,
_remote_addr: packet.remote_addr,
timestamp_ms: packet.timestamp_ms,
source_node_addr: node_addr,
fmp_counter: header.counter,
fmp_flags: header.flags,
fmp_header: header.header_bytes,
fmp_ciphertext_offset: header.ciphertext_offset(),
fallback_tx: self.decrypt_fallback_tx.clone(),
};
workers.dispatch_job(job);
return;
}
}
// Decrypt: try current session first, then previous (drain fallback)
let ciphertext = &packet.data[header.ciphertext_offset()..];
let plaintext = {
let peer = self.peers.get_mut(&node_addr).unwrap();
let session = match peer.noise_session_mut() {
Some(s) => s,
None => {
warn!(
peer = %self.peer_display_name(&node_addr),
"Peer in index map has no session"
);
return;
}
};
match session.decrypt_with_replay_check_and_aad(
ciphertext,
header.counter,
&header.header_bytes,
) {
Ok(p) => {
peer.reset_decrypt_failures();
p
}
Err(e) => {
// Current session failed — try previous session (drain window)
if let Some(prev_session) = peer.previous_session_mut() {
match prev_session.decrypt_with_replay_check_and_aad(
ciphertext,
header.counter,
&header.header_bytes,
) {
Ok(p) => {
peer.reset_decrypt_failures();
p
}
Err(_) => {
self.log_decrypt_failure(&node_addr, &header, &e);
self.handle_decrypt_failure(&node_addr);
return;
}
}
} else {
self.log_decrypt_failure(&node_addr, &header, &e);
self.handle_decrypt_failure(&node_addr);
return;
}
}
}
};
// === PACKET IS AUTHENTIC ===
// Strip inner header (4-byte timestamp + msg_type)
let (timestamp, link_message) = match strip_inner_header(&plaintext) {
Some(parts) => parts,
None => {
debug!(
peer = %self.peer_display_name(&node_addr),
len = plaintext.len(),
"Decrypted payload too short for inner header"
);
return;
}
};
// MMP per-frame processing and statistics
let now_ms = crate::time::mono_ms();
let ce_flag = header.flags & FLAG_CE != 0;
let sp_flag = header.flags & FLAG_SP != 0;
if let Some(peer) = self.peers.get_mut(&node_addr) {
if let Some(mmp) = peer.mmp_mut() {
mmp.receiver.record_recv(
header.counter,
timestamp,
packet.data.len(),
ce_flag,
now_ms,
);
let _spin_rtt = mmp.spin_bit.rx_observe(sp_flag, header.counter, now_ms);
}
peer.set_current_addr(packet.transport_id, packet.remote_addr.clone());
peer.link_stats_mut()
.record_recv(packet.data.len(), packet.timestamp_ms);
peer.touch(packet.timestamp_ms);
}
// Dispatch to link message handler
self.dispatch_link_message(&node_addr, link_message, ce_flag)
.await;
}
/// Log a decryption failure with replay suppression.
fn log_decrypt_failure(
&mut self,
node_addr: &crate::NodeAddr,
header: &EncryptedHeader,
error: &NoiseError,
) {
if matches!(error, NoiseError::ReplayDetected(_)) {
if let Some(peer) = self.peers.get_mut(node_addr) {
let count = peer.increment_replay_suppressed();
if count <= 3 {
debug!(
peer = %self.peer_display_name(node_addr),
counter = header.counter,
error = %error,
"Decryption failed"
);
} else if count == 4 {
debug!(
peer = %self.peer_display_name(node_addr),
"Suppressing further replay detection messages"
);
}
} else {
debug!(
peer = %self.peer_display_name(node_addr),
counter = header.counter,
error = %error,
"Decryption failed"
);
}
} else {
debug!(
peer = %self.peer_display_name(node_addr),
counter = header.counter,
error = %error,
"Decryption failed"
);
}
}
/// Canonical post-FMP-decrypt side-effect site. Used by both the
/// inline rx_loop decrypt path and the decrypt-worker bounce path
/// so the per-peer bookkeeping (stats, MMP, spin-bit RTT, ECN
/// propagation, address-rotation handling, link-message dispatch)
/// happens in exactly one place.
#[allow(clippy::too_many_arguments)]
pub(in crate::node) async fn process_authentic_fmp_plaintext(
&mut self,
node_addr: &crate::NodeAddr,
transport_id: crate::transport::TransportId,
remote_addr: &crate::transport::TransportAddr,
packet_timestamp_ms: u64,
packet_len: usize,
fmp_counter: u64,
ce_flag: bool,
sp_flag: bool,
fmp_plaintext: &[u8],
) {
const INNER_TIMESTAMP_LEN: usize = 4;
let inner_ts = if fmp_plaintext.len() >= INNER_TIMESTAMP_LEN {
u32::from_le_bytes([
fmp_plaintext[0],
fmp_plaintext[1],
fmp_plaintext[2],
fmp_plaintext[3],
])
} else {
return;
};
let now_ms = crate::time::mono_ms();
let mut address_changed = false;
if let Some(peer) = self.peers.get_mut(node_addr) {
peer.reset_decrypt_failures();
address_changed = peer.set_current_addr(transport_id, remote_addr.clone());
peer.link_stats_mut()
.record_recv(packet_len, packet_timestamp_ms);
peer.touch(packet_timestamp_ms);
if let Some(mmp) = peer.mmp_mut() {
mmp.receiver
.record_recv(fmp_counter, inner_ts, packet_len, ce_flag, now_ms);
let _spin_rtt = mmp.spin_bit.rx_observe(sp_flag, fmp_counter, now_ms);
}
}
// Address rotation invalidates the per-peer connect()-ed UDP
// socket. Drop the connected socket + drain so the wildcard
// listen socket takes over until the new 5-tuple settles.
#[cfg(any(target_os = "linux", target_os = "macos"))]
if address_changed {
self.clear_connected_udp_for_peer(node_addr);
}
#[cfg(not(any(target_os = "linux", target_os = "macos")))]
{
let _ = address_changed;
}
let link_message = &fmp_plaintext[INNER_TIMESTAMP_LEN..];
self.dispatch_link_message(node_addr, link_message, ce_flag)
.await;
}
/// Process a decrypt-worker bounce (FMP plaintext only — the
/// worker has already done the AEAD + replay check).
#[cfg(unix)]
pub(in crate::node) async fn process_decrypt_fallback(
&mut self,
fallback: crate::node::decrypt_worker::DecryptFallback,
) {
let ce_flag = fallback.fmp_flags & FLAG_CE != 0;
let sp_flag = fallback.fmp_flags & FLAG_SP != 0;
let plaintext = &fallback.packet_data[fallback.fmp_plaintext_offset
..fallback.fmp_plaintext_offset + fallback.fmp_plaintext_len];
self.process_authentic_fmp_plaintext(
&fallback.source_node_addr,
fallback.transport_id,
&fallback.remote_addr,
fallback.timestamp_ms,
fallback.packet_len,
fallback.fmp_counter,
ce_flag,
sp_flag,
plaintext,
)
.await;
}
/// Process a decrypt-worker failure event.
#[cfg(unix)]
pub(in crate::node) async fn process_decrypt_failure_report(
&mut self,
report: crate::node::decrypt_worker::DecryptFailureReport,
) {
debug!(
peer = %self.peer_display_name(&report.source_node_addr),
counter = report.fmp_counter,
replay_highest = report.fmp_replay_highest,
"Worker FMP AEAD decryption failed"
);
self.handle_decrypt_failure(&report.source_node_addr);
}
/// Dispatch a decrypt-worker event (plaintext bounce or failure
/// report) to the appropriate handler.
#[cfg(unix)]
pub(in crate::node) async fn process_decrypt_worker_event(
&mut self,
event: crate::node::decrypt_worker::DecryptWorkerEvent,
) {
match event {
crate::node::decrypt_worker::DecryptWorkerEvent::Plaintext(fallback) => {
self.process_decrypt_fallback(fallback).await;
}
crate::node::decrypt_worker::DecryptWorkerEvent::DecryptFailure(report) => {
self.process_decrypt_failure_report(report).await;
}
}
}
/// Hand a session's FMP recv cipher + replay window off to a shard
/// of the decrypt worker pool. Idempotent on rekey: re-registering
/// the same cache_key overwrites the worker's entry. Gates the
/// `decrypt_registered_sessions` insert on actual worker acceptance
/// so a `TrySendError::Full` on the per-worker channel doesn't
/// black-hole the session.
#[cfg(unix)]
pub(in crate::node) fn register_decrypt_worker_session(&mut self, node_addr: &crate::NodeAddr) {
let Some(workers) = self.decrypt_workers.as_ref().cloned() else {
return;
};
let (cache_key, state) = {
let Some(peer) = self.peers.get(node_addr) else {
return;
};
let Some(transport_id) = peer.transport_id() else {
return;
};
let Some(our_index) = peer.our_index() else {
return;
};
let cache_key = (transport_id, our_index.as_u32());
let Some(state) = self.build_owned_session_state(node_addr) else {
return;
};
(cache_key, state)
};
if workers.register_session(cache_key, state) {
self.decrypt_registered_sessions.insert(cache_key);
}
}
/// Drop a session from the decrypt worker pool. Mirror of
/// `register_decrypt_worker_session`. Idempotent: safe to call on a
/// `cache_key` that isn't registered, and safe to call when the
/// worker pool is disabled (`FIPS_DECRYPT_WORKERS=0`).
///
/// Called from two sites that already iterate `peers_by_index`:
/// the rekey drain-completion block (after the drain window has
/// expired, the old `our_index` is unreachable to any in-flight
/// OLD-K packet) and `remove_active_peer` (terminal peer cleanup).
/// Without these callers, the per-worker `sessions` HashMap and
/// the Node's `decrypt_registered_sessions` set would grow
/// monotonically per rekey on long-lived peers.
#[cfg(unix)]
pub(in crate::node) fn unregister_decrypt_worker_session(
&mut self,
cache_key: (crate::transport::TransportId, u32),
) {
if let Some(workers) = self.decrypt_workers.as_ref() {
workers.unregister_session(cache_key);
}
self.decrypt_registered_sessions.remove(&cache_key);
}
/// Snapshot the per-peer FMP recv cipher + replay window for the
/// decrypt worker. Returns `None` if the peer / session isn't
/// ready. After hand-off the worker is the sole FMP replay-window
/// authority for this session.
#[cfg(unix)]
fn build_owned_session_state(
&self,
node_addr: &crate::NodeAddr,
) -> Option<crate::node::decrypt_worker::OwnedSessionState> {
let peer = self.peers.get(node_addr)?;
let fmp_session = peer.noise_session()?;
let fmp_cipher = fmp_session.recv_cipher_clone()?;
let fmp_replay = fmp_session.recv_replay_snapshot_owned();
Some(crate::node::decrypt_worker::OwnedSessionState {
fmp_cipher,
fmp_replay,
source_npub: None,
})
}
/// Increment decrypt failure counter and force-remove peer if threshold exceeded.
pub(in crate::node) fn handle_decrypt_failure(&mut self, node_addr: &crate::NodeAddr) {
if let Some(peer) = self.peers.get_mut(node_addr) {
let count = peer.increment_decrypt_failures();
if count >= DECRYPT_FAILURE_THRESHOLD {
warn!(
peer = %self.peer_display_name(node_addr),
consecutive_failures = count,
"Excessive decryption failures, removing peer"
);
let addr = *node_addr;
self.remove_active_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(addr, now_ms);
}
}
}
}
src/node/handlers/forwarding.rs
-488
View File
@@ -1,488 +0,0 @@
//! SessionDatagram forwarding handler.
//!
//! Handles incoming SessionDatagram (0x00) link messages: decodes the
//! envelope, enforces hop limits, performs coordinate cache warming from
//! plaintext session-layer headers, routes to the next hop or delivers
//! locally, and generates error signals on routing failure.
use crate::NodeAddr;
use crate::node::reject::ForwardingReject;
use crate::node::{Node, NodeError, NodeRoutingView};
use crate::proto::fsp::wire::{
FSP_COMMON_PREFIX_SIZE, FSP_HEADER_SIZE, FSP_PHASE_ESTABLISHED, FSP_PHASE_MSG1, FSP_PHASE_MSG2,
FspCommonPrefix, parse_encrypted_coords,
};
use crate::proto::fsp::{SessionAck, SessionSetup};
use crate::proto::link::{SessionDatagram, SessionDatagramRef};
use crate::proto::routing::{DropReason, NextHop, RouteAction, RouteOutcome};
use std::time::{Duration, Instant};
use tracing::{debug, warn};
impl Node {
/// Handle an incoming SessionDatagram from a peer.
///
/// Called by `dispatch_link_message` for msg_type 0x00. The payload
/// has already had its msg_type byte stripped by dispatch.
pub(in crate::node) async fn handle_session_datagram(
&mut self,
_from: &NodeAddr,
payload: &[u8],
incoming_ce: bool,
) {
self.metrics().forwarding.record_received(payload.len());
let datagram_ref = match SessionDatagramRef::decode(payload) {
Ok(dg) => dg,
Err(e) => {
self.metrics()
.forwarding
.record_reject_bytes(ForwardingReject::DecodeError, payload.len());
debug!(error = %e, "Malformed SessionDatagram");
return;
}
};
let my_addr = *self.node_addr();
// Coordinate cache warming from plaintext session-layer headers. Gated
// on a non-exhausted TTL so a datagram the core will drop as
// TTL-exhausted does not warm the cache, matching the pre-refactor
// ordering (warming ran only after the TTL early-return).
if datagram_ref.ttl != 0 {
self.try_warm_coord_cache_ref(&datagram_ref);
}
// Pre-resolve the next hop only for genuine transit packets (TTL > 0
// and not locally destined) so `find_next_hop`'s coord-cache LRU-touch
// side effect keeps the same scope it had inline. Warming above has
// already run, so the resolution observes freshly cached coords.
let next_hop = if datagram_ref.ttl != 0 && datagram_ref.dest_addr != my_addr {
self.resolve_next_hop(&datagram_ref.dest_addr)
} else {
None
};
// Read local congestion once and reuse it for both the CE decision
// (via the view) and the congestion metric/log below, keeping
// `detect_congestion` the single source of truth.
let congested = next_hop
.as_ref()
.map(|nh| self.detect_congestion(&nh.addr))
.unwrap_or(false);
// Borrow the routing tables disjointly from `&mut self.routing` for
// the pure decision, then release both before driving the outcome.
let outcome = {
let view = NodeRoutingView {
coord_cache: &self.coord_cache,
peers: &self.peers,
tree_state: &self.tree_state,
congested,
};
self.routing
.route(&datagram_ref, &my_addr, incoming_ce, next_hop, &view)
};
match outcome {
RouteOutcome::Drop {
reason: DropReason::TtlExhausted,
} => {
self.metrics()
.forwarding
.record_reject_bytes(ForwardingReject::TtlExhausted, payload.len());
debug!(
src = %datagram_ref.src_addr,
dest = %datagram_ref.dest_addr,
"SessionDatagram TTL exhausted, dropping"
);
}
RouteOutcome::DeliverLocal => {
// Local delivery: dispatch to session layer handlers without
// materializing an owned SessionDatagram payload Vec.
self.metrics().forwarding.record_delivered(payload.len());
self.handle_session_payload(
&datagram_ref.src_addr,
datagram_ref.payload,
datagram_ref.path_mtu,
incoming_ce,
)
.await;
}
RouteOutcome::NoRoute => {
self.metrics()
.forwarding
.record_reject_bytes(ForwardingReject::NoRoute, payload.len());
let original = datagram_ref.into_owned();
debug!(
src = %self.peer_display_name(&original.src_addr),
dest = %self.peer_display_name(&original.dest_addr),
bytes = payload.len(),
"Dropping transit SessionDatagram: no route to destination"
);
self.send_routing_error(&original).await;
}
RouteOutcome::Forward {
next_hop,
bytes,
outgoing_ce,
} => {
let dest = datagram_ref.dest_addr;
// ECN CE relay: congestion was detected locally above; emit the
// metric and rate-limited log at the transit chokepoint.
if congested {
self.metrics().congestion.congestion_detected.inc();
let now = Instant::now();
let should_log = self
.last_congestion_log
.map(|t| now.duration_since(t) >= Duration::from_secs(5))
.unwrap_or(true);
if should_log {
self.last_congestion_log = Some(now);
debug!(next_hop = %next_hop, "Congestion detected, CE flag set on forwarded packet");
}
}
match self
.send_encrypted_link_message_with_ce(&next_hop, &bytes, outgoing_ce)
.await
{
Err(NodeError::MtuExceeded { mtu, .. }) => {
self.metrics()
.forwarding
.record_reject_bytes(ForwardingReject::MtuExceeded, payload.len());
self.send_mtu_exceeded_error(dest, datagram_ref.src_addr, mtu)
.await;
}
Err(e) => {
self.metrics()
.forwarding
.record_reject_bytes(ForwardingReject::SendError, payload.len());
debug!(
next_hop = %next_hop,
dest = %dest,
error = %e,
"Failed to forward SessionDatagram"
);
}
Ok(()) => {
self.metrics().forwarding.record_forwarded(bytes.len());
// Classify this transit forward by route class (partition
// of forwarded_packets). Done here, at the data-plane
// chokepoint, so the error-signal routing callers of
// find_next_hop are excluded.
let class = self.classify_forward(&dest, &next_hop);
self.metrics().forwarding.record_route_class(class);
if outgoing_ce {
self.metrics().congestion.ce_forwarded.inc();
}
}
}
}
}
}
/// Resolve the next hop toward `dest` into its address plus the outgoing
/// link's transport MTU. Returns `None` when there is no route.
///
/// The MTU defaults to `u16::MAX` (a no-op min-fold) when the peer's
/// transport is not resolvable, matching the pre-refactor inline behavior
/// where the MTU `if let` chain simply did not fire.
fn resolve_next_hop(&mut self, dest: &NodeAddr) -> Option<NextHop> {
let addr = *self.find_next_hop(dest)?.node_addr();
let link_mtu = if let Some(peer) = self.peers.get(&addr)
&& let Some(tid) = peer.transport_id()
&& let Some(transport) = self.transports.get(&tid)
{
match peer.current_addr() {
Some(link_addr) => transport.link_mtu(link_addr),
None => transport.mtu(),
}
} else {
u16::MAX
};
Some(NextHop { addr, link_mtu })
}
/// Attempt to warm the coordinate cache from session-layer payload headers.
///
/// Transit routers parse the 4-byte FSP common prefix to identify message
/// type, then extract plaintext coordinate fields from:
/// - SessionSetup (phase 0x1): src_coords + dest_coords
/// - SessionAck (phase 0x2): src_coords
/// - Encrypted with CP flag (phase 0x0): cleartext coords between header and ciphertext
///
/// Decode failures are logged and silently ignored — they don't block
/// forwarding.
fn try_warm_coord_cache_ref(&mut self, datagram: &SessionDatagramRef<'_>) {
let prefix = match FspCommonPrefix::parse(datagram.payload) {
Some(p) => p,
None => return,
};
let inner = &datagram.payload[FSP_COMMON_PREFIX_SIZE..];
let now_ms = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as u64)
.unwrap_or(0);
match prefix.phase {
FSP_PHASE_MSG1 => match SessionSetup::decode(inner) {
Ok(setup) => {
self.coord_cache_mut()
.insert(datagram.src_addr, setup.src_coords, now_ms);
self.coord_cache_mut()
.insert(datagram.dest_addr, setup.dest_coords, now_ms);
debug!(
src = %datagram.src_addr,
dest = %datagram.dest_addr,
"Cached coords from SessionSetup"
);
}
Err(e) => {
debug!(error = %e, "Failed to decode SessionSetup for cache warming");
}
},
FSP_PHASE_MSG2 => match SessionAck::decode(inner) {
Ok(ack) => {
self.coord_cache_mut()
.insert(datagram.src_addr, ack.src_coords, now_ms);
self.coord_cache_mut()
.insert(datagram.dest_addr, ack.dest_coords, now_ms);
debug!(
src = %datagram.src_addr,
dest = %datagram.dest_addr,
"Cached coords from SessionAck"
);
}
Err(e) => {
debug!(error = %e, "Failed to decode SessionAck for cache warming");
}
},
FSP_PHASE_ESTABLISHED if prefix.has_coords() => {
// CP flag set: coords in cleartext between header and ciphertext.
// Parse coords from the cleartext section after the 12-byte header.
// inner starts after the 4-byte prefix, so we need 8 more bytes
// for the counter (header is 12 total = 4 prefix + 8 counter).
let coord_data = &datagram.payload[FSP_HEADER_SIZE..];
match parse_encrypted_coords(coord_data) {
Ok((src_coords, dest_coords, _bytes_consumed)) => {
if let Some(coords) = src_coords {
self.coord_cache_mut()
.insert(datagram.src_addr, coords, now_ms);
}
if let Some(coords) = dest_coords {
self.coord_cache_mut()
.insert(datagram.dest_addr, coords, now_ms);
}
debug!(
src = %datagram.src_addr,
dest = %datagram.dest_addr,
"Cached coords from encrypted message"
);
}
Err(e) => {
debug!(error = %e, "Failed to parse coords for cache warming");
}
}
}
_ => {
// Phase 0x0 without CP, error signals, unknown: no coords to cache
}
}
}
/// Generate and send a routing error signal back to the datagram's source.
///
/// If we have cached coords for the destination, send PathBroken (we know
/// where it is but can't reach it). Otherwise send CoordsRequired (we
/// don't know where it is).
///
/// If we can't route the error back to the source either, drop silently.
/// No cascading errors.
async fn send_routing_error(&mut self, original: &SessionDatagram) {
let my_addr = *self.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);
let default_ttl = self.config().node.session.default_ttl;
// Pure decision: rate-limit gate + PathBroken/CoordsRequired choice +
// error-PDU encode. Borrow the routing tables disjointly from
// `&mut self.routing`, then release them before the reverse-hop lookup.
let action = {
let view = NodeRoutingView {
coord_cache: &self.coord_cache,
peers: &self.peers,
tree_state: &self.tree_state,
congested: false,
};
self.routing.synth_routing_error(
&original.dest_addr,
&original.src_addr,
&my_addr,
&view,
now_ms,
default_ttl,
)
};
let RouteAction::SendError { toward, bytes } = match action {
Some(action) => action,
// Rate limited: drop silently. No cascading errors.
None => return,
};
// Resolve the reverse link hop only now, after the gate passed, so
// `find_next_hop`'s coord-cache touch keeps its pre-refactor scope.
let next_hop_addr = match self.find_next_hop(&toward) {
Some(peer) => *peer.node_addr(),
None => {
debug!(
src = %original.src_addr,
dest = %original.dest_addr,
"Cannot route error signal back to source, dropping"
);
return;
}
};
if let Err(e) = self
.send_encrypted_link_message(&next_hop_addr, &bytes)
.await
{
debug!(
next_hop = %next_hop_addr,
error = %e,
"Failed to send routing error signal"
);
} else {
debug!(
original_dest = %original.dest_addr,
error_dest = %original.src_addr,
"Sent routing error signal"
);
}
}
/// Generate and send an MtuExceeded error signal back to the datagram's source.
///
/// Called when `send_encrypted_link_message()` fails with
/// `NodeError::MtuExceeded` during forwarding. The signal tells the
/// source the bottleneck MTU so it can immediately reduce its path MTU.
///
/// `dest` is the failed datagram's destination (rate-limit key); `toward`
/// is its source, where the signal is routed back.
async fn send_mtu_exceeded_error(
&mut self,
dest: NodeAddr,
toward: NodeAddr,
bottleneck_mtu: u16,
) {
let my_addr = *self.node_addr();
let now_ms = Self::now_ms();
let default_ttl = self.config().node.session.default_ttl;
// Pure decision: rate-limit gate + MtuExceeded PDU + encode.
let action = self.routing.synth_mtu_exceeded(
&dest,
&toward,
&my_addr,
bottleneck_mtu,
now_ms,
default_ttl,
);
let RouteAction::SendError { toward, bytes } = match action {
Some(action) => action,
// Rate limited: drop silently. No cascading errors.
None => return,
};
// Resolve the reverse link hop only now, after the gate passed, so
// `find_next_hop`'s coord-cache touch keeps its pre-refactor scope.
let next_hop_addr = match self.find_next_hop(&toward) {
Some(peer) => *peer.node_addr(),
None => {
debug!(
src = %toward,
dest = %dest,
"Cannot route MtuExceeded signal back to source, dropping"
);
return;
}
};
if let Err(e) = self
.send_encrypted_link_message(&next_hop_addr, &bytes)
.await
{
debug!(
next_hop = %next_hop_addr,
error = %e,
"Failed to send MtuExceeded error signal"
);
} else {
debug!(
original_dest = %dest,
error_dest = %toward,
bottleneck_mtu,
"Sent MtuExceeded error signal"
);
}
}
/// Detect congestion for CE marking on forwarded datagrams.
///
/// Checks two signal sources:
/// 1. Outgoing link MMP metrics (loss rate, ETX) against configured thresholds
/// 2. Local transport congestion (kernel drops on any transport)
///
/// Returns `true` if any signal indicates congestion.
pub(in crate::node) fn detect_congestion(&self, next_hop: &NodeAddr) -> bool {
if !self.config().node.ecn.enabled {
return false;
}
// Outgoing link MMP metrics
if let Some(peer) = self.peers.get(next_hop)
&& let Some(mmp) = peer.mmp()
{
let metrics = &mmp.metrics;
if metrics.loss_rate() >= self.config().node.ecn.loss_threshold
|| metrics.etx >= self.config().node.ecn.etx_threshold
{
return true;
}
}
// Local transport congestion (kernel drops)
self.transport_drops.values().any(|s| s.dropping)
}
/// Sample transport congestion indicators.
///
/// Called from the tick handler (1s interval). For each transport,
/// queries the cumulative kernel drop counter and sets the `dropping`
/// flag if new drops occurred since the previous sample.
pub(in crate::node) fn sample_transport_congestion(&mut self) {
let mut new_drop_events = Vec::new();
for (&tid, transport) in &self.transports {
let congestion = transport.congestion();
let state = self.transport_drops.entry(tid).or_default();
if let Some(current) = congestion.recv_drops {
let new_drops = current > state.prev_drops;
if new_drops && !state.dropping {
new_drop_events.push(tid);
}
state.dropping = new_drops;
state.prev_drops = current;
}
}
for tid in new_drop_events {
self.metrics().congestion.kernel_drop_events.inc();
warn!(
transport_id = tid.as_u32(),
"Kernel recv drops first observed on transport"
);
}
}
}
src/node/handlers/mod.rs
+1 -7
View File
@@ -1,14 +1,8 @@
//! RX event loop and message handlers.
//! Message handlers: per-message-type behavior on `impl Node`.
#[cfg(unix)]
pub(crate) mod connected_udp;
mod dispatch;
mod encrypted;
mod forwarding;
mod handshake;
pub(crate) mod lookup;
mod mmp;
mod rekey;
mod rx_loop;
pub(in crate::node) mod session;
mod timeout;
src/node/handlers/rx_loop.rs
-366
View File
@@ -1,366 +0,0 @@
//! RX event loop and packet dispatch.
use crate::control::{ControlSocket, commands};
use crate::node::{Node, NodeError};
use crate::proto::fmp::wire::{
COMMON_PREFIX_SIZE, CommonPrefix, FMP_VERSION, PHASE_ESTABLISHED, PHASE_MSG1, PHASE_MSG2,
};
use crate::transport::ReceivedPacket;
use std::time::Duration;
use tracing::{debug, info, warn};
/// Inside the packet_rx burst drain, run a fallback drain every
/// N packets so bounced FMP plaintexts can't sit behind a full
/// 256-packet UDP burst. Used on unix; on Windows the decrypt-worker
/// pool isn't spawned so the fallback channel is always empty —
/// hold the constants at module scope anyway so the burst-loop
/// dispatch in `run_rx_loop` doesn't need a `#[cfg]` on every site.
const FALLBACK_INTERLEAVE_EVERY: usize = 32;
/// How many fallback events to drain per interleave step. Bounded so
/// the inner loop can keep making forward progress on packet_rx.
#[cfg(unix)]
const FALLBACK_INTERLEAVE_BUDGET: usize = 32;
impl Node {
/// Run the receive event loop.
///
/// Processes packets from all transports, dispatching based on
/// the phase field in the 4-byte common prefix:
/// - Phase 0x0: Encrypted frame (session data)
/// - Phase 0x1: Handshake message 1 (initiator -> responder)
/// - Phase 0x2: Handshake message 2 (responder -> initiator)
///
/// Also processes outbound IPv6 packets from the TUN reader for session
/// encapsulation and routing through the mesh.
///
/// Also processes DNS-resolved identities for identity cache population.
///
/// Also runs a periodic tick (1s) to clean up stale handshake connections
/// that never received a response. This prevents resource leaks when peers
/// are unreachable.
///
/// This method takes ownership of the packet_rx channel and runs
/// until the channel is closed (typically when stop() is called).
pub async fn run_rx_loop(&mut self) -> Result<(), NodeError> {
let mut packet_rx = self.packet_rx.take().ok_or(NodeError::NotStarted)?;
// Take the TUN outbound receiver, or create a dummy channel that never
// produces messages (when TUN is disabled). Holding the sender prevents
// the channel from closing.
let (mut tun_outbound_rx, _tun_guard) = match self.tun_outbound_rx.take() {
Some(rx) => (rx, None),
None => {
let (tx, rx) = tokio::sync::mpsc::channel(1);
(rx, Some(tx))
}
};
// Take the DNS identity receiver, or create a dummy channel (when DNS
// is disabled). Same pattern as TUN outbound.
let (mut dns_identity_rx, _dns_guard) = match self.dns_identity_rx.take() {
Some(rx) => (rx, None),
None => {
let (tx, rx) = tokio::sync::mpsc::channel(1);
(rx, Some(tx))
}
};
let mut tick =
tokio::time::interval(Duration::from_secs(self.config().node.tick_interval_secs));
// Set up control socket channel
let (control_tx, mut control_rx) =
tokio::sync::mpsc::channel::<crate::control::ControlMessage>(32);
if self.config().node.control.enabled {
let config = self.config().node.control.clone();
let tx = control_tx.clone();
let read_handle = self.control_read_handle();
tokio::spawn(async move {
match ControlSocket::bind(&config) {
Ok(socket) => {
socket.accept_loop(tx, read_handle).await;
}
Err(e) => {
warn!(error = %e, "Failed to bind control socket");
}
}
});
}
// Drop unused sender to avoid keeping channel open if control is disabled
drop(control_tx);
// Decrypt-worker fallback receiver. The worker pushes each
// authenticated FMP plaintext here so rx_loop can finish the
// per-peer side-effects (stats, MMP, ECN, link dispatch).
// Always declared so the `tokio::select!` arm doesn't need
// a `cfg` (which the macro doesn't support); on Windows the
// channel just never sees events.
let (mut decrypt_fallback_rx, _decrypt_fallback_guard) = {
#[cfg(unix)]
{
match self.decrypt_fallback_rx.take() {
Some(rx) => (rx, None),
None => {
let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
(rx, Some(tx))
}
}
}
#[cfg(not(unix))]
{
// On non-unix nothing ever sends, but the macro arm
// still needs an existing rx. Keep the sender alive to
// avoid the channel closing into an Err loop.
let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<()>();
(rx, Some(tx))
}
};
info!("RX event loop started");
// Optional per-stage perf profiler (FIPS_PERF=1). No-op otherwise.
crate::perf_profile::maybe_spawn_reporter();
loop {
tokio::select! {
biased;
// Decrypt-worker fallback drains FIRST. Under sustained
// inbound bursts the packet_rx drain (up to 256 packets)
// can starve fallback work for tens of ms — TCP doesn't
// tolerate that (late ACKs → dup-ACK fast retransmits →
// cwnd collapse). Promoting fallback gives the kernel's
// TCP machinery a fair chance to ACK in time.
Some(event) = decrypt_fallback_rx.recv() => {
#[cfg(unix)]
{
self.process_decrypt_worker_event(event).await;
let mut drained = 0;
while drained < 255 {
match decrypt_fallback_rx.try_recv() {
Ok(ev) => {
self.process_decrypt_worker_event(ev).await;
drained += 1;
}
Err(_) => break,
}
}
}
#[cfg(not(unix))]
let _ = event;
}
packet = packet_rx.recv() => {
match packet {
Some(p) => self.process_packet(p).await,
None => break, // channel closed
}
// Drain remaining ready inbound packets in a tight loop
// before yielding back to select! — every yield is a
// futex hop on tokio's multi-thread scheduler, and at
// line rate the kernel UDP queue typically has several
// datagrams available per wake. Caps at a batch
// boundary so other branches (tick, control) eventually
// get a turn even under sustained load.
//
// **Interleave fallback drain** every N packets so
// bounced FMP plaintexts (heartbeats, post-FMP-
// decrypt forwarding payloads, control frames) don't
// sit in the fallback queue for a full 256-packet
// burst. Even with the priority-first ordering of
// the outer select!, once we're inside this inner
// loop only this interleave can free queued
// fallbacks. On multihop forwarding paths this is
// the difference between back-to-back encrypt-
// worker dispatches happening promptly vs piling up
// behind the rx burst.
let mut drained: usize = 1; // count the packet processed above
while drained < 256 {
if drained.is_multiple_of(FALLBACK_INTERLEAVE_EVERY) {
#[cfg(unix)]
{
let mut fb_drained = 0;
while fb_drained < FALLBACK_INTERLEAVE_BUDGET {
match decrypt_fallback_rx.try_recv() {
Ok(ev) => {
self.process_decrypt_worker_event(ev).await;
fb_drained += 1;
}
Err(_) => break,
}
}
}
}
match packet_rx.try_recv() {
Ok(p) => {
self.process_packet(p).await;
drained += 1;
}
Err(_) => break,
}
}
// Trailing fallback drain so the last bounced
// packets of the burst aren't held up by the
// next select! iteration.
#[cfg(unix)]
{
let mut fb_drained = 0;
while fb_drained < 256 {
match decrypt_fallback_rx.try_recv() {
Ok(ev) => {
self.process_decrypt_worker_event(ev).await;
fb_drained += 1;
}
Err(_) => break,
}
}
}
}
Some(ipv6_packet) = tun_outbound_rx.recv() => {
self.handle_tun_outbound(ipv6_packet).await;
let mut drained = 0;
while drained < 256 {
match tun_outbound_rx.try_recv() {
Ok(p) => {
self.handle_tun_outbound(p).await;
drained += 1;
}
Err(_) => break,
}
}
}
Some(identity) = dns_identity_rx.recv() => {
debug!(
node_addr = %identity.node_addr,
"Registering identity from DNS resolution"
);
self.register_identity(identity.node_addr, identity.pubkey);
}
Some((request, response_tx)) = control_rx.recv() => {
// Only mutating COMMAND requests (`connect` / `disconnect`)
// reach the rx_loop now. Every pure-read `show_*` query is
// served off-loop from the read handle in the control accept
// task (`snapshot_dispatch`), so it never round-trips here —
// the data-plane dispatch path carries no `show_*` arm. A
// `show_*` that somehow arrives (none does) falls through to
// `commands::dispatch`, which returns "unknown command".
let response = commands::dispatch(
self,
&request.command,
request.params.as_ref(),
).await;
let _ = response_tx.send(response);
}
_ = tick.tick() => {
self.check_timeouts();
let now_ms = Self::now_ms();
self.reload_peer_acl().await;
// The host map hot-reloads on the same tick as the ACL. It
// is polled separately from `reload_peer_acl` because the
// ACL's embedded alias reloader and this snapshot are
// distinct resources; the `path_mtu_lookup` cache and the
// `nostr_rendezvous` subsystem are deliberately excluded
// from `Reloadable` since neither reloads from a backing
// file (see `node::reloadable`).
self.reload_host_map().await;
self.poll_pending_connects().await;
self.poll_nostr_rendezvous().await;
self.poll_lan_rendezvous().await;
self.resend_pending_handshakes(now_ms).await;
self.resend_pending_rekeys(now_ms).await;
self.resend_pending_session_handshakes(now_ms).await;
self.resend_pending_session_msg3(now_ms).await;
self.purge_idle_sessions(now_ms);
self.process_pending_retries(now_ms).await;
self.check_tree_state().await;
self.check_bloom_state().await;
self.compute_mesh_size();
self.record_stats_history();
self.check_mmp_reports().await;
self.check_session_mmp_reports().await;
self.check_link_heartbeats().await;
self.check_rekey().await;
self.check_session_rekey().await;
self.check_pending_lookups(now_ms).await;
self.poll_transport_discovery().await;
self.sample_transport_congestion();
#[cfg(any(target_os = "linux", target_os = "macos"))]
self.activate_connected_udp_sessions().await;
}
}
}
info!("RX event loop stopped (channel closed)");
Ok(())
}
/// Process a single received packet.
///
/// Dispatches based on the phase field in the 4-byte common prefix.
async fn process_packet(&mut self, packet: ReceivedPacket) {
if packet.data.len() < COMMON_PREFIX_SIZE {
return; // Drop packets too short for common prefix
}
let prefix = match CommonPrefix::parse(&packet.data) {
Some(p) => p,
None => return, // Malformed prefix
};
if prefix.version != FMP_VERSION {
debug!(
version = prefix.version,
transport_id = %packet.transport_id,
"Unknown FMP version, dropping"
);
// If the packet arrived on an adopted Nostr-NAT bootstrap
// transport, the originating peer is necessarily on a
// different FMP-protocol version than us — the discovery
// sweep would otherwise re-traverse them every cycle even
// though no msg1/msg2 exchange can ever succeed. Bump the
// discovery-layer cooldown to the long protocol-mismatch
// window and emit a single WARN per fresh observation.
if self
.nostr_rendezvous
.is_bootstrap_transport(&packet.transport_id)
&& let Some(npub) = self
.nostr_rendezvous
.bootstrap_transport_npub(&packet.transport_id)
.cloned()
&& let Some(handle) = self.nostr_rendezvous_handle()
{
let now_ms = Self::now_ms();
let cooldown_secs = handle.protocol_mismatch_cooldown_secs();
if handle.record_protocol_mismatch(&npub, now_ms) {
warn!(
peer_npub = %npub,
transport_id = %packet.transport_id,
peer_version = prefix.version,
our_version = FMP_VERSION,
cooldown_secs,
"Nostr-discovered peer speaks a different FMP version; suppressing retraversal"
);
}
}
return;
}
match prefix.phase {
PHASE_ESTABLISHED => {
self.handle_encrypted_frame(packet).await;
}
PHASE_MSG1 => {
self.handle_msg1(packet).await;
}
PHASE_MSG2 => {
self.handle_msg2(packet).await;
}
_ => {
debug!(
phase = prefix.phase,
transport_id = %packet.transport_id,
"Unknown FMP phase, dropping"
);
}
}
}
}