diff --git a/CHANGELOG.md b/CHANGELOG.md index 60fa077..0a084f8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -149,6 +149,18 @@ with v0.3.x peers. passes 70/70. The constant and surrounding draw/redraw machinery are kept in place pending diagnosis of why XX cutover state cleanup doesn't absorb variable-interval rekeys the way IK does. +- Receive hot path: removed two per-packet copies. New borrowed + `SessionDatagramRef` decoder is used in the forwarding handler so + local delivery and coordinate-cache warming no longer allocate or + copy the session payload; the owned `SessionDatagram` is materialized + only when re-encoding for the next hop. Owned `SessionDatagram:: + decode` is reimplemented as `Ref::decode + into_owned`, so the two + decoders cannot drift. On Linux + macOS the `recvmmsg` / `recvmsg_x` + receive loop now moves each filled slot buffer into `ReceivedPacket` + via `mem::replace` instead of cloning it, and `TransportAddr` is + formatted directly from the `SocketAddr` without an intermediate + `String`. Focused decode bench: ref 1.6 ns/op vs owned 34.7 ns/op + (21.4x). ### Fixed @@ -191,6 +203,20 @@ with v0.3.x peers. after the test's previous one-shot grep gave up, producing a pre-existing flake on next-branch CI. Success-path cost is unchanged — the helper returns as soon as the pattern appears. +- Nostr-discovered NAT-traversal events (`BootstrapEvent::Established` + and `BootstrapEvent::Failed`) for peers that are already connected + or actively handshaking are now short-circuited at the + `poll_nostr_discovery` dispatch sites before any cooldown + bookkeeping or fallback retry scheduling runs. Stale `Failed` events + previously poisoned the per-peer failure-state cooldown of healthy + peers and could trigger redundant retraversal attempts via + `schedule_retry` / `try_peer_addresses`; stale `Established` + handoffs could attempt to adopt a second socket against a live + connection. A defense-in-depth guard was added to + `adopt_established_traversal` so the same invariant holds if a + future caller bypasses the outer dispatch check. As a side benefit, + narrows a cooldown-poisoning vector previously available to an + attacker injecting stale failure events for an active peer. ## [0.3.0] - 2026-05-11 diff --git a/src/cache/coord_cache.rs b/src/cache/coord_cache.rs index 84fda86..a0c914a 100644 --- a/src/cache/coord_cache.rs +++ b/src/cache/coord_cache.rs @@ -205,6 +205,40 @@ impl CoordCache { self.entries.clear(); } + /// Drop entries whose cached destination ancestry contains the given + /// `NodeAddr`. + /// + /// Used at parent-position-change sites: when our own position in the + /// tree changes, destinations downstream of us (whose cached coordinates + /// embed our previous prefix) have stale path information and must be + /// re-learned. Entries whose ancestry does not include `node_addr` are + /// unaffected by the local position change and are retained. + /// + /// Returns the count of entries removed. + pub fn invalidate_via_node(&mut self, node_addr: &NodeAddr) -> usize { + let len_before = self.entries.len(); + self.entries + .retain(|_, entry| !entry.coords().contains(node_addr)); + len_before - self.entries.len() + } + + /// Drop entries whose cached destination `root_id` differs from + /// `current_root`. + /// + /// Used at root-change sites (become_root, root handover via + /// TreeAnnounce). `find_next_hop` returns `None` for any destination + /// whose root does not match the local root, so entries from a stale + /// root cannot route and would otherwise occupy cache slots until + /// TTL expiry. + /// + /// Returns the count of entries removed. + pub fn invalidate_other_roots(&mut self, current_root: &NodeAddr) -> usize { + let len_before = self.entries.len(); + self.entries + .retain(|_, entry| entry.coords().root_id() == current_root); + len_before - self.entries.len() + } + /// Evict one entry (expired first, then LRU). fn evict_one(&mut self, current_time_ms: u64) { // First try to evict an expired entry @@ -507,4 +541,101 @@ mod tests { assert_eq!(stats.expired, 0); assert_eq!(stats.avg_age_ms, 0); } + + // ===== Surgical invalidation tests ===== + + #[test] + fn test_invalidate_via_node_at_self_depth() { + // Entry whose own NodeAddr (depth 0) is the invalidation target. + let mut cache = CoordCache::new(100, 1000); + let target = make_node_addr(1); + + cache.insert(target, make_coords(&[1, 0]), 0); + assert_eq!(cache.len(), 1); + + let removed = cache.invalidate_via_node(&target); + assert_eq!(removed, 1); + assert_eq!(cache.len(), 0); + } + + #[test] + fn test_invalidate_via_node_interior() { + // Entry whose ancestry contains the target in the interior of the path. + let mut cache = CoordCache::new(100, 1000); + let dest = make_node_addr(5); + // Path: 5 -> 3 -> 1 -> 0 (root). Target 3 appears at depth 1. + cache.insert(dest, make_coords(&[5, 3, 1, 0]), 0); + + let removed = cache.invalidate_via_node(&make_node_addr(3)); + assert_eq!(removed, 1); + assert_eq!(cache.len(), 0); + } + + #[test] + fn test_invalidate_via_node_absent() { + // Entry whose ancestry does NOT contain the target must be retained. + let mut cache = CoordCache::new(100, 1000); + let dest = make_node_addr(5); + cache.insert(dest, make_coords(&[5, 3, 1, 0]), 0); + + let removed = cache.invalidate_via_node(&make_node_addr(99)); + assert_eq!(removed, 0); + assert_eq!(cache.len(), 1); + assert!(cache.contains(&dest, 0)); + } + + #[test] + fn test_invalidate_via_node_empty_cache() { + let mut cache = CoordCache::new(100, 1000); + let removed = cache.invalidate_via_node(&make_node_addr(1)); + assert_eq!(removed, 0); + assert_eq!(cache.len(), 0); + } + + #[test] + fn test_invalidate_other_roots_current_root_kept() { + let mut cache = CoordCache::new(100, 1000); + // Entries rooted at addr(0) + cache.insert(make_node_addr(1), make_coords(&[1, 0]), 0); + cache.insert(make_node_addr(2), make_coords(&[2, 0]), 0); + + let removed = cache.invalidate_other_roots(&make_node_addr(0)); + assert_eq!(removed, 0); + assert_eq!(cache.len(), 2); + } + + #[test] + fn test_invalidate_other_roots_different_root_dropped() { + let mut cache = CoordCache::new(100, 1000); + // Three entries rooted at addr(0), one rooted at addr(9) + cache.insert(make_node_addr(1), make_coords(&[1, 0]), 0); + cache.insert(make_node_addr(2), make_coords(&[2, 0]), 0); + cache.insert(make_node_addr(3), make_coords(&[3, 0]), 0); + cache.insert(make_node_addr(4), make_coords(&[4, 9]), 0); + + let removed = cache.invalidate_other_roots(&make_node_addr(0)); + assert_eq!(removed, 1); + assert_eq!(cache.len(), 3); + assert!(!cache.contains(&make_node_addr(4), 0)); + assert!(cache.contains(&make_node_addr(1), 0)); + } + + #[test] + fn test_invalidate_other_roots_all_match() { + let mut cache = CoordCache::new(100, 1000); + cache.insert(make_node_addr(1), make_coords(&[1, 0]), 0); + cache.insert(make_node_addr(2), make_coords(&[2, 0]), 0); + + let removed = cache.invalidate_other_roots(&make_node_addr(0)); + assert_eq!(removed, 0); + assert_eq!(cache.len(), 2); + } + + #[test] + fn test_invalidate_other_roots_empty_cache() { + let mut cache = CoordCache::new(100, 1000); + let removed = cache.invalidate_other_roots(&make_node_addr(0)); + assert_eq!(removed, 0); + assert_eq!(cache.len(), 0); + } } diff --git a/src/control/listening.rs b/src/control/listening.rs index 6b00ac6..2f271ab 100644 --- a/src/control/listening.rs +++ b/src/control/listening.rs @@ -15,7 +15,9 @@ //! See `docs/design/fips-security.md` for the operator-side narrative //! that motivates the panel. -use std::net::{IpAddr, Ipv6Addr}; +#[cfg(target_os = "linux")] +use std::net::IpAddr; +use std::net::Ipv6Addr; /// Transport protocol of a listening socket. #[derive(Debug, Clone, Copy, PartialEq, Eq)] diff --git a/src/discovery/nostr/runtime.rs b/src/discovery/nostr/runtime.rs index eea46f5..26f3b4b 100644 --- a/src/discovery/nostr/runtime.rs +++ b/src/discovery/nostr/runtime.rs @@ -1537,4 +1537,10 @@ impl NostrDiscovery { let mut cache = self.advert_cache.write().await; cache.insert(npub, advert); } + + /// Queue a bootstrap event directly for lifecycle tests without live relays + /// or a running traversal task. + pub(crate) fn push_event_for_test(&self, event: BootstrapEvent) { + let _ = self.event_tx.send(event); + } } diff --git a/src/node/handlers/forwarding.rs b/src/node/handlers/forwarding.rs index 245004b..e96dfc0 100644 --- a/src/node/handlers/forwarding.rs +++ b/src/node/handlers/forwarding.rs @@ -12,7 +12,8 @@ use crate::node::session_wire::{ }; use crate::node::{Node, NodeError}; use crate::protocol::{ - CoordsRequired, MtuExceeded, PathBroken, SessionAck, SessionDatagram, SessionSetup, + CoordsRequired, MtuExceeded, PathBroken, SessionAck, SessionDatagram, SessionDatagramRef, + SessionSetup, }; use std::time::{Duration, Instant}; use tracing::{debug, warn}; @@ -30,7 +31,7 @@ impl Node { ) { self.stats_mut().forwarding.record_received(payload.len()); - let mut datagram = match SessionDatagram::decode(payload) { + let datagram_ref = match SessionDatagramRef::decode(payload) { Ok(dg) => dg, Err(e) => { self.stats_mut() @@ -41,35 +42,41 @@ impl Node { } }; - // TTL enforcement: decrement and drop if exhausted - if !datagram.decrement_ttl() { + // TTL enforcement: decrement for forwarding and drop only if the + // received datagram was already exhausted. + if datagram_ref.ttl == 0 { self.stats_mut() .forwarding .record_ttl_exhausted(payload.len()); debug!( - src = %datagram.src_addr, - dest = %datagram.dest_addr, + src = %datagram_ref.src_addr, + dest = %datagram_ref.dest_addr, "SessionDatagram TTL exhausted, dropping" ); return; } + let forwarded_ttl = datagram_ref.ttl - 1; // Coordinate cache warming from plaintext session-layer headers - self.try_warm_coord_cache(&datagram); + self.try_warm_coord_cache_ref(&datagram_ref); - // Local delivery: dispatch to session layer handlers - if datagram.dest_addr == *self.node_addr() { + // Local delivery: dispatch to session layer handlers without + // materializing an owned SessionDatagram payload Vec. + if datagram_ref.dest_addr == *self.node_addr() { self.stats_mut().forwarding.record_delivered(payload.len()); self.handle_session_payload( - &datagram.src_addr, - &datagram.payload, - datagram.path_mtu, + &datagram_ref.src_addr, + datagram_ref.payload, + datagram_ref.path_mtu, incoming_ce, ) .await; return; } + let mut datagram = datagram_ref.into_owned(); + datagram.ttl = forwarded_ttl; + // Find next hop toward destination let next_hop_addr = match self.find_next_hop(&datagram.dest_addr) { Some(peer) => *peer.node_addr(), @@ -153,8 +160,8 @@ impl Node { /// /// Decode failures are logged and silently ignored — they don't block /// forwarding. - fn try_warm_coord_cache(&mut self, datagram: &SessionDatagram) { - let prefix = match FspCommonPrefix::parse(&datagram.payload) { + fn try_warm_coord_cache_ref(&mut self, datagram: &SessionDatagramRef<'_>) { + let prefix = match FspCommonPrefix::parse(datagram.payload) { Some(p) => p, None => return, }; diff --git a/src/node/handlers/mmp.rs b/src/node/handlers/mmp.rs index ce28da4..d58ab40 100644 --- a/src/node/handlers/mmp.rs +++ b/src/node/handlers/mmp.rs @@ -155,7 +155,9 @@ impl Node { warn!(error = %e, "Failed to sign declaration after first-RTT parent eval"); return; } - self.coord_cache.clear(); + // Surgical invalidation — see CoordCache::invalidate_via_node doc. + self.coord_cache + .invalidate_via_node(self.identity.node_addr()); self.reset_discovery_backoff(); self.stats_mut().tree.parent_switched += 1; self.stats_mut().tree.parent_switches += 1; @@ -180,7 +182,9 @@ impl Node { warn!(error = %e, "Failed to sign self-root declaration after first-RTT"); return; } - self.coord_cache.clear(); + // Surgical invalidation — see CoordCache::invalidate_other_roots doc. + self.coord_cache + .invalidate_other_roots(self.identity.node_addr()); self.reset_discovery_backoff(); self.stats_mut().tree.parent_switched += 1; self.stats_mut().tree.parent_switches += 1; diff --git a/src/node/lifecycle.rs b/src/node/lifecycle.rs index 729cad7..fe7b0ee 100644 --- a/src/node/lifecycle.rs +++ b/src/node/lifecycle.rs @@ -122,12 +122,7 @@ impl Node { } // Check if connection already in progress to this peer - let already_connecting = self.connections.values().any(|conn| { - conn.expected_identity() - .map(|id| id.node_addr() == &peer_node_addr) - .unwrap_or(false) - }); - if already_connecting { + if self.is_connecting_to_peer(&peer_node_addr) { debug!( npub = %peer_config.npub, "Connection already in progress, skipping" @@ -139,6 +134,14 @@ impl Node { .await } + fn is_connecting_to_peer(&self, peer_node_addr: &NodeAddr) -> bool { + self.connections.values().any(|conn| { + conn.expected_identity() + .map(|id| id.node_addr() == peer_node_addr) + .unwrap_or(false) + }) + } + /// Initiate a connection to a peer on a specific transport and address. /// /// For connectionless transports (UDP, Ethernet): allocates a link, starts @@ -451,6 +454,23 @@ impl Node { match event { BootstrapEvent::Established { traversal } => { let peer_npub = traversal.peer_npub.clone(); + if let Ok(peer_identity) = PeerIdentity::from_npub(&peer_npub) { + let peer_addr = *peer_identity.node_addr(); + if self.peers.contains_key(&peer_addr) { + debug!( + peer_npub = %peer_npub, + "Ignoring established NAT traversal for already-connected peer" + ); + continue; + } + if self.is_connecting_to_peer(&peer_addr) { + debug!( + peer_npub = %peer_npub, + "Ignoring established NAT traversal while peer handshake is already in progress" + ); + continue; + } + } match self.adopt_established_traversal(traversal).await { Ok(_) => { info!(peer_npub = %peer_npub, "Adopted NAT traversal socket"); @@ -467,6 +487,28 @@ impl Node { peer_config, reason, } => { + let peer_identity = match PeerIdentity::from_npub(&peer_config.npub) { + Ok(identity) => identity, + Err(_) => continue, + }; + let node_addr = *peer_identity.node_addr(); + if self.peers.contains_key(&node_addr) { + debug!( + npub = %peer_config.npub, + error = %reason, + "Ignoring failed NAT traversal for already-connected peer" + ); + continue; + } + if self.is_connecting_to_peer(&node_addr) { + debug!( + npub = %peer_config.npub, + error = %reason, + "Ignoring failed NAT traversal while peer handshake is already in progress" + ); + continue; + } + let now_ms = Self::now_ms(); let decision = bootstrap.record_traversal_failure(&peer_config.npub, now_ms); if decision.should_warn { @@ -517,11 +559,6 @@ impl Node { }); } - let peer_identity = match PeerIdentity::from_npub(&peer_config.npub) { - Ok(identity) => identity, - Err(_) => continue, - }; - if self .try_peer_addresses(&peer_config, peer_identity, false) .await @@ -530,7 +567,6 @@ impl Node { continue; } - let node_addr = *peer_identity.node_addr(); self.schedule_retry(node_addr, now_ms); if let Some(cooldown_until_ms) = decision.cooldown_until_ms && let Some(state) = self.retry_pending.get_mut(&node_addr) @@ -1737,6 +1773,22 @@ impl Node { peer_identity: PeerIdentity, allow_bootstrap_nat: bool, ) -> Result<(), NodeError> { + let peer_node_addr = *peer_identity.node_addr(); + if self.peers.contains_key(&peer_node_addr) { + debug!( + npub = %peer_config.npub, + "Peer already exists, skipping address attempts" + ); + return Ok(()); + } + if self.is_connecting_to_peer(&peer_node_addr) { + debug!( + npub = %peer_config.npub, + "Connection already in progress, skipping address attempts" + ); + return Ok(()); + } + // Static-first dialing: avoid delaying configured address attempts on // advert fetch/network latency. let static_addresses = self.static_peer_addresses(peer_config); @@ -1880,6 +1932,20 @@ impl Node { } })?; let peer_node_addr = *peer_identity.node_addr(); + if self.peers.contains_key(&peer_node_addr) { + debug!( + peer_npub = %traversal.peer_npub, + "Ignoring NAT traversal handoff for already-connected peer" + ); + return Err(NodeError::PeerAlreadyExists(peer_node_addr)); + } + if self.is_connecting_to_peer(&peer_node_addr) { + debug!( + peer_npub = %traversal.peer_npub, + "Ignoring NAT traversal handoff while peer handshake is already in progress" + ); + return Err(NodeError::PeerAlreadyExists(peer_node_addr)); + } self.peer_aliases .insert(peer_node_addr, peer_identity.short_npub()); diff --git a/src/node/tests/bootstrap.rs b/src/node/tests/bootstrap.rs index a051cea..43b66df 100644 --- a/src/node/tests/bootstrap.rs +++ b/src/node/tests/bootstrap.rs @@ -133,6 +133,51 @@ async fn test_failed_adopted_traversal_cleans_up_transport() { ); } +#[tokio::test] +async fn test_adopted_traversal_skips_already_connected_peer() { + let mut node = make_node(); + let (packet_tx, packet_rx) = packet_channel(64); + node.packet_tx = Some(packet_tx); + node.packet_rx = Some(packet_rx); + node.state = NodeState::Running; + + let transport_id = TransportId::new(1); + let link_id = LinkId::new(1); + let (conn, peer_identity) = make_completed_connection(&mut node, link_id, transport_id, 1_000); + let peer_node_addr = *peer_identity.node_addr(); + node.add_connection(conn).unwrap(); + node.promote_connection(link_id, peer_identity, 2_000) + .unwrap(); + + let link_count = node.link_count(); + let transport_count = node.transport_count(); + + let adopted_socket = std::net::UdpSocket::bind("127.0.0.1:0").unwrap(); + let handoff = EstablishedTraversal::new( + "sess-stale", + peer_identity.npub(), + "127.0.0.1:9".parse().unwrap(), + adopted_socket, + ) + .with_transport_name("nostr-stale"); + + let result = node.adopt_established_traversal(handoff).await; + assert!( + matches!(result, Err(NodeError::PeerAlreadyExists(addr)) if addr == peer_node_addr), + "stale traversal handoff should be ignored once the peer is already active" + ); + assert_eq!( + node.link_count(), + link_count, + "ignored traversal must not create a duplicate link" + ); + assert_eq!( + node.transport_count(), + transport_count, + "ignored traversal must not leak an adopted transport" + ); +} + #[tokio::test] async fn test_third_peer_can_handshake_via_adopted_transport_socket() { let mut node_a = make_node(); // Existing traversal peer (Alice) diff --git a/src/node/tests/unit.rs b/src/node/tests/unit.rs index 3f46e54..d5a8080 100644 --- a/src/node/tests/unit.rs +++ b/src/node/tests/unit.rs @@ -1,7 +1,9 @@ use super::*; +use crate::discovery::nostr::{BootstrapEvent, NostrDiscovery}; use crate::peer::PromotionResult; use crate::transport::udp::UdpTransport; use crate::transport::{TransportHandle, packet_channel}; +use std::sync::Arc; #[test] fn test_node_creation() { @@ -782,6 +784,135 @@ fn test_schedule_retry_skips_connected_peer() { ); } +#[tokio::test] +async fn test_try_peer_addresses_skips_connected_peer() { + let mut node = make_node(); + let transport_id = TransportId::new(1); + let link_id = LinkId::new(1); + let (conn, peer_identity) = make_completed_connection(&mut node, link_id, transport_id, 1000); + let peer_config = crate::config::PeerConfig::new(peer_identity.npub(), "udp", "127.0.0.1:9"); + + node.add_connection(conn).unwrap(); + node.promote_connection(link_id, peer_identity, 2000) + .unwrap(); + let link_count = node.link_count(); + let connection_count = node.connection_count(); + + node.try_peer_addresses(&peer_config, peer_identity, true) + .await + .unwrap(); + + assert_eq!( + node.link_count(), + link_count, + "stale retry/traversal fallback must not create a duplicate link" + ); + assert_eq!( + node.connection_count(), + connection_count, + "stale retry/traversal fallback must not create a duplicate handshake" + ); +} + +#[tokio::test] +async fn test_try_peer_addresses_skips_connecting_peer() { + let mut node = make_node(); + let peer_identity = make_peer_identity(); + let peer_config = crate::config::PeerConfig::new(peer_identity.npub(), "udp", "127.0.0.1:9"); + let pending = PeerConnection::outbound(LinkId::new(1), peer_identity, 1000); + node.add_connection(pending).unwrap(); + + node.try_peer_addresses(&peer_config, peer_identity, true) + .await + .unwrap(); + + assert_eq!( + node.connection_count(), + 1, + "stale retry/traversal fallback must not start a second handshake" + ); + assert_eq!( + node.link_count(), + 0, + "stale retry/traversal fallback must not allocate a link while a handshake is pending" + ); +} + +#[tokio::test] +async fn test_nostr_traversal_failure_skips_connected_peer() { + let mut node = make_node(); + let transport_id = TransportId::new(1); + let link_id = LinkId::new(1); + let (conn, peer_identity) = make_completed_connection(&mut node, link_id, transport_id, 1000); + node.add_connection(conn).unwrap(); + node.promote_connection(link_id, peer_identity, 2000) + .unwrap(); + + let bootstrap = Arc::new(NostrDiscovery::new_for_test()); + bootstrap.push_event_for_test(BootstrapEvent::Failed { + peer_config: crate::config::PeerConfig::new(peer_identity.npub(), "udp", "127.0.0.1:9"), + reason: "stale traversal failure".to_string(), + }); + node.nostr_discovery = Some(bootstrap.clone()); + + node.poll_nostr_discovery().await; + + assert!( + bootstrap.failure_state_snapshot().is_empty(), + "stale failures for connected peers must not affect traversal cooldown" + ); + assert!( + node.retry_pending.is_empty(), + "stale failures for connected peers must not enqueue reconnect attempts" + ); +} + +#[tokio::test] +async fn test_nostr_traversal_established_skips_connected_peer() { + use crate::discovery::EstablishedTraversal; + use std::net::UdpSocket; + + let mut node = make_node(); + let transport_id = TransportId::new(1); + let link_id = LinkId::new(1); + let (conn, peer_identity) = make_completed_connection(&mut node, link_id, transport_id, 1000); + node.add_connection(conn).unwrap(); + node.promote_connection(link_id, peer_identity, 2000) + .unwrap(); + let link_count = node.link_count(); + let connection_count = node.connection_count(); + + let bootstrap = Arc::new(NostrDiscovery::new_for_test()); + let socket = UdpSocket::bind("127.0.0.1:0").expect("bind local UDP socket"); + let remote_addr = "127.0.0.1:9999".parse().expect("parse remote addr"); + bootstrap.push_event_for_test(BootstrapEvent::Established { + traversal: EstablishedTraversal::new( + "test-session", + peer_identity.npub(), + remote_addr, + socket, + ), + }); + node.nostr_discovery = Some(bootstrap.clone()); + + node.poll_nostr_discovery().await; + + assert_eq!( + node.link_count(), + link_count, + "stale established handoff must not allocate a new link" + ); + assert_eq!( + node.connection_count(), + connection_count, + "stale established handoff must not start a new handshake" + ); + assert!( + node.retry_pending.is_empty(), + "stale established handoff must not enqueue a reconnect" + ); +} + #[tokio::test] async fn test_process_pending_retries_drops_expired_entries() { let mut node = make_node(); diff --git a/src/node/tree.rs b/src/node/tree.rs index 4e166b1..e560384 100644 --- a/src/node/tree.rs +++ b/src/node/tree.rs @@ -255,7 +255,9 @@ impl Node { warn!(error = %e, "Failed to sign declaration after parent switch"); return; } - self.coord_cache.clear(); + // Surgical invalidation — see CoordCache::invalidate_via_node doc. + self.coord_cache + .invalidate_via_node(self.identity.node_addr()); self.reset_discovery_backoff(); self.stats_mut().tree.parent_switched += 1; @@ -266,7 +268,7 @@ impl Node { new_seq = new_seq, new_root = %self.tree_state.root(), depth = self.tree_state.my_coords().depth(), - "Parent switched, flushed coord cache, announcing to all peers" + "Parent switched, invalidated downstream coord cache entries, announcing to all peers" ); if flap_dampened { self.stats_mut().tree.flap_dampened += 1; @@ -286,7 +288,9 @@ impl Node { warn!(error = %e, "Failed to sign self-root declaration"); return; } - self.coord_cache.clear(); + // Surgical invalidation — see CoordCache::invalidate_other_roots doc. + self.coord_cache + .invalidate_other_roots(self.identity.node_addr()); self.reset_discovery_backoff(); self.stats_mut().tree.parent_switched += 1; self.stats_mut().tree.parent_switches += 1; @@ -320,7 +324,12 @@ impl Node { warn!(error = %e, "Failed to sign declaration after loop detection"); return; } - self.coord_cache.clear(); + // handle_parent_lost may promote to root OR find new parent; + // cover both invalidation classes. + self.coord_cache + .invalidate_via_node(self.identity.node_addr()); + self.coord_cache + .invalidate_other_roots(self.tree_state.root()); self.reset_discovery_backoff(); self.send_tree_announce_to_all().await; } @@ -355,7 +364,9 @@ impl Node { warn!(error = %e, "Failed to sign declaration after parent update"); return; } - self.coord_cache.clear(); + // Surgical invalidation — see CoordCache::invalidate_via_node doc. + self.coord_cache + .invalidate_via_node(self.identity.node_addr()); self.reset_discovery_backoff(); let new_addrs: Vec = @@ -445,7 +456,9 @@ impl Node { warn!(error = %e, "Failed to sign declaration after periodic parent re-eval"); return; } - self.coord_cache.clear(); + // Surgical invalidation — see CoordCache::invalidate_via_node doc. + self.coord_cache + .invalidate_via_node(self.identity.node_addr()); self.reset_discovery_backoff(); self.stats_mut().tree.parent_switched += 1; @@ -474,7 +487,9 @@ impl Node { warn!(error = %e, "Failed to sign self-root declaration in periodic reeval"); return; } - self.coord_cache.clear(); + // Surgical invalidation — see CoordCache::invalidate_other_roots doc. + self.coord_cache + .invalidate_other_roots(self.identity.node_addr()); self.reset_discovery_backoff(); self.stats_mut().tree.parent_switched += 1; self.stats_mut().tree.parent_switches += 1; diff --git a/src/protocol/link.rs b/src/protocol/link.rs index f1c7bd2..e8044f3 100644 --- a/src/protocol/link.rs +++ b/src/protocol/link.rs @@ -307,6 +307,19 @@ pub struct SessionDatagram { pub payload: Vec, } +/// Borrowed view of a session datagram payload. +/// +/// This avoids allocating and copying the inner payload when the caller only +/// needs to inspect or locally deliver it. +#[derive(Clone, Copy, Debug)] +pub struct SessionDatagramRef<'a> { + pub src_addr: NodeAddr, + pub dest_addr: NodeAddr, + pub ttl: u8, + pub path_mtu: u16, + pub payload: &'a [u8], +} + /// SessionDatagram fixed header size: msg_type(1) + ttl(1) + path_mtu(2) + src_addr(16) + dest_addr(16). pub const SESSION_DATAGRAM_HEADER_SIZE: usize = 36; @@ -363,6 +376,14 @@ impl SessionDatagram { /// Decode from link-layer payload (after msg_type byte has been consumed). pub fn decode(payload: &[u8]) -> Result { + let view = SessionDatagramRef::decode(payload)?; + Ok(view.into_owned()) + } +} + +impl<'a> SessionDatagramRef<'a> { + /// Decode a borrowed view from link-layer payload after the msg_type byte. + pub fn decode(payload: &'a [u8]) -> Result { // ttl(1) + path_mtu(2) + src_addr(16) + dest_addr(16) = 35 if payload.len() < 35 { return Err(ProtocolError::MessageTooShort { @@ -376,16 +397,26 @@ impl SessionDatagram { src_bytes.copy_from_slice(&payload[3..19]); let mut dest_bytes = [0u8; 16]; dest_bytes.copy_from_slice(&payload[19..35]); - let inner_payload = payload[35..].to_vec(); Ok(Self { src_addr: NodeAddr::from_bytes(src_bytes), dest_addr: NodeAddr::from_bytes(dest_bytes), ttl, path_mtu, - payload: inner_payload, + payload: &payload[35..], }) } + + /// Materialize an owned datagram for forwarding/re-encoding paths. + pub fn into_owned(self) -> SessionDatagram { + SessionDatagram { + src_addr: self.src_addr, + dest_addr: self.dest_addr, + ttl: self.ttl, + path_mtu: self.path_mtu, + payload: self.payload.to_vec(), + } + } } // Legacy type alias for compatibility during transition @@ -561,6 +592,75 @@ mod tests { assert_eq!(decoded.payload, payload); } + #[test] + fn test_session_datagram_ref_decode_borrows_payload() { + let src = make_node_addr(0xAA); + let dest = make_node_addr(0xBB); + let payload = vec![0x10, 0x00, 0x05, 0x00, 1, 2, 3, 4, 5]; + let dg = SessionDatagram::new(src, dest, payload.clone()) + .with_ttl(32) + .with_path_mtu(1400); + + let encoded = dg.encode(); + let decoded = SessionDatagramRef::decode(&encoded[1..]).unwrap(); + + assert_eq!(decoded.src_addr, src); + assert_eq!(decoded.dest_addr, dest); + assert_eq!(decoded.ttl, 32); + assert_eq!(decoded.path_mtu, 1400); + assert_eq!(decoded.payload, payload.as_slice()); + assert_eq!( + decoded.payload.as_ptr(), + encoded[SESSION_DATAGRAM_HEADER_SIZE..].as_ptr() + ); + } + + #[test] + #[ignore = "performance benchmark; run explicitly with --ignored --nocapture"] + fn bench_session_datagram_decode_owned_vs_ref() { + use std::hint::black_box; + use std::time::Instant; + + const ITERS: usize = 300_000; + + let src = make_node_addr(0xAA); + let dest = make_node_addr(0xBB); + let payload = vec![0x5A; 1200]; + let datagram = SessionDatagram::new(src, dest, payload) + .with_ttl(32) + .with_path_mtu(1400); + let encoded = datagram.encode(); + let link_payload = &encoded[1..]; + + let ref_start = Instant::now(); + let mut ref_bytes = 0usize; + for _ in 0..ITERS { + let decoded = SessionDatagramRef::decode(black_box(link_payload)).unwrap(); + ref_bytes = ref_bytes.wrapping_add(decoded.payload.len()); + black_box(decoded); + } + let ref_elapsed = ref_start.elapsed(); + + let owned_start = Instant::now(); + let mut owned_bytes = 0usize; + for _ in 0..ITERS { + let decoded = SessionDatagram::decode(black_box(link_payload)).unwrap(); + owned_bytes = owned_bytes.wrapping_add(decoded.payload.len()); + black_box(decoded); + } + let owned_elapsed = owned_start.elapsed(); + + assert_eq!(ref_bytes, owned_bytes); + println!( + "SessionDatagram decode: ref={:.1} ns/op owned={:.1} ns/op speedup={:.2}x iters={} payload_bytes={}", + ref_elapsed.as_secs_f64() * 1_000_000_000.0 / ITERS as f64, + owned_elapsed.as_secs_f64() * 1_000_000_000.0 / ITERS as f64, + owned_elapsed.as_secs_f64() / ref_elapsed.as_secs_f64(), + ITERS, + link_payload.len() - 35 + ); + } + #[test] fn test_session_datagram_empty_payload() { let dg = SessionDatagram::new(make_node_addr(1), make_node_addr(2), Vec::new()); diff --git a/src/protocol/mod.rs b/src/protocol/mod.rs index 60756ee..da5b746 100644 --- a/src/protocol/mod.rs +++ b/src/protocol/mod.rs @@ -34,7 +34,7 @@ pub use error::ProtocolError; pub use filter::{FilterAnnounce, FilterNack}; pub use link::{ Disconnect, DisconnectReason, HandshakeMessageType, LinkMessageType, - SESSION_DATAGRAM_HEADER_SIZE, SessionDatagram, + SESSION_DATAGRAM_HEADER_SIZE, SessionDatagram, SessionDatagramRef, }; pub use negotiation::{ FMP_FEAT_PROFILE_MASK, FMP_FEAT_PROVIDES_RR, FMP_FEAT_PROVIDES_SR, FMP_FEAT_WANTS_RR, diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 41d0f8d..3fa2c14 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -399,6 +399,15 @@ impl TransportAddr { Self(s.as_bytes().to_vec()) } + /// Create a UDP/TCP transport address directly from a socket address. + pub fn from_socket_addr(addr: std::net::SocketAddr) -> Self { + use std::io::Write; + + let mut buf = Vec::with_capacity(56); + write!(&mut buf, "{addr}").expect("Vec::write_fmt is infallible"); + Self(buf) + } + /// Get the raw bytes. pub fn as_bytes(&self) -> &[u8] { &self.0 @@ -1311,6 +1320,15 @@ mod tests { assert_eq!(addr2.as_str(), Some("hello")); } + #[test] + fn test_transport_addr_from_socket_addr() { + let addr = TransportAddr::from_socket_addr("127.0.0.1:2121".parse().unwrap()); + assert_eq!(addr.as_str(), Some("127.0.0.1:2121")); + + let addr = TransportAddr::from_socket_addr("[::1]:2121".parse().unwrap()); + assert_eq!(addr.as_str(), Some("[::1]:2121")); + } + #[test] fn test_link_stats_basic() { let mut stats = LinkStats::new(); diff --git a/src/transport/udp/mod.rs b/src/transport/udp/mod.rs index 2530def..7fddcbd 100644 --- a/src/transport/udp/mod.rs +++ b/src/transport/udp/mod.rs @@ -413,10 +413,10 @@ impl Drop for UdpTransport { /// UDP receive loop - runs as a spawned task. /// -/// On Linux, drains the kernel UDP queue in 32-packet bursts via `recvmmsg` -/// to amortise the per-syscall + per-task-wakeup overhead. macOS / Windows -/// fall through to single-packet `recv_from`. Either way every datagram -/// is forwarded to `packet_tx` in arrival order. +/// Drains the kernel UDP queue in 32-packet bursts via `recvmmsg` (Linux) or +/// `recvmsg_x` (macOS) to amortise the per-syscall + per-task-wakeup overhead. +/// Other unix targets and Windows fall through to single-packet `recv_from`. +/// Either way every datagram is forwarded to `packet_tx` in arrival order. async fn udp_receive_loop( socket: AsyncUdpSocket, transport_id: TransportId, @@ -426,11 +426,13 @@ async fn udp_receive_loop( ) { debug!(transport_id = %transport_id, "UDP receive loop starting"); - #[cfg(target_os = "linux")] + #[cfg(any(target_os = "linux", target_os = "macos"))] { const BATCH: usize = 32; let buf_size = mtu as usize + 100; - // One contiguous backing alloc; slice it for recvmmsg. + // One Vec per recvmmsg / recvmsg_x slot. When a packet lands, move the + // filled buffer directly into ReceivedPacket and install a fresh empty + // buffer for the next syscall, avoiding a per-packet memcpy. let mut backing: Vec> = (0..BATCH).map(|_| vec![0u8; buf_size]).collect(); let mut addrs: [Option; BATCH] = std::array::from_fn(|_| None); let mut lens: [usize; BATCH] = [0; BATCH]; @@ -454,8 +456,7 @@ async fn udp_receive_loop( }; stats.record_recv(len); - let buf = &backing[i][..len]; - if is_punch_packet(buf) { + if is_punch_packet(&backing[i][..len]) { trace!( transport_id = %transport_id, remote_addr = %remote_addr, @@ -465,8 +466,9 @@ async fn udp_receive_loop( continue; } - let data = buf.to_vec(); - let addr = TransportAddr::from_string(&remote_addr.to_string()); + let mut data = std::mem::replace(&mut backing[i], vec![0u8; buf_size]); + data.truncate(len); + let addr = TransportAddr::from_socket_addr(remote_addr); let packet = ReceivedPacket::new(transport_id, addr, data); trace!( @@ -497,7 +499,7 @@ async fn udp_receive_loop( } } - #[cfg(not(target_os = "linux"))] + #[cfg(not(any(target_os = "linux", target_os = "macos")))] { let mut buf = vec![0u8; mtu as usize + 100]; @@ -955,4 +957,49 @@ mod tests { t1.stop_async().await.unwrap(); t2.stop_async().await.unwrap(); } + + /// Burst more than one datagram into the kernel queue before yielding to + /// the receive loop, then assert all are delivered in arrival order. On + /// Linux/macOS this exercises the recvmmsg / recvmsg_x batching path + /// (multiple datagrams reaped per syscall); on other unix targets it + /// degrades to N single-packet recvmsg calls and still must pass. + #[tokio::test] + async fn test_burst_recv_batch() { + let (tx1, _rx1) = packet_channel(100); + let (tx2, mut rx2) = packet_channel(100); + + let mut t1 = UdpTransport::new(TransportId::new(1), None, make_config(0), tx1); + let mut t2 = UdpTransport::new(TransportId::new(2), None, make_config(0), tx2); + + t1.start_async().await.unwrap(); + t2.start_async().await.unwrap(); + + let addr2 = TransportAddr::from_string(&t2.local_addr().unwrap().to_string()); + + // Fire BURST datagrams back-to-back. Each carries its index in the + // first 4 bytes so we can verify per-datagram boundaries (recvmsg_x + // must not coalesce them). + const BURST: u32 = 10; + for i in 0..BURST { + let mut payload = vec![0u8; 32]; + payload[..4].copy_from_slice(&i.to_be_bytes()); + payload[4..].fill(b'x'); + t1.send_async(&addr2, &payload).await.unwrap(); + } + + // Drain. Order must match send order (UDP loopback is in-order, and + // recvmmsg/recvmsg_x preserve it across the batch). + for expected in 0..BURST { + let packet = timeout(Duration::from_secs(1), rx2.recv()) + .await + .expect("timeout draining burst") + .expect("channel closed"); + assert_eq!(packet.data.len(), 32); + let got = u32::from_be_bytes(packet.data[..4].try_into().unwrap()); + assert_eq!(got, expected, "datagram out of order"); + } + + t1.stop_async().await.unwrap(); + t2.stop_async().await.unwrap(); + } } diff --git a/src/transport/udp/socket.rs b/src/transport/udp/socket.rs index cef256f..d8a7c9e 100644 --- a/src/transport/udp/socket.rs +++ b/src/transport/udp/socket.rs @@ -29,13 +29,41 @@ mod platform { use std::os::unix::io::{AsRawFd, RawFd}; use tokio::io::unix::AsyncFd; - /// Maximum number of datagrams a single recvmmsg / sendmmsg syscall - /// will pull from / push to the kernel. Tuned to amortise syscall + - /// per-task-wakeup overhead across a useful burst without blowing - /// the stack (each slot owns an mmsghdr + sockaddr_storage + iovec). - #[cfg(target_os = "linux")] + /// Maximum number of datagrams a single recvmmsg / recvmsg_x / sendmmsg + /// syscall will pull from / push to the kernel. Tuned to amortise syscall + + /// per-task-wakeup overhead across a useful burst without blowing the + /// stack (each slot owns an mmsghdr/msghdr_x + sockaddr_storage + iovec). + #[cfg(any(target_os = "linux", target_os = "macos"))] const BATCH_SIZE: usize = 32; + /// Darwin-private `msghdr_x` for the `recvmsg_x` / `sendmsg_x` syscalls. + /// Layout matches `bsd/sys/socket_private.h` in xnu — same as `msghdr` plus + /// a trailing `msg_datalen` (per-message bytes-received output, in lieu of + /// the `msg_len` field that `mmsghdr` uses on Linux). + #[cfg(target_os = "macos")] + #[repr(C)] + #[allow(non_camel_case_types)] + struct msghdr_x { + msg_name: *mut libc::c_void, + msg_namelen: libc::socklen_t, + msg_iov: *mut libc::iovec, + msg_iovlen: libc::c_int, + msg_control: *mut libc::c_void, + msg_controllen: libc::socklen_t, + msg_flags: libc::c_int, + msg_datalen: usize, + } + + #[cfg(target_os = "macos")] + unsafe extern "C" { + fn recvmsg_x( + s: libc::c_int, + msgp: *const msghdr_x, + cnt: libc::c_uint, + flags: libc::c_int, + ) -> isize; + } + /// Wrapper around a `socket2::Socket` providing sync send/recv with /// `SO_RXQ_OVFL` ancillary data parsing. pub struct UdpRawSocket { @@ -240,10 +268,10 @@ mod platform { /// value is a cumulative counter since socket creation; it is 0 if /// `SO_RXQ_OVFL` is not supported. /// - /// On Linux the production receive path uses `recv_batch` (recvmmsg); - /// this single-packet variant remains for non-Linux unix targets and - /// for the local `tests` module. - #[cfg_attr(target_os = "linux", allow(dead_code))] + /// The production receive path on Linux/macOS uses `recv_batch` + /// (recvmmsg / recvmsg_x); this single-packet variant remains for + /// other unix targets and for the local `tests` module. + #[cfg_attr(any(target_os = "linux", target_os = "macos"), allow(dead_code))] pub fn recv_from(&self, buf: &mut [u8]) -> std::io::Result<(usize, SocketAddr, u32)> { let fd = self.inner.as_raw_fd(); @@ -391,6 +419,56 @@ mod platform { Ok((count, drops)) } + /// Receive up to `BATCH_SIZE` datagrams in a single `recvmsg_x` syscall + /// (macOS). Same `(count, drops)` contract as the Linux `recv_batch`, + /// except `drops` is always 0 — Darwin has no `SO_RXQ_OVFL` equivalent. + /// + /// `recvmsg_x` is a Darwin-private syscall (not in the public SDK) but + /// is shipped in production xnu and is used by quinn-udp for the same + /// per-syscall-amortisation reason as our Linux `recvmmsg` path. + #[cfg(target_os = "macos")] + pub fn recv_batch( + &self, + bufs: &mut [&mut [u8]], + addrs: &mut [Option], + lens: &mut [usize], + ) -> std::io::Result<(usize, u32)> { + let n = bufs.len().min(addrs.len()).min(lens.len()).min(BATCH_SIZE); + if n == 0 { + return Ok((0, 0)); + } + let fd = self.inner.as_raw_fd(); + + let mut iovs: [libc::iovec; BATCH_SIZE] = unsafe { std::mem::zeroed() }; + let mut storages: [libc::sockaddr_storage; BATCH_SIZE] = unsafe { std::mem::zeroed() }; + let mut msgs: [msghdr_x; BATCH_SIZE] = unsafe { std::mem::zeroed() }; + + for i in 0..n { + iovs[i].iov_base = bufs[i].as_mut_ptr() as *mut libc::c_void; + iovs[i].iov_len = bufs[i].len(); + msgs[i].msg_name = &mut storages[i] as *mut _ as *mut libc::c_void; + msgs[i].msg_namelen = + std::mem::size_of::() as libc::socklen_t; + msgs[i].msg_iov = &mut iovs[i]; + msgs[i].msg_iovlen = 1; + // No cmsg consumption — leave msg_control null. (msg_controllen + // is documented as not overwritten by macOS recvmsg_x; zeroed + // init keeps it sane.) + } + + let r = unsafe { recvmsg_x(fd, msgs.as_ptr(), n as libc::c_uint, 0) }; + if r < 0 { + return Err(std::io::Error::last_os_error()); + } + let count = r as usize; + for i in 0..count { + lens[i] = msgs[i].msg_datalen; + addrs[i] = sockaddr_to_socket_addr(&storages[i]).ok(); + } + + Ok((count, 0)) + } + /// Wrap this socket in a tokio `AsyncFd` for async I/O. pub fn into_async(self) -> Result { let async_fd = AsyncFd::new(self) @@ -440,11 +518,11 @@ mod platform { /// Receive a payload, source address, and kernel drop counter. /// - /// Returns `(bytes_read, source_addr, kernel_drops)`. On Linux the - /// production receive path uses `recv_batch`; this single-packet - /// variant remains for non-Linux unix targets and for the local - /// `tests` module. - #[cfg_attr(target_os = "linux", allow(dead_code))] + /// Returns `(bytes_read, source_addr, kernel_drops)`. On Linux/macOS + /// the production receive path uses `recv_batch`; this single-packet + /// variant remains for other unix targets and for the local `tests` + /// module. + #[cfg_attr(any(target_os = "linux", target_os = "macos"), allow(dead_code))] pub async fn recv_from( &self, buf: &mut [u8], @@ -465,9 +543,10 @@ mod platform { } /// Drain up to `BATCH_SIZE` datagrams from the kernel via - /// `recvmmsg` (Linux). Returns `(count, kernel_drops)`; same - /// buffer / addr / len contract as `UdpRawSocket::recv_batch`. - #[cfg(target_os = "linux")] + /// `recvmmsg` (Linux) or `recvmsg_x` (macOS). Returns + /// `(count, kernel_drops)`; same buffer / addr / len contract as + /// `UdpRawSocket::recv_batch`. `kernel_drops` is always 0 on macOS. + #[cfg(any(target_os = "linux", target_os = "macos"))] pub async fn recv_batch( &self, bufs: &mut [&mut [u8]], @@ -739,4 +818,167 @@ mod tests { assert_eq!(&buf[..n], payload); assert_eq!(src, addr1); } + + /// Microbench: compare per-packet `recv_from` (single recvmsg syscall + + /// task wakeup per datagram — the macOS pre-recvmsg_x baseline) vs + /// `recv_batch` (the new recvmsg_x path, up to 32 datagrams per syscall). + /// Both modes run back-to-back in this binary on loopback so the only + /// thing that varies is the receive-syscall strategy. Sender is a tight + /// `socket.send_to()` loop in a separate task; receiver counts datagrams + /// drained over a fixed wall-clock window per mode. + /// + /// Run with: + /// cargo test --release -p fips --lib transport::udp::socket::tests::bench_udp_recv_amortization -- --ignored --nocapture + /// + /// Sender runs on a dedicated *blocking* OS thread (std::net::UdpSocket + /// in default blocking mode) so it always saturates the kernel rx queue + /// regardless of how the tokio receiver schedules. That's the scenario + /// where recvmmsg / recvmsg_x is meant to win: the receiver wakes up to + /// find N packets already buffered, and one syscall reaps the burst. + #[cfg(any(target_os = "linux", target_os = "macos"))] + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + #[ignore] + async fn bench_udp_recv_amortization() { + use std::sync::Arc; + use std::sync::atomic::{AtomicBool, Ordering}; + use std::time::{Duration, Instant}; + + const RECV_BUF: usize = 4 * 1024 * 1024; + const SEND_BUF: usize = 1024 * 1024; + const PAYLOAD_LEN: usize = 100; + const WINDOW: Duration = Duration::from_secs(3); + const WARMUP: Duration = Duration::from_millis(500); + + async fn run_mode( + label: &str, + batched: bool, + sender_threads: usize, + ) -> (u64, u64, Duration) { + let rx_sock = UdpRawSocket::open("127.0.0.1:0".parse().unwrap(), RECV_BUF, SEND_BUF) + .expect("rx bind"); + let rx_addr = rx_sock.local_addr(); + let rx = rx_sock.into_async().expect("rx into_async"); + + // Senders: N dedicated blocking std threads. More threads → deeper + // kernel rx queue → larger amortization opportunity for recv_batch. + // ENOBUFS / EAGAIN just yield and retry; we want saturation, not + // perfect accounting. Sent count is best-effort. + let stop = Arc::new(AtomicBool::new(false)); + let mut sender_handles = Vec::with_capacity(sender_threads); + for _ in 0..sender_threads { + let stop_tx = stop.clone(); + sender_handles.push(std::thread::spawn(move || { + let sock = std::net::UdpSocket::bind("127.0.0.1:0").expect("tx bind"); + sock.connect(rx_addr).expect("tx connect"); + let payload = vec![0xABu8; PAYLOAD_LEN]; + let mut sent: u64 = 0; + while !stop_tx.load(Ordering::Relaxed) { + match sock.send(&payload) { + Ok(_) => sent += 1, + Err(_) => std::thread::yield_now(), + } + } + sent + })); + } + + // Warm-up: let the sender thread reach steady state and the + // kernel rx queue start filling. + tokio::time::sleep(WARMUP).await; + + let start = Instant::now(); + let deadline = start + WINDOW; + let mut recv_count: u64 = 0; + let mut last_drops: u32 = 0; + + if batched { + const BATCH: usize = 32; + let mut backing: Vec> = + (0..BATCH).map(|_| vec![0u8; PAYLOAD_LEN + 64]).collect(); + let mut addrs: [Option; BATCH] = std::array::from_fn(|_| None); + let mut lens: [usize; BATCH] = [0; BATCH]; + let mut batch_sum: u64 = 0; + let mut batch_calls: u64 = 0; + + while Instant::now() < deadline { + let mut bufs: [&mut [u8]; BATCH] = { + let mut iter = backing.iter_mut(); + std::array::from_fn(|_| iter.next().unwrap().as_mut_slice()) + }; + match rx.recv_batch(&mut bufs, &mut addrs, &mut lens).await { + Ok((n, drops)) => { + recv_count += n as u64; + batch_sum += n as u64; + batch_calls += 1; + last_drops = drops; + } + Err(_) => break, + } + } + let avg_batch = if batch_calls > 0 { + batch_sum as f64 / batch_calls as f64 + } else { + 0.0 + }; + eprintln!( + "[{:>10}] avg_batch_per_call={:.2} ({} calls)", + label, avg_batch, batch_calls + ); + } else { + let mut buf = vec![0u8; PAYLOAD_LEN + 64]; + while Instant::now() < deadline { + match rx.recv_from(&mut buf).await { + Ok((_n, _src, drops)) => { + recv_count += 1; + last_drops = drops; + } + Err(_) => break, + } + } + } + let elapsed = start.elapsed(); + + stop.store(true, Ordering::Relaxed); + drop(rx); + let sent: u64 = sender_handles + .into_iter() + .map(|h| h.join().unwrap_or(0)) + .sum(); + + let pps = (recv_count as f64) / elapsed.as_secs_f64(); + let mbps = + (recv_count as f64) * (PAYLOAD_LEN as f64) * 8.0 / 1e6 / elapsed.as_secs_f64(); + eprintln!( + "[{:>10}] recv={:>10} sent={:>10} elapsed={:?} pps={:>12.0} mbps={:>7.1} kdrops={}", + label, recv_count, sent, elapsed, pps, mbps, last_drops + ); + (recv_count, sent, elapsed) + } + + eprintln!("--- udp recv amortization bench ---"); + eprintln!( + "payload={}B window={:?} warmup={:?} runtime=multi_thread(2)", + PAYLOAD_LEN, WINDOW, WARMUP + ); + + // Sweep sender concurrency. Each level shows how the win scales as + // the rx queue gets deeper (more amortization opportunity). + for senders in [1usize, 2, 4, 8] { + eprintln!("\n=== sender_threads = {} ===", senders); + let (b_recv, _, b_el) = run_mode(" recv_from", false, senders).await; + let (x_recv, _, x_el) = run_mode("recv_batch", true, senders).await; + let (x_recv2, _, x_el2) = run_mode("recv_batch", true, senders).await; + let (b_recv2, _, b_el2) = run_mode(" recv_from", false, senders).await; + + let baseline_pps = + (b_recv as f64 / b_el.as_secs_f64() + b_recv2 as f64 / b_el2.as_secs_f64()) / 2.0; + let batched_pps = + (x_recv as f64 / x_el.as_secs_f64() + x_recv2 as f64 / x_el2.as_secs_f64()) / 2.0; + let speedup = batched_pps / baseline_pps; + eprintln!( + "--- senders={}: baseline={:.0} pps batched={:.0} pps speedup={:.2}x ---", + senders, baseline_pps, batched_pps, speedup + ); + } + } } diff --git a/testing/boringtun/.gitignore b/testing/boringtun/.gitignore new file mode 100644 index 0000000..9ab870d --- /dev/null +++ b/testing/boringtun/.gitignore @@ -0,0 +1 @@ +generated/ diff --git a/testing/boringtun/Dockerfile b/testing/boringtun/Dockerfile new file mode 100644 index 0000000..0b44c21 --- /dev/null +++ b/testing/boringtun/Dockerfile @@ -0,0 +1,28 @@ +# Minimal boringtun-cli image for throughput benchmarking against FIPS. +# Userspace WireGuard (boringtun) + wireguard-tools + iperf3 in one +# image. Run two containers on a docker bridge, configure a WG peer +# between them, then iperf3 client→server across the tunnel. + +FROM rust:1-bookworm AS build +# 0.6.0 (crates.io HEAD) is from 2022 and turns in ~1.7 Mbps in +# this harness — clearly broken. Pull the boringtun-cli binary from +# the cloudflare HEAD instead; it gets the post-0.6 perf fixes and +# the multi-threaded recv loop. +RUN git clone --depth 1 https://github.com/cloudflare/boringtun /usr/src/boringtun \ + && cargo install --path /usr/src/boringtun/boringtun-cli + +FROM debian:bookworm-slim +ENV DEBIAN_FRONTEND=noninteractive +RUN apt-get update \ + && apt-get install -y --no-install-recommends \ + iperf3 \ + iproute2 \ + wireguard-tools \ + iputils-ping \ + netcat-openbsd \ + ca-certificates \ + && rm -rf /var/lib/apt/lists/* +COPY --from=build /usr/local/cargo/bin/boringtun-cli /usr/local/bin/boringtun-cli +COPY entrypoint.sh /entrypoint.sh +RUN chmod +x /entrypoint.sh +ENTRYPOINT ["/entrypoint.sh"] diff --git a/testing/boringtun/README.md b/testing/boringtun/README.md new file mode 100644 index 0000000..aeb1be6 --- /dev/null +++ b/testing/boringtun/README.md @@ -0,0 +1,29 @@ +# BoringTun Throughput Baseline + +This harness runs two userspace WireGuard peers with Cloudflare BoringTun and +measures single-stream TCP throughput with `iperf3`. It is intended as a simple +baseline for comparing FIPS tunnel throughput against another userspace tunnel. + +```bash +docker build -t boringtun-test:latest testing/boringtun +testing/boringtun/scripts/generate-keys.sh +docker compose -f testing/boringtun/docker-compose.yml up -d +testing/boringtun/scripts/bench.sh +docker compose -f testing/boringtun/docker-compose.yml down +``` + +The generated WireGuard key material is written under +`testing/boringtun/generated/` and is ignored by git. + +For FIPS-to-FIPS revision comparisons, use the static topology comparison +script: + +```bash +testing/static/scripts/iperf-compare-refs.sh origin/master HEAD mesh +``` + +That script builds each ref into a separate `fips-test:*` image, runs the +same static `iperf3` topology against both images, and prints a bandwidth +summary for each path. Override `DURATION`, `PARALLEL`, `SETTLE_SECONDS`, or +`IPERF_TIMEOUT` in the environment when needed. Set `RUNS=3` or similar to +repeat each ref and print aggregate results. diff --git a/testing/boringtun/docker-compose.yml b/testing/boringtun/docker-compose.yml new file mode 100644 index 0000000..81c1220 --- /dev/null +++ b/testing/boringtun/docker-compose.yml @@ -0,0 +1,46 @@ +# Minimal 2-node boringtun setup for iperf3 throughput comparison. +# Both containers join the same docker bridge; each runs a boringtun +# userspace WireGuard tunnel and they peer with each other over UDP. +# Inner WG IPs: 10.99.0.1/24 (alice) <-> 10.99.0.2/24 (bob). + +networks: + bt-net: + driver: bridge + ipam: + config: + - subnet: 172.99.0.0/24 + +x-boringtun-common: &boringtun-common + image: boringtun-test:latest + cap_add: + - NET_ADMIN + devices: + - /dev/net/tun:/dev/net/tun + sysctls: + - net.ipv4.conf.all.forwarding=1 + restart: "no" + env_file: + - ./generated/peers.env + +services: + alice: + <<: *boringtun-common + container_name: bt-alice + hostname: alice + environment: + - ROLE=alice + - PEER_HOST=bob + networks: + bt-net: + ipv4_address: 172.99.0.10 + + bob: + <<: *boringtun-common + container_name: bt-bob + hostname: bob + environment: + - ROLE=bob + - PEER_HOST=alice + networks: + bt-net: + ipv4_address: 172.99.0.11 diff --git a/testing/boringtun/entrypoint.sh b/testing/boringtun/entrypoint.sh new file mode 100755 index 0000000..a440125 --- /dev/null +++ b/testing/boringtun/entrypoint.sh @@ -0,0 +1,72 @@ +#!/bin/bash +# Bring up a boringtun-userspace WireGuard tunnel and then sleep +# indefinitely so the benchmark scripts can drive it via docker exec. +# +# Required env: +# ROLE "alice" or "bob" +# ALICE_WG_IP WG inner IP for alice (e.g. 10.99.0.1/24) +# BOB_WG_IP WG inner IP for bob (e.g. 10.99.0.2/24) +# ALICE_PUB alice's WG public key +# BOB_PUB bob's WG public key +# ALICE_PRIV alice's WG private key +# BOB_PRIV bob's WG private key +# PEER_HOST the other container's hostname on the docker bridge +# PEER_PORT the other container's WG UDP port (default 51820) +# +# boringtun-cli runs in foreground (--foreground) so wg-quick-style +# configuration is done manually via `ip`, `wg`, and `wg set`. + +set -e + +PORT="${PEER_PORT:-51820}" +case "$ROLE" in + alice) + OUR_IP="$ALICE_WG_IP" + OUR_PRIV="$ALICE_PRIV" + PEER_PUB="$BOB_PUB" + ;; + bob) + OUR_IP="$BOB_WG_IP" + OUR_PRIV="$BOB_PRIV" + PEER_PUB="$ALICE_PUB" + ;; + *) + echo "ROLE must be alice or bob, got '$ROLE'" >&2 + exit 1 + ;; +esac + +echo "[$ROLE] starting boringtun-cli foreground on wg0" +# `--foreground` keeps the userspace driver in the container +# foreground. boringtun-cli sets WG_TUN_NAME_FILE to /tmp/wg0.name +# when --foreground; we just hardcode the device name. +boringtun-cli --foreground --disable-drop-privileges wg0 & +BORINGTUN_PID=$! + +# wait for the tun device to appear +for i in $(seq 1 50); do + if ip link show wg0 >/dev/null 2>&1; then + break + fi + sleep 0.1 +done + +echo "[$ROLE] configuring wg0 with $OUR_IP listening on $PORT" +PRIV_FILE=$(mktemp) +chmod 600 "$PRIV_FILE" +printf '%s' "$OUR_PRIV" >"$PRIV_FILE" +wg set wg0 private-key "$PRIV_FILE" listen-port "$PORT" +rm -f "$PRIV_FILE" + +ip address add "$OUR_IP" dev wg0 +ip link set up dev wg0 + +echo "[$ROLE] adding peer pubkey, endpoint $PEER_HOST:$PORT" +wg set wg0 peer "$PEER_PUB" allowed-ips 10.99.0.0/24 endpoint "$PEER_HOST:$PORT" persistent-keepalive 25 + +echo "[$ROLE] ready" +wg show wg0 + +# Park here so the container stays up; we'll run iperf3 etc via +# docker exec. +wait $BORINGTUN_PID diff --git a/testing/boringtun/scripts/bench.sh b/testing/boringtun/scripts/bench.sh new file mode 100755 index 0000000..f8acbb9 --- /dev/null +++ b/testing/boringtun/scripts/bench.sh @@ -0,0 +1,25 @@ +#!/bin/bash +# End-to-end iperf3 bandwidth test between two boringtun containers. +# Output mirrors testing/static/scripts/iperf-test.sh so the FIPS +# numbers are directly comparable. +set -euo pipefail + +DURATION="${DURATION:-10}" +PARALLEL="${PARALLEL:-1}" + +echo "=== boringtun iperf3 throughput (single TCP stream, ${DURATION}s) ===" + +# Run iperf3 server on alice (background), client on bob. +docker exec -d bt-alice iperf3 -s -1 -B 10.99.0.1 -p 5201 +sleep 1 + +# wait for tun handshake to settle (boringtun + WG keepalive) +sleep 2 + +# Client: bob → alice over WG (10.99.0.1) +OUT=$(docker exec bt-bob iperf3 -c 10.99.0.1 -p 5201 -t "$DURATION" -P "$PARALLEL" -J) + +# Pull SUM bps. +MBPS=$(echo "$OUT" | python3 -c "import json,sys; d=json.load(sys.stdin); print(f\"{d['end']['sum_received']['bits_per_second'] / 1_000_000:.2f}\")") + +echo "boringtun bob -> alice : ${MBPS} Mbits/sec" diff --git a/testing/boringtun/scripts/generate-keys.sh b/testing/boringtun/scripts/generate-keys.sh new file mode 100755 index 0000000..0c0c7f5 --- /dev/null +++ b/testing/boringtun/scripts/generate-keys.sh @@ -0,0 +1,40 @@ +#!/bin/bash +# Generate WG keypairs for alice and bob and write the +# generated/peers.env file the docker-compose.yml reads. +# Idempotent: skips if file already exists. +set -e + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +OUT_DIR="$SCRIPT_DIR/../generated" +ENV_FILE="$OUT_DIR/peers.env" + +if [ -f "$ENV_FILE" ]; then + echo "$ENV_FILE already exists; reusing keys" + exit 0 +fi + +mkdir -p "$OUT_DIR" + +# Use the built boringtun-test image to generate keys — host may +# not have wireguard-tools installed. The `pubkey` step needs `-i` +# so stdin is piped through; `genkey` doesn't need it but matching +# flags keeps the two calls symmetric. +GEN_CMD="docker run --rm --entrypoint wg boringtun-test:latest" +PUB_CMD="docker run --rm -i --entrypoint wg boringtun-test:latest" + +ALICE_PRIV=$($GEN_CMD genkey) +ALICE_PUB=$(printf '%s' "$ALICE_PRIV" | $PUB_CMD pubkey) +BOB_PRIV=$($GEN_CMD genkey) +BOB_PUB=$(printf '%s' "$BOB_PRIV" | $PUB_CMD pubkey) + +cat >"$ENV_FILE" < [mesh|chain] +# +# Environment: +# DURATION=10 iperf3 duration passed through to iperf-test.sh +# PARALLEL=8 iperf3 parallel streams passed through to iperf-test.sh +# SETTLE_SECONDS=3 topology startup delay passed through to iperf-test.sh +# IPERF_TIMEOUT per-path timeout, defaults to DURATION + 30 +# RUNS=1 total measurement runs per ref +set -euo pipefail + +if [ "$#" -lt 2 ] || [ "$#" -gt 3 ]; then + echo "Usage: $0 [mesh|chain]" >&2 + exit 2 +fi + +BASE_REF="$1" +CANDIDATE_REF="$2" +PROFILE="${3:-mesh}" +RUNS="${RUNS:-1}" + +if ! [[ "$RUNS" =~ ^[1-9][0-9]*$ ]]; then + echo "RUNS must be a positive integer" >&2 + exit 2 +fi + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +PROJECT_ROOT="$(cd "$SCRIPT_DIR/../../.." && pwd)" +COMPOSE_FILE="$PROJECT_ROOT/testing/static/docker-compose.yml" +TMP_DIR="$(mktemp -d "${TMPDIR:-/tmp}/fips-iperf-compare.XXXXXX")" +WORKTREES=() +FAILED_RUNS=0 + +cleanup() { + docker compose -f "$COMPOSE_FILE" --profile "$PROFILE" down >/dev/null 2>&1 || true + for wt in "${WORKTREES[@]}"; do + git -C "$PROJECT_ROOT" worktree remove --force "$wt" >/dev/null 2>&1 || true + done + rm -rf "$TMP_DIR" +} +trap cleanup EXIT + +slug_ref() { + echo "$1" \ + | tr '[:upper:]' '[:lower:]' \ + | sed 's/[^a-z0-9_.-]/-/g; s/^-*//; s/-*$//' \ + | cut -c1-48 +} + +image_tag() { + local label="$1" + local ref="$2" + local slug + slug="$(slug_ref "$ref")" + [ -n "$slug" ] || slug="ref" + printf 'fips-test:compare-%s-%s\n' "$label" "$slug" +} + +build_ref_image() { + local label="$1" + local ref="$2" + local tag="$3" + + local wt="$TMP_DIR/$label" + local target_dir="$PROJECT_ROOT/target/iperf-compare-$label-$(slug_ref "$ref")" + + echo "" + echo "=== Building $label: $ref -> $tag ===" + git -C "$PROJECT_ROOT" worktree add --detach "$wt" "$ref" + WORKTREES+=("$wt") + + mkdir -p "$target_dir" + ln -s "$target_dir" "$wt/target" + ( + cd "$wt" + CARGO_TARGET_DIR="$wt/target" ./testing/scripts/build.sh + ) + docker tag fips-test:latest "$tag" +} + +run_profile() { + local label="$1" + local image="$2" + local run="$3" + local log="$TMP_DIR/$label-run-$run.log" + local duration="${DURATION:-10}" + local parallel="${PARALLEL:-8}" + local settle_seconds="${SETTLE_SECONDS:-3}" + local iperf_timeout="${IPERF_TIMEOUT:-$((duration + 30))}" + + echo "" + echo "=== Running $label run $run/$RUNS with $image ===" + FIPS_TEST_IMAGE="$image" docker compose -f "$COMPOSE_FILE" --profile "$PROFILE" up -d --force-recreate + if DURATION="$duration" PARALLEL="$parallel" \ + SETTLE_SECONDS="$settle_seconds" IPERF_TIMEOUT="$iperf_timeout" \ + "$SCRIPT_DIR/iperf-test.sh" "$PROFILE" | tee "$log"; then + : + else + local status="$?" + echo "=== $label exited with status $status ===" | tee -a "$log" + FAILED_RUNS=1 + fi + docker compose -f "$COMPOSE_FILE" --profile "$PROFILE" down +} + +print_summary() { + local label="$1" + local run="$2" + local log="$TMP_DIR/$label-run-$run.log" + + [ -f "$log" ] || return 0 + + awk -v label="$label" -v run="$run" ' + /^=== .* ===$/ && $0 !~ /FIPS iperf3 Bandwidth Test/ && $0 !~ /Results:/ { + test=$0 + sub(/^=== /, "", test) + sub(/ ===$/, "", test) + } + /^Bandwidth:/ { + print label "\t" run "\t" test "\t" $2 " " $3 + } + ' "$log" +} + +print_aggregate() { + local summary="$1" + + awk -F '\t' ' + NR == 1 { next } + + function to_mbps(value, unit) { + if (unit ~ /^Gbits/) return value * 1000 + if (unit ~ /^Mbits/) return value + if (unit ~ /^Kbits/) return value / 1000 + return value + } + + { + split($4, parts, " ") + mbps = to_mbps(parts[1] + 0, parts[2]) + key = $1 SUBSEP $3 + if (!(key in seen)) { + seen[key] = 1 + order[++order_len] = key + refs[key] = $1 + tests[key] = $3 + min[key] = mbps + max[key] = mbps + } + count[key]++ + sum[key] += mbps + if (mbps < min[key]) min[key] = mbps + if (mbps > max[key]) max[key] = mbps + } + + END { + print "ref\ttest\truns\tavg_mbps\tmin_mbps\tmax_mbps" + for (i = 1; i <= order_len; i++) { + key = order[i] + printf "%s\t%s\t%d\t%.0f\t%.0f\t%.0f\n", + refs[key], tests[key], count[key], + sum[key] / count[key], min[key], max[key] + } + } + ' "$summary" +} + +"$SCRIPT_DIR/generate-configs.sh" "$PROFILE" + +BASE_IMAGE="$(image_tag base "$BASE_REF")" +CANDIDATE_IMAGE="$(image_tag candidate "$CANDIDATE_REF")" + +build_ref_image base "$BASE_REF" "$BASE_IMAGE" +build_ref_image candidate "$CANDIDATE_REF" "$CANDIDATE_IMAGE" + +for run in $(seq 1 "$RUNS"); do + run_profile base "$BASE_IMAGE" "$run" + run_profile candidate "$CANDIDATE_IMAGE" "$run" +done + +echo "" +echo "=== Summary ===" +SUMMARY_FILE="$TMP_DIR/summary.tsv" +{ + printf 'ref\trun\ttest\tbandwidth\n' + for run in $(seq 1 "$RUNS"); do + print_summary base "$run" + print_summary candidate "$run" + done +} | tee "$SUMMARY_FILE" + +echo "" +echo "=== Aggregate ===" +print_aggregate "$SUMMARY_FILE" + +exit "$FAILED_RUNS" diff --git a/testing/static/scripts/iperf-test.sh b/testing/static/scripts/iperf-test.sh index 74d1767..450db5e 100755 --- a/testing/static/scripts/iperf-test.sh +++ b/testing/static/scripts/iperf-test.sh @@ -18,8 +18,10 @@ if [ "$2" = "--live" ] || [ "$1" = "--live" ]; then [ "$1" = "--live" ] && PROFILE="mesh" fi -DURATION=10 -PARALLEL=8 +DURATION="${DURATION:-10}" +PARALLEL="${PARALLEL:-8}" +SETTLE_SECONDS="${SETTLE_SECONDS:-3}" +IPERF_TIMEOUT="${IPERF_TIMEOUT:-$((DURATION + 30))}" PASSED=0 FAILED=0 @@ -47,7 +49,7 @@ iperf_test() { if [ "$LIVE_OUTPUT" = true ]; then # Show live output echo "Running iperf3 test (live output):" - if docker exec "fips-$client_node" iperf3 -c "${dest_npub}.fips" -t "$DURATION" -P "$PARALLEL"; then + if docker exec "fips-$client_node" timeout "$IPERF_TIMEOUT" iperf3 -c "${dest_npub}.fips" -t "$DURATION" -P "$PARALLEL"; then PASSED=$((PASSED + 1)) else echo "FAIL" @@ -57,7 +59,7 @@ iperf_test() { # Capture and summarize output echo -n "Running iperf3 test... " local output - if output=$(docker exec "fips-$client_node" iperf3 -c "${dest_npub}.fips" -t "$DURATION" -P "$PARALLEL" 2>&1); then + if output=$(docker exec "fips-$client_node" timeout "$IPERF_TIMEOUT" iperf3 -c "${dest_npub}.fips" -t "$DURATION" -P "$PARALLEL" 2>&1); then # Check if we got valid results if echo "$output" | grep -q "sender"; then # Extract and display results (get SUM line for aggregate bandwidth) @@ -83,8 +85,8 @@ echo "=== FIPS iperf3 Bandwidth Test ($PROFILE topology) ===" echo "" # Wait for nodes to converge -echo "Waiting 3s for mesh convergence..." -sleep 3 +echo "Waiting ${SETTLE_SECONDS}s for mesh convergence..." +sleep "$SETTLE_SECONDS" if [ "$PROFILE" = "mesh" ] || [ "$PROFILE" = "mesh-public" ]; then # Test key paths in mesh topology @@ -120,4 +122,4 @@ fi echo "" echo "=== Results: $PASSED passed, $FAILED failed ===" -[ "$FAILED" -eq 0 ] && exit 0 || exit 1 \ No newline at end of file +[ "$FAILED" -eq 0 ] && exit 0 || exit 1