diff --git a/src/mdns/mod.rs b/src/mdns/mod.rs index 4da3ebb..bfb3605 100644 --- a/src/mdns/mod.rs +++ b/src/mdns/mod.rs @@ -142,6 +142,11 @@ pub struct LanRendezvous { } impl LanRendezvous { + /// Whether the mDNS event-pump task has exited (runtime liveness). + pub fn is_finished(&self) -> bool { + self.event_pump.is_finished() + } + /// Start the mDNS responder and browser. /// /// `advertised_port` is the UDP port the operational UDP transport diff --git a/src/node/dataplane/rx_loop.rs b/src/node/dataplane/rx_loop.rs index 0582b18..4ebf5a5 100644 --- a/src/node/dataplane/rx_loop.rs +++ b/src/node/dataplane/rx_loop.rs @@ -97,6 +97,18 @@ impl Node { } }; + // Take the runtime child-liveness receiver, or a dummy channel (when the + // node was seeded straight into Running without a start()). Holding the + // dummy sender in the guard keeps the channel open. Same pattern as TUN + // outbound / DNS identity. + let (mut child_exit_rx, _child_exit_guard) = match self.child_exit_rx.take() { + Some(rx) => (rx, None), + None => { + let (tx, rx) = tokio::sync::mpsc::channel(1); + (rx, Some(tx)) + } + }; + let mut tick = tokio::time::interval(Duration::from_secs(self.config().node.tick_interval_secs)); @@ -253,6 +265,27 @@ impl Node { } } } + // Runtime child-liveness. Placed AFTER `packet_rx` so the hot + // inbound path keeps its `biased` priority. A directly-observable + // child (TUN threads, DNS/mDNS/Nostr) exited on its own; feed the + // FSM, which republishes health (Degraded here — a Running node + // always has ≥1 transport up). `on_child_exited` only ever emits + // `PublishState`; other variants are ignored defensively. + maybe_child = child_exit_rx.recv() => { + if let Some(child) = maybe_child { + let actions = self + .supervisor + .fsm + .step(crate::node::lifecycle::supervisor::Event::ChildExited { child }); + for action in actions { + if let crate::node::lifecycle::supervisor::Action::PublishState(ns) = + action + { + self.supervisor.state = ns; + } + } + } + } Some(ipv6_packet) = tun_outbound_rx.recv() => { self.handle_tun_outbound(ipv6_packet).await; let mut drained = 0; diff --git a/src/node/lifecycle/mod.rs b/src/node/lifecycle/mod.rs index 26747f0..48704d8 100644 --- a/src/node/lifecycle/mod.rs +++ b/src/node/lifecycle/mod.rs @@ -1245,6 +1245,16 @@ impl Node { self.supervisor.packet_tx = Some(packet_tx.clone()); self.packet_rx = Some(packet_rx); + // Runtime child-liveness channel (design doc §6). Created before any + // child is spawned so each directly-observable child (TUN threads, the + // DNS task, and the mDNS/Nostr liveness monitor) can clone the sender + // and self-report its `Child` on exit. The sender stored on `self` is + // the keep-alive; the rx_loop takes only the receiver, so the channel + // never closes spuriously while the node runs. + let (child_exit_tx, child_exit_rx) = tokio::sync::mpsc::channel(16); + self.child_exit_tx = Some(child_exit_tx); + self.child_exit_rx = Some(child_exit_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. @@ -1517,9 +1527,16 @@ impl Node { let (writer, tun_tx) = device.create_writer(max_mss, self.path_mtu_lookup.clone())?; - // Spawn writer thread + // Spawn writer thread. On exit it self-reports + // `Child::Tun` (sync context → `blocking_send`); TUN + // is one compound child, so both threads reporting is + // fine (the FSM de-dups via `up.remove`). + let writer_child_tx = self.child_exit_tx.clone(); let writer_handle = thread::spawn(move || { writer.run(); + if let Some(tx) = &writer_child_tx { + let _ = tx.blocking_send(Child::Tun); + } }); // Clone tun_tx for the reader @@ -1530,9 +1547,13 @@ impl Node { let (outbound_tx, outbound_rx) = tokio::sync::mpsc::channel(tun_channel_size); - // Spawn reader thread + // Spawn reader thread. Like the writer, it + // self-reports `Child::Tun` on exit (sync context → + // `blocking_send`). Exactly one cfg variant compiles, + // so the single clone is moved into that closure. let transport_mtu = self.transport_mtu(); let path_mtu_lookup = self.path_mtu_lookup.clone(); + let reader_child_tx = self.child_exit_tx.clone(); #[cfg(target_os = "macos")] let reader_handle = thread::spawn(move || { run_tun_reader( @@ -1545,6 +1566,9 @@ impl Node { path_mtu_lookup, shutdown_read_fd, ); + if let Some(tx) = &reader_child_tx { + let _ = tx.blocking_send(Child::Tun); + } }); #[cfg(not(target_os = "macos"))] let reader_handle = thread::spawn(move || { @@ -1557,6 +1581,9 @@ impl Node { transport_mtu, path_mtu_lookup, ); + if let Some(tx) = &reader_child_tx { + let _ = tx.blocking_send(Child::Tun); + } }); self.tun_state = TunState::Active; @@ -1630,14 +1657,25 @@ impl Node { mesh_ifindex = ?mesh_ifindex, "DNS responder started for .fips domain (auto-reload enabled)" ); - let handle = - tokio::spawn(crate::upper::dns::run_dns_responder( + // Self-report on exit so the supervisor FSM + // routes health when the DNS task dies at + // runtime. On a deliberate stop the task is + // `.abort()`ed before this send; even if it + // fired, the FSM ignores it outside `Running`. + let dns_child_tx = self.child_exit_tx.clone(); + let handle = tokio::spawn(async move { + crate::upper::dns::run_dns_responder( socket, identity_tx, dns_ttl, reloader, mesh_ifindex, - )); + ) + .await; + if let Some(tx) = dns_child_tx { + let _ = tx.send(Child::Dns).await; + } + }); self.supervisor.dns_identity_rx = Some(identity_rx); self.supervisor.dns_task = Some(handle); Event::SubstrateUp { child } @@ -1709,6 +1747,34 @@ impl Node { _ => {} } + // Runtime liveness monitor for the two poll-observable children (mDNS + + // Nostr). Unlike the TUN threads and DNS task, these expose no exit hook, + // so one task polls their `is_finished` accessors every 2s and reports + // `Child::Mdns` / `Child::Nostr` on exit. It self-terminates once both + // have been reported (or were never present), and is only armed when at + // least one of them is actually running. + let mon_lan = self.supervisor.lan_rendezvous.clone(); + let mon_nostr = self.supervisor.nostr_rendezvous.engine_arc(); + if let Some(mon_tx) = self.child_exit_tx.clone() + && (mon_lan.is_some() || mon_nostr.is_some()) + { + tokio::spawn(async move { + let mut mdns_reported = mon_lan.is_none(); + let mut nostr_reported = mon_nostr.is_none(); + while !(mdns_reported && nostr_reported) { + tokio::time::sleep(std::time::Duration::from_secs(2)).await; + if !mdns_reported && mon_lan.as_ref().is_some_and(|l| l.is_finished()) { + let _ = mon_tx.send(Child::Mdns).await; + mdns_reported = true; + } + if !nostr_reported && mon_nostr.as_ref().is_some_and(|n| n.is_finished()) { + let _ = mon_tx.send(Child::Nostr).await; + nostr_reported = true; + } + } + }); + } + info!("Node started:"); info!(" state: {}", self.supervisor.state); info!(" transports: {}", self.transports.len()); diff --git a/src/node/lifecycle/supervisor.rs b/src/node/lifecycle/supervisor.rs index 7e8b9f1..3151328 100644 --- a/src/node/lifecycle/supervisor.rs +++ b/src/node/lifecycle/supervisor.rs @@ -170,8 +170,6 @@ pub(crate) enum Event { /// `Failed`, an optional child out → `Degraded`), but at runtime `Failed` is /// a published health signal, not a teardown — the driver keeps serving. No /// restart (the FSM has no restart action). Inert outside `Running`. - #[allow(dead_code)] - // constructed by the runtime exit-producer wiring in the following commit ChildExited { /// The child whose task/thread exited. child: Child, diff --git a/src/node/mod.rs b/src/node/mod.rs index 372f7e1..4f7ecd6 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -351,6 +351,18 @@ pub struct Node { /// Packet receiver (for event loop). packet_rx: Option, + // === Child Exit Channel === + /// Sender half of the runtime child-liveness channel. Cloned into each + /// directly-observable child (the TUN reader/writer threads, the DNS task, + /// and the mDNS/Nostr liveness monitor) so a child self-reports its + /// [`Child`](crate::node::lifecycle::supervisor::Child) when it exits. Held + /// on `self` for the rx_loop's lifetime as the keep-alive sender so the + /// receiver never observes a spuriously-closed channel. + child_exit_tx: Option>, + /// Receiver half of the runtime child-liveness channel, `take()`-en by the + /// rx_loop select arm that feeds `Event::ChildExited` to the supervisor FSM. + child_exit_rx: Option>, + // === Connections (Handshake Phase) === /// Pending connections (handshake in progress). /// Indexed by LinkId since we don't know the peer's identity yet. @@ -625,6 +637,8 @@ impl Node { links: HashMap::new(), addr_to_link: HashMap::new(), packet_rx: None, + child_exit_tx: None, + child_exit_rx: None, connections: HashMap::new(), peers: HashMap::new(), sessions: HashMap::new(), @@ -772,6 +786,8 @@ impl Node { links: HashMap::new(), addr_to_link: HashMap::new(), packet_rx: None, + child_exit_tx: None, + child_exit_rx: None, connections: HashMap::new(), peers: HashMap::new(), sessions: HashMap::new(), diff --git a/src/nostr/runtime.rs b/src/nostr/runtime.rs index e7cc123..355b6f8 100644 --- a/src/nostr/runtime.rs +++ b/src/nostr/runtime.rs @@ -176,6 +176,25 @@ pub struct NostrRendezvous { } impl NostrRendezvous { + /// Whether the primary Nostr connection task has exited (runtime liveness). + /// + /// "Nostr exited" is defined as the primary `connect_task` having finished. + /// It is `Some` for the engine's whole running life (installed in `start`); + /// `shutdown` takes it, leaving `None` — a taken handle means the engine has + /// been shut down, which counts as finished, so a `None` inner maps to + /// `true` (this lets the liveness poll monitor terminate after a stop rather + /// than spinning forever). `connect_task` is a `tokio::sync::Mutex`, so this + /// sync accessor uses the non-blocking `try_lock`: a momentarily-contended + /// lock (only start/stop hold it, briefly) reports "not finished", the safe + /// direction — the 2s liveness poll re-checks next tick and never spuriously + /// degrades a healthy node. + pub fn is_finished(&self) -> bool { + self.connect_task + .try_lock() + .map(|g| g.as_ref().is_none_or(|h| h.is_finished())) + .unwrap_or(false) + } + pub async fn start( identity: &crate::Identity, config: NostrRendezvousConfig,