diff --git a/CHANGELOG.md b/CHANGELOG.md index c609f3b..5d2a84c 100644 --- a/CHANGELOG.md +++ b/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 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 new file mode 100644 index 0000000..bbe963e --- /dev/null +++ b/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::().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 { + 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, + 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>, +{ + 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(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 { + 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> { + 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!( + 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:?}"); + } +} 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/meter.rs b/crates/lumen-core/src/meter.rs index 2ea0963..994ea09 100644 --- a/crates/lumen-core/src/meter.rs +++ b/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 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); + } + } + 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 8a21ceb..cca3f0d 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,31 +12,59 @@ /// /// 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. [`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> { - 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); + } + } + for backfill in BACKFILLS { + // Literals in BACKFILLS, as above. + busy::retry_busy(|| sqlx::raw_sql(sqlx::AssertSqlSafe(*backfill)).execute(pool)) + .await + .ok(); } 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, 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 = sqlx::sqlite::SqlitePoolOptions::new().connect(url).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 @@ -153,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 new file mode 100644 index 0000000..a6facad --- /dev/null +++ b/crates/lumen-core/tests/busy_ledger.rs @@ -0,0 +1,345 @@ +//! 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 +} + +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(); + 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)); +} + +#[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(); + 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_column_the_ledger_was_busy_to_add_is_added_not_skipped() { + let dir = tempfile::tempdir().unwrap(); + let path = ledger_missing_a_column(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!( + 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 = ledger_missing_a_column(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!(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); +} diff --git a/crates/lumen-core/tests/silent_fallbacks.rs b/crates/lumen-core/tests/silent_fallbacks.rs index e79a839..7853cbe 100644 --- a/crates/lumen-core/tests/silent_fallbacks.rs +++ b/crates/lumen-core/tests/silent_fallbacks.rs @@ -171,9 +171,9 @@ const ALLOWED: &[Allowed] = &[ 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.", + "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", @@ -260,10 +260,9 @@ const ALLOWED: &[Allowed] = &[ 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.", + "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", 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"); +}