mirror of
https://github.com/openai/codex.git
synced 2026-10-29 00:06:55 +00:00
Centralize SQLite connection configuration (#33938)
## What changed - Add `SqliteConfig` as the shared entry point for read-write and read-only SQLite pools. - Apply consistent WAL, synchronization, auto-vacuum, busy-timeout, logging, and pool-size settings to writable Codex databases. - Route state runtime, audit, CLI, and test database connections through the shared configuration. ## Testing - Run migration tests against temporary on-disk databases opened through `SqliteConfig`, including the concurrent-writer repair case. GitOrigin-RevId: e3946b98bde04c47574532ac8b1a7bb2b03edd97
This commit is contained in:
committed by
copyberry
parent
23899f7cb6
commit
6bd3f5e3db
Generated
+2
@@ -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",
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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 (
|
||||
|
||||
@@ -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<assert_cmd::Command> {
|
||||
@@ -17,16 +16,16 @@ fn codex_command(codex_home: &Path) -> Result<assert_cmd::Command> {
|
||||
#[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?;
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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<Vec<ThreadStateAuditRow>> {
|
||||
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
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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::<i64, _>("recency_at"), 1_700_000_300);
|
||||
assert_eq!(seeded.get::<i64, _>("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::<Vec<_>>();
|
||||
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");
|
||||
}
|
||||
|
||||
@@ -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<Arc<Self>> {
|
||||
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<bool> {
|
||||
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<SqlitePool> {
|
||||
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<SqlitePool> {
|
||||
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<SqlitePool> {
|
||||
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<SqlitePool> {
|
||||
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<Sqlite
|
||||
}
|
||||
|
||||
async fn open_sqlite(
|
||||
sqlite: &SqliteConfig,
|
||||
path: &Path,
|
||||
migrator: &Migrator,
|
||||
spec: RuntimeDbSpec,
|
||||
telemetry_override: Option<&dyn DbTelemetry>,
|
||||
) -> anyhow::Result<SqlitePool> {
|
||||
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<RuntimeDbPath> {
|
||||
|
||||
/// Run SQLite's built-in integrity check against an existing database file.
|
||||
pub async fn sqlite_integrity_check(path: &Path) -> anyhow::Result<Vec<String>> {
|
||||
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,
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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());
|
||||
|
||||
@@ -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<SqlitePool, Error> {
|
||||
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<SqlitePool, Error> {
|
||||
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
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user