blockworx/storage/
history.rs

1//! The container's history: one deflated snapshot per settled edit, each with a
2//! small sidecar saying when it happened and what it touched.
3//!
4//! ```text
5//! history/000123.kdl.gz   the whole document, deflated
6//! history/000123.json     {"ts":…, "command":"delete", "changed":["b3"]}
7//! ```
8//!
9//! **The sidecars are the index.** There is no separate index file, so there is
10//! nothing that can go stale, and `rg '"b0:p1->b1:p1"' doc.bwx/history/*.json`
11//! answers "when did this route change?" without reading a single snapshot.
12//!
13//! Entries are write-once and are never rewritten, which is what lets them be
14//! written [`Relaxed`](super::Durability::Relaxed): a lost write is a missing
15//! entry rather than a damaged container, and a torn one fails its gzip checksum
16//! and is skipped. The payload is written *before* its sidecar, so the pair can
17//! only tear one way — into an entry that is still readable, just unlabelled.
18
19use std::io::Write as _;
20
21use serde::{Deserialize, Serialize};
22
23use super::{Durability, Storage};
24
25pub const DIR: &str = "history";
26
27/// One line per time the document was opened.
28const SESSIONS: &str = "history/sessions.jsonl";
29
30/// One entry's metadata: everything a search or a timeline row needs, so
31/// neither has to open a snapshot to decide whether it is interesting.
32#[derive(Clone, PartialEq, Eq, Debug, Default, Serialize, Deserialize)]
33pub struct Entry {
34    /// Milliseconds since the Unix epoch.
35    pub ts: u64,
36    /// The registry name of the command that produced this state, when the edit
37    /// went through one. Absent for direct canvas work — dragging a block is not
38    /// a command — which is why `changed` carries the real information.
39    #[serde(default, skip_serializing_if = "Option::is_none")]
40    pub command: Option<String>,
41    /// What the edit touched, as `document::change` names it.
42    #[serde(default, skip_serializing_if = "Vec::is_empty")]
43    pub changed: Vec<String>,
44}
45
46/// A session: the document was opened at `ts`, and everything from entry `seq`
47/// onwards is that session's work.
48///
49/// Recorded even though nothing reads it yet, because the record has to be made
50/// at the time — a timeline built later cannot work out when past sessions began.
51/// It is also what a "revert to where this session started" verb would stand on,
52/// which is the useful half of the `.bak`-on-startup convention the container
53/// otherwise replaces.
54#[derive(Clone, Copy, PartialEq, Eq, Debug, Serialize, Deserialize)]
55pub struct Session {
56    pub ts: u64,
57    pub seq: u64,
58}
59
60/// Note that a session opened.
61///
62/// The one file in the container that is read-modify-written rather than
63/// appended to write-once, so it is synced: the whole file is replaced, and a
64/// torn write would lose every marker rather than the newest one. It happens
65/// once per open, so the cost lands where nobody is waiting.
66pub fn record_session(storage: &impl Storage, session: &Session) -> std::io::Result<()> {
67    let mut lines = storage.read(SESSIONS).unwrap_or_default();
68    if !lines.is_empty() && !lines.ends_with(b"\n") {
69        lines.push(b'\n');
70    }
71    serde_json::to_writer(&mut lines, session).map_err(std::io::Error::other)?;
72    lines.push(b'\n');
73    storage.write(SESSIONS, &lines, Durability::Sync)
74}
75
76/// Every recorded session, oldest first. A line that does not parse is skipped:
77/// a damaged marker must not hide the sessions around it.
78#[cfg_attr(not(test), allow(dead_code))]
79pub fn sessions(storage: &impl Storage) -> Vec<Session> {
80    let Ok(bytes) = storage.read(SESSIONS) else {
81        return Vec::new();
82    };
83    String::from_utf8_lossy(&bytes)
84        .lines()
85        .filter_map(|line| serde_json::from_str(line).ok())
86        .collect()
87}
88
89/// An entry as found on disk. `meta` is `None` for an entry whose sidecar never
90/// landed — readable, just unlabelled.
91///
92/// Reading the history back is the timeline's job (todo.md P6); the writer only
93/// appends. Kept here rather than deferred because what a torn or damaged entry
94/// looks like on the way *out* is what justifies how it is written.
95#[cfg_attr(not(test), allow(dead_code))]
96#[derive(Clone, PartialEq, Eq, Debug)]
97pub struct Record {
98    pub seq: u64,
99    pub meta: Option<Entry>,
100}
101
102/// Both files making up entry `seq`, for a caller removing it.
103pub fn paths(seq: u64) -> [String; 2] {
104    [payload_path(seq), sidecar_path(seq)]
105}
106
107fn payload_path(seq: u64) -> String {
108    format!("{DIR}/{seq:06}.kdl.gz")
109}
110
111fn sidecar_path(seq: u64) -> String {
112    format!("{DIR}/{seq:06}.json")
113}
114
115/// The sequence number in a history file name, whichever of the pair it is.
116fn seq_of(name: &str) -> Option<u64> {
117    name.split('.').next()?.parse().ok()
118}
119
120/// The next unused sequence number, read from what is already there.
121///
122/// Presentation rather than remembered: the directory is the record, so a counter
123/// cannot drift from it, and a container that was edited elsewhere continues
124/// rather than overwriting.
125pub fn next_seq(storage: &impl Storage) -> std::io::Result<u64> {
126    let highest = storage
127        .list(DIR)?
128        .iter()
129        .filter_map(|name| seq_of(name))
130        .max();
131    Ok(highest.map_or(0, |n| n + 1))
132}
133
134/// Append `document` as entry `seq`. The caller owns the numbering (the writer
135/// holds it, so a burst of edits does not re-scan the directory per entry).
136pub fn append(
137    storage: &impl Storage,
138    seq: u64,
139    meta: &Entry,
140    document: &str,
141) -> std::io::Result<()> {
142    let mut encoder = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default());
143    encoder.write_all(document.as_bytes())?;
144    let payload = encoder.finish()?;
145    storage.write(&payload_path(seq), &payload, Durability::Relaxed)?;
146
147    // Second, always: a sidecar without its payload would be an index entry
148    // pointing at nothing, where a payload without its sidecar is merely
149    // unlabelled.
150    let sidecar = serde_json::to_vec(meta).map_err(std::io::Error::other)?;
151    storage.write(&sidecar_path(seq), &sidecar, Durability::Relaxed)
152}
153
154/// Every entry, oldest first. A sidecar that fails to parse reads as absent
155/// rather than failing the listing — one damaged entry must not hide the rest
156/// of the history.
157#[cfg_attr(not(test), allow(dead_code))]
158pub fn records(storage: &impl Storage) -> std::io::Result<Vec<Record>> {
159    let names = storage.list(DIR)?;
160    let mut seqs: Vec<u64> = names
161        .iter()
162        .filter(|n| n.ends_with(".kdl.gz"))
163        .filter_map(|n| seq_of(n))
164        .collect();
165    seqs.sort_unstable();
166    Ok(seqs
167        .into_iter()
168        .map(|seq| Record {
169            seq,
170            meta: storage
171                .read(&sidecar_path(seq))
172                .ok()
173                .and_then(|bytes| serde_json::from_slice(&bytes).ok()),
174        })
175        .collect())
176}
177
178/// The document stored as entry `seq`.
179#[cfg_attr(not(test), allow(dead_code))]
180pub fn read(storage: &impl Storage, seq: u64) -> std::io::Result<String> {
181    use std::io::Read as _;
182    let payload = storage.read(&payload_path(seq))?;
183    let mut out = String::new();
184    // The checksum is why a torn entry is detectable at all: a truncated
185    // payload fails here instead of decoding to a plausible half-document.
186    flate2::read::GzDecoder::new(&payload[..]).read_to_string(&mut out)?;
187    Ok(out)
188}
189
190/// Milliseconds since the Unix epoch.
191#[cfg(not(target_arch = "wasm32"))]
192pub fn now_millis() -> u64 {
193    std::time::SystemTime::now()
194        .duration_since(std::time::UNIX_EPOCH)
195        .map_or(0, |d| d.as_millis() as u64)
196}
197
198/// The browser's clock; `SystemTime::now` panics on wasm.
199#[cfg(target_arch = "wasm32")]
200pub fn now_millis() -> u64 {
201    js_sys::Date::now() as u64
202}
203
204#[cfg(test)]
205mod tests {
206    use super::*;
207    use crate::storage::atomic::tests::TempDir;
208    use crate::storage::fs::FsStorage;
209
210    fn storage(name: &str) -> (TempDir, FsStorage) {
211        let dir = TempDir::new(name);
212        let s = FsStorage::new(dir.path());
213        (dir, s)
214    }
215
216    fn meta(changed: &[&str]) -> Entry {
217        Entry {
218            ts: 1_700_000_000_000,
219            command: Some("delete".to_string()),
220            changed: changed.iter().map(|s| (*s).to_string()).collect(),
221        }
222    }
223
224    #[test]
225    fn an_entry_round_trips() {
226        let (_dir, s) = storage("history-roundtrip");
227        append(&s, 0, &meta(&["b3"]), "top \"b0\"").unwrap();
228        assert_eq!(read(&s, 0).unwrap(), "top \"b0\"");
229        assert_eq!(
230            records(&s).unwrap(),
231            vec![Record {
232                seq: 0,
233                meta: Some(meta(&["b3"])),
234            }]
235        );
236    }
237
238    #[test]
239    fn a_container_with_no_history_reads_as_empty() {
240        let (_dir, s) = storage("history-empty");
241        assert_eq!(next_seq(&s).unwrap(), 0);
242        assert!(records(&s).unwrap().is_empty());
243    }
244
245    /// The numbering is derived from the directory, so re-opening a container
246    /// continues rather than overwriting what is there.
247    #[test]
248    fn numbering_resumes_from_what_is_on_disk() {
249        let (_dir, s) = storage("history-resume");
250        for seq in 0..3 {
251            append(&s, seq, &meta(&[]), "doc").unwrap();
252        }
253        assert_eq!(next_seq(&s).unwrap(), 3);
254
255        let reopened = FsStorage::new(s.root());
256        assert_eq!(next_seq(&reopened).unwrap(), 3);
257    }
258
259    #[test]
260    fn entries_come_back_oldest_first() {
261        let (_dir, s) = storage("history-order");
262        for seq in [2, 0, 1] {
263            append(&s, seq, &meta(&[]), "doc").unwrap();
264        }
265        let seqs: Vec<u64> = records(&s).unwrap().into_iter().map(|r| r.seq).collect();
266        assert_eq!(seqs, vec![0, 1, 2]);
267    }
268
269    /// The pair can only tear one way. The payload is written first, so an
270    /// interrupted append leaves a readable entry that is merely unlabelled —
271    /// never an index entry pointing at a snapshot that is not there.
272    #[test]
273    fn an_entry_without_its_sidecar_is_unlabelled_not_lost() {
274        let (dir, s) = storage("history-torn");
275        append(&s, 0, &meta(&["b3"]), "the document").unwrap();
276        std::fs::remove_file(dir.path().join(sidecar_path(0))).unwrap();
277
278        assert_eq!(
279            records(&s).unwrap(),
280            vec![Record { seq: 0, meta: None }],
281            "the entry vanished with its label"
282        );
283        assert_eq!(read(&s, 0).unwrap(), "the document");
284    }
285
286    /// One damaged sidecar must not hide the rest of the history.
287    #[test]
288    fn a_corrupt_sidecar_does_not_hide_the_other_entries() {
289        let (dir, s) = storage("history-bad-sidecar");
290        append(&s, 0, &meta(&["b1"]), "one").unwrap();
291        append(&s, 1, &meta(&["b2"]), "two").unwrap();
292        std::fs::write(dir.path().join(sidecar_path(0)), b"{ not json").unwrap();
293
294        let found = records(&s).unwrap();
295        assert_eq!(found.len(), 2);
296        assert!(found[0].meta.is_none());
297        assert_eq!(found[1].meta, Some(meta(&["b2"])));
298    }
299
300    /// What makes a relaxed write safe: a truncated payload fails its checksum
301    /// rather than decoding into a plausible half-document.
302    #[test]
303    fn a_truncated_payload_is_an_error_not_half_a_document() {
304        let (dir, s) = storage("history-truncated");
305        append(
306            &s,
307            0,
308            &meta(&[]),
309            "top \"b0\"\nblock \"b0\" x=0 y=0 w=8 h=8",
310        )
311        .unwrap();
312        let path = dir.path().join(payload_path(0));
313        let whole = std::fs::read(&path).unwrap();
314        // Proves the precondition: there is enough here for a partial read to
315        // have plausibly succeeded.
316        assert!(
317            whole.len() > 8,
318            "payload too small to truncate meaningfully"
319        );
320        std::fs::write(&path, &whole[..whole.len() - 4]).unwrap();
321
322        assert!(read(&s, 0).is_err());
323        // The entry is still listed — a damaged snapshot is not an invisible one.
324        assert_eq!(records(&s).unwrap().len(), 1);
325    }
326
327    #[test]
328    fn sessions_accumulate_across_openings() {
329        let (_dir, s) = storage("history-sessions");
330        assert!(sessions(&s).is_empty(), "nothing opened yet");
331
332        record_session(&s, &Session { ts: 1, seq: 0 }).unwrap();
333        record_session(&s, &Session { ts: 2, seq: 7 }).unwrap();
334        assert_eq!(
335            sessions(&s),
336            vec![Session { ts: 1, seq: 0 }, Session { ts: 2, seq: 7 }],
337            "a later opening must not replace the earlier ones"
338        );
339    }
340
341    /// A damaged line must not hide the sessions around it — the same rule the
342    /// entry sidecars follow.
343    #[test]
344    fn a_damaged_session_line_is_skipped() {
345        let (dir, s) = storage("history-sessions-damaged");
346        record_session(&s, &Session { ts: 1, seq: 0 }).unwrap();
347        record_session(&s, &Session { ts: 2, seq: 7 }).unwrap();
348
349        let path = dir.path().join("history/sessions.jsonl");
350        let good = std::fs::read_to_string(&path).unwrap();
351        let mut lines: Vec<&str> = good.lines().collect();
352        lines.insert(1, "{ not json");
353        std::fs::write(&path, lines.join("\n")).unwrap();
354
355        assert_eq!(
356            sessions(&s),
357            vec![Session { ts: 1, seq: 0 }, Session { ts: 2, seq: 7 }]
358        );
359    }
360
361    /// Catches the classic unit mix-up: seconds would land in 1970 and read as
362    /// long before 2020, which is what the lower bound is for.
363    #[test]
364    fn the_clock_reads_milliseconds_not_seconds() {
365        let now = now_millis();
366        assert!(now > 1_577_836_800_000, "{now} predates 2020");
367        assert!(now < 4_102_444_800_000, "{now} postdates 2100");
368    }
369
370    #[test]
371    fn snapshots_are_deflated() {
372        let (dir, s) = storage("history-deflated");
373        let document = "top \"b0\"\n".repeat(500);
374        append(&s, 0, &meta(&[]), &document).unwrap();
375        let stored = std::fs::metadata(dir.path().join(payload_path(0)))
376            .unwrap()
377            .len();
378        assert!(
379            stored < document.len() as u64 / 4,
380            "{stored} bytes for a {} byte document",
381            document.len()
382        );
383    }
384}
blockworx/storage/history.rs

blockworx/storage/
history.rs
