harmos_runtime/runtime/
boot.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::{AtomicBool, Ordering};
20use std::sync::{Arc, Mutex, Weak};
21use std::task::{Context, Poll};
22use std::time::Duration;
23// Third Party Imports
24use tokio::sync::watch;
25// Local Imports
26use super::Runtime;
27use crate::storage::StorageManager;
28use crate::{Application, Result};
29
30/* === Constants === */
31
32/* === Macros === */
33
34/* === Type Aliases === */
35
36type StartupFuture<A> = Pin<Box<dyn Future<Output = std::result::Result<Runtime<A>, super::Incompatible>> + Send>>;
37type CleanupFuture = Pin<Box<dyn Future<Output = Result<()>> + Send>>;
38type ReconcileFuture = Pin<Box<dyn Future<Output = Result<()>> + Send>>;
39type InitializeFuture = Pin<Box<dyn Future<Output = std::result::Result<(), Box<dyn std::error::Error + Send + Sync>>> + Send>>;
40type Initialize<A> = Box<dyn FnOnce(super::Handle<A>) -> InitializeFuture + Send>;
41pub(super) type Reconciliation<A> =
42    Box<dyn FnOnce(super::StateManager<A>, StorageManager<crate::storage::ErasedCatalog>) -> ReconcileFuture + Send>;
43
44/* === Structs === */
45
46pub(super) struct Initializer<A: Application> {
47    pub(super) timeout: Duration,
48    pub(super) initialize: Initialize<A>,
49}
50
51#[derive(Clone)]
52pub(super) struct Cleanup {
53    close: Arc<dyn Fn() -> CleanupFuture + Send + Sync>,
54    receipt: Weak<CleanupState>,
55}
56
57pub struct Startup<A: Application> {
58    future: Option<StartupFuture<A>>,
59    cleanup: BootstrapCleanup,
60}
61
62#[derive(Clone)]
63pub struct BootstrapCleanup {
64    shared: Arc<CleanupState>,
65}
66
67struct CleanupState {
68    retained: Mutex<Retained>,
69    status: watch::Sender<BootstrapStatus>,
70}
71
72struct Retained {
73    owner: Option<Cleanup>,
74    failure: Option<Arc<crate::Error>>,
75}
76
77pub(super) struct Boot<A: Application> {
78    pub(super) application: Option<super::launch::ApplicationCleanup<A>>,
79    pub(super) identity: Arc<()>,
80    pub(super) storage: Option<StorageManager<crate::storage::ErasedCatalog>>,
81    pub(super) runtime: Option<Runtime<A>>,
82    pub(super) writer: Option<tokio::task::JoinHandle<()>>,
83    pub(super) state: Option<super::StateManager<A>>,
84    armed: bool,
85    cleanup: BootstrapCleanup,
86}
87
88/* === Enums === */
89
90#[derive(Clone, Copy, Debug, PartialEq, Eq)]
91pub enum BootstrapStatus {
92    Starting,
93    Cleaning,
94    Failed,
95    Settled,
96    HandedOff,
97}
98
99/* === Traits === */
100
101/* === Implementations === */
102
103impl<A: Application> Startup<A> {
104    pub(super) fn new(builder: super::Builder<A>, storage: Option<StorageManager<crate::storage::ErasedCatalog>>) -> Self {
105        let cleanup = builder.cleanup.clone();
106        let receipt = cleanup.clone();
107        let future = Box::pin(async move {
108            let outcome = builder.boot(storage.as_ref(), receipt.clone()).await;
109            receipt.settle_unowned();
110            outcome
111        });
112        Self { future: Some(future), cleanup }
113    }
114}
115
116impl<A: Application> Future for Startup<A> {
117    type Output = std::result::Result<Runtime<A>, super::Incompatible>;
118    fn poll(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Self::Output> {
119        let future = self.future.as_mut().expect("startup is polled before completion");
120        let outcome = future.as_mut().poll(context);
121        if outcome.is_ready() {
122            self.future.take();
123        }
124        outcome
125    }
126}
127
128impl<A: Application> Drop for Startup<A> {
129    fn drop(&mut self) {
130        // Drop Boot First So It Can Transfer Every Acquired Owner
131        self.future.take();
132        self.cleanup.settle_unowned();
133    }
134}
135
136impl BootstrapCleanup {
137    pub(super) fn new() -> Self {
138        let (status, _) = watch::channel(BootstrapStatus::Starting);
139        Self {
140            shared: Arc::new(CleanupState {
141                retained: Mutex::new(Retained { owner: None, failure: None }),
142                status,
143            }),
144        }
145    }
146
147    pub fn status(&self) -> BootstrapStatus {
148        *self.shared.status.borrow()
149    }
150
151    pub fn failure(&self) -> Option<Arc<crate::Error>> {
152        self.shared.retained.lock().expect("bootstrap cleanup lock").failure.clone()
153    }
154
155    pub async fn wait(&self) -> Result<()> {
156        let mut status = self.shared.status.subscribe();
157        loop {
158            let current = *status.borrow_and_update();
159            match current {
160                BootstrapStatus::Settled | BootstrapStatus::HandedOff => return Ok(()),
161                BootstrapStatus::Failed => {
162                    let source = {
163                        let retained = self.shared.retained.lock().expect("bootstrap cleanup lock");
164                        if self.status() != BootstrapStatus::Failed {
165                            continue;
166                        }
167                        retained.failure.clone().expect("failed cleanup retains its cause")
168                    };
169                    return Err(crate::Error::TaskFailed { domain: "bootstrap cleanup", source });
170                }
171                BootstrapStatus::Starting | BootstrapStatus::Cleaning => {}
172            }
173            status.changed().await.expect("receipt retains its status owner");
174        }
175    }
176
177    pub async fn retry(&self) -> Result<()> {
178        let owner = {
179            let mut retained = self.shared.retained.lock().expect("bootstrap cleanup lock");
180            if self.status() == BootstrapStatus::Failed {
181                let owner = retained.owner.clone().expect("failed cleanup owns resources");
182                retained.failure = None;
183                self.shared.status.send_replace(BootstrapStatus::Cleaning);
184                Some(owner)
185            } else {
186                None
187            }
188        };
189        if let Some(owner) = owner {
190            self.run(owner);
191        }
192        self.wait().await
193    }
194
195    fn run(&self, owner: Cleanup) {
196        tokio::spawn(async move {
197            let _ = owner.close().await;
198        });
199    }
200
201    pub(super) fn failed(&self, owner: Option<Cleanup>, failure: Arc<crate::Error>) {
202        let mut retained = self.shared.retained.lock().expect("bootstrap cleanup lock");
203        retained.owner = owner;
204        retained.failure = Some(failure);
205        self.shared.status.send_replace(BootstrapStatus::Failed);
206    }
207
208    fn cancelled(&self, owner: Cleanup) {
209        {
210            let mut retained = self.shared.retained.lock().expect("bootstrap cleanup lock");
211            retained.owner = Some(owner.clone());
212            self.shared.status.send_replace(BootstrapStatus::Cleaning);
213        }
214        self.run(owner);
215    }
216
217    pub(super) fn settle_unowned(&self) {
218        let retained = self.shared.retained.lock().expect("bootstrap cleanup lock");
219        if retained.owner.is_none() && self.status() == BootstrapStatus::Starting {
220            self.shared.status.send_replace(BootstrapStatus::Settled);
221        }
222    }
223}
224
225impl<A: Application> Initializer<A> {
226    pub(super) fn new<F, Fut, E>(timeout: Duration, initialize: F) -> Self
227    where
228        F: FnOnce(super::Handle<A>) -> Fut + Send + 'static,
229        Fut: Future<Output = std::result::Result<(), E>> + Send + 'static,
230        E: std::error::Error + Send + Sync + 'static,
231    {
232        Self {
233            timeout,
234            initialize: Box::new(move |handle| Box::pin(async move { initialize(handle).await.map_err(|source| Box::new(source) as _) })),
235        }
236    }
237}
238
239impl Cleanup {
240    pub(super) async fn close(&self) -> Result<()> {
241        let outcome = (self.close)().await;
242        let Some(receipt) = self.receipt.upgrade() else {
243            return outcome;
244        };
245        let mut retained = receipt.retained.lock().expect("bootstrap cleanup lock");
246        match outcome {
247            Ok(()) => {
248                retained.owner = None;
249                retained.failure = None;
250                receipt.status.send_replace(BootstrapStatus::Settled);
251                Ok(())
252            }
253            Err(error) => {
254                let source = Arc::new(error);
255                retained.failure = Some(Arc::clone(&source));
256                receipt.status.send_replace(BootstrapStatus::Failed);
257                Err(crate::Error::TaskFailed { domain: "bootstrap cleanup", source })
258            }
259        }
260    }
261}
262
263impl std::fmt::Debug for Cleanup {
264    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
265        formatter.debug_struct("Cleanup").finish_non_exhaustive()
266    }
267}
268
269impl<A: Application> Boot<A> {
270    pub(super) fn new(storage: Option<&StorageManager<crate::storage::ErasedCatalog>>, cleanup: BootstrapCleanup) -> Result<Self> {
271        let identity = Arc::new(());
272        if let Some(storage) = storage {
273            // Refusal must precede installing a cleanup owner: another runtime's
274            // manager must never be closed by this failed opening attempt.
275            storage.claim_owner(identity.clone())?;
276        }
277
278        Ok(Self {
279            application: None,
280            identity,
281            storage: storage.cloned(),
282            runtime: None,
283            writer: None,
284            state: None,
285            armed: true,
286            cleanup,
287        })
288    }
289
290    pub(super) async fn acquire(&self) -> Result<()> {
291        if let Some(storage) = &self.storage {
292            storage.acquire().await?;
293        }
294        Ok(())
295    }
296
297    pub(super) async fn close(&mut self) -> Result<()> {
298        if self.runtime.is_none()
299            && let Some(application) = &self.application
300        {
301            application.settle().await.map_err(|source| crate::Error::TaskFailed { domain: "application cleanup", source })?;
302        }
303        if self.runtime.is_none() {
304            if let Some(state) = self.state.take() {
305                state.seal();
306            }
307            if let Some(writer) = self.writer.as_mut() {
308                let _ = writer.await;
309            }
310            self.writer.take();
311        }
312        let result = if let Some(runtime) = &self.runtime {
313            runtime.shutdown().await.map(|_| ())
314        } else if let Some(storage) = &self.storage {
315            storage.close().await
316        } else {
317            Ok(())
318        };
319        // Operation finalization may refuse before the state is joined.
320        // Keep the original runtime until success or transfer it to the error.
321        self.armed = false;
322        if result.is_ok() {
323            self.runtime.take();
324        }
325        result
326    }
327
328    pub(super) async fn reject(mut self, source: super::Incompatible) -> super::Incompatible {
329        match self.close().await {
330            Ok(()) => {
331                self.cleanup.shared.status.send_replace(BootstrapStatus::Settled);
332                source
333            }
334            Err(failure) => {
335                let failure = Arc::new(failure);
336                let retained = self.retain_cleanup(true);
337                if retained.is_some() {
338                    self.cleanup.failed(retained.clone(), Arc::clone(&failure));
339                } else {
340                    self.cleanup.shared.status.send_replace(BootstrapStatus::Settled);
341                }
342                super::Incompatible {
343                    cause: super::Cause::Cleanup {
344                        source: Box::new(source),
345                        failure: crate::Error::TaskFailed {
346                            domain: "bootstrap cleanup",
347                            source: failure,
348                        },
349                    },
350                    cleanup: retained,
351                }
352            }
353        }
354    }
355
356    pub(super) fn retain_cleanup(&mut self, failure_reported: bool) -> Option<Cleanup> {
357        let state = self.state.take();
358        let writer = self.writer.take();
359        let runtime = self.runtime.take().filter(|runtime| !failure_reported || !physically_settled(runtime));
360        let application = self.application.take();
361        let storage = self.storage.take().filter(|storage| !storage.status().closed);
362        self.armed = false;
363        if runtime.is_none() && storage.is_none() && writer.is_none() && state.is_none() && application.is_none() {
364            return None;
365        }
366        let writer = Arc::new(tokio::sync::Mutex::new(writer));
367        let runtime = runtime.map(Arc::new);
368        let failure_reported = Arc::new(AtomicBool::new(false));
369        Some(Cleanup {
370            receipt: Arc::downgrade(&self.cleanup.shared),
371            close: Arc::new(move || {
372                let state = state.clone();
373                let writer = Arc::clone(&writer);
374                let runtime = runtime.clone();
375                let storage = storage.clone();
376                let application = application.clone();
377                let failure_reported = failure_reported.clone();
378                Box::pin(async move {
379                    if runtime.is_none() {
380                        if let Some(application) = application {
381                            application
382                                .settle()
383                                .await
384                                .map_err(|source| crate::Error::TaskFailed { domain: "application cleanup", source })?;
385                        }
386                        if let Some(state) = state {
387                            state.seal();
388                        }
389                        let mut writer = writer.lock().await;
390                        if let Some(task) = writer.as_mut() {
391                            let _ = task.await;
392                        }
393                        writer.take();
394                    }
395                    if let Some(runtime) = runtime {
396                        match runtime.shutdown().await {
397                            Err(source) if physically_settled(&runtime) => {
398                                if failure_reported.swap(true, Ordering::AcqRel) {
399                                    Ok(())
400                                } else {
401                                    Err(source)
402                                }
403                            }
404                            outcome => outcome.map(|_| ()),
405                        }
406                    } else if let Some(storage) = storage {
407                        storage.close().await
408                    } else {
409                        Ok(())
410                    }
411                })
412            }),
413        })
414    }
415
416    pub(super) fn finish(mut self) -> Runtime<A> {
417        self.armed = false;
418        self.cleanup.shared.status.send_replace(BootstrapStatus::HandedOff);
419        self.runtime.take().expect("successful boot owns a runtime")
420    }
421}
422
423impl<A: Application> Drop for Boot<A> {
424    fn drop(&mut self) {
425        if !self.armed {
426            return;
427        }
428        if let Some(owner) = self.retain_cleanup(false) {
429            self.cleanup.cancelled(owner);
430        }
431    }
432}
433
434/* === Public Functions === */
435
436/* === Private Functions === */
437
438fn physically_settled<A: Application>(runtime: &Runtime<A>) -> bool {
439    let stopping = runtime.stopping.lock().expect("runtime transition is healthy");
440    matches!(&*stopping, super::Stopping::Closing { outcome, .. } if outcome.get().is_some())
441}
442
443/* === Tests === */
harmos_runtime/runtime/boot.rs

harmos_runtime/runtime/
boot.rs
