Skip to main content

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