From cc68d18783b78d530c6720564c8beed0b37192c1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=94=90=E5=B0=8F=E9=B8=AD?= Date: Sun, 6 Sep 2026 11:08:19 +0800 Subject: [PATCH] fix(iam): publish stamped group writes with the local clock (rustfs#7195) Review finding: `add_users_to_group_at`, `set_group_status_at` and `remove_members_from_group_at` handed the replicated source stamp to the cache as its publication time. `LockedCache::exec` drops a publication whose time is behind the entity's load time, so a group edit whose source time predated this node's startup was persisted but never reached the cache, and the receiver's following status write failed with `NoSuchGroup`. The stamp now stays on `GroupInfo::update_at` only; the group entity and the membership index are published with the local clock. The regression drives `IamSys` through a startup-shaped cache load and then add, status and removal with hour-old source stamps, asserting each is readable at once. --- crates/iam/src/manager.rs | 30 ++++++++++------ crates/iam/src/sys.rs | 75 +++++++++++++++++++++++++++++++++++++-- 2 files changed, 92 insertions(+), 13 deletions(-) diff --git a/crates/iam/src/manager.rs b/crates/iam/src/manager.rs index 8b295059e..f7ad9e6a7 100644 --- a/crates/iam/src/manager.rs +++ b/crates/iam/src/manager.rs @@ -1790,7 +1790,11 @@ 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 = updated_at; + // `updated_at` is the record's stamp only; the cache is published + // with the local clock, because `LockedCache::exec` drops a write + // whose time predates the entity's load time — a replicated edit + // whose source time is older than this node's startup would + // otherwise never reach the cache. let gi = match cache.groups.get(group) { Some(res) => { let mut gi = res.clone(); @@ -1799,15 +1803,16 @@ where uniq_set.extend(members.iter().cloned()); gi.members = uniq_set.into_iter().collect(); - gi.update_at = Some(now); + gi.update_at = Some(updated_at); gi } None => { let mut gi = GroupInfo::new(members.clone()); - gi.update_at = Some(now); + gi.update_at = Some(updated_at); gi } }; + let now = OffsetDateTime::now_utc(); drop(cache); self.api.save_group_info(group, gi.clone()).await?; @@ -1823,7 +1828,7 @@ where }); }); - Ok(now) + Ok(updated_at) } pub async fn set_group_status(&self, name: &str, enable: bool) -> Result { @@ -1850,14 +1855,15 @@ where } else { gi.status = STATUS_DISABLED.to_owned(); } - let now = updated_at; - gi.update_at = Some(now); + gi.update_at = Some(updated_at); self.api.save_group_info(name, gi.clone()).await?; - self.cache.add_or_update_group(name, &gi, now); + // Cache publication time is the local clock, not the record stamp + // (see `add_users_to_group_at`). + self.cache.add_or_update_group(name, &gi, OffsetDateTime::now_utc()); - Ok(now) + Ok(updated_at) } pub async fn get_group_description(&self, name: &str) -> Result { @@ -1953,8 +1959,10 @@ 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 = updated_at; - gi.update_at = Some(now); + gi.update_at = Some(updated_at); + // Cache publication time is the local clock, not the record stamp + // (see `add_users_to_group_at`). + let now = OffsetDateTime::now_utc(); if !update_cache_only { self.api.save_group_info(name, gi.clone()).await?; @@ -1973,7 +1981,7 @@ where }); }); - Ok(now) + Ok(updated_at) } pub async fn remove_users_from_group(&self, group: &str, members: Vec) -> Result { diff --git a/crates/iam/src/sys.rs b/crates/iam/src/sys.rs index 037eb3d4f..16bc8de0a 100644 --- a/crates/iam/src/sys.rs +++ b/crates/iam/src/sys.rs @@ -2326,11 +2326,11 @@ mod tests { } async fn save_group_info(&self, _name: &str, _item: GroupInfo) -> Result<()> { - Err(Error::InvalidArgument) + Ok(()) } async fn delete_group_info(&self, _name: &str) -> Result<()> { - Err(Error::InvalidArgument) + Ok(()) } async fn load_group(&self, name: &str, m: &mut HashMap) -> Result<()> { @@ -2507,6 +2507,77 @@ mod tests { IamSys::new(cache) } + /// Review finding on rustfs#7195: a replicated group edit carries a source + /// stamp that may predate this node's cache load time. The stamp belongs on + /// the record only; publishing the cache with it makes `LockedCache::exec` + /// drop the write, so the group is written to the store but unreadable + /// here and the receiver's next `set_group_status_at` fails with + /// `NoSuchGroup`. Add, status and removal must all publish with the local + /// clock while keeping the source stamp on `GroupInfo::update_at`. + #[tokio::test] + async fn group_writes_stamped_before_the_cache_load_time_still_publish() { + let iam_sys = test_iam_sys().await; + let member = "group-stamp-member"; + let identity = UserIdentity { + version: 1, + credentials: Credentials { + access_key: member.to_string(), + secret_key: "longenoughsecret".to_string(), + status: "on".to_string(), + ..Default::default() + }, + update_at: Some(OffsetDateTime::now_utc()), + }; + iam_sys.store.cache.with_write_lock(|cache| { + cache.add_or_update_user(member, &identity, OffsetDateTime::now_utc()); + // The startup load publishes every entity with the load time. + cache.replace_groups(CacheEntity::new(HashMap::new())); + cache.replace_user_group_memberships(CacheEntity::new(HashMap::new())); + }); + + let group = "group-stamp"; + let source_time = OffsetDateTime::now_utc() - time::Duration::hours(1); + let stamped = iam_sys + .add_users_to_group_at(group, vec![member.to_string()], source_time) + .await + .expect("add members with a source stamp older than the cache load"); + assert_eq!(stamped, source_time, "the returned stamp is the source time"); + let info = iam_sys + .get_group_info(group) + .await + .expect("the group must be readable right after the add"); + assert_eq!(info.members, vec![member.to_string()]); + assert_eq!(info.update_at, Some(source_time), "the record keeps the source stamp"); + let memberships = iam_sys.store.cache.snapshot().user_group_memberships.get(member).cloned(); + assert!( + memberships.is_some_and(|groups| groups.contains(group)), + "the membership index is published too" + ); + + let disabled_at = source_time + time::Duration::seconds(1); + iam_sys + .set_group_status_at(group, false, disabled_at) + .await + .expect("status change with a source stamp older than the cache load"); + let info = iam_sys.get_group_info(group).await.expect("group after status change"); + assert_eq!(info.status, "disabled"); + assert_eq!(info.update_at, Some(disabled_at)); + + let removed_at = source_time + time::Duration::seconds(2); + iam_sys + .remove_users_from_group_at(group, vec![member.to_string()], removed_at) + .await + .expect("removal with a source stamp older than the cache load"); + let info = iam_sys.get_group_info(group).await.expect("group after removal"); + assert!(info.members.is_empty(), "the removal must be visible in the cache"); + assert_eq!(info.update_at, Some(removed_at)); + let memberships = iam_sys.store.cache.snapshot().user_group_memberships.get(member).cloned(); + assert!( + !memberships.is_some_and(|groups| groups.contains(group)), + "the membership index follows the removal" + ); + } + fn service_account_opts(access_key: &str, secret_key: &str) -> NewServiceAccountOpts { NewServiceAccountOpts { access_key: access_key.to_string(),