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