Merge branch 'master' into next

# Conflicts:
#	src/node/tests/bootstrap.rs

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

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,
         }

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<crate::NodeAddr> = 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<crate::NodeAddr> = self.peers.keys().copied().collect();
+                self.bloom_state.mark_all_updates_needed(all_peers);
             }
         }
     }

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!(

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,
         );
 

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);

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.<name>] config matching transport_name",
+    );
+
+    for (_, transport) in node.transports.iter_mut() {
+        transport.stop().await.ok();
+    }
+}

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
 // ============================================================================

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<NodeAddr> = 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<NodeAddr> = 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<NodeAddr> = 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<NodeAddr> = self.peers.keys().copied().collect();
             self.bloom_state.mark_all_updates_needed(all_peers);
         }

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<UdpStats>,
 ) {
-    // 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<Vec<u8>> = (0..BATCH).map(|_| vec![0u8; buf_size]).collect();
+        let mut addrs: [Option<std::net::SocketAddr>; BATCH] = std::array::from_fn(|_| None);
+        let mut lens: [usize; BATCH] = [0; BATCH];
 
-                // 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"
+                    );
+                }
+            }
+        }
+    }
 }
 
 // ============================================================================

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<SocketAddr>],
+            lens: &mut [usize],
+        ) -> std::io::Result<(usize, u32)> {
+            let n = bufs.len().min(addrs.len()).min(lens.len()).min(BATCH_SIZE);
+            if n == 0 {
+                return Ok((0, 0));
+            }
+            let fd = self.inner.as_raw_fd();
+
+            // 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::<libc::sockaddr_storage>() 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<AsyncUdpSocket, TransportError> {
             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<SocketAddr>],
+            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`.

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<NodeAddr> {
+        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<usize> {
         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;
         }
 

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);