Merge branch 'main' into next-v2

src/api/s3/get.rs

@@ -19,12 +19,13 @@ use garage_net::stream::ByteStream;
 use garage_rpc::rpc_helper::OrderTag;
 use garage_table::EmptyKey;
 use garage_util::data::*;
-use garage_util::error::OkOrMessage;
+use garage_util::error::{Error as UtilError, OkOrMessage};
 
 use garage_model::garage::Garage;
 use garage_model::s3::object_table::*;
 use garage_model::s3::version_table::*;
 
+use garage_api_common::common_error::CommonError;
 use garage_api_common::helpers::*;
 use garage_api_common::signature::checksum::{add_checksum_response_headers, X_AMZ_CHECKSUM_MODE};
 
@@ -219,6 +220,7 @@ pub async fn handle_head_without_ctx(
 					.get(&object_version.uuid, &EmptyKey)
 					.await?
 					.ok_or(Error::NoSuchKey)?;
+				check_version_not_deleted(&version)?;
 
 				let (part_offset, part_end) =
 					calculate_part_bounds(&version, pn).ok_or(Error::InvalidPart)?;
@@ -373,6 +375,21 @@ pub async fn handle_get_without_ctx(
 	}
 }
 
+pub(crate) fn check_version_not_deleted(version: &Version) -> Result<(), Error> {
+	if version.deleted.get() {
+		// the version was deleted between when the object_table was consulted
+		// and now, this could mean the object was deleted, or overriden.
+		// Rather than say the key doesn't exist, return a transient error
+		// to signal the client to try again.
+		return Err(CommonError::InternalError(UtilError::Message(
+			"conflict/inconsistency between object and version state, version is deleted"
+				.to_string(),
+		))
+		.into());
+	}
+	Ok(())
+}
+
 async fn handle_get_full(
 	garage: Arc<Garage>,
 	version: &ObjectVersion,
@@ -439,6 +456,7 @@ pub fn full_object_byte_stream(
 						.ok_or_message("channel closed")?;
 
 					let version = version_fut.await.unwrap()?.ok_or(Error::NoSuchKey)?;
+					check_version_not_deleted(&version)?;
 					for (i, (_, vb)) in version.blocks.items().iter().enumerate().skip(1) {
 						let stream_block_i = encryption
 							.get_block(&garage, &vb.hash, Some(order_stream.order(i as u64)))
@@ -454,6 +472,14 @@ pub fn full_object_byte_stream(
 				{
 					Ok(()) => (),
 					Err(e) => {
+						// TODO i think this is a bad idea, we should log
+						// an error and stop there. If the error happens to
+						// be exactly the size of what hasn't been streamed
+						// yet, the client will see the request as a
+						// success
+						// instead truncating the output notify the client
+						// something happened with their download, so that
+						// they can retry it
 						let _ = tx.send(error_stream_item(e)).await;
 					}
 				}
@@ -505,7 +531,7 @@ async fn handle_get_range(
 				.get(&version.uuid, &EmptyKey)
 				.await?
 				.ok_or(Error::NoSuchKey)?;
-
+			check_version_not_deleted(&version)?;
 			let body =
 				body_from_blocks_range(garage, encryption, version.blocks.items(), begin, end);
 			Ok(resp_builder.body(body)?)
@@ -556,6 +582,8 @@ async fn handle_get_part(
 				.await?
 				.ok_or(Error::NoSuchKey)?;
 
+			check_version_not_deleted(&version)?;
+
 			let (begin, end) =
 				calculate_part_bounds(&version, part_number).ok_or(Error::InvalidPart)?;