fix: Auto-recover from corrupted sqlite databases (#26859)

Further investigation of the sqlite incidents showed that the problems
are due to corruption from the older version of SQLite that we recently
upgraded, and that the data is truly corrupted in the root database --
recovery of all data is not possible. Given that the data is
reconstructable from the rollouts on disk, we should just auto-backup
the database and let codex rebuild the rollout info from the disk
rollouts.

The new behavior is that appserver auto-backs-up and rebuilds (with logs
reflecting that behavior). The CLI now pops a message letting you know
this happened and the paths of the backed-up corrupt db and the new
database. There is also context added so that the desktop app can read
the rebuild info from it and inform the user with it.

codex-rs/state/src/lib.rs

@@ -57,15 +57,21 @@ pub use runtime::GoalStore;
 pub use runtime::GoalUpdate;
 pub use runtime::MemoryStore;
 pub use runtime::RemoteControlEnrollmentRecord;
+pub use runtime::RuntimeDbBackup;
 pub use runtime::RuntimeDbPath;
 pub use runtime::ThreadFilterOptions;
+pub use runtime::backup_runtime_db_for_fresh_start;
 pub use runtime::goals_db_filename;
 pub use runtime::goals_db_path;
+pub use runtime::is_sqlite_corruption_error;
 pub use runtime::logs_db_filename;
 pub use runtime::logs_db_path;
 pub use runtime::memories_db_filename;
 pub use runtime::memories_db_path;
+pub use runtime::runtime_db_path_for_corruption_error;
 pub use runtime::runtime_db_paths;
+pub use runtime::sqlite_error_detail_is_corruption;
+pub use runtime::sqlite_error_detail_is_lock;
 pub use runtime::sqlite_integrity_check;
 pub use runtime::state_db_filename;
 pub use runtime::state_db_path;

codex-rs/state/src/runtime.rs

@@ -62,6 +62,7 @@ mod backfill;
 mod goals;
 mod logs;
 mod memories;
+mod recovery;
 mod remote_control;
 #[cfg(test)]
 mod test_support;
@@ -72,6 +73,12 @@ pub use goals::GoalAccountingOutcome;
 pub use goals::GoalStore;
 pub use goals::GoalUpdate;
 pub use memories::MemoryStore;
+pub use recovery::RuntimeDbBackup;
+pub use recovery::backup_runtime_db_for_fresh_start;
+pub use recovery::is_sqlite_corruption_error;
+pub use recovery::runtime_db_path_for_corruption_error;
+pub use recovery::sqlite_error_detail_is_corruption;
+pub use recovery::sqlite_error_detail_is_lock;
 pub use remote_control::RemoteControlEnrollmentRecord;
 pub use threads::ThreadFilterOptions;
 
@@ -200,6 +207,7 @@ impl StateRuntime {
             Ok(db) => Arc::new(db),
             Err(err) => {
                 warn!("failed to open logs db at {}: {err}", logs_path.display());
+                close_sqlite_pools(&[pool.as_ref()]).await;
                 return Err(err);
             }
         };
@@ -208,6 +216,7 @@ impl StateRuntime {
                 Ok(db) => Arc::new(db),
                 Err(err) => {
                     warn!("failed to open goals db at {}: {err}", goals_path.display());
+                    close_sqlite_pools(&[pool.as_ref(), logs_pool.as_ref()]).await;
                     return Err(err);
                 }
             };
@@ -224,6 +233,7 @@ impl StateRuntime {
                     "failed to open memories db at {}: {err}",
                     memories_path.display()
                 );
+                close_sqlite_pools(&[pool.as_ref(), logs_pool.as_ref(), goals_pool.as_ref()]).await;
                 return Err(err);
             }
         };
@@ -236,7 +246,16 @@ impl StateRuntime {
             started.elapsed(),
             &backfill_state_result,
         );
-        backfill_state_result?;
+        if let Err(err) = backfill_state_result {
+            close_sqlite_pools(&[
+                pool.as_ref(),
+                logs_pool.as_ref(),
+                goals_pool.as_ref(),
+                memories_pool.as_ref(),
+            ])
+            .await;
+            return Err(err);
+        }
         let started = Instant::now();
         let thread_updated_at_millis_result: anyhow::Result<Option<i64>> =
             sqlx::query_scalar("SELECT MAX(threads.updated_at_ms) FROM threads")
@@ -250,7 +269,19 @@ impl StateRuntime {
             started.elapsed(),
             &thread_updated_at_millis_result,
         );
-        let thread_updated_at_millis = thread_updated_at_millis_result?;
+        let thread_updated_at_millis = match thread_updated_at_millis_result {
+            Ok(value) => value,
+            Err(err) => {
+                close_sqlite_pools(&[
+                    pool.as_ref(),
+                    logs_pool.as_ref(),
+                    goals_pool.as_ref(),
+                    memories_pool.as_ref(),
+                ])
+                .await;
+                return Err(err);
+            }
+        };
         let thread_updated_at_millis = thread_updated_at_millis.unwrap_or(0);
         let runtime = Arc::new(Self {
             thread_goals: GoalStore::new(Arc::clone(&goals_pool)),
@@ -283,6 +314,14 @@ impl StateRuntime {
         &self.memories
     }
 
+    /// Close all SQLite pools and wait for outstanding pool workers to exit.
+    pub async fn close(&self) {
+        self.memories.close().await;
+        self.thread_goals.close().await;
+        self.logs_pool.close().await;
+        self.pool.close().await;
+    }
+
     pub async fn clear_memory_data_in_sqlite_home(sqlite_home: &Path) -> anyhow::Result<bool> {
         let memories_path = MEMORIES_DB.path(sqlite_home);
         if !tokio::fs::try_exists(&memories_path).await? {
@@ -302,6 +341,12 @@ impl StateRuntime {
     }
 }
 
+async fn close_sqlite_pools(pools: &[&SqlitePool]) {
+    for pool in pools {
+        pool.close().await;
+    }
+}
+
 fn base_sqlite_options(path: &Path) -> SqliteConnectOptions {
     SqliteConnectOptions::new()
         .filename(path)
@@ -367,7 +412,8 @@ async fn open_sqlite(
         started.elapsed(),
         &pool_result,
     );
-    let pool = pool_result?;
+    let pool = pool_result
+        .map_err(|source| recovery::RuntimeDbInitError::new(spec.label, "open", path, source))?;
     let started = Instant::now();
     let migrate_result = migrator.run(&pool).await.map_err(anyhow::Error::from);
     crate::telemetry::record_init_result(
@@ -377,7 +423,10 @@ async fn open_sqlite(
         started.elapsed(),
         &migrate_result,
     );
-    migrate_result?;
+    if let Err(source) = migrate_result {
+        pool.close().await;
+        return Err(recovery::RuntimeDbInitError::new(spec.label, "migrate", path, source).into());
+    }
     Ok(pool)
 }
 

codex-rs/state/src/runtime/goals.rs

@@ -11,6 +11,10 @@ impl GoalStore {
     pub(crate) fn new(pool: Arc<SqlitePool>) -> Self {
         Self { pool }
     }
+
+    pub(crate) async fn close(&self) {
+        self.pool.close().await;
+    }
 }
 
 pub struct GoalUpdate {

codex-rs/state/src/runtime/memories.rs

@@ -35,6 +35,10 @@ impl MemoryStore {
         Self { pool, state_pool }
     }
 
+    pub(crate) async fn close(&self) {
+        self.pool.close().await;
+    }
+
     /// Deletes all persisted memory state in one transaction.
     ///
     /// This removes every `stage1_outputs` row and all `jobs` rows for the

codex-rs/state/src/runtime/recovery.rs

@@ -0,0 +1,243 @@
+//! Backup-and-rebuild support for Codex runtime SQLite databases.
+//!
+//! 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.
+
+use std::borrow::Cow;
+use std::path::Path;
+use std::path::PathBuf;
+
+const BACKUP_DIR_NAME: &str = "db-backups";
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct RuntimeDbBackup {
+    /// Path where the runtime database or sidecar lived before it was moved.
+    pub original_path: PathBuf,
+    /// Path where the runtime database or sidecar was backed up.
+    pub backup_path: PathBuf,
+}
+
+#[derive(Debug)]
+pub(crate) struct RuntimeDbInitError {
+    label: &'static str,
+    operation: &'static str,
+    path: PathBuf,
+    source: anyhow::Error,
+}
+
+impl RuntimeDbInitError {
+    pub(crate) fn new(
+        label: &'static str,
+        operation: &'static str,
+        path: &Path,
+        source: anyhow::Error,
+    ) -> Self {
+        Self {
+            label,
+            operation,
+            path: path.to_path_buf(),
+            source,
+        }
+    }
+
+    fn path(&self) -> &Path {
+        self.path.as_path()
+    }
+}
+
+impl std::fmt::Display for RuntimeDbInitError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "failed to {} {} at {}: {}",
+            self.operation,
+            self.label,
+            self.path.display(),
+            self.source
+        )
+    }
+}
+
+impl std::error::Error for RuntimeDbInitError {
+    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+        Some(self.source.as_ref())
+    }
+}
+
+/// Move one Codex runtime SQLite database out of the way so that database can
+/// be recreated without discarding unrelated runtime databases.
+pub async fn backup_runtime_db_for_fresh_start(
+    db_path: &Path,
+) -> std::io::Result<Vec<RuntimeDbBackup>> {
+    let sqlite_home = db_path.parent().ok_or_else(|| {
+        std::io::Error::other(format!(
+            "database path does not have a parent directory: {}",
+            db_path.display()
+        ))
+    })?;
+    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 => {
+            tokio::fs::create_dir_all(sqlite_home).await?;
+            Err(std::io::Error::other(format!(
+                "no Codex runtime database files were found to back up for {}",
+                db_path.display()
+            )))
+        }
+        Err(err) => Err(err),
+    }
+}
+
+pub fn runtime_db_path_for_corruption_error(err: &anyhow::Error) -> Option<PathBuf> {
+    if !is_sqlite_corruption_error(err) {
+        return None;
+    }
+    err.chain()
+        .find_map(|source| source.downcast_ref::<RuntimeDbInitError>())
+        .map(|err| err.path().to_path_buf())
+}
+
+pub fn is_sqlite_corruption_error(err: &anyhow::Error) -> bool {
+    err.chain().any(sqlite_error_source_is_corruption)
+}
+
+fn sqlite_error_source_is_corruption(source: &(dyn std::error::Error + 'static)) -> bool {
+    let Some(err) = source.downcast_ref::<sqlx::Error>() else {
+        return false;
+    };
+    let sqlx::Error::Database(database_error) = err else {
+        return false;
+    };
+    sqlite_error_detail_is_corruption(database_error.message())
+        || database_error
+            .code()
+            .is_some_and(sqlite_database_code_is_corruption)
+}
+
+fn sqlite_database_code_is_corruption(code: Cow<'_, str>) -> bool {
+    matches!(
+        code.as_ref().to_ascii_lowercase().as_str(),
+        "11" | "26" | "sqlite_corrupt" | "sqlite_notadb"
+    )
+}
+
+pub fn sqlite_error_detail_is_corruption(detail: &str) -> bool {
+    let detail = detail.to_ascii_lowercase();
+    detail.contains("database disk image is malformed")
+        || detail.contains("database schema is malformed")
+        || detail.contains("database is corrupt")
+        || detail.contains("file is not a database")
+        || detail.contains("sqlite_corrupt")
+        || detail.contains("sqlite_notadb")
+        || detail.contains("(code: 11)")
+        || detail.contains("(code: 26)")
+}
+
+pub fn sqlite_error_detail_is_lock(detail: &str) -> bool {
+    let detail = detail.to_ascii_lowercase();
+    detail.contains("database is locked") || detail.contains("database is busy")
+}
+
+async fn backup_runtime_db_files(db_path: &Path) -> std::io::Result<Vec<RuntimeDbBackup>> {
+    let sqlite_home = db_path.parent().ok_or_else(|| {
+        std::io::Error::other(format!(
+            "database path does not have a parent directory: {}",
+            db_path.display()
+        ))
+    })?;
+    backup_sqlite_paths(sqlite_home, sqlite_paths(db_path)).await
+}
+
+async fn backup_sqlite_paths(
+    sqlite_home: &Path,
+    paths: impl IntoIterator<Item = PathBuf>,
+) -> std::io::Result<Vec<RuntimeDbBackup>> {
+    let backup_dir = create_unique_backup_dir(sqlite_home.join(BACKUP_DIR_NAME).as_path()).await?;
+    let mut backups = Vec::new();
+
+    for path in paths {
+        if tokio::fs::try_exists(path.as_path()).await? {
+            let backup_path = backup_dir.join(file_name(path.as_path())?);
+            tokio::fs::rename(path.as_path(), backup_path.as_path()).await?;
+            backups.push(RuntimeDbBackup {
+                original_path: path,
+                backup_path,
+            });
+        }
+    }
+
+    if backups.is_empty() {
+        let _ = tokio::fs::remove_dir(backup_dir).await;
+        return Err(std::io::Error::other(
+            "no Codex runtime database files were found to back up",
+        ));
+    }
+
+    Ok(backups)
+}
+
+async fn backup_blocking_sqlite_home(sqlite_home: &Path) -> std::io::Result<Vec<RuntimeDbBackup>> {
+    let parent = sqlite_home.parent().ok_or_else(|| {
+        std::io::Error::other(format!(
+            "cannot create a backup folder for {}",
+            sqlite_home.display()
+        ))
+    })?;
+    let mut backup_dir_name = file_name(sqlite_home)?.to_os_string();
+    backup_dir_name.push(format!(".{BACKUP_DIR_NAME}"));
+    let backup_parent = parent.join(backup_dir_name);
+    let backup_dir = create_unique_backup_dir(backup_parent.as_path()).await?;
+    let backup_path = backup_dir.join(file_name(sqlite_home)?);
+    tokio::fs::rename(sqlite_home, backup_path.as_path()).await?;
+    tokio::fs::create_dir_all(sqlite_home).await?;
+    Ok(vec![RuntimeDbBackup {
+        original_path: sqlite_home.to_path_buf(),
+        backup_path,
+    }])
+}
+
+fn sqlite_paths(db_path: &Path) -> Vec<PathBuf> {
+    let mut wal_path = db_path.as_os_str().to_os_string();
+    wal_path.push("-wal");
+    let mut shm_path = db_path.as_os_str().to_os_string();
+    shm_path.push("-shm");
+    vec![
+        db_path.to_path_buf(),
+        PathBuf::from(wal_path),
+        PathBuf::from(shm_path),
+    ]
+}
+
+async fn create_unique_backup_dir(backup_parent: &Path) -> std::io::Result<PathBuf> {
+    tokio::fs::create_dir_all(backup_parent).await?;
+    let timestamp = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map_or(0, |duration| duration.as_secs());
+    let mut sequence = 0_u32;
+    loop {
+        let backup_dir = backup_parent.join(format!("sqlite-{timestamp}-{sequence}"));
+        match tokio::fs::create_dir(backup_dir.as_path()).await {
+            Ok(()) => return Ok(backup_dir),
+            Err(err) if err.kind() == std::io::ErrorKind::AlreadyExists => {
+                sequence += 1;
+            }
+            Err(err) => return Err(err),
+        }
+    }
+}
+
+fn file_name(path: &Path) -> std::io::Result<&std::ffi::OsStr> {
+    path.file_name().ok_or_else(|| {
+        std::io::Error::other(format!(
+            "cannot create a backup name for {}",
+            path.display()
+        ))
+    })
+}
+
+#[cfg(test)]
+#[path = "recovery_tests.rs"]
+mod tests;

codex-rs/state/src/runtime/recovery_tests.rs

@@ -0,0 +1,105 @@
+use super::*;
+use crate::runtime::test_support::unique_temp_dir;
+use pretty_assertions::assert_eq;
+
+#[tokio::test]
+async fn backup_moves_only_requested_runtime_db_files_to_backup_folder() -> std::io::Result<()> {
+    let sqlite_home = unique_temp_dir();
+    tokio::fs::create_dir_all(sqlite_home.as_path()).await?;
+    let runtime_paths = super::super::runtime_db_paths(sqlite_home.as_path());
+    let mut expected_paths = Vec::new();
+    for db_path in runtime_paths.iter().map(|db| db.path.as_path()) {
+        for path in sqlite_paths(db_path) {
+            tokio::fs::write(path.as_path(), path.display().to_string()).await?;
+            expected_paths.push(path);
+        }
+    }
+    let failed_db_path = super::super::logs_db_path(sqlite_home.as_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?;
+
+    assert_eq!(backups.len(), failed_paths.len());
+    for path in &failed_paths {
+        assert!(!tokio::fs::try_exists(path.as_path()).await?);
+    }
+    for path in expected_paths
+        .iter()
+        .filter(|path| !failed_paths.contains(path))
+    {
+        assert!(tokio::fs::try_exists(path.as_path()).await?);
+    }
+    for backup in backups {
+        assert!(
+            backup
+                .backup_path
+                .starts_with(sqlite_home.join(BACKUP_DIR_NAME))
+        );
+        assert!(tokio::fs::try_exists(backup.backup_path.as_path()).await?);
+    }
+    Ok(())
+}
+
+#[tokio::test]
+async fn backup_replaces_blocking_sqlite_home_file() -> std::io::Result<()> {
+    let temp_dir = unique_temp_dir();
+    tokio::fs::create_dir_all(temp_dir.as_path()).await?;
+    let sqlite_home = temp_dir.join("sqlite-home");
+    tokio::fs::write(sqlite_home.as_path(), b"not-a-directory").await?;
+
+    let backups = backup_runtime_db_for_fresh_start(
+        super::super::state_db_path(sqlite_home.as_path()).as_path(),
+    )
+    .await?;
+
+    assert_eq!(backups.len(), 1);
+    assert!(tokio::fs::metadata(sqlite_home.as_path()).await?.is_dir());
+    assert!(
+        backups[0]
+            .backup_path
+            .starts_with(temp_dir.join(format!("sqlite-home.{BACKUP_DIR_NAME}")))
+    );
+    assert!(tokio::fs::try_exists(backups[0].backup_path.as_path()).await?);
+    Ok(())
+}
+
+#[test]
+fn sqlite_error_detail_classifies_corruption_and_lock_errors() {
+    assert!(sqlite_error_detail_is_corruption("file is not a database"));
+    assert!(sqlite_error_detail_is_corruption(
+        "error returned from database: (code: 11) database disk image is malformed"
+    ));
+    assert!(!sqlite_error_detail_is_corruption("database is locked"));
+    assert!(sqlite_error_detail_is_lock("database is locked"));
+    assert!(sqlite_error_detail_is_lock("database is busy"));
+}
+
+#[tokio::test]
+async fn runtime_db_path_for_corruption_error_returns_failed_database_path() -> std::io::Result<()>
+{
+    let sqlite_home = unique_temp_dir();
+    tokio::fs::create_dir_all(sqlite_home.as_path()).await?;
+    let path = super::super::state_db_path(sqlite_home.as_path());
+    tokio::fs::write(path.as_path(), b"not sqlite").await?;
+
+    let err = match super::super::StateRuntime::init(sqlite_home, "openai".to_string()).await {
+        Ok(_) => panic!("malformed sqlite should fail to initialize"),
+        Err(err) => err,
+    };
+
+    assert_eq!(runtime_db_path_for_corruption_error(&err), Some(path));
+    Ok(())
+}
+
+#[test]
+fn runtime_db_path_for_corruption_error_ignores_corrupt_word_in_path() {
+    let path = PathBuf::from("/tmp/sqlite_corrupt/state_5.sqlite");
+    let err = anyhow::Error::new(RuntimeDbInitError::new(
+        "state DB",
+        "open",
+        path.as_path(),
+        anyhow::anyhow!("permission denied"),
+    ));
+
+    assert_eq!(runtime_db_path_for_corruption_error(&err), None);
+}