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

[Christian Ebner <c.ebner@proxmox.com>]

Code changes look good to me in general, just a few suggestions to make 
the dependency on the datastore's chunk store instance a bit more explicit.

On 9/15/26 6:00 PM, Shan Shaji wrote:
> 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,

comment: As is, this does not need to be a method of DataStore, the 
reference to self is never used after all.

But I would keep it as such and make the dependency on the datastore and 
it's chunk store explicit.

To do this, I suggest to pass in the chunk store by reference to the 
callback and use the reference for all chunk store operations in the 
callback so the relation becomes stringent. So ...

> +        worker: &dyn WorkerTaskContext,
> +        s3_client: &S3Client,
> +        mut callback: F,
> +    ) -> Result<(), Error>
> +    where
> +        F: FnMut(S3ObjectKey, u64) -> Result<(), Error>,

comment: Instead of the generic, this could use an `impl Trait` 
parameter declaration, which makes the method signature cleaner, and 
with the suggested reference to the chunk store passed along would become:

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()?;
> +

...  here we would now call:
```
callback(self.inner.chunk_store.as_ref(), content.key, content.size)?;
```

> +                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> {

... this would then have the direct reference to the chunk store:

```
|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) {

... so operations on that would use that directly instead of binding to 
self, so here for example
```
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();
>   
> @@ -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)?;