From b46478d37accb794d070130f878c5f7331ef3aaa Mon Sep 17 00:00:00 2001 From: HackPoint <6758579+HackPoint@users.noreply.github.com> Date: Mon, 12 Oct 2026 13:50:37 +0300 Subject: [PATCH 1/5] test(core): opening a ledger a writer holds must wait, not fail busy 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. Switching an old ledger to WAL does exactly that, so two programs opening one in the same instant can fail an open that a short wait would have let through (the Windows upgrade-matrix round that fails with "database is locked"). A migration refused that way is also swallowed today and reads as already applied. Red state: busy.rs is a stub that never retries and open_pool is the old connect-then-init sequence. The unit tests and the six ledger tests fail on it. --- crates/lumen-core/src/busy.rs | 235 +++++++++++++++++++++++++ crates/lumen-core/src/lib.rs | 1 + crates/lumen-core/src/schema.rs | 8 + crates/lumen-core/tests/busy_ledger.rs | 232 ++++++++++++++++++++++++ 4 files changed, 476 insertions(+) create mode 100644 crates/lumen-core/src/busy.rs create mode 100644 crates/lumen-core/tests/busy_ledger.rs diff --git a/crates/lumen-core/src/busy.rs b/crates/lumen-core/src/busy.rs new file mode 100644 index 0000000..86a1196 --- /dev/null +++ b/crates/lumen-core/src/busy.rs @@ -0,0 +1,235 @@ +//! 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; + +/// How long an opener keeps trying a busy ledger, counted from its first attempt. +pub const BUSY_BUDGET: Duration = Duration::from_secs(2); + +/// Whether `e` is SQLITE_BUSY in any of its extended forms. +pub fn is_busy(_e: &sqlx::Error) -> bool { + false +} + +/// Whether `e` is SQLITE_BUSY in any of its extended forms. +pub fn is_busy_rusqlite(_e: &rusqlite::Error) -> bool { + false +} + +/// Run `op`, again while it fails with SQLITE_BUSY, for up to [`BUSY_BUDGET`]. +pub async fn retry_busy(op: F) -> Result +where + F: FnMut() -> Fut, + Fut: Future>, +{ + retry_busy_for(BUSY_BUDGET, op).await +} + +async fn retry_busy_for(_budget: Duration, mut op: F) -> Result +where + F: FnMut() -> Fut, + Fut: Future>, +{ + op().await +} + +/// [`retry_busy`] for the rusqlite openers, which block. +pub fn retry_busy_blocking(op: impl FnMut() -> rusqlite::Result) -> rusqlite::Result { + retry_busy_blocking_for(BUSY_BUDGET, op) +} + +fn retry_busy_blocking_for( + _budget: Duration, + mut op: impl FnMut() -> rusqlite::Result, +) -> rusqlite::Result { + op() +} + +#[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> { + 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) -> Box { + 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!(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!(took >= Duration::from_millis(120), "gave up after {took:?}"); + assert!(took < Duration::from_secs(5), "kept going for {took:?}"); + } +} diff --git a/crates/lumen-core/src/lib.rs b/crates/lumen-core/src/lib.rs index 7e1853e..4aef241 100644 --- a/crates/lumen-core/src/lib.rs +++ b/crates/lumen-core/src/lib.rs @@ -1,3 +1,4 @@ +pub mod busy; pub mod compress; pub mod coverage; pub mod econ; diff --git a/crates/lumen-core/src/schema.rs b/crates/lumen-core/src/schema.rs index 8a21ceb..fc2f096 100644 --- a/crates/lumen-core/src/schema.rs +++ b/crates/lumen-core/src/schema.rs @@ -23,6 +23,14 @@ pub async fn init_schema(pool: &sqlx::SqlitePool) -> Result<(), sqlx::Error> { Ok(()) } +/// Open the ledger at `url` as a pool and bring it up to date: what the daemon runs before it +/// serves anything and what the app runs before it reads. +pub async fn open_pool(url: &str) -> Result { + let pool = sqlx::sqlite::SqlitePoolOptions::new().connect(url).await?; + init_schema(&pool).await?; + Ok(pool) +} + /// Additive migrations run after DDL to upgrade existing DBs. /// Each statement is idempotent (errors for "duplicate column" are swallowed by the caller). pub const MIGRATIONS: &[&str] = &[ diff --git a/crates/lumen-core/tests/busy_ledger.rs b/crates/lumen-core/tests/busy_ledger.rs new file mode 100644 index 0000000..eb889f0 --- /dev/null +++ b/crates/lumen-core/tests/busy_ledger.rs @@ -0,0 +1,232 @@ +//! A ledger that another connection holds for a moment is waited for, not failed on. +//! +//! The case SQLite will not wait for by itself (see `lumen_core::busy`): the ledger is still in +//! the rollback-journal mode, a writer holds the write lock, and the opener, which already holds +//! a read lock, asks for the write lock to switch to WAL. It is refused at once, whatever the +//! busy timeout says. + +use std::path::{Path, PathBuf}; +use std::sync::mpsc; +use std::thread::{self, JoinHandle}; +use std::time::{Duration, Instant}; + +use lumen_core::{busy, meter, schema}; +use rusqlite::Connection; +use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions}; + +const STALE_TURN: &str = "INSERT INTO turns (message_id, session_id, ts, input_tokens, \ + output_tokens, cache_read_input_tokens, cache_creation_input_tokens, source_file, \ + is_subagent) VALUES ('m1', 's1', '2026-01-01T00:00:00Z', 1, 1, 0, 0, \ + '/h/.claude/projects/-p/abc/subagents/agent-1.jsonl', 0)"; + +/// Another connection holding a ledger's write lock, taken before [`Writer::hold`] returns. +struct Writer { + release: mpsc::Sender<()>, + thread: JoinHandle<()>, +} + +impl Writer { + /// Holds the lock until [`Writer::release`] or `at_most`, whichever comes first. + fn hold(path: &Path, at_most: Duration) -> Writer { + let (locked_tx, locked_rx) = mpsc::channel(); + let (release, release_rx) = mpsc::channel::<()>(); + let path = path.to_owned(); + let thread = thread::spawn(move || { + let conn = Connection::open(&path).unwrap(); + conn.execute_batch("BEGIN IMMEDIATE").unwrap(); + locked_tx.send(()).unwrap(); + let _ = release_rx.recv_timeout(at_most); + conn.execute_batch("COMMIT").unwrap(); + }); + locked_rx.recv().unwrap(); + Writer { release, thread } + } + + fn release(self) { + let _ = self.release.send(()); + self.thread.join().unwrap(); + } +} + +fn journal_mode(conn: &Connection) -> String { + conn.query_row("PRAGMA journal_mode", [], |r| r.get(0)) + .unwrap() +} + +fn url(path: &Path) -> String { + format!("sqlite:{}?mode=rwc", path.display()) +} + +/// A ledger with the current schema, left in the rollback-journal mode every SQLite file starts +/// in, holding one turn from a subagent that the backfill migration has not marked yet. +fn old_ledger(dir: &Path) -> PathBuf { + let path = dir.join("lumen.db"); + drop(meter::connect_db(&path).unwrap()); + let conn = Connection::open(&path).unwrap(); + let _: String = conn + .query_row("PRAGMA journal_mode=DELETE", [], |r| r.get(0)) + .unwrap(); + conn.execute_batch(STALE_TURN).unwrap(); + assert_eq!(journal_mode(&conn), "delete"); + path +} + +/// A ledger already in WAL (an install that has run before), with the same stale turn. +fn wal_ledger(dir: &Path) -> PathBuf { + let path = dir.join("lumen.db"); + let conn = meter::connect_db(&path).unwrap(); + conn.execute_batch(STALE_TURN).unwrap(); + assert_eq!(journal_mode(&conn), "wal"); + path +} + +fn backfilled(conn: &Connection) -> bool { + let marked: i64 = conn + .query_row("SELECT is_subagent FROM turns WHERE message_id = 'm1'", [], |r| r.get(0)) + .unwrap(); + marked == 1 +} + +async fn backfilled_in(pool: &sqlx::SqlitePool) -> bool { + let marked: i64 = sqlx::query_scalar("SELECT is_subagent FROM turns WHERE message_id = 'm1'") + .fetch_one(pool) + .await + .unwrap(); + marked == 1 +} + +#[test] +fn the_rusqlite_open_waits_for_a_writer_instead_of_failing_busy() { + let dir = tempfile::tempdir().unwrap(); + let path = old_ledger(dir.path()); + let writer = Writer::hold(&path, Duration::from_millis(300)); + + let conn = meter::connect_db(&path) + .expect("a writer that lets go within a moment must not fail the open"); + writer.release(); + + assert_eq!(journal_mode(&conn), "wal"); + assert!(backfilled(&conn), "the open ran the migrations"); +} + +#[test] +fn the_rusqlite_open_gives_up_busy_once_the_budget_is_spent() { + let dir = tempfile::tempdir().unwrap(); + let path = old_ledger(dir.path()); + let writer = Writer::hold(&path, Duration::from_secs(60)); + + let started = Instant::now(); + let err = meter::connect_db(&path).expect_err("a writer that never lets go fails the open"); + let took = started.elapsed(); + writer.release(); + + assert!(busy::is_busy_rusqlite(&err), "{err}"); + assert!(took >= busy::BUSY_BUDGET, "gave up after {took:?}, which is early"); + assert!( + took < busy::BUSY_BUDGET + Duration::from_secs(3), + "kept trying for {took:?}" + ); + + let conn = meter::connect_db(&path).expect("nothing is left half-done"); + assert_eq!(journal_mode(&conn), "wal"); + assert!(backfilled(&conn)); +} + +#[tokio::test] +async fn the_pool_open_waits_for_a_writer_instead_of_failing_busy() { + let dir = tempfile::tempdir().unwrap(); + let path = old_ledger(dir.path()); + let writer = Writer::hold(&path, Duration::from_millis(300)); + + let pool = schema::open_pool(&url(&path)) + .await + .expect("a writer that lets go within a moment must not fail the open"); + writer.release(); + + let mode: String = sqlx::query_scalar("PRAGMA journal_mode") + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(mode, "wal"); + assert!(backfilled_in(&pool).await, "the open ran the migrations"); +} + +#[tokio::test] +async fn the_pool_open_gives_up_busy_once_the_budget_is_spent() { + let dir = tempfile::tempdir().unwrap(); + let path = old_ledger(dir.path()); + let writer = Writer::hold(&path, Duration::from_secs(60)); + + let started = Instant::now(); + let err = schema::open_pool(&url(&path)) + .await + .expect_err("a writer that never lets go fails the open"); + let took = started.elapsed(); + writer.release(); + + assert!(busy::is_busy(&err), "{err}"); + assert!(took >= busy::BUSY_BUDGET, "gave up after {took:?}, which is early"); + assert!( + took < busy::BUSY_BUDGET + Duration::from_secs(3), + "kept trying for {took:?}" + ); + + let pool = schema::open_pool(&url(&path)) + .await + .expect("nothing is left half-done"); + assert!(backfilled_in(&pool).await); +} + +/// A pool that stops waiting for the write lock after 100 ms, so that a migration which needs +/// the lock is refused while a writer holds it, as one would be after the default five seconds. +async fn impatient_pool(path: &Path) -> sqlx::SqlitePool { + SqlitePoolOptions::new() + .connect_with( + SqliteConnectOptions::new() + .filename(path) + .busy_timeout(Duration::from_millis(100)), + ) + .await + .unwrap() +} + +#[tokio::test] +async fn a_migration_the_ledger_was_busy_for_is_applied_not_skipped() { + let dir = tempfile::tempdir().unwrap(); + let path = wal_ledger(dir.path()); + let pool = impatient_pool(&path).await; + let writer = Writer::hold(&path, Duration::from_millis(400)); + + schema::init_schema(&pool).await.expect("init_schema"); + writer.release(); + + assert!( + backfilled_in(&pool).await, + "the backfill was refused while the writer held the ledger and never ran again" + ); +} + +#[tokio::test] +async fn a_ledger_that_stays_busy_is_an_error_not_an_already_applied_migration() { + let dir = tempfile::tempdir().unwrap(); + let path = wal_ledger(dir.path()); + let pool = impatient_pool(&path).await; + let writer = Writer::hold(&path, Duration::from_secs(60)); + + let started = Instant::now(); + let err = schema::init_schema(&pool) + .await + .expect_err("a migration that could not run is not one that already had"); + let took = started.elapsed(); + writer.release(); + + assert!(busy::is_busy(&err), "{err}"); + assert!(took >= busy::BUSY_BUDGET, "gave up after {took:?}, which is early"); + assert!( + took < busy::BUSY_BUDGET + Duration::from_secs(3), + "kept trying for {took:?}" + ); + + schema::init_schema(&pool).await.expect("the next open finishes the job"); + assert!(backfilled_in(&pool).await); +} From cfe7425da3d3a9f3476126d97a55c11ad029de24 Mon Sep 17 00:00:00 2001 From: HackPoint <6758579+HackPoint@users.noreply.github.com> Date: Mon, 12 Oct 2026 14:03:37 +0300 Subject: [PATCH 2/5] fix(core): wait for a busy ledger instead of failing the open, and never read a busy migration as applied SQLite answers SQLITE_BUSY at once, without the busy timeout, when a connection that already holds a read lock asks for the write lock while another holds it. Switching a ledger from the rollback journal to WAL does that, so two programs opening one ledger in the same instant could hear "database is locked" from an open a short wait would have let through. That is the Windows CI failure in the upgrade matrix's concurrent rounds. Every opener now retries a busy open for up to two seconds: the shared pool open (open_pool, used by the daemon and the app), init_schema, and the rusqlite open_db. A migration refused as busy used to be swallowed and left undone; it is now an error, and the next open runs it again. The silent-fallback allowances for those two swallows are gone. Callers get their own tests: the app's connect and the real daemon binary, each started while a writer holds the ledger. --- CHANGELOG.md | 15 ++ crates/lumen-core/Cargo.toml | 3 + crates/lumen-core/src/busy.rs | 94 ++++++++++--- crates/lumen-core/src/meter.rs | 14 +- crates/lumen-core/src/schema.rs | 27 +++- crates/lumen-core/tests/busy_ledger.rs | 25 +++- crates/lumen-core/tests/silent_fallbacks.rs | 15 -- crates/lumen-daemon/src/main.rs | 6 +- crates/lumen-daemon/tests/busy_start.rs | 130 ++++++++++++++++++ .../tests/upgrade_matrix/openers.rs | 7 +- crates/lumen-stats/Cargo.toml | 2 +- crates/lumen-stats/src/lib.rs | 17 +-- crates/lumen-stats/tests/busy_connect.rs | 44 ++++++ 13 files changed, 329 insertions(+), 70 deletions(-) create mode 100644 crates/lumen-daemon/tests/busy_start.rs create mode 100644 crates/lumen-stats/tests/busy_connect.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 1373870..8f4eedb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,21 @@ 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. 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 diff --git a/crates/lumen-core/Cargo.toml b/crates/lumen-core/Cargo.toml index 0169100..e7c7deb 100644 --- a/crates/lumen-core/Cargo.toml +++ b/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. diff --git a/crates/lumen-core/src/busy.rs b/crates/lumen-core/src/busy.rs index 86a1196..0f20cda 100644 --- a/crates/lumen-core/src/busy.rs +++ b/crates/lumen-core/src/busy.rs @@ -11,22 +11,54 @@ //! Openers retry for [`BUSY_BUDGET`] and then give the error back. use std::future::Future; -use std::time::Duration; +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 { - false +pub fn is_busy(e: &sqlx::Error) -> bool { + let sqlx::Error::Database(db) = e else { + return false; + }; + db.code() + .and_then(|code| code.parse::().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 { - false +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 { + 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(op: F) -> Result where F: FnMut() -> Fut, @@ -35,12 +67,24 @@ where retry_busy_for(BUSY_BUDGET, op).await } -async fn retry_busy_for(_budget: Duration, mut op: F) -> Result +async fn retry_busy_for(budget: Duration, mut op: F) -> Result where F: FnMut() -> Fut, Fut: Future>, { - op().await + 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. @@ -49,10 +93,22 @@ pub fn retry_busy_blocking(op: impl FnMut() -> rusqlite::Result) -> rusqli } fn retry_busy_blocking_for( - _budget: Duration, + budget: Duration, mut op: impl FnMut() -> rusqlite::Result, ) -> rusqlite::Result { - op() + 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)] @@ -166,11 +222,10 @@ mod tests { 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 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()); @@ -182,12 +237,11 @@ mod tests { 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)) - }); + 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); } diff --git a/crates/lumen-core/src/meter.rs b/crates/lumen-core/src/meter.rs index 2ea0963..47e6647 100644 --- a/crates/lumen-core/src/meter.rs +++ b/crates/lumen-core/src/meter.rs @@ -1,3 +1,4 @@ +use crate::busy; use crate::schema::{DDL, MIGRATIONS}; use rusqlite::{Connection, params}; @@ -119,13 +120,20 @@ pub fn resolve_db_path(lumen_db: Option<&str>, home: Option<&str>) -> Option rusqlite::Result { 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); + } } Ok(conn) } diff --git a/crates/lumen-core/src/schema.rs b/crates/lumen-core/src/schema.rs index fc2f096..5ea45e1 100644 --- a/crates/lumen-core/src/schema.rs +++ b/crates/lumen-core/src/schema.rs @@ -1,3 +1,5 @@ +use crate::busy; + /// Bring a database up to date: create anything missing, then migrate. /// /// **Every sqlx consumer must call this instead of executing [`DDL`] alone.** DDL @@ -10,23 +12,36 @@ /// /// Migration errors are swallowed, as they are on the rusqlite side: each /// statement is additive and idempotent, so "duplicate column" is the expected -/// outcome on an already-current database. +/// outcome on an already-current database. Any other failure leaves a column +/// missing, and the insert that binds it fails and is filed (`ingest_failed`, +/// `meter_write_failed`). One error is not read that way: a ledger that stayed +/// busy (see [`crate::busy`]) means the statement did not run, and calling that +/// "already applied" would leave a column or an index missing for good. It is +/// waited for, and returned if the wait runs out; the next open runs the +/// migration again. pub async fn init_schema(pool: &sqlx::SqlitePool) -> Result<(), sqlx::Error> { - sqlx::raw_sql(DDL).execute(pool).await?; + busy::retry_busy(|| sqlx::raw_sql(DDL).execute(pool)).await?; for migration in MIGRATIONS { // AssertSqlSafe is sound here: every statement is a literal in // MIGRATIONS above, so no runtime value can reach the SQL. - let _ = sqlx::raw_sql(sqlx::AssertSqlSafe(*migration)) - .execute(pool) - .await; + let outcome = + busy::retry_busy(|| sqlx::raw_sql(sqlx::AssertSqlSafe(*migration)).execute(pool)).await; + if let Err(e) = outcome + && busy::is_busy(&e) + { + return Err(e); + } } Ok(()) } /// Open the ledger at `url` as a pool and bring it up to date: what the daemon runs before it /// serves anything and what the app runs before it reads. +/// +/// A ledger another program holds at that moment is waited for, not failed on: the connection +/// itself can be refused (it switches the file to WAL) as well as the statements after it. pub async fn open_pool(url: &str) -> Result { - let pool = sqlx::sqlite::SqlitePoolOptions::new().connect(url).await?; + let pool = busy::retry_busy(|| sqlx::sqlite::SqlitePoolOptions::new().connect(url)).await?; init_schema(&pool).await?; Ok(pool) } diff --git a/crates/lumen-core/tests/busy_ledger.rs b/crates/lumen-core/tests/busy_ledger.rs index eb889f0..0cdd6dd 100644 --- a/crates/lumen-core/tests/busy_ledger.rs +++ b/crates/lumen-core/tests/busy_ledger.rs @@ -82,7 +82,11 @@ fn wal_ledger(dir: &Path) -> PathBuf { fn backfilled(conn: &Connection) -> bool { let marked: i64 = conn - .query_row("SELECT is_subagent FROM turns WHERE message_id = 'm1'", [], |r| r.get(0)) + .query_row( + "SELECT is_subagent FROM turns WHERE message_id = 'm1'", + [], + |r| r.get(0), + ) .unwrap(); marked == 1 } @@ -121,7 +125,10 @@ fn the_rusqlite_open_gives_up_busy_once_the_budget_is_spent() { writer.release(); assert!(busy::is_busy_rusqlite(&err), "{err}"); - assert!(took >= busy::BUSY_BUDGET, "gave up after {took:?}, which is early"); + assert!( + took >= busy::BUSY_BUDGET, + "gave up after {took:?}, which is early" + ); assert!( took < busy::BUSY_BUDGET + Duration::from_secs(3), "kept trying for {took:?}" @@ -165,7 +172,10 @@ async fn the_pool_open_gives_up_busy_once_the_budget_is_spent() { writer.release(); assert!(busy::is_busy(&err), "{err}"); - assert!(took >= busy::BUSY_BUDGET, "gave up after {took:?}, which is early"); + assert!( + took >= busy::BUSY_BUDGET, + "gave up after {took:?}, which is early" + ); assert!( took < busy::BUSY_BUDGET + Duration::from_secs(3), "kept trying for {took:?}" @@ -221,12 +231,17 @@ async fn a_ledger_that_stays_busy_is_an_error_not_an_already_applied_migration() writer.release(); assert!(busy::is_busy(&err), "{err}"); - assert!(took >= busy::BUSY_BUDGET, "gave up after {took:?}, which is early"); + assert!( + took >= busy::BUSY_BUDGET, + "gave up after {took:?}, which is early" + ); assert!( took < busy::BUSY_BUDGET + Duration::from_secs(3), "kept trying for {took:?}" ); - schema::init_schema(&pool).await.expect("the next open finishes the job"); + schema::init_schema(&pool) + .await + .expect("the next open finishes the job"); assert!(backfilled_in(&pool).await); } diff --git a/crates/lumen-core/tests/silent_fallbacks.rs b/crates/lumen-core/tests/silent_fallbacks.rs index e79a839..70aeba4 100644 --- a/crates/lumen-core/tests/silent_fallbacks.rs +++ b/crates/lumen-core/tests/silent_fallbacks.rs @@ -168,13 +168,6 @@ const ALLOWED: &[Allowed] = &[ "CLAUDE_CODE_ENTRYPOINT", "Unset selects the default channel label.", ), - allow( - "crates/lumen-core/src/meter.rs", - DroppedWrite, - "execute_batch(migration)", - "A duplicate column is the expected outcome. Any other failure leaves a column \ - missing, and the insert that binds it files meter_write_failed.", - ), allow( "crates/lumen-core/src/meter.rs", DroppedWrite, @@ -257,14 +250,6 @@ const ALLOWED: &[Allowed] = &[ "let html =", "The link shown after filing, once the POST has already succeeded.", ), - allow( - "crates/lumen-core/src/schema.rs", - DroppedWrite, - "AssertSqlSafe(*migration)", - "A duplicate column is the expected outcome. Any other failure leaves a column \ - missing, and the insert that binds it fails and is filed (ingest_failed, \ - meter_write_failed): how 1.1.3's missing column would show now.", - ), allow( "crates/lumen-core/src/update.rs", OrDefault, diff --git a/crates/lumen-daemon/src/main.rs b/crates/lumen-daemon/src/main.rs index 90a93d0..d910aec 100644 --- a/crates/lumen-daemon/src/main.rs +++ b/crates/lumen-daemon/src/main.rs @@ -2,7 +2,6 @@ use futures_util::{SinkExt, StreamExt}; use lumen_core::record::Record; use serde::Serialize; use sqlx::SqlitePool; -use sqlx::sqlite::SqlitePoolOptions; use std::collections::HashMap; use std::io::{Read, Seek, SeekFrom}; use std::path::{Path, PathBuf}; @@ -255,13 +254,12 @@ async fn main() -> Result<(), Box> { } lumen_core::meter::restore_sidecar_write_bits(Path::new(&db_path)); let conn = format!("sqlite:{db_path}?mode=rwc"); - let pool = SqlitePoolOptions::new().connect(&conn).await?; logline!("lumen-daemon using db: {db_path}"); - // init_schema, not raw DDL: DDL alone is CREATE TABLE IF NOT EXISTS, so on an + // open_pool runs init_schema, not raw DDL: DDL alone is CREATE TABLE IF NOT EXISTS, so on an // existing database it is a no-op and the is_subagent column added in 1.1.0 // never appeared — making every insert below fail on an unknown column. - lumen_core::schema::init_schema(&pool).await?; + let pool = lumen_core::schema::open_pool(&conn).await?; let (tx, _rx) = broadcast::channel::(1000); diff --git a/crates/lumen-daemon/tests/busy_start.rs b/crates/lumen-daemon/tests/busy_start.rs new file mode 100644 index 0000000..ca50a25 --- /dev/null +++ b/crates/lumen-daemon/tests/busy_start.rs @@ -0,0 +1,130 @@ +//! A daemon started while another connection holds the ledger waits for it instead of exiting. +//! +//! `main` opens the ledger before it serves anything, so a "database is locked" there ended the +//! process, and the live stream the app shows with it. SQLite refuses at once, whatever the busy +//! timeout says, when the connection that switches a rollback-journal ledger to WAL already holds +//! a read lock (see `lumen_core::busy`); only the real binary shows that the open `main` runs is +//! the one that waits. + +use std::fs::File; +use std::path::Path; +use std::process::{Child, Command, Stdio}; +use std::time::{Duration, Instant}; + +use lumen_core::schema; +use sqlx::Connection; +use sqlx::sqlite::{SqliteConnectOptions, SqliteConnection, SqliteJournalMode}; + +/// The daemon prints this once, with the ledger it is about to open. +const MARKER: &str = "lumen-daemon using db: "; + +const STALE_TURN: &str = "INSERT INTO turns (message_id, session_id, ts, input_tokens, \ + output_tokens, cache_read_input_tokens, cache_creation_input_tokens, source_file, \ + is_subagent) VALUES ('m1', 's1', '2026-01-01T00:00:00Z', 1, 1, 0, 0, \ + '/h/.claude/projects/-p/abc/subagents/agent-1.jsonl', 0)"; + +struct Daemon(Child); + +impl Drop for Daemon { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } +} + +/// Reserve an unused port by binding and releasing it; the daemon has to bind it itself. +fn free_port() -> u16 { + let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("reserve a port"); + listener.local_addr().expect("local addr").port() +} + +fn log_says(log: &Path, marker: &str) -> bool { + std::fs::read_to_string(log).is_ok_and(|text| text.contains(marker)) +} + +#[tokio::test] +async fn a_daemon_started_while_a_writer_holds_the_ledger_serves_once_it_lets_go() { + let dir = tempfile::tempdir().unwrap(); + let db = dir.path().join("ledger.db"); + // A forgotten LUMEN_DB resolves to the developer's own ledger, so this is asserted. + assert!( + db.starts_with(std::env::temp_dir()), + "the ledger must live in a temp dir: {db:?}" + ); + let url = format!("sqlite:{}?mode=rwc", db.display()); + + // A ledger with a turn the backfill has not marked yet, back in the rollback-journal mode + // every ledger starts in, with a writer inside it before the daemon starts. + let pool = schema::open_pool(&url).await.unwrap(); + sqlx::query(STALE_TURN).execute(&pool).await.unwrap(); + pool.close().await; + let mut writer = SqliteConnection::connect_with( + &SqliteConnectOptions::new() + .filename(&db) + .journal_mode(SqliteJournalMode::Delete), + ) + .await + .unwrap(); + sqlx::query("BEGIN IMMEDIATE") + .execute(&mut writer) + .await + .unwrap(); + + let projects = dir.path().join("projects"); + std::fs::create_dir_all(&projects).unwrap(); + let log = dir.path().join("daemon.log"); + let addr = format!("127.0.0.1:{}", free_port()); + let mut daemon = Daemon( + Command::new(env!("CARGO_BIN_EXE_lumen-daemon")) + .env("HOME", dir.path()) + .env("LUMEN_PROJECTS_DIR", &projects) + .env("LUMEN_DB", &db) + .env("LUMEN_FAULT_SPOOL", dir.path().join("faults.jsonl")) + .env("LUMEN_WS_ADDR", &addr) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(File::create(&log).unwrap()) + .spawn() + .expect("spawn lumen-daemon"), + ); + + // It names the ledger just before it opens it; give it time to be refused, and to try again. + let deadline = Instant::now() + Duration::from_secs(30); + while !log_says(&log, MARKER) { + assert!( + Instant::now() < deadline, + "the daemon never named its ledger:\n{}", + std::fs::read_to_string(&log).unwrap_or_default() + ); + tokio::time::sleep(Duration::from_millis(20)).await; + } + tokio::time::sleep(Duration::from_millis(400)).await; + sqlx::query("COMMIT").execute(&mut writer).await.unwrap(); + + // It binds only after the ledger is open and current, so an accepting port means it is. + let deadline = Instant::now() + Duration::from_secs(30); + while tokio::net::TcpStream::connect(&addr).await.is_err() { + if let Some(status) = daemon.0.try_wait().expect("poll the daemon") { + panic!( + "the daemon exited with {status} instead of waiting for the writer:\n{}", + std::fs::read_to_string(&log).unwrap_or_default() + ); + } + assert!( + Instant::now() < deadline, + "the daemon never opened ws://{addr}:\n{}", + std::fs::read_to_string(&log).unwrap_or_default() + ); + tokio::time::sleep(Duration::from_millis(100)).await; + } + drop(daemon); + + let mut check = SqliteConnection::connect_with(&SqliteConnectOptions::new().filename(&db)) + .await + .unwrap(); + let marked: i64 = sqlx::query_scalar("SELECT is_subagent FROM turns WHERE message_id = 'm1'") + .fetch_one(&mut check) + .await + .unwrap(); + assert_eq!(marked, 1, "the daemon ran the migrations before it served"); +} diff --git a/crates/lumen-daemon/tests/upgrade_matrix/openers.rs b/crates/lumen-daemon/tests/upgrade_matrix/openers.rs index f91bed3..4143f68 100644 --- a/crates/lumen-daemon/tests/upgrade_matrix/openers.rs +++ b/crates/lumen-daemon/tests/upgrade_matrix/openers.rs @@ -1,7 +1,6 @@ use std::path::Path; use lumen_core::{meter, schema}; -use sqlx::sqlite::SqlitePoolOptions; /// The code paths that open the ledger and bring its schema up to date. #[derive(Clone, Copy, Debug, PartialEq, Eq)] @@ -24,13 +23,11 @@ impl Opener { match self { Opener::Daemon => { meter::restore_sidecar_write_bits(db); - let pool = SqlitePoolOptions::new() - .connect(&url(db)) + let pool = schema::open_pool(&url(db)) .await .map_err(|e| e.to_string())?; - let init = schema::init_schema(&pool).await.map_err(|e| e.to_string()); pool.close().await; - init + Ok(()) } Opener::Gui => { let pool = lumen_stats::connect(&url(db)).await?; diff --git a/crates/lumen-stats/Cargo.toml b/crates/lumen-stats/Cargo.toml index 28ff3ec..b4b9cb9 100644 --- a/crates/lumen-stats/Cargo.toml +++ b/crates/lumen-stats/Cargo.toml @@ -22,4 +22,4 @@ serde_json = "1" # The tests build a real schema with the production DDL rather than a hand-rolled # copy, so a drift between schema.rs and these queries fails the test. tempfile = "3" -tokio = { version = "1", features = ["macros", "rt-multi-thread"] } +tokio = { version = "1", features = ["macros", "rt-multi-thread", "time"] } diff --git a/crates/lumen-stats/src/lib.rs b/crates/lumen-stats/src/lib.rs index e279751..bc246e9 100644 --- a/crates/lumen-stats/src/lib.rs +++ b/crates/lumen-stats/src/lib.rs @@ -12,7 +12,6 @@ use serde::Serialize; use sqlx::SqlitePool; -use sqlx::sqlite::SqlitePoolOptions; /// The three `routed_via` values that represent a Lumen tool call which returned /// fewer tokens than the full file. `builtin_read` is deliberately excluded — it @@ -34,18 +33,14 @@ pub fn db_url() -> String { } /// Open a pool against `url`. +/// +/// Every query below references turns.is_subagent, which only exists after the migrations run. +/// Opening the GUI before the daemon has started would otherwise hit "no such column" on a +/// database created before 1.1.0, so the pool comes back migrated. pub async fn connect(url: &str) -> Result { - let pool = SqlitePoolOptions::new() - .connect(url) - .await - .map_err(|e| e.to_string())?; - // Every query below references turns.is_subagent, which only exists after the - // migrations run. Opening the GUI before the daemon has started would - // otherwise hit "no such column" on a database created before 1.1.0. - lumen_core::schema::init_schema(&pool) + lumen_core::schema::open_pool(url) .await - .map_err(|e| e.to_string())?; - Ok(pool) + .map_err(|e| e.to_string()) } /// Open a pool against the ambient `LUMEN_DB`. diff --git a/crates/lumen-stats/tests/busy_connect.rs b/crates/lumen-stats/tests/busy_connect.rs new file mode 100644 index 0000000..d4827ad --- /dev/null +++ b/crates/lumen-stats/tests/busy_connect.rs @@ -0,0 +1,44 @@ +//! The GUI opens the ledger through `connect`, and a writer that holds it for a moment must not +//! fail that open with "database is locked": SQLite refuses at once, whatever the busy timeout +//! says, when the connection that switches a rollback-journal ledger to WAL already holds a +//! read lock (see `lumen_core::busy`), so `connect` has to wait for the writer itself. + +use std::time::Duration; + +use lumen_stats::connect; +use sqlx::Connection; +use sqlx::sqlite::{SqliteConnectOptions, SqliteConnection, SqliteJournalMode}; + +#[tokio::test] +async fn connect_waits_for_a_writer_instead_of_failing_busy() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("lumen.db"); + let url = format!("sqlite:{}?mode=rwc", path.display()); + connect(&url).await.unwrap().close().await; + + // The ledger as it is before anything has switched it to WAL, with a writer inside it. + let mut writer = SqliteConnection::connect_with( + &SqliteConnectOptions::new() + .filename(&path) + .journal_mode(SqliteJournalMode::Delete), + ) + .await + .unwrap(); + sqlx::query("BEGIN IMMEDIATE") + .execute(&mut writer) + .await + .unwrap(); + + let let_go = async { + tokio::time::sleep(Duration::from_millis(300)).await; + sqlx::query("COMMIT").execute(&mut writer).await.unwrap(); + }; + let (opened, ()) = tokio::join!(connect(&url), let_go); + + let pool = opened.expect("a writer that lets go within a moment must not fail the open"); + let mode: String = sqlx::query_scalar("PRAGMA journal_mode") + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(mode, "wal"); +} From e25b2024efc7ee93c41f77714e208e9e51d69f5d Mon Sep 17 00:00:00 2001 From: HackPoint <6758579+HackPoint@users.noreply.github.com> Date: Mon, 12 Oct 2026 14:07:57 +0300 Subject: [PATCH 3/5] fix(core): open_pool does not retry the connect, which takes no lock a writer can refuse sqlx leaves the journal mode alone unless asked (its source says so: changing into WAL needs an exclusive lock that the busy timeout cannot wait on), so a plain connect never meets a writer. The switch is the DDL's own pragma, and init_schema retries that. A probe shows it: with a writer inside a rollback-journal ledger the connect takes 2 ms and succeeds, and the DDL is the statement that is refused at once. --- crates/lumen-core/src/schema.rs | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/crates/lumen-core/src/schema.rs b/crates/lumen-core/src/schema.rs index 5ea45e1..f92545a 100644 --- a/crates/lumen-core/src/schema.rs +++ b/crates/lumen-core/src/schema.rs @@ -38,10 +38,11 @@ pub async fn init_schema(pool: &sqlx::SqlitePool) -> Result<(), sqlx::Error> { /// Open the ledger at `url` as a pool and bring it up to date: what the daemon runs before it /// serves anything and what the app runs before it reads. /// -/// A ledger another program holds at that moment is waited for, not failed on: the connection -/// itself can be refused (it switches the file to WAL) as well as the statements after it. +/// A ledger another program holds at that moment is waited for, not failed on, by [`init_schema`]. +/// The connection takes no lock a writer could refuse: sqlx leaves the journal mode alone unless +/// asked, and the switch to WAL is the DDL's. pub async fn open_pool(url: &str) -> Result { - let pool = busy::retry_busy(|| sqlx::sqlite::SqlitePoolOptions::new().connect(url)).await?; + let pool = sqlx::sqlite::SqlitePoolOptions::new().connect(url).await?; init_schema(&pool).await?; Ok(pool) } From eac1e1865c8ff2b918ac554cefbce6cf70731fb5 Mon Sep 17 00:00:00 2001 From: HackPoint <6758579+HackPoint@users.noreply.github.com> Date: Mon, 12 Oct 2026 16:45:52 +0300 Subject: [PATCH 4/5] test(core): a busy backfill does not fail an open, a busy schema migration does, on both openers The backfill UPDATE takes the write lock on every open, even on a ledger already in WAL, so a writer that holds the ledger past the busy timeout made an open fail where it used to go on. Two tests fail on this commit for that (pool and rusqlite). The other new ones pin what is already right: a busy column migration is an error on the rusqlite opener as well as the pool opener, and the retry pauses between tries (it was only checked to try at least three times). --- crates/lumen-core/src/busy.rs | 10 +++ crates/lumen-core/tests/busy_ledger.rs | 106 +++++++++++++++++++++++-- 2 files changed, 111 insertions(+), 5 deletions(-) diff --git a/crates/lumen-core/src/busy.rs b/crates/lumen-core/src/busy.rs index 0f20cda..bbe963e 100644 --- a/crates/lumen-core/src/busy.rs +++ b/crates/lumen-core/src/busy.rs @@ -229,6 +229,11 @@ mod tests { 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:?}"); } @@ -283,6 +288,11 @@ mod tests { 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:?}"); } diff --git a/crates/lumen-core/tests/busy_ledger.rs b/crates/lumen-core/tests/busy_ledger.rs index 0cdd6dd..7289a5b 100644 --- a/crates/lumen-core/tests/busy_ledger.rs +++ b/crates/lumen-core/tests/busy_ledger.rs @@ -99,6 +99,31 @@ async fn backfilled_in(pool: &sqlx::SqlitePool) -> bool { marked == 1 } +const AGENT_ID: &str = + "SELECT COUNT(*) FROM pragma_table_info('read_events') WHERE name = 'agent_id'"; + +/// A ledger in WAL one column short, so the migration that adds the column has to run. +fn ledger_missing_a_column(dir: &Path) -> PathBuf { + let path = wal_ledger(dir); + let conn = Connection::open(&path).unwrap(); + conn.execute_batch("ALTER TABLE read_events DROP COLUMN agent_id") + .unwrap(); + assert!(!has_agent_id(&conn)); + path +} + +fn has_agent_id(conn: &Connection) -> bool { + conn.query_row(AGENT_ID, [], |r| r.get::<_, i64>(0)).unwrap() == 1 +} + +async fn has_agent_id_in(pool: &sqlx::SqlitePool) -> bool { + sqlx::query_scalar::<_, i64>(AGENT_ID) + .fetch_one(pool) + .await + .unwrap() + == 1 +} + #[test] fn the_rusqlite_open_waits_for_a_writer_instead_of_failing_busy() { let dir = tempfile::tempdir().unwrap(); @@ -139,6 +164,41 @@ fn the_rusqlite_open_gives_up_busy_once_the_budget_is_spent() { assert!(backfilled(&conn)); } +#[test] +fn the_rusqlite_open_does_not_count_a_migration_the_ledger_was_busy_for_as_applied() { + let dir = tempfile::tempdir().unwrap(); + let path = ledger_missing_a_column(dir.path()); + let writer = Writer::hold(&path, Duration::from_secs(60)); + + let err = meter::connect_db(&path) + .expect_err("a migration that could not run is not one that already had"); + writer.release(); + + assert!(busy::is_busy_rusqlite(&err), "{err}"); + + let conn = meter::connect_db(&path).expect("the next open finishes the job"); + assert!(has_agent_id(&conn)); +} + +#[test] +fn the_rusqlite_open_goes_on_without_a_backfill_the_ledger_stays_busy_for() { + let dir = tempfile::tempdir().unwrap(); + let path = wal_ledger(dir.path()); + let writer = Writer::hold(&path, Duration::from_secs(60)); + + let conn = meter::connect_db(&path) + .expect("the schema is whole, and the next open does the backfill"); + assert!( + !backfilled(&conn), + "the writer held the ledger throughout, so the backfill cannot have run" + ); + writer.release(); + drop(conn); + + let conn = meter::connect_db(&path).expect("the next open"); + assert!(backfilled(&conn)); +} + #[tokio::test] async fn the_pool_open_waits_for_a_writer_instead_of_failing_busy() { let dir = tempfile::tempdir().unwrap(); @@ -201,9 +261,9 @@ async fn impatient_pool(path: &Path) -> sqlx::SqlitePool { } #[tokio::test] -async fn a_migration_the_ledger_was_busy_for_is_applied_not_skipped() { +async fn a_column_the_ledger_was_busy_to_add_is_added_not_skipped() { let dir = tempfile::tempdir().unwrap(); - let path = wal_ledger(dir.path()); + let path = ledger_missing_a_column(dir.path()); let pool = impatient_pool(&path).await; let writer = Writer::hold(&path, Duration::from_millis(400)); @@ -211,15 +271,15 @@ async fn a_migration_the_ledger_was_busy_for_is_applied_not_skipped() { writer.release(); assert!( - backfilled_in(&pool).await, - "the backfill was refused while the writer held the ledger and never ran again" + has_agent_id_in(&pool).await, + "the migration was refused while the writer held the ledger and never ran again" ); } #[tokio::test] async fn a_ledger_that_stays_busy_is_an_error_not_an_already_applied_migration() { let dir = tempfile::tempdir().unwrap(); - let path = wal_ledger(dir.path()); + let path = ledger_missing_a_column(dir.path()); let pool = impatient_pool(&path).await; let writer = Writer::hold(&path, Duration::from_secs(60)); @@ -243,5 +303,41 @@ async fn a_ledger_that_stays_busy_is_an_error_not_an_already_applied_migration() schema::init_schema(&pool) .await .expect("the next open finishes the job"); + assert!(has_agent_id_in(&pool).await); +} + +#[tokio::test] +async fn a_backfill_the_ledger_was_busy_for_is_run_again_not_skipped() { + let dir = tempfile::tempdir().unwrap(); + let path = wal_ledger(dir.path()); + let pool = impatient_pool(&path).await; + let writer = Writer::hold(&path, Duration::from_millis(400)); + + schema::init_schema(&pool).await.expect("init_schema"); + writer.release(); + + assert!( + backfilled_in(&pool).await, + "the backfill was refused while the writer held the ledger and never ran again" + ); +} + +#[tokio::test] +async fn a_backfill_the_ledger_stays_busy_for_does_not_fail_the_open() { + let dir = tempfile::tempdir().unwrap(); + let path = wal_ledger(dir.path()); + let pool = impatient_pool(&path).await; + let writer = Writer::hold(&path, Duration::from_secs(60)); + + schema::init_schema(&pool) + .await + .expect("the schema is whole, and the next open does the backfill"); + assert!( + !backfilled_in(&pool).await, + "the writer held the ledger throughout, so the backfill cannot have run" + ); + writer.release(); + + schema::init_schema(&pool).await.expect("the next open"); assert!(backfilled_in(&pool).await); } From 7f2a36ef6823611082e139884dab271d2d3be5c3 Mon Sep 17 00:00:00 2001 From: HackPoint <6758579+HackPoint@users.noreply.github.com> Date: Mon, 12 Oct 2026 16:47:38 +0300 Subject: [PATCH 5/5] fix(core): a backfill the ledger stays busy for is skipped, not an open that fails The one row rewrite among the migrations takes the write lock on every open, so on a ledger a writer held past the busy timeout the open now returned the busy error where it used to go on, and a hook lost its row. It moves to BACKFILLS, which both openers run after the migrations and read as before: not an error, and done by the next open. Schema migrations stay strict. open_pool closes its pool when the schema step fails; the MIGRATIONS doc says what the openers now do with a busy statement. --- CHANGELOG.md | 5 ++-- crates/lumen-core/src/meter.rs | 8 ++++-- crates/lumen-core/src/schema.rs | 28 +++++++++++++++------ crates/lumen-core/tests/busy_ledger.rs | 8 +++--- crates/lumen-core/tests/silent_fallbacks.rs | 14 +++++++++++ 5 files changed, 49 insertions(+), 14 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d08b4da..5d2a84c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -24,8 +24,9 @@ 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. Any other migration error (a column that already exists) is still read as -applied. +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 diff --git a/crates/lumen-core/src/meter.rs b/crates/lumen-core/src/meter.rs index 47e6647..994ea09 100644 --- a/crates/lumen-core/src/meter.rs +++ b/crates/lumen-core/src/meter.rs @@ -1,5 +1,5 @@ use crate::busy; -use crate::schema::{DDL, MIGRATIONS}; +use crate::schema::{BACKFILLS, DDL, MIGRATIONS}; use rusqlite::{Connection, params}; /// Detect which Claude Code channel spawned this process. @@ -122,7 +122,8 @@ pub fn resolve_db_path(lumen_db: Option<&str>, home: Option<&str>) -> Option rusqlite::Result { restore_sidecar_write_bits(path); let conn = Connection::open(path)?; @@ -135,6 +136,9 @@ fn open_db(path: &std::path::Path) -> rusqlite::Result { return Err(e); } } + for backfill in BACKFILLS { + busy::retry_busy_blocking(|| conn.execute_batch(backfill)).ok(); + } Ok(conn) } diff --git a/crates/lumen-core/src/schema.rs b/crates/lumen-core/src/schema.rs index f92545a..cca3f0d 100644 --- a/crates/lumen-core/src/schema.rs +++ b/crates/lumen-core/src/schema.rs @@ -18,7 +18,8 @@ use crate::busy; /// busy (see [`crate::busy`]) means the statement did not run, and calling that /// "already applied" would leave a column or an index missing for good. It is /// waited for, and returned if the wait runs out; the next open runs the -/// migration again. +/// migration again. [`BACKFILLS`] are not read that way: they rewrite rows, not the +/// schema, so a ledger that stays busy for one does not fail the open. pub async fn init_schema(pool: &sqlx::SqlitePool) -> Result<(), sqlx::Error> { busy::retry_busy(|| sqlx::raw_sql(DDL).execute(pool)).await?; for migration in MIGRATIONS { @@ -32,6 +33,12 @@ pub async fn init_schema(pool: &sqlx::SqlitePool) -> Result<(), sqlx::Error> { return Err(e); } } + for backfill in BACKFILLS { + // Literals in BACKFILLS, as above. + busy::retry_busy(|| sqlx::raw_sql(sqlx::AssertSqlSafe(*backfill)).execute(pool)) + .await + .ok(); + } Ok(()) } @@ -43,22 +50,21 @@ pub async fn init_schema(pool: &sqlx::SqlitePool) -> Result<(), sqlx::Error> { /// asked, and the switch to WAL is the DDL's. pub async fn open_pool(url: &str) -> Result { let pool = sqlx::sqlite::SqlitePoolOptions::new().connect(url).await?; - init_schema(&pool).await?; + if let Err(e) = init_schema(&pool).await { + pool.close().await; + return Err(e); + } Ok(pool) } /// Additive migrations run after DDL to upgrade existing DBs. -/// Each statement is idempotent (errors for "duplicate column" are swallowed by the caller). +/// Each statement is idempotent: the openers read any error but a busy ledger as "already applied". pub const MIGRATIONS: &[&str] = &[ "ALTER TABLE read_events ADD COLUMN channel TEXT NOT NULL DEFAULT 'unknown'", // Subagent transcripts reuse the parent's sessionId, so their turns land in // the parent session. They are real spend (kept in cost) but carry their own // fresh context, so they must NOT drive the context-fill gauge. "ALTER TABLE turns ADD COLUMN is_subagent INTEGER NOT NULL DEFAULT 0", - // Backfill existing rows from the path they were ingested from. Idempotent, - // so it is safe to re-run on every open. - "UPDATE turns SET is_subagent = 1 \ - WHERE is_subagent = 0 AND source_file LIKE '%/subagents/%'", // ── E7 instrumentation ─────────────────────────────────────────────────── // Existing rows stay NULL. There is no honest way to backfill provenance for // events already recorded, and inventing it would be worse than admitting the @@ -177,6 +183,14 @@ pub const MIGRATIONS: &[&str] = &[ "CREATE INDEX IF NOT EXISTS idx_faults_kind ON faults(kind, variant)", ]; +/// Row rewrites, run after [`MIGRATIONS`]: a ledger that stays busy for one is skipped, not an +/// error, and the next open runs it again. +pub const BACKFILLS: &[&str] = &[ + // Mark the turns ingested from a subagent path. Idempotent, so safe to re-run on every open. + "UPDATE turns SET is_subagent = 1 \ + WHERE is_subagent = 0 AND source_file LIKE '%/subagents/%'", +]; + /// Column count of `read_events` on a current database — DDL plus every ALTER in /// [`MIGRATIONS`]. `lumen report` compares the live table against this to detect a /// database that missed a migration, so adding an ALTER means bumping this number; diff --git a/crates/lumen-core/tests/busy_ledger.rs b/crates/lumen-core/tests/busy_ledger.rs index 7289a5b..a6facad 100644 --- a/crates/lumen-core/tests/busy_ledger.rs +++ b/crates/lumen-core/tests/busy_ledger.rs @@ -113,7 +113,9 @@ fn ledger_missing_a_column(dir: &Path) -> PathBuf { } fn has_agent_id(conn: &Connection) -> bool { - conn.query_row(AGENT_ID, [], |r| r.get::<_, i64>(0)).unwrap() == 1 + conn.query_row(AGENT_ID, [], |r| r.get::<_, i64>(0)) + .unwrap() + == 1 } async fn has_agent_id_in(pool: &sqlx::SqlitePool) -> bool { @@ -186,8 +188,8 @@ fn the_rusqlite_open_goes_on_without_a_backfill_the_ledger_stays_busy_for() { let path = wal_ledger(dir.path()); let writer = Writer::hold(&path, Duration::from_secs(60)); - let conn = meter::connect_db(&path) - .expect("the schema is whole, and the next open does the backfill"); + let conn = + meter::connect_db(&path).expect("the schema is whole, and the next open does the backfill"); assert!( !backfilled(&conn), "the writer held the ledger throughout, so the backfill cannot have run" diff --git a/crates/lumen-core/tests/silent_fallbacks.rs b/crates/lumen-core/tests/silent_fallbacks.rs index 70aeba4..7853cbe 100644 --- a/crates/lumen-core/tests/silent_fallbacks.rs +++ b/crates/lumen-core/tests/silent_fallbacks.rs @@ -168,6 +168,13 @@ const ALLOWED: &[Allowed] = &[ "CLAUDE_CODE_ENTRYPOINT", "Unset selects the default channel label.", ), + allow( + "crates/lumen-core/src/meter.rs", + DroppedWrite, + "execute_batch(backfill)", + "A row rewrite, not a schema change, and idempotent: a ledger that stayed busy for it \ + keeps its rows as they were and the next open runs it again.", + ), allow( "crates/lumen-core/src/meter.rs", DroppedWrite, @@ -250,6 +257,13 @@ const ALLOWED: &[Allowed] = &[ "let html =", "The link shown after filing, once the POST has already succeeded.", ), + allow( + "crates/lumen-core/src/schema.rs", + DroppedWrite, + "AssertSqlSafe(*backfill)", + "A row rewrite, not a schema change, and idempotent: a ledger that stayed busy for it \ + keeps its rows as they were and the next open runs it again.", + ), allow( "crates/lumen-core/src/update.rs", OrDefault,