harmos_runtime/runtime/
lifecycle.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12/* === Re-Exports === */
13
14/* === Imports === */
15
16// Standard Library Imports
17use std::future::Future;
18use std::pin::Pin;
19use std::sync::atomic::Ordering;
20use std::sync::{Arc, OnceLock, Weak};
21use std::task::Poll;
22use std::time::Duration;
23// Third Party Imports
24use tokio::sync::watch;
25use tokio::task::JoinHandle;
26// Local Imports
27use super::{Handle, Participants, Runtime, STOP_GRACE, Stopping};
28use crate::storage::StorageManager;
29use crate::supervision::cleanup::{Cleanup, Finalize, Resource};
30use crate::{Application, Error, Result, state::Position};
31
32/* === Constants === */
33
34/* === Macros === */
35
36/* === Type Aliases === */
37
38pub(super) type Outcome = std::result::Result<Position, Failure>;
39
40/* === Structs === */
41
42pub(super) struct Shutdown<A: Application> {
43    handles: Handle<A>,
44    participants: Participants,
45    journal_handles: Weak<()>,
46    notices: watch::Sender<Option<String>>,
47    state: Drain,
48    outcome: Arc<OnceLock<Outcome>>,
49    progress: watch::Sender<&'static str>,
50}
51
52#[derive(Clone)]
53pub(super) struct Failure {
54    domain: &'static str,
55    source: Arc<dyn std::error::Error + Send + Sync>,
56}
57
58/* === Enums === */
59
60enum Drain {
61    Running(JoinHandle<Position>),
62    Drained(Outcome),
63}
64
65/* === Traits === */
66
67/* === Implementations === */
68
69impl<A: Application> Runtime<A> {
70    pub async fn start_storage<C>(&self, storage: StorageManager<C>) -> Result<()> {
71        let storage = storage.erased();
72        let storage = {
73            let _stopping = self.stopping.lock().expect("runtime transition is healthy");
74            if self.closing.load(Ordering::Acquire) {
75                return Err(Error::Closing { domain: "runtime" });
76            }
77            if let Some(current) = self.storage.get()
78                && !current.same_instance(&storage)
79            {
80                return Err(Error::AlreadyRegistered { domain: "runtime storage", name: "primary".to_owned() });
81            }
82            storage.claim_owner(self.operations.resident_identity())?;
83            self.storage.get_or_init(|| storage)
84        };
85        storage.start_on::<A>(&self.operations).await
86    }
87
88    pub fn storage(&self) -> Option<&StorageManager<crate::storage::ErasedCatalog>> {
89        self.storage.get()
90    }
91
92    pub async fn stop_with_timeout(&self, timeout: Duration) -> Result<Position> {
93        let deadline = tokio::time::Instant::now()
94            .checked_add(timeout)
95            .filter(|_| !timeout.is_zero())
96            .ok_or(Error::InvalidInput { field: "runtime shutdown timeout", reason: "must be positive and representable" })?;
97        match tokio::time::timeout_at(deadline, self.stop()).await {
98            Ok(result) => result,
99            Err(_) => {
100                let stopping = self.stopping.lock().expect("runtime transition is healthy");
101                let phase = match &*stopping {
102                    Stopping::Running(_) => "admission",
103                    Stopping::Closing { progress, .. } => *progress.borrow(),
104                };
105                tracing::error!(phase, ?timeout, "runtime shutdown budget exhausted; original cleanup retained");
106                Err(Error::ShutdownTimeout { phase })
107            }
108        }
109    }
110
111    pub async fn stop(&self) -> Result<Position> {
112        self.closing.store(true, Ordering::Release);
113        #[cfg(feature = "grpc")]
114        self.grpc.close_admission();
115        #[cfg(all(feature = "plugin", not(target_arch = "wasm32")))]
116        self.plugin.close_admission();
117        #[cfg(not(all(feature = "plugin", not(target_arch = "wasm32"))))]
118        {
119            // Notify Residents Before Cancelling Their In-Flight Calls
120            self.operations.close_resident_admission();
121            self.operations.close_admission();
122            #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
123            {
124                self.records.stop_listeners();
125                self.streams.stop_listeners();
126            }
127        }
128        let (cleanup, completed) = {
129            let mut stopping = self.stopping.lock().expect("runtime transition is healthy");
130            if let Stopping::Running(supervisor) = &mut *stopping {
131                let outcome = Arc::new(OnceLock::new());
132                let progress = watch::channel("producers and listeners").0;
133                let original = Shutdown {
134                    handles: self.handle(),
135                    participants: self.participants.clone(),
136                    journal_handles: self.journal_handles.clone(),
137                    notices: self.notices.clone(),
138                    state: Drain::Running(supervisor.take().expect("runtime owns its supervisor")),
139                    outcome: outcome.clone(),
140                    progress: progress.clone(),
141                };
142                *stopping = Stopping::Closing { cleanup: Arc::new(Cleanup::new(original)), outcome, progress };
143            }
144            let Stopping::Closing { cleanup, outcome, .. } = &*stopping else {
145                unreachable!("shutdown retains its driver before awaiting it");
146            };
147            (cleanup.clone(), outcome.clone())
148        };
149        cleanup
150            .settle()
151            .await
152            .map_err(|source| Arc::try_unwrap(source).unwrap_or_else(|source| Error::TaskFailed { domain: "runtime cleanup", source }))?;
153        outcome(completed.get().expect("physical shutdown published its outcome"))
154    }
155}
156
157impl<A: Application> Resource for Shutdown<A> {
158    fn finalize(&mut self) -> Finalize<'_> {
159        Box::pin(async move { self.finish().await.map_err(Arc::new) })
160    }
161}
162
163impl<A: Application> Shutdown<A> {
164    async fn finish(&mut self) -> Result<()> {
165        if let Drain::Running(supervisor) = &mut self.state {
166            #[cfg(all(feature = "plugin", not(target_arch = "wasm32")))]
167            {
168                self.progress.send_replace("plugins");
169                self.handles.plugin.shutdown().await?;
170            }
171            self.handles.operations.close_resident_admission();
172            self.handles.operations.close_admission();
173            #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
174            {
175                self.handles.records.stop_listeners();
176                self.handles.streams.stop_listeners();
177            }
178            self.progress.send_replace("observers and listeners");
179            let observers = self.handles.operations.shutdown_observers();
180            #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
181            let (observers, records, streams) =
182                tokio::join!(observers, self.handles.records.shutdown_listeners(), self.handles.streams.shutdown_listeners(),);
183            #[cfg(not(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar")))]
184            let observers = observers.await;
185            if self.handles.operations.observer_cleanup_pending() {
186                return Err(observers.err().unwrap_or(Error::NotReady { domain: "observer cleanup" }));
187            }
188            #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
189            records?;
190            #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
191            streams?;
192            self.progress.send_replace("application services");
193            let services = self.handles.operations.shutdown_application_services().await;
194            if self.handles.operations.service_cleanup_pending(false) {
195                return Err(services.err().unwrap_or(Error::NotReady { domain: "service cleanup" }));
196            }
197            // Original SDK Shutdown Keeps Its Dispatcher Until Application Cleanup Settles.
198            self.progress.send_replace("operations");
199            self.handles.operations.shutdown().await?;
200            self.progress.send_replace("participants");
201            self.participants.stop().await;
202            self.handles.state.seal();
203            self.progress.send_replace("journal");
204            let mut settled = joined(supervisor, &self.journal_handles, &self.notices).await;
205            if let Err(source) = observers
206                && settled.is_ok()
207            {
208                settled = Err(Failure { domain: "observer", source: Arc::new(source) });
209            }
210            if let Err(source) = services
211                && settled.is_ok()
212            {
213                settled = Err(Failure { domain: "service", source: Arc::new(source) });
214            }
215            self.handles.streams.close();
216            self.state = Drain::Drained(settled);
217        }
218        let Drain::Drained(settled) = &self.state else {
219            unreachable!("the journal is drained before storage closes");
220        };
221        let mut settled = settled.clone();
222        if let Some(storage) = self.handles.storage() {
223            self.progress.send_replace("storage");
224            storage.close().await?;
225        }
226        self.progress.send_replace("infrastructure services");
227        let services = self.handles.operations.shutdown_residents().await;
228        if self.handles.operations.service_cleanup_pending(true) {
229            return Err(services.err().unwrap_or(Error::NotReady { domain: "service cleanup" }));
230        }
231        if let Err(source) = services
232            && settled.is_ok()
233        {
234            settled = Err(Failure { domain: "service", source: Arc::new(source) });
235        }
236        self.outcome.set(settled).unwrap_or_else(|_| unreachable!("shutdown publishes its outcome once"));
237        Ok(())
238    }
239}
240
241impl<A: Application> Handle<A> {
242    pub fn storage(&self) -> Option<&StorageManager<crate::storage::ErasedCatalog>> {
243        self.storage.get()
244    }
245}
246
247/* === Public Functions === */
248
249/* === Private Functions === */
250
251async fn joined(supervisor: &mut JoinHandle<Position>, journal_handles: &Weak<()>, notices: &watch::Sender<Option<String>>) -> Outcome {
252    let mut grace = Box::pin(tokio::time::sleep(STOP_GRACE));
253    let joined = std::future::poll_fn(|context| {
254        if let Poll::Ready(settled) = Pin::new(&mut *supervisor).poll(context) {
255            return Poll::Ready(Some(settled));
256        }
257        if grace.as_mut().poll(context).is_ready() {
258            return Poll::Ready(None);
259        }
260
261        Poll::Pending
262    })
263    .await;
264
265    match joined {
266        Some(settled) => settled,
267        None => {
268            let live = journal_handles.strong_count();
269            notices.send_replace(Some(format!(
270                "the journal stop is still draining; {live} StateManager clone{} outstanding, none \
271                     of which holds it open",
272                if live == 1 { "" } else { "s" }
273            )));
274
275            supervisor.await
276        }
277    }
278    .map_err(|source| Failure { domain: "journal supervisor", source: Arc::new(source) })
279}
280
281fn outcome(settled: &Outcome) -> Result<Position> {
282    match settled {
283        Ok(position) => Ok(*position),
284        Err(failure) => Err(Error::TaskFailed { domain: failure.domain, source: failure.source.clone() }),
285    }
286}
287
288/* === Tests === */
harmos_runtime/runtime/lifecycle.rs

harmos_runtime/runtime/
lifecycle.rs
