From 87d1af02698815b05fa73f662396089edabaf05a Mon Sep 17 00:00:00 2001
From: Martti Malmi <sirius@iki.fi>
Date: Fri, 15 May 2026 15:06:54 +0000
Subject: [PATCH 1/5] nostr: ignore stale traversal for active peers

Skip BootstrapEvent::Established and BootstrapEvent::Failed dispatch
in poll_nostr_discovery for peers that are already connected or
actively handshaking. Without these guards, stale traversal events
arriving after a peer connected through a different path would
either attempt to adopt a redundant socket against the live
connection (Established) or poison the per-peer failure-state
cooldown and trigger redundant retraversal via schedule_retry /
try_peer_addresses (Failed).

The four guard sites use a new is_connecting_to_peer helper extracted
from the existing closure inside initiate_peer_connection; the helper
checks for an in-flight outbound handshake state. adopt_established_traversal
gains a defense-in-depth check returning PeerAlreadyExists when called
against an already-promoted peer, so the invariant holds if a future
caller bypasses the outer dispatch guard.

Side benefit: narrows a cooldown-poisoning vector previously available
to an attacker injecting stale failure events for an active peer.

Test coverage for the new behavior:

- test_try_peer_addresses_skips_connected_peer
- test_try_peer_addresses_skips_connecting_peer
- test_nostr_traversal_failure_skips_connected_peer (Failed-arm event
  injection)
- test_nostr_traversal_established_skips_connected_peer (Established-arm
  event injection, mirror of the Failed test)
- test_adopted_traversal_skips_already_connected_peer
  (adopt_established_traversal defense-in-depth)

CHANGELOG entry under [Unreleased] / Fixed.

Closes #87
---
 CHANGELOG.md                   |  14 ++++
 src/discovery/nostr/runtime.rs |   6 ++
 src/node/lifecycle.rs          |  90 +++++++++++++++++++---
 src/node/tests/bootstrap.rs    |  45 +++++++++++
 src/node/tests/unit.rs         | 131 +++++++++++++++++++++++++++++++++
 5 files changed, 274 insertions(+), 12 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 5a6d592..dd53edd 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -77,6 +77,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   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/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/lifecycle.rs b/src/node/lifecycle.rs
index 4e92418..a8ee430 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
@@ -410,6 +413,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");
@@ -426,6 +446,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 {
@@ -476,11 +518,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
@@ -489,7 +526,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)
@@ -1694,6 +1730,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);
@@ -1837,6 +1889,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 04906fa..4e59cda 100644
--- a/src/node/tests/bootstrap.rs
+++ b/src/node/tests/bootstrap.rs
@@ -117,6 +117,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 03f9558..a4a7a2b 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() {
@@ -779,6 +781,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();

From b05c80e5f5f8690ef6167cf577ebe150193971d2 Mon Sep 17 00:00:00 2001
From: Martti Malmi <sirius@iki.fi>
Date: Fri, 15 May 2026 18:03:42 +0000
Subject: [PATCH 2/5] testing: add boringtun throughput benchmark and iperf
 ref-compare harness

New testing/boringtun/ harness runs two Cloudflare BoringTun userspace
WireGuard containers with iperf3 between them, giving a single-hop
userspace tunnel baseline for comparison against FIPS throughput
numbers. Local WG key generation runs through the harness image so the
host needs no wireguard-tools.

New testing/static/scripts/iperf-compare-refs.sh builds two git refs
into separate fips-test:* images via git worktree and runs the same
static iperf topology against both, with RUNS-based repetition and
aggregate avg/min/max reporting.

testing/static/scripts/iperf-test.sh gains DURATION, PARALLEL,
SETTLE_SECONDS, IPERF_TIMEOUT env knobs and a per-path iperf timeout.
testing/static/docker-compose.yml selects the image under test via
FIPS_TEST_IMAGE; testing/scripts/build.sh respects CARGO_TARGET_DIR.

Author benchmark on aarch64 Docker Desktop:
  boringtun bob -> alice : 1000.13 Mbits/sec
---
 testing/boringtun/.gitignore                 |   1 +
 testing/boringtun/Dockerfile                 |  28 +++
 testing/boringtun/README.md                  |  29 +++
 testing/boringtun/docker-compose.yml         |  46 +++++
 testing/boringtun/entrypoint.sh              |  72 +++++++
 testing/boringtun/scripts/bench.sh           |  25 +++
 testing/boringtun/scripts/generate-keys.sh   |  40 ++++
 testing/scripts/build.sh                     |   6 +-
 testing/static/README.md                     |  13 ++
 testing/static/docker-compose.yml            |   2 +-
 testing/static/scripts/iperf-compare-refs.sh | 202 +++++++++++++++++++
 testing/static/scripts/iperf-test.sh         |  16 +-
 12 files changed, 470 insertions(+), 10 deletions(-)
 create mode 100644 testing/boringtun/.gitignore
 create mode 100644 testing/boringtun/Dockerfile
 create mode 100644 testing/boringtun/README.md
 create mode 100644 testing/boringtun/docker-compose.yml
 create mode 100755 testing/boringtun/entrypoint.sh
 create mode 100755 testing/boringtun/scripts/bench.sh
 create mode 100755 testing/boringtun/scripts/generate-keys.sh
 create mode 100755 testing/static/scripts/iperf-compare-refs.sh

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" <<EOF
+ALICE_PRIV=$ALICE_PRIV
+ALICE_PUB=$ALICE_PUB
+BOB_PRIV=$BOB_PRIV
+BOB_PUB=$BOB_PUB
+ALICE_WG_IP=10.99.0.1/24
+BOB_WG_IP=10.99.0.2/24
+PEER_PORT=51820
+EOF
+chmod 600 "$ENV_FILE"
+echo "wrote $ENV_FILE"
diff --git a/testing/scripts/build.sh b/testing/scripts/build.sh
index b72e47a..35781ea 100755
--- a/testing/scripts/build.sh
+++ b/testing/scripts/build.sh
@@ -40,6 +40,8 @@ done
 UNAME_S=$(uname -s)
 CARGO_TARGET="x86_64-unknown-linux-musl"
 
+TARGET_ROOT="${CARGO_TARGET_DIR:-$PROJECT_ROOT/target}"
+
 if [ "$UNAME_S" = "Darwin" ]; then
     echo "Detected macOS host — using cross-compilation for Linux..."
 
@@ -57,12 +59,12 @@ if [ "$UNAME_S" = "Darwin" ]; then
     echo "Building FIPS for Linux (release) using cargo-zigbuild..."
     cargo zigbuild --release --target "$CARGO_TARGET" --manifest-path="$PROJECT_ROOT/Cargo.toml" "${CARGO_BUILD_ARGS[@]}"
 
-    TARGET_DIR="$PROJECT_ROOT/target/$CARGO_TARGET/release"
+    TARGET_DIR="$TARGET_ROOT/$CARGO_TARGET/release"
 else
     echo "Building FIPS (release)..."
     cargo build --release --manifest-path="$PROJECT_ROOT/Cargo.toml" "${CARGO_BUILD_ARGS[@]}"
 
-    TARGET_DIR="$PROJECT_ROOT/target/release"
+    TARGET_DIR="$TARGET_ROOT/release"
 fi
 
 echo "Copying binaries to $DOCKER_DIR/"
diff --git a/testing/static/README.md b/testing/static/README.md
index 8d22a08..d922035 100644
--- a/testing/static/README.md
+++ b/testing/static/README.md
@@ -270,6 +270,19 @@ Runs iperf3 with:
 - Parallel streams: 8 (`-P 8`)
 - Protocol: TCP over IPv6
 
+For before/after measurements across commits or branches:
+
+```bash
+./testing/static/scripts/iperf-compare-refs.sh origin/master HEAD mesh
+```
+
+The comparison script builds each ref into a separate Docker image, runs the
+same topology and `iperf3` settings for both images, and prints a bandwidth
+summary. Override `DURATION`, `PARALLEL`, `SETTLE_SECONDS`, `IPERF_TIMEOUT`,
+or `RUNS` in the environment when needed. `RUNS` is the total number of
+measurements per ref; for example, `RUNS=3` runs each ref three times and
+prints both per-run and aggregate tables.
+
 ## Network Impairment
 
 The `netem.sh` script simulates adverse network conditions using `tc`/`netem`
diff --git a/testing/static/docker-compose.yml b/testing/static/docker-compose.yml
index fbc40c0..2fcdecb 100644
--- a/testing/static/docker-compose.yml
+++ b/testing/static/docker-compose.yml
@@ -13,7 +13,7 @@ networks:
         - subnet: fd02::/64
 
 x-fips-common: &fips-common
-  image: fips-test:latest
+  image: ${FIPS_TEST_IMAGE:-fips-test:latest}
   cap_add:
     - NET_ADMIN
   devices:
diff --git a/testing/static/scripts/iperf-compare-refs.sh b/testing/static/scripts/iperf-compare-refs.sh
new file mode 100755
index 0000000..ce6ec66
--- /dev/null
+++ b/testing/static/scripts/iperf-compare-refs.sh
@@ -0,0 +1,202 @@
+#!/bin/bash
+# Build two git refs into separate FIPS test images and run the same static
+# iperf3 topology against each one. This keeps performance PR evidence
+# repeatable: compare origin/master with a candidate branch using identical
+# configs, topology, duration, and parallelism.
+#
+# Usage:
+#   ./testing/static/scripts/iperf-compare-refs.sh <base-ref> <candidate-ref> [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 <base-ref> <candidate-ref> [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

From 59225ccfe1a056e5680ab4450c8558e26fbfa562 Mon Sep 17 00:00:00 2001
From: Martti Malmi <sirius@iki.fi>
Date: Fri, 15 May 2026 19:07:30 +0000
Subject: [PATCH 3/5] udp: batch macOS receive with recvmsg_x
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The Linux recv path drains up to 32 datagrams per kernel wakeup via
recvmmsg(2), amortising the per-syscall + per-task-wakeup cost across
the burst. macOS still fell through to single-packet recv_from, so
the same overhead capped inbound rate on Apple builds.

Add an equivalent batch path for Darwin using recvmsg_x(2). It is a
xnu-private syscall (not in the public SDK) but is the canonical
amortisation primitive on macOS — same shape used by quinn-udp for
the same reason. ABI is the public msghdr layout plus a trailing
msg_datalen (per-datagram bytes-received output), declared via
`unsafe extern "C"` against a local repr(C) `msghdr_x`.

Same `(count, kernel_drops)` contract as the Linux `recv_batch`. macOS
has no SO_RXQ_OVFL equivalent, so `kernel_drops` is always 0 — the
1Hz `sample_transport_congestion()` detector simply sees no kernel
drop signal on Apple hosts (it already tolerates that, since the
field has been 0 there pre-batching too).

cmsg buffer is intentionally null: we never consume ancillary data on
this path, and quinn-udp documents that `recvmsg_x` does not overwrite
`msg_controllen` on macOS 10.15+ (zeroed init is the only safe state).

The udp_receive_loop dispatch widens from cfg(linux) to
cfg(any(linux, macos)); the per-packet recv_from path is now used
only on the remaining unix targets (BSDs etc.) and Windows.

Add test_burst_recv_batch exercising 10 in-flight datagrams to
verify per-datagram boundaries and arrival order across the batch.
Add an ignored bench_udp_recv_amortization measuring recv-side
syscall amortization across 1/2/4/8 sender threads on dedicated
blocking std threads (kernel rx queue stays saturated regardless of
tokio scheduling). Sample numbers on aarch64-apple-darwin (100B
payloads, 3s windows):

  senders=1:  recv_from 398k pps   recv_batch 432k pps   1.09x
  senders=2:  recv_from 353k pps   recv_batch 608k pps   1.72x
  senders=4:  recv_from 322k pps   recv_batch 503k pps   1.56x
  senders=8:  recv_from 353k pps   recv_batch 515k pps   1.46x

Gate the Linux-only IpAddr import in control::listening behind a
cfg(target_os = "linux") so the macOS test build is warning-clean
now that test code paths there compile.
---
 src/control/listening.rs    |   4 +-
 src/transport/udp/mod.rs    |  57 +++++++-
 src/transport/udp/socket.rs | 276 +++++++++++++++++++++++++++++++++---
 3 files changed, 313 insertions(+), 24 deletions(-)

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/transport/udp/mod.rs b/src/transport/udp/mod.rs
index 2530def..27acf2d 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,7 +426,7 @@ 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;
@@ -497,7 +497,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 +955,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<SocketAddr>],
+            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::<libc::sockaddr_storage>() 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<AsyncUdpSocket, TransportError> {
             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<Vec<u8>> =
+                    (0..BATCH).map(|_| vec![0u8; PAYLOAD_LEN + 64]).collect();
+                let mut addrs: [Option<SocketAddr>; 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
+            );
+        }
+    }
 }

From b1af151aefc17c47b18c2921c1b578ac3448cbe7 Mon Sep 17 00:00:00 2001
From: Martti Malmi <sirius@iki.fi>
Date: Fri, 15 May 2026 21:35:46 +0000
Subject: [PATCH 4/5] rx: avoid copies in receive hot paths

- 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.
- recvmmsg / recvmsg_x (Linux + macOS) receive loop moves each filled
  slot buffer into ReceivedPacket via mem::replace instead of cloning
  it; a fresh empty buffer is installed for the next syscall.
- TransportAddr is formatted directly from the SocketAddr without
  going through an intermediate String.

Focused decode bench: ref 1.6 ns/op vs owned 34.7 ns/op (21.4x).
End-to-end iperf is neutral as expected for a ~30 ns saving per
packet.

Unit tests added:
  - test_session_datagram_ref_decode_borrows_payload (verifies the
    payload slice pointer equals the input slice's offset 35, a real
    zero-copy invariant guard against accidental future to_vec)
  - bench_session_datagram_decode_owned_vs_ref (ignored, run with
    --ignored --nocapture)
  - test_transport_addr_from_socket_addr
---
 CHANGELOG.md                    |  12 ++++
 src/node/handlers/forwarding.rs |  35 ++++++-----
 src/protocol/link.rs            | 104 +++++++++++++++++++++++++++++++-
 src/protocol/mod.rs             |   2 +-
 src/transport/mod.rs            |  18 ++++++
 src/transport/udp/mod.rs        |  12 ++--
 6 files changed, 161 insertions(+), 22 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index dd53edd..3a98929 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -46,6 +46,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
     is not tied to release tags.
   - Tag-triggered `package-*` release-build workflows remain
     untouched.
+- 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
 
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/protocol/link.rs b/src/protocol/link.rs
index 1c4781d..102f814 100644
--- a/src/protocol/link.rs
+++ b/src/protocol/link.rs
@@ -297,6 +297,19 @@ pub struct SessionDatagram {
     pub payload: Vec<u8>,
 }
 
+/// 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;
 
@@ -353,6 +366,14 @@ impl SessionDatagram {
 
     /// Decode from link-layer payload (after msg_type byte has been consumed).
     pub fn decode(payload: &[u8]) -> Result<Self, ProtocolError> {
+        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<Self, ProtocolError> {
         // ttl(1) + path_mtu(2) + src_addr(16) + dest_addr(16) = 35
         if payload.len() < 35 {
             return Err(ProtocolError::MessageTooShort {
@@ -366,16 +387,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
@@ -547,6 +578,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 1e90519..99e898c 100644
--- a/src/protocol/mod.rs
+++ b/src/protocol/mod.rs
@@ -33,7 +33,7 @@ pub use error::ProtocolError;
 pub use filter::FilterAnnounce;
 pub use link::{
     Disconnect, DisconnectReason, HandshakeMessageType, LinkMessageType,
-    SESSION_DATAGRAM_HEADER_SIZE, SessionDatagram,
+    SESSION_DATAGRAM_HEADER_SIZE, SessionDatagram, SessionDatagramRef,
 };
 pub use session::{
     COORDS_REQUIRED_SIZE, CoordsRequired, FspFlags, FspInnerFlags, MTU_EXCEEDED_SIZE, MtuExceeded,
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<u8>::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 27acf2d..7fddcbd 100644
--- a/src/transport/udp/mod.rs
+++ b/src/transport/udp/mod.rs
@@ -430,7 +430,9 @@ async fn udp_receive_loop(
     {
         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<Vec<u8>> = (0..BATCH).map(|_| vec![0u8; buf_size]).collect();
         let mut addrs: [Option<std::net::SocketAddr>; 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!(

From 49bd21048089d823ac56fc112724a5c0e0dad1e3 Mon Sep 17 00:00:00 2001
From: Johnathan Corgan <johnathan@corganlabs.com>
Date: Sat, 16 May 2026 21:53:55 +0000
Subject: [PATCH 5/5] cache: scope coord cache invalidation to entries actually
 affected by topology change
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Replaces the unconditional `CoordCache::clear()` calls at parent-switch,
become-root, and loop-detection sites with two targeted invalidation
methods scoped to what actually makes an entry stale:

- `invalidate_via_node(node_addr)`: drop entries whose cached
  destination ancestry contains `node_addr`. Used at parent-position-
  change sites — our prefix changed, so destinations downstream of
  us have stale-prefix coords.
- `invalidate_other_roots(current_root)`: drop entries rooted under
  a different root than the current one. Used at root-change sites.

Under the previous global flush, parent switches blanked the cache
across the board, leaving `find_next_hop` returning `None` for every
non-direct-peer destination until the cache passively re-warmed via
incoming TreeAnnounces / SessionSetup. Surgical invalidation
preserves entries that remain correct after the topology change.

The cached coord describes a destination's tree position; that
position only goes stale relative to our own routing decisions when
our own prefix changes (entries we are downstream of) or the root
changes (entries in a different tree). Peer removal does not
invalidate cached coords: `Node::find_next_hop` recomputes the
next-hop decision on every call against the current peer set, bloom
filters, and tree state, and Discovery already triggers on
`no route to destination` errors when a destination becomes
unroutable through us. The peer-removal site retains the original
"no cache invalidation" behavior.

Each method returns the count of entries removed for observability.
Unit tests cover each method against the cases enumerated in the
acceptance criterion.
---
 src/cache/coord_cache.rs | 131 +++++++++++++++++++++++++++++++++++++++
 src/node/handlers/mmp.rs |   8 ++-
 src/node/tree.rs         |  29 ++++++---
 3 files changed, 159 insertions(+), 9 deletions(-)

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/node/handlers/mmp.rs b/src/node/handlers/mmp.rs
index c421c3c..b505b4e 100644
--- a/src/node/handlers/mmp.rs
+++ b/src/node/handlers/mmp.rs
@@ -153,7 +153,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;
@@ -178,7 +180,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/tree.rs b/src/node/tree.rs
index 1d69a5c..336c8b0 100644
--- a/src/node/tree.rs
+++ b/src/node/tree.rs
@@ -250,7 +250,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;
@@ -261,7 +263,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;
@@ -281,7 +283,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;
@@ -315,7 +319,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;
                 }
@@ -350,7 +359,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<NodeAddr> =
@@ -439,7 +450,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;
@@ -468,7 +481,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;