[PATCH proxmox-backup 2/9] 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 | 93 +++++++++++++++++++++-------------
 1 file changed, 58 insertions(+), 35 deletions(-)

diff --git a/pbs-datastore/src/datastore.rs b/pbs-datastore/src/datastore.rs
index a74a6a824..7bb0c187c 100644
--- a/pbs-datastore/src/datastore.rs
+++ b/pbs-datastore/src/datastore.rs
@@ -2505,6 +2505,44 @@ impl DataStore {
         }
     }
 
+    // Fetch chunk objects in batches of max 1000 per request.
+    // For each listed object, invoke `callback` with a reference to the
+    // datastore's local `ChunkStore`, followed by the object's S3 key and size.
+    fn paginate_s3_chunk_objects(
+        &self,
+        worker: &dyn WorkerTaskContext,
+        s3_client: &S3Client,
+        mut callback: impl FnMut(&ChunkStore, S3ObjectKey, u64) -> Result<(), Error>,
+    ) -> 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(&self.inner.chunk_store, 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,
@@ -2596,41 +2634,35 @@ 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,
+                |chunk_store, 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) {
+                    let _chunk_guard = match chunk_store.lock_chunk(&digest, timeout) {
                         Ok(guard) => guard,
-                        Err(_) => continue,
+                        Err(_) => return Ok(()),
                     };
-                    let _guard = self.inner.chunk_store.mutex().lock().unwrap();
+                    let _guard = chunk_store.mutex().lock().unwrap();
 
                     // Check local markers (created or atime updated during phase1) and
                     // keep or delete chunk based on that.
                     let atime = match std::fs::metadata(&chunk_path) {
                         Ok(stat) => Some(stat.accessed()?),
                         Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
-                            if self.inner.chunk_store.clear_chunk_expected_mark(&digest)? {
+                            if chunk_store.clear_chunk_expected_mark(&digest)? {
                                 unsafe {
                                     // chunk store lock held
-                                    self.inner
-                                        .chunk_store
+                                    chunk_store
                                         .replace_chunk_with_marker_or_create_marker(&digest)?;
                                 }
                                 Some(SystemTime::now())
@@ -2645,11 +2677,11 @@ impl DataStore {
                         let atime = atime.duration_since(SystemTime::UNIX_EPOCH)?.as_secs() as i64;
 
                         unsafe {
-                            self.inner.chunk_store.cond_sweep_chunk(
+                            chunk_store.cond_sweep_chunk(
                                 atime,
                                 min_atime,
                                 oldest_writer,
-                                content.size,
+                                size,
                                 bad,
                                 &mut gc_status,
                                 || {
@@ -2660,15 +2692,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;
@@ -2678,19 +2710,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