From ae1cbbe87b8fe0bbb34f3835c8b148b75eb7ef99 Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Sun, 4 Oct 2026 15:19:06 -0400 Subject: [PATCH] Issue #37: Catalog: stream the snapshot into the temp file instead of building two strings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Journal::snapshot serialised the catalog into a String and then copied it into a second `format!`-ed String before writing (2x the snapshot size in memory, ~620 MB at 85k photos). It now streams `{"format":…,"seq":N,"catalog":` + serde_json::to_writer + `}\n` through a 1 MiB BufWriter into the store (new `Store::write_atomic_with`; FsStore writes the temp file directly, then flush + sync_all + rename + dir fsync as before; other stores buffer, as before). Durability and format are unchanged: same temp + fsync + atomic rename; a failed write leaves the old snapshot (and now removes the temp file); the bytes are identical to the old format (test `streamed_snapshot_keeps_the_on_disk_format`, which also loads a legacy snapshot), so no version bump. Also tested: `failed_streamed_write_keeps_the_old_file`. persist_bench, 85k photos (310 MB snapshot), compaction at the 2,000-record threshold, before/after binaries alternated in the same session (load avg ~130-170), building+compacting only (PERSIST_BENCH_NO_LOAD): compaction median: before 3.5 s / 4.6 s, after 1.9 s / 2.5 s peak RSS: before 926 MB / 1263 MB, after 308 MB / 366 MB 10k photos peak RSS: 149-235 MB -> 61-68 MB (times too noisy to compare) Co-Authored-By: Claude Opus 5.5 --- crates/catalog/examples/persist_bench.rs | 17 +++++++- crates/catalog/src/journal.rs | 50 +++++++++++++++------ crates/catalog/src/store.rs | 55 ++++++++++++++++++++++++ crates/catalog/src/tests_journal.rs | 45 +++++++++++++++++++ 4 files changed, 153 insertions(+), 14 deletions(-) diff --git a/crates/catalog/examples/persist_bench.rs b/crates/catalog/examples/persist_bench.rs index e59f3e66..addf3a58 100644 --- a/crates/catalog/examples/persist_bench.rs +++ b/crates/catalog/examples/persist_bench.rs @@ -7,7 +7,8 @@ //! //! `` is a scratch directory (wiped per size). Default sizes: 1000 10000 85000. 90 % of the //! photos are Local browse records, the rest library photos with a few edits and history steps. -//! Peak memory: run under `/usr/bin/time -l` (macOS) / `-v` (Linux) with one size. +//! Peak memory: run under `/usr/bin/time -l` (macOS) / `-v` (Linux) with one size; +//! `PERSIST_BENCH_NO_LOAD=1` skips the final load, so the peak is building + compacting. use std::sync::Arc; use std::time::Instant; @@ -21,7 +22,15 @@ fn ms(t: Instant) -> f64 { fn photo(c: &mut Catalog, i: u64) -> Photo { let id = c.alloc_photo_id(); - let mut p = Photo::new(id, Source::File { path: format!("/Users/someone/Pictures/2026/09/{:05}/IMG_{i:05}.CR3", i / 500) }, &format!("IMG_{i:05}.CR3"), "CR3", 6000, 4000, "2026-09-30T12:00:00"); + let mut p = Photo::new( + id, + Source::File { path: format!("/Users/someone/Pictures/2026/09/{:05}/IMG_{i:05}.CR3", i / 500) }, + &format!("IMG_{i:05}.CR3"), + "CR3", + 6000, + 4000, + "2026-09-30T12:00:00", + ); p.kind = MediaKind::Raw; p.file_size = 25_000_000 + i; p.captured = Some(format!("2026-09-{:02}T10:{:02}:{:02}", 1 + i % 28, i % 60, (i / 60) % 60)); @@ -111,6 +120,10 @@ fn main() -> std::result::Result<(), Box> { println!(" compaction total: {}", stats(totals)); drop(j); + if std::env::var_os("PERSIST_BENCH_NO_LOAD").is_some() { + // peak memory of building + compacting only + continue; + } let t = Instant::now(); let (_, loaded, _) = Journal::open(Box::new(FsStore::open(&dir)?))?; println!(" open (load snapshot + replay): {:.1} ms", ms(t)); diff --git a/crates/catalog/src/journal.rs b/crates/catalog/src/journal.rs index 965fdc45..f0a5fcf8 100644 --- a/crates/catalog/src/journal.rs +++ b/crates/catalog/src/journal.rs @@ -78,9 +78,9 @@ pub struct PersistStats { #[derive(Clone, Copy, Debug, Default, PartialEq, serde::Serialize)] #[serde(rename_all = "camelCase")] pub struct SnapshotTiming { - /// Serialising the catalog to JSON. + /// Serialising the catalog to JSON, streamed into the temp file (buffered writes included). pub serialize_ms: f64, - /// Writing, flushing, `sync_all` and the atomic rename of `catalog.snap`. + /// Flushing, `sync_all` and the atomic rename of `catalog.snap`. pub write_sync_ms: f64, /// Resetting `catalog.log` (an atomic rewrite to empty, fsynced). pub reset_ms: f64, @@ -137,6 +137,29 @@ pub fn decode_record(line: &str) -> Option<(u64, Op)> { serde_json::from_str(body).ok().map(|op| (seq, op)) } +/// Write `catalog.snap` for the state after op `seq`, streaming the JSON into the store (no +/// whole-file string). The bytes are exactly +/// `{"format":"lightcraft-catalog","version":1,"seq":N,"catalog":}\n`, +/// as before streaming. Fills in the serialise / write+sync times and the size. +fn write_snapshot(store: &mut dyn Store, seq: u64, catalog: &Catalog) -> std::io::Result { + let t0 = web_time::Instant::now(); + let mut serialize_ms = 0.0; + let bytes = store.write_atomic_with(SNAPSHOT, &mut |w| { + use std::io::Write; + let t = web_time::Instant::now(); + // serde_json emits many tiny writes: buffer them in a concrete writer (inlined), so only + // 1 MiB chunks go through the store's `dyn Write` + let mut b = std::io::BufWriter::with_capacity(1 << 20, w); + write!(b, "{{\"format\":\"{FORMAT}\",\"version\":{VERSION},\"seq\":{seq},\"catalog\":")?; + serde_json::to_writer(&mut b, catalog).map_err(std::io::Error::other)?; + b.write_all(b"}\n")?; + b.flush()?; + serialize_ms = ms_since(t); + Ok(()) + })?; + Ok(SnapshotTiming { serialize_ms, write_sync_ms: (ms_since(t0) - serialize_ms).max(0.0), bytes, ..Default::default() }) +} + #[derive(serde::Deserialize)] struct SnapFile { format: String, @@ -164,7 +187,15 @@ impl Journal { None => (Catalog::new(), 0), }; report.snapshot_seq = snapshot_seq; - let mut j = Journal { store, seq: snapshot_seq, snapshot_seq, log_records: 0, log_bytes: 0, policy: SnapshotPolicy::default(), stats: PersistStats::default() }; + let mut j = Journal { + store, + seq: snapshot_seq, + snapshot_seq, + log_records: 0, + log_bytes: 0, + policy: SnapshotPolicy::default(), + stats: PersistStats::default(), + }; let log = log.unwrap_or_default(); // Scan records; `good_end` is the byte offset just past the last good record. @@ -262,18 +293,13 @@ impl Journal { /// Write a snapshot of `catalog` (which must reflect every appended op) and reset the log. pub fn snapshot(&mut self, catalog: &Catalog) -> Result<()> { let t0 = web_time::Instant::now(); - let body = serde_json::to_string(catalog).map_err(|e| CatalogError::Invalid(e.to_string()))?; - let file = format!("{{\"format\":\"{FORMAT}\",\"version\":{VERSION},\"seq\":{},\"catalog\":{body}}}\n", self.seq); - let serialize_ms = ms_since(t0); - let t1 = web_time::Instant::now(); - self.store.write_atomic(SNAPSHOT, file.as_bytes()).map_err(io)?; - let write_sync_ms = ms_since(t1); + let mut timing = write_snapshot(self.store.as_mut(), self.seq, catalog).map_err(io)?; let t2 = web_time::Instant::now(); // A crash here leaves old records in the log; they are skipped by seq on load. self.store.write_atomic(LOG, b"").map_err(io)?; - let reset_ms = ms_since(t2); - let timing = - SnapshotTiming { serialize_ms, write_sync_ms, reset_ms, total_ms: ms_since(t0), bytes: file.len() as u64, records: self.log_records }; + timing.reset_ms = ms_since(t2); + timing.total_ms = ms_since(t0); + timing.records = self.log_records; self.record_snapshot(timing); self.snapshot_seq = self.seq; self.log_records = 0; diff --git a/crates/catalog/src/store.rs b/crates/catalog/src/store.rs index 6207173a..56b9eb1a 100644 --- a/crates/catalog/src/store.rs +++ b/crates/catalog/src/store.rs @@ -16,6 +16,18 @@ pub trait Store: Send { /// Replace a file atomically: after a crash either the old or the new content is visible, /// never a mix (native: write temp + fsync + rename + fsync directory). fn write_atomic(&mut self, name: &str, data: &[u8]) -> io::Result<()>; + /// [`Store::write_atomic`] with streamed content: `fill` writes the new content into the + /// given writer (an error aborts and leaves the old file). Same durability: the file is + /// replaced only once all of it is written (native: and fsynced). Returns its length. + /// + /// The default buffers the content in memory; [`FsStore`] streams it into the temp file, so + /// a large snapshot is never held in memory as a whole. + fn write_atomic_with(&mut self, name: &str, fill: &mut dyn FnMut(&mut dyn io::Write) -> io::Result<()>) -> io::Result { + let mut buf = Vec::new(); + fill(&mut buf)?; + self.write_atomic(name, &buf)?; + Ok(buf.len() as u64) + } /// Append and make durable (fsync) before returning. fn append(&mut self, name: &str, data: &[u8]) -> io::Result<()>; /// Cut a file to `len` bytes (drop a torn tail before appending again). @@ -77,6 +89,32 @@ impl Store for FsStore { Ok(()) } + fn write_atomic_with(&mut self, name: &str, fill: &mut dyn FnMut(&mut dyn io::Write) -> io::Result<()>) -> io::Result { + use std::io::Write; + if self.appender.as_ref().is_some_and(|(n, _)| n == name) { + self.appender = None; + } + let tmp = self.path(&format!("{name}.tmp")); + let written = (|| { + let mut w = Counter { inner: io::BufWriter::with_capacity(1 << 20, std::fs::File::create(&tmp)?), n: 0 }; + fill(&mut w)?; + w.flush()?; + let f = w.inner.into_inner().map_err(|e| e.into_error())?; + f.sync_all()?; + Ok(w.n) + })(); + let n = match written { + Ok(n) => n, + Err(e) => { + let _ = std::fs::remove_file(&tmp); + return Err(e); + } + }; + std::fs::rename(&tmp, self.path(name))?; + self.sync_dir(); + Ok(n) + } + fn append(&mut self, name: &str, data: &[u8]) -> io::Result<()> { use std::io::Write; if self.appender.as_ref().is_none_or(|(n, _)| n != name) { @@ -106,6 +144,23 @@ impl Store for FsStore { } } +/// Counts the bytes written through it. +struct Counter { + inner: W, + n: u64, +} + +impl io::Write for Counter { + fn write(&mut self, buf: &[u8]) -> io::Result { + let k = self.inner.write(buf)?; + self.n += k as u64; + Ok(k) + } + fn flush(&mut self) -> io::Result<()> { + self.inner.flush() + } +} + /// In-memory files. Clones share the same files (so a test can "reopen" or corrupt them). #[derive(Clone, Default)] pub struct MemStore { diff --git a/crates/catalog/src/tests_journal.rs b/crates/catalog/src/tests_journal.rs index ec38af57..6bf7c85c 100644 --- a/crates/catalog/src/tests_journal.rs +++ b/crates/catalog/src/tests_journal.rs @@ -274,3 +274,48 @@ fn fs_store_roundtrip() { assert_eq!(c3.to_snapshot(), c.to_snapshot()); let _ = std::fs::remove_dir_all(&dir); } + +/// The streamed snapshot is byte-for-byte the format written before streaming (issue #37), so +/// existing libraries load and older builds can read new snapshots; FsStore and MemStore agree. +#[test] +fn streamed_snapshot_keeps_the_on_disk_format() { + let (m, _, full) = sample_catalog_with_log(); + let (mut j, c, _) = open(&m); + assert_eq!(c.to_snapshot(), full.to_snapshot()); + j.snapshot(&c).unwrap(); + let legacy = format!("{{\"format\":\"lightcraft-catalog\",\"version\":1,\"seq\":{},\"catalog\":{}}}\n", j.seq(), c.to_snapshot()); + assert_eq!(String::from_utf8(m.get(SNAPSHOT).unwrap()).unwrap(), legacy); + assert_eq!(j.stats().last_snapshot.bytes, legacy.len() as u64); + + let dir = std::env::temp_dir().join(format!("lc-journal-stream-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + // an old-style snapshot (written by a build before streaming) loads, and is rewritten the same + std::fs::write(dir.join(SNAPSHOT), legacy.as_bytes()).unwrap(); + let (mut fj, fc, r) = Journal::open(Box::new(FsStore::open(&dir).unwrap())).unwrap(); + assert_eq!((fc.to_snapshot(), r.snapshot_seq), (c.to_snapshot(), j.seq())); + fj.snapshot(&fc).unwrap(); + assert_eq!(std::fs::read(dir.join(SNAPSHOT)).unwrap(), legacy.as_bytes()); + drop(fj); + let _ = std::fs::remove_dir_all(&dir); +} + +/// A snapshot that fails while streaming (serialisation or I/O error) leaves the old file in +/// place and removes its temp file. +#[test] +fn failed_streamed_write_keeps_the_old_file() { + let dir = std::env::temp_dir().join(format!("lc-journal-streamfail-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + let mut s = FsStore::open(&dir).unwrap(); + s.write_atomic(SNAPSHOT, b"old").unwrap(); + let r = s.write_atomic_with(SNAPSHOT, &mut |w| { + w.write_all(b"partial new content")?; + Err(std::io::Error::other("serialisation failed")) + }); + assert!(r.is_err()); + assert_eq!(std::fs::read(dir.join(SNAPSHOT)).unwrap(), b"old"); + assert!(!dir.join(format!("{SNAPSHOT}.tmp")).exists()); + assert_eq!(s.write_atomic_with(SNAPSHOT, &mut |w| w.write_all(b"new")).unwrap(), 3); + assert_eq!(std::fs::read(dir.join(SNAPSHOT)).unwrap(), b"new"); + let _ = std::fs::remove_dir_all(&dir); +}