kcode_session_control_journal/
lib.rs

1//! Durable mechanics for one checksummed `.session-control` record file.
2//!
3//! Domain interpretation, compaction selection, and coordination belong to
4//! Session History or another caller.
5
6use std::{
7    ffi::OsString,
8    fs::{File, OpenOptions},
9    io::{Read, Write},
10    path::{Path, PathBuf},
11};
12
13use anyhow::{Context as _, ensure};
14use serde::{Deserialize, Serialize};
15use serde_json::Value;
16use sha2::{Digest, Sha256};
17use uuid::Uuid;
18
19/// One uninterpreted session-control record.
20#[derive(Clone, Debug, Deserialize, Serialize)]
21#[serde(rename_all = "camelCase")]
22pub struct Record {
23    pub kind: String,
24    pub recorded_at: String,
25    pub value: Value,
26}
27
28/// An opened checksummed record file.
29///
30/// The caller owns all locking and domain policy.
31pub struct Journal {
32    path: PathBuf,
33    records: Vec<Record>,
34}
35
36impl Journal {
37    /// Creates and synchronizes a new empty journal.
38    pub fn create(path: PathBuf) -> anyhow::Result<Self> {
39        let file = OpenOptions::new()
40            .create_new(true)
41            .write(true)
42            .open(&path)
43            .with_context(|| format!("creating {}", path.display()))?;
44        file.sync_all()
45            .with_context(|| format!("syncing {}", path.display()))?;
46        sync_directory(parent(&path))?;
47        Ok(Self {
48            path,
49            records: Vec::new(),
50        })
51    }
52
53    /// Opens an existing journal, or returns `None` when it is absent.
54    ///
55    /// An incomplete final line is truncated and synchronized. Corruption in
56    /// any complete line is rejected.
57    pub fn open(path: PathBuf) -> anyhow::Result<Option<Self>> {
58        let mut file = match OpenOptions::new().read(true).write(true).open(&path) {
59            Ok(file) => file,
60            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
61            Err(error) => {
62                return Err(error).with_context(|| format!("opening {}", path.display()));
63            }
64        };
65        let mut bytes = Vec::new();
66        file.read_to_end(&mut bytes)
67            .with_context(|| format!("reading {}", path.display()))?;
68
69        let mut records = Vec::new();
70        let mut cursor = 0_usize;
71        while cursor < bytes.len() {
72            let Some(relative_end) = bytes[cursor..].iter().position(|byte| *byte == b'\n') else {
73                let repaired_length =
74                    u64::try_from(cursor).context("journal length does not fit in u64")?;
75                file.set_len(repaired_length)
76                    .with_context(|| format!("repairing {}", path.display()))?;
77                file.sync_all()
78                    .with_context(|| format!("syncing repaired {}", path.display()))?;
79                break;
80            };
81            let end = cursor + relative_end;
82            records.push(parse_record(&bytes[cursor..end])?);
83            cursor = end + 1;
84        }
85
86        Ok(Some(Self { path, records }))
87    }
88
89    /// Returns validated records in file order.
90    pub fn records(&self) -> &[Record] {
91        &self.records
92    }
93
94    /// Appends and synchronizes one record before updating memory.
95    pub fn append(
96        &mut self,
97        kind: impl Into<String>,
98        recorded_at: impl Into<String>,
99        value: Value,
100    ) -> anyhow::Result<()> {
101        let record = Record {
102            kind: kind.into(),
103            recorded_at: recorded_at.into(),
104            value,
105        };
106        let encoded = encode_record(&record)?;
107        let mut file = OpenOptions::new()
108            .append(true)
109            .open(&self.path)
110            .with_context(|| format!("opening {} for append", self.path.display()))?;
111        file.write_all(&encoded)
112            .with_context(|| format!("appending {}", self.path.display()))?;
113        file.sync_all()
114            .with_context(|| format!("syncing {}", self.path.display()))?;
115        self.records.push(record);
116        Ok(())
117    }
118
119    /// Atomically replaces the file with caller-selected records.
120    pub fn replace(&mut self, records: impl IntoIterator<Item = Record>) -> anyhow::Result<()> {
121        let records = records.into_iter().collect::<Vec<_>>();
122        let permissions = std::fs::metadata(&self.path)
123            .with_context(|| format!("reading permissions for {}", self.path.display()))?
124            .permissions();
125        let temporary = temporary_path(&self.path)?;
126        let mut temporary_created = false;
127        let result = (|| -> anyhow::Result<()> {
128            let mut file = OpenOptions::new()
129                .create_new(true)
130                .write(true)
131                .open(&temporary)
132                .with_context(|| format!("creating {}", temporary.display()))?;
133            temporary_created = true;
134            file.set_permissions(permissions)
135                .with_context(|| format!("setting permissions on {}", temporary.display()))?;
136            for record in &records {
137                file.write_all(&encode_record(record)?)
138                    .with_context(|| format!("writing {}", temporary.display()))?;
139            }
140            file.sync_all()
141                .with_context(|| format!("syncing {}", temporary.display()))?;
142            drop(file);
143            std::fs::rename(&temporary, &self.path)
144                .with_context(|| format!("replacing {}", self.path.display()))?;
145            sync_directory(parent(&self.path))?;
146            Ok(())
147        })();
148        if result.is_err() && temporary_created {
149            let _ = std::fs::remove_file(&temporary);
150        }
151        result?;
152        self.records = records;
153        Ok(())
154    }
155}
156
157fn parent(path: &Path) -> &Path {
158    path.parent()
159        .filter(|parent| !parent.as_os_str().is_empty())
160        .unwrap_or_else(|| Path::new("."))
161}
162
163fn temporary_path(path: &Path) -> anyhow::Result<PathBuf> {
164    let file_name = path.file_name().context("journal path has no file name")?;
165    let mut temporary_name = OsString::from(".");
166    temporary_name.push(file_name);
167    temporary_name.push(format!(".replace-{}.tmp", Uuid::new_v4()));
168    Ok(parent(path).join(temporary_name))
169}
170
171fn sync_directory(path: &Path) -> anyhow::Result<()> {
172    File::open(path)
173        .with_context(|| format!("opening directory {} for sync", path.display()))?
174        .sync_all()
175        .with_context(|| format!("syncing directory {}", path.display()))
176}
177
178fn parse_record(line: &[u8]) -> anyhow::Result<Record> {
179    let separator = line
180        .iter()
181        .position(|byte| *byte == b' ')
182        .context("session-control record has no checksum separator")?;
183    let expected =
184        std::str::from_utf8(&line[..separator]).context("session-control checksum is not UTF-8")?;
185    let payload = &line[separator + 1..];
186    ensure!(
187        hex_sha256(payload) == expected,
188        "session-control record checksum mismatch"
189    );
190    let payload =
191        std::str::from_utf8(payload).context("session-control JSON payload is not UTF-8")?;
192    serde_json::from_str(payload).context("decoding session-control record")
193}
194
195fn encode_record(record: &Record) -> anyhow::Result<Vec<u8>> {
196    let payload = serde_json::to_vec(record).context("encoding session-control record")?;
197    let checksum = hex_sha256(&payload);
198    let mut encoded = Vec::with_capacity(checksum.len() + payload.len() + 2);
199    encoded.extend_from_slice(checksum.as_bytes());
200    encoded.push(b' ');
201    encoded.extend_from_slice(&payload);
202    encoded.push(b'\n');
203    Ok(encoded)
204}
205
206fn hex_sha256(bytes: &[u8]) -> String {
207    const HEX: &[u8; 16] = b"0123456789abcdef";
208    let mut output = String::with_capacity(64);
209    for byte in Sha256::digest(bytes) {
210        output.push(char::from(HEX[usize::from(byte >> 4)]));
211        output.push(char::from(HEX[usize::from(byte & 0x0f)]));
212    }
213    output
214}
215
216#[cfg(test)]
217mod tests {
218    use std::fs;
219    use std::io::Write as _;
220
221    use serde_json::json;
222
223    use super::*;
224
225    fn path(label: &str) -> PathBuf {
226        let directory = std::env::temp_dir().join(format!(
227            "kcode-session-control-journal-{label}-{}-{}",
228            std::process::id(),
229            Uuid::new_v4()
230        ));
231        fs::create_dir(&directory).unwrap();
232        directory.join("test.session-control")
233    }
234
235    fn remove(path: &Path) {
236        fs::remove_dir_all(path.parent().unwrap()).unwrap();
237    }
238
239    fn record(kind: &str, number: u64) -> Record {
240        Record {
241            kind: kind.into(),
242            recorded_at: format!("2026-08-01T00:00:0{number}Z"),
243            value: json!({"number":number}),
244        }
245    }
246
247    #[test]
248    fn append_encoding_and_reopen_are_exact() {
249        let path = path("append");
250        let mut journal = Journal::create(path.clone()).unwrap();
251        journal
252            .append(
253                "session_lifecycle",
254                "2026-08-01T00:00:00Z",
255                json!({"version":1}),
256            )
257            .unwrap();
258
259        let payload = br#"{"kind":"session_lifecycle","recordedAt":"2026-08-01T00:00:00Z","value":{"version":1}}"#;
260        let checksum = hex_sha256(payload);
261        let mut expected = Vec::new();
262        expected.extend_from_slice(checksum.as_bytes());
263        expected.push(b' ');
264        expected.extend_from_slice(payload);
265        expected.push(b'\n');
266        assert_eq!(fs::read(&path).unwrap(), expected);
267        drop(journal);
268
269        let reopened = Journal::open(path.clone()).unwrap().unwrap();
270        assert_eq!(reopened.records().len(), 1);
271        assert_eq!(reopened.records()[0].kind, "session_lifecycle");
272        assert_eq!(reopened.records()[0].recorded_at, "2026-08-01T00:00:00Z");
273        assert_eq!(reopened.records()[0].value, json!({"version":1}));
274        remove(&path);
275    }
276
277    #[test]
278    fn opening_repairs_only_an_incomplete_final_tail() {
279        let path = path("tail");
280        let mut journal = Journal::create(path.clone()).unwrap();
281        journal.append("first", "time", json!(1)).unwrap();
282        let complete = fs::read(&path).unwrap();
283        drop(journal);
284
285        let mut file = OpenOptions::new().append(true).open(&path).unwrap();
286        file.write_all(b"interrupted record").unwrap();
287        file.sync_all().unwrap();
288        drop(file);
289
290        let reopened = Journal::open(path.clone()).unwrap().unwrap();
291        assert_eq!(reopened.records().len(), 1);
292        assert_eq!(reopened.records()[0].kind, "first");
293        assert_eq!(fs::read(&path).unwrap(), complete);
294        remove(&path);
295    }
296
297    #[test]
298    fn complete_checksum_corruption_is_rejected() {
299        let path = path("corruption");
300        let mut journal = Journal::create(path.clone()).unwrap();
301        journal.append("first", "time", json!(1)).unwrap();
302        drop(journal);
303
304        let mut bytes = fs::read(&path).unwrap();
305        bytes[0] = if bytes[0] == b'0' { b'1' } else { b'0' };
306        let mut file = OpenOptions::new()
307            .write(true)
308            .truncate(true)
309            .open(&path)
310            .unwrap();
311        file.write_all(&bytes).unwrap();
312        file.sync_all().unwrap();
313        drop(file);
314
315        let error = match Journal::open(path.clone()) {
316            Err(error) => error,
317            Ok(_) => panic!("corrupt complete record was accepted"),
318        };
319        assert!(error.to_string().contains("checksum mismatch"));
320        remove(&path);
321    }
322
323    #[test]
324    fn replace_preserves_order_updates_memory_and_reopens() {
325        let path = path("replace");
326        let mut journal = Journal::create(path.clone()).unwrap();
327        journal.append("old", "time", json!(0)).unwrap();
328
329        journal
330            .replace([record("second", 2), record("first", 1)])
331            .unwrap();
332        assert_eq!(journal.records().len(), 2);
333        assert_eq!(journal.records()[0].kind, "second");
334        assert_eq!(journal.records()[1].kind, "first");
335        drop(journal);
336
337        let reopened = Journal::open(path.clone()).unwrap().unwrap();
338        assert_eq!(reopened.records().len(), 2);
339        assert_eq!(reopened.records()[0].kind, "second");
340        assert_eq!(reopened.records()[0].value, json!({"number":2}));
341        assert_eq!(reopened.records()[1].kind, "first");
342        assert_eq!(reopened.records()[1].value, json!({"number":1}));
343        remove(&path);
344    }
345
346    #[test]
347    fn create_rejects_an_existing_path() {
348        let path = path("collision");
349        let journal = Journal::create(path.clone()).unwrap();
350        let error = match Journal::create(path.clone()) {
351            Err(error) => error,
352            Ok(_) => panic!("create replaced an existing journal"),
353        };
354        assert_eq!(
355            error
356                .downcast_ref::<std::io::Error>()
357                .map(std::io::Error::kind),
358            Some(std::io::ErrorKind::AlreadyExists)
359        );
360        drop(journal);
361        remove(&path);
362    }
363
364    #[test]
365    fn create_and_replace_keep_the_target_in_its_parent() {
366        let path = path("durability");
367        assert!(Journal::open(path.clone()).unwrap().is_none());
368        let mut journal = Journal::create(path.clone()).unwrap();
369
370        #[cfg(unix)]
371        {
372            use std::os::unix::fs::PermissionsExt as _;
373            fs::set_permissions(&path, fs::Permissions::from_mode(0o640)).unwrap();
374        }
375
376        journal.replace([record("retained", 1)]).unwrap();
377        assert!(path.is_file());
378        let entries = fs::read_dir(path.parent().unwrap())
379            .unwrap()
380            .map(|entry| entry.unwrap().path())
381            .collect::<Vec<_>>();
382        assert_eq!(entries, vec![path.clone()]);
383
384        #[cfg(unix)]
385        {
386            use std::os::unix::fs::PermissionsExt as _;
387            assert_eq!(
388                fs::metadata(&path).unwrap().permissions().mode() & 0o777,
389                0o640
390            );
391        }
392
393        remove(&path);
394    }
395}
kcode_session_control_journal/lib.rs

kcode_session_control_journal/
lib.rs
