Skip to main content

saluki_core/runtime/
supervisor.rs

1use std::{
2    future::Future,
3    pin::Pin,
4    sync::{
5        atomic::{AtomicU64, AtomicUsize, Ordering},
6        Arc, Mutex,
7    },
8    time::Duration,
9};
10
11use saluki_common::sync::shutdown::ShutdownHandle;
12use saluki_error::GenericError;
13use serde::{Deserialize, Serialize};
14use snafu::{OptionExt as _, Snafu};
15use tokio::{pin, runtime::Handle, select, sync::mpsc};
16use tracing::{debug, error, warn};
17
18use super::{
19    dedicated::{spawn_dedicated_runtime, RuntimeConfiguration, RuntimeMode},
20    halt::HaltWatch,
21    restart::{RestartAction, RestartMode, RestartState, RestartStrategy, RestartType},
22    scope::{Adoption, Edge, UnadoptedScope},
23    supervisable::{InitializationError, ShutdownStrategy, Supervisable},
24    tree::{ChildFacts, ChildKey, NodeConfig, Roster, SupervisionTreeHandle, SupervisorNode},
25    worker_state::WorkerState,
26};
27use crate::runtime::{
28    process::{Process, ProcessExt as _},
29    state::DataspaceRegistry,
30};
31
32/// Process name segment used for a child whose own name can't be turned into a valid process name.
33///
34/// See [`SupervisedChild::create_process`].
35const UNNAMED_CHILD: &str = "unnamed";
36
37/// A `Future` that represents the full lifecycle of a worker, including initialization.
38///
39/// Unlike [`SupervisorFuture`], which only represents the runtime phase, this future first performs async
40/// initialization and then runs the worker. This allows initialization to happen concurrently when multiple workers are
41/// spawned, and keeps the supervisor loop responsive to shutdown signals during initialization.
42pub(super) type WorkerFuture = Pin<Box<dyn Future<Output = Result<(), WorkerError>> + Send>>;
43
44/// Worker lifecycle errors.
45///
46/// Distinguishes between initialization failures (which shouldn't trigger restart logic) and runtime failures (which
47/// are eligible for restart).
48#[derive(Debug)]
49pub(super) enum WorkerError {
50    /// The worker failed during async initialization.
51    ///
52    /// The optional `child_name` carries the name of the original failing child when the error originates from a
53    /// nested supervisor. This allows the parent to include it in its own `FailedToInitialize` error for better
54    /// diagnostics across supervision tree levels.
55    Initialization {
56        child_name: Option<String>,
57        source: InitializationError,
58    },
59
60    /// The worker failed during runtime execution.
61    Runtime(GenericError),
62
63    /// The worker was a nested supervisor that completed a requested shutdown after forcefully aborting one or more of
64    /// its own workers.
65    ///
66    /// Carried as a distinct variant (rather than collapsed into [`Runtime`][WorkerError::Runtime]) so the parent's
67    /// shutdown drain can recover the structured count and merge it into its own tally, aggregating forced aborts up
68    /// the supervision tree.
69    ShutdownTimedOut {
70        /// The number of workers the nested supervisor forcefully aborted, summed across its own supervision tree.
71        aborted: usize,
72    },
73}
74
75impl From<SupervisorError> for WorkerError {
76    fn from(err: SupervisorError) -> Self {
77        match err {
78            // Propagate initialization failures so the parent supervisor does NOT attempt to restart.
79            // Preserve the original child name so the parent can include it in diagnostics.
80            SupervisorError::FailedToInitialize { child_name, source } => WorkerError::Initialization {
81                child_name: Some(child_name),
82                source,
83            },
84            // Preserve the structured abort count so the parent can merge it into its own shutdown tally.
85            SupervisorError::ShutdownTimedOut { aborted } => WorkerError::ShutdownTimedOut { aborted },
86            // All other supervisor errors (shutdown, no children, invalid name) are runtime-level.
87            other => WorkerError::Runtime(other.into()),
88        }
89    }
90}
91
92/// Process errors.
93#[derive(Debug, Snafu)]
94pub enum ProcessError {
95    /// The child process was aborted by the supervisor.
96    #[snafu(display("Child process was aborted by the supervisor."))]
97    Aborted,
98
99    /// The child process panicked.
100    #[snafu(display("Child process panicked."))]
101    Panicked,
102
103    /// The child process terminated with an error.
104    #[snafu(display("Child process terminated with an error: {}", source))]
105    Terminated {
106        /// The error that caused the termination.
107        source: GenericError,
108    },
109}
110
111/// Policy for automatically shutting a supervisor down based on the termination of its _significant_ children.
112///
113/// A significant child (see [`ChildBuilder::with_significant`][crate::runtime::ChildBuilder::with_significant]) is one whose termination -- when it isn't restarted -- can
114/// drive the supervisor to shut down. This mirrors Erlang/OTP's `auto_shutdown` supervisor flag, and is how an
115/// unexpected (or intentional) child exit cascades into the supervisor stopping, and thus propagating up the tree,
116/// without that child being restarted.
117#[derive(Clone, Copy, Debug, PartialEq, Eq, Default, Deserialize, Serialize)]
118#[serde(rename_all = "snake_case")]
119pub enum AutoShutdown {
120    /// Never shut down automatically; significant children have no special effect. This is the default.
121    #[default]
122    Never,
123
124    /// Shut down as soon as _any_ significant child terminates without being restarted.
125    AnySignificant,
126
127    /// Shut down once _all_ significant children have terminated without being restarted.
128    AllSignificant,
129}
130
131/// Supervisor errors.
132#[derive(Debug, Snafu)]
133#[snafu(context(suffix(false)))]
134pub enum SupervisorError {
135    /// Supervisor or worker name is invalid.
136    #[snafu(display("Invalid name for supervisor or worker: '{}'", name))]
137    InvalidName {
138        /// The name of the supervisor is invalid.
139        name: String,
140    },
141
142    /// A child process failed to initialize.
143    ///
144    /// This error indicates that a child couldn't complete its async initialization. This is distinct from runtime
145    /// failures and doesn't trigger restart logic.
146    #[snafu(display("Child process '{}' failed to initialize: {}", child_name, source))]
147    FailedToInitialize {
148        /// The name of the child that failed to initialize.
149        child_name: String,
150
151        /// The underlying initialization error.
152        source: InitializationError,
153    },
154
155    /// The supervisor exceeded its restart limits and was forced to shutdown.
156    #[snafu(display("Supervisor has exceeded restart limits and was forced to shutdown."))]
157    Shutdown,
158
159    /// The supervisor shut down because a child that it cannot work without terminated.
160    ///
161    /// The supervisor stopped and drained its other children, because a [needed][crate::runtime::ChildBuilder::needed]
162    /// child terminated without a restart. A child marked significant can also cause this stop, if the supervisor is
163    /// configured with [`AutoShutdown`].
164    #[snafu(display("Supervisor shut down after a significant child terminated."))]
165    SignificantChildExited,
166
167    /// The supervisor completed a requested shutdown, but one or more workers ignored graceful shutdown and had to be
168    /// forcefully aborted after exceeding their shutdown timeout.
169    ///
170    /// The shutdown itself was requested and otherwise orderly; this variant exists so that having to forcefully stop a
171    /// worker is surfaced as a failure rather than reported as a clean shutdown. The count aggregates forced aborts
172    /// across the entire supervision tree: a parent merges in the counts reported by any child supervisors that also
173    /// timed out, so the value observed at the root supervisor is the total number of workers that had to be
174    /// force-stopped tree-wide.
175    #[snafu(display(
176        "Shutdown completed uncleanly: {} worker(s) were forcefully aborted after exceeding their shutdown timeout.",
177        aborted
178    ))]
179    ShutdownTimedOut {
180        /// The number of workers that had to be forcefully aborted.
181        aborted: usize,
182    },
183}
184
185/// A specification for a process to be added to a [`Supervisor`].
186///
187/// A child specification describes how the supervisor should create and manage a child: the underlying future that
188/// represents the process, along with metadata such as its name and shutdown strategy. All processes in a supervisor,
189/// whether a worker or a (nested) supervisor, are represented by a [`ChildSpecification`].
190///
191/// A specification is a description, not a control surface: it carries no public methods of its own. There are two
192/// ways to obtain one, matching the two levels of control:
193///
194/// - Pass a worker or supervisor directly to [`add_worker`][Supervisor::add_worker], [`spawn`][crate::runtime::spawn],
195///   or [`SupervisorHandle::spawn`], all of which accept a [`Supervisor`] or any [`Supervisable`] and convert it for
196///   you, applying the defaults appropriate to that kind of child -- including the shutdown strategy that lets a
197///   nested supervisor drain its whole subtree.
198/// - Configure one with [`ChildBuilder`][crate::runtime::ChildBuilder], which is the only way to set a restart policy,
199///   significance, placement, or a shutdown deadline. The builder exposes only the settings that make sense for the
200///   kind of child being described, and [`build`][crate::runtime::ChildBuilder::build] hands the result to
201///   [`add_worker`][Supervisor::add_worker].
202///
203/// A nested supervisor has fewer settings, which [`NestedSupervisorBuilder`][crate::runtime::NestedSupervisorBuilder]
204/// configures.
205pub struct ChildSpecification<S = WorkerSpec> {
206    spec_inner: S,
207}
208
209/// Child specification state for a worker.
210pub struct WorkerSpec {
211    worker: Arc<dyn Supervisable>,
212    options: ChildOptions,
213}
214
215/// Child specification state for a supervisor.
216pub struct SupervisorSpec {
217    supervisor: Supervisor,
218    options: ChildOptions,
219}
220
221// The configuration surface below is deliberately crate-internal: `ChildBuilder` is the public front end for all of
222// it, and is what decides which settings are offered for which kind of child. Keeping these methods off the public API
223// means a setting the builder refuses to express -- an explicit permanent restart policy, say -- can't be reached by
224// going around it. In-crate callers use them directly where the builder would be a layering inversion: this module's
225// own tests, which exercise the lowering these methods feed.
226impl ChildSpecification<WorkerSpec> {
227    /// Creates a specification for the given worker.
228    pub(crate) fn worker<T: Supervisable + 'static>(worker: T) -> Self {
229        Self {
230            spec_inner: WorkerSpec {
231                worker: Arc::new(worker),
232                options: ChildOptions::default(),
233            },
234        }
235    }
236
237    /// Gives this worker its own resource group, so that the worker does not count against the resource group of its
238    /// owner.
239    #[must_use]
240    pub(crate) fn with_own_resource_group(mut self) -> Self {
241        self.spec_inner.options.own_resource_group = true;
242        self
243    }
244
245    /// Sets an explicit restart policy for this worker, instead of a derived policy.
246    ///
247    /// If you do not set a policy, the policy comes from two facts: whether the worker can be started again, and
248    /// whether it is needed. See [`ChildBuilder`][crate::runtime::ChildBuilder].
249    ///
250    /// A worker that [cannot be started again][Supervisable::can_restart] is never restarted, so this policy has no
251    /// effect on it.
252    #[must_use]
253    pub(crate) fn with_restart_type(mut self, restart_type: RestartType) -> Self {
254        self.spec_inner.options.restart = Some(restart_type);
255        self
256    }
257
258    /// Sets whether this worker is _significant_.
259    ///
260    /// A significant worker's termination (when it isn't restarted) can drive the supervisor to shut down, per the
261    /// supervisor's [`AutoShutdown`] policy. Only meaningful for non-permanent workers, since a permanent worker is
262    /// always restarted and so never terminates without being restarted.
263    #[must_use]
264    pub(crate) fn with_significant(mut self, significant: bool) -> Self {
265        self.spec_inner.options.significant = significant;
266        self
267    }
268
269    /// Runs this worker on the given Tokio runtime rather than the supervisor's own runtime.
270    ///
271    /// By default, a worker runs on whatever runtime its supervisor runs on. Use this for compute-heavy workers that
272    /// shouldn't contend with the supervisor's runtime -- for example, a topology component offloading encoding work
273    /// onto a shared worker pool.
274    ///
275    /// Note that this only affects where the worker's task is spawned. Supervision itself -- shutdown signalling,
276    /// restart handling, and abort-on-timeout -- is unchanged, and is still driven from the supervisor's runtime.
277    #[must_use]
278    pub(crate) fn with_runtime(mut self, handle: Handle) -> Self {
279        self.spec_inner.options.runtime = Some(handle);
280        self
281    }
282
283    /// Overrides the shutdown strategy for this worker.
284    ///
285    /// By default, a worker's strategy comes from [`Supervisable::shutdown_strategy`], which itself defaults to
286    /// `Graceful(5s)`. Use this when the grace period depends on where the worker is used rather than on the worker
287    /// type: a worker that a component drains during its own shutdown needs at least as long as the component itself,
288    /// otherwise it is forcefully aborted while the component is still waiting on it.
289    #[must_use]
290    pub(crate) fn with_shutdown_strategy(mut self, strategy: ShutdownStrategy) -> Self {
291        self.spec_inner.options.shutdown = ChildShutdown::Explicit(strategy);
292        self
293    }
294
295    /// Sets whether this worker's owner can work without it. See [`Edge`].
296    #[must_use]
297    pub(crate) fn with_edge(mut self, edge: Edge) -> Self {
298        self.spec_inner.options.edge = Some(edge);
299        self
300    }
301
302    /// Makes a failure of this worker fail its owner, whatever its edge is.
303    ///
304    /// The owner is the process whose [`Scope`][super::Scope] contains the worker, or the supervisor of the worker.
305    ///
306    /// Use this for a worker that only holds children of its own, such as the host of a nested scope. This worker fails
307    /// only when one of its children fails, so its failure is the failure of its children. That failure must get to the
308    /// owner of the worker in the same way as a failure of those children. If the worker finishes cleanly, it is only
309    /// reaped.
310    #[must_use]
311    pub(crate) fn forwarding_failures(mut self) -> Self {
312        self.spec_inner.options.forwards_failures = true;
313        self
314    }
315
316    /// Makes `observer` receive how this worker ended, after its scope closed. See [`ExitObserver`].
317    #[cfg(any(test, feature = "test-util"))]
318    #[must_use]
319    pub(crate) fn observing_exit(mut self, observer: ExitObserver) -> Self {
320        self.spec_inner.options.exit_observer = Some(observer);
321        self
322    }
323
324    /// Makes this worker adopt `scope` as its own, so that the worker does not create a scope on demand.
325    ///
326    /// If children are spawned into `scope` before the worker starts, the scope holds them until the worker starts.
327    /// Then they start as children of the worker. Work can be spawned while something is built, before the process that
328    /// will own the work exists. This option is how that process becomes the owner of that work.
329    #[must_use]
330    pub(crate) fn adopting(mut self, scope: UnadoptedScope) -> Self {
331        self.spec_inner.options.adopt = Some(Adoption::new(scope));
332        self
333    }
334}
335
336// Crate-internal for the same reason as the worker surface above: `NestedSupervisorBuilder` is the public front end,
337// and going around it would allow combinations the builder refuses to express.
338//
339// Deliberately narrower than the worker surface, though. A nested supervisor bounds its own subtree through the
340// shutdown timeouts of its children: `SupervisedChild::shutdown_strategy` reports `Graceful(Duration::MAX)` for one.
341// Offering a shutdown setting here would let a caller truncate a drain the subtree is already responsible for, so there
342// isn't one. Placement is likewise absent: a nested supervisor runs wherever its parent does, and its children carry
343// their own placement.
344impl ChildSpecification<SupervisorSpec> {
345    /// Sets whether this nested supervisor's owner can work without it. See [`Edge`].
346    #[must_use]
347    pub(crate) fn with_edge(mut self, edge: Edge) -> Self {
348        self.spec_inner.options.edge = Some(edge);
349        self
350    }
351
352    /// Sets an explicit restart policy for this nested supervisor, instead of a derived policy.
353    #[must_use]
354    pub(crate) fn with_restart_type(mut self, restart_type: RestartType) -> Self {
355        self.spec_inner.options.restart = Some(restart_type);
356        self
357    }
358
359    /// Sets whether this nested supervisor is _significant_.
360    ///
361    /// A significant child's termination (when it isn't restarted) can drive the parent supervisor to shut down, per
362    /// the parent's [`AutoShutdown`] policy.
363    #[must_use]
364    pub(crate) fn with_significant(mut self, significant: bool) -> Self {
365        self.spec_inner.options.significant = significant;
366        self
367    }
368
369    /// Makes the scope that holds this nested supervisor tell it to stop as soon as the owner of the scope is told to
370    /// stop. See [`NestedSupervisorBuilder::stops_with_owner`][builder].
371    ///
372    /// [builder]: crate::runtime::NestedSupervisorBuilder::stops_with_owner
373    #[must_use]
374    pub(crate) fn stopping_with_owner(mut self) -> Self {
375        self.spec_inner.options.stops_with_owner = true;
376        self
377    }
378}
379
380impl<T> From<T> for ChildSpecification<WorkerSpec>
381where
382    T: Supervisable + 'static,
383{
384    fn from(worker: T) -> Self {
385        Self::worker(worker)
386    }
387}
388
389impl From<Supervisor> for ChildSpecification<SupervisorSpec> {
390    fn from(supervisor: Supervisor) -> Self {
391        Self {
392            spec_inner: SupervisorSpec {
393                supervisor,
394                options: ChildOptions::default(),
395            },
396        }
397    }
398}
399
400mod sealed {
401    pub trait Sealed {}
402}
403
404impl sealed::Sealed for WorkerSpec {}
405impl sealed::Sealed for SupervisorSpec {}
406
407/// Child specification state.
408///
409/// This trait is sealed -- it cannot be implemented outside of this crate -- and is implemented only for
410/// [`WorkerSpec`] and [`SupervisorSpec`]. It exists so that [`Supervisor::add_worker`] and
411/// [`SupervisorHandle::spawn`] can both accept a [`ChildSpecification`] in either state (as well as bare workers and
412/// supervisors) while lowering each into the supervisor's internal representation.
413pub trait ChildState: sealed::Sealed + Sized {
414    /// Lowers a specification into the supervisor's internal representation of a child.
415    ///
416    /// `registration` tells how the child is registered. If the specification does not say whether the child is needed
417    /// or hosted, `registration` decides it.
418    #[doc(hidden)]
419    fn into_child_parts(spec: ChildSpecification<Self>, registration: Registration) -> LoweredChild;
420}
421
422/// How a child is registered with its owner.
423#[doc(hidden)]
424#[derive(Clone, Copy, Debug, PartialEq, Eq)]
425pub enum Registration {
426    /// Declared with [`Supervisor::add_worker`] before its owner runs.
427    Declared,
428
429    /// Spawned while its owner runs.
430    Spawned,
431}
432
433/// A child specification lowered into the supervisor's internal representation.
434///
435/// Opaque to callers: it exists only to carry the output of [`ChildState::into_child_parts`] to the supervisor that
436/// registers the child, and is public only because [`ChildState`] is.
437pub struct LoweredChild {
438    spec: SupervisedChild,
439    config: ChildConfig,
440}
441
442impl LoweredChild {
443    /// Splits the lowered child into its runnable form and its configuration.
444    pub(super) fn into_parts(self) -> (SupervisedChild, ChildConfig) {
445        (self.spec, self.config)
446    }
447}
448
449impl ChildState for WorkerSpec {
450    fn into_child_parts(spec: ChildSpecification<Self>, registration: Registration) -> LoweredChild {
451        let WorkerSpec { worker, options } = spec.spec_inner;
452        let restartable = worker.can_restart();
453        LoweredChild {
454            spec: SupervisedChild::Worker(worker),
455            config: options.resolve(registration, restartable),
456        }
457    }
458}
459
460impl ChildState for SupervisorSpec {
461    fn into_child_parts(spec: ChildSpecification<Self>, registration: Registration) -> LoweredChild {
462        let SupervisorSpec { supervisor, options } = spec.spec_inner;
463        LoweredChild {
464            spec: SupervisedChild::Supervisor(supervisor),
465            // A supervisor builds its children again each time it starts, so it can always be started again.
466            config: options.resolve(registration, true),
467        }
468    }
469}
470
471/// The type-erased, runnable form of a child: either a worker or a nested supervisor.
472pub(super) enum SupervisedChild {
473    Worker(Arc<dyn Supervisable>),
474    Supervisor(Supervisor),
475}
476
477impl SupervisedChild {
478    /// Returns whether this child is a nested supervisor rather than a leaf worker.
479    pub(super) fn is_supervisor(&self) -> bool {
480        matches!(self, Self::Supervisor(_))
481    }
482
483    /// This is the link that makes the tree walkable: a parent records its child's node alongside its own, so an
484    /// observer holding the parent's node can descend into the child's.
485    pub(super) fn node(&self) -> Option<Arc<SupervisorNode>> {
486        match self {
487            Self::Worker(_) => None,
488            Self::Supervisor(supervisor) => Some(Arc::clone(&supervisor.node)),
489        }
490    }
491
492    fn process_type(&self) -> &'static str {
493        match self {
494            Self::Worker(_) => "worker",
495            Self::Supervisor(_) => "supervisor",
496        }
497    }
498
499    pub(super) fn name(&self) -> &str {
500        match self {
501            Self::Worker(worker) => worker.name(),
502            Self::Supervisor(supervisor) => &supervisor.supervisor_id,
503        }
504    }
505
506    /// Returns whether this child observes the shutdown signal it is given.
507    ///
508    /// Always true for a nested supervisor: the signal is how it learns to drain its own subtree, and a supervisor on
509    /// a dedicated runtime receives it across the thread boundary through `spawn_dedicated_runtime`, where aborting
510    /// the awaiting future wouldn't stop the runtime thread anyway.
511    pub(super) fn wants_shutdown_signal(&self) -> bool {
512        match self {
513            Self::Worker(worker) => worker.wants_shutdown_signal(),
514            Self::Supervisor(_) => true,
515        }
516    }
517
518    pub(super) fn shutdown_strategy(&self) -> ShutdownStrategy {
519        match self {
520            Self::Worker(worker) => worker.shutdown_strategy(),
521
522            // Supervisors should always be given as much time as necessary shutdown down gracefully to ensure that the
523            // entire supervision subtree can be shutdown cleanly.
524            Self::Supervisor(_) => ShutdownStrategy::Graceful(Duration::MAX),
525        }
526    }
527
528    /// Creates the process for this child under `parent_process`.
529    ///
530    /// A name that sanitizes to nothing at all (an empty string, or one made up entirely of separators) can't be used
531    /// as a process name. Rather than refuse to start the child -- which for a dynamically spawned child would mean
532    /// silently losing work that the caller was told had been accepted -- the child runs under
533    /// [`UNNAMED_CHILD`] instead, and the substitution is logged.
534    pub(super) fn create_process(&self, parent_process: &Process, own_resource_group: bool) -> Process {
535        let name = self.name();
536        let process = match self {
537            Self::Worker(_) if own_resource_group => Process::worker_with_own_resource_group(name, parent_process),
538            Self::Worker(_) => Process::worker(name, parent_process),
539            Self::Supervisor(_) => Process::supervisor(name, Some(parent_process)),
540        };
541
542        process.unwrap_or_else(|| {
543            warn!(
544                parent_process = parent_process.name(),
545                child_name = name,
546                "Child process name is not usable as a process name; falling back to '{}'.",
547                UNNAMED_CHILD
548            );
549
550            match self {
551                Self::Worker(_) => Process::worker(UNNAMED_CHILD, parent_process),
552                Self::Supervisor(_) => Process::supervisor(UNNAMED_CHILD, Some(parent_process)),
553            }
554            .expect("placeholder child name is always a valid process name")
555        })
556    }
557
558    /// Creates the future that runs this child.
559    ///
560    /// Only a nested supervisor uses `halt`. A worker is halted by the `drive` wrapper around its future instead.
561    pub(super) fn create_worker_future(
562        &self, process: Process, process_shutdown: ShutdownHandle, halt: HaltWatch,
563    ) -> Result<WorkerFuture, SupervisorError> {
564        match self {
565            Self::Worker(worker) => {
566                let worker = Arc::clone(worker);
567                Ok(Box::pin(async move {
568                    let run_future =
569                        worker
570                            .initialize(process_shutdown)
571                            .await
572                            .map_err(|source| WorkerError::Initialization {
573                                child_name: None,
574                                source,
575                            })?;
576                    run_future.await.map_err(WorkerError::Runtime)
577                }))
578            }
579            Self::Supervisor(sup) => {
580                match sup.runtime_mode() {
581                    RuntimeMode::Ambient => {
582                        // Run on the parent's ambient runtime.
583                        Ok(sup.as_nested_process(process, process_shutdown, halt))
584                    }
585                    RuntimeMode::Dedicated(config) => {
586                        // Spawn in a dedicated runtime on a new OS thread, passing the parent's
587                        // dataspace so the nested supervisor inherits it across the thread boundary.
588                        //
589                        // TODO: Only the dataspace is carried across, so the supervisor re-roots its own process name
590                        // when it starts (`run_with_shutdown_inner` passes no parent) rather than staying scoped
591                        // under us. That also leaves the process we build here registered as a resource group that
592                        // nothing ever enters, so it reads zero forever. Threading this process through instead would
593                        // fix both, at the cost of renaming the affected resource groups and their metric labels.
594                        let child_name = sup.supervisor_id.to_string();
595                        let dataspace = process.dataspace().clone();
596                        let handle = spawn_dedicated_runtime(
597                            sup.inner_clone(),
598                            config.clone(),
599                            process_shutdown,
600                            halt,
601                            dataspace,
602                        )
603                        .map_err(|e| SupervisorError::FailedToInitialize {
604                            child_name,
605                            source: e.into(),
606                        })?;
607
608                        Ok(Box::pin(async move { handle.await.map_err(WorkerError::from) }))
609                    }
610                }
611            }
612        }
613    }
614}
615
616impl Clone for SupervisedChild {
617    fn clone(&self) -> Self {
618        match self {
619            Self::Worker(worker) => Self::Worker(Arc::clone(worker)),
620            Self::Supervisor(supervisor) => Self::Supervisor(supervisor.inner_clone()),
621        }
622    }
623}
624
625/// How a child's shutdown strategy is determined.
626#[derive(Clone, Copy, Debug, Default)]
627pub(super) enum ChildShutdown {
628    /// Use whatever the worker reports through [`Supervisable::shutdown_strategy`]. This is the default.
629    #[default]
630    Worker,
631
632    /// Use this strategy, overriding whatever the worker reports.
633    Explicit(ShutdownStrategy),
634}
635
636impl ChildShutdown {
637    /// Resolves the shutdown strategy of a child from the strategy that the child reports for itself.
638    pub(super) fn resolve(self, worker_strategy: ShutdownStrategy) -> ShutdownStrategy {
639        match self {
640            Self::Worker => worker_strategy,
641            Self::Explicit(strategy) => strategy,
642        }
643    }
644}
645
646/// Per-child settings as configured on a [`ChildSpecification`], before they are resolved for a specific registration
647/// path.
648///
649/// This type is separate from [`ChildConfig`] because there is no single default for whether a child is needed. A child
650/// that can be started again and that is declared with [`Supervisor::add_worker`] before its owner runs is needed.
651/// Every other child is hosted. This type leaves the edge unset, so that both registration paths can use one
652/// specification type.
653#[derive(Clone, Debug, Default)]
654pub(super) struct ChildOptions {
655    /// An explicit restart policy, which replaces the derived policy.
656    restart: Option<RestartType>,
657    significant: bool,
658
659    /// Runtime to spawn the child on. `None` means the supervisor's own runtime.
660    runtime: Option<Handle>,
661
662    shutdown: ChildShutdown,
663
664    /// Whether the child's owner can work without it, if the specification sets it.
665    edge: Option<Edge>,
666
667    /// Whether a failure of the child fails its owner, whatever its edge is.
668    forwards_failures: bool,
669
670    /// A pre-created scope for the child to adopt as its own.
671    adopt: Option<Adoption>,
672
673    /// Whether the child owns a resource group of its own.
674    own_resource_group: bool,
675
676    /// Whether a nested supervisor is told to stop as soon as the owner of its scope is.
677    stops_with_owner: bool,
678
679    /// Receives how the worker ended, for the component test harness.
680    #[cfg(any(test, feature = "test-util"))]
681    exit_observer: Option<ExitObserver>,
682}
683
684impl ChildOptions {
685    /// Resolves these options into a concrete configuration for a child registered through `registration`.
686    ///
687    /// `restartable` tells whether the child can be started again. For a worker, this comes from
688    /// [`Supervisable::can_restart`].
689    ///
690    /// A child that does not say whether it is needed is needed if it can be started again and is declared before its
691    /// owner runs. Such a child is part of the purpose of its owner. Every other child that does not say is hosted.
692    ///
693    /// The deprecated builder methods can set an explicit restart policy, and explicit restart policies are older than
694    /// edges. Thus, a child with an explicit restart policy is hosted unless it says otherwise, as it was before edges.
695    ///
696    /// Then the restart policy comes from the edge and from whether the child can be started again, unless the child
697    /// has an explicit restart policy. See [`ChildBuilder`][builder].
698    ///
699    /// A child that can only be started once is never restarted, even if it has an explicit restart policy. The
700    /// restart would fail anyway, because the child cannot start again.
701    ///
702    /// [builder]: crate::runtime::ChildBuilder
703    fn resolve(self, registration: Registration, restartable: bool) -> ChildConfig {
704        let edge = self.edge.unwrap_or(
705            if registration == Registration::Declared && restartable && self.restart.is_none() {
706                Edge::Needed
707            } else {
708                Edge::Hosted
709            },
710        );
711        let restart = if restartable {
712            self.restart.unwrap_or(match edge {
713                Edge::Needed => RestartType::Permanent,
714                Edge::Hosted => RestartType::Transient,
715            })
716        } else {
717            RestartType::Temporary
718        };
719
720        ChildConfig {
721            restart,
722            significant: self.significant,
723            runtime: self.runtime,
724            shutdown: self.shutdown,
725            edge,
726            forwards_failures: self.forwards_failures,
727            adopt: self.adopt,
728            own_resource_group: self.own_resource_group,
729            stops_with_owner: self.stops_with_owner,
730            #[cfg(any(test, feature = "test-util"))]
731            exit_observer: self.exit_observer,
732        }
733    }
734}
735
736/// Per-child configuration.
737///
738/// The configuration of a child contains:
739///
740/// - the [`RestartType`] of the child
741/// - whether the child is _significant_ (see [`AutoShutdown`])
742/// - the runtime that the child runs on
743/// - how the shutdown strategy of the child is decided
744/// - how the termination of the child affects a scope that owns it
745/// - the scope that the child adopts, if there is one
746/// - whether a nested supervisor stops as soon as the owner of its scope is told to stop
747#[derive(Clone, Debug)]
748pub(super) struct ChildConfig {
749    restart: RestartType,
750    significant: bool,
751    runtime: Option<Handle>,
752    shutdown: ChildShutdown,
753    edge: Edge,
754    forwards_failures: bool,
755    adopt: Option<Adoption>,
756    own_resource_group: bool,
757    stops_with_owner: bool,
758    #[cfg(any(test, feature = "test-util"))]
759    exit_observer: Option<ExitObserver>,
760}
761
762impl ChildConfig {
763    /// Returns the child's restart policy.
764    pub(super) fn restart(&self) -> RestartType {
765        self.restart
766    }
767
768    /// Returns whether the child is significant to a supervisor.
769    pub(super) fn significant(&self) -> bool {
770        self.significant
771    }
772
773    /// Returns how the child's termination affects a scope that owns it.
774    pub(super) fn edge(&self) -> Edge {
775        self.edge
776    }
777
778    /// Returns whether a failure of the child fails its owner, whatever its edge is.
779    pub(super) fn forwards_failures(&self) -> bool {
780        self.forwards_failures
781    }
782
783    /// Returns the pre-created scope the child adopts as its own, if any.
784    pub(super) fn adopt(&self) -> Option<&Adoption> {
785        self.adopt.as_ref()
786    }
787
788    /// Returns whether the child owns a resource group of its own.
789    pub(super) fn own_resource_group(&self) -> bool {
790        self.own_resource_group
791    }
792
793    /// Returns whether the child is told to stop as soon as the owner of its scope is, rather than when the scope
794    /// closes.
795    pub(super) fn stops_with_owner(&self) -> bool {
796        self.stops_with_owner
797    }
798
799    /// Returns whether the child's termination can stop its owner.
800    ///
801    /// A snapshot reports this value as significance. It is true for a needed child that can terminate without a
802    /// restart, and for a child marked significant. A permanent child is always restarted. Thus, whatever its marks,
803    /// the termination of a permanent child never stops its owner.
804    pub(super) fn can_stop_owner(&self) -> bool {
805        self.restart != RestartType::Permanent && (self.significant || self.edge == Edge::Needed)
806    }
807
808    /// Returns the runtime the child should be spawned on, if it isn't the supervisor's own.
809    pub(super) fn runtime(&self) -> Option<&Handle> {
810        self.runtime.as_ref()
811    }
812
813    /// Returns how the child's shutdown strategy should be determined.
814    pub(super) fn shutdown(&self) -> ChildShutdown {
815        self.shutdown
816    }
817
818    /// Returns the observer that receives how the worker ended, if the test harness set one.
819    #[cfg(any(test, feature = "test-util"))]
820    pub(super) fn exit_observer(&self) -> Option<&ExitObserver> {
821        self.exit_observer.as_ref()
822    }
823}
824
825/// Receives how a worker ended, after its scope closed and its children stopped.
826///
827/// Only the component test harness uses this. When a needed child terminates, its supervisor reports only that, and not
828/// the error of the child. But the `run` of a component can return cleanly while a child that it needs fails as its
829/// scope closes, and then the error of that child is the result of the worker. The harness needs that error to return
830/// it.
831///
832/// The observer receives the result of the first start of the worker only, including a halt (unclean if overdue, clean
833/// if brutal). It receives nothing if the worker panics, or if its task is aborted along with its owner's.
834#[cfg(any(test, feature = "test-util"))]
835#[derive(Clone)]
836pub(crate) struct ExitObserver(Arc<Mutex<Option<ExitSender>>>);
837
838/// Receives how a worker ended, with a description of the error if it failed.
839#[cfg(any(test, feature = "test-util"))]
840type ExitSender = tokio::sync::oneshot::Sender<Result<(), String>>;
841
842#[cfg(any(test, feature = "test-util"))]
843impl ExitObserver {
844    /// Creates an observer that sends how the worker ended to `tx`, with a description of the error if it failed.
845    pub(crate) fn new(tx: ExitSender) -> Self {
846        Self(Arc::new(Mutex::new(Some(tx))))
847    }
848
849    /// Wraps the future of a worker, so that the observer receives its result.
850    pub(super) fn observe(&self, worker: WorkerFuture) -> WorkerFuture {
851        let tx = self.0.lock().unwrap().take();
852        Box::pin(async move {
853            let result = worker.await;
854            if let Some(tx) = tx {
855                let _ = tx.send(result.as_ref().map(|_| ()).map_err(|error| match error {
856                    WorkerError::Runtime(e) => format!("{e:#}"),
857                    WorkerError::Initialization { source, .. } => source.to_string(),
858                    WorkerError::ShutdownTimedOut { aborted } => {
859                        format!("stopped uncleanly after forcefully aborting {aborted} worker(s)")
860                    }
861                }));
862            }
863            result
864        })
865    }
866}
867
868#[cfg(any(test, feature = "test-util"))]
869impl std::fmt::Debug for ExitObserver {
870    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
871        f.debug_struct("ExitObserver").finish_non_exhaustive()
872    }
873}
874
875/// A registered child: its specification together with the configuration chosen at registration time.
876#[derive(Clone)]
877struct ChildEntry {
878    spec: SupervisedChild,
879    config: ChildConfig,
880    /// Whether this child was added dynamically (via [`SupervisorHandle`]) rather than statically before the run. Used
881    /// to maintain the dynamic-children gauge.
882    dynamic: bool,
883}
884
885/// Identifier for a child managed by a [`Supervisor`].
886///
887/// Returned by [`SupervisorHandle::spawn`] for dynamically spawned children. Unique within a single process for the
888/// lifetime of a supervisor run.
889#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
890pub struct ChildId(u64);
891
892impl ChildId {
893    /// Wraps a raw identifier.
894    pub(super) const fn from_raw(id: u64) -> Self {
895        Self(id)
896    }
897
898    /// Returns the raw numeric value of this identifier.
899    pub const fn as_u64(self) -> u64 {
900        self.0
901    }
902}
903
904/// A dynamic spawn request handed from a [`SupervisorHandle`] to the running supervisor.
905struct PendingSpawn {
906    id: u64,
907    spec: SupervisedChild,
908    config: ChildConfig,
909}
910
911/// Number of queued spawn requests the supervisor takes in one go before returning to its loop.
912///
913/// Draining in batches keeps a burst of spawns to a single wake-up rather than one per child, while still bounding how
914/// long the supervisor can spend registering children before it re-checks the rest of its loop (most importantly,
915/// shutdown).
916const SPAWN_DRAIN_BATCH: usize = 64;
917
918/// A handle for spawning dynamic children on a running [`Supervisor`].
919///
920/// Obtained from [`Supervisor::handle`]. Handles are cheap to clone and can be shared across tasks.
921///
922/// Spawning is synchronous and infallible, in the spirit of [`tokio::spawn`]: the child is queued for the running
923/// supervisor and the call returns immediately with the child's [`ChildId`]. Also as with [`tokio::spawn`], being
924/// accepted is not a promise of being run -- if the supervisor isn't running, or shuts down before it gets to the
925/// queued child, the child is never started at all.
926///
927/// # Ambient spawning
928///
929/// Code running under supervision usually doesn't need a handle at all: [`spawn`][crate::runtime::spawn] targets the
930/// supervisor of whatever process is currently running. Use a handle when spawning from outside supervision, or when
931/// targeting a supervisor other than the ambient one. [`scope`][Self::scope] bridges the two by making a handle the
932/// ambient supervisor for a future.
933#[derive(Clone)]
934pub struct SupervisorHandle {
935    name: Arc<str>,
936    // The currently running supervisor publishes its spawn queue here so handles can reach the live run; it's cleared
937    // when no run is active, at which point spawns are accepted and dropped.
938    current_tx: Arc<Mutex<Option<mpsc::UnboundedSender<PendingSpawn>>>>,
939    id_counter: Arc<AtomicU64>,
940    active: Arc<AtomicUsize>,
941}
942
943impl SupervisorHandle {
944    /// Returns the name of the supervisor this handle refers to.
945    pub fn name(&self) -> &str {
946        &self.name
947    }
948
949    /// Spawns a new dynamic child.
950    ///
951    /// Accepts anything [`Supervisor::add_worker`] accepts: a bare [`Supervisable`], a [`Supervisor`] to run as a
952    /// nested supervision subtree, or a [`ChildSpecification`] configured in detail.
953    ///
954    /// Unless [`ChildBuilder`][crate::runtime::ChildBuilder] says otherwise, a dynamic child is hosted. A hosted child
955    /// that can be started again, such as most [`Supervisable`]s and every [`Supervisor`], is restarted only if it
956    /// fails. A child that [cannot be started again][Supervisable::can_restart], such as an
957    /// [`FnWorker`][crate::runtime::FnWorker], is never restarted. See the
958    /// [runtime documentation][crate::runtime#restart-policy].
959    ///
960    /// A dynamic child is not restored when the supervisor itself restarts. The supervisor stops the child when the
961    /// supervisor restarts or terminates.
962    ///
963    /// The returned [`ChildId`] identifies the child for the lifetime of the supervisor run. The child is queued
964    /// rather than started synchronously, so it may not have begun running by the time this returns; if the supervisor
965    /// isn't running, or shuts down before reaching the child, it never runs at all.
966    pub fn spawn<S, T>(&self, child: T) -> ChildId
967    where
968        S: ChildState,
969        T: Into<ChildSpecification<S>>,
970    {
971        let LoweredChild { spec, config } = S::into_child_parts(child.into(), Registration::Spawned);
972
973        // Take the id before we try to enqueue: the caller gets a stable identifier either way, and ids are only
974        // meaningful within a run.
975        let id = self.id_counter.fetch_add(1, Ordering::Relaxed);
976        let pending = PendingSpawn { id, spec, config };
977
978        // Clone the sender out from under the lock rather than sending while holding the guard.
979        //
980        // The queue behind it is unbounded on purpose: a queued child is a child that will be started, so the only
981        // thing a depth limit could buy is discarding work the caller was told had been accepted. A backlog only forms
982        // while the supervisor can't drain -- mid-restart, or mid-drain -- and holding it until it can is the whole
983        // point.
984        let tx = self.current_tx.lock().unwrap().clone();
985        match tx {
986            // Racing a teardown is normal rather than exceptional -- a source that spawns a child per connection will
987            // do it every time it is shut down mid-accept -- so this stays at debug level.
988            Some(tx) => {
989                if let Err(e) = tx.send(pending) {
990                    debug!(
991                        supervisor_id = %self.name,
992                        child_name = e.0.spec.name(),
993                        "Supervisor is shutting down; dynamic child will not be started."
994                    );
995                }
996            }
997            // Spawning against a supervisor that never ran, on the other hand, is a wiring mistake: nothing about the
998            // program's normal operation produces it, and the child is silently lost.
999            None => warn!(
1000                supervisor_id = %self.name,
1001                child_name = pending.spec.name(),
1002                "Supervisor is not running; dynamic child will not be started."
1003            ),
1004        }
1005
1006        ChildId(id)
1007    }
1008
1009    /// Returns whether the supervisor is currently running.
1010    pub fn is_running(&self) -> bool {
1011        self.current_tx.lock().unwrap().is_some()
1012    }
1013
1014    /// Returns the number of dynamic children currently running under the supervisor.
1015    ///
1016    /// Counts children the supervisor has actually started, so a child that has been spawned but not yet picked up
1017    /// isn't included yet.
1018    pub fn active_children(&self) -> usize {
1019        self.active.load(Ordering::Relaxed)
1020    }
1021}
1022
1023/// Supervises a set of workers.
1024///
1025/// # Workers
1026///
1027/// All workers are defined through implementation of the [`Supervisable`] trait, which provides the logic for both
1028/// creating the underlying worker future that's spawned, as well as other metadata, such as the worker's name, how the
1029/// worker should be shutdown, and so on.
1030///
1031/// Supervisors also (indirectly) implement the [`Supervisable`] trait, allowing them to be supervised by other
1032/// supervisors in order to construct _supervision trees_.
1033///
1034/// # Instrumentation
1035///
1036/// Supervisors automatically create their own allocation group
1037/// ([`TrackingAllocator`][saluki_common::resource_tracking::TrackingAllocator]), which is used to track both the memory
1038/// usage of the supervisor itself and its children. Additionally, individual worker processes are wrapped in a
1039/// dedicated [`tracing::Span`] to allow tracing the causal relationship between arbitrary code and the worker executing
1040/// it, and statistics about task polls (poll count, poll duration) are collected.
1041///
1042/// # Restart Strategies
1043///
1044/// As the main purpose of a supervisor, restart behavior is fully configurable. A number of restart strategies are
1045/// available, which generally relate to the purpose of the supervisor: whether the workers being managed are
1046/// independent or interdependent.
1047///
1048/// All restart strategies are configured through [`RestartStrategy`], which has more information on the available
1049/// strategies and configuration settings.
1050pub struct Supervisor {
1051    supervisor_id: Arc<str>,
1052    child_specs: Vec<ChildEntry>,
1053    runtime_mode: RuntimeMode,
1054    // Shared across clones (a nested supervisor is cloned each time it runs) and across all handles. While a run is
1055    // active it holds that run's spawn queue so handles can reach the live supervisor; it's `None` whenever no run is
1056    // active, at which point spawned children are dropped rather than queued. Doubles as the `is_running` signal.
1057    current_tx: Arc<Mutex<Option<mpsc::UnboundedSender<PendingSpawn>>>>,
1058    id_counter: Arc<AtomicU64>,
1059    // Number of dynamic children currently running, shared with handles so it can be surfaced as a gauge.
1060    active: Arc<AtomicUsize>,
1061    // Observable bookkeeping for this supervisor, shared with clones (so every generation writes to the same place),
1062    // with this supervisor's parent (so the tree can be walked downward), and with any tree handle.
1063    node: Arc<SupervisorNode>,
1064}
1065
1066impl Supervisor {
1067    /// Creates an empty `Supervisor` with the default restart strategy.
1068    pub fn new<S: AsRef<str>>(supervisor_id: S) -> Result<Self, SupervisorError> {
1069        // We try to throw an error about invalid names as early as possible. This is a manual check, so we might still
1070        // encounter an error later when actually running the supervisor, but this is a good first step to catch the
1071        // bulk of invalid names.
1072        if supervisor_id.as_ref().is_empty() {
1073            return Err(SupervisorError::InvalidName {
1074                name: supervisor_id.as_ref().to_string(),
1075            });
1076        }
1077
1078        let supervisor_id: Arc<str> = supervisor_id.as_ref().into();
1079
1080        Ok(Self {
1081            node: Arc::new(SupervisorNode::new(Arc::clone(&supervisor_id))),
1082            supervisor_id,
1083            child_specs: Vec::new(),
1084            runtime_mode: RuntimeMode::default(),
1085            current_tx: Arc::new(Mutex::new(None)),
1086            id_counter: Arc::new(AtomicU64::new(0)),
1087            active: Arc::new(AtomicUsize::new(0)),
1088        })
1089    }
1090
1091    /// Returns the supervisor's ID.
1092    pub fn id(&self) -> &str {
1093        &self.supervisor_id
1094    }
1095
1096    /// Sets the restart strategy for the supervisor.
1097    pub fn with_restart_strategy(self, strategy: RestartStrategy) -> Self {
1098        self.node.update_config(|config| config.restart_strategy = strategy);
1099        self
1100    }
1101
1102    /// Sets the supervisor's automatic-shutdown policy.
1103    ///
1104    /// Controls whether the termination of _significant_ children causes the supervisor to shut down. The default is
1105    /// [`AutoShutdown::Never`].
1106    ///
1107    /// A supervisor always stops when a [needed][crate::runtime::ChildBuilder::needed] child terminates without a
1108    /// restart. In the past, that was the purpose of this policy.
1109    #[deprecated(note = "mark children the supervisor can't work without `needed` instead")]
1110    pub fn with_auto_shutdown(self, auto_shutdown: AutoShutdown) -> Self {
1111        self.node.update_config(|config| config.auto_shutdown = auto_shutdown);
1112        self
1113    }
1114
1115    /// Returns a handle for spawning dynamic children on this supervisor while it runs.
1116    ///
1117    /// The handle can be created before the supervisor starts and cloned freely. Spawning through it always succeeds,
1118    /// but a child is only ever started while the supervisor is actually running: one spawned before the supervisor
1119    /// starts, or after it has shut down, is accepted and then dropped.
1120    pub fn handle(&self) -> SupervisorHandle {
1121        SupervisorHandle {
1122            name: Arc::clone(&self.supervisor_id),
1123            current_tx: Arc::clone(&self.current_tx),
1124            id_counter: Arc::clone(&self.id_counter),
1125            active: Arc::clone(&self.active),
1126        }
1127    }
1128
1129    /// Returns a read-only handle for taking snapshots of this supervisor and the subtree beneath it.
1130    ///
1131    /// The handle can be created before the supervisor starts and remains valid across every restart of it. Unlike
1132    /// [`handle`][Self::handle], it grants no ability to affect the supervisor -- only to observe it.
1133    pub fn tree_handle(&self) -> SupervisionTreeHandle {
1134        SupervisionTreeHandle::new(Arc::clone(&self.node))
1135    }
1136
1137    /// Configures this supervisor to run in a dedicated runtime.
1138    ///
1139    /// When this supervisor is added as a child to another supervisor, it will spawn its own OS threads and Tokio
1140    /// runtime instead of running on the parent's ambient runtime.
1141    ///
1142    /// This provides runtime isolation, which can be useful for:
1143    /// - CPU-bound work that shouldn't block the parent's runtime
1144    /// - Isolating failures in one part of the system
1145    /// - Using different runtime configurations (for example, single-threaded vs multi-threaded)
1146    pub fn with_dedicated_runtime(mut self, config: RuntimeConfiguration) -> Self {
1147        let worker_threads = config.worker_threads();
1148        self.node
1149            .update_config(|node_config| node_config.dedicated_threads = Some(worker_threads));
1150        self.runtime_mode = RuntimeMode::Dedicated(config);
1151        self
1152    }
1153
1154    /// Returns the runtime mode for this supervisor.
1155    pub(crate) fn runtime_mode(&self) -> &RuntimeMode {
1156        &self.runtime_mode
1157    }
1158
1159    /// Adds a worker (or nested supervisor) to the supervisor.
1160    ///
1161    /// A worker can be anything that implements the [`Supervisable`] trait. A [`Supervisor`] can also be added as a
1162    /// worker and managed in a nested fashion, known as a supervision tree.
1163    ///
1164    /// Anything that needs configuring -- a restart policy, significance, placement, a shutdown deadline -- is
1165    /// described with [`ChildBuilder`][crate::runtime::ChildBuilder] and handed over via
1166    /// [`build`][crate::runtime::ChildBuilder::build]. See [`ChildSpecification`] for how children are represented
1167    /// internally.
1168    pub fn add_worker<S, T>(&mut self, child: T)
1169    where
1170        S: ChildState,
1171        T: Into<ChildSpecification<S>>,
1172    {
1173        let LoweredChild { spec, config } = S::into_child_parts(child.into(), Registration::Declared);
1174        self.push_child(ChildEntry {
1175            spec,
1176            config,
1177            dynamic: false,
1178        });
1179    }
1180
1181    /// Warns when a child was marked significant but nothing will act on it.
1182    ///
1183    /// Significance only has an effect for a child that can terminate without being restarted, under a supervisor
1184    /// whose [`AutoShutdown`] policy isn't [`Never`][AutoShutdown::Never]. Either mismatch makes the flag inert, which
1185    /// is worth saying out loud: a caller who marked a child significant is asserting that its termination matters,
1186    /// and silently ignoring that is how a supervisor ends up outliving something it can't work without.
1187    ///
1188    /// Called as children are started rather than as they are registered, because the policy half of the question
1189    /// isn't answerable any earlier: [`with_auto_shutdown`][Self::with_auto_shutdown] consumes the supervisor while
1190    /// [`add_worker`][Self::add_worker] borrows it, so a caller is free to add children first and set the policy
1191    /// afterwards. Checking at registration time would flag that -- entirely correct -- ordering as a mistake.
1192    ///
1193    /// Warn-only: the child still starts, since an inert flag is useless rather than unsafe.
1194    fn warn_if_significance_is_inert(&self, config: &ChildConfig, child_name: &str, auto_shutdown: AutoShutdown) {
1195        if !config.significant {
1196            return;
1197        }
1198
1199        if config.restart == RestartType::Permanent {
1200            warn!(
1201                supervisor_id = %self.supervisor_id,
1202                child_name,
1203                "Child is marked significant but is permanent, so it is always restarted and the flag has no effect."
1204            );
1205        }
1206
1207        if auto_shutdown == AutoShutdown::Never {
1208            warn!(
1209                supervisor_id = %self.supervisor_id,
1210                child_name,
1211                "Child is marked significant but the supervisor's auto-shutdown policy is `Never`, so the flag has \
1212                 no effect."
1213            );
1214        }
1215    }
1216
1217    fn push_child(&mut self, entry: ChildEntry) {
1218        debug!(
1219            supervisor_id = %self.supervisor_id,
1220            "Adding new static child process #{}. ({}, {}, {:?})",
1221            self.child_specs.len(),
1222            entry.spec.process_type(),
1223            entry.spec.name(),
1224            entry.config,
1225        );
1226
1227        // A child can be marked significant and still be permanent. The deprecated `with_significant` is available on
1228        // every builder, and a child that can start again and is declared here is permanent by default. Such a flag has
1229        // no effect, and `warn_if_significance_is_inert` reports it when the supervisor starts the child.
1230        self.child_specs.push(entry);
1231    }
1232
1233    /// Assembled here rather than inside the roster because only the supervisor can see a child's specification and
1234    /// resolved configuration.
1235    fn child_facts(entry: &ChildEntry, key: ChildKey) -> ChildFacts {
1236        ChildFacts {
1237            key,
1238            name: entry.spec.name().into(),
1239            node: entry.spec.node(),
1240            restart: entry.config.restart,
1241            significant: entry.config.can_stop_owner(),
1242        }
1243    }
1244
1245    fn spawn_static_children(
1246        &self, roster: &mut Roster<ChildEntry>, worker_state: &mut WorkerState, auto_shutdown: AutoShutdown,
1247    ) -> Result<(), SupervisorError> {
1248        debug!(supervisor_id = %self.supervisor_id, "Spawning all static child processes.");
1249        for (index, entry) in self.child_specs.iter().enumerate() {
1250            self.warn_if_significance_is_inert(&entry.config, entry.spec.name(), auto_shutdown);
1251
1252            let id = self.id_counter.fetch_add(1, Ordering::Relaxed);
1253            let started = worker_state.add_worker(id, &entry.spec, &entry.config)?;
1254            let facts = Self::child_facts(entry, ChildKey::Static(index));
1255            roster.insert(id, entry.clone(), facts, started);
1256        }
1257
1258        Ok(())
1259    }
1260
1261    /// Respawns children after a one-for-all restart, honoring each child's [`RestartType`].
1262    ///
1263    /// Every child except [`RestartType::Temporary`] is restarted, matching Erlang/OTP: a group restart restarts all
1264    /// permanent and transient children -- regardless of how they last exited, including a transient child that had
1265    /// already exited cleanly -- but never temporary children, which are shut down with the group and not brought back.
1266    /// A transient child's "restart only on abnormal exit" rule governs its _own_ termination, not a group restart
1267    /// driven by a sibling. Dynamic children are not restored (they are lost on a supervisor-level restart).
1268    fn respawn_children_one_for_all(
1269        &self, roster: &mut Roster<ChildEntry>, worker_state: &mut WorkerState,
1270    ) -> Result<(), SupervisorError> {
1271        debug!(supervisor_id = %self.supervisor_id, "Restarting all eligible static child processes.");
1272        for (index, entry) in self.child_specs.iter().enumerate() {
1273            // Temporary children are never restarted by a group restart (matching OTP): they are shut down with the
1274            // group but not brought back.
1275            if entry.config.restart == RestartType::Temporary {
1276                continue;
1277            }
1278            let id = self.id_counter.fetch_add(1, Ordering::Relaxed);
1279            let started = worker_state.add_worker(id, &entry.spec, &entry.config)?;
1280            // Keyed by position in the static child list rather than by roster id: a group restart hands out fresh
1281            // ids, and the child's creation time and restart count have to survive that.
1282            let facts = Self::child_facts(entry, ChildKey::Static(index));
1283            roster.insert(id, entry.clone(), facts, started);
1284        }
1285
1286        Ok(())
1287    }
1288
1289    /// Registers one dynamic child into the running supervisor's worker set and roster.
1290    fn spawn_dynamic_child(
1291        &self, spawn: PendingSpawn, worker_state: &mut WorkerState, roster: &mut Roster<ChildEntry>,
1292        significant_remaining: &mut usize, auto_shutdown: AutoShutdown,
1293    ) {
1294        let PendingSpawn { id, spec, config } = spawn;
1295        let entry = ChildEntry {
1296            spec,
1297            config,
1298            dynamic: true,
1299        };
1300        self.warn_if_significance_is_inert(&entry.config, entry.spec.name(), auto_shutdown);
1301
1302        match worker_state.add_worker(id, &entry.spec, &entry.config) {
1303            Ok(started) => {
1304                if entry.config.significant {
1305                    *significant_remaining += 1;
1306                }
1307                self.active.fetch_add(1, Ordering::Relaxed);
1308                let facts = Self::child_facts(&entry, ChildKey::Dynamic(id));
1309                roster.insert(id, entry, facts, started);
1310            }
1311            Err(e) => {
1312                // The only way registration fails now that child names always resolve is a nested supervisor on a
1313                // dedicated runtime failing to get an OS thread. There's no caller left to report it to -- spawning is
1314                // infallible -- so the child is dropped and the failure is logged here.
1315                error!(
1316                    supervisor_id = %self.supervisor_id,
1317                    child_name = entry.spec.name(),
1318                    error = %e,
1319                    "Failed to start dynamic child."
1320                );
1321            }
1322        }
1323    }
1324
1325    async fn run_inner(
1326        &self, process: Process, process_shutdown: ShutdownHandle, halt: HaltWatch,
1327    ) -> Result<(), SupervisorError> {
1328        // Publish a fresh spawn queue for this run so handles can spawn dynamic children into it; while it's set,
1329        // handles observe us as running.
1330        let (cmd_tx, cmd_rx) = mpsc::unbounded_channel();
1331        *self.current_tx.lock().unwrap() = Some(cmd_tx);
1332
1333        // Record the process we're actually running under. It has to come from here rather than from whatever our
1334        // parent created for us, because a supervisor on a dedicated runtime re-roots its process name when it
1335        // starts, and only the resulting name matches the resource group its allocations land in.
1336        self.node.begin_run(&process);
1337
1338        let result = self.supervise(process, process_shutdown, halt, cmd_rx).await;
1339
1340        // The run is over. Clear the sender so later spawns are dropped rather than queued, and reset the
1341        // dynamic-children gauge. Dropping the receiver (owned by `supervise`) already discarded anything in flight.
1342        *self.current_tx.lock().unwrap() = None;
1343        self.active.store(0, Ordering::Relaxed);
1344
1345        // Report as stopped rather than presenting the children of a generation that has ended. Every way out of
1346        // `supervise` -- clean shutdown, a child failing to initialize, the restart limit being exceeded, a
1347        // significant child exiting -- returns through here, so this covers all of them.
1348        self.node.end_run();
1349
1350        result
1351    }
1352
1353    async fn supervise(
1354        &self, process: Process, process_shutdown: ShutdownHandle, mut halt: HaltWatch,
1355        mut cmd_rx: mpsc::UnboundedReceiver<PendingSpawn>,
1356    ) -> Result<(), SupervisorError> {
1357        // Read once: configuration can't change while a run is in flight, and this keeps the reap and spawn arms
1358        // below off the node's lock entirely.
1359        let NodeConfig {
1360            restart_strategy,
1361            auto_shutdown,
1362            ..
1363        } = self.node.config();
1364
1365        let mut restart_state = RestartState::new(restart_strategy);
1366        let mut worker_state = WorkerState::new(process, self.handle(), Arc::clone(&self.node));
1367
1368        // The live roster of children -- both static (seeded below) and dynamic (added via the handle) -- keyed by a
1369        // stable id. A restart re-runs a child by id; a child that isn't restarted is removed from the roster.
1370        //
1371        // Every change also lands in this supervisor's supervision-tree bookkeeping, which is what makes the tree
1372        // observable. Going through the roster for all of it is what keeps the two from drifting apart.
1373        let mut roster = Roster::new(Arc::clone(&self.node));
1374
1375        // Spawn the static children. Initialization is folded into each worker's task, so this returns immediately --
1376        // children initialize concurrently in the background.
1377        self.spawn_static_children(&mut roster, &mut worker_state, auto_shutdown)?;
1378
1379        // Track how many significant children are still running, for `AutoShutdown` evaluation.
1380        let mut significant_remaining = roster.values().filter(|entry| entry.config.significant).count();
1381
1382        // Scratch space reused across every batched drain of the spawn queue.
1383        let mut spawn_batch = Vec::with_capacity(SPAWN_DRAIN_BATCH);
1384
1385        // Now we supervise.
1386        pin!(process_shutdown);
1387
1388        let outcome = loop {
1389            select! {
1390                // Shutdown first, then reaping, then taking on new work -- so neither a flood of spawns nor a stream
1391                // of exiting children can starve anything ranked above it.
1392                biased;
1393
1394                // Halted by our owner: the teardown below sees this, and halts children instead of signalling them.
1395                _ = halt.wait() => break Ok(()),
1396
1397                // Shutdown has been triggered; break out of the loop with a clean outcome and tear down below. (We
1398                // can't touch `cmd_rx` in any arm's handler -- the `recv_many` arm below borrows it for the whole
1399                // `select!` -- so all teardown happens after the loop.)
1400                _ = &mut process_shutdown => break Ok(()),
1401
1402                // Reaping outranks taking on new work: it is the only place children leave the join set, the roster
1403                // and the `active` gauge, so a steady stream of spawns must not be able to starve it. It parks
1404                // whenever there are no children, so it can't starve spawning in return.
1405                (child_id, worker_result) = worker_state.wait_for_next_worker() => {
1406                    // Pull out what we need from the roster before we mutate it.
1407                    let (child_name, config, dynamic) = {
1408                        let entry = roster.get(child_id).expect("completed worker must be present in the roster");
1409                        (entry.spec.name().to_string(), entry.config.clone(), entry.dynamic)
1410                    };
1411
1412                    // Initialization failures are not eligible for restart -- they propagate immediately.
1413                    if let Err(WorkerError::Initialization { child_name: inner, source }) = worker_result {
1414                        // If the error came from a nested supervisor, include the original child name to make the error
1415                        // chain more informative (e.g., "ctrl-pln/privileged-api").
1416                        let full_name = match inner {
1417                            Some(inner) => format!("{}/{}", child_name, inner),
1418                            None => child_name.clone(),
1419                        };
1420
1421                        error!(supervisor_id = %self.supervisor_id, worker_name = full_name, "Child process failed to initialize: {}", source);
1422                        break Err(SupervisorError::FailedToInitialize { child_name: full_name, source });
1423                    }
1424
1425                    // A worker exited abnormally if it returned an error, panicked, or was aborted; a clean exit is
1426                    // `Ok(())`. Together with the worker's restart policy, this determines whether we restart it.
1427                    let abnormal = worker_result.is_err();
1428                    let worker_result = worker_result.map_err(|e| match e {
1429                        WorkerError::Runtime(e) => ProcessError::Terminated { source: e },
1430                        WorkerError::Initialization { .. } => unreachable!("handled above"),
1431                        // A nested supervisor only reports `ShutdownTimedOut` while draining, which is driven by its own
1432                        // `process_shutdown` -- and that fires only when *this* supervisor is itself draining it, i.e.
1433                        // from `shutdown_workers` below, never from this main-loop arm. Treat it as a runtime
1434                        // termination defensively rather than asserting unreachable.
1435                        WorkerError::ShutdownTimedOut { aborted } => ProcessError::Terminated {
1436                            source: SupervisorError::ShutdownTimedOut { aborted }.into(),
1437                        },
1438                    });
1439
1440                    if !config.restart.should_restart(abnormal) {
1441                        // Not eligible for restart given how it exited. Drop it from the roster, and free its slot/gauge
1442                        // if it was dynamic. Crucially, we do NOT consult `evaluate_restart` here: non-restarts must not
1443                        // consume the restart-intensity budget, otherwise a steady stream of terminating temporary
1444                        // children would eventually trip the limit and tear the supervisor (and its siblings) down.
1445                        //
1446                        // An abnormal exit is reported at `warn` rather than `debug`: a child that isn't restarted --
1447                        // every dynamically-spawned child, in practice -- has no other path back to its owner, so this
1448                        // is the only place its failure is surfaced.
1449                        if abnormal {
1450                            warn!(supervisor_id = %self.supervisor_id, worker_name = %child_name, restart = ?config.restart, ?worker_result, "Child process exited with an error and is not eligible for restart.");
1451                        } else {
1452                            debug!(supervisor_id = %self.supervisor_id, worker_name = %child_name, restart = ?config.restart, "Child process exited and is not eligible for restart.");
1453                        }
1454                        roster.remove(child_id);
1455                        if dynamic {
1456                            self.active.fetch_sub(1, Ordering::Relaxed);
1457                        }
1458
1459                        // If a needed child terminates without a restart, the supervisor also stops. The supervisor
1460                        // cannot do its work without the child. It stops, and its own supervisor decides what to do
1461                        // next.
1462                        if config.edge == Edge::Needed {
1463                            warn!(supervisor_id = %self.supervisor_id, worker_name = %child_name, ?worker_result, "Needed child terminated; shutting down supervisor.");
1464                            break Err(SupervisorError::SignificantChildExited);
1465                        }
1466
1467                        // The supervisor also stops if a child that only holds children of its own fails. An example is
1468                        // the host of a pre-created scope. That child fails only when one of its children fails, and
1469                        // their failures must go up the tree. They must not stop at the host.
1470                        if abnormal && config.forwards_failures {
1471                            warn!(supervisor_id = %self.supervisor_id, worker_name = %child_name, ?worker_result, "Child failed on behalf of its own children; shutting down supervisor.");
1472                            break Err(SupervisorError::SignificantChildExited);
1473                        }
1474
1475                        // A significant child terminating without restart can drive the supervisor to shut down, per its
1476                        // `AutoShutdown` policy -- cascading an unexpected (or intentional) child exit into the
1477                        // supervisor stopping and propagating up the tree.
1478                        if config.significant {
1479                            significant_remaining = significant_remaining.saturating_sub(1);
1480                            let auto_shutdown = match auto_shutdown {
1481                                AutoShutdown::Never => false,
1482                                AutoShutdown::AnySignificant => true,
1483                                AutoShutdown::AllSignificant => significant_remaining == 0,
1484                            };
1485                            if auto_shutdown {
1486                                warn!(supervisor_id = %self.supervisor_id, worker_name = %child_name, ?worker_result, "Significant child terminated; shutting down supervisor.");
1487                                break Err(SupervisorError::SignificantChildExited);
1488                            }
1489                        }
1490                    } else {
1491                        match restart_state.evaluate_restart() {
1492                            RestartAction::Restart(mode) => match mode {
1493                                RestartMode::OneForOne => {
1494                                    warn!(supervisor_id = %self.supervisor_id, worker_name = %child_name, ?worker_result, "Child process terminated, restarting.");
1495                                    let spec = roster.get(child_id).expect("present for restart").spec.clone();
1496                                    match worker_state.add_worker(child_id, &spec, &config) {
1497                                        Ok(started) => roster.restart_in_place(child_id, started),
1498                                        Err(e) => break Err(e),
1499                                    }
1500                                }
1501                                RestartMode::OneForAll => {
1502                                    warn!(supervisor_id = %self.supervisor_id, worker_name = %child_name, ?worker_result, "Child process terminated, restarting all processes.");
1503                                    // This drain is part of a restart, not a shutdown: any forced aborts here are
1504                                    // already logged per-worker, and the supervisor keeps running, so the count does
1505                                    // not feed the unclean-shutdown signal.
1506                                    let _ = worker_state.shutdown_workers(&mut halt).await;
1507                                    // A one-for-all restart resets to the static roster; dynamic children are not
1508                                    // restored (they're lost on a supervisor-level restart, matching Erlang/OTP), and
1509                                    // temporary children are not restarted.
1510                                    roster.clear_for_group_restart();
1511                                    self.active.store(0, Ordering::Relaxed);
1512                                    let respawn = self.respawn_children_one_for_all(&mut roster, &mut worker_state);
1513                                    if let Err(e) = respawn {
1514                                        break Err(e);
1515                                    }
1516                                    significant_remaining =
1517                                        roster.values().filter(|entry| entry.config.significant).count();
1518                                }
1519                            },
1520                            // The edge of the child has no effect here. A child that continues to fail makes this
1521                            // supervisor stop, so that its own supervisor can try again.
1522                            RestartAction::Shutdown => {
1523                                error!(supervisor_id = %self.supervisor_id, worker_name = %child_name, ?worker_result, "Supervisor shutting down due to restart limits.");
1524                                break Err(SupervisorError::Shutdown);
1525                            }
1526                        }
1527                    }
1528                }
1529
1530                // A handle asked us to spawn one or more dynamic children. The published sender keeps the queue open
1531                // for the whole run, so this only yields zero once we close it during teardown. Draining in batches
1532                // keeps a burst of spawns to a single wake-up.
1533                _ = cmd_rx.recv_many(&mut spawn_batch, SPAWN_DRAIN_BATCH) => {
1534                    for spawn in spawn_batch.drain(..) {
1535                        self.spawn_dynamic_child(
1536                            spawn,
1537                            &mut worker_state,
1538                            &mut roster,
1539                            &mut significant_remaining,
1540                            auto_shutdown,
1541                        );
1542                    }
1543                }
1544            }
1545        };
1546
1547        // The run is ending -- either cleanly (shutdown was signalled) or with an error (a child failed to initialize
1548        // or restart, the restart limit was exceeded, or a significant child exited). On every path: stop accepting
1549        // spawns and discard anything still queued, rather than starting children only to tear them down immediately,
1550        // and then shut down all children.
1551        cmd_rx.close();
1552        let mut discarded = 0;
1553        while cmd_rx.try_recv().is_ok() {
1554            discarded += 1;
1555        }
1556        if discarded > 0 {
1557            debug!(
1558                supervisor_id = %self.supervisor_id,
1559                discarded,
1560                "Discarded queued dynamic children during shutdown."
1561            );
1562        }
1563        let aborted = worker_state.shutdown_workers(&mut halt).await;
1564
1565        // A requested shutdown that nonetheless had to forcefully abort one or more workers (here or anywhere in the
1566        // subtree below us) is surfaced as an unclean shutdown so it propagates up the tree rather than being reported
1567        // as success. An outcome that already carries an error (initialization, restart limit, significant child)
1568        // takes precedence -- that's the root cause -- and the forced aborts are left to the per-worker warnings.
1569        match outcome {
1570            Ok(()) if aborted > 0 => {
1571                warn!(supervisor_id = %self.supervisor_id, aborted, "Shutdown completed uncleanly; workers were forcefully aborted.");
1572                Err(SupervisorError::ShutdownTimedOut { aborted })
1573            }
1574            outcome => outcome,
1575        }
1576    }
1577
1578    fn as_nested_process(&self, process: Process, process_shutdown: ShutdownHandle, halt: HaltWatch) -> WorkerFuture {
1579        // Simple wrapper around `run_inner` to satisfy the return type signature needed when running the supervisor as
1580        // a nested child process in another supervisor.
1581        debug!(supervisor_id = %self.supervisor_id, "Nested supervisor starting.");
1582
1583        // Create a standalone clone of ourselves so we can fulfill the future signature.
1584        let sup = self.inner_clone();
1585
1586        Box::pin(async move {
1587            sup.run_inner(process, process_shutdown, halt)
1588                .await
1589                .map_err(WorkerError::from)
1590        })
1591    }
1592
1593    /// Runs the supervisor forever.
1594    ///
1595    /// # Errors
1596    ///
1597    /// If the supervisor exceeds its restart limits, or fails to initialize a child process, an error is returned.
1598    pub async fn run(&mut self) -> Result<(), SupervisorError> {
1599        // Create a no-op `ShutdownHandle` to satisfy the `run_inner` function. This is never used since we want to run
1600        // forever, but we need to satisfy the signature.
1601        let process_shutdown = ShutdownHandle::noop();
1602        let process = Process::supervisor(&self.supervisor_id, None).context(InvalidName {
1603            name: self.supervisor_id.to_string(),
1604        })?;
1605
1606        debug!(supervisor_id = %self.supervisor_id, "Supervisor starting.");
1607        self.run_inner(process.clone(), process_shutdown, HaltWatch::never())
1608            .into_process_future(process)
1609            .await
1610    }
1611
1612    /// Runs the supervisor until shutdown is triggered.
1613    ///
1614    /// When `shutdown` resolves, the supervisor will shutdown all child processes according to their shutdown strategy,
1615    /// and then return.
1616    ///
1617    /// # Errors
1618    ///
1619    /// If the supervisor exceeds its restart limits, or fails to initialize a child process, an error is returned.
1620    pub async fn run_with_shutdown<F: Future + Send + 'static>(&mut self, shutdown: F) -> Result<(), SupervisorError> {
1621        // Drive the caller-provided shutdown future into a trigger so the supervisor can begin shutting down its
1622        // children once `shutdown` resolves. The trigger fires at most once (guarded), and otherwise fires on drop if
1623        // the supervisor returns on its own first.
1624        let (shutdown_coordinator, shutdown_handle) = ShutdownHandle::paired();
1625        let run = self.run_with_shutdown_inner(shutdown_handle, HaltWatch::never(), None);
1626        pin!(run, shutdown);
1627
1628        let mut shutdown_coordinator = Some(shutdown_coordinator);
1629        loop {
1630            select! {
1631                result = &mut run => return result,
1632                _ = &mut shutdown, if shutdown_coordinator.is_some() => {
1633                    shutdown_coordinator.take().expect("coordinator present per select guard").shutdown();
1634                }
1635            }
1636        }
1637    }
1638
1639    /// Runs the supervisor until the given `ShutdownHandle` signal is received.
1640    ///
1641    /// This is an internal variant of `run_with_shutdown` that takes a `ShutdownHandle` directly, used when spawning
1642    /// supervisors in dedicated runtimes where the shutdown signal is already wrapped in a `ShutdownHandle`.
1643    ///
1644    /// `halt` lets the owner of a nested supervisor on a dedicated runtime halt it. Other callers pass
1645    /// `HaltWatch::never()`.
1646    ///
1647    /// If `dataspace` is provided, the supervisor will use it instead of creating a new one. This is used to propagate
1648    /// the parent's dataspace across OS thread boundaries for dedicated runtimes.
1649    ///
1650    /// # Errors
1651    ///
1652    /// If the supervisor exceeds its restart limits, or fails to initialize a child process, an error is returned.
1653    pub(crate) async fn run_with_shutdown_inner(
1654        &mut self, process_shutdown: ShutdownHandle, halt: HaltWatch, dataspace: Option<DataspaceRegistry>,
1655    ) -> Result<(), SupervisorError> {
1656        let process =
1657            Process::supervisor_with_dataspace(&self.supervisor_id, None, dataspace).context(InvalidName {
1658                name: self.supervisor_id.to_string(),
1659            })?;
1660
1661        debug!(supervisor_id = %self.supervisor_id, "Supervisor starting.");
1662        self.run_inner(process.clone(), process_shutdown, halt)
1663            .into_process_future(process)
1664            .await
1665    }
1666
1667    fn inner_clone(&self) -> Self {
1668        // This is no different than if we just implemented `Clone` directly, but it allows us to avoid exposing a
1669        // _public_ implementation of `Clone`, which we don't want normal users to be able to do. We only need this
1670        // internally to support nested supervisors.
1671        Self {
1672            supervisor_id: Arc::clone(&self.supervisor_id),
1673            child_specs: self.child_specs.clone(),
1674            runtime_mode: self.runtime_mode.clone(),
1675            current_tx: Arc::clone(&self.current_tx),
1676            id_counter: Arc::clone(&self.id_counter),
1677            active: Arc::clone(&self.active),
1678            node: Arc::clone(&self.node),
1679        }
1680    }
1681}
1682
1683#[cfg(test)]
1684// Significance and explicit restart types are deprecated, but supervisors still use them. These tests make sure that
1685// they continue to work correctly until they are removed.
1686#[allow(deprecated)]
1687mod tests {
1688    use std::{
1689        future::pending,
1690        sync::atomic::{AtomicBool, AtomicUsize, Ordering},
1691    };
1692
1693    use async_trait::async_trait;
1694    use saluki_common::sync::shutdown::ShutdownCoordinator;
1695    use saluki_metrics::test::TestRecorder;
1696    use tokio::{
1697        sync::oneshot,
1698        task::JoinHandle,
1699        time::{sleep, timeout},
1700    };
1701
1702    use super::*;
1703    use crate::runtime::{self, FnWorker, NodeKind, NodeSnapshot, NodeState, SupervisorFuture};
1704    use crate::test_support::wait_until;
1705
1706    /// Behavior for a mock worker during initialization.
1707    #[derive(Clone)]
1708    enum InitBehavior {
1709        /// Initialization succeeds immediately.
1710        Instant,
1711
1712        /// Initialization takes the given duration before succeeding.
1713        Slow(Duration),
1714
1715        /// Initialization fails with the given message.
1716        Fail(&'static str),
1717    }
1718
1719    /// Behavior for a mock worker during runtime (after initialization).
1720    #[derive(Clone)]
1721    enum RunBehavior {
1722        /// Runs until shutdown is received.
1723        UntilShutdown,
1724
1725        /// Fails with the given error message after the given delay.
1726        FailAfter(Duration, &'static str),
1727
1728        /// Completes successfully after the given delay.
1729        CompleteAfter(Duration),
1730
1731        /// On shutdown, sleeps for the given duration before exiting (to exercise concurrent draining).
1732        SlowShutdown(Duration),
1733
1734        /// Ignores shutdown entirely and runs forever (to exercise abort-at-deadline).
1735        IgnoreShutdown,
1736
1737        /// Panics after the given delay, unless shutdown arrives first.
1738        PanicAfter(Duration),
1739    }
1740
1741    /// A configurable mock worker for testing supervisor behavior.
1742    struct MockWorker {
1743        name: &'static str,
1744        init_behavior: InitBehavior,
1745        run_behavior: RunBehavior,
1746        start_count: Arc<AtomicUsize>,
1747        finish_count: Arc<AtomicUsize>,
1748        brutal_shutdown: bool,
1749        graceful_timeout: Duration,
1750        restartable: bool,
1751    }
1752
1753    impl MockWorker {
1754        /// Creates a worker that runs until shutdown.
1755        fn long_running(name: &'static str) -> Self {
1756            Self {
1757                name,
1758                init_behavior: InitBehavior::Instant,
1759                run_behavior: RunBehavior::UntilShutdown,
1760                start_count: Arc::new(AtomicUsize::new(0)),
1761                finish_count: Arc::new(AtomicUsize::new(0)),
1762                brutal_shutdown: false,
1763                graceful_timeout: Duration::from_millis(500),
1764                restartable: true,
1765            }
1766        }
1767
1768        /// Creates a worker that fails after the given delay.
1769        fn failing(name: &'static str, delay: Duration) -> Self {
1770            Self {
1771                name,
1772                init_behavior: InitBehavior::Instant,
1773                run_behavior: RunBehavior::FailAfter(delay, "worker failed"),
1774                start_count: Arc::new(AtomicUsize::new(0)),
1775                finish_count: Arc::new(AtomicUsize::new(0)),
1776                brutal_shutdown: false,
1777                graceful_timeout: Duration::from_millis(500),
1778                restartable: true,
1779            }
1780        }
1781
1782        /// Creates a worker that completes successfully after the given delay.
1783        fn completing(name: &'static str, delay: Duration) -> Self {
1784            Self {
1785                name,
1786                init_behavior: InitBehavior::Instant,
1787                run_behavior: RunBehavior::CompleteAfter(delay),
1788                start_count: Arc::new(AtomicUsize::new(0)),
1789                finish_count: Arc::new(AtomicUsize::new(0)),
1790                brutal_shutdown: false,
1791                graceful_timeout: Duration::from_millis(500),
1792                restartable: true,
1793            }
1794        }
1795
1796        /// Creates a worker that sleeps for `delay` after observing shutdown before exiting.
1797        fn slow_shutdown(name: &'static str, delay: Duration) -> Self {
1798            Self {
1799                name,
1800                init_behavior: InitBehavior::Instant,
1801                run_behavior: RunBehavior::SlowShutdown(delay),
1802                start_count: Arc::new(AtomicUsize::new(0)),
1803                finish_count: Arc::new(AtomicUsize::new(0)),
1804                brutal_shutdown: false,
1805                graceful_timeout: Duration::from_millis(500),
1806                restartable: true,
1807            }
1808        }
1809
1810        /// Creates a worker that never reacts to shutdown.
1811        fn ignore_shutdown(name: &'static str) -> Self {
1812            Self {
1813                name,
1814                init_behavior: InitBehavior::Instant,
1815                run_behavior: RunBehavior::IgnoreShutdown,
1816                start_count: Arc::new(AtomicUsize::new(0)),
1817                finish_count: Arc::new(AtomicUsize::new(0)),
1818                brutal_shutdown: false,
1819                graceful_timeout: Duration::from_millis(500),
1820                restartable: true,
1821            }
1822        }
1823
1824        /// Creates a worker that panics after the given delay.
1825        fn panicking(name: &'static str, delay: Duration) -> Self {
1826            Self {
1827                name,
1828                init_behavior: InitBehavior::Instant,
1829                run_behavior: RunBehavior::PanicAfter(delay),
1830                start_count: Arc::new(AtomicUsize::new(0)),
1831                finish_count: Arc::new(AtomicUsize::new(0)),
1832                brutal_shutdown: false,
1833                graceful_timeout: Duration::from_millis(500),
1834                restartable: true,
1835            }
1836        }
1837
1838        /// Creates a worker that fails during initialization.
1839        fn init_failure(name: &'static str) -> Self {
1840            Self {
1841                name,
1842                init_behavior: InitBehavior::Fail("init failed"),
1843                run_behavior: RunBehavior::UntilShutdown,
1844                start_count: Arc::new(AtomicUsize::new(0)),
1845                finish_count: Arc::new(AtomicUsize::new(0)),
1846                brutal_shutdown: false,
1847                graceful_timeout: Duration::from_millis(500),
1848                restartable: true,
1849            }
1850        }
1851
1852        /// Creates a worker with slow initialization.
1853        fn slow_init(name: &'static str, init_delay: Duration) -> Self {
1854            Self {
1855                name,
1856                init_behavior: InitBehavior::Slow(init_delay),
1857                run_behavior: RunBehavior::UntilShutdown,
1858                start_count: Arc::new(AtomicUsize::new(0)),
1859                finish_count: Arc::new(AtomicUsize::new(0)),
1860                brutal_shutdown: false,
1861                graceful_timeout: Duration::from_millis(500),
1862                restartable: true,
1863            }
1864        }
1865
1866        /// Returns a shared handle to the start count for this worker.
1867        ///
1868        /// The start count ticks up the instant the worker's run future begins executing, which is *before* any
1869        /// programmed delay elapses. It records that the worker started (or was restarted), not that it ran to any
1870        /// particular outcome.
1871        fn start_count(&self) -> Arc<AtomicUsize> {
1872            Arc::clone(&self.start_count)
1873        }
1874
1875        /// Returns a shared handle to the finish count for this worker.
1876        ///
1877        /// The finish count ticks up only when the worker runs to its *own* programmed terminal state -- a
1878        /// [`RunBehavior::FailAfter`] failure, a [`RunBehavior::CompleteAfter`] completion, or a
1879        /// [`RunBehavior::SlowShutdown`] drain that finished -- and not when it is cut short by an abort. Tests use it
1880        /// to wait for a worker to actually fail or complete (rather than merely start) before asserting on restart
1881        /// behavior, so the failure/completion path is genuinely exercised.
1882        fn finish_count(&self) -> Arc<AtomicUsize> {
1883            Arc::clone(&self.finish_count)
1884        }
1885
1886        /// Configures this worker to use a `Brutal` shutdown strategy (immediate abort, no graceful wait).
1887        fn with_brutal_shutdown(mut self) -> Self {
1888            self.brutal_shutdown = true;
1889            self
1890        }
1891
1892        /// Overrides the worker's graceful shutdown timeout (defaults to 500 milliseconds).
1893        fn with_graceful_timeout(mut self, timeout: Duration) -> Self {
1894            self.graceful_timeout = timeout;
1895            self
1896        }
1897
1898        /// Makes this worker report that it cannot be started again, as a worker that consumes its state does.
1899        fn one_shot(mut self) -> Self {
1900            self.restartable = false;
1901            self
1902        }
1903    }
1904
1905    #[async_trait]
1906    impl Supervisable for MockWorker {
1907        fn name(&self) -> &str {
1908            self.name
1909        }
1910
1911        fn shutdown_strategy(&self) -> ShutdownStrategy {
1912            if self.brutal_shutdown {
1913                ShutdownStrategy::Brutal
1914            } else {
1915                ShutdownStrategy::Graceful(self.graceful_timeout)
1916            }
1917        }
1918
1919        fn can_restart(&self) -> bool {
1920            self.restartable
1921        }
1922
1923        async fn initialize(&self, process_shutdown: ShutdownHandle) -> Result<SupervisorFuture, InitializationError> {
1924            match &self.init_behavior {
1925                InitBehavior::Instant => {}
1926                InitBehavior::Slow(delay) => {
1927                    sleep(*delay).await;
1928                }
1929                InitBehavior::Fail(msg) => {
1930                    return Err(InitializationError::Failed {
1931                        source: GenericError::msg(*msg),
1932                    });
1933                }
1934            }
1935
1936            let start_count = Arc::clone(&self.start_count);
1937            let finish_count = Arc::clone(&self.finish_count);
1938            let run_behavior = self.run_behavior.clone();
1939
1940            Ok(Box::pin(async move {
1941                start_count.fetch_add(1, Ordering::SeqCst);
1942
1943                match run_behavior {
1944                    RunBehavior::UntilShutdown => {
1945                        process_shutdown.await;
1946                        Ok(())
1947                    }
1948                    RunBehavior::FailAfter(delay, msg) => {
1949                        select! {
1950                            _ = sleep(delay) => {
1951                                // Ran to our own programmed failure rather than being cut short by shutdown; record
1952                                // it so tests can wait for the failure to actually happen before asserting.
1953                                finish_count.fetch_add(1, Ordering::SeqCst);
1954                                Err(GenericError::msg(msg))
1955                            }
1956                            _ = process_shutdown => {
1957                                Ok(())
1958                            }
1959                        }
1960                    }
1961                    RunBehavior::CompleteAfter(delay) => {
1962                        select! {
1963                            _ = sleep(delay) => {
1964                                // Ran to our own programmed completion rather than being cut short by shutdown.
1965                                finish_count.fetch_add(1, Ordering::SeqCst);
1966                                Ok(())
1967                            }
1968                            _ = process_shutdown => Ok(()),
1969                        }
1970                    }
1971                    RunBehavior::SlowShutdown(delay) => {
1972                        process_shutdown.await;
1973                        sleep(delay).await;
1974                        // Finished draining rather than being aborted partway through it.
1975                        finish_count.fetch_add(1, Ordering::SeqCst);
1976                        Ok(())
1977                    }
1978                    RunBehavior::IgnoreShutdown => {
1979                        // Hold the handle (so the supervisor counts us as outstanding) but never react to it.
1980                        let _hold = process_shutdown;
1981                        pending().await
1982                    }
1983                    RunBehavior::PanicAfter(delay) => {
1984                        select! {
1985                            _ = sleep(delay) => panic!("worker panicked"),
1986                            _ = process_shutdown => Ok(()),
1987                        }
1988                    }
1989                }
1990            }))
1991        }
1992    }
1993
1994    /// Helper: run a supervisor with a oneshot-based shutdown trigger.
1995    ///
1996    /// Returns the shutdown sender and a join handle for the run. The supervisor is polled to a running state (its
1997    /// static children spawned) via readiness polling rather than a blind startup sleep, so callers can rely on it
1998    /// being live on return.
1999    async fn run_supervisor_with_trigger(
2000        supervisor: Supervisor,
2001    ) -> (oneshot::Sender<()>, JoinHandle<Result<(), SupervisorError>>) {
2002        // Grab a handle before moving the supervisor into the run task so we can observe when it actually starts.
2003        let sup_handle = supervisor.handle();
2004        let mut supervisor = supervisor;
2005
2006        let (tx, rx) = oneshot::channel();
2007        let handle = tokio::spawn(async move { supervisor.run_with_shutdown(rx).await });
2008
2009        wait_until("supervisor is running", || sup_handle.is_running()).await;
2010        (tx, handle)
2011    }
2012
2013    /// Helper: awaits a spawned supervisor run to completion under a bounded timeout, unwrapping the join.
2014    ///
2015    /// Collapses the `timeout(..).await.unwrap().unwrap()` suffix repeated across the restart/shutdown tests into one
2016    /// call with useful panic messages.
2017    async fn join_supervisor(handle: JoinHandle<Result<(), SupervisorError>>) -> Result<(), SupervisorError> {
2018        timeout(Duration::from_secs(2), handle)
2019            .await
2020            .expect("supervisor should exit promptly")
2021            .expect("supervisor task should not panic")
2022    }
2023
2024    // -- Supervisor run mode tests ---------------------------------------------------------
2025
2026    #[tokio::test]
2027    async fn standalone_supervisor_shuts_down_cleanly() {
2028        let mut sup = Supervisor::new("test-sup").unwrap();
2029        sup.add_worker(MockWorker::long_running("worker1"));
2030        sup.add_worker(MockWorker::long_running("worker2"));
2031
2032        let (tx, handle) = run_supervisor_with_trigger(sup).await;
2033        tx.send(()).unwrap();
2034
2035        let result = join_supervisor(handle).await;
2036        assert!(result.is_ok());
2037    }
2038
2039    #[tokio::test]
2040    async fn nested_supervisor_shuts_down_cleanly() {
2041        let mut child_sup = Supervisor::new("child-sup").unwrap();
2042        child_sup.add_worker(MockWorker::long_running("inner-worker"));
2043
2044        let mut parent_sup = Supervisor::new("parent-sup").unwrap();
2045        parent_sup.add_worker(MockWorker::long_running("outer-worker"));
2046        parent_sup.add_worker(child_sup);
2047
2048        let (tx, handle) = run_supervisor_with_trigger(parent_sup).await;
2049        tx.send(()).unwrap();
2050
2051        let result = join_supervisor(handle).await;
2052        assert!(result.is_ok());
2053    }
2054
2055    #[tokio::test]
2056    async fn empty_supervisor_idles_until_shutdown() {
2057        // A supervisor with no static children is valid: it idles, waiting for dynamic children, and shuts down
2058        // cleanly when signalled. (Before dynamic children were folded in, this returned a `NoChildren` error.)
2059        let sup = Supervisor::new("empty-sup").unwrap();
2060
2061        let (tx, handle) = run_supervisor_with_trigger(sup).await;
2062        assert!(!handle.is_finished(), "an empty supervisor must idle rather than exit");
2063
2064        tx.send(()).unwrap();
2065        let result = join_supervisor(handle).await;
2066        assert!(result.is_ok());
2067    }
2068
2069    // -- Child restart behavior tests ------------------------------------------------------
2070
2071    #[tokio::test]
2072    async fn one_for_one_restarts_only_failed_child() {
2073        let failing = MockWorker::failing("failing-worker", Duration::from_millis(50));
2074        let failing_count = failing.start_count();
2075
2076        let stable = MockWorker::long_running("stable-worker");
2077        let stable_count = stable.start_count();
2078
2079        let mut sup = Supervisor::new("test-sup").unwrap().with_restart_strategy(
2080            RestartStrategy::one_to_one().with_intensity_and_period(20, Duration::from_secs(10)),
2081        );
2082        sup.add_worker(stable);
2083        sup.add_worker(failing);
2084
2085        let (tx, handle) = run_supervisor_with_trigger(sup).await;
2086
2087        // Wait until the failing worker has actually been restarted (its second start), then shut down.
2088        wait_until("the failing worker has been restarted", || {
2089            failing_count.load(Ordering::SeqCst) >= 2
2090        })
2091        .await;
2092        let _ = tx.send(());
2093
2094        let result = join_supervisor(handle).await;
2095        assert!(result.is_ok());
2096
2097        // The failing worker should have been started multiple times.
2098        assert!(
2099            failing_count.load(Ordering::SeqCst) >= 2,
2100            "failing worker should have been restarted"
2101        );
2102        // The stable worker should only have been started once (never restarted).
2103        assert_eq!(
2104            stable_count.load(Ordering::SeqCst),
2105            1,
2106            "stable worker should not have been restarted"
2107        );
2108    }
2109
2110    #[tokio::test]
2111    async fn one_for_all_restarts_all_children() {
2112        let failing = MockWorker::failing("failing-worker", Duration::from_millis(50));
2113        let failing_count = failing.start_count();
2114
2115        let stable = MockWorker::long_running("stable-worker");
2116        let stable_count = stable.start_count();
2117
2118        let mut sup = Supervisor::new("test-sup").unwrap().with_restart_strategy(
2119            RestartStrategy::one_for_all().with_intensity_and_period(20, Duration::from_secs(10)),
2120        );
2121        sup.add_worker(stable);
2122        sup.add_worker(failing);
2123
2124        let (tx, handle) = run_supervisor_with_trigger(sup).await;
2125
2126        // Wait until a one-for-all cycle has restarted both workers (each on its second start), then shut down.
2127        wait_until("both workers have been restarted", || {
2128            failing_count.load(Ordering::SeqCst) >= 2 && stable_count.load(Ordering::SeqCst) >= 2
2129        })
2130        .await;
2131        let _ = tx.send(());
2132
2133        let result = join_supervisor(handle).await;
2134        assert!(result.is_ok());
2135
2136        // Both workers should have been started multiple times.
2137        assert!(
2138            failing_count.load(Ordering::SeqCst) >= 2,
2139            "failing worker should have been restarted"
2140        );
2141        assert!(
2142            stable_count.load(Ordering::SeqCst) >= 2,
2143            "stable worker should also have been restarted"
2144        );
2145    }
2146
2147    #[tokio::test]
2148    async fn one_for_all_does_not_restart_temporary_children() {
2149        // A permanent worker that fails repeatedly drives one-for-all restarts; a temporary sibling is shut down with
2150        // the group on each cycle but, per OTP semantics, must never be brought back.
2151        let failing = MockWorker::failing("failing-worker", Duration::from_millis(50));
2152        let failing_count = failing.start_count();
2153
2154        let temp = MockWorker::long_running("temp-worker");
2155        let temp_count = temp.start_count();
2156
2157        let mut sup = Supervisor::new("test-sup").unwrap().with_restart_strategy(
2158            RestartStrategy::one_for_all().with_intensity_and_period(20, Duration::from_secs(10)),
2159        );
2160        sup.add_worker(ChildSpecification::worker(temp).with_restart_type(RestartType::Temporary));
2161        sup.add_worker(failing);
2162
2163        let (tx, handle) = run_supervisor_with_trigger(sup).await;
2164
2165        // Wait until the permanent worker has driven at least one one-for-all restart, then shut down.
2166        wait_until("the permanent worker has been restarted", || {
2167            failing_count.load(Ordering::SeqCst) >= 2
2168        })
2169        .await;
2170        let _ = tx.send(());
2171
2172        let result = join_supervisor(handle).await;
2173        assert!(result.is_ok());
2174        assert!(
2175            failing_count.load(Ordering::SeqCst) >= 2,
2176            "permanent worker should have been restarted by one-for-all"
2177        );
2178        assert_eq!(
2179            temp_count.load(Ordering::SeqCst),
2180            1,
2181            "temporary child must not be restarted by a one-for-all group restart"
2182        );
2183    }
2184
2185    #[tokio::test]
2186    async fn one_for_all_restarts_transient_children() {
2187        // A transient child that exits cleanly is not restarted on its own, but a one-for-all restart triggered by a
2188        // sibling restarts it anyway -- matching OTP, where only temporary children are exempt from group restarts.
2189        let transient = MockWorker::completing("transient-worker", Duration::from_millis(30));
2190        let transient_count = transient.start_count();
2191
2192        // Fails after the transient has already exited cleanly, so the group restart is what brings the transient back.
2193        let failing = MockWorker::failing("failing-worker", Duration::from_millis(80));
2194
2195        let mut sup = Supervisor::new("test-sup").unwrap().with_restart_strategy(
2196            RestartStrategy::one_for_all().with_intensity_and_period(20, Duration::from_secs(10)),
2197        );
2198        sup.add_worker(ChildSpecification::worker(transient).with_restart_type(RestartType::Transient));
2199        sup.add_worker(failing);
2200
2201        let (tx, handle) = run_supervisor_with_trigger(sup).await;
2202
2203        wait_until("the transient worker has been restarted by the group", || {
2204            transient_count.load(Ordering::SeqCst) >= 2
2205        })
2206        .await;
2207        let _ = tx.send(());
2208
2209        let result = join_supervisor(handle).await;
2210        assert!(result.is_ok());
2211        assert!(
2212            transient_count.load(Ordering::SeqCst) >= 2,
2213            "transient child must be restarted by a one-for-all group restart, even after a clean exit"
2214        );
2215    }
2216
2217    #[tokio::test]
2218    async fn transient_abnormal_exit_triggers_one_for_all() {
2219        // A transient child's *own* abnormal exit is restartable, so under one-for-all it triggers a whole-group
2220        // restart -- the sibling is restarted too, not just the transient.
2221        let transient = MockWorker::failing("transient-worker", Duration::from_millis(50));
2222        let transient_count = transient.start_count();
2223
2224        let stable = MockWorker::long_running("stable-worker");
2225        let stable_count = stable.start_count();
2226
2227        let mut sup = Supervisor::new("test-sup").unwrap().with_restart_strategy(
2228            RestartStrategy::one_for_all().with_intensity_and_period(20, Duration::from_secs(10)),
2229        );
2230        sup.add_worker(ChildSpecification::worker(transient).with_restart_type(RestartType::Transient));
2231        sup.add_worker(stable);
2232
2233        let (tx, handle) = run_supervisor_with_trigger(sup).await;
2234
2235        wait_until("the abnormal exit has restarted both workers", || {
2236            transient_count.load(Ordering::SeqCst) >= 2 && stable_count.load(Ordering::SeqCst) >= 2
2237        })
2238        .await;
2239        let _ = tx.send(());
2240
2241        let result = join_supervisor(handle).await;
2242        assert!(result.is_ok());
2243        assert!(
2244            transient_count.load(Ordering::SeqCst) >= 2,
2245            "transient worker must be restarted after its own abnormal exit"
2246        );
2247        assert!(
2248            stable_count.load(Ordering::SeqCst) >= 2,
2249            "the transient's abnormal exit must trigger a one-for-all that also restarts the sibling"
2250        );
2251    }
2252
2253    #[tokio::test]
2254    async fn restart_limit_exceeded_shuts_down_supervisor() {
2255        let mut sup = Supervisor::new("test-sup")
2256            .unwrap()
2257            .with_restart_strategy(RestartStrategy::one_to_one().with_intensity_and_period(1, Duration::from_secs(10)));
2258        // This worker fails immediately, which will exhaust the restart budget quickly.
2259        sup.add_worker(MockWorker::failing("fast-fail", Duration::ZERO));
2260
2261        let (tx, rx) = oneshot::channel::<()>();
2262        let handle = tokio::spawn(async move { sup.run_with_shutdown(rx).await });
2263
2264        let result = join_supervisor(handle).await;
2265        drop(tx);
2266
2267        assert!(matches!(result, Err(SupervisorError::Shutdown)));
2268    }
2269
2270    // -- Restart type tests ----------------------------------------------------------------
2271
2272    #[tokio::test]
2273    async fn temporary_child_is_not_restarted() {
2274        // A temporary worker that fails quickly, alongside a long-running worker that keeps the supervisor alive.
2275        let temp = MockWorker::failing("temp-worker", Duration::from_millis(50));
2276        let temp_started = temp.start_count();
2277        let temp_failed = temp.finish_count();
2278
2279        let stable = MockWorker::long_running("stable-worker");
2280
2281        let mut sup = Supervisor::new("test-sup").unwrap().with_restart_strategy(
2282            RestartStrategy::one_to_one().with_intensity_and_period(20, Duration::from_secs(10)),
2283        );
2284        sup.add_worker(stable);
2285        sup.add_worker(ChildSpecification::worker(temp).with_restart_type(RestartType::Temporary));
2286
2287        let (tx, handle) = run_supervisor_with_trigger(sup).await;
2288
2289        // Wait for the worker to *actually fail*, not merely start. `start_count` ticks up the instant the worker
2290        // begins running -- well before its 50ms failure -- so shutting down as soon as it reached 1 would tear the
2291        // supervisor down before the failure -> no-restart path ever ran, hiding a regression that restarted a
2292        // temporary child (or charged the failure against restart intensity). `finish_count` ticks only once the
2293        // worker runs to its own failure, so waiting on it genuinely exercises that path before we shut down.
2294        wait_until("the temporary worker has failed once", || {
2295            temp_failed.load(Ordering::SeqCst) == 1
2296        })
2297        .await;
2298        let _ = tx.send(());
2299
2300        let result = join_supervisor(handle).await;
2301        assert!(result.is_ok());
2302        assert_eq!(
2303            temp_started.load(Ordering::SeqCst),
2304            1,
2305            "temporary worker must not be restarted after it fails"
2306        );
2307    }
2308
2309    #[tokio::test]
2310    async fn transient_child_is_not_restarted_on_clean_exit() {
2311        let transient = MockWorker::completing("transient-worker", Duration::from_millis(50));
2312        let transient_started = transient.start_count();
2313        let transient_finished = transient.finish_count();
2314
2315        let stable = MockWorker::long_running("stable-worker");
2316
2317        let mut sup = Supervisor::new("test-sup").unwrap().with_restart_strategy(
2318            RestartStrategy::one_to_one().with_intensity_and_period(20, Duration::from_secs(10)),
2319        );
2320        sup.add_worker(stable);
2321        sup.add_worker(ChildSpecification::worker(transient).with_restart_type(RestartType::Transient));
2322
2323        let (tx, handle) = run_supervisor_with_trigger(sup).await;
2324
2325        // Wait for the worker to *actually complete*, not merely start: `start_count` ticks the instant it begins
2326        // running, so shutting down as soon as it reached 1 would drive the supervisor's teardown before the clean
2327        // exit -> no-restart path ran, hiding a regression that restarted a transient child after a clean exit.
2328        // `finish_count` ticks only once the worker runs to its own completion.
2329        wait_until("the transient worker has completed once", || {
2330            transient_finished.load(Ordering::SeqCst) == 1
2331        })
2332        .await;
2333        let _ = tx.send(());
2334
2335        let result = join_supervisor(handle).await;
2336        assert!(result.is_ok());
2337        assert_eq!(
2338            transient_started.load(Ordering::SeqCst),
2339            1,
2340            "transient worker must not be restarted after a clean exit"
2341        );
2342    }
2343
2344    #[tokio::test]
2345    async fn transient_child_is_restarted_on_failure() {
2346        let transient = MockWorker::failing("transient-worker", Duration::from_millis(50));
2347        let transient_count = transient.start_count();
2348
2349        let mut sup = Supervisor::new("test-sup").unwrap().with_restart_strategy(
2350            RestartStrategy::one_to_one().with_intensity_and_period(20, Duration::from_secs(10)),
2351        );
2352        sup.add_worker(ChildSpecification::worker(transient).with_restart_type(RestartType::Transient));
2353
2354        let (tx, handle) = run_supervisor_with_trigger(sup).await;
2355
2356        wait_until("the transient worker has been restarted", || {
2357            transient_count.load(Ordering::SeqCst) >= 2
2358        })
2359        .await;
2360        let _ = tx.send(());
2361
2362        let result = join_supervisor(handle).await;
2363        assert!(result.is_ok());
2364        assert!(
2365            transient_count.load(Ordering::SeqCst) >= 2,
2366            "transient worker must be restarted after an abnormal exit"
2367        );
2368    }
2369
2370    #[tokio::test]
2371    async fn transient_child_is_restarted_on_panic() {
2372        // The other half of what makes `transient` the right policy for a worker serving an object with a lifetime of
2373        // its own -- a cache's expiration loop, say. A panic is a bug worth recovering from, so the worker comes back;
2374        // the clean exit that follows the served object being dropped is taken at face value (see
2375        // `transient_child_is_not_restarted_on_clean_exit`), so it stays stopped. Under `Permanent` that clean exit
2376        // would be restarted straight into another clean exit, over and over, until the restart intensity failed the
2377        // whole supervisor.
2378        let transient = MockWorker::panicking("transient-worker", Duration::from_millis(50));
2379        let transient_count = transient.start_count();
2380
2381        let mut sup = Supervisor::new("test-sup").unwrap().with_restart_strategy(
2382            RestartStrategy::one_to_one().with_intensity_and_period(20, Duration::from_secs(10)),
2383        );
2384        sup.add_worker(ChildSpecification::worker(transient).with_restart_type(RestartType::Transient));
2385
2386        let (tx, handle) = run_supervisor_with_trigger(sup).await;
2387
2388        wait_until("the panicking transient worker has been restarted", || {
2389            transient_count.load(Ordering::SeqCst) >= 2
2390        })
2391        .await;
2392        let _ = tx.send(());
2393
2394        let result = join_supervisor(handle).await;
2395        assert!(result.is_ok());
2396        assert!(
2397            transient_count.load(Ordering::SeqCst) >= 2,
2398            "transient worker must be restarted after a panic"
2399        );
2400    }
2401
2402    #[tokio::test]
2403    async fn permanent_child_is_restarted_on_clean_exit() {
2404        // A permanent worker that completes cleanly must still be restarted -- this is what distinguishes
2405        // `Permanent` from `Transient`, which is left stopped after a clean exit.
2406        let permanent = MockWorker::completing("permanent-worker", Duration::from_millis(50));
2407        let permanent_count = permanent.start_count();
2408
2409        let mut sup = Supervisor::new("test-sup").unwrap().with_restart_strategy(
2410            RestartStrategy::one_to_one().with_intensity_and_period(20, Duration::from_secs(10)),
2411        );
2412        // Added with the default restart policy, which is `Permanent`.
2413        sup.add_worker(permanent);
2414
2415        let (tx, handle) = run_supervisor_with_trigger(sup).await;
2416
2417        wait_until("the permanent worker has been restarted", || {
2418            permanent_count.load(Ordering::SeqCst) >= 2
2419        })
2420        .await;
2421        let _ = tx.send(());
2422
2423        let result = join_supervisor(handle).await;
2424        assert!(result.is_ok());
2425        assert!(
2426            permanent_count.load(Ordering::SeqCst) >= 2,
2427            "permanent worker must be restarted even after a clean exit"
2428        );
2429    }
2430
2431    #[tokio::test]
2432    async fn temporary_failures_do_not_consume_restart_intensity() {
2433        // With intensity=1, two *restartable* failures within the period would shut the supervisor down. Here several
2434        // temporary workers all fail quickly. Because temporary exits aren't eligible for restart, they must not consume
2435        // the restart-intensity budget, and the supervisor must stay up.
2436        let mut sup = Supervisor::new("test-sup")
2437            .unwrap()
2438            .with_restart_strategy(RestartStrategy::one_to_one().with_intensity_and_period(1, Duration::from_secs(10)));
2439
2440        let workers = [
2441            MockWorker::failing("temp-0", Duration::from_millis(20)),
2442            MockWorker::failing("temp-1", Duration::from_millis(20)),
2443            MockWorker::failing("temp-2", Duration::from_millis(20)),
2444            MockWorker::failing("temp-3", Duration::from_millis(20)),
2445            MockWorker::failing("temp-4", Duration::from_millis(20)),
2446        ];
2447        let started: Vec<_> = workers.iter().map(|w| w.start_count()).collect();
2448        let failed: Vec<_> = workers.iter().map(|w| w.finish_count()).collect();
2449        for worker in workers {
2450            sup.add_worker(ChildSpecification::worker(worker).with_restart_type(RestartType::Temporary));
2451        }
2452        // A long-running worker so the supervisor doesn't simply idle once the temporaries are gone.
2453        sup.add_worker(MockWorker::long_running("stable-worker"));
2454
2455        let (tx, handle) = run_supervisor_with_trigger(sup).await;
2456        // Wait for every temporary worker to *actually fail* on its own. Keying off `start_count` would let shutdown
2457        // cut them short before their failures ran, so the supervisor would never get the chance to (mis)charge those
2458        // failures against its intensity=1 budget -- hiding the very regression this guards against.
2459        wait_until("every temporary worker has failed once", || {
2460            failed.iter().all(|c| c.load(Ordering::SeqCst) == 1)
2461        })
2462        .await;
2463        let _ = tx.send(());
2464
2465        let result = join_supervisor(handle).await;
2466        assert!(
2467            result.is_ok(),
2468            "supervisor must not trip its restart limit on temporary exits"
2469        );
2470        for count in started {
2471            assert_eq!(
2472                count.load(Ordering::SeqCst),
2473                1,
2474                "each temporary worker runs exactly once"
2475            );
2476        }
2477    }
2478
2479    #[tokio::test]
2480    async fn transient_clean_exits_do_not_consume_restart_intensity() {
2481        // With intensity=1, two *restartable* exits within the period would shut the supervisor down. Here several
2482        // transient workers all complete cleanly. A transient child's clean exit isn't eligible for restart, so it
2483        // must not consume the restart-intensity budget, and the supervisor must stay up.
2484        let mut sup = Supervisor::new("test-sup")
2485            .unwrap()
2486            .with_restart_strategy(RestartStrategy::one_to_one().with_intensity_and_period(1, Duration::from_secs(10)));
2487
2488        let workers = [
2489            MockWorker::completing("transient-0", Duration::from_millis(20)),
2490            MockWorker::completing("transient-1", Duration::from_millis(20)),
2491            MockWorker::completing("transient-2", Duration::from_millis(20)),
2492            MockWorker::completing("transient-3", Duration::from_millis(20)),
2493            MockWorker::completing("transient-4", Duration::from_millis(20)),
2494        ];
2495        let started: Vec<_> = workers.iter().map(|w| w.start_count()).collect();
2496        let finished: Vec<_> = workers.iter().map(|w| w.finish_count()).collect();
2497        for worker in workers {
2498            sup.add_worker(ChildSpecification::worker(worker).with_restart_type(RestartType::Transient));
2499        }
2500        // A long-running worker so the supervisor doesn't simply idle once the transients have completed.
2501        sup.add_worker(MockWorker::long_running("stable-worker"));
2502
2503        let (tx, handle) = run_supervisor_with_trigger(sup).await;
2504        // Wait for every transient to *actually complete* on its own. Keying off `start_count` would let shutdown cut
2505        // the workers short before their clean exits ran, so the supervisor would never get the chance to (mis)charge
2506        // those exits against its intensity=1 budget -- hiding the very regression this guards against.
2507        wait_until("every transient worker has completed once", || {
2508            finished.iter().all(|c| c.load(Ordering::SeqCst) == 1)
2509        })
2510        .await;
2511        let _ = tx.send(());
2512
2513        let result = join_supervisor(handle).await;
2514        assert!(
2515            result.is_ok(),
2516            "supervisor must not trip its restart limit on clean transient exits"
2517        );
2518        for count in started {
2519            assert_eq!(
2520                count.load(Ordering::SeqCst),
2521                1,
2522                "each transient worker runs exactly once"
2523            );
2524        }
2525    }
2526
2527    #[tokio::test]
2528    async fn supervisor_idles_when_all_temporary_children_exit() {
2529        // When every static child is temporary and they all exit, the worker set drains. The supervisor must not panic
2530        // or exit on its own; it must keep running and remain able to accept new (dynamic) work until shutdown is
2531        // triggered.
2532        let temp_a = MockWorker::completing("temp-a", Duration::from_millis(10));
2533        let a_finished = temp_a.finish_count();
2534        let temp_b = MockWorker::completing("temp-b", Duration::from_millis(10));
2535        let b_finished = temp_b.finish_count();
2536
2537        let mut sup = Supervisor::new("test-sup").unwrap();
2538        let handle = sup.handle();
2539        sup.add_worker(ChildSpecification::worker(temp_a).with_restart_type(RestartType::Temporary));
2540        sup.add_worker(ChildSpecification::worker(temp_b).with_restart_type(RestartType::Temporary));
2541
2542        let (tx, run) = run_supervisor_with_trigger(sup).await;
2543
2544        // Wait for both temporary children to actually complete -- draining the worker set to empty -- before probing.
2545        // Keying off `start_count` could spawn the probe child before the set ever emptied, letting a supervisor that
2546        // (wrongly) exited once its last child left slip through.
2547        wait_until("both temporary children have completed", || {
2548            a_finished.load(Ordering::SeqCst) == 1 && b_finished.load(Ordering::SeqCst) == 1
2549        })
2550        .await;
2551
2552        // The supervisor must still be alive after its worker set empties: spawning a new dynamic child succeeds and
2553        // runs, which is only possible if the supervise loop kept running rather than exiting when the last child left.
2554        let dynamic = MockWorker::long_running("late-comer");
2555        let dynamic_count = dynamic.start_count();
2556        handle.spawn(dynamic);
2557        wait_until("the late dynamic child has started", || {
2558            dynamic_count.load(Ordering::SeqCst) == 1
2559        })
2560        .await;
2561        assert!(
2562            handle.is_running(),
2563            "supervisor must keep running after all temporary children exit"
2564        );
2565
2566        tx.send(()).unwrap();
2567        let result = join_supervisor(run).await;
2568        assert!(result.is_ok());
2569    }
2570
2571    // -- Significant child / auto-shutdown tests -------------------------------------------
2572
2573    #[tokio::test]
2574    async fn significant_child_drives_auto_shutdown() {
2575        // With `AnySignificant`, a significant child terminating (even cleanly, and without being restarted) must
2576        // shut the supervisor down, surfacing the significant-exit error.
2577        let mut sup = Supervisor::new("test-sup")
2578            .unwrap()
2579            .with_auto_shutdown(AutoShutdown::AnySignificant);
2580        sup.add_worker(MockWorker::long_running("stable"));
2581        sup.add_worker(
2582            ChildSpecification::worker(MockWorker::completing("significant", Duration::from_millis(50)))
2583                .with_restart_type(RestartType::Temporary)
2584                .with_significant(true),
2585        );
2586
2587        // Hold the shutdown sender so the only thing that can stop the supervisor is the significant child.
2588        let (_tx, rx) = oneshot::channel::<()>();
2589        let result = timeout(Duration::from_secs(2), sup.run_with_shutdown(rx))
2590            .await
2591            .unwrap();
2592        assert!(matches!(result, Err(SupervisorError::SignificantChildExited)));
2593    }
2594
2595    #[tokio::test]
2596    async fn significant_child_added_before_the_auto_shutdown_policy_still_drives_it() {
2597        // `with_auto_shutdown` consumes the supervisor while `add_worker` borrows it, so adding children first and
2598        // setting the policy afterwards is a perfectly good way to build one up. Nothing about registration may
2599        // assume the policy is already final -- an earlier version of the inert-significance check read it at
2600        // registration time and flagged this ordering as a mistake.
2601        let mut sup = Supervisor::new("test-sup").unwrap();
2602        sup.add_worker(MockWorker::long_running("stable"));
2603        sup.add_worker(
2604            runtime::supervisable(MockWorker::completing("significant", Duration::from_millis(50)))
2605                .temporary()
2606                .with_significant(true)
2607                .build(),
2608        );
2609        let mut sup = sup.with_auto_shutdown(AutoShutdown::AnySignificant);
2610
2611        let (_tx, rx) = oneshot::channel::<()>();
2612        let result = timeout(Duration::from_secs(2), sup.run_with_shutdown(rx))
2613            .await
2614            .unwrap();
2615        assert!(
2616            matches!(result, Err(SupervisorError::SignificantChildExited)),
2617            "the policy set after registration should still have applied, got {result:?}"
2618        );
2619    }
2620
2621    #[tokio::test]
2622    async fn non_significant_exit_does_not_auto_shutdown() {
2623        // Even with `AnySignificant` set, a non-significant child exiting must not shut the supervisor down.
2624        let plain = MockWorker::completing("plain", Duration::from_millis(10));
2625        let plain_finished = plain.finish_count();
2626
2627        let mut sup = Supervisor::new("test-sup")
2628            .unwrap()
2629            .with_auto_shutdown(AutoShutdown::AnySignificant);
2630        let handle = sup.handle();
2631        sup.add_worker(MockWorker::long_running("stable"));
2632        sup.add_worker(ChildSpecification::worker(plain).with_restart_type(RestartType::Temporary));
2633
2634        let (tx, run) = run_supervisor_with_trigger(sup).await;
2635
2636        // Let the non-significant child actually run to completion -- not merely start. Its completion is what could
2637        // (wrongly) trip `AnySignificant`, so we must observe the real exit before probing liveness; keying off
2638        // `start_count` could assert before the completion was ever processed.
2639        wait_until("the non-significant child has completed", || {
2640            plain_finished.load(Ordering::SeqCst) == 1
2641        })
2642        .await;
2643
2644        // The supervisor must still be alive after the non-significant child exits (had it been treated as
2645        // significant, `AnySignificant` would have torn the supervisor down). Spawning a dynamic child and observing
2646        // it start proves the supervise loop is still running.
2647        let dynamic = MockWorker::long_running("late-comer");
2648        let dynamic_count = dynamic.start_count();
2649        handle.spawn(dynamic);
2650        wait_until("the late dynamic child has started", || {
2651            dynamic_count.load(Ordering::SeqCst) == 1
2652        })
2653        .await;
2654        assert!(
2655            handle.is_running(),
2656            "a non-significant child exiting must not trigger auto-shutdown"
2657        );
2658
2659        tx.send(()).unwrap();
2660        let result = join_supervisor(run).await;
2661        assert!(result.is_ok());
2662    }
2663
2664    #[tokio::test]
2665    async fn all_significant_waits_for_last() {
2666        // With `AllSignificant`, the supervisor shuts down only once *all* significant children have terminated.
2667        let mut sup = Supervisor::new("test-sup")
2668            .unwrap()
2669            .with_auto_shutdown(AutoShutdown::AllSignificant);
2670        sup.add_worker(
2671            ChildSpecification::worker(MockWorker::completing("sig-a", Duration::from_millis(50)))
2672                .with_restart_type(RestartType::Temporary)
2673                .with_significant(true),
2674        );
2675        sup.add_worker(
2676            ChildSpecification::worker(MockWorker::completing("sig-b", Duration::from_millis(250)))
2677                .with_restart_type(RestartType::Temporary)
2678                .with_significant(true),
2679        );
2680
2681        let (_tx, rx) = oneshot::channel::<()>();
2682        let start = std::time::Instant::now();
2683        let result = timeout(Duration::from_secs(2), sup.run_with_shutdown(rx))
2684            .await
2685            .unwrap();
2686        let elapsed = start.elapsed();
2687
2688        assert!(matches!(result, Err(SupervisorError::SignificantChildExited)));
2689        // The first significant child exits at ~50ms but must NOT trigger shutdown; only the second (~250ms) does.
2690        assert!(
2691            elapsed >= Duration::from_millis(200),
2692            "auto-shutdown must wait for all significant children (took {elapsed:?})"
2693        );
2694    }
2695
2696    // -- Initialization failure tests ------------------------------------------------------
2697
2698    #[tokio::test]
2699    async fn init_failure_propagates_with_child_name() {
2700        let mut sup = Supervisor::new("test-sup").unwrap();
2701        sup.add_worker(MockWorker::long_running("good-worker"));
2702        sup.add_worker(MockWorker::init_failure("bad-worker"));
2703
2704        let (_tx, rx) = oneshot::channel::<()>();
2705        let result = timeout(Duration::from_secs(2), sup.run_with_shutdown(rx))
2706            .await
2707            .unwrap();
2708
2709        match result {
2710            Err(SupervisorError::FailedToInitialize { child_name, .. }) => {
2711                assert_eq!(child_name, "bad-worker");
2712            }
2713            other => panic!("expected FailedToInitialize, got: {:?}", other),
2714        }
2715    }
2716
2717    #[tokio::test]
2718    async fn init_failure_does_not_trigger_restart() {
2719        let init_fail = MockWorker::init_failure("bad-worker");
2720        let start_count = init_fail.start_count();
2721
2722        let mut sup = Supervisor::new("test-sup").unwrap().with_restart_strategy(
2723            RestartStrategy::one_to_one().with_intensity_and_period(10, Duration::from_secs(10)),
2724        );
2725        sup.add_worker(init_fail);
2726
2727        let (_tx, rx) = oneshot::channel::<()>();
2728        let result = timeout(Duration::from_secs(2), sup.run_with_shutdown(rx))
2729            .await
2730            .unwrap();
2731
2732        assert!(matches!(result, Err(SupervisorError::FailedToInitialize { .. })));
2733        // The worker never got past init, so start_count should be 0.
2734        assert_eq!(start_count.load(Ordering::SeqCst), 0);
2735    }
2736
2737    // -- Shutdown responsiveness tests -----------------------------------------------------
2738
2739    #[tokio::test]
2740    async fn shutdown_completes_promptly_in_steady_state() {
2741        let mut sup = Supervisor::new("test-sup").unwrap();
2742        sup.add_worker(MockWorker::long_running("worker1"));
2743        sup.add_worker(MockWorker::long_running("worker2"));
2744
2745        let (tx, handle) = run_supervisor_with_trigger(sup).await;
2746        tx.send(()).unwrap();
2747
2748        // Shutdown should complete well within 1 second (workers respond to shutdown signal immediately).
2749        let result = timeout(Duration::from_secs(1), handle).await;
2750        assert!(result.is_ok(), "shutdown should complete promptly");
2751    }
2752
2753    #[tokio::test]
2754    async fn shutdown_during_slow_init_completes_promptly() {
2755        let mut sup = Supervisor::new("test-sup").unwrap();
2756        // This worker takes 30 seconds to initialize — but we'll trigger shutdown immediately.
2757        sup.add_worker(MockWorker::slow_init("slow-worker", Duration::from_secs(30)));
2758
2759        let (tx, rx) = oneshot::channel();
2760        let handle = tokio::spawn(async move { sup.run_with_shutdown(rx).await });
2761
2762        // Give the supervisor just enough time to spawn the task, then trigger shutdown.
2763        sleep(Duration::from_millis(20)).await;
2764        tx.send(()).unwrap();
2765
2766        // Shutdown should complete quickly even though the worker hasn't finished initializing.
2767        // The supervisor loop sees the shutdown signal and aborts the still-initializing task.
2768        let result = timeout(Duration::from_secs(2), handle).await;
2769        assert!(result.is_ok(), "shutdown during slow init should complete promptly");
2770    }
2771
2772    // -- Dynamic children tests ------------------------------------------------------------
2773
2774    #[tokio::test]
2775    async fn dynamic_children_spawn_after_start() {
2776        let sup = Supervisor::new("dyn-sup").unwrap();
2777        let handle = sup.handle();
2778        let (tx, run) = run_supervisor_with_trigger(sup).await;
2779        wait_until("supervisor is running", || handle.is_running()).await;
2780
2781        let c1 = MockWorker::long_running("c1");
2782        let c2 = MockWorker::long_running("c2");
2783        let c1_count = c1.start_count();
2784        let c2_count = c2.start_count();
2785        handle.spawn(c1);
2786        handle.spawn(c2);
2787
2788        wait_until("both dynamic children have started", || {
2789            c1_count.load(Ordering::SeqCst) == 1 && c2_count.load(Ordering::SeqCst) == 1
2790        })
2791        .await;
2792        assert_eq!(handle.active_children(), 2);
2793
2794        tx.send(()).unwrap();
2795        let result = join_supervisor(run).await;
2796        assert!(result.is_ok());
2797        assert_eq!(
2798            handle.active_children(),
2799            0,
2800            "all dynamic children must be drained on shutdown"
2801        );
2802    }
2803
2804    #[tokio::test]
2805    async fn failing_bare_fn_worker_is_not_restarted() {
2806        // A bare `FnWorker` reaches the supervisor without a builder, through the conversion that every `Supervisable`
2807        // uses. It still reports that it cannot start again, so the supervisor reaps it when it fails. A restart would
2808        // fail to initialize the consumed body, and that would stop the supervisor and the siblings of the worker.
2809        let sup = Supervisor::new("dyn-sup").unwrap();
2810        let handle = sup.handle();
2811        let (tx, run) = run_supervisor_with_trigger(sup).await;
2812        wait_until("supervisor is running", || handle.is_running()).await;
2813
2814        handle.spawn(MockWorker::long_running("sibling"));
2815        let (ran_tx, ran_rx) = oneshot::channel();
2816        handle.spawn(FnWorker::new("bare", async move {
2817            let _ = ran_tx.send(());
2818            Err::<(), _>(GenericError::msg("boom"))
2819        }));
2820        ran_rx.await.expect("the bare worker should run");
2821        wait_until("only the sibling is still running", || handle.active_children() == 1).await;
2822
2823        sleep(Duration::from_millis(50)).await;
2824        assert!(handle.is_running(), "supervisor stays up after the bare worker fails");
2825
2826        tx.send(()).unwrap();
2827        let result = join_supervisor(run).await;
2828        assert!(result.is_ok(), "supervisor should stop cleanly: {result:?}");
2829    }
2830
2831    #[tokio::test]
2832    async fn temporary_dynamic_child_failure_is_isolated() {
2833        // A dynamic child that can only start once (temporary, hosted) is fault-isolated. When it fails, the supervisor
2834        // reaps and removes it, and does not restart it. The failure has no effect on the supervisor or on the siblings
2835        // of the child.
2836        let sup = Supervisor::new("dyn-sup").unwrap();
2837        let handle = sup.handle();
2838        let (tx, run) = run_supervisor_with_trigger(sup).await;
2839        wait_until("supervisor is running", || handle.is_running()).await;
2840
2841        let failing = MockWorker::failing("boom", Duration::from_millis(20));
2842        let failing_count = failing.start_count();
2843        handle.spawn(ChildSpecification::worker(failing.one_shot()));
2844        wait_until("the failing dynamic child has run once", || {
2845            failing_count.load(Ordering::SeqCst) == 1
2846        })
2847        .await;
2848        wait_until("all dynamic children have drained", || handle.active_children() == 0).await;
2849
2850        sleep(Duration::from_millis(50)).await;
2851        assert!(
2852            handle.is_running(),
2853            "supervisor stays up after an isolated child failure"
2854        );
2855        assert_eq!(
2856            failing_count.load(Ordering::SeqCst),
2857            1,
2858            "a temporary child is never restarted"
2859        );
2860
2861        // It still accepts new children.
2862        handle.spawn(MockWorker::long_running("c2"));
2863        wait_until("one dynamic child is running", || handle.active_children() == 1).await;
2864
2865        tx.send(()).unwrap();
2866        let result = join_supervisor(run).await;
2867        assert!(result.is_ok());
2868    }
2869
2870    #[tokio::test]
2871    async fn temporary_dynamic_child_panic_is_isolated() {
2872        // A panicking temporary, non-significant child is isolated exactly like an error exit.
2873        let sup = Supervisor::new("dyn-sup").unwrap();
2874        let handle = sup.handle();
2875        let (tx, run) = run_supervisor_with_trigger(sup).await;
2876        wait_until("supervisor is running", || handle.is_running()).await;
2877
2878        handle.spawn(ChildSpecification::worker(
2879            MockWorker::panicking("boom", Duration::from_millis(20)).one_shot(),
2880        ));
2881        wait_until("all dynamic children have drained", || handle.active_children() == 0).await;
2882
2883        sleep(Duration::from_millis(50)).await;
2884        assert!(handle.is_running(), "supervisor stays up after an isolated child panic");
2885
2886        tx.send(()).unwrap();
2887        let result = join_supervisor(run).await;
2888        assert!(result.is_ok());
2889    }
2890
2891    #[tokio::test]
2892    async fn significant_dynamic_child_failure_shuts_down_supervisor() {
2893        // A dynamic child added as significant, under `AutoShutdown::AnySignificant`, drives the supervisor to shut
2894        // down when it terminates -- the opt-in mechanism that replaces the old escalate-on-error behavior.
2895        let sup = Supervisor::new("dyn-sup")
2896            .unwrap()
2897            .with_auto_shutdown(AutoShutdown::AnySignificant);
2898        let handle = sup.handle();
2899        let (_tx, run) = run_supervisor_with_trigger(sup).await;
2900        wait_until("supervisor is running", || handle.is_running()).await;
2901
2902        handle.spawn(
2903            ChildSpecification::worker(MockWorker::failing("boom", Duration::from_millis(20)))
2904                .with_restart_type(RestartType::Temporary)
2905                .with_significant(true),
2906        );
2907
2908        let result = join_supervisor(run).await;
2909        assert!(matches!(result, Err(SupervisorError::SignificantChildExited)));
2910    }
2911
2912    #[tokio::test]
2913    async fn hosted_dynamic_child_exceeding_the_limit_shuts_down_supervisor() {
2914        // A dynamic child that can be started again is hosted by default, and the supervisor restarts it when it fails.
2915        // But hosted does not mean that its failures are ignored. If the child goes above the restart limit, the
2916        // supervisor stops, so that its own supervisor can try again.
2917        let sup = Supervisor::new("dyn-sup").unwrap();
2918        let handle = sup.handle();
2919        let (_tx, run) = run_supervisor_with_trigger(sup).await;
2920
2921        let failing = MockWorker::failing("boom", Duration::from_millis(20));
2922        let failing_count = failing.start_count();
2923        handle.spawn(failing);
2924
2925        // The default intensity permits one restart in the period. The second failure goes above that limit.
2926        let result = join_supervisor(run).await;
2927        assert!(
2928            matches!(result, Err(SupervisorError::Shutdown)),
2929            "a hosted child exceeding the limit should take the supervisor down, got {result:?}"
2930        );
2931        assert_eq!(failing_count.load(Ordering::SeqCst), 2);
2932    }
2933
2934    #[tokio::test]
2935    async fn needed_dynamic_child_terminating_shuts_down_supervisor() {
2936        // When a child marked needed terminates without a restart, its supervisor also stops. Here, the child can only
2937        // start once, and it exits cleanly.
2938        let sup = Supervisor::new("dyn-sup").unwrap();
2939        let handle = sup.handle();
2940        let (_tx, run) = run_supervisor_with_trigger(sup).await;
2941
2942        handle.spawn(
2943            ChildSpecification::worker(MockWorker::completing("brief", Duration::from_millis(20)).one_shot())
2944                .with_edge(Edge::Needed),
2945        );
2946
2947        let result = join_supervisor(run).await;
2948        assert!(matches!(result, Err(SupervisorError::SignificantChildExited)));
2949    }
2950
2951    #[tokio::test]
2952    async fn dynamic_spawn_outside_a_run_is_accepted_and_dropped() {
2953        // Spawning is infallible in the same sense `tokio::spawn` is: the child is always accepted, but a child handed
2954        // to a supervisor that isn't running is never started. That holds both before a run and after one, and a child
2955        // spawned before the run must not be held over and started by it -- children belong to a run, not to the
2956        // supervisor across runs.
2957        let sup = Supervisor::new("dyn-sup").unwrap();
2958        let handle = sup.handle();
2959
2960        assert!(!handle.is_running());
2961        let before = MockWorker::long_running("before-start");
2962        let before_count = before.start_count();
2963        handle.spawn(before);
2964
2965        // Once it's running, spawns do start children.
2966        let (tx, run) = run_supervisor_with_trigger(sup).await;
2967        wait_until("supervisor is running", || handle.is_running()).await;
2968        let worker = MockWorker::long_running("after-start");
2969        let started = worker.start_count();
2970        handle.spawn(worker);
2971        wait_until("the dynamic child has started", || started.load(Ordering::SeqCst) == 1).await;
2972        assert_eq!(
2973            before_count.load(Ordering::SeqCst),
2974            0,
2975            "a child spawned before the run must not be started by it"
2976        );
2977
2978        tx.send(()).unwrap();
2979        let result = join_supervisor(run).await;
2980        assert!(result.is_ok());
2981
2982        // And once it has shut down there is nothing left to start children either.
2983        wait_until("the supervisor has stopped", || !handle.is_running()).await;
2984        let after = MockWorker::long_running("after-shutdown");
2985        let after_count = after.start_count();
2986        handle.spawn(after);
2987
2988        sleep(Duration::from_millis(50)).await;
2989        assert_eq!(
2990            after_count.load(Ordering::SeqCst),
2991            0,
2992            "a child spawned after shutdown must never start"
2993        );
2994    }
2995
2996    #[tokio::test]
2997    async fn dynamic_spawn_allocates_an_id_eagerly() {
2998        // The id comes back synchronously, before the supervisor has picked the child up, so it can't depend on
2999        // registration having happened. Ids come from the supervisor's shared counter, so with no static children the
3000        // first dynamic child takes id 0.
3001        let sup = Supervisor::new("dyn-sup").unwrap();
3002        let handle = sup.handle();
3003        let (tx, run) = run_supervisor_with_trigger(sup).await;
3004        wait_until("supervisor is running", || handle.is_running()).await;
3005
3006        let worker = MockWorker::long_running("c");
3007        let started = worker.start_count();
3008        let id = handle.spawn(worker);
3009        assert_eq!(id.as_u64(), 0);
3010        wait_until("the dynamic child has started", || started.load(Ordering::SeqCst) == 1).await;
3011
3012        tx.send(()).unwrap();
3013        let result = join_supervisor(run).await;
3014        assert!(result.is_ok());
3015    }
3016
3017    #[tokio::test]
3018    async fn dynamic_child_with_an_unusable_name_runs_under_a_placeholder() {
3019        // A name that sanitizes to nothing can't be used as a process name. Spawning is infallible, so rather than
3020        // quietly discarding work the caller was told had been accepted, the child runs under a placeholder segment.
3021        // The poll metric is what proves the substituted name is what the child actually ran as, rather than the child
3022        // merely having started somehow.
3023        let recorder = TestRecorder::default();
3024        let _guard = metrics::set_default_local_recorder(&recorder);
3025
3026        let sup = Supervisor::new("dyn-sup").unwrap();
3027        let handle = sup.handle();
3028        let (tx, run) = run_supervisor_with_trigger(sup).await;
3029        wait_until("supervisor is running", || handle.is_running()).await;
3030
3031        let worker = MockWorker::long_running("");
3032        let started = worker.start_count();
3033        handle.spawn(worker);
3034        wait_until("the unnamed dynamic child has started", || {
3035            started.load(Ordering::SeqCst) == 1
3036        })
3037        .await;
3038
3039        // The supervisor stays up and still accepts normally-named children.
3040        assert!(handle.is_running());
3041        handle.spawn(MockWorker::long_running("ok"));
3042        wait_until("both dynamic children are running", || handle.active_children() == 2).await;
3043
3044        tx.send(()).unwrap();
3045        let result = join_supervisor(run).await;
3046        assert!(result.is_ok());
3047
3048        let polls = recorder.counter(("runtime_task_poll_count", &[("task_name", "dyn_sup.unnamed")]));
3049        assert!(
3050            polls.is_some_and(|polls| polls > 0),
3051            "the child should have run under the placeholder name, got {polls:?}"
3052        );
3053    }
3054
3055    #[tokio::test]
3056    async fn dynamic_spawns_are_not_capped() {
3057        // Spawn requests are queued without a bound, so a burst well past what any fixed-capacity channel would hold
3058        // still starts every child. Being accepted means being started -- a depth limit could only deliver that by
3059        // discarding work the caller was already told had been taken.
3060        const CHILDREN: usize = 2048;
3061
3062        let sup = Supervisor::new("dyn-sup").unwrap();
3063        let handle = sup.handle();
3064        let (tx, run) = run_supervisor_with_trigger(sup).await;
3065        wait_until("supervisor is running", || handle.is_running()).await;
3066
3067        for _ in 0..CHILDREN {
3068            // A generous deadline: this test is about every child starting, not about how fast a couple of thousand
3069            // of them can be reaped, which is slow enough in a debug build to trip a short per-child timeout.
3070            handle.spawn(MockWorker::long_running("burst").with_graceful_timeout(Duration::from_secs(30)));
3071        }
3072
3073        wait_until("every child in the burst has started", || {
3074            handle.active_children() == CHILDREN
3075        })
3076        .await;
3077
3078        tx.send(()).unwrap();
3079        let result = timeout(Duration::from_secs(30), run)
3080            .await
3081            .expect("supervisor should stop")
3082            .expect("supervisor task should not panic");
3083        assert!(result.is_ok(), "the burst should have drained cleanly: {result:?}");
3084    }
3085
3086    #[tokio::test]
3087    async fn dynamically_spawned_supervisor_runs_and_drains() {
3088        // A dynamic child can be a whole supervision subtree, not just a worker: spawning a `Supervisor` runs it
3089        // nested, and shutting the parent down drains it along with everything under it.
3090        let child_worker = MockWorker::long_running("nested-child");
3091        let child_started = child_worker.start_count();
3092        let mut nested = Supervisor::new("nested-sup").unwrap();
3093        nested.add_worker(child_worker);
3094
3095        let sup = Supervisor::new("dyn-sup").unwrap();
3096        let handle = sup.handle();
3097        let (tx, run) = run_supervisor_with_trigger(sup).await;
3098        wait_until("supervisor is running", || handle.is_running()).await;
3099
3100        handle.spawn(nested);
3101        wait_until("the nested supervisor's own child has started", || {
3102            child_started.load(Ordering::SeqCst) == 1
3103        })
3104        .await;
3105
3106        tx.send(()).unwrap();
3107        let result = join_supervisor(run).await;
3108        assert!(
3109            result.is_ok(),
3110            "the nested subtree should have drained cleanly: {result:?}"
3111        );
3112    }
3113
3114    /// Counts how many times a nested subtree is started, by counting starts of its sole child.
3115    ///
3116    /// The child fails immediately and the subtree has a restart intensity of zero, so the subtree gives up the first
3117    /// time it fails. That makes the child's start count equal to the number of times the *subtree* ran, which is what
3118    /// these tests are actually asserting on -- without the zero intensity the subtree's own one-for-one restart would
3119    /// be indistinguishable from the parent restarting the subtree.
3120    fn failing_subtree(name: &'static str) -> (Supervisor, Arc<AtomicUsize>) {
3121        let worker = MockWorker::failing("nested-child", Duration::from_millis(5));
3122        let started = worker.start_count();
3123        let mut nested = Supervisor::new(name)
3124            .unwrap()
3125            .with_restart_strategy(RestartStrategy::new(RestartMode::OneForOne, 0, Duration::from_secs(30)));
3126        nested.add_worker(worker);
3127
3128        (nested, started)
3129    }
3130
3131    #[tokio::test]
3132    async fn dynamic_nested_supervisor_defaults_to_hosted() {
3133        // A bare `Supervisor` that is spawned gets the dynamic default, which is hosted. A supervisor can always be
3134        // started again. Thus, if a subtree fails, its parent restarts it. If the subtree continues to fail, its parent
3135        // stops at the restart limit.
3136        let (nested, started) = failing_subtree("nested-hosted");
3137
3138        let sup = Supervisor::new("dyn-hosted-sup").unwrap();
3139        let handle = sup.handle();
3140        let (_tx, run) = run_supervisor_with_trigger(sup).await;
3141
3142        handle.spawn(nested);
3143
3144        let result = join_supervisor(run).await;
3145        assert!(
3146            matches!(result, Err(SupervisorError::Shutdown)),
3147            "a subtree exceeding the limit should take its parent down, got {result:?}"
3148        );
3149        assert_eq!(started.load(Ordering::SeqCst), 2, "the subtree is restarted once");
3150    }
3151
3152    #[tokio::test]
3153    async fn dynamic_nested_supervisor_can_be_made_permanent() {
3154        // What `nested_supervisor` buys: the same subtree, spawned permanent, is brought back when it terminates.
3155        let (nested, started) = failing_subtree("nested-perm");
3156
3157        let sup = Supervisor::new("dyn-perm-sup")
3158            .unwrap()
3159            .with_restart_strategy(RestartStrategy::new(
3160                RestartMode::OneForOne,
3161                100,
3162                Duration::from_secs(30),
3163            ));
3164        let handle = sup.handle();
3165        let (tx, run) = run_supervisor_with_trigger(sup).await;
3166
3167        handle.nested_supervisor(nested).spawn();
3168        wait_until("the subtree has been restarted", || started.load(Ordering::SeqCst) >= 2).await;
3169
3170        tx.send(()).unwrap();
3171        assert!(join_supervisor(run).await.is_ok());
3172    }
3173
3174    #[tokio::test]
3175    async fn dynamic_nested_supervisor_can_be_significant() {
3176        // The other half: a subtree its parent can't function without takes the parent with it when it terminates,
3177        // rather than leaving it running with nothing behind it.
3178        let (nested, _started) = failing_subtree("nested-sig");
3179
3180        let sup = Supervisor::new("dyn-sig-sup")
3181            .unwrap()
3182            .with_auto_shutdown(AutoShutdown::AnySignificant);
3183        let handle = sup.handle();
3184        let (tx, run) = run_supervisor_with_trigger(sup).await;
3185
3186        handle
3187            .nested_supervisor(nested)
3188            .temporary()
3189            .with_significant(true)
3190            .spawn();
3191
3192        let result = timeout(Duration::from_secs(5), run)
3193            .await
3194            .expect("supervisor should stop once the significant subtree terminates")
3195            .expect("supervisor task should not panic");
3196        assert!(
3197            matches!(result, Err(SupervisorError::SignificantChildExited)),
3198            "the significant subtree's termination should have stopped the parent, got {result:?}"
3199        );
3200
3201        // The run already ended; the trigger is redundant but keeps the sender alive to the end of the test.
3202        let _ = tx.send(());
3203    }
3204
3205    #[tokio::test]
3206    async fn concurrent_shutdown_drains_many_children_quickly() {
3207        const CHILDREN: usize = 500;
3208        const SHUTDOWN_DELAY: Duration = Duration::from_millis(50);
3209
3210        let sup = Supervisor::new("dyn-sup").unwrap();
3211        let handle = sup.handle();
3212        let (tx, run) = run_supervisor_with_trigger(sup).await;
3213        wait_until("supervisor is running", || handle.is_running()).await;
3214
3215        for _ in 0..CHILDREN {
3216            handle.spawn(MockWorker::slow_shutdown("conn", SHUTDOWN_DELAY));
3217        }
3218        wait_until("all dynamic children are running", || {
3219            handle.active_children() == CHILDREN
3220        })
3221        .await;
3222
3223        // Each child sleeps after observing shutdown. Concurrent shutdown drains them all in roughly one delay; an
3224        // ordered shutdown would take CHILDREN * delay (25s here). Assert it finishes well under that.
3225        let start = std::time::Instant::now();
3226        tx.send(()).unwrap();
3227        let result = timeout(Duration::from_secs(5), run).await.unwrap().unwrap();
3228        let elapsed = start.elapsed();
3229
3230        assert!(result.is_ok());
3231        assert_eq!(handle.active_children(), 0, "active count must return to zero");
3232        assert!(
3233            elapsed < Duration::from_secs(2),
3234            "shutdown must be concurrent (took {elapsed:?})"
3235        );
3236    }
3237
3238    #[tokio::test]
3239    async fn concurrent_shutdown_aborts_unresponsive_children() {
3240        let sup = Supervisor::new("dyn-sup").unwrap();
3241        let handle = sup.handle();
3242        let (tx, run) = run_supervisor_with_trigger(sup).await;
3243        wait_until("supervisor is running", || handle.is_running()).await;
3244
3245        handle.spawn(MockWorker::ignore_shutdown("stuck"));
3246        wait_until("one dynamic child is running", || handle.active_children() == 1).await;
3247
3248        // The child never reacts to shutdown, so it must be aborted once its graceful deadline (500ms) elapses rather
3249        // than hanging the supervisor.
3250        let start = std::time::Instant::now();
3251        tx.send(()).unwrap();
3252        let result = join_supervisor(run).await;
3253        let elapsed = start.elapsed();
3254
3255        // Forcefully aborting an unresponsive child is surfaced as an unclean shutdown rather than reported as success.
3256        assert!(
3257            matches!(result, Err(SupervisorError::ShutdownTimedOut { aborted: 1 })),
3258            "aborting a stuck child must surface as an unclean shutdown, got {result:?}"
3259        );
3260        assert_eq!(handle.active_children(), 0);
3261        assert!(
3262            elapsed < Duration::from_secs(1),
3263            "stuck child must be aborted at the deadline (took {elapsed:?})"
3264        );
3265    }
3266
3267    #[tokio::test]
3268    async fn concurrent_shutdown_honors_per_child_deadline() {
3269        // Each child must be aborted at its OWN graceful deadline, not a single shared one. A responsive child with an
3270        // effectively-infinite timeout (modeling a nested supervisor, which uses `Graceful(Duration::MAX)`) coexists
3271        // with an unresponsive child with a short timeout. Under a shared `max` deadline the short-timeout child would
3272        // never be aborted (the shared deadline would be `MAX`) and shutdown would hang.
3273        let sup = Supervisor::new("dyn-sup").unwrap();
3274        let handle = sup.handle();
3275        let (tx, run) = run_supervisor_with_trigger(sup).await;
3276        wait_until("supervisor is running", || handle.is_running()).await;
3277
3278        // Responds to shutdown promptly, but its deadline is effectively infinite.
3279        handle.spawn(MockWorker::long_running("responsive").with_graceful_timeout(Duration::MAX));
3280        // Never responds; must be aborted at its own short deadline.
3281        handle.spawn(MockWorker::ignore_shutdown("stuck").with_graceful_timeout(Duration::from_millis(200)));
3282        wait_until("both dynamic children are running", || handle.active_children() == 2).await;
3283
3284        let start = std::time::Instant::now();
3285        tx.send(()).unwrap();
3286        let result = join_supervisor(run).await;
3287        let elapsed = start.elapsed();
3288
3289        // Only the stuck child is aborted (the responsive one exits cleanly), so the unclean-shutdown tally is exactly 1.
3290        assert!(
3291            matches!(result, Err(SupervisorError::ShutdownTimedOut { aborted: 1 })),
3292            "aborting the stuck child must surface as an unclean shutdown with a count of 1, got {result:?}"
3293        );
3294        assert_eq!(handle.active_children(), 0);
3295        assert!(
3296            elapsed < Duration::from_secs(1),
3297            "stuck child must be aborted at its own deadline despite an infinite-timeout sibling (took {elapsed:?})"
3298        );
3299    }
3300
3301    #[tokio::test]
3302    async fn unresponsive_child_is_aborted_at_its_deadline() {
3303        // A child that never reacts to shutdown must be aborted once its graceful deadline (500ms) elapses, rather
3304        // than hanging the supervisor indefinitely.
3305        let mut sup = Supervisor::new("test-sup").unwrap();
3306        sup.add_worker(MockWorker::ignore_shutdown("stuck"));
3307
3308        let (tx, handle) = run_supervisor_with_trigger(sup).await;
3309
3310        let start = std::time::Instant::now();
3311        tx.send(()).unwrap();
3312        let result = join_supervisor(handle).await;
3313        let elapsed = start.elapsed();
3314
3315        assert!(
3316            matches!(result, Err(SupervisorError::ShutdownTimedOut { aborted: 1 })),
3317            "aborting a stuck child must surface as an unclean shutdown, got {result:?}"
3318        );
3319        assert!(
3320            elapsed < Duration::from_secs(1),
3321            "unresponsive child must be aborted at its deadline (took {elapsed:?})"
3322        );
3323    }
3324
3325    #[tokio::test]
3326    async fn brutal_shutdown_aborts_child_immediately() {
3327        // A child with a `Brutal` shutdown strategy is aborted at once on shutdown, with no graceful wait -- so even a
3328        // child that ignores shutdown is torn down promptly rather than after the graceful deadline.
3329        let mut sup = Supervisor::new("test-sup").unwrap();
3330        sup.add_worker(MockWorker::ignore_shutdown("brutal-stuck").with_brutal_shutdown());
3331
3332        let (tx, handle) = run_supervisor_with_trigger(sup).await;
3333
3334        let start = std::time::Instant::now();
3335        tx.send(()).unwrap();
3336        let result = join_supervisor(handle).await;
3337        let elapsed = start.elapsed();
3338
3339        // A brutal abort is the configured, expected way to stop this child -- not a graceful-timeout overrun -- so it
3340        // is NOT counted toward the unclean-shutdown tally, and the shutdown reports success.
3341        assert!(result.is_ok());
3342        assert!(
3343            elapsed < Duration::from_millis(200),
3344            "brutal-shutdown child must be aborted immediately, not after a graceful wait (took {elapsed:?})"
3345        );
3346    }
3347
3348    #[tokio::test]
3349    async fn shutdown_timeout_aborts_aggregate_to_root() {
3350        // Forced aborts must surface as an unclean shutdown and aggregate up the tree: a supervisor adds the workers it
3351        // aborts directly to the counts reported by any child supervisors that also timed out. Here the parent aborts
3352        // one direct child and a nested supervisor aborts one of its own, so the root observes a total of 2.
3353        let mut child_sup = Supervisor::new("child-sup").unwrap();
3354        child_sup
3355            .add_worker(MockWorker::ignore_shutdown("child-stuck").with_graceful_timeout(Duration::from_millis(200)));
3356
3357        let mut parent_sup = Supervisor::new("parent-sup").unwrap();
3358        parent_sup
3359            .add_worker(MockWorker::ignore_shutdown("parent-stuck").with_graceful_timeout(Duration::from_millis(200)));
3360        parent_sup.add_worker(MockWorker::long_running("parent-clean"));
3361        parent_sup.add_worker(child_sup);
3362
3363        let (tx, handle) = run_supervisor_with_trigger(parent_sup).await;
3364        tx.send(()).unwrap();
3365
3366        let result = join_supervisor(handle).await;
3367        assert!(
3368            matches!(result, Err(SupervisorError::ShutdownTimedOut { aborted: 2 })),
3369            "forced aborts must aggregate across the tree (1 direct + 1 nested), got {result:?}"
3370        );
3371    }
3372
3373    // -- Restart-policy edge cases ---------------------------------------------------------
3374
3375    #[tokio::test]
3376    async fn restart_intensity_zero_shuts_down_on_first_failure() {
3377        // A restart intensity of zero means the supervisor gives up the moment any restartable child fails: it shuts
3378        // down on the very first failure without ever restarting the worker. (See `RestartState::evaluate_restart`,
3379        // which short-circuits to `Shutdown` when intensity is zero.)
3380        let worker = MockWorker::failing("boom", Duration::from_millis(20));
3381        let start_count = worker.start_count();
3382
3383        let mut sup = Supervisor::new("test-sup")
3384            .unwrap()
3385            .with_restart_strategy(RestartStrategy::new(RestartMode::OneForOne, 0, Duration::from_secs(5)));
3386        sup.add_worker(worker);
3387
3388        let (_tx, rx) = oneshot::channel::<()>();
3389        let result = timeout(Duration::from_secs(2), sup.run_with_shutdown(rx))
3390            .await
3391            .unwrap();
3392
3393        assert!(matches!(result, Err(SupervisorError::Shutdown)));
3394        assert_eq!(
3395            start_count.load(Ordering::SeqCst),
3396            1,
3397            "with intensity zero the worker must run exactly once and never be restarted"
3398        );
3399    }
3400
3401    #[tokio::test]
3402    async fn one_for_all_restart_loses_dynamic_children() {
3403        // Documented one-for-all semantics: a group restart resets to the static roster only -- dynamic children are
3404        // NOT restored (they're lost on a supervisor-level restart, matching Erlang/OTP). A permanent static worker
3405        // that keeps failing drives repeated one-for-all restarts; a dynamic child spawned before the first restart
3406        // must be torn down and never brought back.
3407        let failing = MockWorker::failing("failing-static", Duration::from_millis(50));
3408        let failing_count = failing.start_count();
3409
3410        let sup = Supervisor::new("dyn-sup").unwrap().with_restart_strategy(
3411            RestartStrategy::one_for_all().with_intensity_and_period(20, Duration::from_secs(10)),
3412        );
3413        let handle = sup.handle();
3414        let mut sup = sup;
3415        sup.add_worker(failing);
3416
3417        let (tx, run) = run_supervisor_with_trigger(sup).await;
3418
3419        // Spawn a long-running dynamic child and wait for it to be running.
3420        let dynamic = MockWorker::long_running("dynamic");
3421        let dynamic_count = dynamic.start_count();
3422        handle.spawn(dynamic);
3423        wait_until("the dynamic child is running", || handle.active_children() == 1).await;
3424
3425        // Let the static worker drive at least one one-for-all restart (its second start).
3426        wait_until("the static worker has been restarted", || {
3427            failing_count.load(Ordering::SeqCst) >= 2
3428        })
3429        .await;
3430
3431        // The one-for-all restart must have discarded the dynamic child: the active count returns to zero, and the
3432        // dynamic child ran exactly once (it was never restored).
3433        wait_until("the dynamic child has been discarded", || handle.active_children() == 0).await;
3434        assert_eq!(
3435            dynamic_count.load(Ordering::SeqCst),
3436            1,
3437            "a dynamic child must be lost -- not restored -- across a one-for-all restart"
3438        );
3439
3440        tx.send(()).unwrap();
3441        let result = join_supervisor(run).await;
3442        assert!(result.is_ok());
3443    }
3444
3445    // -- Dedicated-runtime tests -----------------------------------------------------------
3446
3447    #[tokio::test]
3448    async fn dedicated_single_threaded_runtime_runs_nested_worker_and_shuts_down_cleanly() {
3449        // A nested supervisor configured with a dedicated single-threaded runtime spawns its own OS thread and Tokio
3450        // runtime (via `spawn_dedicated_runtime`). Its worker must run there, and a shutdown signalled by the parent
3451        // must propagate across the thread boundary and tear it down cleanly.
3452        let worker = MockWorker::long_running("dedicated-worker");
3453        let worker_count = worker.start_count();
3454
3455        let mut child_sup = Supervisor::new("child-sup")
3456            .unwrap()
3457            .with_dedicated_runtime(RuntimeConfiguration::single_threaded());
3458        child_sup.add_worker(worker);
3459
3460        let mut parent_sup = Supervisor::new("parent-sup").unwrap();
3461        parent_sup.add_worker(child_sup);
3462
3463        let (tx, handle) = run_supervisor_with_trigger(parent_sup).await;
3464
3465        // The worker starts on the dedicated runtime's own thread.
3466        wait_until("the dedicated worker has started", || {
3467            worker_count.load(Ordering::SeqCst) == 1
3468        })
3469        .await;
3470
3471        tx.send(()).unwrap();
3472        let result = join_supervisor(handle).await;
3473        assert!(
3474            result.is_ok(),
3475            "dedicated-runtime supervisor should shut down cleanly, got {result:?}"
3476        );
3477    }
3478
3479    #[tokio::test]
3480    async fn dedicated_multi_threaded_runtime_runs_nested_worker() {
3481        // The same nested-dedicated flow, but exercising the multi-threaded dedicated runtime builder path.
3482        let worker = MockWorker::long_running("dedicated-worker");
3483        let worker_count = worker.start_count();
3484
3485        let mut child_sup = Supervisor::new("child-sup")
3486            .unwrap()
3487            .with_dedicated_runtime(RuntimeConfiguration::multi_threaded(2));
3488        child_sup.add_worker(worker);
3489
3490        let mut parent_sup = Supervisor::new("parent-sup").unwrap();
3491        parent_sup.add_worker(child_sup);
3492
3493        let (tx, handle) = run_supervisor_with_trigger(parent_sup).await;
3494        wait_until("the dedicated worker has started", || {
3495            worker_count.load(Ordering::SeqCst) == 1
3496        })
3497        .await;
3498
3499        tx.send(()).unwrap();
3500        let result = join_supervisor(handle).await;
3501        assert!(
3502            result.is_ok(),
3503            "multi-threaded dedicated-runtime supervisor should shut down cleanly, got {result:?}"
3504        );
3505    }
3506
3507    #[tokio::test]
3508    async fn dedicated_runtime_forced_abort_aggregates_to_root() {
3509        // A worker inside a dedicated-runtime nested supervisor that ignores shutdown must be forcefully aborted at its
3510        // deadline, and that abort tally must survive the OS-thread boundary (`DedicatedRuntimeHandle` -> `WorkerError`)
3511        // and be observed by the root supervisor as `ShutdownTimedOut`.
3512        let stuck = MockWorker::ignore_shutdown("stuck").with_graceful_timeout(Duration::from_millis(200));
3513        let stuck_count = stuck.start_count();
3514
3515        let mut child_sup = Supervisor::new("child-sup")
3516            .unwrap()
3517            .with_dedicated_runtime(RuntimeConfiguration::single_threaded());
3518        child_sup.add_worker(stuck);
3519
3520        let mut parent_sup = Supervisor::new("parent-sup").unwrap();
3521        parent_sup.add_worker(child_sup);
3522
3523        let (tx, handle) = run_supervisor_with_trigger(parent_sup).await;
3524
3525        // Make sure the stuck worker is actually running on the dedicated runtime before signalling shutdown, so the
3526        // forced-abort path (rather than an early exit) is what we exercise.
3527        wait_until("the stuck worker has started", || {
3528            stuck_count.load(Ordering::SeqCst) == 1
3529        })
3530        .await;
3531
3532        tx.send(()).unwrap();
3533        let result = join_supervisor(handle).await;
3534        assert!(
3535            matches!(result, Err(SupervisorError::ShutdownTimedOut { aborted: 1 })),
3536            "a stuck worker in a dedicated runtime must surface as an unclean shutdown aggregated to the root, got {result:?}"
3537        );
3538    }
3539
3540    // -- Per-child override tests ----------------------------------------------------------
3541
3542    #[tokio::test]
3543    async fn child_with_runtime_override_runs_on_that_runtime() {
3544        // `with_runtime` places an individual child's task on a caller-provided runtime instead of the supervisor's
3545        // own. The child reports the name of the thread it's actually running on, which must belong to that runtime.
3546        let child_runtime = tokio::runtime::Builder::new_multi_thread()
3547            .worker_threads(1)
3548            .thread_name("child-rt-test")
3549            .enable_all()
3550            .build()
3551            .expect("should build child runtime");
3552
3553        let (thread_tx, thread_rx) = oneshot::channel();
3554        let worker = FnWorker::new("placed", async move {
3555            let thread_name = std::thread::current().name().unwrap_or_default().to_string();
3556            let _ = thread_tx.send(thread_name);
3557            pending::<()>().await;
3558        });
3559
3560        // The worker only has to stay alive long enough to report where it ran, so it's aborted at shutdown rather
3561        // than given a terminal condition to reach.
3562        let mut sup = Supervisor::new("test-sup").unwrap();
3563        sup.add_worker(
3564            ChildSpecification::worker(worker)
3565                .with_restart_type(RestartType::Temporary)
3566                .with_runtime(child_runtime.handle().clone())
3567                .with_shutdown_strategy(ShutdownStrategy::Brutal),
3568        );
3569
3570        let (tx, handle) = run_supervisor_with_trigger(sup).await;
3571
3572        let thread_name = timeout(Duration::from_secs(2), thread_rx)
3573            .await
3574            .expect("child should report its thread promptly")
3575            .expect("child should not be dropped before reporting");
3576        assert!(
3577            thread_name.starts_with("child-rt-test"),
3578            "child must run on the runtime given to `with_runtime`, but ran on thread {thread_name:?}"
3579        );
3580
3581        tx.send(()).unwrap();
3582        assert!(join_supervisor(handle).await.is_ok());
3583
3584        // Dropping a `Runtime` from within an async context panics, so tear it down without blocking.
3585        child_runtime.shutdown_background();
3586    }
3587
3588    #[tokio::test]
3589    async fn child_shutdown_strategy_override_takes_precedence_over_worker() {
3590        // `with_shutdown_strategy` overrides what the worker reports for itself. The worker below asks for a 30-second
3591        // grace period and then ignores shutdown entirely; the override cuts that to 50ms, so the supervisor must
3592        // abort it and report an unclean shutdown well inside `join_supervisor`'s two-second bound. Without the
3593        // override taking precedence, this test times out.
3594        let worker = MockWorker::ignore_shutdown("stuck").with_graceful_timeout(Duration::from_secs(30));
3595
3596        let mut sup = Supervisor::new("test-sup").unwrap();
3597        sup.add_worker(
3598            ChildSpecification::worker(worker)
3599                .with_restart_type(RestartType::Temporary)
3600                .with_shutdown_strategy(ShutdownStrategy::Graceful(Duration::from_millis(50))),
3601        );
3602
3603        let (tx, handle) = run_supervisor_with_trigger(sup).await;
3604        tx.send(()).unwrap();
3605
3606        let result = join_supervisor(handle).await;
3607        assert!(
3608            matches!(result, Err(SupervisorError::ShutdownTimedOut { aborted: 1 })),
3609            "the overridden 50ms deadline should have aborted the stuck child, got {result:?}"
3610        );
3611    }
3612
3613    /// A worker that waits for shutdown and then drains a [`ShutdownCoordinator`] before exiting.
3614    ///
3615    /// Stands in for a component that owns background work and waits for it during its own shutdown. Unlike a
3616    /// closure-based worker it genuinely needs the shutdown signal, which is exactly the case [`Supervisable`] exists
3617    /// for.
3618    struct DrainWaiter {
3619        coordinator: Mutex<Option<ShutdownCoordinator>>,
3620        finished: Arc<AtomicBool>,
3621    }
3622
3623    #[async_trait]
3624    impl Supervisable for DrainWaiter {
3625        fn name(&self) -> &str {
3626            "waiter"
3627        }
3628
3629        async fn initialize(&self, process_shutdown: ShutdownHandle) -> Result<SupervisorFuture, InitializationError> {
3630            let coordinator = self
3631                .coordinator
3632                .lock()
3633                .expect("drain waiter mutex poisoned")
3634                .take()
3635                .expect("drain waiter runs once");
3636            let finished = Arc::clone(&self.finished);
3637
3638            Ok(Box::pin(async move {
3639                process_shutdown.await;
3640                coordinator.shutdown_and_wait().await;
3641                finished.store(true, Ordering::SeqCst);
3642                Ok(())
3643            }))
3644        }
3645    }
3646
3647    /// Builds two children modelling an owner that drains a background task during shutdown.
3648    ///
3649    /// `stuck` holds a shutdown handle and never releases it voluntarily, so the only way it goes away is a forced
3650    /// abort. `waiter` blocks on that handle being dropped, standing in for a component's `shutdown_and_wait`. The
3651    /// returned flag records whether `waiter` ran to completion rather than being aborted itself.
3652    fn build_drain_pair(
3653        stuck_strategy: ShutdownStrategy, waiter_strategy: ShutdownStrategy,
3654    ) -> (Supervisor, Arc<AtomicBool>) {
3655        let mut coordinator = ShutdownCoordinator::default();
3656        let held_handle = coordinator.register();
3657
3658        let stuck = FnWorker::new("stuck", async move {
3659            // Hold the handle for as long as this future lives, and ignore shutdown entirely.
3660            let _held = held_handle;
3661            pending::<()>().await;
3662        });
3663
3664        let waiter_finished = Arc::new(AtomicBool::new(false));
3665        let waiter = DrainWaiter {
3666            coordinator: Mutex::new(Some(coordinator)),
3667            finished: Arc::clone(&waiter_finished),
3668        };
3669
3670        let mut sup = Supervisor::new("test-sup").unwrap();
3671        sup.add_worker(
3672            ChildSpecification::worker(stuck)
3673                .with_restart_type(RestartType::Temporary)
3674                .with_shutdown_strategy(stuck_strategy),
3675        );
3676        sup.add_worker(
3677            ChildSpecification::worker(waiter)
3678                .with_restart_type(RestartType::Temporary)
3679                .with_shutdown_strategy(waiter_strategy),
3680        );
3681
3682        (sup, waiter_finished)
3683    }
3684
3685    #[tokio::test]
3686    async fn shorter_child_deadline_releases_a_waiting_sibling() {
3687        // Aborting a stuck child drops the shutdown handle it was holding, which is what releases anything waiting on
3688        // it. A child bounded more tightly than its waiter therefore stays recoverable: the child is aborted, the
3689        // waiter unblocks and finishes cleanly, and only one abort is reported.
3690        let (sup, waiter_finished) = build_drain_pair(
3691            ShutdownStrategy::Graceful(Duration::from_millis(100)),
3692            ShutdownStrategy::Graceful(Duration::from_secs(1)),
3693        );
3694
3695        let (tx, handle) = run_supervisor_with_trigger(sup).await;
3696        tx.send(()).unwrap();
3697
3698        let result = join_supervisor(handle).await;
3699        assert!(
3700            matches!(result, Err(SupervisorError::ShutdownTimedOut { aborted: 1 })),
3701            "only the stuck child should have been aborted, got {result:?}"
3702        );
3703        assert!(
3704            waiter_finished.load(Ordering::SeqCst),
3705            "the waiter should have been released by the stuck child's abort and run to completion"
3706        );
3707    }
3708
3709    #[tokio::test]
3710    async fn equal_child_deadlines_abort_the_waiter_too() {
3711        // The counterpart: concurrent shutdown computes every deadline from one shared instant, so identical timeouts
3712        // elapse in the same pass and the waiter is aborted alongside the child it was waiting on. This is also what an
3713        // owner's expired timeout does to everything in its scope.
3714        let (sup, waiter_finished) = build_drain_pair(
3715            ShutdownStrategy::Graceful(Duration::from_millis(100)),
3716            ShutdownStrategy::Graceful(Duration::from_millis(100)),
3717        );
3718
3719        let (tx, handle) = run_supervisor_with_trigger(sup).await;
3720        tx.send(()).unwrap();
3721
3722        let result = join_supervisor(handle).await;
3723        assert!(
3724            matches!(result, Err(SupervisorError::ShutdownTimedOut { aborted: 2 })),
3725            "both children should have been aborted together, got {result:?}"
3726        );
3727        assert!(
3728            !waiter_finished.load(Ordering::SeqCst),
3729            "the waiter should have been aborted mid-wait, not completed"
3730        );
3731    }
3732
3733    #[tokio::test]
3734    async fn every_worker_records_poll_metrics() {
3735        // Poll timing is a property of being supervised, not something a child opts into, so a plain statically
3736        // registered worker gets it too -- tagged with its fully qualified process name.
3737        let recorder = TestRecorder::default();
3738        let _guard = metrics::set_default_local_recorder(&recorder);
3739
3740        // The recorder has to be installed before the worker spawns: its metric handles are resolved once, at spawn.
3741        let mut sup = Supervisor::new("metrics_sup").unwrap();
3742        sup.add_worker(ChildSpecification::worker(MockWorker::long_running("timed").one_shot()));
3743
3744        let (tx, handle) = run_supervisor_with_trigger(sup).await;
3745        tx.send(()).unwrap();
3746        assert!(join_supervisor(handle).await.is_ok());
3747
3748        let polls = recorder.counter(("runtime_task_poll_count", &[("task_name", "metrics_sup.timed")]));
3749        assert!(
3750            polls.is_some_and(|polls| polls > 0),
3751            "a supervised worker should have recorded poll metrics, got {polls:?}"
3752        );
3753    }
3754
3755    #[tokio::test]
3756    async fn near_max_child_timeout_does_not_panic() {
3757        // A timeout just under `Duration::MAX` overflows when added to an instant without `checked_add`.
3758        let mut sup = Supervisor::new("test-sup").unwrap();
3759        sup.add_worker(
3760            ChildSpecification::worker(MockWorker::long_running("prompt").one_shot())
3761                .with_shutdown_strategy(ShutdownStrategy::Graceful(Duration::MAX - Duration::from_nanos(1))),
3762        );
3763
3764        let (tx, handle) = run_supervisor_with_trigger(sup).await;
3765        tx.send(()).unwrap();
3766        assert!(join_supervisor(handle).await.is_ok());
3767    }
3768
3769    // -- Supervision tree snapshot tests ---------------------------------------------------
3770
3771    fn find_node<'a>(nodes: &'a [NodeSnapshot], name: &str) -> &'a NodeSnapshot {
3772        nodes.iter().find(|node| node.name == name).unwrap_or_else(|| {
3773            panic!(
3774                "no node named '{name}' among {:?}",
3775                nodes.iter().map(|n| &n.name).collect::<Vec<_>>()
3776            )
3777        })
3778    }
3779
3780    #[tokio::test]
3781    async fn snapshot_before_run_reports_a_registered_root() {
3782        let mut sup = Supervisor::new("test-sup")
3783            .unwrap()
3784            .with_restart_strategy(RestartStrategy::one_for_all().with_intensity_and_period(7, Duration::from_secs(11)))
3785            .with_auto_shutdown(AutoShutdown::AnySignificant);
3786        sup.add_worker(MockWorker::long_running("worker1"));
3787
3788        // A tree that hasn't started still has a declared shape, and configuration is known from the moment it is
3789        // set, so all of it should be readable without running anything.
3790        let snapshot = sup.tree_handle().snapshot();
3791
3792        assert_eq!(snapshot.root.name, "test-sup");
3793        assert_eq!(snapshot.root.kind, NodeKind::Supervisor);
3794        assert_eq!(snapshot.root.state, NodeState::Registered);
3795        assert_eq!(snapshot.root.process_id, None);
3796        assert_eq!(snapshot.root.process_name, None);
3797        assert_eq!(snapshot.root.started_at, None);
3798        assert_eq!(snapshot.root.uptime_ms, None);
3799        assert!(snapshot.root.children.is_empty(), "no children have been started yet");
3800
3801        let supervision = snapshot.root.supervision.expect("a supervisor reports its settings");
3802        assert_eq!(supervision.restart_mode, RestartMode::OneForAll);
3803        assert_eq!(supervision.restart_intensity, 7);
3804        assert_eq!(supervision.restart_period_ms, 11_000);
3805        assert_eq!(supervision.auto_shutdown, AutoShutdown::AnySignificant);
3806        assert_eq!(supervision.dedicated_threads, None);
3807        assert_eq!(supervision.generation, 0);
3808
3809        assert_eq!(snapshot.totals.supervisors, 1);
3810        assert_eq!(snapshot.totals.workers, 0);
3811        assert_eq!(snapshot.totals.registered, 1);
3812        assert_eq!(snapshot.totals.max_depth, 1);
3813    }
3814
3815    #[tokio::test]
3816    async fn snapshot_lists_static_children_in_declaration_order() {
3817        let mut sup = Supervisor::new("test-sup").unwrap();
3818        sup.add_worker(MockWorker::long_running("worker1"));
3819        sup.add_worker(MockWorker::long_running("worker2"));
3820
3821        let tree = sup.tree_handle();
3822        let (tx, handle) = run_supervisor_with_trigger(sup).await;
3823        wait_until("both children appear in the snapshot", || {
3824            tree.snapshot().root.children.len() == 2
3825        })
3826        .await;
3827
3828        let snapshot = tree.snapshot();
3829        assert_eq!(snapshot.root.state, NodeState::Running);
3830        assert!(snapshot.root.process_id.is_some());
3831        assert_eq!(snapshot.root.process_name.as_deref(), Some("test_sup"));
3832        assert_eq!(snapshot.root.supervision.expect("supervisor settings").generation, 1);
3833
3834        // Declaration order, not hash order: it is how the tree is written, so it is how it should read.
3835        let names: Vec<&str> = snapshot.root.children.iter().map(|c| c.name.as_str()).collect();
3836        assert_eq!(names, vec!["worker1", "worker2"]);
3837
3838        for (child, expected_process) in snapshot
3839            .root
3840            .children
3841            .iter()
3842            .zip(["test_sup.worker1", "test_sup.worker2"])
3843        {
3844            assert_eq!(child.kind, NodeKind::Worker);
3845            assert_eq!(child.state, NodeState::Running);
3846            assert_eq!(child.restart, RestartType::Permanent);
3847            assert_eq!(child.restart_count, 0);
3848            assert_eq!(child.process_name.as_deref(), Some(expected_process));
3849            assert!(child.process_id.is_some(), "a running child has a process");
3850            assert!(child.children.is_empty(), "a worker has no children");
3851            assert!(child.supervision.is_none(), "a worker has no supervision settings");
3852            assert!(
3853                child.created_at <= child.started_at.expect("a running child has a start time"),
3854                "a child cannot start before it is created"
3855            );
3856        }
3857
3858        let first = snapshot.root.children[0].process_id;
3859        let second = snapshot.root.children[1].process_id;
3860        assert_ne!(first, second, "each child runs as its own process");
3861
3862        assert_eq!(snapshot.totals.supervisors, 1);
3863        assert_eq!(snapshot.totals.workers, 2);
3864        assert_eq!(snapshot.totals.running, 3);
3865        assert_eq!(snapshot.totals.max_depth, 2);
3866
3867        tx.send(()).unwrap();
3868        let _ = join_supervisor(handle).await;
3869    }
3870
3871    #[tokio::test]
3872    async fn one_for_one_restart_increments_only_the_failing_child() {
3873        let mut sup = Supervisor::new("test-sup")
3874            .unwrap()
3875            .with_restart_strategy(RestartStrategy::one_to_one().with_intensity_and_period(5, Duration::from_secs(5)));
3876        sup.add_worker(MockWorker::failing("flapper", Duration::from_millis(20)));
3877        sup.add_worker(MockWorker::long_running("stable"));
3878
3879        let tree = sup.tree_handle();
3880        let (tx, handle) = run_supervisor_with_trigger(sup).await;
3881
3882        wait_until("the failing child has appeared", || {
3883            tree.snapshot().root.children.len() == 2
3884        })
3885        .await;
3886        let before = tree.snapshot();
3887        let created_before = find_node(&before.root.children, "flapper").created_at;
3888        let process_before = find_node(&before.root.children, "flapper").process_id;
3889
3890        wait_until("the failing child has been restarted", || {
3891            find_node(&tree.snapshot().root.children, "flapper").restart_count >= 1
3892        })
3893        .await;
3894
3895        let after = tree.snapshot();
3896        let flapper = find_node(&after.root.children, "flapper");
3897        let stable = find_node(&after.root.children, "stable");
3898
3899        assert_eq!(stable.restart_count, 0, "a one-for-one restart leaves siblings alone");
3900        assert_ne!(flapper.process_id, process_before, "a restart is a new process");
3901        assert_eq!(
3902            flapper.created_at, created_before,
3903            "creation time is a property of the child, not of the incarnation"
3904        );
3905
3906        // A single child flapping is distinguishable from the supervisor restarting its whole group.
3907        assert!(after.root.supervision.expect("supervisor settings").restarts_performed >= 1);
3908
3909        tx.send(()).unwrap();
3910        let _ = join_supervisor(handle).await;
3911    }
3912
3913    #[tokio::test]
3914    async fn one_for_all_restart_increments_every_restarted_child() {
3915        let mut sup = Supervisor::new("test-sup")
3916            .unwrap()
3917            .with_restart_strategy(RestartStrategy::one_for_all().with_intensity_and_period(5, Duration::from_secs(5)));
3918        sup.add_worker(MockWorker::failing("flapper", Duration::from_millis(20)));
3919        sup.add_worker(MockWorker::long_running("sibling"));
3920
3921        let tree = sup.tree_handle();
3922        let (tx, handle) = run_supervisor_with_trigger(sup).await;
3923
3924        wait_until("both children have appeared", || {
3925            tree.snapshot().root.children.len() == 2
3926        })
3927        .await;
3928        let before = tree.snapshot();
3929        let sibling_created = find_node(&before.root.children, "sibling").created_at;
3930        let sibling_process = find_node(&before.root.children, "sibling").process_id;
3931
3932        // The whole group is brought back, so every child's count moves -- which is what keeps each child's count
3933        // consistent with the new process and start time in the same record.
3934        wait_until("the group has been restarted", || {
3935            let snapshot = tree.snapshot();
3936            snapshot.root.children.len() == 2 && snapshot.root.children.iter().all(|child| child.restart_count >= 1)
3937        })
3938        .await;
3939
3940        let after = tree.snapshot();
3941        let sibling = find_node(&after.root.children, "sibling");
3942        assert_ne!(sibling.process_id, sibling_process, "a group restart is a new process");
3943        assert_eq!(
3944            sibling.created_at, sibling_created,
3945            "creation time survives a group restart"
3946        );
3947
3948        tx.send(()).unwrap();
3949        let _ = join_supervisor(handle).await;
3950    }
3951
3952    #[tokio::test]
3953    async fn temporary_child_survives_a_group_restart_as_a_tombstone() {
3954        let mut sup = Supervisor::new("test-sup")
3955            .unwrap()
3956            .with_restart_strategy(RestartStrategy::one_for_all().with_intensity_and_period(5, Duration::from_secs(5)));
3957        sup.add_worker(
3958            runtime::supervisable(MockWorker::completing("one-shot", Duration::from_millis(10)))
3959                .temporary()
3960                .build(),
3961        );
3962        sup.add_worker(MockWorker::failing("flapper", Duration::from_millis(30)));
3963
3964        let tree = sup.tree_handle();
3965        let (tx, handle) = run_supervisor_with_trigger(sup).await;
3966
3967        // A temporary child is not brought back by a group restart, but it was declared, so it stays visible as a
3968        // tombstone rather than vanishing -- "declared, ran, stopped for good" is the useful answer.
3969        wait_until("the temporary child has exited and the group has restarted", || {
3970            let snapshot = tree.snapshot();
3971            snapshot
3972                .root
3973                .children
3974                .iter()
3975                .any(|child| child.name == "one-shot" && child.state == NodeState::Exited)
3976                && snapshot
3977                    .root
3978                    .children
3979                    .iter()
3980                    .any(|child| child.name == "flapper" && child.restart_count >= 1)
3981        })
3982        .await;
3983
3984        let snapshot = tree.snapshot();
3985        let one_shot = find_node(&snapshot.root.children, "one-shot");
3986        assert_eq!(one_shot.state, NodeState::Exited);
3987        assert_eq!(one_shot.restart, RestartType::Temporary);
3988        assert_eq!(one_shot.restart_count, 0, "a temporary child is never brought back");
3989        assert!(one_shot.exited_at.is_some(), "a tombstone records when it exited");
3990        assert_eq!(one_shot.uptime_ms, None, "a node that isn't running has no uptime");
3991        assert!(snapshot.totals.exited >= 1);
3992
3993        tx.send(()).unwrap();
3994        let _ = join_supervisor(handle).await;
3995    }
3996
3997    #[tokio::test]
3998    async fn dynamic_children_leave_no_tombstones() {
3999        let mut sup = Supervisor::new("test-sup").unwrap();
4000        sup.add_worker(MockWorker::long_running("static-worker"));
4001
4002        let tree = sup.tree_handle();
4003        let sup_handle = sup.handle();
4004        let (tx, handle) = run_supervisor_with_trigger(sup).await;
4005        wait_until("the static child is running", || {
4006            tree.snapshot().root.children.len() == 1
4007        })
4008        .await;
4009
4010        // The canonical dynamic child is one short-lived task per unit of work, so retaining every one that has ever
4011        // finished would grow without bound. Spawn far more than a tree should ever hold and require the count to
4012        // come back to the static baseline.
4013        for _ in 0..500 {
4014            sup_handle.spawn(MockWorker::completing("ephemeral", Duration::from_millis(1)));
4015        }
4016
4017        wait_until("every dynamic child has been reaped", || {
4018            sup_handle.active_children() == 0 && tree.snapshot().root.children.len() == 1
4019        })
4020        .await;
4021
4022        let snapshot = tree.snapshot();
4023        assert_eq!(snapshot.root.children.len(), 1, "only the static child remains");
4024        assert_eq!(snapshot.root.children[0].name, "static-worker");
4025
4026        tx.send(()).unwrap();
4027        let _ = join_supervisor(handle).await;
4028    }
4029
4030    #[tokio::test]
4031    async fn snapshot_nests_child_supervisors() {
4032        let mut child_sup = Supervisor::new("child-sup").unwrap();
4033        child_sup.add_worker(MockWorker::long_running("inner-worker"));
4034
4035        let mut parent_sup = Supervisor::new("parent-sup").unwrap();
4036        parent_sup.add_worker(MockWorker::long_running("outer-worker"));
4037        parent_sup.add_worker(child_sup);
4038
4039        let tree = parent_sup.tree_handle();
4040        let (tx, handle) = run_supervisor_with_trigger(parent_sup).await;
4041
4042        wait_until("the nested subtree is populated", || {
4043            let snapshot = tree.snapshot();
4044            snapshot
4045                .root
4046                .children
4047                .iter()
4048                .any(|child| child.name == "child-sup" && !child.children.is_empty())
4049        })
4050        .await;
4051
4052        let snapshot = tree.snapshot();
4053        let nested = find_node(&snapshot.root.children, "child-sup");
4054        assert_eq!(nested.kind, NodeKind::Supervisor);
4055        assert!(nested.supervision.is_some(), "a nested supervisor reports its settings");
4056        assert_eq!(nested.process_name.as_deref(), Some("parent_sup.child_sup"));
4057        assert_eq!(nested.children.len(), 1);
4058
4059        let grandchild = &nested.children[0];
4060        assert_eq!(grandchild.name, "inner-worker");
4061        assert_eq!(grandchild.kind, NodeKind::Worker);
4062        assert_eq!(
4063            grandchild.process_name.as_deref(),
4064            Some("parent_sup.child_sup.inner_worker")
4065        );
4066
4067        assert_eq!(snapshot.totals.supervisors, 2);
4068        assert_eq!(snapshot.totals.workers, 2);
4069        assert_eq!(snapshot.totals.max_depth, 3);
4070
4071        tx.send(()).unwrap();
4072        let _ = join_supervisor(handle).await;
4073    }
4074
4075    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
4076    async fn dedicated_runtime_subtree_is_visible_across_the_thread_boundary() {
4077        let mut child_sup = Supervisor::new("child-sup")
4078            .unwrap()
4079            .with_dedicated_runtime(RuntimeConfiguration::single_threaded());
4080        child_sup.add_worker(MockWorker::long_running("inner-worker"));
4081
4082        let mut parent_sup = Supervisor::new("parent-sup").unwrap();
4083        parent_sup.add_worker(child_sup);
4084
4085        // Taken from the parent, before the child is running on an OS thread of its own.
4086        let tree = parent_sup.tree_handle();
4087        let (tx, handle) = run_supervisor_with_trigger(parent_sup).await;
4088
4089        wait_until("the dedicated-runtime subtree is visible", || {
4090            tree.snapshot()
4091                .root
4092                .children
4093                .iter()
4094                .any(|child| child.name == "child-sup" && !child.children.is_empty())
4095        })
4096        .await;
4097
4098        // Read the tree from a thread that is neither the parent's nor the dedicated runtime's.
4099        let probe = tree.clone();
4100        let snapshot = tokio::task::spawn_blocking(move || probe.snapshot())
4101            .await
4102            .expect("probe should not panic");
4103
4104        let nested = find_node(&snapshot.root.children, "child-sup");
4105        assert_eq!(nested.state, NodeState::Running);
4106        assert_eq!(
4107            nested.supervision.expect("supervisor settings").dedicated_threads,
4108            Some(1)
4109        );
4110        assert_eq!(nested.children.len(), 1);
4111
4112        // A supervisor on a dedicated runtime re-roots its process name when it starts, so it is *not* scoped under
4113        // its parent. That is a known defect (see the `Dedicated` branch of `create_worker_future`), not a design
4114        // choice -- but it is also why a node's name has to come from its own run rather than from the process its
4115        // parent created for it, since only the former names the resource group its allocations land in. Pinned here
4116        // so that fixing the defect is a deliberate act rather than a silent regression.
4117        assert_eq!(nested.process_name.as_deref(), Some("child_sup"));
4118        assert_eq!(
4119            nested.children[0].process_name.as_deref(),
4120            Some("child_sup.inner_worker")
4121        );
4122
4123        tx.send(()).unwrap();
4124        let _ = join_supervisor(handle).await;
4125    }
4126
4127    #[tokio::test]
4128    async fn a_running_node_always_has_a_process() {
4129        // A nested supervisor being restarted is briefly stopped, and a snapshot taken in that window must not
4130        // present the previous generation's processes as if they were alive. Sampling exactly inside the window is
4131        // inherently racy, so assert the invariant that has to hold at every instant instead.
4132        let mut child_sup = Supervisor::new("child-sup")
4133            .unwrap()
4134            .with_restart_strategy(RestartStrategy::new(RestartMode::OneForOne, 0, Duration::from_secs(5)));
4135        child_sup.add_worker(MockWorker::failing("inner", Duration::from_millis(10)));
4136
4137        let mut parent_sup = Supervisor::new("parent-sup")
4138            .unwrap()
4139            .with_restart_strategy(RestartStrategy::one_to_one().with_intensity_and_period(50, Duration::from_secs(5)));
4140        parent_sup.add_worker(child_sup);
4141
4142        let tree = parent_sup.tree_handle();
4143        let (tx, handle) = run_supervisor_with_trigger(parent_sup).await;
4144
4145        for _ in 0..60 {
4146            let snapshot = tree.snapshot();
4147            assert_running_nodes_have_processes(&snapshot.root);
4148            tokio::time::sleep(Duration::from_millis(5)).await;
4149        }
4150
4151        wait_until("the nested supervisor has been restarted", || {
4152            find_node(&tree.snapshot().root.children, "child-sup").restart_count >= 1
4153        })
4154        .await;
4155
4156        tx.send(()).unwrap();
4157        let _ = join_supervisor(handle).await;
4158    }
4159
4160    fn assert_running_nodes_have_processes(node: &NodeSnapshot) {
4161        if node.state == NodeState::Running {
4162            assert!(
4163                node.process_id.is_some() && node.process_name.is_some(),
4164                "node '{}' reports as running but names no process",
4165                node.name
4166            );
4167            assert!(
4168                node.uptime_ms.is_some(),
4169                "node '{}' is running but has no uptime",
4170                node.name
4171            );
4172        } else {
4173            assert!(
4174                node.uptime_ms.is_none(),
4175                "node '{}' is not running but reports an uptime",
4176                node.name
4177            );
4178        }
4179
4180        for child in &node.children {
4181            assert_running_nodes_have_processes(child);
4182        }
4183    }
4184
4185    #[tokio::test]
4186    async fn snapshot_after_the_run_reports_a_stopped_tree() {
4187        fn assert_send_sync<T: Send + Sync + 'static>() {}
4188        assert_send_sync::<SupervisionTreeHandle>();
4189
4190        let mut sup = Supervisor::new("test-sup").unwrap();
4191        sup.add_worker(MockWorker::long_running("worker1"));
4192
4193        let tree = sup.tree_handle();
4194        let (tx, handle) = run_supervisor_with_trigger(sup).await;
4195        wait_until("the child is running", || tree.snapshot().root.children.len() == 1).await;
4196
4197        tx.send(()).unwrap();
4198        join_supervisor(handle).await.unwrap();
4199
4200        // A stopped supervisor reports as stopped rather than presenting a subtree of processes that no longer exist.
4201        let snapshot = tree.snapshot();
4202        assert_eq!(snapshot.root.state, NodeState::Registered);
4203        assert!(snapshot.root.children.is_empty());
4204        assert_eq!(
4205            snapshot.root.supervision.expect("supervisor settings").generation,
4206            1,
4207            "the generation count outlives the run"
4208        );
4209
4210        // Not running is not the same as never having run: the node keeps the identity of the run that just ended, so
4211        // that a stopped tree still says what it was. `state` is what distinguishes the two.
4212        assert!(snapshot.root.process_id.is_some(), "a stopped node says what it ran as");
4213        assert_eq!(snapshot.root.process_name.as_deref(), Some("test_sup"));
4214        assert!(snapshot.root.started_at.is_some());
4215        assert_eq!(snapshot.root.uptime_ms, None, "a node that isn't running has no uptime");
4216    }
4217
4218    #[tokio::test]
4219    async fn an_exited_nested_supervisor_says_what_it_ran_as() {
4220        // A nested supervisor that stops for good is kept by its parent as a tombstone, and what it ran as is what a
4221        // postmortem wants from it -- which is what its own workers have always reported in the same situation.
4222        let mut child_sup = Supervisor::new("child-sup")
4223            .unwrap()
4224            .with_auto_shutdown(AutoShutdown::AnySignificant);
4225        child_sup.add_worker(
4226            runtime::supervisable(MockWorker::completing("one-shot", Duration::from_millis(10)))
4227                .temporary()
4228                .with_significant(true)
4229                .build(),
4230        );
4231
4232        // Temporary, so the parent does not bring it back once it stops.
4233        let mut parent_sup = Supervisor::new("parent-sup").unwrap();
4234        parent_sup.add_worker(ChildSpecification::from(child_sup).with_restart_type(RestartType::Temporary));
4235
4236        let tree = parent_sup.tree_handle();
4237        let (tx, handle) = run_supervisor_with_trigger(parent_sup).await;
4238
4239        // Record the identity while it is still running, so the tombstone has to report the same one rather than
4240        // merely reporting something.
4241        let mut running = None;
4242        wait_until("the nested supervisor is running", || {
4243            let snapshot = tree.snapshot();
4244            let nested = find_node(&snapshot.root.children, "child-sup");
4245            if nested.state == NodeState::Running {
4246                running = Some((nested.process_id, nested.process_name.clone(), nested.started_at));
4247                return true;
4248            }
4249            false
4250        })
4251        .await;
4252        let (process_id, process_name, started_at) = running.expect("observed running");
4253        assert!(process_id.is_some(), "a running supervisor has a process");
4254
4255        wait_until("the nested supervisor has exited", || {
4256            find_node(&tree.snapshot().root.children, "child-sup").state == NodeState::Exited
4257        })
4258        .await;
4259
4260        let snapshot = tree.snapshot();
4261        let nested = find_node(&snapshot.root.children, "child-sup");
4262        assert_eq!(nested.process_id, process_id, "an exited supervisor keeps its identity");
4263        assert_eq!(nested.process_name, process_name);
4264        assert_eq!(nested.started_at, started_at);
4265        assert!(nested.exited_at.is_some(), "a tombstone records when it exited");
4266        assert_eq!(nested.uptime_ms, None, "a node that isn't running has no uptime");
4267        assert!(
4268            nested.resource_group.is_some(),
4269            "and the group its allocations were attributed to"
4270        );
4271
4272        tx.send(()).unwrap();
4273        let _ = join_supervisor(handle).await;
4274    }
4275
4276    #[tokio::test]
4277    async fn initialization_failure_leaves_nothing_running() {
4278        let mut sup = Supervisor::new("test-sup").unwrap();
4279        sup.add_worker(MockWorker::init_failure("broken"));
4280
4281        let tree = sup.tree_handle();
4282
4283        // Run directly rather than through the readiness barrier: this supervisor fails almost immediately, so it may
4284        // never be observed running at all.
4285        let mut sup = sup;
4286        let (_tx, rx) = oneshot::channel::<()>();
4287        let result = timeout(Duration::from_secs(2), sup.run_with_shutdown(rx))
4288            .await
4289            .expect("supervisor should exit promptly");
4290        assert!(matches!(result, Err(SupervisorError::FailedToInitialize { .. })));
4291
4292        // The failure path returns through `run_inner` like every other, so the tree is cleaned up either way.
4293        let snapshot = tree.snapshot();
4294        assert_eq!(snapshot.root.state, NodeState::Registered);
4295        assert_eq!(snapshot.totals.running, 0);
4296    }
4297
4298    #[tokio::test]
4299    async fn resources_are_attributed_to_supervisors_not_workers() {
4300        let mut child_sup = Supervisor::new("child-sup").unwrap();
4301        child_sup.add_worker(MockWorker::long_running("inner-worker"));
4302
4303        let mut parent_sup = Supervisor::new("parent-sup").unwrap();
4304        parent_sup.add_worker(child_sup);
4305
4306        let tree = parent_sup.tree_handle();
4307        let (tx, handle) = run_supervisor_with_trigger(parent_sup).await;
4308        wait_until("the nested subtree is populated", || {
4309            tree.snapshot()
4310                .root
4311                .children
4312                .iter()
4313                .any(|child| !child.children.is_empty())
4314        })
4315        .await;
4316
4317        let snapshot = tree.snapshot();
4318
4319        // The tracking allocator is a process-wide facility that the test binary doesn't install, so every byte count
4320        // here reads zero. That is exactly why the snapshot reports whether tracking is on at all: without it, zero
4321        // bytes is indistinguishable from nothing being measured.
4322        assert!(!snapshot.resource_tracking_enabled);
4323
4324        let nested = find_node(&snapshot.root.children, "child-sup");
4325        assert!(
4326            nested.resources.is_some(),
4327            "a supervisor owns the resource group named by its own process"
4328        );
4329        assert_eq!(nested.resource_group.as_deref(), nested.process_name.as_deref());
4330
4331        // A worker inherits its supervisor's group rather than owning one, so it names the group but carries no
4332        // figures of its own -- reporting the supervisor's totals against each of its workers would count them twice.
4333        let worker = &nested.children[0];
4334        assert!(worker.resources.is_none(), "a worker owns no resource group");
4335        assert_eq!(worker.resource_group.as_deref(), nested.process_name.as_deref());
4336
4337        tx.send(()).unwrap();
4338        let _ = join_supervisor(handle).await;
4339    }
4340
4341    #[tokio::test]
4342    async fn snapshot_serializes_to_the_expected_shape() {
4343        let mut child_sup = Supervisor::new("child-sup").unwrap();
4344        child_sup.add_worker(MockWorker::long_running("inner-worker"));
4345
4346        let mut parent_sup = Supervisor::new("parent-sup").unwrap();
4347        parent_sup.add_worker(child_sup);
4348
4349        let tree = parent_sup.tree_handle();
4350        let (tx, handle) = run_supervisor_with_trigger(parent_sup).await;
4351        wait_until("the nested subtree is populated", || {
4352            tree.snapshot()
4353                .root
4354                .children
4355                .iter()
4356                .any(|child| !child.children.is_empty())
4357        })
4358        .await;
4359
4360        let value = serde_json::to_value(tree.snapshot()).expect("snapshot should serialize");
4361
4362        // This is the wire contract an HTTP consumer reads, so pin its shape rather than just its serializability.
4363        assert!(value["captured_at"].is_u64(), "a timestamp is a plain integer");
4364        assert_eq!(value["root"]["kind"], "supervisor");
4365        assert_eq!(value["root"]["state"], "running");
4366        assert_eq!(value["root"]["restart"], "permanent");
4367        assert_eq!(value["root"]["supervision"]["restart_mode"], "one_for_one");
4368        assert_eq!(value["root"]["supervision"]["auto_shutdown"], "never");
4369
4370        let nested = &value["root"]["children"][0];
4371        assert_eq!(nested["kind"], "supervisor");
4372        let worker = &nested["children"][0];
4373        assert_eq!(worker["kind"], "worker");
4374        assert!(
4375            worker["children"]
4376                .as_array()
4377                .expect("children is always an array")
4378                .is_empty(),
4379            "children is present and empty for a worker, so consumers can recurse unconditionally"
4380        );
4381        assert!(worker.get("resources").is_none(), "a worker reports no resource usage");
4382        assert!(
4383            worker.get("supervision").is_none(),
4384            "a worker reports no supervision settings"
4385        );
4386        assert!(
4387            worker.get("children_truncated").is_none(),
4388            "an untruncated node omits the truncation flag"
4389        );
4390
4391        tx.send(()).unwrap();
4392        let _ = join_supervisor(handle).await;
4393    }
4394
4395    #[tokio::test]
4396    async fn a_worker_that_drives_its_own_supervisor_is_shown_as_its_parent() {
4397        // Some workers own a supervisor rather than being one: they build it after initialization and run it inside
4398        // their own future, so their parent has no supervisor value to record and the subtree would otherwise be
4399        // invisible. This is how the largest subtrees in a real process are shaped.
4400        struct SupervisorDrivingWorker;
4401
4402        #[async_trait]
4403        impl Supervisable for SupervisorDrivingWorker {
4404            fn name(&self) -> &str {
4405                "driver"
4406            }
4407
4408            fn shutdown_strategy(&self) -> ShutdownStrategy {
4409                ShutdownStrategy::Graceful(Duration::MAX)
4410            }
4411
4412            async fn initialize(
4413                &self, process_shutdown: ShutdownHandle,
4414            ) -> Result<SupervisorFuture, InitializationError> {
4415                Ok(Box::pin(async move {
4416                    let mut inner = Supervisor::new("inner-sup").expect("valid name");
4417                    inner.add_worker(MockWorker::long_running("inner-worker"));
4418                    inner
4419                        .run_with_shutdown_inner(process_shutdown, HaltWatch::never(), None)
4420                        .await
4421                        .map_err(Into::into)
4422                }))
4423            }
4424        }
4425
4426        let mut sup = Supervisor::new("test-sup").unwrap();
4427        sup.add_worker(SupervisorDrivingWorker);
4428
4429        let tree = sup.tree_handle();
4430        let (tx, handle) = run_supervisor_with_trigger(sup).await;
4431
4432        wait_until("the worker's own supervisor has attached itself", || {
4433            let snapshot = tree.snapshot();
4434            snapshot
4435                .root
4436                .children
4437                .first()
4438                .is_some_and(|child| !child.children.is_empty())
4439        })
4440        .await;
4441
4442        let snapshot = tree.snapshot();
4443        let driver = find_node(&snapshot.root.children, "driver");
4444        assert_eq!(
4445            driver.kind,
4446            NodeKind::Worker,
4447            "the worker is what its parent supervises"
4448        );
4449
4450        let inner_sup = find_node(&driver.children, "inner-sup");
4451        assert_eq!(inner_sup.kind, NodeKind::Supervisor);
4452        assert_eq!(inner_sup.children.len(), 1);
4453        assert_eq!(inner_sup.children[0].name, "inner-worker");
4454        assert_eq!(snapshot.totals.max_depth, 4);
4455
4456        tx.send(()).unwrap();
4457        let _ = join_supervisor(handle).await;
4458    }
4459
4460    #[tokio::test]
4461    async fn mixed_child_traffic_keeps_the_roster_consistent() {
4462        // Exercises every path that mutates the roster in one run -- static spawn, dynamic spawn, restart in place,
4463        // and removal without restart. It asserts little itself: the drift check inside the roster is what is under
4464        // test, and it runs on every mutation.
4465        let mut sup = Supervisor::new("test-sup").unwrap().with_restart_strategy(
4466            RestartStrategy::one_to_one().with_intensity_and_period(100, Duration::from_secs(5)),
4467        );
4468        sup.add_worker(MockWorker::long_running("stable"));
4469        sup.add_worker(MockWorker::failing("flapper", Duration::from_millis(5)));
4470
4471        let tree = sup.tree_handle();
4472        let sup_handle = sup.handle();
4473        let (tx, handle) = run_supervisor_with_trigger(sup).await;
4474
4475        for _ in 0..25 {
4476            sup_handle.spawn(MockWorker::completing("ephemeral", Duration::from_millis(2)));
4477            sup_handle.spawn(MockWorker::long_running("lingering"));
4478            let snapshot = tree.snapshot();
4479            assert_running_nodes_have_processes(&snapshot.root);
4480            tokio::time::sleep(Duration::from_millis(2)).await;
4481        }
4482
4483        wait_until("the flapping child has restarted several times", || {
4484            find_node(&tree.snapshot().root.children, "flapper").restart_count >= 3
4485        })
4486        .await;
4487
4488        tx.send(()).unwrap();
4489        let _ = join_supervisor(handle).await;
4490    }
4491}