fips/src/node/dataplane/rx_loop.rs

//! RX event loop and packet dispatch.

use crate::control::{ControlSocket, commands};
use crate::node::reject::{RejectReason, TransportReject};
use crate::node::{Node, NodeError};
use crate::proto::fmp::wire::{
    COMMON_PREFIX_SIZE, CommonPrefix, FMP_VERSION, PHASE_ESTABLISHED, PHASE_MSG1, PHASE_MSG2,
    expected_payload_len,
};
use crate::transport::ReceivedPacket;
use std::time::Duration;
use tracing::{debug, info, warn};

/// Inside the packet_rx burst drain, run a fallback drain every
/// N packets so bounced FMP plaintexts can't sit behind a full
/// 256-packet UDP burst. Used on unix; on Windows the decrypt-worker
/// pool isn't spawned so the fallback channel is always empty —
/// hold the constants at module scope anyway so the burst-loop
/// dispatch in `run_rx_loop` doesn't need a `#[cfg]` on every site.
const FALLBACK_INTERLEAVE_EVERY: usize = 32;
/// How many fallback events to drain per interleave step. Bounded so
/// the inner loop can keep making forward progress on packet_rx.
#[cfg(unix)]
const FALLBACK_INTERLEAVE_BUDGET: usize = 32;

impl Node {
    /// Run the receive event loop.
    ///
    /// Processes packets from all transports, dispatching based on
    /// the phase field in the 4-byte common prefix:
    /// - Phase 0x0: Encrypted frame (session data)
    /// - Phase 0x1: Handshake message 1 (initiator -> responder)
    /// - Phase 0x2: Handshake message 2 (responder -> initiator)
    ///
    /// Also processes outbound IPv6 packets from the TUN reader for session
    /// encapsulation and routing through the mesh.
    ///
    /// Also processes DNS-resolved identities for identity cache population.
    ///
    /// Also runs a periodic tick (1s) to clean up stale handshake connections
    /// that never received a response. This prevents resource leaks when peers
    /// are unreachable.
    ///
    /// This method takes ownership of the packet_rx channel and runs
    /// until the channel is closed (typically when stop() is called).
    pub async fn run_rx_loop(&mut self) -> Result<(), NodeError> {
        // No shutdown observer → today's infinite loop, byte-identical. All
        // existing callers/tests use this; `pending()` never fires, so the
        // shutdown/deadline arms below stay permanently disabled.
        self.run_rx_loop_with_shutdown(std::future::pending()).await
    }

    /// The rx event loop, which serves until `shutdown` fires and then drains
    /// **in place** before returning.
    ///
    /// The channel receivers are moved into this frame's locals and live across
    /// both serve and drain, so — unlike a `select!`-cancelled loop — they are
    /// never destructively dropped mid-flight; they are released only on clean
    /// exit, after which teardown does not need them.
    ///
    /// - While serving (`drain_deadline == None`) the loop is behaviorally
    ///   identical to before: the shutdown arm, the deadline arm, and the
    ///   peers-empty early-exit are all guarded off, so the hot per-packet path
    ///   and the `biased` order of the real arms are unchanged.
    /// - When `shutdown` fires, the loop calls [`Node::enter_drain`] once
    ///   (broadcast Disconnect, gate the reconciler off) and arms the bounded
    ///   deadline, then keeps servicing inbound/tick/peer-removal until all
    ///   peers clear or the deadline elapses, then returns. The caller
    ///   ([`Node::finish_shutdown`]) closes the window and tears down.
    pub async fn run_rx_loop_with_shutdown(
        &mut self,
        shutdown: impl std::future::Future<Output = ()>,
    ) -> Result<(), NodeError> {
        tokio::pin!(shutdown);
        // `None` = serving; `Some(deadline)` = draining (bounded window).
        let mut drain_deadline: Option<tokio::time::Instant> = None;
        let mut packet_rx = self.packet_rx.take().ok_or(NodeError::NotStarted)?;

        // Take the TUN outbound receiver, or create a dummy channel that never
        // produces messages (when TUN is disabled). Holding the sender prevents
        // the channel from closing.
        let (mut tun_outbound_rx, _tun_guard) = match self.supervisor.tun_outbound_rx.take() {
            Some(rx) => (rx, None),
            None => {
                let (tx, rx) = tokio::sync::mpsc::channel(1);
                (rx, Some(tx))
            }
        };

        // Take the DNS identity receiver, or create a dummy channel (when DNS
        // is disabled). Same pattern as TUN outbound.
        let (mut dns_identity_rx, _dns_guard) = match self.supervisor.dns_identity_rx.take() {
            Some(rx) => (rx, None),
            None => {
                let (tx, rx) = tokio::sync::mpsc::channel(1);
                (rx, Some(tx))
            }
        };

        // Take the runtime child-liveness receiver, or a dummy channel (when the
        // node was seeded straight into Running without a start()). Holding the
        // dummy sender in the guard keeps the channel open. Same pattern as TUN
        // outbound / DNS identity.
        let (mut child_exit_rx, _child_exit_guard) = match self.child_exit_rx.take() {
            Some(rx) => (rx, None),
            None => {
                let (tx, rx) = tokio::sync::mpsc::channel(1);
                (rx, Some(tx))
            }
        };

        // Interface-presence receiver, or a dummy channel — same pattern and
        // same reason as the child-liveness receiver above.
        let (mut presence_rx, _presence_guard) = match self.transport_presence_rx.take() {
            Some(rx) => (rx, None),
            None => {
                let (tx, rx) = tokio::sync::mpsc::channel(1);
                (rx, Some(tx))
            }
        };

        let tick_period = Duration::from_secs(self.config().node.tick_interval_secs);
        let mut tick = tokio::time::interval(tick_period);

        // Set up control socket channel
        let (control_tx, mut control_rx) =
            tokio::sync::mpsc::channel::<crate::control::ControlMessage>(32);

        if self.config().node.control.enabled {
            let config = self.config().node.control.clone();
            let tx = control_tx.clone();
            let read_handle = self.control_read_handle();
            tokio::spawn(async move {
                match ControlSocket::bind(&config) {
                    Ok(socket) => {
                        socket.accept_loop(tx, read_handle).await;
                    }
                    Err(e) => {
                        warn!(error = %e, "Failed to bind control socket");
                    }
                }
            });
        }
        // Drop unused sender to avoid keeping channel open if control is disabled
        drop(control_tx);

        // Native datagram API socket. Experimental, off by default, and built
        // on Linux, FreeBSD and macOS: Windows has no way to pass a descriptor
        // between processes at all, which is the mechanism itself. Bound
        // synchronously so a bad path or a socket already in use is reported
        // here, before the node starts serving, rather than at whatever later
        // moment a spawned bind happened to run.
        // Two channels, as the TUN plane has: registrations go one way and are
        // rare, datagrams go the other and arrive in bursts. Keeping them apart
        // lets the data arm drain in batches without a registration waiting
        // behind a burst of traffic.
        let (native_out_tx, mut native_outbound_rx) =
            tokio::sync::mpsc::channel::<crate::native::link::Outbound>(1024);
        let _native_out_guard = native_out_tx.clone();
        let (mut native_rx, _native_guard) = {
            let (tx, rx) = tokio::sync::mpsc::channel::<crate::native::link::NativeMessage>(64);
            // The `cfg` sits on the binding rather than on the arm that drains
            // the receiver, because `tokio::select!` does not accept one. Where
            // there is no listener the channel exists and nothing ever sends,
            // and the guard keeps it open so the arm never sees a closed
            // receiver.
            #[cfg(any(target_os = "linux", target_os = "freebsd", target_os = "macos"))]
            let guard = {
                let mut guard = Some(tx.clone());
                if self.config().node.native_api.enabled {
                    match crate::native::NativeApi::bind(&self.config().node.native_api) {
                        Ok(socket) => {
                            // The accept loop owns a sender, so the dummy guard
                            // is dropped: the channel closes when the last
                            // client task ends, not while one is still serving.
                            guard = None;
                            let npub = self.npub();
                            tokio::spawn(socket.accept_loop(tx, native_out_tx.clone(), npub));
                        }
                        Err(e) => {
                            warn!(error = %e, "Failed to bind native API socket");
                        }
                    }
                }
                guard
            };
            #[cfg(not(any(target_os = "linux", target_os = "freebsd", target_os = "macos")))]
            let guard = Some(tx.clone());
            (rx, guard)
        };

        // Transport-medium change receiver, or a dummy channel when detection
        // is disabled (or the node was seeded straight into Running without a
        // start()). Same guard pattern as TUN outbound and DNS identity: the
        // held sender keeps the channel open so the arm never sees it closed.
        let (mut netmon_rx, _netmon_guard) = match self.supervisor.netmon_rx.take() {
            Some(rx) => (rx, None),
            None => {
                let (tx, rx) = tokio::sync::mpsc::channel(1);
                (rx, Some(tx))
            }
        };

        // Decrypt-worker fallback receiver. The worker pushes each
        // authenticated FMP plaintext here so rx_loop can finish the
        // per-peer side-effects (stats, MMP, ECN, link dispatch).
        // Always declared so the `tokio::select!` arm doesn't need
        // a `cfg` (which the macro doesn't support); on Windows the
        // channel just never sees events.
        let (mut decrypt_fallback_rx, _decrypt_fallback_guard) = {
            #[cfg(unix)]
            {
                match self.decrypt_fallback_rx.take() {
                    Some(rx) => (rx, None),
                    None => {
                        let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
                        (rx, Some(tx))
                    }
                }
            }
            #[cfg(not(unix))]
            {
                // On non-unix nothing ever sends, but the macro arm
                // still needs an existing rx. Keep the sender alive to
                // avoid the channel closing into an Err loop.
                let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<()>();
                (rx, Some(tx))
            }
        };

        info!("RX event loop started");
        // Optional per-stage perf profiler (FIPS_PERF=1). No-op otherwise.
        crate::perf_profile::maybe_spawn_reporter();

        loop {
            // Bounded drain mode: break as soon as all peers have cleared. In
            // normal mode (`None`) this short-circuits before touching
            // `self.peers`, so the loop is byte-identical.
            if drain_deadline.is_some() && self.peers.is_empty() {
                info!("Drain complete: all peers cleared, ending drain loop");
                break;
            }
            tokio::select! {
                biased;
                // Decrypt-worker fallback drains FIRST. Under sustained
                // inbound bursts the packet_rx drain (up to 256 packets)
                // can starve fallback work for tens of ms — TCP doesn't
                // tolerate that (late ACKs → dup-ACK fast retransmits →
                // cwnd collapse). Promoting fallback gives the kernel's
                // TCP machinery a fair chance to ACK in time.
                Some(event) = decrypt_fallback_rx.recv() => {
                    #[cfg(unix)]
                    {
                        self.process_decrypt_worker_event(event).await;
                        let mut drained = 0;
                        while drained < 255 {
                            match decrypt_fallback_rx.try_recv() {
                                Ok(ev) => {
                                    self.process_decrypt_worker_event(ev).await;
                                    drained += 1;
                                }
                                Err(_) => break,
                            }
                        }
                    }
                    #[cfg(not(unix))]
                    let _ = event;
                }
                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.
                    //
                    // **Interleave fallback drain** every N packets so
                    // bounced FMP plaintexts (heartbeats, post-FMP-
                    // decrypt forwarding payloads, control frames) don't
                    // sit in the fallback queue for a full 256-packet
                    // burst. Even with the priority-first ordering of
                    // the outer select!, once we're inside this inner
                    // loop only this interleave can free queued
                    // fallbacks. On multihop forwarding paths this is
                    // the difference between back-to-back encrypt-
                    // worker dispatches happening promptly vs piling up
                    // behind the rx burst.
                    let mut drained: usize = 1; // count the packet processed above
                    while drained < 256 {
                        if drained.is_multiple_of(FALLBACK_INTERLEAVE_EVERY) {
                            #[cfg(unix)]
                            {
                                let mut fb_drained = 0;
                                while fb_drained < FALLBACK_INTERLEAVE_BUDGET {
                                    match decrypt_fallback_rx.try_recv() {
                                        Ok(ev) => {
                                            self.process_decrypt_worker_event(ev).await;
                                            fb_drained += 1;
                                        }
                                        Err(_) => break,
                                    }
                                }
                            }
                        }
                        match packet_rx.try_recv() {
                            Ok(p) => {
                                self.process_packet(p).await;
                                drained += 1;
                            }
                            Err(_) => break,
                        }
                    }
                    // Trailing fallback drain so the last bounced
                    // packets of the burst aren't held up by the
                    // next select! iteration.
                    #[cfg(unix)]
                    {
                        let mut fb_drained = 0;
                        while fb_drained < 256 {
                            match decrypt_fallback_rx.try_recv() {
                                Ok(ev) => {
                                    self.process_decrypt_worker_event(ev).await;
                                    fb_drained += 1;
                                }
                                Err(_) => break,
                            }
                        }
                    }
                }
                // Runtime child-liveness. Placed AFTER `packet_rx` so the hot
                // inbound path keeps its `biased` priority. A directly-observable
                // child (TUN threads, DNS/mDNS/Nostr) exited on its own; feed the
                // FSM, which republishes health (Degraded here — a Running node
                // always has ≥1 transport up). `on_child_exited` only ever emits
                // `PublishState`; other variants are ignored defensively.
                maybe_child = child_exit_rx.recv() => {
                    if let Some(child) = maybe_child {
                        // Drop anything the dead child published for embedders
                        // (e.g. the DNS responder's bound address) before
                        // republishing health, so nothing outside the node can
                        // observe an address the listener no longer answers on.
                        self.retract_child_publications(child);
                        self.step_child_exited(child);
                        // A transport child exiting leaves the bound set, so
                        // it can be the one that was holding the node's egress
                        // MTU down. `is_bound()` is `is_operational()` plus the
                        // presence refinement, and this moves the first half.
                        self.refresh_tun_mss_ceiling();
                    }
                }
                // Interface presence. An interface-bound transport's binder
                // reports attach and detach; the FSM folds it into health.
                // Unlike `ChildExited` this is reversible in both directions —
                // the interface coming back republishes `Running` — which is
                // the whole point of `Degraded` being a level rather than a
                // latch.
                maybe_presence = presence_rx.recv() => {
                    if let Some(edge) = maybe_presence {
                        // Health is policy-filtered; the MTU floor below is
                        // not. An `optional` interface's absence is normal and
                        // must not move the node off `Full`, but it changes
                        // the bound set all the same.
                        if edge.health_relevant {
                            let child = crate::node::lifecycle::supervisor::Child::Transport(
                                edge.transport_id,
                            );
                            let event = if edge.present {
                                crate::node::lifecycle::supervisor::Event::ChildPresent { child }
                            } else {
                                crate::node::lifecycle::supervisor::Event::ChildAbsent { child }
                            };
                            let actions = self.supervisor.fsm.step(event);
                            for action in actions {
                                if let crate::node::lifecycle::supervisor::Action::PublishState(
                                    ns,
                                ) = action
                                {
                                    self.supervisor.state = ns;
                                }
                            }
                        }
                        // The bound set just changed, so the node's egress MTU
                        // floor may have. Both directions: an interface that
                        // binds can be the narrow one, and one that detaches
                        // can be the reason the clamp was tight.
                        self.refresh_tun_mss_ceiling();

                        // A peer reachable only through an interface that has
                        // gone is not reachable. Withdraw it now rather than
                        // leaving the liveness reaper to notice up to
                        // `link_dead_timeout_secs` later, during which this
                        // node both drops transit traffic in silence and keeps
                        // advertising reachability it does not have.
                        //
                        // Not policy-filtered: whether an interface's absence
                        // is normal is a statement about node *health*, not
                        // about whether the routes over it still work.
                        if !edge.present {
                            let reaped =
                                self.reap_peers_on_transport(edge.transport_id).await;
                            if reaped > 0 {
                                info!(
                                    transport_id = %edge.transport_id,
                                    peers = reaped,
                                    "Withdrew peers whose interface went away"
                                );
                            }
                        }
                    }
                }
                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!(
                        node_addr = %identity.node_addr,
                        "Registering identity from DNS resolution"
                    );
                    self.register_identity(identity.node_addr, identity.pubkey);
                }
                // A transport medium change (WLAN -> LAN -> 5G, a BLE adapter
                // arriving or leaving). Placed after the hot inbound path so
                // the `biased` priority of packet processing is unchanged; the
                // detector coalesces into a single-slot channel, so this arm
                // never sees a burst and needs no drain loop.
                Some(change) = netmon_rx.recv() => {
                    self.handle_net_change(change).await;
                }
                // Native API datagrams a client wrote to its descriptor. Drained
                // in a burst like the TUN arm, for the same reason: one wake-up
                // should clear what a client handed over, not one datagram.
                Some(out) = native_outbound_rx.recv() => {
                    self.handle_native_outbound(out.key, out.peer, out.payload).await;
                    let mut drained = 0;
                    while drained < 256 {
                        match native_outbound_rx.try_recv() {
                            Ok(next) => {
                                self.handle_native_outbound(next.key, next.peer, next.payload).await;
                                drained += 1;
                            }
                            Err(_) => break,
                        }
                    }
                }
                // Native API registry requests. Placed after the hot inbound
                // path so a burst of client registrations cannot delay packet
                // processing. No `cfg` here: `tokio::select!` does not accept
                // one, so the channel exists on every platform and only the
                // listener that feeds it is gated.
                Some(message) = native_rx.recv() => {
                    self.handle_native(message);
                }
                Some((request, response_tx)) = control_rx.recv() => {
                    // Only mutating COMMAND requests (`connect` / `disconnect`)
                    // reach the rx_loop now. Every pure-read `show_*` query is
                    // served off-loop from the read handle in the control accept
                    // task (`snapshot_dispatch`), so it never round-trips here —
                    // the data-plane dispatch path carries no `show_*` arm. A
                    // `show_*` that somehow arrives (none does) falls through to
                    // `commands::dispatch`, which returns "unknown command".
                    let response = commands::dispatch(
                        self,
                        &request.command,
                        request.params.as_ref(),
                    ).await;
                    let _ = response_tx.send(response);
                }
                deadline = tick.tick() => {
                    // Tick-body instrumentation. The gate is read ONCE per tick
                    // into `instr_on`, which is then passed explicitly to every
                    // `instr_step!` invocation — macro hygiene makes a call-site
                    // local invisible inside the macro body. With the
                    // `profiling` feature off, `gate()` is a `const fn`
                    // returning false and the macro is a pure pass-through, so
                    // the whole arm compiles to the uninstrumented sequence.
                    //
                    // `tick_entry` records how late this entry is against the
                    // deadline the interval scheduled it for. That is the
                    // measurement this instrumentation exists for: the arm is
                    // polled LAST under `biased;`, so the
                    // lateness IS the time it spent waiting behind the packet,
                    // TUN and control arms. `tick()` hands back its scheduled
                    // deadline, so this is a subtraction rather than a model.
                    // The whole-tick span below measures the body alone.
                    let instr_on = crate::instr::gate();
                    crate::instr::tick_entry(instr_on, deadline.into_std(), std::time::Instant::now());
                    instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::WholeTick, {
                        instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::CheckTimeouts,
                        self.check_timeouts().await);
                        // Discard flows the rx_loop announced and a listener's
                        // own task never took. Cheap: it walks only the pending
                        // map, which the backlog bounds.
                        self.native_expire();
                        let now_ms = Self::now_ms();
                        instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::ReloadPeerAcl,
                        self.reload_peer_acl().await);
                        // The host map hot-reloads on the same tick as the ACL. It
                        // is polled separately from `reload_peer_acl` because the
                        // ACL's embedded alias reloader and this snapshot are
                        // distinct resources; the `path_mtu_lookup` cache and the
                        // `nostr_rendezvous` subsystem are deliberately excluded
                        // from `Reloadable` since neither reloads from a backing
                        // file (see `node::reloadable`). The `path_mtu_lookup`
                        // cache is nevertheless swept on this tick, by
                        // `purge_expired_path_mtu` below: that is expiry, not
                        // reload.
                        instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::ReloadHostMap,
                        self.reload_host_map().await);
                        instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::PollPendingConnects,
                        self.poll_pending_connects().await);
                        instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::PollNostrRendezvous,
                        self.poll_nostr_rendezvous().await);
                        instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::PollLanRendezvous,
                        self.poll_lan_rendezvous().await);
                        instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::DrivePeerTimers,
                        self.drive_peer_timers(now_ms).await);
                        instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::ResendPendingRekeys,
                        self.resend_pending_rekeys(now_ms).await);
                        instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::ResendPendingSessionHandshakes,
                        self.resend_pending_session_handshakes(now_ms).await);
                        instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::ResendPendingSessionMsg3,
                        self.resend_pending_session_msg3(now_ms).await);
                        instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::PurgeIdleSessions,
                        self.purge_idle_sessions(now_ms));
                        instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::PurgeExpiredPathMtu,
                        self.purge_expired_path_mtu(now_ms));
                        instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::ProcessPendingRetries,
                        self.process_pending_retries(now_ms).await);
                        instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::CheckTreeState,
                        self.check_tree_state().await);
                        instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::CheckBloomState,
                        self.check_bloom_state().await);
                        instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::ComputeMeshSize,
                        self.compute_mesh_size());
                        instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::RecordStatsHistory,
                        self.record_stats_history());
                        instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::CheckMmpReports,
                        self.check_mmp_reports().await);
                        instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::CheckSessionMmpReports,
                        self.check_session_mmp_reports().await);
                        instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::CheckLinkHeartbeats,
                        self.check_link_heartbeats().await);
                        instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::CheckRekey,
                        self.check_rekey().await);
                        instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::CheckSessionRekey,
                        self.check_session_rekey().await);
                        instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::CheckPendingLookups,
                        self.check_pending_lookups(now_ms).await);
                        // After CheckPendingLookups so a probe's resolve stage
                        // observes this tick's lookup progress rather than the
                        // previous tick's.
                        instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::PollProbes,
                        self.poll_probes().await);
                        instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::PollTransportDiscovery,
                        self.poll_transport_discovery().await);
                        instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::SampleTransportCongestion,
                        self.sample_transport_congestion());
                        #[cfg(any(target_os = "linux", target_os = "macos"))]
                        instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::ActivateConnectedUdpSessions,
                        self.activate_connected_udp_sessions().await);
                        // Crypto worker threads that exited since the last tick.
                        #[cfg(unix)]
                        instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::PollWorkerLiveness,
                        self.poll_worker_liveness());
                        // Debug-build sweep of the peer-lifecycle map invariant
                        // (leaked machines / machine-less legs); two map scans,
                        // compiled out of release builds.
                        #[cfg(debug_assertions)]
                        instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::DebugAssertPeerMapsCoherent,
                        self.debug_assert_peer_maps_coherent());
                    });
                    crate::instr::tick_gauges(instr_on, self.peers.len() as u64);
                }
                // Shutdown signal → enter the bounded drain in place, ONCE.
                // Gated on `is_none()` so it only fires while serving; after
                // entering drain the arm is disabled (the completed signal is
                // never polled again) and the deadline arm below bounds the
                // window. Placed after the real arms so their `biased` priority
                // is unchanged, and inert while serving with `pending()`.
                _ = &mut shutdown, if drain_deadline.is_none() => {
                    self.enter_drain().await;
                    drain_deadline =
                        Some(tokio::time::Instant::now() + self.config().node.drain_timeout());
                }
                // Bounded drain deadline (drain mode only). Placed LAST so the
                // `biased` priority of the normal arms is unchanged, and gated
                // on `is_some()` so in normal mode the branch is disabled — the
                // future is created but never polled and never fires.
                _ = tokio::time::sleep_until(
                    drain_deadline.unwrap_or_else(tokio::time::Instant::now)
                ), if drain_deadline.is_some() => {
                    info!("Drain deadline elapsed, ending drain loop");
                    break;
                }
            }
        }

        info!("RX event loop stopped (channel closed)");
        Ok(())
    }

    /// Process a single received packet.
    ///
    /// Dispatches based on the phase field in the 4-byte common prefix.
    ///
    /// Visible to the rest of `crate::node` so tests can drive a single
    /// packet through the dispatch, the same reach `handle_msg1` and
    /// `handle_msg2` already have.
    pub(in crate::node) async fn process_packet(&mut self, packet: ReceivedPacket) {
        if packet.data.len() < COMMON_PREFIX_SIZE {
            return; // Drop packets too short for common prefix
        }

        let prefix = match CommonPrefix::parse(&packet.data) {
            Some(p) => p,
            None => return, // Malformed prefix
        };

        if prefix.version != FMP_VERSION {
            debug!(
                version = prefix.version,
                transport_id = %packet.transport_id,
                "Unknown FMP version, dropping"
            );

            // If the packet arrived on an adopted Nostr-NAT bootstrap
            // transport, the originating peer is necessarily on a
            // different FMP-protocol version than us — the discovery
            // sweep would otherwise re-traverse them every cycle even
            // though no msg1/msg2 exchange can ever succeed. Bump the
            // discovery-layer cooldown to the long protocol-mismatch
            // window and emit a single WARN per fresh observation.
            if self
                .supervisor
                .nostr_rendezvous
                .is_bootstrap_transport(&packet.transport_id)
                && let Some(npub) = self
                    .supervisor
                    .nostr_rendezvous
                    .bootstrap_transport_npub(&packet.transport_id)
                    .cloned()
                && let Some(handle) = self.nostr_rendezvous_handle()
            {
                let now_ms = Self::now_ms();
                let cooldown_secs = handle.protocol_mismatch_cooldown_secs();
                if handle.record_protocol_mismatch(&npub, now_ms) {
                    warn!(
                        peer_npub = %npub,
                        transport_id = %packet.transport_id,
                        peer_version = prefix.version,
                        our_version = FMP_VERSION,
                        cooldown_secs,
                        "Nostr-discovered peer speaks a different FMP version; suppressing retraversal"
                    );
                }
            }
            return;
        }

        // Drop a frame whose declared payload length disagrees with the
        // frame that arrived, before that field can be used as a parsing
        // input.
        //
        // Every transport's packets converge here, but the two families
        // reach this line differently. TCP, Tor and Nym read their frame
        // boundary out of this same field, so for them the comparison holds
        // by construction and never fires. UDP, Ethernet and BLE deliver one
        // whole frame per packet, where the arrived length is known exactly
        // and nothing compares the two today. A short read on those
        // transports is a truncated frame, which fails the AEAD tag or the
        // exact-size handshake parse already; this changes which reason it
        // is dropped for, not whether it is dropped.
        //
        // A `None` means the phase carries no fixed relationship and the
        // frame is left alone rather than rejected, so an unrecognised phase
        // still reaches the dispatch below and is handled there.
        if let Some(expected) = expected_payload_len(prefix.phase, packet.data.len())
            && prefix.payload_len != expected
        {
            debug!(
                phase = prefix.phase,
                declared = prefix.payload_len,
                expected,
                len = packet.data.len(),
                transport_id = %packet.transport_id,
                "FMP payload_len disagrees with frame length, dropping"
            );
            self.stats_mut()
                .record_reject(RejectReason::Transport(TransportReject::PayloadLenMismatch));
            return;
        }

        match prefix.phase {
            PHASE_ESTABLISHED => {
                self.handle_encrypted_frame(packet).await;
            }
            PHASE_MSG1 => {
                self.handle_msg1(packet).await;
            }
            PHASE_MSG2 => {
                self.handle_msg2(packet).await;
            }
            _ => {
                debug!(
                    phase = prefix.phase,
                    transport_id = %packet.transport_id,
                    "Unknown FMP phase, dropping"
                );
            }
        }
    }
}