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

hayson_engine/txn/
journal.rs
