github/codex
4
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-20 08:19:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

Preserve existing SQLite journal modes on startup (#54389)

Browse Source 
## Why

Opening an initialized database in `DELETE` journal mode previously attempted to convert it to `WAL`, which could fail while another connection held a write lock.

## What changed

- Initialize `WAL` and incremental auto-vacuum only for databases without a schema, including files left by interrupted initialization.
- Preserve journal and vacuum modes for initialized databases, use `synchronous = EXTRA` for `DELETE` connections, and retain `NORMAL` for `WAL`.
- Close failed initialization pools before retrying and restrict background page reclamation to `WAL` databases.

## Testing

Extend SQLite tests to cover interrupted initialization, opening `DELETE` databases under a write lock across all pool connections, persistence after reopening, and lock release after failed initialization.

GitOrigin-RevId: 23acecfe472618dde8545815e9a1058a51591235
This commit is contained in:
Dmitriy Kovalenko
2026-10-15 19:15:04 +00:00
committed by copyberry
parent f847a4bd3c
commit 1ef30bcbe7
3 changed files with 216 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/state/src/runtime/reclamation.rs
+5 -4
View File
@@ -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(());
}
codex-rs/state/src/sqlite.rs
+90 -54
View File
@@ -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<SqlitePool, Error> {
// 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;
codex-rs/state/src/sqlite_tests.rs
+121 -29
View File
@@ -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<i64> = 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::<sqlx::Error>()
@@ -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(())
}