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()