diff --git a/Cargo.lock b/Cargo.lock index 53e5cb595a..5e9bb43be0 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/.sqlx/query-1e58cbf7abbc2dda49a2f8127a9752f1a6f39f3daee356fb533cdd2795d8f9fa.json b/crates/db/.sqlx/query-1e58cbf7abbc2dda49a2f8127a9752f1a6f39f3daee356fb533cdd2795d8f9fa.json new file mode 100644 index 0000000000..b086d87215 --- /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 0000000000..d141e49c2b --- /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 0000000000..60cc14eb67 --- /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 0000000000..c94f283ce3 --- /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/Cargo.toml b/crates/db/Cargo.toml index e5d62c7dd0..ca173a0e9e 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 bc5b090e36..e9fc23533e 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() diff --git a/crates/db/src/models/cli_native_file.rs b/crates/db/src/models/cli_native_file.rs index c596b58c59..3e26e7f7ab 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/server/src/error.rs b/crates/server/src/error.rs index 77d8d72fa1..71c1e97e14 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 + ))); + } } diff --git a/crates/services/src/services/claude_transcript_ingest.rs b/crates/services/src/services/claude_transcript_ingest.rs index b869b59b68..ad933751a5 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,