diff --git a/src/node/dataplane/peer_actions.rs b/src/node/dataplane/peer_actions.rs index 719a379..e81b415 100644 --- a/src/node/dataplane/peer_actions.rs +++ b/src/node/dataplane/peer_actions.rs @@ -216,8 +216,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 { .. } => { @@ -467,6 +472,16 @@ impl Node { queue.extend(follow); } } + 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/node/handlers/handshake.rs b/src/node/handlers/handshake.rs index 213133a..0213537 100644 --- a/src/node/handlers/handshake.rs +++ b/src/node/handlers/handshake.rs @@ -16,7 +16,7 @@ use crate::peer::{ActivePeer, PeerConnection}; use crate::proto::fmp::wire::{Msg1Header, Msg2Header, Msg3Header, build_msg2, build_msg3}; use crate::proto::fmp::{ Disconnect, DisconnectReason, EstablishSnapshot, InboundDecision, InboundReject, - NegotiationPayload, PromotionResult, WireOutcome, cross_connection_winner, + NegotiationPayload, OutboundSnapshot, PromotionResult, WireOutcome, cross_connection_winner, decide_fmp_negotiation, }; use crate::transport::{Link, LinkDirection, LinkId, ReceivedPacket}; @@ -25,6 +25,22 @@ use std::time::Duration; use tracing::{debug, info, warn}; impl Node { + /// Snapshot the registry state the outbound establish decision reads about + /// `peer_addr`: whether the identity is already an active peer, and the + /// pre-evaluated cross-connection tie-break for THIS outbound connection + /// (`is_outbound = true`), resolved into a plain `bool` here so the core + /// stays free of the peer helper. + fn outbound_snapshot(&self, peer_addr: &NodeAddr) -> OutboundSnapshot { + OutboundSnapshot { + has_existing_peer: self.peers.contains_key(peer_addr), + our_outbound_wins: cross_connection_winner( + self.identity().node_addr(), + peer_addr, + true, + ), + } + } + /// Feed the peer's control machine the completed-rekey observation after the /// inline `complete_rekey_msg2`. The obs records the peer's new session index /// and advances the rekey phase; it emits no action, so a bare `step` keeps @@ -752,26 +768,82 @@ impl Node { // // This ensures both nodes use the same Noise handshake (the winner's // outbound = the loser's inbound). - if self.peers.contains_key(&peer_node_addr) { + // 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. + // + // Every outbound leg carries a persistent machine by now — identified + // dials persist one at dial, anonymous-discovery legs at leg birth in + // `start_handshake` — so the lookup is expected to hit, and the + // executor's `PromoteToActive` arm can feed `PromotionResolved` back + // via the same lookup. For an anonymous machine this is where its + // identity crystallizes: msg2 revealed who answered, and the learned + // identity lands on the machine before the step (a no-op for + // identified machines), so the Promote arm reads a crystallized + // address. The `pending_outbound` lifecycle stays shell-side — the + // machine never touches it. + let out_snap = self.outbound_snapshot(&peer_node_addr); + let actions = match self.peer_machines.get_mut(&link_id) { + Some(machine) => { + machine.crystallize_identity(peer_identity); + machine.step( + PeerEvent::OutboundMsg2 { + their_index: header.sender_idx, + out: out_snap, + }, + packet.timestamp_ms, + &mut self.index_allocator, + ) + } + None => { + // A miss is a state-machine inconsistency (e.g. a test seeding + // a connection/`pending_outbound` entry directly): rebuild the + // machine defensively and persist it, so the promotion feedback + // below still finds it and the promoted peer keeps a machine. + debug_assert!( + false, + "outbound leg {link_id} reached msg2 without a control machine" + ); + let mut machine = + PeerMachine::new_outbound(link_id, Some(peer_identity), packet.timestamp_ms); + let actions = machine.step( + PeerEvent::OutboundMsg2 { + 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) .and_then(|machine| machine.take_leg()); self.remove_peer_machine(link_id); - let our_outbound_wins = cross_connection_winner( - self.identity().node_addr(), - &peer_node_addr, - true, // this IS our outbound - ); let mut conn = match taken_conn { Some(c) => c, @@ -784,7 +856,7 @@ impl Node { }; let mut cross_conn_outcome: Option = None; - if our_outbound_wins { + 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(); @@ -906,65 +978,23 @@ impl Node { return; } - // Net-new outbound promote, driven by the per-peer machine. The peer-map - // membership test above was false (the cross-connection block returns on - // an existing peer), so this is the net-new path: promote_connection hits - // its normal-promotion branch and returns Promoted. + // === Net-new outbound establish, driven by the machine. === + // The machine's decision was `Promote` (`has_existing_peer == false` — + // the cross-connection block above returns otherwise), so + // `promote_connection` hits its normal-promotion branch and returns + // `Promoted`. The machine survives the promotion and the executor + // crystallizes its state via the `PromotionResolved` feedback. The + // promote tail (info log, tree/bloom/backoff, `pending_outbound` + // removal) lives in the executor's `PromoteToActive` arm. // - // Every outbound leg carries a persistent machine by now — identified - // dials persist one at dial, anonymous-discovery legs at leg birth in - // `start_handshake` — so the lookup is expected to hit. For an anonymous - // machine this is where its identity crystallizes: msg2 revealed who - // answered, and the learned identity lands on the machine before the - // step (a no-op for identified machines). The machine survives the - // promotion and the executor crystallizes its state via the - // `PromotionResolved` feedback. With no outbound decision core to run, - // the machine emits `PromoteToActive` unconditionally (the - // net-new-vs-cross-connection decision was the `peers.contains_key` - // test above, which returns on an existing peer). The promote tail - // (info log, tree/bloom/backoff, `pending_outbound` removal) lives in - // the executor's `PromoteToActive` arm. - let promote_actions = match self.peer_machines.get_mut(&link_id) { - Some(machine) => { - machine.crystallize_identity(peer_identity); - machine.step( - PeerEvent::OutboundMsg2 { - their_index: header.sender_idx, - }, - packet.timestamp_ms, - &mut self.index_allocator, - ) - } - None => { - // A miss is a state-machine inconsistency (e.g. a test seeding - // a connection/`pending_outbound` entry directly): rebuild the - // machine defensively and persist it, so the promotion feedback - // below still finds it and the promoted peer keeps a machine. - debug_assert!( - false, - "outbound leg {link_id} reached msg2 without a control machine" - ); - let mut machine = - PeerMachine::new_outbound(link_id, Some(peer_identity), packet.timestamp_ms); - let actions = machine.step( - PeerEvent::OutboundMsg2 { - their_index: header.sender_idx, - }, - 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 @@ -986,8 +1016,7 @@ impl Node { is_outbound: true, pending_outbound_key: Some(key), }; - self.execute_peer_actions(link_id, &ambient, promote_actions) - .await; + self.execute_peer_actions(link_id, &ambient, actions).await; } /// Handle handshake message 3 (phase 0x3). diff --git a/src/node/lifecycle/mod.rs b/src/node/lifecycle/mod.rs index ba4c9a7..4a6f1fb 100644 --- a/src/node/lifecycle/mod.rs +++ b/src/node/lifecycle/mod.rs @@ -793,10 +793,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, + current_time_ms, + &mut self.index_allocator, + ); + debug_assert!( + actions.is_empty(), + "HandshakeSendFailed must emit no actions" + ); } } } @@ -806,9 +815,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( @@ -816,6 +826,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(), @@ -842,10 +853,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 34e7a3a..52d9e61 100644 --- a/src/peer/machine.rs +++ b/src/peer/machine.rs @@ -54,12 +54,17 @@ //! triggers are live: `handle_msg3` steps the decision machine and the //! executor performs its verdict. //! -//! Likewise the outbound completion is NOT routed through the machine: next has -//! no `establish_outbound` core — `handle_msg2` learns the identity from -//! `conn.expected_identity()` and drives `promote_connection` inline. The -//! machine's outbound role is `Dial`→`OpenTransport`, `start_outbound_handshake`, -//! and the resend/timeout timers; the shell drives promote and feeds the outcome -//! back via [`PromotionResolved`](PeerEvent::PromotionResolved). +//! The outbound completion is routed through the machine the same way: +//! `handle_msg2` crystallizes the learned identity, builds an +//! [`OutboundSnapshot`], and steps the persistent machine once; the machine +//! computes the establish decision via `establish_outbound`. A net-new +//! completion promotes through [`PromoteToActive`](PeerAction::PromoteToActive) +//! (the executor feeds the outcome back via +//! [`PromotionResolved`](PeerEvent::PromotionResolved)), while a +//! cross-connection resolves as a +//! [`ResolveCrossConnection`](PeerAction::ResolveCrossConnection) decision the +//! shell intercepts — its inline Swap/Keep bodies mutate the already-promoted +//! peer, for which no `PeerAction` exists. //! //! ## Realizability notes //! @@ -81,7 +86,8 @@ use crate::peer::PeerConnection; use crate::proto::fmp::{ ConnAction, ConnSnapshot, ConnectionState, EstablishSnapshot, Fmp, InboundDecision, - InboundReject, PeerSnapshot, PromotionResult, RekeyCfg, RekeyResendSnapshot, WireOutcome, + InboundReject, OutboundDecision, OutboundSnapshot, PeerSnapshot, PromotionResult, RekeyCfg, + RekeyResendSnapshot, WireOutcome, }; use crate::proto::link::LinkMessageType; use crate::transport::{LinkId, TransportAddr, TransportId}; @@ -258,6 +264,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: allocate our index and reply /// with msg2, deferring identity/classification to msg3. Carries no identity /// or wire outcome — on XX neither is known at msg1. @@ -275,12 +287,16 @@ pub(crate) enum PeerEvent { our_index: SessionIndex, }, /// Outbound handshake msg2 completed (Noise finalized, identity crystallized - /// shell-side from the connection's expected identity, ACL already gated, - /// msg3 sent). The net-new-vs-cross-connection decision is the shell's - /// peer-map membership test; this event is emitted only on the net-new path, - /// so it maps directly to a promote. `their_index` carries the peer's session - /// index for shape parity with the inbound path. - OutboundMsg2 { their_index: SessionIndex }, + /// onto the machine shell-side, ACL already gated, msg3 sent). Carries the + /// shell-built [`OutboundSnapshot`]; the machine computes the establish + /// decision from it — a net-new promote, or a cross-connection swap/keep + /// conveyed as [`ResolveCrossConnection`](PeerAction::ResolveCrossConnection). + /// `their_index` carries the peer's session index for shape parity with the + /// inbound path. + OutboundMsg2 { + their_index: SessionIndex, + out: OutboundSnapshot, + }, /// `promote_connection` resolved the [`PromoteToActive`](PeerAction::PromoteToActive) /// action shell-side; the machine consumes the outcome (it does not /// re-decide the tie-break). @@ -359,6 +375,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] }, @@ -716,13 +737,16 @@ 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 } => self.on_inbound_msg1(link, now, index_allocator), PeerEvent::InboundMsg3 { wire, est, our_index, } => self.on_inbound_msg3(wire, est, our_index, now, index_allocator), - PeerEvent::OutboundMsg2 { their_index } => self.on_outbound_msg2(their_index, now), + PeerEvent::OutboundMsg2 { their_index, out } => { + self.on_outbound_msg2(their_index, out, now) + } PeerEvent::PromotionResolved { result } => self.on_promotion_resolved(result, now), PeerEvent::RekeyMsg2 { their_index } => self.on_rekey_msg2(their_index), PeerEvent::RekeyConsume { action } => self.map_rekey_action(action, now), @@ -813,6 +837,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 @@ -960,33 +997,63 @@ impl PeerMachine { } } - /// Net-new outbound promote after msg2 completes the initiator handshake. + /// Outbound completion after msg2 completes the initiator handshake: + /// 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. No state guard: anonymous-discovery + /// legs reach msg2 parked in `Discovered` (identified dials in + /// `Handshaking{SentMsg1}`), and the decision reads only the snapshot. /// - /// The net-new-vs-cross-connection decision is made shell-side (via the peer - /// map), so this handler is reached only on the net-new path and maps - /// directly to an unconditional promote — there is no sub-decision to run - /// here. The persistent `Established` machine is created inside - /// `promote_connection`; this machine is a transient decision vehicle that is - /// discarded after the step, so the `set_their_index` and state write below - /// are behaviorally inert (kept to make the transient's intent explicit and - /// to match the established lifecycle shape). - fn on_outbound_msg2(&mut self, their_index: SessionIndex, _now: u64) -> Vec { + /// On `Promote` the persistent machine survives as the active peer's + /// control machine, and `Established` is written here, ahead of the + /// `PromotionResolved` feedback (which re-crystallizes it). The shell + /// crystallizes the learned identity onto the machine before the step, so + /// the `addr()` read is `Some` for anonymous legs too. On Swap/Keep the + /// `set_their_index` write lands on the machine's shadow connection and + /// the shell discards the machine right after the step, so it is + /// unobservable. + fn on_outbound_msg2( + &mut self, + their_index: SessionIndex, + out: OutboundSnapshot, + _now: u64, + ) -> Vec { self.conn.set_their_index(their_index); - let addr = self.addr().unwrap_or_else(zero_addr); - self.state = PeerState::Established { addr }; - // The machine survives promotion (it becomes the active peer's control - // machine), so cancel the outbound handshake timers here or they would - // linger in the driver's store. A late fire would no-op against the - // non-`Handshaking` state, but leaving them armed is a timer leak. - vec![ - PeerAction::CancelTimer { - kind: TimerKind::HandshakeRetransmit, - }, - PeerAction::CancelTimer { - kind: TimerKind::HandshakeTimeout, - }, - PeerAction::PromoteToActive { link: self.link }, - ] + match Fmp::new().establish_outbound(&out) { + OutboundDecision::Promote => { + let addr = self.addr().unwrap_or_else(zero_addr); + self.state = PeerState::Established { addr }; + // The machine survives promotion (it becomes the active peer's + // control machine), so cancel the outbound handshake timers here + // or they would linger in the driver's store. A late fire would + // no-op against the non-`Handshaking` state, but leaving them + // armed is a timer leak. + vec![ + PeerAction::CancelTimer { + kind: TimerKind::HandshakeRetransmit, + }, + PeerAction::CancelTimer { + kind: TimerKind::HandshakeTimeout, + }, + PeerAction::PromoteToActive { link: self.link }, + ] + } + OutboundDecision::CrossConnectionSwap => { + // 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: 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 }] + } + } } /// XX inbound cross-connection at msg3. The Noise session swap and the @@ -1620,7 +1687,17 @@ mod tests { let mut m = PeerMachine::new_outbound(link, Some(id), 0); let their_index = SessionIndex::new(0x77); - let actions = m.step(PeerEvent::OutboundMsg2 { their_index }, 0, &mut alloc); + let actions = m.step( + PeerEvent::OutboundMsg2 { + their_index, + out: OutboundSnapshot { + has_existing_peer: false, + our_outbound_wins: false, + }, + }, + 0, + &mut alloc, + ); assert_eq!( actions, @@ -1650,7 +1727,17 @@ mod tests { let mut m = PeerMachine::new_outbound(link, Some(id), 0); let their_index = SessionIndex::new(0x88); - let actions = m.step(PeerEvent::OutboundMsg2 { their_index }, 0, &mut alloc); + let actions = m.step( + PeerEvent::OutboundMsg2 { + their_index, + out: OutboundSnapshot { + has_existing_peer: false, + our_outbound_wins: false, + }, + }, + 0, + &mut alloc, + ); assert_eq!( actions, @@ -1666,6 +1753,92 @@ mod tests { ); } + // ---- Test: outbound msg2 cross-connection decision arms --------------- + // Swap/Keep are decision-only: a single `ResolveCrossConnection { swap }` + // action, no index actions, no state change — the shell's inline + // resolution owns all effects and discards the machine after the step. + #[test] + fn outbound_msg2_cross_connection_decisions() { + let mut alloc = IndexAllocator::new(); + let peer = peer_identity(); + + // 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), Some(peer), 0); + m2.state = PeerState::Handshaking { + link: LinkId::new(2), + phase: HandshakePhase::SentMsg1, + }; + m2.conn.set_our_index(SessionIndex::new(0x2222)); // outbound index + m2.our_index = Some(SessionIndex::new(0x1111)); // old inbound index + let swap = m2.step( + PeerEvent::OutboundMsg2 { + their_index: SessionIndex::new(0x99), + out: OutboundSnapshot { + has_existing_peer: true, + our_outbound_wins: true, + }, + }, + 400, + &mut alloc, + ); + assert_eq!( + swap, + vec![PeerAction::ResolveCrossConnection { swap: true }] + ); + 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 -> decision only; the + // shell's inline resolution frees the unused outbound index. + let mut m3 = PeerMachine::new_outbound(LinkId::new(3), Some(peer), 0); + m3.state = PeerState::Handshaking { + link: LinkId::new(3), + phase: HandshakePhase::SentMsg1, + }; + m3.conn.set_our_index(SessionIndex::new(0x3333)); + let keep = m3.step( + PeerEvent::OutboundMsg2 { + their_index: SessionIndex::new(0x9A), + out: OutboundSnapshot { + has_existing_peer: true, + our_outbound_wins: false, + }, + }, + 500, + &mut alloc, + ); + assert_eq!( + keep, + 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); + } + // ---- Contract: actions are runtime-agnostic data ---------------------- // // The machine's emitted actions are the message contract between the sync @@ -1700,6 +1873,7 @@ mod tests { PeerAction::PromoteToActive { link: LinkId::new(7), }, + PeerAction::ResolveCrossConnection { swap: true }, PeerAction::SwapSendState { epoch: [1u8; 8] }, PeerAction::CompleteDrain { peer }, PeerAction::InvalidateSendState, @@ -1745,6 +1919,7 @@ mod tests { | PeerAction::SendRekey { .. } | PeerAction::SendLinkMessage { .. } | PeerAction::PromoteToActive { .. } + | PeerAction::ResolveCrossConnection { .. } | PeerAction::SwapSendState { .. } | PeerAction::CompleteDrain { .. } | PeerAction::InvalidateSendState @@ -2461,6 +2636,10 @@ mod tests { let promote = m.step( PeerEvent::OutboundMsg2 { their_index: SessionIndex::new(0x77), + out: OutboundSnapshot { + has_existing_peer: false, + our_outbound_wins: false, + }, }, 300, &mut alloc, @@ -2575,6 +2754,10 @@ mod tests { let promote = m.step( PeerEvent::OutboundMsg2 { their_index: SessionIndex::new(0x77), + out: OutboundSnapshot { + has_existing_peer: false, + our_outbound_wins: false, + }, }, 200, &mut alloc, @@ -2661,6 +2844,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), Some(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() { diff --git a/src/proto/fmp/core.rs b/src/proto/fmp/core.rs index 3043d20..b042380 100644 --- a/src/proto/fmp/core.rs +++ b/src/proto/fmp/core.rs @@ -266,6 +266,21 @@ pub(crate) struct EstablishSnapshot { pub our_node_addr: NodeAddr, } +/// A snapshot of the registry state the *outbound* establish decision reads +/// about the peer whose msg2 just completed our handshake, taken by the shell. +/// +/// Both fields are pre-evaluated shell-side (the tie-break is a pure function of +/// the two node addresses, resolved into a plain `bool` here) so the core never +/// touches live `Node` state or the `crate::peer` tie-break helper. +pub(crate) struct OutboundSnapshot { + /// The peer identity is already a promoted active peer — i.e. this outbound + /// completion is a cross-connection (we also processed their msg1). + pub has_existing_peer: bool, + /// Pre-evaluated cross-connection tie-break: our *outbound* connection wins + /// (we are the smaller NodeAddr). Only meaningful when `has_existing_peer`. + pub our_outbound_wins: bool, +} + /// A registry/transport effect the async shell performs on the core's behalf. /// /// The scaffold subset covers the maintain/teardown half of the lifecycle. The @@ -400,6 +415,30 @@ pub(crate) enum InboundReject { DualRekeyWon, } +/// The classification outcome for one outbound `handle_msg2` completion, decided +/// purely from the [`OutboundSnapshot`]. The shell matches on this and drives +/// the effects; the core consumes nothing and touches no live state. +/// +/// Only the case where the peer is *not* yet a promoted active peer is a plain +/// promotion; when it is, this msg2 completes the outbound half of a +/// cross-connection and the tie-break decides whether we swap our session to the +/// (winning) outbound one or keep our existing inbound session. The rekey-msg2 +/// completion path is handled by a separate shell driver (it mutates +/// `ActivePeer`, not a `PeerConnection`) and never reaches this decision. +#[derive(Debug, PartialEq, Eq)] +pub(crate) enum OutboundDecision { + /// No existing peer for this identity: promote the completed outbound + /// connection to an active peer via the normal promotion path. + Promote, + /// Cross-connection and our outbound wins (smaller NodeAddr): swap the peer + /// to the outbound session + indices, freeing the old inbound index. + CrossConnectionSwap, + /// Cross-connection and our outbound loses (larger NodeAddr): keep the + /// existing inbound session and original `their_index`, freeing the unused + /// outbound index. + CrossConnectionKeep, +} + impl Fmp { /// Decide the teardown choreography for the stale/failed connections the /// shell snapshotted. For each connection, an outbound one with a known @@ -631,6 +670,23 @@ impl Fmp { } } } + + /// Classify one outbound `handle_msg2` completion from the outbound snapshot. + /// Pure: reads only `snap`, mutates nothing. + /// + /// Mirrors the pre-refactor branch exactly: an existing same-identity peer + /// makes this a cross-connection resolved by the (pre-evaluated) tie-break — + /// swap on a win, keep on a loss — otherwise a net-new promote. + pub(crate) fn establish_outbound(&self, snap: &OutboundSnapshot) -> OutboundDecision { + if !snap.has_existing_peer { + return OutboundDecision::Promote; + } + if snap.our_outbound_wins { + OutboundDecision::CrossConnectionSwap + } else { + OutboundDecision::CrossConnectionKeep + } + } } /// Exponential-backoff schedule for the next handshake/rekey msg1 resend: diff --git a/src/proto/fmp/mod.rs b/src/proto/fmp/mod.rs index f259a04..231473b 100644 --- a/src/proto/fmp/mod.rs +++ b/src/proto/fmp/mod.rs @@ -38,7 +38,8 @@ mod tests; pub(crate) use core::{ ConnAction, ConnSnapshot, EstablishSnapshot, InboundDecision, InboundReject, LifecycleView, - PeerSnapshot, RekeyCfg, RekeyResendSnapshot, WireOutcome, decide_fmp_negotiation, + OutboundDecision, OutboundSnapshot, PeerSnapshot, RekeyCfg, RekeyResendSnapshot, WireOutcome, + decide_fmp_negotiation, }; pub use core::{PromotionResult, cross_connection_winner}; pub(crate) use limits::backoff_ms; diff --git a/src/proto/fmp/tests/core.rs b/src/proto/fmp/tests/core.rs index 7f6182d..395e965 100644 --- a/src/proto/fmp/tests/core.rs +++ b/src/proto/fmp/tests/core.rs @@ -5,8 +5,8 @@ use super::util::{ wire_outcome, }; use crate::proto::fmp::{ - ConnAction, Fmp, InboundDecision, InboundReject, NegotiationPayload, NodeProfile, RekeyCfg, - cross_connection_winner, + ConnAction, Fmp, InboundDecision, InboundReject, NegotiationPayload, NodeProfile, + OutboundDecision, OutboundSnapshot, RekeyCfg, cross_connection_winner, }; use crate::testutil::make_node_addr; use crate::transport::LinkId; @@ -539,6 +539,47 @@ fn establish_aged_unhealthy_is_duplicate() { )); } +// =========================================================================== +// establish_outbound — outbound msg2 classification (E4) +// =========================================================================== + +#[test] +fn establish_outbound_no_existing_peer_promotes() { + let fmp = Fmp::new(); + // our_outbound_wins is irrelevant when there is no existing peer. + let snap = OutboundSnapshot { + has_existing_peer: false, + our_outbound_wins: true, + }; + assert_eq!(fmp.establish_outbound(&snap), OutboundDecision::Promote); +} + +#[test] +fn establish_outbound_cross_connection_win_swaps() { + let fmp = Fmp::new(); + let snap = OutboundSnapshot { + has_existing_peer: true, + our_outbound_wins: true, + }; + assert_eq!( + fmp.establish_outbound(&snap), + OutboundDecision::CrossConnectionSwap + ); +} + +#[test] +fn establish_outbound_cross_connection_loss_keeps() { + let fmp = Fmp::new(); + let snap = OutboundSnapshot { + has_existing_peer: true, + our_outbound_wins: false, + }; + assert_eq!( + fmp.establish_outbound(&snap), + OutboundDecision::CrossConnectionKeep + ); +} + // ===== cross_connection_winner tie-break tests ===== #[test]