diff --git a/src/node/handlers/handshake.rs b/src/node/handlers/handshake.rs index ee02865..e479612 100644 --- a/src/node/handlers/handshake.rs +++ b/src/node/handlers/handshake.rs @@ -7,7 +7,8 @@ use crate::node::dataplane::PeerActionCtx; use crate::node::reject::{HandshakeReject, RejectReason}; use crate::node::{Node, NodeError}; use crate::peer::machine::{ - FailReason, HandshakePhase, PeerAction, PeerEvent, PeerMachine, PeerState, TimerKind, + CrossConnOutcome, FailReason, HandshakePhase, PeerAction, PeerEvent, PeerMachine, PeerState, + TimerKind, }; use crate::peer::{ActivePeer, PeerConnection}; use crate::proto::fmp::wire::{Msg1Header, Msg2Header, build_msg2}; @@ -16,6 +17,7 @@ use crate::proto::fmp::{ OutboundSnapshot, PromotionResult, WireOutcome, cross_connection_winner, }; use crate::transport::{Link, LinkDirection, LinkId, ReceivedPacket}; +use crate::utils::index::SessionIndex; use std::time::Duration; use tracing::{debug, info, warn}; @@ -62,6 +64,58 @@ impl EstablishView for Node { } impl Node { + /// 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 + /// the machine coherent without an executor pass. + fn observe_rekey_msg2(&mut self, node_addr: &NodeAddr, their_index: SessionIndex) { + let link = match self.peers.get(node_addr) { + Some(peer) => peer.link_id(), + None => return, + }; + if let Some(machine) = self.peer_machines.get_mut(&link) { + let acts = machine.step( + PeerEvent::RekeyMsg2 { their_index }, + Self::now_ms(), + &mut self.index_allocator, + ); + debug_assert!(acts.is_empty(), "completed-rekey is a pure observation"); + } else { + debug_assert!( + false, + "peer machine present for every established rekey peer" + ); + } + } + + /// Feed the promoted peer's control machine the cross-connection resolution + /// after the inline session surgery. The obs reconciles the machine's shadow + /// session indices (updated on a swap, unchanged on a keep); it emits no + /// action, so a bare `step` keeps the machine coherent without an executor + /// pass. + fn observe_cross_conn_resolved(&mut self, node_addr: &NodeAddr, outcome: CrossConnOutcome) { + let link = match self.peers.get(node_addr) { + Some(peer) => peer.link_id(), + None => return, + }; + if let Some(machine) = self.peer_machines.get_mut(&link) { + let acts = machine.step( + PeerEvent::CrossConnResolved { outcome }, + Self::now_ms(), + &mut self.index_allocator, + ); + debug_assert!( + acts.is_empty(), + "cross-connection resolution is a pure observation" + ); + } else { + debug_assert!( + false, + "peer machine present for the promoted cross-connection peer" + ); + } + } + /// Returns true if an inbound msg1 should be admitted past the /// `accept_connections` gate. /// @@ -853,6 +907,7 @@ impl Node { let display_name = self.peer_display_name(&peer_node_addr); // Complete the rekey handshake on the ActivePeer + let mut rekey_completed = false; if let Some(peer) = self.peers.get_mut(&peer_node_addr) { match peer.complete_rekey_msg2(noise_msg2) { Ok((session, remote_epoch)) => { @@ -890,6 +945,7 @@ impl Node { new_their_index = %header.sender_idx, "Rekey completed (initiator), pending K-bit cutover" ); + rekey_completed = true; } Err(e) => { warn!( @@ -909,6 +965,15 @@ impl Node { } } + // Feed the control machine the completed-rekey observation so its + // shadow index and rekey phase stay coherent. Only on success — + // the failure path above reverts the rekey and leaves the machine + // untouched. The crypto effect already ran inline; this emits no + // action. + if rekey_completed { + self.observe_rekey_msg2(&peer_node_addr, header.sender_idx); + } + self.pending_outbound.remove(&key); return; } @@ -1026,6 +1091,7 @@ impl Node { } }; + let mut cross_conn_outcome: Option = None; if out_decision == OutboundDecision::CrossConnectionSwap { // We're the smaller node. Swap to outbound session + indices. // The peer will keep their inbound session (complement of ours). @@ -1078,6 +1144,11 @@ impl Node { new_their_index = %header.sender_idx, "Cross-connection: swapped to outbound session (our outbound wins)" ); + + cross_conn_outcome = Some(CrossConnOutcome::Swap { + our_index: outbound_our_index, + their_index: header.sender_idx, + }); } } else { // We're the larger node. Keep our inbound session (it pairs @@ -1103,6 +1174,18 @@ impl Node { if let Some(idx) = outbound_our_index { let _ = self.index_allocator.free(idx); } + + cross_conn_outcome = Some(CrossConnOutcome::Keep); + } + + // Feed the promoted peer's control machine the cross-connection + // resolution so its shadow session indices track the inline session + // surgery above (updated on a swap, unchanged on a keep). The + // outbound leg's machine was removed on entry, so this targets the + // still-live promoted peer's machine. The crypto effect already ran + // inline; this emits no action. + if let Some(outcome) = cross_conn_outcome { + self.observe_cross_conn_resolved(&peer_node_addr, outcome); } // Clean up outbound connection state diff --git a/src/peer/machine.rs b/src/peer/machine.rs index f741256..9d1b8f1 100644 --- a/src/peer/machine.rs +++ b/src/peer/machine.rs @@ -189,6 +189,20 @@ pub(crate) enum TimerKind { Liveness, } +/// Outcome of an outbound cross-connection resolution, observed by the control +/// machine after the shell has already applied the effect inline. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum CrossConnOutcome { + /// The outbound session replaced the existing inbound one: the control + /// shadow adopts the new local and remote session indices. + Swap { + our_index: SessionIndex, + their_index: SessionIndex, + }, + /// The existing inbound session was kept: the control shadow is unchanged. + Keep, +} + /// An input to the machine. Cross-registry facts ride in the payload as /// plain-data snapshots ([`WireOutcome`]/[`EstablishSnapshot`]/[`OutboundSnapshot`]) /// built shell-side; `now` is the `step` parameter, never duplicated here. @@ -248,6 +262,11 @@ pub(crate) enum PeerEvent { /// `Maintaining{Rekey(Msg1Sent)}` so the next tick's `Cutover`/`Drain` consume /// transitions from a coherent phase. Emits no action. RekeyInitiated, + /// OBSERVATION: the shell resolved an outbound cross-connection inline (a + /// session swap or keep, with the registry and index surgery already + /// applied). Reconciles the control shadow's session indices with reality + /// on a swap; leaves them untouched on a keep. Emits no action. + CrossConnResolved { outcome: CrossConnOutcome }, /// Data plane observed the responder K-bit flip inline. PeerKbitFlip { epoch: [u8; 8] }, /// A filter announce is due for this peer. @@ -521,6 +540,7 @@ impl PeerMachine { PeerEvent::RekeyMsg2 { their_index } => self.on_rekey_msg2(their_index), PeerEvent::RekeyConsume { action } => self.map_rekey_action(action, now), PeerEvent::RekeyInitiated => self.on_rekey_initiated(), + PeerEvent::CrossConnResolved { outcome } => self.on_cross_conn_resolved(outcome), PeerEvent::PeerKbitFlip { .. } => { // Responder cutover is data-plane-owned: the machine only // schedules the drain-window unregister. NO slot mutation. @@ -918,6 +938,23 @@ impl PeerMachine { Vec::new() } + /// Observation: the shell resolved an outbound cross-connection inline. On a + /// session swap it adopts the new local and remote session indices into the + /// control shadow, mirroring the shell's in-place session replacement; on a + /// keep it leaves the shadow untouched. Emits NO action — the crypto effect + /// already ran shell-side. + fn on_cross_conn_resolved(&mut self, outcome: CrossConnOutcome) -> Vec { + if let CrossConnOutcome::Swap { + our_index, + their_index, + } = outcome + { + self.conn.set_our_index(our_index); + self.conn.set_their_index(their_index); + } + Vec::new() + } + /// Rekey cadence: run `poll_rekey` over this one peer's snapshot and map the /// phase-grouped `ConnAction`s. fn on_rekey_cadence(&mut self, now: u64) -> Vec { @@ -2437,4 +2474,95 @@ mod tests { // No index allocation happened in the machine (shell-side leaf). assert_eq!(alloc.count(), 0); } + + // ---- Test 11: RekeyMsg2 observation ----------------------------------- + // The shell completed the initiated rekey inline; the obs records the peer's + // new index, clears the in-progress flag, advances to PendingCutover, and + // emits nothing. + #[test] + fn rekey_msg2_observation() { + let mut alloc = IndexAllocator::new(); + let id = peer_identity(); + let addr = *id.node_addr(); + let mut m = PeerMachine::new_outbound(LinkId::new(1), id, 0); + m.state = PeerState::Maintaining { + addr, + kind: MaintainKind::Rekey(RekeyPhase::Msg1Sent), + }; + m.rekey_in_progress = true; + + let their = SessionIndex::new(0x4444); + let acts = m.step( + PeerEvent::RekeyMsg2 { their_index: their }, + 6_000, + &mut alloc, + ); + assert!(acts.is_empty()); + assert_eq!(m.conn.their_index(), Some(their)); + assert!(!m.rekey_in_progress); + assert_eq!( + m.state(), + PeerState::Maintaining { + addr, + kind: MaintainKind::Rekey(RekeyPhase::PendingCutover) + } + ); + assert_eq!(alloc.count(), 0); + } + + // ---- Test 12: CrossConnResolved observation --------------------------- + // A swap adopts the new local and remote indices into the control shadow and + // emits nothing; a keep leaves the shadow untouched and emits nothing. + #[test] + fn cross_conn_resolved_swap_updates_shadow() { + let mut alloc = IndexAllocator::new(); + let id = peer_identity(); + let addr = *id.node_addr(); + let mut m = PeerMachine::new_outbound(LinkId::new(1), id, 0); + m.state = PeerState::Active { addr }; + m.conn.set_our_index(SessionIndex::new(0x1111)); + m.conn.set_their_index(SessionIndex::new(0x2222)); + + let our = SessionIndex::new(0xAAAA); + let their = SessionIndex::new(0xBBBB); + let acts = m.step( + PeerEvent::CrossConnResolved { + outcome: CrossConnOutcome::Swap { + our_index: our, + their_index: their, + }, + }, + 7_000, + &mut alloc, + ); + assert!(acts.is_empty()); + assert_eq!(m.conn.our_index(), Some(our)); + assert_eq!(m.conn.their_index(), Some(their)); + assert_eq!(m.state(), PeerState::Active { addr }); + assert_eq!(alloc.count(), 0); + } + + #[test] + fn cross_conn_resolved_keep_is_noop() { + let mut alloc = IndexAllocator::new(); + let id = peer_identity(); + let addr = *id.node_addr(); + let mut m = PeerMachine::new_outbound(LinkId::new(1), id, 0); + m.state = PeerState::Active { addr }; + m.conn.set_our_index(SessionIndex::new(0x1111)); + m.conn.set_their_index(SessionIndex::new(0x2222)); + + let acts = m.step( + PeerEvent::CrossConnResolved { + outcome: CrossConnOutcome::Keep, + }, + 7_000, + &mut alloc, + ); + assert!(acts.is_empty()); + assert_eq!(m.conn.our_index(), Some(SessionIndex::new(0x1111))); + assert_eq!(m.conn.their_index(), Some(SessionIndex::new(0x2222))); + assert_eq!(m.state(), PeerState::Active { addr }); + assert_eq!(alloc.count(), 0); + } }