Merge pull request 'Ensure increasing version timestamps when writing new object versions' (#543) from increasing-timestamps into main

Reviewed-on: #543
2023-10-24 10:07:16 +00:00 · 2023-10-24 10:07:16 +00:00 · 75d5d08ee1
commit 75d5d08ee1
parent a5e8ffeb63 c82d91c6bc
3 changed files with 58 additions and 42 deletions
--- a/src/api/s3/delete.rs
+++ b/src/api/s3/delete.rs
@ -3,12 +3,12 @@ use std::sync::Arc;
 use hyper::{Body, Request, Response, StatusCode};

 use garage_util::data::*;
-use garage_util::time::*;

 use garage_model::garage::Garage;
 use garage_model::s3::object_table::*;

 use crate::s3::error::*;
+use crate::s3::put::next_timestamp;
 use crate::s3::xml as s3_xml;
 use crate::signature::verify_signed_content;

@ -23,40 +23,36 @@ async fn handle_delete_internal(
 		.await?
 		.ok_or(Error::NoSuchKey)?; // No need to delete

-	let interesting_versions = object.versions().iter().filter(|v| {
-		!matches!(
-			v.state,
-			ObjectVersionState::Aborted
-				| ObjectVersionState::Complete(ObjectVersionData::DeleteMarker)
-		)
-	});
+	let del_timestamp = next_timestamp(Some(&object));
+	let del_uuid = gen_uuid();

-	let mut version_to_delete = None;
-	let mut timestamp = now_msec();
-	for v in interesting_versions {
-		if v.timestamp + 1 > timestamp || version_to_delete.is_none() {
-			version_to_delete = Some(v.uuid);
+	let deleted_version = object
+		.versions()
+		.iter()
+		.rev()
+		.find(|v| !matches!(&v.state, ObjectVersionState::Aborted))
+		.or_else(|| object.versions().iter().rev().next());
+	let deleted_version = match deleted_version {
+		Some(dv) => dv.uuid,
+		None => {
+			warn!("Object has no versions: {:?}", object);
+			Uuid::from([0u8; 32])
 		}
-		timestamp = std::cmp::max(timestamp, v.timestamp + 1);
-	}
-
-	let deleted_version = version_to_delete.ok_or(Error::NoSuchKey)?;
-
-	let version_uuid = gen_uuid();
+	};

 	let object = Object::new(
 		bucket_id,
 		key.into(),
 		vec![ObjectVersion {
-			uuid: version_uuid,
-			timestamp,
+			uuid: del_uuid,
+			timestamp: del_timestamp,
 			state: ObjectVersionState::Complete(ObjectVersionData::DeleteMarker),
 		}],
 	);

 	garage.object_table.insert(&object).await?;

-	Ok((deleted_version, version_uuid))
+	Ok((deleted_version, del_uuid))
 }

 pub async fn handle_delete(
--- a/src/api/s3/multipart.rs
+++ b/src/api/s3/multipart.rs
@ -9,7 +9,6 @@ use md5::{Digest as Md5Digest, Md5};
 use garage_table::*;
 use garage_util::async_hash::*;
 use garage_util::data::*;
-use garage_util::time::*;

 use garage_model::bucket_table::Bucket;
 use garage_model::garage::Garage;
@ -30,10 +29,13 @@ pub async fn handle_create_multipart_upload(
 	req: &Request<Body>,
 	bucket_name: &str,
 	bucket_id: Uuid,
-	key: &str,
+	key: &String,
 ) -> Result<Response<Body>, Error> {
+	let existing_object = garage.object_table.get(&bucket_id, &key).await?;
+
 	let upload_id = gen_uuid();
-	let timestamp = now_msec();
+	let timestamp = next_timestamp(existing_object.as_ref());
+
 	let headers = get_headers(req.headers())?;

 	// Create object in object table
@ -233,7 +235,8 @@ pub async fn handle_complete_multipart_upload(

 	// Get object and multipart upload
 	let key = key.to_string();
-	let (_, mut object_version, mpu) = get_upload(&garage, &bucket.id, &key, &upload_id).await?;
+	let (object, mut object_version, mpu) =
+		get_upload(&garage, &bucket.id, &key, &upload_id).await?;

 	if mpu.parts.is_empty() {
 		return Err(Error::bad_request("No data was uploaded"));
@ -331,7 +334,7 @@ pub async fn handle_complete_multipart_upload(
 	// Calculate total size of final object
 	let total_size = parts.iter().map(|x| x.size.unwrap()).sum();

-	if let Err(e) = check_quotas(&garage, bucket, &key, total_size).await {
+	if let Err(e) = check_quotas(&garage, bucket, total_size, Some(&object)).await {
 		object_version.state = ObjectVersionState::Aborted;
 		let final_object = Object::new(bucket.id, key.clone(), vec![object_version]);
 		garage.object_table.insert(&final_object).await?;
--- a/src/api/s3/put.rs
+++ b/src/api/s3/put.rs
@ -3,6 +3,7 @@ use std::sync::Arc;

 use base64::prelude::*;
 use futures::prelude::*;
+use futures::try_join;
 use hyper::body::{Body, Bytes};
 use hyper::header::{HeaderMap, HeaderValue};
 use hyper::{Request, Response};
@ -35,7 +36,7 @@ pub async fn handle_put(
 	garage: Arc<Garage>,
 	req: Request<Body>,
 	bucket: &Bucket,
-	key: &str,
+	key: &String,
 	content_sha256: Option<Hash>,
 ) -> Result<Response<Body>, Error> {
 	// Retrieve interesting headers from request
@ -68,16 +69,24 @@ pub(crate) async fn save_stream<S: Stream<Item = Result<Bytes, Error>> + Unpin>(
 	headers: ObjectVersionHeaders,
 	body: S,
 	bucket: &Bucket,
-	key: &str,
+	key: &String,
 	content_md5: Option<String>,
 	content_sha256: Option<FixedBytes32>,
 ) -> Result<(Uuid, String), Error> {
+	let mut chunker = StreamChunker::new(body, garage.config.block_size);
+	let (first_block_opt, existing_object) = try_join!(
+		chunker.next(),
+		garage
+			.object_table
+			.get(&bucket.id, key)
+			.map_err(Error::from),
+	)?;
+
+	let first_block = first_block_opt.unwrap_or_default();
+
 	// Generate identity of new version
 	let version_uuid = gen_uuid();
-	let version_timestamp = now_msec();
-
-	let mut chunker = StreamChunker::new(body, garage.config.block_size);
-	let first_block = chunker.next().await?.unwrap_or_default();
+	let version_timestamp = next_timestamp(existing_object.as_ref());

 	// If body is small enough, store it directly in the object table
 	// as "inline data". We can then return immediately.
@ -97,7 +106,7 @@ pub(crate) async fn save_stream<S: Stream<Item = Result<Bytes, Error>> + Unpin>(
 			content_sha256,
 		)?;

-		check_quotas(&garage, bucket, key, size).await?;
+		check_quotas(&garage, bucket, size, existing_object.as_ref()).await?;

 		let object_version = ObjectVersion {
 			uuid: version_uuid,
@ -176,7 +185,7 @@ pub(crate) async fn save_stream<S: Stream<Item = Result<Bytes, Error>> + Unpin>(
 		content_sha256,
 	)?;

-	check_quotas(&garage, bucket, key, total_size).await?;
+	check_quotas(&garage, bucket, total_size, existing_object.as_ref()).await?;

 	// Save final object state, marked as Complete
 	let md5sum_hex = hex::encode(data_md5sum);
@ -229,19 +238,19 @@ pub(crate) fn ensure_checksum_matches(
 pub(crate) async fn check_quotas(
 	garage: &Arc<Garage>,
 	bucket: &Bucket,
-	key: &str,
 	size: u64,
+	prev_object: Option<&Object>,
 ) -> Result<(), Error> {
 	let quotas = bucket.state.as_option().unwrap().quotas.get();
 	if quotas.max_objects.is_none() && quotas.max_size.is_none() {
 		return Ok(());
 	};

-	let key = key.to_string();
-	let (prev_object, counters) = futures::try_join!(
-		garage.object_table.get(&bucket.id, &key),
-		garage.object_counter_table.table.get(&bucket.id, &EmptyKey),
-	)?;
+	let counters = garage
+		.object_counter_table
+		.table
+		.get(&bucket.id, &EmptyKey)
+		.await?;

 	let counters = counters
 		.map(|x| x.filtered_values(&garage.system.ring.borrow()))
@ -275,7 +284,7 @@ pub(crate) async fn check_quotas(
 		if cnt_size_diff > 0 && current_size + cnt_size_diff > ms as i64 {
 			return Err(Error::forbidden(format!(
 				"Bucket size quota is reached, maximum total size of objects for this bucket: {}. The bucket is already {} bytes, and this object would add {} bytes.",
-				ms, current_size, size
+				ms, current_size, cnt_size_diff
 			)));
 		}
 	}
@ -519,3 +528,11 @@ pub(crate) fn get_headers(headers: &HeaderMap<HeaderValue>) -> Result<ObjectVers
 		other,
 	})
 }
+
+pub(crate) fn next_timestamp(existing_object: Option<&Object>) -> u64 {
+	existing_object
+		.as_ref()
+		.and_then(|obj| obj.versions().iter().map(|v| v.timestamp).max())
+		.map(|t| std::cmp::max(t + 1, now_msec()))
+		.unwrap_or_else(now_msec)
+}