diff --git a/codex-rs/state/src/sqlite.rs b/codex-rs/state/src/sqlite.rs index 18b0a2b72d..d68c1fa8e4 100644 --- a/codex-rs/state/src/sqlite.rs +++ b/codex-rs/state/src/sqlite.rs @@ -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 { 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 { +#[derive(Clone, Copy, Eq, PartialEq)] +enum JournalModePolicy { + InitializeWal, + InitializeDelete, +} + +async fn connect_once( + options: &SqliteConnectOptions, + journal_mode_policy: JournalModePolicy, +) -> 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); @@ -575,12 +605,14 @@ async fn connect_once(options: &SqliteConnectOptions) -> Result "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("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;