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

harmos_runtime/runtime/
boot.rs
