harmos-storage 0.1.3

Harmos storage: durable journals, snapshots and Arrow-backed stream storage.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112

/* ***
 *
 * Harmos
 *
 * SPDX-License-Identifier: Apache-2.0
 *
 * ***
 */

// State File Ownership And Durable Replacement

/* === Modules === */

/* === Re-Exports === */

/* === Imports === */

// Standard Library Imports
use std::fs::{self, File, OpenOptions};
use std::io::{BufReader, BufWriter, Write};
use std::path::{Path, PathBuf};
use std::sync::Arc;
// Third Party Imports
use serde::{Serialize, de::DeserializeOwned};
// Local Imports
use super::format::{self, Read, corrupt, io};
use crate::{BackendFailure, Error, Result};

/* === Constants === */

/* === Macros === */

/* === Type Aliases === */

pub(super) type JournalOpener = Arc<dyn Fn(&Path) -> std::io::Result<File> + Send + Sync>;

/* === Structs === */

pub(super) struct Files {
    root: PathBuf,
    _lease: File,
    opener: JournalOpener,
}

/* === Enums === */

/* === Traits === */

/* === Implementations === */

impl Files {
    pub(super) fn open(root: &Path, opener: JournalOpener) -> Result<Self> {
        fs::create_dir_all(root).map_err(io)?;
        let lease = OpenOptions::new().create(true).truncate(false).read(true).write(true).open(root.join("state.lock")).map_err(io)?;
        lease.try_lock().map_err(|error| Error::backend("lock state storage", BackendFailure::Conflict, error))?;
        Ok(Self { root: root.to_owned(), _lease: lease, opener })
    }

    pub(super) fn journal(&self) -> Result<File> {
        (self.opener)(&self.root.join("state.jsonl")).map_err(io)
    }

    pub(super) fn snapshot<T: DeserializeOwned>(&self, maximum: usize) -> Result<Option<T>> {
        let file = match File::open(self.root.join("state.json")) {
            Ok(file) => file,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
            Err(error) => return Err(io(error)),
        };
        let mut reader = BufReader::new(file);
        let value = match format::read(&mut reader, maximum)? {
            Read::Value(value) => value,
            Read::End | Read::Incomplete => return Err(corrupt("incomplete state checkpoint")),
        };
        if !matches!(format::read::<serde_json::Value>(&mut reader, maximum)?, Read::End) {
            return Err(corrupt("state checkpoint contains trailing records"));
        }
        Ok(Some(value))
    }

    pub(super) fn write_snapshot<T: Serialize>(&self, value: &T, maximum: usize) -> Result<()> {
        self.replace("state.json", |writer| format::write(writer, value, maximum))
    }

    pub(super) fn replace_journal(&self, write: impl FnOnce(&mut BufWriter<File>) -> Result<()>) -> Result<()> {
        self.replace("state.jsonl", write)
    }

    // Directory Flush Is A POSIX Idiom: Windows Refuses A Directory Opened As A File
    // (`ERROR_ACCESS_DENIED`), And NTFS Makes Rename Metadata Durable Without One.
    pub(super) fn sync_directory(&self) -> Result<()> {
        if cfg!(unix) { File::open(&self.root).map_err(io)?.sync_all().map_err(io) } else { Ok(()) }
    }

    fn replace(&self, name: &str, write: impl FnOnce(&mut BufWriter<File>) -> Result<()>) -> Result<()> {
        // The Exclusive Lease Owns These Temporary Names, Including Crash Remnants.
        let temporary = self.root.join(format!("{name}.pending"));
        let file = OpenOptions::new().create(true).truncate(true).write(true).open(&temporary).map_err(io)?;
        let mut writer = BufWriter::new(file);
        write(&mut writer)?;
        writer.flush().map_err(io)?;
        writer.get_ref().sync_all().map_err(io)?;
        drop(writer);
        fs::rename(&temporary, self.root.join(name)).map_err(io)?;
        self.sync_directory()
    }
}

/* === Public Functions === */

/* === Private Functions === */

/* === Tests === */