Add object listing to storage trait - #6705
Conversation
Expose paginated object metadata incrementally while preserving errors and tracing across lazy S3 page fetches.
Use fallible SystemTime conversion and surface malformed S3 metadata with warning-level diagnostics.
Make the metadata and stream aliases explicit and update all storage implementations consistently.
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 2c05146b29
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
Preserve S3 listing errors when wrapping supported storage and keep prefix paths relative to their storage root.
| ): ListState| { | ||
| let page_span = list_span.clone(); | ||
| async move { | ||
| let continuation_token = match list_position { |
There was a problem hiding this comment.
| let continuation_token = match list_position { | |
| let continuation_token_opt = match list_position { |
| .list_objects_v2() | ||
| .bucket(&bucket) | ||
| .prefix(&key_prefix); | ||
| if let Some(token) = continuation_token.as_deref() { |
There was a problem hiding this comment.
| if let Some(token) = continuation_token.as_deref() { | |
| if let Some(continuation_token) = continuation_token_opt.as_deref() { |
There was a problem hiding this comment.
nit: &continuation_token also works instead of continuation_token.as_deref()
|
|
||
| // Convert S3 metadata to paths relative to the configured storage root. | ||
| let mut page_objects = Vec::with_capacity(response.contents().len()); | ||
| for object in response.contents() { |
There was a problem hiding this comment.
the conversion logic from object to response could be encapsulated in one function. That might make the whole body of list easier to read.
| } | ||
| }; | ||
| let size_bytes = match object.size() { | ||
| Some(size) => match u64::try_from(size) { |
There was a problem hiding this comment.
I think we can safely assume that size is always positive so size as u64 is safe, or we can we can also u64::try_from(size).expect("size` should be positive");
Are you using Codex by any chance? I find that Codex is needlessly defensive and verbose for error handling.
| list_position, | ||
| list_span, | ||
| ): ListState| { | ||
| let page_span = list_span.clone(); |
There was a problem hiding this comment.
I't be nicer to have a span for list and a child span for each call to list_objects
| } | ||
| } | ||
|
|
||
| #[instrument(name = "storage.s3.list", level = "debug", skip(self))] |
There was a problem hiding this comment.
Add prefix for the list call as span attribute?
| } | ||
|
|
||
| let next_list_position = response | ||
| .next_continuation_token() |
There was a problem hiding this comment.
| .next_continuation_token() | |
| .next_continuation_token |
You can take ownership of the continuation token avoid the copy below token.to_string().
| let mut list_request = s3_client | ||
| .list_objects_v2() | ||
| .bucket(&bucket) | ||
| .prefix(&key_prefix); |
There was a problem hiding this comment.
into_paginator exists on this builder, why not use that directly?
|
|
||
| /// Metadata for an object listed from storage. | ||
| #[derive(Clone, Debug, Eq, PartialEq)] | ||
| pub struct ListObjectMetadata { |
There was a problem hiding this comment.
| pub struct ListObjectMetadata { | |
| pub struct ObjectMetadata { |
nit: I think that not so tied to the list API that we can simply can this ObjectMetadata.
| /// Object path relative to this storage root. | ||
| pub path: PathBuf, | ||
| /// Object size in bytes. | ||
| pub size_bytes: u64, |
2 participants
Summary
This PR extends the
Storagetrait by adding alistmethod that allows to receive a stream of objects metadata present in a path.The method returns
ListObjectsStream, a boxed stream ofStorageResult<Vec<ListObjectMetadata>>, where each item is one batch of listed objects.Streaming means callers can start processing results before the whole listing completes, which matters for prefixes holding a large number of objects. Each
ListObjectMetadatacarries:Storageoperation),Errors are reported through the stream rather than aborting it upfront, so a failure on the third page surfaces after the first two pages have already been delivered to the caller.
S3 storage
The S3 implementation uses the
ListObjectsV2API, which returns results in paginated responses. It automatically follows continuation tokens and emits each page as a batch in the returned stream, keeping pagination details hidden from the caller.Page-level failures are surfaced through the stream rather than aborting the listing upfront. A few implementation notes:
ListObjectsV2ErrorimplementsAwsRetryable, so listing gets the same retry classification as the other S3 calls.How was this PR tested?
cargo test -p quickwit-storage test_s3_compatible_storage_list_streams_all_pages --lib— covers multi-page listing against a mocked S3 client, asserting that both pages are streamed, that the continuation token is sent on the second request, and that metadata is converted correctly.cargo check -p quickwit-storage --all-featurescargo clippy -p quickwit-storage --all-features --tests -- -D warningscargo +nightly fmt -p quickwit-storage -- --check