harmos_storage/manager/
lifecycle.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
17
18// Third Party Imports
19
20// Local Imports
21use super::*;
22
23/* === Constants === */
24
25/* === Macros === */
26
27/* === Type Aliases === */
28
29/* === Structs === */
30
31/* === Enums === */
32
33/* === Traits === */
34
35/* === Implementations === */
36
37impl<C> StorageManager<C> {
38    // Acquire Backends Before Giving One Worker To The Host's Supervisor.
39    pub async fn start_with(&self, owner: Arc<()>, spawn: impl FnOnce(Worker) -> Result<Box<dyn WorkerOwner>>) -> Result<()> {
40        self.claim_owner(owner)?;
41        let mut lifecycle = self.inner.lifecycle.write().await;
42        {
43            let state = self.inner.state.lock().unwrap();
44            if state.status.closing {
45                return Err(Error::Closing { domain: "storage" });
46            }
47            if state.ready && lifecycle.startup_complete {
48                return Ok(());
49            }
50            if state.ready {
51                return Err(Error::AlreadyRegistered { domain: "storage", name: WORKER_NAME.into() });
52            }
53        }
54        self.acquire_locked(&mut lifecycle).await?;
55        let mut state = self.inner.state.lock().unwrap();
56        if state.status.closing {
57            return Err(Error::Closing { domain: "storage" });
58        }
59        let worker = spawn(Worker(self.erased()))?;
60        state.ready = true;
61        state.status.running = true;
62        drop(state);
63        lifecycle.owner = Some(worker);
64        lifecycle.startup_complete = true;
65        self.inner.notify();
66        Ok(())
67    }
68
69    pub fn settings(&self) -> &StorageSettings {
70        &self.inner.settings
71    }
72
73    pub async fn acquire(&self) -> Result<()> {
74        let mut lifecycle = self.inner.lifecycle.write().await;
75        self.acquire_locked(&mut lifecycle).await
76    }
77
78    pub(super) async fn acquire_locked(&self, lifecycle: &mut Lifecycle) -> Result<()> {
79        if self.inner.state.lock().unwrap().status.closing {
80            return Err(Error::Closing { domain: "storage" });
81        }
82        if !lifecycle.primary_open {
83            lifecycle.primary_touched = true;
84            let saved = self.inner.bounded(self.inner.primary.open()).await?;
85            self.inner.primary.history().check(&saved.history)?;
86            let mut state = self.inner.state.lock().unwrap();
87            state.status.accepted = saved.clone();
88            state.status.saved = saved;
89            state.data_attempted = state.status.accepted.sequence;
90            state.data_accepted.sequence = state.status.accepted.sequence;
91            lifecycle.primary_open = true;
92        }
93        if !lifecycle.files_recovered {
94            self.recover_files().await?;
95            lifecycle.files_recovered = true;
96        }
97        if !lifecycle.receipts_recovered {
98            self.recover_receipts().await?;
99            lifecycle.receipts_recovered = true;
100        }
101        if !lifecycle.recovery_open {
102            lifecycle.recovery_touched = true;
103            let saved = self.inner.bounded(self.inner.recovery.open()).await?;
104            self.inner.recovery.history().check(&saved.history)?;
105            let mut state = self.inner.state.lock().unwrap();
106            state.status.recovery_accepted = saved.clone();
107            state.status.recovery_saved = saved;
108            state.recovery_written.sequence = state.status.recovery_accepted.sequence;
109            state.recovery_accepted.sequence = state.status.recovery_accepted.sequence;
110            lifecycle.recovery_open = true;
111        }
112        if self.inner.state.lock().unwrap().status.closing {
113            return Err(Error::Closing { domain: "storage" });
114        }
115        Ok(())
116    }
117
118    pub fn same_instance<D>(&self, other: &StorageManager<D>) -> bool {
119        Arc::ptr_eq(&self.inner, &other.inner)
120    }
121
122    pub fn claim_owner(&self, owner: Arc<()>) -> Result<()> {
123        let mut state = self.inner.state.lock().unwrap();
124        if state.status.closing {
125            return Err(Error::Closing { domain: "storage" });
126        }
127        if let Some(current) = &state.service_owner {
128            if !Arc::ptr_eq(current, &owner) {
129                return Err(Error::AlreadyRegistered { domain: "storage owner", name: WORKER_NAME.to_owned() });
130            }
131        } else {
132            state.service_owner = Some(owner);
133        }
134        Ok(())
135    }
136
137    pub fn cursor_accepted(&self) -> StorageCursor {
138        self.inner.state.lock().unwrap().status.accepted.clone()
139    }
140
141    pub fn cursor_saved(&self) -> StorageCursor {
142        self.inner.state.lock().unwrap().status.saved.clone()
143    }
144
145    pub fn status(&self) -> StorageMetrics {
146        self.inner.state.lock().unwrap().status.clone()
147    }
148}
149
150/* === Public Functions === */
151
152/* === Private Functions === */
153
154/* === Tests === */
harmos_storage/manager/lifecycle.rs

harmos_storage/manager/
lifecycle.rs
