more resilience to inconsistent alias states

src/api/admin/bucket.rs

@@ -382,7 +382,7 @@ pub async fn handle_delete_bucket(
 	for ((key_id, alias), _, active) in state.local_aliases.items().iter() {
 		if *active {
 			helper
-				.unset_local_bucket_alias(bucket.id, key_id, alias)
+				.purge_local_bucket_alias(bucket.id, key_id, alias)
 				.await?;
 		}
 	}

src/api/s3/bucket.rs

@@ -241,11 +241,11 @@ pub async fn handle_delete_bucket(ctx: ReqCtx) -> Result<Response<ResBody>, Erro
 		// 1. delete bucket alias
 		if is_local_alias {
 			helper
-				.unset_local_bucket_alias(*bucket_id, &api_key.key_id, bucket_name)
+				.purge_local_bucket_alias(*bucket_id, &api_key.key_id, bucket_name)
 				.await?;
 		} else {
 			helper
-				.unset_global_bucket_alias(*bucket_id, bucket_name)
+				.purge_global_bucket_alias(*bucket_id, bucket_name)
 				.await?;
 		}
 

src/model/helper/locked.rs

@@ -6,6 +6,7 @@ use garage_util::time::*;
 use garage_table::util::*;
 
 use crate::bucket_alias_table::*;
+use crate::bucket_table::*;
 use crate::garage::Garage;
 use crate::helper::bucket::BucketHelper;
 use crate::helper::error::*;
@@ -47,6 +48,10 @@ impl<'a> LockedHelper<'a> {
 		KeyHelper(self.0)
 	}
 
+	// ================================================
+	//      global bucket aliases
+	// ================================================
+
 	/// Sets a new alias for a bucket in global namespace.
 	/// This function fails if:
 	/// - alias name is not valid according to S3 spec
@@ -204,6 +209,10 @@ impl<'a> LockedHelper<'a> {
 		Ok(())
 	}
 
+	// ================================================
+	//      local bucket aliases
+	// ================================================
+
 	/// 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
@@ -274,21 +283,13 @@ impl<'a> LockedHelper<'a> {
 	) -> 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 = self.bucket().get_existing_bucket(bucket_id).await?;
+		let key = key_helper.get_existing_key(key_id).await?;
 
-		let bucket_p = bucket.state.as_option_mut().unwrap();
+		let key_p = key.state.as_option().unwrap();
+		let bucket_p = bucket.state.as_option().unwrap();
 
-		if key
-			.state
-			.as_option()
-			.unwrap()
-			.local_aliases
-			.get(alias_name)
-			.cloned()
-			.flatten()
-			!= Some(bucket_id)
-		{
+		if key_p.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
@@ -305,12 +306,50 @@ impl<'a> LockedHelper<'a> {
 			.local_aliases
 			.items()
 			.iter()
-			.any(|((k, n), _, active)| *k == key.key_id && n == alias_name && *active);
+			.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
+		self.internal_purge_local_bucket_alias(bucket, key, alias_name)
+			.await?;
+
+		Ok(())
+	}
+
+	/// Ensures a bucket does not have a certain local alias.
+	/// Contrarily to unset_local_bucket_alias, this does not
+	/// fail on any condition other than:
+	/// - bucket cannot be found (its fine if it is in deleted state)
+	/// - key cannot be found (its fine if alias in key points to nothing
+	///   or to another bucket)
+	pub async fn purge_local_bucket_alias(
+		&self,
+		bucket_id: Uuid,
+		key_id: &String,
+		alias_name: &String,
+	) -> Result<(), Error> {
+		let key_helper = KeyHelper(self.0);
+
+		let bucket = self.bucket().get_existing_bucket(bucket_id).await?;
+		let key = key_helper.get_existing_key(key_id).await?;
+
+		self.internal_purge_local_bucket_alias(bucket, key, alias_name)
+			.await?;
+
+		Ok(())
+	}
+
+	async fn internal_purge_local_bucket_alias(
+		&self,
+		mut bucket: Bucket,
+		mut key: Key,
+		alias_name: &String,
+	) -> Result<(), Error> {
+		let bucket_p = bucket.state.as_option_mut().unwrap();
+
 		let key_param = key.state.as_option_mut().unwrap();
 		let bucket_p_local_alias_key = (key.key_id.clone(), alias_name.clone());
 
@@ -333,11 +372,15 @@ impl<'a> LockedHelper<'a> {
 		Ok(())
 	}
 
+	// ================================================
+	//      permissions
+	// ================================================
+
 	/// 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"
+	/// - 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,
@@ -384,21 +427,20 @@ impl<'a> LockedHelper<'a> {
 		Ok(())
 	}
 
-	// ----
+	// ================================================
+	//      keys
+	// ================================================
 
 	/// 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)
+				self.purge_local_bucket_alias(*bucket_id, &key.key_id, alias)
 					.await?;
 			}
 		}