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 d212301..ce652b4 100644 --- a/src/config/node.rs +++ b/src/config/node.rs @@ -1045,6 +1045,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, @@ -1127,6 +1140,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(), @@ -1177,6 +1191,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)] @@ -1213,6 +1235,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/encrypted.rs b/src/node/dataplane/encrypted.rs index b5e2826..e55a34c 100644 --- a/src/node/dataplane/encrypted.rs +++ b/src/node/dataplane/encrypted.rs @@ -166,7 +166,7 @@ impl Node { #[cfg(unix)] { let cache_key = (packet.transport_id, header.receiver_idx.as_u32()); - if let Some(workers) = self.decrypt_workers.as_ref().cloned() + if let Some(workers) = self.supervisor.decrypt_workers.as_ref().cloned() && self.decrypt_registered_sessions.contains(&cache_key) { let job = crate::node::decrypt_worker::DecryptJob { @@ -460,7 +460,7 @@ impl Node { /// black-hole the session. #[cfg(unix)] pub(in crate::node) fn register_decrypt_worker_session(&mut self, node_addr: &crate::NodeAddr) { - let Some(workers) = self.decrypt_workers.as_ref().cloned() else { + let Some(workers) = self.supervisor.decrypt_workers.as_ref().cloned() else { return; }; let (cache_key, state) = { @@ -501,7 +501,7 @@ impl Node { &mut self, cache_key: (crate::transport::TransportId, u32), ) { - if let Some(workers) = self.decrypt_workers.as_ref() { + if let Some(workers) = self.supervisor.decrypt_workers.as_ref() { workers.unregister_session(cache_key); } self.decrypt_registered_sessions.remove(&cache_key); diff --git a/src/node/dataplane/rx_loop.rs b/src/node/dataplane/rx_loop.rs index f03bc3c..0582b18 100644 --- a/src/node/dataplane/rx_loop.rs +++ b/src/node/dataplane/rx_loop.rs @@ -44,12 +44,42 @@ 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 // produces messages (when TUN is disabled). Holding the sender prevents // the channel from closing. - let (mut tun_outbound_rx, _tun_guard) = match self.tun_outbound_rx.take() { + let (mut tun_outbound_rx, _tun_guard) = match self.supervisor.tun_outbound_rx.take() { Some(rx) => (rx, None), None => { let (tx, rx) = tokio::sync::mpsc::channel(1); @@ -59,7 +89,7 @@ impl Node { // Take the DNS identity receiver, or create a dummy channel (when DNS // is disabled). Same pattern as TUN outbound. - let (mut dns_identity_rx, _dns_guard) = match self.dns_identity_rx.take() { + let (mut dns_identity_rx, _dns_guard) = match self.supervisor.dns_identity_rx.take() { Some(rx) => (rx, None), None => { let (tx, rx) = tokio::sync::mpsc::channel(1); @@ -124,6 +154,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 @@ -288,6 +325,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; + } } } @@ -323,9 +381,11 @@ impl Node { // discovery-layer cooldown to the long protocol-mismatch // window and emit a single WARN per fresh observation. if self + .supervisor .nostr_rendezvous .is_bootstrap_transport(&packet.transport_id) && let Some(npub) = self + .supervisor .nostr_rendezvous .bootstrap_transport_npub(&packet.transport_id) .cloned() diff --git a/src/node/handlers/session.rs b/src/node/handlers/session.rs index e388fbf..0bd5f20 100644 --- a/src/node/handlers/session.rs +++ b/src/node/handlers/session.rs @@ -365,7 +365,7 @@ impl Node { mark_ipv6_ecn_ce(&mut packet); self.metrics().congestion.ce_received.inc(); } - if let Some(tun_tx) = &self.tun_tx { + if let Some(tun_tx) = &self.supervisor.tun_tx { if let Err(e) = tun_tx.send(packet) { debug!(error = %e, "Failed to deliver decompressed IPv6 packet to TUN"); } @@ -1916,7 +1916,7 @@ impl Node { send: PipelinedSend<'_>, ) -> Result { let dest_addr = send.dest_addr; - let Some(workers) = self.encrypt_workers.as_ref().cloned() else { + let Some(workers) = self.supervisor.encrypt_workers.as_ref().cloned() else { return Ok(false); }; @@ -2528,7 +2528,7 @@ impl Node { let our_ipv6 = FipsAddress::from_node_addr(self.node_addr()).to_ipv6(); if let Some(response) = build_dest_unreachable(original_packet, DestUnreachableCode::NoRoute, our_ipv6) - && let Some(tun_tx) = &self.tun_tx + && let Some(tun_tx) = &self.supervisor.tun_tx { let _ = tun_tx.send(response); } @@ -2565,7 +2565,7 @@ impl Node { // SAFETY: slice is exactly 16 bytes; length validated above (>= 40) let dest_addr = Ipv6Addr::from(<[u8; 16]>::try_from(&original_packet[24..40]).unwrap()); if let Some(response) = build_packet_too_big(original_packet, mtu, dest_addr) - && let Some(tun_tx) = &self.tun_tx + && let Some(tun_tx) = &self.supervisor.tun_tx { debug!( original_src = %src_addr, diff --git a/src/node/lifecycle.rs b/src/node/lifecycle/mod.rs similarity index 71% rename from src/node/lifecycle.rs rename to src/node/lifecycle/mod.rs index b6c4eb8..5645791 100644 --- a/src/node/lifecycle.rs +++ b/src/node/lifecycle/mod.rs @@ -1,6 +1,10 @@ //! Node lifecycle management: start, stop, and peer connection initiation. +pub(crate) mod supervisor; + use super::{Node, NodeError, NodeState}; +use supervisor::{Action, Child, Event, SupervisorFsm}; + use crate::config::{ConnectPolicy, PeerAddress, PeerConfig}; use crate::node::acl::PeerAclContext; use crate::nostr::{BootstrapEvent, NostrRendezvous}; @@ -263,7 +267,7 @@ impl Node { // would loop on the same dead address until expiry. Force a // re-fetch so the next retry tick picks up fresh endpoints. if matches!(e, crate::node::NodeError::NoTransportForType(_)) - && let Some(bootstrap) = self.nostr_rendezvous.engine_arc() + && let Some(bootstrap) = self.supervisor.nostr_rendezvous.engine_arc() { let npub = peer_config.npub.clone(); tokio::spawn(async move { @@ -357,7 +361,7 @@ impl Node { .filter(|(id, handle)| { handle.transport_type().name == "udp" && handle.is_operational() - && !self.nostr_rendezvous.is_bootstrap_transport(id) + && !self.supervisor.nostr_rendezvous.is_bootstrap_transport(id) }) .filter_map(|(id, handle)| { let local_addr = handle.local_addr()?; @@ -734,7 +738,7 @@ impl Node { } pub(super) async fn poll_nostr_rendezvous(&mut self) { - let Some(bootstrap) = self.nostr_rendezvous.engine_arc() else { + let Some(bootstrap) = self.supervisor.nostr_rendezvous.engine_arc() else { return; }; @@ -959,7 +963,7 @@ impl Node { /// The handshake itself is the authentication — a spoofed mDNS advert /// with someone else's npub fails the XX exchange and is dropped. pub(super) async fn poll_lan_rendezvous(&mut self) { - let Some(runtime) = self.lan_rendezvous.clone() else { + let Some(runtime) = self.supervisor.lan_rendezvous.clone() else { return; }; let events = runtime.drain_events().await; @@ -1111,51 +1115,41 @@ impl Node { /// Initializes the TUN interface (if configured), spawns I/O threads, /// and transitions to the Running state. pub async fn start(&mut self) -> Result<(), NodeError> { - if !self.state.can_start() { + if !self.supervisor.state.can_start() { return Err(NodeError::AlreadyStarted); } - self.state = NodeState::Starting; + self.supervisor.state = NodeState::Starting; // Create packet channel for transport -> Node communication let packet_buffer_size = self.config().node.buffers.packet_channel; let (packet_tx, packet_rx) = packet_channel(packet_buffer_size); - self.packet_tx = Some(packet_tx.clone()); + self.supervisor.packet_tx = Some(packet_tx.clone()); self.packet_rx = Some(packet_rx); // Initialize transports first (before TUN, before Nostr discovery). + // Creation allocates each transport's id; the supervisor FSM authors + // the start order over those ids. let transport_handles = self.create_transports(&packet_tx).await; + let transport_ids: Vec = + transport_handles.iter().map(|h| h.transport_id()).collect(); + let mut pending_handles: HashMap<_, _> = transport_handles + .into_iter() + .map(|h| (h.transport_id(), h)) + .collect(); - for mut handle in transport_handles { - let transport_id = handle.transport_id(); - let transport_type = handle.transport_type().name; - let name = handle.name().map(|s| s.to_string()); + // Singleton child booleans, with today's exact enable conditions. + let nostr = self.config().node.rendezvous.nostr.enabled; + let mdns = self.config().node.rendezvous.lan.enabled; + let tun = self.config().tun.enabled; + let dns = self.config().dns.enabled; - match handle.start().await { - Ok(()) => { - self.transports.insert(transport_id, handle); - } - Err(e) => { - if let Some(ref n) = name { - warn!(transport_type, name = %n, error = %e, "Transport failed to start"); - } else { - warn!(transport_type, error = %e, "Transport failed to start"); - } - } - } - } - - if !self.transports.is_empty() { - info!(count = self.transports.len(), "Transports initialized"); - } - - // Spawn the off-task FMP-encrypt + UDP-send worker pool. - // Unix only — the worker issues sendmmsg(2) / sendmsg+UDP_GSO - // calls on raw fds via `AsRawFd`, a unix-only trait. Worker - // count defaults to num_cpus, overridable via FIPS_ENCRYPT_WORKERS. - // Hash-by-destination pins a TCP flow to one worker (preserves - // wire ordering); additional workers light up under multi-flow load. + // Worker-pool booleans + counts. Unix only — the workers issue + // sendmmsg(2) / sendmsg+UDP_GSO on raw fds via `AsRawFd`. Encrypt + // always spawns on unix; decrypt spawns iff FIPS_DECRYPT_WORKERS != 0. + // Counts are parsed up-front so the FSM can decide whether the decrypt + // child exists; the actual spawns run when the SpawnChild actions do. #[cfg(unix)] - { + let (encrypt_workers, decrypt_workers, encrypt_worker_count, decrypt_worker_count) = { let cpu_default = std::thread::available_parallelism() .map(|n| n.get()) .unwrap_or(1) @@ -1165,273 +1159,439 @@ impl Node { .and_then(|s| s.parse().ok()) .unwrap_or(cpu_default) .max(1); - self.encrypt_workers = Some(super::encrypt_worker::EncryptWorkerPool::spawn( - encrypt_worker_count, - )); - info!( - workers = encrypt_worker_count, - "Spawned FMP-encrypt worker pool" - ); - - // `FIPS_DECRYPT_WORKERS=0` disables the pool entirely and - // forces the in-line rx_loop decrypt path (useful as an A/B - // against the worker pipeline). Any non-zero value (env or - // default) spawns the shard-owned decrypt pool. let decrypt_worker_count: usize = std::env::var("FIPS_DECRYPT_WORKERS") .ok() .and_then(|s| s.parse().ok()) .unwrap_or(cpu_default); - if decrypt_worker_count == 0 { - info!("FIPS_DECRYPT_WORKERS=0 → in-line decrypt in rx_loop"); - } else { - self.decrypt_workers = Some(super::decrypt_worker::DecryptWorkerPool::spawn( - decrypt_worker_count, - )); - info!( - workers = decrypt_worker_count, - "Spawned FMP-decrypt worker pool" - ); - } - } - - if self.config().node.rendezvous.nostr.enabled { - match NostrRendezvous::start( - self.identity(), - self.config().node.rendezvous.nostr.clone(), + ( + true, + decrypt_worker_count != 0, + encrypt_worker_count, + decrypt_worker_count, ) - .await - { - Ok(runtime) => { - if let Err(err) = self.refresh_overlay_advert(&runtime).await { - warn!(error = %err, "Failed to publish initial Nostr overlay advert"); - } - self.nostr_rendezvous.set_engine(runtime); - self.nostr_rendezvous.set_started_at_ms(Self::now_ms()); - info!("Nostr overlay discovery enabled"); - } - Err(err) => { - warn!(error = %err, "Failed to start Nostr overlay discovery"); - } + }; + #[cfg(not(unix))] + let (encrypt_workers, decrypt_workers) = (false, false); + + // Ask the supervisor FSM for the canonical spawn order. + let actions = self.supervisor.fsm.step(Event::Start { + transports: transport_ids, + encrypt_workers, + decrypt_workers, + nostr, + mdns, + tun, + dns, + }); + + // 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 + // 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. + let mut start_outcome: Option = None; + for action in &actions { + if let Action::PublishState(ns) = action { + start_outcome = Some(*ns); } } - // mDNS / DNS-SD LAN discovery. Independent of Nostr — runs even - // when Nostr is disabled, since it gives us sub-second pairing - // on the same link without any relay or NAT-traversal roundtrip. - if self.config().node.rendezvous.lan.enabled { - // Advertise the port of a non-bootstrap operational UDP transport. - // Bootstrap transports must be excluded (they are not the node's - // listening data-plane socket), and a stable selector (lowest - // TransportId) is used so the advertised port is deterministic - // across restarts rather than dependent on HashMap iteration - // order. This mirrors find_udp_transport_for_remote_addr. - let advertised_udp_port = self - .transports - .iter() - .filter(|(id, h)| { - h.transport_type().name == "udp" - && h.is_operational() - && !self.nostr_rendezvous.is_bootstrap_transport(id) - }) - .filter_map(|(id, h)| h.local_addr().map(|addr| (*id, addr.port()))) - .min_by_key(|(id, _)| id.as_u32()) - .map(|(_, port)| port) - .unwrap_or(0); - let scope = self.lan_rendezvous_scope(); - match crate::mdns::LanRendezvous::start( - self.identity(), - scope, - advertised_udp_port, - self.config().node.rendezvous.lan.clone(), - ) - .await - { - Ok(runtime) => { - self.lan_rendezvous = Some(runtime); - info!("LAN mDNS discovery enabled"); - } - Err(err) => { - debug!(error = %err, "LAN mDNS discovery not started"); + // Execute each SpawnChild in order, reporting the outcome back so the + // FSM's up-set tracks what actually came up. Optional failures are + // warn/debug-and-continue (today's behavior); start still reaches + // Running. Two driver seams are woven in at their current positions: + // the post-transport-loop "Transports initialized" info!, and the + // peer-connect that today sits after mDNS and before TUN. + let mut transports_info_emitted = false; + let mut peer_connect_done = false; + for action in actions { + let Action::SpawnChild(child) = action else { + continue; + }; + + // Post-transport-loop seam: once, after all transport spawns and + // before the first non-transport child. + if !transports_info_emitted && !matches!(child, Child::Transport(_)) { + if !self.transports.is_empty() { + info!(count = self.transports.len(), "Transports initialized"); } + transports_info_emitted = true; } - } - // Connect to static peers before TUN is active - // This allows handshake messages to be sent before we start accepting packets - self.initiate_peer_connections().await; - - // Initialize TUN interface last, after transports and peers are ready - if self.config().tun.enabled { - let address = *self.identity().address(); - match TunDevice::create(&self.config().tun, address).await { - Ok(device) => { - let mtu = device.mtu(); - let name = device.name().to_string(); - let our_addr = *device.address(); - - info!("TUN device active:"); - info!(" name: {}", name); - info!(" address: {}", device.address()); - info!(" mtu: {}", mtu); - - // Calculate max MSS for TCP clamping - let effective_mtu = self.effective_ipv6_mtu(); - let max_mss = effective_mtu.saturating_sub(40).saturating_sub(20); // IPv6 + TCP headers - - info!("effective MTU: {} bytes", effective_mtu); - debug!(" max TCP MSS: {} bytes", max_mss); - - // On macOS, create a shutdown pipe. Writing to it unblocks the - // reader thread's select() loop without closing the TUN fd - // (which would cause a double-close when TunDevice drops). - #[cfg(target_os = "macos")] - let (shutdown_read_fd, shutdown_write_fd) = { - let mut fds = [0i32; 2]; - if unsafe { libc::pipe(fds.as_mut_ptr()) } < 0 { - return Err(NodeError::Tun(crate::upper::tun::TunError::Configure( - "failed to create shutdown pipe".into(), - ))); - } - (fds[0], fds[1]) - }; - - // Create writer (dups the fd for independent write access). - // Pass path_mtu_lookup so inbound SYN-ACK clamp can read - // per-destination path MTU learned via discovery. - let (writer, tun_tx) = - device.create_writer(max_mss, self.path_mtu_lookup.clone())?; - - // Spawn writer thread - let writer_handle = thread::spawn(move || { - writer.run(); - }); - - // Clone tun_tx for the reader - let reader_tun_tx = tun_tx.clone(); - - // Create outbound channel for TUN reader → Node - let tun_channel_size = self.config().node.buffers.tun_channel; - let (outbound_tx, outbound_rx) = tokio::sync::mpsc::channel(tun_channel_size); - - // Spawn reader thread - let transport_mtu = self.transport_mtu(); - let path_mtu_lookup = self.path_mtu_lookup.clone(); - #[cfg(target_os = "macos")] - let reader_handle = thread::spawn(move || { - run_tun_reader( - device, - mtu, - our_addr, - reader_tun_tx, - outbound_tx, - transport_mtu, - path_mtu_lookup, - shutdown_read_fd, - ); - }); - #[cfg(not(target_os = "macos"))] - let reader_handle = thread::spawn(move || { - run_tun_reader( - device, - mtu, - our_addr, - reader_tun_tx, - outbound_tx, - transport_mtu, - path_mtu_lookup, - ); - }); - - self.tun_state = TunState::Active; - self.tun_name = Some(name); - self.tun_tx = Some(tun_tx); - self.tun_outbound_rx = Some(outbound_rx); - self.tun_reader_handle = Some(reader_handle); - self.tun_writer_handle = Some(writer_handle); - #[cfg(target_os = "macos")] - { - self.tun_shutdown_fd = Some(shutdown_write_fd); - } - } - Err(e) => { - self.tun_state = TunState::Failed; - warn!(error = %e, "Failed to initialize TUN, continuing without it"); - } + // Peer-connect seam: once, immediately before the first Tun-or-Dns + // child. Connect to static peers before TUN is active so handshake + // messages can be sent before we start accepting packets. + if !peer_connect_done && matches!(child, Child::Tun | Child::Dns) { + self.initiate_peer_connections().await; + peer_connect_done = true; } - } - // Initialize DNS responder (independent of TUN). - // - // Default bind_addr is "::1" (IPv6 loopback). The shipped - // fips-dns-setup configures systemd-resolved via a global - // /etc/systemd/resolved.conf.d/fips.conf drop-in pointing at - // [::1]:5354, which sidesteps a Linux IPV6_PKTINFO behaviour - // where self-destined traffic to fips0's address is attributed - // to fips0 in PKTINFO and gets silently dropped by the - // mesh-interface filter in src/upper/dns.rs. - // - // For mesh-reachable resolution (rare), set bind_addr: "::" - // in fips.yaml. The mesh-interface filter remains active to - // prevent hosts-file alias enumeration in that mode. - // `IPV6_V6ONLY=0` is set explicitly so IPv4 clients on - // 127.0.0.1 still reach us regardless of kernel sysctl - // defaults — but only when bind is on a wildcard / IPv6 path. - if self.config().dns.enabled { - let addr_str = self.config().dns.bind_addr(); - match addr_str.parse::() { - Ok(ip) => { - let bind = std::net::SocketAddr::new(ip, self.config().dns.port()); - match Self::bind_dns_socket(bind) { - Ok(socket) => { - let dns_channel_size = self.config().node.buffers.dns_channel; - let (identity_tx, identity_rx) = - tokio::sync::mpsc::channel(dns_channel_size); - let dns_ttl = self.config().dns.ttl(); - let base_hosts = crate::upper::hosts::HostMap::from_peer_configs( - self.config().peers(), - ); - let hosts_path = - std::path::PathBuf::from(crate::upper::hosts::DEFAULT_HOSTS_PATH); - let reloader = - crate::upper::hosts::HostMapReloader::new(base_hosts, hosts_path); - // Resolve the TUN ifindex so the responder can - // drop queries arriving on the mesh interface - // (fips0). Without this, the `::` bind exposes - // /etc/fips/hosts alias probing to any mesh peer. - // When TUN isn't enabled or the name can't be - // resolved, `None` disables the filter (there - // is no mesh surface to defend anyway). - let mesh_ifindex = Self::lookup_mesh_ifindex(self.config().tun.name()); - info!( - bind = %bind, - hosts = reloader.hosts().len(), - mesh_ifindex = ?mesh_ifindex, - "DNS responder started for .fips domain (auto-reload enabled)" - ); - let handle = tokio::spawn(crate::upper::dns::run_dns_responder( - socket, - identity_tx, - dns_ttl, - reloader, - mesh_ifindex, - )); - self.dns_identity_rx = Some(identity_rx); - self.dns_task = Some(handle); + let feedback = match child { + Child::Transport(id) => { + let mut handle = pending_handles + .remove(&id) + .expect("supervisor emitted SpawnChild for a created transport"); + let transport_type = handle.transport_type().name; + let name = handle.name().map(|s| s.to_string()); + + match handle.start().await { + Ok(()) => { + self.transports.insert(id, handle); + Event::SubstrateUp { child } } Err(e) => { - warn!(bind = %bind, error = %e, "Failed to start DNS responder"); + if let Some(ref n) = name { + warn!(transport_type, name = %n, error = %e, "Transport failed to start"); + } else { + warn!(transport_type, error = %e, "Transport failed to start"); + } + Event::SubstrateFailed { child } } } } - Err(e) => { - warn!(addr = %addr_str, error = %e, "Invalid dns.bind_addr; DNS responder not started"); + Child::EncryptWorkers => { + // Hash-by-destination pins a TCP flow to one worker + // (preserves wire ordering); additional workers light up + // under multi-flow load. Infallible → always up. + #[cfg(unix)] + { + self.supervisor.encrypt_workers = Some( + super::encrypt_worker::EncryptWorkerPool::spawn(encrypt_worker_count), + ); + info!( + workers = encrypt_worker_count, + "Spawned FMP-encrypt worker pool" + ); + + // `FIPS_DECRYPT_WORKERS=0` disables the pool entirely + // and forces the in-line rx_loop decrypt path. When 0 + // no DecryptWorkers child is emitted, so this info! + // sits here — exactly where the decrypt spawn would be + // in today's sequence (after the encrypt spawn+info, + // before nostr). + if decrypt_worker_count == 0 { + info!("FIPS_DECRYPT_WORKERS=0 → in-line decrypt in rx_loop"); + } + } + Event::SubstrateUp { child } + } + Child::DecryptWorkers => { + // Shard-owned decrypt pool. Infallible → always up. + #[cfg(unix)] + { + self.supervisor.decrypt_workers = Some( + super::decrypt_worker::DecryptWorkerPool::spawn(decrypt_worker_count), + ); + info!( + workers = decrypt_worker_count, + "Spawned FMP-decrypt worker pool" + ); + } + Event::SubstrateUp { child } + } + Child::Nostr => { + match NostrRendezvous::start( + self.identity(), + self.config().node.rendezvous.nostr.clone(), + ) + .await + { + Ok(runtime) => { + if let Err(err) = self.refresh_overlay_advert(&runtime).await { + warn!(error = %err, "Failed to publish initial Nostr overlay advert"); + } + self.supervisor.nostr_rendezvous.set_engine(runtime); + self.supervisor + .nostr_rendezvous + .set_started_at_ms(Self::now_ms()); + info!("Nostr overlay discovery enabled"); + Event::SubstrateUp { child } + } + Err(err) => { + warn!(error = %err, "Failed to start Nostr overlay discovery"); + Event::SubstrateFailed { child } + } + } + } + Child::Mdns => { + // Advertise the port of a non-bootstrap operational UDP + // transport. Bootstrap transports must be excluded (they + // are not the node's listening data-plane socket), and a + // stable selector (lowest TransportId) is used so the + // advertised port is deterministic across restarts rather + // than dependent on HashMap iteration order. This mirrors + // find_udp_transport_for_remote_addr. + let advertised_udp_port = self + .transports + .iter() + .filter(|(id, h)| { + h.transport_type().name == "udp" + && h.is_operational() + && !self.supervisor.nostr_rendezvous.is_bootstrap_transport(id) + }) + .filter_map(|(id, h)| h.local_addr().map(|addr| (*id, addr.port()))) + .min_by_key(|(id, _)| id.as_u32()) + .map(|(_, port)| port) + .unwrap_or(0); + let scope = self.lan_rendezvous_scope(); + match crate::mdns::LanRendezvous::start( + self.identity(), + scope, + advertised_udp_port, + self.config().node.rendezvous.lan.clone(), + ) + .await + { + Ok(runtime) => { + self.supervisor.lan_rendezvous = Some(runtime); + info!("LAN mDNS discovery enabled"); + Event::SubstrateUp { child } + } + Err(err) => { + debug!(error = %err, "LAN mDNS discovery not started"); + Event::SubstrateFailed { child } + } + } + } + Child::Tun => { + // Initialize TUN interface after transports and peers are + // ready. + let address = *self.identity().address(); + match TunDevice::create(&self.config().tun, address).await { + Ok(device) => { + let mtu = device.mtu(); + let name = device.name().to_string(); + let our_addr = *device.address(); + + info!("TUN device active:"); + info!(" name: {}", name); + info!(" address: {}", device.address()); + info!(" mtu: {}", mtu); + + // Calculate max MSS for TCP clamping + let effective_mtu = self.effective_ipv6_mtu(); + let max_mss = effective_mtu.saturating_sub(40).saturating_sub(20); // IPv6 + TCP headers + + info!("effective MTU: {} bytes", effective_mtu); + debug!(" max TCP MSS: {} bytes", max_mss); + + // On macOS, create a shutdown pipe. Writing to it unblocks the + // reader thread's select() loop without closing the TUN fd + // (which would cause a double-close when TunDevice drops). + #[cfg(target_os = "macos")] + let (shutdown_read_fd, shutdown_write_fd) = { + let mut fds = [0i32; 2]; + if unsafe { libc::pipe(fds.as_mut_ptr()) } < 0 { + return Err(NodeError::Tun( + crate::upper::tun::TunError::Configure( + "failed to create shutdown pipe".into(), + ), + )); + } + (fds[0], fds[1]) + }; + + // Create writer (dups the fd for independent write access). + // Pass path_mtu_lookup so inbound SYN-ACK clamp can read + // per-destination path MTU learned via discovery. + let (writer, tun_tx) = + device.create_writer(max_mss, self.path_mtu_lookup.clone())?; + + // Spawn writer thread + let writer_handle = thread::spawn(move || { + writer.run(); + }); + + // Clone tun_tx for the reader + let reader_tun_tx = tun_tx.clone(); + + // Create outbound channel for TUN reader → Node + let tun_channel_size = self.config().node.buffers.tun_channel; + let (outbound_tx, outbound_rx) = + tokio::sync::mpsc::channel(tun_channel_size); + + // Spawn reader thread + let transport_mtu = self.transport_mtu(); + let path_mtu_lookup = self.path_mtu_lookup.clone(); + #[cfg(target_os = "macos")] + let reader_handle = thread::spawn(move || { + run_tun_reader( + device, + mtu, + our_addr, + reader_tun_tx, + outbound_tx, + transport_mtu, + path_mtu_lookup, + shutdown_read_fd, + ); + }); + #[cfg(not(target_os = "macos"))] + let reader_handle = thread::spawn(move || { + run_tun_reader( + device, + mtu, + our_addr, + reader_tun_tx, + outbound_tx, + transport_mtu, + path_mtu_lookup, + ); + }); + + self.tun_state = TunState::Active; + self.tun_name = Some(name); + self.supervisor.tun_tx = Some(tun_tx); + self.supervisor.tun_outbound_rx = Some(outbound_rx); + self.supervisor.tun_reader_handle = Some(reader_handle); + self.supervisor.tun_writer_handle = Some(writer_handle); + #[cfg(target_os = "macos")] + { + self.supervisor.tun_shutdown_fd = Some(shutdown_write_fd); + } + Event::SubstrateUp { child } + } + Err(e) => { + self.tun_state = TunState::Failed; + warn!(error = %e, "Failed to initialize TUN, continuing without it"); + Event::SubstrateFailed { child } + } + } + } + Child::Dns => { + // Initialize DNS responder (independent of TUN). + // + // Default bind_addr is "::1" (IPv6 loopback). The shipped + // fips-dns-setup configures systemd-resolved via a global + // /etc/systemd/resolved.conf.d/fips.conf drop-in pointing at + // [::1]:5354, which sidesteps a Linux IPV6_PKTINFO behaviour + // where self-destined traffic to fips0's address is attributed + // to fips0 in PKTINFO and gets silently dropped by the + // mesh-interface filter in src/upper/dns.rs. + // + // For mesh-reachable resolution (rare), set bind_addr: "::" + // in fips.yaml. The mesh-interface filter remains active to + // prevent hosts-file alias enumeration in that mode. + // `IPV6_V6ONLY=0` is set explicitly so IPv4 clients on + // 127.0.0.1 still reach us regardless of kernel sysctl + // defaults — but only when bind is on a wildcard / IPv6 path. + let addr_str = self.config().dns.bind_addr(); + match addr_str.parse::() { + Ok(ip) => { + let bind = std::net::SocketAddr::new(ip, self.config().dns.port()); + match Self::bind_dns_socket(bind) { + Ok(socket) => { + let dns_channel_size = self.config().node.buffers.dns_channel; + let (identity_tx, identity_rx) = + tokio::sync::mpsc::channel(dns_channel_size); + let dns_ttl = self.config().dns.ttl(); + let base_hosts = + crate::upper::hosts::HostMap::from_peer_configs( + self.config().peers(), + ); + let hosts_path = std::path::PathBuf::from( + crate::upper::hosts::DEFAULT_HOSTS_PATH, + ); + let reloader = crate::upper::hosts::HostMapReloader::new( + base_hosts, hosts_path, + ); + // Resolve the TUN ifindex so the responder can + // drop queries arriving on the mesh interface + // (fips0). Without this, the `::` bind exposes + // /etc/fips/hosts alias probing to any mesh peer. + // When TUN isn't enabled or the name can't be + // resolved, `None` disables the filter (there + // is no mesh surface to defend anyway). + let mesh_ifindex = + Self::lookup_mesh_ifindex(self.config().tun.name()); + info!( + bind = %bind, + hosts = reloader.hosts().len(), + mesh_ifindex = ?mesh_ifindex, + "DNS responder started for .fips domain (auto-reload enabled)" + ); + let handle = + tokio::spawn(crate::upper::dns::run_dns_responder( + socket, + identity_tx, + dns_ttl, + reloader, + mesh_ifindex, + )); + self.supervisor.dns_identity_rx = Some(identity_rx); + self.supervisor.dns_task = Some(handle); + Event::SubstrateUp { child } + } + Err(e) => { + warn!(bind = %bind, error = %e, "Failed to start DNS responder"); + Event::SubstrateFailed { child } + } + } + } + Err(e) => { + warn!(addr = %addr_str, error = %e, "Invalid dns.bind_addr; DNS responder not started"); + Event::SubstrateFailed { child } + } + } + } + }; + + let feedback_actions = self.supervisor.fsm.step(feedback); + for action in &feedback_actions { + if let Action::PublishState(ns) = action { + start_outcome = Some(*ns); } } } - self.state = NodeState::Running; + // Seams that never triggered inside the loop: the "Transports + // initialized" info! when there was no non-transport child, and the + // peer-connect when there was no Tun/Dns child (today it still runs, + // after mDNS). + if !transports_info_emitted && !self.transports.is_empty() { + info!(count = self.transports.len(), "Transports initialized"); + } + if !peer_connect_done { + self.initiate_peer_connections().await; + } + + // Publish the FSM-resolved start-completion state (design doc §6/§9.1) + // instead of the old unconditional `Running`. + let outcome = start_outcome + .expect("supervisor publishes a start-completion state when bring-up resolves"); + self.supervisor.state = outcome; + + match outcome { + NodeState::Failed => { + // Zero transports came up — fatal. Tear down cleanly any + // children that DID come up (a failed start must not leave the + // node half-up), then return an error. `broadcast_disconnect = + // false`: there is nothing to gracefully disconnect on a start + // that never reached service. The daemon exits on this error. + warn!( + "Node start failed: no operational transports came up; tearing down partially-started children" + ); + let up = self.reconstruct_supervised_up(); + self.supervisor.fsm = SupervisorFsm::running_with(up); + let teardown = self.supervisor.fsm.step(Event::Stop); + self.execute_teardown(teardown, false).await; + return Err(NodeError::NoOperationalTransports); + } + NodeState::Degraded => { + // Operational but missing one or more configured optional + // children. Enumerate them for the operator, then proceed — + // a degraded node serves traffic. + warn!( + degraded_children = ?self.supervisor.fsm.failed(), + "Node started DEGRADED: one or more configured optional children failed to start" + ); + } + _ => {} + } + info!("Node started:"); - info!(" state: {}", self.state); + info!(" state: {}", self.supervisor.state); info!(" transports: {}", self.transports.len()); info!(" connections: {}", self.connections.len()); Ok(()) @@ -1518,99 +1678,278 @@ impl Node { /// Shuts down TUN interface, stops I/O threads, and transitions to /// the Stopped state. pub async fn stop(&mut self) -> Result<(), NodeError> { - if !self.state.can_stop() { + if !self.supervisor.state.can_stop() { return Err(NodeError::NotStarted); } - self.state = NodeState::Stopping; - info!(state = %self.state, "Node stopping"); + self.supervisor.state = NodeState::Stopping; + info!(state = %self.supervisor.state, "Node stopping"); - // Stop DNS responder - if let Some(handle) = self.dns_task.take() { - handle.abort(); - debug!("DNS responder stopped"); - } + // Reconstruct the supervised up-set from observed runtime presence and + // let the FSM author the teardown order (dns → nostr → mdns → + // 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); - // Send disconnect notifications to all active peers before closing transports - self.send_disconnect_to_all_peers(DisconnectReason::Shutdown) - .await; + // 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; - // Stop Nostr overlay discovery background work and withdraw any advert. - if let Some(bootstrap) = self.nostr_rendezvous.take_engine() - && let Err(e) = bootstrap.shutdown().await - { - warn!(error = %e, "Failed to shutdown Nostr overlay discovery"); - } - - // Tear down LAN mDNS responder + browser. Best-effort: the - // OS will eventually time the advert out via its TTL even if - // we don't get a clean unregister out before the daemon exits. - if let Some(lan) = self.lan_rendezvous.take() { - lan.shutdown().await; - } - - // Shutdown transports (they're packet producers) - let transport_ids: Vec<_> = self.transports.keys().cloned().collect(); - for transport_id in transport_ids { - if let Some(mut handle) = self.transports.remove(&transport_id) { - let transport_type = handle.transport_type().name; - match handle.stop().await { - Ok(()) => { - info!(transport_id = %transport_id, transport_type, "Transport stopped"); - } - Err(e) => { - warn!( - transport_id = %transport_id, - transport_type, - error = %e, - "Transport stop failed" - ); - } - } - } - } - - // Drop packet channels - self.packet_tx.take(); - self.packet_rx.take(); - - // Shutdown TUN interface - if let Some(name) = self.tun_name.take() { - info!(name = %name, "Shutting down TUN interface"); - - // Drop the tun_tx to signal the writer to stop - self.tun_tx.take(); - - // Delete the interface (on Linux, causes reader to get EFAULT) - if let Err(e) = shutdown_tun_interface(&name).await { - warn!(name = %name, error = %e, "Failed to shutdown TUN interface"); - } - - // On macOS, signal the reader thread to exit by writing to the - // shutdown pipe. The reader's select() will wake up and break. - #[cfg(target_os = "macos")] - if let Some(fd) = self.tun_shutdown_fd.take() { - unsafe { - libc::write(fd, b"x".as_ptr() as *const libc::c_void, 1); - libc::close(fd); - } - } - - // Wait for threads to finish - if let Some(handle) = self.tun_reader_handle.take() { - let _ = handle.join(); - } - if let Some(handle) = self.tun_writer_handle.take() { - let _ = handle.join(); - } - - self.tun_state = TunState::Disabled; - } - - self.state = NodeState::Stopped; - info!(state = %self.state, "Node stopped"); + 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 { + continue; + }; + + // Seam (a): send disconnect notifications to all active peers + // before closing transports — after any Dns teardown, before + // everything else. + if !disconnect_done && !matches!(child, Child::Dns) { + self.send_disconnect_to_all_peers(DisconnectReason::Shutdown) + .await; + disconnect_done = true; + } + + // Seam (b): drop the packet channels after all transports have + // stopped and before the TUN teardown. + if !packet_taken && matches!(child, Child::Tun) { + self.supervisor.packet_tx.take(); + self.packet_rx.take(); + packet_taken = true; + } + + match child { + Child::Dns => { + // Stop DNS responder + if let Some(handle) = self.supervisor.dns_task.take() { + handle.abort(); + debug!("DNS responder stopped"); + } + } + Child::Nostr => { + // Stop Nostr overlay discovery background work and withdraw + // any advert. + if let Some(bootstrap) = self.supervisor.nostr_rendezvous.take_engine() + && let Err(e) = bootstrap.shutdown().await + { + warn!(error = %e, "Failed to shutdown Nostr overlay discovery"); + } + } + Child::Mdns => { + // Tear down LAN mDNS responder + browser. Best-effort: the + // OS will eventually time the advert out via its TTL even if + // we don't get a clean unregister out before the daemon exits. + if let Some(lan) = self.supervisor.lan_rendezvous.take() { + lan.shutdown().await; + } + } + Child::Transport(id) => { + // Shutdown transport (they're packet producers) + if let Some(mut handle) = self.transports.remove(&id) { + let transport_type = handle.transport_type().name; + match handle.stop().await { + Ok(()) => { + info!(transport_id = %id, transport_type, "Transport stopped"); + } + Err(e) => { + warn!( + transport_id = %id, + transport_type, + error = %e, + "Transport stop failed" + ); + } + } + } + } + Child::Tun => { + // Shutdown TUN interface + if let Some(name) = self.tun_name.take() { + info!(name = %name, "Shutting down TUN interface"); + + // Drop the tun_tx to signal the writer to stop + self.supervisor.tun_tx.take(); + + // Delete the interface (on Linux, causes reader to get EFAULT) + if let Err(e) = shutdown_tun_interface(&name).await { + warn!(name = %name, error = %e, "Failed to shutdown TUN interface"); + } + + // On macOS, signal the reader thread to exit by writing to the + // shutdown pipe. The reader's select() will wake up and break. + #[cfg(target_os = "macos")] + if let Some(fd) = self.supervisor.tun_shutdown_fd.take() { + unsafe { + libc::write(fd, b"x".as_ptr() as *const libc::c_void, 1); + libc::close(fd); + } + } + + // Wait for threads to finish + if let Some(handle) = self.supervisor.tun_reader_handle.take() { + let _ = handle.join(); + } + if let Some(handle) = self.supervisor.tun_writer_handle.take() { + let _ = handle.join(); + } + + self.tun_state = TunState::Disabled; + } + } + Child::EncryptWorkers | Child::DecryptWorkers => { + // Worker pools are never torn down in stop() (matches + // today); the FSM never emits StopChild for them, so this + // is unreachable. + } + } + + self.supervisor.fsm.step(Event::ChildStopped { child }); + } + + // Seams that never triggered inside the loop (no non-Dns child for the + // disconnect fan-out, no Tun child for dropping the packet channels). + if !disconnect_done { + self.send_disconnect_to_all_peers(DisconnectReason::Shutdown) + .await; + } + if !packet_taken { + self.supervisor.packet_tx.take(); + self.packet_rx.take(); + } + } + + /// 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(_) | Action::PublishState(_) => { + // Drain entry never emits child or publish-state actions + // (the `Draining` state is a direct write above); 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. /// /// Best-effort: send failures are logged and ignored since the transport @@ -1703,6 +2042,7 @@ impl Node { continue; } if self + .supervisor .nostr_rendezvous .request_nostr_bootstrap(peer_config) .await @@ -2027,7 +2367,7 @@ impl Node { &mut self, bootstrap: &std::sync::Arc, ) { - if self.nostr_rendezvous.startup_sweep_done() { + if self.supervisor.nostr_rendezvous.startup_sweep_done() { return; } if !self.config().node.rendezvous.nostr.enabled @@ -2035,10 +2375,10 @@ impl Node { != crate::config::NostrRendezvousPolicy::Open { // Mark done so we don't keep re-checking on every tick. - self.nostr_rendezvous.set_startup_sweep_done(); + self.supervisor.nostr_rendezvous.set_startup_sweep_done(); return; } - let Some(started_at_ms) = self.nostr_rendezvous.started_at_ms() else { + let Some(started_at_ms) = self.supervisor.nostr_rendezvous.started_at_ms() else { return; }; let now_ms = Self::now_ms(); @@ -2061,7 +2401,7 @@ impl Node { .startup_sweep_max_age_secs; self.run_open_discovery_sweep(bootstrap, Some(max_age_secs), "startup") .await; - self.nostr_rendezvous.set_startup_sweep_done(); + self.supervisor.nostr_rendezvous.set_startup_sweep_done(); } fn available_outbound_slots(&self) -> usize { @@ -2231,7 +2571,8 @@ impl Node { bootstrap: &std::sync::Arc, ) -> Result<(), crate::nostr::BootstrapError> { let snapshot = self.advert_transport_snapshot(); - self.nostr_rendezvous + self.supervisor + .nostr_rendezvous .refresh_overlay_advert(bootstrap, snapshot, &self.config().node.rendezvous.nostr) .await } @@ -2350,6 +2691,7 @@ impl Node { async fn peer_address_candidates(&self, peer_config: &PeerConfig) -> Vec { let static_addresses = self.static_peer_addresses(peer_config); let overlay_addresses = self + .supervisor .nostr_rendezvous .nostr_peer_fallback_addresses( peer_config, @@ -2424,7 +2766,7 @@ impl Node { }; if peer .transport_id() - .map(|id| self.nostr_rendezvous.is_bootstrap_transport(&id)) + .map(|id| self.supervisor.nostr_rendezvous.is_bootstrap_transport(&id)) .unwrap_or(false) { return false; @@ -2539,11 +2881,15 @@ impl Node { "adopting established traversal socket" ); - if !self.state.is_operational() { + if !self.supervisor.state.is_operational() { return Err(NodeError::NotStarted); } - let packet_tx = self.packet_tx.clone().ok_or(NodeError::NotStarted)?; + let packet_tx = self + .supervisor + .packet_tx + .clone() + .ok_or(NodeError::NotStarted)?; let peer_identity = PeerIdentity::from_npub(&traversal.peer_npub).map_err(|e| { NodeError::InvalidPeerNpub { npub: traversal.peer_npub.clone(), @@ -2620,7 +2966,8 @@ impl Node { transport_id, crate::transport::TransportHandle::Udp(transport), ); - self.nostr_rendezvous + self.supervisor + .nostr_rendezvous .insert_bootstrap_transport(transport_id, traversal.peer_npub.clone()); let remote_addr = TransportAddr::from_string(&traversal.remote_addr.to_string()); @@ -2628,7 +2975,8 @@ impl Node { .initiate_connection(transport_id, remote_addr.clone(), Some(peer_identity)) .await { - self.nostr_rendezvous + self.supervisor + .nostr_rendezvous .remove_bootstrap_transport(&transport_id); if let Some(mut handle) = self.transports.remove(&transport_id) { let _ = handle.stop().await; diff --git a/src/node/lifecycle/supervisor.rs b/src/node/lifecycle/supervisor.rs new file mode 100644 index 0000000..30a1ba3 --- /dev/null +++ b/src/node/lifecycle/supervisor.rs @@ -0,0 +1,1174 @@ +//! Node lifecycle supervisor — sans-IO core (Milestone-1 Step 1a). +//! +//! A synchronous `step(event) -> Vec` finite-state machine over the +//! fixed set of substrate children. It owns the *decision* of what to bring up +//! and tear down and in what order; the async driver in [`super`] +//! (`start()`/`stop()`) performs the actual I/O each [`Action`] names and reports +//! results back as [`Event`]s. The core reads no clock, performs no I/O, and +//! 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"). +//! +//! ## Scope: the behavior-neutral rewrite +//! +//! This is the first of the three Step-1a commits and is strictly +//! behavior-preserving. The machine mirrors today's `start()`/`stop()` exactly: +//! +//! - every configured child is spawned in the current order, and optional +//! failures are warn/debug-and-continue (there is no `Degraded` yet — a +//! failed child simply drains from `pending` and start still reaches +//! `Running`, as today an even-zero-transport node does); +//! - 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*, 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. +//! +//! ## 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. +//! +//! ## 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`]: +//! +//! - [`SupState::Running`] now carries a [`Health`] (`Full` or `Degraded`), and +//! [`SupState::Failed`] is the fatal path. When `Starting.pending` empties (or +//! the degenerate no-children path), the machine resolves health once +//! ([`SupervisorFsm::resolve_start_health`]): **zero transports up → `Failed`** +//! (fatal); **≥1 transport up but a configured optional child failed → +//! `Degraded`**; **everything configured came up → `Full`**. Not-configured +//! children never count (a node never asked to run DNS is not degraded for +//! lacking it); worker-pool failures are `Degraded` at most, never `Failed`. +//! - the health outcome is a fork that a single direct `self.state` write cannot +//! express, so the machine emits [`Action::PublishState`] carrying the resolved +//! [`NodeState`]; the driver writes it. The non-forking transitions +//! (`Starting`/`Draining`/`Stopping`/`Stopped`) keep their direct `self.state` +//! writes — only the start-completion health outcome routes through +//! `PublishState`, to minimize churn. +//! - the degenerate no-children path now resolves to `Failed` (zero transports), +//! **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 +//! resolution is start-framed, and liveness monitoring is a substantial unbuilt +//! mechanism. This commit is start-time health only. + +use std::collections::HashSet; +use std::sync::Arc; +use std::thread::JoinHandle; + +use crate::node::NodeState; +use crate::transport::{PacketTx, TransportId}; +use crate::upper::tun::{TunOutboundRx, TunTx}; + +/// A supervised substrate child (design doc §6 Core 1). +/// +/// 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. +/// The TUN device is a compound unit at the driver (a reader thread plus a +/// writer thread); the supervisor tracks it as the single `Tun` child, and the +/// driver joins both threads when it executes `StopChild(Tun)`. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] +pub(crate) enum Child { + /// A transport instance (UDP / TCP / Ethernet), keyed by its runtime id. + Transport(TransportId), + /// The off-task FMP-encrypt + UDP-send worker pool (`#[cfg(unix)]`). + EncryptWorkers, + /// The off-task FMP-decrypt worker pool (`#[cfg(unix)]`). + DecryptWorkers, + /// Nostr overlay rendezvous/discovery. + Nostr, + /// LAN mDNS / DNS-SD rendezvous. + Mdns, + /// The TUN device (reader + writer threads). + Tun, + /// The `.fips` DNS responder task. + Dns, +} + +/// An input to the supervisor. Results of executing [`Action`]s are fed back as +/// `SubstrateUp` / `SubstrateFailed` / `ChildStopped`. +/// +/// `Tick` and `ChildExited` (design doc §6) are **deferred**: `ChildExited` +/// belongs to the runtime child-liveness monitoring follow-up (this commit is +/// start-time health only). The bounded-drain events (`Drain` / +/// `DrainDeadlineElapsed`) and the start/up/failed/stop events are present here. +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) enum Event { + /// Begin bring-up. `transports` are the ids the driver has already created + /// (their ids are allocated at creation), in creation order; the booleans + /// mark which singleton children are configured. Valid from `Created` or + /// `Stopped`. + Start { + /// Created transport ids, in the order they must be started. + transports: Vec, + /// The `#[cfg(unix)]` encrypt worker pool is configured. + encrypt_workers: bool, + /// The `#[cfg(unix)]` decrypt worker pool is configured. + decrypt_workers: bool, + /// Nostr overlay discovery is enabled. + nostr: bool, + /// LAN mDNS discovery is enabled. + mdns: bool, + /// The TUN device is enabled. + tun: bool, + /// The DNS responder is enabled. + dns: bool, + }, + /// A child the driver was asked to spawn came up. + SubstrateUp { + /// The child that started successfully. + child: Child, + }, + /// A child the driver was asked to spawn failed to start. In the + /// behavior-neutral rewrite this is warn/debug-and-continue: the child + /// drains from `pending` and never joins the up-set (matching today), and + /// start still proceeds to `Running`. + SubstrateFailed { + /// The child that failed to start. + child: Child, + }, + /// 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. + child: Child, + }, +} + +/// 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) 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 +/// `self.state` writes for the non-forking transitions (`Starting`/`Draining`/ +/// `Stopping`/`Stopped`); only the start-completion health outcome routes through +/// `PublishState`. +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) enum Action { + /// Bring up this child (the driver performs the spawn / start I/O and + /// reports `SubstrateUp` or `SubstrateFailed`). + SpawnChild(Child), + /// Publish the given operator-visible [`NodeState`]. Emitted at start + /// completion (when `Starting.pending` empties, or the degenerate + /// 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). + PublishState(NodeState), + /// 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, +} + +/// Start-completion health (design doc §9.1). 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`]. +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) enum Health { + /// Every configured child came up. + Full, + /// ≥1 transport is up, but one or more configured optional children failed + /// to start (a transport beyond the first, Nostr, mDNS, TUN, DNS, or a + /// worker-pool spawn). The node is operational (serving) but degraded. + Degraded { + /// The configured children that failed to start. + reasons: HashSet, + }, +} + +/// Reason for the fatal [`SupState::Failed`] state (design doc §9.1). +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum FailReason { + /// Zero transports came up at start completion. Without a transport the node + /// cannot serve, so this is fatal (the driver tears down and returns an + /// error), unlike the degraded-but-serving optional-child failures. + NoTransports, +} + +/// Internal supervisor state (design doc §6). 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)] +pub(crate) enum SupState { + /// Constructed but not started. + Created, + /// Bringing children up; `pending` is the set not yet resolved. + Starting { + /// Children asked to spawn that have not yet reported up-or-failed. + pending: HashSet, + }, + /// All children resolved and ≥1 transport up; node operational. Carries the + /// resolved [`Health`] (`Full` or `Degraded`). + Running { + /// Resolved start-completion health. + health: Health, + }, + /// Start completed with zero transports up (design doc §9.1) — 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 + /// 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. + pending: HashSet, + }, + /// Fully torn down. + Stopped, +} + +/// The lifecycle supervisor FSM. +/// +/// Construct with [`SupervisorFsm::new`], feed [`Event`]s via [`SupervisorFsm::step`], +/// and execute the returned [`Action`]s. See the module docs for the +/// behavior-neutral scope. +#[derive(Clone, Debug)] +pub(crate) struct SupervisorFsm { + state: SupState, + /// Children currently up (present). Drives the teardown plan. + up: HashSet, + /// Configured children that failed to start during the current bring-up. + /// Feeds the `Degraded` health determination when `pending` empties. + failed: HashSet, +} + +impl SupervisorFsm { + /// A fresh supervisor in `Created`. + pub(crate) fn new() -> Self { + Self { + state: SupState::Created, + up: HashSet::new(), + failed: HashSet::new(), + } + } + + /// A supervisor seeded directly into `Running` with a known up-set. + /// + /// The teardown driver (`stop()`) reconstructs the up-set from observed + /// runtime presence (`dns_task.is_some()`, transports keys, etc.) rather + /// than relying on a live machine persisted across start/stop, so that + /// teardown ordering is authored here regardless of how the node reached + /// `Running`. Feeding `Event::Stop` then yields the ordered `StopChild` + /// plan over exactly the present children. + pub(crate) fn running_with(up: impl IntoIterator) -> Self { + Self { + state: SupState::Running { + health: Health::Full, + }, + up: up.into_iter().collect(), + failed: HashSet::new(), + } + } + + /// Current internal state (for the driver's bookkeeping and for tests). + #[cfg(test)] + pub(crate) fn state(&self) -> &SupState { + &self.state + } + + /// The configured children that failed to start during bring-up. The driver + /// reads this on the `Degraded` start outcome to enumerate the degraded + /// children in an operator-visible `warn!`. + pub(in crate::node) fn failed(&self) -> &HashSet { + &self.failed + } + + /// 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 { + Event::Start { + transports, + encrypt_workers, + decrypt_workers, + nostr, + mdns, + tun, + dns, + } => self.on_start( + transports, + encrypt_workers, + decrypt_workers, + nostr, + mdns, + tun, + dns, + ), + 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), + } + } + + #[allow(clippy::too_many_arguments)] + fn on_start( + &mut self, + transports: Vec, + encrypt_workers: bool, + decrypt_workers: bool, + nostr: bool, + mdns: bool, + tun: bool, + dns: bool, + ) -> Vec { + // Only meaningful from a not-running state (the driver also guards on + // `can_start`). Ignore otherwise. + if !matches!(self.state, SupState::Created | SupState::Stopped) { + return Vec::new(); + } + + // Canonical spawn order, mirroring today's `start()`: + // transports (creation order) → encrypt → decrypt → nostr → mdns → + // tun → dns. (The driver performs the peer-connect between mdns and + // tun; it is not a supervised child.) + let mut order: Vec = transports.into_iter().map(Child::Transport).collect(); + if encrypt_workers { + order.push(Child::EncryptWorkers); + } + if decrypt_workers { + order.push(Child::DecryptWorkers); + } + if nostr { + order.push(Child::Nostr); + } + if mdns { + order.push(Child::Mdns); + } + if tun { + order.push(Child::Tun); + } + if dns { + order.push(Child::Dns); + } + + self.up.clear(); + 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 + // change from the old immediate-`Running`). + if order.is_empty() { + return vec![Action::PublishState(self.resolve_start_health())]; + } + + self.state = SupState::Starting { + pending: order.iter().copied().collect(), + }; + order.into_iter().map(Action::SpawnChild).collect() + } + + fn on_substrate_up(&mut self, child: Child) -> Vec { + let SupState::Starting { pending } = &mut self.state else { + return Vec::new(); + }; + pending.remove(&child); + let emptied = pending.is_empty(); + self.up.insert(child); + if emptied { + vec![Action::PublishState(self.resolve_start_health())] + } else { + Vec::new() + } + } + + fn on_substrate_failed(&mut self, child: Child) -> Vec { + // Record the failed child (design doc §9.1): 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 { + return Vec::new(); + }; + pending.remove(&child); + let emptied = pending.is_empty(); + self.failed.insert(child); + if emptied { + vec![Action::PublishState(self.resolve_start_health())] + } else { + Vec::new() + } + } + + /// Resolve start-completion health (design doc §9.1) and set the resulting + /// state, returning the [`NodeState`] the driver should publish. Called once + /// when `Starting.pending` empties (or the degenerate no-children path): + /// + /// - zero transports up → [`SupState::Failed`] / [`NodeState::Failed`]; + /// - ≥1 transport up but some configured child failed → [`Health::Degraded`] + /// / [`NodeState::Degraded`]; + /// - everything configured came up → [`Health::Full`] / [`NodeState::Running`]. + /// + /// 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). + /// + /// This is start-time health only. Runtime child-liveness monitoring (a + /// `ChildExited` event re-routing health when a task/thread dies at runtime) + /// is a deferred follow-up (design doc §7 / §9.1 is start-framed). + fn resolve_start_health(&mut self) -> NodeState { + let transports_up = self + .up + .iter() + .filter(|c| matches!(c, Child::Transport(_))) + .count(); + if transports_up == 0 { + self.state = SupState::Failed { + reason: FailReason::NoTransports, + }; + NodeState::Failed + } else if !self.failed.is_empty() { + self.state = SupState::Running { + health: Health::Degraded { + reasons: self.failed.clone(), + }, + }; + NodeState::Degraded + } else { + self.state = SupState::Running { + health: Health::Full, + }; + NodeState::Running + } + } + + fn on_stop(&mut self) -> Vec { + 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 (either health). 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(), + }; + order.into_iter().map(Action::StopChild).collect() + } + + fn on_child_stopped(&mut self, child: Child) -> Vec { + if let SupState::Stopping { pending } = &mut self.state { + pending.remove(&child); + self.up.remove(&child); + if pending.is_empty() { + self.state = SupState::Stopped; + } + } + Vec::new() + } + + /// Teardown order over the up-set, mirroring today's `stop()`: + /// dns → nostr → mdns → transports (ascending id) → tun. + /// + /// The encrypt/decrypt worker pools are deliberately excluded: today's + /// `stop()` spawns them in `start()` but never tears them down. Transports + /// are ordered by ascending id for determinism (today's `stop()` iterates + /// them in nondeterministic `HashMap` order, so this is neutral). + fn teardown_order(&self) -> Vec { + let mut order = Vec::new(); + if self.up.contains(&Child::Dns) { + order.push(Child::Dns); + } + if self.up.contains(&Child::Nostr) { + order.push(Child::Nostr); + } + if self.up.contains(&Child::Mdns) { + order.push(Child::Mdns); + } + let mut transports: Vec = self + .up + .iter() + .filter_map(|c| match c { + Child::Transport(id) => Some(*id), + _ => None, + }) + .collect(); + transports.sort_by_key(|id| id.as_u32()); + order.extend(transports.into_iter().map(Child::Transport)); + if self.up.contains(&Child::Tun) { + order.push(Child::Tun); + } + order + } +} + +/// Owner of the node's lifecycle-managed substrate handles plus the sans-IO +/// [`SupervisorFsm`] that authors their spawn/teardown ordering. +/// +/// The fields moved here off `Node` are exactly the children the supervisor +/// governs — the packet-send channel, the TUN reader/writer plumbing, the DNS +/// responder task, the Nostr/LAN rendezvous drivers, and (on unix) the +/// encrypt/decrypt worker pools — together with the published `NodeState`. +/// This is a pure relocation: the driver (`start()`/`stop()`) reaches each +/// field through `self.supervisor.*`, and the initializers are the same ones +/// `Node::new` used. +pub(crate) struct Supervisor { + /// Node operational state (the published `NodeState`; the driver keeps its + /// verbatim writes here at their current positions). + pub(in crate::node) state: NodeState, + + /// Packet sender for transports. + pub(in crate::node) packet_tx: Option, + + /// TUN packet sender channel. + pub(in crate::node) tun_tx: Option, + /// Receiver for outbound packets from the TUN reader. + pub(in crate::node) tun_outbound_rx: Option, + /// TUN reader thread handle. + pub(in crate::node) tun_reader_handle: Option>, + /// TUN writer thread handle. + pub(in crate::node) tun_writer_handle: Option>, + /// Shutdown pipe: writing to this fd unblocks the TUN reader thread on macOS. + /// On Linux, deleting the interface via netlink serves the same purpose. + #[cfg(target_os = "macos")] + pub(in crate::node) tun_shutdown_fd: Option, + + /// Receiver for resolved identities from the DNS responder. + pub(in crate::node) dns_identity_rx: Option, + /// DNS responder task handle. + pub(in crate::node) dns_task: Option>, + + /// Node-side driver state for the Nostr overlay peer-rendezvous + /// subsystem: the engine handle, its startup timestamp, the one-shot + /// startup-sweep latch, and the per-peer bootstrap-transport bookkeeping + /// adopted from NAT-traversal handoffs. + pub(in crate::node) nostr_rendezvous: crate::nostr::RendezvousDriver, + /// mDNS / DNS-SD responder + browser for local-link peer discovery. + /// Identity is unverified at this layer — the Noise XX handshake + /// initiated against an mDNS-observed endpoint is what proves the + /// peer holds the matching private key. + pub(in crate::node) lan_rendezvous: Option>, + + /// Off-task FMP-encrypt + UDP-send worker pool. Unix-only — + /// the worker issues direct sendmmsg(2) / sendmsg+UDP_GSO calls + /// on raw fds via `AsRawFd`. None on Windows or when the worker + /// pool failed to spawn. + #[cfg(unix)] + pub(crate) encrypt_workers: Option, + + /// Off-task FMP decrypt worker pool — receiver-side mirror of + /// `encrypt_workers`. Workers are shards: each owns its session + /// state directly in a thread-local `HashMap` (no `RwLock`, + /// no `Mutex` per packet). Hash-by-cache-key dispatch. + #[cfg(unix)] + pub(crate) decrypt_workers: Option, + + /// The sans-IO lifecycle FSM authoring spawn/teardown ordering. + pub(in crate::node) fsm: SupervisorFsm, +} + +impl Supervisor { + /// A fresh supervisor with all handles empty and the FSM in `Created`, + /// matching the field initializers `Node::new` previously used. + pub(crate) fn new() -> Self { + Self { + state: NodeState::Created, + packet_tx: None, + tun_tx: None, + tun_outbound_rx: None, + tun_reader_handle: None, + tun_writer_handle: None, + #[cfg(target_os = "macos")] + tun_shutdown_fd: None, + dns_identity_rx: None, + dns_task: None, + nostr_rendezvous: crate::nostr::RendezvousDriver::default(), + lan_rendezvous: None, + #[cfg(unix)] + encrypt_workers: None, + #[cfg(unix)] + decrypt_workers: None, + fsm: SupervisorFsm::new(), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn tid(n: u32) -> TransportId { + TransportId::new(n) + } + + fn start_full() -> Event { + Event::Start { + transports: vec![tid(1), tid(2)], + encrypt_workers: true, + decrypt_workers: true, + nostr: true, + mdns: true, + tun: true, + dns: true, + } + } + + #[test] + fn start_emits_spawn_in_canonical_order() { + let mut s = SupervisorFsm::new(); + let actions = s.step(start_full()); + assert_eq!( + actions, + vec![ + Action::SpawnChild(Child::Transport(tid(1))), + Action::SpawnChild(Child::Transport(tid(2))), + Action::SpawnChild(Child::EncryptWorkers), + Action::SpawnChild(Child::DecryptWorkers), + Action::SpawnChild(Child::Nostr), + Action::SpawnChild(Child::Mdns), + Action::SpawnChild(Child::Tun), + Action::SpawnChild(Child::Dns), + ] + ); + assert!(matches!(s.state(), SupState::Starting { .. })); + } + + #[test] + fn all_configured_up_reaches_full() { + // Everything configured came up (2 transports + all optional children) + // → Full; the pending-emptying step publishes `Running`. + let mut s = SupervisorFsm::new(); + let spawns = s.step(start_full()); + let children: Vec = spawns + .into_iter() + .map(|a| match a { + Action::SpawnChild(c) => c, + _ => panic!("unexpected action"), + }) + .collect(); + let last = children.len() - 1; + for (i, child) in children.into_iter().enumerate() { + let out = s.step(Event::SubstrateUp { child }); + if i == last { + assert_eq!(out, vec![Action::PublishState(NodeState::Running)]); + } else { + assert_eq!(out, vec![]); + } + } + assert_eq!( + s.state(), + &SupState::Running { + health: Health::Full + } + ); + assert!(s.failed().is_empty()); + } + + #[test] + fn configured_optional_child_failure_is_degraded() { + // A configured optional child (mDNS) fails but ≥1 transport is up → + // Degraded, with the failed child in `reasons`. The node stays + // operational and tears down cleanly (the failed child never joined the + // up-set, so it is excluded from teardown; workers excluded by design). + let mut s = SupervisorFsm::new(); + s.step(start_full()); + for child in [ + Child::Transport(tid(1)), + Child::Transport(tid(2)), + Child::EncryptWorkers, + Child::DecryptWorkers, + Child::Nostr, + ] { + assert_eq!(s.step(Event::SubstrateUp { child }), vec![]); + } + // mdns fails, tun comes up, dns comes up last (empties pending). + assert_eq!( + s.step(Event::SubstrateFailed { child: Child::Mdns }), + vec![] + ); + assert_eq!(s.step(Event::SubstrateUp { child: Child::Tun }), vec![]); + assert_eq!( + s.step(Event::SubstrateUp { child: Child::Dns }), + vec![Action::PublishState(NodeState::Degraded)] + ); + let mut expected_reasons = HashSet::new(); + expected_reasons.insert(Child::Mdns); + assert_eq!( + s.state(), + &SupState::Running { + health: Health::Degraded { + reasons: expected_reasons.clone() + } + } + ); + assert_eq!(s.failed(), &expected_reasons); + + let stops = s.step(Event::Stop); + // mdns must not appear in teardown; workers excluded by design. + assert_eq!( + stops, + vec![ + Action::StopChild(Child::Dns), + Action::StopChild(Child::Nostr), + Action::StopChild(Child::Transport(tid(1))), + Action::StopChild(Child::Transport(tid(2))), + Action::StopChild(Child::Tun), + ] + ); + } + + #[test] + fn worker_pool_failure_is_degraded_not_failed() { + // A worker-pool spawn failure is Degraded at most, never Failed + // (inline crypto fallback keeps the node correct). One transport is up. + let mut s = SupervisorFsm::new(); + s.step(Event::Start { + transports: vec![tid(1)], + encrypt_workers: true, + decrypt_workers: false, + nostr: false, + mdns: false, + tun: false, + dns: false, + }); + assert_eq!( + s.step(Event::SubstrateUp { + child: Child::Transport(tid(1)) + }), + vec![] + ); + assert_eq!( + s.step(Event::SubstrateFailed { + child: Child::EncryptWorkers + }), + vec![Action::PublishState(NodeState::Degraded)] + ); + let mut expected = HashSet::new(); + expected.insert(Child::EncryptWorkers); + assert_eq!( + s.state(), + &SupState::Running { + health: Health::Degraded { reasons: expected } + } + ); + } + + #[test] + fn not_configured_child_does_not_cause_degraded() { + // A node that never asked to run mDNS/TUN/DNS/Nostr is not degraded for + // lacking them: only a configured-and-failed child counts. One transport + // configured and up, nothing else configured → Full. + let mut s = SupervisorFsm::new(); + s.step(Event::Start { + transports: vec![tid(1)], + encrypt_workers: false, + decrypt_workers: false, + nostr: false, + mdns: false, + tun: false, + dns: false, + }); + assert_eq!( + s.step(Event::SubstrateUp { + child: Child::Transport(tid(1)) + }), + vec![Action::PublishState(NodeState::Running)] + ); + assert_eq!( + s.state(), + &SupState::Running { + health: Health::Full + } + ); + } + + #[test] + fn zero_transports_up_is_failed_via_child_failures() { + // Transports were configured but all failed → zero transports up → + // Failed (fatal), even though other children came up. Failed takes + // priority over Degraded. + let mut s = SupervisorFsm::new(); + s.step(Event::Start { + transports: vec![tid(1)], + encrypt_workers: false, + decrypt_workers: false, + nostr: true, + mdns: false, + tun: false, + dns: false, + }); + assert_eq!( + s.step(Event::SubstrateFailed { + child: Child::Transport(tid(1)) + }), + vec![] + ); + assert_eq!( + s.step(Event::SubstrateUp { + child: Child::Nostr + }), + vec![Action::PublishState(NodeState::Failed)] + ); + assert_eq!( + s.state(), + &SupState::Failed { + reason: FailReason::NoTransports + } + ); + } + + #[test] + fn no_children_is_failed_immediately() { + // The degenerate empty-`Start` path: zero transports → Failed (the + // behavioral change from the old immediate-Running). + let mut s = SupervisorFsm::new(); + let actions = s.step(Event::Start { + transports: vec![], + encrypt_workers: false, + decrypt_workers: false, + nostr: false, + mdns: false, + tun: false, + dns: false, + }); + assert_eq!(actions, vec![Action::PublishState(NodeState::Failed)]); + assert_eq!( + s.state(), + &SupState::Failed { + reason: FailReason::NoTransports + } + ); + } + + #[test] + fn stop_teardown_order_excludes_workers() { + let mut s = SupervisorFsm::new(); + s.step(start_full()); + for child in [ + Child::Transport(tid(2)), + Child::Transport(tid(1)), + Child::EncryptWorkers, + Child::DecryptWorkers, + Child::Nostr, + Child::Mdns, + Child::Tun, + Child::Dns, + ] { + s.step(Event::SubstrateUp { child }); + } + let stops = s.step(Event::Stop); + assert_eq!( + stops, + vec![ + Action::StopChild(Child::Dns), + Action::StopChild(Child::Nostr), + Action::StopChild(Child::Mdns), + // transports ascending by id regardless of up-report order + Action::StopChild(Child::Transport(tid(1))), + Action::StopChild(Child::Transport(tid(2))), + Action::StopChild(Child::Tun), + ] + ); + assert!(matches!(s.state(), SupState::Stopping { .. })); + } + + #[test] + fn all_children_stopped_reaches_stopped() { + let mut s = SupervisorFsm::new(); + s.step(start_full()); + // Every spawned child reports an outcome: five come up, three fail. + // `pending` drains fully; one transport (tid(1)) is up so the node + // reaches `Running`, but a configured transport (tid(2)) and two + // configured optional children failed → Degraded. + for child in [ + Child::Transport(tid(1)), + Child::EncryptWorkers, + Child::Nostr, + Child::Tun, + Child::Dns, + ] { + s.step(Event::SubstrateUp { child }); + } + for child in [Child::Transport(tid(2)), Child::DecryptWorkers, Child::Mdns] { + s.step(Event::SubstrateFailed { child }); + } + assert!(matches!( + s.state(), + SupState::Running { + health: Health::Degraded { .. } + } + )); + + // Only the children that came up are torn down; the failed ones never + // joined the up-set. + let stops = s.step(Event::Stop); + assert_eq!( + stops, + vec![ + Action::StopChild(Child::Dns), + Action::StopChild(Child::Nostr), + Action::StopChild(Child::Transport(tid(1))), + Action::StopChild(Child::Tun), + ] + ); + for a in stops { + let child = match a { + Action::StopChild(c) => c, + _ => panic!("unexpected action"), + }; + assert_eq!(s.step(Event::ChildStopped { child }), vec![]); + } + assert_eq!(s.state(), &SupState::Stopped); + } + + #[test] + fn late_substrate_up_in_running_is_inert() { + let mut s = SupervisorFsm::new(); + s.step(Event::Start { + transports: vec![tid(1)], + encrypt_workers: false, + decrypt_workers: false, + nostr: false, + mdns: false, + tun: false, + dns: false, + }); + s.step(Event::SubstrateUp { + child: Child::Transport(tid(1)), + }); + assert_eq!( + s.state(), + &SupState::Running { + health: Health::Full + } + ); + // A stray event in Running produces nothing and does not change state. + assert_eq!( + s.step(Event::SubstrateUp { + child: Child::Nostr + }), + vec![] + ); + assert_eq!( + s.state(), + &SupState::Running { + health: Health::Full + } + ); + } + + #[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 { + health: Health::Full + } + ); + } +} diff --git a/src/node/mod.rs b/src/node/mod.rs index f0a63ee..52ac05a 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -75,14 +75,13 @@ use crate::transport::{ }; use crate::upper::hosts::HostMap; use crate::upper::icmp_rate_limit::IcmpRateLimiter; -use crate::upper::tun::{TunError, TunOutboundRx, TunState, TunTx}; +use crate::upper::tun::{TunError, TunState, TunTx}; use crate::utils::index::IndexAllocator; use crate::{Config, ConfigError, Identity, IdentityError, NodeAddr, PeerIdentity, TreeCoordinate}; use rand::Rng; use std::collections::{BTreeSet, HashMap, VecDeque}; use std::fmt; use std::sync::Arc; -use std::thread::JoinHandle; use thiserror::Error; /// Errors related to node operations. @@ -172,6 +171,9 @@ pub enum NodeError { #[error("bootstrap handoff failed: {0}")] BootstrapHandoff(String), + + #[error("node start failed: no operational transports")] + NoOperationalTransports, } /// Node operational state. @@ -181,8 +183,23 @@ pub enum NodeState { Created, /// Starting up (initializing transports). Starting, - /// Fully operational. + /// Fully operational — every configured child came up (design doc §9.1). Running, + /// Operational but degraded (design doc §9.1): ≥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 + /// `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 + /// driver tears down any children that did come up and `start()` returns + /// an error. Not operational and not restartable in-process. + Failed, /// Shutting down. Stopping, /// Stopped. @@ -190,19 +207,23 @@ pub enum NodeState { } impl NodeState { - /// Check if node is operational. + /// Check if node is operational. A `Degraded` node is operational — it is + /// serving, just missing an optional child (design doc §9.1). pub fn is_operational(&self) -> bool { - matches!(self, NodeState::Running) + matches!(self, NodeState::Running | NodeState::Degraded) } - /// Check if node can be started. + /// 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 + /// `Stopped` node can start. pub fn can_start(&self) -> bool { matches!(self, NodeState::Created | NodeState::Stopped) } - /// Check if node can be stopped. + /// Check if node can be stopped. Both `Running` and `Degraded` nodes are + /// operational and can be stopped or drained. pub fn can_stop(&self) -> bool { - matches!(self, NodeState::Running) + matches!(self, NodeState::Running | NodeState::Degraded) } } @@ -212,6 +233,9 @@ impl fmt::Display for NodeState { NodeState::Created => "created", NodeState::Starting => "starting", NodeState::Running => "running", + NodeState::Degraded => "degraded", + NodeState::Draining => "draining", + NodeState::Failed => "failed", NodeState::Stopping => "stopping", NodeState::Stopped => "stopped", }; @@ -287,9 +311,13 @@ pub struct Node { /// `update_peers`; readers reach it through the accessors. context: Arc, - // === State === - /// Node operational state. - state: NodeState, + // === Lifecycle Supervisor === + /// Owner of the lifecycle-managed substrate handles (packet-send channel, + /// TUN plumbing, DNS task, Nostr/LAN rendezvous, encrypt/decrypt worker + /// pools) plus the published `NodeState` and the sans-IO supervisor FSM + /// that authors their spawn/teardown ordering. Reached via + /// `self.supervisor.*`. + supervisor: lifecycle::supervisor::Supervisor, // === Spanning Tree === /// Local spanning tree state. @@ -320,8 +348,6 @@ pub struct Node { addr_to_link: HashMap, // === Packet Channel === - /// Packet sender for transports. - packet_tx: Option, /// Packet receiver (for event loop). packet_rx: Option, @@ -398,24 +424,6 @@ pub struct Node { tun_state: TunState, /// TUN interface name (for cleanup). tun_name: Option, - /// TUN packet sender channel. - tun_tx: Option, - /// Receiver for outbound packets from the TUN reader. - tun_outbound_rx: Option, - /// TUN reader thread handle. - tun_reader_handle: Option>, - /// TUN writer thread handle. - tun_writer_handle: Option>, - /// Shutdown pipe: writing to this fd unblocks the TUN reader thread on macOS. - /// On Linux, deleting the interface via netlink serves the same purpose. - #[cfg(target_os = "macos")] - tun_shutdown_fd: Option, - - // === DNS Responder === - /// Receiver for resolved identities from the DNS responder. - dns_identity_rx: Option, - /// DNS responder task handle. - dns_task: Option>, // === Index-Based Session Dispatch === /// Allocator for session indices. @@ -464,17 +472,6 @@ pub struct Node { /// are exhausted. retry_pending: HashMap, - /// Node-side driver state for the Nostr overlay peer-rendezvous - /// subsystem: the engine handle, its startup timestamp, the one-shot - /// startup-sweep latch, and the per-peer bootstrap-transport bookkeeping - /// adopted from NAT-traversal handoffs. - nostr_rendezvous: crate::nostr::RendezvousDriver, - /// mDNS / DNS-SD responder + browser for local-link peer discovery. - /// Identity is unverified at this layer — the Noise XX handshake - /// initiated against an mDNS-observed endpoint is what proves the - /// peer holds the matching private key. - lan_rendezvous: Option>, - // === Periodic Parent Re-evaluation === /// Timestamp of last periodic parent re-evaluation (for pacing). last_parent_reeval: Option, @@ -510,20 +507,6 @@ pub struct Node { /// published through a lock-free snapshot for the display path. host_map: reloadable::HostMapReloadable, - /// Off-task FMP-encrypt + UDP-send worker pool. Unix-only — - /// the worker issues direct sendmmsg(2) / sendmsg+UDP_GSO calls - /// on raw fds via `AsRawFd`. None on Windows or when the worker - /// pool failed to spawn. - #[cfg(unix)] - pub(crate) encrypt_workers: Option, - - /// Off-task FMP decrypt worker pool — receiver-side mirror of - /// `encrypt_workers`. Workers are shards: each owns its session - /// state directly in a thread-local `HashMap` (no `RwLock`, - /// no `Mutex` per packet). Hash-by-cache-key dispatch. - #[cfg(unix)] - pub(crate) decrypt_workers: Option, - /// Sessions whose recv cipher + replay window have been handed /// off to a decrypt shard worker. Lookup gate on the hot receive /// path: if the cache-key is in here, dispatch to worker; else @@ -634,7 +617,7 @@ impl Node { Ok(Self { context, - state: NodeState::Created, + supervisor: lifecycle::supervisor::Supervisor::new(), tree_state, bloom_state, coord_cache, @@ -642,7 +625,6 @@ impl Node { transport_drops: HashMap::new(), links: HashMap::new(), addr_to_link: HashMap::new(), - packet_tx: None, packet_rx: None, connections: HashMap::new(), peers: HashMap::new(), @@ -665,14 +647,6 @@ impl Node { )), tun_state, tun_name: None, - tun_tx: None, - tun_outbound_rx: None, - tun_reader_handle: None, - tun_writer_handle: None, - #[cfg(target_os = "macos")] - tun_shutdown_fd: None, - dns_identity_rx: None, - dns_task: None, index_allocator: IndexAllocator::new(), peers_by_index: HashMap::new(), pending_outbound: HashMap::new(), @@ -692,8 +666,6 @@ impl Node { ), pending_connects: Vec::new(), retry_pending: HashMap::new(), - nostr_rendezvous: crate::nostr::RendezvousDriver::default(), - lan_rendezvous: None, last_parent_reeval: None, last_congestion_log: None, estimated_mesh_size: None, @@ -704,10 +676,6 @@ impl Node { host_map, path_mtu_lookup: Arc::new(std::sync::RwLock::new(HashMap::new())), #[cfg(unix)] - encrypt_workers: None, - #[cfg(unix)] - decrypt_workers: None, - #[cfg(unix)] decrypt_registered_sessions: std::collections::HashSet::new(), #[cfg(unix)] decrypt_fallback_rx: Some(decrypt_fallback_rx), @@ -797,7 +765,7 @@ impl Node { Ok(Self { context, - state: NodeState::Created, + supervisor: lifecycle::supervisor::Supervisor::new(), tree_state, bloom_state, coord_cache, @@ -805,7 +773,6 @@ impl Node { transport_drops: HashMap::new(), links: HashMap::new(), addr_to_link: HashMap::new(), - packet_tx: None, packet_rx: None, connections: HashMap::new(), peers: HashMap::new(), @@ -828,14 +795,6 @@ impl Node { )), tun_state, tun_name: None, - tun_tx: None, - tun_outbound_rx: None, - tun_reader_handle: None, - tun_writer_handle: None, - #[cfg(target_os = "macos")] - tun_shutdown_fd: None, - dns_identity_rx: None, - dns_task: None, index_allocator: IndexAllocator::new(), peers_by_index: HashMap::new(), pending_outbound: HashMap::new(), @@ -852,8 +811,6 @@ impl Node { lookup: Lookup::new(LookupBackoff::new(), LookupForwardRateLimiter::new()), pending_connects: Vec::new(), retry_pending: HashMap::new(), - nostr_rendezvous: crate::nostr::RendezvousDriver::default(), - lan_rendezvous: None, last_parent_reeval: None, last_congestion_log: None, estimated_mesh_size: None, @@ -864,10 +821,6 @@ impl Node { host_map, path_mtu_lookup: Arc::new(std::sync::RwLock::new(HashMap::new())), #[cfg(unix)] - encrypt_workers: None, - #[cfg(unix)] - decrypt_workers: None, - #[cfg(unix)] decrypt_registered_sessions: std::collections::HashSet::new(), #[cfg(unix)] decrypt_fallback_rx: Some(decrypt_fallback_rx), @@ -1234,7 +1187,7 @@ impl Node { /// Get the node state. pub fn state(&self) -> NodeState { - self.state + self.supervisor.state } /// Get the node uptime. @@ -1244,7 +1197,7 @@ impl Node { /// Check if node is operational. pub fn is_running(&self) -> bool { - self.state.is_operational() + self.supervisor.state.is_operational() } /// Check if this is a leaf-only node. @@ -1560,7 +1513,7 @@ impl Node { let snapshot = crate::control::snapshot::StatsSnapshot { history: std::sync::Arc::new(self.stats_history.clone()), estimated_mesh_size: self.estimated_mesh_size, - state: self.state, + state: self.supervisor.state, tun_state: self.tun_state, tun_name: self.tun_name.clone(), effective_ipv6_mtu: self.effective_ipv6_mtu(), @@ -2308,7 +2261,11 @@ impl Node { } pub(crate) fn cleanup_bootstrap_transport_if_unused(&mut self, transport_id: TransportId) { - if !self.nostr_rendezvous.is_bootstrap_transport(&transport_id) { + if !self + .supervisor + .nostr_rendezvous + .is_bootstrap_transport(&transport_id) + { return; } @@ -2338,7 +2295,8 @@ impl Node { "bootstrap transport has no remaining references; dropping" ); - self.nostr_rendezvous + self.supervisor + .nostr_rendezvous .remove_bootstrap_transport(&transport_id); self.transport_drops.remove(&transport_id); self.transports.remove(&transport_id); @@ -2415,7 +2373,7 @@ impl Node { /// Used by control queries (`show_peers` per-peer Nostr-traversal /// state) to read failure-state without taking shared ownership. pub fn nostr_rendezvous_handle(&self) -> Option<&crate::nostr::NostrRendezvous> { - self.nostr_rendezvous.engine() + self.supervisor.nostr_rendezvous.engine() } /// Iterate over all peer node IDs. @@ -2706,7 +2664,7 @@ impl Node { /// /// Returns None if TUN is not active or the node hasn't been started. pub fn tun_tx(&self) -> Option<&TunTx> { - self.tun_tx.as_ref() + self.supervisor.tun_tx.as_ref() } // === Sending === @@ -2800,7 +2758,7 @@ impl Node { { let send_cipher_opt = session.send_cipher_clone(); if let Some(fmp_cipher) = send_cipher_opt - && let Some(workers) = self.encrypt_workers.as_ref().cloned() + && let Some(workers) = self.supervisor.encrypt_workers.as_ref().cloned() && let Some(transport) = self.transports.get(&transport_id) && let TransportHandle::Udp(udp) = transport && let Some(socket) = udp.async_socket() @@ -3097,7 +3055,7 @@ impl fmt::Debug for Node { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("Node") .field("node_addr", self.node_addr()) - .field("state", &self.state) + .field("state", &self.supervisor.state) .field("is_leaf_only", &self.is_leaf_only()) .field("connections", &self.connection_count()) .field("peers", &self.peer_count()) diff --git a/src/node/retry.rs b/src/node/retry.rs index ba422aa..0a97449 100644 --- a/src/node/retry.rs +++ b/src/node/retry.rs @@ -282,7 +282,7 @@ impl Node { // evicts if the relay has nothing, otherwise leaves it. Cheap // (one Filter fetch with 2s timeout) and bounded by the retry // backoff cadence. - if let Some(bootstrap) = self.nostr_rendezvous.engine_arc() { + if let Some(bootstrap) = self.supervisor.nostr_rendezvous.engine_arc() { let _ = bootstrap .refetch_advert_for_stale_check(&peer_config.npub) .await; @@ -318,7 +318,7 @@ impl Node { // entry expires. Force a re-fetch so the next retry tick // picks up fresh endpoints. if matches!(e, NodeError::NoTransportForType(_)) - && let Some(bootstrap) = self.nostr_rendezvous.engine_arc() + && let Some(bootstrap) = self.supervisor.nostr_rendezvous.engine_arc() { let npub = peer_config.npub.clone(); tokio::spawn(async move { diff --git a/src/node/tests/bootstrap.rs b/src/node/tests/bootstrap.rs index aae96bb..6eb31bd 100644 --- a/src/node/tests/bootstrap.rs +++ b/src/node/tests/bootstrap.rs @@ -24,17 +24,17 @@ async fn test_adopted_udp_traversal_completes_handshake() { let (packet_tx_a, packet_rx_a) = packet_channel(64); let (packet_tx_b, packet_rx_b) = packet_channel(64); - node_a.packet_tx = Some(packet_tx_a.clone()); + node_a.supervisor.packet_tx = Some(packet_tx_a.clone()); node_a.packet_rx = Some(packet_rx_a); - node_a.state = NodeState::Running; + node_a.supervisor.state = NodeState::Running; let mut transport_b = UdpTransport::new(transport_id_b, None, udp_config, packet_tx_b.clone()); transport_b.start_async().await.unwrap(); let addr_b = transport_b.local_addr().unwrap(); - node_b.packet_tx = Some(packet_tx_b.clone()); + node_b.supervisor.packet_tx = Some(packet_tx_b.clone()); node_b.packet_rx = Some(packet_rx_b); - node_b.state = NodeState::Running; + node_b.supervisor.state = NodeState::Running; node_b .transports .insert(transport_id_b, TransportHandle::Udp(transport_b)); @@ -107,9 +107,9 @@ async fn test_adopted_udp_traversal_completes_handshake() { async fn test_failed_adopted_traversal_cleans_up_transport() { let mut node = make_node(); let (packet_tx, packet_rx) = packet_channel(64); - node.packet_tx = Some(packet_tx); + node.supervisor.packet_tx = Some(packet_tx); node.packet_rx = Some(packet_rx); - node.state = NodeState::Running; + node.supervisor.state = NodeState::Running; node.index_allocator = IndexAllocator::with_max_attempts(0); let peer = make_node(); @@ -137,9 +137,9 @@ async fn test_failed_adopted_traversal_cleans_up_transport() { async fn test_adopted_traversal_skips_already_connected_peer() { let mut node = make_node(); let (packet_tx, packet_rx) = packet_channel(64); - node.packet_tx = Some(packet_tx); + node.supervisor.packet_tx = Some(packet_tx); node.packet_rx = Some(packet_rx); - node.state = NodeState::Running; + node.supervisor.state = NodeState::Running; let transport_id = TransportId::new(1); let link_id = LinkId::new(1); @@ -196,17 +196,17 @@ async fn test_third_peer_can_handshake_via_adopted_transport_socket() { let (packet_tx_b, packet_rx_b) = packet_channel(64); let (packet_tx_c, packet_rx_c) = packet_channel(64); - node_a.packet_tx = Some(packet_tx_a.clone()); + node_a.supervisor.packet_tx = Some(packet_tx_a.clone()); node_a.packet_rx = Some(packet_rx_a); - node_a.state = NodeState::Running; + node_a.supervisor.state = NodeState::Running; - node_b.packet_tx = Some(packet_tx_b.clone()); + node_b.supervisor.packet_tx = Some(packet_tx_b.clone()); node_b.packet_rx = Some(packet_rx_b); - node_b.state = NodeState::Running; + node_b.supervisor.state = NodeState::Running; - node_c.packet_tx = Some(packet_tx_c.clone()); + node_c.supervisor.packet_tx = Some(packet_tx_c.clone()); node_c.packet_rx = Some(packet_rx_c); - node_c.state = NodeState::Running; + node_c.supervisor.state = NodeState::Running; let mut transport_a = UdpTransport::new(transport_id_a, None, udp_config.clone(), packet_tx_a); transport_a.start_async().await.unwrap(); @@ -341,9 +341,9 @@ async fn test_adopted_udp_inherits_mtu_from_single_primary_config() { let mut node = make_node_with(config); let (packet_tx, packet_rx) = packet_channel(64); - node.packet_tx = Some(packet_tx); + node.supervisor.packet_tx = Some(packet_tx); node.packet_rx = Some(packet_rx); - node.state = NodeState::Running; + node.supervisor.state = NodeState::Running; let peer = make_node(); let adopted_socket = std::net::UdpSocket::bind("127.0.0.1:0").unwrap(); @@ -391,9 +391,9 @@ async fn test_adopted_udp_inherits_mtu_from_named_primary_config() { let mut node = make_node_with(config); let (packet_tx, packet_rx) = packet_channel(64); - node.packet_tx = Some(packet_tx); + node.supervisor.packet_tx = Some(packet_tx); node.packet_rx = Some(packet_rx); - node.state = NodeState::Running; + node.supervisor.state = NodeState::Running; let peer = make_node(); let adopted_socket = std::net::UdpSocket::bind("127.0.0.1:0").unwrap(); diff --git a/src/node/tests/discovery.rs b/src/node/tests/discovery.rs index a743eb3..dd1f692 100644 --- a/src/node/tests/discovery.rs +++ b/src/node/tests/discovery.rs @@ -1217,7 +1217,7 @@ async fn test_check_pending_lookups_default_sequence_unreachable() { // Inject a TUN sender so `send_icmpv6_dest_unreachable` is observable. let (tun_tx, tun_rx) = mpsc::channel::>(); - node.tun_tx = Some(tun_tx); + node.supervisor.tun_tx = Some(tun_tx); // Build a target identity (the unreachable destination). let target_identity = Identity::generate(); diff --git a/src/node/tests/handshake.rs b/src/node/tests/handshake.rs index 3a729d2..82f8bbb 100644 --- a/src/node/tests/handshake.rs +++ b/src/node/tests/handshake.rs @@ -310,8 +310,8 @@ async fn test_run_rx_loop_handshake() { node_b.packet_rx = Some(packet_rx_b); // Set node state to Running (transports need to be operational) - node_a.state = NodeState::Running; - node_b.state = NodeState::Running; + node_a.supervisor.state = NodeState::Running; + node_b.supervisor.state = NodeState::Running; // === Phase 1: Node A initiates handshake to Node B === diff --git a/src/node/tests/mod.rs b/src/node/tests/mod.rs index 2bc5d4f..953b6b7 100644 --- a/src/node/tests/mod.rs +++ b/src/node/tests/mod.rs @@ -29,6 +29,25 @@ pub(super) fn make_node() -> Node { make_node_with(Config::new()) } +/// 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 +/// 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 +/// configured child and it comes up (`Full`). Mirrors the udp config in +/// `test_node_start_does_not_wait_for_nostr_relay_startup`. +pub(super) fn make_healthy_node() -> Node { + let mut config = Config::new(); + config.transports.udp = crate::config::TransportInstances::Single(crate::config::UdpConfig { + bind_addr: Some("127.0.0.1:0".to_string()), + ..Default::default() + }); + config.dns.enabled = false; + make_node_with(config) +} + /// Build a test node from an explicit `Config`. Immutable state lives solely in /// the shared `NodeContext`, built once at construction — there is no /// post-construction field to poke, so set limits/config on the `Config` here. diff --git a/src/node/tests/session.rs b/src/node/tests/session.rs index 1babd75..85e1f7d 100644 --- a/src/node/tests/session.rs +++ b/src/node/tests/session.rs @@ -654,7 +654,7 @@ async fn test_session_100_nodes() { let mut tun_receivers: Vec>> = Vec::with_capacity(NUM_NODES); for tn in nodes.iter_mut() { let (tx, rx) = mpsc::channel(); - tn.node.tun_tx = Some(tx); + tn.node.supervisor.tun_tx = Some(tx); tun_receivers.push(rx); } @@ -1070,7 +1070,7 @@ async fn test_tun_outbound_established_session() { // Install TUN receiver on Node 1 let (tun_tx, tun_rx) = std::sync::mpsc::channel(); - nodes[1].node.tun_tx = Some(tun_tx); + nodes[1].node.supervisor.tun_tx = Some(tun_tx); // Build and inject an IPv6 packet let test_payload = b"data-plane-test-12345"; @@ -1114,7 +1114,7 @@ async fn test_tun_outbound_triggers_session_initiation() { // Install TUN receiver on Node 1 let (tun_tx, tun_rx) = std::sync::mpsc::channel(); - nodes[1].node.tun_tx = Some(tun_tx); + nodes[1].node.supervisor.tun_tx = Some(tun_tx); // Build and inject an IPv6 packet (identity cache populated at peer promotion) let test_payload = b"trigger-session-test"; @@ -1167,7 +1167,7 @@ async fn test_tun_outbound_unknown_destination() { // Install TUN receiver on Node 0 (for ICMPv6 response) let (tun_tx, tun_rx) = std::sync::mpsc::channel(); - nodes[0].node.tun_tx = Some(tun_tx); + nodes[0].node.supervisor.tun_tx = Some(tun_tx); let src_fips = crate::FipsAddress::from_node_addr(nodes[0].node.node_addr()); @@ -1219,7 +1219,7 @@ async fn test_tun_outbound_3node_forwarded() { // Install TUN receiver on Node 2 let (tun_tx, tun_rx) = std::sync::mpsc::channel(); - nodes[2].node.tun_tx = Some(tun_tx); + nodes[2].node.supervisor.tun_tx = Some(tun_tx); // Build and inject an IPv6 packet (triggers session initiation to Node 2) let test_payload = b"forwarded-data-plane"; @@ -1264,7 +1264,7 @@ async fn test_tun_outbound_pending_queue_flush() { // Install TUN receiver on Node 1 let (tun_tx, tun_rx) = std::sync::mpsc::channel(); - nodes[1].node.tun_tx = Some(tun_tx); + nodes[1].node.supervisor.tun_tx = Some(tun_tx); // Send 5 packets before any session exists let mut packets = Vec::new(); @@ -1915,7 +1915,7 @@ async fn test_tun_outbound_path_mtu_generates_ptb() { // Install TUN receiver on source node to capture ICMPv6 PTB let (tun_tx, tun_rx) = std::sync::mpsc::channel(); - nodes[0].node.tun_tx = Some(tun_tx); + nodes[0].node.supervisor.tun_tx = Some(tun_tx); // Build an IPv6 packet that fits local MTU but exceeds path MTU let reduced_ipv6_mtu = crate::upper::icmp::effective_ipv6_mtu(reduced_mtu) as usize; @@ -1972,7 +1972,7 @@ async fn test_tun_outbound_path_mtu_generates_ptb() { // Verify a packet that fits within path MTU passes through (no PTB) let (tun_tx2, tun_rx2) = std::sync::mpsc::channel(); - nodes[0].node.tun_tx = Some(tun_tx2); + nodes[0].node.supervisor.tun_tx = Some(tun_tx2); let fitting_payload = vec![0u8; reduced_ipv6_mtu - 41]; // fits within path MTU let fitting_packet = build_ipv6_packet(&src_fips, &dst_fips, &fitting_payload); assert!(fitting_packet.len() <= reduced_ipv6_mtu); @@ -2111,7 +2111,7 @@ async fn test_multihop_pmtud_heterogeneous_mtu() { // should check PathMtuState and generate ICMPv6 PTB on TUN instead // of forwarding. let (tun_tx2, tun_rx2) = std::sync::mpsc::channel(); - nodes[0].node.tun_tx = Some(tun_tx2); + nodes[0].node.supervisor.tun_tx = Some(tun_tx2); nodes[0].node.handle_tun_outbound(ipv6_packet.clone()).await; @@ -2155,7 +2155,7 @@ async fn test_multihop_pmtud_heterogeneous_mtu() { // Verify a fitting packet still passes through without PTB let (tun_tx3, tun_rx3) = std::sync::mpsc::channel(); - nodes[0].node.tun_tx = Some(tun_tx3); + nodes[0].node.supervisor.tun_tx = Some(tun_tx3); let fitting_payload = vec![0xCDu8; 600 - 40]; // 600-byte IPv6 packet, well within 694 let fitting_packet = build_ipv6_packet(&src_fips, &dst_fips, &fitting_payload); diff --git a/src/node/tests/unit.rs b/src/node/tests/unit.rs index 2d0d256..bd49d9f 100644 --- a/src/node/tests/unit.rs +++ b/src/node/tests/unit.rs @@ -56,7 +56,7 @@ async fn test_nat_bootstrap_failure_falls_back_to_direct_udp_address() { let peer_identity = Identity::generate(); let mut node = make_node(); let (packet_tx, packet_rx) = packet_channel(64); - node.packet_tx = Some(packet_tx.clone()); + node.supervisor.packet_tx = Some(packet_tx.clone()); node.packet_rx = Some(packet_rx); let transport_id = TransportId::new(1); @@ -102,7 +102,7 @@ async fn test_try_peer_addresses_races_all_concrete_udp_candidates() { let peer_identity = Identity::generate(); let mut node = make_node(); let (packet_tx, packet_rx) = packet_channel(64); - node.packet_tx = Some(packet_tx.clone()); + node.supervisor.packet_tx = Some(packet_tx.clone()); node.packet_rx = Some(packet_rx); let transport_id = TransportId::new(1); @@ -151,13 +151,16 @@ async fn test_try_peer_addresses_races_all_concrete_udp_candidates() { #[tokio::test] async fn test_node_state_transitions() { - let mut node = make_node(); + // A transport-less node now resolves to `Failed` on start (design doc + // §9.1), so exercise state transitions with a genuinely healthy node. + let mut node = make_healthy_node(); assert!(!node.is_running()); assert!(node.state().can_start()); node.start().await.unwrap(); assert!(node.is_running()); + assert_eq!(node.state(), NodeState::Running); assert!(!node.state().can_start()); node.stop().await.unwrap(); @@ -165,6 +168,58 @@ async fn test_node_state_transitions() { assert_eq!(node.state(), NodeState::Stopped); } +#[tokio::test] +async fn test_transportless_start_fails_and_publishes_failed() { + // The intended behavioral change (design doc §9.1): 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). + let mut node = make_node(); + assert!(node.state().can_start()); + + let result = node.start().await; + assert!(matches!(result, Err(NodeError::NoOperationalTransports))); + assert_eq!(node.state(), NodeState::Failed); + assert!(!node.is_running()); + assert!(!node.state().is_operational()); + assert!(!node.state().can_start()); +} + +#[tokio::test] +async fn test_drain_publishes_draining_state() { + let mut node = make_healthy_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_healthy_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(); @@ -196,7 +251,7 @@ async fn test_node_start_does_not_wait_for_nostr_relay_startup() { #[tokio::test] async fn test_node_double_start() { - let mut node = make_node(); + let mut node = make_healthy_node(); node.start().await.unwrap(); let result = node.start().await; @@ -543,7 +598,7 @@ async fn test_node_rx_loop_requires_start() { #[tokio::test] async fn test_node_rx_loop_takes_channel() { - let mut node = make_node(); + let mut node = make_healthy_node(); node.start().await.unwrap(); // packet_rx should be available after start @@ -1053,7 +1108,7 @@ async fn update_peers_races_new_alternative_without_dropping_active_peer() { config.peers = vec![old_peer.clone()]; let mut node = make_node_with(config); let (packet_tx, packet_rx) = packet_channel(64); - node.packet_tx = Some(packet_tx.clone()); + node.supervisor.packet_tx = Some(packet_tx.clone()); node.packet_rx = Some(packet_rx); let transport_id = TransportId::new(1); @@ -1133,7 +1188,9 @@ async fn test_nostr_traversal_failure_skips_connected_peer() { peer_config: crate::config::PeerConfig::new(peer_identity.npub(), "udp", "127.0.0.1:9"), reason: "stale traversal failure".to_string(), }); - node.nostr_rendezvous.set_engine(bootstrap.clone()); + node.supervisor + .nostr_rendezvous + .set_engine(bootstrap.clone()); node.poll_nostr_rendezvous().await; @@ -1173,7 +1230,9 @@ async fn test_nostr_traversal_established_skips_connected_peer() { socket, ), }); - node.nostr_rendezvous.set_engine(bootstrap.clone()); + node.supervisor + .nostr_rendezvous + .set_engine(bootstrap.clone()); node.poll_nostr_rendezvous().await; @@ -1427,7 +1486,7 @@ async fn test_transport_mtu_returns_min_across_operational() { // iteration order. This is the core ISSUE-2026-0011 regression test. let mut node = make_node(); let (packet_tx, packet_rx) = packet_channel(64); - node.packet_tx = Some(packet_tx); + node.supervisor.packet_tx = Some(packet_tx); node.packet_rx = Some(packet_rx); let udp1 = make_udp_transport_with_mtu(1, 1497).await; @@ -1464,7 +1523,7 @@ async fn test_transport_mtu_min_with_single_operational() { // operational. let mut node = make_node(); let (packet_tx, packet_rx) = packet_channel(64); - node.packet_tx = Some(packet_tx); + node.supervisor.packet_tx = Some(packet_tx); node.packet_rx = Some(packet_rx); let udp = make_udp_transport_with_mtu(1, 1452).await; @@ -1487,7 +1546,7 @@ async fn test_transport_mtu_min_with_single_operational() { async fn test_seed_path_mtu_inserts_when_empty() { let mut node = make_node(); let (packet_tx, packet_rx) = packet_channel(64); - node.packet_tx = Some(packet_tx); + node.supervisor.packet_tx = Some(packet_tx); node.packet_rx = Some(packet_rx); let udp = make_udp_transport_with_mtu(1, 1452).await; @@ -1520,7 +1579,7 @@ async fn test_seed_path_mtu_inserts_when_empty() { async fn test_seed_path_mtu_keeps_tighter_existing_value() { let mut node = make_node(); let (packet_tx, packet_rx) = packet_channel(64); - node.packet_tx = Some(packet_tx); + node.supervisor.packet_tx = Some(packet_tx); node.packet_rx = Some(packet_rx); let udp = make_udp_transport_with_mtu(1, 1452).await; @@ -1560,7 +1619,7 @@ async fn test_seed_path_mtu_keeps_tighter_existing_value() { async fn test_seed_path_mtu_tightens_looser_existing_value() { let mut node = make_node(); let (packet_tx, packet_rx) = packet_channel(64); - node.packet_tx = Some(packet_tx); + node.supervisor.packet_tx = Some(packet_tx); node.packet_rx = Some(packet_rx); let udp = make_udp_transport_with_mtu(1, 1280).await; @@ -1730,7 +1789,9 @@ async fn poll_nostr_rendezvous_established_gated_at_capacity() { socket, ), }); - node.nostr_rendezvous.set_engine(bootstrap.clone()); + node.supervisor + .nostr_rendezvous + .set_engine(bootstrap.clone()); let before_peers = node.peer_count(); let before_links = node.link_count(); diff --git a/testing/dns-resolver/test.sh b/testing/dns-resolver/test.sh index 218f978..f5df0b0 100755 --- a/testing/dns-resolver/test.sh +++ b/testing/dns-resolver/test.sh @@ -661,11 +661,18 @@ DOCKERFILE # Write a minimal fips.yaml that exercises the new defaults. # tun.enabled: true so the daemon creates fips0 itself; identity # persistent so /etc/fips/fips.pub gives us a stable npub to query. + # A UDP transport is configured so at least one transport comes up: + # a node with zero operational transports is Failed and refuses to + # start. This test exercises the .fips DNS responder, not mesh + # connectivity, so any bound transport suffices. docker exec "$name" bash -c 'cat > /etc/fips/fips.yaml <