aster_forge_runtime/
shutdown.rs

1//! Termination signal helpers.
2//!
3//! This module only waits for process termination signals. Product crates remain
4//! responsible for recording audit events, stopping background tasks, flushing
5//! buffers, and closing database or network handles in their preferred order.
6
7use std::future::Future;
8use std::pin::Pin;
9use std::time::{Duration, Instant};
10
11use tokio::task::JoinHandle;
12use tokio_util::sync::CancellationToken;
13
14type ShutdownFuture = Pin<Box<dyn Future<Output = Result<(), String>> + Send>>;
15type ShutdownPhaseFn = dyn FnMut() -> ShutdownFuture + Send;
16
17/// Termination signal observed by the runtime.
18#[derive(Debug, Clone, Copy, PartialEq, Eq)]
19pub enum TerminationSignal {
20    /// Unix SIGINT or cross-platform Ctrl+C.
21    Interrupt,
22    /// Unix SIGTERM.
23    Terminate,
24}
25
26impl TerminationSignal {
27    /// Returns a stable label for logging and tests.
28    #[must_use]
29    pub const fn as_str(self) -> &'static str {
30        match self {
31            Self::Interrupt => "SIGINT",
32            Self::Terminate => "SIGTERM",
33        }
34    }
35}
36
37/// Errors returned while installing signal listeners.
38#[derive(Debug, thiserror::Error)]
39pub enum RuntimeSignalError {
40    /// Failed to install or await a process signal handler.
41    #[error("failed to install termination signal handler: {0}")]
42    Install(String),
43}
44
45/// Final status for one shutdown phase.
46#[derive(Debug, Clone, PartialEq, Eq)]
47pub enum ShutdownPhaseStatus {
48    /// The phase completed successfully.
49    Succeeded,
50    /// The phase returned an error string.
51    Failed(String),
52    /// The phase exceeded its timeout.
53    TimedOut,
54}
55
56impl ShutdownPhaseStatus {
57    /// Returns whether this phase did not complete successfully.
58    #[must_use]
59    pub const fn is_failure(&self) -> bool {
60        !matches!(self, Self::Succeeded)
61    }
62}
63
64/// Report for one executed shutdown phase.
65#[derive(Debug, Clone, PartialEq, Eq)]
66pub struct ShutdownPhaseReport {
67    /// Stable phase name.
68    pub name: &'static str,
69    /// Phase result.
70    pub status: ShutdownPhaseStatus,
71    /// Execution duration.
72    pub duration: Duration,
73}
74
75/// Aggregate report for a shutdown run.
76#[derive(Debug, Clone, PartialEq, Eq)]
77pub struct ShutdownReport {
78    /// Phase reports in execution order.
79    pub phases: Vec<ShutdownPhaseReport>,
80}
81
82impl ShutdownReport {
83    /// Returns a report from phase entries.
84    #[must_use]
85    pub fn new(phases: Vec<ShutdownPhaseReport>) -> Self {
86        Self { phases }
87    }
88
89    /// Returns whether any phase failed or timed out.
90    #[must_use]
91    pub fn has_failures(&self) -> bool {
92        self.phases.iter().any(|phase| phase.status.is_failure())
93    }
94}
95
96/// Logs the aggregate result of a shutdown run.
97pub fn log_shutdown_report(report: &ShutdownReport) {
98    if report.has_failures() {
99        tracing::warn!("shutdown completed with one or more failed phases");
100    } else {
101        tracing::info!("shutdown complete");
102    }
103}
104
105struct RegisteredShutdownPhase {
106    name: &'static str,
107    timeout: Option<Duration>,
108    phase: Box<ShutdownPhaseFn>,
109}
110
111/// Ordered shutdown phase coordinator.
112///
113/// The coordinator owns phase ordering, timeout handling, duration collection,
114/// and error aggregation. Product crates provide the actual phase closures.
115/// Phases are `FnMut` so shutdown code can move owned handles into the
116/// coordinator and consume them exactly once during the shutdown run.
117#[derive(Default)]
118pub struct ShutdownCoordinator {
119    phases: Vec<RegisteredShutdownPhase>,
120}
121
122impl ShutdownCoordinator {
123    /// Creates an empty shutdown coordinator.
124    #[must_use]
125    pub fn new() -> Self {
126        Self::default()
127    }
128
129    /// Registers a shutdown phase.
130    pub fn phase<F, Fut>(
131        &mut self,
132        name: &'static str,
133        timeout: Option<Duration>,
134        mut phase: F,
135    ) -> &mut Self
136    where
137        F: FnMut() -> Fut + Send + 'static,
138        Fut: Future<Output = Result<(), String>> + Send + 'static,
139    {
140        self.phases.push(RegisteredShutdownPhase {
141            name,
142            timeout,
143            phase: Box::new(move || Box::pin(phase())),
144        });
145        self
146    }
147
148    /// Runs phases sequentially and returns a report.
149    ///
150    /// Later phases still run when an earlier phase fails. This lets product
151    /// shutdown code make best-effort progress through independent resources.
152    pub async fn run(&mut self) -> ShutdownReport {
153        let mut reports = Vec::with_capacity(self.phases.len());
154
155        for phase in &mut self.phases {
156            tracing::info!(phase = phase.name, "starting shutdown phase");
157            let started_at = Instant::now();
158            let future = (phase.phase)();
159            let status = match phase.timeout {
160                Some(timeout) => match tokio::time::timeout(timeout, future).await {
161                    Ok(Ok(())) => ShutdownPhaseStatus::Succeeded,
162                    Ok(Err(error)) => ShutdownPhaseStatus::Failed(error),
163                    Err(_) => ShutdownPhaseStatus::TimedOut,
164                },
165                None => match future.await {
166                    Ok(()) => ShutdownPhaseStatus::Succeeded,
167                    Err(error) => ShutdownPhaseStatus::Failed(error),
168                },
169            };
170            let duration = started_at.elapsed();
171            match &status {
172                ShutdownPhaseStatus::Succeeded => {
173                    tracing::info!(phase = phase.name, ?duration, "shutdown phase completed");
174                }
175                ShutdownPhaseStatus::Failed(error) => {
176                    tracing::error!(phase = phase.name, ?duration, %error, "shutdown phase failed");
177                }
178                ShutdownPhaseStatus::TimedOut => {
179                    tracing::error!(phase = phase.name, ?duration, "shutdown phase timed out");
180                }
181            }
182            reports.push(ShutdownPhaseReport {
183                name: phase.name,
184                status,
185                duration,
186            });
187        }
188
189        ShutdownReport::new(reports)
190    }
191
192    /// Returns how many phases are registered.
193    #[must_use]
194    pub fn len(&self) -> usize {
195        self.phases.len()
196    }
197
198    /// Returns whether no phases are registered.
199    #[must_use]
200    pub fn is_empty(&self) -> bool {
201        self.phases.is_empty()
202    }
203}
204
205/// Waits until the process receives a termination signal.
206///
207/// # Errors
208///
209/// Returns an error when the platform signal listeners cannot be installed or all listener streams
210/// terminate before delivering a signal.
211pub async fn wait_for_termination_signal() -> Result<TerminationSignal, RuntimeSignalError> {
212    let signal = wait_for_signal_impl().await?;
213    tracing::info!(
214        signal = signal.as_str(),
215        "received termination signal, shutting down gracefully..."
216    );
217    Ok(signal)
218}
219
220/// Spawns a task that waits for a termination signal, cancels `shutdown_token`,
221/// and then runs `on_signal`.
222///
223/// This keeps product entrypoints from duplicating the same signal-listener
224/// task while leaving the actual server stop primitive product-specific.
225pub fn spawn_termination_signal_handler<F, Fut>(
226    shutdown_token: CancellationToken,
227    on_signal: F,
228) -> JoinHandle<()>
229where
230    F: FnOnce() -> Fut + Send + 'static,
231    Fut: Future<Output = ()> + Send + 'static,
232{
233    tokio::spawn(async move {
234        if let Err(error) = wait_for_termination_signal().await {
235            tracing::error!(%error, "shutdown signal listener failed");
236        }
237        shutdown_token.cancel();
238        on_signal().await;
239    })
240}
241
242#[cfg(unix)]
243async fn wait_for_signal_impl() -> Result<TerminationSignal, RuntimeSignalError> {
244    use tokio::signal::unix::{SignalKind, signal};
245
246    let mut sigint = signal(SignalKind::interrupt())
247        .map_err(|error| RuntimeSignalError::Install(error.to_string()))?;
248    let mut sigterm = signal(SignalKind::terminate())
249        .map_err(|error| RuntimeSignalError::Install(error.to_string()))?;
250
251    tokio::select! {
252        _ = sigint.recv() => Ok(TerminationSignal::Interrupt),
253        _ = sigterm.recv() => Ok(TerminationSignal::Terminate),
254    }
255}
256
257#[cfg(not(unix))]
258async fn wait_for_signal_impl() -> Result<TerminationSignal, RuntimeSignalError> {
259    tokio::signal::ctrl_c()
260        .await
261        .map_err(|error| RuntimeSignalError::Install(error.to_string()))?;
262    Ok(TerminationSignal::Interrupt)
263}
264
265#[cfg(test)]
266mod tests {
267    use super::{ShutdownCoordinator, ShutdownPhaseStatus, TerminationSignal};
268    use std::time::Duration;
269
270    #[test]
271    fn termination_signal_reports_stable_labels() {
272        assert_eq!(TerminationSignal::Interrupt.as_str(), "SIGINT");
273        assert_eq!(TerminationSignal::Terminate.as_str(), "SIGTERM");
274    }
275
276    #[tokio::test]
277    async fn shutdown_coordinator_runs_all_phases_in_order() {
278        let mut coordinator = ShutdownCoordinator::new();
279        coordinator
280            .phase("tasks", None, || async { Ok(()) })
281            .phase("audit", None, || async { Err("flush failed".to_string()) })
282            .phase("db", None, || async { Ok(()) });
283
284        let report = coordinator.run().await;
285
286        assert_eq!(coordinator.len(), 3);
287        assert!(report.has_failures());
288        assert_eq!(report.phases[0].name, "tasks");
289        assert_eq!(report.phases[0].status, ShutdownPhaseStatus::Succeeded);
290        assert_eq!(
291            report.phases[1].status,
292            ShutdownPhaseStatus::Failed("flush failed".to_string())
293        );
294        assert_eq!(report.phases[2].status, ShutdownPhaseStatus::Succeeded);
295    }
296
297    #[test]
298    fn shutdown_report_logger_accepts_success_and_failure_reports() {
299        super::log_shutdown_report(&super::ShutdownReport::new(vec![
300            super::ShutdownPhaseReport {
301                name: "tasks",
302                status: ShutdownPhaseStatus::Succeeded,
303                duration: Duration::from_millis(1),
304            },
305        ]));
306
307        super::log_shutdown_report(&super::ShutdownReport::new(vec![
308            super::ShutdownPhaseReport {
309                name: "database",
310                status: ShutdownPhaseStatus::Failed("close failed".to_string()),
311                duration: Duration::from_millis(1),
312            },
313        ]));
314    }
315
316    #[tokio::test]
317    async fn shutdown_coordinator_reports_timeouts() {
318        let mut coordinator = ShutdownCoordinator::new();
319        coordinator.phase("slow", Some(Duration::from_millis(1)), || async {
320            tokio::time::sleep(Duration::from_millis(50)).await;
321            Ok(())
322        });
323
324        let report = coordinator.run().await;
325
326        assert!(report.has_failures());
327        assert_eq!(report.phases[0].status, ShutdownPhaseStatus::TimedOut);
328    }
329
330    #[tokio::test]
331    async fn shutdown_coordinator_supports_consumed_phase_handles() {
332        let mut coordinator = ShutdownCoordinator::new();
333        let mut owned_handle = Some("resource");
334        coordinator.phase("owned", None, move || {
335            let handle = owned_handle.take();
336            async move {
337                if handle == Some("resource") {
338                    Ok(())
339                } else {
340                    Err("resource already consumed".to_string())
341                }
342            }
343        });
344
345        let report = coordinator.run().await;
346
347        assert!(!report.has_failures());
348        assert_eq!(report.phases[0].status, ShutdownPhaseStatus::Succeeded);
349    }
350}
aster_forge_runtime/shutdown.rs

aster_forge_runtime/
shutdown.rs
