diff --git a/codex-rs/state/src/migrations_tests.rs b/codex-rs/state/src/migrations_tests.rs index 036b8ed855..4f07d1c11b 100644 --- a/codex-rs/state/src/migrations_tests.rs +++ b/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(); diff --git a/codex-rs/state/src/sqlite.rs b/codex-rs/state/src/sqlite.rs index 6487395c33..4d1c9e9819 100644 --- a/codex-rs/state/src/sqlite.rs +++ b/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; diff --git a/codex-rs/state/src/sqlite_tests.rs b/codex-rs/state/src/sqlite_tests.rs new file mode 100644 index 0000000000..d9372dd3a4 --- /dev/null +++ b/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(()) +}