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::Arc,
28};
29// Third Party Imports
30use tokio::sync::Mutex;
31// Local Imports
32use super::{RecoveryEntry, RecoveryPage, RecoverySnapshot, StateBackend, StorageStateSettings};
33use crate::storage::{ReadLimits, RecoveryPosition, StorageFuture, StoreHistory};
34use crate::{BackendFailure, Error, Result};
35use files::JournalOpener;
36use journal::StateLog;
37
38/* === Constants === */
39
40/* === Macros === */
41
42/* === Type Aliases === */
43
44/* === Structs === */
45
46pub struct FileBackend {
47    root: PathBuf,
48    history: StoreHistory,
49    settings: StorageStateSettings,
50    opener: JournalOpener,
51    journal: Arc<Mutex<Option<StateLog>>>,
52}
53
54/* === Enums === */
55
56/* === Traits === */
57
58/* === Implementations === */
59
60impl FileBackend {
61    pub fn new(root: PathBuf, history: StoreHistory, settings: StorageStateSettings) -> Self {
62        Self::with_journal_opener(root, history, settings, |path| {
63            OpenOptions::new().create(true).truncate(false).read(true).write(true).open(path)
64        })
65    }
66
67    // The Opener Must Return The Named Journal Without Truncating Existing Bytes.
68    // It Is Called Under The Exclusive Root Lease, Including After Replacement.
69    pub fn with_journal_opener(
70        root: PathBuf, history: StoreHistory, settings: StorageStateSettings,
71        opener: impl Fn(&Path) -> std::io::Result<File> + Send + Sync + 'static,
72    ) -> Self {
73        let opener = Arc::new(opener);
74        let journal = Mutex::new(None);
75        let journal = Arc::new(journal);
76        Self { root, history, settings, opener, journal }
77    }
78
79    async fn call<T: Send + 'static>(&self, action: impl FnOnce(&mut StateLog) -> Result<T> + Send + 'static) -> Result<T> {
80        // Acquire Before Dispatch So Cancellation Cannot Enqueue An Older Write.
81        let mut owner = self.journal.clone().lock_owned().await;
82        tokio::task::spawn_blocking(move || {
83            let journal = owner.as_mut().ok_or_else(|| format::invalid("state backend is closed"))?;
84            journal.check()?;
85            action(journal)
86        })
87        .await
88        .map_err(task)?
89    }
90}
91
92impl StateBackend for FileBackend {
93    fn history(&self) -> &StoreHistory {
94        &self.history
95    }
96
97    fn open(&self) -> StorageFuture<'_, RecoveryPosition> {
98        Box::pin(async move {
99            let mut owner = self.journal.clone().lock_owned().await;
100            let root = self.root.clone();
101            let history = self.history.clone();
102            let settings = self.settings;
103            let opener = Arc::clone(&self.opener);
104            tokio::task::spawn_blocking(move || {
105                if owner.is_none() {
106                    *owner = Some(StateLog::open(&root, history, settings, opener)?);
107                }
108                let journal = owner.as_ref().expect("opened journal");
109                journal.check()?;
110                Ok(journal.position())
111            })
112            .await
113            .map_err(task)?
114        })
115    }
116
117    fn read_snapshot(&self, maximum: NonZeroUsize) -> StorageFuture<'_, Option<RecoverySnapshot>> {
118        Box::pin(self.call(move |journal| journal.snapshot(maximum.get())))
119    }
120
121    fn write_snapshot<'a>(&'a self, snapshot: &'a RecoverySnapshot) -> StorageFuture<'a, ()> {
122        Box::pin(async move {
123            self.history.check(&snapshot.through.history)?;
124            let snapshot = snapshot.clone();
125            self.call(move |journal| journal.checkpoint(&snapshot)).await
126        })
127    }
128
129    fn read_entries<'a>(
130        &'a self, after: &'a RecoveryPosition, through: &'a RecoveryPosition, limits: ReadLimits,
131    ) -> StorageFuture<'a, RecoveryPage> {
132        Box::pin(async move {
133            self.history.check(&after.history)?;
134            self.history.check(&through.history)?;
135            let after = after.clone();
136            let through = through.clone();
137            self.call(move |journal| journal.page(&after, &through, limits)).await
138        })
139    }
140
141    fn write_entries<'a>(&'a self, entries: &'a [RecoveryEntry]) -> StorageFuture<'a, ()> {
142        Box::pin(async move {
143            for entry in entries {
144                self.history.check(&entry.position.history)?;
145            }
146            let entries = entries.to_vec();
147            self.call(move |journal| journal.append(&entries)).await
148        })
149    }
150
151    fn flush<'a>(&'a self, through: &'a RecoveryPosition) -> StorageFuture<'a, RecoveryPosition> {
152        Box::pin(async move {
153            self.history.check(&through.history)?;
154            let through = through.clone();
155            self.call(move |journal| journal.flush(&through)).await
156        })
157    }
158
159    fn close(&self) -> StorageFuture<'_, ()> {
160        Box::pin(async move {
161            let mut owner = self.journal.clone().lock_owned().await;
162            tokio::task::spawn_blocking(move || {
163                let Some(mut journal) = owner.take() else {
164                    return Ok(());
165                };
166                journal.check()?;
167                journal.flush(&journal.position())?;
168                Ok(())
169            })
170            .await
171            .map_err(task)?
172        })
173    }
174}
175
176/* === Public Functions === */
177
178/* === Private Functions === */
179
180fn task(error: tokio::task::JoinError) -> Error {
181    Error::backend("state file task", BackendFailure::Permanent, error)
182}
183
184/* === Tests === */
harmos_storage/state/file/mod.rs

harmos_storage/state/file/
mod.rs
