fips/src/node/handlers/mmp.rs

//! MMP report dispatch, periodic report generation, and operator logging.
//!
//! Handles incoming SenderReport / ReceiverReport messages, drives
//! periodic report generation on the tick timer, and emits periodic
//! and teardown metric logs.

use crate::NodeAddr;
use crate::node::Node;
use crate::node::dataplane::PeerActionCtx;
use crate::node::reject::{MmpReject, RejectReason, TreeReject};
use crate::node::tree::sign_declaration;
use crate::peer::machine::PeerEvent;
use crate::proto::link::LinkMessageType;
use crate::proto::mmp::{
    LinkReportKind, LinkReportSnapshot, MmpAction, PeerLivenessSnapshot, ReceiverReport, RrLog,
    SenderReport,
};
use crate::proto::stp::ParentEval;
use crate::transport::{TransportAddr, TransportId};
use std::time::{Duration, Instant};
use tracing::{debug, info, trace, warn};

/// How long a peer whose heartbeat send *failed* waits before the next attempt.
///
/// Applies to the failure path only. Gating a healthy peer on it too would
/// floor `node.heartbeat_interval_secs` at this value without validating or
/// reporting it, which is a configured knob quietly not doing what it says.
///
/// Short against `heartbeat_interval_secs`, because a failed heartbeat means
/// the peer has heard nothing and the point is to recover well inside
/// `link_dead_timeout_secs` rather than after another full interval. Not
/// shorter still, because the send behind it awaits an unbounded `write_all`
/// on a connection-oriented transport, on the rx loop; retrying that every
/// tick would make a stranded stream a stalled node. Once that write is
/// bounded this can come down to the tick.
const HEARTBEAT_RETRY_INTERVAL: Duration = Duration::from_secs(2);

/// Decide whether a peer is due a heartbeat, from the two timestamps it keeps.
///
/// Two gates rather than one. `sent` is when a heartbeat last *landed*, and it
/// alone paces a healthy peer. `attempt` is when one was last *tried*, and it
/// gates only a peer whose last try failed, holding the retry off for
/// [`HEARTBEAT_RETRY_INTERVAL`] so a peer whose send keeps failing is not
/// retried on every tick.
///
/// **The retry gate is deliberately not consulted on the healthy path.** On a
/// peer whose last send succeeded the two timestamps are equal, so gating there
/// would clamp a configured `heartbeat_interval_secs` up to the retry interval,
/// and that setting has no validation floor.
fn heartbeat_due(
    sent: Option<Instant>,
    attempt: Option<Instant>,
    now: Instant,
    interval: Duration,
) -> bool {
    let landed_due = match sent {
        None => true,
        Some(last) => now.duration_since(last) >= interval,
    };

    // An attempt later than the last success is one that failed, and an attempt
    // with no success behind it is the same thing on a peer never reached.
    let retry_due = match (attempt, sent) {
        (Some(last), Some(landed)) if last > landed => {
            now.duration_since(last) >= HEARTBEAT_RETRY_INTERVAL
        }
        (Some(last), None) => now.duration_since(last) >= HEARTBEAT_RETRY_INTERVAL,
        _ => true,
    };

    landed_due && retry_due
}

/// Emit the operator `trace!` point for a processed ReceiverReport outcome.
///
/// These log points used to live inside `MmpMetrics::process_receiver_report`;
/// the sans-IO migration returns the outcome as an [`RrLog`] and re-emits it
/// here, shell-side, preserving the original field set, content, and (relative
/// to the surrounding handler logs) ordering. The original traces carried no
/// peer identifier, so none is added here.
pub(super) fn log_rr_outcome(rr: &ReceiverReport, our_timestamp_ms: u32, log: RrLog) {
    match log {
        RrLog::Stale {
            prev_highest,
            prev_packets,
            prev_bytes,
        } => trace!(
            highest_counter = rr.highest_counter,
            prev_highest_counter = prev_highest,
            cumulative_packets_recv = rr.cumulative_packets_recv,
            prev_cumulative_packets_recv = prev_packets,
            cumulative_bytes_recv = rr.cumulative_bytes_recv,
            prev_cumulative_bytes_recv = prev_bytes,
            "Ignoring stale MMP ReceiverReport"
        ),
        RrLog::RttSample { rtt_ms, srtt_ms } => trace!(
            our_ts = our_timestamp_ms,
            echo = rr.timestamp_echo,
            dwell = u32::from(rr.dwell_time),
            rtt_ms = rtt_ms,
            srtt_ms = srtt_ms,
            "RTT sample from timestamp echo"
        ),
        RrLog::InvalidRtt => trace!(
            our_ts = our_timestamp_ms,
            echo = rr.timestamp_echo,
            dwell = u32::from(rr.dwell_time),
            "Ignoring invalid MMP RTT sample"
        ),
        RrLog::None => {}
    }
}

/// Format bytes/sec as human-readable throughput.
pub(in crate::node) fn format_throughput(bps: f64) -> String {
    if bps == 0.0 {
        "n/a".to_string()
    } else if bps >= 1_000_000.0 {
        format!("{:.1}MB/s", bps / 1_000_000.0)
    } else if bps >= 1_000.0 {
        format!("{:.1}KB/s", bps / 1_000.0)
    } else {
        format!("{:.0}B/s", bps)
    }
}

impl Node {
    /// Handle an incoming SenderReport from a peer.
    ///
    /// The peer is telling us about what they sent. We feed this to our
    /// receiver state for cross-reference (not currently used for metrics,
    /// but stored for future use).
    pub(in crate::node) fn handle_sender_report(&mut self, from: &NodeAddr, payload: &[u8]) {
        let sr = match SenderReport::decode(payload) {
            Ok(sr) => sr,
            Err(e) => {
                self.stats_mut()
                    .record_reject(RejectReason::Mmp(MmpReject::DecodeError));
                debug!(from = %self.peer_display_name(from), error = %e, "Malformed SenderReport");
                return;
            }
        };

        let peer = match self.peers.get_mut(from) {
            Some(p) => p,
            None => {
                self.stats_mut()
                    .record_reject(RejectReason::Mmp(MmpReject::UnknownPeer));
                debug!(from = %self.peer_display_name(from), "SenderReport from unknown peer");
                return;
            }
        };

        if peer.mmp().is_none() {
            return;
        }

        trace!(
            from = %self.peer_display_name(from),
            cum_pkts = sr.cumulative_packets_sent,
            interval_pkts = sr.interval_packets_sent,
            interval_bytes = sr.interval_bytes_sent,
            "Received SenderReport"
        );

        // Store sender's report in receiver state for cross-reference.
        // Currently informational; the receiver already tracks its own
        // counters and echoes timestamps from data frames.
    }

    /// Handle an incoming ReceiverReport from a peer.
    ///
    /// The peer is telling us about what they received from us. We feed
    /// this to our metrics to compute RTT, loss rate, and trend indicators.
    pub(in crate::node) async fn handle_receiver_report(
        &mut self,
        from: &NodeAddr,
        payload: &[u8],
    ) {
        let rr = match ReceiverReport::decode(payload) {
            Ok(rr) => rr,
            Err(e) => {
                self.stats_mut()
                    .record_reject(RejectReason::Mmp(MmpReject::DecodeError));
                debug!(from = %self.peer_display_name(from), error = %e, "Malformed ReceiverReport");
                return;
            }
        };

        let peer_name = self.peer_display_name(from);

        let peer = match self.peers.get_mut(from) {
            Some(p) => p,
            None => {
                self.stats_mut()
                    .record_reject(RejectReason::Mmp(MmpReject::UnknownPeer));
                debug!(from = %peer_name, "ReceiverReport from unknown peer");
                return;
            }
        };

        // Get session timestamp before taking mutable borrow on MMP
        let our_timestamp_ms = peer.session_elapsed_ms();

        let Some(mmp) = peer.mmp_mut() else {
            return;
        };

        // Process the report: computes RTT from timestamp echo, updates
        // loss rate, goodput rate, jitter trend, and ETX.
        let now_ms = crate::time::mono_ms();
        let (first_rtt, rr_log) =
            mmp.metrics
                .process_receiver_report(&rr, our_timestamp_ms, now_ms);
        // Re-emit the operator trace the core used to log mid-decision.
        log_rr_outcome(&rr, our_timestamp_ms, rr_log);

        // Feed SRTT back to sender/receiver report interval tuning
        if let Some(srtt_ms) = mmp.metrics.srtt_ms() {
            let srtt_us = (srtt_ms * 1000.0) as i64;
            mmp.sender.update_report_interval_from_srtt(srtt_us);
            mmp.receiver.update_report_interval_from_srtt(srtt_us);
        }

        // Update reverse delivery ratio from our own receiver state
        // (what fraction of peer's frames we received), using per-interval deltas.
        let our_recv_packets = mmp.receiver.cumulative_packets_recv();
        let peer_highest = mmp.receiver.highest_counter();
        mmp.metrics
            .update_reverse_delivery(our_recv_packets, peer_highest);

        trace!(
            from = %peer_name,
            rtt_ms = ?mmp.metrics.srtt_ms(),
            loss = format_args!("{:.1}%", mmp.metrics.loss_rate() * 100.0),
            etx = format_args!("{:.2}", mmp.metrics.etx),
            "Processed ReceiverReport"
        );

        // First RTT sample — peer is now eligible for parent selection.
        // Trigger re-evaluation so the node doesn't wait for the next
        // periodic tick or TreeAnnounce.
        if first_rtt {
            let peer_costs: std::collections::BTreeMap<crate::NodeAddr, f64> = self
                .peers
                .iter()
                .filter(|(_, p)| p.has_srtt())
                .map(|(a, p)| (*a, p.link_cost()))
                .collect();
            // Wall-clock seconds for the escaping declaration timestamp;
            // monotonic ms for the flap-dampening / hold-down timers.
            let now_secs = std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map(|d| d.as_secs())
                .unwrap_or(0);
            let mono_now_ms = crate::time::mono_ms();
            let skip = self.non_full_peers();
            // Compute the flap-dampening / hold-down veto at the edge; a mandatory
            // switch bypasses it, a discretionary one is taken only if not suppressed.
            let switch_suppressed = self.tree_state.is_switch_suppressed(mono_now_ms);
            let new_parent = match self.tree_state.evaluate_parent(&peer_costs, &skip) {
                ParentEval::Mandatory(p) => Some(p),
                ParentEval::Discretionary(p) if !switch_suppressed => Some(p),
                ParentEval::Discretionary(_) | ParentEval::None => None,
            };
            if let Some(new_parent) = new_parent {
                let new_seq = self.tree_state.my_declaration().sequence() + 1;
                let flap_dampened =
                    self.tree_state
                        .set_parent(new_parent, new_seq, now_secs, mono_now_ms);
                self.tree_state.recompute_coords();
                // Clone identity once: sign_declaration borrows &mut tree_state while
                // the identity() accessor borrows all of &self, so an owned copy avoids
                // the split-borrow conflict on this infrequent parent-switch path.
                let our_identity = self.identity().clone();
                if let Err(e) =
                    sign_declaration(self.tree_state.my_declaration_mut(), &our_identity)
                {
                    warn!(error = %e, "Failed to sign declaration after first-RTT parent eval");
                    self.metrics()
                        .tree
                        .record_reject(TreeReject::OutboundSignFailed);
                    return;
                }
                // Surgical invalidation — see CoordCache::invalidate_via_node doc.
                self.coord_cache
                    .invalidate_via_node(our_identity.node_addr());
                self.reset_lookup_backoff();
                self.metrics().tree.parent_switches.inc();
                info!(
                    new_parent = %self.peer_display_name(&new_parent),
                    new_seq = new_seq,
                    new_root = %self.tree_state.root(),
                    depth = self.tree_state.my_coords().depth(),
                    trigger = "first-rtt",
                    "Parent switched after first RTT measurement"
                );
                if flap_dampened {
                    self.note_flap("first-rtt");
                }
                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(now_secs);
                // Clone identity once (see the parent-switch branch above for why).
                let our_identity = self.identity().clone();
                if let Err(e) =
                    sign_declaration(self.tree_state.my_declaration_mut(), &our_identity)
                {
                    warn!(error = %e, "Failed to sign self-root declaration after first-RTT");
                    self.metrics()
                        .tree
                        .record_reject(TreeReject::OutboundSignFailed);
                    return;
                }
                // Surgical invalidation — see CoordCache::invalidate_other_roots doc.
                self.coord_cache
                    .invalidate_other_roots(our_identity.node_addr());
                self.reset_lookup_backoff();
                self.metrics().tree.parent_switches.inc();
                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);
            }
        }
    }

    /// Check all peers for pending MMP reports and send them.
    ///
    /// Called from the tick handler. Also emits periodic operator logs.
    pub(in crate::node) async fn check_mmp_reports(&mut self) {
        let now_ms = crate::time::mono_ms();

        // Build one report-gating snapshot per peer, resolving every timing read
        // shell-side into a `bool`. `send_sr`/`send_rr` come from the peer's
        // negotiated profile (whether it provides/wants each report); the core
        // ANDs them into the mode/timing gate. The snapshots own only
        // `NodeAddr`/`MmpMode`/`bool`, so the peer-iteration borrow is released
        // before the pure decision runs and the driving loop mutates the
        // reporting state.
        let snapshots: Vec<LinkReportSnapshot> = self
            .peers
            .iter()
            .filter_map(|(node_addr, peer)| {
                let mmp = peer.mmp()?;
                Some(LinkReportSnapshot {
                    peer: *node_addr,
                    mode: mmp.mode(),
                    send_sr: peer.send_sr(),
                    send_rr: peer.send_rr(),
                    sr_due: mmp.sender.should_send_report(now_ms),
                    rr_due: mmp.receiver.should_send_report(now_ms),
                    log_due: mmp.should_log(now_ms),
                })
            })
            .collect();

        let actions = self.mmp.plan_link_reports(&snapshots);

        // Drive the planned actions in their phase-grouped order (all logs, then
        // all SenderReports, then all ReceiverReports). Logs run first because the
        // operator log reads cumulative_packets_sent, which each report send
        // advances (send_encrypted_link_message -> sender.record_sent); the
        // pre-refactor handler logged during its collect pass, before any send.
        // `build_report` (which advances the interval state) is called only on a
        // SendLinkReport action, exactly as the pre-refactor gate did.
        for action in actions {
            match action {
                MmpAction::SendLinkReport { peer, kind } => {
                    let encoded = self
                        .peers
                        .get_mut(&peer)
                        .and_then(|p| p.mmp_mut())
                        .and_then(|mmp| match kind {
                            LinkReportKind::Sender => {
                                mmp.sender.build_report(now_ms).map(|sr| sr.encode())
                            }
                            LinkReportKind::Receiver => {
                                mmp.receiver.build_report(now_ms).map(|rr| rr.encode())
                            }
                        });
                    if let Some(encoded) = encoded
                        && let Err(e) = self.send_encrypted_link_message(&peer, &encoded).await
                    {
                        let label = match kind {
                            LinkReportKind::Sender => "Failed to send SenderReport",
                            LinkReportKind::Receiver => "Failed to send ReceiverReport",
                        };
                        debug!(peer = %self.peer_display_name(&peer), error = %e, "{}", label);
                    }
                }
                MmpAction::LogLink { peer } => {
                    // Resolve the display name exactly as the pre-refactor loop
                    // did (alias, else short_npub) — not `peer_display_name`,
                    // which also consults the host map.
                    let peer_name = self.peer_aliases.get(&peer).cloned().unwrap_or_else(|| {
                        self.peers
                            .get(&peer)
                            .map(|p| p.identity().short_npub())
                            .unwrap_or_default()
                    });
                    if let Some(mmp) = self.peers.get_mut(&peer).and_then(|p| p.mmp_mut()) {
                        Self::log_mmp_metrics(&peer_name, mmp);
                        mmp.mark_logged(now_ms);
                    }
                }
                MmpAction::ReapPeer { .. }
                | MmpAction::Heartbeat { .. }
                | MmpAction::SendSessionReport { .. }
                | MmpAction::LogSession { .. } => {}
            }
        }
    }

    /// Emit periodic MMP metrics for a peer.
    fn log_mmp_metrics(peer_name: &str, mmp: &crate::proto::mmp::MmpPeerState) {
        let m = &mmp.metrics;

        let rtt_str = if m.rtt_trend.initialized() {
            format!("{:.1}ms", m.rtt_trend.long() / 1000.0)
        } else {
            "n/a".to_string()
        };
        let loss_str = if m.loss_trend.initialized() {
            format!("{:.1}%", m.loss_trend.long() * 100.0)
        } else {
            "n/a".to_string()
        };
        let jitter_ms = mmp.receiver.jitter_us() as f64 / 1000.0;

        debug!(
            peer = %peer_name,
            rtt = %rtt_str,
            loss = %loss_str,
            jitter = format_args!("{:.1}ms", jitter_ms),
            goodput = %format_throughput(m.goodput_bps()),
            tx_pkts = mmp.sender.cumulative_packets_sent(),
            rx_pkts = mmp.receiver.cumulative_packets_recv(),
            "MMP link metrics"
        );
    }

    /// Emit a teardown log summarizing lifetime MMP metrics for a removed peer.
    pub(in crate::node) fn log_mmp_teardown(
        peer_name: &str,
        mmp: &crate::proto::mmp::MmpPeerState,
    ) {
        let m = &mmp.metrics;
        let jitter_ms = mmp.receiver.jitter_us() as f64 / 1000.0;

        let rtt_str = match m.srtt_ms() {
            Some(rtt) => format!("{:.1}ms", rtt),
            None => "n/a".to_string(),
        };
        let loss_str = format!("{:.1}%", m.loss_rate() * 100.0);

        debug!(
            peer = %peer_name,
            rtt = %rtt_str,
            loss = %loss_str,
            jitter = format_args!("{:.1}ms", jitter_ms),
            etx = format_args!("{:.2}", m.etx),
            goodput = %format_throughput(m.goodput_bps()),
            tx_pkts = mmp.sender.cumulative_packets_sent(),
            rx_pkts = mmp.receiver.cumulative_packets_recv(),
            rx_bytes = mmp.receiver.cumulative_bytes_recv(),
            "MMP link teardown"
        );
    }

    /// Send heartbeats and remove dead peers.
    ///
    /// Called from the tick handler. Sends a 1-byte heartbeat to each peer
    /// whose heartbeat interval has elapsed, and removes any peer that
    /// hasn't sent us a frame within the link dead timeout.
    pub(in crate::node) async fn check_link_heartbeats(&mut self) {
        let now = Instant::now();
        // Monotonic ms for the MMP receiver's injected-`u64` liveness clock; the
        // Instant `now` is still used for the shell-owned heartbeat timing and
        // the session-start fallback (both `ActivePeer` Instants).
        let now_ms = crate::time::mono_ms();
        let heartbeat_interval = Duration::from_secs(self.config().node.heartbeat_interval_secs);
        let dead_timeout = Duration::from_secs(self.config().node.link_dead_timeout_secs);
        let dead_timeout_ms = dead_timeout.as_millis() as u64;
        let max_resends = self.config().node.rate_limit.handshake_max_resends;
        let heartbeat_msg = [LinkMessageType::Heartbeat.to_byte()];

        // Build one liveness snapshot per peer, resolving every clock read and
        // the rekey-suppression predicate shell-side. The snapshots own only
        // `NodeAddr`/`bool`, so the peer-iteration borrow is released before the
        // pure decision runs and the driving loop mutates the registry.
        let snapshots: Vec<PeerLivenessSnapshot> = self
            .peers
            .iter()
            .map(|(node_addr, peer)| {
                // Check liveness via the MMP receiver's last-received monotonic
                // ms. Fall back to session_start (an `ActivePeer` Instant) for
                // peers that never sent data, keeping that branch in Instant
                // space so no monotonic-ms epoch conversion is needed.
                let time_dead = if let Some(mmp) = peer.mmp() {
                    match mmp.receiver.last_recv_ms() {
                        Some(last_ms) => now_ms.saturating_sub(last_ms) >= dead_timeout_ms,
                        None => now.duration_since(peer.session_start()) >= dead_timeout,
                    }
                } else {
                    false
                };

                // Suppress teardown while an FMP rekey is genuinely in flight
                // with budget left: a rekey-handshake link is not silent —
                // whether mid-msg1 or mid-msg3 retransmit. The resend caps
                // guarantee this terminates (abandon on exhaustion or cutover on
                // completion clears the rekey state), so a truly dead link is
                // reaped on the next cycle.
                let rekey_active = (peer.rekey_in_progress()
                    && peer.rekey_msg1_resend_count() < max_resends
                    && peer.rekey_msg1().is_some())
                    || (peer.rekey_msg3_payload().is_some()
                        && peer.rekey_msg3_resend_count() < max_resends);

                // Check if heartbeat is due. Two gates, not one: a send that
                // failed does not satisfy the interval, so a peer that has
                // heard nothing stays due instead of being suppressed by an
                // attempt that went nowhere, and the retry gap keeps a peer
                // whose send keeps failing from being tried on every tick.
                // Both are decided by `heartbeat_due`, which is a pure
                // function so it can be tested without driving a send.
                let heartbeat_due = heartbeat_due(
                    peer.last_heartbeat_sent(),
                    peer.last_heartbeat_attempt(),
                    now,
                    heartbeat_interval,
                );

                PeerLivenessSnapshot {
                    peer: *node_addr,
                    time_dead,
                    rekey_active,
                    heartbeat_due,
                }
            })
            .collect();

        let actions = self.mmp.plan_heartbeats(&snapshots);

        // Wall-clock basis for reconnect scheduling, sourced once (as before).
        let now_ms = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_millis() as u64)
            .unwrap_or(0);

        // Drive the planned actions: all reaps first (each removed +
        // reconnect-scheduled), then all heartbeats (a just-reaped peer is never
        // heartbeated — the core never emits both for the same peer).
        for action in actions {
            match action {
                MmpAction::ReapPeer { peer } => {
                    // Log SHELL-SIDE before routing so the reap keeps the
                    // `fips::node::handlers::mmp` tracing target (no relocation into
                    // the executor, no target pin needed).
                    debug!(
                        peer = %self.peer_display_name(&peer),
                        timeout_secs = self.config().node.link_dead_timeout_secs,
                        "Removing peer: link dead timeout"
                    );
                    self.route_link_dead(peer, now_ms).await;
                }
                MmpAction::Heartbeat { peer } => {
                    // Attempt first, success after: the attempt is recorded
                    // even if the send below fails or never returns, so the
                    // retry stays spaced; only a send that came back clean
                    // moves the interval that says the peer has heard from us.
                    if let Some(p) = self.peers.get_mut(&peer) {
                        p.mark_heartbeat_attempt(now);
                    }
                    match self
                        .send_encrypted_link_message(&peer, &heartbeat_msg)
                        .await
                    {
                        Ok(()) => {
                            if let Some(p) = self.peers.get_mut(&peer) {
                                p.mark_heartbeat_sent(now);
                            }
                        }
                        Err(e) => {
                            trace!(peer = %self.peer_display_name(&peer), error = %e, "Failed to send heartbeat");
                        }
                    }
                }
                MmpAction::SendLinkReport { .. }
                | MmpAction::LogLink { .. }
                | MmpAction::SendSessionReport { .. }
                | MmpAction::LogSession { .. } => {}
            }
        }
    }

    /// Reap every active peer reachable only through `transport_id`.
    ///
    /// Called on a transport's detach edge. Until this existed, losing an
    /// interface withdrew nothing: the peers stayed in the registry, the
    /// routes through them stayed selectable, and this node kept advertising
    /// reachability it no longer had — so transit traffic was dropped in
    /// silence, and other nodes kept routing toward us for those destinations,
    /// until the liveness reaper noticed up to `link_dead_timeout_secs` later.
    /// Measured on real hardware that was 27 seconds of routing through a link
    /// that had already gone, with four alternative peers available the whole
    /// time.
    ///
    /// The detach edge is both earlier and more certain than inactivity, so it
    /// is the better trigger. This routes through the same
    /// [`Self::route_link_dead`] the liveness reaper uses rather than
    /// open-coding a second teardown — every consequence of losing a peer
    /// (sessions, path MTU, session indices, the link, the control machine,
    /// tree cleanup and re-announce, bloom withdrawal) already hangs off that
    /// one path, and a parallel one would drift from it.
    ///
    /// Deliberately undamped. A flapping interface cannot drive a reap storm
    /// through here, because `ChurnGuard` stops publishing presence edges
    /// after three short-lived bindings and does not resume until one lasts —
    /// so the edges this reacts to are already rate-limited at the source, and
    /// a second damper here would only add a way for the two to disagree.
    ///
    /// Returns how many peers were reaped.
    pub(in crate::node) async fn reap_peers_on_transport(
        &mut self,
        transport_id: TransportId,
    ) -> usize {
        let doomed: Vec<NodeAddr> = self
            .peers
            .iter()
            .filter(|(_, peer)| peer.transport_id() == Some(transport_id))
            .map(|(node_addr, _)| *node_addr)
            .collect();

        if doomed.is_empty() {
            return 0;
        }

        let now_ms = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_millis() as u64)
            .unwrap_or(0);

        let reaped = doomed.len();
        for node_addr in doomed {
            debug!(
                peer = %self.peer_display_name(&node_addr),
                %transport_id,
                "Removing peer: its interface went away"
            );
            self.route_link_dead(node_addr, now_ms).await;
        }
        reaped
    }

    /// Route a link-dead reap through the peer machine + executor. Two callers
    /// decide: the tick sweep's `plan_heartbeats` batch emits a `ReapPeer` for a
    /// peer that has gone quiet, and [`Self::reap_peers_on_transport`] withdraws
    /// a transport's peers when its interface goes away. Mirrors
    /// [`route_rekey_cadence`](Node::route_rekey_cadence): the shell has already
    /// decided by the time this runs, so the machine only CONSUMES the decision
    /// via [`PeerEvent::LinkDeadSuspected`]. The resulting executor arms
    /// (`InvalidateSendState` → `remove_active_peer`, `ReportLost` →
    /// `note_link_dead`) reproduce the pre-refactor inline reap body exactly, in
    /// that order.
    ///
    /// An established peer always has a `peer_machine`. If the peer vanished
    /// between snapshot and effect, the old inline body was already a
    /// no-op, so we return; if the machine is absent (which should be impossible)
    /// we fall back to the byte-identical inline body under a `debug_assert`.
    ///
    /// `now_ms` is the caller's hoisted wall-clock ms (the same value the old
    /// reap fed `note_link_dead`); it flows to the executor `ReportLost` arm via
    /// `ambient.now_ms`. Both callers hoist it once per batch, so every peer
    /// removed in one pass carries the same instant.
    async fn route_link_dead(&mut self, node_addr: NodeAddr, now_ms: u64) {
        let link = match self.peers.get(&node_addr) {
            Some(peer) => peer.link_id(),
            None => return,
        };
        if !self.peer_machines.contains_key(&link) {
            debug_assert!(false, "peer machine present for every established peer");
            self.remove_active_peer(&node_addr);
            self.note_link_dead(node_addr, now_ms);
            return;
        }
        let ambient = self.link_dead_ctx(&node_addr, now_ms);
        self.advance_peer_machine(link, PeerEvent::LinkDeadSuspected, Self::now_ms(), &ambient)
            .await;
    }

    /// Ambient shell facts for the routed liveness reap. The executor reads only
    /// `verified_identity` (`InvalidateSendState` → `remove_active_peer` resolves
    /// its `NodeAddr` from it, so it must equal `node_addr`) and `now_ms`
    /// (`ReportLost` → `note_link_dead`, the wall-clock reconnect basis). The
    /// transport/index/direction fields are unused by these two arms and are
    /// populated best-effort for coherence. `now_ms` is threaded in (rather than
    /// re-read) so the value fed to `note_link_dead` is byte-identical to the old
    /// reap's hoisted wall-clock for every peer in the sweep.
    fn link_dead_ctx(&self, node_addr: &NodeAddr, now_ms: u64) -> PeerActionCtx {
        let peer = &self.peers[node_addr];
        PeerActionCtx {
            verified_identity: *peer.identity(),
            transport_id: peer.transport_id().unwrap_or_else(|| TransportId::new(0)),
            remote_addr: peer
                .current_addr()
                .cloned()
                .unwrap_or_else(|| TransportAddr::new(Vec::new())),
            our_index: peer.our_index(),
            their_index: peer.their_index(),
            now_ms,
            is_outbound: false,
            pending_outbound_key: None,
        }
    }
}

#[cfg(test)]
mod tests {
    use super::{HEARTBEAT_RETRY_INTERVAL, heartbeat_due};
    use std::time::{Duration, Instant};

    const INTERVAL: Duration = Duration::from_secs(10);

    #[test]
    fn a_peer_never_heartbeated_is_due_immediately() {
        let now = Instant::now();
        assert!(heartbeat_due(None, None, now, INTERVAL));
    }

    #[test]
    fn a_peer_whose_heartbeat_landed_waits_the_configured_interval() {
        let landed = Instant::now();
        assert!(!heartbeat_due(
            Some(landed),
            Some(landed),
            landed + INTERVAL - Duration::from_millis(1),
            INTERVAL
        ));
        assert!(heartbeat_due(
            Some(landed),
            Some(landed),
            landed + INTERVAL,
            INTERVAL
        ));
    }

    #[test]
    fn a_healthy_peer_is_paced_by_the_configured_interval_and_not_by_the_retry_floor() {
        // The interval a peer configures can be shorter than the retry floor.
        // Consulting the retry gate on the healthy path would clamp it, and
        // `heartbeat_interval_secs` has no validation floor to prevent that.
        let short = Duration::from_secs(1);
        assert!(short < HEARTBEAT_RETRY_INTERVAL);
        let landed = Instant::now();
        assert!(heartbeat_due(
            Some(landed),
            Some(landed),
            landed + short,
            short
        ));
    }

    #[test]
    fn a_failed_attempt_does_not_suppress_the_next_heartbeat_for_a_full_interval() {
        // A heartbeat landed at t0 and the next attempt, at t0 + INTERVAL,
        // failed. Once the retry interval has passed the peer is due again,
        // rather than waiting another whole interval on a send that never
        // reached it.
        let landed = Instant::now();
        let failed = landed + INTERVAL;
        assert!(heartbeat_due(
            Some(landed),
            Some(failed),
            failed + HEARTBEAT_RETRY_INTERVAL,
            INTERVAL
        ));
    }

    #[test]
    fn a_failed_attempt_is_not_retried_before_the_retry_interval() {
        let landed = Instant::now();
        let failed = landed + INTERVAL;
        assert!(!heartbeat_due(
            Some(landed),
            Some(failed),
            failed + HEARTBEAT_RETRY_INTERVAL - Duration::from_millis(1),
            INTERVAL
        ));
    }

    #[test]
    fn a_peer_never_reached_is_retried_on_the_retry_interval_not_the_heartbeat_interval() {
        // No heartbeat has ever landed, so there is no interval to pace by.
        // The attempt alone spaces the retries.
        let failed = Instant::now();
        assert!(!heartbeat_due(
            None,
            Some(failed),
            failed + Duration::from_millis(1),
            INTERVAL
        ));
        assert!(heartbeat_due(
            None,
            Some(failed),
            failed + HEARTBEAT_RETRY_INTERVAL,
            INTERVAL
        ));
    }
}