harmos_storage/state/file/
mod.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10// Durable JSON State Backend
11
12/* === Modules === */
13
14mod files;
15mod format;
16mod journal;
17
18/* === Re-Exports === */
19
20/* === Imports === */
21
22// Standard Library Imports
23use std::{
24    fs::{File, OpenOptions},
25    num::NonZeroUsize,
26    path::{Path, PathBuf},
27    sync::{
28        Arc,
29        atomic::{AtomicBool, Ordering},
30    },
31};
32// Third Party Imports
33use tokio::sync::Mutex;
34// Local Imports
35use super::{RecoveryEntry, RecoveryPage, RecoverySnapshot, StateBackend, StorageStateSettings};
36use crate::storage::{ReadLimits, RecoveryPosition, StorageBatch, StorageFuture, StoreHistory};
37use crate::{BackendFailure, Error, Result};
38use files::JournalOpener;
39use journal::StateLog;
40
41/* === Constants === */
42
43/* === Macros === */
44
45/* === Type Aliases === */
46
47/* === Structs === */
48
49pub struct FileBackend {
50    root: PathBuf,
51    history: StoreHistory,
52    settings: StorageStateSettings,
53    opener: JournalOpener,
54    journal: Arc<Mutex<Option<StateLog>>>,
55}
56
57struct CancelOpen(Arc<AtomicBool>);
58
59/* === Enums === */
60
61/* === Traits === */
62
63/* === Implementations === */
64
65impl Drop for CancelOpen {
66    fn drop(&mut self) {
67        self.0.store(true, Ordering::Release);
68    }
69}
70
71impl FileBackend {
72    pub fn new(root: PathBuf, history: StoreHistory, settings: StorageStateSettings) -> Self {
73        Self::with_journal_opener(root, history, settings, |path| {
74            OpenOptions::new().create(true).truncate(false).read(true).write(true).open(path)
75        })
76    }
77
78    // The Opener Must Return The Named Journal Without Truncating Existing Bytes.
79    // It Is Called Under The Exclusive Root Lease, Including After Replacement.
80    pub fn with_journal_opener(
81        root: PathBuf, history: StoreHistory, settings: StorageStateSettings,
82        opener: impl Fn(&Path) -> std::io::Result<File> + Send + Sync + 'static,
83    ) -> Self {
84        let opener = Arc::new(opener);
85        let journal = Mutex::new(None);
86        let journal = Arc::new(journal);
87        Self { root, history, settings, opener, journal }
88    }
89
90    #[tracing::instrument(name = "storage.state.file.mod.call", level = "debug", skip_all)]
91    async fn call<T: Send + 'static>(&self, action: impl FnOnce(&mut StateLog) -> Result<T> + Send + 'static) -> Result<T> {
92        // Acquire Before Dispatch So Cancellation Cannot Enqueue An Older Write.
93        let mut owner = tracing::Instrument::instrument(self.journal.clone().lock_owned(), tracing::trace_span!("storage.lock.wait")).await;
94        {
95            let context = crate::execution::Context::current();
96            tokio::task::spawn_blocking(move || {
97                context.blocking(|| {
98                    let journal = owner.as_mut().ok_or_else(|| format::invalid("state backend is closed"))?;
99                    journal.check()?;
100                    action(journal)
101                })
102            })
103        }
104        .await
105        .map_err(task)?
106    }
107}
108
109impl StateBackend for FileBackend {
110    fn history(&self) -> &StoreHistory {
111        &self.history
112    }
113
114    fn open(&self) -> StorageFuture<'_, RecoveryPosition> {
115        Box::pin(async move {
116            let context = crate::execution::Context::current();
117            let future = async move {
118                let span = tracing::debug_span!("storage.state.backend", method = "open", outcome = "cancelled");
119                let result = tracing::Instrument::instrument(
120                    async move {
121                        let mut owner =
122                            tracing::Instrument::instrument(self.journal.clone().lock_owned(), tracing::trace_span!("storage.lock.wait"))
123                                .await;
124                        let root = self.root.clone();
125                        let history = self.history.clone();
126                        let settings = self.settings;
127                        let opener = Arc::clone(&self.opener);
128                        let cancelled = Arc::new(AtomicBool::new(false));
129                        let _cancel = CancelOpen(Arc::clone(&cancelled));
130                        {
131                            let context = crate::execution::Context::current();
132                            tokio::task::spawn_blocking(move || {
133                                context.blocking(|| {
134                                    if owner.is_none() {
135                                        *owner = Some(StateLog::open(&root, history, settings, opener, &mut || {
136                                            if cancelled.load(Ordering::Acquire) {
137                                                Err(Error::Closing { domain: "cancelled state recovery" })
138                                            } else {
139                                                Ok(())
140                                            }
141                                        })?);
142                                    }
143                                    let journal = owner.as_ref().expect("opened journal");
144                                    journal.check()?;
145                                    Ok(journal.position())
146                                })
147                            })
148                        }
149                        .await
150                        .map_err(task)?
151                    },
152                    span.clone(),
153                )
154                .await;
155                span.record("outcome", if result.is_ok() { "completed" } else { "failed" });
156                result
157            };
158            context.scope(future).await
159        })
160    }
161
162    fn read_snapshot(&self, maximum: NonZeroUsize) -> StorageFuture<'_, Option<RecoverySnapshot>> {
163        Box::pin(async move {
164            let context = crate::execution::Context::current();
165            let future = async move {
166                let span = tracing::debug_span!("storage.state.backend", method = "read_snapshot", outcome = "cancelled");
167                let result = tracing::Instrument::instrument(self.call(move |journal| journal.snapshot(maximum.get())), span.clone()).await;
168                span.record("outcome", if result.is_ok() { "completed" } else { "failed" });
169                result
170            };
171            context.scope(future).await
172        })
173    }
174
175    fn write_snapshot(&self, snapshot: StorageBatch<RecoverySnapshot>) -> StorageFuture<'_, ()> {
176        Box::pin(async move {
177            let context = crate::execution::Context::current();
178            let future = async move {
179                let span = tracing::debug_span!("storage.state.backend", method = "write_snapshot", outcome = "cancelled");
180                let result = tracing::Instrument::instrument(
181                    async move {
182                        if snapshot.len() != 1 {
183                            return Err(format::invalid("snapshot write requires exactly one snapshot"));
184                        }
185                        self.history.check(&snapshot.as_slice()[0].through.history)?;
186                        self.call(move |journal| journal.checkpoint(&snapshot.as_slice()[0])).await
187                    },
188                    span.clone(),
189                )
190                .await;
191                span.record("outcome", if result.is_ok() { "completed" } else { "failed" });
192                result
193            };
194            context.scope(future).await
195        })
196    }
197
198    fn read_entries<'a>(
199        &'a self, after: &'a RecoveryPosition, through: &'a RecoveryPosition, limits: ReadLimits,
200    ) -> StorageFuture<'a, RecoveryPage> {
201        Box::pin(async move {
202            let context = crate::execution::Context::current();
203            let future = async move {
204                let span = tracing::debug_span!("storage.state.backend", method = "read_entries", outcome = "cancelled");
205                let result = tracing::Instrument::instrument(
206                    async move {
207                        self.history.check(&after.history)?;
208                        self.history.check(&through.history)?;
209                        let after = after.clone();
210                        let through = through.clone();
211                        self.call(move |journal| journal.page(&after, &through, limits)).await
212                    },
213                    span.clone(),
214                )
215                .await;
216                span.record("outcome", if result.is_ok() { "completed" } else { "failed" });
217                result
218            };
219            context.scope(future).await
220        })
221    }
222
223    fn write_entries(&self, entries: StorageBatch<RecoveryEntry>) -> StorageFuture<'_, ()> {
224        Box::pin(async move {
225            let context = crate::execution::Context::current();
226            let future = async move {
227                let span = tracing::debug_span!("storage.state.backend", method = "write_entries", outcome = "cancelled");
228                let result = tracing::Instrument::instrument(
229                    async move {
230                        for entry in entries.as_slice() {
231                            self.history.check(&entry.position.history)?;
232                        }
233                        self.call(move |journal| journal.append(entries.as_slice())).await
234                    },
235                    span.clone(),
236                )
237                .await;
238                span.record("outcome", if result.is_ok() { "completed" } else { "failed" });
239                result
240            };
241            context.scope(future).await
242        })
243    }
244
245    fn flush<'a>(&'a self, through: &'a RecoveryPosition) -> StorageFuture<'a, RecoveryPosition> {
246        Box::pin(async move {
247            let context = crate::execution::Context::current();
248            let future = async move {
249                let span = tracing::debug_span!("storage.state.backend", method = "flush", outcome = "cancelled");
250                let result = tracing::Instrument::instrument(
251                    async move {
252                        self.history.check(&through.history)?;
253                        let through = through.clone();
254                        self.call(move |journal| journal.flush(&through)).await
255                    },
256                    span.clone(),
257                )
258                .await;
259                span.record("outcome", if result.is_ok() { "completed" } else { "failed" });
260                result
261            };
262            context.scope(future).await
263        })
264    }
265
266    fn close(&self) -> StorageFuture<'_, ()> {
267        Box::pin(async move {
268            let context = crate::execution::Context::current();
269            let future = async move {
270                let span = tracing::debug_span!("storage.state.backend", method = "close", outcome = "cancelled");
271                let result = tracing::Instrument::instrument(
272                    async move {
273                        let mut owner =
274                            tracing::Instrument::instrument(self.journal.clone().lock_owned(), tracing::trace_span!("storage.lock.wait"))
275                                .await;
276                        {
277                            let context = crate::execution::Context::current();
278                            tokio::task::spawn_blocking(move || {
279                                context.blocking(|| {
280                                    let Some(mut journal) = owner.take() else {
281                                        return Ok(());
282                                    };
283                                    journal.check()?;
284                                    journal.flush(&journal.position())?;
285                                    Ok(())
286                                })
287                            })
288                        }
289                        .await
290                        .map_err(task)?
291                    },
292                    span.clone(),
293                )
294                .await;
295                span.record("outcome", if result.is_ok() { "completed" } else { "failed" });
296                result
297            };
298            context.scope(future).await
299        })
300    }
301}
302
303/* === Public Functions === */
304
305/* === Private Functions === */
306
307fn task(error: tokio::task::JoinError) -> Error {
308    Error::backend("state file task", BackendFailure::Permanent, error)
309}
310
311/* === Tests === */
harmos_storage/state/file/mod.rs

harmos_storage/state/file/
mod.rs
