diff --git a/Cargo.lock b/Cargo.lock index 4648745..bb17ea5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2070,9 +2070,9 @@ checksum = "6800badb6cb2082ffd7b6a67e6125bb39f18782f793520caee8cb8846be06112" [[package]] name = "libfuse-fs" -version = "0.1.11" +version = "0.1.12" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ec549f45dd0953d4ccb016b956c9e30e5f463bb24f0e0a5ddc06987ecf402eee" +checksum = "c5aa54193dd60028e43274b0e43fd2b68629af72714589820d1f50b97f0a8f85" dependencies = [ "async-trait", "bitflags 2.11.0", @@ -3568,7 +3568,7 @@ checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" [[package]] name = "scorpiofs" -version = "0.1.0" +version = "0.2.1" dependencies = [ "assert_cmd", "async-recursion", diff --git a/Cargo.toml b/Cargo.toml index 930328e..2014f4f 100644 --- a/Cargo.toml +++ b/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" diff --git a/scorpio.toml b/scorpio.toml index 717fae9..8428492 100644 --- a/scorpio.toml +++ b/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" diff --git a/src/antares/fuse.rs b/src/antares/fuse.rs index 2b24dcc..c1efbdc 100644 --- a/src/antares/fuse.rs +++ b/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(()) } diff --git a/src/antares/mod.rs b/src/antares/mod.rs index cf3d9aa..0058c4f 100644 --- a/src/antares/mod.rs +++ b/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, paths: AntaresPaths, instances: Arc>>, + /// Active FUSE handles keyed by job_id. Stored separately from `AntaresConfig` + /// because `AntaresFuse` is not serializable. + fuse_handles: Arc>>, } 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> { 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); diff --git a/src/daemon/antares.rs b/src/daemon/antares.rs index 2672e52..3369317 100644 --- a/src/daemon/antares.rs +++ b/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, diff --git a/src/dicfuse/async_io.rs b/src/dicfuse/async_io.rs index 7cc379b..8971c39 100644 --- a/src/dicfuse/async_io.rs +++ b/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 { // 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 { + async fn opendir(&self, _req: Request, inode: Inode, _flags: u32) -> Result { + // 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(()) } diff --git a/src/dicfuse/manager.rs b/src/dicfuse/manager.rs index e59ba60..4aa0a3c 100644 --- a/src/dicfuse/manager.rs +++ b/src/dicfuse/manager.rs @@ -105,14 +105,8 @@ impl DicfuseManager { GLOBAL_DICFUSE .get_or_init(|| async { let dicfuse = Arc::new(Dicfuse::new().await); - // Trigger import_arc immediately so directory tree starts loading. - // Guarded so we don't start multiple concurrent imports for the same store. - if dicfuse.store.try_start_import() { - let store_clone = dicfuse.store.clone(); - tokio::spawn(async move { - super::store::import_arc(store_clone).await; - }); - } + // Trigger import immediately so directory tree starts loading. + dicfuse.start_import(); dicfuse }) .await @@ -170,16 +164,10 @@ impl DicfuseManager { Dicfuse::new_with_base_path_and_store_path(&normalized, &store_path).await, ); - // IMPORTANT: Trigger import_arc immediately so the directory tree starts loading. - // This is necessary because `import_arc` is normally called in `Filesystem::init()` - // when FUSE mounts, but callers may need to wait_for_ready() BEFORE mounting + // Trigger import immediately so the directory tree starts loading. + // This is necessary because callers may need to wait_for_ready() BEFORE mounting // (e.g., the Antares daemon needs the root inode to be set up first). - if dicfuse.store.try_start_import() { - let store_clone = dicfuse.store.clone(); - tokio::spawn(async move { - super::store::import_arc(store_clone).await; - }); - } + dicfuse.start_import(); dicfuse }) diff --git a/src/dicfuse/mod.rs b/src/dicfuse/mod.rs index 9af525f..81db201 100644 --- a/src/dicfuse/mod.rs +++ b/src/dicfuse/mod.rs @@ -6,15 +6,11 @@ mod size_store; pub mod store; mod tree_store; -use std::{ - ffi::{OsStr, OsString}, - sync::Arc, - time::Duration, -}; +use std::{ffi::OsStr, sync::Arc, time::Duration}; pub use manager::DicfuseManager; -use crate::{manager::fetch::fetch_tree, util::config}; +use crate::util::config; /// Compute the backing store directory for a given base path. /// @@ -40,12 +36,10 @@ pub(crate) fn compute_store_dir_for_base_path_with_store_root( } use async_trait::async_trait; -use git_internal::internal::object::tree::TreeItemMode; use libfuse_fs::{ context::OperationContext, unionfs::{layer::Layer, Inode}, }; -use reqwest::Client; use rfuse3::{ raw::reply::{ReplyCreated, ReplyEntry}, Result, @@ -206,6 +200,20 @@ impl Dicfuse { } } + /// Start the background import task (directory tree loading + depth-based content prefetch). + /// + /// This should be called as early as possible (e.g., from antares service or DicfuseManager) + /// instead of waiting for the FUSE `init()` callback. The call is idempotent: only the first + /// invocation actually spawns the import task. + pub fn start_import(&self) { + if self.store.try_start_import() { + let s = self.store.clone(); + tokio::spawn(async move { + store::import_arc(s).await; + }); + } + } + pub async fn new_with_store_path(store_path: &str) -> Self { Self { readable: config::dicfuse_readable(), @@ -298,155 +306,6 @@ impl Dicfuse { e.attr.size = self.store.get_persisted_size(item.get_inode()).unwrap_or(0); e } - async fn load_one_file(&self, parent: u64, name: &OsStr) -> std::io::Result<()> { - if !self.readable { - return Ok(()); - } - - let mut parent_item = self.store.find_path(parent).await.unwrap(); - let tree = fetch_tree(&parent_item).await.unwrap(); - - let file_blob_endpoint = config::file_blob_endpoint(); - - let client = Client::new(); - for i in tree.tree_items { - let name_os = OsString::from(&i.name); - if name_os != name { - continue; - } else if i.mode != TreeItemMode::Blob && i.mode != TreeItemMode::BlobExecutable { - return Ok(()); - } - - let url = format!("{}/{}", file_blob_endpoint, i.id); - // Send GET request - let response = client - .get(url) - .send() - .await - .map_err(std::io::Error::other)?; - - // Ensure that the response status is successful - if response.status().is_success() { - // Get the binary data from the response body - let content = response.bytes().await.map_err(std::io::Error::other)?; - - // Store the content in a Vec - let data: Vec = content.to_vec(); - //let child_osstr = OsStr::new(&i.name); - parent_item.push(i.name.clone()); - - let it_temp = self.store.get_by_path(&parent_item.to_string()).await?; - self.store.save_file(it_temp.get_inode(), data); - if i.mode == TreeItemMode::BlobExecutable { - self.store.set_executable(it_temp.get_inode(), true); - } - } else { - eprintln!("Request failed with status: {}", response.status()); - } - break; - } - Ok(()) - } - pub async fn load_files(&self, parent_item: StorageItem, items: &Vec) { - if !self.readable { - return; - } - if self.store.file_exists(parent_item.get_inode()) { - return; - } - let gpath = match self.store.find_path(parent_item.get_inode()).await { - Some(p) => p, - None => { - tracing::warn!( - "load_files: find_path missing for inode {}", - parent_item.get_inode() - ); - return; - } - }; - let tree = match fetch_tree(&gpath).await { - Ok(t) => t, - Err(err) => { - tracing::warn!( - "load_files: fetch_tree failed for path {}: {err}", - gpath.to_string() - ); - return; - } - }; - let mut is_first = true; - let client = Client::new(); - let file_blob_endpoint = config::file_blob_endpoint(); - for i in tree.tree_items { - // Symlinks (TreeItemMode::Link) and subtrees are skipped during - // file preloading. Symlink support requires storing the link - // target in DictionaryStore and returning it via readlink(); this - // is not yet implemented. - if i.mode != TreeItemMode::Blob && i.mode != TreeItemMode::BlobExecutable { - if i.mode == TreeItemMode::Link { - tracing::debug!(name = %i.name, "load_files: skipping symlink (not yet supported)"); - } - continue; - } - let url = format!("{}/{}", file_blob_endpoint, i.id); - // Send GET request - let response = match client.get(url).send().await { - Ok(resp) => resp, - Err(err) => { - tracing::warn!("load_files: request failed for {}: {err}", i.id); - continue; - } - }; - - // Ensure that the response status is successful - if response.status().is_success() { - // Get the binary data from the response body - let content = match response.bytes().await { - Ok(b) => b, - Err(err) => { - tracing::warn!("load_files: read body failed for {}: {err}", i.id); - continue; - } - }; - - // Store the content in a Vec - let data: Vec = content.to_vec(); - - // Get the hit inodes. - let mut hit_inodes: Option = None; - for it in items { - if it.name.eq(&i.name) { - hit_inodes = Some(it.get_inode()); - break; - } - } - let Some(hit_inodes) = hit_inodes else { - tracing::warn!( - "load_files: inode not found for name {} in parent {}", - i.name, - gpath.to_string() - ); - continue; - }; - - // Look up the buff, find Loaded file. - if is_first { - if self.store.file_exists(hit_inodes) { - // if the file is already exists, no need to load again. - break; - } - self.store.save_file(hit_inodes, data); - if i.mode == TreeItemMode::BlobExecutable { - self.store.set_executable(hit_inodes, true); - } - is_first = false; - } - } else { - tracing::warn!(name = %i.name, status = %response.status(), "load_files: HTTP request failed"); - } - } - self.store.save_file(parent_item.get_inode(), Vec::new()); - } } #[cfg(test)] diff --git a/src/dicfuse/store.rs b/src/dicfuse/store.rs index 56afee9..f5724b8 100644 --- a/src/dicfuse/store.rs +++ b/src/dicfuse/store.rs @@ -19,7 +19,7 @@ use once_cell::sync::Lazy; use reqwest::Client; use rfuse3::{raw::reply::ReplyEntry, FileType}; use serde::{Deserialize, Serialize}; -use tokio::sync::{Mutex, Notify, Semaphore}; +use tokio::sync::{Mutex, Notify, RwLock, Semaphore}; use tracing::{debug, info, warn}; use super::{ @@ -770,7 +770,7 @@ pub struct DictionaryStore { /// Per-directory async locks to avoid concurrent loads producing duplicate inodes. dir_locks: Arc>>>, next_inode: AtomicU64, - radix_trie: Arc>>, + radix_trie: Arc>>, persistent_path_store: Arc, // persistent path store for saving and retrieving file paths max_depth: Arc, // max depth for loading directories pub init_notify: Arc, // used in dir_test to notify the start of the test.. @@ -819,7 +819,7 @@ impl DictionaryStore { DictionaryStore { next_inode: AtomicU64::new(1), inodes: Arc::new(Mutex::new(HashMap::new())), - radix_trie: Arc::new(Mutex::new(radix_trie::Trie::new())), + radix_trie: Arc::new(RwLock::new(radix_trie::Trie::new())), persistent_path_store: Arc::new(tree_store), dirs: Arc::new(DashMap::new()), dir_locks: Arc::new(DashMap::new()), @@ -852,7 +852,7 @@ impl DictionaryStore { DictionaryStore { next_inode: AtomicU64::new(1), inodes: Arc::new(Mutex::new(HashMap::new())), - radix_trie: Arc::new(Mutex::new(radix_trie::Trie::new())), + radix_trie: Arc::new(RwLock::new(radix_trie::Trie::new())), persistent_path_store: Arc::new(tree_store), dirs: Arc::new(DashMap::new()), dir_locks: Arc::new(DashMap::new()), @@ -916,7 +916,7 @@ impl DictionaryStore { DictionaryStore { next_inode: AtomicU64::new(1), inodes: Arc::new(Mutex::new(HashMap::new())), - radix_trie: Arc::new(Mutex::new(radix_trie::Trie::new())), + radix_trie: Arc::new(RwLock::new(radix_trie::Trie::new())), persistent_path_store: Arc::new(tree_store), dirs: Arc::new(DashMap::new()), dir_locks: Arc::new(DashMap::new()), @@ -1124,7 +1124,7 @@ impl DictionaryStore { // Use radix trie as the authoritative mapping from path -> inode. let key = GPath::from(item.item.path.clone()).to_string(); let existing = { - let trie = self.radix_trie.lock().await; + let trie = self.radix_trie.read().await; trie.get(&key).copied() }; @@ -1256,7 +1256,7 @@ impl DictionaryStore { if let Ok(pinode) = prw.get_item(parent) { // insert info to a radix_trie for path match. self.radix_trie - .lock() + .write() .await .insert(GPath::from(item.item.path.clone()).to_string(), alloc_inode); prw.insert_item(alloc_inode, parent, item); @@ -1491,7 +1491,7 @@ impl DictionaryStore { .unwrap() .file_list .insert(child_path.to_string_lossy().to_string(), false); - self.radix_trie.lock().await.insert( + self.radix_trie.write().await.insert( GPath::from(child_path.to_string_lossy().to_string()).to_string(), child, ); @@ -1581,7 +1581,7 @@ impl DictionaryStore { let inode = if normalized.is_empty() { 1 } else { - let binding = self.radix_trie.lock().await; + let binding = self.radix_trie.read().await; *binding .get(&normalized) .ok_or(io::Error::new(io::ErrorKind::NotFound, "path not found"))? @@ -1594,7 +1594,7 @@ impl DictionaryStore { let inode = if path.is_empty() || path == "/" { 1 } else { - let binding = self.radix_trie.lock().await; + let binding = self.radix_trie.read().await; *binding .get(&GPath::from(path.to_owned()).to_string()) .ok_or(io::Error::new(io::ErrorKind::NotFound, "path not found"))? @@ -1614,7 +1614,7 @@ impl DictionaryStore { return Ok(PathLookupStatus::Found(1)); } - if let Some(inode) = self.radix_trie.lock().await.get(&normalized).copied() { + if let Some(inode) = self.radix_trie.read().await.get(&normalized).copied() { return Ok(PathLookupStatus::Found(inode)); } @@ -1629,7 +1629,7 @@ impl DictionaryStore { return Ok(PathLookupStatus::Found(1)); } - let trie = self.radix_trie.lock().await; + let trie = self.radix_trie.read().await; let mut ancestor_key: Option = None; let mut ancestor_inode: Option = None; for i in (1..parts.len()).rev() { @@ -1982,7 +1982,7 @@ async fn reset_store_for_import(store: &DictionaryStore) { store.dirs.clear(); store.dir_locks.clear(); store.inodes.lock().await.clear(); - *store.radix_trie.lock().await = radix_trie::Trie::new(); + *store.radix_trie.write().await = radix_trie::Trie::new(); store.next_inode.store(1, Ordering::Relaxed); store.ready.store(false, Ordering::Release); @@ -2106,9 +2106,22 @@ pub async fn load_dir_depth(store: Arc, parent_path: String, ma }, ); } else { - // NOTE: Do NOT prefetch file contents during directory tree loading. - // Dicfuse should fetch file contents on-demand on read() to keep initial load fast, - // especially for large monorepos. + // Prefetch file content within depth limit during import. + // Files discovered at the root level (depth 0) are always within max_depth. + if !it.hash.is_empty() && it.hash != EMPTY_BLOB_OID && !store.file_exists(it_inode) + { + match fetch_file(&it.hash).await { + Ok(content) => { + store.save_file(it_inode, content); + } + Err(e) => { + debug!( + "[load_dir_depth] prefetch file failed (path={path:?} oid={}): {e}", + it.hash + ); + } + } + } } } } @@ -2230,7 +2243,23 @@ pub async fn load_dir_depth(store: Arc, parent_path: String, ma }, ); } else { - // NOTE: Do NOT prefetch file contents during directory tree loading. + // Prefetch file content within depth limit during import. + if !newit.hash.is_empty() + && newit.hash != EMPTY_BLOB_OID + && !store.file_exists(new_inode) + { + match fetch_file(&newit.hash).await { + Ok(content) => { + store.save_file(new_inode, content); + } + Err(e) => { + debug!( + "[load_dir_depth] prefetch file failed (path={tmp_path:?} oid={}): {e}", + newit.hash + ); + } + } + } } } } @@ -2340,18 +2369,17 @@ pub async fn import_arc(store: Arc) { store.inodes.lock().await.insert(1, root_item.into()); ensure_dir_tracked(&store.dirs, &user_root); - // Mark ready as soon as the root inode exists so Antares can mount immediately. - // Directory entries will be populated lazily on lookup/readdir, while import continues. - store.mark_ready(); - // Limit concurrent warmups/imports across stores to avoid remote pressure spikes. let _permit = global_import_semaphore() .acquire_owned() .await .expect("global import semaphore closed"); - // Always do a shallow root listing once (best-effort). This warms the root and persists children, - // while still keeping mount time-to-usable low (root is already ready). + // Always do a shallow root listing once (best-effort). This warms the root and persists children. + // We intentionally do NOT mark_ready() until this succeeds, so that callers (e.g., Antares + // mount_job_at → wait_for_ready()) only proceed once root children are actually queryable. + // Without this, Buck2-like workloads that immediately traverse deep paths would hit ENOENT + // on the very first lookup because the root directory listing hasn't been fetched yet. let seeded_ok = match store.ensure_dir_loaded(1).await { Ok(()) => true, Err(e) => { @@ -2362,6 +2390,10 @@ pub async fn import_arc(store: Arc) { } }; + // Mark ready after root children are loaded. Callers waiting on wait_for_ready() (e.g., + // Antares mount) can now safely resolve root-level lookups without blocking on network IO. + store.mark_ready(); + // Optional deep prewarm: disabled by default for Antares subdir mounts (max_depth=0). if store.max_depth() > 0 { let max_depth = store.max_depth() + 2; @@ -2758,7 +2790,7 @@ impl DictionaryStore { // Keep trie in sync so `get_by_path` works in tests. self.radix_trie - .lock() + .write() .await .insert(GPath::from(full_path).to_string(), inode); } @@ -2954,7 +2986,7 @@ mod tests { DictionaryStore { next_inode: AtomicU64::new(1), inodes: Arc::new(Mutex::new(HashMap::new())), - radix_trie: Arc::new(Mutex::new(radix_trie::Trie::new())), + radix_trie: Arc::new(RwLock::new(radix_trie::Trie::new())), persistent_path_store: Arc::new(tree_store), dirs: Arc::new(DashMap::new()), dir_locks: Arc::new(DashMap::new()), diff --git a/tests/antares_test.rs b/tests/antares_test.rs index d478267..93e9565 100644 --- a/tests/antares_test.rs +++ b/tests/antares_test.rs @@ -419,3 +419,73 @@ async fn test_fuse_multiple_custom_mounts() { 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 +/// while the main task waits for Ctrl-C. Single-thread (`current_thread`) would +/// serialize all FUSE handlers and stall `ls` when Dicfuse does network IO. +/// +/// Run with: +/// ```bash +/// sudo -E cargo test --test antares_test test_mount_job_no_cl_keep_running -- --ignored --nocapture +/// ``` +/// +/// After Ctrl-C the test will attempt `fusermount -u`; if it fails, clean up manually: +/// ```bash +/// fusermount -uz /tmp/.tmp*/mnt/test-job-no-cl +/// ``` +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[ignore = "requires FUSE privileges; runs until Ctrl-C"] +async fn test_mount_job_no_cl_keep_running() { + // Enable tracing so FUSE / Dicfuse logs are visible with --nocapture. + let _ = tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")), + ) + .try_init(); + + init_config(); + + let root = tempdir().unwrap(); + let paths = AntaresPaths::new( + root.path().join("upper"), + root.path().join("cl"), + root.path().join("mnt"), + root.path().join("state.toml"), + ); + let manager = AntaresManager::new(paths).await; + + println!("Mounting (waiting for Dicfuse ready, may take a while if server is slow)..."); + let config = manager.mount_job("test-job-no-cl", None).await.unwrap(); + println!("✓ Mounted at: {}", config.mountpoint.display()); + println!(" job_id : {}", config.job_id); + println!(" upper_dir: {}", config.upper_dir.display()); + assert_eq!(config.job_id, "test-job-no-cl"); + assert!(config.cl_dir.is_none()); + assert!(config.cl_id.is_none()); + assert!(config.mountpoint.exists()); + + println!( + "FUSE is running. Try `ls {}` in another terminal.", + config.mountpoint.display() + ); + println!("Press Ctrl-C to stop."); + tokio::signal::ctrl_c() + .await + .expect("failed to listen for Ctrl-C"); + println!("\nCtrl-C received, unmounting..."); + + // Attempt clean unmount so the mountpoint doesn't become a zombie. + if let Err(e) = manager.umount_job("test-job-no-cl").await { + eprintln!("Warning: umount_job failed: {e}"); + // Fallback: lazy unmount so the test doesn't leave a zombie mount. + let mp = config.mountpoint.to_string_lossy().to_string(); + let _ = tokio::process::Command::new("fusermount") + .args(["-uz", &mp]) + .output() + .await; + } + println!("Done."); +}