laantungir/fips
1
0
Fork 0
mirror of https://github.com/jmcorgan/fips.git synced 2026-07-22 07:48:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'refactor-node' into refactor-node-next

Browse Source
This commit is contained in:
Johnathan Corgan
2026-07-17 07:43:36 +00:00
parent 4fa565e528 252d16fab9
commit f65cd07048
7 changed files with 518 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
src/node/dataplane/peer_actions.rs
View File

@@ -216,8 +216,13 @@ impl Node {
// (`prepare_outbound_msg1`); send the stored wire. The
// machine's empty payload is ignored.
let _ = bytes;
self.send_stored_msg1(link, ambient.transport_id, &ambient.remote_addr)
.await;
self.send_stored_msg1(
link,
ambient.transport_id,
&ambient.remote_addr,
ambient.now_ms,
)
.await;
}
}
PeerAction::SendRekey { .. } => {
@@ -467,6 +472,16 @@ impl Node {
queue.extend(follow);
}
}
PeerAction::ResolveCrossConnection { .. } => {
// A decision token, not an effect: the outbound msg2
// handler intercepts it and runs the inline swap/keep
// resolution itself, so it must never reach the executor.
debug_assert!(
false,
"ResolveCrossConnection is intercepted by the msg2 \
handler and must never reach the executor"
);
}
PeerAction::SwapSendState { .. } => {
// Initiator cutover: the live authoritative rekey-cadence
// path, routed here from `check_rekey` via

167
src/node/handlers/handshake.rs
View File

@@ -16,7 +16,7 @@ use crate::peer::{ActivePeer, PeerConnection};
use crate::proto::fmp::wire::{Msg1Header, Msg2Header, Msg3Header, build_msg2, build_msg3};
use crate::proto::fmp::{
Disconnect, DisconnectReason, EstablishSnapshot, InboundDecision, InboundReject,
NegotiationPayload, PromotionResult, WireOutcome, cross_connection_winner,
NegotiationPayload, OutboundSnapshot, PromotionResult, WireOutcome, cross_connection_winner,
decide_fmp_negotiation,
};
use crate::transport::{Link, LinkDirection, LinkId, ReceivedPacket};
@@ -25,6 +25,22 @@ use std::time::Duration;
use tracing::{debug, info, warn};
impl Node {
/// Snapshot the registry state the outbound establish decision reads about
/// `peer_addr`: whether the identity is already an active peer, and the
/// pre-evaluated cross-connection tie-break for THIS outbound connection
/// (`is_outbound = true`), resolved into a plain `bool` here so the core
/// stays free of the peer helper.
fn outbound_snapshot(&self, peer_addr: &NodeAddr) -> OutboundSnapshot {
OutboundSnapshot {
has_existing_peer: self.peers.contains_key(peer_addr),
our_outbound_wins: cross_connection_winner(
self.identity().node_addr(),
peer_addr,
true,
),
}
}
/// Feed the peer's control machine the completed-rekey observation after the
/// inline `complete_rekey_msg2`. The obs records the peer's new session index
/// and advances the rekey phase; it emits no action, so a bare `step` keeps
@@ -752,26 +768,82 @@ impl Node {
//
// This ensures both nodes use the same Noise handshake (the winner's
// outbound = the loser's inbound).
if self.peers.contains_key(&peer_node_addr) {
// The machine is the sole computation site of the establish decision:
// the shell builds the outbound snapshot, steps the machine once here,
// and routes on the returned decision — a cross-connection resolves as
// a single `ResolveCrossConnection { swap }` action, a net-new
// establish as the promote action sequence. The Swap/Keep resolution
// bodies stay inline in the shell because they mutate the already
// promoted peer via `replace_session`, for which no `PeerAction`
// exists.
//
// Every outbound leg carries a persistent machine by now — identified
// dials persist one at dial, anonymous-discovery legs at leg birth in
// `start_handshake` — so the lookup is expected to hit, and the
// executor's `PromoteToActive` arm can feed `PromotionResolved` back
// via the same lookup. For an anonymous machine this is where its
// identity crystallizes: msg2 revealed who answered, and the learned
// identity lands on the machine before the step (a no-op for
// identified machines), so the Promote arm reads a crystallized
// address. The `pending_outbound` lifecycle stays shell-side — the
// machine never touches it.
let out_snap = self.outbound_snapshot(&peer_node_addr);
let actions = match self.peer_machines.get_mut(&link_id) {
Some(machine) => {
machine.crystallize_identity(peer_identity);
machine.step(
PeerEvent::OutboundMsg2 {
their_index: header.sender_idx,
out: out_snap,
},
packet.timestamp_ms,
&mut self.index_allocator,
)
}
None => {
// A miss is a state-machine inconsistency (e.g. a test seeding
// a connection/`pending_outbound` entry directly): rebuild the
// machine defensively and persist it, so the promotion feedback
// below still finds it and the promoted peer keeps a machine.
debug_assert!(
false,
"outbound leg {link_id} reached msg2 without a control machine"
);
let mut machine =
PeerMachine::new_outbound(link_id, Some(peer_identity), packet.timestamp_ms);
let actions = machine.step(
PeerEvent::OutboundMsg2 {
their_index: header.sender_idx,
out: out_snap,
},
packet.timestamp_ms,
&mut self.index_allocator,
);
self.peer_machines.insert(link_id, machine);
actions
}
};
let cross_swap = actions.iter().find_map(|action| match action {
PeerAction::ResolveCrossConnection { swap } => Some(*swap),
_ => None,
});
if let Some(swap) = cross_swap {
// The cross-connection arms are decision-only: the resolution
// action is the whole vector.
debug_assert_eq!(actions, vec![PeerAction::ResolveCrossConnection { swap }]);
// Extract the outbound connection from its machine FIRST — the
// machine owns it, so disposing the machine before the take would
// destroy the connection. The dial-persisted outbound machine is
// not consumed by the inline Swap/Keep resolution below (which
// mutates the existing promoted peer directly, with no machine),
// so drop it right after the take — unconditionally, whether or
// not a connection was carried — so none of this block's exits
// leave a dangling machine. Matches the pre-persistence path,
// which created no machine for a cross-connection.
// destroy the connection. The machine has delivered its decision
// and the inline resolution below needs no machine, so drop it
// right after the take — unconditionally, whether or not a
// connection was carried — so none of this block's exits leave a
// dangling machine.
let taken_conn = self
.peer_machines
.get_mut(&link_id)
.and_then(|machine| machine.take_leg());
self.remove_peer_machine(link_id);
let our_outbound_wins = cross_connection_winner(
self.identity().node_addr(),
&peer_node_addr,
true, // this IS our outbound
);
let mut conn = match taken_conn {
Some(c) => c,
@@ -784,7 +856,7 @@ impl Node {
};
let mut cross_conn_outcome: Option<CrossConnOutcome> = None;
if our_outbound_wins {
if swap {
// We're the smaller node. Swap to outbound session + indices.
// The peer will keep their inbound session (complement of ours).
let outbound_our_index = conn.our_index();
@@ -906,65 +978,23 @@ impl Node {
return;
}
// Net-new outbound promote, driven by the per-peer machine. The peer-map
// membership test above was false (the cross-connection block returns on
// an existing peer), so this is the net-new path: promote_connection hits
// its normal-promotion branch and returns Promoted.
// === Net-new outbound establish, driven by the machine. ===
// The machine's decision was `Promote` (`has_existing_peer == false` —
// the cross-connection block above returns otherwise), so
// `promote_connection` hits its normal-promotion branch and returns
// `Promoted`. The machine survives the promotion and the executor
// crystallizes its state via the `PromotionResolved` feedback. The
// promote tail (info log, tree/bloom/backoff, `pending_outbound`
// removal) lives in the executor's `PromoteToActive` arm.
//
// Every outbound leg carries a persistent machine by now — identified
// dials persist one at dial, anonymous-discovery legs at leg birth in
// `start_handshake` — so the lookup is expected to hit. For an anonymous
// machine this is where its identity crystallizes: msg2 revealed who
// answered, and the learned identity lands on the machine before the
// step (a no-op for identified machines). The machine survives the
// promotion and the executor crystallizes its state via the
// `PromotionResolved` feedback. With no outbound decision core to run,
// the machine emits `PromoteToActive` unconditionally (the
// net-new-vs-cross-connection decision was the `peers.contains_key`
// test above, which returns on an existing peer). The promote tail
// (info log, tree/bloom/backoff, `pending_outbound` removal) lives in
// the executor's `PromoteToActive` arm.
let promote_actions = match self.peer_machines.get_mut(&link_id) {
Some(machine) => {
machine.crystallize_identity(peer_identity);
machine.step(
PeerEvent::OutboundMsg2 {
their_index: header.sender_idx,
},
packet.timestamp_ms,
&mut self.index_allocator,
)
}
None => {
// A miss is a state-machine inconsistency (e.g. a test seeding
// a connection/`pending_outbound` entry directly): rebuild the
// machine defensively and persist it, so the promotion feedback
// below still finds it and the promoted peer keeps a machine.
debug_assert!(
false,
"outbound leg {link_id} reached msg2 without a control machine"
);
let mut machine =
PeerMachine::new_outbound(link_id, Some(peer_identity), packet.timestamp_ms);
let actions = machine.step(
PeerEvent::OutboundMsg2 {
their_index: header.sender_idx,
},
packet.timestamp_ms,
&mut self.index_allocator,
);
self.peer_machines.insert(link_id, machine);
actions
}
};
// The outbound Msg2 Promote step cancels the two dial-armed handshake
// timers (the machine survives promotion, so they would otherwise linger
// in `peer_timers` until `drive_peer_timers` lazily discards them — the
// promoted leg's pending connection is consumed and the machine has left
// `SentMsg1`, so they can no longer fire) and then promotes.
// `PromoteToActive` is still what performs the promotion.
// `PromoteToActive` is what performs the promotion.
debug_assert_eq!(
promote_actions,
actions,
vec![
PeerAction::CancelTimer {
kind: TimerKind::HandshakeRetransmit
@@ -986,8 +1016,7 @@ impl Node {
is_outbound: true,
pending_outbound_key: Some(key),
};
self.execute_peer_actions(link_id, &ambient, promote_actions)
.await;
self.execute_peer_actions(link_id, &ambient, actions).await;
}
/// Handle handshake message 3 (phase 0x3).

42
src/node/lifecycle/mod.rs
View File

@@ -793,10 +793,19 @@ impl Node {
error = %e,
"Failed to send handshake message"
);
// Mark connection as failed but don't remove it yet
// The event loop can handle retry logic
if let Some(conn) = self.leg_mut(&link_id) {
conn.mark_failed();
// The machine marks its leg failed but retains it —
// the stale-connection sweep reclaims it, and the
// event loop can handle retry logic until then.
if let Some(machine) = self.peer_machines.get_mut(&link_id) {
let actions = machine.step(
PeerEvent::HandshakeSendFailed,
current_time_ms,
&mut self.index_allocator,
);
debug_assert!(
actions.is_empty(),
"HandshakeSendFailed must emit no actions"
);
}
}
}
@@ -806,9 +815,10 @@ impl Node {
}
/// Send the msg1 wire that `prepare_outbound_msg1` armed on the connection.
/// On send error, marks the connection failed and RETAINS it (the legacy
/// resend tick retries); a missing wire or transport is a no-op. This is the
/// body of the executor's `SendHandshake` msg1 action — reached on both the
/// On send error, steps the machine with `HandshakeSendFailed` — the machine
/// marks its embedded leg failed and RETAINS it (the legacy resend tick
/// retries); a missing wire or transport is a no-op. This is the body of the
/// executor's `SendHandshake` msg1 action — reached on both the
/// connectionless dial and the connection-oriented connect-resolution path,
/// after `prepare_outbound_msg1` has armed the wire.
pub(in crate::node) async fn send_stored_msg1(
@@ -816,6 +826,7 @@ impl Node {
link_id: LinkId,
transport_id: TransportId,
remote_addr: &TransportAddr,
now_ms: u64,
) {
let wire_msg1 = match self.leg(&link_id).and_then(|c| c.handshake_msg1()) {
Some(w) => w.to_vec(),
@@ -842,10 +853,19 @@ impl Node {
error = %e,
"Failed to send handshake message"
);
// Mark connection as failed but don't remove it yet
// The event loop can handle retry logic
if let Some(conn) = self.leg_mut(&link_id) {
conn.mark_failed();
// The machine marks its leg failed but retains it —
// the stale-connection sweep reclaims it, and the
// event loop can handle retry logic until then.
if let Some(machine) = self.peer_machines.get_mut(&link_id) {
let actions = machine.step(
PeerEvent::HandshakeSendFailed,
now_ms,
&mut self.index_allocator,
);
debug_assert!(
actions.is_empty(),
"HandshakeSendFailed must emit no actions"
);
}
}
}

312
src/peer/machine.rs
View File

@@ -54,12 +54,17 @@
//! triggers are live: `handle_msg3` steps the decision machine and the
//! executor performs its verdict.
//!
//! Likewise the outbound completion is NOT routed through the machine: next has
//! no `establish_outbound` core — `handle_msg2` learns the identity from
//! `conn.expected_identity()` and drives `promote_connection` inline. The
//! machine's outbound role is `Dial`→`OpenTransport`, `start_outbound_handshake`,
//! and the resend/timeout timers; the shell drives promote and feeds the outcome
//! back via [`PromotionResolved`](PeerEvent::PromotionResolved).
//! The outbound completion is routed through the machine the same way:
//! `handle_msg2` crystallizes the learned identity, builds an
//! [`OutboundSnapshot`], and steps the persistent machine once; the machine
//! computes the establish decision via `establish_outbound`. A net-new
//! completion promotes through [`PromoteToActive`](PeerAction::PromoteToActive)
//! (the executor feeds the outcome back via
//! [`PromotionResolved`](PeerEvent::PromotionResolved)), while a
//! cross-connection resolves as a
//! [`ResolveCrossConnection`](PeerAction::ResolveCrossConnection) decision the
//! shell intercepts — its inline Swap/Keep bodies mutate the already-promoted
//! peer, for which no `PeerAction` exists.
//!
//! ## Realizability notes
//!
@@ -81,7 +86,8 @@
use crate::peer::PeerConnection;
use crate::proto::fmp::{
ConnAction, ConnSnapshot, ConnectionState, EstablishSnapshot, Fmp, InboundDecision,
InboundReject, PeerSnapshot, PromotionResult, RekeyCfg, RekeyResendSnapshot, WireOutcome,
InboundReject, OutboundDecision, OutboundSnapshot, PeerSnapshot, PromotionResult, RekeyCfg,
RekeyResendSnapshot, WireOutcome,
};
use crate::proto::link::LinkMessageType;
use crate::transport::{LinkId, TransportAddr, TransportId};
@@ -258,6 +264,12 @@ pub(crate) enum PeerEvent {
TransportConnected,
/// Transport connect failed.
TransportFailed,
/// The transport accepted the dial but sending a stored handshake
/// initiation failed. The machine marks the embedded leg failed so the
/// stale-connection sweep reclaims it, WITHOUT leaving the handshaking
/// state — the retransmit driver may still resend in the window before
/// the sweep.
HandshakeSendFailed,
/// Inbound handshake msg1 processed shell-side: allocate our index and reply
/// with msg2, deferring identity/classification to msg3. Carries no identity
/// or wire outcome — on XX neither is known at msg1.
@@ -275,12 +287,16 @@ pub(crate) enum PeerEvent {
our_index: SessionIndex,
},
/// Outbound handshake msg2 completed (Noise finalized, identity crystallized
/// shell-side from the connection's expected identity, ACL already gated,
/// msg3 sent). The net-new-vs-cross-connection decision is the shell's
/// peer-map membership test; this event is emitted only on the net-new path,
/// so it maps directly to a promote. `their_index` carries the peer's session
/// index for shape parity with the inbound path.
OutboundMsg2 { their_index: SessionIndex },
/// onto the machine shell-side, ACL already gated, msg3 sent). Carries the
/// shell-built [`OutboundSnapshot`]; the machine computes the establish
/// decision from it — a net-new promote, or a cross-connection swap/keep
/// conveyed as [`ResolveCrossConnection`](PeerAction::ResolveCrossConnection).
/// `their_index` carries the peer's session index for shape parity with the
/// inbound path.
OutboundMsg2 {
their_index: SessionIndex,
out: OutboundSnapshot,
},
/// `promote_connection` resolved the [`PromoteToActive`](PeerAction::PromoteToActive)
/// action shell-side; the machine consumes the outcome (it does not
/// re-decide the tie-break).
@@ -359,6 +375,11 @@ pub(crate) enum PeerAction {
/// Crystallize identity, re-home the map key, publish send-state
/// (`promote_connection`). Resolves to a [`PromotionResolved`](PeerEvent::PromotionResolved).
PromoteToActive { link: LinkId },
/// A DECISION conveyed to the driver, not an effect: emitted by the
/// outbound-msg2 arm when the establish decision is a cross-connection
/// resolution. The shell intercepts it and runs the inline swap/keep
/// resolution; it must never reach the action executor.
ResolveCrossConnection { swap: bool },
/// Initiator-side rekey cutover: swap the published send-state to the pending
/// epoch.
SwapSendState { epoch: [u8; 8] },
@@ -716,13 +737,16 @@ impl PeerMachine {
} => self.on_dial(transport_id, remote_addr, connection_oriented, now),
PeerEvent::TransportConnected => self.on_transport_connected(now),
PeerEvent::TransportFailed => self.on_transport_failed(now),
PeerEvent::HandshakeSendFailed => self.on_handshake_send_failed(),
PeerEvent::InboundMsg1 { link } => self.on_inbound_msg1(link, 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::OutboundMsg2 { their_index, out } => {
self.on_outbound_msg2(their_index, out, now)
}
PeerEvent::PromotionResolved { result } => self.on_promotion_resolved(result, now),
PeerEvent::RekeyMsg2 { their_index } => self.on_rekey_msg2(their_index),
PeerEvent::RekeyConsume { action } => self.map_rekey_action(action, now),
@@ -813,6 +837,19 @@ impl PeerMachine {
actions
}
/// A stored handshake initiation failed to send: mark the embedded leg
/// failed so the stale-connection sweep (which reads the leg's
/// `is_failed`) reclaims it. NO state flip — the machine stays in
/// `Handshaking{SentMsg1}` so retransmit eligibility
/// (`is_handshaking_sent_msg1`) survives until the sweep, and no timer
/// actions are emitted.
fn on_handshake_send_failed(&mut self) -> Vec<PeerAction> {
if let Some(leg) = self.leg.as_mut() {
leg.mark_failed();
}
Vec::new()
}
/// Emit msg1 and arm the retransmit/timeout timers. The Noise msg1
/// construction and its index allocation are shell-side effects performed by
/// the driver when it executes this action; an empty payload is emitted (see
@@ -960,33 +997,63 @@ impl PeerMachine {
}
}
/// Net-new outbound promote after msg2 completes the initiator handshake.
/// Outbound completion after msg2 completes the initiator handshake:
/// compute the establish decision from the snapshot via
/// `establish_outbound`. `Promote` drives promotion via actions; the
/// Swap/Keep outcomes are conveyed as a
/// [`ResolveCrossConnection`](PeerAction::ResolveCrossConnection) decision
/// for the shell's inline resolution, which owns all effects (index frees,
/// session replacement) permanently. No state guard: anonymous-discovery
/// legs reach msg2 parked in `Discovered` (identified dials in
/// `Handshaking{SentMsg1}`), and the decision reads only the snapshot.
///
/// The net-new-vs-cross-connection decision is made shell-side (via the peer
/// map), so this handler is reached only on the net-new path and maps
/// directly to an unconditional promote — there is no sub-decision to run
/// here. The persistent `Established` machine is created inside
/// `promote_connection`; this machine is a transient decision vehicle that is
/// discarded after the step, so the `set_their_index` and state write below
/// are behaviorally inert (kept to make the transient's intent explicit and
/// to match the established lifecycle shape).
fn on_outbound_msg2(&mut self, their_index: SessionIndex, _now: u64) -> Vec<PeerAction> {
/// On `Promote` the persistent machine survives as the active peer's
/// control machine, and `Established` is written here, ahead of the
/// `PromotionResolved` feedback (which re-crystallizes it). The shell
/// crystallizes the learned identity onto the machine before the step, so
/// the `addr()` read is `Some` for anonymous legs too. On Swap/Keep the
/// `set_their_index` write lands on the machine's shadow connection and
/// the shell discards the machine right after the step, so it is
/// unobservable.
fn on_outbound_msg2(
&mut self,
their_index: SessionIndex,
out: OutboundSnapshot,
_now: u64,
) -> Vec<PeerAction> {
self.conn.set_their_index(their_index);
let addr = self.addr().unwrap_or_else(zero_addr);
self.state = PeerState::Established { addr };
// The machine survives promotion (it becomes the active peer's control
// machine), so cancel the outbound handshake timers here or they would
// linger in the driver's store. A late fire would no-op against the
// non-`Handshaking` state, but leaving them armed is a timer leak.
vec![
PeerAction::CancelTimer {
kind: TimerKind::HandshakeRetransmit,
},
PeerAction::CancelTimer {
kind: TimerKind::HandshakeTimeout,
},
PeerAction::PromoteToActive { link: self.link },
]
match Fmp::new().establish_outbound(&out) {
OutboundDecision::Promote => {
let addr = self.addr().unwrap_or_else(zero_addr);
self.state = PeerState::Established { addr };
// The machine survives promotion (it becomes the active peer's
// control machine), so cancel the outbound handshake timers here
// or they would linger in the driver's store. A late fire would
// no-op against the non-`Handshaking` state, but leaving them
// armed is a timer leak.
vec![
PeerAction::CancelTimer {
kind: TimerKind::HandshakeRetransmit,
},
PeerAction::CancelTimer {
kind: TimerKind::HandshakeTimeout,
},
PeerAction::PromoteToActive { link: self.link },
]
}
OutboundDecision::CrossConnectionSwap => {
// Our outbound wins: convey the decision only. The shell's
// inline resolution swaps the peer to the outbound session and
// owns the index frees and session replacement.
vec![PeerAction::ResolveCrossConnection { swap: true }]
}
OutboundDecision::CrossConnectionKeep => {
// Our outbound loses: convey the decision only. The shell's
// inline resolution keeps the existing inbound session and
// frees the unused outbound index.
vec![PeerAction::ResolveCrossConnection { swap: false }]
}
}
}
/// XX inbound cross-connection at msg3. The Noise session swap and the
@@ -1620,7 +1687,17 @@ mod tests {
let mut m = PeerMachine::new_outbound(link, Some(id), 0);
let their_index = SessionIndex::new(0x77);
let actions = m.step(PeerEvent::OutboundMsg2 { their_index }, 0, &mut alloc);
let actions = m.step(
PeerEvent::OutboundMsg2 {
their_index,
out: OutboundSnapshot {
has_existing_peer: false,
our_outbound_wins: false,
},
},
0,
&mut alloc,
);
assert_eq!(
actions,
@@ -1650,7 +1727,17 @@ mod tests {
let mut m = PeerMachine::new_outbound(link, Some(id), 0);
let their_index = SessionIndex::new(0x88);
let actions = m.step(PeerEvent::OutboundMsg2 { their_index }, 0, &mut alloc);
let actions = m.step(
PeerEvent::OutboundMsg2 {
their_index,
out: OutboundSnapshot {
has_existing_peer: false,
our_outbound_wins: false,
},
},
0,
&mut alloc,
);
assert_eq!(
actions,
@@ -1666,6 +1753,92 @@ mod tests {
);
}
// ---- Test: outbound msg2 cross-connection decision arms ---------------
// Swap/Keep are decision-only: a single `ResolveCrossConnection { swap }`
// action, no index actions, no state change — the shell's inline
// resolution owns all effects and discards the machine after the step.
#[test]
fn outbound_msg2_cross_connection_decisions() {
let mut alloc = IndexAllocator::new();
let peer = peer_identity();
// Cross-connection SWAP: our outbound wins -> decision only; the
// shell's inline resolution owns the index frees and session
// replacement.
let mut m2 = PeerMachine::new_outbound(LinkId::new(2), Some(peer), 0);
m2.state = PeerState::Handshaking {
link: LinkId::new(2),
phase: HandshakePhase::SentMsg1,
};
m2.conn.set_our_index(SessionIndex::new(0x2222)); // outbound index
m2.our_index = Some(SessionIndex::new(0x1111)); // old inbound index
let swap = m2.step(
PeerEvent::OutboundMsg2 {
their_index: SessionIndex::new(0x99),
out: OutboundSnapshot {
has_existing_peer: true,
our_outbound_wins: true,
},
},
400,
&mut alloc,
);
assert_eq!(
swap,
vec![PeerAction::ResolveCrossConnection { swap: true }]
);
assert!(!swap.iter().any(|a| matches!(
a,
PeerAction::FreeIndex { .. } | PeerAction::RegisterDecryptSession { .. }
)));
// The decision arm leaves the machine untouched: still Handshaking,
// our_index unchanged.
assert_eq!(
m2.state(),
PeerState::Handshaking {
link: LinkId::new(2),
phase: HandshakePhase::SentMsg1,
}
);
assert_eq!(m2.our_index(), Some(SessionIndex::new(0x1111)));
// Cross-connection KEEP: our outbound loses -> decision only; the
// shell's inline resolution frees the unused outbound index.
let mut m3 = PeerMachine::new_outbound(LinkId::new(3), Some(peer), 0);
m3.state = PeerState::Handshaking {
link: LinkId::new(3),
phase: HandshakePhase::SentMsg1,
};
m3.conn.set_our_index(SessionIndex::new(0x3333));
let keep = m3.step(
PeerEvent::OutboundMsg2 {
their_index: SessionIndex::new(0x9A),
out: OutboundSnapshot {
has_existing_peer: true,
our_outbound_wins: false,
},
},
500,
&mut alloc,
);
assert_eq!(
keep,
vec![PeerAction::ResolveCrossConnection { swap: false }]
);
assert!(!keep.iter().any(|a| matches!(
a,
PeerAction::FreeIndex { .. } | PeerAction::RegisterDecryptSession { .. }
)));
assert_eq!(
m3.state(),
PeerState::Handshaking {
link: LinkId::new(3),
phase: HandshakePhase::SentMsg1,
}
);
assert_eq!(m3.our_index(), None);
}
// ---- Contract: actions are runtime-agnostic data ----------------------
//
// The machine's emitted actions are the message contract between the sync
@@ -1700,6 +1873,7 @@ mod tests {
PeerAction::PromoteToActive {
link: LinkId::new(7),
},
PeerAction::ResolveCrossConnection { swap: true },
PeerAction::SwapSendState { epoch: [1u8; 8] },
PeerAction::CompleteDrain { peer },
PeerAction::InvalidateSendState,
@@ -1745,6 +1919,7 @@ mod tests {
| PeerAction::SendRekey { .. }
| PeerAction::SendLinkMessage { .. }
| PeerAction::PromoteToActive { .. }
| PeerAction::ResolveCrossConnection { .. }
| PeerAction::SwapSendState { .. }
| PeerAction::CompleteDrain { .. }
| PeerAction::InvalidateSendState
@@ -2461,6 +2636,10 @@ mod tests {
let promote = m.step(
PeerEvent::OutboundMsg2 {
their_index: SessionIndex::new(0x77),
out: OutboundSnapshot {
has_existing_peer: false,
our_outbound_wins: false,
},
},
300,
&mut alloc,
@@ -2575,6 +2754,10 @@ mod tests {
let promote = m.step(
PeerEvent::OutboundMsg2 {
their_index: SessionIndex::new(0x77),
out: OutboundSnapshot {
has_existing_peer: false,
our_outbound_wins: false,
},
},
200,
&mut alloc,
@@ -2661,6 +2844,53 @@ mod tests {
);
}
// ---- Test 7e: HandshakeSendFailed marks the leg, keeps the state ------
// A stored-msg1 send failure marks the embedded leg failed (the
// stale-connection sweep reads the leg's `is_failed`) WITHOUT leaving
// `Handshaking{SentMsg1}` — retransmit eligibility
// (`is_handshaking_sent_msg1`) must survive until the sweep — and emits
// no actions. On a machine with no leg it is a defensive no-op.
#[test]
fn handshake_send_failed_marks_leg_without_leaving_handshaking() {
let mut alloc = IndexAllocator::new();
let peer = peer_identity();
// Dial-persisted outbound machine carrying a prepared leg, driven to
// Handshaking{SentMsg1} via the connectionless dial.
let mut m = PeerMachine::new_outbound(LinkId::new(1), Some(peer), 0);
let _ = m.step(
PeerEvent::Dial {
transport_id: TransportId::new(1),
remote_addr: TransportAddr::from_string("127.0.0.1:9999"),
peer_identity: peer,
connection_oriented: false,
},
100,
&mut alloc,
);
m.set_leg(PeerConnection::outbound(LinkId::new(1), peer, 100));
assert!(m.is_handshaking_sent_msg1());
assert!(!m.leg().expect("leg embedded").is_failed());
let actions = m.step(PeerEvent::HandshakeSendFailed, 200, &mut alloc);
assert_eq!(actions, Vec::new(), "HandshakeSendFailed emits no actions");
assert!(
m.is_handshaking_sent_msg1(),
"retransmit eligibility survives a send failure"
);
assert!(
m.leg().expect("leg retained").is_failed(),
"the leg carries the failed mark the sweep reads"
);
// With no leg (e.g. after take_leg) the event is a defensive no-op.
let _ = m.take_leg();
let actions = m.step(PeerEvent::HandshakeSendFailed, 300, &mut alloc);
assert_eq!(actions, Vec::new());
assert!(m.is_handshaking_sent_msg1());
assert!(m.leg().is_none());
}
// ---- Test 8: liveness -> LinkDeadSuspected -> ReportLost --------------
#[test]
fn liveness_to_link_dead() {

56
src/proto/fmp/core.rs
View File

@@ -266,6 +266,21 @@ pub(crate) struct EstablishSnapshot {
pub our_node_addr: NodeAddr,
}
/// A snapshot of the registry state the *outbound* establish decision reads
/// about the peer whose msg2 just completed our handshake, taken by the shell.
///
/// Both fields are pre-evaluated shell-side (the tie-break is a pure function of
/// the two node addresses, resolved into a plain `bool` here) so the core never
/// touches live `Node` state or the `crate::peer` tie-break helper.
pub(crate) struct OutboundSnapshot {
/// The peer identity is already a promoted active peer — i.e. this outbound
/// completion is a cross-connection (we also processed their msg1).
pub has_existing_peer: bool,
/// Pre-evaluated cross-connection tie-break: our *outbound* connection wins
/// (we are the smaller NodeAddr). Only meaningful when `has_existing_peer`.
pub our_outbound_wins: bool,
}
/// A registry/transport effect the async shell performs on the core's behalf.
///
/// The scaffold subset covers the maintain/teardown half of the lifecycle. The
@@ -400,6 +415,30 @@ pub(crate) enum InboundReject {
DualRekeyWon,
}
/// The classification outcome for one outbound `handle_msg2` completion, decided
/// purely from the [`OutboundSnapshot`]. The shell matches on this and drives
/// the effects; the core consumes nothing and touches no live state.
///
/// Only the case where the peer is *not* yet a promoted active peer is a plain
/// promotion; when it is, this msg2 completes the outbound half of a
/// cross-connection and the tie-break decides whether we swap our session to the
/// (winning) outbound one or keep our existing inbound session. The rekey-msg2
/// completion path is handled by a separate shell driver (it mutates
/// `ActivePeer`, not a `PeerConnection`) and never reaches this decision.
#[derive(Debug, PartialEq, Eq)]
pub(crate) enum OutboundDecision {
/// No existing peer for this identity: promote the completed outbound
/// connection to an active peer via the normal promotion path.
Promote,
/// Cross-connection and our outbound wins (smaller NodeAddr): swap the peer
/// to the outbound session + indices, freeing the old inbound index.
CrossConnectionSwap,
/// Cross-connection and our outbound loses (larger NodeAddr): keep the
/// existing inbound session and original `their_index`, freeing the unused
/// outbound index.
CrossConnectionKeep,
}
impl Fmp {
/// Decide the teardown choreography for the stale/failed connections the
/// shell snapshotted. For each connection, an outbound one with a known
@@ -631,6 +670,23 @@ impl Fmp {
}
}
}
/// Classify one outbound `handle_msg2` completion from the outbound snapshot.
/// Pure: reads only `snap`, mutates nothing.
///
/// Mirrors the pre-refactor branch exactly: an existing same-identity peer
/// makes this a cross-connection resolved by the (pre-evaluated) tie-break —
/// swap on a win, keep on a loss — otherwise a net-new promote.
pub(crate) fn establish_outbound(&self, snap: &OutboundSnapshot) -> OutboundDecision {
if !snap.has_existing_peer {
return OutboundDecision::Promote;
}
if snap.our_outbound_wins {
OutboundDecision::CrossConnectionSwap
} else {
OutboundDecision::CrossConnectionKeep
}
}
}
/// Exponential-backoff schedule for the next handshake/rekey msg1 resend:

3
src/proto/fmp/mod.rs
View File

@@ -38,7 +38,8 @@ mod tests;
pub(crate) use core::{
ConnAction, ConnSnapshot, EstablishSnapshot, InboundDecision, InboundReject, LifecycleView,
PeerSnapshot, RekeyCfg, RekeyResendSnapshot, WireOutcome, decide_fmp_negotiation,
OutboundDecision, OutboundSnapshot, PeerSnapshot, RekeyCfg, RekeyResendSnapshot, WireOutcome,
decide_fmp_negotiation,
};
pub use core::{PromotionResult, cross_connection_winner};
pub(crate) use limits::backoff_ms;

45
src/proto/fmp/tests/core.rs
View File

@@ -5,8 +5,8 @@ use super::util::{
wire_outcome,
};
use crate::proto::fmp::{
ConnAction, Fmp, InboundDecision, InboundReject, NegotiationPayload, NodeProfile, RekeyCfg,
cross_connection_winner,
ConnAction, Fmp, InboundDecision, InboundReject, NegotiationPayload, NodeProfile,
OutboundDecision, OutboundSnapshot, RekeyCfg, cross_connection_winner,
};
use crate::testutil::make_node_addr;
use crate::transport::LinkId;
@@ -539,6 +539,47 @@ fn establish_aged_unhealthy_is_duplicate() {
));
}
// ===========================================================================
// establish_outbound — outbound msg2 classification (E4)
// ===========================================================================
#[test]
fn establish_outbound_no_existing_peer_promotes() {
let fmp = Fmp::new();
// our_outbound_wins is irrelevant when there is no existing peer.
let snap = OutboundSnapshot {
has_existing_peer: false,
our_outbound_wins: true,
};
assert_eq!(fmp.establish_outbound(&snap), OutboundDecision::Promote);
}
#[test]
fn establish_outbound_cross_connection_win_swaps() {
let fmp = Fmp::new();
let snap = OutboundSnapshot {
has_existing_peer: true,
our_outbound_wins: true,
};
assert_eq!(
fmp.establish_outbound(&snap),
OutboundDecision::CrossConnectionSwap
);
}
#[test]
fn establish_outbound_cross_connection_loss_keeps() {
let fmp = Fmp::new();
let snap = OutboundSnapshot {
has_existing_peer: true,
our_outbound_wins: false,
};
assert_eq!(
fmp.establish_outbound(&snap),
OutboundDecision::CrossConnectionKeep
);
}
// ===== cross_connection_winner tie-break tests =====
#[test]