From 52eb7bb72fd526797a3957fc9e144003cbdc8ee3 Mon Sep 17 00:00:00 2001
From: Ashley Coleman <ashley.coleman@sifive.com>
Date: Wed, 25 Sep 2024 14:47:09 -0600
Subject: [PATCH] rsc: Delete multiple blob files per task by config (#1652)

* rsc: Delete multiple blob files per task by config

* cleanup

* Update rust/rsc/src/bin/rsc/config.rs

Co-authored-by: Colin Schmidt <colin.schmidt@sifive.com>

---------

Co-authored-by: Colin Schmidt <colin.schmidt@sifive.com>
---
 rust/rsc/.config.json          |  3 ++-
 rust/rsc/src/bin/rsc/config.rs | 16 ++++++++++--
 rust/rsc/src/bin/rsc/main.rs   | 45 ++++++++++++++++++++++------------
 3 files changed, 45 insertions(+), 19 deletions(-)

diff --git a/rust/rsc/.config.json b/rust/rsc/.config.json
index e6b1b9cc6..5bc1ded25 100644
--- a/rust/rsc/.config.json
+++ b/rust/rsc/.config.json
@@ -8,7 +8,8 @@
   "blob_eviction": {
     "tick_rate": 60,
     "ttl": 3600,
-    "chunk_size": 16000
+    "chunk_size": 16000,
+    "file_chunk_size": 100
   },
   "job_eviction": {
     "ttl": {
diff --git a/rust/rsc/src/bin/rsc/config.rs b/rust/rsc/src/bin/rsc/config.rs
index ea600c373..cbdb3f502 100644
--- a/rust/rsc/src/bin/rsc/config.rs
+++ b/rust/rsc/src/bin/rsc/config.rs
@@ -15,8 +15,20 @@ pub struct RSCTTLConfig {
     pub tick_rate: u64,
     // How long an object is allowed to live
     pub ttl: u64,
-    // Maximum number of objects to delete at a time. Must be 1 >= x <= 16000
+    // Maximum number of objects to delete from the db at a time. Must be 1 >= x <= 16000
+    pub chunk_size: u32,
+}
+
+#[derive(Debug, Deserialize, Serialize)]
+pub struct RSCBlobTTLConfig {
+    // How often to run the eviction check in seconds
+    pub tick_rate: u64,
+    // How long an object is allowed to live
+    pub ttl: u64,
+    // Maximum number of blobs to delete from the db at a time. Must be 1 >= x <= 64000
     pub chunk_size: u32,
+    // Maximum number of files to delete from the disk per task
+    pub file_chunk_size: usize,
 }
 
 #[derive(Debug, Deserialize, Serialize)]
@@ -62,7 +74,7 @@ pub struct RSCConfig {
     // The directory that server logs should be written to. If None logs are written to stdout
     pub log_directory: Option<String>,
     // The config to control blob eviction
-    pub blob_eviction: RSCTTLConfig,
+    pub blob_eviction: RSCBlobTTLConfig,
     // The config to control job eviction
     pub job_eviction: RSCJobEvictionConfig,
     // The config to control job size calculation
diff --git a/rust/rsc/src/bin/rsc/main.rs b/rust/rsc/src/bin/rsc/main.rs
index dc4e3ffc1..fffe51bbc 100644
--- a/rust/rsc/src/bin/rsc/main.rs
+++ b/rust/rsc/src/bin/rsc/main.rs
@@ -5,6 +5,7 @@ use axum::{
 };
 use chrono::Utc;
 use clap::Parser;
+use itertools::Itertools;
 use migration::{Migrator, MigratorTrait};
 use rlimit::Resource;
 use rsc::database;
@@ -279,23 +280,34 @@ fn launch_blob_eviction(
             tracing::info!(%deleted, "N blobs deleted for eviction");
 
             // Delete blobs from blob store
-            for blob in blobs {
-                let store = match blob_stores.get(&blob.store_id) {
-                    Some(s) => s.clone(),
-                    None => {
-                        let blob = blob.clone();
-                        tracing::info!(%blob.store_id, %blob.key, "Blob has been orphaned!");
-                        tracing::error!(%blob.store_id, "Blob's store id missing from activated stores");
-                        continue;
-                    }
-                };
+            let chunked: Vec<Vec<database::DeletedBlob>> = blobs
+                .into_iter()
+                .chunks(config.blob_eviction.file_chunk_size)
+                .into_iter()
+                .map(|chunk| chunk.collect())
+                .collect();
+
+            for chunk in chunked {
+                let thread_store = blob_stores.clone();
 
                 tokio::spawn(async move {
-                    store.delete_key(blob.key.clone()).await.unwrap_or_else(|err| {
-                        let blob = blob.clone();
-                        tracing::info!(%blob.store_id, %blob.key, "Blob has been orphaned!");
-                        tracing::error!(%err, "Failed to delete blob from store for eviction. See above for blob info");
-                    });
+                    for blob in chunk {
+                        let store = match thread_store.get(&blob.store_id) {
+                            Some(s) => s.clone(),
+                            None => {
+                                let blob = blob.clone();
+                                tracing::info!(%blob.store_id, %blob.key, "Blob has been orphaned!");
+                                tracing::error!(%blob.store_id, "Blob's store id missing from activated stores");
+                                continue;
+                            }
+                        };
+
+                        store.delete_key(blob.key.clone()).await.unwrap_or_else(|err| {
+                            let blob = blob.clone();
+                            tracing::info!(%blob.store_id, %blob.key, "Blob has been orphaned!");
+                            tracing::error!(%err, "Failed to delete blob from store for eviction. See above for blob info");
+                        });
+                    }
                 });
             }
         }
@@ -481,10 +493,11 @@ mod tests {
             active_store: store_id.to_string(),
             connection_pool_timeout: 10,
             log_directory: None,
-            blob_eviction: config::RSCTTLConfig {
+            blob_eviction: config::RSCBlobTTLConfig {
                 tick_rate: 10,
                 ttl: 100,
                 chunk_size: 100,
+                file_chunk_size: 1,
             },
             job_eviction: config::RSCJobEvictionConfig::TTL(config::RSCTTLConfig {
                 tick_rate: 10,