fix(replication): close IAM snapshot, marker purge and broadcast gaps (#7195)

crates/ecstore/src/api/mod.rs

@@ -159,15 +159,17 @@ pub mod bucket {
             BUCKET_CONFIG_PUBLISH_HOOK, BucketConfigPublishHook, 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,
+            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_on_demand_migration_config_in, 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_quota_if_incarnation, update_under_transaction_lock,
+            update_if_incarnation_at, update_quota_if_incarnation, update_quota_if_incarnation_at, update_under_transaction_lock,
+            update_under_transaction_lock_at,
         };
         #[cfg(feature = "test-util")]
         pub use crate::bucket::metadata_sys::{ConfigWriteLockProbe, test_support};

crates/ecstore/src/bucket/metadata.rs

@@ -791,9 +791,22 @@ impl BucketMetadata {
         }
     }
 
+    /// Replace one config payload and stamp its `*_config_updated_at` with the
+    /// local clock. This is the entry for edits that originate here: the
+    /// local write time is the edit's source time.
     pub fn update_config(&mut self, config_file: &str, data: Vec<u8>) -> Result<OffsetDateTime> {
-        let updated = OffsetDateTime::now_utc();
+        self.update_config_at(config_file, data, OffsetDateTime::now_utc())
+    }
 
+    /// [`Self::update_config`] with an explicit `updated_at` stamp.
+    ///
+    /// For a config replicated from another site the edit's source time is
+    /// the peer's `updated_at`, not the moment it lands here: staleness of
+    /// the next incoming item is judged against the stored stamp, so stamping
+    /// the local apply time would reject a newer source edit that was merely
+    /// delivered late (backlog#2292). Only replication receivers should pass
+    /// a foreign time; local edits keep [`Self::update_config`].
+    pub fn update_config_at(&mut self, config_file: &str, data: Vec<u8>, updated: OffsetDateTime) -> Result<OffsetDateTime> {
         match config_file {
             BUCKET_POLICY_CONFIG => {
                 self.policy_config_json = data;
@@ -1525,6 +1538,39 @@ mod test {
         assert_eq!(metadata.bucket_incarnation_id, incarnation);
     }
 
+    /// backlog#2292: a replicated config is stamped with the source
+    /// `updated_at` it was given, not the local clock, while the plain
+    /// `update_config` entry keeps stamping the local clock.
+    #[test]
+    fn update_config_at_stamps_the_given_time_and_update_config_stamps_now() {
+        let source_time = OffsetDateTime::now_utc() - time::Duration::hours(3);
+        let mut metadata = BucketMetadata::new("source-stamped");
+
+        let stamped = metadata
+            .update_config_at(BUCKET_POLICY_CONFIG, br#"{"Version":"2012-10-17","Statement":[]}"#.to_vec(), source_time)
+            .unwrap();
+        assert_eq!(stamped, source_time);
+        assert_eq!(metadata.policy_config_updated_at, source_time);
+
+        let tagging = b"<Tagging><TagSet><Tag><Key>k</Key><Value>v</Value></Tag></TagSet></Tagging>".to_vec();
+        let stamped = metadata
+            .update_config_at(BUCKET_TAGGING_CONFIG, tagging, source_time)
+            .unwrap();
+        assert_eq!(stamped, source_time);
+        assert_eq!(metadata.tagging_config_updated_at, source_time);
+
+        let before = OffsetDateTime::now_utc();
+        let stamped = metadata
+            .update_config(BUCKET_POLICY_CONFIG, br#"{"Version":"2012-10-17","Statement":[]}"#.to_vec())
+            .unwrap();
+        assert!(stamped >= before, "a local edit is stamped with the local clock");
+        assert_eq!(metadata.policy_config_updated_at, stamped);
+        assert_eq!(
+            metadata.tagging_config_updated_at, source_time,
+            "restamping one config must not move another config's stamp"
+        );
+    }
+
     #[test]
     fn object_locking_requires_lock_metadata_not_plain_versioning() {
         use s3s::dto::ObjectLockEnabled;

crates/ecstore/src/bucket/metadata_sys.rs

@@ -567,6 +567,32 @@ pub async fn update_if_incarnation(
         config_file,
         data,
         Some(expected_incarnation_id),
+        None,
+    ))
+    .await
+}
+
+/// [`update_if_incarnation`] stamping the config with `updated_at` instead of
+/// the local clock.
+///
+/// For a site-replication receiver the edit's source time is the peer's
+/// `updated_at`; persisting it keeps the stored `*_config_updated_at` on the
+/// source clock so the next item's staleness is judged source-time against
+/// source-time (backlog#2292). See [`BucketMetadata::update_config_at`].
+pub async fn update_if_incarnation_at(
+    bucket: &str,
+    config_file: &str,
+    data: Vec<u8>,
+    expected_incarnation_id: Uuid,
+    updated_at: OffsetDateTime,
+) -> Result<OffsetDateTime> {
+    Box::pin(update_with_sys_expected(
+        get_bucket_metadata_sys()?,
+        bucket,
+        config_file,
+        data,
+        Some(expected_incarnation_id),
+        Some(updated_at),
     ))
     .await
 }
@@ -577,6 +603,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<OffsetDateTime> {
+    Box::pin(delete_with_sys_expected(
+        get_bucket_metadata_sys()?,
+        bucket,
+        config_file,
+        Some(expected_incarnation_id),
+        Some(updated_at),
     ))
     .await
 }
@@ -598,34 +648,41 @@ async fn update_with_sys(
     config_file: &str,
     data: Vec<u8>,
 ) -> Result<OffsetDateTime> {
-    update_with_sys_expected(sys, bucket, config_file, data, None).await
+    update_with_sys_expected(sys, bucket, config_file, data, None, None).await
 }
 
+/// `updated_at` is the stamp persisted on the config; `None` uses the local
+/// clock (the edit originates here), `Some` carries a replicated edit's
+/// source time (backlog#2292).
 async fn update_with_sys_expected(
     sys: Arc<RwLock<BucketMetadataSys>>,
     bucket: &str,
     config_file: &str,
     data: Vec<u8>,
     expected_incarnation_id: Option<Uuid>,
+    updated_at: Option<OffsetDateTime>,
 ) -> Result<OffsetDateTime> {
     let guard = acquire_config_write_guard_for_incarnation(sys.clone(), bucket, expected_incarnation_id).await?;
-    update_under_config_write_guard(sys, &guard, config_file, data).await
+    update_under_config_write_guard(sys, &guard, config_file, data, updated_at).await
 }
 
 /// [`delete`] against an explicitly supplied metadata system. See
 /// [`update_with_sys`].
 async fn delete_with_sys(sys: Arc<RwLock<BucketMetadataSys>>, bucket: &str, config_file: &str) -> Result<OffsetDateTime> {
-    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<RwLock<BucketMetadataSys>>,
     bucket: &str,
     config_file: &str,
     expected_incarnation_id: Option<Uuid>,
+    updated_at: Option<OffsetDateTime>,
 ) -> Result<OffsetDateTime> {
     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.
@@ -772,7 +829,21 @@ pub async fn update_under_transaction_lock(
     data: Vec<u8>,
 ) -> Result<OffsetDateTime> {
     guard.ensure_valid(bucket)?;
-    update_under_config_write_guard(get_bucket_metadata_sys()?, guard, config_file, data).await
+    update_under_config_write_guard(get_bucket_metadata_sys()?, guard, config_file, data, None).await
+}
+
+/// [`update_under_transaction_lock`] stamping the config with `updated_at`
+/// (a replicated edit's source time) instead of the local clock; see
+/// [`update_if_incarnation_at`] (backlog#2292).
+pub async fn update_under_transaction_lock_at(
+    guard: &BucketMetadataMutationGuard,
+    bucket: &str,
+    config_file: &str,
+    data: Vec<u8>,
+    updated_at: OffsetDateTime,
+) -> Result<OffsetDateTime> {
+    guard.ensure_valid(bucket)?;
+    update_under_config_write_guard(get_bucket_metadata_sys()?, guard, config_file, data, Some(updated_at)).await
 }
 
 /// Clear one config file while the caller holds this bucket's transaction lock.
@@ -782,7 +853,7 @@ pub async fn delete_under_transaction_lock(
     config_file: &str,
 ) -> Result<OffsetDateTime> {
     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(
@@ -790,6 +861,29 @@ pub async fn update_quota_if_incarnation(
     data: Vec<u8>,
     expected_incarnation_id: Uuid,
     proof: &crate::services::notification_sys::CrossPoolFenceFleetProofToken,
+) -> Result<OffsetDateTime> {
+    update_quota_if_incarnation_stamped(bucket, data, expected_incarnation_id, proof, None).await
+}
+
+/// [`update_quota_if_incarnation`] stamping the quota config with
+/// `updated_at` (a replicated edit's source time) instead of the local
+/// clock; see [`update_if_incarnation_at`] (backlog#2292).
+pub async fn update_quota_if_incarnation_at(
+    bucket: &str,
+    data: Vec<u8>,
+    expected_incarnation_id: Uuid,
+    proof: &crate::services::notification_sys::CrossPoolFenceFleetProofToken,
+    updated_at: OffsetDateTime,
+) -> Result<OffsetDateTime> {
+    update_quota_if_incarnation_stamped(bucket, data, expected_incarnation_id, proof, Some(updated_at)).await
+}
+
+async fn update_quota_if_incarnation_stamped(
+    bucket: &str,
+    data: Vec<u8>,
+    expected_incarnation_id: Uuid,
+    proof: &crate::services::notification_sys::CrossPoolFenceFleetProofToken,
+    updated_at: Option<OffsetDateTime>,
 ) -> Result<OffsetDateTime> {
     let sys = get_bucket_metadata_sys()?;
     let guard = Box::pin(acquire_config_write_guard_for_incarnation(
@@ -807,7 +901,7 @@ pub async fn update_quota_if_incarnation(
             achieved: 0,
         });
     }
-    update_under_config_write_guard(sys, &guard, rustfs_config::QUOTA_CONFIG_FILE, data).await
+    update_under_config_write_guard(sys, &guard, rustfs_config::QUOTA_CONFIG_FILE, data, updated_at).await
 }
 
 pub async fn update_bucket_targets_under_transaction_lock(
@@ -823,6 +917,7 @@ async fn update_under_config_write_guard(
     guard: &BucketMetadataMutationGuard,
     config_file: &str,
     data: Vec<u8>,
+    updated_at: Option<OffsetDateTime>,
 ) -> Result<OffsetDateTime> {
     guard.ensure_valid(&guard.bucket)?;
     let metadata_sys = sys.read().await.clone();
@@ -834,7 +929,7 @@ async fn update_under_config_write_guard(
             Some(&guard.transaction_guard),
             &guard.bucket,
             "bucket config transaction",
-            metadata_sys.update_checked(&guard.bucket, config_file, data, true, guard.incarnation_id),
+            metadata_sys.update_checked(&guard.bucket, config_file, data, true, guard.incarnation_id, updated_at),
         ),
     )
     .await?;
@@ -846,6 +941,7 @@ async fn delete_under_config_write_guard(
     sys: Arc<RwLock<BucketMetadataSys>>,
     guard: &BucketMetadataMutationGuard,
     config_file: &str,
+    updated_at: Option<OffsetDateTime>,
 ) -> Result<OffsetDateTime> {
     guard.ensure_valid(&guard.bucket)?;
     let metadata_sys = sys.read().await.clone();
@@ -857,7 +953,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),
+            metadata_sys.update_checked(&guard.bucket, config_file, Vec::new(), false, guard.incarnation_id, updated_at),
         ),
     )
     .await?;
@@ -1762,15 +1858,17 @@ impl BucketMetadataSys {
     /// `update` and the config read alone). Keep these boxed.
     pub async fn update(&self, bucket: &str, config_file: &str, data: Vec<u8>) -> Result<OffsetDateTime> {
         let incarnation_id = Box::pin(self.get_bucket_incarnation_id(bucket)).await?;
-        Box::pin(self.update_checked(bucket, config_file, data, true, incarnation_id)).await
+        Box::pin(self.update_checked(bucket, config_file, data, true, incarnation_id, None)).await
     }
 
     pub async fn delete(&self, bucket: &str, config_file: &str) -> Result<OffsetDateTime> {
         let incarnation_id = self.get_bucket_incarnation_id(bucket).await?;
-        self.update_checked(bucket, config_file, Vec::new(), false, incarnation_id)
+        self.update_checked(bucket, config_file, Vec::new(), false, incarnation_id, None)
             .await
     }
 
+    /// `updated_at`: `None` stamps the local clock; `Some` persists a
+    /// replicated edit's source time (backlog#2292).
     async fn update_checked(
         &self,
         bucket: &str,
@@ -1778,6 +1876,7 @@ impl BucketMetadataSys {
         data: Vec<u8>,
         parse: bool,
         expected_incarnation_id: Uuid,
+        updated_at: Option<OffsetDateTime>,
     ) -> Result<OffsetDateTime> {
         // Load through this system's own store, the one `save` persists to
         // (backlog#1052 S7). Reading from the ambient handle instead made the
@@ -1788,7 +1887,10 @@ impl BucketMetadataSys {
             return Err(Error::BucketNotFound(bucket.to_string()));
         }
 
-        let updated = bm.update_config(config_file, data)?;
+        let updated = match updated_at {
+            Some(updated_at) => bm.update_config_at(config_file, data, updated_at)?,
+            None => bm.update_config(config_file, data)?,
+        };
 
         Box::pin(self.save(bm)).await?;
 
@@ -3755,6 +3857,106 @@ mod tests {
         );
     }
 
+    /// backlog#2292: the explicit-stamp write path persists the given source
+    /// time as the config's `*_config_updated_at` — through the incarnation
+    /// path and through an already-held transaction guard — and survives a
+    /// reload from disk, while the plain path keeps stamping the local clock.
+    #[tokio::test]
+    async fn explicit_updated_at_is_persisted_as_the_config_stamp() {
+        let (dirs, ecstore) = isolated_store_over_temp_disks().await;
+        let bucket = "source-stamped-config";
+        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 source_time = OffsetDateTime::now_utc() - Duration::from_secs(3 * 3600);
+        let policy = br#"{"Version":"2012-10-17","Statement":[]}"#.to_vec();
+        let tagging = b"<Tagging><TagSet><Tag><Key>k</Key><Value>v</Value></Tag></TagSet></Tagging>".to_vec();
+
+        // Incarnation path (`update_if_incarnation_at` minus the ambient lookup).
+        let stamped =
+            update_with_sys_expected(sys.clone(), bucket, BUCKET_POLICY_CONFIG, policy.clone(), None, Some(source_time))
+                .await
+                .expect("source-stamped policy write should persist");
+        assert_eq!(stamped, source_time);
+
+        // Held-guard path (`update_under_transaction_lock_at` minus the ambient lookup).
+        let guard = acquire_config_write_guard(sys.clone(), bucket).await.expect("write guard");
+        let stamped = update_under_config_write_guard(sys.clone(), &guard, BUCKET_TAGGING_CONFIG, tagging, Some(source_time))
+            .await
+            .expect("source-stamped tagging write should persist");
+        drop(guard);
+        assert_eq!(stamped, source_time);
+
+        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_eq!(reloaded.policy_config_updated_at, source_time);
+        assert_eq!(reloaded.tagging_config_updated_at, source_time);
+
+        // The plain path is unchanged: a local edit is stamped with the local clock.
+        let before = OffsetDateTime::now_utc();
+        let stamped = update_with_sys(sys.clone(), bucket, BUCKET_POLICY_CONFIG, policy)
+            .await
+            .expect("locally stamped policy write should persist");
+        assert!(stamped >= before, "the plain write 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);
+        assert_eq!(
+            reloaded.tagging_config_updated_at, source_time,
+            "an unrelated config keeps its source stamp"
+        );
+    }
+
+    /// 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.
@@ -3971,10 +4173,16 @@ mod tests {
         let new_incarnation = store.bucket_incarnation_id_from_disk(bucket).await.unwrap();
         assert_ne!(old_incarnation, new_incarnation);
 
-        let err =
-            update_with_sys_expected(sys.clone(), bucket, BUCKET_TAGGING_CONFIG, b"<Tagging/>".to_vec(), Some(old_incarnation))
-                .await
-                .expect_err("a request authorized for the deleted incarnation must fail closed");
+        let err = update_with_sys_expected(
+            sys.clone(),
+            bucket,
+            BUCKET_TAGGING_CONFIG,
+            b"<Tagging/>".to_vec(),
+            Some(old_incarnation),
+            None,
+        )
+        .await
+        .expect_err("a request authorized for the deleted incarnation must fail closed");
         assert!(matches!(err, Error::BucketNotFound(name) if name == bucket));
 
         let persisted = sys.read().await.get_config_from_disk(bucket).await.unwrap();
@@ -4009,7 +4217,7 @@ mod tests {
             }],
         })
         .unwrap();
-        update_under_config_write_guard(sys, &guard, BUCKET_TAGGING_CONFIG, tagging)
+        update_under_config_write_guard(sys, &guard, BUCKET_TAGGING_CONFIG, tagging, None)
             .await
             .unwrap();
         assert!(!delete.is_finished());

crates/ecstore/src/bucket/replication/replication_object_decision_boundary.rs

@@ -20,9 +20,9 @@ pub use rustfs_replication::{
 pub(crate) use rustfs_replication::{
     ReplicationDeleteSource, ReplicationMultipartPartInput, ReplicationResyncTargetObject, delete_marker_purge_mrf_entry,
     delete_marker_purge_version_id, delete_replication_creates_marker, delete_replication_missing_source_decision,
-    delete_replication_object_opts, heal_uses_delete_replication_path, is_object_lock_denied_delete,
-    is_retryable_delete_replication_head_error, is_version_delete_replication, replicate_delete_outcome, replication_etags_match,
-    replication_multipart_complete_actual_size, replication_multipart_part_plan, replication_single_put_size_error,
-    resync_existing_delete_replication_info, resync_target_for_object, should_retry_delete_marker_purge,
-    single_part_replica_etag_mismatch, target_delete_version_id,
+    delete_replication_object_opts, delete_replication_target_version_id, heal_uses_delete_replication_path,
+    is_object_lock_denied_delete, is_retryable_delete_replication_head_error, is_version_delete_replication,
+    replicate_delete_outcome, replication_etags_match, replication_multipart_complete_actual_size,
+    replication_multipart_part_plan, replication_single_put_size_error, resync_existing_delete_replication_info,
+    resync_target_for_object, should_retry_delete_marker_purge, single_part_replica_etag_mismatch,
 };

crates/ecstore/src/bucket/replication/replication_pool.rs

@@ -882,6 +882,20 @@ fn reconstructed_heal_delete_info(
 ) -> DeletedObjectReplicationInfo {
     let mut rstate = oi.replication_state();
     rstate.replicate_decision_str = dsc.to_string();
+    // The caller hands us a blank ObjectInfo (the source marker may already be
+    // gone), so the state above carries no target-assigned marker version ids.
+    // Restore them from the journal: `delete_marker_purge_version_id` must hit
+    // the id the target reported, not fall back to the source marker id, which
+    // a target that mints its own ids answers with an idempotent 204 that would
+    // acknowledge the intent while the real marker stays behind (backlog#2290).
+    // The corrupt flag rides along so a refusal stays a refusal after restart.
+    for (arn, version_id) in &entry.target_delete_marker_version_ids {
+        rstate
+            .target_delete_marker_version_ids
+            .entry(arn.clone())
+            .or_insert_with(|| version_id.clone());
+    }
+    rstate.target_delete_marker_version_ids_corrupt |= entry.target_delete_marker_version_ids_corrupt;
 
     let delete_marker_mtime = entry
         .delete_marker_mtime
@@ -6601,4 +6615,87 @@ mod tests {
             replacement_data
         );
     }
+
+    /// backlog#2290: a delete-marker purge intent that survives a restart
+    /// through the MRF journal addresses the marker version the TARGET
+    /// assigned, exactly as the live watcher does (see the
+    /// `requires_delayed_purge` spawn). The journal carries the per-ARN ids
+    /// (`targetDeleteMarkerVersionIDs`) and replay restores them into the
+    /// reconstructed replication state; without that the replay would fall
+    /// back to the source marker id, which a target that mints its own ids
+    /// answers with an idempotent 204 — the entry would be acknowledged while
+    /// the real marker stayed behind.
+    #[test]
+    fn mrf_delete_marker_purge_replay_preserves_target_assigned_marker_version() {
+        use super::super::replication_object_decision_boundary::{delete_marker_purge_mrf_entry, delete_marker_purge_version_id};
+
+        let arn = "arn:minio:replication::generic-target:photos".to_string();
+        let source_marker = uuid::Uuid::new_v4();
+        let remote_marker = "remote-assigned-marker-version".to_string();
+
+        let live_oi = ObjectInfo {
+            bucket: "photos".to_string(),
+            name: "obj".to_string(),
+            version_id: Some(source_marker),
+            delete_marker: true,
+            ..Default::default()
+        };
+        let mut live_state = live_oi.replication_state();
+        live_state.replicate_decision_str = replicate_decision_for_admitted_targets(std::slice::from_ref(&arn)).to_string();
+        live_state
+            .target_delete_marker_version_ids
+            .insert(arn.clone(), remote_marker.clone());
+        let live = DeletedObjectReplicationInfo {
+            delete_object: ReplicationDeletedObject {
+                object_name: "obj".to_string(),
+                delete_marker: true,
+                delete_marker_version_id: Some(source_marker),
+                replication_state: Some(live_state),
+                ..Default::default()
+            },
+            bucket: "photos".to_string(),
+            ..Default::default()
+        };
+        assert_eq!(
+            delete_marker_purge_version_id(live.delete_object.replication_state.as_ref(), &arn, source_marker),
+            Some(Some(remote_marker.clone())),
+            "the live purge addresses the recorded target version"
+        );
+
+        // Watch window exhausted: persist the intent, restart, replay it.
+        let entry = delete_marker_purge_mrf_entry(&live, vec![arn.clone()]);
+        let replay_oi = ObjectInfo {
+            bucket: entry.bucket.clone(),
+            name: entry.object.clone(),
+            version_id: entry.version_id,
+            delete_marker: entry.delete_marker,
+            ..Default::default()
+        };
+        let dsc = replicate_decision_for_admitted_targets(&entry.target_arns);
+        let replayed = reconstructed_heal_delete_info(&entry, &replay_oi, &dsc);
+
+        assert_eq!(
+            delete_marker_purge_version_id(replayed.delete_object.replication_state.as_ref(), &arn, source_marker),
+            Some(Some(remote_marker)),
+            "the MRF replay must address the target-assigned marker version, not source marker {source_marker}"
+        );
+
+        // A refusal (inconsistent recorded ids) must stay a refusal across the
+        // journal round trip instead of degrading into the source-id fallback.
+        let mut refused = live;
+        refused
+            .delete_object
+            .replication_state
+            .as_mut()
+            .expect("state was set above")
+            .target_delete_marker_version_ids_corrupt = true;
+        let entry = delete_marker_purge_mrf_entry(&refused, vec![arn.clone()]);
+        assert!(entry.target_delete_marker_version_ids_corrupt);
+        let replayed = reconstructed_heal_delete_info(&entry, &replay_oi, &dsc);
+        assert_eq!(
+            delete_marker_purge_version_id(replayed.delete_object.replication_state.as_ref(), &arn, source_marker),
+            None,
+            "the MRF replay must keep refusing to guess when the recorded ids were inconsistent"
+        );
+    }
 }

crates/ecstore/src/bucket/replication/replication_resyncer.rs

@@ -32,11 +32,11 @@ use super::replication_msgp_boundary::ReplicationMsgpCodec;
 use super::replication_object_config::{ReplicationConfig, get_replication_config, must_replicate};
 use super::replication_object_decision_boundary::{
     MustReplicateOptions, ReplicationMultipartPartInput, delete_marker_purge_mrf_entry, delete_marker_purge_version_id,
-    delete_replication_creates_marker, heal_uses_delete_replication_path, is_object_lock_denied_delete,
-    is_retryable_delete_replication_head_error, is_version_delete_replication, replicate_delete_outcome, replication_etags_match,
-    replication_multipart_complete_actual_size, replication_multipart_part_plan, replication_single_put_size_error,
-    resync_existing_delete_replication_info, should_retry_delete_marker_purge, single_part_replica_etag_mismatch,
-    target_delete_version_id,
+    delete_replication_creates_marker, delete_replication_target_version_id, heal_uses_delete_replication_path,
+    is_object_lock_denied_delete, is_retryable_delete_replication_head_error, is_version_delete_replication,
+    replicate_delete_outcome, replication_etags_match, replication_multipart_complete_actual_size,
+    replication_multipart_part_plan, replication_single_put_size_error, resync_existing_delete_replication_info,
+    should_retry_delete_marker_purge, single_part_replica_etag_mismatch,
 };
 use super::replication_queue_boundary::{DeletedObjectReplicationInfo, ReplicationQueueAdmission};
 use super::replication_resync_boundary::ResyncStatusType;
@@ -2051,7 +2051,11 @@ pub(crate) async fn replicate_delete_with_outcome<S: ReplicationStorage>(
 
     let is_version_purge = is_version_delete_replication(&dobj.delete_object);
 
-    let requires_delayed_purge = should_retry_delete_marker_purge(&dobj.delete_object);
+    // The watcher exists to purge a replicated marker once the SOURCE marker
+    // vanishes. A version purge is that purge already (its failures reach the
+    // journal as a purge entry), so it must not spawn a second watcher that
+    // journals a duplicate intent (backlog#2290).
+    let requires_delayed_purge = should_retry_delete_marker_purge(&dobj.delete_object) && !is_version_purge;
 
     let (replication_status, prev_status) = if !is_version_purge {
         (
@@ -2761,12 +2765,6 @@ fn unavailable_delete_target_info(dobj: &DeletedObjectReplicationInfo, arn: &str
 }
 
 async fn replicate_delete_to_target(dobj: &DeletedObjectReplicationInfo, tgt_client: Arc<TargetClient>) -> ReplicatedTargetInfo {
-    let version_id = if let Some(version_id) = &dobj.delete_object.delete_marker_version_id {
-        version_id.to_owned()
-    } else {
-        dobj.delete_object.version_id.unwrap_or_default()
-    };
-
     let mut rinfo = dobj
         .delete_object
         .replication_state
@@ -2799,7 +2797,25 @@ async fn replicate_delete_to_target(dobj: &DeletedObjectReplicationInfo, tgt_cli
         return rinfo;
     }
 
-    let version_id = target_delete_version_id(version_id, is_version_purge);
+    // Purging a replicated delete marker addresses the version the target
+    // assigned (recorded when the marker was created there); see
+    // `delete_replication_target_version_id`. A corrupt record is a failure,
+    // not a guess: the entry stays visible until the metadata is repaired.
+    let Some(version_id) = delete_replication_target_version_id(&dobj.delete_object, &tgt_client.arn) else {
+        warn!(
+            event = EVENT_DELETE_MARKER_PURGE_FAILED,
+            component = LOG_COMPONENT_ECSTORE,
+            subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
+            bucket = tgt_client.bucket,
+            object = dobj.delete_object.object_name,
+            arn = %tgt_client.arn,
+            reason = "recorded_target_version_inconsistent",
+            "Replicated version purge refused: recorded target delete-marker version metadata is inconsistent"
+        );
+        rinfo.version_purge_status = VersionPurgeStatusType::Failed;
+        rinfo.error = Some("recorded target delete-marker version metadata is inconsistent".to_string());
+        return rinfo;
+    };
 
     if dobj.delete_object.delete_marker && dobj.delete_object.delete_marker_version_id.is_some() {
         match head_object_for_worker(

crates/iam/src/manager.rs

@@ -429,6 +429,27 @@ where
         }
     }
 
+    /// The cached mapping record for one user or group, looked up in the same
+    /// cache partition `policy_db_set` writes it to (group / STS / regular+service
+    /// user). `None` when no mapping is stored.
+    pub async fn get_mapped_policy_record(&self, name: &str, user_type: UserType, is_group: bool) -> Option<MappedPolicy> {
+        let cache = self.cache.snapshot();
+        if is_group {
+            cache.group_policies.get(name).cloned()
+        } else if user_type == UserType::Sts {
+            cache.sts_policies.get(name).cloned()
+        } else {
+            cache.user_policies.get(name).cloned()
+        }
+    }
+
+    /// The cached group record (members, status, own timestamp) without the
+    /// mapped-policy overlay `get_group_description` applies. `None` when the
+    /// group does not exist.
+    pub async fn get_group_info(&self, name: &str) -> Option<GroupInfo> {
+        self.cache.snapshot().groups.get(name).cloned()
+    }
+
     pub async fn get_policy(&self, name: &str) -> Result<Policy> {
         if name.is_empty() {
             return Err(Error::InvalidArgument);
@@ -534,6 +555,17 @@ where
     }
 
     pub async fn set_policy(&self, name: &str, policy: Policy) -> Result<OffsetDateTime> {
+        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<OffsetDateTime> {
         if name.is_empty() || policy.is_empty() {
             return Err(Error::InvalidArgument);
         }
@@ -544,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<HashMap<String, Policy>> {
@@ -789,6 +820,12 @@ where
 
     /// create a service account and update cache
     pub async fn add_service_account(&self, cred: Credentials) -> Result<OffsetDateTime> {
+        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<OffsetDateTime> {
         if cred.access_key.is_empty() || cred.parent_user.is_empty() {
             return Err(Error::InvalidArgument);
         }
@@ -800,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)
@@ -808,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<OffsetDateTime> {
+        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<OffsetDateTime> {
         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 {
@@ -858,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<String, Value> = if token_without_expiration {
@@ -916,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?;
@@ -1149,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<OffsetDateTime> {
+        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<OffsetDateTime> {
         if name.is_empty() {
             return Err(Error::InvalidArgument);
         }
@@ -1168,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);
@@ -1194,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<OffsetDateTime> {
@@ -1391,6 +1450,17 @@ where
     }
 
     pub async fn add_user(&self, access_key: &str, args: &AddOrUpdateUserReq) -> Result<OffsetDateTime> {
+        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<OffsetDateTime> {
         let cache = self.cache.snapshot();
         let users = Arc::clone(&cache.users);
         if let Some(x) = users.get(access_key) {
@@ -1408,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)
@@ -1421,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<()> {
@@ -1599,6 +1670,17 @@ where
     }
 
     pub async fn set_user_status(&self, access_key: &str, status: AccountStatus) -> Result<OffsetDateTime> {
+        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<OffsetDateTime> {
         if access_key.is_empty() {
             return Err(Error::InvalidArgument);
         }
@@ -1625,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);
 
@@ -1640,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<()> {
@@ -1676,6 +1759,17 @@ where
     }
 
     pub async fn add_users_to_group(&self, group: &str, members: Vec<String>) -> Result<OffsetDateTime> {
+        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<String>,
+        updated_at: OffsetDateTime,
+    ) -> Result<OffsetDateTime> {
         if group.is_empty() {
             return Err(Error::InvalidArgument);
         }
@@ -1693,6 +1787,14 @@ 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.
+        // `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();
@@ -1701,15 +1803,20 @@ where
                 uniq_set.extend(members.iter().cloned());
 
                 gi.members = uniq_set.into_iter().collect();
+                gi.update_at = Some(updated_at);
+                gi
+            }
+            None => {
+                let mut gi = GroupInfo::new(members.clone());
+                gi.update_at = Some(updated_at);
                 gi
             }
-            None => GroupInfo::new(members.clone()),
         };
         drop(cache);
 
         self.api.save_group_info(group, gi.clone()).await?;
 
-        let now = self.cache.with_write_lock(|cache| {
+        self.cache.with_write_lock(|cache| {
             let now = OffsetDateTime::now_utc();
             cache.add_or_update_group(group, &gi, now);
 
@@ -1719,13 +1826,18 @@ where
                 m.insert(group.to_string());
                 cache.add_or_update_user_group_membership(member, &m, now);
             });
-            now
         });
 
-        Ok(now)
+        Ok(updated_at)
     }
 
     pub async fn set_group_status(&self, name: &str, enable: bool) -> Result<OffsetDateTime> {
+        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<OffsetDateTime> {
         if name.is_empty() {
             return Err(Error::InvalidArgument);
         }
@@ -1743,12 +1855,15 @@ where
         } else {
             gi.status = STATUS_DISABLED.to_owned();
         }
+        gi.update_at = Some(updated_at);
 
         self.api.save_group_info(name, gi.clone()).await?;
 
+        // 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(OffsetDateTime::now_utc())
+        Ok(updated_at)
     }
 
     pub async fn get_group_description(&self, name: &str) -> Result<GroupDesc> {
@@ -1818,6 +1933,20 @@ where
         name: &str,
         members: Vec<String>,
         update_cache_only: bool,
+    ) -> Result<OffsetDateTime> {
+        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<String>,
+        update_cache_only: bool,
+        updated_at: OffsetDateTime,
     ) -> Result<OffsetDateTime> {
         let cache = self.cache.snapshot();
         let mut gi = cache
@@ -1830,12 +1959,14 @@ 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>>();
-
+        gi.update_at = Some(updated_at);
         if !update_cache_only {
             self.api.save_group_info(name, gi.clone()).await?;
         }
 
-        let now = self.cache.with_write_lock(|cache| {
+        self.cache.with_write_lock(|cache| {
+            // Sample after storage completes so a concurrent reload cannot
+            // make this publication older than the cache it must update.
             let now = OffsetDateTime::now_utc();
             cache.add_or_update_group(name, &gi, now);
 
@@ -1847,13 +1978,25 @@ where
                     cache.add_or_update_user_group_membership(member, &m, now);
                 }
             });
-            now
         });
 
-        Ok(now)
+        Ok(updated_at)
     }
 
     pub async fn remove_users_from_group(&self, group: &str, members: Vec<String>) -> Result<OffsetDateTime> {
+        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<String>,
+        updated_at: OffsetDateTime,
+    ) -> Result<OffsetDateTime> {
         if group.is_empty() {
             return Err(Error::InvalidArgument);
         }
@@ -1902,18 +2045,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) {
@@ -2235,6 +2377,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<String, PolicyDoc> {
     let default_policies = &DEFAULT_POLICIES;
     default_policies

crates/iam/src/sys.rs

@@ -385,7 +385,14 @@ impl<T: Store> IamSys<T> {
     }
 
     pub async fn set_policy(&self, name: &str, policy: Policy) -> Result<OffsetDateTime> {
-        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<OffsetDateTime> {
+        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<T: Store> IamSys<T> {
     }
 
     pub async fn set_user_status(&self, name: &str, status: rustfs_madmin::AccountStatus) -> Result<OffsetDateTime> {
-        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<OffsetDateTime> {
+        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<T: Store> IamSys<T> {
         parent_user: &str,
         groups: Option<Vec<String>>,
         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<Vec<String>>,
+        opts: NewServiceAccountOpts,
+        updated_at: OffsetDateTime,
     ) -> Result<(Credentials, OffsetDateTime)> {
         if parent_user.is_empty() {
             return Err(IamError::InvalidArgument);
@@ -724,11 +756,18 @@ impl<T: Store> IamSys<T> {
         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<T: Store> IamSys<T> {
     }
 
     pub async fn update_service_account(&self, name: &str, opts: UpdateServiceAccountOpts) -> Result<OffsetDateTime> {
+        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<OffsetDateTime> {
         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<T: Store> IamSys<T> {
     }
 
     pub async fn create_user(&self, access_key: &str, args: &AddOrUpdateUserReq) -> Result<OffsetDateTime> {
+        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<OffsetDateTime> {
         if !is_access_key_valid(access_key) {
             return Err(IamError::InvalidAccessKeyLength);
         }
@@ -952,7 +1014,7 @@ impl<T: Store> IamSys<T> {
             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<T: Store> IamSys<T> {
     }
 
     pub async fn add_users_to_group(&self, group: &str, users: Vec<String>) -> Result<OffsetDateTime> {
+        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<String>,
+        updated_at: OffsetDateTime,
+    ) -> Result<OffsetDateTime> {
         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<T: Store> IamSys<T> {
     }
 
     pub async fn remove_users_from_group(&self, group: &str, users: Vec<String>) -> Result<OffsetDateTime> {
-        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<String>,
+        updated_at: OffsetDateTime,
+    ) -> Result<OffsetDateTime> {
+        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<T: Store> IamSys<T> {
     }
 
     pub async fn set_group_status(&self, group: &str, enable: bool) -> Result<OffsetDateTime> {
-        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<OffsetDateTime> {
+        let updated_at = self.store.set_group_status_at(group, enable, updated_at).await?;
 
         self.notify_for_group(group).await;
 
@@ -1055,6 +1146,22 @@ impl<T: Store> IamSys<T> {
         self.store.get_group_description(group).await
     }
 
+    /// The stored group record itself (see `IamCache::get_group_info`).
+    pub async fn get_group_info(&self, group: &str) -> Option<GroupInfo> {
+        self.store.get_group_info(group).await
+    }
+
+    /// The stored policy document, `Error::NoSuchPolicy` when absent.
+    pub async fn get_policy_doc(&self, name: &str) -> Result<PolicyDoc> {
+        self.store.get_policy_doc(name).await
+    }
+
+    /// The stored mapping record for one user or group (see
+    /// `IamCache::get_mapped_policy_record`).
+    pub async fn get_mapped_policy_record(&self, name: &str, user_type: UserType, is_group: bool) -> Option<MappedPolicy> {
+        self.store.get_mapped_policy_record(name, user_type, is_group).await
+    }
+
     pub async fn list_groups_load(&self) -> Result<Vec<String>> {
         self.store.update_groups().await
     }
@@ -1064,7 +1171,24 @@ impl<T: Store> IamSys<T> {
     }
 
     pub async fn policy_db_set(&self, name: &str, user_type: UserType, is_group: bool, policy: &str) -> Result<OffsetDateTime> {
-        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<OffsetDateTime> {
+        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 {
@@ -1846,6 +1970,11 @@ pub struct NewServiceAccountOpts {
     pub expiration: Option<OffsetDateTime>,
     pub allow_site_replicator_account: bool,
     pub claims: Option<HashMap<String, Value>>,
+    /// 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<String>,
 }
 
 pub struct UpdateServiceAccountOpts {
@@ -2081,6 +2210,9 @@ mod tests {
         block_delete: Arc<std::sync::atomic::AtomicBool>,
         delete_started: Arc<tokio::sync::Notify>,
         release_delete: Arc<tokio::sync::Notify>,
+        block_group_save: Arc<std::sync::atomic::AtomicBool>,
+        group_save_started: Arc<tokio::sync::Notify>,
+        group_save_release: Arc<tokio::sync::Notify>,
     }
 
     impl StsTestMockStore {
@@ -2094,6 +2226,9 @@ mod tests {
                 block_delete: Arc::new(std::sync::atomic::AtomicBool::new(false)),
                 delete_started: Arc::new(tokio::sync::Notify::new()),
                 release_delete: Arc::new(tokio::sync::Notify::new()),
+                block_group_save: Arc::new(std::sync::atomic::AtomicBool::new(false)),
+                group_save_started: Arc::new(tokio::sync::Notify::new()),
+                group_save_release: Arc::new(tokio::sync::Notify::new()),
             }
         }
 
@@ -2197,11 +2332,15 @@ mod tests {
         }
 
         async fn save_group_info(&self, _name: &str, _item: GroupInfo) -> Result<()> {
-            Err(Error::InvalidArgument)
+            if self.block_group_save.load(std::sync::atomic::Ordering::SeqCst) {
+                self.group_save_started.notify_one();
+                self.group_save_release.notified().await;
+            }
+            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<()> {
@@ -2378,6 +2517,140 @@ mod tests {
         IamSys::new(cache)
     }
 
+    async fn assert_group_write_during_reload_is_published(remove: bool) {
+        let iam_sys = Arc::new(temp_env::async_with_vars([("RUSTFS_SKIP_BACKGROUND_TASK", Some("1"))], test_iam_sys()).await);
+        let member = "sts-fallback-test-parent";
+        let group = if remove { "testgroup" } else { "new-published-group" };
+        let source_time = OffsetDateTime::now_utc() - time::Duration::hours(1);
+        iam_sys
+            .store
+            .api
+            .block_group_save
+            .store(true, std::sync::atomic::Ordering::SeqCst);
+        let before = iam_sys.store.cache.snapshot();
+        let writer_iam = iam_sys.clone();
+        let writer = tokio::spawn(async move {
+            if remove {
+                writer_iam
+                    .remove_users_from_group_at(group, vec![member.to_string()], source_time)
+                    .await
+            } else {
+                writer_iam
+                    .add_users_to_group_at(group, vec![member.to_string()], source_time)
+                    .await
+            }
+        });
+        tokio::time::timeout(std::time::Duration::from_secs(5), iam_sys.store.api.group_save_started.notified())
+            .await
+            .expect("group save should reach the barrier");
+        // The pending store write has not changed the cache, so the production
+        // full-reload snapshot guard permits this replacement.
+        assert!(iam_sys.store.cache.with_write_lock(|cache| cache.matches_snapshot(&before)));
+        iam_sys
+            .store
+            .api
+            .load_all(&iam_sys.store.cache)
+            .await
+            .expect("reload while group save is pending");
+        iam_sys.store.api.group_save_release.notify_one();
+        assert_eq!(writer.await.expect("join group writer").expect("group write should succeed"), source_time);
+        let info = iam_sys
+            .get_group_info(group)
+            .await
+            .expect("successful group write must remain readable after reload");
+        assert_eq!(info.update_at, Some(source_time), "source timestamp must remain on the record");
+        assert_eq!(info.members, if remove { Vec::new() } else { vec![member.to_string()] });
+        let groups = iam_sys.store.cache.snapshot().user_group_memberships.get(member).cloned();
+        assert_eq!(
+            groups.is_some_and(|groups| groups.contains(group)),
+            !remove,
+            "membership index must reflect the write"
+        );
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn add_group_write_during_reload_publishes_after_store_save() {
+        assert_group_write_during_reload_is_published(false).await;
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn remove_group_write_during_reload_publishes_after_store_save() {
+        assert_group_write_during_reload_is_published(true).await;
+    }
+
+    /// 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(),

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;

crates/replication/src/delete.rs

@@ -76,6 +76,21 @@ impl ReplicationWorkerOperation for DeletedObjectReplicationInfo {
                 .delete_object
                 .delete_marker_mtime
                 .and_then(|t| i64::try_from(t.unix_timestamp_nanos()).ok()),
+            // Carry the target-assigned marker version ids (and the fail-closed corrupt
+            // flag) into the journal so a purge intent replayed after a restart addresses
+            // the same version the live path did (backlog#2290). Only delete-marker state
+            // ever records these; other deletes serialize an empty map.
+            target_delete_marker_version_ids: self
+                .delete_object
+                .replication_state
+                .as_ref()
+                .map(|state| state.target_delete_marker_version_ids.clone())
+                .unwrap_or_default(),
+            target_delete_marker_version_ids_corrupt: self
+                .delete_object
+                .replication_state
+                .as_ref()
+                .is_some_and(|state| state.target_delete_marker_version_ids_corrupt),
             target_arns: self.admitted_target_arns(),
             force_delete_id: self.delete_object.force_delete_id,
             force_delete_generation: self.delete_object.force_delete_generation,
@@ -238,6 +253,28 @@ pub fn delete_marker_purge_version_id(
     })
 }
 
+/// The version a delete replication addresses on `arn`, or `None` to refuse.
+///
+/// A version purge whose purged version is a delete marker must address the
+/// marker version the TARGET assigned — the recorded mapping, exactly as the
+/// delayed-purge watcher does. The source-side `DELETE ?versionId=<marker>`
+/// replicates as such a purge, and a generic S3 target answers a DELETE of an
+/// unknown versionId with 204 while keeping its marker, so addressing it by
+/// the source id reported success and left the marker behind (backlog#2290,
+/// R6.1 on the VMs). Nothing recorded falls back to the source-derived id
+/// (id-mirroring peers); a corrupt record refuses, as the watcher does.
+pub fn delete_replication_target_version_id(dobj: &DeletedObject, arn: &str) -> Option<Option<String>> {
+    let is_version_purge = is_version_delete_replication(dobj);
+    if is_version_purge
+        && !dobj.delete_marker
+        && let Some(marker) = dobj.delete_marker_version_id
+    {
+        return delete_marker_purge_version_id(dobj.replication_state.as_ref(), arn, marker);
+    }
+    let source_version = dobj.delete_marker_version_id.or(dobj.version_id).unwrap_or_default();
+    Some(target_delete_version_id(source_version, is_version_purge))
+}
+
 /// Shape an exhausted purge intent as a marker-creation delete entry. Replay
 /// reconstructs it with `delete_marker: true`, finds the source marker gone,
 /// and funnels into the stale-marker branch of `replicate_delete_with_outcome`
@@ -258,9 +295,9 @@ mod tests {
 
     use super::{
         DeletedObjectReplicationInfo, delete_marker_purge_mrf_entry, delete_marker_purge_version_id,
-        delete_replication_creates_marker, is_object_lock_denied_delete, is_retryable_delete_replication_head_error,
-        is_version_delete_replication, replicate_delete_outcome, resync_existing_delete_replication_info,
-        should_retry_delete_marker_purge, target_delete_version_id,
+        delete_replication_creates_marker, delete_replication_target_version_id, is_object_lock_denied_delete,
+        is_retryable_delete_replication_head_error, is_version_delete_replication, replicate_delete_outcome,
+        resync_existing_delete_replication_info, should_retry_delete_marker_purge, target_delete_version_id,
     };
     use crate::storage_api::DeletedObject;
     use crate::{
@@ -595,6 +632,76 @@ mod tests {
         assert_eq!(entry.retry_count, 0);
         assert_eq!(entry.bucket, "bucket-a");
         assert_eq!(entry.object, "doc.txt");
+        assert!(
+            entry.target_delete_marker_version_ids.is_empty(),
+            "no recorded target marker ids means the journal carries none"
+        );
+        assert!(!entry.target_delete_marker_version_ids_corrupt);
+    }
+
+    /// backlog#2290: a purge intent journaled to MRF must carry the marker
+    /// version ids the targets assigned, plus the fail-closed corrupt flag,
+    /// so a replay after restart addresses the same version the live path did.
+    #[test]
+    fn delete_marker_purge_mrf_entry_carries_target_assigned_marker_versions() {
+        let delete_marker_version_id = Uuid::new_v4();
+        let mut state = ReplicationState::default();
+        state
+            .target_delete_marker_version_ids
+            .insert("arn:a".to_string(), "remote-marker-a".to_string());
+        state
+            .target_delete_marker_version_ids
+            .insert("arn:b".to_string(), "remote-marker-b".to_string());
+        let mut dobj = DeletedObjectReplicationInfo {
+            delete_object: DeletedObject {
+                object_name: "doc.txt".to_string(),
+                delete_marker: false,
+                version_id: Some(Uuid::new_v4()),
+                delete_marker_version_id: Some(delete_marker_version_id),
+                replication_state: Some(state),
+                ..Default::default()
+            },
+            bucket: "bucket-a".to_string(),
+            ..Default::default()
+        };
+
+        let entry = delete_marker_purge_mrf_entry(&dobj, vec!["arn:a".to_string()]);
+        assert_eq!(
+            entry.target_delete_marker_version_ids,
+            HashMap::from([
+                ("arn:a".to_string(), "remote-marker-a".to_string()),
+                ("arn:b".to_string(), "remote-marker-b".to_string()),
+            ]),
+            "every recorded target marker id survives the journal, regardless of the retried ARN subset"
+        );
+        assert!(!entry.target_delete_marker_version_ids_corrupt);
+        assert_eq!(
+            delete_marker_purge_version_id(
+                Some(&ReplicationState {
+                    target_delete_marker_version_ids: entry.target_delete_marker_version_ids,
+                    ..Default::default()
+                }),
+                "arn:a",
+                delete_marker_version_id
+            ),
+            Some(Some("remote-marker-a".to_string()))
+        );
+
+        // The live path refuses to purge on inconsistent metadata and reports the target
+        // as failed; the journaled intent must keep refusing after a restart.
+        dobj.delete_object
+            .replication_state
+            .as_mut()
+            .expect("state was set above")
+            .target_delete_marker_version_ids_corrupt = true;
+        let entry = delete_marker_purge_mrf_entry(&dobj, vec!["arn:a".to_string()]);
+        assert!(entry.target_delete_marker_version_ids_corrupt);
+
+        // A delete without replication state journals an empty map.
+        dobj.delete_object.replication_state = None;
+        let entry = dobj.to_mrf_entry();
+        assert!(entry.target_delete_marker_version_ids.is_empty());
+        assert!(!entry.target_delete_marker_version_ids_corrupt);
     }
 
     #[test]
@@ -656,4 +763,57 @@ mod tests {
         assert!(!is_object_lock_denied_delete(Some("InternalError"), Some("retention lookup failed")));
         assert!(!is_object_lock_denied_delete(None, Some("legal hold")));
     }
+
+    fn purge_of_marker(marker: Uuid, state: Option<ReplicationState>) -> DeletedObject {
+        DeletedObject {
+            object_name: "obj".to_string(),
+            delete_marker: false,
+            delete_marker_version_id: Some(marker),
+            version_id: None,
+            replication_state: state,
+            ..Default::default()
+        }
+    }
+
+    #[test]
+    fn delete_replication_target_version_id_addresses_recorded_marker_for_purges() {
+        let arn = "arn:minio:replication::generic:photos";
+        let marker = Uuid::new_v4();
+        let mut state = ReplicationState::default();
+        state
+            .target_delete_marker_version_ids
+            .insert(arn.to_string(), "remote-marker".to_string());
+
+        // purge of a replicated marker: the target's own version
+        assert_eq!(
+            delete_replication_target_version_id(&purge_of_marker(marker, Some(state.clone())), arn),
+            Some(Some("remote-marker".to_string()))
+        );
+        // nothing recorded for this arn: the source-derived id (id-mirroring peers)
+        assert_eq!(
+            delete_replication_target_version_id(&purge_of_marker(marker, None), arn),
+            Some(Some(marker.to_string()))
+        );
+        // corrupt record: refuse instead of guessing
+        state.target_delete_marker_version_ids_corrupt = true;
+        assert_eq!(delete_replication_target_version_id(&purge_of_marker(marker, Some(state)), arn), None);
+
+        // marker creation keeps the source id (the target mints its own on a
+        // versionless DELETE; the id only travels in the source header)
+        let creation = DeletedObject {
+            object_name: "obj".to_string(),
+            delete_marker: true,
+            delete_marker_version_id: Some(marker),
+            ..Default::default()
+        };
+        assert_eq!(delete_replication_target_version_id(&creation, arn), Some(Some(marker.to_string())));
+        // plain version purge: the source version id
+        let version = Uuid::new_v4();
+        let purge = DeletedObject {
+            object_name: "obj".to_string(),
+            version_id: Some(version),
+            ..Default::default()
+        };
+        assert_eq!(delete_replication_target_version_id(&purge, arn), Some(Some(version.to_string())));
+    }
 }

crates/replication/src/filemeta.rs

@@ -641,6 +641,26 @@ pub struct MrfReplicateEntry {
     #[serde(rename = "deleteMarkerMtime", skip_serializing_if = "Option::is_none", default)]
     pub delete_marker_mtime: Option<i64>,
 
+    // For delete-marker purge intents: the exact version id each target assigned to the
+    // replicated marker, keyed by target ARN. A generic S3 target mints its own version ids
+    // and answers a DELETE of an unknown id with 204, so a replay that fell back to the source
+    // marker id would be acknowledged while the real marker stayed behind (backlog#2290).
+    // Old files lack this key; default=empty means "unknown" and replay keeps the source-id
+    // fallback it always had.
+    #[serde(rename = "targetDeleteMarkerVersionIDs", skip_serializing_if = "HashMap::is_empty", default)]
+    pub target_delete_marker_version_ids: HashMap<String, String>,
+
+    // Companion to the map above: the source metadata disagreed about the recorded ids when
+    // the intent was journaled, so the live path refused to guess and reported the target as
+    // failed. Replay must keep refusing instead of falling back to the source id. Old files
+    // lack this key; default=false.
+    #[serde(
+        rename = "targetDeleteMarkerVersionIDsCorrupt",
+        skip_serializing_if = "std::ops::Not::not",
+        default
+    )]
+    pub target_delete_marker_version_ids_corrupt: bool,
+
     #[serde(rename = "targetARNs", skip_serializing_if = "Vec::is_empty", default)]
     pub target_arns: Vec<String>,
 

crates/replication/src/lib.rs

@@ -41,9 +41,9 @@ pub use config::{
 };
 pub use delete::{
     DeletedObjectReplicationInfo, delete_marker_purge_mrf_entry, delete_marker_purge_version_id,
-    delete_replication_creates_marker, is_object_lock_denied_delete, is_retryable_delete_replication_head_error,
-    is_version_delete_replication, replicate_delete_outcome, resync_existing_delete_replication_info,
-    should_retry_delete_marker_purge, target_delete_version_id,
+    delete_replication_creates_marker, delete_replication_target_version_id, is_object_lock_denied_delete,
+    is_retryable_delete_replication_head_error, is_version_delete_replication, replicate_delete_outcome,
+    resync_existing_delete_replication_info, should_retry_delete_marker_purge, target_delete_version_id,
 };
 pub use filemeta::{
     NULL_VERSION_ID, REPLICATE_EXISTING, REPLICATE_EXISTING_DELETE, REPLICATE_HEAL, REPLICATE_HEAL_DELETE, REPLICATE_INCOMING,

crates/replication/src/mrf.rs

@@ -31,8 +31,13 @@ const CAPABILITY_OPERATION_KIND: u64 = 1 << 0;
 const CAPABILITY_TARGET_ARNS: u64 = 1 << 1;
 const CAPABILITY_FORCE_DELETE: u64 = 1 << 2;
 const CAPABILITY_DELETE_MARKER_MTIME: u64 = 1 << 3;
-const MRF_KNOWN_CAPABILITIES: u64 =
-    CAPABILITY_OPERATION_KIND | CAPABILITY_TARGET_ARNS | CAPABILITY_FORCE_DELETE | CAPABILITY_DELETE_MARKER_MTIME;
+// Per-ARN target-assigned delete-marker version ids on purge intents (backlog#2290).
+const CAPABILITY_TARGET_DELETE_MARKER_VERSION_IDS: u64 = 1 << 4;
+const MRF_KNOWN_CAPABILITIES: u64 = CAPABILITY_OPERATION_KIND
+    | CAPABILITY_TARGET_ARNS
+    | CAPABILITY_FORCE_DELETE
+    | CAPABILITY_DELETE_MARKER_MTIME
+    | CAPABILITY_TARGET_DELETE_MARKER_VERSION_IDS;
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum MrfCapability {
@@ -40,6 +45,7 @@ pub enum MrfCapability {
     TargetArns,
     ForceDelete,
     DeleteMarkerMtime,
+    TargetDeleteMarkerVersionIds,
 }
 
 impl MrfCapability {
@@ -49,6 +55,7 @@ impl MrfCapability {
             Self::TargetArns => CAPABILITY_TARGET_ARNS,
             Self::ForceDelete => CAPABILITY_FORCE_DELETE,
             Self::DeleteMarkerMtime => CAPABILITY_DELETE_MARKER_MTIME,
+            Self::TargetDeleteMarkerVersionIds => CAPABILITY_TARGET_DELETE_MARKER_VERSION_IDS,
         }
     }
 }
@@ -601,9 +608,17 @@ pub fn decode_mrf_file(data: &[u8]) -> Result<Vec<MrfReplicateEntry>> {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use std::collections::HashMap;
     use uuid::Uuid;
 
+    // Capability word 31 = OperationKind | TargetArns | ForceDelete | DeleteMarkerMtime |
+    // TargetDeleteMarkerVersionIds (backlog#2290).
     const ENVELOPE_FIXTURE: &[u8] = &[
+        b'M', b'R', b'F', b'E', 1, 0, 1, 0, 1, 0, 0, 0, 31, 0, 0, 0, 0, 0, 0, 0, 3, 0, 0, 0, 1, 2, 3,
+    ];
+
+    // The envelope a binary from before backlog#2290 writes: same header, capability word 15.
+    const PRE_TARGET_MARKER_IDS_ENVELOPE_FIXTURE: &[u8] = &[
         b'M', b'R', b'F', b'E', 1, 0, 1, 0, 1, 0, 0, 0, 15, 0, 0, 0, 0, 0, 0, 0, 3, 0, 0, 0, 1, 2, 3,
     ];
 
@@ -626,6 +641,8 @@ mod tests {
                 delete_marker_version_id: None,
                 delete_marker: false,
                 delete_marker_mtime: None,
+                target_delete_marker_version_ids: HashMap::new(),
+                target_delete_marker_version_ids_corrupt: false,
                 target_arns: vec!["arn:target-a".to_string()],
             },
             MrfReplicateEntry {
@@ -642,6 +659,8 @@ mod tests {
                 delete_marker_version_id: None,
                 delete_marker: false,
                 delete_marker_mtime: None,
+                target_delete_marker_version_ids: HashMap::new(),
+                target_delete_marker_version_ids_corrupt: false,
                 target_arns: vec!["arn:target-a".to_string(), "arn:target-b".to_string()],
             },
             MrfReplicateEntry {
@@ -658,6 +677,11 @@ mod tests {
                 delete_marker_version_id: Some(del_vid),
                 delete_marker: true,
                 delete_marker_mtime: Some(1_705_312_200_123_456_789),
+                target_delete_marker_version_ids: HashMap::from([
+                    ("arn:target-a".to_string(), "remote-marker-a".to_string()),
+                    ("arn:target-b".to_string(), "remote-marker-b".to_string()),
+                ]),
+                target_delete_marker_version_ids_corrupt: false,
                 target_arns: vec!["arn:target-a".to_string()],
             },
         ];
@@ -685,6 +709,54 @@ mod tests {
             Some(1_705_312_200_123_456_789),
             "delete-marker mtime must survive the MRF disk round-trip"
         );
+        assert!(decoded[0].target_delete_marker_version_ids.is_empty());
+        assert!(decoded[1].target_delete_marker_version_ids.is_empty());
+        assert_eq!(
+            decoded[2].target_delete_marker_version_ids,
+            HashMap::from([
+                ("arn:target-a".to_string(), "remote-marker-a".to_string()),
+                ("arn:target-b".to_string(), "remote-marker-b".to_string()),
+            ]),
+            "target-assigned marker version ids must survive the MRF disk round-trip (backlog#2290)"
+        );
+        assert!(!decoded[2].target_delete_marker_version_ids_corrupt);
+    }
+
+    /// backlog#2290: the corrupt flag rides the same journal round trip, and an
+    /// entry that carries neither field encodes exactly as it did before the
+    /// field existed (both keys are skipped when empty/false).
+    #[test]
+    fn mrf_file_round_trips_target_marker_ids_corrupt_flag_and_skips_empty_keys() {
+        let corrupt = MrfReplicateEntry {
+            bucket: "bucket-a".to_string(),
+            object: "delete-a".to_string(),
+            op: MrfOpKind::Delete,
+            delete_marker: true,
+            delete_marker_version_id: Some(Uuid::new_v4()),
+            target_delete_marker_version_ids_corrupt: true,
+            target_arns: vec!["arn:target-a".to_string()],
+            ..Default::default()
+        };
+        let decoded = decode_mrf_file(&encode_mrf_file(std::slice::from_ref(&corrupt)).expect("mrf file should encode"))
+            .expect("mrf file should decode");
+        assert_eq!(decoded, vec![corrupt]);
+        assert!(decoded[0].target_delete_marker_version_ids_corrupt);
+
+        let plain = MrfReplicateEntry {
+            bucket: "bucket-a".to_string(),
+            object: "delete-a".to_string(),
+            op: MrfOpKind::Delete,
+            delete_marker: true,
+            target_arns: vec!["arn:target-a".to_string()],
+            ..Default::default()
+        };
+        let encoded = encode_mrf_file(std::slice::from_ref(&plain)).expect("mrf file should encode");
+        let payload = String::from_utf8_lossy(&encoded);
+        assert!(
+            !payload.contains("targetDeleteMarkerVersionIDs"),
+            "an entry without recorded ids must not grow the new keys: {payload}"
+        );
+        assert_eq!(decode_mrf_file(&encoded).expect("mrf file should decode"), vec![plain]);
     }
 
     #[test]
@@ -719,6 +791,99 @@ mod tests {
         // Old files lack the deleteMarkerMtime key; it must default to None so replay keeps the
         // pre-#867 fallback to the current time.
         assert_eq!(decoded[0].delete_marker_mtime, None);
+        // Old files also lack the target marker id keys; they must default to an empty map
+        // and a clear corrupt flag so replay keeps the pre-#2290 source-id fallback.
+        assert!(decoded[0].target_delete_marker_version_ids.is_empty());
+        assert!(!decoded[0].target_delete_marker_version_ids_corrupt);
+    }
+
+    /// backlog#2290: a delete-marker entry written by a binary that predates the
+    /// `targetDeleteMarkerVersionIDs` key decodes with an empty map and a clear
+    /// corrupt flag — the exact shape replay handled before the field existed.
+    #[test]
+    fn mrf_pre_target_marker_ids_delete_entry_decodes_with_empty_map() {
+        let marker_version_id = Uuid::new_v4();
+        let mut payload = Vec::new();
+        rmp::encode::write_array_len(&mut payload, 1).expect("array len should encode");
+        rmp::encode::write_map_len(&mut payload, 9).expect("map len should encode");
+        rmp::encode::write_str(&mut payload, "bucket").expect("bucket key should encode");
+        rmp::encode::write_str(&mut payload, "old-bucket").expect("bucket value should encode");
+        rmp::encode::write_str(&mut payload, "object").expect("object key should encode");
+        rmp::encode::write_str(&mut payload, "old-key").expect("object value should encode");
+        rmp::encode::write_str(&mut payload, "retryCount").expect("retry key should encode");
+        rmp::encode::write_i32(&mut payload, 0).expect("retry value should encode");
+        rmp::encode::write_str(&mut payload, "size").expect("size key should encode");
+        rmp::encode::write_i64(&mut payload, 0).expect("size value should encode");
+        rmp::encode::write_str(&mut payload, "op").expect("op key should encode");
+        rmp::encode::write_str(&mut payload, "delete").expect("op value should encode");
+        rmp::encode::write_str(&mut payload, "forceDelete").expect("forceDelete key should encode");
+        rmp::encode::write_bool(&mut payload, false).expect("forceDelete value should encode");
+        rmp::encode::write_str(&mut payload, "deleteMarkerVersionID").expect("marker id key should encode");
+        // Uuid serializes as a 16-byte bin in the MessagePack journal.
+        rmp::encode::write_bin(&mut payload, marker_version_id.as_bytes()).expect("marker id value should encode");
+        rmp::encode::write_str(&mut payload, "deleteMarker").expect("deleteMarker key should encode");
+        rmp::encode::write_bool(&mut payload, true).expect("deleteMarker value should encode");
+        rmp::encode::write_str(&mut payload, "targetARNs").expect("targetARNs key should encode");
+        rmp::encode::write_array_len(&mut payload, 1).expect("targetARNs len should encode");
+        rmp::encode::write_str(&mut payload, "arn:target-a").expect("targetARNs value should encode");
+
+        let mut data = Vec::with_capacity(4 + payload.len());
+        data.extend_from_slice(&MRF_META_FORMAT.to_le_bytes());
+        data.extend_from_slice(&MRF_META_VERSION.to_le_bytes());
+        data.extend_from_slice(&payload);
+
+        let decoded = decode_mrf_file(&data).expect("pre-#2290 delete-marker entry should decode");
+
+        assert_eq!(decoded.len(), 1);
+        assert_eq!(decoded[0].op, MrfOpKind::Delete);
+        assert!(decoded[0].delete_marker);
+        assert_eq!(decoded[0].delete_marker_version_id, Some(marker_version_id));
+        assert_eq!(decoded[0].target_arns, vec!["arn:target-a".to_string()]);
+        assert!(decoded[0].target_delete_marker_version_ids.is_empty());
+        assert!(!decoded[0].target_delete_marker_version_ids_corrupt);
+    }
+
+    /// backlog#2290: the new field is fenced by its own capability bit exactly
+    /// like the earlier optional fields — a reader without the bit refuses an
+    /// envelope that advertises it, while the current reader still accepts the
+    /// pre-#2290 envelope.
+    #[test]
+    fn envelope_target_marker_ids_capability_is_fenced_and_backward_compatible() {
+        assert!(MrfCapabilities::current().contains(MrfCapability::TargetDeleteMarkerVersionIds));
+        assert_eq!(MrfCapabilities::with(MrfCapability::TargetDeleteMarkerVersionIds).bits(), 1 << 4);
+
+        // Old envelope, current reader: accepted, and the negotiated set lacks the new bit.
+        let legacy = MrfEnvelope::decode(PRE_TARGET_MARKER_IDS_ENVELOPE_FIXTURE, MrfProtocolCapabilities::current())
+            .expect("pre-#2290 envelope should decode");
+        assert_eq!(legacy.protocol().capabilities().bits(), 15);
+        assert!(
+            !legacy
+                .protocol()
+                .capabilities()
+                .contains(MrfCapability::TargetDeleteMarkerVersionIds)
+        );
+        assert_eq!(legacy.payload(), &[1, 2, 3]);
+
+        // Current envelope, reader that only knows the pre-#2290 bits: refused.
+        let pre_2290_reader = MrfProtocolCapabilities::new(1, 1, MrfCapabilities::from_bits(15).expect("known bits"));
+        assert_eq!(
+            MrfEnvelope::decode(ENVELOPE_FIXTURE, pre_2290_reader),
+            Err(MrfEnvelopeError::MissingCapabilities {
+                required: 31,
+                available: 15,
+            })
+        );
+
+        // Negotiation with such a peer drops the bit instead of failing.
+        let negotiated = MrfProtocolCapabilities::current()
+            .negotiate(pre_2290_reader)
+            .expect("negotiation with a pre-#2290 peer should succeed");
+        assert!(
+            !negotiated
+                .capabilities()
+                .contains(MrfCapability::TargetDeleteMarkerVersionIds)
+        );
+        assert!(negotiated.capabilities().contains(MrfCapability::DeleteMarkerMtime));
     }
 
     #[test]