Merge remote-tracking branch 'origin/main' into overtrue/fix/odm-application-regressions

(cherry picked from commit eb1b17802c)

.github/workflows/ci.yml

@@ -850,6 +850,11 @@ jobs:
           cache-save-if: 'false'
           install-build-packaging-tools: 'false'
 
+      - name: Install network fault-injection tools
+        run: |
+          sudo apt-get install -y iptables
+          sudo -n iptables --version
+
       - name: Set up Python
         uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6.3.0
         with:

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/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,14 +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.
-        // `updated_at` is the record's stamp only; the cache is published
-        // with the local clock, because `LockedCache::exec` drops a write
-        // whose time predates the entity's load time — a replicated edit
-        // whose source time is older than this node's startup would
-        // otherwise never reach the cache.
         let gi = match cache.groups.get(group) {
             Some(res) => {
                 let mut gi = res.clone();
@@ -1803,20 +1701,15 @@ where
                 uniq_set.extend(members.iter().cloned());
 
                 gi.members = uniq_set.into_iter().collect();
-                gi.update_at = Some(updated_at);
-                gi
-            }
-            None => {
-                let mut gi = GroupInfo::new(members.clone());
-                gi.update_at = Some(updated_at);
                 gi
             }
+            None => GroupInfo::new(members.clone()),
         };
         drop(cache);
 
         self.api.save_group_info(group, gi.clone()).await?;
 
-        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);
 
@@ -1826,18 +1719,13 @@ where
                 m.insert(group.to_string());
                 cache.add_or_update_user_group_membership(member, &m, now);
             });
+            now
         });
 
-        Ok(updated_at)
+        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);
         }
@@ -1855,15 +1743,12 @@ where
         } else {
             gi.status = STATUS_DISABLED.to_owned();
         }
-        gi.update_at = Some(updated_at);
 
         self.api.save_group_info(name, gi.clone()).await?;
 
-        // Cache publication time is the local clock, not the record stamp
-        // (see `add_users_to_group_at`).
         self.cache.add_or_update_group(name, &gi, OffsetDateTime::now_utc());
 
-        Ok(updated_at)
+        Ok(OffsetDateTime::now_utc())
     }
 
     pub async fn get_group_description(&self, name: &str) -> Result<GroupDesc> {
@@ -1933,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
@@ -1959,14 +1830,12 @@ where
         let s: HashSet<&String> = HashSet::from_iter(gi.members.iter());
         let d: HashSet<&String> = HashSet::from_iter(members.iter());
         gi.members = s.difference(&d).map(|v| v.to_string()).collect::<Vec<String>>();
-        gi.update_at = Some(updated_at);
+
         if !update_cache_only {
             self.api.save_group_info(name, gi.clone()).await?;
         }
 
-        self.cache.with_write_lock(|cache| {
-            // Sample after storage completes so a concurrent reload cannot
-            // make this publication older than the cache it must update.
+        let now = self.cache.with_write_lock(|cache| {
             let now = OffsetDateTime::now_utc();
             cache.add_or_update_group(name, &gi, now);
 
@@ -1978,25 +1847,13 @@ where
                     cache.add_or_update_user_group_membership(member, &m, now);
                 }
             });
+            now
         });
 
-        Ok(updated_at)
+        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);
         }
@@ -2045,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) {
@@ -2377,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 {
@@ -2210,9 +2081,6 @@ mod tests {
         block_delete: Arc<std::sync::atomic::AtomicBool>,
         delete_started: Arc<tokio::sync::Notify>,
         release_delete: Arc<tokio::sync::Notify>,
-        block_group_save: Arc<std::sync::atomic::AtomicBool>,
-        group_save_started: Arc<tokio::sync::Notify>,
-        group_save_release: Arc<tokio::sync::Notify>,
     }
 
     impl StsTestMockStore {
@@ -2226,9 +2094,6 @@ mod tests {
                 block_delete: Arc::new(std::sync::atomic::AtomicBool::new(false)),
                 delete_started: Arc::new(tokio::sync::Notify::new()),
                 release_delete: Arc::new(tokio::sync::Notify::new()),
-                block_group_save: Arc::new(std::sync::atomic::AtomicBool::new(false)),
-                group_save_started: Arc::new(tokio::sync::Notify::new()),
-                group_save_release: Arc::new(tokio::sync::Notify::new()),
             }
         }
 
@@ -2332,15 +2197,11 @@ mod tests {
         }
 
         async fn save_group_info(&self, _name: &str, _item: GroupInfo) -> Result<()> {
-            if self.block_group_save.load(std::sync::atomic::Ordering::SeqCst) {
-                self.group_save_started.notify_one();
-                self.group_save_release.notified().await;
-            }
-            Ok(())
+            Err(Error::InvalidArgument)
         }
 
         async fn delete_group_info(&self, _name: &str) -> Result<()> {
-            Ok(())
+            Err(Error::InvalidArgument)
         }
 
         async fn load_group(&self, name: &str, m: &mut HashMap<String, GroupInfo>) -> Result<()> {
@@ -2517,140 +2378,6 @@ mod tests {
         IamSys::new(cache)
     }
 
-    async fn assert_group_write_during_reload_is_published(remove: bool) {
-        let iam_sys = Arc::new(temp_env::async_with_vars([("RUSTFS_SKIP_BACKGROUND_TASK", Some("1"))], test_iam_sys()).await);
-        let member = "sts-fallback-test-parent";
-        let group = if remove { "testgroup" } else { "new-published-group" };
-        let source_time = OffsetDateTime::now_utc() - time::Duration::hours(1);
-        iam_sys
-            .store
-            .api
-            .block_group_save
-            .store(true, std::sync::atomic::Ordering::SeqCst);
-        let before = iam_sys.store.cache.snapshot();
-        let writer_iam = iam_sys.clone();
-        let writer = tokio::spawn(async move {
-            if remove {
-                writer_iam
-                    .remove_users_from_group_at(group, vec![member.to_string()], source_time)
-                    .await
-            } else {
-                writer_iam
-                    .add_users_to_group_at(group, vec![member.to_string()], source_time)
-                    .await
-            }
-        });
-        tokio::time::timeout(std::time::Duration::from_secs(5), iam_sys.store.api.group_save_started.notified())
-            .await
-            .expect("group save should reach the barrier");
-        // The pending store write has not changed the cache, so the production
-        // full-reload snapshot guard permits this replacement.
-        assert!(iam_sys.store.cache.with_write_lock(|cache| cache.matches_snapshot(&before)));
-        iam_sys
-            .store
-            .api
-            .load_all(&iam_sys.store.cache)
-            .await
-            .expect("reload while group save is pending");
-        iam_sys.store.api.group_save_release.notify_one();
-        assert_eq!(writer.await.expect("join group writer").expect("group write should succeed"), source_time);
-        let info = iam_sys
-            .get_group_info(group)
-            .await
-            .expect("successful group write must remain readable after reload");
-        assert_eq!(info.update_at, Some(source_time), "source timestamp must remain on the record");
-        assert_eq!(info.members, if remove { Vec::new() } else { vec![member.to_string()] });
-        let groups = iam_sys.store.cache.snapshot().user_group_memberships.get(member).cloned();
-        assert_eq!(
-            groups.is_some_and(|groups| groups.contains(group)),
-            !remove,
-            "membership index must reflect the write"
-        );
-    }
-
-    #[tokio::test]
-    #[serial]
-    async fn add_group_write_during_reload_publishes_after_store_save() {
-        assert_group_write_during_reload_is_published(false).await;
-    }
-
-    #[tokio::test]
-    #[serial]
-    async fn remove_group_write_during_reload_publishes_after_store_save() {
-        assert_group_write_during_reload_is_published(true).await;
-    }
-
-    /// Review finding on rustfs#7195: a replicated group edit carries a source
-    /// stamp that may predate this node's cache load time. The stamp belongs on
-    /// the record only; publishing the cache with it makes `LockedCache::exec`
-    /// drop the write, so the group is written to the store but unreadable
-    /// here and the receiver's next `set_group_status_at` fails with
-    /// `NoSuchGroup`. Add, status and removal must all publish with the local
-    /// clock while keeping the source stamp on `GroupInfo::update_at`.
-    #[tokio::test]
-    async fn group_writes_stamped_before_the_cache_load_time_still_publish() {
-        let iam_sys = test_iam_sys().await;
-        let member = "group-stamp-member";
-        let identity = UserIdentity {
-            version: 1,
-            credentials: Credentials {
-                access_key: member.to_string(),
-                secret_key: "longenoughsecret".to_string(),
-                status: "on".to_string(),
-                ..Default::default()
-            },
-            update_at: Some(OffsetDateTime::now_utc()),
-        };
-        iam_sys.store.cache.with_write_lock(|cache| {
-            cache.add_or_update_user(member, &identity, OffsetDateTime::now_utc());
-            // The startup load publishes every entity with the load time.
-            cache.replace_groups(CacheEntity::new(HashMap::new()));
-            cache.replace_user_group_memberships(CacheEntity::new(HashMap::new()));
-        });
-
-        let group = "group-stamp";
-        let source_time = OffsetDateTime::now_utc() - time::Duration::hours(1);
-        let stamped = iam_sys
-            .add_users_to_group_at(group, vec![member.to_string()], source_time)
-            .await
-            .expect("add members with a source stamp older than the cache load");
-        assert_eq!(stamped, source_time, "the returned stamp is the source time");
-        let info = iam_sys
-            .get_group_info(group)
-            .await
-            .expect("the group must be readable right after the add");
-        assert_eq!(info.members, vec![member.to_string()]);
-        assert_eq!(info.update_at, Some(source_time), "the record keeps the source stamp");
-        let memberships = iam_sys.store.cache.snapshot().user_group_memberships.get(member).cloned();
-        assert!(
-            memberships.is_some_and(|groups| groups.contains(group)),
-            "the membership index is published too"
-        );
-
-        let disabled_at = source_time + time::Duration::seconds(1);
-        iam_sys
-            .set_group_status_at(group, false, disabled_at)
-            .await
-            .expect("status change with a source stamp older than the cache load");
-        let info = iam_sys.get_group_info(group).await.expect("group after status change");
-        assert_eq!(info.status, "disabled");
-        assert_eq!(info.update_at, Some(disabled_at));
-
-        let removed_at = source_time + time::Duration::seconds(2);
-        iam_sys
-            .remove_users_from_group_at(group, vec![member.to_string()], removed_at)
-            .await
-            .expect("removal with a source stamp older than the cache load");
-        let info = iam_sys.get_group_info(group).await.expect("group after removal");
-        assert!(info.members.is_empty(), "the removal must be visible in the cache");
-        assert_eq!(info.update_at, Some(removed_at));
-        let memberships = iam_sys.store.cache.snapshot().user_group_memberships.get(member).cloned();
-        assert!(
-            !memberships.is_some_and(|groups| groups.contains(group)),
-            "the membership index follows the removal"
-        );
-    }
-
     fn service_account_opts(access_key: &str, secret_key: &str) -> NewServiceAccountOpts {
         NewServiceAccountOpts {
             access_key: access_key.to_string(),

crates/madmin/fixtures/on_demand_migration/get_response.json

@@ -1 +1 @@
-{"bucket":"photos","config":{"version":1,"enabled":true,"source":{"provider":"minio","endpoint":"https://source.example.com:9000","region":"us-east-1","bucket":"legacy-photos","path_style":"auto","credentials":{"access_key":"AKIASOURCE","secret_key":"REDACTED","session_token":null},"tls":{"skip_verify":false,"ca_cert_pem":null},"azure":null,"gcs":null},"filter":{"prefix":null,"source_prefix":"photos/"},"policy":{"head":"proxy","range_get":"serve_and_backfill","source_error":"propagate","list_through":false,"respect_local_delete_marker":true,"preserve_etag":true,"copy_tags":false,"emit_events":true,"negative_cache_ttl_secs":30,"inline_max_bytes":16777216,"multipart_part_size_bytes":67108864,"max_concurrent_pulls":8,"pull_queue_capacity":1024,"source_timeout":{"connect_ms":5000,"first_byte_ms":15000,"idle_ms":30000},"bandwidth_limit_bytes_per_sec":null}},"updated_at":"2026-09-02T10:00:00Z"}
+{"bucket":"photos","config":{"version":1,"enabled":true,"source":{"provider":"minio","endpoint":"https://source.example.com:9000","region":"us-east-1","bucket":"legacy-photos","path_style":"auto","credentials":{"access_key":"AKIASOURCE","secret_key":"REDACTED","session_token":null},"tls":{"skip_verify":false,"ca_cert_pem":null}},"filter":{"prefix":null,"source_prefix":"photos/"},"policy":{"head":"proxy","range_get":"serve_and_backfill","source_error":"propagate","list_through":false,"respect_local_delete_marker":true,"preserve_etag":true,"copy_tags":false,"emit_events":true,"negative_cache_ttl_secs":30,"inline_max_bytes":16777216,"multipart_part_size_bytes":67108864,"max_concurrent_pulls":8,"pull_queue_capacity":1024,"source_timeout":{"connect_ms":5000,"first_byte_ms":15000,"idle_ms":30000},"bandwidth_limit_bytes_per_sec":null}},"updated_at":"2026-09-02T10:00:00Z"}

crates/madmin/fixtures/on_demand_migration/set_request.json

@@ -1 +1 @@
-{"version":1,"enabled":true,"source":{"provider":"minio","endpoint":"https://source.example.com:9000","region":"us-east-1","bucket":"legacy-photos","path_style":"auto","credentials":{"access_key":"AKIASOURCE","secret_key":"sourceSecretKey123","session_token":null},"tls":{"skip_verify":false,"ca_cert_pem":null},"azure":null,"gcs":null},"filter":{"prefix":null,"source_prefix":"photos/"},"policy":{"head":"proxy","range_get":"serve_and_backfill","source_error":"propagate","list_through":false,"respect_local_delete_marker":true,"preserve_etag":true,"copy_tags":false,"emit_events":true,"negative_cache_ttl_secs":30,"inline_max_bytes":16777216,"multipart_part_size_bytes":67108864,"max_concurrent_pulls":8,"pull_queue_capacity":1024,"source_timeout":{"connect_ms":5000,"first_byte_ms":15000,"idle_ms":30000},"bandwidth_limit_bytes_per_sec":null}}
+{"version":1,"enabled":true,"source":{"provider":"minio","endpoint":"https://source.example.com:9000","region":"us-east-1","bucket":"legacy-photos","path_style":"auto","credentials":{"access_key":"AKIASOURCE","secret_key":"sourceSecretKey123","session_token":null},"tls":{"skip_verify":false,"ca_cert_pem":null}},"filter":{"prefix":null,"source_prefix":"photos/"},"policy":{"head":"proxy","range_get":"serve_and_backfill","source_error":"propagate","list_through":false,"respect_local_delete_marker":true,"preserve_etag":true,"copy_tags":false,"emit_events":true,"negative_cache_ttl_secs":30,"inline_max_bytes":16777216,"multipart_part_size_bytes":67108864,"max_concurrent_pulls":8,"pull_queue_capacity":1024,"source_timeout":{"connect_ms":5000,"first_byte_ms":15000,"idle_ms":30000},"bandwidth_limit_bytes_per_sec":null}}

crates/madmin/fixtures/on_demand_migration/set_response.json

@@ -1 +1 @@
-{"bucket":"photos","dry_run":false,"config":{"version":1,"enabled":true,"source":{"provider":"minio","endpoint":"https://source.example.com:9000","region":"us-east-1","bucket":"legacy-photos","path_style":"auto","credentials":{"access_key":"AKIASOURCE","secret_key":"REDACTED","session_token":null},"tls":{"skip_verify":false,"ca_cert_pem":null},"azure":null,"gcs":null},"filter":{"prefix":null,"source_prefix":"photos/"},"policy":{"head":"proxy","range_get":"serve_and_backfill","source_error":"propagate","list_through":false,"respect_local_delete_marker":true,"preserve_etag":true,"copy_tags":false,"emit_events":true,"negative_cache_ttl_secs":30,"inline_max_bytes":16777216,"multipart_part_size_bytes":67108864,"max_concurrent_pulls":8,"pull_queue_capacity":1024,"source_timeout":{"connect_ms":5000,"first_byte_ms":15000,"idle_ms":30000},"bandwidth_limit_bytes_per_sec":null}},"updated_at":"2026-09-02T10:00:00Z","probe":{"reachable":true,"listable":true,"sample_key":"photos/2024/01.jpg"}}
+{"bucket":"photos","dry_run":false,"config":{"version":1,"enabled":true,"source":{"provider":"minio","endpoint":"https://source.example.com:9000","region":"us-east-1","bucket":"legacy-photos","path_style":"auto","credentials":{"access_key":"AKIASOURCE","secret_key":"REDACTED","session_token":null},"tls":{"skip_verify":false,"ca_cert_pem":null}},"filter":{"prefix":null,"source_prefix":"photos/"},"policy":{"head":"proxy","range_get":"serve_and_backfill","source_error":"propagate","list_through":false,"respect_local_delete_marker":true,"preserve_etag":true,"copy_tags":false,"emit_events":true,"negative_cache_ttl_secs":30,"inline_max_bytes":16777216,"multipart_part_size_bytes":67108864,"max_concurrent_pulls":8,"pull_queue_capacity":1024,"source_timeout":{"connect_ms":5000,"first_byte_ms":15000,"idle_ms":30000},"bandwidth_limit_bytes_per_sec":null}},"updated_at":"2026-09-02T10:00:00Z","probe":{"reachable":true,"listable":true,"sample_key":"photos/2024/01.jpg"}}

crates/madmin/fixtures/on_demand_migration/source_config_e2a.rs

@@ -0,0 +1,80 @@
+// Strict source reader frozen from e2a921bc1608823c8efec955d7463ab8350a8a01.
+// Wire declarations and credential Debug are copied verbatim; runtime methods are omitted.
+use serde::{Deserialize, Serialize};
+use std::fmt;
+
+const REDACTED: &str = "REDACTED";
+
+/// The external S3-compatible source bucket.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct SourceConfig {
+    pub provider: Provider,
+    /// `http(s)://host[:port]` with no path or query. Optional only for
+    /// [`Provider::Aws`], where it is derived from `region`.
+    #[serde(default)]
+    pub endpoint: Option<String>,
+    pub region: String,
+    pub bucket: String,
+    #[serde(default)]
+    pub path_style: PathStyle,
+    /// `None` means anonymous access to a public source bucket.
+    #[serde(default)]
+    pub credentials: Option<SourceCredentials>,
+    #[serde(default)]
+    pub tls: TlsConfig,
+}
+
+/// Source vendor family. `azure` is deliberately absent from this version.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum Provider {
+    /// Generic S3-compatible endpoint.
+    S3,
+    Aws,
+    Minio,
+    Rustfs,
+    R2,
+    /// GCS XML interoperability API with HMAC keys.
+    Gcs,
+}
+
+/// Bucket addressing style. `auto` is resolved by the source client builder.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum PathStyle {
+    #[default]
+    Auto,
+    Path,
+    Virtual,
+}
+
+/// Static credentials for the source. `Debug` never prints the secret or
+/// the session token.
+#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct SourceCredentials {
+    pub access_key: String,
+    pub secret_key: String,
+    #[serde(default)]
+    pub session_token: Option<String>,
+}
+
+impl fmt::Debug for SourceCredentials {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("SourceCredentials")
+            .field("access_key", &self.access_key)
+            .field("secret_key", &REDACTED)
+            .field("session_token", &self.session_token.as_ref().map(|_| REDACTED))
+            .finish()
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct TlsConfig {
+    #[serde(default)]
+    pub skip_verify: bool,
+    #[serde(default)]
+    pub ca_cert_pem: Option<String>,
+}

crates/madmin/src/on_demand_migration.rs

@@ -85,10 +85,10 @@ pub struct OnDemandMigrationSource {
     #[serde(default)]
     pub tls: OnDemandMigrationTls,
     /// Required for `azure` and rejected for every other provider.
-    #[serde(default)]
+    #[serde(default, skip_serializing_if = "Option::is_none")]
     pub azure: Option<OnDemandMigrationAzure>,
     /// Required for `gcs_native` and rejected for every other provider.
-    #[serde(default)]
+    #[serde(default, skip_serializing_if = "Option::is_none")]
     pub gcs: Option<OnDemandMigrationGcs>,
 }
 
@@ -651,6 +651,10 @@ mod tests {
     use super::*;
     use crate::test_support::TestServer;
 
+    mod before_native_sources {
+        include!("../fixtures/on_demand_migration/source_config_e2a.rs");
+    }
+
     const SET_REQUEST_FIXTURE: &str = include_str!("../fixtures/on_demand_migration/set_request.json");
     const SET_RESPONSE_FIXTURE: &str = include_str!("../fixtures/on_demand_migration/set_response.json");
     const GET_RESPONSE_FIXTURE: &str = include_str!("../fixtures/on_demand_migration/get_response.json");
@@ -684,6 +688,20 @@ mod tests {
         assert_eq!(config.source.tls, OnDemandMigrationTls::default());
     }
 
+    #[test]
+    fn s3_admin_writes_remain_readable_by_the_strict_pre_native_server() {
+        for provider in ["s3", "aws", "minio", "rustfs", "r2", "gcs"] {
+            let historical = SET_REQUEST_FIXTURE.replace("\"provider\":\"minio\"", &format!("\"provider\":\"{provider}\""));
+            let config: OnDemandMigrationConfig = serde_json::from_str(&historical).expect("historical set request");
+            let wire = serde_json::to_string(&config).expect("current admin set request");
+            let actual: serde_json::Value = serde_json::from_str(&wire).expect("admin request JSON");
+            let old_source: before_native_sources::SourceConfig = serde_json::from_value(actual["source"].clone())
+                .expect("the strict e2a server must accept an ordinary S3 source from the new admin client");
+            assert_eq!(serde_json::to_value(old_source).expect("old source wire"), actual["source"]);
+            assert_eq!(wire, historical.trim(), "provider={provider}: preserve the historical request bytes");
+        }
+    }
+
     #[test]
     fn set_response_fixture_round_trips_and_is_redacted() {
         let response: OnDemandMigrationSetResponse = round_trip(SET_RESPONSE_FIXTURE);
@@ -878,11 +896,11 @@ mod tests {
         for (label, json) in [
             (
                 "azure",
-                r#"{"provider":"azure","endpoint":null,"region":"auto","bucket":"legacy-photos","path_style":"auto","credentials":null,"tls":{"skip_verify":false,"ca_cert_pem":null},"azure":{"account":"legacyaccount","account_key":null,"sas_token":"sv=2021-08-06&sig=topsecret"},"gcs":null}"#,
+                r#"{"provider":"azure","endpoint":null,"region":"auto","bucket":"legacy-photos","path_style":"auto","credentials":null,"tls":{"skip_verify":false,"ca_cert_pem":null},"azure":{"account":"legacyaccount","account_key":null,"sas_token":"sv=2021-08-06&sig=topsecret"}}"#,
             ),
             (
                 "gcs_native",
-                r#"{"provider":"gcs_native","endpoint":null,"region":"auto","bucket":"legacy-photos","path_style":"auto","credentials":null,"tls":{"skip_verify":false,"ca_cert_pem":null},"azure":null,"gcs":{"service_account_json":"{\"type\":\"service_account\"}"}}"#,
+                r#"{"provider":"gcs_native","endpoint":null,"region":"auto","bucket":"legacy-photos","path_style":"auto","credentials":null,"tls":{"skip_verify":false,"ca_cert_pem":null},"gcs":{"service_account_json":"{\"type\":\"service_account\"}"}}"#,
             ),
         ] {
             let source: OnDemandMigrationSource = serde_json::from_str(json).unwrap_or_else(|err| panic!("{label}: {err}"));
@@ -891,6 +909,16 @@ mod tests {
                 json,
                 "{label} must reproduce the server wire shape byte for byte"
             );
+            let mut wire: serde_json::Value = serde_json::from_str(json).expect("native wire fixture");
+            assert!(
+                serde_json::from_value::<before_native_sources::SourceConfig>(wire.clone()).is_err(),
+                "native provider names and fields still require an upgraded server"
+            );
+            wire[if label == "azure" { "gcs" } else { "azure" }] = serde_json::Value::Null;
+            assert_eq!(
+                serde_json::from_value::<OnDemandMigrationSource>(wire).expect("the prior explicit-null wire still decodes"),
+                source
+            );
         }
 
         let azure = OnDemandMigrationAzure {
@@ -945,6 +973,10 @@ mod tests {
                 .is_some_and(|auth| auth.starts_with("AWS4-HMAC-SHA256"))
         );
         assert_eq!(request.body, SET_REQUEST_FIXTURE.trim(), "the body is the canonical config document");
+        let body: serde_json::Value = serde_json::from_str(&request.body).expect("signed admin request JSON");
+        let old_source: before_native_sources::SourceConfig = serde_json::from_value(body["source"].clone())
+            .expect("the strict pre-native server must accept the actual signed PUT source");
+        assert_eq!(old_source.provider, before_native_sources::Provider::Minio);
     }
 
     #[tokio::test]

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]

docs/architecture/compat-cleanup-register.md

@@ -11,6 +11,7 @@
 
 ## Open Items
 
+- `odm-list-bare-envelope` historical ODM continuation tokens: preserve complete bare v1/v2 envelopes and default bare issuance while framed readers deploy. Remove the legacy classifier and default-off framing issuance gate only after every supported reader accepts framing and outstanding bare listings have drained or clients explicitly restarted them; tokens have no automatic expiry. Exact full-envelope object keys remain intrinsically ambiguous during this compatibility period.
 - `backlog-2263` legacy heal MRF inspection: retained per-record journals remain readable while committed-snapshot ownership and writer activation are staged. Remove legacy import only after all supported direct-upgrade and rollback readers understand committed snapshots and migration tooling confirms that no retained or restorable legacy journal requires it. This does not enable a new writer or change the automatic legacy consumer.
 - `backlog-1337` legacy restore orphan recovery: releases that predate the restore worker-lock marker can leave a valid operation-id and `ongoing-request="true"` after cancellation or process failure, with no durable liveness proof. New servers allow an exact, non-nil legacy generation to be superseded only when its consistently parsed request date is at least 24 hours old. Remove the clock-based legacy fallback after the minimum supported direct-upgrade release writes the v1 worker-lock marker on every restore and operators have resolved every retained pre-v1 ongoing generation.
 - `backlog-2133-tier-delete-chunk-parent` bounded tier-delete dispatch compatibility: prefixes at or below the legacy manifest limit keep the byte-compatible v1 single-manifest protocol, while larger prefixes place a chunk-parent sentinel at the original deterministic root path and use operation-scoped child manifests. Older binaries reject the sentinel and child paths, preserving the v6 sole-owner downgrade fence instead of starting a competing local delete. Remove the v1 reader and fail-closed mixed-version sentinel only after every supported rollback release validates the parent/child protocol and migration tooling confirms that no retained v1 dispatch manifest remains.

docs/operations/on-demand-migration.md

@@ -23,13 +23,20 @@ Both builds can read, redact and preserve GCS configuration. A build without `gc
 
 ## List continuation token rollout
 
-`RUSTFS_ON_DEMAND_MIGRATION_LIST_V2_TOKENS` defaults to `false`; unset or invalid boolean values also keep it off. It controls only whether a v1 listing may first issue a v2 continuation token after an empty truncated merged page. Every node with this reader support accepts existing v2 tokens and continues their budget even with the switch off. Ordinary pages that consume an object or common prefix retain the original v1 token shape.
+Two independent rollout switches default to `false`; unset or invalid boolean values also keep them off. Both are node environment variables, not bucket settings:
 
-Leave the switch off while deploying v2 reader support to every node that can receive a continuation request, including nodes behind other load-balancer routes. Then set it to `true` in each node's environment and restart those nodes to enable issuance. A v1-only binary rejects v2 with `400 InvalidArgument` before the source-error policy runs; neither `not_found` nor turning off list-through makes that old reader compatible. With issuance still off, a new v1 chain retains the existing limitation: an empty source cursor cycle spanning requests can continue indefinitely. The default rollout does not claim to fix that chain until issuance is enabled.
+- `RUSTFS_ON_DEMAND_MIGRATION_LIST_V2_TOKENS` allows a v1 listing to first issue a v2 token after an empty truncated merged page. Existing v2 tokens keep their budget even on reader-only nodes. Consuming an object/common prefix or reaching a new EOF resets the budget to v1 without changing the chain's framing.
+- `RUSTFS_ON_DEMAND_MIGRATION_LIST_FRAMED_TOKENS` allows a bare/new merged listing to first issue a NUL-prefixed JSON envelope inside the existing base64 encoding. Existing framed chains stay framed even with this switch off, including a reset to v1 and local continuation after list-through is disabled. With framing issuance off, new bare v1 output keeps its historical bytes; ordinary local listings remain unchanged. This switch does not enable the v2 budget.
 
-An active v2 budget rejects the sixteenth consecutive merged page that consumes no new object/common prefix and reaches no new end-of-list state. The first fifteen empty pages can be resumed; with the existing two-fetch-per-side limit, that interval costs at most 32 fetches per side, including the failing request. A key, common prefix, or a newly exhausted side on the sixteenth request succeeds and resets the budget. A side that was already exhausted does not reset it again. This is a resource bound, not proof of a cursor cycle: an unusually long but valid empty source-page chain also reaches the limit. Tokens are unsigned base64 JSON, so this budget applies to clients that continue with the returned token unchanged; replaying or editing a token can reset it, and it is not a malicious-client defense or a global request quota. The two-fetch-per-side request limit and existing source rate limiter still apply. A source failure follows `policy.source_error`: `propagate` returns `424 SourceUnavailable` with `invalid_pagination`; `not_found` returns the fetched local listing with `x-rustfs-on-demand-migration-list: local_only`. A blocking local-side failure returns `InternalError`, without silently discarding local entries.
+Deploy readers before enabling either writer switch. Every node serving continuation requests must understand the selected token version and framing, including nodes behind other load-balancer routes. This build reads both complete historical bare envelopes and framed v1/v2 envelopes with the same strict version/count validation. A bare-v1-only binary rejects bare v2 with `400 InvalidArgument`; an old bare reader mistakes framed input for a local marker, while a framed-only reader mistakes bare input for one. Neither format mismatch is safe: it can restart a merged scan and lose its budget rather than returning an error. The interim framed-only build from #7187 must be replaced on every serving node before a mixed-format rollout. Enable framing only after all readers support both formats; enable the budget only after all readers support v2. Restart nodes after updating their environment.
 
-For rollback, first turn issuance off on every node. Keep v2-capable readers available for outstanding v2 chains: switching issuance off does not erase their budgets, and tokens have no expiration that proves those chains have drained. Route those continuations to compatible readers or have clients explicitly restart their listings before restoring v1-only binaries. Restarting a listing is a new scan and can repeat entries. Do not roll back readers while assuming the issuance switch makes existing v2 tokens disappear.
+Partial JSON-shaped object keys remain local markers. To retain already issued cursors, a bare JSON object with the ODM tag and every historical writer field (`v`, `local`, `local_done`, `source`, `source_done`, `last_key`) is treated as an envelope, then strictly validated. A valid object key can be identical to that complete envelope: the two byte strings are indistinguishable, so legacy compatibility necessarily gives the envelope interpretation precedence. Framing identifies new merged tokens unambiguously, but dual-format readers do not eliminate this old full-envelope key collision. There is no signature, session store, or automatic format negotiation.
+
+An active v2 budget rejects the sixteenth consecutive merged page that consumes no new object/common prefix and reaches no new end-of-list state. The first fifteen empty pages can be resumed; with the existing two-fetch-per-side limit, that interval costs at most 32 fetches per side, including the failing request. A key, common prefix, or a newly exhausted side on the sixteenth request succeeds and resets the budget. A side that was already exhausted does not reset it again. A zero-sized request does not spend an existing budget. This is a resource bound, not proof of a cursor cycle: an unusually long but valid empty source-page chain also reaches the limit. Tokens are unsigned base64-encoded JSON, optionally framed, so this budget applies to clients that continue with the returned token unchanged; replaying or editing a token can reset it, and it is not a malicious-client defense or a global request quota. The two-fetch-per-side request limit and existing source rate limiter still apply. A source failure follows `policy.source_error`: `propagate` returns `424 SourceUnavailable` with `invalid_pagination`; `not_found` returns the fetched local listing with `x-rustfs-on-demand-migration-list: local_only`. A blocking local-side failure returns `InternalError`, without silently discarding local entries.
+
+With budget issuance off, a new v1 chain retains the existing limitation: an empty source cursor cycle spanning requests can continue indefinitely. Default rollout does not fix that chain until the v2 switch is enabled. Framing alone does not impose the budget.
+
+For rollback, first turn both issuance switches off on every node. Keep readers compatible with outstanding framed and v2 chains: neither switch rewrites existing tokens, and tokens have no expiration that proves those chains have drained. Route those continuations to compatible readers or have clients explicitly restart their listings before restoring older binaries. Restarting a listing is a new scan and can repeat entries. Do not assume switching issuance off makes outstanding framed or v2 tokens disappear.
 
 ## Positioning
 
@@ -188,9 +195,9 @@ No write, delete, ACL or versioning permission is required or used. Scope the po
 
 Behaviour a client can observe. The "Test" column names the case that pins it: `*_test.rs` files live under `crates/e2e_test/src/on_demand_migration/`, and the unit tests live next to the code in `rustfs/src/app/object/get.rs`, `head.rs` and `shared.rs`.
 
-ODM merged continuation tokens use a NUL-prefixed JSON envelope inside the existing base64 encoding. NUL is not valid in a local object key, so a legitimate JSON-shaped key can never be mistaken for a merged cursor. Upgrade every node before using list-through, and restart any in-progress ODM listing issued by an older build: its unframed JSON tokens cannot be distinguished from legitimate local keys. Ordinary local listing tokens remain unchanged. Tokens issued by this build can still resume the local side after list-through is disabled.
+ODM merged continuation tokens use bare or NUL-prefixed JSON inside the existing base64 encoding. The default writer preserves bare output; compatible readers accept both formats and retain existing budgets. See [List continuation token rollout](#list-continuation-token-rollout) for the independent issuance switches, rolling-upgrade requirements, and the unavoidable ambiguity between a complete historical envelope and an identically named local key.
 
-Source `HEAD` responses with status 404 require a successful bucket probe before being negative-cached. The source credential therefore needs permission for `HeadBucket` (S3 `ListBucket`); a prefix-restricted ListBucket policy can deny that probe, in which case the response is a source failure rather than a cached miss. A missing/inaccessible source bucket, a missing source version, or an ambiguous GET 404 is not proof that the requested key is absent. Conditional GET validators are checked against the actual source GET metadata as well as the advisory HEAD; a missing required validator fails with 424. Source LIST entries without a key or a non-negative size fail the page rather than fabricating an empty object.
+Source `HEAD` responses with status 404 require a successful bucket probe before being negative-cached. The source credential therefore needs permission for `HeadBucket` (S3 `ListBucket`); a prefix-restricted ListBucket policy can deny that probe, in which case the response is a source failure rather than a cached miss. A missing/inaccessible source bucket or a missing source version is not proof that the requested key is absent. Native GCS verifies the bucket after either HEAD or GET returns 404 and preserves a failed probe as a source error. Azure accepts explicit `BlobNotFound` only on an unversioned object read with status 404; an ambiguous HEAD may make one container probe, while an ambiguous GET remains a source error. Native probes add at most one request and retain the existing per-request timeouts, rather than a single deadline for the pair. Conditional GET validators are checked against the actual source GET metadata as well as the advisory HEAD; a missing required validator fails with 424. Source LIST entries without a key or a non-negative size fail the page rather than fabricating an empty object.
 
 Write-back currently requires namespace locking enabled and exactly one pool with one erasure set. Other topologies fail write-back explicitly as `unsupported`: source reads remain available, but backfill cannot complete successfully or certify cutover. This restriction avoids relying on a set-local condition across distinct pool or lock domains; it does not restrict ordinary S3 writes. Full cross-pool migration requires a globally fenced commit protocol.
 

rustfs/fixtures/on_demand_migration/source_config_e2a.rs

@@ -0,0 +1,80 @@
+// Strict source reader frozen from e2a921bc1608823c8efec955d7463ab8350a8a01.
+// Wire declarations and credential Debug are copied verbatim; runtime methods are omitted.
+use serde::{Deserialize, Serialize};
+use std::fmt;
+
+const REDACTED: &str = "REDACTED";
+
+/// The external S3-compatible source bucket.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct SourceConfig {
+    pub provider: Provider,
+    /// `http(s)://host[:port]` with no path or query. Optional only for
+    /// [`Provider::Aws`], where it is derived from `region`.
+    #[serde(default)]
+    pub endpoint: Option<String>,
+    pub region: String,
+    pub bucket: String,
+    #[serde(default)]
+    pub path_style: PathStyle,
+    /// `None` means anonymous access to a public source bucket.
+    #[serde(default)]
+    pub credentials: Option<SourceCredentials>,
+    #[serde(default)]
+    pub tls: TlsConfig,
+}
+
+/// Source vendor family. `azure` is deliberately absent from this version.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum Provider {
+    /// Generic S3-compatible endpoint.
+    S3,
+    Aws,
+    Minio,
+    Rustfs,
+    R2,
+    /// GCS XML interoperability API with HMAC keys.
+    Gcs,
+}
+
+/// Bucket addressing style. `auto` is resolved by the source client builder.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum PathStyle {
+    #[default]
+    Auto,
+    Path,
+    Virtual,
+}
+
+/// Static credentials for the source. `Debug` never prints the secret or
+/// the session token.
+#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct SourceCredentials {
+    pub access_key: String,
+    pub secret_key: String,
+    #[serde(default)]
+    pub session_token: Option<String>,
+}
+
+impl fmt::Debug for SourceCredentials {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("SourceCredentials")
+            .field("access_key", &self.access_key)
+            .field("secret_key", &REDACTED)
+            .field("session_token", &self.session_token.as_ref().map(|_| REDACTED))
+            .finish()
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct TlsConfig {
+    #[serde(default)]
+    pub skip_verify: bool,
+    #[serde(default)]
+    pub ca_cert_pem: Option<String>,
+}

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/bucket_list_through.rs

@@ -33,9 +33,10 @@ use super::storage_api::bucket_usecase::s3_api::bucket::ListObjectsV2Params;
 use crate::app::object::shared::{odm_source_unavailable_error, odm_state_error_class};
 use crate::error::ApiError;
 use crate::on_demand_migration::{
-    BucketOdmState, ListEntryKey, ListPageError, ListThroughCursor, ListThroughMerger, ListThroughToken, ListThroughTokenError,
-    MergeSide, OnDemandMigrationSys, SOURCE_LIST_MAX_RATE_WAIT, SourceClient, SourceError, SourceErrorPolicy, SourceListPlan,
-    SourceListRequest, SourceObject, SourcePage, decode_continuation_token, source_list_plan,
+    BucketOdmState, LIST_THROUGH_TOKEN_VERSION, ListEntryKey, ListPageError, ListThroughCursor, ListThroughMerger,
+    ListThroughToken, ListThroughTokenError, MergeSide, OnDemandMigrationSys, SOURCE_LIST_MAX_RATE_WAIT, SourceClient,
+    SourceError, SourceErrorPolicy, SourceListPlan, SourceListRequest, SourceObject, SourcePage, decode_continuation_token,
+    source_list_plan,
 };
 use futures::StreamExt;
 use rustfs_utils::http::{SUFFIX_SOURCE_PROXY_REQUEST, get_header};
@@ -53,6 +54,9 @@ const SOURCE_STORAGE_CLASS: &str = "STANDARD";
 /// Enable only after every node serving continuation requests can read v2.
 const ENV_LIST_PROGRESS_TOKENS: &str = "RUSTFS_ON_DEMAND_MIGRATION_LIST_V2_TOKENS";
 
+/// Independent of the budget version: bare-v2 readers cannot read framing.
+const ENV_LIST_FRAMED_TOKENS: &str = "RUSTFS_ON_DEMAND_MIGRATION_LIST_FRAMED_TOKENS";
+
 /// Concurrent local metadata probes when a versioned bucket has to check
 /// source-only keys for a shadowing delete marker.
 const DELETE_MARKER_PROBE_CONCURRENCY: usize = 32;
@@ -89,6 +93,37 @@ pub(crate) fn local_cursor(decoded: Option<&str>, merged: Option<&ListThroughTok
     }
 }
 
+/// A framed chain keeps its envelope when the bucket stops consulting source.
+pub(crate) fn preserve_framed_local_cursor(info: &mut ListObjectsV2Info, previous: Option<&ListThroughToken>) {
+    let Some(previous) = previous.filter(|token| token.framed) else {
+        return;
+    };
+    let Some(next) = info.next_continuation_token.take() else {
+        return;
+    };
+    let mut token = previous.clone();
+    token.local = Some(next);
+    token.local_done = false;
+    if let Some(last_key) = info
+        .objects
+        .iter()
+        .map(|object| object.name.as_str())
+        .chain(info.prefixes.iter().map(String::as_str))
+        .max()
+    {
+        token.last_key = Some(
+            token
+                .last_key
+                .as_deref()
+                .map_or(last_key, |previous| previous.max(last_key))
+                .to_string(),
+        );
+        token.v = LIST_THROUGH_TOKEN_VERSION;
+        token.no_progress = None;
+    }
+    info.next_continuation_token = Some(token.encode());
+}
+
 fn invalid_continuation_token(err: &ListThroughTokenError) -> S3Error {
     debug!(error = %err, "rejected an on-demand migration list continuation token");
     S3Error::with_message(S3ErrorCode::InvalidArgument, "Invalid continuation token".to_string())
@@ -289,6 +324,7 @@ pub(crate) async fn merged_list_objects_v2(
     }
 
     let issue_progress_tokens = rustfs_utils::get_env_bool(ENV_LIST_PROGRESS_TOKENS, false);
+    let framed = token.is_some_and(|token| token.framed) || rustfs_utils::get_env_bool(ENV_LIST_FRAMED_TOKENS, false);
     let outcome = match merger.finish(issue_progress_tokens) {
         Ok(outcome) => outcome,
         Err(ListPageError::NoProgress(MergeSide::Source)) => {
@@ -326,7 +362,10 @@ pub(crate) async fn merged_list_objects_v2(
         info: ListObjectsV2Info {
             is_truncated: outcome.is_truncated,
             continuation_token: None,
-            next_continuation_token: outcome.next_token.map(|token| token.encode()),
+            next_continuation_token: outcome.next_token.map(|mut token| {
+                token.framed = framed;
+                token.encode()
+            }),
             objects,
             prefixes,
         },
@@ -481,6 +520,7 @@ mod tests {
 
     fn token(local: Option<&str>, local_done: bool) -> ListThroughToken {
         ListThroughToken {
+            framed: false,
             t: "odm-list".to_string(),
             v: 1,
             local: local.map(str::to_string),
@@ -558,20 +598,23 @@ mod tests {
 
     #[test]
     fn a_v2_token_keeps_the_local_cursor_when_list_through_is_turned_off() {
-        let mut resume = token(Some("local-2"), false);
-        resume.v = 2;
-        resume.no_progress = Some(MAX_LIST_NO_PROGRESS_PAGES - 1);
-        let encoded = resume.encode();
-        let decoded = decode_list_cursor(Some(&encoded)).expect("a v2 envelope decodes");
-        assert_eq!(decoded.as_ref(), Some(&resume));
-        assert!(matches!(
-            local_cursor(Some(&encoded), decoded.as_ref()),
-            LocalListCursor::Token(Some(local)) if local == "local-2"
-        ));
-        resume.local_done = true;
-        let encoded = resume.encode();
-        let decoded = decode_list_cursor(Some(&encoded)).expect("v2 with local EOF decodes");
-        assert!(matches!(local_cursor(Some(&encoded), decoded.as_ref()), LocalListCursor::Exhausted));
+        for framed in [false, true] {
+            let mut resume = token(Some("local-2"), false);
+            resume.framed = framed;
+            resume.v = 2;
+            resume.no_progress = Some(MAX_LIST_NO_PROGRESS_PAGES - 1);
+            let encoded = resume.encode();
+            let decoded = decode_list_cursor(Some(&encoded)).expect("a v2 envelope decodes");
+            assert_eq!(decoded.as_ref(), Some(&resume));
+            assert!(matches!(
+                local_cursor(Some(&encoded), decoded.as_ref()),
+                LocalListCursor::Token(Some(local)) if local == "local-2"
+            ));
+            resume.local_done = true;
+            let encoded = resume.encode();
+            let decoded = decode_list_cursor(Some(&encoded)).expect("v2 with local EOF decodes");
+            assert!(matches!(local_cursor(Some(&encoded), decoded.as_ref()), LocalListCursor::Exhausted));
+        }
     }
 
     #[test]
@@ -762,6 +805,7 @@ mod tests {
         );
         let continuation_token = resume_source.map(|source| {
             let token = ListThroughToken {
+                framed: false,
                 t: "odm-list".into(),
                 v: 1,
                 local: None,
@@ -809,6 +853,286 @@ mod tests {
         .expect("listing must complete within its bounded source budget")
     }
 
+    async fn native_list_source(
+        provider: Provider,
+        pages: Vec<(String, String)>,
+    ) -> (
+        String,
+        tokio_util::task::AbortOnDropHandle<Vec<String>>,
+        tokio_util::sync::CancellationToken,
+    ) {
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
+            .await
+            .expect("bind native listing source");
+        let address = listener.local_addr().expect("native source address");
+        let stop = tokio_util::sync::CancellationToken::new();
+        let server_stop = stop.clone();
+        let server = tokio::spawn(async move {
+            let mut pages = pages.into_iter();
+            let mut requests = Vec::new();
+            loop {
+                let (mut stream, _) = tokio::select! {
+                    _ = server_stop.cancelled() => break,
+                    accepted = listener.accept() => accepted.expect("accept native source request"),
+                };
+                let (target, body) = pages.next().expect("native source must not receive an extra request");
+                let mut request = Vec::new();
+                let mut chunk = [0; 4096];
+                while !request.windows(4).any(|window| window == b"\r\n\r\n") {
+                    let count = stream.read(&mut chunk).await.expect("read native source request");
+                    assert!(count > 0, "native request needs complete headers");
+                    request.extend_from_slice(&chunk[..count]);
+                    assert!(request.len() <= 32 * 1024, "native request headers must be bounded");
+                }
+                let text = String::from_utf8(request).expect("native HTTP request text");
+                let first_line = text.lines().next().expect("native request line");
+                assert_eq!(first_line, format!("GET {target} HTTP/1.1"));
+                let authorization = text
+                    .lines()
+                    .filter_map(|line| line.split_once(':'))
+                    .find(|(name, _)| name.eq_ignore_ascii_case("authorization"))
+                    .map(|(_, value)| value.trim())
+                    .expect("native credential must be used");
+                assert!(authorization.starts_with(if provider == Provider::Azure {
+                    "SharedKey acct:"
+                } else {
+                    "Bearer "
+                }));
+                requests.push(first_line.to_string());
+                let response = format!("HTTP/1.1 200 OK\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}", body.len());
+                stream.write_all(response.as_bytes()).await.expect("write native source page");
+                stream.shutdown().await.expect("finish native source response");
+            }
+            assert!(pages.next().is_none(), "every scripted native page must have been requested");
+            requests
+        });
+        (format!("http://{address}"), tokio_util::task::AbortOnDropHandle::new(server), stop)
+    }
+
+    fn native_list_target(provider: Provider, cursor: bool, max_keys: i32) -> String {
+        if provider == Provider::Azure {
+            format!(
+                "/source-bucket?restype=container&comp=list{}&maxresults={max_keys}",
+                if cursor { "&marker=opaque%2B%2F%3D" } else { "" }
+            )
+        } else {
+            format!(
+                "/storage/v1/b/source-bucket/o?{}maxResults={max_keys}",
+                if cursor { "pageToken=opaque%2B%2F%3D&" } else { "" }
+            )
+        }
+    }
+
+    fn native_list_body(provider: Provider, entries: &str, prefixes: bool, next: bool) -> String {
+        if provider == Provider::Azure {
+            format!(
+                "<EnumerationResults><Blobs>{entries}{}</Blobs><NextMarker>{}</NextMarker></EnumerationResults>",
+                if prefixes {
+                    "<BlobPrefix><Name>目录/子/</Name></BlobPrefix>"
+                } else {
+                    ""
+                },
+                if next { "opaque+/=" } else { "" }
+            )
+        } else {
+            format!(
+                r#"{{"items":[{entries}],"prefixes":{},"nextPageToken":{}}}"#,
+                if prefixes { r#"["目录/子/"]"# } else { "[]" },
+                if next { r#""opaque+/=""# } else { "null" }
+            )
+        }
+    }
+
+    #[cfg(feature = "gcs")]
+    fn native_test_service_account() -> String {
+        // The real Google credentials implementation signs locally. Generate a
+        // disposable key instead of storing private key material in the fixture.
+        let key = rcgen::KeyPair::generate_for(&rcgen::PKCS_RSA_SHA256).expect("generate fixture service-account key");
+        serde_json::json!({
+            "type": "service_account",
+            "client_email": "fixture@example.invalid",
+            "private_key_id": "fixture-key",
+            "private_key": key.serialize_pem(),
+            "project_id": "fixture-project"
+        })
+        .to_string()
+    }
+
+    async fn native_source_policy_request(
+        provider: Provider,
+        policy: SourceErrorPolicy,
+        pages: Vec<(String, String)>,
+        service_account: &str,
+        max_keys: i32,
+    ) -> (S3Result<S3Response<ListObjectsV2Output>>, Vec<String>) {
+        let (endpoint, server, stop) = native_list_source(provider, pages).await;
+        let (_state_guard, mut input) = source_policy_input(endpoint.clone(), policy, None, None).await;
+        input.max_keys = Some(max_keys);
+        let sys = OnDemandMigrationSys::get();
+        let installed = sys.state(&input.bucket).expect("installed source state");
+        let mut config = installed.config().clone();
+        config.source = serde_json::from_value(serde_json::json!({
+            "provider": provider,
+            "endpoint": endpoint,
+            "region": "us-east-1",
+            "bucket": "source-bucket",
+            "azure": if provider == Provider::Azure { serde_json::json!({ "account": "acct", "account_key": "c2VjcmV0LWtleQ==" }) } else { serde_json::Value::Null },
+            "gcs": if provider == Provider::GcsNative { serde_json::json!({ "service_account_json": service_account }) } else { serde_json::Value::Null }
+        })).expect("native source configuration");
+        sys.apply_for_incarnation(&input.bucket, installed.incarnation_id(), Some(&config))
+            .await;
+        let state = sys.state(&input.bucket).expect("native source state");
+        state
+            .client()
+            .unwrap_or_else(|error| panic!("{provider:?} native client must build: {error:?}"));
+        let result = execute_source_list(input).await;
+        stop.cancel();
+        let requests = tokio::time::timeout(Duration::from_secs(5), server)
+            .await
+            .expect("native source server must finish")
+            .expect("native source requests must match the script");
+        (result, requests)
+    }
+
+    #[test]
+    #[serial_test::serial]
+    fn native_list_through_malformed_fields_follow_both_source_policies() {
+        run_large_stack_test("native-list-through-fields", || async {
+            temp_env::async_with_vars(
+                [("RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET", Some("true")), ("HTTP_PROXY", None), ("HTTPS_PROXY", None),
+                 ("ALL_PROXY", None), ("http_proxy", None), ("https_proxy", None), ("all_proxy", None),
+                 ("NO_PROXY", Some("*")), ("no_proxy", Some("*"))],
+                async {
+                    #[cfg(feature = "gcs")]
+                    let service_account = native_test_service_account();
+                    #[cfg(not(feature = "gcs"))]
+                    let service_account = String::new();
+                    for provider in [Provider::Azure, #[cfg(feature = "gcs")] Provider::GcsNative] {
+                        let invalid = if provider == Provider::Azure {
+                            ["<Blob><Name>bad</Name></Blob>",
+                             "<Blob><Name>bad</Name><Properties><Content-Length>-1</Content-Length></Properties></Blob>",
+                             "<Blob><Name>bad</Name><Properties><Content-Length>18446744073709551616</Content-Length></Properties></Blob>",
+                             "<Blob><Properties><Content-Length>1</Content-Length></Properties></Blob>"]
+                        } else {
+                            [r#"{"name":"bad"}"#, r#"{"name":"bad","size":"-1"}"#,
+                             r#"{"name":"bad","size":"18446744073709551616"}"#, r#"{"size":"1"}"#]
+                        };
+                        let valid = if provider == Provider::Azure {
+                            "<Blob><Name>a-source</Name><Properties><Content-Length>1</Content-Length></Properties></Blob>"
+                        } else { r#"{"name":"a-source","size":"1"}"# };
+                        for policy in [SourceErrorPolicy::Propagate, SourceErrorPolicy::NotFound] {
+                            for entry in invalid {
+                                for refill in [false, true] {
+                                    let mut pages = Vec::new();
+                                    if refill {
+                                        pages.push((native_list_target(provider, false, 2), native_list_body(provider, valid, false, true)));
+                                    }
+                                    let entries = if refill { entry.to_string() } else if provider == Provider::Azure {
+                                        format!("{valid}{entry}")
+                                    } else { format!("{valid},{entry}") };
+                                    pages.push((native_list_target(provider, refill, 2), native_list_body(provider, &entries, false, false)));
+                                    let (result, requests) = native_source_policy_request(provider, policy, pages, &service_account, 2).await;
+                                    assert_eq!(requests.len(), if refill { 2 } else { 1 }, "{provider:?} {policy:?} {entry}");
+                                    if policy == SourceErrorPolicy::Propagate {
+                                        let err = result.expect_err("malformed native page must propagate");
+                                        assert_eq!(err.status_code(), Some(http::StatusCode::FAILED_DEPENDENCY));
+                                        assert_eq!(err.code(), &S3ErrorCode::Custom("SourceUnavailable".into()));
+                                        assert_eq!(err.message(), Some("other"));
+                                    } else {
+                                        // Reuse the complete local-only assertions, including no
+                                        // leaked source objects, no cursor and the degraded header.
+                                        assert_source_policy_result(result, policy);
+                                    }
+                                }
+                            }
+                        }
+                    }
+                },
+            ).await;
+        });
+    }
+
+    #[test]
+    #[serial_test::serial]
+    fn native_list_through_preserves_valid_empty_pages_and_zero_size_objects() {
+        run_large_stack_test("native-list-through-valid", || async {
+            temp_env::async_with_vars(
+                [
+                    ("RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET", Some("true")),
+                    ("HTTP_PROXY", None),
+                    ("HTTPS_PROXY", None),
+                    ("ALL_PROXY", None),
+                    ("http_proxy", None),
+                    ("https_proxy", None),
+                    ("all_proxy", None),
+                    ("NO_PROXY", Some("*")),
+                    ("no_proxy", Some("*")),
+                ],
+                async {
+                    #[cfg(feature = "gcs")]
+                    let service_account = native_test_service_account();
+                    #[cfg(not(feature = "gcs"))]
+                    let service_account = String::new();
+                    for provider in [
+                        Provider::Azure,
+                        #[cfg(feature = "gcs")]
+                        Provider::GcsNative,
+                    ] {
+                        let valid = if provider == Provider::Azure {
+                            "<Blob><Name>目录/空</Name><Properties><Content-Length>0</Content-Length></Properties></Blob>"
+                        } else {
+                            r#"{"name":"目录/空","size":"0"}"#
+                        };
+                        for policy in [SourceErrorPolicy::Propagate, SourceErrorPolicy::NotFound] {
+                            let (result, requests) = native_source_policy_request(
+                                provider,
+                                policy,
+                                vec![
+                                    (native_list_target(provider, false, 3), native_list_body(provider, "", false, true)),
+                                    (native_list_target(provider, true, 3), native_list_body(provider, valid, true, false)),
+                                ],
+                                &service_account,
+                                3,
+                            )
+                            .await;
+                            assert_eq!(requests.len(), 2);
+                            let response = result.expect("valid native listing must succeed under either policy");
+                            assert_ne!(
+                                response
+                                    .headers
+                                    .get("x-rustfs-on-demand-migration-list")
+                                    .and_then(|value| value.to_str().ok()),
+                                Some("local_only")
+                            );
+                            let output = response.output;
+                            let objects = output.contents.expect("local and source objects");
+                            assert_eq!(
+                                objects
+                                    .iter()
+                                    .map(|object| (object.key.as_deref(), object.size))
+                                    .collect::<Vec<_>>(),
+                                vec![(Some("z-local"), Some(1)), (Some("目录/空"), Some(0))]
+                            );
+                            assert_eq!(
+                                output
+                                    .common_prefixes
+                                    .expect("native prefix")
+                                    .into_iter()
+                                    .map(|prefix| prefix.prefix)
+                                    .collect::<Vec<_>>(),
+                                vec![Some("目录/子/".to_string())]
+                            );
+                            assert_eq!(output.key_count, Some(3));
+                            assert_eq!(output.is_truncated, Some(false));
+                            assert!(output.next_continuation_token.is_none());
+                        }
+                    }
+                },
+            )
+            .await;
+        });
+    }
+
     async fn source_policy_request(
         pages: Vec<String>,
         policy: SourceErrorPolicy,
@@ -1162,6 +1486,7 @@ mod tests {
             temp_env::async_with_vars(
                 [
                     (ENV_LIST_PROGRESS_TOKENS, Some("true")),
+                    (ENV_LIST_FRAMED_TOKENS, None),
                     ("RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET", Some("true")),
                     ("HTTP_PROXY", None),
                     ("HTTPS_PROXY", None),
@@ -1205,6 +1530,7 @@ mod tests {
                                     seen.insert(next.clone()),
                                     "a cross-request source cursor cycle must not return an identical empty merged token"
                                 );
+                                assert!(!decode_wire_token(&next).framed, "the budget switch cannot enable framing");
                                 empty_pages += 1;
                                 input.continuation_token = Some(next);
                             }
@@ -1241,6 +1567,7 @@ mod tests {
             temp_env::async_with_vars(
                 [
                     (ENV_LIST_PROGRESS_TOKENS, None),
+                    (ENV_LIST_FRAMED_TOKENS, None),
                     ("RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET", Some("true")),
                     ("HTTP_PROXY", None),
                     ("HTTPS_PROXY", None),
@@ -1252,7 +1579,10 @@ mod tests {
                     ("no_proxy", Some("*")),
                 ],
                 async {
-                    for policy in [SourceErrorPolicy::Propagate, SourceErrorPolicy::NotFound] {
+                    for (policy, framed) in [SourceErrorPolicy::Propagate, SourceErrorPolicy::NotFound]
+                        .into_iter()
+                        .flat_map(|policy| [false, true].map(|framed| (policy, framed)))
+                    {
                         let pages = ["B", "C", "A"].map(|next| source_xml(Some(next), true, None));
                         let (endpoint, server, stop) = list_source(pages.into_iter().cycle()).await;
                         let (_state_guard, mut input) = source_policy_input(endpoint, policy, Some("A"), None).await;
@@ -1266,6 +1596,7 @@ mod tests {
                             let raw = base64_simd::STANDARD.decode_to_vec(&next).expect("base64 continuation token");
                             let decoded = std::str::from_utf8(&raw).expect("JSON token");
                             let token = decode_list_cursor(Some(decoded)).expect("v1 reader").expect("merged token");
+                            assert!(!token.framed, "the default cannot begin issuing framed tokens");
                             assert_eq!(token.v, 1, "the default rollout cannot begin issuing v2");
                             assert_eq!(token.no_progress, None);
                             assert!(!decoded.contains("no_progress"), "ordinary v1 wire shape stays unchanged");
@@ -1279,6 +1610,7 @@ mod tests {
                         let mut token = decode_list_cursor(Some(std::str::from_utf8(&raw).expect("JSON token")))
                             .expect("v1 reader")
                             .expect("merged token");
+                        token.framed = framed;
                         token.v = 2;
                         token.no_progress = Some(MAX_LIST_NO_PROGRESS_PAGES - 2);
                         input.continuation_token = Some(base64_simd::STANDARD.encode_to_string(token.encode().as_bytes()));
@@ -1290,6 +1622,7 @@ mod tests {
                         let token = decode_list_cursor(Some(std::str::from_utf8(&raw).expect("JSON token")))
                             .expect("v2 reader")
                             .expect("merged token");
+                        assert_eq!(token.framed, framed, "reader-only nodes retain the incoming framing");
                         assert_eq!(token.v, 2);
                         assert_eq!(token.no_progress, Some(MAX_LIST_NO_PROGRESS_PAGES - 1));
                         input.continuation_token = Some(next);
@@ -1318,6 +1651,7 @@ mod tests {
             temp_env::async_with_vars(
                 [
                     (ENV_LIST_PROGRESS_TOKENS, Some("true")),
+                    (ENV_LIST_FRAMED_TOKENS, None),
                     ("RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET", Some("true")),
                     ("HTTP_PROXY", None),
                     ("HTTPS_PROXY", None),
@@ -1393,6 +1727,346 @@ mod tests {
         });
     }
 
+    fn decode_wire_token(wire: &str) -> ListThroughToken {
+        let raw = base64_simd::STANDARD.decode_to_vec(wire).expect("base64 continuation token");
+        decode_list_cursor(Some(std::str::from_utf8(&raw).expect("UTF-8 cursor")))
+            .expect("valid continuation token")
+            .expect("merged continuation token")
+    }
+
+    #[test]
+    fn framed_local_continuations_preserve_json_markers_and_zero_sized_budgets() {
+        let json_key = r#"{"t":"odm-list","v":1,"local_done":true}"#;
+        let mut resume = token(Some("local-2"), false);
+        resume.framed = true;
+        resume.v = 2;
+        resume.no_progress = Some(15);
+        let mut page = ListObjectsV2Info {
+            is_truncated: true,
+            next_continuation_token: Some(json_key.to_string()),
+            objects: vec![info(json_key)],
+            ..Default::default()
+        };
+        preserve_framed_local_cursor(&mut page, Some(&resume));
+        let raw = page.next_continuation_token.expect("local continuation");
+        assert!(raw.starts_with("\0odm-list:"));
+        let decoded = decode_list_cursor(Some(&raw))
+            .expect("framed local continuation")
+            .expect("envelope");
+        assert!(decoded.framed);
+        assert_eq!(
+            decoded.local.as_deref(),
+            Some(json_key),
+            "the local marker is embedded without another encoding"
+        );
+        assert_eq!(decoded.source, resume.source);
+        assert_eq!(decoded.last_key.as_deref(), Some(json_key));
+        assert_eq!(decoded.v, 1);
+        assert_eq!(decoded.no_progress, None);
+        assert!(matches!(local_cursor(Some(&raw), Some(&decoded)), LocalListCursor::Token(Some(local)) if local == json_key));
+
+        let mut prefix_page = ListObjectsV2Info {
+            is_truncated: true,
+            next_continuation_token: Some("photos/".to_string()),
+            prefixes: vec!["photos/".to_string()],
+            ..Default::default()
+        };
+        preserve_framed_local_cursor(&mut prefix_page, Some(&resume));
+        let prefix = decode_list_cursor(prefix_page.next_continuation_token.as_deref())
+            .expect("prefix continuation")
+            .expect("framed prefix envelope");
+        assert!(prefix.framed);
+        assert_eq!(prefix.last_key.as_deref(), Some("photos/"));
+        assert_eq!(prefix.v, 1);
+        assert_eq!(prefix.no_progress, None);
+
+        let mut zero = ListObjectsV2Info {
+            is_truncated: true,
+            next_continuation_token: resume.local.clone(),
+            ..Default::default()
+        };
+        preserve_framed_local_cursor(&mut zero, Some(&resume));
+        assert_eq!(
+            decode_list_cursor(zero.next_continuation_token.as_deref()).expect("zero-sized continuation"),
+            Some(resume.clone())
+        );
+
+        resume.framed = false;
+        let mut ordinary = ListObjectsV2Info {
+            is_truncated: true,
+            next_continuation_token: Some(json_key.to_string()),
+            ..Default::default()
+        };
+        preserve_framed_local_cursor(&mut ordinary, Some(&resume));
+        assert_eq!(ordinary.next_continuation_token.as_deref(), Some(json_key));
+    }
+
+    #[test]
+    #[serial_test::serial]
+    fn list_through_historical_v2_budget_exhausts_on_the_next_reader_only_request() {
+        run_large_stack_test("list-through-historical-budget", || async {
+            temp_env::async_with_vars(
+                [
+                    (ENV_LIST_PROGRESS_TOKENS, None),
+                    (ENV_LIST_FRAMED_TOKENS, None),
+                    ("RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET", Some("true")),
+                    ("HTTP_PROXY", None), ("HTTPS_PROXY", None), ("ALL_PROXY", None),
+                    ("http_proxy", None), ("https_proxy", None), ("all_proxy", None),
+                    ("NO_PROXY", Some("*")), ("no_proxy", Some("*")),
+                ],
+                async {
+                    for policy in [SourceErrorPolicy::Propagate, SourceErrorPolicy::NotFound] {
+                        let (endpoint, server) = scripted_list_source(vec![
+                            source_xml(Some("B"), true, None), source_xml(Some("C"), true, None),
+                        ]).await;
+                        let (_state_guard, mut input) = source_policy_input(endpoint, policy, None, None).await;
+                        // Fixed bytes from the pre-framing writer, independent of today's encoder.
+                        input.continuation_token = Some("eyJ0Ijoib2RtLWxpc3QiLCJ2IjoyLCJsb2NhbCI6bnVsbCwibG9jYWxfZG9uZSI6ZmFsc2UsInNvdXJjZSI6IkEiLCJzb3VyY2VfZG9uZSI6ZmFsc2UsImxhc3Rfa2V5IjpudWxsLCJub19wcm9ncmVzcyI6MTV9".to_string());
+                        assert_source_policy_result(execute_source_list(input).await, policy);
+                        let requests = tokio::time::timeout(Duration::from_secs(5), server).await
+                            .expect("finite source server must finish").expect("source server must not panic");
+                        assert_eq!(requests.len(), 2, "the old count=15 must terminate without starting a new budget");
+                        for (request, cursor) in requests.iter().zip(["A", "B"]) {
+                            assert!(request.contains(&format!("continuation-token={cursor}")), "{request}");
+                        }
+                    }
+                },
+            ).await;
+        });
+    }
+
+    #[test]
+    #[serial_test::serial]
+    fn list_through_framing_survives_budget_reset_and_local_only_pagination() {
+        run_large_stack_test("list-through-framing-local-pagination", || async {
+            temp_env::async_with_vars(
+                [
+                    (ENV_LIST_PROGRESS_TOKENS, None),
+                    (ENV_LIST_FRAMED_TOKENS, Some("true")),
+                    ("RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET", Some("true")),
+                    ("HTTP_PROXY", None),
+                    ("HTTPS_PROXY", None),
+                    ("ALL_PROXY", None),
+                    ("http_proxy", None),
+                    ("https_proxy", None),
+                    ("all_proxy", None),
+                    ("NO_PROXY", Some("*")),
+                    ("no_proxy", Some("*")),
+                ],
+                async {
+                    let (endpoint, server) = scripted_list_source(vec![
+                        source_xml(Some("A"), true, None),
+                        source_xml(Some("B"), true, None),
+                        source_xml(Some("C"), true, None),
+                        source_xml(Some("D"), true, None),
+                        source_xml(None, false, Some("0-source")),
+                    ])
+                    .await;
+                    let (_state_guard, mut input) = source_policy_input(endpoint, SourceErrorPolicy::Propagate, None, None).await;
+                    let json_key = r#"{"t":"odm-list","v":1,"local_done":true}"#;
+                    let expected_local = ["a-local", "b-local", "z-local", json_key, "~last"];
+                    let store = shared_gating_ecstore().await;
+                    for key in ["a-local", "b-local", json_key, "~last"] {
+                        store
+                            .put_object(
+                                &input.bucket,
+                                key,
+                                &mut StoragePutObjReader::from_vec(vec![1]),
+                                &StorageObjectOptions::default(),
+                            )
+                            .await
+                            .expect("seed paginated local keys");
+                    }
+                    input.max_keys = Some(1);
+                    let first = execute_source_list(input.clone()).await.expect("legitimate empty page");
+                    assert_eq!(first.output.key_count, Some(0));
+                    assert_eq!(first.output.is_truncated, Some(true));
+                    let next = first.output.next_continuation_token.expect("first framed cursor");
+                    let token = decode_wire_token(&next);
+                    assert!(token.framed, "the independent switch permits first framing issuance");
+                    assert_eq!(token.v, 1, "framing issuance cannot enable the no-progress budget");
+                    assert_eq!(token.no_progress, None);
+                    input.continuation_token = Some(next);
+                    let budget_page =
+                        temp_env::async_with_vars([(ENV_LIST_PROGRESS_TOKENS, Some("true"))], execute_source_list(input.clone()))
+                            .await
+                            .expect("another valid empty page starts a budget only when enabled");
+                    assert_eq!(budget_page.output.key_count, Some(0));
+                    assert_eq!(budget_page.output.is_truncated, Some(true));
+                    let next = budget_page.output.next_continuation_token.expect("framed v2 cursor");
+                    let token = decode_wire_token(&next);
+                    assert!(token.framed);
+                    assert_eq!(token.v, 2);
+                    assert_eq!(token.no_progress, Some(1));
+                    input.continuation_token = Some(next);
+
+                    temp_env::async_with_vars(
+                        [(ENV_LIST_PROGRESS_TOKENS, None::<&str>), (ENV_LIST_FRAMED_TOKENS, None)],
+                        async {
+                            let second = execute_source_list(input.clone())
+                                .await
+                                .expect("reader-only node reaches source data");
+                            assert_eq!(second.output.key_count, Some(1));
+                            assert_eq!(second.output.is_truncated, Some(true));
+                            assert_eq!(second.output.contents.expect("source object")[0].key.as_deref(), Some("0-source"));
+                            let next = second.output.next_continuation_token.expect("remaining local listing");
+                            let token = decode_wire_token(&next);
+                            assert!(token.framed, "resetting the budget must not downgrade the framing");
+                            assert_eq!(token.v, 1);
+                            assert_eq!(token.no_progress, None);
+                            assert!(token.source_done);
+                            input.continuation_token = Some(next);
+
+                            OnDemandMigrationSys::get().remove(&input.bucket);
+                            for (index, key) in expected_local.iter().enumerate() {
+                                let page = execute_source_list(input.clone()).await.expect("local-only continuation");
+                                assert!(!page.headers.contains_key("x-rustfs-on-demand-migration-list"));
+                                assert_eq!(page.output.key_count, Some(1));
+                                let keys: Vec<_> = page
+                                    .output
+                                    .contents
+                                    .expect("one local object")
+                                    .into_iter()
+                                    .map(|object| object.key.expect("local key"))
+                                    .collect();
+                                assert_eq!(
+                                    keys,
+                                    vec![key.to_string()],
+                                    "no duplicate or omitted key after disabling list-through"
+                                );
+                                let truncated = index + 1 < expected_local.len();
+                                assert_eq!(page.output.is_truncated, Some(truncated));
+                                if truncated {
+                                    let next = page.output.next_continuation_token.expect("local side still has keys");
+                                    let token = decode_wire_token(&next);
+                                    assert!(token.framed);
+                                    assert_eq!(token.v, 1);
+                                    assert_eq!(token.no_progress, None);
+                                    assert!(token.local.as_deref().expect("local marker").starts_with(*key));
+                                    assert_eq!(token.last_key.as_deref(), Some(*key));
+                                    input.continuation_token = Some(next);
+                                } else {
+                                    assert!(page.output.next_continuation_token.is_none());
+                                }
+                            }
+                        },
+                    )
+                    .await;
+                    let requests = tokio::time::timeout(Duration::from_secs(5), server)
+                        .await
+                        .expect("finite source server must finish")
+                        .expect("source server must not panic");
+                    assert_eq!(
+                        requests.len(),
+                        5,
+                        "format changes and local-only continuation perform no additional source I/O"
+                    );
+                    assert!(!requests[0].contains("continuation-token="));
+                    for (request, cursor) in requests[1..].iter().zip(["A", "B", "C", "D"]) {
+                        assert!(request.contains(&format!("continuation-token={cursor}")), "{request}");
+                    }
+                },
+            )
+            .await;
+        });
+    }
+
+    #[test]
+    #[serial_test::serial]
+    fn list_through_zero_sized_framed_request_preserves_its_budget() {
+        run_large_stack_test("list-through-framed-zero-size", || async {
+            temp_env::async_with_vars(
+                [
+                    (ENV_LIST_PROGRESS_TOKENS, None), (ENV_LIST_FRAMED_TOKENS, None),
+                    ("RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET", Some("true")),
+                    ("HTTP_PROXY", None), ("HTTPS_PROXY", None), ("ALL_PROXY", None),
+                    ("http_proxy", None), ("https_proxy", None), ("all_proxy", None),
+                    ("NO_PROXY", Some("*")), ("no_proxy", Some("*")),
+                ],
+                async {
+                    let (endpoint, server) = scripted_list_source(vec![
+                        source_xml(Some("B"), true, None), source_xml(Some("C"), true, None),
+                    ]).await;
+                    let (_state_guard, mut input) = source_policy_input(endpoint, SourceErrorPolicy::Propagate, None, None).await;
+                    let wire = concat!("\0odm-list:", r#"{"t":"odm-list","v":2,"local":null,"local_done":true,"source":"A","source_done":false,"last_key":null,"no_progress":15}"#);
+                    input.continuation_token = Some(base64_simd::STANDARD.encode_to_string(wire.as_bytes()));
+                    input.max_keys = Some(0);
+                    let zero = execute_source_list(input.clone()).await.expect("zero-sized request does not spend the budget");
+                    assert_eq!(zero.output.key_count, Some(0));
+                    assert_eq!(zero.output.is_truncated, Some(true));
+                    let next = zero.output.next_continuation_token.expect("unconsumed source");
+                    let token = decode_wire_token(&next);
+                    assert!(token.framed);
+                    assert_eq!(token.v, 2);
+                    assert_eq!(token.no_progress, Some(15));
+                    assert_eq!(token.source.as_deref(), Some("A"));
+                    assert_eq!(next, input.continuation_token.as_ref().expect("original cursor").as_str());
+                    let state = OnDemandMigrationSys::get().state(&input.bucket).expect("source state");
+                    assert_eq!(state.stats().snapshot(state.breaker().state()).source_latency.count, 0, "zero-sized request must not fetch the source");
+                    input.continuation_token = Some(next);
+                    input.max_keys = Some(2);
+                    assert_source_policy_result(execute_source_list(input).await, SourceErrorPolicy::Propagate);
+                    let requests = tokio::time::timeout(Duration::from_secs(5), server).await
+                        .expect("finite source server must finish").expect("source server must not panic");
+                    assert_eq!(requests.len(), 2, "only the resumed nonzero request fetches the source");
+                    assert_eq!(state.stats().snapshot(state.breaker().state()).source_latency.count, 2);
+                    for (request, cursor) in requests.iter().zip(["A", "B"]) {
+                        assert!(request.contains(&format!("continuation-token={cursor}")), "{request}");
+                    }
+                },
+            ).await;
+        });
+    }
+
+    #[test]
+    #[serial_test::serial]
+    fn zero_sized_merged_cursors_preserve_each_side_and_wire_format() {
+        run_large_stack_test("list-through-zero-side-matrix", || async {
+            temp_env::async_with_vars(
+                [
+                    (ENV_LIST_PROGRESS_TOKENS, Some("true")),
+                    (ENV_LIST_FRAMED_TOKENS, Some("true")),
+                    ("RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET", Some("true")),
+                    ("HTTP_PROXY", None), ("HTTPS_PROXY", None), ("ALL_PROXY", None),
+                    ("http_proxy", None), ("https_proxy", None), ("all_proxy", None),
+                    ("NO_PROXY", Some("*")), ("no_proxy", Some("*")),
+                ],
+                async {
+                    for framed in [false, true] {
+                        for (local_done, source_done) in [(false, true), (true, false), (false, false), (true, true)] {
+                            let (endpoint, server, stop) = list_source(std::iter::repeat(source_xml(Some("unexpected"), true, None))).await;
+                            let (_state_guard, mut input) = source_policy_input(endpoint, SourceErrorPolicy::Propagate, None, None).await;
+                            let json = format!(r#"{{"t":"odm-list","v":2,"local":"local-marker","local_done":{local_done},"source":"source-marker","source_done":{source_done},"last_key":null,"no_progress":15}}"#);
+                            let wire = if framed { format!("\0odm-list:{json}") } else { json };
+                            let original = base64_simd::STANDARD.encode_to_string(wire.as_bytes());
+                            input.continuation_token = Some(original.clone());
+                            input.max_keys = Some(0);
+                            let output = execute_source_list(input).await.expect("zero page remains local").output;
+                            let has_more = !local_done || !source_done;
+                            assert_eq!(output.key_count, Some(0));
+                            assert_eq!(output.is_truncated, Some(has_more), "framed={framed}, local_done={local_done}, source_done={source_done}");
+                            assert_eq!(output.next_continuation_token.as_deref(), has_more.then_some(original.as_str()));
+                            if let Some(next) = output.next_continuation_token {
+                                let token = decode_wire_token(&next);
+                                assert_eq!(token.framed, framed);
+                                assert_eq!(token.v, 2);
+                                assert_eq!(token.no_progress, Some(15));
+                                assert_eq!(token.local_done, local_done);
+                                assert_eq!(token.source_done, source_done);
+                                assert_eq!(token.local.as_deref(), Some("local-marker"));
+                                assert_eq!(token.source.as_deref(), Some("source-marker"));
+                            }
+                            stop.cancel();
+                            let requests = tokio::time::timeout(Duration::from_secs(5), server).await
+                                .expect("unused source must finish").expect("source server must not panic");
+                            assert!(requests.is_empty(), "zero-sized request must not access the source: {requests:?}");
+                        }
+                    }
+                },
+            ).await;
+        });
+    }
+
     fn assert_source_policy_result(result: S3Result<S3Response<ListObjectsV2Output>>, policy: SourceErrorPolicy) {
         match policy {
             SourceErrorPolicy::Propagate => {

rustfs/src/app/bucket_usecase.rs

@@ -2768,8 +2768,21 @@ impl DefaultBucketUsecase {
         } else {
             (None, None)
         };
-        let (object_infos, degraded) = match source_state {
-            Some(state) => {
+        let (object_infos, degraded) = match (source_state, merged_token.as_ref()) {
+            (None, Some(token)) if params.max_keys == 0 => {
+                // No source was consulted, so retain every unconsumed side and
+                // the original wire format without spending its progress budget.
+                let is_truncated = !token.local_done || !token.source_done;
+                (
+                    StorageListObjectsV2Info {
+                        is_truncated,
+                        next_continuation_token: params.decoded_continuation_token.clone().filter(|_| is_truncated),
+                        ..Default::default()
+                    },
+                    false,
+                )
+            }
+            (Some(state), _) => {
                 let outcome = list_through::merged_list_objects_v2(
                     &store,
                     &state,
@@ -2782,12 +2795,12 @@ impl DefaultBucketUsecase {
                 .await?;
                 (outcome.info, outcome.degraded)
             }
-            None => {
+            (None, _) => {
                 let cursor = list_through::local_cursor(params.decoded_continuation_token.as_deref(), merged_token.as_ref());
                 match cursor {
                     list_through::LocalListCursor::Exhausted => (StorageListObjectsV2Info::default(), false),
                     list_through::LocalListCursor::Token(token) => {
-                        let infos = store
+                        let mut infos = store
                             .list_objects_v2(
                                 &bucket,
                                 &params.prefix,
@@ -2800,6 +2813,7 @@ impl DefaultBucketUsecase {
                             )
                             .await
                             .map_err(ApiError::from)?;
+                        list_through::preserve_framed_local_cursor(&mut infos, merged_token.as_ref());
                         (infos, false)
                     }
                 }

rustfs/src/app/object/on_demand_migration_put.rs

@@ -692,9 +692,16 @@ mod tests {
         }
     }
 
-    #[tokio::test]
+    #[test]
     #[serial_test::serial]
-    async fn write_back_multipart_completion_preserves_a_client_put_after_staging() {
+    fn write_back_multipart_completion_preserves_a_client_put_after_staging() {
+        crate::app::gating_test_env::run_large_stack_test(
+            "odm-write-back-multipart-client-put-race",
+            write_back_multipart_completion_preserves_a_client_put_after_staging_inner,
+        );
+    }
+
+    async fn write_back_multipart_completion_preserves_a_client_put_after_staging_inner() {
         let (store, bucket) = write_back_test_bucket("odm-mpu-race", false).await;
         let write_back = OnDemandMigrationWriteBack::new();
         let req = request(

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/on_demand_migration/azure.rs

@@ -162,6 +162,32 @@ impl AzureSourceBackend {
         Ok(request)
     }
 
+    /// A missing blob is distinct from a missing container or version. Only
+    /// object reads may use BlobNotFound as positive evidence of absence.
+    async fn send_object_request(&self, request: reqwest::Request) -> Result<reqwest::Response, SourceError> {
+        let is_head = request.method() == Method::HEAD;
+        let versioned = request
+            .url()
+            .query_pairs()
+            .any(|(name, _)| name.eq_ignore_ascii_case("versionid") || name.eq_ignore_ascii_case("snapshot"));
+        let response = self.http.execute(request).await?;
+        if response.status() == http::StatusCode::NOT_FOUND && !versioned {
+            match header(response.headers(), HEADER_ERROR_CODE) {
+                Some("BlobNotFound") => return Err(SourceError::NotFound),
+                None | Some("ResourceNotFound") if is_head => {
+                    // HEAD may omit an error code. One successful container
+                    // probe proves key absence; a failed probe keeps its error.
+                    // These are two independently timed requests, not one deadline.
+                    drop(response);
+                    self.probe().await?;
+                    return Err(SourceError::NotFound);
+                }
+                _ => {}
+            }
+        }
+        NativeHttp::check_response(response, Some(HEADER_ERROR_CODE))
+    }
+
     /// Shared mapping for Get Blob and Get Blob Properties.
     fn head_from_response(headers: &HeaderMap) -> Result<SourceHead, SourceError> {
         // A customer-provided key means the service holds ciphertext it cannot
@@ -188,7 +214,7 @@ impl AzureSourceBackend {
 impl SourceBackend for AzureSourceBackend {
     async fn head(&self, key: &str) -> Result<SourceHead, SourceError> {
         let request = self.request(Method::HEAD, self.blob_url(key)?, HeaderMap::new())?;
-        let response = self.http.send(request, HEADER_ERROR_CODE).await?;
+        let response = self.send_object_request(request).await?;
         Self::head_from_response(response.headers())
     }
 
@@ -201,7 +227,7 @@ impl SourceBackend for AzureSourceBackend {
             );
         }
         let request = self.request(Method::GET, self.blob_url(key)?, headers)?;
-        let response = self.http.send(request, HEADER_ERROR_CODE).await?;
+        let response = self.send_object_request(request).await?;
         let head = Self::head_from_response(response.headers())?;
         let content_range = header(response.headers(), "content-range").map(str::to_string);
         Ok(SourceGet {
@@ -239,7 +265,7 @@ impl SourceBackend for AzureSourceBackend {
         }
 
         let request = self.request(Method::GET, url, HeaderMap::new())?;
-        let response = self.http.send(request, HEADER_ERROR_CODE).await?;
+        let response = self.http.send(request, Some(HEADER_ERROR_CODE)).await?;
         let body = read_text(response, MAX_XML_BYTES).await?;
         let listing = parse_list_blobs(&body)?;
 
@@ -255,7 +281,7 @@ impl SourceBackend for AzureSourceBackend {
         let mut url = self.blob_url(key)?;
         url.query_pairs_mut().append_pair("comp", "tags");
         let request = self.request(Method::GET, url, HeaderMap::new())?;
-        let response = self.http.send(request, HEADER_ERROR_CODE).await?;
+        let response = self.http.send(request, Some(HEADER_ERROR_CODE)).await?;
         let body = read_text(response, MAX_XML_BYTES).await?;
         parse_blob_tags(&body)
     }
@@ -264,7 +290,7 @@ impl SourceBackend for AzureSourceBackend {
         let mut url = self.container_url()?;
         url.query_pairs_mut().append_pair("restype", "container");
         let request = self.request(Method::HEAD, url, HeaderMap::new())?;
-        self.http.send(request, HEADER_ERROR_CODE).await?;
+        self.http.send(request, Some(HEADER_ERROR_CODE)).await?;
         Ok(())
     }
 }
@@ -342,9 +368,9 @@ struct AzureListing {
 
 #[derive(Default)]
 struct BlobEntry {
-    name: String,
+    name: Option<String>,
     etag: Option<String>,
-    size: u64,
+    size: Option<u64>,
     last_modified: Option<std::time::SystemTime>,
     access_tier: Option<String>,
 }
@@ -357,6 +383,7 @@ fn parse_list_blobs(xml: &str) -> Result<AzureListing, SourceError> {
     let mut next_marker = None;
     let mut blob: Option<BlobEntry> = None;
     let mut in_blob_prefix = false;
+    let mut blob_prefix: Option<String> = None;
     // Open container elements. quick-xml reports a truncated document as a
     // plain end of input, so a non-zero depth at EOF is the only signal that
     // the page was cut short and must not be read as a complete listing.
@@ -366,6 +393,9 @@ fn parse_list_blobs(xml: &str) -> Result<AzureListing, SourceError> {
         match reader.read_event() {
             Ok(Event::Start(start)) => {
                 let name = local_name(start.name().as_ref());
+                if matches!(name.as_str(), "blob" | "blobprefix") && (blob.is_some() || in_blob_prefix) {
+                    return Err(SourceError::Other("source listing entries must not be nested".to_string()));
+                }
                 match name.as_str() {
                     "blob" => {
                         depth += 1;
@@ -379,22 +409,30 @@ fn parse_list_blobs(xml: &str) -> Result<AzureListing, SourceError> {
                     _ => {
                         let end = start.to_end().into_owned();
                         let text = leaf_text(&mut reader, end.name())?;
-                        apply_list_field(&name, text, &mut blob, &mut prefixes, &mut next_marker, in_blob_prefix);
+                        apply_list_field(&name, text, &mut blob, &mut blob_prefix, &mut next_marker, in_blob_prefix)?;
                     }
                 }
             }
             Ok(Event::Empty(empty)) => {
                 let name = local_name(empty.name().as_ref());
-                apply_list_field(&name, String::new(), &mut blob, &mut prefixes, &mut next_marker, in_blob_prefix);
+                if matches!(name.as_str(), "blob" | "blobprefix") {
+                    return Err(SourceError::Other("source listing entry has no name".to_string()));
+                }
+                apply_list_field(&name, String::new(), &mut blob, &mut blob_prefix, &mut next_marker, in_blob_prefix)?;
             }
             Ok(Event::End(end)) => match local_name(end.name().as_ref()).as_str() {
                 "blob" => {
                     depth = depth.saturating_sub(1);
                     if let Some(entry) = blob.take() {
                         objects.push(SourceObject {
-                            key: entry.name,
+                            key: entry
+                                .name
+                                .filter(|name| !name.is_empty())
+                                .ok_or_else(|| SourceError::Other("source listing object has no name".to_string()))?,
                             etag: entry.etag,
-                            size: entry.size,
+                            size: entry
+                                .size
+                                .ok_or_else(|| SourceError::Other("source listing object has no valid size".to_string()))?,
                             last_modified: entry.last_modified,
                             storage_class: entry.access_tier,
                             // Azure ETags carry no part count; the listing
@@ -406,6 +444,12 @@ fn parse_list_blobs(xml: &str) -> Result<AzureListing, SourceError> {
                 "blobprefix" => {
                     depth = depth.saturating_sub(1);
                     in_blob_prefix = false;
+                    prefixes.push(
+                        blob_prefix
+                            .take()
+                            .filter(|name| !name.is_empty())
+                            .ok_or_else(|| SourceError::Other("source listing prefix has no name".to_string()))?,
+                    );
                 }
                 "properties" | "blobs" | "enumerationresults" => depth = depth.saturating_sub(1),
                 _ => {}
@@ -430,16 +474,22 @@ fn apply_list_field(
     name: &str,
     text: String,
     blob: &mut Option<BlobEntry>,
-    prefixes: &mut Vec<String>,
+    blob_prefix: &mut Option<String>,
     next_marker: &mut Option<String>,
     in_blob_prefix: bool,
-) {
+) -> Result<(), SourceError> {
     match name {
         "name" => {
             if in_blob_prefix {
-                prefixes.push(text);
+                if blob_prefix.is_some() {
+                    return Err(SourceError::Other("source listing prefix has duplicate names".to_string()));
+                }
+                *blob_prefix = Some(text);
             } else if let Some(entry) = blob.as_mut() {
-                entry.name = text;
+                if entry.name.is_some() {
+                    return Err(SourceError::Other("source listing object has duplicate names".to_string()));
+                }
+                entry.name = Some(text);
             }
         }
         "nextmarker" => *next_marker = Some(text),
@@ -450,7 +500,14 @@ fn apply_list_field(
         }
         "content-length" => {
             if let Some(entry) = blob.as_mut() {
-                entry.size = text.trim().parse().unwrap_or(0);
+                if entry.size.is_some() {
+                    return Err(SourceError::Other("source listing object has duplicate sizes".to_string()));
+                }
+                entry.size = Some(
+                    text.trim()
+                        .parse()
+                        .map_err(|_| SourceError::Other("source listing object has no valid size".to_string()))?,
+                );
             }
         }
         "last-modified" => {
@@ -465,6 +522,7 @@ fn apply_list_field(
         }
         _ => {}
     }
+    Ok(())
 }
 
 /// Parses a `Get Blob Tags` response.
@@ -551,7 +609,7 @@ mod tests {
     use super::*;
     use crate::on_demand_migration::backend_contract::{BackendCapabilities, assert_backend_contract};
     use crate::on_demand_migration::source_client::SourceError;
-    use crate::on_demand_migration::test_http_fixture::{ScriptedResponse, scripted_server};
+    use crate::on_demand_migration::test_http_fixture::{ScriptedResponse, assert_requests, scripted_server};
 
     const LIST_PAGE: &str = r#"<?xml version="1.0" encoding="utf-8"?>
 <EnumerationResults ServiceEndpoint="https://acct.blob.core.windows.net/" ContainerName="legacy">
@@ -633,6 +691,107 @@ mod tests {
         assert!(parse_blob_tags("<Tags><TagSet>").is_err(), "a truncated tag set must fail");
     }
 
+    #[tokio::test]
+    async fn native_listing_rejects_missing_or_invalid_required_object_fields() {
+        for entry in [
+            "<Blob />",
+            "<Blob><Properties><Content-Length>1</Content-Length></Properties></Blob>",
+            "<Blob><Name /><Properties><Content-Length>1</Content-Length></Properties></Blob>",
+            "<Blob><Name>broken</Name></Blob>",
+            "<Blob><Name>broken</Name><Properties><Content-Length /></Properties></Blob>",
+            "<Blob><Name>broken</Name><Properties><Content-Length>-1</Content-Length></Properties></Blob>",
+            "<Blob><Name>broken</Name><Properties><Content-Length>18446744073709551616</Content-Length></Properties></Blob>",
+            "<Blob><Name>broken</Name><Properties><Content-Length>not-a-size</Content-Length></Properties></Blob>",
+            "<BlobPrefix />",
+            "<BlobPrefix><Name /></BlobPrefix>",
+            "<BlobPrefix></BlobPrefix>",
+        ] {
+            // Reject the entire page even if a valid object precedes the bad
+            // entry, so callers cannot expose partial data or advance its cursor.
+            let body = format!(
+                "<EnumerationResults><Blobs><Blob><Name>valid</Name><Properties><Content-Length>1</Content-Length></Properties></Blob>{entry}</Blobs><NextMarker>next</NextMarker></EnumerationResults>"
+            );
+            let (endpoint, recorded) = scripted_server(vec![ScriptedResponse::new(200, Vec::new(), body)]).await;
+            let err = backend(&endpoint, Credential::SharedKey(vec![7_u8; 32]))
+                .list(&SourceListRequest {
+                    prefix: Some("dir/"),
+                    delimiter: Some("/"),
+                    continuation_token: Some("opaque+/="),
+                    max_keys: 2,
+                    ..Default::default()
+                })
+                .await
+                .expect_err("malformed object must reject the complete native page");
+            assert!(matches!(err, SourceError::Other(_)), "{entry}: {err:?}");
+            assert!(!err.is_retryable());
+            assert_requests(
+                &recorded,
+                &[(
+                    "GET",
+                    "/legacy?restype=container&comp=list&prefix=dir%2F&delimiter=%2F&marker=opaque%2B%2F%3D&maxresults=2",
+                )],
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn native_listing_rejects_duplicate_fields_and_nested_entries() {
+        for entry in [
+            "<Blob><Name>a</Name><Name>b</Name><Properties><Content-Length>1</Content-Length></Properties></Blob>",
+            "<Blob><Name /><Name>b</Name><Properties><Content-Length>1</Content-Length></Properties></Blob>",
+            "<Blob><Name>a</Name><Properties><Content-Length>1</Content-Length><Content-Length>2</Content-Length></Properties></Blob>",
+            "<BlobPrefix><Name>a/</Name><Name>b/</Name></BlobPrefix>",
+            "<BlobPrefix><Name /><Name>b/</Name></BlobPrefix>",
+            "<Blob><Name>a</Name><Properties><Content-Length>1</Content-Length></Properties><Blob><Name>b</Name><Properties><Content-Length>2</Content-Length></Properties></Blob></Blob>",
+            "<Blob><Name>a</Name><Properties><Content-Length>1</Content-Length></Properties><BlobPrefix><Name>b/</Name></BlobPrefix></Blob>",
+            "<BlobPrefix><Name>a/</Name><Blob><Name>b</Name><Properties><Content-Length>2</Content-Length></Properties></Blob></BlobPrefix>",
+            "<BlobPrefix><Name>a/</Name><BlobPrefix><Name>b/</Name></BlobPrefix></BlobPrefix>",
+        ] {
+            let body = format!(
+                "<EnumerationResults><Blobs><Blob><Name>valid</Name><Properties><Content-Length>0</Content-Length></Properties></Blob>{entry}</Blobs><NextMarker>next</NextMarker></EnumerationResults>"
+            );
+            let (endpoint, recorded) = scripted_server(vec![ScriptedResponse::new(200, Vec::new(), body)]).await;
+            let result = backend(&endpoint, Credential::SharedKey(vec![7_u8; 32]))
+                .list(&SourceListRequest {
+                    delimiter: Some("/"),
+                    continuation_token: Some("opaque+/="),
+                    max_keys: 2,
+                    ..Default::default()
+                })
+                .await;
+            let err = result.expect_err("ambiguous entries must reject the entire page and its cursor");
+            assert!(matches!(err, SourceError::Other(_)), "{entry}: {err:?}");
+            assert!(!err.is_retryable(), "{entry}: {err:?}");
+            assert_requests(
+                &recorded,
+                &[(
+                    "GET",
+                    "/legacy?restype=container&comp=list&delimiter=%2F&marker=opaque%2B%2F%3D&maxresults=2",
+                )],
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn native_listing_preserves_zero_size_unicode_prefixes_and_opaque_cursors() {
+        let body = "<EnumerationResults><Blobs><Blob><Name>目录/空 &amp; file</Name><Properties><Content-Length>0</Content-Length></Properties></Blob><BlobPrefix><Name>目录/子/</Name></BlobPrefix></Blobs><NextMarker>opaque+/=</NextMarker></EnumerationResults>";
+        let (endpoint, recorded) = scripted_server(vec![ScriptedResponse::new(200, Vec::new(), body.to_string())]).await;
+        let page = backend(&endpoint, Credential::SharedKey(vec![7_u8; 32]))
+            .list(&SourceListRequest {
+                max_keys: 2,
+                ..Default::default()
+            })
+            .await
+            .expect("valid native page");
+        assert_eq!(page.objects.len(), 1);
+        assert_eq!(page.objects[0].key, "目录/空 & file");
+        assert_eq!(page.objects[0].size, 0);
+        assert_eq!(page.common_prefixes, ["目录/子/"]);
+        assert!(page.is_truncated);
+        assert_eq!(page.next_continuation_token.as_deref(), Some("opaque+/="));
+        assert_requests(&recorded, &[("GET", "/legacy?restype=container&comp=list&maxresults=2")]);
+    }
+
     #[test]
     fn blob_tags_parse_into_the_shared_tag_map() {
         let tags = parse_blob_tags(TAGS).expect("tags should parse");
@@ -991,6 +1150,184 @@ mod tests {
         ]
     }
 
+    #[tokio::test]
+    async fn object_not_found_requires_provider_evidence_or_one_successful_head_probe() {
+        for method in [Method::HEAD, Method::GET] {
+            for (status, code, expected) in [
+                (404, Some("BlobNotFound"), "not_found"),
+                (403, Some("BlobNotFound"), "access_denied"),
+                (404, Some("ContainerNotFound"), "other"),
+                (404, Some("BlobVersionNotFound"), "other"),
+                (404, Some("UnrecognizedError"), "other"),
+                (404, None, if method == Method::HEAD { "not_found" } else { "other" }),
+                (404, Some("ResourceNotFound"), if method == Method::HEAD { "not_found" } else { "other" }),
+            ] {
+                let probes = method == Method::HEAD && status == 404 && matches!(code, None | Some("ResourceNotFound"));
+                let headers = code
+                    .map(|value| vec![(HEADER_ERROR_CODE, value.to_string())])
+                    .unwrap_or_default();
+                let mut responses = vec![ScriptedResponse::new(status, headers, "untrusted-error-body".to_string())];
+                if probes {
+                    responses.push(ScriptedResponse::new(200, Vec::new(), String::new()));
+                }
+                let (endpoint, recorded) = scripted_server(responses).await;
+                let backend = backend(&endpoint, Credential::SharedKey(vec![7_u8; 32]));
+                let result = if method == Method::HEAD {
+                    backend.head("missing").await.map(|_| ())
+                } else {
+                    backend.get("missing", None).await.map(|_| ())
+                };
+                let err = result.expect_err("object error must remain an error");
+                assert_eq!(err.class_label(), expected, "{method} {status} {code:?}: {err:?}");
+                assert!(!err.is_retryable(), "{err:?}");
+                assert!(!err.to_string().contains("untrusted-error-body"));
+                let mut requests = vec![(method.as_str(), "/legacy/missing")];
+                if probes {
+                    requests.push(("HEAD", "/legacy?restype=container"));
+                }
+                assert_requests(&recorded, &requests);
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn s3_not_found_alias_never_proves_native_object_absence() {
+        for selector in [None, Some("versionid"), Some("snapshot")] {
+            for operation in ["head", "get", "list", "tags", "probe"] {
+                if selector.is_some() && !matches!(operation, "head" | "get") {
+                    continue;
+                }
+                for (status, expected, retryable) in [
+                    (403, "access_denied", false),
+                    (404, "other", false),
+                    (416, "other", false),
+                    (500, "server_error", true),
+                ] {
+                    let (endpoint, recorded) = scripted_server(vec![ScriptedResponse::new(
+                        status,
+                        vec![(HEADER_ERROR_CODE, "NoSuchKey".to_string())],
+                        "untrusted-error-body".to_string(),
+                    )])
+                    .await;
+                    let credential = selector.map_or_else(
+                        || Credential::SharedKey(vec![7_u8; 32]),
+                        |selector| Credential::Sas(vec![(selector.to_string(), "old-version".to_string())]),
+                    );
+                    let backend = backend(&endpoint, credential);
+                    let result = match operation {
+                        "head" => backend.head("missing").await.map(|_| ()),
+                        "get" => backend.get("missing", None).await.map(|_| ()),
+                        "list" => backend.list(&SourceListRequest::default()).await.map(|_| ()),
+                        "tags" => backend.tagging("missing").await.map(|_| ()),
+                        "probe" => backend.probe().await,
+                        _ => unreachable!(),
+                    };
+                    let err = result.expect_err("an S3 error alias is not Azure absence evidence");
+                    assert_eq!(err.class_label(), expected, "{operation} {selector:?} HTTP {status}: {err:?}");
+                    assert_eq!(err.is_retryable(), retryable, "{operation} {selector:?} HTTP {status}: {err:?}");
+                    if status == 500 {
+                        assert!(matches!(err, SourceError::ServerError(500)));
+                    }
+                    assert!(!err.to_string().contains("untrusted-error-body"));
+                    let (method, mut target) = match operation {
+                        "head" => ("HEAD", "/legacy/missing".to_string()),
+                        "get" => ("GET", "/legacy/missing".to_string()),
+                        "list" => ("GET", "/legacy?restype=container&comp=list".to_string()),
+                        "tags" => ("GET", "/legacy/missing?comp=tags".to_string()),
+                        "probe" => ("HEAD", "/legacy?restype=container".to_string()),
+                        _ => unreachable!(),
+                    };
+                    if let Some(selector) = selector {
+                        target.push_str(&format!("?{selector}=old-version"));
+                    }
+                    assert_requests(&recorded, &[(method, target.as_str())]);
+                }
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn ambiguous_head_preserves_the_container_probe_failure() {
+        for (status, expected, retryable) in [
+            (403, "access_denied", false),
+            (404, "other", false),
+            (429, "throttled", true),
+            (500, "server_error", true),
+            (503, "throttled", true),
+        ] {
+            let (endpoint, recorded) = scripted_server(vec![
+                ScriptedResponse::new(404, Vec::new(), String::new()),
+                // A BlobNotFound header on a container request cannot prove
+                // that the object is missing, regardless of this status.
+                ScriptedResponse::new(status, vec![(HEADER_ERROR_CODE, "BlobNotFound".to_string())], String::new()),
+            ])
+            .await;
+            let err = backend(&endpoint, Credential::SharedKey(vec![7_u8; 32]))
+                .head("missing")
+                .await
+                .expect_err("failed probe must not become object absence");
+            assert_eq!(err.class_label(), expected, "probe {status}: {err:?}");
+            assert_eq!(err.is_retryable(), retryable, "probe {status}: {err:?}");
+            if status == 500 {
+                assert!(matches!(err, SourceError::ServerError(500)));
+            }
+            assert_requests(&recorded, &[("HEAD", "/legacy/missing"), ("HEAD", "/legacy?restype=container")]);
+        }
+    }
+
+    #[tokio::test]
+    async fn version_and_snapshot_absence_are_not_missing_current_blobs() {
+        for selector in ["versionid", "snapshot"] {
+            for code in [None, Some("BlobNotFound"), Some("ResourceNotFound")] {
+                for method in [Method::HEAD, Method::GET] {
+                    let headers = code
+                        .map(|value| vec![(HEADER_ERROR_CODE, value.to_string())])
+                        .unwrap_or_default();
+                    let (endpoint, recorded) = scripted_server(vec![ScriptedResponse::new(404, headers, String::new())]).await;
+                    let backend = backend(&endpoint, Credential::Sas(vec![(selector.to_string(), "old-version".to_string())]));
+                    let result = if method == Method::HEAD {
+                        backend.head("object").await.map(|_| ())
+                    } else {
+                        backend.get("object", None).await.map(|_| ())
+                    };
+                    let err = result.expect_err("missing selected version must remain a source error");
+                    assert!(matches!(err, SourceError::Other(_)), "{method} {selector} {code:?}: {err:?}");
+                    assert_requests(&recorded, &[(method.as_str(), &format!("/legacy/object?{selector}=old-version"))]);
+                }
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn blob_not_found_header_is_not_object_absence_for_list_or_tags() {
+        for tags in [false, true] {
+            let (endpoint, recorded) = scripted_server(vec![ScriptedResponse::new(
+                404,
+                vec![(HEADER_ERROR_CODE, "BlobNotFound".to_string())],
+                String::new(),
+            )])
+            .await;
+            let backend = backend(&endpoint, Credential::SharedKey(vec![7_u8; 32]));
+            let result = if tags {
+                backend.tagging("missing").await.map(|_| ())
+            } else {
+                backend.list(&SourceListRequest::default()).await.map(|_| ())
+            };
+            assert!(matches!(result, Err(SourceError::Other(_))), "tags={tags}: {result:?}");
+            assert_requests(
+                &recorded,
+                &[(
+                    "GET",
+                    if tags {
+                        "/legacy/missing?comp=tags"
+                    } else {
+                        "/legacy?restype=container&comp=list"
+                    },
+                )],
+            );
+        }
+    }
+
     #[tokio::test]
     async fn azure_backend_satisfies_the_shared_backend_contract() {
         let mut ranged = contract_blob_headers();
@@ -998,7 +1335,7 @@ mod tests {
         // A HEAD reports the object size with no body, exactly as Azure does.
         let mut head_only = contract_blob_headers();
         head_only.push(("Content-Length", "5".to_string()));
-        let (endpoint, _) = scripted_server(vec![
+        let (endpoint, recorded) = scripted_server(vec![
             ScriptedResponse::new(200, head_only, String::new()),
             ScriptedResponse::new(200, contract_blob_headers(), "hello".to_string()),
             ScriptedResponse::new(206, ranged, "ell".to_string()),
@@ -1028,6 +1365,23 @@ mod tests {
             },
         )
         .await;
+        assert_requests(
+            &recorded,
+            &[
+                ("HEAD", "/legacy/dir/a.txt"),
+                ("GET", "/legacy/dir/a.txt"),
+                ("GET", "/legacy/dir/a.txt"),
+                ("GET", "/legacy?restype=container&comp=list&prefix=dir%2F&delimiter=%2F&maxresults=2"),
+                (
+                    "GET",
+                    "/legacy?restype=container&comp=list&prefix=dir%2F&delimiter=%2F&marker=cursor-1&maxresults=2",
+                ),
+                ("GET", "/legacy/dir/a.txt?comp=tags"),
+                ("HEAD", "/legacy?restype=container"),
+                ("HEAD", "/legacy/missing"),
+                ("HEAD", "/legacy/secret"),
+            ],
+        );
     }
 
     #[tokio::test]

rustfs/src/on_demand_migration/config.rs

@@ -106,12 +106,12 @@ pub struct SourceConfig {
     pub tls: TlsConfig,
     /// Required for [`Provider::Azure`] and rejected for every other
     /// provider.
-    #[serde(default)]
+    #[serde(default, skip_serializing_if = "Option::is_none")]
     pub azure: Option<AzureSourceConfig>,
     /// Required for [`Provider::GcsNative`] and rejected for every other
     /// provider. [`Provider::Gcs`] keeps using `credentials` because it
     /// speaks the S3 interoperability API.
-    #[serde(default)]
+    #[serde(default, skip_serializing_if = "Option::is_none")]
     pub gcs: Option<GcsSourceConfig>,
 }
 
@@ -808,6 +808,10 @@ impl EndpointKey {
 mod tests {
     use super::*;
 
+    mod before_native_sources {
+        include!("../../fixtures/on_demand_migration/source_config_e2a.rs");
+    }
+
     const FULL_JSON: &str = r#"{
   "version": 1,
   "enabled": true,
@@ -878,6 +882,32 @@ mod tests {
         assert_eq!(minimal.policy.source_timeout.first_byte_ms, 15_000);
     }
 
+    #[test]
+    fn s3_config_writes_remain_readable_by_the_strict_pre_native_reader() {
+        // FULL_JSON is the complete config fixture already present in e2a921bc.
+        for provider in ["s3", "aws", "minio", "rustfs", "r2", "gcs"] {
+            let mut old_wire: serde_json::Value = serde_json::from_str(FULL_JSON).expect("historical config fixture");
+            old_wire["source"]["provider"] = provider.into();
+            let config = OnDemandMigrationConfig::from_json(&serde_json::to_vec(&old_wire).expect("historical wire"))
+                .expect("current reader accepts the historical source");
+            let wire = config.to_json().expect("persist current config");
+            let actual: serde_json::Value = serde_json::from_slice(&wire).expect("persisted config JSON");
+            let old_source: before_native_sources::SourceConfig = serde_json::from_value(actual["source"].clone())
+                .expect("an existing S3 source must remain readable by the strict e2a source consumer");
+            assert_eq!(serde_json::to_value(old_source).expect("old reader wire"), old_wire["source"]);
+            assert_eq!(actual, old_wire, "provider={provider}: no existing config field or value may change");
+
+            for field in ["azure", "gcs"] {
+                let mut rejected = old_wire["source"].clone();
+                rejected[field] = serde_json::Value::Null;
+                assert!(
+                    serde_json::from_value::<before_native_sources::SourceConfig>(rejected).is_err(),
+                    "the frozen old reader must reject {field}, even when null"
+                );
+            }
+        }
+    }
+
     #[test]
     fn unknown_fields_are_rejected_at_every_level() {
         for (label, json) in [
@@ -1080,6 +1110,19 @@ mod tests {
         for cfg in [azure_cfg(), gcs_native_cfg()] {
             let json = cfg.to_json().expect("config must serialize");
             assert_eq!(OnDemandMigrationConfig::from_json(&json).expect("config must parse"), cfg);
+            let wire: serde_json::Value = serde_json::from_slice(&json).expect("native config JSON");
+            let (present, absent, expected) = match cfg.source.provider {
+                Provider::Azure => ("azure", "gcs", serde_json::to_value(&cfg.source.azure).expect("Azure block")),
+                Provider::GcsNative => ("gcs", "azure", serde_json::to_value(&cfg.source.gcs).expect("GCS block")),
+                _ => unreachable!("native fixture"),
+            };
+            assert!(expected.is_object(), "native credentials must be present");
+            assert_eq!(wire["source"][present], expected);
+            assert!(wire["source"].get(absent).is_none());
+            assert!(
+                serde_json::from_value::<before_native_sources::SourceConfig>(wire["source"].clone()).is_err(),
+                "native providers still require upgraded readers"
+            );
         }
         // The wire labels are part of the admin contract.
         assert!(

rustfs/src/on_demand_migration/gcs.rs

@@ -55,10 +55,6 @@ use url::Url;
 /// Read-only object scope: this backend never writes to the source.
 const READ_ONLY_SCOPE: &str = "https://www.googleapis.com/auth/devstorage.read_only";
 const METADATA_PREFIX: &str = "x-goog-meta-";
-/// GCS reports its error code in the response body, not a header; the shared
-/// transport takes a header name, so it is given one that never matches and
-/// classification falls back to the status.
-const NO_ERROR_CODE_HEADER: &str = "x-goog-unused-error-code";
 /// One `objects.list` page is small; refuse an unbounded document.
 const MAX_JSON_BYTES: usize = 8 * 1024 * 1024;
 
@@ -125,7 +121,7 @@ impl GcsNativeSourceBackend {
     }
 
     async fn send_object(&self, request: reqwest::Request) -> Result<reqwest::Response, SourceError> {
-        match self.http.send_object(request, NO_ERROR_CODE_HEADER).await {
+        match self.http.send_object(request, None).await {
             Err(SourceError::NotFound) => {
                 // An XML object URL also returns 404 when its bucket is gone.
                 // Reuse the read-only listing probe before caching a key miss.
@@ -225,7 +221,7 @@ impl SourceBackend for GcsNativeSourceBackend {
         }
 
         let request = self.request(Method::GET, url, HeaderMap::new()).await?;
-        let response = self.http.send(request, NO_ERROR_CODE_HEADER).await?;
+        let response = self.http.send(request, None).await?;
         let body = read_text(response, MAX_JSON_BYTES).await?;
         parse_objects_list(&body)
     }
@@ -245,7 +241,7 @@ impl SourceBackend for GcsNativeSourceBackend {
         let mut url = self.objects_url()?;
         url.query_pairs_mut().append_pair("maxResults", "1");
         let request = self.request(Method::GET, url, HeaderMap::new()).await?;
-        let response = self.http.send(request, NO_ERROR_CODE_HEADER).await?;
+        let response = self.http.send(request, None).await?;
         read_text(response, MAX_JSON_BYTES)
             .await
             .and_then(|body| parse_objects_list(&body))?;
@@ -284,28 +280,38 @@ struct ListedObject {
 fn parse_objects_list(body: &str) -> Result<SourcePage, SourceError> {
     let listing: ObjectsList =
         serde_json::from_str(body).map_err(|err| SourceError::Other(format!("source listing is not valid JSON: {err}")))?;
+    if listing.prefixes.iter().any(|prefix| prefix.is_empty()) {
+        return Err(SourceError::Other("source listing prefix has no name".to_string()));
+    }
     let next_continuation_token = listing.next_page_token.filter(|token| !token.is_empty());
     let objects = listing
         .items
         .into_iter()
         .map(|item| {
+            if item.name.is_empty() {
+                return Err(SourceError::Other("source listing object has no name".to_string()));
+            }
+            let size = item
+                .size
+                .and_then(|size| size.parse::<u64>().ok())
+                .ok_or_else(|| SourceError::Other("source listing object has no valid size".to_string()))?;
             let etag = item
                 .md5_hash
                 .as_deref()
                 .and_then(base64_md5_to_hex)
                 .or_else(|| item.etag.map(|etag| etag.trim_matches('"').to_string()))
                 .filter(|etag| !etag.is_empty());
-            SourceObject {
+            Ok(SourceObject {
                 key: item.name,
                 etag,
-                size: item.size.and_then(|size| size.parse().ok()).unwrap_or(0),
+                size,
                 last_modified: item.updated.as_deref().and_then(parse_http_timestamp),
                 storage_class: item.storage_class,
                 // GCS never encodes a part count in a digest or an ETag.
                 is_multipart_etag: false,
-            }
+            })
         })
-        .collect();
+        .collect::<Result<_, SourceError>>()?;
 
     Ok(SourcePage {
         objects,
@@ -319,7 +325,7 @@ fn parse_objects_list(body: &str) -> Result<SourcePage, SourceError> {
 mod tests {
     use super::*;
     use crate::on_demand_migration::backend_contract::{BackendCapabilities, assert_backend_contract};
-    use crate::on_demand_migration::test_http_fixture::{ScriptedResponse, scripted_server};
+    use crate::on_demand_migration::test_http_fixture::{ScriptedResponse, assert_requests, scripted_server};
     use google_cloud_auth::credentials::anonymous::Builder as AnonymousBuilder;
 
     const LIST_PAGE_ONE: &str = r#"{
@@ -561,4 +567,195 @@ mod tests {
             }
         }
     }
+
+    #[tokio::test]
+    async fn native_listing_rejects_missing_or_invalid_required_object_fields() {
+        for entry in [
+            r#"{"size":"1"}"#,
+            r#"{"name":"","size":"1"}"#,
+            r#"{"name":"broken"}"#,
+            r#"{"name":"broken","size":null}"#,
+            r#"{"name":"broken","size":""}"#,
+            r#"{"name":"broken","size":"-1"}"#,
+            r#"{"name":"broken","size":"18446744073709551616"}"#,
+            r#"{"name":"broken","size":"not-a-size"}"#,
+            r#"{"name":"broken","size":1}"#,
+        ] {
+            let body = format!(r#"{{"items":[{{"name":"valid","size":"1"}},{entry}],"nextPageToken":"next"}}"#);
+            let (endpoint, recorded) = scripted_server(vec![ScriptedResponse::new(200, Vec::new(), body)]).await;
+            let err = backend(&endpoint)
+                .list(&SourceListRequest {
+                    prefix: Some("dir/"),
+                    delimiter: Some("/"),
+                    continuation_token: Some("opaque+/="),
+                    max_keys: 2,
+                    ..Default::default()
+                })
+                .await
+                .expect_err("malformed object must reject the complete native page");
+            assert!(matches!(err, SourceError::Other(_)), "{entry}: {err:?}");
+            assert!(!err.is_retryable());
+            assert_requests(
+                &recorded,
+                &[(
+                    "GET",
+                    "/storage/v1/b/legacy/o?prefix=dir%2F&delimiter=%2F&pageToken=opaque%2B%2F%3D&maxResults=2",
+                )],
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn native_listing_rejects_empty_prefix_entries() {
+        for body in [
+            r#"{"items":[{"name":"valid","size":"1"}],"prefixes":[""],"nextPageToken":"next"}"#,
+            r#"{"prefixes":[""],"nextPageToken":"next"}"#,
+            r#"{"prefixes":["目录/子/",""],"nextPageToken":"next"}"#,
+        ] {
+            let (endpoint, recorded) = scripted_server(vec![ScriptedResponse::new(200, Vec::new(), body.to_string())]).await;
+            let result = backend(&endpoint)
+                .list(&SourceListRequest {
+                    delimiter: Some("/"),
+                    continuation_token: Some("opaque+/="),
+                    max_keys: 2,
+                    ..Default::default()
+                })
+                .await;
+            let err = result.expect_err("an empty prefix must reject the entire page and its cursor");
+            assert!(matches!(err, SourceError::Other(_)), "{body}: {err:?}");
+            assert!(!err.is_retryable());
+            assert_requests(
+                &recorded,
+                &[("GET", "/storage/v1/b/legacy/o?delimiter=%2F&pageToken=opaque%2B%2F%3D&maxResults=2")],
+            );
+        }
+
+        let body = r#"{"prefixes":["目录/子/"],"nextPageToken":"opaque+/="}"#;
+        let (endpoint, recorded) = scripted_server(vec![ScriptedResponse::new(200, Vec::new(), body.to_string())]).await;
+        let page = backend(&endpoint)
+            .list(&SourceListRequest {
+                delimiter: Some("/"),
+                max_keys: 1,
+                ..Default::default()
+            })
+            .await
+            .expect("a valid prefix-only page must remain usable");
+        assert!(page.objects.is_empty());
+        assert_eq!(page.common_prefixes, ["目录/子/"]);
+        assert!(page.is_truncated);
+        assert_eq!(page.next_continuation_token.as_deref(), Some("opaque+/="));
+        assert_requests(&recorded, &[("GET", "/storage/v1/b/legacy/o?delimiter=%2F&maxResults=1")]);
+    }
+
+    #[tokio::test]
+    async fn native_listing_preserves_zero_size_unicode_prefixes_and_opaque_cursors() {
+        let body = r#"{"items":[{"name":"目录/空 & file","size":"0"}],"prefixes":["目录/子/"],"nextPageToken":"opaque+/="}"#;
+        let (endpoint, recorded) = scripted_server(vec![ScriptedResponse::new(200, Vec::new(), body.to_string())]).await;
+        let page = backend(&endpoint)
+            .list(&SourceListRequest {
+                max_keys: 2,
+                ..Default::default()
+            })
+            .await
+            .expect("valid native page");
+        assert_eq!(page.objects.len(), 1);
+        assert_eq!(page.objects[0].key, "目录/空 & file");
+        assert_eq!(page.objects[0].size, 0);
+        assert_eq!(page.common_prefixes, ["目录/子/"]);
+        assert!(page.is_truncated);
+        assert_eq!(page.next_continuation_token.as_deref(), Some("opaque+/="));
+        assert_requests(&recorded, &[("GET", "/storage/v1/b/legacy/o?maxResults=2")]);
+    }
+
+    #[tokio::test]
+    async fn missing_object_head_requires_one_successful_bucket_probe() {
+        for (status, body, expected, retryable) in [
+            (200, "{}", "not_found", false),
+            (403, "", "access_denied", false),
+            (404, "", "other", false),
+            (429, "", "throttled", true),
+            (500, "", "server_error", true),
+            (503, "", "throttled", true),
+            (200, "not JSON", "other", false),
+        ] {
+            let (endpoint, recorded) = scripted_server(vec![
+                ScriptedResponse::new(404, Vec::new(), String::new()),
+                ScriptedResponse::new(status, Vec::new(), body.to_string()),
+            ])
+            .await;
+            let err = backend(&endpoint).head("missing").await.expect_err("missing HEAD must fail");
+            assert_eq!(err.class_label(), expected, "probe {status} {body:?}: {err:?}");
+            assert_eq!(err.is_retryable(), retryable, "probe {status} {body:?}: {err:?}");
+            if status == 500 {
+                assert!(matches!(err, SourceError::ServerError(500)));
+            }
+            assert_requests(&recorded, &[("HEAD", "/legacy/missing"), ("GET", "/storage/v1/b/legacy/o?maxResults=1")]);
+        }
+    }
+
+    #[tokio::test]
+    async fn denied_object_reads_do_not_probe_or_become_object_absence() {
+        for method in [Method::HEAD, Method::GET] {
+            let (endpoint, recorded) = scripted_server(vec![ScriptedResponse::new(
+                403,
+                vec![("x-goog-unused-error-code", "NoSuchKey".to_string())],
+                "untrusted-error-body".to_string(),
+            )])
+            .await;
+            let backend = backend(&endpoint);
+            let result = if method == Method::HEAD {
+                backend.head("missing").await.map(|_| ())
+            } else {
+                backend.get("missing", None).await.map(|_| ())
+            };
+            let err = result.expect_err("denied object read must remain a failure");
+            assert_eq!(err.class_label(), "access_denied");
+            assert!(!err.is_retryable());
+            assert!(!err.to_string().contains("untrusted-error-body"));
+            assert_requests(&recorded, &[(method.as_str(), "/legacy/missing")]);
+        }
+    }
+    #[tokio::test]
+    async fn non_object_errors_ignore_untrusted_error_code_headers() {
+        for probe in [false, true] {
+            for (status, expected, retryable) in [(403, "access_denied", false), (500, "server_error", true)] {
+                let (endpoint, recorded) = scripted_server(vec![ScriptedResponse::new(
+                    status,
+                    vec![("x-goog-unused-error-code", "NoSuchKey".to_string())],
+                    "untrusted-error-body".to_string(),
+                )])
+                .await;
+                let backend = backend(&endpoint);
+                let result = if probe {
+                    backend.probe().await
+                } else {
+                    backend
+                        .list(&SourceListRequest {
+                            max_keys: 2,
+                            ..Default::default()
+                        })
+                        .await
+                        .map(|_| ())
+                };
+                let err = result.expect_err("a synthetic provider header cannot change the source status");
+                assert_eq!(err.class_label(), expected, "probe={probe} status={status}: {err:?}");
+                assert_eq!(err.is_retryable(), retryable);
+                assert!(!err.to_string().contains("untrusted-error-body"));
+                if status == 500 {
+                    assert!(matches!(err, SourceError::ServerError(500)));
+                }
+                assert_requests(
+                    &recorded,
+                    &[(
+                        "GET",
+                        if probe {
+                            "/storage/v1/b/legacy/o?maxResults=1"
+                        } else {
+                            "/storage/v1/b/legacy/o?maxResults=2"
+                        },
+                    )],
+                );
+            }
+        }
+    }
 }

rustfs/src/on_demand_migration/list_through.rs

@@ -94,13 +94,16 @@ pub struct MergePick {
 }
 
 /// The continuation-token envelope. Opaque to clients: it is serialized as
-/// framed JSON and then base64-encoded by the same helper as a local marker.
+/// JSON, optionally framed, then base64-encoded like a local marker.
 ///
 /// A `null` cursor with `done = false` means "list that side from the start";
 /// `done = true` means the side is finished and must not be listed again.
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(deny_unknown_fields)]
 pub struct ListThroughToken {
+    /// Transport framing observed by the decoder, never an envelope field.
+    #[serde(skip)]
+    pub framed: bool,
     /// Envelope marker, always [`LIST_THROUGH_TOKEN_TAG`].
     pub t: String,
     pub v: u32,
@@ -127,6 +130,7 @@ pub struct ListThroughToken {
 impl ListThroughToken {
     fn new(local: SideCursor, source: SideCursor, last_key: Option<String>) -> Self {
         Self {
+            framed: false,
             t: LIST_THROUGH_TOKEN_TAG.to_string(),
             v: LIST_THROUGH_TOKEN_VERSION,
             local: local.token,
@@ -141,7 +145,12 @@ impl ListThroughToken {
     pub fn encode(&self) -> String {
         // The envelope is built here from owned strings, so serialization
         // cannot fail; the fallback keeps the signature infallible.
-        format!("{LIST_THROUGH_TOKEN_PREFIX}{}", serde_json::to_string(self).unwrap_or_default())
+        let json = serde_json::to_string(self).unwrap_or_default();
+        if self.framed {
+            format!("{LIST_THROUGH_TOKEN_PREFIX}{json}")
+        } else {
+            json
+        }
     }
 }
 
@@ -165,16 +174,30 @@ pub enum ListThroughTokenError {
 
 /// Classifies an already base64-decoded continuation token.
 ///
-/// Only a framed JSON object is read as a merged token;
-/// anything else is a local marker, so a bucket that turns `list_through` off
-/// keeps paginating with the tokens it handed out. A token that *is* an
-/// envelope but was tampered with (unknown version, unknown field, truncated
-/// JSON) is an error, never a silent fallback.
+/// Framed envelopes and complete historical writer envelopes are merged tokens.
+/// Partial JSON-shaped keys remain local markers. A key identical to a complete
+/// historical envelope is inherently ambiguous and retains merged semantics.
+/// Recognized envelopes share the same version, count and field validation.
 pub fn decode_continuation_token(decoded: &str) -> Result<ListThroughCursor, ListThroughTokenError> {
-    let Some(payload) = decoded.strip_prefix(LIST_THROUGH_TOKEN_PREFIX) else {
-        return Ok(ListThroughCursor::Local(decoded.to_string()));
+    let (payload, framed) = match decoded.strip_prefix(LIST_THROUGH_TOKEN_PREFIX) {
+        Some(payload) => (payload, true),
+        None if decoded.starts_with('{') => (decoded, false),
+        None => return Ok(ListThroughCursor::Local(decoded.to_string())),
     };
-    let value = serde_json::from_str::<serde_json::Value>(payload).map_err(|_| ListThroughTokenError::Malformed)?;
+    let value = match serde_json::from_str::<serde_json::Value>(payload) {
+        Ok(value) => value,
+        Err(_) if framed => return Err(ListThroughTokenError::Malformed),
+        Err(_) => return Ok(ListThroughCursor::Local(decoded.to_string())),
+    };
+    // RUSTFS_COMPAT_TODO(odm-list-bare-envelope): old writers issued bare JSON. Remove after all supported readers understand framing and outstanding bare listings have drained or explicitly restarted.
+    if !framed
+        && (value.get("t").and_then(serde_json::Value::as_str) != Some(LIST_THROUGH_TOKEN_TAG)
+            || ["v", "local", "local_done", "source", "source_done", "last_key"]
+                .iter()
+                .any(|field| value.get(field).is_none()))
+    {
+        return Ok(ListThroughCursor::Local(decoded.to_string()));
+    }
     if value.get("t").and_then(serde_json::Value::as_str) != Some(LIST_THROUGH_TOKEN_TAG) {
         return Err(ListThroughTokenError::Malformed);
     }
@@ -198,7 +221,10 @@ pub fn decode_continuation_token(decoded: &str) -> Result<ListThroughCursor, Lis
         None => return Err(ListThroughTokenError::Malformed),
     }
     serde_json::from_value::<ListThroughToken>(value)
-        .map(|token| ListThroughCursor::Merged(Box::new(token)))
+        .map(|mut token| {
+            token.framed = framed;
+            ListThroughCursor::Merged(Box::new(token))
+        })
         .map_err(|_| ListThroughTokenError::Malformed)
 }
 
@@ -797,6 +823,7 @@ mod tests {
     #[test]
     fn a_degraded_page_keeps_the_source_cursor_for_the_next_one() {
         let resume = ListThroughToken {
+            framed: false,
             t: LIST_THROUGH_TOKEN_TAG.to_string(),
             v: LIST_THROUGH_TOKEN_VERSION,
             local: Some("local-1".to_string()),
@@ -1033,7 +1060,7 @@ mod tests {
 
     #[test]
     fn token_round_trips_and_rejects_tampering() {
-        let token = ListThroughToken::new(
+        let mut token = ListThroughToken::new(
             SideCursor {
                 token: Some("l".to_string()),
                 done: false,
@@ -1041,6 +1068,7 @@ mod tests {
             SideCursor { token: None, done: true },
             Some("k".to_string()),
         );
+        token.framed = true;
         let encoded = token.encode();
         assert_eq!(decode_continuation_token(&encoded), Ok(ListThroughCursor::Merged(Box::new(token))));
 
@@ -1089,35 +1117,108 @@ mod tests {
 
     #[test]
     fn progress_tokens_preserve_v1_bytes_and_validate_v2_counts() {
-        fn framed(payload: &str) -> String {
-            format!("{LIST_THROUGH_TOKEN_PREFIX}{payload}")
-        }
-
         let token = progress_token(None, true, false);
         assert_eq!(
             token.encode(),
-            concat!(
-                "\0odm-list:",
-                r#"{"t":"odm-list","v":1,"local":null,"local_done":true,"source":"A","source_done":false,"last_key":"last-key"}"#
-            )
+            r#"{"t":"odm-list","v":1,"local":null,"local_done":true,"source":"A","source_done":false,"last_key":"last-key"}"#
         );
-        for count in 1..MAX_LIST_NO_PROGRESS_PAGES {
-            let token = progress_token(Some(count), true, false);
-            assert_eq!(decode_continuation_token(&token.encode()), Ok(ListThroughCursor::Merged(Box::new(token))));
-        }
-        for version in [1, 2] {
-            for value in ["null", "0", "16", "-1", "1.5", "256", "18446744073709551616", "\"1\""] {
-                let encoded = framed(&format!(r#"{{"t":"odm-list","v":{version},"no_progress":{value}}}"#));
+        for framed in [false, true] {
+            let prefix = if framed { LIST_THROUGH_TOKEN_PREFIX } else { "" };
+            for count in 1..MAX_LIST_NO_PROGRESS_PAGES {
+                let mut token = progress_token(Some(count), true, false);
+                token.framed = framed;
+                assert_eq!(decode_continuation_token(&token.encode()), Ok(ListThroughCursor::Merged(Box::new(token))));
+            }
+            // Bare recognition requires the complete shape emitted by old writers;
+            // partial JSON objects are also valid local keys.
+            for version in [1, 2] {
+                for value in ["null", "0", "16", "-1", "1.5", "256", "18446744073709551616", "\"1\""] {
+                    let encoded = format!(
+                        r#"{prefix}{{"t":"odm-list","v":{version},"local":null,"local_done":true,"source":"A","source_done":false,"last_key":"last-key","no_progress":{value}}}"#
+                    );
+                    assert_eq!(decode_continuation_token(&encoded), Err(ListThroughTokenError::Malformed), "{encoded}");
+                }
+            }
+            for encoded in [
+                r#"{"t":"odm-list","v":1,"local":null,"local_done":true,"source":"A","source_done":false,"last_key":"last-key","no_progress":1}"#,
+                r#"{"t":"odm-list","v":2,"local":null,"local_done":true,"source":"A","source_done":false,"last_key":"last-key"}"#,
+                r#"{"t":"odm-list","v":2,"local":null,"local_done":true,"source":"A","source_done":false,"last_key":"last-key","no_progress":1,"extra":true}"#,
+                r#"{"t":"odm-list","v":2,"local":null,"local_done":true,"source":"A","source_done":false,"last_key":"last-key","no_progress":1,"framed":true}"#,
+            ] {
+                let encoded = format!("{prefix}{encoded}");
                 assert_eq!(decode_continuation_token(&encoded), Err(ListThroughTokenError::Malformed), "{encoded}");
             }
+            let bumped = format!("{prefix}{}", token.encode().replace("\"v\":1", "\"v\":9"));
+            assert_eq!(decode_continuation_token(&bumped), Err(ListThroughTokenError::UnsupportedVersion(9)));
         }
-        for payload in [
-            r#"{"t":"odm-list","v":1,"no_progress":1}"#,
-            r#"{"t":"odm-list","v":2}"#,
-            r#"{"t":"odm-list","v":2,"no_progress":1,"extra":true}"#,
+    }
+
+    // Frozen decoder from 447f3c704, before framing was introduced. Keeping this
+    // independent of the current decoder catches a default-writer rollout break.
+    fn decode_before_framing(decoded: &str) -> Result<ListThroughCursor, ListThroughTokenError> {
+        if !decoded.starts_with('{') {
+            return Ok(ListThroughCursor::Local(decoded.to_string()));
+        }
+        let Ok(value) = serde_json::from_str::<serde_json::Value>(decoded) else {
+            // Not JSON at all: an object key may legitimately start with '{'.
+            return Ok(ListThroughCursor::Local(decoded.to_string()));
+        };
+        if value.get("t").and_then(serde_json::Value::as_str) != Some(LIST_THROUGH_TOKEN_TAG) {
+            return Ok(ListThroughCursor::Local(decoded.to_string()));
+        }
+        match value.get("v").and_then(serde_json::Value::as_u64) {
+            Some(version) if version == u64::from(LIST_THROUGH_TOKEN_VERSION) => {
+                // v1 readers reject this field even when it is null or zero.
+                if value.get("no_progress").is_some() {
+                    return Err(ListThroughTokenError::Malformed);
+                }
+            }
+            Some(version) if version == u64::from(LIST_THROUGH_PROGRESS_TOKEN_VERSION) => {
+                if !value
+                    .get("no_progress")
+                    .and_then(serde_json::Value::as_u64)
+                    .is_some_and(|count| (1..u64::from(MAX_LIST_NO_PROGRESS_PAGES)).contains(&count))
+                {
+                    return Err(ListThroughTokenError::Malformed);
+                }
+            }
+            Some(version) => return Err(ListThroughTokenError::UnsupportedVersion(version.min(u64::from(u32::MAX)) as u32)),
+            None => return Err(ListThroughTokenError::Malformed),
+        }
+        serde_json::from_value::<ListThroughToken>(value)
+            .map(|token| ListThroughCursor::Merged(Box::new(token)))
+            .map_err(|_| ListThroughTokenError::Malformed)
+    }
+
+    #[test]
+    fn historical_writer_fixtures_and_default_output_remain_readable() {
+        for (wire, version, count) in [
+            (
+                r#"{"t":"odm-list","v":1,"local":"local-2","local_done":false,"source":"source-2","source_done":false,"last_key":"k"}"#,
+                1,
+                None,
+            ),
+            (
+                r#"{"t":"odm-list","v":2,"local":"local-2","local_done":false,"source":"source-2","source_done":false,"last_key":"k","no_progress":15}"#,
+                2,
+                Some(15),
+            ),
         ] {
-            let encoded = framed(payload);
-            assert_eq!(decode_continuation_token(&encoded), Err(ListThroughTokenError::Malformed), "{encoded}");
+            let ListThroughCursor::Merged(mut token) = decode_continuation_token(wire).expect("historical issued token") else {
+                panic!("a historical cursor must not silently become a local marker, even if a key has identical JSON");
+            };
+            assert_eq!(token.local.as_deref(), Some("local-2"));
+            assert_eq!(token.source.as_deref(), Some("source-2"));
+            assert_eq!(token.last_key.as_deref(), Some("k"));
+            assert_eq!(token.v, version);
+            assert_eq!(token.no_progress, count);
+            assert!(!token.framed);
+            assert_eq!(token.encode(), wire, "bare output retains the historical bytes");
+            assert_eq!(decode_before_framing(&token.encode()), Ok(ListThroughCursor::Merged(token.clone())));
+            token.framed = true;
+            let framed = format!("\0odm-list:{wire}");
+            assert_eq!(token.encode(), framed, "framing leaves the JSON payload unchanged");
+            assert_eq!(decode_continuation_token(&framed), Ok(ListThroughCursor::Merged(token)));
         }
     }
 

rustfs/src/on_demand_migration/native_http.rs

@@ -99,6 +99,7 @@ impl NativeHttp {
     pub(super) fn for_test(endpoint: Url) -> Self {
         Self {
             client: reqwest::Client::builder()
+                .no_proxy()
                 .redirect(reqwest::redirect::Policy::none())
                 .build()
                 .expect("test http client should build"),
@@ -122,13 +123,13 @@ impl NativeHttp {
     }
 
     /// Sends the request and returns the response only for a 2xx status.
-    /// Non-2xx statuses are classified from the status and the provider's own
+    /// Non-2xx statuses are classified from the status and an optional provider
     /// error-code header; response bodies are not read, so no provider message
     /// can smuggle credentials or markup into a log line.
     pub(super) async fn send(
         &self,
         request: reqwest::Request,
-        error_code_header: &str,
+        error_code_header: Option<&str>,
     ) -> Result<reqwest::Response, SourceError> {
         self.send_classified(request, error_code_header, false).await
     }
@@ -137,7 +138,7 @@ impl NativeHttp {
     pub(super) async fn send_object(
         &self,
         request: reqwest::Request,
-        error_code_header: &str,
+        error_code_header: Option<&str>,
     ) -> Result<reqwest::Response, SourceError> {
         self.send_classified(request, error_code_header, true).await
     }
@@ -145,26 +146,42 @@ impl NativeHttp {
     async fn send_classified(
         &self,
         request: reqwest::Request,
-        error_code_header: &str,
+        error_code_header: Option<&str>,
         not_found_on_404_without_code: bool,
     ) -> Result<reqwest::Response, SourceError> {
-        let response = self.client.execute(request).await.map_err(classify_transport_error)?;
+        let response = self.execute(request).await?;
+        let status = response.status();
+        match Self::check_response(response, error_code_header) {
+            Err(SourceError::Other(_)) if not_found_on_404_without_code && status.as_u16() == 404 => Err(SourceError::NotFound),
+            result => result,
+        }
+    }
+
+    pub(super) async fn execute(&self, request: reqwest::Request) -> Result<reqwest::Response, SourceError> {
+        self.client.execute(request).await.map_err(classify_transport_error)
+    }
+
+    pub(super) fn check_response(
+        response: reqwest::Response,
+        error_code_header: Option<&str>,
+    ) -> Result<reqwest::Response, SourceError> {
         let status = response.status();
         if status.is_success() {
             return Ok(response);
         }
-        let code = response
-            .headers()
-            .get(error_code_header)
+        let code = error_code_header
+            .and_then(|header| response.headers().get(header))
             .and_then(|value| value.to_str().ok())
             .map(str::to_string);
         let message = match &code {
             Some(code) => format!("source returned HTTP {status} ({code})"),
             None => format!("source returned HTTP {status}"),
         };
-        match classify_status(status.as_u16(), code.as_deref(), message) {
-            SourceError::Other(_) if not_found_on_404_without_code && status.as_u16() == 404 => Err(SourceError::NotFound),
-            err => Err(err),
+        match classify_status(status.as_u16(), code.as_deref(), message.clone()) {
+            // Native object absence needs provider-specific evidence or a
+            // successful bucket probe, never an alias from the S3 classifier.
+            SourceError::NotFound => Err(classify_status(status.as_u16(), None, message)),
+            error => Err(error),
         }
     }
 }

rustfs/src/on_demand_migration/source_client.rs

@@ -337,7 +337,7 @@ const THROTTLE_CODES: &[&str] = &[
     "RequestThrottled",
     "ServerBusy",
 ];
-const NOT_FOUND_CODES: &[&str] = &["NoSuchKey", "BlobNotFound"];
+const NOT_FOUND_CODES: &[&str] = &["NoSuchKey"];
 const ACCESS_DENIED_CODES: &[&str] = &[
     "AccessDenied",
     "InvalidAccessKeyId",
@@ -1813,10 +1813,11 @@ mod tests {
 
     /// The S3 backend behind the scripted connector, without the prefix-mapping
     /// client on top: the contract is a property of the backend itself.
-    async fn scripted_s3_backend(responses: Vec<Scripted>) -> S3SourceBackend {
+    async fn scripted_s3_backend(responses: Vec<Scripted>) -> (S3SourceBackend, Recorded) {
         let spec = spec(None);
+        let requests: Recorded = Arc::new(Mutex::new(Vec::new()));
         let connector = SharedHttpConnector::new(ScriptedConnector {
-            requests: Arc::new(Mutex::new(Vec::new())),
+            requests: Arc::clone(&requests),
             responses: Arc::new(Mutex::new(responses.into_iter().collect())),
         });
         let http_client = http_client_fn(move |_settings, _components| connector.clone());
@@ -1826,17 +1827,20 @@ mod tests {
             .expect("test spec should build")
             .http_client(http_client)
             .interceptor(SourceProxyMarkerInterceptor::new());
-        S3SourceBackend {
-            client: S3Client::from_conf(config.build()),
-            bucket: spec.bucket.clone(),
-        }
+        (
+            S3SourceBackend {
+                client: S3Client::from_conf(config.build()),
+                bucket: spec.bucket.clone(),
+            },
+            requests,
+        )
     }
 
     #[tokio::test]
     async fn s3_backend_satisfies_the_shared_backend_contract() {
         let mut ranged = contract_object_headers(3);
         ranged.push(("content-range", "bytes 1-3/5".to_string()));
-        let backend = scripted_s3_backend(vec![
+        let (backend, requests) = scripted_s3_backend(vec![
             ok(contract_object_headers(5), ""),
             ok(contract_object_headers(5), "hello"),
             ok(ranged, "ell"),
@@ -1845,6 +1849,7 @@ mod tests {
             ok(Vec::new(), CONTRACT_TAGGING),
             ok(Vec::new(), ""),
             status(404, ""),
+            // An object HEAD 404 requires the existing S3 bucket HEAD probe.
             ok(Vec::new(), ""),
             status(403, ACCESS_DENIED_BODY),
         ])
@@ -1859,6 +1864,32 @@ mod tests {
             },
         )
         .await;
+        let requests = recorded(&requests);
+        let actual: Vec<_> = requests
+            .iter()
+            .map(|request| {
+                (
+                    request.method.as_str(),
+                    url::Url::parse(&request.uri).expect("recorded S3 URL").path().to_string(),
+                )
+            })
+            .collect();
+        let expected = [
+            ("HEAD", "/source-bucket/dir/a.txt"),
+            ("GET", "/source-bucket/dir/a.txt"),
+            ("GET", "/source-bucket/dir/a.txt"),
+            ("GET", "/source-bucket/"),
+            ("GET", "/source-bucket/"),
+            ("GET", "/source-bucket/dir/a.txt"),
+            ("HEAD", "/source-bucket/"),
+            ("HEAD", "/source-bucket/missing"),
+            ("HEAD", "/source-bucket/"),
+            ("HEAD", "/source-bucket/secret"),
+        ];
+        assert_eq!(actual, expected.map(|(method, path)| (method, path.to_string())));
+        for request in &requests {
+            assert_outbound_markers(request);
+        }
     }
 
     fn prefix_client(prefix: Option<String>) -> SourceClient {

rustfs/src/on_demand_migration/test_http_fixture.rs

@@ -56,6 +56,16 @@ impl RecordedRequest {
 
 pub(super) type Recorder = Arc<Mutex<Vec<RecordedRequest>>>;
 
+/// Checks the full request sequence, including the absence of extra probes.
+pub(super) fn assert_requests(recorder: &Recorder, expected: &[(&str, &str)]) {
+    let recorded = recorder.lock().expect("recorder lock");
+    let actual: Vec<_> = recorded
+        .iter()
+        .map(|request| (request.method.as_str(), request.target.as_str()))
+        .collect();
+    assert_eq!(actual, expected, "unexpected native source request sequence");
+}
+
 /// Binds a loopback listener that answers `responses` in order and returns its
 /// origin plus the recorder. The task ends once the script is exhausted.
 pub(super) async fn scripted_server(responses: Vec<ScriptedResponse>) -> (Url, Recorder) {

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<()> {