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
32const UNNAMED_CHILD: &str = "unnamed";
36
37pub(super) type WorkerFuture = Pin<Box<dyn Future<Output = Result<(), WorkerError>> + Send>>;
43
44#[derive(Debug)]
49pub(super) enum WorkerError {
50 Initialization {
56 child_name: Option<String>,
57 source: InitializationError,
58 },
59
60 Runtime(GenericError),
62
63 ShutdownTimedOut {
70 aborted: usize,
72 },
73}
74
75impl From<SupervisorError> for WorkerError {
76 fn from(err: SupervisorError) -> Self {
77 match err {
78 SupervisorError::FailedToInitialize { child_name, source } => WorkerError::Initialization {
81 child_name: Some(child_name),
82 source,
83 },
84 SupervisorError::ShutdownTimedOut { aborted } => WorkerError::ShutdownTimedOut { aborted },
86 other => WorkerError::Runtime(other.into()),
88 }
89 }
90}
91
92#[derive(Debug, Snafu)]
94pub enum ProcessError {
95 #[snafu(display("Child process was aborted by the supervisor."))]
97 Aborted,
98
99 #[snafu(display("Child process panicked."))]
101 Panicked,
102
103 #[snafu(display("Child process terminated with an error: {}", source))]
105 Terminated {
106 source: GenericError,
108 },
109}
110
111#[derive(Clone, Copy, Debug, PartialEq, Eq, Default, Deserialize, Serialize)]
118#[serde(rename_all = "snake_case")]
119pub enum AutoShutdown {
120 #[default]
122 Never,
123
124 AnySignificant,
126
127 AllSignificant,
129}
130
131#[derive(Debug, Snafu)]
133#[snafu(context(suffix(false)))]
134pub enum SupervisorError {
135 #[snafu(display("Invalid name for supervisor or worker: '{}'", name))]
137 InvalidName {
138 name: String,
140 },
141
142 #[snafu(display("Child process '{}' failed to initialize: {}", child_name, source))]
147 FailedToInitialize {
148 child_name: String,
150
151 source: InitializationError,
153 },
154
155 #[snafu(display("Supervisor has exceeded restart limits and was forced to shutdown."))]
157 Shutdown,
158
159 #[snafu(display("Supervisor shut down after a significant child terminated."))]
165 SignificantChildExited,
166
167 #[snafu(display(
176 "Shutdown completed uncleanly: {} worker(s) were forcefully aborted after exceeding their shutdown timeout.",
177 aborted
178 ))]
179 ShutdownTimedOut {
180 aborted: usize,
182 },
183}
184
185pub struct ChildSpecification<S = WorkerSpec> {
206 spec_inner: S,
207}
208
209pub struct WorkerSpec {
211 worker: Arc<dyn Supervisable>,
212 options: ChildOptions,
213}
214
215pub struct SupervisorSpec {
217 supervisor: Supervisor,
218 options: ChildOptions,
219}
220
221impl ChildSpecification<WorkerSpec> {
227 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 #[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 #[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 #[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 #[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 #[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 #[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 #[must_use]
311 pub(crate) fn forwarding_failures(mut self) -> Self {
312 self.spec_inner.options.forwards_failures = true;
313 self
314 }
315
316 #[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 #[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
336impl ChildSpecification<SupervisorSpec> {
345 #[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 #[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 #[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 #[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
407pub trait ChildState: sealed::Sealed + Sized {
414 #[doc(hidden)]
419 fn into_child_parts(spec: ChildSpecification<Self>, registration: Registration) -> LoweredChild;
420}
421
422#[doc(hidden)]
424#[derive(Clone, Copy, Debug, PartialEq, Eq)]
425pub enum Registration {
426 Declared,
428
429 Spawned,
431}
432
433pub struct LoweredChild {
438 spec: SupervisedChild,
439 config: ChildConfig,
440}
441
442impl LoweredChild {
443 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 config: options.resolve(registration, true),
467 }
468 }
469}
470
471pub(super) enum SupervisedChild {
473 Worker(Arc<dyn Supervisable>),
474 Supervisor(Supervisor),
475}
476
477impl SupervisedChild {
478 pub(super) fn is_supervisor(&self) -> bool {
480 matches!(self, Self::Supervisor(_))
481 }
482
483 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 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 Self::Supervisor(_) => ShutdownStrategy::Graceful(Duration::MAX),
525 }
526 }
527
528 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 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 Ok(sup.as_nested_process(process, process_shutdown, halt))
584 }
585 RuntimeMode::Dedicated(config) => {
586 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#[derive(Clone, Copy, Debug, Default)]
627pub(super) enum ChildShutdown {
628 #[default]
630 Worker,
631
632 Explicit(ShutdownStrategy),
634}
635
636impl ChildShutdown {
637 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#[derive(Clone, Debug, Default)]
654pub(super) struct ChildOptions {
655 restart: Option<RestartType>,
657 significant: bool,
658
659 runtime: Option<Handle>,
661
662 shutdown: ChildShutdown,
663
664 edge: Option<Edge>,
666
667 forwards_failures: bool,
669
670 adopt: Option<Adoption>,
672
673 own_resource_group: bool,
675
676 stops_with_owner: bool,
678
679 #[cfg(any(test, feature = "test-util"))]
681 exit_observer: Option<ExitObserver>,
682}
683
684impl ChildOptions {
685 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#[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 pub(super) fn restart(&self) -> RestartType {
765 self.restart
766 }
767
768 pub(super) fn significant(&self) -> bool {
770 self.significant
771 }
772
773 pub(super) fn edge(&self) -> Edge {
775 self.edge
776 }
777
778 pub(super) fn forwards_failures(&self) -> bool {
780 self.forwards_failures
781 }
782
783 pub(super) fn adopt(&self) -> Option<&Adoption> {
785 self.adopt.as_ref()
786 }
787
788 pub(super) fn own_resource_group(&self) -> bool {
790 self.own_resource_group
791 }
792
793 pub(super) fn stops_with_owner(&self) -> bool {
796 self.stops_with_owner
797 }
798
799 pub(super) fn can_stop_owner(&self) -> bool {
805 self.restart != RestartType::Permanent && (self.significant || self.edge == Edge::Needed)
806 }
807
808 pub(super) fn runtime(&self) -> Option<&Handle> {
810 self.runtime.as_ref()
811 }
812
813 pub(super) fn shutdown(&self) -> ChildShutdown {
815 self.shutdown
816 }
817
818 #[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#[cfg(any(test, feature = "test-util"))]
835#[derive(Clone)]
836pub(crate) struct ExitObserver(Arc<Mutex<Option<ExitSender>>>);
837
838#[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 pub(crate) fn new(tx: ExitSender) -> Self {
846 Self(Arc::new(Mutex::new(Some(tx))))
847 }
848
849 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#[derive(Clone)]
877struct ChildEntry {
878 spec: SupervisedChild,
879 config: ChildConfig,
880 dynamic: bool,
883}
884
885#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
890pub struct ChildId(u64);
891
892impl ChildId {
893 pub(super) const fn from_raw(id: u64) -> Self {
895 Self(id)
896 }
897
898 pub const fn as_u64(self) -> u64 {
900 self.0
901 }
902}
903
904struct PendingSpawn {
906 id: u64,
907 spec: SupervisedChild,
908 config: ChildConfig,
909}
910
911const SPAWN_DRAIN_BATCH: usize = 64;
917
918#[derive(Clone)]
934pub struct SupervisorHandle {
935 name: Arc<str>,
936 current_tx: Arc<Mutex<Option<mpsc::UnboundedSender<PendingSpawn>>>>,
939 id_counter: Arc<AtomicU64>,
940 active: Arc<AtomicUsize>,
941}
942
943impl SupervisorHandle {
944 pub fn name(&self) -> &str {
946 &self.name
947 }
948
949 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 let id = self.id_counter.fetch_add(1, Ordering::Relaxed);
976 let pending = PendingSpawn { id, spec, config };
977
978 let tx = self.current_tx.lock().unwrap().clone();
985 match tx {
986 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 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 pub fn is_running(&self) -> bool {
1011 self.current_tx.lock().unwrap().is_some()
1012 }
1013
1014 pub fn active_children(&self) -> usize {
1019 self.active.load(Ordering::Relaxed)
1020 }
1021}
1022
1023pub struct Supervisor {
1051 supervisor_id: Arc<str>,
1052 child_specs: Vec<ChildEntry>,
1053 runtime_mode: RuntimeMode,
1054 current_tx: Arc<Mutex<Option<mpsc::UnboundedSender<PendingSpawn>>>>,
1058 id_counter: Arc<AtomicU64>,
1059 active: Arc<AtomicUsize>,
1061 node: Arc<SupervisorNode>,
1064}
1065
1066impl Supervisor {
1067 pub fn new<S: AsRef<str>>(supervisor_id: S) -> Result<Self, SupervisorError> {
1069 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 pub fn id(&self) -> &str {
1093 &self.supervisor_id
1094 }
1095
1096 pub fn with_restart_strategy(self, strategy: RestartStrategy) -> Self {
1098 self.node.update_config(|config| config.restart_strategy = strategy);
1099 self
1100 }
1101
1102 #[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 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 pub fn tree_handle(&self) -> SupervisionTreeHandle {
1134 SupervisionTreeHandle::new(Arc::clone(&self.node))
1135 }
1136
1137 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 pub(crate) fn runtime_mode(&self) -> &RuntimeMode {
1156 &self.runtime_mode
1157 }
1158
1159 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 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 self.child_specs.push(entry);
1231 }
1232
1233 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 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 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 let facts = Self::child_facts(entry, ChildKey::Static(index));
1283 roster.insert(id, entry.clone(), facts, started);
1284 }
1285
1286 Ok(())
1287 }
1288
1289 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 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 let (cmd_tx, cmd_rx) = mpsc::unbounded_channel();
1331 *self.current_tx.lock().unwrap() = Some(cmd_tx);
1332
1333 self.node.begin_run(&process);
1337
1338 let result = self.supervise(process, process_shutdown, halt, cmd_rx).await;
1339
1340 *self.current_tx.lock().unwrap() = None;
1343 self.active.store(0, Ordering::Relaxed);
1344
1345 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 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 let mut roster = Roster::new(Arc::clone(&self.node));
1374
1375 self.spawn_static_children(&mut roster, &mut worker_state, auto_shutdown)?;
1378
1379 let mut significant_remaining = roster.values().filter(|entry| entry.config.significant).count();
1381
1382 let mut spawn_batch = Vec::with_capacity(SPAWN_DRAIN_BATCH);
1384
1385 pin!(process_shutdown);
1387
1388 let outcome = loop {
1389 select! {
1390 biased;
1393
1394 _ = halt.wait() => break Ok(()),
1396
1397 _ = &mut process_shutdown => break Ok(()),
1401
1402 (child_id, worker_result) = worker_state.wait_for_next_worker() => {
1406 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 if let Err(WorkerError::Initialization { child_name: inner, source }) = worker_result {
1414 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 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 WorkerError::ShutdownTimedOut { aborted } => ProcessError::Terminated {
1436 source: SupervisorError::ShutdownTimedOut { aborted }.into(),
1437 },
1438 });
1439
1440 if !config.restart.should_restart(abnormal) {
1441 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 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 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 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 let _ = worker_state.shutdown_workers(&mut halt).await;
1507 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 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 _ = 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 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 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 debug!(supervisor_id = %self.supervisor_id, "Nested supervisor starting.");
1582
1583 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 pub async fn run(&mut self) -> Result<(), SupervisorError> {
1599 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 pub async fn run_with_shutdown<F: Future + Send + 'static>(&mut self, shutdown: F) -> Result<(), SupervisorError> {
1621 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 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 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#[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 #[derive(Clone)]
1708 enum InitBehavior {
1709 Instant,
1711
1712 Slow(Duration),
1714
1715 Fail(&'static str),
1717 }
1718
1719 #[derive(Clone)]
1721 enum RunBehavior {
1722 UntilShutdown,
1724
1725 FailAfter(Duration, &'static str),
1727
1728 CompleteAfter(Duration),
1730
1731 SlowShutdown(Duration),
1733
1734 IgnoreShutdown,
1736
1737 PanicAfter(Duration),
1739 }
1740
1741 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 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 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 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 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 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 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 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 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 fn start_count(&self) -> Arc<AtomicUsize> {
1872 Arc::clone(&self.start_count)
1873 }
1874
1875 fn finish_count(&self) -> Arc<AtomicUsize> {
1883 Arc::clone(&self.finish_count)
1884 }
1885
1886 fn with_brutal_shutdown(mut self) -> Self {
1888 self.brutal_shutdown = true;
1889 self
1890 }
1891
1892 fn with_graceful_timeout(mut self, timeout: Duration) -> Self {
1894 self.graceful_timeout = timeout;
1895 self
1896 }
1897
1898 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 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 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 finish_count.fetch_add(1, Ordering::SeqCst);
1976 Ok(())
1977 }
1978 RunBehavior::IgnoreShutdown => {
1979 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 async fn run_supervisor_with_trigger(
2000 supervisor: Supervisor,
2001 ) -> (oneshot::Sender<()>, JoinHandle<Result<(), SupervisorError>>) {
2002 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 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 #[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 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 #[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 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 assert!(
2099 failing_count.load(Ordering::SeqCst) >= 2,
2100 "failing worker should have been restarted"
2101 );
2102 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("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 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 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 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 let transient = MockWorker::completing("transient-worker", Duration::from_millis(30));
2190 let transient_count = transient.start_count();
2191
2192 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 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 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 #[tokio::test]
2273 async fn temporary_child_is_not_restarted() {
2274 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_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_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 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 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 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 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 sup.add_worker(MockWorker::long_running("stable-worker"));
2454
2455 let (tx, handle) = run_supervisor_with_trigger(sup).await;
2456 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 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 sup.add_worker(MockWorker::long_running("stable-worker"));
2502
2503 let (tx, handle) = run_supervisor_with_trigger(sup).await;
2504 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 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_until("both temporary children have completed", || {
2548 a_finished.load(Ordering::SeqCst) == 1 && b_finished.load(Ordering::SeqCst) == 1
2549 })
2550 .await;
2551
2552 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 #[tokio::test]
2574 async fn significant_child_drives_auto_shutdown() {
2575 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 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 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 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 wait_until("the non-significant child has completed", || {
2640 plain_finished.load(Ordering::SeqCst) == 1
2641 })
2642 .await;
2643
2644 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 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 assert!(
2691 elapsed >= Duration::from_millis(200),
2692 "auto-shutdown must wait for all significant children (took {elapsed:?})"
2693 );
2694 }
2695
2696 #[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 assert_eq!(start_count.load(Ordering::SeqCst), 0);
2735 }
2736
2737 #[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 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 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 sleep(Duration::from_millis(20)).await;
2764 tx.send(()).unwrap();
2765
2766 let result = timeout(Duration::from_secs(2), handle).await;
2769 assert!(result.is_ok(), "shutdown during slow init should complete promptly");
2770 }
2771
2772 #[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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 let start = std::time::Instant::now();
3251 tx.send(()).unwrap();
3252 let result = join_supervisor(run).await;
3253 let elapsed = start.elapsed();
3254
3255 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 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 handle.spawn(MockWorker::long_running("responsive").with_graceful_timeout(Duration::MAX));
3280 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 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 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 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 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 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 #[tokio::test]
3376 async fn restart_intensity_zero_shuts_down_on_first_failure() {
3377 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 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 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 wait_until("the static worker has been restarted", || {
3427 failing_count.load(Ordering::SeqCst) >= 2
3428 })
3429 .await;
3430
3431 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 #[tokio::test]
3448 async fn dedicated_single_threaded_runtime_runs_nested_worker_and_shuts_down_cleanly() {
3449 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 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 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 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 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 #[tokio::test]
3543 async fn child_with_runtime_override_runs_on_that_runtime() {
3544 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 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 child_runtime.shutdown_background();
3586 }
3587
3588 #[tokio::test]
3589 async fn child_shutdown_strategy_override_takes_precedence_over_worker() {
3590 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 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 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 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 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 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 let recorder = TestRecorder::default();
3738 let _guard = metrics::set_default_local_recorder(&recorder);
3739
3740 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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}