From 7fe3388f2f1c06d1f1eaaf567e9a1ccf33b8bb83 Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Fri, 17 Jul 2026 13:47:50 +0000 Subject: [PATCH 1/2] peer: return the inbound establish decision from the machine arm The inbound msg1 machine arm re-derived the establish decision for its own action selection while the shell matched on a separately computed copy. Expose the arm as a method returning the decision alongside the actions so a driver can route on it directly; the step dispatch delegates and keeps its action-only shape. The dead rekey-respond arm is stripped to decision-only and its helper deleted: it allocated from the real index allocator, wrote the rekey shadow index, emitted a wrong-framed rekey send, stamped the dampening clock, and flipped state, none of which happens on the live path, where the inline respond body owns all effects. The resend-msg2 arm's send emission is stripped the same way; the decision already carries the stored bytes. --- src/peer/machine.rs | 151 ++++++++++++++++++++++++++------------------ 1 file changed, 88 insertions(+), 63 deletions(-) diff --git a/src/peer/machine.rs b/src/peer/machine.rs index 14be7e7..b3bd776 100644 --- a/src/peer/machine.rs +++ b/src/peer/machine.rs @@ -574,7 +574,8 @@ impl PeerMachine { PeerEvent::TransportFailed => self.on_transport_failed(now), PeerEvent::HandshakeSendFailed => self.on_handshake_send_failed(), PeerEvent::InboundMsg1 { link, wire, est } => { - self.on_inbound_msg1(link, wire, est, now, index_allocator) + let (_decision, actions) = self.inbound_msg1(link, &wire, est, now); + actions } PeerEvent::Msg2 { their_index, out } => { self.on_msg2(their_index, out, now, index_allocator) @@ -584,7 +585,8 @@ impl PeerMachine { PeerEvent::PromotionResolved { result } => self.on_promotion_resolved(result, now), PeerEvent::RekeyMsg1 { wire, est } => { // A rekey msg1 is a msg1 on an established peer — same core. - self.on_inbound_msg1(self.link, wire, est, now, index_allocator) + let (_decision, actions) = self.inbound_msg1(self.link, &wire, est, now); + actions } PeerEvent::RekeyMsg2 { their_index } => self.on_rekey_msg2(their_index), PeerEvent::RekeyConsume { action } => self.map_rekey_action(action, now), @@ -764,15 +766,22 @@ impl PeerMachine { // Inbound establish // ------------------------------------------------------------------ - fn on_inbound_msg1( + /// Inbound msg1 (fresh or rekey): compute the establish decision for the + /// driver, alongside any machine-phase actions. The single + /// `establish_inbound` evaluation happens here; the driver routes on the + /// returned [`InboundDecision`] and owns the effect-bearing arm bodies + /// (the rekey-respond abandon/alloc/send/store, the duplicate resend, the + /// reject bookkeeping). Only the `Promote`/`RestartThenPromote` phase-1 + /// actions (and the fresh-context reject state flip) are machine-side. + pub(crate) fn inbound_msg1( &mut self, link: LinkId, - wire: WireOutcome, + wire: &WireOutcome, est: EstablishSnapshot, - now: u64, - alloc: &mut IndexAllocator, - ) -> Vec { - match Fmp::new().establish_inbound(&est, &wire) { + _now: u64, + ) -> (InboundDecision, Vec) { + let decision = Fmp::new().establish_inbound(&est, wire); + let actions = match &decision { InboundDecision::Reject { .. } => { // In an establish-leg context this fails the leg; on an // established peer (rekey context) the msg1 is dropped and the @@ -783,15 +792,15 @@ impl PeerMachine { self.fail(FailReason::Rejected) } } - InboundDecision::ResendMsg2 { msg2 } => match msg2 { - Some(bytes) => vec![PeerAction::SendHandshake { bytes }], - None => Vec::new(), - }, - InboundDecision::RekeyRespond { - peer, - abandon_first, - } => self.rekey_respond(peer, abandon_first, &wire, now, alloc), + // The decision carries the stored msg2 bytes; the driver's inline + // resend owns the send. No machine state is touched. + InboundDecision::ResendMsg2 { .. } => Vec::new(), + // Decision-only: the driver's inline body owns the abandon, the + // index allocation, the framed msg2 send, the pending-session + // store, and the dampening stamp. The machine mutates nothing. + InboundDecision::RekeyRespond { .. } => Vec::new(), InboundDecision::RestartThenPromote { peer } => { + let peer = *peer; let mut actions = vec![PeerAction::InvalidateSendState]; if let Some(idx) = self.our_index.take() { actions.push(PeerAction::UnregisterDecryptSession { index: idx }); @@ -802,11 +811,12 @@ impl PeerMachine { peer, kind: LostKind::LinkDead, }); - actions.extend(self.inbound_classify(link, &wire)); + actions.extend(self.inbound_classify(link, wire)); actions } - InboundDecision::Promote => self.inbound_classify(link, &wire), - } + InboundDecision::Promote => self.inbound_classify(link, wire), + }; + (decision, actions) } /// Inbound **Phase 1**: classify the fresh leg *without* @@ -867,42 +877,6 @@ impl PeerMachine { ] } - /// Rekey responder: (optionally) abandon our in-flight rekey, allocate a new - /// index, send the rekey msg2, record the peer rekey (dampening). - fn rekey_respond( - &mut self, - _peer: NodeAddr, - abandon_first: bool, - wire: &WireOutcome, - now: u64, - alloc: &mut IndexAllocator, - ) -> Vec { - let mut actions = Vec::new(); - if abandon_first { - if let Some(idx) = self.rekey_our_index.take() { - actions.push(PeerAction::FreeIndex { index: idx }); - } - self.rekey_in_progress = false; - self.rekey_msg1 = None; - } - let new_index = alloc.allocate().ok(); - if let Some(idx) = new_index { - self.rekey_our_index = Some(idx); - } - actions.push(PeerAction::SendRekey { - bytes: wire.msg2_payload.clone(), - }); - self.last_peer_rekey_ms = now; - let addr = self - .addr() - .unwrap_or_else(|| *wire.peer_identity.node_addr()); - self.state = PeerState::Maintaining { - addr, - kind: MaintainKind::Rekey(RekeyPhase::Msg1Sent), - }; - actions - } - // ------------------------------------------------------------------ // Promotion feedback // ------------------------------------------------------------------ @@ -1703,18 +1677,25 @@ mod tests { let wire = wire_outcome(peer, Some([1u8; 8]), 0x77); let actions = m.step(PeerEvent::RekeyMsg1 { wire, est }, 1_000, &mut alloc); - // We win the tie-break: drop the peer's msg1, no rekey response. + // We win the tie-break: drop the peer's msg1, no rekey response, + // established-context state untouched (the peer keeps running). assert!(actions.is_empty()); assert!( !actions .iter() .any(|a| matches!(a, PeerAction::SendRekey { .. })) ); + assert_eq!( + m.state(), + PeerState::Maintaining { + addr: peer_addr, + kind: MaintainKind::Rekey(RekeyPhase::Msg1Sent) + } + ); } // Case B: PEER is smaller -> we lose -> RekeyRespond{abandon_first:true}. { - let mut alloc = IndexAllocator::new(); let our = *larger.node_addr(); let peer = smaller; let mut m = PeerMachine::new_outbound(LinkId::new(2), peer, 0); @@ -1734,18 +1715,62 @@ mod tests { est.rekey_in_progress = true; let wire = wire_outcome(peer, Some([1u8; 8]), 0x77); - let actions = m.step(PeerEvent::RekeyMsg1 { wire, est }, 1_000, &mut alloc); - // abandon_first -> FreeIndex(old rekey index) then SendRekey(msg2). + let (decision, actions) = m.inbound_msg1(LinkId::new(2), &wire, est, 1_000); + // We lose the tie-break: the decision names the responder path with + // the abandon flag; the machine emits nothing and mutates nothing + // (the driver's inline body owns the abandon/alloc/send/store). + assert!(matches!( + decision, + InboundDecision::RekeyRespond { + abandon_first: true, + .. + } + )); + assert!(actions.is_empty()); assert_eq!( - actions[0], - PeerAction::FreeIndex { - index: SessionIndex::new(0x55) + m.state(), + PeerState::Maintaining { + addr: peer_addr, + kind: MaintainKind::Rekey(RekeyPhase::Msg1Sent) } ); - assert!(matches!(actions[1], PeerAction::SendRekey { .. })); + assert_eq!(m.rekey_our_index, Some(SessionIndex::new(0x55))); + assert!(m.rekey_in_progress); } } + // ---- Test 3b: duplicate msg1 -> resend decision only ------------------ + #[test] + fn inbound_resend_msg2_decision_only() { + let peer = peer_identity(); + let mut m = PeerMachine::new_inbound(LinkId::new(1), 0); + let our = *peer_identity().node_addr(); + let mut est = est_new_peer(our); + est.has_existing_peer = true; + est.existing_peer_epoch = Some([1u8; 8]); + est.has_session = true; + est.is_healthy = true; + est.existing_session_age_secs = 5; // young session -> duplicate, not rekey + est.existing_msg2 = Some(vec![0xC4; 16]); + let wire = wire_outcome(peer, Some([1u8; 8]), 0x77); + + let (decision, actions) = m.inbound_msg1(LinkId::new(1), &wire, est, 1_000); + // The decision carries the stored msg2 bytes; no SendHandshake action, + // no state change (the driver's inline resend owns the send). + assert!(matches!( + &decision, + InboundDecision::ResendMsg2 { msg2: Some(bytes) } if bytes.as_slice() == [0xC4; 16] + )); + assert!(actions.is_empty()); + assert_eq!( + m.state(), + PeerState::Handshaking { + link: LinkId::new(1), + phase: HandshakePhase::Initial + } + ); + } + // ---- Test 4: restart-override ----------------------------------------- #[test] fn restart_override() { From b38f8c6ffbfbcc5999bca7cf86537f0dc8b39df6 Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Fri, 17 Jul 2026 19:17:25 +0000 Subject: [PATCH 2/2] node: classify inbound msg1 via a single machine decision site handle_msg1 now creates one local machine before classification and routes on the InboundDecision it returns, instead of calling establish_inbound in the shell and re-deriving the decision inside each arm. The promote and restart arms drop their own per-arm machine construction and the redundant InboundMsg1 step; their phase-1 actions come from the single call. The effect-bearing arm bodies (rekey-respond, duplicate resend, reject bookkeeping, and the promote/restart tails) are unchanged. Behavior-neutral: the same establish_inbound evaluation over the same snapshot and wire, done once in the machine method rather than once in the shell, with byte-identical arm bodies and no change to index allocation, wire sends, machine state, or telemetry. --- src/node/handlers/handshake.rs | 105 +++++++++++++++------------------ 1 file changed, 49 insertions(+), 56 deletions(-) diff --git a/src/node/handlers/handshake.rs b/src/node/handlers/handshake.rs index 62f580b..80ccc2c 100644 --- a/src/node/handlers/handshake.rs +++ b/src/node/handlers/handshake.rs @@ -324,36 +324,33 @@ impl Node { // registry, matching the pre-refactor read points. let est = self.establish_snapshot(&peer_node_addr); - // === PHASE C: structured classification (pure core) === - // The decision reads only the snapshot + wire outcome; the shell below - // drives the effects. `Promote`/`RestartThenPromote` fall through to the + // === PHASE C: structured classification === + // Evaluate the inbound decision once on a local establish leg and route + // on it. The single `establish_inbound` evaluation lives in + // `inbound_msg1`, which also returns the machine-phase actions the + // Promote/Restart arms drive; the effect-bearing arm bodies stay inline + // in the shell below. `Promote`/`RestartThenPromote` fall through to the // shared authorize → allocate → send-msg2 → promote tail; the other - // variants complete the rate-limiter and return here. - match self.fmp.establish_inbound(&est, &wire) { + // variants complete the rate-limiter and return here. The local machine + // enters `peer_machines` only at the promote tails. + let mut machine = PeerMachine::new_inbound(link_id, packet.timestamp_ms); + let (decision, actions) = machine.inbound_msg1(link_id, &wire, est, packet.timestamp_ms); + match decision { InboundDecision::Reject { reason: InboundReject::AtMaxPeers, } => { - // Net-new arm: drive the reject through the machine so a - // net-new msg1 at the max-peers cap reaches `Failed{Rejected}` - // with the index allocator untouched (no allocate before the - // reject). The transient machine is discarded (never inserted - // into `peer_machines`); `conn`/`link_id` were never inserted - // into the registry either. + // Net-new arm at the max-peers cap: the classification already + // drove the local establish leg to `Failed{Rejected}` with the + // index allocator untouched (no allocate before the reject). + // That local machine is discarded (never inserted into + // `peer_machines`); `conn`/`link_id` were never inserted into + // the registry either. + let _ = actions; debug!( peer = %self.peer_display_name(&peer_node_addr), max = self.max_peers(), "Silent-dropping Msg1 at max_peers cap (early gate; no Msg2 sent)" ); - let mut machine = PeerMachine::new_inbound(link_id, packet.timestamp_ms); - let _ = machine.step( - PeerEvent::InboundMsg1 { - link: link_id, - wire, - est, - }, - packet.timestamp_ms, - &mut self.index_allocator, - ); debug_assert!(matches!( machine.state(), PeerState::Failed { @@ -367,8 +364,10 @@ impl Node { InboundDecision::Reject { reason: reason @ (InboundReject::PendingSession | InboundReject::DualRekeyWon), } => { - // Existing-peer rekey rejects — still inline. Byte-unchanged - // from the pre-refactor shared reject tail. + // Existing-peer rekey rejects: the classification took the + // fresh-context fail path (no actions) and the local machine is + // dropped; the reject bookkeeping below is the whole effect. + debug_assert!(actions.is_empty()); match reason { InboundReject::PendingSession => debug!( peer = %self.peer_display_name(&peer_node_addr), @@ -387,6 +386,10 @@ impl Node { .record_reject(RejectReason::Handshake(HandshakeReject::BadState)); } InboundDecision::ResendMsg2 { msg2 } => { + // Duplicate msg1 at the same epoch: the decision carries the + // stored msg2 bytes and the inline resend below owns the send; + // the classification touched no state. + debug_assert!(actions.is_empty()); if let Some(msg2) = msg2.as_deref() && let Some(transport) = self.transports.get(&packet.transport_id) { @@ -408,6 +411,11 @@ impl Node { peer, abandon_first, } => { + // Rekey responder: the decision carries the routing; the inline + // body below owns the abandon, index allocation, framed msg2 + // send, pending-session store, and dampening stamp. The + // classification machine mutates nothing. + debug_assert!(actions.is_empty()); if abandon_first { // Dual-initiation loser: abandon our own in-flight rekey and // free its index before responding as the rekey responder. @@ -517,26 +525,18 @@ impl Node { "Peer restart detected (epoch mismatch), removing stale session" ); - // Keep the shell's own copies of the msg2 framing inputs before - // the machine event consumes `wire` (WireOutcome is not Clone). + // Snapshot the msg2 framing inputs (`their_index` and the opaque + // payload) for the `build_msg2` call at the promote tail below. + // The classification borrows `wire`, so these locals carry the + // two fields the later framing needs. let msg2_payload = wire.msg2_payload.clone(); let their_index = wire.their_index; - let mut machine = PeerMachine::new_inbound(link_id, packet.timestamp_ms); - - // Phase 1: classify + emit the old-peer teardown. For a restart the - // fresh leg has `our_index == None`, so the emitted sequence is - // exactly [InvalidateSendState, ReportLost{peer}]; the machine then - // parks at Handshaking{ReceivedMsg1} (no allocation). - let phase1 = machine.step( - PeerEvent::InboundMsg1 { - link: link_id, - wire, - est, - }, - packet.timestamp_ms, - &mut self.index_allocator, - ); + // The classification parked the machine at + // `Handshaking{ReceivedMsg1}` (no allocation) and returned the + // old-peer teardown as `actions`. For a restart the fresh leg + // has `our_index == None`, so that sequence is exactly + // [InvalidateSendState, ReportLost{peer}]. debug_assert!(matches!( machine.state(), PeerState::Handshaking { @@ -564,7 +564,7 @@ impl Node { now_ms: packet.timestamp_ms, is_outbound: false, }; - self.execute_peer_actions(link_id, &teardown_ctx, phase1) + self.execute_peer_actions(link_id, &teardown_ctx, actions) .await; // Shell interposition: late-ACL authorize BEFORE any allocation. @@ -698,24 +698,17 @@ impl Node { // matching the pre-refactor authorize-before-allocate ordering // exactly. - // Keep the shell's own copies of the msg2 framing inputs before - // the machine event consumes `wire` (WireOutcome is not Clone). + // Snapshot the msg2 framing inputs (`their_index` and the opaque + // payload) for the `build_msg2` call at the promote tail below. + // The classification borrows `wire`, so these locals carry the + // two fields the later framing needs. let msg2_payload = wire.msg2_payload.clone(); let their_index = wire.their_index; - let mut machine = PeerMachine::new_inbound(link_id, packet.timestamp_ms); - - // Phase 1: classify (no allocation, no actions for a net-new leg). - let phase1 = machine.step( - PeerEvent::InboundMsg1 { - link: link_id, - wire, - est, - }, - packet.timestamp_ms, - &mut self.index_allocator, - ); - debug_assert!(phase1.is_empty()); + // Phase 1: a net-new leg classifies with no allocation and emits + // no actions; the classification parked the machine at + // `Handshaking{ReceivedMsg1}` awaiting the late-ACL gate. + debug_assert!(actions.is_empty()); debug_assert!(matches!( machine.state(), PeerState::Handshaking {