fix(filesystem_store): bound background delete concurrency with a semaphore

nativelink-config/src/stores.rs

@@ -835,6 +835,32 @@ pub struct FilesystemSpec {
     #[serde(default, deserialize_with = "convert_numeric_with_shellexpand")]
     pub max_concurrent_writes: usize,
 
+    /// Maximum number of concurrent background file-deletion operations.
+    /// When an entry leaves the store, its backing file is unlinked by a
+    /// background task. Without a bound, an eviction storm (cache wipe or a
+    /// flood of overwrites) can spawn these without limit: memory grows with the
+    /// queued task frames and disk/inode reclamation lags behind what the async
+    /// runtime can schedule. Bounding it applies backpressure the same way
+    /// `max_concurrent_writes` does — once this many deletes are in flight, a
+    /// further delete runs synchronously inline in the evicting path instead of
+    /// spawning, which throttles eviction to delete throughput and never leaks
+    /// a file.
+    ///
+    /// Bounded by default, unlike `max_concurrent_writes`: an unbounded delete
+    /// backlog is a latent memory/inode blowup, and leaving the protection off
+    /// by default just moves that risk from a high-context code review to a
+    /// 4 AM config change nobody remembers to make. The default is generous
+    /// enough that normal steady-state eviction never reaches it (so the inline
+    /// path never fires in ordinary operation); it only engages under a genuine
+    /// storm, which is exactly when the backpressure is wanted.
+    /// A value of 0 is the explicit escape hatch for unlimited.
+    /// Default: 1024.
+    #[serde(
+        default = "default_max_concurrent_deletes",
+        deserialize_with = "convert_numeric_with_shellexpand"
+    )]
+    pub max_concurrent_deletes: usize,
+
     /// When true, advise the kernel to drop the page cache for each blob after
     /// it is written or read (`posix_fadvise` with `POSIX_FADV_DONTNEED`). On
     /// real filesystems this takes a globally serialized, all-CPU kernel path
@@ -1747,6 +1773,14 @@ const fn default_read_batching_dispatch_slots() -> usize {
     4
 }
 
+/// Generous enough that normal eviction never reaches it (so the synchronous
+/// inline delete path never fires in ordinary operation), low enough that the
+/// in-flight backlog stays bounded to a few MB under a storm. Tune per
+/// deployment; 0 disables the bound. See `FilesystemSpec::max_concurrent_deletes`.
+const fn default_max_concurrent_deletes() -> usize {
+    1024
+}
+
 const fn default_read_batching_max_queued_bytes() -> u64 {
     32 * 1024 * 1024 // 32 MiB.
 }

nativelink-store/src/filesystem_store.rs

@@ -102,6 +102,17 @@ pub struct SharedContext {
     // deleted (similar to how it is done in tests).
     #[metric(help = "Number of active drop spawns")]
     pub active_drop_spawns: AtomicU64,
+    /// Count of deletes that ran synchronously inline because the delete
+    /// semaphore was saturated. A nonzero, growing value means eviction is
+    /// outrunning background delete throughput — the backpressure signal.
+    #[metric(help = "Number of deletes performed inline due to delete backpressure")]
+    pub inline_deletes: AtomicU64,
+    /// Bounds the number of in-flight background file deletions. `None` means
+    /// unlimited (the historical behavior). When the permits are exhausted,
+    /// a drop deletes its file synchronously inline instead of spawning, so
+    /// an eviction storm applies backpressure rather than spawning without
+    /// limit. See `FilesystemSpec::max_concurrent_deletes`.
+    delete_semaphore: Option<Arc<Semaphore>>,
     #[metric(help = "Path to the configured temp path")]
     temp_path: String,
     #[metric(help = "Path to the configured content path")]
@@ -187,6 +198,40 @@ impl Drop for EncodedFilePath {
 
         let file_path = self.get_file_path().to_os_string();
         let shared_context = self.shared_context.clone();
+
+        // Bound the number of in-flight background deletes. `drop` is sync, so
+        // we cannot `.await` a permit the way the write path does; instead we
+        // `try_acquire_owned` (never blocks) and, when the permits are
+        // exhausted, delete synchronously inline. That inline path is the
+        // backpressure: the evicting thread pays the unlink cost directly
+        // instead of queueing an unbounded number of tasks, and the file is
+        // always removed (never leaked). `None` keeps the historical
+        // unbounded-spawn behavior.
+        let delete_permit = match &shared_context.delete_semaphore {
+            Some(sem) => {
+                if let Ok(permit) = Arc::clone(sem).try_acquire_owned() {
+                    Some(permit)
+                } else {
+                    // Saturated: delete synchronously inline, which is the
+                    // backpressure, then bail out (no spawn).
+                    shared_context
+                        .inline_deletes
+                        .fetch_add(1, Ordering::Relaxed);
+                    match std::fs::remove_file(&file_path) {
+                        Ok(()) => {
+                            debug!(?file_path, "File deleted inline (delete backpressure)");
+                        }
+                        Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
+                            debug!(?file_path, "File already gone, nothing to delete");
+                        }
+                        Err(err) => error!(?file_path, ?err, "Failed to delete file inline"),
+                    }
+                    return;
+                }
+            }
+            None => None,
+        };
+
         // .fetch_add returns previous value, so we add one to get approximate current value
         let current_active_drop_spawns = shared_context
             .active_drop_spawns
@@ -198,6 +243,9 @@ impl Drop for EncodedFilePath {
             "Spawned a filesystem_delete_file"
         );
         background_spawn!("filesystem_delete_file", async move {
+            // Hold the permit until the delete actually completes so the bound
+            // reflects deletes in flight, not merely scheduled.
+            let _delete_permit = delete_permit;
             match fs::remove_file(&file_path).await {
                 Ok(()) => debug!(?file_path, "File deleted"),
                 // The file already being gone is the desired end state of a
@@ -1433,8 +1481,15 @@ impl<Fe: FileEntry> FilesystemStore<Fe> {
             enabled
         };
 
+        let delete_semaphore = if spec.max_concurrent_deletes > 0 {
+            Some(Arc::new(Semaphore::new(spec.max_concurrent_deletes)))
+        } else {
+            None
+        };
         let shared_context = Arc::new(SharedContext {
             active_drop_spawns: AtomicU64::new(0),
+            inline_deletes: AtomicU64::new(0),
+            delete_semaphore,
             temp_path: spec.temp_path.clone(),
             content_path: spec.content_path.clone(),
         });

nativelink-store/tests/filesystem_store_test.rs

@@ -711,6 +711,62 @@ async fn file_gets_cleans_up_on_cache_eviction() -> Result<(), Error> {
     check_storage_dir_empty(&temp_path).await
 }
 
+// Bounding background deletes with `max_concurrent_deletes` must not change the
+// correctness of eviction: an evicted entry's backing file is still removed.
+// This exercises the bounded (semaphore) delete path — permit acquired, held
+// for the lifetime of the delete, then released. The default unbounded path is
+// covered by the rest of this suite; the inline-overflow fallback (which fires
+// only under saturation) is exercised under real load by the furnace via the
+// `inline_deletes` metric.
+#[nativelink_test]
+async fn bounded_delete_semaphore_still_removes_evicted_files() -> Result<(), Error> {
+    static DELETES_FINISHED: AtomicU32 = AtomicU32::new(0);
+    struct LocalHooks {}
+    impl FileEntryHooks for LocalHooks {
+        fn on_drop<Fe: FileEntry>(_file_entry: &Fe) {
+            DELETES_FINISHED.fetch_add(1, Ordering::Relaxed);
+        }
+    }
+
+    let digest1 = DigestInfo::try_new(HASH1, VALUE1.len())?;
+    let digest2 = DigestInfo::try_new(HASH2, VALUE2.len())?;
+    let content_path = make_temp_path("content_path");
+    let temp_path = make_temp_path("temp_path");
+
+    let store = Arc::new(
+        FilesystemStore::<TestFileEntry<LocalHooks>>::new(&FilesystemSpec {
+            content_path: content_path.clone(),
+            temp_path: temp_path.clone(),
+            eviction_policy: Some(EvictionPolicy {
+                max_count: 1,
+                ..Default::default()
+            }),
+            // The bound under test. With it set, the delete routes through the
+            // semaphore path rather than the unconditional spawn.
+            max_concurrent_deletes: 1,
+            block_size: 1,
+            read_buffer_size: 1,
+            ..Default::default()
+        })
+        .await?,
+    );
+
+    // Insert, then evict digest1 by inserting digest2 (max_count == 1).
+    store.update_oneshot(digest1, VALUE1.into()).await?;
+    store.update_oneshot(digest2, VALUE2.into()).await?;
+
+    // Wait for the eviction's background delete to finish.
+    loop {
+        if DELETES_FINISHED.load(Ordering::Relaxed) == 1 {
+            break;
+        }
+        tokio::task::yield_now().await;
+    }
+
+    // The evicted file must be gone from the temp staging area.
+    check_storage_dir_empty(&temp_path).await
+}
+
 // Test to ensure that if we are holding a reference to `FileEntry` and the contents are
 // replaced, the `FileEntry` continues to use the old data.
 // `FileEntry` file contents should be immutable for the lifetime of the object.