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
29/// order, 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
312    /// while component health is evaluated.
313    #[must_use]
314    pub fn health_all(&self) -> Vec<Health> {
315        let snapshots = self.snapshots();
316        snapshots
317            .into_iter()
318            .map(|snapshot| snapshot.component.health())
319            .collect()
320    }
321
322    /// Return the tracked state for the named component.
323    #[must_use]
324    pub fn state(&self, name: &str) -> Option<State> {
325        self.components
326            .read()
327            .iter()
328            .find(|entry| entry.name == name)
329            .map(|entry| entry.state)
330    }
331
332    /// Return the number of registered components.
333    #[must_use]
334    pub fn len(&self) -> usize {
335        self.components.read().len()
336    }
337
338    /// Return `true` if no components have been registered.
339    #[must_use]
340    pub fn is_empty(&self) -> bool {
341        self.components.read().is_empty()
342    }
343
344    fn snapshots(&self) -> Vec<ComponentSnapshot> {
345        let components = self.components.read();
346        components
347            .iter()
348            .enumerate()
349            .map(|(index, entry)| ComponentSnapshot {
350                index,
351                name: entry.name.clone(),
352                component: Arc::clone(&entry.component),
353                state: entry.state,
354            })
355            .collect()
356    }
357
358    fn set_state(&self, index: usize, state: State) {
359        if let Some(entry) = self.components.write().get_mut(index) {
360            entry.state = state;
361        }
362    }
363
364    async fn rollback_started(&self, names: &[String]) -> Vec<StopResult> {
365        let names = names.iter().cloned().collect::<HashSet<_>>();
366        let snapshots = self
367            .snapshots()
368            .into_iter()
369            .rev()
370            .filter(|snapshot| names.contains(&snapshot.name) && snapshot.state.should_stop())
371            .collect::<Vec<_>>();
372
373        let mut results = Vec::with_capacity(snapshots.len());
374        for snapshot in snapshots {
375            self.set_state(snapshot.index, State::Stopping);
376            match tokio::time::timeout(self.config.stop_timeout, snapshot.component.stop()).await {
377                Ok(Ok(())) => {
378                    self.set_state(snapshot.index, State::Stopped);
379                    results.push(StopResult {
380                        name: snapshot.name,
381                        error: None,
382                    });
383                }
384                Ok(Err(error)) => {
385                    self.set_state(snapshot.index, State::Failed);
386                    results.push(StopResult {
387                        name: snapshot.name,
388                        error: Some(error),
389                    });
390                }
391                Err(_) => {
392                    self.set_state(snapshot.index, State::Failed);
393                    results.push(StopResult {
394                        name: snapshot.name.clone(),
395                        error: Some(AppError::new(
396                            ErrorCode::Timeout,
397                            format!("component '{}' stop timed out", snapshot.name),
398                        )),
399                    });
400                }
401            }
402        }
403        results
404    }
405}
406
407async fn start_component_snapshot(
408    components: Arc<RwLock<Vec<RegisteredComponent>>>,
409    snapshot: ComponentSnapshot,
410    cancel: CancellationToken,
411    start_timeout: Duration,
412) -> AppResult<()> {
413    if cancel.is_cancelled() {
414        restore_state(&components, snapshot.index, snapshot.state);
415        tracing::warn!(
416            component = snapshot.name,
417            "startup cancelled before dispatch"
418        );
419        return Ok(());
420    }
421
422    tracing::debug!(component = snapshot.name, "starting component (concurrent)");
423    let start = tokio::time::timeout(start_timeout, snapshot.component.start());
424    tokio::pin!(start);
425    tokio::select! {
426        () = cancel.cancelled() => {
427            restore_state(&components, snapshot.index, snapshot.state);
428            tracing::warn!(
429                component = snapshot.name,
430                "startup cancelled during dispatch"
431            );
432            Ok(())
433        }
434        result = &mut start => match result {
435            Ok(Ok(())) => {
436            update_state(&components, snapshot.index, State::Running);
437            tracing::debug!(component = snapshot.name, "component started");
438            Ok(())
439        }
440            Ok(Err(error)) => {
441            update_state(&components, snapshot.index, State::Failed);
442            Err(error)
443        }
444            Err(_) => {
445            update_state(&components, snapshot.index, State::Failed);
446            Err(AppError::new(
447                ErrorCode::Timeout,
448                format!("component '{}' start timed out", snapshot.name),
449            ))
450        }
451        }
452    }
453}
454
455fn update_state(components: &Arc<RwLock<Vec<RegisteredComponent>>>, index: usize, state: State) {
456    if let Some(entry) = components.write().get_mut(index) {
457        entry.state = state;
458    }
459}
460
461fn restore_state(
462    components: &Arc<RwLock<Vec<RegisteredComponent>>>,
463    index: usize,
464    original_state: State,
465) {
466    update_state(components, index, original_state);
467}
468
469#[cfg(test)]
470mod tests {
471    use std::sync::Arc;
472    use std::sync::atomic::{AtomicUsize, Ordering};
473    use std::thread;
474    use std::time::Duration;
475
476    use parking_lot::Mutex;
477    use rskit_errors::AppError;
478    use tokio_util::sync::CancellationToken;
479
480    use super::{Registry, RegistryConfig, restore_state};
481    use crate::{Component, Health, State};
482
483    struct MockComponent {
484        name: String,
485        start_count: Arc<AtomicUsize>,
486        stop_count: Arc<AtomicUsize>,
487        fail_on_start: bool,
488        delay: Option<Duration>,
489    }
490
491    impl MockComponent {
492        fn new(name: impl Into<String>) -> Self {
493            Self {
494                name: name.into(),
495                start_count: Arc::new(AtomicUsize::new(0)),
496                stop_count: Arc::new(AtomicUsize::new(0)),
497                fail_on_start: false,
498                delay: None,
499            }
500        }
501
502        fn with_fail_on_start(mut self) -> Self {
503            self.fail_on_start = true;
504            self
505        }
506
507        fn with_delay(mut self, delay: Duration) -> Self {
508            self.delay = Some(delay);
509            self
510        }
511    }
512
513    #[async_trait::async_trait]
514    impl Component for MockComponent {
515        fn name(&self) -> &str {
516            &self.name
517        }
518
519        async fn start(&self) -> rskit_errors::AppResult<()> {
520            if let Some(delay) = self.delay {
521                tokio::time::sleep(delay).await;
522            }
523            if self.fail_on_start {
524                return Err(AppError::service_unavailable(self.name.clone()));
525            }
526            self.start_count.fetch_add(1, Ordering::SeqCst);
527            Ok(())
528        }
529
530        async fn stop(&self) -> rskit_errors::AppResult<()> {
531            self.stop_count.fetch_add(1, Ordering::SeqCst);
532            Ok(())
533        }
534
535        fn health(&self) -> Health {
536            Health::healthy(&self.name)
537        }
538    }
539
540    struct BlockingHealthComponent {
541        name: String,
542        gate: Arc<Mutex<()>>,
543        health_count: Arc<AtomicUsize>,
544    }
545
546    #[async_trait::async_trait]
547    impl Component for BlockingHealthComponent {
548        fn name(&self) -> &str {
549            &self.name
550        }
551
552        async fn start(&self) -> rskit_errors::AppResult<()> {
553            Ok(())
554        }
555
556        async fn stop(&self) -> rskit_errors::AppResult<()> {
557            Ok(())
558        }
559
560        fn health(&self) -> Health {
561            let _guard = self.gate.lock();
562            self.health_count.fetch_add(1, Ordering::SeqCst);
563            Health::healthy(&self.name)
564        }
565    }
566
567    #[tokio::test]
568    async fn state_transitions_track_start_stop_and_restart() {
569        let component = Arc::new(MockComponent::new("svc"));
570        let mut registry = Registry::new();
571        registry.register(component);
572
573        assert_eq!(registry.state("svc"), Some(State::Created));
574
575        registry.start_all().await.expect("start should succeed");
576        assert_eq!(registry.state("svc"), Some(State::Running));
577
578        registry.stop_all().await.expect("stop should succeed");
579        assert_eq!(registry.state("svc"), Some(State::Stopped));
580
581        registry.start_all().await.expect("restart should succeed");
582        assert_eq!(registry.state("svc"), Some(State::Running));
583    }
584
585    #[tokio::test]
586    async fn start_all_skips_running_components_and_rejects_in_progress_states() {
587        let component = Arc::new(MockComponent::new("svc"));
588        let start_count = Arc::clone(&component.start_count);
589        let mut registry = Registry::new();
590        registry.register(component);
591
592        registry.start_all().await.expect("start should succeed");
593        registry
594            .start_all()
595            .await
596            .expect("running components should be skipped");
597        assert_eq!(start_count.load(Ordering::SeqCst), 1);
598
599        registry.set_state(0, State::Starting);
600        let error = registry.start_all().await.unwrap_err();
601        assert_eq!(error.code(), rskit_errors::ErrorCode::Conflict);
602        assert!(error.message().contains("cannot start from state starting"));
603    }
604
605    #[test]
606    fn registry_accessors_and_private_state_helpers_cover_empty_and_missing_indexes() {
607        let mut registry = Registry::new();
608        assert!(registry.is_empty());
609        assert_eq!(registry.len(), 0);
610        registry.set_state(42, State::Failed);
611
612        let component = Arc::new(MockComponent::new("svc"));
613        assert!(component.health().is_healthy());
614        registry.register(component);
615        assert!(!registry.is_empty());
616        assert_eq!(registry.len(), 1);
617
618        restore_state(&registry.components, 42, State::Created);
619        restore_state(&registry.components, 0, State::Stopped);
620        assert_eq!(registry.state("svc"), Some(State::Stopped));
621    }
622
623    #[tokio::test]
624    async fn start_failure_rolls_back_started_components() {
625        let started = Arc::new(MockComponent::new("started"));
626        let failing = Arc::new(MockComponent::new("failing").with_fail_on_start());
627        let never = Arc::new(MockComponent::new("never"));
628
629        let started_stop_count = Arc::clone(&started.stop_count);
630        let never_start_count = Arc::clone(&never.start_count);
631
632        let mut registry = Registry::new();
633        registry.register(started);
634        registry.register(failing);
635        registry.register(never);
636
637        let result = registry.start_all().await;
638        assert!(result.is_err());
639        assert_eq!(registry.state("started"), Some(State::Stopped));
640        assert_eq!(registry.state("failing"), Some(State::Failed));
641        assert_eq!(registry.state("never"), Some(State::Created));
642        assert_eq!(started_stop_count.load(Ordering::SeqCst), 1);
643        assert_eq!(never_start_count.load(Ordering::SeqCst), 0);
644    }
645
646    #[tokio::test(start_paused = true)]
647    async fn start_timeout_marks_component_failed() {
648        let component = Arc::new(MockComponent::new("slow").with_delay(Duration::from_secs(60)));
649        let mut registry = Registry::with_config(RegistryConfig {
650            start_timeout: Duration::from_secs(1),
651            ..RegistryConfig::default()
652        });
653        registry.register(component);
654
655        let error = registry
656            .start_all()
657            .await
658            .expect_err("slow component should time out");
659        assert_eq!(error.code(), rskit_errors::ErrorCode::Timeout);
660        assert_eq!(registry.state("slow"), Some(State::Failed));
661    }
662
663    #[tokio::test(start_paused = true)]
664    async fn start_timeout_reports_rollback_stop_failures() {
665        struct StopFailComponent(&'static str);
666
667        #[async_trait::async_trait]
668        impl Component for StopFailComponent {
669            fn name(&self) -> &str {
670                self.0
671            }
672
673            async fn start(&self) -> rskit_errors::AppResult<()> {
674                Ok(())
675            }
676
677            async fn stop(&self) -> rskit_errors::AppResult<()> {
678                Err(AppError::service_unavailable(self.0))
679            }
680
681            fn health(&self) -> Health {
682                Health::healthy(self.0)
683            }
684        }
685
686        let mut registry = Registry::with_config(RegistryConfig {
687            start_timeout: Duration::from_secs(1),
688            stop_timeout: Duration::from_secs(1),
689            ..RegistryConfig::default()
690        });
691        registry.register(Arc::new(StopFailComponent("started")));
692        registry.register(Arc::new(
693            MockComponent::new("slow").with_delay(Duration::from_secs(60)),
694        ));
695
696        let error = registry.start_all().await.unwrap_err();
697
698        assert_eq!(error.code(), rskit_errors::ErrorCode::Timeout);
699        assert!(error.to_string().contains("rollback failures"));
700        assert_eq!(registry.state("started"), Some(State::Failed));
701        assert_eq!(registry.state("slow"), Some(State::Failed));
702    }
703
704    #[tokio::test(start_paused = true)]
705    async fn start_failure_reports_rollback_stop_timeout() {
706        struct SlowStopComponent;
707
708        #[async_trait::async_trait]
709        impl Component for SlowStopComponent {
710            fn name(&self) -> &str {
711                "slow-stop-started"
712            }
713
714            async fn start(&self) -> rskit_errors::AppResult<()> {
715                Ok(())
716            }
717
718            async fn stop(&self) -> rskit_errors::AppResult<()> {
719                tokio::time::sleep(Duration::from_secs(60)).await;
720                Ok(())
721            }
722
723            fn health(&self) -> Health {
724                Health::healthy(self.name())
725            }
726        }
727
728        let mut registry = Registry::with_config(RegistryConfig {
729            stop_timeout: Duration::from_secs(1),
730            ..RegistryConfig::default()
731        });
732        registry.register(Arc::new(SlowStopComponent));
733        registry.register(Arc::new(MockComponent::new("failing").with_fail_on_start()));
734
735        let error = registry.start_all().await.unwrap_err();
736
737        assert_eq!(error.code(), rskit_errors::ErrorCode::ServiceUnavailable);
738        assert!(error.to_string().contains("rollback failures"));
739        assert_eq!(registry.state("slow-stop-started"), Some(State::Failed));
740        assert_eq!(registry.state("failing"), Some(State::Failed));
741    }
742
743    #[tokio::test(start_paused = true)]
744    async fn concurrent_start_timeout_marks_component_failed_in_internal_suite() {
745        let mut registry = Registry::with_config(RegistryConfig {
746            start_timeout: Duration::from_secs(1),
747            ..RegistryConfig::default()
748        });
749        registry.register(Arc::new(
750            MockComponent::new("slow-concurrent").with_delay(Duration::from_secs(60)),
751        ));
752
753        let error = registry
754            .start_all_concurrent(CancellationToken::new())
755            .await
756            .unwrap_err();
757
758        assert_eq!(error.code(), rskit_errors::ErrorCode::Timeout);
759        assert_eq!(registry.state("slow-concurrent"), Some(State::Failed));
760    }
761
762    #[test]
763    fn health_all_supports_concurrent_snapshots() {
764        let gate = Arc::new(Mutex::new(()));
765        let health_count = Arc::new(AtomicUsize::new(0));
766
767        let mut registry = Registry::with_config(RegistryConfig::default());
768        registry.register(Arc::new(BlockingHealthComponent {
769            name: "one".to_string(),
770            gate: Arc::clone(&gate),
771            health_count: Arc::clone(&health_count),
772        }));
773        registry.register(Arc::new(BlockingHealthComponent {
774            name: "two".to_string(),
775            gate,
776            health_count: Arc::clone(&health_count),
777        }));
778
779        let registry = Arc::new(registry);
780        let handles = (0..4)
781            .map(|_| {
782                let registry = Arc::clone(&registry);
783                thread::spawn(move || registry.health_all())
784            })
785            .collect::<Vec<_>>();
786
787        for handle in handles {
788            let results = handle.join().expect("health thread should succeed");
789            assert_eq!(results.len(), 2);
790        }
791
792        assert_eq!(health_count.load(Ordering::SeqCst), 8);
793    }
794
795    #[tokio::test]
796    async fn blocking_health_component_start_stop_and_health_are_all_exercised() {
797        let gate = Arc::new(Mutex::new(()));
798        let health_count = Arc::new(AtomicUsize::new(0));
799        let mut registry = Registry::new();
800        registry.register(Arc::new(BlockingHealthComponent {
801            name: "blocking".to_string(),
802            gate,
803            health_count: Arc::clone(&health_count),
804        }));
805
806        registry.start_all().await.expect("start should succeed");
807        assert!(registry.health_all()[0].is_healthy());
808        registry.stop_all().await.expect("stop should succeed");
809
810        assert_eq!(health_count.load(Ordering::SeqCst), 1);
811        assert_eq!(registry.state("blocking"), Some(State::Stopped));
812    }
813
814    #[tokio::test]
815    async fn stop_all_detailed_collects_all_errors() {
816        struct StopFailComponent(&'static str);
817
818        #[async_trait::async_trait]
819        impl Component for StopFailComponent {
820            fn name(&self) -> &str {
821                self.0
822            }
823
824            async fn start(&self) -> rskit_errors::AppResult<()> {
825                Ok(())
826            }
827
828            async fn stop(&self) -> rskit_errors::AppResult<()> {
829                Err(AppError::service_unavailable(self.0))
830            }
831
832            fn health(&self) -> Health {
833                Health::healthy(self.0)
834            }
835        }
836
837        let mut registry = Registry::new();
838        registry.register(Arc::new(StopFailComponent("a")));
839        registry.register(Arc::new(StopFailComponent("b")));
840        registry.start_all().await.expect("start should succeed");
841        assert!(registry.health_all().iter().all(Health::is_healthy));
842
843        let results = registry.stop_all_detailed().await;
844        assert_eq!(results.len(), 2);
845        assert!(results.iter().all(|result| result.error.is_some()));
846        assert_eq!(registry.state("a"), Some(State::Failed));
847        assert_eq!(registry.state("b"), Some(State::Failed));
848    }
849
850    #[tokio::test(start_paused = true)]
851    async fn stop_timeout_is_reported_per_component() {
852        struct SlowStopComponent;
853
854        #[async_trait::async_trait]
855        impl Component for SlowStopComponent {
856            fn name(&self) -> &str {
857                "slow-stop"
858            }
859
860            async fn start(&self) -> rskit_errors::AppResult<()> {
861                Ok(())
862            }
863
864            async fn stop(&self) -> rskit_errors::AppResult<()> {
865                tokio::time::sleep(Duration::from_secs(60)).await;
866                Ok(())
867            }
868
869            fn health(&self) -> Health {
870                Health::healthy(self.name())
871            }
872        }
873
874        let mut registry = Registry::with_config(RegistryConfig {
875            stop_timeout: Duration::from_secs(1),
876            ..RegistryConfig::default()
877        });
878        registry.register(Arc::new(SlowStopComponent));
879        registry.start_all().await.expect("start should succeed");
880        assert!(registry.health_all()[0].is_healthy());
881
882        let results = registry.stop_all_detailed().await;
883        assert_eq!(results.len(), 1);
884        assert_eq!(
885            results[0].error.as_ref().map(|error| error.code()),
886            Some(rskit_errors::ErrorCode::Timeout)
887        );
888        assert_eq!(registry.state("slow-stop"), Some(State::Failed));
889    }
890
891    #[tokio::test]
892    async fn concurrent_start_all_rolls_back_running_components_after_failure() {
893        let ok = Arc::new(MockComponent::new("ok").with_delay(Duration::from_millis(5)));
894        let fail = Arc::new(MockComponent::new("fail").with_fail_on_start());
895
896        let ok_stop_count = Arc::clone(&ok.stop_count);
897
898        let mut registry = Registry::with_config(RegistryConfig {
899            concurrency: 2,
900            ..RegistryConfig::default()
901        });
902        registry.register(ok);
903        registry.register(fail);
904
905        let result = registry
906            .start_all_concurrent(CancellationToken::new())
907            .await;
908        assert!(result.is_err());
909        let ok_state = registry.state("ok");
910        assert!(matches!(ok_state, Some(State::Created | State::Stopped)));
911        assert_eq!(registry.state("fail"), Some(State::Failed));
912        if ok_state == Some(State::Stopped) {
913            assert_eq!(ok_stop_count.load(Ordering::SeqCst), 1);
914        }
915    }
916}
rskit_component/registry.rs

rskit_component/
registry.rs
