Merge pull request 'Garage v0.9' (#473) from next into main

Reviewed-on: https://git.deuxfleurs.fr/Deuxfleurs/garage/pulls/473
author: Alex <alex@adnab.me> 2023-10-10 13:28:28 +0000
committer: Alex <alex@adnab.me> 2023-10-10 13:28:28 +0000
commit: e6df7089a19d06696ac2178a4dae7ac9f27fd55e (patch)
tree: e1f7120d582f67fd104dc8e746d883befffb4daa /src/api
parent: e75fe2157d0973eb4320bbc87be5e54a1cb91a63 (diff)
parent: 952c9570c494468643353ee1ae9052b510353665 (diff)
download: garage-e6df7089a19d06696ac2178a4dae7ac9f27fd55e.tar.gz
garage-e6df7089a19d06696ac2178a4dae7ac9f27fd55e.zip
20 files changed, 1619 insertions, 777 deletions
diff --git a/src/api/Cargo.toml b/src/api/Cargo.toml
index cb9e2e55..e8cbc1c8 100644
--- a/src/api/Cargo.toml
+++ b/src/api/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "garage_api"
-version = "0.8.4"
+version = "0.9.0"
 authors = ["Alex Auvolat <alex@adnab.me>"]
 edition = "2018"
 license = "AGPL-3.0"
diff --git a/src/api/admin/api_server.rs b/src/api/admin/api_server.rs
index 6f1e44e5..4779f924 100644
--- a/src/api/admin/api_server.rs
+++ b/src/api/admin/api_server.rs
@@ -25,7 +25,8 @@ use crate::admin::bucket::*;
 use crate::admin::cluster::*;
 use crate::admin::error::*;
 use crate::admin::key::*;
-use crate::admin::router::{Authorization, Endpoint};
+use crate::admin::router_v0;
+use crate::admin::router_v1::{Authorization, Endpoint};
 use crate::helpers::host_to_bucket;
 
 pub struct AdminApiServer {
@@ -229,7 +230,12 @@ impl ApiHandler for AdminApiServer {
 	type Error = Error;
 
 	fn parse_endpoint(&self, req: &Request<Body>) -> Result<Endpoint, Error> {
-		Endpoint::from_request(req)
+		if req.uri().path().starts_with("/v0/") {
+			let endpoint_v0 = router_v0::Endpoint::from_request(req)?;
+			Endpoint::from_v0(endpoint_v0)
+		} else {
+			Endpoint::from_request(req)
+		}
 	}
 
 	async fn handle(
@@ -276,8 +282,13 @@ impl ApiHandler for AdminApiServer {
 			Endpoint::RevertClusterLayout => handle_revert_cluster_layout(&self.garage, req).await,
 			// Keys
 			Endpoint::ListKeys => handle_list_keys(&self.garage).await,
-			Endpoint::GetKeyInfo { id, search } => {
-				handle_get_key_info(&self.garage, id, search).await
+			Endpoint::GetKeyInfo {
+				id,
+				search,
+				show_secret_key,
+			} => {
+				let show_secret_key = show_secret_key.map(|x| x == "true").unwrap_or(false);
+				handle_get_key_info(&self.garage, id, search, show_secret_key).await
 			}
 			Endpoint::CreateKey => handle_create_key(&self.garage, req).await,
 			Endpoint::ImportKey => handle_import_key(&self.garage, req).await,
diff --git a/src/api/admin/bucket.rs b/src/api/admin/bucket.rs
index f0a4a9e7..17f46c30 100644
--- a/src/api/admin/bucket.rs
+++ b/src/api/admin/bucket.rs
@@ -14,6 +14,7 @@ use garage_model::bucket_alias_table::*;
 use garage_model::bucket_table::*;
 use garage_model::garage::Garage;
 use garage_model::permission::*;
+use garage_model::s3::mpu_table;
 use garage_model::s3::object_table::*;
 
 use crate::admin::error::*;
@@ -124,6 +125,14 @@ async fn bucket_info_results(
 		.map(|x| x.filtered_values(&garage.system.ring.borrow()))
 		.unwrap_or_default();
 
+	let mpu_counters = garage
+		.mpu_counter_table
+		.table
+		.get(&bucket_id, &EmptyKey)
+		.await?
+		.map(|x| x.filtered_values(&garage.system.ring.borrow()))
+		.unwrap_or_default();
+
 	let mut relevant_keys = HashMap::new();
 	for (k, _) in bucket
 		.state
@@ -208,12 +217,12 @@ async fn bucket_info_results(
 					}
 				})
 				.collect::<Vec<_>>(),
-			objects: counters.get(OBJECTS).cloned().unwrap_or_default(),
-			bytes: counters.get(BYTES).cloned().unwrap_or_default(),
-			unfinished_uploads: counters
-				.get(UNFINISHED_UPLOADS)
-				.cloned()
-				.unwrap_or_default(),
+			objects: *counters.get(OBJECTS).unwrap_or(&0),
+			bytes: *counters.get(BYTES).unwrap_or(&0),
+			unfinished_uploads: *counters.get(UNFINISHED_UPLOADS).unwrap_or(&0),
+			unfinished_multipart_uploads: *mpu_counters.get(mpu_table::UPLOADS).unwrap_or(&0),
+			unfinished_multipart_upload_parts: *mpu_counters.get(mpu_table::PARTS).unwrap_or(&0),
+			unfinished_multipart_upload_bytes: *mpu_counters.get(mpu_table::BYTES).unwrap_or(&0),
 			quotas: ApiBucketQuotas {
 				max_size: quotas.max_size,
 				max_objects: quotas.max_objects,
@@ -235,6 +244,9 @@ struct GetBucketInfoResult {
 	objects: i64,
 	bytes: i64,
 	unfinished_uploads: i64,
+	unfinished_multipart_uploads: i64,
+	unfinished_multipart_upload_parts: i64,
+	unfinished_multipart_upload_bytes: i64,
 	quotas: ApiBucketQuotas,
 }
 
diff --git a/src/api/admin/cluster.rs b/src/api/admin/cluster.rs
index 98bf2b5a..c8107b82 100644
--- a/src/api/admin/cluster.rs
+++ b/src/api/admin/cluster.rs
@@ -1,14 +1,13 @@
-use std::collections::HashMap;
 use std::net::SocketAddr;
 use std::sync::Arc;
 
-use hyper::{Body, Request, Response, StatusCode};
+use hyper::{Body, Request, Response};
 use serde::{Deserialize, Serialize};
 
 use garage_util::crdt::*;
 use garage_util::data::*;
 
-use garage_rpc::layout::*;
+use garage_rpc::layout;
 
 use garage_model::garage::Garage;
 
@@ -26,26 +25,37 @@ pub async fn handle_get_cluster_status(garage: &Arc<Garage>) -> Result<Response<
 			.system
 			.get_known_nodes()
 			.into_iter()
-			.map(|i| {
-				(
-					hex::encode(i.id),
-					KnownNodeResp {
-						addr: i.addr,
-						is_up: i.is_up,
-						last_seen_secs_ago: i.last_seen_secs_ago,
-						hostname: i.status.hostname,
-					},
-				)
+			.map(|i| KnownNodeResp {
+				id: hex::encode(i.id),
+				addr: i.addr,
+				is_up: i.is_up,
+				last_seen_secs_ago: i.last_seen_secs_ago,
+				hostname: i.status.hostname,
 			})
 			.collect(),
-		layout: get_cluster_layout(garage),
+		layout: format_cluster_layout(&garage.system.get_cluster_layout()),
 	};
 
 	Ok(json_ok_response(&res)?)
 }
 
 pub async fn handle_get_cluster_health(garage: &Arc<Garage>) -> Result<Response<Body>, Error> {
+	use garage_rpc::system::ClusterHealthStatus;
 	let health = garage.system.health();
+	let health = ClusterHealth {
+		status: match health.status {
+			ClusterHealthStatus::Healthy => "healthy",
+			ClusterHealthStatus::Degraded => "degraded",
+			ClusterHealthStatus::Unavailable => "unavailable",
+		},
+		known_nodes: health.known_nodes,
+		connected_nodes: health.connected_nodes,
+		storage_nodes: health.storage_nodes,
+		storage_nodes_ok: health.storage_nodes_ok,
+		partitions: health.partitions,
+		partitions_quorum: health.partitions_quorum,
+		partitions_all_ok: health.partitions_all_ok,
+	};
 	Ok(json_ok_response(&health)?)
 }
 
@@ -74,33 +84,68 @@ pub async fn handle_connect_cluster_nodes(
 }
 
 pub async fn handle_get_cluster_layout(garage: &Arc<Garage>) -> Result<Response<Body>, Error> {
-	let res = get_cluster_layout(garage);
+	let res = format_cluster_layout(&garage.system.get_cluster_layout());
 
 	Ok(json_ok_response(&res)?)
 }
 
-fn get_cluster_layout(garage: &Arc<Garage>) -> GetClusterLayoutResponse {
-	let layout = garage.system.get_cluster_layout();
+fn format_cluster_layout(layout: &layout::ClusterLayout) -> GetClusterLayoutResponse {
+	let roles = layout
+		.roles
+		.items()
+		.iter()
+		.filter_map(|(k, _, v)| v.0.clone().map(|x| (k, x)))
+		.map(|(k, v)| NodeRoleResp {
+			id: hex::encode(k),
+			zone: v.zone.clone(),
+			capacity: v.capacity,
+			tags: v.tags.clone(),
+		})
+		.collect::<Vec<_>>();
+
+	let staged_role_changes = layout
+		.staging_roles
+		.items()
+		.iter()
+		.filter(|(k, _, v)| layout.roles.get(k) != Some(v))
+		.map(|(k, _, v)| match &v.0 {
+			None => NodeRoleChange {
+				id: hex::encode(k),
+				action: NodeRoleChangeEnum::Remove { remove: true },
+			},
+			Some(r) => NodeRoleChange {
+				id: hex::encode(k),
+				action: NodeRoleChangeEnum::Update {
+					zone: r.zone.clone(),
+					capacity: r.capacity,
+					tags: r.tags.clone(),
+				},
+			},
+		})
+		.collect::<Vec<_>>();
 
 	GetClusterLayoutResponse {
 		version: layout.version,
-		roles: layout
-			.roles
-			.items()
-			.iter()
-			.filter(|(_, _, v)| v.0.is_some())
-			.map(|(k, _, v)| (hex::encode(k), v.0.clone()))
-			.collect(),
-		staged_role_changes: layout
-			.staging
-			.items()
-			.iter()
-			.filter(|(k, _, v)| layout.roles.get(k) != Some(v))
-			.map(|(k, _, v)| (hex::encode(k), v.0.clone()))
-			.collect(),
+		roles,
+		staged_role_changes,
 	}
 }
 
+// ----
+
+#[derive(Debug, Clone, Copy, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ClusterHealth {
+	status: &'static str,
+	known_nodes: usize,
+	connected_nodes: usize,
+	storage_nodes: usize,
+	storage_nodes_ok: usize,
+	partitions: usize,
+	partitions_quorum: usize,
+	partitions_all_ok: usize,
+}
+
 #[derive(Serialize)]
 #[serde(rename_all = "camelCase")]
 struct GetClusterStatusResponse {
@@ -109,11 +154,19 @@ struct GetClusterStatusResponse {
 	garage_features: Option<&'static [&'static str]>,
 	rust_version: &'static str,
 	db_engine: String,
-	known_nodes: HashMap<String, KnownNodeResp>,
+	known_nodes: Vec<KnownNodeResp>,
+	layout: GetClusterLayoutResponse,
+}
+
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+struct ApplyClusterLayoutResponse {
+	message: Vec<String>,
 	layout: GetClusterLayoutResponse,
 }
 
 #[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
 struct ConnectClusterNodesResponse {
 	success: bool,
 	error: Option<String>,
@@ -123,18 +176,31 @@ struct ConnectClusterNodesResponse {
 #[serde(rename_all = "camelCase")]
 struct GetClusterLayoutResponse {
 	version: u64,
-	roles: HashMap<String, Option<NodeRole>>,
-	staged_role_changes: HashMap<String, Option<NodeRole>>,
+	roles: Vec<NodeRoleResp>,
+	staged_role_changes: Vec<NodeRoleChange>,
 }
 
 #[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+struct NodeRoleResp {
+	id: String,
+	zone: String,
+	capacity: Option<u64>,
+	tags: Vec<String>,
+}
+
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
 struct KnownNodeResp {
+	id: String,
 	addr: SocketAddr,
 	is_up: bool,
 	last_seen_secs_ago: Option<u64>,
 	hostname: String,
 }
 
+// ---- update functions ----
+
 pub async fn handle_update_cluster_layout(
 	garage: &Arc<Garage>,
 	req: Request<Body>,
@@ -144,22 +210,35 @@ pub async fn handle_update_cluster_layout(
 	let mut layout = garage.system.get_cluster_layout();
 
 	let mut roles = layout.roles.clone();
-	roles.merge(&layout.staging);
+	roles.merge(&layout.staging_roles);
 
-	for (node, role) in updates {
-		let node = hex::decode(node).ok_or_bad_request("Invalid node identifier")?;
+	for change in updates {
+		let node = hex::decode(&change.id).ok_or_bad_request("Invalid node identifier")?;
 		let node = Uuid::try_from(&node).ok_or_bad_request("Invalid node identifier")?;
 
+		let new_role = match change.action {
+			NodeRoleChangeEnum::Remove { remove: true } => None,
+			NodeRoleChangeEnum::Update {
+				zone,
+				capacity,
+				tags,
+			} => Some(layout::NodeRole {
+				zone,
+				capacity,
+				tags,
+			}),
+			_ => return Err(Error::bad_request("Invalid layout change")),
+		};
+
 		layout
-			.staging
-			.merge(&roles.update_mutator(node, NodeRoleV(role)));
+			.staging_roles
+			.merge(&roles.update_mutator(node, layout::NodeRoleV(new_role)));
 	}
 
 	garage.system.update_cluster_layout(&layout).await?;
 
-	Ok(Response::builder()
-		.status(StatusCode::NO_CONTENT)
-		.body(Body::empty())?)
+	let res = format_cluster_layout(&layout);
+	Ok(json_ok_response(&res)?)
 }
 
 pub async fn handle_apply_cluster_layout(
@@ -169,12 +248,15 @@ pub async fn handle_apply_cluster_layout(
 	let param = parse_json_body::<ApplyRevertLayoutRequest>(req).await?;
 
 	let layout = garage.system.get_cluster_layout();
-	let layout = layout.apply_staged_changes(Some(param.version))?;
+	let (layout, msg) = layout.apply_staged_changes(Some(param.version))?;
+
 	garage.system.update_cluster_layout(&layout).await?;
 
-	Ok(Response::builder()
-		.status(StatusCode::NO_CONTENT)
-		.body(Body::empty())?)
+	let res = ApplyClusterLayoutResponse {
+		message: msg,
+		layout: format_cluster_layout(&layout),
+	};
+	Ok(json_ok_response(&res)?)
 }
 
 pub async fn handle_revert_cluster_layout(
@@ -187,14 +269,39 @@ pub async fn handle_revert_cluster_layout(
 	let layout = layout.revert_staged_changes(Some(param.version))?;
 	garage.system.update_cluster_layout(&layout).await?;
 
-	Ok(Response::builder()
-		.status(StatusCode::NO_CONTENT)
-		.body(Body::empty())?)
+	let res = format_cluster_layout(&layout);
+	Ok(json_ok_response(&res)?)
 }
 
-type UpdateClusterLayoutRequest = HashMap<String, Option<NodeRole>>;
+// ----
+
+type UpdateClusterLayoutRequest = Vec<NodeRoleChange>;
 
 #[derive(Deserialize)]
+#[serde(rename_all = "camelCase")]
 struct ApplyRevertLayoutRequest {
 	version: u64,
 }
+
+// ----
+
+#[derive(Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct NodeRoleChange {
+	id: String,
+	#[serde(flatten)]
+	action: NodeRoleChangeEnum,
+}
+
+#[derive(Serialize, Deserialize)]
+#[serde(untagged)]
+enum NodeRoleChangeEnum {
+	#[serde(rename_all = "camelCase")]
+	Remove { remove: bool },
+	#[serde(rename_all = "camelCase")]
+	Update {
+		zone: String,
+		capacity: Option<u64>,
+		tags: Vec<String>,
+	},
+}
diff --git a/src/api/admin/key.rs b/src/api/admin/key.rs
index d74ca361..8d1c6890 100644
--- a/src/api/admin/key.rs
+++ b/src/api/admin/key.rs
@@ -10,7 +10,7 @@ use garage_model::garage::Garage;
 use garage_model::key_table::*;
 
 use crate::admin::error::*;
-use crate::helpers::{json_ok_response, parse_json_body};
+use crate::helpers::{is_default, json_ok_response, parse_json_body};
 
 pub async fn handle_list_keys(garage: &Arc<Garage>) -> Result<Response<Body>, Error> {
 	let res = garage
@@ -34,6 +34,7 @@ pub async fn handle_list_keys(garage: &Arc<Garage>) -> Result<Response<Body>, Er
 }
 
 #[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
 struct ListKeyResultItem {
 	id: String,
 	name: String,
@@ -43,6 +44,7 @@ pub async fn handle_get_key_info(
 	garage: &Arc<Garage>,
 	id: Option<String>,
 	search: Option<String>,
+	show_secret_key: bool,
 ) -> Result<Response<Body>, Error> {
 	let key = if let Some(id) = id {
 		garage.key_helper().get_existing_key(&id).await?
@@ -55,7 +57,7 @@ pub async fn handle_get_key_info(
 		unreachable!();
 	};
 
-	key_info_results(garage, key).await
+	key_info_results(garage, key, show_secret_key).await
 }
 
 pub async fn handle_create_key(
@@ -64,15 +66,16 @@ pub async fn handle_create_key(
 ) -> Result<Response<Body>, Error> {
 	let req = parse_json_body::<CreateKeyRequest>(req).await?;
 
-	let key = Key::new(&req.name);
+	let key = Key::new(req.name.as_deref().unwrap_or("Unnamed key"));
 	garage.key_table.insert(&key).await?;
 
-	key_info_results(garage, key).await
+	key_info_results(garage, key, true).await
 }
 
 #[derive(Deserialize)]
+#[serde(rename_all = "camelCase")]
 struct CreateKeyRequest {
-	name: String,
+	name: Option<String>,
 }
 
 pub async fn handle_import_key(
@@ -86,10 +89,15 @@ pub async fn handle_import_key(
 		return Err(Error::KeyAlreadyExists(req.access_key_id.to_string()));
 	}
 
-	let imported_key = Key::import(&req.access_key_id, &req.secret_access_key, &req.name);
+	let imported_key = Key::import(
+		&req.access_key_id,
+		&req.secret_access_key,
+		req.name.as_deref().unwrap_or("Imported key"),
+	)
+	.ok_or_bad_request("Invalid key format")?;
 	garage.key_table.insert(&imported_key).await?;
 
-	key_info_results(garage, imported_key).await
+	key_info_results(garage, imported_key, false).await
 }
 
 #[derive(Deserialize)]
@@ -97,7 +105,7 @@ pub async fn handle_import_key(
 struct ImportKeyRequest {
 	access_key_id: String,
 	secret_access_key: String,
-	name: String,
+	name: Option<String>,
 }
 
 pub async fn handle_update_key(
@@ -127,10 +135,11 @@ pub async fn handle_update_key(
 
 	garage.key_table.insert(&key).await?;
 
-	key_info_results(garage, key).await
+	key_info_results(garage, key, false).await
 }
 
 #[derive(Deserialize)]
+#[serde(rename_all = "camelCase")]
 struct UpdateKeyRequest {
 	name: Option<String>,
 	allow: Option<KeyPerm>,
@@ -149,7 +158,11 @@ pub async fn handle_delete_key(garage: &Arc<Garage>, id: String) -> Result<Respo
 		.body(Body::empty())?)
 }
 
-async fn key_info_results(garage: &Arc<Garage>, key: Key) -> Result<Response<Body>, Error> {
+async fn key_info_results(
+	garage: &Arc<Garage>,
+	key: Key,
+	show_secret: bool,
+) -> Result<Response<Body>, Error> {
 	let mut relevant_buckets = HashMap::new();
 
 	let key_state = key.state.as_option().unwrap();
@@ -178,7 +191,11 @@ async fn key_info_results(garage: &Arc<Garage>, key: Key) -> Result<Response<Bod
 	let res = GetKeyInfoResult {
 		name: key_state.name.get().clone(),
 		access_key_id: key.key_id.clone(),
-		secret_access_key: key_state.secret_key.clone(),
+		secret_access_key: if show_secret {
+			Some(key_state.secret_key.clone())
+		} else {
+			None
+		},
 		permissions: KeyPerm {
 			create_bucket: *key_state.allow_create_bucket.get(),
 		},
@@ -224,7 +241,8 @@ async fn key_info_results(garage: &Arc<Garage>, key: Key) -> Result<Response<Bod
 struct GetKeyInfoResult {
 	name: String,
 	access_key_id: String,
-	secret_access_key: String,
+	#[serde(skip_serializing_if = "is_default")]
+	secret_access_key: Option<String>,
 	permissions: KeyPerm,
 	buckets: Vec<KeyInfoBucketResult>,
 }
@@ -246,6 +264,7 @@ struct KeyInfoBucketResult {
 }
 
 #[derive(Serialize, Deserialize, Default)]
+#[serde(rename_all = "camelCase")]
 pub(crate) struct ApiBucketKeyPerm {
 	#[serde(default)]
 	pub(crate) read: bool,
diff --git a/src/api/admin/mod.rs b/src/api/admin/mod.rs
index c4857c10..43a8c59c 100644
--- a/src/api/admin/mod.rs
+++ b/src/api/admin/mod.rs
@@ -1,6 +1,7 @@
 pub mod api_server;
 mod error;
-mod router;
+mod router_v0;
+mod router_v1;
 
 mod bucket;
 mod cluster;
diff --git a/src/api/admin/router.rs b/src/api/admin/router_v0.rs
index 0225f18b..68676445 100644
--- a/src/api/admin/router.rs
+++ b/src/api/admin/router_v0.rs
@@ -5,12 +5,6 @@ use hyper::{Method, Request};
 use crate::admin::error::*;
 use crate::router_macros::*;
 
-pub enum Authorization {
-	None,
-	MetricsToken,
-	AdminToken,
-}
-
 router_match! {@func
 
 /// List of all Admin API endpoints.
@@ -134,15 +128,6 @@ impl Endpoint {
 
 		Ok(res)
 	}
-	/// Get the kind of authorization which is required to perform the operation.
-	pub fn authorization_type(&self) -> Authorization {
-		match self {
-			Self::Health => Authorization::None,
-			Self::CheckDomain => Authorization::None,
-			Self::Metrics => Authorization::MetricsToken,
-			_ => Authorization::AdminToken,
-		}
-	}
 }
 
 generateQueryParameters! {
diff --git a/src/api/admin/router_v1.rs b/src/api/admin/router_v1.rs
new file mode 100644
index 00000000..cc5ff2ec
--- /dev/null
+++ b/src/api/admin/router_v1.rs
@@ -0,0 +1,235 @@
+use std::borrow::Cow;
+
+use hyper::{Method, Request};
+
+use crate::admin::error::*;
+use crate::admin::router_v0;
+use crate::router_macros::*;
+
+pub enum Authorization {
+	None,
+	MetricsToken,
+	AdminToken,
+}
+
+router_match! {@func
+
+/// List of all Admin API endpoints.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum Endpoint {
+	Options,
+	CheckDomain,
+	Health,
+	Metrics,
+	GetClusterStatus,
+	GetClusterHealth,
+	ConnectClusterNodes,
+	// Layout
+	GetClusterLayout,
+	UpdateClusterLayout,
+	ApplyClusterLayout,
+	RevertClusterLayout,
+	// Keys
+	ListKeys,
+	CreateKey,
+	ImportKey,
+	GetKeyInfo {
+		id: Option<String>,
+		search: Option<String>,
+		show_secret_key: Option<String>,
+	},
+	DeleteKey {
+		id: String,
+	},
+	UpdateKey {
+		id: String,
+	},
+	// Buckets
+	ListBuckets,
+	CreateBucket,
+	GetBucketInfo {
+		id: Option<String>,
+		global_alias: Option<String>,
+	},
+	DeleteBucket {
+		id: String,
+	},
+	UpdateBucket {
+		id: String,
+	},
+	// Bucket-Key Permissions
+	BucketAllowKey,
+	BucketDenyKey,
+	// Bucket aliases
+	GlobalAliasBucket {
+		id: String,
+		alias: String,
+	},
+	GlobalUnaliasBucket {
+		id: String,
+		alias: String,
+	},
+	LocalAliasBucket {
+		id: String,
+		access_key_id: String,
+		alias: String,
+	},
+	LocalUnaliasBucket {
+		id: String,
+		access_key_id: String,
+		alias: String,
+	},
+}}
+
+impl Endpoint {
+	/// Determine which S3 endpoint a request is for using the request, and a bucket which was
+	/// possibly extracted from the Host header.
+	/// Returns Self plus bucket name, if endpoint is not Endpoint::ListBuckets
+	pub fn from_request<T>(req: &Request<T>) -> Result<Self, Error> {
+		let uri = req.uri();
+		let path = uri.path();
+		let query = uri.query();
+
+		let mut query = QueryParameters::from_query(query.unwrap_or_default())?;
+
+		let res = router_match!(@gen_path_parser (req.method(), path, query) [
+			OPTIONS _ => Options,
+			GET "/check" => CheckDomain,
+			GET "/health" => Health,
+			GET "/metrics" => Metrics,
+			GET "/v1/status" => GetClusterStatus,
+			GET "/v1/health" => GetClusterHealth,
+			POST "/v1/connect" => ConnectClusterNodes,
+			// Layout endpoints
+			GET "/v1/layout" => GetClusterLayout,
+			POST "/v1/layout" => UpdateClusterLayout,
+			POST "/v1/layout/apply" => ApplyClusterLayout,
+			POST "/v1/layout/revert" => RevertClusterLayout,
+			// API key endpoints
+			GET "/v1/key" if id => GetKeyInfo (query_opt::id, query_opt::search, query_opt::show_secret_key),
+			GET "/v1/key" if search => GetKeyInfo (query_opt::id, query_opt::search, query_opt::show_secret_key),
+			POST "/v1/key" if id => UpdateKey (query::id),
+			POST "/v1/key" => CreateKey,
+			POST "/v1/key/import" => ImportKey,
+			DELETE "/v1/key" if id => DeleteKey (query::id),
+			GET "/v1/key" => ListKeys,
+			// Bucket endpoints
+			GET "/v1/bucket" if id => GetBucketInfo (query_opt::id, query_opt::global_alias),
+			GET "/v1/bucket" if global_alias => GetBucketInfo (query_opt::id, query_opt::global_alias),
+			GET "/v1/bucket" => ListBuckets,
+			POST "/v1/bucket" => CreateBucket,
+			DELETE "/v1/bucket" if id => DeleteBucket (query::id),
+			PUT "/v1/bucket" if id => UpdateBucket (query::id),
+			// Bucket-key permissions
+			POST "/v1/bucket/allow" => BucketAllowKey,
+			POST "/v1/bucket/deny" => BucketDenyKey,
+			// Bucket aliases
+			PUT "/v1/bucket/alias/global" => GlobalAliasBucket (query::id, query::alias),
+			DELETE "/v1/bucket/alias/global" => GlobalUnaliasBucket (query::id, query::alias),
+			PUT "/v1/bucket/alias/local" => LocalAliasBucket (query::id, query::access_key_id, query::alias),
+			DELETE "/v1/bucket/alias/local" => LocalUnaliasBucket (query::id, query::access_key_id, query::alias),
+		]);
+
+		if let Some(message) = query.nonempty_message() {
+			debug!("Unused query parameter: {}", message)
+		}
+
+		Ok(res)
+	}
+	/// Some endpoints work exactly the same in their v1/ version as they did in their v0/ version.
+	/// For these endpoints, we can convert a v0/ call to its equivalent as if it was made using
+	/// its v1/ URL.
+	pub fn from_v0(v0_endpoint: router_v0::Endpoint) -> Result<Self, Error> {
+		match v0_endpoint {
+			// Cluster endpoints
+			router_v0::Endpoint::ConnectClusterNodes => Ok(Self::ConnectClusterNodes),
+			// - GetClusterStatus: response format changed
+			// - GetClusterHealth: response format changed
+
+			// Layout endpoints
+			router_v0::Endpoint::RevertClusterLayout => Ok(Self::RevertClusterLayout),
+			// - GetClusterLayout: response format changed
+			// - UpdateClusterLayout: query format changed
+			// - ApplyCusterLayout: response format changed
+
+			// Key endpoints
+			router_v0::Endpoint::ListKeys => Ok(Self::ListKeys),
+			router_v0::Endpoint::CreateKey => Ok(Self::CreateKey),
+			router_v0::Endpoint::GetKeyInfo { id, search } => Ok(Self::GetKeyInfo {
+				id,
+				search,
+				show_secret_key: Some("true".into()),
+			}),
+			router_v0::Endpoint::DeleteKey { id } => Ok(Self::DeleteKey { id }),
+			// - UpdateKey: response format changed (secret key no longer returned)
+
+			// Bucket endpoints
+			router_v0::Endpoint::GetBucketInfo { id, global_alias } => {
+				Ok(Self::GetBucketInfo { id, global_alias })
+			}
+			router_v0::Endpoint::ListBuckets => Ok(Self::ListBuckets),
+			router_v0::Endpoint::CreateBucket => Ok(Self::CreateBucket),
+			router_v0::Endpoint::DeleteBucket { id } => Ok(Self::DeleteBucket { id }),
+			router_v0::Endpoint::UpdateBucket { id } => Ok(Self::UpdateBucket { id }),
+
+			// Bucket-key permissions
+			router_v0::Endpoint::BucketAllowKey => Ok(Self::BucketAllowKey),
+			router_v0::Endpoint::BucketDenyKey => Ok(Self::BucketDenyKey),
+
+			// Bucket alias endpoints
+			router_v0::Endpoint::GlobalAliasBucket { id, alias } => {
+				Ok(Self::GlobalAliasBucket { id, alias })
+			}
+			router_v0::Endpoint::GlobalUnaliasBucket { id, alias } => {
+				Ok(Self::GlobalUnaliasBucket { id, alias })
+			}
+			router_v0::Endpoint::LocalAliasBucket {
+				id,
+				access_key_id,
+				alias,
+			} => Ok(Self::LocalAliasBucket {
+				id,
+				access_key_id,
+				alias,
+			}),
+			router_v0::Endpoint::LocalUnaliasBucket {
+				id,
+				access_key_id,
+				alias,
+			} => Ok(Self::LocalUnaliasBucket {
+				id,
+				access_key_id,
+				alias,
+			}),
+
+			// For endpoints that have different body content syntax, issue
+			// deprecation warning
+			_ => Err(Error::bad_request(format!(
+				"v0/ endpoint is no longer supported: {}",
+				v0_endpoint.name()
+			))),
+		}
+	}
+	/// Get the kind of authorization which is required to perform the operation.
+	pub fn authorization_type(&self) -> Authorization {
+		match self {
+			Self::Health => Authorization::None,
+			Self::CheckDomain => Authorization::None,
+			Self::Metrics => Authorization::MetricsToken,
+			_ => Authorization::AdminToken,
+		}
+	}
+}
+
+generateQueryParameters! {
+	keywords: [],
+	fields: [
+		"format" => format,
+		"id" => id,
+		"search" => search,
+		"globalAlias" => global_alias,
+		"alias" => alias,
+		"accessKeyId" => access_key_id,
+		"showSecretKey" => show_secret_key
+	]
+}
diff --git a/src/api/helpers.rs b/src/api/helpers.rs
index 642dbc42..1d55ebd5 100644
--- a/src/api/helpers.rs
+++ b/src/api/helpers.rs
@@ -152,6 +152,10 @@ pub fn json_ok_response<T: Serialize>(res: &T) -> Result<Response<Body>, Error>
 		.body(Body::from(resp_json))?)
 }
 
+pub fn is_default<T: Default + PartialEq>(v: &T) -> bool {
+	*v == T::default()
+}
+
 #[cfg(test)]
 mod tests {
 	use super::*;
diff --git a/src/api/router_macros.rs b/src/api/router_macros.rs
index 07b5570c..cfecbc92 100644
--- a/src/api/router_macros.rs
+++ b/src/api/router_macros.rs
@@ -26,6 +26,7 @@ macro_rules! router_match {
      $($meth:ident $path:pat $(if $required:ident)? => $api:ident $(($($conv:ident :: $param:ident),*))?,)*
      ]) => {{
         {
+            #[allow(unused_parens)]
             match ($method, $reqpath) {
                 $(
                     (&Method::$meth, $path) if true $(&& $query.$required.is_some())? => Endpoint::$api {
@@ -128,12 +129,6 @@ macro_rules! router_match {
             }
         }
     };
-    (@if ($($cond:tt)+) then ($($then:tt)*) else ($($else:tt)*)) => {
-        $($then)*
-    };
-    (@if () then ($($then:tt)*) else ($($else:tt)*)) => {
-        $($else)*
-    };
 }
 
 /// This macro is used to generate part of the code in this module. It must be called only one, and
diff --git a/src/api/s3/api_server.rs b/src/api/s3/api_server.rs
index ecfb48b6..d675ab61 100644
--- a/src/api/s3/api_server.rs
+++ b/src/api/s3/api_server.rs
@@ -26,7 +26,9 @@ use crate::s3::copy::*;
 use crate::s3::cors::*;
 use crate::s3::delete::*;
 use crate::s3::get::*;
+use crate::s3::lifecycle::*;
 use crate::s3::list::*;
+use crate::s3::multipart::*;
 use crate::s3::post_object::handle_post_object;
 use crate::s3::put::*;
 use crate::s3::router::Endpoint;
@@ -256,7 +258,7 @@ impl ApiHandler for S3ApiServer {
 							bucket_name,
 							bucket_id,
 							delimiter: delimiter.map(|d| d.to_string()),
-							page_size: max_keys.map(|p| p.clamp(1, 1000)).unwrap_or(1000),
+							page_size: max_keys.unwrap_or(1000).clamp(1, 1000),
 							prefix: prefix.unwrap_or_default(),
 							urlencode_resp: encoding_type.map(|e| e == "url").unwrap_or(false),
 						},
@@ -286,7 +288,7 @@ impl ApiHandler for S3ApiServer {
 								bucket_name,
 								bucket_id,
 								delimiter: delimiter.map(|d| d.to_string()),
-								page_size: max_keys.map(|p| p.clamp(1, 1000)).unwrap_or(1000),
+								page_size: max_keys.unwrap_or(1000).clamp(1, 1000),
 								urlencode_resp: encoding_type.map(|e| e == "url").unwrap_or(false),
 								prefix: prefix.unwrap_or_default(),
 							},
@@ -319,7 +321,7 @@ impl ApiHandler for S3ApiServer {
 							bucket_name,
 							bucket_id,
 							delimiter: delimiter.map(|d| d.to_string()),
-							page_size: max_uploads.map(|p| p.clamp(1, 1000)).unwrap_or(1000),
+							page_size: max_uploads.unwrap_or(1000).clamp(1, 1000),
 							prefix: prefix.unwrap_or_default(),
 							urlencode_resp: encoding_type.map(|e| e == "url").unwrap_or(false),
 						},
@@ -343,7 +345,7 @@ impl ApiHandler for S3ApiServer {
 						key,
 						upload_id,
 						part_number_marker: part_number_marker.map(|p| p.clamp(1, 10000)),
-						max_parts: max_parts.map(|p| p.clamp(1, 1000)).unwrap_or(1000),
+						max_parts: max_parts.unwrap_or(1000).clamp(1, 1000),
 					},
 				)
 				.await
@@ -353,14 +355,21 @@ impl ApiHandler for S3ApiServer {
 			}
 			Endpoint::GetBucketWebsite {} => handle_get_website(&bucket).await,
 			Endpoint::PutBucketWebsite {} => {
-				handle_put_website(garage, bucket_id, req, content_sha256).await
+				handle_put_website(garage, bucket.clone(), req, content_sha256).await
 			}
-			Endpoint::DeleteBucketWebsite {} => handle_delete_website(garage, bucket_id).await,
+			Endpoint::DeleteBucketWebsite {} => handle_delete_website(garage, bucket.clone()).await,
 			Endpoint::GetBucketCors {} => handle_get_cors(&bucket).await,
 			Endpoint::PutBucketCors {} => {
-				handle_put_cors(garage, bucket_id, req, content_sha256).await
+				handle_put_cors(garage, bucket.clone(), req, content_sha256).await
+			}
+			Endpoint::DeleteBucketCors {} => handle_delete_cors(garage, bucket.clone()).await,
+			Endpoint::GetBucketLifecycleConfiguration {} => handle_get_lifecycle(&bucket).await,
+			Endpoint::PutBucketLifecycleConfiguration {} => {
+				handle_put_lifecycle(garage, bucket.clone(), req, content_sha256).await
+			}
+			Endpoint::DeleteBucketLifecycle {} => {
+				handle_delete_lifecycle(garage, bucket.clone()).await
 			}
-			Endpoint::DeleteBucketCors {} => handle_delete_cors(garage, bucket_id).await,
 			endpoint => Err(Error::NotImplemented(endpoint.name().to_owned())),
 		};
 
diff --git a/src/api/s3/copy.rs b/src/api/s3/copy.rs
index 7eb6459d..68b4f0c9 100644
--- a/src/api/s3/copy.rs
+++ b/src/api/s3/copy.rs
@@ -2,7 +2,7 @@ use std::pin::Pin;
 use std::sync::Arc;
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
-use futures::{stream, stream::Stream, StreamExt, TryFutureExt};
+use futures::{stream, stream::Stream, StreamExt};
 use md5::{Digest as Md5Digest, Md5};
 
 use bytes::Bytes;
@@ -18,12 +18,14 @@ use garage_util::time::*;
 use garage_model::garage::Garage;
 use garage_model::key_table::Key;
 use garage_model::s3::block_ref_table::*;
+use garage_model::s3::mpu_table::*;
 use garage_model::s3::object_table::*;
 use garage_model::s3::version_table::*;
 
 use crate::helpers::parse_bucket_key;
 use crate::s3::error::*;
-use crate::s3::put::{decode_upload_id, get_headers};
+use crate::s3::multipart;
+use crate::s3::put::get_headers;
 use crate::s3::xml::{self as s3_xml, xmlns_tag};
 
 pub async fn handle_copy(
@@ -92,7 +94,10 @@ pub async fn handle_copy(
 			let tmp_dest_object_version = ObjectVersion {
 				uuid: new_uuid,
 				timestamp: new_timestamp,
-				state: ObjectVersionState::Uploading(new_meta.headers.clone()),
+				state: ObjectVersionState::Uploading {
+					headers: new_meta.headers.clone(),
+					multipart: false,
+				},
 			};
 			let tmp_dest_object = Object::new(
 				dest_bucket_id,
@@ -105,8 +110,14 @@ pub async fn handle_copy(
 			// this means that the BlockRef entries linked to this version cannot be
 			// marked as deleted (they are marked as deleted only if the Version
 			// doesn't exist or is marked as deleted).
-			let mut dest_version =
-				Version::new(new_uuid, dest_bucket_id, dest_key.to_string(), false);
+			let mut dest_version = Version::new(
+				new_uuid,
+				VersionBacklink::Object {
+					bucket_id: dest_bucket_id,
+					key: dest_key.to_string(),
+				},
+				false,
+			);
 			garage.version_table.insert(&dest_version).await?;
 
 			// Fill in block list for version and insert block refs
@@ -179,17 +190,13 @@ pub async fn handle_upload_part_copy(
 ) -> Result<Response<Body>, Error> {
 	let copy_precondition = CopyPreconditionHeaders::parse(req)?;
 
-	let dest_version_uuid = decode_upload_id(upload_id)?;
+	let dest_upload_id = multipart::decode_upload_id(upload_id)?;
 
 	let dest_key = dest_key.to_string();
-	let (source_object, dest_object) = futures::try_join!(
+	let (source_object, (_, _, mut dest_mpu)) = futures::try_join!(
 		get_copy_source(&garage, api_key, req),
-		garage
-			.object_table
-			.get(&dest_bucket_id, &dest_key)
-			.map_err(Error::from),
+		multipart::get_upload(&garage, &dest_bucket_id, &dest_key, &dest_upload_id)
 	)?;
-	let dest_object = dest_object.ok_or(Error::NoSuchKey)?;
 
 	let (source_object_version, source_version_data, source_version_meta) =
 		extract_source_info(&source_object)?;
@@ -217,15 +224,6 @@ pub async fn handle_upload_part_copy(
 		},
 	};
 
-	// Check destination version is indeed in uploading state
-	if !dest_object
-		.versions()
-		.iter()
-		.any(|v| v.uuid == dest_version_uuid && v.is_uploading())
-	{
-		return Err(Error::NoSuchUpload);
-	}
-
 	// Check source version is not inlined
 	match source_version_data {
 		ObjectVersionData::DeleteMarker => unreachable!(),
@@ -242,23 +240,11 @@ pub async fn handle_upload_part_copy(
 
 	// Fetch source versin with its block list,
 	// and destination version to check part hasn't yet been uploaded
-	let (source_version, dest_version) = futures::try_join!(
-		garage
-			.version_table
-			.get(&source_object_version.uuid, &EmptyKey),
-		garage.version_table.get(&dest_version_uuid, &EmptyKey),
-	)?;
-	let source_version = source_version.ok_or(Error::NoSuchKey)?;
-
-	// Check this part number hasn't yet been uploaded
-	if let Some(dv) = dest_version {
-		if dv.has_part_number(part_number) {
-			return Err(Error::bad_request(format!(
-				"Part number {} has already been uploaded",
-				part_number
-			)));
-		}
-	}
+	let source_version = garage
+		.version_table
+		.get(&source_object_version.uuid, &EmptyKey)
+		.await?
+		.ok_or(Error::NoSuchKey)?;
 
 	// We want to reuse blocks from the source version as much as possible.
 	// However, we still need to get the data from these blocks
@@ -299,6 +285,33 @@ pub async fn handle_upload_part_copy(
 		current_offset = block_end;
 	}
 
+	// Calculate the identity of destination part: timestamp, version id
+	let dest_version_id = gen_uuid();
+	let dest_mpu_part_key = MpuPartKey {
+		part_number,
+		timestamp: dest_mpu.next_timestamp(part_number),
+	};
+
+	// Create the uploaded part
+	dest_mpu.parts.clear();
+	dest_mpu.parts.put(
+		dest_mpu_part_key,
+		MpuPart {
+			version: dest_version_id,
+			etag: None,
+			size: None,
+		},
+	);
+	garage.mpu_table.insert(&dest_mpu).await?;
+
+	let mut dest_version = Version::new(
+		dest_version_id,
+		VersionBacklink::MultipartUpload {
+			upload_id: dest_upload_id,
+		},
+		false,
+	);
+
 	// Now, actually copy the blocks
 	let mut md5hasher = Md5::new();
 
@@ -348,8 +361,8 @@ pub async fn handle_upload_part_copy(
 		let must_upload = existing_block_hash.is_none();
 		let final_hash = existing_block_hash.unwrap_or_else(|| blake2sum(&data[..]));
 
-		let mut version = Version::new(dest_version_uuid, dest_bucket_id, dest_key.clone(), false);
-		version.blocks.put(
+		dest_version.blocks.clear();
+		dest_version.blocks.put(
 			VersionBlockKey {
 				part_number,
 				offset: current_offset,
@@ -363,7 +376,7 @@ pub async fn handle_upload_part_copy(
 
 		let block_ref = BlockRef {
 			block: final_hash,
-			version: dest_version_uuid,
+			version: dest_version_id,
 			deleted: false.into(),
 		};
 
@@ -378,23 +391,33 @@ pub async fn handle_upload_part_copy(
 					Ok(())
 				}
 			},
-			// Thing 2: we need to insert the block in the version
-			garage.version_table.insert(&version),
-			// Thing 3: we need to add a block reference
-			garage.block_ref_table.insert(&block_ref),
+			async {
+				// Thing 2: we need to insert the block in the version
+				garage.version_table.insert(&dest_version).await?;
+				// Thing 3: we need to add a block reference
+				garage.block_ref_table.insert(&block_ref).await
+			},
 			// Thing 4: we need to prefetch the next block
 			defragmenter.next(),
 		)?;
-		next_block = res.3;
+		next_block = res.2;
 	}
 
+	assert_eq!(current_offset, source_range.length);
+
 	let data_md5sum = md5hasher.finalize();
 	let etag = hex::encode(data_md5sum);
 
 	// Put the part's ETag in the Versiontable
-	let mut version = Version::new(dest_version_uuid, dest_bucket_id, dest_key.clone(), false);
-	version.parts_etags.put(part_number, etag.clone());
-	garage.version_table.insert(&version).await?;
+	dest_mpu.parts.put(
+		dest_mpu_part_key,
+		MpuPart {
+			version: dest_version_id,
+			etag: Some(etag.clone()),
+			size: Some(current_offset),
+		},
+	);
+	garage.mpu_table.insert(&dest_mpu).await?;
 
 	// LGTM
 	let resp_xml = s3_xml::to_xml_with_header(&CopyPartResult {
diff --git a/src/api/s3/cors.rs b/src/api/s3/cors.rs
index c7273464..49097ad1 100644
--- a/src/api/s3/cors.rs
+++ b/src/api/s3/cors.rs
@@ -44,14 +44,11 @@ pub async fn handle_get_cors(bucket: &Bucket) -> Result<Response<Body>, Error> {
 
 pub async fn handle_delete_cors(
 	garage: Arc<Garage>,
-	bucket_id: Uuid,
+	mut bucket: Bucket,
 ) -> Result<Response<Body>, Error> {
-	let mut bucket = garage
-		.bucket_helper()
-		.get_existing_bucket(bucket_id)
-		.await?;
-
-	let param = bucket.params_mut().unwrap();
+	let param = bucket
+		.params_mut()
+		.ok_or_internal_error("Bucket should not be deleted at this point")?;
 
 	param.cors_config.update(None);
 	garage.bucket_table.insert(&bucket).await?;
@@ -63,7 +60,7 @@ pub async fn handle_delete_cors(
 
 pub async fn handle_put_cors(
 	garage: Arc<Garage>,
-	bucket_id: Uuid,
+	mut bucket: Bucket,
 	req: Request<Body>,
 	content_sha256: Option<Hash>,
 ) -> Result<Response<Body>, Error> {
@@ -73,12 +70,9 @@ pub async fn handle_put_cors(
 		verify_signed_content(content_sha256, &body[..])?;
 	}
 
-	let mut bucket = garage
-		.bucket_helper()
-		.get_existing_bucket(bucket_id)
-		.await?;
-
-	let param = bucket.params_mut().unwrap();
+	let param = bucket
+		.params_mut()
+		.ok_or_internal_error("Bucket should not be deleted at this point")?;
 
 	let conf: CorsConfiguration = from_reader(&body as &[u8])?;
 	conf.validate()?;
diff --git a/src/api/s3/get.rs b/src/api/s3/get.rs
index cde7b461..5e682726 100644
--- a/src/api/s3/get.rs
+++ b/src/api/s3/get.rs
@@ -149,7 +149,6 @@ pub async fn handle_head(
 
 				let (part_offset, part_end) =
 					calculate_part_bounds(&version, pn).ok_or(Error::InvalidPart)?;
-				let n_parts = version.parts_etags.items().len();
 
 				Ok(object_headers(object_version, version_meta)
 					.header(CONTENT_LENGTH, format!("{}", part_end - part_offset))
@@ -162,7 +161,7 @@ pub async fn handle_head(
 							version_meta.size
 						),
 					)
-					.header(X_AMZ_MP_PARTS_COUNT, format!("{}", n_parts))
+					.header(X_AMZ_MP_PARTS_COUNT, format!("{}", version.n_parts()?))
 					.status(StatusCode::PARTIAL_CONTENT)
 					.body(Body::empty())?)
 			}
@@ -376,7 +375,6 @@ async fn handle_get_part(
 
 			let (begin, end) =
 				calculate_part_bounds(&version, part_number).ok_or(Error::InvalidPart)?;
-			let n_parts = version.parts_etags.items().len();
 
 			let body = body_from_blocks_range(garage, version.blocks.items(), begin, end);
 
@@ -386,7 +384,7 @@ async fn handle_get_part(
 					CONTENT_RANGE,
 					format!("bytes {}-{}/{}", begin, end - 1, version_meta.size),
 				)
-				.header(X_AMZ_MP_PARTS_COUNT, format!("{}", n_parts))
+				.header(X_AMZ_MP_PARTS_COUNT, format!("{}", version.n_parts()?))
 				.body(body)?)
 		}
 		_ => unreachable!(),
diff --git a/src/api/s3/lifecycle.rs b/src/api/s3/lifecycle.rs
new file mode 100644
index 00000000..1e7d6755
--- /dev/null
+++ b/src/api/s3/lifecycle.rs
@@ -0,0 +1,401 @@
+use quick_xml::de::from_reader;
+use std::sync::Arc;
+
+use hyper::{Body, Request, Response, StatusCode};
+
+use serde::{Deserialize, Serialize};
+
+use crate::s3::error::*;
+use crate::s3::xml::{to_xml_with_header, xmlns_tag, IntValue, Value};
+use crate::signature::verify_signed_content;
+
+use garage_model::bucket_table::{
+	parse_lifecycle_date, Bucket, LifecycleExpiration as GarageLifecycleExpiration,
+	LifecycleFilter as GarageLifecycleFilter, LifecycleRule as GarageLifecycleRule,
+};
+use garage_model::garage::Garage;
+use garage_util::data::*;
+
+pub async fn handle_get_lifecycle(bucket: &Bucket) -> Result<Response<Body>, Error> {
+	let param = bucket
+		.params()
+		.ok_or_internal_error("Bucket should not be deleted at this point")?;
+
+	if let Some(lifecycle) = param.lifecycle_config.get() {
+		let wc = LifecycleConfiguration::from_garage_lifecycle_config(lifecycle);
+		let xml = to_xml_with_header(&wc)?;
+		Ok(Response::builder()
+			.status(StatusCode::OK)
+			.header(http::header::CONTENT_TYPE, "application/xml")
+			.body(Body::from(xml))?)
+	} else {
+		Ok(Response::builder()
+			.status(StatusCode::NO_CONTENT)
+			.body(Body::empty())?)
+	}
+}
+
+pub async fn handle_delete_lifecycle(
+	garage: Arc<Garage>,
+	mut bucket: Bucket,
+) -> Result<Response<Body>, Error> {
+	let param = bucket
+		.params_mut()
+		.ok_or_internal_error("Bucket should not be deleted at this point")?;
+
+	param.lifecycle_config.update(None);
+	garage.bucket_table.insert(&bucket).await?;
+
+	Ok(Response::builder()
+		.status(StatusCode::NO_CONTENT)
+		.body(Body::empty())?)
+}
+
+pub async fn handle_put_lifecycle(
+	garage: Arc<Garage>,
+	mut bucket: Bucket,
+	req: Request<Body>,
+	content_sha256: Option<Hash>,
+) -> Result<Response<Body>, Error> {
+	let body = hyper::body::to_bytes(req.into_body()).await?;
+
+	if let Some(content_sha256) = content_sha256 {
+		verify_signed_content(content_sha256, &body[..])?;
+	}
+
+	let param = bucket
+		.params_mut()
+		.ok_or_internal_error("Bucket should not be deleted at this point")?;
+
+	let conf: LifecycleConfiguration = from_reader(&body as &[u8])?;
+	let config = conf
+		.validate_into_garage_lifecycle_config()
+		.ok_or_bad_request("Invalid lifecycle configuration")?;
+
+	param.lifecycle_config.update(Some(config));
+	garage.bucket_table.insert(&bucket).await?;
+
+	Ok(Response::builder()
+		.status(StatusCode::OK)
+		.body(Body::empty())?)
+}
+
+// ---- SERIALIZATION AND DESERIALIZATION TO/FROM S3 XML ----
+
+#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)]
+pub struct LifecycleConfiguration {
+	#[serde(serialize_with = "xmlns_tag", skip_deserializing)]
+	pub xmlns: (),
+	#[serde(rename = "Rule")]
+	pub lifecycle_rules: Vec<LifecycleRule>,
+}
+
+#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)]
+pub struct LifecycleRule {
+	#[serde(rename = "ID")]
+	pub id: Option<Value>,
+	#[serde(rename = "Status")]
+	pub status: Value,
+	#[serde(rename = "Filter", default)]
+	pub filter: Option<Filter>,
+	#[serde(rename = "Expiration", default)]
+	pub expiration: Option<Expiration>,
+	#[serde(rename = "AbortIncompleteMultipartUpload", default)]
+	pub abort_incomplete_mpu: Option<AbortIncompleteMpu>,
+}
+
+#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord, Default)]
+pub struct Filter {
+	#[serde(rename = "And")]
+	pub and: Option<Box<Filter>>,
+	#[serde(rename = "Prefix")]
+	pub prefix: Option<Value>,
+	#[serde(rename = "ObjectSizeGreaterThan")]
+	pub size_gt: Option<IntValue>,
+	#[serde(rename = "ObjectSizeLessThan")]
+	pub size_lt: Option<IntValue>,
+}
+
+#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)]
+pub struct Expiration {
+	#[serde(rename = "Days")]
+	pub days: Option<IntValue>,
+	#[serde(rename = "Date")]
+	pub at_date: Option<Value>,
+}
+
+#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)]
+pub struct AbortIncompleteMpu {
+	#[serde(rename = "DaysAfterInitiation")]
+	pub days: IntValue,
+}
+
+impl LifecycleConfiguration {
+	pub fn validate_into_garage_lifecycle_config(
+		self,
+	) -> Result<Vec<GarageLifecycleRule>, &'static str> {
+		let mut ret = vec![];
+		for rule in self.lifecycle_rules {
+			ret.push(rule.validate_into_garage_lifecycle_rule()?);
+		}
+		Ok(ret)
+	}
+
+	pub fn from_garage_lifecycle_config(config: &[GarageLifecycleRule]) -> Self {
+		Self {
+			xmlns: (),
+			lifecycle_rules: config
+				.iter()
+				.map(LifecycleRule::from_garage_lifecycle_rule)
+				.collect(),
+		}
+	}
+}
+
+impl LifecycleRule {
+	pub fn validate_into_garage_lifecycle_rule(self) -> Result<GarageLifecycleRule, &'static str> {
+		let enabled = match self.status.0.as_str() {
+			"Enabled" => true,
+			"Disabled" => false,
+			_ => return Err("invalid value for <Status>"),
+		};
+
+		let filter = self
+			.filter
+			.map(Filter::validate_into_garage_lifecycle_filter)
+			.transpose()?
+			.unwrap_or_default();
+
+		let abort_incomplete_mpu_days = self.abort_incomplete_mpu.map(|x| x.days.0 as usize);
+
+		let expiration = self
+			.expiration
+			.map(Expiration::validate_into_garage_lifecycle_expiration)
+			.transpose()?;
+
+		Ok(GarageLifecycleRule {
+			id: self.id.map(|x| x.0),
+			enabled,
+			filter,
+			abort_incomplete_mpu_days,
+			expiration,
+		})
+	}
+
+	pub fn from_garage_lifecycle_rule(rule: &GarageLifecycleRule) -> Self {
+		Self {
+			id: rule.id.as_deref().map(Value::from),
+			status: if rule.enabled {
+				Value::from("Enabled")
+			} else {
+				Value::from("Disabled")
+			},
+			filter: Filter::from_garage_lifecycle_filter(&rule.filter),
+			abort_incomplete_mpu: rule
+				.abort_incomplete_mpu_days
+				.map(|days| AbortIncompleteMpu {
+					days: IntValue(days as i64),
+				}),
+			expiration: rule
+				.expiration
+				.as_ref()
+				.map(Expiration::from_garage_lifecycle_expiration),
+		}
+	}
+}
+
+impl Filter {
+	pub fn count(&self) -> i32 {
+		fn count<T>(x: &Option<T>) -> i32 {
+			x.as_ref().map(|_| 1).unwrap_or(0)
+		}
+		count(&self.prefix) + count(&self.size_gt) + count(&self.size_lt)
+	}
+
+	pub fn validate_into_garage_lifecycle_filter(
+		self,
+	) -> Result<GarageLifecycleFilter, &'static str> {
+		if self.count() > 0 && self.and.is_some() {
+			Err("Filter tag cannot contain both <And> and another condition")
+		} else if let Some(and) = self.and {
+			if and.and.is_some() {
+				return Err("Nested <And> tags");
+			}
+			Ok(and.internal_into_garage_lifecycle_filter())
+		} else if self.count() > 1 {
+			Err("Multiple Filter conditions must be wrapped in an <And> tag")
+		} else {
+			Ok(self.internal_into_garage_lifecycle_filter())
+		}
+	}
+
+	fn internal_into_garage_lifecycle_filter(self) -> GarageLifecycleFilter {
+		GarageLifecycleFilter {
+			prefix: self.prefix.map(|x| x.0),
+			size_gt: self.size_gt.map(|x| x.0 as u64),
+			size_lt: self.size_lt.map(|x| x.0 as u64),
+		}
+	}
+
+	pub fn from_garage_lifecycle_filter(rule: &GarageLifecycleFilter) -> Option<Self> {
+		let filter = Filter {
+			and: None,
+			prefix: rule.prefix.as_deref().map(Value::from),
+			size_gt: rule.size_gt.map(|x| IntValue(x as i64)),
+			size_lt: rule.size_lt.map(|x| IntValue(x as i64)),
+		};
+		match filter.count() {
+			0 => None,
+			1 => Some(filter),
+			_ => Some(Filter {
+				and: Some(Box::new(filter)),
+				..Default::default()
+			}),
+		}
+	}
+}
+
+impl Expiration {
+	pub fn validate_into_garage_lifecycle_expiration(
+		self,
+	) -> Result<GarageLifecycleExpiration, &'static str> {
+		match (self.days, self.at_date) {
+			(Some(_), Some(_)) => Err("cannot have both <Days> and <Date> in <Expiration>"),
+			(None, None) => Err("<Expiration> must contain either <Days> or <Date>"),
+			(Some(days), None) => Ok(GarageLifecycleExpiration::AfterDays(days.0 as usize)),
+			(None, Some(date)) => {
+				parse_lifecycle_date(&date.0)?;
+				Ok(GarageLifecycleExpiration::AtDate(date.0))
+			}
+		}
+	}
+
+	pub fn from_garage_lifecycle_expiration(exp: &GarageLifecycleExpiration) -> Self {
+		match exp {
+			GarageLifecycleExpiration::AfterDays(days) => Expiration {
+				days: Some(IntValue(*days as i64)),
+				at_date: None,
+			},
+			GarageLifecycleExpiration::AtDate(date) => Expiration {
+				days: None,
+				at_date: Some(Value(date.to_string())),
+			},
+		}
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+
+	use quick_xml::de::from_str;
+
+	#[test]
+	fn test_deserialize_lifecycle_config() -> Result<(), Error> {
+		let message = r#"<?xml version="1.0" encoding="UTF-8"?>
+<LifecycleConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
+  <Rule>
+    <ID>id1</ID>
+    <Status>Enabled</Status>
+    <Filter>
+       <Prefix>documents/</Prefix>
+    </Filter>
+    <AbortIncompleteMultipartUpload>
+       <DaysAfterInitiation>7</DaysAfterInitiation>
+    </AbortIncompleteMultipartUpload>
+  </Rule>
+  <Rule>
+    <ID>id2</ID>
+    <Status>Enabled</Status>
+    <Filter>
+       <And>
+          <Prefix>logs/</Prefix>
+          <ObjectSizeGreaterThan>1000000</ObjectSizeGreaterThan>
+       </And>
+    </Filter>
+    <Expiration>
+      <Days>365</Days>
+    </Expiration>
+  </Rule>
+</LifecycleConfiguration>"#;
+		let conf: LifecycleConfiguration = from_str(message).unwrap();
+		let ref_value = LifecycleConfiguration {
+			xmlns: (),
+			lifecycle_rules: vec![
+				LifecycleRule {
+					id: Some("id1".into()),
+					status: "Enabled".into(),
+					filter: Some(Filter {
+						prefix: Some("documents/".into()),
+						..Default::default()
+					}),
+					expiration: None,
+					abort_incomplete_mpu: Some(AbortIncompleteMpu { days: IntValue(7) }),
+				},
+				LifecycleRule {
+					id: Some("id2".into()),
+					status: "Enabled".into(),
+					filter: Some(Filter {
+						and: Some(Box::new(Filter {
+							prefix: Some("logs/".into()),
+							size_gt: Some(IntValue(1000000)),
+							..Default::default()
+						})),
+						..Default::default()
+					}),
+					expiration: Some(Expiration {
+						days: Some(IntValue(365)),
+						at_date: None,
+					}),
+					abort_incomplete_mpu: None,
+				},
+			],
+		};
+		assert_eq! {
+			ref_value,
+			conf
+		};
+
+		let message2 = to_xml_with_header(&ref_value)?;
+
+		let cleanup = |c: &str| c.replace(char::is_whitespace, "");
+		assert_eq!(cleanup(message), cleanup(&message2));
+
+		// Check validation
+		let validated = ref_value
+			.validate_into_garage_lifecycle_config()
+			.ok_or_bad_request("invalid xml config")?;
+
+		let ref_config = vec![
+			GarageLifecycleRule {
+				id: Some("id1".into()),
+				enabled: true,
+				filter: GarageLifecycleFilter {
+					prefix: Some("documents/".into()),
+					..Default::default()
+				},
+				expiration: None,
+				abort_incomplete_mpu_days: Some(7),
+			},
+			GarageLifecycleRule {
+				id: Some("id2".into()),
+				enabled: true,
+				filter: GarageLifecycleFilter {
+					prefix: Some("logs/".into()),
+					size_gt: Some(1000000),
+					..Default::default()
+				},
+				expiration: Some(GarageLifecycleExpiration::AfterDays(365)),
+				abort_incomplete_mpu_days: None,
+			},
+		];
+		assert_eq!(validated, ref_config);
+
+		let message3 = to_xml_with_header(&LifecycleConfiguration::from_garage_lifecycle_config(
+			&validated,
+		))?;
+		assert_eq!(cleanup(message), cleanup(&message3));
+
+		Ok(())
+	}
+}
diff --git a/src/api/s3/list.rs b/src/api/s3/list.rs
index 5cb0d65a..33d62518 100644
--- a/src/api/s3/list.rs
+++ b/src/api/s3/list.rs
@@ -1,4 +1,3 @@
-use std::cmp::Ordering;
 use std::collections::{BTreeMap, BTreeSet};
 use std::iter::{Iterator, Peekable};
 use std::sync::Arc;
@@ -11,15 +10,15 @@ use garage_util::error::Error as GarageError;
 use garage_util::time::*;
 
 use garage_model::garage::Garage;
+use garage_model::s3::mpu_table::*;
 use garage_model::s3::object_table::*;
-use garage_model::s3::version_table::Version;
 
-use garage_table::{EmptyKey, EnumerationOrder};
+use garage_table::EnumerationOrder;
 
 use crate::encoding::*;
 use crate::helpers::key_after_prefix;
 use crate::s3::error::*;
-use crate::s3::put as s3_put;
+use crate::s3::multipart as s3_multipart;
 use crate::s3::xml as s3_xml;
 
 const DUMMY_NAME: &str = "Dummy Key";
@@ -176,7 +175,9 @@ pub async fn handle_list_multipart_upload(
 			t.get_range(
 				&bucket,
 				key,
-				Some(ObjectFilter::IsUploading),
+				Some(ObjectFilter::IsUploading {
+					check_multipart: Some(true),
+				}),
 				count,
 				EnumerationOrder::Forward,
 			)
@@ -272,24 +273,26 @@ pub async fn handle_list_parts(
 ) -> Result<Response<Body>, Error> {
 	debug!("ListParts {:?}", query);
 
-	let upload_id = s3_put::decode_upload_id(&query.upload_id)?;
+	let upload_id = s3_multipart::decode_upload_id(&query.upload_id)?;
 
-	let (object, version) = futures::try_join!(
-		garage.object_table.get(&query.bucket_id, &query.key),
-		garage.version_table.get(&upload_id, &EmptyKey),
-	)?;
+	let (_, _, mpu) =
+		s3_multipart::get_upload(&garage, &query.bucket_id, &query.key, &upload_id).await?;
 
-	let (info, next) = fetch_part_info(query, object, version, upload_id)?;
+	let (info, next) = fetch_part_info(query, &mpu)?;
 
 	let result = s3_xml::ListPartsResult {
 		xmlns: (),
+
+		// Query parameters
 		bucket: s3_xml::Value(query.bucket_name.to_string()),
 		key: s3_xml::Value(query.key.to_string()),
 		upload_id: s3_xml::Value(query.upload_id.to_string()),
 		part_number_marker: query.part_number_marker.map(|e| s3_xml::IntValue(e as i64)),
-		next_part_number_marker: next.map(|e| s3_xml::IntValue(e as i64)),
 		max_parts: s3_xml::IntValue(query.max_parts as i64),
-		is_truncated: s3_xml::Value(next.map(|_| "true").unwrap_or("false").to_string()),
+
+		// Result values
+		next_part_number_marker: next.map(|e| s3_xml::IntValue(e as i64)),
+		is_truncated: s3_xml::Value(format!("{}", next.is_some())),
 		parts: info
 			.iter()
 			.map(|part| s3_xml::PartItem {
@@ -299,6 +302,8 @@ pub async fn handle_list_parts(
 				size: s3_xml::IntValue(part.size as i64),
 			})
 			.collect(),
+
+		// Dummy result values (unsupported features)
 		initiator: s3_xml::Initiator {
 			display_name: s3_xml::Value(DUMMY_NAME.to_string()),
 			id: s3_xml::Value(DUMMY_KEY.to_string()),
@@ -335,8 +340,8 @@ struct UploadInfo {
 }
 
 #[derive(Debug, PartialEq)]
-struct PartInfo {
-	etag: String,
+struct PartInfo<'a> {
+	etag: &'a str,
 	timestamp: u64,
 	part_number: u64,
 	size: u64,
@@ -456,107 +461,51 @@ where
 	}
 }
 
-fn fetch_part_info(
+fn fetch_part_info<'a>(
 	query: &ListPartsQuery,
-	object: Option<Object>,
-	version: Option<Version>,
-	upload_id: Uuid,
-) -> Result<(Vec<PartInfo>, Option<u64>), Error> {
-	// Check results
-	let object = object.ok_or(Error::NoSuchKey)?;
-
-	let obj_version = object
-		.versions()
-		.iter()
-		.find(|v| v.uuid == upload_id && v.is_uploading())
-		.ok_or(Error::NoSuchUpload)?;
-
-	let version = version.ok_or(Error::NoSuchKey)?;
-
-	// Cut the beginning of our 2 vectors if required
-	let (etags, blocks) = match &query.part_number_marker {
-		Some(marker) => {
-			let next = marker + 1;
-
-			let part_idx = into_ok_or_err(
-				version
-					.parts_etags
-					.items()
-					.binary_search_by(|(part_num, _)| part_num.cmp(&next)),
-			);
-			let parts = &version.parts_etags.items()[part_idx..];
-
-			let block_idx = into_ok_or_err(
-				version
-					.blocks
-					.items()
-					.binary_search_by(|(vkey, _)| vkey.part_number.cmp(&next)),
-			);
-			let blocks = &version.blocks.items()[block_idx..];
-
-			(parts, blocks)
-		}
-		None => (version.parts_etags.items(), version.blocks.items()),
-	};
-
-	// Use the block vector to compute a (part_number, size) vector
-	let mut size = Vec::<(u64, u64)>::new();
-	blocks.iter().for_each(|(key, val)| {
-		let mut new_size = val.size;
-		match size.pop() {
-			Some((part_number, size)) if part_number == key.part_number => new_size += size,
-			Some(v) => size.push(v),
-			None => (),
-		}
-		size.push((key.part_number, new_size))
-	});
-
-	// Merge the etag vector and size vector to build a PartInfo vector
-	let max_parts = query.max_parts as usize;
-	let (mut etag_iter, mut size_iter) = (etags.iter().peekable(), size.iter().peekable());
-
-	let mut info = Vec::<PartInfo>::with_capacity(max_parts);
-
-	while info.len() < max_parts {
-		match (etag_iter.peek(), size_iter.peek()) {
-			(Some((ep, etag)), Some((sp, size))) => match ep.cmp(sp) {
-				Ordering::Less => {
-					debug!("ETag information ignored due to missing corresponding block information. Query: {:?}", query);
-					etag_iter.next();
-				}
-				Ordering::Equal => {
-					info.push(PartInfo {
-						etag: etag.to_string(),
-						timestamp: obj_version.timestamp,
-						part_number: *ep,
-						size: *size,
-					});
-					etag_iter.next();
-					size_iter.next();
+	mpu: &'a MultipartUpload,
+) -> Result<(Vec<PartInfo<'a>>, Option<u64>), Error> {
+	assert!((1..=1000).contains(&query.max_parts)); // see s3/api_server.rs
+
+	// Parse multipart upload part list, removing parts not yet finished
+	// and failed part uploads that were overwritten
+	let mut parts: Vec<PartInfo<'a>> = Vec::with_capacity(mpu.parts.items().len());
+	for (pk, p) in mpu.parts.items().iter() {
+		if let (Some(etag), Some(size)) = (&p.etag, p.size) {
+			let part_info = PartInfo {
+				part_number: pk.part_number,
+				timestamp: pk.timestamp,
+				etag,
+				size,
+			};
+			match parts.last_mut() {
+				Some(lastpart) if lastpart.part_number == pk.part_number => {
+					*lastpart = part_info;
 				}
-				Ordering::Greater => {
-					debug!("Block information ignored due to missing corresponding ETag information. Query: {:?}", query);
-					size_iter.next();
+				_ => {
+					parts.push(part_info);
 				}
-			},
-			(None, None) => return Ok((info, None)),
-			_ => {
-				debug!(
-					"Additional block or ETag information ignored. Query: {:?}",
-					query
-				);
-				return Ok((info, None));
 			}
 		}
 	}
 
-	match info.last() {
-		Some(part_info) => {
-			let pagination = Some(part_info.part_number);
-			Ok((info, pagination))
-		}
-		None => Ok((info, None)),
+	// Cut the beginning if we have a marker
+	if let Some(marker) = &query.part_number_marker {
+		let next = marker + 1;
+		let part_idx = parts
+			.binary_search_by(|part| part.part_number.cmp(&next))
+			.unwrap_or_else(|x| x);
+		parts = parts.split_off(part_idx);
+	}
+
+	// Cut the end if we have too many parts
+	if parts.len() > query.max_parts as usize {
+		parts.truncate(query.max_parts as usize);
+		let pagination = Some(parts.last().unwrap().part_number);
+		return Ok((parts, pagination));
 	}
+
+	Ok((parts, None))
 }
 
 /*
@@ -651,7 +600,7 @@ impl ListMultipartUploadsQuery {
 				}),
 				uuid => Ok(RangeBegin::AfterUpload {
 					key: key_marker.to_string(),
-					upload: s3_put::decode_upload_id(uuid)?,
+					upload: s3_multipart::decode_upload_id(uuid)?,
 				}),
 			},
 
@@ -843,7 +792,7 @@ impl ExtractAccumulator for UploadAccumulator {
 		let mut uploads_for_key = object
 			.versions()
 			.iter()
-			.filter(|x| x.is_uploading())
+			.filter(|x| x.is_uploading(Some(true)))
 			.collect::<Vec<&ObjectVersion>>();
 
 		// S3 logic requires lexicographically sorted upload ids.
@@ -918,14 +867,6 @@ impl ExtractAccumulator for UploadAccumulator {
  * Utility functions
  */
 
-/// This is a stub for Result::into_ok_or_err that is not yet in Rust stable
-fn into_ok_or_err<T>(r: Result<T, T>) -> T {
-	match r {
-		Ok(r) => r,
-		Err(r) => r,
-	}
-}
-
 /// Returns the common prefix of the object given the query prefix and delimiter
 fn common_prefix<'a>(object: &'a Object, query: &ListQueryCommon) -> Option<&'a str> {
 	match &query.delimiter {
@@ -951,7 +892,6 @@ fn uriencode_maybe(s: &str, yes: bool) -> s3_xml::Value {
 #[cfg(test)]
 mod tests {
 	use super::*;
-	use garage_model::s3::version_table::*;
 	use garage_util::*;
 	use std::iter::FromIterator;
 
@@ -991,10 +931,13 @@ mod tests {
 		ObjectVersion {
 			uuid: Uuid::from(uuid),
 			timestamp: TS,
-			state: ObjectVersionState::Uploading(ObjectVersionHeaders {
-				content_type: "text/plain".to_string(),
-				other: BTreeMap::<String, String>::new(),
-			}),
+			state: ObjectVersionState::Uploading {
+				multipart: true,
+				headers: ObjectVersionHeaders {
+					content_type: "text/plain".to_string(),
+					other: BTreeMap::<String, String>::new(),
+				},
+			},
 		}
 	}
 
@@ -1169,83 +1112,77 @@ mod tests {
 		Ok(())
 	}
 
-	fn version() -> Version {
+	fn mpu() -> MultipartUpload {
 		let uuid = Uuid::from([0x08; 32]);
 
-		let blocks = vec![
+		let parts = vec![
 			(
-				VersionBlockKey {
+				MpuPartKey {
 					part_number: 1,
-					offset: 1,
+					timestamp: TS,
 				},
-				VersionBlock {
-					hash: uuid,
-					size: 3,
+				MpuPart {
+					version: uuid,
+					size: Some(3),
+					etag: Some("etag1".into()),
 				},
 			),
 			(
-				VersionBlockKey {
-					part_number: 1,
-					offset: 2,
+				MpuPartKey {
+					part_number: 2,
+					timestamp: TS,
 				},
-				VersionBlock {
-					hash: uuid,
-					size: 2,
+				MpuPart {
+					version: uuid,
+					size: None,
+					etag: None,
 				},
 			),
 			(
-				VersionBlockKey {
-					part_number: 2,
-					offset: 1,
+				MpuPartKey {
+					part_number: 3,
+					timestamp: TS,
 				},
-				VersionBlock {
-					hash: uuid,
-					size: 8,
+				MpuPart {
+					version: uuid,
+					size: Some(10),
+					etag: Some("etag2".into()),
 				},
 			),
 			(
-				VersionBlockKey {
+				MpuPartKey {
 					part_number: 5,
-					offset: 1,
+					timestamp: TS,
 				},
-				VersionBlock {
-					hash: uuid,
-					size: 7,
+				MpuPart {
+					version: uuid,
+					size: Some(7),
+					etag: Some("etag3".into()),
 				},
 			),
 			(
-				VersionBlockKey {
+				MpuPartKey {
 					part_number: 8,
-					offset: 1,
+					timestamp: TS,
 				},
-				VersionBlock {
-					hash: uuid,
-					size: 5,
+				MpuPart {
+					version: uuid,
+					size: Some(5),
+					etag: Some("etag4".into()),
 				},
 			),
 		];
-		let etags = vec![
-			(1, "etag1".to_string()),
-			(3, "etag2".to_string()),
-			(5, "etag3".to_string()),
-			(8, "etag4".to_string()),
-			(9, "etag5".to_string()),
-		];
 
-		Version {
-			bucket_id: uuid,
-			key: "a".to_string(),
-			uuid,
+		MultipartUpload {
+			upload_id: uuid,
+			timestamp: TS,
 			deleted: false.into(),
-			blocks: crdt::Map::<VersionBlockKey, VersionBlock>::from_iter(blocks),
-			parts_etags: crdt::Map::<u64, String>::from_iter(etags),
+			parts: crdt::Map::<MpuPartKey, MpuPart>::from_iter(parts),
+			bucket_id: uuid,
+			key: "a".into(),
 		}
 	}
 
-	fn obj() -> Object {
-		Object::new(bucket(), "d".to_string(), vec![objup_version([0x08; 32])])
-	}
-
 	#[test]
 	fn test_fetch_part_info() -> Result<(), Error> {
 		let uuid = Uuid::from([0x08; 32]);
@@ -1258,82 +1195,85 @@ mod tests {
 			max_parts: 2,
 		};
 
-		assert!(
-			fetch_part_info(&query, None, None, uuid).is_err(),
-			"No object and version should fail"
-		);
-		assert!(
-			fetch_part_info(&query, Some(obj()), None, uuid).is_err(),
-			"No version should faild"
-		);
-		assert!(
-			fetch_part_info(&query, None, Some(version()), uuid).is_err(),
-			"No object should fail"
-		);
+		let mpu = mpu();
 
 		// Start from the beginning but with limited size to trigger pagination
-		let (info, pagination) = fetch_part_info(&query, Some(obj()), Some(version()), uuid)?;
-		assert_eq!(pagination.unwrap(), 5);
+		let (info, pagination) = fetch_part_info(&query, &mpu)?;
+		assert_eq!(pagination.unwrap(), 3);
 		assert_eq!(
 			info,
 			vec![
 				PartInfo {
-					etag: "etag1".to_string(),
+					etag: "etag1",
 					timestamp: TS,
 					part_number: 1,
-					size: 5
+					size: 3
 				},
 				PartInfo {
-					etag: "etag3".to_string(),
+					etag: "etag2",
 					timestamp: TS,
-					part_number: 5,
-					size: 7
+					part_number: 3,
+					size: 10
 				},
 			]
 		);
 
 		// Use previous pagination to make a new request
 		query.part_number_marker = Some(pagination.unwrap());
-		let (info, pagination) = fetch_part_info(&query, Some(obj()), Some(version()), uuid)?;
+		let (info, pagination) = fetch_part_info(&query, &mpu)?;
 		assert!(pagination.is_none());
 		assert_eq!(
 			info,
-			vec![PartInfo {
-				etag: "etag4".to_string(),
-				timestamp: TS,
-				part_number: 8,
-				size: 5
-			},]
+			vec![
+				PartInfo {
+					etag: "etag3",
+					timestamp: TS,
+					part_number: 5,
+					size: 7
+				},
+				PartInfo {
+					etag: "etag4",
+					timestamp: TS,
+					part_number: 8,
+					size: 5
+				},
+			]
 		);
 
 		// Trying to access a part that is way larger than registered ones
 		query.part_number_marker = Some(9999);
-		let (info, pagination) = fetch_part_info(&query, Some(obj()), Some(version()), uuid)?;
+		let (info, pagination) = fetch_part_info(&query, &mpu)?;
 		assert!(pagination.is_none());
 		assert_eq!(info, vec![]);
 
 		// Try without any limitation
 		query.max_parts = 1000;
 		query.part_number_marker = None;
-		let (info, pagination) = fetch_part_info(&query, Some(obj()), Some(version()), uuid)?;
+		let (info, pagination) = fetch_part_info(&query, &mpu)?;
 		assert!(pagination.is_none());
 		assert_eq!(
 			info,
 			vec![
 				PartInfo {
-					etag: "etag1".to_string(),
+					etag: "etag1",
 					timestamp: TS,
 					part_number: 1,
-					size: 5
+					size: 3
+				},
+				PartInfo {
+					etag: "etag2",
+					timestamp: TS,
+					part_number: 3,
+					size: 10
 				},
 				PartInfo {
-					etag: "etag3".to_string(),
+					etag: "etag3",
 					timestamp: TS,
 					part_number: 5,
 					size: 7
 				},
 				PartInfo {
-					etag: "etag4".to_string(),
+					etag: "etag4",
 					timestamp: TS,
 					part_number: 8,
 					size: 5
diff --git a/src/api/s3/mod.rs b/src/api/s3/mod.rs
index 7b56d4d8..cbdb94ab 100644
--- a/src/api/s3/mod.rs
+++ b/src/api/s3/mod.rs
@@ -6,7 +6,9 @@ mod copy;
 pub mod cors;
 mod delete;
 pub mod get;
+mod lifecycle;
 mod list;
+mod multipart;
 mod post_object;
 mod put;
 mod website;
diff --git a/src/api/s3/multipart.rs b/src/api/s3/multipart.rs
new file mode 100644
index 00000000..52ea8e78
--- /dev/null
+++ b/src/api/s3/multipart.rs
@@ -0,0 +1,465 @@
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use futures::prelude::*;
+use hyper::body::Body;
+use hyper::{Request, Response};
+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;
+use garage_model::s3::block_ref_table::*;
+use garage_model::s3::mpu_table::*;
+use garage_model::s3::object_table::*;
+use garage_model::s3::version_table::*;
+
+use crate::s3::error::*;
+use crate::s3::put::*;
+use crate::s3::xml as s3_xml;
+use crate::signature::verify_signed_content;
+
+// ----
+
+pub async fn handle_create_multipart_upload(
+	garage: Arc<Garage>,
+	req: &Request<Body>,
+	bucket_name: &str,
+	bucket_id: Uuid,
+	key: &str,
+) -> Result<Response<Body>, Error> {
+	let upload_id = gen_uuid();
+	let timestamp = now_msec();
+	let headers = get_headers(req.headers())?;
+
+	// Create object in object table
+	let object_version = ObjectVersion {
+		uuid: upload_id,
+		timestamp,
+		state: ObjectVersionState::Uploading {
+			multipart: true,
+			headers,
+		},
+	};
+	let object = Object::new(bucket_id, key.to_string(), vec![object_version]);
+	garage.object_table.insert(&object).await?;
+
+	// Create multipart upload in mpu table
+	// This multipart upload will hold references to uploaded parts
+	// (which are entries in the Version table)
+	let mpu = MultipartUpload::new(upload_id, timestamp, bucket_id, key.into(), false);
+	garage.mpu_table.insert(&mpu).await?;
+
+	// Send success response
+	let result = s3_xml::InitiateMultipartUploadResult {
+		xmlns: (),
+		bucket: s3_xml::Value(bucket_name.to_string()),
+		key: s3_xml::Value(key.to_string()),
+		upload_id: s3_xml::Value(hex::encode(upload_id)),
+	};
+	let xml = s3_xml::to_xml_with_header(&result)?;
+
+	Ok(Response::new(Body::from(xml.into_bytes())))
+}
+
+pub async fn handle_put_part(
+	garage: Arc<Garage>,
+	req: Request<Body>,
+	bucket_id: Uuid,
+	key: &str,
+	part_number: u64,
+	upload_id: &str,
+	content_sha256: Option<Hash>,
+) -> Result<Response<Body>, Error> {
+	let upload_id = decode_upload_id(upload_id)?;
+
+	let content_md5 = match req.headers().get("content-md5") {
+		Some(x) => Some(x.to_str()?.to_string()),
+		None => None,
+	};
+
+	// Read first chuck, and at the same time try to get object to see if it exists
+	let key = key.to_string();
+
+	let body = req.into_body().map_err(Error::from);
+	let mut chunker = StreamChunker::new(body, garage.config.block_size);
+
+	let ((_, _, mut mpu), first_block) = futures::try_join!(
+		get_upload(&garage, &bucket_id, &key, &upload_id),
+		chunker.next(),
+	)?;
+
+	// Check object is valid and part can be accepted
+	let first_block = first_block.ok_or_bad_request("Empty body")?;
+
+	// Calculate part identity: timestamp, version id
+	let version_uuid = gen_uuid();
+	let mpu_part_key = MpuPartKey {
+		part_number,
+		timestamp: mpu.next_timestamp(part_number),
+	};
+
+	// The following consists in many steps that can each fail.
+	// Keep track that some cleanup will be needed if things fail
+	// before everything is finished (cleanup is done using the Drop trait).
+	let mut interrupted_cleanup = InterruptedCleanup(Some(InterruptedCleanupInner {
+		garage: garage.clone(),
+		upload_id,
+		version_uuid,
+	}));
+
+	// Create version and link version from MPU
+	mpu.parts.clear();
+	mpu.parts.put(
+		mpu_part_key,
+		MpuPart {
+			version: version_uuid,
+			etag: None,
+			size: None,
+		},
+	);
+	garage.mpu_table.insert(&mpu).await?;
+
+	let version = Version::new(
+		version_uuid,
+		VersionBacklink::MultipartUpload { upload_id },
+		false,
+	);
+	garage.version_table.insert(&version).await?;
+
+	// Copy data to version
+	let first_block_hash = async_blake2sum(first_block.clone()).await;
+
+	let (total_size, data_md5sum, data_sha256sum) = read_and_put_blocks(
+		&garage,
+		&version,
+		part_number,
+		first_block,
+		first_block_hash,
+		&mut chunker,
+	)
+	.await?;
+
+	// Verify that checksums map
+	ensure_checksum_matches(
+		data_md5sum.as_slice(),
+		data_sha256sum,
+		content_md5.as_deref(),
+		content_sha256,
+	)?;
+
+	// Store part etag in version
+	let data_md5sum_hex = hex::encode(data_md5sum);
+	mpu.parts.put(
+		mpu_part_key,
+		MpuPart {
+			version: version_uuid,
+			etag: Some(data_md5sum_hex.clone()),
+			size: Some(total_size),
+		},
+	);
+	garage.mpu_table.insert(&mpu).await?;
+
+	// We were not interrupted, everything went fine.
+	// We won't have to clean up on drop.
+	interrupted_cleanup.cancel();
+
+	let response = Response::builder()
+		.header("ETag", format!("\"{}\"", data_md5sum_hex))
+		.body(Body::empty())
+		.unwrap();
+	Ok(response)
+}
+
+struct InterruptedCleanup(Option<InterruptedCleanupInner>);
+struct InterruptedCleanupInner {
+	garage: Arc<Garage>,
+	upload_id: Uuid,
+	version_uuid: Uuid,
+}
+
+impl InterruptedCleanup {
+	fn cancel(&mut self) {
+		drop(self.0.take());
+	}
+}
+impl Drop for InterruptedCleanup {
+	fn drop(&mut self) {
+		if let Some(info) = self.0.take() {
+			tokio::spawn(async move {
+				let version = Version::new(
+					info.version_uuid,
+					VersionBacklink::MultipartUpload {
+						upload_id: info.upload_id,
+					},
+					true,
+				);
+				if let Err(e) = info.garage.version_table.insert(&version).await {
+					warn!("Cannot cleanup after aborted UploadPart: {}", e);
+				}
+			});
+		}
+	}
+}
+
+pub async fn handle_complete_multipart_upload(
+	garage: Arc<Garage>,
+	req: Request<Body>,
+	bucket_name: &str,
+	bucket: &Bucket,
+	key: &str,
+	upload_id: &str,
+	content_sha256: Option<Hash>,
+) -> Result<Response<Body>, Error> {
+	let body = hyper::body::to_bytes(req.into_body()).await?;
+
+	if let Some(content_sha256) = content_sha256 {
+		verify_signed_content(content_sha256, &body[..])?;
+	}
+
+	let body_xml = roxmltree::Document::parse(std::str::from_utf8(&body)?)?;
+	let body_list_of_parts = parse_complete_multipart_upload_body(&body_xml)
+		.ok_or_bad_request("Invalid CompleteMultipartUpload XML")?;
+	debug!(
+		"CompleteMultipartUpload list of parts: {:?}",
+		body_list_of_parts
+	);
+
+	let upload_id = decode_upload_id(upload_id)?;
+
+	// Get object and multipart upload
+	let key = key.to_string();
+	let (_, 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"));
+	}
+
+	let headers = match object_version.state {
+		ObjectVersionState::Uploading { headers, .. } => headers,
+		_ => unreachable!(),
+	};
+
+	// Check that part numbers are an increasing sequence.
+	// (it doesn't need to start at 1 nor to be a continuous sequence,
+	// see discussion in #192)
+	if body_list_of_parts.is_empty() {
+		return Err(Error::EntityTooSmall);
+	}
+	if !body_list_of_parts
+		.iter()
+		.zip(body_list_of_parts.iter().skip(1))
+		.all(|(p1, p2)| p1.part_number < p2.part_number)
+	{
+		return Err(Error::InvalidPartOrder);
+	}
+
+	// Check that the list of parts they gave us corresponds to parts we have here
+	debug!("Parts stored in multipart upload: {:?}", mpu.parts.items());
+	let mut have_parts = HashMap::new();
+	for (pk, pv) in mpu.parts.items().iter() {
+		have_parts.insert(pk.part_number, pv);
+	}
+	let mut parts = vec![];
+	for req_part in body_list_of_parts.iter() {
+		match have_parts.get(&req_part.part_number) {
+			Some(part) if part.etag.as_ref() == Some(&req_part.etag) && part.size.is_some() => {
+				parts.push(*part)
+			}
+			_ => return Err(Error::InvalidPart),
+		}
+	}
+
+	let grg = &garage;
+	let parts_versions = futures::future::try_join_all(parts.iter().map(|p| async move {
+		grg.version_table
+			.get(&p.version, &EmptyKey)
+			.await?
+			.ok_or_internal_error("Part version missing from version table")
+	}))
+	.await?;
+
+	// Create final version and block refs
+	let mut final_version = Version::new(
+		upload_id,
+		VersionBacklink::Object {
+			bucket_id: bucket.id,
+			key: key.to_string(),
+		},
+		false,
+	);
+	for (part_number, part_version) in parts_versions.iter().enumerate() {
+		if part_version.deleted.get() {
+			return Err(Error::InvalidPart);
+		}
+		for (vbk, vb) in part_version.blocks.items().iter() {
+			final_version.blocks.put(
+				VersionBlockKey {
+					part_number: (part_number + 1) as u64,
+					offset: vbk.offset,
+				},
+				*vb,
+			);
+		}
+	}
+	garage.version_table.insert(&final_version).await?;
+
+	let block_refs = final_version.blocks.items().iter().map(|(_, b)| BlockRef {
+		block: b.hash,
+		version: upload_id,
+		deleted: false.into(),
+	});
+	garage.block_ref_table.insert_many(block_refs).await?;
+
+	// Calculate etag of final object
+	// To understand how etags are calculated, read more here:
+	// https://teppen.io/2018/06/23/aws_s3_etags/
+	let mut etag_md5_hasher = Md5::new();
+	for part in parts.iter() {
+		etag_md5_hasher.update(part.etag.as_ref().unwrap().as_bytes());
+	}
+	let etag = format!(
+		"{}-{}",
+		hex::encode(etag_md5_hasher.finalize()),
+		parts.len()
+	);
+
+	// 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 {
+		object_version.state = ObjectVersionState::Aborted;
+		let final_object = Object::new(bucket.id, key.clone(), vec![object_version]);
+		garage.object_table.insert(&final_object).await?;
+
+		return Err(e);
+	}
+
+	// Write final object version
+	object_version.state = ObjectVersionState::Complete(ObjectVersionData::FirstBlock(
+		ObjectVersionMeta {
+			headers,
+			size: total_size,
+			etag: etag.clone(),
+		},
+		final_version.blocks.items()[0].1.hash,
+	));
+
+	let final_object = Object::new(bucket.id, key.clone(), vec![object_version]);
+	garage.object_table.insert(&final_object).await?;
+
+	// Send response saying ok we're done
+	let result = s3_xml::CompleteMultipartUploadResult {
+		xmlns: (),
+		location: None,
+		bucket: s3_xml::Value(bucket_name.to_string()),
+		key: s3_xml::Value(key),
+		etag: s3_xml::Value(format!("\"{}\"", etag)),
+	};
+	let xml = s3_xml::to_xml_with_header(&result)?;
+
+	Ok(Response::new(Body::from(xml.into_bytes())))
+}
+
+pub async fn handle_abort_multipart_upload(
+	garage: Arc<Garage>,
+	bucket_id: Uuid,
+	key: &str,
+	upload_id: &str,
+) -> Result<Response<Body>, Error> {
+	let upload_id = decode_upload_id(upload_id)?;
+
+	let (_, mut object_version, _) =
+		get_upload(&garage, &bucket_id, &key.to_string(), &upload_id).await?;
+
+	object_version.state = ObjectVersionState::Aborted;
+	let final_object = Object::new(bucket_id, key.to_string(), vec![object_version]);
+	garage.object_table.insert(&final_object).await?;
+
+	Ok(Response::new(Body::from(vec![])))
+}
+
+// ======== helpers ============
+
+#[allow(clippy::ptr_arg)]
+pub(crate) async fn get_upload(
+	garage: &Garage,
+	bucket_id: &Uuid,
+	key: &String,
+	upload_id: &Uuid,
+) -> Result<(Object, ObjectVersion, MultipartUpload), Error> {
+	let (object, mpu) = futures::try_join!(
+		garage.object_table.get(bucket_id, key).map_err(Error::from),
+		garage
+			.mpu_table
+			.get(upload_id, &EmptyKey)
+			.map_err(Error::from),
+	)?;
+
+	let object = object.ok_or(Error::NoSuchUpload)?;
+	let mpu = mpu.ok_or(Error::NoSuchUpload)?;
+
+	let object_version = object
+		.versions()
+		.iter()
+		.find(|v| v.uuid == *upload_id && v.is_uploading(Some(true)))
+		.ok_or(Error::NoSuchUpload)?
+		.clone();
+
+	Ok((object, object_version, mpu))
+}
+
+pub fn decode_upload_id(id: &str) -> Result<Uuid, Error> {
+	let id_bin = hex::decode(id).map_err(|_| Error::NoSuchUpload)?;
+	if id_bin.len() != 32 {
+		return Err(Error::NoSuchUpload);
+	}
+	let mut uuid = [0u8; 32];
+	uuid.copy_from_slice(&id_bin[..]);
+	Ok(Uuid::from(uuid))
+}
+
+#[derive(Debug)]
+struct CompleteMultipartUploadPart {
+	etag: String,
+	part_number: u64,
+}
+
+fn parse_complete_multipart_upload_body(
+	xml: &roxmltree::Document,
+) -> Option<Vec<CompleteMultipartUploadPart>> {
+	let mut parts = vec![];
+
+	let root = xml.root();
+	let cmu = root.first_child()?;
+	if !cmu.has_tag_name("CompleteMultipartUpload") {
+		return None;
+	}
+
+	for item in cmu.children() {
+		// Only parse <Part> nodes
+		if !item.is_element() {
+			continue;
+		}
+
+		if item.has_tag_name("Part") {
+			let etag = item.children().find(|e| e.has_tag_name("ETag"))?.text()?;
+			let part_number = item
+				.children()
+				.find(|e| e.has_tag_name("PartNumber"))?
+				.text()?;
+			parts.push(CompleteMultipartUploadPart {
+				etag: etag.trim_matches('"').to_string(),
+				part_number: part_number.parse().ok()?,
+			});
+		} else {
+			return None;
+		}
+	}
+
+	Some(parts)
+}
diff --git a/src/api/s3/put.rs b/src/api/s3/put.rs
index 350ab884..c7ac5030 100644
--- a/src/api/s3/put.rs
+++ b/src/api/s3/put.rs
@@ -1,4 +1,4 @@
-use std::collections::{BTreeMap, BTreeSet, HashMap};
+use std::collections::{BTreeMap, HashMap};
 use std::sync::Arc;
 
 use base64::prelude::*;
@@ -30,8 +30,6 @@ use garage_model::s3::object_table::*;
 use garage_model::s3::version_table::*;
 
 use crate::s3::error::*;
-use crate::s3::xml as s3_xml;
-use crate::signature::verify_signed_content;
 
 pub async fn handle_put(
 	garage: Arc<Garage>,
@@ -123,20 +121,23 @@ pub(crate) async fn save_stream<S: Stream<Item = Result<Bytes, Error>> + Unpin>(
 	// The following consists in many steps that can each fail.
 	// Keep track that some cleanup will be needed if things fail
 	// before everything is finished (cleanup is done using the Drop trait).
-	let mut interrupted_cleanup = InterruptedCleanup(Some((
-		garage.clone(),
-		bucket.id,
-		key.into(),
+	let mut interrupted_cleanup = InterruptedCleanup(Some(InterruptedCleanupInner {
+		garage: garage.clone(),
+		bucket_id: bucket.id,
+		key: key.into(),
 		version_uuid,
 		version_timestamp,
-	)));
+	}));
 
 	// Write version identifier in object table so that we have a trace
 	// that we are uploading something
 	let mut object_version = ObjectVersion {
 		uuid: version_uuid,
 		timestamp: version_timestamp,
-		state: ObjectVersionState::Uploading(headers.clone()),
+		state: ObjectVersionState::Uploading {
+			headers: headers.clone(),
+			multipart: false,
+		},
 	};
 	let object = Object::new(bucket.id, key.into(), vec![object_version.clone()]);
 	garage.object_table.insert(&object).await?;
@@ -145,7 +146,14 @@ pub(crate) async fn save_stream<S: Stream<Item = Result<Bytes, Error>> + Unpin>(
 	// Write this entry now, even with empty block list,
 	// to prevent block_ref entries from being deleted (they can be deleted
 	// if the reference a version that isn't found in the version table)
-	let version = Version::new(version_uuid, bucket.id, key.into(), false);
+	let version = Version::new(
+		version_uuid,
+		VersionBacklink::Object {
+			bucket_id: bucket.id,
+			key: key.into(),
+		},
+		false,
+	);
 	garage.version_table.insert(&version).await?;
 
 	// Transfer data and verify checksum
@@ -192,7 +200,7 @@ pub(crate) async fn save_stream<S: Stream<Item = Result<Bytes, Error>> + Unpin>(
 
 /// Validate MD5 sum against content-md5 header
 /// and sha256sum against signed content-sha256
-fn ensure_checksum_matches(
+pub(crate) fn ensure_checksum_matches(
 	data_md5sum: &[u8],
 	data_sha256sum: garage_util::data::FixedBytes32,
 	content_md5: Option<&str>,
@@ -218,7 +226,7 @@ fn ensure_checksum_matches(
 }
 
 /// Check that inserting this object with this size doesn't exceed bucket quotas
-async fn check_quotas(
+pub(crate) async fn check_quotas(
 	garage: &Arc<Garage>,
 	bucket: &Bucket,
 	key: &str,
@@ -275,7 +283,7 @@ async fn check_quotas(
 	Ok(())
 }
 
-async fn read_and_put_blocks<S: Stream<Item = Result<Bytes, Error>> + Unpin>(
+pub(crate) async fn read_and_put_blocks<S: Stream<Item = Result<Bytes, Error>> + Unpin>(
 	garage: &Garage,
 	version: &Version,
 	part_number: u64,
@@ -381,7 +389,7 @@ async fn put_block_meta(
 	Ok(())
 }
 
-struct StreamChunker<S: Stream<Item = Result<Bytes, Error>>> {
+pub(crate) struct StreamChunker<S: Stream<Item = Result<Bytes, Error>>> {
 	stream: S,
 	read_all: bool,
 	block_size: usize,
@@ -389,7 +397,7 @@ struct StreamChunker<S: Stream<Item = Result<Bytes, Error>>> {
 }
 
 impl<S: Stream<Item = Result<Bytes, Error>> + Unpin> StreamChunker<S> {
-	fn new(stream: S, block_size: usize) -> Self {
+	pub(crate) fn new(stream: S, block_size: usize) -> Self {
 		Self {
 			stream,
 			read_all: false,
@@ -398,7 +406,7 @@ impl<S: Stream<Item = Result<Bytes, Error>> + Unpin> StreamChunker<S> {
 		}
 	}
 
-	async fn next(&mut self) -> Result<Option<Bytes>, Error> {
+	pub(crate) async fn next(&mut self) -> Result<Option<Bytes>, Error> {
 		while !self.read_all && self.buf.len() < self.block_size {
 			if let Some(block) = self.stream.next().await {
 				let bytes = block?;
@@ -425,7 +433,14 @@ pub fn put_response(version_uuid: Uuid, md5sum_hex: String) -> Response<Body> {
 		.unwrap()
 }
 
-struct InterruptedCleanup(Option<(Arc<Garage>, Uuid, String, Uuid, u64)>);
+struct InterruptedCleanup(Option<InterruptedCleanupInner>);
+struct InterruptedCleanupInner {
+	garage: Arc<Garage>,
+	bucket_id: Uuid,
+	key: String,
+	version_uuid: Uuid,
+	version_timestamp: u64,
+}
 
 impl InterruptedCleanup {
 	fn cancel(&mut self) {
@@ -434,15 +449,15 @@ impl InterruptedCleanup {
 }
 impl Drop for InterruptedCleanup {
 	fn drop(&mut self) {
-		if let Some((garage, bucket_id, key, version_uuid, version_ts)) = self.0.take() {
+		if let Some(info) = self.0.take() {
 			tokio::spawn(async move {
 				let object_version = ObjectVersion {
-					uuid: version_uuid,
-					timestamp: version_ts,
+					uuid: info.version_uuid,
+					timestamp: info.version_timestamp,
 					state: ObjectVersionState::Aborted,
 				};
-				let object = Object::new(bucket_id, key, vec![object_version]);
-				if let Err(e) = garage.object_table.insert(&object).await {
+				let object = Object::new(info.bucket_id, info.key, vec![object_version]);
+				if let Err(e) = info.garage.object_table.insert(&object).await {
 					warn!("Cannot cleanup after aborted PutObject: {}", e);
 				}
 			});
@@ -450,326 +465,9 @@ impl Drop for InterruptedCleanup {
 	}
 }
 
-// ----
-
-pub async fn handle_create_multipart_upload(
-	garage: Arc<Garage>,
-	req: &Request<Body>,
-	bucket_name: &str,
-	bucket_id: Uuid,
-	key: &str,
-) -> Result<Response<Body>, Error> {
-	let version_uuid = gen_uuid();
-	let headers = get_headers(req.headers())?;
-
-	// Create object in object table
-	let object_version = ObjectVersion {
-		uuid: version_uuid,
-		timestamp: now_msec(),
-		state: ObjectVersionState::Uploading(headers),
-	};
-	let object = Object::new(bucket_id, key.to_string(), vec![object_version]);
-	garage.object_table.insert(&object).await?;
-
-	// Insert empty version so that block_ref entries refer to something
-	// (they are inserted concurrently with blocks in the version table, so
-	// there is the possibility that they are inserted before the version table
-	// is created, in which case it is allowed to delete them, e.g. in repair_*)
-	let version = Version::new(version_uuid, bucket_id, key.into(), false);
-	garage.version_table.insert(&version).await?;
-
-	// Send success response
-	let result = s3_xml::InitiateMultipartUploadResult {
-		xmlns: (),
-		bucket: s3_xml::Value(bucket_name.to_string()),
-		key: s3_xml::Value(key.to_string()),
-		upload_id: s3_xml::Value(hex::encode(version_uuid)),
-	};
-	let xml = s3_xml::to_xml_with_header(&result)?;
-
-	Ok(Response::new(Body::from(xml.into_bytes())))
-}
-
-pub async fn handle_put_part(
-	garage: Arc<Garage>,
-	req: Request<Body>,
-	bucket_id: Uuid,
-	key: &str,
-	part_number: u64,
-	upload_id: &str,
-	content_sha256: Option<Hash>,
-) -> Result<Response<Body>, Error> {
-	let version_uuid = decode_upload_id(upload_id)?;
-
-	let content_md5 = match req.headers().get("content-md5") {
-		Some(x) => Some(x.to_str()?.to_string()),
-		None => None,
-	};
-
-	// Read first chuck, and at the same time try to get object to see if it exists
-	let key = key.to_string();
-
-	let body = req.into_body().map_err(Error::from);
-	let mut chunker = StreamChunker::new(body, garage.config.block_size);
-
-	let (object, version, first_block) = futures::try_join!(
-		garage
-			.object_table
-			.get(&bucket_id, &key)
-			.map_err(Error::from),
-		garage
-			.version_table
-			.get(&version_uuid, &EmptyKey)
-			.map_err(Error::from),
-		chunker.next(),
-	)?;
-
-	// Check object is valid and multipart block can be accepted
-	let first_block = first_block.ok_or_bad_request("Empty body")?;
-	let object = object.ok_or_bad_request("Object not found")?;
+// ============ helpers ============
 
-	if !object
-		.versions()
-		.iter()
-		.any(|v| v.uuid == version_uuid && v.is_uploading())
-	{
-		return Err(Error::NoSuchUpload);
-	}
-
-	// Check part hasn't already been uploaded
-	if let Some(v) = version {
-		if v.has_part_number(part_number) {
-			return Err(Error::bad_request(format!(
-				"Part number {} has already been uploaded",
-				part_number
-			)));
-		}
-	}
-
-	// Copy block to store
-	let version = Version::new(version_uuid, bucket_id, key, false);
-
-	let first_block_hash = async_blake2sum(first_block.clone()).await;
-
-	let (_, data_md5sum, data_sha256sum) = read_and_put_blocks(
-		&garage,
-		&version,
-		part_number,
-		first_block,
-		first_block_hash,
-		&mut chunker,
-	)
-	.await?;
-
-	// Verify that checksums map
-	ensure_checksum_matches(
-		data_md5sum.as_slice(),
-		data_sha256sum,
-		content_md5.as_deref(),
-		content_sha256,
-	)?;
-
-	// Store part etag in version
-	let data_md5sum_hex = hex::encode(data_md5sum);
-	let mut version = version;
-	version
-		.parts_etags
-		.put(part_number, data_md5sum_hex.clone());
-	garage.version_table.insert(&version).await?;
-
-	let response = Response::builder()
-		.header("ETag", format!("\"{}\"", data_md5sum_hex))
-		.body(Body::empty())
-		.unwrap();
-	Ok(response)
-}
-
-pub async fn handle_complete_multipart_upload(
-	garage: Arc<Garage>,
-	req: Request<Body>,
-	bucket_name: &str,
-	bucket: &Bucket,
-	key: &str,
-	upload_id: &str,
-	content_sha256: Option<Hash>,
-) -> Result<Response<Body>, Error> {
-	let body = hyper::body::to_bytes(req.into_body()).await?;
-
-	if let Some(content_sha256) = content_sha256 {
-		verify_signed_content(content_sha256, &body[..])?;
-	}
-
-	let body_xml = roxmltree::Document::parse(std::str::from_utf8(&body)?)?;
-	let body_list_of_parts = parse_complete_multipart_upload_body(&body_xml)
-		.ok_or_bad_request("Invalid CompleteMultipartUpload XML")?;
-	debug!(
-		"CompleteMultipartUpload list of parts: {:?}",
-		body_list_of_parts
-	);
-
-	let version_uuid = decode_upload_id(upload_id)?;
-
-	// Get object and version
-	let key = key.to_string();
-	let (object, version) = futures::try_join!(
-		garage.object_table.get(&bucket.id, &key),
-		garage.version_table.get(&version_uuid, &EmptyKey),
-	)?;
-
-	let object = object.ok_or(Error::NoSuchKey)?;
-	let mut object_version = object
-		.versions()
-		.iter()
-		.find(|v| v.uuid == version_uuid && v.is_uploading())
-		.cloned()
-		.ok_or(Error::NoSuchUpload)?;
-
-	let version = version.ok_or(Error::NoSuchKey)?;
-	if version.blocks.is_empty() {
-		return Err(Error::bad_request("No data was uploaded"));
-	}
-
-	let headers = match object_version.state {
-		ObjectVersionState::Uploading(headers) => headers,
-		_ => unreachable!(),
-	};
-
-	// Check that part numbers are an increasing sequence.
-	// (it doesn't need to start at 1 nor to be a continuous sequence,
-	// see discussion in #192)
-	if body_list_of_parts.is_empty() {
-		return Err(Error::EntityTooSmall);
-	}
-	if !body_list_of_parts
-		.iter()
-		.zip(body_list_of_parts.iter().skip(1))
-		.all(|(p1, p2)| p1.part_number < p2.part_number)
-	{
-		return Err(Error::InvalidPartOrder);
-	}
-
-	// Garage-specific restriction, see #204: part numbers must be
-	// consecutive starting at 1
-	if body_list_of_parts[0].part_number != 1
-		|| !body_list_of_parts
-			.iter()
-			.zip(body_list_of_parts.iter().skip(1))
-			.all(|(p1, p2)| p1.part_number + 1 == p2.part_number)
-	{
-		return Err(Error::NotImplemented("Garage does not support completing a Multipart upload with non-consecutive part numbers. This is a restriction of Garage's data model, which might be fixed in a future release. See issue #204 for more information on this topic.".into()));
-	}
-
-	// Check that the list of parts they gave us corresponds to the parts we have here
-	debug!("Expected parts from request: {:?}", body_list_of_parts);
-	debug!("Parts stored in version: {:?}", version.parts_etags.items());
-	let parts = version
-		.parts_etags
-		.items()
-		.iter()
-		.map(|pair| (&pair.0, &pair.1));
-	let same_parts = body_list_of_parts
-		.iter()
-		.map(|x| (&x.part_number, &x.etag))
-		.eq(parts);
-	if !same_parts {
-		return Err(Error::InvalidPart);
-	}
-
-	// Check that all blocks belong to one of the parts
-	let block_parts = version
-		.blocks
-		.items()
-		.iter()
-		.map(|(bk, _)| bk.part_number)
-		.collect::<BTreeSet<_>>();
-	let same_parts = body_list_of_parts
-		.iter()
-		.map(|x| x.part_number)
-		.eq(block_parts.into_iter());
-	if !same_parts {
-		return Err(Error::bad_request(
-			"Part numbers in block list and part list do not match. This can happen if a part was partially uploaded. Please abort the multipart upload and try again."
-		));
-	}
-
-	// Calculate etag of final object
-	// To understand how etags are calculated, read more here:
-	// https://teppen.io/2018/06/23/aws_s3_etags/
-	let num_parts = body_list_of_parts.len();
-	let mut etag_md5_hasher = Md5::new();
-	for (_, etag) in version.parts_etags.items().iter() {
-		etag_md5_hasher.update(etag.as_bytes());
-	}
-	let etag = format!("{}-{}", hex::encode(etag_md5_hasher.finalize()), num_parts);
-
-	// Calculate total size of final object
-	let total_size = version.blocks.items().iter().map(|x| x.1.size).sum();
-
-	if let Err(e) = check_quotas(&garage, bucket, &key, total_size).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?;
-
-		return Err(e);
-	}
-
-	// Write final object version
-	object_version.state = ObjectVersionState::Complete(ObjectVersionData::FirstBlock(
-		ObjectVersionMeta {
-			headers,
-			size: total_size,
-			etag: etag.clone(),
-		},
-		version.blocks.items()[0].1.hash,
-	));
-
-	let final_object = Object::new(bucket.id, key.clone(), vec![object_version]);
-	garage.object_table.insert(&final_object).await?;
-
-	// Send response saying ok we're done
-	let result = s3_xml::CompleteMultipartUploadResult {
-		xmlns: (),
-		location: None,
-		bucket: s3_xml::Value(bucket_name.to_string()),
-		key: s3_xml::Value(key),
-		etag: s3_xml::Value(format!("\"{}\"", etag)),
-	};
-	let xml = s3_xml::to_xml_with_header(&result)?;
-
-	Ok(Response::new(Body::from(xml.into_bytes())))
-}
-
-pub async fn handle_abort_multipart_upload(
-	garage: Arc<Garage>,
-	bucket_id: Uuid,
-	key: &str,
-	upload_id: &str,
-) -> Result<Response<Body>, Error> {
-	let version_uuid = decode_upload_id(upload_id)?;
-
-	let object = garage
-		.object_table
-		.get(&bucket_id, &key.to_string())
-		.await?;
-	let object = object.ok_or(Error::NoSuchKey)?;
-
-	let object_version = object
-		.versions()
-		.iter()
-		.find(|v| v.uuid == version_uuid && v.is_uploading());
-	let mut object_version = match object_version {
-		None => return Err(Error::NoSuchUpload),
-		Some(x) => x.clone(),
-	};
-
-	object_version.state = ObjectVersionState::Aborted;
-	let final_object = Object::new(bucket_id, key.to_string(), vec![object_version]);
-	garage.object_table.insert(&final_object).await?;
-
-	Ok(Response::new(Body::from(vec![])))
-}
-
-fn get_mime_type(headers: &HeaderMap<HeaderValue>) -> Result<String, Error> {
+pub(crate) fn get_mime_type(headers: &HeaderMap<HeaderValue>) -> Result<String, Error> {
 	Ok(headers
 		.get(hyper::header::CONTENT_TYPE)
 		.map(|x| x.to_str())
@@ -821,54 +519,3 @@ pub(crate) fn get_headers(headers: &HeaderMap<HeaderValue>) -> Result<ObjectVers
 		other,
 	})
 }
-
-pub fn decode_upload_id(id: &str) -> Result<Uuid, Error> {
-	let id_bin = hex::decode(id).map_err(|_| Error::NoSuchUpload)?;
-	if id_bin.len() != 32 {
-		return Err(Error::NoSuchUpload);
-	}
-	let mut uuid = [0u8; 32];
-	uuid.copy_from_slice(&id_bin[..]);
-	Ok(Uuid::from(uuid))
-}
-
-#[derive(Debug)]
-struct CompleteMultipartUploadPart {
-	etag: String,
-	part_number: u64,
-}
-
-fn parse_complete_multipart_upload_body(
-	xml: &roxmltree::Document,
-) -> Option<Vec<CompleteMultipartUploadPart>> {
-	let mut parts = vec![];
-
-	let root = xml.root();
-	let cmu = root.first_child()?;
-	if !cmu.has_tag_name("CompleteMultipartUpload") {
-		return None;
-	}
-
-	for item in cmu.children() {
-		// Only parse <Part> nodes
-		if !item.is_element() {
-			continue;
-		}
-
-		if item.has_tag_name("Part") {
-			let etag = item.children().find(|e| e.has_tag_name("ETag"))?.text()?;
-			let part_number = item
-				.children()
-				.find(|e| e.has_tag_name("PartNumber"))?
-				.text()?;
-			parts.push(CompleteMultipartUploadPart {
-				etag: etag.trim_matches('"').to_string(),
-				part_number: part_number.parse().ok()?,
-			});
-		} else {
-			return None;
-		}
-	}
-
-	Some(parts)
-}
diff --git a/src/api/s3/website.rs b/src/api/s3/website.rs
index 77738971..7f2ab925 100644
--- a/src/api/s3/website.rs
+++ b/src/api/s3/website.rs
@@ -43,14 +43,11 @@ pub async fn handle_get_website(bucket: &Bucket) -> Result<Response<Body>, Error
 
 pub async fn handle_delete_website(
 	garage: Arc<Garage>,
-	bucket_id: Uuid,
+	mut bucket: Bucket,
 ) -> Result<Response<Body>, Error> {
-	let mut bucket = garage
-		.bucket_helper()
-		.get_existing_bucket(bucket_id)
-		.await?;
-
-	let param = bucket.params_mut().unwrap();
+	let param = bucket
+		.params_mut()
+		.ok_or_internal_error("Bucket should not be deleted at this point")?;
 
 	param.website_config.update(None);
 	garage.bucket_table.insert(&bucket).await?;
@@ -62,7 +59,7 @@ pub async fn handle_delete_website(
 
 pub async fn handle_put_website(
 	garage: Arc<Garage>,
-	bucket_id: Uuid,
+	mut bucket: Bucket,
 	req: Request<Body>,
 	content_sha256: Option<Hash>,
 ) -> Result<Response<Body>, Error> {
@@ -72,12 +69,9 @@ pub async fn handle_put_website(
 		verify_signed_content(content_sha256, &body[..])?;
 	}
 
-	let mut bucket = garage
-		.bucket_helper()
-		.get_existing_bucket(bucket_id)
-		.await?;
-
-	let param = bucket.params_mut().unwrap();
+	let param = bucket
+		.params_mut()
+		.ok_or_internal_error("Bucket should not be deleted at this point")?;
 
 	let conf: WebsiteConfiguration = from_reader(&body as &[u8])?;
 	conf.validate()?;
author	Alex <alex@adnab.me>	2023-10-10 13:28:28 +0000
committer	Alex <alex@adnab.me>	2023-10-10 13:28:28 +0000
commit	e6df7089a19d06696ac2178a4dae7ac9f27fd55e (patch)
tree	e1f7120d582f67fd104dc8e746d883befffb4daa /src/api
parent	e75fe2157d0973eb4320bbc87be5e54a1cb91a63 (diff)
parent	952c9570c494468643353ee1ae9052b510353665 (diff)
download	garage-e6df7089a19d06696ac2178a4dae7ac9f27fd55e.tar.gz garage-e6df7089a19d06696ac2178a4dae7ac9f27fd55e.zip