gsadmin/garage
1
0
Fork 0
mirror of https://github.com/deuxfleurs-org/garage.git synced 2026-07-26 07:58:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request 'replace Crdt impl on Option by explicit CancelingOption and MergingOption types' (#1451) from option-crdt into main-v2

Browse Source 
Reviewed-on: https://git.deuxfleurs.fr/Deuxfleurs/garage/pulls/1451
This commit is contained in:
Alex
2026-05-13 09:56:29 +00:00
parent eb37a3e11a a646180d7e
commit 91573eb028
23 changed files with 258 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files
fuzz/fuzz_targets/admin_api_token_crdt.rs
+2 -1
View File
@@ -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<String>,
crdt::Lww<Option<u64>>,
crdt::Lww<crdt::MergingOption<ExpirationTime>>,
crdt::Lww<AdminApiTokenScope>,
);
fuzz/fuzz_targets/key_crdt.rs
+3 -3
View File
@@ -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<String>,
crdt::Lww<Option<u64>>,
crdt::Lww<crdt::MergingOption<ExpirationTime>>,
crdt::Lww<bool>,
crdt::Map<Uuid, BucketKeyPerm>,
crdt::LwwMap<String, Option<Uuid>>,
crdt::LwwMap<String, crdt::CancelingOption<Uuid>>,
);
fn make(input: Input) -> Key {
src/api/admin/admin_token.rs
+5 -4
View File
@@ -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::*;
@@ -244,8 +245,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.0 as i64).expect("invalid timestamp stored in db")
}),
expired: params.is_expired(now),
scope: params.scope.get().0.clone(),
@@ -279,10 +280,10 @@ fn apply_token_updates(
if let Some(expiration) = updates.expiration {
params
.expiration
.update(Some(expiration.timestamp_millis() as u64));
.update(Some(ExpirationTime(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));
src/api/admin/bucket.rs
+23 -20
View File
@@ -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::<Vec<_>>(),
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::<Vec<_>>()
}),
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::<Vec<_>>()
@@ -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::<Vec<_>>(),
})
src/api/admin/key.rs
+8 -7
View File
@@ -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::*;
@@ -40,8 +41,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.0 as i64)
.expect("invalid timestamp stored in db")
}),
expired: p.is_expired(now),
@@ -201,7 +202,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 +218,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.0 as i64).expect("invalid timestamp stored in db")
}),
expired: key_state.is_expired(now_msec()),
access_key_id: key.key_id.clone(),
@@ -283,10 +284,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(ExpirationTime(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 {
src/api/admin/special.rs
+1 -1
View File
@@ -164,7 +164,7 @@ async fn check_domain(garage: &Arc<Garage>, domain: &str) -> Result<bool, Error>
}
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),
src/api/common/cors.rs
+13 -10
View File
@@ -19,7 +19,7 @@ pub fn find_matching_cors_rule<'a, B>(
bucket_params: &'a BucketParams,
req: &'a Request<B>,
) -> Result<Option<(&'a GarageCorsRule, &'a str)>, 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<B>(
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
}
src/api/s3/bucket.rs
+6 -3
View File
@@ -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<Response<ResBody>, 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.
src/api/s3/cors.rs
+3 -3
View File
@@ -13,7 +13,7 @@ use crate::xml::to_xml_with_header;
pub async fn handle_get_cors(ctx: ReqCtx) -> Result<Response<ResBody>, 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<Response<ResBody>, 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))
src/api/s3/lifecycle.rs
+3 -3
View File
@@ -14,7 +14,7 @@ use garage_model::bucket_table::Bucket;
pub async fn handle_get_lifecycle(ctx: ReqCtx) -> Result<Response<ResBody>, 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<Response<ResBody>, 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))
src/api/s3/website.rs
+3 -3
View File
@@ -16,7 +16,7 @@ pub const X_AMZ_WEBSITE_REDIRECT_LOCATION: HeaderName =
pub async fn handle_get_website(ctx: ReqCtx) -> Result<Response<ResBody>, 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<Response<ResBody>, 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))
src/model/admin_token_table.rs
+5 -4
View File
@@ -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<String>,
/// The optional time of expiration of the token
pub expiration: crdt::Lww<Option<u64>>,
pub expiration: crdt::Lww<crdt::MergingOption<ExpirationTime>>,
/// The scope of the token, i.e. list of authorized admin API calls
pub scope: crdt::Lww<AdminApiTokenScope>,
@@ -106,7 +107,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 +148,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.0,
}
}
src/model/bucket_alias_table.rs
+3 -3
View File
@@ -13,7 +13,7 @@ mod v08 {
#[derive(PartialEq, Eq, Clone, Debug, Serialize, Deserialize)]
pub struct BucketAlias {
pub(super) name: String,
pub state: crdt::Lww<Option<Uuid>>,
pub state: crdt::Lww<crdt::CancelingOption<Uuid>>,
}
impl garage_util::migrate::InitialFormat for BucketAlias {}
@@ -25,12 +25,12 @@ impl BucketAlias {
pub fn new(name: String, ts: u64, bucket_id: Option<Uuid>) -> 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
src/model/bucket_table.rs
+9 -9
View File
@@ -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<Option<WebsiteConfig>>,
pub website_config: crdt::Lww<crdt::CancelingOption<WebsiteConfig>>,
/// CORS rules
pub cors_config: crdt::Lww<Option<Vec<CorsRule>>>,
pub cors_config: crdt::Lww<crdt::CancelingOption<Vec<CorsRule>>>,
/// Lifecycle configuration
#[serde(default)]
pub lifecycle_config: crdt::Lww<Option<Vec<LifecycleRule>>>,
pub lifecycle_config: crdt::Lww<crdt::CancelingOption<Vec<LifecycleRule>>>,
/// Bucket quotas
#[serde(default)]
pub quotas: crdt::Lww<BucketQuotas>,
@@ -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<Option<WebsiteConfig>>,
pub website_config: crdt::Lww<crdt::CancelingOption<WebsiteConfig>>,
/// CORS rules
pub cors_config: crdt::Lww<Option<Vec<CorsRule>>>,
pub cors_config: crdt::Lww<crdt::CancelingOption<Vec<CorsRule>>>,
/// Lifecycle configuration
pub lifecycle_config: crdt::Lww<Option<Vec<LifecycleRule>>>,
pub lifecycle_config: crdt::Lww<crdt::CancelingOption<Vec<LifecycleRule>>>,
/// Bucket quotas
pub quotas: crdt::Lww<BucketQuotas>,
}
@@ -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()),
}
}
src/model/helper/bucket.rs
+15 -13
View File
@@ -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
src/model/helper/locked.rs
+34 -18
View File
@@ -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;
}
}
src/model/key_table.rs
+8 -7
View File
@@ -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<String, Option<Uuid>>,
pub local_aliases: crdt::LwwMap<String, crdt::CancelingOption<Uuid>>,
}
impl garage_util::migrate::InitialFormat for Key {}
@@ -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<String>,
/// The optional time of expiration of the key
pub expiration: crdt::Lww<Option<u64>>,
pub expiration: crdt::Lww<crdt::MergingOption<ExpirationTime>>,
/// Flag to allow users having this key to create buckets
pub allow_create_bucket: crdt::Lww<bool>,
@@ -91,7 +92,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<String, Option<Uuid>>,
pub local_aliases: crdt::LwwMap<String, crdt::CancelingOption<Uuid>>,
}
impl garage_util::migrate::Migrate for Key {
@@ -106,7 +107,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 +125,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 +230,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.0,
}
}
}
src/model/permission.rs
+12
View File
@@ -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);
}
}
src/model/s3/lifecycle_worker.rs
+1 -1
View File
@@ -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) {
src/util/crdt/crdt.rs
-22
View File
@@ -26,28 +26,6 @@ pub trait Crdt {
fn merge(&mut self, other: &Self);
}
/// `Option<T>` 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<T>` 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<T> Crdt for Option<T>
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.
src/util/crdt/mod.rs
+2
View File
@@ -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::*;
src/util/crdt/option.rs
+97
View File
@@ -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<T>(pub Option<T>);
/// `CancelingOption<T>` 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<T>` 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<T> Crdt for CancelingOption<T>
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<T> CancelingOption<T> {
pub fn inner(&self) -> Option<&T> {
self.0.as_ref()
}
pub fn into_inner(self) -> Option<T> {
self.0
}
pub fn map<U>(self, f: impl FnOnce(T) -> U) -> CancelingOption<U> {
CancelingOption(self.0.map(f))
}
}
impl<T> From<Option<T>> for CancelingOption<T> {
fn from(x: Option<T>) -> Self {
Self(x)
}
}
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, Copy, Default)]
#[cfg_attr(feature = "arbitrary", derive(arbitrary::Arbitrary))]
#[serde(transparent)]
pub struct MergingOption<T>(pub Option<T>);
/// `MergingOption<T>` implements `Crdt` when `T` implements `Crdt`:
/// None is a bottom value, and different Some values get merged according
/// to their Crdt operator.
impl<T> Crdt for MergingOption<T>
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<T> MergingOption<T> {
pub fn inner(&self) -> Option<&T> {
self.0.as_ref()
}
pub fn into_inner(self) -> Option<T> {
self.0
}
pub fn map<U>(self, f: impl FnOnce(T) -> U) -> MergingOption<U> {
MergingOption(self.0.map(f))
}
}
impl<T> From<Option<T>> for MergingOption<T> {
fn from(x: Option<T>) -> Self {
Self(x)
}
}
src/web/web_server.rs
+2 -2
View File
@@ -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