model: store x-amz-checksum-type (full_object | composite)

src/api/s3/copy.rs

@@ -78,8 +78,23 @@ pub async fn handle_copy(
 		},
 	)?;
 
+	let was_multipart = source_version_meta.etag.contains('-'); // HACK
+
 	// Extract source checksum info before source_object_meta_inner is consumed
 	let source_checksum = source_object_meta_inner.checksum;
+	let source_checksum_type = match (source_object_meta_inner.checksum_type, source_checksum) {
+		(Some(ct), _) => Some(ct),
+		(None, Some(_)) => {
+			// Migrated object from garage v1.x or older
+			// determine checksum type depending if this is a multipart upload or not
+			if was_multipart {
+				Some(ChecksumType::Composite)
+			} else {
+				Some(ChecksumType::FullObject)
+			}
+		}
+		(None, None) => None,
+	};
 	let source_checksum_algorithm = source_checksum.map(|x| x.algorithm());
 
 	// If source object has a checksum, the destination object must as well.
@@ -88,7 +103,6 @@ pub async fn handle_copy(
 	let checksum_algorithm = checksum_algorithm.or(source_checksum_algorithm);
 
 	// Determine metadata of destination object
-	let was_multipart = source_version_meta.etag.contains('-');
 	let dest_object_meta = ObjectVersionMetaInner {
 		headers: match req.headers().get("x-amz-metadata-directive") {
 			Some(v) if v == hyper::header::HeaderValue::from_static("REPLACE") => {
@@ -97,6 +111,7 @@ pub async fn handle_copy(
 			_ => source_object_meta_inner.into_owned().headers,
 		},
 		checksum: source_checksum,
+		checksum_type: source_checksum_type,
 	};
 
 	// Do actual object copying
@@ -144,8 +159,12 @@ pub async fn handle_copy(
 		} else {
 			ChecksumMode::Verify(&expected_checksum)
 		};
-		// If source and dest encryption use different keys,
-		// we must decrypt content and re-encrypt, so rewrite all data blocks.
+		// For multipart uploads that had a composite checksum, set checksum type
+		// to full object as it will be recalculated.
+		let dest_object_meta = ObjectVersionMetaInner {
+			checksum_type: checksum_algorithm.map(|_| ChecksumType::FullObject),
+			..dest_object_meta
+		};
 		handle_copy_reencrypt(
 			ctx,
 			dest_key,
@@ -247,6 +266,7 @@ async fn handle_copy_metaonly(
 				state: ObjectVersionState::Uploading {
 					encryption: new_meta.encryption.clone(),
 					checksum_algorithm: None,
+					checksum_type: None,
 					multipart: false,
 				},
 			};

src/api/s3/multipart.rs

@@ -54,6 +54,7 @@ pub async fn handle_create_multipart_upload(
 	let meta = ObjectVersionMetaInner {
 		headers,
 		checksum: None,
+		checksum_type: None,
 	};
 
 	// Determine whether object should be encrypted, and if so the key
@@ -78,6 +79,8 @@ pub async fn handle_create_multipart_upload(
 			multipart: true,
 			encryption: object_encryption,
 			checksum_algorithm,
+			// TODO: add support for full-object checksums
+			checksum_type: checksum_algorithm.map(|_| ChecksumType::Composite),
 		},
 	};
 	let object = Object::new(*bucket_id, key.to_string(), vec![object_version]);
@@ -440,6 +443,8 @@ pub async fn handle_complete_multipart_upload(
 			let new_meta = ObjectVersionMetaInner {
 				headers: meta.into_owned().headers,
 				checksum: checksum_extra,
+				// TODO: add support for full-object checksums
+				checksum_type: checksum_extra.map(|_| ChecksumType::Composite),
 			};
 			encryption.encrypt_meta(new_meta)?
 		}

src/api/s3/post_object.rs

@@ -233,6 +233,7 @@ pub async fn handle_post_object(
 	let meta = ObjectVersionMetaInner {
 		headers,
 		checksum: expected_checksums.extra,
+		checksum_type: expected_checksums.extra.map(|_| ChecksumType::FullObject),
 	};
 
 	let encryption = EncryptionParams::new_from_headers(

src/api/s3/put.rs

@@ -83,6 +83,7 @@ pub async fn handle_put(
 	let meta = ObjectVersionMetaInner {
 		headers,
 		checksum: expected_checksums.extra,
+		checksum_type: expected_checksums.extra.map(|_| ChecksumType::FullObject),
 	};
 
 	// Determine whether object should be encrypted, and if so the key
@@ -243,6 +244,7 @@ pub(crate) async fn save_stream<S: Stream<Item = Result<Bytes, Error>> + Unpin>(
 		state: ObjectVersionState::Uploading {
 			encryption: encryption.encrypt_meta(meta.clone())?,
 			checksum_algorithm: None, // don't care; overwritten later
+			checksum_type: None,
 			multipart: false,
 		},
 	};

src/model/s3/object_table.rs

@@ -259,7 +259,9 @@ mod v010 {
 			compressed: bool,
 			/// Whether the encryption uses an Object Encryption Key derived
 			/// from the master SSE-C key, instead of the master SSE-C key itself.
-			/// This is the case of objects created in Garage v2+
+			/// This is the case of objects created in Garage v2+.
+			/// This field is kept for compatibility with Garage v2.0.0-beta1,
+			/// which did not yet implement the v2 module below.
 			#[serde(default)]
 			use_oek: bool,
 		},
@@ -378,7 +380,221 @@ mod v010 {
 	}
 }
 
-pub use v010::*;
+mod v2 {
+	use garage_util::data::{Hash, Uuid};
+	use garage_util::migrate::Migrate;
+	use serde::{Deserialize, Serialize};
+
+	use super::v010;
+	pub use v010::{ChecksumAlgorithm, ChecksumValue};
+
+	/// An object
+	#[derive(PartialEq, Eq, Clone, Debug, Serialize, Deserialize)]
+	pub struct Object {
+		/// The bucket in which the object is stored, used as partition key
+		pub bucket_id: Uuid,
+
+		/// The key at which the object is stored in its bucket, used as sorting key
+		pub key: String,
+
+		/// The list of currently stored versions of the object
+		pub(super) versions: Vec<ObjectVersion>,
+	}
+
+	/// Information about a version of an object
+	#[derive(PartialEq, Eq, Clone, Debug, Serialize, Deserialize)]
+	pub struct ObjectVersion {
+		/// Id of the version
+		pub uuid: Uuid,
+		/// Timestamp of when the object was created
+		pub timestamp: u64,
+		/// State of the version
+		pub state: ObjectVersionState,
+	}
+
+	/// State of an object version
+	#[derive(PartialEq, Eq, Clone, Debug, Serialize, Deserialize)]
+	pub enum ObjectVersionState {
+		/// The version is being received
+		Uploading {
+			/// Indicates whether this is a multipart upload
+			multipart: bool,
+			/// Checksum algorithm to use
+			checksum_algorithm: Option<ChecksumAlgorithm>,
+			/// Checksum algorithm type (full object or composite)
+			checksum_type: Option<ChecksumType>,
+			/// Encryption params + headers to be included in the final object
+			encryption: ObjectVersionEncryption,
+		},
+		/// The version is fully received
+		Complete(ObjectVersionData),
+		/// The version uploaded containded errors or the upload was explicitly aborted
+		Aborted,
+	}
+
+	/// Data stored in object version
+	#[derive(PartialEq, Eq, PartialOrd, Ord, Clone, Debug, Serialize, Deserialize)]
+	pub enum ObjectVersionData {
+		/// The object was deleted, this Version is a tombstone to mark it as such
+		DeleteMarker,
+		/// The object is short, it's stored inlined.
+		/// It is never compressed. For encrypted objects, it is encrypted using
+		/// AES256-GCM, like the encrypted headers.
+		Inline(ObjectVersionMeta, #[serde(with = "serde_bytes")] Vec<u8>),
+		/// The object is not short, Hash of first block is stored here, next segments hashes are
+		/// stored in the version table
+		FirstBlock(ObjectVersionMeta, Hash),
+	}
+
+	/// Metadata about the object version
+	#[derive(PartialEq, Eq, PartialOrd, Ord, Clone, Debug, Serialize, Deserialize)]
+	pub struct ObjectVersionMeta {
+		/// Size of the object. If object is encrypted/compressed,
+		/// this is always the size of the unencrypted/uncompressed data
+		pub size: u64,
+		/// etag of the object
+		pub etag: String,
+		/// Encryption params + headers (encrypted or plaintext)
+		pub encryption: ObjectVersionEncryption,
+	}
+
+	/// Encryption information + metadata
+	#[derive(PartialEq, Eq, PartialOrd, Ord, Clone, Debug, Serialize, Deserialize)]
+	pub enum ObjectVersionEncryption {
+		SseC {
+			/// Encrypted serialized ObjectVersionInner struct.
+			/// This is never compressed, just encrypted using AES256-GCM.
+			#[serde(with = "serde_bytes")]
+			inner: Vec<u8>,
+			/// Whether data blocks are compressed in addition to being encrypted
+			/// (compression happens before encryption, whereas for non-encrypted
+			/// objects, compression is handled at the level of the block manager)
+			compressed: bool,
+			/// Whether the encryption uses an Object Encryption Key derived
+			/// from the master SSE-C key, instead of the master SSE-C key itself.
+			/// This is the case of objects created in Garage v2+
+			use_oek: bool,
+		},
+		Plaintext {
+			/// Plain-text headers
+			inner: ObjectVersionMetaInner,
+		},
+	}
+
+	/// Vector of headers, as tuples of the format (header name, header value)
+	/// Note: checksum can be Some(_) with checksum_type = None for objects that
+	/// have been migrated from Garage version before v2.0, as the distinction between
+	/// full-object and composite checksums was not implemented yet.
+	#[derive(PartialEq, Eq, PartialOrd, Ord, Clone, Debug, Serialize, Deserialize)]
+	pub struct ObjectVersionMetaInner {
+		pub headers: HeaderList,
+		pub checksum: Option<ChecksumValue>,
+		pub checksum_type: Option<ChecksumType>,
+	}
+
+	pub type HeaderList = Vec<(String, String)>;
+
+	#[derive(PartialEq, Eq, PartialOrd, Ord, Clone, Copy, Debug, Serialize, Deserialize)]
+	pub enum ChecksumType {
+		FullObject,
+		Composite,
+	}
+
+	impl garage_util::migrate::Migrate for Object {
+		const VERSION_MARKER: &'static [u8] = b"G2s3ob";
+
+		type Previous = v010::Object;
+
+		fn migrate(old: v010::Object) -> Object {
+			Object {
+				bucket_id: old.bucket_id,
+				key: old.key,
+				versions: old.versions.into_iter().map(migrate_version).collect(),
+			}
+		}
+	}
+
+	fn migrate_version(old: v010::ObjectVersion) -> ObjectVersion {
+		ObjectVersion {
+			uuid: old.uuid,
+			timestamp: old.timestamp,
+			state: match old.state {
+				v010::ObjectVersionState::Uploading {
+					multipart,
+					checksum_algorithm,
+					encryption,
+				} => ObjectVersionState::Uploading {
+					multipart,
+					checksum_algorithm,
+					checksum_type: Some(match multipart {
+						false => ChecksumType::FullObject,
+						true => ChecksumType::Composite,
+					}),
+					encryption: migrate_encryption(encryption),
+				},
+				v010::ObjectVersionState::Complete(d) => {
+					ObjectVersionState::Complete(migrate_data(d))
+				}
+				v010::ObjectVersionState::Aborted => ObjectVersionState::Aborted,
+			},
+		}
+	}
+
+	fn migrate_data(old: v010::ObjectVersionData) -> ObjectVersionData {
+		match old {
+			v010::ObjectVersionData::DeleteMarker => ObjectVersionData::DeleteMarker,
+			v010::ObjectVersionData::Inline(meta, data) => {
+				ObjectVersionData::Inline(migrate_meta(meta), data)
+			}
+			v010::ObjectVersionData::FirstBlock(meta, fb) => {
+				ObjectVersionData::FirstBlock(migrate_meta(meta), fb)
+			}
+		}
+	}
+
+	fn migrate_meta(old: v010::ObjectVersionMeta) -> ObjectVersionMeta {
+		ObjectVersionMeta {
+			size: old.size,
+			etag: old.etag,
+			encryption: migrate_encryption(old.encryption),
+		}
+	}
+
+	fn migrate_encryption(old: v010::ObjectVersionEncryption) -> ObjectVersionEncryption {
+		match old {
+			v010::ObjectVersionEncryption::SseC {
+				inner,
+				compressed,
+				use_oek,
+			} => ObjectVersionEncryption::SseC {
+				inner,
+				compressed,
+				use_oek,
+			},
+			v010::ObjectVersionEncryption::Plaintext { inner } => {
+				ObjectVersionEncryption::Plaintext {
+					inner: ObjectVersionMetaInner::migrate(inner),
+				}
+			}
+		}
+	}
+
+	impl Migrate for ObjectVersionMetaInner {
+		const VERSION_MARKER: &'static [u8] = b"G2s3om";
+
+		type Previous = v010::ObjectVersionMetaInner;
+
+		fn migrate(old: v010::ObjectVersionMetaInner) -> ObjectVersionMetaInner {
+			ObjectVersionMetaInner {
+				headers: old.headers,
+				checksum: old.checksum,
+				checksum_type: None,
+			}
+		}
+	}
+}
+
+pub use v2::*;
 
 impl Object {
 	/// Initialize an Object struct from parts