mirror of
https://github.com/openai/codex.git
synced 2026-10-20 08:19:50 +00:00
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
This commit is contained in:
@@ -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<String, String>,
|
||||
}
|
||||
|
||||
/// Runs `codex --cd <workspace> doctor --json` and returns a best-effort
|
||||
/// Runs `codex --cd <workspace> 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!(
|
||||
|
||||
+49
-36
@@ -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<String> {
|
||||
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<String>,
|
||||
integrity_failures: &mut Vec<String>,
|
||||
label: &str,
|
||||
path: &Path,
|
||||
) {
|
||||
deadline: Option<Instant>,
|
||||
) -> 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
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -20,7 +20,9 @@ pub struct ThreadStateAuditRow {
|
||||
pub async fn read_thread_state_audit_rows(
|
||||
sqlite: &SqliteConfig,
|
||||
) -> Result<Vec<ThreadStateAuditRow>> {
|
||||
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
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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");
|
||||
|
||||
@@ -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<String>),
|
||||
TimedOut(Vec<String>),
|
||||
}
|
||||
|
||||
/// Run SQLite's built-in integrity check against an existing database file.
|
||||
pub async fn sqlite_integrity_check(
|
||||
sqlite: &SqliteConfig,
|
||||
path: &Path,
|
||||
) -> anyhow::Result<Vec<String>> {
|
||||
let pool = sqlite.open_read_only_pool(path).await?;
|
||||
let rows = sqlx::query_scalar::<_, String>("PRAGMA integrity_check")
|
||||
.fetch_all(&pool)
|
||||
deadline: Option<Instant>,
|
||||
) -> anyhow::Result<SqliteIntegrityCheck> {
|
||||
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::<Sqlite>::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::<String>::new();
|
||||
// Keep corruption rows even when a later step interrupts the scan.
|
||||
let result = sqlx::query::<Sqlite>("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::<i32>().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;
|
||||
}
|
||||
|
||||
|
||||
@@ -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<SqlitePool, Error> {
|
||||
let options = SqliteConnectOptions::new()
|
||||
pub async fn open_read_only_pool(
|
||||
&self,
|
||||
path: &Path,
|
||||
busy_timeout: Option<Duration>,
|
||||
) -> Result<SqlitePool, Error> {
|
||||
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)
|
||||
|
||||
Reference in New Issue
Block a user