diff --git a/src/node/dataplane/peer_actions.rs b/src/node/dataplane/peer_actions.rs index b879ce1..61216e4 100644 --- a/src/node/dataplane/peer_actions.rs +++ b/src/node/dataplane/peer_actions.rs @@ -345,6 +345,12 @@ impl Node { "peers_by_index should contain pre-registered new index after cutover" ); debug!( + // Pin the target to the pre-refactor module: this + // cutover log relocated from handlers/rekey.rs into + // the executor, but operators (and the test harness) + // filter it under fips::node::handlers::rekey. Keeping + // the target preserves the observable log contract. + target: "fips::node::handlers::rekey", peer = %self.peer_display_name(&node_addr), "Rekey cutover complete (initiator), K-bit flipped" ); @@ -385,6 +391,10 @@ impl Node { } let _ = self.index_allocator.free(old_our_index); trace!( + // Pin to the pre-refactor module (see the cutover log + // above) so the relocated drain log stays visible under + // the operator's fips::node::handlers::rekey filter. + target: "fips::node::handlers::rekey", peer = %self.peer_display_name(&node_addr), old_index = %old_our_index, "Drain complete, previous session erased" diff --git a/src/node/handlers/rekey.rs b/src/node/handlers/rekey.rs index b8b309a..57e8f7d 100644 --- a/src/node/handlers/rekey.rs +++ b/src/node/handlers/rekey.rs @@ -7,13 +7,16 @@ use crate::NodeAddr; use crate::node::Node; +use crate::node::dataplane::PeerActionCtx; use crate::noise::HandshakeState; +use crate::peer::machine::PeerEvent; use crate::proto::fmp::wire::build_msg1; use crate::proto::fmp::{ConnAction, LifecycleView, PeerSnapshot, RekeyCfg, RekeyResendSnapshot}; use crate::proto::fsp::{ FspAction, RekeyMsg3ResendSnapshot, SessionSetup, SessionSnapshot, cutover_timer_elapsed, }; use crate::proto::link::SessionDatagram; +use crate::transport::{TransportAddr, TransportId}; use tracing::{debug, trace, warn}; /// Keep previous session alive for this long after cutover. @@ -46,72 +49,36 @@ impl Node { // The shell snapshots each healthy peer's rekey ages/flags (every clock // read resolved here); the core decides cutover/drain/trigger with no // clock, phase-grouped to preserve the pre-refactor execution order. + // The batch `poll_rekey` + snapshots STAY SHELL-SIDE and BYTE-UNCHANGED + // (Finding B): the cross-peer phase-grouping (all Cutover → all Drain → + // all InitiateRekey) governs the shared `index_allocator` free-then-alloc + // SEQUENCE that appears on the wire. The machine must NOT re-poll; it + // CONSUMES each decided `ConnAction` in the same order the batch returned. let snapshots = self.rekey_peers(); for action in self.fmp.poll_rekey(snapshots, &cfg) { match action { - // Execute cutover for initiator side. + // Initiator cutover: route the decided action through the peer + // machine + executor (C4-1). The executor's `SwapSendState` arm + // reproduces the pre-refactor cutover body EXACTLY. ConnAction::Cutover { peer: node_addr } => { - let did_cutover = if let Some(peer) = self.peers.get_mut(&node_addr) { - if let Some(_old_our_index) = peer.cutover_to_new_session() { - // New index was pre-registered in peers_by_index - // during msg2 handling (handshake.rs). - debug_assert!( - peer.transport_id().is_some() - && peer.our_index().is_some() - && self.peers_by_index.contains_key(&( - peer.transport_id().unwrap(), - peer.our_index().unwrap().as_u32() - )), - "peers_by_index should contain pre-registered new index after cutover" - ); - debug!( - peer = %self.peer_display_name(&node_addr), - "Rekey cutover complete (initiator), K-bit flipped" - ); - true - } else { - false - } - } else { - false - }; - // Re-register the new session with the decrypt worker — the - // cache_key (transport_id, our_index) just changed, so the - // old worker entry is stale and every packet on the new - // session would miss the worker's HashMap lookup. - #[cfg(unix)] - if did_cutover { - self.register_decrypt_worker_session(&node_addr); - } - #[cfg(not(unix))] - let _ = did_cutover; + self.route_rekey_cadence(node_addr, ConnAction::Cutover { peer: node_addr }) + .await; } - // Execute drain completion. + // Drain completion: route through the machine + executor. The + // executor's `CompleteDrain` arm reads the REAL previous index + // from `complete_drain()` and frees it at the same point the old + // inline body did (index-order preserving). ConnAction::Drain { peer: node_addr } => { - // Extract the old index and transport_id under the peer - // borrow, then drop the borrow so the cache_key cleanup - // below can take &mut self for unregister_decrypt_worker_session. - let drained = self.peers.get_mut(&node_addr).and_then(|peer| { - peer.complete_drain().map(|idx| (idx, peer.transport_id())) - }); - if let Some((old_our_index, transport_id)) = drained { - if let Some(tid) = transport_id { - let cache_key = (tid, old_our_index.as_u32()); - self.peers_by_index.remove(&cache_key); - #[cfg(unix)] - self.unregister_decrypt_worker_session(cache_key); - } - let _ = self.index_allocator.free(old_our_index); - trace!( - peer = %self.peer_display_name(&node_addr), - old_index = %old_our_index, - "Drain complete, previous session erased" - ); - } + self.route_rekey_cadence(node_addr, ConnAction::Drain { peer: node_addr }) + .await; } - // Initiate a new rekey. + // Initiate a new rekey: STAYS INLINE (the Noise msg1 build + + // index allocation are a shell-side leaf, byte-unchanged). Feed + // the machine a `RekeyInitiated` observation afterward so its + // control state stays coherent for the next tick's Cutover/Drain. ConnAction::InitiateRekey { peer: node_addr } => { self.initiate_rekey(&node_addr).await; + self.observe_rekey_initiated(&node_addr); } #[allow(unreachable_patterns)] _ => {} @@ -119,6 +86,155 @@ impl Node { } } + /// Route a cadence-decided `Cutover`/`Drain` `ConnAction` through the peer + /// machine + executor (C4-1). The shell already decided (batch `poll_rekey`); + /// the machine consumes via [`PeerEvent::RekeyConsume`] WITHOUT re-polling, + /// preserving the phase order. The `SwapSendState`/`CompleteDrain` executor + /// arms reproduce the pre-refactor inline effect bodies exactly. + /// + /// Finding A: an established peer always has a `peer_machine`. If the peer + /// vanished between snapshot and effect, the old inline body was a no-op, so + /// we do nothing; if the machine is absent (impossible per Finding A) we fall + /// back to the byte-identical inline body under a `debug_assert`. + async fn route_rekey_cadence(&mut self, node_addr: NodeAddr, action: ConnAction) { + let link = match self.peers.get(&node_addr) { + Some(peer) => peer.link_id(), + None => return, + }; + if !self.peer_machines.contains_key(&link) { + debug_assert!( + false, + "peer machine present for every established rekey peer (Finding A)" + ); + match action { + ConnAction::Cutover { peer } => self.cutover_peer_inline(&peer), + ConnAction::Drain { peer } => self.drain_peer_inline(&peer), + _ => {} + } + return; + } + let ambient = self.rekey_cadence_ctx(&node_addr); + self.advance_peer_machine( + link, + PeerEvent::RekeyConsume { action }, + Self::now_ms(), + &ambient, + ) + .await; + } + + /// Feed the machine the `RekeyInitiated` observation after the inline + /// `initiate_rekey` (C4-1). The obs emits no action, so there is no executor + /// pass — a bare `step` keeps the machine's control state coherent. + fn observe_rekey_initiated(&mut self, node_addr: &NodeAddr) { + 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::RekeyInitiated, + Self::now_ms(), + &mut self.index_allocator, + ); + debug_assert!(acts.is_empty(), "RekeyInitiated is a pure observation"); + } else { + debug_assert!( + false, + "peer machine present for every established rekey peer (Finding A)" + ); + } + } + + /// Ambient shell facts for the routed cadence Cutover/Drain step. Only + /// `verified_identity` is read by the `SwapSendState`/`CompleteDrain` + /// executor arms — `SwapSendState` resolves its `NodeAddr` from it (so it must + /// equal `node_addr`), and `CompleteDrain` carries its peer in the action + /// payload. The transport/index/direction fields are unused by these two arms + /// (they matter only to `PromoteToActive`, never emitted on this path) and are + /// populated best-effort for coherence. + fn rekey_cadence_ctx(&self, node_addr: &NodeAddr) -> PeerActionCtx { + let peer = &self.peers[node_addr]; + PeerActionCtx { + verified_identity: *peer.identity(), + transport_id: peer.transport_id().unwrap_or_else(|| TransportId::new(0)), + remote_addr: peer + .current_addr() + .cloned() + .unwrap_or_else(|| TransportAddr::new(Vec::new())), + our_index: peer.our_index(), + their_index: peer.their_index(), + now_ms: Self::now_ms(), + is_outbound: false, + } + } + + /// Pre-refactor initiator cutover body, retained as the release fallback for + /// the (Finding-A-impossible) missing-machine case. Byte-identical to the old + /// inline `ConnAction::Cutover` arm and to the executor's `SwapSendState` arm. + fn cutover_peer_inline(&mut self, node_addr: &NodeAddr) { + let did_cutover = if let Some(peer) = self.peers.get_mut(node_addr) { + if let Some(_old_our_index) = peer.cutover_to_new_session() { + // New index was pre-registered in peers_by_index during msg2 + // handling (handshake.rs). + debug_assert!( + peer.transport_id().is_some() + && peer.our_index().is_some() + && self.peers_by_index.contains_key(&( + peer.transport_id().unwrap(), + peer.our_index().unwrap().as_u32() + )), + "peers_by_index should contain pre-registered new index after cutover" + ); + debug!( + peer = %self.peer_display_name(node_addr), + "Rekey cutover complete (initiator), K-bit flipped" + ); + true + } else { + false + } + } else { + false + }; + // Re-register the new session with the decrypt worker — the cache_key + // (transport_id, our_index) just changed, so the old worker entry is + // stale and every packet on the new session would miss the lookup. + #[cfg(unix)] + if did_cutover { + self.register_decrypt_worker_session(node_addr); + } + #[cfg(not(unix))] + let _ = did_cutover; + } + + /// Pre-refactor drain-completion body, retained as the release fallback for + /// the (Finding-A-impossible) missing-machine case. Byte-identical to the old + /// inline `ConnAction::Drain` arm and to the executor's `CompleteDrain` arm. + fn drain_peer_inline(&mut self, node_addr: &NodeAddr) { + // Extract the old index and transport_id under the peer borrow, then drop + // the borrow so the cache_key cleanup below can take &mut self for + // unregister_decrypt_worker_session. + let drained = self + .peers + .get_mut(node_addr) + .and_then(|peer| peer.complete_drain().map(|idx| (idx, peer.transport_id()))); + if let Some((old_our_index, transport_id)) = drained { + if let Some(tid) = transport_id { + let cache_key = (tid, old_our_index.as_u32()); + self.peers_by_index.remove(&cache_key); + #[cfg(unix)] + self.unregister_decrypt_worker_session(cache_key); + } + let _ = self.index_allocator.free(old_our_index); + trace!( + peer = %self.peer_display_name(node_addr), + old_index = %old_our_index, + "Drain complete, previous session erased" + ); + } + } + /// Snapshot every healthy peer with a session for the rekey decision, /// pre-computing its monotonic ages and timer predicates so the pure core /// applies the thresholds without reading a clock (see [`PeerSnapshot`]). diff --git a/src/peer/machine.rs b/src/peer/machine.rs index 61b358a..b7f3bf4 100644 --- a/src/peer/machine.rs +++ b/src/peer/machine.rs @@ -224,6 +224,19 @@ pub(crate) enum PeerEvent { }, /// Inbound rekey msg2 (completes our initiated rekey). RekeyMsg2 { their_index: SessionIndex }, + /// A cadence-decided rekey `ConnAction` to CONSUME (C4-1). The shell ran the + /// batch `poll_rekey` across the whole peer set (phase-grouped, index-order + /// preserving — Finding B) and routes each decided action here; the machine + /// applies the control-tier transition + emits the send-state write + /// (`SwapSendState`/`CompleteDrain`) WITHOUT re-polling. Carries only + /// `Cutover`/`Drain` on the C4-1 path (`InitiateRekey` stays inline shell-side + /// with a [`RekeyInitiated`](PeerEvent::RekeyInitiated) observation). + RekeyConsume { action: ConnAction }, + /// OBSERVATION: the shell initiated an outbound rekey inline (the Noise msg1 + /// leaf + index allocation are shell-side). Advances the control state to + /// `Maintaining{Rekey(Msg1Sent)}` so the next tick's `Cutover`/`Drain` consume + /// transitions from a coherent phase. Emits no action. + RekeyInitiated, /// Data plane observed the responder K-bit flip inline (§3.7). PeerKbitFlip { epoch: [u8; 8] }, /// A filter announce is due for this peer. @@ -448,6 +461,8 @@ impl PeerMachine { self.on_inbound_msg1(self.link, wire, est, now, index_allocator) } 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::PeerKbitFlip { .. } => { // Responder cutover is data-plane-owned (§3.7): the machine only // schedules the drain-window unregister. NO slot mutation. @@ -789,6 +804,31 @@ impl PeerMachine { Vec::new() } + /// OBS (C4-1): the shell ran `initiate_rekey` inline — the Noise msg1 leaf, + /// the index allocation, the wire send, and the `set_rekey_state` on the + /// `ActivePeer` all happened shell-side. This is a pure observation that + /// advances the machine's control state to `Maintaining{Rekey(Msg1Sent)}` so + /// the subsequent cadence `Cutover`/`Drain` consume transitions from a + /// coherent phase. Emits NO action (nothing left to do). No-op unless the peer + /// is in an established-like state (defensive; the shell only initiates on + /// healthy established peers). + fn on_rekey_initiated(&mut self) -> Vec { + let addr = match self.addr() { + Some(a) => a, + None => return Vec::new(), + }; + if !self.is_established_context() { + return Vec::new(); + } + self.rekey_in_progress = true; + self.rekey_resend_count = 0; + self.state = PeerState::Maintaining { + addr, + kind: MaintainKind::Rekey(RekeyPhase::Msg1Sent), + }; + 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 { @@ -838,6 +878,14 @@ impl PeerMachine { // `ActivePeer::complete_drain` and does the peers_by_index / // decrypt-worker / index-free cleanup, replacing the old // shadow-index emission (which could drift from the real index). + // + // Clear the shadow `draining_index` set by the Cutover arm: the + // real previous index is now retired by `CompleteDrain`, so a + // leftover `Some(stale)` would double-free if a later + // `CrossConnectionWon` consumed it in `on_promotion_resolved` + // (C4-0 latent item 1). Post-rekey cross-connection promotion is + // not a live path, but clearing here removes the hazard outright. + self.draining_index = None; self.state = PeerState::Active { addr: peer }; vec![PeerAction::CompleteDrain { peer }] } @@ -1835,4 +1883,93 @@ mod tests { // guarantee: loss is reported only via ReportLost, and no retry-schedule // action exists in the PeerAction vocabulary at all (reconciler-owned). } + + // ---- Test 9: cadence CONSUME (C4-1) ----------------------------------- + // The shell polls the batch `poll_rekey` and routes each decided ConnAction + // as `RekeyConsume` — the machine maps it WITHOUT re-polling, yielding the + // same action sequence + transition as the machine-driven cadence (Test 1), + // and the Drain consume clears the shadow `draining_index`. + #[test] + fn rekey_consume_cutover_then_drain() { + 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::PendingCutover), + }; + m.rekey_our_index = Some(SessionIndex::new(0x2222)); + m.our_index = Some(SessionIndex::new(0x1111)); + m.remote_epoch = Some([9u8; 8]); + + // Consume the shell-decided Cutover: identical sequence to Test 1. + let cut = m.step( + PeerEvent::RekeyConsume { + action: ConnAction::Cutover { peer: addr }, + }, + 10_000, + &mut alloc, + ); + assert_eq!( + cut, + vec![ + PeerAction::SwapSendState { epoch: [9u8; 8] }, + PeerAction::RegisterDecryptSession { + index: SessionIndex::new(0x2222) + }, + PeerAction::SetTimer { + kind: TimerKind::DrainExpiry, + at_ms: 10_000 + DRAIN_WINDOW_MS + }, + ] + ); + assert_eq!( + m.state(), + PeerState::Maintaining { + addr, + kind: MaintainKind::Rekey(RekeyPhase::Draining) + } + ); + // Cutover stashed the old index in the drain shadow. + assert_eq!(m.draining_index, Some(SessionIndex::new(0x1111))); + + // Consume the shell-decided Drain: single CompleteDrain, Active, and the + // shadow drain index is CLEARED (double-free guard, C4-0 latent item 1). + let drain = m.step( + PeerEvent::RekeyConsume { + action: ConnAction::Drain { peer: addr }, + }, + 20_000, + &mut alloc, + ); + assert_eq!(drain, vec![PeerAction::CompleteDrain { peer: addr }]); + assert_eq!(m.state(), PeerState::Active { addr }); + assert_eq!(m.draining_index, None); + } + + // ---- Test 10: RekeyInitiated observation (C4-1) ----------------------- + // The shell ran `initiate_rekey` inline; the obs advances control state to + // Msg1Sent and emits nothing. + #[test] + fn rekey_initiated_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::Established { addr }; + + let acts = m.step(PeerEvent::RekeyInitiated, 5_000, &mut alloc); + assert!(acts.is_empty()); + assert_eq!( + m.state(), + PeerState::Maintaining { + addr, + kind: MaintainKind::Rekey(RekeyPhase::Msg1Sent) + } + ); + assert!(m.rekey_in_progress); + // No index allocation happened in the machine (shell-side leaf). + assert_eq!(alloc.count(), 0); + } }