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    #[tracing::instrument(name = "storage.manager.lifecycle.start_with", level = "debug", skip_all, fields(outcome = "cancelled"))]
40    pub async fn start_with(&self, owner: Arc<()>, spawn: impl FnOnce(Worker) -> Result<Box<dyn WorkerOwner>>) -> Result<()> {
41        let result = async {
42            self.claim_owner(owner)?;
43            let mut lifecycle =
44                tracing::Instrument::instrument(self.inner.lifecycle.write(), tracing::trace_span!("storage.lock.wait")).await;
45            {
46                let state = self.inner.state.lock().unwrap();
47                if state.status.closing {
48                    return Err(Error::Closing { domain: "storage" });
49                }
50                if state.ready && lifecycle.startup_complete {
51                    return Ok(());
52                }
53                if state.ready {
54                    return Err(Error::AlreadyRegistered { domain: "storage", name: WORKER_NAME.into() });
55                }
56            }
57            self.acquire_locked(&mut lifecycle).await?;
58            let mut state = self.inner.state.lock().unwrap();
59            if state.status.closing {
60                return Err(Error::Closing { domain: "storage" });
61            }
62            let worker = spawn(Worker(self.erased()))?;
63            state.ready = true;
64            state.status.running = true;
65            drop(state);
66            lifecycle.owner = Some(worker);
67            lifecycle.startup_complete = true;
68            self.inner.notify();
69            Ok(())
70        }
71        .await;
72        tracing::Span::current().record("outcome", if result.is_ok() { "completed" } else { "failed" });
73        result
74    }
75
76    pub fn settings(&self) -> &StorageSettings {
77        &self.inner.settings
78    }
79
80    #[tracing::instrument(name = "storage.manager.lifecycle.acquire", level = "debug", skip_all, fields(outcome = "cancelled"))]
81    pub async fn acquire(&self) -> Result<()> {
82        let result = async {
83            let mut lifecycle =
84                tracing::Instrument::instrument(self.inner.lifecycle.write(), tracing::trace_span!("storage.lock.wait")).await;
85            self.acquire_locked(&mut lifecycle).await
86        }
87        .await;
88        tracing::Span::current().record("outcome", if result.is_ok() { "completed" } else { "failed" });
89        result
90    }
91
92    #[tracing::instrument(name = "storage.manager.lifecycle.acquire_locked", level = "debug", skip_all)]
93    pub(super) async fn acquire_locked(&self, lifecycle: &mut Lifecycle) -> Result<()> {
94        if self.inner.state.lock().unwrap().status.closing {
95            return Err(Error::Closing { domain: "storage" });
96        }
97        if !lifecycle.primary_open {
98            lifecycle.primary_touched = true;
99            let saved = self.inner.bounded("primary.open", self.inner.primary.open()).await?;
100            self.inner.primary.history().check(&saved.history)?;
101            let mut state = self.inner.state.lock().unwrap();
102            state.status.accepted = saved.clone();
103            state.status.saved = saved;
104            state.data_attempted = state.status.accepted.sequence;
105            state.data_accepted.sequence = state.status.accepted.sequence;
106            lifecycle.primary_open = true;
107        }
108        if !lifecycle.files_recovered {
109            self.recover_files().await?;
110            lifecycle.files_recovered = true;
111        }
112        if !lifecycle.receipts_recovered {
113            self.recover_receipts().await?;
114            lifecycle.receipts_recovered = true;
115        }
116        if !lifecycle.recovery_open {
117            lifecycle.recovery_touched = true;
118            let saved = self.inner.bounded("recovery.open", self.inner.recovery.open()).await?;
119            self.inner.recovery.history().check(&saved.history)?;
120            let mut state = self.inner.state.lock().unwrap();
121            state.status.recovery_accepted = saved.clone();
122            state.status.recovery_saved = saved;
123            state.recovery_written.sequence = state.status.recovery_accepted.sequence;
124            state.recovery_accepted.sequence = state.status.recovery_accepted.sequence;
125            lifecycle.recovery_open = true;
126        }
127        if self.inner.state.lock().unwrap().status.closing {
128            return Err(Error::Closing { domain: "storage" });
129        }
130        Ok(())
131    }
132
133    pub fn same_instance<D>(&self, other: &StorageManager<D>) -> bool {
134        Arc::ptr_eq(&self.inner, &other.inner)
135    }
136
137    #[tracing::instrument(name = "storage.manager.lifecycle.claim_owner", level = "trace", skip_all)]
138    pub fn claim_owner(&self, owner: Arc<()>) -> Result<()> {
139        let mut state = self.inner.state.lock().unwrap();
140        if state.status.closing {
141            return Err(Error::Closing { domain: "storage" });
142        }
143        if let Some(current) = &state.service_owner {
144            if !Arc::ptr_eq(current, &owner) {
145                return Err(Error::AlreadyRegistered { domain: "storage owner", name: WORKER_NAME.to_owned() });
146            }
147        } else {
148            state.service_owner = Some(owner);
149        }
150        Ok(())
151    }
152
153    pub fn cursor_accepted(&self) -> StorageCursor {
154        self.inner.state.lock().unwrap().status.accepted.clone()
155    }
156
157    pub fn cursor_saved(&self) -> StorageCursor {
158        self.inner.state.lock().unwrap().status.saved.clone()
159    }
160
161    #[tracing::instrument(name = "storage.manager.lifecycle.status", level = "trace", skip_all)]
162    pub fn status(&self) -> StorageMetrics {
163        let mut state = self.inner.state.lock().unwrap();
164        state.collect();
165        state.status.clone()
166    }
167}
168
169/* === Public Functions === */
170
171/* === Private Functions === */
172
173/* === Tests === */
harmos_storage/manager/lifecycle.rs

harmos_storage/manager/
lifecycle.rs
