diff --git a/src/bin/fips.rs b/src/bin/fips.rs index 493da2a..cc33687 100644 --- a/src/bin/fips.rs +++ b/src/bin/fips.rs @@ -8,7 +8,7 @@ use fips::config::{IdentitySource, resolve_identity}; use fips::version; use fips::{Config, Node}; use std::path::PathBuf; -use tracing::{debug, error, info, warn}; +use tracing::{debug, error, info}; use tracing_subscriber::{EnvFilter, fmt}; /// FIPS mesh network daemon @@ -157,26 +157,23 @@ async fn run_daemon( info!("FIPS running"); - // Run the RX event loop until shutdown signal. - // stop() drops the packet channel, causing run_rx_loop to exit. - tokio::select! { - result = node.run_rx_loop() => { - match result { - Ok(()) => info!("RX loop exited"), - Err(e) => error!("RX loop error: {}", e), - } - } - _ = shutdown_signal => { - info!("Shutdown signal received"); - } + // Serve until the shutdown signal, then drain in place before returning. + // The rx loop observes the signal directly, so its channels are never + // destructively cancelled — they live in the loop's locals across serve and + // drain, and are dropped only on clean exit (after which teardown does not + // need them). On the signal the loop broadcasts a shutdown Disconnect and + // waits (bounded by node.drain_timeout_secs) for peers to clear. + match node.run_rx_loop_with_shutdown(shutdown_signal).await { + Ok(()) => info!("RX loop exited"), + Err(e) => error!("RX loop error: {}", e), } info!("FIPS shutting down"); - // Stop the node (shuts down transports, TUN, I/O threads) - if let Err(e) = node.stop().await { - warn!("Error during shutdown: {}", e); - } + // Close the drain window (if the loop drained) and tear down. A drained + // loop tears down without re-broadcasting; a loop that exited some other + // way falls back to the immediate stop(). + node.finish_shutdown().await; info!("FIPS shutdown complete"); } diff --git a/src/config/node.rs b/src/config/node.rs index 8f8173b..198bf38 100644 --- a/src/config/node.rs +++ b/src/config/node.rs @@ -1030,6 +1030,19 @@ pub struct NodeConfig { #[serde(default = "NodeConfig::default_link_dead_timeout_secs")] pub link_dead_timeout_secs: u64, + /// Graceful-shutdown drain deadline in seconds (`node.drain_timeout_secs`). + /// The bounded `Draining` phase broadcasts a shutdown `Disconnect` and then + /// waits up to this long for peers to clear before tearing down, early- + /// exiting as soon as all peers are gone. `None` selects the 2-second + /// default (see [`NodeConfig::drain_timeout`]). + /// + /// Kept `Option` deliberately: `NodeConfig` has no `deny_unknown_fields`, so + /// a naive non-`Option` add with a `default` fn would silently rewrite the + /// value into deployed configs on the next serialize. The `Option` + + /// `skip_serializing_if` keeps absent configs absent. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub drain_timeout_secs: Option, + /// Resource limits (`node.limits.*`). #[serde(default)] pub limits: LimitsConfig, @@ -1111,6 +1124,7 @@ impl Default for NodeConfig { base_rtt_ms: 100, heartbeat_interval_secs: 10, link_dead_timeout_secs: 30, + drain_timeout_secs: None, limits: LimitsConfig::default(), rate_limit: RateLimitConfig::default(), retry: RetryConfig::default(), @@ -1161,6 +1175,14 @@ impl NodeConfig { fn default_link_dead_timeout_secs() -> u64 { 30 } + + /// Graceful-shutdown drain deadline as a `Duration`. + /// + /// Returns the configured `drain_timeout_secs`, or the 2-second default + /// when unset. Used by the daemon's bounded `Draining` phase. + pub fn drain_timeout(&self) -> std::time::Duration { + std::time::Duration::from_secs(self.drain_timeout_secs.unwrap_or(2)) + } } #[cfg(test)] @@ -1197,6 +1219,28 @@ owd_window_size: 48 assert_eq!(config.owd_window_size, DEFAULT_OWD_WINDOW_SIZE); } + #[test] + fn test_drain_timeout_default_and_override() { + // Unset → the 2-second default. + let c = NodeConfig::default(); + assert_eq!(c.drain_timeout_secs, None); + assert_eq!(c.drain_timeout(), std::time::Duration::from_secs(2)); + + // Explicit override is honored. + let c2 = NodeConfig { + drain_timeout_secs: Some(10), + ..NodeConfig::default() + }; + assert_eq!(c2.drain_timeout(), std::time::Duration::from_secs(10)); + + // A zero override is a valid (immediate) drain, not the default. + let c3 = NodeConfig { + drain_timeout_secs: Some(0), + ..NodeConfig::default() + }; + assert_eq!(c3.drain_timeout(), std::time::Duration::from_secs(0)); + } + #[test] fn test_ecn_config_defaults() { let c = EcnConfig::default(); diff --git a/src/node/dataplane/rx_loop.rs b/src/node/dataplane/rx_loop.rs index e056110..6ab5f58 100644 --- a/src/node/dataplane/rx_loop.rs +++ b/src/node/dataplane/rx_loop.rs @@ -42,6 +42,36 @@ impl Node { /// This method takes ownership of the packet_rx channel and runs /// until the channel is closed (typically when stop() is called). pub async fn run_rx_loop(&mut self) -> Result<(), NodeError> { + // No shutdown observer → today's infinite loop, byte-identical. All + // existing callers/tests use this; `pending()` never fires, so the + // shutdown/deadline arms below stay permanently disabled. + self.run_rx_loop_with_shutdown(std::future::pending()).await + } + + /// The rx event loop, which serves until `shutdown` fires and then drains + /// **in place** before returning. + /// + /// The channel receivers are moved into this frame's locals and live across + /// both serve and drain, so — unlike a `select!`-cancelled loop — they are + /// never destructively dropped mid-flight; they are released only on clean + /// exit, after which teardown does not need them. + /// + /// - While serving (`drain_deadline == None`) the loop is behaviorally + /// identical to before: the shutdown arm, the deadline arm, and the + /// peers-empty early-exit are all guarded off, so the hot per-packet path + /// and the `biased` order of the real arms are unchanged. + /// - When `shutdown` fires, the loop calls [`Node::enter_drain`] once + /// (broadcast Disconnect, gate the reconciler off) and arms the bounded + /// deadline, then keeps servicing inbound/tick/peer-removal until all + /// peers clear or the deadline elapses, then returns. The caller + /// ([`Node::finish_shutdown`]) closes the window and tears down. + pub async fn run_rx_loop_with_shutdown( + &mut self, + shutdown: impl std::future::Future, + ) -> Result<(), NodeError> { + tokio::pin!(shutdown); + // `None` = serving; `Some(deadline)` = draining (bounded window). + let mut drain_deadline: Option = None; let mut packet_rx = self.packet_rx.take().ok_or(NodeError::NotStarted)?; // Take the TUN outbound receiver, or create a dummy channel that never @@ -122,6 +152,13 @@ impl Node { crate::perf_profile::maybe_spawn_reporter(); loop { + // Bounded drain mode: break as soon as all peers have cleared. In + // normal mode (`None`) this short-circuits before touching + // `self.peers`, so the loop is byte-identical. + if drain_deadline.is_some() && self.peers.is_empty() { + info!("Drain complete: all peers cleared, ending drain loop"); + break; + } tokio::select! { biased; // Decrypt-worker fallback drains FIRST. Under sustained @@ -285,6 +322,27 @@ impl Node { #[cfg(any(target_os = "linux", target_os = "macos"))] self.activate_connected_udp_sessions().await; } + // Shutdown signal → enter the bounded drain in place, ONCE. + // Gated on `is_none()` so it only fires while serving; after + // entering drain the arm is disabled (the completed signal is + // never polled again) and the deadline arm below bounds the + // window. Placed after the real arms so their `biased` priority + // is unchanged, and inert while serving with `pending()`. + _ = &mut shutdown, if drain_deadline.is_none() => { + self.enter_drain().await; + drain_deadline = + Some(tokio::time::Instant::now() + self.config().node.drain_timeout()); + } + // Bounded drain deadline (drain mode only). Placed LAST so the + // `biased` priority of the normal arms is unchanged, and gated + // on `is_some()` so in normal mode the branch is disabled — the + // future is created but never polled and never fires. + _ = tokio::time::sleep_until( + drain_deadline.unwrap_or_else(tokio::time::Instant::now) + ), if drain_deadline.is_some() => { + info!("Drain deadline elapsed, ending drain loop"); + break; + } } } diff --git a/src/node/lifecycle/mod.rs b/src/node/lifecycle/mod.rs index 2053dbc..c0aa86f 100644 --- a/src/node/lifecycle/mod.rs +++ b/src/node/lifecycle/mod.rs @@ -1576,32 +1576,39 @@ impl Node { // Reconstruct the supervised up-set from observed runtime presence and // let the FSM author the teardown order (dns → nostr → mdns → - // transports (ascending id) → tun). Worker pools are deliberately - // absent from the up-set: today's stop() never tears them down. - let mut up: Vec = Vec::new(); - if self.supervisor.dns_task.is_some() { - up.push(Child::Dns); - } - if self.supervisor.nostr_rendezvous.engine().is_some() { - up.push(Child::Nostr); - } - if self.supervisor.lan_rendezvous.is_some() { - up.push(Child::Mdns); - } - for id in self.transports.keys() { - up.push(Child::Transport(*id)); - } - if self.tun_name.is_some() { - up.push(Child::Tun); - } + // transports (ascending id) → tun). + let up = self.reconstruct_supervised_up(); self.supervisor.fsm = SupervisorFsm::running_with(up); let actions = self.supervisor.fsm.step(Event::Stop); - // Execute each StopChild as today's verbatim block. Two driver seams - // are woven in at their current positions: the shutdown-disconnect - // fan-out (after Dns, before Nostr), and dropping the packet channels - // (after all transports, before TUN). - let mut disconnect_done = false; + // Execute the teardown plan. `broadcast_disconnect = true`: the + // immediate-stop path owns the single shutdown-Disconnect fan-out, + // which the helper emits at the Dns→rest seam. + self.execute_teardown(actions, true).await; + + self.supervisor.state = NodeState::Stopped; + info!(state = %self.supervisor.state, "Node stopped"); + Ok(()) + } + + /// Execute an ordered `StopChild` teardown plan authored by the supervisor + /// FSM, reporting each `ChildStopped` back so the machine reaches `Stopped`. + /// + /// This is the teardown body factored out of [`Self::stop`] so the drain + /// path can reuse the exact same per-child teardown and channel-drop + /// ordering. Two driver seams are woven in at their current positions: + /// + /// - **Seam (a), the shutdown-Disconnect fan-out** (after any Dns teardown, + /// before everything else) is gated on `broadcast_disconnect`. The + /// immediate `stop()` path passes `true` and emits it here; the drain path + /// passes `false` because it already broadcast once at drain entry and + /// must not re-broadcast. + /// - **Seam (b), dropping the packet channels** (after all transports, before + /// TUN) always runs. + async fn execute_teardown(&mut self, actions: Vec, broadcast_disconnect: bool) { + // `broadcast_disconnect = false` (drain path) marks the fan-out already + // done so neither the in-loop seam nor the trailing seam fires. + let mut disconnect_done = !broadcast_disconnect; let mut packet_taken = false; for action in actions { let Action::StopChild(child) = action else { @@ -1723,10 +1730,112 @@ impl Node { self.supervisor.packet_tx.take(); self.packet_rx.take(); } + } - self.supervisor.state = NodeState::Stopped; - info!(state = %self.supervisor.state, "Node stopped"); - Ok(()) + /// Reconstruct the supervised up-set from observed runtime presence, so the + /// FSM authors the teardown order regardless of how the node reached + /// `Running`. Worker pools are deliberately excluded: today's teardown never + /// stops them. Shared by [`Self::stop`] and [`Self::enter_drain`]. + fn reconstruct_supervised_up(&self) -> Vec { + let mut up: Vec = Vec::new(); + if self.supervisor.dns_task.is_some() { + up.push(Child::Dns); + } + if self.supervisor.nostr_rendezvous.engine().is_some() { + up.push(Child::Nostr); + } + if self.supervisor.lan_rendezvous.is_some() { + up.push(Child::Mdns); + } + for id in self.transports.keys() { + up.push(Child::Transport(*id)); + } + if self.tun_name.is_some() { + up.push(Child::Tun); + } + up + } + + /// Enter the bounded graceful drain **in place**, called once by + /// [`Self::run_rx_loop_with_shutdown`] when the shutdown signal fires. + /// + /// 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 + /// loop's deadline arm). Teardown is deferred to [`Self::finish_shutdown`]. + /// + /// Called from an rx-loop `select!` arm body: the channel receivers are + /// already moved into the loop's locals, so borrowing `self` here is sound. + pub(in crate::node) async fn enter_drain(&mut self) { + let up = self.reconstruct_supervised_up(); + self.supervisor.fsm = SupervisorFsm::running_with(up); + + let drain_timeout = self.config().node.drain_timeout(); + // Absolute driver-clock ms, carried into `Draining`/`SetTimer` for + // observability; the real bounded wait is the rx loop's deadline arm. + let deadline_ms = Self::now_ms().saturating_add(drain_timeout.as_millis() as u64); + + 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 + // 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. + self.supervisor.state = NodeState::Draining; + info!(state = %self.supervisor.state, "Node draining"); + + for action in actions { + match action { + Action::BroadcastDisconnect => { + self.send_disconnect_to_all_peers(DisconnectReason::Shutdown) + .await; + } + Action::SetTimer(_, _) => { + // The rx loop owns the bounded wait; the FSM's timer is + // carried for observability only. No-op here. + } + Action::SetPeeringDesired(_) | Action::SuspendReplenish => { + // §8 reconciler drain-gate. Documented no-op in this commit: + // the homeostatic reconciler that consumes these lands in + // Step 1b. Without it there is nothing to reconnect the peers + // the drain closes, so the gate is implicitly satisfied. + } + Action::SpawnChild(_) | Action::StopChild(_) => { + // Drain entry never emits child actions; ignore defensively. + } + } + } + + info!( + drain_timeout_secs = drain_timeout.as_secs(), + peers = self.peers.len(), + "Draining: broadcast shutdown Disconnect, waiting for peers to clear" + ); + } + + /// Finish shutdown after [`Self::run_rx_loop_with_shutdown`] returns. + /// + /// Branches on the supervisor's state: + /// - if the loop drained (FSM in `Draining`), close the window + /// (`DrainDeadlineElapsed`) and tear down **without re-broadcasting** — + /// the fan-out already went out at drain entry; + /// - otherwise the loop exited some other way (the packet channel closed + /// while still `Running` — the degenerate/error path), so fall back to the + /// immediate [`Self::stop`] (which broadcasts and tears down). + pub async fn finish_shutdown(&mut self) { + if self.supervisor.fsm.is_draining() { + self.supervisor.state = NodeState::Stopping; + info!(state = %self.supervisor.state, "Node stopping (drain complete)"); + let stop_actions = self.supervisor.fsm.step(Event::DrainDeadlineElapsed); + self.execute_teardown(stop_actions, false).await; + self.supervisor.state = NodeState::Stopped; + info!(state = %self.supervisor.state, "Node stopped"); + } else if let Err(e) = self.stop().await { + warn!(error = %e, "Error during shutdown"); + } } /// Send disconnect notifications to all active peers. diff --git a/src/node/lifecycle/supervisor.rs b/src/node/lifecycle/supervisor.rs index 00f96f2..b0edd22 100644 --- a/src/node/lifecycle/supervisor.rs +++ b/src/node/lifecycle/supervisor.rs @@ -22,15 +22,30 @@ //! - teardown runs in today's order and, faithfully, does **not** stop the //! encrypt/decrypt worker pools (they are spawned in `start()` but never torn //! down in `stop()`); -//! - the machine authors only the `SpawnChild`/`StopChild` *ordering*. The -//! `self.state` field writes stay verbatim in the driver at their current -//! positions, so the published `NodeState` transitions are byte-for-byte -//! unchanged. FSM-owned published state (`PublishState`) is introduced with -//! the `Draining`/`Degraded` additions, which need it. +//! - the machine authors only the `SpawnChild`/`StopChild` *ordering*, and the +//! driver keeps its `self.state` writes at their current positions. The +//! behavior-neutral relocation left the published `NodeState` transitions +//! byte-for-byte unchanged; the bounded-drain phase below adds exactly one new +//! published transition (`Draining`), written directly like the others. //! -//! The `Draining` phase and the `Running{Full|Degraded}` health split (design -//! doc §6/§9.1) land as the two subsequent, separately-flagged commits and -//! extend the [`SupState`], [`Event`], and [`Action`] enums below. +//! ## Scope: the bounded `Draining` phase (this commit) +//! +//! This commit adds the operator-visible bounded-drain additions and nothing +//! else: the [`SupState::Draining`] state, the [`Event::Drain`] / +//! [`Event::DrainDeadlineElapsed`] events, the drain [`Action`]s +//! ([`Action::BroadcastDisconnect`], [`Action::SetTimer`], +//! [`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` +//! path is untouched. `Draining` and `Stop` share a single teardown-plan author +//! (`begin_stopping`), so the teardown ordering is defined once. +//! +//! What is deferred is only the FSM-owned `PublishState` *action* (published +//! state authored by the machine rather than the driver): it lands with the +//! `Running{Full|Degraded}` health split (design doc §6/§9.1), which needs it +//! because a single direct `self.state` write cannot express the health fork. +//! The `Draining` published state itself is **not** deferred — it is here. use std::collections::HashSet; use std::sync::Arc; @@ -68,9 +83,9 @@ pub(crate) enum Child { /// An input to the supervisor. Results of executing [`Action`]s are fed back as /// `SubstrateUp` / `SubstrateFailed` / `ChildStopped`. /// -/// Only the events the behavior-neutral rewrite needs are present; `Tick`, -/// `ChildExited`, and `DrainDeadlineElapsed` (design doc §6) arrive with the -/// `Draining` / `Degraded` commits. +/// `Tick` and `ChildExited` (design doc §6) arrive with the `Degraded`/health +/// commit; the bounded-drain events (`Drain` / `DrainDeadlineElapsed`) are +/// present here. #[derive(Clone, Debug, PartialEq, Eq)] pub(crate) enum Event { /// Begin bring-up. `transports` are the ids the driver has already created @@ -106,8 +121,23 @@ pub(crate) enum Event { /// The child that failed to start. child: Child, }, - /// Begin teardown. Valid from `Running`. + /// Begin an immediate teardown (no drain). Valid from `Running`. This is + /// the path `node.stop()` uses; unchanged from the behavior-neutral rewrite. Stop, + /// Begin a bounded graceful drain. Valid from `Running`. Emits the drain + /// entry actions (broadcast Disconnect, arm the deadline timer, gate the + /// reconciler off) and moves to `Draining`; the driver then runs the + /// bounded drain window before feeding `DrainDeadlineElapsed`. + Drain { + /// Absolute drain deadline in driver-clock milliseconds, carried into + /// `Draining` and the `SetTimer` action for observability. The driver + /// owns the actual bounded wait. + deadline_ms: u64, + }, + /// The bounded drain window has closed — either the deadline elapsed or all + /// peers drained early. Valid from `Draining`; begins the (shared) teardown + /// plan, transitioning to `Stopping`. + DrainDeadlineElapsed, /// A child the driver was asked to stop has finished stopping. ChildStopped { /// The child that has been torn down. @@ -115,11 +145,33 @@ 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. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum Timer { + /// Fires when the bounded drain window closes. The driver feeds + /// [`Event::DrainDeadlineElapsed`] when it elapses (or earlier, when all + /// peers have drained). + 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 +/// 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 + /// homeostat off"). + Empty, +} + /// An effect the driver must perform. The core never performs I/O itself. /// -/// `PublishState` (design doc §6) is intentionally absent from the -/// behavior-neutral rewrite: the driver keeps its verbatim `self.state` writes, -/// so no published-state action is needed until `Draining`/`Degraded`. +/// `PublishState` (design doc §6) is intentionally absent until the +/// `Running{Full|Degraded}` health commit: the driver keeps its verbatim +/// `self.state` writes, so no published-state action is needed yet. #[derive(Clone, Debug, PartialEq, Eq)] pub(crate) enum Action { /// Bring up this child (the driver performs the spawn / start I/O and @@ -128,6 +180,21 @@ pub(crate) enum Action { /// Tear down this child (the driver performs the stop / join I/O and /// reports `ChildStopped`). StopChild(Child), + /// Broadcast a shutdown `Disconnect` to all sendable peers. Emitted once, + /// at drain entry; the drain teardown does not re-broadcast. + BroadcastDisconnect, + /// Arm a driver timer at the given absolute driver-clock milliseconds. In + /// this commit only `DrainDeadline` exists; the driver notes the deadline + /// 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. + SetPeeringDesired(PeeringDesired), + /// Suspend peer replenishment (§8 drain gate). Documented **no-op in this + /// commit** for the same reason as `SetPeeringDesired`. + SuspendReplenish, } /// Internal supervisor state (design doc §6). Richer than the published @@ -145,6 +212,16 @@ pub(crate) enum SupState { }, /// All children resolved; node operational. Running, + /// Bounded graceful-drain window (design doc §6/§8). 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 + /// `Stopping`. + Draining { + /// Absolute drain deadline in driver-clock milliseconds (carried for + /// observability; the driver owns the actual wait). + deadline_ms: u64, + }, /// Tearing children down; `pending` is the set not yet stopped. Stopping { /// Children asked to stop that have not yet reported stopped. @@ -196,6 +273,13 @@ impl SupervisorFsm { &self.state } + /// Whether the machine is in the bounded-drain window. The driver uses this + /// after the rx loop returns to decide between the drain-teardown path and + /// the immediate-`stop()` fallback. + pub(crate) fn is_draining(&self) -> bool { + matches!(self.state, SupState::Draining { .. }) + } + /// Advance the machine by one event, returning the effects to perform. pub(crate) fn step(&mut self, event: Event) -> Vec { match event { @@ -219,6 +303,8 @@ impl SupervisorFsm { Event::SubstrateUp { child } => self.on_substrate_up(child), Event::SubstrateFailed { child } => self.on_substrate_failed(child), Event::Stop => self.on_stop(), + Event::Drain { deadline_ms } => self.on_drain(deadline_ms), + Event::DrainDeadlineElapsed => self.on_drain_deadline_elapsed(), Event::ChildStopped { child } => self.on_child_stopped(child), } } @@ -306,6 +392,43 @@ impl SupervisorFsm { if !matches!(self.state, SupState::Running) { return Vec::new(); } + self.begin_stopping() + } + + fn on_drain(&mut self, deadline_ms: u64) -> Vec { + // Only a graceful drain from a running node. Inert otherwise (matching + // `Stop`'s guard). + if !matches!(self.state, SupState::Running) { + return Vec::new(); + } + 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. + vec![ + Action::BroadcastDisconnect, + Action::SetTimer(Timer::DrainDeadline, deadline_ms), + Action::SetPeeringDesired(PeeringDesired::Empty), + Action::SuspendReplenish, + ] + } + + fn on_drain_deadline_elapsed(&mut self) -> Vec { + // The bounded drain window closed (deadline or all-peers-gone). Author + // the same teardown plan `Stop` produces. + if !matches!(self.state, SupState::Draining { .. }) { + return Vec::new(); + } + self.begin_stopping() + } + + /// Author the teardown plan over the current up-set, transition to + /// `Stopping`, and return the ordered `StopChild` actions. Shared by the + /// immediate `Stop` path ([`Self::on_stop`]) and the drain-window-close path + /// ([`Self::on_drain_deadline_elapsed`]) so the teardown ordering is defined + /// exactly once. + fn begin_stopping(&mut self) -> Vec { let order = self.teardown_order(); self.state = SupState::Stopping { pending: order.iter().copied().collect(), @@ -655,4 +778,101 @@ mod tests { ); assert_eq!(s.state(), &SupState::Running); } + + #[test] + fn drain_from_running_emits_entry_actions_and_enters_draining() { + let mut s = SupervisorFsm::running_with([ + Child::Dns, + Child::Nostr, + Child::Transport(tid(1)), + Child::Tun, + ]); + let actions = s.step(Event::Drain { deadline_ms: 5_000 }); + // Order matters: broadcast → arm timer → gate reconciler off. + assert_eq!( + actions, + vec![ + Action::BroadcastDisconnect, + Action::SetTimer(Timer::DrainDeadline, 5_000), + Action::SetPeeringDesired(PeeringDesired::Empty), + Action::SuspendReplenish, + ] + ); + assert_eq!(s.state(), &SupState::Draining { deadline_ms: 5_000 }); + } + + #[test] + fn drain_deadline_elapsed_yields_teardown_and_enters_stopping() { + let mut s = SupervisorFsm::running_with([ + Child::Dns, + Child::Nostr, + Child::Mdns, + Child::Transport(tid(2)), + Child::Transport(tid(1)), + Child::Tun, + ]); + s.step(Event::Drain { deadline_ms: 2_000 }); + let stops = s.step(Event::DrainDeadlineElapsed); + // Same ordering the immediate `Stop` path authors: dns → nostr → mdns → + // transports (ascending id) → tun. + assert_eq!( + stops, + vec![ + Action::StopChild(Child::Dns), + Action::StopChild(Child::Nostr), + Action::StopChild(Child::Mdns), + Action::StopChild(Child::Transport(tid(1))), + Action::StopChild(Child::Transport(tid(2))), + Action::StopChild(Child::Tun), + ] + ); + assert!(matches!(s.state(), SupState::Stopping { .. })); + } + + #[test] + fn drain_teardown_matches_immediate_stop_teardown() { + // The drain path and the immediate-stop path must produce the identical + // StopChild plan over the same up-set (single teardown author). + let up = [ + Child::Dns, + Child::Nostr, + Child::Mdns, + Child::Transport(tid(1)), + Child::Transport(tid(3)), + Child::Tun, + ]; + let mut immediate = SupervisorFsm::running_with(up); + let stop_plan = immediate.step(Event::Stop); + + let mut drained = SupervisorFsm::running_with(up); + drained.step(Event::Drain { deadline_ms: 1_000 }); + let drain_plan = drained.step(Event::DrainDeadlineElapsed); + + assert_eq!(stop_plan, drain_plan); + } + + #[test] + fn drain_is_inert_from_non_running() { + // From `Created`. + let mut s = SupervisorFsm::new(); + assert_eq!(s.step(Event::Drain { deadline_ms: 1_000 }), vec![]); + assert_eq!(s.state(), &SupState::Created); + + // From `Stopping` (seed a drain, close its window, then try to drain + // again — inert). + let mut s2 = SupervisorFsm::running_with([Child::Transport(tid(1))]); + s2.step(Event::Drain { deadline_ms: 1_000 }); + s2.step(Event::DrainDeadlineElapsed); + assert!(matches!(s2.state(), SupState::Stopping { .. })); + assert_eq!(s2.step(Event::Drain { deadline_ms: 1_000 }), vec![]); + assert!(matches!(s2.state(), SupState::Stopping { .. })); + } + + #[test] + fn drain_deadline_elapsed_is_inert_outside_draining() { + // Inert from `Running` (no drain in progress). + let mut s = SupervisorFsm::running_with([Child::Transport(tid(1))]); + assert_eq!(s.step(Event::DrainDeadlineElapsed), vec![]); + assert_eq!(s.state(), &SupState::Running); + } } diff --git a/src/node/mod.rs b/src/node/mod.rs index 9c779ee..94054ec 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -168,6 +168,12 @@ pub enum NodeState { Starting, /// Fully operational. Running, + /// Bounded graceful drain in progress (design doc §6): 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, /// Shutting down. Stopping, /// Stopped. @@ -197,6 +203,7 @@ impl fmt::Display for NodeState { NodeState::Created => "created", NodeState::Starting => "starting", NodeState::Running => "running", + NodeState::Draining => "draining", NodeState::Stopping => "stopping", NodeState::Stopped => "stopped", }; diff --git a/src/node/tests/unit.rs b/src/node/tests/unit.rs index f5aab85..e9775f2 100644 --- a/src/node/tests/unit.rs +++ b/src/node/tests/unit.rs @@ -165,6 +165,41 @@ async fn test_node_state_transitions() { assert_eq!(node.state(), NodeState::Stopped); } +#[tokio::test] +async fn test_drain_publishes_draining_state() { + let mut node = make_node(); + node.start().await.unwrap(); + assert_eq!(node.state(), NodeState::Running); + assert!(node.state().is_operational()); + + // Enter the bounded drain in place: publishes the operator-visible + // `Draining` state (not operational) without tearing down. + node.enter_drain().await; + assert_eq!(node.state(), NodeState::Draining); + assert!(!node.state().is_operational()); + // `Draining` is neither startable nor externally stoppable; the daemon + // drain finishes via the supervisor's `DrainDeadlineElapsed`, not `stop()`. + assert!(!node.state().can_start()); + assert!(!node.state().can_stop()); + + // Finishing shutdown from `Draining` tears down to `Stopped`. + node.finish_shutdown().await; + assert_eq!(node.state(), NodeState::Stopped); +} + +#[tokio::test] +async fn test_immediate_stop_never_publishes_draining() { + let mut node = make_node(); + node.start().await.unwrap(); + assert_eq!(node.state(), NodeState::Running); + + // The immediate stop() path (used by tests and the stop-now path) + // transitions Running → Stopping → Stopped and never enters `Draining`. + node.stop().await.unwrap(); + assert_eq!(node.state(), NodeState::Stopped); + assert_ne!(node.state(), NodeState::Draining); +} + #[tokio::test] async fn test_node_start_does_not_wait_for_nostr_relay_startup() { let mut config = Config::new();