From cff702a951cb5bb193c7a891ababfd1d962ae9ed Mon Sep 17 00:00:00 2001 From: Alex Auvolat Date: Thu, 22 Feb 2024 12:28:21 +0100 Subject: [lock-createbucket] Add node-global lock for bucket/key operations (fix #723) --- src/api/admin/bucket.rs | 63 +++---- src/api/admin/key.rs | 6 +- src/api/s3/api_server.rs | 11 +- src/api/s3/bucket.rs | 51 +++--- src/garage/admin/bucket.rs | 40 ++--- src/garage/admin/key.rs | 7 +- src/model/garage.rs | 9 + src/model/helper/bucket.rs | 342 +------------------------------------ src/model/helper/key.rs | 37 ---- src/model/helper/locked.rs | 410 +++++++++++++++++++++++++++++++++++++++++++++ src/model/helper/mod.rs | 1 + src/model/migrate.rs | 10 +- 12 files changed, 511 insertions(+), 476 deletions(-) create mode 100644 src/model/helper/locked.rs (limited to 'src') 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, Error> { let req = parse_json_body::(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, id: String, ) -> Result, 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, Error> { let req = parse_json_body::(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, 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, 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, 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, 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, id: String, ) -> Result, 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, content_sha256: Option, - api_key: Key, + api_key_id: &String, bucket_name: String, ) -> Result, 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, 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 { - 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 { - 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 { - 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 { - 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 { - 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 { - 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>, + /// Lock to prevent concurrent modification of buckets and access keys + bucket_lock: tokio::sync::Mutex<()>, + /// Table containing S3 objects pub object_table: Arc>, /// 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 { 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, -- cgit v1.2.3