fix(core): opening the ledger while another program holds it waits instead of failing busy

CHANGELOG.md

@@ -12,6 +12,22 @@ as they were, so if the same path comes back it is tailed, or resynced, exactly
 missing file is excused. A transcript that cannot be read for any other reason (permissions, a
 directory with its name, bytes that are not text) is still recorded.
 
+### Two programs opening the ledger at the same moment no longer fail with "database is locked"
+
+A ledger that is not in WAL mode yet (a first run, or one last written by an old build) is
+switched by whichever program opens it first, and SQLite does not let a second opener wait for
+that: it refuses at once, whatever the busy timeout says, because the two could otherwise wait on
+each other for good. The daemon and a hook, or the app and the daemon, starting together could so
+fail an open that a moment's wait would have let through (seen on Windows in the upgrade tests).
+Every opener (hooks, MCP server, CLI, daemon, app) now tries again for up to two seconds before it
+reports the error.
+
+A migration refused for the same reason used to be skipped and counted as applied, which could
+leave a column or an index missing. A busy ledger is now an error, and the next open runs the
+migration again. The one exception is the step that marks subagent turns: it rewrites rows, not
+the schema, and every open runs it, so an open that finds the ledger busy for it goes on without
+it, as before. Any other migration error (a column that already exists) is still read as applied.
+
 ### The app shows which version it is
 
 The version was only in a startup log line, so someone filing a report could not say which build

crates/lumen-core/Cargo.toml

@@ -8,6 +8,9 @@ rust-version.workspace = true
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 sqlx = { version = "0.9", features = ["runtime-tokio", "sqlite"] }
+# Only `time`, for the pause between tries in `busy`: sqlx's runtime-tokio already pulls the
+# same crate in with it, so this names a dependency the build has and adds none.
+tokio = { version = "1", default-features = false, features = ["time"] }
 # `bundled` compiles SQLite from source via libsqlite3-sys — no system libsqlite3
 # needed on any OS. Feature unifies with sqlx-sqlite (shared libsqlite3-sys 0.37),
 # so the whole workspace builds on macOS, Linux, and Windows without a system lib.

crates/lumen-core/src/busy.rs

@@ -0,0 +1,299 @@
+//! Waiting for a ledger that another connection is using for a moment.
+//!
+//! SQLite answers SQLITE_BUSY at once, without consulting the busy timeout, when a connection
+//! that already holds a read lock asks for the write lock while another connection holds it:
+//! waiting could deadlock the two. Switching a ledger to WAL does exactly that, and every ledger
+//! starts in the rollback-journal mode, so two programs that open one in the same instant (the
+//! daemon and a hook, the app and the daemon) can hear "database is locked" from an open that
+//! a short wait would have let through. The loser has dropped its lock by the time it hears of
+//! the failure, and the winner finishes the switch for everyone.
+//!
+//! Openers retry for [`BUSY_BUDGET`] and then give the error back.
+
+use std::future::Future;
+use std::time::{Duration, Instant};
+
+/// How long an opener keeps trying a busy ledger, counted from its first attempt.
+pub const BUSY_BUDGET: Duration = Duration::from_secs(2);
+
+/// The wait after the first refusal; it doubles up to [`LONGEST_PAUSE`].
+const FIRST_PAUSE: Duration = Duration::from_millis(5);
+const LONGEST_PAUSE: Duration = Duration::from_millis(25);
+
+/// SQLITE_BUSY. Its extended forms (BUSY_RECOVERY 261, BUSY_SNAPSHOT 517, BUSY_TIMEOUT 773)
+/// carry it in the low byte; SQLITE_LOCKED (6), a table held inside one process, is not it.
+const SQLITE_BUSY: i32 = 5;
+
+/// Whether `e` is SQLITE_BUSY in any of its extended forms.
+pub fn is_busy(e: &sqlx::Error) -> bool {
+    let sqlx::Error::Database(db) = e else {
+        return false;
+    };
+    db.code()
+        .and_then(|code| code.parse::<i32>().ok())
+        .is_some_and(|code| code & 0xff == SQLITE_BUSY)
+}
+
+/// Whether `e` is SQLITE_BUSY in any of its extended forms.
+pub fn is_busy_rusqlite(e: &rusqlite::Error) -> bool {
+    matches!(
+        e,
+        rusqlite::Error::SqliteFailure(failure, _)
+            if failure.code == rusqlite::ErrorCode::DatabaseBusy
+    )
+}
+
+/// How long to wait before the next try, or `None` once `budget` is spent.
+fn pause_before_retry(started: Instant, budget: Duration, retries: u32) -> Option<Duration> {
+    let left = budget
+        .checked_sub(started.elapsed())
+        .filter(|left| !left.is_zero())?;
+    Some(
+        (FIRST_PAUSE * (1 << retries.min(3)))
+            .min(LONGEST_PAUSE)
+            .min(left),
+    )
+}
+
+/// Run `op`, again while it fails with SQLITE_BUSY, for up to [`BUSY_BUDGET`].
+///
+/// Any other error is returned as it is, and so is the last busy one when the budget runs out.
+/// An attempt that itself used the budget (it waited out a busy timeout) is not repeated.
+pub async fn retry_busy<T, F, Fut>(op: F) -> Result<T, sqlx::Error>
+where
+    F: FnMut() -> Fut,
+    Fut: Future<Output = Result<T, sqlx::Error>>,
+{
+    retry_busy_for(BUSY_BUDGET, op).await
+}
+
+async fn retry_busy_for<T, F, Fut>(budget: Duration, mut op: F) -> Result<T, sqlx::Error>
+where
+    F: FnMut() -> Fut,
+    Fut: Future<Output = Result<T, sqlx::Error>>,
+{
+    let started = Instant::now();
+    let mut retries = 0;
+    loop {
+        let busy = match op().await {
+            Err(e) if is_busy(&e) => e,
+            done => return done,
+        };
+        let Some(pause) = pause_before_retry(started, budget, retries) else {
+            return Err(busy);
+        };
+        tokio::time::sleep(pause).await;
+        retries += 1;
+    }
+}
+
+/// [`retry_busy`] for the rusqlite openers, which block.
+pub fn retry_busy_blocking<T>(op: impl FnMut() -> rusqlite::Result<T>) -> rusqlite::Result<T> {
+    retry_busy_blocking_for(BUSY_BUDGET, op)
+}
+
+fn retry_busy_blocking_for<T>(
+    budget: Duration,
+    mut op: impl FnMut() -> rusqlite::Result<T>,
+) -> rusqlite::Result<T> {
+    let started = Instant::now();
+    let mut retries = 0;
+    loop {
+        let busy = match op() {
+            Err(e) if is_busy_rusqlite(&e) => e,
+            done => return done,
+        };
+        let Some(pause) = pause_before_retry(started, budget, retries) else {
+            return Err(busy);
+        };
+        std::thread::sleep(pause);
+        retries += 1;
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::borrow::Cow;
+    use std::cell::Cell;
+    use std::fmt;
+    use std::time::Instant;
+
+    /// A database error that carries only a result code, as sqlx's SQLite driver reports one.
+    #[derive(Debug)]
+    struct Coded(i32);
+
+    impl fmt::Display for Coded {
+        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+            write!(f, "(code: {}) coded", self.0)
+        }
+    }
+
+    impl std::error::Error for Coded {}
+
+    impl sqlx::error::DatabaseError for Coded {
+        fn message(&self) -> &str {
+            "coded"
+        }
+        fn code(&self) -> Option<Cow<'_, str>> {
+            Some(self.0.to_string().into())
+        }
+        fn as_error(&self) -> &(dyn std::error::Error + Send + Sync + 'static) {
+            self
+        }
+        fn as_error_mut(&mut self) -> &mut (dyn std::error::Error + Send + Sync + 'static) {
+            self
+        }
+        fn into_error(self: Box<Self>) -> Box<dyn std::error::Error + Send + Sync + 'static> {
+            self
+        }
+        fn kind(&self) -> sqlx::error::ErrorKind {
+            sqlx::error::ErrorKind::Other
+        }
+    }
+
+    fn sqlx_code(code: i32) -> sqlx::Error {
+        sqlx::Error::Database(Box::new(Coded(code)))
+    }
+
+    fn rusqlite_code(code: i32) -> rusqlite::Error {
+        rusqlite::Error::SqliteFailure(rusqlite::ffi::Error::new(code), None)
+    }
+
+    #[test]
+    fn sqlx_busy_is_every_extended_form_of_code_5_and_nothing_else() {
+        // 5 BUSY, 261 BUSY_RECOVERY, 517 BUSY_SNAPSHOT, 773 BUSY_TIMEOUT.
+        for code in [5, 261, 517, 773] {
+            assert!(is_busy(&sqlx_code(code)), "code {code} is SQLITE_BUSY");
+        }
+        // 1 ERROR, 6 LOCKED (a table held inside one process, which waiting does not cure),
+        // 19 CONSTRAINT, 1555 CONSTRAINT_PRIMARYKEY.
+        for code in [1, 6, 19, 1555] {
+            assert!(!is_busy(&sqlx_code(code)), "code {code} is not SQLITE_BUSY");
+        }
+        assert!(!is_busy(&sqlx::Error::PoolTimedOut));
+        assert!(!is_busy(&sqlx::Error::RowNotFound));
+    }
+
+    #[test]
+    fn rusqlite_busy_is_every_extended_form_of_code_5_and_nothing_else() {
+        for code in [5, 261, 517, 773] {
+            assert!(
+                is_busy_rusqlite(&rusqlite_code(code)),
+                "code {code} is SQLITE_BUSY"
+            );
+        }
+        for code in [1, 6, 19, 1555] {
+            assert!(
+                !is_busy_rusqlite(&rusqlite_code(code)),
+                "code {code} is not SQLITE_BUSY"
+            );
+        }
+        assert!(!is_busy_rusqlite(&rusqlite::Error::QueryReturnedNoRows));
+    }
+
+    #[test]
+    fn blocking_retry_runs_again_until_the_call_works() {
+        let calls = Cell::new(0);
+        let out = retry_busy_blocking(|| {
+            calls.set(calls.get() + 1);
+            if calls.get() < 4 {
+                Err(rusqlite_code(5))
+            } else {
+                Ok(calls.get())
+            }
+        });
+        assert_eq!(out.unwrap(), 4);
+        assert_eq!(calls.get(), 4);
+    }
+
+    #[test]
+    fn blocking_retry_does_not_repeat_an_error_that_is_not_busy() {
+        let calls = Cell::new(0);
+        let out: rusqlite::Result<()> = retry_busy_blocking(|| {
+            calls.set(calls.get() + 1);
+            Err(rusqlite_code(1))
+        });
+        assert!(out.is_err());
+        assert_eq!(calls.get(), 1, "a duplicate column is not a reason to wait");
+    }
+
+    #[test]
+    fn blocking_retry_gives_the_busy_error_back_when_the_budget_is_spent() {
+        let calls = Cell::new(0);
+        let started = Instant::now();
+        let out: rusqlite::Result<()> = retry_busy_blocking_for(Duration::from_millis(120), || {
+            calls.set(calls.get() + 1);
+            Err(rusqlite_code(5))
+        });
+        let took = started.elapsed();
+        assert!(is_busy_rusqlite(&out.unwrap_err()));
+        assert!(calls.get() >= 3, "tried {} times in 120 ms", calls.get());
+        assert!(
+            calls.get() <= 30,
+            "tried {} times in 120 ms: the pauses are gone",
+            calls.get()
+        );
+        assert!(took >= Duration::from_millis(120), "gave up after {took:?}");
+        assert!(took < Duration::from_secs(5), "kept going for {took:?}");
+    }
+
+    #[test]
+    fn blocking_retry_does_not_repeat_a_call_that_used_the_budget_by_itself() {
+        // A call that waited the whole busy timeout and still failed was already given its chance.
+        let calls = Cell::new(0);
+        let out: rusqlite::Result<()> = retry_busy_blocking_for(Duration::from_millis(30), || {
+            calls.set(calls.get() + 1);
+            std::thread::sleep(Duration::from_millis(60));
+            Err(rusqlite_code(5))
+        });
+        assert!(is_busy_rusqlite(&out.unwrap_err()));
+        assert_eq!(calls.get(), 1);
+    }
+
+    #[tokio::test]
+    async fn async_retry_runs_again_until_the_call_works() {
+        let calls = Cell::new(0);
+        let out = retry_busy(|| {
+            calls.set(calls.get() + 1);
+            let n = calls.get();
+            async move { if n < 4 { Err(sqlx_code(5)) } else { Ok(n) } }
+        })
+        .await;
+        assert_eq!(out.unwrap(), 4);
+        assert_eq!(calls.get(), 4);
+    }
+
+    #[tokio::test]
+    async fn async_retry_does_not_repeat_an_error_that_is_not_busy() {
+        let calls = Cell::new(0);
+        let out: Result<(), sqlx::Error> = retry_busy(|| {
+            calls.set(calls.get() + 1);
+            async { Err(sqlx_code(1)) }
+        })
+        .await;
+        assert!(out.is_err());
+        assert_eq!(calls.get(), 1);
+    }
+
+    #[tokio::test]
+    async fn async_retry_gives_the_busy_error_back_when_the_budget_is_spent() {
+        let calls = Cell::new(0);
+        let started = Instant::now();
+        let out: Result<(), sqlx::Error> = retry_busy_for(Duration::from_millis(120), || {
+            calls.set(calls.get() + 1);
+            async { Err(sqlx_code(5)) }
+        })
+        .await;
+        let took = started.elapsed();
+        assert!(is_busy(&out.unwrap_err()));
+        assert!(calls.get() >= 3, "tried {} times in 120 ms", calls.get());
+        assert!(
+            calls.get() <= 30,
+            "tried {} times in 120 ms: the pauses are gone",
+            calls.get()
+        );
+        assert!(took >= Duration::from_millis(120), "gave up after {took:?}");
+        assert!(took < Duration::from_secs(5), "kept going for {took:?}");
+    }
+}

crates/lumen-core/src/lib.rs

@@ -1,3 +1,4 @@
+pub mod busy;
 pub mod compress;
 pub mod coverage;
 pub mod econ;

crates/lumen-core/src/meter.rs

@@ -1,4 +1,5 @@
-use crate::schema::{DDL, MIGRATIONS};
+use crate::busy;
+use crate::schema::{BACKFILLS, DDL, MIGRATIONS};
 use rusqlite::{Connection, params};
 
 /// Detect which Claude Code channel spawned this process.
@@ -119,13 +120,24 @@ pub fn resolve_db_path(lumen_db: Option<&str>, home: Option<&str>) -> Option<std
 }
 
 /// Open (or create) the lumen SQLite DB, apply DDL + additive migrations.
+///
+/// A ledger another program holds is waited for (see [`crate::busy`]); the migrations'
+/// other errors are "already applied", as in [`crate::schema::init_schema`], and so is a
+/// backfill that stays busy.
 fn open_db(path: &std::path::Path) -> rusqlite::Result<Connection> {
     restore_sidecar_write_bits(path);
     let conn = Connection::open(path)?;
-    conn.execute_batch(DDL)?;
+    busy::retry_busy_blocking(|| conn.execute_batch(DDL))?;
     for migration in MIGRATIONS {
-        // Ignore "duplicate column" errors — migration already applied.
-        let _ = conn.execute_batch(migration);
+        let outcome = busy::retry_busy_blocking(|| conn.execute_batch(migration));
+        if let Err(e) = outcome
+            && busy::is_busy_rusqlite(&e)
+        {
+            return Err(e);
+        }
+    }
+    for backfill in BACKFILLS {
+        busy::retry_busy_blocking(|| conn.execute_batch(backfill)).ok();
     }
     Ok(conn)
 }