Merge branch 'refactor-node' into refactor-node-next

Re-express the rekey-msg2 and cross-connection observation feeds onto
the XX cores. The rekey-msg2 observation lands inside the block that
installs the pending session (where XX also generates msg3), so it fires
only when the install actually ran. The cross-connection observation
sits at the outbound resolution arms and targets the promoted peer's
machine; the inbound session-swap effects already own their executor
path here and are untouched. Crypto effect bodies are byte-unchanged.

src/node/handlers/handshake.rs

@@ -5,12 +5,13 @@
 //! - msg2 (responder → initiator): responder identity + epoch + negotiation
 //! - msg3 (initiator → responder): initiator identity + epoch + negotiation
 
+use crate::NodeAddr;
 use crate::PeerIdentity;
 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::{PeerAction, PeerEvent, PeerMachine, TimerKind};
+use crate::peer::machine::{CrossConnOutcome, PeerAction, PeerEvent, PeerMachine, TimerKind};
 use crate::peer::{ActivePeer, PeerConnection};
 use crate::proto::fmp::wire::{Msg1Header, Msg2Header, Msg3Header, build_msg2, build_msg3};
 use crate::proto::fmp::{
@@ -19,10 +20,63 @@ use crate::proto::fmp::{
     decide_fmp_negotiation,
 };
 use crate::transport::{Link, LinkDirection, LinkId, ReceivedPacket};
+use crate::utils::index::SessionIndex;
 use std::time::Duration;
 use tracing::{debug, info, warn};
 
 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.
     ///
@@ -379,6 +433,7 @@ impl Node {
                     self.config().node.rate_limit.handshake_resend_interval_ms;
                 let msg3_now_ms = Self::now_ms();
 
+                let mut rekey_completed = false;
                 if let Some(peer) = self.peers.get_mut(&peer_node_addr) {
                     match peer.complete_rekey_msg2(noise_msg2) {
                         Ok((msg3_bytes, session, remote_epoch)) => {
@@ -469,6 +524,8 @@ impl Node {
                                     new_their_index = %header.sender_idx,
                                     "rekey-msg2 initiator: pending session set, awaiting K-bit cutover"
                                 );
+
+                                rekey_completed = true;
                             } else {
                                 // msg3 send failed — abandon rekey
                                 if let Some(idx) = peer.abandon_rekey() {
@@ -500,6 +557,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;
             }
@@ -685,6 +751,7 @@ impl Node {
                 }
             };
 
+            let mut cross_conn_outcome: Option<CrossConnOutcome> = None;
             if our_outbound_wins {
                 // We're the smaller node. Swap to outbound session + indices.
                 // The peer will keep their inbound session (complement of ours).
@@ -741,6 +808,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
@@ -766,6 +838,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

src/peer/machine.rs

@@ -214,6 +214,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`]) built shell-side;
 /// `now` is the `step` parameter, never duplicated here.
@@ -276,6 +290,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.
@@ -617,6 +636,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.
@@ -1006,6 +1026,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<PeerAction> {
+        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<PeerAction> {
@@ -2659,4 +2696,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);
+    }
 }