diff --git a/nativelink-store/src/filesystem_store.rs b/nativelink-store/src/filesystem_store.rs index 91afe82c0..d65ad4749 100644 --- a/nativelink-store/src/filesystem_store.rs +++ b/nativelink-store/src/filesystem_store.rs @@ -12,7 +12,8 @@ // See the License for the specific language governing permissions and // limitations under the License. -use core::fmt::{Debug, Formatter}; +use core::cmp; +use core::fmt::{Debug, Display, Formatter}; use core::pin::Pin; use core::sync::atomic::{AtomicU64, Ordering}; use core::time::Duration; @@ -104,6 +105,25 @@ enum PathType { Custom(OsString), } +#[derive(Debug, Clone, Copy)] +pub struct Generation(u64); + +impl Generation { + pub const fn new(generation: u64) -> Self { + Self(generation) + } + + pub const fn inner(&self) -> u64 { + self.0 + } +} + +impl Display for Generation { + fn fmt(&self, f: &mut Formatter<'_>) -> core::fmt::Result { + write!(f, "{}", self.0) + } +} + /// [`EncodedFilePath`] stores the path to the file /// including the context, path type and key to the file. /// The whole [`StoreKey`] is stored as opposed to solely @@ -114,12 +134,18 @@ pub struct EncodedFilePath { shared_context: Arc, path_type: PathType, key: StoreKey<'static>, + generation: Generation, } impl EncodedFilePath { #[inline] fn get_file_path(&self) -> Cow<'_, OsStr> { - get_file_path_raw(&self.path_type, self.shared_context.as_ref(), &self.key) + get_file_path_raw( + &self.path_type, + self.shared_context.as_ref(), + &self.key, + self.generation, + ) } } @@ -128,13 +154,14 @@ fn get_file_path_raw<'a>( path_type: &'a PathType, shared_context: &SharedContext, key: &StoreKey<'a>, + generation: Generation, ) -> Cow<'a, OsStr> { let folder = match path_type { PathType::Content => &shared_context.content_path, PathType::Temp => &shared_context.temp_path, PathType::Custom(path) => return Cow::Borrowed(path), }; - Cow::Owned(to_full_path_from_key(folder, key)) + Cow::Owned(to_full_path_from_key(folder, key, generation)) } impl Drop for EncodedFilePath { @@ -193,10 +220,12 @@ impl Drop for EncodedFilePath { /// Previously, only the string representation of the [`DigestInfo`] was /// used with no prefix #[inline] -fn to_full_path_from_key(folder: &str, key: &StoreKey<'_>) -> OsString { +fn to_full_path_from_key(folder: &str, key: &StoreKey<'_>, generation: Generation) -> OsString { match key { - StoreKey::Str(str) => format!("{folder}/{STR_FOLDER}/{str}"), - StoreKey::Digest(digest_info) => format!("{folder}/{DIGEST_FOLDER}/{digest_info}"), + StoreKey::Str(str) => format!("{folder}/{STR_FOLDER}/{str}-{generation}"), + StoreKey::Digest(digest_info) => { + format!("{folder}/{DIGEST_FOLDER}/{digest_info}-{generation}") + } } .into() } @@ -677,7 +706,11 @@ impl LenEntry for FileEntryImpl { let from_path = encoded_file_path.get_file_path(); let new_key = make_temp_key(&encoded_file_path.key); - let to_path = to_full_path_from_key(&encoded_file_path.shared_context.temp_path, &new_key); + let to_path = to_full_path_from_key( + &encoded_file_path.shared_context.temp_path, + &new_key, + encoded_file_path.generation, + ); if let Err(err) = fs::rename(&from_path, &to_path).await { // ENOENT from rename is ambiguous: the source may be gone, or @@ -736,11 +769,19 @@ fn digest_from_filename(file_name: &str) -> Result { DigestInfo::try_new(hash, size) } -pub fn key_from_file(file_name: &str, file_type: FileType) -> Result, Error> { - match file_type { - FileType::String => Ok(StoreKey::new_str(file_name)), - FileType::Digest => digest_from_filename(file_name).map(StoreKey::Digest), - } +pub fn key_and_generation_from_file( + file_name: &str, + file_type: FileType, +) -> Result<(StoreKey<'_>, Generation), Error> { + let (key, generation) = file_name.rsplit_once('-').err_tip(|| "")?; + let generation = Generation::new(generation.parse::()?); + + let key = match file_type { + FileType::String => StoreKey::new_str(key), + FileType::Digest => digest_from_filename(key).map(StoreKey::Digest)?, + }; + + Ok((key, generation)) } /// The number of files to read the metadata for at the same time when running @@ -756,7 +797,7 @@ async fn add_files_to_cache( shared_context: &Arc, block_size: u64, rename_fn: fn(&OsStr, &OsStr) -> Result<(), std::io::Error>, -) -> Result<(), Error> { +) -> Result { #[expect(clippy::too_many_arguments)] async fn process_entry( evicting_map: &FsEvictingMap<'_, Fe>, @@ -767,8 +808,8 @@ async fn add_files_to_cache( block_size: u64, anchor_time: &SystemTime, shared_context: &Arc, - ) -> Result<(), Error> { - let key = key_from_file(file_name, file_type)?; + ) -> Result { + let (key, generation) = key_and_generation_from_file(file_name, file_type)?; let file_entry = Fe::create( data_size, @@ -777,6 +818,7 @@ async fn add_files_to_cache( shared_context: shared_context.clone(), path_type: PathType::Content, key: key.borrow().into_owned(), + generation, }), ); let time_since_anchor = if let Ok(d) = anchor_time.duration_since(atime) { @@ -797,7 +839,7 @@ async fn add_files_to_cache( i32::try_from(time_since_anchor.as_secs()).unwrap_or(i32::MAX), ) .await; - Ok(()) + Ok(generation) } async fn read_files( @@ -877,13 +919,38 @@ async fn add_files_to_cache( Ok(()) } + async fn move_old_cache_2( + shared_context: &Arc, + rename_fn: fn(&OsStr, &OsStr) -> Result<(), std::io::Error>, + ) -> Result<(), Error> { + let file_infos = read_files(Some(DIGEST_FOLDER), shared_context).await?; + let folder_path = format!("{}/{DIGEST_FOLDER}", shared_context.content_path); + + for (file_name, _, _, _) in file_infos.into_iter().filter(|x| x.3) { + if file_name.matches('-').count() != 1 { + continue; + } + + let from_file: OsString = format!("{folder_path}/{file_name}").into(); + let to_file: OsString = format!("{folder_path}/{file_name}-0").into(); + + if let Err(err) = rename_fn(&from_file, &to_file) { + warn!(?from_file, ?to_file, ?err, "Failed to rename file",); + } else { + debug!(?from_file, ?to_file, "Renamed file (old cache)",); + } + } + + Ok(()) + } + async fn add_files_to_cache( evicting_map: &FsEvictingMap<'_, Fe>, anchor_time: &SystemTime, shared_context: &Arc, block_size: u64, folder: &str, - ) -> Result<(), Error> { + ) -> Result { let file_infos = read_files(Some(folder), shared_context).await?; let file_type = match folder { STR_FOLDER => FileType::String, @@ -893,6 +960,8 @@ async fn add_files_to_cache( let path_root = format!("{}/{folder}", shared_context.content_path); + let mut max_generation = 0; + for (file_name, atime, data_size, _) in file_infos.into_iter().filter(|x| x.3) { let result = process_entry( evicting_map, @@ -905,18 +974,26 @@ async fn add_files_to_cache( shared_context, ) .await; - if let Err(err) = result { - warn!(?file_name, ?err, "Failed to add file to eviction cache",); - // Ignore result. - drop(fs::remove_file(format!("{path_root}/{file_name}")).await); + + match result { + Ok(generation) => { + max_generation = cmp::max(max_generation, generation.inner()); + } + Err(err) => { + warn!(?file_name, ?err, "Failed to add file to eviction cache",); + // Ignore result. + drop(fs::remove_file(format!("{path_root}/{file_name}")).await); + } } } - Ok(()) + Ok(Generation::new(max_generation)) } move_old_cache(shared_context, rename_fn).await?; - add_files_to_cache( + move_old_cache_2(shared_context, rename_fn).await?; + + let max_digest_generation = add_files_to_cache( evicting_map, anchor_time, shared_context, @@ -925,7 +1002,7 @@ async fn add_files_to_cache( ) .await?; - add_files_to_cache( + let max_str_generation = add_files_to_cache( evicting_map, anchor_time, shared_context, @@ -933,7 +1010,10 @@ async fn add_files_to_cache( STR_FOLDER, ) .await?; - Ok(()) + + Ok(Generation::new( + cmp::max(max_digest_generation.inner(), max_str_generation.inner()) + 1, + )) } async fn prune_temp_path(temp_path: &str) -> Result<(), Error> { @@ -1119,6 +1199,8 @@ pub struct FilesystemStore { rename_fn: fn(&OsStr, &OsStr) -> Result<(), std::io::Error>, /// Limits concurrent write operations to prevent disk I/O saturation. write_semaphore: Option, + /// A monotonic counter by which we stamp newly added files. + next_generation: AtomicU64, /// See [`FlushCoalescer`]: amortizes the per-blob `F_FULLFSYNC` cost /// across concurrent uploads without weakening durability. `None` when /// the sentinel could not be created (e.g. read-only content volume); @@ -1209,7 +1291,7 @@ impl FilesystemStore { } else { spec.block_size }; - add_files_to_cache( + let next_generation = add_files_to_cache( evicting_map.as_ref(), &now, &shared_context, @@ -1251,6 +1333,7 @@ impl FilesystemStore { evict_page_cache: spec.evict_page_cache, weak_self: weak_self.clone(), rename_fn, + next_generation: AtomicU64::new(next_generation.inner()), write_semaphore, #[cfg(target_os = "macos")] flush_coalescer, @@ -1279,6 +1362,10 @@ impl FilesystemStore { .await; } + fn get_and_update_generation(&self) -> Generation { + Generation::new(self.next_generation.fetch_add(1, Ordering::Relaxed)) + } + /// Path of the read-only executable (0o555) variant for `digest`. #[cfg(unix)] fn executable_variant_path(&self, digest: &DigestInfo) -> OsString { @@ -1664,6 +1751,7 @@ impl FilesystemStore { &PathType::Content, encoded_file_path.shared_context.as_ref(), &key, + encoded_file_path.generation, ); let from_path = encoded_file_path.get_file_path(); @@ -1811,6 +1899,7 @@ impl FilesystemStore { shared_context: self.shared_context.clone(), path_type: PathType::Temp, key: temp_key, + generation: self.get_and_update_generation(), }, ) .await @@ -1971,6 +2060,7 @@ impl StoreDriver for FilesystemStore { shared_context: self.shared_context.clone(), path_type: PathType::Custom(path), key: key.borrow().into_owned(), + generation: self.get_and_update_generation(), }), ); // We are done with the file, if we hold a reference to the file here, it could diff --git a/nativelink-store/tests/filesystem_store_test.rs b/nativelink-store/tests/filesystem_store_test.rs index 17da82e59..6018ea3e2 100644 --- a/nativelink-store/tests/filesystem_store_test.rs +++ b/nativelink-store/tests/filesystem_store_test.rs @@ -32,7 +32,7 @@ use nativelink_error::{Code, Error, ErrorContext, ResultExt, make_err}; use nativelink_macro::nativelink_test; use nativelink_store::filesystem_store::{ DIGEST_FOLDER, EncodedFilePath, FileEntry, FileEntryImpl, FileType, FilesystemStore, - STR_FOLDER, check_duplicate_files, key_from_file, make_temp_key, + STR_FOLDER, check_duplicate_files, key_and_generation_from_file, make_temp_key, }; use nativelink_util::buf_channel::make_buf_channel_pair; use nativelink_util::common::{DigestInfo, fs, make_temp_path}; @@ -69,6 +69,9 @@ trait FileEntryHooks { core::future::ready(Ok(())) } fn on_unref(_entry: &Fe) {} + fn on_unref_async(_entry: &Fe) -> impl Future + Send { + core::future::ready(()) + } fn on_drop(_entry: &Fe) {} } @@ -165,6 +168,7 @@ impl LenEntry for TestFileEntry Result<(), Error> { Ok(()) } +/// Find a file backing `key` in the content directory. +async fn find_content_file( + content_path: &str, + key: &StoreKey<'_>, +) -> Result, Error> { + let (folder, prefix) = match key { + StoreKey::Digest(digest) => (DIGEST_FOLDER, format!("{digest}-")), + StoreKey::Str(name) => (STR_FOLDER, format!("{name}-")), + }; + let (_permit, dir_handle) = fs::read_dir(format!("{content_path}/{folder}")) + .await + .err_tip(|| "Failed opening content directory")? + .into_inner(); + let mut read_dir_stream = ReadDirStream::new(dir_handle); + while let Some(dir_entry) = read_dir_stream.next().await { + let dir_entry = dir_entry?; + if dir_entry.file_name().to_string_lossy().starts_with(&prefix) { + return Ok(Some(dir_entry.path().into_os_string())); + } + } + Ok(None) +} + +/// As [`find_content_file`], erroring when nothing backs `key`. +async fn content_file_path(content_path: &str, key: &StoreKey<'_>) -> Result { + find_content_file(content_path, key) + .await? + .err_tip(|| format!("No content file for {key:?} under {content_path}")) +} + +/// Whether the content directory holds a file for `key`. +async fn content_file_exists(content_path: &str, key: &StoreKey<'_>) -> Result { + Ok(find_content_file(content_path, key).await?.is_some()) +} + /// Helper function to ensure there are no temporary or content files left. async fn check_storage_dir_empty(storage_path: &str) -> Result<(), Error> { let (_permit, temp_dir_handle) = fs::read_dir(format!("{storage_path}/{DIGEST_FOLDER}")) @@ -374,10 +413,10 @@ async fn temp_files_get_deleted_on_replace_test() -> Result<(), Error> { store.update_oneshot(digest1, VALUE1.into()).await?; - let expected_file_name = OsString::from(format!("{content_path}/{DIGEST_FOLDER}/{digest1}")); { // Check to ensure our file exists where it should and content matches. - let data = read_file_contents(&expected_file_name).await?; + let content_file = content_file_path(&content_path, &StoreKey::Digest(digest1)).await?; + let data = read_file_contents(&content_file).await?; assert_eq!( &data[..], VALUE1.as_bytes(), @@ -389,8 +428,9 @@ async fn temp_files_get_deleted_on_replace_test() -> Result<(), Error> { store.update_oneshot(digest1, VALUE2.into()).await?; { - // Check to ensure our file now has new content. - let data = read_file_contents(&expected_file_name).await?; + // Replacing the content publishes a new generation, so re-discover it. + let content_file = content_file_path(&content_path, &StoreKey::Digest(digest1)).await?; + let data = read_file_contents(&content_file).await?; assert_eq!( &data[..], VALUE2.as_bytes(), @@ -673,7 +713,7 @@ async fn eviction_on_insert_calls_unref_once() -> Result<(), Error> { fn on_unref(file_entry: &Fe) { block_on(file_entry.get_file_path_locked(move |path_str| async move { let path = Path::new(&path_str); - let digest = key_from_file( + let (digest, _generation) = key_and_generation_from_file( path.file_name().unwrap().to_str().unwrap(), FileType::Digest, ) @@ -1151,9 +1191,10 @@ async fn update_file_future_drops_before_rename() -> Result<(), Error> { // Ensure the entry we inserted was properly flagged as moved (from temp -> content dir). new_file_entry .get_file_path_locked(move |file_path| async move { - assert_eq!( - file_path, - OsString::from(format!("{content_path}/{DIGEST_FOLDER}/{digest}")) + let expected_prefix = format!("{content_path}/{DIGEST_FOLDER}/{digest}-"); + assert!( + file_path.to_string_lossy().starts_with(&expected_prefix), + "expected {file_path:?} to be a content file for {digest}" ); Ok(()) }) @@ -1183,7 +1224,7 @@ async fn deleted_file_removed_from_store() -> Result<(), Error> { store.update_oneshot(digest, VALUE1.into()).await?; - let stored_file_path = OsString::from(format!("{content_path}/{DIGEST_FOLDER}/{digest}")); + let stored_file_path = content_file_path(&content_path, &StoreKey::Digest(digest)).await?; std::fs::remove_file(stored_file_path)?; let get_part_res = store.get_part_unchunked(digest, 0, None).await; @@ -1198,7 +1239,8 @@ async fn deleted_file_removed_from_store() -> Result<(), Error> { .await .unwrap(); - let stored_file_path = OsString::from(format!("{content_path}/{STR_FOLDER}/{STRING_NAME}")); + let stored_file_path = + content_file_path(&content_path, &StoreKey::new_str(STRING_NAME)).await?; std::fs::remove_file(stored_file_path)?; let string_digest_get_part_res = store.get_part_unchunked(string_key, 0, None).await; @@ -1345,7 +1387,7 @@ async fn update_with_whole_file_uses_same_inode() -> Result<(), Error> { original_inode }; - let expected_file_name = OsString::from(format!("{content_path}/{DIGEST_FOLDER}/{digest}")); + let expected_file_name = content_file_path(&content_path, &StoreKey::Digest(digest)).await?; // Content blobs are stored read-only (0o444), so they cannot be opened for // write (`fs::create_file` opens read+write+truncate). Stat the path // directly to read the inode instead of opening the file. @@ -1511,7 +1553,7 @@ async fn add_too_early_files() -> Result<(), Error> { let demo_file_folder = format!("{content_path}/s"); fs::create_dir_all(&demo_file_folder).await?; - let demo_file_path = format!("{demo_file_folder}/foo"); + let demo_file_path = format!("{demo_file_folder}/foo-0"); std::fs::write(&demo_file_path, "demo text") .err_tip(|| format!("writing to {demo_file_path}"))?; debug!(%demo_file_path, "demo file path"); @@ -1535,7 +1577,7 @@ async fn add_too_early_files() -> Result<(), Error> { .err_tip(|| "during FileSystemStore::new")?; assert!(logs_contain( - "File access time newer than FilesystemStore start time file_name=foo atime=20" + "File access time newer than FilesystemStore start time file_name=foo-0 atime=20" )); Ok(()) @@ -1619,7 +1661,7 @@ async fn executable_hardlink_source_created_once_and_readonly() -> Result<(), Er store.update_oneshot(digest, VALUE1.into()).await?; // The CAS blob itself is stored read-only 0o444. - let blob_path = OsString::from(format!("{content_path}/{DIGEST_FOLDER}/{digest}")); + let blob_path = content_file_path(&content_path, &StoreKey::Digest(digest)).await?; let blob_meta = fs::metadata(&blob_path).await?; assert_eq!( blob_meta.mode() & 0o777, @@ -1960,7 +2002,7 @@ async fn get_part_on_map_disk_divergence_warns_and_removes_entry() -> Result<(), store.update_oneshot(digest, VALUE1.into()).await?; // Delete the backing file out from under the still-present map entry. - let content_file = OsString::from(format!("{content_path}/{DIGEST_FOLDER}/{digest}")); + let content_file = content_file_path(&content_path, &StoreKey::Digest(digest)).await?; fs::remove_file(&content_file).await?; let err = store @@ -2009,7 +2051,7 @@ async fn unref_is_idempotent_when_file_already_gone() -> Result<(), Error> { store.update_oneshot(digest, VALUE1.into()).await?; let file_entry = store.get_file_entry_for_digest(&digest).await?; - let content_file = OsString::from(format!("{content_path}/{DIGEST_FOLDER}/{digest}")); + let content_file = content_file_path(&content_path, &StoreKey::Digest(digest)).await?; fs::remove_file(&content_file).await?; // First unref: rename hits ENOENT (benign) and flips the entry to Temp. @@ -2060,7 +2102,7 @@ async fn unref_does_not_orphan_content_file_when_temp_dir_missing() -> Result<() "an intact content file must not take the benign vanished-source path" ); // The content file must still exist — not orphaned by a wrong Temp flip. - let content_file = OsString::from(format!("{content_path}/{DIGEST_FOLDER}/{digest}")); + let content_file = content_file_path(&content_path, &StoreKey::Digest(digest)).await?; assert_eq!( read_file_contents(&content_file).await?, VALUE1.as_bytes(), @@ -2140,3 +2182,172 @@ async fn concurrent_upload_burst_round_trip_test() -> Result<(), Error> { } Ok(()) } + +/// Regression test. Operations on the eviction map (add/removal) can run at different times +/// w.r.t. moving files on a filesystem. In the past, file names contained only the digest and size. +/// +/// The following happened in past. +/// - An entry was evicted from the map. The removal fs task was scheduled. +/// - Meanwhile, a client reuploads the file. A new entry is added to the map, and the file is moved +/// into the cache on the filesystem. +/// - Now the fs removal task runs, removing the file on the filesystem, leaving an entry in the map. +/// +/// This was mitigated by making file names unique using a generation counter. +#[nativelink_test] +async fn stale_unref_must_not_delete_reuploaded_file() -> Result<(), Error> { + static UNREF_GATE: Semaphore = Semaphore::const_new(0); + + struct GatedUnrefHooks; + impl FileEntryHooks for GatedUnrefHooks { + async fn on_unref_async(entry: &Fe) { + if entry.len() == VALUE1.len() as u64 { + let _permit = UNREF_GATE + .acquire() + .await + .expect("unref gate closed unexpectedly"); + } + } + } + + // Sized apart from VALUE1 so the hook can tell the two unrefs apart. + let other_value = "y".repeat(64); + let digest1 = DigestInfo::try_new(HASH1, VALUE1.len())?; + let digest2 = DigestInfo::try_new(HASH2, other_value.len())?; + let content_path = make_temp_path("content_path"); + let temp_path = make_temp_path("temp_path"); + + let store = Arc::new( + FilesystemStore::>::new(&FilesystemSpec { + content_path: content_path.clone(), + temp_path: temp_path.clone(), + eviction_policy: Some(EvictionPolicy { + max_count: 1, + ..Default::default() + }), + block_size: 1, + ..Default::default() + }) + .await?, + ); + store.update_oneshot(digest1, VALUE1.into()).await?; + assert!( + content_file_exists(&content_path, &StoreKey::Digest(digest1)).await?, + "digest1 should be on disk after its upload" + ); + + // Evict digest1, suspending its unref before the rename. + let evicting_store = store.clone(); + let eviction = spawn!("stale_unref_eviction", async move { + evicting_store + .update_oneshot(digest2, other_value.into()) + .await + }); + while store.has(digest1).await?.is_some() { + tokio::task::yield_now().await; + } + assert!( + content_file_exists(&content_path, &StoreKey::Digest(digest1)).await?, + "digest1's file should outlive its map entry until unref runs" + ); + + // The client sees the blob as missing and re-uploads it. + store.update_oneshot(digest1, VALUE1.into()).await?; + assert_eq!( + store.has(digest1).await?, + Some(VALUE1.len() as u64), + "the re-uploaded blob should be resident" + ); + assert_eq!( + store.get_part_unchunked(digest1, 0, None).await?, + VALUE1.as_bytes(), + "the re-uploaded blob should be readable" + ); + + // Let the evicted entry finish its unref. + UNREF_GATE.add_permits(1); + eviction + .await + .expect("eviction task panicked") + .err_tip(|| "Failed to insert digest2")?; + + assert!( + content_file_exists(&content_path, &StoreKey::Digest(digest1)).await?, + "stale cleanup deleted the re-uploaded file for {digest1}" + ); + assert_eq!( + store.get_part_unchunked(digest1, 0, None).await?, + VALUE1.as_bytes(), + "the re-uploaded blob should still be readable after the stale unref" + ); + Ok(()) +} + +/// Control for [`stale_unref_must_not_delete_reuploaded_file`]: the same +/// evict-then-reupload sequence, with a difference that `unref()` is allowed to +/// finish before the reupload rather than after it. +#[nativelink_test] +async fn reupload_after_completed_unref_is_readable() -> Result<(), Error> { + static UNREF_GATE: Semaphore = Semaphore::const_new(0); + + struct GatedUnrefHooks; + impl FileEntryHooks for GatedUnrefHooks { + async fn on_unref_async(entry: &Fe) { + if entry.len() == VALUE1.len() as u64 { + let _permit = UNREF_GATE + .acquire() + .await + .expect("unref gate closed unexpectedly"); + } + } + } + + let other_value = "y".repeat(64); + let digest1 = DigestInfo::try_new(HASH1, VALUE1.len())?; + let digest2 = DigestInfo::try_new(HASH2, other_value.len())?; + let content_path = make_temp_path("content_path"); + let temp_path = make_temp_path("temp_path"); + + let store = Arc::new( + FilesystemStore::>::new(&FilesystemSpec { + content_path: content_path.clone(), + temp_path: temp_path.clone(), + eviction_policy: Some(EvictionPolicy { + max_count: 1, + ..Default::default() + }), + block_size: 1, + ..Default::default() + }) + .await?, + ); + store.update_oneshot(digest1, VALUE1.into()).await?; + + // Same eviction, drained fully before the client re-uploads. + let evicting_store = store.clone(); + let eviction = spawn!("completed_unref_eviction", async move { + evicting_store + .update_oneshot(digest2, other_value.into()) + .await + }); + while store.has(digest1).await?.is_some() { + tokio::task::yield_now().await; + } + UNREF_GATE.add_permits(1); + eviction + .await + .expect("eviction task panicked") + .err_tip(|| "Failed to insert digest2")?; + + // No stale unref is outstanding by the time this lands. + store.update_oneshot(digest1, VALUE1.into()).await?; + assert!( + content_file_exists(&content_path, &StoreKey::Digest(digest1)).await?, + "the re-uploaded file should be on disk for {digest1}" + ); + assert_eq!( + store.get_part_unchunked(digest1, 0, None).await?, + VALUE1.as_bytes(), + "the re-uploaded blob should be readable" + ); + Ok(()) +}