shifathosn
/
lightcraft
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Issue #37: Catalog: stream the snapshot into the temp file instead of building two strings 

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 <noreply@anthropic.com>
This commit is contained in:
Brandon Thomas 2026-10-04 15:19:06 -04:00
parent db6242d2d3
commit ae1cbbe87b
4 changed files with 153 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
crates/catalog/examples/persist_bench.rs
View File

@ -7,7 +7,8 @@
//!
//! `<dir>` 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<dyn std::error::Error>> {
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));

50
crates/catalog/src/journal.rs
View File

@ -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":<serde_json of the 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<SnapshotTiming> {
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;

55
crates/catalog/src/store.rs
View File

@ -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<u64> {
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<u64> {
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<W> {
inner: W,
n: u64,
}
impl<W: io::Write> io::Write for Counter<W> {
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
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 {

45
crates/catalog/src/tests_journal.rs
View File

@ -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);
}