diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 93ce3f832..a4fd436ac 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -199,16 +199,16 @@ pub mod bucket { pub use crate::bucket::metadata_sys::{ BucketMetadataMutationGuard, BucketMetadataSys, ObjectLockConfigState, acquire_bucket_metadata_transaction_lock, acquire_bucket_metadata_transaction_lock_for_incarnation, acquire_scanner_bucket_incarnation_fence, - capture_bucket_metadata_incarnation, delete, delete_if_incarnation, delete_under_transaction_lock, get, - get_accelerate_config, get_bucket_policy, get_bucket_policy_raw, get_bucket_targets_config, get_config_from_disk, - get_cors_config, get_durability_config, get_global_bucket_metadata_sys, get_lifecycle_config, get_logging_config, - get_notification_config, get_object_lock_config, get_object_lock_config_state, get_on_demand_migration_config, - get_public_access_block_config, get_quota_config, get_replication_config, get_request_payment_config, get_sse_config, - get_tagging_config, get_versioning_config, get_website_config, init_bucket_metadata_sys, list_bucket_targets, - reload_bucket_metadata, remove_bucket_metadata, set_bucket_metadata, update, - update_bucket_targets_under_transaction_lock, update_config_with, update_if_incarnation, update_if_incarnation_at, - update_quota_if_incarnation, update_quota_if_incarnation_at, update_under_transaction_lock, - update_under_transaction_lock_at, + capture_bucket_metadata_incarnation, delete, delete_if_incarnation, delete_if_incarnation_at, + delete_under_transaction_lock, get, get_accelerate_config, get_bucket_policy, get_bucket_policy_raw, + get_bucket_targets_config, get_config_from_disk, get_cors_config, get_durability_config, + get_global_bucket_metadata_sys, get_lifecycle_config, get_logging_config, get_notification_config, + get_object_lock_config, get_object_lock_config_state, get_on_demand_migration_config, get_public_access_block_config, + get_quota_config, get_replication_config, get_request_payment_config, get_sse_config, get_tagging_config, + get_versioning_config, get_website_config, init_bucket_metadata_sys, list_bucket_targets, reload_bucket_metadata, + remove_bucket_metadata, set_bucket_metadata, update, update_bucket_targets_under_transaction_lock, + update_config_with, update_if_incarnation, update_if_incarnation_at, update_quota_if_incarnation, + update_quota_if_incarnation_at, update_under_transaction_lock, update_under_transaction_lock_at, }; } diff --git a/crates/ecstore/src/bucket/metadata_sys.rs b/crates/ecstore/src/bucket/metadata_sys.rs index b04dd9989..91f51414e 100644 --- a/crates/ecstore/src/bucket/metadata_sys.rs +++ b/crates/ecstore/src/bucket/metadata_sys.rs @@ -617,6 +617,30 @@ pub async fn delete_if_incarnation(bucket: &str, config_file: &str, expected_inc bucket, config_file, Some(expected_incarnation_id), + None, + )) + .await +} + +/// [`delete_if_incarnation`] stamping the cleared config with `updated_at` +/// (a replicated deletion's source time) instead of the local clock. +/// +/// The stamp survives the deletion as the config's `*_config_updated_at`, and +/// that is what the next incoming item is judged against: a local stamp on +/// the delete would reject a newer source re-create that was merely delivered +/// later (backlog#2292). See [`update_if_incarnation_at`]. +pub async fn delete_if_incarnation_at( + bucket: &str, + config_file: &str, + expected_incarnation_id: Uuid, + updated_at: OffsetDateTime, +) -> Result { + Box::pin(delete_with_sys_expected( + get_bucket_metadata_sys()?, + bucket, + config_file, + Some(expected_incarnation_id), + Some(updated_at), )) .await } @@ -659,17 +683,20 @@ async fn update_with_sys_expected( /// [`delete`] against an explicitly supplied metadata system. See /// [`update_with_sys`]. async fn delete_with_sys(sys: Arc>, bucket: &str, config_file: &str) -> Result { - delete_with_sys_expected(sys, bucket, config_file, None).await + delete_with_sys_expected(sys, bucket, config_file, None, None).await } +/// `updated_at`: `None` stamps the local clock; `Some` persists a replicated +/// deletion's source time (backlog#2292). async fn delete_with_sys_expected( sys: Arc>, bucket: &str, config_file: &str, expected_incarnation_id: Option, + updated_at: Option, ) -> Result { let guard = acquire_config_write_guard_for_incarnation(sys.clone(), bucket, expected_incarnation_id).await?; - delete_under_config_write_guard(sys, &guard, config_file).await + delete_under_config_write_guard(sys, &guard, config_file, updated_at).await } /// Owns the complete bucket-config mutation fence. @@ -840,7 +867,7 @@ pub async fn delete_under_transaction_lock( config_file: &str, ) -> Result { guard.ensure_valid(bucket)?; - delete_under_config_write_guard(get_bucket_metadata_sys()?, guard, config_file).await + delete_under_config_write_guard(get_bucket_metadata_sys()?, guard, config_file, None).await } pub async fn update_quota_if_incarnation( @@ -928,6 +955,7 @@ async fn delete_under_config_write_guard( sys: Arc>, guard: &BucketMetadataMutationGuard, config_file: &str, + updated_at: Option, ) -> Result { guard.ensure_valid(&guard.bucket)?; let metadata_sys = sys.read().await.clone(); @@ -939,7 +967,7 @@ async fn delete_under_config_write_guard( Some(&guard.transaction_guard), &guard.bucket, "bucket config deletion transaction", - metadata_sys.update_checked(&guard.bucket, config_file, Vec::new(), false, guard.incarnation_id, None), + metadata_sys.update_checked(&guard.bucket, config_file, Vec::new(), false, guard.incarnation_id, updated_at), ), ) .await?; @@ -3890,6 +3918,55 @@ mod tests { ); } + /// backlog#2292: a replicated delete persists the source time as the + /// cleared config's `*_config_updated_at`, so the receive-side gate + /// (source time against stored stamp) lets a newer source re-create land + /// even when the delete was applied later than the re-create's source + /// time; the plain delete keeps stamping the local clock. + #[tokio::test] + async fn explicit_updated_at_is_persisted_by_a_delete() { + let (dirs, ecstore) = isolated_store_over_temp_disks().await; + let bucket = "source-stamped-delete"; + for dir in &dirs { + std::fs::create_dir_all(dir.path().join(bucket)).expect("bucket volume should be created"); + } + let sys = Arc::new(RwLock::new(BucketMetadataSys::new(ecstore))); + let policy = br#"{"Version":"2012-10-17","Statement":[]}"#.to_vec(); + let created_at = OffsetDateTime::now_utc() - Duration::from_secs(3 * 3600); + let deleted_at = created_at + Duration::from_secs(60); + let recreated_at = deleted_at + Duration::from_secs(60); + + update_with_sys_expected(sys.clone(), bucket, BUCKET_POLICY_CONFIG, policy.clone(), None, Some(created_at)) + .await + .expect("source-stamped policy write should persist"); + let stamped = delete_with_sys_expected(sys.clone(), bucket, BUCKET_POLICY_CONFIG, None, Some(deleted_at)) + .await + .expect("source-stamped policy delete should persist"); + assert_eq!(stamped, deleted_at); + + let metadata_sys = sys.read().await.clone(); + metadata_sys.metadata_map.write().await.clear(); + let reloaded = metadata_sys.get_config_from_disk(bucket).await.expect("reload from disk"); + assert!(reloaded.policy_config_json.is_empty(), "the delete cleared the payload"); + assert_eq!(reloaded.policy_config_updated_at, deleted_at, "the delete kept the source stamp"); + assert!( + recreated_at >= reloaded.policy_config_updated_at, + "a re-create newer than the delete's source time is not stale against the stored stamp" + ); + + // The plain delete path is unchanged: stamped with the local clock. + update_with_sys_expected(sys.clone(), bucket, BUCKET_POLICY_CONFIG, policy, None, Some(recreated_at)) + .await + .expect("re-create should persist"); + let before = OffsetDateTime::now_utc(); + let stamped = delete_with_sys_expected(sys.clone(), bucket, BUCKET_POLICY_CONFIG, None, None) + .await + .expect("locally stamped delete should persist"); + assert!(stamped >= before, "the plain delete path must keep stamping the local clock"); + let reloaded = metadata_sys.get_config_from_disk(bucket).await.expect("reload from disk"); + assert_eq!(reloaded.policy_config_updated_at, stamped); + } + /// The load and the persisted write share one write guard, so concurrent /// rewrites of the same config compose instead of clobbering each other. /// Moving the load outside that guard loses all but the last tag. diff --git a/crates/iam/src/manager.rs b/crates/iam/src/manager.rs index 105254456..8b295059e 100644 --- a/crates/iam/src/manager.rs +++ b/crates/iam/src/manager.rs @@ -555,6 +555,17 @@ where } pub async fn set_policy(&self, name: &str, policy: Policy) -> Result { + self.set_policy_at(name, policy, OffsetDateTime::now_utc()).await + } + + /// [`Self::set_policy`] stamping the document with `updated_at` instead + /// of the local clock. + /// + /// A site-replication receiver passes the edit's source time: the next + /// incoming revision is judged against the stored `UpdateDate`, so a + /// local stamp would reject a newer source edit that was merely delivered + /// later (backlog#2291). The returned stamp is the one persisted. + pub async fn set_policy_at(&self, name: &str, policy: Policy, updated_at: OffsetDateTime) -> Result { if name.is_empty() || policy.is_empty() { return Err(Error::InvalidArgument); } @@ -565,18 +576,17 @@ where .get(name) .map(|v| { let mut p = v.clone(); - p.update(policy.clone()); + p.update_at(policy.clone(), updated_at); p }) - .unwrap_or_else(|| PolicyDoc::new(policy)); + .unwrap_or_else(|| PolicyDoc::new_at(policy, updated_at)); self.api.save_policy_doc(name, policy_doc.clone()).await?; - let now = OffsetDateTime::now_utc(); + self.cache + .add_or_update_policy_doc(name, &policy_doc, OffsetDateTime::now_utc()); - self.cache.add_or_update_policy_doc(name, &policy_doc, now); - - Ok(now) + Ok(updated_at) } pub async fn list_policies(&self, bucket_name: &str) -> Result> { @@ -810,6 +820,12 @@ where /// create a service account and update cache pub async fn add_service_account(&self, cred: Credentials) -> Result { + self.add_service_account_at(cred, OffsetDateTime::now_utc()).await + } + + /// [`Self::add_service_account`] stamping the identity with `updated_at` + /// instead of the local clock; see [`Self::set_policy_at`] (backlog#2291). + pub async fn add_service_account_at(&self, cred: Credentials, updated_at: OffsetDateTime) -> Result { if cred.access_key.is_empty() || cred.parent_user.is_empty() { return Err(Error::InvalidArgument); } @@ -821,7 +837,8 @@ where } drop(cache); - let u = UserIdentity::new(cred); + let mut u = UserIdentity::new(cred); + u.update_at = Some(updated_at); self.api .save_user_identity(&u.credentials.access_key, UserType::Svc, u.clone(), None) @@ -829,10 +846,22 @@ where self.update_user_with_claims(&u.credentials.access_key, u.clone())?; - Ok(OffsetDateTime::now_utc()) + Ok(updated_at) } pub async fn update_service_account(&self, name: &str, opts: UpdateServiceAccountOpts) -> Result { + self.update_service_account_at(name, opts, OffsetDateTime::now_utc()).await + } + + /// [`Self::update_service_account`] stamping the identity with + /// `updated_at` instead of the local clock; see [`Self::set_policy_at`] + /// (backlog#2291). + pub async fn update_service_account_at( + &self, + name: &str, + opts: UpdateServiceAccountOpts, + updated_at: OffsetDateTime, + ) -> Result { let _mutation_guard = self.cache.service_account_mutation_lock().lock().await; let cache = self.cache.snapshot(); let Some(ui) = cache.users.get(name).cloned() else { @@ -879,13 +908,7 @@ where } if let Some(status) = opts.status { - match status.as_str() { - val if val == AccountStatus::Enabled.as_ref() => cr.status = auth::ACCOUNT_ON.to_owned(), - val if val == AccountStatus::Disabled.as_ref() => cr.status = auth::ACCOUNT_OFF.to_owned(), - auth::ACCOUNT_ON => cr.status = auth::ACCOUNT_ON.to_owned(), - auth::ACCOUNT_OFF => cr.status = auth::ACCOUNT_OFF.to_owned(), - _ => cr.status = auth::ACCOUNT_OFF.to_owned(), - } + cr.status = account_status_flag(&status).to_owned(); } let mut m: HashMap = if token_without_expiration { @@ -937,8 +960,8 @@ where cr.session_token = jwt_sign(&m, &cr.secret_key)?; - let u = UserIdentity::new(cr); - let updated_at = u.update_at.unwrap_or_else(OffsetDateTime::now_utc); + let mut u = UserIdentity::new(cr); + u.update_at = Some(updated_at); self.api .save_user_identity(&u.credentials.access_key, UserType::Svc, u.clone(), None) .await?; @@ -1170,6 +1193,20 @@ where Ok((policies.into_iter().collect(), update_at)) } pub async fn policy_db_set(&self, name: &str, user_type: UserType, is_group: bool, policy: &str) -> Result { + self.policy_db_set_at(name, user_type, is_group, policy, OffsetDateTime::now_utc()) + .await + } + + /// [`Self::policy_db_set`] stamping the mapping with `updated_at` instead + /// of the local clock; see [`Self::set_policy_at`] (backlog#2291). + pub async fn policy_db_set_at( + &self, + name: &str, + user_type: UserType, + is_group: bool, + policy: &str, + updated_at: OffsetDateTime, + ) -> Result { if name.is_empty() { return Err(Error::InvalidArgument); } @@ -1189,10 +1226,11 @@ where self.cache.delete_user_policy(name, OffsetDateTime::now_utc()); } - return Ok(OffsetDateTime::now_utc()); + return Ok(updated_at); } - let mp = MappedPolicy::new(policy); + let mut mp = MappedPolicy::new(policy); + mp.update_at = updated_at; let cache = self.cache.snapshot(); let policy_docs_cache = Arc::clone(&cache.policy_docs); @@ -1215,7 +1253,7 @@ where self.cache.add_or_update_user_policy(name, &mp, OffsetDateTime::now_utc()); } - Ok(OffsetDateTime::now_utc()) + Ok(updated_at) } pub async fn set_temp_user(&self, access_key: &str, cred: &Credentials, policy_name: Option<&str>) -> Result { @@ -1412,6 +1450,17 @@ where } pub async fn add_user(&self, access_key: &str, args: &AddOrUpdateUserReq) -> Result { + self.add_user_at(access_key, args, OffsetDateTime::now_utc()).await + } + + /// [`Self::add_user`] stamping the identity with `updated_at` instead of + /// the local clock; see [`Self::set_policy_at`] (backlog#2291). + pub async fn add_user_at( + &self, + access_key: &str, + args: &AddOrUpdateUserReq, + updated_at: OffsetDateTime, + ) -> Result { let cache = self.cache.snapshot(); let users = Arc::clone(&cache.users); if let Some(x) = users.get(access_key) { @@ -1429,12 +1478,13 @@ where _ => auth::ACCOUNT_OFF, } }; - let user_entry = UserIdentity::from(Credentials { + let mut user_entry = UserIdentity::from(Credentials { access_key: access_key.to_string(), secret_key: args.secret_key.to_string(), status: status.to_owned(), ..Default::default() }); + user_entry.update_at = Some(updated_at); self.api .save_user_identity(access_key, UserType::Reg, user_entry.clone(), None) @@ -1442,7 +1492,7 @@ where self.update_user_with_claims(access_key, user_entry)?; - Ok(OffsetDateTime::now_utc()) + Ok(updated_at) } pub async fn delete_user(&self, access_key: &str, utype: UserType) -> Result<()> { @@ -1620,6 +1670,17 @@ where } pub async fn set_user_status(&self, access_key: &str, status: AccountStatus) -> Result { + self.set_user_status_at(access_key, status, OffsetDateTime::now_utc()).await + } + + /// [`Self::set_user_status`] stamping the identity with `updated_at` + /// instead of the local clock; see [`Self::set_policy_at`] (backlog#2291). + pub async fn set_user_status_at( + &self, + access_key: &str, + status: AccountStatus, + updated_at: OffsetDateTime, + ) -> Result { if access_key.is_empty() { return Err(Error::InvalidArgument); } @@ -1646,12 +1707,13 @@ where } }; - let user_entry = UserIdentity::from(Credentials { + let mut user_entry = UserIdentity::from(Credentials { access_key: access_key.to_string(), secret_key: u.credentials.secret_key.clone(), status: status.to_owned(), ..Default::default() }); + user_entry.update_at = Some(updated_at); drop(cache); drop(users); @@ -1661,7 +1723,7 @@ where self.update_user_with_claims(access_key, user_entry)?; - Ok(OffsetDateTime::now_utc()) + Ok(updated_at) } fn update_user_with_claims(&self, k: &str, u: UserIdentity) -> Result<()> { @@ -1697,6 +1759,17 @@ where } pub async fn add_users_to_group(&self, group: &str, members: Vec) -> Result { + self.add_users_to_group_at(group, members, OffsetDateTime::now_utc()).await + } + + /// [`Self::add_users_to_group`] stamping the group with `updated_at` + /// instead of the local clock; see [`Self::set_policy_at`] (backlog#2291). + pub async fn add_users_to_group_at( + &self, + group: &str, + members: Vec, + updated_at: OffsetDateTime, + ) -> Result { if group.is_empty() { return Err(Error::InvalidArgument); } @@ -1717,7 +1790,7 @@ where // The group's own timestamp moves with every membership or status // change: site replication judges an incoming group item against it // (backlog#2291), so it must reflect the last change, not creation. - let now = OffsetDateTime::now_utc(); + let now = updated_at; let gi = match cache.groups.get(group) { Some(res) => { let mut gi = res.clone(); @@ -1729,7 +1802,11 @@ where gi.update_at = Some(now); gi } - None => GroupInfo::new(members.clone()), + None => { + let mut gi = GroupInfo::new(members.clone()); + gi.update_at = Some(now); + gi + } }; drop(cache); @@ -1750,6 +1827,12 @@ where } pub async fn set_group_status(&self, name: &str, enable: bool) -> Result { + self.set_group_status_at(name, enable, OffsetDateTime::now_utc()).await + } + + /// [`Self::set_group_status`] stamping the group with `updated_at` instead + /// of the local clock; see [`Self::set_policy_at`] (backlog#2291). + pub async fn set_group_status_at(&self, name: &str, enable: bool, updated_at: OffsetDateTime) -> Result { if name.is_empty() { return Err(Error::InvalidArgument); } @@ -1767,7 +1850,7 @@ where } else { gi.status = STATUS_DISABLED.to_owned(); } - let now = OffsetDateTime::now_utc(); + let now = updated_at; gi.update_at = Some(now); self.api.save_group_info(name, gi.clone()).await?; @@ -1844,6 +1927,20 @@ where name: &str, members: Vec, update_cache_only: bool, + ) -> Result { + self.remove_members_from_group_at(name, members, update_cache_only, OffsetDateTime::now_utc()) + .await + } + + /// [`Self::remove_members_from_group`] stamping the group with + /// `updated_at` instead of the local clock; see [`Self::set_policy_at`] + /// (backlog#2291). + pub async fn remove_members_from_group_at( + &self, + name: &str, + members: Vec, + update_cache_only: bool, + updated_at: OffsetDateTime, ) -> Result { let cache = self.cache.snapshot(); let mut gi = cache @@ -1856,7 +1953,7 @@ where let s: HashSet<&String> = HashSet::from_iter(gi.members.iter()); let d: HashSet<&String> = HashSet::from_iter(members.iter()); gi.members = s.difference(&d).map(|v| v.to_string()).collect::>(); - let now = OffsetDateTime::now_utc(); + let now = updated_at; gi.update_at = Some(now); if !update_cache_only { @@ -1880,6 +1977,19 @@ where } pub async fn remove_users_from_group(&self, group: &str, members: Vec) -> Result { + self.remove_users_from_group_at(group, members, OffsetDateTime::now_utc()) + .await + } + + /// [`Self::remove_users_from_group`] stamping the group with `updated_at` + /// instead of the local clock; a group delete (no members) leaves no + /// record and returns the stamp unchanged (backlog#2291). + pub async fn remove_users_from_group_at( + &self, + group: &str, + members: Vec, + updated_at: OffsetDateTime, + ) -> Result { if group.is_empty() { return Err(Error::InvalidArgument); } @@ -1928,18 +2038,17 @@ where return Err(err); } - let now = self.cache.with_write_lock(|cache| { + self.cache.with_write_lock(|cache| { let now = OffsetDateTime::now_utc(); self.remove_group_from_memberships_map_unlocked(cache, group, now); cache.delete_group(group, now); cache.delete_group_policy(group, now); - now }); - return Ok(now); + return Ok(updated_at); } - self.remove_members_from_group(group, members, false).await + self.remove_members_from_group_at(group, members, false, updated_at).await } fn remove_group_from_memberships_map_unlocked(&self, cache: &mut LockedCache, group: &str, now: OffsetDateTime) { @@ -2261,6 +2370,19 @@ where } } +/// The stored `status` flag for a service-account status given on the admin +/// or replication wire: the madmin `enabled` / `disabled` words and the stored +/// `on` / `off` flags are both accepted; anything else disables the account. +pub(crate) fn account_status_flag(status: &str) -> &'static str { + match status { + val if val == AccountStatus::Enabled.as_ref() => auth::ACCOUNT_ON, + val if val == AccountStatus::Disabled.as_ref() => auth::ACCOUNT_OFF, + auth::ACCOUNT_ON => auth::ACCOUNT_ON, + auth::ACCOUNT_OFF => auth::ACCOUNT_OFF, + _ => auth::ACCOUNT_OFF, + } +} + pub fn get_default_policies() -> HashMap { let default_policies = &DEFAULT_POLICIES; default_policies diff --git a/crates/iam/src/sys.rs b/crates/iam/src/sys.rs index 58ccbeb9a..037eb3d4f 100644 --- a/crates/iam/src/sys.rs +++ b/crates/iam/src/sys.rs @@ -385,7 +385,14 @@ impl IamSys { } pub async fn set_policy(&self, name: &str, policy: Policy) -> Result { - let updated_at = self.store.set_policy(name, policy).await?; + self.set_policy_at(name, policy, OffsetDateTime::now_utc()).await + } + + /// [`Self::set_policy`] stamping the document with `updated_at` (a + /// replicated edit's source time) instead of the local clock; see + /// `IamCache::set_policy_at` (backlog#2291). + pub async fn set_policy_at(&self, name: &str, policy: Policy, updated_at: OffsetDateTime) -> Result { + let updated_at = self.store.set_policy_at(name, policy, updated_at).await?; if !self.has_watcher() { for r in notify_iam_load_policy(name).await { @@ -643,7 +650,18 @@ impl IamSys { } pub async fn set_user_status(&self, name: &str, status: rustfs_madmin::AccountStatus) -> Result { - let updated_at = self.store.set_user_status(name, status).await?; + self.set_user_status_at(name, status, OffsetDateTime::now_utc()).await + } + + /// [`Self::set_user_status`] stamping the identity with `updated_at` (a + /// replicated edit's source time) instead of the local clock (backlog#2291). + pub async fn set_user_status_at( + &self, + name: &str, + status: rustfs_madmin::AccountStatus, + updated_at: OffsetDateTime, + ) -> Result { + let updated_at = self.store.set_user_status_at(name, status, updated_at).await?; self.notify_for_user(name, false).await; @@ -655,6 +673,20 @@ impl IamSys { parent_user: &str, groups: Option>, opts: NewServiceAccountOpts, + ) -> Result<(Credentials, OffsetDateTime)> { + self.new_service_account_at(parent_user, groups, opts, OffsetDateTime::now_utc()) + .await + } + + /// [`Self::new_service_account`] stamping the identity with `updated_at` + /// (a replicated edit's source time) instead of the local clock + /// (backlog#2291). + pub async fn new_service_account_at( + &self, + parent_user: &str, + groups: Option>, + opts: NewServiceAccountOpts, + updated_at: OffsetDateTime, ) -> Result<(Credentials, OffsetDateTime)> { if parent_user.is_empty() { return Err(IamError::InvalidArgument); @@ -724,11 +756,18 @@ impl IamSys { let mut cred = create_new_credentials_with_metadata(&access_key, &secret_key, &m, &secret_key)?; cred.parent_user = parent_user.to_owned(); cred.groups = groups; - cred.status = ACCOUNT_ON.to_owned(); + // The status is part of the created identity: a replicated disabled + // account must never exist enabled, not even between a create and a + // follow-up status write (backlog#2289). + cred.status = opts + .status + .as_deref() + .map_or(ACCOUNT_ON, crate::manager::account_status_flag) + .to_owned(); cred.name = opts.name; cred.description = opts.description; - let create_at = self.store.add_service_account(cred.clone()).await?; + let create_at = self.store.add_service_account_at(cred.clone(), updated_at).await?; self.notify_for_service_account(&cred.access_key).await; @@ -736,11 +775,23 @@ impl IamSys { } pub async fn update_service_account(&self, name: &str, opts: UpdateServiceAccountOpts) -> Result { + self.update_service_account_at(name, opts, OffsetDateTime::now_utc()).await + } + + /// [`Self::update_service_account`] stamping the identity with + /// `updated_at` (a replicated edit's source time) instead of the local + /// clock (backlog#2291). + pub async fn update_service_account_at( + &self, + name: &str, + opts: UpdateServiceAccountOpts, + updated_at: OffsetDateTime, + ) -> Result { if name == SITE_REPLICATOR_SERVICE_ACCOUNT && !opts.allow_site_replicator_account { return Err(IamError::IAMActionNotAllowed); } - let updated_at = self.store.update_service_account(name, opts).await?; + let updated_at = self.store.update_service_account_at(name, opts, updated_at).await?; self.notify_for_service_account(name).await; @@ -940,6 +991,17 @@ impl IamSys { } pub async fn create_user(&self, access_key: &str, args: &AddOrUpdateUserReq) -> Result { + self.create_user_at(access_key, args, OffsetDateTime::now_utc()).await + } + + /// [`Self::create_user`] stamping the identity with `updated_at` (a + /// replicated edit's source time) instead of the local clock (backlog#2291). + pub async fn create_user_at( + &self, + access_key: &str, + args: &AddOrUpdateUserReq, + updated_at: OffsetDateTime, + ) -> Result { if !is_access_key_valid(access_key) { return Err(IamError::InvalidAccessKeyLength); } @@ -952,7 +1014,7 @@ impl IamSys { return Err(IamError::InvalidSecretKeyLength); } - let updated_at = self.store.add_user(access_key, args).await?; + let updated_at = self.store.add_user_at(access_key, args, updated_at).await?; self.load_user(access_key, UserType::Reg).await?; self.notify_for_user(access_key, false).await; @@ -1026,10 +1088,21 @@ impl IamSys { } pub async fn add_users_to_group(&self, group: &str, users: Vec) -> Result { + self.add_users_to_group_at(group, users, OffsetDateTime::now_utc()).await + } + + /// [`Self::add_users_to_group`] stamping the group with `updated_at` (a + /// replicated edit's source time) instead of the local clock (backlog#2291). + pub async fn add_users_to_group_at( + &self, + group: &str, + users: Vec, + updated_at: OffsetDateTime, + ) -> Result { if contains_reserved_chars(group) { return Err(IamError::GroupNameContainsReservedChars); } - let updated_at = self.store.add_users_to_group(group, users).await?; + let updated_at = self.store.add_users_to_group_at(group, users, updated_at).await?; self.notify_for_group(group).await; @@ -1037,7 +1110,19 @@ impl IamSys { } pub async fn remove_users_from_group(&self, group: &str, users: Vec) -> Result { - let updated_at = self.store.remove_users_from_group(group, users).await?; + self.remove_users_from_group_at(group, users, OffsetDateTime::now_utc()).await + } + + /// [`Self::remove_users_from_group`] stamping the group with `updated_at` + /// (a replicated edit's source time) instead of the local clock + /// (backlog#2291). + pub async fn remove_users_from_group_at( + &self, + group: &str, + users: Vec, + updated_at: OffsetDateTime, + ) -> Result { + let updated_at = self.store.remove_users_from_group_at(group, users, updated_at).await?; self.notify_for_group(group).await; @@ -1045,7 +1130,13 @@ impl IamSys { } pub async fn set_group_status(&self, group: &str, enable: bool) -> Result { - let updated_at = self.store.set_group_status(group, enable).await?; + self.set_group_status_at(group, enable, OffsetDateTime::now_utc()).await + } + + /// [`Self::set_group_status`] stamping the group with `updated_at` (a + /// replicated edit's source time) instead of the local clock (backlog#2291). + pub async fn set_group_status_at(&self, group: &str, enable: bool, updated_at: OffsetDateTime) -> Result { + let updated_at = self.store.set_group_status_at(group, enable, updated_at).await?; self.notify_for_group(group).await; @@ -1080,7 +1171,24 @@ impl IamSys { } pub async fn policy_db_set(&self, name: &str, user_type: UserType, is_group: bool, policy: &str) -> Result { - let updated_at = self.store.policy_db_set(name, user_type, is_group, policy).await?; + self.policy_db_set_at(name, user_type, is_group, policy, OffsetDateTime::now_utc()) + .await + } + + /// [`Self::policy_db_set`] stamping the mapping with `updated_at` (a + /// replicated edit's source time) instead of the local clock (backlog#2291). + pub async fn policy_db_set_at( + &self, + name: &str, + user_type: UserType, + is_group: bool, + policy: &str, + updated_at: OffsetDateTime, + ) -> Result { + let updated_at = self + .store + .policy_db_set_at(name, user_type, is_group, policy, updated_at) + .await?; if !self.has_watcher() { for r in notify_iam_load_policy_mapping(name, user_type.to_u64(), is_group).await { @@ -1862,6 +1970,11 @@ pub struct NewServiceAccountOpts { pub expiration: Option, pub allow_site_replicator_account: bool, pub claims: Option>, + /// Status the account is created with (`enabled` / `disabled` or the + /// stored `on` / `off` flags); `None` creates it enabled. Site + /// replication passes the source account's status so a disabled account + /// is never enabled on the peer, not even transiently (backlog#2289). + pub status: Option, } pub struct UpdateServiceAccountOpts { diff --git a/crates/policy/src/policy/doc.rs b/crates/policy/src/policy/doc.rs index dc2b83fa9..30ef4ca03 100644 --- a/crates/policy/src/policy/doc.rs +++ b/crates/policy/src/policy/doc.rs @@ -45,18 +45,33 @@ pub struct PolicyDoc { impl PolicyDoc { pub fn new(policy: Policy) -> Self { + Self::new_at(policy, OffsetDateTime::now_utc()) + } + + /// [`Self::new`] with an explicit `UpdateDate` (and `CreateDate`). + /// + /// A replicated document keeps the edit's source time: the receiver + /// judges the next incoming revision against the stored stamp, so a + /// local stamp would reject a newer source edit that was merely + /// delivered later. + pub fn new_at(policy: Policy, at: OffsetDateTime) -> Self { Self { version: 1, policy, - create_date: Some(OffsetDateTime::now_utc()), - update_date: Some(OffsetDateTime::now_utc()), + create_date: Some(at), + update_date: Some(at), } } pub fn update(&mut self, policy: Policy) { + self.update_at(policy, OffsetDateTime::now_utc()); + } + + /// [`Self::update`] with an explicit `UpdateDate`; see [`Self::new_at`]. + pub fn update_at(&mut self, policy: Policy, at: OffsetDateTime) { self.version += 1; self.policy = policy; - self.update_date = Some(OffsetDateTime::now_utc()); + self.update_date = Some(at); if self.create_date.is_none() { self.create_date = self.update_date; diff --git a/rustfs/src/admin/handlers/site_replication.rs b/rustfs/src/admin/handlers/site_replication.rs index df667513c..318c8a093 100644 --- a/rustfs/src/admin/handlers/site_replication.rs +++ b/rustfs/src/admin/handlers/site_replication.rs @@ -87,7 +87,7 @@ use std::sync::{LazyLock, Mutex as StdMutex}; use std::time::Duration; use time::OffsetDateTime; use tokio::sync::Mutex; -use tracing::{debug, info, warn}; +use tracing::{info, warn}; use url::Url; use url::form_urlencoded; use uuid::Uuid; @@ -1478,6 +1478,7 @@ async fn set_site_replicator_service_account_secret(parent_user: &str, secret_ke expiration: None, allow_site_replicator_account: true, claims: None, + status: None, }, ) .await @@ -1720,6 +1721,7 @@ async fn reconcile_site_replicator_service_account() -> S3Result<()> { expiration: None, allow_site_replicator_account: true, claims: None, + status: None, }, ) .await @@ -5620,10 +5622,11 @@ async fn apply_bucket_meta_item(item: SRBucketMeta) -> S3Result<()> { }; // Persist the SOURCE `updated_at` as the stored `*_config_updated_at` - // stamp (backlog#2292). The staleness gate above compares the next item's - // source time against that stamp, so stamping the local apply time would - // reject a newer source edit that was merely delivered after this write - // (two quick edits under delivery delay, or a peer clock ahead of ours). + // stamp, on writes and on deletes alike (backlog#2292). The staleness + // gate above compares the next item's source time against that stamp, so + // stamping the local apply time would reject a newer source edit that was + // merely delivered after this write or delete (two quick edits under + // delivery delay, or a peer clock ahead of ours). // Items without a source time keep the local stamp; lc-config keeps it // too: its staleness axis is the in-document `expiry_updated_at` the merge // above records, and the whole-config time is only its deletion / legacy @@ -5684,9 +5687,23 @@ async fn apply_bucket_meta_item(item: SRBucketMeta) -> S3Result<()> { .await .map_err(ApiError::from)?; } else { - metadata_sys::delete_if_incarnation(&item.bucket, config_file, expected_incarnation_id) - .await - .map_err(ApiError::from)?; + // A delete is stamped like a write: the source time survives + // as the config's `*_config_updated_at`, so a newer source + // re-create delivered later is not judged stale against the + // local time this delete landed (backlog#2292). + match source_updated_at { + Some(source_updated_at) => { + metadata_sys::delete_if_incarnation_at( + &item.bucket, + config_file, + expected_incarnation_id, + source_updated_at, + ) + .await + } + None => metadata_sys::delete_if_incarnation(&item.bucket, config_file, expected_incarnation_id).await, + } + .map_err(ApiError::from)?; } } } @@ -5801,26 +5818,18 @@ async fn apply_iam_item(item: SRIAMItem) -> S3Result<()> { let Some(iam_sys) = current_iam_handle() else { return Err(s3_error!(InvalidRequest, "iam not init")); }; - let incoming_updated_at = item.updated_at; - let deletion_mark_entities = iam_item_deletion_mark_entities(&item); - - let verdict = match item.r#type.as_str() { - "policy" => apply_iam_policy_item(&iam_sys, &item.name, item.policy, incoming_updated_at).await?, - "policy-mapping" => apply_iam_policy_mapping_item(&iam_sys, item.policy_mapping, incoming_updated_at).await?, - "group-info" => apply_iam_group_info_item(&iam_sys, item.group_info, incoming_updated_at).await?, + match item.r#type.as_str() { // MinIO madmin-go sends `SRIAMItemSTSAcc = "sts-account"`. The legacy alias // `sts-credential` (emitted by older RustFS releases) stays accepted permanently // so mixed-version RustFS sites keep replicating STS credentials during rolling // upgrades; it is a compatibility layer, not temporary code. + // + // STS credentials carry no source revision and leave no deletion mark, + // so they stay outside the ordered transaction below. SR_IAM_ITEM_STS_ACC | SR_IAM_ITEM_STS_ACC_LEGACY => { - apply_iam_sts_account_item(&iam_sys, item.sts_credential).await?; - IamItemVerdict::Apply - } - "iam-user" => apply_iam_user_item(&iam_sys, item.iam_user, incoming_updated_at).await?, - "service-account" => { - apply_iam_service_account_item(&iam_sys, item.svc_acc_change, incoming_updated_at).await?; - IamItemVerdict::Apply + return apply_iam_sts_account_item(&iam_sys, item.sts_credential).await; } + "policy" | "policy-mapping" | "group-info" | "iam-user" | "service-account" => {} _ => { return Err(s3_error!( NotImplemented, @@ -5828,55 +5837,64 @@ async fn apply_iam_item(item: SRIAMItem) -> S3Result<()> { item.r#type )); } - }; - // A committed deletion leaves no record for the gate to judge later items - // against, so its source timestamp is kept as a mark (backlog#2291). The - // mark is part of applying the deletion: failing here makes the sender - // retry the (idempotent) deletion rather than leave a revoke that a stale - // grant could still undo. - if verdict == IamItemVerdict::Apply - && let Some(deleted_at) = incoming_updated_at.filter(|_| !deletion_mark_entities.is_empty()) - { - commit_iam_deletion_marks(deletion_mark_entities, deleted_at).await?; } - Ok(()) + + // One transaction per item (backlog#2291). The staleness verdict, the IAM + // write and the deletion-mark commit run under the distributed + // state-object lock, so an older grant and a newer revoke delivered + // concurrently — to this node or to a sibling node of this site — are + // applied one after the other, each judged against what the other left + // behind. The write stamps the record with the item's source + // `updated_at`, which is what the next item is judged against: stamping + // the local apply time would reject a newer source edit that was merely + // delivered later. A committed deletion leaves no record, so its source + // timestamp is kept as a mark in the same commit; failing to persist the + // mark fails the item, and the sender retries the (idempotent) deletion + // rather than leaving a revoke that a stale grant could still undo. + with_site_replication_state_transaction(move |mut state| async move { + let incoming_updated_at = item.updated_at; + let deletion_mark_entities = iam_item_deletion_mark_entities(&item); + let verdict = match item.r#type.as_str() { + "policy" => apply_iam_policy_item(&iam_sys, &state, &item.name, item.policy, incoming_updated_at).await?, + "policy-mapping" => apply_iam_policy_mapping_item(&iam_sys, &state, item.policy_mapping, incoming_updated_at).await?, + "group-info" => apply_iam_group_info_item(&iam_sys, &state, item.group_info, incoming_updated_at).await?, + "iam-user" => apply_iam_user_item(&iam_sys, &state, item.iam_user, incoming_updated_at).await?, + "service-account" => { + apply_iam_service_account_item(&iam_sys, &state, item.svc_acc_change, incoming_updated_at).await? + } + _ => unreachable!("unsupported IAM item types are rejected before the transaction"), + }; + let changed = verdict == IamItemVerdict::Apply + && incoming_updated_at + .filter(|_| !deletion_mark_entities.is_empty()) + .is_some_and(|deleted_at| record_iam_deletion_marks(&mut state, &deletion_mark_entities, deleted_at)); + Ok(((), changed.then_some(state))) + }) + .await } -/// The deletion mark consulted by the staleness gate when the targeted record -/// is absent: the newest recorded deletion of any of `entities`. An -/// unreadable state falls back to today's behaviour (no mark, the item is -/// applied) — the gate must not turn a state-object outage into rejected -/// IAM replication. -async fn local_iam_deletion_mark(entities: &[String]) -> Option { - match load_site_replication_state().await { - Ok(state) => iam_deletion_mark(&state, entities), - Err(err) => { - debug!( - component = LOG_COMPONENT_ADMIN, - subsystem = LOG_SUBSYSTEM_SITE_REPLICATION, - event = EVENT_ADMIN_SITE_REPLICATION_STATE, - result = "iam_deletion_mark_unavailable", - error = ?err, - "site replication state unreadable; applying IAM item without a deletion mark" - ); - None - } - } +/// The stamp a replicated write persists on the record: the item's source +/// `updated_at`, or the local clock for an item from a peer that predates +/// timestamps (those keep last-writer-wins, see [`judge_iam_item_staleness`]). +fn replicated_write_stamp(incoming_updated_at: Option) -> OffsetDateTime { + incoming_updated_at.unwrap_or_else(OffsetDateTime::now_utc) } async fn apply_iam_policy_item( iam_sys: &IamSys, + marks: &SiteReplicationState, name: &str, policy: Option, incoming_updated_at: Option, ) -> S3Result { - // Judge the item against the local document's own timestamp so a delayed - // older body (or delete) cannot overwrite a newer edit; once the document - // is deleted, its deletion mark stands in for it (backlog#2291). + // Judge the item against the local document's own timestamp — the source + // time of the edit that wrote it — so a delayed older body (or delete) + // cannot overwrite a newer edit; once the document is deleted, its + // deletion mark stands in for it (backlog#2291). let local_updated_at = match iam_sys.get_policy_doc(name).await { Ok(doc) => Some(doc.update_date.unwrap_or(OffsetDateTime::UNIX_EPOCH)), Err(err) if rustfs_iam::error::is_err_no_such_policy(&err) => { - local_iam_deletion_mark(&[iam_policy_deletion_mark_entity(name)]).await + iam_deletion_mark(marks, &[iam_policy_deletion_mark_entity(name)]) } Err(err) => return Err(ApiError::from(err).into()), }; @@ -5886,7 +5904,10 @@ async fn apply_iam_policy_item( if let Some(policy) = policy { let policy: Policy = serde_json::from_value(policy).map_err(|e| s3_error!(InvalidRequest, "invalid policy body: {}", e))?; - iam_sys.set_policy(name, policy).await.map_err(ApiError::from)?; + iam_sys + .set_policy_at(name, policy, replicated_write_stamp(incoming_updated_at)) + .await + .map_err(ApiError::from)?; } else { // Idempotent delete: the retry drain replays recorded deletions, and // an entity already absent here IS the converged outcome — erroring @@ -5902,6 +5923,7 @@ async fn apply_iam_policy_item( async fn apply_iam_policy_mapping_item( iam_sys: &IamSys, + marks: &SiteReplicationState, policy_mapping: Option, incoming_updated_at: Option, ) -> S3Result { @@ -5918,20 +5940,26 @@ async fn apply_iam_policy_mapping_item( .await { Some(record) => Some(record.update_at), - None => { - local_iam_deletion_mark(&[iam_policy_mapping_deletion_mark_entity( + None => iam_deletion_mark( + marks, + &[iam_policy_mapping_deletion_mark_entity( &mapping.user_or_group, mapping.user_type, mapping.is_group, - )]) - .await - } + )], + ), }; if judge_iam_item_staleness(local_updated_at, incoming_updated_at) == IamItemVerdict::SkipStale { return Ok(IamItemVerdict::SkipStale); } iam_sys - .policy_db_set(&mapping.user_or_group, user_type, mapping.is_group, &mapping.policy) + .policy_db_set_at( + &mapping.user_or_group, + user_type, + mapping.is_group, + &mapping.policy, + replicated_write_stamp(incoming_updated_at), + ) .await .map_err(ApiError::from)?; Ok(IamItemVerdict::Apply) @@ -5939,6 +5967,7 @@ async fn apply_iam_policy_mapping_item( async fn apply_iam_group_info_item( iam_sys: &IamSys, + marks: &SiteReplicationState, group_info: Option, incoming_updated_at: Option, ) -> S3Result { @@ -5963,12 +5992,13 @@ async fn apply_iam_group_info_item( .map(|member| iam_group_member_deletion_mark_entity(&update.group, member)), ) .collect(); - local_iam_deletion_mark(&entities).await + iam_deletion_mark(marks, &entities) } }; if judge_iam_item_staleness(local_updated_at, incoming_updated_at) == IamItemVerdict::SkipStale { return Ok(IamItemVerdict::SkipStale); } + let stamp = replicated_write_stamp(incoming_updated_at); if !group_info_requires_upsert(&update) { // Idempotent removal: a replayed deletion may find the group or a // member already gone (deleted here earlier, or the user tombstone @@ -5985,7 +6015,7 @@ async fn apply_iam_group_info_item( if members.is_empty() && !update.members.is_empty() { return Ok(IamItemVerdict::Apply); } - match iam_sys.remove_users_from_group(&update.group, members).await { + match iam_sys.remove_users_from_group_at(&update.group, members, stamp).await { Ok(_) => {} Err(err) if rustfs_iam::error::is_err_no_such_group(&err) => {} Err(err) => return Err(ApiError::from(err).into()), @@ -5994,11 +6024,11 @@ async fn apply_iam_group_info_item( } iam_sys - .add_users_to_group(&update.group, update.members) + .add_users_to_group_at(&update.group, update.members, stamp) .await .map_err(ApiError::from)?; iam_sys - .set_group_status(&update.group, matches!(update.status, GroupStatus::Enabled)) + .set_group_status_at(&update.group, matches!(update.status, GroupStatus::Enabled), stamp) .await .map_err(ApiError::from)?; Ok(IamItemVerdict::Apply) @@ -6041,6 +6071,7 @@ async fn apply_iam_sts_account_item(iam_sys: &IamSys, sts_credentia async fn apply_iam_user_item( iam_sys: &IamSys, + marks: &SiteReplicationState, iam_user: Option, incoming_updated_at: Option, ) -> S3Result { @@ -6051,11 +6082,12 @@ async fn apply_iam_user_item( // record so a stale re-create cannot resurrect it (backlog#2291). let local_updated_at = match iam_sys.get_user(&user.access_key).await { Some(local) => Some(local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH)), - None => local_iam_deletion_mark(&[iam_user_deletion_mark_entity(&user.access_key)]).await, + None => iam_deletion_mark(marks, &[iam_user_deletion_mark_entity(&user.access_key)]), }; if judge_iam_item_staleness(local_updated_at, incoming_updated_at) == IamItemVerdict::SkipStale { return Ok(IamItemVerdict::SkipStale); } + let stamp = replicated_write_stamp(incoming_updated_at); if user.is_delete_req { iam_sys.delete_user(&user.access_key, true).await.map_err(ApiError::from)?; } else { @@ -6065,12 +6097,12 @@ async fn apply_iam_user_item( let is_status_only_update = user_req.secret_key.is_empty() && user_req.policy.is_none(); if is_status_only_update { iam_sys - .set_user_status(&user.access_key, user_req.status) + .set_user_status_at(&user.access_key, user_req.status, stamp) .await .map_err(ApiError::from)?; } else { iam_sys - .create_user(&user.access_key, &user_req) + .create_user_at(&user.access_key, &user_req, stamp) .await .map_err(ApiError::from)?; } @@ -6080,13 +6112,15 @@ async fn apply_iam_user_item( async fn apply_iam_service_account_item( iam_sys: &IamSys, + marks: &SiteReplicationState, svc_acc_change: Option, incoming_updated_at: Option, -) -> S3Result<()> { +) -> S3Result { let Some(change) = svc_acc_change else { return Err(s3_error!(InvalidRequest, "serviceAccountChange is required")); }; let envelope = change.oidc_service_account_envelope; + let stamp = replicated_write_stamp(incoming_updated_at); if let Some(create) = change.create { // Like the user path: with the account already deleted here, the // recorded deletion mark is the timestamp a stale create/update @@ -6094,11 +6128,11 @@ async fn apply_iam_service_account_item( let local_updated_at = match iam_sys.get_user(&create.access_key).await { Some(local) => Some(local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH)), None if create.access_key == SITE_REPLICATOR_SERVICE_ACCOUNT => None, - None => local_iam_deletion_mark(&[format!("svc-acc:{}", create.access_key)]).await, + None => iam_deletion_mark(marks, &[format!("svc-acc:{}", create.access_key)]), }; let replicated_policy = if create.access_key == SITE_REPLICATOR_SERVICE_ACCOUNT { if local_updated_at.is_some_and(|local_updated_at| is_stale_update(local_updated_at, incoming_updated_at)) { - return Ok(()); + return Ok(IamItemVerdict::SkipStale); } ReplicatedServiceAccountPolicy { policy: Some(site_replicator_service_account_policy()?), @@ -6108,7 +6142,7 @@ async fn apply_iam_service_account_item( let Some(replicated_policy) = decode_service_account_replication_policy(&create, envelope.as_ref(), incoming_updated_at, local_updated_at)? else { - return Ok(()); + return Ok(IamItemVerdict::SkipStale); }; replicated_policy }; @@ -6122,7 +6156,7 @@ async fn apply_iam_service_account_item( )); } iam_sys - .update_service_account( + .update_service_account_at( &create.access_key, UpdateServiceAccountOpts { name: replicated_policy.metadata_for_existing_account(create.name), @@ -6134,15 +6168,19 @@ async fn apply_iam_service_account_item( parent_user: None, allow_site_replicator_account: create.access_key == SITE_REPLICATOR_SERVICE_ACCOUNT, }, + stamp, ) .await .map_err(ApiError::from)?; } Err(err) if is_err_no_such_service_account(&err) => { - let access_key = create.access_key.clone(); - let status = create.status.clone(); + // A snapshot (bootstrap / repair / retry resend) carries the + // account's current status, and the account is created with + // it in the same write: a disabled account must never exist + // enabled here, not even between a create and a follow-up + // status write that might fail (backlog#2289). iam_sys - .new_service_account( + .new_service_account_at( &create.parent, Some(create.groups), NewServiceAccountOpts { @@ -6154,43 +6192,23 @@ async fn apply_iam_service_account_item( expiration: create.expiration, allow_site_replicator_account: true, claims: Some(create.claims), + status: (!create.status.is_empty()).then_some(create.status), }, + stamp, ) .await .map_err(ApiError::from)?; - // A snapshot (bootstrap / repair / retry resend) carries the - // account's current status; creation always enables, so a - // disabled account must be switched off in a second step or - // the peer keeps accepting credentials the source rejects. - if !status.is_empty() && status != "on" { - iam_sys - .update_service_account( - &access_key, - UpdateServiceAccountOpts { - session_policy: None, - secret_key: None, - name: None, - description: None, - expiration: None, - status: Some(status), - parent_user: None, - allow_site_replicator_account: false, - }, - ) - .await - .map_err(ApiError::from)?; - } } Err(err) => return Err(ApiError::from(err).into()), } - return Ok(()); + return Ok(IamItemVerdict::Apply); } if let Some(update) = change.update { if let Some(local) = iam_sys.get_user(&update.access_key).await && is_stale_update(local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH), incoming_updated_at) { - return Ok(()); + return Ok(IamItemVerdict::SkipStale); } let allow_site_replicator_account = update.access_key == SITE_REPLICATOR_SERVICE_ACCOUNT; let session_policy = if allow_site_replicator_account { @@ -6199,7 +6217,7 @@ async fn apply_iam_service_account_item( update.session_policy.as_str().and_then(|raw| serde_json::from_str(raw).ok()) }; iam_sys - .update_service_account( + .update_service_account_at( &update.access_key, UpdateServiceAccountOpts { session_policy, @@ -6213,23 +6231,24 @@ async fn apply_iam_service_account_item( parent_user: None, allow_site_replicator_account, }, + stamp, ) .await .map_err(ApiError::from)?; - return Ok(()); + return Ok(IamItemVerdict::Apply); } if let Some(delete) = change.delete { if let Some(local) = iam_sys.get_user(&delete.access_key).await && is_stale_update(local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH), incoming_updated_at) { - return Ok(()); + return Ok(IamItemVerdict::SkipStale); } iam_sys .delete_service_account(&delete.access_key, true) .await .map_err(ApiError::from)?; - return Ok(()); + return Ok(IamItemVerdict::Apply); } Err(s3_error!(InvalidRequest, "serviceAccountChange is empty")) @@ -6892,6 +6911,7 @@ async fn apply_peer_join_service_account(join_req: SRPeerJoinReq) -> S3Result<() expiration: None, allow_site_replicator_account: join_req.svc_acct_access_key == SITE_REPLICATOR_SERVICE_ACCOUNT, claims: None, + status: None, }, ) .await @@ -8225,6 +8245,343 @@ mod tests { } } + // --- Review regressions on rustfs#7195 (backlog#2289 / #2291 / #2292): + // delivery order and concurrency through the real receiver. + + /// Two-peer state so the apply transaction's persist keeps the state + /// object (a single-peer state is cleared on write) and the deletion + /// marks it records survive between items. + async fn seed_two_peer_state_for_iam_apply() { + let seed = SiteReplicationState { + peers: BTreeMap::from([ + ( + "site-a".to_string(), + PeerInfo { + deployment_id: "site-a".to_string(), + ..peer("site-a", "https://a.example:9000") + }, + ), + ( + "site-b".to_string(), + PeerInfo { + deployment_id: "site-b".to_string(), + ..peer("site-b", "https://b.example:9000") + }, + ), + ]), + ..Default::default() + }; + save_site_replication_state(&seed).await.expect("seed state"); + } + + async fn clear_seeded_state() { + save_site_replication_state(&SiteReplicationState::default()) + .await + .expect("clear state"); + } + + fn sr_item(item_type: &str, updated_at: OffsetDateTime) -> SRIAMItem { + SRIAMItem { + r#type: item_type.to_string(), + updated_at: Some(updated_at), + api_version: Some(SITE_REPL_API_VERSION.to_string()), + ..Default::default() + } + } + + fn allow_actions_policy(actions: &[&str]) -> serde_json::Value { + serde_json::json!({ + "Version": "2012-10-17", + "Statement": [{"Effect": "Allow", "Action": actions, "Resource": ["arn:aws:s3:::*"]}] + }) + } + + fn sr_policy_item(name: &str, body: serde_json::Value, updated_at: OffsetDateTime) -> SRIAMItem { + let mut item = sr_item("policy", updated_at); + item.name = name.to_string(); + item.policy = Some(body); + item + } + + fn sr_mapping_item(user: &str, policy: &str, updated_at: OffsetDateTime) -> SRIAMItem { + let mut item = sr_item("policy-mapping", updated_at); + item.policy_mapping = Some(SRPolicyMapping { + user_or_group: user.to_string(), + user_type: 0, + is_group: false, + policy: policy.to_string(), + ..Default::default() + }); + item + } + + fn sr_group_item(group: &str, members: &[&str], is_remove: bool, updated_at: OffsetDateTime) -> SRIAMItem { + let mut item = sr_item("group-info", updated_at); + item.group_info = Some(SRGroupInfo { + update_req: rustfs_madmin::GroupAddRemove { + group: group.to_string(), + members: members.iter().map(|member| member.to_string()).collect(), + status: rustfs_madmin::GroupStatus::Enabled, + is_remove, + }, + api_version: Some(SITE_REPL_API_VERSION.to_string()), + }); + item + } + + fn sr_user_item( + access_key: &str, + user_req: Option, + updated_at: OffsetDateTime, + ) -> SRIAMItem { + let mut item = sr_item("iam-user", updated_at); + item.iam_user = Some(SRIAMUser { + access_key: access_key.to_string(), + is_delete_req: user_req.is_none(), + user_req, + api_version: Some(SITE_REPL_API_VERSION.to_string()), + }); + item + } + + fn user_req(secret_key: &str, status: rustfs_madmin::AccountStatus) -> rustfs_madmin::AddOrUpdateUserReq { + rustfs_madmin::AddOrUpdateUserReq { + secret_key: secret_key.to_string(), + policy: None, + status, + } + } + + fn sr_service_account_create_item(parent: &str, access_key: &str, status: &str, updated_at: OffsetDateTime) -> SRIAMItem { + let mut item = sr_item("service-account", updated_at); + item.svc_acc_change = Some(SRSvcAccChange { + create: Some(SRSvcAccCreate { + parent: parent.to_string(), + access_key: access_key.to_string(), + secret_key: "replicated-svc-secret-123".to_string(), + groups: Vec::new(), + claims: HashMap::new(), + session_policy: rustfs_madmin::SRSessionPolicy::default(), + status: status.to_string(), + name: String::new(), + description: String::new(), + expiration: None, + api_version: Some(SITE_REPL_API_VERSION.to_string()), + }), + ..Default::default() + }); + item + } + + async fn stored_policy_json(name: &str) -> (String, Option) { + let iam = current_iam_handle().expect("test IAM"); + let doc = iam.get_policy_doc(name).await.expect("policy doc"); + (serde_json::to_string(&doc.policy).expect("serialize policy"), doc.update_date) + } + + /// Review finding on rustfs#7195 (P1): two source edits T1 < T2 that both + /// predate their delivery. The record T1 writes must carry T1 — not the + /// later receive time — or T2 is judged stale against it and the newer + /// revoke is silently dropped. Exercised through the real receiver for + /// every gated item type. + #[tokio::test] + #[serial] + async fn apply_iam_item_applies_delayed_in_order_updates_through_the_receiver() { + publish_ready_iam_context().await; + seed_two_peer_state_for_iam_apply().await; + let iam = current_iam_handle().expect("test IAM"); + let t1 = OffsetDateTime::now_utc() - time::Duration::hours(2); + let t2 = t1 + time::Duration::minutes(5); + + // policy: the grant, then the narrower revision. + let policy = "sr-delayed-order-policy"; + apply_iam_item(sr_policy_item(policy, allow_actions_policy(&["s3:GetObject", "s3:PutObject"]), t1)) + .await + .expect("T1 grant"); + apply_iam_item(sr_policy_item(policy, allow_actions_policy(&["s3:GetObject"]), t2)) + .await + .expect("T2 narrowed body"); + let (stored, stamp) = stored_policy_json(policy).await; + assert_eq!(stamp, Some(t2), "the stored stamp is the source time of the last applied edit"); + assert!( + !stored.contains("s3:PutObject"), + "the narrower T2 body must replace the T1 grant: {stored}" + ); + + // policy-mapping: attach the wide policy, then the narrow one. + for (name, actions) in [ + ("sr-delayed-order-wide", &["s3:*"][..]), + ("sr-delayed-order-narrow", &["s3:GetObject"][..]), + ] { + let body: rustfs_policy::policy::Policy = serde_json::from_value(allow_actions_policy(actions)).expect("policy body"); + iam.set_policy(name, body).await.expect("local policy"); + } + let user = "sr-delayed-order-user"; + apply_iam_item(sr_mapping_item(user, "sr-delayed-order-wide", t1)) + .await + .expect("T1 attach"); + apply_iam_item(sr_mapping_item(user, "sr-delayed-order-narrow", t2)) + .await + .expect("T2 attach"); + let mapping = iam + .get_mapped_policy_record(user, rustfs_iam::store::UserType::Reg, false) + .await + .expect("mapping"); + assert_eq!(mapping.policies, "sr-delayed-order-narrow"); + assert_eq!(mapping.update_at, t2); + + // group: add the member, then remove it. + let member = "sr-delayed-order-member"; + iam.create_user(member, &user_req("member-secret-key-123", rustfs_madmin::AccountStatus::Enabled)) + .await + .expect("member"); + let group = "sr-delayed-order-group"; + apply_iam_item(sr_group_item(group, &[member], false, t1)) + .await + .expect("T1 add"); + apply_iam_item(sr_group_item(group, &[member], true, t2)) + .await + .expect("T2 remove"); + let info = iam.get_group_info(group).await.expect("group"); + assert!(info.members.is_empty(), "the T2 removal must land after the delayed T1 add"); + assert_eq!(info.update_at, Some(t2)); + + // iam-user: create enabled, then the status-only disable. + let access_key = "sr-delayed-order-account"; + apply_iam_item(sr_user_item( + access_key, + Some(user_req("account-secret-key-123", rustfs_madmin::AccountStatus::Enabled)), + t1, + )) + .await + .expect("T1 create"); + apply_iam_item(sr_user_item(access_key, Some(user_req("", rustfs_madmin::AccountStatus::Disabled)), t2)) + .await + .expect("T2 disable"); + let identity = iam.get_user(access_key).await.expect("user"); + assert_eq!(identity.credentials.status, "off", "the T2 disable must land after the delayed T1 create"); + assert_eq!(identity.update_at, Some(t2)); + + clear_seeded_state().await; + } + + /// Review finding on rustfs#7195: an older grant and a newer revoke for + /// the same record delivered concurrently must always leave the revoke, + /// whichever request reaches the transaction first — the verdict and + /// the write of one cannot interleave with the other's. + #[tokio::test] + #[serial] + async fn apply_iam_item_serializes_a_concurrent_older_grant_and_newer_revoke() { + publish_ready_iam_context().await; + seed_two_peer_state_for_iam_apply().await; + let t1 = OffsetDateTime::now_utc() - time::Duration::hours(2); + let t2 = t1 + time::Duration::minutes(5); + + for round in 0..6u32 { + let policy = format!("sr-race-policy-{round}"); + let grant = tokio::spawn(apply_iam_item(sr_policy_item( + &policy, + allow_actions_policy(&["s3:GetObject", "s3:PutObject"]), + t1, + ))); + let revoke = tokio::spawn(apply_iam_item(sr_policy_item(&policy, allow_actions_policy(&["s3:GetObject"]), t2))); + let (grant, revoke) = if round % 2 == 0 { + tokio::join!(grant, revoke) + } else { + let (revoke, grant) = tokio::join!(revoke, grant); + (grant, revoke) + }; + grant.expect("join grant").expect("grant delivery is acknowledged"); + revoke.expect("join revoke").expect("revoke delivery is acknowledged"); + let (stored, stamp) = stored_policy_json(&policy).await; + assert!( + !stored.contains("s3:PutObject"), + "round {round}: the grant won over the newer revoke: {stored}" + ); + assert_eq!(stamp, Some(t2), "round {round}"); + } + + clear_seeded_state().await; + } + + /// Review finding on rustfs#7195: a replicated delete followed by the + /// delayed delivery of the older create must not resurrect the entity, + /// and the mark that fences it is committed by the same transaction that + /// applied the delete; a genuinely newer create still lands. + #[tokio::test] + #[serial] + async fn apply_iam_item_rejects_a_stale_recreate_after_a_replicated_delete() { + publish_ready_iam_context().await; + seed_two_peer_state_for_iam_apply().await; + let iam = current_iam_handle().expect("test IAM"); + let t1 = OffsetDateTime::now_utc() - time::Duration::hours(2); + let t3 = t1 + time::Duration::minutes(10); + let t4 = t3 + time::Duration::minutes(10); + let access_key = "sr-recreate-account"; + let create = |at| { + sr_user_item( + access_key, + Some(user_req("recreate-secret-key-123", rustfs_madmin::AccountStatus::Enabled)), + at, + ) + }; + + apply_iam_item(create(t1)).await.expect("T1 create"); + assert!(iam.get_user(access_key).await.is_some()); + apply_iam_item(sr_user_item(access_key, None, t3)).await.expect("T3 delete"); + assert!(iam.get_user(access_key).await.is_none()); + let state = load_site_replication_state().await.expect("state"); + assert_eq!( + state.iam_deletion_marks.get(&iam_user_deletion_mark_entity(access_key)), + Some(&t3), + "the delete's mark is committed with the delete" + ); + + apply_iam_item(create(t1)).await.expect("the stale replay is acknowledged"); + assert!( + iam.get_user(access_key).await.is_none(), + "a create older than the recorded deletion must not re-create the user" + ); + + apply_iam_item(create(t4)).await.expect("T4 create"); + let identity = iam.get_user(access_key).await.expect("a newer create lands"); + assert_eq!(identity.update_at, Some(t4)); + + clear_seeded_state().await; + } + + /// Review finding on rustfs#7195 (backlog#2289): a replicated disabled + /// service account is created disabled in one write, never enabled and + /// then switched off, and carries the source stamp. + #[tokio::test] + #[serial] + async fn apply_iam_item_creates_a_replicated_service_account_with_its_status() { + publish_ready_iam_context().await; + seed_two_peer_state_for_iam_apply().await; + let iam = current_iam_handle().expect("test IAM"); + let t1 = OffsetDateTime::now_utc() - time::Duration::hours(2); + let parent = "sr-svc-parent"; + iam.create_user(parent, &user_req("parent-secret-key-123", rustfs_madmin::AccountStatus::Enabled)) + .await + .expect("parent"); + + for (access_key, status, expected) in [ + ("sr-svc-disabled", "off", "off"), + ("sr-svc-enabled", "on", "on"), + ("sr-svc-default", "", "on"), + ] { + apply_iam_item(sr_service_account_create_item(parent, access_key, status, t1)) + .await + .expect("service account create"); + let (credentials, _) = iam.get_service_account(access_key).await.expect("service account"); + assert_eq!(credentials.status, expected, "{access_key}"); + let identity = iam.get_user(access_key).await.expect("identity"); + assert_eq!(identity.update_at, Some(t1), "{access_key} carries the source stamp"); + } + + clear_seeded_state().await; + } + #[tokio::test] #[serial] async fn apply_iam_item_accepts_minio_sts_account_item_type() { @@ -12501,7 +12858,8 @@ mod tests { && incoming_is_delete && let Some(deleted_at) = incoming_updated_at { - record_iam_deletion_marks(marks, &entities, deleted_at); + // Pruning is judged from the deletion's own clock in the model. + record_iam_deletion_marks_at(marks, &entities, deleted_at, deleted_at); } verdict } @@ -12593,8 +12951,8 @@ mod tests { let mut marks = SiteReplicationState::default(); let bob = iam_group_member_deletion_mark_entity("devs", "bob"); let group = iam_group_deletion_mark_entity("devs"); - record_iam_deletion_marks(&mut marks, std::slice::from_ref(&bob), at(20)); - record_iam_deletion_marks(&mut marks, std::slice::from_ref(&group), at(30)); + record_iam_deletion_marks_at(&mut marks, std::slice::from_ref(&bob), at(20), at(20)); + record_iam_deletion_marks_at(&mut marks, std::slice::from_ref(&group), at(30), at(30)); // The gate for an add of `bob` to the (deleted) group. let add_bob = [group.clone(), bob.clone()]; @@ -12625,43 +12983,98 @@ mod tests { } /// Cheap wiring guard for backlog#2291: every one of the `policy`, - /// `policy-mapping`, `group-info` and `iam-user` apply paths must route - /// through the shared staleness verdict before it writes or deletes - /// anything, and must fall back to the deletion mark when the record is - /// absent; `apply_iam_item` must record the mark of a committed deletion. - /// The ordering rule itself is covered by the `test_iam_item_*` behaviour - /// tests above; this only pins that no path bypasses it again. + /// `policy-mapping`, `group-info`, `iam-user` and `service-account` apply + /// paths must judge the item against the local record (falling back to + /// the deletion marks of the transaction's state when the record is + /// absent) before it writes or deletes anything, must stamp every write + /// with the item's source time, and `apply_iam_item` must run verdict, + /// write and mark commit inside one state transaction. The ordering rule + /// itself is covered by the `test_iam_item_*` model tests above and the + /// `apply_iam_item_*` receiver tests; this only pins that no path bypasses + /// it again. #[test] fn test_iam_policy_mapping_and_group_items_gate_on_incoming_updated_at() { let source = include_str!("site_replication.rs"); - for (start, end) in [ - ("async fn apply_iam_policy_item(", "async fn apply_iam_policy_mapping_item("), - ("async fn apply_iam_policy_mapping_item(", "async fn apply_iam_group_info_item("), - ("async fn apply_iam_group_info_item(", "async fn apply_iam_sts_account_item("), - ("async fn apply_iam_user_item(", "async fn apply_iam_service_account_item("), + let locally_stamped_writes = [ + ".set_policy(", + ".policy_db_set(", + ".add_users_to_group(", + ".remove_users_from_group(", + ".set_group_status(", + ".create_user(", + ".set_user_status(", + ".new_service_account(", + ".update_service_account(", + ]; + for (start, end, judged_by_shared_verdict) in [ + ("async fn apply_iam_policy_item(", "async fn apply_iam_policy_mapping_item(", true), + ("async fn apply_iam_policy_mapping_item(", "async fn apply_iam_group_info_item(", true), + ("async fn apply_iam_group_info_item(", "async fn apply_iam_sts_account_item(", true), + ("async fn apply_iam_user_item(", "async fn apply_iam_service_account_item(", true), + ("async fn apply_iam_service_account_item(", "fn claims_unix_timestamp(", false), ] { let body = source .split(start) .nth(1) .and_then(|rest| rest.split(end).next()) .expect(start); + if judged_by_shared_verdict { + assert!( + body.contains("judge_iam_item_staleness(local_updated_at, incoming_updated_at)"), + "{start} must judge the item against the local record before applying it" + ); + } else { + assert!( + body.contains("is_stale_update(local_updated_at, incoming_updated_at)"), + "{start} must judge the item against the local record before applying it" + ); + } assert!( - body.contains("judge_iam_item_staleness(local_updated_at, incoming_updated_at)"), - "{start} must judge the item against the local record before applying it" + body.contains("iam_deletion_mark("), + "{start} must fall back to the deletion marks of the transaction's state when the record is absent" ); assert!( - body.contains("local_iam_deletion_mark("), - "{start} must fall back to the deletion mark when the record is absent" + body.contains("replicated_write_stamp(incoming_updated_at)"), + "{start} must stamp its writes with the item's source time" ); + for write in locally_stamped_writes { + assert!( + !body.contains(write), + "{start} must not stamp a replicated write with the local clock ({write})" + ); + } } let dispatch = source .split("async fn apply_iam_item(") .nth(1) - .and_then(|rest| rest.split("async fn local_iam_deletion_mark(").next()) + .and_then(|rest| rest.split("fn replicated_write_stamp(").next()) .expect("apply_iam_item"); assert!( - dispatch.contains("commit_iam_deletion_marks(deletion_mark_entities, deleted_at)"), - "apply_iam_item must record the mark of a deletion it committed" + dispatch.contains("with_site_replication_state_transaction(move |mut state| async move {"), + "apply_iam_item must run verdict, write and mark commit in one state transaction" + ); + assert!( + dispatch.contains("record_iam_deletion_marks(&mut state, &deletion_mark_entities, deleted_at)"), + "apply_iam_item must record the mark of a deletion it committed in the same transaction" + ); + assert!( + !dispatch.contains("commit_iam_deletion_marks("), + "the mark must not be committed in a second transaction" + ); + // backlog#2289: a replicated service account is created with its + // status; a second status write could fail and leave it enabled. + let create_branch = source + .split("Err(err) if is_err_no_such_service_account(&err) => {") + .nth(1) + .and_then(|rest| rest.split("Err(err) => return Err(ApiError::from(err).into()),").next()) + .expect("service account create branch"); + assert!( + create_branch.contains("status: (!create.status.is_empty()).then_some(create.status),"), + "the service account must be created with the source status" + ); + assert!( + !create_branch.contains("update_service_account"), + "the created service account's status must not depend on a second write" ); } @@ -14259,6 +14672,10 @@ mod tests { "durable quota must carry the source stamp" ); assert!(apply.contains("update_if_incarnation_at("), "bucket configs must carry the source stamp"); + assert!( + apply.contains("delete_if_incarnation_at("), + "bucket config deletes must carry the source stamp too, or a newer re-create is judged stale" + ); assert!( !apply.contains("metadata_sys::update_if_incarnation(&item.bucket"), "no replicated config write may bypass the source stamp" diff --git a/rustfs/src/admin/handlers/user.rs b/rustfs/src/admin/handlers/user.rs index a9d4ee4f0..04d28290f 100644 --- a/rustfs/src/admin/handlers/user.rs +++ b/rustfs/src/admin/handlers/user.rs @@ -1131,6 +1131,7 @@ impl Operation for ImportIam { expiration: req.expiration, allow_site_replicator_account: false, claims: Some(req.claims), + status: None, }; let groups = if req.groups.is_empty() { None } else { Some(req.groups) }; diff --git a/rustfs/src/admin/storage_api.rs b/rustfs/src/admin/storage_api.rs index 809c46589..4377e7ea0 100644 --- a/rustfs/src/admin/storage_api.rs +++ b/rustfs/src/admin/storage_api.rs @@ -455,6 +455,18 @@ pub(crate) mod metadata_sys { super::ecstore_bucket::metadata_sys::delete_if_incarnation(bucket, config_file, expected_incarnation_id).await } + /// [`delete_if_incarnation`] stamping the cleared config with a replicated + /// deletion's source `updated_at` instead of the local clock (backlog#2292). + pub(crate) async fn delete_if_incarnation_at( + bucket: &str, + config_file: &str, + expected_incarnation_id: uuid::Uuid, + updated_at: OffsetDateTime, + ) -> Result { + super::ecstore_bucket::metadata_sys::delete_if_incarnation_at(bucket, config_file, expected_incarnation_id, updated_at) + .await + } + pub(crate) async fn get_bucket_policy(bucket: &str) -> Result<(BucketPolicy, OffsetDateTime)> { super::ecstore_bucket::metadata_sys::get_bucket_policy(bucket).await } diff --git a/rustfs/src/site_replication/state.rs b/rustfs/src/site_replication/state.rs index 0d9bf5f6e..efc8b16e5 100644 --- a/rustfs/src/site_replication/state.rs +++ b/rustfs/src/site_replication/state.rs @@ -68,23 +68,49 @@ pub(crate) struct SiteReplicationState { /// keyed by the deleted entity (`iam_item_deletion_mark_entities`). A /// deletion leaves no local record to judge a later item against, so this /// is what lets the receive-side staleness gate reject a grant that is - /// older than the revoke it would otherwise undo (backlog#2291). Bounded - /// by [`SITE_REPLICATION_IAM_DELETION_MARK_LIMIT`]; the oldest mark is - /// evicted first. + /// older than the revoke it would otherwise undo (backlog#2291). Marks + /// are kept for [`SITE_REPLICATION_IAM_DELETION_MARK_RETENTION`] and never + /// evicted by count: see that constant for why a count bound would open + /// exactly the window the marks exist to close. #[serde(default, with = "rfc3339_map", skip_serializing_if = "BTreeMap::is_empty")] pub(crate) iam_deletion_marks: BTreeMap, } -/// Upper bound on [`SiteReplicationState::iam_deletion_marks`]. -pub(crate) const SITE_REPLICATION_IAM_DELETION_MARK_LIMIT: usize = 1024; +/// How long an IAM deletion mark outlives the deletion it records. +/// +/// A mark fences the delivery paths that can still carry an older grant for +/// the deleted entity: a live delivery delayed in transit, the same grant +/// arriving on a sibling node while the revoke is being applied, and a +/// snapshot (bootstrap / repair / resend) built by a peer that has not yet +/// received the deletion — which is bounded by this site's own retry queue +/// towards that peer, whose backoff tops out at one day +/// (`SITE_REPLICATION_RETRY_DRAIN_MAX_BACKOFF_SECS`). The retry drain itself +/// never replays a stale grant: it resends snapshots of the current records +/// and the recorded deletion bodies. Thirty days is an order of magnitude +/// beyond every one of those windows. Marks are pruned by age only — a count +/// bound would drop a mark that is still inside the delivery window as soon +/// as enough newer deletions happen, letting the delayed grant re-create the +/// entity, which is the very hole the marks close. +pub(crate) const SITE_REPLICATION_IAM_DELETION_MARK_RETENTION: time::Duration = time::Duration::days(30); /// Record that deletions of `entities` with source timestamp `deleted_at` /// were committed here. Newest wins per entity: an older deletion never -/// lowers a mark. Returns whether the state changed. +/// lowers a mark. Marks older than the retention are pruned in the same +/// pass. Returns whether the state changed. pub(crate) fn record_iam_deletion_marks( state: &mut SiteReplicationState, entities: &[String], deleted_at: OffsetDateTime, +) -> bool { + record_iam_deletion_marks_at(state, entities, deleted_at, OffsetDateTime::now_utc()) +} + +/// [`record_iam_deletion_marks`] pruning against an explicit `now`. +pub(crate) fn record_iam_deletion_marks_at( + state: &mut SiteReplicationState, + entities: &[String], + deleted_at: OffsetDateTime, + now: OffsetDateTime, ) -> bool { let mut changed = false; for entity in entities { @@ -98,18 +124,10 @@ pub(crate) fn record_iam_deletion_marks( state.iam_deletion_marks.insert(entity.clone(), deleted_at); changed = true; } - while state.iam_deletion_marks.len() > SITE_REPLICATION_IAM_DELETION_MARK_LIMIT { - let Some(oldest) = state - .iam_deletion_marks - .iter() - .min_by_key(|(_, deleted_at)| **deleted_at) - .map(|(entity, _)| entity.clone()) - else { - break; - }; - state.iam_deletion_marks.remove(&oldest); - } - changed + let expired_before = now - SITE_REPLICATION_IAM_DELETION_MARK_RETENTION; + let before = state.iam_deletion_marks.len(); + state.iam_deletion_marks.retain(|_, deleted_at| *deleted_at >= expired_before); + changed || state.iam_deletion_marks.len() != before } /// Newest deletion mark among `entities`, or `None` when no deletion of any @@ -403,6 +421,33 @@ where update_site_replication_state_when_changed(move |state| update(state).map(StateCommit::Changed)).await } +/// The state transaction for work that has to await inside it: an IAM write +/// that must be ordered with the staleness verdict taken before it and the +/// deletion mark committed after it (backlog#2291). Same boundary as +/// [`update_site_replication_state`] — load and persist under the +/// distributed state-object write lock, so two nodes of this site cannot +/// interleave their verdicts and writes — and the same rules inside: no peer +/// network calls and no other config locks. The closure hands the state back +/// as `Some` when it changed it; `None` skips the write. +pub(crate) async fn with_site_replication_state_transaction(transaction: F) -> S3Result +where + T: Send + 'static, + F: FnOnce(SiteReplicationState) -> Fut + Send + 'static, + Fut: std::future::Future)>> + Send + 'static, +{ + with_site_replication_state_lock(move || async move { + let store = current_object_store_handle() + .ok_or_else(|| S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()))?; + let state = load_site_replication_state_no_lock(store.clone()).await?; + let (result, changed) = transaction(state).await?; + if let Some(state) = changed { + persist_site_replication_state_no_lock(store, state).await?; + } + Ok(result) + }) + .await +} + /// [`update_site_replication_state`] for closures that may find nothing to /// do — see [`StateCommit`]. pub(crate) async fn update_site_replication_state_when_changed(update: F) -> S3Result diff --git a/rustfs/src/site_replication/tests.rs b/rustfs/src/site_replication/tests.rs index a91d95d35..f1c06131f 100644 --- a/rustfs/src/site_replication/tests.rs +++ b/rustfs/src/site_replication/tests.rs @@ -622,40 +622,64 @@ fn test_iam_item_deletion_mark_entities_shapes() { assert!(iam_item_deletion_mark_entities(&user_create).is_empty()); } -/// Newest wins per entity, the map stays bounded by evicting the oldest -/// mark, and the timestamps survive the state object as RFC 3339. +/// Newest wins per entity, marks are pruned by age only (never by count: a +/// count bound would drop a mark still inside the delivery window as soon as +/// enough newer deletions happen), and the timestamps survive the state +/// object as RFC 3339. #[test] -fn test_record_iam_deletion_marks_newest_wins_and_stays_bounded() { +fn test_record_iam_deletion_marks_newest_wins_and_expires_by_age_only() { let at = |seconds: i64| OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(seconds); + let now = at(1_000_000); let mut state = SiteReplicationState::default(); let alice = vec!["iam-user:alice".to_string()]; - assert!(record_iam_deletion_marks(&mut state, &alice, at(20))); + assert!(record_iam_deletion_marks_at(&mut state, &alice, at(20), now)); assert!( - !record_iam_deletion_marks(&mut state, &alice, at(10)), + !record_iam_deletion_marks_at(&mut state, &alice, at(10), now), "an older deletion does not move the mark" ); assert!( - !record_iam_deletion_marks(&mut state, &alice, at(20)), + !record_iam_deletion_marks_at(&mut state, &alice, at(20), now), "a replayed deletion is not a change" ); assert_eq!(iam_deletion_mark(&state, &alice), Some(at(20))); - assert!(record_iam_deletion_marks(&mut state, &alice, at(30))); + assert!(record_iam_deletion_marks_at(&mut state, &alice, at(30), now)); assert_eq!(iam_deletion_mark(&state, &alice), Some(at(30))); assert_eq!(iam_deletion_mark(&state, &["iam-user:bob".to_string()]), None); - assert!(!record_iam_deletion_marks(&mut state, &[], at(40))); + assert!(!record_iam_deletion_marks_at(&mut state, &[], at(40), now)); - // Fill past the bound with marks older than alice's; the oldest go first. - let members: Vec = (0..SITE_REPLICATION_IAM_DELETION_MARK_LIMIT) - .map(|index| format!("group-member:devs:user-{index:04}")) - .collect(); + // Many newer deletions never evict an older mark that is still within the retention. + let members: Vec = (0..4096).map(|index| format!("group-member:devs:user-{index:04}")).collect(); for (index, member) in members.iter().enumerate() { - record_iam_deletion_marks(&mut state, std::slice::from_ref(member), at(index as i64 - 2000)); + record_iam_deletion_marks_at(&mut state, std::slice::from_ref(member), at(100 + index as i64), now); } - assert_eq!(state.iam_deletion_marks.len(), SITE_REPLICATION_IAM_DELETION_MARK_LIMIT); - assert_eq!(iam_deletion_mark(&state, &alice), Some(at(30)), "the newest mark survives eviction"); - assert_eq!(iam_deletion_mark(&state, &members[..1]), None, "the oldest mark is evicted first"); - assert_eq!(iam_deletion_mark(&state, &members[1..2]), Some(at(-1999))); + assert_eq!(state.iam_deletion_marks.len(), members.len() + 1); + assert_eq!(iam_deletion_mark(&state, &alice), Some(at(30)), "no count-based eviction"); + + // Marks older than the retention are pruned, on the pass that records a + // newer one and on a pass that changes nothing else; younger ones stay. + let later = at(100) + SITE_REPLICATION_IAM_DELETION_MARK_RETENTION; + assert!( + record_iam_deletion_marks_at(&mut state, &["iam-user:carol".to_string()], at(200_000), later), + "pruning alone is a change" + ); + assert_eq!(iam_deletion_mark(&state, &alice), None, "alice's mark aged out"); + assert_eq!( + iam_deletion_mark(&state, &members[..1]), + Some(at(100)), + "a mark exactly at the retention edge stays, and so do the younger ones" + ); + assert_eq!(state.iam_deletion_marks.len(), members.len() + 1); + assert_eq!(iam_deletion_mark(&state, &["iam-user:carol".to_string()]), Some(at(200_000))); + let mut state = SiteReplicationState::default(); + record_iam_deletion_marks_at(&mut state, &alice, at(30), now); + let past_edge = at(30) + SITE_REPLICATION_IAM_DELETION_MARK_RETENTION + time::Duration::seconds(1); + assert!( + record_iam_deletion_marks_at(&mut state, &[], at(0), past_edge), + "a pass that only prunes reports the change" + ); + assert_eq!(iam_deletion_mark(&state, &alice), None); + record_iam_deletion_marks_at(&mut state, &alice, at(30), now); let json = serde_json::to_value(&state).expect("serialize state"); assert_eq!(json["iam_deletion_marks"]["iam-user:alice"], serde_json::json!("1970-01-01T00:00:30Z"));