diff --git a/crates/catalog/examples/persist_bench.rs b/crates/catalog/examples/persist_bench.rs index addf3a58d..60ec46aa7 100644 --- a/crates/catalog/examples/persist_bench.rs +++ b/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 -- [photos…] @@ -119,6 +120,39 @@ fn main() -> std::result::Result<(), Box> { } 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 diff --git a/crates/catalog/proptest-regressions/tests_background.txt b/crates/catalog/proptest-regressions/tests_background.txt new file mode 100644 index 000000000..e84ff8c01 --- /dev/null +++ b/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)] diff --git a/crates/catalog/src/journal.rs b/crates/catalog/src/journal.rs index f0a5fcf8b..c146b8b42 100644 --- a/crates/catalog/src/journal.rs +++ b/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-`, 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, + /// 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, + 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>, } 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, ©)); + 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 { + 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}"); + } + } +} diff --git a/crates/catalog/src/lib.rs b/crates/catalog/src/lib.rs index cf7b7a232..041c9c7d0 100644 --- a/crates/catalog/src/lib.rs +++ b/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; diff --git a/crates/catalog/src/store.rs b/crates/catalog/src/store.rs index 56b9eb1aa..069878c71 100644 --- a/crates/catalog/src/store.rs +++ b/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> { + 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> { + Some(Box::new(FsStore { dir: self.dir.clone(), appender: None })) + } } /// Counts the bytes written through it. @@ -165,6 +176,8 @@ impl io::Write for Counter { #[derive(Clone, Default)] pub struct MemStore { pub files: Arc>>>, + /// 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> { + self.background.then(|| Box::new(self.clone()) as Box) + } } diff --git a/crates/catalog/src/tests_background.rs b/crates/catalog/src/tests_background.rs new file mode 100644 index 000000000..0e06936e0 --- /dev/null +++ b/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, Condvar)>, + fail: Arc, +} + +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 { + 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>> { + 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 { + 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> { + 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 = log.lines().map(|l| journal::decode_record(l).unwrap().0).collect(); + assert_eq!(seqs, ((n + 1)..=j.seq()).collect::>()); + 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()); + } +} diff --git a/crates/engine/src/library.rs b/crates/engine/src/library.rs index 6644efa35..24660ea51 100644 --- a/crates/engine/src/library.rs +++ b/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(store: &mut dyn Store, name: &str) -> Option { 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); } } diff --git a/crates/engine/src/tests_library.rs b/crates/engine/src/tests_library.rs index 952c2e1ef..fd09fdc82 100644 --- a/crates/engine/src/tests_library.rs +++ b/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(©).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(©, true); + assert_eq!(s2.catalog.to_snapshot(), expect); + let _ = std::fs::remove_dir_all(©); + } + } + 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"); diff --git a/docs/control-protocol.md b/docs/control-protocol.md index 186adf406..f9e32f2a9 100644 --- a/docs/control-protocol.md +++ b/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 |