Fix Antares append and unmount FUSE races

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 {

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::<u64>() {
+            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<Option<AntaresConfig>> {
         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);
+    }
+}

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