rskit_component/
registry.rs

1use std::collections::HashSet;
2use std::time::Duration;
3use std::{collections::HashMap, sync::Arc};
4
5use parking_lot::RwLock;
6use rskit_errors::{AppError, AppResult, ErrorCode};
7use tokio::task::JoinSet;
8use tokio_util::sync::CancellationToken;
9
10use crate::{Component, Health, RegistryConfig, State, StopResult};
11
12#[derive(Clone)]
13struct ComponentSnapshot {
14    index: usize,
15    name: String,
16    component: Arc<dyn Component>,
17    state: State,
18}
19
20struct RegisteredComponent {
21    name: String,
22    component: Arc<dyn Component>,
23    state: State,
24}
25
26/// Ordered component registry.
27///
28/// Components start in registration order and stop in reverse registration order,
29/// ensuring dependants shut down before their dependencies.
30pub struct Registry {
31    components: Arc<RwLock<Vec<RegisteredComponent>>>,
32    config: RegistryConfig,
33}
34
35impl Default for Registry {
36    fn default() -> Self {
37        Self {
38            components: Arc::new(RwLock::new(Vec::new())),
39            config: RegistryConfig::default(),
40        }
41    }
42}
43
44impl Registry {
45    /// Create an empty [`Registry`] with default settings.
46    #[must_use]
47    pub fn new() -> Self {
48        Self::default()
49    }
50
51    /// Create a registry with custom [`RegistryConfig`].
52    #[must_use]
53    pub fn with_config(config: RegistryConfig) -> Self {
54        Self {
55            components: Arc::new(RwLock::new(Vec::new())),
56            config,
57        }
58    }
59
60    /// Register a component. Registration order determines startup order.
61    pub fn register(&mut self, component: Arc<dyn Component>) {
62        let entry = RegisteredComponent {
63            name: component.name().to_string(),
64            component,
65            state: State::Created,
66        };
67        self.components.write().push(entry);
68    }
69
70    /// Start all startable components in registration order.
71    pub async fn start_all(&self) -> AppResult<()> {
72        let mut started_names = Vec::new();
73        for snapshot in self.snapshots() {
74            match snapshot.state {
75                state if state.can_start() => {}
76                State::Running => continue,
77                state => {
78                    return Err(AppError::new(
79                        ErrorCode::Conflict,
80                        format!(
81                            "component '{}' cannot start from state {state}",
82                            snapshot.name
83                        ),
84                    ));
85                }
86            }
87
88            self.set_state(snapshot.index, State::Starting);
89            tracing::debug!(component = snapshot.name, "starting component");
90            match tokio::time::timeout(self.config.start_timeout, snapshot.component.start()).await
91            {
92                Ok(Ok(())) => {
93                    self.set_state(snapshot.index, State::Running);
94                    started_names.push(snapshot.name);
95                    tracing::debug!(component = snapshot.component.name(), "component started");
96                }
97                Ok(Err(error)) => {
98                    self.set_state(snapshot.index, State::Failed);
99                    let rollback_results = self.rollback_started(&started_names).await;
100                    let rollback_errors = rollback_results
101                        .into_iter()
102                        .filter_map(|result| {
103                            result.error.map(|err| format!("{}: {err}", result.name))
104                        })
105                        .collect::<Vec<_>>();
106                    return if rollback_errors.is_empty() {
107                        Err(error)
108                    } else {
109                        Err(error
110                            .context(format!("rollback failures: {}", rollback_errors.join("; "))))
111                    };
112                }
113                Err(_) => {
114                    self.set_state(snapshot.index, State::Failed);
115                    match tokio::time::timeout(self.config.stop_timeout, snapshot.component.stop())
116                        .await
117                    {
118                        Ok(Ok(())) => {
119                            tracing::debug!(
120                                component = snapshot.component.name(),
121                                "component stopped after start timeout"
122                            );
123                        }
124                        Ok(Err(error)) => {
125                            tracing::error!(
126                                component = snapshot.component.name(),
127                                error = %error,
128                                "component start timed out and stop cleanup failed"
129                            );
130                        }
131                        Err(_) => {
132                            tracing::error!(
133                                component = snapshot.component.name(),
134                                "component start timed out and stop cleanup also timed out"
135                            );
136                        }
137                    }
138                    let rollback_results = self.rollback_started(&started_names).await;
139                    let rollback_errors = rollback_results
140                        .into_iter()
141                        .filter_map(|result| {
142                            result.error.map(|err| format!("{}: {err}", result.name))
143                        })
144                        .collect::<Vec<_>>();
145                    let error = AppError::new(
146                        ErrorCode::Timeout,
147                        format!("component '{}' start timed out", snapshot.name),
148                    );
149                    return if rollback_errors.is_empty() {
150                        Err(error)
151                    } else {
152                        Err(error
153                            .context(format!("rollback failures: {}", rollback_errors.join("; "))))
154                    };
155                }
156            }
157        }
158        Ok(())
159    }
160
161    /// Start all components concurrently, limited by [`RegistryConfig::concurrency`].
162    pub async fn start_all_concurrent(&self, cancel: CancellationToken) -> AppResult<()> {
163        let candidates = self
164            .snapshots()
165            .into_iter()
166            .filter(|snapshot| snapshot.state.can_start())
167            .collect::<Vec<_>>();
168
169        if candidates.is_empty() {
170            return Ok(());
171        }
172
173        let concurrency = if self.config.concurrency == 0 {
174            candidates.len().max(1)
175        } else {
176            self.config.concurrency
177        };
178
179        let semaphore = Arc::new(tokio::sync::Semaphore::new(concurrency.max(1)));
180        let mut join_set = JoinSet::new();
181        let mut task_indexes = HashMap::new();
182        let candidate_names = candidates
183            .iter()
184            .map(|snapshot| snapshot.name.clone())
185            .collect::<Vec<_>>();
186
187        for snapshot in candidates {
188            self.set_state(snapshot.index, State::Starting);
189            let snapshot_index = snapshot.index;
190            let components = Arc::clone(&self.components);
191            let semaphore = Arc::clone(&semaphore);
192            let cancel = cancel.clone();
193            let start_timeout = self.config.start_timeout;
194            let abort_handle = join_set.spawn(async move {
195                let _permit = semaphore.acquire_owned().await.map_err(|_| {
196                    AppError::new(ErrorCode::Cancelled, "component startup was cancelled")
197                })?;
198                start_component_snapshot(components, snapshot, cancel, start_timeout).await
199            });
200            task_indexes.insert(abort_handle.id(), snapshot_index);
201        }
202
203        let mut first_error = None;
204        while let Some(join_result) = join_set.join_next_with_id().await {
205            match join_result {
206                Ok((_id, Ok(()))) => {}
207                Ok((_id, Err(error))) => {
208                    if first_error.is_none() {
209                        cancel.cancel();
210                        first_error = Some(error);
211                    }
212                }
213                Err(error) => {
214                    if let Some(index) = task_indexes.get(&error.id()) {
215                        self.set_state(*index, State::Failed);
216                    }
217                    if first_error.is_none() {
218                        cancel.cancel();
219                        first_error = Some(AppError::internal(error));
220                    }
221                }
222            }
223        }
224
225        if let Some(error) = first_error {
226            let rollback_results = self.rollback_started(&candidate_names).await;
227            let rollback_errors = rollback_results
228                .into_iter()
229                .filter_map(|result| result.error.map(|err| format!("{}: {err}", result.name)))
230                .collect::<Vec<_>>();
231            return if rollback_errors.is_empty() {
232                Err(error)
233            } else {
234                Err(error.context(format!("rollback failures: {}", rollback_errors.join("; "))))
235            };
236        }
237
238        Ok(())
239    }
240
241    /// Stop all running components in reverse registration order.
242    pub async fn stop_all(&self) -> AppResult<()> {
243        let results = self.stop_all_detailed().await;
244        let failures = results
245            .iter()
246            .filter_map(|result| {
247                result
248                    .error
249                    .as_ref()
250                    .map(|error| format!("{}: {error}", result.name))
251            })
252            .collect::<Vec<_>>();
253
254        if failures.is_empty() {
255            Ok(())
256        } else {
257            Err(AppError::new(
258                ErrorCode::Internal,
259                format!("failed to stop components: {}", failures.join("; ")),
260            ))
261        }
262    }
263
264    /// Stop all running components in reverse registration order with per-component results.
265    pub async fn stop_all_detailed(&self) -> Vec<StopResult> {
266        let snapshots = self
267            .snapshots()
268            .into_iter()
269            .rev()
270            .filter(|snapshot| snapshot.state.should_stop())
271            .collect::<Vec<_>>();
272
273        let mut results = Vec::with_capacity(snapshots.len());
274        for snapshot in snapshots {
275            self.set_state(snapshot.index, State::Stopping);
276            tracing::debug!(component = snapshot.name, "stopping component");
277            match tokio::time::timeout(self.config.stop_timeout, snapshot.component.stop()).await {
278                Ok(Ok(())) => {
279                    self.set_state(snapshot.index, State::Stopped);
280                    tracing::debug!(component = snapshot.component.name(), "component stopped");
281                    results.push(StopResult {
282                        name: snapshot.name,
283                        error: None,
284                    });
285                }
286                Ok(Err(error)) => {
287                    self.set_state(snapshot.index, State::Failed);
288                    tracing::warn!(component = snapshot.component.name(), error = %error, "error stopping component");
289                    results.push(StopResult {
290                        name: snapshot.name,
291                        error: Some(error),
292                    });
293                }
294                Err(_) => {
295                    self.set_state(snapshot.index, State::Failed);
296                    let error = AppError::new(
297                        ErrorCode::Timeout,
298                        format!("component '{}' stop timed out", snapshot.name),
299                    );
300                    tracing::warn!(component = snapshot.component.name(), error = %error, "error stopping component");
301                    results.push(StopResult {
302                        name: snapshot.name,
303                        error: Some(error),
304                    });
305                }
306            }
307        }
308        results
309    }
310
311    /// Collect health for all registered components without holding the registry lock while component health is evaluated.
312    #[must_use]
313    pub fn health_all(&self) -> Vec<Health> {
314        let snapshots = self.snapshots();
315        snapshots
316            .into_iter()
317            .map(|snapshot| snapshot.component.health())
318            .collect()
319    }
320
321    /// Return the tracked state for the named component.
322    #[must_use]
323    pub fn state(&self, name: &str) -> Option<State> {
324        self.components
325            .read()
326            .iter()
327            .find(|entry| entry.name == name)
328            .map(|entry| entry.state)
329    }
330
331    /// Return the number of registered components.
332    #[must_use]
333    pub fn len(&self) -> usize {
334        self.components.read().len()
335    }
336
337    /// Return `true` if no components have been registered.
338    #[must_use]
339    pub fn is_empty(&self) -> bool {
340        self.components.read().is_empty()
341    }
342
343    fn snapshots(&self) -> Vec<ComponentSnapshot> {
344        let components = self.components.read();
345        components
346            .iter()
347            .enumerate()
348            .map(|(index, entry)| ComponentSnapshot {
349                index,
350                name: entry.name.clone(),
351                component: Arc::clone(&entry.component),
352                state: entry.state,
353            })
354            .collect()
355    }
356
357    fn set_state(&self, index: usize, state: State) {
358        if let Some(entry) = self.components.write().get_mut(index) {
359            entry.state = state;
360        }
361    }
362
363    async fn rollback_started(&self, names: &[String]) -> Vec<StopResult> {
364        let names = names.iter().cloned().collect::<HashSet<_>>();
365        let snapshots = self
366            .snapshots()
367            .into_iter()
368            .rev()
369            .filter(|snapshot| names.contains(&snapshot.name) && snapshot.state.should_stop())
370            .collect::<Vec<_>>();
371
372        let mut results = Vec::with_capacity(snapshots.len());
373        for snapshot in snapshots {
374            self.set_state(snapshot.index, State::Stopping);
375            match tokio::time::timeout(self.config.stop_timeout, snapshot.component.stop()).await {
376                Ok(Ok(())) => {
377                    self.set_state(snapshot.index, State::Stopped);
378                    results.push(StopResult {
379                        name: snapshot.name,
380                        error: None,
381                    });
382                }
383                Ok(Err(error)) => {
384                    self.set_state(snapshot.index, State::Failed);
385                    results.push(StopResult {
386                        name: snapshot.name,
387                        error: Some(error),
388                    });
389                }
390                Err(_) => {
391                    self.set_state(snapshot.index, State::Failed);
392                    results.push(StopResult {
393                        name: snapshot.name.clone(),
394                        error: Some(AppError::new(
395                            ErrorCode::Timeout,
396                            format!("component '{}' stop timed out", snapshot.name),
397                        )),
398                    });
399                }
400            }
401        }
402        results
403    }
404}
405
406async fn start_component_snapshot(
407    components: Arc<RwLock<Vec<RegisteredComponent>>>,
408    snapshot: ComponentSnapshot,
409    cancel: CancellationToken,
410    start_timeout: Duration,
411) -> AppResult<()> {
412    if cancel.is_cancelled() {
413        restore_state(&components, snapshot.index, snapshot.state);
414        tracing::warn!(
415            component = snapshot.name,
416            "startup cancelled before dispatch"
417        );
418        return Ok(());
419    }
420
421    tracing::debug!(component = snapshot.name, "starting component (concurrent)");
422    let start = tokio::time::timeout(start_timeout, snapshot.component.start());
423    tokio::pin!(start);
424    tokio::select! {
425        () = cancel.cancelled() => {
426            restore_state(&components, snapshot.index, snapshot.state);
427            tracing::warn!(
428                component = snapshot.name,
429                "startup cancelled during dispatch"
430            );
431            Ok(())
432        }
433        result = &mut start => match result {
434            Ok(Ok(())) => {
435            update_state(&components, snapshot.index, State::Running);
436            tracing::debug!(component = snapshot.name, "component started");
437            Ok(())
438        }
439            Ok(Err(error)) => {
440            update_state(&components, snapshot.index, State::Failed);
441            Err(error)
442        }
443            Err(_) => {
444            update_state(&components, snapshot.index, State::Failed);
445            Err(AppError::new(
446                ErrorCode::Timeout,
447                format!("component '{}' start timed out", snapshot.name),
448            ))
449        }
450        }
451    }
452}
453
454fn update_state(components: &Arc<RwLock<Vec<RegisteredComponent>>>, index: usize, state: State) {
455    if let Some(entry) = components.write().get_mut(index) {
456        entry.state = state;
457    }
458}
459
460fn restore_state(
461    components: &Arc<RwLock<Vec<RegisteredComponent>>>,
462    index: usize,
463    original_state: State,
464) {
465    update_state(components, index, original_state);
466}
467
468#[cfg(test)]
469mod tests {
470    use std::sync::Arc;
471    use std::sync::atomic::{AtomicUsize, Ordering};
472    use std::thread;
473    use std::time::Duration;
474
475    use parking_lot::Mutex;
476    use rskit_errors::AppError;
477    use tokio_util::sync::CancellationToken;
478
479    use super::{Registry, RegistryConfig, restore_state};
480    use crate::{Component, Health, State};
481
482    struct MockComponent {
483        name: String,
484        start_count: Arc<AtomicUsize>,
485        stop_count: Arc<AtomicUsize>,
486        fail_on_start: bool,
487        delay: Option<Duration>,
488    }
489
490    impl MockComponent {
491        fn new(name: impl Into<String>) -> Self {
492            Self {
493                name: name.into(),
494                start_count: Arc::new(AtomicUsize::new(0)),
495                stop_count: Arc::new(AtomicUsize::new(0)),
496                fail_on_start: false,
497                delay: None,
498            }
499        }
500
501        fn with_fail_on_start(mut self) -> Self {
502            self.fail_on_start = true;
503            self
504        }
505
506        fn with_delay(mut self, delay: Duration) -> Self {
507            self.delay = Some(delay);
508            self
509        }
510    }
511
512    #[async_trait::async_trait]
513    impl Component for MockComponent {
514        fn name(&self) -> &str {
515            &self.name
516        }
517
518        async fn start(&self) -> rskit_errors::AppResult<()> {
519            if let Some(delay) = self.delay {
520                tokio::time::sleep(delay).await;
521            }
522            if self.fail_on_start {
523                return Err(AppError::service_unavailable(self.name.clone()));
524            }
525            self.start_count.fetch_add(1, Ordering::SeqCst);
526            Ok(())
527        }
528
529        async fn stop(&self) -> rskit_errors::AppResult<()> {
530            self.stop_count.fetch_add(1, Ordering::SeqCst);
531            Ok(())
532        }
533
534        fn health(&self) -> Health {
535            Health::healthy(&self.name)
536        }
537    }
538
539    struct BlockingHealthComponent {
540        name: String,
541        gate: Arc<Mutex<()>>,
542        health_count: Arc<AtomicUsize>,
543    }
544
545    #[async_trait::async_trait]
546    impl Component for BlockingHealthComponent {
547        fn name(&self) -> &str {
548            &self.name
549        }
550
551        async fn start(&self) -> rskit_errors::AppResult<()> {
552            Ok(())
553        }
554
555        async fn stop(&self) -> rskit_errors::AppResult<()> {
556            Ok(())
557        }
558
559        fn health(&self) -> Health {
560            let _guard = self.gate.lock();
561            self.health_count.fetch_add(1, Ordering::SeqCst);
562            Health::healthy(&self.name)
563        }
564    }
565
566    #[tokio::test]
567    async fn state_transitions_track_start_stop_and_restart() {
568        let component = Arc::new(MockComponent::new("svc"));
569        let mut registry = Registry::new();
570        registry.register(component);
571
572        assert_eq!(registry.state("svc"), Some(State::Created));
573
574        registry.start_all().await.expect("start should succeed");
575        assert_eq!(registry.state("svc"), Some(State::Running));
576
577        registry.stop_all().await.expect("stop should succeed");
578        assert_eq!(registry.state("svc"), Some(State::Stopped));
579
580        registry.start_all().await.expect("restart should succeed");
581        assert_eq!(registry.state("svc"), Some(State::Running));
582    }
583
584    #[tokio::test]
585    async fn start_all_skips_running_components_and_rejects_in_progress_states() {
586        let component = Arc::new(MockComponent::new("svc"));
587        let start_count = Arc::clone(&component.start_count);
588        let mut registry = Registry::new();
589        registry.register(component);
590
591        registry.start_all().await.expect("start should succeed");
592        registry
593            .start_all()
594            .await
595            .expect("running components should be skipped");
596        assert_eq!(start_count.load(Ordering::SeqCst), 1);
597
598        registry.set_state(0, State::Starting);
599        let error = registry.start_all().await.unwrap_err();
600        assert_eq!(error.code(), rskit_errors::ErrorCode::Conflict);
601        assert!(error.message().contains("cannot start from state starting"));
602    }
603
604    #[test]
605    fn registry_accessors_and_private_state_helpers_cover_empty_and_missing_indexes() {
606        let mut registry = Registry::new();
607        assert!(registry.is_empty());
608        assert_eq!(registry.len(), 0);
609        registry.set_state(42, State::Failed);
610
611        let component = Arc::new(MockComponent::new("svc"));
612        assert!(component.health().is_healthy());
613        registry.register(component);
614        assert!(!registry.is_empty());
615        assert_eq!(registry.len(), 1);
616
617        restore_state(&registry.components, 42, State::Created);
618        restore_state(&registry.components, 0, State::Stopped);
619        assert_eq!(registry.state("svc"), Some(State::Stopped));
620    }
621
622    #[tokio::test]
623    async fn start_failure_rolls_back_started_components() {
624        let started = Arc::new(MockComponent::new("started"));
625        let failing = Arc::new(MockComponent::new("failing").with_fail_on_start());
626        let never = Arc::new(MockComponent::new("never"));
627
628        let started_stop_count = Arc::clone(&started.stop_count);
629        let never_start_count = Arc::clone(&never.start_count);
630
631        let mut registry = Registry::new();
632        registry.register(started);
633        registry.register(failing);
634        registry.register(never);
635
636        let result = registry.start_all().await;
637        assert!(result.is_err());
638        assert_eq!(registry.state("started"), Some(State::Stopped));
639        assert_eq!(registry.state("failing"), Some(State::Failed));
640        assert_eq!(registry.state("never"), Some(State::Created));
641        assert_eq!(started_stop_count.load(Ordering::SeqCst), 1);
642        assert_eq!(never_start_count.load(Ordering::SeqCst), 0);
643    }
644
645    #[tokio::test(start_paused = true)]
646    async fn start_timeout_marks_component_failed() {
647        let component = Arc::new(MockComponent::new("slow").with_delay(Duration::from_secs(60)));
648        let mut registry = Registry::with_config(RegistryConfig {
649            start_timeout: Duration::from_secs(1),
650            ..RegistryConfig::default()
651        });
652        registry.register(component);
653
654        let error = registry
655            .start_all()
656            .await
657            .expect_err("slow component should time out");
658        assert_eq!(error.code(), rskit_errors::ErrorCode::Timeout);
659        assert_eq!(registry.state("slow"), Some(State::Failed));
660    }
661
662    #[tokio::test(start_paused = true)]
663    async fn start_timeout_reports_rollback_stop_failures() {
664        struct StopFailComponent(&'static str);
665
666        #[async_trait::async_trait]
667        impl Component for StopFailComponent {
668            fn name(&self) -> &str {
669                self.0
670            }
671
672            async fn start(&self) -> rskit_errors::AppResult<()> {
673                Ok(())
674            }
675
676            async fn stop(&self) -> rskit_errors::AppResult<()> {
677                Err(AppError::service_unavailable(self.0))
678            }
679
680            fn health(&self) -> Health {
681                Health::healthy(self.0)
682            }
683        }
684
685        let mut registry = Registry::with_config(RegistryConfig {
686            start_timeout: Duration::from_secs(1),
687            stop_timeout: Duration::from_secs(1),
688            ..RegistryConfig::default()
689        });
690        registry.register(Arc::new(StopFailComponent("started")));
691        registry.register(Arc::new(
692            MockComponent::new("slow").with_delay(Duration::from_secs(60)),
693        ));
694
695        let error = registry.start_all().await.unwrap_err();
696
697        assert_eq!(error.code(), rskit_errors::ErrorCode::Timeout);
698        assert!(error.to_string().contains("rollback failures"));
699        assert_eq!(registry.state("started"), Some(State::Failed));
700        assert_eq!(registry.state("slow"), Some(State::Failed));
701    }
702
703    #[tokio::test(start_paused = true)]
704    async fn start_failure_reports_rollback_stop_timeout() {
705        struct SlowStopComponent;
706
707        #[async_trait::async_trait]
708        impl Component for SlowStopComponent {
709            fn name(&self) -> &str {
710                "slow-stop-started"
711            }
712
713            async fn start(&self) -> rskit_errors::AppResult<()> {
714                Ok(())
715            }
716
717            async fn stop(&self) -> rskit_errors::AppResult<()> {
718                tokio::time::sleep(Duration::from_secs(60)).await;
719                Ok(())
720            }
721
722            fn health(&self) -> Health {
723                Health::healthy(self.name())
724            }
725        }
726
727        let mut registry = Registry::with_config(RegistryConfig {
728            stop_timeout: Duration::from_secs(1),
729            ..RegistryConfig::default()
730        });
731        registry.register(Arc::new(SlowStopComponent));
732        registry.register(Arc::new(MockComponent::new("failing").with_fail_on_start()));
733
734        let error = registry.start_all().await.unwrap_err();
735
736        assert_eq!(error.code(), rskit_errors::ErrorCode::ServiceUnavailable);
737        assert!(error.to_string().contains("rollback failures"));
738        assert_eq!(registry.state("slow-stop-started"), Some(State::Failed));
739        assert_eq!(registry.state("failing"), Some(State::Failed));
740    }
741
742    #[tokio::test(start_paused = true)]
743    async fn concurrent_start_timeout_marks_component_failed_in_internal_suite() {
744        let mut registry = Registry::with_config(RegistryConfig {
745            start_timeout: Duration::from_secs(1),
746            ..RegistryConfig::default()
747        });
748        registry.register(Arc::new(
749            MockComponent::new("slow-concurrent").with_delay(Duration::from_secs(60)),
750        ));
751
752        let error = registry
753            .start_all_concurrent(CancellationToken::new())
754            .await
755            .unwrap_err();
756
757        assert_eq!(error.code(), rskit_errors::ErrorCode::Timeout);
758        assert_eq!(registry.state("slow-concurrent"), Some(State::Failed));
759    }
760
761    #[test]
762    fn health_all_supports_concurrent_snapshots() {
763        let gate = Arc::new(Mutex::new(()));
764        let health_count = Arc::new(AtomicUsize::new(0));
765
766        let mut registry = Registry::with_config(RegistryConfig::default());
767        registry.register(Arc::new(BlockingHealthComponent {
768            name: "one".to_string(),
769            gate: Arc::clone(&gate),
770            health_count: Arc::clone(&health_count),
771        }));
772        registry.register(Arc::new(BlockingHealthComponent {
773            name: "two".to_string(),
774            gate,
775            health_count: Arc::clone(&health_count),
776        }));
777
778        let registry = Arc::new(registry);
779        let handles = (0..4)
780            .map(|_| {
781                let registry = Arc::clone(&registry);
782                thread::spawn(move || registry.health_all())
783            })
784            .collect::<Vec<_>>();
785
786        for handle in handles {
787            let results = handle.join().expect("health thread should succeed");
788            assert_eq!(results.len(), 2);
789        }
790
791        assert_eq!(health_count.load(Ordering::SeqCst), 8);
792    }
793
794    #[tokio::test]
795    async fn blocking_health_component_start_stop_and_health_are_all_exercised() {
796        let gate = Arc::new(Mutex::new(()));
797        let health_count = Arc::new(AtomicUsize::new(0));
798        let mut registry = Registry::new();
799        registry.register(Arc::new(BlockingHealthComponent {
800            name: "blocking".to_string(),
801            gate,
802            health_count: Arc::clone(&health_count),
803        }));
804
805        registry.start_all().await.expect("start should succeed");
806        assert!(registry.health_all()[0].is_healthy());
807        registry.stop_all().await.expect("stop should succeed");
808
809        assert_eq!(health_count.load(Ordering::SeqCst), 1);
810        assert_eq!(registry.state("blocking"), Some(State::Stopped));
811    }
812
813    #[tokio::test]
814    async fn stop_all_detailed_collects_all_errors() {
815        struct StopFailComponent(&'static str);
816
817        #[async_trait::async_trait]
818        impl Component for StopFailComponent {
819            fn name(&self) -> &str {
820                self.0
821            }
822
823            async fn start(&self) -> rskit_errors::AppResult<()> {
824                Ok(())
825            }
826
827            async fn stop(&self) -> rskit_errors::AppResult<()> {
828                Err(AppError::service_unavailable(self.0))
829            }
830
831            fn health(&self) -> Health {
832                Health::healthy(self.0)
833            }
834        }
835
836        let mut registry = Registry::new();
837        registry.register(Arc::new(StopFailComponent("a")));
838        registry.register(Arc::new(StopFailComponent("b")));
839        registry.start_all().await.expect("start should succeed");
840        assert!(registry.health_all().iter().all(Health::is_healthy));
841
842        let results = registry.stop_all_detailed().await;
843        assert_eq!(results.len(), 2);
844        assert!(results.iter().all(|result| result.error.is_some()));
845        assert_eq!(registry.state("a"), Some(State::Failed));
846        assert_eq!(registry.state("b"), Some(State::Failed));
847    }
848
849    #[tokio::test(start_paused = true)]
850    async fn stop_timeout_is_reported_per_component() {
851        struct SlowStopComponent;
852
853        #[async_trait::async_trait]
854        impl Component for SlowStopComponent {
855            fn name(&self) -> &str {
856                "slow-stop"
857            }
858
859            async fn start(&self) -> rskit_errors::AppResult<()> {
860                Ok(())
861            }
862
863            async fn stop(&self) -> rskit_errors::AppResult<()> {
864                tokio::time::sleep(Duration::from_secs(60)).await;
865                Ok(())
866            }
867
868            fn health(&self) -> Health {
869                Health::healthy(self.name())
870            }
871        }
872
873        let mut registry = Registry::with_config(RegistryConfig {
874            stop_timeout: Duration::from_secs(1),
875            ..RegistryConfig::default()
876        });
877        registry.register(Arc::new(SlowStopComponent));
878        registry.start_all().await.expect("start should succeed");
879        assert!(registry.health_all()[0].is_healthy());
880
881        let results = registry.stop_all_detailed().await;
882        assert_eq!(results.len(), 1);
883        assert_eq!(
884            results[0].error.as_ref().map(|error| error.code()),
885            Some(rskit_errors::ErrorCode::Timeout)
886        );
887        assert_eq!(registry.state("slow-stop"), Some(State::Failed));
888    }
889
890    #[tokio::test]
891    async fn concurrent_start_all_rolls_back_running_components_after_failure() {
892        let ok = Arc::new(MockComponent::new("ok").with_delay(Duration::from_millis(5)));
893        let fail = Arc::new(MockComponent::new("fail").with_fail_on_start());
894
895        let ok_stop_count = Arc::clone(&ok.stop_count);
896
897        let mut registry = Registry::with_config(RegistryConfig {
898            concurrency: 2,
899            ..RegistryConfig::default()
900        });
901        registry.register(ok);
902        registry.register(fail);
903
904        let result = registry
905            .start_all_concurrent(CancellationToken::new())
906            .await;
907        assert!(result.is_err());
908        let ok_state = registry.state("ok");
909        assert!(matches!(ok_state, Some(State::Created | State::Stopped)));
910        assert_eq!(registry.state("fail"), Some(State::Failed));
911        if ok_state == Some(State::Stopped) {
912            assert_eq!(ok_stop_count.load(Ordering::SeqCst), 1);
913        }
914    }
915}
rskit_component/registry.rs

rskit_component/
registry.rs
