From c8cd461f82f8a7d7198b86f55341e0b04eec900c Mon Sep 17 00:00:00 2001 From: Miguel Rasero Date: Sun, 26 Jul 2026 11:48:10 +0000 Subject: [PATCH 1/3] perf(db): default to WAL with a delete fallback and a 30s busy timeout Rollback-journal mode gives SQLite one writer whose exclusive lock blocks every reader. With the transcript-ingest, activity-monitor and PR-monitor services all writing continuously against a 215MB database, ordinary requests spent their whole busy timeout queued behind that lock and then failed: roughly two thirds of observed slow statements were plain SELECTs, and workspace creation lost the race often enough to 500 repeatedly. Default to WAL so readers no longer block the writer, raise the busy timeout from sqlx's 5s to 30s so contention delays a request instead of failing it, and pair WAL with synchronous=NORMAL to drop an fsync per commit. Upstream pinned `delete` in #1882 after reverting #1806 without recording a reason. The plausible one is portability: WAL needs an mmap-able `-shm` sidecar and so fails on some network mounts. Rather than assume, connect falls back to `delete` automatically when WAL cannot be established, and VIBE_KANBAN_SQLITE_JOURNAL_MODE forces either mode outright. Route all three connection sites through one helper so the settings cannot drift, and cover the pragmas that SQLite actually settles on -- journal_mode is a property of the file, so requesting WAL is not the same as getting it. --- Cargo.lock | 1 + crates/db/Cargo.toml | 1 + crates/db/src/lib.rs | 236 ++++++++++++++++++++++++++++++++++++------- 3 files changed, 204 insertions(+), 34 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 53e5cb595aa..5e9bb43be05 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2028,6 +2028,7 @@ dependencies = [ "sqlx", "strum", "strum_macros 0.27.2", + "tempfile", "thiserror 2.0.18", "tokio", "tracing", diff --git a/crates/db/Cargo.toml b/crates/db/Cargo.toml index e5d62c7dd05..ca173a0e9ed 100644 --- a/crates/db/Cargo.toml +++ b/crates/db/Cargo.toml @@ -22,3 +22,4 @@ futures = "0.3.32" [dev-dependencies] tokio = { workspace = true } +tempfile = "3" diff --git a/crates/db/src/lib.rs b/crates/db/src/lib.rs index bc5b090e36f..e9fc23533ea 100644 --- a/crates/db/src/lib.rs +++ b/crates/db/src/lib.rs @@ -1,14 +1,109 @@ -use std::sync::Arc; +use std::{sync::Arc, time::Duration}; use sqlx::{ - ConnectOptions, Error, Pool, Sqlite, SqlitePool, + ConnectOptions, Error, Pool, Sqlite, migrate::MigrateError, - sqlite::{SqliteConnectOptions, SqliteConnection, SqliteJournalMode, SqlitePoolOptions}, + sqlite::{ + SqliteConnectOptions, SqliteConnection, SqliteJournalMode, SqlitePoolOptions, + SqliteSynchronous, + }, }; use utils::assets::{DB_FILE_NAME, asset_dir}; pub mod models; +/// Overrides the journal mode, accepting `wal` or `delete`. +/// +/// We default to WAL: in rollback-journal mode a writer holds an exclusive lock +/// that blocks every reader, so a continuously-writing background service (CLI +/// transcript ingest, activity monitors, PR monitor) starves ordinary requests +/// until they exhaust the busy timeout and surface as 500s. Upstream pinned +/// `delete` in #1882 after reverting #1806, but recorded no reason for the +/// revert; the likely one is that WAL needs an mmap-able `-shm` sidecar and so +/// fails on network mounts. [`connect_pool`] detects that and falls back, and +/// this variable forces the old behaviour outright. +const JOURNAL_MODE_ENV: &str = "VIBE_KANBAN_SQLITE_JOURNAL_MODE"; + +/// SQLite serialises writers, so contention is normal and waiting is correct. +/// sqlx defaults to 5s, which a slow commit can exceed — and losing that race +/// fails the request rather than delaying it. +const BUSY_TIMEOUT: Duration = Duration::from_secs(30); + +fn preferred_journal_mode() -> SqliteJournalMode { + match std::env::var(JOURNAL_MODE_ENV) { + Err(_) => SqliteJournalMode::Wal, + Ok(raw) => match raw.trim().to_ascii_lowercase().as_str() { + "wal" => SqliteJournalMode::Wal, + "delete" => SqliteJournalMode::Delete, + other => { + tracing::warn!( + "{JOURNAL_MODE_ENV}={other:?} is not a supported journal mode \ + (expected `wal` or `delete`); using `wal`" + ); + SqliteJournalMode::Wal + } + }, + } +} + +/// Applies the concurrency settings to any connection, separately from which +/// file it points at, so the settings can be exercised against a scratch +/// database in tests. +fn tune(options: SqliteConnectOptions, journal_mode: SqliteJournalMode) -> SqliteConnectOptions { + options + .journal_mode(journal_mode) + .busy_timeout(BUSY_TIMEOUT) + // WAL already fsyncs the log before a checkpoint can discard it, so + // NORMAL only risks the most recent commits on host power loss — never + // on process crash — and removes an fsync from every commit. + .synchronous(match journal_mode { + SqliteJournalMode::Wal => SqliteSynchronous::Normal, + _ => SqliteSynchronous::Full, + }) +} + +fn connect_options(journal_mode: SqliteJournalMode) -> SqliteConnectOptions { + tune( + SqliteConnectOptions::new() + .filename(asset_dir().join(DB_FILE_NAME)) + .create_if_missing(true), + journal_mode, + ) +} + +/// Connects with the preferred journal mode, retrying once in `delete` mode so +/// a filesystem that cannot host WAL's shared-memory file still opens. +async fn connect_pool( + pool_options: SqlitePoolOptions, + disable_statement_logging: bool, +) -> Result, Error> { + let build = |journal_mode| { + let options = connect_options(journal_mode); + if disable_statement_logging { + options.disable_statement_logging() + } else { + options + } + }; + + let preferred = preferred_journal_mode(); + match pool_options.clone().connect_with(build(preferred)).await { + Ok(pool) => Ok(pool), + Err(err) if preferred == SqliteJournalMode::Wal => { + tracing::warn!( + %err, + "could not open the database in WAL mode (the filesystem may not support \ + shared memory); falling back to `delete`. Set {JOURNAL_MODE_ENV}=delete \ + to select it explicitly." + ); + pool_options + .connect_with(build(SqliteJournalMode::Delete)) + .await + } + Err(err) => Err(err), + } +} + async fn run_migrations(pool: &Pool) -> Result<(), Error> { use std::collections::HashSet; @@ -79,25 +174,13 @@ pub struct DBService { impl DBService { pub async fn new() -> Result { - let options = SqliteConnectOptions::new() - .filename(asset_dir().join(DB_FILE_NAME)) - .create_if_missing(true) - .journal_mode(SqliteJournalMode::Delete); - let pool = SqlitePool::connect_with(options).await?; + let pool = connect_pool(SqlitePoolOptions::new(), false).await?; run_migrations(&pool).await?; Ok(DBService { pool }) } pub async fn new_migration_pool() -> Result, Error> { - let options = SqliteConnectOptions::new() - .filename(asset_dir().join(DB_FILE_NAME)) - .create_if_missing(true) - .journal_mode(SqliteJournalMode::Delete) - .disable_statement_logging(); - SqlitePoolOptions::new() - .max_connections(64) - .connect_with(options) - .await + connect_pool(SqlitePoolOptions::new().max_connections(64), true).await } pub async fn new_with_after_connect(after_connect: F) -> Result @@ -124,25 +207,17 @@ impl DBService { + Sync + 'static, { - let options = SqliteConnectOptions::new() - .filename(asset_dir().join(DB_FILE_NAME)) - .create_if_missing(true) - .journal_mode(SqliteJournalMode::Delete); - - let pool = if let Some(hook) = after_connect { - SqlitePoolOptions::new() - .after_connect(move |conn, _meta| { - let hook = hook.clone(); - Box::pin(async move { - hook(conn).await?; - Ok(()) - }) + let pool_options = match after_connect { + Some(hook) => SqlitePoolOptions::new().after_connect(move |conn, _meta| { + let hook = hook.clone(); + Box::pin(async move { + hook(conn).await?; + Ok(()) }) - .connect_with(options) - .await? - } else { - SqlitePool::connect_with(options).await? + }), + None => SqlitePoolOptions::new(), }; + let pool = connect_pool(pool_options, false).await?; run_migrations(&pool).await?; Ok(pool) @@ -154,6 +229,99 @@ mod tests { use sqlx::sqlite::SqlitePoolOptions; use uuid::Uuid; + use super::*; + + /// Opens a scratch database with the real tuning and reports the pragmas + /// SQLite actually settled on — `journal_mode` is a property of the file, so + /// asking for WAL is not the same as getting it. + async fn effective_pragmas(journal_mode: SqliteJournalMode) -> (String, i64) { + let dir = tempfile::tempdir().unwrap(); + let pool = SqlitePoolOptions::new() + .max_connections(1) + .connect_with(tune( + SqliteConnectOptions::new() + .filename(dir.path().join("pragmas.sqlite")) + .create_if_missing(true), + journal_mode, + )) + .await + .unwrap(); + let mode = sqlx::query_scalar::<_, String>("PRAGMA journal_mode") + .fetch_one(&pool) + .await + .unwrap(); + let timeout = sqlx::query_scalar::<_, i64>("PRAGMA busy_timeout") + .fetch_one(&pool) + .await + .unwrap(); + pool.close().await; + (mode, timeout) + } + + #[tokio::test] + async fn wal_and_a_generous_busy_timeout_are_actually_applied() { + let (mode, busy_timeout) = effective_pragmas(SqliteJournalMode::Wal).await; + assert_eq!(mode, "wal"); + assert_eq!(busy_timeout, BUSY_TIMEOUT.as_millis() as i64); + } + + #[tokio::test] + async fn the_delete_fallback_remains_available() { + let (mode, busy_timeout) = effective_pragmas(SqliteJournalMode::Delete).await; + assert_eq!(mode, "delete"); + assert_eq!(busy_timeout, BUSY_TIMEOUT.as_millis() as i64); + } + + #[tokio::test] + async fn a_reader_does_not_block_a_writer_under_wal() { + // The behaviour the whole change exists for: in `delete` mode an open + // read transaction holds a shared lock that fails the writer, which is + // what surfaced as 500s. + let dir = tempfile::tempdir().unwrap(); + let options = tune( + SqliteConnectOptions::new() + .filename(dir.path().join("concurrent.sqlite")) + .create_if_missing(true), + SqliteJournalMode::Wal, + ) + // Without waiting, any blocking would surface immediately as an error. + .busy_timeout(Duration::ZERO); + let pool = SqlitePoolOptions::new() + .max_connections(2) + .connect_with(options) + .await + .unwrap(); + sqlx::query("CREATE TABLE t (v INTEGER)") + .execute(&pool) + .await + .unwrap(); + sqlx::query("INSERT INTO t (v) VALUES (1)") + .execute(&pool) + .await + .unwrap(); + + let mut reader = pool.begin().await.unwrap(); + let seen = sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM t") + .fetch_one(&mut *reader) + .await + .unwrap(); + assert_eq!(seen, 1); + + sqlx::query("INSERT INTO t (v) VALUES (2)") + .execute(&pool) + .await + .expect("an open read transaction must not block a writer under WAL"); + + // The reader keeps its original snapshot. + let still_seen = sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM t") + .fetch_one(&mut *reader) + .await + .unwrap(); + assert_eq!(still_seen, 1); + drop(reader); + pool.close().await; + } + #[tokio::test] async fn archived_at_migration_backfills_preexisting_archived_rows() { let pool = SqlitePoolOptions::new() From 23875a17cee9c67e6e9d038e0d7b8858758147c6 Mon Sep 17 00:00:00 2001 From: Miguel Rasero Date: Sun, 26 Jul 2026 11:48:18 +0000 Subject: [PATCH 2/3] fix(server): report SQLite contention as 503, not a generic internal error Every database failure mapped to ErrorInfo::internal, so a request that merely lost the write race answered 500 with "An internal error occurred. Please try again." -- the real cause never reached the browser and a week of lock contention read as a workspace-creation bug. Contention is retryable and the server is healthy, so answer 503 with a message that says so. Match on the error source chain rather than adding cases per variant: the same sqlx error arrives wrapped in Container, Workspace or ScratchError depending on the route. Two details the tests pin down, both of which silently defeated an earlier attempt at this. sqlx implements StdError for Box, so the chain element is that box and not the driver's concrete SqliteError. And the code sqlx reports is SQLite's extended result code, so the primary code has to be masked out of the low byte -- WAL raises SQLITE_BUSY_SNAPSHOT (517), not 5. --- crates/server/src/error.rs | 133 +++++++++++++++++++++++++++++++++++++ 1 file changed, 133 insertions(+) diff --git a/crates/server/src/error.rs b/crates/server/src/error.rs index 77d8d72fa13..71c1e97e140 100644 --- a/crates/server/src/error.rs +++ b/crates/server/src/error.rs @@ -27,6 +27,7 @@ use services::services::{ remote_client::RemoteClientError, repo::RepoError as RepoServiceError, }; +use sqlx::error::DatabaseError; use thiserror::Error; use trusted_key_auth::error::TrustedKeyAuthError; use utils::response::ApiResponse; @@ -343,8 +344,45 @@ fn remote_client_error(err: &RemoteClientError) -> ErrorInfo { } } +/// SQLite serialises writers, so a request that loses the race exhausts the +/// busy timeout and fails with `SQLITE_BUSY`/`SQLITE_LOCKED`. That is +/// contention, not a defect: the request is retryable and the server is +/// healthy, so it should not be reported as an opaque internal error. +fn sqlite_contention(err: &(dyn std::error::Error + 'static)) -> bool { + // The failure reaches here wrapped in whichever domain error the route + // happened to use, so walk the source chain rather than enumerating every + // variant that can carry one. The chain ends at the driver error: those + // wrappers are `#[error(transparent)]`, and that forwards `source()` past + // the error it wraps, so the intermediate `sqlx::Error` is never yielded. + std::iter::successors(Some(err), |err| err.source()).any(|err| { + // The chain element is the `Box` itself, not the + // driver's concrete error type: sqlx writes `impl StdError for + // Box`, so the box is what carries the vtable here. + err.downcast_ref::>() + .and_then(|err| err.code()) + // sqlx reports SQLite's *extended* result code, so the primary code + // is the low byte: plain SQLITE_BUSY is 5, but WAL also raises + // SQLITE_BUSY_SNAPSHOT (517) and recovery raises 261. + .and_then(|code| code.parse::().ok()) + .is_some_and(|code| matches!(code & 0xff, SQLITE_BUSY | SQLITE_LOCKED)) + }) +} + +const SQLITE_BUSY: u32 = 5; +const SQLITE_LOCKED: u32 = 6; + impl IntoResponse for ApiError { fn into_response(self) -> Response { + if sqlite_contention(&self) { + tracing::warn!( + error = ?self, + "request failed on database contention; retryable" + ); + let response = + ApiResponse::<()>::error("The database is busy right now. Please try that again."); + return (StatusCode::SERVICE_UNAVAILABLE, Json(response)).into_response(); + } + let info = match &self { ApiError::Repo(RepoError::Database(_)) => ErrorInfo::internal("RepoError"), ApiError::Repo(RepoError::NotFound) => { @@ -658,6 +696,11 @@ impl From for ApiError { #[cfg(test)] mod tests { + use sqlx::{ + Connection, + sqlite::{SqliteConnectOptions, SqliteConnection}, + }; + use super::*; #[test] @@ -675,4 +718,94 @@ mod tests { StatusCode::NOT_FOUND ); } + + /// Provokes a genuine `SQLITE_BUSY` rather than a stand-in: `SqliteError` + /// cannot be constructed outside sqlx, and the behaviour under test is + /// precisely that a *real* error's source chain is matched. Wrappers here + /// are `#[error(transparent)]`, which forwards `source()` past the error it + /// wraps, so a hand-built chain would not prove anything. + async fn real_busy_error() -> sqlx::Error { + let dir = tempfile::tempdir().unwrap(); + let options = SqliteConnectOptions::new() + .filename(dir.path().join("busy.sqlite")) + .create_if_missing(true) + // A zero timeout makes losing the write race immediate, so the test + // is deterministic rather than dependent on timing. + .busy_timeout(std::time::Duration::ZERO); + + let mut holder = SqliteConnection::connect_with(&options).await.unwrap(); + sqlx::query("CREATE TABLE t (v INTEGER)") + .execute(&mut holder) + .await + .unwrap(); + let mut contender = SqliteConnection::connect_with(&options).await.unwrap(); + sqlx::query("BEGIN EXCLUSIVE") + .execute(&mut holder) + .await + .unwrap(); + + let err = sqlx::query("INSERT INTO t (v) VALUES (1)") + .execute(&mut contender) + .await + .expect_err("the exclusive lock must block this write"); + drop(dir); + err + } + + async fn real_constraint_error() -> sqlx::Error { + let mut conn = SqliteConnection::connect("sqlite::memory:").await.unwrap(); + for statement in [ + "CREATE TABLE t (v INTEGER PRIMARY KEY)", + "INSERT INTO t (v) VALUES (1)", + ] { + sqlx::query(statement).execute(&mut conn).await.unwrap(); + } + sqlx::query("INSERT INTO t (v) VALUES (1)") + .execute(&mut conn) + .await + .expect_err("duplicate primary key") + } + + #[tokio::test] + async fn contention_is_detected_through_every_wrapper_a_route_uses() { + let err = real_busy_error().await; + assert!( + matches!(&err, sqlx::Error::Database(db) if db.code().as_deref() == Some("5")), + "expected SQLITE_BUSY, got {err:?}" + ); + assert!(sqlite_contention(&ApiError::Database(err))); + // The variants that produced the observed 500s. + assert!(sqlite_contention(&ApiError::Workspace( + WorkspaceError::Database(real_busy_error().await) + ))); + assert!(sqlite_contention(&ApiError::Container( + ContainerError::Sqlx(real_busy_error().await) + ))); + } + + #[tokio::test] + async fn contention_answers_service_unavailable_not_internal_error() { + let response = + ApiError::Container(ContainerError::Sqlx(real_busy_error().await)).into_response(); + assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE); + } + + #[tokio::test] + async fn other_sqlite_errors_still_report_internal() { + // Masking the extended code to its low byte must not catch unrelated + // failures: a primary-key violation is 1555, whose low byte is 19. + let err = ApiError::Database(real_constraint_error().await); + assert!(!sqlite_contention(&err)); + assert_eq!( + err.into_response().status(), + StatusCode::INTERNAL_SERVER_ERROR + ); + } + + #[test] + fn non_database_errors_are_unaffected() { + assert!(!sqlite_contention(&ApiError::Database( + sqlx::Error::RowNotFound + ))); + } } From 43dcfa995c5555a09cd5e526532d66dd381ee3d4 Mon Sep 17 00:00:00 2001 From: Miguel Rasero Date: Sun, 26 Jul 2026 11:48:41 +0000 Subject: [PATCH 3/3] perf(cli-native): stop rewriting unchanged registrations, reclaim dead transcripts Two sources of avoidable write-lock pressure from transcript ingest. CliNativeFile::register wrote unconditionally on every scan of every tracked file, so a poll that observed no change still took the single write lock just to restamp updated_at -- which nothing reads, and which a real import restamps anyway. It then re-read the row it had just written. This was the single hottest slow statement observed (3,643 of 14,071). Compare against the row find_latest_by_path already loaded and skip both round trips when nothing changed. Nothing ever reclaimed raw transcript records, which had grown to 196MB of a 215MB database in six days, and a larger database makes every commit and so every lock hold slower. cli_native_files has no foreign key to sessions, so deleting a session stranded its transcript permanently; the cascades that would have collected the rest never fire because PRAGMA foreign_keys defaults to off on every connection, which is also why the sweep deletes children explicitly. A file is only prunable once it is past the retention window and unreachable: no claude_session_links row, so no session can render it and the unassigned-CLI adoption list has had the full window to claim it, and no outbox row belonging to a session that still exists, so no live feed can be replaying it. Sweeps are capped per run and each file is its own transaction, because holding the write lock is the problem retention exists to relieve. Retention defaults to 14 days; VIBE_KANBAN_CLI_TRANSCRIPT_RETENTION_DAYS tunes it and 0 disables it. --- ...752f1a6f39f3daee356fb533cdd2795d8f9fa.json | 12 + ...ac7f75887ad9e1db6e91e78c0a1812a5d0fe6.json | 12 + ...6ca98ed29261fd4e78895e594e1def76ec3ba.json | 20 + ...fc9586dc5b07162f090016c352e1695f2c4f6.json | 12 + crates/db/src/models/cli_native_file.rs | 478 +++++++++++++++++- .../src/services/claude_transcript_ingest.rs | 65 +++ 6 files changed, 595 insertions(+), 4 deletions(-) create mode 100644 crates/db/.sqlx/query-1e58cbf7abbc2dda49a2f8127a9752f1a6f39f3daee356fb533cdd2795d8f9fa.json create mode 100644 crates/db/.sqlx/query-69fe66fcf15c9bee1894df9740bac7f75887ad9e1db6e91e78c0a1812a5d0fe6.json create mode 100644 crates/db/.sqlx/query-9d3606475632e9d6f8b8455101e6ca98ed29261fd4e78895e594e1def76ec3ba.json create mode 100644 crates/db/.sqlx/query-ca6f9129f197133edeece41f44cfc9586dc5b07162f090016c352e1695f2c4f6.json diff --git a/crates/db/.sqlx/query-1e58cbf7abbc2dda49a2f8127a9752f1a6f39f3daee356fb533cdd2795d8f9fa.json b/crates/db/.sqlx/query-1e58cbf7abbc2dda49a2f8127a9752f1a6f39f3daee356fb533cdd2795d8f9fa.json new file mode 100644 index 00000000000..b086d87215f --- /dev/null +++ b/crates/db/.sqlx/query-1e58cbf7abbc2dda49a2f8127a9752f1a6f39f3daee356fb533cdd2795d8f9fa.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "DELETE FROM cli_native_records WHERE file_id = $1", + "describe": { + "columns": [], + "parameters": { + "Right": 1 + }, + "nullable": [] + }, + "hash": "1e58cbf7abbc2dda49a2f8127a9752f1a6f39f3daee356fb533cdd2795d8f9fa" +} diff --git a/crates/db/.sqlx/query-69fe66fcf15c9bee1894df9740bac7f75887ad9e1db6e91e78c0a1812a5d0fe6.json b/crates/db/.sqlx/query-69fe66fcf15c9bee1894df9740bac7f75887ad9e1db6e91e78c0a1812a5d0fe6.json new file mode 100644 index 00000000000..d141e49c2b2 --- /dev/null +++ b/crates/db/.sqlx/query-69fe66fcf15c9bee1894df9740bac7f75887ad9e1db6e91e78c0a1812a5d0fe6.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "DELETE FROM cli_native_files WHERE id = $1", + "describe": { + "columns": [], + "parameters": { + "Right": 1 + }, + "nullable": [] + }, + "hash": "69fe66fcf15c9bee1894df9740bac7f75887ad9e1db6e91e78c0a1812a5d0fe6" +} diff --git a/crates/db/.sqlx/query-9d3606475632e9d6f8b8455101e6ca98ed29261fd4e78895e594e1def76ec3ba.json b/crates/db/.sqlx/query-9d3606475632e9d6f8b8455101e6ca98ed29261fd4e78895e594e1def76ec3ba.json new file mode 100644 index 00000000000..60cc14eb673 --- /dev/null +++ b/crates/db/.sqlx/query-9d3606475632e9d6f8b8455101e6ca98ed29261fd4e78895e594e1def76ec3ba.json @@ -0,0 +1,20 @@ +{ + "db_name": "SQLite", + "query": "SELECT f.id AS \"id!: Uuid\"\n FROM cli_native_files f\n WHERE COALESCE(f.last_import_at, f.updated_at, f.created_at)\n < datetime('now', $1)\n AND NOT EXISTS (\n SELECT 1 FROM claude_session_links l\n WHERE l.claude_session_id = f.claude_session_id\n )\n AND NOT EXISTS (\n SELECT 1 FROM cli_ingest_outbox o\n JOIN sessions s ON s.id = o.session_id\n WHERE o.file_id = f.id\n )\n ORDER BY COALESCE(f.last_import_at, f.updated_at, f.created_at) ASC\n LIMIT $2", + "describe": { + "columns": [ + { + "name": "id!: Uuid", + "ordinal": 0, + "type_info": "Blob" + } + ], + "parameters": { + "Right": 2 + }, + "nullable": [ + true + ] + }, + "hash": "9d3606475632e9d6f8b8455101e6ca98ed29261fd4e78895e594e1def76ec3ba" +} diff --git a/crates/db/.sqlx/query-ca6f9129f197133edeece41f44cfc9586dc5b07162f090016c352e1695f2c4f6.json b/crates/db/.sqlx/query-ca6f9129f197133edeece41f44cfc9586dc5b07162f090016c352e1695f2c4f6.json new file mode 100644 index 00000000000..c94f283ce38 --- /dev/null +++ b/crates/db/.sqlx/query-ca6f9129f197133edeece41f44cfc9586dc5b07162f090016c352e1695f2c4f6.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "DELETE FROM cli_ingest_outbox WHERE file_id = $1", + "describe": { + "columns": [], + "parameters": { + "Right": 1 + }, + "nullable": [] + }, + "hash": "ca6f9129f197133edeece41f44cfc9586dc5b07162f090016c352e1695f2c4f6" +} diff --git a/crates/db/src/models/cli_native_file.rs b/crates/db/src/models/cli_native_file.rs index c596b58c59c..3e26e7f7ab8 100644 --- a/crates/db/src/models/cli_native_file.rs +++ b/crates/db/src/models/cli_native_file.rs @@ -69,9 +69,26 @@ impl CliNativeFile { pool: &SqlitePool, data: &RegisterCliNativeFile<'_>, ) -> Result { - if let Some(existing) = + if let Some(mut existing) = Self::find_latest_by_path(pool, data.dir_path, data.file_name).await? { + // Registration runs on every scan of every tracked file, so writing + // unconditionally would take the database's single write lock once + // per file per poll just to restamp `updated_at` — which nothing + // reads, and which a real import restamps anyway. Only the observed + // stat fields and a first-time workspace attribution are worth a + // write; `COALESCE` in the statement below means a `None` discovery + // never clears an existing one, so it is not a change either. + let discovery = data + .discovered_workspace_id + .filter(|id| Some(*id) != existing.discovered_workspace_id); + let changed = discovery.is_some() + || existing.observed_size != data.observed_size + || existing.observed_mtime_ms != data.observed_mtime_ms; + if !changed { + return Ok(existing); + } + sqlx::query!( r#"UPDATE cli_native_files SET discovered_workspace_id = COALESCE($1, discovered_workspace_id), @@ -86,9 +103,13 @@ impl CliNativeFile { ) .execute(pool) .await?; - return Ok(Self::find_by_id(pool, existing.id) - .await? - .expect("row exists")); + // Re-reading only to observe writes we just made is another lock + // acquisition on the hot path; apply them to the row we already hold. + existing.discovered_workspace_id = discovery.or(existing.discovered_workspace_id); + existing.observed_size = data.observed_size; + existing.observed_mtime_ms = data.observed_mtime_ms; + existing.updated_at = Utc::now(); + return Ok(existing); } Self::insert_generation(pool, data, 0).await @@ -190,4 +211,453 @@ impl CliNativeFile { .fetch_all(pool) .await } + + /// Deletes ingested transcripts that nothing can reach any more. + /// + /// Raw transcript text is by far the largest thing this schema stores and + /// nothing ever reclaimed it: `cli_native_files` has no foreign key to + /// `sessions`, so deleting a session stranded its transcript, and the + /// cascades that would have collected the rest never run because + /// `PRAGMA foreign_keys` defaults to off on every connection. + /// + /// A file is prunable only once it is older than `retention_days` *and* + /// unreachable: it has no [`claude_session_links`] row, so no session can + /// render it and the unassigned-CLI adoption list has had the whole window + /// to claim it, and it has no outbox row belonging to a session that still + /// exists, so no live feed can be replaying it. + /// + /// Each file is deleted in its own transaction and `file_limit` bounds the + /// sweep: holding the single write lock is the precise problem retention + /// exists to relieve, so a backlog is worked off across sweeps. + pub async fn prune_unreachable( + pool: &SqlitePool, + retention_days: u32, + file_limit: i64, + ) -> Result { + let cutoff = format!("-{retention_days} days"); + let candidates = sqlx::query_scalar!( + r#"SELECT f.id AS "id!: Uuid" + FROM cli_native_files f + WHERE COALESCE(f.last_import_at, f.updated_at, f.created_at) + < datetime('now', $1) + AND NOT EXISTS ( + SELECT 1 FROM claude_session_links l + WHERE l.claude_session_id = f.claude_session_id + ) + AND NOT EXISTS ( + SELECT 1 FROM cli_ingest_outbox o + JOIN sessions s ON s.id = o.session_id + WHERE o.file_id = f.id + ) + ORDER BY COALESCE(f.last_import_at, f.updated_at, f.created_at) ASC + LIMIT $2"#, + cutoff, + file_limit + ) + .fetch_all(pool) + .await?; + + let mut pruned = PrunedTranscripts::default(); + for file_id in candidates { + let mut tx = pool.begin().await?; + // Children first and explicitly, since ON DELETE CASCADE is inert + // while foreign keys are disabled. + sqlx::query!("DELETE FROM cli_ingest_outbox WHERE file_id = $1", file_id) + .execute(&mut *tx) + .await?; + let records = + sqlx::query!("DELETE FROM cli_native_records WHERE file_id = $1", file_id) + .execute(&mut *tx) + .await? + .rows_affected(); + let files = sqlx::query!("DELETE FROM cli_native_files WHERE id = $1", file_id) + .execute(&mut *tx) + .await? + .rows_affected(); + tx.commit().await?; + pruned.records += records; + pruned.files += files; + } + Ok(pruned) + } +} + +/// What one retention sweep reclaimed. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct PrunedTranscripts { + pub files: u64, + pub records: u64, +} + +impl PrunedTranscripts { + pub fn is_empty(self) -> bool { + self.files == 0 && self.records == 0 + } +} + +#[cfg(test)] +mod tests { + use sqlx::sqlite::SqlitePoolOptions; + + use super::*; + use crate::models::{ + session::{CreateSession, Session}, + workspace::{CreateWorkspace, Workspace}, + }; + + const SID: &str = "06a7eacd-664b-4d9c-83f3-d4774a6216a8"; + /// Distinguishable from any timestamp a write would produce. + const SENTINEL: &str = "2000-01-01 00:00:00.000"; + + async fn pool() -> SqlitePool { + let pool = SqlitePoolOptions::new() + .max_connections(1) + .connect("sqlite::memory:") + .await + .unwrap(); + crate::run_migrations_for_tests(&pool).await.unwrap(); + pool + } + + fn registration( + observed_size: i64, + observed_mtime_ms: Option, + ) -> RegisterCliNativeFile<'static> { + RegisterCliNativeFile { + claude_session_id: SID, + dir_path: "/home/dev/project", + file_name: "session.jsonl", + discovered_workspace_id: None, + dev: 1, + inode: 2, + observed_size, + observed_mtime_ms, + } + } + + /// Backdates the bookkeeping timestamps so a later write is unmistakable + /// without depending on clock resolution. + async fn backdate(pool: &SqlitePool, id: Uuid) { + sqlx::query( + "UPDATE cli_native_files + SET updated_at = ?1, last_import_at = ?1, created_at = ?1 + WHERE id = ?2", + ) + .bind(SENTINEL) + .bind(id) + .execute(pool) + .await + .unwrap(); + } + + async fn updated_at(pool: &SqlitePool, id: Uuid) -> String { + sqlx::query_scalar::<_, String>("SELECT updated_at FROM cli_native_files WHERE id = ?") + .bind(id) + .fetch_one(pool) + .await + .unwrap() + } + + async fn add_records(pool: &SqlitePool, file_id: Uuid, count: i64) { + for line_seq in 0..count { + sqlx::query( + "INSERT INTO cli_native_records + (file_id, line_seq, claude_session_id, kind, raw, disposition) + VALUES (?, ?, ?, 'user', '{}', 'renderable')", + ) + .bind(file_id) + .bind(line_seq) + .bind(SID) + .execute(pool) + .await + .unwrap(); + } + } + + async fn remaining_records(pool: &SqlitePool) -> i64 { + sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM cli_native_records") + .fetch_one(pool) + .await + .unwrap() + } + + #[tokio::test] + async fn rescan_with_unchanged_stats_does_not_write() { + let pool = pool().await; + let file = CliNativeFile::register(&pool, ®istration(120, Some(9))) + .await + .unwrap(); + backdate(&pool, file.id).await; + + let again = CliNativeFile::register(&pool, ®istration(120, Some(9))) + .await + .unwrap(); + + assert_eq!(again.id, file.id); + assert_eq!( + updated_at(&pool, file.id).await, + SENTINEL, + "an unchanged rescan must not take the write lock" + ); + } + + #[tokio::test] + async fn rescan_with_new_stats_writes_them() { + let pool = pool().await; + let file = CliNativeFile::register(&pool, ®istration(120, Some(9))) + .await + .unwrap(); + backdate(&pool, file.id).await; + + let grown = CliNativeFile::register(&pool, ®istration(310, Some(11))) + .await + .unwrap(); + + assert_eq!(grown.observed_size, 310); + assert_eq!(grown.observed_mtime_ms, Some(11)); + assert_ne!(updated_at(&pool, file.id).await, SENTINEL); + // The returned row must match what a fresh read would see. + let reread = CliNativeFile::find_by_id(&pool, file.id) + .await + .unwrap() + .unwrap(); + assert_eq!(reread.observed_size, 310); + assert_eq!(reread.observed_mtime_ms, Some(11)); + } + + #[tokio::test] + async fn first_workspace_attribution_is_written_then_never_rewritten() { + let pool = pool().await; + let workspace_id = Uuid::new_v4(); + Workspace::create( + &pool, + &CreateWorkspace { + branch: "main".to_string(), + name: Some("retention test".to_string()), + }, + workspace_id, + ) + .await + .unwrap(); + + let file = CliNativeFile::register(&pool, ®istration(120, Some(9))) + .await + .unwrap(); + assert_eq!(file.discovered_workspace_id, None); + backdate(&pool, file.id).await; + + let attributed = CliNativeFile::register( + &pool, + &RegisterCliNativeFile { + discovered_workspace_id: Some(workspace_id), + ..registration(120, Some(9)) + }, + ) + .await + .unwrap(); + assert_eq!(attributed.discovered_workspace_id, Some(workspace_id)); + assert_ne!(updated_at(&pool, file.id).await, SENTINEL); + + // Re-attributing the same workspace is not a change. + backdate(&pool, file.id).await; + CliNativeFile::register( + &pool, + &RegisterCliNativeFile { + discovered_workspace_id: Some(workspace_id), + ..registration(120, Some(9)) + }, + ) + .await + .unwrap(); + assert_eq!(updated_at(&pool, file.id).await, SENTINEL); + } + + #[tokio::test] + async fn prune_removes_unreachable_transcripts_past_retention() { + let pool = pool().await; + let file = CliNativeFile::register(&pool, ®istration(120, Some(9))) + .await + .unwrap(); + add_records(&pool, file.id, 3).await; + backdate(&pool, file.id).await; + + let pruned = CliNativeFile::prune_unreachable(&pool, 14, 25) + .await + .unwrap(); + + assert_eq!( + pruned, + PrunedTranscripts { + files: 1, + records: 3 + } + ); + assert_eq!(remaining_records(&pool).await, 0); + assert!( + CliNativeFile::find_by_id(&pool, file.id) + .await + .unwrap() + .is_none() + ); + } + + #[tokio::test] + async fn prune_keeps_transcripts_inside_the_retention_window() { + let pool = pool().await; + let file = CliNativeFile::register(&pool, ®istration(120, Some(9))) + .await + .unwrap(); + add_records(&pool, file.id, 3).await; + + let pruned = CliNativeFile::prune_unreachable(&pool, 14, 25) + .await + .unwrap(); + + assert!(pruned.is_empty()); + assert_eq!(remaining_records(&pool).await, 3); + } + + #[tokio::test] + async fn prune_keeps_transcripts_a_session_still_owns() { + let pool = pool().await; + let file = CliNativeFile::register(&pool, ®istration(120, Some(9))) + .await + .unwrap(); + add_records(&pool, file.id, 3).await; + backdate(&pool, file.id).await; + + let workspace_id = Uuid::new_v4(); + Workspace::create( + &pool, + &CreateWorkspace { + branch: "main".to_string(), + name: Some("owner".to_string()), + }, + workspace_id, + ) + .await + .unwrap(); + let session = Session::create( + &pool, + &CreateSession { + executor: Some("CLAUDE_CODE".to_string()), + name: None, + }, + Uuid::new_v4(), + workspace_id, + ) + .await + .unwrap(); + sqlx::query( + "INSERT INTO claude_session_links + (claude_session_id, session_id, workspace_id, cwd, bound_via) + VALUES (?, ?, ?, '/home/dev/project', 'executor')", + ) + .bind(SID) + .bind(session.id) + .bind(workspace_id) + .execute(&pool) + .await + .unwrap(); + + let pruned = CliNativeFile::prune_unreachable(&pool, 14, 25) + .await + .unwrap(); + + assert!(pruned.is_empty(), "a linked transcript is still renderable"); + assert_eq!(remaining_records(&pool).await, 3); + } + + #[tokio::test] + async fn prune_keeps_unlinked_transcripts_a_live_session_can_replay() { + let pool = pool().await; + let file = CliNativeFile::register(&pool, ®istration(120, Some(9))) + .await + .unwrap(); + add_records(&pool, file.id, 2).await; + backdate(&pool, file.id).await; + + let workspace_id = Uuid::new_v4(); + Workspace::create( + &pool, + &CreateWorkspace { + branch: "main".to_string(), + name: Some("replay".to_string()), + }, + workspace_id, + ) + .await + .unwrap(); + let session = Session::create( + &pool, + &CreateSession { + executor: Some("CLAUDE_CODE".to_string()), + name: None, + }, + Uuid::new_v4(), + workspace_id, + ) + .await + .unwrap(); + // The ownership link is gone but the publication log still points here. + sqlx::query( + "INSERT INTO cli_ingest_outbox (session_id, seq, file_id, line_seq) + VALUES (?, 1, ?, 0)", + ) + .bind(session.id) + .bind(file.id) + .execute(&pool) + .await + .unwrap(); + + let pruned = CliNativeFile::prune_unreachable(&pool, 14, 25) + .await + .unwrap(); + + assert!( + pruned.is_empty(), + "a live session's feed can still replay this" + ); + assert_eq!(remaining_records(&pool).await, 2); + } + + #[tokio::test] + async fn prune_bounds_each_sweep_to_the_file_limit() { + let pool = pool().await; + for index in 0..3 { + let file = CliNativeFile::register( + &pool, + &RegisterCliNativeFile { + file_name: match index { + 0 => "a.jsonl", + 1 => "b.jsonl", + _ => "c.jsonl", + }, + ..registration(120, Some(9)) + }, + ) + .await + .unwrap(); + add_records(&pool, file.id, 1).await; + backdate(&pool, file.id).await; + } + + let first = CliNativeFile::prune_unreachable(&pool, 14, 2) + .await + .unwrap(); + assert_eq!(first.files, 2); + + let second = CliNativeFile::prune_unreachable(&pool, 14, 2) + .await + .unwrap(); + assert_eq!(second.files, 1); + + assert!( + CliNativeFile::prune_unreachable(&pool, 14, 2) + .await + .unwrap() + .is_empty() + ); + assert_eq!(remaining_records(&pool).await, 0); + } } diff --git a/crates/services/src/services/claude_transcript_ingest.rs b/crates/services/src/services/claude_transcript_ingest.rs index b869b59b689..ad933751a5a 100644 --- a/crates/services/src/services/claude_transcript_ingest.rs +++ b/crates/services/src/services/claude_transcript_ingest.rs @@ -72,6 +72,35 @@ const REGISTRY_RECONCILE_INTERVAL: Duration = Duration::from_secs(30); const OUTBOX_SAFETY_POLL_INTERVAL: Duration = Duration::from_secs(30); const IMPORT_BATCH_LINE_LIMIT: usize = 256; +/// Days an unreachable transcript is kept before retention removes it. `0` +/// disables retention and lets the store grow without bound. +const RETENTION_ENV: &str = "VIBE_KANBAN_CLI_TRANSCRIPT_RETENTION_DAYS"; +const DEFAULT_RETENTION_DAYS: u32 = 14; +const RETENTION_SWEEP_INTERVAL: Duration = Duration::from_secs(6 * 60 * 60); +/// Retention competes for the same write lock as ingest, so the first sweep +/// waits for start-up scanning to settle. +const RETENTION_STARTUP_DELAY: Duration = Duration::from_secs(5 * 60); +/// Files per sweep. Each is one transaction, so this bounds how long retention +/// can hold the write lock at a time; a backlog drains over several sweeps. +const RETENTION_FILES_PER_SWEEP: i64 = 25; + +fn retention_days() -> Option { + match std::env::var(RETENTION_ENV) { + Err(_) => Some(DEFAULT_RETENTION_DAYS), + Ok(raw) => match raw.trim().parse::() { + Ok(0) => None, + Ok(days) => Some(days), + Err(_) => { + tracing::warn!( + "{RETENTION_ENV}={raw:?} is not a whole number of days; \ + using {DEFAULT_RETENTION_DAYS}" + ); + Some(DEFAULT_RETENTION_DAYS) + } + }, + } +} + #[derive(Debug, Clone)] pub(crate) struct NativeLinkPersisted { pub execution_process_id: Uuid, @@ -237,9 +266,45 @@ impl ClaudeTranscriptIngest { .run_native_link_invalidation(native_link_updates, shutdown.child_token()), ); tokio::spawn(service.clone().run_registry(shutdown.child_token(), true)); + tokio::spawn(service.clone().run_retention(shutdown.child_token())); Some(service) } + /// Periodically reclaims transcripts no session can reach. Without this the + /// raw-record table grows for the lifetime of the install, and a larger + /// database makes every commit — and so every write-lock hold — slower. + async fn run_retention(self: Arc, shutdown: CancellationToken) { + let Some(days) = retention_days() else { + tracing::info!("CLI transcript retention disabled by {RETENTION_ENV}"); + return; + }; + let mut interval = tokio::time::interval_at( + tokio::time::Instant::now() + RETENTION_STARTUP_DELAY, + RETENTION_SWEEP_INTERVAL, + ); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + loop { + tokio::select! { + _ = shutdown.cancelled() => return, + _ = interval.tick() => {} + } + match CliNativeFile::prune_unreachable(&self.db.pool, days, RETENTION_FILES_PER_SWEEP) + .await + { + Ok(pruned) if pruned.is_empty() => {} + Ok(pruned) => tracing::info!( + files = pruned.files, + records = pruned.records, + retention_days = days, + "pruned unreachable CLI transcripts" + ), + Err(err) => { + tracing::warn!(%err, "CLI transcript retention sweep failed") + } + } + } + } + fn new_with_probe( db: DBService, projects_dir: PathBuf,