diff --git a/src/antares/fuse.rs b/src/antares/fuse.rs index 1bc43bf..2d04d66 100644 --- a/src/antares/fuse.rs +++ b/src/antares/fuse.rs @@ -94,8 +94,11 @@ impl AntaresFuse { let overlay = self.build_overlay().await?; let logfs = LoggingFileSystem::new(overlay); + // Keep Antares mounts on the safer non-writeback path for now. + // With writeback cache enabled, reopening an existing file in append mode + // can fail inside libfuse-fs passthrough I/O with EBADF. let handle = - mount_filesystem_with_antares_cache(logfs, self.mountpoint.as_os_str(), true).await; + mount_filesystem_with_antares_cache(logfs, self.mountpoint.as_os_str(), false).await; // Spawn background task to run the FUSE session let fuse_task = tokio::spawn(async move { diff --git a/src/antares/mod.rs b/src/antares/mod.rs index 0058c4f..5feb5de 100644 --- a/src/antares/mod.rs +++ b/src/antares/mod.rs @@ -73,6 +73,24 @@ use crate::{ util::config, }; +fn unmount_grace_duration() -> std::time::Duration { + const DEFAULT_MS: u64 = 150; + match std::env::var("ANTARES_UNMOUNT_GRACE_MS") { + Ok(raw) => match raw.trim().parse::() { + Ok(ms) => std::time::Duration::from_millis(ms.clamp(0, 3_000)), + Err(_) => { + tracing::warn!( + value = %raw, + default_ms = DEFAULT_MS, + "invalid ANTARES_UNMOUNT_GRACE_MS, using default" + ); + std::time::Duration::from_millis(DEFAULT_MS) + } + }, + Err(_) => std::time::Duration::from_millis(DEFAULT_MS), + } +} + /// Global paths used by Antares to place layers and state. #[derive(Debug, Clone)] pub struct AntaresPaths { @@ -308,15 +326,19 @@ impl AntaresManager { pub async fn umount_job(&self, job_id: &str) -> std::io::Result> { use tracing::{info, warn}; - // Lock and get the config, but do not remove yet - let mut instances = self.instances.lock().await; - let config = match instances.get(job_id) { + // Look up config first so we can quiesce without holding the state lock. + let config = match self.instances.lock().await.get(job_id) { Some(cfg) => cfg.clone(), None => return Ok(None), }; let mount_path = &config.mountpoint; info!("Attempting to unmount FUSE mount at {:?}", mount_path); + let grace = unmount_grace_duration(); + if !grace.is_zero() { + info!("Quiescing {:?} for {:?} before unmount", mount_path, grace); + tokio::time::sleep(grace).await; + } // Try to unmount via the stored FUSE handle first (proper teardown) let mut fuse_handles = self.fuse_handles.lock().await; @@ -366,6 +388,7 @@ impl AntaresManager { drop(fuse_handles); // Remove from bookkeeping and persist (even if unmount failed) + let mut instances = self.instances.lock().await; let removed = instances.remove(job_id); drop(instances); self.persist_state().await?; @@ -411,3 +434,58 @@ impl AntaresManager { Ok(()) } } + +#[cfg(test)] +mod tests { + use super::unmount_grace_duration; + use serial_test::serial; + + fn set_unmount_grace_env(value: Option<&str>) { + // SAFETY: tests mutate process env in a controlled way and do not run in parallel here. + unsafe { + match value { + Some(value) => std::env::set_var("ANTARES_UNMOUNT_GRACE_MS", value), + None => std::env::remove_var("ANTARES_UNMOUNT_GRACE_MS"), + } + } + } + + #[test] + #[serial] + fn test_unmount_grace_duration_defaults_to_150ms() { + set_unmount_grace_env(None); + assert_eq!( + unmount_grace_duration(), + std::time::Duration::from_millis(150) + ); + } + + #[test] + #[serial] + fn test_unmount_grace_duration_accepts_explicit_value() { + set_unmount_grace_env(Some("275")); + assert_eq!( + unmount_grace_duration(), + std::time::Duration::from_millis(275) + ); + set_unmount_grace_env(None); + } + + #[test] + #[serial] + fn test_unmount_grace_duration_clamps_and_falls_back() { + set_unmount_grace_env(Some("50000")); + assert_eq!( + unmount_grace_duration(), + std::time::Duration::from_millis(3_000) + ); + + set_unmount_grace_env(Some("not-a-number")); + assert_eq!( + unmount_grace_duration(), + std::time::Duration::from_millis(150) + ); + + set_unmount_grace_env(None); + } +} diff --git a/tests/antares_test.rs b/tests/antares_test.rs index 93e9565..daaef11 100644 --- a/tests/antares_test.rs +++ b/tests/antares_test.rs @@ -21,6 +21,7 @@ use scorpiofs::{ util::config, }; use serial_test::serial; +use std::io::Write; use std::path::PathBuf; use tempfile::tempdir; use tokio::time::{sleep, Duration}; @@ -420,6 +421,78 @@ async fn test_fuse_multiple_custom_mounts() { } } +/// Test that appending to an existing file on an Antares FUSE mount succeeds. +/// +/// This covers the reproduced regression where reopening an existing file with +/// append mode could fail with `EBADF`. +/// +/// Run with: +/// sudo -E cargo test --test antares_test test_fuse_append_existing_file -- --exact --ignored --nocapture +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[ignore] +#[serial] +async fn test_fuse_append_existing_file() { + let test_future = async { + init_config(); + + if !fuse_prereqs_available() { + return; + } + + let test_id = Uuid::new_v4(); + let base = PathBuf::from(format!("/tmp/antares_append_test_{test_id}")); + let _ = std::fs::remove_dir_all(&base); + + let mountpoint = base.join("workspace"); + let paths = AntaresPaths::new( + base.join("upper"), + base.join("cl"), + base.join("mnt"), + base.join("state.toml"), + ); + let manager = AntaresManager::new(paths).await; + + let config = manager + .mount_job_at("append-test-job", mountpoint.clone(), None) + .await + .expect("mount_job_at should succeed"); + sleep(Duration::from_millis(500)).await; + + let file_path = config.mountpoint.join("event.jsonl"); + + std::fs::write(&file_path, b"{\"seq\":1}\n").expect("initial write should succeed"); + + let mut append_file = std::fs::OpenOptions::new() + .append(true) + .open(&file_path) + .expect("opening existing file in append mode should succeed"); + append_file + .write_all(b"{\"seq\":2}\n") + .expect("appending to existing file should succeed"); + append_file + .flush() + .expect("flush after append should succeed"); + append_file + .sync_all() + .expect("sync_all after append should succeed"); + drop(append_file); + + let content = std::fs::read_to_string(&file_path).expect("reading appended file"); + assert_eq!(content, "{\"seq\":1}\n{\"seq\":2}\n"); + + manager + .umount_job("append-test-job") + .await + .expect("manager umount should succeed"); + let _ = std::fs::remove_dir_all(&base); + }; + + match tokio::time::timeout(Duration::from_secs(120), test_future).await { + Ok(_) => println!("✓ Test passed"), + Err(_) => panic!("Test timed out"), + } +} + /// Mount a job (cl = None) and keep FUSE running until Ctrl-C. /// /// Uses multi-thread tokio runtime so FUSE can handle concurrent kernel requests