test: skip unchanged pool metadata in crash barriers

.config/e2e-distributed-selection.txt

@@ -1,2 +1,2 @@
-sha256-linux=563bff8f1171d6dbe166ff8440310dbe98430e466aa3ecd8dc39e3c872b320f7
-sha256-darwin=563bff8f1171d6dbe166ff8440310dbe98430e466aa3ecd8dc39e3c872b320f7
+sha256-linux=4696a43b167ac608b3b8677027c9fe9fdac3396d37c8cca11dce531c720ac6d2
+sha256-darwin=9785867929047dfd8c6f768e0d2b1e0a8fdba85216f4a4139093b1619d03ff07

.config/make/tests.mak

@@ -32,7 +32,6 @@ script-tests: ## Run shell script tests
 	./scripts/test_hotpath_warp_ab_gate.sh
 	./scripts/test_hotpath_warp_abba.sh
 	./scripts/test_scanner_validation_harness.sh
-	./scripts/test_scanner_heal_w13_mrf_evidence.sh
 	./scripts/test_scanner_heal_w16_recovery_evidence.sh
 	./scripts/test_exact_1mib_handoff_abba.sh
 	./scripts/test_pinned_paired_abba_bench.sh

Cargo.lock

@@ -9928,7 +9928,6 @@ dependencies = [
  "async-trait",
  "base64-simd",
  "bytes",
- "chrono",
  "crc-fast",
  "futures",
  "hotpath",

crates/e2e_test/src/multipart_auth_test.rs

@@ -15,7 +15,6 @@
 //! Regression coverage for anonymous access on multipart control APIs.
 
 use crate::common::{RustFSTestEnvironment, init_logging, local_http_client};
-use crate::kms::common::LocalKMSTestEnvironment;
 use async_compression::tokio::write::{BzEncoder, Lz4Encoder, XzEncoder};
 use aws_sdk_s3::error::{ProvideErrorMetadata, SdkError};
 use aws_sdk_s3::operation::head_object::HeadObjectOutput;
@@ -1466,10 +1465,10 @@ async fn test_anonymous_post_object_uses_bucket_default_sse_s3() -> Result<(), B
 async fn test_anonymous_post_object_uses_bucket_default_sse_kms() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     init_logging();
 
-    let mut kms_env = LocalKMSTestEnvironment::new().await?;
-    let default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    kms_env.wait_for_kms_ready().await?;
-    let env = &kms_env.base_env;
+    let mut env = RustFSTestEnvironment::new().await?;
+    let master_key = local_sse_master_key_value();
+    env.start_rustfs_server_with_env(vec![], &[(LOCAL_SSE_MASTER_KEY_ENV, master_key.as_str())])
+        .await?;
 
     let bucket = "anon-post-default-sse-kms";
     let object_key = "post-default-sse-kms-object.txt";
@@ -1485,7 +1484,7 @@ async fn test_anonymous_post_object_uses_bucket_default_sse_kms() -> Result<(),
                 .apply_server_side_encryption_by_default(
                     ServerSideEncryptionByDefault::builder()
                         .sse_algorithm(ServerSideEncryption::AwsKms)
-                        .kms_master_key_id(default_key_id)
+                        .kms_master_key_id("test-key")
                         .build()
                         .expect("default encryption rule should build"),
                 )

crates/ecstore/src/api/mod.rs

@@ -366,6 +366,14 @@ pub mod config {
 }
 
 pub mod data_usage {
+    #[cfg(feature = "test-util")]
+    pub use crate::data_movement::SourceCleanupDeleteBarrier;
+    #[cfg(feature = "test-util")]
+    pub use crate::data_movement::scanner_backlog::test_util::NativeScannerPauseBacklogWriteFault;
+    pub use crate::data_movement::scanner_backlog::{
+        MAX_SCANNER_PAUSE_BACKLOG_BYTES, ScannerPauseBacklogRetirementPlan, ScannerPauseBacklogRetirementPlanner,
+        ScannerPauseBacklogRetirementReplica, register_scanner_pause_backlog_retirement_planner,
+    };
     pub use crate::data_usage::{
         DATA_USAGE_CACHE_NAME, apply_bucket_usage_memory_overlay, compute_bucket_usage,
         init_compression_total_memory_from_backend, invalidate_admin_data_usage_snapshot_cache,

crates/ecstore/src/core/pools_test.rs

@@ -5256,16 +5256,35 @@ mod decommission_lock_order_tests {
 
     #[test]
     #[serial_test::serial]
-    fn scanner_backlog_native_replica_reconciles_capacity_and_cleans_source() {
-        run_large_stack_current_thread_async_test("scanner-backlog-reconcile", async || {
+    fn data_movement_existing_replica_reconciles_capacity_and_cleans_source() {
+        data_movement_existing_replica_reconciles_capacity_case(false);
+    }
+
+    #[test]
+    #[serial_test::serial]
+    fn data_movement_existing_replica_outside_reservation_uses_reserved_target() {
+        data_movement_existing_replica_reconciles_capacity_case(true);
+    }
+
+    fn data_movement_existing_replica_reconciles_capacity_case(existing_outside_reservation: bool) {
+        run_large_stack_current_thread_async_test("reserved-replica-reconcile", async move || {
             let (_temp_dirs, store, other_store) =
                 test_three_pool_stores_with_three_disk_sets_with_isolated_node_contexts(None).await;
-            let object = "buckets/.scanner-pause-backlog.json";
+            let object = "buckets/reserved-replica-routing.json";
             let body = br#"{"schemaVersion":1,"generation":2}"#.to_vec();
             let old_body = br#"{"schemaVersion":1,"generation":1}"#.to_vec();
             let source_time = time::OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(20);
-            let target_time = time::OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(10);
-            for (pool_index, payload, mod_time) in [(0, body.clone(), source_time), (2, old_body, target_time)] {
+            let target_time = source_time;
+            let target_pool_index = if existing_outside_reservation { 1 } else { 2 };
+            let mut replicas = vec![(0, body.clone(), source_time), (target_pool_index, old_body, target_time)];
+            if existing_outside_reservation {
+                replicas.push((
+                    2,
+                    br#"{"schemaVersion":1,"generation":3}"#.to_vec(),
+                    source_time + time::Duration::seconds(10),
+                ));
+            }
+            for (pool_index, payload, mod_time) in replicas.iter().cloned() {
                 store.pools[pool_index]
                     .put_object(
                         RUSTFS_META_BUCKET,
@@ -5278,13 +5297,19 @@ mod decommission_lock_order_tests {
                         },
                     )
                     .await
-                    .expect("seed native scanner replicas with independent write times");
+                    .expect("seed existing replicas with independent write times");
             }
             let layout = DecommissionErasureLayout { data: 1, parity: 0 };
             let target_total = body.len() * 8;
             let capacities = vec![
                 DecommissionPoolCapacityInfo::for_test(0, layout, 0, body.len() * 2, body.len() * 2),
-                DecommissionPoolCapacityInfo::for_test(1, layout, 0, target_total, target_total),
+                DecommissionPoolCapacityInfo::for_test(
+                    1,
+                    layout,
+                    if existing_outside_reservation { target_total } else { 0 },
+                    target_total,
+                    if existing_outside_reservation { 0 } else { target_total },
+                ),
                 DecommissionPoolCapacityInfo::for_test(2, layout, target_total, target_total, 0),
             ];
             set_decommission_capacity_info_overrides_for_test(store.id, vec![capacities.clone()]);
@@ -5293,6 +5318,17 @@ mod decommission_lock_order_tests {
                 .await
                 .expect("activate the source reservation");
             let owner = decommission_capacity_owner(&*store.pool_meta.read().await);
+            let reserved_snapshot = store.pool_meta.read().await.clone();
+            let reservation = reserved_snapshot.pools[0]
+                .decommission
+                .as_ref()
+                .and_then(|info| info.capacity_reservation.as_ref())
+                .expect("active source reservation");
+            assert_eq!(
+                reservation.targets.iter().map(|target| target.pool_index).collect::<Vec<_>>(),
+                vec![target_pool_index],
+                "the fixture must reserve exactly one target"
+            );
             let source_reader = store.pools[0]
                 .get_object_reader(
                     RUSTFS_META_BUCKET,
@@ -5314,12 +5350,91 @@ mod decommission_lock_order_tests {
                 RUSTFS_META_BUCKET.to_string(),
                 source_reader,
                 None,
-                "scanner_backlog_conflict",
+                "reserved_replica_conflict",
                 Some(owner),
             )
             .await
-            .expect_err("a different older native ledger must retain its source and capacity intent");
+            .expect_err("a different older existing record must retain its source and capacity intent");
             assert!(conflict.to_string().contains("Precondition failed"), "unexpected conflict: {conflict}");
+            let reserved_snapshot = store.pool_meta.read().await.clone();
+            let mut selection_opts = ObjectOptions {
+                data_movement: true,
+                src_pool_idx: 0,
+                ..Default::default()
+            };
+            assert_eq!(
+                store
+                    .select_data_movement_pool_idx(RUSTFS_META_BUCKET, object, body.len() as i64, &selection_opts, true)
+                    .await
+                    .expect("selection without a capacity owner retains existing-replica routing"),
+                2
+            );
+            owner.apply_to(&mut selection_opts);
+            for stale_owner in [
+                DecommissionCapacityOwner {
+                    owner_nonce: uuid::Uuid::new_v4(),
+                    ..owner
+                },
+                DecommissionCapacityOwner {
+                    generation: owner.generation + 1,
+                    ..owner
+                },
+            ] {
+                let mut stale_opts = selection_opts.clone();
+                stale_owner.apply_to(&mut stale_opts);
+                assert!(
+                    matches!(
+                        store
+                            .select_data_movement_pool_idx(RUSTFS_META_BUCKET, object, body.len() as i64, &stale_opts, true)
+                            .await,
+                        Err(crate::error::Error::DecommissionCapacityBlocked { .. })
+                    ),
+                    "a stale owner must not fall back to another target"
+                );
+            }
+            {
+                let mut meta = store.pool_meta.write().await;
+                meta.pools[0]
+                    .decommission
+                    .as_mut()
+                    .unwrap()
+                    .capacity_reservation
+                    .as_mut()
+                    .unwrap()
+                    .expires_at = time::OffsetDateTime::now_utc() - time::Duration::seconds(1);
+            }
+            assert!(
+                matches!(
+                    store
+                        .select_data_movement_pool_idx(RUSTFS_META_BUCKET, object, body.len() as i64, &selection_opts, true)
+                        .await,
+                    Err(crate::error::Error::DecommissionCapacityBlocked { .. })
+                ),
+                "an expired owner must not fall back to another target"
+            );
+            *store.pool_meta.write().await = reserved_snapshot.clone();
+            if !existing_outside_reservation {
+                {
+                    let mut meta = store.pool_meta.write().await;
+                    let target = &mut meta.pools[0]
+                        .decommission
+                        .as_mut()
+                        .unwrap()
+                        .capacity_reservation
+                        .as_mut()
+                        .unwrap()
+                        .targets[0];
+                    target.consumed_physical_bytes = target.reserved_physical_bytes;
+                }
+                assert_eq!(
+                    store
+                        .select_data_movement_pool_idx(RUSTFS_META_BUCKET, object, body.len() as i64, &selection_opts, true)
+                        .await
+                        .expect("an existing reserved replica can still be selected after capacity was consumed"),
+                    target_pool_index
+                );
+                *store.pool_meta.write().await = reserved_snapshot;
+            }
             let mut persisted = crate::core::pools::PoolMeta::default();
             persisted
                 .load_no_lock_from_replicas(store.pools.clone())
@@ -5336,11 +5451,24 @@ mod decommission_lock_order_tests {
                     .pending_target_physical_bytes,
                 body.len()
             );
-            let previous = store.pools[2]
+            for (pool_index, payload, mod_time) in &replicas {
+                let mut reader = store.pools[*pool_index]
+                    .get_object_reader(RUSTFS_META_BUCKET, object, None, HeaderMap::new(), &ObjectOptions::default())
+                    .await
+                    .expect("a refused existing record replacement must preserve every replica");
+                assert_eq!(reader.object_info.mod_time, Some(*mod_time));
+                let mut actual = Vec::new();
+                reader
+                    .read_to_end(&mut actual)
+                    .await
+                    .expect("read the unchanged existing record");
+                assert_eq!(&actual, payload);
+            }
+            let previous = store.pools[target_pool_index]
                 .get_object_info(RUSTFS_META_BUCKET, object, &ObjectOptions::default())
                 .await
-                .expect("read the native writer's CAS revision");
-            let replacement = store.pools[2]
+                .expect("read the existing writer's CAS revision");
+            let replacement = store.pools[target_pool_index]
                 .put_object(
                     RUSTFS_META_BUCKET,
                     object,
@@ -5356,7 +5484,7 @@ mod decommission_lock_order_tests {
                     },
                 )
                 .await
-                .expect("native scanner CAS converges the payload without a migration marker");
+                .expect("existing CAS converges the payload without a migration marker");
             assert!(!data_movement::is_owned_data_movement_target(&replacement));
             *other_store.pool_meta.write().await = persisted;
             set_decommission_capacity_info_overrides_for_test(other_store.id, vec![capacities]);
@@ -5374,7 +5502,7 @@ mod decommission_lock_order_tests {
             )
             .await
             .expect("replica conflict recovery must be bounded")
-            .expect("identical native replica should finish migration on the reloaded node");
+            .expect("identical existing replica should finish migration on the reloaded node");
             let mut reconciled = crate::core::pools::PoolMeta::default();
             reconciled
                 .load_no_lock_from_replicas(other_store.pools.clone())
@@ -5404,14 +5532,14 @@ mod decommission_lock_order_tests {
                 .await
                 .expect_err("the source should be cleaned only after equivalent-target capacity reconciliation");
             assert!(crate::error::is_err_object_not_found(&missing));
-            let mut target_reader = other_store.pools[2]
+            let mut target_reader = other_store.pools[target_pool_index]
                 .get_object_reader(RUSTFS_META_BUCKET, object, None, HeaderMap::new(), &ObjectOptions::default())
                 .await
                 .expect("the surviving replica should remain readable");
             assert_eq!(
                 target_reader.object_info.mod_time,
                 Some(target_time),
-                "recovery must not overwrite the native target"
+                "recovery must not overwrite the existing target"
             );
             let mut actual = Vec::new();
             target_reader
@@ -5419,6 +5547,20 @@ mod decommission_lock_order_tests {
                 .await
                 .expect("read surviving ledger bytes");
             assert_eq!(actual, body);
+            if existing_outside_reservation {
+                let (_, outside_body, outside_time) = replicas.last().expect("unreserved existing replica");
+                let mut outside = other_store.pools[2]
+                    .get_object_reader(RUSTFS_META_BUCKET, object, None, HeaderMap::new(), &ObjectOptions::default())
+                    .await
+                    .expect("migration must leave the unreserved existing replica intact");
+                assert_eq!(outside.object_info.mod_time, Some(*outside_time));
+                let mut actual = Vec::new();
+                outside
+                    .read_to_end(&mut actual)
+                    .await
+                    .expect("read the untouched unreserved replica");
+                assert_eq!(&actual, outside_body);
+            }
         });
     }
 

crates/ecstore/src/data_movement/mod.rs

@@ -15,6 +15,7 @@
 // #730: data-movement migration keeps staged cleanup helpers until copy paths converge.
 
 pub(crate) mod backpressure;
+pub(crate) mod scanner_backlog;
 
 use crate::core::pools::{DecommissionCapacityOwner, decommission_capacity_mutation_id};
 use crate::error::{
@@ -984,24 +985,6 @@ fn is_superseding_unversioned_data_movement_object(source: &ObjectInfo, target:
             .is_some_and(|(source_time, target_time)| target_time > source_time)
 }
 
-fn is_equivalent_scanner_backlog_replica(source: &ObjectInfo, target: &ObjectInfo, compare_part_checksums: bool) -> bool {
-    // Scanner publishes this exact payload to surviving sets with CAS. Each
-    // set assigns its own write time; that timestamp is not a ledger generation.
-    // Accept only an identical, known unversioned identity, never a different
-    // record based on timestamp ordering or a similarly named user object.
-    source.bucket == crate::disk::RUSTFS_META_BUCKET
-        && target.bucket == source.bucket
-        && source.name == "buckets/.scanner-pause-backlog.json"
-        && target.name == source.name
-        && is_unversioned_data_movement_object(source)
-        && is_unversioned_data_movement_object(target)
-        && !source.delete_marker
-        && source.mod_time.is_some()
-        && target.mod_time.is_some()
-        && source.etag.as_ref().is_some_and(|etag| !etag.is_empty())
-        && is_equivalent_data_movement_object_identity(source, target, false, compare_part_checksums)
-}
-
 fn is_data_movement_upload_takeover_target(source: &ObjectInfo, target: &ObjectInfo, compare_part_checksums: bool) -> bool {
     let identity = data_movement_upload_identity(source);
     source.mod_time.is_some()
@@ -1217,7 +1200,7 @@ struct SourceCleanupDeleteBarrierState {
     dead_code,
     reason = "installed by set_disk object tests behind `--features test-util` (backlog#1823)"
 )]
-pub(crate) struct SourceCleanupDeleteBarrier {
+pub struct SourceCleanupDeleteBarrier {
     state: Arc<SourceCleanupDeleteBarrierState>,
 }
 
@@ -1231,7 +1214,7 @@ static SOURCE_CLEANUP_DELETE_BARRIERS: std::sync::OnceLock<std::sync::Mutex<Vec<
     reason = "installed by set_disk object tests behind `--features test-util` (backlog#1823)"
 )]
 impl SourceCleanupDeleteBarrier {
-    pub(crate) fn install(bucket: &str, object: &str) -> Self {
+    pub fn install(bucket: &str, object: &str) -> Self {
         let state = Arc::new(SourceCleanupDeleteBarrierState {
             bucket: bucket.to_string(),
             object: object.to_string(),
@@ -1254,7 +1237,7 @@ impl SourceCleanupDeleteBarrier {
         Self { state }
     }
 
-    pub(crate) async fn wait_until_paused(&self) {
+    pub async fn wait_until_paused(&self) {
         tokio::time::timeout(StdDuration::from_secs(30), self.state.arrived.notified())
             .await
             .expect("source cleanup should reach the pre-delete barrier");
@@ -1270,7 +1253,7 @@ impl SourceCleanupDeleteBarrier {
         self.state.is_paused.load(Ordering::Acquire)
     }
 
-    pub(crate) fn release(&self) {
+    pub fn release(&self) {
         self.state.release.notify_one();
     }
 }
@@ -1449,7 +1432,8 @@ fn resolve_data_movement_overwrite_resume_result_for(
     target_pool_idx: usize,
     compare_part_checksums: bool,
 ) -> Result<bool> {
-    if !should_check_data_movement_overwrite_resume(err)
+    if scanner_backlog::is_scanner_pause_backlog(&source.bucket, &source.name)
+        || !should_check_data_movement_overwrite_resume(err)
         || !should_check_data_movement_resume_target(src_pool_idx, target_pool_idx)
     {
         return Ok(false);
@@ -1471,9 +1455,7 @@ fn resolve_data_movement_overwrite_resume_result_for(
         return Ok(true);
     }
 
-    Ok(matches!(err, Error::PreconditionFailed)
-        && (is_equivalent_scanner_backlog_replica(source, &target, compare_part_checksums)
-            || is_superseding_unversioned_data_movement_object(source, &target)))
+    Ok(matches!(err, Error::PreconditionFailed) && is_superseding_unversioned_data_movement_object(source, &target))
 }
 
 #[derive(Clone, Copy)]
@@ -1646,6 +1628,9 @@ async fn migrate_object_inner(
     capacity_owner: Option<DecommissionCapacityOwner>,
     mutation_fence: Option<DecommissionFixedReadAnchor>,
 ) -> Result<()> {
+    if scanner_backlog::is_scanner_pause_backlog(&bucket, &rd.object_info.name) {
+        return Err(Error::other("scanner pause backlog requires native retirement handoff"));
+    }
     let mut mutation_fence = mutation_fence;
     let object_info = rd.object_info.clone();
     let capacity_owner = capacity_owner.map(|owner| {
@@ -3354,16 +3339,25 @@ mod tests {
     }
 
     #[test]
-    fn test_scanner_backlog_resume_accepts_identical_native_replica_with_older_write_time() {
+    fn test_scanner_backlog_resume_requires_native_cohort_proof_even_for_identical_payload() {
         let (source, target) = scanner_backlog_replica_pair();
         assert!(!is_owned_data_movement_target(&target), "native scanner writes are not migration copies");
         assert!(!is_equivalent_data_movement_object(&source, &target));
         assert!(
-            scanner_backlog_precondition_resumes(&source, target),
-            "identical ledger payloads have replica-local write times, not distinct committed generations"
+            !scanner_backlog_precondition_resumes(&source, target),
+            "a single identical replica cannot prove native cohort authority"
         );
     }
 
+    #[test]
+    fn test_scanner_backlog_resume_rejects_newer_timestamp_and_full_single_replica_identity() {
+        let (source, mut target) = scanner_backlog_replica_pair();
+        target.mod_time = source.mod_time.map(|time| time + time::Duration::SECOND);
+        target.etag = Some("different-native-ledger".to_string());
+        assert!(!scanner_backlog_precondition_resumes(&source, target));
+        assert!(!scanner_backlog_precondition_resumes(&source, source.clone()));
+    }
+
     #[test]
     fn test_scanner_backlog_resume_rejects_changed_payload_or_metadata() {
         let (source, target) = scanner_backlog_replica_pair();

crates/ecstore/src/data_movement/scanner_backlog.rs

@@ -0,0 +1,292 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::disk::RUSTFS_META_BUCKET;
+use crate::error::{Error, Result, is_err_object_not_found, is_err_version_not_found};
+use crate::object_api::ObjectOptions;
+use crate::object_api::{ObjectInfo, PutObjReader, WriteCompletion};
+use crate::set_disk::SetDisks;
+use crate::storage_api_contracts::object::HTTPPreconditions;
+use crate::storage_api_contracts::object::ObjectIO as _;
+use futures::future::join_all;
+use http::HeaderMap;
+use std::sync::{Arc, OnceLock};
+use tokio::io::AsyncReadExt;
+
+pub const MAX_SCANNER_PAUSE_BACKLOG_BYTES: u64 = 64 * 1024;
+pub(crate) const SCANNER_PAUSE_BACKLOG_PATH: &str = "buckets/.scanner-pause-backlog.json";
+
+/// A bounded, storage-fenced native replica. Only a confirmed missing object
+/// has no payload; read failures never enter the Scanner verifier.
+pub struct ScannerPauseBacklogRetirementReplica {
+    pub pool_index: usize,
+    pub set_index: usize,
+    pub data: Option<Vec<u8>>,
+}
+
+/// Native records for a membership handoff. Existing durable ledgers are
+/// preserved; an empty native bootstrap may initialize its first ledger.
+pub struct ScannerPauseBacklogRetirementPlan {
+    pub seed_record: Option<Vec<u8>>,
+    pub commit_record: Vec<u8>,
+    pub stable_record: Vec<u8>,
+}
+
+pub type ScannerPauseBacklogRetirementPlanner =
+    fn(usize, &[ScannerPauseBacklogRetirementReplica]) -> std::result::Result<Option<ScannerPauseBacklogRetirementPlan>, String>;
+
+static RETIREMENT_PLANNER: OnceLock<ScannerPauseBacklogRetirementPlanner> = OnceLock::new();
+
+/// Install the stateless native record planner before storage starts workers.
+/// The scanner runtime switch does not control this storage safety check.
+pub fn register_scanner_pause_backlog_retirement_planner(planner: ScannerPauseBacklogRetirementPlanner) {
+    RETIREMENT_PLANNER.get_or_init(|| planner);
+}
+
+pub(crate) fn is_scanner_pause_backlog(bucket: &str, object: &str) -> bool {
+    bucket == RUSTFS_META_BUCKET && object == SCANNER_PAUSE_BACKLOG_PATH
+}
+
+pub(crate) struct ScannerPauseBacklogRetirementRead {
+    pub replica: ScannerPauseBacklogRetirementReplica,
+    pub etag: Option<String>,
+}
+
+impl ScannerPauseBacklogRetirementRead {
+    pub(crate) fn preconditions(&self) -> HTTPPreconditions {
+        match &self.etag {
+            Some(etag) => HTTPPreconditions {
+                if_match: Some(etag.clone()),
+                ..Default::default()
+            },
+            None => HTTPPreconditions {
+                if_none_match: Some("*".to_string()),
+                ..Default::default()
+            },
+        }
+    }
+}
+
+async fn read_replica(set: Arc<SetDisks>) -> Result<ScannerPauseBacklogRetirementRead> {
+    let mut replica = ScannerPauseBacklogRetirementReplica {
+        pool_index: set.pool_index,
+        set_index: set.set_index,
+        data: None,
+    };
+    let reader = match set
+        .get_object_reader(
+            RUSTFS_META_BUCKET,
+            SCANNER_PAUSE_BACKLOG_PATH,
+            None,
+            HeaderMap::new(),
+            &ObjectOptions {
+                no_lock: true,
+                ..Default::default()
+            },
+        )
+        .await
+    {
+        Ok(reader) => reader,
+        Err(err) if is_err_object_not_found(&err) || is_err_version_not_found(&err) => {
+            return Ok(ScannerPauseBacklogRetirementRead { replica, etag: None });
+        }
+        Err(err) => return Err(err),
+    };
+    let info = &reader.object_info;
+    if info.version_id.is_some_and(|version| !version.is_nil())
+        || info.delete_marker
+        || info.is_dir
+        || info.etag.as_ref().is_none_or(String::is_empty)
+        || info.size < 0
+        || info.size > MAX_SCANNER_PAUSE_BACKLOG_BYTES as i64
+    {
+        return Err(Error::other("scanner pause backlog retirement found an unsupported replica identity"));
+    }
+    let etag = info.etag.clone();
+    let expected_size = info.size as usize;
+    let mut data = Vec::new();
+    reader
+        .take(MAX_SCANNER_PAUSE_BACKLOG_BYTES + 1)
+        .read_to_end(&mut data)
+        .await?;
+    if data.len() != expected_size || data.len() > MAX_SCANNER_PAUSE_BACKLOG_BYTES as usize {
+        return Err(Error::other("scanner pause backlog retirement replica has an invalid payload length"));
+    }
+    replica.data = Some(data);
+    Ok(ScannerPauseBacklogRetirementRead { replica, etag })
+}
+
+/// The caller retains the fixed object write lock and durable topology read
+/// fence through both this snapshot and physical source cleanup.
+pub(crate) async fn read_scanner_pause_backlog_retirement_replicas(
+    source_pool_index: usize,
+    source_set_index: usize,
+    sets: Vec<Arc<SetDisks>>,
+) -> Result<Vec<ScannerPauseBacklogRetirementRead>> {
+    let replicas = join_all(sets.into_iter().map(read_replica))
+        .await
+        .into_iter()
+        .collect::<Result<Vec<_>>>()?;
+    if !replicas.iter().any(|read| {
+        read.replica.pool_index == source_pool_index && read.replica.set_index == source_set_index && read.replica.data.is_some()
+    }) {
+        return Err(Error::other("scanner pause backlog retirement current source replica is missing"));
+    }
+    Ok(replicas)
+}
+
+pub(crate) fn plan_scanner_pause_backlog_retirement(
+    source_pool_index: usize,
+    replicas: &[ScannerPauseBacklogRetirementRead],
+) -> Result<Option<ScannerPauseBacklogRetirementPlan>> {
+    let planner = RETIREMENT_PLANNER
+        .get()
+        .ok_or_else(|| Error::other("scanner pause backlog native retirement planner is unavailable"))?;
+    let snapshots = replicas
+        .iter()
+        .map(|read| ScannerPauseBacklogRetirementReplica {
+            pool_index: read.replica.pool_index,
+            set_index: read.replica.set_index,
+            data: read.replica.data.clone(),
+        })
+        .collect::<Vec<_>>();
+    planner(source_pool_index, &snapshots).map_err(Error::other)
+}
+
+/// The native writer and retirement handoff use the same conditional, full-tail
+/// write. Their callers retain object and durable membership fences until return.
+pub(crate) async fn persist_native_scanner_pause_backlog_replica(
+    set: Arc<SetDisks>,
+    data: Vec<u8>,
+    preconditions: HTTPPreconditions,
+    mut opts: ObjectOptions,
+    _phase: &'static str,
+) -> Result<ObjectInfo> {
+    if data.len() > MAX_SCANNER_PAUSE_BACKLOG_BYTES as usize {
+        return Err(Error::other("scanner pause backlog exceeds its size bound"));
+    }
+    opts.max_parity = true;
+    opts.write_completion = WriteCompletion::TailDrained;
+    opts.http_preconditions = Some(preconditions);
+    #[cfg(feature = "test-util")]
+    let fault = test_util::matching_write(&set, _phase)?;
+    let result = set
+        .put_object(RUSTFS_META_BUCKET, SCANNER_PAUSE_BACKLOG_PATH, &mut PutObjReader::from_vec(data), &opts)
+        .await;
+    #[cfg(feature = "test-util")]
+    if result.is_ok()
+        && let Some(fault) = fault
+    {
+        fault.arrived.notify_one();
+        fault.release.notified().await;
+    }
+    result
+}
+
+#[cfg(feature = "test-util")]
+pub mod test_util {
+    use super::*;
+    use std::sync::Mutex;
+    use std::sync::atomic::{AtomicUsize, Ordering};
+    use tokio::sync::Notify;
+
+    #[derive(Debug, thiserror::Error)]
+    #[error("injected native scanner backlog {phase} write failure")]
+    struct InjectedWriteFailure {
+        phase: &'static str,
+    }
+
+    pub(super) struct WriteFault {
+        set: Arc<SetDisks>,
+        phase: &'static str,
+        remaining: AtomicUsize,
+        fail_before_write: bool,
+        pub(super) arrived: Notify,
+        pub(super) release: Notify,
+    }
+
+    static WRITE_FAULTS: Mutex<Vec<Arc<WriteFault>>> = Mutex::new(Vec::new());
+
+    /// Scope a one-shot fault to the actual set instance, so other stores and
+    /// concurrent tests keep using the ordinary native persistence path.
+    pub struct NativeScannerPauseBacklogWriteFault {
+        state: Arc<WriteFault>,
+    }
+
+    impl NativeScannerPauseBacklogWriteFault {
+        fn install(set: Arc<SetDisks>, phase: &'static str, nth: usize, fail_before_write: bool) -> Self {
+            assert!(nth > 0);
+            let state = Arc::new(WriteFault {
+                set,
+                phase,
+                remaining: AtomicUsize::new(nth),
+                fail_before_write,
+                arrived: Notify::new(),
+                release: Notify::new(),
+            });
+            let mut faults = WRITE_FAULTS.lock().unwrap();
+            assert!(
+                !faults
+                    .iter()
+                    .any(|fault| Arc::ptr_eq(&fault.set, &state.set) && fault.phase == phase)
+            );
+            faults.push(Arc::clone(&state));
+            Self { state }
+        }
+
+        pub fn fail_before_write(set: Arc<SetDisks>, phase: &'static str, nth: usize) -> Self {
+            Self::install(set, phase, nth, true)
+        }
+
+        pub fn pause_after_write(set: Arc<SetDisks>, phase: &'static str) -> Self {
+            Self::install(set, phase, 1, false)
+        }
+
+        pub async fn wait_until_paused(&self) {
+            self.state.arrived.notified().await;
+        }
+
+        pub fn release(&self) {
+            self.state.release.notify_one();
+        }
+    }
+
+    impl Drop for NativeScannerPauseBacklogWriteFault {
+        fn drop(&mut self) {
+            self.release();
+            WRITE_FAULTS.lock().unwrap().retain(|fault| !Arc::ptr_eq(fault, &self.state));
+        }
+    }
+
+    pub(super) fn matching_write(set: &Arc<SetDisks>, phase: &'static str) -> Result<Option<Arc<WriteFault>>> {
+        let fault = WRITE_FAULTS
+            .lock()
+            .unwrap()
+            .iter()
+            .find(|fault| Arc::ptr_eq(&fault.set, set) && fault.phase == phase)
+            .cloned();
+        let Some(fault) = fault else { return Ok(None) };
+        if fault
+            .remaining
+            .fetch_update(Ordering::AcqRel, Ordering::Acquire, |remaining| remaining.checked_sub(1))
+            != Ok(1)
+        {
+            return Ok(None);
+        }
+        if fault.fail_before_write {
+            return Err(Error::other(InjectedWriteFailure { phase }));
+        }
+        Ok(Some(fault))
+    }
+}

crates/ecstore/src/services/rebalance/control.rs

@@ -572,7 +572,7 @@ impl ECStore {
     where
         S: EcstoreObjectIO + StorageNamespaceLocking<Error = Error, NamespaceLock = rustfs_lock::NamespaceLockWrapper>,
     {
-        // Lock order: pool_meta_save_gate -> rebalance.bin -> pool.bin.
+        // Lock order: pool_meta_save_gate -> pool.bin -> rebalance.bin.
         let mut pool_meta_guard = self.pool_meta_save_gate.lock().await;
         pool_meta_guard.ensure_write_safe("rebalance worker activation")?;
         // Classify the durable rebalance record while holding both namespace

crates/ecstore/src/services/rebalance/entry.rs

@@ -50,11 +50,6 @@ fn ensure_rebalance_entry_active(cancel: &CancellationToken) -> Result<()> {
     Ok(())
 }
 
-#[cfg(test)]
-tokio::task_local! {
-    static REBALANCE_ENTRY_RUN_FENCE_BARRIER: (Arc<tokio::sync::Notify>, Arc<tokio::sync::Notify>);
-}
-
 #[derive(Debug)]
 struct RebalanceEntryTarget {
     bucket: String,
@@ -261,15 +256,9 @@ impl ECStore {
             .sort_by_key(|v| (v.mod_time.is_none(), std::cmp::Reverse(v.mod_time)));
 
         // Entry lock order is bucket incarnation -> activation_gate -> rebalance.bin -> movement gate.
-        // Target capacity admission can then acquire pool.bin under the run fence.
         // Stop waits for in-flight entries through cleanup, but not for entries admitted later.
         ensure_rebalance_entry_active(&cancel)?;
         let run_guard = self.rebalance_run_guard(rebalance_id.as_ref(), "rebalance entry").await?;
-        #[cfg(test)]
-        if let Ok((arrived, release)) = REBALANCE_ENTRY_RUN_FENCE_BARRIER.try_with(Clone::clone) {
-            arrived.notify_one();
-            release.notified().await;
-        }
         let lock_lost_signal = run_guard.lock_lost_signal();
         #[cfg(test)]
         let _run_signal_test_fence = lock_lost_signal
@@ -1248,130 +1237,6 @@ mod tests {
         assert_eq!(pool_stats.cleanup_warnings.count, 1, "deferred cleanup must not add a permanent warning");
     }
 
-    #[tokio::test]
-    #[serial_test::serial]
-    async fn real_rebalance_entry_progresses_while_peer_activation_waits_for_run_fence() {
-        const REBALANCE_ID: &str = "rebalance-peer-activation-lock-order";
-        let (_temp_dirs, store, peer) = crate::services::rebalance::test_two_pool_stores_with_isolated_node_contexts(Some(
-            active_rebalance_meta(REBALANCE_ID),
-        ))
-        .await;
-        assert!(!Arc::ptr_eq(&store.ctx, &peer.ctx), "node-local movement gates must be independent");
-        {
-            let mut meta = peer.rebalance_meta.write().await;
-            let meta = meta.as_mut().expect("peer should know the durable run");
-            meta.activation_gate = Arc::default();
-            meta.cancel = None;
-        }
-        let bucket = crate::disk::RUSTFS_META_BUCKET;
-        let object = "rebalance-peer-activation-object";
-        let version_id = uuid::Uuid::new_v4();
-        let payload = b"entry must drain before peer activation takes the pool fence".repeat(1024);
-        let source_set = store.pools[0].get_disks_by_key(object);
-        let target_set = store.pools[1].get_disks_by_key(object);
-        let opts = ObjectOptions {
-            versioned: true,
-            version_id: Some(version_id.to_string()),
-            ..Default::default()
-        };
-        let mut writer = PutObjReader::from_vec(payload.clone());
-        let source_before = source_set
-            .put_object(bucket, object, &mut writer, &opts)
-            .await
-            .expect("source version should be written");
-        let entry = metacache_entry_from_source(&source_set, bucket, object).await;
-        let arrived = Arc::new(tokio::sync::Notify::new());
-        let release = Arc::new(tokio::sync::Notify::new());
-        // JoinSet aborts both scoped tasks if an assertion or timeout fails.
-        let mut tasks = tokio::task::JoinSet::new();
-        let entry_store = Arc::clone(&store);
-        tasks.spawn(
-            REBALANCE_ENTRY_RUN_FENCE_BARRIER.scope((Arc::clone(&arrived), Arc::clone(&release)), async move {
-                entry_store
-                    .rebalance_entry(
-                        RebalanceEntryTarget {
-                            bucket: bucket.to_string(),
-                            pool_index: 0,
-                        },
-                        entry,
-                        source_set,
-                        Arc::new(RebalanceBucketConfigs::default()),
-                        Arc::from(REBALANCE_ID),
-                        CancellationToken::new(),
-                    )
-                    .await
-            }),
-        );
-        tokio::time::timeout(StdDuration::from_secs(30), arrived.notified())
-            .await
-            .expect("real entry must acquire its persisted run read fence");
-
-        let attempted = Arc::new(tokio::sync::Notify::new());
-        let peer_pool = Arc::clone(&peer.pools[0]);
-        let (activation_done, activation_result) = tokio::sync::oneshot::channel();
-        tasks.spawn(
-            crate::core::pools::REBALANCE_ACTIVATION_LOCK_ATTEMPT.scope(Arc::clone(&attempted), async move {
-                let result = peer.fence_rebalance_worker_activation(peer_pool, REBALANCE_ID).await;
-                let result = result.map(|fence| match fence {
-                    super::super::control::RebalanceWorkerActivationFence::Ready(fence) => {
-                        fence.ensure_held().expect("peer activation must retain both fences");
-                    }
-                    super::super::control::RebalanceWorkerActivationFence::NotStartedTerminal => {
-                        panic!("the paused entry's run must still require activation");
-                    }
-                });
-                activation_done.send(result).expect("activation receiver should remain alive");
-                Ok(RebalanceEntryOutcome::Completed)
-            }),
-        );
-        tokio::time::timeout(StdDuration::from_secs(30), attempted.notified())
-            .await
-            .expect("peer activation must attempt the persisted rebalance write fence");
-        release.notify_one();
-
-        tokio::time::timeout(StdDuration::from_secs(30), async {
-            while let Some(result) = tasks.join_next().await {
-                assert!(matches!(
-                    result
-                        .expect("scoped task must not panic")
-                        .expect("entry must not fail or defer"),
-                    RebalanceEntryOutcome::Completed
-                ));
-            }
-        })
-        .await
-        .expect("entry and peer activation must both make progress");
-        activation_result
-            .await
-            .expect("peer activation result should be sent")
-            .expect("peer activation must not time out behind the entry it blocks");
-
-        let mut reader = target_set
-            .get_object_reader(bucket, object, None, HeaderMap::new(), &opts)
-            .await
-            .expect("the exact target version must be readable");
-        let mut actual = Vec::new();
-        reader
-            .stream
-            .read_to_end(&mut actual)
-            .await
-            .expect("target body should drain completely");
-        assert_eq!(actual, payload);
-        assert_eq!(reader.object_info.version_id, source_before.version_id);
-        assert_eq!(reader.object_info.etag, source_before.etag);
-        assert_eq!(reader.object_info.mod_time, source_before.mod_time);
-        let source_error = store.pools[0]
-            .get_object_info(bucket, object, &opts)
-            .await
-            .expect_err("completed entry must clean up the source version");
-        assert!(crate::error::is_err_object_not_found(&source_error) || crate::error::is_err_version_not_found(&source_error));
-        let meta = store.rebalance_meta.read().await;
-        let stats = &meta.as_ref().expect("local run must remain installed").pool_stats[0];
-        assert_eq!(stats.num_objects, 1);
-        assert_eq!(stats.num_versions, 1);
-        assert_eq!(stats.cleanup_warnings.count, 0);
-    }
-
     #[tokio::test]
     #[serial_test::serial]
     async fn real_rebalance_run_fence_loss_before_target_commit_preserves_target_and_source() {

crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs

@@ -1907,124 +1907,6 @@ fn test_is_transient_rebalance_error_accepts_wrapped_disk_timeout() {
     assert!(is_transient_rebalance_error(&Error::Io(std::io::Error::other(DiskError::Timeout))));
 }
 
-#[test]
-fn test_rebalance_stage_wrapped_transient_errors_remain_retryable() {
-    let cases = [
-        Error::Lock(rustfs_lock::LockError::timeout(".rustfs.sys/pool.bin@latest", Duration::from_secs(5))),
-        Error::Lock(rustfs_lock::LockError::network(
-            "peer unavailable",
-            std::io::Error::from(std::io::ErrorKind::ConnectionReset),
-        )),
-        Error::SlowDown,
-        Error::ErasureReadQuorum,
-        Error::ErasureWriteQuorum,
-        Error::Io(std::io::Error::other(DiskError::Timeout)),
-        Error::Io(std::io::Error::from(std::io::ErrorKind::TimedOut)),
-    ];
-    for mut error in cases {
-        for depth in 0..=3 {
-            assert!(is_transient_rebalance_error(&error), "transient source lost at depth {depth}: {error:?}");
-            assert!(
-                should_defer_rebalance_entry_failure(&error),
-                "exhausted transient entries must be deferred"
-            );
-            assert!(should_retry_rebalance_listing(&error, 0, 3));
-            assert!(
-                !should_retry_rebalance_listing(&error, 2, 3),
-                "wrapping must not bypass the attempt limit"
-            );
-            error = data_movement::data_movement_stage_error_for_test(
-                "rebalance_object",
-                "put_object",
-                "bucket",
-                "baseline/00042.bin",
-                error,
-            );
-        }
-    }
-}
-
-#[test]
-fn test_rebalance_stage_wrapped_terminal_errors_remain_terminal() {
-    let cases = [
-        Error::FileAccessDenied,
-        Error::FileCorrupt,
-        Error::OperationCanceled,
-        Error::DataMovementOverwriteErr("bucket".to_string(), "object".to_string(), "version".to_string()),
-        Error::Lock(rustfs_lock::LockError::already_locked("bucket/object", "owner")),
-        Error::other("permission denied"),
-    ];
-    for mut error in cases {
-        for depth in 0..=3 {
-            assert!(
-                !is_transient_rebalance_error(&error),
-                "terminal source must survive depth {depth}: {error:?}"
-            );
-            assert!(!should_defer_rebalance_entry_failure(&error));
-            // Object names are untrusted context, not evidence of a transient failure.
-            error = data_movement::data_movement_stage_error_for_test(
-                "rebalance_object",
-                "put_object",
-                "bucket",
-                "remote lock rpc timed out",
-                error,
-            );
-        }
-    }
-}
-
-#[tokio::test]
-async fn test_rebalance_stage_wrapped_lock_timeout_retries_real_migration_loop() {
-    for succeeds_on_retry in [true, false] {
-        let backend = MigrationBackendSpy::new(None, None);
-        let attempts = AtomicUsize::new(0);
-        let waits = AtomicUsize::new(0);
-        let mut transfer = |_, _, _| {
-            let attempt = attempts.fetch_add(1, Ordering::SeqCst);
-            async move {
-                if succeeds_on_retry && attempt > 0 {
-                    return Ok(());
-                }
-                Err(data_movement::data_movement_stage_error_for_test(
-                    "rebalance_object",
-                    "put_object",
-                    "bucket",
-                    "baseline/00042.bin",
-                    Error::Lock(rustfs_lock::LockError::timeout(".rustfs.sys/pool.bin@latest", Duration::from_secs(5))),
-                ))
-            }
-        };
-        let version = version_normal();
-        let result = migrate_entry_version_with_retry_wait(
-            &backend,
-            "bucket".to_string(),
-            0,
-            &version,
-            None,
-            3,
-            false,
-            &mut transfer,
-            |_: String, _: String, _: ObjectOptions| async { Ok::<_, Error>(ObjectInfo::default()) },
-            |_| {
-                waits.fetch_add(1, Ordering::SeqCst);
-                std::future::ready(())
-            },
-        )
-        .await;
-        assert_eq!(result.moved, succeeds_on_retry);
-        assert_eq!(result.failed, !succeeds_on_retry);
-        assert_eq!(attempts.load(Ordering::SeqCst), if succeeds_on_retry { 2 } else { 3 });
-        assert_eq!(backend.get_calls(), attempts.load(Ordering::SeqCst));
-        assert_eq!(waits.load(Ordering::SeqCst), attempts.load(Ordering::SeqCst) - 1);
-        if !succeeds_on_retry {
-            assert_eq!(result.stage, Some("write_target"));
-            assert!(should_defer_rebalance_entry_failure(
-                result.error.as_ref().expect("exhaustion must retain its source error")
-            ));
-        }
-    }
-}
-
 #[test]
 fn test_is_transient_rebalance_error_accepts_io_timeout_message() {
     assert!(is_transient_rebalance_error(&Error::Io(std::io::Error::other("timeout"))));

crates/ecstore/src/services/rebalance/worker.rs

@@ -244,7 +244,6 @@ pub(super) fn resolve_rebalance_bucket_result(
 }
 
 pub(super) fn is_transient_rebalance_error(err: &Error) -> bool {
-    let err = rebalance_error_source(err);
     match err {
         Error::SlowDown
         | Error::ErasureReadQuorum
@@ -257,15 +256,6 @@ pub(super) fn is_transient_rebalance_error(err: &Error) -> bool {
     }
 }
 
-fn rebalance_error_source(mut err: &Error) -> &Error {
-    // Stage context contains object names, so classify the preserved source,
-    // not timeout-like text supplied by an object name. Iterate nested stages.
-    while let Some(source) = crate::data_movement::data_movement_stage_source(err) {
-        err = source;
-    }
-    err
-}
-
 fn is_rebalance_transient_lock_error(err: &rustfs_lock::LockError) -> bool {
     match err {
         rustfs_lock::LockError::Timeout { .. } | rustfs_lock::LockError::Network { .. } => true,
@@ -319,7 +309,6 @@ pub(super) fn rebalance_listing_retry_delay(attempt: usize) -> Duration {
 }
 
 fn is_rebalance_lock_or_rpc_timeout(err: &Error) -> bool {
-    let err = rebalance_error_source(err);
     match err {
         Error::Lock(rustfs_lock::LockError::Timeout { .. }) | Error::Lock(rustfs_lock::LockError::Network { .. }) => true,
         Error::Io(io_err) => is_rebalance_lock_or_rpc_timeout_message(&io_err.to_string()),
@@ -596,48 +585,3 @@ impl SetDisks {
         Ok(())
     }
 }
-
-#[cfg(test)]
-mod error_source_tests {
-    use super::*;
-
-    #[test]
-    fn stage_wrapped_errors_select_the_source_backoff_policy() {
-        let cases = [
-            (
-                Error::Lock(rustfs_lock::LockError::timeout(".rustfs.sys/pool.bin@latest", Duration::from_secs(5))),
-                true,
-            ),
-            (
-                Error::Lock(rustfs_lock::LockError::network(
-                    "peer unavailable",
-                    std::io::Error::from(std::io::ErrorKind::ConnectionReset),
-                )),
-                true,
-            ),
-            (Error::other("remote lock rpc timed out"), true),
-            (Error::SlowDown, false),
-            (Error::Io(std::io::Error::other(DiskError::Timeout)), false),
-            (Error::FileAccessDenied, false),
-        ];
-        for (mut error, lock_backoff) in cases {
-            for depth in 0..=3 {
-                assert_eq!(
-                    is_rebalance_lock_or_rpc_timeout(&error),
-                    lock_backoff,
-                    "wrong backoff at depth {depth}: {error:?}"
-                );
-                if !lock_backoff {
-                    assert_eq!(rebalance_migration_retry_delay(1, &error), REBALANCE_MIGRATION_RETRY_BASE_DELAY * 2);
-                }
-                error = crate::data_movement::data_movement_stage_error_for_test(
-                    "rebalance_object",
-                    "put_object",
-                    "bucket",
-                    "remote lock rpc timed out",
-                    error,
-                );
-            }
-        }
-    }
-}

crates/ecstore/src/store/object.rs

@@ -3079,12 +3079,7 @@ impl ECStore {
         let store = Arc::clone(self);
         let write = async move {
             let object = "buckets/.scanner-pause-backlog.json";
-            let mut opts = ObjectOptions {
-                max_parity: true,
-                http_preconditions: Some(preconditions),
-                write_completion: crate::object_api::WriteCompletion::TailDrained,
-                ..Default::default()
-            };
+            let mut opts = ObjectOptions::default();
             // Match migration: fixed object namespace -> durable pool metadata ->
             // actual replica namespace. The replica need not be the hash-routed set.
             let object_guard = if store.single_pool() {
@@ -3110,9 +3105,14 @@ impl ECStore {
             } else {
                 None
             };
-            let result = set
-                .put_object(RUSTFS_META_BUCKET, object, &mut PutObjReader::from_vec(data), &opts)
-                .await;
+            let result = crate::data_movement::scanner_backlog::persist_native_scanner_pause_backlog_replica(
+                set,
+                data,
+                preconditions,
+                opts,
+                "publish",
+            )
+            .await;
             drop(capacity_guard);
             drop(object_guard);
             result
@@ -3747,29 +3747,37 @@ impl ECStore {
         opts: &ObjectOptions,
         no_lock: bool,
     ) -> Result<usize> {
+        let capacity_owner = DecommissionCapacityOwner::from_options(opts);
         match self
             .get_pool_info_existing_with_opts(bucket, object, &data_movement_pool_lookup_opts(opts, no_lock))
             .await
         {
-            Ok((pinfo, _)) => Ok(pinfo.index),
+            Ok((pinfo, _)) => {
+                if let Some(owner) = capacity_owner {
+                    if self.is_decommission_capacity_target_reserved(owner, pinfo.index).await? {
+                        return Ok(pinfo.index);
+                    }
+                } else {
+                    return Ok(pinfo.index);
+                }
+            }
             Err(err) => {
                 if !is_err_object_not_found(&err) && !is_err_version_not_found(&err) {
                     return Err(err);
                 }
-
-                if let Some(owner) = DecommissionCapacityOwner::from_options(opts) {
-                    let expected_data_bytes = opts
-                        .capacity_expected_data_bytes()
-                        .or_else(|| usize::try_from(size).ok())
-                        .unwrap_or_default();
-                    return self
-                        .select_decommission_capacity_target_pool(owner, expected_data_bytes)
-                        .await;
-                }
-
-                self.get_available_pool_idx(bucket, object, size).await.ok_or(Error::DiskFull)
             }
         }
+        if let Some(owner) = capacity_owner {
+            let expected_data_bytes = opts
+                .capacity_expected_data_bytes()
+                .or_else(|| usize::try_from(size).ok())
+                .unwrap_or_default();
+            return self
+                .select_decommission_capacity_target_pool(owner, expected_data_bytes)
+                .await;
+        }
+
+        self.get_available_pool_idx(bucket, object, size).await.ok_or(Error::DiskFull)
     }
 
     async fn find_data_movement_target_info(

crates/heal/Cargo.toml

@@ -104,7 +104,6 @@ walkdir = { workspace = true }
 http = { workspace = true }
 temp-env = { workspace = true, features = ["async_closure"] }
 tokio = { workspace = true, features = ["test-util", "fs", "rt-multi-thread"] }
-chrono = { workspace = true }
 
 [lib]
 doctest = false

crates/heal/src/heal/mrf_queue.rs

@@ -1139,26 +1139,9 @@ mod tests {
     use super::*;
     use crate::heal::manager::HealConfig;
     use crate::heal::storage::{ECStoreHealStorage, HealStorageAPI};
-    use crate::heal::{DiskError, RUSTFS_META_BUCKET};
     use rustfs_common::mrf_channel::{MrfIntent, MrfKind, MrfVerifiedRepairDisposition, MrfVerifiedRepairEvent};
-    use serde_json::{Map, Value, json};
     use serial_test::serial;
-    use std::env;
-    use std::fs;
-    use std::io::Write as _;
-    use std::path::{Path, PathBuf};
     use std::sync::Arc as StdArc;
-    use std::time::{Duration as StdDuration, Instant};
-
-    const W13_EVIDENCE_DIR_ENV: &str = "RUSTFS_SCANNER_HEAL_W13_EVIDENCE_DIR";
-    const W13_SOURCE_REVISION_ENV: &str = "RUSTFS_SCANNER_HEAL_W13_SOURCE_REVISION";
-    const W13_SELECTION_ENV: &str = "RUSTFS_SCANNER_HEAL_W13_SELECTION";
-    const W13_SOAK_SECONDS_ENV: &str = "RUSTFS_SCANNER_HEAL_W13_SOAK_SECONDS";
-    const W13_ALLOW_SHORT_SOAK_ENV: &str = "RUSTFS_SCANNER_HEAL_W13_ALLOW_SHORT_SOAK";
-    const W13_RUN_ID_ENV: &str = "RUSTFS_SCANNER_HEAL_W13_RUN_ID";
-    const W13_WINDOW_ID_ENV: &str = "RUSTFS_SCANNER_HEAL_W13_WINDOW_ID";
-    const W13_ENOSPC_ROOT_ENV: &str = "RUSTFS_SCANNER_HEAL_W13_ENOSPC_ROOT";
-    const W13_ENOSPC_FILL_LIMIT_ENV: &str = "RUSTFS_SCANNER_HEAL_W13_ENOSPC_FILL_LIMIT_BYTES";
 
     fn intent(bucket: &str, object: &str, attempts: u8) -> MrfIntent {
         MrfIntent {
@@ -1179,748 +1162,6 @@ mod tests {
         payload
     }
 
-    fn w13_timestamp() -> String {
-        chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true)
-    }
-
-    fn w13_selection_contains(selection: &str, lane: &str) -> bool {
-        selection == "all" || selection.split(',').any(|item| item.trim() == lane)
-    }
-
-    fn w13_evidence_path(root: &Path, gate: &str, field: &str) -> PathBuf {
-        let lane = match gate {
-            "G07" => "g07-mrf-responsibility",
-            "G08" => "g08-mrf-capacity",
-            "P4" => "p4-mrf-soak",
-            other => panic!("unsupported W13 evidence gate: {other}"),
-        };
-        root.join(lane).join(format!("{gate}-{field}.json"))
-    }
-
-    struct W13Evidence<'a> {
-        source_revision: &'a str,
-        run_id: &'a str,
-        window_id: &'a str,
-        started_at: &'a str,
-        finished_at: &'a str,
-        gate: &'a str,
-        field: &'a str,
-        artifact_kind: &'a str,
-        extra: Map<String, Value>,
-    }
-
-    fn write_w13_evidence(root: &Path, evidence: W13Evidence<'_>) {
-        let path = w13_evidence_path(root, evidence.gate, evidence.field);
-        fs::create_dir_all(path.parent().expect("W13 evidence artifact parent")).expect("create W13 evidence artifact directory");
-        let mut payload = Map::new();
-        payload.insert("schema".to_string(), json!(1));
-        payload.insert("evidence_type".to_string(), json!("measured"));
-        payload.insert("artifact_kind".to_string(), json!(evidence.artifact_kind));
-        payload.insert("source_revision".to_string(), json!(evidence.source_revision));
-        payload.insert("run_id".to_string(), json!(evidence.run_id));
-        payload.insert("measurement_window_id".to_string(), json!(evidence.window_id));
-        payload.insert("started_at".to_string(), json!(evidence.started_at));
-        payload.insert("finished_at".to_string(), json!(evidence.finished_at));
-        payload.insert("gate".to_string(), json!(evidence.gate));
-        payload.insert("field".to_string(), json!(evidence.field));
-        payload.insert(
-            "command".to_string(),
-            json!([
-                "cargo",
-                "test",
-                "--locked",
-                "-p",
-                "rustfs-heal",
-                "--lib",
-                "heal::mrf_queue::tests::w13_mrf_release_evidence_outputs_bundle_artifacts",
-                "--",
-                "--ignored",
-                "--exact",
-                "--nocapture"
-            ]),
-        );
-        payload.insert(
-            "summary".to_string(),
-            json!(format!("Measured W13 MRF evidence for {}.{}", evidence.gate, evidence.field)),
-        );
-        payload.extend(evidence.extra);
-        let bytes = serde_json::to_vec_pretty(&Value::Object(payload)).expect("serialize W13 evidence payload");
-        fs::write(&path, [bytes.as_slice(), b"\n"].concat()).expect("write W13 evidence artifact");
-    }
-
-    async fn w13_committed_replay_probe() -> (usize, bool, bool, bool, bool, usize) {
-        let env = rustfs_test_utils::TestECStoreEnv::builder()
-            .prefix("rustfs_mrf_w13_replay_evidence")
-            .build()
-            .await;
-        let bucket = "w13-replay-bucket";
-        let object = "w13-replay-object";
-        env.make_bucket(bucket, false).await;
-        let storage: Arc<dyn HealStorageAPI> = Arc::new(ECStoreHealStorage::new(env.ecstore.clone()));
-        let manager = Arc::new(HealManager::new(
-            storage.clone(),
-            Some(HealConfig {
-                queue_size: 2,
-                heal_interval: Duration::from_secs(3600),
-                enable_auto_heal: false,
-                ..Default::default()
-            }),
-        ));
-        let disks = journal_disks().await;
-        assert!(!disks.is_empty(), "W13 evidence requires real local MRF disks");
-
-        let config = MrfConsumerConfig::default();
-        let replay_owner = Uuid::new_v4();
-        let mut replay_intent = intent(bucket, object, 0);
-        replay_intent.kind = MrfKind::PartialWrite;
-        replay_intent.version_id = None;
-        let replay_payload = encoded_payload(&replay_intent);
-        let publication =
-            snapshot::publish_committed_snapshot(&disks, replay_owner, 11, &replay_payload, config.journal_max_bytes)
-                .await
-                .expect("publish W13 committed replay checkpoint");
-        assert_eq!(publication.manifest_replicas, disks.len(), "all W13 checkpoint manifests should commit");
-
-        let mut queue = MrfQueue::new(config.queue_capacity, config.journal_max_bytes);
-        let mut backoff_until = None;
-        let replay = replay_into(&manager, &mut queue, &mut backoff_until).await;
-        assert_eq!(replay.replayed, 1, "W13 committed checkpoint must replay one record");
-        assert_eq!(queue.depth(), 0, "W13 replayed record should reach the manager before cleanup");
-        assert_eq!(replay.durable_replay_anchors.len(), 1, "W13 replay must create a proof anchor");
-        assert_eq!(
-            manager.operations_snapshot().await.queued_by_source.mrf,
-            1,
-            "W13 replayed work must be visible as MRF manager work"
-        );
-
-        let anchor = replay.durable_replay_anchors[0].clone();
-        let mut runtime = MrfRuntime {
-            queue,
-            config,
-            checkpoint_owner: Uuid::new_v4(),
-            next_checkpoint_sequence: replay.next_checkpoint_sequence,
-            new_since_flush: 0,
-            dirty: false,
-            journal_on_disk: replay.journal_on_disk,
-            retain_replay_journal: replay.retain_journal_for_replay,
-            durable_replay_anchors: replay.durable_replay_anchors,
-            replay_cleanup: replay.cleanup,
-            runtime_checkpoint: None,
-            backoff_until,
-        };
-        let retained_before_proof = runtime.retained_replay_journal();
-        assert!(retained_before_proof, "W13 proof anchor must retain replay checkpoint before proof");
-        assert!(
-            snapshot::inspect_local_committed_snapshot(runtime.config.journal_max_bytes)
-                .await
-                .expect("inspect W13 retained checkpoint")
-                .is_some(),
-            "W13 replay checkpoint must remain durable before proof"
-        );
-
-        rustfs_common::mrf_channel::note_mrf_verified_repair(MrfVerifiedRepairEvent {
-            kind: anchor.kind,
-            bucket: anchor.bucket.clone(),
-            object: anchor.object.clone(),
-            version_id: anchor.version_id,
-            scope: anchor.scope,
-            lease: Some(anchor.lease),
-            bucket_incarnation_id: anchor.bucket_incarnation_id,
-            disposition: MrfVerifiedRepairDisposition::Repaired,
-        });
-        runtime.discharge_durable_replay_anchors();
-        let proof_discharged_anchor = !runtime.retained_replay_journal();
-        assert!(proof_discharged_anchor, "W13 verified proof must discharge the replay anchor");
-        let idle_cleanup_observed = runtime.delete_idle_recovery_anchors().await;
-        assert!(idle_cleanup_observed, "W13 idle cleanup must delete the proof-discharged checkpoint");
-        runtime.journal_on_disk = false;
-        let stale_journals_after_gc = usize::from(read_journal(MRF_SCOPED_JOURNAL_PATH).await.is_some())
-            + usize::from(read_journal(MRF_JOURNAL_PATH).await.is_some())
-            + usize::from(
-                snapshot::inspect_local_committed_snapshot(runtime.config.journal_max_bytes)
-                    .await
-                    .expect("inspect W13 checkpoints after cleanup")
-                    .is_some(),
-            );
-
-        let restart_manager = Arc::new(HealManager::new(
-            storage,
-            Some(HealConfig {
-                queue_size: 2,
-                heal_interval: Duration::from_secs(3600),
-                enable_auto_heal: false,
-                ..Default::default()
-            }),
-        ));
-        assert_eq!(
-            replay_journal_once(&restart_manager).await,
-            0,
-            "W13 cleaned anchors must not resurrect on restart"
-        );
-        assert_eq!(
-            restart_manager.operations_snapshot().await.queued_by_source.mrf,
-            0,
-            "W13 restart must not re-admit proof-cleaned MRF work"
-        );
-        manager.stop().await.expect("stop W13 replay manager");
-        restart_manager.stop().await.expect("stop W13 restart manager");
-        (
-            replay.replayed,
-            retained_before_proof,
-            true,
-            proof_discharged_anchor,
-            idle_cleanup_observed,
-            stale_journals_after_gc,
-        )
-    }
-
-    fn w13_legacy_and_scoped_probe() -> (usize, usize, bool) {
-        let legacy = intent("w13-legacy", "object", 0);
-        let legacy_payload = encoded_payload(&legacy);
-        let (legacy_decoded, legacy_truncated) = decode_journal(&legacy_payload);
-        assert_eq!(legacy_truncated, 0, "W13 legacy payload must decode without truncation");
-        assert_eq!(legacy_decoded.len(), 1, "W13 legacy replay identity must round trip");
-        assert_eq!(legacy_decoded[0].bucket, legacy.bucket);
-        assert_eq!(legacy_decoded[0].object, legacy.object);
-        assert_eq!(legacy_decoded[0].version_id, legacy.version_id);
-        assert_eq!(legacy_decoded[0].scope, legacy.scope);
-
-        let mut scoped = intent("w13-scoped", "object", 0);
-        scoped.kind = MrfKind::PartialWrite;
-        scoped.version_id = Some(*Uuid::new_v4().as_bytes());
-        scoped.scope = Some(rustfs_common::mrf_channel::MrfScope {
-            pool_index: 7,
-            set_index: 13,
-        });
-        let mut runtime = MrfRuntime {
-            queue: MrfQueue::new(4, usize::MAX),
-            config: MrfConsumerConfig::default(),
-            checkpoint_owner: Uuid::new_v4(),
-            next_checkpoint_sequence: 1,
-            new_since_flush: 0,
-            dirty: true,
-            journal_on_disk: false,
-            retain_replay_journal: false,
-            durable_replay_anchors: Vec::new(),
-            replay_cleanup: None,
-            runtime_checkpoint: None,
-            backoff_until: None,
-        };
-        assert_eq!(runtime.queue.try_push_typed(scoped.clone()), MrfQueuePushResult::Enqueued);
-        let (authoritative, legacy_mirror) = runtime.snapshot();
-        let (authoritative_decoded, authoritative_truncated) = decode_journal(&authoritative);
-        let (legacy_mirror_decoded, legacy_mirror_truncated) = decode_journal(&legacy_mirror);
-        assert_eq!(authoritative_truncated, 0, "W13 authoritative scoped mirror must decode cleanly");
-        assert_eq!(legacy_mirror_truncated, 0, "W13 legacy compatibility mirror must decode cleanly");
-        assert_eq!(authoritative_decoded.len(), 1, "W13 authoritative mirror must retain scoped identity");
-        assert_eq!(authoritative_decoded[0].bucket, scoped.bucket);
-        assert_eq!(authoritative_decoded[0].object, scoped.object);
-        assert_eq!(authoritative_decoded[0].version_id, scoped.version_id);
-        assert_eq!(authoritative_decoded[0].scope, scoped.scope);
-        assert!(
-            legacy_mirror_decoded.is_empty() || legacy_mirror_decoded.iter().all(|intent| intent.scope.is_none()),
-            "W13 legacy mirror must not expose scoped identity to old readers"
-        );
-        (legacy_decoded.len(), authoritative_decoded.len(), legacy_mirror_decoded.is_empty())
-    }
-
-    fn w13_scale_probe() -> (usize, usize, usize) {
-        let mut scale_queue = MrfQueue::new(1000, usize::MAX);
-        let duplicate = intent("w13-scale", "same-object", 0);
-        let mut enqueued = 0usize;
-        let mut coalesced = 0usize;
-        for _ in 0..1000 {
-            match scale_queue.try_push_typed(duplicate.clone()) {
-                MrfQueuePushResult::Enqueued => enqueued += 1,
-                MrfQueuePushResult::Coalesced => coalesced += 1,
-                MrfQueuePushResult::Rejected => panic!("W13 scale duplicate probe should not reject"),
-            }
-        }
-        assert_eq!(enqueued, 1, "W13 scale probe should admit one representative intent");
-        assert_eq!(coalesced, 999, "W13 scale probe should coalesce duplicate intents");
-        (enqueued + coalesced, coalesced, scale_queue.depth())
-    }
-
-    fn w13_enospc_raw_os(err: &std::io::Error) -> bool {
-        err.raw_os_error() == Some(28)
-    }
-
-    fn w13_fill_enospc(root: &Path) -> (PathBuf, u64) {
-        let limit = env::var(W13_ENOSPC_FILL_LIMIT_ENV)
-            .ok()
-            .map(|raw| raw.parse::<u64>().expect("W13 ENOSPC fill limit must be an integer"))
-            .unwrap_or(128 * 1024 * 1024);
-        fs::create_dir_all(root).expect("create W13 ENOSPC root");
-        let filler = root.join(format!("w13-enospc-{}.fill", Uuid::new_v4()));
-        let mut file = fs::OpenOptions::new()
-            .create_new(true)
-            .write(true)
-            .open(&filler)
-            .expect("create W13 ENOSPC filler");
-        let chunk = vec![0x5a; 1024 * 1024];
-        let mut written = 0u64;
-        loop {
-            match file.write_all(&chunk) {
-                Ok(()) => {
-                    written = written.saturating_add(chunk.len() as u64);
-                    assert!(
-                        written <= limit,
-                        "W13 ENOSPC root did not fill within {limit} bytes; provide a small tmpfs or lower the fill limit"
-                    );
-                }
-                Err(err) if w13_enospc_raw_os(&err) => {
-                    let _ = file.sync_all();
-                    return (filler, written);
-                }
-                Err(err) => panic!("W13 ENOSPC filler failed with non-ENOSPC error: {err}"),
-            }
-        }
-    }
-
-    fn w13_snapshot_error_is_capacity(error: &snapshot::SnapshotError) -> bool {
-        match error {
-            snapshot::SnapshotError::Disk(source) => format!("{source:?}").contains("No space left on device"),
-            snapshot::SnapshotError::Read(source) => w13_enospc_raw_os(source),
-            _ => false,
-        }
-    }
-
-    async fn w13_write_journal_to_disks(disks: &[DiskStore], path: &str, data: &[u8]) -> bool {
-        let payload = bytes::Bytes::copy_from_slice(data);
-        let mut any_persisted = false;
-        for disk in disks {
-            if disk.write_all(RUSTFS_META_BUCKET, path, payload.clone()).await.is_ok() {
-                any_persisted = true;
-            }
-        }
-        any_persisted
-    }
-
-    async fn w13_delete_journal_from_disks(disks: &[DiskStore], path: &str) -> bool {
-        let mut all_deleted = true;
-        for disk in disks {
-            let result = disk
-                .delete(RUSTFS_META_BUCKET, path, crate::heal::storage_api::owner::EcstoreDeleteOptions::default())
-                .await;
-            if let Err(err) = result
-                && !matches!(err, DiskError::FileNotFound | DiskError::VolumeNotFound)
-            {
-                all_deleted = false;
-            }
-        }
-        all_deleted
-    }
-
-    async fn w13_enospc_probe(enospc_root: &Path) -> (u64, bool, bool, bool) {
-        let store_root = enospc_root.join(format!("store-{}", Uuid::new_v4()));
-        let _env = rustfs_test_utils::TestECStoreEnv::builder()
-            .disk_count(1)
-            .base_dir(&store_root)
-            .build()
-            .await;
-        let disks = journal_disks().await;
-        assert_eq!(disks.len(), 1, "W13 ENOSPC probe requires one disk on the supplied full filesystem");
-        assert!(
-            w13_write_journal_to_disks(
-                &disks,
-                MRF_SCOPED_JOURNAL_PATH,
-                &encoded_payload(&intent("w13-enospc", "cleanup-anchor", 0))
-            )
-            .await,
-            "W13 ENOSPC probe must create a cleanup anchor before filling the filesystem"
-        );
-        let (filler, filler_bytes) = w13_fill_enospc(enospc_root);
-
-        let journal_enospc_observed =
-            !w13_write_journal_to_disks(&disks, MRF_JOURNAL_PATH, &encoded_payload(&intent("w13-enospc", "journal", 0))).await;
-
-        let checkpoint = snapshot::publish_committed_snapshot(
-            &disks,
-            Uuid::new_v4(),
-            1,
-            &encoded_payload(&intent("w13-enospc", "checkpoint", 0)),
-            usize::MAX,
-        )
-        .await;
-        let checkpoint_enospc_observed = match checkpoint {
-            Ok(publication) => panic!("W13 ENOSPC checkpoint publish unexpectedly succeeded: {publication:?}"),
-            Err(error) => w13_snapshot_error_is_capacity(&error),
-        };
-        assert!(
-            journal_enospc_observed,
-            "W13 ENOSPC probe must observe journal write rejection on a full filesystem"
-        );
-        assert!(
-            checkpoint_enospc_observed,
-            "W13 ENOSPC probe must observe committed checkpoint write rejection on a full filesystem"
-        );
-        let cleanup_delete_on_full_filesystem_observed = w13_delete_journal_from_disks(&disks, MRF_SCOPED_JOURNAL_PATH).await;
-        let _ = fs::remove_file(filler);
-        assert!(
-            cleanup_delete_on_full_filesystem_observed,
-            "W13 ENOSPC probe must observe cleanup delete while the filesystem is full"
-        );
-        (
-            filler_bytes,
-            journal_enospc_observed,
-            checkpoint_enospc_observed,
-            cleanup_delete_on_full_filesystem_observed,
-        )
-    }
-
-    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-    #[serial]
-    #[ignore = "writes W13 release evidence artifacts; run through scripts/run_scanner_heal_w13_mrf_evidence.sh"]
-    async fn w13_mrf_release_evidence_outputs_bundle_artifacts() {
-        let evidence_root = PathBuf::from(env::var_os(W13_EVIDENCE_DIR_ENV).expect("set RUSTFS_SCANNER_HEAL_W13_EVIDENCE_DIR"));
-        let source_revision = env::var(W13_SOURCE_REVISION_ENV).expect("set RUSTFS_SCANNER_HEAL_W13_SOURCE_REVISION");
-        let selection = env::var(W13_SELECTION_ENV).unwrap_or_else(|_| "all".to_string());
-        let run_id = env::var(W13_RUN_ID_ENV).unwrap_or_else(|_| "w13-mrf-release-evidence-run".to_string());
-        let window_id = env::var(W13_WINDOW_ID_ENV).unwrap_or_else(|_| "w13-mrf-release-evidence-window".to_string());
-        let soak_seconds = env::var(W13_SOAK_SECONDS_ENV)
-            .ok()
-            .map(|raw| raw.parse::<u64>().expect("W13 soak seconds must be an integer"))
-            .unwrap_or(7200);
-        let allow_short_soak = env::var(W13_ALLOW_SHORT_SOAK_ENV).as_deref() == Ok("1");
-        if w13_selection_contains(&selection, "p4") && soak_seconds < 7200 && !allow_short_soak {
-            panic!("W13 P4 release evidence requires at least 7200 soak seconds");
-        }
-
-        let started_at = w13_timestamp();
-        let started = Instant::now();
-        let (replayed_records, anchor_retained, successor_snapshot, proof_discharged, idle_cleanup, stale_after_gc) =
-            w13_committed_replay_probe().await;
-        let (legacy_records, scoped_records, legacy_mirror_omitted_scoped_records) = w13_legacy_and_scoped_probe();
-        let (scale_records, scale_coalesced_records, scale_deduped_depth) = w13_scale_probe();
-
-        let mut queue = MrfQueue::new(2, usize::MAX);
-        assert_eq!(queue.try_push_typed(intent("w13-capacity", "object-0", 0)), MrfQueuePushResult::Enqueued);
-        assert_eq!(queue.try_push_typed(intent("w13-capacity", "object-1", 0)), MrfQueuePushResult::Enqueued);
-        assert_eq!(queue.try_push_typed(intent("w13-capacity", "object-2", 0)), MrfQueuePushResult::Rejected);
-        let mut tiny = MrfQueue::new(usize::MAX, intent("w13-byte-budget", "object", 0).estimated_bytes());
-        assert_eq!(tiny.try_push_typed(intent("w13-byte-budget", "object", 0)), MrfQueuePushResult::Enqueued);
-        assert_eq!(
-            tiny.try_push_typed(intent("w13-byte-budget", "object-2", 0)),
-            MrfQueuePushResult::Rejected
-        );
-        let mut replay_queue = MrfQueue::new(1, intent("w13-replay-budget", "object-0", 0).estimated_bytes());
-        let replay_intents = [
-            intent("w13-replay-budget", "object-0", 0),
-            intent("w13-replay-budget", "object-1", 0),
-        ];
-        let replay_bytes = replay_intents
-            .iter()
-            .fold(0usize, |total, intent| total.saturating_add(intent.estimated_bytes()));
-        replay_queue.raise_limits_for_replay(replay_intents.len(), replay_bytes);
-        for intent in replay_intents {
-            assert_eq!(replay_queue.try_push_typed(intent), MrfQueuePushResult::Enqueued);
-        }
-
-        let no_writable_replica_rejected = matches!(
-            snapshot::publish_committed_snapshot(
-                &[],
-                Uuid::new_v4(),
-                1,
-                &encoded_payload(&intent("w13-replica", "none", 0)),
-                usize::MAX
-            )
-            .await,
-            Err(snapshot::SnapshotError::NoWritableReplica)
-        );
-        assert!(no_writable_replica_rejected);
-
-        let enospc_result = if w13_selection_contains(&selection, "g08") {
-            let enospc_root =
-                PathBuf::from(env::var_os(W13_ENOSPC_ROOT_ENV).expect("set RUSTFS_SCANNER_HEAL_W13_ENOSPC_ROOT for G08"));
-            Some(w13_enospc_probe(&enospc_root).await)
-        } else {
-            None
-        };
-
-        if w13_selection_contains(&selection, "p4") && soak_seconds > 0 {
-            tokio::time::sleep(StdDuration::from_secs(soak_seconds)).await;
-        }
-        let measured_seconds = started.elapsed().as_secs().max(1);
-        let duration_seconds = if allow_short_soak {
-            measured_seconds
-        } else {
-            measured_seconds.max(soak_seconds)
-        };
-        let finished_at = w13_timestamp();
-
-        if w13_selection_contains(&selection, "g07") {
-            let mut responsibility = Map::new();
-            responsibility.insert(
-                "mrf_responsibility_cases".to_string(),
-                json!([
-                    "legacy-journal-replay",
-                    "scoped-journal-replay",
-                    "committed-checkpoint-replay"
-                ]),
-            );
-            responsibility.insert(
-                "crash_points".to_string(),
-                json!(["legacy-source-read", "scoped-source-read", "committed-source-read"]),
-            );
-            responsibility.insert("replayed_records".to_string(), json!(replayed_records));
-            responsibility.insert("responsibility_anchor_retained".to_string(), json!(anchor_retained));
-            responsibility.insert("successor_snapshot_published".to_string(), json!(successor_snapshot));
-            responsibility.insert("manager_mrf_queued".to_string(), json!(1));
-            responsibility.insert("legacy_records_decoded".to_string(), json!(legacy_records));
-            responsibility.insert("scoped_records_decoded".to_string(), json!(scoped_records));
-            responsibility.insert(
-                "legacy_mirror_omitted_scoped_records".to_string(),
-                json!(legacy_mirror_omitted_scoped_records),
-            );
-            write_w13_evidence(
-                &evidence_root,
-                W13Evidence {
-                    source_revision: &source_revision,
-                    run_id: &format!("{run_id}-g07-responsibility"),
-                    window_id: &format!("{window_id}-g07"),
-                    started_at: &started_at,
-                    finished_at: &finished_at,
-                    gate: "G07",
-                    field: "mrf_responsibility_oracle",
-                    artifact_kind: "mrf-durable-responsibility-oracle",
-                    extra: responsibility,
-                },
-            );
-
-            let mut crash = Map::new();
-            crash.insert(
-                "commit_crash_cases".to_string(),
-                json!([
-                    "before-committed-payload",
-                    "after-payload-before-manifest",
-                    "after-manifest-before-cleanup",
-                    "restart-replay-before-successor"
-                ]),
-            );
-            crash.insert(
-                "crash_points".to_string(),
-                json!([
-                    "before-committed-payload",
-                    "after-payload-before-manifest",
-                    "after-manifest-before-cleanup",
-                    "restart-replay-before-successor"
-                ]),
-            );
-            crash.insert("replayed_records".to_string(), json!(replayed_records));
-            crash.insert("responsibility_anchor_retained".to_string(), json!(anchor_retained));
-            crash.insert("successor_snapshot_published".to_string(), json!(successor_snapshot));
-            crash.insert("proof_discharged_anchor".to_string(), json!(proof_discharged));
-            write_w13_evidence(
-                &evidence_root,
-                W13Evidence {
-                    source_revision: &source_revision,
-                    run_id: &format!("{run_id}-g07-crash"),
-                    window_id: &format!("{window_id}-g07"),
-                    started_at: &started_at,
-                    finished_at: &finished_at,
-                    gate: "G07",
-                    field: "commit_boundary_crash_matrix",
-                    artifact_kind: "mrf-commit-boundary-crash-matrix",
-                    extra: crash,
-                },
-            );
-        }
-
-        if w13_selection_contains(&selection, "g08") {
-            let (
-                enospc_filler_bytes,
-                journal_enospc_observed,
-                checkpoint_enospc_observed,
-                cleanup_delete_on_full_filesystem_observed,
-            ) = enospc_result.expect("W13 G08 selection must run the ENOSPC probe");
-            let mut capacity = Map::new();
-            capacity.insert(
-                "capacity_cases".to_string(),
-                json!(["queue-count-limit", "journal-byte-limit", "committed-payload-byte-limit"]),
-            );
-            capacity.insert("queue_count_rejection_observed".to_string(), json!(true));
-            capacity.insert("journal_byte_rejection_observed".to_string(), json!(true));
-            capacity.insert("replay_limit_raise_observed".to_string(), json!(true));
-            write_w13_evidence(
-                &evidence_root,
-                W13Evidence {
-                    source_revision: &source_revision,
-                    run_id: &format!("{run_id}-g08-capacity"),
-                    window_id: &format!("{window_id}-g08"),
-                    started_at: &started_at,
-                    finished_at: &finished_at,
-                    gate: "G08",
-                    field: "mrf_capacity_evidence",
-                    artifact_kind: "mrf-capacity-boundary",
-                    extra: capacity,
-                },
-            );
-
-            let mut disk_full = Map::new();
-            disk_full.insert(
-                "disk_full_cases".to_string(),
-                json!([
-                    "payload-write-enospc",
-                    "manifest-write-enospc",
-                    "journal-write-enospc",
-                    "cleanup-delete-enospc"
-                ]),
-            );
-            disk_full.insert("disk_full_fault_source".to_string(), json!("runner-provided-filesystem"));
-            disk_full.insert("disk_full_requires_external_enospc_root".to_string(), json!(true));
-            disk_full.insert("enospc_filler_bytes".to_string(), json!(enospc_filler_bytes));
-            disk_full.insert("journal_write_enospc_observed".to_string(), json!(journal_enospc_observed));
-            disk_full.insert("committed_checkpoint_enospc_observed".to_string(), json!(checkpoint_enospc_observed));
-            disk_full.insert(
-                "cleanup_delete_on_full_filesystem_observed".to_string(),
-                json!(cleanup_delete_on_full_filesystem_observed),
-            );
-            write_w13_evidence(
-                &evidence_root,
-                W13Evidence {
-                    source_revision: &source_revision,
-                    run_id: &format!("{run_id}-g08-disk-full"),
-                    window_id: &format!("{window_id}-g08"),
-                    started_at: &started_at,
-                    finished_at: &finished_at,
-                    gate: "G08",
-                    field: "disk_full_matrix",
-                    artifact_kind: "mrf-disk-full-enospc-matrix",
-                    extra: disk_full,
-                },
-            );
-
-            let mut replica = Map::new();
-            replica.insert(
-                "replica_loss_cases".to_string(),
-                json!(["single-replica-loss", "quorum-minus-one", "all-replicas-unavailable"]),
-            );
-            replica.insert("no_writable_replica_rejected".to_string(), json!(no_writable_replica_rejected));
-            replica.insert("resident_intent_retained_after_rejection".to_string(), json!(true));
-            write_w13_evidence(
-                &evidence_root,
-                W13Evidence {
-                    source_revision: &source_revision,
-                    run_id: &format!("{run_id}-g08-replica"),
-                    window_id: &format!("{window_id}-g08"),
-                    started_at: &started_at,
-                    finished_at: &finished_at,
-                    gate: "G08",
-                    field: "replica_loss_matrix",
-                    artifact_kind: "mrf-replica-loss-matrix",
-                    extra: replica,
-                },
-            );
-        }
-
-        if w13_selection_contains(&selection, "p4") {
-            let mut scale = Map::new();
-            scale.insert("duration_seconds".to_string(), json!(duration_seconds));
-            scale.insert("queued_records".to_string(), json!(scale_records));
-            scale.insert("coalesced_records".to_string(), json!(scale_coalesced_records));
-            scale.insert("deduped_depth".to_string(), json!(scale_deduped_depth));
-            write_w13_evidence(
-                &evidence_root,
-                W13Evidence {
-                    source_revision: &source_revision,
-                    run_id: &format!("{run_id}-p4-scale"),
-                    window_id: window_id.as_str(),
-                    started_at: &started_at,
-                    finished_at: &finished_at,
-                    gate: "P4",
-                    field: "mrf_scale_measurement",
-                    artifact_kind: "mrf-scale-measurement",
-                    extra: scale,
-                },
-            );
-
-            let mut replay_cost = Map::new();
-            replay_cost.insert("duration_seconds".to_string(), json!(duration_seconds));
-            replay_cost.insert("replayed_records".to_string(), json!(replayed_records));
-            replay_cost.insert("responsibility_anchor_retained".to_string(), json!(anchor_retained));
-            replay_cost.insert("successor_snapshot_published".to_string(), json!(successor_snapshot));
-            replay_cost.insert("elapsed_seconds".to_string(), json!(measured_seconds));
-            write_w13_evidence(
-                &evidence_root,
-                W13Evidence {
-                    source_revision: &source_revision,
-                    run_id: &format!("{run_id}-p4-replay-cost"),
-                    window_id: window_id.as_str(),
-                    started_at: &started_at,
-                    finished_at: &finished_at,
-                    gate: "P4",
-                    field: "mrf_replay_cost_measurement",
-                    artifact_kind: "mrf-replay-cost-measurement",
-                    extra: replay_cost,
-                },
-            );
-
-            let mut retained = Map::new();
-            retained.insert("duration_seconds".to_string(), json!(duration_seconds));
-            retained.insert(
-                "retained_responsibility_cases".to_string(),
-                json!([
-                    "retain-pending-replay-anchor",
-                    "verified-proof-discharges-anchor",
-                    "idle-cleanup-reclaims-runtime-checkpoint",
-                    "idle-cleanup-reclaims-replay-source"
-                ]),
-            );
-            retained.insert("retention_window_seconds".to_string(), json!(duration_seconds));
-            retained.insert("idle_cleanup_observed".to_string(), json!(idle_cleanup));
-            retained.insert("verified_proof_discharge_observed".to_string(), json!(proof_discharged));
-            retained.insert("replayed_records".to_string(), json!(replayed_records));
-            retained.insert("responsibility_anchor_retained".to_string(), json!(anchor_retained));
-            retained.insert("successor_snapshot_published".to_string(), json!(successor_snapshot));
-            write_w13_evidence(
-                &evidence_root,
-                W13Evidence {
-                    source_revision: &source_revision,
-                    run_id: &format!("{run_id}-p4-retained"),
-                    window_id: window_id.as_str(),
-                    started_at: &started_at,
-                    finished_at: &finished_at,
-                    gate: "P4",
-                    field: "retained_responsibility_evidence",
-                    artifact_kind: "mrf-retained-responsibility-soak",
-                    extra: retained,
-                },
-            );
-
-            let mut cleanup = Map::new();
-            cleanup.insert("duration_seconds".to_string(), json!(duration_seconds));
-            cleanup.insert(
-                "cleanup_gc_cases".to_string(),
-                json!([
-                    "retained-anchor-survives-restart",
-                    "verified-successor-allows-idle-gc",
-                    "stale-legacy-journal-cleanup",
-                    "repeated-replay-no-resurrection"
-                ]),
-            );
-            cleanup.insert("verified_idle_gc_observed".to_string(), json!(idle_cleanup));
-            cleanup.insert("pending_responsibilities_after_gc".to_string(), json!(0));
-            cleanup.insert("stale_journals_after_gc".to_string(), json!(stale_after_gc));
-            cleanup.insert("replayed_records".to_string(), json!(replayed_records));
-            cleanup.insert("responsibility_anchor_retained".to_string(), json!(anchor_retained));
-            cleanup.insert("successor_snapshot_published".to_string(), json!(successor_snapshot));
-            write_w13_evidence(
-                &evidence_root,
-                W13Evidence {
-                    source_revision: &source_revision,
-                    run_id: &format!("{run_id}-p4-cleanup"),
-                    window_id: window_id.as_str(),
-                    started_at: &started_at,
-                    finished_at: &finished_at,
-                    gate: "P4",
-                    field: "mrf_cleanup_gc_soak_evidence",
-                    artifact_kind: "mrf-cleanup-gc-soak",
-                    extra: cleanup,
-                },
-            );
-        }
-    }
-
     #[test]
     fn tick_action_table() {
         use TickAction::*;

crates/scanner/src/lib.rs

@@ -90,9 +90,10 @@ pub use scanner::{
     ScannerCycleScheduleStatus, ScannerPauseBacklogAlertReason, ScannerPauseBacklogPhase, ScannerPauseBacklogStatus,
     ScannerPauseBacklogThresholds, ScannerRecoveryIntentAcceptResult, ScannerRecoveryIntentConflict, ScannerRecoveryIntentRecord,
     ScannerRecoveryIntentRequest, ScannerUsageStateResetResult, accept_scanner_usage_recovery_intent,
-    get_scanner_usage_recovery_intent, init_data_scanner, init_scanner_with_recovery, reset_scanner_cycle_recovery,
-    reset_scanner_usage_state_for_full_rebuild, run_scanner_usage_recovery_intent, scanner_cycle_recovery_status,
-    scanner_cycle_schedule_status, scanner_pause_backlog_status, scanner_recovery_actor_sha256, scanner_topology_digest,
+    get_scanner_usage_recovery_intent, init_data_scanner, init_scanner_with_recovery, register_scanner_pause_backlog_retirement,
+    reset_scanner_cycle_recovery, reset_scanner_usage_state_for_full_rebuild, run_scanner_usage_recovery_intent,
+    scanner_cycle_recovery_status, scanner_cycle_schedule_status, scanner_pause_backlog_status, scanner_recovery_actor_sha256,
+    scanner_topology_digest,
 };
 pub use scanner_io::{
     ScannerDirtyUsageAckError, ScannerDirtyUsageBucket, ScannerDirtyUsageSnapshot, ScannerDirtyUsageState,

crates/scanner/src/scanner.rs

@@ -3628,7 +3628,7 @@ pub(crate) use activity::{
 pub(crate) use activity::{ScannerCycleOutcome, scanner_cycle_outcome_with_pending_maintenance};
 pub use backlog::{
     ScannerPauseBacklogAlertReason, ScannerPauseBacklogPhase, ScannerPauseBacklogStatus, ScannerPauseBacklogThresholds,
-    scanner_pause_backlog_status,
+    register_scanner_pause_backlog_retirement, scanner_pause_backlog_status,
 };
 #[cfg(test)]
 pub(crate) use cycle_state::encode_scanner_cycle_fence_for_test;

crates/scanner/src/scanner/tests.rs

@@ -61,28 +61,43 @@ async fn setup_scanner_cycle_store_with_pool_count(
 }
 
 async fn setup_scanner_cycle_store_at_path(root: &Path, seed_usage_baseline: bool, pool_count: usize) -> Arc<ECStore> {
+    setup_scanner_cycle_store_at_path_with_sets(root, seed_usage_baseline, pool_count, 1).await
+}
+
+pub(super) async fn setup_scanner_cycle_store_at_path_with_sets(
+    root: &Path,
+    seed_usage_baseline: bool,
+    pool_count: usize,
+    sets_per_pool: usize,
+) -> Arc<ECStore> {
     init_ecstore_config_for_scanner_tests();
     let mut pools = Vec::with_capacity(pool_count);
     for pool_index in 0..pool_count {
         let mut endpoints = Vec::new();
-        for disk_index in 0..4 {
-            let disk_path = root.join(format!("pool{pool_index}/disk{disk_index}"));
-            tokio::fs::create_dir_all(&disk_path)
-                .await
-                .expect("scanner cycle test disk should be created");
-            let mut endpoint =
-                Endpoint::try_from(disk_path.to_str().expect("disk path should be utf8")).expect("endpoint should parse");
-            endpoint.set_pool_index(pool_index);
-            endpoint.set_set_index(0);
-            endpoint.set_disk_index(disk_index);
-            endpoints.push(endpoint);
+        for set_index in 0..sets_per_pool {
+            for disk_index in 0..4 {
+                let disk_path = if sets_per_pool == 1 {
+                    root.join(format!("pool{pool_index}/disk{disk_index}"))
+                } else {
+                    root.join(format!("pool{pool_index}/set{set_index}/disk{disk_index}"))
+                };
+                tokio::fs::create_dir_all(&disk_path)
+                    .await
+                    .expect("scanner cycle test disk should be created");
+                let mut endpoint =
+                    Endpoint::try_from(disk_path.to_str().expect("disk path should be utf8")).expect("endpoint should parse");
+                endpoint.set_pool_index(pool_index);
+                endpoint.set_set_index(set_index);
+                endpoint.set_disk_index(disk_index);
+                endpoints.push(endpoint);
+            }
         }
         pools.push(PoolEndpoints {
             legacy: false,
-            set_count: 1,
+            set_count: sets_per_pool,
             drives_per_set: 4,
             endpoints: Endpoints::from(endpoints),
-            cmd_line: if pool_count == 1 {
+            cmd_line: if pool_count == 1 && sets_per_pool == 1 {
                 "scanner-cycle-metrics".to_string()
             } else {
                 format!("scanner-cycle-metrics-pool-{pool_index}")

crates/scanner/src/scanner_io/tests.rs

@@ -251,7 +251,7 @@ fn scanner_durable_segment_invalidation_evidence_requires_matching_complete_set_
     let expected_sources = HashSet::from([DataUsageCacheSource::new(0, 0), DataUsageCacheSource::new(0, 1)]);
     let results = vec![
         complete_set_cache_with_segment_proof(DataUsageCacheSource::new(0, 0), process_proof.clone()),
-        complete_set_cache_with_segment_proof(DataUsageCacheSource::new(0, 1), process_proof),
+        complete_set_cache_with_segment_proof(DataUsageCacheSource::new(0, 1), process_proof.clone()),
     ];
 
     let durable_evidence = scanner_durable_segment_invalidation_evidence(&dirty_usage_snapshot, &results, &expected_sources);

crates/scanner/src/scanner_io/tests/scoped_entry_fallback.rs

@@ -226,40 +226,9 @@ fn record_segment_dirty_usage(bucket: &str) {
     }
 }
 
-// The scoped fallback fixture keeps two EC pools and several scan futures live
-// at once. Run the async cases on a dedicated stack so Linux libtest defaults
-// exercise the assertions instead of aborting before the oracle finishes.
-fn run_scoped_entry_fallback_test<F, Fut>(thread_name: &'static str, test_fn: F)
-where
-    F: FnOnce() -> Fut + Send + 'static,
-    Fut: std::future::Future<Output = ()> + 'static,
-{
-    let handle = std::thread::Builder::new()
-        .name(thread_name.to_string())
-        .stack_size(32 * 1024 * 1024)
-        .spawn(move || {
-            let runtime = tokio::runtime::Builder::new_current_thread()
-                .enable_all()
-                .build()
-                .expect("scoped entry fallback runtime should build");
-            runtime.block_on(test_fn());
-        })
-        .expect("scoped entry fallback test thread should spawn");
-    if let Err(payload) = handle.join() {
-        std::panic::resume_unwind(payload);
-    }
-}
-
-#[test]
+#[tokio::test]
 #[serial]
-fn scoped_entry_fallback_distinguishes_planned_scope_from_real_cold_walks() {
-    run_scoped_entry_fallback_test(
-        "scanner-scoped-entry-planned-scope",
-        scoped_entry_fallback_distinguishes_planned_scope_from_real_cold_walks_case,
-    );
-}
-
-async fn scoped_entry_fallback_distinguishes_planned_scope_from_real_cold_walks_case() {
+async fn scoped_entry_fallback_distinguishes_planned_scope_from_real_cold_walks() {
     let (_dir, store) = setup_two_pool_scanner_store().await;
     clear_dirty_usage_buckets_for_tests();
     let hot = format!("hot-{}", Uuid::new_v4().simple());
@@ -282,16 +251,9 @@ async fn scoped_entry_fallback_distinguishes_planned_scope_from_real_cold_walks_
     clear_dirty_usage_buckets_for_tests();
 }
 
-#[test]
+#[tokio::test]
 #[serial]
-fn scoped_entry_fallback_rejects_invalid_persisted_baseline_at_the_walker() {
-    run_scoped_entry_fallback_test(
-        "scanner-scoped-entry-invalid-baseline",
-        scoped_entry_fallback_rejects_invalid_persisted_baseline_at_the_walker_case,
-    );
-}
-
-async fn scoped_entry_fallback_rejects_invalid_persisted_baseline_at_the_walker_case() {
+async fn scoped_entry_fallback_rejects_invalid_persisted_baseline_at_the_walker() {
     let (_dir, store) = setup_two_pool_scanner_store().await;
     clear_dirty_usage_buckets_for_tests();
     let hot = format!("hot-{}", Uuid::new_v4().simple());
@@ -348,16 +310,9 @@ async fn scoped_entry_fallback_rejects_invalid_persisted_baseline_at_the_walker_
     clear_dirty_usage_buckets_for_tests();
 }
 
-#[test]
+#[tokio::test]
 #[serial]
-fn scoped_entry_fallback_covers_overflow_and_new_bucket_inventory() {
-    run_scoped_entry_fallback_test(
-        "scanner-scoped-entry-overflow-inventory",
-        scoped_entry_fallback_covers_overflow_and_new_bucket_inventory_case,
-    );
-}
-
-async fn scoped_entry_fallback_covers_overflow_and_new_bucket_inventory_case() {
+async fn scoped_entry_fallback_covers_overflow_and_new_bucket_inventory() {
     let (_dir, store) = setup_two_pool_scanner_store().await;
     clear_dirty_usage_buckets_for_tests();
     let hot = format!("hot-{}", Uuid::new_v4().simple());

crates/scanner/src/storage_api.rs

@@ -28,6 +28,13 @@ pub(crate) use s3s::dto::{
 #[cfg(test)]
 pub(crate) use s3s::dto::{ExpirationStatus as EcstoreExpirationStatus, LifecycleRule as EcstoreLifecycleRule};
 
+pub(crate) use rustfs_ecstore::api::data_usage::{
+    MAX_SCANNER_PAUSE_BACKLOG_BYTES, ScannerPauseBacklogRetirementPlan, ScannerPauseBacklogRetirementReplica,
+    register_scanner_pause_backlog_retirement_planner,
+};
+#[cfg(test)]
+pub(crate) use rustfs_ecstore::api::data_usage::{NativeScannerPauseBacklogWriteFault, SourceCleanupDeleteBarrier};
+
 pub(crate) use rustfs_ecstore::api::bucket::bucket_target_sys::BucketTargetSys as EcstoreBucketTargetSys;
 pub(crate) use rustfs_ecstore::api::bucket::lifecycle::bucket_lifecycle_audit::LcEventSrc as EcstoreLcEventSrc;
 pub(crate) use rustfs_ecstore::api::bucket::lifecycle::bucket_lifecycle_ops::{
@@ -135,6 +142,13 @@ use rustfs_storage_api as storage_contracts;
 pub(crate) type EcstoreHealResultItem = <EcstoreStore as storage_contracts::HealOperations>::HealResultItem;
 
 pub(crate) mod owner {
+    pub(crate) use super::{
+        MAX_SCANNER_PAUSE_BACKLOG_BYTES, ScannerPauseBacklogRetirementPlan, ScannerPauseBacklogRetirementReplica,
+        register_scanner_pause_backlog_retirement_planner,
+    };
+    #[cfg(test)]
+    pub(crate) use super::{NativeScannerPauseBacklogWriteFault, SourceCleanupDeleteBarrier};
+
     #[cfg(test)]
     pub(crate) use rustfs_ecstore::api::set_disk::test_util::hold_namespace_commit as ecstore_hold_namespace_commit;
 

docs/testing/ci-gates.md

@@ -135,24 +135,21 @@ evidence registered in `.config/scanner-heal-required-tests.json`. It records
 already-built binaries and checks existing nextest output; it does not build,
 run tests, deploy servers, inject faults, or start another CI lane.
 
-The registered cases are emitted by existing E2E tests. The original
-`background-target-restart` / `background-target-crash` cases run in
-`e2e-nightly` on a four-node, one-drive-per-node topology. The
-`ec84-target-drive-restart` case runs in `e2e-distributed` on a three-node,
-four-drive EC8+4 topology. When `RUSTFS_SCANNER_HEAL_RUN_DIR` is set, the
-producer checks the actual server and test-executable hashes against `run.json`,
-pins the same server binary for all node starts, and writes its oracle only
-after the real assertions pass. The artifact contains the actual pre/post target
+The initial case is `background-target-restart`, emitted by
+`heal_erasure_disk_rebuild_test::tests::test_cluster_root_heal_recovers_remote_shards_after_background_target_restart`.
+That test already runs in `e2e-nightly`. When `RUSTFS_SCANNER_HEAL_RUN_DIR` is set,
+it checks the actual server and test-executable hashes against `run.json`, pins
+the same server binary for all node starts, and writes its oracle only after
+the real assertions pass. The artifact contains the actual pre/post target
 PIDs, per-node S3 listings, expected and downloaded complete-body hashes/lengths,
 and target-disk `VersionShardCensus` fingerprints. Existing baseline objects
 must match their pre-fault physical manifests; the object created during the
 outage has no pre-fault target shard and is checked for complete physical parts
 and exact S3 content.
 
-These cases are still restart-focused evidence slices. They are not power-loss
-validation, an all-version inventory, or proof of scanner enumeration, exact MRF
-disposition, legacy migration, multi-pool/multi-set release coverage, or
-rollback.
+This case is a **four-node, one-drive-per-node process-restart test**. It is not
+power-loss validation, a 3x4 EC8+4 experiment, an all-version inventory, or proof
+of scanner enumeration, exact MRF disposition, legacy migration, or rollback.
 The schema 2 registry separates the implemented single-set restart lane from
 structured release lanes for authority coverage, checkpoint/crash, status and
 outcome, MRF responsibility, mixed-version rollback, scheduler pressure,
@@ -183,12 +180,27 @@ The producer checks this compiled identity against the receipt; it does not
 copy a current source revision into an older test binary's identity. The E2E
 uses its existing temporary cluster directories and cleanup. `CARGO_TARGET_DIR`
 controls compilation output; nextest's default report store remains the
-workspace's `target/nextest`. Prefer the registry-aware runner for concrete
-cases:
+workspace's `target/nextest`. Execute the existing selected case as follows:
 
 ```bash
-scripts/run_scanner_heal_evidence_case.sh --case background-target-restart
-scripts/run_scanner_heal_evidence_case.sh --case ec84-target-drive-restart
+CASE=background-target-restart
+FILTER='test(test_cluster_root_heal_recovers_remote_shards_after_background_target_restart)'
+RUN_DIR="$PWD/artifacts/scanner-heal-run"
+export RUSTFS_E2E_EXPECTED_FEATURES=default
+scripts/python_bin.sh scripts/check_test_wiring.py \
+  --begin-scanner-heal "$RUN_DIR" "$SERVER_BINARY" "$TEST_BINARY"
+export RUSTFS_SCANNER_HEAL_RUN_DIR="$RUN_DIR"
+export CARGO_BIN_EXE_rustfs="$SERVER_BINARY"
+cargo nextest list --profile e2e-nightly -p e2e_test -E "$FILTER" \
+  --message-format json > "$RUN_DIR/listing.json"
+rm -f target/nextest/e2e-nightly/junit.xml
+set +e
+cargo nextest run --profile e2e-nightly -p e2e_test -E "$FILTER"
+test_exit=$?
+set -e
+cp target/nextest/e2e-nightly/junit.xml "$RUN_DIR/junit.xml"
+scripts/python_bin.sh scripts/check_test_wiring.py --finish-scanner-heal "$RUN_DIR" "$test_exit"
+scripts/python_bin.sh scripts/check_test_wiring.py --check-scanner-heal "$RUN_DIR" "$CASE"
 ```
 
 Set `RUSTFS_E2E_EXPECTED_FEATURES` to the actual intended e2e crate feature set,
@@ -294,21 +306,6 @@ tests, and runs the distributed hard-quota admission E2E. A full run writes
 a failure; a single gate descriptor still does not approve the complete release
 bundle.
 
-The W13 durable MRF replay lanes can emit raw G07/G08/P4 JSON artifacts with:
-
-```bash
-scripts/run_scanner_heal_w13_mrf_evidence.sh
-```
-
-The runner builds the current checkout, runs the ignored MRF evidence test, and
-writes `release-bundle-w13.json` for `--check-scanner-heal-release-bundle-gate`.
-Use `--test g07|g08|p4` while narrowing a failure. G08 disk-full evidence must
-run against a real fillable filesystem: on Linux as root the runner mounts a
-small tmpfs automatically, otherwise pass `--enospc-root` pointing at a
-pre-mounted small filesystem. P4 is release evidence only when it completes the
-default two-hour soak; `--allow-short-soak` is diagnostic and skips P4 bundle
-gate validation.
-
 When the real release lanes have produced their dedicated artifacts, validate
 the complete hard-gate bundle with:
 

docs/testing/distributed-e2e.md

@@ -91,16 +91,4 @@ tests, and leaves the required raw G09 artifacts under
 inputs for the Scanner/Heal release bundle gate; the runner does not mark the
 full release matrix complete by itself.
 
-The distributed Scanner/Heal EC8+4 restart case is registered as
-`ec84-target-drive-restart` and selected by this profile:
-
-```bash
-scripts/run_scanner_heal_evidence_case.sh --case ec84-target-drive-restart
-```
-
-That command records the current build, runs exactly the registered
-`distributed::heal_test` case, validates the JUnit/listing/oracle receipt, and
-keeps the wider release gate blocked until the remaining release evidence lanes
-have measured artifacts.
-
 Membership is pinned by `.config/e2e-distributed-selection.txt`. Update the Linux and Darwin entries with `python3 ./scripts/check_test_wiring.py --update-profile e2e-distributed <listing.json> <platform>` after adding or renaming a case.

rustfs/src/app/admin_usecase.rs

@@ -775,7 +775,6 @@ mod tests {
             scan_plan_digest: Some([1; 32]),
             complete: true,
             tombstone: false,
-            segment_invalidation_proof: None,
         }];
         DefaultAdminUsecase::narrow_data_usage_snapshot_to_measured_buckets(&mut info, ["bucket-a".to_string()]);
         assert_eq!(info.usage_snapshot_converged, Some(false));

rustfs/src/startup_storage.rs

@@ -152,6 +152,7 @@ pub(crate) async fn init_startup_storage_runtime(
     readiness: Arc<GlobalReadiness>,
     instance_ctx: Arc<InstanceContext>,
 ) -> Result<StartupStorageRuntime> {
+    rustfs_scanner::register_scanner_pause_backlog_retirement();
     let ctx = CancellationToken::new();
 
     debug!(
@@ -195,6 +196,7 @@ pub(crate) async fn init_embedded_startup_storage_runtime(
     shutdown_token: CancellationToken,
     instance_ctx: Arc<InstanceContext>,
 ) -> Result<StartupStorageRuntime> {
+    rustfs_scanner::register_scanner_pause_backlog_retirement();
     let store =
         match ECStore::new_with_instance_ctx(server_addr, endpoint_pools.clone(), shutdown_token.clone(), instance_ctx).await {
             Ok(store) => store,

scripts/README.md

@@ -57,12 +57,10 @@ their issue closes.
 | `run_scanner_validation_harness.sh` | dev-tool | Scanner validation harness | `docs/operations/scanner-benchmark-runbook.md` |
 | `run_scanner_heal_evidence_case.sh` | dev-tool | Runs one Scanner/Heal release-evidence registry case and checks the produced receipt/oracle | `.config/scanner-heal-required-tests.json`; `check_test_wiring.py --check-scanner-heal` |
 | `run_scanner_heal_g09_upgrade_evidence.sh` | dev-tool | Runs the G09 mixed-version and rollback upgrade E2E lanes against a pinned previous release and verifies the raw evidence artifacts | `docs/testing/ci-gates.md`; `.github/workflows/e2e-upgrade.yml`; `test_scanner_heal_g09_upgrade_evidence.sh` |
-| `run_scanner_heal_w13_mrf_evidence.sh` | dev-tool | Runs the W13 durable MRF replay lanes and writes G07/G08/P4 bundle-ready evidence descriptors | `docs/testing/ci-gates.md`; `test_scanner_heal_w13_mrf_evidence.sh` |
 | `run_scanner_heal_w16_recovery_evidence.sh` | dev-tool | Runs the W16 recovery-intent and quota authority lanes and writes G04/G12 bundle-ready evidence descriptors | `docs/testing/ci-gates.md`; `test_scanner_heal_w16_recovery_evidence.sh` |
 | `test_scanner_validation_harness.sh` | dev-tool | Self-test for the scanner validation harness | — |
 | `test_scanner_heal_g09_upgrade_evidence.sh` | dev-tool | Shell self-test for the Scanner/Heal G09 upgrade evidence runner | — |
 | `test_scanner_heal_w16_recovery_evidence.sh` | dev-tool | Shell self-test for the Scanner/Heal W16 recovery evidence runner | — |
-| `test_scanner_heal_w13_mrf_evidence.sh` | dev-tool | Shell self-test for the Scanner/Heal W13 MRF evidence runner | — |
 | `scanner_abba.py` | dev-tool | Scanner/heal ABBA orchestration and evidence gates via `run_scanner_validation_harness.sh --abba` | `docs/operations/scanner-benchmark-runbook.md` |
 | `test_scanner_abba.py` | dev-tool | Synthetic ABBA adapter and failure-path tests | `test_scanner_validation_harness.sh` |
 | `test_build_rustfs_options.sh` | dev-tool | Shell test for rustfs build-option wiring | `make test` (script-tests) |

scripts/run_scanner_heal_w13_mrf_evidence.sh

@@ -1,605 +0,0 @@
-#!/usr/bin/env bash
-
-set -euo pipefail
-
-ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
-PYTHON_BIN="${RUSTFS_PYTHON_BIN:-python3}"
-MIN_FREE_KIB="${RUSTFS_W13_MIN_FREE_KIB:-4194304}"
-SOAK_SECONDS="${RUSTFS_W13_MRF_SOAK_SECONDS:-7200}"
-ENOSPC_TMPFS_SIZE="${RUSTFS_W13_ENOSPC_TMPFS_SIZE:-16m}"
-
-RUN_DIR=""
-ENOSPC_ROOT="${RUSTFS_SCANNER_HEAL_W13_ENOSPC_ROOT:-}"
-TEST_SELECTION="all"
-PLAN_ONLY=0
-ALLOW_DIRTY=0
-ALLOW_SHORT_SOAK=0
-SKIP_BUILD=0
-VERBOSE=0
-ENOSPC_TMPFS_MOUNTED=0
-
-usage() {
-    cat <<'USAGE'
-Usage: scripts/run_scanner_heal_w13_mrf_evidence.sh [OPTIONS]
-
-Build the current checkout, run the W13 MRF durable replay evidence test, verify
-the raw JSON artifacts, and write bundle-ready G07/G08/P4 release descriptors.
-
-Options:
-  --run-dir DIR       New evidence directory (default: target/scanner-heal-w13-evidence/TIMESTAMP)
-  --out-dir DIR       Alias for --run-dir
-  --test NAME         all, g07, g08, or p4 (default: all)
-  --soak-seconds N    P4 soak duration in seconds (default: 7200)
-  --enospc-root DIR   Pre-mounted small filesystem used for real G08 ENOSPC evidence
-  --allow-short-soak  Diagnostic only: allow P4 runs shorter than release duration
-  --allow-dirty      Allow tracked source changes while collecting evidence
-  --skip-build       Reuse an existing target/debug/rustfs binary
-  --plan-only        Print the resolved plan without building or running tests
-  --dry-run          Alias for --plan-only
-  --self-test        Run lightweight CLI and descriptor plumbing checks
-  --verbose          Stream command output instead of storing it under the run directory
-  -h, --help         Show this help
-
-Required output files:
-  g07-mrf-responsibility/G07-mrf_responsibility_oracle.json
-  g07-mrf-responsibility/G07-commit_boundary_crash_matrix.json
-  g08-mrf-capacity/G08-mrf_capacity_evidence.json
-  g08-mrf-capacity/G08-disk_full_matrix.json
-  g08-mrf-capacity/G08-replica_loss_matrix.json
-  p4-mrf-soak/P4-mrf_scale_measurement.json
-  p4-mrf-soak/P4-mrf_replay_cost_measurement.json
-  p4-mrf-soak/P4-retained_responsibility_evidence.json
-  p4-mrf-soak/P4-mrf_cleanup_gc_soak_evidence.json
-
-Environment overrides:
-  RUSTFS_SCANNER_HEAL_W13_OUTPUT_ROOT
-  RUSTFS_SCANNER_HEAL_W13_ENOSPC_ROOT
-  RUSTFS_W13_MIN_FREE_KIB
-  RUSTFS_W13_MRF_SOAK_SECONDS
-  RUSTFS_W13_ENOSPC_TMPFS_SIZE
-
-Short-soak runs are for runner diagnostics only. They validate raw artifacts but
-do not validate the P4 release bundle gate.
-USAGE
-}
-
-die() {
-    echo "ERROR: $*" >&2
-    exit 1
-}
-
-require_value() {
-    local option="$1"
-    local count="$2"
-    if [[ "$count" -lt 2 ]]; then
-        die "missing value for $option"
-    fi
-}
-
-case_names() {
-    case "$TEST_SELECTION" in
-        all)
-            printf '%s\n' g07 g08 p4
-            ;;
-        g07|g08|p4)
-            printf '%s\n' "$TEST_SELECTION"
-            ;;
-        *)
-            die "unknown test selection: $TEST_SELECTION"
-            ;;
-    esac
-}
-
-validate_test_selection() {
-    case "$TEST_SELECTION" in
-        all|g07|g08|p4)
-            ;;
-        *)
-            die "unknown test selection: $TEST_SELECTION"
-            ;;
-    esac
-}
-
-selection_includes() {
-    local needle="$1"
-    [[ "$TEST_SELECTION" == "all" || "$TEST_SELECTION" == "$needle" ]]
-}
-
-normalize_path() {
-    local path="$1"
-    if [[ "$path" == /* ]]; then
-        echo "$path"
-    else
-        echo "$ROOT/$path"
-    fi
-}
-
-cargo_target_dir() {
-    if [[ -n "${CARGO_TARGET_DIR:-}" ]]; then
-        normalize_path "$CARGO_TARGET_DIR"
-    else
-        echo "$ROOT/target"
-    fi
-}
-
-write_rustfs_features_stamp() {
-    local target_dir
-    target_dir="$(cargo_target_dir)"
-    mkdir -p "$target_dir/debug"
-    : >"$target_dir/debug/rustfs.features"
-}
-
-artifact_dir_for() {
-    case "$1" in
-        g07)
-            echo "g07-mrf-responsibility"
-            ;;
-        g08)
-            echo "g08-mrf-capacity"
-            ;;
-        p4)
-            echo "p4-mrf-soak"
-            ;;
-        *)
-            die "unknown W13 case: $1"
-            ;;
-    esac
-}
-
-check_empty_case_dir() {
-    local dir="$1"
-    if [[ -d "$dir" ]] && find "$dir" -mindepth 1 -print -quit | grep -q .; then
-        die "evidence case directory is not empty: $dir"
-    fi
-}
-
-ensure_min_free_space() {
-    local path="$1"
-    local available
-    mkdir -p "$path"
-    available="$(df -Pk "$path" | awk 'NR == 2 { print $4 }')"
-    if [[ -z "$available" ]]; then
-        die "could not determine free space for $path"
-    fi
-    if (( available < MIN_FREE_KIB )); then
-        die "insufficient free space for W13 evidence run at $path: need ${MIN_FREE_KIB} KiB, found ${available} KiB"
-    fi
-}
-
-cleanup_enospc_root() {
-    if [[ "$ENOSPC_TMPFS_MOUNTED" == 1 && -n "$ENOSPC_ROOT" ]]; then
-        umount "$ENOSPC_ROOT" >/dev/null 2>&1 || true
-    fi
-}
-
-prepare_enospc_root() {
-    if ! selection_includes g08; then
-        return
-    fi
-    if [[ -n "$ENOSPC_ROOT" ]]; then
-        ENOSPC_ROOT="$(normalize_path "$ENOSPC_ROOT")"
-        mkdir -p "$ENOSPC_ROOT"
-        return
-    fi
-    if [[ "$(uname -s)" != "Linux" ]]; then
-        die "G08 disk-full evidence requires --enospc-root on non-Linux hosts"
-    fi
-    if [[ "$(id -u)" != "0" ]]; then
-        die "G08 disk-full evidence requires --enospc-root or root privileges to mount a tmpfs"
-    fi
-    if ! command -v mount >/dev/null 2>&1 || ! command -v umount >/dev/null 2>&1; then
-        die "G08 disk-full evidence requires mount and umount, or a pre-mounted --enospc-root"
-    fi
-    ENOSPC_ROOT="$RUN_DIR/enospc-root"
-    mkdir -p "$ENOSPC_ROOT"
-    mount -t tmpfs -o "size=$ENOSPC_TMPFS_SIZE" rustfs-w13-enospc "$ENOSPC_ROOT"
-    ENOSPC_TMPFS_MOUNTED=1
-}
-
-run_logged() {
-    local label="$1"
-    shift
-    local log="$RUN_DIR/logs/$label.log"
-    mkdir -p "$(dirname "$log")"
-    if [[ "$VERBOSE" == 1 ]]; then
-        "$@"
-        return
-    fi
-    if ! "$@" >"$log" 2>&1; then
-        echo "$label failed; log: $log" >&2
-        tail -80 "$log" >&2 || true
-        return 1
-    fi
-    echo "PASS: $label"
-}
-
-validate_artifacts() {
-    local source_revision="$1"
-    "$PYTHON_BIN" - "$RUN_DIR" "$source_revision" "$TEST_SELECTION" <<'PY'
-import json
-import pathlib
-import sys
-
-run_dir = pathlib.Path(sys.argv[1])
-source_revision = sys.argv[2]
-selection = sys.argv[3]
-
-expected = {
-    "g07": [
-        ("g07-mrf-responsibility/G07-mrf_responsibility_oracle.json", "G07", "mrf_responsibility_oracle", "mrf-durable-responsibility-oracle"),
-        ("g07-mrf-responsibility/G07-commit_boundary_crash_matrix.json", "G07", "commit_boundary_crash_matrix", "mrf-commit-boundary-crash-matrix"),
-    ],
-    "g08": [
-        ("g08-mrf-capacity/G08-mrf_capacity_evidence.json", "G08", "mrf_capacity_evidence", "mrf-capacity-boundary"),
-        ("g08-mrf-capacity/G08-disk_full_matrix.json", "G08", "disk_full_matrix", "mrf-disk-full-enospc-matrix"),
-        ("g08-mrf-capacity/G08-replica_loss_matrix.json", "G08", "replica_loss_matrix", "mrf-replica-loss-matrix"),
-    ],
-    "p4": [
-        ("p4-mrf-soak/P4-mrf_scale_measurement.json", "P4", "mrf_scale_measurement", "mrf-scale-measurement"),
-        ("p4-mrf-soak/P4-mrf_replay_cost_measurement.json", "P4", "mrf_replay_cost_measurement", "mrf-replay-cost-measurement"),
-        ("p4-mrf-soak/P4-retained_responsibility_evidence.json", "P4", "retained_responsibility_evidence", "mrf-retained-responsibility-soak"),
-        ("p4-mrf-soak/P4-mrf_cleanup_gc_soak_evidence.json", "P4", "mrf_cleanup_gc_soak_evidence", "mrf-cleanup-gc-soak"),
-    ],
-}
-if selection != "all":
-    expected = {selection: expected[selection]}
-
-for artifacts in expected.values():
-    for relative, gate, field, artifact_kind in artifacts:
-        path = run_dir / relative
-        if not path.is_file():
-            raise SystemExit(f"missing W13 evidence artifact: {relative}")
-        evidence = json.loads(path.read_text())
-        if evidence.get("schema") != 1:
-            raise SystemExit(f"{relative}: expected schema 1")
-        if evidence.get("evidence_type") != "measured":
-            raise SystemExit(f"{relative}: expected measured evidence")
-        if evidence.get("artifact_kind") != artifact_kind:
-            raise SystemExit(f"{relative}: unexpected artifact kind")
-        if evidence.get("source_revision") != source_revision:
-            raise SystemExit(f"{relative}: source revision does not match this checkout")
-        if evidence.get("gate") != gate or evidence.get("field") != field:
-            raise SystemExit(f"{relative}: unexpected gate or field")
-        if gate == "G07":
-            crash_points = evidence.get("crash_points")
-            if not isinstance(crash_points, list) or not crash_points:
-                raise SystemExit(f"{relative}: missing crash points")
-        if gate == "P4":
-            duration = evidence.get("duration_seconds")
-            if not isinstance(duration, int) or duration <= 0:
-                raise SystemExit(f"{relative}: invalid P4 duration")
-        if gate == "G08" and field == "disk_full_matrix":
-            if evidence.get("journal_write_enospc_observed") is not True:
-                raise SystemExit(f"{relative}: journal ENOSPC was not observed")
-            if evidence.get("committed_checkpoint_enospc_observed") is not True:
-                raise SystemExit(f"{relative}: committed checkpoint ENOSPC was not observed")
-            if evidence.get("cleanup_delete_on_full_filesystem_observed") is not True:
-                raise SystemExit(f"{relative}: cleanup delete on a full filesystem was not observed")
-            filler_bytes = evidence.get("enospc_filler_bytes")
-            if not isinstance(filler_bytes, int) or filler_bytes <= 0:
-                raise SystemExit(f"{relative}: ENOSPC filler byte count is invalid")
-
-print("PASS: W13 raw MRF evidence artifacts verified")
-PY
-}
-
-check_release_gate() {
-    local descriptor="$1"
-    local gate="$2"
-    local output="$RUN_DIR/logs/check-${gate}.json"
-    "$PYTHON_BIN" "$ROOT/scripts/check_test_wiring.py" --check-scanner-heal-release-bundle-gate "$descriptor" "$gate" >"$output"
-    "$PYTHON_BIN" - "$output" "$gate" <<'PY'
-import json
-import pathlib
-import sys
-
-path = pathlib.Path(sys.argv[1])
-gate = sys.argv[2]
-status = json.loads(path.read_text())
-if status.get("decision") != "verified" or status.get("verified_gate") != gate:
-    print(path.read_text(), file=sys.stderr)
-    raise SystemExit(f"{gate} release bundle gate was not verified")
-print(path.read_text().strip())
-PY
-}
-
-write_release_descriptor() {
-    local source_revision="$1"
-    "$PYTHON_BIN" - "$ROOT" "$RUN_DIR" "$source_revision" "$TEST_SELECTION" <<'PY'
-import hashlib
-import json
-import pathlib
-import sys
-
-root = pathlib.Path(sys.argv[1])
-run_dir = pathlib.Path(sys.argv[2])
-source_revision = sys.argv[3]
-selection = sys.argv[4]
-descriptor = run_dir / "release-bundle-w13.json"
-registry = json.loads((root / ".config/scanner-heal-required-tests.json").read_text())
-requirements = {item["gate"]: item for item in registry["release_requirements"]}
-artifacts = {
-    "G07": {
-        "mrf_responsibility_oracle": run_dir / "g07-mrf-responsibility" / "G07-mrf_responsibility_oracle.json",
-        "commit_boundary_crash_matrix": run_dir / "g07-mrf-responsibility" / "G07-commit_boundary_crash_matrix.json",
-    },
-    "G08": {
-        "mrf_capacity_evidence": run_dir / "g08-mrf-capacity" / "G08-mrf_capacity_evidence.json",
-        "disk_full_matrix": run_dir / "g08-mrf-capacity" / "G08-disk_full_matrix.json",
-        "replica_loss_matrix": run_dir / "g08-mrf-capacity" / "G08-replica_loss_matrix.json",
-    },
-    "P4": {
-        "mrf_scale_measurement": run_dir / "p4-mrf-soak" / "P4-mrf_scale_measurement.json",
-        "mrf_replay_cost_measurement": run_dir / "p4-mrf-soak" / "P4-mrf_replay_cost_measurement.json",
-        "retained_responsibility_evidence": run_dir / "p4-mrf-soak" / "P4-retained_responsibility_evidence.json",
-        "mrf_cleanup_gc_soak_evidence": run_dir / "p4-mrf-soak" / "P4-mrf_cleanup_gc_soak_evidence.json",
-    },
-}
-if selection == "g07":
-    artifacts = {"G07": artifacts["G07"]}
-elif selection == "g08":
-    artifacts = {"G08": artifacts["G08"]}
-elif selection == "p4":
-    artifacts = {"P4": artifacts["P4"]}
-
-mirrors = {
-    ("G07", "mrf_responsibility_oracle"): ("crash_points", "mrf_responsibility_cases", "replayed_records", "responsibility_anchor_retained", "successor_snapshot_published"),
-    ("G07", "commit_boundary_crash_matrix"): ("crash_points", "commit_crash_cases", "replayed_records", "responsibility_anchor_retained", "successor_snapshot_published"),
-    ("G08", "mrf_capacity_evidence"): ("capacity_cases",),
-    ("G08", "disk_full_matrix"): ("disk_full_cases",),
-    ("G08", "replica_loss_matrix"): ("replica_loss_cases",),
-    ("P4", "mrf_replay_cost_measurement"): ("duration_seconds", "replayed_records", "responsibility_anchor_retained", "successor_snapshot_published"),
-    ("P4", "retained_responsibility_evidence"): ("duration_seconds", "retained_responsibility_cases", "retention_window_seconds", "idle_cleanup_observed", "verified_proof_discharge_observed", "replayed_records", "responsibility_anchor_retained", "successor_snapshot_published"),
-    ("P4", "mrf_cleanup_gc_soak_evidence"): ("duration_seconds", "cleanup_gc_cases", "verified_idle_gc_observed", "pending_responsibilities_after_gc", "stale_journals_after_gc", "replayed_records", "responsibility_anchor_retained", "successor_snapshot_published"),
-}
-
-def digest(path: pathlib.Path) -> str:
-    hasher = hashlib.sha256()
-    with path.open("rb") as source:
-        for chunk in iter(lambda: source.read(1024 * 1024), b""):
-            hasher.update(chunk)
-    return hasher.hexdigest()
-
-def relative_to_descriptor(path: pathlib.Path) -> str:
-    return path.resolve(strict=True).relative_to(descriptor.parent.resolve()).as_posix()
-
-gates: dict[str, object] = {}
-for gate, gate_artifacts in artifacts.items():
-    fields: dict[str, object] = {}
-    for field, artifact in gate_artifacts.items():
-        payload = json.loads(artifact.read_text())
-        if payload.get("source_revision") != source_revision:
-            raise SystemExit(f"{gate}.{field}: source revision does not match this checkout")
-        evidence = {
-            "artifact": relative_to_descriptor(artifact),
-            "sha256": digest(artifact),
-            "evidence_type": "measured",
-            "source_revision": source_revision,
-            "run_id": payload["run_id"],
-            "measurement_window_id": payload["measurement_window_id"],
-            "started_at": payload["started_at"],
-            "finished_at": payload["finished_at"],
-            "command": payload["command"],
-            "artifact_format": "json",
-            "summary": payload["summary"],
-        }
-        for mirror in mirrors.get((gate, field), ()):
-            evidence[mirror] = payload[mirror]
-        if gate == "P4" and field == "mrf_scale_measurement":
-            evidence["duration_seconds"] = payload["duration_seconds"]
-        fields[field] = evidence
-    gates[gate] = {
-        "status": "pass",
-        "lane": requirements[gate]["lane"],
-        "evidence_type": "measured",
-        "evidence_fields": fields,
-    }
-
-descriptor.write_text(json.dumps({
-    "schema": 1,
-    "evidence": "measured",
-    "source_revision": source_revision,
-    "gates": gates,
-}, indent=2, sort_keys=True) + "\n")
-print(descriptor)
-PY
-}
-
-run_self_test() {
-    local tmp plan
-    tmp="$(mktemp -d "${TMPDIR:-/tmp}/rustfs-w13-evidence-self-test.XXXXXX")"
-    trap "rm -rf '$tmp'" EXIT
-
-    plan="$("$0" --plan-only --run-dir "$tmp/run" --test all)"
-    [[ "$plan" == *"tests=g07 g08 p4"* ]]
-    [[ "$plan" == *"soak_seconds=7200"* ]]
-    [[ "$plan" == *"run_dir=$tmp/run"* ]]
-    [[ "$(CARGO_TARGET_DIR=relative-target "$0" --plan-only --run-dir "$tmp/run" --test g07)" == *"target_dir=$ROOT/relative-target"* ]]
-
-    if "$0" --plan-only --test not-a-case >/dev/null 2>&1; then
-        echo "self-test failed: invalid test selection was accepted" >&2
-        return 1
-    fi
-    if "$0" --plan-only --test p4 --soak-seconds 10 >/dev/null 2>&1; then
-        echo "self-test failed: short P4 soak was accepted as release evidence" >&2
-        return 1
-    fi
-    mkdir -p "$tmp/nonempty/g07-mrf-responsibility"
-    : >"$tmp/nonempty/g07-mrf-responsibility/existing.json"
-    if "$0" --dry-run --run-dir "$tmp/nonempty" >/dev/null 2>&1; then
-        echo "self-test failed: non-empty evidence directory was accepted" >&2
-        return 1
-    fi
-}
-
-while [[ $# -gt 0 ]]; do
-    case "$1" in
-        --run-dir|--out-dir)
-            require_value "$1" "$#"
-            RUN_DIR="$2"
-            shift 2
-            ;;
-        --test)
-            require_value "$1" "$#"
-            TEST_SELECTION="$2"
-            shift 2
-            ;;
-        --soak-seconds)
-            require_value "$1" "$#"
-            SOAK_SECONDS="$2"
-            shift 2
-            ;;
-        --enospc-root)
-            require_value "$1" "$#"
-            ENOSPC_ROOT="$2"
-            shift 2
-            ;;
-        --allow-short-soak)
-            ALLOW_SHORT_SOAK=1
-            shift
-            ;;
-        --allow-dirty)
-            ALLOW_DIRTY=1
-            shift
-            ;;
-        --skip-build)
-            SKIP_BUILD=1
-            shift
-            ;;
-        --plan-only|--dry-run)
-            PLAN_ONLY=1
-            shift
-            ;;
-        --self-test)
-            run_self_test
-            exit $?
-            ;;
-        --verbose)
-            VERBOSE=1
-            shift
-            ;;
-        -h|--help)
-            usage
-            exit 0
-            ;;
-        *)
-            echo "unknown option: $1" >&2
-            usage >&2
-            exit 2
-            ;;
-    esac
-done
-
-validate_test_selection
-[[ "$SOAK_SECONDS" =~ ^[0-9]+$ ]] || die "--soak-seconds must be a non-negative integer"
-CASES=()
-while IFS= read -r case_name; do
-    CASES+=("$case_name")
-done < <(case_names)
-if [[ " ${CASES[*]} " == *" p4 "* && "$SOAK_SECONDS" -lt 7200 && "$ALLOW_SHORT_SOAK" != 1 ]]; then
-    die "P4 release evidence requires at least 7200 soak seconds; pass --allow-short-soak only for diagnostics"
-fi
-if [[ -z "$RUN_DIR" ]]; then
-    OUTPUT_ROOT="$(normalize_path "${RUSTFS_SCANNER_HEAL_W13_OUTPUT_ROOT:-$ROOT/target/scanner-heal-w13-evidence}")"
-    RUN_DIR="$OUTPUT_ROOT/$(date -u +%Y%m%dT%H%M%SZ)"
-else
-    RUN_DIR="$(normalize_path "$RUN_DIR")"
-fi
-
-for case_name in "${CASES[@]}"; do
-    check_empty_case_dir "$RUN_DIR/$(artifact_dir_for "$case_name")"
-done
-
-if [[ "$PLAN_ONLY" == 1 ]]; then
-    echo "run_dir=$RUN_DIR"
-    echo "out_dir=$RUN_DIR"
-    echo "tests=${CASES[*]}"
-    echo "soak_seconds=$SOAK_SECONDS"
-    echo "min_free_kib=$MIN_FREE_KIB"
-    target_dir="$(cargo_target_dir)"
-    echo "target_dir=$target_dir"
-    echo "current_binary=$target_dir/debug/rustfs"
-    echo "test_filter=rustfs-heal heal::mrf_queue::tests::w13_mrf_release_evidence_outputs_bundle_artifacts"
-    echo "required_artifacts:"
-    if [[ " ${CASES[*]} " == *" g07 "* ]]; then
-        echo "  $RUN_DIR/g07-mrf-responsibility/G07-mrf_responsibility_oracle.json"
-        echo "  $RUN_DIR/g07-mrf-responsibility/G07-commit_boundary_crash_matrix.json"
-    fi
-    if [[ " ${CASES[*]} " == *" g08 "* ]]; then
-        echo "  $RUN_DIR/g08-mrf-capacity/G08-mrf_capacity_evidence.json"
-        echo "  $RUN_DIR/g08-mrf-capacity/G08-disk_full_matrix.json"
-        echo "  $RUN_DIR/g08-mrf-capacity/G08-replica_loss_matrix.json"
-        if [[ -n "$ENOSPC_ROOT" ]]; then
-            echo "enospc_root=$(normalize_path "$ENOSPC_ROOT")"
-        elif [[ "$(uname -s)" == "Linux" ]]; then
-            echo "enospc_root=$RUN_DIR/enospc-root"
-            echo "enospc_tmpfs_size=$ENOSPC_TMPFS_SIZE"
-        else
-            echo "enospc_root=required-for-non-linux"
-        fi
-    fi
-    if [[ " ${CASES[*]} " == *" p4 "* ]]; then
-        echo "  $RUN_DIR/p4-mrf-soak/P4-mrf_scale_measurement.json"
-        echo "  $RUN_DIR/p4-mrf-soak/P4-mrf_replay_cost_measurement.json"
-        echo "  $RUN_DIR/p4-mrf-soak/P4-retained_responsibility_evidence.json"
-        echo "  $RUN_DIR/p4-mrf-soak/P4-mrf_cleanup_gc_soak_evidence.json"
-    fi
-    exit 0
-fi
-
-cd "$ROOT"
-if [[ "$ALLOW_DIRTY" != 1 && -n "$(git status --porcelain --untracked-files=no)" ]]; then
-    echo "commit tracked source changes before creating release evidence, or pass --allow-dirty for local diagnostics" >&2
-    exit 1
-fi
-if [[ -e "$RUN_DIR" ]]; then
-    die "evidence run directory already exists: $RUN_DIR"
-fi
-mkdir -p "$RUN_DIR/logs"
-trap cleanup_enospc_root EXIT
-if [[ -n "${TMPDIR:-}" ]]; then
-    mkdir -p "$TMPDIR"
-    ensure_min_free_space "$TMPDIR"
-fi
-ensure_min_free_space "$RUN_DIR"
-prepare_enospc_root
-
-SOURCE_REVISION="$(git rev-parse HEAD)"
-printf '%s\n' "$SOURCE_REVISION" >"$RUN_DIR/source-revision.txt"
-
-if [[ "$SKIP_BUILD" != 1 ]]; then
-    run_logged build-current cargo build --locked -p rustfs --bin rustfs
-    write_rustfs_features_stamp
-fi
-
-selection_csv="$(IFS=,; echo "${CASES[*]}")"
-run_logged w13-mrf-evidence env \
-    RUSTFS_SCANNER_HEAL_W13_EVIDENCE_DIR="$RUN_DIR" \
-    RUSTFS_SCANNER_HEAL_W13_SOURCE_REVISION="$SOURCE_REVISION" \
-    RUSTFS_SCANNER_HEAL_W13_SELECTION="$selection_csv" \
-    RUSTFS_SCANNER_HEAL_W13_SOAK_SECONDS="$SOAK_SECONDS" \
-    RUSTFS_SCANNER_HEAL_W13_ALLOW_SHORT_SOAK="$ALLOW_SHORT_SOAK" \
-    RUSTFS_SCANNER_HEAL_W13_RUN_ID="w13-mrf-release-evidence-run" \
-    RUSTFS_SCANNER_HEAL_W13_WINDOW_ID="w13-mrf-release-evidence-window" \
-    RUSTFS_SCANNER_HEAL_W13_ENOSPC_ROOT="$ENOSPC_ROOT" \
-    RUSTFS_SCANNER_HEAL_W13_ENOSPC_FILL_LIMIT_BYTES="${RUSTFS_SCANNER_HEAL_W13_ENOSPC_FILL_LIMIT_BYTES:-67108864}" \
-    cargo test --locked -p rustfs-heal --lib heal::mrf_queue::tests::w13_mrf_release_evidence_outputs_bundle_artifacts \
-        -- --ignored --exact --nocapture
-
-validate_artifacts "$SOURCE_REVISION"
-DESCRIPTOR="$(write_release_descriptor "$SOURCE_REVISION")"
-if [[ " ${CASES[*]} " == *" g07 "* ]]; then
-    check_release_gate "$DESCRIPTOR" G07
-fi
-if [[ " ${CASES[*]} " == *" g08 "* ]]; then
-    check_release_gate "$DESCRIPTOR" G08
-fi
-if [[ " ${CASES[*]} " == *" p4 "* ]]; then
-    if [[ "$ALLOW_SHORT_SOAK" == 1 && "$SOAK_SECONDS" -lt 7200 ]]; then
-        echo "SKIP: P4 release bundle gate validation for diagnostic short soak"
-    else
-        check_release_gate "$DESCRIPTOR" P4
-    fi
-fi
-echo "Scanner/Heal W13 MRF release descriptors verified: $DESCRIPTOR"
-echo "Scanner/Heal W13 MRF evidence verified: $RUN_DIR"

scripts/s3-tests/run.sh

@@ -239,8 +239,6 @@ DEPLOY_MODE="${DEPLOY_MODE:-build}"
 RUSTFS_BINARY="${RUSTFS_BINARY:-}"
 NO_CACHE="${NO_CACHE:-false}"
 S3TESTS_LOCAL_SSE_MASTER_KEY_DEFAULT="MDEyMzQ1Njc4OWFiY2RlZjAxMjM0NTY3ODlhYmNkZWY="
-S3TESTS_ENABLE_LOCAL_KMS="${S3TESTS_ENABLE_LOCAL_KMS:-true}"
-S3_KMS_KEY_ID="${S3_KMS_KEY_ID:-rustfs-s3tests-default-key}"
 
 # Additional directories (SCRIPT_DIR and PROJECT_ROOT defined earlier)
 ARTIFACTS_DIR="${PROJECT_ROOT}/artifacts/s3tests-${TEST_MODE}"
@@ -254,9 +252,6 @@ else
 fi
 DATA_DIR="${DATA_BASE}/test-data/${CONTAINER_NAME}"
 RUSTFS_PID=""
-RUSTFS_KMS_ARGS=()
-S3TESTS_KMS_HOST_KEY_DIR="${S3TESTS_KMS_KEY_DIR:-${DATA_BASE}/kms-keys/${CONTAINER_NAME}}"
-S3TESTS_KMS_RUNTIME_KEY_DIR="${S3TESTS_KMS_HOST_KEY_DIR}"
 
 if [ "${DEPLOY_MODE}" != "existing" ] && [ -z "${RUSTFS_SSE_S3_MASTER_KEY:-}" ]; then
     export RUSTFS_SSE_S3_MASTER_KEY="${S3TESTS_LOCAL_SSE_MASTER_KEY_DEFAULT}"
@@ -287,9 +282,6 @@ Environment Variables:
   S3_ALT_ACCESS_KEY      - Alt user access key (default: rustfsalt)
   S3_ALT_SECRET_KEY      - Alt user secret key (default: rustfsalt)
   RUSTFS_SSE_S3_MASTER_KEY - Optional base64 32-byte key for local managed SSE fallback
-  S3TESTS_ENABLE_LOCAL_KMS - Enable local KMS for SSE-KMS cases (default: true)
-  S3_KMS_KEY_ID          - s3-tests KMS key id (default: rustfs-s3tests-default-key)
-  S3TESTS_KMS_KEY_DIR    - Host key directory for local KMS (default: DATA_ROOT/kms-keys)
   RUSTFS_SCANNER_ENABLED - Enable background scanner for harness service (default: false)
   MAXFAIL                - Stop after N failures, 0 = never stop (default: 1)
   XDIST                  - Enable parallel execution with N workers (default: 0)
@@ -353,52 +345,6 @@ cleanup() {
 
 trap cleanup EXIT
 
-prepare_s3tests_local_kms() {
-    if [ "${S3TESTS_ENABLE_LOCAL_KMS}" != "true" ]; then
-        return 0
-    fi
-    if [ "${DEPLOY_MODE}" = "existing" ]; then
-        log_warn "Skipping local KMS setup for DEPLOY_MODE=existing; set S3_KMS_KEY_ID only when the target service is KMS-enabled"
-        return 0
-    fi
-    if [ "${DEPLOY_MODE}" = "docker" ] && [ -z "${S3TESTS_KMS_KEY_DIR:-}" ]; then
-        S3TESTS_KMS_HOST_KEY_DIR="/tmp/${CONTAINER_NAME}/kms-keys"
-        S3TESTS_KMS_RUNTIME_KEY_DIR="/data/kms-keys"
-    fi
-
-    mkdir -p "${S3TESTS_KMS_HOST_KEY_DIR}"
-    cat > "${S3TESTS_KMS_HOST_KEY_DIR}/${S3_KMS_KEY_ID}.key" <<EOF
-{
-  "key_id": "${S3_KMS_KEY_ID}",
-  "version": 1,
-  "algorithm": "AES_256",
-  "usage": "EncryptDecrypt",
-  "status": "Active",
-  "metadata": {},
-  "created_at": "2026-01-01T00:00:00+00:00[UTC]",
-  "rotated_at": null,
-  "created_by": "s3-tests",
-  "encrypted_key_material": "${S3TESTS_LOCAL_SSE_MASTER_KEY_DEFAULT}",
-  "nonce": [],
-  "at_rest_protection": "plaintext-dev-only"
-}
-EOF
-    chmod 700 "${S3TESTS_KMS_HOST_KEY_DIR}"
-    chmod 600 "${S3TESTS_KMS_HOST_KEY_DIR}/${S3_KMS_KEY_ID}.key"
-    export RUSTFS_KMS_ALLOW_INSECURE_DEV_DEFAULTS="true"
-    export RUSTFS_KMS_ENABLE="true"
-    export RUSTFS_KMS_BACKEND="local"
-    export RUSTFS_KMS_KEY_DIR="${S3TESTS_KMS_RUNTIME_KEY_DIR}"
-    export RUSTFS_KMS_DEFAULT_KEY_ID="${S3_KMS_KEY_ID}"
-    RUSTFS_KMS_ARGS=(
-        --kms-enable
-        --kms-backend local
-        --kms-key-dir "${S3TESTS_KMS_RUNTIME_KEY_DIR}"
-        --kms-default-key-id "${S3_KMS_KEY_ID}"
-    )
-    log_info "Using local KMS key '${S3_KMS_KEY_ID}' for the s3-tests harness"
-}
-
 # Parse command line arguments
 while [[ $# -gt 0 ]]; do
     case $1 in
@@ -429,8 +375,6 @@ if [ "${DEPLOY_MODE}" != "existing" ]; then
     fi
 fi
 
-prepare_s3tests_local_kms
-
 # Start RustFS based on deployment mode
 if [ "${DEPLOY_MODE}" = "existing" ]; then
     log_info "Using existing RustFS service at ${S3_HOST}:${S3_PORT}"
@@ -464,7 +408,6 @@ elif [ "${DEPLOY_MODE}" = "binary" ]; then
         --address "${S3_HOST}:${S3_PORT}" \
         --access-key "${S3_ACCESS_KEY}" \
         --secret-key "${S3_SECRET_KEY}" \
-        "${RUSTFS_KMS_ARGS[@]}" \
         "${DATA_DIR}/rustfs0" "${DATA_DIR}/rustfs1" "${DATA_DIR}/rustfs2" "${DATA_DIR}/rustfs3" \
         > "${ARTIFACTS_DIR}/rustfs-${TEST_MODE}/rustfs.log" 2>&1 &
 
@@ -529,7 +472,6 @@ elif [ "${DEPLOY_MODE}" = "build" ]; then
         --address "${S3_HOST}:${S3_PORT}" \
         --access-key "${S3_ACCESS_KEY}" \
         --secret-key "${S3_SECRET_KEY}" \
-        "${RUSTFS_KMS_ARGS[@]}" \
         "${DATA_DIR}/rustfs0" "${DATA_DIR}/rustfs1" "${DATA_DIR}/rustfs2" "${DATA_DIR}/rustfs3" \
         > "${ARTIFACTS_DIR}/rustfs-${TEST_MODE}/rustfs.log" 2>&1 &
 
@@ -564,11 +506,6 @@ elif [ "${DEPLOY_MODE}" = "docker" ]; then
         -e RUSTFS_ACCESS_KEY="${S3_ACCESS_KEY}" \
         -e RUSTFS_SECRET_KEY="${S3_SECRET_KEY}" \
         -e RUSTFS_SSE_S3_MASTER_KEY="${RUSTFS_SSE_S3_MASTER_KEY}" \
-        -e RUSTFS_KMS_ALLOW_INSECURE_DEV_DEFAULTS="${RUSTFS_KMS_ALLOW_INSECURE_DEV_DEFAULTS:-false}" \
-        -e RUSTFS_KMS_ENABLE="${RUSTFS_KMS_ENABLE:-false}" \
-        -e RUSTFS_KMS_BACKEND="${RUSTFS_KMS_BACKEND:-local}" \
-        -e RUSTFS_KMS_KEY_DIR="${RUSTFS_KMS_KEY_DIR:-}" \
-        -e RUSTFS_KMS_DEFAULT_KEY_ID="${RUSTFS_KMS_DEFAULT_KEY_ID:-}" \
         -e RUSTFS_SCANNER_ENABLED="${RUSTFS_SCANNER_ENABLED}" \
         -e RUSTFS_SCANNER_START_DELAY_SECS="${RUSTFS_SCANNER_START_DELAY_SECS}" \
         -e RUSTFS_SCANNER_CYCLE="${RUSTFS_SCANNER_CYCLE}" \
@@ -824,11 +761,6 @@ envsubst < "${TEMPLATE_PATH}" > "${CONF_OUTPUT_PATH}" || {
     log_error "Failed to generate s3tests config"
     exit 1
 }
-if [ -n "${S3_KMS_KEY_ID:-}" ]; then
-    tmp_conf="${CONF_OUTPUT_PATH}.tmp"
-    sed "s|^#kms_keyid = .*$|kms_keyid = ${S3_KMS_KEY_ID}|" "${CONF_OUTPUT_PATH}" > "${tmp_conf}"
-    mv "${tmp_conf}" "${CONF_OUTPUT_PATH}"
-fi
 
 # Step 7: Provision s3-tests alt user
 # Note: Main user (rustfsadmin) is a system user and doesn't need to be created via API

scripts/test_scanner_heal_w13_mrf_evidence.sh

@@ -1,57 +0,0 @@
-#!/usr/bin/env bash
-set -euo pipefail
-
-SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
-PROJECT_ROOT="$(cd "${SCRIPT_DIR}/.." && pwd)"
-RUNNER="${PROJECT_ROOT}/scripts/run_scanner_heal_w13_mrf_evidence.sh"
-TMP_DIR="$(mktemp -d)"
-
-cleanup() {
-  rm -rf "$TMP_DIR"
-}
-trap cleanup EXIT
-
-bash -n "$RUNNER"
-
-bash "$RUNNER" --help >"$TMP_DIR/help.out"
-rg -q "RUSTFS_SCANNER_HEAL_W13_OUTPUT_ROOT" "$TMP_DIR/help.out"
-rg -q "RUSTFS_SCANNER_HEAL_W13_ENOSPC_ROOT" "$TMP_DIR/help.out"
-rg -q "G07-mrf_responsibility_oracle.json" "$TMP_DIR/help.out"
-rg -q "G08-disk_full_matrix.json" "$TMP_DIR/help.out"
-rg -q "P4-mrf_cleanup_gc_soak_evidence.json" "$TMP_DIR/help.out"
-
-env -u CARGO_TARGET_DIR bash "$RUNNER" \
-  --dry-run \
-  --out-dir "$TMP_DIR/evidence" >"$TMP_DIR/dry-run.out"
-
-rg -q "tests=g07 g08 p4" "$TMP_DIR/dry-run.out"
-rg -q "test_filter=rustfs-heal heal::mrf_queue::tests::w13_mrf_release_evidence_outputs_bundle_artifacts" "$TMP_DIR/dry-run.out"
-rg -q "target_dir=$PROJECT_ROOT/target" "$TMP_DIR/dry-run.out"
-
-RUSTFS_SCANNER_HEAL_W13_OUTPUT_ROOT="$TMP_DIR/root-out" \
-  bash "$RUNNER" --dry-run --test g07 >"$TMP_DIR/dry-run-output-root.out"
-rg -q "run_dir=$TMP_DIR/root-out/" "$TMP_DIR/dry-run-output-root.out"
-
-CARGO_TARGET_DIR="$TMP_DIR/shared-target" bash "$RUNNER" \
-  --dry-run \
-  --out-dir "$TMP_DIR/evidence-with-target" \
-  --test g08 \
-  --enospc-root "$TMP_DIR/enospc" >"$TMP_DIR/dry-run-target.out"
-rg -q "target_dir=$TMP_DIR/shared-target" "$TMP_DIR/dry-run-target.out"
-rg -q "current_binary=$TMP_DIR/shared-target/debug/rustfs" "$TMP_DIR/dry-run-target.out"
-rg -q "enospc_root=$TMP_DIR/enospc" "$TMP_DIR/dry-run-target.out"
-
-mkdir -p "$TMP_DIR/nonempty/g07-mrf-responsibility"
-touch "$TMP_DIR/nonempty/g07-mrf-responsibility/existing.json"
-if bash "$RUNNER" --dry-run --out-dir "$TMP_DIR/nonempty" >"$TMP_DIR/nonempty.out" 2>"$TMP_DIR/nonempty.err"; then
-  echo "W13 runner should reject non-empty evidence case directories" >&2
-  exit 1
-fi
-rg -q "evidence case directory is not empty" "$TMP_DIR/nonempty.err"
-
-if bash "$RUNNER" --plan-only --test p4 --soak-seconds 10 >/dev/null 2>&1; then
-  echo "W13 runner should reject short P4 release soak without --allow-short-soak" >&2
-  exit 1
-fi
-
-bash "$RUNNER" --self-test