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

node: rewrite planning-phase locators out of source comments

Browse Source 
Comments across the per-peer machine, executor, lifecycle supervisor,
and peering reconciler carried internal rollout labels and design-note
section references. Rewrite them to describe the code's behavior
directly. Comment-text only; no code changes.
This commit is contained in:
Johnathan Corgan
2026-07-14 01:38:52 +00:00
parent 5d5da69a5b
commit 6bebca88ac
15 changed files with 301 additions and 309 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/node/dataplane/dispatch.rs
+1 -1
View File
@@ -187,7 +187,7 @@ impl Node {
// Remove link and address mapping
self.remove_link(&link_id);
// Bound `peer_machines` (C3-2b follow-up #2): drop this peer's machine
// Bound `peer_machines`: drop this peer's machine
// entry, keyed by the `link_id` derived above BEFORE the `peers` removal.
// This cleans up the OLD peer's machine on an inbound restart and prevents
// unbounded growth on the establish success path. NEUTRAL: nothing on the
src/node/dataplane/peer_actions.rs
+42 -42
View File
@@ -1,4 +1,4 @@
//! Executor for the per-peer control machine's [`PeerAction`]s (Step 2 / C3).
//! Executor for the per-peer control machine's [`PeerAction`]s.
//!
//! The per-peer FSM in [`crate::peer::machine`] is a sans-IO reducer: it decides
//! *what* must happen and returns a `Vec<PeerAction>`; this module is the *doing*
@@ -6,18 +6,18 @@
//! stands for (`build_msg2` + `transport.send`, `promote_connection`,
//! `remove_active_peer`, `index_allocator.free`, `note_link_dead`, …).
//!
//! ## C3-1 skeleton (SHADOW-ONLY)
//! ## Shadow-only skeleton
//!
//! This is the **C3-1** increment: the machine home (`Node.peer_machines`), the
//! The machine home (`Node.peer_machines`), the
//! executor, and the disjoint-borrow advance helper. It is **unwired** — the live
//! `handle_msg1`/`handle_msg2` path does not drive it yet, so every method here is
//! `#[allow(dead_code)]`. The inbound cutover (`handle_msg1` → `step(InboundMsg1)`)
//! lands in **C3-2**, the outbound cutover (`handle_msg2` / dial) in **C3-3**.
//! and the outbound cutover (`handle_msg2` / dial) are wired later.
//!
//! Arms the C3 ladder does not yet exercise are inert stubs carrying the sub-commit
//! that realizes them (`OpenTransport`→C3-3, `SendRekey`/`SwapSendState`→C4,
//! `SendLinkMessage`→C4/C5, `SetTimer`/`CancelTimer`→C5 inert, connected-UDP→C6).
//! `RegisterDecryptSession` is a deliberate no-op — see its arm for the C3-2 note.
//! Arms not yet exercised are inert stubs (outbound dial, rekey/crypto installs,
//! link-control frames, timers, and the connected-UDP plane are inert stubs
//! realized as those planes are wired).
//! `RegisterDecryptSession` is a deliberate no-op — see its arm for the note.
use crate::node::Node;
use crate::node::reject::{HandshakeReject, RejectReason};
@@ -37,20 +37,20 @@ use tracing::{debug, trace, 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 `promote_connection`'s ambient args). It is
/// built fresh per driven step by the caller at cutover time (C3-2/C3-3).
/// built fresh per driven step by the caller at cutover time.
#[allow(dead_code)]
pub(in crate::node) struct PeerActionCtx {
/// The authenticated peer identity (GAP-1: `PromoteToActive` /
/// `InvalidateSendState` resolve their `NodeAddr` from this).
/// The authenticated peer identity: `PromoteToActive` /
/// `InvalidateSendState` resolve their `NodeAddr` from this.
pub(in crate::node) verified_identity: PeerIdentity,
/// The transport the exchange is happening over (msg2 send target, decrypt
/// cache-key transport half).
pub(in crate::node) transport_id: TransportId,
/// The peer's wire address (msg2 send target).
pub(in crate::node) remote_addr: TransportAddr,
/// Our session index for this exchange (GAP-2: msg2 framing sender_idx).
/// Our session index for this exchange (msg2 framing sender_idx).
pub(in crate::node) our_index: Option<SessionIndex>,
/// The peer's session index for this exchange (GAP-2: msg2 framing
/// The peer's session index for this exchange (msg2 framing
/// receiver_idx).
pub(in crate::node) their_index: Option<SessionIndex>,
/// The wire timestamp driving this step (promotion ts / loss-report clock).
@@ -69,7 +69,7 @@ impl Node {
/// Advance the machine for `link` by one event and execute the resulting
/// actions.
///
/// The borrow structure the whole seam turns on (spec risk #8): the machine
/// The borrow structure the whole seam turns on: the machine
/// needs `&mut IndexAllocator` as a synchronous capability *while it is
/// itself borrowed mutably out of `peer_machines`*. `peer_machines` and
/// `index_allocator` are **distinct `Node` fields**, so the collect below is
@@ -92,10 +92,10 @@ impl Node {
self.execute_peer_actions(link, ambient, actions).await;
}
/// Map each [`PeerAction`] onto its shell call (spec's executor table).
/// Map each [`PeerAction`] onto its shell call.
///
/// `PromoteToActive` feeds its [`PromotionResult`](crate::proto::fmp::PromotionResult)
/// back into the machine (GAP-1) and appends the follow-up actions to the same
/// back into the machine and appends the follow-up actions to the same
/// worklist — a queue rather than self-recursion so the async executor stays a
/// single flat future (no boxing) and the emitted order is preserved (the
/// establish sequences always end in `PromoteToActive`, so its follow-ups run
@@ -111,21 +111,21 @@ impl Node {
while let Some(action) = queue.pop_front() {
match action {
PeerAction::OpenTransport { .. } => {
// C3-3: outbound dial (`initiate_connection`,
// Outbound dial (`initiate_connection`,
// `lifecycle/mod.rs:470`). Outbound establish is not cut over
// until C3-3; inert in the C3-1 skeleton.
// yet; inert in the shadow-only skeleton.
}
PeerAction::SendHandshake { bytes } => {
// GAP-2: the machine payload is the UNFRAMED Noise msg2 payload;
// The machine payload is the UNFRAMED Noise msg2 payload;
// frame it with our/their index (mirrors `handshake.rs:472`'s
// `build_msg2(our_index, their_index, &payload)`) before the
// wire send. A fresh-outbound msg1 (empty payload → build msg1
// from indices) is framed differently and lands in C3-3.
// from indices) is framed differently and is wired later.
if let (Some(sender_idx), Some(receiver_idx)) =
(ambient.our_index, ambient.their_index)
{
let frame = build_msg2(sender_idx, receiver_idx, &bytes);
// GAP-5: surface the send Result. A missing transport skips
// Surface the send Result. A missing transport skips
// the send and continues (mirrors `handle_msg1`'s
// `if let Some(transport)` guard); a send *error* runs the
// pre-refactor msg2-send-failure cleanup (`handle_msg1`
@@ -157,15 +157,15 @@ impl Node {
}
}
PeerAction::SendRekey { .. } => {
// C4: rekey msg2 framing (`build_msg2(our_new_index, …)`,
// `handshake.rs:365`) + send. Rekey fold is out of C3 scope.
// Rekey msg2 framing (`build_msg2(our_new_index, …)`,
// `handshake.rs:365`) + send. Rekey fold is not yet wired.
}
PeerAction::SendLinkMessage { .. } => {
// C4/C5: encrypt + send a link-control frame (heartbeat / filter
// / tree / disconnect). Data-plane-owned; out of C3 scope.
// Encrypt + send a link-control frame (heartbeat / filter
// / tree / disconnect). Data-plane-owned; not yet wired.
}
PeerAction::PromoteToActive { link: promote_link } => {
// GAP-1: ambient supplies the verified identity + promotion ts
// Ambient supplies the verified identity + promotion ts
// that `promote_connection` needs (resolved from the wire ctx).
match self.promote_connection(
promote_link,
@@ -173,7 +173,7 @@ impl Node {
ambient.now_ms,
) {
Ok(result) => {
// R1 (C3-3b): the decrypt-worker registration relocated
// 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`
@@ -211,7 +211,7 @@ impl Node {
};
queue.extend(follow);
// Defensive cross-connection loser-link surgery (C3-3b).
// Defensive cross-connection loser-link surgery.
// 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
@@ -276,7 +276,7 @@ impl Node {
}
}
Err(e) => {
// GAP-4: promotion failed. `promote_connection` already
// Promotion failed. `promote_connection` already
// removed `connections[link]` and (on error) handled its
// own index internally. The pre-refactor inbound and
// outbound promote-Err arms were NOT byte-identical, so
@@ -295,7 +295,7 @@ impl Node {
// connection").
//
// The transient outbound machine was inserted BEFORE
// execute (Model A); it is additive C3-1 state that
// execute; it is additive state that
// did not exist pre-refactor, so removing the just-
// inserted machine on failure is neutral vs old and
// prevents a leak.
@@ -323,13 +323,13 @@ impl Node {
}
}
PeerAction::SwapSendState { .. } => {
// C4-0: initiator cutover. Reproduces the `ConnAction::Cutover`
// Initiator cutover. Reproduces the `ConnAction::Cutover`
// body in `handlers/rekey.rs:53-88` EXACTLY. `addr` is resolved
// from the ambient verified identity (as `InvalidateSendState`
// does). The decrypt re-register folds HERE, gated on
// `did_cutover` — the generic `RegisterDecryptSession` arm stays a
// no-op so a promote never double-registers. Shadow-only until the
// cadence fold routes here (C4-1).
// cadence fold routes here.
let node_addr = *ambient.verified_identity.node_addr();
let did_cutover = if let Some(peer) = self.peers.get_mut(&node_addr) {
if let Some(_old_our_index) = peer.cutover_to_new_session() {
@@ -373,12 +373,12 @@ impl Node {
let _ = did_cutover;
}
PeerAction::CompleteDrain { peer: node_addr } => {
// C4-0: initiator drain completion. Reproduces the
// Initiator drain completion. Reproduces the
// `ConnAction::Drain` body in `handlers/rekey.rs:90-111` EXACTLY.
// Extract the real previous index + transport_id under the peer
// borrow, drop the borrow, then run the cache_key cleanup (which
// takes &mut self for unregister_decrypt_worker_session).
// Shadow-only until the cadence fold routes here (C4-1).
// Shadow-only until the cadence fold routes here.
let drained = self.peers.get_mut(&node_addr).and_then(|peer| {
peer.complete_drain().map(|idx| (idx, peer.transport_id()))
});
@@ -402,7 +402,7 @@ impl Node {
}
}
PeerAction::InvalidateSendState => {
// GAP-4 (biggest): the FULL teardown. `remove_active_peer`
// The FULL teardown. `remove_active_peer`
// (`dispatch.rs:107`) frees the four index slots
// (current/rekey/pending/previous), drops `peers_by_index`,
// unregisters the decrypt worker, removes the FSP `sessions`
@@ -412,8 +412,8 @@ impl Node {
}
PeerAction::RegisterDecryptSession { index } => {
let _ = index;
// No-op by design. C3-3b (R1-a) relocated the decrypt-worker
// registration into the `PromoteToActive` Ok arm above, gated on
// No-op by design. The decrypt-worker
// registration relocated 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;
@@ -434,12 +434,12 @@ impl Node {
let _ = self.index_allocator.free(index);
}
PeerAction::ActivateConnectedUdp | PeerAction::TeardownConnectedUdp => {
// C6: connected-UDP plane ownership (`connected_udp.rs`).
// Connected-UDP plane ownership (`connected_udp.rs`).
}
PeerAction::SetTimer { .. } | PeerAction::CancelTimer { .. } => {
// C5: timers become actions on the existing quantized tick.
// INERT in C3 — the legacy tick timers still run, so driving
// these would double-schedule (spec risk #7).
// Timers become actions on the existing quantized tick.
// INERT — the legacy tick timers still run, so driving
// these would double-schedule.
}
PeerAction::ReportLost { peer } => {
// The single loss token → the reconciler reflex (`driver.rs:48`).
@@ -450,7 +450,7 @@ impl Node {
}
/// `ReportLost` → `note_link_dead` (kept as a named seam so the ambient clock
/// source is explicit and C5 can thread the reconciler-computed backoff).
/// source is explicit and the reconciler-computed backoff can be threaded later).
#[allow(dead_code)]
fn report_peer_lost(&mut self, peer: NodeAddr, now_ms: u64) {
self.note_link_dead(peer, now_ms);
src/node/handlers/handshake.rs
+40 -40
View File
@@ -279,12 +279,12 @@ impl Node {
InboundDecision::Reject {
reason: InboundReject::AtMaxPeers,
} => {
// C3-2a net-new arm: drive the reject through the machine to
// prove realization A — 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: 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.
debug!(
peer = %self.peer_display_name(&peer_node_addr),
max = self.max_peers(),
@@ -313,7 +313,7 @@ impl Node {
InboundDecision::Reject {
reason: reason @ (InboundReject::PendingSession | InboundReject::DualRekeyWon),
} => {
// Existing-peer rekey rejects — still inline (C4). Byte-unchanged
// Existing-peer rekey rejects — still inline. Byte-unchanged
// from the pre-refactor shared reject tail.
match reason {
InboundReject::PendingSession => debug!(
@@ -440,16 +440,16 @@ impl Node {
self.msg1_rate_limiter.complete_handshake();
}
InboundDecision::RestartThenPromote { peer } => {
// === C3-2b: restart inbound establish, driven by the machine. ===
// === Restart inbound establish, driven by the machine. ===
// Epoch mismatch — the peer restarted. The fresh leg is promoted
// exactly like a net-new inbound (realization A two-phase
// authorize); the OLD peer's teardown is the machine's Phase-1
// exactly like a net-new inbound (two-phase authorize); the OLD
// peer's teardown is the machine's Phase-1
// `[InvalidateSendState, ReportLost{peer}]`:
// InvalidateSendState → remove_active_peer(old): frees the four
// index slots + `peers_by_index` + decrypt unregister + FSP
// `sessions` + `pending_tun_packets` (GAP-4). The fresh leg's
// `sessions` + `pending_tun_packets`. The fresh leg's
// `our_index` is None, so the machine emits NO
// UnregisterDecryptSession (N1).
// UnregisterDecryptSession.
// ReportLost{peer} → note_link_dead(old): reconnect backoff.
// These execute BEFORE authorize/allocate, preserving the
// pre-refactor order exactly (remove_active_peer → note_link_dead →
@@ -473,7 +473,7 @@ impl Node {
// 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 — realization A).
// parks at Handshaking{ReceivedMsg1} (no allocation).
let phase1 = machine.step(
PeerEvent::InboundMsg1 {
link: link_id,
@@ -493,13 +493,13 @@ impl Node {
// Execute the Phase-1 teardown, in emitted order
// (InvalidateSendState before ReportLost, both before
// authorize/alloc). N2 CLOCK NOTE — INTENTIONAL DIVERGENCE: the
// authorize/alloc). CLOCK NOTE — INTENTIONAL DIVERGENCE: the
// pre-refactor arm timestamped `note_link_dead` with
// `SystemTime::now()` wall-clock; routing `ReportLost` through the
// executor uses `ambient.now_ms == packet.timestamp_ms`. This is an
// accepted sub-millisecond reconnect-backoff timing shift — NOT
// on-wire, NOT index/metrics — see design/step2-c3-2-blueprint.md
// N2. The machine is not yet in `peer_machines`, but these two
// on-wire, NOT index/metrics. The machine is not yet in
// `peer_machines`, but these two
// actions do not touch the map, so executing them here is safe.
let teardown_ctx = PeerActionCtx {
verified_identity: peer_identity,
@@ -556,7 +556,7 @@ impl Node {
}
};
// Shell registry surgery (Option A1), in the pre-refactor order:
// Shell registry surgery, in the pre-refactor order:
// set indices on the shell connection, insert link / reverse map /
// connection, then build + store the framed msg2. The old index was
// already freed by `remove_active_peer` above, BEFORE this fresh
@@ -603,14 +603,14 @@ impl Node {
// Established); a send/promote failure removed the machine and
// already cleaned up.
//
// DEFENSIVE CROSS-CONNECTION (risk #5): the machine's
// DEFENSIVE CROSS-CONNECTION: the machine's
// `PromotionResolved{CrossConnectionWon/Lost}` follow-ups run the
// index-level cleanup generically in the executor, but the loser-
// link surgery (close_connection → remove_link → addr_to_link) is
// NOT reproduced here — it is UNREACHABLE on the driven restart
// path: Phase-1 `remove_active_peer` removed `peers[addr]`, so
// `promote_connection` returns `Promoted`. The full cross-connection
// link surgery lands in C3-3 (blueprint risk #5 / N3); the
// link surgery is wired later; the
// debug_assert below catches any regression that reaches a non-
// Established, non-absent state.
debug_assert!(matches!(
@@ -637,13 +637,13 @@ impl Node {
self.msg1_rate_limiter.complete_handshake();
}
InboundDecision::Promote => {
// === C3-2a: net-new inbound establish, driven by the machine. ===
// Realization A (two-phase authorize): Phase 1 classifies with no
// allocation; the shell interposes the late-ACL gate here; Phase 2
// allocates the single index and emits [SendHandshake,
// PromoteToActive]. A rejected/unauthorized msg1 therefore
// allocates NO index — matching the pre-refactor
// authorize-before-allocate ordering exactly.
// === Net-new inbound establish, driven by the machine. ===
// Two-phase authorize: Phase 1 classifies with no allocation; the
// shell interposes the late-ACL gate here; Phase 2 allocates the
// single index and emits [SendHandshake, PromoteToActive]. A
// rejected/unauthorized msg1 therefore allocates NO index —
// 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).
@@ -713,7 +713,7 @@ impl Node {
}
};
// Shell registry surgery (Option A1), in the pre-refactor order:
// Shell registry surgery, in the pre-refactor order:
// set indices on the shell connection, insert link / reverse map /
// connection, then build + store the framed msg2.
conn.set_our_index(our_index);
@@ -740,7 +740,7 @@ impl Node {
// Execute [SendHandshake, PromoteToActive]. The executor frames +
// sends msg2 (bytes identical to `wire_msg2`), promotes via
// `promote_connection`, feeds PromotionResolved back, and runs the
// inert RegisterDecryptSession (R2 — register stays in
// inert RegisterDecryptSession (register stays in
// `promote_connection`). Its send-failure / promote-failure arms
// run the pre-refactor cleanup and remove the machine, leaving it
// absent (not Established).
@@ -1119,22 +1119,22 @@ impl Node {
return;
}
// === C3-3a: net-new outbound establish, driven by the machine. ===
// === Net-new outbound establish, driven by the machine. ===
// ONLY the `establish_outbound == Promote` arm is cut over here. The
// Swap/Keep cross-connection arms and the rekey-msg2 completion branch
// above STAY INLINE (§0): they mutate an existing already-promoted peer
// via `replace_session` with no PeerAction, so the machine's C1 Swap/Keep
// above STAY INLINE: they mutate an existing already-promoted peer
// via `replace_session` with no PeerAction, so the machine's Swap/Keep
// arms cannot be neutral until `PeerSendState` expresses `replace_session`.
//
// This arm is `has_existing_peer == false` only, so `promote_connection`
// always hits its else branch and returns `Promoted`; the defensive
// `CrossConnectionWon/Lost` follow-ups are UNREACHABLE here (their
// loser-link surgery lands in C3-3b). Direct analog of the C3-2a inbound
// loser-link surgery is wired later). Direct analog of the inbound
// net-new arm — no ordering constraint, lowest risk.
//
// Model A: build a TRANSIENT outbound machine, step `Msg2 →
// Build a TRANSIENT outbound machine, step `Msg2 →
// [PromoteToActive]`, execute it (→ `promote_connection` →
// `PromotionResolved{Promoted}` → inert `RegisterDecryptSession`, R2), and
// `PromotionResolved{Promoted}` → inert `RegisterDecryptSession`), and
// insert into `peer_machines` only on the Promoted (Established) tail. The
// outbound `our_index` was allocated at DIAL (unchanged), the outbound
// promote sends nothing on the wire, and `promote_connection` frees
@@ -1146,8 +1146,8 @@ impl Node {
let mut machine = PeerMachine::new_outbound(link_id, peer_identity, packet.timestamp_ms);
// Step `Msg2 → [PromoteToActive]`. The machine re-runs the pure
// `establish_outbound` on the snapshot (a harmless second pure call, as in
// C3-2a); `has_existing_peer == false` reproduces the `Promote` decision.
// `establish_outbound` on the snapshot (a harmless second pure call);
// `has_existing_peer == false` reproduces the `Promote` decision.
let promote_actions = machine.step(
PeerEvent::Msg2 {
their_index: header.sender_idx,
@@ -1165,13 +1165,13 @@ impl Node {
// all returned without inserting). Inserted BEFORE execute so the
// executor's `PromoteToActive` arm can feed `PromotionResolved` back into
// it via the `peer_machines` lookup. The outbound `link_id` was allocated
// at dial and the dial path never inserts a machine (Model A), so this
// at dial and the dial path never inserts a machine, so this
// cannot collide with an existing entry.
self.peer_machines.insert(link_id, machine);
// 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
// session (the register was relocated 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
@@ -1367,7 +1367,7 @@ impl Node {
"Cross-connection resolved: this connection won"
);
// R1 (C3-3b): the decrypt-worker registration is no longer done
// 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`).
@@ -1478,7 +1478,7 @@ impl Node {
"Connection promoted to active peer"
);
// R1 (C3-3b): the decrypt-worker registration relocated OUT of
// 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
src/node/handlers/mmp.rs
+6 -9
View File
@@ -532,17 +532,17 @@ impl Node {
}
}
/// Route a link-dead liveness reap through the peer machine + executor
/// (C5-1). Mirrors [`route_rekey_cadence`](Node::route_rekey_cadence): the
/// Route a link-dead liveness reap through the peer machine + executor.
/// Mirrors [`route_rekey_cadence`](Node::route_rekey_cadence): the
/// shell already decided (the tick sweep's `plan_heartbeats` batch emitted
/// this `ReapPeer` in phase order), so the machine only CONSUMES the decision
/// via [`PeerEvent::LinkDeadSuspected`]. The resulting executor arms
/// (`InvalidateSendState` → `remove_active_peer`, `ReportLost` →
/// `note_link_dead`) reproduce the pre-refactor inline reap body exactly.
///
/// Finding A: an established peer always has a `peer_machine`. If the peer
/// vanished between snapshot and effect, the old inline body was already a
/// no-op, so we return; if the machine is absent (impossible per Finding A)
/// An established peer always has a `peer_machine`. If the peer vanished
/// between snapshot and effect, the old inline body was already a
/// no-op, so we return; if the machine is absent (which should be impossible)
/// we fall back to the byte-identical inline body under a `debug_assert`.
///
/// `now_ms` is the sweep's hoisted wall-clock ms (the same value the old reap
@@ -554,10 +554,7 @@ impl Node {
None => return,
};
if !self.peer_machines.contains_key(&link) {
debug_assert!(
false,
"peer machine present for every established peer (Finding A)"
);
debug_assert!(false, "peer machine present for every established peer");
self.remove_active_peer(&node_addr);
self.note_link_dead(node_addr, now_ms);
return;
src/node/handlers/rekey.rs
+12 -12
View File
@@ -49,8 +49,8 @@ impl Node {
// The shell snapshots each healthy peer's rekey ages/flags (every clock
// read resolved here); the core decides cutover/drain/trigger with no
// clock, phase-grouped to preserve the pre-refactor execution order.
// The batch `poll_rekey` + snapshots STAY SHELL-SIDE and BYTE-UNCHANGED
// (Finding B): the cross-peer phase-grouping (all Cutover → all Drain →
// The batch `poll_rekey` + snapshots STAY SHELL-SIDE and BYTE-UNCHANGED:
// the cross-peer phase-grouping (all Cutover → all Drain →
// all InitiateRekey) governs the shared `index_allocator` free-then-alloc
// SEQUENCE that appears on the wire. The machine must NOT re-poll; it
// CONSUMES each decided `ConnAction` in the same order the batch returned.
@@ -58,7 +58,7 @@ impl Node {
for action in self.fmp.poll_rekey(snapshots, &cfg) {
match action {
// Initiator cutover: route the decided action through the peer
// machine + executor (C4-1). The executor's `SwapSendState` arm
// machine + executor. The executor's `SwapSendState` arm
// reproduces the pre-refactor cutover body EXACTLY.
ConnAction::Cutover { peer: node_addr } => {
self.route_rekey_cadence(node_addr, ConnAction::Cutover { peer: node_addr })
@@ -87,14 +87,14 @@ impl Node {
}
/// Route a cadence-decided `Cutover`/`Drain` `ConnAction` through the peer
/// machine + executor (C4-1). The shell already decided (batch `poll_rekey`);
/// machine + executor. The shell already decided (batch `poll_rekey`);
/// the machine consumes via [`PeerEvent::RekeyConsume`] WITHOUT re-polling,
/// preserving the phase order. The `SwapSendState`/`CompleteDrain` executor
/// arms reproduce the pre-refactor inline effect bodies exactly.
///
/// Finding A: an established peer always has a `peer_machine`. If the peer
/// vanished between snapshot and effect, the old inline body was a no-op, so
/// we do nothing; if the machine is absent (impossible per Finding A) we fall
/// An established peer always has a `peer_machine`. If the peer vanished
/// between snapshot and effect, the old inline body was a no-op, so we do
/// nothing; if the machine is absent (which should be impossible) we fall
/// back to the byte-identical inline body under a `debug_assert`.
async fn route_rekey_cadence(&mut self, node_addr: NodeAddr, action: ConnAction) {
let link = match self.peers.get(&node_addr) {
@@ -104,7 +104,7 @@ impl Node {
if !self.peer_machines.contains_key(&link) {
debug_assert!(
false,
"peer machine present for every established rekey peer (Finding A)"
"peer machine present for every established rekey peer"
);
match action {
ConnAction::Cutover { peer } => self.cutover_peer_inline(&peer),
@@ -124,7 +124,7 @@ impl Node {
}
/// Feed the machine the `RekeyInitiated` observation after the inline
/// `initiate_rekey` (C4-1). The obs emits no action, so there is no executor
/// `initiate_rekey`. The obs emits no action, so there is no executor
/// pass — a bare `step` keeps the machine's control state coherent.
fn observe_rekey_initiated(&mut self, node_addr: &NodeAddr) {
let link = match self.peers.get(node_addr) {
@@ -141,7 +141,7 @@ impl Node {
} else {
debug_assert!(
false,
"peer machine present for every established rekey peer (Finding A)"
"peer machine present for every established rekey peer"
);
}
}
@@ -170,7 +170,7 @@ impl Node {
}
/// Pre-refactor initiator cutover body, retained as the release fallback for
/// the (Finding-A-impossible) missing-machine case. Byte-identical to the old
/// the (should-be-impossible) missing-machine case. Byte-identical to the old
/// inline `ConnAction::Cutover` arm and to the executor's `SwapSendState` arm.
fn cutover_peer_inline(&mut self, node_addr: &NodeAddr) {
let did_cutover = if let Some(peer) = self.peers.get_mut(node_addr) {
@@ -209,7 +209,7 @@ impl Node {
}
/// Pre-refactor drain-completion body, retained as the release fallback for
/// the (Finding-A-impossible) missing-machine case. Byte-identical to the old
/// the (should-be-impossible) missing-machine case. Byte-identical to the old
/// inline `ConnAction::Drain` arm and to the executor's `CompleteDrain` arm.
fn drain_peer_inline(&mut self, node_addr: &NodeAddr) {
// Extract the old index and transport_id under the peer borrow, then drop
src/node/lifecycle/mod.rs
+20 -20
View File
@@ -248,7 +248,7 @@ impl Node {
}
// Collect the auto-connect peer configs and build the mandatory-floor
// reconcile inputs. This is the startup-gate seam (design §3 D4): the
// reconcile inputs. This is the startup-gate seam: the
// substrate is up (transports created, before TUN) but the published
// NodeState is still `Starting`, so pass `Gate::Reconciling` EXPLICITLY
// rather than deriving it from the published state (which would map to
@@ -621,7 +621,7 @@ impl Node {
// Drain each auto-connect transport's discovery buffer (the I/O) and
// apply the driver-only prefilters that read live path-granular state the
// sans-IO core cannot observe (obligation O7): self-skip, the active-peer
// sans-IO core cannot observe: self-skip, the active-peer
// "fresh enough to skip" check, and the "already connecting on this exact
// path" check. The surviving beacons become the opportunistic pool, in
// transport-then-beacon iteration order; the core owns the connected /
@@ -982,7 +982,7 @@ impl Node {
// discovery budget or per-peer cap, only the connected/connecting guard,
// applied in event order.
//
// First-wins per-peer dedup (obligation O7): mdns-sd emits one
// First-wins per-peer dedup: mdns-sd emits one
// `Discovered` event per interface IP of a multi-homed responder, and the
// old inline-dial loop dialed the first compatible address then skipped
// the rest via `is_connecting_to_peer` (which turned true after that
@@ -1183,7 +1183,7 @@ impl Node {
self.supervisor.packet_tx = Some(packet_tx.clone());
self.packet_rx = Some(packet_rx);
// Runtime child-liveness channel (design doc §6). Created before any
// Runtime child-liveness channel. Created before any
// child is spawned so each directly-observable child (TUN threads, the
// DNS task, and the mDNS/Nostr liveness monitor) can clone the sender
// and self-report its `Child` on exit. The sender stored on `self` is
@@ -1252,8 +1252,8 @@ impl Node {
});
// The FSM resolves start-completion health (Full/Degraded/Failed) when
// `Starting.pending` empties and emits it as a `PublishState` action
// (design doc §6/§9.1). Capture that outcome — from the degenerate
// `Starting.pending` empties and emits it as a `PublishState`
// action. Capture that outcome — from the degenerate
// no-children path (published on the `Event::Start` step itself) or from
// the final `SubstrateUp`/`SubstrateFailed` below — to drive the
// start-completion behavior after the spawn loop.
@@ -1651,7 +1651,7 @@ impl Node {
self.initiate_peer_connections().await;
}
// Publish the FSM-resolved start-completion state (design doc §6/§9.1)
// Publish the FSM-resolved start-completion state
// instead of the old unconditional `Running`.
let outcome = start_outcome
.expect("supervisor publishes a start-completion state when bring-up resolves");
@@ -1994,9 +1994,9 @@ impl Node {
///
/// Seeds the FSM at `Running` from observed presence (same pattern as
/// [`Self::stop`]), steps it into `Draining`, and executes the entry
/// actions: broadcast a single shutdown `Disconnect`, and no-op the §8
/// reconciler-gate actions (the reconciler that consumes them lands in
/// Step 1b; the `SetTimer` is likewise a no-op — the bounded wait is the rx
/// actions: broadcast a single shutdown `Disconnect`, and no-op the
/// reconciler-gate actions (the reconciler that consumes them is not yet
/// built; the `SetTimer` is likewise a no-op — the bounded wait is the rx
/// loop's deadline arm). Teardown is deferred to [`Self::finish_shutdown`].
///
/// Called from an rx-loop `select!` arm body: the channel receivers are
@@ -2013,7 +2013,7 @@ impl Node {
let actions = self.supervisor.fsm.step(Event::Drain { deadline_ms });
// Publish the operator-visible `Draining` state (a direct write, like
// the other `self.state` transitions this milestone uses). The
// the other `self.state` transitions this module uses). The
// FSM-owned `PublishState` *action* is not needed for this single
// transition; it arrives with the Full/Degraded health split (c), which
// a direct write cannot express.
@@ -2031,13 +2031,13 @@ impl Node {
// carried for observability only. No-op here.
}
Action::SetPeeringDesired(PeeringDesired::Empty) => {
// §8 reconciler drain-gate (obligation O4): clear the queued
// reconciler drain-gate: clear the queued
// retry schedule so the disconnects the drain itself causes
// cannot leave reconnect entries behind.
self.peering.reconciler.retry_pending.clear();
}
Action::SuspendReplenish => {
// §8 reconciler drain-gate: no extra latch needed. The whole
// reconciler drain-gate: no extra latch needed. The whole
// drain window is `Gate::Suspended`
// (`Gate::from_state(NodeState::Draining)`), so the per-tick
// retry-dial reconcile and every peer-loss reflex read that
@@ -2288,7 +2288,7 @@ impl Node {
///
/// The driver builds the [`DiscoveryPools`] overlay input from
/// `bootstrap.cached_open_discovery_candidates(64)` (the I/O), excluding the
/// node's own advert (obligation O6 — the sans-IO core has no self-identity
/// node's own advert (the sans-IO core has no self-identity
/// input), and supplies the configured-npub set, the per-npub cooldown set,
/// and the startup-sweep max-age. `max_age_secs` is `None` for the per-tick
/// sweep and `Some(startup_sweep_max_age_secs)` for the one-shot startup
@@ -2297,8 +2297,8 @@ impl Node {
/// The core reproduces the old sweep's full skip order, configured-advert
/// expedite, and enqueue budget internally: it inserts due-now entries into
/// the relocated `retry_pending`, and the retry-slot `process_pending_retries`
/// dials them later in the same tick (two-phase, design §2.5). Per design §10
/// this calls the reconciler's `reconcile_overlay` (the overlay layer only) —
/// dials them later in the same tick (two-phase). This
/// calls the reconciler's `reconcile_overlay` (the overlay layer only) —
/// NOT the monolithic `reconcile()` — so the always-on retry-dial phase does
/// not re-fire at this slot (which would double-dial the due entries and
/// apply the per-tick 16-cap twice). The returned `ScheduleRetry` actions
@@ -2392,7 +2392,7 @@ impl Node {
let budget = self.build_peering_budget();
let gate = Gate::from_state(self.supervisor.state);
// Two-phase enqueue (design §2.5): reconcile_overlay inserts due-now
// Two-phase enqueue: reconcile_overlay inserts due-now
// entries into retry_pending; the retry slot dials them. The core emits
// a `ScheduleRetry` for exactly the NEW enqueues (the configured-advert
// expedite bumps `retry_after_ms` without emitting), which is precisely
@@ -2524,7 +2524,7 @@ impl Node {
/// Build the reconciler [`Policy`] from config. `auto_connect_peers` is
/// filled by the caller: the startup floor and the reflex wrappers pass the
/// configured auto-connect set; the per-tick retry-dial slot passes an empty
/// set so the config floor stays silent (cadence contract, design §3 D4).
/// set so the config floor stays silent (cadence contract).
pub(in crate::node) fn build_peering_policy(
&self,
auto_connect_peers: Vec<PeerConfig>,
@@ -2559,7 +2559,7 @@ impl Node {
///
/// The `connected` / `connecting` sets gate the floor, retry-dial, overlay,
/// and LAN layers; `in_flight_by_peer` feeds the opportunistic layer's
/// per-peer parallel cap (obligation O7), computed exactly as the deleted
/// per-peer parallel cap, computed exactly as the deleted
/// `path_candidate_attempt_budget` did: `connections(expected == addr) +
/// pending_connects(addr)`. The scalar counts stay unpopulated at the
/// ceiling-only posture (no layer reads them; see [`Observed`]).
@@ -2591,7 +2591,7 @@ impl Node {
/// Build the admission [`Budget`] from the live maps. This is the surviving
/// home for the slot arithmetic; the shared helpers it wraps stay until the
/// overlay/opportunistic cutovers consume them (obligation O5).
/// overlay/opportunistic cutovers consume them.
pub(in crate::node) fn build_peering_budget(&self) -> Budget {
let peer_slots = if self.max_peers() == 0 {
usize::MAX
src/node/lifecycle/supervisor.rs
+37 -37
View File
@@ -1,4 +1,4 @@
//! Node lifecycle supervisor — sans-IO core (Milestone-1 Step 1a).
//! Node lifecycle supervisor — sans-IO core.
//!
//! A synchronous `step(event) -> Vec<Action>` finite-state machine over the
//! fixed set of substrate children. It owns the *decision* of what to bring up
@@ -8,11 +8,11 @@
//! holds no runtime handles — time enters only as inputs (a future `Tick`/
//! `DrainDeadlineElapsed`, added with the `Draining` phase) — so it is
//! unit-testable with synthetic sequences and survives a later thread-boundary
//! move (design doc §6 Core 1, §8 "cores are sans-IO").
//! move (cores are sans-IO).
//!
//! ## Scope: the behavior-neutral rewrite
//!
//! This is the first of the three Step-1a commits and is strictly
//! This module is strictly
//! behavior-preserving. The machine mirrors today's `start()`/`stop()` exactly:
//!
//! - every configured child is spawned in the current order, and optional
@@ -37,14 +37,14 @@
//! [`Action::SetPeeringDesired`], [`Action::SuspendReplenish`]), and the new
//! published [`NodeState::Draining`](crate::node::NodeState::Draining) —
//! written directly by the driver at drain entry, exactly like the other
//! `self.state` transitions this milestone uses. The existing immediate `Stop`
//! `self.state` transitions this module uses. The existing immediate `Stop`
//! path is untouched. `Draining` and `Stop` share a single teardown-plan author
//! (`begin_stopping`), so the teardown ordering is defined once.
//!
//! ## Scope: the `Running{Full|Degraded}` + `Failed` health split (this commit)
//!
//! This commit lands the operator-visible start-completion health policy
//! (design doc §6/§9.1) and, with it, the FSM-owned [`Action::PublishState`]:
//! This module implements the operator-visible start-completion health policy
//! and, with it, the FSM-owned [`Action::PublishState`]:
//!
//! - [`SupState::Running`] now carries a [`Health`] (`Full` or `Degraded`), and
//! [`SupState::Failed`] is the fatal path. When `Starting.pending` empties (or
@@ -64,7 +64,7 @@
//! **not** the old immediate-`Running`.
//!
//! Runtime child-liveness monitoring (a `ChildExited` event re-routing health
//! when a task/thread dies at runtime) is **deferred** (design doc §7): §9.1's
//! when a task/thread dies at runtime) is **deferred**: start-completion health
//! resolution is start-framed, and liveness monitoring is a substantial unbuilt
//! mechanism. This commit is start-time health only.
@@ -76,7 +76,7 @@ use crate::node::NodeState;
use crate::transport::{PacketTx, TransportId};
use crate::upper::tun::{TunOutboundRx, TunTx};
/// A supervised substrate child (design doc §6 Core 1).
/// A supervised substrate child.
///
/// Each transport is an individual child keyed by its id so the later
/// required-vs-optional health policy can reason about partial N-of-M bring-up.
@@ -104,7 +104,7 @@ pub(crate) enum Child {
/// An input to the supervisor. Results of executing [`Action`]s are fed back as
/// `SubstrateUp` / `SubstrateFailed` / `ChildStopped`.
///
/// `Tick` (design doc §6) is **deferred** (the per-tick reconciler backstop lands
/// `Tick` is **deferred** (the per-tick reconciler backstop lands
/// with the cadence work). `ChildExited` is present: it feeds runtime
/// child-liveness monitoring, routing health the same way a start failure does.
#[derive(Clone, Debug, PartialEq, Eq)]
@@ -164,8 +164,8 @@ pub(crate) enum Event {
/// The child that has been torn down.
child: Child,
},
/// A supervised child's task or thread exited on its own at runtime (design
/// doc §6) — not in response to a `StopChild`. Valid from `Running`; routes
/// A supervised child's task or thread exited on its own at runtime — not
/// in response to a `StopChild`. Valid from `Running`; routes
/// health the same way a start failure does (the last transport out →
/// `Failed`, an optional child out → `Degraded`), but at runtime `Failed` is
/// a published health signal, not a teardown — the driver keeps serving. No
@@ -176,9 +176,9 @@ pub(crate) enum Event {
},
}
/// A driver-scheduled timer the supervisor can arm (design doc §6). Only the
/// drain deadline exists for now; the handshake/rekey/liveness timers named in
/// §8 arrive with later cores.
/// A driver-scheduled timer the supervisor can arm. Only the
/// drain deadline exists for now; the handshake/rekey/liveness timers
/// arrive with later cores.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum Timer {
/// Fires when the bounded drain window closes. The driver feeds
@@ -187,20 +187,20 @@ pub(crate) enum Timer {
DrainDeadline,
}
/// The reconciler's desired peering set (design doc §8 drain gate). Only
/// `Empty` is needed in this commit; the populated variants that the Step-1b
/// The reconciler's desired peering set (drain gate). Only
/// `Empty` is needed for now; the populated variants that the
/// homeostatic reconciler converges toward land with that core.
#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) enum PeeringDesired {
/// No peers desired. Entering `Draining` sets this so the reconciler stops
/// reconnecting the peers the drain just closed (§8: "Draining switches the
/// reconnecting the peers the drain just closed ("Draining switches the
/// homeostat off").
Empty,
}
/// An effect the driver must perform. The core never performs I/O itself.
///
/// `PublishState` (design doc §6) lands here, with the `Running{Full|Degraded}`
/// `PublishState` lands here, with the `Running{Full|Degraded}`
/// health split: the start-completion health outcome is a fork
/// (`Full`/`Degraded`/`Failed`) that a single direct `self.state` write cannot
/// express, so the machine authors it as an action. The driver keeps its direct
@@ -217,7 +217,7 @@ pub(crate) enum Action {
/// no-children path) to carry the resolved health outcome —
/// [`NodeState::Running`] (Full), [`NodeState::Degraded`], or
/// [`NodeState::Failed`] — to the driver, which writes it to the published
/// state (design doc §6/§9.1).
/// state.
PublishState(NodeState),
/// Tear down this child (the driver performs the stop / join I/O and
/// reports `ChildStopped`).
@@ -230,16 +230,16 @@ pub(crate) enum Action {
/// and owns the bounded drain wait, so this is a documented no-op beyond
/// bookkeeping.
SetTimer(Timer, u64),
/// Set the reconciler's desired peering set (§8 drain gate). Documented
/// **no-op in this commit** — the reconciler that consumes it lands in
/// Step 1b; the driver logs/ignores it for now.
/// Set the reconciler's desired peering set (drain gate). Documented
/// **no-op for now** — the reconciler that consumes it is not yet
/// built; the driver logs/ignores it for now.
SetPeeringDesired(PeeringDesired),
/// Suspend peer replenishment (§8 drain gate). Documented **no-op in this
/// commit** for the same reason as `SetPeeringDesired`.
/// Suspend peer replenishment (drain gate). Documented **no-op for
/// now** for the same reason as `SetPeeringDesired`.
SuspendReplenish,
}
/// Start-completion health (design doc §9.1). Resolved once when
/// Start-completion health. Resolved once when
/// `Starting.pending` empties: `Full` iff every configured child came up,
/// `Degraded` iff ≥1 transport is up but some configured optional child failed.
/// Zero transports up is not a health — it is the fatal [`SupState::Failed`].
@@ -256,7 +256,7 @@ pub(crate) enum Health {
},
}
/// Reason for the fatal [`SupState::Failed`] state (design doc §9.1).
/// Reason for the fatal [`SupState::Failed`] state.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum FailReason {
/// Zero transports came up at start completion. Without a transport the node
@@ -265,7 +265,7 @@ pub(crate) enum FailReason {
NoTransports,
}
/// Internal supervisor state (design doc §6). Richer than the published
/// Internal supervisor state. Richer than the published
/// [`NodeState`](crate::node::NodeState): `Starting`/`Stopping` carry the set of
/// children still resolving, and `Running` carries the resolved [`Health`].
#[derive(Clone, Debug, PartialEq, Eq)]
@@ -283,13 +283,13 @@ pub(crate) enum SupState {
/// Resolved start-completion health.
health: Health,
},
/// Start completed with zero transports up (design doc §9.1) — fatal. The
/// Start completed with zero transports up — fatal. The
/// driver tears down any children that did come up and returns an error.
Failed {
/// Why the start failed.
reason: FailReason,
},
/// Bounded graceful-drain window (design doc §6/§8). Broadcast Disconnect
/// Bounded graceful-drain window. Broadcast Disconnect
/// has gone out and the reconciler is gated off (desired peering set
/// emptied, replenishment suspended); teardown begins when
/// `DrainDeadlineElapsed` arrives. Logically sits between `Running` and
@@ -446,7 +446,7 @@ impl SupervisorFsm {
self.failed.clear();
// A node with no children at all resolves health immediately. Zero
// transports up → `Failed` (design doc §9.1; this is the behavioral
// transports up → `Failed` (this is the behavioral
// change from the old immediate-`Running`).
if order.is_empty() {
return vec![Action::PublishState(self.resolve_start_health())];
@@ -473,7 +473,7 @@ impl SupervisorFsm {
}
fn on_substrate_failed(&mut self, child: Child) -> Vec<Action> {
// Record the failed child (design doc §9.1): a configured child that
// Record the failed child: a configured child that
// failed to start drives the `Degraded` determination when `pending`
// empties. It drains from `pending` and never joins the up-set.
let SupState::Starting { pending } = &mut self.state else {
@@ -489,7 +489,7 @@ impl SupervisorFsm {
}
}
/// Resolve start-completion health (design doc §9.1). Called once when
/// Resolve start-completion health. Called once when
/// `Starting.pending` empties (or the degenerate no-children path); the
/// classification is shared with runtime child-exit via
/// [`Self::classify_health`].
@@ -509,7 +509,7 @@ impl SupervisorFsm {
///
/// Worker-pool failures are captured in `failed` like any other optional
/// child, so they contribute `Degraded` (never `Failed`) — the inline crypto
/// fallback keeps the node correct without the pools (design doc §9.1).
/// fallback keeps the node correct without the pools.
fn classify_health(&mut self) -> NodeState {
let transports_up = self
.up
@@ -552,8 +552,8 @@ impl SupervisorFsm {
self.state = SupState::Draining { deadline_ms };
// Drain entry, in order: broadcast the shutdown Disconnect, arm the
// deadline timer, then gate the reconciler off (desired = ∅, suspend
// replenishment) so it cannot reconnect the peers the drain just closed
// (§8). The up-set is left intact for the eventual teardown plan.
// replenishment) so it cannot reconnect the peers the drain just
// closed. The up-set is left intact for the eventual teardown plan.
vec![
Action::BroadcastDisconnect,
Action::SetTimer(Timer::DrainDeadline, deadline_ms),
@@ -595,7 +595,7 @@ impl SupervisorFsm {
Vec::new()
}
/// A supervised child exited on its own at runtime (design doc §6). Only
/// A supervised child exited on its own at runtime. Only
/// meaningful while `Running`: startup (`Starting`), drain (`Draining`), and
/// teardown (`Stopping`) own their own child bookkeeping through the
/// `pending` / `up` sets and the `SubstrateUp` / `SubstrateFailed` /
@@ -603,7 +603,7 @@ impl SupervisorFsm {
///
/// The exit is routed exactly like a start-time failure via
/// [`Self::classify_health`] — the last transport out → `Failed`, an optional
/// child out → `Degraded{+child}` — the runtime analogue of the §9.1
/// child out → `Degraded{+child}` — the runtime analogue of the
/// start-time policy. Two differences from start: `Failed` here is a
/// published health signal only (the driver keeps serving, per the resolved
/// runtime policy — no auto-teardown), and there is no restart (the FSM has
src/node/mod.rs
+9 -9
View File
@@ -170,20 +170,20 @@ pub enum NodeState {
Created,
/// Starting up (initializing transports).
Starting,
/// Fully operational — every configured child came up (design doc §9.1).
/// Fully operational — every configured child came up.
Running,
/// Operational but degraded (design doc §9.1): ≥1 transport is up and the
/// Operational but degraded: ≥1 transport is up and the
/// node is serving, but one or more configured optional children (a
/// transport beyond the first, Nostr, mDNS, TUN, DNS, or a worker pool)
/// failed to start. Still operational — a degraded node serves traffic.
Degraded,
/// Bounded graceful drain in progress (design doc §6): a shutdown
/// Bounded graceful drain in progress: a shutdown
/// `Disconnect` has been broadcast and the node is waiting for peers to
/// clear (bounded by `node.drain_timeout_secs`) before teardown. Not
/// operational; the daemon drain path advances to `Stopping` via the
/// supervisor's `DrainDeadlineElapsed`, never through `stop()`.
Draining,
/// Start failed fatally: zero transports came up (design doc §9.1). The
/// Start failed fatally: zero transports came up. The
/// driver tears down any children that did come up and `start()` returns
/// an error. Not operational and not restartable in-process.
Failed,
@@ -195,13 +195,13 @@ pub enum NodeState {
impl NodeState {
/// Check if node is operational. A `Degraded` node is operational — it is
/// serving, just missing an optional child (design doc §9.1).
/// serving, just missing an optional child.
pub fn is_operational(&self) -> bool {
matches!(self, NodeState::Running | NodeState::Degraded)
}
/// Check if node can be started. A `Failed` node is not restartable
/// in-process (design doc §9.1) — only a fresh `Created` or a cleanly
/// in-process — only a fresh `Created` or a cleanly
/// `Stopped` node can start.
pub fn can_start(&self) -> bool {
matches!(self, NodeState::Created | NodeState::Stopped)
@@ -353,14 +353,14 @@ pub struct Node {
/// Indexed by LinkId since we don't know the peer's identity yet.
connections: HashMap<LinkId, PeerConnection>,
// === Per-Peer Control Machines (Step 2 / C3) ===
// === Per-Peer Control Machines ===
/// Per-peer lifecycle control FSMs, keyed by the stable `LinkId` that spans
/// the handshake→active lifetime. A NEW parallel structure introduced by the
/// node-runtime decomposition: `connections`/`peers` stay byte-unchanged (hot
/// path pristine) and are cut over to this machine home path-by-path. Unwired
/// in C3-1 — the executor (`dataplane/peer_actions.rs`) and advance helper
/// initially — the executor (`dataplane/peer_actions.rs`) and advance helper
/// exist but the live `handle_msg1`/`handle_msg2` path does not drive them yet;
/// the inbound cutover lands in C3-2.
/// the inbound cutover is wired later.
#[allow(dead_code)]
peer_machines: HashMap<LinkId, PeerMachine>,
src/node/peering/driver.rs
+11 -11
View File
@@ -21,10 +21,10 @@ impl Node {
/// Reflex: an outbound handshake timed out (replaces the old
/// `Node::schedule_retry` call sites).
///
/// Replicates `schedule_retry`'s connected-guard (obligation O2) — the pure
/// core cannot observe the peers map, so the driver drops the event when the
/// peer is already connected — then feeds the gate-guarded reconciler reflex
/// with the gate derived from the live published state (obligation O4).
/// Replicates `schedule_retry`'s connected-guard — the pure core cannot
/// observe the peers map, so the driver drops the event when the peer is
/// already connected — then feeds the gate-guarded reconciler reflex with
/// the gate derived from the live published state.
pub(in crate::node) fn note_handshake_timeout(&mut self, node_addr: NodeAddr, now_ms: u64) {
if self.peers.contains_key(&node_addr) {
return;
@@ -44,7 +44,7 @@ impl Node {
/// No connected-guard — the peer is already gone by the time a link-dead /
/// disconnect event fires (`schedule_reconnect` had none). The gate is
/// derived from the live published state so a drain self-suppresses the
/// reconnect (obligation O4, design §8 correctness trap).
/// reconnect.
pub(in crate::node) fn note_link_dead(&mut self, node_addr: NodeAddr, now_ms: u64) {
let policy =
self.build_peering_policy(self.config().auto_connect_peers().cloned().collect());
@@ -63,16 +63,16 @@ impl Node {
/// (bumping their `retry_after_ms` past the handshake window). This driver
/// performs the advert-refetch + dial I/O each emitted `Connect` names, and
/// on an immediate dial error feeds the `on_handshake_timeout` reflex so the
/// optimistic re-fire suppression is overwritten by proper backoff
/// (obligation O3). During a drain the gate is `Suspended`, so the reconcile
/// clears the schedule and emits nothing (the design §8 gate).
/// optimistic re-fire suppression is overwritten by proper backoff. During a
/// drain the gate is `Suspended`, so the reconcile clears the schedule and
/// emits nothing.
pub(in crate::node) async fn process_pending_retries(&mut self, now_ms: u64) {
if self.peering.reconciler.retry_pending.is_empty() {
return;
}
// Retry-dial cadence slot: empty config floor (design §3 D4) and empty
// discovery pools, so only the retry-dial phase acts.
// Retry-dial cadence slot: empty config floor and empty discovery
// pools, so only the retry-dial phase acts.
let policy = self.build_peering_policy(Vec::new());
let observed = self.observe_peering();
let budget = self.build_peering_budget();
@@ -139,7 +139,7 @@ impl Node {
});
}
// Immediate failure counts as an attempt: overwrite the
// optimistic re-fire suppression with backoff (obligation O3).
// optimistic re-fire suppression with backoff.
self.note_handshake_timeout(node_addr, now_ms);
}
}
src/node/peering/reconcile.rs
+58 -63
View File
@@ -1,4 +1,4 @@
//! Peering homeostatic reconciler — sans-IO core (Milestone-1 Step 1b).
//! Peering homeostatic reconciler — sans-IO core.
//!
//! A synchronous `reconcile(inputs) -> Vec<PeeringAction>` decision core over
//! the node's peer set. It owns the *decision* of which peers to dial and how
@@ -10,10 +10,10 @@
//! I/O, and holds no runtime handles — time enters only as the `now` input, and
//! the live dataplane maps enter only as an immutable [`Observed`] snapshot — so
//! it is unit-testable with synthetic inputs and survives a later
//! thread-boundary move (design doc §6 Core 2, §8 "cores are sans-IO"). It
//! mirrors the [`crate::node::lifecycle::supervisor`] template.
//! thread-boundary move. It mirrors the
//! [`crate::node::lifecycle::supervisor`] template.
//!
//! ## Scope: the four reconcile layers (behavior-neutral rewrite)
//! ## The four reconcile layers
//!
//! [`PeeringReconciler::reconcile`] runs four layers in priority order, each
//! subsuming today's imperative `Node` methods verbatim:
@@ -24,7 +24,7 @@
//! the retry-dial phase is gated only by [`Budget::admission_ok`], matching
//! today.
//! 2. **Overlay pool** — Nostr open-discovery, **ceiling-only** enqueue with no
//! set-point floor (design doc §9.2 option b; subsumes
//! set-point floor (option b; subsumes
//! `run_open_discovery_sweep`). Enqueues into the durable retry schedule; the
//! dial happens on a later retry-slot invocation (the two-phase timing,
//! below).
@@ -33,10 +33,10 @@
//! (subsumes `poll_lan_rendezvous`, connected/connecting skip only).
//! 4. **Ceiling** — not a separate pass; the `node.limits` triple plus the
//! per-tick and per-peer caps are enforced inline in every layer through the
//! [`Budget`] the driver builds. "Any limb binds → stop growing" (design
//! §6:673). Ceiling-only posture: never emits [`PeeringAction::Disconnect`].
//! [`Budget`] the driver builds. "Any limb binds → stop growing".
//! Ceiling-only posture: never emits [`PeeringAction::Disconnect`].
//!
//! ### Two-phase overlay timing (design §2.5)
//! ### Two-phase overlay timing
//!
//! The durable `retry_pending` schedule survives between the reconcile
//! invocations at the two relevant cadence slots. The overlay layer at the
@@ -49,24 +49,21 @@
//! ### Cadence contract (driver responsibility)
//!
//! Behavior-neutrality across the tick depends on the driver populating only the
//! input relevant to each cadence slot (design §2 intro): the config-peer floor
//! input relevant to each cadence slot: the config-peer floor
//! runs only when `policy.auto_connect_peers` is non-empty (the startup
//! peer-connect seam, design §3 D4); the overlay/opportunistic layers run only
//! peer-connect seam); the overlay/opportunistic layers run only
//! when their pools are populated. The driver also excludes the node's own
//! identity and the driver-only "candidate fresh enough to skip" /
//! "already connecting on this exact path" predicates when building the pools —
//! those read the live freshness cache / connection table at path granularity,
//! which the pure core does not observe (see the deviation notes in the C2
//! disposition).
//! which the pure core does not observe.
//!
//! ## Scope: the drain gate (consumed by the C3 cutover)
//! ## The drain gate
//!
//! The `Gate` derived from the published `NodeState` gates the whole core: a
//! `Suspended` (draining) gate clears the retry schedule and returns no actions,
//! and the reflexes self-suppress, so the drain does not reconnect the peers it
//! just closed (design §8 correctness trap). `NotRunning` is the inert startup
//! gate. This core is **unwired** in this commit — nothing calls it on the hot
//! path yet; the driver cutover lands in the following commits.
//! just closed. `NotRunning` is the inert startup gate.
use std::collections::{HashMap, HashSet};
@@ -80,8 +77,8 @@ use crate::transport::{TransportAddr, TransportId};
use super::retry::RetryState;
/// Drain/run gate derived from the supervisor's published [`NodeState`]
/// (design §6). Not a bare bool: it must also express "not yet running" (the
/// Drain/run gate derived from the supervisor's published [`NodeState`].
/// Not a bare bool: it must also express "not yet running" (the
/// startup gate) distinctly from "draining" (the suspend gate).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum Gate {
@@ -95,7 +92,7 @@ pub(crate) enum Gate {
}
impl Gate {
/// Map the published [`NodeState`] to the gate (design §6:636):
/// Map the published [`NodeState`] to the gate:
/// `Running`/`Degraded` → `Reconciling`; `Draining` → `Suspended`;
/// everything else (`Created`/`Starting`/`Stopping`/`Stopped`/`Failed`) →
/// `NotRunning`.
@@ -112,7 +109,7 @@ impl Gate {
}
}
/// Admission observations the ceiling needs (design §6:654), built by the driver
/// Admission observations the ceiling needs, built by the driver
/// from the live maps at each cadence point. This is the only place the deleted
/// `available_outbound_slots` / `outbound_handshake_slots` / `outbound_link_slots`
/// arithmetic survives.
@@ -137,31 +134,30 @@ pub(crate) struct Budget {
}
/// What the reconciler observes of today's dataplane maps (it never owns them).
/// Peer/leg identity is anonymous-capable so re-applying over
/// refactor-node-next's XX first-contact path stays neutral (design §7:1181).
/// Peer/leg identity is anonymous-capable so the XX first-contact path stays
/// neutral.
#[derive(Clone, Debug, Default)]
pub(crate) struct Observed {
/// `self.peers.len()`.
// The scalar counts are carried for the ceiling's future set-point use
// (design §6:654). At Step 1b's ceiling-only posture the ceiling is enforced
// through `Budget` (admission arithmetic) and the per-peer cap through
// `in_flight_by_peer`, so no reconcile layer reads these scalars — they
// remain unread after all three driver cutovers, awaiting the Step 2
// set-point.
// The scalar counts are carried for the ceiling's future set-point use.
// At the ceiling-only posture the ceiling is enforced through `Budget`
// (admission arithmetic) and the per-peer cap through `in_flight_by_peer`,
// so no reconcile layer reads these scalars — they remain unread for now,
// awaiting a future set-point.
#[allow(dead_code)]
// ceiling-only posture: no layer reads the scalar counts (Step 2 set-point)
// ceiling-only posture: no layer reads the scalar counts (future set-point)
pub peers: usize,
/// `self.connections.len()`.
#[allow(dead_code)]
// ceiling-only posture: no layer reads the scalar counts (Step 2 set-point)
// ceiling-only posture: no layer reads the scalar counts (future set-point)
pub connections: usize,
/// `self.links.len()`.
#[allow(dead_code)]
// ceiling-only posture: no layer reads the scalar counts (Step 2 set-point)
// ceiling-only posture: no layer reads the scalar counts (future set-point)
pub links: usize,
/// `self.pending_connects.len()`.
#[allow(dead_code)]
// ceiling-only posture: no layer reads the scalar counts (Step 2 set-point)
// ceiling-only posture: no layer reads the scalar counts (future set-point)
pub pending_connects: usize,
/// The `peers` map keys (fully authenticated peers).
pub connected: HashSet<NodeAddr>,
@@ -173,7 +169,7 @@ pub(crate) struct Observed {
pub in_flight_by_peer: HashMap<NodeAddr, usize>,
}
/// A dialable candidate. Mirrors next's discovery tuple exactly
/// A dialable candidate. Mirrors the discovery tuple exactly
/// (`(TransportId, TransportAddr, Option<PeerIdentity>, bool)`), identity
/// anonymous-capable.
#[derive(Clone, Debug)]
@@ -185,9 +181,9 @@ pub(crate) struct Candidate {
pub transport_id: TransportId,
/// The remote address to dial.
pub remote_addr: TransportAddr,
/// The peer identity; `None` is an anonymous first-contact leg (design §7).
/// The peer identity; `None` is an anonymous first-contact leg.
pub identity: Option<PeerIdentity>,
/// The trailing bool in next's tuple: whether this is an active-refresh dial
/// The trailing bool in the tuple: whether this is an active-refresh dial
/// against an already-connected peer.
pub active_refresh: bool,
}
@@ -217,7 +213,7 @@ impl Candidate {
}
}
/// Read-only candidate pools the driver drained this tick (design §6:650). Plain
/// Read-only candidate pools the driver drained this tick. Plain
/// data, no runtime handles.
#[derive(Clone, Debug, Default)]
pub(crate) struct DiscoveryPools {
@@ -236,7 +232,7 @@ pub(crate) struct DiscoveryPools {
pub startup_sweep_max_age_secs: Option<u64>,
}
/// The desired-state policy (design §5 knobs). Built once per invocation from
/// The desired-state policy knobs. Built once per invocation from
/// config.
#[derive(Clone, Debug)]
pub(crate) struct Policy {
@@ -268,11 +264,11 @@ pub(crate) struct Policy {
pub open_discovery_max_pending: usize,
/// `advert_ttl_secs * 1000 * OPEN_DISCOVERY_RETRY_LIFETIME_MULTIPLIER`.
// NOTE: there is deliberately NO set-point N knob. The overlay pool is
// CEILING-ONLY (design §9.2 option b).
// CEILING-ONLY.
pub open_discovery_expires_ms: u64,
}
/// The action vocabulary (design §6:643). `Connect` carries the
/// The action vocabulary. `Connect` carries the
/// anonymous-capable [`Candidate`]; `ScheduleRetry` reports the retry-schedule
/// delta the reconciler owns (for driver observability; the durable mutation is
/// internal).
@@ -280,9 +276,9 @@ pub(crate) struct Policy {
pub(crate) enum PeeringAction {
/// Dial this candidate (driver performs the connect / handshake I/O).
Connect(Candidate),
/// Shed this peer. Reserved and unused at the ceiling-only posture of Step
/// 1b (design §9.3 refuse-to-grow, no shed).
#[allow(dead_code)] // Step 2: shedding/eviction emits this
/// Shed this peer. Reserved and unused at the ceiling-only posture
/// (refuse-to-grow, no shed).
#[allow(dead_code)] // shedding/eviction will emit this
Disconnect(NodeAddr),
/// A retry was (re)scheduled for `peer` at `backoff_ms` delay. Observability
/// only; the durable mutation is on the reconciler's `retry_pending`.
@@ -336,7 +332,7 @@ pub(crate) struct OverlaySweepTally {
pub(crate) struct PeeringReconciler {
/// Cross-attempt retry schedule (moved off `Node.retry_pending`). The
/// `retry_count` lives here, not per-connection, so escalating backoff
/// survives a fresh connection per re-dial (design §6:704 / §7:1150). Keyed
/// survives a fresh connection per re-dial. Keyed
/// by [`NodeAddr`].
pub(in crate::node) retry_pending: HashMap<NodeAddr, RetryState>,
}
@@ -354,11 +350,11 @@ impl PeeringReconciler {
now: u64,
gate: Gate,
) -> Vec<PeeringAction> {
// Gate prologue (design §2.0).
// Gate prologue.
match gate {
// Startup gate: inert before the substrate is ready (design §2.3).
// Startup gate: inert before the substrate is ready.
Gate::NotRunning => return Vec::new(),
// Drain gate: clear the schedule and desire nothing (design §2.3).
// Drain gate: clear the schedule and desire nothing.
Gate::Suspended => {
self.retry_pending.clear();
return Vec::new();
@@ -369,7 +365,7 @@ impl PeeringReconciler {
let mut actions = Vec::new();
// Layer 1: mandatory floor. The config floor precedes the retry-dial
// phase; the retry-dial phase precedes the overlay enqueue so a
// same-call overlay insert never dials in its own call (two-phase, §2.5).
// same-call overlay insert never dials in its own call (two-phase).
self.layer_config_floor(policy, observed, &mut actions);
self.layer_retry_dial(policy, observed, budget, now, &mut actions);
// Layer 2: overlay pool (ceiling-only enqueue). The tally is
@@ -389,7 +385,7 @@ impl PeeringReconciler {
actions
}
/// Per-layer wrapper: run **only** the overlay-enqueue layer (design §10).
/// Per-layer wrapper: run **only** the overlay-enqueue layer.
///
/// The monolithic [`reconcile`] runs the always-on retry-dial phase on every
/// call, so the driver must NOT call it at the overlay (nostr-poll) cadence
@@ -401,9 +397,9 @@ impl PeeringReconciler {
/// On `NotRunning` / `Suspended` it returns no actions. It does **not**
/// clear `retry_pending` on `Suspended` — that clear is owned by the drain
/// gate (`enter_drain`) and the retry-slot [`reconcile`], not the overlay
/// slot (design §10). The returned [`PeeringAction::ScheduleRetry`] items are
/// slot. The returned [`PeeringAction::ScheduleRetry`] items are
/// observability only; the durable mutation is the insert the layer performed
/// into `retry_pending`, dialed at the later retry slot (two-phase, §2.5).
/// into `retry_pending`, dialed at the later retry slot (two-phase).
pub(in crate::node) fn reconcile_overlay(
&mut self,
policy: &Policy,
@@ -433,8 +429,7 @@ impl PeeringReconciler {
(actions, tally)
}
/// Per-layer wrapper: run **only** the opportunistic-growth layer (design
/// §10).
/// Per-layer wrapper: run **only** the opportunistic-growth layer.
///
/// The monolithic [`reconcile`] runs the always-on retry-dial phase on every
/// call, so the driver must NOT call it at the opportunistic (transport
@@ -448,7 +443,7 @@ impl PeeringReconciler {
/// On `NotRunning` / `Suspended` it returns no actions. It does **not** clear
/// `retry_pending` on `Suspended` — that clear is owned by the drain gate
/// (`enter_drain`) and the retry-slot [`reconcile`], not the opportunistic
/// slot (design §10). `policy` and `now` are accepted for wrapper-family
/// slot. `policy` and `now` are accepted for wrapper-family
/// uniformity with [`reconcile`] / [`reconcile_overlay`]; the ceiling-only
/// opportunistic layer reads neither (it grows only against the live
/// [`Budget`] / [`Observed`], with no time- or config-derived input).
@@ -475,7 +470,7 @@ impl PeeringReconciler {
/// Emits a `Connect` for every auto-connect peer not already connected or
/// connecting. Not budget-gated (the startup floor is mandatory); the driver
/// populates `policy.auto_connect_peers` only at the startup peer-connect
/// seam so this does not re-fire every tick (design §3 D4).
/// seam so this does not re-fire every tick.
fn layer_config_floor(
&self,
policy: &Policy,
@@ -571,9 +566,9 @@ impl PeeringReconciler {
/// Layer 2 — overlay pool enqueue (subsumes `run_open_discovery_sweep`).
///
/// Ceiling-only: the sole bound is the enqueue budget (pending cap ∧
/// available outbound slots). No `>= N` set-point floor (design §9.2 option
/// b). Enqueues due-now retry entries; the dial happens on a later
/// retry-slot invocation (two-phase, §2.5).
/// available outbound slots). No `>= N` set-point floor. Enqueues due-now
/// retry entries; the dial happens on a later
/// retry-slot invocation (two-phase).
#[allow(clippy::too_many_arguments)]
fn layer_overlay_enqueue(
&mut self,
@@ -593,7 +588,7 @@ impl PeeringReconciler {
// open_discovery_enqueue_budget: cap_remaining ∧ available_outbound_slots
// (== handshake_slots ∧ peer_slots). Note this is min(cap, handshake,
// peer) — the real helper uses available_outbound_slots, NOT peer_slots
// alone (see deviation note).
// alone.
let current_open_discovery_pending = self
.retry_pending
.values()
@@ -783,11 +778,11 @@ impl PeeringReconciler {
/// Reflex: an outbound handshake timed out (== `Node::schedule_retry`,
/// retry.rs:58). Byte-identical retry_count/backoff math. No-op when the gate
/// is `Suspended` (drain trap, §2.3, review decision D1).
/// is `Suspended` (drain trap).
///
/// NOTE: today's `schedule_retry` also returns early if the peer is already
/// connected (`self.peers.contains_key`). The reflex takes no [`Observed`]
/// (D1 keeps the signature `(addr, now, policy, gate)`), so that
/// (the reflex signature is `(addr, now, policy, gate)`), so that
/// connected-guard is the driver's responsibility at the call site.
pub(crate) fn on_handshake_timeout(
&mut self,
@@ -848,7 +843,7 @@ impl PeeringReconciler {
/// Reflex: a link went dead (== `Node::schedule_reconnect`, retry.rs:134).
/// Byte-identical retry_count/backoff math, including preserving accumulated
/// backoff across repeated link-dead events. No-op when the gate is
/// `Suspended` (drain trap, §2.3, review decision D1).
/// `Suspended` (drain trap).
pub(crate) fn on_link_dead(
&mut self,
addr: NodeAddr,
@@ -1051,7 +1046,7 @@ mod tests {
let _ = overlay_addr;
// auto_connect_peers stays empty: these reconcile calls model non-startup
// cadence slots, where the config floor does not run (design §3 D4).
// cadence slots, where the config floor does not run.
let mut policy = base_policy();
policy.open_discovery_enabled = true;
@@ -1178,7 +1173,7 @@ mod tests {
assert_eq!(st.retry_after_ms, 1_000 + backoff_ms(0, BASE_MS, CAP_MS));
// A later retry-slot reconcile (now past retry_after) dials once. The
// retry-slot policy has an empty config floor (cadence contract, §3 D4),
// retry-slot policy has an empty config floor (cadence contract),
// so the single Connect comes only from the retry-dial phase.
let now = 1_000 + backoff_ms(0, BASE_MS, CAP_MS);
let tick_policy = base_policy();
src/node/peering/retry.rs
+1 -1
View File
@@ -11,7 +11,7 @@
use crate::config::PeerConfig;
/// Per-tick cap on retry-dial connection attempts (design §6 ceiling).
/// Per-tick cap on retry-dial connection attempts (ceiling).
pub(in crate::node) const MAX_RETRY_CONNECTIONS_PER_TICK: usize = 16;
/// Tracks retry state for a peer across connection attempts.
src/node/session/mod.rs
+1 -1
View File
@@ -1269,7 +1269,7 @@ mod overlapping_epoch_tests {
// ========================================================================
// Rekey-policy characterization (pins `check_session_rekey`'s decision
// boundaries before the `Fsp::poll_rekey` hoist — these thresholds have no
// other test module; see plan §10).
// other test module).
// ========================================================================
/// The initiator liveness-cutover delay used by `check_session_rekey`
src/node/tests/mod.rs
+1 -1
View File
@@ -33,7 +33,7 @@ pub(super) fn make_node() -> Node {
/// A test node that reaches `Full` health on `start()`.
///
/// A default [`make_node`] configures no transports, so its `start()` now
/// resolves to `NodeState::Failed` (zero transports up, design doc §9.1) and
/// resolves to `NodeState::Failed` (zero transports up) and
/// returns `NoOperationalTransports`. Lifecycle-state tests that need a running
/// node build one with a single loopback UDP transport (ephemeral port) as the
/// sole configured child — DNS disabled — so bring-up has exactly one
src/node/tests/unit.rs
+3 -3
View File
@@ -151,8 +151,8 @@ async fn test_try_peer_addresses_races_all_concrete_udp_candidates() {
#[tokio::test]
async fn test_node_state_transitions() {
// A transport-less node now resolves to `Failed` on start (design doc
// §9.1), so exercise state transitions with a genuinely healthy node.
// A transport-less node now resolves to `Failed` on start, so exercise
// state transitions with a genuinely healthy node.
let mut node = make_healthy_node();
assert!(!node.is_running());
@@ -170,7 +170,7 @@ async fn test_node_state_transitions() {
#[tokio::test]
async fn test_transportless_start_fails_and_publishes_failed() {
// The intended behavioral change (design doc §9.1): a node with zero
// The intended behavioral change: a node with zero
// transports up cannot serve, so `start()` returns `NoOperationalTransports`
// and leaves the published state at `Failed` (not operational, not
// restartable in-process).
src/peer/machine.rs
+59 -59
View File
@@ -1,11 +1,11 @@
//! Per-peer FMP control FSM (sans-IO reducer).
//!
//! The unified per-peer lifecycle state machine that Step 2 of the node-runtime
//! decomposition folds the scattered `connections`/`peers`/rekey state carriers
//! into. This module is the **C1** increment: the FSM types, the machine struct
//! (control-tier state only), and the pure `step` reducer, plus its unit tests.
//! It is **unwired** — nothing in the codebase calls it yet; the driver wiring
//! (C3b) and the send-state boundary (C2) land in later commits.
//! The unified per-peer lifecycle state machine that folds the scattered
//! `connections`/`peers`/rekey state carriers into one place. It provides the
//! FSM types, the machine struct (control-tier state only), and the pure `step`
//! reducer, plus its unit tests. It is **unwired** — nothing in the codebase
//! calls it yet; the driver wiring and the send-state boundary land in later
//! commits.
//!
//! ## Shape
//!
@@ -14,7 +14,7 @@
//! [`crate::proto::fmp`] ([`Fmp::establish_inbound`]/[`establish_outbound`],
//! [`Fmp::poll_timeouts`]/[`poll_resends`]/[`poll_rekey`]/[`poll_rekey_resends`],
//! and `cross_connection_winner`) — this module writes **no new decision
//! core** (R4). The machine only (a) builds the plain-data snapshots those cores
//! core**. The machine only (a) builds the plain-data snapshots those cores
//! consume from its control-tier state, (b) maps the returned
//! [`ConnAction`]/[`InboundDecision`]/[`OutboundDecision`]/[`PromotionResult`]
//! into the [`PeerAction`] vocabulary the driver executes, and (c) advances its
@@ -22,16 +22,16 @@
//! registry surgery, late ACL `authorize_peer`, decrypt-worker register/unregister)
//! are **emitted as actions**, never performed here.
//!
//! ## Control / send-state split (§6 Core 3)
//! ## Control / send-state split
//!
//! The machine holds **control-tier** state only. The hot send-critical state
//! (the three epoch slots, transport target, connected-UDP handle, hot counters)
//! becomes `PeerSendState` in **C2** and is *not* built here; the machine emits
//! becomes `PeerSendState` and is *not* built here; the machine emits
//! actions (`PromoteToActive`, `SwapSendState`, `RegisterDecryptSession`, …) that
//! the driver applies to the published send-state. `remote_epoch` is
//! establish-path-only, hence control-tier, and lives here.
//!
//! ## C1 realizability notes (see `design/step2-machine-spec.md` caveats)
//! ## Realizability notes
//!
//! - `SendHandshake`/`SendRekey`/`SendLinkMessage` carry **opaque bytes**
//! (`Vec<u8>`) — the driver applies outer wire framing / encryption. On the
@@ -39,16 +39,16 @@
//! rekey msg2 they are the Noise payload the shell already produced
//! ([`WireOutcome::msg2_payload`]). A fresh outbound msg1 has no bytes the
//! control machine can build (the Noise step is shell-side), so it is emitted
//! with an empty payload and a `C3b` note — that path is not exercised by the
//! C1 tests.
//! with an empty payload and a note that the driver fills it in — that path is
//! not exercised by the tests.
//! - `SendLinkMessage { msg }` is opaque plaintext (there is **no** unifying
//! `LinkMessage` type in the tree today — heartbeat is a bare `[0x51]` byte,
//! while filter/tree/disconnect are distinct concrete types). The machine
//! builds the real heartbeat and disconnect frames; filter/tree announce
//! payloads are data-plane-owned and threaded in at C3b (empty here).
//! payloads are data-plane-owned and threaded in by the driver (empty here).
//! - `PeerSnapshot::counter` (the Noise send counter) is a send-state fact the
//! control machine cannot see; it is passed as `0` (the message-count rekey
//! trigger is threaded from `PeerSendState` at C4). Irrelevant to every C1
//! trigger is threaded from `PeerSendState`). Irrelevant to every
//! test (cutover/drain ignore it).
#![allow(dead_code)]
@@ -66,8 +66,8 @@ use crate::{NodeAddr, PeerIdentity};
// ============================================================================
// Timing placeholders
//
// C1 is unwired; the real intervals come from `NodeConfig` when the driver is
// wired (C3b/C5). The `poll_*` cores already take the interval/backoff as
// This module is unwired; the real intervals come from `NodeConfig` when the
// driver is wired. The `poll_*` cores already take the interval/backoff as
// arguments, so these are only used to compute `SetTimer{at_ms}` deadlines and
// the `Closed{backoff_deadline_ms}` park time. The unit tests assert on timer
// *kinds*, not exact deadlines.
@@ -88,7 +88,7 @@ const REKEY_DAMPEN_MS: u64 = 30_000;
const CLOSED_BACKOFF_MS: u64 = 5_000;
// ============================================================================
// FSM types (spec §1)
// FSM types
// ============================================================================
/// The unified per-peer lifecycle state (subsumes today's `HandshakeState`,
@@ -224,12 +224,12 @@ pub(crate) enum PeerEvent {
},
/// Inbound rekey msg2 (completes our initiated rekey).
RekeyMsg2 { their_index: SessionIndex },
/// A cadence-decided rekey `ConnAction` to CONSUME (C4-1). The shell ran the
/// A cadence-decided rekey `ConnAction` to CONSUME. The shell ran the
/// batch `poll_rekey` across the whole peer set (phase-grouped, index-order
/// preserving — Finding B) and routes each decided action here; the machine
/// preserving) and routes each decided action here; the machine
/// applies the control-tier transition + emits the send-state write
/// (`SwapSendState`/`CompleteDrain`) WITHOUT re-polling. Carries only
/// `Cutover`/`Drain` on the C4-1 path (`InitiateRekey` stays inline shell-side
/// `Cutover`/`Drain` (`InitiateRekey` stays inline shell-side
/// with a [`RekeyInitiated`](PeerEvent::RekeyInitiated) observation).
RekeyConsume { action: ConnAction },
/// OBSERVATION: the shell initiated an outbound rekey inline (the Noise msg1
@@ -237,7 +237,7 @@ pub(crate) enum PeerEvent {
/// `Maintaining{Rekey(Msg1Sent)}` so the next tick's `Cutover`/`Drain` consume
/// transitions from a coherent phase. Emits no action.
RekeyInitiated,
/// Data plane observed the responder K-bit flip inline (§3.7).
/// Data plane observed the responder K-bit flip inline.
PeerKbitFlip { epoch: [u8; 8] },
/// A filter announce is due for this peer.
FilterAnnounce,
@@ -271,7 +271,7 @@ pub(crate) enum PeerAction {
/// Transmit rekey handshake bytes.
SendRekey { bytes: Vec<u8> },
/// Transmit an (encrypted) plaintext link-control frame. Opaque `Vec<u8>`
/// pending a unifying `LinkMessage` type — C3b: type against
/// pending a unifying `LinkMessage` type; a future revision could type against
/// `proto::bloom::FilterAnnounce` / `proto::stp::TreeAnnounce` /
/// `proto::fmp::Disconnect` / a heartbeat marker.
SendLinkMessage { msg: Vec<u8> },
@@ -308,11 +308,11 @@ pub(crate) enum PeerAction {
}
// ============================================================================
// The machine (control tier — spec §2)
// The machine (control tier)
// ============================================================================
/// Per-peer control FSM. Holds control-tier lifecycle state only; the
/// send-critical state is published as `PeerSendState` (C2) and mutated via the
/// send-critical state is published as `PeerSendState` and mutated via the
/// emitted [`PeerAction`]s.
pub(crate) struct PeerMachine {
state: PeerState,
@@ -323,7 +323,7 @@ pub(crate) struct PeerMachine {
conn: ConnectionState,
/// Remote startup epoch (establish-path-only; NOT in send-state).
remote_epoch: Option<[u8; 8]>,
/// Inbound two-phase authorize (realization A): the opaque Noise msg2
/// Inbound two-phase authorize: the opaque Noise msg2
/// payload stashed in Phase 1 (`InboundMsg1`) and emitted in Phase 2
/// (`on_authorized`), so a rejected/unauthorized msg1 allocates no index.
pending_msg2_payload: Option<Vec<u8>>,
@@ -348,7 +348,7 @@ pub(crate) struct PeerMachine {
// The machine owns decrypt-worker register/unregister via actions, so it
// tracks which index it registered (to later unregister/free). This is
// control knowledge of the registration lifecycle, distinct from the hot
// send-state slots (C2).
// send-state slots.
/// The currently-registered decrypt index (post-establish).
our_index: Option<SessionIndex>,
/// The previous index held open during a post-cutover drain window.
@@ -414,7 +414,7 @@ impl PeerMachine {
/// The index we allocated for this peer's inbound session, once Phase 2
/// (`on_authorized`) has run. `None` before allocation (and after a
/// rejected/unauthorized msg1). The inbound cutover reads this to perform
/// the shell registry surgery with the machine-owned index (Option A1).
/// the shell registry surgery with the machine-owned index.
pub(crate) fn our_index(&self) -> Option<SessionIndex> {
self.our_index
}
@@ -464,7 +464,7 @@ impl PeerMachine {
PeerEvent::RekeyConsume { action } => self.map_rekey_action(action, now),
PeerEvent::RekeyInitiated => self.on_rekey_initiated(),
PeerEvent::PeerKbitFlip { .. } => {
// Responder cutover is data-plane-owned (§3.7): the machine only
// Responder cutover is data-plane-owned: the machine only
// schedules the drain-window unregister. NO slot mutation.
vec![PeerAction::SetTimer {
kind: TimerKind::DrainExpiry,
@@ -472,12 +472,12 @@ impl PeerMachine {
}]
}
PeerEvent::FilterAnnounce => vec![PeerAction::SendLinkMessage {
// C3b: filter-announce payload is data-plane-owned; threaded in
// Filter-announce payload is data-plane-owned; threaded in
// at wiring time.
msg: Vec::new(),
}],
PeerEvent::TreeAnnounceDue => vec![PeerAction::SendLinkMessage {
// C3b: tree-announce payload is data-plane-owned.
// Tree-announce payload is data-plane-owned.
msg: Vec::new(),
}],
PeerEvent::PeerHeard => self.on_peer_heard(now),
@@ -490,7 +490,7 @@ impl PeerMachine {
}
// ------------------------------------------------------------------
// Outbound establish (§3.1)
// Outbound establish
// ------------------------------------------------------------------
fn on_dial(
@@ -504,7 +504,7 @@ impl PeerMachine {
}
self.conn.set_transport_id(transport_id);
// Connection-oriented transports open the transport first; connectionless
// ones send msg1 immediately. C1 models the connection-oriented arm
// ones send msg1 immediately. This models the connection-oriented arm
// (OpenTransport) — the reconciler's candidate carries the transport
// kind at wiring time; connectionless dial reuses `start_handshake`.
self.state = PeerState::Connecting { link: self.link };
@@ -537,8 +537,8 @@ impl PeerMachine {
/// 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; C1 emits an empty payload (see
/// module note). This path is not exercised by the C1 tests.
/// the driver when it executes this action; an empty payload is emitted (see
/// module note). This path is not exercised by the tests.
fn start_outbound_handshake(&mut self, now: u64) -> Vec<PeerAction> {
let bytes = Vec::new();
self.state = PeerState::Handshaking {
@@ -606,7 +606,7 @@ impl PeerMachine {
}
// ------------------------------------------------------------------
// Inbound establish (§3.2)
// Inbound establish
// ------------------------------------------------------------------
fn on_inbound_msg1(
@@ -649,7 +649,7 @@ impl PeerMachine {
}
}
/// Inbound **Phase 1** (realization A): classify the fresh leg *without*
/// Inbound **Phase 1**: classify the fresh leg *without*
/// allocating an index. Records identity/epoch/their-index and stashes the
/// opaque msg2 payload, parking at `Handshaking{ReceivedMsg1}` — the
/// "awaiting Authorized" marker. The index allocation and the msg2/promote
@@ -668,10 +668,10 @@ impl PeerMachine {
Vec::new()
}
/// Inbound **Phase 2** (realization A): the late-ACL gate passed shell-side.
/// Inbound **Phase 2**: the late-ACL gate passed shell-side.
/// Allocate our index NOW — the single inbound allocation point — record it
/// on `conn`, and emit the msg2 send + promotion. `RegisterDecryptSession`
/// follows on the `PromotionResolved{Promoted}` feedback (§3.2). Guarded to
/// follows on the `PromotionResolved{Promoted}` feedback. Guarded to
/// the inbound `ReceivedMsg1` phase so the benign outbound `Authorized`
/// confirmation stays a no-op (state `Handshaking{SentMsg1}` and every other
/// state fall through to `Vec::new()`).
@@ -744,7 +744,7 @@ impl PeerMachine {
}
// ------------------------------------------------------------------
// Promotion feedback (§3.3)
// Promotion feedback
// ------------------------------------------------------------------
fn on_promotion_resolved(&mut self, result: PromotionResult, now: u64) -> Vec<PeerAction> {
@@ -787,7 +787,7 @@ impl PeerMachine {
}
// ------------------------------------------------------------------
// Rekey (initiator) + cutover (§3.4)
// Rekey (initiator) + cutover
// ------------------------------------------------------------------
fn on_rekey_msg2(&mut self, their_index: SessionIndex) -> Vec<PeerAction> {
@@ -804,7 +804,7 @@ impl PeerMachine {
Vec::new()
}
/// OBS (C4-1): the shell ran `initiate_rekey` inline — the Noise msg1 leaf,
/// Observation: the shell ran `initiate_rekey` inline — the Noise msg1 leaf,
/// the index allocation, the wire send, and the `set_rekey_state` on the
/// `ActivePeer` all happened shell-side. This is a pure observation that
/// advances the machine's control state to `Maintaining{Rekey(Msg1Sent)}` so
@@ -882,8 +882,8 @@ impl PeerMachine {
// Clear the shadow `draining_index` set by the Cutover arm: the
// real previous index is now retired by `CompleteDrain`, so a
// leftover `Some(stale)` would double-free if a later
// `CrossConnectionWon` consumed it in `on_promotion_resolved`
// (C4-0 latent item 1). Post-rekey cross-connection promotion is
// `CrossConnectionWon` consumed it in `on_promotion_resolved`.
// Post-rekey cross-connection promotion is
// not a live path, but clearing here removes the hazard outright.
self.draining_index = None;
self.state = PeerState::Active { addr: peer };
@@ -891,7 +891,7 @@ impl PeerMachine {
}
ConnAction::InitiateRekey { peer } => {
// Fresh outbound rekey: allocate our new index, send msg1 (Noise
// leaf is shell-side → empty payload in C1), arm the resend timer.
// leaf is shell-side → empty payload here), arm the resend timer.
self.rekey_in_progress = true;
self.rekey_resend_count = 0;
self.rekey_msg1 = Some(Vec::new());
@@ -961,7 +961,7 @@ impl PeerMachine {
}
// ------------------------------------------------------------------
// Liveness (§3.5)
// Liveness
// ------------------------------------------------------------------
fn on_heartbeat_due(&mut self, now: u64) -> Vec<PeerAction> {
@@ -1007,9 +1007,9 @@ impl PeerMachine {
}
// `InvalidateSendState` maps to the executor's `remove_active_peer`, which
// unregisters the decrypt worker by the REAL current index. The machine's
// shadow `our_index` is deliberately NOT used to unregister here (C5-0): it
// shadow `our_index` is deliberately NOT used to unregister here: it
// can drift to a reused index and wrongly unregister ANOTHER peer's worker
// session. `TeardownConnectedUdp` is inert (C6 — the old reap had no
// session. `TeardownConnectedUdp` is inert (the old reap had no
// connected-UDP teardown, so inert is neutral); `ReportLost` drives the
// loss reflex (`note_link_dead`).
let mut actions = vec![
@@ -1026,7 +1026,7 @@ impl PeerMachine {
}
// ------------------------------------------------------------------
// Timeout / teardown / close (§3.6)
// Timeout / teardown / close
// ------------------------------------------------------------------
fn on_timeout(&mut self, kind: TimerKind, now: u64) -> Vec<PeerAction> {
@@ -1082,7 +1082,7 @@ impl PeerMachine {
let snap = self.conn_snapshot();
// poll_timeouts emits [ScheduleRetry?, Teardown]; the machine REMAPS
// ScheduleRetry -> ReportLost (single loss token) and Teardown ->
// FreeIndex{our_index}, emitting FreeIndex before ReportLost (§3.6).
// FreeIndex{our_index}, emitting FreeIndex before ReportLost.
let mut free = Vec::new();
let mut lost = Vec::new();
for act in Fmp::new().poll_timeouts(vec![snap]) {
@@ -1131,8 +1131,8 @@ impl PeerMachine {
fn on_tick(&mut self, now: u64) -> Vec<PeerAction> {
// The driver evaluates due machine timers on the quantized tick and
// re-enters the Timeout{kind} handlers. C1 is unwired; the deadline
// bookkeeping is threaded from the driver at C5, so Tick is a no-op here.
// re-enters the Timeout{kind} handlers. This module is unwired; the
// deadline bookkeeping is threaded from the driver, so Tick is a no-op here.
let _ = now;
Vec::new()
}
@@ -1189,7 +1189,7 @@ impl PeerMachine {
}
/// Build this peer's rekey snapshot from control-tier state. `counter` is a
/// send-state fact (C4); passed as 0 here (see module note).
/// send-state fact; passed as 0 here (see module note).
fn peer_snapshot(&self, addr: NodeAddr, now: u64) -> PeerSnapshot {
let phase = match self.state {
PeerState::Maintaining {
@@ -1229,8 +1229,8 @@ fn disconnect_frame(reason: CloseReason) -> Vec<u8> {
}
// ============================================================================
// Unit tests (spec §5) — assert on ACTION SEQUENCES + STATE transitions using
// hand-built synthetic snapshots (caveat #4: no real crypto sessions).
// Unit tests — assert on ACTION SEQUENCES + STATE transitions using
// hand-built synthetic snapshots (no real crypto sessions).
// ============================================================================
#[cfg(test)]
@@ -1475,7 +1475,7 @@ mod tests {
);
// Phase 1: restart tail only (invalidate, unregister old, report lost),
// then park at ReceivedMsg1 — no index allocated yet (realization A).
// then park at ReceivedMsg1 — no index allocated yet.
assert_eq!(
actions,
vec![
@@ -1632,7 +1632,7 @@ mod tests {
let wire = wire_outcome(peer, Some([4u8; 8]), 0x77);
// Phase 1 (InboundMsg1): classify only — no actions, no allocation,
// parked at ReceivedMsg1 (realization A).
// parked at ReceivedMsg1.
let phase1 = m.step(
PeerEvent::InboundMsg1 {
link: LinkId::new(1),
@@ -1893,7 +1893,7 @@ mod tests {
// action exists in the PeerAction vocabulary at all (reconciler-owned).
}
// ---- Test 9: cadence CONSUME (C4-1) -----------------------------------
// ---- Test 9: cadence CONSUME ------------------------------------------
// The shell polls the batch `poll_rekey` and routes each decided ConnAction
// as `RekeyConsume` — the machine maps it WITHOUT re-polling, yielding the
// same action sequence + transition as the machine-driven cadence (Test 1),
@@ -1944,7 +1944,7 @@ mod tests {
assert_eq!(m.draining_index, Some(SessionIndex::new(0x1111)));
// Consume the shell-decided Drain: single CompleteDrain, Active, and the
// shadow drain index is CLEARED (double-free guard, C4-0 latent item 1).
// shadow drain index is CLEARED (double-free guard).
let drain = m.step(
PeerEvent::RekeyConsume {
action: ConnAction::Drain { peer: addr },
@@ -1957,7 +1957,7 @@ mod tests {
assert_eq!(m.draining_index, None);
}
// ---- Test 10: RekeyInitiated observation (C4-1) -----------------------
// ---- Test 10: RekeyInitiated observation ------------------------------
// The shell ran `initiate_rekey` inline; the obs advances control state to
// Msg1Sent and emits nothing.
#[test]