From 480ab2d0061e004bc594a9f3f1ac07994e853522 Mon Sep 17 00:00:00 2001 From: Dmitriy Kovalenko Date: Sun, 25 Oct 2026 16:19:41 +0000 Subject: [PATCH] Add managed enforcement of DELETE journaling on NFS (#57225) ## Why Existing databases retain their journal mode, so selecting `DELETE` for new NFS databases leaves existing WAL databases unchanged. ## What changed - Add the managed `unsafe_enforce_sqlite_journal_mode` requirement to migrate existing NFS databases before opening writable pools and require `DELETE` on each connection. - Share migration code between startup and `codex migrate-nfs-db`, preserving WAL backups and resuming interrupted replacements, including lazily opened databases. - Hold the migration lock through pool initialization and keep migration failures out of automatic corruption recovery. - Preserve local logs behavior and skip legacy NFS logs cleanup when enforcement is enabled. ## Testing Add coverage for requirement precedence, database settings, data and backup preservation across restarts, interrupted replacement recovery, refusal of active WAL databases, migration lock waiting, and local logs remaining in WAL mode. GitOrigin-RevId: da41ec1b58b1ee8092591cdf8b358280928ff9d4 --- codex-rs/Cargo.lock | 3 +- codex-rs/cli/Cargo.toml | 1 - codex-rs/cli/src/bin/logs_client.rs | 3 +- codex-rs/cli/src/migrate_nfs_db.rs | 254 ++--------------- codex-rs/cli/tests/doctor_path_safety.rs | 4 +- codex-rs/cli/tests/worktree.rs | 2 +- codex-rs/config/src/config_requirements.rs | 27 ++ .../config/src/requirements_layers/stack.rs | 5 + .../src/requirements_layers/stack_tests.rs | 3 + codex-rs/core/src/config/config_tests.rs | 17 +- codex-rs/core/src/config/mod.rs | 8 +- .../examples/configure_board.rs | 3 +- codex-rs/state/Cargo.toml | 2 + codex-rs/state/src/lib.rs | 3 + codex-rs/state/src/nfs_migration.rs | 262 ++++++++++++++++++ .../src/nfs_migration}/access.rs | 0 .../src/nfs_migration}/replacement.rs | 0 codex-rs/state/src/runtime.rs | 2 +- codex-rs/state/src/sqlite.rs | 148 +++------- codex-rs/state/src/sqlite/connection.rs | 122 ++++++++ codex-rs/state/src/sqlite/nfs.rs | 66 +++++ codex-rs/state/src/sqlite_tests.rs | 164 ++++++++++- codex-rs/thread-manager-sample/src/main.rs | 5 +- codex-rs/tui/src/debug_config.rs | 1 + 24 files changed, 747 insertions(+), 358 deletions(-) create mode 100644 codex-rs/state/src/nfs_migration.rs rename codex-rs/{cli/src/migrate_nfs_db => state/src/nfs_migration}/access.rs (100%) rename codex-rs/{cli/src/migrate_nfs_db => state/src/nfs_migration}/replacement.rs (100%) create mode 100644 codex-rs/state/src/sqlite/connection.rs create mode 100644 codex-rs/state/src/sqlite/nfs.rs diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 962dfa6219..12a8a8bb0f 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -2571,7 +2571,6 @@ dependencies = [ "insta", "jsonwebtoken", "libc", - "libsqlite3-sys", "os_info", "owo-colors", "predicates", @@ -4803,6 +4802,7 @@ dependencies = [ "codex-protocol", "codex-utils-absolute-path", "file-id", + "libc", "libsqlite3-sys", "log", "pretty_assertions", @@ -4813,6 +4813,7 @@ dependencies = [ "sqlx-macros", "sqlx-sqlite", "strum 0.27.2", + "tempfile", "tokio", "tracing", "tracing-subscriber", diff --git a/codex-rs/cli/Cargo.toml b/codex-rs/cli/Cargo.toml index c812cf9887..8cc0bce5d8 100644 --- a/codex-rs/cli/Cargo.toml +++ b/codex-rs/cli/Cargo.toml @@ -87,7 +87,6 @@ crossterm = { workspace = true, features = ["event-stream"] } futures = { workspace = true } http = { workspace = true } libc = { workspace = true } -libsqlite3-sys = { workspace = true } os_info = { workspace = true } owo-colors = { workspace = true } regex-lite = { workspace = true } diff --git a/codex-rs/cli/src/bin/logs_client.rs b/codex-rs/cli/src/bin/logs_client.rs index 32ef9323ed..e959c33df1 100644 --- a/codex-rs/cli/src/bin/logs_client.rs +++ b/codex-rs/cli/src/bin/logs_client.rs @@ -134,8 +134,9 @@ async fn resolve_sqlite_config(args: &Args) -> anyhow::Result { .parent() .map(ToOwned::to_owned) .unwrap_or_else(|| PathBuf::from(".")); - return Ok(SqliteConfig::from_sqlite_home( + return Ok(SqliteConfig::from_sqlite_home_with_nfs_enforcement( AbsolutePathBuf::relative_to_current_dir(sqlite_home)?, + /*enforce_delete_on_nfs*/ false, )); } diff --git a/codex-rs/cli/src/migrate_nfs_db.rs b/codex-rs/cli/src/migrate_nfs_db.rs index 254102d032..02c296d4f0 100644 --- a/codex-rs/cli/src/migrate_nfs_db.rs +++ b/codex-rs/cli/src/migrate_nfs_db.rs @@ -1,37 +1,17 @@ -//! Convert configured databases to DELETE mode and keep their WAL originals in backup directories. -//! Require confirmation before opening database files. All clients must remain stopped. -//! Keep source descriptors open until SQLite closes. Do not checkpoint the WAL source. -//! NFS must enforce POSIX locks across clients. Local descriptor checks cannot prove this. +//! Confirms explicit database migration and reports the shared migrator's results. +//! Other SQLite clients must remain stopped during replacement, and NFS must enforce +//! cross-client POSIX locks. use anyhow::Context; use anyhow::ensure; use clap::Parser; use codex_config::LoaderOverrides; use codex_core::config::ConfigBuilder; +use codex_state::NfsDbMigration; +use codex_state::NfsDbMigrationOutcome; use codex_utils_cli::CliConfigOverrides; use owo_colors::OwoColorize; -use sqlx::ConnectOptions; -use sqlx::Connection; -use sqlx::SqliteConnection; -use sqlx::sqlite::SqliteConnectOptions; -use sqlx::sqlite::SqliteJournalMode; -use sqlx::sqlite::SqliteLockingMode; -use sqlx::sqlite::SqliteSynchronous; -use std::fs::File; -use std::io::ErrorKind; -use std::io::Read; -use std::io::Write; -use std::path::Path; use std::path::PathBuf; -use std::time::Duration; - -#[cfg(unix)] -mod access; -mod replacement; - -use replacement::Recovery; -use replacement::Replacement; -use replacement::regular_file_exists; #[derive(Debug, Parser)] #[command( @@ -83,226 +63,28 @@ pub(crate) async fn run( println!("No databases found."); return Ok(()); } - // Keep this file after release. Removing it lets two migrations lock different inodes. - let migration_lock = std::fs::OpenOptions::new() - .read(true) - .write(true) - .create(true) - .truncate(false) - .open(sqlite.home().join(".codex-nfs-migration.lock"))?; - migration_lock - .try_lock() - .context("another database migration is running")?; + let mut migration = NfsDbMigration::lock(sqlite.home())?; for database in sqlite.runtime_db_paths() { if skip_logs && database.path == sqlite.logs_db_path() { println!("Skipped logs database: log_dir is configured."); continue; } - migrate(&database.path, &command.staging_dir).await.with_context(|| format!( + let outcome = migration.migrate(&database.path, &command.staging_dir).await.with_context(|| format!( "migration failed for {}. Keep all Codex instances stopped. Correct the error and rerun codex migrate-nfs-db", database.path.display() ))?; + match outcome { + NfsDbMigrationOutcome::Missing => {} + NfsDbMigrationOutcome::AlreadyDelete => { + println!("Already in DELETE mode: {}", database.path.display()) + } + NfsDbMigrationOutcome::Migrated { backup_directory } => println!( + "Migrated database: {}\nWAL backup: {}", + database.path.display(), + backup_directory.display() + ), + } } println!("Migration completed successfully. You can restart Codex."); Ok(()) } - -#[expect( - clippy::disallowed_methods, - reason = "Migration needs dedicated exclusive connections, not the runtime's WAL pool" -)] -async fn migrate(source: &Path, staging_dir: &Path) -> anyhow::Result<()> { - let replacement = Replacement::new(source)?; - if let Recovery::Installed = replacement.resume().await? { - println!( - "Migrated database: {}\nWAL backup: {}", - source.display(), - replacement.backup_directory.display() - ); - return Ok(()); - } - let exists = regular_file_exists(source)?; - for suffix in ["-wal", "-shm", "-journal"] { - let mut path = source.as_os_str().to_owned(); - path.push(suffix); - let sidecar_exists = regular_file_exists(Path::new(&path))?; - ensure!( - exists || !sidecar_exists, - "database is missing but a SQLite sidecar exists" - ); - } - if !exists { - return Ok(()); - } - let source_file = File::open(source)?; - #[cfg(unix)] - access::require_private_database(&source_file)?; - // The private directory protects WAL contents even on a shared /tmp. - let staging = tempfile::Builder::new() - .prefix("codex-nfs-migration-") - .tempdir_in(staging_dir) - .context("cannot create local staging directory")?; - let staged_database = staging.path().join("database.sqlite"); - let options = SqliteConnectOptions::new() - .filename(source) - // Set this mode before SQLite first accesses the WAL. - // SQLite then locks the database file without shared memory across NFS client kernels. - .locking_mode(SqliteLockingMode::Exclusive) - .synchronous(SqliteSynchronous::Full) - .busy_timeout(Duration::ZERO) - .disable_statement_logging(); - let mut connection = SqliteConnection::connect_with(&options).await.context( - "cannot open source database exclusively; close its SQLite clients on all hosts", - )?; - - // Closing any descriptor for an inode can release this process's POSIX record locks - // for that inode. - // Keep both source files open until SQLite closes its connection, including after an error. - let mut source_files = vec![source_file]; - let result: anyhow::Result = async { - { - let mut handle = connection.lock_handle().await?; - // SAFETY: SQLx's guard prevents its worker from accessing this open connection. - // This db_config option accepts an int and an optional output pointer to an int. - let status = unsafe { - libsqlite3_sys::sqlite3_db_config( - handle.as_raw_handle().as_ptr(), - libsqlite3_sys::SQLITE_DBCONFIG_NO_CKPT_ON_CLOSE, - /*onoff*/ 1, - std::ptr::null_mut::(), - ) - }; - ensure!( - status == libsqlite3_sys::SQLITE_OK, - "cannot disable source checkpoint on close: {status}" - ); - } - let mode: String = sqlx::query_scalar("PRAGMA journal_mode") - .fetch_one(&mut connection) - .await - .context( - "cannot access source database exclusively; close its SQLite clients on all hosts", - )?; - if mode == "delete" { - ensure!( - !regular_file_exists(&replacement.replacement_path())?, - "unfinished backup exists" - ); - println!("Already in DELETE mode: {}", source.display()); - return Ok(false); - } - ensure!( - mode == "wal", - "source database must be in WAL mode (found {mode})" - ); - // BEGIN EXCLUSIVE alone blocks only WAL writers. EXCLUSIVE locking_mode also blocks readers. - // The mode keeps the database lock after ROLLBACK. - sqlx::raw_sql("BEGIN EXCLUSIVE; ROLLBACK;") - .execute(&mut connection) - .await - .context( - "cannot lock source database exclusively; close its SQLite clients on all hosts", - )?; - - let mut wal_path = source.as_os_str().to_owned(); - wal_path.push("-wal"); - match File::open(wal_path) { - Ok(file) => source_files.push(file), - Err(error) if error.kind() == ErrorKind::NotFound => {} - Err(error) => return Err(error).context("cannot open source WAL"), - } - for (file, name) in source_files - .iter_mut() - .zip(["database.sqlite", "database.sqlite-wal"]) - { - let mut target = File::create(staging.path().join(name))?; - copy_buffered(file, &mut target).context("cannot stage source database and WAL")?; - } - let mut copy = SqliteConnection::connect_with( - &SqliteConnectOptions::new() - .filename(&staged_database) - .locking_mode(SqliteLockingMode::Exclusive) - .journal_mode(SqliteJournalMode::Delete) - .synchronous(SqliteSynchronous::Full) - .disable_statement_logging(), - ) - .await - .context("cannot convert staged database to DELETE")?; - let validated = validate_delete_database(&mut copy) - .await - .context("staged database failed validation"); - let closed = copy.close().await; - validated?; - closed?; - - replacement.prepare_backup_directory()?; - let mut temporary = tempfile::NamedTempFile::new_in(&replacement.backup_directory)?; - copy_buffered(&mut File::open(&staged_database)?, temporary.as_file_mut()) - .context("cannot write replacement copy")?; - // Use the open handle. Closing another source descriptor can release SQLite's locks. - temporary - .as_file() - .set_permissions(source_files[0].metadata()?.permissions())?; - #[cfg(unix)] - access::require_private_database(temporary.as_file())?; - temporary.as_file().sync_all()?; - temporary - .into_temp_path() - .persist_noclobber(replacement.replacement_path()) - .context("cannot save replacement copy")?; - File::open(&replacement.backup_directory)?.sync_all()?; - for file in &source_files { - file.sync_all()?; - } - replacement.move_original_to_backup()?; - Ok(true) - } - .await; - let closed = connection.close().await; - drop(source_files); - let changed = result?; - closed.context("cannot close source database. Keep clients stopped")?; - if changed { - replacement.install()?; - println!( - "Migrated database: {}\nWAL backup: {}", - source.display(), - replacement.backup_directory.display() - ); - } - Ok(()) -} - -async fn validate_delete_database(connection: &mut SqliteConnection) -> anyhow::Result<()> { - let mode: String = sqlx::query_scalar("PRAGMA journal_mode") - .fetch_one(&mut *connection) - .await?; - ensure!( - mode == "delete", - "database did not enter DELETE journal mode" - ); - let integrity: String = sqlx::query_scalar("PRAGMA integrity_check(1)") - .fetch_one(connection) - .await - .context("database failed integrity check")?; - ensure!( - integrity == "ok", - "database failed integrity check: {integrity}" - ); - Ok(()) -} - -fn copy_buffered(source: &mut File, destination: &mut File) -> std::io::Result<()> { - // Use large reads to avoid the fallback to small reads for copy_file_range on NFSv4.1. - let mut buffer = vec![0; 1024 * 1024]; - loop { - let bytes = match source.read(&mut buffer) { - Err(error) if error.kind() == ErrorKind::Interrupted => continue, - result => result?, - }; - if bytes == 0 { - return Ok(()); - } - destination.write_all(&buffer[..bytes])?; - } -} diff --git a/codex-rs/cli/tests/doctor_path_safety.rs b/codex-rs/cli/tests/doctor_path_safety.rs index cd5d7182ec..66d6d2c0cf 100644 --- a/codex-rs/cli/tests/doctor_path_safety.rs +++ b/codex-rs/cli/tests/doctor_path_safety.rs @@ -217,7 +217,7 @@ fn doctor_reports_failed_database_paths() -> Result<()> { let fixture = Fixture::new()?; let sqlite_home = fixture.root.path().join("sqlite"); std::fs::create_dir(&sqlite_home)?; - let sqlite = SqliteConfig::from_sqlite_home(AbsolutePathBuf::try_from(sqlite_home.clone())?); + let sqlite = SqliteConfig::new_for_testing(AbsolutePathBuf::try_from(sqlite_home.clone())?); let config_file = fixture.home.join("config.toml"); let config = std::fs::read_to_string(&config_file)?; let sqlite_home = toml::Value::String(sqlite_home.display().to_string()); @@ -311,7 +311,7 @@ fn doctor_redacts_failed_database_paths_in_json() -> Result<()> { let fixture = Fixture::new()?; let sqlite_home = fixture.root.path().join("doctor-secret-sqlite"); std::fs::create_dir(&sqlite_home)?; - let sqlite = SqliteConfig::from_sqlite_home(AbsolutePathBuf::try_from(sqlite_home.clone())?); + let sqlite = SqliteConfig::new_for_testing(AbsolutePathBuf::try_from(sqlite_home.clone())?); let config_file = fixture.home.join("config.toml"); let config = std::fs::read_to_string(&config_file)?; let sqlite_home = toml::Value::String(sqlite_home.display().to_string()); diff --git a/codex-rs/cli/tests/worktree.rs b/codex-rs/cli/tests/worktree.rs index 55cc4dfd98..13b6914968 100644 --- a/codex-rs/cli/tests/worktree.rs +++ b/codex-rs/cli/tests/worktree.rs @@ -819,7 +819,7 @@ trust_level = "trusted" assert!(context.contains("managed cloud instructions"), "{context}"); if previous.is_empty() { // A legacy summary still says launcher A, but persisted turns use checkout B. - let sqlite = codex_state::SqliteConfig::from_sqlite_home( + let sqlite = codex_state::SqliteConfig::new_for_testing( codex_utils_absolute_path::AbsolutePathBuf::from_absolute_path(&home)?, ); let db = sqlite.open_read_write_pool(&sqlite.state_db_path()).await?; diff --git a/codex-rs/config/src/config_requirements.rs b/codex-rs/config/src/config_requirements.rs index 961e3c3d78..e641058844 100644 --- a/codex-rs/config/src/config_requirements.rs +++ b/codex-rs/config/src/config_requirements.rs @@ -166,6 +166,7 @@ pub struct ConfigRequirements { pub cli_auth_credentials_store: Option>, pub chatgpt_base_url: Option>, pub sqlite_home: Option>, + pub unsafe_enforce_sqlite_journal_mode: Option>, pub log_dir: Option>, pub model_catalog_json: Option>, pub model_provider: Option>, @@ -211,6 +212,7 @@ impl Default for ConfigRequirements { cli_auth_credentials_store: None, chatgpt_base_url: None, sqlite_home: None, + unsafe_enforce_sqlite_journal_mode: None, log_dir: None, model_catalog_json: None, model_provider: None, @@ -1049,6 +1051,8 @@ pub struct ConfigRequirementsToml { pub cli_auth_credentials_store: Option, pub chatgpt_base_url: Option, pub sqlite_home: Option, + /// Require migration of existing WAL databases on NFS before opening writable pools. + pub unsafe_enforce_sqlite_journal_mode: Option, pub log_dir: Option, pub model_catalog_json: Option, /// Exact provider selection, overriding local and session configuration. @@ -1161,6 +1165,7 @@ pub struct ConfigRequirementsWithSources { pub cli_auth_credentials_store: Option>, pub chatgpt_base_url: Option>, pub sqlite_home: Option>, + pub unsafe_enforce_sqlite_journal_mode: Option>, pub log_dir: Option>, pub model_catalog_json: Option>, pub model_provider: Option>, @@ -1225,6 +1230,7 @@ impl ConfigRequirementsWithSources { cli_auth_credentials_store: _, chatgpt_base_url: _, sqlite_home: _, + unsafe_enforce_sqlite_journal_mode: _, log_dir: _, model_catalog_json: _, model_provider: _, @@ -1291,6 +1297,7 @@ impl ConfigRequirementsWithSources { cli_auth_credentials_store, chatgpt_base_url, sqlite_home, + unsafe_enforce_sqlite_journal_mode, log_dir, model_catalog_json, model_provider, @@ -1378,6 +1385,7 @@ impl ConfigRequirementsWithSources { cli_auth_credentials_store, chatgpt_base_url, sqlite_home, + unsafe_enforce_sqlite_journal_mode, log_dir, model_catalog_json, model_provider, @@ -1423,6 +1431,8 @@ impl ConfigRequirementsWithSources { cli_auth_credentials_store: cli_auth_credentials_store.map(|sourced| sourced.value), chatgpt_base_url: chatgpt_base_url.map(|sourced| sourced.value), sqlite_home: sqlite_home.map(|sourced| sourced.value), + unsafe_enforce_sqlite_journal_mode: unsafe_enforce_sqlite_journal_mode + .map(|sourced| sourced.value), log_dir: log_dir.map(|sourced| sourced.value), model_catalog_json: model_catalog_json.map(|sourced| sourced.value), model_provider: model_provider.map(|sourced| sourced.value), @@ -1536,6 +1546,7 @@ impl ConfigRequirementsToml { && self.cli_auth_credentials_store.is_none() && self.chatgpt_base_url.is_none() && self.sqlite_home.is_none() + && self.unsafe_enforce_sqlite_journal_mode.is_none() && self.log_dir.is_none() && self.model_catalog_json.is_none() && self.model_provider.is_none() @@ -1780,6 +1791,7 @@ impl TryFrom for ConfigRequirements { cli_auth_credentials_store, chatgpt_base_url, sqlite_home, + unsafe_enforce_sqlite_journal_mode, log_dir, model_catalog_json, model_provider, @@ -2170,6 +2182,7 @@ impl TryFrom for ConfigRequirements { cli_auth_credentials_store, chatgpt_base_url, sqlite_home, + unsafe_enforce_sqlite_journal_mode, log_dir, model_catalog_json, model_provider, @@ -2334,6 +2347,7 @@ mod tests { cli_auth_credentials_store, chatgpt_base_url, sqlite_home, + unsafe_enforce_sqlite_journal_mode, log_dir, model_catalog_json, model_provider, @@ -2385,6 +2399,8 @@ mod tests { chatgpt_base_url: chatgpt_base_url .map(|value| Sourced::new(value, RequirementSource::Unknown)), sqlite_home: sqlite_home.map(|value| Sourced::new(value, RequirementSource::Unknown)), + unsafe_enforce_sqlite_journal_mode: unsafe_enforce_sqlite_journal_mode + .map(|value| Sourced::new(value, RequirementSource::Unknown)), log_dir: log_dir.map(|value| Sourced::new(value, RequirementSource::Unknown)), model_catalog_json: model_catalog_json .map(|value| Sourced::new(value, RequirementSource::Unknown)), @@ -2916,6 +2932,7 @@ mod tests { cli_auth_credentials_store: Some(AuthCredentialsStoreMode::Keyring), chatgpt_base_url: Some("https://managed.example/backend-api/".to_string()), sqlite_home: Some(sqlite_home.clone()), + unsafe_enforce_sqlite_journal_mode: Some(true), log_dir: Some(log_dir.clone()), model_catalog_json: Some(model_catalog_json.clone()), model_provider: Some("gateway".to_string()), @@ -2979,6 +2996,10 @@ mod tests { source.clone(), )), sqlite_home: Some(Sourced::new(sqlite_home, source.clone())), + unsafe_enforce_sqlite_journal_mode: Some(Sourced::new( + /*value*/ true, + source.clone() + )), log_dir: Some(Sourced::new(log_dir, source.clone())), model_catalog_json: Some(Sourced::new(model_catalog_json, source.clone())), model_provider: Some(Sourced::new("gateway".to_string(), source.clone())), @@ -3109,6 +3130,7 @@ mod tests { let mut populated_target = ConfigRequirementsWithSources::default(); let populated_requirements: ConfigRequirementsToml = from_str( r#" + unsafe_enforce_sqlite_journal_mode = true allowed_approval_policies = ["never"] guardian_extra_policy = "Use the managed extra policy." "#, @@ -3117,6 +3139,7 @@ mod tests { let source: ConfigRequirementsToml = from_str( r#" + unsafe_enforce_sqlite_journal_mode = false allowed_approval_policies = ["on-request"] guardian_extra_policy = "Use the lower-priority extra policy." "#, @@ -3130,6 +3153,10 @@ mod tests { assert_eq!( populated_target, ConfigRequirementsWithSources { + unsafe_enforce_sqlite_journal_mode: Some(Sourced::new( + /*value*/ true, + existing_source.clone() + )), allowed_approval_policies: Some(Sourced::new( vec![AskForApproval::Never], existing_source.clone(), diff --git a/codex-rs/config/src/requirements_layers/stack.rs b/codex-rs/config/src/requirements_layers/stack.rs index 68f4261bb2..4dd6e255ba 100644 --- a/codex-rs/config/src/requirements_layers/stack.rs +++ b/codex-rs/config/src/requirements_layers/stack.rs @@ -223,6 +223,7 @@ fn populate_merged_regular_fields_with_sources( cli_auth_credentials_store, chatgpt_base_url, sqlite_home, + unsafe_enforce_sqlite_journal_mode, log_dir, model_catalog_json, model_provider, @@ -269,6 +270,10 @@ fn populate_merged_regular_fields_with_sources( set_sourced!(cli_auth_credentials_store, &["cli_auth_credentials_store"]); set_sourced!(chatgpt_base_url, &["chatgpt_base_url"]); set_sourced!(sqlite_home, &["sqlite_home"]); + set_sourced!( + unsafe_enforce_sqlite_journal_mode, + &["unsafe_enforce_sqlite_journal_mode"] + ); set_sourced!(log_dir, &["log_dir"]); set_sourced!(model_catalog_json, &["model_catalog_json"]); set_sourced!(model_provider, &["model_provider"]); diff --git a/codex-rs/config/src/requirements_layers/stack_tests.rs b/codex-rs/config/src/requirements_layers/stack_tests.rs index 2e9bebde11..e1c9d9fe22 100644 --- a/codex-rs/config/src/requirements_layers/stack_tests.rs +++ b/codex-rs/config/src/requirements_layers/stack_tests.rs @@ -102,6 +102,7 @@ allowed_approval_policies = ["on-request"] allowed_sandbox_modes = ["workspace-write"] default_permissions = ":workspace" allow_remote_control = true +unsafe_enforce_sqlite_journal_mode = false additional_developer_instructions = "Lower-priority instructions." [allowed_permission_profiles] @@ -117,6 +118,7 @@ allowed_approval_policies = ["never"] allowed_sandbox_modes = ["read-only"] default_permissions = ":read-only" allow_remote_control = false +unsafe_enforce_sqlite_journal_mode = true additional_developer_instructions = "" [allowed_permission_profiles] @@ -136,6 +138,7 @@ allowed_approval_policies = ["never"] allowed_sandbox_modes = ["read-only"] default_permissions = ":read-only" allow_remote_control = false +unsafe_enforce_sqlite_journal_mode = true additional_developer_instructions = "" [allowed_permission_profiles] diff --git a/codex-rs/core/src/config/config_tests.rs b/codex-rs/core/src/config/config_tests.rs index 16c6fc8ef5..97217c5ce6 100644 --- a/codex-rs/core/src/config/config_tests.rs +++ b/codex-rs/core/src/config/config_tests.rs @@ -10660,6 +10660,7 @@ async fn test_requirements_web_search_mode_allowlist_does_not_warn_when_unset() cli_auth_credentials_store: None, chatgpt_base_url: None, sqlite_home: None, + unsafe_enforce_sqlite_journal_mode: None, log_dir: None, model_catalog_json: None, check_for_update_on_startup: None, @@ -13835,8 +13836,12 @@ async fn load_with_enterprise_requirement( .await } +#[test_case::test_case(true; "enforce_sqlite_journal_mode")] +#[test_case::test_case(false; "preserve_sqlite_journal_mode")] #[tokio::test] -async fn exact_requirements_apply_to_runtime_config() -> std::io::Result<()> { +async fn exact_requirements_apply_to_runtime_config( + enforce_delete_on_nfs: bool, +) -> std::io::Result<()> { let codex_home = TempDir::new()?; let unused_diagnostics = codex_home.path().join("unused-diagnostics"); std::fs::write(&unused_diagnostics, "not a directory")?; @@ -13855,6 +13860,7 @@ async fn exact_requirements_apply_to_runtime_config() -> std::io::Result<()> { log_dir = {unused_diagnostics:?} check_for_update_on_startup = true allow_login_shell = true +unsafe_enforce_sqlite_journal_mode = false [feedback] enabled = true @@ -13867,6 +13873,7 @@ enabled = true let requirements = format!( r#" sqlite_home = {:?} +unsafe_enforce_sqlite_journal_mode = {enforce_delete_on_nfs} log_dir = {:?} model_catalog_json = {:?} check_for_update_on_startup = false @@ -13881,7 +13888,13 @@ enabled = false ); let config = load_with_enterprise_requirement(&codex_home, requirements).await?; - assert_eq!(config.sqlite.home(), required_sqlite_home.as_path()); + assert_eq!( + config.sqlite, + codex_state::SqliteConfig::new_for_testing(AbsolutePathBuf::try_from( + required_sqlite_home.clone() + )?) + .enable_delete_mode_on_nfs(enforce_delete_on_nfs) + ); assert_eq!(config.log_dir, required_log_dir); assert_eq!( ( diff --git a/codex-rs/core/src/config/mod.rs b/codex-rs/core/src/config/mod.rs index e30c8db539..feb68b2523 100644 --- a/codex-rs/core/src/config/mod.rs +++ b/codex-rs/core/src/config/mod.rs @@ -3429,6 +3429,7 @@ impl Config { cli_auth_credentials_store, chatgpt_base_url: _, sqlite_home: _, + unsafe_enforce_sqlite_journal_mode, log_dir: _, model_catalog_json: _, model_provider: _, @@ -4312,7 +4313,12 @@ impl Config { .cloned() .or(sqlite_home_env) .unwrap_or_else(|| codex_home.clone()); - let mut sqlite = codex_state::SqliteConfig::from_sqlite_home(sqlite_home); + let enforce_delete_on_nfs = + unsafe_enforce_sqlite_journal_mode.is_some_and(|required| required.value); + let mut sqlite = codex_state::SqliteConfig::from_sqlite_home_with_nfs_enforcement( + sqlite_home, + enforce_delete_on_nfs, + ); let requirements = config_layer_stack.requirements_toml(); if cfg.log_dir.is_some() && requirements.sqlite_home.is_none() { sqlite = sqlite.with_local_logs_home(log_dir.clone()); diff --git a/codex-rs/ext/agent-message-board/examples/configure_board.rs b/codex-rs/ext/agent-message-board/examples/configure_board.rs index 978c611b83..712011b3cf 100644 --- a/codex-rs/ext/agent-message-board/examples/configure_board.rs +++ b/codex-rs/ext/agent-message-board/examples/configure_board.rs @@ -13,8 +13,9 @@ async fn main() -> Result<(), Box> { let [_, sqlite_home, board, input] = args.as_slice() else { return Err("Usage: configure_board ".into()); }; - let sqlite = SqliteConfig::from_sqlite_home( + let sqlite = SqliteConfig::from_sqlite_home_with_nfs_enforcement( std::path::absolute(PathBuf::from(sqlite_home))?.try_into()?, + /*enforce_delete_on_nfs*/ false, ); let board = SessionId::from_string(board.to_str().ok_or("invalid board ID")?)?; let input = tokio::fs::read(input).await?; diff --git a/codex-rs/state/Cargo.toml b/codex-rs/state/Cargo.toml index 06eb338963..bc32feccab 100644 --- a/codex-rs/state/Cargo.toml +++ b/codex-rs/state/Cargo.toml @@ -12,6 +12,7 @@ codex-history = { workspace = true } codex-protocol = { workspace = true } codex-utils-absolute-path = { workspace = true } file-id = { workspace = true } +libc = { workspace = true } libsqlite3-sys = { workspace = true } log = { workspace = true } serde = { workspace = true, features = ["derive"] } @@ -20,6 +21,7 @@ sqlx = { workspace = true } sqlx-macros = { workspace = true } sqlx-sqlite = { workspace = true } strum = { workspace = true, features = ["derive"] } +tempfile = { workspace = true } tokio = { workspace = true, features = ["fs", "io-util", "macros", "rt-multi-thread", "sync", "time"] } tracing = { workspace = true } tracing-subscriber = { workspace = true } diff --git a/codex-rs/state/src/lib.rs b/codex-rs/state/src/lib.rs index 78aea77025..5c5ecad54b 100644 --- a/codex-rs/state/src/lib.rs +++ b/codex-rs/state/src/lib.rs @@ -15,6 +15,7 @@ mod extract; pub mod log_db; mod migrations; mod model; +mod nfs_migration; mod paths; mod runtime; mod sqlite; @@ -34,6 +35,8 @@ pub use model::QueuedUserSubmissionRecord; pub use model::RolloutMigrationCursor; pub use model::RolloutMigrationSkippedRollout; pub use model::RolloutMigrationState; +pub use nfs_migration::NfsDbMigration; +pub use nfs_migration::NfsDbMigrationOutcome; pub use runtime::GuardianReviewRecord; pub use runtime::MAX_GUARDIAN_REVIEW_BYTES; pub use runtime::MAX_GUARDIAN_REVIEW_RECORDS; diff --git a/codex-rs/state/src/nfs_migration.rs b/codex-rs/state/src/nfs_migration.rs new file mode 100644 index 0000000000..160eabd046 --- /dev/null +++ b/codex-rs/state/src/nfs_migration.rs @@ -0,0 +1,262 @@ +//! Converts WAL databases to DELETE mode while preserving recoverable WAL backups. +//! Callers serialize startup and migration with the directory lock. Other SQLite clients +//! must remain stopped during replacement; NFS must enforce cross-client POSIX locks. + +use anyhow::Context; +use anyhow::ensure; +use sqlx::ConnectOptions; +use sqlx::Connection; +use sqlx::SqliteConnection; +use sqlx::sqlite::SqliteConnectOptions; +use sqlx::sqlite::SqliteJournalMode; +use sqlx::sqlite::SqliteLockingMode; +use sqlx::sqlite::SqliteSynchronous; +use std::fs::File; +use std::io::ErrorKind; +use std::io::Read; +use std::io::Write; +use std::path::Path; +use std::path::PathBuf; +use std::time::Duration; + +#[cfg(unix)] +mod access; +mod replacement; + +use replacement::Recovery; +use replacement::Replacement; +use replacement::regular_file_exists; + +/// Holds the directory lock shared by automatic startup and the migration command. +pub struct NfsDbMigration { + _lock: File, +} + +/// Result of checking or converting a database, for caller-owned reporting. +pub enum NfsDbMigrationOutcome { + Missing, + AlreadyDelete, + Migrated { backup_directory: PathBuf }, +} + +impl NfsDbMigration { + pub fn lock(sqlite_home: &Path) -> anyhow::Result { + ensure!( + cfg!(unix), + "NFS database migration requires Unix filesystem locking and directory fsync" + ); + // Never remove the lock: concurrent openers must lock the same inode. + let lock = std::fs::OpenOptions::new() + .read(true) + .write(true) + .create(true) + .truncate(false) + .open(sqlite_home.join(".codex-nfs-migration.lock"))?; + lock.try_lock() + .context("another database migration is running")?; + Ok(Self { _lock: lock }) + } + + /// Convert an existing WAL database or resume an interrupted replacement. + /// All other clients must remain stopped until this returns successfully. + #[expect( + clippy::disallowed_methods, + reason = "Migration needs a dedicated exclusive connection, not a runtime pool" + )] + pub async fn migrate( + &mut self, + source: &Path, + staging_dir: &Path, + ) -> anyhow::Result { + let replacement = Replacement::new(source)?; + if let Recovery::Installed = replacement.resume().await? { + return Ok(NfsDbMigrationOutcome::Migrated { + backup_directory: replacement.backup_directory, + }); + } + let exists = regular_file_exists(source)?; + for suffix in ["-wal", "-shm", "-journal"] { + let mut path = source.as_os_str().to_owned(); + path.push(suffix); + let sidecar_exists = regular_file_exists(Path::new(&path))?; + ensure!( + exists || !sidecar_exists, + "database is missing but a SQLite sidecar exists" + ); + } + if !exists { + return Ok(NfsDbMigrationOutcome::Missing); + } + let options = SqliteConnectOptions::new() + .filename(source) + // Set this mode before SQLite first accesses the WAL. + // SQLite then locks the database file without shared memory across NFS client kernels. + .locking_mode(SqliteLockingMode::Exclusive) + .synchronous(SqliteSynchronous::Full) + .busy_timeout(Duration::ZERO) + .disable_statement_logging(); + let mut connection = SqliteConnection::connect_with(&options).await.context( + "cannot open source database exclusively; close its SQLite clients on all hosts", + )?; + + // Closing any descriptor for an inode can release this process's POSIX record locks + // for that inode. + // Keep both source files open until SQLite closes its connection, including after an error. + let mut source_files = Vec::new(); + let result: anyhow::Result = async { + { + let mut handle = connection.lock_handle().await?; + // SAFETY: SQLx's guard prevents its worker from accessing this open connection. + // This db_config option accepts an int and an optional output pointer to an int. + let status = unsafe { + libsqlite3_sys::sqlite3_db_config( + handle.as_raw_handle().as_ptr(), + libsqlite3_sys::SQLITE_DBCONFIG_NO_CKPT_ON_CLOSE, + /*onoff*/ 1, + std::ptr::null_mut::(), + ) + }; + ensure!( + status == libsqlite3_sys::SQLITE_OK, + "cannot disable source checkpoint on close: {status}" + ); + } + let mode: String = sqlx::query_scalar("PRAGMA journal_mode") + .fetch_one(&mut connection) + .await + .context( + "cannot access source database exclusively; close its SQLite clients on all hosts", + )?; + if mode == "delete" { + ensure!( + !regular_file_exists(&replacement.replacement_path())?, + "unfinished backup exists" + ); + return Ok(false); + } + ensure!( + mode == "wal", + "source database must be in WAL mode (found {mode})" + ); + // BEGIN EXCLUSIVE alone blocks only WAL writers. EXCLUSIVE locking_mode also blocks readers. + // The mode keeps the database lock after ROLLBACK. + sqlx::raw_sql("BEGIN EXCLUSIVE; ROLLBACK;") + .execute(&mut connection) + .await + .context( + "cannot lock source database exclusively; close its SQLite clients on all hosts", + )?; + + // A DELETE no-op must not open raw descriptors: closing them could release + // POSIX locks held by another pool in this process. + source_files.push(File::open(source)?); + #[cfg(unix)] + access::require_private_database(&source_files[0])?; + // The private directory protects WAL contents even on a shared /tmp. + let staging = tempfile::Builder::new() + .prefix("codex-nfs-migration-") + .tempdir_in(staging_dir) + .context("cannot create local staging directory")?; + let staged_database = staging.path().join("database.sqlite"); + + let mut wal_path = source.as_os_str().to_owned(); + wal_path.push("-wal"); + match File::open(wal_path) { + Ok(file) => source_files.push(file), + Err(error) if error.kind() == ErrorKind::NotFound => {} + Err(error) => return Err(error).context("cannot open source WAL"), + } + for (file, name) in source_files + .iter_mut() + .zip(["database.sqlite", "database.sqlite-wal"]) + { + let mut target = File::create(staging.path().join(name))?; + copy_buffered(file, &mut target).context("cannot stage source database and WAL")?; + } + let mut copy = SqliteConnection::connect_with( + &SqliteConnectOptions::new() + .filename(&staged_database) + .locking_mode(SqliteLockingMode::Exclusive) + .journal_mode(SqliteJournalMode::Delete) + .synchronous(SqliteSynchronous::Full) + .disable_statement_logging(), + ) + .await + .context("cannot convert staged database to DELETE")?; + let validated = validate_delete_database(&mut copy) + .await + .context("staged database failed validation"); + let closed = copy.close().await; + validated?; + closed?; + + replacement.prepare_backup_directory()?; + let mut temporary = tempfile::NamedTempFile::new_in(&replacement.backup_directory)?; + copy_buffered(&mut File::open(&staged_database)?, temporary.as_file_mut()) + .context("cannot write replacement copy")?; + // Use the open handle. Closing another source descriptor can release SQLite's locks. + temporary + .as_file() + .set_permissions(source_files[0].metadata()?.permissions())?; + #[cfg(unix)] + access::require_private_database(temporary.as_file())?; + temporary.as_file().sync_all()?; + temporary + .into_temp_path() + .persist_noclobber(replacement.replacement_path()) + .context("cannot save replacement copy")?; + File::open(&replacement.backup_directory)?.sync_all()?; + for file in &source_files { + file.sync_all()?; + } + replacement.move_original_to_backup()?; + Ok(true) + } + .await; + let closed = connection.close().await; + drop(source_files); + let changed = result?; + closed.context("cannot close source database. Keep clients stopped")?; + if changed { + replacement.install()?; + return Ok(NfsDbMigrationOutcome::Migrated { + backup_directory: replacement.backup_directory, + }); + } + Ok(NfsDbMigrationOutcome::AlreadyDelete) + } +} + +async fn validate_delete_database(connection: &mut SqliteConnection) -> anyhow::Result<()> { + let mode: String = sqlx::query_scalar("PRAGMA journal_mode") + .fetch_one(&mut *connection) + .await?; + ensure!( + mode == "delete", + "database did not enter DELETE journal mode" + ); + let integrity: String = sqlx::query_scalar("PRAGMA integrity_check(1)") + .fetch_one(connection) + .await + .context("database failed integrity check")?; + ensure!( + integrity == "ok", + "database failed integrity check: {integrity}" + ); + Ok(()) +} + +fn copy_buffered(source: &mut File, destination: &mut File) -> std::io::Result<()> { + // Use large reads to avoid the fallback to small reads for copy_file_range on NFSv4.1. + let mut buffer = vec![0; 1024 * 1024]; + loop { + let bytes = match source.read(&mut buffer) { + Err(error) if error.kind() == ErrorKind::Interrupted => continue, + result => result?, + }; + if bytes == 0 { + return Ok(()); + } + destination.write_all(&buffer[..bytes])?; + } +} diff --git a/codex-rs/cli/src/migrate_nfs_db/access.rs b/codex-rs/state/src/nfs_migration/access.rs similarity index 100% rename from codex-rs/cli/src/migrate_nfs_db/access.rs rename to codex-rs/state/src/nfs_migration/access.rs diff --git a/codex-rs/cli/src/migrate_nfs_db/replacement.rs b/codex-rs/state/src/nfs_migration/replacement.rs similarity index 100% rename from codex-rs/cli/src/migrate_nfs_db/replacement.rs rename to codex-rs/state/src/nfs_migration/replacement.rs diff --git a/codex-rs/state/src/runtime.rs b/codex-rs/state/src/runtime.rs index 58dd668469..2ecd9592c6 100644 --- a/codex-rs/state/src/runtime.rs +++ b/codex-rs/state/src/runtime.rs @@ -148,7 +148,6 @@ impl StateRuntime { let goals_path = sqlite.goals_db_path(); let memories_path = sqlite.memories_db_path(); let queue_path = sqlite.queue_db_path(); - let has_memories_v2 = tokio::fs::try_exists(sqlite.memories_v2_db_path()).await?; let pool = match sqlite .open_state_db(&state_migrator, telemetry_override) .await @@ -159,6 +158,7 @@ impl StateRuntime { return Err(err); } }; + let has_memories_v2 = tokio::fs::try_exists(sqlite.memories_v2_db_path()).await?; let logs_pool = match sqlite .open_logs_db(&logs_migrator, telemetry_override) .await diff --git a/codex-rs/state/src/sqlite.rs b/codex-rs/state/src/sqlite.rs index d68c1fa8e4..8719ac0ad8 100644 --- a/codex-rs/state/src/sqlite.rs +++ b/codex-rs/state/src/sqlite.rs @@ -1,6 +1,6 @@ //! Shared SQLite connection configuration. //! New databases use DELETE on NFS and WAL elsewhere. -//! Writable pools preserve the journal and vacuum modes of initialized databases. +//! Writable pools preserve existing settings unless managed policy requires DELETE on NFS. //! Startup retries SQLITE_BUSY and reports other errors to callers. //! The configured logs directory is created and locked only when opening its database. @@ -32,11 +32,15 @@ use std::time::Instant; use tokio::sync::OnceCell; mod busy_retry; +mod connection; +mod nfs; mod validation; pub use busy_retry::SqlitePoolWriteExt; pub(crate) use busy_retry::SqliteWriteQueryExt; pub(crate) use busy_retry::retry_on_busy; +use connection::JournalModePolicy; +use connection::connect_once; use validation::SqliteQuickCheckManager; const DEFAULT_BUSY_TIMEOUT: Duration = Duration::from_secs(5); @@ -171,6 +175,7 @@ pub struct SqliteConfig { local_logs_dir: Option, is_nfs: Arc>, logs_is_nfs: Arc>, + enforce_delete_on_nfs: bool, quick_check_manager: SqliteQuickCheckManager, } @@ -179,6 +184,7 @@ pub struct SqliteConfig { impl PartialEq for SqliteConfig { fn eq(&self, other: &Self) -> bool { self.sqlite_home == other.sqlite_home + && self.enforce_delete_on_nfs == other.enforce_delete_on_nfs && self.local_logs_dir.as_ref().map(LocalLogsDir::path) == other.local_logs_dir.as_ref().map(LocalLogsDir::path) } @@ -187,18 +193,42 @@ impl PartialEq for SqliteConfig { impl Eq for SqliteConfig {} impl SqliteConfig { + #[deprecated( + note = "Use SqliteConfig::from_sqlite_home_with_nfs_enforcement with the managed unsafe_enforce_sqlite_journal_mode requirement, or SqliteConfig::new_for_testing in tests" + )] pub fn from_sqlite_home(sqlite_home: AbsolutePathBuf) -> Self { + Self::from_sqlite_home_with_nfs_enforcement( + sqlite_home, + /*enforce_delete_on_nfs*/ false, + ) + } + + /// Construct the configuration with the resolved `unsafe_enforce_sqlite_journal_mode` requirement. + pub fn from_sqlite_home_with_nfs_enforcement( + sqlite_home: AbsolutePathBuf, + enforce_delete_on_nfs: bool, + ) -> Self { Self { sqlite_home, local_logs_dir: None, is_nfs: Arc::new(OnceCell::new()), logs_is_nfs: Arc::new(OnceCell::new()), + enforce_delete_on_nfs, quick_check_manager: SqliteQuickCheckManager::new(), } } + /// When enforcement is enabled, migrate existing WAL databases on NFS before opening them. + pub fn enable_delete_mode_on_nfs(mut self, enabled: bool) -> Self { + self.enforce_delete_on_nfs = enabled; + self + } + pub fn new_for_testing(sqlite_home: AbsolutePathBuf) -> Self { - Self::from_sqlite_home(sqlite_home) + Self::from_sqlite_home_with_nfs_enforcement( + sqlite_home, + /*enforce_delete_on_nfs*/ false, + ) } pub fn home(&self) -> &Path { @@ -428,7 +458,16 @@ impl SqliteConfig { } else { probe_nfs(directory, &OnceCell::new()).await? }; - let journal_mode_policy = if is_nfs { + // Keep the migration lock through pool initialization so another managed startup + // cannot open the old inode or initialize a database during cutover. + let migration = if self.enforce_delete_on_nfs && is_nfs { + Some(nfs::prepare_database(self, path).await?) + } else { + None + }; + let journal_mode_policy = if migration.is_some() { + JournalModePolicy::RequireDelete + } else if is_nfs { JournalModePolicy::InitializeDelete } else { JournalModePolicy::InitializeWal @@ -506,6 +545,7 @@ impl SqliteConfig { } /// Open old logs for cleanup without creating, migrating, or rebuilding the database. + /// Skip legacy NFS logs when managed DELETE enforcement is active. pub(crate) async fn open_legacy_logs_for_cleanup(&self) -> anyhow::Result> { let Some(path) = self.legacy_logs_db_path() else { return Ok(None); @@ -513,6 +553,9 @@ impl SqliteConfig { if !tokio::fs::try_exists(&path).await? { return Ok(None); } + if self.enforce_delete_on_nfs && self.is_nfs().await? { + return Ok(None); + } let options = SqliteConnectOptions::new() .filename(path) .create_if_missing(false) @@ -539,105 +582,6 @@ impl SqliteConfig { } } -#[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); - 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?; - } - 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?; - } - // 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?; - if journal == "wal" { - 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. - connection.return_to_pool().await; - Ok(()) - } - .await; - if let Err(error) = initialized { - pool.close().await; - return Err(error); - } - Ok(pool) -} - async fn probe_nfs(home: &Path, probe: &OnceCell) -> anyhow::Result { probe .get_or_try_init(|| async { diff --git a/codex-rs/state/src/sqlite/connection.rs b/codex-rs/state/src/sqlite/connection.rs new file mode 100644 index 0000000000..0d74e91532 --- /dev/null +++ b/codex-rs/state/src/sqlite/connection.rs @@ -0,0 +1,122 @@ +//! Initializes writable connections while preserving existing database settings. +//! Only empty databases adopt the requested journal mode; DELETE connections use EXTRA sync. +//! Managed DELETE requirements also reject other modes on every connection. + +#![expect( + clippy::disallowed_methods, + reason = "centralized SQLite connection initialization" +)] + +use sqlx::Error; +use sqlx::SqlitePool; +use sqlx::sqlite::SqliteConnectOptions; +use sqlx::sqlite::SqlitePoolOptions; + +#[derive(Clone, Copy, Eq, PartialEq)] +pub(super) enum JournalModePolicy { + InitializeWal, + InitializeDelete, + RequireDelete, +} + +pub(super) 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); + 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_mode_policy == JournalModePolicy::RequireDelete + && journal != "delete" + { + return Err(Error::Protocol(format!( + "NFS database requires DELETE journaling, found {journal}" + ))); + } + 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?; + } + let pragma = match journal_mode_policy { + JournalModePolicy::InitializeWal => "PRAGMA journal_mode = WAL", + JournalModePolicy::InitializeDelete | JournalModePolicy::RequireDelete => { + "PRAGMA journal_mode = DELETE" + } + }; + sqlx::query(pragma).execute(&mut *connection).await?; + } + // 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?; + if journal == "wal" { + 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. + connection.return_to_pool().await; + Ok(()) + } + .await; + if let Err(error) = initialized { + pool.close().await; + return Err(error); + } + Ok(pool) +} diff --git a/codex-rs/state/src/sqlite/nfs.rs b/codex-rs/state/src/sqlite/nfs.rs new file mode 100644 index 0000000000..1a0386462f --- /dev/null +++ b/codex-rs/state/src/sqlite/nfs.rs @@ -0,0 +1,66 @@ +//! Migrates databases after the caller checks the managed policy and detects NFS. +//! Migration errors must not trigger corruption recovery that would discard a WAL backup. + +use super::SqliteConfig; +use crate::NfsDbMigration; +use crate::NfsDbMigrationOutcome; +use anyhow::Context; +use std::fs::TryLockError; +use std::path::Path; +use std::time::Duration; +use tokio::time::Instant; + +pub(super) async fn prepare_database( + sqlite: &SqliteConfig, + path: &Path, +) -> anyhow::Result { + // Replacement syncs the source directory, which must not be an empty relative parent. + let absolute_path = std::path::absolute(path)?; + let path = absolute_path.as_path(); + let home = path + .parent() + .context("database path has no parent directory")?; + // Concurrent managed startups can briefly hold this lock even after migration is complete. + let deadline = Instant::now() + super::DEFAULT_BUSY_TIMEOUT; + let mut migration = loop { + match NfsDbMigration::lock(home) { + Ok(migration) => break migration, + Err(error) + if matches!( + error.downcast_ref::(), + Some(TryLockError::WouldBlock) + ) && Instant::now() < deadline => + { + tokio::time::sleep(Duration::from_millis(/*millis*/ 100)).await; + } + Err(error) => return Err(error), + } + }; + // The primary runtime opens state first. Include lazy databases so startup also + // resumes interrupted migrations whose source path has not been reinstalled yet. + // A separate logs directory is checked and locked when its own database opens. + let paths = if path == sqlite.state_db_path() { + sqlite + .runtime_db_paths() + .into_iter() + .map(|db| db.path) + .filter(|path| path.parent() == Some(home)) + .collect() + } else { + vec![path.to_path_buf()] + }; + for path in paths { + let outcome = migration.migrate(&path, &std::env::temp_dir()).await.map_err(|error| { + // Do not expose a SQLite corruption error to automatic backup-and-rebuild recovery: + // an interrupted cutover must be resumed by the migration implementation itself. + anyhow::anyhow!( + "required NFS database migration failed for {}: {error:#}. Stop Codex instances on all hosts and rerun codex migrate-nfs-db", + path.display() + ) + })?; + if let NfsDbMigrationOutcome::Migrated { backup_directory } = outcome { + tracing::info!(database = %path.display(), backup = %backup_directory.display(), "Migrated NFS database to DELETE journaling"); + } + } + Ok(migration) +} diff --git a/codex-rs/state/src/sqlite_tests.rs b/codex-rs/state/src/sqlite_tests.rs index bb674101b5..e6d0548304 100644 --- a/codex-rs/state/src/sqlite_tests.rs +++ b/codex-rs/state/src/sqlite_tests.rs @@ -163,19 +163,26 @@ 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 (nfs, partial_journal) in [ - (false, None), - (false, Some(SqliteJournalMode::Delete)), - (true, None), - (true, Some(SqliteJournalMode::Delete)), - (true, Some(SqliteJournalMode::Wal)), + for (enforced, nfs, partial_journal) in [ + (false, false, None), + (false, false, Some(SqliteJournalMode::Delete)), + (false, true, None), + (false, true, Some(SqliteJournalMode::Delete)), + (false, true, Some(SqliteJournalMode::Wal)), + (true, false, None), + (true, false, Some(SqliteJournalMode::Delete)), + (true, cfg!(unix), None), + (true, cfg!(unix), Some(SqliteJournalMode::Delete)), ] { 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 sqlite = SqliteConfig::from_sqlite_home_with_nfs_enforcement( + sqlite_home.as_path().abs(), + enforced, + ); sqlite.is_nfs.set(nfs)?; let path = sqlite.logs_db_path(); if let Some(journal_mode) = partial_journal { @@ -311,7 +318,9 @@ async fn open_read_write_pool_preserves_delete_under_write_lock() -> anyhow::Res 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 sqlite = SqliteConfig::new_for_testing(sqlite_home.as_path().abs()) + .enable_delete_mode_on_nfs(/*enabled*/ true); + sqlite.is_nfs.set(cfg!(unix))?; let database_path = sqlite.logs_db_path(); let existing_pool = SqlitePoolOptions::new() .max_connections(1) @@ -425,3 +434,142 @@ async fn open_read_write_pool_preserves_empty_database_lock_error() -> anyhow::R pool.close().await; Ok(()) } + +#[cfg(unix)] +#[tokio::test] +async fn required_nfs_migration_preserves_runtime_data_and_backups_after_restart() +-> anyhow::Result<()> { + use std::os::unix::fs::PermissionsExt; + + let home = tempfile::tempdir()?; + let original = SqliteConfig::new_for_testing(home.path().abs()); + let runtime = crate::StateRuntime::init(original.clone(), "test-provider".into()).await?; + runtime.checkpoint_backfill("before-migration").await?; + let before = runtime.get_backfill_state().await?; + runtime.close().await; + original.open_memories_v2_db().await?.close().await; + crate::open_thread_history_db(&original) + .await? + .close() + .await; + let mut backups = Vec::new(); + for db in original + .runtime_db_paths() + .into_iter() + .filter(|db| db.path.exists()) + { + std::fs::set_permissions(&db.path, std::fs::Permissions::from_mode(/*mode*/ 0o600))?; + let name = db.path.file_name().unwrap().to_str().unwrap(); + backups.push(( + home.path() + .join(format!("{name}_wal_migration_backup")) + .join(name), + std::fs::read(&db.path)?, + )); + } + let sqlite = SqliteConfig::from_sqlite_home_with_nfs_enforcement( + home.path().abs(), + /*enforce_delete_on_nfs*/ true, + ); + sqlite.is_nfs.set(/*value*/ true)?; + let migrated = crate::StateRuntime::init(sqlite.clone(), "test-provider".into()).await?; + assert_eq!(migrated.get_backfill_state().await?, before); + migrated.checkpoint_backfill("after-migration").await?; + let after = migrated.get_backfill_state().await?; + migrated.close().await; + + // Simulate a crash after moving the original but before reinstalling a lazy database. + let lazy = sqlite.memories_v2_db_path(); + let name = lazy.file_name().unwrap().to_str().unwrap(); + std::fs::rename( + &lazy, + home.path() + .join(format!("{name}_wal_migration_backup/delete.sqlite")), + )?; + let restarted = crate::StateRuntime::init(sqlite.clone(), "test-provider".into()).await?; + assert_eq!(restarted.get_backfill_state().await?, after); + restarted.close().await; + // Include lazily created databases: they must also start in DELETE mode. + for db in sqlite.runtime_db_paths() { + let pool = sqlite.open_read_write_pool(&db.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, ("delete".into(), 3)); + pool.close().await; + } + for (path, bytes) in backups { + assert_eq!(std::fs::read(path)?, bytes); + } + Ok(()) +} + +#[cfg(unix)] +#[tokio::test] +async fn required_nfs_migration_refuses_active_wal_and_retries_after_close() -> anyhow::Result<()> { + use std::future::Future; + use std::os::unix::fs::PermissionsExt; + use std::task::Poll; + + let home = tempfile::tempdir()?; + let logs_home = home.path().join("local-logs"); + let sqlite = + SqliteConfig::new_for_testing(home.path().abs()).with_local_logs_home(logs_home.clone()); + let logs = sqlite.open_read_write_pool(&sqlite.logs_db_path()).await?; + sqlx::query("CREATE TABLE existing_logs (id INTEGER PRIMARY KEY)") + .execute(&logs) + .await?; + let path = sqlite.state_db_path(); + let pool = sqlite.open_read_write_pool(&path).await?; + sqlx::raw_sql("CREATE TABLE entries(id); INSERT INTO entries VALUES(42);") + .execute(&pool) + .await?; + std::fs::set_permissions(&path, std::fs::Permissions::from_mode(/*mode*/ 0o600))?; + let required = SqliteConfig::new_for_testing(home.path().abs()) + .with_local_logs_home(logs_home) + .enable_delete_mode_on_nfs(/*enabled*/ true); + required.is_nfs.set(/*value*/ true)?; + required.logs_is_nfs.set(/*value*/ false)?; + let error = required.open_read_write_pool(&path).await.unwrap_err(); + assert!( + error + .to_string() + .contains("required NFS database migration failed"), + "{error:#}" + ); + assert!(!crate::is_sqlite_corruption_error(&error)); + pool.close().await; + let migration_lock = crate::NfsDbMigration::lock(home.path())?; + let mut opening = Box::pin(required.open_read_write_pool(&path)); + std::future::poll_fn(|cx| { + assert!(opening.as_mut().poll(cx).is_pending()); + Poll::Ready(()) + }) + .await; + drop(migration_lock); + let migrated = opening.await?; + let rows: Vec = sqlx::query_scalar("SELECT id FROM entries") + .fetch_all(&migrated) + .await?; + assert_eq!(rows, vec![42]); + migrated.close().await; + // An active local logs database must remain in WAL while the NFS state database migrates. + let journal: String = sqlx::query_scalar("PRAGMA journal_mode") + .fetch_one(&logs) + .await?; + assert_eq!(journal, "wal"); + logs.close().await; + // Old versions may create legacy WAL logs after startup. Managed cleanup must not write them. + let legacy = sqlite + .open_read_write_pool(&sqlite.legacy_logs_db_path().unwrap()) + .await?; + assert!(required.open_legacy_logs_for_cleanup().await?.is_none()); + let journal: String = sqlx::query_scalar("PRAGMA journal_mode") + .fetch_one(&legacy) + .await?; + assert_eq!(journal, "wal"); + legacy.close().await; + Ok(()) +} diff --git a/codex-rs/thread-manager-sample/src/main.rs b/codex-rs/thread-manager-sample/src/main.rs index afc6155974..47792f00ac 100644 --- a/codex-rs/thread-manager-sample/src/main.rs +++ b/codex-rs/thread-manager-sample/src/main.rs @@ -338,7 +338,10 @@ async fn new_config( agent_max_depth: 1, agent_roles: BTreeMap::new(), memories: MemoriesConfig::default(), - sqlite: SqliteConfig::from_sqlite_home(codex_home.clone()), + sqlite: SqliteConfig::from_sqlite_home_with_nfs_enforcement( + codex_home.clone(), + /*enforce_delete_on_nfs*/ false, + ), log_dir: codex_home.join("log").to_path_buf(), codex_home, history: History::default(), diff --git a/codex-rs/tui/src/debug_config.rs b/codex-rs/tui/src/debug_config.rs index 01a7802d06..01842c2f1e 100644 --- a/codex-rs/tui/src/debug_config.rs +++ b/codex-rs/tui/src/debug_config.rs @@ -955,6 +955,7 @@ interrupt_message = false cli_auth_credentials_store: None, chatgpt_base_url: None, sqlite_home: Some(sqlite_home), + unsafe_enforce_sqlite_journal_mode: None, log_dir: Some(log_dir), model_catalog_json: Some(model_catalog_json), model_provider: None,