[RFC PATCH proxmox-backup 1/4] datastore: factor out s3 chunk objects pagination into a helper

[Shan Shaji <s.shaji@proxmox.com>]

This is in preparation to re-use the chunk iteration logic for creating
empty marker files during s3 refresh.

Signed-off-by: Shan Shaji <s.shaji@proxmox.com>
---
 pbs-datastore/src/datastore.rs | 85 ++++++++++++++++++++++------------
 1 file changed, 56 insertions(+), 29 deletions(-)

diff --git a/pbs-datastore/src/datastore.rs b/pbs-datastore/src/datastore.rs
index c3db52269..f3eab8a28 100644
--- a/pbs-datastore/src/datastore.rs
+++ b/pbs-datastore/src/datastore.rs
@@ -2382,6 +2382,47 @@ impl DataStore {
         }
     }
 
+    /// Fetch chunk objects in batches of max 1000 per request.
+    /// For each objects listed, the `callback` will be invoked with its
+    /// S3 object key and size.
+    fn paginate_s3_chunk_objects<F>(
+        &self,
+        worker: &dyn WorkerTaskContext,
+        s3_client: &S3Client,
+        mut callback: F,
+    ) -> Result<(), Error>
+    where
+        F: FnMut(S3ObjectKey, u64) -> Result<(), Error>,
+    {
+        let prefix = S3PathPrefix::Some(".chunks/".to_string());
+        let mut list_bucket_result =
+            proxmox_async::runtime::block_on(s3_client.list_objects_v2(&prefix, None))
+                .context("failed to list chunk in s3 object store")?;
+
+        loop {
+            for content in list_bucket_result.contents {
+                worker.check_abort()?;
+                worker.fail_on_shutdown()?;
+
+                callback(content.key, content.size)?;
+            }
+
+            // Process next batch of chunks if there are more.
+            if list_bucket_result.is_truncated {
+                list_bucket_result = proxmox_async::runtime::block_on(s3_client.list_objects_v2(
+                    &prefix,
+                    list_bucket_result.next_continuation_token.as_deref(),
+                ))?;
+
+                continue;
+            }
+
+            break;
+        }
+
+        Ok(())
+    }
+
     fn garbage_collection_impl(
         &self,
         worker: &dyn WorkerTaskContext,
@@ -2473,28 +2514,23 @@ impl DataStore {
 
         if let Some(ref s3_client) = s3_client {
             let mut chunk_count = 0;
-            let prefix = S3PathPrefix::Some(".chunks/".to_string());
-            // Operates in batches of 1000 objects max per request
-            let mut list_bucket_result =
-                proxmox_async::runtime::block_on(s3_client.list_objects_v2(&prefix, None))
-                    .context("failed to list chunk in s3 object store")?;
-
             let mut delete_list =
                 S3DeleteList::with_thresholds(S3_DELETE_BATCH_LIMIT, S3_DELETE_DEFER_LIMIT_SECONDS);
-            loop {
-                for content in list_bucket_result.contents {
-                    worker.check_abort()?;
-                    worker.fail_on_shutdown()?;
+
+            self.paginate_s3_chunk_objects(
+                worker,
+                s3_client,
+                |object_key, size| -> Result<(), Error> {
                     let (chunk_path, digest, bad) =
-                        match self.chunk_path_from_object_key(&content.key) {
+                        match self.chunk_path_from_object_key(&object_key) {
                             Some(path) => path,
-                            None => continue,
+                            None => return Ok(()),
                         };
 
                     let timeout = std::time::Duration::from_secs(0);
                     let _chunk_guard = match self.inner.chunk_store.lock_chunk(&digest, timeout) {
                         Ok(guard) => guard,
-                        Err(_) => continue,
+                        Err(_) => return Ok(()),
                     };
                     let _guard = self.inner.chunk_store.mutex().lock().unwrap();
 
@@ -2526,7 +2562,7 @@ impl DataStore {
                                 atime,
                                 min_atime,
                                 oldest_writer,
-                                content.size,
+                                size,
                                 bad,
                                 &mut gc_status,
                                 || {
@@ -2537,15 +2573,15 @@ impl DataStore {
                                             std::fs::remove_file(chunk_path)?;
                                         }
                                     }
-                                    delete_list.push(content.key, _chunk_guard);
+                                    delete_list.push(object_key, _chunk_guard);
                                     Ok(())
                                 },
                             )?;
                         }
                     } else {
                         gc_status.removed_chunks += 1;
-                        gc_status.removed_bytes += content.size;
-                        delete_list.push(content.key, _chunk_guard);
+                        gc_status.removed_bytes += size;
+                        delete_list.push(object_key, _chunk_guard);
                     }
 
                     chunk_count += 1;
@@ -2555,19 +2591,10 @@ impl DataStore {
 
                     // limit pending deletes to avoid holding too many chunk flocks
                     delete_list.conditional_delete_and_drop_locks(s3_client)?;
-                }
-                // Process next batch of chunks if there is more
-                if list_bucket_result.is_truncated {
-                    list_bucket_result =
-                        proxmox_async::runtime::block_on(s3_client.list_objects_v2(
-                            &prefix,
-                            list_bucket_result.next_continuation_token.as_deref(),
-                        ))?;
-                    continue;
-                }
 
-                break;
-            }
+                    Ok(())
+                },
+            )?;
 
             // delete the last batch of objects, if there are any remaining
             delete_list.delete_and_drop_locks(s3_client)?;
-- 
2.47.3