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::{Manager, resident::Lease};
27use crate::services::{Once, Resident, ServiceContext, 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 {
73            engine: harmos_storage::StorageManager::with_execution(backends, settings, execution)?,
74        })
75    }
76}
77
78impl<C> StorageManager<C> {
79    pub(crate) fn claim_owner(&self, owner: Arc<()>) -> Result<()> {
80        self.engine.claim_owner(owner).map_err(Into::into)
81    }
82    #[tracing::instrument(target = "harmos::storage", name = "storage.acquire", level = "debug", skip_all)]
83    pub(crate) async fn acquire(&self) -> Result<()> {
84        self.engine.acquire().await.map_err(Into::into)
85    }
86    #[tracing::instrument(target = "harmos::storage", name = "storage.close", level = "debug", skip_all)]
87    pub async fn close(&self) -> Result<()> {
88        self.engine.close().await.map_err(Into::into)
89    }
90
91    pub(crate) fn engine_erased(&self) -> harmos_storage::StorageManager<ErasedCatalog> {
92        self.engine.erased()
93    }
94    pub(crate) fn erased(&self) -> StorageManager<ErasedCatalog> {
95        StorageManager { engine: self.engine.erased() }
96    }
97    pub(crate) fn same_instance<D>(&self, other: &StorageManager<D>) -> bool {
98        self.engine.same_instance(&other.engine)
99    }
100    #[tracing::instrument(target = "harmos::storage", name = "storage.start", level = "debug", skip_all)]
101    pub async fn start<A: Application>(&self, runtime: &crate::Runtime<A>) -> Result<()> {
102        self.start_on::<A>(&runtime.operations).await
103    }
104    #[tracing::instrument(target = "harmos::storage", name = "storage.start_on", level = "debug", skip_all)]
105    pub(crate) async fn start_on<A: Application>(&self, operations: &Manager) -> Result<()> {
106        self.engine
107            .start_with(operations.resident_identity(), |worker| {
108                let lease = operations
109                    .start_service::<A>(Once::new(ServiceDefinition::new("harmos-storage"), Worker(worker)), true, false)
110                    .map_err(|error| {
111                        harmos_storage::Error::backend("start storage worker", harmos_storage::BackendFailure::Permanent, error)
112                    })?;
113                Ok(Box::new(Owner { leases: Mutex::new(vec![lease]) }))
114            })
115            .await
116            .map_err(Into::into)
117    }
118}
119
120impl<A: Application> Resident<A> for Worker {
121    type Error = Error;
122    async fn cleanup(&mut self) -> Result<()> {
123        self.0.cleanup().await.map_err(Into::into)
124    }
125    async fn run(&mut self, mut context: ServiceContext<A>) -> std::result::Result<(), Error> {
126        context.ready();
127        context.until_stopping(self.0.run()).await.unwrap_or(Ok(())).map_err(|error| error.into())
128    }
129}
130
131impl harmos_storage::WorkerOwner for Owner {
132    fn stop(&self) -> harmos_storage::StorageFuture<'_, ()> {
133        Box::pin(async move {
134            let receipts: Vec<_> = self.leases.lock().unwrap().iter_mut().map(Lease::retire).collect();
135            let mut failure = None;
136            for receipt in receipts {
137                if let Err(error) = receipt.await {
138                    failure.get_or_insert_with(|| {
139                        harmos_storage::Error::backend("stop storage worker", harmos_storage::BackendFailure::Permanent, error)
140                    });
141                }
142            }
143            failure.map_or(Ok(()), Err)
144        })
145    }
146}
147
148/* === Public Functions === */
149
150/* === Private Functions === */
151
152/* === Tests === */
harmos_runtime/storage/manager.rs

harmos_runtime/storage/
manager.rs
