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, StorageBatch, 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    #[tracing::instrument(name = "storage.state.file.mod.call", level = "debug", skip_all)]
80    async fn call<T: Send + 'static>(&self, action: impl FnOnce(&mut StateLog) -> Result<T> + Send + 'static) -> Result<T> {
81        // Acquire Before Dispatch So Cancellation Cannot Enqueue An Older Write.
82        let mut owner = tracing::Instrument::instrument(self.journal.clone().lock_owned(), tracing::trace_span!("storage.lock.wait")).await;
83        {
84            let context = crate::execution::Context::current();
85            tokio::task::spawn_blocking(move || {
86                context.blocking(|| {
87                    let journal = owner.as_mut().ok_or_else(|| format::invalid("state backend is closed"))?;
88                    journal.check()?;
89                    action(journal)
90                })
91            })
92        }
93        .await
94        .map_err(task)?
95    }
96    async fn read<T: Send + 'static>(
97        &self, memory: crate::arrow::RetainedMemory, base: usize,
98        action: impl FnOnce(&mut StateLog, &crate::arrow::RetainedMemory) -> Result<T> + Send + 'static,
99    ) -> Result<T> {
100        memory.try_grow_to(base.saturating_add(8192))?;
101        let mut owner = tracing::Instrument::instrument(self.journal.clone().lock_owned(), tracing::trace_span!("storage.lock.wait")).await;
102        let (result, _memory) = read_task(memory, move |memory| {
103            let journal = owner.as_mut().ok_or_else(|| format::invalid("state backend is closed"))?;
104            journal.check()?;
105            action(journal, memory)
106        })
107        .await
108        .map_err(task)?;
109        result
110    }
111}
112
113impl StateBackend for FileBackend {
114    fn history(&self) -> &StoreHistory {
115        &self.history
116    }
117
118    fn open(&self) -> StorageFuture<'_, RecoveryPosition> {
119        Box::pin(async move {
120            let context = crate::execution::Context::current();
121            let future = async move {
122                let span = tracing::debug_span!("storage.state.backend", method = "open", outcome = "cancelled");
123                let result = tracing::Instrument::instrument(
124                    async move {
125                        let mut owner =
126                            tracing::Instrument::instrument(self.journal.clone().lock_owned(), tracing::trace_span!("storage.lock.wait"))
127                                .await;
128                        let root = self.root.clone();
129                        let history = self.history.clone();
130                        let settings = self.settings;
131                        let opener = Arc::clone(&self.opener);
132                        {
133                            let context = crate::execution::Context::current();
134                            tokio::task::spawn_blocking(move || {
135                                context.blocking(|| {
136                                    if owner.is_none() {
137                                        *owner = Some(StateLog::open(&root, history, settings, opener)?);
138                                    }
139                                    let journal = owner.as_ref().expect("opened journal");
140                                    journal.check()?;
141                                    Ok(journal.position())
142                                })
143                            })
144                        }
145                        .await
146                        .map_err(task)?
147                    },
148                    span.clone(),
149                )
150                .await;
151                span.record("outcome", if result.is_ok() { "completed" } else { "failed" });
152                result
153            };
154            context.scope(future).await
155        })
156    }
157
158    fn read_snapshot(&self, maximum: NonZeroUsize, memory: crate::arrow::RetainedMemory) -> StorageFuture<'_, Option<RecoverySnapshot>> {
159        Box::pin(async move {
160            let context = crate::execution::Context::current();
161            let future = async move {
162                let span = tracing::debug_span!("storage.state.backend", method = "read_snapshot", outcome = "cancelled");
163                let result = tracing::Instrument::instrument(
164                    self.read(memory, crate::manager::ENTRY_OVERHEAD, move |journal, memory| {
165                        journal.snapshot_retained(maximum.get(), memory, crate::manager::ENTRY_OVERHEAD)
166                    }),
167                    span.clone(),
168                )
169                .await;
170                span.record("outcome", if result.is_ok() { "completed" } else { "failed" });
171                result
172            };
173            context.scope(future).await
174        })
175    }
176
177    fn write_snapshot(&self, snapshot: StorageBatch<RecoverySnapshot>) -> StorageFuture<'_, ()> {
178        Box::pin(async move {
179            let context = crate::execution::Context::current();
180            let future = async move {
181                let span = tracing::debug_span!("storage.state.backend", method = "write_snapshot", outcome = "cancelled");
182                let result = tracing::Instrument::instrument(
183                    async move {
184                        if snapshot.len() != 1 {
185                            return Err(format::invalid("snapshot write requires exactly one snapshot"));
186                        }
187                        self.history.check(&snapshot.as_slice()[0].through.history)?;
188                        self.call(move |journal| journal.checkpoint(&snapshot.as_slice()[0])).await
189                    },
190                    span.clone(),
191                )
192                .await;
193                span.record("outcome", if result.is_ok() { "completed" } else { "failed" });
194                result
195            };
196            context.scope(future).await
197        })
198    }
199
200    fn read_entries<'a>(
201        &'a self, after: &'a RecoveryPosition, through: &'a RecoveryPosition, limits: ReadLimits, memory: crate::arrow::RetainedMemory,
202    ) -> StorageFuture<'a, RecoveryPage> {
203        Box::pin(async move {
204            let context = crate::execution::Context::current();
205            let future = async move {
206                let span = tracing::debug_span!("storage.state.backend", method = "read_entries", outcome = "cancelled");
207                let result = tracing::Instrument::instrument(
208                    async move {
209                        self.history.check(&after.history)?;
210                        self.history.check(&through.history)?;
211                        let base = crate::manager::ENTRY_OVERHEAD
212                            .saturating_add(crate::MemorySize::memory_size(&after.history))
213                            .saturating_add(crate::MemorySize::memory_size(&through.history));
214                        memory.try_grow_to(base)?;
215                        let after = after.clone();
216                        let through = through.clone();
217                        self.read(memory, base, move |journal, memory| journal.page(&after, &through, limits, memory, base)).await
218                    },
219                    span.clone(),
220                )
221                .await;
222                span.record("outcome", if result.is_ok() { "completed" } else { "failed" });
223                result
224            };
225            context.scope(future).await
226        })
227    }
228
229    fn write_entries(&self, entries: StorageBatch<RecoveryEntry>) -> StorageFuture<'_, ()> {
230        Box::pin(async move {
231            let context = crate::execution::Context::current();
232            let future = async move {
233                let span = tracing::debug_span!("storage.state.backend", method = "write_entries", outcome = "cancelled");
234                let result = tracing::Instrument::instrument(
235                    async move {
236                        for entry in entries.as_slice() {
237                            self.history.check(&entry.position.history)?;
238                        }
239                        self.call(move |journal| journal.append(entries.as_slice())).await
240                    },
241                    span.clone(),
242                )
243                .await;
244                span.record("outcome", if result.is_ok() { "completed" } else { "failed" });
245                result
246            };
247            context.scope(future).await
248        })
249    }
250
251    fn flush<'a>(&'a self, through: &'a RecoveryPosition) -> StorageFuture<'a, RecoveryPosition> {
252        Box::pin(async move {
253            let context = crate::execution::Context::current();
254            let future = async move {
255                let span = tracing::debug_span!("storage.state.backend", method = "flush", outcome = "cancelled");
256                let result = tracing::Instrument::instrument(
257                    async move {
258                        self.history.check(&through.history)?;
259                        let through = through.clone();
260                        self.call(move |journal| journal.flush(&through)).await
261                    },
262                    span.clone(),
263                )
264                .await;
265                span.record("outcome", if result.is_ok() { "completed" } else { "failed" });
266                result
267            };
268            context.scope(future).await
269        })
270    }
271
272    fn close(&self) -> StorageFuture<'_, ()> {
273        Box::pin(async move {
274            let context = crate::execution::Context::current();
275            let future = async move {
276                let span = tracing::debug_span!("storage.state.backend", method = "close", outcome = "cancelled");
277                let result = tracing::Instrument::instrument(
278                    async move {
279                        let mut owner =
280                            tracing::Instrument::instrument(self.journal.clone().lock_owned(), tracing::trace_span!("storage.lock.wait"))
281                                .await;
282                        {
283                            let context = crate::execution::Context::current();
284                            tokio::task::spawn_blocking(move || {
285                                context.blocking(|| {
286                                    let Some(mut journal) = owner.take() else {
287                                        return Ok(());
288                                    };
289                                    journal.check()?;
290                                    journal.flush(&journal.position())?;
291                                    Ok(())
292                                })
293                            })
294                        }
295                        .await
296                        .map_err(task)?
297                    },
298                    span.clone(),
299                )
300                .await;
301                span.record("outcome", if result.is_ok() { "completed" } else { "failed" });
302                result
303            };
304            context.scope(future).await
305        })
306    }
307}
308
309/* === Public Functions === */
310
311/* === Private Functions === */
312
313fn read_task<T: Send + 'static>(
314    memory: crate::arrow::RetainedMemory, action: impl FnOnce(&crate::arrow::RetainedMemory) -> Result<T> + Send + 'static,
315) -> tokio::task::JoinHandle<(Result<T>, crate::arrow::RetainedMemory)> {
316    let context = crate::execution::Context::current();
317    tokio::task::spawn_blocking(move || {
318        let result = context.blocking(|| action(&memory));
319        (result, memory)
320    })
321}
322
323fn task(error: tokio::task::JoinError) -> Error {
324    Error::backend("state file task", BackendFailure::Permanent, error)
325}
326
327/* === Tests === */
328#[cfg(test)]
329mod tests {
330    use super::*;
331    use crate::test_backend::{Data, Recovery, history, storage_settings};
332
333    fn budget() -> crate::StorageManager {
334        crate::StorageManager::new(
335            crate::StorageBackends { primary: Arc::new(Data::<()>::new(history())), state: Arc::new(Recovery::new(history())) },
336            storage_settings(),
337        )
338        .unwrap()
339    }
340
341    #[tokio::test]
342    async fn detached_read_retained() {
343        let budget = budget();
344        let memory = budget.read_memory(NonZeroUsize::new(8192).unwrap()).await.unwrap();
345        let (entered, ready) = tokio::sync::oneshot::channel();
346        let (release, gate) = std::sync::mpsc::channel();
347        let task = read_task(memory, move |_| {
348            let bytes = vec![7u8; 4096];
349            entered.send(()).unwrap();
350            gate.recv().unwrap();
351            Ok(bytes)
352        });
353        ready.await.unwrap();
354        drop(task);
355        assert_eq!(budget.status().read_bytes, 8192);
356        let mut changed = budget.inner.changed.subscribe();
357        release.send(()).unwrap();
358        while budget.status().read_bytes != 0 {
359            changed.changed().await.unwrap();
360        }
361    }
362
363    #[tokio::test]
364    async fn completed_read_retained() {
365        let budget = budget();
366        let memory = budget.read_memory(NonZeroUsize::new(8192).unwrap()).await.unwrap();
367        let task = read_task(memory, |_| Ok(vec![7u8; 4096]));
368        while !task.is_finished() {
369            tokio::task::yield_now().await;
370        }
371        assert_eq!(budget.status().read_bytes, 8192);
372        drop(task);
373        assert_eq!(budget.status().read_bytes, 0);
374    }
375}
harmos_storage/state/file/mod.rs

harmos_storage/state/file/
mod.rs
