1 files changed, 31 insertions, 12 deletions

diff --git a/src/api/s3/get.rs b/src/api/s3/get.rs
index 7fa1a177..dfc284fe 100644
--- a/src/api/s3/get.rs
+++ b/src/api/s3/get.rs
@@ -10,6 +10,7 @@ use http::header::{
 use hyper::body::Bytes;
 use hyper::{Body, Request, Response, StatusCode};
 
+use garage_rpc::rpc_helper::OrderTag;
 use garage_table::EmptyKey;
 use garage_util::data::*;
 
@@ -242,10 +243,15 @@ pub async fn handle_get(
 			Ok(resp_builder.body(body)?)
 		}
 		ObjectVersionData::FirstBlock(_, first_block_hash) => {
-			let read_first_block = garage.block_manager.rpc_get_block(first_block_hash);
+			let order_stream = OrderTag::stream();
+
+			let read_first_block = garage
+				.block_manager
+				.rpc_get_block_streaming(first_block_hash, Some(order_stream.order(0)));
 			let get_next_blocks = garage.version_table.get(&last_v.uuid, &EmptyKey);
 
-			let (first_block, version) = futures::try_join!(read_first_block, get_next_blocks)?;
+			let (first_block_stream, version) =
+				futures::try_join!(read_first_block, get_next_blocks)?;
 			let version = version.ok_or(Error::NoSuchKey)?;
 
 			let mut blocks = version
@@ -254,24 +260,33 @@ pub async fn handle_get(
 				.iter()
 				.map(|(_, vb)| (vb.hash, None))
 				.collect::<Vec<_>>();
-			blocks[0].1 = Some(first_block);
+			blocks[0].1 = Some(first_block_stream);
 
 			let body_stream = futures::stream::iter(blocks)
-				.map(move |(hash, data_opt)| {
+				.enumerate()
+				.map(move |(i, (hash, stream_opt))| {
 					let garage = garage.clone();
 					async move {
-						if let Some(data) = data_opt {
-							Ok(Bytes::from(data))
+						if let Some(stream) = stream_opt {
+							stream
 						} else {
 							garage
 								.block_manager
-								.rpc_get_block(&hash)
+								.rpc_get_block_streaming(&hash, Some(order_stream.order(i as u64)))
 								.await
-								.map(Bytes::from)
+								.unwrap_or_else(|_| {
+									Box::pin(futures::stream::once(async move {
+										Err(std::io::Error::new(
+											std::io::ErrorKind::Other,
+											"Could not get next block",
+										))
+									}))
+								})
 						}
 					}
 				})
-				.buffered(2);
+				.buffered(2)
+				.flatten();
 
 			let body = hyper::body::Body::wrap_stream(body_stream);
 			Ok(resp_builder.body(body)?)
@@ -434,12 +449,16 @@ fn body_from_blocks_range(
 		true_offset += b.size;
 	}
 
+	let order_stream = OrderTag::stream();
 	let body_stream = futures::stream::iter(blocks)
-		.map(move |(block, true_offset)| {
+		.enumerate()
+		.map(move |(i, (block, true_offset))| {
 			let garage = garage.clone();
 			async move {
-				let data = garage.block_manager.rpc_get_block(&block.hash).await?;
-				let data = Bytes::from(data);
+				let data = garage
+					.block_manager
+					.rpc_get_block(&block.hash, Some(order_stream.order(i as u64)))
+					.await?;
 				let start_in_block = if true_offset > begin {
 					0
 				} else {