Issue #37: Catalog: compact the log on a background thread with ordered journal boundaries

At the 2,000-record threshold the UI thread serialised and fsynced the whole catalog (1-2 s
at 85k photos after streaming). Now `persist()` starts `Journal::snapshot_in_background`:

- the worker writes `catalog.snap` for op N from a clone of the catalog (photos are Arc-shared,
  so the clone is cheap) through `Store::background_writer` (a second FsStore handle; MemStore
  opt-in; none on wasm/web, which stays synchronous);
- appends go on as before (fsynced before every command returns) and are also kept in memory
  as the "tail" (records > N);
- `Journal::poll` (from `persist` and the per-frame `persist_if_dirty`) sees the worker done and
  atomically replaces the log with exactly the tail.

Crash points: before the snapshot rename -> old snapshot + whole log; between the rename and the
log rewrite -> new snapshot + whole log (records <= N skipped as stale, existing rule); after ->
new snapshot + tail. A failed snapshot keeps the log whole and retries after another
max_records records (no retry storm). At most one snapshot in flight; a synchronous snapshot
(close, Optimize Library), reopening a library and dropping the Journal first wait for it, so an
older snapshot can never replace a newer one. On-disk format unchanged.

Tests (crates/catalog/src/tests_background.rs, a gated store holds the worker mid-flight and
"crashes" by copying the files): edits during a background snapshot survive crashes at each
point and the trimmed log holds exactly seq N+1.. in order; failure loses nothing and backs
off; sync snapshot waits (no regression); drop waits; real FsStore with a leftover partial
temp file; a proptest over interleavings of edits / start / land / finish / sync snapshot /
crash. Mutations checked: trimming the log to empty, or not waiting in `snapshot`, fail them.
Engine: `threshold_compaction_runs_in_the_background` (crash copy mid-run, reopen without close).
`library.info` -> persistence gains snapshotRunning, failedSnapshots, maxBlockingMs and
lastSnapshot.{background, blockingMs}.

persist_bench, same session (load avg ~160), UI-thread time per threshold compaction:
  85k photos: synchronous 0.85-1.7 s (median 1.25 / 1.34 s) -> background: blocking median
              45 / 74 ms (clone 1.3-154 ms + log rewrite 9-23 ms); worker 1.2-2.4 s; 230-380
              single-op appends during it at median 4-6 ms (same as without)
  10k photos: synchronous 72-233 ms -> blocking median 12.5 / 14 ms

Co-Authored-By: Claude Opus 5.5 <[email protected]>

crates/catalog/examples/persist_bench.rs

@@ -1,5 +1,6 @@
 //! Catalog persistence costs on large synthetic catalogs (issue #37): log append, compaction
-//! (snapshot serialise / write+fsync / log reset), load, and a catalog clone.
+//! (snapshot serialise / write+fsync / log reset), synchronous and on the background worker, load,
+//! and a catalog clone.
 //!
 //! ```text
 //! cargo run --release -p lightcraft-catalog --example persist_bench -- <dir> [photos…]
@@ -119,6 +120,39 @@ fn main() -> std::result::Result<(), Box<dyn std::error::Error>> {
         }
         println!("  compaction total: {}", stats(totals));
 
+        // the same at the threshold, written by the worker while single-op appends go on
+        let mut blocking = Vec::new();
+        for _ in 0..3 {
+            while !j.wants_snapshot() {
+                let op = Op::SetRating { id: ids.first().copied().unwrap_or(PhotoId(1)), rating: (j.seq() % 6) as u8 };
+                cat.apply(op.clone())?;
+                j.append(&[op])?;
+            }
+            let t = Instant::now();
+            j.snapshot_in_background(&cat)?;
+            let start_ms = ms(t);
+            let mut during = Vec::new();
+            while !j.snapshot_written() {
+                let op = Op::SetRating { id: ids.get(1).copied().unwrap_or(PhotoId(1)), rating: (j.seq() % 6) as u8 };
+                cat.apply(op.clone())?;
+                j.append(&[op])?;
+                during.push(j.stats().last_append_ms);
+            }
+            let t = Instant::now();
+            j.poll()?;
+            let finish_ms = ms(t);
+            let s = j.stats().last_snapshot;
+            println!(
+                "  background compaction ({} records): start {start_ms:.1} ms + finish {finish_ms:.1} ms blocking; worker {:.1} ms; {} appends during it: {}",
+                s.records,
+                s.total_ms,
+                during.len(),
+                stats(during)
+            );
+            blocking.push(s.blocking_ms);
+        }
+        println!("  background compaction blocking: {}", stats(blocking));
+
         drop(j);
         if std::env::var_os("PERSIST_BENCH_NO_LOAD").is_some() {
             // peak memory of building + compacting only

crates/catalog/proptest-regressions/tests_background.txt

@@ -0,0 +1,8 @@
+# Seeds for failure cases proptest has generated in the past. It is
+# automatically read and these particular cases re-run before any
+# novel cases are generated.
+#
+# It is recommended to check this file in to source control so that
+# everyone who runs the test benefits from these saved cases.
+cc 0227748e171237a13833c9175fe02c7fe6ff332266798961a6a43ec37b81d1b4 # shrinks to actions = [(3, 0), (0, 0), (4, 0), (0, 0)]
+cc 0b0ab2531bd46fbe261b55ecffebea9d3e8949900ab7bc33fb27ca19f5cd7b8b # shrinks to actions = [(4, 443), (3, 474), (2, 27), (5, 858), (5, 466), (0, 901), (4, 907), (3, 839), (0, 630), (7, 578), (4, 13), (2, 361), (6, 136), (6, 12), (6, 223), (1, 394), (6, 78), (5, 837), (6, 765), (4, 207), (2, 185), (2, 583), (2, 56), (5, 430), (1, 702), (1, 547), (5, 552), (4, 762), (0, 443), (0, 851), (7, 240), (4, 438), (3, 599), (6, 70), (4, 694), (1, 327), (7, 859), (6, 814), (4, 336), (3, 438), (0, 131), (0, 831), (3, 18), (2, 963), (6, 10), (3, 114)]

crates/catalog/src/journal.rs

@@ -14,6 +14,19 @@
 //!   the log) are skipped;
 //! - a bad record **followed by good ones** is real damage: replay stops there, the log is kept
 //!   as `catalog.log.damaged-<seq>`, and a fresh snapshot is written so the library stays usable.
+//!
+//! **Background compaction** ([`Journal::snapshot_in_background`], when the store offers a
+//! [`Store::background_writer`]): a worker thread writes the snapshot of a copy of the catalog at
+//! op `N` while appends go on (durably, as always) to the same log. Once the snapshot is durable,
+//! [`Journal::poll`] atomically replaces the log with the records appended since the snapshot
+//! started (`seq > N`, kept in memory). Every crash point is covered by the rules above:
+//! - before the rename of `catalog.snap`: the old snapshot + the whole log (nothing was removed);
+//! - between the rename and the log rewrite: the new snapshot + the whole log, whose records
+//!   `<= N` are skipped as stale;
+//! - after the log rewrite: the new snapshot + exactly the records `> N`.
+//!
+//! At most one snapshot is in flight; a synchronous [`Journal::snapshot`] and dropping the journal
+//! first wait for it, so an older snapshot can never replace a newer one.
 
 use crate::store::Store;
 use crate::{Catalog, CatalogError, Op, Result};
@@ -72,6 +85,13 @@ pub struct PersistStats {
     pub last_snapshot: SnapshotTiming,
     /// Total time of the slowest snapshot.
     pub max_snapshot_ms: f64,
+    /// Longest time a snapshot blocked the caller (for background snapshots: copying the catalog
+    /// and starting the worker, plus rewriting the log when it is done).
+    pub max_blocking_ms: f64,
+    /// A background snapshot is being written.
+    pub snapshot_running: bool,
+    /// Background snapshots that failed (the log was kept whole; retried later).
+    pub failed_snapshots: u64,
 }
 
 /// One snapshot (compaction), by stage.
@@ -82,9 +102,15 @@ pub struct SnapshotTiming {
     pub serialize_ms: f64,
     /// Flushing, `sync_all` and the atomic rename of `catalog.snap`.
     pub write_sync_ms: f64,
-    /// Resetting `catalog.log` (an atomic rewrite to empty, fsynced).
+    /// Resetting `catalog.log` (an atomic rewrite to empty, or to the records appended while a
+    /// background snapshot ran; fsynced).
     pub reset_ms: f64,
+    /// Start to finish (for a background snapshot: until [`Journal::poll`] saw it done).
     pub total_ms: f64,
+    /// How long the caller was blocked (equal to `total_ms` unless `background`).
+    pub blocking_ms: f64,
+    /// Written by a worker thread.
+    pub background: bool,
     /// Size of `catalog.snap`.
     pub bytes: u64,
     /// Log records the snapshot compacted.
@@ -110,6 +136,25 @@ pub struct Journal {
     log_bytes: u64,
     pub policy: SnapshotPolicy,
     stats: PersistStats,
+    /// The background snapshot in flight.
+    pending: Option<Pending>,
+    /// After a failed background snapshot: don't retry before the log has this many records.
+    retry_at_records: u64,
+}
+
+/// A background snapshot in flight.
+struct Pending {
+    /// The snapshot holds the state after this op.
+    seq: u64,
+    /// Log bytes / records appended since it started (all `> seq`): the log once it's done.
+    tail: Vec<u8>,
+    tail_records: u64,
+    /// Log records it compacts.
+    records: u64,
+    started: web_time::Instant,
+    /// Time the caller spent starting it.
+    start_ms: f64,
+    worker: std::thread::JoinHandle<std::io::Result<SnapshotTiming>>,
 }
 
 fn io(e: std::io::Error) -> CatalogError {
@@ -195,6 +240,8 @@ impl Journal {
             log_bytes: 0,
             policy: SnapshotPolicy::default(),
             stats: PersistStats::default(),
+            pending: None,
+            retry_at_records: 0,
         };
         let log = log.unwrap_or_default();
 
@@ -277,6 +324,10 @@ impl Journal {
             buf.push('\n');
         }
         self.store.append(LOG, buf.as_bytes()).map_err(io)?;
+        if let Some(p) = self.pending.as_mut() {
+            p.tail.extend_from_slice(buf.as_bytes());
+            p.tail_records += ops.len() as u64;
+        }
         self.seq = seq;
         self.log_records += ops.len() as u64;
         self.log_bytes += buf.len() as u64;
@@ -292,6 +343,10 @@ 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<()> {
+        // never two snapshot writers; the one in flight is older, so it must land first
+        if let Err(e) = self.wait() {
+            log::warn!("catalog: background snapshot failed ({e}); writing one now");
+        }
         let t0 = web_time::Instant::now();
         let mut timing = write_snapshot(self.store.as_mut(), self.seq, catalog).map_err(io)?;
         let t2 = web_time::Instant::now();
@@ -299,11 +354,103 @@ impl Journal {
         self.store.write_atomic(LOG, b"").map_err(io)?;
         timing.reset_ms = ms_since(t2);
         timing.total_ms = ms_since(t0);
+        timing.blocking_ms = timing.total_ms;
         timing.records = self.log_records;
         self.record_snapshot(timing);
         self.snapshot_seq = self.seq;
         self.log_records = 0;
         self.log_bytes = 0;
+        self.retry_at_records = 0;
+        Ok(())
+    }
+
+    /// Compact like [`Journal::snapshot`], but write the snapshot on a worker thread from a copy
+    /// of `catalog` (cheap: photos are shared) when the store supports it (else synchronously).
+    /// `catalog` must reflect every appended op. Appends go on meanwhile; [`Journal::poll`]
+    /// finishes the compaction. No-op while a snapshot is already in flight.
+    pub fn snapshot_in_background(&mut self, catalog: &Catalog) -> Result<()> {
+        if self.pending.is_some() {
+            return Ok(());
+        }
+        let t0 = web_time::Instant::now();
+        let Some(mut writer) = self.store.background_writer() else { return self.snapshot(catalog) };
+        let copy = catalog.clone();
+        let seq = self.seq;
+        let spawned = std::thread::Builder::new().name("catalog-snapshot".into()).spawn(move || write_snapshot(writer.as_mut(), seq, &copy));
+        let worker = match spawned {
+            Ok(w) => w,
+            Err(e) => {
+                log::warn!("catalog: can't start a snapshot thread ({e}); writing it now");
+                return self.snapshot(catalog);
+            }
+        };
+        self.pending =
+            Some(Pending { seq, tail: Vec::new(), tail_records: 0, records: self.log_records, started: t0, start_ms: ms_since(t0), worker });
+        Ok(())
+    }
+
+    /// Finish a background snapshot if its worker is done (cheap otherwise). Returns whether one
+    /// finished. An error means the snapshot failed: nothing was lost (the log is whole), and
+    /// compaction is retried once the log has grown by another [`SnapshotPolicy::max_records`].
+    pub fn poll(&mut self) -> Result<bool> {
+        if !self.pending.as_ref().is_some_and(|p| p.worker.is_finished()) {
+            return Ok(false);
+        }
+        self.finish().map(|()| true)
+    }
+
+    /// Wait for the background snapshot in flight, if any, and finish it.
+    pub fn wait(&mut self) -> Result<()> {
+        if self.pending.is_none() {
+            return Ok(());
+        }
+        self.finish()
+    }
+
+    /// A background snapshot is in flight (until [`Journal::poll`] or [`Journal::wait`] finish it).
+    pub fn snapshot_running(&self) -> bool {
+        self.pending.is_some()
+    }
+
+    /// The background worker is done (its result not yet applied by [`Journal::poll`]).
+    pub fn snapshot_written(&self) -> bool {
+        self.pending.as_ref().is_some_and(|p| p.worker.is_finished())
+    }
+
+    /// Join the worker; on success replace the log with the records appended meanwhile.
+    fn finish(&mut self) -> Result<()> {
+        let Some(p) = self.pending.take() else { return Ok(()) };
+        let t0 = web_time::Instant::now();
+        let written = p.worker.join().unwrap_or_else(|_| Err(std::io::Error::other("snapshot thread panicked")));
+        let mut timing = match written {
+            Ok(t) => t,
+            Err(e) => {
+                // the log still holds every record: nothing lost, retry later
+                self.stats.failed_snapshots += 1;
+                self.retry_at_records = self.log_records + self.policy.max_records;
+                log::error!("catalog: background snapshot failed: {e}");
+                return Err(io(e));
+            }
+        };
+        // `catalog.snap` (op `p.seq`) is durable; the log's records `<= p.seq` are now stale.
+        // Keep exactly the ones appended since (a crash before this rewrite skips the stale ones).
+        let t1 = web_time::Instant::now();
+        self.snapshot_seq = p.seq;
+        if let Err(e) = self.store.write_atomic(LOG, &p.tail) {
+            // the log stays whole (stale records are skipped on load); compact again later
+            self.retry_at_records = self.log_records + self.policy.max_records;
+            log::error!("catalog: can't trim the log after a snapshot: {e}");
+            return Err(io(e));
+        }
+        timing.reset_ms = ms_since(t1);
+        timing.total_ms = ms_since(p.started);
+        timing.blocking_ms = p.start_ms + ms_since(t0);
+        timing.background = true;
+        timing.records = p.records;
+        self.record_snapshot(timing);
+        self.log_records = p.tail_records;
+        self.log_bytes = p.tail.len() as u64;
+        self.retry_at_records = 0;
         Ok(())
     }
 
@@ -311,22 +458,32 @@ impl Journal {
         self.stats.snapshots += 1;
         self.stats.last_snapshot = t;
         self.stats.max_snapshot_ms = self.stats.max_snapshot_ms.max(t.total_ms);
+        self.stats.max_blocking_ms = self.stats.max_blocking_ms.max(t.blocking_ms);
         if profiling() {
             eprintln!(
-                "catalog: snapshot of {} records, {} B: serialize {:.1} ms, write+sync {:.1} ms, log reset {:.1} ms, total {:.1} ms",
-                t.records, t.bytes, t.serialize_ms, t.write_sync_ms, t.reset_ms, t.total_ms
+                "catalog: {}snapshot of {} records, {} B: serialize {:.1} ms, write+sync {:.1} ms, log reset {:.1} ms, total {:.1} ms, blocking {:.1} ms",
+                if t.background { "background " } else { "" },
+                t.records,
+                t.bytes,
+                t.serialize_ms,
+                t.write_sync_ms,
+                t.reset_ms,
+                t.total_ms,
+                t.blocking_ms
             );
         }
     }
 
     /// Where persistence time went so far.
     pub fn stats(&self) -> PersistStats {
-        self.stats
+        PersistStats { snapshot_running: self.pending.is_some(), ..self.stats }
     }
 
     /// The log is long enough to be worth compacting.
     pub fn wants_snapshot(&self) -> bool {
-        self.log_records >= self.policy.max_records || self.log_bytes >= self.policy.max_bytes
+        self.pending.is_none()
+            && self.log_records >= self.retry_at_records
+            && (self.log_records >= self.policy.max_records || self.log_bytes >= self.policy.max_bytes)
     }
 
     pub fn seq(&self) -> u64 {
@@ -345,3 +502,16 @@ impl Journal {
         self.store.describe()
     }
 }
+
+impl Drop for Journal {
+    /// Wait for a background snapshot (never leave a writer behind that could replace a newer
+    /// snapshot of a journal reopened on the same files). The log is left whole: the snapshot's
+    /// records in it are skipped as stale on the next load.
+    fn drop(&mut self) {
+        if let Some(p) = self.pending.take()
+            && let Ok(Err(e)) = p.worker.join()
+        {
+            log::error!("catalog: background snapshot failed: {e}");
+        }
+    }
+}

crates/catalog/src/lib.rs

@@ -622,4 +622,6 @@ impl Catalog {
 #[cfg(test)]
 mod tests;
 #[cfg(test)]
+mod tests_background;
+#[cfg(test)]
 mod tests_journal;

crates/catalog/src/store.rs

@@ -34,6 +34,12 @@ pub trait Store: Send {
     fn truncate(&mut self, name: &str, len: u64) -> io::Result<()>;
     /// Human-readable location (diagnostics).
     fn describe(&self) -> String;
+    /// Another handle on the same files that a worker thread can write `catalog.snap` through
+    /// while this one keeps appending to the log (background compaction). `None` (the default):
+    /// snapshots are written synchronously.
+    fn background_writer(&self) -> Option<Box<dyn Store>> {
+        None
+    }
 }
 
 /// Files in a directory.
@@ -142,6 +148,11 @@ impl Store for FsStore {
     fn describe(&self) -> String {
         self.dir.display().to_string()
     }
+
+    #[cfg(not(target_arch = "wasm32"))]
+    fn background_writer(&self) -> Option<Box<dyn Store>> {
+        Some(Box::new(FsStore { dir: self.dir.clone(), appender: None }))
+    }
 }
 
 /// Counts the bytes written through it.
@@ -165,6 +176,8 @@ impl<W: io::Write> io::Write for Counter<W> {
 #[derive(Clone, Default)]
 pub struct MemStore {
     pub files: Arc<Mutex<BTreeMap<String, Vec<u8>>>>,
+    /// Offer a [`Store::background_writer`] (off by default: snapshots stay synchronous).
+    pub background: bool,
 }
 
 impl MemStore {
@@ -200,4 +213,7 @@ impl Store for MemStore {
     fn describe(&self) -> String {
         "memory".into()
     }
+    fn background_writer(&self) -> Option<Box<dyn Store>> {
+        self.background.then(|| Box::new(self.clone()) as Box<dyn Store>)
+    }
 }

crates/catalog/src/tests_background.rs

@@ -0,0 +1,338 @@
+//! Background compaction (issue #37): a snapshot written by a worker while appends go on.
+//! "Crashes" are simulated by copying the files at a given moment and loading the copy: every
+//! acknowledged (appended) op must be there, in order, whatever the snapshot was doing.
+
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Arc, Condvar, Mutex};
+use std::time::Duration;
+
+use proptest::prelude::*;
+
+use crate::journal::{LOG, SNAPSHOT};
+use crate::*;
+
+/// A MemStore whose snapshot writes wait for a gate (to hold a background snapshot mid-flight)
+/// and can be made to fail.
+#[derive(Clone)]
+struct Gated {
+    files: MemStore,
+    gate: Arc<(Mutex<bool>, Condvar)>,
+    fail: Arc<AtomicBool>,
+}
+
+impl Gated {
+    fn new() -> Gated {
+        Gated { files: MemStore::new(), gate: Arc::new((Mutex::new(false), Condvar::new())), fail: Arc::new(AtomicBool::new(false)) }
+    }
+    fn set_gate(&self, open: bool) {
+        *self.gate.0.lock().unwrap() = open;
+        self.gate.1.notify_all();
+    }
+    /// The files as a crash at this moment would leave them.
+    fn crash(&self) -> MemStore {
+        let m = MemStore::new();
+        *m.files.lock().unwrap() = self.files.files.lock().unwrap().clone();
+        m
+    }
+    fn snapshot_seq_on_disk(&self) -> Option<u64> {
+        let v: serde_json::Value = serde_json::from_slice(&self.files.get(SNAPSHOT)?).ok()?;
+        v["seq"].as_u64()
+    }
+}
+
+impl Store for Gated {
+    fn read(&mut self, name: &str) -> std::io::Result<Option<Vec<u8>>> {
+        self.files.read(name)
+    }
+    fn write_atomic(&mut self, name: &str, data: &[u8]) -> std::io::Result<()> {
+        self.files.write_atomic(name, data)
+    }
+    fn write_atomic_with(&mut self, name: &str, fill: &mut dyn FnMut(&mut dyn std::io::Write) -> std::io::Result<()>) -> std::io::Result<u64> {
+        if name == SNAPSHOT {
+            let (lock, cv) = &*self.gate;
+            let (open, timeout) = cv.wait_timeout_while(lock.lock().unwrap(), Duration::from_secs(20), |open| !*open).unwrap();
+            if !*open || timeout.timed_out() {
+                return Err(std::io::Error::other("gate never opened"));
+            }
+        }
+        let mut buf = Vec::new();
+        fill(&mut buf)?;
+        if self.fail.load(Ordering::SeqCst) {
+            return Err(std::io::Error::other("disk full"));
+        }
+        self.files.write_atomic(name, &buf)?;
+        Ok(buf.len() as u64)
+    }
+    fn append(&mut self, name: &str, data: &[u8]) -> std::io::Result<()> {
+        self.files.append(name, data)
+    }
+    fn truncate(&mut self, name: &str, len: u64) -> std::io::Result<()> {
+        self.files.truncate(name, len)
+    }
+    fn describe(&self) -> String {
+        "gated".into()
+    }
+    fn background_writer(&self) -> Option<Box<dyn Store>> {
+        Some(Box::new(self.clone()))
+    }
+}
+
+/// Apply and durably append one edit (two ops in one append every fourth time).
+fn edit(j: &mut Journal, live: &mut Catalog, k: u64) {
+    let one = |live: &mut Catalog, k: u64| {
+        let op = if live.len() < 4 || k.is_multiple_of(3) {
+            let id = live.alloc_photo_id();
+            Op::AddPhoto { photo: Box::new(Photo::new(id, Source::File { path: format!("/p/{}.jpg", id.0) }, "p.jpg", "JPEG", 4, 3, "t")) }
+        } else {
+            Op::SetRating { id: PhotoId(1 + k % live.len() as u64), rating: (k % 6) as u8 }
+        };
+        live.apply(op.clone()).unwrap();
+        op
+    };
+    let mut ops = vec![one(live, k)];
+    if k.is_multiple_of(4) {
+        ops.push(one(live, k / 4 + 1));
+    }
+    j.append(&ops).unwrap();
+}
+
+/// Load the files a crash now would leave: they must hold exactly the live state.
+fn assert_recovers(g: &Gated, live: &Catalog) -> LoadReport {
+    let (_, c, r) = Journal::open(Box::new(g.crash())).unwrap();
+    assert_eq!(c.to_snapshot(), live.to_snapshot(), "{r:?}");
+    assert_eq!((r.failed, r.torn_bytes, r.damaged.clone()), (0, 0, None));
+    r
+}
+
+fn wait_written(j: &Journal) {
+    for _ in 0..20_000 {
+        if j.snapshot_written() {
+            return;
+        }
+        std::thread::sleep(Duration::from_millis(1));
+    }
+    panic!("background snapshot never finished");
+}
+
+#[test]
+fn background_snapshot_keeps_edits_made_while_it_runs() {
+    let g = Gated::new();
+    let (mut j, mut live, _) = Journal::open(Box::new(g.clone())).unwrap();
+    for k in 0..30 {
+        edit(&mut j, &mut live, k);
+    }
+    let n = j.seq();
+    j.snapshot_in_background(&live).unwrap();
+    assert!(j.snapshot_running() && !j.wants_snapshot() && j.stats().snapshot_running);
+
+    // edits while the snapshot is being written are appended (durably) as usual
+    let mut during = 0;
+    for k in 30..50 {
+        edit(&mut j, &mut live, k);
+        during = j.seq() - n;
+    }
+    // crash before the snapshot landed: no snapshot yet, the whole log replays
+    let r = assert_recovers(&g, &live);
+    assert_eq!((r.snapshot_seq, r.replayed as u64), (0, j.seq()));
+
+    // the snapshot lands (op n); crash before the log was trimmed: its records are stale
+    g.set_gate(true);
+    wait_written(&j);
+    assert_eq!(g.snapshot_seq_on_disk(), Some(n));
+    for k in 50..55 {
+        edit(&mut j, &mut live, k);
+        during = j.seq() - n;
+    }
+    let r = assert_recovers(&g, &live);
+    assert_eq!((r.snapshot_seq, r.stale as u64, r.replayed as u64), (n, n, during));
+
+    // finishing trims the log to exactly the records after the snapshot, in order
+    assert!(j.poll().unwrap());
+    assert!(!j.snapshot_running());
+    assert_eq!((j.snapshot_seq(), j.log_records()), (n, during));
+    let log = String::from_utf8(g.files.get(LOG).unwrap()).unwrap();
+    let seqs: Vec<u64> = log.lines().map(|l| journal::decode_record(l).unwrap().0).collect();
+    assert_eq!(seqs, ((n + 1)..=j.seq()).collect::<Vec<_>>());
+    assert_eq!(j.log_bytes(), log.len() as u64);
+    let r = assert_recovers(&g, &live);
+    assert_eq!((r.snapshot_seq, r.stale, r.replayed as u64), (n, 0, during));
+    let s = j.stats();
+    assert!(s.last_snapshot.background && s.last_snapshot.records == n && !s.snapshot_running, "{s:?}");
+
+    // and appending goes on after it
+    edit(&mut j, &mut live, 99);
+    assert_recovers(&g, &live);
+}
+
+#[test]
+fn failed_background_snapshot_loses_nothing_and_retries_later() {
+    let g = Gated::new();
+    let (mut j, mut live, _) = Journal::open(Box::new(g.clone())).unwrap();
+    j.policy = SnapshotPolicy { max_records: 10, max_bytes: u64::MAX };
+    for k in 0..12 {
+        edit(&mut j, &mut live, k);
+    }
+    assert!(j.wants_snapshot());
+    g.fail.store(true, Ordering::SeqCst);
+    g.set_gate(true);
+    j.snapshot_in_background(&live).unwrap();
+    edit(&mut j, &mut live, 12);
+    wait_written(&j);
+    assert!(j.poll().is_err());
+    assert_eq!((j.snapshot_seq(), j.stats().failed_snapshots), (0, 1));
+    assert!(g.files.get(SNAPSHOT).is_none());
+    assert_recovers(&g, &live);
+    // no retry storm: the next attempt waits for another `max_records` records
+    assert!(!j.wants_snapshot());
+    g.fail.store(false, Ordering::SeqCst);
+    while !j.wants_snapshot() {
+        let k = j.seq();
+        edit(&mut j, &mut live, k);
+    }
+    j.snapshot_in_background(&live).unwrap();
+    j.wait().unwrap();
+    assert_eq!((j.snapshot_seq(), j.log_records()), (j.seq(), 0));
+    let r = assert_recovers(&g, &live);
+    assert_eq!((r.snapshot_seq, r.replayed), (j.seq(), 0));
+}
+
+/// A synchronous snapshot (close, "Optimize Library") waits for the background one, so the older
+/// snapshot can't land after (and replace) the newer one.
+#[test]
+fn sync_snapshot_waits_for_the_background_one() {
+    let g = Gated::new();
+    let (mut j, mut live, _) = Journal::open(Box::new(g.clone())).unwrap();
+    for k in 0..10 {
+        edit(&mut j, &mut live, k);
+    }
+    j.snapshot_in_background(&live).unwrap();
+    for k in 10..20 {
+        edit(&mut j, &mut live, k);
+    }
+    let opener = {
+        let g = g.clone();
+        std::thread::spawn(move || {
+            std::thread::sleep(Duration::from_millis(50));
+            g.set_gate(true);
+        })
+    };
+    j.snapshot(&live).unwrap();
+    opener.join().unwrap();
+    assert!(!j.snapshot_running());
+    assert_eq!(g.snapshot_seq_on_disk(), Some(j.seq()));
+    assert_eq!((j.snapshot_seq(), j.log_records()), (j.seq(), 0));
+    assert!(g.files.get(LOG).unwrap().is_empty());
+    assert_eq!(j.stats().snapshots, 2);
+    assert_recovers(&g, &live);
+}
+
+/// Dropping the journal (e.g. reopening the library) waits for the snapshot in flight and leaves
+/// the log whole: reopening finds every op.
+#[test]
+fn dropping_the_journal_waits_for_its_snapshot() {
+    let g = Gated::new();
+    let (mut j, mut live, _) = Journal::open(Box::new(g.clone())).unwrap();
+    for k in 0..10 {
+        edit(&mut j, &mut live, k);
+    }
+    let n = j.seq();
+    j.snapshot_in_background(&live).unwrap();
+    for k in 10..15 {
+        edit(&mut j, &mut live, k);
+    }
+    let opener = {
+        let g = g.clone();
+        std::thread::spawn(move || {
+            std::thread::sleep(Duration::from_millis(50));
+            g.set_gate(true);
+        })
+    };
+    let total = j.seq();
+    drop(j);
+    opener.join().unwrap();
+    assert_eq!(g.snapshot_seq_on_disk(), Some(n));
+    let (_, c, r) = Journal::open(Box::new(g.files.clone())).unwrap();
+    assert_eq!(c.to_snapshot(), live.to_snapshot());
+    assert_eq!((r.snapshot_seq, r.stale as u64, r.replayed as u64), (n, n, total - n));
+}
+
+/// On disk, with a real worker thread: a crash mid-write leaves only a temp file, which loading
+/// ignores; edits during the snapshot survive; reopening resumes appends.
+#[test]
+fn fs_store_background_snapshot() {
+    let dir = std::env::temp_dir().join(format!("lc-journal-bg-{}", std::process::id()));
+    let _ = std::fs::remove_dir_all(&dir);
+    let (mut j, mut live, _) = Journal::open(Box::new(FsStore::open(&dir).unwrap())).unwrap();
+    for k in 0..200 {
+        edit(&mut j, &mut live, k);
+    }
+    // a previous crash mid-snapshot left a partial temp file
+    std::fs::write(dir.join(format!("{SNAPSHOT}.tmp")), b"{\"format\":\"lightcraft-catal").unwrap();
+    let (mut j2, c2, r) = Journal::open(Box::new(FsStore::open(&dir).unwrap())).unwrap();
+    assert_eq!((c2.to_snapshot(), r.snapshot_seq), (live.to_snapshot(), 0));
+    drop(j);
+    let n = j2.seq();
+    j2.snapshot_in_background(&live).unwrap();
+    for k in 200..260 {
+        edit(&mut j2, &mut live, k);
+    }
+    j2.wait().unwrap();
+    assert_eq!((j2.snapshot_seq(), j2.log_records()), (n, j2.seq() - n));
+    assert!(!dir.join(format!("{SNAPSHOT}.tmp")).exists());
+    edit(&mut j2, &mut live, 999);
+    drop(j2);
+    let (_, c3, r) = Journal::open(Box::new(FsStore::open(&dir).unwrap())).unwrap();
+    assert_eq!(c3.to_snapshot(), live.to_snapshot());
+    assert_eq!((r.snapshot_seq, r.stale), (n, 0));
+    let _ = std::fs::remove_dir_all(&dir);
+}
+
+proptest! {
+    #![proptest_config(ProptestConfig::with_cases(48))]
+
+    /// Any interleaving of edits, background snapshots (started, landed, finished), synchronous
+    /// snapshots and crashes: a crash at any point recovers exactly the acknowledged ops.
+    #[test]
+    fn background_compaction_never_loses_acknowledged_ops(actions in proptest::collection::vec((0u8..8, 0u64..1000), 1..60)) {
+        let g = Gated::new();
+        let (mut j, mut live, _) = Journal::open(Box::new(g.clone())).unwrap();
+        for (ctl, k) in actions {
+            match ctl {
+                0..=2 => edit(&mut j, &mut live, k),
+                3 => j.snapshot_in_background(&live).unwrap(),
+                4 | 5 => {
+                    // the worker writes its snapshot; finish it now or later
+                    if j.snapshot_running() {
+                        g.set_gate(true);
+                        wait_written(&j);
+                        g.set_gate(false);
+                        if ctl == 4 {
+                            prop_assert!(j.poll().unwrap());
+                        }
+                    }
+                }
+                6 => {
+                    g.set_gate(true);
+                    j.snapshot(&live).unwrap();
+                    g.set_gate(false);
+                    prop_assert_eq!(g.snapshot_seq_on_disk(), Some(j.seq()));
+                }
+                _ => {
+                    // a crash now
+                    let (_, c, r) = Journal::open(Box::new(g.crash())).unwrap();
+                    prop_assert_eq!(c.to_snapshot(), live.to_snapshot());
+                    prop_assert_eq!(r.failed, 0);
+                    prop_assert!(r.damaged.is_none());
+                }
+            }
+            // the snapshot on disk never covers more than what was appended
+            prop_assert!(g.snapshot_seq_on_disk().unwrap_or(0) <= j.seq());
+        }
+        g.set_gate(true);
+        let (_, c, _) = Journal::open(Box::new(g.crash())).unwrap();
+        prop_assert_eq!(c.to_snapshot(), live.to_snapshot());
+        drop(j);
+        let (_, c, _) = Journal::open(Box::new(g.files.clone())).unwrap();
+        prop_assert_eq!(c.to_snapshot(), live.to_snapshot());
+    }
+}

crates/engine/src/library.rs

@@ -15,7 +15,8 @@
 //!
 //! Every top-level [`Session::execute`] persists the ops it produced (fsynced) before returning,
 //! so a crash loses at most the command in flight. The log is compacted into a snapshot when it
-//! grows (see [`lightcraft_catalog::SnapshotPolicy`]) and on [`Session::close_library`].
+//! grows (see [`lightcraft_catalog::SnapshotPolicy`]; written by a worker thread on native, see
+//! [`lightcraft_catalog::journal`]) and on [`Session::close_library`].
 
 use std::path::{Path, PathBuf};
 
@@ -91,6 +92,10 @@ impl Library {
     pub fn journal(&self) -> &Journal {
         &self.journal
     }
+    #[cfg(test)]
+    pub(crate) fn journal_mut(&mut self) -> &mut Journal {
+        &mut self.journal
+    }
     pub fn thumbs_dir(&self) -> PathBuf {
         self.dir.join("thumbs")
     }
@@ -141,6 +146,15 @@ fn presets_json(s: &Session) -> String {
     serde_json::to_string_pretty(&f).unwrap_or_default()
 }
 
+/// Finish a background compaction that is done (cheap otherwise). A failed one lost nothing (the
+/// log is kept whole) and is retried later; it is reported like other persistence errors.
+fn poll_compaction(lib: &mut Library) {
+    if let Err(e) = lib.journal.poll() {
+        log::error!("library: compaction: {e}");
+        lib.last_error = Some(format!("compaction: {e}"));
+    }
+}
+
 fn read_json<T: serde::de::DeserializeOwned>(store: &mut dyn Store, name: &str) -> Option<T> {
     store.read(name).ok().flatten().and_then(|b| serde_json::from_slice(&b).ok())
 }
@@ -160,6 +174,12 @@ impl Session {
     pub fn open_library_in(&mut self, stores: LibraryStores, seed_demo: bool) -> Result<&LoadReport> {
         let LibraryStores { dir, catalog, mut files, on_disk } = stores;
         self.media.smart_dir = on_disk.then(|| crate::smart::dir(&dir));
+        // the current library may be these same files: let its background snapshot land first
+        if let Some(old) = self.library.as_mut()
+            && let Err(e) = old.journal.wait()
+        {
+            log::error!("library: {e}");
+        }
         let (mut journal, catalog, report) = Journal::open(catalog)?;
         self.catalog = catalog;
         self.undo.clear();
@@ -245,9 +265,11 @@ impl Session {
             self.pending_log.clear();
             lib.last_error = None;
         }
+        poll_compaction(lib);
         // Never snapshot mid-interaction: the catalog then holds an uncommitted preview value.
+        // The snapshot is written by a worker thread (natively); appends go on meanwhile.
         if lib.journal.wants_snapshot() && self.interaction.is_none() {
-            lib.journal.snapshot(&self.catalog)?;
+            lib.journal.snapshot_in_background(&self.catalog)?;
         }
         let presets = presets_json(self);
         let Some(lib) = self.library.as_mut() else { return Ok(()) };
@@ -266,6 +288,8 @@ impl Session {
     pub fn persist_if_dirty(&mut self) {
         if self.library.is_some() && !self.pending_log.is_empty() {
             let _ = self.persist();
+        } else if let Some(lib) = self.library.as_mut() {
+            poll_compaction(lib);
         }
     }
 

crates/engine/src/tests_library.rs

@@ -52,6 +52,57 @@ fn edits_survive_restart_without_close() {
     let _ = std::fs::remove_dir_all(&dir);
 }
 
+/// Compaction at the threshold runs on a worker thread (issue #37): commands go on (and stay
+/// durable) meanwhile, the frame loop's `persist_if_dirty` finishes it, and a crash at any time
+/// keeps every command.
+#[test]
+fn threshold_compaction_runs_in_the_background() {
+    let dir = temp_dir("bg-compact");
+    let mut s = open(&dir, true);
+    s.library.as_mut().unwrap().journal_mut().policy = lightcraft_catalog::SnapshotPolicy { max_records: 8, max_bytes: u64::MAX };
+    let ids: Vec<_> = s.catalog.photos().map(|p| p.id).collect();
+    let mut background = 0;
+    for k in 0..60usize {
+        s.selection = crate::Selection::single(ids[k % ids.len()]);
+        s.execute("photo.rate", &json!({"rating": k % 6})).unwrap();
+        if s.execute("library.info", &json!({})).unwrap()["persistence"]["snapshotRunning"].as_bool().unwrap() {
+            background += 1;
+        }
+        if k == 30 {
+            // a crash while a compaction may be in flight
+            let expect = s.catalog.to_snapshot();
+            let copy = temp_dir("bg-compact-crash");
+            std::fs::create_dir_all(&copy).unwrap();
+            for f in ["catalog.snap", "catalog.log"] {
+                if dir.join(f).exists() {
+                    std::fs::copy(dir.join(f), copy.join(f)).unwrap();
+                }
+            }
+            let s2 = open(&copy, true);
+            assert_eq!(s2.catalog.to_snapshot(), expect);
+            let _ = std::fs::remove_dir_all(&copy);
+        }
+    }
+    assert!(background > 0, "compactions ran in the background");
+    for _ in 0..10_000 {
+        s.persist_if_dirty();
+        if !s.library.as_ref().unwrap().journal().snapshot_running() {
+            break;
+        }
+        std::thread::sleep(std::time::Duration::from_millis(1));
+    }
+    let info = s.execute("library.info", &json!({})).unwrap();
+    let p = &info["persistence"];
+    assert!(p["snapshots"].as_u64() >= Some(2) && p["lastSnapshot"]["background"] == true, "{p}");
+    // the log holds exactly the records after the snapshot
+    assert_eq!(info["logRecords"].as_u64().unwrap(), info["seq"].as_u64().unwrap() - info["snapshotSeq"].as_u64().unwrap(), "{info}");
+    let expect = s.catalog.to_snapshot();
+    drop(s); // no close: like a crash
+    let s2 = open(&dir, true);
+    assert_eq!(s2.catalog.to_snapshot(), expect);
+    let _ = std::fs::remove_dir_all(&dir);
+}
+
 #[test]
 fn close_writes_snapshot_and_view_state() {
     let dir = temp_dir("close");

docs/control-protocol.md

@@ -21,7 +21,7 @@ mode ([mcp.md](mcp.md)) is a thin layer over this channel. Implementation:
 | `engine.commands` | — | Engine + UI commands: id, label, menu, shortcut, params doc, enabled |
 | `ui.menu.list` | — | Menu entries (flat: id, label, menu path, shortcut, enabled) |
 | `ui.menu.tree` | — | The menu bar as shown (File … Help): items `{id, params?, label, shortcut?, enabled, checked?}`, separators, submenus — the model behind the native macOS menu bar and the in-window menus |
-| `ui.inspect` | — | UI state, window, canvas/image rects, `scroll: {grid, filmstrip}` (scroll offsets in points, `null` until drawn), active photo, selection, perf, status, memory (bytes per cache, see `library.memory`; plus stage caches and textures), `export: {running: {total, done, current} \| null, last}` |
+| `ui.inspect` | — | UI state, window, canvas/image rects, `scroll: {grid, filmstrip}` (scroll offsets in points, `null` until drawn), active photo, selection, perf (`frameMs` = layout, `logicMs` = per-frame logic before it, `updateMs` = both, `maxUpdateMs`, `fps`, render queue …), status, memory (bytes per cache, see `library.memory`; plus stage caches and textures), `export: {running: {total, done, current} \| null, last}` |
 | `ui.widgets` | `{filter?}` | On-screen widgets `{id, rect: [x, y, w, h]}` (screen points) |
 | `ui.clickWidget` / `ui.dragWidget` / `ui.hoverWidget` | `{id, count?, fx?, fy?}` / `{id, toX?, toY?, dx?, dy?, steps?}` / `{id, fx?, fy?}` | Real egui input on a widget (hover: the pointer rests on it, e.g. for preset/profile previews) |
 | `ui.move` / `ui.click` / `ui.drag` | `{x, y, count?, button?}` / `{x, y, toX, toY, steps?}` | Raw pointer input, screen points |