From e05b868cf8e80460a33753732612ed292230f784 Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Mon, 13 Jul 2026 16:20:45 +0000 Subject: [PATCH] node: drive rekey cadence cutover and drain through the peer machine Route each Cutover and Drain that the shell-side batch poll_rekey decides through the per-peer machine and the executor, replacing the inline effect bodies in check_rekey. The batch decision and its per-peer snapshots stay shell-side and byte-unchanged: poll_rekey phase-groups all cutovers, then all drains, then all initiations across the peer set, and that ordering governs the shared index allocator's free-then-allocate sequence that appears on the wire, so the machine only consumes the already-decided actions (a new RekeyConsume event) without re-deciding. InitiateRekey stays inline (its Noise msg1 build is a shell-side leaf) with a RekeyInitiated observation feeding the machine so its control state stays coherent for the next tick's cutover. The cutover and drain logs, which relocated into the executor in the prior commit, are pinned back to the fips::node::handlers::rekey tracing target so they stay visible under the operator's module log filter. Also clears the machine's shadow draining_index on drain so a later cross-connection resolution cannot double-free the already-freed index. --- src/node/dataplane/peer_actions.rs | 10 ++ src/node/handlers/rekey.rs | 230 ++++++++++++++++++++++------- src/peer/machine.rs | 137 +++++++++++++++++ 3 files changed, 320 insertions(+), 57 deletions(-) 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); + } }