Merge pull request 'Dependency upgrades: http, hyper, aws-sdk, smaller deps' (#703) from dep-upgrade-202402 into main

Reviewed-on: https://git.deuxfleurs.fr/Deuxfleurs/garage/pulls/703
author: Alex <alex@adnab.me> 2024-02-07 14:59:40 +0000
committer: Alex <alex@adnab.me> 2024-02-07 14:59:40 +0000
commit: 5d941e0100489fff552b59f0679a2a010403a21c (patch)
tree: 5c5cef9af72d48dd7347922341e43f0013380c60 /src/api/s3/get.rs
parent: feeb076b7f5db7fe6fdbe3d2903fae054cde6219 (diff)
parent: e011941964b1c1e0b90f85014d166d64a83ae8e2 (diff)
download: garage-5d941e0100489fff552b59f0679a2a010403a21c.tar.gz
garage-5d941e0100489fff552b59f0679a2a010403a21c.zip
1 files changed, 67 insertions, 51 deletions
diff --git a/src/api/s3/get.rs b/src/api/s3/get.rs
index 5e682726..f70dad7d 100644
--- a/src/api/s3/get.rs
+++ b/src/api/s3/get.rs
@@ -8,10 +8,11 @@ use http::header::{
 	ACCEPT_RANGES, CONTENT_LENGTH, CONTENT_RANGE, CONTENT_TYPE, ETAG, IF_MODIFIED_SINCE,
 	IF_NONE_MATCH, LAST_MODIFIED, RANGE,
 };
-use hyper::{Body, Request, Response, StatusCode};
+use hyper::{body::Body, Request, Response, StatusCode};
 use tokio::sync::mpsc;
 
-use garage_rpc::rpc_helper::{netapp::stream::ByteStream, OrderTag};
+use garage_block::manager::BlockStream;
+use garage_rpc::rpc_helper::OrderTag;
 use garage_table::EmptyKey;
 use garage_util::data::*;
 use garage_util::error::OkOrMessage;
@@ -20,6 +21,8 @@ use garage_model::garage::Garage;
 use garage_model::s3::object_table::*;
 use garage_model::s3::version_table::*;
 
+use crate::helpers::*;
+use crate::s3::api_server::ResBody;
 use crate::s3::error::*;
 
 const X_AMZ_MP_PARTS_COUNT: &str = "x-amz-mp-parts-count";
@@ -52,8 +55,8 @@ fn object_headers(
 fn try_answer_cached(
 	version: &ObjectVersion,
 	version_meta: &ObjectVersionMeta,
-	req: &Request<Body>,
-) -> Option<Response<Body>> {
+	req: &Request<impl Body>,
+) -> Option<Response<ResBody>> {
 	// <trinity> It is possible, and is even usually the case, [that both If-None-Match and
 	// If-Modified-Since] are present in a request. In this situation If-None-Match takes
 	// precedence and If-Modified-Since is ignored (as per 6.Precedence from rfc7232). The rational
@@ -80,7 +83,7 @@ fn try_answer_cached(
 		Some(
 			Response::builder()
 				.status(StatusCode::NOT_MODIFIED)
-				.body(Body::empty())
+				.body(empty_body())
 				.unwrap(),
 		)
 	} else {
@@ -91,11 +94,11 @@ fn try_answer_cached(
 /// Handle HEAD request
 pub async fn handle_head(
 	garage: Arc<Garage>,
-	req: &Request<Body>,
+	req: &Request<impl Body>,
 	bucket_id: Uuid,
 	key: &str,
 	part_number: Option<u64>,
-) -> Result<Response<Body>, Error> {
+) -> Result<Response<ResBody>, Error> {
 	let object = garage
 		.object_table
 		.get(&bucket_id, &key.to_string())
@@ -138,7 +141,7 @@ pub async fn handle_head(
 					)
 					.header(X_AMZ_MP_PARTS_COUNT, "1")
 					.status(StatusCode::PARTIAL_CONTENT)
-					.body(Body::empty())?)
+					.body(empty_body())?)
 			}
 			ObjectVersionData::FirstBlock(_, _) => {
 				let version = garage
@@ -163,7 +166,7 @@ pub async fn handle_head(
 					)
 					.header(X_AMZ_MP_PARTS_COUNT, format!("{}", version.n_parts()?))
 					.status(StatusCode::PARTIAL_CONTENT)
-					.body(Body::empty())?)
+					.body(empty_body())?)
 			}
 			_ => unreachable!(),
 		}
@@ -171,18 +174,18 @@ pub async fn handle_head(
 		Ok(object_headers(object_version, version_meta)
 			.header(CONTENT_LENGTH, format!("{}", version_meta.size))
 			.status(StatusCode::OK)
-			.body(Body::empty())?)
+			.body(empty_body())?)
 	}
 }
 
 /// Handle GET request
 pub async fn handle_get(
 	garage: Arc<Garage>,
-	req: &Request<Body>,
+	req: &Request<impl Body>,
 	bucket_id: Uuid,
 	key: &str,
 	part_number: Option<u64>,
-) -> Result<Response<Body>, Error> {
+) -> Result<Response<ResBody>, Error> {
 	let object = garage
 		.object_table
 		.get(&bucket_id, &key.to_string())
@@ -240,11 +243,10 @@ pub async fn handle_get(
 	match &last_v_data {
 		ObjectVersionData::DeleteMarker => unreachable!(),
 		ObjectVersionData::Inline(_, bytes) => {
-			let body: Body = Body::from(bytes.to_vec());
-			Ok(resp_builder.body(body)?)
+			Ok(resp_builder.body(bytes_body(bytes.to_vec().into()))?)
 		}
 		ObjectVersionData::FirstBlock(_, first_block_hash) => {
-			let (tx, rx) = mpsc::channel(2);
+			let (tx, rx) = mpsc::channel::<BlockStream>(2);
 
 			let order_stream = OrderTag::stream();
 			let first_block_hash = *first_block_hash;
@@ -282,20 +284,12 @@ pub async fn handle_get(
 				{
 					Ok(()) => (),
 					Err(e) => {
-						let err = std::io::Error::new(
-							std::io::ErrorKind::Other,
-							format!("Error while getting object data: {}", e),
-						);
-						let _ = tx
-							.send(Box::pin(stream::once(future::ready(Err(err)))))
-							.await;
+						let _ = tx.send(error_stream_item(e)).await;
 					}
 				}
 			});
 
-			let body_stream = tokio_stream::wrappers::ReceiverStream::new(rx).flatten();
-
-			let body = hyper::body::Body::wrap_stream(body_stream);
+			let body = response_body_from_block_stream(rx);
 			Ok(resp_builder.body(body)?)
 		}
 	}
@@ -308,7 +302,7 @@ async fn handle_get_range(
 	version_meta: &ObjectVersionMeta,
 	begin: u64,
 	end: u64,
-) -> Result<Response<Body>, Error> {
+) -> Result<Response<ResBody>, Error> {
 	let resp_builder = object_headers(version, version_meta)
 		.header(CONTENT_LENGTH, format!("{}", end - begin))
 		.header(
@@ -321,7 +315,7 @@ async fn handle_get_range(
 		ObjectVersionData::DeleteMarker => unreachable!(),
 		ObjectVersionData::Inline(_meta, bytes) => {
 			if end as usize <= bytes.len() {
-				let body: Body = Body::from(bytes[begin as usize..end as usize].to_vec());
+				let body = bytes_body(bytes[begin as usize..end as usize].to_vec().into());
 				Ok(resp_builder.body(body)?)
 			} else {
 				Err(Error::internal_error(
@@ -348,7 +342,7 @@ async fn handle_get_part(
 	version_data: &ObjectVersionData,
 	version_meta: &ObjectVersionMeta,
 	part_number: u64,
-) -> Result<Response<Body>, Error> {
+) -> Result<Response<ResBody>, Error> {
 	let resp_builder =
 		object_headers(object_version, version_meta).status(StatusCode::PARTIAL_CONTENT);
 
@@ -364,7 +358,7 @@ async fn handle_get_part(
 					format!("bytes {}-{}/{}", 0, bytes.len() - 1, bytes.len()),
 				)
 				.header(X_AMZ_MP_PARTS_COUNT, "1")
-				.body(Body::from(bytes.to_vec()))?)
+				.body(bytes_body(bytes.to_vec().into()))?)
 		}
 		ObjectVersionData::FirstBlock(_, _) => {
 			let version = garage
@@ -392,7 +386,7 @@ async fn handle_get_part(
 }
 
 fn parse_range_header(
-	req: &Request<Body>,
+	req: &Request<impl Body>,
 	total_size: u64,
 ) -> Result<Option<http_range::HttpRange>, Error> {
 	let range = match req.headers().get(RANGE) {
@@ -434,7 +428,7 @@ fn body_from_blocks_range(
 	all_blocks: &[(VersionBlockKey, VersionBlock)],
 	begin: u64,
 	end: u64,
-) -> Body {
+) -> ResBody {
 	// We will store here the list of blocks that have an intersection with the requested
 	// range, as well as their "true offset", which is their actual offset in the complete
 	// file (whereas block.offset designates the offset of the block WITHIN THE PART
@@ -456,17 +450,17 @@ fn body_from_blocks_range(
 	}
 
 	let order_stream = OrderTag::stream();
-	let body_stream = futures::stream::iter(blocks)
-		.enumerate()
-		.map(move |(i, (block, block_offset))| {
+	let (tx, rx) = mpsc::channel::<BlockStream>(2);
+
+	tokio::spawn(async move {
+		match async {
 			let garage = garage.clone();
-			async move {
-				garage
+			for (i, (block, block_offset)) in blocks.iter().enumerate() {
+				let block_stream = garage
 					.block_manager
 					.rpc_get_block_streaming(&block.hash, Some(order_stream.order(i as u64)))
-					.await
-					.unwrap_or_else(|e| error_stream(i, e))
-					.scan(block_offset, move |chunk_offset, chunk| {
+					.await?
+					.scan(*block_offset, move |chunk_offset, chunk| {
 						let r = match chunk {
 							Ok(chunk_bytes) => {
 								let chunk_len = chunk_bytes.len() as u64;
@@ -502,20 +496,42 @@ fn body_from_blocks_range(
 						};
 						futures::future::ready(r)
 					})
-					.filter_map(futures::future::ready)
+					.filter_map(futures::future::ready);
+
+				let block_stream: BlockStream = Box::pin(block_stream);
+				tx.send(Box::pin(block_stream))
+					.await
+					.ok_or_message("channel closed")?;
 			}
-		})
-		.buffered(2)
-		.flatten();
 
-	hyper::body::Body::wrap_stream(body_stream)
+			Ok::<(), Error>(())
+		}
+		.await
+		{
+			Ok(()) => (),
+			Err(e) => {
+				let _ = tx.send(error_stream_item(e)).await;
+			}
+		}
+	});
+
+	response_body_from_block_stream(rx)
+}
+
+fn response_body_from_block_stream(rx: mpsc::Receiver<BlockStream>) -> ResBody {
+	let body_stream = tokio_stream::wrappers::ReceiverStream::new(rx)
+		.flatten()
+		.map(|x| {
+			x.map(hyper::body::Frame::data)
+				.map_err(|e| Error::from(garage_util::error::Error::from(e)))
+		});
+	ResBody::new(http_body_util::StreamBody::new(body_stream))
 }
 
-fn error_stream(i: usize, e: garage_util::error::Error) -> ByteStream {
-	Box::pin(futures::stream::once(async move {
-		Err(std::io::Error::new(
-			std::io::ErrorKind::Other,
-			format!("Could not get block {}: {}", i, e),
-		))
-	}))
+fn error_stream_item<E: std::fmt::Display>(e: E) -> BlockStream {
+	let err = std::io::Error::new(
+		std::io::ErrorKind::Other,
+		format!("Error while getting object data: {}", e),
+	);
+	Box::pin(stream::once(future::ready(Err(err))))
 }
author	Alex <alex@adnab.me>	2024-02-07 14:59:40 +0000
committer	Alex <alex@adnab.me>	2024-02-07 14:59:40 +0000
commit	5d941e0100489fff552b59f0679a2a010403a21c (patch)
tree	5c5cef9af72d48dd7347922341e43f0013380c60 /src/api/s3/get.rs
parent	feeb076b7f5db7fe6fdbe3d2903fae054cde6219 (diff)
parent	e011941964b1c1e0b90f85014d166d64a83ae8e2 (diff)
download	garage-5d941e0100489fff552b59f0679a2a010403a21c.tar.gz garage-5d941e0100489fff552b59f0679a2a010403a21c.zip