From 2fad6be5e6ba7e17b93b99a7daa7abe3f450f0ef Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Wed, 15 Jul 2026 23:12:48 +0000 Subject: [PATCH] node: execute inbound session-swap effects through the per-peer machine MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Move the two Noise-XX inbound establish effects — the cross-connection session swap and the rekey-respond session install — out of the inline handle_msg3 arms and into their per-peer machine executor actions (SwapToInboundSession, RekeyRespondTrigger). The inbound establish decision now falls through to the shared machine-drive arm, matching the promote/restart arms. Carry the session index on the InboundMsg3 event and seed it into the transient machine before dispatch. The swap actions build their index from the machine's connection state, which is unset on a fresh inbound transient; without the seed the actions would emit empty and the swap would silently no-op while leaking the allocated index. Give the inbound reject and msg2-resend machine arms their terminal Failed state and index free, matching the inline arms that own those paths, so a later timer cannot free the index or report loss for a peer whose handshake has already resolved. Preserve exactly: the three distinct index frees and their paired index and pending-outbound map removals; the link-teardown asymmetry (the cross-connection path drops the address mapping, the rekey-respond path keeps it); the session take-then-install ordering; and every reject kind on the error sub-paths. --- src/node/dataplane/peer_actions.rs | 176 ++++++++++++++++--- src/node/handlers/handshake.rs | 190 +++------------------ src/peer/machine.rs | 263 ++++++++++++++++++++++++++--- 3 files changed, 419 insertions(+), 210 deletions(-) diff --git a/src/node/dataplane/peer_actions.rs b/src/node/dataplane/peer_actions.rs index 5ea6b5c..1db6c42 100644 --- a/src/node/dataplane/peer_actions.rs +++ b/src/node/dataplane/peer_actions.rs @@ -116,7 +116,6 @@ impl Node { ambient: &PeerActionCtx, actions: Vec, ) { - let _ = link; let mut queue: VecDeque = actions.into(); while let Some(action) = queue.pop_front() { match action { @@ -525,30 +524,161 @@ impl Node { // double-free. self.remove_active_peer(ambient.verified_identity.node_addr()); } - PeerAction::SwapToInboundSession { .. } => { - // DEFERRED: XX inbound cross-connection resolved at msg3. The - // session swap (`take_session`/`replace_session`, - // `peers_by_index` surgery, free the peer's OLD index — or free - // `our_index` on the losing side) is written, wired, and - // ci-local-validated when the inbound establish path is wired. - // Unreachable now (nothing drives the machine). - debug_assert!( - false, - "SwapToInboundSession executor body lands when inbound establish is wired" - ); + PeerAction::SwapToInboundSession { + peer, + our_index, + our_inbound_wins, + } => { + // Simultaneous-init cross-connection resolved at msg3 (msg2-then- + // msg3 ordering): apply the same tie-breaker the inverse ordering + // uses so both sides converge on a single Noise session pair. + let their_index = ambient + .their_index + .expect("cross-connection swap carries the peer session index"); + if our_inbound_wins { + // Larger node side: swap to the inbound session so it pairs + // with the peer's kept outbound session. + let inbound_session = match self + .connections + .get_mut(&link) + .and_then(|c| c.take_session()) + { + Some(s) => s, + None => { + self.connections.remove(&link); + self.remove_link(&link); + self.stats_mut().record_reject(RejectReason::Handshake( + HandshakeReject::BadState, + )); + return; + } + }; + if let Some(peer_ref) = self.peers.get_mut(&peer) { + let old_our_index = + peer_ref.replace_session(inbound_session, our_index, their_index); + let Some(transport_id) = peer_ref.transport_id() else { + self.connections.remove(&link); + self.remove_link(&link); + self.stats_mut().record_reject(RejectReason::Handshake( + HandshakeReject::BadState, + )); + return; + }; + if let Some(old_idx) = old_our_index { + self.peers_by_index + .remove(&(transport_id, old_idx.as_u32())); + let _ = self.index_allocator.free(old_idx); + } + self.peers_by_index + .insert((transport_id, our_index.as_u32()), peer); + + debug!( + peer = %self.peer_display_name(&peer), + new_our_index = %our_index, + new_their_index = %their_index, + "Simultaneous-init (msg3): swapped to inbound session (our inbound wins)" + ); + } + } else { + // Smaller node side: keep the existing outbound session, drop + // the inbound leg's allocated index. + let _ = self.index_allocator.free(our_index); + debug!( + peer = %self.peer_display_name(&peer), + "Simultaneous-init (msg3): keeping outbound session (our outbound wins)" + ); + } + + // Both branches tear down the temporary inbound link fully + // (including its `addr_to_link` mapping) via `remove_link`. + self.connections.remove(&link); + self.remove_link(&link); + return; } - PeerAction::RekeyRespondTrigger { .. } => { - // DEFERRED: XX rekey-responder resolved at msg3. The session - // move (`abandon_rekey` + index/`peers_by_index`/ - // `pending_outbound` cleanup, then `take_session` + - // `set_pending_session` + `record_peer_rekey` + - // `peers_by_index.insert`) is written, wired, and - // ci-local-validated when the inbound establish path is wired. - // Unreachable now (nothing drives the machine). - debug_assert!( - false, - "RekeyRespondTrigger executor body lands when inbound establish is wired" + PeerAction::RekeyRespondTrigger { + peer, + our_index, + abandon_first, + } => { + // Rekey-responder resolved at msg3: store the new session as + // pending on the existing peer, awaiting the K-bit cutover. + let their_index = ambient + .their_index + .expect("rekey-responder trigger carries the peer session index"); + if abandon_first { + // We lose the dual-rekey tie-break (larger addr): abandon our + // own rekey/pending and fall through as responder. + // `abandon_rekey` clears both the in-progress flag and any + // pending session state, returning whichever index needs + // freeing. + info!( + peer = %self.peer_display_name(&peer), + our_addr = %self.identity().node_addr(), + their_addr = %peer, + "rekey-msg3 tie-break: we lose (larger addr), abandon ours" + ); + if let Some(peer_ref) = self.peers.get_mut(&peer) + && let Some(idx) = peer_ref.abandon_rekey() + { + if let Some(tid) = peer_ref.transport_id() { + self.peers_by_index.remove(&(tid, idx.as_u32())); + self.pending_outbound.remove(&(tid, idx.as_u32())); + } + let _ = self.index_allocator.free(idx); + } + } + + // Rekey: process as responder, store new session as pending. + let noise_session = { + let Some(conn) = self.connections.get_mut(&link) else { + warn!(link_id = %link, "Connection removed during rekey msg3 processing"); + self.links.remove(&link); + self.stats_mut().record_reject(RejectReason::Handshake( + HandshakeReject::UnknownConnection, + )); + return; + }; + conn.take_session() + }; + let our_new_index = our_index; + + let noise_session = match noise_session { + Some(s) => s, + None => { + warn!("Rekey msg3: no session from handshake"); + self.connections.remove(&link); + self.links.remove(&link); + self.stats_mut() + .record_reject(RejectReason::Handshake(HandshakeReject::BadState)); + return; + } + }; + + // Store pending session on the existing peer + if let Some(peer_ref) = self.peers.get_mut(&peer) { + peer_ref.set_pending_session(noise_session, our_new_index, their_index); + peer_ref.record_peer_rekey(); + } + + // Register new index in peers_by_index + self.peers_by_index + .insert((ambient.transport_id, our_new_index.as_u32()), peer); + + // Clean up: remove the temporary connection/link. Do NOT remove + // addr_to_link — the entry must remain pointing to the original + // link so the established peer stays routable, so this uses the + // bare `links.remove` rather than the full `remove_link`. + self.connections.remove(&link); + self.links.remove(&link); + + debug!( + peer = %self.peer_display_name(&peer), + our_addr = %self.identity().node_addr(), + new_our_index = %our_new_index, + new_their_index = %their_index, + "rekey-msg3 responder: pending session set, awaiting K-bit cutover" ); + return; } PeerAction::RegisterDecryptSession { index } => { let _ = index; diff --git a/src/node/handlers/handshake.rs b/src/node/handlers/handshake.rs index 8492d72..3951fcc 100644 --- a/src/node/handlers/handshake.rs +++ b/src/node/handlers/handshake.rs @@ -1157,177 +1157,41 @@ impl Node { self.links.remove(&link_id); return; } - InboundDecision::CrossConnect { - peer, - our_inbound_wins, - } => { - debug_assert_eq!(peer, peer_node_addr); - // Simultaneous-init cross-connection (msg2-then-msg3 ordering): - // apply the same tie-breaker handle_msg2 uses for the inverse - // ordering so both sides converge on a single Noise session pair. - if our_inbound_wins { - // Larger node side: swap to the inbound session so it pairs - // with the peer's kept outbound session. - let inbound_session = match self - .connections - .get_mut(&link_id) - .and_then(|c| c.take_session()) - { - Some(s) => s, - None => { - self.connections.remove(&link_id); - self.remove_link(&link_id); - self.stats_mut() - .record_reject(RejectReason::Handshake(HandshakeReject::BadState)); - return; - } - }; - if let Some(peer) = self.peers.get_mut(&peer_node_addr) { - let old_our_index = - peer.replace_session(inbound_session, our_index, header.sender_idx); - let Some(transport_id) = peer.transport_id() else { - self.connections.remove(&link_id); - self.remove_link(&link_id); - self.stats_mut() - .record_reject(RejectReason::Handshake(HandshakeReject::BadState)); - return; - }; - if let Some(old_idx) = old_our_index { - self.peers_by_index - .remove(&(transport_id, old_idx.as_u32())); - let _ = self.index_allocator.free(old_idx); - } - self.peers_by_index - .insert((transport_id, our_index.as_u32()), peer_node_addr); - - debug!( - peer = %self.peer_display_name(&peer_node_addr), - new_our_index = %our_index, - new_their_index = %header.sender_idx, - "Simultaneous-init (msg3): swapped to inbound session (our inbound wins)" - ); - } - } else { - // Smaller node side: keep the existing outbound session, drop - // the inbound's allocated index. - let _ = self.index_allocator.free(our_index); - debug!( - peer = %self.peer_display_name(&peer_node_addr), - "Simultaneous-init (msg3): keeping outbound session (our outbound wins)" - ); - } - - self.connections.remove(&link_id); - self.remove_link(&link_id); - return; - } - InboundDecision::RekeyRespond { - peer, - abandon_first, - } => { - debug_assert_eq!(peer, peer_node_addr); - if abandon_first { - // We lose — abandon our rekey/pending, fall through as - // responder. abandon_rekey clears both rekey_in_progress and - // any pending session state, returning whichever index needs - // freeing. - info!( - peer = %self.peer_display_name(&peer_node_addr), - our_addr = %our_node_addr, - their_addr = %peer_node_addr, - rekey_in_progress = snap.rekey_in_progress, - pending_new_session = snap.pending_new_session, - "rekey-msg3 tie-break: we lose (larger addr), abandon ours" - ); - if let Some(peer) = self.peers.get_mut(&peer_node_addr) - && let Some(idx) = peer.abandon_rekey() - { - if let Some(tid) = peer.transport_id() { - self.peers_by_index.remove(&(tid, idx.as_u32())); - self.pending_outbound.remove(&(tid, idx.as_u32())); - } - let _ = self.index_allocator.free(idx); - } - } - - // Rekey: process as responder, store new session as pending. - let noise_session = { - let Some(conn) = self.connections.get_mut(&link_id) else { - warn!(link_id = %link_id, "Connection removed during rekey msg3 processing"); - self.links.remove(&link_id); - self.stats_mut().record_reject(RejectReason::Handshake( - HandshakeReject::UnknownConnection, - )); - return; - }; - conn.take_session() - }; - let our_new_index = our_index; - - let noise_session = match noise_session { - Some(s) => s, - None => { - warn!("Rekey msg3: no session from handshake"); - self.connections.remove(&link_id); - self.links.remove(&link_id); - self.stats_mut() - .record_reject(RejectReason::Handshake(HandshakeReject::BadState)); - return; - } - }; - - // Store pending session on the existing peer - if let Some(peer) = self.peers.get_mut(&peer_node_addr) { - peer.set_pending_session(noise_session, our_new_index, header.sender_idx); - peer.record_peer_rekey(); - } - - // Register new index in peers_by_index - self.peers_by_index.insert( - (packet.transport_id, our_new_index.as_u32()), - peer_node_addr, - ); - - // Clean up: remove the temporary connection/link. - // Do NOT remove addr_to_link — the entry must remain pointing - // to the original link. - self.connections.remove(&link_id); - self.links.remove(&link_id); - - debug!( - peer = %self.peer_display_name(&peer_node_addr), - our_addr = %self.identity().node_addr(), - new_our_index = %our_new_index, - new_their_index = %header.sender_idx, - "rekey-msg3 responder: pending session set, awaiting K-bit cutover" - ); - return; - } - decision @ (InboundDecision::RestartThenPromote { .. } | InboundDecision::Promote) => { - // Preserve next's epoch-mismatch restart breadcrumb — fires before - // the machine's teardown actions run, matching next's inline order - // (breadcrumb → remove_active_peer → note_link_dead → promote). + decision @ (InboundDecision::RestartThenPromote { .. } + | InboundDecision::Promote + | InboundDecision::CrossConnect { .. } + | InboundDecision::RekeyRespond { .. }) => { + // Preserve the epoch-mismatch restart breadcrumb — it fires before + // the machine's teardown actions run, matching the pre-refactor + // order (breadcrumb → remove_active_peer → note_link_dead → promote). if let InboundDecision::RestartThenPromote { peer } = &decision { debug!( peer = %self.peer_display_name(peer), "Peer restart detected (epoch mismatch), removing stale session" ); } - // Machine-driven establish promote. A TRANSIENT inbound machine - // re-derives the decision and emits the action stream: - // `[PromoteToActive]` for `Promote`, or `[InvalidateSendState, - // ReportLost, PromoteToActive]` for `RestartThenPromote` (whose two - // teardown actions map to `remove_active_peer` / `note_link_dead`, - // in that order — the same order next ran them inline). The transient - // is never inserted into `peer_machines`; the persistent - // `established()` machine is created inside `promote_connection`, - // so there is no double-insert. The relocated promote body, - // and the `RestartThenPromote` teardown ordering equivalence, live in - // the executor's `PromoteToActive` / `InvalidateSendState` / - // `ReportLost` arms. + // Machine-driven inbound establish/rekey resolution. A TRANSIENT + // inbound machine, seeded with the msg1-allocated index, re-derives + // the decision from the same snapshot and emits the action stream: + // `[PromoteToActive]` for `Promote`; + // `[InvalidateSendState, ReportLost, PromoteToActive]` for + // `RestartThenPromote` (the two teardown actions map to + // `remove_active_peer` / `note_link_dead`, in that order); + // `[SwapToInboundSession]` for a simultaneous-init cross-connection; + // `[RekeyRespondTrigger]` for a rekey-responder tie-break. + // The transient is never inserted into `peer_machines`; the + // persistent `established()` machine is created inside + // `promote_connection`, so there is no double-insert. The relocated + // session-swap / promote / teardown bodies live in the executor's + // `SwapToInboundSession` / `RekeyRespondTrigger` / `PromoteToActive` + // / `InvalidateSendState` / `ReportLost` arms. let mut machine = PeerMachine::new_inbound(link_id, packet.timestamp_ms); let actions = machine.step( - PeerEvent::InboundMsg3 { wire, est: snap }, + PeerEvent::InboundMsg3 { + wire, + est: snap, + our_index, + }, packet.timestamp_ms, &mut self.index_allocator, ); diff --git a/src/peer/machine.rs b/src/peer/machine.rs index c46bb12..c09c107 100644 --- a/src/peer/machine.rs +++ b/src/peer/machine.rs @@ -241,9 +241,14 @@ pub(crate) enum PeerEvent { /// Inbound handshake msg3 completed (Noise finalized, identity crystallized /// shell-side, ACL already gated). Drives the establish classification. Used /// for both a fresh inbound establish and a rekey msg3 on an established peer. + /// `our_index` is the index allocated for this leg at msg1; it is carried on + /// the event so a fresh classification machine can be seeded with it before + /// dispatch (the cross-connection and rekey-responder decisions read it back + /// to build their session-swap trigger, and would otherwise emit nothing). InboundMsg3 { wire: WireOutcome, est: EstablishSnapshot, + our_index: SessionIndex, }, /// Outbound handshake msg2 completed (Noise finalized, identity crystallized /// shell-side from the connection's expected identity, ACL already gated, @@ -347,7 +352,7 @@ pub(crate) enum PeerAction { /// NodeAddr) swaps to the inbound session (`take_session`/`replace_session`, /// `peers_by_index` surgery, free the peer's OLD index); otherwise it frees /// `our_index` (the msg1-allocated leg index) and keeps the outbound session. - /// Either way the temporary inbound link is torn down. Provisional / unwired. + /// Either way the temporary inbound link is torn down. SwapToInboundSession { peer: NodeAddr, our_index: SessionIndex, @@ -362,7 +367,6 @@ pub(crate) enum PeerAction { /// `record_peer_rekey` + `peers_by_index.insert`. `our_index` is the /// msg1-allocated leg index that becomes the pending session's index. On XX /// there is NO responder-side msg2 send here (it went out at msg1). - /// Provisional / unwired. RekeyRespondTrigger { peer: NodeAddr, our_index: SessionIndex, @@ -603,9 +607,11 @@ impl PeerMachine { PeerEvent::TransportConnected => self.on_transport_connected(now), PeerEvent::TransportFailed => self.on_transport_failed(now), PeerEvent::InboundMsg1 { link } => self.on_inbound_msg1(link, now, index_allocator), - PeerEvent::InboundMsg3 { wire, est } => { - self.on_inbound_msg3(wire, est, 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::PromotionResolved { result } => self.on_promotion_resolved(result, now), PeerEvent::RekeyMsg2 { their_index } => self.on_rekey_msg2(their_index), @@ -769,6 +775,7 @@ impl PeerMachine { &mut self, wire: WireOutcome, est: EstablishSnapshot, + our_index: SessionIndex, now: u64, _alloc: &mut IndexAllocator, ) -> Vec { @@ -776,6 +783,14 @@ impl PeerMachine { // address + epoch; the full static key stays shell-side). self.node_addr = Some(wire.peer_node_addr); self.remote_epoch = wire.remote_epoch; + // Seed the leg's index from the event. On a fresh classification machine + // the index was allocated at msg1 shell-side and is not otherwise known + // here; the cross-connection and rekey-responder decisions read it back to + // build their session-swap trigger, and the tie-break/duplicate arms use + // it to return the index. Without this seed those arms would emit nothing + // and leak the index. + self.conn.set_our_index(our_index); + self.our_index = Some(our_index); match Fmp::new().establish_inbound(&est, &wire) { InboundDecision::Reject { @@ -784,19 +799,24 @@ impl PeerMachine { // Dual-init rekey tie-break: we win (smaller NodeAddr), drop the // peer's msg3 and keep driving our own rekey. The existing peer // (a separate machine/registry entry) is untouched; this temporary - // leg is discarded. NOTE: next's `handle_msg3` Reject arm removes - // conn+link but does NOT free the msg1-allocated index — so no - // FreeIndex is emitted here, matching ground truth. - self.fail(FailReason::Rejected) + // leg is discarded, returning the msg1-allocated index rather than + // orphaning it, then terminating this leg. + let actions = vec![PeerAction::FreeIndex { index: our_index }]; + let _ = self.fail(FailReason::Rejected); + actions } InboundDecision::ResendMsg2 { msg2 } => { - // Same-epoch duplicate: resend the existing peer's stored msg2, - // leaving the active peer untouched. NOTE: next's `handle_msg3` - // ResendMsg2 arm likewise does NOT free the msg1-allocated index. - match msg2 { - Some(bytes) => vec![PeerAction::SendHandshake { bytes }], - None => Vec::new(), + // Same-epoch duplicate: resend the existing peer's stored msg2 (if + // any), leaving the active peer untouched, return the msg1-allocated + // index, then terminate this leg so a later timeout on a persistent + // machine cannot fire against the healthy established peer. + let mut actions = Vec::new(); + if let Some(bytes) = msg2 { + actions.push(PeerAction::SendHandshake { bytes }); } + actions.push(PeerAction::FreeIndex { index: our_index }); + let _ = self.fail(FailReason::Rejected); + actions } InboundDecision::CrossConnect { peer, @@ -1759,9 +1779,23 @@ mod tests { est.rekey_in_progress = true; let wire = wire_outcome(peer_addr, Some([1u8; 8])); - let actions = m.step(PeerEvent::InboundMsg3 { wire, est }, 1_000, &mut alloc); - // We win the tie-break: drop the peer's msg3, no response, no free. - assert!(actions.is_empty()); + let actions = m.step( + PeerEvent::InboundMsg3 { + wire, + est, + our_index: SessionIndex::new(0x55), + }, + 1_000, + &mut alloc, + ); + // We win the tie-break: drop the peer's msg3 and return the + // msg1-allocated index, then terminate this leg. + assert_eq!( + actions, + vec![PeerAction::FreeIndex { + index: SessionIndex::new(0x55) + }] + ); assert_eq!( m.state(), PeerState::Failed { @@ -1793,7 +1827,15 @@ mod tests { est.rekey_in_progress = true; let wire = wire_outcome(peer_addr, Some([1u8; 8])); - let actions = m.step(PeerEvent::InboundMsg3 { wire, est }, 1_000, &mut alloc); + let actions = m.step( + PeerEvent::InboundMsg3 { + wire, + est, + our_index: SessionIndex::new(0x55), + }, + 1_000, + &mut alloc, + ); // We lose: emit the rekey-respond trigger with abandon_first=true and // the msg1-allocated index. The session/registry surgery is the // executor's; the machine emits ONLY the plain-data trigger (no @@ -1845,7 +1887,16 @@ mod tests { est.existing_peer_epoch = Some([1u8; 8]); // old let wire = wire_outcome(peer_addr, Some([2u8; 8])); // new epoch - let actions = m.step(PeerEvent::InboundMsg3 { wire, est }, 1_000, &mut alloc); + let our_index = m.our_index().unwrap(); + let actions = m.step( + PeerEvent::InboundMsg3 { + wire, + est, + our_index, + }, + 1_000, + &mut alloc, + ); assert_eq!( actions, vec![ @@ -1893,10 +1944,12 @@ mod tests { 100, &mut alloc, ); + let winner_index = winner.our_index().unwrap(); let wp = winner.step( PeerEvent::InboundMsg3 { wire: wire_outcome(peer_addr, Some([3u8; 8])), est: est_new_peer(our), + our_index: winner_index, }, 100, &mut alloc, @@ -1930,10 +1983,12 @@ mod tests { 100, &mut alloc, ); + let loser_index_seed = loser.our_index().unwrap(); let lp = loser.step( PeerEvent::InboundMsg3 { wire: wire_outcome(peer_addr, Some([3u8; 8])), est: est_new_peer(our), + our_index: loser_index_seed, }, 100, &mut alloc, @@ -2004,10 +2059,12 @@ mod tests { // msg3: net-new promote. let our = *peer_identity().node_addr(); + let our_index = m.our_index().unwrap(); let msg3 = m.step( PeerEvent::InboundMsg3 { wire: wire_outcome(peer_addr, Some([4u8; 8])), est: est_new_peer(our), + our_index, }, 200, &mut alloc, @@ -2091,7 +2148,15 @@ mod tests { est.existing_session_age_secs = 10; // < floor(60) -> cross-connection let wire = wire_outcome(peer_addr, Some([5u8; 8])); - let actions = m.step(PeerEvent::InboundMsg3 { wire, est }, 200, &mut alloc); + let actions = m.step( + PeerEvent::InboundMsg3 { + wire, + est, + our_index, + }, + 200, + &mut alloc, + ); assert_eq!( actions, vec![PeerAction::SwapToInboundSession { @@ -2102,6 +2167,147 @@ mod tests { ); } + // ---- Test 7a: cross-connection index seeded ONLY from the event ------- + // A fresh classification machine (no msg1 step, so `conn.our_index()` is + // None) must still emit a NON-EMPTY `SwapToInboundSession` carrying the index + // carried on the event. Without the event-seed the cross-connection arm reads + // a None index and emits nothing — a silent session-swap no-op that leaks the + // index. This guards that the seed closes that hole. + #[test] + fn cross_connect_index_seeded_from_event() { + let (smaller, larger) = ordered_identities(); + let our = *larger.node_addr(); + let peer_addr = *smaller.node_addr(); + + let mut alloc = IndexAllocator::new(); + // Fresh machine, deliberately NOT stepped through msg1 — the only index + // provenance is the event field. + let mut m = PeerMachine::new_inbound(LinkId::new(3), 0); + assert_eq!(m.our_index(), None); + let seed = SessionIndex::new(0xAB); + + let mut est = est_new_peer(our); + est.has_existing_peer = true; + est.existing_peer_epoch = Some([5u8; 8]); + est.has_session = true; + est.is_healthy = true; + est.different_link = true; + est.existing_session_age_secs = 10; // < floor(60) -> cross-connection + let wire = wire_outcome(peer_addr, Some([5u8; 8])); + + let actions = m.step( + PeerEvent::InboundMsg3 { + wire, + est, + our_index: seed, + }, + 200, + &mut alloc, + ); + assert_eq!( + actions, + vec![PeerAction::SwapToInboundSession { + peer: peer_addr, + our_index: seed, + our_inbound_wins: true, + }] + ); + } + + // ---- Test 7d: rekey-responder index seeded ONLY from the event ------- + // The rekey-responder counterpart of Test 7a: a fresh machine seeded only via + // the event emits a NON-EMPTY `RekeyRespondTrigger` carrying that index (the + // pending session's index), rather than the empty no-op an unseeded index + // would produce. + #[test] + fn rekey_respond_index_seeded_from_event() { + let peer = peer_identity(); + let peer_addr = *peer.node_addr(); + let our = *peer_identity().node_addr(); + + let mut alloc = IndexAllocator::new(); + let mut m = PeerMachine::new_inbound(LinkId::new(4), 0); + assert_eq!(m.our_index(), None); + let seed = SessionIndex::new(0xCD); + + // Aged, healthy session, same epoch, same link, no rekey in progress -> + // plain rekey responder (abandon_first: false). + let mut est = est_new_peer(our); + est.has_existing_peer = true; + est.existing_peer_epoch = Some([7u8; 8]); + est.has_session = true; + est.is_healthy = true; + est.existing_session_age_secs = 120; // >= floor -> rekey path + let wire = wire_outcome(peer_addr, Some([7u8; 8])); + + let actions = m.step( + PeerEvent::InboundMsg3 { + wire, + est, + our_index: seed, + }, + 200, + &mut alloc, + ); + assert_eq!( + actions, + vec![PeerAction::RekeyRespondTrigger { + peer: peer_addr, + our_index: seed, + abandon_first: false, + }] + ); + } + + // ---- Test 7e: same-epoch duplicate frees the index and terminates ----- + // A same-epoch duplicate handshake (no cross-connection, no rekey) resends the + // stored msg2, returns the msg1-allocated index, and terminates the leg. The + // terminal transition matters: a persistent machine parked here would keep its + // handshake-timeout armed and later free the index + report loss against the + // healthy established peer. + #[test] + fn resend_msg2_frees_index_and_terminates() { + let peer = peer_identity(); + let peer_addr = *peer.node_addr(); + let our = *peer_identity().node_addr(); + + let mut alloc = IndexAllocator::new(); + let mut m = PeerMachine::new_inbound(LinkId::new(5), 0); + let seed = SessionIndex::new(0xEF); + let stored_msg2 = vec![1u8, 2, 3, 4]; + + // Same epoch, same link, rekey disabled -> duplicate handshake. + let mut est = est_new_peer(our); + est.has_existing_peer = true; + est.existing_peer_epoch = Some([9u8; 8]); + est.rekey_enabled = false; + est.existing_msg2 = Some(stored_msg2.clone()); + let wire = wire_outcome(peer_addr, Some([9u8; 8])); + + let actions = m.step( + PeerEvent::InboundMsg3 { + wire, + est, + our_index: seed, + }, + 200, + &mut alloc, + ); + assert_eq!( + actions, + vec![ + PeerAction::SendHandshake { bytes: stored_msg2 }, + PeerAction::FreeIndex { index: seed }, + ] + ); + assert_eq!( + m.state(), + PeerState::Failed { + reason: FailReason::Rejected + } + ); + } + // ---- Test 7b: dial-persisted outbound promote leaves our_index unset --- // An outbound machine persisted at DIAL (`new_outbound`, `Discovered`, with // `conn.our_index` UNSET — the shell owns the index on its own @@ -2163,13 +2369,22 @@ mod tests { assert_eq!(m.state(), PeerState::Established { addr: peer_addr }); assert_eq!(m.our_index(), None); - // A subsequent inbound restart (peer restart, new epoch) must NOT emit - // UnregisterDecryptSession, because our_index is None. + // A subsequent inbound restart (peer restart, new epoch) must NOT emit a + // separate UnregisterDecryptSession: the restart teardown is the full + // InvalidateSendState (remove_active_peer), which owns the index cleanup. let mut est = est_new_peer(our); est.has_existing_peer = true; est.existing_peer_epoch = Some([1u8; 8]); let wire = wire_outcome(peer_addr, Some([2u8; 8])); - let restart = m.step(PeerEvent::InboundMsg3 { wire, est }, 1_000, &mut alloc); + let restart = m.step( + PeerEvent::InboundMsg3 { + wire, + est, + our_index: SessionIndex::new(0x99), + }, + 1_000, + &mut alloc, + ); assert!( !restart .iter()