Let a catalog writer wait for its turn instead of losing its work

SQLite's busy timeout defaults to zero, and nothing ever set one: the
loser of a write race got SQLITE_BUSY at the moment it asked. WAL does
not cover this — it makes one writer and many readers free, and this
application constantly has two writers, the face sweep committing a
batch while the derived sync imports shards or reclustering reads.

The cost was not a retry but lost work. A sweep that had already paid
for the detection and the embedding — seconds per image, the expensive
part — discarded the result on "storing faces for 214: database is
locked" and moved on to the next image. Both the desktop and the tablet
logged runs of those on consecutive images, which is a face sweep
quietly failing to store the faces it had just computed.

Ten seconds, on every connection, set in configure() so that nothing
can open the catalog without it — the figure the job runner's own tests
have used for this reason since they were written. It is far longer
than any transaction here, so it bounds pathology rather than making
anyone wait.

core/dr-catalog/src/schema.rs

@@ -208,11 +208,38 @@ pub fn backfill(conn: &Connection) -> Result<Vec<(&'static str, usize)>, Catalog
     Ok(out)
 }
 
+/// How long a connection waits for a writer to finish before giving up.
+///
+/// TRACES: NFR-R1
+/// SQLite's default is **zero** — the loser of a race gets `SQLITE_BUSY` at
+/// once rather than a turn — and WAL does not change that for two writers. One
+/// writer and many readers is the case WAL makes free; this is the other one,
+/// and this application has it constantly: the face sweep commits a batch while
+/// reclustering reads, the derived sync imports shards while the sweep writes.
+///
+/// Without a timeout that contention was *lost work*, not a retry. A face
+/// sweep that had already paid for the detection and the embedding — the
+/// expensive part, seconds per image — threw the result away on
+/// `storing faces for 214: database is locked` and moved on, and both the
+/// desktop and the tablet logged runs of those on consecutive images.
+///
+/// Ten seconds, matching the figure the job runner's tests already use for the
+/// same reason. It is far longer than any transaction here (a sweep batch is
+/// sub-second; the slowest is a WAL checkpoint of a 130 MB catalog), so in
+/// practice it is a bound on pathology rather than a wait anyone sits through.
+/// The tension with NFR-P9 is real but one-sided: a query on the UI thread
+/// would rather wait for its turn than fail, because the failure is what the
+/// user sees as "cannot open catalog".
+const BUSY_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
+
 /// Connection setup applied on every open, migration or not.
 ///
 /// WAL is required by NFR-R1: it survives power loss without corruption, and
 /// it lets a background job write while the grid reads.
 pub fn configure(conn: &Connection) -> Result<(), CatalogError> {
+    // Before the pragmas, so that a connection racing a migration waits for it
+    // rather than failing on the first statement it tries.
+    conn.busy_timeout(BUSY_TIMEOUT)?;
     conn.pragma_update(None, "journal_mode", "WAL")?;
     // NORMAL rather than FULL: with WAL this is durable across process death
     // (which is what FR-PLAT-AND-3 cares about) and only risks the last
@@ -1148,6 +1175,60 @@ CREATE INDEX jobs_ready ON jobs(state, priority DESC, not_before);
 
 #[cfg(test)]
 mod tests {
+
+    #[test]
+    fn a_writer_waits_for_its_turn_rather_than_losing_its_work() {
+        // The failure this exists for: a face sweep that had already paid for
+        // the detection and the embedding threw the result away on
+        // "database is locked" and moved on. WAL does not help here — it makes
+        // one writer and many readers free, and this is two writers.
+        let dir = std::env::temp_dir().join(format!(
+            "dr-busy-{}-{:?}",
+            std::process::id(),
+            std::thread::current().id()
+        ));
+        let _ = std::fs::remove_dir_all(&dir);
+        std::fs::create_dir_all(&dir).unwrap();
+        let path = dir.join("catalog.sqlite");
+
+        let held = rusqlite::Connection::open(&path).unwrap();
+        configure(&held).unwrap();
+        migrate(&held).unwrap();
+
+        let other = rusqlite::Connection::open(&path).unwrap();
+        configure(&other).unwrap();
+
+        // Every connection carries the timeout, which is what makes the wait
+        // below a wait rather than an immediate error.
+        let timeout: i64 = other
+            .query_row("PRAGMA busy_timeout", [], |r| r.get(0))
+            .unwrap();
+        assert_eq!(timeout, BUSY_TIMEOUT.as_millis() as i64);
+
+        // A writer holds the database; the other one must still get its turn
+        // once the first commits, rather than failing at the moment it asks.
+        let writing = held.unchecked_transaction().unwrap();
+        held.execute(
+            "INSERT INTO roots(id, kind, label) VALUES (1, 'local', 'lib')",
+            [],
+        )
+        .unwrap();
+
+        let handle = std::thread::spawn(move || {
+            other.execute(
+                "INSERT INTO roots(id, kind, label) VALUES (2, 'local', 'two')",
+                [],
+            )
+        });
+        std::thread::sleep(std::time::Duration::from_millis(150));
+        writing.commit().unwrap();
+
+        assert!(
+            handle.join().unwrap().is_ok(),
+            "the second writer waited and then wrote, rather than erroring"
+        );
+        let _ = std::fs::remove_dir_all(&dir);
+    }
     use super::*;
 
     fn mem() -> Connection {