From bf0a24ea69f462d7d1315aecd8a84c9fa0def99e Mon Sep 17 00:00:00 2001 From: Alex Auvolat Date: Mon, 4 May 2026 19:51:35 +0200 Subject: [PATCH 1/3] replace Option CRDT by explicit CancelingOption and MergingOption types --- src/api/admin/admin_token.rs | 8 +-- src/api/admin/bucket.rs | 43 +++++++------- src/api/admin/key.rs | 14 ++--- src/api/admin/special.rs | 2 +- src/api/common/cors.rs | 23 ++++---- src/api/s3/bucket.rs | 9 ++- src/api/s3/cors.rs | 6 +- src/api/s3/lifecycle.rs | 6 +- src/api/s3/website.rs | 6 +- src/model/admin_token_table.rs | 8 +-- src/model/bucket_alias_table.rs | 6 +- src/model/bucket_table.rs | 18 +++--- src/model/helper/bucket.rs | 28 ++++----- src/model/helper/locked.rs | 52 +++++++++++------ src/model/key_table.rs | 14 ++--- src/model/s3/lifecycle_worker.rs | 2 +- src/util/crdt/crdt.rs | 22 -------- src/util/crdt/mod.rs | 2 + src/util/crdt/option.rs | 97 ++++++++++++++++++++++++++++++++ src/web/web_server.rs | 4 +- 20 files changed, 237 insertions(+), 133 deletions(-) create mode 100644 src/util/crdt/option.rs diff --git a/src/api/admin/admin_token.rs b/src/api/admin/admin_token.rs index 242c9958..1dd83c7c 100644 --- a/src/api/admin/admin_token.rs +++ b/src/api/admin/admin_token.rs @@ -244,8 +244,8 @@ fn admin_token_info_results(token: &AdminApiToken, now: u64) -> GetAdminTokenInf .expect("invalid timestamp stored in db"), ), name: params.name.get().to_string(), - expiration: params.expiration.get().map(|x| { - DateTime::from_timestamp_millis(x as i64).expect("invalid timestamp stored in db") + expiration: params.expiration.get().inner().map(|x| { + DateTime::from_timestamp_millis(*x as i64).expect("invalid timestamp stored in db") }), expired: params.is_expired(now), scope: params.scope.get().0.clone(), @@ -279,10 +279,10 @@ fn apply_token_updates( if let Some(expiration) = updates.expiration { params .expiration - .update(Some(expiration.timestamp_millis() as u64)); + .update(Some(expiration.timestamp_millis() as u64).into()); } if updates.never_expires { - params.expiration.update(None); + params.expiration.update(None.into()); } if let Some(scope) = updates.scope { params.scope.update(AdminApiTokenScope(scope)); diff --git a/src/api/admin/bucket.rs b/src/api/admin/bucket.rs index e723d3b4..ad4844bb 100644 --- a/src/api/admin/bucket.rs +++ b/src/api/admin/bucket.rs @@ -90,7 +90,7 @@ impl RequestHandler for GetBucketInfoRequest { .bucket_alias_table .get(&EmptyKey, &ga) .await? - .and_then(|x| *x.state.get()) + .and_then(|x| x.state.get().into_inner()) .ok_or_else(|| HelperError::NoSuchBucket(ga.to_string()))?, (None, None, Some(search)) => { let helper = garage.bucket_helper(); @@ -168,7 +168,7 @@ impl RequestHandler for CreateBucketRequest { } if let Some(alias) = garage.bucket_alias_table.get(&EmptyKey, ga).await? { - if alias.state.get().is_some() { + if alias.state.get().inner().is_some() { return Err(CommonError::BucketAlreadyExists.into()); } } @@ -297,7 +297,7 @@ impl RequestHandler for UpdateBucketRequest { let redirect_all = state .website_config .get() - .as_ref() + .inner() .and_then(|wc| wc.redirect_all.clone()); let routing_rules = if let Some(rr) = wa.routing_rules { @@ -311,26 +311,29 @@ impl RequestHandler for UpdateBucketRequest { state .website_config .get() - .as_ref() + .inner() .map(|wc| wc.routing_rules.clone()) .unwrap_or_default() }; - state.website_config.update(Some(WebsiteConfig { - index_document: wa.index_document.ok_or_bad_request( - "Please specify indexDocument when enabling website access.", - )?, - error_document: wa.error_document, - redirect_all, - routing_rules, - })); + state.website_config.update( + Some(WebsiteConfig { + index_document: wa.index_document.ok_or_bad_request( + "Please specify indexDocument when enabling website access.", + )?, + error_document: wa.error_document, + redirect_all, + routing_rules, + }) + .into(), + ); } else { if wa.index_document.is_some() || wa.error_document.is_some() { return Err(Error::bad_request( "Cannot specify indexDocument or errorDocument when disabling website access.", )); } - state.website_config.update(None); + state.website_config.update(None.into()); } } @@ -353,7 +356,7 @@ impl RequestHandler for UpdateBucketRequest { Some(cc.into_garage_cors_config()?) }; - state.cors_config.update(cors_config); + state.cors_config.update(cors_config.into()); } if let Some(lr) = self.body.lifecycle_rules { @@ -370,7 +373,7 @@ impl RequestHandler for UpdateBucketRequest { ) }; - state.lifecycle_config.update(lifecycle_config); + state.lifecycle_config.update(lifecycle_config.into()); } garage.bucket_table.insert(&bucket).await?; @@ -739,8 +742,8 @@ async fn bucket_info_results( .filter(|(_, _, a)| *a) .map(|(n, _, _)| n.to_string()) .collect::>(), - website_access: state.website_config.get().is_some(), - website_config: state.website_config.get().clone().map(|wsc| { + website_access: state.website_config.get().inner().is_some(), + website_config: state.website_config.get().inner().cloned().map(|wsc| { GetBucketInfoWebsiteResponse { index_document: wsc.index_document, error_document: wsc.error_document, @@ -752,13 +755,13 @@ async fn bucket_info_results( ), } }), - cors_rules: state.cors_config.get().as_ref().map(|rules| { + cors_rules: state.cors_config.get().inner().map(|rules| { rules .iter() .map(xml::cors::CorsRule::from_garage_cors_rule) .collect::>() }), - lifecycle_rules: state.lifecycle_config.get().as_ref().map(|lc| { + lifecycle_rules: state.lifecycle_config.get().inner().map(|lc| { lc.iter() .map(xml::lifecycle::LifecycleRule::from_garage_lifecycle_rule) .collect::>() @@ -784,7 +787,7 @@ async fn bucket_info_results( .local_aliases .items() .iter() - .filter(|(_, _, b)| *b == Some(bucket.id)) + .filter(|(_, _, b)| b.into_inner() == Some(bucket.id)) .map(|(n, _, _)| n.to_string()) .collect::>(), }) diff --git a/src/api/admin/key.rs b/src/api/admin/key.rs index e8c8ad95..e5024cf8 100644 --- a/src/api/admin/key.rs +++ b/src/api/admin/key.rs @@ -40,8 +40,8 @@ impl RequestHandler for ListKeysRequest { DateTime::from_timestamp_millis(x as i64) .expect("invalid timestamp stored in db") }), - expiration: p.expiration.get().map(|x| { - DateTime::from_timestamp_millis(x as i64) + expiration: p.expiration.get().inner().map(|x| { + DateTime::from_timestamp_millis(*x as i64) .expect("invalid timestamp stored in db") }), expired: p.is_expired(now), @@ -201,7 +201,7 @@ async fn key_info_results( .local_aliases .items() .iter() - .filter_map(|(_, _, v)| v.as_ref()), + .filter_map(|(_, _, v)| v.inner()), ) { if !relevant_buckets.contains_key(id) { if let Some(b) = garage.bucket_table.get(&EmptyKey, id).await? { @@ -217,8 +217,8 @@ async fn key_info_results( created: key_state.created.map(|x| { DateTime::from_timestamp_millis(x as i64).expect("invalid timestamp stored in db") }), - expiration: key_state.expiration.get().map(|x| { - DateTime::from_timestamp_millis(x as i64).expect("invalid timestamp stored in db") + expiration: key_state.expiration.get().inner().map(|x| { + DateTime::from_timestamp_millis(*x as i64).expect("invalid timestamp stored in db") }), expired: key_state.is_expired(now_msec()), access_key_id: key.key_id.clone(), @@ -283,10 +283,10 @@ fn apply_key_updates(key: &mut Key, updates: UpdateKeyRequestBody) -> Result<(), if let Some(expiration) = updates.expiration { key_state .expiration - .update(Some(expiration.timestamp_millis() as u64)); + .update(Some(expiration.timestamp_millis() as u64).into()); } if updates.never_expires { - key_state.expiration.update(None); + key_state.expiration.update(None.into()); } if let Some(allow) = updates.allow { if allow.create_bucket { diff --git a/src/api/admin/special.rs b/src/api/admin/special.rs index 0a4e6705..ddeedfa8 100644 --- a/src/api/admin/special.rs +++ b/src/api/admin/special.rs @@ -164,7 +164,7 @@ async fn check_domain(garage: &Arc, domain: &str) -> Result } let bucket_state = bucket.state.as_option().unwrap(); - let bucket_website_config = bucket_state.website_config.get(); + let bucket_website_config = bucket_state.website_config.get().inner(); match bucket_website_config { Some(_v) => Ok(true), diff --git a/src/api/common/cors.rs b/src/api/common/cors.rs index 17158acd..305843c1 100644 --- a/src/api/common/cors.rs +++ b/src/api/common/cors.rs @@ -19,7 +19,7 @@ pub fn find_matching_cors_rule<'a, B>( bucket_params: &'a BucketParams, req: &'a Request, ) -> Result, CommonError> { - if let Some(cors_config) = bucket_params.cors_config.get() { + if let Some(cors_config) = bucket_params.cors_config.get().inner() { if let Some(origin) = req.headers().get("Origin") { let origin = origin.to_str()?; let request_headers = match req.headers().get(ACCESS_CONTROL_REQUEST_HEADERS) { @@ -158,7 +158,7 @@ pub fn handle_options_for_bucket( None => vec![], }; - if let Some(cors_config) = bucket_params.cors_config.get() { + if let Some(cors_config) = bucket_params.cors_config.get().inner() { let matching_rule = cors_config .iter() .find(|rule| cors_rule_matches(rule, origin, request_method, request_headers.iter())); @@ -192,14 +192,17 @@ mod tests { fn bucket_params_with_rule(allow_origins: Vec<&str>) -> BucketParams { let mut bucket_params = BucketParams::default(); - bucket_params.cors_config.update(Some(vec![GarageCorsRule { - id: Some("cors-test".into()), - max_age_seconds: None, - allow_origins: allow_origins.into_iter().map(str::to_string).collect(), - allow_methods: vec!["GET".into(), "PUT".into()], - allow_headers: vec!["*".into()], - expose_headers: vec![], - }])); + bucket_params.cors_config.update( + Some(vec![GarageCorsRule { + id: Some("cors-test".into()), + max_age_seconds: None, + allow_origins: allow_origins.into_iter().map(str::to_string).collect(), + allow_methods: vec!["GET".into(), "PUT".into()], + allow_headers: vec!["*".into()], + expose_headers: vec![], + }]) + .into(), + ); bucket_params } diff --git a/src/api/s3/bucket.rs b/src/api/s3/bucket.rs index 901961e9..bcccdaf7 100644 --- a/src/api/s3/bucket.rs +++ b/src/api/s3/bucket.rs @@ -122,7 +122,7 @@ pub async fn handle_list_buckets( for (alias, _, _active) in bucket.aliases().iter().filter(|(_, _, active)| *active) { let alias_opt = garage.bucket_alias_table.get(&EmptyKey, alias).await?; if let Some(alias_ent) = alias_opt { - if *alias_ent.state.get() == Some(*bucket_id) { + if alias_ent.state.get().inner() == Some(bucket_id) { aliases.insert(alias_ent.name().to_string(), *bucket_id); } } @@ -134,7 +134,7 @@ pub async fn handle_list_buckets( } for (alias, _, id_opt) in key_p.local_aliases.items() { - if let Some(id) = id_opt { + if let Some(id) = id_opt.inner() { aliases.insert(alias.clone(), *id); } } @@ -256,7 +256,10 @@ pub async fn handle_delete_bucket(ctx: ReqCtx) -> Result, Erro let key_params = api_key.params().unwrap(); - let is_local_alias = matches!(key_params.local_aliases.get(bucket_name), Some(Some(_))); + let is_local_alias = matches!( + key_params.local_aliases.get(bucket_name).map(|x| x.inner()), + Some(Some(_)) + ); // If the bucket has no other aliases, this is a true deletion. // Otherwise, it is just an alias removal. diff --git a/src/api/s3/cors.rs b/src/api/s3/cors.rs index 9356983f..d5837d4f 100644 --- a/src/api/s3/cors.rs +++ b/src/api/s3/cors.rs @@ -13,7 +13,7 @@ use crate::xml::to_xml_with_header; pub async fn handle_get_cors(ctx: ReqCtx) -> Result, Error> { let ReqCtx { bucket_params, .. } = ctx; - if let Some(cors) = bucket_params.cors_config.get() { + if let Some(cors) = bucket_params.cors_config.get().inner() { let wc = CorsConfiguration { xmlns: (), cors_rules: cors @@ -38,7 +38,7 @@ pub async fn handle_delete_cors(ctx: ReqCtx) -> Result, Error> mut bucket_params, .. } = ctx; - bucket_params.cors_config.update(None); + bucket_params.cors_config.update(None.into()); garage .bucket_table .insert(&Bucket::present(bucket_id, bucket_params)) @@ -67,7 +67,7 @@ pub async fn handle_put_cors( bucket_params .cors_config - .update(Some(conf.into_garage_cors_config()?)); + .update(Some(conf.into_garage_cors_config()?).into()); garage .bucket_table .insert(&Bucket::present(bucket_id, bucket_params)) diff --git a/src/api/s3/lifecycle.rs b/src/api/s3/lifecycle.rs index 83064bca..18b80b43 100644 --- a/src/api/s3/lifecycle.rs +++ b/src/api/s3/lifecycle.rs @@ -14,7 +14,7 @@ use garage_model::bucket_table::Bucket; pub async fn handle_get_lifecycle(ctx: ReqCtx) -> Result, Error> { let ReqCtx { bucket_params, .. } = ctx; - if let Some(lifecycle) = bucket_params.lifecycle_config.get() { + if let Some(lifecycle) = bucket_params.lifecycle_config.get().inner() { let wc = LifecycleConfiguration::from_garage_lifecycle_config(lifecycle); let xml = to_xml_with_header(&wc)?; Ok(Response::builder() @@ -33,7 +33,7 @@ pub async fn handle_delete_lifecycle(ctx: ReqCtx) -> Result, E mut bucket_params, .. } = ctx; - bucket_params.lifecycle_config.update(None); + bucket_params.lifecycle_config.update(None.into()); garage .bucket_table .insert(&Bucket::present(bucket_id, bucket_params)) @@ -62,7 +62,7 @@ pub async fn handle_put_lifecycle( .validate_into_garage_lifecycle_config() .ok_or_bad_request("Invalid lifecycle configuration")?; - bucket_params.lifecycle_config.update(Some(config)); + bucket_params.lifecycle_config.update(Some(config).into()); garage .bucket_table .insert(&Bucket::present(bucket_id, bucket_params)) diff --git a/src/api/s3/website.rs b/src/api/s3/website.rs index 0010a300..0b4f1f2b 100644 --- a/src/api/s3/website.rs +++ b/src/api/s3/website.rs @@ -16,7 +16,7 @@ pub const X_AMZ_WEBSITE_REDIRECT_LOCATION: HeaderName = pub async fn handle_get_website(ctx: ReqCtx) -> Result, Error> { let ReqCtx { bucket_params, .. } = ctx; - if let Some(website) = bucket_params.website_config.get() { + if let Some(website) = bucket_params.website_config.get().inner() { let wc = WebsiteConfiguration { xmlns: (), error_document: website.error_document.as_ref().map(|v| Key { @@ -54,7 +54,7 @@ pub async fn handle_delete_website(ctx: ReqCtx) -> Result, Err mut bucket_params, .. } = ctx; - bucket_params.website_config.update(None); + bucket_params.website_config.update(None.into()); garage .bucket_table .insert(&Bucket::present(bucket_id, bucket_params)) @@ -83,7 +83,7 @@ pub async fn handle_put_website( bucket_params .website_config - .update(Some(conf.into_garage_website_config()?)); + .update(Some(conf.into_garage_website_config()?).into()); garage .bucket_table .insert(&Bucket::present(bucket_id, bucket_params)) diff --git a/src/model/admin_token_table.rs b/src/model/admin_token_table.rs index 97159651..21ce2c08 100644 --- a/src/model/admin_token_table.rs +++ b/src/model/admin_token_table.rs @@ -35,7 +35,7 @@ mod v2 { pub name: crdt::Lww, /// The optional time of expiration of the token - pub expiration: crdt::Lww>, + pub expiration: crdt::Lww>, /// The scope of the token, i.e. list of authorized admin API calls pub scope: crdt::Lww, @@ -106,7 +106,7 @@ impl AdminApiToken { created: now_msec(), token_hash: hashed_token, name: crdt::Lww::new(name.to_string()), - expiration: crdt::Lww::new(None), + expiration: crdt::Lww::new(None.into()), scope: crdt::Lww::new(AdminApiTokenScope(vec!["*".to_string()])), }), }; @@ -147,9 +147,9 @@ impl AdminApiToken { impl AdminApiTokenParams { pub fn is_expired(&self, ts_now: u64) -> bool { - match *self.expiration.get() { + match self.expiration.get().inner() { None => false, - Some(exp) => ts_now >= exp, + Some(exp) => ts_now >= *exp, } } diff --git a/src/model/bucket_alias_table.rs b/src/model/bucket_alias_table.rs index 276d0d1c..1b9bb2d4 100644 --- a/src/model/bucket_alias_table.rs +++ b/src/model/bucket_alias_table.rs @@ -13,7 +13,7 @@ mod v08 { #[derive(PartialEq, Eq, Clone, Debug, Serialize, Deserialize)] pub struct BucketAlias { pub(super) name: String, - pub state: crdt::Lww>, + pub state: crdt::Lww>, } impl garage_util::migrate::InitialFormat for BucketAlias {} @@ -25,12 +25,12 @@ impl BucketAlias { pub fn new(name: String, ts: u64, bucket_id: Option) -> Self { BucketAlias { name, - state: crdt::Lww::raw(ts, bucket_id), + state: crdt::Lww::raw(ts, CancelingOption(bucket_id)), } } pub fn is_deleted(&self) -> bool { - self.state.get().is_none() + self.state.get().inner().is_none() } pub fn name(&self) -> &str { &self.name diff --git a/src/model/bucket_table.rs b/src/model/bucket_table.rs index c2a5a135..22464e9d 100644 --- a/src/model/bucket_table.rs +++ b/src/model/bucket_table.rs @@ -45,12 +45,12 @@ mod v08 { /// Whether this bucket is allowed for website access /// (under all of its global alias names), /// and if so, the website configuration XML document - pub website_config: crdt::Lww>, + pub website_config: crdt::Lww>, /// CORS rules - pub cors_config: crdt::Lww>>, + pub cors_config: crdt::Lww>>, /// Lifecycle configuration #[serde(default)] - pub lifecycle_config: crdt::Lww>>, + pub lifecycle_config: crdt::Lww>>, /// Bucket quotas #[serde(default)] pub quotas: crdt::Lww, @@ -164,11 +164,11 @@ mod v2 { /// Whether this bucket is allowed for website access /// (under all of its global alias names), /// and if so, the website configuration XML document - pub website_config: crdt::Lww>, + pub website_config: crdt::Lww>, /// CORS rules - pub cors_config: crdt::Lww>>, + pub cors_config: crdt::Lww>>, /// Lifecycle configuration - pub lifecycle_config: crdt::Lww>>, + pub lifecycle_config: crdt::Lww>>, /// Bucket quotas pub quotas: crdt::Lww, } @@ -259,9 +259,9 @@ impl BucketParams { authorized_keys: crdt::Map::new(), aliases: crdt::LwwMap::new(), local_aliases: crdt::LwwMap::new(), - website_config: crdt::Lww::new(None), - cors_config: crdt::Lww::new(None), - lifecycle_config: crdt::Lww::new(None), + website_config: crdt::Lww::new(None.into()), + cors_config: crdt::Lww::new(None.into()), + lifecycle_config: crdt::Lww::new(None.into()), quotas: crdt::Lww::new(BucketQuotas::default()), } } diff --git a/src/model/helper/bucket.rs b/src/model/helper/bucket.rs index b1caded5..55c14cf2 100644 --- a/src/model/helper/bucket.rs +++ b/src/model/helper/bucket.rs @@ -52,7 +52,7 @@ impl<'a> BucketHelper<'a> { .0 .bucket_alias_table .get_local(&EmptyKey, bucket_name)? - .and_then(|x| *x.state.get()); + .and_then(|x| x.state.get().into_inner()); match alias { Some(id) => id, None => return Ok(None), @@ -91,15 +91,18 @@ impl<'a> BucketHelper<'a> { .as_option() .ok_or_message("Key should not be deleted at this point")?; - let bucket_opt = - if let Some(Some(bucket_id)) = api_key_params.local_aliases.get(bucket_name) { - self.0 - .bucket_table - .get_local(&EmptyKey, bucket_id)? - .filter(|x| !x.state.is_deleted()) - } else { - self.resolve_global_bucket_fast(bucket_name)? - }; + let bucket_opt = if let Some(bucket_id) = api_key_params + .local_aliases + .get(bucket_name) + .and_then(|x| x.inner()) + { + self.0 + .bucket_table + .get_local(&EmptyKey, bucket_id)? + .filter(|x| !x.state.is_deleted()) + } else { + self.resolve_global_bucket_fast(bucket_name)? + }; bucket_opt.ok_or_else(|| Error::NoSuchBucket(bucket_name.to_string())) } @@ -125,7 +128,7 @@ impl<'a> BucketHelper<'a> { .bucket_alias_table .get(&EmptyKey, bucket_name) .await? - .and_then(|x| *x.state.get()); + .and_then(|x| x.state.get().into_inner()); match alias { Some(id) => id, None => return Ok(None), @@ -163,8 +166,7 @@ impl<'a> BucketHelper<'a> { .ok_or_else(|| GarageError::Message(format!("access key {} has been deleted", key_id)))? .local_aliases .get(bucket_name) - .copied() - .flatten(); + .and_then(|x| x.inner().copied()); if let Some(bucket_id) = local_alias { Ok(self diff --git a/src/model/helper/locked.rs b/src/model/helper/locked.rs index a1ad5b2b..5d751dce 100644 --- a/src/model/helper/locked.rs +++ b/src/model/helper/locked.rs @@ -74,8 +74,8 @@ impl<'a> LockedHelper<'a> { 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 { + if let Some(p_bucket) = existing_alias.state.get().into_inner() { + if p_bucket != bucket_id { return Err(Error::BadRequest(format!( "Alias {} already exists and points to different bucket: {:?}", alias_name, p_bucket @@ -98,7 +98,7 @@ impl<'a> LockedHelper<'a> { let alias = match alias { None => BucketAlias::new(alias_name.clone(), alias_ts, Some(bucket_id)), Some(mut a) => { - a.state = Lww::raw(alias_ts, Some(bucket_id)); + a.state = Lww::raw(alias_ts, Some(bucket_id).into()); a } }; @@ -128,7 +128,13 @@ impl<'a> LockedHelper<'a> { .bucket_alias_table .get(&EmptyKey, alias_name) .await? - .filter(|a| a.state.get().map(|x| x == bucket_id).unwrap_or(false)) + .filter(|a| { + a.state + .get() + .into_inner() + .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 @@ -157,7 +163,7 @@ impl<'a> LockedHelper<'a> { // ---- timestamp-ensured causality barrier ---- // writes are now done and all writes use timestamp alias_ts - alias.state = Lww::raw(alias_ts, None); + alias.state = Lww::raw(alias_ts, None.into()); self.0.bucket_alias_table.insert(&alias).await?; bucket_state.aliases = LwwMap::raw_item(alias_name.clone(), alias_ts, false); @@ -199,8 +205,8 @@ impl<'a> LockedHelper<'a> { // ---- 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); + if alias.state.get().inner() == Some(&bucket_id) { + alias.state = Lww::raw(alias_ts, None.into()); self.0.bucket_alias_table.insert(&alias).await?; } @@ -237,7 +243,11 @@ impl<'a> LockedHelper<'a> { let key_param = key.state.as_option_mut().unwrap(); - if let Some(Some(existing_alias)) = key_param.local_aliases.get(alias_name) { + if let Some(Some(existing_alias)) = key_param + .local_aliases + .get(alias_name) + .map(CancelingOption::inner) + { 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))); } @@ -261,7 +271,8 @@ impl<'a> LockedHelper<'a> { // ---- 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)); + key_param.local_aliases = + LwwMap::raw_item(alias_name.clone(), alias_ts, Some(bucket_id).into()); self.0.key_table.insert(&key).await?; bucket_p.local_aliases = LwwMap::raw_item(bucket_p_local_alias_key, alias_ts, true); @@ -288,7 +299,12 @@ impl<'a> LockedHelper<'a> { let key_p = key.state.as_option().unwrap(); let bucket_p = bucket.state.as_option_mut().unwrap(); - if key_p.local_aliases.get(alias_name).cloned().flatten() != Some(bucket_id) { + if key_p + .local_aliases + .get(alias_name) + .and_then(CancelingOption::inner) + != Some(&bucket_id) + { return Err(GarageError::Message(format!( "Bucket {:?} does not have alias {} in namespace of key {}", bucket_id, alias_name, key_id @@ -325,7 +341,7 @@ impl<'a> LockedHelper<'a> { // writes are now done and all writes use timestamp alias_ts key.state.as_option_mut().unwrap().local_aliases = - LwwMap::raw_item(alias_name.clone(), alias_ts, None); + LwwMap::raw_item(alias_name.clone(), alias_ts, None.into()); self.0.key_table.insert(&key).await?; bucket_p.local_aliases = LwwMap::raw_item(bucket_p_local_alias_key, alias_ts, false); @@ -367,7 +383,7 @@ impl<'a> LockedHelper<'a> { // writes are now done and all writes use timestamp alias_ts if let Some(kp) = key.state.as_option_mut() { - kp.local_aliases = LwwMap::raw_item(alias_name.clone(), alias_ts, None); + kp.local_aliases = LwwMap::raw_item(alias_name.clone(), alias_ts, None.into()); self.0.key_table.insert(&key).await?; } @@ -444,8 +460,8 @@ impl<'a> LockedHelper<'a> { // 1. Delete local aliases for (alias, _, to) in state.local_aliases.items().iter() { - if let Some(bucket_id) = to { - self.purge_local_bucket_alias(*bucket_id, &key.key_id, alias) + if let Some(bucket_id) = to.into_inner() { + self.purge_local_bucket_alias(bucket_id, &key.key_id, alias) .await?; } } @@ -501,7 +517,7 @@ impl<'a> LockedHelper<'a> { .data .decode_entry(&(item?.1)) .map_err(db::TxError::Abort)?; - if let Some(id) = alias.state.get() { + if let Some(id) = alias.state.get().inner() { if all_buckets.contains(id) { // keep aliases global_aliases.insert(alias.name().to_string(), *id); @@ -512,7 +528,7 @@ impl<'a> LockedHelper<'a> { alias.name(), id ); - alias.state.update(None); + alias.state.update(None.into()); delete_global.push(alias); } } @@ -544,7 +560,7 @@ impl<'a> LockedHelper<'a> { }; let mut has_changes = false; for (name, _, to) in p.local_aliases.items().to_vec() { - if let Some(id) = to { + if let Some(id) = to.into_inner() { if all_buckets.contains(&id) { local_aliases.insert((key.key_id.clone(), name), id); } else { @@ -552,7 +568,7 @@ impl<'a> LockedHelper<'a> { "local alias: remove ({}, {}) -> {:?} (bucket is deleted)", key.key_id, name, id ); - p.local_aliases.update_in_place(name, None); + p.local_aliases.update_in_place(name, None.into()); has_changes = true; } } diff --git a/src/model/key_table.rs b/src/model/key_table.rs index 72ca0182..da0dcb7d 100644 --- a/src/model/key_table.rs +++ b/src/model/key_table.rs @@ -43,7 +43,7 @@ mod v08 { /// A key can have a local view of buckets names it is /// the only one to see, this is the namespace for these aliases - pub local_aliases: crdt::LwwMap>, + pub local_aliases: crdt::LwwMap>, } impl garage_util::migrate::InitialFormat for Key {} @@ -79,7 +79,7 @@ mod v2 { /// Name for the key pub name: crdt::Lww, /// The optional time of expiration of the key - pub expiration: crdt::Lww>, + pub expiration: crdt::Lww>, /// Flag to allow users having this key to create buckets pub allow_create_bucket: crdt::Lww, @@ -91,7 +91,7 @@ mod v2 { /// A key can have a local view of buckets names it is /// the only one to see, this is the namespace for these aliases - pub local_aliases: crdt::LwwMap>, + pub local_aliases: crdt::LwwMap>, } impl garage_util::migrate::Migrate for Key { @@ -106,7 +106,7 @@ mod v2 { created: None, secret_key: x.secret_key, name: x.name, - expiration: crdt::Lww::raw(0, None), + expiration: crdt::Lww::raw(0, None.into()), allow_create_bucket: x.allow_create_bucket, authorized_buckets: x.authorized_buckets, local_aliases: x.local_aliases, @@ -124,7 +124,7 @@ impl KeyParams { created: Some(now_msec()), secret_key: secret_key.to_string(), name: crdt::Lww::new(name.to_string()), - expiration: crdt::Lww::new(None), + expiration: crdt::Lww::new(None.into()), allow_create_bucket: crdt::Lww::new(false), authorized_buckets: crdt::Map::new(), local_aliases: crdt::LwwMap::new(), @@ -229,9 +229,9 @@ impl Key { impl KeyParams { pub fn is_expired(&self, ts_now: u64) -> bool { - match *self.expiration.get() { + match self.expiration.get().inner() { None => false, - Some(exp) => ts_now >= exp, + Some(exp) => ts_now >= *exp, } } } diff --git a/src/model/s3/lifecycle_worker.rs b/src/model/s3/lifecycle_worker.rs index f014337f..24c858ae 100644 --- a/src/model/s3/lifecycle_worker.rs +++ b/src/model/s3/lifecycle_worker.rs @@ -271,7 +271,7 @@ async fn process_object( let lifecycle_policy: &[LifecycleRule] = bucket .state .as_option() - .and_then(|s| s.lifecycle_config.get().as_deref()) + .and_then(|s| s.lifecycle_config.get().inner().map(|x| &x[..])) .unwrap_or_default(); if lifecycle_policy.iter().all(|x| !x.enabled) { diff --git a/src/util/crdt/crdt.rs b/src/util/crdt/crdt.rs index ebdc66d1..7b700cde 100644 --- a/src/util/crdt/crdt.rs +++ b/src/util/crdt/crdt.rs @@ -26,28 +26,6 @@ pub trait Crdt { fn merge(&mut self, other: &Self); } -/// `Option` implements Crdt for any type T, even if T doesn't implement CRDT itself: when -/// different values are detected, they are always merged to None. This can be used for value -/// types which shoulnd't be merged, instead of trying to merge things when we know we don't want -/// to merge them (which is what the `AutoCrdt` trait is used for most of the time). This cases -/// arises very often, for example with a Lww or a `LwwMap`: the value type has to be a CRDT so that -/// we have a rule for what to do when timestamps aren't enough to disambiguate (in a distributed -/// system, anything can happen!), and with `AutoCrdt` the rule is to make an arbitrary (but -/// deterministic) choice between the two. When using an `Option` instead with this impl, ambiguity -/// cases are explicitly stored as None, which allows us to detect the ambiguity and handle it in -/// the way we want. (this can only work if we are happy with losing the value when an ambiguity -/// arises) -impl Crdt for Option -where - T: Eq, -{ - fn merge(&mut self, other: &Self) { - if self != other { - *self = None; - } - } -} - /// All types that implement `Ord` (a total order) can also implement a trivial CRDT /// defined by the merge rule: `a ⊔ b = max(a, b)`. Implement this trait for your type /// to enable this behavior. diff --git a/src/util/crdt/mod.rs b/src/util/crdt/mod.rs index 64f0984e..563e820a 100644 --- a/src/util/crdt/mod.rs +++ b/src/util/crdt/mod.rs @@ -16,6 +16,7 @@ mod deletable; mod lww; mod lww_map; mod map; +mod option; pub use self::bool::*; pub use crdt::*; @@ -23,3 +24,4 @@ pub use deletable::*; pub use lww::*; pub use lww_map::*; pub use map::*; +pub use option::*; diff --git a/src/util/crdt/option.rs b/src/util/crdt/option.rs new file mode 100644 index 00000000..15d5b8a2 --- /dev/null +++ b/src/util/crdt/option.rs @@ -0,0 +1,97 @@ +use serde::{Deserialize, Serialize}; + +use crate::crdt::Crdt; + +#[derive(Serialize, Deserialize, Clone, Default, Copy, PartialEq, Eq, Debug)] +#[cfg_attr(feature = "arbitrary", derive(arbitrary::Arbitrary))] +#[serde(transparent)] +pub struct CancelingOption(pub Option); + +/// `CancelingOption` implements Crdt for any type T, even if T doesn't implement CRDT itself: when +/// different values are detected, they are always merged to None. This can be used for value +/// types which shoulnd't be merged, instead of trying to merge things when we know we don't want +/// to merge them (which is what the `AutoCrdt` trait is used for most of the time). This cases +/// arises very often, for example with a Lww or a `LwwMap`: the value type has to be a CRDT so that +/// we have a rule for what to do when timestamps aren't enough to disambiguate (in a distributed +/// system, anything can happen!), and with `AutoCrdt` the rule is to make an arbitrary (but +/// deterministic) choice between the two. When using an `CancelingOption` instead with this impl, ambiguity +/// cases are explicitly stored as None, which allows us to detect the ambiguity and handle it in +/// the way we want. (this can only work if we are happy with losing the value when an ambiguity +/// arises) +impl Crdt for CancelingOption +where + T: Eq + Clone, +{ + fn merge(&mut self, other: &Self) { + match (self.0.as_ref(), other.0.as_ref()) { + (Some(a), Some(b)) if a != b => { + self.0 = None; + } + (None, Some(b)) => { + self.0 = Some(b.clone()); + } + _ => {} + } + } +} + +impl CancelingOption { + pub fn inner(&self) -> Option<&T> { + self.0.as_ref() + } + + pub fn into_inner(self) -> Option { + self.0 + } + + pub fn map(self, f: impl FnOnce(T) -> U) -> CancelingOption { + CancelingOption(self.0.map(f)) + } +} + +impl From> for CancelingOption { + fn from(x: Option) -> Self { + Self(x) + } +} + +#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, Copy, Default)] +#[cfg_attr(feature = "arbitrary", derive(arbitrary::Arbitrary))] +#[serde(transparent)] +pub struct MergingOption(pub Option); + +/// `MergingOption` implements `Crdt` when `T` implements `Crdt`: +/// None is a bottom value, and different Some values get merged according +/// to their Crdt operator. +impl Crdt for MergingOption +where + T: Crdt + Clone, +{ + fn merge(&mut self, other: &Self) { + if let (Some(a), Some(b)) = (self.0.as_mut(), other.0.as_ref()) { + a.merge(b); + } else { + self.0 = self.0.take().or_else(|| other.0.clone()); + } + } +} + +impl MergingOption { + pub fn inner(&self) -> Option<&T> { + self.0.as_ref() + } + + pub fn into_inner(self) -> Option { + self.0 + } + + pub fn map(self, f: impl FnOnce(T) -> U) -> MergingOption { + MergingOption(self.0.map(f)) + } +} + +impl From> for MergingOption { + fn from(x: Option) -> Self { + Self(x) + } +} diff --git a/src/web/web_server.rs b/src/web/web_server.rs index 6aac096e..eaff9dc7 100644 --- a/src/web/web_server.rs +++ b/src/web/web_server.rs @@ -241,7 +241,7 @@ impl WebServer { .bucket_alias_table .get(&EmptyKey, &bucket_name.to_string()) .await? - .and_then(|x| x.state.take()) + .and_then(|x| x.state.get().into_inner()) .ok_or(Error::NotFound)?; // Check bucket isn't deleted and has website access enabled @@ -256,7 +256,7 @@ impl WebServer { let website_config = bucket_params .website_config .get() - .as_ref() + .inner() .ok_or(Error::NotFound)?; // Get path From bacc6c98b26b4e15b69fe9f4b60c50faa460e294 Mon Sep 17 00:00:00 2001 From: Alex Auvolat Date: Wed, 13 May 2026 10:53:56 +0200 Subject: [PATCH 2/3] replace expiration field with custom type that merges to min value --- src/api/admin/admin_token.rs | 5 +++-- src/api/admin/key.rs | 7 ++++--- src/model/admin_token_table.rs | 5 +++-- src/model/key_table.rs | 5 +++-- src/model/permission.rs | 12 ++++++++++++ 5 files changed, 25 insertions(+), 9 deletions(-) diff --git a/src/api/admin/admin_token.rs b/src/api/admin/admin_token.rs index 1dd83c7c..b954b53e 100644 --- a/src/api/admin/admin_token.rs +++ b/src/api/admin/admin_token.rs @@ -7,6 +7,7 @@ use garage_util::time::now_msec; use garage_model::admin_token_table::*; use garage_model::garage::Garage; +use garage_model::permission::ExpirationTime; use crate::api::*; use crate::error::*; @@ -245,7 +246,7 @@ fn admin_token_info_results(token: &AdminApiToken, now: u64) -> GetAdminTokenInf ), name: params.name.get().to_string(), expiration: params.expiration.get().inner().map(|x| { - DateTime::from_timestamp_millis(*x as i64).expect("invalid timestamp stored in db") + DateTime::from_timestamp_millis(x.0 as i64).expect("invalid timestamp stored in db") }), expired: params.is_expired(now), scope: params.scope.get().0.clone(), @@ -279,7 +280,7 @@ fn apply_token_updates( if let Some(expiration) = updates.expiration { params .expiration - .update(Some(expiration.timestamp_millis() as u64).into()); + .update(Some(ExpirationTime(expiration.timestamp_millis() as u64)).into()); } if updates.never_expires { params.expiration.update(None.into()); diff --git a/src/api/admin/key.rs b/src/api/admin/key.rs index e5024cf8..6a9cecd2 100644 --- a/src/api/admin/key.rs +++ b/src/api/admin/key.rs @@ -8,6 +8,7 @@ use garage_util::time::now_msec; use garage_model::garage::Garage; use garage_model::key_table::*; +use garage_model::permission::ExpirationTime; use crate::api::*; use crate::error::*; @@ -41,7 +42,7 @@ impl RequestHandler for ListKeysRequest { .expect("invalid timestamp stored in db") }), expiration: p.expiration.get().inner().map(|x| { - DateTime::from_timestamp_millis(*x as i64) + DateTime::from_timestamp_millis(x.0 as i64) .expect("invalid timestamp stored in db") }), expired: p.is_expired(now), @@ -218,7 +219,7 @@ async fn key_info_results( DateTime::from_timestamp_millis(x as i64).expect("invalid timestamp stored in db") }), expiration: key_state.expiration.get().inner().map(|x| { - DateTime::from_timestamp_millis(*x as i64).expect("invalid timestamp stored in db") + DateTime::from_timestamp_millis(x.0 as i64).expect("invalid timestamp stored in db") }), expired: key_state.is_expired(now_msec()), access_key_id: key.key_id.clone(), @@ -283,7 +284,7 @@ fn apply_key_updates(key: &mut Key, updates: UpdateKeyRequestBody) -> Result<(), if let Some(expiration) = updates.expiration { key_state .expiration - .update(Some(expiration.timestamp_millis() as u64).into()); + .update(Some(ExpirationTime(expiration.timestamp_millis() as u64)).into()); } if updates.never_expires { key_state.expiration.update(None.into()); diff --git a/src/model/admin_token_table.rs b/src/model/admin_token_table.rs index 21ce2c08..2159bdb6 100644 --- a/src/model/admin_token_table.rs +++ b/src/model/admin_token_table.rs @@ -8,6 +8,7 @@ use garage_table::{EmptyKey, Entry, TableSchema}; pub use crate::key_table::KeyFilter; mod v2 { + use crate::permission::ExpirationTime; use garage_util::crdt; use serde::{Deserialize, Serialize}; @@ -35,7 +36,7 @@ mod v2 { pub name: crdt::Lww, /// The optional time of expiration of the token - pub expiration: crdt::Lww>, + pub expiration: crdt::Lww>, /// The scope of the token, i.e. list of authorized admin API calls pub scope: crdt::Lww, @@ -149,7 +150,7 @@ impl AdminApiTokenParams { pub fn is_expired(&self, ts_now: u64) -> bool { match self.expiration.get().inner() { None => false, - Some(exp) => ts_now >= *exp, + Some(exp) => ts_now >= exp.0, } } diff --git a/src/model/key_table.rs b/src/model/key_table.rs index da0dcb7d..13ad4c8e 100644 --- a/src/model/key_table.rs +++ b/src/model/key_table.rs @@ -51,6 +51,7 @@ mod v08 { mod v2 { use crate::permission::BucketKeyPerm; + use crate::permission::ExpirationTime; use garage_util::crdt; use garage_util::data::Uuid; use serde::{Deserialize, Serialize}; @@ -79,7 +80,7 @@ mod v2 { /// Name for the key pub name: crdt::Lww, /// The optional time of expiration of the key - pub expiration: crdt::Lww>, + pub expiration: crdt::Lww>, /// Flag to allow users having this key to create buckets pub allow_create_bucket: crdt::Lww, @@ -231,7 +232,7 @@ impl KeyParams { pub fn is_expired(&self, ts_now: u64) -> bool { match self.expiration.get().inner() { None => false, - Some(exp) => ts_now >= *exp, + Some(exp) => ts_now >= exp.0, } } } diff --git a/src/model/permission.rs b/src/model/permission.rs index ed9d9c68..6bb53e17 100644 --- a/src/model/permission.rs +++ b/src/model/permission.rs @@ -63,3 +63,15 @@ impl Crdt for BucketKeyPerm { } } } + +/// Expiration date for a key or token +#[derive(PartialOrd, Ord, PartialEq, Eq, Clone, Copy, Debug, Serialize, Deserialize)] +#[cfg_attr(feature = "arbitrary", derive(arbitrary::Arbitrary))] +#[serde(transparent)] +pub struct ExpirationTime(pub u64); + +impl Crdt for ExpirationTime { + fn merge(&mut self, other: &Self) { + self.0 = std::cmp::min(self.0, other.0); + } +} From a646180d7e000477068338e620af0dcf22dd0ae5 Mon Sep 17 00:00:00 2001 From: Alex Auvolat Date: Wed, 13 May 2026 11:47:57 +0200 Subject: [PATCH 3/3] fix fuzz targets --- fuzz/fuzz_targets/admin_api_token_crdt.rs | 3 ++- fuzz/fuzz_targets/key_crdt.rs | 6 +++--- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/fuzz/fuzz_targets/admin_api_token_crdt.rs b/fuzz/fuzz_targets/admin_api_token_crdt.rs index a69ee735..fa1dbec3 100644 --- a/fuzz/fuzz_targets/admin_api_token_crdt.rs +++ b/fuzz/fuzz_targets/admin_api_token_crdt.rs @@ -2,13 +2,14 @@ use garage_fuzz::check_crdt_laws; use garage_model::admin_token_table::{AdminApiToken, AdminApiTokenParams, AdminApiTokenScope}; +use garage_model::permission::ExpirationTime; use garage_util::crdt; use libfuzzer_sys::fuzz_target; type Input = ( bool, crdt::Lww, - crdt::Lww>, + crdt::Lww>, crdt::Lww, ); diff --git a/fuzz/fuzz_targets/key_crdt.rs b/fuzz/fuzz_targets/key_crdt.rs index 0c7b9d07..6e055fa1 100644 --- a/fuzz/fuzz_targets/key_crdt.rs +++ b/fuzz/fuzz_targets/key_crdt.rs @@ -2,7 +2,7 @@ use garage_fuzz::check_crdt_laws; use garage_model::key_table::{Key, KeyParams}; -use garage_model::permission::BucketKeyPerm; +use garage_model::permission::{BucketKeyPerm, ExpirationTime}; use garage_util::crdt; use garage_util::data::Uuid; use libfuzzer_sys::fuzz_target; @@ -10,10 +10,10 @@ use libfuzzer_sys::fuzz_target; type Input = ( bool, crdt::Lww, - crdt::Lww>, + crdt::Lww>, crdt::Lww, crdt::Map, - crdt::LwwMap>, + crdt::LwwMap>, ); fn make(input: Input) -> Key {