diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 6138bddc8d..1e3bae8936 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -3985,11 +3985,13 @@ dependencies = [ "clap", "codex-git-utils", "codex-protocol", + "codex-utils-absolute-path", "dirs", "libsqlite3-sys", "log", "owo-colors", "pretty_assertions", + "scopeguard", "serde", "serde_json", "sqlx", diff --git a/codex-rs/Cargo.toml b/codex-rs/Cargo.toml index 376bad36a5..501a4106a8 100644 --- a/codex-rs/Cargo.toml +++ b/codex-rs/Cargo.toml @@ -394,6 +394,7 @@ serde_with = "3.17" serde_yaml = "0.9" serial_test = "3.2.0" sha1 = "0.10.6" +scopeguard = "1.2.0" sha2 = "0.10" shlex = "1.3.0" similar = "2.7.0" diff --git a/codex-rs/cli/src/doctor/thread_inventory.rs b/codex-rs/cli/src/doctor/thread_inventory.rs index ecef4e6fdc..2eb9003ba5 100644 --- a/codex-rs/cli/src/doctor/thread_inventory.rs +++ b/codex-rs/cli/src/doctor/thread_inventory.rs @@ -727,8 +727,6 @@ mod tests { use super::*; use codex_protocol::ThreadId; use pretty_assertions::assert_eq; - use sqlx::sqlite::SqliteConnectOptions; - use sqlx::sqlite::SqlitePoolOptions; use tempfile::TempDir; #[tokio::test] @@ -1152,14 +1150,11 @@ mod tests { async fn insert_thread_row(&self, id: &str, rollout_path: &Path, archived: bool) { let state_db_path = codex_state::state_db_path(self.sqlite_home.path()); - let options = SqliteConnectOptions::new() - .filename(state_db_path) - .create_if_missing(false); - let pool = SqlitePoolOptions::new() - .max_connections(1) - .connect_with(options) - .await - .expect("sqlite pool"); + let pool = + codex_state::SqliteConfig::new_for_testing(self.sqlite_home.path().to_path_buf()) + .open_read_write_pool(&state_db_path) + .await + .expect("sqlite pool"); sqlx::query( r#" INSERT INTO threads ( diff --git a/codex-rs/cli/tests/debug_clear_memories.rs b/codex-rs/cli/tests/debug_clear_memories.rs index be1c30da8a..0ff50053cb 100644 --- a/codex-rs/cli/tests/debug_clear_memories.rs +++ b/codex-rs/cli/tests/debug_clear_memories.rs @@ -5,7 +5,6 @@ use codex_state::StateRuntime; use codex_state::memories_db_path; use codex_state::state_db_path; use predicates::str::contains; -use sqlx::SqlitePool; use tempfile::TempDir; fn codex_command(codex_home: &Path) -> Result { @@ -17,16 +16,16 @@ fn codex_command(codex_home: &Path) -> Result { #[tokio::test] async fn debug_clear_memories_resets_state_and_removes_memory_dir() -> Result<()> { let codex_home = TempDir::new()?; + let sqlite = codex_state::SqliteConfig::new_for_testing(codex_home.path().to_path_buf()); let runtime = StateRuntime::init(codex_home.path().to_path_buf(), "test-provider".to_string()).await?; drop(runtime); let thread_id = "00000000-0000-0000-0000-000000000123"; let db_path = state_db_path(codex_home.path()); - let pool = SqlitePool::connect(&format!("sqlite://{}", db_path.display())).await?; + let pool = sqlite.open_read_write_pool(&db_path).await?; let memories_db_path = memories_db_path(codex_home.path()); - let memories_pool = - SqlitePool::connect(&format!("sqlite://{}", memories_db_path.display())).await?; + let memories_pool = sqlite.open_read_write_pool(&memories_db_path).await?; sqlx::query( r#" @@ -118,7 +117,7 @@ INSERT INTO jobs ( .success() .stdout(contains("Cleared memory state")); - let pool = SqlitePool::connect(&format!("sqlite://{}", memories_db_path.display())).await?; + let pool = sqlite.open_read_write_pool(&memories_db_path).await?; let stage1_outputs_count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM stage1_outputs") .fetch_one(&pool) .await?; @@ -140,14 +139,14 @@ INSERT INTO jobs ( #[tokio::test] async fn debug_clear_memories_resets_memories_db_without_state_db() -> Result<()> { let codex_home = TempDir::new()?; + let sqlite = codex_state::SqliteConfig::new_for_testing(codex_home.path().to_path_buf()); let runtime = StateRuntime::init(codex_home.path().to_path_buf(), "test-provider".to_string()).await?; runtime.close().await; let db_path = state_db_path(codex_home.path()); let memories_db_path = memories_db_path(codex_home.path()); - let memories_pool = - SqlitePool::connect(&format!("sqlite://{}", memories_db_path.display())).await?; + let memories_pool = sqlite.open_read_write_pool(&memories_db_path).await?; sqlx::query( r#" @@ -177,7 +176,7 @@ INSERT INTO stage1_outputs ( .success() .stdout(contains("Cleared memory state")); - let pool = SqlitePool::connect(&format!("sqlite://{}", memories_db_path.display())).await?; + let pool = sqlite.open_read_write_pool(&memories_db_path).await?; let stage1_outputs_count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM stage1_outputs") .fetch_one(&pool) .await?; diff --git a/codex-rs/state/Cargo.toml b/codex-rs/state/Cargo.toml index cbcec171ee..464d633bfc 100644 --- a/codex-rs/state/Cargo.toml +++ b/codex-rs/state/Cargo.toml @@ -9,6 +9,7 @@ anyhow = { workspace = true } chrono = { workspace = true } clap = { workspace = true, features = ["derive", "env"] } codex-protocol = { workspace = true } +codex-utils-absolute-path = { workspace = true } dirs = { workspace = true } libsqlite3-sys = { workspace = true } log = { workspace = true } @@ -25,6 +26,7 @@ uuid = { workspace = true } [dev-dependencies] codex-git-utils = { workspace = true } pretty_assertions = { workspace = true } +scopeguard = { workspace = true } [lints] workspace = true diff --git a/codex-rs/state/src/audit.rs b/codex-rs/state/src/audit.rs index 0ac4aadfaf..eedde2b7b2 100644 --- a/codex-rs/state/src/audit.rs +++ b/codex-rs/state/src/audit.rs @@ -1,11 +1,8 @@ //! Read-only state database queries for diagnostics. use anyhow::Result; -use log::LevelFilter; -use sqlx::ConnectOptions; +use codex_utils_absolute_path::AbsolutePathBuf; use sqlx::Row; -use sqlx::sqlite::SqliteConnectOptions; -use sqlx::sqlite::SqlitePoolOptions; use std::path::Path; use std::path::PathBuf; @@ -21,15 +18,10 @@ pub struct ThreadStateAuditRow { /// Read persisted thread rows from a state DB without creating, migrating, or repairing it. pub async fn read_thread_state_audit_rows(path: &Path) -> Result> { - let options = SqliteConnectOptions::new() - .filename(path) - .create_if_missing(false) - .read_only(true) - .log_statements(LevelFilter::Off); - let pool = SqlitePoolOptions::new() - .max_connections(1) - .connect_with(options) - .await?; + let sqlite = crate::SqliteConfig::from_sqlite_home(AbsolutePathBuf::try_from( + path.parent().unwrap_or(path), + )?); + let pool = sqlite.open_read_only_pool(path).await?; let rows = sqlx::query( r#" SELECT id, rollout_path, archived, source, model_provider diff --git a/codex-rs/state/src/lib.rs b/codex-rs/state/src/lib.rs index 4ab6349f7c..e0e8e44565 100644 --- a/codex-rs/state/src/lib.rs +++ b/codex-rs/state/src/lib.rs @@ -16,6 +16,7 @@ mod migrations; mod model; mod paths; mod runtime; +mod sqlite; mod telemetry; pub use model::LogEntry; @@ -24,6 +25,7 @@ pub use model::LogRow; pub use model::Phase2JobClaimOutcome; /// Preferred entrypoint: owns configuration and metrics. pub use runtime::StateRuntime; +pub use sqlite::SqliteConfig; pub use audit::ThreadStateAuditRow; pub use audit::read_thread_state_audit_rows; diff --git a/codex-rs/state/src/migrations_tests.rs b/codex-rs/state/src/migrations_tests.rs index 7cb4813605..77722f2585 100644 --- a/codex-rs/state/src/migrations_tests.rs +++ b/codex-rs/state/src/migrations_tests.rs @@ -1,18 +1,12 @@ -use std::borrow::Cow; -use std::time::Duration; - use sqlx::Connection; use sqlx::Row; -use sqlx::SqliteConnection; use sqlx::migrate::Migration; use sqlx::migrate::Migrator; -use sqlx::sqlite::SqliteConnectOptions; -use sqlx::sqlite::SqliteJournalMode; -use sqlx::sqlite::SqlitePoolOptions; -use uuid::Uuid; +use std::borrow::Cow; use super::STATE_MIGRATOR; use super::repair_legacy_recency_migration_version; +use crate::state_db_path; fn migrator_through(version: i64) -> Migrator { Migrator { @@ -34,11 +28,18 @@ fn migrator_through(version: i64) -> Migrator { #[tokio::test] async fn recency_migration_backfills_and_seeds_old_binary_inserts() { - let pool = SqlitePoolOptions::new() - .max_connections(1) - .connect("sqlite::memory:") + let sqlite_home = crate::runtime::test_support::unique_temp_dir(); + tokio::fs::create_dir_all(&sqlite_home) .await - .expect("in-memory database should open"); + .expect("sqlite home should be created"); + let _cleanup = scopeguard::guard(sqlite_home.clone(), |sqlite_home| { + let _ = std::fs::remove_dir_all(sqlite_home); + }); + let sqlite = crate::SqliteConfig::new_for_testing(sqlite_home.clone()); + let pool = sqlite + .open_read_write_pool(&state_db_path(&sqlite_home)) + .await + .expect("sqlite database should open"); migrator_through(/*version*/ 37) .run(&pool) .await @@ -134,15 +135,24 @@ INSERT INTO threads ( .expect("old-binary row should load"); assert_eq!(seeded.get::("recency_at"), 1_700_000_300); assert_eq!(seeded.get::("recency_at_ms"), 1_700_000_300_456); + + pool.close().await; } #[tokio::test] async fn repairs_recency_migration_that_was_applied_as_version_38() { - let pool = SqlitePoolOptions::new() - .max_connections(1) - .connect("sqlite::memory:") + let sqlite_home = crate::runtime::test_support::unique_temp_dir(); + tokio::fs::create_dir_all(&sqlite_home) .await - .expect("in-memory database should open"); + .expect("sqlite home should be created"); + let _cleanup = scopeguard::guard(sqlite_home.clone(), |sqlite_home| { + let _ = std::fs::remove_dir_all(sqlite_home); + }); + let sqlite = crate::SqliteConfig::new_for_testing(sqlite_home.clone()); + let pool = sqlite + .open_read_write_pool(&state_db_path(&sqlite_home)) + .await + .expect("sqlite database should open"); migrator_through(/*version*/ 37) .run(&pool) .await @@ -201,44 +211,47 @@ async fn repairs_recency_migration_that_was_applied_as_version_38() { .map(|migration| (migration.version, migration.checksum.to_vec())) .collect::>(); assert_eq!(applied, expected); + + pool.close().await; } #[tokio::test] async fn repair_recency_migration_succeeds_while_another_connection_holds_writer_slot() { - let database_path = std::env::temp_dir().join(format!( - "codex-state-migrations-test-{}.sqlite", - Uuid::new_v4() - )); - let options = SqliteConnectOptions::new() - .filename(&database_path) - .create_if_missing(true) - .journal_mode(SqliteJournalMode::Wal) - .busy_timeout(Duration::from_millis(100)); - let pool = SqlitePoolOptions::new() - .max_connections(1) - .connect_with(options.clone()) + let sqlite_home = crate::runtime::test_support::unique_temp_dir(); + tokio::fs::create_dir_all(&sqlite_home) + .await + .expect("sqlite home should be created"); + let _cleanup = scopeguard::guard(sqlite_home.clone(), |sqlite_home| { + let _ = std::fs::remove_dir_all(sqlite_home); + }); + let sqlite = crate::SqliteConfig::new_for_testing(sqlite_home.clone()); + let state_path = state_db_path(&sqlite_home); + let pool = sqlite + .open_read_write_pool(&state_path) .await .expect("database should open"); STATE_MIGRATOR .run(&pool) .await .expect("current migrations should apply"); - let mut write_connection = SqliteConnection::connect_with(&options) + let read_pool = sqlite + .open_read_only_pool(&state_path) .await - .expect("write connection should open"); + .expect("read-only pool should open"); + let mut write_connection = pool.acquire().await.expect("write connection should open"); let write_transaction = write_connection .begin_with("BEGIN IMMEDIATE") .await .expect("write transaction should acquire the writer slot"); - let repair_result = repair_legacy_recency_migration_version(&pool, &STATE_MIGRATOR).await; + let repair_result = repair_legacy_recency_migration_version(&read_pool, &STATE_MIGRATOR).await; write_transaction .rollback() .await .expect("write transaction should roll back"); drop(write_connection); + read_pool.close().await; pool.close().await; - let _ = tokio::fs::remove_file(database_path).await; repair_result.expect("current migration history should not need the writer slot"); } diff --git a/codex-rs/state/src/runtime.rs b/codex-rs/state/src/runtime.rs index 0055d262da..d3073df9a2 100644 --- a/codex-rs/state/src/runtime.rs +++ b/codex-rs/state/src/runtime.rs @@ -13,6 +13,7 @@ use crate::LogRow; use crate::MEMORIES_DB_FILENAME; use crate::STATE_DB_FILENAME; use crate::SortKey; +use crate::SqliteConfig; use crate::THREAD_HISTORY_DB_FILENAME; use crate::ThreadMetadata; use crate::ThreadMetadataBuilder; @@ -37,26 +38,19 @@ use chrono::DateTime; use chrono::Utc; use codex_protocol::ThreadId; use codex_protocol::protocol::RolloutItem; -use log::LevelFilter; +use codex_utils_absolute_path::AbsolutePathBuf; use serde_json::Value; -use sqlx::ConnectOptions; use sqlx::QueryBuilder; use sqlx::Row; use sqlx::Sqlite; use sqlx::SqliteConnection; use sqlx::SqlitePool; use sqlx::migrate::Migrator; -use sqlx::sqlite::SqliteAutoVacuum; -use sqlx::sqlite::SqliteConnectOptions; -use sqlx::sqlite::SqliteJournalMode; -use sqlx::sqlite::SqlitePoolOptions; -use sqlx::sqlite::SqliteSynchronous; use std::collections::BTreeSet; use std::path::Path; use std::path::PathBuf; use std::sync::Arc; use std::sync::atomic::AtomicI64; -use std::time::Duration; use std::time::Instant; use tracing::warn; @@ -69,7 +63,7 @@ mod memories; mod recovery; mod remote_control; #[cfg(test)] -mod test_support; +pub(crate) mod test_support; mod threads; pub use external_agent_config_imports::ExternalAgentConfigImportDetailsRecord; @@ -204,6 +198,7 @@ impl StateRuntime { default_provider: String, telemetry_override: Option<&dyn DbTelemetry>, ) -> anyhow::Result> { + let sqlite = SqliteConfig::from_sqlite_home(AbsolutePathBuf::try_from(codex_home.clone())?); tokio::fs::create_dir_all(&codex_home).await?; let state_migrator = runtime_state_migrator(); let logs_migrator = runtime_logs_migrator(); @@ -213,32 +208,46 @@ impl StateRuntime { let logs_path = LOGS_DB.path(codex_home.as_path()); let goals_path = GOALS_DB.path(codex_home.as_path()); let memories_path = MEMORIES_DB.path(codex_home.as_path()); - let pool = match open_state_sqlite(&state_path, &state_migrator, telemetry_override).await { + let pool = match open_state_sqlite( + &sqlite, + &state_path, + &state_migrator, + telemetry_override, + ) + .await + { Ok(db) => Arc::new(db), Err(err) => { warn!("failed to open state db at {}: {err}", state_path.display()); return Err(err); } }; - let logs_pool = match open_logs_sqlite(&logs_path, &logs_migrator, telemetry_override).await - { - 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); - } - }; - let goals_pool = - match open_goals_sqlite(&goals_path, &goals_migrator, telemetry_override).await { + let logs_pool = + match open_logs_sqlite(&sqlite, &logs_path, &logs_migrator, telemetry_override).await { 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; + warn!("failed to open logs db at {}: {err}", logs_path.display()); + close_sqlite_pools(&[pool.as_ref()]).await; return Err(err); } }; + let goals_pool = match open_goals_sqlite( + &sqlite, + &goals_path, + &goals_migrator, + telemetry_override, + ) + .await + { + 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); + } + }; let memories_pool = match open_memories_sqlite( + &sqlite, &memories_path, &memories_migrator, telemetry_override, @@ -346,6 +355,7 @@ impl StateRuntime { } pub async fn clear_memory_data_in_sqlite_home(sqlite_home: &Path) -> anyhow::Result { + let sqlite = SqliteConfig::from_sqlite_home(AbsolutePathBuf::try_from(sqlite_home)?); let memories_path = MEMORIES_DB.path(sqlite_home); if !tokio::fs::try_exists(&memories_path).await? { return Ok(false); @@ -353,6 +363,7 @@ impl StateRuntime { let memories_migrator = runtime_memories_migrator(); let pool = open_memories_sqlite( + &sqlite, &memories_path, &memories_migrator, /*telemetry_override*/ None, @@ -370,17 +381,8 @@ async fn close_sqlite_pools(pools: &[&SqlitePool]) { } } -fn base_sqlite_options(path: &Path) -> SqliteConnectOptions { - SqliteConnectOptions::new() - .filename(path) - .create_if_missing(true) - .journal_mode(SqliteJournalMode::Wal) - .synchronous(SqliteSynchronous::Normal) - .busy_timeout(Duration::from_secs(5)) - .log_statements(LevelFilter::Off) -} - async fn open_state_sqlite( + sqlite: &SqliteConfig, path: &Path, migrator: &Migrator, telemetry_override: Option<&dyn DbTelemetry>, @@ -388,37 +390,42 @@ async fn open_state_sqlite( // New state DBs should use incremental auto-vacuum, but retrofitting an // existing DB requires a full VACUUM. Do not attempt that during process // startup: it is maintenance work that can contend with foreground writers. - open_sqlite(path, migrator, STATE_DB, telemetry_override).await + open_sqlite(sqlite, path, migrator, STATE_DB, telemetry_override).await } async fn open_logs_sqlite( + sqlite: &SqliteConfig, path: &Path, migrator: &Migrator, telemetry_override: Option<&dyn DbTelemetry>, ) -> anyhow::Result { - open_sqlite(path, migrator, LOGS_DB, telemetry_override).await + open_sqlite(sqlite, path, migrator, LOGS_DB, telemetry_override).await } async fn open_goals_sqlite( + sqlite: &SqliteConfig, path: &Path, migrator: &Migrator, telemetry_override: Option<&dyn DbTelemetry>, ) -> anyhow::Result { - open_sqlite(path, migrator, GOALS_DB, telemetry_override).await + open_sqlite(sqlite, path, migrator, GOALS_DB, telemetry_override).await } async fn open_memories_sqlite( + sqlite: &SqliteConfig, path: &Path, migrator: &Migrator, telemetry_override: Option<&dyn DbTelemetry>, ) -> anyhow::Result { - open_sqlite(path, migrator, MEMORIES_DB, telemetry_override).await + open_sqlite(sqlite, path, migrator, MEMORIES_DB, telemetry_override).await } /// Open and migrate the rebuildable paginated thread-history database. pub async fn open_thread_history_db(sqlite_home: &Path) -> anyhow::Result { + let sqlite = SqliteConfig::from_sqlite_home(AbsolutePathBuf::try_from(sqlite_home)?); let migrator = runtime_thread_history_migrator(); open_sqlite( + &sqlite, thread_history_db_path(sqlite_home).as_path(), &migrator, THREAD_HISTORY_DB, @@ -428,16 +435,15 @@ pub async fn open_thread_history_db(sqlite_home: &Path) -> anyhow::Result, ) -> anyhow::Result { - let options = base_sqlite_options(path).auto_vacuum(SqliteAutoVacuum::Incremental); let started = Instant::now(); - let pool_result = SqlitePoolOptions::new() - .max_connections(5) - .connect_with(options) + let pool_result = sqlite + .open_read_write_pool(path) .await .map_err(anyhow::Error::from); crate::telemetry::record_init_result( @@ -551,15 +557,9 @@ pub fn runtime_db_paths(codex_home: &Path) -> Vec { /// Run SQLite's built-in integrity check against an existing database file. pub async fn sqlite_integrity_check(path: &Path) -> anyhow::Result> { - let options = SqliteConnectOptions::new() - .filename(path) - .create_if_missing(false) - .read_only(true) - .log_statements(LevelFilter::Off); - let pool = SqlitePoolOptions::new() - .max_connections(1) - .connect_with(options) - .await?; + let sqlite = + SqliteConfig::from_sqlite_home(AbsolutePathBuf::try_from(path.parent().unwrap_or(path))?); + let pool = sqlite.open_read_only_pool(path).await?; let rows = sqlx::query_scalar::<_, String>("PRAGMA integrity_check") .fetch_all(&pool) .await?; @@ -581,7 +581,6 @@ mod tests { use pretty_assertions::assert_eq; use sqlx::SqlitePool; use sqlx::migrate::MigrateError; - use sqlx::sqlite::SqliteConnectOptions; use std::collections::BTreeMap; use std::collections::BTreeSet; use std::path::Path; @@ -639,13 +638,10 @@ mod tests { } async fn open_db_pool(path: &Path) -> SqlitePool { - SqlitePool::connect_with( - SqliteConnectOptions::new() - .filename(path) - .create_if_missing(false), - ) - .await - .expect("open sqlite pool") + crate::SqliteConfig::new_for_testing(path.parent().unwrap_or(path).to_path_buf()) + .open_read_write_pool(path) + .await + .expect("open sqlite pool") } #[tokio::test] @@ -655,13 +651,10 @@ mod tests { .await .expect("create codex home"); let path = state_db_path(codex_home.as_path()); - let pool = SqlitePool::connect_with( - SqliteConnectOptions::new() - .filename(&path) - .create_if_missing(true), - ) - .await - .expect("open sqlite db"); + let pool = crate::SqliteConfig::new_for_testing(codex_home.clone()) + .open_read_write_pool(&path) + .await + .expect("open sqlite db"); sqlx::query("CREATE TABLE sample (id INTEGER PRIMARY KEY)") .execute(&pool) .await @@ -683,13 +676,10 @@ mod tests { .await .expect("create codex home"); let state_path = state_db_path(codex_home.as_path()); - let pool = SqlitePool::connect_with( - SqliteConnectOptions::new() - .filename(&state_path) - .create_if_missing(true), - ) - .await - .expect("open state db"); + let pool = crate::SqliteConfig::new_for_testing(codex_home.clone()) + .open_read_write_pool(&state_path) + .await + .expect("open state db"); STATE_MIGRATOR .run(&pool) .await @@ -717,6 +707,7 @@ mod tests { let tolerant_migrator = runtime_state_migrator(); let tolerant_pool = open_state_sqlite( + &crate::SqliteConfig::new_for_testing(codex_home.clone()), state_path.as_path(), &tolerant_migrator, /*telemetry_override*/ None, diff --git a/codex-rs/state/src/runtime/backfill.rs b/codex-rs/state/src/runtime/backfill.rs index 3bfa74fb9a..629e579ad6 100644 --- a/codex-rs/state/src/runtime/backfill.rs +++ b/codex-rs/state/src/runtime/backfill.rs @@ -110,7 +110,6 @@ WHERE id = 1 #[cfg(test)] mod tests { use super::StateRuntime; - use super::base_sqlite_options; use super::test_support::unique_temp_dir; use chrono::Utc; use pretty_assertions::assert_eq; @@ -175,11 +174,11 @@ mod tests { let runtime = StateRuntime::init(codex_home.clone(), "test-provider".to_string()) .await .expect("initialize runtime"); - let mut write_connection = sqlx::SqliteConnection::connect_with(&base_sqlite_options( - &crate::state_db_path(codex_home.as_path()), - )) - .await - .expect("open write connection"); + let write_pool = crate::SqliteConfig::new_for_testing(codex_home.clone()) + .open_read_write_pool(&crate::state_db_path(codex_home.as_path())) + .await + .expect("open write pool"); + let mut write_connection = write_pool.acquire().await.expect("open write connection"); let write_transaction = write_connection .begin_with("BEGIN IMMEDIATE") .await diff --git a/codex-rs/state/src/runtime/logs.rs b/codex-rs/state/src/runtime/logs.rs index 75b9ce82b2..a9ac2dbf7e 100644 --- a/codex-rs/state/src/runtime/logs.rs +++ b/codex-rs/state/src/runtime/logs.rs @@ -551,18 +551,14 @@ mod tests { use pretty_assertions::assert_eq; use sqlx::SqlitePool; use sqlx::migrate::Migrator; - use sqlx::sqlite::SqliteConnectOptions; use std::borrow::Cow; use std::path::Path; async fn open_db_pool(path: &Path) -> SqlitePool { - SqlitePool::connect_with( - SqliteConnectOptions::new() - .filename(path) - .create_if_missing(false), - ) - .await - .expect("open sqlite pool") + crate::SqliteConfig::new_for_testing(path.parent().unwrap_or(path).to_path_buf()) + .open_read_write_pool(path) + .await + .expect("open sqlite pool") } async fn log_row_count(path: &Path) -> i64 { @@ -621,13 +617,10 @@ mod tests { table_name: LOGS_MIGRATOR.table_name.clone(), create_schemas: LOGS_MIGRATOR.create_schemas.clone(), }; - let pool = SqlitePool::connect_with( - SqliteConnectOptions::new() - .filename(&logs_path) - .create_if_missing(true), - ) - .await - .expect("open old logs db"); + let pool = crate::SqliteConfig::new_for_testing(codex_home.clone()) + .open_read_write_pool(&logs_path) + .await + .expect("open old logs db"); old_logs_migrator .run(&pool) .await diff --git a/codex-rs/state/src/runtime/remote_control.rs b/codex-rs/state/src/runtime/remote_control.rs index 3482dbd68a..440da87b17 100644 --- a/codex-rs/state/src/runtime/remote_control.rs +++ b/codex-rs/state/src/runtime/remote_control.rs @@ -159,9 +159,7 @@ mod tests { use crate::migrations::STATE_MIGRATOR; use crate::state_db_path; use pretty_assertions::assert_eq; - use sqlx::SqlitePool; use sqlx::migrate::Migrator; - use sqlx::sqlite::SqliteConnectOptions; use std::borrow::Cow; #[tokio::test] @@ -338,13 +336,10 @@ mod tests { table_name: STATE_MIGRATOR.table_name.clone(), create_schemas: STATE_MIGRATOR.create_schemas.clone(), }; - let pool = SqlitePool::connect_with( - SqliteConnectOptions::new() - .filename(state_db_path(codex_home.as_path())) - .create_if_missing(true), - ) - .await - .expect("open old state db"); + let pool = crate::SqliteConfig::new_for_testing(codex_home.clone()) + .open_read_write_pool(&state_db_path(codex_home.as_path())) + .await + .expect("open old state db"); old_state_migrator .run(&pool) .await diff --git a/codex-rs/state/src/runtime/test_support.rs b/codex-rs/state/src/runtime/test_support.rs index 8c63066842..d5dbe7e0af 100644 --- a/codex-rs/state/src/runtime/test_support.rs +++ b/codex-rs/state/src/runtime/test_support.rs @@ -27,7 +27,7 @@ use uuid::Uuid; use crate::ThreadMetadata; #[cfg(test)] -pub(super) fn unique_temp_dir() -> PathBuf { +pub(crate) fn unique_temp_dir() -> PathBuf { let nanos = SystemTime::now() .duration_since(UNIX_EPOCH) .map_or(0, |duration| duration.as_nanos()); diff --git a/codex-rs/state/src/sqlite.rs b/codex-rs/state/src/sqlite.rs new file mode 100644 index 0000000000..b8e364f816 --- /dev/null +++ b/codex-rs/state/src/sqlite.rs @@ -0,0 +1,70 @@ +//! Shared SQLite connection configuration. + +use codex_utils_absolute_path::AbsolutePathBuf; +use log::LevelFilter; +use sqlx::ConnectOptions; +use sqlx::Error; +use sqlx::SqlitePool; +use sqlx::sqlite::SqliteAutoVacuum; +use sqlx::sqlite::SqliteConnectOptions; +use sqlx::sqlite::SqliteJournalMode; +use sqlx::sqlite::SqlitePoolOptions; +use sqlx::sqlite::SqliteSynchronous; +use std::path::Path; +use std::path::PathBuf; +use std::time::Duration; + +/// Resolved configuration shared by all Codex SQLite connections. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct SqliteConfig { + sqlite_home: AbsolutePathBuf, +} + +impl SqliteConfig { + pub fn from_sqlite_home(sqlite_home: AbsolutePathBuf) -> Self { + Self { sqlite_home } + } + + #[expect( + clippy::expect_used, + reason = "test sqlite homes must already be absolute" + )] + pub fn new_for_testing(sqlite_home: PathBuf) -> Self { + Self::from_sqlite_home( + AbsolutePathBuf::try_from(sqlite_home).expect("sqlite home should be absolute"), + ) + } + + pub fn home(&self) -> &Path { + self.sqlite_home.as_path() + } + + /// Open a writable Codex SQLite database, creating it if necessary. + pub async fn open_read_write_pool(&self, path: &Path) -> Result { + let options = SqliteConnectOptions::new() + .filename(path) + .create_if_missing(true) + .journal_mode(SqliteJournalMode::Wal) + .synchronous(SqliteSynchronous::Normal) + .auto_vacuum(SqliteAutoVacuum::Incremental) + .busy_timeout(Duration::from_secs(5)) + .log_statements(LevelFilter::Off); + SqlitePoolOptions::new() + .max_connections(5) + .connect_with(options) + .await + } + + /// Open an existing Codex SQLite database without creating or modifying it. + pub async fn open_read_only_pool(&self, path: &Path) -> Result { + let options = SqliteConnectOptions::new() + .filename(path) + .create_if_missing(false) + .read_only(true) + .log_statements(LevelFilter::Off); + SqlitePoolOptions::new() + .max_connections(1) + .connect_with(options) + .await + } +}