camel_api/
step_lifecycle.rs

1use crate::CamelError;
2use async_trait::async_trait;
3
4/// Why a stateful pipeline step is being shut down.
5#[derive(Debug, Clone, Copy, PartialEq, Eq)]
6#[non_exhaustive]
7pub enum StepShutdownReason {
8    /// The route is stopping (`stop_route`).
9    RouteStop,
10    /// The pipeline is being replaced via hot reload (Restart path).
11    HotSwap,
12}
13
14/// Lifecycle hook for **stateful** pipeline steps that own background work
15/// (timers, buckets, gap-detectors, queues) beyond a single `process()` call.
16///
17/// Stateless processors do NOT implement this trait. The runtime collects
18/// `Arc<dyn StepLifecycle>` at compile time and drains them in route order
19/// during `stop_route` and hot-swap. See ADR-0022.
20///
21/// **Why `&self`, not `&mut self`?** `Lifecycle` uses `&mut self` for exclusive
22/// start/stop of services. `StepLifecycle` is dispatched through
23/// `Arc<dyn StepLifecycle>` carried inside `ArcSwap` pipeline snapshots, so it
24/// MUST be `&self` (shared-reference, interior-mutability) for `Arc` cloning and
25/// concurrent snapshots to work. See ADR-0022.
26///
27/// `shutdown` MUST be idempotent. By the time it is called, intake is cancelled
28/// and the pipeline task has been joined, so no `process()` is in flight.
29/// `Err` is best-effort: the runtime logs and continues (it does NOT fail
30/// `stop_route`), mirroring `CamelContext::stop` service handling.
31#[async_trait]
32pub trait StepLifecycle: std::fmt::Debug + Send + Sync + 'static {
33    /// Stable name for logging/diagnostics.
34    fn name(&self) -> &'static str;
35
36    async fn shutdown(&self, reason: StepShutdownReason) -> Result<(), CamelError>;
37
38    /// Called when the route starts.
39    ///
40    /// Default is a no-op so existing implementors remain source-compatible.
41    /// Stateful steps that own background work (timers, buckets, gap-detectors)
42    /// should override this to start that work after construction. Mirrors
43    /// `Lifecycle::start` semantics, but with `&self` (shared-reference,
44    /// interior-mutability) — see ADR-0022.
45    // TODO(reload-start): `start()` is currently invoked only from `start_route`.
46    // The hot-reload path (swap_pipeline / swap_pipeline_raw) stores lifecycle
47    // handles but does not yet call `start()` on rebuilt handles; whether they
48    // need re-start is a future decision (ADR-0022).
49    async fn start(&self) -> Result<(), CamelError> {
50        Ok(())
51    }
52}
53
54#[cfg(test)]
55mod tests {
56    use super::*;
57    use std::sync::{Arc, Mutex};
58
59    #[derive(Debug)]
60    struct FakeStep {
61        shutdowns: Mutex<Vec<StepShutdownReason>>,
62    }
63
64    #[async_trait]
65    impl StepLifecycle for FakeStep {
66        fn name(&self) -> &'static str {
67            "fake"
68        }
69        async fn shutdown(&self, reason: StepShutdownReason) -> Result<(), CamelError> {
70            self.shutdowns.lock().unwrap().push(reason);
71            Ok(())
72        }
73    }
74
75    #[tokio::test]
76    async fn dyn_dispatch_works() {
77        // Keep concrete handle for assertion, dispatch through Arc<dyn StepLifecycle>
78        // (the ArcSwap-snapshot shape).
79        let inner = Arc::new(FakeStep {
80            shutdowns: Mutex::new(vec![]),
81        });
82        let step: Arc<dyn StepLifecycle> = inner.clone();
83        step.shutdown(StepShutdownReason::RouteStop).await.unwrap();
84        step.shutdown(StepShutdownReason::HotSwap).await.unwrap();
85        assert_eq!(
86            *inner.shutdowns.lock().unwrap(),
87            vec![StepShutdownReason::RouteStop, StepShutdownReason::HotSwap,]
88        );
89    }
90
91    #[tokio::test]
92    async fn start_default_is_noop() {
93        // FakeStep does not override `start`; the default impl on the trait
94        // must return `Ok(())` so existing implementors (ResequencerService,
95        // AggregatorService, test fakes) are unaffected.
96        let step: Arc<dyn StepLifecycle> = Arc::new(FakeStep {
97            shutdowns: Mutex::new(vec![]),
98        });
99        let result = step.start().await;
100        assert!(result.is_ok());
101    }
102
103    #[tokio::test]
104    async fn shutdown_err_is_not_fatal() {
105        #[derive(Debug)]
106        struct FailingStep;
107        #[async_trait]
108        impl StepLifecycle for FailingStep {
109            fn name(&self) -> &'static str {
110                "fail"
111            }
112            async fn shutdown(&self, _: StepShutdownReason) -> Result<(), CamelError> {
113                Err(CamelError::ProcessorError("boom".into()))
114            }
115        }
116        let step: Arc<dyn StepLifecycle> = Arc::new(FailingStep);
117        let result = step.shutdown(StepShutdownReason::RouteStop).await;
118        assert!(result.is_err());
119        // But typical drain loop continues (log + skip), see Task 5.
120    }
121}
camel_api/step_lifecycle.rs

camel_api/
step_lifecycle.rs
