fix: file lock and acc file load

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "scorpiofs"
-version = "0.1.0"
+version = "0.2.1"
 edition = "2021"
 description = "FUSE-based virtual filesystem with Antares overlay for monorepo builds"
 license = "MIT OR Apache-2.0"
@@ -39,7 +39,7 @@ async-recursion = "1.1.1"
 bytes = "1.11.1"
 futures = "0.3.31"
 uuid = { version = "1.20.0", features = ["v4"] }
-libfuse-fs = { version = "0.1.11"}
+libfuse-fs = { version = "0.1.12"}
 whoami = "1.6.0"
 thiserror = "2.0.18"
 crossbeam = "0.8.4"

scorpio.toml

@@ -6,7 +6,7 @@ git_email = "admin@mega.org"
 workspace = "/tmp/scorpio-megadir/mount"
 base_url = "http://git.gitmega.com"
 dicfuse_readable = "true"
-load_dir_depth = "3"
+load_dir_depth = "5"
 fetch_file_thread = "10"
 dicfuse_import_concurrency = "4"
 dicfuse_dir_sync_ttl_secs = "5"

src/antares/fuse.rs

@@ -105,10 +105,38 @@ impl AntaresFuse {
 
         self.fuse_task = Some(fuse_task);
 
-        tracing::info!(
-            "Mount spawned for {}; FUSE session running (Dicfuse may still be loading in background)",
-            self.mountpoint.display()
-        );
+        // Readiness probe: wait until the FUSE mount is actually servicing requests.
+        // Without this, callers (e.g., Buck2) that immediately stat() the mountpoint
+        // may race against the kernel FUSE_INIT handshake and get ENOTCONN (errno 107).
+        let mp = self.mountpoint.clone();
+        let probe_timeout = std::time::Duration::from_secs(10);
+        let probe_interval = std::time::Duration::from_millis(50);
+        let probe_start = std::time::Instant::now();
+        loop {
+            match tokio::fs::metadata(&mp).await {
+                Ok(_) => {
+                    tracing::info!(
+                        "FUSE mount ready at {} (probe took {:.2}s)",
+                        mp.display(),
+                        probe_start.elapsed().as_secs_f64()
+                    );
+                    break;
+                }
+                Err(e) => {
+                    if probe_start.elapsed() >= probe_timeout {
+                        tracing::warn!(
+                            "FUSE mount probe timed out for {} after {:.1}s: {}",
+                            mp.display(),
+                            probe_timeout.as_secs_f64(),
+                            e
+                        );
+                        break;
+                    }
+                    tokio::time::sleep(probe_interval).await;
+                }
+            }
+        }
+
         Ok(())
     }
 

src/antares/mod.rs

@@ -129,12 +129,13 @@ struct AntaresState {
 }
 
 /// Manager responsible for creating and tracking Antares overlay instances.
-/// This scaffold currently wires directory creation and bookkeeping; the unionfs
-/// integration will be added once the layer stack is finalized.
 pub struct AntaresManager {
     dic: Arc<Dicfuse>,
     paths: AntaresPaths,
     instances: Arc<Mutex<HashMap<String, AntaresConfig>>>,
+    /// Active FUSE handles keyed by job_id. Stored separately from `AntaresConfig`
+    /// because `AntaresFuse` is not serializable.
+    fuse_handles: Arc<Mutex<HashMap<String, fuse::AntaresFuse>>>,
 }
 
 impl AntaresManager {
@@ -146,6 +147,7 @@ impl AntaresManager {
             dic,
             paths,
             instances: Arc::new(Mutex::new(instances)),
+            fuse_handles: Arc::new(Mutex::new(HashMap::new())),
         }
     }
 
@@ -222,6 +224,51 @@ impl AntaresManager {
         }
         std::fs::create_dir_all(&mountpoint)?;
 
+        // Wait for Dicfuse directory tree to be fully loaded before mounting.
+        // Without this, the FUSE mount would start with an empty directory tree
+        // and callers would get "file not found" errors (e.g., buck2 looking for .buckconfig).
+        const DICFUSE_INIT_TIMEOUT_SECS: u64 = 120;
+        tracing::info!(
+            "antares: mount_job_at waiting for Dicfuse ready (timeout: {}s)",
+            DICFUSE_INIT_TIMEOUT_SECS
+        );
+        match tokio::time::timeout(
+            std::time::Duration::from_secs(DICFUSE_INIT_TIMEOUT_SECS),
+            self.dic.store.wait_for_ready(),
+        )
+        .await
+        {
+            Ok(_) => {
+                tracing::info!("antares: mount_job_at Dicfuse ready");
+            }
+            Err(_) => {
+                return Err(std::io::Error::new(
+                    std::io::ErrorKind::TimedOut,
+                    format!(
+                        "Dicfuse initialization timed out after {}s for job {}",
+                        DICFUSE_INIT_TIMEOUT_SECS, job_id
+                    ),
+                ));
+            }
+        }
+
+        // Create AntaresFuse and mount the union filesystem
+        let mut antares_fuse = fuse::AntaresFuse::new(
+            mountpoint.clone(),
+            self.dic.clone(),
+            upper_dir.clone(),
+            cl_dir.clone(),
+        )
+        .await?;
+
+        antares_fuse.mount().await?;
+
+        tracing::info!(
+            "antares: mount_job_at FUSE mounted job_id={} mountpoint={}",
+            job_id,
+            mountpoint.display()
+        );
+
         let instance = AntaresConfig {
             job_id: job_id.to_string(),
             mountpoint,
@@ -236,6 +283,12 @@ impl AntaresManager {
             .await
             .insert(job_id.to_string(), instance.clone());
 
+        // Store the FUSE handle for later unmount
+        self.fuse_handles
+            .lock()
+            .await
+            .insert(job_id.to_string(), antares_fuse);
+
         self.persist_state().await?;
 
         tracing::info!(
@@ -249,9 +302,9 @@ impl AntaresManager {
 
     /// Unmount the FUSE filesystem and remove bookkeeping for a job.
     ///
-    /// Attempts to unmount the filesystem using `fusermount -u`. If the filesystem
-    /// is not mounted (e.g., it was never mounted or already unmounted), the unmount
-    /// attempt will fail but the function will still remove the bookkeeping entry.
+    /// First attempts to unmount using the stored FUSE handle (proper teardown).
+    /// Falls back to `fusermount -u` if no handle is available.
+    /// Bookkeeping is always removed regardless of unmount outcome.
     pub async fn umount_job(&self, job_id: &str) -> std::io::Result<Option<AntaresConfig>> {
         use tracing::{info, warn};
 
@@ -262,36 +315,55 @@ impl AntaresManager {
             None => return Ok(None),
         };
 
-        // Attempt to unmount the FUSE mount
         let mount_path = &config.mountpoint;
         info!("Attempting to unmount FUSE mount at {:?}", mount_path);
 
-        let output = tokio::process::Command::new("fusermount")
-            .arg("-u")
-            .arg(mount_path)
-            .output()
-            .await?;
-
-        if !output.status.success() {
-            let error_msg = String::from_utf8_lossy(&output.stderr);
-            // Check if the error is because the filesystem is not mounted
-            // In this case, we still proceed to remove bookkeeping
-            if error_msg.contains("not mounted") || error_msg.contains("Invalid argument") {
-                warn!(
-                    "Filesystem at {:?} is not mounted, removing bookkeeping only: {}",
-                    mount_path, error_msg
-                );
-            } else {
-                warn!(
-                    "fusermount -u failed with status {} for {:?}: {}",
-                    output.status, mount_path, error_msg
-                );
-                // For other errors, we still remove bookkeeping to avoid stale entries
-                // but log the warning
+        // Try to unmount via the stored FUSE handle first (proper teardown)
+        let mut fuse_handles = self.fuse_handles.lock().await;
+        if let Some(mut fuse) = fuse_handles.remove(job_id) {
+            match fuse.unmount().await {
+                Ok(()) => {
+                    info!("Successfully unmounted {:?} via FUSE handle", mount_path);
+                }
+                Err(e) => {
+                    warn!(
+                        "FUSE handle unmount failed for {:?}: {}, falling back to fusermount",
+                        mount_path, e
+                    );
+                    // Fallback to fusermount -u
+                    let _ = tokio::process::Command::new("fusermount")
+                        .arg("-u")
+                        .arg(mount_path)
+                        .output()
+                        .await;
+                }
             }
         } else {
-            info!("Successfully unmounted {:?}", mount_path);
+            // No FUSE handle available, use fusermount directly
+            let output = tokio::process::Command::new("fusermount")
+                .arg("-u")
+                .arg(mount_path)
+                .output()
+                .await?;
+
+            if !output.status.success() {
+                let error_msg = String::from_utf8_lossy(&output.stderr);
+                if error_msg.contains("not mounted") || error_msg.contains("Invalid argument") {
+                    warn!(
+                        "Filesystem at {:?} is not mounted, removing bookkeeping only: {}",
+                        mount_path, error_msg
+                    );
+                } else {
+                    warn!(
+                        "fusermount -u failed with status {} for {:?}: {}",
+                        output.status, mount_path, error_msg
+                    );
+                }
+            } else {
+                info!("Successfully unmounted {:?} via fusermount", mount_path);
+            }
         }
+        drop(fuse_handles);
 
         // Remove from bookkeeping and persist (even if unmount failed)
         let removed = instances.remove(job_id);

src/daemon/antares.rs

@@ -667,6 +667,9 @@ impl AntaresServiceImpl {
             Some(d) => d,
             None => DicfuseManager::global().await,
         };
+        // Trigger import as early as possible so directory tree loading begins
+        // before any mount requests arrive. Idempotent: no-op if already started.
+        dic.start_import();
         let state_file = PathBuf::from(crate::util::config::antares_state_file());
         Self {
             dicfuse: dic,

src/dicfuse/async_io.rs

@@ -25,7 +25,7 @@ impl Filesystem for Dicfuse {
 
         // Spawn import_arc as a background task to avoid blocking FUSE mount.
         // Guarded so we don't start multiple concurrent imports for the same store
-        // (DicfuseManager may already have started one).
+        // (DicfuseManager or Antares service may already have started one via start_import()).
         if s.try_start_import() {
             tokio::spawn(async move {
                 super::store::import_arc(s).await;
@@ -73,8 +73,15 @@ impl Filesystem for Dicfuse {
     async fn lookup(&self, _req: Request, parent: Inode, name: &OsStr) -> Result<ReplyEntry> {
         // Keep lookup mostly non-blocking: wait a short budget for directory refresh,
         // then continue with best-effort cache lookup and only retry once on miss.
-        const LOOKUP_REFRESH_WAIT_BUDGET_MS: u64 = 20;
-        const LOOKUP_MISS_RETRY_WAIT_BUDGET_MS: u64 = 200;
+        //
+        // Budget rationale: fetch_dir() involves an HTTP round-trip to the monorepo server.
+        // With typical server latency of 50-200ms:
+        //   - 200ms initial wait covers most warm-cache responses and avoids unnecessary
+        //     "miss → retry" round-trips that double the latency for Buck2-style deep traversals.
+        //   - 2000ms retry wait gives enough headroom for cold paths or transient server slowness
+        //     to avoid returning spurious ENOENT that break build tool path resolution.
+        const LOOKUP_REFRESH_WAIT_BUDGET_MS: u64 = 200;
+        const LOOKUP_MISS_RETRY_WAIT_BUDGET_MS: u64 = 2000;
 
         let store = self.store.clone();
 
@@ -361,7 +368,22 @@ impl Filesystem for Dicfuse {
     /// I/O and not store anything in `fh`.  A file system need not implement this method if it
     /// sets [`MountOptions::no_open_dir_support`][crate::MountOptions::no_open_dir_support] and
     /// if the kernel supports `FUSE_NO_OPENDIR_SUPPORT`.
-    async fn opendir(&self, _req: Request, _inode: Inode, _flags: u32) -> Result<ReplyOpen> {
+    async fn opendir(&self, _req: Request, inode: Inode, _flags: u32) -> Result<ReplyOpen> {
+        // Prefetch directory children in the background so that the subsequent
+        // readdir/readdirplus call finds the data already cached.
+        // This is a fire-and-forget optimisation: if it finishes before readdir
+        // arrives the listing is instant; if not, readdir's own ensure_dir_loaded
+        // will wait for the same lock-guarded fetch.
+        let store = self.store.clone();
+        tokio::spawn(async move {
+            if let Err(e) = store.ensure_dir_loaded(inode).await {
+                tracing::debug!(
+                    "dicfuse: opendir prefetch for inode {} failed: {}",
+                    inode,
+                    e
+                );
+            }
+        });
         Ok(ReplyOpen { fh: 0, flags: 0 })
     }
 
@@ -391,6 +413,29 @@ impl Filesystem for Dicfuse {
         }
 
         tracing::debug!("dicfuse: open inode {} (read-only)", inode);
+
+        // Eagerly fetch file content on open so that subsequent read() calls can
+        // be served from cache.  This is the ONLY place (besides import_arc
+        // depth-based prefetch) where Dicfuse triggers content downloads.
+        if self.readable {
+            let item = self.store.get_inode(inode).await?;
+            if !item.is_dir()
+                && !item.hash.is_empty()
+                && item.hash != super::store::EMPTY_BLOB_OID
+                && !self.store.file_exists(inode)
+            {
+                if let Err(e) = self.store.fetch_file_content(inode, &item.hash).await {
+                    tracing::warn!(
+                        "dicfuse: open prefetch failed for inode {} oid {}: {}",
+                        inode,
+                        item.hash,
+                        e
+                    );
+                    // Non-fatal: read() will retry on-demand.
+                }
+            }
+        }
+
         Ok(ReplyOpen { fh: 0, flags: 0 })
     }
     /// read data. Read should send exactly the number of bytes requested except on EOF or error,
@@ -535,12 +580,11 @@ impl Filesystem for Dicfuse {
         Err(std::io::Error::from_raw_os_error(libc::EIO).into())
     }
     async fn access(&self, _req: Request, inode: Inode, _mask: u32) -> Result<()> {
-        // Access is a metadata permission check; keep it lightweight.
-        // For directories, ensure at least one children listing exists (lazy).
-        let item = self.store.get_inode(inode).await?;
-        if item.is_dir() {
-            self.store.ensure_dir_loaded(inode).await?;
-        }
+        // Read-only filesystem: all inodes are accessible. Just verify the inode exists.
+        // Do NOT trigger ensure_dir_loaded here — it causes an unnecessary network
+        // round-trip during the access() → opendir() → readdir() sequence.
+        // Directory children will be loaded lazily in opendir()/readdir().
+        let _item = self.store.get_inode(inode).await?;
         Ok(())
     }