Preserve SQLite vacuum modes and surface pool initialization errors (#49102)

## Why

Changing an existing database from `FULL` to `INCREMENTAL` auto-vacuum requires a writer lock and can block pool initialization. SQLx also retries connection initialization failures, masking the original error as `PoolTimedOut`.

## What changed

Preserve existing auto-vacuum modes when opening writable pools, while continuing to initialize empty databases with incremental auto-vacuum. Return the first connection initialization error directly to the caller, retaining SQLx's normal error handling for later lazy connections.

## Testing

Add regression tests for fresh database settings, opening an existing `FULL` auto-vacuum database under a write lock, and preserving the WAL conversion lock error while allowing a subsequent open to succeed after the lock is released.

GitOrigin-RevId: b586c68e43a5a0d3b47268378716919d01b749cf

codex-rs/state/src/migrations_tests.rs

@@ -1025,7 +1025,7 @@ async fn writable_pool_reads_do_not_wait_for_an_existing_writer() {
 
     for (index, (setup, expected_mode)) in [
         ("PRAGMA auto_vacuum = INCREMENTAL", 2_i64),
-        ("PRAGMA auto_vacuum = FULL", 2),
+        ("PRAGMA auto_vacuum = FULL", 1),
         ("PRAGMA auto_vacuum = NONE; VACUUM", 0),
     ]
     .into_iter()
@@ -1048,7 +1048,7 @@ async fn writable_pool_reads_do_not_wait_for_an_existing_writer() {
             .unwrap();
         pool.close().await;
 
-        // Reopening converts FULL, but preserves legacy NONE without VACUUM.
+        // Reopening preserves existing FULL and legacy NONE without taking a writer lock.
         let pool = sqlite.open_read_write_pool(&path).await.unwrap();
         let mut connection = pool.acquire().await.unwrap();
         let writer = connection.begin_with("BEGIN IMMEDIATE").await.unwrap();

codex-rs/state/src/sqlite.rs

@@ -1,4 +1,5 @@
 //! Shared SQLite connection configuration.
+//! Writable pools preserve existing vacuum modes and report initialization errors to callers.
 
 #![expect(
     clippy::disallowed_methods,
@@ -313,38 +314,54 @@ impl SqliteConfig {
             .synchronous(SqliteSynchronous::Normal)
             .busy_timeout(Duration::from_secs(5))
             .log_statements(LevelFilter::Off);
-        SqlitePoolOptions::new()
-            .max_connections(5)
-            .after_connect(|connection, _metadata| {
+        // 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 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 it before WAL creates the first database page. Existing
-                    // NONE databases require VACUUM to convert; leave those alone.
-                    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 mode == 1 || empty {
-                        sqlx::query("PRAGMA auto_vacuum = INCREMENTAL")
-                            .execute(&mut *connection)
+                    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(),
+                    })
                 })
-            })
-            .connect_with(options)
-            .await
+            },
+        );
+        tokio::select! {
+            biased;
+            Some(error) = init_error_rx.recv() => Err(error),
+            result = pool_options.connect_with(options) => result,
+        }
     }
 
     /// Open an existing Codex SQLite database without creating or modifying it.
@@ -367,3 +384,7 @@ impl SqliteConfig {
             .await
     }
 }
+
+#[cfg(test)]
+#[path = "sqlite_tests.rs"]
+mod tests;

codex-rs/state/src/sqlite_tests.rs

@@ -0,0 +1,123 @@
+//! Regression coverage for SQLite initialization under competing database locks.
+
+use super::SqliteConfig;
+use codex_utils_absolute_path::test_support::PathExt;
+use pretty_assertions::assert_eq;
+use sqlx::sqlite::SqliteAutoVacuum;
+use sqlx::sqlite::SqliteConnectOptions;
+use sqlx::sqlite::SqliteJournalMode;
+use sqlx::sqlite::SqlitePoolOptions;
+
+#[tokio::test]
+async fn open_read_write_pool_initializes_fresh_database_settings() -> anyhow::Result<()> {
+    let sqlite_home = crate::runtime::test_support::unique_temp_dir();
+    tokio::fs::create_dir_all(&sqlite_home).await?;
+    let _cleanup = scopeguard::guard(sqlite_home.clone(), |sqlite_home| {
+        let _ = std::fs::remove_dir_all(sqlite_home);
+    });
+    let sqlite = SqliteConfig::new_for_testing(sqlite_home.as_path().abs());
+    let pool = sqlite.open_read_write_pool(&sqlite.logs_db_path()).await?;
+
+    let journal_mode = sqlx::query_scalar::<_, String>("PRAGMA journal_mode")
+        .fetch_one(&pool)
+        .await?;
+    let auto_vacuum = sqlx::query_scalar::<_, i64>("PRAGMA auto_vacuum")
+        .fetch_one(&pool)
+        .await?;
+
+    assert_eq!(
+        (journal_mode, auto_vacuum),
+        ("wal".to_string(), SqliteAutoVacuum::Incremental as i64)
+    );
+    pool.close().await;
+    Ok(())
+}
+
+#[tokio::test]
+async fn open_read_write_pool_preserves_existing_settings_under_write_lock() -> anyhow::Result<()> {
+    let sqlite_home = crate::runtime::test_support::unique_temp_dir();
+    tokio::fs::create_dir_all(&sqlite_home).await?;
+    let _cleanup = scopeguard::guard(sqlite_home.clone(), |sqlite_home| {
+        let _ = std::fs::remove_dir_all(sqlite_home);
+    });
+    let sqlite = SqliteConfig::new_for_testing(sqlite_home.as_path().abs());
+    let database_path = sqlite.logs_db_path();
+    let existing_pool = SqlitePoolOptions::new()
+        .max_connections(1)
+        .connect_with(
+            SqliteConnectOptions::new()
+                .filename(&database_path)
+                .create_if_missing(true)
+                .auto_vacuum(SqliteAutoVacuum::Full)
+                .journal_mode(SqliteJournalMode::Wal),
+        )
+        .await?;
+    sqlx::query("CREATE TABLE existing (id INTEGER PRIMARY KEY)")
+        .execute(&existing_pool)
+        .await?;
+
+    let mut lock_holder = existing_pool.acquire().await?;
+    sqlx::query("BEGIN IMMEDIATE")
+        .execute(&mut *lock_holder)
+        .await?;
+
+    let pool = sqlite.open_read_write_pool(&database_path).await?;
+    let auto_vacuum = sqlx::query_scalar::<_, i64>("PRAGMA auto_vacuum")
+        .fetch_one(&pool)
+        .await?;
+
+    assert_eq!(auto_vacuum, SqliteAutoVacuum::Full as i64);
+    sqlx::query("ROLLBACK").execute(&mut *lock_holder).await?;
+    drop(lock_holder);
+    pool.close().await;
+    existing_pool.close().await;
+    Ok(())
+}
+
+#[tokio::test]
+async fn open_read_write_pool_preserves_wal_conversion_lock_error() -> anyhow::Result<()> {
+    let sqlite_home = crate::runtime::test_support::unique_temp_dir();
+    tokio::fs::create_dir_all(&sqlite_home).await?;
+    let _cleanup = scopeguard::guard(sqlite_home.clone(), |sqlite_home| {
+        let _ = std::fs::remove_dir_all(sqlite_home);
+    });
+    let sqlite = SqliteConfig::new_for_testing(sqlite_home.as_path().abs());
+    let database_path = sqlite.logs_db_path();
+    let existing_pool = SqlitePoolOptions::new()
+        .max_connections(1)
+        .connect_with(
+            SqliteConnectOptions::new()
+                .filename(&database_path)
+                .create_if_missing(true)
+                .journal_mode(SqliteJournalMode::Delete),
+        )
+        .await?;
+    sqlx::query("CREATE TABLE existing (id INTEGER PRIMARY KEY)")
+        .execute(&existing_pool)
+        .await?;
+    let mut lock_holder = existing_pool.acquire().await?;
+    sqlx::query("BEGIN IMMEDIATE")
+        .execute(&mut *lock_holder)
+        .await?;
+
+    let result = sqlite.open_read_write_pool(&database_path).await;
+    sqlx::query("ROLLBACK").execute(&mut *lock_holder).await?;
+    drop(lock_holder);
+    existing_pool.close().await;
+
+    let error = result.expect_err("WAL conversion requires the writer lock");
+    assert_eq!(
+        error
+            .as_database_error()
+            .and_then(sqlx::error::DatabaseError::code),
+        Some("5".into())
+    );
+    assert!(crate::sqlite_error_detail_is_lock(&error.to_string()));
+    // Failure must release the connection so a later startup can succeed.
+    sqlite
+        .open_read_write_pool(&database_path)
+        .await?
+        .close()
+        .await;
+    Ok(())
+}