Merge pull request 'Add node-global lock for bucket/key operations (fix #723)' (#728) from lock-createbucket into main

Reviewed-on: https://git.deuxfleurs.fr/Deuxfleurs/garage/pulls/728

12 files changed, 511 insertions, 476 deletions

diff --git a/src/api/admin/bucket.rs b/src/api/admin/bucket.rs
index 1b22dd03..cfe8a6c4 100644
--- a/src/api/admin/bucket.rs
+++ b/src/api/admin/bucket.rs
@@ -273,6 +273,8 @@ pub async fn handle_create_bucket(
 ) -> Result<Response<ResBody>, Error> {
 	let req = parse_json_body::<CreateBucketRequest, _, Error>(req).await?;
 
+	let helper = garage.locked_helper().await;
+
 	if let Some(ga) = &req.global_alias {
 		if !is_valid_bucket_name(ga) {
 			return Err(Error::bad_request(format!(
@@ -296,10 +298,7 @@ pub async fn handle_create_bucket(
 			)));
 		}
 
-		let key = garage
-			.key_helper()
-			.get_existing_key(&la.access_key_id)
-			.await?;
+		let key = helper.key().get_existing_key(&la.access_key_id).await?;
 		let state = key.state.as_option().unwrap();
 		if matches!(state.local_aliases.get(&la.alias), Some(_)) {
 			return Err(Error::bad_request("Local alias already exists"));
@@ -310,21 +309,16 @@ pub async fn handle_create_bucket(
 	garage.bucket_table.insert(&bucket).await?;
 
 	if let Some(ga) = &req.global_alias {
-		garage
-			.bucket_helper()
-			.set_global_bucket_alias(bucket.id, ga)
-			.await?;
+		helper.set_global_bucket_alias(bucket.id, ga).await?;
 	}
 
 	if let Some(la) = &req.local_alias {
-		garage
-			.bucket_helper()
+		helper
 			.set_local_bucket_alias(bucket.id, &la.access_key_id, &la.alias)
 			.await?;
 
 		if la.allow.read || la.allow.write || la.allow.owner {
-			garage
-				.bucket_helper()
+			helper
 				.set_bucket_key_permissions(
 					bucket.id,
 					&la.access_key_id,
@@ -362,15 +356,15 @@ pub async fn handle_delete_bucket(
 	garage: &Arc<Garage>,
 	id: String,
 ) -> Result<Response<ResBody>, Error> {
-	let helper = garage.bucket_helper();
+	let helper = garage.locked_helper().await;
 
 	let bucket_id = parse_bucket_id(&id)?;
 
-	let mut bucket = helper.get_existing_bucket(bucket_id).await?;
+	let mut bucket = helper.bucket().get_existing_bucket(bucket_id).await?;
 	let state = bucket.state.as_option().unwrap();
 
 	// Check bucket is empty
-	if !helper.is_bucket_empty(bucket_id).await? {
+	if !helper.bucket().is_bucket_empty(bucket_id).await? {
 		return Err(CommonError::BucketNotEmpty.into());
 	}
 
@@ -476,18 +470,14 @@ pub async fn handle_bucket_change_key_perm(
 ) -> Result<Response<ResBody>, Error> {
 	let req = parse_json_body::<BucketKeyPermChangeRequest, _, Error>(req).await?;
 
+	let helper = garage.locked_helper().await;
+
 	let bucket_id = parse_bucket_id(&req.bucket_id)?;
 
-	let bucket = garage
-		.bucket_helper()
-		.get_existing_bucket(bucket_id)
-		.await?;
+	let bucket = helper.bucket().get_existing_bucket(bucket_id).await?;
 	let state = bucket.state.as_option().unwrap();
 
-	let key = garage
-		.key_helper()
-		.get_existing_key(&req.access_key_id)
-		.await?;
+	let key = helper.key().get_existing_key(&req.access_key_id).await?;
 
 	let mut perm = state
 		.authorized_keys
@@ -505,8 +495,7 @@ pub async fn handle_bucket_change_key_perm(
 		perm.allow_owner = new_perm_flag;
 	}
 
-	garage
-		.bucket_helper()
+	helper
 		.set_bucket_key_permissions(bucket.id, &key.key_id, perm)
 		.await?;
 
@@ -530,10 +519,9 @@ pub async fn handle_global_alias_bucket(
 ) -> Result<Response<ResBody>, Error> {
 	let bucket_id = parse_bucket_id(&bucket_id)?;
 
-	garage
-		.bucket_helper()
-		.set_global_bucket_alias(bucket_id, &alias)
-		.await?;
+	let helper = garage.locked_helper().await;
+
+	helper.set_global_bucket_alias(bucket_id, &alias).await?;
 
 	bucket_info_results(garage, bucket_id).await
 }
@@ -545,10 +533,9 @@ pub async fn handle_global_unalias_bucket(
 ) -> Result<Response<ResBody>, Error> {
 	let bucket_id = parse_bucket_id(&bucket_id)?;
 
-	garage
-		.bucket_helper()
-		.unset_global_bucket_alias(bucket_id, &alias)
-		.await?;
+	let helper = garage.locked_helper().await;
+
+	helper.unset_global_bucket_alias(bucket_id, &alias).await?;
 
 	bucket_info_results(garage, bucket_id).await
 }
@@ -561,8 +548,9 @@ pub async fn handle_local_alias_bucket(
 ) -> Result<Response<ResBody>, Error> {
 	let bucket_id = parse_bucket_id(&bucket_id)?;
 
-	garage
-		.bucket_helper()
+	let helper = garage.locked_helper().await;
+
+	helper
 		.set_local_bucket_alias(bucket_id, &access_key_id, &alias)
 		.await?;
 
@@ -577,8 +565,9 @@ pub async fn handle_local_unalias_bucket(
 ) -> Result<Response<ResBody>, Error> {
 	let bucket_id = parse_bucket_id(&bucket_id)?;
 
-	garage
-		.bucket_helper()
+	let helper = garage.locked_helper().await;
+
+	helper
 		.unset_local_bucket_alias(bucket_id, &access_key_id, &alias)
 		.await?;
 
diff --git a/src/api/admin/key.rs b/src/api/admin/key.rs
index 1efaca16..291b6d54 100644
--- a/src/api/admin/key.rs
+++ b/src/api/admin/key.rs
@@ -151,11 +151,11 @@ pub async fn handle_delete_key(
 	garage: &Arc<Garage>,
 	id: String,
 ) -> Result<Response<ResBody>, Error> {
-	let mut key = garage.key_helper().get_existing_key(&id).await?;
+	let helper = garage.locked_helper().await;
 
-	key.state.as_option().unwrap();
+	let mut key = helper.key().get_existing_key(&id).await?;
 
-	garage.key_helper().delete_key(&mut key).await?;
+	helper.delete_key(&mut key).await?;
 
 	Ok(Response::builder()
 		.status(StatusCode::NO_CONTENT)
diff --git a/src/api/s3/api_server.rs b/src/api/s3/api_server.rs
index 7fac6261..08405923 100644
--- a/src/api/s3/api_server.rs
+++ b/src/api/s3/api_server.rs
@@ -148,7 +148,14 @@ impl ApiHandler for S3ApiServer {
 
 		// Special code path for CreateBucket API endpoint
 		if let Endpoint::CreateBucket {} = endpoint {
-			return handle_create_bucket(&garage, req, content_sha256, api_key, bucket_name).await;
+			return handle_create_bucket(
+				&garage,
+				req,
+				content_sha256,
+				&api_key.key_id,
+				bucket_name,
+			)
+			.await;
 		}
 
 		let bucket_id = garage
@@ -261,7 +268,7 @@ impl ApiHandler for S3ApiServer {
 				Ok(response)
 			}
 			Endpoint::DeleteBucket {} => {
-				handle_delete_bucket(&garage, bucket_id, bucket_name, api_key).await
+				handle_delete_bucket(&garage, bucket_id, bucket_name, &api_key.key_id).await
 			}
 			Endpoint::GetBucketLocation {} => handle_get_bucket_location(garage),
 			Endpoint::GetBucketVersioning {} => handle_get_bucket_versioning(),
diff --git a/src/api/s3/bucket.rs b/src/api/s3/bucket.rs
index fa2f1b6d..fa337566 100644
--- a/src/api/s3/bucket.rs
+++ b/src/api/s3/bucket.rs
@@ -122,7 +122,7 @@ pub async fn handle_create_bucket(
 	garage: &Garage,
 	req: Request<ReqBody>,
 	content_sha256: Option<Hash>,
-	api_key: Key,
+	api_key_id: &String,
 	bucket_name: String,
 ) -> Result<Response<ResBody>, Error> {
 	let body = BodyExt::collect(req.into_body()).await?.to_bytes();
@@ -144,16 +144,18 @@ pub async fn handle_create_bucket(
 		}
 	}
 
-	let key_params = api_key
-		.params()
-		.ok_or_internal_error("Key should not be deleted at this point")?;
+	let helper = garage.locked_helper().await;
+
+	// refetch API key after taking lock to ensure up-to-date data
+	let api_key = helper.key().get_existing_key(api_key_id).await?;
+	let key_params = api_key.params().unwrap();
 
 	let existing_bucket = if let Some(Some(bucket_id)) = key_params.local_aliases.get(&bucket_name)
 	{
 		Some(*bucket_id)
 	} else {
-		garage
-			.bucket_helper()
+		helper
+			.bucket()
 			.resolve_global_bucket_name(&bucket_name)
 			.await?
 	};
@@ -187,13 +189,11 @@ pub async fn handle_create_bucket(
 		let bucket = Bucket::new();
 		garage.bucket_table.insert(&bucket).await?;
 
-		garage
-			.bucket_helper()
+		helper
 			.set_bucket_key_permissions(bucket.id, &api_key.key_id, BucketKeyPerm::ALL_PERMISSIONS)
 			.await?;
 
-		garage
-			.bucket_helper()
+		helper
 			.set_local_bucket_alias(bucket.id, &api_key.key_id, &bucket_name)
 			.await?;
 	}
@@ -208,18 +208,16 @@ pub async fn handle_delete_bucket(
 	garage: &Garage,
 	bucket_id: Uuid,
 	bucket_name: String,
-	api_key: Key,
+	api_key_id: &String,
 ) -> Result<Response<ResBody>, Error> {
-	let key_params = api_key
-		.params()
-		.ok_or_internal_error("Key should not be deleted at this point")?;
+	let helper = garage.locked_helper().await;
+
+	let api_key = helper.key().get_existing_key(api_key_id).await?;
+	let key_params = api_key.params().unwrap();
 
 	let is_local_alias = matches!(key_params.local_aliases.get(&bucket_name), Some(Some(_)));
 
-	let mut bucket = garage
-		.bucket_helper()
-		.get_existing_bucket(bucket_id)
-		.await?;
+	let mut bucket = helper.bucket().get_existing_bucket(bucket_id).await?;
 	let bucket_state = bucket.state.as_option().unwrap();
 
 	// If the bucket has no other aliases, this is a true deletion.
@@ -243,28 +241,25 @@ pub async fn handle_delete_bucket(
 		// Delete bucket
 
 		// Check bucket is empty
-		if !garage.bucket_helper().is_bucket_empty(bucket_id).await? {
+		if !helper.bucket().is_bucket_empty(bucket_id).await? {
 			return Err(CommonError::BucketNotEmpty.into());
 		}
 
 		// --- done checking, now commit ---
 		// 1. delete bucket alias
 		if is_local_alias {
-			garage
-				.bucket_helper()
+			helper
 				.unset_local_bucket_alias(bucket_id, &api_key.key_id, &bucket_name)
 				.await?;
 		} else {
-			garage
-				.bucket_helper()
+			helper
 				.unset_global_bucket_alias(bucket_id, &bucket_name)
 				.await?;
 		}
 
 		// 2. delete authorization from keys that had access
 		for (key_id, _) in bucket.authorized_keys() {
-			garage
-				.bucket_helper()
+			helper
 				.set_bucket_key_permissions(bucket.id, key_id, BucketKeyPerm::NO_PERMISSIONS)
 				.await?;
 		}
@@ -274,14 +269,12 @@ pub async fn handle_delete_bucket(
 		garage.bucket_table.insert(&bucket).await?;
 	} else if is_local_alias {
 		// Just unalias
-		garage
-			.bucket_helper()
+		helper
 			.unset_local_bucket_alias(bucket_id, &api_key.key_id, &bucket_name)
 			.await?;
 	} else {
 		// Just unalias (but from global namespace)
-		garage
-			.bucket_helper()
+		helper
 			.unset_global_bucket_alias(bucket_id, &bucket_name)
 			.await?;
 	}
diff --git a/src/garage/admin/bucket.rs b/src/garage/admin/bucket.rs
index 0781cb8b..803b55bd 100644
--- a/src/garage/admin/bucket.rs
+++ b/src/garage/admin/bucket.rs
@@ -134,6 +134,8 @@ impl AdminRpcHandler {
 			)));
 		}
 
+		let helper = self.garage.locked_helper().await;
+
 		if let Some(alias) = self.garage.bucket_alias_table.get(&EmptyKey, name).await? {
 			if alias.state.get().is_some() {
 				return Err(Error::BadRequest(format!("Bucket {} already exists", name)));
@@ -145,18 +147,16 @@ impl AdminRpcHandler {
 		let bucket = Bucket::new();
 		self.garage.bucket_table.insert(&bucket).await?;
 
-		self.garage
-			.bucket_helper()
-			.set_global_bucket_alias(bucket.id, name)
-			.await?;
+		helper.set_global_bucket_alias(bucket.id, name).await?;
 
 		Ok(AdminRpc::Ok(format!("Bucket {} was created.", name)))
 	}
 
 	async fn handle_delete_bucket(&self, query: &DeleteBucketOpt) -> Result<AdminRpc, Error> {
-		let helper = self.garage.bucket_helper();
+		let helper = self.garage.locked_helper().await;
 
 		let bucket_id = helper
+			.bucket()
 			.resolve_global_bucket_name(&query.name)
 			.await?
 			.ok_or_bad_request("Bucket not found")?;
@@ -174,7 +174,7 @@ impl AdminRpcHandler {
 			.await?;
 
 		// Check bucket doesn't have other aliases
-		let mut bucket = helper.get_existing_bucket(bucket_id).await?;
+		let mut bucket = helper.bucket().get_existing_bucket(bucket_id).await?;
 		let bucket_state = bucket.state.as_option().unwrap();
 		if bucket_state
 			.aliases
@@ -195,7 +195,7 @@ impl AdminRpcHandler {
 		}
 
 		// Check bucket is empty
-		if !helper.is_bucket_empty(bucket_id).await? {
+		if !helper.bucket().is_bucket_empty(bucket_id).await? {
 			return Err(Error::BadRequest(format!(
 				"Bucket {} is not empty",
 				query.name
@@ -231,16 +231,16 @@ impl AdminRpcHandler {
 	}
 
 	async fn handle_alias_bucket(&self, query: &AliasBucketOpt) -> Result<AdminRpc, Error> {
-		let helper = self.garage.bucket_helper();
-		let key_helper = self.garage.key_helper();
+		let helper = self.garage.locked_helper().await;
 
 		let bucket_id = helper
+			.bucket()
 			.resolve_global_bucket_name(&query.existing_bucket)
 			.await?
 			.ok_or_bad_request("Bucket not found")?;
 
 		if let Some(key_pattern) = &query.local {
-			let key = key_helper.get_existing_matching_key(key_pattern).await?;
+			let key = helper.key().get_existing_matching_key(key_pattern).await?;
 
 			helper
 				.set_local_bucket_alias(bucket_id, &key.key_id, &query.new_name)
@@ -261,11 +261,10 @@ impl AdminRpcHandler {
 	}
 
 	async fn handle_unalias_bucket(&self, query: &UnaliasBucketOpt) -> Result<AdminRpc, Error> {
-		let helper = self.garage.bucket_helper();
-		let key_helper = self.garage.key_helper();
+		let helper = self.garage.locked_helper().await;
 
 		if let Some(key_pattern) = &query.local {
-			let key = key_helper.get_existing_matching_key(key_pattern).await?;
+			let key = helper.key().get_existing_matching_key(key_pattern).await?;
 
 			let bucket_id = key
 				.state
@@ -287,6 +286,7 @@ impl AdminRpcHandler {
 			)))
 		} else {
 			let bucket_id = helper
+				.bucket()
 				.resolve_global_bucket_name(&query.name)
 				.await?
 				.ok_or_bad_request("Bucket not found")?;
@@ -303,14 +303,15 @@ impl AdminRpcHandler {
 	}
 
 	async fn handle_bucket_allow(&self, query: &PermBucketOpt) -> Result<AdminRpc, Error> {
-		let helper = self.garage.bucket_helper();
-		let key_helper = self.garage.key_helper();
+		let helper = self.garage.locked_helper().await;
 
 		let bucket_id = helper
+			.bucket()
 			.resolve_global_bucket_name(&query.bucket)
 			.await?
 			.ok_or_bad_request("Bucket not found")?;
-		let key = key_helper
+		let key = helper
+			.key()
 			.get_existing_matching_key(&query.key_pattern)
 			.await?;
 
@@ -338,14 +339,15 @@ impl AdminRpcHandler {
 	}
 
 	async fn handle_bucket_deny(&self, query: &PermBucketOpt) -> Result<AdminRpc, Error> {
-		let helper = self.garage.bucket_helper();
-		let key_helper = self.garage.key_helper();
+		let helper = self.garage.locked_helper().await;
 
 		let bucket_id = helper
+			.bucket()
 			.resolve_global_bucket_name(&query.bucket)
 			.await?
 			.ok_or_bad_request("Bucket not found")?;
-		let key = key_helper
+		let key = helper
+			.key()
 			.get_existing_matching_key(&query.key_pattern)
 			.await?;
 
diff --git a/src/garage/admin/key.rs b/src/garage/admin/key.rs
index 1c92670c..bd010d2c 100644
--- a/src/garage/admin/key.rs
+++ b/src/garage/admin/key.rs
@@ -76,9 +76,10 @@ impl AdminRpcHandler {
 	}
 
 	async fn handle_delete_key(&self, query: &KeyDeleteOpt) -> Result<AdminRpc, Error> {
-		let key_helper = self.garage.key_helper();
+		let helper = self.garage.locked_helper().await;
 
-		let mut key = key_helper
+		let mut key = helper
+			.key()
 			.get_existing_matching_key(&query.key_pattern)
 			.await?;
 
@@ -88,7 +89,7 @@ impl AdminRpcHandler {
 			));
 		}
 
-		key_helper.delete_key(&mut key).await?;
+		helper.delete_key(&mut key).await?;
 
 		Ok(AdminRpc::Ok(format!(
 			"Key {} was deleted successfully.",
diff --git a/src/model/garage.rs b/src/model/garage.rs
index c10d64c5..fe38a760 100644
--- a/src/model/garage.rs
+++ b/src/model/garage.rs
@@ -56,6 +56,9 @@ pub struct Garage {
 	/// Table containing api keys
 	pub key_table: Arc<Table<KeyTable, TableFullReplication>>,
 
+	/// Lock to prevent concurrent modification of buckets and access keys
+	bucket_lock: tokio::sync::Mutex<()>,
+
 	/// Table containing S3 objects
 	pub object_table: Arc<Table<ObjectTable, TableShardedReplication>>,
 	/// Counting table containing object counters
@@ -343,6 +346,7 @@ impl Garage {
 			bucket_table,
 			bucket_alias_table,
 			key_table,
+			bucket_lock: tokio::sync::Mutex::new(()),
 			object_table,
 			object_counter_table,
 			mpu_table,
@@ -385,6 +389,11 @@ impl Garage {
 	pub fn key_helper(&self) -> helper::key::KeyHelper {
 		helper::key::KeyHelper(self)
 	}
+
+	pub async fn locked_helper(&self) -> helper::locked::LockedHelper {
+		let lock = self.bucket_lock.lock().await;
+		helper::locked::LockedHelper(self, lock)
+	}
 }
 
 #[cfg(feature = "k2v")]
diff --git a/src/model/helper/bucket.rs b/src/model/helper/bucket.rs
index e9842a91..f4e669c3 100644
--- a/src/model/helper/bucket.rs
+++ b/src/model/helper/bucket.rs
@@ -1,19 +1,15 @@
 use std::time::Duration;
 
-use garage_util::crdt::*;
 use garage_util::data::*;
-use garage_util::error::{Error as GarageError, OkOrMessage};
+use garage_util::error::OkOrMessage;
 use garage_util::time::*;
 
 use garage_table::util::*;
 
-use crate::bucket_alias_table::*;
 use crate::bucket_table::*;
 use crate::garage::Garage;
 use crate::helper::error::*;
-use crate::helper::key::KeyHelper;
 use crate::key_table::*;
-use crate::permission::BucketKeyPerm;
 use crate::s3::object_table::*;
 
 pub struct BucketHelper<'a>(pub(crate) &'a Garage);
@@ -96,341 +92,7 @@ impl<'a> BucketHelper<'a> {
 			.ok_or_else(|| Error::NoSuchBucket(hex::encode(bucket_id)))
 	}
 
-	/// Sets a new alias for a bucket in global namespace.
-	/// This function fails if:
-	/// - alias name is not valid according to S3 spec
-	/// - bucket does not exist or is deleted
-	/// - alias already exists and points to another bucket
-	pub async fn set_global_bucket_alias(
-		&self,
-		bucket_id: Uuid,
-		alias_name: &String,
-	) -> Result<(), Error> {
-		if !is_valid_bucket_name(alias_name) {
-			return Err(Error::InvalidBucketName(alias_name.to_string()));
-		}
-
-		let mut bucket = self.get_existing_bucket(bucket_id).await?;
-
-		let alias = self.0.bucket_alias_table.get(&EmptyKey, alias_name).await?;
-
-		if let Some(existing_alias) = alias.as_ref() {
-			if let Some(p_bucket) = existing_alias.state.get() {
-				if *p_bucket != bucket_id {
-					return Err(Error::BadRequest(format!(
-						"Alias {} already exists and points to different bucket: {:?}",
-						alias_name, p_bucket
-					)));
-				}
-			}
-		}
-
-		// Checks ok, add alias
-		let bucket_p = bucket.state.as_option_mut().unwrap();
-
-		let alias_ts = increment_logical_clock_2(
-			bucket_p.aliases.get_timestamp(alias_name),
-			alias.as_ref().map(|a| a.state.timestamp()).unwrap_or(0),
-		);
-
-		// ---- timestamp-ensured causality barrier ----
-		// writes are now done and all writes use timestamp alias_ts
-
-		let alias = match alias {
-			None => BucketAlias::new(alias_name.clone(), alias_ts, Some(bucket_id))
-				.ok_or_else(|| Error::InvalidBucketName(alias_name.clone()))?,
-			Some(mut a) => {
-				a.state = Lww::raw(alias_ts, Some(bucket_id));
-				a
-			}
-		};
-		self.0.bucket_alias_table.insert(&alias).await?;
-
-		bucket_p.aliases = LwwMap::raw_item(alias_name.clone(), alias_ts, true);
-		self.0.bucket_table.insert(&bucket).await?;
-
-		Ok(())
-	}
-
-	/// Unsets an alias for a bucket in global namespace.
-	/// This function fails if:
-	/// - bucket does not exist or is deleted
-	/// - alias does not exist or maps to another bucket (-> internal error)
-	/// - bucket has no other aliases (global or local)
-	pub async fn unset_global_bucket_alias(
-		&self,
-		bucket_id: Uuid,
-		alias_name: &String,
-	) -> Result<(), Error> {
-		let mut bucket = self.get_existing_bucket(bucket_id).await?;
-		let bucket_state = bucket.state.as_option_mut().unwrap();
-
-		let mut alias = self
-			.0
-			.bucket_alias_table
-			.get(&EmptyKey, alias_name)
-			.await?
-			.filter(|a| a.state.get().map(|x| x == bucket_id).unwrap_or(false))
-			.ok_or_message(format!(
-				"Internal error: alias not found or does not point to bucket {:?}",
-				bucket_id
-			))?;
-
-		let has_other_global_aliases = bucket_state
-			.aliases
-			.items()
-			.iter()
-			.any(|(name, _, active)| name != alias_name && *active);
-		let has_other_local_aliases = bucket_state
-			.local_aliases
-			.items()
-			.iter()
-			.any(|(_, _, active)| *active);
-		if !has_other_global_aliases && !has_other_local_aliases {
-			return Err(Error::BadRequest(format!("Bucket {} doesn't have other aliases, please delete it instead of just unaliasing.", alias_name)));
-		}
-
-		// Checks ok, remove alias
-		let alias_ts = increment_logical_clock_2(
-			alias.state.timestamp(),
-			bucket_state.aliases.get_timestamp(alias_name),
-		);
-
-		// ---- timestamp-ensured causality barrier ----
-		// writes are now done and all writes use timestamp alias_ts
-
-		alias.state = Lww::raw(alias_ts, None);
-		self.0.bucket_alias_table.insert(&alias).await?;
-
-		bucket_state.aliases = LwwMap::raw_item(alias_name.clone(), alias_ts, false);
-		self.0.bucket_table.insert(&bucket).await?;
-
-		Ok(())
-	}
-
-	/// Ensures a bucket does not have a certain global alias.
-	/// Contrarily to unset_global_bucket_alias, this does not
-	/// fail on any condition other than:
-	/// - bucket cannot be found (its fine if it is in deleted state)
-	/// - alias cannot be found (its fine if it points to nothing or
-	///   to another bucket)
-	pub async fn purge_global_bucket_alias(
-		&self,
-		bucket_id: Uuid,
-		alias_name: &String,
-	) -> Result<(), Error> {
-		let mut bucket = self.get_internal_bucket(bucket_id).await?;
-
-		let mut alias = self
-			.0
-			.bucket_alias_table
-			.get(&EmptyKey, alias_name)
-			.await?
-			.ok_or_else(|| Error::NoSuchBucket(alias_name.to_string()))?;
-
-		// Checks ok, remove alias
-		let alias_ts = match bucket.state.as_option() {
-			Some(bucket_state) => increment_logical_clock_2(
-				alias.state.timestamp(),
-				bucket_state.aliases.get_timestamp(alias_name),
-			),
-			None => increment_logical_clock(alias.state.timestamp()),
-		};
-
-		// ---- timestamp-ensured causality barrier ----
-		// writes are now done and all writes use timestamp alias_ts
-
-		if alias.state.get() == &Some(bucket_id) {
-			alias.state = Lww::raw(alias_ts, None);
-			self.0.bucket_alias_table.insert(&alias).await?;
-		}
-
-		if let Some(bucket_state) = bucket.state.as_option_mut() {
-			bucket_state.aliases = LwwMap::raw_item(alias_name.clone(), alias_ts, false);
-			self.0.bucket_table.insert(&bucket).await?;
-		}
-
-		Ok(())
-	}
-
-	/// Sets a new alias for a bucket in the local namespace of a key.
-	/// This function fails if:
-	/// - alias name is not valid according to S3 spec
-	/// - bucket does not exist or is deleted
-	/// - key does not exist or is deleted
-	/// - alias already exists and points to another bucket
-	pub async fn set_local_bucket_alias(
-		&self,
-		bucket_id: Uuid,
-		key_id: &String,
-		alias_name: &String,
-	) -> Result<(), Error> {
-		let key_helper = KeyHelper(self.0);
-
-		if !is_valid_bucket_name(alias_name) {
-			return Err(Error::InvalidBucketName(alias_name.to_string()));
-		}
-
-		let mut bucket = self.get_existing_bucket(bucket_id).await?;
-		let mut key = key_helper.get_existing_key(key_id).await?;
-
-		let key_param = key.state.as_option_mut().unwrap();
-
-		if let Some(Some(existing_alias)) = key_param.local_aliases.get(alias_name) {
-			if *existing_alias != bucket_id {
-				return Err(Error::BadRequest(format!("Alias {} already exists in namespace of key {} and points to different bucket: {:?}", alias_name,  key.key_id, existing_alias)));
-			}
-		}
-
-		// Checks ok, add alias
-		let bucket_p = bucket.state.as_option_mut().unwrap();
-		let bucket_p_local_alias_key = (key.key_id.clone(), alias_name.clone());
-
-		// Calculate the timestamp to assign to this aliasing in the two local_aliases maps
-		// (the one from key to bucket, and the reverse one stored in the bucket iself)
-		// so that merges on both maps in case of a concurrent operation resolve
-		// to the same alias being set
-		let alias_ts = increment_logical_clock_2(
-			key_param.local_aliases.get_timestamp(alias_name),
-			bucket_p
-				.local_aliases
-				.get_timestamp(&bucket_p_local_alias_key),
-		);
-
-		// ---- timestamp-ensured causality barrier ----
-		// writes are now done and all writes use timestamp alias_ts
-
-		key_param.local_aliases = LwwMap::raw_item(alias_name.clone(), alias_ts, Some(bucket_id));
-		self.0.key_table.insert(&key).await?;
-
-		bucket_p.local_aliases = LwwMap::raw_item(bucket_p_local_alias_key, alias_ts, true);
-		self.0.bucket_table.insert(&bucket).await?;
-
-		Ok(())
-	}
-
-	/// Unsets an alias for a bucket in the local namespace of a key.
-	/// This function fails if:
-	/// - bucket does not exist or is deleted
-	/// - key does not exist or is deleted
-	/// - alias does not exist or maps to another bucket (-> internal error)
-	/// - bucket has no other aliases (global or local)
-	pub async fn unset_local_bucket_alias(
-		&self,
-		bucket_id: Uuid,
-		key_id: &String,
-		alias_name: &String,
-	) -> Result<(), Error> {
-		let key_helper = KeyHelper(self.0);
-
-		let mut bucket = self.get_existing_bucket(bucket_id).await?;
-		let mut key = key_helper.get_existing_key(key_id).await?;
-
-		let bucket_p = bucket.state.as_option_mut().unwrap();
-
-		if key
-			.state
-			.as_option()
-			.unwrap()
-			.local_aliases
-			.get(alias_name)
-			.cloned()
-			.flatten() != Some(bucket_id)
-		{
-			return Err(GarageError::Message(format!(
-				"Bucket {:?} does not have alias {} in namespace of key {}",
-				bucket_id, alias_name, key_id
-			))
-			.into());
-		}
-
-		let has_other_global_aliases = bucket_p
-			.aliases
-			.items()
-			.iter()
-			.any(|(_, _, active)| *active);
-		let has_other_local_aliases = bucket_p
-			.local_aliases
-			.items()
-			.iter()
-			.any(|((k, n), _, active)| *k == key.key_id && n == alias_name && *active);
-		if !has_other_global_aliases && !has_other_local_aliases {
-			return Err(Error::BadRequest(format!("Bucket {} doesn't have other aliases, please delete it instead of just unaliasing.", alias_name)));
-		}
-
-		// Checks ok, remove alias
-		let key_param = key.state.as_option_mut().unwrap();
-		let bucket_p_local_alias_key = (key.key_id.clone(), alias_name.clone());
-
-		let alias_ts = increment_logical_clock_2(
-			key_param.local_aliases.get_timestamp(alias_name),
-			bucket_p
-				.local_aliases
-				.get_timestamp(&bucket_p_local_alias_key),
-		);
-
-		// ---- timestamp-ensured causality barrier ----
-		// writes are now done and all writes use timestamp alias_ts
-
-		key_param.local_aliases = LwwMap::raw_item(alias_name.clone(), alias_ts, None);
-		self.0.key_table.insert(&key).await?;
-
-		bucket_p.local_aliases = LwwMap::raw_item(bucket_p_local_alias_key, alias_ts, false);
-		self.0.bucket_table.insert(&bucket).await?;
-
-		Ok(())
-	}
-
-	/// Sets permissions for a key on a bucket.
-	/// This function fails if:
-	/// - bucket or key cannot be found at all (its ok if they are in deleted state)
-	/// - bucket or key is in deleted state and we are trying to set permissions other than "deny
-	/// all"
-	pub async fn set_bucket_key_permissions(
-		&self,
-		bucket_id: Uuid,
-		key_id: &String,
-		mut perm: BucketKeyPerm,
-	) -> Result<(), Error> {
-		let key_helper = KeyHelper(self.0);
-
-		let mut bucket = self.get_internal_bucket(bucket_id).await?;
-		let mut key = key_helper.get_internal_key(key_id).await?;
-
-		if let Some(bstate) = bucket.state.as_option() {
-			if let Some(kp) = bstate.authorized_keys.get(key_id) {
-				perm.timestamp = increment_logical_clock_2(perm.timestamp, kp.timestamp);
-			}
-		} else if perm.is_any() {
-			return Err(Error::BadRequest(
-				"Trying to give permissions on a deleted bucket".into(),
-			));
-		}
-
-		if let Some(kstate) = key.state.as_option() {
-			if let Some(bp) = kstate.authorized_buckets.get(&bucket_id) {
-				perm.timestamp = increment_logical_clock_2(perm.timestamp, bp.timestamp);
-			}
-		} else if perm.is_any() {
-			return Err(Error::BadRequest(
-				"Trying to give permissions to a deleted key".into(),
-			));
-		}
-
-		// ---- timestamp-ensured causality barrier ----
-
-		if let Some(bstate) = bucket.state.as_option_mut() {
-			bstate.authorized_keys = Map::put_mutator(key_id.clone(), perm);
-			self.0.bucket_table.insert(&bucket).await?;
-		}
-
-		if let Some(kstate) = key.state.as_option_mut() {
-			kstate.authorized_buckets = Map::put_mutator(bucket_id, perm);
-			self.0.key_table.insert(&key).await?;
-		}
-
-		Ok(())
-	}
+	// ----
 
 	pub async fn is_bucket_empty(&self, bucket_id: Uuid) -> Result<bool, Error> {
 		let objects = self
diff --git a/src/model/helper/key.rs b/src/model/helper/key.rs
index c1a8e974..b8a99d55 100644
--- a/src/model/helper/key.rs
+++ b/src/model/helper/key.rs
@@ -1,12 +1,9 @@
 use garage_table::util::*;
-use garage_util::crdt::*;
 use garage_util::error::OkOrMessage;
 
 use crate::garage::Garage;
-use crate::helper::bucket::BucketHelper;
 use crate::helper::error::*;
 use crate::key_table::{Key, KeyFilter};
-use crate::permission::BucketKeyPerm;
 
 pub struct KeyHelper<'a>(pub(crate) &'a Garage);
 
@@ -65,38 +62,4 @@ impl<'a> KeyHelper<'a> {
 			Ok(candidates.into_iter().next().unwrap())
 		}
 	}
-
-	/// Deletes an API access key
-	pub async fn delete_key(&self, key: &mut Key) -> Result<(), Error> {
-		let bucket_helper = BucketHelper(self.0);
-
-		let state = key.state.as_option_mut().unwrap();
-
-		// --- done checking, now commit ---
-		// (the step at unset_local_bucket_alias will fail if a bucket
-		// does not have another alias, the deletion will be
-		// interrupted in the middle if that happens)
-
-		// 1. Delete local aliases
-		for (alias, _, to) in state.local_aliases.items().iter() {
-			if let Some(bucket_id) = to {
-				bucket_helper
-					.unset_local_bucket_alias(*bucket_id, &key.key_id, alias)
-					.await?;
-			}
-		}
-
-		// 2. Remove permissions on all authorized buckets
-		for (ab_id, _auth) in state.authorized_buckets.items().iter() {
-			bucket_helper
-				.set_bucket_key_permissions(*ab_id, &key.key_id, BucketKeyPerm::NO_PERMISSIONS)
-				.await?;
-		}
-
-		// 3. Actually delete key
-		key.state = Deletable::delete();
-		self.0.key_table.insert(key).await?;
-
-		Ok(())
-	}
 }
diff --git a/src/model/helper/locked.rs b/src/model/helper/locked.rs
new file mode 100644
index 00000000..f8e06add
--- /dev/null
+++ b/src/model/helper/locked.rs
@@ -0,0 +1,410 @@
+use garage_util::crdt::*;
+use garage_util::data::*;
+use garage_util::error::{Error as GarageError, OkOrMessage};
+use garage_util::time::*;
+
+use garage_table::util::*;
+
+use crate::bucket_alias_table::*;
+use crate::garage::Garage;
+use crate::helper::bucket::BucketHelper;
+use crate::helper::error::*;
+use crate::helper::key::KeyHelper;
+use crate::key_table::*;
+use crate::permission::BucketKeyPerm;
+
+/// A LockedHelper is the mandatory struct to hold when doing operations
+/// that modify access keys or bucket aliases. This structure takes
+/// a lock to a unit value that is in the globally-shared Garage struct.
+///
+/// This avoid several concurrent requests to modify the list of buckets
+/// and aliases at the same time, ending up in inconsistent states.
+/// This DOES NOT FIX THE FUNDAMENTAL ISSUE as CreateBucket requests handled
+/// by different API nodes can still break the cluster, but it is a first
+/// fix that allows consistency to be maintained if all such requests are
+/// directed to a single node, which is doable for many deployments.
+///
+/// See issues: #649, #723
+pub struct LockedHelper<'a>(
+	pub(crate) &'a Garage,
+	pub(crate) tokio::sync::MutexGuard<'a, ()>,
+);
+
+#[allow(clippy::ptr_arg)]
+impl<'a> LockedHelper<'a> {
+	pub fn bucket(&self) -> BucketHelper<'a> {
+		BucketHelper(self.0)
+	}
+
+	pub fn key(&self) -> KeyHelper<'a> {
+		KeyHelper(self.0)
+	}
+
+	/// Sets a new alias for a bucket in global namespace.
+	/// This function fails if:
+	/// - alias name is not valid according to S3 spec
+	/// - bucket does not exist or is deleted
+	/// - alias already exists and points to another bucket
+	pub async fn set_global_bucket_alias(
+		&self,
+		bucket_id: Uuid,
+		alias_name: &String,
+	) -> Result<(), Error> {
+		if !is_valid_bucket_name(alias_name) {
+			return Err(Error::InvalidBucketName(alias_name.to_string()));
+		}
+
+		let mut bucket = self.bucket().get_existing_bucket(bucket_id).await?;
+
+		let alias = self.0.bucket_alias_table.get(&EmptyKey, alias_name).await?;
+
+		if let Some(existing_alias) = alias.as_ref() {
+			if let Some(p_bucket) = existing_alias.state.get() {
+				if *p_bucket != bucket_id {
+					return Err(Error::BadRequest(format!(
+						"Alias {} already exists and points to different bucket: {:?}",
+						alias_name, p_bucket
+					)));
+				}
+			}
+		}
+
+		// Checks ok, add alias
+		let bucket_p = bucket.state.as_option_mut().unwrap();
+
+		let alias_ts = increment_logical_clock_2(
+			bucket_p.aliases.get_timestamp(alias_name),
+			alias.as_ref().map(|a| a.state.timestamp()).unwrap_or(0),
+		);
+
+		// ---- timestamp-ensured causality barrier ----
+		// writes are now done and all writes use timestamp alias_ts
+
+		let alias = match alias {
+			None => BucketAlias::new(alias_name.clone(), alias_ts, Some(bucket_id))
+				.ok_or_else(|| Error::InvalidBucketName(alias_name.clone()))?,
+			Some(mut a) => {
+				a.state = Lww::raw(alias_ts, Some(bucket_id));
+				a
+			}
+		};
+		self.0.bucket_alias_table.insert(&alias).await?;
+
+		bucket_p.aliases = LwwMap::raw_item(alias_name.clone(), alias_ts, true);
+		self.0.bucket_table.insert(&bucket).await?;
+
+		Ok(())
+	}
+
+	/// Unsets an alias for a bucket in global namespace.
+	/// This function fails if:
+	/// - bucket does not exist or is deleted
+	/// - alias does not exist or maps to another bucket (-> internal error)
+	/// - bucket has no other aliases (global or local)
+	pub async fn unset_global_bucket_alias(
+		&self,
+		bucket_id: Uuid,
+		alias_name: &String,
+	) -> Result<(), Error> {
+		let mut bucket = self.bucket().get_existing_bucket(bucket_id).await?;
+		let bucket_state = bucket.state.as_option_mut().unwrap();
+
+		let mut alias = self
+			.0
+			.bucket_alias_table
+			.get(&EmptyKey, alias_name)
+			.await?
+			.filter(|a| a.state.get().map(|x| x == bucket_id).unwrap_or(false))
+			.ok_or_message(format!(
+				"Internal error: alias not found or does not point to bucket {:?}",
+				bucket_id
+			))?;
+
+		let has_other_global_aliases = bucket_state
+			.aliases
+			.items()
+			.iter()
+			.any(|(name, _, active)| name != alias_name && *active);
+		let has_other_local_aliases = bucket_state
+			.local_aliases
+			.items()
+			.iter()
+			.any(|(_, _, active)| *active);
+		if !has_other_global_aliases && !has_other_local_aliases {
+			return Err(Error::BadRequest(format!("Bucket {} doesn't have other aliases, please delete it instead of just unaliasing.", alias_name)));
+		}
+
+		// Checks ok, remove alias
+		let alias_ts = increment_logical_clock_2(
+			alias.state.timestamp(),
+			bucket_state.aliases.get_timestamp(alias_name),
+		);
+
+		// ---- timestamp-ensured causality barrier ----
+		// writes are now done and all writes use timestamp alias_ts
+
+		alias.state = Lww::raw(alias_ts, None);
+		self.0.bucket_alias_table.insert(&alias).await?;
+
+		bucket_state.aliases = LwwMap::raw_item(alias_name.clone(), alias_ts, false);
+		self.0.bucket_table.insert(&bucket).await?;
+
+		Ok(())
+	}
+
+	/// Ensures a bucket does not have a certain global alias.
+	/// Contrarily to unset_global_bucket_alias, this does not
+	/// fail on any condition other than:
+	/// - bucket cannot be found (its fine if it is in deleted state)
+	/// - alias cannot be found (its fine if it points to nothing or
+	///   to another bucket)
+	pub async fn purge_global_bucket_alias(
+		&self,
+		bucket_id: Uuid,
+		alias_name: &String,
+	) -> Result<(), Error> {
+		let mut bucket = self.bucket().get_internal_bucket(bucket_id).await?;
+
+		let mut alias = self
+			.0
+			.bucket_alias_table
+			.get(&EmptyKey, alias_name)
+			.await?
+			.ok_or_else(|| Error::NoSuchBucket(alias_name.to_string()))?;
+
+		// Checks ok, remove alias
+		let alias_ts = match bucket.state.as_option() {
+			Some(bucket_state) => increment_logical_clock_2(
+				alias.state.timestamp(),
+				bucket_state.aliases.get_timestamp(alias_name),
+			),
+			None => increment_logical_clock(alias.state.timestamp()),
+		};
+
+		// ---- timestamp-ensured causality barrier ----
+		// writes are now done and all writes use timestamp alias_ts
+
+		if alias.state.get() == &Some(bucket_id) {
+			alias.state = Lww::raw(alias_ts, None);
+			self.0.bucket_alias_table.insert(&alias).await?;
+		}
+
+		if let Some(bucket_state) = bucket.state.as_option_mut() {
+			bucket_state.aliases = LwwMap::raw_item(alias_name.clone(), alias_ts, false);
+			self.0.bucket_table.insert(&bucket).await?;
+		}
+
+		Ok(())
+	}
+
+	/// Sets a new alias for a bucket in the local namespace of a key.
+	/// This function fails if:
+	/// - alias name is not valid according to S3 spec
+	/// - bucket does not exist or is deleted
+	/// - key does not exist or is deleted
+	/// - alias already exists and points to another bucket
+	pub async fn set_local_bucket_alias(
+		&self,
+		bucket_id: Uuid,
+		key_id: &String,
+		alias_name: &String,
+	) -> Result<(), Error> {
+		let key_helper = KeyHelper(self.0);
+
+		if !is_valid_bucket_name(alias_name) {
+			return Err(Error::InvalidBucketName(alias_name.to_string()));
+		}
+
+		let mut bucket = self.bucket().get_existing_bucket(bucket_id).await?;
+		let mut key = key_helper.get_existing_key(key_id).await?;
+
+		let key_param = key.state.as_option_mut().unwrap();
+
+		if let Some(Some(existing_alias)) = key_param.local_aliases.get(alias_name) {
+			if *existing_alias != bucket_id {
+				return Err(Error::BadRequest(format!("Alias {} already exists in namespace of key {} and points to different bucket: {:?}", alias_name,  key.key_id, existing_alias)));
+			}
+		}
+
+		// Checks ok, add alias
+		let bucket_p = bucket.state.as_option_mut().unwrap();
+		let bucket_p_local_alias_key = (key.key_id.clone(), alias_name.clone());
+
+		// Calculate the timestamp to assign to this aliasing in the two local_aliases maps
+		// (the one from key to bucket, and the reverse one stored in the bucket iself)
+		// so that merges on both maps in case of a concurrent operation resolve
+		// to the same alias being set
+		let alias_ts = increment_logical_clock_2(
+			key_param.local_aliases.get_timestamp(alias_name),
+			bucket_p
+				.local_aliases
+				.get_timestamp(&bucket_p_local_alias_key),
+		);
+
+		// ---- timestamp-ensured causality barrier ----
+		// writes are now done and all writes use timestamp alias_ts
+
+		key_param.local_aliases = LwwMap::raw_item(alias_name.clone(), alias_ts, Some(bucket_id));
+		self.0.key_table.insert(&key).await?;
+
+		bucket_p.local_aliases = LwwMap::raw_item(bucket_p_local_alias_key, alias_ts, true);
+		self.0.bucket_table.insert(&bucket).await?;
+
+		Ok(())
+	}
+
+	/// Unsets an alias for a bucket in the local namespace of a key.
+	/// This function fails if:
+	/// - bucket does not exist or is deleted
+	/// - key does not exist or is deleted
+	/// - alias does not exist or maps to another bucket (-> internal error)
+	/// - bucket has no other aliases (global or local)
+	pub async fn unset_local_bucket_alias(
+		&self,
+		bucket_id: Uuid,
+		key_id: &String,
+		alias_name: &String,
+	) -> Result<(), Error> {
+		let key_helper = KeyHelper(self.0);
+
+		let mut bucket = self.bucket().get_existing_bucket(bucket_id).await?;
+		let mut key = key_helper.get_existing_key(key_id).await?;
+
+		let bucket_p = bucket.state.as_option_mut().unwrap();
+
+		if key
+			.state
+			.as_option()
+			.unwrap()
+			.local_aliases
+			.get(alias_name)
+			.cloned()
+			.flatten() != Some(bucket_id)
+		{
+			return Err(GarageError::Message(format!(
+				"Bucket {:?} does not have alias {} in namespace of key {}",
+				bucket_id, alias_name, key_id
+			))
+			.into());
+		}
+
+		let has_other_global_aliases = bucket_p
+			.aliases
+			.items()
+			.iter()
+			.any(|(_, _, active)| *active);
+		let has_other_local_aliases = bucket_p
+			.local_aliases
+			.items()
+			.iter()
+			.any(|((k, n), _, active)| *k == key.key_id && n == alias_name && *active);
+		if !has_other_global_aliases && !has_other_local_aliases {
+			return Err(Error::BadRequest(format!("Bucket {} doesn't have other aliases, please delete it instead of just unaliasing.", alias_name)));
+		}
+
+		// Checks ok, remove alias
+		let key_param = key.state.as_option_mut().unwrap();
+		let bucket_p_local_alias_key = (key.key_id.clone(), alias_name.clone());
+
+		let alias_ts = increment_logical_clock_2(
+			key_param.local_aliases.get_timestamp(alias_name),
+			bucket_p
+				.local_aliases
+				.get_timestamp(&bucket_p_local_alias_key),
+		);
+
+		// ---- timestamp-ensured causality barrier ----
+		// writes are now done and all writes use timestamp alias_ts
+
+		key_param.local_aliases = LwwMap::raw_item(alias_name.clone(), alias_ts, None);
+		self.0.key_table.insert(&key).await?;
+
+		bucket_p.local_aliases = LwwMap::raw_item(bucket_p_local_alias_key, alias_ts, false);
+		self.0.bucket_table.insert(&bucket).await?;
+
+		Ok(())
+	}
+
+	/// Sets permissions for a key on a bucket.
+	/// This function fails if:
+	/// - bucket or key cannot be found at all (its ok if they are in deleted state)
+	/// - bucket or key is in deleted state and we are trying to set permissions other than "deny
+	/// all"
+	pub async fn set_bucket_key_permissions(
+		&self,
+		bucket_id: Uuid,
+		key_id: &String,
+		mut perm: BucketKeyPerm,
+	) -> Result<(), Error> {
+		let key_helper = KeyHelper(self.0);
+
+		let mut bucket = self.bucket().get_internal_bucket(bucket_id).await?;
+		let mut key = key_helper.get_internal_key(key_id).await?;
+
+		if let Some(bstate) = bucket.state.as_option() {
+			if let Some(kp) = bstate.authorized_keys.get(key_id) {
+				perm.timestamp = increment_logical_clock_2(perm.timestamp, kp.timestamp);
+			}
+		} else if perm.is_any() {
+			return Err(Error::BadRequest(
+				"Trying to give permissions on a deleted bucket".into(),
+			));
+		}
+
+		if let Some(kstate) = key.state.as_option() {
+			if let Some(bp) = kstate.authorized_buckets.get(&bucket_id) {
+				perm.timestamp = increment_logical_clock_2(perm.timestamp, bp.timestamp);
+			}
+		} else if perm.is_any() {
+			return Err(Error::BadRequest(
+				"Trying to give permissions to a deleted key".into(),
+			));
+		}
+
+		// ---- timestamp-ensured causality barrier ----
+
+		if let Some(bstate) = bucket.state.as_option_mut() {
+			bstate.authorized_keys = Map::put_mutator(key_id.clone(), perm);
+			self.0.bucket_table.insert(&bucket).await?;
+		}
+
+		if let Some(kstate) = key.state.as_option_mut() {
+			kstate.authorized_buckets = Map::put_mutator(bucket_id, perm);
+			self.0.key_table.insert(&key).await?;
+		}
+
+		Ok(())
+	}
+
+	// ----
+
+	/// Deletes an API access key
+	pub async fn delete_key(&self, key: &mut Key) -> Result<(), Error> {
+		let state = key.state.as_option_mut().unwrap();
+
+		// --- done checking, now commit ---
+		// (the step at unset_local_bucket_alias will fail if a bucket
+		// does not have another alias, the deletion will be
+		// interrupted in the middle if that happens)
+
+		// 1. Delete local aliases
+		for (alias, _, to) in state.local_aliases.items().iter() {
+			if let Some(bucket_id) = to {
+				self.unset_local_bucket_alias(*bucket_id, &key.key_id, alias)
+					.await?;
+			}
+		}
+
+		// 2. Remove permissions on all authorized buckets
+		for (ab_id, _auth) in state.authorized_buckets.items().iter() {
+			self.set_bucket_key_permissions(*ab_id, &key.key_id, BucketKeyPerm::NO_PERMISSIONS)
+				.await?;
+		}
+
+		// 3. Actually delete key
+		key.state = Deletable::delete();
+		self.0.key_table.insert(key).await?;
+
+		Ok(())
+	}
+}
diff --git a/src/model/helper/mod.rs b/src/model/helper/mod.rs
index dd947c86..5534ae3d 100644
--- a/src/model/helper/mod.rs
+++ b/src/model/helper/mod.rs
@@ -1,3 +1,4 @@
 pub mod bucket;
 pub mod error;
 pub mod key;
+pub mod locked;
diff --git a/src/model/migrate.rs b/src/model/migrate.rs
index 4c74b43b..8528382a 100644
--- a/src/model/migrate.rs
+++ b/src/model/migrate.rs
@@ -67,6 +67,8 @@ impl Migrate {
 			None
 		};
 
+		let helper = self.garage.locked_helper().await;
+
 		self.garage
 			.bucket_table
 			.insert(&Bucket {
@@ -84,14 +86,10 @@ impl Migrate {
 			})
 			.await?;
 
-		self.garage
-			.bucket_helper()
-			.set_global_bucket_alias(bucket_id, &new_name)
-			.await?;
+		helper.set_global_bucket_alias(bucket_id, &new_name).await?;
 
 		for (k, ts, perm) in old_bucket_p.authorized_keys.items().iter() {
-			self.garage
-				.bucket_helper()
+			helper
 				.set_bucket_key_permissions(
 					bucket_id,
 					k,