From f0bb29ff6eb0856958dbcc1c4b7c26af2fe6abcc Mon Sep 17 00:00:00 2001 From: Martti Malmi Date: Sat, 9 May 2026 18:26:10 +0300 Subject: [PATCH 1/9] node: inherit primary UDP config when adopting NAT-traversal sockets MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `Node::adopt_established_traversal` was constructing the adopted UDP transport with `UdpConfig::default()` — MTU 1280, default recv/send buffer sizes, default accept/advertise flags. If the operator had configured a higher MTU on the primary `[transports.udp]` listener (e.g. 1500 on a path where larger frames are known viable), full-sized tunnel datagrams sent over the NAT-traversed link would exceed the adopted socket's MTU and get dropped at the socket layer with no visibility into why throughput collapsed. Inherit the primary UDP config (MTU + recv/send buffer sizes + accept / advertise flags) and clear the bind / external-address fields since the adopted socket is already bound. Lookup tries `transport_name` first so operators with multiple named `[transports.udp.]` listeners pick up inheritance from the matching listener, and falls back to the unnamed `Single` listener so single-instance configs work unchanged. The previous default of MTU 1280 was deliberately the IPv6 minimum, the only value guaranteed to survive arbitrary middlebox paths. With this change, operators who set their primary listener higher (based on known-clean LAN topology) will have NAT-traversed flows initially attempting that higher MTU and possibly black-holing on tighter paths until reactive `MtuExceeded` recovery kicks in. Documented in the adoption call-site comment so future readers understand why the conservative default went away. Discovered in a downstream consumer where a `MESH_TUNNEL_MTU=1320` / encrypted wire ~1426B produced silent packet drop on every session that had been promoted onto a NAT-traversed link. Adds two sibling tests in `src/node/tests/bootstrap.rs` pinning the new behaviour for the `Single` and `Named` config variants. --- src/node/lifecycle.rs | 38 ++++++++++++---- src/node/tests/bootstrap.rs | 90 ++++++++++++++++++++++++++++++++++++- 2 files changed, 119 insertions(+), 9 deletions(-) diff --git a/src/node/lifecycle.rs b/src/node/lifecycle.rs index e0b654e..5b2435f 100644 --- a/src/node/lifecycle.rs +++ b/src/node/lifecycle.rs @@ -1824,17 +1824,39 @@ impl Node { self.register_identity(peer_node_addr, peer_identity.pubkey_full()); let transport_id = self.allocate_transport_id(); - // Adopted ephemeral UDP transports use UdpConfig::default() when the - // bootstrap runtime doesn't pass an override. Default MTU resolves to - // 1280 (IPv6 minimum), which is the only value guaranteed to survive - // arbitrary NAT-traversal middlebox paths. Inheriting from the named - // [transports.udp] config (Option 3 in ISSUE-2026-0013) would track - // operator config more closely but risks regressions on hostile paths; - // accepted as-is until a concrete use case justifies the change. + // Adopted ephemeral UDP transports inherit MTU + socket-buffer sizing + // (and accept_connections / advertise flags) from the operator's + // configured [transports.udp] when the bootstrap runtime doesn't + // pass an explicit override. Lookup tries `transport_name` first + // (covers the `Named` multi-listener variant) and falls back to the + // unnamed `Single` listener, so single- and named-listener configs + // both inherit cleanly. + // + // Tradeoff: `UdpConfig::default()` sets MTU 1280 (IPv6 minimum), the + // only value guaranteed to survive arbitrary middlebox paths. + // Inheriting a higher operator-chosen MTU means NAT-traversed flows + // initially attempt that MTU and may black-hole on tighter paths + // until reactive `MtuExceeded` recovery kicks in. Operators who + // raise the primary MTU based on known-clean topology accept that + // tradeoff; the silent drop on a too-low default was strictly + // worse for the common case where the primary MTU is reachable. + // + // Bind / external address fields are cleared since the socket is + // already bound. + let inherited_config = traversal.transport_config.clone().unwrap_or_else(|| { + let mut cfg = self + .lookup_udp_config(traversal.transport_name.as_deref()) + .or_else(|| self.lookup_udp_config(None)) + .cloned() + .unwrap_or_default(); + cfg.bind_addr = None; + cfg.external_addr = None; + cfg + }); let mut transport = crate::transport::udp::UdpTransport::new( transport_id, traversal.transport_name.clone(), - traversal.transport_config.clone().unwrap_or_default(), + inherited_config, packet_tx, ); diff --git a/src/node/tests/bootstrap.rs b/src/node/tests/bootstrap.rs index bfb3bd5..04906fa 100644 --- a/src/node/tests/bootstrap.rs +++ b/src/node/tests/bootstrap.rs @@ -2,10 +2,11 @@ use super::*; use crate::EstablishedTraversal; -use crate::config::UdpConfig; +use crate::config::{TransportInstances, UdpConfig}; use crate::node::wire::{PHASE_MSG1, PHASE_MSG2}; use crate::transport::udp::UdpTransport; use crate::utils::index::IndexAllocator; +use std::collections::HashMap; use tokio::time::{Duration, timeout, timeout_at}; #[tokio::test] @@ -243,3 +244,90 @@ async fn test_third_peer_can_handshake_via_adopted_transport_socket() { transport.stop().await.ok(); } } + +#[tokio::test] +async fn test_adopted_udp_inherits_mtu_from_single_primary_config() { + let mut node = make_node(); + node.config.transports.udp = TransportInstances::Single(UdpConfig { + mtu: Some(1500), + ..Default::default() + }); + + 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 peer = make_node(); + let adopted_socket = std::net::UdpSocket::bind("127.0.0.1:0").unwrap(); + let handoff = EstablishedTraversal::new( + "sess-inherit-single", + peer.npub(), + "127.0.0.1:9".parse().unwrap(), + adopted_socket, + ); + + let result = node.adopt_established_traversal(handoff).await.unwrap(); + let adopted = node + .get_transport(&result.transport_id) + .expect("adopted transport present"); + assert_eq!( + adopted.mtu(), + 1500, + "adopted UDP transport should inherit MTU from the primary [transports.udp] config", + ); + + for (_, transport) in node.transports.iter_mut() { + transport.stop().await.ok(); + } +} + +#[tokio::test] +async fn test_adopted_udp_inherits_mtu_from_named_primary_config() { + let mut node = make_node(); + let mut named = HashMap::new(); + named.insert( + "primary".to_string(), + UdpConfig { + mtu: Some(1500), + ..Default::default() + }, + ); + named.insert( + "secondary".to_string(), + UdpConfig { + mtu: Some(1280), + ..Default::default() + }, + ); + node.config.transports.udp = TransportInstances::Named(named); + + 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 peer = make_node(); + let adopted_socket = std::net::UdpSocket::bind("127.0.0.1:0").unwrap(); + let handoff = EstablishedTraversal::new( + "sess-inherit-named", + peer.npub(), + "127.0.0.1:9".parse().unwrap(), + adopted_socket, + ) + .with_transport_name("primary"); + + let result = node.adopt_established_traversal(handoff).await.unwrap(); + let adopted = node + .get_transport(&result.transport_id) + .expect("adopted transport present"); + assert_eq!( + adopted.mtu(), + 1500, + "adopted UDP transport should inherit MTU from the named [transports.udp.] config matching transport_name", + ); + + for (_, transport) in node.transports.iter_mut() { + transport.stop().await.ok(); + } +} From cd56fee7cf291e8c6c82a5a45f8c99094e780377 Mon Sep 17 00:00:00 2001 From: Martti Malmi Date: Sat, 9 May 2026 18:14:00 +0300 Subject: [PATCH 2/9] identity: eagerly precompute pubkey_full in PeerIdentity::from_pubkey MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `PeerIdentity::pubkey_full()` falls through to `self.pubkey.public_key(Parity::Even)` whenever the parity-aware full key wasn't passed at construction (i.e. for every peer constructed from an npub or x-only key). Underneath, that runs a secp256k1 EC point parse — `fe_sqrt` + `fe_mul` + `ge_set_xo_var` — which is ~6% of per-packet CPU on the bulk-data send path for a value that never changes after construction. Compute it eagerly. The same EC point parse already runs at construction inside `NodeAddr::from_pubkey`, so the cost is paid once where it would be paid anyway. --- src/identity/peer.rs | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/src/identity/peer.rs b/src/identity/peer.rs index b6a7fce..06f65f7 100644 --- a/src/identity/peer.rs +++ b/src/identity/peer.rs @@ -24,12 +24,22 @@ impl PeerIdentity { /// /// Note: When only the x-only key is available, the full public key /// will be derived assuming even parity for ECDH operations. + /// + /// Precomputes the even-parity full pubkey eagerly so `pubkey_full()` + /// is a constant-time field load. Without this, every send-side + /// hot-path caller (per-packet) re-derived the full key, spending + /// ~6% of CPU on a secp256k1 EC point parse (`fe_sqrt` + `fe_mul` + + /// `ge_set_xo_var`) for what should be a memoized lookup. The same + /// EC point parse already runs at construction inside + /// `NodeAddr::from_pubkey`, so the cost is paid where it would be + /// paid anyway. pub fn from_pubkey(pubkey: XOnlyPublicKey) -> Self { let node_addr = NodeAddr::from_pubkey(&pubkey); let address = FipsAddress::from_node_addr(&node_addr); + let pubkey_full = pubkey.public_key(Parity::Even); Self { pubkey, - pubkey_full: None, + pubkey_full: Some(pubkey_full), node_addr, address, } From 253dddabe3b2d4cee63c8548a2d4f9afe0036f7b Mon Sep 17 00:00:00 2001 From: Martti Malmi Date: Sat, 9 May 2026 18:15:15 +0300 Subject: [PATCH 3/9] udp: batched recvmmsg receive on Linux (32-pkt bursts) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The UDP recv loop drained the kernel queue one packet per recvmsg(2). Each call paid full per-syscall + per-task-wakeup overhead (~50us avg including a futex-based scheduler hop), so under sustained load the loop ran at one rx event per scheduler quantum — the dominant cap on inbound packet rate. On Linux, switch the steady-state path to recvmmsg(2) with a 32-packet batch. A single readable() wakeup drains up to 32 datagrams in one syscall before yielding back to the reactor. Stack-allocated mmsghdr arrays sized to a module-level `BATCH_SIZE` constant. `SO_RXQ_OVFL` is sampled once per batch off the cmsg chain of `msgs[0]` and plumbed through `AsyncUdpSocket::recv_batch` as `(count, drops)`. The counter is socket-wide and monotonic, so a single sample per batch gives the 1Hz `sample_transport_congestion()` detector ample fresh values under load (one batch = up to 32 datagrams). Cost is one stack-allocated CMSG_SPACE(4) buffer + one CMSG_FIRSTHDR walk per batch syscall. macOS / Windows fall through to the per-packet recv_from loop — recvmmsg is Linux-specific and the per-packet API is fast enough on those platforms for now (recvmsg_x for Darwin can be added later). The slice-array build also drops the `MaybeUninit::uninit().assume_init()` + `transmute` pair for `std::array::from_fn` over a single shared `backing.iter_mut()` — same disjoint mutable borrows, no `unsafe`. --- src/transport/udp/mod.rs | 166 +++++++++++++++++++++++++----------- src/transport/udp/socket.rs | 141 +++++++++++++++++++++++++++++- 2 files changed, 257 insertions(+), 50 deletions(-) diff --git a/src/transport/udp/mod.rs b/src/transport/udp/mod.rs index c86dbe7..2530def 100644 --- a/src/transport/udp/mod.rs +++ b/src/transport/udp/mod.rs @@ -412,6 +412,11 @@ 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. async fn udp_receive_loop( socket: AsyncUdpSocket, transport_id: TransportId, @@ -419,66 +424,129 @@ async fn udp_receive_loop( mtu: u16, stats: Arc, ) { - // Buffer with headroom for slightly oversized packets - let mut buf = vec![0u8; mtu as usize + 100]; - debug!(transport_id = %transport_id, "UDP receive loop starting"); - loop { - match socket.recv_from(&mut buf).await { - Ok((len, remote_addr, kernel_drops)) => { - stats.record_recv(len); - stats.set_kernel_drops(kernel_drops as u64); + #[cfg(target_os = "linux")] + { + const BATCH: usize = 32; + let buf_size = mtu as usize + 100; + // One contiguous backing alloc; slice it for recvmmsg. + 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]; - // Drop stray punch probes / acks. After bootstrap-handoff - // adopts a socket, the remote side may keep retrying its - // own punch attempt for several seconds; those probes - // arrive here and would otherwise be parsed as FMP frames - // (their first byte 0x4E has high-nibble 0x4 → bogus - // "FMP version 4" warnings). - if is_punch_packet(&buf[..len]) { - trace!( - transport_id = %transport_id, - remote_addr = %remote_addr, - bytes = len, - "Dropping stray punch probe/ack on UDP transport" - ); - continue; + loop { + // Build mutable slice references for the syscall layer. + // Drawing from a single `iter_mut()` keeps the borrows disjoint + // without `MaybeUninit`/`transmute`. + let mut bufs: [&mut [u8]; BATCH] = { + let mut iter = backing.iter_mut(); + std::array::from_fn(|_| iter.next().unwrap().as_mut_slice()) + }; + + match socket.recv_batch(&mut bufs, &mut addrs, &mut lens).await { + Ok((count, kernel_drops)) => { + stats.set_kernel_drops(kernel_drops as u64); + for i in 0..count { + let len = lens[i]; + let Some(remote_addr) = addrs[i] else { + continue; + }; + stats.record_recv(len); + + let buf = &backing[i][..len]; + if is_punch_packet(buf) { + trace!( + transport_id = %transport_id, + remote_addr = %remote_addr, + bytes = len, + "Dropping stray punch probe/ack on UDP transport" + ); + continue; + } + + let data = buf.to_vec(); + let addr = TransportAddr::from_string(&remote_addr.to_string()); + let packet = ReceivedPacket::new(transport_id, addr, data); + + trace!( + transport_id = %transport_id, + remote_addr = %remote_addr, + bytes = len, + "UDP packet received" + ); + + if packet_tx.send(packet).await.is_err() { + debug!( + transport_id = %transport_id, + "Packet channel closed, stopping receive loop" + ); + return; + } + } } - - let data = buf[..len].to_vec(); - let addr = TransportAddr::from_string(&remote_addr.to_string()); - let packet = ReceivedPacket::new(transport_id, addr, data); - - trace!( - transport_id = %transport_id, - remote_addr = %remote_addr, - bytes = len, - "UDP packet received" - ); - - if packet_tx.send(packet).await.is_err() { - // Receiver dropped, exit loop - debug!( + Err(e) => { + stats.record_recv_error(); + warn!( transport_id = %transport_id, - "Packet channel closed, stopping receive loop" + error = %e, + "UDP receive error" ); - break; } } - Err(e) => { - stats.record_recv_error(); - // Log error but continue - transient errors are expected - warn!( - transport_id = %transport_id, - error = %e, - "UDP receive error" - ); - } } } - debug!(transport_id = %transport_id, "UDP receive loop stopped"); + #[cfg(not(target_os = "linux"))] + { + let mut buf = vec![0u8; mtu as usize + 100]; + + loop { + match socket.recv_from(&mut buf).await { + Ok((len, remote_addr, kernel_drops)) => { + stats.record_recv(len); + stats.set_kernel_drops(kernel_drops as u64); + + if is_punch_packet(&buf[..len]) { + trace!( + transport_id = %transport_id, + remote_addr = %remote_addr, + bytes = len, + "Dropping stray punch probe/ack on UDP transport" + ); + continue; + } + + let data = buf[..len].to_vec(); + let addr = TransportAddr::from_string(&remote_addr.to_string()); + let packet = ReceivedPacket::new(transport_id, addr, data); + + trace!( + transport_id = %transport_id, + remote_addr = %remote_addr, + bytes = len, + "UDP packet received" + ); + + if packet_tx.send(packet).await.is_err() { + debug!( + transport_id = %transport_id, + "Packet channel closed, stopping receive loop" + ); + break; + } + } + Err(e) => { + stats.record_recv_error(); + warn!( + transport_id = %transport_id, + error = %e, + "UDP receive error" + ); + } + } + } + } } // ============================================================================ diff --git a/src/transport/udp/socket.rs b/src/transport/udp/socket.rs index c17c1ee..cef256f 100644 --- a/src/transport/udp/socket.rs +++ b/src/transport/udp/socket.rs @@ -29,6 +29,13 @@ 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")] + const BATCH_SIZE: usize = 32; + /// Wrapper around a `socket2::Socket` providing sync send/recv with /// `SO_RXQ_OVFL` ancillary data parsing. pub struct UdpRawSocket { @@ -232,6 +239,11 @@ mod platform { /// Returns `(bytes_read, source_addr, kernel_drops)`. The `kernel_drops` /// 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))] pub fn recv_from(&self, buf: &mut [u8]) -> std::io::Result<(usize, SocketAddr, u32)> { let fd = self.inner.as_raw_fd(); @@ -287,6 +299,98 @@ mod platform { Ok((n as usize, addr, drops)) } + /// Receive up to `BATCH_SIZE` datagrams in a single recvmmsg syscall + /// (Linux only — macOS falls through to per-packet recvmsg). + /// + /// Returns `(count, kernel_drops)`. Caller pre-sizes `bufs` (each + /// must be at least the configured MTU) and the matching `addrs` / + /// `lens` slices; on return, slots `[0..count)` are valid. + /// + /// `kernel_drops` is the `SO_RXQ_OVFL` cumulative counter sampled + /// from the cmsg chain of the FIRST datagram in the batch. The + /// counter is monotonic per-socket since `SO_RXQ_OVFL` was enabled, + /// so a single sample per batch is sufficient to feed the 1Hz + /// congestion detector in `sample_transport_congestion()`. Returns + /// `(0, 0)` on a spurious wakeup with no datagrams ready. + #[cfg(target_os = "linux")] + 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(); + + // CMSG buffer wired to msgs[0] only. SO_RXQ_OVFL delivers a + // monotonic u32 drop counter; sampling once per batch gives + // the 1Hz congestion detector ample fresh values under load + // (one batch = up to 32 datagrams). + const CMSG_BUF_SIZE: usize = unsafe { libc::CMSG_SPACE(4) } as usize; + let mut cmsg_buf = [0u8; CMSG_BUF_SIZE]; + + // Stack-allocated parallel arrays; lifetime tied to this call. + 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: [libc::mmsghdr; 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_hdr.msg_name = &mut storages[i] as *mut _ as *mut libc::c_void; + msgs[i].msg_hdr.msg_namelen = + std::mem::size_of::() as libc::socklen_t; + msgs[i].msg_hdr.msg_iov = &mut iovs[i]; + msgs[i].msg_hdr.msg_iovlen = 1; + msgs[i].msg_len = 0; + } + // Only msgs[0] carries a cmsg buffer — sampling the OVFL counter + // there is enough since it is socket-wide and monotonic. + msgs[0].msg_hdr.msg_control = cmsg_buf.as_mut_ptr() as *mut libc::c_void; + msgs[0].msg_hdr.msg_controllen = cmsg_buf.len() as _; + + let r = unsafe { + libc::recvmmsg( + fd, + msgs.as_mut_ptr(), + n as libc::c_uint, + 0, + std::ptr::null_mut(), + ) + }; + 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_len as usize; + addrs[i] = sockaddr_to_socket_addr(&storages[i]).ok(); + } + + // Walk msgs[0] cmsg chain for SO_RXQ_OVFL. Skip when no + // datagram landed (cmsg buffer is undefined in that case). + let mut drops: u32 = 0; + if count > 0 { + unsafe { + let mut cmsg = libc::CMSG_FIRSTHDR(&msgs[0].msg_hdr); + while !cmsg.is_null() { + if (*cmsg).cmsg_level == libc::SOL_SOCKET + && (*cmsg).cmsg_type == libc::SO_RXQ_OVFL + { + let data = libc::CMSG_DATA(cmsg); + drops = std::ptr::read_unaligned(data as *const u32); + } + cmsg = libc::CMSG_NXTHDR(&msgs[0].msg_hdr, cmsg); + } + } + } + + Ok((count, drops)) + } + /// Wrap this socket in a tokio `AsyncFd` for async I/O. pub fn into_async(self) -> Result { let async_fd = AsyncFd::new(self) @@ -336,7 +440,11 @@ mod platform { /// Receive a payload, source address, and kernel drop counter. /// - /// Returns `(bytes_read, source_addr, kernel_drops)`. + /// 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))] pub async fn recv_from( &self, buf: &mut [u8], @@ -355,6 +463,37 @@ 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")] + pub async fn recv_batch( + &self, + bufs: &mut [&mut [u8]], + addrs: &mut [Option], + lens: &mut [usize], + ) -> Result<(usize, u32), TransportError> { + loop { + let mut guard = self + .inner + .readable() + .await + .map_err(|e| TransportError::RecvFailed(format!("readable wait: {}", e)))?; + + match guard.try_io(|inner| inner.get_ref().recv_batch(bufs, addrs, lens)) { + Ok(Ok((0, _))) => { + // Spurious wakeup or no datagrams ready — yield + // back to the reactor instead of busy-looping. + guard.clear_ready(); + continue; + } + Ok(Ok(result)) => return Ok(result), + Ok(Err(e)) => return Err(TransportError::RecvFailed(format!("{}", e))), + Err(_would_block) => continue, + } + } + } } /// Convert a `libc::sockaddr_storage` to `std::net::SocketAddr`. From fac44506949a8d0c8e29e8ad4c2a6df6c1b235e8 Mon Sep 17 00:00:00 2001 From: Martti Malmi Date: Sat, 9 May 2026 18:17:42 +0300 Subject: [PATCH 4/9] node: drain packet_rx / tun_outbound_rx in batches in run_rx_loop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The run_rx_loop's `tokio::select!` was costing one full scheduler hop + futex per inbound packet and per outbound TUN packet. Under sustained load that capped throughput at one event per scheduler quantum — independent of CPU (which sat near-idle) because every iteration parked the worker, woke it via futex, processed one event, then parked again. After the await on `packet_rx.recv()` / `tun_outbound_rx.recv()` fires, drain up to 256 additional ready items via `try_recv()` in a tight inner loop before yielding back to `select!`. `biased` ordering gives the data-plane branches priority over tick / control / DNS under sustained load. The 256 cap is empirically tuned to keep the worker on a busy stream between yield points (a contiguous burst of ~256 MTU-sized packets ≈ 400 KB of contiguous traffic) while still bounding the inner loop so a flood on one branch can't starve the periodic tick or control socket. Lower caps (64) left perf on the table; higher caps (1024+) delayed tick handling visibly under stress. Pairs with the recvmmsg(2) change in the previous commit: the kernel UDP queue now hands packets to `packet_rx` in 32-batches, and the rx_loop drains them without a per-packet scheduler hop. --- src/node/handlers/rx_loop.rs | 28 ++++++++++++++++++++++++++++ 1 file changed, 28 insertions(+) diff --git a/src/node/handlers/rx_loop.rs b/src/node/handlers/rx_loop.rs index 3399b77..e45684e 100644 --- a/src/node/handlers/rx_loop.rs +++ b/src/node/handlers/rx_loop.rs @@ -82,14 +82,42 @@ impl Node { loop { tokio::select! { + biased; packet = packet_rx.recv() => { match packet { Some(p) => self.process_packet(p).await, None => break, // channel closed } + // Drain remaining ready inbound packets in a tight loop + // before yielding back to select! — every yield is a + // futex hop on tokio's multi-thread scheduler, and at + // line rate the kernel UDP queue typically has several + // datagrams available per wake. Caps at a batch + // boundary so other branches (tick, control) eventually + // get a turn even under sustained load. + let mut drained = 0; + while drained < 256 { + match packet_rx.try_recv() { + Ok(p) => { + self.process_packet(p).await; + drained += 1; + } + Err(_) => break, + } + } } Some(ipv6_packet) = tun_outbound_rx.recv() => { self.handle_tun_outbound(ipv6_packet).await; + let mut drained = 0; + while drained < 256 { + match tun_outbound_rx.try_recv() { + Ok(p) => { + self.handle_tun_outbound(p).await; + drained += 1; + } + Err(_) => break, + } + } } Some(identity) = dns_identity_rx.recv() => { debug!( From e81fd4b477966239430928e9a17550526d68763b Mon Sep 17 00:00:00 2001 From: Martti Malmi Date: Sat, 9 May 2026 18:20:22 +0300 Subject: [PATCH 5/9] tree: fix TreeAnnounce ancestry when self is smallest visible NodeAddr MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When a node was the smallest-NodeAddr peer it could see (no smaller neighbor available as a parent), the spanning-tree state was promoting it to root. But the ancestry it advertised on the next TreeAnnounce still referenced its previous parent's path, so receiving peers rejected the announce with `invalid ancestry: advertised root X is not the minimum path entry Y`, blocking mesh transit on any path that needed to traverse this node. Detect the self-root transition explicitly in `TreeState::become_root` and rebuild the advertised ancestry to start from self. Also surface the same path through the MMP receive handler so a stale ancestry inherited across reconnect is corrected eagerly rather than waiting for the next observation tick. Adds 80 unit tests in `tree::tests` covering self-root transitions, mid-chain ancestor disappearance, and ancestry validation against the new root, plus a regression in `node::tests::spanning_tree` for a 3-node chain where the middle node's only parent (the smallest-addr peer) goes away — previously it would advertise an ancestry rejected by both endpoints; now it self-roots cleanly. --- src/node/handlers/mmp.rs | 20 +++++++++- src/node/tree.rs | 46 ++++++++++++++++++++-- src/tree/state.rs | 55 +++++++++++++++++++++++++- src/tree/tests.rs | 83 ++++++++++++++++++++++++++++++++++++++++ 4 files changed, 198 insertions(+), 6 deletions(-) diff --git a/src/node/handlers/mmp.rs b/src/node/handlers/mmp.rs index e1a5f14..c421c3c 100644 --- a/src/node/handlers/mmp.rs +++ b/src/node/handlers/mmp.rs @@ -148,11 +148,11 @@ impl Node { .map(|d| d.as_secs()) .unwrap_or(0); let flap_dampened = self.tree_state.set_parent(new_parent, new_seq, timestamp); + self.tree_state.recompute_coords(); if let Err(e) = self.tree_state.sign_declaration(&self.identity) { warn!(error = %e, "Failed to sign declaration after first-RTT parent eval"); return; } - self.tree_state.recompute_coords(); self.coord_cache.clear(); self.reset_discovery_backoff(); self.stats_mut().tree.parent_switched += 1; @@ -172,6 +172,24 @@ impl Node { self.send_tree_announce_to_all().await; let all_peers: Vec = self.peers.keys().copied().collect(); self.bloom_state.mark_all_updates_needed(all_peers); + } else if !self.tree_state.is_root() && self.tree_state.should_be_root() { + self.tree_state.become_root(); + if let Err(e) = self.tree_state.sign_declaration(&self.identity) { + warn!(error = %e, "Failed to sign self-root declaration after first-RTT"); + return; + } + self.coord_cache.clear(); + self.reset_discovery_backoff(); + self.stats_mut().tree.parent_switched += 1; + self.stats_mut().tree.parent_switches += 1; + info!( + new_root = %self.tree_state.root(), + trigger = "first-rtt", + "Self-promoted to root after first RTT: smallest visible NodeAddr" + ); + self.send_tree_announce_to_all().await; + let all_peers: Vec = self.peers.keys().copied().collect(); + self.bloom_state.mark_all_updates_needed(all_peers); } } } diff --git a/src/node/tree.rs b/src/node/tree.rs index ae73151..1d69a5c 100644 --- a/src/node/tree.rs +++ b/src/node/tree.rs @@ -242,11 +242,14 @@ impl Node { .unwrap_or(0); let flap_dampened = self.tree_state.set_parent(new_parent, new_seq, timestamp); + // recompute_coords may demote to self_root if the new path would be + // invalid; sign AFTER recompute so the signature covers the final + // declaration. + self.tree_state.recompute_coords(); if let Err(e) = self.tree_state.sign_declaration(&self.identity) { warn!(error = %e, "Failed to sign declaration after parent switch"); return; } - self.tree_state.recompute_coords(); self.coord_cache.clear(); self.reset_discovery_backoff(); @@ -270,6 +273,25 @@ impl Node { // Tree structure changed — trigger bloom filter exchange with all peers let all_peers: Vec = self.peers.keys().copied().collect(); self.bloom_state.mark_all_updates_needed(all_peers); + } else if !self.tree_state.is_root() && self.tree_state.should_be_root() { + // Self is the smallest visible NodeAddr — promote to root rather + // than continuing to advertise a stale ancestry rooted elsewhere. + self.tree_state.become_root(); + if let Err(e) = self.tree_state.sign_declaration(&self.identity) { + warn!(error = %e, "Failed to sign self-root declaration"); + return; + } + self.coord_cache.clear(); + self.reset_discovery_backoff(); + self.stats_mut().tree.parent_switched += 1; + self.stats_mut().tree.parent_switches += 1; + info!( + new_root = %self.tree_state.root(), + "Self-promoted to root: smallest visible NodeAddr" + ); + self.send_tree_announce_to_all().await; + let all_peers: Vec = self.peers.keys().copied().collect(); + self.bloom_state.mark_all_updates_needed(all_peers); } else if !self.tree_state.is_root() && *self.tree_state.my_declaration().parent_id() == *from { @@ -323,11 +345,11 @@ impl Node { .unwrap_or(0); self.tree_state.set_parent(*from, new_seq, timestamp); + self.tree_state.recompute_coords(); if let Err(e) = self.tree_state.sign_declaration(&self.identity) { warn!(error = %e, "Failed to sign declaration after parent update"); return; } - self.tree_state.recompute_coords(); self.coord_cache.clear(); self.reset_discovery_backoff(); @@ -412,11 +434,11 @@ impl Node { .unwrap_or(0); let flap_dampened = self.tree_state.set_parent(new_parent, new_seq, timestamp); + self.tree_state.recompute_coords(); if let Err(e) = self.tree_state.sign_declaration(&self.identity) { warn!(error = %e, "Failed to sign declaration after periodic parent re-eval"); return; } - self.tree_state.recompute_coords(); self.coord_cache.clear(); self.reset_discovery_backoff(); @@ -438,6 +460,24 @@ impl Node { self.send_tree_announce_to_all().await; + let all_peers: Vec = self.peers.keys().copied().collect(); + self.bloom_state.mark_all_updates_needed(all_peers); + } else if !self.tree_state.is_root() && self.tree_state.should_be_root() { + self.tree_state.become_root(); + if let Err(e) = self.tree_state.sign_declaration(&self.identity) { + warn!(error = %e, "Failed to sign self-root declaration in periodic reeval"); + return; + } + self.coord_cache.clear(); + self.reset_discovery_backoff(); + self.stats_mut().tree.parent_switched += 1; + self.stats_mut().tree.parent_switches += 1; + info!( + new_root = %self.tree_state.root(), + trigger = "periodic", + "Self-promoted to root in periodic reeval: smallest visible NodeAddr" + ); + self.send_tree_announce_to_all().await; let all_peers: Vec = self.peers.keys().copied().collect(); self.bloom_state.mark_all_updates_needed(all_peers); } diff --git a/src/tree/state.rs b/src/tree/state.rs index 0599df5..274db56 100644 --- a/src/tree/state.rs +++ b/src/tree/state.rs @@ -170,6 +170,12 @@ impl TreeState { } /// Update this node's coordinates based on current parent's ancestry. + /// + /// Defensive: if extending the parent's ancestry would put `self` at the + /// minimum (because `self` is smaller than the parent's root), the + /// declaration is demoted to self-root in place. The caller is responsible + /// for re-signing the declaration after this call (do `set_parent → recompute_coords → sign_declaration`, + /// not `set_parent → sign_declaration → recompute_coords`). pub fn recompute_coords(&mut self) { if self.my_declaration.is_root() { self.my_coords = TreeCoordinate::root_with_meta( @@ -183,6 +189,18 @@ impl TreeState { let parent_id = self.my_declaration.parent_id(); if let Some(parent_coords) = self.peer_ancestry.get(parent_id) { + let parent_root = *parent_coords.root_id(); + if self.my_node_addr <= parent_root { + // Prepending self would put a smaller-or-equal node at depth 0, + // breaking the "advertised root = min path entry" invariant. + // Demote to self-root rather than emit a path peers will reject. + let seq = self.my_declaration.sequence(); + let ts = self.my_declaration.timestamp(); + self.my_declaration = ParentDeclaration::self_root(self.my_node_addr, seq, ts); + self.my_coords = TreeCoordinate::root_with_meta(self.my_node_addr, seq, ts); + self.root = self.my_node_addr; + return; + } // Our coords = [self_entry] ++ parent_coords entries let self_entry = CoordEntry::new( self.my_node_addr, @@ -196,6 +214,33 @@ impl TreeState { } } + /// Smallest root_id visible across known peers. + pub fn smallest_visible_root(&self) -> Option { + self.peer_ancestry.values().map(|c| *c.root_id()).min() + } + + /// Whether this node should be the tree root: either there are no peers, + /// or our NodeAddr is `<=` every visible root. + pub fn should_be_root(&self) -> bool { + match self.smallest_visible_root() { + Some(sr) => self.my_node_addr <= sr, + None => true, + } + } + + /// Promote self to root with an incremented sequence number. + /// + /// Caller must `sign_declaration` afterwards before sending the result. + pub fn become_root(&mut self) { + let new_seq = self.my_declaration.sequence() + 1; + let timestamp = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0); + self.my_declaration = ParentDeclaration::self_root(self.my_node_addr, new_seq, timestamp); + self.recompute_coords(); + } + /// Calculate tree distance to a peer. pub fn distance_to_peer(&self, peer_id: &NodeAddr) -> Option { self.peer_ancestry @@ -324,8 +369,14 @@ impl TreeState { let smallest_root = smallest_root?; - // If we are the smallest node in the network, stay root - if self.my_node_addr <= smallest_root && self.is_root() { + // If our own NodeAddr is smaller than (or equal to) the smallest visible + // root, we are the network's smallest node and must be root. Returning + // `None` lets the caller promote us via `become_root` / `should_be_root`. + // Picking any peer here would produce an invalid path, since prepending + // `self` to that peer's ancestry would put `self` at depth 0 and the + // peer's larger root at the tail — violating "advertised root = min path + // entry" and getting rejected by recipients' `validate_semantics`. + if self.my_node_addr <= smallest_root { return None; } diff --git a/src/tree/tests.rs b/src/tree/tests.rs index 2ebb4dc..9c3cde1 100644 --- a/src/tree/tests.rs +++ b/src/tree/tests.rs @@ -568,6 +568,89 @@ fn test_handle_parent_lost_finds_alternative() { assert!(!state.is_root()); } +#[test] +fn test_handle_parent_lost_becomes_root_when_self_smaller_than_remaining() { + // Regression: self (NodeAddr 1) had peer 0 as parent. Peer 0 disappears, + // leaving only peers with bigger NodeAddrs (and bigger roots). The old + // evaluate_parent() picked one of them — recompute_coords() then + // produced [self, peer, ..., peer_root] where last (peer_root) > min + // (self), an ancestry that recipients reject as + // "advertised root X is not the minimum path entry Y". This is the + // bug seen in production where ubuntu-dev (3847a4..) advertised itself + // as root while its path still contained mac (312c79..). + let my_node = make_node_addr(1); // our addr is the smallest + let mut state = TreeState::new(my_node); + + let smaller = make_node_addr(0); + let bigger1 = make_node_addr(2); + let bigger2 = make_node_addr(3); + + // Initially: peer 0 (smaller) is our parent. + state.update_peer( + ParentDeclaration::self_root(smaller, 1, 1000), + make_coords(&[0]), + ); + // Bigger peers exist, both rooted at themselves (no smaller node visible + // through them). + state.update_peer( + ParentDeclaration::self_root(bigger1, 1, 1000), + make_coords(&[2]), + ); + state.update_peer( + ParentDeclaration::self_root(bigger2, 1, 1000), + make_coords(&[3]), + ); + + state.set_parent(smaller, 2, 2000); + state.recompute_coords(); + assert_eq!(state.my_coords().entries().len(), 2); + assert_eq!(state.root(), &smaller); + + // Smaller peer disconnects. + state.remove_peer(&smaller); + let changed = state.handle_parent_lost(&HashMap::new()); + assert!(changed); + + // Must become root (we're the smallest visible), NOT pick bigger1/bigger2. + assert!( + state.is_root(), + "must self-root when no smaller peer remains" + ); + assert_eq!(state.root(), &my_node); + assert_eq!(state.my_coords().entries().len(), 1); + + // The resulting ancestry must be valid: last == min. + let entries = state.my_coords().entries(); + let min = entries.iter().map(|e| e.node_addr).min().unwrap(); + assert_eq!(*state.my_coords().root_id(), min); +} + +#[test] +fn test_recompute_coords_demotes_when_self_smaller_than_parent_root() { + // Defensive: even if set_parent is called with a parent whose root is + // bigger than us (e.g., a stale evaluate_parent decision in some legacy + // path), recompute_coords must produce a valid ancestry by demoting to + // self-root rather than emit [self, peer, peer_root] with last > min. + let my_node = make_node_addr(5); + let mut state = TreeState::new(my_node); + + let bigger_peer = make_node_addr(7); + state.update_peer( + ParentDeclaration::self_root(bigger_peer, 1, 1000), + make_coords(&[7]), + ); + + state.set_parent(bigger_peer, 2, 2000); + state.recompute_coords(); + + assert!(state.is_root(), "recompute_coords demoted to self-root"); + assert_eq!(state.root(), &my_node); + assert_eq!(state.my_coords().entries().len(), 1); + let entries = state.my_coords().entries(); + let min = entries.iter().map(|e| e.node_addr).min().unwrap(); + assert_eq!(*state.my_coords().root_id(), min); +} + #[test] fn test_handle_parent_lost_becomes_root() { let my_node = make_node_addr(5); From 6ce1406664678c2b458afc2934e4b41353b615ac Mon Sep 17 00:00:00 2001 From: Martti Malmi Date: Fri, 8 May 2026 21:46:14 +0300 Subject: [PATCH 6/9] Refetch overlay advert before every retry, not only on NoTransportForType MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The previous fix (6ebca3e) only refetched the advert when retry returned NodeError::NoTransportForType (cache returned no addresses at all). But the much more common stale-cache failure mode is: cache returns an endpoint that LOOKS valid (the address it had last week, before the peer's NAT rebound), the dial succeeds at the IP layer, the handshake times out, MMP fires, schedule_reconnect adds the entry back to retry_pending, next retry hits the same cached endpoint, dials it again, times out again. Loop forever — no NoTransportForType ever fires because the cache has data, just dead data. Move refetch_advert_for_stale_check to before each retry attempt unconditionally. Cheap (one Filter query against advert_relays with a 2s timeout, bounded by the retry backoff cadence), and replaces the cache only if the relay has a newer advert or evicts if the relay has nothing. Keeps the retry loop pinned to relay ground truth instead of whatever the cache happened to learn at startup. Co-Authored-By: Claude Opus 4.7 (1M context) --- src/node/retry.rs | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/src/node/retry.rs b/src/node/retry.rs index 5b25df2..0eb371d 100644 --- a/src/node/retry.rs +++ b/src/node/retry.rs @@ -259,6 +259,25 @@ impl Node { let peer_config = state.peer_config.clone(); + // Refresh the peer's overlay advert before retrying. The cache is + // read-only on hit (see fetch_advert), so every retry without a + // refetch dials the same cached endpoint — and the most common + // reason a peer ended up in retry_pending is that the cached + // endpoint just stopped working (NAT rebind, port change, peer + // restart on a different port). Without this refresh the retry + // loop dials the same dead address forever. + // + // refetch_advert_for_stale_check uses the relay's advert as + // ground truth: replaces the cache if there's a newer one, + // evicts if the relay has nothing, otherwise leaves it. Cheap + // (one Filter fetch with 2s timeout) and bounded by the retry + // backoff cadence. + if let Some(bootstrap) = self.nostr_discovery.clone() { + let _ = bootstrap + .refetch_advert_for_stale_check(&peer_config.npub) + .await; + } + match self.initiate_peer_connection(&peer_config).await { Ok(()) => { // Push retry_after_ms past the handshake timeout window so From 64cc30df1214c4062550107d47d29e746c0bc3f6 Mon Sep 17 00:00:00 2001 From: Martti Malmi Date: Fri, 8 May 2026 19:53:53 +0300 Subject: [PATCH 7/9] Schedule retry on startup peer-init failure MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When initiate_peer_connections() runs at boot, address resolution can fail for an entire peer (no operational transport for the configured transport types, all addresses unreachable, NAT rebind invalidated cached endpoints, etc.). Before this change the failure was logged and silently forgotten — the peer entry stayed in a dead state forever, accepting incoming pings but unable to answer them, until the daemon was manually restarted. The retry plumbing (schedule_retry / process_pending_retries with exponential backoff) already exists and is wired into the post-handshake failure paths (BootstrapEvent::Failed, MMP dead-link timeout, handshake timeout). The startup loop just wasn't calling it. Mirror the BootstrapEvent::Failed path: on a startup peer-init error, parse the peer's npub and call schedule_retry so the peer recovers without operator intervention. Includes a regression test that asserts retry_pending is populated when initiate_peer_connections() fails for a peer with no operational transport. Co-Authored-By: Claude Opus 4.7 (1M context) --- src/node/lifecycle.rs | 6 ++++++ src/node/tests/unit.rs | 32 ++++++++++++++++++++++++++++++++ 2 files changed, 38 insertions(+) diff --git a/src/node/lifecycle.rs b/src/node/lifecycle.rs index 5b2435f..d4ec4f5 100644 --- a/src/node/lifecycle.rs +++ b/src/node/lifecycle.rs @@ -73,6 +73,12 @@ impl Node { error = %e, "Failed to initiate peer connection" ); + // Schedule a retry so transient address-resolution failures + // (e.g. cached endpoints stale, NAT rebinds, all addresses + // currently unreachable) recover without a daemon restart. + if let Ok(peer_identity) = PeerIdentity::from_npub(&peer_config.npub) { + self.schedule_retry(*peer_identity.node_addr(), Self::now_ms()); + } } } } diff --git a/src/node/tests/unit.rs b/src/node/tests/unit.rs index 43d5944..03f9558 100644 --- a/src/node/tests/unit.rs +++ b/src/node/tests/unit.rs @@ -952,6 +952,38 @@ fn test_promote_clears_retry_pending() { ); } +/// Initial peer-init failure at startup must enqueue a retry. Otherwise a peer +/// whose addresses cannot be dialed at boot (no operational transport for the +/// configured transport types, all addresses unreachable, NAT rebind, etc.) +/// stays dead forever — pings arrive but cannot be answered until the daemon +/// is manually restarted. +#[tokio::test] +async fn test_initiate_peer_connections_schedules_retry_on_no_transport() { + let peer_identity = Identity::generate(); + let peer_npub = peer_identity.npub(); + let peer_node_addr = *PeerIdentity::from_npub(&peer_npub).unwrap().node_addr(); + + let mut config = Config::new(); + // udp address but no UDP transport registered on the node — every dial + // attempt resolves to NodeError::NoTransportForType. + config.peers.push(crate::config::PeerConfig::new( + peer_npub, + "udp", + "10.0.0.2:2121", + )); + + let mut node = Node::new(config).unwrap(); + assert!(node.retry_pending.is_empty()); + + node.initiate_peer_connections().await; + + assert!( + node.retry_pending.contains_key(&peer_node_addr), + "startup peer-init failure must enqueue a retry so the peer can recover \ + without a daemon restart" + ); +} + // ============================================================================ // transport_mtu() — ISSUE-2026-0011 regression coverage // ============================================================================ From 2d18d019d6322e9f0ef7c97e6e7dc87b42854c7e Mon Sep 17 00:00:00 2001 From: Martti Malmi Date: Fri, 8 May 2026 21:03:51 +0300 Subject: [PATCH 8/9] Evict stale overlay advert when retry hits NoTransportForType MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The advert cache inside fetch_advert is read-only on hit — once a peer's overlay advert is cached, every subsequent lookup returns the same endpoints regardless of whether they still work. So when a peer rebinds its NAT (or its STUN-discovered port flaps), connection retries to that peer dial the same dead address forever, even with exponential backoff firing at the right cadence. Observed in deployment: macOS daemon's view of a Linux peer would "regress" — peer marked rch=False after a brief link-dead window, then hours of "Retry connection initiation failed: no operational transport for any of 's addresses" with no recovery. Manual pause+resume of the daemon (which restarts the FIPS endpoint and forces fresh advert fetches) was the only way out. When initiate_peer_connection / a retry tick returns NodeError::NoTransportForType, fire-and-forget refetch_advert_for_stale_check on the peer's npub. This re-fetches kind 37195 from advert_relays; if the relay has a newer advert it replaces the cached entry, if it has nothing it evicts the cached entry. Either way the next retry tick goes to fresh data instead of looping on the same dead endpoint. Mirrors the existing stale-advert sweep that runs from the BootstrapEvent::Failed (NAT-traversal-streak) path, but covers the direct-UDP-retry path which never crosses that streak threshold. Co-Authored-By: Claude Opus 4.7 (1M context) --- src/node/lifecycle.rs | 13 +++++++++++++ src/node/retry.rs | 16 +++++++++++++++- 2 files changed, 28 insertions(+), 1 deletion(-) diff --git a/src/node/lifecycle.rs b/src/node/lifecycle.rs index d4ec4f5..4e92418 100644 --- a/src/node/lifecycle.rs +++ b/src/node/lifecycle.rs @@ -79,6 +79,19 @@ impl Node { if let Ok(peer_identity) = PeerIdentity::from_npub(&peer_config.npub) { self.schedule_retry(*peer_identity.node_addr(), Self::now_ms()); } + // No-transport failures most often mean the cached overlay + // advert is pointing at a dead post-NAT-rebind address. The + // advert cache is read-only inside fetch_advert, so retries + // would loop on the same dead address until expiry. Force a + // re-fetch so the next retry tick picks up fresh endpoints. + if matches!(e, crate::node::NodeError::NoTransportForType(_)) + && let Some(bootstrap) = self.nostr_discovery.clone() + { + let npub = peer_config.npub.clone(); + tokio::spawn(async move { + let _ = bootstrap.refetch_advert_for_stale_check(&npub).await; + }); + } } } } diff --git a/src/node/retry.rs b/src/node/retry.rs index 0eb371d..8ddf6dc 100644 --- a/src/node/retry.rs +++ b/src/node/retry.rs @@ -4,7 +4,7 @@ //! automatically retry with exponential backoff. Retry state lives on Node //! (not PeerConnection) because each retry creates a fresh connection. -use super::Node; +use super::{Node, NodeError}; use crate::PeerIdentity; use crate::config::PeerConfig; use crate::identity::NodeAddr; @@ -301,6 +301,20 @@ impl Node { error = %e, "Retry connection initiation failed" ); + // No-transport failures usually mean the cached overlay + // advert is stale (peer rebound NAT, switched relay, etc.). + // The advert cache is read-only inside fetch_advert, so + // every retry returns the same dead address until the + // entry expires. Force a re-fetch so the next retry tick + // picks up fresh endpoints. + if matches!(e, NodeError::NoTransportForType(_)) + && let Some(bootstrap) = self.nostr_discovery.clone() + { + let npub = peer_config.npub.clone(); + tokio::spawn(async move { + let _ = bootstrap.refetch_advert_for_stale_check(&npub).await; + }); + } // Immediate failure counts as an attempt — schedule next retry // (reconnect flag is preserved on existing retry_pending entry) self.schedule_retry(node_addr, now_ms); From 0cc3de3daa450ddcca31061ee0abca5244d4dd32 Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Sat, 9 May 2026 22:37:25 +0000 Subject: [PATCH 9/9] changelog: update [Unreleased] Three Changed entries for the rx-path performance work (Linux UDP recvmmsg batched receive, run_rx_loop drain batching, and eager pubkey_full precompute on PeerIdentity construction) and five Fixed entries: adopted NAT-traversed UDP transports inheriting the primary listener's MTU and buffer config, TreeAnnounce ancestry on self-root transitions, unconditional overlay-advert refetch before each retry, stale overlay-advert eviction on NoTransportForType, and scheduled retry on startup peer-init failure. Pure CHANGELOG addition (+117 lines, no edits to existing entries). Bullets are wrapped at 80 columns and attribute external contributions to the originating PR and author. --- CHANGELOG.md | 117 +++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 117 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 31390fd..da671be 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -246,6 +246,45 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- Linux UDP receive path uses `recvmmsg(2)` with a 32-packet batch + in place of single-packet `recvmsg(2)`. A single `readable()` + wakeup drains up to 32 datagrams in one syscall before yielding + back to the reactor, eliminating the per-packet scheduler-hop + + futex cost that previously capped inbound rate at one event per + scheduler quantum independent of CPU. `SO_RXQ_OVFL` is sampled + once per batch from the cmsg chain of `msgs[0]` and surfaced + through `AsyncUdpSocket::recv_batch` so the 1Hz + `sample_transport_congestion()` detector continues to feed the + per-transport `dropping` flag. macOS / Windows fall through to + the per-packet path; `recvmmsg` is Linux-specific + ([#81](https://github.com/jmcorgan/fips/pull/81), + [@mmalmi](https://github.com/mmalmi)) +- `Node::run_rx_loop` drains up to 256 additional ready items via + `try_recv()` after each `tokio::select!` await fires on + `packet_rx` / `tun_outbound_rx`, in a tight inner loop before + yielding. Previously the select cost a full scheduler hop + + futex per packet, capping throughput at one event per scheduler + quantum with the worker near-idle. `biased` ordering keeps + data-plane branches priority over tick / control / DNS under + sustained load; the 256 cap is empirically tuned to keep the + worker on a busy stream between yield points (≈ 400 KB of + contiguous traffic) while still bounding the inner loop so a + flood on one branch can't starve the periodic tick or control + socket. Pairs with the UDP `recvmmsg` change above + ([#81](https://github.com/jmcorgan/fips/pull/81), + [@mmalmi](https://github.com/mmalmi)) +- `PeerIdentity::pubkey_full()` now precomputes the parity-aware + full public key at construction in `from_pubkey`. Previously the + method fell through to a secp256k1 EC point parse (`fe_sqrt` + + `fe_mul` + `ge_set_xo_var`) on every call when the full key + wasn't passed at construction (i.e. for every peer constructed + from an npub or x-only key) — ~6% of per-packet CPU on the + bulk-data send path for a value that never changed after + construction. The same EC point parse already runs at + construction inside `NodeAddr::from_pubkey`, so the cost is paid + once where it would be paid anyway + ([#81](https://github.com/jmcorgan/fips/pull/81), + [@mmalmi](https://github.com/mmalmi)) - Cargo feature flags `tui`, `ble`, `gateway`, and `nostr-discovery` removed; subsystem inclusion is now driven by platform `cfg` gates so plain `cargo build` compiles everything @@ -330,6 +369,84 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Adopted NAT-traversed UDP transports inherit the primary listener's + MTU and buffer config. `Node::adopt_established_traversal` + constructed the adopted UDP transport with `UdpConfig::default()` + (MTU 1280, default recv/send buffer sizes, default accept/advertise + flags) regardless of the operator's primary `[transports.udp]` + listener. Operators who set the primary MTU higher (e.g. 1500 on + a known-clean LAN path) silently dropped full-sized tunnel + datagrams over the NAT-traversed link with no log explaining why + throughput collapsed. Lookup now tries `transport_name` first (so + multiple named listeners pick up inheritance from the matching + one) and falls back to the unnamed `Single` listener; bind / + external-address fields are cleared since the adopted socket is + already bound. The 1280 default was deliberately the IPv6 minimum + (the only value guaranteed across arbitrary middlebox paths); + with this change, operators who raise the primary MTU accept the + tradeoff that NAT-traversed flows initially attempt the higher + MTU and may black-hole on tighter paths until reactive + `MtuExceeded` recovery kicks in + ([#83](https://github.com/jmcorgan/fips/pull/83), + [@mmalmi](https://github.com/mmalmi)) +- TreeAnnounce ancestry on self-root transitions. When a node had + no smaller-NodeAddr peer to use as a parent, the spanning-tree + state correctly promoted it to root, but the ancestry it + advertised on the next `TreeAnnounce` still referenced its + previous parent's path. Receiving peers rejected the announce + with `invalid ancestry: advertised root X is not the minimum + path entry Y`, blocking mesh transit on any path that needed to + traverse the node. The self-root transition is now detected + explicitly in `TreeState::become_root` and the advertised + ancestry rebuilt to start from self. The MMP receive handler + surfaces the same path so stale ancestry inherited across + reconnect is corrected eagerly rather than waiting for the next + observation tick + ([#82](https://github.com/jmcorgan/fips/pull/82), + [@mmalmi](https://github.com/mmalmi)) +- Auto-connect retry refetches the cached overlay advert + unconditionally before each retry attempt, not only when + `fetch_advert` returns zero endpoints (`NoTransportForType`). + The much more common stale-cache failure was: cache returned an + endpoint that *looked* valid (the address learned before the + peer's NAT rebound), the dial succeeded at the IP layer, the + handshake timed out, MMP fired, the next retry hit the same + cached endpoint, looped forever — no `NoTransportForType` ever + fired because the cache had data, just dead data. Refetch now + runs unconditionally before each retry attempt (one Filter query + against `advert_relays` with a 2s per-attempt timeout, bounded + by the retry backoff cadence). Keeps the retry loop pinned to + relay ground truth instead of whatever the cache happened to + learn at startup + ([#82](https://github.com/jmcorgan/fips/pull/82), + [@mmalmi](https://github.com/mmalmi)) +- Stale overlay-advert eviction on `NoTransportForType`. Mirrors + the existing stale-advert sweep that ran from the + `BootstrapEvent::Failed` (NAT-traversal-streak) path, but covers + the case where `initiate_peer_connection` / a retry tick returns + `NodeError::NoTransportForType` — the cache had no addresses for + the peer at all. A fire-and-forget `refetch_advert_for_stale_check` + against the peer's npub re-fetches kind `37195` from + `advert_relays`; if the relay has a newer advert it replaces the + cached entry, if it has nothing it evicts the entry. Either way + the next retry tick goes to fresh data instead of looping on the + same dead endpoint. Resolves a deployment regression where a + macOS daemon's view of a Linux peer would flap after NAT rebind + with no recovery short of a daemon restart + ([#82](https://github.com/jmcorgan/fips/pull/82), + [@mmalmi](https://github.com/mmalmi)) +- Schedule retry on startup peer-init failure. When + `initiate_peer_connections()` ran at boot, an address-resolution + failure (no operational transport for the configured transport + types, all addresses unreachable, NAT rebind invalidating cached + endpoints) was logged and silently forgotten — the peer entry + stayed in a dead state forever, accepting incoming pings but + unable to answer them, until the daemon was manually restarted. + Now mirrors the `BootstrapEvent::Failed` path: on a startup + peer-init error, parse the peer's npub and call `schedule_retry` + so the peer recovers without operator intervention + ([#82](https://github.com/jmcorgan/fips/pull/82), + [@mmalmi](https://github.com/mmalmi)) - Default control-socket path resolution: daemon and client tools now use a shared resolver, eliminating a divergence where `fipsctl` / `fipstop` could connect to a socket the daemon never bound (notably