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

@@ -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<OffsetDateTime> {
@@ -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<GroupDesc> {
@@ -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::<Vec<String>>();
-        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<String>) -> Result<OffsetDateTime> {

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<String, GroupInfo>) -> 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(),