From e9112cc1bb2c854f5d0c77bcef5d670e241ee346 Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Mon, 13 Jul 2026 13:42:08 +0000 Subject: [PATCH] node: drive net-new outbound establish through the per-peer machine Cut the net-new outbound handshake completion (a received msg2 that promotes a fresh outbound leg to a new peer) over to the per-peer control machine, mirroring the inbound cutover. handle_msg2 still runs the msg2 prologue, the ACL check, and the cross-connection swap/keep arms inline; for the net-new promote it now builds a transient machine, steps it, and drives promote_connection through the executor. The session index was already allocated at dial, so there is no two-phase authorize here. The wire, index sequence, and peer registry state after promote are byte-neutral. To keep the promote-failure path neutral, the executor's cleanup now distinguishes inbound from outbound: an outbound promote failure records the reject only, matching the prior handler, rather than the inbound path's link and index teardown. Cross-connection swap/keep, rekey-msg2, and the dial path stay inline; the loser-link surgery for the currently unreachable driven cross-connection case lands with the register relocation next. --- src/node/dataplane/peer_actions.rs | 64 +++++++--- src/node/handlers/handshake.rs | 184 +++++++++++++++++------------ src/proto/fmp/core.rs | 7 ++ 3 files changed, 162 insertions(+), 93 deletions(-) diff --git a/src/node/dataplane/peer_actions.rs b/src/node/dataplane/peer_actions.rs index f1f8aab..c3c3593 100644 --- a/src/node/dataplane/peer_actions.rs +++ b/src/node/dataplane/peer_actions.rs @@ -54,6 +54,14 @@ pub(in crate::node) struct PeerActionCtx { pub(in crate::node) their_index: Option, /// The wire timestamp driving this step (promotion ts / loss-report clock). pub(in crate::node) now_ms: u64, + /// Establish direction for this exchange. Discriminates the + /// `PromoteToActive` failure cleanup: the pre-refactor inbound + /// (`handle_msg1`) and outbound (`handle_msg2`) promote-Err arms were NOT + /// byte-identical, so the executor must reproduce each. `false` = inbound + /// (drop link + reverse map + free index), `true` = outbound (record the + /// reject only; leave the dead link/`addr_to_link` for the stale-connection + /// reaper, matching old `handle_msg2`). + pub(in crate::node) is_outbound: bool, } impl Node { @@ -179,22 +187,48 @@ impl Node { } Err(e) => { // GAP-4: promotion failed. `promote_connection` already - // removed `connections[link]`; mirror the pre-refactor - // cleanup (`handle_msg1` L587-591): drop the link + - // reverse map, free our index, discard the machine, and - // record the reject. The queue is drained (PromoteToActive - // is the last establish action), so no explicit abort. - // - // Restored pre-refactor promote-failure warn! - // (`handle_msg1` L757). - warn!(link_id = %promote_link, error = %e, "Failed to promote inbound connection"); - self.remove_link(&promote_link); - if let Some(idx) = ambient.our_index { - let _ = self.index_allocator.free(idx); + // removed `connections[link]` and (on error) handled its + // own index internally. The pre-refactor inbound and + // outbound promote-Err arms were NOT byte-identical, so + // discriminate on `ambient.is_outbound`. The queue is + // drained (PromoteToActive is the last establish action), + // so no explicit abort. + if ambient.is_outbound { + // OLD outbound (`handle_msg2` promote-Err): warn + + // record_reject ONLY. NO `remove_link`, NO + // `index_allocator.free`, NO `addr_to_link` removal — + // the dead link/addr_to_link/pending_outbound were + // left for the 30s stale-connection reaper + // (`promote_connection` already handled + // `connections[link]`/its index on error). Restored + // pre-refactor outbound warn! ("Failed to promote + // connection"). + // + // The transient outbound machine was inserted BEFORE + // execute (Model A); it is additive C3-1 state that + // did not exist pre-refactor, so removing the just- + // inserted machine on failure is neutral vs old and + // prevents a leak. + warn!(link_id = %promote_link, error = %e, "Failed to promote connection"); + self.stats_mut().record_reject(RejectReason::Handshake( + HandshakeReject::BadState, + )); + self.peer_machines.remove(&promote_link); + } else { + // OLD inbound (`handle_msg1` L587-591): drop the link + // + reverse map, free our index, discard the machine, + // and record the reject. Restored pre-refactor inbound + // promote-failure warn! (`handle_msg1` L757). + warn!(link_id = %promote_link, error = %e, "Failed to promote inbound connection"); + self.remove_link(&promote_link); + if let Some(idx) = ambient.our_index { + let _ = self.index_allocator.free(idx); + } + self.peer_machines.remove(&promote_link); + self.stats_mut().record_reject(RejectReason::Handshake( + HandshakeReject::BadState, + )); } - self.peer_machines.remove(&promote_link); - self.stats_mut() - .record_reject(RejectReason::Handshake(HandshakeReject::BadState)); } } } diff --git a/src/node/handlers/handshake.rs b/src/node/handlers/handshake.rs index ba1b7f0..953beb7 100644 --- a/src/node/handlers/handshake.rs +++ b/src/node/handlers/handshake.rs @@ -6,7 +6,9 @@ use crate::node::acl::PeerAclContext; use crate::node::dataplane::PeerActionCtx; use crate::node::reject::{HandshakeReject, RejectReason}; use crate::node::{Node, NodeError}; -use crate::peer::machine::{FailReason, HandshakePhase, PeerEvent, PeerMachine, PeerState}; +use crate::peer::machine::{ + FailReason, HandshakePhase, PeerAction, PeerEvent, PeerMachine, PeerState, +}; use crate::peer::{ActivePeer, PeerConnection}; use crate::proto::fmp::wire::{Msg1Header, Msg2Header, build_msg2}; use crate::proto::fmp::{ @@ -506,6 +508,7 @@ impl Node { our_index: None, their_index: Some(their_index), now_ms: packet.timestamp_ms, + is_outbound: false, }; self.execute_peer_actions(link_id, &teardown_ctx, phase1) .await; @@ -590,6 +593,7 @@ impl Node { our_index: Some(our_index), their_index: Some(their_index), now_ms: packet.timestamp_ms, + is_outbound: false, }; self.execute_peer_actions(link_id, &ambient, promote_actions) .await; @@ -747,6 +751,7 @@ impl Node { our_index: Some(our_index), their_index: Some(their_index), now_ms: packet.timestamp_ms, + is_outbound: false, }; self.execute_peer_actions(link_id, &ambient, promote_actions) .await; @@ -1114,85 +1119,102 @@ impl Node { return; } - // Normal path: promote to active peer - let promote = ConnAction::PromoteToActive { link: link_id }; - match self.drive_promote_to_active(promote, peer_identity, packet.timestamp_ms) { - Ok(result) => { - // Clean up pending_outbound - self.pending_outbound.remove(&key); + // === C3-3a: net-new outbound establish, driven by the machine. === + // ONLY the `establish_outbound == Promote` arm is cut over here. The + // Swap/Keep cross-connection arms and the rekey-msg2 completion branch + // above STAY INLINE (§0): they mutate an existing already-promoted peer + // via `replace_session` with no PeerAction, so the machine's C1 Swap/Keep + // arms cannot be neutral until `PeerSendState` expresses `replace_session`. + // + // This arm is `has_existing_peer == false` only, so `promote_connection` + // always hits its else branch and returns `Promoted`; the defensive + // `CrossConnectionWon/Lost` follow-ups are UNREACHABLE here (their + // loser-link surgery lands in C3-3b). Direct analog of the C3-2a inbound + // net-new arm — no ordering constraint, lowest risk. + // + // Model A: build a TRANSIENT outbound machine, step `Msg2 → + // [PromoteToActive]`, execute it (→ `promote_connection` → + // `PromotionResolved{Promoted}` → inert `RegisterDecryptSession`, R2), and + // insert into `peer_machines` only on the Promoted (Established) tail. The + // outbound `our_index` was allocated at DIAL (unchanged), the outbound + // promote sends nothing on the wire, and `promote_connection` frees + // nothing new — so the index sequence, `peers`/`peers_by_index`/ + // `addr_to_link` state, and metrics are byte-identical to the pre-refactor + // Promoted arm. `pending_outbound` lifecycle stays shell-side (removed on + // the Established tail, exactly where the pre-refactor Ok arm removed it); + // the machine never touches it. + let mut machine = PeerMachine::new_outbound(link_id, peer_identity, packet.timestamp_ms); - match result { - PromotionResult::Promoted(node_addr) => { - info!( - peer = %self.peer_display_name(&node_addr), - "Peer promoted to active" - ); - // Send initial tree announce to new peer - if let Err(e) = self.send_tree_announce_to_peer(&node_addr).await { - debug!(peer = %self.peer_display_name(&node_addr), error = %e, "Failed to send initial TreeAnnounce"); - } - // Schedule filter announce (sent on next tick via debounce) - self.bloom_state.mark_update_needed(node_addr); - self.reset_lookup_backoff(); - } - PromotionResult::CrossConnectionWon { - loser_link_id, - node_addr, - } => { - // Close the losing TCP connection (no-op for connectionless) - 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; - } - } - // Clean up the losing connection's link - self.remove_link(&loser_link_id); - // Ensure addr_to_link points to the winning link - self.addr_to_link - .insert((packet.transport_id, packet.remote_addr.clone()), link_id); - debug!( - peer = %self.peer_display_name(&node_addr), - loser_link_id = %loser_link_id, - "Outbound cross-connection won, loser link cleaned up" - ); - // Send initial tree announce to peer (new or reconnected) - if let Err(e) = self.send_tree_announce_to_peer(&node_addr).await { - debug!(peer = %self.peer_display_name(&node_addr), error = %e, "Failed to send initial TreeAnnounce"); - } - // Schedule filter announce (sent on next tick via debounce) - self.bloom_state.mark_update_needed(node_addr); - self.reset_lookup_backoff(); - } - PromotionResult::CrossConnectionLost { winner_link_id } => { - // Close the losing TCP connection (no-op for connectionless) - if let Some(transport) = self.transports.get(&packet.transport_id) { - transport.close_connection(&packet.remote_addr).await; - } - // This connection lost — clean up its link - self.remove_link(&link_id); - // Ensure addr_to_link points to the winner's link - self.addr_to_link.insert( - (packet.transport_id, packet.remote_addr.clone()), - winner_link_id, - ); - debug!( - winner_link_id = %winner_link_id, - "Outbound cross-connection lost, keeping existing" - ); - } - } - } - Err(e) => { - warn!( - link_id = %link_id, - error = %e, - "Failed to promote connection" - ); - self.stats_mut() - .record_reject(RejectReason::Handshake(HandshakeReject::BadState)); + // Step `Msg2 → [PromoteToActive]`. The machine re-runs the pure + // `establish_outbound` on the snapshot (a harmless second pure call, as in + // C3-2a); `has_existing_peer == false` reproduces the `Promote` decision. + let promote_actions = machine.step( + PeerEvent::Msg2 { + their_index: header.sender_idx, + out: out_snap, + }, + packet.timestamp_ms, + &mut self.index_allocator, + ); + debug_assert_eq!( + promote_actions, + vec![PeerAction::PromoteToActive { link: link_id }] + ); + + // Register the machine (Promote tail only — the Swap/Keep/rekey arms above + // all returned without inserting). Inserted BEFORE execute so the + // executor's `PromoteToActive` arm can feed `PromotionResolved` back into + // it via the `peer_machines` lookup. The outbound `link_id` was allocated + // at dial and the dial path never inserts a machine (Model A), so this + // 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). + let ambient = PeerActionCtx { + verified_identity: peer_identity, + transport_id: packet.transport_id, + remote_addr: packet.remote_addr.clone(), + our_index, + their_index: Some(header.sender_idx), + now_ms: packet.timestamp_ms, + is_outbound: true, + }; + self.execute_peer_actions(link_id, &ambient, promote_actions) + .await; + + // Post-`Promoted` shell tail (byte-identical to the pre-refactor Promoted + // arm), reached only when promotion succeeded (machine now Established). + // `pending_outbound.remove` runs here — exactly where the pre-refactor Ok + // arm removed it, before the TreeAnnounce/bloom/backoff tail. A promote + // failure removed the machine and skips the whole tail (the pre-refactor + // Err arm likewise left `pending_outbound` in place and only recorded the + // reject, which the executor's Err arm already did). + debug_assert!(matches!( + self.peer_machines.get(&link_id).map(|m| m.state()), + Some(PeerState::Established { .. }) | None + )); + if matches!( + self.peer_machines.get(&link_id).map(|m| m.state()), + Some(PeerState::Established { .. }) + ) { + self.pending_outbound.remove(&key); + info!( + peer = %self.peer_display_name(&peer_node_addr), + "Peer promoted to active" + ); + // Send initial tree announce to new peer + if let Err(e) = self.send_tree_announce_to_peer(&peer_node_addr).await { + debug!(peer = %self.peer_display_name(&peer_node_addr), error = %e, "Failed to send initial TreeAnnounce"); } + // Schedule filter announce (sent on next tick via debounce) + self.bloom_state.mark_update_needed(peer_node_addr); + self.reset_lookup_backoff(); } } @@ -1206,6 +1228,12 @@ impl Node { /// [`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, diff --git a/src/proto/fmp/core.rs b/src/proto/fmp/core.rs index 5fa27d9..44d2cf4 100644 --- a/src/proto/fmp/core.rs +++ b/src/proto/fmp/core.rs @@ -319,6 +319,13 @@ pub(crate) enum ConnAction { /// 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 }, }