From 800cfb23e3f3f32714e5bdebb483ea209bd9706a Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Mon, 13 Jul 2026 14:20:56 +0000 Subject: [PATCH] node: relocate decrypt-session registration to the establish executor Move register_decrypt_worker_session out of promote_connection into the executor's PromoteToActive handler, gated on a promoted or cross-connection-won result. Every live promote now flows through that one executor path, so registration still fires exactly once at the same synchronous point; the direct test callers of promote_connection spawn no worker pool, so the call was already a no-op for them. Add the cross-connection loser-link teardown (close the losing transport, remove its link, re-point addr_to_link at the winner) to the executor as a guarded follow-up. It is unreachable on the current driven establish paths, which only promote net-new peers, and asserts so, but keeps the executor complete for when that case is driven. Remove the now-dead drive_promote_to_active and ConnAction::PromoteToActive. --- src/node/dataplane/peer_actions.rs | 113 ++++++++++++++++++++++++++--- src/node/handlers/handshake.rs | 73 ++++++------------- src/proto/fmp/core.rs | 15 ---- 3 files changed, 125 insertions(+), 76 deletions(-) diff --git a/src/node/dataplane/peer_actions.rs b/src/node/dataplane/peer_actions.rs index c3c3593..65097da 100644 --- a/src/node/dataplane/peer_actions.rs +++ b/src/node/dataplane/peer_actions.rs @@ -22,6 +22,7 @@ use crate::node::Node; use crate::node::reject::{HandshakeReject, RejectReason}; use crate::peer::machine::{PeerAction, PeerEvent}; +use crate::proto::fmp::PromotionResult; use crate::proto::fmp::wire::build_msg2; use crate::transport::{LinkId, TransportAddr, TransportId}; use crate::utils::index::SessionIndex; @@ -35,7 +36,7 @@ use tracing::warn; /// /// Unlike a machine event/action payload this is **executor-side**, so it may /// hold real values resolved from the wire context (cf. `handle_msg1`'s -/// `wire`/`packet` locals and `drive_promote_to_active`'s ambient args). It is +/// `wire`/`packet` locals and `promote_connection`'s ambient args). It is /// built fresh per driven step by the caller at cutover time (C3-2/C3-3). #[allow(dead_code)] pub(in crate::node) struct PeerActionCtx { @@ -165,15 +166,40 @@ impl Node { } PeerAction::PromoteToActive { link: promote_link } => { // GAP-1: ambient supplies the verified identity + promotion ts - // that `promote_connection` needs (cf. `drive_promote_to_active`). + // that `promote_connection` needs (resolved from the wire ctx). match self.promote_connection( promote_link, ambient.verified_identity, ambient.now_ms, ) { Ok(result) => { + // R1 (C3-3b): the decrypt-worker registration relocated + // OUT of `promote_connection` into THIS single executor + // arm — the one live caller of `promote_connection` (both + // the inbound `handle_msg1` and outbound `handle_msg2` + // net-new establish paths reach it here). Register iff the + // promotion actually created or replaced a peer + // (`Promoted | CrossConnectionWon`), NEVER on + // `CrossConnectionLost`. Run synchronously right after + // `promote_connection` returns, before feeding + // `PromotionResolved` and before any await — the exact + // synchronous point (and Promoted/Won gating) of the + // pre-refactor in-`promote_connection` call. No-op when + // the worker pool isn't spawned (`register_...` early- + // returns), so the direct `promote_connection` test + // callers (which bypass this executor) are unaffected. + #[cfg(unix)] + match result { + PromotionResult::Promoted(node_addr) + | PromotionResult::CrossConnectionWon { node_addr, .. } => { + self.register_decrypt_worker_session(&node_addr); + } + PromotionResult::CrossConnectionLost { .. } => {} + } + // Feed the outcome back into the machine and fold the - // follow-up actions (RegisterDecryptSession, cross-conn + // follow-up actions (RegisterDecryptSession — now a + // redundant no-op, see its arm — and the cross-conn index // frees) into the worklist. Disjoint field borrow again. let follow = match self.peer_machines.get_mut(&promote_link) { Some(machine) => machine.step( @@ -184,6 +210,70 @@ impl Node { None => Vec::new(), }; queue.extend(follow); + + // Defensive cross-connection loser-link surgery (C3-3b). + // LINK-ONLY: close the losing transport connection, drop + // its link, and re-point `addr_to_link`, reproducing the + // pre-refactor inline `handle_msg2`/`handle_msg1` per-arm + // order EXACTLY. The index-plane frees/unregisters are + // owned by the machine's `PromotionResolved{Won/Lost}` + // follow-up (queued just above), so NOTHING here touches + // an index — no double-free. + // + // UNREACHABLE on every current driven path: the inbound + // and outbound net-new establish arms only route to the + // machine when no promoted peer exists for the node_addr + // (and `RestartThenPromote` removes the old peer first), + // so `promote_connection` always returns `Promoted`. The + // `debug_assert!(false, ..)` catches any future path that + // drives a cross-connection through the executor without + // the matching send-state handling. + match result { + PromotionResult::CrossConnectionWon { loser_link_id, .. } => { + debug_assert!( + false, + "executor CrossConnectionWon is unreachable on \ + driven net-new establish paths" + ); + // Close the losing transport connection (no-op for + // connectionless) via the LOSER link's own + // transport/addr, then drop the losing link. + if let Some(loser_link) = self.links.get(&loser_link_id) { + let loser_tid = loser_link.transport_id(); + let loser_addr = loser_link.remote_addr().clone(); + if let Some(transport) = self.transports.get(&loser_tid) { + transport.close_connection(&loser_addr).await; + } + } + self.remove_link(&loser_link_id); + // Point `addr_to_link` at the winning (current) + // link. + self.addr_to_link.insert( + (ambient.transport_id, ambient.remote_addr.clone()), + promote_link, + ); + } + PromotionResult::CrossConnectionLost { winner_link_id } => { + debug_assert!( + false, + "executor CrossConnectionLost is unreachable on \ + driven net-new establish paths" + ); + // Close this (losing) connection, drop its link, + // and restore `addr_to_link` to the winner. + if let Some(transport) = + self.transports.get(&ambient.transport_id) + { + transport.close_connection(&ambient.remote_addr).await; + } + self.remove_link(&promote_link); + self.addr_to_link.insert( + (ambient.transport_id, ambient.remote_addr.clone()), + winner_link_id, + ); + } + PromotionResult::Promoted(_) => {} + } } Err(e) => { // GAP-4: promotion failed. `promote_connection` already @@ -247,14 +337,15 @@ impl Node { } PeerAction::RegisterDecryptSession { index } => { let _ = index; - // C3-2 (HALT-reported): the decrypt-worker registration still - // runs INSIDE `promote_connection` (`handshake.rs:1193/1305`), - // which is the single source of truth for its ~40 direct - // `promote_connection` callers (unit/integration tests) and the - // two live handlers. Relocating it out (GAP-3) would perturb the - // live promote path, so C3-1 keeps it there and drives this - // action as a no-op; the relocation lands with the inbound - // cutover in C3-2. + // No-op by design. C3-3b (R1-a) relocated the decrypt-worker + // registration into the `PromoteToActive` Ok arm above, gated on + // the returned `PromotionResult`, so it runs once per live + // promote (Promoted/Won) at the pre-refactor synchronous point. + // This machine-emitted action is now redundant with that arm; + // kept as an inert no-op (rather than removing the emission) so + // the machine's action sequence and its unit tests stay + // unchanged. The keyed-by-NodeAddr register does not need the + // machine's `index` payload. } PeerAction::UnregisterDecryptSession { index } => { // Executor supplies `transport_id` from ambient; keyed by diff --git a/src/node/handlers/handshake.rs b/src/node/handlers/handshake.rs index 953beb7..acae1d3 100644 --- a/src/node/handlers/handshake.rs +++ b/src/node/handlers/handshake.rs @@ -12,7 +12,7 @@ use crate::peer::machine::{ use crate::peer::{ActivePeer, PeerConnection}; use crate::proto::fmp::wire::{Msg1Header, Msg2Header, build_msg2}; use crate::proto::fmp::{ - ConnAction, EstablishSnapshot, EstablishView, InboundDecision, InboundReject, OutboundDecision, + EstablishSnapshot, EstablishView, InboundDecision, InboundReject, OutboundDecision, OutboundSnapshot, PromotionResult, WireOutcome, cross_connection_winner, }; use crate::transport::{Link, LinkDirection, LinkId, ReceivedPacket}; @@ -1169,13 +1169,14 @@ impl Node { // cannot collide with an existing entry. self.peer_machines.insert(link_id, machine); - // Execute `[PromoteToActive]`. The executor calls `promote_connection` - // (identical to the pre-refactor `drive_promote_to_active`), feeds - // `PromotionResolved{Promoted}` back, and runs the inert - // `RegisterDecryptSession` (R2 — register stays in `promote_connection`). - // A promote failure (e.g. `MaxPeersExceeded` if peers filled between dial - // and msg2) runs the executor's Err cleanup and removes the machine, - // leaving it absent (not Established). + // Execute `[PromoteToActive]`. The executor calls `promote_connection`, + // feeds `PromotionResolved{Promoted}` back, registers the decrypt-worker + // session (R1 — C3-3b relocated the register into the executor's + // `PromoteToActive` Ok arm, gated on the result), and runs the now-inert + // `RegisterDecryptSession` follow-up. A promote failure (e.g. + // `MaxPeersExceeded` if peers filled between dial and msg2) runs the + // executor's Err cleanup and removes the machine, leaving it absent (not + // Established). let ambient = PeerActionCtx { verified_identity: peer_identity, transport_id: packet.transport_id, @@ -1218,36 +1219,6 @@ impl Node { } } - /// Execute a [`ConnAction::PromoteToActive`] from the establish machine. - /// - /// The decision to promote is made by the establish handlers (and, from the - /// establish-core stage on, the pure decision in `proto::fmp`); this is the - /// executor half of the seam. It runs the promotion through - /// [`Self::promote_connection`], resolving the verified identity and - /// promotion timestamp from the ambient wire context, and returns the - /// [`PromotionResult`] so the caller can drive the site-specific - /// post-promotion tail (TreeAnnounce, bloom mark, discovery-backoff reset, - /// loser-link cleanup). - /// - // C3-3a cut the last live caller (the inline outbound `Promote` arm) over to - // the executor's `PromoteToActive` path, so this is now unused. Its caller - // census / retirement is C3-3b (blueprint § C3-3b); kept here (allowed) until - // then so the diff stays scoped to the outbound Promote cutover. - #[allow(dead_code)] - fn drive_promote_to_active( - &mut self, - action: ConnAction, - verified_identity: PeerIdentity, - current_time_ms: u64, - ) -> Result { - match action { - ConnAction::PromoteToActive { link } => { - self.promote_connection(link, verified_identity, current_time_ms) - } - _ => unreachable!("drive_promote_to_active requires a PromoteToActive action"), - } - } - /// Promote a connection to active peer after successful authentication. /// /// Handles cross-connection detection and resolution using tie-breaker rules. @@ -1396,11 +1367,12 @@ impl Node { "Cross-connection resolved: this connection won" ); - // Hand the FMP recv cipher + replay window to the - // decrypt shard worker. (Same as normal-promotion tail - // below.) - #[cfg(unix)] - self.register_decrypt_worker_session(&peer_node_addr); + // R1 (C3-3b): the decrypt-worker registration is no longer done + // here — it relocated OUT of `promote_connection` into the single + // executor `PromoteToActive` Ok arm (`peer_actions.rs`), gated on + // the returned `PromotionResult` (`Promoted | CrossConnectionWon`). + // The executor runs it synchronously right after this call returns, + // before any await, so the live establish behaviour is unchanged. Ok(PromotionResult::CrossConnectionWon { loser_link_id, @@ -1506,13 +1478,14 @@ impl Node { "Connection promoted to active peer" ); - // Hand the FMP recv cipher + replay window to the - // decrypt shard worker. From this point on the worker - // is the sole authority on FMP replay protection for - // this session. No-op when the worker pool isn't - // spawned (unit-test path or `FIPS_DECRYPT_WORKERS=0`). - #[cfg(unix)] - self.register_decrypt_worker_session(&peer_node_addr); + // R1 (C3-3b): the decrypt-worker registration relocated OUT of + // `promote_connection` into the single executor `PromoteToActive` Ok + // arm (`peer_actions.rs`), gated on the returned `PromotionResult` + // (`Promoted | CrossConnectionWon`, never `CrossConnectionLost`). The + // executor runs it synchronously right after this call returns, before + // any await — same point, same effect as the pre-refactor in-place call + // (no-op when the worker pool isn't spawned; unit-test path or + // `FIPS_DECRYPT_WORKERS=0`). Ok(PromotionResult::Promoted(peer_node_addr)) } diff --git a/src/proto/fmp/core.rs b/src/proto/fmp/core.rs index 44d2cf4..a3d1e07 100644 --- a/src/proto/fmp/core.rs +++ b/src/proto/fmp/core.rs @@ -312,21 +312,6 @@ pub(crate) enum ConnAction { bytes: Vec, next_resend_at_ms: u64, }, - /// Promote the completed handshake connection on `link` to an active peer - /// (`promote_connection`): moves the Noise session out of the - /// `PeerConnection`, resolves cross-connection precedence via the - /// tie-breaker, and installs the `ActivePeer`. The shell executes the - /// promotion (resolving the verified identity and promotion timestamp from - /// the ambient wire context) and then runs the post-promotion tail - /// (TreeAnnounce, bloom mark, discovery-backoff reset, loser-link cleanup). - // - // C3-3a cut the inline outbound `Promote` arm — the last constructor of this - // variant — over to the machine's `PeerAction::PromoteToActive` seam. The only - // remaining reference is the (now dead-code-allowed) `drive_promote_to_active` - // matcher; both are retired together in C3-3b, so this variant is allowed - // until then to keep the C3-3a diff scoped to the outbound Promote cutover. - #[allow(dead_code)] - PromoteToActive { link: LinkId }, } /// Read-only view of FMP connection/peer state the lifecycle core needs.