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

harmos_storage/manager/
lifecycle.rs
