From 32ce703a80ef8828a07d452e2af5630cc0825aae Mon Sep 17 00:00:00 2001 From: chess Date: Tue, 25 Aug 2026 19:53:47 +0000 Subject: [PATCH] Bound SQLite integrity checks for diagnostic attachments (#40688) ## Why Diagnostic uploads skip the doctor attachment if report generation exceeds its overall timeout. SQLite integrity scans can be delayed by large databases or lock contention. ## What changed - Give each database integrity scan a one-second deadline when collecting a diagnostic attachment. - Interrupt scans and lock waits at the deadline, preserve any corruption rows already returned, and report an incomplete clean scan as a warning. - Keep direct `codex doctor` integrity checks unbounded. ## Testing - Cover interrupted scans, lock waits, retries without a deadline, and corruption results returned before interruption. GitOrigin-RevId: 4817fb459ac53953669a9ba5f03397a6fc82c72e --- .../feedback_doctor_report.rs | 10 +- codex-rs/cli/src/doctor.rs | 85 ++++++----- codex-rs/state/src/audit.rs | 4 +- codex-rs/state/src/lib.rs | 1 + codex-rs/state/src/migrations_tests.rs | 2 +- codex-rs/state/src/runtime.rs | 134 ++++++++++++++++-- codex-rs/state/src/sqlite.rs | 11 +- 7 files changed, 194 insertions(+), 53 deletions(-) diff --git a/codex-rs/app-server/src/request_processors/feedback_doctor_report.rs b/codex-rs/app-server/src/request_processors/feedback_doctor_report.rs index 9350357e57..d039cad99d 100644 --- a/codex-rs/app-server/src/request_processors/feedback_doctor_report.rs +++ b/codex-rs/app-server/src/request_processors/feedback_doctor_report.rs @@ -2,10 +2,10 @@ //! //! Feedback upload should never depend on doctor succeeding. This module runs //! the configured Codex executable as a subprocess, accepts only valid JSON from -//! `codex doctor --json`, derives a small set of Sentry tags, and otherwise +//! `codex doctor --json --feedback`, derives a small set of Sentry tags, and otherwise //! skips the attachment with a warning. Keeping the report generation out of the //! app-server process avoids sharing doctor internals across crates while still -//! attaching exactly the same JSON a user could copy from the CLI. +//! using the CLI's JSON report format with bounded database scans. use std::collections::BTreeMap; use std::path::Path; @@ -31,7 +31,7 @@ pub(crate) struct DoctorFeedbackReport { pub(crate) tags: BTreeMap, } -/// Runs `codex --cd doctor --json` and returns a best-effort +/// Runs `codex --cd doctor --json --feedback` and returns a best-effort /// feedback attachment. /// /// Failure to spawn Codex, finish before the timeout, or parse JSON means the @@ -110,6 +110,7 @@ fn doctor_command(executable: &Path, cwd: &Path, codex_home: &Path) -> Command { .arg(cwd) .arg("doctor") .arg("--json") + .arg("--feedback") .current_dir(codex_home) .env("CODEX_HOME", codex_home); command @@ -211,7 +212,8 @@ mod tests { "--cd".as_ref(), workspace.as_os_str(), "doctor".as_ref(), - "--json".as_ref() + "--json".as_ref(), + "--feedback".as_ref() ] ); assert_eq!( diff --git a/codex-rs/cli/src/doctor.rs b/codex-rs/cli/src/doctor.rs index 3fba0e2fdf..b43e3c3e5c 100644 --- a/codex-rs/cli/src/doctor.rs +++ b/codex-rs/cli/src/doctor.rs @@ -162,7 +162,7 @@ const NARROW_TERMINAL_ROWS: u16 = 24; /// Options for building a local Codex diagnostic report. /// -/// The command always runs the full bounded diagnostic set. Human output includes +/// The command always runs the full diagnostic set. Human output includes /// detailed diagnostics by default; --summary keeps the terminal output compact. #[derive(Debug, Parser)] pub struct DoctorCommand { @@ -170,6 +170,10 @@ pub struct DoctorCommand { #[arg(long, default_value_t = false)] json: bool, + /// Limit database integrity scans when collecting a feedback attachment. + #[arg(long, hide = true, default_value_t = false)] + feedback: bool, + /// Only show grouped check rows and the final count summary. #[arg(long, default_value_t = false)] summary: bool, @@ -452,7 +456,7 @@ async fn build_report( terminal_title_check(config) }) }, - run_async_check("state", progress.clone(), state_check(config)), + run_async_check("state", progress.clone(), state_check(config, command)), run_async_check( "thread inventory", progress.clone(), @@ -2179,35 +2183,36 @@ fn non_empty_trimmed(value: String) -> Option { if value.is_empty() { None } else { Some(value) } } -async fn state_check(config: &Config) -> DoctorCheck { +async fn state_check(config: &Config, command: &DoctorCommand) -> DoctorCheck { let mut details = Vec::new(); path_readiness(&mut details, "CODEX_HOME", &config.codex_home); path_readiness(&mut details, "log dir", &config.log_dir); path_readiness(&mut details, "sqlite home", config.sqlite_config().home()); - let mut integrity_failures = Vec::new(); + let mut status = CheckStatus::Ok; for db in config.sqlite_config().runtime_db_paths() { path_readiness(&mut details, db.label, &db.path); - sqlite_integrity_detail( - config.sqlite_config(), - &mut details, - &mut integrity_failures, - db.label, - &db.path, - ) - .await; + // Feedback collection gives each database its own budget; direct runs scan fully. + let deadline = command + .feedback + .then(|| Instant::now() + Duration::from_secs(1)); + status = status.max( + sqlite_integrity_detail( + config.sqlite_config(), + &mut details, + db.label, + &db.path, + deadline, + ) + .await, + ); } rollout_stats_details(&mut details, &config.codex_home); standalone_release_cache_details(&mut details); - let status = if integrity_failures.is_empty() { - CheckStatus::Ok - } else { - CheckStatus::Fail - }; - let summary = if status == CheckStatus::Ok { - "state paths and databases are inspectable" - } else { - "state database integrity check failed" + let summary = match status { + CheckStatus::Ok => "state paths and databases are inspectable", + CheckStatus::Warning => "some database integrity checks exceeded their time limit", + CheckStatus::Fail => "state database integrity check failed", }; let mut check = DoctorCheck::new("state.paths", "state", status, summary).details(details); if status == CheckStatus::Fail { @@ -2221,29 +2226,37 @@ async fn state_check(config: &Config) -> DoctorCheck { async fn sqlite_integrity_detail( sqlite: &codex_state::SqliteConfig, details: &mut Vec, - integrity_failures: &mut Vec, label: &str, path: &Path, -) { + deadline: Option, +) -> CheckStatus { if !path.is_file() { details.push(format!("{label} integrity: skipped (missing)")); - return; + return CheckStatus::Ok; } - match codex_state::sqlite_integrity_check(sqlite, path).await { - Ok(rows) if rows.iter().all(|row| row == "ok") => { - details.push(format!("{label} integrity: ok")); - } - Ok(rows) => { - let message = format!("{label} integrity: {}", rows.join("; ")); - integrity_failures.push(message.clone()); - details.push(message); - } + let (rows, timed_out) = match codex_state::sqlite_integrity_check(sqlite, path, deadline).await + { + Ok(codex_state::SqliteIntegrityCheck::Complete(rows)) => (rows, false), + Ok(codex_state::SqliteIntegrityCheck::TimedOut(rows)) => (rows, true), Err(err) => { - let message = format!("{label} integrity: {err}"); - integrity_failures.push(message.clone()); - details.push(message); + details.push(format!("{label} integrity: {err}")); + return CheckStatus::Fail; } + }; + if timed_out { + details.push(format!( + "{label} integrity: incomplete (1 second time limit)" + )); + } + if rows.iter().any(|row| row != "ok") { + details.push(format!("{label} integrity: {}", rows.join("; "))); + CheckStatus::Fail + } else if timed_out { + CheckStatus::Warning + } else { + details.push(format!("{label} integrity: ok")); + CheckStatus::Ok } } diff --git a/codex-rs/state/src/audit.rs b/codex-rs/state/src/audit.rs index 6772b916e7..57ef9f0673 100644 --- a/codex-rs/state/src/audit.rs +++ b/codex-rs/state/src/audit.rs @@ -20,7 +20,9 @@ pub struct ThreadStateAuditRow { pub async fn read_thread_state_audit_rows( sqlite: &SqliteConfig, ) -> Result> { - let pool = sqlite.open_read_only_pool(&sqlite.state_db_path()).await?; + let pool = sqlite + .open_read_only_pool(&sqlite.state_db_path(), /*busy_timeout*/ None) + .await?; let rows = sqlx::query( r#" SELECT id, rollout_path, archived, source, model_provider diff --git a/codex-rs/state/src/lib.rs b/codex-rs/state/src/lib.rs index 8176aed59c..b2c61b4860 100644 --- a/codex-rs/state/src/lib.rs +++ b/codex-rs/state/src/lib.rs @@ -78,6 +78,7 @@ pub use runtime::GoalUpdate; pub use runtime::MemoryStore; pub use runtime::RemoteControlEnrollmentRecord; pub use runtime::RuntimeDbBackup; +pub use runtime::SqliteIntegrityCheck; pub use runtime::SqliteQueueStore; pub use runtime::ThreadFilterOptions; pub use runtime::backup_runtime_db_for_fresh_start; diff --git a/codex-rs/state/src/migrations_tests.rs b/codex-rs/state/src/migrations_tests.rs index c80e851d18..99de1f6b00 100644 --- a/codex-rs/state/src/migrations_tests.rs +++ b/codex-rs/state/src/migrations_tests.rs @@ -843,7 +843,7 @@ async fn repair_recency_migration_succeeds_while_another_connection_holds_writer .await .expect("current migrations should apply"); let read_pool = sqlite - .open_read_only_pool(&state_path) + .open_read_only_pool(&state_path, /*busy_timeout*/ None) .await .expect("read-only pool should open"); let mut write_connection = pool.acquire().await.expect("write connection should open"); diff --git a/codex-rs/state/src/runtime.rs b/codex-rs/state/src/runtime.rs index 95765f98f0..50ccf8e43e 100644 --- a/codex-rs/state/src/runtime.rs +++ b/codex-rs/state/src/runtime.rs @@ -353,21 +353,74 @@ ON CONFLICT(id) DO NOTHING Ok(()) } +/// Integrity-check rows, including those emitted before interruption. +#[derive(Debug, Eq, PartialEq)] +pub enum SqliteIntegrityCheck { + Complete(Vec), + TimedOut(Vec), +} + /// Run SQLite's built-in integrity check against an existing database file. pub async fn sqlite_integrity_check( sqlite: &SqliteConfig, path: &Path, -) -> anyhow::Result> { - let pool = sqlite.open_read_only_pool(path).await?; - let rows = sqlx::query_scalar::<_, String>("PRAGMA integrity_check") - .fetch_all(&pool) + deadline: Option, +) -> anyhow::Result { + let pool = sqlite + .open_read_only_pool( + path, + deadline.map(|deadline| deadline.saturating_duration_since(Instant::now())), + ) .await?; + let mut connection = pool.acquire().await?; + if let Some(deadline) = deadline { + // Lock waits do not invoke the progress handler; share the remaining budget. + QueryBuilder::::new("PRAGMA busy_timeout = ") + .push( + deadline + .saturating_duration_since(Instant::now()) + .as_millis(), + ) + .build() + .execute(&mut *connection) + .await?; + // Interrupt SQLite itself: dropping a timed-out future leaves its worker scanning. + connection + .lock_handle() + .await? + .set_progress_handler(/*num_ops*/ 1_000, move || Instant::now() < deadline); + } + let mut rows = Vec::::new(); + // Keep corruption rows even when a later step interrupts the scan. + let result = sqlx::query::("PRAGMA integrity_check") + .try_map(|row| { + rows.push(row.try_get(/*index*/ 0)?); + Ok(()) + }) + .fetch_all(&mut *connection) + .await; + drop(connection); pool.close().await; - Ok(rows) + match result { + Ok(_) => Ok(SqliteIntegrityCheck::Complete(rows)), + Err(sqlx::Error::Database(error)) + if deadline.is_some() + && error.code().is_some_and(|code| { + matches!( + code.parse::().ok().map(|code| code & 0xff), + Some(libsqlite3_sys::SQLITE_INTERRUPT | libsqlite3_sys::SQLITE_BUSY) + ) + }) => + { + Ok(SqliteIntegrityCheck::TimedOut(rows)) + } + Err(error) => Err(error.into()), + } } #[cfg(test)] mod tests { + use super::SqliteIntegrityCheck; use super::StateRuntime; use super::runtime_state_migrator; use super::sqlite_integrity_check; @@ -386,6 +439,8 @@ mod tests { use std::path::Path; use std::sync::Mutex; use std::sync::atomic::Ordering; + use std::time::Duration; + use std::time::Instant; #[derive(Default)] struct TestTelemetry { @@ -446,7 +501,7 @@ mod tests { } #[tokio::test] - async fn sqlite_integrity_check_reports_ok_for_valid_db() { + async fn sqlite_integrity_check_can_be_interrupted_and_retried() { let codex_home = unique_temp_dir(); tokio::fs::create_dir_all(&codex_home) .await @@ -457,17 +512,78 @@ mod tests { .open_read_write_pool(&path) .await .expect("open sqlite db"); - sqlx::query("CREATE TABLE sample (id INTEGER PRIMARY KEY)") + sqlx::query("CREATE TABLE sample (id INTEGER PRIMARY KEY, value INTEGER, other INTEGER)") .execute(&pool) .await .expect("create sample table"); + sqlx::query("CREATE UNIQUE INDEX sample_value ON sample(value)") + .execute(&pool) + .await + .expect("create sample index"); + sqlx::query( + "WITH RECURSIVE rows(id) AS (SELECT 1 UNION ALL SELECT id + 1 FROM rows WHERE id < 2048) INSERT INTO sample SELECT id, id, CASE WHEN id = 1 THEN 0 ELSE id END FROM rows", + ) + .execute(&pool) + .await + .expect("populate enough rows to invoke the progress handler"); pool.close().await; - let result = sqlite_integrity_check(&sqlite, &path) + assert_eq!( + sqlite_integrity_check(&sqlite, &path, Some(Instant::now())) + .await + .expect("interrupt integrity check"), + SqliteIntegrityCheck::TimedOut(Vec::new()), + ); + + let result = sqlite_integrity_check(&sqlite, &path, /*deadline*/ None) .await .expect("integrity check should run"); - assert_eq!(result, vec!["ok".to_string()]); + assert_eq!( + result, + SqliteIntegrityCheck::Complete(vec!["ok".to_string()]) + ); + + let pool = sqlite + .open_read_write_pool(&path) + .await + .expect("reopen sqlite db"); + let mut connection = pool.acquire().await.expect("acquire writer"); + // WAL readers do not wait on this writer, so use a rollback journal. + sqlx::query("PRAGMA journal_mode=DELETE; BEGIN EXCLUSIVE") + .execute(&mut *connection) + .await + .expect("hold an exclusive lock"); + let started = Instant::now(); + let result = + sqlite_integrity_check(&sqlite, &path, Some(started + Duration::from_millis(50))) + .await + .expect("interrupt lock wait"); + let elapsed = started.elapsed(); + sqlx::query("ROLLBACK") + .execute(&mut *connection) + .await + .expect("release lock"); + assert_eq!(result, SqliteIntegrityCheck::TimedOut(Vec::new())); + assert!(elapsed < Duration::from_secs(1), "lock wait: {elapsed:?}"); + + // Misdescribe one index entry so SQLite emits an error before its first progress callback. + sqlx::query( + "PRAGMA writable_schema=ON; UPDATE sqlite_schema SET sql='CREATE UNIQUE INDEX sample_value ON sample(other)' WHERE name='sample_value'; PRAGMA writable_schema=OFF", + ) + .execute(&mut *connection) + .await + .expect("introduce an index mismatch"); + drop(connection); + pool.close().await; + assert_eq!( + sqlite_integrity_check(&sqlite, &path, Some(Instant::now())) + .await + .expect("retain corruption before interruption"), + SqliteIntegrityCheck::TimedOut(vec![ + "row 1 missing from index sample_value".to_string() + ]), + ); let _ = tokio::fs::remove_dir_all(codex_home).await; } diff --git a/codex-rs/state/src/sqlite.rs b/codex-rs/state/src/sqlite.rs index 4bb65c6e35..536d56befa 100644 --- a/codex-rs/state/src/sqlite.rs +++ b/codex-rs/state/src/sqlite.rs @@ -291,12 +291,19 @@ impl SqliteConfig { } /// Open an existing Codex SQLite database without creating or modifying it. - pub async fn open_read_only_pool(&self, path: &Path) -> Result { - let options = SqliteConnectOptions::new() + pub async fn open_read_only_pool( + &self, + path: &Path, + busy_timeout: Option, + ) -> Result { + let mut options = SqliteConnectOptions::new() .filename(path) .create_if_missing(false) .read_only(true) .log_statements(LevelFilter::Off); + if let Some(busy_timeout) = busy_timeout { + options = options.busy_timeout(busy_timeout); + } SqlitePoolOptions::new() .max_connections(1) .connect_with(options)