[lock-createbucket] Add node-global lock for bucket/key operations (fix #723)lock-createbucket

4 files changed, 60 insertions, 71 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?;
 	}