1use 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
28pub trait RuntimeComponentBundle {
35 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#[derive(Debug, Clone, Copy, PartialEq, Eq)]
50pub enum RuntimeComponentKind {
51 Core,
53 Database,
55 Cache,
57 Storage,
59 Mail,
61 Tasks,
63 ExternalAuth,
65 Product,
67}
68
69impl RuntimeComponentKind {
70 #[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#[derive(Debug, Clone, Copy, PartialEq, Eq)]
88pub struct RuntimeShutdownDescriptor {
89 pub phase_name: &'static str,
91 pub timeout: Option<Duration>,
93}
94
95#[derive(Debug, Clone, Copy, PartialEq, Eq)]
97pub struct RuntimeStartupDescriptor {
98 pub phase_name: &'static str,
100 pub failure_policy: StartupPhaseFailurePolicy,
102}
103
104#[derive(Debug, Clone, Copy, PartialEq, Eq)]
106pub struct RuntimeTaskDescriptor {
107 pub task_name: &'static str,
109 pub display_name: &'static str,
111}
112
113#[derive(Debug, Clone, PartialEq, Eq)]
115pub struct RuntimeComponentDescriptor {
116 pub name: &'static str,
118 pub kind: RuntimeComponentKind,
120 pub dependencies: Vec<&'static str>,
122 pub startup: Vec<RuntimeStartupDescriptor>,
124 pub health_checks: Vec<HealthCheckDescriptor>,
126 pub tasks: Vec<RuntimeTaskDescriptor>,
128 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#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
148pub enum RuntimeComponentGraphError {
149 #[error("runtime component '{component}' depends on missing component '{dependency}'")]
151 MissingDependency {
152 component: &'static str,
154 dependency: &'static str,
156 },
157 #[error("runtime component dependency cycle detected: {cycle}")]
159 Cycle {
160 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#[derive(Default)]
174pub struct RuntimeComponentRegistry {
175 components: Vec<RuntimeComponentDescriptor>,
176 startup: StartupCoordinator,
177 health: HealthCheckRegistry,
178 shutdown: Vec<RuntimeComponentShutdownPhase>,
179}
180
181impl RuntimeComponentRegistry {
182 #[must_use]
184 pub fn new() -> Self {
185 Self::default()
186 }
187
188 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 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 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 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 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 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 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 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 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 #[must_use]
329 pub fn descriptors(&self) -> &[RuntimeComponentDescriptor] {
330 &self.components
331 }
332
333 #[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 #[must_use]
343 pub const fn health_registry(&self) -> &HealthCheckRegistry {
344 &self.health
345 }
346
347 pub const fn health_registry_mut(&mut self) -> &mut HealthCheckRegistry {
349 &mut self.health
350 }
351
352 pub async fn run_health(&mut self, scope: HealthCheckScope) -> SystemHealthReport {
354 self.health.run_scope(scope).await
355 }
356
357 pub async fn startup(&mut self) -> StartupReport {
359 self.startup.run().await
360 }
361
362 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 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 if let Some(indices) = phase_indices_by_component.get(component_name) {
566 ordered.extend(indices);
567 }
568 }
569
570 #[must_use]
572 pub fn len(&self) -> usize {
573 self.components.len()
574 }
575
576 #[must_use]
578 pub fn is_empty(&self) -> bool {
579 self.components.is_empty()
580 }
581}
582
583pub struct RuntimeComponentBuilder<'a> {
585 registry: &'a mut RuntimeComponentRegistry,
586 index: usize,
587}
588
589impl RuntimeComponentBuilder<'_> {
590 pub fn kind(&mut self, kind: RuntimeComponentKind) -> &mut Self {
592 self.descriptor_mut().kind = kind;
593 self
594 }
595
596 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 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 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 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 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 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 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}