harmos_runtime/storage/
manager.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12/* === Re-Exports === */
13
14pub use harmos_storage::ErasedCatalog;
15pub(crate) use harmos_storage::encoded_size;
16
17/* === Imports === */
18
19// Standard Library Imports
20use std::sync::{Arc, Mutex};
21
22// Third Party Imports
23
24// Local Imports
25use super::{StorageBackends, StorageSettings};
26use crate::operations::{OperationManager, resident::Lease};
27use crate::services::{Fault, Once, Resident, Service, ServiceDefinition};
28use crate::supervision::executor;
29use crate::{Application, Error, Result};
30
31/* === Constants === */
32
33/* === Macros === */
34
35/* === Type Aliases === */
36
37/* === Structs === */
38
39pub struct StorageManager<C = ()> {
40    engine: harmos_storage::StorageManager<C>,
41}
42
43struct Worker(harmos_storage::Worker);
44
45struct Owner {
46    leases: Mutex<Vec<Lease>>,
47}
48
49/* === Enums === */
50
51/* === Traits === */
52
53/* === Implementations === */
54
55impl<C> Clone for StorageManager<C> {
56    fn clone(&self) -> Self {
57        Self { engine: self.engine.clone() }
58    }
59}
60
61impl<C> std::ops::Deref for StorageManager<C> {
62    type Target = harmos_storage::StorageManager<C>;
63    fn deref(&self) -> &Self::Target {
64        &self.engine
65    }
66}
67
68impl<C: super::ObjectCatalog> StorageManager<C> {
69    #[tracing::instrument(target = "harmos::storage", name = "storage.new", level = "debug", skip_all)]
70    pub fn new(backends: StorageBackends, settings: StorageSettings) -> Result<Self> {
71        let execution = harmos_storage::execution::Execution::new(executor::now, |duration| Box::pin(executor::sleep(duration)));
72        Ok(Self { engine: harmos_storage::StorageManager::with_execution(backends, settings, execution)? })
73    }
74}
75
76impl<C> StorageManager<C> {
77    pub(crate) fn claim_owner(&self, owner: Arc<()>) -> Result<()> {
78        self.engine.claim_owner(owner).map_err(Into::into)
79    }
80    #[tracing::instrument(target = "harmos::storage", name = "storage.acquire", level = "debug", skip_all)]
81    pub(crate) async fn acquire(&self) -> Result<()> {
82        self.engine.acquire().await.map_err(Into::into)
83    }
84    #[tracing::instrument(target = "harmos::storage", name = "storage.close", level = "debug", skip_all)]
85    pub async fn close(&self) -> Result<()> {
86        self.engine.close().await.map_err(Into::into)
87    }
88
89    pub(crate) fn erased(&self) -> StorageManager<ErasedCatalog> {
90        StorageManager { engine: self.engine.erased() }
91    }
92    pub(crate) fn same_instance<D>(&self, other: &StorageManager<D>) -> bool {
93        self.engine.same_instance(&other.engine)
94    }
95    #[tracing::instrument(target = "harmos::storage", name = "storage.start", level = "debug", skip_all)]
96    pub async fn start<A: Application>(&self, runtime: &crate::Runtime<A>) -> Result<()> {
97        self.start_on::<A>(&runtime.operations).await
98    }
99    #[tracing::instrument(target = "harmos::storage", name = "storage.start_on", level = "debug", skip_all)]
100    pub(crate) async fn start_on<A: Application>(&self, operations: &OperationManager) -> Result<()> {
101        self.engine
102            .start_with(operations.resident_identity(), |worker| {
103                let lease = operations
104                    .start_service::<A>(Once::new(ServiceDefinition::new("harmos-storage"), Worker(worker)), true, false)
105                    .map_err(|error| {
106                        harmos_storage::Error::backend("start storage worker", harmos_storage::BackendFailure::Permanent, error)
107                    })?;
108                Ok(Box::new(Owner { leases: Mutex::new(vec![lease]) }))
109            })
110            .await
111            .map_err(Into::into)
112    }
113}
114
115impl<A: Application> Resident<A> for Worker {
116    type Error = Error;
117    async fn cleanup(&mut self) -> Result<()> {
118        self.0.cleanup().await.map_err(Into::into)
119    }
120    async fn run(&mut self, mut context: Service<A>) -> std::result::Result<(), Fault<Error>> {
121        context.ready();
122        context.until_stopping(self.0.run()).await.unwrap_or(Ok(())).map_err(|error| Fault::Environmental(error.into()))
123    }
124}
125
126impl harmos_storage::WorkerOwner for Owner {
127    fn stop(&self) -> harmos_storage::StorageFuture<'_, ()> {
128        Box::pin(async move {
129            let receipts: Vec<_> = self.leases.lock().unwrap().iter_mut().map(Lease::retire).collect();
130            let mut failure = None;
131            for receipt in receipts {
132                if let Err(error) = receipt.await {
133                    failure.get_or_insert_with(|| {
134                        harmos_storage::Error::backend("stop storage worker", harmos_storage::BackendFailure::Permanent, error)
135                    });
136                }
137            }
138            failure.map_or(Ok(()), Err)
139        })
140    }
141}
142
143/* === Public Functions === */
144
145/* === Private Functions === */
146
147/* === Tests === */
harmos_runtime/storage/manager.rs

harmos_runtime/storage/
manager.rs
