aleleba/lightcraft
1
0
Fork 0
mirror of https://github.com/storytold/lightcraft.git synced 2026-11-13 09:20:07 -06:00
Code Issues Packages Projects Releases Wiki Activity

PERF-STEP-STALL: stepping and flagging never wait for the drive, and no change is ever dropped (#3557)

Browse Source 
* PERF-STEP-STALL: stepping and flagging never wait for the drive; the catalog log is written and synced by its own thread, and nothing handed to it is ever dropped

On Windows a write to the catalog log waited for a sync of the same file in
flight, and each flag, rating or step synced on the UI thread, so culling on
a busy drive froze the window. FsStore now hands log appends to a thread
that writes and syncs them in order, batching what arrives during a sync.

No change is lost: the thread keeps every byte it has not yet synced. A
write or sync that fails is held and retried (cutting any fragment first),
later appends are refused with the reason (the engine keeps them queued, as
for any failed save), the held records count as unsaved so the quit prompt
warns, and a snapshot is made due. Cuts, replaces and reads act on the
logical log (synced bytes plus held ones), so the log never has a gap.
A clean quit's closing snapshot holds everything; dropping the store waits
for the thread.

Tests drive a slow drive (a sync hook that sleeps) and a failing one
through test-only fault hooks (doc-hidden, keyed by library folder): with 300 ms syncs the slowest flag is
0.2 ms (305 ms with the log written on the caller, as before); every change
made during slow saves reopens after a flush, a drop and a close; a failed
write is reported, retried and lands with no gap.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* PERF-STEP-STALL: review fixes: a reopen reads what the log thread was handed, the quit check reads the held status now, a covering snapshot clears it, a cut to the synced length reaches the file

- open_stores flushes the old journal's log thread before the same
  library's files are read again (Settings, Libraries; a reopen).
- Session::refresh_unsaved, called by the quit check, reads what the log
  thread holds at that moment, not as of the last frame.
- Journal::unwritten is None once a snapshot holds every op (the closing
  snapshot after a failed log reset no longer makes the quit prompt say
  changes would be lost); wants_snapshot follows it.
- A cut to exactly the synced length with bytes held cuts the file, so
  bytes a failed sync left there never come back on the next open.
- A caller that waits over a minute for the drive gets a timed-out error,
  not "the writer stopped".

Each fix has a test that fails without it.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* PERF-STEP-STALL: a reopen and the quit check wait for what the log thread was handed

Review of #3557:
- Opening the same library again while its log thread can't write what
  it was handed now fails and keeps the open library (with the held
  change, retried). Before, the new catalog was read from files missing
  it.
- The quit check flushes the log thread first instead of reading only
  the failure it had already reported. A change accepted but not yet
  written, then failed, is saved by the closing snapshot before the
  window may close; if that fails too, the unsaved prompt shows.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* PERF-STEP-STALL: a reopen goes ahead when the snapshot already holds every change

Review of #3557: the reopen was refused whenever the old log's flush
failed. When a completed snapshot covers every op, nothing is missing
from the files, so the open now goes ahead and only logs the error.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: FredApps <153173854+FredApps@users.noreply.github.com>
This commit is contained in:
OneSlip
2026-11-05 05:25:04 +01:00
committed by GitHub
co-authored by Claude Opus 5.5 FredApps
parent d977d3e1a4
commit 2ddb43896b
17 changed files with 1288 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/catalog/src/fault.rs
+70
View File
@@ -0,0 +1,70 @@
//! Fault injection for the catalog log's writer thread (tests only: nothing in the product
//! installs a hook, and with none installed a check is one uncontended lock of an empty list).
//!
//! A test makes the drive under one library directory slow (a hook that sleeps) or failing (a
//! hook that returns an error) and proves that a flag never waits for it and that nothing handed
//! over is lost: `crates/catalog/src/tests_log_thread.rs`, `crates/engine/src/tests/slow_drive.rs`.
use std::io;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex, PoisonError};
/// Where the log thread is about to touch the drive.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Step {
/// Writing bytes to the log (or to the file that replaces it).
Write,
/// Syncing them.
Sync,
}
/// Runs before each [`Step`] in the hooked directory: an error fails that step.
pub type Hook = Arc<dyn Fn(Step) -> io::Result<()> + Send + Sync>;
static HOOKS: Mutex<Vec<(PathBuf, Hook)>> = Mutex::new(Vec::new());
/// Run `hook` before every write and sync of the catalog log in `dir` (replacing an earlier one).
#[doc(hidden)]
pub fn install(dir: &Path, hook: Hook) {
let mut hooks = HOOKS.lock().unwrap_or_else(PoisonError::into_inner);
hooks.retain(|(d, _)| d != dir);
hooks.push((dir.to_path_buf(), hook));
}
/// Remove the hook of `dir`.
#[doc(hidden)]
pub fn remove(dir: &Path) {
HOOKS.lock().unwrap_or_else(PoisonError::into_inner).retain(|(d, _)| d != dir);
}
static ON_CALLER: Mutex<Vec<PathBuf>> = Mutex::new(Vec::new());
/// Write the log in `dir` on the caller's thread, syncing before each append returns, as builds
/// before PERF-STEP-STALL did: a test's "before" (stores opened afterwards; the hooks run there too).
#[doc(hidden)]
pub fn write_on_caller(dir: &Path, on: bool) {
let mut dirs = ON_CALLER.lock().unwrap_or_else(PoisonError::into_inner);
dirs.retain(|d| d != dir);
if on {
dirs.push(dir.to_path_buf());
}
}
pub(crate) fn on_caller(dir: &Path) -> bool {
let dirs = ON_CALLER.lock().unwrap_or_else(PoisonError::into_inner);
!dirs.is_empty() && dirs.iter().any(|d| d == dir)
}
pub(crate) fn check(dir: &Path, step: Step) -> io::Result<()> {
let hook = {
let hooks = HOOKS.lock().unwrap_or_else(PoisonError::into_inner);
if hooks.is_empty() {
return Ok(());
}
hooks.iter().find(|(d, _)| d == dir).map(|(_, h)| h.clone())
};
match hook {
Some(h) => h(step),
None => Ok(()),
}
}
crates/catalog/src/journal.rs
+32 -2
View File
@@ -6,7 +6,10 @@
//! rename).
//! - `catalog.log` — JSON lines, one per op applied after the snapshot:
//! `{"seq":N,"crc":C,"op":{…}}` where `C` is the CRC-32 of the op's JSON text. Appends are
//! fsynced before [`Journal::append`] returns.
//! fsynced before [`Journal::append`] returns, except natively: [`crate::FsStore`] hands them to
//! a thread of its own that writes and syncs them in order right after (so a flag never waits
//! for the drive), holds and retries whatever fails ([`Journal::unwritten`]) and refuses later
//! appends meanwhile, so the log never has a gap and nothing handed over is dropped.
//!
//! Loading = snapshot + replay of the records with `seq > snapshot seq`. Recovery rules:
//! - a **torn final record** (crash mid-append: truncated line or bad CRC at the end) is dropped
@@ -195,6 +198,10 @@ pub struct Journal {
/// A failed append may have left part of its records after `log_bytes`, and cutting them off
/// failed too: the next append cuts the log back to `log_bytes` first.
tail_dirty: bool,
/// An append failed (natively that can be the store's own log thread failing to write records
/// handed over before, which it holds and retries): the next chance writes a snapshot, a fresh
/// file holding everything, in case only the log's file is broken.
snapshot_due: bool,
}
/// A background snapshot in flight.
@@ -383,6 +390,7 @@ impl Journal {
pending: None,
retry_at_records: 0,
tail_dirty: false,
snapshot_due: false,
};
let log = log.unwrap_or_default();
@@ -516,6 +524,7 @@ impl Journal {
self.tail_dirty = false;
}
if let Err(e) = self.store.append(LOG, buf.as_bytes()) {
self.snapshot_due = true;
// Part of the batch may be in the file. Cut it back to the last whole record, so the
// retry (same seqs) starts on a clean line instead of after a fragment that would
// read back as damage.
@@ -678,6 +687,7 @@ impl Journal {
}
fn record_snapshot(&mut self, t: SnapshotTiming) {
self.snapshot_due = false;
self.stats.snapshots += 1;
self.stats.last_snapshot = t;
self.stats.max_snapshot_ms = self.stats.max_snapshot_ms.max(t.total_ms);
@@ -697,6 +707,23 @@ impl Journal {
}
}
/// Wait until every appended record is in the file and synced (natively the store writes the
/// log on a thread of its own, [`crate::FsStore`]): before the files are read or copied from
/// outside. An error says why some aren't yet: they are held and retried, never dropped.
pub fn flush(&mut self) -> Result<()> {
self.store.flush().map_err(io)
}
/// Appended records the store holds because writing them failed (retried), and why; `None`
/// when nothing is held back. Never waits for the drive.
pub fn unwritten(&self) -> Option<(usize, String)> {
// a snapshot that holds every op holds whatever the log's thread still holds
if self.snapshot_seq >= self.seq && self.pending.is_none() {
return None;
}
self.store.unwritten()
}
/// Where persistence time went so far.
pub fn stats(&self) -> PersistStats {
PersistStats { snapshot_running: self.pending.is_some(), ..self.stats }
@@ -706,7 +733,10 @@ impl Journal {
pub fn wants_snapshot(&self) -> bool {
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)
&& (self.snapshot_due
|| self.unwritten().is_some()
|| self.log_records >= self.policy.max_records
|| self.log_bytes >= self.policy.max_bytes)
}
pub fn seq(&self) -> u64 {
crates/catalog/src/lib.rs
+7
View File
@@ -14,12 +14,17 @@
#![deny(clippy::unwrap_used, clippy::expect_used, clippy::panic, clippy::unimplemented, clippy::todo, clippy::unreachable)]
pub mod dates;
#[cfg(not(target_arch = "wasm32"))]
#[doc(hidden)]
pub mod fault;
pub mod folders;
mod index;
pub mod journal;
pub mod keywords;
pub mod local;
pub mod lock;
#[cfg(not(target_arch = "wasm32"))]
mod log_writer;
pub mod model;
pub mod query;
pub mod rules;
@@ -1490,6 +1495,8 @@ mod tests_journal;
mod tests_local;
#[cfg(test)]
mod tests_lock;
#[cfg(all(test, not(target_arch = "wasm32")))]
mod tests_log_thread;
#[cfg(test)]
mod tests_repair;
#[cfg(test)]
crates/catalog/src/log_writer.rs
+377
View File
@@ -0,0 +1,377 @@
//! The thread that writes the catalog log for [`crate::FsStore`] (natively).
//!
//! An append hands its bytes over and returns; the thread writes them in order and makes each batch
//! durable (`sync_data`) before the next. The caller (the UI thread, for the app) never waits for
//! the drive: on Windows a write to a file waits for a sync of that file in flight, so with the log
//! written and synced on the UI thread every flag, rating and step waited for the drive, 30–90 ms
//! on a busy disk in a test and over a second on a photographer's machine while previews were
//! written.
//!
//! **Nothing handed over is ever dropped.** The store's log is the file's first `good` bytes (written
//! and synced) followed by `held`, the bytes handed over and not yet synced. A write or sync that
//! fails keeps `held` and retries it (on every later message, and every [`RETRY`] meanwhile), first
//! cutting the file back to `good` so a fragment of the failed write never stays in it. While a
//! failure stands, [`LogWriter::append`] refuses new bytes with the failure (the journal keeps those
//! records queued, as for any failed append) and [`LogWriter::unwritten`] says how many records are
//! held, so the app counts them as not saved and its quit prompt warns before anything could be
//! lost. Cutting the log back, replacing it and reading it act on that logical content, after
//! everything handed over before; the caller waits for them. Dropping the writer waits for
//! everything handed over and tries a last time.
use std::io::{self, Seek, Write};
use std::path::PathBuf;
use std::sync::mpsc::{Receiver, RecvTimeoutError, Sender, channel};
use std::sync::{Arc, Mutex, PoisonError};
use std::time::Duration;
/// How long a caller waits for the thread at most (a cut, a replace, a read, a flush).
const WAIT: Duration = Duration::from_secs(60);
/// How often held bytes are retried while a failure stands and nothing else arrives.
pub(crate) const RETRY: Duration = Duration::from_millis(500);
enum Msg {
Append(Vec<u8>),
Truncate(u64, Sender<io::Result<()>>),
Replace(Vec<u8>, Sender<io::Result<()>>),
Read(Sender<io::Result<Option<Vec<u8>>>>),
/// Reply once everything before is written and synced (or why it isn't).
Flush(Sender<io::Result<()>>),
}
/// What the caller can see without waiting.
#[derive(Default)]
struct Shared {
/// Why the held bytes couldn't be written (until they are, or the log is cut or replaced).
failed: Option<String>,
/// Records (lines) handed over and not yet written and synced, while `failed` stands.
held_records: usize,
}
pub(crate) struct LogWriter {
pub(crate) name: String,
tx: Option<Sender<Msg>>,
shared: Arc<Mutex<Shared>>,
thread: Option<std::thread::JoinHandle<()>>,
}
fn stopped() -> io::Error {
io::Error::other("the catalog log writer stopped")
}
fn not_written(why: &str) -> io::Error {
io::Error::other(format!("the catalog log could not be written: {why}"))
}
impl LogWriter {
/// A writer for `dir/name` (`None` when no thread can start: the store then writes itself).
pub(crate) fn start(dir: PathBuf, name: &str) -> Option<LogWriter> {
let (tx, rx) = channel::<Msg>();
let shared = Arc::new(Mutex::new(Shared::default()));
let sh = shared.clone();
let mut w = Writer::new(dir, name);
let thread = std::thread::Builder::new().name("catalog-log".into()).spawn(move || w.run(&rx, &sh)).ok()?;
Some(LogWriter { name: name.to_string(), tx: Some(tx), shared, thread: Some(thread) })
}
fn send(&self, m: Msg) -> io::Result<()> {
self.tx.as_ref().ok_or_else(stopped)?.send(m).map_err(|_| stopped())
}
fn ask<T>(&self, make: impl FnOnce(Sender<io::Result<T>>) -> Msg) -> io::Result<T> {
let (tx, rx) = channel();
self.send(make(tx))?;
rx.recv_timeout(WAIT).map_err(|e| match e {
RecvTimeoutError::Timeout => io::Error::new(io::ErrorKind::TimedOut, "the drive took over a minute to write the catalog log"),
RecvTimeoutError::Disconnected => stopped(),
})?
}
/// Hand `data` over (never waits for the drive). Refused while bytes handed over before can't
/// be written: the caller keeps `data` and appends it again later.
pub(crate) fn append(&self, data: &[u8]) -> io::Result<()> {
if let Some(why) = self.shared.lock().unwrap_or_else(PoisonError::into_inner).failed.clone() {
return Err(not_written(&why));
}
self.send(Msg::Append(data.to_vec()))
}
/// Cut the log to `len` bytes, after everything handed over; waits for it.
pub(crate) fn truncate(&self, len: u64) -> io::Result<()> {
self.ask(|r| Msg::Truncate(len, r))
}
/// Replace the log with `data` (atomically, durably), after everything handed over; waits.
pub(crate) fn replace(&self, data: &[u8]) -> io::Result<()> {
self.ask(|r| Msg::Replace(data.to_vec(), r))
}
/// The log as appended so far (written or still held), after everything handed over.
pub(crate) fn read(&self) -> io::Result<Option<Vec<u8>>> {
self.ask(Msg::Read)
}
/// Wait until everything handed over is written and synced; an error says why it isn't (the
/// bytes are still held and retried).
pub(crate) fn flush(&self) -> io::Result<()> {
self.ask(Msg::Flush)
}
/// Records handed over that can't be written yet, and why (`None`: nothing is held back by a
/// failure). Never waits.
pub(crate) fn unwritten(&self) -> Option<(usize, String)> {
let sh = self.shared.lock().unwrap_or_else(PoisonError::into_inner);
sh.failed.clone().map(|why| (sh.held_records.max(1), why))
}
}
impl Drop for LogWriter {
fn drop(&mut self) {
drop(self.tx.take());
if let Some(t) = self.thread.take() {
let _ = t.join();
}
}
}
/// The thread's side: the file and what is held for it.
struct Writer {
dir: PathBuf,
path: PathBuf,
tmp: PathBuf,
file: Option<std::fs::File>,
/// Bytes of the file written whole and synced (`None` until the file was first opened).
good: Option<u64>,
/// Handed over, not yet written and synced.
held: Vec<u8>,
failed: Option<String>,
}
fn lines(b: &[u8]) -> usize {
b.iter().filter(|c| **c == b'\n').count()
}
impl Writer {
fn new(dir: PathBuf, name: &str) -> Writer {
let path = dir.join(name);
let tmp = dir.join(format!("{name}.tmp"));
Writer { dir, path, tmp, file: None, good: None, held: Vec::new(), failed: None }
}
fn hook(&self, step: crate::fault::Step) -> io::Result<()> {
crate::fault::check(&self.dir, step)
}
/// Open the file positioned at `good` (a fragment past it, from a write that failed, is cut).
fn open(&mut self) -> io::Result<u64> {
if let (Some(_), Some(g)) = (&self.file, self.good) {
return Ok(g);
}
// (a link or a device planted at the name meanwhile is refused, as at the first append)
crate::store::check_regular_directory_entry(&self.path)?;
let mut f = std::fs::OpenOptions::new().write(true).create(true).truncate(false).open(&self.path)?;
let len = f.metadata()?.len();
let good = match self.good {
None => len,
Some(g) if len > g => {
f.set_len(g)?;
g
}
Some(g) if len < g => {
// shorter than what was synced: someone else cut it; continue at its end
log::error!("catalog: {} is {len} bytes, {g} were written", self.path.display());
len
}
Some(g) => g,
};
f.seek(io::SeekFrom::Start(good))?;
self.good = Some(good);
self.file = Some(f);
Ok(good)
}
/// Write and sync what is held.
fn write_held(&mut self) -> io::Result<()> {
if self.held.is_empty() {
return Ok(());
}
let r = (|| {
self.hook(crate::fault::Step::Write)?;
self.open()?;
let f = self.file.as_mut().ok_or_else(|| io::Error::other("catalog log not open"))?;
f.write_all(&self.held)?;
self.hook(crate::fault::Step::Sync)?;
self.file.as_ref().ok_or_else(|| io::Error::other("catalog log not open"))?.sync_data()
})();
match r {
Ok(()) => {
let n = self.held.len() as u64;
self.good = Some(self.good.unwrap_or(0).saturating_add(n));
self.held.clear();
Ok(())
}
Err(e) => {
// the handle may be dead (a network share that reconnected, a removed volume): the
// retry opens the file again and cuts what this attempt left past `good`
self.file = None;
Err(e)
}
}
}
/// Cut the logical log (file then held) to `len` bytes.
fn truncate(&mut self, len: u64) -> io::Result<()> {
let good = match self.good {
Some(g) => g,
None => match self.open() {
Ok(g) => g,
Err(e) if e.kind() == io::ErrorKind::NotFound => 0,
Err(e) => return Err(e),
},
};
// (to exactly `good` with bytes held: the file is cut, since a write that failed may have
// left some of them past `good`)
if len > good || (len == good && self.held.is_empty()) {
// within what is held: nothing on the drive changes (what a failed write left past
// `good` is cut before the next write)
let keep = usize::try_from(len - good).unwrap_or(usize::MAX).min(self.held.len());
self.held.truncate(keep);
return Ok(());
}
self.file = None;
let f = match std::fs::OpenOptions::new().write(true).open(&self.path) {
Ok(f) => f,
Err(e) if e.kind() == io::ErrorKind::NotFound => {
self.good = None;
self.held.clear();
return Ok(());
}
Err(e) => return Err(e),
};
f.set_len(len)?;
f.sync_all()?;
self.good = Some(len);
self.held.clear();
Ok(())
}
fn replace(&mut self, data: &[u8]) -> io::Result<()> {
self.file = None;
let written = (|| {
self.hook(crate::fault::Step::Write)?;
let mut f = crate::store::create_staging(&self.tmp)?;
f.write_all(data)?;
self.hook(crate::fault::Step::Sync)?;
f.sync_all()?;
drop(f);
std::fs::rename(&self.tmp, &self.path)
})();
match written {
Ok(()) => {
// what was held is part of what the caller replaced the log with (a snapshot holds
// the records before it, `data` the ones after)
self.good = Some(data.len() as u64);
self.held.clear();
self.failed = None;
Ok(())
}
Err(e) => {
let _ = std::fs::remove_file(&self.tmp);
Err(e)
}
}
}
fn read(&mut self) -> io::Result<Option<Vec<u8>>> {
let mut bytes = match std::fs::read(&self.path) {
Ok(b) => b,
Err(e) if e.kind() == io::ErrorKind::NotFound && self.held.is_empty() => return Ok(None),
Err(e) if e.kind() == io::ErrorKind::NotFound => Vec::new(),
Err(e) => return Err(e),
};
if let Some(g) = self.good {
bytes.truncate(usize::try_from(g).unwrap_or(usize::MAX));
}
bytes.extend_from_slice(&self.held);
Ok(Some(bytes))
}
fn publish(&self, shared: &Mutex<Shared>) {
let mut sh = shared.lock().unwrap_or_else(PoisonError::into_inner);
sh.failed = self.failed.clone().filter(|_| !self.held.is_empty());
sh.held_records = if sh.failed.is_some() { lines(&self.held) } else { 0 };
}
fn run(&mut self, rx: &Receiver<Msg>, shared: &Mutex<Shared>) {
loop {
let first = if self.failed.is_some() && !self.held.is_empty() {
match rx.recv_timeout(RETRY) {
Ok(m) => Some(m),
Err(RecvTimeoutError::Timeout) => None,
Err(RecvTimeoutError::Disconnected) => break,
}
} else {
match rx.recv() {
Ok(m) => Some(m),
Err(_) => break,
}
};
let mut batch: Vec<Msg> = first.into_iter().collect();
while let Ok(m) = rx.try_recv() {
batch.push(m);
}
let mut flushes = Vec::new();
for m in batch {
match m {
Msg::Append(data) => self.held.extend_from_slice(&data),
// (published before the reply: the caller reads what it left at once)
Msg::Truncate(len, reply) => {
let r = self.truncate(len);
self.publish(shared);
let _ = reply.send(r);
}
Msg::Replace(data, reply) => {
let r = self.replace(&data);
self.publish(shared);
let _ = reply.send(r);
}
Msg::Read(reply) => {
let _ = reply.send(self.read());
}
Msg::Flush(reply) => flushes.push(reply),
}
}
self.step();
self.publish(shared);
for r in flushes {
let _ = r.send(match &self.failed {
Some(why) if !self.held.is_empty() => Err(not_written(why)),
_ => Ok(()),
});
}
}
// the store is closing: everything handed over gets one more try
self.step();
self.publish(shared);
if !self.held.is_empty() {
log::error!("catalog: {} record(s) could not be written to {} (the closing snapshot holds them)", lines(&self.held), self.path.display());
}
}
/// Write what is held; a failure is kept (and logged when it is new).
fn step(&mut self) {
match self.write_held() {
Ok(()) => {
if self.failed.take().is_some() {
log::info!("catalog: {} written again", self.path.display());
}
}
Err(e) => {
let why = e.to_string();
if self.failed.as_deref() != Some(why.as_str()) {
log::error!("catalog: {}: {why} ({} record(s) held, retried)", self.path.display(), lines(&self.held));
}
self.failed = Some(why);
}
}
}
}
crates/catalog/src/store.rs
+100 -5
View File
@@ -28,7 +28,9 @@ pub trait Store: Send {
self.write_atomic(name, &buf)?;
Ok(buf.len() as u64)
}
/// Append and make durable (fsync) before returning. On an error, part of `data` may have
/// Append and make durable (fsync): before returning, or, natively, by the store's own log
/// thread right after ([`FsStore`]: a write that fails is returned by a later append, and a
/// read through the store sees every byte handed over). On an error, part of `data` may have
/// been written (a full disk, a dropped network share): the journal cuts the file back to its
/// previous length with [`Store::truncate`] before appending again.
fn append(&mut self, name: &str, data: &[u8]) -> io::Result<()>;
@@ -43,12 +45,23 @@ pub trait Store: Send {
fn background_writer(&self) -> Option<Box<dyn Store>> {
None
}
/// Wait until every append handed over is in the file and synced ([`FsStore`] writes them on a
/// thread of its own; the default has nothing to wait for). An error says why some aren't: they
/// are still held and retried, never dropped.
fn flush(&mut self) -> io::Result<()> {
Ok(())
}
/// Records handed over by [`Store::append`] that can't be written yet (a write or sync on the
/// store's own thread failed: they are held and retried), and why. `None`: nothing held back.
fn unwritten(&self) -> Option<(usize, String)> {
None
}
}
/// Appends and truncates open the name's own file, so anything else at the name is the library's
/// problem: a link planted there redirects the write outside the directory (CWE-59), and a device
/// hangs the open. Refused with an error the open can show, before anything is written through.
fn check_regular_directory_entry(path: &Path) -> io::Result<()> {
pub(crate) fn check_regular_directory_entry(path: &Path) -> io::Result<()> {
match std::fs::symlink_metadata(path) {
Ok(md) if md.is_file() => Ok(()),
Ok(md) => Err(io::Error::other(format!("{} is a {:?}, not a regular file", path.display(), md.file_type()))),
@@ -60,15 +73,24 @@ fn check_regular_directory_entry(path: &Path) -> io::Result<()> {
/// Files in a directory.
pub struct FsStore {
dir: PathBuf,
/// Open append handle (kept between appends; dropped on rewrite/truncate).
/// Open append handle (kept between appends; dropped on rewrite/truncate). Used when no thread
/// writes the log (the browser build, or one that couldn't start).
appender: Option<(String, std::fs::File)>,
/// The thread that writes the log (natively; started by the first append).
#[cfg(not(target_arch = "wasm32"))]
log: Option<crate::log_writer::LogWriter>,
}
impl FsStore {
/// Use `dir` (created if missing).
pub fn open(dir: impl AsRef<Path>) -> io::Result<FsStore> {
std::fs::create_dir_all(dir.as_ref())?;
Ok(FsStore { dir: dir.as_ref().to_path_buf(), appender: None })
Ok(FsStore {
dir: dir.as_ref().to_path_buf(),
appender: None,
#[cfg(not(target_arch = "wasm32"))]
log: None,
})
}
pub fn dir(&self) -> &Path {
&self.dir
@@ -76,6 +98,21 @@ impl FsStore {
fn path(&self, name: &str) -> PathBuf {
self.dir.join(name)
}
/// The log thread, when it writes `name`.
#[cfg(not(target_arch = "wasm32"))]
fn log_for(&self, name: &str) -> Option<&crate::log_writer::LogWriter> {
self.log.as_ref().filter(|w| w.name == name)
}
/// Wait until the log thread has written and synced everything handed to it.
pub fn flush_log(&self) -> io::Result<()> {
#[cfg(not(target_arch = "wasm32"))]
if let Some(w) = &self.log {
return w.flush();
}
Ok(())
}
fn sync_dir(&self) {
// Makes the rename durable on POSIX; opening a directory fails on Windows (where
// `rename` is already journaled by NTFS), so errors are ignored.
@@ -97,6 +134,11 @@ pub(crate) fn create_staging(tmp: &std::path::Path) -> io::Result<std::fs::File>
impl Store for FsStore {
fn read(&mut self, name: &str) -> io::Result<Option<Vec<u8>>> {
#[cfg(not(target_arch = "wasm32"))]
if let Some(w) = self.log_for(name) {
// what was handed over, written or still held
return w.read();
}
match std::fs::read(self.path(name)) {
Ok(b) => Ok(Some(b)),
Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(None),
@@ -106,6 +148,13 @@ impl Store for FsStore {
fn write_atomic(&mut self, name: &str, data: &[u8]) -> io::Result<()> {
use std::io::Write;
#[cfg(not(target_arch = "wasm32"))]
if let Some(w) = self.log_for(name) {
// (after the appends handed over before)
w.replace(data)?;
self.sync_dir();
return Ok(());
}
if self.appender.as_ref().is_some_and(|(n, _)| n == name) {
self.appender = None;
}
@@ -122,6 +171,15 @@ impl Store for FsStore {
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;
#[cfg(not(target_arch = "wasm32"))]
if let Some(w) = self.log_for(name) {
// (the log is small: buffered, and replaced by its thread after the appends before)
let mut buf = Vec::new();
fill(&mut buf)?;
w.replace(&buf)?;
self.sync_dir();
return Ok(buf.len() as u64);
}
if self.appender.as_ref().is_some_and(|(n, _)| n == name) {
self.appender = None;
}
@@ -148,6 +206,23 @@ impl Store for FsStore {
fn append(&mut self, name: &str, data: &[u8]) -> io::Result<()> {
use std::io::Write;
#[cfg(not(target_arch = "wasm32"))]
{
if self.log.is_none() && !crate::fault::on_caller(&self.dir) {
check_regular_directory_entry(&self.path(name))?;
if !self.path(name).exists() {
// created here, so the open's errors are this append's and its entry is durable
std::fs::OpenOptions::new().create(true).append(true).open(self.path(name))?;
self.sync_dir();
}
self.appender = None;
self.log = crate::log_writer::LogWriter::start(self.dir.clone(), name);
}
// (another name than the log's is written here, as without a thread)
if let Some(w) = self.log_for(name) {
return w.append(data);
}
}
if self.appender.as_ref().is_none_or(|(n, _)| n != name) {
check_regular_directory_entry(&self.path(name))?;
let existed = self.path(name).exists();
@@ -158,6 +233,12 @@ impl Store for FsStore {
self.appender = Some((name.to_string(), f));
}
let Some((_, f)) = self.appender.as_mut() else { return Err(io::Error::other("appender not open")) };
#[cfg(not(target_arch = "wasm32"))]
let written = crate::fault::check(&self.dir, crate::fault::Step::Write)
.and_then(|()| f.write_all(data))
.and_then(|()| crate::fault::check(&self.dir, crate::fault::Step::Sync))
.and_then(|()| f.sync_data());
#[cfg(target_arch = "wasm32")]
let written = f.write_all(data).and_then(|()| f.sync_data());
if written.is_err() {
// The handle may be dead (a network share that reconnected, a removed volume): the next
@@ -169,6 +250,10 @@ impl Store for FsStore {
}
fn truncate(&mut self, name: &str, len: u64) -> io::Result<()> {
#[cfg(not(target_arch = "wasm32"))]
if let Some(w) = self.log_for(name) {
return w.truncate(len);
}
if self.appender.as_ref().is_some_and(|(n, _)| n == name) {
self.appender = None;
}
@@ -189,7 +274,16 @@ impl Store for FsStore {
#[cfg(not(target_arch = "wasm32"))]
fn background_writer(&self) -> Option<Box<dyn Store>> {
Some(Box::new(FsStore { dir: self.dir.clone(), appender: None }))
Some(Box::new(FsStore { dir: self.dir.clone(), appender: None, log: None }))
}
fn flush(&mut self) -> io::Result<()> {
self.flush_log()
}
#[cfg(not(target_arch = "wasm32"))]
fn unwritten(&self) -> Option<(usize, String)> {
self.log.as_ref().and_then(|w| w.unwritten())
}
}
@@ -256,6 +350,7 @@ impl Store for MemStore {
}
}
// (its one test plants a symlink, so the module is Unix-only: elsewhere `use super::*` would be unused)
#[cfg(all(test, unix))]
mod tests_staging {
use super::*;
crates/catalog/src/tests_background.rs
+2 -1
View File
@@ -269,9 +269,10 @@ fn fs_store_background_snapshot() {
}
// a previous crash mid-snapshot left a partial temp file
std::fs::write(dir.join(format!("{SNAPSHOT}.tmp")), b"{\"format\":\"lightcraft-catal").unwrap();
// (closed first: its log thread writes what was handed over)
drop(j);
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 {
crates/catalog/src/tests_log_thread.rs
+232
View File
@@ -0,0 +1,232 @@
//! PERF-STEP-STALL: the catalog log's own thread ([`crate::log_writer`]). Appends return at once
//! however slow the drive is; everything handed over is written in order; a write that fails is
//! held, reported and retried, never dropped, and the log never has a gap.
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::time::{Duration, Instant};
use crate::fault::{self, Step};
use crate::store::{FsStore, Store};
const LOG: &str = "catalog.log";
fn temp_dir(tag: &str) -> std::path::PathBuf {
let d = std::env::temp_dir().join(format!("lc-log-thread-{tag}-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&d);
d
}
fn wait_until(what: &str, mut done: impl FnMut() -> bool) {
let t = Instant::now();
while !done() {
assert!(t.elapsed() < Duration::from_secs(20), "{what}");
std::thread::sleep(Duration::from_millis(10));
}
}
/// Appends are written by the thread in order; a read through the store sees every one; a cut
/// and a replace come after the appends before them; dropping the store writes what is left.
#[test]
fn appends_are_written_in_order_and_cuts_and_replaces_come_after_them() {
let dir = temp_dir("order");
let mut store = FsStore::open(&dir).unwrap();
let mut want = Vec::new();
for i in 0..200u32 {
let line = format!("{i}\n");
store.append(LOG, line.as_bytes()).unwrap();
want.extend_from_slice(line.as_bytes());
if i % 50 == 0 {
assert_eq!(store.read(LOG).unwrap().unwrap(), want, "all of it, through the store");
}
}
store.flush().unwrap();
assert_eq!(std::fs::read(dir.join(LOG)).unwrap(), want, "all of it, in the file");
store.truncate(LOG, 4).unwrap();
assert_eq!(std::fs::read(dir.join(LOG)).unwrap(), b"0\n1\n");
// never longer than it is
store.truncate(LOG, 400).unwrap();
assert_eq!(std::fs::read(dir.join(LOG)).unwrap(), b"0\n1\n");
store.append(LOG, b"x\n").unwrap();
store.write_atomic(LOG, b"tail\n").unwrap();
assert_eq!(std::fs::read(dir.join(LOG)).unwrap(), b"tail\n");
store.append(LOG, b"more\n").unwrap();
drop(store);
assert_eq!(std::fs::read(dir.join(LOG)).unwrap(), b"tail\nmore\n");
let _ = std::fs::remove_dir_all(&dir);
}
/// (a) Each sync takes 300 ms: 100 appends still return at once, and a flush then finds all of
/// them in the file (before, each append waited for its own sync: 30 s for these).
#[test]
fn an_append_never_waits_for_a_slow_sync() {
let dir = temp_dir("slow");
let mut store = FsStore::open(&dir).unwrap();
store.append(LOG, b"first\n").unwrap();
let syncs = Arc::new(AtomicUsize::new(0));
let n = syncs.clone();
fault::install(
&dir,
Arc::new(move |step| {
if step == Step::Sync {
n.fetch_add(1, Ordering::SeqCst);
std::thread::sleep(Duration::from_millis(300));
}
Ok(())
}),
);
let mut want = b"first\n".to_vec();
let mut slowest = Duration::ZERO;
for i in 0..100u32 {
let line = format!("{i}\n");
let t = Instant::now();
store.append(LOG, line.as_bytes()).unwrap();
slowest = slowest.max(t.elapsed());
want.extend_from_slice(line.as_bytes());
std::thread::sleep(Duration::from_millis(2));
}
assert!(slowest < Duration::from_millis(50), "an append waited {slowest:?}");
store.flush().unwrap();
assert_eq!(std::fs::read(dir.join(LOG)).unwrap(), want);
// batched: far fewer syncs than appends
assert!(syncs.load(Ordering::SeqCst) < 50, "{} syncs", syncs.load(Ordering::SeqCst));
fault::remove(&dir);
let _ = std::fs::remove_dir_all(&dir);
}
/// (c) Writes fail for a while: what was handed over is held (reported, with the reason), later
/// appends are refused so the caller keeps them, and once writes work again the held bytes land,
/// in order, with no fragment of the failed attempts and no gap.
#[test]
fn a_failed_write_is_held_reported_and_retried() {
let dir = temp_dir("fail");
let mut store = FsStore::open(&dir).unwrap();
store.append(LOG, b"0\n").unwrap();
store.flush().unwrap();
let fail = Arc::new(AtomicBool::new(true));
let f = fail.clone();
fault::install(&dir, Arc::new(move |_| if f.load(Ordering::SeqCst) { Err(std::io::Error::other("disk full (test)")) } else { Ok(()) }));
store.append(LOG, b"1\n").unwrap();
store.append(LOG, b"2\n").unwrap();
wait_until("the failure is published", || store.unwritten().is_some());
let (held, why) = store.unwritten().unwrap();
assert!(why.contains("disk full"), "{why}");
assert!(held >= 1, "{held}");
let refused = store.append(LOG, b"3\n");
assert!(refused.is_err(), "refused while held");
assert!(store.flush().is_err(), "a flush says it isn't written");
// a read through the store still sees what was handed over
assert_eq!(store.read(LOG).unwrap().unwrap(), b"0\n1\n2\n");
// a cut within what is held keeps it (the journal cuts back to its own length on an error)
store.truncate(LOG, 6).unwrap();
assert_eq!(store.read(LOG).unwrap().unwrap(), b"0\n1\n2\n");
// a fragment of a failed write left in the file is cut before the retry
{
use std::io::Write;
let mut file = std::fs::OpenOptions::new().append(true).open(dir.join(LOG)).unwrap();
file.write_all(b"{\"seq\":").unwrap();
}
fail.store(false, Ordering::SeqCst);
wait_until("retried on its own", || store.unwritten().is_none());
store.append(LOG, b"3\n").unwrap();
store.flush().unwrap();
assert_eq!(std::fs::read(dir.join(LOG)).unwrap(), b"0\n1\n2\n3\n");
fault::remove(&dir);
let _ = std::fs::remove_dir_all(&dir);
}
/// (b, c) Through the journal: ops appended while the drive is slow or failing all reopen, and a
/// journal over a log that failed has no gap (no damage on load).
#[test]
fn the_journal_reopens_every_op_after_a_slow_then_failing_drive() {
use crate::{Catalog, Journal, Op, Photo, PhotoId, Source};
let dir = temp_dir("journal");
let (mut j, mut c, _) = Journal::open(Box::new(FsStore::open(&dir).unwrap())).unwrap();
let fail = Arc::new(AtomicBool::new(false));
let f = fail.clone();
fault::install(
&dir,
Arc::new(move |step| {
if f.load(Ordering::SeqCst) {
return Err(std::io::Error::other("network share dropped (test)"));
}
if step == Step::Sync {
std::thread::sleep(Duration::from_millis(40));
}
Ok(())
}),
);
let mut queued: Vec<Op> = Vec::new();
for id in 1..=60u64 {
if id == 20 {
fail.store(true, Ordering::SeqCst);
}
if id == 40 {
fail.store(false, Ordering::SeqCst);
}
let p = Photo::new(PhotoId(id), Source::Demo { scene: 0 }, &format!("p{id}.jpg"), "JPEG", 4, 3, "2026-11-02");
let op = Op::AddPhoto { photo: Box::new(p) };
c.apply(op.clone()).unwrap();
queued.push(op);
// as the engine does: a failed append keeps its ops queued for the next one
if j.append(&queued).is_ok() {
queued.clear();
}
std::thread::sleep(Duration::from_millis(5));
}
wait_until("everything written", || {
if !queued.is_empty() && j.append(&queued).is_ok() {
queued.clear();
}
queued.is_empty() && j.unwritten().is_none()
});
j.flush().unwrap();
let expect = c.to_snapshot();
drop(j);
fault::remove(&dir);
let (_, again, report): (_, Catalog, _) = Journal::open(Box::new(FsStore::open(&dir).unwrap())).unwrap();
assert!(report.damaged.is_none(), "{report:?}");
assert_eq!(again.to_snapshot(), expect);
assert_eq!(again.len(), 60);
let _ = std::fs::remove_dir_all(&dir);
}
/// A sync that fails leaves the written bytes past the synced length in the file; a cut to exactly
/// that length removes them from the file too, so they never come back on the next open (review
/// of #3557).
#[test]
fn a_cut_to_the_synced_length_removes_bytes_a_failed_sync_left() {
let dir = temp_dir("cut-synced");
let mut store = FsStore::open(&dir).unwrap();
store
.append(
LOG, b"0
",
)
.unwrap();
store.flush().unwrap();
fault::install(&dir, Arc::new(|step| if step == Step::Sync { Err(std::io::Error::other("sync failed (test)")) } else { Ok(()) }));
store
.append(
LOG, b"1
",
)
.unwrap();
wait_until("the failure is published", || store.unwritten().is_some());
assert_eq!(
std::fs::read(dir.join(LOG)).unwrap(),
b"0
1
",
"written, not synced"
);
store.truncate(LOG, 2).unwrap();
fault::remove(&dir);
drop(store);
assert_eq!(
std::fs::read(dir.join(LOG)).unwrap(),
b"0
"
);
let _ = std::fs::remove_dir_all(&dir);
}
crates/catalog/src/tests_torn_append.rs
+3
View File
@@ -88,6 +88,9 @@ fn partial_append_then_retry_loads_cleanly() {
assert!(j.append(&batch).is_err(), "tear at {tear}");
// cut back to the last whole record
assert_eq!(t.files.get(LOG).unwrap().len(), good_len, "tear at {tear}");
// and a snapshot is due (records handed over before may be missing from a log written
// by a thread of its own: note 2 of 2026-10-25)
assert!(j.wants_snapshot(), "tear at {tear}");
j.append(&batch).unwrap();
let more = apply(&mut c, vec![Op::SetLabel { id, label: Some(ColorLabel::ALL[0]) }]);
j.append(&more).unwrap();
crates/engine/src/lib.rs
+2 -1
View File
@@ -614,7 +614,8 @@ impl Session {
}
}
if self.depth == 0 && self.library.is_some() {
// Make the command durable before reporting success. When the command's own records
// Make the command durable before reporting success (natively: handed to the catalog
// log's own thread, which syncs it right after and holds it until it can). When the command's own records
// can't be appended, it fails with `NotSaved`: the change stays applied in memory and
// queued, and the next save (any later command, or the frame loop) retries it. Other
// persistence trouble (a failed compaction, an older queued change still unwritten by
crates/engine/src/library.rs
+64 -12
View File
@@ -13,8 +13,12 @@
//! The files live in [`Store`]s: a directory ([`FsStore`]) natively, or any other implementation
//! via [`Session::open_library_in`] (the browser build keeps them in OPFS or IndexedDB).
//!
//! Every top-level [`Session::execute`] persists the ops it produced (fsynced) before returning,
//! so a crash loses at most the command in flight. When that write fails the command returns
//! Every top-level [`Session::execute`] persists the ops it produced before returning: fsynced
//! there, or natively handed to the catalog log's own thread, which writes and syncs them right
//! after (the UI thread never waits for the drive), so a crash or power cut loses at most the
//! batch in flight (what arrived during one sync). A write that thread
//! can't make is held and retried, counts as unsaved ([`Session::unsaved`]) and makes a snapshot
//! due. When that write fails the command returns
//! [`EngineError::NotSaved`]: its change stays applied in memory and queued, and every later save
//! retries the queue ([`Session::unsaved`] reports it meanwhile), so nothing is lost once the
//! disk is writable again. The log is compacted into a snapshot when it
@@ -134,6 +138,10 @@ pub struct Library {
/// Why the last write of presets.json failed; `None` once it is written. While set, the preset
/// changes exist only in memory and count as unsaved ([`Session::unsaved`], issue #2266).
presets_error: Option<String>,
/// Records the catalog log's own thread holds because writing them failed (it retries them),
/// and why; refreshed by every persist and every frame ([`Session::persist_if_dirty`]). While
/// set they count as unsaved ([`Session::unsaved`]), so quitting asks first.
log_held: Option<(usize, String)>,
/// When the frame loop may retry a failed append ([`Session::persist_if_dirty`] backs off).
retry_at: Option<web_time::Instant>,
/// What forgetting untouched Local records did when the library opened.
@@ -322,6 +330,8 @@ fn poll_compaction(lib: &mut Library) {
log::error!("library: compaction: {e}");
lib.last_error = Some(format!("compaction: {e}"));
}
// (never waits: the log's thread publishes what it holds)
lib.log_held = lib.journal.unwritten();
}
/// How long the frame loop waits before retrying a failed append (commands retry at once).
@@ -501,13 +511,24 @@ impl Session {
/// Open `stores`; the new library takes `lock` once nothing can fail any more.
fn open_stores(&mut self, stores: LibraryStores, seed_demo: bool, lock: &mut Option<LibraryLock>) -> Result<()> {
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}");
// the current library may be these same files: let its background snapshot land first, and
// what its log's thread was handed reach the file before the new journal reads it. When
// that can't be written, the open fails and the current library stays open with what its
// thread holds (retried): read now, the files would miss it (review of #3557).
if let Some(old) = self.library.as_mut() {
if let Err(e) = old.journal.wait() {
log::error!("library: {e}");
}
if let Err(e) = old.journal.flush() {
old.log_held = old.journal.unwritten();
// a snapshot that holds every op leaves nothing for the files to miss
if old.journal.snapshot_seq() < old.journal.seq() {
return Err(EngineError::NotSaved(format!("the open library's changes aren't on disk yet, so it stays open: {e}")));
}
log::error!("library: {e} (every change is in its snapshot)");
}
}
self.media.smart_dir = on_disk.then(|| crate::smart::dir(&dir));
// Photo ids belong to one library; denoise work must not carry into the next.
self.denoise.library_changed();
self.media.denoise.clear();
@@ -631,6 +652,7 @@ impl Session {
last_error: None,
unsaved_error: None,
presets_error: None,
log_held: None,
retry_at: None,
forgot_local: None,
presets_saved,
@@ -661,7 +683,7 @@ impl Session {
Ok(())
}
/// Write pending ops to the log (fsynced), compact when due, and save changed presets.
/// Write pending ops to the log (fsynced, or handed to the log's own thread natively), compact when due, and save changed presets.
/// Called after every top-level command; cheap when nothing changed. Fails (with
/// [`EngineError::NotSaved`]) only when the queued ops couldn't be written; they stay queued.
pub fn persist(&mut self) -> Result<()> {
@@ -726,13 +748,40 @@ impl Session {
pub fn persist_if_dirty(&mut self) {
let backing_off = self.library.as_ref().and_then(|l| l.retry_at).is_some_and(|t| web_time::Instant::now() < t);
let presets_unsaved = self.library.as_ref().is_some_and(|l| l.presets_error.is_some());
if self.library.is_some() && (!self.pending_log.is_empty() || presets_unsaved) && !backing_off {
// records the log's thread can't write: a snapshot (a fresh file holding everything) is due
let log_held = self.library.as_ref().is_some_and(|l| l.log_held.is_some());
if self.library.is_some() && (!self.pending_log.is_empty() || presets_unsaved || log_held) && !backing_off {
let _ = self.persist();
} else if let Some(lib) = self.library.as_mut() {
poll_compaction(lib);
}
}
/// Read again what the catalog log's own thread holds because writing it failed (never waits):
/// before a check of [`Session::unsaved`] that must be current, such as the quit prompt's.
pub fn refresh_unsaved(&mut self) {
if let Some(lib) = self.library.as_mut() {
lib.log_held = lib.journal.unwritten();
}
}
/// Every change applied is in the library's last snapshot (a closing snapshot landed), whatever
/// the catalog log's thread still holds or failed to reset. `false` without a library.
pub fn snapshot_holds_everything(&self) -> bool {
self.pending_log.is_empty() && self.library.as_ref().is_some_and(|l| l.journal.snapshot_seq() >= l.journal.seq())
}
/// Wait until what the catalog log's thread was handed is on disk, then read again what it
/// holds ([`Session::refresh_unsaved`]). Before the quit check: a change the thread accepted
/// but hasn't written yet is in no count of [`Session::unsaved`]. An error (a write failed, or
/// the drive took over a minute) means some changes may not be on disk yet.
pub fn flush_log(&mut self) -> Result<()> {
let Some(lib) = self.library.as_mut() else { return Ok(()) };
let flushed = lib.journal.flush();
lib.log_held = lib.journal.unwritten();
Ok(flushed?)
}
/// Changes applied in memory but not yet written to disk because saving failed: the number of
/// queued ops (plus one for presets.json) and why (`None` when everything is saved, or no library
/// is open). Retried by every command and by [`Session::persist_if_dirty`].
@@ -740,8 +789,9 @@ impl Session {
let lib = self.library.as_ref()?;
let ops = lib.unsaved_error.as_deref().filter(|_| !self.pending_log.is_empty());
let presets = lib.presets_error.as_deref();
let n = ops.map_or(0, |_| self.pending_log.len()) + usize::from(presets.is_some());
ops.or(presets).map(|why| (n, why))
let held = lib.log_held.as_ref().map(|(n, why)| (*n, why.as_str()));
let n = ops.map_or(0, |_| self.pending_log.len()) + held.map_or(0, |(n, _)| n) + usize::from(presets.is_some());
ops.or(held.map(|(_, why)| why)).or(presets).map(|why| (n, why))
}
/// Flush everything and write a snapshot (on quit). Ends an open interaction first.
@@ -760,6 +810,8 @@ impl Session {
let Some(lib) = self.library.as_mut() else { return Ok(()) };
let before = lib.journal.seq();
let snapshot = lib.journal.snapshot_with_unlogged(&self.catalog, unlogged);
// (a snapshot that landed holds what the log's thread held)
lib.log_held = lib.journal.unwritten();
if unlogged > 0 && lib.journal.seq() == before + unlogged {
// the snapshot holds the queued ops: they are saved, never append them again
log::info!("library: {unlogged} queued change(s) saved in the closing snapshot");
crates/engine/src/tests.rs
+2
View File
@@ -49,6 +49,8 @@ mod settings_files;
mod shelved_folder;
mod slide;
mod slideshow_export;
#[cfg(not(target_arch = "wasm32"))]
mod slow_drive;
mod spots;
mod sync;
mod view_follow;
crates/engine/src/tests/library.rs
+5 -1
View File
@@ -69,7 +69,9 @@ fn threshold_compaction_runs_in_the_background() {
background += 1;
}
if k == 30 {
// a crash while a compaction may be in flight
// a crash while a compaction may be in flight (once the log's thread has written what it
// was handed: a process that crashes in LightCraft's code keeps running it)
s.library.as_mut().unwrap().journal_mut().flush().unwrap();
let expect = s.catalog.to_snapshot();
let copy = temp_dir("bg-compact-crash");
std::fs::create_dir_all(&copy).unwrap();
@@ -544,6 +546,8 @@ fn a_library_open_elsewhere_is_refused() {
let dir = temp_dir("locked");
let mut s = open(&dir, true);
s.execute("photo.rate", &json!({"rating": 3})).unwrap();
// (written by the log's own thread)
s.library.as_mut().unwrap().journal_mut().flush().unwrap();
let log = std::fs::read(dir.join("catalog.log")).unwrap();
let mut other = Session::new();
crates/engine/src/tests/slow_drive.rs
+317
View File
@@ -0,0 +1,317 @@
//! PERF-STEP-STALL: culling on a slow or failing drive. A flag or rating never waits for the drive
//! (the catalog log is written and synced by a thread of its own), and nothing made meanwhile is
//! lost: it is on disk once the save lands, after a clean quit, after a quit with no close, and,
//! when the drive fails, held, retried and reported as unsaved until it is written.
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::{Duration, Instant};
use lightcraft_catalog::fault::{self, Step};
use lightcraft_catalog::{Flag, PhotoId};
use serde_json::json;
use crate::Session;
fn temp_dir(tag: &str) -> std::path::PathBuf {
let d = std::env::temp_dir().join(format!("lc-engine-slow-drive-{tag}-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&d);
d
}
fn open(dir: &std::path::Path) -> Session {
let mut s = Session::new();
s.open_library(dir, true).unwrap();
s
}
fn ids(s: &Session) -> Vec<PhotoId> {
let mut v: Vec<PhotoId> = s.catalog.photos().map(|p| p.id).collect();
v.sort();
v
}
/// Every sync of the log in `dir` takes `sync` (a busy disk: Windows makes a write wait for a sync
/// of the same file in flight); `fail` makes every write fail while it is set.
fn slow_drive(dir: &std::path::Path, sync: Duration, fail: Arc<AtomicBool>) {
fault::install(
dir,
Arc::new(move |step| {
if fail.load(Ordering::SeqCst) {
return Err(std::io::Error::other("the drive is gone (test)"));
}
if step == Step::Sync {
std::thread::sleep(sync);
}
Ok(())
}),
);
}
/// The flag or rating each photo got, as `(id, flag, rating)`.
fn culled(s: &Session) -> Vec<(PhotoId, Flag, u8)> {
ids(s).into_iter().filter_map(|id| s.catalog.photo(id).map(|p| (id, p.flag, p.rating))).collect()
}
/// Flag and rate every photo the way a photographer culls: pick / reject in turn, a rating on
/// every other one. Returns the slowest command, in ms.
fn cull(s: &mut Session, rounds: usize) -> f64 {
let all = ids(s);
let mut slowest = 0.0f64;
for k in 0..rounds {
for (i, id) in all.iter().enumerate() {
let flag = if (i + k) % 2 == 0 { "pick" } else { "reject" };
let t = Instant::now();
s.execute("photo.flag", &json!({"flag": flag, "ids": [id.0]})).unwrap();
slowest = slowest.max(t.elapsed().as_secs_f64() * 1000.0);
if i % 2 == 0 {
let t = Instant::now();
s.execute("photo.rate", &json!({"rating": ((i + k) % 5) as u8 + 1, "ids": [id.0]})).unwrap();
slowest = slowest.max(t.elapsed().as_secs_f64() * 1000.0);
}
}
}
slowest
}
/// (a) With every sync of the log taking 300 ms, a flag or rating returns long before one sync.
/// The same culling with the log written on the caller's thread (as before PERF-STEP-STALL) waits
/// for a sync on every key, which is what this test would see without the fix.
#[test]
fn a_flag_never_waits_for_a_slow_drive() {
let sync = Duration::from_millis(300);
let mut seen = Vec::new();
for on_caller in [true, false] {
let dir = temp_dir(if on_caller { "wait-before" } else { "wait" });
fault::write_on_caller(&dir, on_caller);
let mut s = open(&dir);
slow_drive(&dir, sync, Arc::new(AtomicBool::new(false)));
let t = Instant::now();
let slowest = cull(&mut s, 1);
let total = t.elapsed().as_secs_f64() * 1000.0;
let n = s.library.as_ref().map_or(0, |l| l.journal().stats().appends);
eprintln!(
"{} drive syncing in 300 ms: slowest command {slowest:.1} ms, {total:.0} ms for the culling",
if on_caller { "before (log written on the caller):" } else { "after (log on its own thread):" }
);
seen.push((slowest, total, n));
drop(s);
fault::remove(&dir);
fault::write_on_caller(&dir, false);
let _ = std::fs::remove_dir_all(&dir);
}
let ((before, _, _), (after, _, _)) = (seen[0], seen[1]);
assert!(before >= 300.0, "the caller waited for its sync before: {before:.1} ms");
// a debug build of the whole workspace's tests at once leaves room; one sync is 300 ms
assert!(after < 120.0, "a command waited for the drive: {after:.1} ms");
}
/// (b) Everything flagged and rated while slow saves are in flight is on disk once they land
/// (a flush), after a quit with no close (the process ends normally: the log's thread finishes),
/// and after a clean quit (`close_library`).
#[test]
fn every_change_made_during_a_slow_save_is_on_disk() {
for how in ["flush", "drop", "close"] {
let dir = temp_dir(&format!("kept-{how}"));
let mut s = open(&dir);
slow_drive(&dir, Duration::from_millis(150), Arc::new(AtomicBool::new(false)));
cull(&mut s, 3);
let want = culled(&s);
let expect = s.catalog.to_snapshot();
match how {
"flush" => {
s.library.as_mut().unwrap().journal_mut().flush().unwrap();
// a copy of the files taken now (the next crash) holds everything
let copy = temp_dir("kept-copy");
std::fs::create_dir_all(&copy).unwrap();
for f in [lightcraft_catalog::journal::SNAPSHOT, lightcraft_catalog::journal::LOG] {
if dir.join(f).exists() {
std::fs::copy(dir.join(f), copy.join(f)).unwrap();
}
}
let (_, c, _) = lightcraft_catalog::Journal::open(Box::new(lightcraft_catalog::store::FsStore::open(&copy).unwrap())).unwrap();
assert_eq!(c.to_snapshot(), expect, "{how}");
let _ = std::fs::remove_dir_all(&copy);
}
"close" => s.close_library().unwrap(),
_ => {}
}
assert!(s.unsaved().is_none(), "{how}: {:?}", s.unsaved());
drop(s);
fault::remove(&dir);
let again = open(&dir);
assert_eq!(culled(&again), want, "{how}");
assert_eq!(again.catalog.to_snapshot(), expect, "{how}");
drop(again);
let _ = std::fs::remove_dir_all(&dir);
}
}
/// Wait (bounded) until `done` holds, calling the frame loop's save meanwhile.
fn frames_until(s: &mut Session, what: &str, mut done: impl FnMut(&Session) -> bool) {
let t = Instant::now();
while !done(s) {
assert!(t.elapsed() < Duration::from_secs(20), "{what}: {:?}", s.unsaved());
s.persist_if_dirty();
std::thread::sleep(Duration::from_millis(20));
}
}
/// (c) A drive that fails: the flags handed over before the failure are held and retried, never
/// dropped; the failure is reported (`unsaved`, the quit prompt's source) and later flags fail
/// with `NotSaved` and stay queued; once the drive is back, all of it is written, in order, with
/// no gap in the log.
#[test]
fn a_failed_write_is_reported_and_retried_never_dropped() {
let dir = temp_dir("fail");
let mut s = open(&dir);
let fail = Arc::new(AtomicBool::new(false));
slow_drive(&dir, Duration::from_millis(5), fail.clone());
let all = ids(&s);
fail.store(true, Ordering::SeqCst);
// handed over before the thread found out: accepted, then held
s.execute("photo.flag", &json!({"flag": "pick", "ids": [all[0].0]})).unwrap();
// the quit check reads it at once, not as of the last frame (review of #3557)
let t = Instant::now();
while s.library.as_ref().unwrap().journal().unwritten().is_none() {
assert!(t.elapsed() < Duration::from_secs(20), "the log thread never reported the failure");
std::thread::sleep(Duration::from_millis(10));
}
s.refresh_unsaved();
assert!(s.unsaved().is_some(), "refreshed for the quit check");
let (n, why) = s.unsaved().map(|(n, w)| (n, w.to_string())).unwrap();
assert!(n >= 1 && why.contains("the drive is gone"), "{n} {why}");
// later flags: refused with the reason, kept in memory and queued
let r = s.execute("photo.flag", &json!({"flag": "reject", "ids": [all[1].0]}));
assert!(matches!(r, Err(crate::EngineError::NotSaved(_))), "{r:?}");
assert_eq!(s.catalog.photo(all[1]).unwrap().flag, Flag::Reject, "applied in memory");
let _ = s.execute("photo.rate", &json!({"rating": 4, "ids": [all[2].0]}));
assert!(s.unsaved().is_some_and(|(n, _)| n >= 3), "{:?}", s.unsaved());
// the drive is back: the held record and the queued ones are written by the frame loop
fail.store(false, Ordering::SeqCst);
frames_until(&mut s, "written once the drive is back", |s| s.unsaved().is_none() && s.pending_log.is_empty());
let want = culled(&s);
let expect = s.catalog.to_snapshot();
// a crash now: the files alone hold everything, and the log has no gap (no damage on load)
s.library.as_mut().unwrap().journal_mut().flush().unwrap();
let copy = temp_dir("fail-copy");
std::fs::create_dir_all(&copy).unwrap();
for f in [lightcraft_catalog::journal::SNAPSHOT, lightcraft_catalog::journal::LOG] {
if dir.join(f).exists() {
std::fs::copy(dir.join(f), copy.join(f)).unwrap();
}
}
let (_, c, report) = lightcraft_catalog::Journal::open(Box::new(lightcraft_catalog::store::FsStore::open(&copy).unwrap())).unwrap();
assert!(report.damaged.is_none(), "{report:?}");
assert_eq!(c.to_snapshot(), expect);
let _ = std::fs::remove_dir_all(&copy);
fault::remove(&dir);
drop(s);
assert_eq!(culled(&open(&dir)), want);
let _ = std::fs::remove_dir_all(&dir);
}
/// (c) The drive stays broken for the log until the quit: the quit prompt is told (unsaved), and
/// the closing snapshot (a fresh file) saves the held flags, so they are there on the next open.
#[test]
fn a_log_that_stays_broken_is_saved_by_the_closing_snapshot() {
let dir = temp_dir("broken");
let mut s = open(&dir);
let all = ids(&s);
let fail = Arc::new(AtomicBool::new(false));
slow_drive(&dir, Duration::from_millis(5), fail.clone());
fail.store(true, Ordering::SeqCst);
s.execute("photo.flag", &json!({"flag": "reject", "ids": [all[0].0]})).unwrap();
frames_until(&mut s, "the failure is reported", |s| s.unsaved().is_some());
let _ = s.execute("photo.flag", &json!({"flag": "pick", "ids": [all[1].0]}));
let want = culled(&s);
// (the snapshot itself is written by the store, which the hook leaves alone; the log's reset
// after it fails, so close reports it, but the snapshot holds everything)
let _ = s.close_library();
// the snapshot holds them: the quit prompt must not say they would be lost (review of #3557)
assert!(s.unsaved().is_none(), "{:?}", s.unsaved());
drop(s);
fault::remove(&dir);
let again = open(&dir);
assert!(again.library.as_ref().unwrap().report.damaged.is_none());
assert_eq!(culled(&again), want);
drop(again);
let _ = std::fs::remove_dir_all(&dir);
}
/// The same library opened again in the same session (Settings, Libraries; a reopen after an
/// error): everything the old session's log thread was handed is in the files before the new one
/// reads them (review of #3557).
#[test]
fn reopening_the_same_library_reads_what_its_log_thread_was_handed() {
let dir = temp_dir("reopen");
let mut s = open(&dir);
fault::install(
&dir,
Arc::new(|step| {
if step == Step::Write {
std::thread::sleep(Duration::from_millis(300));
}
Ok(())
}),
);
let all = ids(&s);
s.execute("photo.flag", &json!({"flag": "reject", "ids": [all[0].0]})).unwrap();
s.open_library(&dir, true).unwrap();
fault::remove(&dir);
assert_eq!(s.catalog.photo(all[0]).unwrap().flag, Flag::Reject);
drop(s);
let _ = std::fs::remove_dir_all(&dir);
}
/// The same library opened again while its log's thread can't write what it was handed: the open
/// fails and the current library stays open with the change (held, retried), instead of a new
/// catalog read from files that miss it (review of #3557). Once the drive is back, the reopen
/// reads it.
#[test]
fn a_reopen_waits_for_what_the_log_thread_could_not_write() {
let dir = temp_dir("reopen-fail");
let mut s = open(&dir);
let all = ids(&s);
let fail = Arc::new(AtomicBool::new(false));
slow_drive(&dir, Duration::from_millis(5), fail.clone());
fail.store(true, Ordering::SeqCst);
s.execute("photo.flag", &json!({"flag": "reject", "ids": [all[0].0]})).unwrap();
let r = s.open_library(&dir, true);
assert!(matches!(r, Err(crate::EngineError::NotSaved(_))), "{r:?}");
assert_eq!(s.catalog.photo(all[0]).unwrap().flag, Flag::Reject, "the open library keeps the change");
assert!(s.unsaved().is_some(), "and says it isn't saved yet");
fail.store(false, Ordering::SeqCst);
let t = Instant::now();
while s.library.as_mut().unwrap().journal_mut().flush().is_err() {
assert!(t.elapsed() < Duration::from_secs(20), "never written");
std::thread::sleep(Duration::from_millis(50));
}
s.open_library(&dir, true).unwrap();
fault::remove(&dir);
assert_eq!(s.catalog.photo(all[0]).unwrap().flag, Flag::Reject);
drop(s);
let _ = std::fs::remove_dir_all(&dir);
}
/// A reopen while the log still can't be written, but after a snapshot that holds every change
/// (the log's reset after it failed): nothing is missing from the files, so the open goes ahead
/// (review of #3557).
#[test]
fn a_reopen_after_a_snapshot_that_holds_everything_goes_ahead() {
let dir = temp_dir("reopen-snap");
let mut s = open(&dir);
let all = ids(&s);
let fail = Arc::new(AtomicBool::new(false));
slow_drive(&dir, Duration::from_millis(5), fail.clone());
fail.store(true, Ordering::SeqCst);
s.execute("photo.flag", &json!({"flag": "reject", "ids": [all[0].0]})).unwrap();
let _ = s.close_library();
assert!(s.snapshot_holds_everything());
s.open_library(&dir, true).unwrap();
assert_eq!(s.catalog.photo(all[0]).unwrap().flag, Flag::Reject);
fault::remove(&dir);
drop(s);
assert_eq!(open(&dir).catalog.photo(all[0]).unwrap().flag, Flag::Reject);
let _ = std::fs::remove_dir_all(&dir);
}
crates/ui-egui/src/panels/notices.rs
+21 -8
View File
@@ -76,21 +76,34 @@ pub fn may_close(app: &mut LightcraftApp) -> bool {
app.quit_prompt = Some(QuitPrompt::Tasks(tasks_message(names.len(), &names.join(", "))));
return false;
}
if app.session.unsaved().is_none() {
// what the catalog log's thread was handed is written first, and what it holds read now: a
// change it accepted but hasn't written yet must not pass for saved (review of #3557)
let flushed = app.session.flush_log();
if flushed.is_ok() && app.session.unsaved().is_none() {
return true;
}
let _ = app.session.persist();
if app.session.unsaved().is_some()
&& let Err(e) = app.session.close_library()
{
log::error!("quit: library not saved: {e}");
// a snapshot of everything in memory saves what the log couldn't take
let mut closed = Ok(());
if flushed.is_err() || app.session.unsaved().is_some() {
closed = app.session.close_library();
if let Err(e) = &closed {
log::error!("quit: library not saved: {e}");
}
}
match app.session.unsaved() {
None => true,
Some((n, e)) => {
match (app.session.unsaved(), flushed, closed) {
(None, Ok(()), _) | (None, _, Ok(())) => true,
(Some((n, e)), _, _) => {
app.quit_prompt = Some(QuitPrompt::Unsaved(unsaved_message(n, e)));
false
}
// the snapshot landed; only the log's reset after it failed
(None, Err(_), Err(_)) if app.session.snapshot_holds_everything() => true,
// the log's thread didn't finish and the snapshot failed: changes may be lost
(None, Err(_), Err(e)) => {
app.quit_prompt = Some(QuitPrompt::Unsaved(unsaved_message(1, &e.to_string())));
false
}
}
}
crates/ui-egui/src/tests/quit_unsaved.rs
+48
View File
@@ -164,3 +164,51 @@ fn damaged_settings_file_is_shown() {
h.step();
assert!(!has(&h, "button:noticeOk"));
}
/// A change the catalog log's thread accepted but hadn't written yet (a drive that takes a moment,
/// then fails) is not taken for saved: the quit check waits for the thread, finds the failure and
/// saves everything with the closing snapshot before the window may close (review of #3557). It
/// used to read only the failure the thread had already reported, and close with the change in
/// no file.
#[test]
fn quitting_waits_for_what_the_log_thread_has_not_written_yet() {
use lightcraft_catalog::fault::{self, Step};
let dir = std::env::temp_dir().join(format!("lc-ui-quit-log-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
let mut s = lightcraft_engine::Session::new();
s.open_library(&dir, true).unwrap();
let id = s.catalog.photos().map(|p| p.id).min().unwrap();
fault::install(
&dir,
Arc::new(|step| {
if step == Step::Write {
std::thread::sleep(Duration::from_millis(400));
return Err(std::io::Error::other("the drive is gone (test)"));
}
Ok(())
}),
);
let app = LightcraftApp::new(s, Services { png: None, ..Default::default() });
let mut h = Headless::new(app, [1200.0, 800.0], 1.0);
let r = h.request("engine.execute", json!({"command": "photo.flag", "params": {"flag": "reject", "ids": [id.0]}}), T);
assert_eq!(r["ok"], true, "{r}");
let closes = crate::panels::notices::may_close(&mut h.app);
let prompt = h.app.quit_prompt.as_ref().map(|p| p.text().to_string());
// the files alone, as the next start reads them, hold the change
let copy = dir.with_extension("copy");
let _ = std::fs::remove_dir_all(&copy);
std::fs::create_dir_all(&copy).unwrap();
for f in [lightcraft_catalog::journal::SNAPSHOT, lightcraft_catalog::journal::LOG] {
if dir.join(f).exists() {
std::fs::copy(dir.join(f), copy.join(f)).unwrap();
}
}
let (_, c, _) = lightcraft_catalog::Journal::open(Box::new(lightcraft_catalog::store::FsStore::open(&copy).unwrap())).unwrap();
let saved = c.photo(id).map(|p| p.flag);
fault::remove(&dir);
drop(h);
let _ = std::fs::remove_dir_all(&copy);
let _ = std::fs::remove_dir_all(&dir);
assert!(!closes || saved == Some(lightcraft_catalog::Flag::Reject), "closed with the flag in no file: {saved:?}");
assert!(closes, "the closing snapshot saved it, so the window may close: {:?} {saved:?}", prompt);
}
docs/gaps.md
+3 -2
View File
File diff suppressed because one or more lines are too long
docs/parity-checklist.md
+3 -2
View File
File diff suppressed because one or more lines are too long