From 3620b2caf8afecfcedd5c5c9520c0d239c880f26 Mon Sep 17 00:00:00 2001 From: Dmitriy Kovalenko Date: Wed, 30 Sep 2026 18:12:51 +0000 Subject: [PATCH] Detect SQLite corruption during startup and preserve recovery backups (#49701) ## Why A database can open successfully while containing corruption that startup previously missed. Detecting it before runtime initialization completes allows recoverable databases to be rebuilt while preserving the damaged files. ## What changed - Run `PRAGMA quick_check(1)` when opening writable databases, with a 100 ms scan budget. Share attempts across cloned `SqliteConfig` instances by file identity so repeated opens skip validation and replacement files are checked again. - Back up and rebuild runtime databases that support recovery. Report corruption without rebuilding thread history or other databases without an automatic recovery policy. Treat interrupted or locked checks as incomplete rather than confirmed corruption. - Collect startup backups across recovery retries and include their locations in the app server warning. Explain that saved conversations can restore the thread list and history, while some database-only metadata may be unavailable. - Record confirmed corruption in `codex.sqlite.corruption.count`. ## Testing Add tests for recovery policies, backup preservation, validation caching and concurrency, cancelled and interrupted checks, and locked databases. Add app server coverage for recovery warnings and saved conversation restoration, a TUI warning snapshot, and message board post preservation. GitOrigin-RevId: 43e26328888e0301def642d69bd324acee867566 --- MODULE.bazel.lock | 1 + codex-rs/Cargo.lock | 10 + codex-rs/Cargo.toml | 1 + codex-rs/app-server/src/lib.rs | 119 ++-- codex-rs/app-server/tests/suite/v2/mod.rs | 1 + .../tests/suite/v2/sqlite_recovery.rs | 196 ++++++ .../src/local/lifecycle.rs | 1 - .../agent-message-board/tests/local_board.rs | 61 ++ codex-rs/state/Cargo.toml | 1 + codex-rs/state/src/lib.rs | 3 + codex-rs/state/src/runtime.rs | 1 + codex-rs/state/src/runtime/recovery.rs | 42 +- codex-rs/state/src/runtime/recovery_tests.rs | 6 +- codex-rs/state/src/sqlite.rs | 184 ++++-- codex-rs/state/src/sqlite/validation.rs | 80 +++ codex-rs/state/src/sqlite/validation_tests.rs | 586 ++++++++++++++++++ codex-rs/state/src/sqlite_tests.rs | 3 +- codex-rs/state/src/telemetry.rs | 17 +- .../tui/src/chatwidget/tests/app_server.rs | 22 + ...__app_server__sqlite_recovery_warning.snap | 10 + 20 files changed, 1227 insertions(+), 118 deletions(-) create mode 100644 codex-rs/app-server/tests/suite/v2/sqlite_recovery.rs create mode 100644 codex-rs/state/src/sqlite/validation.rs create mode 100644 codex-rs/state/src/sqlite/validation_tests.rs create mode 100644 codex-rs/tui/src/chatwidget/tests/snapshots/codex_tui__chatwidget__tests__app_server__sqlite_recovery_warning.snap diff --git a/MODULE.bazel.lock b/MODULE.bazel.lock index b0077619b6..a34e326095 100644 --- a/MODULE.bazel.lock +++ b/MODULE.bazel.lock @@ -1015,6 +1015,7 @@ "fdeflate_0.3.7": "{\"dependencies\":[{\"kind\":\"dev\",\"name\":\"miniz_oxide\",\"req\":\"^0.7.1\"},{\"kind\":\"dev\",\"name\":\"rand\",\"req\":\"^0.8.5\"},{\"name\":\"simd-adler32\",\"req\":\"^0.3.4\"}],\"features\":{}}", "ff_0.13.1": "{\"dependencies\":[{\"default_features\":false,\"name\":\"bitvec\",\"optional\":true,\"req\":\"^1\"},{\"kind\":\"dev\",\"name\":\"blake2b_simd\",\"req\":\"^1\"},{\"default_features\":false,\"name\":\"byteorder\",\"optional\":true,\"req\":\"^1\"},{\"name\":\"ff_derive\",\"optional\":true,\"req\":\"^0.13.1\"},{\"kind\":\"dev\",\"name\":\"rand\",\"req\":\"^0.8\"},{\"default_features\":false,\"name\":\"rand_core\",\"req\":\"^0.6\"},{\"default_features\":false,\"features\":[\"i128\"],\"name\":\"subtle\",\"req\":\"^2.2.1\"}],\"features\":{\"alloc\":[],\"bits\":[\"bitvec\"],\"default\":[\"bits\",\"std\"],\"derive\":[\"byteorder\",\"ff_derive\"],\"derive_bits\":[\"bits\",\"ff_derive/bits\"],\"std\":[\"alloc\"]}}", "fiat-crypto_0.2.9": "{\"dependencies\":[],\"features\":{\"default\":[\"std\"],\"std\":[]}}", + "file-id_0.2.3": "{\"dependencies\":[{\"features\":[\"derive\"],\"name\":\"serde\",\"optional\":true,\"req\":\"^1.0.89\"},{\"kind\":\"dev\",\"name\":\"tempfile\",\"req\":\"^3.10.0\"},{\"features\":[\"Win32_Storage_FileSystem\",\"Win32_Foundation\"],\"name\":\"windows-sys\",\"req\":\"^0.60.1\",\"target\":\"cfg(windows)\"}],\"features\":{}}", "filedescriptor_0.8.3": "{\"dependencies\":[{\"name\":\"libc\",\"req\":\"^0.2\"},{\"name\":\"thiserror\",\"req\":\"^1.0\"},{\"features\":[\"winuser\",\"handleapi\",\"fileapi\",\"namedpipeapi\",\"processthreadsapi\",\"winsock2\",\"processenv\"],\"name\":\"winapi\",\"req\":\"^0.3\",\"target\":\"cfg(windows)\"}],\"features\":{}}", "filetime_0.2.27": "{\"dependencies\":[{\"name\":\"cfg-if\",\"req\":\"^1.0.0\"},{\"name\":\"libc\",\"req\":\"^0.2.27\",\"target\":\"cfg(unix)\"},{\"name\":\"libredox\",\"req\":\"^0.1.0\",\"target\":\"cfg(target_os = \\\"redox\\\")\"},{\"kind\":\"dev\",\"name\":\"tempfile\",\"req\":\"^3\"}],\"features\":{}}", "filetime_0.2.29": "{\"dependencies\":[{\"name\":\"cfg-if\",\"req\":\"^1.0.0\"},{\"name\":\"libc\",\"req\":\"^0.2.27\",\"target\":\"cfg(unix)\"},{\"kind\":\"dev\",\"name\":\"tempfile\",\"req\":\"^3\"}],\"features\":{}}", diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index f799dfda14..ae76b2d1a7 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -4620,6 +4620,7 @@ dependencies = [ "codex-history", "codex-protocol", "codex-utils-absolute-path", + "file-id", "libsqlite3-sys", "log", "pretty_assertions", @@ -6986,6 +6987,15 @@ version = "0.2.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "28dea519a9695b9977216879a3ebfddf92f1c08c05d984f8996aecd6ecdc811d" +[[package]] +name = "file-id" +version = "0.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e1fc6a637b6dc58414714eddd9170ff187ecb0933d4c7024d1abbd23a3cc26e9" +dependencies = [ + "windows-sys 0.60.2", +] + [[package]] name = "filedescriptor" version = "0.8.3" diff --git a/codex-rs/Cargo.toml b/codex-rs/Cargo.toml index fc1becc762..717891fc4b 100644 --- a/codex-rs/Cargo.toml +++ b/codex-rs/Cargo.toml @@ -376,6 +376,7 @@ dunce = "1.0.4" ed25519-dalek = { version = "2.2.0", features = ["pkcs8"] } encoding_rs = "0.8.35" eventsource-stream = "0.2.3" +file-id = "0.2.3" flate2 = "1.1.8" futures = { version = "0.3", default-features = false } gethostname = "1.1.0" diff --git a/codex-rs/app-server/src/lib.rs b/codex-rs/app-server/src/lib.rs index d1a622ca8c..bf1bb93285 100644 --- a/codex-rs/app-server/src/lib.rs +++ b/codex-rs/app-server/src/lib.rs @@ -83,7 +83,7 @@ use tracing_subscriber::layer::SubscriberExt; use tracing_subscriber::registry::Registry; use tracing_subscriber::util::SubscriberInitExt; -const SQLITE_RECOVERY_CONFIG_WARNING_SUMMARY: &str = "Codex rebuilt its local database."; +const SQLITE_RECOVERY_CONFIG_WARNING_SUMMARY: &str = "Codex rebuilt its local database"; fn is_unsupported_untrusted_approval_policy_error(err: &std::io::Error) -> bool { err.get_ref().is_some_and( @@ -1379,74 +1379,71 @@ struct StateDbInitResult { async fn init_sqlite_state_db_with_fresh_start_on_corruption( config: &Config, ) -> anyhow::Result { - let mut attempted_backups = HashSet::new(); - let mut recovered_databases = Vec::new(); - loop { - let err = match rollout_state_db::try_init(config).await { - Ok(state_db) => { - let recovery_notice = sqlite_recovery_notice(&recovered_databases); - if recovery_notice.is_some() { - emit_state_db_backup_warning(SQLITE_RECOVERY_CONFIG_WARNING_SUMMARY); - for recovered_database in &recovered_databases { - emit_state_db_backup_warning(&format!( - "Database path: {}", - recovered_database.database_path - )); - emit_state_db_backup_warning(&format!( - "Backup folder: {}", - recovered_database.backup_folder - )); - } - } - return Ok(StateDbInitResult { - state_db: Some(state_db), - recovery_notice, - }); + let (result, backups) = codex_state::collect_runtime_db_backups(async { + let mut attempted_backups = HashSet::new(); + loop { + let err = match rollout_state_db::try_init(config).await { + Ok(state_db) => return Ok(state_db), + Err(err) => err, + }; + let database_path = codex_state::runtime_db_path_for_corruption_error(&err) + .unwrap_or_else(|| config.sqlite_config().state_db_path()); + if !codex_state::is_sqlite_corruption_error(&err) + && !sqlite_home_is_blocking_file(database_path.as_path()) + { + return Err(err); } - Err(err) => err, - }; - let database_path = codex_state::runtime_db_path_for_corruption_error(&err) - .unwrap_or_else(|| config.sqlite_config().state_db_path()); - if !codex_state::is_sqlite_corruption_error(&err) - && !sqlite_home_is_blocking_file(database_path.as_path()) - { - return Err(err); - } - if !attempted_backups.insert(database_path.clone()) { - return Err(anyhow::anyhow!( - "failed to initialize sqlite state runtime after moving damaged database file into a backup folder: {err}" - )); - } + if !attempted_backups.insert(database_path.clone()) { + return Err(anyhow::anyhow!( + "failed to initialize sqlite state runtime after moving damaged database file into a backup folder: {err}" + )); + } - let original_error = err.to_string(); - emit_state_db_backup_warning(&format!( - "Codex local database at {} appears damaged. Moving it into a backup folder so the app server can rebuild it from saved data.", - database_path.display() - )); - let backups = codex_state::backup_runtime_db_for_fresh_start(database_path.as_path()) - .await - .map_err(|backup_err| { - anyhow::anyhow!( - "failed to move damaged sqlite state database files into a backup folder: {backup_err}; original error: {original_error}" - ) - })?; - for backup in &backups { + let original_error = err.to_string(); emit_state_db_backup_warning(&format!( - "Moved damaged Codex local database file {} to {}", - backup.original_path.display(), - backup.backup_path.display() + "Codex local database at {} appears damaged. Moving it into a backup folder so the app server can rebuild it from saved data.", + database_path.display() )); + let backups = codex_state::backup_runtime_db_for_fresh_start(database_path.as_path()) + .await + .map_err(|backup_err| { + anyhow::anyhow!( + "failed to move damaged sqlite state database files into a backup folder: {backup_err}; original error: {original_error}" + ) + })?; + for backup in &backups { + emit_state_db_backup_warning(&format!( + "Moved damaged Codex local database file {} to {}", + backup.original_path.display(), + backup.backup_path.display() + )); + } } - if let Some(first_backup) = backups.first() - && let Some(backup_folder) = first_backup.backup_path.parent() + }) + .await; + let state_db = result?; + let mut recovered_databases = Vec::new(); + let mut backup_folders = HashSet::new(); + for backup in backups { + if let Some(folder) = backup.backup_path.parent() + && backup_folders.insert(folder.to_path_buf()) { recovered_databases.push(RecoveredSqliteDatabase { - database_path: first_backup.original_path.display().to_string(), - backup_folder: backup_folder.display().to_string(), + database_path: backup.original_path.display().to_string(), + backup_folder: folder.display().to_string(), }); } } + let recovery_notice = sqlite_recovery_notice(&recovered_databases); + if let Some(notice) = &recovery_notice { + emit_state_db_backup_warning(SQLITE_RECOVERY_CONFIG_WARNING_SUMMARY); + emit_state_db_backup_warning(¬ice.details); + } + Ok(StateDbInitResult { + state_db: Some(state_db), + recovery_notice, + }) } fn sqlite_home_is_blocking_file(database_path: &Path) -> bool { @@ -1473,7 +1470,11 @@ fn sqlite_recovery_notice( }) .collect::>() .join("\n\n"); - Some(SqliteRecoveryNotice { details }) + Some(SqliteRecoveryNotice { + details: format!( + "Damaged local databases were rebuilt. Saved conversations remain in rollout files and can restore the thread list and history. Some database-only metadata may be unavailable. The original database files were preserved at the backup locations below.\n\n{details}" + ), + }) } fn emit_state_db_backup_warning(message: &str) { diff --git a/codex-rs/app-server/tests/suite/v2/mod.rs b/codex-rs/app-server/tests/suite/v2/mod.rs index 2d82aa8221..b94a7a45c6 100644 --- a/codex-rs/app-server/tests/suite/v2/mod.rs +++ b/codex-rs/app-server/tests/suite/v2/mod.rs @@ -125,6 +125,7 @@ mod server_diagnostics; mod session_end; mod skills_list; mod sleep; +mod sqlite_recovery; mod thread_archive; mod thread_attachments; mod thread_delete; diff --git a/codex-rs/app-server/tests/suite/v2/sqlite_recovery.rs b/codex-rs/app-server/tests/suite/v2/sqlite_recovery.rs new file mode 100644 index 0000000000..70dab94da1 --- /dev/null +++ b/codex-rs/app-server/tests/suite/v2/sqlite_recovery.rs @@ -0,0 +1,196 @@ +//! Verifies startup recovery preserves corrupt databases and allows thread creation. + +use anyhow::Result; +use app_test_support::TestAppServer; +use codex_app_server_protocol::ConfigWarningNotification; +use codex_app_server_protocol::ThreadStartParams; +use codex_state::SqliteConfig; +use codex_state::StateRuntime; +use codex_utils_absolute_path::test_support::PathExt; +use pretty_assertions::assert_eq; +use std::time::Duration; +use tempfile::TempDir; +use tokio::time::timeout; + +#[tokio::test] +async fn startup_reports_shared_and_fallback_recovery() -> Result<()> { + let home = TempDir::new()?; + let sqlite = SqliteConfig::new_for_testing(home.path().canonicalize()?.as_path().abs()); + let runtime = StateRuntime::init(sqlite.clone(), "openai".to_string()).await?; + runtime.close().await; + let logs_path = sqlite.logs_db_path(); + let pool = sqlite.open_read_write_pool(&logs_path).await?; + // The schema and migrations remain readable; opening the database alone succeeds. + sqlx::raw_sql( + "CREATE TABLE sample(value INTEGER); + INSERT INTO sample VALUES (NULL); + PRAGMA writable_schema=ON; + UPDATE sqlite_schema SET sql='CREATE TABLE sample(value INTEGER NOT NULL)' WHERE name='sample'; + PRAGMA writable_schema=OFF;", + ) + .execute(&pool) + .await?; + pool.close().await; + + // Logs recover inside the pool opener; the unreadable goals header forces + // the outer startup fallback and a retry after that first recovery. + tokio::fs::write(sqlite.goals_db_path(), b"damaged goals database").await?; + + let read_timeout = Duration::from_secs(/*secs*/ 30); + let mut server = TestAppServer::builder() + .with_codex_home(sqlite.home()) + .build_initialized_with_timeout(read_timeout) + .await?; + let warning: ConfigWarningNotification = + timeout(read_timeout, server.read_notification("configWarning")).await??; + assert_eq!(warning.summary, "Codex rebuilt its local database"); + let details = warning.details.expect("recovery details"); + assert!(details.contains("Some database-only metadata may be unavailable")); + let backups = std::fs::read_dir(sqlite.home().join("db-backups"))? + .collect::>>()?; + assert_eq!(backups.len(), 2); + for backup in &backups { + assert!(details.contains(&backup.path().display().to_string())); + } + assert!(details.contains(&sqlite.logs_db_path().display().to_string())); + assert!(details.contains(&sqlite.goals_db_path().display().to_string())); + let backup_dir = backups + .iter() + .map(std::fs::DirEntry::path) + .find(|path| { + path.join(logs_path.file_name().expect("log filename")) + .exists() + }) + .expect("preserved logs database"); + let backup_pool = sqlite + .open_read_only_pool( + &backup_dir.join(logs_path.file_name().expect("log filename")), + /*busy_timeout*/ None, + ) + .await?; + assert_eq!( + sqlx::query_scalar::<_, Option>("SELECT value FROM sample") + .fetch_all(&backup_pool) + .await?, + vec![None] + ); + backup_pool.close().await; + server.start_thread(ThreadStartParams::default()).await?; + timeout(read_timeout, server.shutdown_gracefully()).await??; + Ok(()) +} + +#[tokio::test] +async fn state_recovery_restores_saved_threads() -> Result<()> { + use codex_app_server_protocol::ThreadListParams; + use codex_app_server_protocol::ThreadListResponse; + use codex_app_server_protocol::ThreadReadParams; + use codex_app_server_protocol::ThreadReadResponse; + use codex_app_server_protocol::TurnStartParams; + use codex_app_server_protocol::TurnStatus; + use codex_app_server_protocol::UserInput; + + let home = TempDir::new()?; + let mock = + app_test_support::create_mock_responses_server_repeating_assistant("Saved reply").await; + app_test_support::MockResponsesConfig::new(&mock.uri()).write(home.path())?; + let sqlite = SqliteConfig::new_for_testing(home.path().canonicalize()?.as_path().abs()); + let read_timeout = Duration::from_secs(/*secs*/ 30); + let mut server = TestAppServer::builder() + .with_codex_home(sqlite.home()) + .build_initialized_with_timeout(read_timeout) + .await?; + let thread_id = server + .start_thread(ThreadStartParams::default()) + .await? + .thread + .id; + let completed = server + .start_turn_and_wait_for_completion(TurnStartParams { + thread_id: thread_id.clone(), + input: vec![UserInput::Text { + text: "Keep this conversation".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }) + .await?; + assert_eq!(completed.turn.status, TurnStatus::Completed); + let request = server + .send_thread_read_request(ThreadReadParams { + thread_id: thread_id.clone(), + include_turns: true, + }) + .await?; + let before: ThreadReadResponse = timeout(read_timeout, server.read_response(request)).await??; + timeout(read_timeout, server.shutdown_gracefully()).await??; + + // Introduce corruption only after the owning process has closed its pools. + let pool = sqlite.open_read_write_pool(&sqlite.state_db_path()).await?; + sqlx::raw_sql( + "CREATE TABLE sample(value INTEGER); + INSERT INTO sample VALUES (NULL); + PRAGMA writable_schema=ON; + UPDATE sqlite_schema SET sql='CREATE TABLE sample(value INTEGER NOT NULL)' WHERE name='sample'; + PRAGMA writable_schema=OFF;", + ).execute(&pool).await?; + pool.close().await; + + let mut server = TestAppServer::builder() + .with_codex_home(sqlite.home()) + .build_initialized_with_timeout(read_timeout) + .await?; + let request = server + .send_thread_list_request(ThreadListParams { + originators: None, + cursor: None, + limit: None, + sort_key: None, + sort_direction: None, + model_providers: None, + source_kinds: None, + archived: None, + section_id: None, + project_id: None, + cwd: None, + use_state_db_only: true, + search_term: None, + parent_thread_id: None, + ancestor_thread_id: None, + }) + .await?; + let listed: ThreadListResponse = timeout(read_timeout, server.read_response(request)).await??; + assert_eq!( + listed + .data + .iter() + .map(|thread| thread.id.as_str()) + .collect::>(), + vec![thread_id.as_str()] + ); + let request = server + .send_thread_read_request(ThreadReadParams { + thread_id, + include_turns: true, + }) + .await?; + let after: ThreadReadResponse = timeout(read_timeout, server.read_response(request)).await??; + assert_eq!(after.thread.turns, before.thread.turns); + let warning: ConfigWarningNotification = + timeout(read_timeout, server.read_notification("configWarning")).await??; + assert_eq!(warning.summary, "Codex rebuilt its local database"); + let details = warning.details.expect("recovery details"); + assert!(details.contains("Some database-only metadata may be unavailable")); + let backups = std::fs::read_dir(sqlite.home().join("db-backups"))? + .collect::>>()?; + assert_eq!(backups.len(), 1); + assert!(details.contains(&backups[0].path().display().to_string())); + assert!( + backups[0] + .path() + .join(sqlite.state_db_path().file_name().expect("state filename")) + .is_file() + ); + timeout(read_timeout, server.shutdown_gracefully()).await??; + Ok(()) +} diff --git a/codex-rs/ext/agent-message-board/src/local/lifecycle.rs b/codex-rs/ext/agent-message-board/src/local/lifecycle.rs index 787d02bf77..d44109de9c 100644 --- a/codex-rs/ext/agent-message-board/src/local/lifecycle.rs +++ b/codex-rs/ext/agent-message-board/src/local/lifecycle.rs @@ -38,7 +38,6 @@ impl LocalAgentMessageBoard { let pool = match sqlite.open_read_write_pool(&path).await { Ok(pool) => pool, Err(error) => { - let error = error.into(); if !codex_state::is_sqlite_corruption_error(&error) { return Err(storage_error(error)); } diff --git a/codex-rs/ext/agent-message-board/tests/local_board.rs b/codex-rs/ext/agent-message-board/tests/local_board.rs index c117b587dd..f7870b8fbd 100644 --- a/codex-rs/ext/agent-message-board/tests/local_board.rs +++ b/codex-rs/ext/agent-message-board/tests/local_board.rs @@ -87,6 +87,67 @@ impl MessageBoardHost for Host { } } +#[tokio::test] +async fn opening_board_preserves_posts_when_corruption_has_no_startup_recovery() { + let dir = tempfile::tempdir().unwrap(); + let sqlite = SqliteConfig::new_for_testing(dir.path().to_path_buf().try_into().unwrap()); + let root = ThreadId::new(); + let host = Arc::new(Host { + clock: AtomicI64::default(), + agent_path_calls: AtomicUsize::default(), + members: [(root, AgentPath::root())].into(), + fail_notifications: AtomicBool::new(false), + active: AtomicBool::new(true), + notifications: Mutex::default(), + }); + let path = dir.path().join("agent_message_board_1.sqlite"); + // Keep the validated fixture alive and copy it to a new file identity so the + // board's first open still checks the corruption. + let fixture = path.with_extension("fixture"); + let pool = sqlite.open_read_write_pool(&fixture).await.unwrap(); + sqlx::raw_sql( + "CREATE TABLE unrelated(value INTEGER); + INSERT INTO unrelated VALUES (NULL); + PRAGMA writable_schema=ON; + UPDATE sqlite_schema SET sql='CREATE TABLE unrelated(value INTEGER NOT NULL)' WHERE name='unrelated'; + PRAGMA writable_schema=OFF;", + ) + .execute(&pool) + .await + .unwrap(); + pool.close().await; + tokio::fs::copy(&fixture, &path).await.unwrap(); + let pool = sqlite + .open_read_only_pool(&path, /*busy_timeout*/ None) + .await + .unwrap(); + assert_eq!( + sqlx::query_scalar::<_, String>("PRAGMA quick_check(1)") + .fetch_one(&pool) + .await + .unwrap(), + "NULL value in unrelated.value" + ); + pool.close().await; + + let tree = SessionId::from(root); + let board = LocalAgentMessageBoard::open(&sqlite, tree, host.clone()) + .await + .unwrap(); + let request = PostRequest { + request_id: "preserved-post".into(), + destination: PostDestination::NewChannel("proofs".into()), + text: "Preserve this post.".into(), + agents_to_notify: Vec::new(), + }; + let metadata = board.post(root, request.clone()).await.unwrap(); + drop(board); + let resumed = LocalAgentMessageBoard::open(&sqlite, tree, host) + .await + .unwrap(); + assert_eq!(resumed.post(root, request).await.unwrap(), metadata); +} + #[tokio::test] async fn shared_handles_resume_posts_and_preserve_subscription_rules() { let dir = tempfile::tempdir().unwrap(); diff --git a/codex-rs/state/Cargo.toml b/codex-rs/state/Cargo.toml index c6c2215fb4..d98de0da11 100644 --- a/codex-rs/state/Cargo.toml +++ b/codex-rs/state/Cargo.toml @@ -10,6 +10,7 @@ chrono = { workspace = true } codex-history = { workspace = true } codex-protocol = { workspace = true } codex-utils-absolute-path = { workspace = true } +file-id = { workspace = true } libsqlite3-sys = { workspace = true } log = { workspace = true } serde = { workspace = true, features = ["derive"] } diff --git a/codex-rs/state/src/lib.rs b/codex-rs/state/src/lib.rs index be969b08cd..5dbf2a80ec 100644 --- a/codex-rs/state/src/lib.rs +++ b/codex-rs/state/src/lib.rs @@ -86,6 +86,7 @@ pub use runtime::SqliteIntegrityCheck; pub use runtime::SqliteQueueStore; pub use runtime::ThreadFilterOptions; pub use runtime::backup_runtime_db_for_fresh_start; +pub use runtime::collect_runtime_db_backups; pub use runtime::is_sqlite_corruption_error; pub use runtime::open_thread_history_db; pub use runtime::recovery::RuntimeDbBackup; @@ -133,6 +134,8 @@ pub const DB_ERROR_METRIC: &str = "codex.db.error"; pub const DB_METRIC_BACKFILL: &str = "codex.db.backfill"; /// Metrics on backfill duration. Tags: [status] pub const DB_METRIC_BACKFILL_DURATION_MS: &str = "codex.db.backfill.duration_ms"; +/// Confirmed SQLite quick-check corruption findings. Tags: [db] +pub const DB_CORRUPTION_METRIC: &str = "codex.sqlite.corruption.count"; /// SQLite initialization attempts. Tags: [status, phase, db, error] pub const DB_INIT_METRIC: &str = "codex.sqlite.init.count"; /// SQLite initialization latency. Tags: [status, phase, db, error] diff --git a/codex-rs/state/src/runtime.rs b/codex-rs/state/src/runtime.rs index 3cf1622226..2178378082 100644 --- a/codex-rs/state/src/runtime.rs +++ b/codex-rs/state/src/runtime.rs @@ -71,6 +71,7 @@ pub use goals::GoalUpdate; pub use memories::MemoryStore; pub use queued_items::SqliteQueueStore; pub use recovery::backup_runtime_db_for_fresh_start; +pub use recovery::collect_runtime_db_backups; pub use recovery::is_sqlite_corruption_error; pub use recovery::runtime_db_path_for_corruption_error; pub use recovery::sqlite_error_detail_is_corruption; diff --git a/codex-rs/state/src/runtime/recovery.rs b/codex-rs/state/src/runtime/recovery.rs index d9cf0bcdc9..72b47a0253 100644 --- a/codex-rs/state/src/runtime/recovery.rs +++ b/codex-rs/state/src/runtime/recovery.rs @@ -3,14 +3,39 @@ //! Codex keeps several independent runtime SQLite databases under one SQLite //! home. When SQLite reports that one of them is corrupt, automatic recovery //! moves only that database file and its sidecars into a backup folder so the -//! other databases keep their data. +//! other databases keep their data. Startup callers may collect backups in a +//! task-local scope to notify users after recovery; backup behavior is unchanged +//! outside that scope. use std::borrow::Cow; +use std::cell::RefCell; +use std::future::Future; use std::path::Path; use std::path::PathBuf; const BACKUP_DIR_NAME: &str = "db-backups"; +tokio::task_local! { + static RECOVERY_BACKUPS: RefCell>; +} + +/// Collect backups made while polling startup, including recovery across retries. +/// +/// The report is scoped to this task: unrelated processes and tasks are excluded. +/// Recovery in spawned tasks is not collected. Callers should report successful +/// recovery only after startup has completed successfully. +pub async fn collect_runtime_db_backups( + startup: impl Future, +) -> (T, Vec) { + RECOVERY_BACKUPS + .scope(RefCell::new(Vec::new()), async { + let result = startup.await; + let backups = RECOVERY_BACKUPS.with(std::cell::RefCell::take); + (result, backups) + }) + .await +} + #[derive(Clone, Debug, Eq, PartialEq)] pub struct RuntimeDbBackup { /// Path where the runtime database or sidecar lived before it was moved. @@ -77,7 +102,7 @@ pub async fn backup_runtime_db_for_fresh_start( db_path.display() )) })?; - match tokio::fs::metadata(sqlite_home).await { + let backups = match tokio::fs::metadata(sqlite_home).await { Ok(metadata) if metadata.is_dir() => backup_runtime_db_files(db_path).await, Ok(_) => backup_blocking_sqlite_home(sqlite_home).await, Err(err) if err.kind() == std::io::ErrorKind::NotFound => { @@ -88,7 +113,9 @@ pub async fn backup_runtime_db_for_fresh_start( ))) } Err(err) => Err(err), - } + }?; + let _ = RECOVERY_BACKUPS.try_with(|report| report.borrow_mut().extend(backups.iter().cloned())); + Ok(backups) } pub fn runtime_db_path_for_corruption_error(err: &anyhow::Error) -> Option { @@ -105,12 +132,21 @@ pub fn is_sqlite_corruption_error(err: &anyhow::Error) -> bool { } fn sqlite_error_source_is_corruption(source: &(dyn std::error::Error + 'static)) -> bool { + if let Some(err) = source.downcast_ref::() { + return matches!( + err.code, + libsqlite3_sys::ErrorCode::DatabaseCorrupt | libsqlite3_sys::ErrorCode::NotADatabase + ); + } + let Some(err) = source.downcast_ref::() else { return false; }; + let sqlx::Error::Database(database_error) = err else { return false; }; + sqlite_error_detail_is_corruption(database_error.message()) || database_error .code() diff --git a/codex-rs/state/src/runtime/recovery_tests.rs b/codex-rs/state/src/runtime/recovery_tests.rs index 3861000dd3..6caca58d85 100644 --- a/codex-rs/state/src/runtime/recovery_tests.rs +++ b/codex-rs/state/src/runtime/recovery_tests.rs @@ -19,7 +19,11 @@ async fn backup_moves_only_requested_runtime_db_files_to_backup_folder() -> std: let failed_db_path = sqlite.logs_db_path(); let failed_paths = sqlite_paths(failed_db_path.as_path()); - let backups = backup_runtime_db_for_fresh_start(failed_db_path.as_path()).await?; + let (result, report) = + collect_runtime_db_backups(backup_runtime_db_for_fresh_start(failed_db_path.as_path())) + .await; + let backups = result?; + assert_eq!(report, backups); assert_eq!(backups.len(), failed_paths.len()); for path in &failed_paths { diff --git a/codex-rs/state/src/sqlite.rs b/codex-rs/state/src/sqlite.rs index 4d1c9e9819..e35a074d68 100644 --- a/codex-rs/state/src/sqlite.rs +++ b/codex-rs/state/src/sqlite.rs @@ -25,6 +25,12 @@ use std::path::PathBuf; use std::time::Duration; use std::time::Instant; +mod validation; + +use validation::SqliteQuickCheckManager; + +const DEFAULT_BUSY_TIMEOUT: Duration = Duration::from_secs(5); + const LOGS_DB_FILENAME: &str = "logs_2.sqlite"; const GOALS_DB_FILENAME: &str = "goals_1.sqlite"; const MEMORIES_DB_FILENAME: &str = "memories_1.sqlite"; @@ -32,6 +38,12 @@ const QUEUE_DB_FILENAME: &str = "queue_1.sqlite"; const STATE_DB_FILENAME: &str = "state_5.sqlite"; const THREAD_HISTORY_DB_FILENAME: &str = "thread_history_1.sqlite"; +#[derive(Clone, Copy, Eq, PartialEq)] +enum RecoveryMode { + BackupAndRebuild, + Unavailable, +} + #[derive(Clone, Copy)] struct RuntimeDbSpec { label: &'static str, @@ -42,6 +54,7 @@ struct RuntimeDbSpec { /// Opt in only after auditing writers for deferred read-to-write upgrades: /// an intervening reclamation commit can make those fail with SQLITE_BUSY_SNAPSHOT. background_reclamation: bool, + recovery: RecoveryMode, } impl RuntimeDbSpec { @@ -57,6 +70,7 @@ const STATE_DB: RuntimeDbSpec = RuntimeDbSpec { open_phase: "open_state", migrate_phase: "migrate_state", background_reclamation: false, + recovery: RecoveryMode::BackupAndRebuild, }; const LOGS_DB: RuntimeDbSpec = RuntimeDbSpec { @@ -67,6 +81,7 @@ const LOGS_DB: RuntimeDbSpec = RuntimeDbSpec { migrate_phase: "migrate_logs", // Log transactions write before reading, so they already hold the writer lock. background_reclamation: true, + recovery: RecoveryMode::BackupAndRebuild, }; const GOALS_DB: RuntimeDbSpec = RuntimeDbSpec { @@ -76,6 +91,7 @@ const GOALS_DB: RuntimeDbSpec = RuntimeDbSpec { open_phase: "open_goals", migrate_phase: "migrate_goals", background_reclamation: false, + recovery: RecoveryMode::BackupAndRebuild, }; const MEMORIES_DB: RuntimeDbSpec = RuntimeDbSpec { @@ -85,6 +101,7 @@ const MEMORIES_DB: RuntimeDbSpec = RuntimeDbSpec { open_phase: "open_memories", migrate_phase: "migrate_memories", background_reclamation: false, + recovery: RecoveryMode::BackupAndRebuild, }; const MEMORIES_V2_DB: RuntimeDbSpec = RuntimeDbSpec { @@ -101,6 +118,7 @@ const QUEUE_DB: RuntimeDbSpec = RuntimeDbSpec { open_phase: "open_queue", migrate_phase: "migrate_queue", background_reclamation: false, + recovery: RecoveryMode::BackupAndRebuild, }; const THREAD_HISTORY_DB: RuntimeDbSpec = RuntimeDbSpec { @@ -110,6 +128,7 @@ const THREAD_HISTORY_DB: RuntimeDbSpec = RuntimeDbSpec { open_phase: "open_thread_history", migrate_phase: "migrate_thread_history", background_reclamation: false, + recovery: RecoveryMode::Unavailable, }; const RUNTIME_DBS: [RuntimeDbSpec; 7] = [ @@ -130,14 +149,28 @@ pub struct RuntimeDbPath { } /// Resolved configuration shared by all Codex SQLite connections. -#[derive(Clone, Debug, Eq, PartialEq)] +/// Clones share quick-check attempts; a newly constructed config checks independently. +#[derive(Clone, Debug)] pub struct SqliteConfig { sqlite_home: AbsolutePathBuf, + quick_check_manager: SqliteQuickCheckManager, } +// Validation history is operational state, not part of configuration equality. +impl PartialEq for SqliteConfig { + fn eq(&self, other: &Self) -> bool { + self.sqlite_home == other.sqlite_home + } +} + +impl Eq for SqliteConfig {} + impl SqliteConfig { pub fn from_sqlite_home(sqlite_home: AbsolutePathBuf) -> Self { - Self { sqlite_home } + Self { + sqlite_home, + quick_check_manager: SqliteQuickCheckManager::new(), + } } pub fn new_for_testing(sqlite_home: AbsolutePathBuf) -> Self { @@ -269,9 +302,8 @@ impl SqliteConfig { let path = spec.path(self.home()); let started = Instant::now(); let pool_result = self - .open_read_write_pool(&path) - .await - .map_err(anyhow::Error::from); + .open_read_write_pool_with_spec(path.as_path(), Some(spec), telemetry_override) + .await; telemetry::record_init_result( telemetry_override, spec.kind, @@ -307,61 +339,111 @@ impl SqliteConfig { } /// Open a writable Codex SQLite database, creating it if necessary. - pub async fn open_read_write_pool(&self, path: &Path) -> Result { + pub async fn open_read_write_pool(&self, path: &Path) -> anyhow::Result { + self.open_read_write_pool_with_spec( + path, /*recover_spec*/ None, /*telemetry_override*/ None, + ) + .await + } + + async fn open_read_write_pool_with_spec( + &self, + path: &Path, + recover_spec: Option, + telemetry_override: Option<&dyn DbTelemetry>, + ) -> anyhow::Result { + let recovery = recover_spec.map_or(RecoveryMode::Unavailable, |spec| spec.recovery); let options = SqliteConnectOptions::new() .filename(path) .create_if_missing(true) .synchronous(SqliteSynchronous::Normal) - .busy_timeout(Duration::from_secs(5)) + .busy_timeout(DEFAULT_BUSY_TIMEOUT) .log_statements(LevelFilter::Off); - // SQLx retries after_connect errors, eventually replacing them with PoolTimedOut. - // Return the first initialization error directly while opening this pool. - let (init_error_tx, mut init_error_rx) = tokio::sync::mpsc::channel(/*buffer*/ 1); - let pool_options = SqlitePoolOptions::new().max_connections(5).after_connect( - move |connection, _metadata| { - let init_error_tx = init_error_tx.clone(); - Box::pin(async move { - let result = async { - let mode: i64 = sqlx::query_scalar("PRAGMA auto_vacuum") - .fetch_one(&mut *connection) - .await?; - // The setter takes the writer lock even when the mode is unchanged. - // Initialize before WAL creates the first database page; preserve - // existing modes, including FULL, without taking the writer lock. - let empty = if mode == 0 { - sqlx::query_scalar::<_, bool>( - "SELECT NOT EXISTS (SELECT 1 FROM sqlite_schema)", - ) - .fetch_one(&mut *connection) - .await? - } else { - false - }; - if empty { - sqlx::query("PRAGMA auto_vacuum = INCREMENTAL") - .execute(&mut *connection) + let connect = || async { + // SQLx retries after_connect errors, eventually replacing them with PoolTimedOut. + // Return the first initialization error directly while opening this pool. + let (init_error_tx, mut init_error_rx) = tokio::sync::mpsc::channel(/*buffer*/ 1); + let pool_options = SqlitePoolOptions::new().max_connections(5).after_connect( + move |connection, _metadata| { + let init_error_tx = init_error_tx.clone(); + Box::pin(async move { + let result = async { + let mode: i64 = sqlx::query_scalar("PRAGMA auto_vacuum") + .fetch_one(&mut *connection) .await?; + // The setter takes the writer lock even when the mode is unchanged. + // Initialize before WAL creates the first database page; preserve + // existing modes, including FULL, without taking the writer lock. + let empty = if mode == 0 { + sqlx::query_scalar::<_, bool>( + "SELECT NOT EXISTS (SELECT 1 FROM sqlite_schema)", + ) + .fetch_one(&mut *connection) + .await? + } else { + false + }; + if empty { + sqlx::query("PRAGMA auto_vacuum = INCREMENTAL") + .execute(&mut *connection) + .await?; + } + sqlx::query("PRAGMA journal_mode = WAL") + .execute(connection) + .await?; + Ok(()) } - sqlx::query("PRAGMA journal_mode = WAL") - .execute(connection) - .await?; - Ok(()) - } - .await; - result.map_err(|error| match init_error_tx.try_send(error) { - // The opener owns the original error and cancels connection retries. - Ok(()) => Error::PoolClosed, - // After opening, lazy connections retain SQLx's normal error handling. - Err(error) => error.into_inner(), + .await; + result.map_err(|error| match init_error_tx.try_send(error) { + // The opener owns the original error and cancels connection retries. + Ok(()) => Error::PoolClosed, + // After opening, lazy connections retain SQLx's normal error handling. + Err(error) => error.into_inner(), + }) }) - }) - }, - ); - tokio::select! { - biased; - Some(error) = init_error_rx.recv() => Err(error), - result = pool_options.connect_with(options) => result, + }, + ); + tokio::select! { + biased; + Some(error) = init_error_rx.recv() => Err(error), + result = pool_options.connect_with(options.clone()) => result, + } + }; + let pool = connect().await?; + let validation_result = self + .quick_check_manager + .quick_check_once( + &pool, + path, + // Limit startup validation to 100 ms. + Duration::from_millis(/*millis*/ 100), + ) + .await; + + let Ok(result) = validation_result else { + pool.close().await; + return validation_result.map(|_| pool); + }; + + if result == validation::QuickCheckOutcome::CorruptedNeedsFixed { + tracing::error!(database = %path.display(), "SQLite quick check detected corruption"); + telemetry::record_corruption(telemetry_override, recover_spec.map(|spec| spec.kind)); + if recovery == RecoveryMode::BackupAndRebuild { + pool.close().await; + let backups = crate::backup_runtime_db_for_fresh_start(path).await?; + for backup in backups { + tracing::warn!( + database = %backup.original_path.display(), + backup = %backup.backup_path.display(), + "Preserved corrupt SQLite database before rebuilding" + ); + } + + return connect().await.map_err(anyhow::Error::from); + } } + + Ok(pool) } /// Open an existing Codex SQLite database without creating or modifying it. diff --git a/codex-rs/state/src/sqlite/validation.rs b/codex-rs/state/src/sqlite/validation.rs new file mode 100644 index 0000000000..f606d80f00 --- /dev/null +++ b/codex-rs/state/src/sqlite/validation.rs @@ -0,0 +1,80 @@ +//! Budgeted integrity checks that report confirmed corruption to database owners. +//! +//! Each file identity is checked at most once per shared SQLite configuration. +//! Attempts are recorded before validation, including failed or cancelled checks. + +use file_id::FileId; +use sqlx::SqlitePool; +use std::collections::HashSet; +use std::path::Path; +use std::sync::Arc; +use std::time::Duration; +use std::time::Instant; +use tokio::sync::Mutex; + +#[derive(Debug, Eq, PartialEq)] +pub(super) enum QuickCheckOutcome { + Complete, + Incomplete, + CorruptedNeedsFixed, + Skipped, +} + +/// Keeps track of what databases were verified with quick_check() already +#[derive(Clone, Debug, Default)] +pub(super) struct SqliteQuickCheckManager(Arc>>); + +impl SqliteQuickCheckManager { + pub fn new() -> Self { + Self::default() + } + + /// Returns this attempt's outcome, or `Skipped` if validation was already attempted. + pub async fn quick_check_once( + &self, + pool: &SqlitePool, + path: &Path, + budget: Duration, + ) -> anyhow::Result { + let id = file_id::get_file_id(path)?; + + if !self.0.lock().await.insert(id) { + // Only check each file once for this owner. + return Ok(QuickCheckOutcome::Skipped); + } + + quick_check(pool, Instant::now() + budget).await + } +} + +pub(super) async fn quick_check( + pool: &SqlitePool, + deadline: Instant, +) -> anyhow::Result { + let mut connection = pool.acquire().await?; + connection + .lock_handle() + .await? + .set_progress_handler(/*num_ops*/ 1_000, move || Instant::now() < deadline); + + // quick_check(1) returns either "ok" or the first corruption finding + let result = sqlx::query_scalar::<_, String>("PRAGMA quick_check(1)") + .fetch_one(&mut *connection) + .await; + connection.lock_handle().await?.remove_progress_handler(); + + connection.return_to_pool().await; + match result.map_err(anyhow::Error::from) { + Ok(result) if result == "ok" => Ok(QuickCheckOutcome::Complete), + Ok(_) => Ok(QuickCheckOutcome::CorruptedNeedsFixed), + Err(error) if crate::is_sqlite_corruption_error(&error) => { + Ok(QuickCheckOutcome::CorruptedNeedsFixed) + } + // timeouts or other unhandled errors + Err(_) => Ok(QuickCheckOutcome::Incomplete), + } +} + +#[cfg(test)] +#[path = "validation_tests.rs"] +mod tests; diff --git a/codex-rs/state/src/sqlite/validation_tests.rs b/codex-rs/state/src/sqlite/validation_tests.rs new file mode 100644 index 0000000000..f3d3853c8a --- /dev/null +++ b/codex-rs/state/src/sqlite/validation_tests.rs @@ -0,0 +1,586 @@ +//! Exercises startup validation, per-file caching, and report-only lazy databases. + +use super::QuickCheckOutcome; +use super::SqliteQuickCheckManager; +use super::quick_check; +use crate::SqliteConfig; +use crate::StateRuntime; +use crate::runtime::test_support::unique_temp_dir; +use anyhow::Context; +use codex_utils_absolute_path::test_support::PathExt; +use pretty_assertions::assert_eq; +use std::collections::BTreeMap; +use std::sync::Mutex; +use std::time::Duration; +use std::time::Instant; + +#[derive(Debug, Eq, PartialEq)] +struct CorruptionEvent { + count: i64, + tags: BTreeMap, +} + +#[derive(Default)] +struct CorruptionTelemetry(Mutex>); + +impl crate::DbTelemetry for CorruptionTelemetry { + fn counter(&self, name: &str, inc: i64, tags: &[(&str, &str)]) { + if name == crate::DB_CORRUPTION_METRIC { + self.0 + .lock() + .expect("telemetry lock") + .push(CorruptionEvent { + count: inc, + tags: tags + .iter() + .map(|(key, value)| (key.to_string(), value.to_string())) + .collect(), + }); + } + } + + fn histogram(&self, _name: &str, _value: i64, _tags: &[(&str, &str)]) {} + + fn record_duration(&self, _name: &str, _duration: Duration, _tags: &[(&str, &str)]) {} +} + +#[tokio::test] +async fn runtime_opens_recover_or_report_corruption_by_database_policy() -> anyhow::Result<()> { + let root = unique_temp_dir(); + let _cleanup = scopeguard::guard(root.clone(), |root| { + let _ = std::fs::remove_dir_all(root); + }); + for spec in super::super::RUNTIME_DBS { + let home = root.join(spec.label); + tokio::fs::create_dir_all(&home).await?; + let sqlite = SqliteConfig::new_for_testing(home.as_path().abs()); + let path = spec.path(&home); + let fixture = path.with_extension("fixture"); + let pool = sqlx::SqlitePool::connect_with( + sqlx::sqlite::SqliteConnectOptions::new() + .filename(&fixture) + .create_if_missing(true), + ) + .await?; + sqlx::raw_sql( + "CREATE TABLE sample(value INTEGER); + INSERT INTO sample VALUES (NULL); + PRAGMA writable_schema=ON; + UPDATE sqlite_schema SET sql='CREATE TABLE sample(value INTEGER NOT NULL)' WHERE name='sample'; + PRAGMA writable_schema=OFF;", + ) + .execute(&pool) + .await?; + pool.close().await; + tokio::fs::rename(&fixture, &path).await?; + + if path == sqlite.thread_history_db_path() { + let telemetry = CorruptionTelemetry::default(); + // Unrelated corruption must not disable lazy history reads when the + // caller has no recovery path. Exercise both migration and reopening. + for _ in 0..2 { + let pool = sqlite + .open_thread_history_db( + &crate::migrations::runtime_thread_history_migrator(), + Some(&telemetry), + ) + .await?; + assert_eq!( + sqlx::query_scalar::<_, i64>("SELECT count(*) FROM thread_turns") + .fetch_one(&pool) + .await?, + 0 + ); + assert_eq!( + quick_check(&pool, Instant::now() + Duration::from_secs(/*secs*/ 5)).await?, + QuickCheckOutcome::CorruptedNeedsFixed + ); + pool.close().await; + } + assert_eq!( + *telemetry.0.lock().expect("telemetry lock"), + vec![CorruptionEvent { + count: 1, + tags: BTreeMap::from([("db".to_string(), "thread_history".to_string())]), + }] + ); + continue; + } + + // Recovery happens inside the database opener, without a CLI or app server. + for _ in 0..2 { + let runtime = StateRuntime::init(sqlite.clone(), "openai".to_string()).await?; + runtime.close().await; + } + let backups = std::fs::read_dir(home.join("db-backups")) + .with_context(|| format!("{} corruption did not create a backup", spec.label))? + .collect::>>()?; + assert_eq!(backups.len(), 1); + let backup_path = backups[0] + .path() + .join(path.file_name().expect("database filename")); + let pool = sqlite + .open_read_only_pool(&backup_path, /*busy_timeout*/ None) + .await?; + assert_eq!( + sqlx::query_scalar::<_, String>("PRAGMA quick_check(1)") + .fetch_one(&pool) + .await?, + "NULL value in sample.value" + ); + assert_eq!( + sqlx::query_scalar::<_, String>("SELECT name FROM sqlite_schema WHERE type = 'table'") + .fetch_all(&pool) + .await?, + vec!["sample".to_string()] + ); + pool.close().await; + } + Ok(()) +} + +#[tokio::test] +async fn quick_check_failure_is_recovered_before_runtime_init_returns() -> anyhow::Result<()> { + let home = unique_temp_dir(); + let _cleanup = scopeguard::guard(home.clone(), |home| { + let _ = std::fs::remove_dir_all(home); + }); + let telemetry = CorruptionTelemetry::default(); + let sqlite = SqliteConfig::new_for_testing(home.as_path().abs()); + let runtime = StateRuntime::init_with_telemetry_for_tests( + sqlite.clone(), + "openai".to_string(), + &telemetry, + ) + .await?; + runtime.close().await; + let state_pool = sqlite.open_read_write_pool(&sqlite.state_db_path()).await?; + sqlx::query("CREATE TABLE preserved(value); INSERT INTO preserved VALUES ('keep me')") + .execute(&state_pool) + .await?; + state_pool.close().await; + let logs_path = sqlite.logs_db_path(); + sqlite + .open_logs_db( + &crate::migrations::runtime_logs_migrator(), + /*telemetry_override*/ None, + ) + .await? + .close() + .await; + let fixture = logs_path.with_extension("fixture"); + let pool = sqlx::SqlitePool::connect_with( + sqlx::sqlite::SqliteConnectOptions::new() + .filename(&fixture) + .create_if_missing(true), + ) + .await?; + sqlx::raw_sql( + "CREATE TABLE sample(value INTEGER); + INSERT INTO sample VALUES (NULL); + PRAGMA writable_schema=ON; + UPDATE sqlite_schema SET sql='CREATE TABLE sample(value INTEGER NOT NULL)' WHERE name='sample'; + PRAGMA writable_schema=OFF;", + ) + .execute(&pool) + .await?; + pool.close().await; + // Replacing an already-validated file must trigger a new check at the same path. + tokio::fs::remove_file(&logs_path).await?; + tokio::fs::rename(&fixture, &logs_path).await?; + let runtime = StateRuntime::init_with_telemetry_for_tests( + sqlite.clone(), + "openai".to_string(), + &telemetry, + ) + .await?; + runtime.close().await; + let runtime = StateRuntime::init_with_telemetry_for_tests( + sqlite.clone(), + "openai".to_string(), + &telemetry, + ) + .await?; + runtime.close().await; + assert_eq!( + *telemetry.0.lock().expect("telemetry lock"), + vec![CorruptionEvent { + count: 1, + tags: BTreeMap::from([("db".to_string(), "logs".to_string())]), + }] + ); + let backups = + std::fs::read_dir(home.join("db-backups"))?.collect::>>()?; + assert_eq!(backups.len(), 1); + let backup_path = backups[0] + .path() + .join(logs_path.file_name().expect("logs filename")); + let backup_pool = sqlite + .open_read_only_pool(&backup_path, /*busy_timeout*/ None) + .await?; + assert_eq!( + sqlx::query_scalar::<_, Option>("SELECT value FROM sample") + .fetch_all(&backup_pool) + .await?, + vec![None], + ); + backup_pool.close().await; + let pool = sqlite.open_read_write_pool(&sqlite.state_db_path()).await?; + assert_eq!( + sqlx::query_scalar::<_, String>("SELECT value FROM preserved") + .fetch_all(&pool) + .await?, + vec!["keep me".to_string()] + ); + pool.close().await; + Ok(()) +} + +#[tokio::test] +async fn repeated_pool_opens_share_completed_and_incomplete_attempts() -> anyhow::Result<()> { + for budget in [Duration::from_secs(/*secs*/ 5), Duration::ZERO] { + let home = unique_temp_dir(); + tokio::fs::create_dir_all(&home).await?; + let _cleanup = scopeguard::guard(home.clone(), |home| { + let _ = std::fs::remove_dir_all(home); + }); + let sqlite = SqliteConfig::new_for_testing(home.as_path().abs()); + let fixture = home.join("fixture.sqlite"); + let pool = sqlx::SqlitePool::connect_with( + sqlx::sqlite::SqliteConnectOptions::new() + .filename(&fixture) + .create_if_missing(true), + ) + .await?; + sqlx::raw_sql( + "CREATE TABLE sample(value INTEGER); + WITH RECURSIVE rows(id) AS (SELECT 1 UNION ALL SELECT id + 1 FROM rows WHERE id < 2048) + INSERT INTO sample SELECT id FROM rows;", + ) + .execute(&pool) + .await?; + pool.close().await; + let path = sqlite.logs_db_path(); + tokio::fs::rename(&fixture, &path).await?; + let pool = sqlite + .open_read_only_pool(&path, /*busy_timeout*/ None) + .await?; + sqlite + .quick_check_manager + .quick_check_once(&pool, &path, budget) + .await?; + pool.close().await; + + let migrator = crate::migrations::runtime_logs_migrator(); + let pool = sqlite + .open_logs_db(&migrator, /*telemetry_override*/ None) + .await?; + sqlx::raw_sql( + "UPDATE sample SET value=NULL WHERE value=1; + PRAGMA writable_schema=ON; + UPDATE sqlite_schema SET sql='CREATE TABLE sample(value INTEGER NOT NULL)' WHERE name='sample'; + PRAGMA writable_schema=OFF;", + ) + .execute(&pool) + .await?; + pool.close().await; + // Corruption introduced after the first attempt exposes any repeated scan. + let cloned_sqlite = sqlite.clone(); + let (first, second) = tokio::join!( + sqlite.open_logs_db(&migrator, /*telemetry_override*/ None), + cloned_sqlite.open_logs_db(&migrator, /*telemetry_override*/ None), + ); + let first = first?; + let second = second?; + assert_eq!( + quick_check(&first, Instant::now() + Duration::from_secs(/*secs*/ 5)).await?, + QuickCheckOutcome::CorruptedNeedsFixed, + "an uncached check must still see the corruption" + ); + first.close().await; + second.close().await; + + // A separately constructed configuration must check and recover the file. + let independent_sqlite = SqliteConfig::new_for_testing(home.as_path().abs()); + let repaired = independent_sqlite + .open_logs_db(&migrator, /*telemetry_override*/ None) + .await?; + assert_eq!( + quick_check(&repaired, Instant::now() + Duration::from_secs(/*secs*/ 5)).await?, + QuickCheckOutcome::Complete + ); + assert!(home.join("db-backups").is_dir()); + repaired.close().await; + } + Ok(()) +} + +#[tokio::test] +async fn concurrent_first_checks_share_an_in_flight_attempt() -> anyhow::Result<()> { + let home = unique_temp_dir(); + tokio::fs::create_dir_all(&home).await?; + let _cleanup = scopeguard::guard(home.clone(), |home| { + let _ = std::fs::remove_dir_all(home); + }); + let validation = SqliteQuickCheckManager::new(); + let sqlite = SqliteConfig::new_for_testing(home.as_path().abs()); + let path = sqlite.logs_db_path(); + sqlx::SqlitePool::connect_with( + sqlx::sqlite::SqliteConnectOptions::new() + .filename(&path) + .create_if_missing(true), + ) + .await? + .close() + .await; + let path = tokio::fs::canonicalize(path).await?; + let first = sqlite + .open_read_only_pool(&path, /*busy_timeout*/ None) + .await?; + let second = sqlite + .open_read_only_pool(&path, /*busy_timeout*/ None) + .await?; + // A second scan would fail on this pool instead of silently passing twice. + second.close().await; + let mut connection = first.acquire().await?; + let id = file_id::get_file_id(&path)?; + let check = tokio::spawn({ + let validation = validation.clone(); + let pool = first.clone(); + let path = path.clone(); + async move { + validation + .quick_check_once(&pool, &path, Duration::from_secs(/*secs*/ 5)) + .await + } + }); + tokio::time::timeout(Duration::from_secs(/*secs*/ 5), async { + while !validation.0.lock().await.contains(&id) { + tokio::task::yield_now().await; + } + }) + .await?; + // The concurrent caller skips immediately, while the first check is blocked. + assert_eq!( + validation + .quick_check_once(&second, &path, Duration::from_secs(/*secs*/ 5)) + .await?, + QuickCheckOutcome::Skipped + ); + connection.return_to_pool().await; + assert_eq!(check.await??, QuickCheckOutcome::Complete); + first.close().await; + Ok(()) +} + +#[tokio::test] +async fn cancelled_check_counts_as_an_attempt() -> anyhow::Result<()> { + let home = unique_temp_dir(); + tokio::fs::create_dir_all(&home).await?; + let _cleanup = scopeguard::guard(home.clone(), |home| { + let _ = std::fs::remove_dir_all(home); + }); + let validation = SqliteQuickCheckManager::new(); + let sqlite = SqliteConfig::new_for_testing(home.as_path().abs()); + let path = sqlite.logs_db_path(); + sqlx::SqlitePool::connect_with( + sqlx::sqlite::SqliteConnectOptions::new() + .filename(&path) + .create_if_missing(true), + ) + .await? + .close() + .await; + let id = file_id::get_file_id(&path)?; + let pool = sqlite + .open_read_only_pool(&path, /*busy_timeout*/ None) + .await?; + // Occupy the sole connection so cancellation happens during initialization. + let mut connection = pool.acquire().await?; + let check = tokio::spawn({ + let validation = validation.clone(); + let pool = pool.clone(); + let path = path.clone(); + async move { + validation + .quick_check_once(&pool, &path, Duration::from_secs(/*secs*/ 5)) + .await + } + }); + tokio::time::timeout(Duration::from_secs(/*secs*/ 5), async { + loop { + if validation.0.lock().await.contains(&id) { + break; + } + tokio::task::yield_now().await; + } + }) + .await?; + check.abort(); + assert!( + check + .await + .expect_err("check should be cancelled") + .is_cancelled() + ); + connection.return_to_pool().await; + assert_eq!( + tokio::time::timeout( + Duration::from_secs(/*secs*/ 5), + validation.quick_check_once(&pool, &path, Duration::from_secs(/*secs*/ 5)), + ) + .await??, + QuickCheckOutcome::Skipped + ); + pool.close().await; + Ok(()) +} + +#[tokio::test] +async fn corruption_attempts_are_cached_per_owner() -> anyhow::Result<()> { + let home = unique_temp_dir(); + tokio::fs::create_dir_all(&home).await?; + let _cleanup = scopeguard::guard(home.clone(), |home| { + let _ = std::fs::remove_dir_all(home); + }); + let validation = SqliteQuickCheckManager::new(); + let sqlite = SqliteConfig::new_for_testing(home.as_path().abs()); + let path = sqlite.logs_db_path(); + let pool = sqlx::SqlitePool::connect_with( + sqlx::sqlite::SqliteConnectOptions::new() + .filename(&path) + .create_if_missing(true), + ) + .await?; + let mut connection = pool.acquire().await?; + sqlx::raw_sql( + "CREATE TABLE sample(value INTEGER); + PRAGMA writable_schema=ON; + UPDATE sqlite_schema SET rootpage=2147483647 WHERE name='sample'; + PRAGMA writable_schema=OFF; + PRAGMA schema_version=99;", + ) + .execute(&mut *connection) + .await?; + connection.return_to_pool().await; + + assert_eq!( + validation + .quick_check_once(&pool, &path, Duration::from_secs(/*secs*/ 5)) + .await?, + QuickCheckOutcome::CorruptedNeedsFixed + ); + // Another owner must check the same file independently. + let independent_validation = SqliteQuickCheckManager::new(); + assert_eq!( + independent_validation + .quick_check_once(&pool, &path, Duration::from_secs(/*secs*/ 5)) + .await?, + QuickCheckOutcome::CorruptedNeedsFixed + ); + pool.close().await; + // A closed pool proves the completed attempt is skipped without another query. + assert_eq!( + validation + .quick_check_once(&pool, &path, Duration::from_secs(/*secs*/ 5)) + .await?, + QuickCheckOutcome::Skipped + ); + Ok(()) +} + +#[tokio::test] +async fn interrupted_check_leaves_connection_usable() -> anyhow::Result<()> { + let home = unique_temp_dir(); + tokio::fs::create_dir_all(&home).await?; + let _cleanup = scopeguard::guard(home.clone(), |home| { + let _ = std::fs::remove_dir_all(home); + }); + let sqlite = SqliteConfig::new_for_testing(home.as_path().abs()); + let path = sqlite.logs_db_path(); + let pool = sqlite.open_read_write_pool(&path).await?; + sqlx::raw_sql( + "CREATE TABLE sample(value INTEGER); + WITH RECURSIVE rows(id) AS (SELECT 1 UNION ALL SELECT id + 1 FROM rows WHERE id < 2048) + INSERT INTO sample SELECT id FROM rows;", + ) + .execute(&pool) + .await?; + pool.close().await; + // One connection ensures every assertion exercises the connection that was interrupted. + let pool = sqlite + .open_read_only_pool(&path, Some(Duration::from_secs(/*secs*/ 5))) + .await?; + assert_eq!( + quick_check(&pool, Instant::now()).await?, + QuickCheckOutcome::Incomplete + ); + assert_eq!( + sqlx::query_scalar::<_, i64>("SELECT sum(value) FROM sample") + .fetch_one(&pool) + .await?, + 2_098_176 + ); + assert_eq!( + quick_check(&pool, Instant::now() + Duration::from_secs(/*secs*/ 5)).await?, + QuickCheckOutcome::Complete + ); + pool.close().await; + + let pool = sqlite.open_read_write_pool(&path).await?; + sqlx::raw_sql( + "UPDATE sample SET value=NULL WHERE value=1; + PRAGMA writable_schema=ON; + UPDATE sqlite_schema SET sql='CREATE TABLE sample(value INTEGER NOT NULL)' WHERE name='sample'; + PRAGMA writable_schema=OFF;", + ) + .execute(&pool) + .await?; + pool.close().await; + let pool = sqlite + .open_read_only_pool(&path, /*busy_timeout*/ None) + .await?; + assert_eq!( + quick_check(&pool, Instant::now()).await?, + QuickCheckOutcome::CorruptedNeedsFixed, + "retain a finding even if a later scan step exceeds the budget" + ); + pool.close().await; + Ok(()) +} + +#[tokio::test] +async fn locked_database_does_not_count_as_corruption() -> anyhow::Result<()> { + let home = unique_temp_dir(); + tokio::fs::create_dir_all(&home).await?; + let _cleanup = scopeguard::guard(home.clone(), |home| { + let _ = std::fs::remove_dir_all(home); + }); + let sqlite = SqliteConfig::new_for_testing(home.as_path().abs()); + let path = sqlite.logs_db_path(); + let writer_pool = sqlite.open_read_write_pool(&path).await?; + let mut writer = writer_pool.acquire().await?; + sqlx::query("PRAGMA journal_mode=DELETE") + .execute(&mut *writer) + .await?; + let reader_pool = sqlite + .open_read_only_pool(&path, Some(Duration::from_millis(/*millis*/ 50))) + .await?; + sqlx::query("BEGIN EXCLUSIVE").execute(&mut *writer).await?; + let started = Instant::now(); + let result = quick_check(&reader_pool, started + Duration::from_millis(/*millis*/ 50)).await; + sqlx::query("ROLLBACK").execute(&mut *writer).await?; + drop(writer); + writer_pool.close().await; + assert_eq!(result?, QuickCheckOutcome::Incomplete); + assert_eq!( + quick_check( + &reader_pool, + Instant::now() + Duration::from_secs(/*secs*/ 5) + ) + .await?, + QuickCheckOutcome::Complete + ); + reader_pool.close().await; + Ok(()) +} diff --git a/codex-rs/state/src/sqlite_tests.rs b/codex-rs/state/src/sqlite_tests.rs index d9372dd3a4..327031b909 100644 --- a/codex-rs/state/src/sqlite_tests.rs +++ b/codex-rs/state/src/sqlite_tests.rs @@ -108,7 +108,8 @@ async fn open_read_write_pool_preserves_wal_conversion_lock_error() -> anyhow::R let error = result.expect_err("WAL conversion requires the writer lock"); assert_eq!( error - .as_database_error() + .downcast_ref::() + .and_then(sqlx::Error::as_database_error) .and_then(sqlx::error::DatabaseError::code), Some("5".into()) ); diff --git a/codex-rs/state/src/telemetry.rs b/codex-rs/state/src/telemetry.rs index d9692b1943..e9acd99632 100644 --- a/codex-rs/state/src/telemetry.rs +++ b/codex-rs/state/src/telemetry.rs @@ -3,6 +3,7 @@ use std::sync::Arc; use std::sync::OnceLock; use std::time::Duration; +use crate::DB_CORRUPTION_METRIC; use crate::DB_FALLBACK_METRIC; use crate::DB_INIT_DURATION_METRIC; use crate::DB_INIT_METRIC; @@ -17,7 +18,7 @@ use crate::runtime::reclamation::ReclamationPass; use tracing::debug; -/// Low-cardinality sink for SQLite startup, fallback, log-write, and reclamation telemetry. +/// Low-cardinality sink for SQLite startup, corruption, fallback, log-write, and reclamation telemetry. /// /// Implementations should absorb delivery failures locally. Database behavior /// must not depend on whether telemetry export succeeds. @@ -68,6 +69,14 @@ impl DbKind { } } +pub(crate) fn record_corruption(telemetry: Option<&dyn DbTelemetry>, db: Option) { + record_counter( + telemetry, + DB_CORRUPTION_METRIC, + &[("db", db.map_or("other", DbKind::as_str))], + ); +} + pub(crate) fn record_init_result( telemetry: Option<&dyn DbTelemetry>, db: DbKind, @@ -222,7 +231,11 @@ pub(crate) fn classify_error(err: &anyhow::Error) -> &'static str { return "io"; } } - "unknown" + if crate::is_sqlite_corruption_error(err) { + "corrupt" + } else { + "unknown" + } } fn classify_sqlx_error(err: &sqlx::Error) -> &'static str { diff --git a/codex-rs/tui/src/chatwidget/tests/app_server.rs b/codex-rs/tui/src/chatwidget/tests/app_server.rs index b19aa7b8c5..e6b7b1d2b8 100644 --- a/codex-rs/tui/src/chatwidget/tests/app_server.rs +++ b/codex-rs/tui/src/chatwidget/tests/app_server.rs @@ -1120,6 +1120,28 @@ async fn config_warning_during_turn_retains_transcript_details() { ); } +#[tokio::test] +async fn sqlite_recovery_warning_shows_backup_and_metadata_limitations() { + let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await; + chat.handle_server_notification( + ServerNotification::ConfigWarning(ConfigWarningNotification { + summary: "Codex rebuilt its local database".into(), + details: Some("Damaged local databases were rebuilt. Saved conversations remain in rollout files and can restore the thread list and history. Some database-only metadata may be unavailable. The original database files were preserved at the backup locations below.\n\nDatabase path: /codex/state_5.sqlite\nBackup folder: /codex/db-backups/recovery".into()), + path: None, + range: None, + }), + /*replay_kind*/ None, + ); + let cells = drain_insert_history_transcript(&mut rx); + insta::assert_snapshot!( + "sqlite_recovery_warning", + cells + .iter() + .map(|lines| lines_to_single_string(lines)) + .collect::() + ); +} + #[tokio::test] async fn startup_config_warning_is_not_repeated_by_thread() { let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await; diff --git a/codex-rs/tui/src/chatwidget/tests/snapshots/codex_tui__chatwidget__tests__app_server__sqlite_recovery_warning.snap b/codex-rs/tui/src/chatwidget/tests/snapshots/codex_tui__chatwidget__tests__app_server__sqlite_recovery_warning.snap new file mode 100644 index 0000000000..e635743447 --- /dev/null +++ b/codex-rs/tui/src/chatwidget/tests/snapshots/codex_tui__chatwidget__tests__app_server__sqlite_recovery_warning.snap @@ -0,0 +1,10 @@ +--- +source: tui/src/chatwidget/tests/app_server.rs +expression: "cells.iter().map(|lines| lines_to_single_string(lines)).collect::()" +--- +⚠ Codex rebuilt its local database: Damaged local databases were rebuilt. Saved + conversations remain in rollout files and can restore the thread list and + history. Some database-only metadata may be unavailable. The original database + files were preserved at the backup locations below. + Database path: /codex/state_5.sqlite + Backup folder: /codex/db-backups/recovery