hayson_engine/txn/
journal.rs

1use std::fs;
2use std::io;
3use std::path::Path;
4
5use serde::{Deserialize, Serialize};
6
7use crate::atomic::write_atomic;
8use crate::crypto::Sealer;
9
10pub const JOURNAL_FILE: &str = ".hayson.txn";
11pub const JOURNAL_VERSION: u32 = 1;
12
13#[derive(Debug, Clone, Serialize, Deserialize)]
14#[serde(rename_all = "camelCase")]
15pub struct JournalHeader {
16    // NOTE: "hayson" is the pre-5.0.0 key — read it, or an interrupted unit is missed at recovery.
17    #[serde(alias = "hayson")]
18    pub journal_version: u32,
19    pub kind: String,
20    pub id: String,
21    // NOTE: `started_at` is what 4.0.0 wrote before this alias — read it, or its unit is dropped.
22    #[serde(alias = "started_at")]
23    pub started_at: i64,
24    pub tables: Vec<String>,
25}
26
27#[derive(Debug, Clone, Serialize, Deserialize)]
28pub struct JournalEntry {
29    pub t: String,
30    pub s: u64,
31    pub l: String,
32}
33
34#[derive(Debug, Clone)]
35pub struct Journal {
36    pub header: JournalHeader,
37    pub entries: Vec<JournalEntry>,
38}
39
40pub fn write_journal(path: &Path, journal: &Journal) -> io::Result<()> {
41    write_journal_sealed(path, journal, None)
42}
43
44/// [`write_journal`] whose file is sealed with `sealer` when one is given: a journal holds the very
45/// lines about to be appended to the logs, so in an encrypted database it must not be plain.
46pub fn write_journal_sealed(path: &Path, journal: &Journal, sealer: Option<&Sealer>) -> io::Result<()> {
47    let mut lines = Vec::with_capacity(journal.entries.len() + 1);
48    lines.push(serde_json::to_string(&journal.header)?);
49    for entry in &journal.entries {
50        lines.push(serde_json::to_string(entry)?);
51    }
52    let mut contents = lines.join("\n");
53    contents.push('\n');
54    match sealer {
55        Some(sealer) => write_atomic(path, &sealer.seal_blob("journal", contents.as_bytes()), None),
56        None => write_atomic(path, contents.as_bytes(), None),
57    }
58}
59
60/// Reads back a journal written by [`write_journal`]. Any failure to parse it — missing file,
61/// truncated write, a header from a future/foreign format — is reported as "nothing to recover"
62/// rather than an error: an unreadable journal cannot be finished, only left alone.
63pub fn read_journal(path: &Path) -> Option<Journal> {
64    read_journal_sealed(path, None)
65}
66
67/// [`read_journal`] for a sealed journal. One that does not open is unreadable like any other.
68pub fn read_journal_sealed(path: &Path, sealer: Option<&Sealer>) -> Option<Journal> {
69    let bytes = fs::read(path).ok()?;
70    let bytes = match sealer {
71        Some(sealer) => sealer.open_blob("journal", &bytes).ok()?,
72        None => bytes,
73    };
74    let raw = String::from_utf8(bytes).ok()?;
75    let mut lines = raw.lines();
76    let header: JournalHeader = serde_json::from_str(lines.next()?).ok()?;
77    if header.journal_version != JOURNAL_VERSION || header.kind != "unit" {
78        return None;
79    }
80    let mut entries = Vec::new();
81    for line in lines {
82        if line.is_empty() {
83            continue;
84        }
85        entries.push(serde_json::from_str(line).ok()?);
86    }
87    Some(Journal { header, entries })
88}
89
90pub fn remove_journal(path: &Path) {
91    let _ = fs::remove_file(path);
92    let _ = fs::remove_file(format!("{}.tmp", path.display()));
93}
94
95#[cfg(test)]
96mod tests {
97    use super::*;
98    use tempfile::tempdir;
99
100    #[test]
101    fn round_trips_header_and_entries() {
102        let dir = tempdir().unwrap();
103        let path = dir.path().join(JOURNAL_FILE);
104        let journal = Journal {
105            header: JournalHeader {
106                journal_version: JOURNAL_VERSION,
107                kind: "unit".into(),
108                id: "abc".into(),
109                started_at: 42,
110                tables: vec!["users".into(), "orders".into()],
111            },
112            entries: vec![
113                JournalEntry {
114                    t: "users".into(),
115                    s: 1,
116                    l: "{\"_k\":\"u1\"}".into(),
117                },
118                JournalEntry {
119                    t: "orders".into(),
120                    s: 1,
121                    l: "{\"_k\":\"o1\"}".into(),
122                },
123            ],
124        };
125
126        write_journal(&path, &journal).unwrap();
127        let read = read_journal(&path).unwrap();
128        assert_eq!(read.header.id, "abc");
129        assert_eq!(read.header.tables, vec!["users".to_string(), "orders".to_string()]);
130        assert_eq!(read.entries.len(), 2);
131
132        remove_journal(&path);
133        assert!(read_journal(&path).is_none());
134    }
135
136    #[test]
137    fn a_missing_journal_reads_as_nothing_to_recover() {
138        let dir = tempdir().unwrap();
139        let path = dir.path().join(JOURNAL_FILE);
140        assert!(read_journal(&path).is_none());
141    }
142
143    #[test]
144    fn writes_the_header_key_the_format_documents() {
145        let dir = tempdir().unwrap();
146        let path = dir.path().join(JOURNAL_FILE);
147        let journal = Journal {
148            header: JournalHeader {
149                journal_version: JOURNAL_VERSION,
150                kind: "unit".into(),
151                id: "abc".into(),
152                started_at: 1_786_995_100_337,
153                tables: vec!["users".into()],
154            },
155            entries: Vec::new(),
156        };
157
158        write_journal(&path, &journal).unwrap();
159
160        let header = fs::read_to_string(&path).unwrap().lines().next().unwrap().to_string();
161        assert!(header.contains("\"startedAt\":1786995100337"), "{header}");
162        assert!(!header.contains("started_at"), "{header}");
163        assert!(header.contains("\"journalVersion\":1"), "{header}");
164        assert!(
165            !header.contains("\"hayson\""),
166            "the old marker key is never written again: {header}"
167        );
168    }
169
170    /// A unit left unfinished by a release that wrote the marker under `"hayson"` still has to be
171    /// finished — nothing migrates a journal, and a missed one loses a half-applied write.
172    #[test]
173    fn reads_the_pre_rename_marker_key_an_older_release_left_behind() {
174        let dir = tempdir().unwrap();
175        let path = dir.path().join(JOURNAL_FILE);
176        fs::write(
177            &path,
178            "{\"hayson\":1,\"kind\":\"unit\",\"id\":\"abc\",\"startedAt\":42,\"tables\":[\"users\"]}\n{\"t\":\"users\",\"s\":7,\"l\":\"{}\"}\n",
179        )
180        .unwrap();
181
182        let read = read_journal(&path).expect("a journal under the old marker key is still recoverable");
183        assert_eq!(read.header.journal_version, JOURNAL_VERSION);
184        assert_eq!(read.header.id, "abc");
185        assert_eq!(read.entries.len(), 1);
186    }
187
188    /// A unit left unfinished by a release that wrote the snake-case key still has to be finished,
189    /// not silently dropped — the whole point of the journal is that nothing gets lost.
190    #[test]
191    fn reads_the_snake_case_header_an_older_release_left_behind() {
192        let dir = tempdir().unwrap();
193        let path = dir.path().join(JOURNAL_FILE);
194        fs::write(
195            &path,
196            "{\"hayson\":1,\"kind\":\"unit\",\"id\":\"abc\",\"started_at\":42,\"tables\":[\"users\"]}\n{\"t\":\"users\",\"s\":7,\"l\":\"{}\"}\n",
197        )
198        .unwrap();
199
200        let read = read_journal(&path).expect("an older journal is still recoverable");
201        assert_eq!(read.header.started_at, 42);
202        assert_eq!(read.entries.len(), 1);
203    }
204}
hayson_engine/txn/journal.rs

hayson_engine/txn/
journal.rs
