From 40eeb6e8a89ef421c25d4c40e06fa1d40ce66b4f Mon Sep 17 00:00:00 2001 From: jif Date: Mon, 21 Sep 2026 14:12:16 +0000 Subject: [PATCH] Recover corrupt message-board storage during thread deletion (#47042) ## Why A corrupt agent message-board database can block thread deletion even when the feature is disabled. ## What changed When corruption prevents opening the database for board deletion, back it up and recreate it before recording deletion tombstones. Recovery resets all boards, retaining their previous storage in the backup. Other database open failures continue to return an error without deleting the thread or replacing storage. ## Testing Add an app-server regression test with the feature disabled that verifies ordinary open failures preserve storage and the thread, while corruption recovery preserves the backup, deletes the requested thread, retains another thread, and records the board tombstone in the recreated database. GitOrigin-RevId: 85f7d6303ae5c359782018cad4a78461d86c43db --- .../tests/suite/v2/thread_delete.rs | 79 +++++++++++++++++++ .../src/local/lifecycle.rs | 25 +++++- 2 files changed, 100 insertions(+), 4 deletions(-) diff --git a/codex-rs/app-server/tests/suite/v2/thread_delete.rs b/codex-rs/app-server/tests/suite/v2/thread_delete.rs index f5ed2b139f..e3c2a354dd 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_delete.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_delete.rs @@ -423,6 +423,85 @@ async fn thread_delete_removes_persisted_board_even_with_feature_disabled() -> R Ok(()) } +#[tokio::test] +async fn thread_delete_recovers_only_corrupt_board_with_feature_disabled() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; + let home = TempDir::new()?; + MockResponsesConfig::new(&server.uri()) + .disable_feature(Feature::AgentMessageBoard) + .write(home.path())?; + let thread_id = create_delete_test_rollout(home.path(), /*minute*/ 0, "Delete me")?; + let retained_id = create_delete_test_rollout(home.path(), /*minute*/ 1, "Keep me")?; + let corrupt = b"not a SQLite database"; + let database = home.path().join("agent_message_board_1.sqlite"); + std::fs::create_dir(&database)?; + let mut app = TestAppServer::builder() + .with_codex_home(home.path()) + .build_initialized() + .await?; + + // An ordinary open failure must leave both storage and the thread intact. + let request_id = app + .send_thread_delete_request(ThreadDeleteParams { + thread_id: thread_id.clone(), + }) + .await?; + let error = timeout( + DEFAULT_READ_TIMEOUT, + app.read_stream_until_error_message(RequestId::Integer(request_id)), + ) + .await??; + assert_eq!(error.error.code, -32603); + assert!(database.is_dir()); + assert!(!home.path().join("db-backups").exists()); + assert!( + find_thread_path_by_id_str(home.path(), &thread_id, /*state_db_ctx*/ None) + .await? + .is_some() + ); + std::fs::remove_dir(&database)?; + std::fs::write(&database, corrupt)?; + + let _: ThreadDeleteResponse = app + .request(|request_id| ClientRequest::ThreadDelete { + request_id, + params: ThreadDeleteParams { + thread_id: thread_id.clone(), + }, + }) + .await?; + assert_eq!( + find_thread_path_by_id_str(home.path(), &thread_id, /*state_db_ctx*/ None).await?, + None + ); + assert!( + find_thread_path_by_id_str(home.path(), &retained_id, /*state_db_ctx*/ None) + .await? + .is_some() + ); + let backups = + std::fs::read_dir(home.path().join("db-backups"))?.collect::>>()?; + assert_eq!(backups.len(), 1); + assert_eq!( + std::fs::read(backups[0].path().join("agent_message_board_1.sqlite"))?, + corrupt + ); + let sqlite = SqliteConfig::new_for_testing(home.path().abs()); + let pool = sqlite + .open_read_only_pool(&database, /*busy_timeout*/ None) + .await?; + assert_eq!(board_counts(&pool).await?, (0, 0, 0)); + assert_eq!( + sqlx::query_scalar::<_, String>("SELECT board FROM deleted_boards") + .fetch_all(&pool) + .await?, + vec![thread_id] + ); + pool.close().await; + app.shutdown_gracefully().await?; + Ok(()) +} + async fn board_counts(pool: &sqlx::SqlitePool) -> Result<(i64, i64, i64)> { Ok(sqlx::query_as( "SELECT (SELECT COUNT(*) FROM channels), 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 45779c875c..0be62986d8 100644 --- a/codex-rs/ext/agent-message-board/src/local/lifecycle.rs +++ b/codex-rs/ext/agent-message-board/src/local/lifecycle.rs @@ -1,6 +1,7 @@ //! Permanent board deletion, serialized with writes across all open handles. //! //! Keep only a root ID tombstone so delayed writes cannot resurrect deleted data. +//! If corruption prevents opening the database, back it up before recreating it. use super::DATABASE_FILE; use super::LocalAgentMessageBoard; @@ -17,15 +18,31 @@ impl LocalAgentMessageBoard { /// Permanently removes boards owned by these roots, including their posts and subscriptions. /// A child's ID does not match its parent's board. Unload and archive must not call this. /// Safe to retry and independent of whether the feature is currently enabled. + /// Recovering an unopenable corrupt database resets all boards, retaining a backup. pub async fn delete_boards(sqlite: &SqliteConfig, roots: &[SessionId]) -> Result<()> { let path = sqlite.home().join(DATABASE_FILE); if roots.is_empty() || !tokio::fs::try_exists(&path).await? { return Ok(()); } - let pool = sqlite - .open_read_write_pool(&path) - .await - .map_err(storage_error)?; + 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)); + } + let backups = codex_state::backup_runtime_db_for_fresh_start(&path).await?; + tracing::warn!( + database = %path.display(), + ?backups, + "Backed up corrupt agent message-board storage; recreating it for thread deletion. Existing data for all boards is unavailable." + ); + sqlite + .open_read_write_pool(&path) + .await + .map_err(storage_error)? + } + }; // Also handles databases created before permanent deletion was supported. sqlx::raw_sql(SCHEMA) .execute(&pool)