siderite_core/lifespan/
managed.rs

1//! A single cleanup supervisor owns hooks across caller cancellation.
2
3use super::Lifespan;
4use crate::{ApiError, ServerError};
5use std::time::Duration;
6use tokio::sync::oneshot;
7use tokio::task::JoinHandle;
8use tokio::time::Instant;
9
10type Outcome = Result<(), ServerError>;
11type Ready = oneshot::Receiver<Outcome>;
12type Worker = JoinHandle<Outcome>;
13
14/// Owns a running lifespan and its cancellation-safe cleanup supervisor.
15///
16/// Dropping this owner requests teardown. Cleanup continues on its existing
17/// supervisor until the deadline; no per-resource cleanup tasks are spawned.
18/// The Tokio runtime must stay alive for cleanup to run. A process crash or
19/// runtime destruction cannot guarantee asynchronous teardown.
20#[derive(Debug)]
21pub struct ManagedLifespan {
22    ready: Option<Ready>,
23    stop: Option<oneshot::Sender<Instant>>,
24    worker: Option<Worker>,
25    started: bool,
26    budget: Duration,
27}
28
29impl ManagedLifespan {
30    pub(super) fn new(mut lifespan: Lifespan) -> Result<Self, ServerError> {
31        let runtime = tokio::runtime::Handle::try_current().map_err(|_| {
32            let message = "supervision needs a Tokio runtime";
33            ServerError::Configuration(message.into())
34        })?;
35        let budget = lifespan.shutdown_budget;
36        let (ready_tx, ready) = oneshot::channel();
37        let (stop, mut stop_rx) = oneshot::channel();
38        let worker = runtime.spawn(async move {
39            let startup = tokio::select! {
40                result = lifespan.start_hooks() => result,
41                deadline = &mut stop_rx => {
42                    let deadline = deadline.unwrap_or_else(|_| {
43                        Instant::now() + budget
44                    });
45                    let _ = ready_tx.send(Err(ServerError::Configuration(
46                        "startup was cancelled".into(),
47                    )));
48                    return lifespan.shutdown_at(deadline).await;
49                }
50            };
51            match startup {
52                Ok(()) => {
53                    if ready_tx.send(Ok(())).is_err() {
54                        let at = Instant::now() + budget;
55                        return lifespan.shutdown_at(at).await;
56                    }
57                    let received = stop_rx.await;
58                    let at = received.unwrap_or_else(|_| Instant::now() + budget);
59                    lifespan.shutdown_at(at).await
60                }
61                Err(error) => {
62                    // Preserve the startup error after all cleanup attempts.
63                    let at = Instant::now() + budget;
64                    let _ = lifespan.shutdown_at(at).await;
65                    let _ = ready_tx.send(Err(error));
66                    Ok(())
67                }
68            }
69        });
70        Ok(Self {
71            ready: Some(ready),
72            stop: Some(stop),
73            worker: Some(worker),
74            started: false,
75            budget,
76        })
77    }
78
79    /// Wait for successful startup, including cleanup on startup failure.
80    ///
81    /// Args:
82    ///     self: The supervisor owner.
83    ///
84    /// Returns:
85    ///     Success after initialization, or the original startup error.
86    ///
87    /// # Errors
88    /// Startup or supervisor failure. Dropping this future requests cleanup
89    /// only if its owner is dropped too; otherwise call [`Self::shutdown`].
90    pub async fn ready(&mut self) -> Result<(), ServerError> {
91        if self.started {
92            return Ok(());
93        }
94        let receiver = self.ready.take().ok_or_else(|| {
95            let message = "readiness consumed";
96            ServerError::Configuration(message.into())
97        })?;
98        receiver.await.map_err(|_| supervisor_error())??;
99        self.started = true;
100        Ok(())
101    }
102
103    /// Request shutdown and await reverse-order, bounded resource cleanup.
104    ///
105    /// Args:
106    ///     self: Owner consumed by shutdown.
107    ///
108    /// Returns:
109    ///     The first cleanup error, if any.
110    ///
111    /// # Errors
112    /// Cleanup failure or exceeded budget. Cancelling this future leaves the
113    /// supervisor running its bounded cleanup.
114    pub async fn shutdown(self) -> Result<(), ServerError> {
115        let deadline = Instant::now() + self.budget;
116        self.shutdown_at(deadline).await
117    }
118
119    pub(crate) async fn shutdown_at(mut self, at: Instant) -> Outcome {
120        if let Some(stop) = self.stop.take() {
121            let _ = stop.send(at);
122        }
123        let worker = self.worker.take().ok_or_else(supervisor_error)?;
124        worker.await.map_err(|_| supervisor_error())?
125    }
126}
127
128fn supervisor_error() -> ServerError {
129    ServerError::Lifespan(ApiError::internal("lifespan supervisor failed"))
130}
siderite_core/lifespan/managed.rs

siderite_core/lifespan/
managed.rs
