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/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<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_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)
+}

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<NfsDbMigration> {
+    // 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::<TryLockError>(),
+                    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)
+}