diff --git a/CHANGELOG.md b/CHANGELOG.md index 86af880..3373767 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -338,6 +338,45 @@ with v0.2.x peers. ### 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 @@ -422,6 +461,84 @@ with v0.2.x peers. ### 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 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, } diff --git a/src/node/handlers/mmp.rs b/src/node/handlers/mmp.rs index 255ad08..ce28da4 100644 --- a/src/node/handlers/mmp.rs +++ b/src/node/handlers/mmp.rs @@ -150,11 +150,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; @@ -174,6 +174,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/handlers/rx_loop.rs b/src/node/handlers/rx_loop.rs index 1f53912..545a8f6 100644 --- a/src/node/handlers/rx_loop.rs +++ b/src/node/handlers/rx_loop.rs @@ -84,14 +84,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!( diff --git a/src/node/lifecycle.rs b/src/node/lifecycle.rs index b2a65d8..729cad7 100644 --- a/src/node/lifecycle.rs +++ b/src/node/lifecycle.rs @@ -73,6 +73,25 @@ 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()); + } + // 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; + }); + } } } } @@ -1867,17 +1886,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/retry.rs b/src/node/retry.rs index 5b25df2..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; @@ -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 @@ -282,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); diff --git a/src/node/tests/bootstrap.rs b/src/node/tests/bootstrap.rs index a969566..a051cea 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, PHASE_MSG3}; use crate::transport::udp::UdpTransport; use crate::utils::index::IndexAllocator; +use std::collections::HashMap; use tokio::time::{Duration, timeout, timeout_at}; #[tokio::test] @@ -284,3 +285,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(); + } +} diff --git a/src/node/tests/unit.rs b/src/node/tests/unit.rs index dcdf792..3f46e54 100644 --- a/src/node/tests/unit.rs +++ b/src/node/tests/unit.rs @@ -955,6 +955,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 // ============================================================================ diff --git a/src/node/tree.rs b/src/node/tree.rs index 0cfd10b..4e166b1 100644 --- a/src/node/tree.rs +++ b/src/node/tree.rs @@ -247,11 +247,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(); @@ -275,6 +278,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 { @@ -328,11 +350,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(); @@ -418,11 +440,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(); @@ -444,6 +466,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/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`. diff --git a/src/tree/state.rs b/src/tree/state.rs index 586394a..b0672a4 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 @@ -341,8 +386,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 a2c46b9..21fddff 100644 --- a/src/tree/tests.rs +++ b/src/tree/tests.rs @@ -580,6 +580,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);