New filesystem test for eviction breaking

Author: palfrey

nativelink-store/src/filesystem_store.rs

@@ -220,12 +220,16 @@ pub trait FileEntry: LenEntry + Send + Sync + Debug + 'static {
 pub struct FileEntryImpl {
     data_size: u64,
     block_size: u64,
-    encoded_file_path: RwLock<EncodedFilePath>,
+    encoded_file_path: Arc<RwLock<EncodedFilePath>>,
 }
 
 impl FileEntryImpl {
-    pub fn get_shared_context_for_test(&mut self) -> Arc<SharedContext> {
-        self.encoded_file_path.get_mut().shared_context.clone()
+    pub async fn get_shared_context_for_test(&mut self) -> Arc<SharedContext> {
+        self.encoded_file_path
+            .read_arc()
+            .await
+            .shared_context
+            .clone()
     }
 }
 
@@ -234,7 +238,7 @@ impl FileEntry for FileEntryImpl {
         Self {
             data_size,
             block_size,
-            encoded_file_path,
+            encoded_file_path: Arc::new(encoded_file_path),
         }
     }
 
@@ -362,8 +366,11 @@ impl LenEntry for FileEntryImpl {
     // target file location to the new temp file. `unref()` should only ever be called once.
     #[inline]
     async fn unref(&self) {
-        {
-            let mut encoded_file_path = self.encoded_file_path.write().await;
+        // If this is called during emplace_file, it'll block, so we need to spawn a new task.
+        let local_encoded_path = self.encoded_file_path.clone();
+        debug!("Spawning filesystem_unref");
+        background_spawn!("filesystem_unref", async move {
+            let mut encoded_file_path = local_encoded_path.write().await;
             if encoded_file_path.path_type == PathType::Temp {
                 // We are already a temp file that is now marked for deletion on drop.
                 // This is very rare, but most likely the rename into the content path failed.
@@ -393,7 +400,7 @@ impl LenEntry for FileEntryImpl {
                 encoded_file_path.path_type = PathType::Temp;
                 encoded_file_path.key = new_key;
             }
-        }
+        });
     }
 }
 

nativelink-store/tests/fast_slow_store_test.rs

@@ -18,7 +18,9 @@ use std::sync::{Arc, Mutex};
 
 use async_trait::async_trait;
 use bytes::Bytes;
-use nativelink_config::stores::{FastSlowSpec, MemorySpec, NoopSpec, StoreDirection, StoreSpec};
+use nativelink_config::stores::{
+    EvictionPolicy, FastSlowSpec, MemorySpec, NoopSpec, StoreDirection, StoreSpec,
+};
 use nativelink_error::{Code, Error, ResultExt, make_err};
 use nativelink_macro::nativelink_test;
 use nativelink_metric::MetricsComponent;
@@ -32,6 +34,7 @@ use nativelink_util::store_trait::{RemoveItemCallback, Store, StoreDriver, Store
 use pretty_assertions::assert_eq;
 use rand::rngs::SmallRng;
 use rand::{Rng, SeedableRng};
+use tracing::debug;
 
 const MEGABYTE_SZ: usize = 1024 * 1024;
 
@@ -570,3 +573,97 @@ async fn fast_readonly_only_not_updated_on_get() -> Result<(), Error> {
     );
     Ok(())
 }
+
+#[ignore]
+#[nativelink_test]
+async fn slow_store_safe_eviction() -> Result<(), Error> {
+    let fast_store_spec = MemorySpec {
+        eviction_policy: Some(EvictionPolicy {
+            max_bytes: 100,
+            ..Default::default()
+        }),
+    };
+    let slow_store_spec = MemorySpec {
+        eviction_policy: Some(EvictionPolicy {
+            max_bytes: 101,
+            ..Default::default()
+        }),
+    };
+    let fast_store = Store::new(MemoryStore::new(&fast_store_spec));
+    let slow_store = Store::new(MemoryStore::new(&slow_store_spec));
+    let fast_slow_store = Store::new(FastSlowStore::new(
+        &FastSlowSpec {
+            fast: StoreSpec::Memory(fast_store_spec),
+            slow: StoreSpec::Memory(slow_store_spec),
+            fast_direction: StoreDirection::default(),
+            slow_direction: StoreDirection::default(),
+        },
+        fast_store.clone(),
+        slow_store.clone(),
+    ));
+
+    let data1 = make_random_data(100);
+    let digest1 = DigestInfo::try_new(VALID_HASH, data1.len()).unwrap();
+
+    let data2 = make_random_data(10);
+    let digest2 = DigestInfo::try_new(VALID_HASH, data2.len()).unwrap();
+
+    assert_eq!(
+        fast_slow_store
+            .has(digest1)
+            .await
+            .merge(fast_slow_store.has(digest2).await),
+        Ok(None),
+        "Expected data to not exist in store"
+    );
+
+    debug!("Uploading first data...");
+    fast_slow_store
+        .update_oneshot(digest1, data1.clone().into())
+        .await?;
+
+    debug!("Uploading second data...");
+    fast_slow_store
+        .update_oneshot(digest2, data2.clone().into())
+        .await?;
+
+    debug!("Both uploads complete");
+
+    assert_eq!(
+        fast_slow_store.has(digest1).await,
+        Ok(Some(100)),
+        "Expected first data to not exist in store, because eviction"
+    );
+
+    let (tx, mut rx) = make_buf_channel_pair();
+
+    assert_eq!(
+        fast_slow_store.get(digest1, tx).await,
+        Ok(()),
+        "Expected first data to not exist in store, because eviction"
+    );
+
+    rx.recv().await.unwrap();
+    assert_eq!(rx.get_bytes_received(), 100);
+
+    assert_eq!(
+        fast_slow_store.has(digest2).await,
+        Ok(Some(10)),
+        "Expected first data to not exist in store, because eviction"
+    );
+
+    let (tx, mut rx) = make_buf_channel_pair();
+
+    assert_eq!(
+        fast_slow_store.get(digest2, tx).await,
+        Ok(()),
+        "Expected second data to not exist in store, because eviction"
+    );
+
+    rx.recv().await.unwrap();
+    assert_eq!(rx.get_bytes_received(), 10);
+
+    assert!(false);
+
+    Ok(())
+}