1use crate::child_runner::run_exit::TaskExit;
7use crate::child_runner::runner::{ChildRunHandle, ChildRunReport, ChildRunner, wait_for_report};
8use crate::control::command::{CommandMeta, CommandResult, ControlCommand, CurrentState};
9use crate::control::outcome::{
10 ChildAttemptStatus, ChildControlFailure, ChildControlFailurePhase, ChildControlOperation,
11 ChildControlResult, ChildLivenessState, ChildStopState, GenerationFenceDecision,
12 GenerationFenceOutcome, GenerationFencePhase, PendingRestart, RestartLimitState,
13 StaleAttemptReport, StaleReportHandling,
14};
15use crate::error::types::SupervisorError;
16use crate::event::payload::{ProtectionAction, SupervisorEvent, ThrottleGateOwner, What, Where};
17use crate::event::time::{CorrelationId, EventSequenceSource, EventTime, When};
18use crate::id::types::{ChildId, ChildStartCount, Generation, SupervisorPath};
19use crate::observe::fairness::FairnessProbe;
20use crate::observe::pipeline::{ObservabilityPipeline, PipelineStageDiagnostic};
21use crate::policy::backoff::BackoffPolicy;
22use crate::policy::decision::{
23 PolicyEngine, RestartDecision, RestartPolicy, TaskExit as PolicyTaskExit,
24};
25use crate::policy::failure_window::FailureWindow;
26use crate::policy::meltdown::MeltdownTracker;
27use crate::policy::task_role_defaults::{EffectivePolicy, OnSuccessAction};
28use crate::readiness::signal::ReadinessState;
29use crate::registry::entry::{ChildRuntime, ChildRuntimeStatus};
30use crate::registry::store::RegistryStore;
31use crate::runtime::admission::{AdmissionConflict, AdmissionSet};
32use crate::runtime::child_slot::{
33 ChildExitSummary, ChildSlot, DEFAULT_HEARTBEAT_TIMEOUT_SECS, RuntimeTimeBase,
34};
35use crate::runtime::concurrent_gate::GatePermit;
36use crate::runtime::lifecycle::RuntimeExitReport;
37use crate::runtime::message::{ChildStartMessage, ControlPlaneMessage, RuntimeLoopMessage};
38use crate::runtime::pipeline::{ExitClassification, PipelineContext, SupervisionPipeline};
39use crate::runtime::shutdown::{reconcile_shutdown_slots, shutdown_tree_fanout};
40use crate::runtime::shutdown_pipeline::ShutdownPipeline;
41use crate::shutdown::coordinator::{ShutdownCoordinator, ShutdownResult};
42use crate::shutdown::report::{
43 ChildShutdownOutcome, ChildShutdownOutcomeInput, ChildShutdownStatus, ShutdownPipelineReport,
44 ShutdownReconcileReport,
45};
46use crate::shutdown::stage::{ShutdownCause, ShutdownPhase};
47use crate::spec::child::{ChildSpec, RestartPolicy as ChildRestartPolicy};
48use crate::spec::child_declaration::{ChildDeclaration, validate_child_declaration};
49use crate::spec::shutdown::ShutdownBudget;
50use crate::spec::supervisor::{RestartLimit, SupervisorSpec};
51use crate::task::factory_registry::TaskFactoryRegistry;
52use crate::tree::builder::SupervisorTree;
53use crate::tree::order::{restart_execution_plan, shutdown_order, startup_order};
54use std::collections::HashMap;
55use std::sync::{Arc, Mutex};
56use std::time::{Duration, SystemTime, UNIX_EPOCH};
57use tokio::sync::{broadcast, mpsc};
58use tokio::time::{Instant, timeout};
59
60#[derive(Debug)]
62struct PendingRuntimeEvent {
63 child_id: ChildId,
65 path: SupervisorPath,
67 generation: Option<Generation>,
69 attempt: Option<ChildStartCount>,
71 correlation_id: CorrelationId,
73 what: What,
75}
76
77#[derive(Debug)]
79pub struct RuntimeControlState {
80 shutdown: ShutdownCoordinator,
82 shutdown_pipeline: ShutdownPipeline,
84 slots: HashMap<ChildId, ChildSlot>,
86 #[allow(dead_code)]
88 admission_set: AdmissionSet,
89 time_base: RuntimeTimeBase,
91 event_sequences: EventSequenceSource,
93 observability: Arc<Mutex<ObservabilityPipeline>>,
95 supervision_pipeline: SupervisionPipeline,
97 concurrent_gate: crate::runtime::concurrent_gate::SupervisorInstanceGate,
99 fairness_probe: FairnessProbe,
101 manifests: Vec<String>,
103 task_factory_registry: TaskFactoryRegistry,
105 registry: RegistryStore,
107 tree: SupervisorTree,
109 spec: SupervisorSpec,
111 policy_engine: PolicyEngine,
113 command_sender: mpsc::Sender<RuntimeLoopMessage>,
115 pub orphan_count: u64,
118 exit_handler: Arc<dyn crate::exit_handler::ExitHandler>,
120 active_gate_permits: HashMap<(ChildId, Generation, ChildStartCount), GatePermit>,
127}
128
129fn build_initial_slots(
131 registry: &RegistryStore,
132 spec: &SupervisorSpec,
133) -> HashMap<ChildId, ChildSlot> {
134 registry
135 .declaration_order()
136 .iter()
137 .filter_map(|child_id| {
138 registry.child(child_id).map(|runtime| {
139 let mut slot = ChildSlot::new_placeholder(runtime.id.clone(), runtime.path.clone());
140 if let Some(child_spec) = spec.children.iter().find(|c| &c.id == child_id) {
142 slot.stale_after = child_spec.health_policy.stale_after;
143 slot.group = child_spec.group.clone();
144 }
145 (child_id.clone(), slot)
146 })
147 })
148 .collect::<HashMap<_, _>>()
149}
150
151#[allow(dead_code)]
152impl RuntimeControlState {
153 pub fn new(
164 spec: SupervisorSpec,
165 command_sender: mpsc::Sender<RuntimeLoopMessage>,
166 observability: Arc<Mutex<ObservabilityPipeline>>,
167 ) -> Result<Self, SupervisorError> {
168 Self::new_with_factory_registry(
169 spec,
170 command_sender,
171 observability,
172 TaskFactoryRegistry::new(),
173 )
174 }
175
176 pub fn new_with_factory_registry(
189 spec: SupervisorSpec,
190 command_sender: mpsc::Sender<RuntimeLoopMessage>,
191 observability: Arc<Mutex<ObservabilityPipeline>>,
192 task_factory_registry: TaskFactoryRegistry,
193 ) -> Result<Self, SupervisorError> {
194 let tree_shutdown = spec.tree_shutdown;
195 let tree = SupervisorTree::build(&spec)?;
196 let mut registry = RegistryStore::new();
197 registry.register_tree(&tree)?;
198 let time_base = RuntimeTimeBase::new();
199 let slots = build_initial_slots(®istry, &spec);
200
201 let meltdown_tracker = MeltdownTracker::new(spec.meltdown_policy);
202 let failure_window = FailureWindow::new(spec.failure_window_config);
203 let supervision_pipeline = SupervisionPipeline::with_backpressure_config(
204 spec.pipeline_journal_capacity,
205 spec.pipeline_subscriber_capacity,
206 meltdown_tracker,
207 failure_window,
208 spec.restart_budget_config.clone(),
209 spec.group_dependencies.clone(),
210 spec.backpressure_config.clone(),
211 );
212
213 let concurrent_gate = crate::runtime::concurrent_gate::SupervisorInstanceGate::new(
214 spec.concurrent_restart_limit,
215 );
216
217 let now_unix_nanos = SystemTime::now()
219 .duration_since(UNIX_EPOCH)
220 .unwrap_or_default()
221 .as_nanos();
222 let fairness_probe = FairnessProbe::new(now_unix_nanos);
223
224 Ok(Self {
225 shutdown: ShutdownCoordinator::new(tree_shutdown),
226 shutdown_pipeline: ShutdownPipeline::new(),
227 slots,
228 admission_set: AdmissionSet::new(),
229 time_base,
230 event_sequences: EventSequenceSource::new(),
231 observability,
232 supervision_pipeline,
233 concurrent_gate,
234 fairness_probe,
235 manifests: Vec::new(),
236 task_factory_registry,
237 registry,
238 tree,
239 spec,
240 policy_engine: PolicyEngine::new(),
241 command_sender,
242 orphan_count: 0,
243 exit_handler: Arc::new(crate::exit_handler::DefaultExitHandler),
244 active_gate_permits: HashMap::new(),
245 })
246 }
247
248 fn child_shutdown_budget(&self, child_id: &ChildId) -> ShutdownBudget {
258 self.spec.shutdown_budget_for(child_id)
259 }
260
261 pub async fn start_declared_children(&mut self) {
280 let child_ids = startup_order(&self.tree)
281 .into_iter()
282 .map(|node| node.child.id.clone())
283 .collect::<Vec<_>>();
284 for child_id in &child_ids {
285 if let Some(deps) = self
289 .registry
290 .child(child_id)
291 .map(|r| r.spec.dependencies.clone())
292 .filter(|d| !d.is_empty())
293 {
294 self.wait_for_dependencies(&deps, child_id).await;
295 }
296 self.spawn_child_start(child_id.clone(), false, Duration::ZERO);
297 }
298 }
299
300 async fn wait_for_dependencies(&mut self, deps: &[ChildId], child_id: &ChildId) {
316 for dep_id in deps {
317 let mut receiver = self
318 .slots
319 .get(dep_id)
320 .and_then(|slot| slot.readiness_receiver.clone());
321
322 let Some(ref mut rx) = receiver else {
323 continue;
325 };
326
327 if *rx.borrow_and_update() == ReadinessState::Ready {
329 continue;
330 }
331
332 tracing::info!(
333 child_id = %child_id,
334 dependency = %dep_id,
335 "waiting for dependency readiness",
336 );
337
338 loop {
340 let result = rx.changed().await;
341 if result.is_err() {
342 break;
344 }
345 if *rx.borrow() == ReadinessState::Ready {
346 break;
347 }
348 }
349
350 tracing::info!(
351 child_id = %child_id,
352 dependency = %dep_id,
353 "dependency ready, proceeding",
354 );
355 }
356 }
357
358 pub fn set_exit_handler(&mut self, handler: Arc<dyn crate::exit_handler::ExitHandler>) {
386 self.exit_handler = handler;
387 }
388
389 fn attach_spawned_child_handle(
404 &mut self,
405 child_id: ChildId,
406 path: SupervisorPath,
407 generation: Generation,
408 attempt: ChildStartCount,
409 handle: ChildRunHandle,
410 ) {
411 let mut completion_receiver = handle.completion_receiver.clone();
412 let sender = self.command_sender.clone();
413 self.slots
414 .entry(child_id.clone())
415 .or_insert_with(|| ChildSlot::new_placeholder(child_id.clone(), path))
416 .activate(generation, attempt, ChildAttemptStatus::Running, handle);
417 tokio::spawn(async move {
418 let result = wait_for_report(&mut completion_receiver).await;
419 send_child_result(sender, child_id, result).await;
420 });
421 }
422
423 pub async fn execute_control(
434 &mut self,
435 command: ControlCommand,
436 event_sender: &broadcast::Sender<String>,
437 ) -> Result<CommandResult, SupervisorError> {
438 command.validate_audit_metadata()?;
439 self.reconcile_stop_deadlines();
440 match command {
441 ControlCommand::AddChild {
442 child_manifest,
443 meta: _,
444 target: _,
445 } => {
446 self.ensure_dynamic_child_allowed()?;
447
448 if self.shutdown.phase() != ShutdownPhase::Idle {
450 return Err(SupervisorError::fatal_config(
451 "Cannot add child: supervisor is shutting down",
452 ));
453 }
454
455 let declaration: ChildDeclaration =
457 serde_yaml::from_str(&child_manifest).map_err(|e| {
458 SupervisorError::fatal_config(format!(
459 "Failed to parse child manifest: {e}"
460 ))
461 })?;
462
463 let all_names: std::collections::HashSet<String> =
465 self.spec.children.iter().map(|c| c.name.clone()).collect();
466 validate_child_declaration(&declaration, &all_names).map_err(|e| {
467 SupervisorError::fatal_config(format!(
468 "Child validation failed at {}: {}",
469 e.field_path, e.reason
470 ))
471 })?;
472
473 let child_spec =
475 crate::spec::child::ChildSpec::try_from(declaration).map_err(|e| {
476 SupervisorError::fatal_config(format!("Child conversion failed: {e:?}"))
477 })?;
478 let mut child_spec = child_spec;
479 crate::config::factory_binding::bind_child_factory(
480 &mut child_spec,
481 &self.task_factory_registry,
482 )?;
483
484 let child_id = child_spec.id.clone();
485 let path = crate::id::types::SupervisorPath::root().join(
486 crate::id::types::ChildId::new(&child_spec.name)
487 .value
488 .clone(),
489 );
490
491 let runtime = crate::registry::entry::ChildRuntime::new(child_spec.clone(), path);
493 self.registry.register(runtime.clone()).map_err(|e| {
494 SupervisorError::fatal_config(format!("Registry registration failed: {e}"))
495 })?;
496
497 let slot = crate::runtime::child_slot::ChildSlot::new(
499 child_id.clone(),
500 runtime.path.clone(),
501 std::time::Duration::from_secs(60),
502 );
503 self.slots.insert(child_id.clone(), slot);
504
505 self.spec.children.push(child_spec);
507 self.manifests.push(child_manifest.clone());
508
509 self.spawn_child_start(child_id.clone(), false, Duration::ZERO);
511
512 Ok(CommandResult::ChildAdded { child_manifest })
513 }
514 ControlCommand::RemoveChild { meta, child_id } => Ok(self.execute_stop_child_control(
515 child_id,
516 ChildControlOperation::Removed,
517 "remove_child",
518 &meta,
519 event_sender,
520 )),
521 ControlCommand::RestartChild { meta, child_id } => {
522 Ok(self.execute_restart_child_control(child_id, &meta, event_sender))
523 }
524 ControlCommand::PauseChild { meta, child_id } => Ok(self.execute_stop_child_control(
525 child_id,
526 ChildControlOperation::Paused,
527 "pause_child",
528 &meta,
529 event_sender,
530 )),
531 ControlCommand::ResumeChild { child_id, .. } => {
532 Ok(self.set_child_state(child_id, ChildControlOperation::Active))
533 }
534 ControlCommand::QuarantineChild { meta, child_id } => Ok(self
535 .execute_stop_child_control(
536 child_id,
537 ChildControlOperation::Quarantined,
538 "quarantine_child",
539 &meta,
540 event_sender,
541 )),
542 ControlCommand::ShutdownTree { meta } => {
543 let result = self
544 .execute_shutdown(meta.requested_by, meta.reason, event_sender)
545 .await?;
546 Ok(CommandResult::Shutdown { result })
547 }
548 ControlCommand::CurrentState { .. } => {
549 self.reconcile_stop_deadlines();
550 Ok(CommandResult::CurrentState {
551 state: self.build_current_state(),
552 })
553 }
554 }
555 }
556
557 pub fn handle_child_exit(
568 &mut self,
569 report: ChildRunReport,
570 event_sender: &broadcast::Sender<String>,
571 ) {
572 let child_id = report.runtime.id.clone();
573 let generation = report.runtime.generation;
574 let attempt = report.runtime.child_start_count;
575 let exit_kind = report.exit.clone();
576 let mut pending_events = Vec::new();
577 let was_active = self
578 .slots
579 .get(&child_id)
580 .is_some_and(ChildSlot::has_active_attempt);
581 let matches_pending_fence = self
582 .slots
583 .get(&child_id)
584 .and_then(|state| state.generation_fence.pending_restart.as_ref())
585 .is_some_and(|pending_restart| {
586 pending_restart.old_generation == generation
587 && pending_restart.old_attempt == attempt
588 });
589 let matches_active_attempt = self.slots.get(&child_id).is_some_and(|state| {
590 state.has_active_attempt()
591 && state.generation == Some(generation)
592 && state.attempt == Some(attempt)
593 });
594
595 if matches_active_attempt || matches_pending_fence {
601 self.active_gate_permits
602 .remove(&(child_id.clone(), generation, attempt));
603 }
604 let manual_stop_requested = self
605 .slots
606 .get(&child_id)
607 .is_some_and(|state| state.stop_state == ChildStopState::CancelDelivered);
608 let mut stale_idle_report = false;
609 let count_restart_failure = self.slots.get(&child_id).is_some_and(|state| {
610 state.operation == ChildControlOperation::Active
611 && restart_limit_counts_exit(&exit_kind)
612 });
613 let late_report = !was_active && self.shutdown.phase() == ShutdownPhase::Completed;
614 let mut fence_pending_release = None::<PendingRestart>;
615
616 if let Some(runtime_state) = self.slots.get_mut(&child_id) {
617 if matches_pending_fence {
618 if runtime_state.stop_state == ChildStopState::CancelDelivered {
619 runtime_state.stop_state = ChildStopState::Completed;
620 pending_events.push(PendingRuntimeEvent {
621 child_id: child_id.clone(),
622 path: runtime_state.path.clone(),
623 generation: Some(generation),
624 attempt: Some(attempt),
625 correlation_id: CorrelationId::from_uuid(
626 runtime_state
627 .generation_fence
628 .pending_restart
629 .as_ref()
630 .expect("matches pending implies Some")
631 .command_id,
632 ),
633 what: What::ChildControlStopCompleted {
634 child_id: child_id.clone(),
635 generation,
636 attempt,
637 exit_kind: exit_kind.clone(),
638 },
639 });
640 }
641 fence_pending_release = runtime_state.generation_fence.pending_restart.take();
642 if let Some(pending_release) = fence_pending_release.as_ref() {
643 let drained_correlation_id =
644 CorrelationId::from_uuid(pending_release.command_id);
645 pending_events.push(PendingRuntimeEvent {
646 child_id: child_id.clone(),
647 path: runtime_state.path.clone(),
648 generation: Some(generation),
649 attempt: Some(attempt),
650 correlation_id: drained_correlation_id,
651 what: What::ChildRestartFencePendingDrained {
652 child_id: child_id.clone(),
653 },
654 });
655 }
656 runtime_state.generation_fence.phase = GenerationFencePhase::ReadyToStart;
657 runtime_state.status = ChildAttemptStatus::Stopped;
658 runtime_state.clear_instance();
659 } else if matches_active_attempt
660 || late_report
661 || self.shutdown.phase() != ShutdownPhase::Idle
662 {
663 if runtime_state.stop_state == ChildStopState::CancelDelivered {
664 runtime_state.stop_state = ChildStopState::Completed;
665 pending_events.push(PendingRuntimeEvent {
666 child_id: child_id.clone(),
667 path: runtime_state.path.clone(),
668 generation: Some(generation),
669 attempt: Some(attempt),
670 correlation_id: CorrelationId::new(),
671 what: What::ChildControlStopCompleted {
672 child_id: child_id.clone(),
673 generation,
674 attempt,
675 exit_kind: exit_kind.clone(),
676 },
677 });
678 }
679 runtime_state.status = ChildAttemptStatus::Stopped;
680 runtime_state.clear_instance();
681 } else {
682 stale_idle_report = true;
683 let observed_at_unix_nanos = self.time_base.now_unix_nanos();
684 let current_generation = runtime_state.generation;
685 let current_attempt = runtime_state.attempt;
686 let stale_fact = StaleAttemptReport::new(
687 child_id.clone(),
688 generation,
689 attempt,
690 current_generation,
691 current_attempt,
692 exit_kind.clone(),
693 StaleReportHandling::RecordedForAudit,
694 observed_at_unix_nanos,
695 );
696 runtime_state.generation_fence.last_stale_report = Some(stale_fact);
697 pending_events.push(PendingRuntimeEvent {
698 child_id: child_id.clone(),
699 path: runtime_state.path.clone(),
700 generation: Some(generation),
701 attempt: Some(attempt),
702 correlation_id: CorrelationId::new(),
703 what: What::ChildAttemptStaleReport {
704 child_id: child_id.clone(),
705 reported_generation: generation,
706 reported_attempt: attempt,
707 current_generation,
708 current_attempt,
709 exit_kind: exit_kind.clone(),
710 handled_as: StaleReportHandling::RecordedForAudit,
711 },
712 });
713 }
714 }
715
716 if stale_idle_report {
717 let _ignored = event_sender.send(format!("child_exit:{child_id}"));
718 for event in pending_events {
719 self.emit_pending_event(event);
720 }
721 self.reconcile_stop_deadlines();
722 return;
723 }
724
725 self.record_child_exit(report);
726 let restart_limit_refreshed =
727 self.refresh_restart_limit_for_child(&child_id, count_restart_failure);
728 if let Some((path, restart_limit)) = restart_limit_refreshed.clone() {
729 pending_events.push(PendingRuntimeEvent {
730 child_id: child_id.clone(),
731 path,
732 generation: Some(generation),
733 attempt: Some(attempt),
734 correlation_id: CorrelationId::new(),
735 what: What::ChildRuntimeRestartLimitUpdated {
736 child_id: child_id.clone(),
737 restart_limit,
738 },
739 });
740 }
741 let _ignored = event_sender.send(format!("child_exit:{child_id}"));
742 if late_report {
743 let _ignored = event_sender.send(format!("child_shutdown_late_report:{child_id}"));
744 }
745
746 let sequence = self.event_sequences.next().value;
748 let correlation_id_str = uuid::Uuid::new_v4().to_string();
750 let supervisor_path = self
751 .slots
752 .get(&child_id)
753 .map(|state| state.path.clone())
754 .unwrap_or_else(|| SupervisorPath::root().join(child_id.value.clone()));
755
756 let mut pipeline_ctx = PipelineContext::new(
757 child_id.clone(),
758 supervisor_path,
759 sequence,
760 correlation_id_str,
761 );
762 pipeline_ctx.exit_classification = Some(classify_exit_for_pipeline(
763 &exit_kind,
764 manual_stop_requested,
765 ));
766 pipeline_ctx.effective_policy = self
767 .registry
768 .child(&child_id)
769 .map(|runtime| prepare_effective_policy(&runtime.spec));
770
771 let policy_exit = policy_task_exit(&exit_kind);
773
774 let pipeline_result = self.supervision_pipeline.execute_pipeline(
776 pipeline_ctx,
777 policy_exit,
778 &self.spec,
779 &self.tree,
780 );
781 self.record_pipeline_stage_diagnostics(&pipeline_result.stage_diagnostics);
782
783 self.fairness_probe.record_opportunity(&child_id);
785
786 self.check_fairness_probe(event_sender);
788
789 if let Some(ref budget_eval) = pipeline_result.budget_evaluation
791 && matches!(
792 budget_eval.meltdown_outcome,
793 crate::policy::meltdown::MeltdownOutcome::GroupFuse
794 )
795 && let Some(ref group_id) = pipeline_result.group_id
796 {
797 let _ignored = event_sender.send(format!("group_fuse_active:{group_id}:{child_id}"));
798 for (_cid, slot) in self.slots.iter_mut() {
800 if slot.group.as_deref() == Some(group_id.as_str()) {
801 slot.last_control_failure = Some(ChildControlFailure::new(
802 ChildControlFailurePhase::WaitCompletion,
803 format!("group_fuse_active:{group_id}"),
804 false,
805 ));
806 }
807 }
808 }
809
810 if let Some(pending) = fence_pending_release {
811 for event in pending_events {
812 self.emit_pending_event(event);
813 }
814 self.spawn_pending_restart_target(child_id.clone(), pending, exit_kind.clone());
815 self.reconcile_stop_deadlines();
816 return;
817 }
818
819 if !self.should_apply_automatic_policy(&child_id) {
820 if self
821 .slots
822 .get(&child_id)
823 .is_some_and(|state| state.operation == ChildControlOperation::Removed)
824 && let Some(removed) = self.slots.remove(&child_id)
825 {
826 pending_events.push(PendingRuntimeEvent {
827 child_id: child_id.clone(),
828 path: removed.path.clone(),
829 generation: Some(generation),
830 attempt: Some(attempt),
831 correlation_id: CorrelationId::new(),
832 what: What::ChildRuntimeStateRemoved {
833 child_id: child_id.clone(),
834 path: removed.path,
835 final_status: Some(ChildAttemptStatus::Stopped),
836 },
837 });
838 }
839 for event in pending_events {
840 self.emit_pending_event(event);
841 }
842 self.reconcile_stop_deadlines();
843 return;
844 }
845
846 let action_decision = pipeline_result.action_decision.as_ref();
848
849 let pipeline_driven_decision = if let Some(decision) = action_decision {
851 match decision.action {
852 ProtectionAction::RestartAllowed => {
853 if role_policy_restarts_success(&pipeline_result) {
854 Some(RestartDecision::RestartAfter {
855 delay: Duration::ZERO,
856 })
857 } else {
858 self.restart_decision(&child_id)
859 }
860 }
861 ProtectionAction::RestartQueued => {
862 None
864 }
865 ProtectionAction::RestartDenied
866 | ProtectionAction::SupervisionPaused
867 | ProtectionAction::Escalated
868 | ProtectionAction::SupervisedStop => {
869 None
871 }
872 }
873 } else {
874 self.restart_decision(&child_id)
876 };
877
878 let Some(decision) = pipeline_driven_decision else {
879 for event in pending_events {
880 self.emit_pending_event(event);
881 }
882 self.reconcile_stop_deadlines();
883 return;
884 };
885
886 if restart_limit_refreshed
887 .as_ref()
888 .is_some_and(|(_path, restart_limit)| {
889 restart_limit.used > restart_limit.limit
890 && matches!(decision, RestartDecision::RestartAfter { .. })
891 })
892 {
893 let _ignored = event_sender.send(format!("child_restart_limit_exhausted:{child_id}"));
894 for event in pending_events {
895 self.emit_pending_event(event);
896 }
897 self.reconcile_stop_deadlines();
898 return;
899 }
900 self.execute_restart_decision(child_id, decision, event_sender);
901 for event in pending_events {
902 self.emit_pending_event(event);
903 }
904 self.reconcile_stop_deadlines();
905 }
906
907 pub fn handle_child_start_failed(
919 &mut self,
920 child_id: ChildId,
921 message: String,
922 event_sender: &broadcast::Sender<String>,
923 ) {
924 let _ignored = event_sender.send(format!("child_start_failed:{child_id}:{message}"));
925
926 let mut fenced_spawn_recovery = Option::<(Generation, ChildStartCount, u64)>::None;
927 let mut repaired_fenced_spawn = false;
928
929 if let Some(runtime_state) = self.slots.get_mut(&child_id)
930 && runtime_state.generation_fence.phase == GenerationFencePhase::ReadyToStart
931 {
932 fenced_spawn_recovery = runtime_state
933 .registry_identity_anchor_for_spawn_attempt
934 .take();
935 repaired_fenced_spawn = true;
936 runtime_state.generation_fence.phase = GenerationFencePhase::Open;
937 runtime_state.last_control_failure = Some(ChildControlFailure::new(
938 ChildControlFailurePhase::WaitCompletion,
939 message,
940 true,
941 ));
942 }
943
944 if repaired_fenced_spawn {
945 if let Some((generation, attempt, restart_count)) = fenced_spawn_recovery
946 && let Some(registry_runtime) = self.registry.child_mut(&child_id)
947 {
948 registry_runtime.generation = generation;
949 registry_runtime.child_start_count = attempt;
950 registry_runtime.restart_count = restart_count;
951 }
952 return;
953 }
954
955 let _result = self.set_child_state(child_id, ChildControlOperation::Quarantined);
956 }
957
958 pub(crate) async fn execute_shutdown(
986 &mut self,
987 requested_by: String,
988 reason: String,
989 event_sender: &broadcast::Sender<String>,
990 ) -> Result<ShutdownResult, SupervisorError> {
991 if let Some(report) = self.shutdown_pipeline.cached_report() {
992 return Ok(self
993 .shutdown
994 .result_with_report(report.as_idempotent(), true));
995 }
996
997 let global_deadline = self.shutdown.policy.effective_global_deadline();
998 match tokio::time::timeout(
999 global_deadline,
1000 self.execute_shutdown_body(requested_by, reason, event_sender),
1001 )
1002 .await
1003 {
1004 Ok(result) => result,
1005 Err(_elapsed) => {
1006 let _ = event_sender.send("shutdown_global_timeout".to_owned());
1011
1012 let wait_order: Vec<ChildId> = self.slots.keys().cloned().collect();
1013 for child_id in &wait_order {
1014 if let Some(diag) = crate::runtime::shutdown::emergency_force_kill(
1015 &mut self.slots,
1016 child_id,
1017 &mut self.orphan_count,
1018 ) {
1019 let _ = event_sender.send(format!(
1020 "child_orphaned:{diag}:shutdown_phase=global_timeout"
1021 ));
1022 }
1023 }
1024
1025 self.advance_shutdown_phase(event_sender);
1028 self.shutdown.complete();
1029
1030 let threshold = self.shutdown.policy.effective_max_orphan_threshold() as u64;
1032 if self.orphan_count >= threshold {
1033 let _ = event_sender.send(format!(
1034 "fatal_orphan_overflow:count={}:threshold={}",
1035 self.orphan_count, threshold,
1036 ));
1037 self.exit_handler.exit(1);
1038 } else if self.orphan_count > 0 {
1039 let _ = event_sender.send(format!(
1040 "orphan_count:{}:below_threshold:{}:supervisor_degraded",
1041 self.orphan_count, threshold,
1042 ));
1043 }
1044
1045 Err(SupervisorError::FatalConfig {
1046 message: "shutdown global timeout exceeded".to_owned(),
1047 })
1048 }
1049 }
1050 }
1051
1052 async fn execute_shutdown_body(
1058 &mut self,
1059 requested_by: String,
1060 reason: String,
1061 event_sender: &broadcast::Sender<String>,
1062 ) -> Result<ShutdownResult, SupervisorError> {
1063 let cause = ShutdownCause::new(requested_by.clone(), reason.clone());
1064 let requested = self.shutdown.request_stop(cause);
1065 let started_at_unix_nanos = unix_epoch_nanos();
1066 let wait_order = self.shutdown_wait_order();
1067 let mut outcomes = HashMap::<ChildId, ChildShutdownOutcome>::new();
1068 let _ignored = event_sender.send(format!(
1069 "shutdown_phase_changed:{:?}:{:?}",
1070 ShutdownPhase::Idle,
1071 self.shutdown.phase()
1072 ));
1073 self.deliver_shutdown_cancellations(&wait_order, event_sender);
1074
1075 self.advance_shutdown_phase(event_sender);
1076 self.drain_graceful_children(&wait_order, &mut outcomes, event_sender)
1077 .await;
1078
1079 self.advance_shutdown_phase(event_sender);
1080 self.abort_remaining_children(&wait_order, &mut outcomes, event_sender)
1081 .await;
1082
1083 self.advance_shutdown_phase(event_sender);
1084 self.reconcile_shutdown_outcomes(&wait_order, &mut outcomes);
1085 for child_id in &wait_order {
1087 if let Some(diag) = crate::runtime::shutdown::emergency_force_kill(
1088 &mut self.slots,
1089 child_id,
1090 &mut self.orphan_count,
1091 ) {
1092 let _ =
1093 event_sender.send(format!("child_orphaned:{diag}:shutdown_phase=reconcile"));
1094 }
1095 }
1096 let reconcile = ShutdownReconcileReport::core_runtime_completed();
1097
1098 let from = self.shutdown.phase();
1099 self.shutdown.complete();
1100 let _ignored = event_sender.send(format!(
1101 "shutdown_phase_changed:{from:?}:{:?}",
1102 self.shutdown.phase()
1103 ));
1104 let ordered_outcomes = wait_order
1105 .iter()
1106 .filter_map(|child_id| outcomes.remove(child_id))
1107 .collect::<Vec<_>>();
1108 let report = ShutdownPipelineReport {
1109 cause: requested.cause,
1110 started_at_unix_nanos,
1111 completed_at_unix_nanos: unix_epoch_nanos(),
1112 phase: self.shutdown.phase(),
1113 outcomes: ordered_outcomes,
1114 reconcile,
1115 idempotent: false,
1116 };
1117 let _ignored = event_sender.send(format!(
1118 "shutdown_completed:{len}",
1119 len = report.outcomes.len()
1120 ));
1121 self.shutdown_pipeline.cache_report(report.clone());
1122
1123 let threshold = self.shutdown.policy.effective_max_orphan_threshold() as u64;
1126 if self.orphan_count >= threshold {
1127 let _ = event_sender.send(format!(
1128 "fatal_orphan_overflow:count={}:threshold={}",
1129 self.orphan_count, threshold,
1130 ));
1131 self.exit_handler.exit(1);
1135 } else if self.orphan_count > 0 {
1136 let _ = event_sender.send(format!(
1137 "orphan_count:{}:below_threshold:{}:supervisor_degraded",
1138 self.orphan_count, threshold,
1139 ));
1140 }
1141
1142 Ok(self.shutdown.result_with_report(report, false))
1143 }
1144
1145 fn advance_shutdown_phase(
1155 &mut self,
1156 event_sender: &broadcast::Sender<String>,
1157 ) -> ShutdownPhase {
1158 let from = self.shutdown.phase();
1159 let to = self.shutdown.advance();
1160 let _ignored = event_sender.send(format!("shutdown_phase_changed:{from:?}:{to:?}"));
1161 to
1162 }
1163
1164 fn shutdown_wait_order(&self) -> Vec<ChildId> {
1174 shutdown_order(&self.tree)
1175 .into_iter()
1176 .map(|node| node.child.id.clone())
1177 .collect()
1178 }
1179
1180 fn deliver_shutdown_cancellations(
1191 &mut self,
1192 wait_order: &[ChildId],
1193 event_sender: &broadcast::Sender<String>,
1194 ) {
1195 for child_id in wait_order {
1196 let Some(runtime_state) = self.slots.get_mut(child_id) else {
1197 continue;
1198 };
1199 if runtime_state.operation == ChildControlOperation::Removed {
1200 continue;
1201 }
1202 if !runtime_state.has_active_attempt() {
1203 continue;
1204 };
1205 runtime_state.cancel();
1206 let _ignored = event_sender.send(format!(
1207 "child_shutdown_cancel_delivered:{}:{}:{}",
1208 runtime_state.child_id,
1209 runtime_state
1210 .generation
1211 .map_or(0, |generation| generation.value),
1212 runtime_state.attempt.map_or(0, |attempt| attempt.value)
1213 ));
1214 }
1215 }
1216
1217 async fn drain_graceful_children(
1229 &mut self,
1230 wait_order: &[ChildId],
1231 outcomes: &mut HashMap<ChildId, ChildShutdownOutcome>,
1232 event_sender: &broadcast::Sender<String>,
1233 ) {
1234 for child_id in wait_order {
1235 if outcomes.contains_key(child_id) {
1236 continue;
1237 }
1238 let budget = self.child_shutdown_budget(child_id);
1239 let deadline = Instant::now() + budget.graceful_timeout;
1240 let Some(mut runtime_state) = self.slots.remove(child_id) else {
1241 continue;
1242 };
1243 if runtime_state.operation == ChildControlOperation::Removed {
1244 outcomes.insert(
1245 child_id.clone(),
1246 removed_runtime_state_shutdown_outcome(
1247 &runtime_state,
1248 ShutdownPhase::GracefulDrain,
1249 ),
1250 );
1251 self.reinsert_slot_unless_replaced(child_id.clone(), runtime_state);
1252 continue;
1253 }
1254 if !runtime_state.has_active_attempt() {
1255 self.reinsert_slot_unless_replaced(child_id.clone(), runtime_state);
1256 continue;
1257 };
1258 let completed = match remaining_duration(deadline) {
1259 Some(remaining) => timeout(remaining, runtime_state.wait_for_report())
1260 .await
1261 .ok(),
1262 None => None,
1263 };
1264 match completed {
1265 Some(Ok(report)) => {
1266 let outcome = outcome_from_report(
1267 &runtime_state,
1268 &report,
1269 ChildShutdownStatus::Graceful,
1270 ShutdownPhase::GracefulDrain,
1271 "child completed during graceful drain",
1272 );
1273 self.record_child_exit(report);
1274 runtime_state.clear_instance();
1275 self.reinsert_slot_unless_replaced(child_id.clone(), runtime_state);
1276 let _ignored = event_sender.send(format!("child_shutdown_graceful:{child_id}"));
1277 outcomes.insert(child_id.clone(), outcome);
1278 }
1279 Some(Err(error)) => {
1280 outcomes.insert(
1281 child_id.clone(),
1282 outcome_from_error(
1283 &runtime_state,
1284 ChildShutdownStatus::Graceful,
1285 ShutdownPhase::GracefulDrain,
1286 error,
1287 ),
1288 );
1289 self.reinsert_slot_unless_replaced(child_id.clone(), runtime_state);
1290 }
1291 None => {
1292 self.reinsert_slot_unless_replaced(child_id.clone(), runtime_state);
1293 }
1294 }
1295 }
1296 }
1297
1298 async fn abort_remaining_children(
1310 &mut self,
1311 wait_order: &[ChildId],
1312 outcomes: &mut HashMap<ChildId, ChildShutdownOutcome>,
1313 event_sender: &broadcast::Sender<String>,
1314 ) {
1315 let policy = self.shutdown.policy;
1316 for child_id in wait_order {
1317 if outcomes.contains_key(child_id) {
1318 continue;
1319 }
1320 let Some(mut runtime_state) = self.slots.remove(child_id) else {
1321 continue;
1322 };
1323 if runtime_state.operation == ChildControlOperation::Removed {
1324 outcomes.insert(
1325 child_id.clone(),
1326 removed_runtime_state_shutdown_outcome(
1327 &runtime_state,
1328 ShutdownPhase::AbortStragglers,
1329 ),
1330 );
1331 self.reinsert_slot_unless_replaced(child_id.clone(), runtime_state);
1332 continue;
1333 }
1334 if !runtime_state.has_active_attempt() {
1335 self.reinsert_slot_unless_replaced(child_id.clone(), runtime_state);
1336 continue;
1337 };
1338 if !policy.abort_after_timeout {
1339 let budget = self.child_shutdown_budget(child_id);
1340 self.wait_for_late_report(
1341 child_id,
1342 runtime_state,
1343 budget.abort_wait,
1344 outcomes,
1345 event_sender,
1346 )
1347 .await;
1348 continue;
1349 }
1350 runtime_state.abort();
1351 let _ignored = event_sender.send(format!(
1352 "child_shutdown_abort_requested:{}",
1353 runtime_state.child_id
1354 ));
1355 let budget = self.child_shutdown_budget(child_id);
1356 match timeout(budget.abort_wait, runtime_state.wait_for_report()).await {
1357 Ok(Ok(report)) => {
1358 let outcome = outcome_from_report(
1359 &runtime_state,
1360 &report,
1361 ChildShutdownStatus::Aborted,
1362 ShutdownPhase::AbortStragglers,
1363 "child completed after abort request",
1364 );
1365 self.record_child_exit(report);
1366 runtime_state.clear_instance();
1367 self.reinsert_slot_unless_replaced(child_id.clone(), runtime_state);
1368 let _ignored = event_sender.send(format!("child_shutdown_aborted:{child_id}"));
1369 outcomes.insert(child_id.clone(), outcome);
1370 }
1371 Ok(Err(error)) => {
1372 outcomes.insert(
1373 child_id.clone(),
1374 outcome_from_error(
1375 &runtime_state,
1376 ChildShutdownStatus::AbortFailed,
1377 ShutdownPhase::AbortStragglers,
1378 error,
1379 ),
1380 );
1381 self.reinsert_slot_unless_replaced(child_id.clone(), runtime_state);
1382 }
1383 Err(_elapsed) => {
1384 outcomes.insert(
1385 child_id.clone(),
1386 ChildShutdownOutcome::new(ChildShutdownOutcomeInput {
1387 child_id: runtime_state.child_id.clone(),
1388 path: runtime_state.path.clone(),
1389 generation: runtime_state
1390 .generation
1391 .unwrap_or_else(Generation::initial),
1392 child_start_count: runtime_state
1393 .attempt
1394 .unwrap_or_else(ChildStartCount::first),
1395 status: ChildShutdownStatus::AbortFailed,
1396 cancel_delivered: runtime_state.attempt_cancel_delivered,
1397 exit: None,
1398 phase: ShutdownPhase::AbortStragglers,
1399 reason: "child did not complete after abort request".to_owned(),
1400 }),
1401 );
1402 self.reinsert_slot_unless_replaced(child_id.clone(), runtime_state);
1403 }
1404 }
1405 }
1406 }
1407
1408 async fn wait_for_late_report(
1422 &mut self,
1423 child_id: &ChildId,
1424 mut runtime_state: ChildSlot,
1425 wait: Duration,
1426 outcomes: &mut HashMap<ChildId, ChildShutdownOutcome>,
1427 event_sender: &broadcast::Sender<String>,
1428 ) {
1429 match timeout(wait, runtime_state.wait_for_report()).await {
1430 Ok(Ok(report)) => {
1431 let outcome = outcome_from_report(
1432 &runtime_state,
1433 &report,
1434 ChildShutdownStatus::LateReport,
1435 ShutdownPhase::AbortStragglers,
1436 "child reported after graceful timeout",
1437 );
1438 self.record_child_exit(report);
1439 runtime_state.clear_instance();
1440 self.reinsert_slot_unless_replaced(child_id.clone(), runtime_state);
1441 let _ignored = event_sender.send(format!("child_shutdown_late_report:{child_id}"));
1442 outcomes.insert(child_id.clone(), outcome);
1443 }
1444 Ok(Err(error)) => {
1445 outcomes.insert(
1446 child_id.clone(),
1447 outcome_from_error(
1448 &runtime_state,
1449 ChildShutdownStatus::LateReport,
1450 ShutdownPhase::AbortStragglers,
1451 error,
1452 ),
1453 );
1454 self.reinsert_slot_unless_replaced(child_id.clone(), runtime_state);
1455 }
1456 Err(_elapsed) => {
1457 outcomes.insert(
1458 child_id.clone(),
1459 ChildShutdownOutcome::new(ChildShutdownOutcomeInput {
1460 child_id: runtime_state.child_id.clone(),
1461 path: runtime_state.path.clone(),
1462 generation: runtime_state.generation.unwrap_or_else(Generation::initial),
1463 child_start_count: runtime_state
1464 .attempt
1465 .unwrap_or_else(ChildStartCount::first),
1466 status: ChildShutdownStatus::AbortFailed,
1467 cancel_delivered: runtime_state.attempt_cancel_delivered,
1468 exit: None,
1469 phase: ShutdownPhase::AbortStragglers,
1470 reason: "abort disabled and child did not report before reconcile"
1471 .to_owned(),
1472 }),
1473 );
1474 self.reinsert_slot_unless_replaced(child_id.clone(), runtime_state);
1475 }
1476 }
1477 }
1478
1479 fn reinsert_slot_unless_replaced(&mut self, child_id: ChildId, slot: ChildSlot) {
1493 self.slots.entry(child_id).or_insert(slot);
1494 }
1498
1499 fn reconcile_shutdown_outcomes(
1510 &self,
1511 wait_order: &[ChildId],
1512 outcomes: &mut HashMap<ChildId, ChildShutdownOutcome>,
1513 ) {
1514 for child_id in wait_order {
1515 if outcomes.contains_key(child_id) {
1516 continue;
1517 }
1518 let Some(runtime) = self.registry.child(child_id) else {
1519 continue;
1520 };
1521 outcomes.insert(
1522 child_id.clone(),
1523 ChildShutdownOutcome::new(ChildShutdownOutcomeInput {
1524 child_id: runtime.id.clone(),
1525 path: runtime.path.clone(),
1526 generation: runtime.generation,
1527 child_start_count: runtime.child_start_count,
1528 status: ChildShutdownStatus::AlreadyExited,
1529 cancel_delivered: false,
1530 exit: runtime.last_exit.clone(),
1531 phase: ShutdownPhase::Reconcile,
1532 reason: "child had no active child_start_count during shutdown".to_owned(),
1533 }),
1534 );
1535 }
1536 }
1537
1538 fn set_child_state(
1549 &mut self,
1550 child_id: ChildId,
1551 operation: ChildControlOperation,
1552 ) -> CommandResult {
1553 if !self.slots.contains_key(&child_id) {
1554 let placeholder = self
1555 .registry
1556 .child(&child_id)
1557 .map(|runtime| ChildSlot::new_placeholder(runtime.id.clone(), runtime.path.clone()))
1558 .unwrap_or_else(|| {
1559 ChildSlot::new_placeholder(
1560 child_id.clone(),
1561 crate::id::types::SupervisorPath::root().join(child_id.value.clone()),
1562 )
1563 });
1564 self.slots.insert(child_id.clone(), placeholder);
1565 }
1566 let runtime_state = self
1567 .slots
1568 .get_mut(&child_id)
1569 .expect("child runtime state should exist after insertion");
1570 let operation_before = runtime_state.operation;
1571 runtime_state.operation = operation;
1572 let outcome = ChildControlResult::new(
1573 child_id,
1574 runtime_state.attempt,
1575 runtime_state.generation,
1576 operation_before,
1577 runtime_state.operation,
1578 Some(runtime_state.status),
1579 false,
1580 if runtime_state.has_active_attempt() {
1581 runtime_state.stop_state
1582 } else {
1583 ChildStopState::NoActiveAttempt
1584 },
1585 runtime_state.restart_limit.clone(),
1586 runtime_state.observe_liveness(self.time_base.now_unix_nanos(), &self.time_base),
1587 operation_before == operation,
1588 runtime_state.last_control_failure.clone(),
1589 None,
1590 );
1591 CommandResult::ChildControl { outcome }
1592 }
1593
1594 fn execute_stop_child_control(
1608 &mut self,
1609 child_id: ChildId,
1610 target_operation: ChildControlOperation,
1611 command_name: &'static str,
1612 meta: &CommandMeta,
1613 event_sender: &broadcast::Sender<String>,
1614 ) -> CommandResult {
1615 if !self.slots.contains_key(&child_id) && self.registry.child(&child_id).is_none() {
1619 return CommandResult::ChildControl {
1620 outcome: ChildControlResult {
1621 child_id: child_id.clone(),
1622 operation_before: ChildControlOperation::Active,
1623 operation_after: ChildControlOperation::Active,
1624 cancel_delivered: false,
1625 idempotent: true,
1626 attempt: None,
1627 generation: None,
1628 stop_state: ChildStopState::Idle,
1629 restart_limit: RestartLimitState::default(),
1630 liveness: ChildLivenessState::new(
1631 None,
1632 false,
1633 crate::readiness::signal::ReadinessState::Unreported,
1634 ),
1635 failure: None,
1636 generation_fence: None,
1637 status: None,
1638 admission_conflict: None,
1639 },
1640 };
1641 }
1642
1643 if !self.slots.contains_key(&child_id) {
1644 let placeholder = self
1645 .registry
1646 .child(&child_id)
1647 .map(|runtime| ChildSlot::new_placeholder(runtime.id.clone(), runtime.path.clone()))
1648 .unwrap_or_else(|| {
1649 ChildSlot::new_placeholder(
1650 child_id.clone(),
1651 crate::id::types::SupervisorPath::root().join(child_id.value.clone()),
1652 )
1653 });
1654 self.slots.insert(child_id.clone(), placeholder);
1655 }
1656
1657 let remove_after_outcome;
1658 let correlation_id = CorrelationId::from_uuid(meta.command_id.value);
1659 let mut pending_events = Vec::new();
1660 let graceful_stop_budget = self.child_shutdown_budget(&child_id).graceful_timeout;
1661 let outcome = {
1662 let runtime_state = self
1663 .slots
1664 .get_mut(&child_id)
1665 .expect("child runtime state should exist after insertion");
1666 let stop = apply_stop_control_to_runtime_state(
1667 runtime_state,
1668 target_operation,
1669 command_name,
1670 &meta.command_id.value.to_string(),
1671 correlation_id,
1672 self.time_base
1673 .now_unix_nanos()
1674 .saturating_add(graceful_stop_budget.as_nanos()),
1675 event_sender,
1676 &mut pending_events,
1677 );
1678 remove_after_outcome = stop.remove_after_outcome;
1679 build_child_control_outcome(
1680 stop.operation_before,
1681 stop.cancel_delivered,
1682 stop.idempotent,
1683 runtime_state.last_control_failure.clone(),
1684 runtime_state,
1685 &self.time_base,
1686 None,
1687 )
1688 };
1689
1690 for event in pending_events {
1691 self.emit_pending_event(event);
1692 }
1693
1694 let outcome_path = self
1695 .slots
1696 .get(&child_id)
1697 .map(|state| state.path.clone())
1698 .or_else(|| {
1699 self.registry
1700 .child(&child_id)
1701 .map(|runtime| runtime.path.clone())
1702 })
1703 .unwrap_or_else(|| SupervisorPath::root().join(child_id.value.clone()));
1704 self.emit_pending_event(PendingRuntimeEvent {
1705 child_id: child_id.clone(),
1706 path: outcome_path,
1707 generation: outcome.generation,
1708 attempt: outcome.attempt,
1709 correlation_id,
1710 what: What::ChildControlCommandCompleted {
1711 child_id: child_id.clone(),
1712 command: command_name.to_owned(),
1713 command_id: meta.command_id.value.to_string(),
1714 requested_by: meta.requested_by.clone(),
1715 reason: meta.reason.clone(),
1716 result: child_control_result_label(&outcome).to_owned(),
1717 outcome: Box::new(outcome.clone()),
1718 },
1719 });
1720
1721 if remove_after_outcome {
1722 if let Some(removed) = self.slots.remove(&child_id) {
1723 self.emit_pending_event(PendingRuntimeEvent {
1724 child_id: child_id.clone(),
1725 path: removed.path.clone(),
1726 generation: removed.generation,
1727 attempt: removed.attempt,
1728 correlation_id,
1729 what: What::ChildRuntimeStateRemoved {
1730 child_id: child_id.clone(),
1731 path: removed.path,
1732 final_status: None,
1733 },
1734 });
1735 }
1736 let _ignored = event_sender.send(format!("child_runtime_state_removed:{child_id}"));
1737 }
1738
1739 CommandResult::ChildControl { outcome }
1740 }
1741
1742 fn record_child_exit(&mut self, report: ChildRunReport) {
1752 let child_id = report.runtime.id.clone();
1753 if let Some(runtime) = self.registry.child_mut(&child_id) {
1754 runtime.last_exit = Some(report.exit);
1755 runtime.status = ChildRuntimeStatus::Exited;
1756 runtime.generation = report.runtime.generation;
1757 runtime.child_start_count = report.runtime.child_start_count;
1758 runtime.restart_count = report.runtime.restart_count;
1759 }
1760 }
1761
1762 fn should_apply_automatic_policy(&self, child_id: &ChildId) -> bool {
1772 if self.shutdown.phase() != crate::shutdown::stage::ShutdownPhase::Idle {
1773 return false;
1774 }
1775 !matches!(
1776 self.slots.get(child_id).map(|state| state.operation),
1777 Some(
1778 ChildControlOperation::Paused
1779 | ChildControlOperation::Quarantined
1780 | ChildControlOperation::Removed
1781 )
1782 )
1783 }
1784
1785 fn restart_decision(&self, child_id: &ChildId) -> Option<RestartDecision> {
1795 let runtime = self.registry.child(child_id)?;
1796 let exit = runtime.last_exit.as_ref()?;
1797 let policy_exit = policy_task_exit(exit);
1798 let restart_policy = restart_policy(runtime.spec.restart_policy);
1799 let backoff = backoff_policy(runtime.spec.backoff_policy);
1800 Some(self.policy_engine.decide(
1801 restart_policy,
1802 policy_exit,
1803 runtime.child_start_count.value,
1804 &backoff,
1805 ))
1806 }
1807
1808 fn refresh_restart_limit_for_child(
1819 &mut self,
1820 child_id: &ChildId,
1821 count_failure: bool,
1822 ) -> Option<(SupervisorPath, crate::control::outcome::RestartLimitState)> {
1823 let restart_limit = restart_limit_for_child_in_spec(&self.tree, &self.spec, child_id);
1824 let runtime_state = self.slots.get_mut(child_id)?;
1825 let updated = runtime_state.refresh_restart_limit(
1826 restart_limit.window,
1827 restart_limit.max_restarts,
1828 count_failure,
1829 &self.time_base,
1830 );
1831 Some((runtime_state.path.clone(), updated))
1832 }
1833
1834 fn execute_restart_decision(
1846 &mut self,
1847 failed_child: ChildId,
1848 decision: RestartDecision,
1849 event_sender: &broadcast::Sender<String>,
1850 ) {
1851 match decision {
1852 RestartDecision::RestartAfter { delay } => {
1853 self.restart_strategy_scope(failed_child, delay, event_sender);
1854 }
1855 RestartDecision::Quarantine => {
1856 let _result =
1857 self.set_child_state(failed_child, ChildControlOperation::Quarantined);
1858 }
1859 RestartDecision::ShutdownTree => {
1860 let cause = ShutdownCause::new("runtime", "policy requested tree shutdown");
1861 let _result = self.shutdown.request_stop(cause);
1862 }
1863 RestartDecision::EscalateToParent | RestartDecision::DoNotRestart => {}
1864 }
1865 }
1866
1867 fn restart_strategy_scope(
1879 &mut self,
1880 failed_child: ChildId,
1881 delay: Duration,
1882 event_sender: &broadcast::Sender<String>,
1883 ) {
1884 let plan = restart_execution_plan(&self.tree, &self.spec, &failed_child);
1885 let scope_label = child_scope_label(&plan.scope);
1886 let group_label = plan.group.as_deref().unwrap_or("supervisor");
1887
1888 let _ignored = event_sender.send(format!(
1889 "restart_plan:{:?}:{group_label}:{scope_label}",
1890 plan.strategy
1891 ));
1892
1893 for child_id in &plan.scope {
1898 let Some(permit) = self.concurrent_gate.try_acquire() else {
1900 let _ignored = event_sender.send(format!(
1902 "restart_throttled:concurrent_gate_saturated:{group_label}:{scope_label}:{child_id}",
1903 ));
1904 self.emit_throttle_gate_event(
1905 child_id,
1906 plan.group.as_deref(),
1907 ThrottleGateOwner::SupervisorInstance,
1908 );
1909 continue;
1910 };
1911 self.spawn_child_start(child_id.clone(), true, delay);
1917 if let Some(slot) = self.slots.get(child_id)
1920 && let (Some(gen_val), Some(att_val)) = (slot.generation, slot.attempt)
1921 {
1922 self.active_gate_permits
1923 .insert((child_id.clone(), gen_val, att_val), permit);
1924 }
1925 }
1926 }
1927
1928 fn emit_throttle_gate_event(
1940 &mut self,
1941 child_id: &ChildId,
1942 group_id: Option<&str>,
1943 owner: ThrottleGateOwner,
1944 ) {
1945 let now = Instant::now();
1946 let uptime = now
1947 .duration_since(self.time_base.base_instant)
1948 .as_millis()
1949 .min(u128::from(u64::MAX)) as u64;
1950 let monotonic_nanos = now.duration_since(self.time_base.base_instant).as_nanos();
1951 let path = self
1952 .slots
1953 .get(child_id)
1954 .map(|state| state.path.clone())
1955 .unwrap_or_else(|| SupervisorPath::root().join(child_id.value.clone()));
1956 let child_name = self
1957 .registry
1958 .child(child_id)
1959 .map(|runtime| runtime.spec.name.clone())
1960 .unwrap_or_else(|| child_id.to_string());
1961 let mut event = SupervisorEvent::new(
1962 When::new(EventTime::from_parts(
1963 monotonic_nanos,
1964 uptime,
1965 Generation::initial(),
1966 ChildStartCount::first(),
1967 )),
1968 Where::new(path).with_child(child_id.clone(), child_name),
1969 What::ChildFailed {
1970 failure: crate::error::types::TaskFailure::new(
1971 crate::error::types::TaskFailureKind::Error,
1972 "restart_throttled",
1973 format!(
1974 "restart denied by throttle gate {} for group {}",
1975 owner,
1976 group_id.unwrap_or("supervisor")
1977 ),
1978 ),
1979 },
1980 self.event_sequences.next(),
1981 CorrelationId::new(),
1982 1,
1983 );
1984 event.effective_protective_action = Some(ProtectionAction::RestartDenied);
1985 event.throttle_gate_owner = owner;
1986 if let Some(runtime) = self.registry.child(child_id) {
1987 let effective_policy = prepare_effective_policy(&runtime.spec);
1988 event.task_role = Some(effective_policy.task_role);
1989 event.used_fallback_default = effective_policy.used_fallback;
1990 event.effective_policy_source = Some(effective_policy.source);
1991 }
1992 if let Ok(mut observability) = self.observability.lock() {
1993 let _lagged = observability.emit(event);
1994 }
1995 }
1996
1997 fn record_pipeline_stage_diagnostics(&self, diagnostics: &[PipelineStageDiagnostic]) {
2007 if let Ok(mut observability) = self.observability.lock() {
2008 observability.record_pipeline_stage_diagnostics(diagnostics);
2009 }
2010 }
2011
2012 fn check_fairness_probe(&mut self, event_sender: &broadcast::Sender<String>) {
2022 let now_unix_nanos = self.time_base.now_unix_nanos();
2023 let all_child_ids: Vec<ChildId> = self.slots.keys().cloned().collect();
2024 if let Some(alert) = self.fairness_probe.check(now_unix_nanos, &all_child_ids) {
2025 let path = self
2027 .slots
2028 .get(&alert.starved_child_id)
2029 .map(|slot| slot.path.clone())
2030 .unwrap_or_else(|| {
2031 SupervisorPath::root().join(alert.starved_child_id.value.clone())
2032 });
2033 let generation = self
2034 .slots
2035 .get(&alert.starved_child_id)
2036 .and_then(|slot| slot.generation);
2037 let attempt = self
2038 .slots
2039 .get(&alert.starved_child_id)
2040 .and_then(|slot| slot.attempt);
2041 let pending = PendingRuntimeEvent {
2042 child_id: alert.starved_child_id.clone(),
2043 path,
2044 generation,
2045 attempt,
2046 correlation_id: CorrelationId::new(),
2047 what: What::FairnessProbeStarvation {
2048 starved_child_id: alert.starved_child_id.clone(),
2049 skip_count: alert.skip_count,
2050 probe_start_unix_nanos: alert.probe_start_unix_nanos,
2051 probe_end_unix_nanos: alert.probe_end_unix_nanos,
2052 },
2053 };
2054 self.emit_pending_event(pending);
2055
2056 let _ignored = event_sender.send(format!(
2058 "fairness_starvation:{}:skip_count={}:window_start={}:window_end={}",
2059 alert.starved_child_id,
2060 alert.skip_count,
2061 alert.probe_start_unix_nanos,
2062 alert.probe_end_unix_nanos,
2063 ));
2064 }
2065 }
2066
2067 fn ensure_dynamic_child_allowed(&self) -> Result<(), SupervisorError> {
2077 let current_child_count = self.dynamic_child_count();
2078 if self
2079 .spec
2080 .dynamic_supervisor_policy
2081 .allows_addition(current_child_count)
2082 {
2083 return Ok(());
2084 }
2085 Err(SupervisorError::InvalidTransition {
2086 message: "dynamic supervisor child limit reached".to_owned(),
2087 })
2088 }
2089
2090 fn dynamic_child_count(&self) -> usize {
2100 self.registry.declaration_order().len()
2101 }
2102
2103 fn execute_restart_child_control(
2115 &mut self,
2116 child_id: ChildId,
2117 meta: &CommandMeta,
2118 event_sender: &broadcast::Sender<String>,
2119 ) -> CommandResult {
2120 let correlation_id = CorrelationId::from_uuid(meta.command_id.value);
2121
2122 if self.registry.child(&child_id).is_none() {
2123 let outcome = restart_child_unknown_outcome(child_id.clone());
2124 self.emit_restart_child_completed(
2125 outcome.clone(),
2126 meta,
2127 correlation_id,
2128 event_sender,
2129 Vec::new(),
2130 );
2131 return CommandResult::ChildControl { outcome };
2132 }
2133
2134 if self.shutdown.phase() != ShutdownPhase::Idle {
2135 return self.restart_child_blocked_by_shutdown(
2136 &child_id,
2137 meta,
2138 correlation_id,
2139 event_sender,
2140 );
2141 }
2142
2143 if !self.slots.contains_key(&child_id) {
2144 let placeholder = self
2145 .registry
2146 .child(&child_id)
2147 .map(|runtime| ChildSlot::new_placeholder(runtime.id.clone(), runtime.path.clone()))
2148 .unwrap_or_else(|| {
2149 ChildSlot::new_placeholder(
2150 child_id.clone(),
2151 SupervisorPath::root().join(child_id.value.clone()),
2152 )
2153 });
2154 self.slots.insert(child_id.clone(), placeholder);
2155 }
2156
2157 let mut pending_events = Vec::new();
2158
2159 enum RestartPrep {
2161 Completed(Box<ChildControlResult>),
2163 DeferredImmediate {
2165 operation_before: ChildControlOperation,
2167 },
2168 }
2169
2170 let restart_prep = {
2171 let graceful_stop_budget = self.child_shutdown_budget(&child_id).graceful_timeout;
2172 let runtime_state = self
2173 .slots
2174 .get_mut(&child_id)
2175 .expect("runtime state exists after insertion");
2176 if runtime_state.generation_fence.pending_restart.is_some() {
2177 let pending = runtime_state
2178 .generation_fence
2179 .pending_restart
2180 .as_mut()
2181 .expect("checked pending restart");
2182 pending.duplicate_request_count = pending.duplicate_request_count.saturating_add(1);
2183 let pending_for_conflict = pending.clone();
2184 pending_events.push(PendingRuntimeEvent {
2185 child_id: child_id.clone(),
2186 path: runtime_state.path.clone(),
2187 generation: Some(pending_for_conflict.old_generation),
2188 attempt: Some(pending_for_conflict.old_attempt),
2189 correlation_id: CorrelationId::from_uuid(meta.command_id.value),
2190 what: What::ChildRestartConflict {
2191 child_id: child_id.clone(),
2192 current_generation: Some(pending_for_conflict.old_generation),
2193 current_attempt: Some(pending_for_conflict.old_attempt),
2194 target_generation: Some(pending_for_conflict.target_generation),
2195 command_id: meta.command_id.value.to_string(),
2196 decision: "already_pending".to_owned(),
2197 reason: "duplicate restart merged into pending restart".to_owned(),
2198 },
2199 });
2200 let fence = GenerationFenceOutcome::new(
2201 GenerationFenceDecision::AlreadyPending,
2202 Some(pending_for_conflict.old_generation),
2203 Some(pending_for_conflict.old_attempt),
2204 Some(pending_for_conflict.target_generation),
2205 false,
2206 pending_for_conflict.abort_requested,
2207 None,
2208 );
2209 let operation_before = runtime_state.operation;
2210 RestartPrep::Completed(Box::new(build_child_control_outcome(
2211 operation_before,
2212 false,
2213 false,
2214 runtime_state.last_control_failure.clone(),
2215 runtime_state,
2216 &self.time_base,
2217 Some(fence),
2218 )))
2219 } else if !runtime_state.has_active_attempt() {
2220 RestartPrep::DeferredImmediate {
2221 operation_before: runtime_state.operation,
2222 }
2223 } else {
2224 let old_generation = runtime_state
2225 .generation
2226 .expect("active attempt owns a generation");
2227 let old_attempt = runtime_state
2228 .attempt
2229 .expect("active attempt owns an attempt counter");
2230 let cancel_delivered = runtime_state.cancel();
2231 let deadline = self
2232 .time_base
2233 .now_unix_nanos()
2234 .saturating_add(graceful_stop_budget.as_nanos());
2235 runtime_state.stop_deadline_at_unix_nanos = Some(deadline);
2236 if cancel_delivered {
2237 runtime_state.stop_state = ChildStopState::CancelDelivered;
2238 }
2239 let target_generation = old_generation.next();
2240 let requested_at = self.time_base.now_unix_nanos();
2241 let pending = PendingRestart::new(
2242 meta.command_id.value,
2243 meta.requested_by.clone(),
2244 meta.reason.clone(),
2245 old_generation,
2246 old_attempt,
2247 target_generation,
2248 requested_at,
2249 deadline,
2250 false,
2251 0,
2252 );
2253 runtime_state.generation_fence.pending_restart = Some(pending.clone());
2254 runtime_state.generation_fence.phase = GenerationFencePhase::WaitingForOldStop;
2255
2256 if cancel_delivered {
2257 pending_events.push(PendingRuntimeEvent {
2258 child_id: child_id.clone(),
2259 path: runtime_state.path.clone(),
2260 generation: Some(old_generation),
2261 attempt: Some(old_attempt),
2262 correlation_id,
2263 what: What::ChildControlCancelDelivered {
2264 child_id: child_id.clone(),
2265 generation: old_generation,
2266 attempt: old_attempt,
2267 command: "restart_child".to_owned(),
2268 command_id: meta.command_id.value.to_string(),
2269 },
2270 });
2271 let _ignored = event_sender.send(format!(
2272 "child_control_cancel_delivered:{child_id}:restart_child"
2273 ));
2274 }
2275
2276 pending_events.push(PendingRuntimeEvent {
2277 child_id: child_id.clone(),
2278 path: runtime_state.path.clone(),
2279 generation: Some(old_generation),
2280 attempt: Some(old_attempt),
2281 correlation_id,
2282 what: What::ChildRestartFenceEntered {
2283 child_id: child_id.clone(),
2284 old_generation,
2285 old_attempt,
2286 target_generation,
2287 command_id: meta.command_id.value.to_string(),
2288 requested_by: meta.requested_by.clone(),
2289 reason: meta.reason.clone(),
2290 stop_deadline_at_unix_nanos: deadline,
2291 },
2292 });
2293
2294 let operation_before = runtime_state.operation;
2295 let fence = GenerationFenceOutcome::new(
2296 GenerationFenceDecision::QueuedAfterStop,
2297 Some(old_generation),
2298 Some(old_attempt),
2299 Some(target_generation),
2300 cancel_delivered,
2301 false,
2302 None,
2303 );
2304 RestartPrep::Completed(Box::new(build_child_control_outcome(
2305 operation_before,
2306 cancel_delivered,
2307 false,
2308 None,
2309 runtime_state,
2310 &self.time_base,
2311 Some(fence),
2312 )))
2313 }
2314 };
2315
2316 let outcome = match restart_prep {
2317 RestartPrep::Completed(outcome) => *outcome,
2318 RestartPrep::DeferredImmediate { operation_before } => {
2319 self.spawn_child_start(child_id.clone(), true, Duration::ZERO);
2320 let runtime_state = self.slots.get_mut(&child_id).expect("runtime state exists");
2321 let target_generation = self
2322 .registry
2323 .child(&child_id)
2324 .map(|runtime| runtime.generation);
2325 let fence = GenerationFenceOutcome::new(
2326 GenerationFenceDecision::StartedImmediately,
2327 None,
2328 None,
2329 target_generation,
2330 false,
2331 false,
2332 None,
2333 );
2334 build_child_control_outcome(
2335 operation_before,
2336 false,
2337 false,
2338 runtime_state.last_control_failure.clone(),
2339 runtime_state,
2340 &self.time_base,
2341 Some(fence),
2342 )
2343 }
2344 };
2345
2346 self.emit_restart_child_completed(
2347 outcome.clone(),
2348 meta,
2349 correlation_id,
2350 event_sender,
2351 pending_events,
2352 );
2353
2354 CommandResult::ChildControl { outcome }
2355 }
2356
2357 fn emit_restart_child_completed(
2371 &mut self,
2372 outcome: ChildControlResult,
2373 meta: &CommandMeta,
2374 correlation_id: CorrelationId,
2375 event_sender: &broadcast::Sender<String>,
2376 mut pending_events: Vec<PendingRuntimeEvent>,
2377 ) {
2378 for event in pending_events.drain(..) {
2379 self.emit_pending_event(event);
2380 }
2381 let outcome_identifier = outcome.child_id.clone();
2382 let outcome_path = self
2383 .slots
2384 .get(&outcome.child_id)
2385 .map(|state| state.path.clone())
2386 .or_else(|| {
2387 self.registry
2388 .child(&outcome.child_id)
2389 .map(|runtime| runtime.path.clone())
2390 })
2391 .unwrap_or_else(|| SupervisorPath::root().join(outcome.child_id.value.clone()));
2392 self.emit_pending_event(PendingRuntimeEvent {
2393 child_id: outcome.child_id.clone(),
2394 path: outcome_path,
2395 generation: outcome.generation,
2396 attempt: outcome.attempt,
2397 correlation_id,
2398 what: What::ChildControlCommandCompleted {
2399 child_id: outcome.child_id.clone(),
2400 command: "restart_child".to_owned(),
2401 command_id: meta.command_id.value.to_string(),
2402 requested_by: meta.requested_by.clone(),
2403 reason: meta.reason.clone(),
2404 result: child_control_result_label(&outcome).to_owned(),
2405 outcome: Box::new(outcome),
2406 },
2407 });
2408 let _ignored = event_sender.send(format!(
2409 "child_control_command_completed:{outcome_identifier}:restart_child"
2410 ));
2411 }
2412
2413 fn restart_child_blocked_by_shutdown(
2426 &mut self,
2427 child_id: &ChildId,
2428 meta: &CommandMeta,
2429 correlation_id: CorrelationId,
2430 event_sender: &broadcast::Sender<String>,
2431 ) -> CommandResult {
2432 if !self.slots.contains_key(child_id) {
2433 let placeholder = self
2434 .registry
2435 .child(child_id)
2436 .map(|runtime| ChildSlot::new_placeholder(runtime.id.clone(), runtime.path.clone()))
2437 .unwrap_or_else(|| {
2438 ChildSlot::new_placeholder(
2439 child_id.clone(),
2440 SupervisorPath::root().join(child_id.value.clone()),
2441 )
2442 });
2443 self.slots.insert(child_id.clone(), placeholder);
2444 }
2445
2446 let outcome = {
2447 let runtime_state = self.slots.get_mut(child_id).expect("runtime state exists");
2448 runtime_state.generation_fence.phase = GenerationFencePhase::Closed;
2449 let failure = ChildControlFailure::new(
2450 ChildControlFailurePhase::WaitCompletion,
2451 "supervisor tree is shutting down",
2452 false,
2453 );
2454 let fence = GenerationFenceOutcome::new(
2455 GenerationFenceDecision::BlockedByShutdown,
2456 runtime_state.generation,
2457 runtime_state.attempt,
2458 None,
2459 false,
2460 false,
2461 Some(failure.clone()),
2462 );
2463 let operation_before = runtime_state.operation;
2464 runtime_state.last_control_failure = Some(failure);
2465 build_child_control_outcome(
2466 operation_before,
2467 false,
2468 false,
2469 runtime_state.last_control_failure.clone(),
2470 runtime_state,
2471 &self.time_base,
2472 Some(fence),
2473 )
2474 };
2475
2476 let blocked_events = match self.slots.get(child_id).map(|runtime_state| {
2477 (
2478 runtime_state.path.clone(),
2479 runtime_state.generation,
2480 runtime_state.attempt,
2481 )
2482 }) {
2483 Some((path, current_generation, current_attempt)) => {
2484 vec![PendingRuntimeEvent {
2485 child_id: child_id.clone(),
2486 path,
2487 generation: current_generation,
2488 attempt: current_attempt,
2489 correlation_id,
2490 what: What::ChildRestartConflict {
2491 child_id: child_id.clone(),
2492 current_generation,
2493 current_attempt,
2494 target_generation: None,
2495 command_id: meta.command_id.value.to_string(),
2496 decision: "rejected".to_owned(),
2497 reason: "restart rejected while supervisor tree is shutting down"
2498 .to_owned(),
2499 },
2500 }]
2501 }
2502 None => Vec::new(),
2503 };
2504
2505 self.emit_restart_child_completed(
2506 outcome.clone(),
2507 meta,
2508 correlation_id,
2509 event_sender,
2510 blocked_events,
2511 );
2512
2513 CommandResult::ChildControl { outcome }
2514 }
2515
2516 fn build_current_state(&mut self) -> CurrentState {
2526 let mut child_runtime_records = Vec::new();
2527 let mut pending_events = Vec::new();
2528 let declaration_order = self.registry.declaration_order().to_vec();
2529 for child_id in declaration_order {
2530 if let Some(runtime_state) = self.slots.get_mut(&child_id) {
2531 let liveness = runtime_state
2532 .observe_liveness(self.time_base.now_unix_nanos(), &self.time_base);
2533 if let Some(event) = heartbeat_stale_event(runtime_state, &liveness) {
2534 pending_events.push(event);
2535 }
2536 child_runtime_records.push(runtime_state.to_record(liveness));
2537 }
2538 }
2539 for (child_id, runtime_state) in &mut self.slots {
2540 if self.registry.child(child_id).is_some() {
2541 continue;
2542 }
2543 let liveness =
2544 runtime_state.observe_liveness(self.time_base.now_unix_nanos(), &self.time_base);
2545 if let Some(event) = heartbeat_stale_event(runtime_state, &liveness) {
2546 pending_events.push(event);
2547 }
2548 child_runtime_records.push(runtime_state.to_record(liveness));
2549 }
2550 for event in pending_events {
2551 self.emit_pending_event(event);
2552 }
2553 CurrentState {
2554 child_count: self.dynamic_child_count(),
2555 shutdown_completed: self.shutdown.phase()
2556 == crate::shutdown::stage::ShutdownPhase::Completed,
2557 child_runtime_records,
2558 }
2559 }
2560
2561 fn spawn_pending_restart_target(
2573 &mut self,
2574 child_id: ChildId,
2575 pending: PendingRestart,
2576 old_exit: TaskExit,
2577 ) {
2578 let Some(registry_identity_anchor) = self.registry.child(&child_id).map(|runtime| {
2579 (
2580 runtime.generation,
2581 runtime.child_start_count,
2582 runtime.restart_count,
2583 )
2584 }) else {
2585 return;
2586 };
2587 let path = self
2588 .slots
2589 .get(&child_id)
2590 .map(|state| state.path.clone())
2591 .unwrap_or_else(|| SupervisorPath::root().join(child_id.value.clone()));
2592 let correlation_id = CorrelationId::from_uuid(pending.command_id);
2593
2594 if let Some(runtime_state) = self.slots.get_mut(&child_id) {
2595 runtime_state.registry_identity_anchor_for_spawn_attempt =
2596 Some(registry_identity_anchor);
2597 }
2598
2599 {
2600 let Some(registry_runtime) = self.registry.child_mut(&child_id) else {
2601 return;
2602 };
2603 registry_runtime.generation = pending.target_generation;
2604 registry_runtime.child_start_count = registry_runtime.child_start_count.next();
2605 registry_runtime.restart_count = registry_runtime.restart_count.saturating_add(1);
2606 registry_runtime.status = ChildRuntimeStatus::Starting;
2607 }
2608
2609 let Some(runtime) = self.registry.child(&child_id).cloned() else {
2610 return;
2611 };
2612 let new_generation = runtime.generation;
2613 let new_attempt = runtime.child_start_count;
2614
2615 let path_for_handles = path.clone();
2616
2617 match ChildRunner::new().spawn_once(runtime) {
2618 Ok(handle) => {
2619 let sender = self.command_sender.clone();
2620 self.emit_pending_event(PendingRuntimeEvent {
2621 child_id: child_id.clone(),
2622 path,
2623 generation: Some(new_generation),
2624 attempt: Some(new_attempt),
2625 correlation_id,
2626 what: What::ChildRestartFenceReleased {
2627 child_id: child_id.clone(),
2628 old_generation: pending.old_generation,
2629 old_attempt: pending.old_attempt,
2630 target_generation: pending.target_generation,
2631 exit_kind: old_exit.clone(),
2632 },
2633 });
2634 let mut completion_receiver = handle.completion_receiver.clone();
2635 self.slots
2636 .entry(child_id.clone())
2637 .or_insert_with(|| {
2638 ChildSlot::new_placeholder(child_id.clone(), path_for_handles)
2639 })
2640 .activate(
2641 new_generation,
2642 new_attempt,
2643 ChildAttemptStatus::Running,
2644 handle,
2645 );
2646 tokio::spawn(async move {
2647 let result = wait_for_report(&mut completion_receiver).await;
2648 send_child_result(sender, child_id, result).await;
2649 });
2650 }
2651 Err(error) => {
2652 let message = error.to_string();
2653 if let Some(runtime_state) = self.slots.get_mut(&child_id) {
2654 let identity_anchor_triple_opt = runtime_state
2655 .registry_identity_anchor_for_spawn_attempt
2656 .take();
2657 if let Some((generation, attempt, restart_count)) = identity_anchor_triple_opt {
2658 if let Some(registry_runtime) = self.registry.child_mut(&child_id) {
2659 registry_runtime.generation = generation;
2660 registry_runtime.child_start_count = attempt;
2661 registry_runtime.restart_count = restart_count;
2662 }
2663 runtime_state.generation = Some(generation);
2665 runtime_state.attempt = Some(attempt);
2666 runtime_state.status = ChildAttemptStatus::Stopped;
2667 }
2668 runtime_state.generation_fence.phase = GenerationFencePhase::Open;
2669 runtime_state.last_control_failure = Some(ChildControlFailure::new(
2670 ChildControlFailurePhase::WaitCompletion,
2671 message,
2672 true,
2673 ));
2674 }
2675 }
2677 }
2678 }
2679
2680 fn reconcile_stop_deadlines(&mut self) {
2690 let now = self.time_base.now_unix_nanos();
2691 let mut pending_events = Vec::new();
2692 for runtime_state in self.slots.values_mut() {
2693 let fence_escalation = if let Some(pending_restart) =
2694 runtime_state.generation_fence.pending_restart.as_ref()
2695 {
2696 if pending_restart.abort_requested {
2697 None
2698 } else if runtime_state.generation_fence.phase
2699 == GenerationFencePhase::WaitingForOldStop
2700 && runtime_state.stop_state == ChildStopState::CancelDelivered
2701 && runtime_state.has_active_attempt()
2702 && now >= pending_restart.stop_deadline_at_unix_nanos
2703 {
2704 match (runtime_state.generation, runtime_state.attempt) {
2705 (Some(old_generation), Some(old_attempt)) => Some((
2706 pending_restart.command_id,
2707 pending_restart.target_generation,
2708 pending_restart.stop_deadline_at_unix_nanos,
2709 runtime_state.child_id.clone(),
2710 runtime_state.path.clone(),
2711 old_generation,
2712 old_attempt,
2713 )),
2714 _ => None,
2715 }
2716 } else {
2717 None
2718 }
2719 } else {
2720 None
2721 };
2722
2723 if let Some((
2724 command_id,
2725 target_generation,
2726 deadline_ns,
2727 fence_child_id,
2728 fence_path,
2729 old_generation,
2730 old_attempt,
2731 )) = fence_escalation
2732 {
2733 let delivered = runtime_state.abort();
2734 if delivered {
2735 if let Some(pending_mut) = &mut runtime_state.generation_fence.pending_restart {
2736 pending_mut.abort_requested = true;
2737 }
2738 runtime_state.generation_fence.phase = GenerationFencePhase::AbortingOld;
2739 pending_events.push(PendingRuntimeEvent {
2740 child_id: fence_child_id.clone(),
2741 path: fence_path,
2742 generation: Some(old_generation),
2743 attempt: Some(old_attempt),
2744 correlation_id: CorrelationId::from_uuid(command_id),
2745 what: What::ChildRestartFenceAbortRequested {
2746 child_id: fence_child_id,
2747 old_generation,
2748 old_attempt,
2749 target_generation,
2750 command_id: command_id.to_string(),
2751 deadline_unix_nanos: deadline_ns,
2752 },
2753 });
2754 }
2755 }
2756
2757 if runtime_state.generation_fence.pending_restart.is_some() {
2758 continue;
2759 }
2760
2761 if matches!(
2762 runtime_state.generation_fence.phase,
2763 GenerationFencePhase::WaitingForOldStop | GenerationFencePhase::AbortingOld
2764 ) {
2765 continue;
2766 }
2767
2768 if runtime_state.stop_state != ChildStopState::CancelDelivered {
2769 continue;
2770 }
2771 let Some(deadline) = runtime_state.stop_deadline_at_unix_nanos else {
2772 continue;
2773 };
2774 if deadline > now || !runtime_state.has_active_attempt() {
2775 continue;
2776 }
2777 let Some(generation) = runtime_state.generation else {
2778 continue;
2779 };
2780 let Some(attempt) = runtime_state.attempt else {
2781 continue;
2782 };
2783 let status = runtime_state.status;
2784 let failure = ChildControlFailure::new(
2785 ChildControlFailurePhase::WaitCompletion,
2786 "child did not complete before stop deadline",
2787 true,
2788 );
2789 runtime_state.status = status;
2790 runtime_state.stop_state = ChildStopState::Failed;
2791 runtime_state.last_control_failure = Some(failure.clone());
2792 pending_events.push(PendingRuntimeEvent {
2793 child_id: runtime_state.child_id.clone(),
2794 path: runtime_state.path.clone(),
2795 generation: Some(generation),
2796 attempt: Some(attempt),
2797 correlation_id: CorrelationId::new(),
2798 what: What::ChildControlStopFailed {
2799 child_id: runtime_state.child_id.clone(),
2800 generation,
2801 attempt,
2802 status,
2803 stop_state: ChildStopState::Failed,
2804 phase: failure.phase,
2805 reason: failure.reason,
2806 recoverable: failure.recoverable,
2807 },
2808 });
2809 }
2810 for event in pending_events {
2811 self.emit_pending_event(event);
2812 }
2813 }
2814
2815 fn emit_pending_event(&mut self, pending: PendingRuntimeEvent) {
2825 let now = Instant::now();
2826 let uptime = now
2827 .duration_since(self.time_base.base_instant)
2828 .as_millis()
2829 .min(u128::from(u64::MAX)) as u64;
2830 let monotonic_nanos = now.duration_since(self.time_base.base_instant).as_nanos();
2831 let child_name = self
2832 .registry
2833 .child(&pending.child_id)
2834 .map(|runtime| runtime.spec.name.clone())
2835 .unwrap_or_else(|| pending.child_id.to_string());
2836 let event = SupervisorEvent::new(
2837 When::new(EventTime::from_parts(
2838 monotonic_nanos,
2839 uptime,
2840 pending.generation.unwrap_or_else(Generation::initial),
2841 pending.attempt.unwrap_or_else(ChildStartCount::first),
2842 )),
2843 Where::new(pending.path).with_child(pending.child_id, child_name),
2844 pending.what,
2845 self.event_sequences.next(),
2846 pending.correlation_id,
2847 1,
2848 );
2849 if let Ok(mut observability) = self.observability.lock() {
2850 let _lagged = observability.emit(event);
2851 }
2852 }
2853
2854 fn spawn_child_start(&mut self, child_id: ChildId, is_restart: bool, delay: Duration) {
2866 if self.shutdown.phase() != ShutdownPhase::Idle {
2867 return;
2868 }
2869 if let Some(runtime_state) = self.slots.get(&child_id) {
2870 if runtime_state.generation_fence.pending_restart.is_some() {
2871 if is_restart {
2872 let path = runtime_state.path.clone();
2873 let generation = runtime_state.generation;
2874 let attempt = runtime_state.attempt;
2875 let pending_target = runtime_state
2876 .generation_fence
2877 .pending_restart
2878 .as_ref()
2879 .map(|pending| pending.target_generation);
2880 self.emit_pending_event(PendingRuntimeEvent {
2881 child_id: child_id.clone(),
2882 path,
2883 generation,
2884 attempt,
2885 correlation_id: CorrelationId::new(),
2886 what: What::ChildRestartConflict {
2887 child_id: child_id.clone(),
2888 current_generation: generation,
2889 current_attempt: attempt,
2890 target_generation: pending_target,
2891 command_id: "runtime-policy".to_owned(),
2892 decision: "rejected".to_owned(),
2893 reason: "automatic restart suppressed while pending manual restart holds the fence".to_owned(),
2894 },
2895 });
2896 }
2897 return;
2898 }
2899 if matches!(
2900 runtime_state.generation_fence.phase,
2901 GenerationFencePhase::WaitingForOldStop
2902 | GenerationFencePhase::AbortingOld
2903 | GenerationFencePhase::Closed
2904 | GenerationFencePhase::ReadyToStart
2905 ) {
2906 return;
2907 }
2908 }
2909 let Some(mut runtime) = self.prepare_child_start(&child_id, is_restart) else {
2910 return;
2911 };
2912
2913 if let Some(slot) = self.slots.get(&child_id)
2918 && slot.orphaned
2919 {
2920 runtime.spec.isolation = crate::spec::child::Isolation::BlockingPool;
2921 }
2922
2923 for path in &runtime.spec.cleanup_paths {
2928 if let Err(error) = std::fs::remove_file(path)
2929 && error.kind() != std::io::ErrorKind::NotFound
2930 {
2931 tracing::warn!(
2932 child_id = %child_id,
2933 path = %path.display(),
2934 ?error,
2935 "failed to clean up orphaned resource before spawn",
2936 );
2937 }
2938 }
2939
2940 let sender = self.command_sender.clone();
2941 if !delay.is_zero() {
2942 tokio::spawn(async move {
2943 tokio::time::sleep(delay).await;
2944 let child_id_for_msg = runtime.id.clone();
2945 let path = runtime.path.clone();
2946 let generation = runtime.generation;
2947 let attempt = runtime.child_start_count;
2948 match ChildRunner::new().spawn_once(runtime) {
2949 Ok(handle) => {
2950 let _ignored = sender
2951 .send(RuntimeLoopMessage::ChildStart(
2952 ChildStartMessage::DelayedSpawnAttached {
2953 child_id: child_id_for_msg,
2954 path,
2955 generation,
2956 attempt,
2957 handle,
2958 },
2959 ))
2960 .await;
2961 }
2962 Err(error) => {
2963 tokio::spawn(async move {
2964 send_child_result(sender, child_id_for_msg, Err(error)).await;
2965 });
2966 }
2967 }
2968 });
2969 return;
2970 }
2971
2972 let child_id_cloned = runtime.id.clone();
2973 let path = runtime.path.clone();
2974 let generation = runtime.generation;
2975 let child_start_count = runtime.child_start_count;
2976 match ChildRunner::new().spawn_once(runtime) {
2977 Ok(handle) => {
2978 self.attach_spawned_child_handle(
2979 child_id_cloned,
2980 path,
2981 generation,
2982 child_start_count,
2983 handle,
2984 );
2985 }
2986 Err(error) => {
2987 tokio::spawn(async move {
2988 send_child_result(sender, child_id_cloned, Err(error)).await;
2989 });
2990 }
2991 }
2992 }
2993
2994 fn prepare_child_start(
3005 &mut self,
3006 child_id: &ChildId,
3007 bump_restart_counters: bool,
3008 ) -> Option<ChildRuntime> {
3009 let runtime = self.registry.child_mut(child_id)?;
3010 if bump_restart_counters {
3011 runtime.child_start_count = runtime.child_start_count.next();
3012 runtime.generation = runtime.generation.next();
3013 runtime.restart_count = runtime.restart_count.saturating_add(1);
3014 }
3015 runtime.status = ChildRuntimeStatus::Starting;
3016 if let Some(runtime_state) = self.slots.get_mut(child_id) {
3017 runtime_state.operation = ChildControlOperation::Active;
3018 }
3019 Some(runtime.clone())
3020 }
3021
3022 pub(crate) async fn handle_shutdown_tree(
3037 &mut self,
3038 requested_by: String,
3039 reason: String,
3040 event_sender: &broadcast::Sender<String>,
3041 ) -> Result<ShutdownResult, SupervisorError> {
3042 let policy = self.shutdown.policy;
3043 let reason_copy = reason.clone();
3044 let cause = ShutdownCause::new(requested_by, reason);
3045 let _started = self.shutdown.request_stop(cause);
3046 let _ignored = event_sender.send(format!(
3047 "shutdown_phase_changed:{:?}:{:?}",
3048 ShutdownPhase::Idle,
3049 self.shutdown.phase()
3050 ));
3051 self.advance_shutdown_phase(event_sender);
3052 self.advance_shutdown_phase(event_sender);
3053
3054 let outcomes = shutdown_tree_fanout(
3055 &mut self.slots,
3056 &policy,
3057 &mut self.admission_set,
3058 &mut self.orphan_count,
3059 )
3060 .await;
3061 let reconcile = reconcile_shutdown_slots(&self.slots);
3062
3063 if !reconcile.verified_clean {
3065 let _ignored = event_sender.send(format!(
3066 "shutdown_reconcile_warning: orphan_slots={:?}",
3067 reconcile.orphan_slots
3068 ));
3069 }
3070
3071 self.advance_shutdown_phase(event_sender);
3072 self.advance_shutdown_phase(event_sender);
3073 let _completed = self.shutdown.complete();
3074
3075 let report = ShutdownPipelineReport {
3076 cause: ShutdownCause::new("slot-shutdown", reason_copy),
3077 started_at_unix_nanos: unix_epoch_nanos(),
3078 completed_at_unix_nanos: unix_epoch_nanos(),
3079 phase: ShutdownPhase::Completed,
3080 outcomes,
3081 reconcile: ShutdownReconcileReport::core_runtime_completed(),
3082 idempotent: false,
3083 };
3084 self.shutdown_pipeline.cache_report(report.clone());
3085 let _ignored = event_sender.send(format!(
3086 "shutdown_completed:{len}",
3087 len = report.outcomes.len()
3088 ));
3089
3090 let threshold = self.shutdown.policy.effective_max_orphan_threshold() as u64;
3092 if self.orphan_count >= threshold {
3093 let _ = event_sender.send(format!(
3094 "fatal_orphan_overflow:count={}:threshold={}",
3095 self.orphan_count, threshold,
3096 ));
3097 self.exit_handler.exit(1);
3098 } else if self.orphan_count > 0 {
3099 let _ = event_sender.send(format!(
3100 "orphan_count:{}:below_threshold:{}:supervisor_degraded",
3101 self.orphan_count, threshold,
3102 ));
3103 }
3104
3105 Ok(self.shutdown.result_with_report(report, false))
3106 }
3107
3108 pub(crate) fn handle_command_on_slot(
3119 &mut self,
3120 child_id: &ChildId,
3121 operation: ChildControlOperation,
3122 ) -> bool {
3123 let Some(slot) = self.slots.get_mut(child_id) else {
3124 return false;
3125 };
3126 slot.operation = operation;
3127 if matches!(
3128 operation,
3129 ChildControlOperation::Quarantined | ChildControlOperation::Removed
3130 ) && slot.has_active_attempt()
3131 {
3132 slot.cancel();
3133 }
3134 true
3135 }
3136
3137 pub(crate) fn process_child_exit_on_slot(
3149 &mut self,
3150 child_id: &ChildId,
3151 report: &ChildRunReport,
3152 ) -> Option<ChildExitSummary> {
3153 let slot = self.slots.get_mut(child_id)?;
3154 let now_nanos = self.time_base.now_unix_nanos();
3155 let summary = ChildExitSummary::from_report(report, now_nanos);
3156 slot.deactivate(summary.clone());
3157 self.admission_set.release(child_id);
3158 Some(summary)
3159 }
3160
3161 pub(crate) fn observe_slot_liveness(
3172 &mut self,
3173 event_sender: &broadcast::Sender<String>,
3174 ) -> usize {
3175 let mut stale_count = 0usize;
3176 let threshold_nanos = Duration::from_secs(DEFAULT_HEARTBEAT_TIMEOUT_SECS).as_nanos();
3177 let now_nanos = self.time_base.now_unix_nanos();
3178
3179 for (child_id, slot) in self.slots.iter_mut() {
3180 if !slot.has_active_attempt() {
3181 continue;
3182 }
3183 if let Some(last_hb) = slot.last_heartbeat_at
3184 && now_nanos.saturating_sub(last_hb) >= threshold_nanos
3185 {
3186 stale_count += 1;
3187 let _ignored = event_sender.send(format!(
3188 "child_liveness_stale: child_id={child_id} last_heartbeat_at={last_hb}"
3189 ));
3190 }
3191 }
3192 stale_count
3193 }
3194
3195 pub(crate) fn check_slot_restart_eligibility(
3210 &self,
3211 child_id: &ChildId,
3212 request_generation: Generation,
3213 request_attempt: ChildStartCount,
3214 ) -> Result<(), AdmissionConflict> {
3215 let Some(slot) = self.slots.get(child_id) else {
3216 return Ok(());
3217 };
3218 if slot.pending_restart {
3219 return Err(AdmissionConflict::new(
3220 child_id.clone(),
3221 slot.generation.unwrap_or(Generation::initial()),
3222 slot.attempt.unwrap_or(ChildStartCount::first()),
3223 "restart rejected: pending restart already exists",
3224 ));
3225 }
3226 if let (Some(active_gen), Some(active_att)) = (slot.generation, slot.attempt)
3227 && (request_generation != active_gen || request_attempt != active_att)
3228 {
3229 return Err(AdmissionConflict::new(
3230 child_id.clone(),
3231 active_gen,
3232 active_att,
3233 "restart conflicts with active attempt",
3234 ));
3235 }
3236 Ok(())
3237 }
3238
3239 pub(crate) fn ensure_slot_exists(&mut self, child_id: ChildId, path: SupervisorPath) -> bool {
3251 if self.slots.contains_key(&child_id) {
3252 return false;
3253 }
3254 let slot = ChildSlot::new(
3255 child_id.clone(),
3256 path,
3257 Duration::from_secs(60), );
3259 self.slots.insert(child_id, slot);
3260 true
3261 }
3262}
3263
3264fn build_child_control_outcome(
3280 operation_before: ChildControlOperation,
3281 cancel_delivered: bool,
3282 idempotent: bool,
3283 failure: Option<ChildControlFailure>,
3284 runtime_state: &mut ChildSlot,
3285 time_base: &RuntimeTimeBase,
3286 generation_fence: Option<GenerationFenceOutcome>,
3287) -> ChildControlResult {
3288 let liveness = runtime_state.observe_liveness(time_base.now_unix_nanos(), time_base);
3289 let status = if runtime_state.attempt.is_some() {
3291 Some(runtime_state.status)
3292 } else {
3293 None
3294 };
3295 ChildControlResult::new(
3296 runtime_state.child_id.clone(),
3297 runtime_state.attempt,
3298 runtime_state.generation,
3299 operation_before,
3300 runtime_state.operation,
3301 status,
3302 cancel_delivered,
3303 runtime_state.stop_state,
3304 runtime_state.restart_limit.clone(),
3305 liveness,
3306 idempotent,
3307 failure,
3308 generation_fence,
3309 )
3310}
3311
3312#[derive(Debug, Clone, Copy)]
3314struct StopControlApplication {
3315 operation_before: ChildControlOperation,
3317 cancel_delivered: bool,
3319 idempotent: bool,
3321 remove_after_outcome: bool,
3323}
3324
3325#[allow(clippy::too_many_arguments)]
3342fn apply_stop_control_to_runtime_state(
3343 runtime_state: &mut ChildSlot,
3344 target_operation: ChildControlOperation,
3345 command_name: &'static str,
3346 command_id: &str,
3347 correlation_id: CorrelationId,
3348 stop_deadline_at_unix_nanos: u128,
3349 event_sender: &broadcast::Sender<String>,
3350 pending_events: &mut Vec<PendingRuntimeEvent>,
3351) -> StopControlApplication {
3352 let child_id = runtime_state.child_id.clone();
3353 let operation_before = runtime_state.operation;
3354 let had_active_attempt = runtime_state.has_active_attempt();
3355 let already_cancelled_for_target = had_active_attempt
3356 && operation_before == target_operation
3357 && runtime_state.attempt_cancel_delivered;
3358 let idempotent = if had_active_attempt {
3359 already_cancelled_for_target
3360 } else {
3361 operation_before == target_operation && target_operation != ChildControlOperation::Removed
3362 };
3363
3364 if operation_before != target_operation {
3365 runtime_state.operation = target_operation;
3366 pending_events.push(PendingRuntimeEvent {
3367 child_id: child_id.clone(),
3368 path: runtime_state.path.clone(),
3369 generation: runtime_state.generation,
3370 attempt: runtime_state.attempt,
3371 correlation_id,
3372 what: What::ChildControlOperationChanged {
3373 child_id: child_id.clone(),
3374 from: operation_before,
3375 to: target_operation,
3376 command: command_name.to_owned(),
3377 command_id: command_id.to_owned(),
3378 },
3379 });
3380 let _ignored = event_sender.send(format!(
3381 "child_control_operation_changed:{child_id}:{operation_before:?}:{target_operation:?}"
3382 ));
3383 }
3384
3385 let cancel_delivered = if had_active_attempt && !already_cancelled_for_target {
3386 let delivered = runtime_state.cancel();
3387 if delivered {
3388 runtime_state.stop_state = ChildStopState::CancelDelivered;
3389 runtime_state.stop_deadline_at_unix_nanos = Some(stop_deadline_at_unix_nanos);
3390 if let (Some(generation), Some(attempt)) =
3391 (runtime_state.generation, runtime_state.attempt)
3392 {
3393 pending_events.push(PendingRuntimeEvent {
3394 child_id: child_id.clone(),
3395 path: runtime_state.path.clone(),
3396 generation: Some(generation),
3397 attempt: Some(attempt),
3398 correlation_id,
3399 what: What::ChildControlCancelDelivered {
3400 child_id: child_id.clone(),
3401 generation,
3402 attempt,
3403 command: command_name.to_owned(),
3404 command_id: command_id.to_owned(),
3405 },
3406 });
3407 }
3408 let _ignored = event_sender.send(format!(
3409 "child_control_cancel_delivered:{child_id}:{command_name}"
3410 ));
3411 }
3412 delivered
3413 } else {
3414 if !had_active_attempt {
3415 runtime_state.stop_state = ChildStopState::NoActiveAttempt;
3416 }
3417 false
3418 };
3419
3420 StopControlApplication {
3421 operation_before,
3422 cancel_delivered,
3423 idempotent,
3424 remove_after_outcome: target_operation == ChildControlOperation::Removed
3425 && !had_active_attempt,
3426 }
3427}
3428
3429fn heartbeat_stale_event(
3440 runtime_state: &mut ChildSlot,
3441 liveness: &ChildLivenessState,
3442) -> Option<PendingRuntimeEvent> {
3443 let Some(attempt) = runtime_state.attempt else {
3444 runtime_state.stale_event_attempt = None;
3445 return None;
3446 };
3447 if !liveness.heartbeat_stale {
3448 runtime_state.stale_event_attempt = None;
3449 return None;
3450 }
3451 if runtime_state.stale_event_attempt == Some(attempt) {
3452 return None;
3453 }
3454 let since_unix_nanos = liveness.last_heartbeat_at_unix_nanos?;
3455 runtime_state.stale_event_attempt = Some(attempt);
3456 Some(PendingRuntimeEvent {
3457 child_id: runtime_state.child_id.clone(),
3458 path: runtime_state.path.clone(),
3459 generation: runtime_state.generation,
3460 attempt: Some(attempt),
3461 correlation_id: CorrelationId::new(),
3462 what: What::ChildHeartbeatStale {
3463 child_id: runtime_state.child_id.clone(),
3464 attempt,
3465 since_unix_nanos,
3466 },
3467 })
3468}
3469
3470pub async fn run_control_loop(
3482 mut state: RuntimeControlState,
3483 mut receiver: mpsc::Receiver<RuntimeLoopMessage>,
3484 event_sender: broadcast::Sender<String>,
3485) -> RuntimeExitReport {
3486 state.start_declared_children().await;
3487 while let Some(message) = receiver.recv().await {
3488 match message {
3489 RuntimeLoopMessage::Control {
3490 command,
3491 reply_sender,
3492 } => {
3493 let command_name = command_name(&command);
3494 let result = state.execute_control(command, &event_sender).await;
3495 let _ignored = event_sender.send(format!("control_command:{command_name}"));
3496 let _ignored = reply_sender.send(result);
3497 }
3498 RuntimeLoopMessage::ChildStart(ChildStartMessage::Exited { report }) => {
3499 state.handle_child_exit(*report, &event_sender);
3500 }
3501 RuntimeLoopMessage::ChildStart(ChildStartMessage::StartFailed {
3502 child_id,
3503 message,
3504 }) => {
3505 state.handle_child_start_failed(child_id, message, &event_sender);
3506 }
3507 RuntimeLoopMessage::ChildStart(ChildStartMessage::DelayedSpawnAttached {
3508 child_id,
3509 path,
3510 generation,
3511 attempt,
3512 handle,
3513 }) => {
3514 state.attach_spawned_child_handle(child_id, path, generation, attempt, handle);
3515 }
3516 RuntimeLoopMessage::ControlPlane(ControlPlaneMessage::ReplayChildExitForTest {
3517 report,
3518 }) => {
3519 state.handle_child_exit(*report, &event_sender);
3520 }
3521 RuntimeLoopMessage::ControlPlane(ControlPlaneMessage::Shutdown {
3522 meta,
3523 reply_sender,
3524 }) => {
3525 let _ignored = event_sender.send(format!(
3526 "runtime_control_loop_shutdown_requested:{}:{}",
3527 meta.requested_by, meta.reason
3528 ));
3529 match meta.validate() {
3530 Ok(()) => {
3531 let report = RuntimeExitReport::completed(
3532 "shutdown",
3533 format!(
3534 "runtime control plane shutdown requested: {reason}",
3535 reason = meta.reason
3536 ),
3537 );
3538 let _ignored = reply_sender.send(Ok(report.clone()));
3539 return report;
3540 }
3541 Err(error) => {
3542 let _ignored = reply_sender.send(Err(error));
3543 continue;
3544 }
3545 }
3546 }
3547 }
3548 }
3549 RuntimeExitReport::completed("message_loop", "runtime command channel closed")
3550}
3551
3552fn command_name(command: &ControlCommand) -> &'static str {
3562 match command {
3563 ControlCommand::AddChild { .. } => "add_child",
3564 ControlCommand::RemoveChild { .. } => "remove_child",
3565 ControlCommand::RestartChild { .. } => "restart_child",
3566 ControlCommand::PauseChild { .. } => "pause_child",
3567 ControlCommand::ResumeChild { .. } => "resume_child",
3568 ControlCommand::QuarantineChild { .. } => "quarantine_child",
3569 ControlCommand::ShutdownTree { .. } => "shutdown_tree",
3570 ControlCommand::CurrentState { .. } => "current_state",
3571 }
3572}
3573
3574async fn send_child_result(
3586 sender: mpsc::Sender<RuntimeLoopMessage>,
3587 child_id: ChildId,
3588 result: Result<ChildRunReport, SupervisorError>,
3589) {
3590 let message = match result {
3591 Ok(report) => RuntimeLoopMessage::ChildStart(ChildStartMessage::Exited {
3592 report: Box::new(report),
3593 }),
3594 Err(error) => RuntimeLoopMessage::ChildStart(ChildStartMessage::StartFailed {
3595 child_id,
3596 message: error.to_string(),
3597 }),
3598 };
3599 let _ignored = sender.send(message).await;
3600}
3601
3602fn restart_policy(policy: ChildRestartPolicy) -> RestartPolicy {
3612 match policy {
3613 ChildRestartPolicy::Permanent => RestartPolicy::Permanent,
3614 ChildRestartPolicy::Transient => RestartPolicy::Transient,
3615 ChildRestartPolicy::Temporary => RestartPolicy::Temporary,
3616 }
3617}
3618
3619fn backoff_policy(policy: crate::spec::child::BackoffPolicy) -> BackoffPolicy {
3633 let jitter_percent = (policy.jitter_ratio * 100.0).round().clamp(0.0, 100.0) as u8;
3634 BackoffPolicy::new(
3635 policy.initial_delay,
3636 policy.max_delay,
3637 jitter_percent,
3638 policy.max_delay,
3639 )
3640 .with_full_jitter(42) }
3642
3643fn policy_task_exit(exit: &TaskExit) -> PolicyTaskExit {
3653 match exit.failure_kind() {
3654 Some(kind) => PolicyTaskExit::Failed { kind: kind.into() },
3655 None => PolicyTaskExit::Succeeded,
3656 }
3657}
3658
3659fn classify_exit_for_pipeline(exit: &TaskExit, manual_stop_requested: bool) -> ExitClassification {
3672 match exit {
3673 TaskExit::Succeeded => ExitClassification::Success,
3674 TaskExit::Cancelled if manual_stop_requested => ExitClassification::ManualStop,
3675 TaskExit::Cancelled => ExitClassification::ExternalCancel,
3676 TaskExit::Failed(failure) => {
3677 match failure.kind {
3679 crate::error::types::TaskFailureKind::Cancelled if manual_stop_requested => {
3680 ExitClassification::ManualStop
3681 }
3682 crate::error::types::TaskFailureKind::Cancelled => {
3683 ExitClassification::ExternalCancel
3684 }
3685 crate::error::types::TaskFailureKind::Timeout => ExitClassification::Timeout,
3686 _ => ExitClassification::NonZeroExit { exit_code: -1 },
3687 }
3688 }
3689 TaskExit::Panicked(_) => ExitClassification::Crash {
3690 reason: "panic".to_string(),
3691 },
3692 TaskExit::TimedOut => ExitClassification::Timeout,
3693 }
3694}
3695
3696fn role_policy_restarts_success(pipeline_result: &PipelineContext) -> bool {
3706 pipeline_result.exit_classification == Some(ExitClassification::Success)
3707 && pipeline_result
3708 .effective_policy
3709 .as_ref()
3710 .is_some_and(|policy| policy.policy_pack.on_success_exit == OnSuccessAction::Restart)
3711}
3712
3713fn prepare_effective_policy(child_spec: &ChildSpec) -> EffectivePolicy {
3723 EffectivePolicy::for_child(child_spec)
3724}
3725
3726fn restart_limit_counts_exit(exit: &TaskExit) -> bool {
3736 matches!(
3737 exit,
3738 TaskExit::Failed(_) | TaskExit::Panicked(_) | TaskExit::TimedOut
3739 )
3740}
3741
3742fn restart_limit_for_child_in_spec(
3754 tree: &SupervisorTree,
3755 spec: &SupervisorSpec,
3756 child_id: &ChildId,
3757) -> RestartLimit {
3758 restart_execution_plan(tree, spec, child_id)
3759 .restart_limit
3760 .unwrap_or_else(default_restart_limit)
3761}
3762
3763fn default_restart_limit() -> RestartLimit {
3773 RestartLimit::new(u32::MAX, Duration::from_secs(60))
3774}
3775
3776fn restart_child_unknown_outcome(child_id: ChildId) -> ChildControlResult {
3786 let conflict = ChildControlFailure::new(
3787 ChildControlFailurePhase::WaitCompletion,
3788 "unknown child",
3789 false,
3790 );
3791 let fence = GenerationFenceOutcome::new(
3792 GenerationFenceDecision::Rejected,
3793 None,
3794 None,
3795 None,
3796 false,
3797 false,
3798 Some(conflict.clone()),
3799 );
3800 ChildControlResult::new(
3801 child_id,
3802 None,
3803 None,
3804 ChildControlOperation::Active,
3805 ChildControlOperation::Active,
3806 None,
3807 false,
3808 ChildStopState::NoActiveAttempt,
3809 RestartLimitState::default(),
3810 ChildLivenessState::new(
3811 None,
3812 false,
3813 crate::readiness::signal::ReadinessState::Unreported,
3814 ),
3815 false,
3816 Some(conflict),
3817 Some(fence),
3818 )
3819}
3820
3821fn child_control_result_label(outcome: &ChildControlResult) -> &'static str {
3831 if outcome.failure.is_some() || outcome.stop_state == ChildStopState::Failed {
3832 "failed"
3833 } else if outcome.idempotent {
3834 "idempotent"
3835 } else {
3836 "accepted"
3837 }
3838}
3839
3840fn child_scope_label(scope: &[ChildId]) -> String {
3850 scope
3851 .iter()
3852 .map(|child_id| child_id.value.clone())
3853 .collect::<Vec<_>>()
3854 .join(",")
3855}
3856
3857fn outcome_from_report(
3880 runtime_state: &ChildSlot,
3881 report: &ChildRunReport,
3882 status: ChildShutdownStatus,
3883 phase: ShutdownPhase,
3884 reason: impl Into<String>,
3885) -> ChildShutdownOutcome {
3886 ChildShutdownOutcome::new(ChildShutdownOutcomeInput {
3887 child_id: runtime_state.child_id.clone(),
3888 path: runtime_state.path.clone(),
3889 generation: runtime_state.generation.unwrap_or_else(Generation::initial),
3890 child_start_count: runtime_state.attempt.unwrap_or_else(ChildStartCount::first),
3891 status,
3892 cancel_delivered: runtime_state.attempt_cancel_delivered,
3893 exit: Some(report.exit.clone()),
3894 phase,
3895 reason: reason.into(),
3896 })
3897}
3898
3899fn removed_runtime_state_shutdown_outcome(
3910 runtime_state: &ChildSlot,
3911 phase: ShutdownPhase,
3912) -> ChildShutdownOutcome {
3913 ChildShutdownOutcome::new(ChildShutdownOutcomeInput {
3914 child_id: runtime_state.child_id.clone(),
3915 path: runtime_state.path.clone(),
3916 generation: runtime_state.generation.unwrap_or_else(Generation::initial),
3917 child_start_count: runtime_state.attempt.unwrap_or_else(ChildStartCount::first),
3918 status: ChildShutdownStatus::AlreadyExited,
3919 cancel_delivered: false,
3920 exit: None,
3921 phase,
3922 reason: "child runtime state was already removed before shutdown".to_owned(),
3923 })
3924}
3925
3926fn outcome_from_error(
3939 runtime_state: &ChildSlot,
3940 status: ChildShutdownStatus,
3941 phase: ShutdownPhase,
3942 error: SupervisorError,
3943) -> ChildShutdownOutcome {
3944 ChildShutdownOutcome::new(ChildShutdownOutcomeInput {
3945 child_id: runtime_state.child_id.clone(),
3946 path: runtime_state.path.clone(),
3947 generation: runtime_state.generation.unwrap_or_else(Generation::initial),
3948 child_start_count: runtime_state.attempt.unwrap_or_else(ChildStartCount::first),
3949 status,
3950 cancel_delivered: runtime_state.attempt_cancel_delivered,
3951 exit: None,
3952 phase,
3953 reason: error.to_string(),
3954 })
3955}
3956
3957fn remaining_duration(deadline: Instant) -> Option<Duration> {
3967 deadline.checked_duration_since(Instant::now())
3968}
3969
3970fn unix_epoch_nanos() -> u128 {
3980 SystemTime::now()
3981 .duration_since(UNIX_EPOCH)
3982 .map_or(0, |duration| duration.as_nanos())
3983}