1use 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
29pub trait RuntimeComponentBundle {
36 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#[derive(Debug, Clone, Copy, PartialEq, Eq)]
51pub enum RuntimeComponentKind {
52 Core,
54 Database,
56 Cache,
58 Storage,
60 Mail,
62 Tasks,
64 ExternalAuth,
66 Product,
68}
69
70impl RuntimeComponentKind {
71 #[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#[derive(Debug, Clone, Copy, PartialEq, Eq)]
89pub struct RuntimeShutdownDescriptor {
90 pub phase_name: &'static str,
92 pub timeout: Option<Duration>,
94}
95
96#[derive(Debug, Clone, Copy, PartialEq, Eq)]
98pub struct RuntimeStartupDescriptor {
99 pub phase_name: &'static str,
101 pub failure_policy: StartupPhaseFailurePolicy,
103}
104
105#[derive(Debug, Clone, Copy, PartialEq, Eq)]
107pub struct RuntimeTaskDescriptor {
108 pub task_name: &'static str,
110 pub display_name: &'static str,
112}
113
114#[derive(Debug, Clone, PartialEq, Eq)]
116pub struct RuntimeComponentDescriptor {
117 pub name: &'static str,
119 pub kind: RuntimeComponentKind,
121 pub dependencies: Vec<&'static str>,
123 pub startup: Vec<RuntimeStartupDescriptor>,
125 pub health_checks: Vec<HealthCheckDescriptor>,
127 pub tasks: Vec<RuntimeTaskDescriptor>,
129 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#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
149pub enum RuntimeComponentGraphError {
150 #[error("runtime component '{component}' depends on missing component '{dependency}'")]
152 MissingDependency {
153 component: &'static str,
155 dependency: &'static str,
157 },
158 #[error("runtime component dependency cycle detected: {cycle}")]
160 Cycle {
161 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#[derive(Default)]
175pub struct RuntimeComponentRegistry {
176 components: Vec<RuntimeComponentDescriptor>,
177 startup: StartupCoordinator,
178 health: HealthCheckRegistry,
179 shutdown: Vec<RuntimeComponentShutdownPhase>,
180}
181
182impl RuntimeComponentRegistry {
183 #[must_use]
185 pub fn new() -> Self {
186 Self::default()
187 }
188
189 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 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 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 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 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 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 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 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 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 #[must_use]
330 pub fn descriptors(&self) -> &[RuntimeComponentDescriptor] {
331 &self.components
332 }
333
334 #[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 #[must_use]
344 pub const fn health_registry(&self) -> &HealthCheckRegistry {
345 &self.health
346 }
347
348 pub const fn health_registry_mut(&mut self) -> &mut HealthCheckRegistry {
350 &mut self.health
351 }
352
353 pub async fn run_health(&mut self, scope: HealthCheckScope) -> SystemHealthReport {
355 self.health.run_scope(scope).await
356 }
357
358 pub async fn startup(&mut self) -> StartupReport {
360 self.startup.run().await
361 }
362
363 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 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 if let Some(indices) = phase_indices_by_component.get(component_name) {
559 ordered.extend(indices);
560 }
561 }
562
563 #[must_use]
565 pub fn len(&self) -> usize {
566 self.components.len()
567 }
568
569 #[must_use]
571 pub fn is_empty(&self) -> bool {
572 self.components.is_empty()
573 }
574}
575
576pub struct RuntimeComponentBuilder<'a> {
578 registry: &'a mut RuntimeComponentRegistry,
579 index: usize,
580}
581
582impl RuntimeComponentBuilder<'_> {
583 pub fn kind(&mut self, kind: RuntimeComponentKind) -> &mut Self {
585 self.descriptor_mut().kind = kind;
586 self
587 }
588
589 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 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 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 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 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 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 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}