diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 41e6cd96ac..23e5f20b74 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -3485,6 +3485,7 @@ dependencies = [ "codex-utils-absolute-path", "codex-utils-path-uri", "futures", + "rustix 1.1.4", "serde", ] @@ -4658,6 +4659,7 @@ version = "0.0.0" dependencies = [ "anyhow", "chrono", + "codex-file-system", "codex-git-utils", "codex-history", "codex-protocol", diff --git a/codex-rs/file-system/Cargo.toml b/codex-rs/file-system/Cargo.toml index 382eacb43d..006574084f 100644 --- a/codex-rs/file-system/Cargo.toml +++ b/codex-rs/file-system/Cargo.toml @@ -15,6 +15,9 @@ codex-utils-path-uri = { workspace = true } futures = { workspace = true } serde = { workspace = true, features = ["derive"] } +[target.'cfg(target_os = "linux")'.dependencies] +rustix = { workspace = true, features = ["fs"] } + [lib] test = false doctest = false diff --git a/codex-rs/file-system/src/lib.rs b/codex-rs/file-system/src/lib.rs index 55ecf43f8d..1687d006f5 100644 --- a/codex-rs/file-system/src/lib.rs +++ b/codex-rs/file-system/src/lib.rs @@ -1,6 +1,7 @@ mod environment_accessor; mod exec_permission_profile_serde; mod find_up; +mod nfs; use bytes::Bytes; use codex_protocol::config_types::WindowsSandboxLevel; @@ -28,6 +29,7 @@ pub use find_up::FindUpErrorPolicy; pub use find_up::find_nearest_ancestor_with_markers; pub use find_up::find_nearest_native_ancestor_with_markers; use futures::Stream; +pub use nfs::is_nfs; use serde::Deserialize; use serde::Serialize; use std::future::Future; diff --git a/codex-rs/file-system/src/nfs.rs b/codex-rs/file-system/src/nfs.rs new file mode 100644 index 0000000000..ae5718be82 --- /dev/null +++ b/codex-rs/file-system/src/nfs.rs @@ -0,0 +1,16 @@ +//! Detects NFS on Linux; other platforms retain their existing filesystem behavior. +//! Closing an extra file descriptor could release process-wide POSIX locks on its inode. + +use std::io; +use std::path::Path; + +/// On linux we sometimes do a special treatment for NFS related path. We don't support windows NFS +#[cfg(target_os = "linux")] +pub fn is_nfs(path: &Path) -> io::Result { + Ok(rustix::fs::statfs(path)?.f_type == rustix::fs::NFS_SUPER_MAGIC) +} + +#[cfg(not(target_os = "linux"))] +pub fn is_nfs(_path: &Path) -> io::Result { + Ok(false) +} diff --git a/codex-rs/state/Cargo.toml b/codex-rs/state/Cargo.toml index d98de0da11..06eb338963 100644 --- a/codex-rs/state/Cargo.toml +++ b/codex-rs/state/Cargo.toml @@ -7,6 +7,7 @@ license.workspace = true [dependencies] anyhow = { workspace = true } chrono = { workspace = true } +codex-file-system = { workspace = true } codex-history = { workspace = true } codex-protocol = { workspace = true } codex-utils-absolute-path = { workspace = true } diff --git a/codex-rs/state/src/runtime/reclamation.rs b/codex-rs/state/src/runtime/reclamation.rs index 27f89d612a..952aead663 100644 --- a/codex-rs/state/src/runtime/reclamation.rs +++ b/codex-rs/state/src/runtime/reclamation.rs @@ -1,4 +1,5 @@ //! Reclaims unused SQLite pages in opted-in databases, reducing batches after stalled passes. +//! NFS on Linux skips opening a maintenance connection. use crate::RuntimeDbPath; use crate::SqliteConfig; use log::LevelFilter; @@ -64,7 +65,7 @@ impl SqliteReclamationWorker { if owner.is_none() { continue; } - delay = visit(&mut scheduled, &receiver).await; + delay = visit(&sqlite, &mut scheduled, &receiver).await; } }); @@ -79,6 +80,7 @@ impl SqliteReclamationWorker { } async fn visit( + sqlite: &SqliteConfig, scheduled: &mut [(RuntimeDbPath, Instant, ReclamationState)], shutdown: &watch::Receiver<()>, ) -> Duration { @@ -91,6 +93,7 @@ async fn visit( continue; } let result = reclaim( + sqlite, &db.path, ReclamationOptions { budget: Budget { @@ -206,10 +209,18 @@ impl From for ReclamationOptions { } async fn reclaim( + sqlite: &SqliteConfig, path: &Path, options: impl Into, shutdown: &watch::Receiver<()>, ) -> anyhow::Result { + // do not perform autovacuum and cleanup on NFS as it creates a lot of load and contention on IO + if sqlite.is_nfs().await? { + return Ok(ReclamationPass { + pages: 0, + outcome: PassOutcome::Idle, + }); + } if !tokio::fs::try_exists(path).await? { return Ok(ReclamationPass { pages: 0, diff --git a/codex-rs/state/src/runtime/reclamation_tests.rs b/codex-rs/state/src/runtime/reclamation_tests.rs index 4ae711acf9..fb15e3602d 100644 --- a/codex-rs/state/src/runtime/reclamation_tests.rs +++ b/codex-rs/state/src/runtime/reclamation_tests.rs @@ -93,6 +93,7 @@ async fn interrupted_reclamation_releases_writer_and_resumes_without_data_loss() .await?; let (_shutdown, receiver) = watch::channel(()); let resumed = reclaim( + &sqlite, &sqlite.logs_db_path(), Budget { deadline: Some(Instant::now() + Duration::from_secs(30)), @@ -104,16 +105,21 @@ async fn interrupted_reclamation_releases_writer_and_resumes_without_data_loss() let free_after_retry: i64 = sqlx::query_scalar("PRAGMA freelist_count") .fetch_one(&pool) .await?; - assert!(resumed.pages > 0); assert_eq!( free_before_retry - free_after_retry, i64::from(resumed.pages) ); let bytes_after_retry = tokio::fs::metadata(sqlite.logs_db_path()).await?.len(); - assert!( - bytes_after_retry < bytes_before_retry, - "reclamation must shrink the SQLite file: {bytes_before_retry} -> {bytes_after_retry} bytes" - ); + if sqlite.is_nfs().await? { + assert_eq!(resumed.pages, 0); + assert_eq!(bytes_after_retry, bytes_before_retry); + } else { + assert!(resumed.pages > 0); + assert!( + bytes_after_retry < bytes_before_retry, + "reclamation must shrink the SQLite file: {bytes_before_retry} -> {bytes_after_retry} bytes" + ); + } assert_eq!( sqlx::query_as::<_, (i64, String)>(rows) .fetch_all(&pool) diff --git a/codex-rs/state/src/sqlite.rs b/codex-rs/state/src/sqlite.rs index b082f2622d..5ac878f35d 100644 --- a/codex-rs/state/src/sqlite.rs +++ b/codex-rs/state/src/sqlite.rs @@ -23,8 +23,10 @@ use sqlx::sqlite::SqlitePoolOptions; use sqlx::sqlite::SqliteSynchronous; use std::path::Path; use std::path::PathBuf; +use std::sync::Arc; use std::time::Duration; use std::time::Instant; +use tokio::sync::OnceCell; mod busy_retry; mod validation; @@ -154,14 +156,16 @@ pub struct RuntimeDbPath { } /// Resolved configuration shared by all Codex SQLite connections. -/// Clones share quick-check attempts; a newly constructed config checks independently. +/// Clones share filesystem probes and quick-check attempts; a new config checks independently. #[derive(Clone, Debug)] pub struct SqliteConfig { sqlite_home: AbsolutePathBuf, + is_nfs: Arc>, quick_check_manager: SqliteQuickCheckManager, } -// Validation history is operational state, not part of configuration equality. +// Cached filesystem information and validation history are operational state, +// not part of configuration equality. impl PartialEq for SqliteConfig { fn eq(&self, other: &Self) -> bool { self.sqlite_home == other.sqlite_home @@ -174,6 +178,7 @@ impl SqliteConfig { pub fn from_sqlite_home(sqlite_home: AbsolutePathBuf) -> Self { Self { sqlite_home, + is_nfs: Arc::new(OnceCell::new()), quick_check_manager: SqliteQuickCheckManager::new(), } } @@ -186,6 +191,20 @@ impl SqliteConfig { self.sqlite_home.as_path() } + /// Detects if the sqlite_home directory is under the NFS mount (once per the lifetime) + pub async fn is_nfs(&self) -> anyhow::Result { + self.is_nfs + .get_or_try_init(|| async { + let sqlite_home = self.home().to_path_buf(); + Ok( + tokio::task::spawn_blocking(move || codex_file_system::is_nfs(&sqlite_home)) + .await??, + ) + }) + .await + .copied() + } + /// Return the path to the primary state database. pub fn state_db_path(&self) -> PathBuf { STATE_DB.path(self.home()) diff --git a/codex-rs/state/src/sqlite_tests.rs b/codex-rs/state/src/sqlite_tests.rs index 5ebbab9cab..19bc89432f 100644 --- a/codex-rs/state/src/sqlite_tests.rs +++ b/codex-rs/state/src/sqlite_tests.rs @@ -1,4 +1,4 @@ -//! Regression coverage for SQLite initialization under competing database locks. +//! Regression coverage for SQLite initialization, cached filesystem detection, and competing locks. use super::SqliteConfig; use codex_utils_absolute_path::test_support::PathExt; @@ -8,6 +8,60 @@ use sqlx::sqlite::SqliteConnectOptions; use sqlx::sqlite::SqliteJournalMode; use sqlx::sqlite::SqlitePoolOptions; +#[cfg(target_os = "linux")] +#[tokio::test] +async fn sqlite_home_filesystem_probe_is_shared_by_clones_until_config_replacement() +-> 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 clone = sqlite.clone(); + let is_nfs = sqlite.is_nfs().await?; + tokio::fs::remove_dir(&sqlite_home).await?; + // This clone has never queried the home. It must reuse the original's probe, + // because a fresh filesystem lookup would fail now that the home is absent. + assert_eq!(clone.is_nfs().await?, is_nfs); + + let fresh = SqliteConfig::new_for_testing(sqlite_home.as_path().abs()); + assert_eq!( + fresh + .is_nfs() + .await + .unwrap_err() + .downcast_ref::() + .unwrap() + .kind(), + std::io::ErrorKind::NotFound + ); + Ok(()) +} + +#[cfg(target_os = "linux")] +#[tokio::test] +async fn failed_sqlite_home_filesystem_probe_recovers_after_home_creation() -> anyhow::Result<()> { + let sqlite_home = crate::runtime::test_support::unique_temp_dir(); + 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()); + assert_eq!( + sqlite + .is_nfs() + .await + .unwrap_err() + .downcast_ref::() + .unwrap() + .kind(), + std::io::ErrorKind::NotFound + ); + tokio::fs::create_dir_all(&sqlite_home).await?; + sqlite.is_nfs().await?; + Ok(()) +} + #[tokio::test] async fn open_read_write_pool_initializes_fresh_database_settings() -> anyhow::Result<()> { for interrupted in [false, true] {