From c15e63e101f4c83d6d783637cecd2f9a6c697fbf Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Mon, 13 Jul 2026 20:52:58 +0000 Subject: [PATCH] node: drive the link-dead peer reap through the per-peer machine MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Route the liveness reap through the per-peer machine: on a link-dead timeout, the mmp reap advances the peer's machine with a link-dead event and executes the resulting actions, instead of calling remove_active_peer and note_link_dead inline. The effect is unchanged — the machine emits InvalidateSendState then ReportLost, which the executor maps back to remove_active_peer(peer) then note_link_dead(peer, now), the same two calls in the same order. The reap log stays shell-side so it keeps the handlers::mmp target, and a guard falls back to the inline path if a machine is somehow absent. --- src/node/dataplane/mod.rs | 2 ++ src/node/handlers/mmp.rs | 68 +++++++++++++++++++++++++++++++++++++-- 2 files changed, 68 insertions(+), 2 deletions(-) diff --git a/src/node/dataplane/mod.rs b/src/node/dataplane/mod.rs index e0643a1..bcf5d61 100644 --- a/src/node/dataplane/mod.rs +++ b/src/node/dataplane/mod.rs @@ -14,3 +14,5 @@ mod encrypted; mod forwarding; mod peer_actions; mod rx_loop; + +pub(in crate::node) use peer_actions::PeerActionCtx; diff --git a/src/node/handlers/mmp.rs b/src/node/handlers/mmp.rs index aef0b4a..46a2c81 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,63 @@ impl Node { } } } + + /// Route a link-dead liveness reap through the peer machine + executor. + /// 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, in + /// that order. + /// + /// 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. 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, + } + } }