diff --git a/src/node/handlers/mmp.rs b/src/node/handlers/mmp.rs index e1f9866..43fe932 100644 --- a/src/node/handlers/mmp.rs +++ b/src/node/handlers/mmp.rs @@ -6,14 +6,17 @@ 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}; @@ -500,13 +503,15 @@ impl Node { 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.remove_active_peer(&peer); - self.note_link_dead(peer, now_ms); + self.route_link_dead(peer, now_ms).await; } MmpAction::Heartbeat { peer } => { if let Some(p) = self.peers.get_mut(&peer) { @@ -526,4 +531,64 @@ impl Node { } } } + + /// Route a link-dead liveness reap through the peer machine + executor + /// (C5-1). Mirrors [`route_rekey_cadence`](Node::route_rekey_cadence): the + /// shell already decided (the tick sweep's `plan_heartbeats` batch emitted + /// this `ReapPeer` in phase order), 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. + /// + /// Finding A: 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 (impossible per Finding A) + /// we fall back to the byte-identical inline body under a `debug_assert`. + /// + /// `now_ms` is the sweep'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`. + 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 (Finding A)" + ); + 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. Mirrors + /// [`rekey_cadence_ctx`](Node::rekey_cadence_ctx). 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, + } + } } diff --git a/src/peer/machine.rs b/src/peer/machine.rs index b7f3bf4..ea91952 100644 --- a/src/peer/machine.rs +++ b/src/peer/machine.rs @@ -996,14 +996,26 @@ impl PeerMachine { } fn on_link_dead(&mut self, now: u64) -> Vec { - if !self.is_active_like() { + // Guard the full established set (Established | Active | Maintaining), not + // just `is_active_like()`: a peer that never rekeyed stays parked in + // `Established` (the machine reaches `Active` only via a rekey `Drain`), + // yet the pre-refactor liveness reap tore down EVERY dead established peer. + // A too-narrow `is_active_like()` guard here would silently skip the common + // (never-rekeyed) reap target. Mirrors `on_disconnect`'s guard. + if !self.is_established_context() { return Vec::new(); } - let mut actions = vec![PeerAction::InvalidateSendState]; - if let Some(idx) = self.our_index.take() { - actions.push(PeerAction::UnregisterDecryptSession { index: idx }); - } - actions.push(PeerAction::TeardownConnectedUdp); + // `InvalidateSendState` maps to the executor's `remove_active_peer`, which + // unregisters the decrypt worker by the REAL current index. The machine's + // shadow `our_index` is deliberately NOT used to unregister here (C5-0): it + // can drift to a reused index and wrongly unregister ANOTHER peer's worker + // session. `TeardownConnectedUdp` is inert (C6 — the old reap had no + // connected-UDP teardown, so inert is neutral); `ReportLost` drives the + // loss reflex (`note_link_dead`). + let mut actions = vec![ + PeerAction::InvalidateSendState, + PeerAction::TeardownConnectedUdp, + ]; if let Some(peer) = self.addr() { actions.push(PeerAction::ReportLost { peer }); } @@ -1871,9 +1883,6 @@ mod tests { dead, vec![ PeerAction::InvalidateSendState, - PeerAction::UnregisterDecryptSession { - index: SessionIndex::new(0x4242) - }, PeerAction::TeardownConnectedUdp, PeerAction::ReportLost { peer: addr }, ]