mirror of
https://github.com/jmcorgan/fips.git
synced 2026-08-09 16:24:45 +00:00
node: relocate decrypt-session registration to the establish executor
Move register_decrypt_worker_session out of promote_connection into the executor's PromoteToActive handler, gated on a promoted or cross-connection-won result. Every live promote now flows through that one executor path, so registration still fires exactly once at the same synchronous point; the direct test callers of promote_connection spawn no worker pool, so the call was already a no-op for them. Add the cross-connection loser-link teardown (close the losing transport, remove its link, re-point addr_to_link at the winner) to the executor as a guarded follow-up. It is unreachable on the current driven establish paths, which only promote net-new peers, and asserts so, but keeps the executor complete for when that case is driven. Remove the now-dead drive_promote_to_active and ConnAction::PromoteToActive.
This commit is contained in:
@@ -22,6 +22,7 @@
|
||||
use crate::node::Node;
|
||||
use crate::node::reject::{HandshakeReject, RejectReason};
|
||||
use crate::peer::machine::{PeerAction, PeerEvent};
|
||||
use crate::proto::fmp::PromotionResult;
|
||||
use crate::proto::fmp::wire::build_msg2;
|
||||
use crate::transport::{LinkId, TransportAddr, TransportId};
|
||||
use crate::utils::index::SessionIndex;
|
||||
@@ -35,7 +36,7 @@ use tracing::warn;
|
||||
///
|
||||
/// Unlike a machine event/action payload this is **executor-side**, so it may
|
||||
/// hold real values resolved from the wire context (cf. `handle_msg1`'s
|
||||
/// `wire`/`packet` locals and `drive_promote_to_active`'s ambient args). It is
|
||||
/// `wire`/`packet` locals and `promote_connection`'s ambient args). It is
|
||||
/// built fresh per driven step by the caller at cutover time (C3-2/C3-3).
|
||||
#[allow(dead_code)]
|
||||
pub(in crate::node) struct PeerActionCtx {
|
||||
@@ -165,15 +166,40 @@ impl Node {
|
||||
}
|
||||
PeerAction::PromoteToActive { link: promote_link } => {
|
||||
// GAP-1: ambient supplies the verified identity + promotion ts
|
||||
// that `promote_connection` needs (cf. `drive_promote_to_active`).
|
||||
// that `promote_connection` needs (resolved from the wire ctx).
|
||||
match self.promote_connection(
|
||||
promote_link,
|
||||
ambient.verified_identity,
|
||||
ambient.now_ms,
|
||||
) {
|
||||
Ok(result) => {
|
||||
// R1 (C3-3b): the decrypt-worker registration relocated
|
||||
// OUT of `promote_connection` into THIS single executor
|
||||
// arm — the one live caller of `promote_connection` (both
|
||||
// the inbound `handle_msg1` and outbound `handle_msg2`
|
||||
// net-new establish paths reach it here). Register iff the
|
||||
// promotion actually created or replaced a peer
|
||||
// (`Promoted | CrossConnectionWon`), NEVER on
|
||||
// `CrossConnectionLost`. Run synchronously right after
|
||||
// `promote_connection` returns, before feeding
|
||||
// `PromotionResolved` and before any await — the exact
|
||||
// synchronous point (and Promoted/Won gating) of the
|
||||
// pre-refactor in-`promote_connection` call. No-op when
|
||||
// the worker pool isn't spawned (`register_...` early-
|
||||
// returns), so the direct `promote_connection` test
|
||||
// callers (which bypass this executor) are unaffected.
|
||||
#[cfg(unix)]
|
||||
match result {
|
||||
PromotionResult::Promoted(node_addr)
|
||||
| PromotionResult::CrossConnectionWon { node_addr, .. } => {
|
||||
self.register_decrypt_worker_session(&node_addr);
|
||||
}
|
||||
PromotionResult::CrossConnectionLost { .. } => {}
|
||||
}
|
||||
|
||||
// Feed the outcome back into the machine and fold the
|
||||
// follow-up actions (RegisterDecryptSession, cross-conn
|
||||
// follow-up actions (RegisterDecryptSession — now a
|
||||
// redundant no-op, see its arm — and the cross-conn index
|
||||
// frees) into the worklist. Disjoint field borrow again.
|
||||
let follow = match self.peer_machines.get_mut(&promote_link) {
|
||||
Some(machine) => machine.step(
|
||||
@@ -184,6 +210,70 @@ impl Node {
|
||||
None => Vec::new(),
|
||||
};
|
||||
queue.extend(follow);
|
||||
|
||||
// Defensive cross-connection loser-link surgery (C3-3b).
|
||||
// LINK-ONLY: close the losing transport connection, drop
|
||||
// its link, and re-point `addr_to_link`, reproducing the
|
||||
// pre-refactor inline `handle_msg2`/`handle_msg1` per-arm
|
||||
// order EXACTLY. The index-plane frees/unregisters are
|
||||
// owned by the machine's `PromotionResolved{Won/Lost}`
|
||||
// follow-up (queued just above), so NOTHING here touches
|
||||
// an index — no double-free.
|
||||
//
|
||||
// UNREACHABLE on every current driven path: the inbound
|
||||
// and outbound net-new establish arms only route to the
|
||||
// machine when no promoted peer exists for the node_addr
|
||||
// (and `RestartThenPromote` removes the old peer first),
|
||||
// so `promote_connection` always returns `Promoted`. The
|
||||
// `debug_assert!(false, ..)` catches any future path that
|
||||
// drives a cross-connection through the executor without
|
||||
// the matching send-state handling.
|
||||
match result {
|
||||
PromotionResult::CrossConnectionWon { loser_link_id, .. } => {
|
||||
debug_assert!(
|
||||
false,
|
||||
"executor CrossConnectionWon is unreachable on \
|
||||
driven net-new establish paths"
|
||||
);
|
||||
// Close the losing transport connection (no-op for
|
||||
// connectionless) via the LOSER link's own
|
||||
// transport/addr, then drop the losing link.
|
||||
if let Some(loser_link) = self.links.get(&loser_link_id) {
|
||||
let loser_tid = loser_link.transport_id();
|
||||
let loser_addr = loser_link.remote_addr().clone();
|
||||
if let Some(transport) = self.transports.get(&loser_tid) {
|
||||
transport.close_connection(&loser_addr).await;
|
||||
}
|
||||
}
|
||||
self.remove_link(&loser_link_id);
|
||||
// Point `addr_to_link` at the winning (current)
|
||||
// link.
|
||||
self.addr_to_link.insert(
|
||||
(ambient.transport_id, ambient.remote_addr.clone()),
|
||||
promote_link,
|
||||
);
|
||||
}
|
||||
PromotionResult::CrossConnectionLost { winner_link_id } => {
|
||||
debug_assert!(
|
||||
false,
|
||||
"executor CrossConnectionLost is unreachable on \
|
||||
driven net-new establish paths"
|
||||
);
|
||||
// Close this (losing) connection, drop its link,
|
||||
// and restore `addr_to_link` to the winner.
|
||||
if let Some(transport) =
|
||||
self.transports.get(&ambient.transport_id)
|
||||
{
|
||||
transport.close_connection(&ambient.remote_addr).await;
|
||||
}
|
||||
self.remove_link(&promote_link);
|
||||
self.addr_to_link.insert(
|
||||
(ambient.transport_id, ambient.remote_addr.clone()),
|
||||
winner_link_id,
|
||||
);
|
||||
}
|
||||
PromotionResult::Promoted(_) => {}
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
// GAP-4: promotion failed. `promote_connection` already
|
||||
@@ -247,14 +337,15 @@ impl Node {
|
||||
}
|
||||
PeerAction::RegisterDecryptSession { index } => {
|
||||
let _ = index;
|
||||
// C3-2 (HALT-reported): the decrypt-worker registration still
|
||||
// runs INSIDE `promote_connection` (`handshake.rs:1193/1305`),
|
||||
// which is the single source of truth for its ~40 direct
|
||||
// `promote_connection` callers (unit/integration tests) and the
|
||||
// two live handlers. Relocating it out (GAP-3) would perturb the
|
||||
// live promote path, so C3-1 keeps it there and drives this
|
||||
// action as a no-op; the relocation lands with the inbound
|
||||
// cutover in C3-2.
|
||||
// No-op by design. C3-3b (R1-a) relocated the decrypt-worker
|
||||
// registration into the `PromoteToActive` Ok arm above, gated on
|
||||
// the returned `PromotionResult`, so it runs once per live
|
||||
// promote (Promoted/Won) at the pre-refactor synchronous point.
|
||||
// This machine-emitted action is now redundant with that arm;
|
||||
// kept as an inert no-op (rather than removing the emission) so
|
||||
// the machine's action sequence and its unit tests stay
|
||||
// unchanged. The keyed-by-NodeAddr register does not need the
|
||||
// machine's `index` payload.
|
||||
}
|
||||
PeerAction::UnregisterDecryptSession { index } => {
|
||||
// Executor supplies `transport_id` from ambient; keyed by
|
||||
|
||||
@@ -12,7 +12,7 @@ use crate::peer::machine::{
|
||||
use crate::peer::{ActivePeer, PeerConnection};
|
||||
use crate::proto::fmp::wire::{Msg1Header, Msg2Header, build_msg2};
|
||||
use crate::proto::fmp::{
|
||||
ConnAction, EstablishSnapshot, EstablishView, InboundDecision, InboundReject, OutboundDecision,
|
||||
EstablishSnapshot, EstablishView, InboundDecision, InboundReject, OutboundDecision,
|
||||
OutboundSnapshot, PromotionResult, WireOutcome, cross_connection_winner,
|
||||
};
|
||||
use crate::transport::{Link, LinkDirection, LinkId, ReceivedPacket};
|
||||
@@ -1169,13 +1169,14 @@ impl Node {
|
||||
// cannot collide with an existing entry.
|
||||
self.peer_machines.insert(link_id, machine);
|
||||
|
||||
// Execute `[PromoteToActive]`. The executor calls `promote_connection`
|
||||
// (identical to the pre-refactor `drive_promote_to_active`), feeds
|
||||
// `PromotionResolved{Promoted}` back, and runs the inert
|
||||
// `RegisterDecryptSession` (R2 — register stays in `promote_connection`).
|
||||
// A promote failure (e.g. `MaxPeersExceeded` if peers filled between dial
|
||||
// and msg2) runs the executor's Err cleanup and removes the machine,
|
||||
// leaving it absent (not Established).
|
||||
// Execute `[PromoteToActive]`. The executor calls `promote_connection`,
|
||||
// feeds `PromotionResolved{Promoted}` back, registers the decrypt-worker
|
||||
// session (R1 — C3-3b relocated the register into the executor's
|
||||
// `PromoteToActive` Ok arm, gated on the result), and runs the now-inert
|
||||
// `RegisterDecryptSession` follow-up. A promote failure (e.g.
|
||||
// `MaxPeersExceeded` if peers filled between dial and msg2) runs the
|
||||
// executor's Err cleanup and removes the machine, leaving it absent (not
|
||||
// Established).
|
||||
let ambient = PeerActionCtx {
|
||||
verified_identity: peer_identity,
|
||||
transport_id: packet.transport_id,
|
||||
@@ -1218,36 +1219,6 @@ impl Node {
|
||||
}
|
||||
}
|
||||
|
||||
/// Execute a [`ConnAction::PromoteToActive`] from the establish machine.
|
||||
///
|
||||
/// The decision to promote is made by the establish handlers (and, from the
|
||||
/// establish-core stage on, the pure decision in `proto::fmp`); this is the
|
||||
/// executor half of the seam. It runs the promotion through
|
||||
/// [`Self::promote_connection`], resolving the verified identity and
|
||||
/// promotion timestamp from the ambient wire context, and returns the
|
||||
/// [`PromotionResult`] so the caller can drive the site-specific
|
||||
/// post-promotion tail (TreeAnnounce, bloom mark, discovery-backoff reset,
|
||||
/// loser-link cleanup).
|
||||
///
|
||||
// C3-3a cut the last live caller (the inline outbound `Promote` arm) over to
|
||||
// the executor's `PromoteToActive` path, so this is now unused. Its caller
|
||||
// census / retirement is C3-3b (blueprint § C3-3b); kept here (allowed) until
|
||||
// then so the diff stays scoped to the outbound Promote cutover.
|
||||
#[allow(dead_code)]
|
||||
fn drive_promote_to_active(
|
||||
&mut self,
|
||||
action: ConnAction,
|
||||
verified_identity: PeerIdentity,
|
||||
current_time_ms: u64,
|
||||
) -> Result<PromotionResult, NodeError> {
|
||||
match action {
|
||||
ConnAction::PromoteToActive { link } => {
|
||||
self.promote_connection(link, verified_identity, current_time_ms)
|
||||
}
|
||||
_ => unreachable!("drive_promote_to_active requires a PromoteToActive action"),
|
||||
}
|
||||
}
|
||||
|
||||
/// Promote a connection to active peer after successful authentication.
|
||||
///
|
||||
/// Handles cross-connection detection and resolution using tie-breaker rules.
|
||||
@@ -1396,11 +1367,12 @@ impl Node {
|
||||
"Cross-connection resolved: this connection won"
|
||||
);
|
||||
|
||||
// Hand the FMP recv cipher + replay window to the
|
||||
// decrypt shard worker. (Same as normal-promotion tail
|
||||
// below.)
|
||||
#[cfg(unix)]
|
||||
self.register_decrypt_worker_session(&peer_node_addr);
|
||||
// R1 (C3-3b): the decrypt-worker registration is no longer done
|
||||
// here — it relocated OUT of `promote_connection` into the single
|
||||
// executor `PromoteToActive` Ok arm (`peer_actions.rs`), gated on
|
||||
// the returned `PromotionResult` (`Promoted | CrossConnectionWon`).
|
||||
// The executor runs it synchronously right after this call returns,
|
||||
// before any await, so the live establish behaviour is unchanged.
|
||||
|
||||
Ok(PromotionResult::CrossConnectionWon {
|
||||
loser_link_id,
|
||||
@@ -1506,13 +1478,14 @@ impl Node {
|
||||
"Connection promoted to active peer"
|
||||
);
|
||||
|
||||
// Hand the FMP recv cipher + replay window to the
|
||||
// decrypt shard worker. From this point on the worker
|
||||
// is the sole authority on FMP replay protection for
|
||||
// this session. No-op when the worker pool isn't
|
||||
// spawned (unit-test path or `FIPS_DECRYPT_WORKERS=0`).
|
||||
#[cfg(unix)]
|
||||
self.register_decrypt_worker_session(&peer_node_addr);
|
||||
// R1 (C3-3b): the decrypt-worker registration relocated OUT of
|
||||
// `promote_connection` into the single executor `PromoteToActive` Ok
|
||||
// arm (`peer_actions.rs`), gated on the returned `PromotionResult`
|
||||
// (`Promoted | CrossConnectionWon`, never `CrossConnectionLost`). The
|
||||
// executor runs it synchronously right after this call returns, before
|
||||
// any await — same point, same effect as the pre-refactor in-place call
|
||||
// (no-op when the worker pool isn't spawned; unit-test path or
|
||||
// `FIPS_DECRYPT_WORKERS=0`).
|
||||
|
||||
Ok(PromotionResult::Promoted(peer_node_addr))
|
||||
}
|
||||
|
||||
@@ -312,21 +312,6 @@ pub(crate) enum ConnAction {
|
||||
bytes: Vec<u8>,
|
||||
next_resend_at_ms: u64,
|
||||
},
|
||||
/// Promote the completed handshake connection on `link` to an active peer
|
||||
/// (`promote_connection`): moves the Noise session out of the
|
||||
/// `PeerConnection`, resolves cross-connection precedence via the
|
||||
/// tie-breaker, and installs the `ActivePeer`. The shell executes the
|
||||
/// promotion (resolving the verified identity and promotion timestamp from
|
||||
/// the ambient wire context) and then runs the post-promotion tail
|
||||
/// (TreeAnnounce, bloom mark, discovery-backoff reset, loser-link cleanup).
|
||||
//
|
||||
// C3-3a cut the inline outbound `Promote` arm — the last constructor of this
|
||||
// variant — over to the machine's `PeerAction::PromoteToActive` seam. The only
|
||||
// remaining reference is the (now dead-code-allowed) `drive_promote_to_active`
|
||||
// matcher; both are retired together in C3-3b, so this variant is allowed
|
||||
// until then to keep the C3-3a diff scoped to the outbound Promote cutover.
|
||||
#[allow(dead_code)]
|
||||
PromoteToActive { link: LinkId },
|
||||
}
|
||||
|
||||
/// Read-only view of FMP connection/peer state the lifecycle core needs.
|
||||
|
||||
Reference in New Issue
Block a user