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}