diff --git a/codex-rs/state/src/runtime/reclamation.rs b/codex-rs/state/src/runtime/reclamation.rs index 29c46e4fb1..27f89d612a 100644 --- a/codex-rs/state/src/runtime/reclamation.rs +++ b/codex-rs/state/src/runtime/reclamation.rs @@ -291,15 +291,16 @@ async fn reclaim_pages( .is_none_or(|deadline| Instant::now() < deadline) }); - let (mode, page_size, pages, free): (i64, i64, i64, i64) = sqlx::query_as( - "SELECT auto_vacuum, page_size, page_count, freelist_count \ - FROM pragma_auto_vacuum, pragma_page_size, pragma_page_count, pragma_freelist_count", + let (journal, mode, page_size, pages, free): (String, i64, i64, i64, i64) = sqlx::query_as( + "SELECT journal_mode, auto_vacuum, page_size, page_count, freelist_count \ + FROM pragma_journal_mode, pragma_auto_vacuum, pragma_page_size, pragma_page_count, pragma_freelist_count", ) .fetch_one(&mut *connection) .await?; + // This worker's checkpoint and contention rules apply only to WAL mode. // Require incremental auto-vacuum and at least 64 MiB and 25% free space. - if mode != 2 || free * page_size < MIN_FREE_BYTES || free < pages / 4 { + if journal != "wal" || mode != 2 || free * page_size < MIN_FREE_BYTES || free < pages / 4 { outcome = PassOutcome::Idle; return Ok(()); } diff --git a/codex-rs/state/src/sqlite.rs b/codex-rs/state/src/sqlite.rs index d8c72c6031..b082f2622d 100644 --- a/codex-rs/state/src/sqlite.rs +++ b/codex-rs/state/src/sqlite.rs @@ -1,6 +1,6 @@ //! Shared SQLite connection configuration. -//! Writable pools preserve existing vacuum modes and report initialization errors to callers. -//! External lock contention while opening gets a bounded retry before the original error surfaces. +//! Writable pools preserve the journal and vacuum modes of initialized databases. +//! Startup retries SQLITE_BUSY and reports other errors to callers. #![expect( clippy::disallowed_methods, @@ -364,57 +364,9 @@ impl SqliteConfig { .synchronous(SqliteSynchronous::Normal) .busy_timeout(DEFAULT_BUSY_TIMEOUT) .log_statements(LevelFilter::Off); - 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(()) - } - .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.clone()) => result, - } - }; - let pool = retry_on_busy(|| async { Ok(connect().await?) }).await?; + // Each failed attempt closes its pool before initialization is retried. + let connect = || async { connect_once(&options).await.map_err(anyhow::Error::from) }; + let pool = retry_on_busy(connect).await?; let validation_result = self .quick_check_manager .quick_check_once( @@ -444,7 +396,7 @@ impl SqliteConfig { ); } - return retry_on_busy(|| async { Ok(connect().await?) }).await; + return retry_on_busy(connect).await; } } @@ -472,6 +424,90 @@ impl SqliteConfig { } } +async fn connect_once(options: &SqliteConnectOptions) -> Result { + // 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 journal: String = sqlx::query_scalar("PRAGMA journal_mode") + .fetch_one(&mut *connection) + .await?; + if journal == "delete" { + // If the filesystem honors sync requests, EXTRA syncs the directory + // after SQLite deletes a rollback journal. + sqlx::query("PRAGMA synchronous = EXTRA") + .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(), + }) + }) + }); + let pool = tokio::select! { + biased; + Some(error) = init_error_rx.recv() => Err(error), + result = pool_options.connect_with(options.clone()) => result, + }?; + // Database settings persist on disk. + // Initialize these settings once before callers can use the pool. + // Connections that open later need only settings that apply to each connection. + let initialized: Result<(), Error> = async { + let mut connection = pool.acquire().await?; + // Auto-vacuum alone can write a header to disk. + // If creation stops before the schema exists, startup must initialize the file. + let empty = + sqlx::query_scalar::<_, bool>("SELECT NOT EXISTS (SELECT 1 FROM sqlite_schema)") + .fetch_one(&mut *connection) + .await?; + if empty { + let mode: i64 = sqlx::query_scalar("PRAGMA auto_vacuum") + .fetch_one(&mut *connection) + .await?; + if mode == 0 { + sqlx::query("PRAGMA auto_vacuum = INCREMENTAL") + .execute(&mut *connection) + .await?; + } + sqlx::query("PRAGMA journal_mode = WAL") + .execute(&mut *connection) + .await?; + } + // Another process can initialize WAL after this connection selects EXTRA. + // Apply NORMAL to WAL even if that process already created the schema. + let journal: String = sqlx::query_scalar("PRAGMA journal_mode") + .fetch_one(&mut *connection) + .await?; + if journal == "wal" { + sqlx::query("PRAGMA synchronous = NORMAL") + .execute(&mut *connection) + .await?; + } + // Wait for the connection to return to the pool before startup requests it again. + // An asynchronous drop can otherwise cause the pool to open a second connection. + connection.return_to_pool().await; + Ok(()) + } + .await; + if let Err(error) = initialized { + pool.close().await; + return Err(error); + } + Ok(pool) +} + #[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 index 327031b909..5ebbab9cab 100644 --- a/codex-rs/state/src/sqlite_tests.rs +++ b/codex-rs/state/src/sqlite_tests.rs @@ -10,26 +10,45 @@ 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; + for interrupted in [false, true] { + 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 path = sqlite.logs_db_path(); + if interrupted { + // Auto-vacuum writes a header before WAL setup. + // Startup must initialize the database even if the file is not empty. + let partial = SqlitePoolOptions::new() + .max_connections(/*max*/ 1) + .connect_with( + SqliteConnectOptions::new() + .filename(&path) + .create_if_missing(/*create*/ true) + .auto_vacuum(SqliteAutoVacuum::Incremental), + ) + .await?; + partial.close().await; + assert!(tokio::fs::metadata(&path).await?.len() > 0); + } + let pool = sqlite.open_read_write_pool(&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?; + let synchronous = sqlx::query_scalar::<_, i64>("PRAGMA synchronous") + .fetch_one(&pool) + .await?; + assert_eq!( + (journal_mode, auto_vacuum, synchronous), + ("wal".to_string(), SqliteAutoVacuum::Incremental as i64, 1) + ); + pool.close().await; + } Ok(()) } @@ -75,7 +94,7 @@ async fn open_read_write_pool_preserves_existing_settings_under_write_lock() -> } #[tokio::test] -async fn open_read_write_pool_preserves_wal_conversion_lock_error() -> anyhow::Result<()> { +async fn open_read_write_pool_preserves_delete_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| { @@ -100,12 +119,72 @@ async fn open_read_write_pool_preserves_wal_conversion_lock_error() -> anyhow::R .execute(&mut *lock_holder) .await?; - let result = sqlite.open_read_write_pool(&database_path).await; + let pool = sqlite.open_read_write_pool(&database_path).await?; + // Keep each acquired connection open to test connections that the pool initializes later. + let mut connections = Vec::new(); + let mut settings = Vec::new(); + for _ in 0..5 { + let mut connection = pool.acquire().await?; + let journal: String = sqlx::query_scalar("PRAGMA journal_mode") + .fetch_one(&mut *connection) + .await?; + let synchronous: i64 = sqlx::query_scalar("PRAGMA synchronous") + .fetch_one(&mut *connection) + .await?; + settings.push((journal, synchronous)); + connections.push(connection); + } + assert_eq!(settings, vec![("delete".to_string(), 3); 5]); + drop(connections); 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"); + sqlx::query("INSERT INTO existing VALUES(42)") + .execute(&pool) + .await?; + pool.close().await; + let reopened = sqlite.open_read_write_pool(&database_path).await?; + let journal: String = sqlx::query_scalar("PRAGMA journal_mode") + .fetch_one(&reopened) + .await?; + let rows: Vec = sqlx::query_scalar("SELECT id FROM existing") + .fetch_all(&reopened) + .await?; + assert_eq!((journal, rows), ("delete".into(), vec![42])); + reopened.close().await; + Ok(()) +} + +#[tokio::test] +async fn open_read_write_pool_preserves_empty_database_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(/*max*/ 1) + .connect_with( + SqliteConnectOptions::new() + .filename(&database_path) + .create_if_missing(/*create*/ true) + .auto_vacuum(SqliteAutoVacuum::Incremental) + .journal_mode(SqliteJournalMode::Delete), + ) + .await?; + // Leave the schema empty so startup must initialize WAL. + let mut lock_holder = existing_pool.acquire().await?; + sqlx::query("BEGIN IMMEDIATE") + .execute(&mut *lock_holder) + .await?; + + let error = sqlite + .open_read_write_pool(&database_path) + .await + .expect_err("WAL setup must fail while another connection holds the writer lock"); assert_eq!( error .downcast_ref::() @@ -114,11 +193,24 @@ async fn open_read_write_pool_preserves_wal_conversion_lock_error() -> anyhow::R 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; + + sqlx::query("ROLLBACK").execute(&mut *lock_holder).await?; + // A failed opener must release its locks before another writer can proceed. + sqlx::query("BEGIN EXCLUSIVE") + .execute(&mut *lock_holder) + .await?; + sqlx::query("ROLLBACK").execute(&mut *lock_holder).await?; + drop(lock_holder); + existing_pool.close().await; + + let pool = sqlite.open_read_write_pool(&database_path).await?; + sqlx::query("CREATE TABLE initialized (id INTEGER PRIMARY KEY)") + .execute(&pool) + .await?; + let journal: String = sqlx::query_scalar("PRAGMA journal_mode") + .fetch_one(&pool) + .await?; + assert_eq!(journal, "wal"); + pool.close().await; Ok(()) }