test(scanner): require real mixed-version revisions

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

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

.config/e2e-full-selection.txt

@@ -1,2 +1,2 @@
-sha256-darwin=cca6d0bc1487f472dc354bbffcc2b5c7410edfccb499c6bce6638113fbe6bbec
-sha256-linux=3b71936f6f4ea0cca3b5db6c2f387c990dddda7e96c982315eb42e3b2e6c92b8
+sha256-darwin=f0c78fdb93471575d9a64c5c46eae6c806bdd0bc10a6e33d7fb574aabd8db5a3
+sha256-linux=03ed7016cab672de9320e31375a0358eceacb4408b0e79cf063614fa7c878b87

.config/e2e-nightly-selection.txt

@@ -1,2 +1,2 @@
-sha256-darwin=83a7dcaffd5a789517ae9f02a224f66a9713937885cff96fca2ad7e216f197ae
-sha256-linux=626c10f8c964507ff987b6c86069e9019dc6d2ae7fb02db9be5df5aa8cc5145b
+sha256-darwin=a5665318c9bdc0947514fb7008ba1b83b114b739fac775c3c446f207058b7c7a
+sha256-linux=45d80e1723de5d25bb5b81f3ef5c82f583efc3e4f036a8cd2bb99e4f1eca9e51

.config/e2e-repl-nightly-selection.txt

@@ -1 +1 @@
-sha256=0e338d305260229e17ccfb2adc48a6212dbdfea36a9ebfb5a4e0d38658e6cc45
+sha256=0fe8408874ccec3620262a9812d67920ddd72dc9edf0e36e0d0aed3f8bad026e

.config/scanner-heal-required-tests.json

@@ -43,6 +43,8 @@
       "min_objects": 5,
       "max_objects": 5,
       "topology": {"nodes": 3, "drives_per_node": 4},
+      "erasure": {"data_blocks": 8, "parity_blocks": 4},
+      "erasure_set_drive_count": 12,
       "scope": "3-node x 4-drive single-set EC8+4, graceful target restart, preformatted replacement drive, exact unversioned S3 bodies and physical target shards; not mixed-version, multi-pool or long-window ABBA."
     },
     "background-target-restart-ec8-4": {

.github/workflows/audit.yml

@@ -16,7 +16,7 @@ name: Security Audit
 
 on:
   push:
-    branches: [ main, release ]
+    branches: [ main ]
     paths:
       - '**/Cargo.toml'
       - '**/Cargo.lock'
@@ -32,7 +32,7 @@ on:
       - 'scripts/security/check_workflow_pins.sh'
   pull_request:
     types: [ opened, synchronize, reopened, closed ]
-    branches: [ main, release ]
+    branches: [ main ]
     paths:
       - '**/Cargo.toml'
       - '**/Cargo.lock'

.github/workflows/e2e-distributed.yml

@@ -19,7 +19,7 @@
 # case is a two-site 4-node 1-drive pair or a 4-node upgrade). Membership is
 # `[profile.e2e-distributed]` in `.config/nextest.toml`. Storage-sensitive PRs,
 # nightly runs, and manual dispatches all execute the same fail-closed suite.
-# Upgrade cases use an independent 1.0.0-rc.2 pin defined below.
+# Upgrade cases download the same pinned previous release as e2e-upgrade.yml.
 #
 # Isolated pool filesystems: expand/decommission/rebalance cases require
 # independent `statfs` capacity. This job runs on GitHub-hosted
@@ -87,7 +87,7 @@ jobs:
       NO_PROXY: 127.0.0.1,localhost
       HTTP_PROXY: ""
       HTTPS_PROXY: ""
-      # Independent 1.0.0-rc.2 source pin for distributed::upgrade_test.
+      # Pinned previous release used by distributed::upgrade_test (same pin as e2e-upgrade.yml).
       UPGRADE_SOURCE_VERSION: 1.0.0-rc.2
       UPGRADE_SOURCE_ASSET: rustfs-linux-x86_64-gnu-v1.0.0-rc.2.zip
       UPGRADE_SOURCE_SHA256: 7c789386bf85278f865b8e0d359bf4edb84d5aa408cc3fa54a18c25ca74cd6e7

crates/e2e_test/src/chaos.rs

@@ -44,7 +44,6 @@ use sha2::{Digest, Sha256};
 use std::collections::{BTreeMap, BTreeSet};
 use std::error::Error;
 use std::path::{Path, PathBuf};
-use tokio::time::{Duration, Instant, sleep};
 use tracing::info;
 use uuid::Uuid;
 use walkdir::WalkDir;
@@ -375,33 +374,6 @@ pub(crate) fn census_object_version_on_disk(
     })
 }
 
-/// Wait for the background PUT tail to commit every physical part on one disk.
-/// Invalid metadata remains an immediate error instead of a retryable absence.
-pub(crate) async fn wait_for_complete_physical_shard_on_disk(
-    disk: &Path,
-    bucket: &str,
-    key: &str,
-    version_id: Option<&str>,
-    timeout: Duration,
-) -> ChaosResult<VersionShardCensus> {
-    let deadline = Instant::now() + timeout;
-    loop {
-        let census = census_object_version_on_disk(disk, bucket, key, version_id)?;
-        if census.is_complete() && !census.expected_part_numbers.is_empty() {
-            return Ok(census);
-        }
-        let remaining = deadline.saturating_duration_since(Instant::now());
-        if remaining.is_zero() {
-            return Err(format!(
-                "physical shard for {bucket}/{key}@{version_id:?} on {} did not become complete within {timeout:?}: {census:?}",
-                disk.display()
-            )
-            .into());
-        }
-        sleep(remaining.min(Duration::from_millis(50))).await;
-    }
-}
-
 /// `POST` a signed (SigV4, service `s3`) admin request without relying on the
 /// external `awscurl` binary. Mirrors the admin heal calls used by the heal
 /// regression suite.
@@ -479,30 +451,4 @@ mod tests {
         assert!(expected.matches_manifest(&expected));
         assert!(!changed.matches_manifest(&expected));
     }
-
-    #[tokio::test]
-    async fn physical_shard_readiness_fails_closed_with_last_census() {
-        let disk = tempfile::tempdir().expect("temporary disk");
-        let error = wait_for_complete_physical_shard_on_disk(disk.path(), "bucket", "missing", None, Duration::ZERO)
-            .await
-            .expect_err("missing physical shards must fail the baseline gate");
-        assert!(error.to_string().contains("has_xl_meta: false"));
-        assert!(error.to_string().contains("bucket/missing"));
-    }
-
-    #[tokio::test]
-    async fn physical_shard_readiness_does_not_retry_invalid_metadata() {
-        let disk = tempfile::tempdir().expect("temporary disk");
-        let object = disk.path().join("bucket").join("corrupt");
-        std::fs::create_dir_all(&object).expect("object directory");
-        std::fs::write(object.join("xl.meta"), b"invalid metadata").expect("corrupt metadata fixture");
-        let error = tokio::time::timeout(
-            Duration::from_secs(1),
-            wait_for_complete_physical_shard_on_disk(disk.path(), "bucket", "corrupt", None, Duration::from_secs(30)),
-        )
-        .await
-        .expect("corrupt metadata must fail immediately")
-        .expect_err("invalid metadata must not be accepted as a complete baseline");
-        assert!(!error.to_string().contains("did not become complete"));
-    }
 }

crates/e2e_test/src/distributed/heal_test.rs

@@ -12,15 +12,18 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use super::harness::{DistCluster, DistLayout, TestResult, assert_inventory, payload_for, put_object, unique_bucket, wait_until};
-use crate::chaos::{
-    VersionShardCensus, census_object_version_on_disk, sha256_hex, signed_admin_post, wait_for_complete_physical_shard_on_disk,
+use super::harness::{
+    DistCluster, DistLayout, TestResult, assert_inventory, get_object_bytes, payload_for, put_object, sha256_hex, unique_bucket,
+    wait_until,
 };
-use crate::common::{init_logging, rustfs_binary_path};
-use crate::scanner_heal_evidence::{EvidenceTopology, RestartObservation, ScannerHealEvidenceCase, restart_evidence_run};
+use crate::chaos::{VersionShardCensus, census_object_version_on_disk, signed_admin_post};
+use crate::common::init_logging;
 use aws_sdk_s3::Client;
 use aws_sdk_s3::primitives::ByteStream;
+use serde_json::Value;
+use sha2::{Digest, Sha256};
 use std::collections::{BTreeMap, HashSet};
+use std::io::{Read, Write};
 use std::path::{Path, PathBuf};
 use std::time::Duration;
 
@@ -28,6 +31,8 @@ const EC84_NODE_COUNT: usize = 3;
 const EC84_DRIVES_PER_NODE: usize = 4;
 const EC84_DATA_BLOCKS: usize = 8;
 const EC84_PARITY_BLOCKS: usize = 4;
+const EC84_TARGET_DRIVE_RESTART_CASE: &str = "ec84-target-drive-restart";
+const EC84_TARGET_DRIVE_RESTART_ORACLE: &str = "ec84-target-drive-restart.json";
 
 #[derive(Clone)]
 struct ExpectedShard {
@@ -36,6 +41,79 @@ struct ExpectedShard {
     baseline: VersionShardCensus,
 }
 
+struct ScannerHealEvidenceContext {
+    directory: PathBuf,
+    run: Value,
+}
+
+fn file_sha256(path: &Path) -> TestResult<String> {
+    let mut file = std::fs::File::open(path)?;
+    let mut digest = Sha256::new();
+    let mut buffer = [0_u8; 64 * 1024];
+    loop {
+        let read = file.read(&mut buffer)?;
+        if read == 0 {
+            break;
+        }
+        digest.update(&buffer[..read]);
+    }
+    Ok(digest.finalize().iter().map(|byte| format!("{byte:02x}")).collect())
+}
+
+fn compiled_test_identity() -> Value {
+    serde_json::json!({
+        "source_revision": env!("RUSTFS_E2E_BUILD_COMMIT"),
+        "dirty": env!("RUSTFS_E2E_BUILD_DIRTY") != "false",
+        "lock_blob": env!("RUSTFS_E2E_BUILD_LOCK"),
+        "features": env!("RUSTFS_E2E_BUILD_FEATURES"),
+        "target": env!("RUSTFS_E2E_BUILD_TARGET"),
+        "profile": env!("RUSTFS_E2E_BUILD_PROFILE"),
+        "rustflags_hex": env!("RUSTFS_E2E_BUILD_RUSTFLAGS_HEX"),
+    })
+}
+
+fn string_field<'a>(value: &'a Value, path: &str) -> TestResult<&'a str> {
+    let mut current = value;
+    for segment in path.split('.') {
+        current = current
+            .get(segment)
+            .ok_or_else(|| format!("scanner/heal run receipt missing {path}"))?;
+    }
+    current
+        .as_str()
+        .filter(|text| !text.is_empty())
+        .ok_or_else(|| format!("scanner/heal run receipt has invalid {path}").into())
+}
+
+fn scanner_heal_evidence_context() -> TestResult<Option<ScannerHealEvidenceContext>> {
+    let Some(directory) = std::env::var_os("RUSTFS_SCANNER_HEAL_RUN_DIR") else {
+        return Ok(None);
+    };
+    let directory = PathBuf::from(directory);
+    let receipt = directory.join("run.json");
+    if receipt.metadata()?.len() > 1024 * 1024 {
+        return Err("oversized scanner/heal execution receipt".into());
+    }
+    let run: Value = serde_json::from_slice(&std::fs::read(receipt)?)?;
+    let built = compiled_test_identity();
+    for key in ["source_revision", "dirty", "lock_blob", "features"] {
+        if built[key] != run["test_build"][key] {
+            return Err(format!("compiled test identity differs for {key}").into());
+        }
+    }
+    let binary_path = PathBuf::from(string_field(&run, "binary.path")?);
+    if file_sha256(&binary_path)? != string_field(&run, "binary.sha256")? {
+        return Err("server binary must match the run receipt".into());
+    }
+    if file_sha256(&std::env::current_exe()?)? != string_field(&run, "test_binary.sha256")? {
+        return Err("test executable must match the run receipt".into());
+    }
+    if directory.join(EC84_TARGET_DRIVE_RESTART_ORACLE).exists() {
+        return Err("scanner/heal oracle already exists; create a new execution receipt".into());
+    }
+    Ok(Some(ScannerHealEvidenceContext { directory, run }))
+}
+
 fn assert_ec84_geometry(census: &VersionShardCensus, key: &str) -> TestResult {
     if census.data_blocks != Some(EC84_DATA_BLOCKS) || census.parity_blocks != Some(EC84_PARITY_BLOCKS) {
         return Err(format!("object {key} did not use EC8+4 geometry: {census:?}").into());
@@ -52,6 +130,76 @@ fn assert_ec84_geometry(census: &VersionShardCensus, key: &str) -> TestResult {
     Ok(())
 }
 
+async fn write_scanner_heal_evidence(
+    context: ScannerHealEvidenceContext,
+    dist: &DistCluster,
+    bucket: &str,
+    expected: &[ExpectedShard],
+    outage_key: &str,
+    outage_body: &[u8],
+    replaced_drive: &Path,
+    pid_before: u32,
+    pid_after: u32,
+    node_listings: Vec<Vec<String>>,
+) -> TestResult {
+    let verifier = dist.client(0)?;
+    let mut objects = Vec::new();
+    for item in expected {
+        let actual = get_object_bytes(&verifier, bucket, &item.key).await?;
+        let physical = census_object_version_on_disk(replaced_drive, bucket, &item.key, None)?;
+        objects.push(serde_json::json!({
+            "key": item.key,
+            "version_id": null,
+            "expected_bytes": item.body.len(),
+            "actual_bytes": actual.len(),
+            "expected_sha256": sha256_hex(&item.body),
+            "actual_sha256": sha256_hex(&actual),
+            "expected_physical": item.baseline,
+            "physical": physical,
+        }));
+    }
+    let actual = get_object_bytes(&verifier, bucket, outage_key).await?;
+    let physical = census_object_version_on_disk(replaced_drive, bucket, outage_key, None)?;
+    objects.push(serde_json::json!({
+        "key": outage_key,
+        "version_id": null,
+        "expected_bytes": outage_body.len(),
+        "actual_bytes": actual.len(),
+        "expected_sha256": sha256_hex(outage_body),
+        "actual_sha256": sha256_hex(&actual),
+        "expected_physical": null,
+        "physical": physical,
+    }));
+
+    let evidence = serde_json::json!({
+        "schema": 1,
+        "case": EC84_TARGET_DRIVE_RESTART_CASE,
+        "evidence": "process-restart",
+        "run_id": string_field(&context.run, "run_id")?,
+        "source_revision": string_field(&context.run, "source_revision")?,
+        "test_build": compiled_test_identity(),
+        "binary_sha256": string_field(&context.run, "binary.sha256")?,
+        "test_binary_sha256": string_field(&context.run, "test_binary.sha256")?,
+        "topology": {"nodes": EC84_NODE_COUNT, "drives_per_node": EC84_DRIVES_PER_NODE},
+        "pid_before": pid_before,
+        "pid_after": pid_after,
+        "unclean_shutdown_marker": false,
+        "objects": objects,
+        "node_listings": node_listings,
+    });
+    let data = serde_json::to_vec(&evidence)?;
+    if data.len() > 1024 * 1024 {
+        return Err("scanner/heal oracle exceeds the 1 MiB artifact budget".into());
+    }
+    let mut output = std::fs::OpenOptions::new()
+        .write(true)
+        .create_new(true)
+        .open(context.directory.join(EC84_TARGET_DRIVE_RESTART_ORACLE))?;
+    output.write_all(&data)?;
+    output.sync_all()?;
+    Ok(())
+}
+
 fn assert_replaced_drive_empty(drive: &Path, bucket: &str, keys: &[String]) -> TestResult {
     for key in keys {
         let census = census_object_version_on_disk(drive, bucket, key, None)?;
@@ -90,19 +238,7 @@ async fn put_large_inventory(client: &Client, bucket: &str) -> TestResult<Vec<Ex
 #[tokio::test]
 async fn three_node_four_drive_ec8_4_root_heal_rebuilds_replaced_drive_after_restart() -> TestResult {
     init_logging();
-    let server_binary = rustfs_binary_path();
-    let evidence_run = restart_evidence_run(
-        &server_binary,
-        ScannerHealEvidenceCase {
-            id: "ec84-target-drive-restart",
-            oracle: "ec84-target-drive-restart.json",
-            evidence: "process-restart",
-            unclean_shutdown_marker: false,
-            topology: EvidenceTopology::new(3, 4),
-            storage_class_standard: Some("EC:4"),
-            erasure_set_drive_count: Some("12"),
-        },
-    )?;
+    let evidence_context = scanner_heal_evidence_context()?;
     let mut dist = DistCluster::start_with_env(
         DistLayout::ThreeByFourEc84,
         &[
@@ -126,24 +262,18 @@ async fn three_node_four_drive_ec8_4_root_heal_rebuilds_replaced_drive_after_res
     let replaced_drive = PathBuf::from(&dist.cluster.nodes[replaced_node].data_dirs[replaced_drive_index]);
 
     for item in &mut expected {
-        item.baseline =
-            wait_for_complete_physical_shard_on_disk(&replaced_drive, &bucket, &item.key, None, Duration::from_secs(10)).await?;
+        item.baseline = census_object_version_on_disk(&replaced_drive, &bucket, &item.key, None)?;
         assert_ec84_geometry(&item.baseline, &item.key)?;
     }
 
     let format_path = replaced_drive.join(".rustfs.sys").join("format.json");
     let format_json = std::fs::read(&format_path)?;
-    let pid_before = dist.cluster.nodes[replaced_node]
+    let target_pid_before = dist.cluster.nodes[replaced_node]
         .process
         .as_ref()
-        .ok_or("target process is absent")?
+        .ok_or("target process is absent before graceful restart")?
         .id();
     dist.cluster.stop_node_gracefully(replaced_node).await?;
-    let unclean_shutdown_marker = Path::new(&dist.cluster.nodes[replaced_node].data_dir)
-        .join(".rustfs.sys")
-        .join("unclean-shutdown")
-        .is_file();
-    assert!(!unclean_shutdown_marker, "graceful target shutdown must remove its unclean marker");
     let retired_drive = PathBuf::from(format!("{}.retired", replaced_drive.display()));
     std::fs::rename(&replaced_drive, &retired_drive)?;
     std::fs::create_dir_all(format_path.parent().ok_or("replacement format path has no parent")?)?;
@@ -165,6 +295,11 @@ async fn three_node_four_drive_ec8_4_root_heal_rebuilds_replaced_drive_after_res
         .await?;
 
     dist.cluster.start_node(replaced_node).await?;
+    let target_pid_after = dist.cluster.nodes[replaced_node]
+        .process
+        .as_ref()
+        .ok_or("target process is absent after restart")?
+        .id();
     let heal_body =
         r#"{"recursive":true,"dryRun":false,"remove":false,"recreate":true,"scanMode":2,"updateParity":false,"nolock":false}"#;
     let heal_url = format!("{}/rustfs/admin/v3/heal/{bucket}?forceStart=true", dist.cluster.nodes[0].url);
@@ -205,45 +340,24 @@ async fn three_node_four_drive_ec8_4_root_heal_rebuilds_replaced_drive_after_res
             .filter_map(|object| object.key().map(str::to_owned))
             .collect::<HashSet<_>>();
         assert_eq!(observed, expected_keys, "node {node_index} listing diverged after EC8+4 heal");
-        let mut keys = observed.into_iter().collect::<Vec<_>>();
-        keys.sort();
-        node_listings.push(keys);
+        let mut observed = observed.into_iter().collect::<Vec<_>>();
+        observed.sort();
+        node_listings.push(observed);
     }
-
-    if let Some(evidence_run) = evidence_run {
-        let target_client = dist.client(replaced_node)?;
-        let mut objects = Vec::with_capacity(inventory.len());
-        for (key, body) in &inventory {
-            let response = target_client.get_object().bucket(&bucket).key(key).send().await?;
-            let actual = response.body.collect().await?.into_bytes();
-            assert_eq!(actual.as_ref(), body.as_slice(), "object body changed for {key}");
-            let physical = census_object_version_on_disk(&replaced_drive, &bucket, key, None)?;
-            assert_ec84_geometry(&physical, key)?;
-            let baseline = expected.iter().find(|item| item.key == *key).map(|item| &item.baseline);
-            objects.push(serde_json::json!({
-                "key": key, "version_id": null,
-                "expected_bytes": body.len(), "actual_bytes": actual.len(),
-                "expected_sha256": sha256_hex(body), "actual_sha256": sha256_hex(&actual),
-                "expected_physical": baseline, "physical": physical,
-            }));
-        }
-        let pid_after = dist.cluster.nodes[replaced_node]
-            .process
-            .as_ref()
-            .ok_or("restarted target is absent")?
-            .id();
-        evidence_run.write(
-            &server_binary,
-            RestartObservation {
-                nodes: dist.cluster.nodes.len(),
-                drives_per_node: dist.cluster.topology.drives_per_node,
-                pid_before,
-                pid_after,
-                unclean_shutdown_marker,
-                objects,
-                node_listings,
-            },
-        )?;
+    if let Some(context) = evidence_context {
+        write_scanner_heal_evidence(
+            context,
+            &dist,
+            &bucket,
+            &expected,
+            outage_key,
+            &outage_body,
+            &replaced_drive,
+            target_pid_before,
+            target_pid_after,
+            node_listings,
+        )
+        .await?;
     }
 
     Ok(())

crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs

@@ -16,20 +16,18 @@
 
 #[cfg(test)]
 mod tests {
-    use crate::chaos::{
-        VersionShardCensus, census_object_version_on_disk, sha256_hex, signed_admin_post,
-        wait_for_complete_physical_shard_on_disk,
-    };
+    use crate::chaos::{VersionShardCensus, census_object_version_on_disk, sha256_hex, signed_admin_post};
     use crate::common::{
-        FAST_DATA_USAGE_SCANNER_ENV, RustFSTestClusterEnvironment, RustFSTestEnvironment, admin_request, init_logging,
-        rustfs_binary_path,
+        ClusterTopology, FAST_DATA_USAGE_SCANNER_ENV, RustFSTestClusterEnvironment, RustFSTestEnvironment, admin_request,
+        init_logging, rustfs_binary_path,
     };
-    use crate::scanner_heal_evidence::{EvidenceTopology, RestartObservation, ScannerHealEvidenceCase, restart_evidence_run};
     use crate::storage_api::RUSTFS_META_BUCKET;
     use aws_sdk_s3::primitives::ByteStream;
     use http::Method;
+    use sha2::{Digest, Sha256};
     use std::collections::HashSet;
     use std::error::Error;
+    use std::io::{Read, Write};
     use std::net::SocketAddr;
     use std::path::{Path, PathBuf};
     use std::process::Command;
@@ -41,6 +39,52 @@ mod tests {
 
     const POOL_METADATA_OBJECT: &str = "pool.bin";
 
+    #[derive(serde::Deserialize)]
+    struct EvidenceBuild {
+        sha256: String,
+    }
+
+    #[derive(serde::Deserialize)]
+    struct RestartEvidenceRun {
+        schema: u32,
+        run_id: String,
+        source_revision: String,
+        test_build: serde_json::Value,
+        binary: EvidenceBuild,
+        test_binary: EvidenceBuild,
+    }
+
+    #[derive(Clone, Copy)]
+    struct ScannerHealEvidenceCase {
+        id: &'static str,
+        oracle: &'static str,
+        evidence: &'static str,
+        unclean_shutdown_marker: bool,
+        topology: EvidenceTopology,
+        storage_class_standard: Option<&'static str>,
+        erasure_set_drive_count: Option<&'static str>,
+    }
+
+    #[derive(Clone, Copy)]
+    struct EvidenceTopology {
+        nodes: usize,
+        drives_per_node: usize,
+    }
+
+    impl EvidenceTopology {
+        const fn new(nodes: usize, drives_per_node: usize) -> Self {
+            Self { nodes, drives_per_node }
+        }
+
+        fn total_drives(self) -> usize {
+            self.nodes * self.drives_per_node
+        }
+
+        fn cluster_topology(self) -> ClusterTopology {
+            ClusterTopology::single_pool_multidrive(self.nodes, self.drives_per_node)
+        }
+    }
+
     const BACKGROUND_TARGET_RESTART_EVIDENCE: ScannerHealEvidenceCase = ScannerHealEvidenceCase {
         id: "background-target-restart",
         oracle: "background-target-restart.json",
@@ -81,6 +125,81 @@ mod tests {
         erasure_set_drive_count: Some("12"),
     };
 
+    struct RestartEvidenceContext {
+        directory: PathBuf,
+        run: RestartEvidenceRun,
+        case: ScannerHealEvidenceCase,
+    }
+
+    fn file_sha256(path: &Path) -> Result<String, Box<dyn Error + Send + Sync>> {
+        let mut file = std::fs::File::open(path)?;
+        let mut digest = Sha256::new();
+        let mut buffer = [0_u8; 64 * 1024];
+        loop {
+            let read = file.read(&mut buffer)?;
+            if read == 0 {
+                break;
+            }
+            digest.update(&buffer[..read]);
+        }
+        Ok(digest.finalize().iter().map(|byte| format!("{byte:02x}")).collect())
+    }
+
+    fn restart_evidence_run(
+        binary: &Path,
+        case: ScannerHealEvidenceCase,
+    ) -> Result<Option<RestartEvidenceContext>, Box<dyn Error + Send + Sync>> {
+        let Some(directory) = std::env::var_os("RUSTFS_SCANNER_HEAL_RUN_DIR") else {
+            return Ok(None);
+        };
+        if case.id.is_empty()
+            || case.oracle.is_empty()
+            || !case.oracle.ends_with(".json")
+            || case.oracle.contains('/')
+            || case.oracle.contains('\\')
+            || case.oracle.contains("..")
+            || !matches!(case.evidence, "process-restart" | "process-crash-restart")
+            || (case.evidence == "process-crash-restart") != case.unclean_shutdown_marker
+        {
+            return Err("invalid scanner/heal evidence case".into());
+        }
+        let directory = PathBuf::from(directory);
+        let receipt = directory.join("run.json");
+        if receipt.metadata()?.len() > 1024 * 1024 {
+            return Err("oversized scanner/heal execution receipt".into());
+        }
+        let run: RestartEvidenceRun = serde_json::from_slice(&std::fs::read(receipt)?)?;
+        if run.schema != 1 || run.run_id.len() != 32 || run.source_revision.len() != 40 {
+            return Err("invalid scanner/heal execution identity".into());
+        }
+        let built = compiled_test_identity();
+        for key in ["source_revision", "dirty", "lock_blob", "features"] {
+            assert_eq!(built[key], run.test_build[key], "compiled test identity differs for {key}");
+        }
+        assert_eq!(file_sha256(binary)?, run.binary.sha256, "server binary must match the run receipt");
+        assert_eq!(
+            file_sha256(&std::env::current_exe()?)?,
+            run.test_binary.sha256,
+            "test executable must match the run receipt"
+        );
+        if directory.join(case.oracle).exists() {
+            return Err("scanner/heal oracle already exists; create a new execution receipt".into());
+        }
+        Ok(Some(RestartEvidenceContext { directory, run, case }))
+    }
+
+    fn compiled_test_identity() -> serde_json::Value {
+        serde_json::json!({
+            "source_revision": env!("RUSTFS_E2E_BUILD_COMMIT"),
+            "dirty": env!("RUSTFS_E2E_BUILD_DIRTY") != "false",
+            "lock_blob": env!("RUSTFS_E2E_BUILD_LOCK"),
+            "features": env!("RUSTFS_E2E_BUILD_FEATURES"),
+            "target": env!("RUSTFS_E2E_BUILD_TARGET"),
+            "profile": env!("RUSTFS_E2E_BUILD_PROFILE"),
+            "rustflags_hex": env!("RUSTFS_E2E_BUILD_RUSTFLAGS_HEX"),
+        })
+    }
+
     struct TcpPortBlackhole {
         port: u16,
         comment: String,
@@ -821,10 +940,7 @@ mod tests {
         cluster: &RustFSTestClusterEnvironment,
         previous_cycle_end: u64,
     ) -> Result<u64, Box<dyn Error + Send + Sync>> {
-        let started = Instant::now();
-        let mut deadline = started + Duration::from_secs(60);
-        let catch_up_deadline = deadline + Duration::from_secs(300);
-        let mut catch_up_wait_observed = false;
+        let deadline = Instant::now() + Duration::from_secs(60);
         loop {
             let mut latest_cycle_end = 0;
             let mut versions_observed = false;
@@ -852,61 +968,24 @@ mod tests {
                 let versions_scanned = metrics["versions_scanned"]
                     .as_u64()
                     .ok_or("scanner status is missing its version-coverage counter")?;
-                let cycle_result = metrics["last_cycle_result"]
-                    .as_str()
-                    .ok_or("scanner status is missing its cycle result")?;
-                if cycle_result == "success" {
-                    latest_cycle_end = latest_cycle_end.max(cycle_end);
-                }
+                latest_cycle_end = latest_cycle_end.max(cycle_end);
                 versions_observed |= versions_scanned > 0;
-                let backlog = &status["pause_backlog"];
-                if !catch_up_wait_observed
-                    && backlog["persistence_state"].as_str() == Some("healthy")
-                    && backlog["durable"].as_bool() == Some(true)
-                    && backlog["phase"].as_str() == Some("catching_up")
-                    && backlog["rate_limited"].as_bool() == Some(true)
-                    && backlog["retry_exhausted"].as_bool() == Some(false)
-                {
-                    let next_attempt = backlog["next_attempt_at_unix_secs"]
-                        .as_u64()
-                        .ok_or("rate-limited scanner backlog is missing its next attempt")?;
-                    let interval = backlog["thresholds"]["catch_up_min_interval_seconds"]
-                        .as_u64()
-                        .ok_or("rate-limited scanner backlog is missing its catch-up interval")?;
-                    let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH)?.as_secs();
-                    let remaining = next_attempt.saturating_sub(now);
-                    if remaining > 0 {
-                        if interval > 300 || remaining > interval {
-                            return Err(
-                                format!("scanner catch-up schedule exceeds the bounded recovery budget: {backlog}").into()
-                            );
-                        }
-                        // The durable catch-up interval overrides SCANNER_CYCLE=1.
-                        // Honor one observed retry without restarting the deadline on every poll.
-                        deadline = deadline
-                            .max(Instant::now() + Duration::from_secs(remaining + 60))
-                            .min(catch_up_deadline);
-                        catch_up_wait_observed = true;
-                    }
-                }
                 observations.push(format!(
-                    "node{node_index}: end={cycle_end}, versions={versions_scanned}, cycle={}, active={}, leader={}, result={}, backlog={}",
+                    "node{node_index}: end={cycle_end}, versions={versions_scanned}, cycle={}, active={}, leader={}, result={}",
                     metrics["current_cycle"],
                     metrics["current_cycle_active"],
                     metrics["leader_lock_state"],
                     metrics["last_cycle_result"],
-                    backlog,
                 ));
             }
-            // Only a successful coordinator cycle counts as completion; deferred
-            // and superseded attempts also advance its end timestamp. Remote
-            // workers record version coverage, so the witnesses can span nodes.
+            // The coordinator records cycle completion, but remote workers
+            // record scanned versions. Both witnesses need not share a node.
             if latest_cycle_end > previous_cycle_end && versions_observed {
                 return Ok(latest_cycle_end);
             }
             if Instant::now() >= deadline {
                 return Err(format!(
-                    "enabled scanner did not complete a successful object-scanning cycle after {previous_cycle_end}: {observations:?}"
+                    "enabled scanner did not complete an object-scanning cycle after {previous_cycle_end}: {observations:?}"
                 )
                 .into());
             }
@@ -926,7 +1005,7 @@ mod tests {
     async fn test_cluster_root_heal_recovers_remote_shards_after_background_target_restart()
     -> Result<(), Box<dyn Error + Send + Sync>> {
         timeout(
-            Duration::from_secs(720),
+            Duration::from_secs(420),
             run_cluster_root_heal_interruption(InterruptionScenario::BackgroundTargetRestart),
         )
         .await?
@@ -936,7 +1015,7 @@ mod tests {
     async fn test_cluster_root_heal_recovers_remote_shards_after_background_target_crash()
     -> Result<(), Box<dyn Error + Send + Sync>> {
         timeout(
-            Duration::from_secs(720),
+            Duration::from_secs(420),
             run_cluster_root_heal_interruption(InterruptionScenario::BackgroundTargetCrash),
         )
         .await?
@@ -946,7 +1025,7 @@ mod tests {
     async fn test_cluster_root_heal_recovers_ec84_shards_after_background_target_restart()
     -> Result<(), Box<dyn Error + Send + Sync>> {
         timeout(
-            Duration::from_secs(720),
+            Duration::from_secs(420),
             run_cluster_root_heal_interruption(InterruptionScenario::BackgroundTargetRestartEc84),
         )
         .await?
@@ -956,7 +1035,7 @@ mod tests {
     async fn test_cluster_root_heal_recovers_ec84_shards_after_background_target_crash()
     -> Result<(), Box<dyn Error + Send + Sync>> {
         timeout(
-            Duration::from_secs(720),
+            Duration::from_secs(420),
             run_cluster_root_heal_interruption(InterruptionScenario::BackgroundTargetCrashEc84),
         )
         .await?
@@ -966,7 +1045,7 @@ mod tests {
     async fn test_cluster_root_heal_recovers_remote_shards_after_coordinator_restart() -> Result<(), Box<dyn Error + Send + Sync>>
     {
         timeout(
-            Duration::from_secs(720),
+            Duration::from_secs(420),
             run_cluster_root_heal_interruption(InterruptionScenario::BackgroundCoordinatorRestart),
         )
         .await?
@@ -1073,19 +1152,10 @@ mod tests {
         let server_rust_log = std::env::var("RUSTFS_HEAL_CHAOS_SERVER_RUST_LOG")
             .unwrap_or_else(|_| "rustfs::heal::task=info,rustfs=error".to_string());
         cluster.set_env("RUST_LOG", server_rust_log);
-        let log_dir = if let Some(directory) = std::env::var_os("RUSTFS_HEAL_CHAOS_LOG_DIR") {
-            PathBuf::from(directory)
-        } else if let Some(directory) = std::env::var_os("RUSTFS_E2E_LOG_DIR") {
-            let cluster_name = Path::new(&cluster.temp_dir)
-                .file_name()
-                .ok_or("cluster directory has no name")?;
-            PathBuf::from(directory).join(cluster_name).join("heal")
-        } else {
-            PathBuf::from(&cluster.temp_dir).join("logs")
-        };
+        let log_dir = std::env::var("RUSTFS_HEAL_CHAOS_LOG_DIR").unwrap_or_else(|_| format!("{}/logs", cluster.temp_dir));
         std::fs::create_dir_all(&log_dir)?;
         for node_index in 0..cluster.nodes.len() {
-            cluster.set_node_capture_log_path(node_index, log_dir.join(format!("node{node_index}.log")).to_string_lossy())?;
+            cluster.set_node_capture_log_path(node_index, format!("{log_dir}/node{node_index}.log"))?;
         }
         cluster.start_with_binary(&server_binary).await?;
         let clients = cluster.create_all_clients()?;
@@ -1140,8 +1210,6 @@ mod tests {
                 attempt_count += 1;
                 continue;
             }
-            let shard_census =
-                wait_for_complete_physical_shard_on_disk(&replaced_disk, bucket, &key, None, Duration::from_secs(10)).await?;
             assert!(
                 shard_census.is_complete(),
                 "node 1 should hold a complete baseline shard for {key}: {shard_census:?}"
@@ -1362,7 +1430,7 @@ mod tests {
         let pre_interrupt_status: serde_json::Value = serde_json::from_str(&pre_interrupt_status_body)
             .map_err(|err| format!("pre-interrupt background heal status is not JSON ({err}): {pre_interrupt_status_body}"))?;
         let pre_interrupt_replacement = replacement_recovery_status(&cluster).await?;
-        let coordinator_log = std::fs::read_to_string(log_dir.join("node0.log"))?;
+        let coordinator_log = std::fs::read_to_string(format!("{log_dir}/node0.log"))?;
         assert!(
             coordinator_log
                 .lines()
@@ -1499,11 +1567,10 @@ mod tests {
                 "Restored target endpoint forwarding"
             );
         } else {
-            let graceful_restart = matches!(
+            if matches!(
                 scenario,
                 InterruptionScenario::BackgroundTargetRestart | InterruptionScenario::BackgroundTargetRestartEc84
-            );
-            if graceful_restart {
+            ) {
                 cluster.stop_node_gracefully(interruption_node).await?;
             } else {
                 cluster.stop_node(interruption_node)?;
@@ -1525,7 +1592,10 @@ mod tests {
             if background_enabled {
                 let marker_exists = unclean_shutdown_marker.is_file();
                 unclean_shutdown_marker_observed = Some(marker_exists);
-                let expected_marker = !graceful_restart;
+                let expected_marker = matches!(
+                    scenario,
+                    InterruptionScenario::BackgroundTargetCrash | InterruptionScenario::BackgroundTargetCrashEc84
+                );
                 assert!(
                     marker_exists == expected_marker,
                     "background restart/crash lane observed unexpected unclean-shutdown marker state"
@@ -1728,18 +1798,32 @@ mod tests {
         if let Some(evidence_context) = evidence_run {
             let restarted_pid = cluster.nodes[1].process.as_ref().ok_or("restarted target is absent")?.id();
             assert_ne!(target_pid, restarted_pid, "target must be a new process");
-            evidence_context.write(
-                &server_binary,
-                RestartObservation {
-                    nodes: cluster.nodes.len(),
-                    drives_per_node: cluster.nodes[0].data_dirs.len(),
-                    pid_before: target_pid,
-                    pid_after: restarted_pid,
-                    unclean_shutdown_marker: unclean_shutdown_marker_observed.ok_or("missing shutdown marker observation")?,
-                    objects: evidence_objects,
-                    node_listings,
-                },
-            )?;
+            assert_eq!(
+                file_sha256(&server_binary)?,
+                evidence_context.run.binary.sha256,
+                "server build changed during restart"
+            );
+            let evidence = serde_json::json!({
+                "schema": 1, "case": evidence_context.case.id, "evidence": evidence_context.case.evidence,
+                "run_id": evidence_context.run.run_id, "source_revision": evidence_context.run.source_revision,
+                "test_build": compiled_test_identity(),
+                "binary_sha256": evidence_context.run.binary.sha256,
+                "test_binary_sha256": evidence_context.run.test_binary.sha256,
+                "topology": {"nodes": cluster.nodes.len(), "drives_per_node": cluster.nodes[0].data_dirs.len()},
+                "pid_before": target_pid, "pid_after": restarted_pid,
+                "unclean_shutdown_marker": unclean_shutdown_marker_observed.unwrap_or(false),
+                "objects": evidence_objects, "node_listings": node_listings,
+            });
+            let data = serde_json::to_vec(&evidence)?;
+            if data.len() > 1024 * 1024 {
+                return Err("scanner/heal oracle exceeds the 1 MiB artifact budget".into());
+            }
+            let mut output = std::fs::OpenOptions::new()
+                .write(true)
+                .create_new(true)
+                .open(evidence_context.directory.join(evidence_context.case.oracle))?;
+            output.write_all(&data)?;
+            output.sync_all()?;
         }
 
         Ok(())

crates/e2e_test/src/inline_fast_path_cluster_test.rs

@@ -21,7 +21,7 @@
 //! One S3 GET can select readers on multiple EC nodes, so the counter tracks
 //! distributed reader selection rather than HTTP request count.
 
-use crate::common::{RustFSTestClusterEnvironment, RustFSTestEnvironment, init_logging, signal_process};
+use crate::common::{RustFSTestClusterEnvironment, RustFSTestEnvironment, init_logging};
 use aws_sdk_s3::Client;
 use aws_sdk_s3::primitives::ByteStream;
 use aws_sdk_s3::types::{
@@ -2207,33 +2207,6 @@ async fn four_node_manual_transition_job_status_survives_node_restart() -> TestR
     Ok(())
 }
 
-struct SuspendedTransitionTarget<'a> {
-    // Keep the owned child borrowed until it is resumed so its PID cannot be reused.
-    child: &'a std::process::Child,
-    suspended: bool,
-}
-
-impl<'a> SuspendedTransitionTarget<'a> {
-    fn suspend(child: &'a std::process::Child) -> TestResult<Self> {
-        signal_process(child.id(), "STOP")?;
-        Ok(Self { child, suspended: true })
-    }
-
-    fn resume(&mut self) -> TestResult {
-        signal_process(self.child.id(), "CONT")?;
-        self.suspended = false;
-        Ok(())
-    }
-}
-
-impl Drop for SuspendedTransitionTarget<'_> {
-    fn drop(&mut self) {
-        if self.suspended {
-            let _ = signal_process(self.child.id(), "CONT");
-        }
-    }
-}
-
 #[tokio::test]
 async fn four_node_manual_transition_distributed_admission_conflict_reports_status_and_backpressure() -> TestResult {
     init_logging();
@@ -2261,6 +2234,7 @@ async fn four_node_manual_transition_distributed_admission_conflict_reports_stat
     let bucket = format!("distributed-admission-{}", Uuid::new_v4().simple());
     let prefix = "transition/distributed-admission/";
     hot_client.create_bucket().bucket(&bucket).send().await?;
+    put_lifecycle_with_transition_retry(&hot_client, &bucket, &tier_name).await?;
     for index in 0u8..64 {
         let key = format!("{prefix}object-{index:02}.bin");
         hot_client
@@ -2271,21 +2245,6 @@ async fn four_node_manual_transition_distributed_admission_conflict_reports_stat
             .send()
             .await?;
     }
-    // Lifecycle PUT starts its own backfill. Keep its first page on a separate
-    // node and stop it at queue backpressure before it reaches the tested prefix:
-    // one active worker, one queued item, then the first rejected item.
-    for index in 0u8..3 {
-        hot_client
-            .put_object()
-            .bucket(&bucket)
-            .key(format!("transition/automatic-admission/object-{index:02}.bin"))
-            .body(ByteStream::from(payload(KIB, index)))
-            .send()
-            .await?;
-    }
-    let mut suspended_cold = SuspendedTransitionTarget::suspend(cold.process.as_ref().ok_or("cold-tier process missing")?)?;
-    let lifecycle_client = hot.create_s3_client(2)?;
-    put_lifecycle_with_transition_retry(&lifecycle_client, &bucket, &tier_name).await?;
 
     let (node0, node1) = tokio::join!(
         start_manual_transition_job_on_node(&hot, 0, &bucket, prefix, &tier_name, false, 64),
@@ -2345,31 +2304,6 @@ async fn four_node_manual_transition_distributed_admission_conflict_reports_stat
     assert_eq!(status["job_id"].as_str(), Some(job_id));
     assert_eq!(status["status_endpoint"].as_str(), Some(status_endpoint));
 
-    let deadline = Instant::now() + Duration::from_secs(30);
-    loop {
-        let status = read_manual_transition_job_status_endpoint(&hot, accepted.0, status_endpoint).await?;
-        assert_eq!(
-            status["status"].as_str(),
-            Some("running"),
-            "blocked cold tier must keep the admitted job running: {status}"
-        );
-        if status["report"]["skipped_queue_full"].as_u64().is_some_and(|count| count > 0) {
-            assert!(
-                status["report"]["enqueued"].as_u64().is_some_and(|count| count > 0),
-                "the job must own pending transitions while the cold tier is suspended: {status}"
-            );
-            break;
-        }
-        if Instant::now() >= deadline {
-            return Err(format!(
-                "manual transition job did not reach queue backpressure while the cold tier was suspended: {status}"
-            )
-            .into());
-        }
-        sleep(Duration::from_millis(50)).await;
-    }
-
-    suspended_cold.resume()?;
     let terminal = wait_for_manual_transition_job_terminal(&hot, conflict.0, job_id, false).await?;
     assert_eq!(terminal["job_id"].as_str(), Some(job_id));
     assert_eq!(terminal["bucket"].as_str(), Some(bucket.as_str()));

crates/e2e_test/src/lib.rs

@@ -23,9 +23,6 @@ pub mod common;
 #[cfg(test)]
 pub mod chaos;
 
-#[cfg(test)]
-mod scanner_heal_evidence;
-
 // Programmable S3 target for replication failure-path tests (backlog#1147 repl-8)
 // and on-demand-migration source scenarios (backlog#2151).
 #[cfg(test)]

crates/e2e_test/src/protocols/sftp_compliance.rs

@@ -79,11 +79,6 @@ pub async fn test_sftp_compliance_suite() -> Result<()> {
             .await
             .map_err(|e| anyhow!("{}", e))?;
 
-        // Protocol listeners can accept connections before IAM is initialized.
-        // A signed S3 request establishes readiness before the first SFTP login.
-        let s3 = build_test_s3_client(&format!("http://{COMPLIANCE_RW_S3_ADDRESS}"));
-        wait_for_s3_ready(&s3, 30).await?;
-
         let (session, sftp) = connect_sftp_to(COMPLIANCE_RW_SFTP_ADDRESS).await?;
 
         cmptst_01::run_medium_binary_round_trip(&sftp).await?;
@@ -106,6 +101,8 @@ pub async fn test_sftp_compliance_suite() -> Result<()> {
         // reach the finalised object as x-amz-meta-* user metadata
         // through the CreateMultipartUpload input field. The S3 client
         // connects to the same rustfs process this suite already drives.
+        let s3 = build_test_s3_client(&format!("http://{COMPLIANCE_RW_S3_ADDRESS}"));
+        wait_for_s3_ready(&s3, 30).await?;
         cmptst_34::run_open_attrs_round_trip_multipart(&sftp, &s3).await?;
 
         drop(sftp);

crates/e2e_test/src/protocols/sftp_core.rs

@@ -168,10 +168,6 @@ pub async fn test_sftp_core_operations() -> Result<()> {
             .await
             .map_err(|e| anyhow!("{}", e))?;
 
-        // Protocol listeners can accept connections before IAM is initialized.
-        let s3 = build_test_s3_client(S3_ENDPOINT);
-        wait_for_s3_ready(&s3, S3_READY_ATTEMPTS).await?;
-
         let (session, sftp) = connect_sftp().await?;
 
         // --- 1. Subsystem canary: SFTP session reachable after password auth ---
@@ -352,6 +348,16 @@ pub async fn test_sftp_core_operations() -> Result<()> {
         let _ = bad_session.disconnect(russh::Disconnect::ByApplication, "", "en").await;
         info!("PASS: bad-password authentication rejected");
 
+        // --- Cross-protocol setup: aws-sdk-s3 client against the same server ---
+        // The rustfs binary spawned for this suite serves both SFTP on port
+        // 9022 and S3 on port 9000. The S3 stack may need a moment to finish
+        // initialising after TCP is listening, so list_buckets is polled
+        // until it succeeds before any cross-protocol assertion runs.
+        info!("Testing SFTP: prepare aws-sdk-s3 client and wait for S3 readiness");
+        let s3 = build_test_s3_client(S3_ENDPOINT);
+        wait_for_s3_ready(&s3, S3_READY_ATTEMPTS).await?;
+        info!("PASS: S3 endpoint reachable from cross-protocol client");
+
         // --- SFTP write, S3 read: SHA256 round-trip ---
         // SFTP creates the object, then assert_cross_protocol_sha_match
         // fetches it via both S3 GetObject and SFTP READ and compares
@@ -516,9 +522,6 @@ pub async fn test_sftp_idle_timeout_disconnects() -> Result<()> {
             .await
             .map_err(|e| anyhow!("{}", e))?;
 
-        let s3 = build_test_s3_client(&format!("http://{IDLE_S3_ADDRESS}"));
-        wait_for_s3_ready(&s3, S3_READY_ATTEMPTS).await?;
-
         let (session, sftp) = connect_sftp_to(IDLE_SFTP_ADDRESS).await?;
 
         // Confirm the session is live before the wait so a failure in the

crates/e2e_test/src/scanner_heal_evidence.rs

@@ -1,181 +0,0 @@
-// Copyright 2026 RustFS Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-//! Build-bound evidence for scanner and heal restart tests.
-
-use crate::common::ClusterTopology;
-use sha2::{Digest, Sha256};
-use std::error::Error;
-use std::io::{Read, Write};
-use std::path::{Path, PathBuf};
-
-#[derive(serde::Deserialize)]
-struct EvidenceBuild {
-    sha256: String,
-}
-
-#[derive(serde::Deserialize)]
-struct RestartEvidenceRun {
-    schema: u32,
-    run_id: String,
-    source_revision: String,
-    test_build: serde_json::Value,
-    binary: EvidenceBuild,
-    test_binary: EvidenceBuild,
-}
-
-#[derive(Clone, Copy)]
-pub(crate) struct ScannerHealEvidenceCase {
-    pub(crate) id: &'static str,
-    pub(crate) oracle: &'static str,
-    pub(crate) evidence: &'static str,
-    pub(crate) unclean_shutdown_marker: bool,
-    pub(crate) topology: EvidenceTopology,
-    pub(crate) storage_class_standard: Option<&'static str>,
-    pub(crate) erasure_set_drive_count: Option<&'static str>,
-}
-
-#[derive(Clone, Copy)]
-pub(crate) struct EvidenceTopology {
-    pub(crate) nodes: usize,
-    pub(crate) drives_per_node: usize,
-}
-
-impl EvidenceTopology {
-    pub(crate) const fn new(nodes: usize, drives_per_node: usize) -> Self {
-        Self { nodes, drives_per_node }
-    }
-
-    pub(crate) fn total_drives(self) -> usize {
-        self.nodes * self.drives_per_node
-    }
-
-    pub(crate) fn cluster_topology(self) -> ClusterTopology {
-        ClusterTopology::single_pool_multidrive(self.nodes, self.drives_per_node)
-    }
-}
-
-pub(crate) struct RestartEvidenceContext {
-    directory: PathBuf,
-    run: RestartEvidenceRun,
-    case: ScannerHealEvidenceCase,
-}
-
-fn file_sha256(path: &Path) -> Result<String, Box<dyn Error + Send + Sync>> {
-    let mut file = std::fs::File::open(path)?;
-    let mut digest = Sha256::new();
-    let mut buffer = [0_u8; 64 * 1024];
-    loop {
-        let read = file.read(&mut buffer)?;
-        if read == 0 {
-            break;
-        }
-        digest.update(&buffer[..read]);
-    }
-    Ok(digest.finalize().iter().map(|byte| format!("{byte:02x}")).collect())
-}
-
-pub(crate) fn restart_evidence_run(
-    binary: &Path,
-    case: ScannerHealEvidenceCase,
-) -> Result<Option<RestartEvidenceContext>, Box<dyn Error + Send + Sync>> {
-    let Some(directory) = std::env::var_os("RUSTFS_SCANNER_HEAL_RUN_DIR") else {
-        return Ok(None);
-    };
-    if case.id.is_empty()
-        || case.oracle.is_empty()
-        || !case.oracle.ends_with(".json")
-        || case.oracle.contains('/')
-        || case.oracle.contains('\\')
-        || case.oracle.contains("..")
-        || !matches!(case.evidence, "process-restart" | "process-crash-restart")
-        || (case.evidence == "process-crash-restart") != case.unclean_shutdown_marker
-    {
-        return Err("invalid scanner/heal evidence case".into());
-    }
-    let directory = PathBuf::from(directory);
-    let receipt = directory.join("run.json");
-    if receipt.metadata()?.len() > 1024 * 1024 {
-        return Err("oversized scanner/heal execution receipt".into());
-    }
-    let run: RestartEvidenceRun = serde_json::from_slice(&std::fs::read(receipt)?)?;
-    if run.schema != 1 || run.run_id.len() != 32 || run.source_revision.len() != 40 {
-        return Err("invalid scanner/heal execution identity".into());
-    }
-    let built = compiled_test_identity();
-    for key in ["source_revision", "dirty", "lock_blob", "features"] {
-        assert_eq!(built[key], run.test_build[key], "compiled test identity differs for {key}");
-    }
-    assert_eq!(file_sha256(binary)?, run.binary.sha256, "server binary must match the run receipt");
-    assert_eq!(
-        file_sha256(&std::env::current_exe()?)?,
-        run.test_binary.sha256,
-        "test executable must match the run receipt"
-    );
-    if directory.join(case.oracle).exists() {
-        return Err("scanner/heal oracle already exists; create a new execution receipt".into());
-    }
-    Ok(Some(RestartEvidenceContext { directory, run, case }))
-}
-
-fn compiled_test_identity() -> serde_json::Value {
-    serde_json::json!({
-        "source_revision": env!("RUSTFS_E2E_BUILD_COMMIT"),
-        "dirty": env!("RUSTFS_E2E_BUILD_DIRTY") != "false",
-        "lock_blob": env!("RUSTFS_E2E_BUILD_LOCK"),
-        "features": env!("RUSTFS_E2E_BUILD_FEATURES"),
-        "target": env!("RUSTFS_E2E_BUILD_TARGET"),
-        "profile": env!("RUSTFS_E2E_BUILD_PROFILE"),
-        "rustflags_hex": env!("RUSTFS_E2E_BUILD_RUSTFLAGS_HEX"),
-    })
-}
-
-pub(crate) struct RestartObservation {
-    pub(crate) nodes: usize,
-    pub(crate) drives_per_node: usize,
-    pub(crate) pid_before: u32,
-    pub(crate) pid_after: u32,
-    pub(crate) unclean_shutdown_marker: bool,
-    pub(crate) objects: Vec<serde_json::Value>,
-    pub(crate) node_listings: Vec<Vec<String>>,
-}
-
-impl RestartEvidenceContext {
-    pub(crate) fn write(self, binary: &Path, observed: RestartObservation) -> Result<(), Box<dyn Error + Send + Sync>> {
-        assert_ne!(observed.pid_before, observed.pid_after, "target must be a new process");
-        assert_eq!(file_sha256(binary)?, self.run.binary.sha256, "server build changed during restart");
-        let evidence = serde_json::json!({
-            "schema": 1, "case": self.case.id, "evidence": self.case.evidence,
-            "run_id": self.run.run_id, "source_revision": self.run.source_revision,
-            "test_build": compiled_test_identity(),
-            "binary_sha256": self.run.binary.sha256,
-            "test_binary_sha256": self.run.test_binary.sha256,
-            "topology": {"nodes": observed.nodes, "drives_per_node": observed.drives_per_node},
-            "pid_before": observed.pid_before, "pid_after": observed.pid_after,
-            "unclean_shutdown_marker": observed.unclean_shutdown_marker,
-            "objects": observed.objects, "node_listings": observed.node_listings,
-        });
-        let data = serde_json::to_vec(&evidence)?;
-        if data.len() > 1024 * 1024 {
-            return Err("scanner/heal oracle exceeds the 1 MiB artifact budget".into());
-        }
-        let mut output = std::fs::OpenOptions::new()
-            .write(true)
-            .create_new(true)
-            .open(self.directory.join(self.case.oracle))?;
-        output.write_all(&data)?;
-        output.sync_all()?;
-        Ok(())
-    }
-}

crates/ecstore/src/disk/local.rs

@@ -22463,6 +22463,86 @@ mod test {
         );
     }
 
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn conditional_mrf_manifest_dir_fsync_failure_keeps_recovery_anchors() {
+        use tempfile::tempdir;
+
+        const MRF_COMMIT_MANIFEST_SLOT_0: &str = ".heal-mrf-commit.0.bin";
+        const MRF_COMMIT_MANIFEST_SLOT_1: &str = ".heal-mrf-commit.1.bin";
+        const MRF_SCOPED_JOURNAL_PATH: &str = "buckets/.heal/mrf/journal-scoped.bin";
+
+        let _mode = durability_mode_override::set(DurabilityMode::Relaxed);
+        let dir = tempdir().expect("temp dir should be created");
+        let endpoint = Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse");
+        let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created");
+        let previous_manifest = Bytes::from_static(b"mrf-committed-manifest-v1");
+        let successor_manifest = Bytes::from_static(b"mrf-committed-manifest-v2");
+        let legacy_journal = Bytes::from_static(b"legacy-mrf-journal-records");
+
+        assert_eq!(
+            disk.compare_and_update_file(RUSTFS_META_BUCKET, MRF_COMMIT_MANIFEST_SLOT_0, None, Some(previous_manifest.clone()),)
+                .await
+                .expect("previous MRF manifest should commit"),
+            ConditionalFileUpdate::Updated
+        );
+        disk.write_all(RUSTFS_META_BUCKET, MRF_SCOPED_JOURNAL_PATH, legacy_journal.clone())
+            .await
+            .expect("legacy MRF journal should be retained");
+
+        let manifest_path = disk
+            .get_object_path(RUSTFS_META_BUCKET, MRF_COMMIT_MANIFEST_SLOT_0)
+            .expect("MRF manifest path should resolve");
+        let parent = manifest_path.parent().expect("MRF manifest path should have a parent");
+        assert!(
+            os::fsync_dir_recorder::was_fsynced(parent),
+            "system metadata MRF manifest publication must fsync the metadata directory even under relaxed durability"
+        );
+        os::fsync_dir_recorder::set_failure(parent, ErrorKind::Other);
+
+        let err = disk
+            .compare_and_update_file(
+                RUSTFS_META_BUCKET,
+                MRF_COMMIT_MANIFEST_SLOT_0,
+                Some(previous_manifest.clone()),
+                Some(successor_manifest),
+            )
+            .await
+            .expect_err("directory fsync failure must fail the MRF manifest successor commit");
+        assert!(matches!(err, DiskError::Io(ref err) if err.kind() == ErrorKind::Other));
+        assert_eq!(
+            disk.read_all(RUSTFS_META_BUCKET, MRF_COMMIT_MANIFEST_SLOT_0)
+                .await
+                .expect("previous committed MRF manifest should remain readable after rollback"),
+            previous_manifest
+        );
+
+        os::fsync_dir_recorder::set_failure(parent, ErrorKind::Other);
+        let err = disk
+            .compare_and_update_file(
+                RUSTFS_META_BUCKET,
+                MRF_COMMIT_MANIFEST_SLOT_1,
+                None,
+                Some(Bytes::from_static(b"first-successor-manifest")),
+            )
+            .await
+            .expect_err("directory fsync failure must fail first MRF manifest commit");
+        assert!(matches!(err, DiskError::Io(ref err) if err.kind() == ErrorKind::Other));
+        assert!(
+            matches!(
+                disk.read_all(RUSTFS_META_BUCKET, MRF_COMMIT_MANIFEST_SLOT_1).await,
+                Err(DiskError::FileNotFound)
+            ),
+            "uncommitted first MRF manifest must be removed when no committed anchor exists"
+        );
+        assert_eq!(
+            disk.read_all(RUSTFS_META_BUCKET, MRF_SCOPED_JOURNAL_PATH)
+                .await
+                .expect("legacy MRF journal should remain readable after failed manifest publication"),
+            legacy_journal
+        );
+    }
+
     #[cfg(unix)]
     #[tokio::test]
     async fn conditional_file_update_dir_fsync_failure_removes_new_file_without_anchor() {

crates/ecstore/src/erasure/coding/encode.rs

@@ -431,7 +431,7 @@ impl<'a> MultiWriter<'a> {
             errs = ?self.errs,
             "Erasure encode write quorum unavailable: {summary_text}"
         );
-        Err(write_err.into())
+        Err(std::io::Error::other(format!("Failed to write data: {summary_text}")))
     }
 
     async fn shutdown_writer(writer_opt: &mut Option<BitrotWriterWrapper>, err: &mut Option<Error>) {
@@ -503,7 +503,7 @@ impl<'a> MultiWriter<'a> {
             errs = ?self.errs,
             "Erasure encode shutdown quorum unavailable: {summary_text}"
         );
-        Err(write_err.into())
+        Err(std::io::Error::other(format!("Failed to shutdown writers: {summary_text}")))
     }
 }
 
@@ -1002,7 +1002,6 @@ impl Erasure {
 mod tests {
     use super::*;
     use crate::erasure::coding::{BitrotWriterWrapper, CustomWriter};
-    use crate::error::StorageError;
     use rustfs_rio::HardLimitReader;
     use rustfs_utils::HashAlgorithm;
     use std::future::Future;
@@ -1452,14 +1451,7 @@ mod tests {
             Ok(_) => panic!("writer quorum failure should fail the encode pipeline"),
             Err(err) => err,
         };
-        let err = StorageError::from(err);
-        assert!(matches!(
-            &err,
-            StorageError::Io(source)
-                if source.kind() == std::io::ErrorKind::Other
-                    && source.to_string() == "injected write failure after producer blocks"
-        ));
-        assert!(!err.is_quorum_error());
+        assert!(err.to_string().contains("Failed to write data"));
         tokio::time::timeout(Duration::from_secs(1), reader_dropped)
             .await
             .expect("writer failure should abort the blocked producer")
@@ -1652,7 +1644,7 @@ mod tests {
 
     #[tokio::test]
     async fn multi_writer_short_write_fails_before_shutdown() {
-        let mut writers = vec![Some(bitrot_writer(ShortWriteWriter, 32))];
+        let mut writers = vec![Some(bitrot_writer(ShortWriteWriter, 16))];
         let err = {
             let mut writer = MultiWriter::new(&mut writers, 1);
             writer
@@ -1661,93 +1653,63 @@ mod tests {
                 .expect_err("short writes must fail the shard writer")
         };
 
-        let err = StorageError::from(err);
-        assert!(matches!(&err, StorageError::Io(source) if source.kind() == std::io::ErrorKind::WriteZero));
-        assert!(!err.is_quorum_error());
+        assert!(err.to_string().contains("Failed to write data"));
         assert!(writers[0].is_none(), "short-write shard must be removed before commit");
     }
 
     #[tokio::test]
     async fn multi_writer_reports_fallback_summary_when_only_offline_writers_remain() {
         let mut writers = vec![None, None];
-        let (err, summary) = {
+        let err = {
             let mut writer = MultiWriter::new(&mut writers, 1);
-            let err = writer
+            writer
                 .write(vec![Bytes::from_static(b"offline-a"), Bytes::from_static(b"offline-b")])
                 .await
-                .expect_err("offline writers cannot satisfy write quorum");
-            let summary = build_write_quorum_failure_summary(&writer.errs, OBJECT_OP_IGNORED_ERRS, writer.write_quorum);
-            (err, format_write_quorum_failure(&summary))
+                .expect_err("offline writers cannot satisfy write quorum")
         };
 
-        assert_eq!(
-            err.get_ref().and_then(|source| source.downcast_ref::<Error>()),
-            Some(&Error::ErasureWriteQuorum),
-        );
-        let err = StorageError::from(err);
-        assert_eq!(err, StorageError::ErasureWriteQuorum);
-        assert!(err.is_quorum_error());
-        assert!(summary.contains("offline-disks=2/2"));
-        assert!(summary.contains("required=1"));
+        let err = err.to_string();
+        assert!(err.contains("Failed to write data"));
+        assert!(err.contains("offline-disks=2/2"));
+        assert!(err.contains("required=1"));
 
-        let (shutdown_err, summary) = {
+        let shutdown_err = {
             let mut writer = MultiWriter::new(&mut writers, 1);
-            let err = writer
+            writer
                 .shutdown()
                 .await
-                .expect_err("offline writers cannot satisfy shutdown quorum");
-            let summary = build_write_quorum_failure_summary(&writer.errs, OBJECT_OP_IGNORED_ERRS, writer.write_quorum);
-            (err, format_write_quorum_failure(&summary))
+                .expect_err("offline writers cannot satisfy shutdown quorum")
         };
 
-        assert_eq!(
-            shutdown_err.get_ref().and_then(|source| source.downcast_ref::<Error>()),
-            Some(&Error::ErasureWriteQuorum),
-        );
-        let shutdown_err = StorageError::from(shutdown_err);
-        assert_eq!(shutdown_err, StorageError::ErasureWriteQuorum);
-        assert!(shutdown_err.is_quorum_error());
-        assert!(summary.contains("offline-disks=2/2"));
-        assert!(summary.contains("required=1"));
+        let shutdown_err = shutdown_err.to_string();
+        assert!(shutdown_err.contains("Failed to shutdown writers"));
+        assert!(shutdown_err.contains("offline-disks=2/2"));
+        assert!(shutdown_err.contains("required=1"));
     }
 
     #[tokio::test]
     async fn multi_writer_reports_quorum_failure_when_quorum_exceeds_writer_count() {
         let committed = Arc::new(Mutex::new(Vec::new()));
-        let mut writers = vec![Some(bitrot_writer(DeferredCommitWriter::new(committed), 32))];
+        let mut writers = vec![Some(bitrot_writer(DeferredCommitWriter::new(committed), 16))];
         let mut writer = MultiWriter::new(&mut writers, 2);
 
         let err = writer
             .write(vec![Bytes::from_static(b"quorum impossible")])
             .await
             .expect_err("write quorum above writer count must fail");
-        assert_eq!(
-            err.get_ref().and_then(|source| source.downcast_ref::<Error>()),
-            Some(&Error::ErasureWriteQuorum),
-        );
-        let err = StorageError::from(err);
-        assert_eq!(err, StorageError::ErasureWriteQuorum);
-        assert!(err.is_quorum_error());
-        let summary = build_write_quorum_failure_summary(&writer.errs, OBJECT_OP_IGNORED_ERRS, writer.write_quorum);
-        let summary = format_write_quorum_failure(&summary);
-        assert!(summary.contains("required=2"));
-        assert!(summary.contains("erasure write quorum"));
+        let err = err.to_string();
+        assert!(err.contains("Failed to write data"));
+        assert!(err.contains("required=2"));
+        assert!(err.contains("erasure write quorum"));
 
         let shutdown_err = writer
             .shutdown()
             .await
             .expect_err("shutdown quorum above writer count must fail");
-        assert_eq!(
-            shutdown_err.get_ref().and_then(|source| source.downcast_ref::<Error>()),
-            Some(&Error::ErasureWriteQuorum),
-        );
-        let shutdown_err = StorageError::from(shutdown_err);
-        assert_eq!(shutdown_err, StorageError::ErasureWriteQuorum);
-        assert!(shutdown_err.is_quorum_error());
-        let summary = build_write_quorum_failure_summary(&writer.errs, OBJECT_OP_IGNORED_ERRS, writer.write_quorum);
-        let summary = format_write_quorum_failure(&summary);
-        assert!(summary.contains("required=2"));
-        assert!(summary.contains("erasure write quorum"));
+        let shutdown_err = shutdown_err.to_string();
+        assert!(shutdown_err.contains("Failed to shutdown writers"));
+        assert!(shutdown_err.contains("required=2"));
+        assert!(shutdown_err.contains("erasure write quorum"));
     }
 
     // The production wiring (`MultiWriter::new`) must arm a real deadline by
@@ -1832,13 +1794,7 @@ mod tests {
             .write(four_shards())
             .await
             .expect_err("two stalled writers must fail the write quorum instead of hanging");
-        assert_eq!(
-            err.get_ref().and_then(|source| source.downcast_ref::<Error>()),
-            Some(&Error::ErasureWriteQuorum),
-        );
-        let err = StorageError::from(err);
-        assert_eq!(err, StorageError::ErasureWriteQuorum);
-        assert!(err.is_quorum_error());
+        assert!(err.to_string().contains("Failed to write data"));
     }
 
     // A small object whose bytes were fully buffered leaves `write` succeeding
@@ -1883,13 +1839,7 @@ mod tests {
             .shutdown()
             .await
             .expect_err("two shutdown stalls must fail the shutdown quorum instead of hanging");
-        assert_eq!(
-            err.get_ref().and_then(|source| source.downcast_ref::<Error>()),
-            Some(&Error::ErasureWriteQuorum),
-        );
-        let err = StorageError::from(err);
-        assert_eq!(err, StorageError::ErasureWriteQuorum);
-        assert!(err.is_quorum_error());
+        assert!(err.to_string().contains("Failed to shutdown writers"));
     }
 
     // A slow-but-honest writer that keeps completing shards (delay < stall
@@ -2171,13 +2121,7 @@ mod tests {
             .await
             .expect_err("streaming encode must fail when write quorum is unavailable");
 
-        assert_eq!(
-            err.get_ref().and_then(|source| source.downcast_ref::<Error>()),
-            Some(&Error::ErasureWriteQuorum),
-        );
-        let err = StorageError::from(err);
-        assert_eq!(err, StorageError::ErasureWriteQuorum);
-        assert!(err.is_quorum_error());
+        assert!(err.to_string().contains("Failed to write data"));
     }
 
     #[tokio::test]
@@ -2201,13 +2145,7 @@ mod tests {
             .await
             .expect_err("write quorum failure must fail the inline encode");
 
-        assert_eq!(
-            err.get_ref().and_then(|source| source.downcast_ref::<Error>()),
-            Some(&Error::ErasureWriteQuorum),
-        );
-        let err = StorageError::from(err);
-        assert_eq!(err, StorageError::ErasureWriteQuorum);
-        assert!(err.is_quorum_error());
+        assert!(err.to_string().contains("Failed to write data"));
         assert!(
             committed.lock().expect("committed buffer should be lockable").is_empty(),
             "successful writer must not be committed when write quorum fails before shutdown"
@@ -2235,13 +2173,7 @@ mod tests {
             .await
             .expect_err("shutdown quorum failure must fail the inline encode");
 
-        let err = StorageError::from(err);
-        assert!(matches!(
-            &err,
-            StorageError::Io(source)
-                if source.kind() == std::io::ErrorKind::Other && source.to_string() == "injected shutdown failure"
-        ));
-        assert!(!err.is_quorum_error());
+        assert!(err.to_string().contains("Failed to shutdown writers"));
         assert!(
             !committed.lock().expect("committed buffer should be lockable").is_empty(),
             "the successful writer should have committed before shutdown quorum failure was reported"
@@ -2463,13 +2395,7 @@ mod tests {
             .await
             .expect_err("batched encode must fail when write quorum is unavailable");
 
-        assert_eq!(
-            err.get_ref().and_then(|source| source.downcast_ref::<Error>()),
-            Some(&Error::ErasureWriteQuorum),
-        );
-        let err = StorageError::from(err);
-        assert_eq!(err, StorageError::ErasureWriteQuorum);
-        assert!(err.is_quorum_error());
+        assert!(err.to_string().contains("Failed to write data"));
     }
 
     #[tokio::test]

crates/ecstore/src/store/heal.rs

@@ -378,26 +378,6 @@ impl ECStore {
         Ok(result)
     }
 
-    /// Whether this replacement set owns the pool's metadata replica.
-    ///
-    /// Pool metadata follows normal object placement within each pool. A valid
-    /// non-owner set has no replica to repair; missing metadata on the owner
-    /// set still requires healing and target-specific readback.
-    pub fn replacement_pool_metadata_applies(&self, pool_index: usize, set_index: usize) -> Result<bool> {
-        let pool = self
-            .pools
-            .get(pool_index)
-            .ok_or_else(|| invalid_heal_pool_index(pool_index, self.pools.len()))?;
-        let selected = pool.get_disks_for_heal_object(
-            POOL_META_NAME,
-            &HealOpts {
-                set: Some(set_index),
-                ..Default::default()
-            },
-        )?;
-        Ok(Arc::ptr_eq(&selected, &pool.get_disks_by_key(POOL_META_NAME)))
-    }
-
     #[instrument(skip(self, targets), fields(pool_index, set_index, target_count = targets.len()))]
     pub async fn replacement_targets_have_version(
         &self,
@@ -849,73 +829,6 @@ mod tests {
         }
     }
 
-    #[tokio::test]
-    async fn replacement_pool_metadata_applies_to_the_written_replica_in_each_pool() {
-        let mut store = minimal_heal_store().await;
-        for pool_index in 0..store.pools.len() {
-            assert!(
-                store
-                    .replacement_pool_metadata_applies(pool_index, 0)
-                    .expect("a valid single-set pool should have a metadata owner")
-            );
-        }
-        store.ctx = Arc::new(InstanceContext::new());
-        for algorithm in [
-            crate::disk::format::DistributionAlgoVersion::V1,
-            crate::disk::format::DistributionAlgoVersion::V2,
-            crate::disk::format::DistributionAlgoVersion::V3,
-        ] {
-            let mut temp_dirs = Vec::new();
-            for pool_index in 0..store.pools.len() {
-                let (dirs, mut pool) =
-                    crate::core::sets::make_local_two_set_sets_for_pool_with_ctx(Arc::clone(&store.ctx), pool_index).await;
-                temp_dirs.extend(dirs);
-                Arc::get_mut(&mut pool)
-                    .expect("fixture pool should have one owner")
-                    .distribution_algo = algorithm.clone();
-                store.pools[pool_index] = pool;
-            }
-            for (pool_index, pool) in store.pools.iter().enumerate() {
-                let mut required_sets = 0;
-                for set_index in 0..pool.disk_set.len() {
-                    required_sets += usize::from(
-                        store
-                            .replacement_pool_metadata_applies(pool_index, set_index)
-                            .expect("valid replacement topology should be classified before metadata exists"),
-                    );
-                }
-                assert_eq!(required_sets, 1, "missing metadata cannot exempt the owner set");
-                save_config(pool.clone(), POOL_META_NAME, b"pool metadata placement".to_vec())
-                    .await
-                    .expect("normal config writes should persist one metadata replica per pool");
-                for (set_index, set) in pool.disk_set.iter().enumerate() {
-                    let applies = store
-                        .replacement_pool_metadata_applies(pool_index, set_index)
-                        .expect("valid replacement topology should be classified");
-                    let disks = set.disks.read().await.clone();
-                    for disk in disks.iter().flatten() {
-                        let replica = disk.read_xl(RUSTFS_META_BUCKET, POOL_META_NAME, false).await;
-                        if applies {
-                            replica.expect("the metadata owner must match actual persisted shards");
-                        } else {
-                            assert!(
-                                matches!(replica, Err(crate::disk::error::DiskError::FileNotFound)),
-                                "non-owner sets must have no persisted metadata shard; observed error: {:?}",
-                                replica.as_ref().err()
-                            );
-                        }
-                    }
-                }
-                assert!(
-                    store
-                        .replacement_pool_metadata_applies(pool_index, pool.disk_set.len())
-                        .is_err()
-                );
-            }
-        }
-        assert!(store.replacement_pool_metadata_applies(store.pools.len(), 0).is_err());
-    }
-
     async fn remove_pool_meta_shard(store: &ECStore, pool_idx: usize) -> DiskStore {
         let target_set = store.pools[pool_idx].get_disks_by_key(POOL_META_NAME);
         let missing_disk = target_set.disks.read().await[0]

crates/heal/src/heal/erasure_healer.rs

@@ -113,6 +113,7 @@ pub struct ErasureSetHealer {
     heal_opts: HealOpts,
     source: HealRequestSource,
     target_endpoints: Arc<[String]>,
+    pool_metadata_target_endpoints: Arc<[String]>,
     replacement_task_id: Option<String>,
     replacement_target_identities: Option<Arc<[ReplacementTargetIdentity]>>,
     mainline_pacer: Option<Arc<super::pacing::MainlinePacer>>,
@@ -362,6 +363,7 @@ impl ErasureSetHealer {
             heal_opts,
             source,
             target_endpoints: Vec::new().into(),
+            pool_metadata_target_endpoints: Vec::new().into(),
             replacement_task_id: None,
             replacement_target_identities: None,
             mainline_pacer: None,
@@ -385,6 +387,13 @@ impl ErasureSetHealer {
         self
     }
 
+    pub(crate) fn with_pool_metadata_targets(mut self, mut target_endpoints: Vec<String>) -> Self {
+        target_endpoints.sort_unstable();
+        target_endpoints.dedup();
+        self.pool_metadata_target_endpoints = target_endpoints.into();
+        self
+    }
+
     pub(crate) fn with_replacement_identity_fence(
         mut self,
         replacement_target_identities: Option<Vec<ReplacementTargetIdentity>>,
@@ -948,19 +957,21 @@ impl ErasureSetHealer {
         resume_manager: &ResumeManager,
         checkpoint_manager: &CheckpointManager,
     ) -> Result<()> {
-        if self.replacement_task_id.is_none() {
+        let target_endpoints = if self.pool_metadata_target_endpoints.is_empty() {
+            self.target_endpoints.as_ref()
+        } else {
+            self.pool_metadata_target_endpoints.as_ref()
+        };
+        let target_scoped_recreate = !self.heal_opts.dry_run && self.heal_opts.recreate && !target_endpoints.is_empty();
+        if self.replacement_task_id.is_none() && !target_scoped_recreate {
             return Ok(());
         }
-        if self.target_endpoints.is_empty() {
+        if target_endpoints.is_empty() {
             return Err(Error::TaskExecutionFailed {
                 message: "Replacement pool metadata heal requires target endpoints".to_string(),
             });
         }
 
-        if !self.storage.replacement_pool_metadata_applies(&self.heal_opts).await? {
-            return Ok(());
-        }
-
         let object_key = format!("{RUSTFS_META_BUCKET}/{POOL_META_NAME}");
         let checkpoint_key = compose_key(&object_key, None);
         let checkpoint = checkpoint_manager.get_checkpoint().await;
@@ -982,17 +993,11 @@ impl ErasureSetHealer {
             .heal_object(RUSTFS_META_BUCKET, POOL_META_NAME, None, &self.heal_opts)
             .await
         {
-            Ok((result, None)) if target_outcomes_complete(&result, &self.target_endpoints) => {
+            Ok((result, None)) if target_outcomes_complete(&result, target_endpoints) => {
                 let object_size = result_object_size_u64(&result);
                 match self
                     .storage
-                    .replacement_targets_have_version(
-                        RUSTFS_META_BUCKET,
-                        POOL_META_NAME,
-                        None,
-                        &self.heal_opts,
-                        &self.target_endpoints,
-                    )
+                    .replacement_targets_have_version(RUSTFS_META_BUCKET, POOL_META_NAME, None, &self.heal_opts, target_endpoints)
                     .await
                 {
                     Ok(true) => (object_size, Ok(())),
@@ -2033,8 +2038,6 @@ mod resume_loop_tests {
     #[derive(Clone)]
     enum HealOutcome {
         Ok,
-        /// The object has no metadata on any disk in the selected set.
-        FileNotFound,
         /// The version vanished before heal ran (deleted mid-heal).
         VersionNotFound,
         /// A transient infrastructure condition (offline disk / unmet quorum):
@@ -2060,8 +2063,6 @@ mod resume_loop_tests {
         /// Target-specific physical readback evidence per `compose_key`; the
         /// fake models a healthy backend unless a test explicitly revokes it.
         replacement_commit_evidence: Mutex<HashMap<String, ReplacementCommitEvidence>>,
-        pool_metadata_not_applicable: AtomicBool,
-        fail_pool_metadata_scope: AtomicBool,
         lifecycle_expired: Mutex<HashSet<String>>,
         /// every heal_object call recorded as (name, version_id)
         heal_calls: Mutex<Vec<(String, Option<String>)>>,
@@ -2163,7 +2164,6 @@ mod resume_loop_tests {
             let outcome = self.outcomes.lock().unwrap().get(&key).cloned().unwrap_or(HealOutcome::Ok);
             match outcome {
                 HealOutcome::Ok => Ok((self.results.lock().unwrap().get(&key).cloned().unwrap_or_default(), None)),
-                HealOutcome::FileNotFound => Ok((HealResultItem::default(), Some(Error::Storage(EcstoreError::FileNotFound)))),
                 HealOutcome::VersionNotFound => {
                     Ok((HealResultItem::default(), Some(Error::Storage(EcstoreError::FileVersionNotFound))))
                 }
@@ -2177,17 +2177,6 @@ mod resume_loop_tests {
         async fn heal_format(&self, _dry: bool) -> Result<(HealResultItem, Option<Error>)> {
             Ok((HealResultItem::default(), None))
         }
-        async fn replacement_pool_metadata_applies(&self, opts: &HealOpts) -> Result<bool> {
-            if self.fail_pool_metadata_scope.load(Ordering::SeqCst) {
-                return Err(Error::other("injected pool metadata scope failure"));
-            }
-            if self.pool_metadata_not_applicable.load(Ordering::SeqCst) {
-                assert_eq!(opts.pool, Some(0));
-                assert_eq!(opts.set, Some(1));
-                return Ok(false);
-            }
-            Ok(true)
-        }
         async fn replacement_targets_have_version(
             &self,
             _bucket: &str,
@@ -2733,111 +2722,6 @@ mod resume_loop_tests {
         drop(checkpoint);
     }
 
-    #[tokio::test]
-    async fn replacement_pool_metadata_non_owner_completes_but_missing_owner_retries() {
-        for owns_pool_metadata in [false, true] {
-            let env = make_env_with_targets(vec!["replacement-a".to_string()]).await;
-            let replacement_task_id = ResumeUtils::generate_task_id();
-            let set_index = usize::from(!owns_pool_metadata);
-            let set_disk_id = format!("pool_0_set_{set_index}");
-            ResumeManager::new_replacement_intent(
-                env.healer.disk.clone(),
-                replacement_task_id.clone(),
-                set_disk_id.clone(),
-                vec!["b".to_string()],
-                vec!["replacement-a".to_string()],
-                vec![crate::heal::resume::ReplacementTargetIdentity {
-                    endpoint: "replacement-a".to_string(),
-                    canonical_path: "/mnt/replacement-a".to_string(),
-                    physical_device_ids: vec!["device-a".to_string()],
-                    filesystem_identity: "1:2:3".to_string(),
-                }],
-            )
-            .await
-            .expect("replacement intent should persist");
-            env.storage
-                .pool_metadata_not_applicable
-                .store(!owns_pool_metadata, Ordering::SeqCst);
-            env.storage.set_outcome(POOL_META_NAME, None, HealOutcome::FileNotFound);
-            let healer = ErasureSetHealer::new(
-                env.storage.clone(),
-                Arc::new(RwLock::new(HealProgress::new())),
-                CancellationToken::new(),
-                env.healer.disk.clone(),
-                HealOpts {
-                    pool: Some(0),
-                    set: Some(set_index),
-                    ..Default::default()
-                },
-                HealRequestSource::AutoHeal,
-            )
-            .with_replacement_targets(vec!["replacement-a".to_string()], Some(replacement_task_id.clone()));
-
-            let result = healer.heal_erasure_set(&["b".to_string()], &set_disk_id).await;
-            let state = ResumeManager::load_replacement_intent(env.healer.disk.clone(), &replacement_task_id)
-                .await
-                .expect("replacement state must remain until marker cleanup")
-                .get_state()
-                .await;
-            if owns_pool_metadata {
-                let error = result.expect_err("missing metadata in the owner set must keep replacement incomplete");
-                assert!(error.to_string().contains("Replacement erasure set heal incomplete"));
-                assert!(!state.completed);
-                assert_eq!(state.replacement_phase, crate::heal::resume::ReplacementPhase::Intent);
-                assert_eq!(state.retry_count, 1);
-                assert_eq!(env.storage.calls(), vec![(POOL_META_NAME.to_string(), None)]);
-            } else {
-                result.expect("a non-owner set must complete without a pool metadata replica");
-                assert!(state.completed);
-                assert_eq!(state.replacement_phase, crate::heal::resume::ReplacementPhase::Verified);
-                assert_eq!(state.retry_count, 0);
-                assert!(env.storage.calls().is_empty(), "non-owner sets must not attempt pool metadata repair");
-            }
-        }
-    }
-
-    #[tokio::test]
-    async fn replacement_pool_metadata_unknown_scope_cannot_complete() {
-        let env = make_env_with_targets(vec!["replacement-a".to_string()]).await;
-        let healer = ErasureSetHealer::new(
-            env.storage.clone(),
-            Arc::new(RwLock::new(HealProgress::new())),
-            CancellationToken::new(),
-            env.healer.disk.clone(),
-            HealOpts {
-                pool: Some(0),
-                set: Some(0),
-                ..Default::default()
-            },
-            HealRequestSource::AutoHeal,
-        )
-        .with_replacement_targets(vec!["replacement-a".to_string()], Some("generation-a".to_string()));
-        env.storage.fail_pool_metadata_scope.store(true, Ordering::SeqCst);
-        env.storage
-            .set_result(POOL_META_NAME, None, replacement_target_ok_result("replacement-a", POOL_META_NAME));
-        let mut processed_objects = 0;
-        let mut successful_objects = 0;
-        let mut failed_objects = 0;
-        let mut skipped_objects = 0;
-        let error = healer
-            .heal_replacement_pool_metadata(
-                "pool_0_set_0",
-                &mut super::ErasureSetPassCounters {
-                    processed_objects: &mut processed_objects,
-                    successful_objects: &mut successful_objects,
-                    failed_objects: &mut failed_objects,
-                    skipped_objects: &mut skipped_objects,
-                },
-                &env.resume,
-                &env.checkpoint,
-            )
-            .await
-            .expect_err("unknown metadata placement must keep replacement incomplete");
-        assert!(error.to_string().contains("injected pool metadata scope failure"));
-        assert!(env.storage.calls().is_empty());
-        assert_eq!((processed_objects, successful_objects, failed_objects, skipped_objects), (0, 0, 0, 0));
-    }
-
     #[tokio::test]
     async fn replacement_pool_metadata_readback_failure_schedules_retry() {
         let env = make_env_with_targets(vec!["replacement-a".to_string()]).await;
@@ -2887,6 +2771,68 @@ mod resume_loop_tests {
         assert_eq!(env.storage.calls(), vec![(POOL_META_NAME.to_string(), None)]);
     }
 
+    #[tokio::test]
+    async fn admin_recreate_target_heals_pool_metadata_before_completion() {
+        let env = make_env_with_targets(vec!["replacement-a".to_string()]).await;
+        let healer = ErasureSetHealer::new(
+            env.storage.clone(),
+            Arc::new(RwLock::new(HealProgress::new())),
+            CancellationToken::new(),
+            env.healer.disk.clone(),
+            HealOpts {
+                recreate: true,
+                pool: Some(0),
+                set: Some(0),
+                ..Default::default()
+            },
+            HealRequestSource::Admin,
+        )
+        .with_pool_metadata_targets(vec!["replacement-a".to_string()]);
+        env.storage
+            .set_result(POOL_META_NAME, None, replacement_target_ok_result("replacement-a", POOL_META_NAME));
+
+        healer
+            .execute_heal_with_resume(&["b".to_string()], "pool_0_set_0", &env.resume, &env.checkpoint)
+            .await
+            .expect("admin recreate should heal and verify pool metadata");
+
+        assert!(env.resume.get_state().await.completed);
+        assert_eq!(env.storage.calls(), vec![(POOL_META_NAME.to_string(), None)]);
+    }
+
+    #[tokio::test]
+    async fn admin_recreate_pool_metadata_readback_failure_keeps_resume_state() {
+        let env = make_env_with_targets(vec!["replacement-a".to_string()]).await;
+        let healer = ErasureSetHealer::new(
+            env.storage.clone(),
+            Arc::new(RwLock::new(HealProgress::new())),
+            CancellationToken::new(),
+            env.healer.disk.clone(),
+            HealOpts {
+                recreate: true,
+                pool: Some(0),
+                set: Some(0),
+                ..Default::default()
+            },
+            HealRequestSource::Admin,
+        )
+        .with_pool_metadata_targets(vec!["replacement-a".to_string()]);
+        env.storage
+            .set_result(POOL_META_NAME, None, replacement_target_ok_result("replacement-a", POOL_META_NAME));
+        env.storage.set_replacement_commit_evidence(POOL_META_NAME, None, false);
+
+        let error = healer
+            .execute_heal_with_resume(&["b".to_string()], "pool_0_set_0", &env.resume, &env.checkpoint)
+            .await
+            .expect_err("unconfirmed admin recreate pool metadata must keep the set incomplete");
+
+        assert!(error.to_string().contains("Erasure set heal incomplete"));
+        let state = env.resume.get_state().await;
+        assert!(!state.completed);
+        assert_eq!(state.retry_count, 1);
+        assert_eq!(env.storage.calls(), vec![(POOL_META_NAME.to_string(), None)]);
+    }
+
     #[tokio::test]
     async fn retry_exhaustion_keeps_resume_artifacts_for_recovery() {
         let env = make_env().await;

crates/heal/src/heal/manager.rs

@@ -1527,7 +1527,7 @@ impl HealManager {
         request: HealRequest,
         preserve_alias: bool,
     ) -> Result<HealAdmissionReceipt> {
-        self.submit_heal_request_with_receipt_alias_and_mrf_notice(request, preserve_alias, None)
+        self.submit_heal_request_with_receipt_alias_and_mrf_notice(request, preserve_alias, true, None)
             .await
     }
 
@@ -1563,7 +1563,7 @@ impl HealManager {
         request: HealRequest,
         mrf_notice_target: MrfRepairNoticeTarget,
     ) -> Result<HealAdmissionReceipt> {
-        self.submit_heal_request_with_receipt_alias_and_mrf_notice(request, true, Some(mrf_notice_target))
+        self.submit_heal_request_with_receipt_alias_and_mrf_notice(request, true, true, Some(mrf_notice_target))
             .await
     }
 
@@ -1583,6 +1583,7 @@ impl HealManager {
         &self,
         request: HealRequest,
         preserve_alias: bool,
+        accept_same_request_id_replay: bool,
         mrf_notice_target: Option<MrfRepairNoticeTarget>,
     ) -> Result<HealAdmissionReceipt> {
         let admission_start = Instant::now();
@@ -1661,7 +1662,11 @@ impl HealManager {
             });
         if let Some((matches_existing, duplicate_state)) = request_id_admission {
             let admission = if matches_existing {
-                HealAdmissionResult::Accepted
+                if accept_same_request_id_replay {
+                    HealAdmissionResult::Accepted
+                } else {
+                    Self::duplicate_admission_for_request(&request, &config)
+                }
             } else {
                 HealAdmissionResult::Dropped(HealAdmissionDropReason::AlreadyRunning)
             };
@@ -1908,7 +1913,10 @@ impl HealManager {
 
     /// Submit heal request.
     pub async fn submit_heal_request(&self, request: HealRequest) -> Result<HealAdmissionResult> {
-        Ok(self.submit_heal_request_with_receipt_and_alias(request, true).await?.result)
+        Ok(self
+            .submit_heal_request_with_receipt_alias_and_mrf_notice(request, true, false, None)
+            .await?
+            .result)
     }
 
     /// Get task status

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

@@ -1869,7 +1869,7 @@ async fn test_submit_heal_request_returns_merged_for_duplicate() {
     let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
     let manager = HealManager::new(storage, None);
 
-    let mut request = HealRequest::new(
+    let request = HealRequest::new(
         HealType::Object {
             bucket: "bucket".to_string(),
             object: "object".to_string(),
@@ -1886,7 +1886,6 @@ async fn test_submit_heal_request_returns_merged_for_duplicate() {
             .expect("first request should be accepted"),
         HealAdmissionResult::Accepted
     );
-    request.id = uuid::Uuid::new_v4().to_string();
     assert_eq!(
         manager
             .submit_heal_request(request)
@@ -3726,7 +3725,7 @@ async fn test_submit_heal_request_returns_merged_before_full_for_duplicate() {
         }),
     );
 
-    let mut request = HealRequest::new(
+    let request = HealRequest::new(
         HealType::Object {
             bucket: "bucket".to_string(),
             object: "object".to_string(),
@@ -3743,7 +3742,6 @@ async fn test_submit_heal_request_returns_merged_before_full_for_duplicate() {
             .expect("first request should be accepted"),
         HealAdmissionResult::Accepted
     );
-    request.id = uuid::Uuid::new_v4().to_string();
     assert_eq!(
         manager
             .submit_heal_request(request)

crates/heal/src/heal/mrf_queue.rs

@@ -38,7 +38,7 @@ use crate::heal::manager::{HealManager, MrfRepairNoticeTarget};
 use metrics::{counter, gauge};
 use rustfs_common::mrf_channel::{MRF_MAX_ATTEMPTS, MrfDurableRepairAnchor, MrfIngressResult, MrfIntent};
 use rustfs_heal_contracts::heal_channel::{HealAdmissionDropReason, HealAdmissionResult};
-use std::collections::{HashMap, HashSet, VecDeque};
+use std::collections::{HashSet, VecDeque};
 use std::sync::Arc;
 use std::time::Duration;
 use tokio::sync::mpsc;
@@ -134,10 +134,7 @@ struct MrfQueueKey {
 }
 
 fn queue_key(intent: &MrfIntent) -> MrfQueueKey {
-    let version_id = (!matches!(intent.kind, rustfs_common::mrf_channel::MrfKind::MetadataCorruption))
-        .then_some(intent.version_id)
-        .flatten()
-        .filter(|bytes| *bytes != [0; 16]);
+    let version_id = intent.version_id.filter(|bytes| *bytes != [0; 16]);
     let scope = (!matches!(intent.kind, rustfs_common::mrf_channel::MrfKind::MetadataCorruption))
         .then_some(intent.scope)
         .flatten();
@@ -297,7 +294,11 @@ fn decode_one(data: &[u8]) -> Option<(MrfIntent, usize)> {
     };
     let attempts = data[3];
     let enqueued_at_ms = u64::from_le_bytes(data[4..12].try_into().ok()?);
-    let has_version = data[12] != 0;
+    let has_version = match data[12] {
+        0 => false,
+        1 => true,
+        _ => return None,
+    };
     let mut cursor = MRF_RECORD_FIXED_HEAD;
     let version_id = if has_version {
         if data.len() < cursor + 16 {
@@ -536,101 +537,29 @@ struct MrfRuntime {
     /// Partial-write responsibilities accepted from replay and waiting for an
     /// exact storage-owned proof before the startup journal can be deleted.
     durable_replay_anchors: Vec<MrfDurableRepairAnchor>,
-    /// Startup responsibilities remain in every successor snapshot until an
-    /// exact verified repair discharges them. Live admissions never grow this
-    /// set, so its size is bounded by the decoded startup journal.
-    retained_replay_intents: HashMap<MrfQueueKey, MrfIntent>,
     /// Earliest instant a full-admission retry may proceed.
     backoff_until: Option<tokio::time::Instant>,
 }
 
 impl MrfRuntime {
-    fn enqueue_batch(&mut self, intents: impl IntoIterator<Item = MrfIntent>) -> usize {
-        // Reserve retained startup work before admitting live hints. Computing
-        // the union once per batch avoids scanning it for every incoming hint.
-        let mut snapshot_count = self.queue.depth();
-        let mut snapshot_bytes = self.queue.bytes();
-        for (key, intent) in &self.retained_replay_intents {
-            if !self.queue.pending_keys.contains(key) {
-                snapshot_count = snapshot_count.saturating_add(1);
-                snapshot_bytes = snapshot_bytes.saturating_add(intent.estimated_bytes());
-            }
-        }
-        let mut enqueued = 0;
-        for intent in intents {
-            let key = queue_key(&intent);
-            let retained = self.retained_replay_intents.get(&key);
-            let additional = retained.is_none() && !self.queue.pending_keys.contains(&key);
-            let next_count = snapshot_count.saturating_add(usize::from(additional));
-            let next_bytes = snapshot_bytes.saturating_add(if additional { intent.estimated_bytes() } else { 0 });
-            let result = if retained.is_some_and(|retained| retained.lease != intent.lease)
-                || next_count > self.queue.capacity
-                || next_bytes > self.queue.byte_budget
-            {
-                counter!("rustfs_heal_mrf_dropped_total", "reason" => "queue_overflow").increment(1);
-                MrfQueuePushResult::Rejected
-            } else {
-                self.queue.try_push_typed(intent.clone())
-            };
-            match result {
-                MrfQueuePushResult::Enqueued => {
-                    snapshot_count = next_count;
-                    snapshot_bytes = next_bytes;
-                    enqueued += 1;
-                    self.new_since_flush += 1;
-                    self.dirty = true;
-                }
-                MrfQueuePushResult::Coalesced | MrfQueuePushResult::Rejected => {
-                    rustfs_common::mrf_channel::release_mrf_intent(&intent);
-                }
-            }
-        }
-        enqueued
-    }
-
-    fn snapshot(&self) -> Option<(Vec<u8>, Vec<u8>)> {
+    fn snapshot(&self) -> (Vec<u8>, Vec<u8>) {
         let mut authoritative = Vec::new();
         let mut legacy = Vec::new();
-        let mut encoded = HashSet::new();
-        for intent in self.queue.intents().chain(self.retained_replay_intents.values()) {
-            let key = queue_key(intent);
-            if self
-                .retained_replay_intents
-                .get(&key)
-                .is_some_and(|retained| retained.lease != intent.lease)
-            {
-                // Legacy records cannot distinguish concurrent responsibilities
-                // with different leases. Preserve the existing disk anchor.
-                return None;
-            }
-            if !encoded.insert(key) {
-                continue;
-            }
-            if encoded.len() > self.queue.capacity {
-                return None;
-            }
+        for intent in self.queue.intents() {
             let scoped_identity =
                 !matches!(intent.kind, rustfs_common::mrf_channel::MrfKind::MetadataCorruption) && intent.scope.is_some();
             if !encode_intent(intent, &mut authoritative) {
                 counter!("rustfs_heal_mrf_dropped_total", "reason" => "journal_identity_oversized").increment(1);
-                return None;
-            }
-            if authoritative.len() > self.queue.byte_budget {
-                return None;
             }
             if !scoped_identity && !encode_intent(intent, &mut legacy) {
                 counter!("rustfs_heal_mrf_dropped_total", "reason" => "journal_identity_oversized").increment(1);
-                return None;
             }
         }
-        Some((authoritative, legacy))
+        (authoritative, legacy)
     }
 
     async fn flush(&mut self) {
-        let Some((authoritative, legacy)) = self.snapshot() else {
-            self.dirty = true;
-            return;
-        };
+        let (authoritative, legacy) = self.snapshot();
         let authoritative_persisted = write_journal(MRF_SCOPED_JOURNAL_PATH, &authoritative).await;
         if !authoritative.is_empty() {
             counter!("rustfs_heal_mrf_journal_fsync_total").increment(1);
@@ -669,31 +598,11 @@ impl MrfRuntime {
             // attempts counter) changes the encoded snapshot; mark it dirty
             // either way.
             self.dirty = true;
-            let replay_key = queue_key(&intent);
-            let replayed = self
-                .retained_replay_intents
-                .get(&replay_key)
-                .is_some_and(|retained| retained.lease == intent.lease);
-            if replayed {
-                if !matches!(
-                    rustfs_common::mrf_channel::try_rearm_mrf_replay_intent(&mut intent),
-                    MrfIngressResult::Enqueued
-                ) {
-                    self.retain_replay_journal = true;
-                    self.queue.push_back(intent);
-                    self.backoff_until = Some(tokio::time::Instant::now() + self.config.admission_backoff);
-                    break;
-                }
-                self.retained_replay_intents.insert(replay_key, intent.clone());
-            }
             match submit_mrf_heal_request(manager, &intent).await {
-                // Admission removes executable work from the pending queue,
-                // but startup responsibilities still require an exact proof.
-                Ok(HealAdmissionResult::Accepted) | Ok(HealAdmissionResult::Merged) => {
-                    if replayed && let Some(anchor) = manager.durable_mrf_repair_anchor(&intent).await {
-                        self.durable_replay_anchors.push(anchor);
-                    }
-                }
+                // Accepted intents leave the pending set; the next flush persists the
+                // smaller snapshot. This is not a durable successor receipt and
+                // does not discharge the producer's existing retry hints.
+                Ok(HealAdmissionResult::Accepted) | Ok(HealAdmissionResult::Merged) => {}
                 Ok(HealAdmissionResult::Full) | Ok(HealAdmissionResult::Dropped(HealAdmissionDropReason::QueueFull)) => {
                     intent.attempts = intent.attempts.saturating_add(1);
                     if intent.attempts >= MRF_MAX_ATTEMPTS {
@@ -727,14 +636,13 @@ impl MrfRuntime {
     }
 
     fn retained_replay_journal(&self) -> bool {
-        self.retain_replay_journal || !self.retained_replay_intents.is_empty()
+        self.retain_replay_journal || !self.durable_replay_anchors.is_empty()
     }
 
     fn discharge_durable_replay_anchors(&mut self) {
         if self.durable_replay_anchors.is_empty() {
             return;
         }
-        let mut discharged_leases: HashSet<_> = self.durable_replay_anchors.iter().map(|anchor| anchor.lease).collect();
         let mut buckets: Vec<Arc<str>> = self
             .durable_replay_anchors
             .iter()
@@ -748,13 +656,6 @@ impl MrfRuntime {
                 &mut self.durable_replay_anchors,
             );
         }
-        for anchor in &self.durable_replay_anchors {
-            discharged_leases.remove(&anchor.lease);
-        }
-        let before = self.retained_replay_intents.len();
-        self.retained_replay_intents
-            .retain(|_, intent| !intent.lease.is_some_and(|lease| discharged_leases.contains(&lease)));
-        self.dirty |= self.retained_replay_intents.len() != before;
     }
 }
 
@@ -807,7 +708,6 @@ struct ReplayOutcome {
     journal_on_disk: bool,
     retain_journal_for_replay: bool,
     durable_replay_anchors: Vec<MrfDurableRepairAnchor>,
-    retained_replay_intents: HashMap<MrfQueueKey, MrfIntent>,
 }
 
 fn replay_must_retain_journal(
@@ -822,7 +722,7 @@ fn replay_must_retain_journal(
 #[derive(Clone, Copy)]
 enum ReplayCleanup {
     Legacy,
-    Committed { sequence: u64 },
+    Committed { owner: Uuid, sequence: u64 },
 }
 
 struct ReplaySource {
@@ -835,6 +735,7 @@ async fn read_replay_source(max_bytes: usize) -> Result<Option<ReplaySource>, sn
         return Ok(Some(ReplaySource {
             data: committed.payload().to_vec(),
             cleanup: ReplayCleanup::Committed {
+                owner: committed.owner(),
                 sequence: committed.sequence(),
             },
         }));
@@ -858,18 +759,20 @@ async fn read_replay_source(max_bytes: usize) -> Result<Option<ReplaySource>, sn
 async fn delete_replay_source(cleanup: ReplayCleanup, max_bytes: usize) -> bool {
     let committed_deleted = match cleanup {
         ReplayCleanup::Legacy => true,
-        ReplayCleanup::Committed { sequence } => match snapshot::delete_committed_snapshots_through(sequence, max_bytes).await {
-            Ok(deleted) => deleted,
-            Err(err) => {
-                tracing::warn!(
-                    target: "rustfs::heal::mrf",
-                    error = %err,
-                    sequence,
-                    "MRF committed replay checkpoint cleanup failed"
-                );
-                false
+        ReplayCleanup::Committed { owner, sequence } => {
+            match snapshot::delete_committed_snapshots_through(owner, sequence, max_bytes).await {
+                Ok(deleted) => deleted,
+                Err(err) => {
+                    tracing::warn!(
+                        target: "rustfs::heal::mrf",
+                        error = %err,
+                        sequence,
+                        "MRF committed replay checkpoint cleanup failed"
+                    );
+                    false
+                }
             }
-        },
+        }
     };
     committed_deleted && delete_journals().await
 }
@@ -890,7 +793,6 @@ async fn replay_into(
                 journal_on_disk: false,
                 retain_journal_for_replay: false,
                 durable_replay_anchors: Vec::new(),
-                retained_replay_intents: HashMap::new(),
             };
         }
         Err(err) => {
@@ -904,7 +806,6 @@ async fn replay_into(
                 journal_on_disk: true,
                 retain_journal_for_replay: true,
                 durable_replay_anchors: Vec::new(),
-                retained_replay_intents: HashMap::new(),
             };
         }
     };
@@ -943,7 +844,6 @@ async fn replay_into(
             }
         }
     }
-    let mut retained_replay_intents: HashMap<_, _> = queue.intents().map(|intent| (queue_key(intent), intent.clone())).collect();
 
     // Drain the replayed intents immediately; whatever the manager refuses
     // stays armed in `queue` for the consumer's retry loop.
@@ -958,7 +858,6 @@ async fn replay_into(
                 *backoff_until = Some(tokio::time::Instant::now());
                 break;
             }
-            retained_replay_intents.insert(queue_key(&intent), intent.clone());
             match submit_mrf_heal_request(manager, &intent).await {
                 Ok(HealAdmissionResult::Accepted) | Ok(HealAdmissionResult::Merged) => {
                     if let Some(anchor) = manager.durable_mrf_repair_anchor(&intent).await {
@@ -980,7 +879,6 @@ async fn replay_into(
                     break;
                 }
                 Ok(HealAdmissionResult::Dropped(_)) => {
-                    retained_replay_intents.remove(&queue_key(&intent));
                     rustfs_common::mrf_channel::release_mrf_intent(&intent);
                 }
                 Err(_) => {
@@ -1015,7 +913,6 @@ async fn replay_into(
         journal_on_disk,
         retain_journal_for_replay,
         durable_replay_anchors,
-        retained_replay_intents,
     }
 }
 
@@ -1031,7 +928,6 @@ async fn run_mrf_consumer(manager: Arc<HealManager>, mut receiver: mpsc::Receive
         journal_on_disk: false,
         retain_replay_journal: false,
         durable_replay_anchors: Vec::new(),
-        retained_replay_intents: HashMap::new(),
         backoff_until: None,
     };
 
@@ -1041,7 +937,6 @@ async fn run_mrf_consumer(manager: Arc<HealManager>, mut receiver: mpsc::Receive
     runtime.journal_on_disk = replay.journal_on_disk;
     runtime.retain_replay_journal = replay.retain_journal_for_replay;
     runtime.durable_replay_anchors = replay.durable_replay_anchors;
-    runtime.retained_replay_intents = replay.retained_replay_intents;
     // Anything still pending (e.g. the manager was full and backoff armed)
     // must be re-persisted by the next flush before replay can delete the
     // startup anchor.
@@ -1068,7 +963,17 @@ async fn run_mrf_consumer(manager: Arc<HealManager>, mut receiver: mpsc::Receive
                     );
                     return;
                 }
-                runtime.enqueue_batch(batch.drain(..));
+                for intent in batch.drain(..) {
+                    match runtime.queue.try_push_typed(intent.clone()) {
+                        MrfQueuePushResult::Enqueued => {
+                            runtime.new_since_flush += 1;
+                            runtime.dirty = true;
+                        }
+                        MrfQueuePushResult::Coalesced | MrfQueuePushResult::Rejected => {
+                            rustfs_common::mrf_channel::release_mrf_intent(&intent);
+                        }
+                    }
+                }
                 runtime.dispatch(manager.as_ref()).await;
                 if runtime.new_since_flush >= runtime.config.flush_threshold {
                     runtime.flush().await;
@@ -1220,7 +1125,6 @@ mod tests {
             journal_on_disk: true,
             retain_replay_journal: false,
             durable_replay_anchors: vec![anchor],
-            retained_replay_intents: HashMap::from([(queue_key(&intent), intent.clone())]),
             backoff_until: None,
         };
         rustfs_common::mrf_channel::note_mrf_verified_repair(MrfVerifiedRepairEvent {
@@ -1238,196 +1142,14 @@ mod tests {
             runtime.retained_replay_journal(),
             "anchor must retain the startup journal before proof is consumed"
         );
-        assert_eq!(
-            decode_journal(&runtime.snapshot().expect("retained snapshot").0).0.len(),
-            1,
-            "an admitted responsibility must remain in the successor before proof"
-        );
         runtime.discharge_durable_replay_anchors();
         assert!(
             !runtime.retained_replay_journal(),
             "matching verified proof discharges the durable replay anchor"
         );
-        assert!(runtime.dirty, "proof removal must be persisted by the next flush");
-        assert!(runtime.snapshot().expect("discharged snapshot").0.is_empty());
         rustfs_common::mrf_channel::release_mrf_intent(&intent);
     }
 
-    #[test]
-    fn runtime_snapshot_retains_admitted_replay_and_pending_successor() {
-        let accepted = intent("snapshot-bucket", "accepted", 0);
-        let pending = intent("snapshot-bucket", "pending", 2);
-        let mut queue = MrfQueue::new(4, 4096);
-        assert!(queue.try_push(pending.clone()));
-        let runtime = MrfRuntime {
-            queue,
-            config: MrfConsumerConfig::default(),
-            new_since_flush: 0,
-            dirty: true,
-            journal_on_disk: true,
-            retain_replay_journal: true,
-            durable_replay_anchors: Vec::new(),
-            retained_replay_intents: HashMap::from([
-                (queue_key(&accepted), accepted),
-                (queue_key(&pending), intent("snapshot-bucket", "pending", 0)),
-            ]),
-            backoff_until: None,
-        };
-
-        let (authoritative, legacy) = runtime.snapshot().expect("complete successor should fit");
-        for snapshot in [authoritative, legacy] {
-            let (recovered, truncated) = decode_journal(&snapshot);
-            assert_eq!(truncated, 0);
-            assert_eq!(recovered.len(), 2, "admission must not discard an unproven startup responsibility");
-            assert!(recovered.iter().any(|intent| intent.object.as_ref() == "accepted"));
-            assert!(
-                recovered
-                    .iter()
-                    .any(|intent| intent.object.as_ref() == "pending" && intent.attempts == 2)
-            );
-        }
-    }
-
-    #[test]
-    fn runtime_snapshot_preserves_anchor_when_successor_exceeds_budget_or_changes_lease() {
-        let mut retained = intent("bounded-bucket", "retained", 0);
-        assert_eq!(
-            rustfs_common::mrf_channel::try_rearm_mrf_replay_intent(&mut retained),
-            MrfIngressResult::Enqueued
-        );
-        let mut queue = MrfQueue::new(1, 4096);
-        assert!(queue.try_push(intent("bounded-bucket", "pending", 0)));
-        let mut runtime = MrfRuntime {
-            queue,
-            config: MrfConsumerConfig::default(),
-            new_since_flush: 0,
-            dirty: true,
-            journal_on_disk: true,
-            retain_replay_journal: false,
-            durable_replay_anchors: Vec::new(),
-            retained_replay_intents: HashMap::from([(queue_key(&retained), retained.clone())]),
-            backoff_until: None,
-        };
-        assert!(runtime.snapshot().is_none(), "combined count must honor the queue ceiling");
-        runtime.queue.capacity = 2;
-        runtime.queue.byte_budget = 1;
-        assert!(runtime.snapshot().is_none(), "oversized successor must not replace the startup journal");
-
-        runtime.queue = MrfQueue::new(2, 4096);
-        let mut newer = retained.clone();
-        newer.lease = None;
-        assert_eq!(
-            rustfs_common::mrf_channel::try_rearm_mrf_replay_intent(&mut newer),
-            MrfIngressResult::Enqueued
-        );
-        assert_ne!(retained.lease, newer.lease);
-        assert!(runtime.queue.try_push(newer));
-        assert!(
-            runtime.snapshot().is_none(),
-            "legacy encoding cannot conflate distinct responsibilities for one object"
-        );
-    }
-
-    #[test]
-    fn runtime_admission_reserves_replay_budget_until_verified_repair() {
-        for count_limited in [true, false] {
-            let mut retained = intent("reserved-bucket", "retained", 0);
-            assert_eq!(
-                rustfs_common::mrf_channel::try_rearm_mrf_replay_intent(&mut retained),
-                MrfIngressResult::Enqueued
-            );
-            let incarnation = Uuid::new_v4();
-            let anchor = MrfDurableRepairAnchor::from_intent(&retained, incarnation).expect("replay anchor");
-            let mut runtime = MrfRuntime {
-                queue: MrfQueue::new(if count_limited { 1 } else { 2 }, retained.estimated_bytes()),
-                config: MrfConsumerConfig::default(),
-                new_since_flush: 0,
-                dirty: false,
-                journal_on_disk: true,
-                retain_replay_journal: false,
-                durable_replay_anchors: vec![anchor],
-                retained_replay_intents: HashMap::from([(queue_key(&retained), retained.clone())]),
-                backoff_until: None,
-            };
-            if count_limited {
-                runtime.queue.byte_budget = 4096;
-            }
-            let pending = intent("reserved-bucket", "pending", 0);
-            assert_eq!(runtime.enqueue_batch([pending.clone()]), 0, "retained work consumes admission budget");
-            assert_eq!(runtime.queue.depth(), 0);
-            let (snapshot, _) = runtime.snapshot().expect("rejection must leave a writable retained snapshot");
-            let (recovered, truncated) = decode_journal(&snapshot);
-            assert_eq!(truncated, 0);
-            assert_eq!(recovered.len(), 1);
-            assert_eq!(recovered[0].object.as_ref(), "retained");
-
-            rustfs_common::mrf_channel::note_mrf_verified_repair(MrfVerifiedRepairEvent {
-                kind: retained.kind,
-                bucket: retained.bucket.clone(),
-                object: retained.object.clone(),
-                version_id: retained.version_id,
-                scope: retained.scope,
-                lease: retained.lease,
-                bucket_incarnation_id: incarnation,
-                disposition: MrfVerifiedRepairDisposition::Repaired,
-            });
-            runtime.discharge_durable_replay_anchors();
-            assert_eq!(runtime.enqueue_batch([pending]), 1, "proof must release admission capacity for retry");
-            let (snapshot, _) = runtime.snapshot().expect("new admitted work must be persistable");
-            let (recovered, truncated) = decode_journal(&snapshot);
-            assert_eq!(truncated, 0);
-            assert_eq!(recovered.len(), 1);
-            assert_eq!(recovered[0].object.as_ref(), "pending");
-        }
-    }
-
-    #[test]
-    fn runtime_admission_rejects_new_lease_without_blocking_other_successors() {
-        let mut retained = intent("lease-bucket", "retained", 0);
-        assert_eq!(
-            rustfs_common::mrf_channel::try_rearm_mrf_replay_intent(&mut retained),
-            MrfIngressResult::Enqueued
-        );
-        let incarnation = Uuid::new_v4();
-        let anchor = MrfDurableRepairAnchor::from_intent(&retained, incarnation).expect("replay anchor");
-        let mut runtime = MrfRuntime {
-            queue: MrfQueue::new(2, 4096),
-            config: MrfConsumerConfig::default(),
-            new_since_flush: 0,
-            dirty: false,
-            journal_on_disk: true,
-            retain_replay_journal: false,
-            durable_replay_anchors: vec![anchor],
-            retained_replay_intents: HashMap::from([(queue_key(&retained), retained.clone())]),
-            backoff_until: None,
-        };
-        let mut newer = retained.clone();
-        newer.lease = None;
-        assert_eq!(
-            rustfs_common::mrf_channel::try_rearm_mrf_replay_intent(&mut newer),
-            MrfIngressResult::Enqueued
-        );
-        assert_ne!(newer.lease, retained.lease);
-        assert_eq!(runtime.enqueue_batch([newer.clone(), intent("lease-bucket", "pending", 0)]), 1);
-        assert_eq!(runtime.queue.intents().next().expect("unrelated successor").object.as_ref(), "pending");
-        assert_eq!(decode_journal(&runtime.snapshot().expect("unblocked successor").0).0.len(), 2);
-
-        rustfs_common::mrf_channel::note_mrf_verified_repair(MrfVerifiedRepairEvent {
-            kind: retained.kind,
-            bucket: retained.bucket.clone(),
-            object: retained.object.clone(),
-            version_id: retained.version_id,
-            scope: retained.scope,
-            lease: retained.lease,
-            bucket_incarnation_id: incarnation,
-            disposition: MrfVerifiedRepairDisposition::Repaired,
-        });
-        runtime.discharge_durable_replay_anchors();
-        assert_eq!(runtime.enqueue_batch([newer.clone()]), 1);
-        assert!(runtime.queue.intents().any(|intent| intent.lease == newer.lease));
-        assert_eq!(decode_journal(&runtime.snapshot().expect("new lease successor").0).0.len(), 2);
-    }
-
     #[test]
     fn durable_replay_acquires_a_fresh_lease_before_manager_admission() {
         let unique = uuid::Uuid::new_v4();
@@ -1462,46 +1184,6 @@ mod tests {
         rustfs_common::mrf_channel::release_mrf_intent(&replay);
     }
 
-    #[test]
-    fn metadata_replay_canonicalization_preserves_one_bounded_responsibility() {
-        let mut legacy = intent("metadata-replay-bucket", "object", 0);
-        legacy.kind = MrfKind::MetadataCorruption;
-        let mut bytes = Vec::new();
-        assert!(encode_intent(&legacy, &mut bytes));
-        let (mut decoded, truncated) = decode_journal(&bytes);
-        assert_eq!(truncated, 0);
-        let mut replay = decoded.pop().expect("legacy metadata record");
-        assert!(replay.version_id.is_some(), "the legacy wire record carries an ignored version");
-        let original_key = queue_key(&replay);
-        let mut retained_replay_intents = HashMap::from([(original_key.clone(), replay.clone())]);
-        assert_eq!(
-            rustfs_common::mrf_channel::try_rearm_mrf_replay_intent(&mut replay),
-            MrfIngressResult::Enqueued
-        );
-        assert!(replay.version_id.is_none());
-        assert_eq!(queue_key(&replay), original_key, "rearm must not create another retained key");
-        retained_replay_intents.insert(queue_key(&replay), replay);
-        assert_eq!(retained_replay_intents.len(), 1);
-        let runtime = MrfRuntime {
-            queue: MrfQueue::new(1, 4096),
-            config: MrfConsumerConfig::default(),
-            new_since_flush: 0,
-            dirty: false,
-            journal_on_disk: true,
-            retain_replay_journal: true,
-            durable_replay_anchors: Vec::new(),
-            retained_replay_intents,
-            backoff_until: None,
-        };
-        let (snapshot, _) = runtime
-            .snapshot()
-            .expect("canonical metadata fits the original one-record budget");
-        let (recovered, truncated) = decode_journal(&snapshot);
-        assert_eq!(truncated, 0);
-        assert_eq!(recovered.len(), 1);
-        assert!(recovered[0].version_id.is_none());
-    }
-
     #[test]
     fn replay_can_arm_more_records_than_live_queue_budget() {
         let mut queue = MrfQueue::new(1, intent("bucket", "object-0", 0).estimated_bytes());
@@ -1669,6 +1351,25 @@ mod tests {
         assert_eq!(truncated, corrupt.len());
     }
 
+    #[test]
+    fn journal_rejects_unknown_version_presence_flag_even_with_valid_crc() {
+        let mut versioned = intent("rollback-bucket", "object", 0);
+        versioned.version_id = Some([9; 16]);
+        let mut buf = Vec::new();
+        assert!(encode_intent(&versioned, &mut buf));
+
+        buf[12] = 2;
+        let crc_offset = buf.len() - 4;
+        let mut hasher = crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc32IsoHdlc);
+        hasher.update(&buf[..crc_offset]);
+        let checksum = u32::try_from(hasher.finalize()).expect("CRC32 fits");
+        buf[crc_offset..].copy_from_slice(&checksum.to_le_bytes());
+
+        let (decoded, truncated) = decode_journal(&buf);
+        assert!(decoded.is_empty(), "unknown boolean encodings are not rollback-compatible payloads");
+        assert_eq!(truncated, buf.len());
+    }
+
     #[test]
     fn heal_request_mapping_follows_priority_matrix() {
         let decode = build_heal_request(&intent("b", "o", 0));

crates/heal/src/heal/mrf_queue/snapshot.rs

@@ -555,20 +555,25 @@ pub async fn inspect_local_committed_snapshot(max_bytes: usize) -> Result<Option
     read_committed(&super::journal_disks().await, max_bytes).await
 }
 
-/// Remove committed manifests whose sequence is no newer than
+/// Remove committed manifests from `owner` whose sequence is no newer than
 /// `committed_through`.
 ///
 /// Payload files are intentionally left as orphans after their manifest is
 /// removed. Readers cannot discover a payload without its matching manifest,
 /// and deleting manifests first prevents an older retained slot from becoming
 /// visible again after the newest replay has been fully discharged.
-pub async fn delete_committed_snapshots_through(committed_through: u64, max_bytes: usize) -> Result<bool, SnapshotError> {
+pub async fn delete_committed_snapshots_through(
+    owner: Uuid,
+    committed_through: u64,
+    max_bytes: usize,
+) -> Result<bool, SnapshotError> {
     let disks = super::journal_disks().await;
-    delete_committed_snapshots_through_on(&disks, committed_through, max_bytes).await
+    delete_committed_snapshots_through_on(&disks, owner, committed_through, max_bytes).await
 }
 
 async fn delete_committed_snapshots_through_on(
     disks: &[EcstoreDiskStore],
+    owner: Uuid,
     committed_through: u64,
     max_bytes: usize,
 ) -> Result<bool, SnapshotError> {
@@ -598,7 +603,7 @@ async fn delete_committed_snapshots_through_on(
                     continue;
                 }
             };
-            if manifest.sequence > committed_through {
+            if manifest.owner != owner || manifest.sequence > committed_through {
                 continue;
             }
             match EcstoreDiskAPI::compare_and_update_file(
@@ -1200,6 +1205,83 @@ mod tests {
         assert_eq!(recovered.manifest.sequence, 1);
     }
 
+    #[tokio::test]
+    async fn manifest_cas_failure_after_payload_write_keeps_previous_anchor() {
+        let root = TempDir::new().expect("test directory");
+        let store = disk(&root, "disk").await;
+        let owner = Uuid::new_v4();
+        let old = payload("old");
+        let next = payload("next");
+        let damaged_manifest = b"damaged successor manifest".to_vec();
+        commit(&store, 0, owner, 1, &old).await;
+
+        let expected_manifest = EcstoreDiskAPI::read_all(store.as_ref(), RUSTFS_META_BUCKET, MANIFEST_PATHS[1])
+            .await
+            .ok();
+        assert_eq!(
+            cas_replace(&store, PAYLOAD_PATHS[1], &next, 4096)
+                .await
+                .expect("successor payload CAS"),
+            EcstoreConditionalFileUpdate::Updated
+        );
+        install(&store, MANIFEST_PATHS[1], &damaged_manifest).await;
+
+        let manifest_update = cas_replace_expected(&store, MANIFEST_PATHS[1], expected_manifest, &manifest(owner, 2, &next))
+            .await
+            .expect("successor manifest CAS");
+        assert_eq!(manifest_update, EcstoreConditionalFileUpdate::Mismatch);
+
+        let reopened = disk(&root, "disk").await;
+        let recovered = read_committed(std::slice::from_ref(&reopened), 4096)
+            .await
+            .expect("read committed snapshot after failed successor CAS")
+            .expect("previous committed anchor");
+        assert_eq!(recovered.sequence(), 1);
+        assert_eq!(recovered.slot(), 0);
+        assert_eq!(recovered.payload(), old.as_slice());
+        assert_eq!(
+            EcstoreDiskAPI::read_all(reopened.as_ref(), RUSTFS_META_BUCKET, PAYLOAD_PATHS[1])
+                .await
+                .expect("successor payload remains non-authoritative")
+                .as_ref(),
+            next.as_slice()
+        );
+        assert_eq!(
+            EcstoreDiskAPI::read_all(reopened.as_ref(), RUSTFS_META_BUCKET, MANIFEST_PATHS[1])
+                .await
+                .expect("failed successor manifest retained")
+                .as_ref(),
+            damaged_manifest.as_slice()
+        );
+    }
+
+    #[tokio::test]
+    async fn torn_successor_on_one_replica_does_not_hide_previous_anchor_on_peer() {
+        let root = TempDir::new().expect("test directory");
+        let first = disk(&root, "first").await;
+        let second = disk(&root, "second").await;
+        let owner = Uuid::new_v4();
+        let old = payload("old");
+        let next = payload("next");
+        let damaged_manifest = b"damaged successor manifest".to_vec();
+        commit(&first, 0, owner, 1, &old).await;
+        commit(&second, 0, owner, 1, &old).await;
+        install(&first, PAYLOAD_PATHS[1], &next).await;
+        install(&first, MANIFEST_PATHS[1], &damaged_manifest).await;
+
+        let mut stats = SnapshotReadStats::default();
+        let recovered = read_committed_with_stats(&[first, second], 4096, Some(&mut stats))
+            .await
+            .expect("read committed snapshot across torn successor")
+            .expect("previous committed anchor");
+
+        assert_eq!(recovered.sequence(), 1);
+        assert_eq!(recovered.payload(), old.as_slice());
+        assert_eq!(stats.file_reads, 5);
+        assert_eq!(stats.bytes_read, (MANIFEST_LEN * 2) + (old.len() * 2) + damaged_manifest.len());
+        assert_eq!(stats.peak_file_bytes, old.len().max(next.len()).max(MANIFEST_LEN));
+    }
+
     #[tokio::test]
     async fn manifest_cas_publication_transitions_from_legacy_without_losing_anchor() {
         let root = TempDir::new().expect("test directory");
@@ -1283,7 +1365,7 @@ mod tests {
         commit(&disk, 1, owner, 4, &newer).await;
 
         assert!(
-            delete_committed_snapshots_through_on(std::slice::from_ref(&disk), 3, 4096)
+            delete_committed_snapshots_through_on(std::slice::from_ref(&disk), owner, 3, 4096)
                 .await
                 .expect("delete old manifest"),
             "old committed manifest should be removed"
@@ -1310,6 +1392,47 @@ mod tests {
         assert_eq!(recovered.payload(), newer);
     }
 
+    #[tokio::test]
+    async fn committed_cleanup_preserves_other_owner_manifests_within_sequence_window() {
+        let root = TempDir::new().expect("test directory");
+        let disk = disk(&root, "disk").await;
+        let replay_owner = Uuid::new_v4();
+        let other_owner = Uuid::new_v4();
+        let replay_payload = payload("replay-owner");
+        let other_payload = payload("other-owner");
+        commit(&disk, 0, replay_owner, 9, &replay_payload).await;
+        commit(&disk, 1, other_owner, 4, &other_payload).await;
+
+        assert!(
+            delete_committed_snapshots_through_on(std::slice::from_ref(&disk), replay_owner, 9, 4096)
+                .await
+                .expect("delete replay-owner manifest"),
+            "the matched owner manifest should be removed"
+        );
+
+        assert!(
+            matches!(
+                EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, MANIFEST_PATHS[0]).await,
+                Err(EcstoreDiskError::FileNotFound | EcstoreDiskError::VolumeNotFound)
+            ),
+            "the replay owner's manifest is reclaimed"
+        );
+        assert_eq!(
+            EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, MANIFEST_PATHS[1])
+                .await
+                .expect("other owner manifest retained")
+                .as_ref(),
+            manifest(other_owner, 4, &other_payload)
+        );
+        assert_eq!(
+            EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, PAYLOAD_PATHS[1])
+                .await
+                .expect("other owner payload retained")
+                .as_ref(),
+            other_payload
+        );
+    }
+
     #[tokio::test]
     async fn legacy_import_requires_complete_consistent_replicas() {
         let root = TempDir::new().expect("test directory");

crates/heal/src/heal/storage.rs

@@ -436,14 +436,6 @@ pub trait HealStorageAPI: Send + Sync {
         Err(Error::other("target-scoped replacement format is unsupported"))
     }
 
-    /// Whether the selected replacement set owns the pool metadata replica.
-    ///
-    /// Only a topology-aware backend may exempt a valid non-owner set. The
-    /// conservative default requires the existing repair and readback checks.
-    async fn replacement_pool_metadata_applies(&self, _opts: &HealOpts) -> Result<bool> {
-        Ok(true)
-    }
-
     /// Read target-specific physical evidence for one replacement version.
     ///
     /// This is only used by automatic replacement healing after the normal
@@ -1276,18 +1268,6 @@ impl HealStorageAPI for ECStoreHealStorage {
             .map_err(Error::Storage)
     }
 
-    async fn replacement_pool_metadata_applies(&self, opts: &HealOpts) -> Result<bool> {
-        let pool_index = opts
-            .pool
-            .ok_or_else(|| Error::other("replacement pool metadata is missing pool scope"))?;
-        let set_index = opts
-            .set
-            .ok_or_else(|| Error::other("replacement pool metadata is missing set scope"))?;
-        self.ecstore
-            .replacement_pool_metadata_applies(pool_index, set_index)
-            .map_err(Error::Storage)
-    }
-
     async fn replacement_targets_have_version(
         &self,
         bucket: &str,

crates/heal/src/heal/task.rs

@@ -45,7 +45,7 @@ use tokio::sync::RwLock;
 use tracing::{debug, error, info, warn};
 use uuid::Uuid;
 
-use super::{BUCKET_META_PREFIX, DATA_USAGE_CACHE_NAME, RUSTFS_META_BUCKET};
+use super::{BUCKET_META_PREFIX, DATA_USAGE_CACHE_NAME, POOL_META_NAME, RUSTFS_META_BUCKET};
 
 #[cfg(test)]
 pub(crate) struct OutcomeFinishTestHook {

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

@@ -340,9 +340,88 @@ impl HealTask {
             return Err(self.record_batch_failure(failure).await);
         }
 
+        if self.options.recreate_missing && !self.options.dry_run {
+            self.heal_cluster_pool_metadata().await?;
+        }
+
         Ok(())
     }
 
+    async fn heal_cluster_pool_metadata(&self) -> Result<()> {
+        let heal_opts = HealOpts {
+            recursive: false,
+            dry_run: self.options.dry_run,
+            remove: false,
+            recreate: self.options.recreate_missing,
+            scan_mode: self.options.scan_mode,
+            update_parity: self.options.update_parity,
+            no_lock: self.options.no_lock,
+            read_repair: false,
+            pool: self.options.pool_index,
+            set: self.options.set_index,
+        };
+
+        let heal_result = self
+            .await_with_control(self.storage.heal_object(RUSTFS_META_BUCKET, POOL_META_NAME, None, &heal_opts))
+            .await;
+        match heal_result {
+            Ok((result, None)) => {
+                debug!(
+                    target: "rustfs::heal::task",
+                    event = EVENT_HEAL_BUCKET_RESULT,
+                    component = LOG_COMPONENT_HEAL,
+                    subsystem = LOG_SUBSYSTEM_TASK,
+                    task_id = %self.id,
+                    bucket = RUSTFS_META_BUCKET,
+                    object = POOL_META_NAME,
+                    drives_healed = result.drives_healed(),
+                    drives_total = result.drives_reported(),
+                    result = "pool_metadata_ok",
+                    "Heal cluster pool metadata repaired"
+                );
+                self.record_result_item(result).await;
+                Ok(())
+            }
+            Ok((result, Some(err))) => {
+                self.record_result_item(result).await;
+                warn!(
+                    target: "rustfs::heal::task",
+                    event = EVENT_HEAL_BUCKET_RESULT,
+                    component = LOG_COMPONENT_HEAL,
+                    subsystem = LOG_SUBSYSTEM_TASK,
+                    task_id = %self.id,
+                    bucket = RUSTFS_META_BUCKET,
+                    object = POOL_META_NAME,
+                    result = "pool_metadata_failed",
+                    error = %err,
+                    "Heal cluster pool metadata failed"
+                );
+                Err(Error::TaskExecutionFailed {
+                    message: format!("Failed to heal cluster pool metadata: {err}"),
+                })
+            }
+            Err(Error::TaskCancelled) => Err(Error::TaskCancelled),
+            Err(Error::TaskTimeout) => Err(Error::TaskTimeout),
+            Err(err) => {
+                warn!(
+                    target: "rustfs::heal::task",
+                    event = EVENT_HEAL_BUCKET_RESULT,
+                    component = LOG_COMPONENT_HEAL,
+                    subsystem = LOG_SUBSYSTEM_TASK,
+                    task_id = %self.id,
+                    bucket = RUSTFS_META_BUCKET,
+                    object = POOL_META_NAME,
+                    result = "pool_metadata_failed",
+                    error = %err,
+                    "Heal cluster pool metadata failed"
+                );
+                Err(Error::TaskExecutionFailed {
+                    message: format!("Failed to heal cluster pool metadata: {err}"),
+                })
+            }
+        }
+    }
+
     pub(super) async fn heal_prefix(&self, bucket: &str, prefix: &str) -> Result<()> {
         debug!(
             target: "rustfs::heal::task",

crates/heal/src/heal/task/heal_erasure_set.rs

@@ -451,6 +451,11 @@ impl HealTask {
             self.source,
         )
         .with_replacement_targets(replacement_targets, is_auto_replacement.then(|| self.id.clone()))
+        .with_pool_metadata_targets(if self.options.recreate_missing && !self.options.dry_run {
+            self.heal_endpoints.clone()
+        } else {
+            Vec::new()
+        })
         .with_replacement_identity_fence(replacement_target_identities.clone())
         .with_mainline_pacer(self.mainline_pacer.clone());
 

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

@@ -66,7 +66,7 @@ mod canonical_outcome {
         assert_eq!(task.get_progress().await.objects_scanned, 2);
         assert_eq!(
             storage.heal_object_calls.lock().expect("object calls").as_slice(),
-            ["object-a", "object-b"]
+            ["object-a", "object-b", POOL_META_NAME]
         );
         assert_eq!(
             storage.listing_tokens.lock().expect("listing tokens").as_slice(),
@@ -1339,10 +1339,6 @@ struct MockStorage {
     heal_object_outcomes: Mutex<HashMap<String, VecDeque<MockHealObjectOutcome>>>,
     heal_object_receipts: Mutex<HashMap<String, VecDeque<HealObjectReceipt>>>,
     bucket_incarnation_id: Mutex<Option<Uuid>>,
-    bucket_incarnation_calls: AtomicU64,
-    bucket_incarnation_error: Mutex<Option<Error>>,
-    block_bucket_incarnation: bool,
-    bucket_incarnation_started: tokio::sync::Notify,
     bucket_incarnation_after_object_heal: Mutex<Option<Uuid>>,
     bucket_incarnation_unavailable: Mutex<bool>,
     format_no_heal_required: Mutex<bool>,
@@ -1486,7 +1482,7 @@ async fn object_heal_records_matching_positive_storage_receipt() {
     });
     let task = HealTask::from_request(
         HealRequest::object("bucket-a".to_string(), "object-a".to_string(), Some("version-a".to_string())),
-        storage.clone(),
+        storage,
     );
 
     task.execute().await.expect("mock object heal should complete");
@@ -1499,114 +1495,6 @@ async fn object_heal_records_matching_positive_storage_receipt() {
     assert_eq!(object.identity.version_id.as_deref(), Some("version-a"));
     assert!(object.identity.bucket_incarnation_id.is_some());
     assert_eq!(object.disposition, HealObjectDisposition::Repaired);
-    assert_eq!(
-        storage.bucket_incarnation_calls.load(Ordering::Relaxed),
-        1,
-        "latch the owner exactly once before repair"
-    );
-}
-
-#[tokio::test]
-async fn object_heal_owner_lookup_failure_preserves_unverified_repair() {
-    let storage = Arc::new(MockStorage {
-        bucket_incarnation_error: Mutex::new(Some(Error::other("owner metadata unavailable"))),
-        heal_object_receipts: Mutex::new(HashMap::from([(
-            "object-a".to_string(),
-            VecDeque::from([object_receipt(
-                "object-a",
-                None,
-                HealObjectDisposition::Repaired,
-                Uuid::new_v4(),
-            )]),
-        )])),
-        ..Default::default()
-    });
-    let task = HealTask::from_request(HealRequest::object("bucket-a".to_string(), "object-a".to_string(), None), storage.clone());
-
-    task.execute().await.expect("missing receipt owner must not prevent repair");
-
-    assert_eq!(storage.heal_object_calls.lock().expect("heal calls").as_slice(), ["object-a"]);
-    let outcome = task.get_outcome().await;
-    assert_eq!(outcome.counters.healed, 0);
-    assert_eq!(outcome.counters.unknown, 1);
-    assert_eq!(
-        outcome.objects.front().expect("unverified outcome").disposition,
-        HealObjectDisposition::Unknown
-    );
-}
-
-#[tokio::test]
-async fn object_heal_dry_run_skips_owner_lookup_and_positive_receipts() {
-    let storage = Arc::new(MockStorage {
-        bucket_incarnation_error: Mutex::new(Some(Error::other("dry-run must not query the receipt owner"))),
-        heal_object_receipts: Mutex::new(HashMap::from([(
-            "object-a".to_string(),
-            VecDeque::from([object_receipt(
-                "object-a",
-                None,
-                HealObjectDisposition::Repaired,
-                Uuid::new_v4(),
-            )]),
-        )])),
-        ..Default::default()
-    });
-    let mut request = HealRequest::object("bucket-a".to_string(), "object-a".to_string(), None);
-    request.options.dry_run = true;
-    let task = HealTask::from_request(request, storage.clone());
-
-    task.execute().await.expect("dry-run should complete without owner metadata");
-
-    assert!(storage.object_heal_opts.lock().expect("heal options")[0].dry_run);
-    assert_eq!(storage.bucket_incarnation_calls.load(Ordering::Relaxed), 0);
-    let outcome = task.get_outcome().await;
-    assert_eq!(outcome.counters.healed, 0);
-    assert_eq!(outcome.counters.unknown, 0);
-    assert_eq!(outcome.counters.skipped, 1);
-    assert_eq!(
-        outcome.objects.front().expect("dry-run outcome").disposition,
-        HealObjectDisposition::DryRunObserved
-    );
-}
-
-#[tokio::test(start_paused = true)]
-async fn object_heal_owner_lookup_obeys_task_timeout() {
-    let storage = Arc::new(MockStorage {
-        block_bucket_incarnation: true,
-        ..Default::default()
-    });
-    let mut request = HealRequest::object("bucket-a".to_string(), "object-a".to_string(), None);
-    request.options.timeout = Some(Duration::from_secs(5));
-    let task = HealTask::from_request(request, storage.clone());
-
-    let result = tokio::time::timeout(Duration::from_secs(60), task.execute())
-        .await
-        .expect("owner lookup must honor the task deadline");
-
-    assert!(matches!(result, Err(Error::TaskTimeout)));
-    assert!(storage.heal_object_calls.lock().expect("heal calls").is_empty());
-}
-
-#[tokio::test]
-async fn object_heal_owner_lookup_obeys_cancellation() {
-    let storage = Arc::new(MockStorage {
-        block_bucket_incarnation: true,
-        ..Default::default()
-    });
-    let mut request = HealRequest::object("bucket-a".to_string(), "object-a".to_string(), None);
-    request.options.timeout = None;
-    let task = HealTask::from_request(request, storage.clone());
-
-    let (result, ()) = tokio::time::timeout(Duration::from_secs(5), async {
-        tokio::join!(task.execute(), async {
-            storage.bucket_incarnation_started.notified().await;
-            task.cancel().await.expect("cancel pending owner lookup");
-        })
-    })
-    .await
-    .expect("cancellation must interrupt owner lookup");
-
-    assert!(matches!(result, Err(Error::TaskCancelled)));
-    assert!(storage.heal_object_calls.lock().expect("heal calls").is_empty());
 }
 
 #[tokio::test]
@@ -1956,14 +1844,6 @@ impl HealStorageAPI for MockStorage {
     }
 
     async fn bucket_incarnation_id(&self, _bucket: &str) -> Result<Option<Uuid>> {
-        self.bucket_incarnation_calls.fetch_add(1, Ordering::Relaxed);
-        self.bucket_incarnation_started.notify_one();
-        if self.block_bucket_incarnation {
-            std::future::pending::<()>().await;
-        }
-        if let Some(error) = self.bucket_incarnation_error.lock().expect("owner lookup error").take() {
-            return Err(error);
-        }
         if *self.bucket_incarnation_unavailable.lock().unwrap() {
             return Err(Error::Other("bucket incarnation unavailable".to_string()));
         }
@@ -2702,11 +2582,95 @@ async fn test_cluster_heal_visits_bucket_objects() {
 
     assert_eq!(
         storage.healed_objects.lock().unwrap().as_slice(),
-        ["object-a".to_string(), "object-b".to_string()]
+        ["object-a".to_string(), "object-b".to_string(), POOL_META_NAME.to_string()]
     );
     assert!(matches!(task.get_status().await, HealTaskStatus::Completed));
 }
 
+#[tokio::test]
+async fn cluster_recreate_heals_pool_metadata_after_user_buckets() {
+    let storage = Arc::new(MockStorage::default());
+    let request = HealRequest::new(
+        HealType::Cluster,
+        HealOptions {
+            recursive: true,
+            recreate_missing: true,
+            timeout: None,
+            ..Default::default()
+        },
+        HealPriority::Normal,
+    );
+    let task = HealTask::from_request(request, storage.clone());
+
+    task.execute()
+        .await
+        .expect("cluster recreate heal should include pool metadata");
+
+    assert_eq!(
+        storage.heal_object_calls.lock().expect("object calls").as_slice(),
+        ["object-a".to_string(), "object-b".to_string(), POOL_META_NAME.to_string()]
+    );
+    let opts = storage.object_heal_opts.lock().expect("object opts");
+    assert!(opts.last().expect("pool metadata opts").recreate);
+}
+
+#[tokio::test]
+async fn cluster_recreate_fails_when_pool_metadata_heal_fails() {
+    let storage = Arc::new(MockStorage::default());
+    storage.heal_object_outcomes.lock().expect("object outcomes").insert(
+        POOL_META_NAME.to_string(),
+        VecDeque::from([MockHealObjectOutcome::ErrOther("pool metadata missing")]),
+    );
+    let request = HealRequest::new(
+        HealType::Cluster,
+        HealOptions {
+            recursive: true,
+            recreate_missing: true,
+            timeout: None,
+            ..Default::default()
+        },
+        HealPriority::Normal,
+    );
+    let task = HealTask::from_request(request, storage.clone());
+
+    let err = task
+        .execute()
+        .await
+        .expect_err("cluster recreate heal must not hide pool metadata failure");
+
+    assert!(matches!(err, Error::TaskExecutionFailed { .. }));
+    assert_eq!(
+        storage.heal_object_calls.lock().expect("object calls").as_slice(),
+        ["object-a".to_string(), "object-b".to_string(), POOL_META_NAME.to_string()]
+    );
+}
+
+#[tokio::test]
+async fn cluster_dry_run_does_not_heal_pool_metadata() {
+    let storage = Arc::new(MockStorage::default());
+    let request = HealRequest::new(
+        HealType::Cluster,
+        HealOptions {
+            recursive: true,
+            dry_run: true,
+            recreate_missing: true,
+            timeout: None,
+            ..Default::default()
+        },
+        HealPriority::Normal,
+    );
+    let task = HealTask::from_request(request, storage.clone());
+
+    task.execute()
+        .await
+        .expect("dry-run cluster heal should preserve existing coverage");
+
+    assert_eq!(
+        storage.heal_object_calls.lock().expect("object calls").as_slice(),
+        ["object-a".to_string(), "object-b".to_string()]
+    );
+}
+
 #[tokio::test]
 async fn object_heal_skips_dangling_delete_grace_without_failing_task() {
     let storage = Arc::new(MockStorage {

crates/heal/tests/mrf_pipeline_test.rs

@@ -626,8 +626,7 @@ fn mrf_successor_flush_child_process_fixture() {
             }),
         ));
         mrf_queue::spawn_mrf_consumer(manager.clone());
-        let mut expected_successor = journal_record(1, "successor-bucket", "second-object", None, 2);
-        expected_successor.extend(journal_record(1, "successor-bucket", "first-object", None, 0));
+        let expected_successor = journal_record(1, "successor-bucket", "second-object", None, 2);
         let flushed = wait_until(Duration::from_secs(10), || async {
             manager.operations_snapshot().await.queued_by_source.mrf == 1
                 && journal_matches_on_all_disks(&disk_paths, SCOPED_JOURNAL_REL, &expected_successor)
@@ -674,8 +673,7 @@ fn mrf_successor_flush_waiting_child_process_fixture() {
             }),
         ));
         mrf_queue::spawn_mrf_consumer(manager.clone());
-        let mut expected_successor = journal_record(1, "service-kill-bucket", "second-object", None, 2);
-        expected_successor.extend(journal_record(1, "service-kill-bucket", "first-object", None, 0));
+        let expected_successor = journal_record(1, "service-kill-bucket", "second-object", None, 2);
         let flushed = wait_until(Duration::from_secs(10), || async {
             manager.operations_snapshot().await.queued_by_source.mrf == 1
                 && journal_matches_on_all_disks(&disk_paths, SCOPED_JOURNAL_REL, &expected_successor)
@@ -715,8 +713,7 @@ fn mrf_authoritative_fsync_waiting_child_process_fixture() {
         write_journal_path_to_disks(&disk_paths, SCOPED_JOURNAL_REL, &startup);
         write_journal_path_to_disks(&disk_paths, JOURNAL_REL, &startup);
 
-        let mut successor = journal_record(1, "fsync-kill-bucket", "second-object", None, 2);
-        successor.extend(journal_record(1, "fsync-kill-bucket", "first-object", None, 0));
+        let successor = journal_record(1, "fsync-kill-bucket", "second-object", None, 2);
         write_journal_path_to_disks_synced(&disk_paths, SCOPED_JOURNAL_REL, &successor);
         assert!(
             journal_matches_on_all_disks(&disk_paths, SCOPED_JOURNAL_REL, &successor)
@@ -773,8 +770,9 @@ async fn journal_replay_retains_child_process_anchor_when_manager_is_full() {
     );
 }
 
-/// A successor flush must preserve both pending work and accepted work whose
-/// repair has not been proven when the process restarts.
+/// If a process crashes after flushing a smaller successor snapshot but before
+/// deleting the startup anchor, the restarted process must replay the
+/// successor tail rather than losing it or merging it with stale records.
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
 #[serial]
 async fn journal_replay_survives_successor_flush_before_delete() {
@@ -789,8 +787,7 @@ async fn journal_replay_survives_successor_flush_before_delete() {
     assert_eq!(status.code(), Some(78), "child process did not reach the successor flush boundary");
 
     let (disk_paths, storage) = heal_env_at(Some(temp_dir.path())).await;
-    let mut expected_successor = journal_record(1, "successor-bucket", "second-object", None, 2);
-    expected_successor.extend(journal_record(1, "successor-bucket", "first-object", None, 0));
+    let expected_successor = journal_record(1, "successor-bucket", "second-object", None, 2);
     assert!(
         journal_matches_on_all_disks(&disk_paths, SCOPED_JOURNAL_REL, &expected_successor),
         "restarted process must see the pending successor snapshot"
@@ -798,11 +795,11 @@ async fn journal_replay_survives_successor_flush_before_delete() {
 
     let restarted = make_manager(storage);
     let replayed = mrf_queue::replay_journal_once(&restarted).await;
-    assert_eq!(replayed, 2, "restart must replay both the admitted and pending responsibilities");
+    assert_eq!(replayed, 1, "restart after successor flush must replay only the still-pending tail");
     assert_eq!(
         restarted.operations_snapshot().await.queued_by_source.mrf,
-        2,
-        "both unproven successor responsibilities must be accepted after restart"
+        1,
+        "the successor tail must be accepted after restart"
     );
     assert!(
         disk_paths.iter().all(|path| {
@@ -814,8 +811,8 @@ async fn journal_replay_survives_successor_flush_before_delete() {
 }
 
 /// A service-style hard kill after successor flush must be equivalent to a
-/// crash at the flush-before-delete boundary: restart must recover every
-/// unproven responsibility from the successor snapshot.
+/// crash at the flush-before-delete boundary: restart may replay the smaller
+/// successor snapshot, but must not lose or merge stale startup records.
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
 #[serial]
 #[cfg(unix)]
@@ -843,8 +840,7 @@ async fn journal_replay_survives_service_kill_after_successor_flush() {
     assert!(!status.success(), "child fixture must be terminated instead of exiting cleanly");
 
     let (disk_paths, storage) = heal_env_at(Some(temp_dir.path())).await;
-    let mut expected_successor = journal_record(1, "service-kill-bucket", "second-object", None, 2);
-    expected_successor.extend(journal_record(1, "service-kill-bucket", "first-object", None, 0));
+    let expected_successor = journal_record(1, "service-kill-bucket", "second-object", None, 2);
     assert!(
         journal_matches_on_all_disks(&disk_paths, SCOPED_JOURNAL_REL, &expected_successor),
         "restarted process must see the successor snapshot produced before the kill"
@@ -852,11 +848,11 @@ async fn journal_replay_survives_service_kill_after_successor_flush() {
 
     let restarted = make_manager(storage);
     let replayed = mrf_queue::replay_journal_once(&restarted).await;
-    assert_eq!(replayed, 2, "service-kill restart must preserve every unproven responsibility");
+    assert_eq!(replayed, 1, "restart after service kill must replay only the still-pending tail");
     assert_eq!(
         restarted.operations_snapshot().await.queued_by_source.mrf,
-        2,
-        "both unproven responsibilities must be accepted after service kill restart"
+        1,
+        "the successor tail must be accepted after service kill restart"
     );
     assert!(
         disk_paths.iter().all(|path| {
@@ -868,9 +864,9 @@ async fn journal_replay_survives_service_kill_after_successor_flush() {
 }
 
 /// A hard kill between the authoritative successor fsync and the legacy mirror
-/// rewrite must prefer the canonical successor over the stale legacy startup
-/// epoch while retaining every unproven responsibility. New readers must not
-/// merge epochs, while the old mirror remains crash-visible.
+/// rewrite must prefer the canonical successor tail over the stale legacy
+/// startup epoch. This models the mixed-version boundary conservatively: new
+/// readers must not merge epochs, while the old mirror remains crash-visible.
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
 #[serial]
 #[cfg(unix)]
@@ -898,8 +894,7 @@ async fn journal_replay_survives_sigkill_after_authoritative_successor_fsync_bef
     assert!(!status.success(), "child fixture must be terminated instead of exiting cleanly");
 
     let (disk_paths, storage) = heal_env_at(Some(temp_dir.path())).await;
-    let mut expected_successor = journal_record(1, "fsync-kill-bucket", "second-object", None, 2);
-    expected_successor.extend(journal_record(1, "fsync-kill-bucket", "first-object", None, 0));
+    let expected_successor = journal_record(1, "fsync-kill-bucket", "second-object", None, 2);
     let stale_startup = {
         let mut startup = journal_record(1, "fsync-kill-bucket", "first-object", None, 0);
         startup.extend(journal_record(1, "fsync-kill-bucket", "second-object", None, 0));
@@ -916,11 +911,11 @@ async fn journal_replay_survives_sigkill_after_authoritative_successor_fsync_bef
 
     let restarted = make_manager(storage);
     let replayed = mrf_queue::replay_journal_once(&restarted).await;
-    assert_eq!(replayed, 2, "new reader must recover every responsibility in the authoritative successor");
+    assert_eq!(replayed, 1, "new reader must replay only the authoritative successor tail");
     assert_eq!(
         restarted.operations_snapshot().await.queued_by_source.mrf,
-        2,
-        "both responsibilities must be accepted after the fsync-boundary restart"
+        1,
+        "the successor tail must be accepted after the fsync-boundary restart"
     );
     assert!(
         disk_paths.iter().all(|path| {

crates/kms/src/backends/local.rs

@@ -1233,8 +1233,6 @@ impl LocalKmsClient {
     async fn decode_stored_key(&self, key_id: &str) -> Result<(StoredMasterKey, Vec<u8>)> {
         let key_path = self.master_key_path(key_id)?;
         if !fs::try_exists(&key_path).await? {
-            // A missing key is a caller error only while its storage directory is available.
-            let _ = fs::read_dir(&self.config.key_dir).await?;
             return Err(KmsError::key_not_found(key_id));
         }
 
@@ -2097,7 +2095,11 @@ impl KmsBackend for LocalKmsBackend {
         let _write_guard = self.client.lock_key_for_write(key_id).await;
 
         // First, load the key from disk to get the master key
-        let mut master_key = self.client.load_master_key(key_id).await?;
+        let mut master_key = self
+            .client
+            .load_master_key(key_id)
+            .await
+            .map_err(|_| KmsError::key_not_found(format!("Key {key_id} not found")))?;
 
         let (deletion_date_str, deletion_date_dt) = if request.force_immediate.unwrap_or(false) {
             // Tombstone first: mark the record Deleted before removing the
@@ -2200,7 +2202,11 @@ impl KmsBackend for LocalKmsBackend {
         let _write_guard = self.client.lock_key_for_write(key_id).await;
 
         // Load the key from disk to get the master key
-        let mut master_key = self.client.load_master_key(key_id).await?;
+        let mut master_key = self
+            .client
+            .load_master_key(key_id)
+            .await
+            .map_err(|_| KmsError::key_not_found(format!("Key {key_id} not found")))?;
 
         if master_key.status != KeyStatus::PendingDeletion {
             return Err(KmsError::invalid_key_state(format!("Key {key_id} is not pending deletion")));
@@ -2959,146 +2965,6 @@ mod tests {
         assert!(matches!(error, KmsError::InvalidKey { .. }));
     }
 
-    #[tokio::test]
-    async fn delete_key_preserves_directory_io_error() {
-        let (client, temp_dir) = create_dev_mode_client().await;
-        client.create_key("existing-key", "AES_256", None).await.expect("create key");
-        let backend = LocalKmsBackend { client };
-        let offline_dir = TempDir::new().expect("create offline directory");
-        let offline_key_dir = offline_dir.path().join("keys");
-        fs::rename(temp_dir.path(), &offline_key_dir)
-            .await
-            .expect("move key directory offline");
-        fs::write(temp_dir.path(), b"not a directory")
-            .await
-            .expect("replace key directory with a file");
-
-        let error = backend
-            .delete_key(DeleteKeyRequest {
-                key_id: "existing-key".to_string(),
-                ..Default::default()
-            })
-            .await
-            .expect_err("unreadable storage must prevent scheduling deletion");
-
-        fs::remove_file(temp_dir.path()).await.expect("remove replacement file");
-        fs::rename(&offline_key_dir, temp_dir.path())
-            .await
-            .expect("restore key directory");
-        assert!(matches!(error, KmsError::IoError { .. }), "got {error:?}");
-        let key = backend
-            .client
-            .load_master_key("existing-key")
-            .await
-            .expect("read retained key");
-        assert_eq!(key.status, KeyStatus::Active, "failed deletion must not mutate key state");
-    }
-
-    #[tokio::test]
-    async fn cancel_key_deletion_preserves_directory_io_error() {
-        let (client, temp_dir) = create_dev_mode_client().await;
-        client.create_key("existing-key", "AES_256", None).await.expect("create key");
-        let backend = LocalKmsBackend { client };
-        backend
-            .delete_key(DeleteKeyRequest {
-                key_id: "existing-key".to_string(),
-                ..Default::default()
-            })
-            .await
-            .expect("schedule key deletion");
-        let offline_dir = TempDir::new().expect("create offline directory");
-        let offline_key_dir = offline_dir.path().join("keys");
-        fs::rename(temp_dir.path(), &offline_key_dir)
-            .await
-            .expect("move key directory offline");
-        fs::write(temp_dir.path(), b"not a directory")
-            .await
-            .expect("replace key directory with a file");
-
-        let error = backend
-            .cancel_key_deletion(CancelKeyDeletionRequest {
-                key_id: "existing-key".to_string(),
-            })
-            .await
-            .expect_err("unreadable storage must prevent cancelling deletion");
-
-        fs::remove_file(temp_dir.path()).await.expect("remove replacement file");
-        fs::rename(&offline_key_dir, temp_dir.path())
-            .await
-            .expect("restore key directory");
-        assert!(matches!(error, KmsError::IoError { .. }), "got {error:?}");
-        let key = backend
-            .client
-            .load_master_key("existing-key")
-            .await
-            .expect("read retained key");
-        assert_eq!(
-            key.status,
-            KeyStatus::PendingDeletion,
-            "failed cancellation must retain the deletion state"
-        );
-    }
-
-    #[tokio::test]
-    async fn test_load_master_key_missing_key_remains_not_found() {
-        let (client, _temp_dir) = create_dev_mode_client().await;
-
-        let error = client
-            .load_master_key("missing-key")
-            .await
-            .expect_err("missing key must fail");
-
-        assert!(matches!(error, KmsError::KeyNotFound { key_id } if key_id == "missing-key"));
-    }
-
-    #[tokio::test]
-    async fn test_load_master_key_unavailable_directory_is_io_error() {
-        let (client, temp_dir) = create_dev_mode_client().await;
-        client.create_key("existing-key", "AES_256", None).await.expect("create key");
-        let offline_dir = TempDir::new().expect("create offline directory");
-        let offline_key_dir = offline_dir.path().join("keys");
-        fs::rename(temp_dir.path(), &offline_key_dir)
-            .await
-            .expect("move key directory offline");
-
-        let error = client
-            .load_master_key("existing-key")
-            .await
-            .expect_err("unavailable key directory must fail");
-
-        fs::rename(&offline_key_dir, temp_dir.path())
-            .await
-            .expect("restore key directory");
-        assert!(matches!(error, KmsError::IoError { .. }), "got {error:?}");
-        let key = client.load_master_key("existing-key").await.expect("read restored key");
-        assert_eq!(key.key_id, "existing-key");
-    }
-
-    #[tokio::test]
-    async fn test_load_master_key_directory_replaced_by_file_is_io_error() {
-        let (client, temp_dir) = create_dev_mode_client().await;
-        client.create_key("existing-key", "AES_256", None).await.expect("create key");
-        let offline_dir = TempDir::new().expect("create offline directory");
-        let offline_key_dir = offline_dir.path().join("keys");
-        fs::rename(temp_dir.path(), &offline_key_dir)
-            .await
-            .expect("move key directory offline");
-        fs::write(temp_dir.path(), b"not a directory")
-            .await
-            .expect("replace key directory with a file");
-
-        let error = client
-            .load_master_key("existing-key")
-            .await
-            .expect_err("a file in place of the key directory must fail");
-
-        fs::remove_file(temp_dir.path()).await.expect("remove replacement file");
-        fs::rename(&offline_key_dir, temp_dir.path())
-            .await
-            .expect("restore key directory");
-        assert!(matches!(error, KmsError::IoError { .. }), "got {error:?}");
-    }
-
     #[tokio::test]
     async fn test_load_master_key_accepts_legacy_rfc3339_timestamp() {
         let (client, _temp_dir) = create_dev_mode_client().await;

docs/testing/ci-gates.md

@@ -43,18 +43,17 @@ Promotion rule: never promote a report-only lane to required from one green run.
 | PR, non-doc change | `End-to-End Tests` | `ci.yml` `e2e-tests` | Report-only | `cargo nextest run --profile e2e-smoke -p e2e_test`, then `./scripts/e2e-run.sh ./target/debug/rustfs <data-dir>`; membership guards `scripts/check_test_wiring.py --check-profile e2e-smoke <listing.json>` and `scripts/check_security_smoke_count.sh check <listing.json>` |
 | PR, non-doc change | `S3 Implemented Tests` | `ci.yml` `s3-implemented-tests` | Report-only | build `rustfs`, then `scripts/s3-tests/run.sh` with the job's `DEPLOY_MODE` / `TEST_MODE` / `MAXFAIL` env |
 | PR, non-doc change | `S3 Lifecycle Behavior Tests` | `ci.yml` `s3-lifecycle-behavior-tests` | Report-only | `scripts/s3-tests/run.sh` with the job's accelerated-scanner env |
-| PR to `main` or `release` touching `paths` in `audit.yml` | `Cargo Deny`, `Workflow Pin Report`, `Dependency Review` | `audit.yml` `cargo-deny`, `workflow-pin-report`, `dependency-review` | Report-only | `cargo deny check`; `scripts/security/check_workflow_pins.sh` |
-| Push to `main` or `release` touching `paths` in `audit.yml` | `Cargo Deny`, `Workflow Pin Report` | `audit.yml` `cargo-deny`, `workflow-pin-report` | Report-only | `cargo deny check`; `scripts/security/check_workflow_pins.sh` |
+| PR touching `paths` in `audit.yml` | `Cargo Deny`, `Workflow Pin Report`, `Dependency Review` | `audit.yml` `cargo-deny`, `workflow-pin-report`, `dependency-review` | Report-only | `cargo deny check`; `scripts/security/check_workflow_pins.sh` |
 | PR touching `paths` in `architecture-migration-rules.yml` | `Architecture Migration Rules` | `architecture-migration-rules.yml` `architecture-migration-rules` | Report-only | `scripts/check_architecture_migration_rules.sh` |
 | PR touching `paths` in `nix.yml` | `Nix Build & Check` | `nix.yml` `nix-validation` | Report-only | `nix flake check` |
 | PR touching `paths` in `fuzz.yml` | `Build Fuzz Harness`, `Smoke / <target>` | `fuzz.yml` `fuzz-build`, `pr-fuzz-smoke` | Report-only | `MAX_TOTAL_TIME=60 ./scripts/fuzz/run.sh` |
 | PR touching `paths` in `windows-filesystem.yml` | `Rename Safety` | `windows-filesystem.yml` `rename-safety` | Report-only | the `cargo test -p rustfs-ecstore --lib <filter>` commands in the job, on Windows |
 | PR touching `paths` in `coverage.yml` | `Workspace line coverage` | `coverage.yml` `coverage` | Report-only | `make coverage`; `python3 scripts/check_security_coverage.py target/llvm-cov/coverage.json` |
-| PR touching `paths` in `e2e-upgrade.yml` | `Direct upgrade from the previous release`, `Mixed-version rolling upgrade from the previous release`, `Bucket configuration survives the upgrade`, `Rollback reads current bucket metadata`, `ODM configuration recovery after rc.5 rollback`, `Multipart layouts survive the rc.5 upgrade`, `rc.5 multipart replication baseline` | `e2e-upgrade.yml` `upgrade` matrix | Report-only | the `cargo test --locked -p e2e_test` command in the job with `RUSTFS_UPGRADE_SOURCE_BINARY` pointing at the pinned previous release (`UPGRADE_SOURCE_VERSION`) |
+| PR touching `paths` in `e2e-upgrade.yml` | `Direct upgrade from the previous release`, `Mixed-version rolling upgrade from the previous release`, `Bucket configuration survives the upgrade`, `Rollback reads current bucket metadata` | `e2e-upgrade.yml` `upgrade` matrix | Report-only | the `cargo test --locked -p e2e_test` command in the job with `RUSTFS_UPGRADE_SOURCE_BINARY` pointing at the pinned previous release (`UPGRADE_SOURCE_VERSION`) |
 | PR touching `paths` in `oidc-keycloak.yml` | `OIDC Keycloak live gate` | `oidc-keycloak.yml` `oidc-keycloak-live` | Report-only | `cargo build --locked -p rustfs --bin rustfs`, then `bash scripts/test/oidc_keycloak_live.sh ./target/debug/rustfs` |
 | PR touching `paths` in `targets-integration.yml` | `PostgreSQL, MySQL, AMQP, and NATS` | `targets-integration.yml` `targets-live` | Report-only | start the containers as in the job, export the `RUSTFS_TEST_*` DSNs, then the job's `cargo test --locked -p rustfs-targets --test <name> -- --ignored --test-threads=1` commands |
 | PR limited to main-CI-excluded paths | `Quick Checks`, `Test and Lint` | `ci-docs-only.yml` `quick-checks`, `test-and-lint` | Required | `git diff --check`; `make doc-paths-check`; `scripts/check_no_planning_docs.sh` |
-| `merge_group`; push to `main` or `release` | `End-to-End Tests (full merge gate)` | `ci.yml` `e2e-full` | Report-only | `cargo nextest run --profile e2e-full -p e2e_test` |
+| `merge_group`; push to `main` | `End-to-End Tests (full merge gate)` | `ci.yml` `e2e-full` | Report-only | `cargo nextest run --profile e2e-full -p e2e_test` |
 
 e2e filters live in `.config/nextest.toml`; extend a profile instead of adding a second selector. Before a profile runs, `scripts/check_test_wiring.py` compares its listing to the committed digest in `.config/e2e-<profile>-selection.txt`, so a silent test drop fails closed.
 
@@ -63,7 +62,7 @@ cost. `data_usage_test` runs in the PR `e2e-smoke` lane so changes that affect
 authoritative scanner usage publication, quota-visible usage, or admin usage
 snapshots get an end-to-end signal before merge review. `heal_erasure_disk_rebuild_test`
 runs in `e2e-full` so core erasure heal rebuild regressions are caught no later
-than the merge queue or `main`/`release` push lane; it also remains in `e2e-nightly` with
+than the merge queue or `main` push lane; it also remains in `e2e-nightly` with
 the serialized cluster fault-domain suites for scheduled soak signal.
 
 ## Scheduled validation
@@ -86,7 +85,7 @@ Scheduled lanes never block a PR. Their workflow-local gate fails the run, sched
 | `mint.yml` (weekly) | `mint` | report-only by design; per-suite PASS/FAIL/NA and raw `log.json` | yes | pinned Docker sequence in the workflow |
 | `coverage.yml` (weekly) | `coverage` | report-only trend; lcov and JSON artifact | yes | `make coverage` |
 | `runner-hygiene.yml` (monthly) | `check-ephemerality` | runner ephemerality | yes | dispatch |
-| `e2e-upgrade.yml` (weekly) | `upgrade` (7-case matrix) | upgrade and rollback gate; server logs | no | see the PR row |
+| `e2e-upgrade.yml` (weekly) | `upgrade` (4-case matrix) | upgrade and rollback gate; server logs | no | see the PR row |
 | `oidc-keycloak.yml` (weekly) | `oidc-keycloak-live` | live OIDC gate | no | see the PR row |
 | `targets-integration.yml` (nightly) | `targets-live` | live target gate; container logs | no | see the PR row |
 | `scheduled-validation-freshness.yml` (nightly) | `check-freshness` | fails on a never-created or stale schedule | n/a | dispatch |

scripts/check_test_wiring.py

@@ -956,6 +956,9 @@ def scanner_heal_oracle_names(root: Path) -> tuple[str, ...]:
     names = set()
     for case_id, requirement in cases.items():
         require(isinstance(case_id, str) and case_id, "invalid scanner/heal case identity")
+        lane = requirement.get("lane")
+        require(isinstance(lane, str) and re.fullmatch(r"[a-z0-9-]+", lane) is not None,
+                f"invalid nextest profile lane for {case_id}")
         oracle = requirement.get("oracle")
         require(isinstance(oracle, str) and oracle.endswith(".json"), f"invalid oracle for {case_id}")
         path = Path(oracle)
@@ -1337,9 +1340,12 @@ def validate_release_bundle_artifact(bundle_path: Path, source_revision: str, ga
         evidence_integer(evidence.get("duration_seconds"), f"{gate}.{field}.duration_seconds", 1, 86400)
     if gate in ("G03", "G09", "R-L"):
         versions = evidence.get("versions")
-        require(isinstance(versions, list) and
-                len({version for version in versions if isinstance(version, str) and version.strip()}) >= 2,
-                f"{gate}.{field} requires mixed-version evidence")
+        revisions = {
+            version for version in versions
+            if isinstance(version, str) and re.fullmatch(r"[0-9a-f]{40}", version) is not None
+        } if isinstance(versions, list) else set()
+        require(len(revisions) >= 2, f"{gate}.{field} requires at least two 40-hex mixed-version revisions")
+        require(source_revision in revisions, f"{gate}.{field} must include the tested source revision")
     if gate in ("G04", "G07", "R-E", "R-L"):
         crash_points = evidence.get("crash_points")
         require(isinstance(crash_points, list) and crash_points,
@@ -1667,6 +1673,16 @@ class SelfTests(unittest.TestCase):
         finish_scanner_heal_receipt(run_dir, 0, root)
         return root, run_dir
 
+    def test_scanner_heal_case_lane_is_required_for_runner_profile(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root, _ = self.scanner_heal_fixture(Path(tmp))
+            registry_path = root / ".config/scanner-heal-required-tests.json"
+            registry = read_json(registry_path)
+            del registry["cases"]["ec84-target-drive-restart"]["lane"]
+            write_json(registry_path, registry)
+            with self.assertRaisesRegex(ValueError, "invalid nextest profile lane for ec84-target-drive-restart"):
+                scanner_heal_oracle_names(root)
+
     def scanner_heal_release_bundle_fixture(self, directory: Path) -> tuple[Path, Path]:
         """Parser fixtures only; the bundle is not runtime evidence."""
         root, _ = self.scanner_heal_fixture(directory)
@@ -1704,7 +1720,7 @@ class SelfTests(unittest.TestCase):
                     evidence["duration_seconds"] = duration
                 evidence["finished_at"] = (started + timedelta(seconds=duration)).isoformat().replace("+00:00", "Z")
                 if gate in ("G03", "G09", "R-L"):
-                    evidence["versions"] = ["previous", "candidate"]
+                    evidence["versions"] = ["a" * 40, source_revision]
                 if gate in ("G04", "G07", "R-E", "R-L"):
                     evidence["crash_points"] = ["before-commit"]
                 if gate == "G14" and field == "ec8_4_evidence":
@@ -1769,6 +1785,8 @@ class SelfTests(unittest.TestCase):
             ("duration", "P3", "two_hour_pressure_measurement", lambda item: item.update({"duration_seconds": 7199}), "two hours"),
             ("profile", "P1", "profile_evidence", lambda item: item.pop("resolved_samples"), "resolved_samples"),
             ("versions", "G09", "mixed_version_reader_evidence", lambda item: item.update({"versions": [1, 2]}), "mixed-version"),
+            ("placeholder-versions", "G09", "mixed_version_writer_evidence", lambda item: item.update({"versions": ["previous", "candidate"]}), "40-hex"),
+            ("missing-tested-revision", "R-L", "migration_gap_evidence", lambda item: item.update({"versions": ["a" * 40, "c" * 40]}), "tested source revision"),
         ):
             with self.subTest(fault=fault), tempfile.TemporaryDirectory() as tmp:
                 root, bundle = self.scanner_heal_release_bundle_fixture(Path(tmp))

scripts/error-other-format-baseline.txt

@@ -33,7 +33,7 @@
 1|crates/ecstore/src/disk/mod.rs
 5|crates/ecstore/src/erasure/codec/bridge.rs
 1|crates/ecstore/src/erasure/coding/decode_reader.rs
-8|crates/ecstore/src/erasure/coding/encode.rs
+10|crates/ecstore/src/erasure/coding/encode.rs
 25|crates/ecstore/src/erasure/coding/erasure.rs
 3|crates/ecstore/src/layout/disks_layout.rs
 2|crates/ecstore/src/layout/endpoint.rs

scripts/run_scanner_heal_evidence_case.sh

@@ -4,7 +4,7 @@ set -euo pipefail
 
 ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
 PYTHON_BIN="${RUSTFS_PYTHON_BIN:-python3}"
-PROFILE="e2e-nightly"
+PROFILE=""
 CASE_ID="background-target-crash"
 RUN_DIR=""
 PLAN_ONLY=0
@@ -18,7 +18,7 @@ then validate the produced receipt, nextest listing, JUnit, and case oracle.
 
 Options:
   --case CASE          Registry case to run (default: background-target-crash)
-  --profile PROFILE   Nextest profile to use (default: e2e-nightly)
+  --profile PROFILE   Nextest profile to use (default: registry lane)
   --run-dir DIR       New evidence directory (default: target/scanner-heal-evidence/CASE-TIMESTAMP)
   --plan-only         Validate registry selection and print the exact filter without running cargo
   --self-test         Run lightweight CLI/registry checks without building Rust
@@ -31,6 +31,18 @@ port allocator when the default 20000..30000 test range is unavailable.
 USAGE
 }
 
+case_ids() {
+    "$PYTHON_BIN" - "$ROOT/.config/scanner-heal-required-tests.json" <<'PY'
+import json
+import pathlib
+import sys
+
+registry = json.loads(pathlib.Path(sys.argv[1]).read_text())
+for case_id in sorted(registry["cases"]):
+    print(case_id)
+PY
+}
+
 case_field() {
     local case_id="$1"
     local field="$2"
@@ -109,20 +121,24 @@ PY
 }
 
 run_self_test() {
-    local filter
-    filter="$(test_filter_for background-target-crash)"
-    case "$filter" in
-        *background_target_crash*) ;;
-        *)
-            echo "self-test failed: crash case filter missing" >&2
-            return 1
-            ;;
-    esac
     if "$0" --case release --plan-only >/dev/null 2>&1; then
         echo "self-test failed: release pseudo-case must not be runnable" >&2
         return 1
     fi
-    "$0" --case background-target-crash --plan-only >/dev/null
+    local case_id
+    while IFS= read -r case_id; do
+        local expected_filter expected_profile plan
+        expected_filter="$(test_filter_for "$case_id")"
+        expected_profile="$(case_field "$case_id" lane)"
+        plan="$("$0" --case "$case_id" --plan-only)"
+        if [[ "$plan" != *"case=$case_id"* ]] ||
+            [[ "$plan" != *"profile=$expected_profile"* ]] ||
+            [[ "$plan" != *"filter=$expected_filter"* ]] ||
+            [[ "$plan" != *"run_dir=$ROOT/target/scanner-heal-evidence/$case_id-"* ]]; then
+            echo "self-test failed: registry case plan mismatch for $case_id" >&2
+            return 1
+        fi
+    done < <(case_ids)
 }
 
 while [[ $# -gt 0 ]]; do
@@ -166,6 +182,9 @@ fi
 
 case_field "$CASE_ID" name >/dev/null
 TEST_FILTER="$(test_filter_for "$CASE_ID")"
+if [[ -z "$PROFILE" ]]; then
+    PROFILE="$(case_field "$CASE_ID" lane)"
+fi
 case "$CASE_ID" in
   background-target-crash|background-target-restart)
     export RUSTFS_HEAL_CHAOS_OBJECT_COUNT="${RUSTFS_HEAL_CHAOS_OBJECT_COUNT:-64}"