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

camel_api/
step_lifecycle.rs
