Merge branch 'main' into houseme/feat/heal-outcomes-delivery

Limit synchronous mock mutex guards to pre-await scopes in canonical outcome tests.

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

crates/e2e_test/src/on_demand_migration/concurrency_test.rs

@@ -20,9 +20,10 @@
 //! journal (`count_requests`) carries the assertion in every one of them.
 
 use super::common::{BoxError, OdmTestEnv, RawResponse, SeedObject, start_configured_env};
-use crate::fake_s3_target::Operation;
+use crate::fake_s3_target::{FaultAction, Operation};
 use aws_sdk_s3::types::{BucketVersioningStatus, VersioningConfiguration};
 use bytes::Bytes;
+use futures::{StreamExt, TryStreamExt};
 use std::time::Duration;
 
 type TestResult = Result<(), BoxError>;
@@ -145,14 +146,38 @@ async fn test_odm_range_burst_overflows_the_pull_queue_without_failing_clients()
     .await?;
 
     let body = payload(128 * 1024);
+    let blocker = "queue/blocker.bin";
+    env.seed_source(SOURCE_BUCKET, &[SeedObject::new(blocker, body.clone())]);
+    // The one-chunk range completes immediately; its full background pull
+    // occupies the only slot while the remaining requests fill the queue.
+    env.source.inject_for_key(
+        Operation::GetObject,
+        blocker,
+        FaultAction::SlowSendBody {
+            chunk_bytes: 1024,
+            delay: Duration::from_millis(100),
+        },
+        2,
+    );
+    let response = env
+        .raw_object_request(http::Method::GET, bucket, blocker, &[("range", "bytes=0-1023")])
+        .await?;
+    assert_eq!(response.status, 206);
+    assert_eq!(response.body, body.slice(0..1024));
+    env.wait_for_status_counter(bucket, "/inflight_pulls", 1, SETTLE).await?;
+
     let keys: Vec<String> = (0..REQUESTS).map(|index| format!("queue/object-{index:03}.bin")).collect();
     let seeds: Vec<SeedObject> = keys.iter().map(|key| SeedObject::new(key.clone(), body.clone())).collect();
     env.seed_source(SOURCE_BUCKET, &seeds);
 
-    let responses: Vec<RawResponse> = futures::future::try_join_all(
+    // Bound source connections below the fixture's limit while still
+    // submitting all 100 requests to the eight-slot background queue.
+    let responses: Vec<RawResponse> = futures::stream::iter(
         keys.iter()
             .map(|key| env.raw_object_request(http::Method::GET, bucket, key, &[("range", "bytes=0-1023")])),
     )
+    .buffered(16)
+    .try_collect()
     .await?;
     for (key, response) in keys.iter().zip(&responses) {
         assert_eq!(response.status, 206, "{key}: {}", String::from_utf8_lossy(&response.body));
@@ -168,6 +193,15 @@ async fn test_odm_range_burst_overflows_the_pull_queue_without_failing_clients()
         .wait_for_status_counter(bucket, "/counters/pull_failures_total/queue_full", 1, SETTLE)
         .await?;
     assert!(queue_full > 0, "a 100-deep burst must overflow an 8-slot queue");
+    let queue_full = usize::try_from(queue_full)?;
+    assert!(queue_full <= REQUESTS);
+    env.wait_for_status_counter(
+        bucket,
+        "/counters/pulled_objects_total/background",
+        u64::try_from(REQUESTS + 1 - queue_full)?,
+        SETTLE,
+    )
+    .await?;
 
     let ranged_reads: usize = keys.iter().map(|key| source_get_count(&env, key)).sum();
     assert!(
@@ -175,9 +209,6 @@ async fn test_odm_range_burst_overflows_the_pull_queue_without_failing_clients()
         "every reader is served from the source: {ranged_reads} GETs for {REQUESTS} readers"
     );
     let dropped = keys.iter().filter(|key| source_get_count(&env, key) == 1).count();
-    assert!(
-        dropped > 0,
-        "the overflowed keys are the ones with no backfill GET, but every key got one"
-    );
+    assert_eq!(dropped, queue_full, "only overflowed keys remain without a background GET");
     Ok(())
 }

crates/e2e_test/src/on_demand_migration/list_through_test.rs

@@ -265,16 +265,13 @@ async fn list_through_rejects_a_tampered_continuation_token() -> TestResult {
     let decoded = String::from_utf8(base64_simd::STANDARD.decode_to_vec(token.as_bytes())?)?;
     assert!(decoded.contains("\"t\":\"odm-list\""), "the merged token is an envelope: {decoded}");
 
-    let tampered = base64_simd::STANDARD.encode_to_string(decoded.replace("\"v\":1", "\"v\":2").as_bytes());
-    let rejected = env
-        .raw_list_objects_v2(bucket, &format!("continuation-token={tampered}"))
-        .await?;
-    assert_eq!(
-        rejected.status,
-        400,
-        "a bumped token version is a client error: {}",
-        String::from_utf8_lossy(&rejected.body)
-    );
+    let tampered = base64_simd::STANDARD.encode_to_string(decoded.replace("\"v\":1", "\"v\":3").as_bytes());
+    assert_ne!(tampered, token, "the test must change the token version");
+    let query = serde_urlencoded::to_string([("continuation-token", tampered.as_str())])?;
+    let rejected = env.raw_list_objects_v2(bucket, &query).await?;
+    let error_body = String::from_utf8_lossy(&rejected.body);
+    assert_eq!(rejected.status, 400, "a bumped token version is a client error: {}", error_body);
+    assert!(error_body.contains("<Code>InvalidArgument</Code>"), "{error_body}");
     Ok(())
 }
 

crates/ecstore/src/api/mod.rs

@@ -151,17 +151,15 @@ 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_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,
+            capture_bucket_metadata_incarnation, delete, delete_if_incarnation, delete_under_transaction_lock, get,
+            get_accelerate_config, get_bucket_policy, get_bucket_policy_raw, get_bucket_targets_config, get_config_from_disk,
+            get_cors_config, get_durability_config, get_global_bucket_metadata_sys, get_lifecycle_config, get_logging_config,
+            get_notification_config, get_object_lock_config, get_object_lock_config_state, get_on_demand_migration_config,
             get_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_if_incarnation_at, update_quota_if_incarnation, update_quota_if_incarnation_at, update_under_transaction_lock,
-            update_under_transaction_lock_at,
+            update_quota_if_incarnation, update_under_transaction_lock,
         };
         #[cfg(feature = "test-util")]
         pub use crate::bucket::metadata_sys::{ConfigWriteLockProbe, test_support};

crates/ecstore/src/bucket/metadata.rs

@@ -791,22 +791,9 @@ 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> {
-        self.update_config_at(config_file, data, OffsetDateTime::now_utc())
-    }
+        let updated = 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;
@@ -1538,39 +1525,6 @@ 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,32 +567,6 @@ 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
 }
@@ -603,30 +577,6 @@ 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
 }
@@ -648,41 +598,34 @@ async fn update_with_sys(
     config_file: &str,
     data: Vec<u8>,
 ) -> Result<OffsetDateTime> {
-    update_with_sys_expected(sys, bucket, config_file, data, None, None).await
+    update_with_sys_expected(sys, bucket, config_file, data, 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, updated_at).await
+    update_under_config_write_guard(sys, &guard, config_file, data).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, None).await
+    delete_with_sys_expected(sys, bucket, config_file, 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, updated_at).await
+    delete_under_config_write_guard(sys, &guard, config_file).await
 }
 
 /// Owns the complete bucket-config mutation fence.
@@ -829,21 +772,7 @@ 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, 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
+    update_under_config_write_guard(get_bucket_metadata_sys()?, guard, config_file, data).await
 }
 
 /// Clear one config file while the caller holds this bucket's transaction lock.
@@ -853,7 +782,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, None).await
+    delete_under_config_write_guard(get_bucket_metadata_sys()?, guard, config_file).await
 }
 
 pub async fn update_quota_if_incarnation(
@@ -861,29 +790,6 @@ 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(
@@ -901,7 +807,7 @@ async fn update_quota_if_incarnation_stamped(
             achieved: 0,
         });
     }
-    update_under_config_write_guard(sys, &guard, rustfs_config::QUOTA_CONFIG_FILE, data, updated_at).await
+    update_under_config_write_guard(sys, &guard, rustfs_config::QUOTA_CONFIG_FILE, data).await
 }
 
 pub async fn update_bucket_targets_under_transaction_lock(
@@ -917,7 +823,6 @@ 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();
@@ -929,7 +834,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, updated_at),
+            metadata_sys.update_checked(&guard.bucket, config_file, data, true, guard.incarnation_id),
         ),
     )
     .await?;
@@ -941,7 +846,6 @@ 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();
@@ -953,7 +857,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, updated_at),
+            metadata_sys.update_checked(&guard.bucket, config_file, Vec::new(), false, guard.incarnation_id),
         ),
     )
     .await?;
@@ -1858,17 +1762,15 @@ 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, None)).await
+        Box::pin(self.update_checked(bucket, config_file, data, true, incarnation_id)).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, None)
+        self.update_checked(bucket, config_file, Vec::new(), false, incarnation_id)
             .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,
@@ -1876,7 +1778,6 @@ 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
@@ -1887,10 +1788,7 @@ impl BucketMetadataSys {
             return Err(Error::BucketNotFound(bucket.to_string()));
         }
 
-        let updated = match updated_at {
-            Some(updated_at) => bm.update_config_at(config_file, data, updated_at)?,
-            None => bm.update_config(config_file, data)?,
-        };
+        let updated = bm.update_config(config_file, data)?;
 
         Box::pin(self.save(bm)).await?;
 
@@ -3857,106 +3755,6 @@ 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.
@@ -4173,16 +3971,10 @@ 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),
-            None,
-        )
-        .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))
+                .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();
@@ -4217,7 +4009,7 @@ mod tests {
             }],
         })
         .unwrap();
-        update_under_config_write_guard(sys, &guard, BUCKET_TAGGING_CONFIG, tagging, None)
+        update_under_config_write_guard(sys, &guard, BUCKET_TAGGING_CONFIG, tagging)
             .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, 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,
+    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,
 };

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

@@ -882,20 +882,6 @@ 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
@@ -6615,87 +6601,4 @@ 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, 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,
+    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,
 };
 use super::replication_queue_boundary::{DeletedObjectReplicationInfo, ReplicationQueueAdmission};
 use super::replication_resync_boundary::ResyncStatusType;
@@ -2051,11 +2051,7 @@ pub(crate) async fn replicate_delete_with_outcome<S: ReplicationStorage>(
 
     let is_version_purge = is_version_delete_replication(&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 requires_delayed_purge = should_retry_delete_marker_purge(&dobj.delete_object);
 
     let (replication_status, prev_status) = if !is_version_purge {
         (
@@ -2765,6 +2761,12 @@ 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
@@ -2797,25 +2799,7 @@ async fn replicate_delete_to_target(dobj: &DeletedObjectReplicationInfo, tgt_cli
         return rinfo;
     }
 
-    // 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;
-    };
+    let version_id = target_delete_version_id(version_id, is_version_purge);
 
     if dobj.delete_object.delete_marker && dobj.delete_object.delete_marker_version_id.is_some() {
         match head_object_for_worker(

crates/heal/src/error.rs

@@ -54,6 +54,15 @@ pub enum Error {
     #[error("Heal task execution failed: {message}")]
     TaskExecutionFailed { message: String },
 
+    /// The current page already exhausted its local retry budget. Retrying
+    /// the enclosing bucket would replay pages whose results were counted.
+    #[error("Heal listing failed for bucket {bucket}: {source}")]
+    HealListingFailed {
+        bucket: String,
+        #[source]
+        source: Box<Error>,
+    },
+
     #[error("Invalid heal type: {heal_type}")]
     InvalidHealType { heal_type: String },
 

crates/heal/src/heal/channel.rs

@@ -447,6 +447,7 @@ impl HealChannelProcessor {
                 progress,
                 next_seq,
                 min_seq,
+                ..
             }) => (
                 "running".to_string(),
                 None,
@@ -463,6 +464,7 @@ impl HealChannelProcessor {
                 progress,
                 next_seq,
                 min_seq,
+                ..
             }) => (
                 "running".to_string(),
                 Some(format!("heal task retrying after recoverable failure, attempt {retry_attempt}: {error}")),
@@ -479,6 +481,7 @@ impl HealChannelProcessor {
                 progress,
                 next_seq,
                 min_seq,
+                ..
             }) => (
                 "finished".to_string(),
                 None,
@@ -495,6 +498,7 @@ impl HealChannelProcessor {
                 progress,
                 next_seq,
                 min_seq,
+                ..
             }) => (
                 "stopped".to_string(),
                 Some("heal task cancelled".to_string()),
@@ -511,6 +515,7 @@ impl HealChannelProcessor {
                 progress,
                 next_seq,
                 min_seq,
+                ..
             }) => (
                 "stopped".to_string(),
                 Some("heal task timed out".to_string()),
@@ -527,6 +532,7 @@ impl HealChannelProcessor {
                 progress,
                 next_seq,
                 min_seq,
+                ..
             }) => (
                 "stopped".to_string(),
                 Some(error),

crates/heal/src/heal/manager.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use crate::heal::{
+    outcome::HealTaskOutcome,
     progress::{HealProgress, HealStatistics},
     resume::{ReplacementPhase, ResumeGc, ResumeManager, ResumeState, ResumeUtils},
     storage::HealStorageAPI,
@@ -185,6 +186,7 @@ fn record_displaced_terminal(
     request: &HealRequest,
 ) -> Arc<CompletedHealStatus> {
     let terminal = Arc::new(CompletedHealStatus {
+        outcome: None,
         progress: None,
         retained_bytes: std::sync::OnceLock::new(),
         heal_type: request.heal_type.clone(),
@@ -268,6 +270,7 @@ async fn publish_completed_heal(
 
 #[derive(Debug, Clone)]
 pub struct HealTaskReport {
+    pub outcome: Option<Arc<HealTaskOutcome>>,
     pub status: HealTaskStatus,
     pub result_items: Vec<HealResultItem>,
     pub result_items_truncated: bool,
@@ -285,6 +288,7 @@ async fn active_task_report(task: &HealTask, since: Option<u64>) -> HealTaskRepo
     let window = task.get_result_items_since(since).await;
     HealTaskReport {
         status: task.get_status().await,
+        outcome: Some(Arc::new(task.get_outcome().await)),
         result_items: window.items,
         // The legacy flag stays set once anything was evicted; a lagging
         // incremental cursor additionally marks this response truncated so
@@ -298,6 +302,7 @@ async fn active_task_report(task: &HealTask, since: Option<u64>) -> HealTaskRepo
 
 fn empty_task_report(status: HealTaskStatus) -> HealTaskReport {
     HealTaskReport {
+        outcome: None,
         status,
         result_items: Vec::new(),
         result_items_truncated: false,
@@ -325,6 +330,7 @@ fn completed_task_report(completed: &CompletedHealStatus, since: Option<u64>) ->
     };
     HealTaskReport {
         status: completed.status.clone(),
+        outcome: completed.outcome.clone(),
         result_items,
         result_items_truncated: completed.result_items_truncated || lagged,
         progress: completed.progress.clone(),

crates/heal/src/heal/manager/queue.rs

@@ -83,6 +83,7 @@ pub(super) struct CompletedHealStatus {
     pub(super) heal_type: HealType,
     pub(super) status: HealTaskStatus,
     pub(super) progress: Option<HealProgress>,
+    pub(super) outcome: Option<Arc<HealTaskOutcome>>,
     pub(super) retained_bytes: std::sync::OnceLock<usize>,
     pub(super) result_items_truncated: bool,
     pub(super) completed_at: SystemTime,
@@ -105,6 +106,7 @@ impl CompletedHealStatus {
     fn measure_retained_bytes(&self) -> usize {
         let mut bytes = size_of::<Self>();
         let mut add = |amount: usize| bytes = bytes.saturating_add(amount);
+        add(self.outcome.as_ref().map_or(0, |outcome| outcome.retained_bytes()));
         match &self.heal_type {
             HealType::Cluster => {}
             HealType::Bucket { bucket } => add(bucket.capacity()),
@@ -209,6 +211,7 @@ impl CompletedHealStatus {
             heal_type: task.heal_type.clone(),
             status,
             progress: Some(task.get_progress().await),
+            outcome: Some(Arc::new(task.get_outcome().await)),
             retained_bytes: std::sync::OnceLock::new(),
             result_items_truncated: task.result_items_truncated(),
             completed_at: SystemTime::now(),

crates/heal/src/heal/manager/scheduler.rs

@@ -298,6 +298,7 @@ impl HealManager {
                     if cancelled_completion {
                         completed_status = HealTaskStatus::Cancelled;
                         completed_status_entry.status = HealTaskStatus::Cancelled;
+                        completed_status_entry.outcome = Some(Arc::new(task.get_outcome().await));
                     }
                     let terminal_completion = !matches!(completed_status, HealTaskStatus::Retrying { .. });
                     let successful_completion = matches!(completed_status, HealTaskStatus::Completed);

crates/heal/src/heal/manager/tests.rs

@@ -103,6 +103,7 @@ struct MockStorage;
 
 fn completed_retention_fixture(completed_at: SystemTime) -> CompletedHealStatus {
     CompletedHealStatus {
+        outcome: None,
         heal_type: HealType::Cluster,
         status: HealTaskStatus::Completed,
         progress: Some(HealProgress {
@@ -287,6 +288,59 @@ pub(super) async fn pause_completed_retention_before_publish(task_id: &str, stat
     }
 }
 
+#[tokio::test]
+async fn canonical_outcome_cancel_wins_before_worker_finalizes_success() {
+    use crate::heal::outcome::{HealAbortReason, HealExecutionOutcome};
+    use crate::heal::task::{OUTCOME_FINISH_TEST_HOOK, OutcomeFinishTestHook};
+    let bucket = "canonical-outcome-cancel-before-finish";
+    let manager = HealManager::new(Arc::new(MockStorage), None);
+    let request = HealRequest::object(bucket.to_string(), "object".to_string(), None);
+    let task_id = request.id.clone();
+    let duplicate = HealRequest::object(bucket.to_string(), "object".to_string(), None);
+    let alias = duplicate.id.clone();
+    let retention_hook = Arc::new(CompletedRetentionHook::default());
+    {
+        let mut hooks = COMPLETED_RETENTION_HOOKS.lock().await;
+        hooks.insert(bucket.to_string(), retention_hook.clone());
+        hooks.insert(task_id.clone(), retention_hook.clone());
+    }
+    let finish_hook = Arc::new(OutcomeFinishTestHook {
+        task_id: task_id.clone(),
+        reached: Notify::new(),
+        release: Notify::new(),
+    });
+    *OUTCOME_FINISH_TEST_HOOK.lock().await = Some(finish_hook.clone());
+    manager.submit_heal_request(request).await.expect("admit original");
+    manager.submit_heal_request(duplicate).await.expect("admit alias");
+    process_manager_queue_once(&manager).await;
+    tokio::time::timeout(Duration::from_secs(5), retention_hook.started.notified())
+        .await
+        .expect("storage started");
+    retention_hook.execute.notify_one();
+    tokio::time::timeout(Duration::from_secs(5), finish_hook.reached.notified())
+        .await
+        .expect("storage returned before outcome finalization");
+    manager.cancel_task(&alias).await.expect("cancel wins publication");
+    finish_hook.release.notify_one();
+    tokio::time::timeout(Duration::from_secs(5), retention_hook.handoff.notified())
+        .await
+        .expect("scheduler completes cancelled handoff");
+    for token in [&task_id, &alias] {
+        let report = manager.get_task_report(token).await.expect("cancelled token retained");
+        assert_eq!(report.status, HealTaskStatus::Cancelled);
+        assert_eq!(
+            report.outcome.as_ref().expect("frozen outcome").execution,
+            HealExecutionOutcome::Aborted(HealAbortReason::Cancelled)
+        );
+    }
+    retention_hook.finish.notify_one();
+    *OUTCOME_FINISH_TEST_HOOK.lock().await = None;
+    COMPLETED_RETENTION_HOOKS
+        .lock()
+        .await
+        .retain(|key, _| key != bucket && key != &task_id);
+}
+
 #[tokio::test]
 async fn completed_retention_cancel_wins_over_a_prepared_retry_snapshot() {
     let bucket = "completed-retention-retry-cancel";
@@ -325,6 +379,10 @@ async fn completed_retention_cancel_wins_over_a_prepared_retry_snapshot() {
     for token in [&task_id, &alias] {
         let report = manager.get_task_report(token).await.expect("cancelled token retained");
         assert_eq!(report.status, HealTaskStatus::Cancelled);
+        assert_eq!(
+            report.outcome.as_ref().expect("cancelled outcome retained").execution,
+            crate::heal::outcome::HealExecutionOutcome::Aborted(crate::heal::outcome::HealAbortReason::Cancelled)
+        );
         assert_eq!(report.progress.expect("frozen progress").objects_scanned, 1);
     }
     assert!(!manager.retrying_heals.lock().await.contains_key(&task_id));
@@ -391,6 +449,7 @@ async fn completed_retention_scheduler_preserves_progress_aliases_and_atomic_han
             .expect("scheduler archives terminal");
         assert!(!manager.active_heals.lock().await.contains_key(&task_id));
         let expected = task.get_progress().await;
+        let expected_outcome = task.get_outcome().await;
         for token in [&task_id, &alias] {
             assert_eq!(manager.get_task_progress(token).await.expect("terminal progress query"), expected);
             let report = manager
@@ -398,6 +457,7 @@ async fn completed_retention_scheduler_preserves_progress_aliases_and_atomic_han
                 .await
                 .expect("terminal token remains queryable at handoff");
             assert_eq!(report.progress.as_ref(), Some(&expected));
+            assert_eq!(report.outcome.as_deref(), Some(&expected_outcome));
             assert!(report.result_items.is_empty());
             match outcome {
                 "success" => assert_eq!(report.status, HealTaskStatus::Completed),
@@ -1976,6 +2036,7 @@ async fn insert_retrying_request(manager: &HealManager, request: HealRequest) ->
         task_id,
         Arc::new(CompletedHealStatus {
             progress: None,
+            outcome: None,
             retained_bytes: std::sync::OnceLock::new(),
             heal_type: request.heal_type,
             status: HealTaskStatus::Retrying {
@@ -2700,6 +2761,7 @@ async fn test_retrying_completion_outranks_the_queue_for_the_same_id() {
         task_id.clone(),
         Arc::new(CompletedHealStatus {
             progress: None,
+            outcome: None,
             retained_bytes: std::sync::OnceLock::new(),
             heal_type: request.heal_type.clone(),
             status: HealTaskStatus::Retrying {
@@ -2737,6 +2799,7 @@ async fn test_get_task_status_reads_recent_completed_status() {
         "completed-token".to_string(),
         Arc::new(CompletedHealStatus {
             progress: None,
+            outcome: None,
             retained_bytes: std::sync::OnceLock::new(),
             heal_type: HealType::Bucket {
                 bucket: "bucket".to_string(),
@@ -2768,6 +2831,7 @@ async fn test_get_task_report_for_path_reads_completed_items() {
         "completed-token".to_string(),
         Arc::new(CompletedHealStatus {
             progress: None,
+            outcome: None,
             retained_bytes: std::sync::OnceLock::new(),
             heal_type: HealType::Object {
                 bucket: "bucket".to_string(),

crates/heal/src/heal/mod.rs

@@ -16,6 +16,7 @@ pub mod channel;
 pub mod erasure_healer;
 pub mod manager;
 pub mod mrf_queue;
+pub mod outcome;
 pub mod progress;
 pub(crate) mod replacement_readiness;
 pub mod resume;

crates/heal/src/heal/outcome.rs

@@ -0,0 +1,305 @@
+// Copyright 2026 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Execution results are separate from repair responsibility. A legacy
+//! successful storage call supplies no authoritative repair receipt.
+
+use std::{collections::VecDeque, time::SystemTime};
+use uuid::Uuid;
+
+const MAX_OUTCOME_ITEMS: usize = 128;
+const MAX_OUTCOME_BYTES: usize = 64 * 1024;
+const MAX_OUTCOME_DETAIL_BYTES: usize = 1024;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum HealObjectKind {
+    Object,
+    Metadata,
+    Decode,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct HealObjectIdentity {
+    pub kind: HealObjectKind,
+    pub bucket: String,
+    pub object: String,
+    /// The requested version; None remains unresolved, never an absence proof.
+    pub version_id: Option<String>,
+    pub bucket_incarnation_id: Option<Uuid>,
+    pub pool_index: Option<usize>,
+    pub set_index: Option<usize>,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum HealDeferredReason {
+    DanglingDeleteGrace,
+    TransientUsageCache,
+    TransientExistenceCheck,
+    Deadline,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum HealFailureClass {
+    Recoverable,
+    RetryExhausted,
+    Permanent,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum HealObjectDisposition {
+    /// The legacy storage response does not prove the requested check or commit.
+    Unknown,
+    Repaired,
+    VerifiedHealthy,
+    AuthoritativelyAbsent,
+    Deferred {
+        reason: HealDeferredReason,
+        retry_not_before: Option<SystemTime>,
+    },
+    Failed(HealFailureClass),
+    Cancelled,
+    DryRunObserved,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct HealObjectOutcome {
+    pub identity: HealObjectIdentity,
+    pub disposition: HealObjectDisposition,
+    pub detail: Option<String>,
+}
+
+impl HealObjectOutcome {
+    fn retained_bytes(&self) -> usize {
+        size_of::<Self>()
+            .saturating_add(self.identity.bucket.capacity())
+            .saturating_add(self.identity.object.capacity())
+            .saturating_add(self.identity.version_id.as_ref().map_or(0, String::capacity))
+            .saturating_add(self.detail.as_ref().map_or(0, String::capacity))
+    }
+}
+
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+pub enum HealTraversalCoverage {
+    #[default]
+    Unknown,
+    Partial,
+    Complete,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum HealAbortReason {
+    Cancelled,
+    Deadline,
+    Untraversable,
+}
+
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+pub enum HealExecutionOutcome {
+    #[default]
+    Pending,
+    Running,
+    Completed,
+    CompletedWithErrors,
+    Aborted(HealAbortReason),
+}
+
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+pub struct HealOutcomeCounters {
+    pub processed: u64,
+    pub healed: u64,
+    pub unchanged: u64,
+    /// Deferred, cancelled, dry-run and unverified results remain unresolved.
+    pub skipped: u64,
+    pub failed: u64,
+    pub unknown: u64,
+    pub attempt_failures: u64,
+    pub overflowed: bool,
+}
+
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+pub struct HealTaskOutcome {
+    pub execution: HealExecutionOutcome,
+    pub coverage: HealTraversalCoverage,
+    pub counters: HealOutcomeCounters,
+    /// A bounded diagnostic window, not a complete responsibility ledger.
+    pub objects: VecDeque<HealObjectOutcome>,
+    pub objects_truncated: bool,
+    retained_object_bytes: usize,
+    untraversable: bool,
+}
+
+impl HealTaskOutcome {
+    pub(crate) fn start(&mut self) {
+        if self.execution != HealExecutionOutcome::Aborted(HealAbortReason::Cancelled) {
+            self.execution = HealExecutionOutcome::Running;
+        }
+        self.coverage = HealTraversalCoverage::Partial;
+    }
+
+    pub(crate) fn attempt_failed(&mut self) {
+        self.counters.overflowed |= !super::progress::increment_counter(&mut self.counters.attempt_failures);
+    }
+
+    pub(crate) fn mark_untraversable(&mut self) {
+        self.untraversable = true;
+        self.coverage = HealTraversalCoverage::Partial;
+    }
+
+    pub(crate) fn finish(&mut self, abort: Option<HealAbortReason>) {
+        if self.execution == HealExecutionOutcome::Aborted(HealAbortReason::Cancelled) {
+            return;
+        }
+        let abort = abort.or(self.untraversable.then_some(HealAbortReason::Untraversable));
+        self.execution = match abort {
+            Some(reason) => HealExecutionOutcome::Aborted(reason),
+            None if self.counters.failed > 0 => HealExecutionOutcome::CompletedWithErrors,
+            None => HealExecutionOutcome::Completed,
+        };
+        self.coverage = if abort.is_none() && !self.counters.overflowed {
+            HealTraversalCoverage::Complete
+        } else {
+            HealTraversalCoverage::Partial
+        };
+    }
+
+    pub(crate) fn record(&mut self, mut item: HealObjectOutcome) {
+        use super::progress::increment_counter;
+        let counters = &mut self.counters;
+        counters.overflowed |= !increment_counter(&mut counters.processed);
+        let counter = match item.disposition {
+            HealObjectDisposition::Repaired => &mut counters.healed,
+            HealObjectDisposition::VerifiedHealthy | HealObjectDisposition::AuthoritativelyAbsent => &mut counters.unchanged,
+            HealObjectDisposition::Failed(_) => &mut counters.failed,
+            HealObjectDisposition::Unknown => {
+                counters.overflowed |= !increment_counter(&mut counters.unknown);
+                &mut counters.skipped
+            }
+            _ => &mut counters.skipped,
+        };
+        counters.overflowed |= !increment_counter(counter);
+        if let Some(detail) = &mut item.detail {
+            let mut end = detail.len().min(MAX_OUTCOME_DETAIL_BYTES);
+            while !detail.is_char_boundary(end) {
+                end -= 1;
+            }
+            self.objects_truncated |= end < detail.len();
+            detail.truncate(end);
+            detail.shrink_to_fit();
+        }
+        let bytes = item.retained_bytes();
+        if bytes > MAX_OUTCOME_BYTES {
+            self.objects_truncated = true;
+            return;
+        }
+        while self.objects.len() >= MAX_OUTCOME_ITEMS || self.retained_object_bytes.saturating_add(bytes) > MAX_OUTCOME_BYTES {
+            let Some(oldest) = self.objects.pop_front() else { break };
+            self.retained_object_bytes = self.retained_object_bytes.saturating_sub(oldest.retained_bytes());
+            self.objects_truncated = true;
+        }
+        self.retained_object_bytes = self.retained_object_bytes.saturating_add(bytes);
+        self.objects.push_back(item);
+    }
+
+    pub(crate) fn retained_bytes(&self) -> usize {
+        size_of::<Self>()
+            .saturating_add(self.retained_object_bytes)
+            .saturating_add(self.objects.capacity().saturating_mul(size_of::<HealObjectOutcome>()))
+    }
+}
+
+#[cfg(test)]
+mod canonical_outcome_tests {
+    use super::*;
+
+    fn item(disposition: HealObjectDisposition) -> HealObjectOutcome {
+        HealObjectOutcome {
+            identity: HealObjectIdentity {
+                kind: HealObjectKind::Object,
+                bucket: "bucket".to_string(),
+                object: "object".to_string(),
+                version_id: None,
+                bucket_incarnation_id: None,
+                pool_index: None,
+                set_index: None,
+            },
+            disposition,
+            detail: None,
+        }
+    }
+
+    #[test]
+    fn canonical_outcome_categories_have_one_terminal_count() {
+        let mut outcome = HealTaskOutcome::default();
+        for disposition in [
+            HealObjectDisposition::Unknown,
+            HealObjectDisposition::Repaired,
+            HealObjectDisposition::VerifiedHealthy,
+            HealObjectDisposition::AuthoritativelyAbsent,
+            HealObjectDisposition::Deferred {
+                reason: HealDeferredReason::DanglingDeleteGrace,
+                retry_not_before: None,
+            },
+            HealObjectDisposition::Failed(HealFailureClass::Permanent),
+            HealObjectDisposition::Cancelled,
+            HealObjectDisposition::DryRunObserved,
+        ] {
+            outcome.record(item(disposition));
+        }
+        let c = &outcome.counters;
+        assert_eq!((c.processed, c.healed, c.unchanged, c.skipped, c.failed, c.unknown), (8, 1, 2, 4, 1, 1));
+        assert_eq!(c.processed, c.healed + c.unchanged + c.skipped + c.failed);
+    }
+
+    #[test]
+    fn canonical_outcome_window_count_bytes_and_oversize_keep_total_counts() {
+        let mut outcome = HealTaskOutcome::default();
+        for _ in 0..MAX_OUTCOME_ITEMS {
+            outcome.record(item(HealObjectDisposition::Unknown));
+        }
+        assert_eq!(outcome.objects.len(), MAX_OUTCOME_ITEMS);
+        assert!(!outcome.objects_truncated);
+        outcome.record(item(HealObjectDisposition::Unknown));
+        assert_eq!(outcome.objects.len(), MAX_OUTCOME_ITEMS);
+        assert!(outcome.objects_truncated);
+        let mut oversized = item(HealObjectDisposition::Failed(HealFailureClass::Permanent));
+        oversized.identity.object = "x".repeat(MAX_OUTCOME_BYTES);
+        outcome.record(oversized);
+        assert_eq!(outcome.counters.processed, u64::try_from(MAX_OUTCOME_ITEMS + 2).expect("bounded count"));
+        assert_eq!(outcome.counters.failed, 1);
+        assert!(outcome.retained_object_bytes <= MAX_OUTCOME_BYTES);
+        for _ in 0..MAX_OUTCOME_ITEMS {
+            let mut failed = item(HealObjectDisposition::Failed(HealFailureClass::Permanent));
+            failed.detail = Some("\u{4fee}".repeat(MAX_OUTCOME_DETAIL_BYTES));
+            outcome.record(failed);
+        }
+        assert!(outcome.retained_object_bytes <= MAX_OUTCOME_BYTES);
+        assert!(outcome.objects.iter().all(|item| {
+            item.detail
+                .as_ref()
+                .is_none_or(|detail| detail.len() <= MAX_OUTCOME_DETAIL_BYTES)
+        }));
+        assert!(outcome.objects.len() < MAX_OUTCOME_ITEMS);
+    }
+
+    #[test]
+    fn canonical_outcome_counter_overflow_cannot_claim_complete_coverage() {
+        let mut outcome = HealTaskOutcome::default();
+        outcome.counters.processed = u64::MAX;
+        outcome.record(item(HealObjectDisposition::Unknown));
+        outcome.finish(None);
+        assert!(outcome.counters.overflowed);
+        assert_eq!(outcome.counters.processed, u64::MAX);
+        assert_eq!(outcome.coverage, HealTraversalCoverage::Partial);
+    }
+}

crates/heal/src/heal/task.rs

@@ -15,6 +15,10 @@
 use crate::heal::{
     DiskError, EcstoreError, ErasureSetHealer, HealDiskExt as _,
     erasure_healer::target_outcomes_complete,
+    outcome::{
+        HealAbortReason, HealDeferredReason, HealFailureClass, HealObjectDisposition, HealObjectIdentity, HealObjectKind,
+        HealObjectOutcome, HealTaskOutcome,
+    },
     progress::HealProgress,
     resume::{
         CheckpointManager, ReplacementPhase, ReplacementTargetIdentity, ResumeManager, replacement_target_identities_match,
@@ -43,6 +47,26 @@ use uuid::Uuid;
 
 use super::{BUCKET_META_PREFIX, DATA_USAGE_CACHE_NAME, RUSTFS_META_BUCKET};
 
+#[cfg(test)]
+pub(crate) struct OutcomeFinishTestHook {
+    pub(crate) task_id: String,
+    pub(crate) reached: tokio::sync::Notify,
+    pub(crate) release: tokio::sync::Notify,
+}
+
+#[cfg(test)]
+pub(crate) static OUTCOME_FINISH_TEST_HOOK: std::sync::LazyLock<tokio::sync::Mutex<Option<Arc<OutcomeFinishTestHook>>>> =
+    std::sync::LazyLock::new(|| tokio::sync::Mutex::new(None));
+
+#[cfg(test)]
+async fn pause_outcome_finish(task_id: &str) {
+    let hook = OUTCOME_FINISH_TEST_HOOK.lock().await.clone();
+    if let Some(hook) = hook.filter(|hook| hook.task_id == task_id) {
+        hook.reached.notify_one();
+        hook.release.notified().await;
+    }
+}
+
 const LOG_COMPONENT_HEAL: &str = "heal";
 const LOG_SUBSYSTEM_TASK: &str = "task";
 const LOG_SUBSYSTEM_OBJECT: &str = "object";
@@ -394,6 +418,7 @@ pub struct HealTask {
     pub status: Arc<RwLock<HealTaskStatus>>,
     /// Progress tracking
     pub progress: Arc<RwLock<HealProgress>>,
+    outcome: Arc<RwLock<HealTaskOutcome>>,
     /// Result items collected from storage heal calls, each stamped with a
     /// monotonically increasing sequence number for incremental consumption
     /// (the client passes the last seen seq back and receives only newer
@@ -460,6 +485,7 @@ impl HealTask {
             result_items_truncated: Arc::new(AtomicBool::new(false)),
             batch_failure: Arc::new(RwLock::new(None)),
             batch_failure_recorded: Arc::new(AtomicBool::new(false)),
+            outcome: Arc::new(RwLock::new(HealTaskOutcome::default())),
             created_at: request.created_at,
             enqueued_at: request.enqueued_at,
             started_at: Arc::new(RwLock::new(None)),
@@ -507,6 +533,66 @@ impl HealTask {
         self.heal_type.kind_label()
     }
 
+    pub async fn get_outcome(&self) -> HealTaskOutcome {
+        self.outcome.read().await.clone()
+    }
+
+    fn outcome_identity(
+        &self,
+        bucket: &str,
+        object: &str,
+        version_id: Option<&str>,
+        pool_index: Option<usize>,
+        set_index: Option<usize>,
+    ) -> HealObjectIdentity {
+        HealObjectIdentity {
+            kind: match self.heal_type {
+                HealType::Metadata { .. } => HealObjectKind::Metadata,
+                HealType::ECDecode { .. } => HealObjectKind::Decode,
+                _ => HealObjectKind::Object,
+            },
+            bucket: bucket.to_owned(),
+            object: object.to_owned(),
+            version_id: version_id.map(ToOwned::to_owned),
+            bucket_incarnation_id: None,
+            pool_index,
+            set_index,
+        }
+    }
+
+    fn single_object_identity(&self) -> Option<HealObjectIdentity> {
+        let (bucket, object, version) = match &self.heal_type {
+            HealType::Object {
+                bucket,
+                object,
+                version_id,
+            }
+            | HealType::ECDecode {
+                bucket,
+                object,
+                version_id,
+            } => (bucket, object, version_id.as_deref()),
+            HealType::Metadata { bucket, object } => (bucket, object, None),
+            _ => return None,
+        };
+        Some(self.outcome_identity(bucket, object, version, self.options.pool_index, self.options.set_index))
+    }
+
+    async fn record_deferred_object(&self, reason: HealDeferredReason) {
+        if let Some(identity) = self.single_object_identity() {
+            let mut outcome = self.outcome.write().await;
+            outcome.attempt_failed();
+            outcome.record(HealObjectOutcome {
+                identity,
+                disposition: HealObjectDisposition::Deferred {
+                    reason,
+                    retry_not_before: None,
+                },
+                detail: None,
+            });
+        }
+    }
+
     pub(crate) fn has_batch_failure(&self) -> bool {
         self.batch_failure_recorded.load(Ordering::Acquire)
     }
@@ -634,6 +720,7 @@ impl HealTask {
     }
 
     async fn skip_due_to_transient_object_exists(&self, bucket: &str, object: &str, err: &Error) -> Result<()> {
+        self.record_deferred_object(HealDeferredReason::TransientExistenceCheck).await;
         warn!(
             target: "rustfs::heal::task",
             event = EVENT_HEAL_OBJECT_RESULT,
@@ -733,6 +820,8 @@ impl HealTask {
             return false;
         }
 
+        self.record_deferred_object(HealDeferredReason::TransientUsageCache).await;
+
         warn!(
             target: "rustfs::heal::task",
             event = EVENT_HEAL_OBJECT_RESULT,
@@ -755,6 +844,8 @@ impl HealTask {
             return false;
         }
 
+        self.record_deferred_object(HealDeferredReason::DanglingDeleteGrace).await;
+
         warn!(
             target: "rustfs::heal::task",
             event = EVENT_HEAL_OBJECT_RESULT,
@@ -801,6 +892,7 @@ impl HealTask {
     #[tracing::instrument(skip(self), fields(task_id = %self.id, heal_type = ?self.heal_type))]
     #[hotpath::measure]
     pub async fn execute(&self) -> Result<()> {
+        self.outcome.write().await.start();
         // update status and timestamps atomically to avoid race conditions
         let now = SystemTime::now();
         let start_instant = Instant::now();
@@ -860,6 +952,45 @@ impl HealTask {
             HealType::ErasureSet { buckets, set_disk_id } => self.heal_erasure_set(buckets.clone(), set_disk_id.clone()).await,
         };
 
+        #[cfg(test)]
+        pause_outcome_finish(&self.id).await;
+        {
+            let mut outcome = self.outcome.write().await;
+            if outcome.counters.processed == 0
+                && let Some(identity) = self.single_object_identity()
+            {
+                let disposition = match &result {
+                    Ok(()) if self.options.dry_run => HealObjectDisposition::DryRunObserved,
+                    Ok(()) => HealObjectDisposition::Unknown,
+                    Err(Error::TaskCancelled) => HealObjectDisposition::Cancelled,
+                    Err(Error::TaskTimeout) => HealObjectDisposition::Deferred {
+                        reason: HealDeferredReason::Deadline,
+                        retry_not_before: None,
+                    },
+                    Err(error) => {
+                        outcome.attempt_failed();
+                        HealObjectDisposition::Failed(if error.is_recoverable_heal() {
+                            HealFailureClass::Recoverable
+                        } else {
+                            HealFailureClass::Permanent
+                        })
+                    }
+                };
+                outcome.record(HealObjectOutcome {
+                    identity,
+                    disposition,
+                    detail: result.as_ref().err().map(ToString::to_string),
+                });
+            }
+            let abort = match &result {
+                Err(Error::TaskCancelled) => Some(HealAbortReason::Cancelled),
+                Err(Error::TaskTimeout) => Some(HealAbortReason::Deadline),
+                Err(_) if !self.has_batch_failure() && !self.heal_type.is_per_object() => Some(HealAbortReason::Untraversable),
+                _ => None,
+            };
+            outcome.finish(abort);
+        }
+
         // update completed time and status
         {
             let mut completed_at = self.completed_at.write().await;
@@ -944,6 +1075,7 @@ impl HealTask {
 
     pub async fn cancel(&self) -> Result<()> {
         self.cancel_token.cancel();
+        self.outcome.write().await.finish(Some(HealAbortReason::Cancelled));
         let mut status = self.status.write().await;
         *status = HealTaskStatus::Cancelled;
         debug!(

crates/heal/src/heal/task/heal_bucket.rs

@@ -214,6 +214,7 @@ impl HealTask {
                             continue;
                         }
                         failed = failed.saturating_add(1);
+                        self.outcome.write().await.mark_untraversable();
                         if err.is_recoverable_heal() {
                             retryable = retryable.saturating_add(1);
                         } else {
@@ -260,6 +261,7 @@ impl HealTask {
 
     #[hotpath::measure]
     async fn heal_bucket_objects(&self, bucket: &str, prefix: &str) -> Result<()> {
+        let previous_progress = self.get_progress().await;
         let mut scanned = 0u64;
         let mut healed = 0u64;
         let mut failed = 0u64;
@@ -304,23 +306,47 @@ impl HealTask {
             let mut continuation_token: Option<String> = None;
             loop {
                 self.check_control_flags().await?;
-                let (objects, next_token, is_truncated) = if let Some(set_disk_id) = set_disk_id.as_deref() {
-                    self.await_with_control(self.storage.list_versions_for_heal_page_disk_walk(
-                        set_disk_id,
-                        bucket,
-                        prefix,
-                        continuation_token.as_deref(),
-                        false,
-                    ))
-                    .await?
-                } else {
-                    self.await_with_control(self.storage.list_objects_for_heal_page(
-                        bucket,
-                        prefix,
-                        continuation_token.as_deref(),
-                        false,
-                    ))
-                    .await?
+                let mut listing_attempt = 0;
+                let (objects, next_token, is_truncated) = loop {
+                    let page = if let Some(set_disk_id) = set_disk_id.as_deref() {
+                        self.await_with_control(self.storage.list_versions_for_heal_page_disk_walk(
+                            set_disk_id,
+                            bucket,
+                            prefix,
+                            continuation_token.as_deref(),
+                            false,
+                        ))
+                        .await
+                    } else {
+                        self.await_with_control(self.storage.list_objects_for_heal_page(
+                            bucket,
+                            prefix,
+                            continuation_token.as_deref(),
+                            false,
+                        ))
+                        .await
+                    };
+                    match page {
+                        Ok(page) => break page,
+                        Err(error @ (Error::TaskCancelled | Error::TaskTimeout)) => return Err(error),
+                        Err(error) => {
+                            self.outcome.write().await.attempt_failed();
+                            if error.is_recoverable_heal() && listing_attempt < MAX_BUCKET_OBJECT_HEAL_RETRIES {
+                                listing_attempt += 1;
+                                self.await_with_control(async {
+                                    tokio::time::sleep(self.bucket_object_retry_delay(listing_attempt)).await;
+                                    Ok(())
+                                })
+                                .await?;
+                                continue;
+                            }
+                            self.outcome.write().await.mark_untraversable();
+                            return Err(Error::HealListingFailed {
+                                bucket: bucket.to_string(),
+                                source: Box::new(error),
+                            });
+                        }
+                    }
                 };
 
                 let mut pending = objects;
@@ -338,6 +364,14 @@ impl HealTask {
                         self.check_control_flags().await?;
                         let mut telemetry_unknown = false;
                         let object = item.name.as_str();
+                        let identity =
+                            self.outcome_identity(bucket, object, item.version_id.as_deref(), heal_opts.pool, heal_opts.set);
+                        let mut disposition = if heal_opts.dry_run {
+                            HealObjectDisposition::DryRunObserved
+                        } else {
+                            HealObjectDisposition::Unknown
+                        };
+                        let mut detail = None;
                         {
                             let mut progress = self.progress.write().await;
                             progress.set_current_object(Some(format!("{bucket}/{object}")));
@@ -380,7 +414,31 @@ impl HealTask {
                         };
 
                         if let Some(err) = error {
+                            match err {
+                                Error::TaskCancelled | Error::TaskTimeout => {
+                                    let disposition = if matches!(err, Error::TaskCancelled) {
+                                        HealObjectDisposition::Cancelled
+                                    } else {
+                                        HealObjectDisposition::Deferred {
+                                            reason: HealDeferredReason::Deadline,
+                                            retry_not_before: None,
+                                        }
+                                    };
+                                    self.outcome.write().await.record(HealObjectOutcome {
+                                        identity,
+                                        disposition,
+                                        detail: None,
+                                    });
+                                    return Err(err);
+                                }
+                                _ => self.outcome.write().await.attempt_failed(),
+                            }
+                            detail = Some(err.to_string());
                             if Self::is_dangling_delete_grace_error(&err) {
+                                disposition = HealObjectDisposition::Deferred {
+                                    reason: HealDeferredReason::DanglingDeleteGrace,
+                                    retry_not_before: None,
+                                };
                                 telemetry_unknown |= !increment_counter(&mut skipped);
                                 warn!(
                                     target: "rustfs::heal::task",
@@ -395,6 +453,10 @@ impl HealTask {
                                     "Heal bucket object dangling cleanup deferred by grace window"
                                 );
                             } else if Self::should_skip_data_usage_cache_heal_error(bucket, object, &err) {
+                                disposition = HealObjectDisposition::Deferred {
+                                    reason: HealDeferredReason::TransientUsageCache,
+                                    retry_not_before: None,
+                                };
                                 telemetry_unknown |= !increment_counter(&mut skipped);
                                 warn!(
                                     target: "rustfs::heal::task",
@@ -425,6 +487,11 @@ impl HealTask {
                                 );
                                 retry.push(item);
                             } else {
+                                disposition = HealObjectDisposition::Failed(if err.is_recoverable_heal() {
+                                    HealFailureClass::RetryExhausted
+                                } else {
+                                    HealFailureClass::Permanent
+                                });
                                 telemetry_unknown |= !increment_counter(&mut failed);
                                 if err.is_recoverable_heal() {
                                     retryable_failed = retryable_failed.saturating_add(1);
@@ -459,8 +526,20 @@ impl HealTask {
                             continue;
                         }
 
+                        self.outcome.write().await.record(HealObjectOutcome {
+                            identity,
+                            disposition,
+                            detail,
+                        });
+
                         let mut progress = self.progress.write().await;
-                        progress.update_object_progress(scanned, healed, failed, skipped, bytes);
+                        progress.update_object_progress(
+                            previous_progress.objects_scanned.saturating_add(scanned),
+                            previous_progress.objects_healed.saturating_add(healed),
+                            previous_progress.objects_failed.saturating_add(failed),
+                            previous_progress.skipped_objects.saturating_add(skipped),
+                            previous_progress.bytes_processed.saturating_add(bytes),
+                        );
                         if telemetry_unknown {
                             progress.mark_unknown();
                         }
@@ -475,7 +554,7 @@ impl HealTask {
 
                 continuation_token = next_heal_listing_token(bucket, prefix, next_token, is_truncated)?;
                 if continuation_token.is_none() {
-                    // Truncated but no continuation token: end of listing.
+                    // Truncated without a continuation token is a compatibility EOF.
                     break;
                 }
             }

crates/heal/src/heal/task/heal_metadata.rs

@@ -261,8 +261,8 @@ impl HealTask {
             update_parity: true,
             no_lock: self.options.no_lock,
             read_repair: false,
-            pool: None,
-            set: None,
+            pool: self.options.pool_index,
+            set: self.options.set_index,
         };
 
         let heal_result = self

crates/heal/src/heal/task/tests.rs

@@ -14,6 +14,364 @@
 
 use super::super::{DiskOption, DiskStore, Endpoint, new_disk};
 use super::*;
+
+mod canonical_outcome {
+    use super::*;
+    use crate::heal::outcome::{HealExecutionOutcome, HealTraversalCoverage};
+
+    fn bucket_task(storage: Arc<MockStorage>) -> HealTask {
+        HealTask::from_request(
+            HealRequest::new(
+                HealType::Bucket {
+                    bucket: "bucket-a".to_string(),
+                },
+                HealOptions {
+                    recursive: true,
+                    timeout: None,
+                    ..Default::default()
+                },
+                HealPriority::Normal,
+            ),
+            storage,
+        )
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn cluster_retries_only_the_failed_listing_page() {
+        let storage = Arc::new(MockStorage {
+            recoverable_second_page_failures: Mutex::new(Some(1)),
+            ..Default::default()
+        });
+        let task = HealTask::from_request(
+            HealRequest::new(
+                HealType::Cluster,
+                HealOptions {
+                    recursive: true,
+                    timeout: None,
+                    ..Default::default()
+                },
+                HealPriority::Normal,
+            ),
+            storage.clone(),
+        );
+        task.execute().await.expect("second-page retry succeeds");
+        let outcome = task.get_outcome().await;
+        assert_eq!(outcome.execution, HealExecutionOutcome::Completed);
+        assert_eq!(outcome.coverage, HealTraversalCoverage::Complete);
+        assert_eq!(outcome.counters.processed, 2);
+        assert_eq!(outcome.counters.attempt_failures, 1);
+        assert_eq!(task.get_progress().await.objects_scanned, 2);
+        assert_eq!(
+            storage.heal_object_calls.lock().expect("object calls").as_slice(),
+            ["object-a", "object-b"]
+        );
+        assert_eq!(
+            storage.listing_tokens.lock().expect("listing tokens").as_slice(),
+            [None, Some("second".to_string()), Some("second".to_string())]
+        );
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn exhausted_listing_page_cannot_restart_the_bucket() {
+        let storage = Arc::new(MockStorage {
+            recoverable_second_page_failures: Mutex::new(Some(4)),
+            ..Default::default()
+        });
+        let task = HealTask::from_request(
+            HealRequest::new(
+                HealType::Cluster,
+                HealOptions {
+                    recursive: true,
+                    timeout: None,
+                    ..Default::default()
+                },
+                HealPriority::Normal,
+            ),
+            storage.clone(),
+        );
+        task.execute().await.expect_err("listing page budget exhausted");
+        let outcome = task.get_outcome().await;
+        assert_eq!(outcome.execution, HealExecutionOutcome::Aborted(HealAbortReason::Untraversable));
+        assert_eq!(outcome.coverage, HealTraversalCoverage::Partial);
+        assert_eq!(outcome.counters.processed, 1);
+        assert_eq!(outcome.counters.attempt_failures, 4);
+        assert_eq!(task.get_progress().await.objects_scanned, 1);
+        assert_eq!(storage.heal_object_calls.lock().expect("object calls").as_slice(), ["object-a"]);
+        assert_eq!(storage.bucket_heal_calls.lock().expect("bucket calls").as_slice(), ["bucket-a"]);
+    }
+
+    #[tokio::test]
+    async fn listing_failure_preserves_processed_objects_and_partial_coverage() {
+        let storage = Arc::new(MockStorage {
+            fail_second_listing_page: true,
+            ..Default::default()
+        });
+        let task = bucket_task(storage);
+        task.execute().await.expect_err("second page cannot be traversed");
+        let outcome = task.get_outcome().await;
+        assert_eq!(outcome.execution, HealExecutionOutcome::Aborted(HealAbortReason::Untraversable));
+        assert_eq!(outcome.coverage, HealTraversalCoverage::Partial);
+        assert_eq!(outcome.counters.processed, 1);
+        assert_eq!(outcome.objects[0].identity.object, "object-a");
+        assert_eq!(task.get_progress().await.objects_scanned, 1);
+    }
+
+    #[tokio::test]
+    async fn cluster_preserves_cumulative_progress_across_buckets() {
+        let storage = Arc::new(MockStorage {
+            list_each_bucket: true,
+            listed_buckets: Mutex::new(Some(vec!["bucket-a".to_string(), "bucket-b".to_string()])),
+            ..Default::default()
+        });
+        let task = HealTask::from_request(
+            HealRequest::new(
+                HealType::Cluster,
+                HealOptions {
+                    recursive: true,
+                    timeout: None,
+                    ..Default::default()
+                },
+                HealPriority::Normal,
+            ),
+            storage,
+        );
+        task.execute().await.expect("both buckets complete");
+        let outcome = task.get_outcome().await;
+        assert_eq!(outcome.counters.processed, 4);
+        assert_eq!(outcome.coverage, HealTraversalCoverage::Complete);
+        let progress = task.get_progress().await;
+        assert_eq!((progress.objects_scanned, progress.objects_healed), (4, 4));
+        assert_eq!(
+            outcome
+                .objects
+                .iter()
+                .filter(|item| item.identity.bucket == "bucket-b")
+                .count(),
+            2
+        );
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn exhausted_object_does_not_abort_other_objects_or_erase_counts() {
+        let storage = Arc::new(MockStorage::default());
+        storage.heal_object_outcomes.lock().expect("outcomes").insert(
+            "object-a".to_string(),
+            (0..4).map(|_| MockHealObjectOutcome::RetryableReadQuorum).collect(),
+        );
+        let task = bucket_task(storage.clone());
+        task.execute().await.expect_err("legacy adapter retains batch failure");
+        let outcome = task.get_outcome().await;
+        assert_eq!(outcome.execution, HealExecutionOutcome::CompletedWithErrors);
+        assert_eq!(outcome.coverage, HealTraversalCoverage::Complete);
+        assert_eq!((outcome.counters.processed, outcome.counters.failed, outcome.counters.unknown), (2, 1, 1));
+        assert_eq!(outcome.counters.attempt_failures, 4);
+        let failed = outcome
+            .objects
+            .iter()
+            .find(|item| item.identity.object == "object-a")
+            .expect("failed object");
+        assert_eq!(failed.disposition, HealObjectDisposition::Failed(HealFailureClass::RetryExhausted));
+        let object_b_calls = {
+            let calls = storage.heal_object_calls.lock().expect("calls");
+            calls.iter().filter(|object| object.as_str() == "object-b").count()
+        };
+        assert_eq!(object_b_calls, 1);
+        let progress = task.get_progress().await;
+        assert_eq!((progress.objects_scanned, progress.objects_healed, progress.objects_failed), (2, 1, 1));
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn retry_success_counts_one_terminal_outcome() {
+        let storage = Arc::new(MockStorage::default());
+        storage
+            .heal_object_outcomes
+            .lock()
+            .expect("outcomes")
+            .insert("object-a".to_string(), VecDeque::from([MockHealObjectOutcome::RetryableReadQuorum]));
+        let task = bucket_task(storage);
+        task.execute().await.expect("retry should recover");
+        let outcome = task.get_outcome().await;
+        assert_eq!(outcome.execution, HealExecutionOutcome::Completed);
+        assert_eq!(outcome.counters.processed, 2);
+        assert_eq!(outcome.counters.failed, 0);
+        assert_eq!(outcome.counters.attempt_failures, 1);
+        assert_eq!(
+            outcome
+                .objects
+                .iter()
+                .filter(|item| item.identity.object == "object-a")
+                .count(),
+            1
+        );
+        assert_eq!(
+            outcome.counters.processed,
+            outcome.counters.healed + outcome.counters.unchanged + outcome.counters.skipped + outcome.counters.failed
+        );
+    }
+
+    #[tokio::test]
+    async fn mixed_grace_and_legacy_success_keep_distinct_dispositions() {
+        let storage = Arc::new(MockStorage::default());
+        storage
+            .heal_object_outcomes
+            .lock()
+            .expect("outcomes")
+            .insert("object-a".to_string(), VecDeque::from([MockHealObjectOutcome::DanglingGraceDeferred]));
+        let task = bucket_task(storage);
+        task.execute().await.expect("grace permits traversal completion");
+        let outcome = task.get_outcome().await;
+        assert_eq!(outcome.coverage, HealTraversalCoverage::Complete);
+        assert_eq!(outcome.counters.processed, 2);
+        assert_eq!(outcome.counters.healed, 0, "legacy result is not a repair receipt");
+        assert!(matches!(
+            outcome.objects[0].disposition,
+            HealObjectDisposition::Deferred {
+                reason: HealDeferredReason::DanglingDeleteGrace,
+                ..
+            }
+        ));
+        assert_eq!(outcome.objects[1].disposition, HealObjectDisposition::Unknown);
+        assert!(
+            outcome
+                .objects
+                .iter()
+                .all(|item| item.identity.bucket_incarnation_id.is_none())
+        );
+        assert_eq!(
+            task.get_progress().await.objects_healed,
+            1,
+            "legacy display count remains distinct from proof"
+        );
+    }
+
+    #[tokio::test]
+    async fn grace_single_object_is_completed_but_deferred() {
+        let storage = Arc::new(MockStorage {
+            heal_object_outcome: Mutex::new(Some(MockHealObjectOutcome::DanglingGraceDeferred)),
+            ..Default::default()
+        });
+        let task = HealTask::from_request(HealRequest::object("bucket-a".to_string(), "recent.txt".to_string(), None), storage);
+        task.execute().await.expect("grace is deferred");
+        let outcome = task.get_outcome().await;
+        assert_eq!(task.get_status().await, HealTaskStatus::Completed);
+        assert_eq!(outcome.counters.processed, 1);
+        assert!(matches!(
+            outcome.objects[0].disposition,
+            HealObjectDisposition::Deferred {
+                reason: HealDeferredReason::DanglingDeleteGrace,
+                ..
+            }
+        ));
+        assert_eq!(outcome.counters.attempt_failures, 1);
+    }
+
+    #[tokio::test]
+    async fn dry_run_and_transient_existence_do_not_prove_repair() {
+        for transient in [false, true] {
+            let storage = Arc::new(MockStorage::default());
+            if transient {
+                storage
+                    .object_exists_by_name
+                    .lock()
+                    .expect("existence fixture")
+                    .insert("object".to_string(), MockObjectExists::TransientSkip("retry later"));
+            }
+            let mut request = HealRequest::object("bucket-a".to_string(), "object".to_string(), None);
+            request.options.dry_run = !transient;
+            let task = HealTask::from_request(request, storage);
+            task.execute().await.expect("observation may complete");
+            let outcome = task.get_outcome().await;
+            assert_eq!(outcome.counters.healed, 0);
+            if transient {
+                assert!(matches!(
+                    outcome.objects[0].disposition,
+                    HealObjectDisposition::Deferred {
+                        reason: HealDeferredReason::TransientExistenceCheck,
+                        ..
+                    }
+                ));
+            } else {
+                assert_eq!(outcome.objects[0].disposition, HealObjectDisposition::DryRunObserved);
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn untraversable_bucket_does_not_claim_complete_cluster_coverage() {
+        let storage = Arc::new(MockStorage {
+            listed_buckets: Mutex::new(Some(vec!["bucket-a".to_string(), "bucket-b".to_string()])),
+            bucket_heal_errors: Mutex::new(HashMap::from([("bucket-a".to_string(), VecDeque::from(["metadata unavailable"]))])),
+            ..Default::default()
+        });
+        let task = HealTask::from_request(
+            HealRequest::new(
+                HealType::Cluster,
+                HealOptions {
+                    recursive: true,
+                    timeout: None,
+                    ..Default::default()
+                },
+                HealPriority::Normal,
+            ),
+            storage.clone(),
+        );
+        task.execute().await.expect_err("structural bucket error");
+        let outcome = task.get_outcome().await;
+        assert_eq!(outcome.execution, HealExecutionOutcome::Aborted(HealAbortReason::Untraversable));
+        assert_eq!(outcome.coverage, HealTraversalCoverage::Partial);
+        assert_eq!(
+            storage.bucket_heal_calls.lock().expect("bucket calls").as_slice(),
+            ["bucket-a", "bucket-b"]
+        );
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn cancellation_and_deadline_leave_partial_coverage() {
+        for cancel in [false, true] {
+            let storage = Arc::new(MockStorage {
+                block_heal_object: Mutex::new(true),
+                ..Default::default()
+            });
+            let mut request = HealRequest::object("bucket-a".to_string(), "object".to_string(), None);
+            request.options.timeout = Some(Duration::from_secs(1));
+            let task = HealTask::from_request(request, storage);
+            if cancel {
+                task.cancel().await.expect("cancel request");
+            }
+            task.execute().await.expect_err("control interruption");
+            let outcome = task.get_outcome().await;
+            assert_eq!(outcome.coverage, HealTraversalCoverage::Partial);
+            assert_eq!(
+                outcome.execution,
+                HealExecutionOutcome::Aborted(if cancel {
+                    HealAbortReason::Cancelled
+                } else {
+                    HealAbortReason::Deadline
+                })
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn decode_keeps_the_requested_pool_and_set() {
+        let storage = Arc::new(MockStorage::default());
+        let mut request = HealRequest::ec_decode("bucket-a".to_string(), "object".to_string(), Some("version-a".to_string()));
+        request.options.pool_index = Some(2);
+        request.options.set_index = Some(3);
+        let task = HealTask::from_request(request, storage.clone());
+        task.execute().await.expect("decode fixture");
+        let pool_and_set = {
+            let options = storage.object_heal_opts.lock().expect("storage options");
+            (options[0].pool, options[0].set)
+        };
+        assert_eq!(pool_and_set, (Some(2), Some(3)));
+        let outcome = task.get_outcome().await;
+        let identity = &outcome.objects[0].identity;
+        assert_eq!((identity.pool_index, identity.set_index), (Some(2), Some(3)));
+        assert_eq!(identity.version_id.as_deref(), Some("version-a"));
+        assert_eq!(outcome.objects[0].disposition, HealObjectDisposition::Unknown);
+    }
+}
 use crate::heal::storage::{HealListItem, HealObjectInfo};
 use rustfs_common::trace_bus::{TraceEvent, TraceFunc, TraceKind, TraceSubscription, TraceVal, subscribe_trace_events};
 use rustfs_madmin::heal_commands::{HealDriveInfo, HealResultItem, Infos};
@@ -582,6 +940,10 @@ async fn verified_recovery_keeps_state_when_marker_clear_fails() {
 #[derive(Default)]
 struct MockStorage {
     listed: Mutex<bool>,
+    list_each_bucket: bool,
+    fail_second_listing_page: bool,
+    recoverable_second_page_failures: Mutex<Option<usize>>,
+    listing_tokens: Mutex<Vec<Option<String>>>,
     healed_objects: Mutex<Vec<String>>,
     heal_object_calls: Mutex<Vec<String>>,
     heal_object_version_ids: Mutex<Vec<Option<String>>>,
@@ -995,12 +1357,41 @@ impl HealStorageAPI for MockStorage {
         _include_lifecycle_object_info: bool,
     ) -> Result<(Vec<HealListItem>, Option<String>, bool)> {
         self.listed_prefixes.lock().unwrap().push(prefix.to_string());
+        self.listing_tokens
+            .lock()
+            .expect("listing tokens")
+            .push(continuation_token.map(ToOwned::to_owned));
+        if let Some(remaining) = self
+            .recoverable_second_page_failures
+            .lock()
+            .expect("listing failures")
+            .as_mut()
+        {
+            if continuation_token.is_none() {
+                return Ok((vec![heal_item("object-a")], Some("second".to_string()), true));
+            }
+            if *remaining > 0 {
+                *remaining -= 1;
+                return Err(Error::Storage(EcstoreError::InsufficientReadQuorum(
+                    bucket.to_string(),
+                    "page".to_string(),
+                )));
+            }
+            return Ok((vec![heal_item("object-b")], None, false));
+        }
+        if self.fail_second_listing_page {
+            return if continuation_token.is_none() {
+                Ok((vec![heal_item("object-a")], Some("next-page".to_string()), true))
+            } else {
+                Err(Error::other("listing unavailable"))
+            };
+        }
         if *self.truncate_without_token.lock().unwrap() {
             return Ok((vec![heal_item("object-a")], None, true));
         }
 
         let mut listed = self.listed.lock().unwrap();
-        if continuation_token.is_none() && !*listed {
+        if continuation_token.is_none() && (!*listed || self.list_each_bucket) {
             *listed = true;
             let objects = if bucket == RUSTFS_META_BUCKET {
                 vec![
@@ -1393,6 +1784,8 @@ async fn test_recursive_bucket_heal_skips_object_dir_candidates() {
 
 #[tokio::test]
 async fn test_recursive_bucket_heal_treats_missing_continuation_token_as_end() {
+    use crate::heal::outcome::{HealExecutionOutcome, HealTraversalCoverage};
+
     // A version listing can report the final page as truncated with no
     // continuation token. That is treated as end-of-listing (not an error),
     // so the returned page is healed and the pass terminates cleanly instead
@@ -1414,10 +1807,16 @@ async fn test_recursive_bucket_heal_treats_missing_continuation_token_as_end() {
     );
     let task = HealTask::from_request(request, storage.clone());
 
-    task.heal_bucket("bucket-a")
+    task.execute()
         .await
         .expect("truncated-without-token must terminate cleanly, not loop or error");
 
+    assert_eq!(task.get_status().await, HealTaskStatus::Completed);
+    let outcome = task.get_outcome().await;
+    assert_eq!(outcome.execution, HealExecutionOutcome::Completed);
+    assert_eq!(outcome.coverage, HealTraversalCoverage::Complete);
+    assert_eq!(outcome.counters.processed, 1);
+
     assert_eq!(
         storage.healed_objects.lock().unwrap().as_slice(),
         ["object-a".to_string()],

crates/iam/src/manager.rs

@@ -429,27 +429,6 @@ 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);
@@ -555,17 +534,6 @@ 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);
         }
@@ -576,17 +544,18 @@ where
             .get(name)
             .map(|v| {
                 let mut p = v.clone();
-                p.update_at(policy.clone(), updated_at);
+                p.update(policy.clone());
                 p
             })
-            .unwrap_or_else(|| PolicyDoc::new_at(policy, updated_at));
+            .unwrap_or_else(|| PolicyDoc::new(policy));
 
         self.api.save_policy_doc(name, policy_doc.clone()).await?;
 
-        self.cache
-            .add_or_update_policy_doc(name, &policy_doc, OffsetDateTime::now_utc());
+        let now = OffsetDateTime::now_utc();
 
-        Ok(updated_at)
+        self.cache.add_or_update_policy_doc(name, &policy_doc, now);
+
+        Ok(now)
     }
 
     pub async fn list_policies(&self, bucket_name: &str) -> Result<HashMap<String, Policy>> {
@@ -820,12 +789,6 @@ 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);
         }
@@ -837,8 +800,7 @@ where
         }
         drop(cache);
 
-        let mut u = UserIdentity::new(cred);
-        u.update_at = Some(updated_at);
+        let u = UserIdentity::new(cred);
 
         self.api
             .save_user_identity(&u.credentials.access_key, UserType::Svc, u.clone(), None)
@@ -846,22 +808,10 @@ where
 
         self.update_user_with_claims(&u.credentials.access_key, u.clone())?;
 
-        Ok(updated_at)
+        Ok(OffsetDateTime::now_utc())
     }
 
     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 {
@@ -908,7 +858,13 @@ where
         }
 
         if let Some(status) = opts.status {
-            cr.status = account_status_flag(&status).to_owned();
+            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(),
+            }
         }
 
         let mut m: HashMap<String, Value> = if token_without_expiration {
@@ -960,8 +916,8 @@ where
 
         cr.session_token = jwt_sign(&m, &cr.secret_key)?;
 
-        let mut u = UserIdentity::new(cr);
-        u.update_at = Some(updated_at);
+        let u = UserIdentity::new(cr);
+        let updated_at = u.update_at.unwrap_or_else(OffsetDateTime::now_utc);
         self.api
             .save_user_identity(&u.credentials.access_key, UserType::Svc, u.clone(), None)
             .await?;
@@ -1193,20 +1149,6 @@ 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);
         }
@@ -1226,11 +1168,10 @@ where
                 self.cache.delete_user_policy(name, OffsetDateTime::now_utc());
             }
 
-            return Ok(updated_at);
+            return Ok(OffsetDateTime::now_utc());
         }
 
-        let mut mp = MappedPolicy::new(policy);
-        mp.update_at = updated_at;
+        let mp = MappedPolicy::new(policy);
 
         let cache = self.cache.snapshot();
         let policy_docs_cache = Arc::clone(&cache.policy_docs);
@@ -1253,7 +1194,7 @@ where
             self.cache.add_or_update_user_policy(name, &mp, OffsetDateTime::now_utc());
         }
 
-        Ok(updated_at)
+        Ok(OffsetDateTime::now_utc())
     }
 
     pub async fn set_temp_user(&self, access_key: &str, cred: &Credentials, policy_name: Option<&str>) -> Result<OffsetDateTime> {
@@ -1450,17 +1391,6 @@ 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) {
@@ -1478,13 +1408,12 @@ where
                 _ => auth::ACCOUNT_OFF,
             }
         };
-        let mut user_entry = UserIdentity::from(Credentials {
+        let 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)
@@ -1492,7 +1421,7 @@ where
 
         self.update_user_with_claims(access_key, user_entry)?;
 
-        Ok(updated_at)
+        Ok(OffsetDateTime::now_utc())
     }
 
     pub async fn delete_user(&self, access_key: &str, utype: UserType) -> Result<()> {
@@ -1670,17 +1599,6 @@ 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);
         }
@@ -1707,13 +1625,12 @@ where
             }
         };
 
-        let mut user_entry = UserIdentity::from(Credentials {
+        let 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);
 
@@ -1723,7 +1640,7 @@ where
 
         self.update_user_with_claims(access_key, user_entry)?;
 
-        Ok(updated_at)
+        Ok(OffsetDateTime::now_utc())
     }
 
     fn update_user_with_claims(&self, k: &str, u: UserIdentity) -> Result<()> {
@@ -1759,17 +1676,6 @@ 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);
         }
@@ -1787,10 +1693,6 @@ where
             }
         }
 
-        // The group's own timestamp moves with every membership or status
-        // change: site replication judges an incoming group item against it
-        // (backlog#2291), so it must reflect the last change, not creation.
-        let now = updated_at;
         let gi = match cache.groups.get(group) {
             Some(res) => {
                 let mut gi = res.clone();
@@ -1799,20 +1701,16 @@ where
                 uniq_set.extend(members.iter().cloned());
 
                 gi.members = uniq_set.into_iter().collect();
-                gi.update_at = Some(now);
-                gi
-            }
-            None => {
-                let mut gi = GroupInfo::new(members.clone());
-                gi.update_at = Some(now);
                 gi
             }
+            None => GroupInfo::new(members.clone()),
         };
         drop(cache);
 
         self.api.save_group_info(group, gi.clone()).await?;
 
-        self.cache.with_write_lock(|cache| {
+        let now = self.cache.with_write_lock(|cache| {
+            let now = OffsetDateTime::now_utc();
             cache.add_or_update_group(group, &gi, now);
 
             let user_group_memberships = Arc::clone(&cache.state().user_group_memberships);
@@ -1821,18 +1719,13 @@ where
                 m.insert(group.to_string());
                 cache.add_or_update_user_group_membership(member, &m, now);
             });
+            now
         });
 
         Ok(now)
     }
 
     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);
         }
@@ -1850,14 +1743,12 @@ where
         } else {
             gi.status = STATUS_DISABLED.to_owned();
         }
-        let now = updated_at;
-        gi.update_at = Some(now);
 
         self.api.save_group_info(name, gi.clone()).await?;
 
-        self.cache.add_or_update_group(name, &gi, now);
+        self.cache.add_or_update_group(name, &gi, OffsetDateTime::now_utc());
 
-        Ok(now)
+        Ok(OffsetDateTime::now_utc())
     }
 
     pub async fn get_group_description(&self, name: &str) -> Result<GroupDesc> {
@@ -1927,20 +1818,6 @@ 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
@@ -1953,14 +1830,13 @@ where
         let s: HashSet<&String> = HashSet::from_iter(gi.members.iter());
         let d: HashSet<&String> = HashSet::from_iter(members.iter());
         gi.members = s.difference(&d).map(|v| v.to_string()).collect::<Vec<String>>();
-        let now = updated_at;
-        gi.update_at = Some(now);
 
         if !update_cache_only {
             self.api.save_group_info(name, gi.clone()).await?;
         }
 
-        self.cache.with_write_lock(|cache| {
+        let now = self.cache.with_write_lock(|cache| {
+            let now = OffsetDateTime::now_utc();
             cache.add_or_update_group(name, &gi, now);
 
             let user_group_memberships = Arc::clone(&cache.state().user_group_memberships);
@@ -1971,25 +1847,13 @@ where
                     cache.add_or_update_user_group_membership(member, &m, now);
                 }
             });
+            now
         });
 
         Ok(now)
     }
 
     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);
         }
@@ -2038,17 +1902,18 @@ where
                 return Err(err);
             }
 
-            self.cache.with_write_lock(|cache| {
+            let now = 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(updated_at);
+            return Ok(now);
         }
 
-        self.remove_members_from_group_at(group, members, false, updated_at).await
+        self.remove_members_from_group(group, members, false).await
     }
 
     fn remove_group_from_memberships_map_unlocked(&self, cache: &mut LockedCache, group: &str, now: OffsetDateTime) {
@@ -2370,19 +2235,6 @@ 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,14 +385,7 @@ impl<T: Store> IamSys<T> {
     }
 
     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` (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?;
+        let updated_at = self.store.set_policy(name, policy).await?;
 
         if !self.has_watcher() {
             for r in notify_iam_load_policy(name).await {
@@ -650,18 +643,7 @@ impl<T: Store> IamSys<T> {
     }
 
     pub async fn set_user_status(&self, name: &str, status: rustfs_madmin::AccountStatus) -> Result<OffsetDateTime> {
-        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?;
+        let updated_at = self.store.set_user_status(name, status).await?;
 
         self.notify_for_user(name, false).await;
 
@@ -673,20 +655,6 @@ 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);
@@ -756,18 +724,11 @@ 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;
-        // 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.status = ACCOUNT_ON.to_owned();
         cred.name = opts.name;
         cred.description = opts.description;
 
-        let create_at = self.store.add_service_account_at(cred.clone(), updated_at).await?;
+        let create_at = self.store.add_service_account(cred.clone()).await?;
 
         self.notify_for_service_account(&cred.access_key).await;
 
@@ -775,23 +736,11 @@ 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_at(name, opts, updated_at).await?;
+        let updated_at = self.store.update_service_account(name, opts).await?;
 
         self.notify_for_service_account(name).await;
 
@@ -991,17 +940,6 @@ 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);
         }
@@ -1014,7 +952,7 @@ impl<T: Store> IamSys<T> {
             return Err(IamError::InvalidSecretKeyLength);
         }
 
-        let updated_at = self.store.add_user_at(access_key, args, updated_at).await?;
+        let updated_at = self.store.add_user(access_key, args).await?;
         self.load_user(access_key, UserType::Reg).await?;
 
         self.notify_for_user(access_key, false).await;
@@ -1088,21 +1026,10 @@ 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_at(group, users, updated_at).await?;
+        let updated_at = self.store.add_users_to_group(group, users).await?;
 
         self.notify_for_group(group).await;
 
@@ -1110,19 +1037,7 @@ impl<T: Store> IamSys<T> {
     }
 
     pub async fn remove_users_from_group(&self, group: &str, users: Vec<String>) -> Result<OffsetDateTime> {
-        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?;
+        let updated_at = self.store.remove_users_from_group(group, users).await?;
 
         self.notify_for_group(group).await;
 
@@ -1130,13 +1045,7 @@ impl<T: Store> IamSys<T> {
     }
 
     pub async fn set_group_status(&self, group: &str, enable: bool) -> Result<OffsetDateTime> {
-        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?;
+        let updated_at = self.store.set_group_status(group, enable).await?;
 
         self.notify_for_group(group).await;
 
@@ -1146,22 +1055,6 @@ 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
     }
@@ -1171,24 +1064,7 @@ impl<T: Store> IamSys<T> {
     }
 
     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` (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?;
+        let updated_at = self.store.policy_db_set(name, user_type, is_group, policy).await?;
 
         if !self.has_watcher() {
             for r in notify_iam_load_policy_mapping(name, user_type.to_u64(), is_group).await {
@@ -1970,11 +1846,6 @@ 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 {

crates/policy/src/policy/doc.rs

@@ -45,33 +45,18 @@ 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(at),
-            update_date: Some(at),
+            create_date: Some(OffsetDateTime::now_utc()),
+            update_date: Some(OffsetDateTime::now_utc()),
         }
     }
 
     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(at);
+        self.update_date = Some(OffsetDateTime::now_utc());
 
         if self.create_date.is_none() {
             self.create_date = self.update_date;

crates/replication/src/delete.rs

@@ -76,21 +76,6 @@ 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,
@@ -253,28 +238,6 @@ 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`
@@ -295,9 +258,9 @@ mod tests {
 
     use super::{
         DeletedObjectReplicationInfo, delete_marker_purge_mrf_entry, delete_marker_purge_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,
+        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,
     };
     use crate::storage_api::DeletedObject;
     use crate::{
@@ -632,76 +595,6 @@ 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]
@@ -763,57 +656,4 @@ 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,26 +641,6 @@ 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, 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,
+    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,
 };
 pub use filemeta::{
     NULL_VERSION_ID, REPLICATE_EXISTING, REPLICATE_EXISTING_DELETE, REPLICATE_HEAL, REPLICATE_HEAL_DELETE, REPLICATE_INCOMING,

crates/replication/src/mrf.rs

@@ -31,13 +31,8 @@ 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;
-// 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;
+const MRF_KNOWN_CAPABILITIES: u64 =
+    CAPABILITY_OPERATION_KIND | CAPABILITY_TARGET_ARNS | CAPABILITY_FORCE_DELETE | CAPABILITY_DELETE_MARKER_MTIME;
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum MrfCapability {
@@ -45,7 +40,6 @@ pub enum MrfCapability {
     TargetArns,
     ForceDelete,
     DeleteMarkerMtime,
-    TargetDeleteMarkerVersionIds,
 }
 
 impl MrfCapability {
@@ -55,7 +49,6 @@ 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,
         }
     }
 }
@@ -608,17 +601,9 @@ 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,
     ];
 
@@ -641,8 +626,6 @@ 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 {
@@ -659,8 +642,6 @@ 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 {
@@ -677,11 +658,6 @@ 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()],
             },
         ];
@@ -709,54 +685,6 @@ 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]
@@ -791,99 +719,6 @@ 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]

rustfs/src/admin/handlers/user.rs

@@ -1131,7 +1131,6 @@ impl Operation for ImportIam {
                         expiration: req.expiration,
                         allow_site_replicator_account: false,
                         claims: Some(req.claims),
-                        status: None,
                     };
 
                     let groups = if req.groups.is_empty() { None } else { Some(req.groups) };

rustfs/src/admin/storage_api.rs

@@ -325,25 +325,6 @@ pub(crate) mod metadata_sys {
         super::ecstore_bucket::metadata_sys::update_if_incarnation(bucket, config_file, data, expected_incarnation_id).await
     }
 
-    /// [`update_if_incarnation`] stamping the config with a replicated edit's
-    /// source `updated_at` instead of the local clock (backlog#2292).
-    pub(crate) async fn update_if_incarnation_at(
-        bucket: &str,
-        config_file: &str,
-        data: Vec<u8>,
-        expected_incarnation_id: uuid::Uuid,
-        updated_at: OffsetDateTime,
-    ) -> Result<OffsetDateTime> {
-        super::ecstore_bucket::metadata_sys::update_if_incarnation_at(
-            bucket,
-            config_file,
-            data,
-            expected_incarnation_id,
-            updated_at,
-        )
-        .await
-    }
-
     pub(crate) async fn update_quota_if_incarnation(
         bucket: &str,
         data: Vec<u8>,
@@ -353,25 +334,6 @@ pub(crate) mod metadata_sys {
         super::ecstore_bucket::metadata_sys::update_quota_if_incarnation(bucket, data, expected_incarnation_id, proof).await
     }
 
-    /// [`update_quota_if_incarnation`] stamping the quota with a replicated
-    /// edit's source `updated_at` instead of the local clock (backlog#2292).
-    pub(crate) async fn update_quota_if_incarnation_at(
-        bucket: &str,
-        data: Vec<u8>,
-        expected_incarnation_id: uuid::Uuid,
-        proof: &super::ecstore_notification::CrossPoolFenceFleetProofToken,
-        updated_at: OffsetDateTime,
-    ) -> Result<OffsetDateTime> {
-        super::ecstore_bucket::metadata_sys::update_quota_if_incarnation_at(
-            bucket,
-            data,
-            expected_incarnation_id,
-            proof,
-            updated_at,
-        )
-        .await
-    }
-
     pub(crate) async fn capture_bucket_metadata_incarnation(bucket: &str) -> Result<uuid::Uuid> {
         super::ecstore_bucket::metadata_sys::capture_bucket_metadata_incarnation(bucket).await
     }
@@ -426,18 +388,6 @@ pub(crate) mod metadata_sys {
         super::ecstore_bucket::metadata_sys::delete_if_incarnation(bucket, config_file, expected_incarnation_id).await
     }
 
-    /// [`delete_if_incarnation`] stamping the cleared config with a replicated
-    /// deletion's source `updated_at` instead of the local clock (backlog#2292).
-    pub(crate) async fn delete_if_incarnation_at(
-        bucket: &str,
-        config_file: &str,
-        expected_incarnation_id: uuid::Uuid,
-        updated_at: OffsetDateTime,
-    ) -> Result<OffsetDateTime> {
-        super::ecstore_bucket::metadata_sys::delete_if_incarnation_at(bucket, config_file, expected_incarnation_id, updated_at)
-            .await
-    }
-
     pub(crate) async fn get_bucket_policy(bucket: &str) -> Result<(BucketPolicy, OffsetDateTime)> {
         super::ecstore_bucket::metadata_sys::get_bucket_policy(bucket).await
     }

rustfs/src/app/storage_api.rs

@@ -712,8 +712,6 @@ pub(crate) mod bucket {
                     delete_marker_version_id: None,
                     delete_marker: false,
                     delete_marker_mtime: None,
-                    target_delete_marker_version_ids: Default::default(),
-                    target_delete_marker_version_ids_corrupt: false,
                     target_arns,
                     force_delete_id: Some(operation_id),
                     force_delete_generation: Some(i64::try_from(generation.unix_timestamp_nanos()).unwrap_or(i64::MAX)),

rustfs/src/site_replication/hooks.rs

@@ -302,164 +302,7 @@ pub(crate) fn site_replication_state_replicates_ilm_expiry(state: &SiteReplicati
     state.peers.values().any(|peer| peer.replicate_ilm_expiry)
 }
 
-/// Secret-bearing half of the IAM snapshot. `SRInfo` is served to admin
-/// callers (`site-replication/info`, status, add preflight) and must stay
-/// secret-free, so the bootstrap plan receives credentials through this
-/// separate value, built only on the paths that deliver to peers (site add
-/// bootstrap, repair, retry snapshot resend). Never persisted, never served.
-#[derive(Debug, Clone, Default)]
-pub(crate) struct SiteReplicationIamCredentials {
-    /// Built-in users (access key -> credential); temp and service accounts
-    /// are excluded, external/IdP users never appear here.
-    pub(crate) users: BTreeMap<String, SiteReplicationUserCredential>,
-    /// Every service account except the site replicator's own, already
-    /// shaped as the `service-account` create item the live hook emits.
-    pub(crate) service_accounts: Vec<SiteReplicationServiceAccountSnapshot>,
-}
-
-#[derive(Debug, Clone)]
-pub(crate) struct SiteReplicationUserCredential {
-    pub(crate) secret_key: String,
-    pub(crate) status: AccountStatus,
-    /// The user record's own update time (the axis the receiver's staleness
-    /// check compares against), unlike `UserInfo::updated_at` which
-    /// `list_users` overwrites with the policy mapping's time.
-    pub(crate) updated_at: Option<OffsetDateTime>,
-}
-
-#[derive(Debug, Clone)]
-pub(crate) struct SiteReplicationServiceAccountSnapshot {
-    pub(crate) create: SRSvcAccCreate,
-    pub(crate) envelope: Option<SRSvcAccReplicationEnvelope>,
-    pub(crate) updated_at: Option<OffsetDateTime>,
-}
-
-pub(crate) const SERVICE_ACCOUNT_ENVELOPE_VERSION: u64 = 2;
-
-pub(crate) fn encode_service_account_replication_policy(
-    claims: &HashMap<String, Value>,
-    session_policy: Option<&str>,
-) -> S3Result<(SRSessionPolicy, Option<SRSvcAccReplicationEnvelope>)> {
-    if !claims.contains_key(OIDC_VIRTUAL_PARENT_CLAIM) {
-        return session_policy
-            .map(SRSessionPolicy::from_json)
-            .transpose()
-            .map(|policy| policy.unwrap_or_default())
-            .map(|policy| (policy, None))
-            .map_err(|err| s3_error!(InvalidArgument, "marshal policy failed: {:?}", err));
-    }
-
-    let policy = match session_policy {
-        Some(policy) => serde_json::from_str::<Policy>(policy)
-            .map_err(|err| s3_error!(InvalidArgument, "invalid service account replication policy: {:?}", err))?,
-        None => Policy::default(),
-    };
-    if policy.statements.is_empty() && (!policy.id.is_empty() || !policy.version.is_empty())
-        || policy.version.is_empty() && !policy.statements.is_empty()
-    {
-        return Err(s3_error!(InvalidArgument, "service account replication policy is not normalized"));
-    }
-    let policy = serde_json::to_string(&policy)
-        .map_err(|err| s3_error!(InternalError, "marshal service account replication policy failed: {:?}", err))?;
-    let policy = SRSessionPolicy::from_json(&policy)
-        .map_err(|err| s3_error!(InternalError, "marshal service account replication policy failed: {:?}", err))?;
-    Ok((
-        policy,
-        Some(SRSvcAccReplicationEnvelope {
-            version: SERVICE_ACCOUNT_ENVELOPE_VERSION,
-        }),
-    ))
-}
-
-/// Read the credentials the IAM snapshot needs straight from the IAM store:
-/// `list_users` deliberately strips secret keys and skips service accounts,
-/// which is right for an admin listing and wrong for a peer snapshot (the
-/// plan builder used to drop every user for lack of a secret, so a status
-/// change or secret rotation committed while a peer was unreachable never
-/// reached it — backlog#2289).
-pub(crate) async fn build_sr_iam_credentials() -> S3Result<SiteReplicationIamCredentials> {
-    let mut credentials = SiteReplicationIamCredentials::default();
-    let Some(iam_sys) = current_iam_handle() else {
-        return Ok(credentials);
-    };
-
-    let mut users = HashMap::new();
-    iam_sys.load_users(UserType::Reg, &mut users).await.map_err(ApiError::from)?;
-    for (access_key, identity) in users {
-        if identity.credentials.is_temp() || identity.credentials.is_service_account() {
-            continue;
-        }
-        credentials.users.insert(
-            access_key,
-            SiteReplicationUserCredential {
-                secret_key: identity.credentials.secret_key,
-                status: if identity.credentials.status == "off" {
-                    AccountStatus::Disabled
-                } else {
-                    AccountStatus::Enabled
-                },
-                updated_at: identity.update_at,
-            },
-        );
-    }
-
-    let mut service_accounts = HashMap::new();
-    iam_sys
-        .load_users(UserType::Svc, &mut service_accounts)
-        .await
-        .map_err(ApiError::from)?;
-    let mut service_accounts: Vec<_> = service_accounts.into_iter().collect();
-    service_accounts.sort_by(|(a, _), (b, _)| a.cmp(b));
-    for (access_key, identity) in service_accounts {
-        // The replicator account is installed by join / rotate, never by a snapshot.
-        if access_key == SITE_REPLICATOR_SERVICE_ACCOUNT || !identity.credentials.is_service_account() {
-            continue;
-        }
-        let claims = iam_sys.get_claims_for_svc_acc(&access_key).await.map_err(ApiError::from)?;
-        let (account, session_policy) = iam_sys.get_service_account(&access_key).await.map_err(ApiError::from)?;
-        let session_policy = session_policy
-            .map(|policy| serde_json::to_string(&policy))
-            .transpose()
-            .map_err(|err| {
-                S3Error::with_message(
-                    S3ErrorCode::InternalError,
-                    format!("marshal service account session policy failed: {err:?}"),
-                )
-            })?;
-        let (session_policy, envelope) = encode_service_account_replication_policy(&claims, session_policy.as_deref())?;
-        credentials.service_accounts.push(SiteReplicationServiceAccountSnapshot {
-            create: SRSvcAccCreate {
-                parent: identity.credentials.parent_user,
-                access_key,
-                secret_key: identity.credentials.secret_key,
-                groups: identity.credentials.groups.unwrap_or_default(),
-                claims,
-                session_policy,
-                status: identity.credentials.status,
-                name: account.name.unwrap_or_default(),
-                description: account.description.unwrap_or_default(),
-                expiration: account.expiration,
-                api_version: Some(SITE_REPL_API_VERSION.to_string()),
-            },
-            envelope,
-            updated_at: identity.update_at,
-        });
-    }
-
-    Ok(credentials)
-}
-
-/// The bootstrap plan for peer delivery: `info` (secret-free) plus the IAM
-/// credentials read at this moment.
-pub(crate) async fn build_site_replication_bootstrap_plan(info: &SRInfo) -> S3Result<SiteReplicationBootstrapPlan> {
-    let credentials = build_sr_iam_credentials().await?;
-    site_replication_bootstrap_plan(info, &credentials)
-}
-
-pub(crate) fn site_replication_bootstrap_plan(
-    info: &SRInfo,
-    credentials: &SiteReplicationIamCredentials,
-) -> S3Result<SiteReplicationBootstrapPlan> {
+pub(crate) fn site_replication_bootstrap_plan(info: &SRInfo) -> S3Result<SiteReplicationBootstrapPlan> {
     let mut plan = SiteReplicationBootstrapPlan::default();
     let replicate_ilm_expiry = site_replication_info_replicates_ilm_expiry(info);
 
@@ -475,57 +318,24 @@ pub(crate) fn site_replication_bootstrap_plan(
     }
 
     for (access_key, user) in &info.user_info_map {
-        // Credentials come from the store snapshot; an inline `secret_key` on
-        // the SRInfo entry (older callers, tests) is accepted as a fallback.
-        // Users with neither (external / IdP identities) have nothing a peer
-        // could install and are skipped.
-        let credential = credentials.users.get(access_key);
-        let Some(secret_key) = credential
-            .map(|credential| credential.secret_key.clone())
-            .or_else(|| user.secret_key.clone())
-            .filter(|secret_key| !secret_key.is_empty())
-        else {
-            continue;
-        };
-        let status = credential
-            .map(|credential| credential.status.clone())
-            .unwrap_or_else(|| user.status.clone());
-        let updated_at = credential.and_then(|credential| credential.updated_at).or(user.updated_at);
-        plan.iam_items.push(SRIAMItem {
-            r#type: "iam-user".to_string(),
-            iam_user: Some(rustfs_madmin::SRIAMUser {
-                access_key: access_key.clone(),
-                is_delete_req: false,
-                user_req: Some(AddOrUpdateUserReq {
-                    secret_key,
-                    policy: user.policy_name.clone(),
-                    status,
+        if let Some(secret_key) = &user.secret_key {
+            plan.iam_items.push(SRIAMItem {
+                r#type: "iam-user".to_string(),
+                iam_user: Some(rustfs_madmin::SRIAMUser {
+                    access_key: access_key.clone(),
+                    is_delete_req: false,
+                    user_req: Some(AddOrUpdateUserReq {
+                        secret_key: secret_key.clone(),
+                        policy: user.policy_name.clone(),
+                        status: user.status.clone(),
+                    }),
+                    api_version: Some(SITE_REPL_API_VERSION.to_string()),
                 }),
-                api_version: Some(SITE_REPL_API_VERSION.to_string()),
-            }),
-            updated_at,
-            api_version: Some(SITE_REPL_API_VERSION.to_string()),
-            ..Default::default()
-        });
-    }
-
-    // Service accounts follow their parents: the receiver creates a missing
-    // account under `parent` and updates an existing one (secret, status,
-    // session policy), so a rotation or disable committed during an outage
-    // converges through the same snapshot as users do.
-    for account in &credentials.service_accounts {
-        plan.iam_items.push(SRIAMItem {
-            r#type: "service-account".to_string(),
-            svc_acc_change: Some(SRSvcAccChange {
-                create: Some(account.create.clone()),
-                oidc_service_account_envelope: account.envelope.clone(),
+                updated_at: user.updated_at,
                 api_version: Some(SITE_REPL_API_VERSION.to_string()),
                 ..Default::default()
-            }),
-            updated_at: account.updated_at,
-            api_version: Some(SITE_REPL_API_VERSION.to_string()),
-            ..Default::default()
-        });
+            });
+        }
     }
 
     for (name, desc) in &info.group_desc_map {
@@ -708,12 +518,7 @@ pub(crate) async fn broadcast_site_replication_make_bucket(
     } else {
         path
     };
-    // Both steps run to completion on their own: the broadcast attempts every
-    // peer and reports the first failure (backlog#2293), so stopping here on
-    // that error would skip `configure-replication` for the peers whose
-    // `make` just succeeded — and nothing records a retry for that gap. The
-    // failed peer's retry events cover both steps independently.
-    let make_result = broadcast_site_replication_json_using_runtime(runtime, &path, &serde_json::json!({})).await;
+    broadcast_site_replication_json_using_runtime(runtime, &path, &serde_json::json!({})).await?;
 
     let configure_path = bootstrap_bucket_op_path(bucket, "configure-replication");
     let configure_path = if let Some(token) = bootstrap_token {
@@ -721,8 +526,7 @@ pub(crate) async fn broadcast_site_replication_make_bucket(
     } else {
         configure_path
     };
-    let configure_result = broadcast_site_replication_json_using_runtime(runtime, &configure_path, &serde_json::json!({})).await;
-    make_result.and(configure_result)
+    broadcast_site_replication_json_using_runtime(runtime, &configure_path, &serde_json::json!({})).await
 }
 
 const SITE_REPLICATION_DELETE_INTENT_PENDING: &str =
@@ -1028,21 +832,6 @@ pub async fn site_replication_iam_change_hook(item: SRIAMItem) -> S3Result<()> {
     let Some(runtime) = runtime_site_replication_targets().await? else {
         return Ok(());
     };
-    // A local revoke must out-rank a stale grant a peer delivers later, so its
-    // mark is committed before the broadcast (backlog#2291). The broadcast
-    // still goes out when the mark cannot be persisted: the peers' own records
-    // remain the primary gate, the mark only covers the deleted case.
-    if let Err(err) = record_iam_deletion_marks_for_item(&item).await {
-        warn!(
-            component = LOG_COMPONENT_ADMIN,
-            subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
-            event = EVENT_ADMIN_SITE_REPLICATION_STATE,
-            item_type = %item.r#type,
-            result = "iam_deletion_mark_not_recorded",
-            error = ?err,
-            "failed to record local IAM deletion mark before broadcast"
-        );
-    }
     let mut first_error: Option<S3Error> = None;
     for peer in runtime.state.peers.values() {
         if peer.deployment_id == runtime.local_peer.deployment_id

rustfs/src/site_replication/mod.rs

@@ -79,16 +79,13 @@ use http::header::{CONTENT_TYPE, HOST};
 use http::{HeaderMap, HeaderValue, Uri};
 use hyper::{Method, StatusCode};
 use rustfs_config::{DEFAULT_CONSOLE_ADDRESS, DEFAULT_RUSTFS_TLS_PATH, ENV_RUSTFS_CONSOLE_ADDRESS, ENV_RUSTFS_TLS_PATH};
-use rustfs_iam::federation::OIDC_VIRTUAL_PARENT_CLAIM;
 use rustfs_iam::store::{MappedPolicy, UserType, sr_wire_user_type};
 use rustfs_iam::sys::SITE_REPLICATOR_SERVICE_ACCOUNT;
 use rustfs_madmin::{
-    AccountStatus, AddOrUpdateUserReq, GroupAddRemove, GroupStatus, PeerInfo, PeerSite, ReplicateEditStatus,
-    SITE_REPL_API_VERSION, SRBucketInfo, SRBucketMeta, SRGroupInfo, SRIAMItem, SRIAMPolicy, SRInfo, SRPolicyMapping, SRRemoveReq,
-    SRResyncOpStatus, SRRetryStats, SRSessionPolicy, SRStateInfo, SRSvcAccChange, SRSvcAccCreate, SRSvcAccDelete,
-    SRSvcAccReplicationEnvelope, SyncStatus,
+    AddOrUpdateUserReq, GroupAddRemove, GroupStatus, PeerInfo, PeerSite, ReplicateEditStatus, SITE_REPL_API_VERSION,
+    SRBucketInfo, SRBucketMeta, SRGroupInfo, SRIAMItem, SRIAMPolicy, SRInfo, SRPolicyMapping, SRRemoveReq, SRResyncOpStatus,
+    SRRetryStats, SRStateInfo, SyncStatus,
 };
-use rustfs_policy::policy::Policy;
 use rustfs_signer::constants::UNSIGNED_PAYLOAD;
 use rustfs_signer::sign_v4;
 use rustfs_tls_runtime::{GlobalPublishedOutboundTlsState, TlsGeneration};
@@ -110,26 +107,6 @@ use tracing::{info, warn};
 use url::{Url, form_urlencoded};
 use uuid::Uuid;
 
-/// Serialize `value` with every JSON object's keys sorted, for hashing and
-/// equality checks. `HashMap` fields (service-account claims) iterate in a
-/// per-instance random order and `serde_json` is built with `preserve_order`,
-/// so two identical plans would otherwise hash differently: the repair
-/// preflight token went stale between dry-run and execute, and a retry
-/// snapshot resend never looked "stable" (backlog#2289 follow-up).
-pub(crate) fn canonical_json_vec<T: Serialize>(value: &T) -> serde_json::Result<Vec<u8>> {
-    fn sort_keys(value: Value) -> Value {
-        match value {
-            Value::Object(map) => {
-                let sorted: BTreeMap<String, Value> = map.into_iter().map(|(key, value)| (key, sort_keys(value))).collect();
-                Value::Object(sorted.into_iter().collect())
-            }
-            Value::Array(items) => Value::Array(items.into_iter().map(sort_keys).collect()),
-            other => other,
-        }
-    }
-    serde_json::to_vec(&sort_keys(serde_json::to_value(value)?))
-}
-
 pub(crate) const LOG_COMPONENT_ADMIN: &str = "admin";
 
 pub(crate) const LOG_SUBSYSTEM_SITE_REPLICATION: &str = "site_replication";

rustfs/src/site_replication/repair.rs

@@ -234,9 +234,9 @@ impl SiteReplicationRepairTask<'_> {
 
     pub(crate) fn id(&self) -> S3Result<String> {
         let payload = match self {
-            Self::Iam(item) => canonical_json_vec(item),
+            Self::Iam(item) => serde_json::to_vec(item),
             Self::BucketMake(_) | Self::Replication(_) => serde_json::to_vec(&serde_json::json!({})),
-            Self::BucketMetadata(item) => canonical_json_vec(item),
+            Self::BucketMetadata(item) => serde_json::to_vec(item),
         }
         .map_err(|err| S3Error::with_message(S3ErrorCode::InternalError, format!("serialize repair task failed: {err}")))?;
         let mut digest = Sha256::new();
@@ -726,7 +726,7 @@ pub(crate) async fn execute_site_replication_repair_locked(
         return Err(s3_error!(InvalidRequest, "site replication is not configured"));
     }
     let info = build_sr_info(&state, &request.local_peer).await?;
-    let plan = build_site_replication_bootstrap_plan(&info).await?;
+    let plan = site_replication_bootstrap_plan(&info)?;
     let plan_token = site_replication_repair_plan_token(&state, &plan)?;
     let preflight_token = site_replication_repair_preflight_token(&state, &plan, request.signing_key.as_bytes())?;
     let sites = site_replication_repair_sites(&state, &request.local_peer, &plan, request.signing_key.as_bytes())?;

rustfs/src/site_replication/retry.rs

@@ -397,12 +397,12 @@ pub(crate) fn iam_deletion_replay_matches(record: &SiteReplicationIamDeletionRep
 /// newer revision of one another.
 pub(crate) fn iam_item_deletion_entity(item: &SRIAMItem) -> Option<String> {
     match item.r#type.as_str() {
-        "policy" if item.policy.is_none() => Some(iam_policy_deletion_mark_entity(&item.name)),
+        "policy" if item.policy.is_none() => Some(format!("policy:{}", item.name)),
         "iam-user" => item
             .iam_user
             .as_ref()
             .filter(|user| user.is_delete_req)
-            .map(|user| iam_user_deletion_mark_entity(&user.access_key)),
+            .map(|user| format!("iam-user:{}", user.access_key)),
         "group-info" => item
             .group_info
             .as_ref()
@@ -416,7 +416,7 @@ pub(crate) fn iam_item_deletion_entity(item: &SRIAMItem) -> Option<String> {
             .policy_mapping
             .as_ref()
             .filter(|mapping| mapping.policy.is_empty())
-            .map(|mapping| iam_policy_mapping_deletion_mark_entity(&mapping.user_or_group, mapping.user_type, mapping.is_group)),
+            .map(|mapping| format!("policy-mapping:{}:{}:{}", mapping.user_or_group, mapping.user_type, mapping.is_group)),
         "service-account" => item
             .svc_acc_change
             .as_ref()
@@ -426,82 +426,6 @@ pub(crate) fn iam_item_deletion_entity(item: &SRIAMItem) -> Option<String> {
     }
 }
 
-/// The entities whose deletion a deletion-shaped IAM item commits, keyed the
-/// way the receive-side staleness gate looks them up once the local record is
-/// gone (backlog#2291); empty for creates and updates. Group member removal
-/// yields one entity per removed member so a stale re-add of that member can
-/// be judged, and a group delete (no members) yields the group itself.
-pub(crate) fn iam_item_deletion_mark_entities(item: &SRIAMItem) -> Vec<String> {
-    if item.r#type == "group-info" {
-        let Some(update) = item
-            .group_info
-            .as_ref()
-            .map(|group| &group.update_req)
-            .filter(|update| update.is_remove)
-        else {
-            return Vec::new();
-        };
-        if update.members.is_empty() {
-            return vec![iam_group_deletion_mark_entity(&update.group)];
-        }
-        return update
-            .members
-            .iter()
-            .map(|member| iam_group_member_deletion_mark_entity(&update.group, member))
-            .collect();
-    }
-    iam_item_deletion_entity(item).into_iter().collect()
-}
-
-pub(crate) fn iam_policy_deletion_mark_entity(name: &str) -> String {
-    format!("policy:{name}")
-}
-
-pub(crate) fn iam_user_deletion_mark_entity(access_key: &str) -> String {
-    format!("iam-user:{access_key}")
-}
-
-/// `user_type` is the SR wire integer, as carried by the item on both sides.
-pub(crate) fn iam_policy_mapping_deletion_mark_entity(user_or_group: &str, user_type: i64, is_group: bool) -> String {
-    format!("policy-mapping:{user_or_group}:{user_type}:{is_group}")
-}
-
-pub(crate) fn iam_group_deletion_mark_entity(group: &str) -> String {
-    format!("group:{group}")
-}
-
-pub(crate) fn iam_group_member_deletion_mark_entity(group: &str, member: &str) -> String {
-    format!("group-member:{group}:{member}")
-}
-
-/// Persist the deletion marks of `item` (its source `updated_at` per entity
-/// of [`iam_item_deletion_mark_entities`]) through the state transaction.
-/// No-op for creates/updates and for items without a source timestamp
-/// (older peers): a mark without a source clock could not be ordered against
-/// later items. Called before a local deletion is broadcast and after a
-/// replicated deletion is applied, so both sides out-rank a stale grant that
-/// arrives later.
-pub(crate) async fn record_iam_deletion_marks_for_item(item: &SRIAMItem) -> S3Result<()> {
-    let entities = iam_item_deletion_mark_entities(item);
-    let Some(deleted_at) = item.updated_at.filter(|_| !entities.is_empty()) else {
-        return Ok(());
-    };
-    commit_iam_deletion_marks(entities, deleted_at).await
-}
-
-/// [`record_iam_deletion_marks`] under the state transaction; the write is
-/// skipped when no mark moves.
-pub(crate) async fn commit_iam_deletion_marks(entities: Vec<String>, deleted_at: OffsetDateTime) -> S3Result<()> {
-    update_site_replication_state_when_changed(move |state| {
-        Ok(if record_iam_deletion_marks(state, &entities, deleted_at) {
-            StateCommit::Changed(())
-        } else {
-            StateCommit::Unchanged(())
-        })
-    })
-    .await
-}
-
 /// Failure bookkeeping for one IAM item delivery: upsert the collapsed retry
 /// event and, when the item is a deletion, record its body for replay. Both
 /// live in the same state so the caller commits them in one transaction — a
@@ -867,8 +791,8 @@ impl RetrySnapshot {
 
     pub(crate) fn fingerprint(&self) -> S3Result<Vec<Vec<u8>>> {
         let mut payloads = match self {
-            Self::Iam(items) => items.iter().map(canonical_json_vec).collect::<Result<Vec<_>, _>>(),
-            Self::BucketMetadata(items) => items.iter().map(canonical_json_vec).collect::<Result<Vec<_>, _>>(),
+            Self::Iam(items) => items.iter().map(serde_json::to_vec).collect::<Result<Vec<_>, _>>(),
+            Self::BucketMetadata(items) => items.iter().map(serde_json::to_vec).collect::<Result<Vec<_>, _>>(),
         }
         .map_err(|err| S3Error::with_message(S3ErrorCode::InternalError, format!("serialize retry snapshot failed: {err}")))?;
         payloads.sort_unstable();
@@ -1030,7 +954,6 @@ pub(crate) enum IamSnapshotKey {
     User(String),
     Group(String),
     PolicyMapping { target: String, user_type: i64, is_group: bool },
-    ServiceAccount(String),
 }
 
 pub(crate) fn iam_snapshot_key(item: &SRIAMItem) -> Option<IamSnapshotKey> {
@@ -1049,11 +972,6 @@ pub(crate) fn iam_snapshot_key(item: &SRIAMItem) -> Option<IamSnapshotKey> {
             user_type: mapping.user_type,
             is_group: mapping.is_group,
         }),
-        "service-account" => item
-            .svc_acc_change
-            .as_ref()
-            .and_then(|change| change.create.as_ref())
-            .map(|create| IamSnapshotKey::ServiceAccount(create.access_key.clone())),
         _ => None,
     }
 }
@@ -1088,24 +1006,6 @@ pub(crate) fn iam_snapshot_tombstones(item: &SRIAMItem, observed_at: OffsetDateT
                 mapping.policy.clear();
             }
         }
-        "service-account" => {
-            let Some(access_key) = item
-                .svc_acc_change
-                .as_ref()
-                .and_then(|change| change.create.as_ref())
-                .map(|create| create.access_key.clone())
-            else {
-                return Vec::new();
-            };
-            tombstone.svc_acc_change = Some(SRSvcAccChange {
-                delete: Some(SRSvcAccDelete {
-                    access_key,
-                    api_version: Some(SITE_REPL_API_VERSION.to_string()),
-                }),
-                api_version: Some(SITE_REPL_API_VERSION.to_string()),
-                ..Default::default()
-            });
-        }
         _ => return Vec::new(),
     }
     vec![tombstone]
@@ -1801,7 +1701,7 @@ pub(crate) async fn drain_site_replication_retry_queue_locked(
     // tick and only when a snapshot resend is actually due.
     let plan = if needs_plan {
         let info = build_sr_info(&runtime.state, &runtime.local_peer).await?;
-        Some(build_site_replication_bootstrap_plan(&info).await?)
+        Some(site_replication_bootstrap_plan(&info)?)
     } else {
         None
     };
@@ -1941,7 +1841,7 @@ pub(crate) async fn drain_one_site_replication_retry_event(
                     }
                 }
                 let fresh_info = build_sr_info(&runtime.state, &runtime.local_peer).await?;
-                let fresh_plan = build_site_replication_bootstrap_plan(&fresh_info).await?;
+                let fresh_plan = site_replication_bootstrap_plan(&fresh_info)?;
                 let fresh_snapshot = RetrySnapshot::from_plan(&action, &fresh_plan).expect("snapshot action has a snapshot");
                 if fresh_snapshot.fingerprint()? == current_fingerprint {
                     if is_iam {

rustfs/src/site_replication/state.rs

@@ -64,104 +64,6 @@ pub(crate) struct SiteReplicationState {
     /// newer edit that already landed.
     #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
     pub(crate) applied_edit_generations: BTreeMap<String, u64>,
-    /// Source timestamp of the newest IAM deletion committed on this site,
-    /// keyed by the deleted entity (`iam_item_deletion_mark_entities`). A
-    /// deletion leaves no local record to judge a later item against, so this
-    /// is what lets the receive-side staleness gate reject a grant that is
-    /// older than the revoke it would otherwise undo (backlog#2291). Marks
-    /// are kept for [`SITE_REPLICATION_IAM_DELETION_MARK_RETENTION`] and never
-    /// evicted by count: see that constant for why a count bound would open
-    /// exactly the window the marks exist to close.
-    #[serde(default, with = "rfc3339_map", skip_serializing_if = "BTreeMap::is_empty")]
-    pub(crate) iam_deletion_marks: BTreeMap<String, OffsetDateTime>,
-}
-
-/// How long an IAM deletion mark outlives the deletion it records.
-///
-/// A mark fences the delivery paths that can still carry an older grant for
-/// the deleted entity: a live delivery delayed in transit, the same grant
-/// arriving on a sibling node while the revoke is being applied, and a
-/// snapshot (bootstrap / repair / resend) built by a peer that has not yet
-/// received the deletion — which is bounded by this site's own retry queue
-/// towards that peer, whose backoff tops out at one day
-/// (`SITE_REPLICATION_RETRY_DRAIN_MAX_BACKOFF_SECS`). The retry drain itself
-/// never replays a stale grant: it resends snapshots of the current records
-/// and the recorded deletion bodies. Thirty days is an order of magnitude
-/// beyond every one of those windows. Marks are pruned by age only — a count
-/// bound would drop a mark that is still inside the delivery window as soon
-/// as enough newer deletions happen, letting the delayed grant re-create the
-/// entity, which is the very hole the marks close.
-pub(crate) const SITE_REPLICATION_IAM_DELETION_MARK_RETENTION: time::Duration = time::Duration::days(30);
-
-/// Record that deletions of `entities` with source timestamp `deleted_at`
-/// were committed here. Newest wins per entity: an older deletion never
-/// lowers a mark. Marks older than the retention are pruned in the same
-/// pass. Returns whether the state changed.
-pub(crate) fn record_iam_deletion_marks(
-    state: &mut SiteReplicationState,
-    entities: &[String],
-    deleted_at: OffsetDateTime,
-) -> bool {
-    record_iam_deletion_marks_at(state, entities, deleted_at, OffsetDateTime::now_utc())
-}
-
-/// [`record_iam_deletion_marks`] pruning against an explicit `now`.
-pub(crate) fn record_iam_deletion_marks_at(
-    state: &mut SiteReplicationState,
-    entities: &[String],
-    deleted_at: OffsetDateTime,
-    now: OffsetDateTime,
-) -> bool {
-    let mut changed = false;
-    for entity in entities {
-        if state
-            .iam_deletion_marks
-            .get(entity)
-            .is_some_and(|existing| *existing >= deleted_at)
-        {
-            continue;
-        }
-        state.iam_deletion_marks.insert(entity.clone(), deleted_at);
-        changed = true;
-    }
-    let expired_before = now - SITE_REPLICATION_IAM_DELETION_MARK_RETENTION;
-    let before = state.iam_deletion_marks.len();
-    state.iam_deletion_marks.retain(|_, deleted_at| *deleted_at >= expired_before);
-    changed || state.iam_deletion_marks.len() != before
-}
-
-/// Newest deletion mark among `entities`, or `None` when no deletion of any
-/// of them was recorded here. The receive-side staleness gate feeds this in
-/// as the local timestamp when the targeted record is absent.
-pub(crate) fn iam_deletion_mark(state: &SiteReplicationState, entities: &[String]) -> Option<OffsetDateTime> {
-    entities
-        .iter()
-        .filter_map(|entity| state.iam_deletion_marks.get(entity).copied())
-        .max()
-}
-
-/// RFC 3339 map values, matching the other timestamps in the state object
-/// (`time::serde::rfc3339` only applies to a single field).
-mod rfc3339_map {
-    use serde::{Deserialize, Deserializer, Serialize, Serializer};
-    use std::collections::BTreeMap;
-    use time::OffsetDateTime;
-
-    #[derive(Serialize, Deserialize)]
-    #[serde(transparent)]
-    struct Stamp(#[serde(with = "time::serde::rfc3339")] OffsetDateTime);
-
-    pub(super) fn serialize<S: Serializer>(map: &BTreeMap<String, OffsetDateTime>, serializer: S) -> Result<S::Ok, S::Error> {
-        serializer.collect_map(map.iter().map(|(entity, deleted_at)| (entity, Stamp(*deleted_at))))
-    }
-
-    pub(super) fn deserialize<'de, D: Deserializer<'de>>(deserializer: D) -> Result<BTreeMap<String, OffsetDateTime>, D::Error> {
-        let map = BTreeMap::<String, Stamp>::deserialize(deserializer)?;
-        Ok(map
-            .into_iter()
-            .map(|(entity, Stamp(deleted_at))| (entity, deleted_at))
-            .collect())
-    }
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, Default)]
@@ -421,33 +323,6 @@ where
     update_site_replication_state_when_changed(move |state| update(state).map(StateCommit::Changed)).await
 }
 
-/// The state transaction for work that has to await inside it: an IAM write
-/// that must be ordered with the staleness verdict taken before it and the
-/// deletion mark committed after it (backlog#2291). Same boundary as
-/// [`update_site_replication_state`] — load and persist under the
-/// distributed state-object write lock, so two nodes of this site cannot
-/// interleave their verdicts and writes — and the same rules inside: no peer
-/// network calls and no other config locks. The closure hands the state back
-/// as `Some` when it changed it; `None` skips the write.
-pub(crate) async fn with_site_replication_state_transaction<T, F, Fut>(transaction: F) -> S3Result<T>
-where
-    T: Send + 'static,
-    F: FnOnce(SiteReplicationState) -> Fut + Send + 'static,
-    Fut: std::future::Future<Output = S3Result<(T, Option<SiteReplicationState>)>> + Send + 'static,
-{
-    with_site_replication_state_lock(move || async move {
-        let store = current_object_store_handle()
-            .ok_or_else(|| S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()))?;
-        let state = load_site_replication_state_no_lock(store.clone()).await?;
-        let (result, changed) = transaction(state).await?;
-        if let Some(state) = changed {
-            persist_site_replication_state_no_lock(store, state).await?;
-        }
-        Ok(result)
-    })
-    .await
-}
-
 /// [`update_site_replication_state`] for closures that may find nothing to
 /// do — see [`StateCommit`].
 pub(crate) async fn update_site_replication_state_when_changed<T, F>(update: F) -> S3Result<T>

rustfs/src/site_replication/tests.rs

@@ -554,145 +554,6 @@ fn test_iam_item_deletion_entity_shapes() {
     assert!(iam_item_deletion_entity(&policy_set).is_none());
 }
 
-/// Deletion marks (backlog#2291) key on the same entities as the replay
-/// records, except that a group member removal is marked per member (so a
-/// stale re-add of one member can be judged) and a group delete marks the
-/// group itself. Creates and updates leave no mark.
-#[test]
-fn test_iam_item_deletion_mark_entities_shapes() {
-    assert_eq!(
-        iam_item_deletion_mark_entities(&user_delete_item("alice")),
-        vec!["iam-user:alice".to_string()]
-    );
-    assert_eq!(
-        iam_item_deletion_mark_entities(&policy_delete_item("readonly")),
-        vec!["policy:readonly".to_string()]
-    );
-
-    let mut group_remove = SRIAMItem {
-        r#type: "group-info".to_string(),
-        group_info: Some(SRGroupInfo {
-            update_req: GroupAddRemove {
-                group: "devs".to_string(),
-                members: vec!["bob".to_string(), "alice".to_string()],
-                status: GroupStatus::Enabled,
-                is_remove: true,
-            },
-            api_version: None,
-        }),
-        ..Default::default()
-    };
-    assert_eq!(
-        iam_item_deletion_mark_entities(&group_remove),
-        vec!["group-member:devs:bob".to_string(), "group-member:devs:alice".to_string()]
-    );
-    group_remove
-        .group_info
-        .as_mut()
-        .expect("group info")
-        .update_req
-        .members
-        .clear();
-    assert_eq!(
-        iam_item_deletion_mark_entities(&group_remove),
-        vec!["group:devs".to_string()],
-        "a removal without members deletes the group"
-    );
-    group_remove.group_info.as_mut().expect("group info").update_req.is_remove = false;
-    assert!(iam_item_deletion_mark_entities(&group_remove).is_empty());
-
-    let mapping_clear = SRIAMItem {
-        r#type: "policy-mapping".to_string(),
-        policy_mapping: Some(SRPolicyMapping {
-            user_or_group: "alice".to_string(),
-            user_type: 0,
-            is_group: false,
-            policy: String::new(),
-            ..Default::default()
-        }),
-        ..Default::default()
-    };
-    assert_eq!(
-        iam_item_deletion_mark_entities(&mapping_clear),
-        vec!["policy-mapping:alice:0:false".to_string()]
-    );
-
-    let mut user_create = user_delete_item("alice");
-    user_create.iam_user.as_mut().expect("iam user").is_delete_req = false;
-    assert!(iam_item_deletion_mark_entities(&user_create).is_empty());
-}
-
-/// Newest wins per entity, marks are pruned by age only (never by count: a
-/// count bound would drop a mark still inside the delivery window as soon as
-/// enough newer deletions happen), and the timestamps survive the state
-/// object as RFC 3339.
-#[test]
-fn test_record_iam_deletion_marks_newest_wins_and_expires_by_age_only() {
-    let at = |seconds: i64| OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(seconds);
-    let now = at(1_000_000);
-    let mut state = SiteReplicationState::default();
-    let alice = vec!["iam-user:alice".to_string()];
-
-    assert!(record_iam_deletion_marks_at(&mut state, &alice, at(20), now));
-    assert!(
-        !record_iam_deletion_marks_at(&mut state, &alice, at(10), now),
-        "an older deletion does not move the mark"
-    );
-    assert!(
-        !record_iam_deletion_marks_at(&mut state, &alice, at(20), now),
-        "a replayed deletion is not a change"
-    );
-    assert_eq!(iam_deletion_mark(&state, &alice), Some(at(20)));
-    assert!(record_iam_deletion_marks_at(&mut state, &alice, at(30), now));
-    assert_eq!(iam_deletion_mark(&state, &alice), Some(at(30)));
-    assert_eq!(iam_deletion_mark(&state, &["iam-user:bob".to_string()]), None);
-    assert!(!record_iam_deletion_marks_at(&mut state, &[], at(40), now));
-
-    // Many newer deletions never evict an older mark that is still within the retention.
-    let members: Vec<String> = (0..4096).map(|index| format!("group-member:devs:user-{index:04}")).collect();
-    for (index, member) in members.iter().enumerate() {
-        record_iam_deletion_marks_at(&mut state, std::slice::from_ref(member), at(100 + index as i64), now);
-    }
-    assert_eq!(state.iam_deletion_marks.len(), members.len() + 1);
-    assert_eq!(iam_deletion_mark(&state, &alice), Some(at(30)), "no count-based eviction");
-
-    // Marks older than the retention are pruned, on the pass that records a
-    // newer one and on a pass that changes nothing else; younger ones stay.
-    let later = at(100) + SITE_REPLICATION_IAM_DELETION_MARK_RETENTION;
-    assert!(
-        record_iam_deletion_marks_at(&mut state, &["iam-user:carol".to_string()], at(200_000), later),
-        "pruning alone is a change"
-    );
-    assert_eq!(iam_deletion_mark(&state, &alice), None, "alice's mark aged out");
-    assert_eq!(
-        iam_deletion_mark(&state, &members[..1]),
-        Some(at(100)),
-        "a mark exactly at the retention edge stays, and so do the younger ones"
-    );
-    assert_eq!(state.iam_deletion_marks.len(), members.len() + 1);
-    assert_eq!(iam_deletion_mark(&state, &["iam-user:carol".to_string()]), Some(at(200_000)));
-    let mut state = SiteReplicationState::default();
-    record_iam_deletion_marks_at(&mut state, &alice, at(30), now);
-    let past_edge = at(30) + SITE_REPLICATION_IAM_DELETION_MARK_RETENTION + time::Duration::seconds(1);
-    assert!(
-        record_iam_deletion_marks_at(&mut state, &[], at(0), past_edge),
-        "a pass that only prunes reports the change"
-    );
-    assert_eq!(iam_deletion_mark(&state, &alice), None);
-    record_iam_deletion_marks_at(&mut state, &alice, at(30), now);
-
-    let json = serde_json::to_value(&state).expect("serialize state");
-    assert_eq!(json["iam_deletion_marks"]["iam-user:alice"], serde_json::json!("1970-01-01T00:00:30Z"));
-    let reloaded = parse_site_replication_state(&serde_json::to_vec(&state).expect("serialize state")).expect("parse state");
-    assert_eq!(reloaded.iam_deletion_marks, state.iam_deletion_marks);
-    assert!(
-        parse_site_replication_state(br#"{"name":"a","service_account_access_key":"","service_account_parent":"","peers":{},"updated_at":null,"resync_status":{}}"#)
-            .expect("state without marks")
-            .iam_deletion_marks
-            .is_empty()
-    );
-}
-
 /// A failed deletion delivery persists a replay record next to the collapsed
 /// retry entry; a fresh entry is stamped `deletions_recorded` so a later
 /// replay can settle it, and a repeated deletion of the same entity keeps the
@@ -1818,8 +1679,7 @@ fn test_site_replication_bootstrap_plan_includes_replayable_snapshot_items() {
         },
     );
 
-    let plan =
-        site_replication_bootstrap_plan(&info, &SiteReplicationIamCredentials::default()).expect("bootstrap plan should build");
+    let plan = site_replication_bootstrap_plan(&info).expect("bootstrap plan should build");
 
     assert_eq!(plan.iam_items.iter().map(|item| item.r#type.as_str()).collect::<Vec<_>>(), {
         vec!["policy", "iam-user", "group-info", "policy-mapping"]
@@ -1857,8 +1717,7 @@ fn test_site_replication_bootstrap_plan_skips_lifecycle_by_default() {
         },
     );
 
-    let plan =
-        site_replication_bootstrap_plan(&info, &SiteReplicationIamCredentials::default()).expect("bootstrap plan should build");
+    let plan = site_replication_bootstrap_plan(&info).expect("bootstrap plan should build");
 
     assert!(!plan.bucket_items.iter().any(|item| item.r#type == "lc-config"));
 }
@@ -1889,8 +1748,7 @@ fn test_site_replication_bootstrap_plan_emits_timestamped_lifecycle_delete() {
         },
     );
 
-    let plan =
-        site_replication_bootstrap_plan(&info, &SiteReplicationIamCredentials::default()).expect("bootstrap plan should build");
+    let plan = site_replication_bootstrap_plan(&info).expect("bootstrap plan should build");
 
     let item = plan
         .bucket_items
@@ -2077,8 +1935,8 @@ fn test_site_replication_repair_preflight_token_is_deterministic_for_equal_state
         },
     );
 
-    let plan_a = site_replication_bootstrap_plan(&info, &SiteReplicationIamCredentials::default()).expect("first plan");
-    let plan_b = site_replication_bootstrap_plan(&info, &SiteReplicationIamCredentials::default()).expect("second plan");
+    let plan_a = site_replication_bootstrap_plan(&info).expect("first plan");
+    let plan_b = site_replication_bootstrap_plan(&info).expect("second plan");
     let token_a = site_replication_repair_preflight_token(&state, &plan_a, b"test-signing-key").expect("first token");
     let token_b = site_replication_repair_preflight_token(&state, &plan_b, b"test-signing-key").expect("second token");
 
@@ -3361,345 +3219,3 @@ fn test_reconcile_adds_missing_peer_rules_to_existing_config() {
     assert!(rule_ids.contains(&"site-repl-dep-b"));
     assert!(rule_ids.contains(&"site-repl-dep-c"));
 }
-
-/// backlog#2289: the IAM snapshot (retry resend, repair, site-add bootstrap)
-/// used to be built from `list_users`, whose `UserInfo` never carries a
-/// secret key, so the plan dropped every user and a status change or secret
-/// rotation committed while a peer was unreachable never reached it. The
-/// credentials now come from a separate store read; SRInfo stays secret-free.
-#[test]
-fn test_bootstrap_plan_carries_users_from_the_credential_snapshot() {
-    let mut info = SRInfo::default();
-    // Exactly what `list_users` builds: status, policy, updated_at — never secret_key.
-    info.user_info_map.insert(
-        "alice".to_string(),
-        rustfs_madmin::UserInfo {
-            status: rustfs_madmin::AccountStatus::Disabled,
-            policy_name: Some("readwrite".to_string()),
-            updated_at: Some(OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("timestamp")),
-            ..Default::default()
-        },
-    );
-    info.user_info_map.insert(
-        "external-idp-user".to_string(),
-        rustfs_madmin::UserInfo {
-            status: rustfs_madmin::AccountStatus::Enabled,
-            ..Default::default()
-        },
-    );
-    let user_updated_at = OffsetDateTime::from_unix_timestamp(1_700_000_500).expect("timestamp");
-    let mut credentials = SiteReplicationIamCredentials::default();
-    credentials.users.insert(
-        "alice".to_string(),
-        SiteReplicationUserCredential {
-            secret_key: "alice-secret".to_string(),
-            status: rustfs_madmin::AccountStatus::Disabled,
-            updated_at: Some(user_updated_at),
-        },
-    );
-
-    let plan = site_replication_bootstrap_plan(&info, &credentials).expect("bootstrap plan should build");
-
-    let users: Vec<_> = plan.iam_items.iter().filter(|item| item.r#type == "iam-user").collect();
-    assert_eq!(users.len(), 1, "only the user with a credential travels: {:?}", plan.iam_items);
-    let alice = users[0].iam_user.as_ref().expect("iam user body");
-    assert_eq!(alice.access_key, "alice");
-    let req = alice.user_req.as_ref().expect("user request");
-    assert_eq!(req.secret_key, "alice-secret");
-    assert_eq!(req.status, rustfs_madmin::AccountStatus::Disabled);
-    assert_eq!(req.policy.as_deref(), Some("readwrite"));
-    // the user record's own axis, not the policy-mapping time list_users reports
-    assert_eq!(users[0].updated_at, Some(user_updated_at));
-}
-
-fn service_account_snapshot(access_key: &str, parent: &str, status: &str) -> SiteReplicationServiceAccountSnapshot {
-    SiteReplicationServiceAccountSnapshot {
-        create: rustfs_madmin::SRSvcAccCreate {
-            parent: parent.to_string(),
-            access_key: access_key.to_string(),
-            secret_key: format!("{access_key}-secret"),
-            groups: Vec::new(),
-            claims: HashMap::new(),
-            session_policy: SRSessionPolicy::default(),
-            status: status.to_string(),
-            name: String::new(),
-            description: String::new(),
-            expiration: None,
-            api_version: Some(SITE_REPL_API_VERSION.to_string()),
-        },
-        envelope: None,
-        updated_at: Some(OffsetDateTime::from_unix_timestamp(1_700_000_600).expect("timestamp")),
-    }
-}
-
-/// backlog#2289: service accounts were absent from every snapshot (the
-/// listing filters them). They now travel as the create item the live hook
-/// emits — after their parents — carrying secret and status.
-#[test]
-fn test_bootstrap_plan_emits_service_accounts_after_their_parents() {
-    let mut info = SRInfo::default();
-    info.user_info_map
-        .insert("alice".to_string(), rustfs_madmin::UserInfo::default());
-    let mut credentials = SiteReplicationIamCredentials::default();
-    credentials.users.insert(
-        "alice".to_string(),
-        SiteReplicationUserCredential {
-            secret_key: "alice-secret".to_string(),
-            status: rustfs_madmin::AccountStatus::Enabled,
-            updated_at: None,
-        },
-    );
-    credentials
-        .service_accounts
-        .push(service_account_snapshot("alice-svc", "alice", "off"));
-
-    let plan = site_replication_bootstrap_plan(&info, &credentials).expect("bootstrap plan should build");
-
-    let types: Vec<_> = plan.iam_items.iter().map(|item| item.r#type.as_str()).collect();
-    assert_eq!(types, vec!["iam-user", "service-account"]);
-    let change = plan.iam_items[1].svc_acc_change.as_ref().expect("service account change");
-    let create = change.create.as_ref().expect("create body");
-    assert_eq!((create.access_key.as_str(), create.parent.as_str()), ("alice-svc", "alice"));
-    assert_eq!(create.secret_key, "alice-svc-secret");
-    assert_eq!(create.status, "off", "a disabled account must arrive disabled");
-    assert!(change.delete.is_none() && change.update.is_none());
-}
-
-/// A service account present in the previous snapshot but gone from the
-/// fresh one is replayed as an explicit delete, like the other IAM kinds.
-#[test]
-fn test_retry_snapshot_tombstones_removed_service_accounts() {
-    let observed_at = OffsetDateTime::from_unix_timestamp(1_700_001_000).expect("timestamp");
-    let mut info = SRInfo::default();
-    info.user_info_map
-        .insert("alice".to_string(), rustfs_madmin::UserInfo::default());
-    let mut credentials = SiteReplicationIamCredentials::default();
-    credentials.users.insert(
-        "alice".to_string(),
-        SiteReplicationUserCredential {
-            secret_key: "alice-secret".to_string(),
-            status: rustfs_madmin::AccountStatus::Enabled,
-            updated_at: None,
-        },
-    );
-    let mut with_account = credentials.clone();
-    with_account
-        .service_accounts
-        .push(service_account_snapshot("alice-svc", "alice", "on"));
-    let previous = site_replication_bootstrap_plan(&info, &with_account).expect("previous plan");
-    let fresh = site_replication_bootstrap_plan(&info, &credentials).expect("fresh plan");
-
-    let replay = RetrySnapshot::replay_after_change(
-        &RetrySnapshot::Iam(previous.iam_items),
-        &RetrySnapshot::Iam(fresh.iam_items),
-        observed_at,
-    );
-    let RetrySnapshot::Iam(items) = replay else {
-        panic!("IAM snapshot expected");
-    };
-    let tombstone = items
-        .iter()
-        .find(|item| item.r#type == "service-account")
-        .expect("service account tombstone");
-    let change = tombstone.svc_acc_change.as_ref().expect("change");
-    assert_eq!(change.delete.as_ref().map(|delete| delete.access_key.as_str()), Some("alice-svc"));
-    assert!(change.create.is_none());
-    assert_eq!(tombstone.updated_at, Some(observed_at));
-}
-
-/// Spawns a one-shot HTTP peer that answers 200 and flips the returned flag
-/// once a request head has arrived.
-async fn spawn_reached_probe_peer() -> (String, Arc<AtomicBool>, tokio::task::JoinHandle<()>) {
-    let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind healthy peer");
-    let endpoint = format!("http://{}", listener.local_addr().expect("healthy peer address"));
-    let reached = Arc::new(AtomicBool::new(false));
-    let reached_by_server = reached.clone();
-    let server = tokio::spawn(async move {
-        let Ok((mut stream, _)) = listener.accept().await else {
-            return;
-        };
-        let mut request = Vec::new();
-        let mut buffer = [0_u8; 1024];
-        loop {
-            let Ok(read) = stream.read(&mut buffer).await else {
-                return;
-            };
-            if read == 0 {
-                return;
-            }
-            request.extend_from_slice(&buffer[..read]);
-            if request.windows(4).any(|window| window == b"\r\n\r\n") {
-                break;
-            }
-        }
-        reached_by_server.store(true, Ordering::SeqCst);
-        let _ = stream
-            .write_all(b"HTTP/1.1 200 OK\r\ncontent-length: 2\r\nconnection: close\r\n\r\nok")
-            .await;
-    });
-    (endpoint, reached, server)
-}
-
-/// Three-peer runtime whose local peer is `local`; BTreeMap order visits the
-/// failing peer `b` before the healthy peer `c`.
-fn broadcast_runtime_with_failing_peer_before_healthy(failing_endpoint: &str, healthy_endpoint: &str) -> SiteReplicationRuntime {
-    let local_peer = PeerInfo {
-        deployment_id: "local".to_string(),
-        ..peer("local", "http://127.0.0.1:9")
-    };
-    let mut state = SiteReplicationState {
-        name: "local".to_string(),
-        service_account_access_key: "site-replicator-0".to_string(),
-        ..Default::default()
-    };
-    state.peers.insert("local".to_string(), local_peer.clone());
-    state.peers.insert(
-        "b".to_string(),
-        PeerInfo {
-            deployment_id: "b".to_string(),
-            ..peer("b", failing_endpoint)
-        },
-    );
-    state.peers.insert(
-        "c".to_string(),
-        PeerInfo {
-            deployment_id: "c".to_string(),
-            ..peer("c", healthy_endpoint)
-        },
-    );
-    SiteReplicationRuntime {
-        state,
-        local_peer,
-        service_account_secret_key: "site-replicator-secret".to_string(),
-    }
-}
-
-const BROADCAST_PROBE_DELETE_BUCKET_PATH: &str =
-    "/rustfs/admin/v3/site-replication/peer/bucket-ops?bucket=photos&operation=delete-bucket";
-
-/// The generic JSON broadcast (bucket make/delete, bucket-meta hook, bucket
-/// ops) attempts every remote peer: a peer whose request fails must not stop
-/// delivery to the peers that follow it in deployment-id order, and the
-/// failure is still reported to the caller (backlog#2293).
-#[tokio::test]
-#[serial]
-async fn test_broadcast_json_reaches_healthy_peers_after_a_failed_peer() {
-    // Peer "b": nothing listens on the port, so the connect is refused.
-    let refused = TcpListener::bind("127.0.0.1:0").await.expect("bind refused-peer probe");
-    let refused_endpoint = format!("http://{}", refused.local_addr().expect("refused-peer address"));
-    drop(refused);
-
-    let (healthy_endpoint, reached, server) = spawn_reached_probe_peer().await;
-    let runtime = broadcast_runtime_with_failing_peer_before_healthy(&refused_endpoint, &healthy_endpoint);
-
-    let result = temp_env::async_with_vars([(ALLOW_LOOPBACK_REPLICATION_TARGET_ENV, Some("true"))], async {
-        broadcast_site_replication_json_with_runtime(&runtime, BROADCAST_PROBE_DELETE_BUCKET_PATH, &serde_json::json!({})).await
-    })
-    .await;
-
-    let err = result.expect_err("peer b refuses connections, the broadcast must report it");
-    assert!(
-        reached.load(Ordering::SeqCst),
-        "peer c never received the broadcast once peer b failed: {err}"
-    );
-    server.abort();
-}
-
-/// Same guarantee when the failing peer never gets a transport: an endpoint
-/// that `PeerTransport::for_runtime_peer` rejects must be skipped past (and
-/// reported), not abort the broadcast before the healthy peers (backlog#2293).
-#[tokio::test]
-#[serial]
-async fn test_broadcast_json_reaches_healthy_peers_after_a_peer_without_transport() {
-    // Peer "b": a scheme the peer connection validator refuses outright.
-    let forbidden_endpoint = "ftp://peer-b.example.com";
-
-    let (healthy_endpoint, reached, server) = spawn_reached_probe_peer().await;
-    let runtime = broadcast_runtime_with_failing_peer_before_healthy(forbidden_endpoint, &healthy_endpoint);
-
-    let result = temp_env::async_with_vars([(ALLOW_LOOPBACK_REPLICATION_TARGET_ENV, Some("true"))], async {
-        broadcast_site_replication_json_with_runtime(&runtime, BROADCAST_PROBE_DELETE_BUCKET_PATH, &serde_json::json!({})).await
-    })
-    .await;
-
-    let err = result.expect_err("peer b has no usable transport, the broadcast must report it");
-    assert!(
-        err.to_string().contains("invalid persisted site replication peer"),
-        "the reported error must be peer b's transport failure: {err}"
-    );
-    assert!(
-        reached.load(Ordering::SeqCst),
-        "peer c never received the broadcast once peer b failed to get a transport: {err}"
-    );
-    server.abort();
-}
-
-fn service_account_item_with_claims(order: &[&str]) -> SRIAMItem {
-    let mut claims = HashMap::new();
-    for key in order {
-        claims.insert((*key).to_string(), serde_json::json!(format!("value-of-{key}")));
-    }
-    SRIAMItem {
-        r#type: "service-account".to_string(),
-        svc_acc_change: Some(SRSvcAccChange {
-            create: Some(rustfs_madmin::SRSvcAccCreate {
-                parent: "alice".to_string(),
-                access_key: "alice-svc".to_string(),
-                secret_key: "alice-svc-secret".to_string(),
-                groups: Vec::new(),
-                claims,
-                session_policy: SRSessionPolicy::default(),
-                status: "on".to_string(),
-                name: String::new(),
-                description: String::new(),
-                expiration: None,
-                api_version: Some(SITE_REPL_API_VERSION.to_string()),
-            }),
-            api_version: Some(SITE_REPL_API_VERSION.to_string()),
-            ..Default::default()
-        }),
-        updated_at: Some(OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("timestamp")),
-        api_version: Some(SITE_REPL_API_VERSION.to_string()),
-        ..Default::default()
-    }
-}
-
-/// The repair preflight token and the retry-snapshot fingerprint hash the
-/// serialized items. Service-account claims live in a `HashMap`, whose
-/// iteration order differs between instances, so the hash must not depend on
-/// it (the real-VM repair returned 412 "preflight is stale" between dry-run
-/// and execute once snapshots carried service accounts).
-#[test]
-fn test_repair_task_id_and_retry_fingerprint_ignore_claim_map_order() {
-    let forward = service_account_item_with_claims(&["accessKey", "exp", "parent", "sa-policy", "sub", "tenant"]);
-    let backward = service_account_item_with_claims(&["tenant", "sub", "sa-policy", "parent", "exp", "accessKey"]);
-
-    let canonical = canonical_json_vec(&forward).expect("canonical json");
-    let text = String::from_utf8(canonical).expect("utf8");
-    let positions: Vec<usize> = [
-        "\"accessKey\"",
-        "\"exp\"",
-        "\"parent\"",
-        "\"sa-policy\"",
-        "\"sub\"",
-        "\"tenant\"",
-    ]
-    .iter()
-    .map(|key| text.find(key).expect("claim key present"))
-    .collect();
-    assert!(
-        positions.windows(2).all(|pair| pair[0] < pair[1]),
-        "claim keys must serialize sorted: {text}"
-    );
-
-    assert_eq!(
-        SiteReplicationRepairTask::Iam(&forward).id().expect("id"),
-        SiteReplicationRepairTask::Iam(&backward).id().expect("id"),
-        "identical items must yield the same repair task id regardless of claim map order"
-    );
-    assert_eq!(
-        RetrySnapshot::Iam(vec![forward]).fingerprint().expect("fingerprint"),
-        RetrySnapshot::Iam(vec![backward]).fingerprint().expect("fingerprint"),
-        "identical snapshots must fingerprint equal regardless of claim map order"
-    );
-}

rustfs/src/site_replication/transport.rs

@@ -876,14 +876,6 @@ pub(crate) async fn broadcast_site_replication_json<T: Serialize>(path: &str, bo
     broadcast_site_replication_json_with_runtime(&runtime, path, body).await
 }
 
-/// PUT `body` to `path` on every remote peer of the runtime.
-///
-/// Every peer is attempted: one peer's failure — transport construction
-/// included — must not skip the peers that follow it in deployment-id order,
-/// or they silently miss the change with no retry record (backlog#2293). A
-/// success settles the peer/path's queued retry event, a failure enqueues one
-/// under the request `path` (so the drain classifies it as today), and the
-/// first error is returned once all peers were attempted.
 pub(crate) async fn broadcast_site_replication_json_with_runtime<T: Serialize>(
     runtime: &SiteReplicationRuntime,
     path: &str,
@@ -891,30 +883,20 @@ pub(crate) async fn broadcast_site_replication_json_with_runtime<T: Serialize>(
 ) -> S3Result<()> {
     let state = &runtime.state;
     let local_peer = &runtime.local_peer;
-    let mut first_error: Option<S3Error> = None;
 
     for peer in state.peers.values() {
         if peer.deployment_id == local_peer.deployment_id || same_identity_endpoint(&peer.endpoint, &local_peer.endpoint) {
             continue;
         }
 
-        let sent = match PeerTransport::for_runtime_peer(peer).await {
-            Ok(transport) => PeerAdminRequest::put(&transport.connection, path, &state.service_account_access_key)
-                .with_client(&transport.client)
-                .send_with_retry_event(peer, &runtime.service_account_secret_key, body)
-                .await
-                .map(|_| ()),
-            Err(err) => {
-                enqueue_site_replication_retry_event(peer, path, &err).await;
-                Err(err)
-            }
-        };
-        if let Err(err) = sent {
-            first_error.get_or_insert(err);
-        }
+        let transport = PeerTransport::for_runtime_peer(peer).await?;
+        PeerAdminRequest::put(&transport.connection, path, &state.service_account_access_key)
+            .with_client(&transport.client)
+            .send_with_retry_event(peer, &runtime.service_account_secret_key, body)
+            .await?;
     }
 
-    first_error.map_or(Ok(()), Err)
+    Ok(())
 }
 
 pub(crate) fn parse_endpoint_refresh_status(peer: &PeerInfo, body: &[u8]) -> S3Result<()> {