mirror of
https://github.com/jmcorgan/fips.git
synced 2026-08-11 00:47:39 +00:00
Moves both AEAD layers (ChaCha20-Poly1305, one round per layer per packet) plus the sendmsg syscall off the rx_loop task onto a per-shard worker pool, adds per-peer connect(2)-ed UDP with SO_REUSEPORT, and uses Linux UDP GSO (sendmsg+UDP_SEGMENT — kernel splits one super-skb into N on-the-wire datagrams in a single TX-stack walk) when packets in a batch are uniform-size. Same kernel primitive WireGuard's in-kernel module and BoringTun use to hit 2.5–3.2 Gbps single-stream. Single TCP stream on a 5-node docker-bridge mesh, 5 x 15 s x P=1: A→D: 1379 → 2708 Mbps (1.96x, RTT +0.12 ms) A→E: 1394 → 2663 Mbps (1.91x, RTT +0.11 ms) E→A: 1406 → 2624 Mbps (1.87x, RTT +0.19 ms) Static-peer pairs only — every CoV under 3%, 0 outliers, 0% ICMP loss. The ~+100 µs RTT is the worker queue handoff cost; AEAD + sendmmsg now run on a separate core in exchange. What lands: - src/node/encrypt_worker.rs: std::thread + crossbeam_channel workers; hash-by-destination dispatch pins a TCP flow to one worker so wire ordering is preserved; per-worker sendmmsg(2) batching up to 32; Linux uses sendmsg(2)+UDP_SEGMENT when packets in a group are uniform-size. - src/node/decrypt_worker.rs: receive-side mirror. Each shard owns its session's recv cipher + replay window in a thread-local HashMap (no shared RwLock/Mutex). Sessions are handed off at promote_connection and re-registered on K-bit flip / rekey cutover. - src/node/handlers/session.rs try_send_session_data_pipelined: FSP+FMP both seal in-place in the worker on one wire-buffer alloc; no intermediate inner_plaintext / fsp_payload Vecs. - src/transport/udp/connected_peer.rs + peer_drain.rs: per-peer connect(2)-ed UDP socket with SO_REUSEPORT (set on the listen socket too — without that, EADDRINUSE on activation and every packet falls back to the wildcard path); the worker sends with msg_name=NULL and the kernel uses its cached 5-tuple. Tick- driven activation in handlers/connected_udp.rs, idempotent. - src/transport/udp/mod.rs: mem::replace the recvmmsg backing buffer instead of buf.to_vec() per packet — single pointer swap, no MTU-sized memcpy. - src/protocol/link.rs SessionDatagramRef: zero-copy borrowed view used by handle_session_datagram for the bulk local-delivery path; handle_session_payload takes the borrowed payload directly (no payload[35..].to_vec()). - src/transport/mod.rs TransportAddr::from_socket_addr: collapses the two-alloc from_string(addr.to_string()) pattern to one. - src/node/handlers/rx_loop.rs: decrypt-fallback drain promoted ahead of packet_rx in the select! (TCP ACK starvation fix); interleaved fallback drain every 32 packets inside the rx burst loop. - noise::Session: send_cipher_clone / recv_cipher_clone / recv_replay_snapshot_owned / take_send_counter / accept_replay so off-task workers can hold a cloned cipher + reserved counter while the dispatcher keeps replay/counter sequencing serial. CipherState::cipher_clone returns a refcount-bumped LessSafeKey. AsyncUdpSocket: AsRawFd so workers issue raw sendmmsg / sendmsg without going through the tokio reactor. - Worker pool sizing: both default to num_cpus, overridable via FIPS_ENCRYPT_WORKERS=N / FIPS_DECRYPT_WORKERS=N. Per-peer connected UDP can be disabled via FIPS_CONNECTED_UDP=0. - src/perf_profile.rs: optional per-stage timing reporter under FIPS_PERF=1 (or FIPS_PIPELINE_TRACE=1). Off by default; zero overhead when disabled. - All cfg(unix)-gated. Windows continues on the existing tokio- based send/recv. Decrypt worker session lifecycle: - Node::unregister_decrypt_worker_session mirrors the existing register helper. Wired at the two natural sites that already iterate peers_by_index: the rekey drain-completion block in handlers/rekey.rs (drops the worker entry for the old our_index once the drain window has expired and the cache_key is unreachable to any in-flight OLD-K packet), and remove_active_peer in handlers/dispatch.rs (drops the worker entry for each of the four index slots: current, rekey, pending, previous). Only our_index is normally registered; unregister_session is fire- and-forget for missing entries, so calling unconditionally on all four slots is correct and bounds the cleanup without per- slot accounting. 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. Testing: - testing/static/scripts/bench-multirun.sh: multi-run iperf3 + ping bench. N reruns (default 5), median / min / max / CoV % / per-run outlier flag, avg ping RTT, ICMP loss %, TCP retransmit total. Plain client→dest labels + topology header. Pre-bench peer-convergence check (FIPS_BENCH_CONVERGE_SECS, default 15); per-path route verification via stats.bytes_sent deltas — fails fast if traffic exits via a non-static-peer link. - testing/static/docker-compose.yml: passes FIPS_ENCRYPT_WORKERS / FIPS_DECRYPT_WORKERS / FIPS_PERF through to containers for A/B benchmarking without rebuilds. - testing/static/scripts/iperf-test.sh: same plain client→dest labels + topology header (was multihop/direct/N hop, which conflated topology distance with on-wire path). - .config/nextest.toml: synthetic UDP node tests serialized through a max-threads=1 test group. Localhost handshakes drop on shared CI runners under parallel load; one-at-a-time keeps assertions reliable. - src/node/tests/spanning_tree.rs: repair_missing_edge_handshakes — retries up to 5 times for synthetic edges whose msg1 was dropped, with a drain after each edge retry instead of after each attempt's full burst. - src/node/decrypt_worker.rs::tests: two unit tests asserting WorkerMsg::UnregisterSession removes the worker-thread session HashMap entry (handle_msg_unregister_session_removes_entry) and is a no-op for never-seen cache_keys (handle_msg_unregister_session_idempotent_on_unknown_key), which is the safety invariant the unconditional unregister calls at the four index slots in remove_active_peer rely on. - src/node/encrypt_worker.rs::unix_tests pipelined_send_wire_layout_roundtrips_canonical_decoders: mirrors the encoder geometry of try_send_session_data_pipelined (no coords, the common established-session path), runs the worker's real seal + send via flush_direct_batch_sync, and decodes the resulting wire packet using only canonical receive-side decoders (EncryptedHeader::parse, SessionDatagramRef::decode, FSP header parse, noise::open). Any divergence between the hand-rolled encoder offsets (fsp_aad_offset, fsp_plaintext_offset) and the decoders fails at one of the parse / open / decode steps before the inner-plaintext assertion fires. Complements the existing fsp_preseal_runs_before_outer_fmp_seal test which covers the seal-ordering invariant with synthetic headers but does not exercise the wire-layout invariant. CHANGELOG.md [Unreleased] # Changed entry added describing the worker-pool threading model, hash-by-destination dispatch, sendmmsg/UDP_GSO, per-peer connected UDP, the operator-facing env vars, and the bench numbers above. Cherry-picks from mmalmi/master (paths translated from crates/fips-core/src/ to src/): 9b7c723, 0deb5cb, 13f7339, e036c0e, 3740a68, 3792f83, 8510193, 4910b07, e53f545, e4e2896, 5fe4af5, 1d01ada, 8c37008, e12469e, 6eb2860. Co-authored-by: Johnathan Corgan <johnathan@corganlabs.com>
217 lines
8.2 KiB
Rust
217 lines
8.2 KiB
Rust
//! 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::protocol::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);
|
|
|
|
info!(
|
|
peer = %self.peer_display_name(node_addr),
|
|
link_id = %link_id,
|
|
tree_changed = tree_changed,
|
|
"Peer removed and state cleaned up"
|
|
);
|
|
}
|
|
}
|