Fetch multiple pieces in the piece fetcher

shared/subspace-data-retrieval/src/piece_fetcher.rs

@@ -17,8 +17,7 @@
 
 use crate::object_fetcher::Error;
 use crate::piece_getter::ObjectPieceGetter;
-use futures::stream::FuturesOrdered;
-use futures::TryStreamExt;
+use futures::{StreamExt, TryStreamExt};
 use subspace_core_primitives::pieces::{Piece, PieceIndex};
 use tracing::{debug, trace};
 
@@ -42,40 +41,18 @@ where
     );
 
     // TODO:
-    // - consider using a semaphore to limit the number of concurrent requests, like
-    //   download_segment_pieces()
     // - if we're close to the number of pieces in a segment, use segment downloading and piece
     //   reconstruction instead
     // Currently most objects are limited to 4 pieces, so this isn't needed yet.
-    let received_pieces = piece_indexes
-        .iter()
-        .map(|piece_index| async move {
-            match piece_getter.get_piece(*piece_index).await {
-                Ok(Some(piece)) => {
-                    trace!(?piece_index, "Piece request succeeded",);
-                    Ok(piece)
-                }
-                Ok(None) => {
-                    trace!(?piece_index, "Piece not found");
-                    Err(Error::PieceNotFound {
-                        piece_index: *piece_index,
-                    }
-                    .into())
-                }
-                Err(error) => {
-                    trace!(
-                        %error,
-                        ?piece_index,
-                        "Piece request caused an error",
-                    );
-                    Err(error)
-                }
-            }
-        })
-        .collect::<FuturesOrdered<_>>();
+    let received_pieces = piece_getter
+        .get_pieces(piece_indexes.iter().copied())
+        .await?;
 
     // We want exact pieces, so any errors are fatal.
-    let received_pieces: Vec<Piece> = received_pieces.try_collect().await?;
+    let received_pieces: Vec<Piece> = received_pieces
+        .map(|(piece_index, maybe_piece)| maybe_piece?.ok_or(Error::PieceNotFound { piece_index }))
+        .try_collect()
+        .await?;
 
     trace!(
         count = piece_indexes.len(),