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 {
142            shared: Arc::new(CleanupState {
143                retained: Mutex::new(Retained { owner: None, failure: None }),
144                status,
145            }),
146        }
147    }
148
149    pub fn status(&self) -> BootstrapStatus {
150        *self.shared.status.borrow()
151    }
152
153    pub fn failure(&self) -> Option<Arc<crate::Error>> {
154        self.shared.retained.lock().expect("bootstrap cleanup lock").failure.clone()
155    }
156
157    pub async fn wait(&self) -> Result<()> {
158        let mut status = self.shared.status.subscribe();
159        loop {
160            let current = *status.borrow_and_update();
161            match current {
162                BootstrapStatus::Settled | BootstrapStatus::HandedOff => return Ok(()),
163                BootstrapStatus::Failed => {
164                    let source = {
165                        let retained = self.shared.retained.lock().expect("bootstrap cleanup lock");
166                        if self.status() != BootstrapStatus::Failed {
167                            continue;
168                        }
169                        retained.failure.clone().expect("failed cleanup retains its cause")
170                    };
171                    return Err(crate::Error::TaskFailed { domain: "bootstrap cleanup", source });
172                }
173                BootstrapStatus::Starting | BootstrapStatus::Cleaning => {}
174            }
175            status.changed().await.expect("receipt retains its status owner");
176        }
177    }
178
179    pub async fn retry(&self) -> Result<()> {
180        let owner = {
181            let mut retained = self.shared.retained.lock().expect("bootstrap cleanup lock");
182            if self.status() == BootstrapStatus::Failed {
183                let owner = retained.owner.clone().expect("failed cleanup owns resources");
184                retained.failure = None;
185                self.shared.status.send_replace(BootstrapStatus::Cleaning);
186                Some(owner)
187            } else {
188                None
189            }
190        };
191        if let Some(owner) = owner {
192            self.run(owner);
193        }
194        self.wait().await
195    }
196
197    fn run(&self, owner: Cleanup) {
198        tokio::spawn(async move {
199            let _ = owner.close().await;
200        });
201    }
202
203    pub(super) fn failed(&self, owner: Option<Cleanup>, failure: Arc<crate::Error>) {
204        let mut retained = self.shared.retained.lock().expect("bootstrap cleanup lock");
205        retained.owner = owner;
206        retained.failure = Some(failure);
207        self.shared.status.send_replace(BootstrapStatus::Failed);
208    }
209
210    fn cancelled(&self, owner: Cleanup) {
211        {
212            let mut retained = self.shared.retained.lock().expect("bootstrap cleanup lock");
213            retained.owner = Some(owner.clone());
214            self.shared.status.send_replace(BootstrapStatus::Cleaning);
215        }
216        self.run(owner);
217    }
218
219    fn settle_unowned(&self) {
220        let retained = self.shared.retained.lock().expect("bootstrap cleanup lock");
221        if retained.owner.is_none() && self.status() == BootstrapStatus::Starting {
222            self.shared.status.send_replace(BootstrapStatus::Settled);
223        }
224    }
225}
226
227impl<A: Application> Initializer<A> {
228    pub(super) fn new<F, Fut, E>(timeout: Duration, initialize: F) -> Self
229    where
230        F: FnOnce(super::Handle<A>) -> Fut + Send + 'static,
231        Fut: Future<Output = std::result::Result<(), E>> + Send + 'static,
232        E: std::error::Error + Send + Sync + 'static,
233    {
234        Self {
235            timeout,
236            initialize: Box::new(move |handle| Box::pin(async move { initialize(handle).await.map_err(|source| Box::new(source) as _) })),
237        }
238    }
239}
240
241impl Cleanup {
242    pub(super) async fn close(&self) -> Result<()> {
243        let outcome = (self.close)().await;
244        let Some(receipt) = self.receipt.upgrade() else {
245            return outcome;
246        };
247        let mut retained = receipt.retained.lock().expect("bootstrap cleanup lock");
248        match outcome {
249            Ok(()) => {
250                retained.owner = None;
251                retained.failure = None;
252                receipt.status.send_replace(BootstrapStatus::Settled);
253                Ok(())
254            }
255            Err(error) => {
256                let source = Arc::new(error);
257                retained.failure = Some(Arc::clone(&source));
258                receipt.status.send_replace(BootstrapStatus::Failed);
259                Err(crate::Error::TaskFailed { domain: "bootstrap cleanup", source })
260            }
261        }
262    }
263}
264
265impl std::fmt::Debug for Cleanup {
266    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
267        formatter.debug_struct("Cleanup").finish_non_exhaustive()
268    }
269}
270
271impl<A: Application> Boot<A> {
272    pub(super) fn new(storage: Option<&StorageManager<crate::storage::ErasedCatalog>>, cleanup: BootstrapCleanup) -> Result<Self> {
273        let identity = Arc::new(());
274        if let Some(storage) = storage {
275            // Refusal must precede installing a cleanup owner: another runtime's
276            // manager must never be closed by this failed opening attempt.
277            storage.claim_owner(identity.clone())?;
278        }
279
280        Ok(Self {
281            identity,
282            storage: storage.cloned(),
283            runtime: None,
284            writer: None,
285            state: None,
286            armed: true,
287            cleanup,
288        })
289    }
290
291    pub(super) async fn acquire(&self) -> Result<()> {
292        if let Some(storage) = &self.storage {
293            storage.acquire().await?;
294        }
295        Ok(())
296    }
297
298    pub(super) async fn close(&mut self) -> Result<()> {
299        if self.runtime.is_none() {
300            if let Some(state) = self.state.take() {
301                state.seal();
302            }
303            if let Some(writer) = self.writer.as_mut() {
304                let _ = writer.await;
305            }
306            self.writer.take();
307        }
308        let result = if let Some(runtime) = &self.runtime {
309            runtime.stop().await.map(|_| ())
310        } else if let Some(storage) = &self.storage {
311            storage.close().await
312        } else {
313            Ok(())
314        };
315        // Operation finalization may refuse before the state is joined.
316        // Keep the original runtime until success or transfer it to the error.
317        self.armed = false;
318        if result.is_ok() {
319            self.runtime.take();
320        }
321        result
322    }
323
324    pub(super) fn retain_cleanup(&mut self) -> Option<Cleanup> {
325        if let Some(runtime) = self.runtime.take() {
326            let runtime = Arc::new(runtime);
327            return Some(Cleanup {
328                receipt: Arc::downgrade(&self.cleanup.shared),
329                close: Arc::new(move || {
330                    let runtime = Arc::clone(&runtime);
331                    Box::pin(async move { runtime.stop().await.map(|_| ()) })
332                }),
333            });
334        }
335        self.storage.take().map(|storage| Cleanup {
336            receipt: Arc::downgrade(&self.cleanup.shared),
337            close: Arc::new(move || {
338                let storage = storage.clone();
339                Box::pin(async move { storage.close().await })
340            }),
341        })
342    }
343
344    pub(super) fn finish(mut self) -> Runtime<A> {
345        self.armed = false;
346        self.cleanup.shared.status.send_replace(BootstrapStatus::HandedOff);
347        self.runtime.take().expect("successful boot owns a runtime")
348    }
349}
350
351impl<A: Application> Drop for Boot<A> {
352    fn drop(&mut self) {
353        if !self.armed {
354            return;
355        }
356        let state = self.state.take();
357        let writer = self.writer.take();
358        let runtime = self.runtime.take();
359        let storage = self.storage.take();
360        if runtime.is_none() && storage.is_none() && writer.is_none() {
361            return;
362        }
363        // Retain Partial Writer And StateManager Ownership Alongside The Full Runtime
364        let writer = Arc::new(tokio::sync::Mutex::new(writer));
365        let runtime = runtime.map(Arc::new);
366        let owner = Cleanup {
367            receipt: Arc::downgrade(&self.cleanup.shared),
368            close: Arc::new(move || {
369                let state = state.clone();
370                let writer = Arc::clone(&writer);
371                let runtime = runtime.clone();
372                let storage = storage.clone();
373                Box::pin(async move {
374                    if runtime.is_none() {
375                        if let Some(state) = state {
376                            state.seal();
377                        }
378                        let mut writer = writer.lock().await;
379                        if let Some(task) = writer.as_mut() {
380                            let _ = task.await;
381                        }
382                        writer.take();
383                    }
384                    if let Some(runtime) = runtime {
385                        runtime.stop().await.map(|_| ())
386                    } else if let Some(storage) = storage {
387                        storage.close().await
388                    } else {
389                        Ok(())
390                    }
391                })
392            }),
393        };
394        self.cleanup.cancelled(owner);
395    }
396}
397
398/* === Public Functions === */
399
400/* === Private Functions === */
401
402/* === Tests === */
harmos_runtime/runtime/boot.rs

harmos_runtime/runtime/
boot.rs
