From 6c5fd3f4b080d38adecfa1721e10abf02e9cd89f Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Sun, 12 Jul 2026 21:03:46 +0000 Subject: [PATCH] node: extract lifecycle supervisor FSM, migrate substrate fields (behavior-neutral) Introduce a sans-IO lifecycle supervisor: a synchronous step(event) -> [action] state machine (SupervisorFsm) that authors the substrate-child spawn and teardown order, plus an owner struct (Supervisor) holding the substrate runtime fields and embedding the FSM. The substrate-lifecycle fields leave Node's flat list into the owner - state, packet_tx, the TUN reader/writer handles + shutdown fd + channels, the DNS task + identity channel, the Nostr/LAN rendezvous drivers, and the encrypt/decrypt worker pools; the dataplane keeps packet_rx. start()/stop() become the driver executing the FSM's SpawnChild/StopChild actions: same children, same order, same warn/debug-and-continue on optional failures, same logs, same NodeState transitions. Behavior- and wire-neutral. The only determinism change is that transports now tear down in ascending-id order, previously nondeterministic HashMap iteration. The bounded Draining phase and the Running{Full|Degraded} health split land as separate follow-on commits. --- src/node/dataplane/encrypted.rs | 6 +- src/node/dataplane/rx_loop.rs | 6 +- src/node/handlers/session.rs | 8 +- src/node/{lifecycle.rs => lifecycle/mod.rs} | 914 ++++++++++++-------- src/node/lifecycle/supervisor.rs | 658 ++++++++++++++ src/node/mod.rs | 115 +-- src/node/retry.rs | 4 +- src/node/tests/bootstrap.rs | 36 +- src/node/tests/discovery.rs | 2 +- src/node/tests/handshake.rs | 4 +- src/node/tests/session.rs | 20 +- src/node/tests/unit.rs | 28 +- 12 files changed, 1293 insertions(+), 508 deletions(-) rename src/node/{lifecycle.rs => lifecycle/mod.rs} (75%) create mode 100644 src/node/lifecycle/supervisor.rs diff --git a/src/node/dataplane/encrypted.rs b/src/node/dataplane/encrypted.rs index 4a5e544..2d69adc 100644 --- a/src/node/dataplane/encrypted.rs +++ b/src/node/dataplane/encrypted.rs @@ -172,7 +172,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 { @@ -451,7 +451,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) = { @@ -492,7 +492,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 0eca9d8..e056110 100644 --- a/src/node/dataplane/rx_loop.rs +++ b/src/node/dataplane/rx_loop.rs @@ -47,7 +47,7 @@ impl Node { // 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); @@ -57,7 +57,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); @@ -320,9 +320,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 de447f6..2f67e39 100644 --- a/src/node/handlers/session.rs +++ b/src/node/handlers/session.rs @@ -374,7 +374,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"); } @@ -1832,7 +1832,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); }; @@ -2448,7 +2448,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); } @@ -2483,7 +2483,7 @@ impl Node { // causes a PMTUD blackhole when both src and ICMP-src are local. 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 75% rename from src/node/lifecycle.rs rename to src/node/lifecycle/mod.rs index 9547396..2053dbc 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 { @@ -354,7 +358,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()?; @@ -677,7 +681,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; }; @@ -902,7 +906,7 @@ impl Node { /// The handshake itself is the authentication — a spoofed mDNS advert /// with someone else's npub fails the IK 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; @@ -1052,51 +1056,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) @@ -1106,273 +1100,388 @@ 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, + }); + + // 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; } - } - // 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"); - } + // 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; } - } - // 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; + 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()); - // 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"); - } - } - } - - // 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); + 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 } + } + } + } + }; + + self.supervisor.fsm.step(feedback); } - 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; + } + + self.supervisor.state = NodeState::Running; info!("Node started:"); - info!(" state: {}", self.state); + info!(" state: {}", self.supervisor.state); info!(" transports: {}", self.transports.len()); info!(" connections: {}", self.connections.len()); Ok(()) @@ -1459,96 +1568,164 @@ 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). Worker pools are deliberately + // absent from the up-set: today's stop() never tears them down. + let mut up: Vec = Vec::new(); + if self.supervisor.dns_task.is_some() { + up.push(Child::Dns); } - - // Send disconnect notifications to all active peers before closing transports - self.send_disconnect_to_all_peers(DisconnectReason::Shutdown) - .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"); + if self.supervisor.nostr_rendezvous.engine().is_some() { + up.push(Child::Nostr); } - - // 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; + 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); + } + self.supervisor.fsm = SupervisorFsm::running_with(up); + let actions = self.supervisor.fsm.step(Event::Stop); - // 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" - ); + // Execute each StopChild as today's verbatim block. Two driver seams + // are woven in at their current positions: the shutdown-disconnect + // fan-out (after Dns, before Nostr), and dropping the packet channels + // (after all transports, before TUN). + let mut disconnect_done = false; + 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 packet channels - self.packet_tx.take(); - self.packet_rx.take(); + // Drop the tun_tx to signal the writer to stop + self.supervisor.tun_tx.take(); - // Shutdown TUN interface - if let Some(name) = self.tun_name.take() { - info!(name = %name, "Shutting down TUN interface"); + // 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"); + } - // Drop the tun_tx to signal the writer to stop - self.tun_tx.take(); + // 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); + } + } - // 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"); - } + // 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(); + } - // 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); + 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. } } - // 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.supervisor.fsm.step(Event::ChildStopped { child }); } - self.state = NodeState::Stopped; - info!(state = %self.state, "Node stopped"); + // 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(); + } + + self.supervisor.state = NodeState::Stopped; + info!(state = %self.supervisor.state, "Node stopped"); Ok(()) } @@ -1644,6 +1821,7 @@ impl Node { continue; } if self + .supervisor .nostr_rendezvous .request_nostr_bootstrap(peer_config) .await @@ -1968,7 +2146,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 @@ -1976,10 +2154,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(); @@ -2002,7 +2180,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 { @@ -2167,7 +2345,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 } @@ -2286,6 +2465,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, @@ -2360,7 +2540,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; @@ -2475,11 +2655,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(), @@ -2556,7 +2740,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()); @@ -2564,7 +2749,8 @@ impl Node { .initiate_connection(transport_id, remote_addr.clone(), 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..00f96f2 --- /dev/null +++ b/src/node/lifecycle/supervisor.rs @@ -0,0 +1,658 @@ +//! 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*. The +//! `self.state` field writes stay verbatim in the driver at their current +//! positions, so the published `NodeState` transitions are byte-for-byte +//! unchanged. FSM-owned published state (`PublishState`) is introduced with +//! the `Draining`/`Degraded` additions, which need it. +//! +//! The `Draining` phase and the `Running{Full|Degraded}` health split (design +//! doc §6/§9.1) land as the two subsequent, separately-flagged commits and +//! extend the [`SupState`], [`Event`], and [`Action`] enums below. + +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`. +/// +/// Only the events the behavior-neutral rewrite needs are present; `Tick`, +/// `ChildExited`, and `DrainDeadlineElapsed` (design doc §6) arrive with the +/// `Draining` / `Degraded` commits. +#[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 teardown. Valid from `Running`. + Stop, + /// A child the driver was asked to stop has finished stopping. + ChildStopped { + /// The child that has been torn down. + child: Child, + }, +} + +/// An effect the driver must perform. The core never performs I/O itself. +/// +/// `PublishState` (design doc §6) is intentionally absent from the +/// behavior-neutral rewrite: the driver keeps its verbatim `self.state` writes, +/// so no published-state action is needed until `Draining`/`Degraded`. +#[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), + /// Tear down this child (the driver performs the stop / join I/O and + /// reports `ChildStopped`). + StopChild(Child), +} + +/// Internal supervisor state (design doc §6). Richer than the published +/// [`NodeState`](crate::node::NodeState): `Starting`/`Stopping` carry the set of +/// children still resolving. `Draining{deadline}`, `Running{Full|Degraded}`, and +/// `Failed{reason}` are added by the later flagged commits. +#[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; node operational. + Running, + /// 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, +} + +impl SupervisorFsm { + /// A fresh supervisor in `Created`. + pub(crate) fn new() -> Self { + Self { + state: SupState::Created, + up: 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, + up: up.into_iter().collect(), + } + } + + /// Current internal state (for the driver's bookkeeping and for tests). + #[cfg(test)] + pub(crate) fn state(&self) -> &SupState { + &self.state + } + + /// 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::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(); + + // A node with no children at all still reaches `Running` (today: even + // zero started transports proceeds to `Running`). + if order.is_empty() { + self.state = SupState::Running; + return Vec::new(); + } + + 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 { + if let SupState::Starting { pending } = &mut self.state { + pending.remove(&child); + self.up.insert(child); + if pending.is_empty() { + self.state = SupState::Running; + } + } + Vec::new() + } + + fn on_substrate_failed(&mut self, child: Child) -> Vec { + // Behavior-neutral: warn/continue. The child drains from `pending` and + // does not join the up-set; start still reaches `Running`. + if let SupState::Starting { pending } = &mut self.state { + pending.remove(&child); + if pending.is_empty() { + self.state = SupState::Running; + } + } + Vec::new() + } + + fn on_stop(&mut self) -> Vec { + if !matches!(self.state, SupState::Running) { + return Vec::new(); + } + 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_children_up_reaches_running() { + let mut s = SupervisorFsm::new(); + let spawns = s.step(start_full()); + for a in spawns { + let child = match a { + Action::SpawnChild(c) => c, + _ => panic!("unexpected action"), + }; + assert_eq!(s.step(Event::SubstrateUp { child }), vec![]); + } + assert_eq!(s.state(), &SupState::Running); + } + + #[test] + fn failed_child_still_reaches_running_and_is_not_up() { + // Behavior-neutral: a failed optional child does not block Running and + // is excluded from teardown (never joined the up-set). + 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, + ] { + s.step(Event::SubstrateUp { child }); + } + // mdns fails, tun+dns come up + s.step(Event::SubstrateFailed { child: Child::Mdns }); + s.step(Event::SubstrateUp { child: Child::Tun }); + s.step(Event::SubstrateUp { child: Child::Dns }); + assert_eq!(s.state(), &SupState::Running); + + 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 no_children_reaches_running_immediately() { + 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![]); + assert_eq!(s.state(), &SupState::Running); + } + + #[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 + // (warn/continue). `pending` drains fully, so the node still reaches + // `Running` — as it does today. + 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_eq!(s.state(), &SupState::Running); + + // 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); + // 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); + } +} diff --git a/src/node/mod.rs b/src/node/mod.rs index 03ee49b..9c779ee 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -61,14 +61,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. @@ -271,9 +270,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. @@ -304,8 +307,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, @@ -382,24 +383,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. @@ -444,17 +427,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, @@ -490,20 +462,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 @@ -612,7 +570,7 @@ impl Node { Ok(Self { context, - state: NodeState::Created, + supervisor: lifecycle::supervisor::Supervisor::new(), tree_state, bloom_state, coord_cache, @@ -620,7 +578,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(), @@ -643,14 +600,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(), @@ -669,8 +618,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, @@ -681,10 +628,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), @@ -773,7 +716,7 @@ impl Node { Ok(Self { context, - state: NodeState::Created, + supervisor: lifecycle::supervisor::Supervisor::new(), tree_state, bloom_state, coord_cache, @@ -781,7 +724,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(), @@ -804,14 +746,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(), @@ -827,8 +761,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, @@ -839,10 +771,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), @@ -1210,7 +1138,7 @@ impl Node { /// Get the node state. pub fn state(&self) -> NodeState { - self.state + self.supervisor.state } /// Get the node uptime. @@ -1220,7 +1148,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. @@ -1520,7 +1448,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(), @@ -2269,7 +2197,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; } @@ -2299,7 +2231,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); @@ -2376,7 +2309,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. @@ -2669,7 +2602,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 === @@ -2765,7 +2698,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() @@ -2999,7 +2932,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 38ab60b..69d3b20 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)); @@ -91,9 +91,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(); @@ -121,9 +121,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); @@ -180,17 +180,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(); @@ -300,9 +300,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(); @@ -350,9 +350,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 52be3b3..c2c3084 100644 --- a/src/node/tests/discovery.rs +++ b/src/node/tests/discovery.rs @@ -1095,7 +1095,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 2f2e088..cc999c5 100644 --- a/src/node/tests/handshake.rs +++ b/src/node/tests/handshake.rs @@ -285,8 +285,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/session.rs b/src/node/tests/session.rs index 3915ea2..8d50d1f 100644 --- a/src/node/tests/session.rs +++ b/src/node/tests/session.rs @@ -657,7 +657,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); } @@ -1073,7 +1073,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"; @@ -1117,7 +1117,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"; @@ -1170,7 +1170,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()); @@ -1222,7 +1222,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"; @@ -1267,7 +1267,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(); @@ -1918,7 +1918,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; @@ -1975,7 +1975,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); @@ -2114,7 +2114,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; @@ -2158,7 +2158,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 9fe75ec..f5aab85 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); @@ -1050,7 +1050,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); @@ -1130,7 +1130,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; @@ -1170,7 +1172,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; @@ -1424,7 +1428,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; @@ -1461,7 +1465,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; @@ -1484,7 +1488,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; @@ -1517,7 +1521,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; @@ -1557,7 +1561,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; @@ -1727,7 +1731,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();