Skip to main content

saluki_core/runtime/tree/
mod.rs

1//! Supervision-tree introspection.
2//!
3//! A supervisor knows the shape of the subtree below it -- which children it has, whether each is a worker or a
4//! nested supervisor, how each is configured, and how each has fared -- but that knowledge lives inside the future
5//! that drives the supervisor and is unreachable from anywhere else. This module makes it observable.
6//!
7//! # How it works
8//!
9//! Every [`Supervisor`][super::Supervisor] owns an [`Arc<SupervisorNode>`][SupervisorNode]: a small piece of shared
10//! bookkeeping that the supervisor's run loop writes to as children start, exit, and restart. A parent records a clone
11//! of each child supervisor's node, so the nodes form a graph mirroring the supervision tree, and walking it from any
12//! node yields the subtree rooted there -- which is what [`SupervisionTreeHandle::snapshot`] does. Nodes are shared
13//! rather than copied, so a handle taken before a supervisor starts observes every subsequent generation of it,
14//! including one running on a dedicated runtime on another OS thread.
15//!
16//! Note that this module owns more than an observer's view: [`Roster`] holds the supervisor's own live child roster,
17//! not a mirror of it. That is deliberate -- it is what makes the two impossible to drift apart -- but it means
18//! changes here affect supervision itself and not only what is reported about it.
19
20use std::{
21    sync::{Arc, Mutex},
22    time::{Duration, Instant, SystemTime, UNIX_EPOCH},
23};
24
25use saluki_common::{
26    collections::{FastHashMap, FastHashSet, FastIndexMap},
27    resource_tracking::{ResourceGroupRegistry, ResourceStatsSnapshot},
28};
29use tracing::warn;
30
31use super::{
32    process::Process,
33    restart::{RestartStrategy, RestartType},
34    supervisor::AutoShutdown,
35    ProcessId,
36};
37
38mod api;
39pub use self::api::{SupervisionTreeAPIHandler, SupervisionTreeState};
40
41mod worker;
42pub use self::worker::SupervisionTreeWorker;
43
44mod snapshot;
45pub use self::snapshot::{
46    NodeKind, NodeSnapshot, NodeState, ResourceUsage, SupervisionSettings, TreeSnapshot, TreeTotals, UnixMillis,
47};
48
49/// API route serving a snapshot of a supervision tree.
50///
51/// Exported so that a client can address the route without restating the path.
52pub const SUPERVISION_TREE_ROUTE: &str = "/runtime/processes";
53
54/// Maximum depth the snapshot walk descends before it stops and reports the subtree as truncated.
55///
56/// The node graph cannot contain a cycle -- a supervisor can't be its own ancestor, since
57/// [`add_worker`][super::Supervisor::add_worker] takes its child by value -- so this is not a termination condition
58/// but a stack guard. The walk is recursive and runs wherever the caller asks for a snapshot, which may be a
59/// request-serving task with a modest stack, and a diagnostics endpoint should not be able to abort the process even
60/// if a future change does manage to build a pathological tree.
61const MAX_TREE_DEPTH: usize = 64;
62
63/// A wall-clock and monotonic reading of the same instant.
64///
65/// Both are needed. Wall clock is what an operator wants to see (and what correlates with logs), but it can step
66/// backwards, so ages computed from it can come out negative or absurd. The monotonic reading can't, so every
67/// duration is derived from it and every displayed timestamp from the wall clock.
68///
69/// Uses [`std::time::Instant`] rather than [`tokio::time::Instant`] deliberately: a snapshot may be taken from a
70/// thread with no Tokio runtime, and it should report real elapsed time rather than a test's virtual clock.
71#[derive(Clone, Copy)]
72pub(super) struct Stamp {
73    wall: SystemTime,
74    mono: Instant,
75}
76
77impl Stamp {
78    pub(super) fn now() -> Self {
79        Self {
80            wall: SystemTime::now(),
81            mono: Instant::now(),
82        }
83    }
84
85    fn wall_millis(&self) -> UnixMillis {
86        // A pre-epoch clock makes `duration_since` fail rather than return a negative duration; report it as the
87        // epoch rather than panicking inside a diagnostics path.
88        UnixMillis(duration_millis(
89            self.wall.duration_since(UNIX_EPOCH).unwrap_or_default(),
90        ))
91    }
92
93    fn elapsed_millis(&self) -> u64 {
94        duration_millis(self.mono.elapsed())
95    }
96}
97
98/// Identity of a child that survives the id churn of a restart.
99///
100/// A supervisor keys its live roster by an id drawn from a monotonic counter, and a restart -- whether a
101/// [`OneForAll`][super::RestartMode::OneForAll] restart of the group or a restart of the supervisor itself --
102/// discards the whole roster and re-registers every eligible child under a *fresh* id. Facts that must outlive that
103/// -- when the child was first created, how many times it has been restarted -- therefore can't be keyed by roster
104/// id.
105#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
106pub(super) enum ChildKey {
107    /// A child declared before the run, identified by its index in the supervisor's static child list.
108    ///
109    /// That list is fixed once the supervisor starts, so the index is a stable identity across generations.
110    Static(usize),
111
112    /// A dynamically spawned child, identified by its roster id.
113    ///
114    /// A dynamic child is never restored across generations, so its roster id is as stable an identity as it needs.
115    /// Ids come from a monotonic counter, so ordering by one is ordering by spawn.
116    Dynamic(u64),
117}
118
119impl ChildKey {
120    fn is_dynamic(&self) -> bool {
121        matches!(self, Self::Dynamic(_))
122    }
123}
124
125/// The process a child was started under.
126///
127/// Produced by the supervisor's worker bookkeeping at the moment a child's task is spawned, which is the only place
128/// the child's [`Process`] exists and therefore the only place its identity can be captured.
129#[derive(Clone)]
130pub(super) struct StartedChild {
131    process_id: ProcessId,
132    process_name: Arc<str>,
133    at: Stamp,
134}
135
136impl StartedChild {
137    pub(super) fn new(process: &Process, process_name: Arc<str>) -> Self {
138        Self {
139            process_id: *process.id(),
140            process_name,
141            at: Stamp::now(),
142        }
143    }
144}
145
146/// Facts about a child that its supervisor knows at registration time.
147///
148/// Assembled by the supervisor (which can see its own child specifications and configuration) and handed to
149/// [`Roster::insert`], so that this module needs no visibility into either.
150pub(super) struct ChildFacts {
151    /// Identity that survives a group restart.
152    pub(super) key: ChildKey,
153
154    pub(super) name: Arc<str>,
155
156    /// The child's own node, if the child is a nested supervisor.
157    ///
158    /// This is what links a parent's bookkeeping to its child's, and so what makes the tree walkable.
159    pub(super) node: Option<Arc<SupervisorNode>>,
160
161    pub(super) restart: RestartType,
162
163    /// Whether the child's termination can drive its supervisor to shut down.
164    pub(super) significant: bool,
165}
166
167/// What a child is, from its parent's point of view.
168enum RecordKind {
169    Worker,
170
171    Supervisor(Arc<SupervisorNode>),
172}
173
174/// A supervisor's record of one child in the current generation.
175struct ChildRecord {
176    key: ChildKey,
177    name: Arc<str>,
178    kind: RecordKind,
179    restart: RestartType,
180    significant: bool,
181    /// The process the child is currently running under, or last ran under. `None` only for a child that never
182    /// successfully started.
183    start: Option<StartedChild>,
184    /// When the child exited for good, if it has. `None` while it is running.
185    exited: Option<Stamp>,
186}
187
188/// Cumulative facts about a child that outlive any single generation of it.
189struct ChildHistory {
190    /// When the child first entered its supervisor's roster.
191    ///
192    /// Deliberately not its registration time: registration happens once, during program construction, and never
193    /// again on a later generation, so it measures how long ago a builder method was called rather than anything
194    /// about the running tree.
195    created: Stamp,
196    restarts: u32,
197}
198
199/// A supervisor's configuration.
200///
201/// Owned by the supervisor's node rather than by the [`Supervisor`][super::Supervisor] itself, so that there is one
202/// copy rather than two that have to be kept in step, and so that it can be read before the supervisor has ever run.
203#[derive(Clone, Copy)]
204pub(super) struct NodeConfig {
205    pub(super) restart_strategy: RestartStrategy,
206    pub(super) auto_shutdown: AutoShutdown,
207    pub(super) dedicated_threads: Option<usize>,
208}
209
210/// The mutable half of a [`SupervisorNode`].
211struct NodeInner {
212    config: NodeConfig,
213    /// The supervisor's most recent run. `None` only before its first one.
214    ///
215    /// Retained after a run ends rather than cleared, so that a stopped supervisor still reports what it last ran as
216    /// -- which is what a postmortem asks of it, and what its own workers have always answered. Whether this run is
217    /// still in progress is carried by `running`, not by the presence of an identity.
218    ///
219    /// The process name here is recorded by the supervisor itself rather than by its parent, because the two can
220    /// differ: a supervisor on a dedicated runtime re-roots its process name when it starts (a known defect), so only
221    /// the supervisor knows what it actually runs as -- and only that name matches the resource group its allocations
222    /// are attributed to.
223    run: Option<StartedChild>,
224    /// Whether `run` is in progress rather than ended.
225    running: bool,
226    generation: u64,
227    restarts_performed: u64,
228    /// Children of the current generation, in declaration order.
229    ///
230    /// Ordered rather than hashed so the rendered tree and the serialized JSON come out in the order children were
231    /// declared, which is how the tree is written and therefore how it reads.
232    children: FastIndexMap<u64, ChildRecord>,
233    /// Per-child facts that outlive a generation, keyed by an identity that survives a restart.
234    ///
235    /// Only statically declared children keep an entry across a restart of the supervisor itself, since only they
236    /// have an identity that a later generation can look up again.
237    history: FastHashMap<ChildKey, ChildHistory>,
238    /// Supervisors that a worker child drives internally, keyed by that worker's roster id.
239    ///
240    /// Some workers own a supervisor rather than being one -- they build it after initialization and run it inside
241    /// their own future -- so the parent has no supervisor value to record at registration time and the subtree would
242    /// otherwise be invisible. Such a supervisor registers itself here when it starts.
243    adopted: FastHashMap<u64, Arc<SupervisorNode>>,
244    /// Scopes owned by worker children, keyed by the roster id of the worker.
245    ///
246    /// The scope of a worker holds all children that the worker spawns as its own. Like an adopted supervisor, the
247    /// scope exists only after the worker starts. Thus, the scope is not known at registration time, and it registers
248    /// itself here.
249    scoped: FastHashMap<u64, Arc<SupervisorNode>>,
250}
251
252impl NodeInner {
253    /// Discards everything specific to a single generation of the supervisor.
254    ///
255    /// Used when a run begins or ends, which discards the generation wholesale. A group restart within a run keeps
256    /// the tombstones it won't bring back, and so prunes the roster itself.
257    ///
258    /// A dynamic child's history belongs to the generation and goes with it: its key is a roster id drawn from a
259    /// counter that is never reset, so once the generation is gone nothing can look the key up again -- no later
260    /// generation restores the child, and no later spawn reuses the id. A statically declared child's history is
261    /// kept, since the next generation re-registers it under the same identity.
262    fn clear_generation(&mut self) {
263        self.children.clear();
264        self.adopted.clear();
265        self.scoped.clear();
266        self.history.retain(|key, _| !key.is_dynamic());
267    }
268
269    fn running_children(&self) -> usize {
270        self.children.values().filter(|r| r.exited.is_none()).count()
271    }
272}
273
274/// The supervised child slot a worker is running as.
275///
276/// Installed around every supervised worker task so that a supervisor a worker builds and runs inside its own future
277/// can attach itself to the tree, which it could not otherwise do: the parent has no supervisor value to record at
278/// registration time, because the supervisor does not exist until the worker is already running.
279#[derive(Clone)]
280pub(super) struct TreeParent {
281    node: Arc<SupervisorNode>,
282    child_id: u64,
283}
284
285impl TreeParent {
286    pub(super) fn new(node: Arc<SupervisorNode>, child_id: u64) -> Self {
287        Self { node, child_id }
288    }
289
290    /// Records `scope` as the scope of the worker in this slot.
291    pub(super) fn attach_scope(&self, scope: Arc<SupervisorNode>) {
292        self.node.state().scoped.insert(self.child_id, scope);
293    }
294
295    /// Removes `scope` as the scope of the worker in this slot, if it is still the scope of that worker.
296    ///
297    /// This function compares by identity, and does not remove a different scope that it finds in the slot. A restarted
298    /// worker attaches a new scope under the same slot. A late detach of the old scope must not remove the new scope.
299    pub(super) fn detach_scope(&self, scope: &Arc<SupervisorNode>) {
300        let mut state = self.node.state();
301        if state
302            .scoped
303            .get(&self.child_id)
304            .is_some_and(|attached| Arc::ptr_eq(attached, scope))
305        {
306            state.scoped.remove(&self.child_id);
307        }
308    }
309}
310
311tokio::task_local! {
312    /// The child slot the currently running worker occupies in its supervisor.
313    pub(super) static CURRENT_TREE_PARENT: TreeParent;
314}
315
316/// Shared bookkeeping for one supervisor.
317///
318/// Held by the supervisor itself, by every internal clone of it (so all generations write to the same place), by its
319/// parent's record of it, and by any [`SupervisionTreeHandle`] pointed at it.
320pub(super) struct SupervisorNode {
321    id: Arc<str>,
322    created: Stamp,
323    state: Mutex<NodeInner>,
324}
325
326impl SupervisorNode {
327    pub(super) fn new(id: Arc<str>) -> Self {
328        Self {
329            id,
330            created: Stamp::now(),
331            state: Mutex::new(NodeInner {
332                config: NodeConfig {
333                    restart_strategy: RestartStrategy::default(),
334                    auto_shutdown: AutoShutdown::default(),
335                    dedicated_threads: None,
336                },
337                run: None,
338                running: false,
339                generation: 0,
340                restarts_performed: 0,
341                children: FastIndexMap::default(),
342                history: FastHashMap::default(),
343                adopted: FastHashMap::default(),
344                scoped: FastHashMap::default(),
345            }),
346        }
347    }
348
349    fn state(&self) -> std::sync::MutexGuard<'_, NodeInner> {
350        self.state.lock().expect("supervision-tree node lock poisoned")
351    }
352
353    pub(super) fn config(&self) -> NodeConfig {
354        self.state().config
355    }
356
357    pub(super) fn update_config(&self, f: impl FnOnce(&mut NodeConfig)) {
358        f(&mut self.state().config);
359    }
360
361    /// Records that the supervisor has begun a run under `process`.
362    ///
363    /// Resets everything specific to a generation, so a run that ended abruptly -- a panicked task, or a dedicated
364    /// runtime thread that unwound without reaching its cleanup -- can't leave stale children behind for the next
365    /// generation to be confused with. Counts the new generation as a restart of every statically declared child,
366    /// since every one of them is about to be re-registered under a fresh process.
367    pub(super) fn begin_run(self: &Arc<Self>, process: &Process) {
368        let mut state = self.state();
369
370        if state.running {
371            // Two overlapping runs share one node, one spawn queue, one id counter and one child gauge, so the second
372            // silently clobbers the first. This is unreachable from outside the crate -- `add_worker` takes its child
373            // by value and `Supervisor` has no public `Clone` -- so what this really guards is a restart that begins
374            // before the previous generation has finished stopping.
375            debug_assert!(
376                false,
377                "supervisor '{}' began a run while another was still active",
378                self.id
379            );
380            warn!(
381                supervisor_id = %self.id,
382                "Supervisor began a run while another was still active; a supervisor must not be supervised by two \
383                 parents."
384            );
385        }
386
387        state.run = Some(StartedChild::new(process, process.name().into()));
388        state.running = true;
389        state.generation += 1;
390        state.clear_generation();
391
392        // A statically declared child is identified by its position in a list fixed before the run, so this
393        // generation re-registers it under the same identity and its history carries over. Every one of them comes
394        // back -- `spawn_static_children` filters on nothing -- and each comes back under a fresh process, so this
395        // generation is a restart of each of them.
396        //
397        // Only static children are left to count here: `clear_generation` has just discarded the rest. There is
398        // nothing at all to count on the first run, which has no history to carry over.
399        for history in state.history.values_mut() {
400            history.restarts += 1;
401        }
402        drop(state);
403
404        // If we were started inside a worker's task, attach ourselves to the tree under that worker. This is how a
405        // supervisor built and run inside a worker's future -- rather than handed to a parent as a child -- becomes
406        // visible. Absent when the supervisor is a root, or is running on a dedicated runtime (whose thread has no
407        // task-locals), and in the latter case the parent already recorded us directly.
408        let _ = CURRENT_TREE_PARENT.try_with(|parent| {
409            parent.node.state().adopted.insert(parent.child_id, Arc::clone(self));
410        });
411    }
412
413    /// Records that the supervisor's run has ended.
414    ///
415    /// Keeps everything cumulative -- generation count, restarts performed, static children's history, and the
416    /// identity of the run itself -- and discards the generation's children, so a stopped supervisor reports as
417    /// stopped rather than presenting a subtree of processes that no longer exist.
418    pub(super) fn end_run(&self) {
419        let mut state = self.state();
420        state.running = false;
421        state.clear_generation();
422        drop(state);
423
424        let _ = CURRENT_TREE_PARENT.try_with(|parent| {
425            parent.node.state().adopted.remove(&parent.child_id);
426        });
427    }
428}
429
430/// A supervisor's children, in both the form the supervisor runs from and the form an observer reads.
431///
432/// The supervisor's own roster is a plain map it owns exclusively; the observable bookkeeping lives behind a lock and
433/// is shared. Both are updated through this one type so that they cannot drift: there is no way to add, restart or
434/// remove a child in one without doing so in the other.
435pub(super) struct Roster<E> {
436    live: FastHashMap<u64, E>,
437    node: Arc<SupervisorNode>,
438}
439
440impl<E> Roster<E> {
441    pub(super) fn new(node: Arc<SupervisorNode>) -> Self {
442        Self {
443            live: FastHashMap::default(),
444            node,
445        }
446    }
447
448    pub(super) fn get(&self, id: u64) -> Option<&E> {
449        self.live.get(&id)
450    }
451
452    pub(super) fn values(&self) -> impl Iterator<Item = &E> {
453        self.live.values()
454    }
455
456    /// Registers a newly started child.
457    ///
458    /// `facts` describes the child as its supervisor configured it, and `started` identifies the process it was
459    /// started under. A child re-registered under a fresh id after a group restart keeps the creation time and
460    /// restart count recorded against its [`ChildKey`].
461    pub(super) fn insert(&mut self, id: u64, entry: E, facts: ChildFacts, started: StartedChild) {
462        self.live.insert(id, entry);
463
464        let mut state = self.node.state();
465        state.history.entry(facts.key).or_insert_with(|| ChildHistory {
466            created: started.at,
467            restarts: 0,
468        });
469
470        let kind = match facts.node {
471            Some(node) => RecordKind::Supervisor(node),
472            None => RecordKind::Worker,
473        };
474
475        state.children.insert(
476            id,
477            ChildRecord {
478                key: facts.key,
479                name: facts.name,
480                kind,
481                restart: facts.restart,
482                significant: facts.significant,
483                start: Some(started),
484                exited: None,
485            },
486        );
487        drop(state);
488
489        self.debug_assert_consistent();
490    }
491
492    pub(super) fn restart_in_place(&mut self, id: u64, started: StartedChild) {
493        let mut state = self.node.state();
494        state.restarts_performed += 1;
495
496        // Any supervisor that the previous instance of the child adopted is now stopped. The new instance registers again
497        // if it has one.
498        //
499        // The scope of the previous instance is not removed here. That scope detached itself when the previous instance
500        // was dropped, before its exit was reported. The new instance is already running at this point, and it can
501        // already have attached its own scope, so a removal here could only remove the scope of the new instance.
502        state.adopted.remove(&id);
503
504        if let Some(record) = state.children.get_mut(&id) {
505            let key = record.key;
506            record.start = Some(started);
507            record.exited = None;
508            if let Some(history) = state.history.get_mut(&key) {
509                history.restarts += 1;
510            }
511        }
512        drop(state);
513
514        self.debug_assert_consistent();
515    }
516
517    /// Removes the child under `id`, which has exited and will not be restarted.
518    ///
519    /// A statically declared child is kept as a tombstone rather than dropped: it remains part of the tree's declared
520    /// shape, and "declared, ran, and then stopped for good" is precisely what an operator needs to see. A
521    /// dynamically spawned child is dropped outright -- the canonical dynamic child is one short-lived task per unit
522    /// of work, so retaining every one that has ever finished would grow without bound.
523    pub(super) fn remove(&mut self, id: u64) -> Option<E> {
524        let entry = self.live.remove(&id);
525
526        let mut state = self.node.state();
527        state.adopted.remove(&id);
528        state.scoped.remove(&id);
529        match state.children.get(&id).map(|record| record.key) {
530            // Order is restored by sorting on `ChildKey` when a snapshot is taken, so the roster itself doesn't need
531            // to preserve it -- which lets a dynamic child, of which there may be one per unit of work, leave in
532            // constant time rather than shifting every entry behind it.
533            Some(key) if key.is_dynamic() => {
534                state.children.swap_remove(&id);
535                state.history.remove(&key);
536            }
537            Some(_) => {
538                if let Some(record) = state.children.get_mut(&id) {
539                    record.exited = Some(Stamp::now());
540                }
541            }
542            None => {}
543        }
544        drop(state);
545
546        self.debug_assert_consistent();
547
548        entry
549    }
550
551    /// Clears the roster ahead of a group restart, counting it as a restart of every child it will bring back.
552    ///
553    /// The children about to be re-registered each get a new process and a new start time, so counting the restart
554    /// against every one of them is what keeps their restart counts consistent with the rest of their own record.
555    /// Children that a group restart doesn't bring back keep their tombstones.
556    pub(super) fn clear_for_group_restart(&mut self) {
557        self.live.clear();
558
559        let mut state = self.node.state();
560        state.restarts_performed += 1;
561
562        let stopped_at = Stamp::now();
563        let NodeInner { children, history, .. } = &mut *state;
564        for record in children.values() {
565            if record.key.is_dynamic() {
566                // A dynamic child is not restored, and its roster id is never handed out again, so anything keyed to
567                // it is unreachable from here on rather than merely stale.
568                history.remove(&record.key);
569                continue;
570            }
571
572            // Exactly the children `respawn_children_one_for_all` brings back: every statically declared child that
573            // isn't temporary, whatever state it was last in. That includes a transient child which had already
574            // exited cleanly, since a group restart starts it afresh rather than honoring the rule that governs its
575            // own termination.
576            if record.restart != RestartType::Temporary {
577                if let Some(history) = history.get_mut(&record.key) {
578                    history.restarts += 1;
579                }
580            }
581        }
582
583        // Keep the tombstones of statically declared children a group restart won't bring back, since they remain
584        // part of the tree's declared shape. Everything else is about to be re-registered under a fresh id, and
585        // keeping the old record too would show each of those children twice.
586        //
587        // A temporary child still running is shut down along with the group and never brought back, so this is where
588        // it stops for good: record it as exited rather than dropping it, or it would vanish from the tree entirely.
589        state.children.retain(|_, record| {
590            if record.key.is_dynamic() || record.restart != RestartType::Temporary {
591                return false;
592            }
593
594            if record.exited.is_none() {
595                record.exited = Some(stopped_at);
596            }
597            true
598        });
599        state.adopted.clear();
600        state.scoped.clear();
601        drop(state);
602
603        self.debug_assert_consistent();
604    }
605
606    /// Checks that the supervisor's roster and the observable bookkeeping still agree.
607    ///
608    /// Every mutating method ends here, so the crate's existing supervision tests double as drift tests for this
609    /// module without needing to know it exists.
610    fn debug_assert_consistent(&self) {
611        debug_assert_eq!(
612            self.live.len(),
613            self.node.state().running_children(),
614            "supervisor '{}' roster and supervision-tree bookkeeping diverged",
615            self.node.id
616        );
617
618        saluki_antithesis::always_or_unreachable!(
619            self.live.len() == self.node.state().running_children(),
620            "supervision-tree bookkeeping tracks the supervisor's live roster"
621        );
622    }
623}
624
625/// A read-only handle for taking snapshots of a supervision tree.
626///
627/// Obtained from [`Supervisor::tree_handle`][super::Supervisor::tree_handle]. Cheap to clone, safe to share across
628/// tasks and threads, and deliberately incapable of anything but observation -- it cannot spawn children or
629/// otherwise affect the tree it reports on.
630///
631/// A handle can be taken before the supervisor starts, and remains valid across every restart of it.
632#[derive(Clone)]
633pub struct SupervisionTreeHandle {
634    node: Arc<SupervisorNode>,
635}
636
637impl SupervisionTreeHandle {
638    pub(super) fn new(node: Arc<SupervisorNode>) -> Self {
639        Self { node }
640    }
641
642    /// Returns the identifier of the supervisor at the root of this handle's view.
643    pub fn name(&self) -> &str {
644        &self.node.id
645    }
646
647    /// Renders the current state of the supervision tree as pretty-printed JSON.
648    ///
649    /// Returns a JSON object describing the failure if the tree can't be serialized, so that a caller writing a
650    /// diagnostic artifact always has something to write.
651    pub fn snapshot_json(&self) -> String {
652        match serde_json::to_string_pretty(&self.snapshot()) {
653            Ok(json) => json,
654            Err(e) => {
655                warn!(error = %e, "Failed to serialize supervision tree.");
656                String::from(r#"{"error": "failed to serialize supervision tree"}"#)
657            }
658        }
659    }
660
661    /// Creates a [`SupervisionTreeAPIHandler`] serving snapshots of this tree.
662    pub fn api_handler(&self) -> SupervisionTreeAPIHandler {
663        SupervisionTreeAPIHandler::from_handle(self.clone())
664    }
665
666    /// Creates a [`SupervisionTreeWorker`] that publishes this tree over the control plane.
667    pub fn worker(&self) -> SupervisionTreeWorker {
668        SupervisionTreeWorker::new(self.clone())
669    }
670
671    /// Captures the current state of the supervision tree.
672    ///
673    /// Descends from this handle's supervisor through every child, nesting each child's own children beneath it.
674    ///
675    /// # Consistency
676    ///
677    /// The tree is read one supervisor at a time, so a snapshot is not a globally atomic view: a child may start or
678    /// exit while the walk is in progress, and the result then mixes observations from slightly different instants.
679    /// This is deliberate. Reading the whole tree atomically would mean holding every supervisor's bookkeeping at
680    /// once, which would let an operator-facing diagnostics call stall supervision across the entire process -- a far
681    /// worse property than a snapshot whose nodes are microseconds apart.
682    pub fn snapshot(&self) -> TreeSnapshot {
683        // Read every resource group once, before touching any node. Doing it per node would take the global resource
684        // registry's lock once per node, each acquisition contending with the group registration that happens
685        // whenever any supervisor anywhere in the process starts a child.
686        let groups = collect_resource_groups();
687
688        let mut walk = Walk {
689            groups,
690            totals: TreeTotals::default(),
691            counted_groups: FastHashSet::default(),
692        };
693        let root = walk.supervisor(&self.node, ParentFacts::root(&self.node), 1);
694
695        TreeSnapshot {
696            captured_at: Stamp::now().wall_millis(),
697            resource_tracking_enabled: ResourceGroupRegistry::allocator_installed(),
698            totals: walk.totals,
699            root,
700        }
701    }
702}
703
704fn collect_resource_groups() -> FastHashMap<Arc<str>, ResourceUsage> {
705    let mut groups = FastHashMap::default();
706
707    ResourceGroupRegistry::global().visit_resource_groups(|name, stats| {
708        groups.insert(
709            Arc::<str>::from(name),
710            ResourceUsage::from(&stats.snapshot_delta(&ResourceStatsSnapshot::empty())),
711        );
712    });
713
714    groups
715}
716
717/// What a node's parent contributes to its snapshot.
718///
719/// A node's own bookkeeping knows what it is doing; its parent knows how it was configured, when it was created, and
720/// how many times it has been restarted. Both halves are needed, and only the parent has the second one -- except at
721/// the root, which has no parent.
722#[derive(Clone)]
723struct ParentFacts {
724    name: Arc<str>,
725    restart: RestartType,
726    significant: bool,
727    created: Stamp,
728    restarts: u32,
729    exited: Option<Stamp>,
730}
731
732impl ParentFacts {
733    /// Facts for a node with no parent: the root of the walk, or a supervisor a worker attached to the tree itself.
734    fn root(node: &SupervisorNode) -> Self {
735        Self {
736            name: Arc::clone(&node.id),
737            restart: RestartType::default(),
738            significant: false,
739            created: node.created,
740            restarts: 0,
741            exited: None,
742        }
743    }
744}
745
746struct RunFacts {
747    process_id: Option<ProcessId>,
748    process_name: Option<Arc<str>>,
749    started: Option<Stamp>,
750    state: NodeState,
751    /// The group this node's allocations are attributed to, which for a worker is its supervisor's rather than its
752    /// own.
753    resource_group: Option<Arc<str>>,
754    resources: Option<ResourceUsage>,
755}
756
757/// A node its parent has seen exit is stopped whatever else is true; otherwise `running` decides.
758///
759/// Each caller supplies `running` from what it actually knows, which is not the same thing in both cases. A
760/// supervisor's own bookkeeping says whether its run is in progress, and must be asked: it keeps the identity of its
761/// most recent run after that run ends, so reading the identity would say "running" forever. A worker has no such
762/// bookkeeping, but its record is created at the moment it starts and keeps an exit stamp once it stops, so for a
763/// worker having an identity and no exit stamp is the same statement.
764fn node_state(exited: Option<Stamp>, running: bool) -> NodeState {
765    match (exited, running) {
766        (Some(_), _) => NodeState::Exited,
767        (None, true) => NodeState::Running,
768        (None, false) => NodeState::Registered,
769    }
770}
771
772struct Walk {
773    groups: FastHashMap<Arc<str>, ResourceUsage>,
774    totals: TreeTotals,
775    /// Resource groups already folded into the totals.
776    ///
777    /// Group registration is idempotent by name, so two supervisors whose fully qualified names coincide share one
778    /// group. Totalling per node would then count the same bytes twice.
779    counted_groups: FastHashSet<Arc<str>>,
780}
781
782impl Walk {
783    fn supervisor(&mut self, node: &Arc<SupervisorNode>, parent: ParentFacts, depth: usize) -> NodeSnapshot {
784        // Copy out everything needed, then release the lock before descending: holding a parent's lock while taking a
785        // child's is the only way this walk could ever deadlock against a running supervisor, and not doing it is
786        // simpler to guarantee than any ordering rule.
787        let (run, running, config, generation, restarts_performed, children) = {
788            let state = node.state();
789            (
790                state.run.clone(),
791                state.running,
792                state.config,
793                state.generation,
794                state.restarts_performed,
795                pending_children(&state),
796            )
797        };
798
799        let (process_id, process_name, started) = split_run(run.as_ref());
800
801        // Taken from `running` rather than from whether an identity is present: the identity is that of the most
802        // recent run and outlives it, so deriving the state from it would have every supervisor that has ever run
803        // report itself as running forever.
804        let state = node_state(parent.exited, running);
805        let resources = process_name.as_ref().and_then(|name| self.groups.get(name).cloned());
806
807        // A worker inherits its supervisor's resource group rather than owning one, so this is what the children below
808        // are attributed to.
809        let worker_group = process_name.clone();
810
811        let mut snapshot = self.node_snapshot(
812            NodeKind::Supervisor,
813            parent,
814            RunFacts {
815                process_id,
816                // A supervisor owns its resource group, and it is named by the process it actually runs as.
817                resource_group: process_name.clone(),
818                process_name,
819                started,
820                state,
821                resources,
822            },
823            depth,
824        );
825
826        snapshot.supervision = Some(SupervisionSettings {
827            restart_mode: config.restart_strategy.mode(),
828            restart_intensity: config.restart_strategy.intensity(),
829            restart_period_ms: duration_millis(config.restart_strategy.period()),
830            auto_shutdown: config.auto_shutdown,
831            dedicated_threads: config.dedicated_threads,
832            restarts_performed,
833            generation,
834        });
835
836        if children.is_empty() || !self.may_descend(&node.id, depth) {
837            return snapshot;
838        }
839
840        snapshot.children.reserve(children.len());
841        for child in children {
842            snapshot
843                .children
844                .push(self.child(child, worker_group.as_ref(), depth + 1));
845        }
846
847        snapshot
848    }
849
850    fn child(&mut self, child: PendingChild, worker_group: Option<&Arc<str>>, depth: usize) -> NodeSnapshot {
851        match child.kind {
852            // A nested supervisor reports its own process and its own children; its parent only contributes how it
853            // was configured and how it has fared.
854            PendingKind::Supervisor(node) => self.supervisor(&node, child.facts, depth),
855
856            PendingKind::Worker { adopted, scope } => {
857                // Usually, a worker counts against the resource group of its supervisor. But a worker can represent
858                // something with its own accounting, for example a topology component. Such a worker owns a group with
859                // the name of its process.
860                let own_group = child
861                    .start
862                    .as_ref()
863                    .map(|start| &start.process_name)
864                    .filter(|name| self.groups.contains_key(*name))
865                    .cloned();
866                let worker_group = own_group.as_ref().or(worker_group);
867
868                let mut snapshot = self.worker(child.facts, child.start, worker_group, own_group.is_some(), depth);
869
870                // The worker was found to drive a supervisor of its own. The parent supervises the worker, so the
871                // worker stays the node. The supervisor that the worker drives appears under the worker.
872                if let Some(adopted) = adopted {
873                    if self.may_descend(&adopted.id, depth) {
874                        let nested = self.supervisor(&adopted, ParentFacts::root(&adopted), depth + 1);
875                        snapshot.children.push(nested);
876                    }
877                }
878
879                // These are the children that the worker spawned into its own scope. They belong to the worker, so
880                // they appear directly under the worker. The scope is how the worker holds them, and it is not a
881                // separate node. The children run in the process of the worker, so they count against the same
882                // resource group as the worker.
883                if let Some(scope) = scope {
884                    if self.may_descend(&scope.id, depth) {
885                        let children = pending_children(&scope.state());
886                        for child in children {
887                            let nested = self.child(child, worker_group, depth + 1);
888                            snapshot.children.push(nested);
889                        }
890                    }
891                }
892
893                snapshot
894            }
895        }
896    }
897
898    fn worker(
899        &mut self, facts: ParentFacts, start: Option<StartedChild>, worker_group: Option<&Arc<str>>, owns_group: bool,
900        depth: usize,
901    ) -> NodeSnapshot {
902        let (process_id, process_name, started) = split_run(start.as_ref());
903        let state = node_state(facts.exited, process_id.is_some());
904
905        // Usually, a worker owns no resource group. Its allocations count against the supervisor named above. If each
906        // worker reports the totals of its supervisor, the same totals are counted many times. A worker that owns a
907        // resource group reports its figures, as a supervisor does.
908        let resources = owns_group
909            .then(|| worker_group.and_then(|group| self.groups.get(group).cloned()))
910            .flatten();
911
912        self.node_snapshot(
913            NodeKind::Worker,
914            facts,
915            RunFacts {
916                process_id,
917                process_name,
918                started,
919                state,
920                resource_group: worker_group.cloned(),
921                resources,
922            },
923            depth,
924        )
925    }
926
927    /// Assembles a node from its two halves, and folds it into the tree totals.
928    fn node_snapshot(&mut self, kind: NodeKind, parent: ParentFacts, run: RunFacts, depth: usize) -> NodeSnapshot {
929        self.totals.max_depth = self.totals.max_depth.max(depth);
930
931        match kind {
932            NodeKind::Supervisor => self.totals.supervisors += 1,
933            NodeKind::Worker => self.totals.workers += 1,
934        }
935        match run.state {
936            NodeState::Running => self.totals.running += 1,
937            NodeState::Exited => self.totals.exited += 1,
938            NodeState::Registered => self.totals.registered += 1,
939        }
940        self.totals.restarts += u64::from(parent.restarts);
941
942        // Several nodes can share one resource group, so fold each group in once rather than once per node.
943        if let (Some(usage), Some(group)) = (&run.resources, &run.process_name) {
944            if self.counted_groups.insert(Arc::clone(group)) {
945                self.totals.live_bytes += usage.live_bytes;
946                self.totals.cpu_time_nanos += usage.cpu_time_nanos;
947            }
948        }
949
950        NodeSnapshot {
951            name: parent.name.to_string(),
952            kind,
953            process_name: run.process_name.as_deref().map(str::to_string),
954            process_id: run.process_id.map(|id| id.as_usize() as u64),
955            state: run.state,
956            restart: parent.restart,
957            significant: parent.significant,
958            created_at: parent.created.wall_millis(),
959            started_at: run.started.map(|started| started.wall_millis()),
960            uptime_ms: match run.state {
961                NodeState::Running => run.started.map(|started| started.elapsed_millis()),
962                _ => None,
963            },
964            restart_count: parent.restarts,
965            exited_at: parent.exited.map(|exited| exited.wall_millis()),
966            resource_group: run.resource_group.as_deref().map(str::to_string),
967            resources: run.resources,
968            supervision: None,
969            children: Vec::new(),
970        }
971    }
972
973    fn may_descend(&self, supervisor_id: &str, depth: usize) -> bool {
974        if depth < MAX_TREE_DEPTH {
975            return true;
976        }
977
978        warn!(
979            supervisor_id,
980            depth, "Supervision tree is deeper than the snapshot walk descends; subtree omitted."
981        );
982        false
983    }
984}
985
986fn split_run(started: Option<&StartedChild>) -> (Option<ProcessId>, Option<Arc<str>>, Option<Stamp>) {
987    match started {
988        Some(start) => (
989            Some(start.process_id),
990            Some(Arc::clone(&start.process_name)),
991            Some(start.at),
992        ),
993        None => (None, None, None),
994    }
995}
996
997/// A child copied out from under its supervisor's lock, ready to be walked.
998struct PendingChild {
999    key: ChildKey,
1000    facts: ParentFacts,
1001    kind: PendingKind,
1002    start: Option<StartedChild>,
1003}
1004
1005enum PendingKind {
1006    Worker {
1007        /// A supervisor that the worker drives inside its own future, and that appears under the worker.
1008        adopted: Option<Arc<SupervisorNode>>,
1009
1010        /// The worker's own scope, whose children appear under the worker.
1011        scope: Option<Arc<SupervisorNode>>,
1012    },
1013    Supervisor(Arc<SupervisorNode>),
1014}
1015
1016/// Copies the children of a node out of the locked node state, in the order that a snapshot shows them.
1017fn pending_children(state: &NodeInner) -> Vec<PendingChild> {
1018    let mut children = state
1019        .children
1020        .iter()
1021        .map(|(id, record)| {
1022            let history = state.history.get(&record.key);
1023            PendingChild {
1024                key: record.key,
1025                facts: ParentFacts {
1026                    name: Arc::clone(&record.name),
1027                    restart: record.restart,
1028                    significant: record.significant,
1029                    created: history
1030                        .map(|h| h.created)
1031                        .or_else(|| record.start.as_ref().map(|start| start.at))
1032                        .unwrap_or_else(Stamp::now),
1033                    restarts: history.map(|h| h.restarts).unwrap_or(0),
1034                    exited: record.exited,
1035                },
1036                kind: match &record.kind {
1037                    RecordKind::Supervisor(node) => PendingKind::Supervisor(Arc::clone(node)),
1038                    RecordKind::Worker => PendingKind::Worker {
1039                        adopted: state.adopted.get(id).cloned(),
1040                        scope: state.scoped.get(id).cloned(),
1041                    },
1042                },
1043                start: record.start.clone(),
1044            }
1045        })
1046        .collect::<Vec<_>>();
1047
1048    // A restart registers the child again under a new id. Thus, after any restart, the insertion order is not the same
1049    // as the declaration order. `ChildKey` orders static children by declaration and dynamic children by spawn. This
1050    // order keeps successive snapshots easy to diff.
1051    children.sort_by_key(|child| child.key);
1052
1053    children
1054}
1055
1056/// Converts a duration to whole milliseconds, saturating rather than overflowing.
1057///
1058/// A restart period comes from configuration, and it can be as large as [`Duration::MAX`].
1059fn duration_millis(duration: Duration) -> u64 {
1060    duration.as_millis().min(u128::from(u64::MAX)) as u64
1061}
1062
1063#[cfg(test)]
1064mod tests {
1065    use super::*;
1066
1067    fn test_node(id: &str) -> Arc<SupervisorNode> {
1068        Arc::new(SupervisorNode::new(Arc::from(id)))
1069    }
1070
1071    fn test_process(name: &str) -> Process {
1072        Process::supervisor(name, None).expect("valid process name")
1073    }
1074
1075    fn test_started(name: &str) -> StartedChild {
1076        StartedChild::new(&test_process(name), Arc::from(name))
1077    }
1078
1079    fn test_facts(key: ChildKey, name: &str, restart: RestartType) -> ChildFacts {
1080        ChildFacts {
1081            key,
1082            name: Arc::from(name),
1083            node: None,
1084            restart,
1085            significant: false,
1086        }
1087    }
1088
1089    fn restarts_of(node: &SupervisorNode, key: ChildKey) -> Option<u32> {
1090        node.state().history.get(&key).map(|history| history.restarts)
1091    }
1092
1093    /// Whether `key` still has a record, and if so whether it has been recorded as exited.
1094    fn record_of(node: &SupervisorNode, key: ChildKey) -> Option<bool> {
1095        node.state()
1096            .children
1097            .values()
1098            .find(|record| record.key == key)
1099            .map(|record| record.exited.is_some())
1100    }
1101
1102    #[test]
1103    fn group_restart_tombstones_a_temporary_child_that_was_still_running() {
1104        let node = test_node("sup");
1105        let mut roster = Roster::new(Arc::clone(&node));
1106        roster.insert(
1107            0,
1108            (),
1109            test_facts(ChildKey::Static(0), "one-shot", RestartType::Temporary),
1110            test_started("sup.one_shot"),
1111        );
1112        roster.insert(
1113            1,
1114            (),
1115            test_facts(ChildKey::Static(1), "flapper", RestartType::Permanent),
1116            test_started("sup.flapper"),
1117        );
1118
1119        roster.clear_for_group_restart();
1120
1121        // A group restart shuts the temporary child down and never brings it back, so this is where it stops for
1122        // good. It was declared, so it stays part of the tree's shape as a tombstone rather than vanishing.
1123        assert_eq!(
1124            record_of(&node, ChildKey::Static(0)),
1125            Some(true),
1126            "a temporary child stopped by a group restart is retained, and records that it stopped"
1127        );
1128        assert_eq!(
1129            restarts_of(&node, ChildKey::Static(0)),
1130            Some(0),
1131            "a child that is not brought back has not been restarted"
1132        );
1133
1134        // The permanent child is about to be re-registered under a fresh id, so keeping the old record too would
1135        // show it twice -- and it really is being restarted.
1136        assert_eq!(record_of(&node, ChildKey::Static(1)), None);
1137        assert_eq!(restarts_of(&node, ChildKey::Static(1)), Some(1));
1138    }
1139
1140    #[test]
1141    fn group_restart_counts_an_exited_transient_child_it_brings_back() {
1142        let node = test_node("sup");
1143        let mut roster = Roster::new(Arc::clone(&node));
1144        roster.insert(
1145            0,
1146            (),
1147            test_facts(ChildKey::Static(0), "transient", RestartType::Transient),
1148            test_started("sup.transient"),
1149        );
1150
1151        // A clean exit leaves a transient child stopped, as a tombstone.
1152        roster.remove(0);
1153        assert_eq!(record_of(&node, ChildKey::Static(0)), Some(true));
1154
1155        roster.clear_for_group_restart();
1156
1157        // A group restart starts every non-temporary static child afresh, including one that had already exited
1158        // cleanly, so the restart is counted against it and its tombstone gives way to the new registration.
1159        assert_eq!(restarts_of(&node, ChildKey::Static(0)), Some(1));
1160        assert_eq!(record_of(&node, ChildKey::Static(0)), None);
1161    }
1162
1163    #[test]
1164    fn group_restart_forgets_dynamic_children_entirely() {
1165        let node = test_node("sup");
1166        let mut roster = Roster::new(Arc::clone(&node));
1167        roster.insert(
1168            0,
1169            (),
1170            test_facts(ChildKey::Static(0), "static", RestartType::Permanent),
1171            test_started("sup.static"),
1172        );
1173        roster.insert(
1174            7,
1175            (),
1176            test_facts(ChildKey::Dynamic(7), "ephemeral", RestartType::Temporary),
1177            test_started("sup.ephemeral"),
1178        );
1179
1180        roster.clear_for_group_restart();
1181
1182        // A dynamic child is not restored, and its roster id is never handed out again, so nothing keyed to it can
1183        // ever be looked up again. Keeping it would strand one entry per dynamic child alive at every restart.
1184        let state = node.state();
1185        assert!(!state.history.contains_key(&ChildKey::Dynamic(7)));
1186        assert!(state.history.contains_key(&ChildKey::Static(0)));
1187    }
1188
1189    #[test]
1190    fn a_new_generation_counts_as_a_restart_of_every_static_child() {
1191        let node = test_node("sup");
1192        node.begin_run(&test_process("sup"));
1193
1194        let mut roster = Roster::new(Arc::clone(&node));
1195        roster.insert(
1196            0,
1197            (),
1198            test_facts(ChildKey::Static(0), "worker", RestartType::Permanent),
1199            test_started("sup.worker"),
1200        );
1201        roster.insert(
1202            5,
1203            (),
1204            test_facts(ChildKey::Dynamic(5), "ephemeral", RestartType::Temporary),
1205            test_started("sup.ephemeral"),
1206        );
1207        assert_eq!(restarts_of(&node, ChildKey::Static(0)), Some(0));
1208        drop(roster);
1209
1210        // The supervisor itself stops and is restarted from above. Its children go down with it, and every static one
1211        // is re-registered under a fresh process by the next generation, which is a restart of each of them.
1212        node.end_run();
1213        node.begin_run(&test_process("sup"));
1214
1215        assert_eq!(restarts_of(&node, ChildKey::Static(0)), Some(1));
1216        assert!(
1217            !node.state().history.contains_key(&ChildKey::Dynamic(5)),
1218            "a dynamic child is never restored, so its history cannot be looked up again"
1219        );
1220    }
1221
1222    #[test]
1223    fn restarting_a_child_keeps_the_scope_that_its_new_instance_attached() {
1224        let node = test_node("sup");
1225        let mut roster = Roster::new(Arc::clone(&node));
1226        roster.insert(
1227            0,
1228            (),
1229            test_facts(ChildKey::Dynamic(0), "owner", RestartType::Permanent),
1230            test_started("sup.owner"),
1231        );
1232
1233        // The restarted instance runs as soon as it is spawned. It can attach its scope before the roster records the
1234        // restart.
1235        let scope = test_node("sup.owner");
1236        TreeParent::new(Arc::clone(&node), 0).attach_scope(Arc::clone(&scope));
1237        roster.restart_in_place(0, test_started("sup.owner"));
1238
1239        let state = node.state();
1240        let attached = state
1241            .scoped
1242            .get(&0)
1243            .expect("the scope of the new instance must stay attached");
1244        assert!(Arc::ptr_eq(attached, &scope));
1245    }
1246
1247    #[test]
1248    fn a_supervisor_keeps_the_identity_of_its_most_recent_run() {
1249        let node = test_node("sup");
1250        let handle = SupervisionTreeHandle::new(Arc::clone(&node));
1251
1252        // Never run: there is no identity to report.
1253        let before = handle.snapshot();
1254        assert_eq!(before.root.state, NodeState::Registered);
1255        assert_eq!(before.root.process_id, None);
1256        assert_eq!(before.root.process_name, None);
1257        assert_eq!(before.root.started_at, None);
1258
1259        node.begin_run(&test_process("sup"));
1260        let running = handle.snapshot();
1261        assert_eq!(running.root.state, NodeState::Running);
1262        assert!(running.root.process_id.is_some());
1263        assert!(running.root.uptime_ms.is_some());
1264
1265        // Stopped, but it still says what it ran as -- as an exited worker always has. `state` carries the difference
1266        // between "not running" and "never ran", so the identity does not have to.
1267        node.end_run();
1268        let stopped = handle.snapshot();
1269        assert_eq!(stopped.root.state, NodeState::Registered);
1270        assert_eq!(stopped.root.process_id, running.root.process_id);
1271        assert_eq!(stopped.root.process_name, running.root.process_name);
1272        assert_eq!(stopped.root.started_at, running.root.started_at);
1273        assert_eq!(stopped.root.uptime_ms, None, "a node that isn't running has no uptime");
1274    }
1275}