From 7790eb86bdb5a5ef9c18d71df9eaea7e541be4ed Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Fri, 17 Jul 2026 06:44:09 +0000 Subject: [PATCH 1/3] peer: make the outbound msg2 swap/keep machine arms decision-only The machine's on_msg2 cross-connection arms crystallized state and emitted FreeIndex/RegisterDecryptSession actions that duplicate the inline shell resolution, but they are unreachable on the live path (the shell removes the machine before running the swap/keep bodies). Making them live in that shape would double-free the outbound index through the executor. Strip both arms to a single new ResolveCrossConnection { swap } action: a decision conveyed to the driver, not an effect. The shell intercepts it and runs the inline resolution, which owns all effects permanently. The action executor gets a defensive unreachable arm. The Promote arm and set_their_index are unchanged. --- src/node/dataplane/peer_actions.rs | 10 ++++ src/peer/machine.rs | 95 +++++++++++++++++------------- 2 files changed, 65 insertions(+), 40 deletions(-) diff --git a/src/node/dataplane/peer_actions.rs b/src/node/dataplane/peer_actions.rs index 1e2bc62..88d5d93 100644 --- a/src/node/dataplane/peer_actions.rs +++ b/src/node/dataplane/peer_actions.rs @@ -387,6 +387,16 @@ impl Node { } } } + PeerAction::ResolveCrossConnection { .. } => { + // A decision token, not an effect: the outbound msg2 + // handler intercepts it and runs the inline swap/keep + // resolution itself, so it must never reach the executor. + debug_assert!( + false, + "ResolveCrossConnection is intercepted by the msg2 \ + handler and must never reach the executor" + ); + } PeerAction::SwapSendState { .. } => { // Initiator cutover: the live authoritative rekey-cadence // path, routed here from `check_rekey` via diff --git a/src/peer/machine.rs b/src/peer/machine.rs index 36d375f..4251476 100644 --- a/src/peer/machine.rs +++ b/src/peer/machine.rs @@ -330,6 +330,11 @@ pub(crate) enum PeerAction { /// Crystallize identity, re-home the map key, publish send-state /// (`promote_connection`). Resolves to a [`PromotionResolved`](PeerEvent::PromotionResolved). PromoteToActive { link: LinkId }, + /// A DECISION conveyed to the driver, not an effect: emitted by the + /// outbound-msg2 arm when the establish decision is a cross-connection + /// resolution. The shell intercepts it and runs the inline swap/keep + /// resolution; it must never reach the action executor. + ResolveCrossConnection { swap: bool }, /// Initiator-side rekey cutover: swap the published send-state to the pending /// epoch. SwapSendState { epoch: [u8; 8] }, @@ -686,13 +691,17 @@ impl PeerMachine { ] } - /// Outbound completion: classify via `establish_outbound` and drive the - /// promote / cross-connection resolution. + /// Outbound completion: compute the establish decision from the snapshot + /// via `establish_outbound`. `Promote` drives promotion via actions; the + /// Swap/Keep outcomes are conveyed as a + /// [`ResolveCrossConnection`](PeerAction::ResolveCrossConnection) decision + /// for the shell's inline resolution, which owns all effects (index + /// frees, session replacement) permanently. fn on_msg2( &mut self, their_index: SessionIndex, out: OutboundSnapshot, - now: u64, + _now: u64, _alloc: &mut IndexAllocator, ) -> Vec { self.conn.set_their_index(their_index); @@ -717,31 +726,16 @@ impl PeerMachine { ] } OutboundDecision::CrossConnectionSwap => { - // Our outbound wins: swap the peer to the outbound session, - // freeing the old inbound index. Resolved in-step (peer exists). - let outbound_index = self.conn.our_index(); - let old_inbound_index = self.our_index; - self.crystallize(now); - let mut actions = Vec::new(); - if let Some(idx) = old_inbound_index { - actions.push(PeerAction::FreeIndex { index: idx }); - } - if let Some(idx) = outbound_index { - self.our_index = Some(idx); - actions.push(PeerAction::RegisterDecryptSession { index: idx }); - } - actions + // Our outbound wins: convey the decision only. The shell's + // inline resolution swaps the peer to the outbound session and + // owns the index frees and session replacement. + vec![PeerAction::ResolveCrossConnection { swap: true }] } OutboundDecision::CrossConnectionKeep => { - // Our outbound loses: keep the existing inbound session, free the - // unused outbound index. - let outbound_index = self.conn.our_index(); - self.crystallize(now); - let mut actions = Vec::new(); - if let Some(idx) = outbound_index { - actions.push(PeerAction::FreeIndex { index: idx }); - } - actions + // Our outbound loses: convey the decision only. The shell's + // inline resolution keeps the existing inbound session and + // frees the unused outbound index. + vec![PeerAction::ResolveCrossConnection { swap: false }] } } } @@ -1491,6 +1485,7 @@ mod tests { PeerAction::PromoteToActive { link: LinkId::new(7), }, + PeerAction::ResolveCrossConnection { swap: true }, PeerAction::SwapSendState { epoch: [1u8; 8] }, PeerAction::CompleteDrain { peer }, PeerAction::InvalidateSendState, @@ -1526,6 +1521,7 @@ mod tests { | PeerAction::SendRekey { .. } | PeerAction::SendLinkMessage { .. } | PeerAction::PromoteToActive { .. } + | PeerAction::ResolveCrossConnection { .. } | PeerAction::SwapSendState { .. } | PeerAction::CompleteDrain { .. } | PeerAction::InvalidateSendState @@ -2086,7 +2082,9 @@ mod tests { ); assert_eq!(m.state(), PeerState::Established { addr: peer_addr }); - // Cross-connection SWAP: our outbound wins -> free old inbound, register outbound. + // Cross-connection SWAP: our outbound wins -> decision only; the + // shell's inline resolution owns the index frees and session + // replacement. let mut m2 = PeerMachine::new_outbound(LinkId::new(2), peer, 0); m2.state = PeerState::Handshaking { link: LinkId::new(2), @@ -2108,18 +2106,25 @@ mod tests { ); assert_eq!( swap, - vec![ - PeerAction::FreeIndex { - index: SessionIndex::new(0x1111) - }, - PeerAction::RegisterDecryptSession { - index: SessionIndex::new(0x2222) - }, - ] + vec![PeerAction::ResolveCrossConnection { swap: true }] ); - assert_eq!(m2.state(), PeerState::Established { addr: peer_addr }); + assert!(!swap.iter().any(|a| matches!( + a, + PeerAction::FreeIndex { .. } | PeerAction::RegisterDecryptSession { .. } + ))); + // The decision arm leaves the machine untouched: still Handshaking, + // our_index unchanged. + assert_eq!( + m2.state(), + PeerState::Handshaking { + link: LinkId::new(2), + phase: HandshakePhase::SentMsg1, + } + ); + assert_eq!(m2.our_index(), Some(SessionIndex::new(0x1111))); - // Cross-connection KEEP: our outbound loses -> free unused outbound index. + // Cross-connection KEEP: our outbound loses -> decision only; the + // shell's inline resolution frees the unused outbound index. let mut m3 = PeerMachine::new_outbound(LinkId::new(3), peer, 0); m3.state = PeerState::Handshaking { link: LinkId::new(3), @@ -2140,10 +2145,20 @@ mod tests { ); assert_eq!( keep, - vec![PeerAction::FreeIndex { - index: SessionIndex::new(0x3333) - }] + vec![PeerAction::ResolveCrossConnection { swap: false }] ); + assert!(!keep.iter().any(|a| matches!( + a, + PeerAction::FreeIndex { .. } | PeerAction::RegisterDecryptSession { .. } + ))); + assert_eq!( + m3.state(), + PeerState::Handshaking { + link: LinkId::new(3), + phase: HandshakePhase::SentMsg1, + } + ); + assert_eq!(m3.our_index(), None); } // ---- Test 7b: dial-persisted outbound promote leaves our_index unset --- From 94d7b9124425d953fccec1973bea8ebab15b0cd2 Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Fri, 17 Jul 2026 06:59:30 +0000 Subject: [PATCH 2/3] node: route the outbound msg2 establish decision through the machine handle_msg2 no longer pre-computes establish_outbound alongside the machine's own evaluation of the same snapshot. The shell now builds the snapshot, steps the persistent outbound machine once at the decision point, and routes on what comes back: the promote action vector drives promotion through the executor as before, and the ResolveCrossConnection decision selects the inline swap/keep resolution bodies, which are unchanged. The machine step and its defensive transient-rebuild move up from the promote arm to the decision point; the cross-connection path keeps its take-leg-then-dispose ordering with the step preceding both. Comment prose at the touched sites refreshed to describe the single-decision-site shape. --- src/node/handlers/handshake.rs | 134 +++++++++++++++------------------ 1 file changed, 60 insertions(+), 74 deletions(-) diff --git a/src/node/handlers/handshake.rs b/src/node/handlers/handshake.rs index 0707ddc..62f580b 100644 --- a/src/node/handlers/handshake.rs +++ b/src/node/handlers/handshake.rs @@ -13,8 +13,8 @@ use crate::peer::machine::{ use crate::peer::{ActivePeer, PeerConnection}; use crate::proto::fmp::wire::{Msg1Header, Msg2Header, build_msg2}; use crate::proto::fmp::{ - EstablishSnapshot, EstablishView, InboundDecision, InboundReject, OutboundDecision, - OutboundSnapshot, PromotionResult, WireOutcome, cross_connection_winner, + EstablishSnapshot, EstablishView, InboundDecision, InboundReject, OutboundSnapshot, + PromotionResult, WireOutcome, cross_connection_winner, }; use crate::transport::{Link, LinkDirection, LinkId, ReceivedPacket}; use crate::utils::index::SessionIndex; @@ -1074,21 +1074,62 @@ impl Node { // // This ensures both nodes use the same Noise handshake (the winner's // outbound = the loser's inbound). - // Structured classification (pure core): cross-connection swap/keep, or - // a net-new promote. The tie-break is pre-evaluated in the snapshot; the - // effect bodies below are unchanged. + // The machine is the sole computation site of the establish decision: + // the shell builds the outbound snapshot, steps the machine once here, + // and routes on the returned decision — a cross-connection resolves as + // a single `ResolveCrossConnection { swap }` action, a net-new + // establish as the promote action sequence. The Swap/Keep resolution + // bodies stay inline in the shell because they mutate the already + // promoted peer via `replace_session`, for which no `PeerAction` + // exists. The machine was persisted at DIAL, so the executor's + // `PromoteToActive` arm can feed `PromotionResolved` back via the same + // lookup; the `pending_outbound` lifecycle stays shell-side — the + // machine never touches it. let out_snap = self.outbound_snapshot(&peer_node_addr); - let out_decision = self.fmp.establish_outbound(&out_snap); - if out_decision != OutboundDecision::Promote { + let actions = match self.peer_machines.get_mut(&link_id) { + Some(machine) => machine.step( + PeerEvent::Msg2 { + their_index: header.sender_idx, + out: out_snap, + }, + packet.timestamp_ms, + &mut self.index_allocator, + ), + None => { + // No machine persisted at dial (e.g. a test that seeds + // `connections`/`pending_outbound` directly, or any path that + // reaches msg2 without `start_handshake`): reproduce the + // pre-persistence transient exactly. + let mut machine = + PeerMachine::new_outbound(link_id, peer_identity, packet.timestamp_ms); + let actions = machine.step( + PeerEvent::Msg2 { + their_index: header.sender_idx, + out: out_snap, + }, + packet.timestamp_ms, + &mut self.index_allocator, + ); + self.peer_machines.insert(link_id, machine); + actions + } + }; + + let cross_swap = actions.iter().find_map(|action| match action { + PeerAction::ResolveCrossConnection { swap } => Some(*swap), + _ => None, + }); + if let Some(swap) = cross_swap { + // The cross-connection arms are decision-only: the resolution + // action is the whole vector. + debug_assert_eq!(actions, vec![PeerAction::ResolveCrossConnection { swap }]); // Extract the outbound connection from its machine FIRST — the // machine owns it, so disposing the machine before the take would - // destroy the connection. The dial-persisted outbound machine is - // not consumed by the inline Swap/Keep resolution below (which - // mutates the existing promoted peer directly, with no machine), - // so drop it right after the take — unconditionally, whether or - // not a connection was carried — so none of this block's exits - // leave a dangling machine. Matches the pre-persistence path, - // which created no machine for a cross-connection. + // destroy the connection. The machine has delivered its decision + // and the inline resolution below needs no machine, so drop it + // right after the take — unconditionally, whether or not a + // connection was carried — so none of this block's exits leave a + // dangling machine. let taken_conn = self .peer_machines .get_mut(&link_id) @@ -1105,7 +1146,7 @@ impl Node { }; let mut cross_conn_outcome: Option = None; - if out_decision == OutboundDecision::CrossConnectionSwap { + if swap { // We're the smaller node. Swap to outbound session + indices. // The peer will keep their inbound session (complement of ours). let outbound_our_index = conn.our_index(); @@ -1224,72 +1265,18 @@ impl Node { } // === 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: they mutate an existing already-promoted peer - // via `replace_session` with no PeerAction, so the machine's 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 is wired later). Direct analog of the inbound - // net-new arm — no ordering constraint, lowest risk. + // `CrossConnectionWon/Lost` follow-ups are UNREACHABLE here. // - // Look up the outbound machine persisted at DIAL, and step `Msg2 → - // [PromoteToActive]` in place. Both dial paths reach msg2 in - // `Handshaking{SentMsg1}` — each drives the machine to send msg1 before - // msg2 — and `on_msg2` is state-independent regardless: it decides from - // the outbound snapshot, not the - // machine's state, and reads the machine's unset `conn.our_index` - // as `None`, so stepping the persisted (vs the former transient) machine - // is byte-identical: same `Promote` decision (`has_existing_peer == false`), - // same `our_index == None`. The machine is already in `peer_machines` - // (inserted at dial), so the executor's `PromoteToActive` arm can feed - // `PromotionResolved` back via the same lookup. A defensive transient - // reproduces the pre-persistence path if the machine is somehow absent — a - // state-machine inconsistency, since every dialed leg persists one. The - // outbound `our_index` was allocated at DIAL (unchanged), the promote - // sends nothing on the wire, and `promote_connection` frees nothing new, - // so index sequence, `peers`/`peers_by_index`/`addr_to_link` state, and - // metrics are byte-identical. `pending_outbound` lifecycle stays shell-side - // (removed on the Established tail); the machine never touches it. - let promote_actions = match self.peer_machines.get_mut(&link_id) { - Some(machine) => machine.step( - PeerEvent::Msg2 { - their_index: header.sender_idx, - out: out_snap, - }, - packet.timestamp_ms, - &mut self.index_allocator, - ), - None => { - // No machine persisted at dial (e.g. a test that seeds - // `connections`/`pending_outbound` directly, or any path that - // reaches msg2 without `start_handshake`): reproduce the - // pre-persistence transient exactly. - let mut machine = - PeerMachine::new_outbound(link_id, peer_identity, packet.timestamp_ms); - let actions = machine.step( - PeerEvent::Msg2 { - their_index: header.sender_idx, - out: out_snap, - }, - packet.timestamp_ms, - &mut self.index_allocator, - ); - self.peer_machines.insert(link_id, machine); - actions - } - }; // The outbound Msg2 Promote step cancels the two dial-armed handshake // timers (the machine survives promotion, so they would otherwise linger // in `peer_timers` until `drive_peer_timers` lazily discards them — the // promoted leg's pending connection is consumed and the machine has left // `SentMsg1`, so they can no longer fire) and then promotes. - // `PromoteToActive` is still what performs the promotion. + // `PromoteToActive` is what performs the promotion. debug_assert_eq!( - promote_actions, + actions, vec![ PeerAction::CancelTimer { kind: TimerKind::HandshakeRetransmit @@ -1318,8 +1305,7 @@ impl Node { now_ms: packet.timestamp_ms, is_outbound: true, }; - self.execute_peer_actions(link_id, &ambient, promote_actions) - .await; + self.execute_peer_actions(link_id, &ambient, actions).await; // Post-`Promoted` shell tail (byte-identical to the pre-refactor Promoted // arm), reached only when promotion succeeded (machine now Established). From 252d16fab987fcc8860c8c4b5a8b81e0c0e66fb8 Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Fri, 17 Jul 2026 07:12:40 +0000 Subject: [PATCH 3/3] node: feed msg1 send failure to the peer machine as an event send_stored_msg1 marked the embedded leg failed by writing it directly from the shell. Route the write through a new HandshakeSendFailed machine event instead: the machine marks its leg so the stale-connection sweep reclaims it, without leaving the handshaking state, so retransmit eligibility survives the window between the failed send and the sweep exactly as before. send_stored_msg1 gains a now_ms parameter threaded from the action executor for the step call; the failure arm itself ignores it. --- src/node/dataplane/peer_actions.rs | 9 +++- src/node/lifecycle/mod.rs | 25 +++++++---- src/peer/machine.rs | 67 ++++++++++++++++++++++++++++++ 3 files changed, 92 insertions(+), 9 deletions(-) diff --git a/src/node/dataplane/peer_actions.rs b/src/node/dataplane/peer_actions.rs index 88d5d93..09e6c1c 100644 --- a/src/node/dataplane/peer_actions.rs +++ b/src/node/dataplane/peer_actions.rs @@ -208,8 +208,13 @@ impl Node { // (`prepare_outbound_msg1`); send the stored wire. The // machine's empty payload is ignored. let _ = bytes; - self.send_stored_msg1(link, ambient.transport_id, &ambient.remote_addr) - .await; + self.send_stored_msg1( + link, + ambient.transport_id, + &ambient.remote_addr, + ambient.now_ms, + ) + .await; } } PeerAction::SendRekey { .. } => { diff --git a/src/node/lifecycle/mod.rs b/src/node/lifecycle/mod.rs index fb9c2e0..5bfbaa3 100644 --- a/src/node/lifecycle/mod.rs +++ b/src/node/lifecycle/mod.rs @@ -651,9 +651,10 @@ impl Node { } /// Send the msg1 wire that `prepare_outbound_msg1` armed on the connection. - /// On send error, marks the connection failed and RETAINS it (the legacy - /// resend tick retries); a missing wire or transport is a no-op. This is the - /// body of the executor's `SendHandshake` msg1 action — reached on both the + /// On send error, steps the machine with `HandshakeSendFailed` — the machine + /// marks its embedded leg failed and RETAINS it (the legacy resend tick + /// retries); a missing wire or transport is a no-op. This is the body of the + /// executor's `SendHandshake` msg1 action — reached on both the /// connectionless dial and the connection-oriented connect-resolution path, /// after `prepare_outbound_msg1` has armed the wire. pub(in crate::node) async fn send_stored_msg1( @@ -661,6 +662,7 @@ impl Node { link_id: LinkId, transport_id: TransportId, remote_addr: &TransportAddr, + now_ms: u64, ) { let wire_msg1 = match self.leg(&link_id).and_then(|c| c.handshake_msg1()) { Some(w) => w.to_vec(), @@ -687,10 +689,19 @@ impl Node { error = %e, "Failed to send handshake message" ); - // Mark connection as failed but don't remove it yet - // The event loop can handle retry logic - if let Some(conn) = self.leg_mut(&link_id) { - conn.mark_failed(); + // The machine marks its leg failed but retains it — + // the stale-connection sweep reclaims it, and the + // event loop can handle retry logic until then. + if let Some(machine) = self.peer_machines.get_mut(&link_id) { + let actions = machine.step( + PeerEvent::HandshakeSendFailed, + now_ms, + &mut self.index_allocator, + ); + debug_assert!( + actions.is_empty(), + "HandshakeSendFailed must emit no actions" + ); } } } diff --git a/src/peer/machine.rs b/src/peer/machine.rs index 4251476..14be7e7 100644 --- a/src/peer/machine.rs +++ b/src/peer/machine.rs @@ -230,6 +230,12 @@ pub(crate) enum PeerEvent { TransportConnected, /// Transport connect failed. TransportFailed, + /// The transport accepted the dial but sending a stored handshake + /// initiation failed. The machine marks the embedded leg failed so the + /// stale-connection sweep reclaims it, WITHOUT leaving the handshaking + /// state — the retransmit driver may still resend in the window before + /// the sweep. + HandshakeSendFailed, /// Inbound handshake msg1 processed shell-side (Noise + snapshot). InboundMsg1 { link: LinkId, @@ -566,6 +572,7 @@ impl PeerMachine { } => self.on_dial(transport_id, remote_addr, connection_oriented, now), PeerEvent::TransportConnected => self.on_transport_connected(now), PeerEvent::TransportFailed => self.on_transport_failed(now), + PeerEvent::HandshakeSendFailed => self.on_handshake_send_failed(), PeerEvent::InboundMsg1 { link, wire, est } => { self.on_inbound_msg1(link, wire, est, now, index_allocator) } @@ -668,6 +675,19 @@ impl PeerMachine { actions } + /// A stored handshake initiation failed to send: mark the embedded leg + /// failed so the stale-connection sweep (which reads the leg's + /// `is_failed`) reclaims it. NO state flip — the machine stays in + /// `Handshaking{SentMsg1}` so retransmit eligibility + /// (`is_handshaking_sent_msg1`) survives until the sweep, and no timer + /// actions are emitted. + fn on_handshake_send_failed(&mut self) -> Vec { + if let Some(leg) = self.leg.as_mut() { + leg.mark_failed(); + } + Vec::new() + } + /// Emit msg1 and arm the retransmit/timeout timers. The Noise msg1 /// construction and its index allocation are shell-side effects performed by /// the driver when it executes this action; an empty payload is emitted (see @@ -2396,6 +2416,53 @@ mod tests { ); } + // ---- Test 7e: HandshakeSendFailed marks the leg, keeps the state ------ + // A stored-msg1 send failure marks the embedded leg failed (the + // stale-connection sweep reads the leg's `is_failed`) WITHOUT leaving + // `Handshaking{SentMsg1}` — retransmit eligibility + // (`is_handshaking_sent_msg1`) must survive until the sweep — and emits + // no actions. On a machine with no leg it is a defensive no-op. + #[test] + fn handshake_send_failed_marks_leg_without_leaving_handshaking() { + let mut alloc = IndexAllocator::new(); + let peer = peer_identity(); + + // Dial-persisted outbound machine carrying a prepared leg, driven to + // Handshaking{SentMsg1} via the connectionless dial. + let mut m = PeerMachine::new_outbound(LinkId::new(1), peer, 0); + let _ = m.step( + PeerEvent::Dial { + transport_id: TransportId::new(1), + remote_addr: TransportAddr::from_string("127.0.0.1:9999"), + peer_identity: peer, + connection_oriented: false, + }, + 100, + &mut alloc, + ); + m.set_leg(PeerConnection::outbound(LinkId::new(1), peer, 100)); + assert!(m.is_handshaking_sent_msg1()); + assert!(!m.leg().expect("leg embedded").is_failed()); + + let actions = m.step(PeerEvent::HandshakeSendFailed, 200, &mut alloc); + assert_eq!(actions, Vec::new(), "HandshakeSendFailed emits no actions"); + assert!( + m.is_handshaking_sent_msg1(), + "retransmit eligibility survives a send failure" + ); + assert!( + m.leg().expect("leg retained").is_failed(), + "the leg carries the failed mark the sweep reads" + ); + + // With no leg (e.g. after take_leg) the event is a defensive no-op. + let _ = m.take_leg(); + let actions = m.step(PeerEvent::HandshakeSendFailed, 300, &mut alloc); + assert_eq!(actions, Vec::new()); + assert!(m.is_handshaking_sent_msg1()); + assert!(m.leg().is_none()); + } + // ---- Test 8: liveness -> LinkDeadSuspected -> ReportLost -------------- #[test] fn liveness_to_link_dead() {