remote_storage: add list_streaming API call

[arpad-m]

This adds the ability to list many prefixes in a streaming fashion to the RemoteStorage crate as well as GenericRemoteStorage.

• The list function of the RemoteStorage trait is implemented by default in terms of list_streaming.
• For the production users (S3, Azure), list_streaming is implemented and the default list implementation is used.
• For LocalFs, we keep the list implementation and make list_streaming call it.

The list_streaming function is implemented for both S3 and Azure.

A TODO for later is retries, which the scrubber currently has while the list_streaming implementations lack them.

part of #8457 and #7547

[github-actions]

3126 tests run: 3005 passed, 0 failed, 121 skipped (full report)

---

Code coverage* (full report)

• functions: 32.6% (7005 of 21502 functions)
• lines: 49.9% (55167 of 110530 lines)

* collected from Rust tests only

---

_{The comment gets automatically updated with the latest test results
73741ec at 2024-07-23T18:03:37.197Z :recycle:}

[skyzh]

LGTM but I'm not a big fan of async_stream::stream. fmt + clippy cannot check the async stream body :( I used https://crates.io/crates/futures-async-stream in the past but it's only on Rust nightly.

[arpad-m]

A TODO for later is retries, which the scrubber currently has while the list_streaming implementations lack them.

Taking a look at this, it seems that the azure SDK resets the internal state to Done upon any error obtained. The stream will always be returning None once the state is Done.

The other relevant file to look at is list_blobs.rs.

The good news is that I see a path towards being able to do retries on Azure in a not too ugly way:

• ListBlobsResponse impls the Continuable trait which is public so one can always obtain the latest NextMarker.
• The NextMarker can be fed into the builder. Builders are cloneable, and the only changed state is that marker.
• So if one keeps a record of the latest NextMarker, and keeps the ListBlobsBuilder around, then one can avoid the state management of the Pageable.