openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-24 18:18:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

Initialize SQLite databases with DELETE journaling on NFS (#56552)

Browse Source 
## What changed

Choose `DELETE` journaling for new SQLite databases on NFS and `WAL` elsewhere, using the database's own directory to detect the filesystem. Probe a separately configured logs directory independently of the state directory.

Preserve journal modes for databases with an existing schema, and apply `synchronous = EXTRA` when initialization results in `DELETE` mode.

## Testing

Extend SQLite tests to cover NFS and non-NFS initialization, interrupted initialization, lazy pool connections, separate logs directories, and preservation of existing settings under a write lock.

GitOrigin-RevId: bde35861e517881b9ce23b28b199475cee70bd76
This commit is contained in:
Dmitriy Kovalenko
2026-10-23 18:30:44 +00:00
committed by copyberry
parent f970f4be67
commit 0d7f3ea03b
2 changed files with 136 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/state/src/sqlite.rs
+43 -7
View File
@@ -1,4 +1,5 @@
//! Shared SQLite connection configuration.
//! New databases use DELETE on NFS and WAL elsewhere.
//! Writable pools preserve the journal and vacuum modes of initialized databases.
//! Startup retries SQLITE_BUSY and reports other errors to callers.
//! The configured logs directory is created and locked only when opening its database.
@@ -416,6 +417,22 @@ impl SqliteConfig {
telemetry_override: Option<&dyn DbTelemetry>,
) -> anyhow::Result<SqlitePool> {
self.prepare_logs_directory(path).await?;
let directory = path
.parent()
.filter(|parent| !parent.as_os_str().is_empty())
.unwrap_or_else(|| Path::new("."));
let is_nfs = if self.local_logs_dir.is_some() && directory == self.logs_home() {
self.logs_are_on_nfs().await?
} else if directory == self.home() {
self.is_nfs().await?
} else {
probe_nfs(directory, &OnceCell::new()).await?
};
let journal_mode_policy = if is_nfs {
JournalModePolicy::InitializeDelete
} else {
JournalModePolicy::InitializeWal
};
let recovery = recover_spec.map_or(RecoveryMode::Unavailable, |spec| spec.recovery);
let options = SqliteConnectOptions::new()
.filename(path)
@@ -424,7 +441,11 @@ impl SqliteConfig {
.busy_timeout(DEFAULT_BUSY_TIMEOUT)
.log_statements(LevelFilter::Off);
// Each failed attempt closes its pool before initialization is retried.
let connect = || async { connect_once(&options).await.map_err(anyhow::Error::from) };
let connect = || async {
connect_once(&options, journal_mode_policy)
.await
.map_err(anyhow::Error::from)
};
let pool = retry_on_busy(connect).await?;
let validation_result = self
.quick_check_manager
@@ -518,7 +539,16 @@ impl SqliteConfig {
}
}
async fn connect_once(options: &SqliteConnectOptions) -> Result<SqlitePool, Error> {
#[derive(Clone, Copy, Eq, PartialEq)]
enum JournalModePolicy {
InitializeWal,
InitializeDelete,
}
async fn connect_once(
options: &SqliteConnectOptions,
journal_mode_policy: JournalModePolicy,
) -> 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);
@@ -575,12 +605,14 @@ async fn connect_once(options: &SqliteConnectOptions) -> Result<SqlitePool, Erro
.execute(&mut *connection)
.await?;
}
sqlx::query("PRAGMA journal_mode = WAL")
.execute(&mut *connection)
.await?;
let pragma = match journal_mode_policy {
JournalModePolicy::InitializeWal => "PRAGMA journal_mode = WAL",
JournalModePolicy::InitializeDelete => "PRAGMA journal_mode = DELETE",
};
sqlx::query(pragma).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.
// Initialization or another process can change the journal mode after after_connect.
// Match this connection's synchronization setting to the resulting journal mode.
let journal: String = sqlx::query_scalar("PRAGMA journal_mode")
.fetch_one(&mut *connection)
.await?;
@@ -588,6 +620,10 @@ async fn connect_once(options: &SqliteConnectOptions) -> Result<SqlitePool, Erro
sqlx::query("PRAGMA synchronous = NORMAL")
.execute(&mut *connection)
.await?;
} else if journal == "delete" {
sqlx::query("PRAGMA synchronous = EXTRA")
.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.
codex-rs/state/src/sqlite_tests.rs
+93 -19
View File
@@ -19,8 +19,10 @@ async fn configured_log_directory_is_initialized_on_use_and_retries_after_failur
let logs_home = sqlite_home.join("logs");
tokio::fs::write(&logs_home, "not a directory").await?;
let sqlite = SqliteConfig::new_for_testing(sqlite_home.as_path().abs());
sqlite.is_nfs.set(/*value*/ true)?;
let fallback_logs_path = sqlite.logs_db_path();
let sqlite = sqlite.with_local_logs_home(logs_home.clone());
sqlite.logs_is_nfs.set(/*value*/ false)?;
// Opening unrelated state must not require the configured logs directory.
let pool = sqlite.open_read_write_pool(&sqlite.state_db_path()).await?;
@@ -38,6 +40,20 @@ async fn configured_log_directory_is_initialized_on_use_and_retries_after_failur
assert!(sqlite.logs_db_path().is_file());
assert!(!fallback_logs_path.exists());
runtime.close().await;
// The configured logs directory has its own filesystem, independent of the NFS state home.
let mut modes = Vec::new();
for path in [sqlite.state_db_path(), sqlite.logs_db_path()] {
let pool = sqlite
.open_read_only_pool(&path, /*busy_timeout*/ None)
.await?;
modes.push(
sqlx::query_scalar::<_, String>("PRAGMA journal_mode")
.fetch_one(&pool)
.await?,
);
pool.close().await;
}
assert_eq!(modes, vec!["delete".to_string(), "wal".to_string()]);
Ok(())
}
@@ -147,42 +163,96 @@ async fn failed_sqlite_home_filesystem_probe_recovers_after_home_creation() -> a
#[tokio::test]
async fn open_read_write_pool_initializes_fresh_database_settings() -> anyhow::Result<()> {
for interrupted in [false, true] {
for (nfs, partial_journal) in [
(false, None),
(false, Some(SqliteJournalMode::Delete)),
(true, None),
(true, Some(SqliteJournalMode::Delete)),
(true, Some(SqliteJournalMode::Wal)),
] {
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());
sqlite.is_nfs.set(nfs)?;
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.
if let Some(journal_mode) = partial_journal {
// Startup must finish initialization when settings were written before the schema.
let partial = SqlitePoolOptions::new()
.max_connections(/*max*/ 1)
.connect_with(
SqliteConnectOptions::new()
.filename(&path)
.create_if_missing(/*create*/ true)
.auto_vacuum(SqliteAutoVacuum::Incremental),
.auto_vacuum(SqliteAutoVacuum::Incremental)
.journal_mode(journal_mode),
)
.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)
// Hold connections open so the assertions cover lazy pool connections too.
let mut connections = Vec::new();
let mut settings = Vec::new();
for _ in 0..5 {
let mut connection = pool.acquire().await?;
let current: (String, i64, i64) = sqlx::query_as(
"SELECT journal_mode, auto_vacuum, synchronous
FROM pragma_journal_mode, pragma_auto_vacuum, pragma_synchronous",
)
.fetch_one(&mut *connection)
.await?;
settings.push(current);
connections.push(connection);
}
assert_eq!(
(journal_mode, auto_vacuum, synchronous),
("wal".to_string(), SqliteAutoVacuum::Incremental as i64, 1)
settings,
vec![
(
if nfs { "delete" } else { "wal" }.to_string(),
SqliteAutoVacuum::Incremental as i64,
if nfs { 3 } else { 1 },
);
5
]
);
drop(connections);
pool.close().await;
}
Ok(())
}
#[tokio::test]
async fn open_read_write_pool_uses_database_directory_for_filesystem_probe() -> anyhow::Result<()> {
let directory = crate::runtime::test_support::unique_temp_dir();
tokio::fs::create_dir_all(&directory).await?;
let _cleanup = scopeguard::guard(directory.clone(), |directory| {
let _ = std::fs::remove_dir_all(directory);
});
let nfs = codex_file_system::is_nfs(&directory)?;
let missing_home = directory.join("absent");
for (index, cached_home_nfs) in [None, Some(!nfs)].into_iter().enumerate() {
let sqlite = SqliteConfig::new_for_testing(missing_home.as_path().abs());
if let Some(nfs) = cached_home_nfs {
sqlite.is_nfs.set(nfs)?;
}
// An unrelated missing home or its cached filesystem must not affect this database.
let path = directory.join(format!("database-{index}.sqlite"));
let pool = sqlite.open_read_write_pool(&path).await?;
let settings: (String, i64) = sqlx::query_as(
"SELECT journal_mode, synchronous FROM pragma_journal_mode, pragma_synchronous",
)
.fetch_one(&pool)
.await?;
assert_eq!(
settings,
(
if nfs { "delete" } else { "wal" }.to_string(),
if nfs { 3 } else { 1 },
)
);
pool.close().await;
}
@@ -197,6 +267,7 @@ async fn open_read_write_pool_preserves_existing_settings_under_write_lock() ->
let _ = std::fs::remove_dir_all(sqlite_home);
});
let sqlite = SqliteConfig::new_for_testing(sqlite_home.as_path().abs());
sqlite.is_nfs.set(/*value*/ true)?;
let database_path = sqlite.logs_db_path();
let existing_pool = SqlitePoolOptions::new()
.max_connections(1)
@@ -218,11 +289,14 @@ async fn open_read_write_pool_preserves_existing_settings_under_write_lock() ->
.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?;
let settings: (String, i64, i64) = sqlx::query_as(
"SELECT journal_mode, auto_vacuum, synchronous
FROM pragma_journal_mode, pragma_auto_vacuum, pragma_synchronous",
)
.fetch_one(&pool)
.await?;
assert_eq!(auto_vacuum, SqliteAutoVacuum::Full as i64);
assert_eq!(settings, ("wal".into(), SqliteAutoVacuum::Full as i64, 1));
sqlx::query("ROLLBACK").execute(&mut *lock_holder).await?;
drop(lock_holder);
pool.close().await;