aster_forge_runtime/
component.rs

1//! Runtime component registration primitives.
2//!
3//! This module ties together the reusable runtime registries that already live
4//! in this crate. It lets product crates describe subsystems once, then attach
5//! health checks and shutdown phases without duplicating central dispatch
6//! tables. Product crates still own resource construction, application state
7//! assembly, business-specific startup ordering, and how reports are exposed.
8//!
9//! Component dependencies are enforced by [`RuntimeComponentRegistry`] when it
10//! runs component-owned shutdown phases. Lower-level coordinators such as
11//! [`crate::ShutdownCoordinator`] remain simple ordered executors for callers
12//! that already have a fixed sequence.
13
14use std::collections::{HashMap, HashSet};
15use std::future::Future;
16use std::pin::Pin;
17use std::time::Duration;
18
19use crate::{
20    HealthCheckDescriptor, HealthCheckOptions, HealthCheckRegistry, HealthCheckScope,
21    HealthComponentReport, ShutdownPhaseReport, ShutdownPhaseStatus, ShutdownReport,
22    StartupCoordinator, StartupPhaseFailurePolicy, StartupReport, SystemHealthReport,
23};
24
25type RuntimeShutdownFuture = Pin<Box<dyn Future<Output = Result<(), String>> + Send>>;
26type RuntimeShutdownPhaseFn = dyn FnMut() -> RuntimeShutdownFuture + Send;
27
28/// Product-owned runtime component bundle.
29///
30/// A bundle is useful when registration needs to consume owned handles such as database pools,
31/// background task collections, or other shutdown-only resources. Product subsystems should expose
32/// component factory functions that return a bundle registration instead of asking entrypoints to
33/// call low-level registry functions directly.
34pub trait RuntimeComponentBundle {
35    /// Registers this bundle into the runtime component registry.
36    fn register(self, registry: &mut RuntimeComponentRegistry);
37}
38
39impl<F> RuntimeComponentBundle for F
40where
41    F: FnOnce(&mut RuntimeComponentRegistry),
42{
43    fn register(self, registry: &mut RuntimeComponentRegistry) {
44        self(registry);
45    }
46}
47
48/// Broad category for a registered runtime component.
49#[derive(Debug, Clone, Copy, PartialEq, Eq)]
50pub enum RuntimeComponentKind {
51    /// Core process-level component.
52    Core,
53    /// Database or database connection pool component.
54    Database,
55    /// Cache component.
56    Cache,
57    /// Object storage or file storage component.
58    Storage,
59    /// Mail sender, outbox, or delivery component.
60    Mail,
61    /// Background task scheduler or worker component.
62    Tasks,
63    /// External authentication connector component.
64    ExternalAuth,
65    /// Product-specific component that does not fit another shared kind.
66    Product,
67}
68
69impl RuntimeComponentKind {
70    /// Returns a stable lowercase wire value.
71    #[must_use]
72    pub const fn as_str(self) -> &'static str {
73        match self {
74            Self::Core => "core",
75            Self::Database => "database",
76            Self::Cache => "cache",
77            Self::Storage => "storage",
78            Self::Mail => "mail",
79            Self::Tasks => "tasks",
80            Self::ExternalAuth => "external_auth",
81            Self::Product => "product",
82        }
83    }
84}
85
86/// Static shutdown metadata for a registered component.
87#[derive(Debug, Clone, Copy, PartialEq, Eq)]
88pub struct RuntimeShutdownDescriptor {
89    /// Stable shutdown phase name.
90    pub phase_name: &'static str,
91    /// Optional phase timeout.
92    pub timeout: Option<Duration>,
93}
94
95/// Static startup metadata for a registered component.
96#[derive(Debug, Clone, Copy, PartialEq, Eq)]
97pub struct RuntimeStartupDescriptor {
98    /// Stable startup phase name.
99    pub phase_name: &'static str,
100    /// Failure policy used by this startup phase.
101    pub failure_policy: StartupPhaseFailurePolicy,
102}
103
104/// Static runtime task metadata for a registered component.
105#[derive(Debug, Clone, Copy, PartialEq, Eq)]
106pub struct RuntimeTaskDescriptor {
107    /// Stable task name used in logs, persisted runtime payloads, or admin UI.
108    pub task_name: &'static str,
109    /// Operator-facing display name.
110    pub display_name: &'static str,
111}
112
113/// Static metadata for a registered runtime component.
114#[derive(Debug, Clone, PartialEq, Eq)]
115pub struct RuntimeComponentDescriptor {
116    /// Stable component name.
117    pub name: &'static str,
118    /// Broad component category.
119    pub kind: RuntimeComponentKind,
120    /// Stable names of components that should be initialized before this one.
121    pub dependencies: Vec<&'static str>,
122    /// Registered startup phases owned by this component.
123    pub startup: Vec<RuntimeStartupDescriptor>,
124    /// Registered health checks owned by this component.
125    pub health_checks: Vec<HealthCheckDescriptor>,
126    /// Registered runtime tasks owned by this component.
127    pub tasks: Vec<RuntimeTaskDescriptor>,
128    /// Registered shutdown phases owned by this component.
129    pub shutdown: Vec<RuntimeShutdownDescriptor>,
130}
131
132impl RuntimeComponentDescriptor {
133    fn new(name: &'static str) -> Self {
134        Self {
135            name,
136            kind: RuntimeComponentKind::Product,
137            dependencies: Vec::new(),
138            startup: Vec::new(),
139            health_checks: Vec::new(),
140            tasks: Vec::new(),
141            shutdown: Vec::new(),
142        }
143    }
144}
145
146/// Component dependency graph validation error.
147#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
148pub enum RuntimeComponentGraphError {
149    /// A component depends on a component that was never registered.
150    #[error("runtime component '{component}' depends on missing component '{dependency}'")]
151    MissingDependency {
152        /// Component declaring the dependency.
153        component: &'static str,
154        /// Missing dependency component name.
155        dependency: &'static str,
156    },
157    /// Component dependencies contain a cycle.
158    #[error("runtime component dependency cycle detected: {cycle}")]
159    Cycle {
160        /// Human-readable cycle path.
161        cycle: String,
162    },
163}
164
165struct RuntimeComponentShutdownPhase {
166    component_name: &'static str,
167    phase_name: &'static str,
168    timeout: Option<Duration>,
169    phase: Box<RuntimeShutdownPhaseFn>,
170}
171
172/// Registry for runtime component metadata and lifecycle hooks.
173#[derive(Default)]
174pub struct RuntimeComponentRegistry {
175    components: Vec<RuntimeComponentDescriptor>,
176    startup: StartupCoordinator,
177    health: HealthCheckRegistry,
178    shutdown: Vec<RuntimeComponentShutdownPhase>,
179}
180
181impl RuntimeComponentRegistry {
182    /// Creates an empty component registry.
183    #[must_use]
184    pub fn new() -> Self {
185        Self::default()
186    }
187
188    /// Creates a registry and applies one product or subsystem registration function.
189    pub fn configured<F>(configure: F) -> Self
190    where
191        F: FnOnce(&mut Self),
192    {
193        let mut registry = Self::new();
194        registry.configure(configure);
195        registry
196    }
197
198    /// Applies a product or subsystem registration function.
199    pub fn configure<F>(&mut self, configure: F) -> &mut Self
200    where
201        F: FnOnce(&mut Self),
202    {
203        configure(self);
204        self
205    }
206
207    /// Registers one product-owned component bundle.
208    pub fn register_bundle<B>(&mut self, bundle: B) -> &mut Self
209    where
210        B: RuntimeComponentBundle,
211    {
212        bundle.register(self);
213        self
214    }
215
216    /// Registers a component health check with explicit options.
217    pub fn component_health_with_options<F, Fut>(
218        &mut self,
219        component_name: &'static str,
220        kind: RuntimeComponentKind,
221        check_name: &'static str,
222        options: HealthCheckOptions,
223        check: F,
224    ) -> &mut Self
225    where
226        F: Fn() -> Fut + Send + Sync + 'static,
227        Fut: Future<Output = HealthComponentReport> + Send + 'static,
228    {
229        self.component(component_name)
230            .kind(kind)
231            .health_with_options(check_name, options, check);
232        self
233    }
234
235    /// Registers a component startup phase.
236    pub fn component_startup<F, Fut>(
237        &mut self,
238        component_name: &'static str,
239        kind: RuntimeComponentKind,
240        phase_name: &'static str,
241        failure_policy: StartupPhaseFailurePolicy,
242        phase: F,
243    ) -> &mut Self
244    where
245        F: FnMut() -> Fut + Send + 'static,
246        Fut: Future<Output = Result<(), String>> + Send + 'static,
247    {
248        self.component(component_name)
249            .kind(kind)
250            .startup(phase_name, failure_policy, phase);
251        self
252    }
253
254    /// Registers a component-owned runtime task descriptor.
255    pub fn component_task(
256        &mut self,
257        component_name: &'static str,
258        kind: RuntimeComponentKind,
259        task_name: &'static str,
260        display_name: &'static str,
261    ) -> &mut Self {
262        self.component(component_name)
263            .kind(kind)
264            .task(task_name, display_name);
265        self
266    }
267
268    /// Registers a component shutdown phase.
269    pub fn component_shutdown<F, Fut>(
270        &mut self,
271        component_name: &'static str,
272        kind: RuntimeComponentKind,
273        phase_name: &'static str,
274        timeout: Option<Duration>,
275        phase: F,
276    ) -> &mut Self
277    where
278        F: FnMut() -> Fut + Send + 'static,
279        Fut: Future<Output = Result<(), String>> + Send + 'static,
280    {
281        self.component(component_name)
282            .kind(kind)
283            .shutdown(phase_name, timeout, phase);
284        self
285    }
286
287    /// Registers a component shutdown phase that consumes one owned value at most once.
288    pub fn component_shutdown_once<T, F, Fut>(
289        &mut self,
290        component_name: &'static str,
291        kind: RuntimeComponentKind,
292        phase_name: &'static str,
293        timeout: Option<Duration>,
294        value: T,
295        phase: F,
296    ) -> &mut Self
297    where
298        T: Send + 'static,
299        F: FnOnce(T) -> Fut + Send + 'static,
300        Fut: Future<Output = Result<(), String>> + Send + 'static,
301    {
302        self.component(component_name)
303            .kind(kind)
304            .shutdown_once(phase_name, timeout, value, phase);
305        self
306    }
307
308    /// Returns a builder for `name`, creating the component when needed.
309    pub fn component(&mut self, name: &'static str) -> RuntimeComponentBuilder<'_> {
310        let index = if let Some(index) = self
311            .components
312            .iter()
313            .position(|component| component.name == name)
314        {
315            index
316        } else {
317            self.components.push(RuntimeComponentDescriptor::new(name));
318            self.components.len() - 1
319        };
320
321        RuntimeComponentBuilder {
322            registry: self,
323            index,
324        }
325    }
326
327    /// Returns registered component descriptors in registration order.
328    #[must_use]
329    pub fn descriptors(&self) -> &[RuntimeComponentDescriptor] {
330        &self.components
331    }
332
333    /// Returns one descriptor by component name.
334    #[must_use]
335    pub fn descriptor(&self, name: &str) -> Option<&RuntimeComponentDescriptor> {
336        self.components
337            .iter()
338            .find(|component| component.name == name)
339    }
340
341    /// Returns the underlying health registry.
342    #[must_use]
343    pub const fn health_registry(&self) -> &HealthCheckRegistry {
344        &self.health
345    }
346
347    /// Returns the underlying health registry mutably.
348    pub const fn health_registry_mut(&mut self) -> &mut HealthCheckRegistry {
349        &mut self.health
350    }
351
352    /// Runs health checks registered for `scope`.
353    pub async fn run_health(&mut self, scope: HealthCheckScope) -> SystemHealthReport {
354        self.health.run_scope(scope).await
355    }
356
357    /// Runs registered startup phases.
358    pub async fn startup(&mut self) -> StartupReport {
359        self.startup.run().await
360    }
361
362    /// Validates that the component dependency graph is resolvable.
363    ///
364    /// # Errors
365    ///
366    /// Returns an error when a dependency is missing or the component graph contains a cycle.
367    pub fn validate(&self) -> Result<(), RuntimeComponentGraphError> {
368        let descriptor_by_name = self
369            .components
370            .iter()
371            .map(|component| (component.name, component))
372            .collect::<HashMap<_, _>>();
373
374        for component in &self.components {
375            for dependency in &component.dependencies {
376                if !descriptor_by_name.contains_key(dependency) {
377                    return Err(RuntimeComponentGraphError::MissingDependency {
378                        component: component.name,
379                        dependency,
380                    });
381                }
382            }
383        }
384
385        let mut visiting = Vec::new();
386        let mut visited = HashSet::new();
387        for component in &self.components {
388            Self::validate_component_dependencies(
389                component.name,
390                &descriptor_by_name,
391                &mut visiting,
392                &mut visited,
393            )?;
394        }
395
396        Ok(())
397    }
398
399    fn validate_component_dependencies(
400        component_name: &'static str,
401        descriptor_by_name: &HashMap<&'static str, &RuntimeComponentDescriptor>,
402        visiting: &mut Vec<&'static str>,
403        visited: &mut HashSet<&'static str>,
404    ) -> Result<(), RuntimeComponentGraphError> {
405        if visited.contains(component_name) {
406            return Ok(());
407        }
408        if let Some(position) = visiting
409            .iter()
410            .position(|visiting_name| *visiting_name == component_name)
411        {
412            let mut cycle = visiting[position..].to_vec();
413            cycle.push(component_name);
414            return Err(RuntimeComponentGraphError::Cycle {
415                cycle: cycle.join(" -> "),
416            });
417        }
418
419        visiting.push(component_name);
420        if let Some(descriptor) = descriptor_by_name.get(component_name) {
421            for dependency in &descriptor.dependencies {
422                Self::validate_component_dependencies(
423                    dependency,
424                    descriptor_by_name,
425                    visiting,
426                    visited,
427                )?;
428            }
429        }
430        visiting.pop();
431        visited.insert(component_name);
432        Ok(())
433    }
434
435    /// Runs registered shutdown phases in component dependency order.
436    ///
437    /// A component's dependencies run before that component when both sides
438    /// have shutdown phases. Multiple phases registered by one component run in
439    /// registration order. Dependencies without shutdown phases are kept as
440    /// descriptor metadata and do not block execution. Cycles are reported as
441    /// warnings and the registry still makes best-effort progress without
442    /// executing a phase more than once.
443    pub async fn shutdown(&mut self) -> ShutdownReport {
444        let mut reports = Vec::with_capacity(self.shutdown.len());
445        for index in self.shutdown_order() {
446            let registered = &mut self.shutdown[index];
447            tracing::info!(phase = registered.phase_name, "starting shutdown phase");
448            let started_at = std::time::Instant::now();
449            let future = (registered.phase)();
450            let status = match registered.timeout {
451                Some(timeout) => match tokio::time::timeout(timeout, future).await {
452                    Ok(Ok(())) => ShutdownPhaseStatus::Succeeded,
453                    Ok(Err(error)) => ShutdownPhaseStatus::Failed(error),
454                    Err(_) => ShutdownPhaseStatus::TimedOut,
455                },
456                None => match future.await {
457                    Ok(()) => ShutdownPhaseStatus::Succeeded,
458                    Err(error) => ShutdownPhaseStatus::Failed(error),
459                },
460            };
461            let duration = started_at.elapsed();
462            match &status {
463                ShutdownPhaseStatus::Succeeded => {
464                    tracing::info!(
465                        phase = registered.phase_name,
466                        ?duration,
467                        "shutdown phase completed"
468                    );
469                }
470                ShutdownPhaseStatus::Failed(error) => {
471                    tracing::error!(
472                        phase = registered.phase_name,
473                        ?duration,
474                        %error,
475                        "shutdown phase failed"
476                    );
477                }
478                ShutdownPhaseStatus::TimedOut => {
479                    tracing::error!(
480                        phase = registered.phase_name,
481                        ?duration,
482                        "shutdown phase timed out"
483                    );
484                }
485            }
486            reports.push(ShutdownPhaseReport {
487                name: registered.phase_name,
488                status,
489                duration,
490            });
491        }
492
493        ShutdownReport::new(reports)
494    }
495
496    fn shutdown_order(&self) -> Vec<usize> {
497        let mut phase_indices_by_component: HashMap<&'static str, Vec<usize>> = HashMap::new();
498        for (index, phase) in self.shutdown.iter().enumerate() {
499            phase_indices_by_component
500                .entry(phase.component_name)
501                .or_default()
502                .push(index);
503        }
504        let descriptor_by_name = self
505            .components
506            .iter()
507            .map(|component| (component.name, component))
508            .collect::<HashMap<_, _>>();
509        let mut visiting = HashSet::new();
510        let mut visited = HashSet::new();
511        let mut ordered = Vec::with_capacity(self.shutdown.len());
512
513        for phase in &self.shutdown {
514            Self::push_shutdown_component_order(
515                phase.component_name,
516                &phase_indices_by_component,
517                &descriptor_by_name,
518                &mut visiting,
519                &mut visited,
520                &mut ordered,
521            );
522        }
523
524        ordered
525    }
526
527    fn push_shutdown_component_order(
528        component_name: &'static str,
529        phase_indices_by_component: &HashMap<&'static str, Vec<usize>>,
530        descriptor_by_name: &HashMap<&'static str, &RuntimeComponentDescriptor>,
531        visiting: &mut HashSet<&'static str>,
532        visited: &mut HashSet<&'static str>,
533        ordered: &mut Vec<usize>,
534    ) {
535        if visited.contains(component_name) {
536            return;
537        }
538        if !visiting.insert(component_name) {
539            tracing::warn!(
540                component = component_name,
541                "runtime component dependency cycle detected during shutdown ordering"
542            );
543            return;
544        }
545
546        if let Some(descriptor) = descriptor_by_name.get(component_name) {
547            for dependency in &descriptor.dependencies {
548                if phase_indices_by_component.contains_key(dependency) {
549                    Self::push_shutdown_component_order(
550                        dependency,
551                        phase_indices_by_component,
552                        descriptor_by_name,
553                        visiting,
554                        visited,
555                        ordered,
556                    );
557                }
558            }
559        }
560
561        visiting.remove(component_name);
562        visited.insert(component_name);
563        // Dependencies are per-component, so the DFS stays per-component; when a
564        // component is emitted, every phase it registered runs in registration order.
565        if let Some(indices) = phase_indices_by_component.get(component_name) {
566            ordered.extend(indices);
567        }
568    }
569
570    /// Returns how many components are registered.
571    #[must_use]
572    pub fn len(&self) -> usize {
573        self.components.len()
574    }
575
576    /// Returns whether no components are registered.
577    #[must_use]
578    pub fn is_empty(&self) -> bool {
579        self.components.is_empty()
580    }
581}
582
583/// Builder for one runtime component registration.
584pub struct RuntimeComponentBuilder<'a> {
585    registry: &'a mut RuntimeComponentRegistry,
586    index: usize,
587}
588
589impl RuntimeComponentBuilder<'_> {
590    /// Sets the component category.
591    pub fn kind(&mut self, kind: RuntimeComponentKind) -> &mut Self {
592        self.descriptor_mut().kind = kind;
593        self
594    }
595
596    /// Adds one component dependency.
597    pub fn depends_on(&mut self, dependency: &'static str) -> &mut Self {
598        let descriptor = self.descriptor_mut();
599        if !descriptor.dependencies.contains(&dependency) {
600            descriptor.dependencies.push(dependency);
601        }
602        self
603    }
604
605    /// Adds component dependencies in order.
606    pub fn depends_on_all(&mut self, dependencies: &[&'static str]) -> &mut Self {
607        for dependency in dependencies {
608            self.depends_on(dependency);
609        }
610        self
611    }
612
613    /// Registers a component startup phase.
614    pub fn startup<F, Fut>(
615        &mut self,
616        phase_name: &'static str,
617        failure_policy: StartupPhaseFailurePolicy,
618        phase: F,
619    ) -> &mut Self
620    where
621        F: FnMut() -> Fut + Send + 'static,
622        Fut: Future<Output = Result<(), String>> + Send + 'static,
623    {
624        self.registry
625            .startup
626            .phase(phase_name, failure_policy, phase);
627        self.descriptor_mut()
628            .startup
629            .push(RuntimeStartupDescriptor {
630                phase_name,
631                failure_policy,
632            });
633        self
634    }
635
636    /// Registers a component-owned runtime task descriptor.
637    pub fn task(&mut self, task_name: &'static str, display_name: &'static str) -> &mut Self {
638        self.descriptor_mut().tasks.push(RuntimeTaskDescriptor {
639            task_name,
640            display_name,
641        });
642        self
643    }
644
645    /// Registers a component health check with explicit options.
646    pub fn health_with_options<F, Fut>(
647        &mut self,
648        check_name: &'static str,
649        options: HealthCheckOptions,
650        check: F,
651    ) -> &mut Self
652    where
653        F: Fn() -> Fut + Send + Sync + 'static,
654        Fut: Future<Output = HealthComponentReport> + Send + 'static,
655    {
656        self.registry
657            .health
658            .register_with_options(check_name, options, check);
659        self.descriptor_mut()
660            .health_checks
661            .push(HealthCheckDescriptor {
662                name: check_name,
663                requirement: options.requirement,
664                timeout: options.timeout,
665                scopes: options.scopes,
666            });
667        self
668    }
669
670    /// Registers a component shutdown phase.
671    ///
672    /// A component may register multiple shutdown phases; they run in registration
673    /// order, after the shutdown phases of the component's dependencies. Registering
674    /// the same phase name twice for one component is almost always a duplicated
675    /// registration accident, so it is reported with a warning, but every registered
676    /// phase still runs.
677    pub fn shutdown<F, Fut>(
678        &mut self,
679        phase_name: &'static str,
680        timeout: Option<Duration>,
681        mut phase: F,
682    ) -> &mut Self
683    where
684        F: FnMut() -> Fut + Send + 'static,
685        Fut: Future<Output = Result<(), String>> + Send + 'static,
686    {
687        let component_name = self.descriptor_mut().name;
688        if self.registry.shutdown.iter().any(|registered| {
689            registered.component_name == component_name && registered.phase_name == phase_name
690        }) {
691            tracing::warn!(
692                component = component_name,
693                phase = phase_name,
694                "duplicate shutdown phase registration"
695            );
696        }
697        self.registry.shutdown.push(RuntimeComponentShutdownPhase {
698            component_name,
699            phase_name,
700            timeout,
701            phase: Box::new(move || Box::pin(phase())),
702        });
703        self.descriptor_mut()
704            .shutdown
705            .push(RuntimeShutdownDescriptor {
706                phase_name,
707                timeout,
708            });
709        self
710    }
711
712    /// Registers a shutdown phase that consumes one owned value at most once.
713    ///
714    /// This is the common shape for database handles, background task sets,
715    /// and other shutdown-only resources. Re-running the registry after the
716    /// value has already been consumed becomes a no-op success.
717    pub fn shutdown_once<T, F, Fut>(
718        &mut self,
719        phase_name: &'static str,
720        timeout: Option<Duration>,
721        value: T,
722        phase: F,
723    ) -> &mut Self
724    where
725        T: Send + 'static,
726        F: FnOnce(T) -> Fut + Send + 'static,
727        Fut: Future<Output = Result<(), String>> + Send + 'static,
728    {
729        let mut value = Some(value);
730        let mut phase = Some(phase);
731        self.shutdown(phase_name, timeout, move || {
732            let value = value.take();
733            let phase = phase.take();
734            async move {
735                if let (Some(value), Some(phase)) = (value, phase) {
736                    phase(value).await?;
737                }
738                Ok(())
739            }
740        })
741    }
742
743    fn descriptor_mut(&mut self) -> &mut RuntimeComponentDescriptor {
744        &mut self.registry.components[self.index]
745    }
746}
747
748#[cfg(test)]
749mod tests {
750    use std::sync::{Arc, Mutex};
751    use std::time::Duration;
752
753    use super::{RuntimeComponentBundle, RuntimeComponentKind, RuntimeComponentRegistry};
754    use crate::{
755        HealthCheckOptions, HealthCheckScope, HealthCheckScopes, HealthComponentReport,
756        HealthStatus, ShutdownPhaseStatus, StartupPhaseFailurePolicy,
757    };
758
759    #[tokio::test]
760    async fn registry_runs_component_health_checks_by_scope() {
761        let mut registry = RuntimeComponentRegistry::new();
762        registry.component_health_with_options(
763            "database",
764            RuntimeComponentKind::Database,
765            "database",
766            HealthCheckOptions::required(Some(Duration::from_secs(1)))
767                .with_scopes(HealthCheckScopes::readiness_and_diagnostics()),
768            || async { HealthComponentReport::healthy("database", "ok") },
769        );
770        registry.component_health_with_options(
771            "cache",
772            RuntimeComponentKind::Cache,
773            "cache",
774            HealthCheckOptions::optional(None).with_scopes(HealthCheckScopes::diagnostics()),
775            || async { HealthComponentReport::degraded("cache", "fallback") },
776        );
777
778        let readiness = registry.run_health(HealthCheckScope::Readiness).await;
779        let diagnostics = registry.run_health(HealthCheckScope::Diagnostics).await;
780
781        assert_eq!(readiness.components.len(), 1);
782        assert_eq!(readiness.components[0].name, "database");
783        assert_eq!(diagnostics.components.len(), 2);
784        assert_eq!(diagnostics.status(), HealthStatus::Degraded);
785        assert_eq!(registry.descriptors()[0].health_checks[0].name, "database");
786        assert_eq!(registry.descriptors()[1].health_checks[0].name, "cache");
787    }
788
789    #[tokio::test]
790    async fn registry_runs_component_startup_and_records_task_descriptors() {
791        let startup_events = Arc::new(Mutex::new(Vec::new()));
792        let mut registry = RuntimeComponentRegistry::new();
793        registry
794            .component("mail")
795            .kind(RuntimeComponentKind::Mail)
796            .startup("mail_templates", StartupPhaseFailurePolicy::Required, {
797                let startup_events = Arc::clone(&startup_events);
798                move || {
799                    let startup_events = Arc::clone(&startup_events);
800                    async move {
801                        startup_events.lock().unwrap().push("mail_templates");
802                        Ok(())
803                    }
804                }
805            })
806            .task("mail-outbox-dispatch", "Mail outbox dispatch");
807
808        let report = registry.startup().await;
809
810        assert!(!report.aborted());
811        assert_eq!(
812            startup_events.lock().unwrap().as_slice(),
813            ["mail_templates"]
814        );
815        let descriptor = registry.descriptor("mail").expect("mail descriptor");
816        assert_eq!(descriptor.startup[0].phase_name, "mail_templates");
817        assert_eq!(
818            descriptor.startup[0].failure_policy,
819            StartupPhaseFailurePolicy::Required
820        );
821        assert_eq!(descriptor.tasks[0].task_name, "mail-outbox-dispatch");
822        assert_eq!(descriptor.tasks[0].display_name, "Mail outbox dispatch");
823    }
824
825    fn register_database_component(registry: &mut RuntimeComponentRegistry) {
826        registry.component_health_with_options(
827            "database",
828            RuntimeComponentKind::Database,
829            "database",
830            HealthCheckOptions::required(None),
831            || async { HealthComponentReport::healthy("database", "ok") },
832        );
833    }
834
835    fn register_cache_component(registry: &mut RuntimeComponentRegistry) {
836        registry
837            .component("cache")
838            .kind(RuntimeComponentKind::Cache)
839            .depends_on("database")
840            .health_with_options(
841                "cache",
842                HealthCheckOptions::optional(None).with_scopes(HealthCheckScopes::diagnostics()),
843                || async { HealthComponentReport::healthy("cache", "ok") },
844            );
845    }
846
847    #[tokio::test]
848    async fn registry_runs_shutdown_phases_in_dependency_order() {
849        let order = Arc::new(Mutex::new(Vec::new()));
850        let mut registry = RuntimeComponentRegistry::new();
851
852        registry
853            .component("database")
854            .kind(RuntimeComponentKind::Database)
855            .depends_on_all(&["tasks"])
856            .shutdown("database", None, {
857                let order = Arc::clone(&order);
858                move || {
859                    let order = Arc::clone(&order);
860                    async move {
861                        order.lock().unwrap().push("database");
862                        Err("close failed".to_string())
863                    }
864                }
865            });
866        registry.component_shutdown("tasks", RuntimeComponentKind::Tasks, "tasks", None, {
867            let order = Arc::clone(&order);
868            move || {
869                let order = Arc::clone(&order);
870                async move {
871                    order.lock().unwrap().push("tasks");
872                    Ok(())
873                }
874            }
875        });
876
877        let report = registry.shutdown().await;
878
879        assert_eq!(order.lock().unwrap().as_slice(), ["tasks", "database"]);
880        assert!(report.has_failures());
881        assert_eq!(report.phases[0].status, ShutdownPhaseStatus::Succeeded);
882        assert_eq!(
883            report.phases[1].status,
884            ShutdownPhaseStatus::Failed("close failed".to_string())
885        );
886        assert_eq!(
887            registry
888                .descriptor("database")
889                .expect("database component should exist")
890                .dependencies,
891            vec!["tasks"]
892        );
893    }
894
895    #[tokio::test]
896    async fn registry_runs_deep_shutdown_graph_before_dependents() {
897        let order = Arc::new(Mutex::new(Vec::new()));
898        let mut registry = RuntimeComponentRegistry::new();
899
900        for (component, kind, dependencies) in [
901            (
902                "database",
903                RuntimeComponentKind::Database,
904                &["audit_manager"][..],
905            ),
906            (
907                "audit_manager",
908                RuntimeComponentKind::Product,
909                &["audit_logs"][..],
910            ),
911            (
912                "audit_logs",
913                RuntimeComponentKind::Product,
914                &["mail_outbox"][..],
915            ),
916            (
917                "mail_outbox",
918                RuntimeComponentKind::Mail,
919                &["background_tasks"][..],
920            ),
921            ("background_tasks", RuntimeComponentKind::Tasks, &[][..]),
922        ] {
923            registry
924                .component(component)
925                .kind(kind)
926                .depends_on_all(dependencies)
927                .shutdown(component, None, {
928                    let order = Arc::clone(&order);
929                    move || {
930                        let order = Arc::clone(&order);
931                        async move {
932                            order.lock().unwrap().push(component);
933                            Ok(())
934                        }
935                    }
936                });
937        }
938
939        let report = registry.shutdown().await;
940
941        assert!(!report.has_failures());
942        assert_eq!(
943            order.lock().unwrap().as_slice(),
944            [
945                "background_tasks",
946                "mail_outbox",
947                "audit_logs",
948                "audit_manager",
949                "database"
950            ]
951        );
952        assert_eq!(
953            report
954                .phases
955                .iter()
956                .map(|phase| phase.name)
957                .collect::<Vec<_>>(),
958            vec![
959                "background_tasks",
960                "mail_outbox",
961                "audit_logs",
962                "audit_manager",
963                "database"
964            ]
965        );
966    }
967
968    #[tokio::test]
969    async fn registry_runs_all_shutdown_phases_of_one_component_in_registration_order() {
970        let order = Arc::new(Mutex::new(Vec::new()));
971        let mut registry = RuntimeComponentRegistry::new();
972
973        {
974            let mut component = registry.component("audit");
975            for phase_name in ["audit_stop", "audit_flush", "audit_close"] {
976                component.shutdown(phase_name, None, {
977                    let order = Arc::clone(&order);
978                    move || {
979                        let order = Arc::clone(&order);
980                        async move {
981                            order.lock().unwrap().push(phase_name);
982                            Ok(())
983                        }
984                    }
985                });
986            }
987        }
988
989        let report = registry.shutdown().await;
990
991        assert!(!report.has_failures());
992        assert_eq!(
993            order.lock().unwrap().as_slice(),
994            ["audit_stop", "audit_flush", "audit_close"]
995        );
996        assert_eq!(
997            report
998                .phases
999                .iter()
1000                .map(|phase| phase.name)
1001                .collect::<Vec<_>>(),
1002            vec!["audit_stop", "audit_flush", "audit_close"]
1003        );
1004        assert_eq!(
1005            registry
1006                .descriptor("audit")
1007                .expect("audit component should exist")
1008                .shutdown
1009                .iter()
1010                .map(|descriptor| descriptor.phase_name)
1011                .collect::<Vec<_>>(),
1012            vec!["audit_stop", "audit_flush", "audit_close"]
1013        );
1014    }
1015
1016    #[tokio::test]
1017    async fn registry_runs_dependency_phases_before_all_dependent_phases() {
1018        let order = Arc::new(Mutex::new(Vec::new()));
1019        let mut registry = RuntimeComponentRegistry::new();
1020
1021        {
1022            let mut database = registry.component("database");
1023            for phase_name in ["database_stop", "database_close"] {
1024                database.shutdown(phase_name, None, {
1025                    let order = Arc::clone(&order);
1026                    move || {
1027                        let order = Arc::clone(&order);
1028                        async move {
1029                            order.lock().unwrap().push(phase_name);
1030                            Ok(())
1031                        }
1032                    }
1033                });
1034            }
1035        }
1036        {
1037            let mut audit = registry.component("audit");
1038            audit.depends_on("database");
1039            for phase_name in ["audit_flush", "audit_close"] {
1040                audit.shutdown(phase_name, None, {
1041                    let order = Arc::clone(&order);
1042                    move || {
1043                        let order = Arc::clone(&order);
1044                        async move {
1045                            order.lock().unwrap().push(phase_name);
1046                            Ok(())
1047                        }
1048                    }
1049                });
1050            }
1051        }
1052
1053        let report = registry.shutdown().await;
1054
1055        assert!(!report.has_failures());
1056        assert_eq!(
1057            order.lock().unwrap().as_slice(),
1058            [
1059                "database_stop",
1060                "database_close",
1061                "audit_flush",
1062                "audit_close"
1063            ]
1064        );
1065    }
1066
1067    #[derive(Clone)]
1068    struct SharedLogBuffer(Arc<Mutex<Vec<u8>>>);
1069
1070    impl std::io::Write for SharedLogBuffer {
1071        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
1072            self.0.lock().unwrap().extend_from_slice(buf);
1073            Ok(buf.len())
1074        }
1075
1076        fn flush(&mut self) -> std::io::Result<()> {
1077            Ok(())
1078        }
1079    }
1080
1081    #[tokio::test]
1082    async fn duplicate_shutdown_phase_registration_warns_and_still_runs_both() {
1083        let order = Arc::new(Mutex::new(Vec::new()));
1084        let buffer = SharedLogBuffer(Arc::new(Mutex::new(Vec::new())));
1085        let writer = buffer.clone();
1086        let subscriber = tracing_subscriber::fmt()
1087            .with_writer(move || writer.clone())
1088            .with_ansi(false)
1089            .without_time()
1090            .finish();
1091
1092        let mut registry = RuntimeComponentRegistry::new();
1093        tracing::subscriber::with_default(subscriber, || {
1094            let mut component = registry.component("audit");
1095            for run in [1_u8, 2] {
1096                component.shutdown("audit_flush", None, {
1097                    let order = Arc::clone(&order);
1098                    move || {
1099                        let order = Arc::clone(&order);
1100                        async move {
1101                            order.lock().unwrap().push(run);
1102                            Ok(())
1103                        }
1104                    }
1105                });
1106            }
1107        });
1108
1109        let logs = String::from_utf8(buffer.0.lock().unwrap().clone())
1110            .expect("log output should be valid UTF-8");
1111        assert!(
1112            logs.contains("duplicate shutdown phase registration"),
1113            "expected duplicate registration warning, got: {logs}"
1114        );
1115
1116        let report = registry.shutdown().await;
1117        assert!(!report.has_failures());
1118        assert_eq!(order.lock().unwrap().as_slice(), [1, 2]);
1119        assert_eq!(report.phases.len(), 2);
1120    }
1121
1122    #[tokio::test]
1123    async fn registry_ignores_shutdown_dependencies_without_shutdown_phase() {
1124        let order = Arc::new(Mutex::new(Vec::new()));
1125        let mut registry = RuntimeComponentRegistry::new();
1126
1127        registry
1128            .component("cache")
1129            .kind(RuntimeComponentKind::Cache);
1130        registry
1131            .component("database")
1132            .kind(RuntimeComponentKind::Database)
1133            .depends_on("cache")
1134            .shutdown("database", None, {
1135                let order = Arc::clone(&order);
1136                move || {
1137                    let order = Arc::clone(&order);
1138                    async move {
1139                        order.lock().unwrap().push("database");
1140                        Ok(())
1141                    }
1142                }
1143            });
1144
1145        let report = registry.shutdown().await;
1146
1147        assert!(!report.has_failures());
1148        assert_eq!(order.lock().unwrap().as_slice(), ["database"]);
1149        assert_eq!(report.phases.len(), 1);
1150        assert_eq!(report.phases[0].name, "database");
1151    }
1152
1153    #[test]
1154    fn registry_validate_rejects_missing_component_dependencies() {
1155        let mut registry = RuntimeComponentRegistry::new();
1156        registry
1157            .component("database")
1158            .kind(RuntimeComponentKind::Database)
1159            .depends_on("cache");
1160
1161        let error = registry
1162            .validate()
1163            .expect_err("missing dependency should fail validation");
1164
1165        assert_eq!(
1166            error,
1167            crate::RuntimeComponentGraphError::MissingDependency {
1168                component: "database",
1169                dependency: "cache"
1170            }
1171        );
1172    }
1173
1174    #[test]
1175    fn registry_validate_rejects_dependency_cycles() {
1176        let mut registry = RuntimeComponentRegistry::new();
1177        registry.component("database").depends_on("audit");
1178        registry.component("audit").depends_on("database");
1179
1180        let error = registry
1181            .validate()
1182            .expect_err("dependency cycle should fail validation");
1183
1184        assert_eq!(
1185            error,
1186            crate::RuntimeComponentGraphError::Cycle {
1187                cycle: "database -> audit -> database".to_string()
1188            }
1189        );
1190    }
1191
1192    #[tokio::test]
1193    async fn registry_shutdown_dependency_cycle_does_not_repeat_phases() {
1194        let order = Arc::new(Mutex::new(Vec::new()));
1195        let mut registry = RuntimeComponentRegistry::new();
1196
1197        for (component, dependency) in [("database", "audit"), ("audit", "database")] {
1198            registry
1199                .component(component)
1200                .kind(RuntimeComponentKind::Product)
1201                .depends_on(dependency)
1202                .shutdown(component, None, {
1203                    let order = Arc::clone(&order);
1204                    move || {
1205                        let order = Arc::clone(&order);
1206                        async move {
1207                            order.lock().unwrap().push(component);
1208                            Ok(())
1209                        }
1210                    }
1211                });
1212        }
1213
1214        let report = registry.shutdown().await;
1215        let order = order.lock().unwrap();
1216
1217        assert!(!report.has_failures());
1218        assert_eq!(order.len(), 2);
1219        assert!(order.contains(&"database"));
1220        assert!(order.contains(&"audit"));
1221        assert_eq!(report.phases.len(), 2);
1222    }
1223
1224    #[tokio::test]
1225    async fn registry_shutdown_once_consumes_owned_value_once() {
1226        let values = Arc::new(Mutex::new(Vec::new()));
1227        let mut registry = RuntimeComponentRegistry::new();
1228
1229        registry.component_shutdown_once(
1230            "database",
1231            RuntimeComponentKind::Database,
1232            "database",
1233            None,
1234            "writer",
1235            {
1236                let values = Arc::clone(&values);
1237                move |value| async move {
1238                    values.lock().unwrap().push(value);
1239                    Ok(())
1240                }
1241            },
1242        );
1243
1244        let first = registry.shutdown().await;
1245        let second = registry.shutdown().await;
1246
1247        assert!(!first.has_failures());
1248        assert!(!second.has_failures());
1249        assert_eq!(values.lock().unwrap().as_slice(), ["writer"]);
1250        assert_eq!(
1251            registry
1252                .descriptor("database")
1253                .map(|descriptor| descriptor.kind),
1254            Some(RuntimeComponentKind::Database)
1255        );
1256    }
1257
1258    struct TestShutdownBundle {
1259        values: Arc<Mutex<Vec<&'static str>>>,
1260    }
1261
1262    impl RuntimeComponentBundle for TestShutdownBundle {
1263        fn register(self, registry: &mut RuntimeComponentRegistry) {
1264            registry.component_shutdown("audit", RuntimeComponentKind::Product, "audit", None, {
1265                let values = Arc::clone(&self.values);
1266                move || {
1267                    let values = Arc::clone(&values);
1268                    async move {
1269                        values.lock().unwrap().push("audit");
1270                        Ok(())
1271                    }
1272                }
1273            });
1274            registry
1275                .component("database")
1276                .kind(RuntimeComponentKind::Database)
1277                .depends_on("audit")
1278                .shutdown_once("database", None, self.values, |values| async move {
1279                    values.lock().unwrap().push("database");
1280                    Ok(())
1281                });
1282        }
1283    }
1284
1285    #[tokio::test]
1286    async fn registry_accepts_owned_component_bundle() {
1287        let values = Arc::new(Mutex::new(Vec::new()));
1288        let mut registry = RuntimeComponentRegistry::new();
1289        registry.register_bundle(TestShutdownBundle {
1290            values: Arc::clone(&values),
1291        });
1292
1293        assert_eq!(registry.len(), 2);
1294        assert_eq!(
1295            registry
1296                .descriptor("database")
1297                .expect("database descriptor should exist")
1298                .dependencies,
1299            vec!["audit"]
1300        );
1301
1302        let report = registry.shutdown().await;
1303
1304        assert!(!report.has_failures());
1305        assert_eq!(values.lock().unwrap().as_slice(), ["audit", "database"]);
1306    }
1307
1308    #[test]
1309    fn registry_accepts_closure_component_bundle() {
1310        let mut registry = RuntimeComponentRegistry::new();
1311        registry.register_bundle(|registry: &mut RuntimeComponentRegistry| {
1312            register_database_component(registry);
1313        });
1314
1315        assert_eq!(registry.len(), 1);
1316        assert_eq!(registry.descriptors()[0].name, "database");
1317    }
1318
1319    #[test]
1320    fn registry_register_bundle_chains_multiple_component_bundles() {
1321        let mut registry = RuntimeComponentRegistry::new();
1322        registry
1323            .register_bundle(register_database_component)
1324            .register_bundle(register_cache_component);
1325
1326        assert_eq!(registry.len(), 2);
1327        assert_eq!(registry.descriptors()[0].name, "database");
1328        assert_eq!(registry.descriptors()[1].name, "cache");
1329    }
1330
1331    #[tokio::test]
1332    async fn registry_can_shutdown_registered_component_bundle() {
1333        let values = Arc::new(Mutex::new(Vec::new()));
1334
1335        let mut registry = RuntimeComponentRegistry::new();
1336        registry.register_bundle(TestShutdownBundle {
1337            values: Arc::clone(&values),
1338        });
1339        let report = registry.shutdown().await;
1340
1341        assert!(!report.has_failures());
1342        assert_eq!(values.lock().unwrap().as_slice(), ["audit", "database"]);
1343    }
1344}