fix: close replacement and protocol validation gaps

Add focused oracles for transient quorum results that carry a matching receipt and for mixed grace plus repaired receipt batches. Only the repaired object may produce positive proof.

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=f0c78fdb93471575d9a64c5c46eae6c806bdd0bc10a6e33d7fb574aabd8db5a3
-sha256-linux=03ed7016cab672de9320e31375a0358eceacb4408b0e79cf063614fa7c878b87
+sha256-darwin=cca6d0bc1487f472dc354bbffcc2b5c7410edfccb499c6bce6638113fbe6bbec
+sha256-linux=3b71936f6f4ea0cca3b5db6c2f387c990dddda7e96c982315eb42e3b2e6c92b8

.config/e2e-nightly-selection.txt

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

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

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

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

@@ -13,6 +13,7 @@
       "min_objects": 9,
       "max_objects": 65,
       "topology": {"nodes": 4, "drives_per_node": 1},
+      "erasure": {"data_blocks": 2, "parity_blocks": 2},
       "scope": "Target process restart, exact unversioned S3 bodies and replacement-disk shards; not power loss or EC8+4."
     },
     "background-target-crash": {
@@ -27,6 +28,7 @@
       "min_objects": 9,
       "max_objects": 65,
       "topology": {"nodes": 4, "drives_per_node": 1},
+      "erasure": {"data_blocks": 2, "parity_blocks": 2},
       "scope": "Target process killed during partial background rebuild, real unclean-shutdown marker, exact unversioned S3 bodies and replacement-disk shards; not power loss or EC8+4."
     },
     "ec84-target-drive-restart": {
@@ -42,6 +44,38 @@
       "max_objects": 5,
       "topology": {"nodes": 3, "drives_per_node": 4},
       "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": {
+      "gate": "G14",
+      "task": "W21",
+      "lane": "e2e-nightly",
+      "suite": "e2e_test",
+      "name": "heal_erasure_disk_rebuild_test::tests::test_cluster_root_heal_recovers_ec84_shards_after_background_target_restart",
+      "oracle": "background-target-restart-ec8-4.json",
+      "evidence": "process-restart",
+      "unclean_shutdown_marker": false,
+      "min_objects": 9,
+      "max_objects": 65,
+      "topology": {"nodes": 3, "drives_per_node": 4},
+      "erasure": {"data_blocks": 8, "parity_blocks": 4},
+      "erasure_set_drive_count": 12,
+      "scope": "Target process restart during partial background rebuild on a single 3x4 EC8+4 set; exact unversioned S3 bodies and replacement-drive shards; not power loss, multi-set, or multi-pool."
+    },
+    "background-target-crash-ec8-4": {
+      "gate": "G14",
+      "task": "W21",
+      "lane": "e2e-nightly",
+      "suite": "e2e_test",
+      "name": "heal_erasure_disk_rebuild_test::tests::test_cluster_root_heal_recovers_ec84_shards_after_background_target_crash",
+      "oracle": "background-target-crash-ec8-4.json",
+      "evidence": "process-crash-restart",
+      "unclean_shutdown_marker": true,
+      "min_objects": 9,
+      "max_objects": 65,
+      "topology": {"nodes": 3, "drives_per_node": 4},
+      "erasure": {"data_blocks": 8, "parity_blocks": 4},
+      "erasure_set_drive_count": 12,
+      "scope": "Target process killed during partial background rebuild on a single 3x4 EC8+4 set; real unclean-shutdown marker, exact unversioned S3 bodies and replacement-drive shards; not power loss, multi-set, or multi-pool."
     }
   },
   "release_lanes": {

.github/workflows/audit.yml

@@ -16,7 +16,7 @@ name: Security Audit
 
 on:
   push:
-    branches: [ main ]
+    branches: [ main, release ]
     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 ]
+    branches: [ main, release ]
     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 download the same pinned previous release as e2e-upgrade.yml.
+# Upgrade cases use an independent 1.0.0-rc.2 pin defined below.
 #
 # 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: ""
-      # Pinned previous release used by distributed::upgrade_test (same pin as e2e-upgrade.yml).
+      # Independent 1.0.0-rc.2 source pin for distributed::upgrade_test.
       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,6 +44,7 @@ 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;
@@ -374,6 +375,33 @@ 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.
@@ -451,4 +479,30 @@ 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

@@ -13,8 +13,11 @@
 // 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, signed_admin_post};
-use crate::common::init_logging;
+use crate::chaos::{
+    VersionShardCensus, census_object_version_on_disk, sha256_hex, signed_admin_post, wait_for_complete_physical_shard_on_disk,
+};
+use crate::common::{init_logging, rustfs_binary_path};
+use crate::scanner_heal_evidence::{EvidenceTopology, RestartObservation, ScannerHealEvidenceCase, restart_evidence_run};
 use aws_sdk_s3::Client;
 use aws_sdk_s3::primitives::ByteStream;
 use std::collections::{BTreeMap, HashSet};
@@ -87,6 +90,19 @@ 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 mut dist = DistCluster::start_with_env(
         DistLayout::ThreeByFourEc84,
         &[
@@ -110,13 +126,24 @@ 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 = census_object_version_on_disk(&replaced_drive, &bucket, &item.key, None)?;
+        item.baseline =
+            wait_for_complete_physical_shard_on_disk(&replaced_drive, &bucket, &item.key, None, Duration::from_secs(10)).await?;
         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]
+        .process
+        .as_ref()
+        .ok_or("target process is absent")?
+        .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")?)?;
@@ -167,6 +194,7 @@ async fn three_node_four_drive_ec8_4_root_heal_rebuilds_replaced_drive_after_res
         .chain(std::iter::once((outage_key.to_string(), outage_body.clone())))
         .collect::<BTreeMap<_, _>>();
     let expected_keys = inventory.keys().cloned().collect::<HashSet<_>>();
+    let mut node_listings = Vec::new();
     for node_index in 0..dist.cluster.nodes.len() {
         let client = dist.client(node_index)?;
         assert_inventory(&client, &bucket, &inventory).await?;
@@ -177,6 +205,45 @@ 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);
+    }
+
+    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,
+            },
+        )?;
     }
 
     Ok(())

crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs

@@ -16,18 +16,20 @@
 
 #[cfg(test)]
 mod tests {
-    use crate::chaos::{VersionShardCensus, census_object_version_on_disk, sha256_hex, signed_admin_post};
+    use crate::chaos::{
+        VersionShardCensus, census_object_version_on_disk, sha256_hex, signed_admin_post,
+        wait_for_complete_physical_shard_on_disk,
+    };
     use crate::common::{
         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;
@@ -39,34 +41,14 @@ 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,
-    }
-
     const BACKGROUND_TARGET_RESTART_EVIDENCE: ScannerHealEvidenceCase = ScannerHealEvidenceCase {
         id: "background-target-restart",
         oracle: "background-target-restart.json",
         evidence: "process-restart",
         unclean_shutdown_marker: false,
+        topology: EvidenceTopology::new(4, 1),
+        storage_class_standard: None,
+        erasure_set_drive_count: None,
     };
 
     const BACKGROUND_TARGET_CRASH_EVIDENCE: ScannerHealEvidenceCase = ScannerHealEvidenceCase {
@@ -74,82 +56,30 @@ mod tests {
         oracle: "background-target-crash.json",
         evidence: "process-crash-restart",
         unclean_shutdown_marker: true,
+        topology: EvidenceTopology::new(4, 1),
+        storage_class_standard: None,
+        erasure_set_drive_count: None,
     };
 
-    struct RestartEvidenceContext {
-        directory: PathBuf,
-        run: RestartEvidenceRun,
-        case: ScannerHealEvidenceCase,
-    }
+    const BACKGROUND_TARGET_RESTART_EC84_EVIDENCE: ScannerHealEvidenceCase = ScannerHealEvidenceCase {
+        id: "background-target-restart-ec8-4",
+        oracle: "background-target-restart-ec8-4.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"),
+    };
 
-    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"),
-        })
-    }
+    const BACKGROUND_TARGET_CRASH_EC84_EVIDENCE: ScannerHealEvidenceCase = ScannerHealEvidenceCase {
+        id: "background-target-crash-ec8-4",
+        oracle: "background-target-crash-ec8-4.json",
+        evidence: "process-crash-restart",
+        unclean_shutdown_marker: true,
+        topology: EvidenceTopology::new(3, 4),
+        storage_class_standard: Some("EC:4"),
+        erasure_set_drive_count: Some("12"),
+    };
 
     struct TcpPortBlackhole {
         port: u16,
@@ -332,11 +262,10 @@ mod tests {
 
     // Healing may rewrite non-identity bookkeeping in xl.meta. The census
     // therefore compares the canonical selected metadata fields plus every
-    // physical shard, while the payload seed makes object mix-ups observable.
+    // physical shard.
     #[derive(Debug)]
     struct PhysicalObjectManifest {
         key: String,
-        payload_seed: u8,
         shard_census: VersionShardCensus,
     }
 
@@ -892,7 +821,10 @@ mod tests {
         cluster: &RustFSTestClusterEnvironment,
         previous_cycle_end: u64,
     ) -> Result<u64, Box<dyn Error + Send + Sync>> {
-        let deadline = Instant::now() + Duration::from_secs(60);
+        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;
         loop {
             let mut latest_cycle_end = 0;
             let mut versions_observed = false;
@@ -920,24 +852,61 @@ mod tests {
                 let versions_scanned = metrics["versions_scanned"]
                     .as_u64()
                     .ok_or("scanner status is missing its version-coverage counter")?;
-                latest_cycle_end = latest_cycle_end.max(cycle_end);
+                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);
+                }
                 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={}",
+                    "node{node_index}: end={cycle_end}, versions={versions_scanned}, cycle={}, active={}, leader={}, result={}, backlog={}",
                     metrics["current_cycle"],
                     metrics["current_cycle_active"],
                     metrics["leader_lock_state"],
                     metrics["last_cycle_result"],
+                    backlog,
                 ));
             }
-            // The coordinator records cycle completion, but remote workers
-            // record scanned versions. Both witnesses need not share a node.
+            // 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.
             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 an object-scanning cycle after {previous_cycle_end}: {observations:?}"
+                    "enabled scanner did not complete a successful object-scanning cycle after {previous_cycle_end}: {observations:?}"
                 )
                 .into());
             }
@@ -957,7 +926,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(420),
+            Duration::from_secs(720),
             run_cluster_root_heal_interruption(InterruptionScenario::BackgroundTargetRestart),
         )
         .await?
@@ -967,17 +936,37 @@ 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(420),
+            Duration::from_secs(720),
             run_cluster_root_heal_interruption(InterruptionScenario::BackgroundTargetCrash),
         )
         .await?
     }
 
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_cluster_root_heal_recovers_ec84_shards_after_background_target_restart()
+    -> Result<(), Box<dyn Error + Send + Sync>> {
+        timeout(
+            Duration::from_secs(720),
+            run_cluster_root_heal_interruption(InterruptionScenario::BackgroundTargetRestartEc84),
+        )
+        .await?
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_cluster_root_heal_recovers_ec84_shards_after_background_target_crash()
+    -> Result<(), Box<dyn Error + Send + Sync>> {
+        timeout(
+            Duration::from_secs(720),
+            run_cluster_root_heal_interruption(InterruptionScenario::BackgroundTargetCrashEc84),
+        )
+        .await?
+    }
+
     #[tokio::test(flavor = "multi_thread")]
     async fn test_cluster_root_heal_recovers_remote_shards_after_coordinator_restart() -> Result<(), Box<dyn Error + Send + Sync>>
     {
         timeout(
-            Duration::from_secs(420),
+            Duration::from_secs(720),
             run_cluster_root_heal_interruption(InterruptionScenario::BackgroundCoordinatorRestart),
         )
         .await?
@@ -1010,29 +999,38 @@ mod tests {
         IsolatedTargetRestart,
         BackgroundTargetRestart,
         BackgroundTargetCrash,
+        BackgroundTargetRestartEc84,
+        BackgroundTargetCrashEc84,
         BackgroundCoordinatorRestart,
         TargetEndpointBlackhole,
     }
 
     async fn run_cluster_root_heal_interruption(scenario: InterruptionScenario) -> Result<(), Box<dyn Error + Send + Sync>> {
         let server_binary = rustfs_binary_path();
-        let evidence_run = match scenario {
-            InterruptionScenario::BackgroundTargetRestart => {
-                restart_evidence_run(&server_binary, BACKGROUND_TARGET_RESTART_EVIDENCE)?
-            }
-            InterruptionScenario::BackgroundTargetCrash => {
-                restart_evidence_run(&server_binary, BACKGROUND_TARGET_CRASH_EVIDENCE)?
-            }
+        let evidence_case = match scenario {
+            InterruptionScenario::BackgroundTargetRestart => Some(BACKGROUND_TARGET_RESTART_EVIDENCE),
+            InterruptionScenario::BackgroundTargetCrash => Some(BACKGROUND_TARGET_CRASH_EVIDENCE),
+            InterruptionScenario::BackgroundTargetRestartEc84 => Some(BACKGROUND_TARGET_RESTART_EC84_EVIDENCE),
+            InterruptionScenario::BackgroundTargetCrashEc84 => Some(BACKGROUND_TARGET_CRASH_EC84_EVIDENCE),
             _ => None,
         };
+        let evidence_run = match evidence_case {
+            Some(case) => restart_evidence_run(&server_binary, case)?,
+            None => None,
+        };
         let mut evidence_objects = Vec::new();
         let (background_enabled, interruption_node, interruption_kind) = match scenario {
             InterruptionScenario::IsolatedTargetRestart => (false, 1, "target_restart"),
             InterruptionScenario::BackgroundTargetRestart => (true, 1, "background_target_restart"),
             InterruptionScenario::BackgroundTargetCrash => (true, 1, "background_target_crash"),
+            InterruptionScenario::BackgroundTargetRestartEc84 => (true, 1, "background_target_restart_ec8_4"),
+            InterruptionScenario::BackgroundTargetCrashEc84 => (true, 1, "background_target_crash_ec8_4"),
             InterruptionScenario::BackgroundCoordinatorRestart => (true, 0, "coordinator_restart"),
             InterruptionScenario::TargetEndpointBlackhole => (false, 1, "target_endpoint_blackhole"),
         };
+        let topology = evidence_case
+            .map(|case| case.topology)
+            .unwrap_or_else(|| EvidenceTopology::new(4, 1));
         init_logging();
         info!(
             event = "heal_interruption_started",
@@ -1044,9 +1042,15 @@ mod tests {
             "Starting root-heal interruption test"
         );
 
-        let mut cluster = RustFSTestClusterEnvironment::new(4).await?;
+        let mut cluster = RustFSTestClusterEnvironment::with_topology(topology.cluster_topology()).await?;
         cluster.set_env("RUSTFS_UNSAFE_BYPASS_DISK_CHECK", "true");
         cluster.set_env("RUSTFS_HEAL_ENABLED", "true");
+        if let Some(storage_class) = evidence_case.and_then(|case| case.storage_class_standard) {
+            cluster.set_env("RUSTFS_STORAGE_CLASS_STANDARD", storage_class);
+        }
+        if let Some(erasure_set_drive_count) = evidence_case.and_then(|case| case.erasure_set_drive_count) {
+            cluster.set_env("RUSTFS_ERASURE_SET_DRIVE_COUNT", erasure_set_drive_count);
+        }
         // Heal control uses the first lexicographically sorted grid host.
         // Keep that coordinator distinct from the remote target at index 1.
         cluster.nodes.sort_by(|left, right| left.url.cmp(&right.url));
@@ -1069,10 +1073,19 @@ 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 = std::env::var("RUSTFS_HEAL_CHAOS_LOG_DIR").unwrap_or_else(|_| format!("{}/logs", cluster.temp_dir));
+        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")
+        };
         std::fs::create_dir_all(&log_dir)?;
         for node_index in 0..cluster.nodes.len() {
-            cluster.set_node_capture_log_path(node_index, format!("{log_dir}/node{node_index}.log"))?;
+            cluster.set_node_capture_log_path(node_index, log_dir.join(format!("node{node_index}.log")).to_string_lossy())?;
         }
         cluster.start_with_binary(&server_binary).await?;
         let clients = cluster.create_all_clients()?;
@@ -1095,11 +1108,22 @@ mod tests {
             .and_then(|value| value.parse::<usize>().ok())
             .unwrap_or(4 * 1024 * 1024)
             .clamp(1024 * 1024, 16 * 1024 * 1024);
+        let mut created_online_objects = Vec::with_capacity(online_object_count);
         let mut expected_manifests = Vec::with_capacity(online_object_count);
         let mut unclean_shutdown_marker_observed = None;
-        for index in 0..online_object_count {
-            let key = format!("cluster/online/object-{index:04}.bin");
-            let payload_seed = u8::try_from(index + 1).expect("clamped object count must fit in u8");
+        let mut attempt_count = 0usize;
+        let max_online_attempts = online_object_count.saturating_mul(topology.total_drives().max(1));
+        while expected_manifests.len() < online_object_count {
+            if attempt_count >= max_online_attempts {
+                return Err(format!(
+                    "target replacement drive held only {}/{} baseline object shards after {attempt_count} writes",
+                    expected_manifests.len(),
+                    online_object_count
+                )
+                .into());
+            }
+            let key = format!("cluster/online/object-{attempt_count:04}.bin");
+            let payload_seed = ((attempt_count % 251) + 1) as u8;
             timeout(
                 Duration::from_secs(30),
                 clients[0]
@@ -1111,6 +1135,13 @@ mod tests {
             )
             .await??;
             let shard_census = census_object_version_on_disk(&replaced_disk, bucket, &key, None)?;
+            if !shard_census.has_xl_meta {
+                timeout(Duration::from_secs(30), clients[0].delete_object().bucket(bucket).key(&key).send()).await??;
+                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:?}"
@@ -1119,11 +1150,9 @@ mod tests {
                 !shard_census.expected_part_numbers.is_empty(),
                 "chaos objects must use physical part shards rather than inline data: {shard_census:?}"
             );
-            expected_manifests.push(PhysicalObjectManifest {
-                key,
-                payload_seed,
-                shard_census,
-            });
+            created_online_objects.push((key.clone(), payload_seed));
+            expected_manifests.push(PhysicalObjectManifest { key, shard_census });
+            attempt_count += 1;
         }
 
         let expected_pool_metadata = if background_enabled {
@@ -1169,31 +1198,38 @@ mod tests {
             if node_index == 1 {
                 continue;
             }
-            let census = census_object_version_on_disk(Path::new(&node.data_dir), bucket, outage_key, None)?;
-            assert!(
-                census.is_complete(),
-                "online node {node_index} must hold a complete outage-object shard: {census:?}"
-            );
-            let erasure_index = census
-                .erasure_index
-                .ok_or_else(|| format!("online node {node_index} outage-object shard has no erasure index: {census:?}"))?;
-            assert!(
-                (1..=cluster.nodes.len()).contains(&erasure_index),
-                "online node {node_index} outage-object erasure index is out of range: {census:?}"
-            );
-            assert!(
-                outage_peer_erasure_indices.insert(erasure_index),
-                "outage-object erasure index {erasure_index} is duplicated across online nodes"
-            );
+            for (drive_index, drive) in node.data_dirs.iter().enumerate() {
+                let census = census_object_version_on_disk(Path::new(drive), bucket, outage_key, None)?;
+                assert!(
+                    census.is_complete(),
+                    "online node {node_index} drive {drive_index} must hold a complete outage-object shard: {census:?}"
+                );
+                let erasure_index = census.erasure_index.ok_or_else(|| {
+                    format!("online node {node_index} drive {drive_index} outage-object shard has no erasure index: {census:?}")
+                })?;
+                assert!(
+                    (1..=topology.total_drives()).contains(&erasure_index),
+                    "online node {node_index} drive {drive_index} outage-object erasure index is out of range: {census:?}"
+                );
+                assert!(
+                    outage_peer_erasure_indices.insert(erasure_index),
+                    "outage-object erasure index {erasure_index} is duplicated across online drives"
+                );
+            }
         }
         assert_eq!(
             outage_peer_erasure_indices.len(),
-            cluster.nodes.len().saturating_sub(1),
-            "every online node must contribute one unique outage-object erasure index"
+            topology.total_drives().saturating_sub(cluster.nodes[1].data_dirs.len()),
+            "every online drive must contribute one unique outage-object erasure index"
+        );
+        let missing_outage_erasure_indices = (1..=topology.total_drives())
+            .filter(|index| !outage_peer_erasure_indices.contains(index))
+            .collect::<HashSet<_>>();
+        assert_eq!(
+            missing_outage_erasure_indices.len(),
+            cluster.nodes[1].data_dirs.len(),
+            "the stopped node must account for every missing outage-object erasure index"
         );
-        let expected_outage_target_erasure_index = (1..=cluster.nodes.len())
-            .find(|index| !outage_peer_erasure_indices.contains(index))
-            .ok_or("online outage-object shards leave no erasure index for the replacement target")?;
 
         let heal_body = r#"{"recursive":true,"dryRun":false,"remove":false,"recreate":true,"scanMode":2,"updateParity":false,"nolock":false}"#;
         if !background_enabled {
@@ -1326,7 +1362,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(format!("{log_dir}/node0.log"))?;
+        let coordinator_log = std::fs::read_to_string(log_dir.join("node0.log"))?;
         assert!(
             coordinator_log
                 .lines()
@@ -1463,7 +1499,11 @@ mod tests {
                 "Restored target endpoint forwarding"
             );
         } else {
-            if scenario == InterruptionScenario::BackgroundTargetRestart {
+            let graceful_restart = matches!(
+                scenario,
+                InterruptionScenario::BackgroundTargetRestart | InterruptionScenario::BackgroundTargetRestartEc84
+            );
+            if graceful_restart {
                 cluster.stop_node_gracefully(interruption_node).await?;
             } else {
                 cluster.stop_node(interruption_node)?;
@@ -1485,7 +1525,7 @@ mod tests {
             if background_enabled {
                 let marker_exists = unclean_shutdown_marker.is_file();
                 unclean_shutdown_marker_observed = Some(marker_exists);
-                let expected_marker = !matches!(scenario, InterruptionScenario::BackgroundTargetRestart);
+                let expected_marker = !graceful_restart;
                 assert!(
                     marker_exists == expected_marker,
                     "background restart/crash lane observed unexpected unclean-shutdown marker state"
@@ -1586,29 +1626,35 @@ mod tests {
             "outage object must have a complete target shard: {outage_census:?}"
         );
         assert_eq!(
+            outage_census
+                .erasure_index
+                .filter(|index| missing_outage_erasure_indices.contains(index)),
             outage_census.erasure_index,
-            Some(expected_outage_target_erasure_index),
-            "the outage object must be rebuilt into its own missing erasure slot"
+            "the outage object must be rebuilt into one of the stopped node's missing erasure slots"
         );
 
         if let Some(cycle_end) = scanner_cycle_floor {
             wait_for_scanner_cycle_after(&cluster, cycle_end).await?;
         }
 
-        let mut expected_keys = expected_manifests
+        let mut expected_keys = created_online_objects
             .iter()
-            .map(|manifest| manifest.key.clone())
+            .map(|(key, _)| key.clone())
             .collect::<HashSet<_>>();
         assert!(expected_keys.insert(outage_key.to_string()));
         let node_listings = assert_all_nodes_list_exact_keys(&clients, bucket, &expected_keys).await?;
 
         let target_client = cluster.create_s3_client(1)?;
-        for expected in &expected_manifests {
-            let response = target_client.get_object().bucket(bucket).key(&expected.key).send().await?;
+        for (key, payload_seed) in &created_online_objects {
+            let response = target_client.get_object().bucket(bucket).key(key).send().await?;
             let actual = response.body.collect().await?.into_bytes();
-            let expected_body = deterministic_object_body(object_size_bytes, expected.payload_seed);
-            assert_eq!(actual.as_ref(), expected_body.as_slice(), "object body changed for {}", expected.key);
-            if evidence_run.is_some() {
+            let expected_body = deterministic_object_body(object_size_bytes, *payload_seed);
+            assert_eq!(actual.as_ref(), expected_body.as_slice(), "object body changed for {key}");
+            if evidence_run.is_some()
+                && let Some(expected) = expected_manifests
+                    .iter()
+                    .find(|manifest| manifest.key.as_str() == key.as_str())
+            {
                 evidence_objects.push(serde_json::json!({
                     "key": expected.key, "version_id": expected.shard_census.version_id,
                     "expected_bytes": expected_body.len(), "actual_bytes": actual.len(),
@@ -1682,32 +1728,18 @@ 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");
-            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()?;
+            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,
+                },
+            )?;
         }
 
         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};
+use crate::common::{RustFSTestClusterEnvironment, RustFSTestEnvironment, init_logging, signal_process};
 use aws_sdk_s3::Client;
 use aws_sdk_s3::primitives::ByteStream;
 use aws_sdk_s3::types::{
@@ -2207,6 +2207,33 @@ 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();
@@ -2234,7 +2261,6 @@ 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
@@ -2245,6 +2271,21 @@ 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),
@@ -2304,6 +2345,31 @@ 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,6 +23,9 @@ 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,6 +79,11 @@ 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?;
@@ -101,8 +106,6 @@ 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,6 +168,10 @@ 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 ---
@@ -348,16 +352,6 @@ 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
@@ -522,6 +516,9 @@ 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

@@ -0,0 +1,181 @@
+// 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/api/mod.rs

@@ -563,7 +563,7 @@ pub mod store_list {
 }
 
 pub mod storage {
-    pub use crate::core::pools::HealLifecycleExpiryContext;
+    pub use crate::core::pools::{HealLifecycleExpiryContext, POOL_META_NAME};
     pub use crate::store::HealWalkVersion;
     pub use crate::store::{
         BootstrapLocalTarget, ECStore, SCANNER_PUBLICATION_LEASE_TTL_MS, ScannerDataMovementPauseStatus, all_local_disk,

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(std::io::Error::other(format!("Failed to write data: {summary_text}")))
+        Err(write_err.into())
     }
 
     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(std::io::Error::other(format!("Failed to shutdown writers: {summary_text}")))
+        Err(write_err.into())
     }
 }
 
@@ -1002,6 +1002,7 @@ 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;
@@ -1451,7 +1452,14 @@ mod tests {
             Ok(_) => panic!("writer quorum failure should fail the encode pipeline"),
             Err(err) => err,
         };
-        assert!(err.to_string().contains("Failed to write data"));
+        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());
         tokio::time::timeout(Duration::from_secs(1), reader_dropped)
             .await
             .expect("writer failure should abort the blocked producer")
@@ -1644,7 +1652,7 @@ mod tests {
 
     #[tokio::test]
     async fn multi_writer_short_write_fails_before_shutdown() {
-        let mut writers = vec![Some(bitrot_writer(ShortWriteWriter, 16))];
+        let mut writers = vec![Some(bitrot_writer(ShortWriteWriter, 32))];
         let err = {
             let mut writer = MultiWriter::new(&mut writers, 1);
             writer
@@ -1653,63 +1661,93 @@ mod tests {
                 .expect_err("short writes must fail the shard writer")
         };
 
-        assert!(err.to_string().contains("Failed to write data"));
+        let err = StorageError::from(err);
+        assert!(matches!(&err, StorageError::Io(source) if source.kind() == std::io::ErrorKind::WriteZero));
+        assert!(!err.is_quorum_error());
         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 = {
+        let (err, summary) = {
             let mut writer = MultiWriter::new(&mut writers, 1);
-            writer
+            let err = writer
                 .write(vec![Bytes::from_static(b"offline-a"), Bytes::from_static(b"offline-b")])
                 .await
-                .expect_err("offline writers cannot satisfy write quorum")
+                .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))
         };
 
-        let err = err.to_string();
-        assert!(err.contains("Failed to write data"));
-        assert!(err.contains("offline-disks=2/2"));
-        assert!(err.contains("required=1"));
+        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 shutdown_err = {
+        let (shutdown_err, summary) = {
             let mut writer = MultiWriter::new(&mut writers, 1);
-            writer
+            let err = writer
                 .shutdown()
                 .await
-                .expect_err("offline writers cannot satisfy shutdown quorum")
+                .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))
         };
 
-        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"));
+        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"));
     }
 
     #[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), 16))];
+        let mut writers = vec![Some(bitrot_writer(DeferredCommitWriter::new(committed), 32))];
         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");
-        let err = err.to_string();
-        assert!(err.contains("Failed to write data"));
-        assert!(err.contains("required=2"));
-        assert!(err.contains("erasure 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());
+        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 = writer
             .shutdown()
             .await
             .expect_err("shutdown quorum above writer count must fail");
-        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"));
+        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"));
     }
 
     // The production wiring (`MultiWriter::new`) must arm a real deadline by
@@ -1794,7 +1832,13 @@ mod tests {
             .write(four_shards())
             .await
             .expect_err("two stalled writers must fail the write quorum instead of hanging");
-        assert!(err.to_string().contains("Failed to write data"));
+        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());
     }
 
     // A small object whose bytes were fully buffered leaves `write` succeeding
@@ -1839,7 +1883,13 @@ mod tests {
             .shutdown()
             .await
             .expect_err("two shutdown stalls must fail the shutdown quorum instead of hanging");
-        assert!(err.to_string().contains("Failed to shutdown writers"));
+        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());
     }
 
     // A slow-but-honest writer that keeps completing shards (delay < stall
@@ -2121,7 +2171,13 @@ mod tests {
             .await
             .expect_err("streaming encode must fail when write quorum is unavailable");
 
-        assert!(err.to_string().contains("Failed to write data"));
+        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());
     }
 
     #[tokio::test]
@@ -2145,7 +2201,13 @@ mod tests {
             .await
             .expect_err("write quorum failure must fail the inline encode");
 
-        assert!(err.to_string().contains("Failed to write data"));
+        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!(
             committed.lock().expect("committed buffer should be lockable").is_empty(),
             "successful writer must not be committed when write quorum fails before shutdown"
@@ -2173,7 +2235,13 @@ mod tests {
             .await
             .expect_err("shutdown quorum failure must fail the inline encode");
 
-        assert!(err.to_string().contains("Failed to shutdown writers"));
+        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!(
             !committed.lock().expect("committed buffer should be lockable").is_empty(),
             "the successful writer should have committed before shutdown quorum failure was reported"
@@ -2395,7 +2463,13 @@ mod tests {
             .await
             .expect_err("batched encode must fail when write quorum is unavailable");
 
-        assert!(err.to_string().contains("Failed to write data"));
+        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());
     }
 
     #[tokio::test]

crates/ecstore/src/store/heal.rs

@@ -378,6 +378,26 @@ 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,
@@ -829,6 +849,73 @@ 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

@@ -34,7 +34,7 @@ use std::time::{Duration, UNIX_EPOCH};
 use tokio::sync::{RwLock, Semaphore};
 use tracing::{debug, error, warn};
 
-use super::{DiskStore, EcstoreError};
+use super::{DiskStore, EcstoreError, POOL_META_NAME, RUSTFS_META_BUCKET};
 
 /// Outcome of classifying an error returned by [`HealStorageAPI::heal_object`].
 enum HealObjectOutcome {
@@ -68,6 +68,13 @@ struct PageConcurrencyGuard {
     set_label: String,
 }
 
+struct ErasureSetPassCounters<'a> {
+    processed_objects: &'a mut u64,
+    successful_objects: &'a mut u64,
+    failed_objects: &'a mut u64,
+    skipped_objects: &'a mut u64,
+}
+
 impl PageConcurrencyGuard {
     fn new(in_flight: Arc<AtomicUsize>, set_label: String) -> Self {
         let current = in_flight.fetch_add(1, Ordering::SeqCst) + 1;
@@ -834,6 +841,21 @@ impl ErasureSetHealer {
             current_object_index = 0;
         }
 
+        if failed_objects == 0 && skipped_objects == 0 && failed_buckets == 0 {
+            self.heal_replacement_pool_metadata(
+                set_disk_id,
+                &mut ErasureSetPassCounters {
+                    processed_objects: &mut processed_objects,
+                    successful_objects: &mut successful_objects,
+                    failed_objects: &mut failed_objects,
+                    skipped_objects: &mut skipped_objects,
+                },
+                resume_manager,
+                checkpoint_manager,
+            )
+            .await?;
+        }
+
         // 5. finalize. Only declare the set healed when nothing failed AND
         // nothing was transiently skipped — otherwise the resume/checkpoint
         // state must survive so the failed/skipped versions are retried instead
@@ -919,6 +941,211 @@ impl ErasureSetHealer {
         Ok(())
     }
 
+    async fn heal_replacement_pool_metadata(
+        &self,
+        set_disk_id: &str,
+        counters: &mut ErasureSetPassCounters<'_>,
+        resume_manager: &ResumeManager,
+        checkpoint_manager: &CheckpointManager,
+    ) -> Result<()> {
+        if self.replacement_task_id.is_none() {
+            return Ok(());
+        }
+        if self.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;
+        if checkpoint.processed_objects.contains(&checkpoint_key)
+            || checkpoint.failed_objects.contains(&checkpoint_key)
+            || checkpoint.skipped_objects.contains(&checkpoint_key)
+        {
+            return Ok(());
+        }
+        drop(checkpoint);
+
+        self.verify_replacement_identity_fence("pool metadata").await?;
+        resume_manager
+            .set_current_item(Some(RUSTFS_META_BUCKET.to_string()), Some(POOL_META_NAME.to_string()))
+            .await?;
+
+        let result = match self
+            .storage
+            .heal_object(RUSTFS_META_BUCKET, POOL_META_NAME, None, &self.heal_opts)
+            .await
+        {
+            Ok((result, None)) if target_outcomes_complete(&result, &self.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,
+                    )
+                    .await
+                {
+                    Ok(true) => (object_size, Ok(())),
+                    Ok(false) => (
+                        object_size,
+                        Err(Error::transient_skip(
+                            "Skipped replacement pool metadata heal because target readback did not confirm the committed version",
+                        )),
+                    ),
+                    Err(err) => (
+                        object_size,
+                        Err(Error::transient_skip(format!(
+                            "Skipped replacement pool metadata heal because target readback failed: {err}"
+                        ))),
+                    ),
+                }
+            }
+            Ok((result, None)) => (
+                result_object_size_u64(&result),
+                Err(Error::transient_skip(
+                    "Skipped replacement pool metadata heal because a replacement target was not committed",
+                )),
+            ),
+            Ok((result, Some(err))) => {
+                let object_size = result_object_size_u64(&result);
+                match Self::classify_heal_object_error(&err) {
+                    HealObjectOutcome::Absent | HealObjectOutcome::Transient => (
+                        object_size,
+                        Err(Error::transient_skip(format!(
+                            "Skipped replacement pool metadata heal due to transient error: {err}"
+                        ))),
+                    ),
+                    HealObjectOutcome::Failed => (object_size, Err(err)),
+                }
+            }
+            Err(err @ Error::TaskCancelled) | Err(err @ Error::TaskTimeout) => return Err(err),
+            Err(err) => match Self::classify_heal_object_error(&err) {
+                HealObjectOutcome::Absent | HealObjectOutcome::Transient => (
+                    0,
+                    Err(Error::transient_skip(format!(
+                        "Skipped replacement pool metadata heal due to transient error: {err}"
+                    ))),
+                ),
+                HealObjectOutcome::Failed => (0, Err(err)),
+            },
+        };
+
+        let (object_size, result) = result;
+        let mut bytes_processed = self.progress.read().await.bytes_processed;
+        let mut telemetry_unknown = false;
+        let checkpoint_outcome = match result {
+            Ok(()) => {
+                telemetry_unknown |= !increment_counter(counters.successful_objects);
+                telemetry_unknown |= !add_bytes(&mut bytes_processed, object_size);
+                debug!(
+                    target: "rustfs::heal::erasure_healer",
+                    event = EVENT_HEAL_ERASURE_OBJECT_STATE,
+                    component = LOG_COMPONENT_HEAL,
+                    subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
+                    set_disk_id,
+                    bucket = RUSTFS_META_BUCKET,
+                    object = POOL_META_NAME,
+                    state = "healed",
+                    "Replacement pool metadata healed"
+                );
+                CheckpointObjectOutcome::Processed
+            }
+            Err(Error::TransientSkip { message }) => {
+                telemetry_unknown |= !increment_counter(counters.skipped_objects);
+                telemetry_unknown |= !add_bytes(&mut bytes_processed, object_size);
+                warn!(
+                    target: "rustfs::heal::erasure_healer",
+                    event = EVENT_HEAL_ERASURE_OBJECT_STATE,
+                    component = LOG_COMPONENT_HEAL,
+                    subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
+                    set_disk_id,
+                    bucket = RUSTFS_META_BUCKET,
+                    object = POOL_META_NAME,
+                    state = "transient_skip",
+                    error = %message,
+                    "Replacement pool metadata heal skipped due to transient error"
+                );
+                CheckpointObjectOutcome::Skipped
+            }
+            Err(err) => {
+                telemetry_unknown |= !increment_counter(counters.failed_objects);
+                telemetry_unknown |= !add_bytes(&mut bytes_processed, object_size);
+                warn!(
+                    target: "rustfs::heal::erasure_healer",
+                    event = EVENT_HEAL_ERASURE_OBJECT_STATE,
+                    component = LOG_COMPONENT_HEAL,
+                    subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
+                    set_disk_id,
+                    bucket = RUSTFS_META_BUCKET,
+                    object = POOL_META_NAME,
+                    state = "failed",
+                    error = %err,
+                    "Replacement pool metadata heal failed"
+                );
+                CheckpointObjectOutcome::Failed
+            }
+        };
+
+        telemetry_unknown |= !increment_counter(counters.processed_objects);
+        let (outcome_record, counter_unknown, skipped_new_versions, skipped_ilm_expired) = {
+            let mut progress = self.progress.write().await;
+            progress.set_current_object(Some(object_key.clone()));
+            progress.update_object_progress(
+                *counters.processed_objects,
+                *counters.successful_objects,
+                *counters.failed_objects,
+                *counters.skipped_objects,
+                bytes_processed,
+            );
+            if telemetry_unknown {
+                progress.mark_unknown();
+            }
+            (
+                CheckpointObjectOutcomeRecord {
+                    object: checkpoint_key,
+                    outcome: checkpoint_outcome,
+                    successful: progress.objects_healed,
+                    failed: progress.objects_failed,
+                    skipped: progress.skipped_objects,
+                    bytes: progress.bytes_processed,
+                    skipped_new_versions: progress.skipped_new_versions,
+                    skipped_ilm_expired: progress.skipped_ilm_expired,
+                    counter_unknown: progress.counter_unknown,
+                },
+                progress.counter_unknown,
+                progress.skipped_new_versions,
+                progress.skipped_ilm_expired,
+            )
+        };
+        checkpoint_manager.record_object_outcome(outcome_record).await?;
+        resume_manager
+            .update_progress_with_bytes(
+                *counters.processed_objects,
+                *counters.successful_objects,
+                *counters.failed_objects,
+                *counters.skipped_objects,
+                bytes_processed,
+            )
+            .await?;
+        resume_manager
+            .set_skipped_version_counts(skipped_new_versions, skipped_ilm_expired)
+            .await?;
+        if counter_unknown {
+            resume_manager.mark_counter_unknown().await?;
+        }
+        Ok(())
+    }
+
     /// heal single bucket with resume
     #[allow(clippy::too_many_arguments)]
     #[tracing::instrument(skip(self, current_object_index, processed_objects, successful_objects, failed_objects, skipped_objects, resume_manager, checkpoint_manager), fields(bucket = %bucket, bucket_index = bucket_index))]
@@ -1679,7 +1906,8 @@ mod resume_loop_tests {
     use crate::heal::storage::{HealLifecycleExpiryContext, HealListItem, HealObjectInfo, HealStorageAPI};
     use crate::heal::storage_api::status::BucketInfo;
     use crate::heal::{
-        BUCKET_META_PREFIX, DiskOption, DiskStore, EcstoreError, Endpoint, HealDiskExt as _, RUSTFS_META_BUCKET, new_disk,
+        BUCKET_META_PREFIX, DiskOption, DiskStore, EcstoreError, Endpoint, HealDiskExt as _, POOL_META_NAME, RUSTFS_META_BUCKET,
+        new_disk,
     };
     use crate::{Error, Result};
     use rustfs_heal_contracts::heal_channel::{HealOpts, HealRequestSource};
@@ -1773,6 +2001,28 @@ mod resume_loop_tests {
         assert!(!target_outcomes_complete(&duplicate, &["replacement-a".to_string()]));
     }
 
+    fn replacement_target_ok_result(endpoint: &str, object: &str) -> HealResultItem {
+        HealResultItem {
+            object: object.to_string(),
+            object_size: 1024,
+            before: Infos {
+                drives: vec![HealDriveInfo {
+                    endpoint: endpoint.to_string(),
+                    state: "missing".to_string(),
+                    ..Default::default()
+                }],
+            },
+            after: Infos {
+                drives: vec![HealDriveInfo {
+                    endpoint: endpoint.to_string(),
+                    state: "ok".to_string(),
+                    ..Default::default()
+                }],
+            },
+            ..Default::default()
+        }
+    }
+
     #[derive(Clone)]
     struct Page {
         items: Vec<HealListItem>,
@@ -1783,6 +2033,8 @@ 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):
@@ -1808,6 +2060,8 @@ 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>)>>,
@@ -1909,6 +2163,7 @@ 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))))
                 }
@@ -1922,6 +2177,17 @@ 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,
@@ -2444,6 +2710,8 @@ mod resume_loop_tests {
             HealRequestSource::AutoHeal,
         )
         .with_replacement_targets(vec!["replacement-a".to_string()], Some(replacement_task_id.clone()));
+        env.storage
+            .set_result(POOL_META_NAME, None, replacement_target_ok_result("replacement-a", POOL_META_NAME));
 
         healer
             .heal_erasure_set(&["b".to_string()], "pool_0_set_0")
@@ -2461,9 +2729,164 @@ mod resume_loop_tests {
             CheckpointManager::has_checkpoint(&env.healer.disk, &replacement_task_id).await,
             "the checkpoint must survive until the caller clears the healing marker"
         );
+        assert_eq!(env.storage.calls(), vec![(POOL_META_NAME.to_string(), None)]);
         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;
+        let replacement_task_id = ResumeUtils::generate_task_id();
+        ResumeManager::new_replacement_intent(
+            env.healer.disk.clone(),
+            replacement_task_id.clone(),
+            "pool_0_set_0".to_string(),
+            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
+            .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 healer = ErasureSetHealer::new(
+            env.storage.clone(),
+            Arc::new(RwLock::new(HealProgress::new())),
+            CancellationToken::new(),
+            env.healer.disk.clone(),
+            HealOpts::default(),
+            HealRequestSource::AutoHeal,
+        )
+        .with_replacement_targets(vec!["replacement-a".to_string()], Some(replacement_task_id.clone()));
+
+        let error = healer
+            .heal_erasure_set(&["b".to_string()], "pool_0_set_0")
+            .await
+            .expect_err("unconfirmed pool metadata readback must keep the replacement incomplete");
+
+        assert!(error.to_string().contains("Replacement erasure set heal incomplete"));
+        let state = ResumeManager::load_replacement_intent(env.healer.disk.clone(), &replacement_task_id)
+            .await
+            .expect("replacement retry state must remain")
+            .get_state()
+            .await;
+        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)]);
+    }
+
     #[tokio::test]
     async fn retry_exhaustion_keeps_resume_artifacts_for_recovery() {
         let env = make_env().await;

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 request = HealRequest::new(
+    let mut request = HealRequest::new(
         HealType::Object {
             bucket: "bucket".to_string(),
             object: "object".to_string(),
@@ -1886,6 +1886,7 @@ 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)
@@ -3725,7 +3726,7 @@ async fn test_submit_heal_request_returns_merged_before_full_for_duplicate() {
         }),
     );
 
-    let request = HealRequest::new(
+    let mut request = HealRequest::new(
         HealType::Object {
             bucket: "bucket".to_string(),
             object: "object".to_string(),
@@ -3742,6 +3743,7 @@ 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/mod.rs

@@ -27,10 +27,10 @@ pub mod task;
 pub mod utils;
 
 use storage_api::owner::{
-    ECSTORE_BUCKET_META_PREFIX, ECSTORE_DATA_USAGE_CACHE_NAME, ECSTORE_HEALING_MARKER_PATH, ECSTORE_RUSTFS_META_BUCKET,
-    EcstoreConditionalFileUpdate, EcstoreDeleteOptions, EcstoreDiskAPI, EcstoreDiskBytes, EcstoreDiskError, EcstoreDiskOption,
-    EcstoreDiskResult, EcstoreDiskStore, EcstoreEndpoint, EcstoreErrorType, EcstoreStorageError, EcstoreStore, ObjectIO,
-    ObjectOperations, ecstore_local_disk_map_read, ecstore_new_disk,
+    ECSTORE_BUCKET_META_PREFIX, ECSTORE_DATA_USAGE_CACHE_NAME, ECSTORE_HEALING_MARKER_PATH, ECSTORE_POOL_META_NAME,
+    ECSTORE_RUSTFS_META_BUCKET, EcstoreConditionalFileUpdate, EcstoreDeleteOptions, EcstoreDiskAPI, EcstoreDiskBytes,
+    EcstoreDiskError, EcstoreDiskOption, EcstoreDiskResult, EcstoreDiskStore, EcstoreEndpoint, EcstoreErrorType,
+    EcstoreStorageError, EcstoreStore, ObjectIO, ObjectOperations, ecstore_local_disk_map_read, ecstore_new_disk,
 };
 
 pub use erasure_healer::ErasureSetHealer;
@@ -41,6 +41,7 @@ pub use task::{HealOptions, HealPriority, HealRequest, HealTask, HealType};
 pub(crate) const DATA_USAGE_CACHE_NAME: &str = ECSTORE_DATA_USAGE_CACHE_NAME;
 pub(crate) const BUCKET_META_PREFIX: &str = ECSTORE_BUCKET_META_PREFIX;
 pub(crate) const RUSTFS_META_BUCKET: &str = ECSTORE_RUSTFS_META_BUCKET;
+pub(crate) const POOL_META_NAME: &str = ECSTORE_POOL_META_NAME;
 
 /// Marker written to every local disk while the process runs; removed by
 /// [`clear_unclean_shutdown_markers`] on graceful shutdown. Finding it at

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::{HashSet, VecDeque};
+use std::collections::{HashMap, HashSet, VecDeque};
 use std::sync::Arc;
 use std::time::Duration;
 use tokio::sync::mpsc;
@@ -134,7 +134,10 @@ struct MrfQueueKey {
 }
 
 fn queue_key(intent: &MrfIntent) -> MrfQueueKey {
-    let version_id = intent.version_id.filter(|bytes| *bytes != [0; 16]);
+    let version_id = (!matches!(intent.kind, rustfs_common::mrf_channel::MrfKind::MetadataCorruption))
+        .then_some(intent.version_id)
+        .flatten()
+        .filter(|bytes| *bytes != [0; 16]);
     let scope = (!matches!(intent.kind, rustfs_common::mrf_channel::MrfKind::MetadataCorruption))
         .then_some(intent.scope)
         .flatten();
@@ -533,29 +536,101 @@ 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 snapshot(&self) -> (Vec<u8>, Vec<u8>) {
+    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>)> {
         let mut authoritative = Vec::new();
         let mut legacy = Vec::new();
-        for intent in self.queue.intents() {
+        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;
+            }
             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;
             }
         }
-        (authoritative, legacy)
+        Some((authoritative, legacy))
     }
 
     async fn flush(&mut self) {
-        let (authoritative, legacy) = self.snapshot();
+        let Some((authoritative, legacy)) = self.snapshot() else {
+            self.dirty = true;
+            return;
+        };
         let authoritative_persisted = write_journal(MRF_SCOPED_JOURNAL_PATH, &authoritative).await;
         if !authoritative.is_empty() {
             counter!("rustfs_heal_mrf_journal_fsync_total").increment(1);
@@ -594,11 +669,31 @@ 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 {
-                // 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) => {}
+                // 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);
+                    }
+                }
                 Ok(HealAdmissionResult::Full) | Ok(HealAdmissionResult::Dropped(HealAdmissionDropReason::QueueFull)) => {
                     intent.attempts = intent.attempts.saturating_add(1);
                     if intent.attempts >= MRF_MAX_ATTEMPTS {
@@ -632,13 +727,14 @@ impl MrfRuntime {
     }
 
     fn retained_replay_journal(&self) -> bool {
-        self.retain_replay_journal || !self.durable_replay_anchors.is_empty()
+        self.retain_replay_journal || !self.retained_replay_intents.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()
@@ -652,6 +748,13 @@ 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;
     }
 }
 
@@ -704,6 +807,7 @@ 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(
@@ -786,6 +890,7 @@ 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) => {
@@ -799,6 +904,7 @@ async fn replay_into(
                 journal_on_disk: true,
                 retain_journal_for_replay: true,
                 durable_replay_anchors: Vec::new(),
+                retained_replay_intents: HashMap::new(),
             };
         }
     };
@@ -837,6 +943,7 @@ 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.
@@ -851,6 +958,7 @@ 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 {
@@ -872,6 +980,7 @@ async fn replay_into(
                     break;
                 }
                 Ok(HealAdmissionResult::Dropped(_)) => {
+                    retained_replay_intents.remove(&queue_key(&intent));
                     rustfs_common::mrf_channel::release_mrf_intent(&intent);
                 }
                 Err(_) => {
@@ -906,6 +1015,7 @@ async fn replay_into(
         journal_on_disk,
         retain_journal_for_replay,
         durable_replay_anchors,
+        retained_replay_intents,
     }
 }
 
@@ -921,6 +1031,7 @@ 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,
     };
 
@@ -930,6 +1041,7 @@ 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.
@@ -956,17 +1068,7 @@ async fn run_mrf_consumer(manager: Arc<HealManager>, mut receiver: mpsc::Receive
                     );
                     return;
                 }
-                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.enqueue_batch(batch.drain(..));
                 runtime.dispatch(manager.as_ref()).await;
                 if runtime.new_since_flush >= runtime.config.flush_threshold {
                     runtime.flush().await;
@@ -1118,6 +1220,7 @@ 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 {
@@ -1135,14 +1238,196 @@ 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();
@@ -1177,6 +1462,46 @@ 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());

crates/heal/src/heal/storage.rs

@@ -436,6 +436,14 @@ 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
@@ -1268,6 +1276,18 @@ 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/storage_api.rs

@@ -29,6 +29,7 @@ pub(crate) use rustfs_ecstore::api::error::{Error as EcstoreErrorType, StorageEr
 pub(crate) use rustfs_ecstore::api::runtime::local_disk_map_read as ecstore_local_disk_map_read;
 pub(crate) use rustfs_ecstore::api::storage::{
     ECStore as EcstoreStore, HealLifecycleExpiryContext as EcstoreHealLifecycleExpiryContext,
+    POOL_META_NAME as ECSTORE_POOL_META_NAME,
 };
 use rustfs_storage_api as storage_contracts;
 
@@ -36,10 +37,11 @@ pub(crate) mod owner {
     pub(crate) use super::storage_contracts::{ObjectIO, ObjectOperations};
 
     pub(crate) use super::{
-        ECSTORE_BUCKET_META_PREFIX, ECSTORE_DATA_USAGE_CACHE_NAME, ECSTORE_HEALING_MARKER_PATH, ECSTORE_RUSTFS_META_BUCKET,
-        EcstoreConditionalFileUpdate, EcstoreDeleteOptions, EcstoreDiskAPI, EcstoreDiskBytes, EcstoreDiskError,
-        EcstoreDiskResult, EcstoreDiskStore, EcstoreEndpoint, EcstoreErrorType, EcstoreHealLifecycleExpiryContext,
-        EcstoreStorageError, EcstoreStore, ecstore_load_admin_data_usage_from_backend_cached, ecstore_local_disk_map_read,
+        ECSTORE_BUCKET_META_PREFIX, ECSTORE_DATA_USAGE_CACHE_NAME, ECSTORE_HEALING_MARKER_PATH, ECSTORE_POOL_META_NAME,
+        ECSTORE_RUSTFS_META_BUCKET, EcstoreConditionalFileUpdate, EcstoreDeleteOptions, EcstoreDiskAPI, EcstoreDiskBytes,
+        EcstoreDiskError, EcstoreDiskResult, EcstoreDiskStore, EcstoreEndpoint, EcstoreErrorType,
+        EcstoreHealLifecycleExpiryContext, EcstoreStorageError, EcstoreStore, ecstore_load_admin_data_usage_from_backend_cached,
+        ecstore_local_disk_map_read,
     };
 
     pub(crate) use super::{EcstoreDiskOption, ecstore_new_disk};

crates/heal/src/heal/task.rs

@@ -574,6 +574,18 @@ impl HealTask {
         }
     }
 
+    pub(super) async fn outcome_bucket_incarnation_id(&self, bucket: &str, dry_run: bool) -> Result<Option<Uuid>> {
+        if dry_run {
+            return Ok(None);
+        }
+        match self.await_with_control(self.storage.bucket_incarnation_id(bucket)).await {
+            Ok(incarnation_id) => Ok(incarnation_id),
+            Err(Error::TaskCancelled) => Err(Error::TaskCancelled),
+            Err(Error::TaskTimeout) => Err(Error::TaskTimeout),
+            Err(_) => Ok(None),
+        }
+    }
+
     fn single_object_identity(&self) -> Option<HealObjectIdentity> {
         let (bucket, object, version) = match &self.heal_type {
             HealType::Object {
@@ -597,6 +609,9 @@ impl HealTask {
         expected: HealObjectIdentity,
         receipt: Option<HealObjectReceipt>,
     ) -> bool {
+        if self.options.dry_run || self.cancel_token.is_cancelled() {
+            return false;
+        }
         let Some(receipt) = receipt else {
             return false;
         };

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

@@ -403,6 +403,7 @@ impl HealTask {
         };
 
         for (set_disk_id, heal_opts) in listing_scopes {
+            let bucket_incarnation_id = self.outcome_bucket_incarnation_id(bucket, heal_opts.dry_run).await?;
             let mut continuation_token: Option<String> = None;
             let mut deferred = DeferredWindow::default();
             let mut inline_retry: Option<DeferredObject> = None;
@@ -501,13 +502,15 @@ impl HealTask {
                     let retry_attempt = item.attempt;
                     let mut telemetry_unknown = false;
                     let object = item.name.as_str();
-                    let identity =
+                    let mut identity =
                         self.outcome_identity(bucket, object, item.version_id.as_deref(), heal_opts.pool, heal_opts.set);
+                    identity.bucket_incarnation_id = bucket_incarnation_id;
                     let mut disposition = if heal_opts.dry_run {
                         HealObjectDisposition::DryRunObserved
                     } else {
                         HealObjectDisposition::Unknown
                     };
+                    let mut recorded_authoritative_outcome = false;
                     let mut detail = None;
                     {
                         let mut progress = self.progress.write().await;
@@ -520,23 +523,37 @@ impl HealTask {
                         Some(Error::other("heal object retry age exhausted"))
                     } else {
                         match self
-                            .await_with_control(
-                                self.storage
-                                    .heal_object(bucket, object, item.version_id.as_deref(), &heal_opts),
-                            )
+                            .await_with_control(self.storage.heal_object_with_receipt(
+                                bucket,
+                                object,
+                                item.version_id.as_deref(),
+                                &heal_opts,
+                            ))
                             .await
                         {
-                            Ok((result, None)) => match unavailable_recreate_error(&result, &heal_opts) {
-                                Some(error) => Some(error),
-                                None => {
-                                    telemetry_unknown |= !increment_counter(&mut healed);
-                                    telemetry_unknown |=
-                                        !add_bytes(&mut bytes, u64::try_from(result.object_size).unwrap_or(u64::MAX));
-                                    self.record_result_item(result).await;
-                                    None
+                            Ok(storage_result) if storage_result.error.is_none() => {
+                                match unavailable_recreate_error(&storage_result.item, &heal_opts) {
+                                    Some(error) => Some(error),
+                                    None => {
+                                        telemetry_unknown |= !increment_counter(&mut healed);
+                                        telemetry_unknown |= !add_bytes(
+                                            &mut bytes,
+                                            u64::try_from(storage_result.item.object_size).unwrap_or(u64::MAX),
+                                        );
+                                        recorded_authoritative_outcome = self
+                                            .record_verified_storage_receipt(identity.clone(), storage_result.receipt)
+                                            .await;
+                                        self.record_result_item(storage_result.item).await;
+                                        None
+                                    }
                                 }
-                            },
-                            Ok((_, Some(err))) if is_missing_object_dir_heal_result(object, &err) => {
+                            }
+                            Ok(storage_result)
+                                if storage_result
+                                    .error
+                                    .as_ref()
+                                    .is_some_and(|err| is_missing_object_dir_heal_result(object, err)) =>
+                            {
                                 telemetry_unknown |= !increment_counter(&mut healed);
                                 debug!(
                                     target: "rustfs::heal::task",
@@ -551,7 +568,8 @@ impl HealTask {
                                 );
                                 None
                             }
-                            Ok((_, Some(err))) | Err(err) => Some(err),
+                            Ok(storage_result) => storage_result.error,
+                            Err(err) => Some(err),
                         }
                     };
 
@@ -674,11 +692,13 @@ impl HealTask {
                         continue;
                     }
 
-                    self.outcome.write().await.record(HealObjectOutcome {
-                        identity,
-                        disposition,
-                        detail,
-                    });
+                    if !recorded_authoritative_outcome {
+                        self.outcome.write().await.record(HealObjectOutcome {
+                            identity,
+                            disposition,
+                            detail,
+                        });
+                    }
 
                     let mut progress = self.progress.write().await;
                     progress.update_object_progress(

crates/heal/src/heal/task/heal_object.rs

@@ -162,6 +162,9 @@ impl HealTask {
             pool: self.options.pool_index,
             set: self.options.set_index,
         };
+        let mut expected_identity =
+            self.outcome_identity(bucket, object, version_id, self.options.pool_index, self.options.set_index);
+        expected_identity.bucket_incarnation_id = self.outcome_bucket_incarnation_id(bucket, self.options.dry_run).await?;
 
         let heal_fut = self.storage.heal_object_with_receipt(bucket, object, version_id, &heal_opts);
         let heal_result = if self.source == HealRequestSource::ReadRepair {
@@ -266,10 +269,6 @@ impl HealTask {
                     let mut progress = self.progress.write().await;
                     progress.update_object_progress(1, 1, 0, 0, object_size);
                 }
-                let expected_bucket_incarnation_id = self.storage.bucket_incarnation_id(bucket).await?;
-                let mut expected_identity =
-                    self.outcome_identity(bucket, object, version_id, self.options.pool_index, self.options.set_index);
-                expected_identity.bucket_incarnation_id = expected_bucket_incarnation_id;
                 self.record_verified_storage_receipt(expected_identity, storage_result.receipt)
                     .await;
                 self.record_result_item(result).await;

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

@@ -154,6 +154,126 @@ mod canonical_outcome {
         );
     }
 
+    #[tokio::test]
+    async fn bucket_heal_records_matching_positive_storage_receipts() {
+        let incarnation = Uuid::new_v4();
+        let storage = Arc::new(MockStorage {
+            heal_object_receipts: Mutex::new(HashMap::from([
+                (
+                    "object-a".to_string(),
+                    VecDeque::from([object_receipt("object-a", None, HealObjectDisposition::Repaired, incarnation)]),
+                ),
+                (
+                    "object-b".to_string(),
+                    VecDeque::from([object_receipt("object-b", None, HealObjectDisposition::Repaired, incarnation)]),
+                ),
+            ])),
+            bucket_incarnation_id: Mutex::new(Some(incarnation)),
+            ..Default::default()
+        });
+        let task = bucket_task(storage);
+
+        task.execute()
+            .await
+            .expect("bucket heal should record verified object receipts");
+
+        let outcome = task.get_outcome().await;
+        assert_eq!(outcome.execution, HealExecutionOutcome::Completed);
+        assert_eq!(outcome.counters.healed, 2);
+        assert_eq!(outcome.counters.unknown, 0);
+        assert_eq!(outcome.objects.len(), 2);
+        assert!(outcome.objects.iter().all(|item| {
+            item.identity.bucket_incarnation_id == Some(incarnation) && item.disposition == HealObjectDisposition::Repaired
+        }));
+    }
+
+    #[tokio::test]
+    async fn bucket_heal_keeps_repairing_when_bucket_incarnation_is_unavailable() {
+        let storage = Arc::new(MockStorage {
+            heal_object_receipts: Mutex::new(HashMap::from([(
+                "object-a".to_string(),
+                VecDeque::from([object_receipt(
+                    "object-a",
+                    None,
+                    HealObjectDisposition::Repaired,
+                    Uuid::new_v4(),
+                )]),
+            )])),
+            bucket_incarnation_unavailable: Mutex::new(true),
+            ..Default::default()
+        });
+        let task = bucket_task(storage.clone());
+
+        task.execute()
+            .await
+            .expect("bucket heal should continue when only proof ownership is unavailable");
+
+        let outcome = task.get_outcome().await;
+        assert_eq!(outcome.execution, HealExecutionOutcome::Completed);
+        assert_eq!(outcome.counters.healed, 0);
+        assert_eq!(outcome.counters.unknown, 2);
+        assert!(
+            outcome
+                .objects
+                .iter()
+                .all(|item| item.disposition == HealObjectDisposition::Unknown)
+        );
+        assert_eq!(storage.healed_objects.lock().expect("healed objects").len(), 2);
+    }
+
+    #[tokio::test]
+    async fn bucket_heal_rejects_stale_receipts_without_double_recording() {
+        let expected_incarnation = Uuid::new_v4();
+        let storage = Arc::new(MockStorage {
+            heal_object_receipts: Mutex::new(HashMap::from([
+                (
+                    "object-a".to_string(),
+                    VecDeque::from([object_receipt(
+                        "object-a",
+                        None,
+                        HealObjectDisposition::Repaired,
+                        Uuid::new_v4(),
+                    )]),
+                ),
+                (
+                    "object-b".to_string(),
+                    VecDeque::from([object_receipt(
+                        "object-b",
+                        None,
+                        HealObjectDisposition::Repaired,
+                        expected_incarnation,
+                    )]),
+                ),
+            ])),
+            bucket_incarnation_id: Mutex::new(Some(expected_incarnation)),
+            ..Default::default()
+        });
+        let task = bucket_task(storage);
+
+        task.execute()
+            .await
+            .expect("stale bucket receipt should not fail the legacy heal");
+
+        let outcome = task.get_outcome().await;
+        assert_eq!(outcome.execution, HealExecutionOutcome::Completed);
+        assert_eq!(outcome.counters.healed, 1);
+        assert_eq!(outcome.counters.unknown, 1);
+        assert_eq!(outcome.objects.len(), 2);
+        let object_a = outcome
+            .objects
+            .iter()
+            .find(|item| item.identity.object == "object-a")
+            .expect("stale receipt object outcome");
+        assert_eq!(object_a.disposition, HealObjectDisposition::Unknown);
+        let object_b = outcome
+            .objects
+            .iter()
+            .find(|item| item.identity.object == "object-b")
+            .expect("matching receipt object outcome");
+        assert_eq!(object_b.disposition, HealObjectDisposition::Repaired);
+        assert_eq!(object_b.identity.bucket_incarnation_id, Some(expected_incarnation));
+    }
+
     #[tokio::test(start_paused = true)]
     async fn exhausted_object_does_not_abort_other_objects_or_erase_counts() {
         let storage = Arc::new(MockStorage::default());
@@ -313,6 +433,96 @@ mod canonical_outcome {
         );
     }
 
+    #[tokio::test]
+    async fn mixed_grace_and_repaired_receipt_transfer_only_repaired_responsibility() {
+        let incarnation = Uuid::new_v4();
+        let storage = Arc::new(MockStorage {
+            heal_object_outcomes: Mutex::new(HashMap::from([(
+                "object-a".to_string(),
+                VecDeque::from([MockHealObjectOutcome::DanglingGraceDeferred]),
+            )])),
+            heal_object_receipts: Mutex::new(HashMap::from([(
+                "object-b".to_string(),
+                VecDeque::from([object_receipt("object-b", None, HealObjectDisposition::Repaired, incarnation)]),
+            )])),
+            bucket_incarnation_id: Mutex::new(Some(incarnation)),
+            ..Default::default()
+        });
+        let task = bucket_task(storage);
+
+        task.execute()
+            .await
+            .expect("mixed grace and repaired receipt should complete");
+
+        let outcome = task.get_outcome().await;
+        assert_eq!(outcome.coverage, HealTraversalCoverage::Complete);
+        assert_eq!(outcome.counters.processed, 2);
+        assert_eq!(outcome.counters.healed, 1);
+        assert_eq!(outcome.counters.skipped, 1);
+        let deferred = outcome
+            .objects
+            .iter()
+            .find(|item| item.identity.object == "object-a")
+            .expect("grace object should remain recorded");
+        assert!(matches!(
+            deferred.disposition,
+            HealObjectDisposition::Deferred {
+                reason: HealDeferredReason::DanglingDeleteGrace,
+                ..
+            }
+        ));
+        assert_ne!(deferred.disposition, HealObjectDisposition::Repaired);
+        let repaired = outcome
+            .objects
+            .iter()
+            .find(|item| item.identity.object == "object-b")
+            .expect("receipt-backed object should be recorded");
+        assert_eq!(repaired.identity.bucket_incarnation_id, Some(incarnation));
+        assert_eq!(repaired.disposition, HealObjectDisposition::Repaired);
+    }
+
+    #[tokio::test]
+    async fn bucket_heal_records_matching_positive_storage_receipt() {
+        let incarnation = Uuid::new_v4();
+        let storage = Arc::new(MockStorage {
+            heal_object_receipts: Mutex::new(HashMap::from([(
+                "object-a".to_string(),
+                VecDeque::from([object_receipt("object-a", None, HealObjectDisposition::Repaired, incarnation)]),
+            )])),
+            bucket_incarnation_id: Mutex::new(Some(incarnation)),
+            ..Default::default()
+        });
+        let task = bucket_task(storage);
+
+        task.execute().await.expect("bucket heal should complete");
+
+        let outcome = task.get_outcome().await;
+        assert_eq!(outcome.counters.processed, 2);
+        assert_eq!(outcome.counters.healed, 1);
+        assert_eq!(outcome.counters.unknown, 1);
+        assert_eq!(
+            outcome
+                .objects
+                .iter()
+                .filter(|item| item.identity.object == "object-a")
+                .count(),
+            1
+        );
+        let repaired = outcome
+            .objects
+            .iter()
+            .find(|item| item.identity.object == "object-a")
+            .expect("receipt-backed bucket object should be recorded");
+        assert_eq!(repaired.identity.bucket_incarnation_id, Some(incarnation));
+        assert_eq!(repaired.disposition, HealObjectDisposition::Repaired);
+        let legacy = outcome
+            .objects
+            .iter()
+            .find(|item| item.identity.object == "object-b")
+            .expect("legacy bucket object should still be recorded");
+        assert_eq!(legacy.disposition, HealObjectDisposition::Unknown);
+    }
+
     #[tokio::test]
     async fn grace_single_object_is_completed_but_deferred() {
         let storage = Arc::new(MockStorage {
@@ -344,6 +554,18 @@ mod canonical_outcome {
                     .lock()
                     .expect("existence fixture")
                     .insert("object".to_string(), MockObjectExists::TransientSkip("retry later"));
+            } else {
+                let incarnation = Uuid::new_v4();
+                *storage.bucket_incarnation_id.lock().expect("bucket incarnation") = Some(incarnation);
+                storage.heal_object_receipts.lock().expect("receipt fixture").insert(
+                    "object".to_string(),
+                    VecDeque::from([object_receipt(
+                        "object",
+                        None,
+                        HealObjectDisposition::VerifiedHealthy,
+                        incarnation,
+                    )]),
+                );
             }
             let mut request = HealRequest::object("bucket-a".to_string(), "object".to_string(), None);
             request.options.dry_run = !transient;
@@ -1117,6 +1339,12 @@ 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>,
     format_error: Mutex<Option<Error>>,
     global_format_calls: Mutex<u32>,
@@ -1258,7 +1486,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,
+        storage.clone(),
     );
 
     task.execute().await.expect("mock object heal should complete");
@@ -1271,6 +1499,260 @@ 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]
+async fn cancelled_object_heal_rejects_matching_positive_storage_receipt() {
+    let incarnation = Uuid::new_v4();
+    let storage = Arc::new(MockStorage::default());
+    let task = HealTask::from_request(
+        HealRequest::object("bucket-a".to_string(), "object-a".to_string(), Some("version-a".to_string())),
+        storage,
+    );
+    task.cancel().await.expect("task cancellation should succeed");
+
+    let expected = HealObjectIdentity {
+        kind: HealObjectKind::Object,
+        bucket: "bucket-a".to_string(),
+        object: "object-a".to_string(),
+        version_id: Some("version-a".to_string()),
+        bucket_incarnation_id: Some(incarnation),
+        pool_index: None,
+        set_index: None,
+    };
+    let accepted = task
+        .record_verified_storage_receipt(
+            expected,
+            Some(object_receipt(
+                "object-a",
+                Some("version-a"),
+                HealObjectDisposition::Repaired,
+                incarnation,
+            )),
+        )
+        .await;
+
+    let outcome = task.get_outcome().await;
+    assert!(!accepted);
+    assert_eq!(outcome.counters.healed, 0);
+    assert!(outcome.objects.is_empty());
+}
+
+#[tokio::test]
+async fn failed_object_heal_rejects_matching_positive_storage_receipt() {
+    let incarnation = Uuid::new_v4();
+    let storage = Arc::new(MockStorage {
+        heal_object_outcome: Mutex::new(Some(MockHealObjectOutcome::OkWithOtherError("commit failed"))),
+        heal_object_receipts: Mutex::new(HashMap::from([(
+            "object-a".to_string(),
+            VecDeque::from([object_receipt(
+                "object-a",
+                Some("version-a"),
+                HealObjectDisposition::Repaired,
+                incarnation,
+            )]),
+        )])),
+        bucket_incarnation_id: Mutex::new(Some(incarnation)),
+        ..Default::default()
+    });
+    let task = HealTask::from_request(
+        HealRequest::object("bucket-a".to_string(), "object-a".to_string(), Some("version-a".to_string())),
+        storage,
+    );
+
+    let result = task.execute().await;
+
+    let outcome = task.get_outcome().await;
+    assert!(result.is_err());
+    assert_eq!(outcome.counters.healed, 0);
+    assert_eq!(outcome.counters.unchanged, 0);
+    assert!(outcome.objects.iter().all(|object| {
+        !matches!(
+            object.disposition,
+            HealObjectDisposition::Repaired
+                | HealObjectDisposition::VerifiedHealthy
+                | HealObjectDisposition::AuthoritativelyAbsent
+        )
+    }));
+}
+
+#[tokio::test]
+async fn transient_quorum_object_heal_rejects_matching_positive_storage_receipt() {
+    let incarnation = Uuid::new_v4();
+    let storage = Arc::new(MockStorage {
+        heal_object_outcome: Mutex::new(Some(MockHealObjectOutcome::OkWithReadQuorum)),
+        heal_object_receipts: Mutex::new(HashMap::from([(
+            "object-a".to_string(),
+            VecDeque::from([object_receipt(
+                "object-a",
+                Some("version-a"),
+                HealObjectDisposition::Repaired,
+                incarnation,
+            )]),
+        )])),
+        bucket_incarnation_id: Mutex::new(Some(incarnation)),
+        ..Default::default()
+    });
+    let task = HealTask::from_request(
+        HealRequest::object("bucket-a".to_string(), "object-a".to_string(), Some("version-a".to_string())),
+        storage,
+    );
+
+    let result = task.execute().await;
+
+    let outcome = task.get_outcome().await;
+    assert!(result.is_err());
+    assert_eq!(outcome.counters.healed, 0);
+    assert_eq!(outcome.counters.unchanged, 0);
+    assert!(outcome.objects.iter().all(|object| {
+        !matches!(
+            object.disposition,
+            HealObjectDisposition::Repaired
+                | HealObjectDisposition::VerifiedHealthy
+                | HealObjectDisposition::AuthoritativelyAbsent
+        )
+    }));
+}
+
+#[tokio::test]
+async fn object_heal_latches_expected_incarnation_before_repair() {
+    let original_incarnation = Uuid::new_v4();
+    let successor_incarnation = Uuid::new_v4();
+    let storage = Arc::new(MockStorage {
+        heal_object_receipts: Mutex::new(HashMap::from([(
+            "object-a".to_string(),
+            VecDeque::from([object_receipt(
+                "object-a",
+                Some("version-a"),
+                HealObjectDisposition::VerifiedHealthy,
+                original_incarnation,
+            )]),
+        )])),
+        bucket_incarnation_id: Mutex::new(Some(original_incarnation)),
+        bucket_incarnation_after_object_heal: Mutex::new(Some(successor_incarnation)),
+        ..Default::default()
+    });
+    let task = HealTask::from_request(
+        HealRequest::object("bucket-a".to_string(), "object-a".to_string(), Some("version-a".to_string())),
+        storage,
+    );
+
+    task.execute().await.expect("mock object heal should complete");
+
+    let outcome = task.get_outcome().await;
+    assert_eq!(outcome.counters.unchanged, 1);
+    assert_eq!(outcome.counters.unknown, 0);
+    let object = outcome.objects.front().expect("latched receipt should be recorded");
+    assert_eq!(object.identity.bucket_incarnation_id, Some(original_incarnation));
+    assert_eq!(object.disposition, HealObjectDisposition::VerifiedHealthy);
 }
 
 #[tokio::test]
@@ -1377,6 +1859,7 @@ enum MockHealObjectOutcome {
     RetryableLock,
     RetryableLockTimeout,
     OkWithOtherError(&'static str),
+    OkWithReadQuorum,
     ErrOther(&'static str),
     DanglingGraceDeferred,
     UnavailableDrive(DriveState),
@@ -1473,6 +1956,17 @@ 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()));
+        }
         Ok(*self.bucket_incarnation_id.lock().unwrap())
     }
 
@@ -1533,6 +2027,13 @@ impl HealStorageAPI for MockStorage {
                 MockHealObjectOutcome::RetryableSlowDown => {
                     Ok((HealResultItem::default(), Some(Error::Storage(EcstoreError::SlowDown))))
                 }
+                MockHealObjectOutcome::OkWithReadQuorum => Ok((
+                    HealResultItem::default(),
+                    Some(Error::Storage(EcstoreError::InsufficientReadQuorum(
+                        bucket.to_string(),
+                        object.to_string(),
+                    ))),
+                )),
                 MockHealObjectOutcome::PermanentOther(message) => Err(Error::other(message)),
                 MockHealObjectOutcome::OkWithOtherError(message) => Ok((HealResultItem::default(), Some(Error::other(message)))),
                 MockHealObjectOutcome::ErrOther(message) => Err(Error::other(message)),
@@ -1572,6 +2073,13 @@ impl HealStorageAPI for MockStorage {
                 MockHealObjectOutcome::RetryableSlowDown => {
                     Ok((HealResultItem::default(), Some(Error::Storage(EcstoreError::SlowDown))))
                 }
+                MockHealObjectOutcome::OkWithReadQuorum => Ok((
+                    HealResultItem::default(),
+                    Some(Error::Storage(EcstoreError::InsufficientReadQuorum(
+                        bucket.to_string(),
+                        object.to_string(),
+                    ))),
+                )),
             };
         }
         if bucket == RUSTFS_META_BUCKET && object == format!("{BUCKET_META_PREFIX}/{DATA_USAGE_CACHE_NAME}") {
@@ -1586,6 +2094,9 @@ impl HealStorageAPI for MockStorage {
             return Ok((HealResultItem::default(), Some(Error::Disk(DiskError::FileNotFound))));
         }
         self.healed_objects.lock().unwrap().push(object.to_string());
+        if let Some(bucket_incarnation_id) = self.bucket_incarnation_after_object_heal.lock().unwrap().take() {
+            *self.bucket_incarnation_id.lock().unwrap() = Some(bucket_incarnation_id);
+        }
         Ok((
             HealResultItem {
                 object_size: 1,

crates/heal/tests/mrf_pipeline_test.rs

@@ -364,6 +364,41 @@ async fn committed_snapshot_replay_takes_precedence_over_stale_legacy_mirror() {
     );
 }
 
+/// A damaged committed checkpoint is ambiguous: replay must not fall back to
+/// older legacy bytes or delete any recovery anchor until another process can
+/// publish a valid successor.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+#[serial]
+async fn damaged_committed_snapshot_blocks_legacy_fallback_and_retains_anchors() {
+    let (disk_paths, storage) = heal_env().await;
+    register_local_disks(&disk_paths, "mrf-damaged-committed-replay-test").await;
+
+    let committed = scoped_journal_record(3, "damaged-committed-bucket", "committed-object", Some([8u8; 16]), 0, 0, 0);
+    let stale_legacy = journal_record(1, "damaged-legacy-bucket", "legacy-object", None, 0);
+    write_journal_path_to_disks(&disk_paths, COMMITTED_PAYLOAD_REL, &committed);
+    let mut manifest = committed_manifest(uuid::Uuid::new_v4(), 9, &committed);
+    manifest[25] ^= 1;
+    write_journal_path_to_disks(&disk_paths, COMMITTED_MANIFEST_REL, &manifest);
+    write_journal_path_to_disks(&disk_paths, SCOPED_JOURNAL_REL, &stale_legacy);
+    write_journal_path_to_disks(&disk_paths, JOURNAL_REL, &stale_legacy);
+
+    let manager = make_manager(storage);
+    let replayed = mrf_queue::replay_journal_once(&manager).await;
+    assert_eq!(replayed, 0, "damaged committed state must fail closed");
+    assert_eq!(
+        manager.operations_snapshot().await.queued_by_source.mrf,
+        0,
+        "stale legacy bytes must not be replayed when committed state is ambiguous"
+    );
+    assert!(
+        journal_exists_on_all_disks(&disk_paths, COMMITTED_MANIFEST_REL)
+            && journal_exists_on_all_disks(&disk_paths, COMMITTED_PAYLOAD_REL)
+            && journal_matches_on_all_disks(&disk_paths, SCOPED_JOURNAL_REL, &stale_legacy)
+            && journal_matches_on_all_disks(&disk_paths, JOURNAL_REL, &stale_legacy),
+        "all recovery anchors must remain after a fail-closed committed read"
+    );
+}
+
 /// A canonical snapshot and its compatibility mirror may differ after a
 /// partial flush. Replay must choose the complete canonical epoch instead of
 /// combining records that never coexisted in memory.
@@ -591,7 +626,8 @@ fn mrf_successor_flush_child_process_fixture() {
             }),
         ));
         mrf_queue::spawn_mrf_consumer(manager.clone());
-        let expected_successor = journal_record(1, "successor-bucket", "second-object", None, 2);
+        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 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)
@@ -638,7 +674,8 @@ fn mrf_successor_flush_waiting_child_process_fixture() {
             }),
         ));
         mrf_queue::spawn_mrf_consumer(manager.clone());
-        let expected_successor = journal_record(1, "service-kill-bucket", "second-object", None, 2);
+        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 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)
@@ -678,7 +715,8 @@ 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 successor = journal_record(1, "fsync-kill-bucket", "second-object", None, 2);
+        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));
         write_journal_path_to_disks_synced(&disk_paths, SCOPED_JOURNAL_REL, &successor);
         assert!(
             journal_matches_on_all_disks(&disk_paths, SCOPED_JOURNAL_REL, &successor)
@@ -735,9 +773,8 @@ async fn journal_replay_retains_child_process_anchor_when_manager_is_full() {
     );
 }
 
-/// 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.
+/// A successor flush must preserve both pending work and accepted work whose
+/// repair has not been proven when the process restarts.
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
 #[serial]
 async fn journal_replay_survives_successor_flush_before_delete() {
@@ -752,7 +789,8 @@ 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 expected_successor = journal_record(1, "successor-bucket", "second-object", None, 2);
+    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));
     assert!(
         journal_matches_on_all_disks(&disk_paths, SCOPED_JOURNAL_REL, &expected_successor),
         "restarted process must see the pending successor snapshot"
@@ -760,11 +798,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, 1, "restart after successor flush must replay only the still-pending tail");
+    assert_eq!(replayed, 2, "restart must replay both the admitted and pending responsibilities");
     assert_eq!(
         restarted.operations_snapshot().await.queued_by_source.mrf,
-        1,
-        "the successor tail must be accepted after restart"
+        2,
+        "both unproven successor responsibilities must be accepted after restart"
     );
     assert!(
         disk_paths.iter().all(|path| {
@@ -776,8 +814,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 may replay the smaller
-/// successor snapshot, but must not lose or merge stale startup records.
+/// crash at the flush-before-delete boundary: restart must recover every
+/// unproven responsibility from the successor snapshot.
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
 #[serial]
 #[cfg(unix)]
@@ -805,7 +843,8 @@ 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 expected_successor = journal_record(1, "service-kill-bucket", "second-object", None, 2);
+    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));
     assert!(
         journal_matches_on_all_disks(&disk_paths, SCOPED_JOURNAL_REL, &expected_successor),
         "restarted process must see the successor snapshot produced before the kill"
@@ -813,11 +852,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, 1, "restart after service kill must replay only the still-pending tail");
+    assert_eq!(replayed, 2, "service-kill restart must preserve every unproven responsibility");
     assert_eq!(
         restarted.operations_snapshot().await.queued_by_source.mrf,
-        1,
-        "the successor tail must be accepted after service kill restart"
+        2,
+        "both unproven responsibilities must be accepted after service kill restart"
     );
     assert!(
         disk_paths.iter().all(|path| {
@@ -829,9 +868,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 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.
+/// 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.
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
 #[serial]
 #[cfg(unix)]
@@ -859,7 +898,8 @@ 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 expected_successor = journal_record(1, "fsync-kill-bucket", "second-object", None, 2);
+    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 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));
@@ -876,11 +916,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, 1, "new reader must replay only the authoritative successor tail");
+    assert_eq!(replayed, 2, "new reader must recover every responsibility in the authoritative successor");
     assert_eq!(
         restarted.operations_snapshot().await.queued_by_source.mrf,
-        1,
-        "the successor tail must be accepted after the fsync-boundary restart"
+        2,
+        "both responsibilities must be accepted after the fsync-boundary restart"
     );
     assert!(
         disk_paths.iter().all(|path| {

crates/kms/src/backends/local.rs

@@ -1233,6 +1233,8 @@ 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));
         }
 
@@ -2095,11 +2097,7 @@ 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
-            .map_err(|_| KmsError::key_not_found(format!("Key {key_id} not found")))?;
+        let mut master_key = self.client.load_master_key(key_id).await?;
 
         let (deletion_date_str, deletion_date_dt) = if request.force_immediate.unwrap_or(false) {
             // Tombstone first: mark the record Deleted before removing the
@@ -2202,11 +2200,7 @@ 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
-            .map_err(|_| KmsError::key_not_found(format!("Key {key_id} not found")))?;
+        let mut master_key = self.client.load_master_key(key_id).await?;
 
         if master_key.status != KeyStatus::PendingDeletion {
             return Err(KmsError::invalid_key_state(format!("Key {key_id} is not pending deletion")));
@@ -2965,6 +2959,146 @@ 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,17 +43,18 @@ 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 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 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 `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` | `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`, `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 `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` | `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` or `release` | `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.
 
@@ -62,7 +63,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` push lane; it also remains in `e2e-nightly` with
+than the merge queue or `main`/`release` push lane; it also remains in `e2e-nightly` with
 the serialized cluster fault-domain suites for scheduled soak signal.
 
 ## Scheduled validation
@@ -85,7 +86,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` (4-case matrix) | upgrade and rollback gate; server logs | no | see the PR row |
+| `e2e-upgrade.yml` (weekly) | `upgrade` (7-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 |
@@ -274,14 +275,23 @@ The bundle checker is intentionally stricter than the case checker. It requires
 schema 2 registry metadata, `evidence: measured`, the current checkout revision,
 all G01-G14/P1-P4/R-E/R-D/R-L gates, per-gate `status: pass`, lane identity,
 relative artifact paths, matching SHA256 hashes, and non-empty summaries. It
-also binds the hard evidence shape for the release claims: mixed-version gates
-must name at least two participating versions, crash/durable replay gates must
-include crash-boundary evidence, G14 must record EC8+4 with at least three nodes
-and four drives per node plus multi-set and multi-pool evidence, performance
-gates need measured durations, P3's pressure run needs at least two hours, and
-P1 needs a symbolized profile summary with resolved samples. Missing, synthetic,
-stale, tampered, undersized, or topology-mismatched evidence returns a compact
-blocked or invalid JSON result and a nonzero exit.
+also binds each evidence field to its own run provenance: `source_revision`,
+`run_id`, `measurement_window_id`, timezone-qualified `started_at` and
+`finished_at`, command arguments, and artifact format. The field
+`source_revision` must match the bundle revision, and measured performance
+duration cannot exceed the recorded run window.
+
+The hard evidence shape remains claim-specific: mixed-version gates must name at
+least two participating versions, crash/durable replay gates must include
+crash-boundary evidence, G14 must record EC8+4 with at least three nodes and four
+drives per node plus multi-set and multi-pool evidence, performance gates need
+measured durations, P3's pressure run needs at least two hours, and P1 needs a
+symbolized profile summary with resolved samples. Every G14 field and every
+performance gate's fields must also share one `measurement_window_id`, so EC8+4,
+multi-set/multi-pool, ABBA, throughput, and profiling artifacts cannot be
+stitched together from unrelated runs. Missing, synthetic, stale, tampered,
+undersized, or topology-mismatched evidence returns a compact blocked or invalid
+JSON result and a nonzero exit.
 
 This command validates the evidence package; it does not create evidence. A
 handwritten JSON file, a synthetic harness pass, a single focused case, or a

scripts/check_test_wiring.py

@@ -13,7 +13,7 @@ import tempfile
 import unittest
 import uuid
 import xml.etree.ElementTree as ET
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 from unittest import mock
 from pathlib import Path
 from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
@@ -927,6 +927,23 @@ def evidence_integer(value: object, name: str, minimum: int, maximum: int) -> in
     return value
 
 
+def evidence_string(value: object, name: str, pattern: str | None = None) -> str:
+    require(isinstance(value, str) and value.strip(), f"invalid string {name}")
+    if pattern is not None:
+        require(re.fullmatch(pattern, value) is not None, f"invalid string {name}")
+    return value
+
+
+def evidence_timestamp(value: object, name: str) -> datetime:
+    text = evidence_string(value, name)
+    try:
+        parsed = datetime.fromisoformat(text.replace("Z", "+00:00"))
+    except ValueError as error:
+        raise ValueError(f"invalid timestamp {name}") from error
+    require(parsed.tzinfo is not None, f"{name} must include timezone")
+    return parsed
+
+
 def scanner_heal_registry_schema(registry: dict[str, object]) -> int:
     return evidence_integer(registry.get("schema"), "registry schema", 1, SCANNER_HEAL_REGISTRY_SCHEMA_MAX)
 
@@ -947,6 +964,19 @@ def scanner_heal_oracle_names(root: Path) -> tuple[str, ...]:
                 f"invalid evidence for {case_id}")
         require(type(requirement.get("unclean_shutdown_marker")) is bool,
                 f"invalid unclean-shutdown marker expectation for {case_id}")
+        if "erasure" in requirement:
+            erasure = requirement["erasure"]
+            require(isinstance(erasure, dict), f"invalid erasure expectation for {case_id}")
+            data_blocks = evidence_integer(erasure.get("data_blocks"), f"{case_id} data_blocks", 1, 16)
+            parity_blocks = evidence_integer(erasure.get("parity_blocks"), f"{case_id} parity_blocks", 1, 16)
+            require(data_blocks >= parity_blocks, f"invalid erasure geometry for {case_id}")
+            require(data_blocks + parity_blocks == requirement["topology"]["nodes"] * requirement["topology"]["drives_per_node"],
+                    f"erasure geometry differs from topology for {case_id}")
+        if "erasure_set_drive_count" in requirement:
+            erasure_set_drive_count = evidence_integer(requirement.get("erasure_set_drive_count"),
+                                                       f"{case_id} erasure_set_drive_count", 1, 64)
+            require(erasure_set_drive_count == requirement["topology"]["nodes"] * requirement["topology"]["drives_per_node"],
+                    f"erasure set drive count differs from topology for {case_id}")
         names.add(oracle)
     return tuple(sorted(names))
 
@@ -1175,6 +1205,19 @@ def check_scanner_heal_evidence(root: Path, directory: Path, case_id: str) -> li
             require(oracle.get("topology") == requirement["topology"], "oracle topology mismatch")
             for key in ("nodes", "drives_per_node"):
                 evidence_integer(oracle["topology"][key], f"observed {key}", 1, 16)
+            expected_erasure = requirement.get("erasure")
+            if expected_erasure is not None:
+                require(isinstance(expected_erasure, dict), "invalid erasure expectation")
+                expected_data_blocks = evidence_integer(expected_erasure.get("data_blocks"), "expected EC data blocks", 1, 16)
+                expected_parity_blocks = evidence_integer(expected_erasure.get("parity_blocks"), "expected EC parity blocks", 1, 16)
+                require(
+                    expected_data_blocks + expected_parity_blocks
+                    == oracle["topology"]["nodes"] * oracle["topology"]["drives_per_node"],
+                    "expected EC geometry differs from topology",
+                )
+            else:
+                expected_data_blocks = None
+                expected_parity_blocks = None
             evidence_integer(oracle.get("pid_before"), "pid_before", 1, 2**32 - 1)
             evidence_integer(oracle.get("pid_after"), "pid_after", 1, 2**32 - 1)
             require(oracle["pid_before"] != oracle["pid_after"], "no process restart witnessed")
@@ -1198,6 +1241,9 @@ def check_scanner_heal_evidence(root: Path, directory: Path, case_id: str) -> li
                     parity = evidence_integer(geometry["parity_blocks"], "EC parity blocks", 1, 16)
                     require(data + parity == oracle["topology"]["nodes"] * oracle["topology"]["drives_per_node"],
                             "EC geometry differs from this case's single set")
+                    if expected_data_blocks is not None:
+                        require(data == expected_data_blocks and parity == expected_parity_blocks,
+                                "EC data/parity geometry differs from the required case")
                     evidence_integer(geometry["erasure_index"], "target erasure index", 1, data + parity)
                 require(physical["has_xl_meta"] is True and physical["version_id"] is None, "missing target metadata")
                 parts = physical["expected_part_numbers"]
@@ -1260,14 +1306,30 @@ def release_bundle_artifact_path(bundle_path: Path, raw_path: object, gate: str,
     return resolved
 
 
-def validate_release_bundle_artifact(bundle_path: Path, gate: str, field: str, evidence: dict[str, object]) -> None:
+def validate_release_bundle_artifact(bundle_path: Path, source_revision: str, gate: str, field: str,
+                                     evidence: dict[str, object]) -> str:
     require(evidence.get("evidence_type") == "measured", f"{gate}.{field} must be measured evidence")
+    require(evidence.get("source_revision") == source_revision, f"{gate}.{field} source revision mismatch")
+    run_id = evidence_string(evidence.get("run_id"), f"{gate}.{field}.run_id", r"[A-Za-z0-9][A-Za-z0-9._:-]{7,127}")
+    window_id = evidence_string(evidence.get("measurement_window_id"), f"{gate}.{field}.measurement_window_id",
+                                r"[A-Za-z0-9][A-Za-z0-9._:-]{7,127}")
+    require(window_id != run_id, f"{gate}.{field} must separate run and measurement-window identities")
+    started = evidence_timestamp(evidence.get("started_at"), f"{gate}.{field}.started_at")
+    finished = evidence_timestamp(evidence.get("finished_at"), f"{gate}.{field}.finished_at")
+    require(started <= finished, f"{gate}.{field} evidence timestamps are inverted")
+    command = evidence.get("command")
+    require(isinstance(command, list) and command and
+            all(isinstance(part, str) and part.strip() for part in command),
+            f"{gate}.{field} missing command provenance")
+    evidence_string(evidence.get("artifact_format"), f"{gate}.{field}.artifact_format",
+                    r"[A-Za-z0-9][A-Za-z0-9._+:-]{1,63}")
     artifact = release_bundle_artifact_path(bundle_path, evidence.get("artifact"), gate, field)
     require(sha(evidence.get("sha256")) and digest(artifact) == evidence["sha256"], f"{gate}.{field} artifact hash mismatch")
     summary = evidence.get("summary")
     require(isinstance(summary, str) and summary.strip(), f"{gate}.{field} missing human summary")
     if gate.startswith("P"):
         duration = evidence_integer(evidence.get("duration_seconds"), f"{gate}.{field}.duration_seconds", 1, 86400)
+        require((finished - started).total_seconds() + 1 >= duration, f"{gate}.{field} duration exceeds run window")
         require(duration >= 900, f"{gate}.{field} requires at least 900 seconds")
         if gate == "P3" and field == "two_hour_pressure_measurement":
             require(duration >= 7200, f"{gate}.{field} requires at least two hours")
@@ -1297,6 +1359,13 @@ def validate_release_bundle_artifact(bundle_path: Path, gate: str, field: str, e
             evidence_integer(evidence.get("pools"), "G14 multi_pool_evidence.pools", 2, 1024)
     if field == "profile_evidence":
         evidence_integer(evidence.get("resolved_samples"), f"{gate}.{field}.resolved_samples", 1, 2**63 - 1)
+    return window_id
+
+
+def validate_release_bundle_gate_windows(gate: str, field_windows: dict[str, str]) -> None:
+    if gate == "G14" or gate.startswith("P"):
+        windows = sorted(set(field_windows.values()))
+        require(len(windows) == 1, f"{gate} evidence fields must share one measurement window")
 
 
 def scanner_heal_release_bundle_status(root: Path, bundle_path: Path) -> dict[str, object]:
@@ -1309,7 +1378,10 @@ def scanner_heal_release_bundle_status(root: Path, bundle_path: Path) -> dict[st
     require(bundle.get("schema") == 1, "unsupported scanner/heal release evidence bundle schema")
     require(bundle.get("evidence") == "measured", "scanner/heal release evidence bundle must be measured")
     revision = subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=root, text=True).strip()
-    require(bundle.get("source_revision") == revision, "scanner/heal release evidence source revision mismatch")
+    source_revision = bundle.get("source_revision")
+    require(isinstance(source_revision, str) and re.fullmatch(r"[0-9a-f]{40}", source_revision) is not None and
+            source_revision == revision,
+            "scanner/heal release evidence source revision mismatch")
     raw_gates = bundle.get("gates")
     require(isinstance(raw_gates, dict), "scanner/heal release evidence bundle missing gates")
 
@@ -1335,6 +1407,7 @@ def scanner_heal_release_bundle_status(root: Path, bundle_path: Path) -> dict[st
         missing_fields = [field for field in required_fields if field not in fields]
         if missing_fields:
             gate_errors.append(f"missing required fields: {', '.join(missing_fields)}")
+        field_windows: dict[str, str] = {}
         for field in required_fields:
             if field not in fields:
                 continue
@@ -1343,9 +1416,14 @@ def scanner_heal_release_bundle_status(root: Path, bundle_path: Path) -> dict[st
                 gate_errors.append(f"{field} must be an object")
                 continue
             try:
-                validate_release_bundle_artifact(bundle_path, gate, field, evidence)
+                field_windows[field] = validate_release_bundle_artifact(bundle_path, source_revision, gate, field, evidence)
             except (OSError, KeyError, TypeError, ValueError) as error:
                 gate_errors.append(str(error))
+        if not gate_errors:
+            try:
+                validate_release_bundle_gate_windows(gate, field_windows)
+            except ValueError as error:
+                gate_errors.append(str(error))
         if gate_errors:
             rejected[gate] = gate_errors
         else:
@@ -1554,8 +1632,13 @@ class SelfTests(unittest.TestCase):
         def oracle_objects(requirement: dict[str, object]) -> list[dict[str, object]]:
             topology = requirement["topology"]
             total_blocks = topology["nodes"] * topology["drives_per_node"]
-            parity_blocks = 4 if total_blocks == 12 else total_blocks // 2
-            data_blocks = total_blocks - parity_blocks
+            erasure = requirement.get("erasure")
+            if erasure is None:
+                parity_blocks = 4 if total_blocks == 12 else total_blocks // 2
+                data_blocks = total_blocks - parity_blocks
+            else:
+                data_blocks = erasure["data_blocks"]
+                parity_blocks = erasure["parity_blocks"]
             physical = {"has_xl_meta": True, "version_id": None, "data_dir": "data-generation",
                         "erasure_index": 1, "data_blocks": data_blocks, "parity_blocks": parity_blocks,
                         "expected_part_numbers": [1],
@@ -1593,21 +1676,33 @@ class SelfTests(unittest.TestCase):
         registry = read_json(root / ".config/scanner-heal-required-tests.json")
         requirements, _, _ = scanner_heal_release_requirements(registry)
         gates = {}
+        source_revision = "b" * 40
+        started = datetime(2026, 9, 8, 0, 0, tzinfo=timezone.utc)
         for gate, requirement in requirements.items():
             fields = {}
             for field in SCANNER_HEAL_RELEASE_BUNDLE_REQUIRED_EVIDENCE_FIELDS[gate]:
                 artifact = artifact_dir / f"{gate}-{field}.json"
                 write_json(artifact, {"gate": gate, "field": field, "fixture": True})
+                duration = 60
                 evidence = {
                     "artifact": artifact.relative_to(bundle_dir).as_posix(),
                     "sha256": digest(artifact),
                     "evidence_type": "measured",
+                    "source_revision": source_revision,
+                    "run_id": f"{gate.lower()}-{field.replace('_', '-')}-run",
+                    "measurement_window_id": f"{gate.lower()}-window",
+                    "started_at": started.isoformat().replace("+00:00", "Z"),
+                    "command": ["cargo", "nextest", "run", requirement["description"]],
+                    "artifact_format": "json",
                     "summary": f"parser fixture for {gate}.{field}",
                 }
                 if gate.startswith("P"):
-                    evidence["duration_seconds"] = 900
+                    duration = 900
+                    evidence["duration_seconds"] = duration
                 if gate == "P3" and field == "two_hour_pressure_measurement":
-                    evidence["duration_seconds"] = 7200
+                    duration = 7200
+                    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"]
                 if gate in ("G04", "G07", "R-E", "R-L"):
@@ -1628,7 +1723,7 @@ class SelfTests(unittest.TestCase):
                 "evidence_fields": fields,
             }
         bundle = bundle_dir / "release-evidence.json"
-        write_json(bundle, {"schema": 1, "evidence": "measured", "source_revision": "b" * 40, "gates": gates})
+        write_json(bundle, {"schema": 1, "evidence": "measured", "source_revision": source_revision, "gates": gates})
         return root, bundle
 
     def test_scanner_heal_release_bundle_accepts_complete_measured_evidence(self) -> None:
@@ -1687,6 +1782,43 @@ class SelfTests(unittest.TestCase):
                 self.assertFalse(status["release_approved"])
                 self.assertTrue(any(expected in error for error in status["rejected_gates"][gate]))
 
+    def test_scanner_heal_release_bundle_requires_field_provenance(self) -> None:
+        for fault, mutation, expected in (
+            ("source", lambda item: item.update({"source_revision": "c" * 40}), "source revision mismatch"),
+            ("run-id", lambda item: item.pop("run_id"), "run_id"),
+            ("window-id", lambda item: item.update({"measurement_window_id": item["run_id"]}), "separate run"),
+            ("started-at", lambda item: item.update({"started_at": "not-a-time"}), "timestamp"),
+            ("finished-at", lambda item: item.update({"finished_at": "2026-09-07T00:00:00Z"}), "timestamps are inverted"),
+            ("command", lambda item: item.update({"command": []}), "command provenance"),
+            ("artifact-format", lambda item: item.pop("artifact_format"), "artifact_format"),
+        ):
+            with self.subTest(fault=fault), tempfile.TemporaryDirectory() as tmp:
+                root, bundle = self.scanner_heal_release_bundle_fixture(Path(tmp))
+                data = read_json(bundle)
+                mutation(data["gates"]["G01"]["evidence_fields"]["root_authority_evidence"])
+                write_json(bundle, data)
+
+                with mock.patch("subprocess.check_output", return_value="b" * 40):
+                    status = scanner_heal_release_bundle_status(root, bundle)
+                self.assertEqual(status["decision"], "blocked")
+                self.assertFalse(status["release_approved"])
+                self.assertTrue(any(expected in error for error in status["rejected_gates"]["G01"]), fault)
+
+    def test_scanner_heal_release_bundle_requires_same_gate_measurement_window(self) -> None:
+        for gate, field in (("G14", "multi_pool_evidence"), ("P1", "profile_evidence"), ("P3", "heal_capacity_measurement")):
+            with self.subTest(gate=gate), tempfile.TemporaryDirectory() as tmp:
+                root, bundle = self.scanner_heal_release_bundle_fixture(Path(tmp))
+                data = read_json(bundle)
+                data["gates"][gate]["evidence_fields"][field]["measurement_window_id"] = f"{gate.lower()}-different-window"
+                write_json(bundle, data)
+
+                with mock.patch("subprocess.check_output", return_value="b" * 40):
+                    status = scanner_heal_release_bundle_status(root, bundle)
+                self.assertEqual(status["decision"], "blocked")
+                self.assertFalse(status["release_approved"])
+                self.assertTrue(any("must share one measurement window" in error
+                                    for error in status["rejected_gates"][gate]))
+
     def test_scanner_heal_case_does_not_approve_pending_release(self) -> None:
         with tempfile.TemporaryDirectory() as tmp:
             root, run_dir = self.scanner_heal_fixture(Path(tmp))
@@ -1778,6 +1910,27 @@ class SelfTests(unittest.TestCase):
                         requirement["evidence_fields"] = list(SCANNER_HEAL_RELEASE_REQUIRED_EVIDENCE_FIELDS[gate])
                         break
 
+    def test_scanner_heal_ec84_case_rejects_wrong_erasure_geometry(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root, run_dir = self.scanner_heal_fixture(Path(tmp))
+            self.assertEqual(check_scanner_heal_evidence(root, run_dir, "background-target-crash-ec8-4"), [])
+
+            path = run_dir / "background-target-crash-ec8-4.json"
+            oracle = read_json(path)
+            oracle["objects"][0]["physical"]["data_blocks"] = 10
+            oracle["objects"][0]["physical"]["parity_blocks"] = 2
+            oracle["objects"][0]["expected_physical"]["data_blocks"] = 10
+            oracle["objects"][0]["expected_physical"]["parity_blocks"] = 2
+            write_json(path, oracle)
+            (run_dir / "execution.json").unlink()
+            finish_scanner_heal_receipt(run_dir, 0, root)
+
+            errors = check_scanner_heal_evidence(root, run_dir, "background-target-crash-ec8-4")
+            self.assertTrue(
+                any("EC data/parity geometry differs from the required case" in error for error in errors),
+                errors,
+            )
+
     def test_scanner_heal_pending_gate_cannot_map_to_implemented_lane(self) -> None:
         with tempfile.TemporaryDirectory() as tmp:
             root, run_dir = self.scanner_heal_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
-10|crates/ecstore/src/erasure/coding/encode.rs
+8|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