chore(cubestore): Set metastore deletion batch size based on upload_concurrency option

Author: srh

rust/cubestore/cubestore/src/metastore/rocks_fs.rs

@@ -55,7 +55,8 @@ pub struct BaseRocksStoreFs {
     name: &'static str,
     minimum_snapshots_count: u64,
     snapshots_lifetime: u64,
-    remote_files_cleanup_batch_size: u64,
+    // A copy of the upload-concurrency config -- we multiply this for our deletes.
+    deletion_batch_size: u64,
 }
 
 impl BaseRocksStoreFs {
@@ -65,13 +66,13 @@ impl BaseRocksStoreFs {
     ) -> Arc<Self> {
         let minimum_snapshots_count = config.minimum_metastore_snapshots_count();
         let snapshots_lifetime = config.metastore_snapshots_lifetime();
-        let remote_files_cleanup_batch_size = config.remote_files_cleanup_batch_size();
+        let deletion_batch_size = Self::deletion_batch_size(config.as_ref());
         Arc::new(Self {
             remote_fs,
             name: "metastore",
             minimum_snapshots_count,
             snapshots_lifetime,
-            remote_files_cleanup_batch_size,
+            deletion_batch_size,
         })
     }
     pub fn new_for_cachestore(
@@ -80,16 +81,23 @@ impl BaseRocksStoreFs {
     ) -> Arc<Self> {
         let minimum_snapshots_count = config.minimum_cachestore_snapshots_count();
         let snapshots_lifetime = config.cachestore_snapshots_lifetime();
-        let remote_files_cleanup_batch_size = config.remote_files_cleanup_batch_size();
+        let deletion_batch_size = Self::deletion_batch_size(config.as_ref());
         Arc::new(Self {
             remote_fs,
             name: "cachestore",
             minimum_snapshots_count,
             snapshots_lifetime,
-            remote_files_cleanup_batch_size,
+            deletion_batch_size,
         })
     }
 
+    fn deletion_batch_size(config: &dyn ConfigObj) -> u64 {
+        // Pick a large enough batch size that batching is not a significant slowdown, given the
+        // upload_concurrency parameter.
+        const MULTIPLIER: u64 = 25;
+        config.upload_concurrency().saturating_mul(MULTIPLIER)
+    }
+
     pub fn get_name(&self) -> &'static str {
         &self.name
     }
@@ -214,11 +222,7 @@ impl BaseRocksStoreFs {
             );
         }
 
-        for batch in to_delete.chunks(
-            self.remote_files_cleanup_batch_size
-                .try_into()
-                .unwrap_or(usize::MAX),
-        ) {
+        for batch in to_delete.chunks(self.deletion_batch_size.try_into().unwrap_or(usize::MAX)) {
             for v in join_all(
                 batch
                     .iter()