diff --git a/nativelink-store/src/filesystem_store.rs b/nativelink-store/src/filesystem_store.rs index 8654254ec..6adb0b565 100644 --- a/nativelink-store/src/filesystem_store.rs +++ b/nativelink-store/src/filesystem_store.rs @@ -20,6 +20,7 @@ use std::borrow::Cow; #[cfg(unix)] use std::collections::HashMap; use std::ffi::{OsStr, OsString}; +use std::path::Path; use std::sync::{Arc, Weak}; use std::time::SystemTime; @@ -1407,9 +1408,11 @@ impl FilesystemStore { "filesystem_store_executable_variant", move || -> Result<(), Error> { use std::os::unix::fs::PermissionsExt; - std::fs::copy(&src_path, &temp_owned).map_err(|e| { - make_err!(Code::Internal, "executable-variant copy failed: {e:?}") - })?; + // Keep the io error's code: an ENOENT here means the CAS blob + // this variant copies from has diverged, and callers heal on + // `NotFound`. + std::fs::copy(&src_path, &temp_owned) + .map_err(|e| Error::from(e).append("executable-variant copy failed"))?; std::fs::set_permissions(&temp_owned, std::fs::Permissions::from_mode(0o555)) .map_err(|e| { make_err!( @@ -1477,6 +1480,136 @@ impl FilesystemStore { .ok_or_else(|| make_err!(Code::NotFound, "{digest} not found in filesystem store. This may indicate the file was evicted due to cache pressure. Consider increasing 'max_bytes' in your filesystem store's eviction_policy configuration.")) } + /// Hardlinks the blob for `digest` to `dest`, which must not already exist. + /// + /// `is_executable` links the 0o555 executable variant (see + /// [`Self::get_executable_hardlink_source`]), otherwise the shared 0o444 CAS + /// blob, which is linked under the entry's path lock so eviction cannot + /// rename it away mid-link. + /// + /// Fails with `NotFound` when the blob is absent from this store, and if it + /// is indexed but missing on disk the entry is dropped before returning — + /// the next `populate_fast_store` then re-fetches it instead of + /// short-circuiting on `has()`, which only consults the index. + pub async fn hardlink_to( + &self, + digest: &DigestInfo, + is_executable: bool, + dest: impl AsRef, + ) -> Result<(), Error> { + let dest = dest.as_ref(); + if is_executable { + let src_path = match self.get_executable_hardlink_source(digest).await { + Ok(src_path) => src_path, + Err(err) => { + // The variant is built by copying the CAS blob, so a + // divergence surfaces here rather than at the hardlink. + return Err(self + .heal_and_describe_failed_hardlink(digest, None, dest, err) + .await) + .err_tip(|| "Resolving executable hardlink source"); + } + }; + return match fs::hard_link(&src_path, dest).await { + Ok(()) => Ok(()), + Err(err) => Err(self + .heal_and_describe_failed_hardlink(digest, Some(&src_path), dest, err) + .await), + }; + } + + let entry = self + .get_file_entry_for_digest(digest) + .await + .err_tip(|| "During hard link")?; + // Link under the entry's path lock: `unref()` takes it for writing to + // rename the blob away, so the source cannot move mid-link. + // TODO: add a test for #2051: deadlock with large number of files + let dest_owned = dest.to_owned(); + let (src_path, link_result) = entry + .get_file_path_locked(|src_path| async move { + let link_result = fs::hard_link(&src_path, &dest_owned).await; + Ok((src_path, link_result)) + }) + .await?; + match link_result { + Ok(()) => Ok(()), + // Healing re-takes the path lock, so it must happen out here. + Err(err) => Err(self + .heal_and_describe_failed_hardlink(digest, Some(&src_path), dest, err) + .await), + } + } + + /// Logs a failed hardlink, drops `digest`'s entry if its file is missing, + /// and returns `err` annotated for the caller. + async fn heal_and_describe_failed_hardlink( + &self, + digest: &DigestInfo, + src_path: Option<&OsStr>, + dest: &Path, + err: Error, + ) -> Error { + let src_metadata = src_path.map(std::fs::metadata); + let dest_metadata = std::fs::metadata(dest); + let dest_parent_metadata = dest.parent().map(Path::metadata); + let snapshot = self.get_eviction_snapshot(); + warn!(?err, fs_eviction_snapshot = %snapshot, ?src_path, ?src_metadata, ?dest, ?dest_metadata, ?dest_parent_metadata, "Could not make hardlink"); + let src_display = src_path.map_or_else( + || format!("{digest}"), + |src_path| src_path.display().to_string(), + ); + if err.code != Code::NotFound { + return err.append(format!( + "Could not make hardlink from {src_display} to {}", + dest.display() + )); + } + self.remove_entry_if_file_missing(digest).await; + err.append(format!( + "Could not make hardlink from {src_display} to {}, file was likely evicted from cache.\n\ + This error often occurs when the filesystem store's max_bytes is too small for your workload.\n\ + To fix this issue:\n\ + 1. Increase the 'max_bytes' value in your filesystem store configuration\n\ + 2. Example: Change 'max_bytes: 10000000000' to 'max_bytes: 50000000000' (or higher)\n\ + 3. The setting is typically found in your nativelink.json config under:\n\ + stores -> [your_filesystem_store] -> filesystem -> eviction_policy -> max_bytes\n\ + 4. Restart NativeLink after making the change\n\n\ + If this error persists after increasing max_bytes several times, please report at:\n\ + https://github.com/TraceMachina/nativelink/issues\n\ + Include your config file and both server and client logs to help us assist you.", + dest.display() + )) + } + + /// Removes `digest`'s entry if its file is missing from disk, and returns + /// whether it removed one. The check runs under the entry's path lock; an + /// entry whose file is present, or unreadable for some other reason, is + /// left alone. + async fn remove_entry_if_file_missing(&self, digest: &DigestInfo) -> bool { + let key: StoreKey<'static> = (*digest).into(); + let Some(entry) = self.evicting_map.get(&key).await else { + return false; + }; + let stat_result = entry + .get_file_path_locked(|path| async move { fs::metadata(&path).await }) + .await; + let Err(err) = stat_result else { + return false; + }; + // An unreadable-but-present file (e.g. EACCES) is not a divergence; + // leave the entry for a human to notice. + if err.code != Code::NotFound { + return false; + } + warn!( + ?key, + "Filesystem store map/disk divergence: removing entry; next populate will repair it from the slow store", + ); + self.evicting_map.remove(&key).await; + true + } + async fn update_file( self: Pin<&Self>, mut entry: Fe, diff --git a/nativelink-store/tests/fast_slow_store_test.rs b/nativelink-store/tests/fast_slow_store_test.rs index b6ee7ee5a..eaac810c5 100644 --- a/nativelink-store/tests/fast_slow_store_test.rs +++ b/nativelink-store/tests/fast_slow_store_test.rs @@ -20,17 +20,20 @@ use std::sync::{Arc, Mutex}; use async_trait::async_trait; use bytes::Bytes; use futures::future::join_all; -use nativelink_config::stores::{FastSlowSpec, MemorySpec, NoopSpec, StoreDirection, StoreSpec}; +use nativelink_config::stores::{ + FastSlowSpec, FilesystemSpec, MemorySpec, NoopSpec, StoreDirection, StoreSpec, +}; use nativelink_error::{Code, Error, ResultExt, make_err}; use nativelink_macro::nativelink_test; use nativelink_metric::MetricsComponent; use nativelink_store::fast_slow_store::FastSlowStore; +use nativelink_store::filesystem_store::{FileEntry, FileEntryImpl, FilesystemStore}; use nativelink_store::memory_store::MemoryStore; use nativelink_store::noop_store::NoopStore; use nativelink_util::buf_channel::{ DropCloserReadHalf, DropCloserWriteHalf, make_buf_channel_pair, }; -use nativelink_util::common::DigestInfo; +use nativelink_util::common::{DigestInfo, fs, make_temp_path}; use nativelink_util::health_utils::{HealthStatusIndicator, default_health_status_indicator}; use nativelink_util::store_trait::{ RemoveCallback, Store, StoreDriver, StoreKey, StoreLike, UploadSizeInfo, @@ -2000,3 +2003,88 @@ async fn get_part_propagates_not_found_after_partial_fast_read() -> Result<(), E Ok(()) } + +/// `populate_fast_store` short-circuits on `fast_store.has()`, an in-memory index, so +/// after `hardlink_to` drops a diverged entry the next populate must repair it. +#[nativelink_test] +async fn healed_divergence_is_repaired_by_next_populate() -> Result<(), Error> { + const VALUE: &str = "0123456789"; + + let fast_spec = FilesystemSpec { + content_path: make_temp_path("content_path"), + temp_path: make_temp_path("temp_path"), + eviction_policy: None, + ..Default::default() + }; + let slow_spec = MemorySpec::default(); + let fast_store = FilesystemStore::::new(&fast_spec).await?; + let slow_store = MemoryStore::new(&slow_spec); + let fast_slow = FastSlowStore::new( + &FastSlowSpec { + fast: StoreSpec::Filesystem(fast_spec), + slow: StoreSpec::Memory(slow_spec), + fast_direction: StoreDirection::default(), + slow_direction: StoreDirection::default(), + bypass_dedup_threshold_bytes: 0, + }, + Store::new(fast_store.clone()), + Store::new(slow_store.clone()), + ); + + let digest = DigestInfo::try_new(VALID_HASH, VALUE.len())?; + slow_store + .as_ref() + .update_oneshot(digest, VALUE.into()) + .await?; + + // Diverge the fast tier: delete the content file, leave the map entry. + fast_slow.populate_fast_store(digest.into()).await?; + let content_file = fast_store + .get_file_entry_for_digest(&digest) + .await? + .get_file_path_locked(|path| async move { Ok(path) }) + .await?; + fs::remove_file(&content_file).await?; + + assert!( + fast_store.as_ref().has(digest).await?.is_some(), + "precondition: the index still advertises the blob it no longer has", + ); + + let dest_dir = make_temp_path("hardlink_dest"); + fs::create_dir_all(&dest_dir).await?; + let dest = format!("{dest_dir}/staged"); + let err = fast_store + .hardlink_to(&digest, false, &dest) + .await + .expect_err("hardlinking a missing blob must fail"); + assert_eq!(err.code, Code::NotFound, "got: {err:?}"); + assert!( + fast_store.as_ref().has(digest).await?.is_none(), + "the failed hardlink must have dropped the diverged entry", + ); + + // With the stale entry gone, populate must miss and repair. + fast_slow + .populate_fast_store(digest.into()) + .await + .err_tip(|| "Populate after healing must repair the fast tier")?; + fast_store + .hardlink_to(&digest, false, &dest) + .await + .err_tip(|| "Hardlink after the repairing populate")?; + assert_eq!(fs::read(&dest).await?, VALUE.as_bytes()); + + // A hardlink can also fail with ENOENT for reasons that are nothing to do with the + // source (here: no such destination directory), so a healthy entry must survive. + let err = fast_store + .hardlink_to(&digest, false, "no_such_dir/staged") + .await + .expect_err("hardlinking into a missing directory must fail"); + assert_eq!(err.code, Code::NotFound, "got: {err:?}"); + assert!( + fast_store.as_ref().has(digest).await?.is_some(), + "a healthy entry must survive a hardlink failure", + ); + Ok(()) +} diff --git a/nativelink-worker/src/running_actions_manager.rs b/nativelink-worker/src/running_actions_manager.rs index 9e8e77208..cb5aea4e5 100644 --- a/nativelink-worker/src/running_actions_manager.rs +++ b/nativelink-worker/src/running_actions_manager.rs @@ -367,48 +367,9 @@ pub fn download_to_directory<'a>( // per-digest 0o555 variant created once off the hot // path (the 0o444 CAS blob is shared and cannot carry // +x); non-executables hardlink the 0o444 CAS blob. - let src_path = if is_executable { - filesystem_store - .get_executable_hardlink_source(&digest) - .await - .err_tip(|| "Resolving executable hardlink source")? - } else { - let file_entry = filesystem_store - .get_file_entry_for_digest(&digest) - .await - .err_tip(|| "During hard link")?; - // TODO: add a test for #2051: deadlock with large number of files - file_entry - .get_file_path_locked(|src| async move { Ok(src) }) - .await? - }; - fs::hard_link(&src_path, &dest) - .await - .map_err(|e| { - let src_metadata = std::fs::metadata(&src_path); - let dest_metadata = std::fs::metadata(&dest); - let dest_parent_metadata = Path::new(&dest).parent().map(Path::metadata); - let snapshot = filesystem_store.get_eviction_snapshot(); - warn!(?e, fs_eviction_snapshot = %snapshot, ?src_path, ?src_metadata, %dest, ?dest_metadata, ?dest_parent_metadata, "Could not make hardlink"); - if e.code == Code::NotFound { - e.append( - format!( - "Could not make hardlink from {} to {dest}, file was likely evicted from cache.\n\ - This error often occurs when the filesystem store's max_bytes is too small for your workload.\n\ - To fix this issue:\n\ - 1. Increase the 'max_bytes' value in your filesystem store configuration\n\ - 2. Example: Change 'max_bytes: 10000000000' to 'max_bytes: 50000000000' (or higher)\n\ - 3. The setting is typically found in your nativelink.json config under:\n\ - stores -> [your_filesystem_store] -> filesystem -> eviction_policy -> max_bytes\n\ - 4. Restart NativeLink after making the change\n\n\ - If this error persists after increasing max_bytes several times, please report at:\n\ - https://github.com/TraceMachina/nativelink/issues\n\ - Include your config file and both server and client logs to help us assist you.", src_path.display() - )) - } else { - e.append(format!("Could not make hardlink from {} to {dest}", src_path.display())) - } - })?; + filesystem_store + .hardlink_to(&digest, is_executable, &dest) + .await?; // Hardlinked inodes are already correct (the 0o444 // blob or the 0o555 executable variant) and carry no // per-file metadata, so there is nothing to stamp. diff --git a/nativelink-worker/tests/running_actions_manager_test.rs b/nativelink-worker/tests/running_actions_manager_test.rs index e2007c150..87581f935 100644 --- a/nativelink-worker/tests/running_actions_manager_test.rs +++ b/nativelink-worker/tests/running_actions_manager_test.rs @@ -56,7 +56,7 @@ mod tests { use nativelink_store::ac_utils::compute_buf_digest; use nativelink_store::ac_utils::{get_and_decode_digest, serialize_and_upload_message}; use nativelink_store::fast_slow_store::FastSlowStore; - use nativelink_store::filesystem_store::FilesystemStore; + use nativelink_store::filesystem_store::{FileEntry, FilesystemStore}; use nativelink_store::memory_store::MemoryStore; #[cfg(target_family = "unix")] use nativelink_util::action_messages::DirectoryInfo; @@ -258,6 +258,108 @@ mod tests { Ok(()) } + /// When the fast tier's eviction map and content directory disagree about the presence of a + /// blob, the worker should notice when it fails to hardlink and trigger some kind of + /// recovery so that later attempts succeed. + /// Test by deleting a content file, simulating some kind of race or failed emplace etc + /// in the underlying store. + #[nativelink_test] + async fn download_to_directory_recovers_from_fast_store_divergence() + -> Result<(), Box> { + const FILE1_NAME: &str = "file1.txt"; + const FILE1_CONTENT: &str = "HELLOFILE1"; + + let (fast_store, slow_store, cas_store, _ac_store) = setup_stores().await?; + + let file1_content_digest = DigestInfo::new([2u8; 32], FILE1_CONTENT.len() as u64); + slow_store + .as_ref() + .update_oneshot(file1_content_digest, FILE1_CONTENT.into()) + .await?; + + let root_directory_digest = DigestInfo::new([1u8; 32], 32); + let root_directory = Directory { + files: vec![FileNode { + name: FILE1_NAME.to_string(), + digest: Some(file1_content_digest.into()), + is_executable: false, + node_properties: None, + }], + ..Default::default() + }; + slow_store + .as_ref() + .update_oneshot(root_directory_digest, root_directory.encode_to_vec().into()) + .await?; + + let download_once = |dir_name: &'static str| { + let cas_store = cas_store.clone(); + let fast_store = fast_store.clone(); + async move { + let download_dir = make_temp_path(dir_name); + fs::create_dir_all(&download_dir) + .await + .err_tip(|| format!("Could not make download_dir : {download_dir}"))?; + download_to_directory( + cas_store.as_ref(), + fast_store.as_pin(), + &root_directory_digest, + &download_dir, + ) + .await?; + Result::::Ok(download_dir) + } + }; + + // Attempt 1 populates the fast tier from the slow tier and succeeds. + let first_dir = download_once("divergence_attempt1").await?; + assert_eq!( + from_utf8(&fs::read(format!("{first_dir}/{FILE1_NAME}")).await?)?, + FILE1_CONTENT, + ); + + // Diverge the fast tier: delete the content file, leave the map entry. + let content_file = fast_store + .get_file_entry_for_digest(&file1_content_digest) + .await? + .get_file_path_locked(|path| async move { Ok(path) }) + .await?; + fs::remove_file(&content_file).await?; + + // Only presence is asserted; the filesystem store reports block-rounded sizes. + assert!( + fast_store + .as_ref() + .has(file1_content_digest) + .await? + .is_some(), + "precondition: the fast tier still advertises the blob it no longer has", + ); + + // The scheduler's retries: a fix may repair in place (2) or heal the stale + // entry and recover on the next attempt (3). Either must stage the blob. + let second = download_once("divergence_attempt2").await; + let third = download_once("divergence_attempt3").await; + + if let (Err(second_err), Err(third_err)) = (&second, &third) { + panic!( + "every retry failed even though the slow store still holds the blob; \ + the action can never recover.\nattempt 2: {second_err:?}\nattempt 3: {third_err:?}" + ); + } + + let ((Ok(recovered_dir), _) | (Err(_), Ok(recovered_dir))) = (second, third) else { + unreachable!("handled above") + }; + assert_eq!( + from_utf8(&fs::read(format!("{recovered_dir}/{FILE1_NAME}")).await?)?, + FILE1_CONTENT, + "recovered file must have the full contents from the slow tier", + ); + + Ok(()) + } + #[nativelink_test] async fn download_to_directory_folder_download_test() -> Result<(), Box> {