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

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

Browse Source 
# Conflicts:
#	src/node/dataplane/dispatch.rs
#	src/node/dataplane/peer_actions.rs
#	src/node/handlers/handshake.rs
#	src/node/handlers/mmp.rs
#	src/node/handlers/rekey.rs
#	src/node/mod.rs
#	src/peer/machine.rs
This commit is contained in:
Johnathan Corgan
2026-07-14 01:40:01 +00:00
parent 1e934e7603 6bebca88ac
commit 22a5f1f807
15 changed files with 208 additions and 215 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/node/dataplane/dispatch.rs
+2 -2
View File
@@ -187,10 +187,10 @@ impl Node {
// Remove link and address mapping
self.remove_link(&link_id);
// Finding A: drop this peer's inert machine, keyed by the link_id
// Drop this peer's inert machine, keyed by the link_id
// derived above (a peer's link_id is immutable, so the key never
// moved). Keeps peers <-> peer_machines in exact correspondence on
// teardown. NEUTRAL — nothing reads peer_machines yet.
// teardown. NEUTRAL: nothing reads peer_machines yet.
self.peer_machines.remove(&link_id);
if let Some(transport_id) = transport_id {
self.cleanup_bootstrap_transport_if_unused(transport_id);
src/node/dataplane/peer_actions.rs
+18 -18
View File
@@ -1,4 +1,4 @@
//! Executor for the per-peer control machine's [`PeerAction`]s (Step 2 / M2).
//! 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,14 +6,14 @@
//! stands for (`build_msg2` + `transport.send`, `promote_connection`,
//! `remove_active_peer`, `index_allocator.free`, `note_link_dead`, …).
//!
//! ## M2 skeleton (SHADOW-ONLY)
//! ## Shadow-only skeleton
//!
//! This is the **M2** 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** — no live
//! handler path drives the machine and nothing inserts into `peer_machines`, so
//! every method here is `#[allow(dead_code)]` and behaviorally inert.
//!
//! The arm bodies that a later reap/rekey/timer step will drive for real are
//! The arm bodies that a later reap/rekey/timer commit will drive for real are
//! ported 1:1 from the IK-lineage executor and adapted to next's Node API — they
//! reproduce next's inline shell bodies exactly:
//!
@@ -30,8 +30,8 @@
//! for the establish promote it stays INSIDE `promote_connection` (unlike the
//! IK lineage), so `PromoteToActive` does NOT re-register.
//!
//! Actions not yet exercised are inert stubs carrying the step that realizes
//! them: the establish send-path (`SendHandshake` framing) lands when the inbound
//! Actions not yet exercised are inert stubs: the establish send-path
//! (`SendHandshake` framing) lands when the inbound
//! establish send is wired; `SendRekey`/`SendLinkMessage`, the timers
//! (`SetTimer`/`CancelTimer`), the outbound dial (`OpenTransport`), and the
//! connected-UDP plane are inert here.
@@ -62,7 +62,7 @@ use tracing::{debug, info, 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_msg3`'s
/// `wire`/`packet` locals and `promote_connection`'s ambient args). It is built
/// fresh per driven step by the caller at cutover time (a later establish step).
/// fresh per driven step by the caller at cutover time.
#[allow(dead_code)]
pub(in crate::node) struct PeerActionCtx {
/// The authenticated peer identity (`PromoteToActive` / `InvalidateSendState`
@@ -119,7 +119,7 @@ impl Node {
self.execute_peer_actions(link, ambient, actions).await;
}
/// Map each [`PeerAction`] onto its shell call (the executor table).
/// Map each [`PeerAction`] onto its shell call.
///
/// The worklist is a `VecDeque` rather than self-recursion so the async
/// executor stays a single flat future (no boxing) and the emitted order is
@@ -138,22 +138,22 @@ impl Node {
match action {
PeerAction::OpenTransport { .. } => {
// Outbound dial (`initiate_connection`). Outbound establish is
// not cut over yet; inert in the M2 skeleton.
// not cut over yet; inert in the shadow-only skeleton.
}
PeerAction::SendHandshake { .. } => {
// Inbound establish send-path: frame the unframed Noise msg2
// payload with our/their index (`build_msg2(our_index,
// their_index, &payload)`) and send, with the msg2-send-failure
// cleanup + queue abort. Lands with `PromoteToActive` when the
// inbound establish path is wired; inert in the M2 skeleton.
// inbound establish path is wired; inert in the shadow-only skeleton.
}
PeerAction::SendRekey { .. } => {
// Rekey msg framing (`build_msg2(our_new_index, …)`) + send.
// Rekey fold is out of M2 scope.
// Rekey fold is out of scope here.
}
PeerAction::SendLinkMessage { .. } => {
// Encrypt + send a link-control frame (heartbeat / filter /
// tree / disconnect). Data-plane-owned; out of M2 scope.
// tree / disconnect). Data-plane-owned; out of scope here.
}
PeerAction::PromoteToActive { link: promote_link } => {
// Establish promote, driven through the machine. Transcribes
@@ -505,12 +505,12 @@ impl Node {
let _ = self.index_allocator.free(index);
}
PeerAction::ActivateConnectedUdp | PeerAction::TeardownConnectedUdp => {
// Connected-UDP plane ownership (`connected_udp.rs`). Out of M2
// scope.
// Connected-UDP plane ownership (`connected_udp.rs`). Out of
// scope for now.
}
PeerAction::SetTimer { .. } | PeerAction::CancelTimer { .. } => {
// Timers become actions on the existing quantized tick. INERT in
// M2 — the legacy tick timers still run, so driving these would
// Timers become actions on the existing quantized tick. INERT:
// the legacy tick timers still run, so driving these would
// double-schedule.
}
PeerAction::ReportLost { peer } => {
@@ -522,8 +522,8 @@ impl Node {
}
/// `ReportLost` → `note_link_dead` (kept as a named seam so the ambient clock
/// source is explicit and a later step 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
+6 -6
View File
@@ -1270,8 +1270,8 @@ impl Node {
// teardown actions map to `remove_active_peer` / `note_link_dead`,
// in that order — the same order next ran them inline). The transient
// is never inserted into `peer_machines`; the persistent
// `established()` machine is created inside `promote_connection`
// (M3), so there is no double-insert. The relocated promote body,
// `established()` machine is created inside `promote_connection`,
// so there is no double-insert. The relocated promote body,
// and the `RestartThenPromote` teardown ordering equivalence, live in
// the executor's `PromoteToActive` / `InvalidateSendState` /
// `ReportLost` arms.
@@ -1401,11 +1401,11 @@ impl Node {
};
let loser_link_id = old_peer.link_id();
// Finding A: the replaced (losing) peer was established and so
// The replaced (losing) peer was established and so
// carried a machine keyed by its OWN link_id (loser_link_id);
// drop it so no machine orphans when its ActivePeer is removed.
// The winning connection's machine is inserted below keyed by
// the winner link_id. NEUTRAL — nothing reads peer_machines yet.
// the winner link_id. NEUTRAL: nothing reads peer_machines yet.
self.peer_machines.remove(&loser_link_id);
// Clean up old peer's index from peers_by_index
@@ -1465,7 +1465,7 @@ impl Node {
);
self.peers.insert(peer_node_addr, new_peer);
// Finding A: populate the inert per-peer machine so every
// Populate the inert per-peer machine so every
// established peer has exactly one machine, keyed by its
// link_id (the winner link here). Nothing drives it yet.
self.peer_machines.insert(
@@ -1596,7 +1596,7 @@ impl Node {
}
self.peers.insert(peer_node_addr, new_peer);
// Finding A: populate the inert per-peer machine so every
// Populate the inert per-peer machine so every
// established peer has exactly one machine, keyed by its link_id.
// Nothing drives it yet.
self.peer_machines.insert(
src/node/handlers/mmp.rs
+4 -7
View File
@@ -540,9 +540,9 @@ impl Node {
/// `note_link_dead`) reproduce the pre-refactor inline reap body exactly, in
/// that order.
///
/// 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
+11 -11
View File
@@ -54,7 +54,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 })
@@ -83,14 +83,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) {
@@ -100,7 +100,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),
@@ -120,7 +120,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) {
@@ -137,7 +137,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"
);
}
}
@@ -167,7 +167,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) {
@@ -212,7 +212,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
@@ -243,7 +243,7 @@ impl Node {
/// applies the thresholds without reading a clock (see [`PeerSnapshot`]).
///
/// Lives here, beside the drain/dampening constants and the FSP analog, so
/// the forward-merge onto `next` reconciles rekey timing in one place.
/// rekey timing is reconciled in one place.
pub(in crate::node) fn rekey_peer_snapshots(&self) -> Vec<PeerSnapshot> {
self.peers
.iter()
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
@@ -646,7 +646,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 /
@@ -1038,7 +1038,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
@@ -1245,7 +1245,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
@@ -1314,8 +1314,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.
@@ -1713,7 +1713,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");
@@ -2056,9 +2056,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
@@ -2075,7 +2075,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.
@@ -2093,13 +2093,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
@@ -2350,7 +2350,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
@@ -2359,8 +2359,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
@@ -2454,7 +2454,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
@@ -2586,7 +2586,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>,
@@ -2621,7 +2621,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`]).
@@ -2656,7 +2656,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
+10 -10
View File
@@ -184,20 +184,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,
@@ -209,13 +209,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)
@@ -369,14 +369,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 / M2) ===
// === 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 M2 — the executor (`dataplane/peer_actions.rs`) and advance helper exist
/// initially: the executor (`dataplane/peer_actions.rs`) and advance helper exist
/// but no live handler path drives them yet and nothing inserts into this map;
/// the inbound establish cutover lands in a later step.
/// the inbound establish cutover lands in a later commit.
#[allow(dead_code)]
peer_machines: HashMap<LinkId, PeerMachine>,
@@ -2393,7 +2393,7 @@ impl Node {
/// the production teardown path is [`remove_active_peer`](Self::remove_active_peer),
/// which drops the peer's `peer_machines` entry. This helper deliberately
/// does NOT touch `peer_machines`: it never runs against an established peer
/// in production, so it cannot orphan a machine (Finding A). If it is ever
/// in production, so it cannot orphan a machine. If it is ever
/// wired to remove established peers, mirror the `remove_active_peer` cleanup
/// (`self.peer_machines.remove(&peer.link_id())`) here.
pub fn remove_peer(&mut self, node_addr: &NodeAddr) -> Option<ActivePeer> {
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()
@@ -792,11 +787,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,
@@ -857,7 +852,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,
@@ -1060,7 +1055,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;
@@ -1187,7 +1182,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
@@ -1268,7 +1268,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
@@ -32,7 +32,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
+25 -24
View File
@@ -1,10 +1,10 @@
//! Per-peer FMP control FSM (sans-IO reducer) — XX re-derivation.
//!
//! The unified per-peer lifecycle state machine, ported onto next's XX cores
//! from the IK-lineage template. This is the **M1** increment: the FSM types,
//! from the IK-lineage template. 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 action executor and the msg3/rekey/reap wiring land in later steps.
//! yet; the action executor and the msg3/rekey/reap wiring land in later commits.
//!
//! ## Shape
//!
@@ -46,7 +46,7 @@
//! [`RekeyRespondTrigger`](PeerAction::RekeyRespondTrigger)) carrying indices +
//! the tie-break flag, and the executor performs the session/registry surgery
//! shell-side, matching next's inline `handle_msg3` verbatim. These trigger
//! variants are **provisional** (the wiring step may refine them) and unwired.
//! variants are **provisional** (the wiring may refine them) and unwired.
//!
//! Likewise the outbound completion is NOT routed through the machine: next has
//! no `establish_outbound` core — `handle_msg2` learns the identity from
@@ -55,19 +55,20 @@
//! and the resend/timeout timers; the shell drives promote and feeds the outcome
//! back via [`PromotionResolved`](PeerEvent::PromotionResolved).
//!
//! ## M1 realizability notes
//! ## Realizability notes
//!
//! - `SendHandshake`/`SendRekey`/`SendLinkMessage` carry **opaque bytes**
//! (`Vec<u8>`); the driver applies outer wire framing / encryption. A fresh
//! outbound msg1 and a fresh inbound msg2 have no bytes the control machine can
//! build (the Noise step / `build_msg2` are shell-side), so they are emitted
//! with an empty payload and threaded at wiring time. Not exercised by the M1
//! with an empty payload and threaded at wiring time. Not exercised by the
//! tests (which assert on action *kinds* / index-plane facts).
//! - `PeerSnapshot::rekey_msg3_pending` is sourced from a control field defaulting
//! `false`; the real wiring to `peer.rekey_msg3_payload().is_some()` and the
//! [`RekeyMsg3Resend`](TimerKind::RekeyMsg3Resend) driver land at the rekey step.
//! [`RekeyMsg3Resend`](TimerKind::RekeyMsg3Resend) driver land when the rekey
//! path is wired.
//! - `PeerSnapshot::counter` (the Noise send counter) is a send-state fact the
//! control machine cannot see; passed as `0`. Irrelevant to every M1 test.
//! control machine cannot see; passed as `0`. Irrelevant to every test.
#![allow(dead_code)]
@@ -83,10 +84,10 @@ use crate::{NodeAddr, PeerIdentity};
// ============================================================================
// Timing placeholders
//
// M1 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
// 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.
// ============================================================================
@@ -456,13 +457,13 @@ impl PeerMachine {
}
/// New machine for an ALREADY-established peer: the post-handshake state a
/// promoted peer occupies before any rekey. M3 inserts one of these into
/// `Node.peer_machines` at each `promote_connection` establishment site so
/// every established peer has exactly one machine keyed by its `LinkId`
/// (Finding A). The machine is **inert** — nothing drives it yet — and is
/// promoted peer occupies before any rekey. The driver inserts one of these
/// into `Node.peer_machines` at each `promote_connection` establishment site
/// so every established peer has exactly one machine keyed by its `LinkId`.
/// The machine is **inert** — nothing drives it yet — and is
/// parked at [`PeerState::Established`] so a later reap sees
/// [`is_established_context`](Self::is_established_context) true and a later
/// rekey step finds it. `our_index` is the peer's msg1-allocated session
/// rekey finds it. `our_index` is the peer's msg1-allocated session
/// index; `remote_epoch` is the crystallized peer's startup epoch.
pub(crate) fn established(
link: LinkId,
@@ -593,7 +594,7 @@ impl PeerMachine {
}
self.conn.set_transport_id(transport_id);
// Connection-oriented transports open the transport first; connectionless
// ones send msg1 immediately. M1 models the connection-oriented arm.
// ones send msg1 immediately. This models the connection-oriented arm.
self.state = PeerState::Connecting { link: self.link };
vec![PeerAction::OpenTransport {
transport_id,
@@ -624,8 +625,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; M1 emits an empty payload (see
/// module note). This path is not exercised by the M1 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 {
@@ -951,7 +952,7 @@ impl PeerMachine {
}
ConnAction::InitiateRekey { peer } => {
// Fresh outbound rekey: the Noise leaf + index allocation are
// shell-side (empty payload in M1), arm the resend timer.
// 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());
@@ -1187,7 +1188,7 @@ 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. M1 is unwired; the deadline
// re-enters the Timeout{kind} handlers. This module is unwired; the deadline
// bookkeeping is threaded from the driver at wiring time, so Tick is a
// no-op here.
let _ = now;
@@ -1242,8 +1243,8 @@ impl PeerMachine {
/// Build this peer's rekey snapshot from control-tier state. `counter` is a
/// send-state fact; passed as 0 here (see module note). `rekey_msg3_pending`
/// is sourced from the control field (default `false`; real wiring at the
/// rekey step).
/// is sourced from the control field (default `false`; real wiring when the
/// rekey path is wired).
fn peer_snapshot(&self, addr: NodeAddr, now: u64) -> PeerSnapshot {
let phase = match self.state {
PeerState::Maintaining {
@@ -1339,7 +1340,7 @@ mod tests {
}
}
// ---- Test 0: established constructor (Finding A populate) --------------
// ---- Test 0: established constructor --------------
#[test]
fn established_constructor_yields_established_context() {
let id = peer_identity();