fix: close replacement and protocol validation gaps

.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

.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::common::{
-        ClusterTopology, FAST_DATA_USAGE_SCANNER_ENV, RustFSTestClusterEnvironment, RustFSTestEnvironment, admin_request,
-        init_logging, rustfs_binary_path,
+    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,52 +41,6 @@ mod tests {
 
     const POOL_METADATA_OBJECT: &str = "pool.bin";
 
-    #[derive(serde::Deserialize)]
-    struct EvidenceBuild {
-        sha256: String,
-    }
-
-    #[derive(serde::Deserialize)]
-    struct RestartEvidenceRun {
-        schema: u32,
-        run_id: String,
-        source_revision: String,
-        test_build: serde_json::Value,
-        binary: EvidenceBuild,
-        test_binary: EvidenceBuild,
-    }
-
-    #[derive(Clone, Copy)]
-    struct ScannerHealEvidenceCase {
-        id: &'static str,
-        oracle: &'static str,
-        evidence: &'static str,
-        unclean_shutdown_marker: bool,
-        topology: EvidenceTopology,
-        storage_class_standard: Option<&'static str>,
-        erasure_set_drive_count: Option<&'static str>,
-    }
-
-    #[derive(Clone, Copy)]
-    struct EvidenceTopology {
-        nodes: usize,
-        drives_per_node: usize,
-    }
-
-    impl EvidenceTopology {
-        const fn new(nodes: usize, drives_per_node: usize) -> Self {
-            Self { nodes, drives_per_node }
-        }
-
-        fn total_drives(self) -> usize {
-            self.nodes * self.drives_per_node
-        }
-
-        fn cluster_topology(self) -> ClusterTopology {
-            ClusterTopology::single_pool_multidrive(self.nodes, self.drives_per_node)
-        }
-    }
-
     const BACKGROUND_TARGET_RESTART_EVIDENCE: ScannerHealEvidenceCase = ScannerHealEvidenceCase {
         id: "background-target-restart",
         oracle: "background-target-restart.json",
@@ -125,81 +81,6 @@ mod tests {
         erasure_set_drive_count: Some("12"),
     };
 
-    struct RestartEvidenceContext {
-        directory: PathBuf,
-        run: RestartEvidenceRun,
-        case: ScannerHealEvidenceCase,
-    }
-
-    fn file_sha256(path: &Path) -> Result<String, Box<dyn Error + Send + Sync>> {
-        let mut file = std::fs::File::open(path)?;
-        let mut digest = Sha256::new();
-        let mut buffer = [0_u8; 64 * 1024];
-        loop {
-            let read = file.read(&mut buffer)?;
-            if read == 0 {
-                break;
-            }
-            digest.update(&buffer[..read]);
-        }
-        Ok(digest.finalize().iter().map(|byte| format!("{byte:02x}")).collect())
-    }
-
-    fn restart_evidence_run(
-        binary: &Path,
-        case: ScannerHealEvidenceCase,
-    ) -> Result<Option<RestartEvidenceContext>, Box<dyn Error + Send + Sync>> {
-        let Some(directory) = std::env::var_os("RUSTFS_SCANNER_HEAL_RUN_DIR") else {
-            return Ok(None);
-        };
-        if case.id.is_empty()
-            || case.oracle.is_empty()
-            || !case.oracle.ends_with(".json")
-            || case.oracle.contains('/')
-            || case.oracle.contains('\\')
-            || case.oracle.contains("..")
-            || !matches!(case.evidence, "process-restart" | "process-crash-restart")
-            || (case.evidence == "process-crash-restart") != case.unclean_shutdown_marker
-        {
-            return Err("invalid scanner/heal evidence case".into());
-        }
-        let directory = PathBuf::from(directory);
-        let receipt = directory.join("run.json");
-        if receipt.metadata()?.len() > 1024 * 1024 {
-            return Err("oversized scanner/heal execution receipt".into());
-        }
-        let run: RestartEvidenceRun = serde_json::from_slice(&std::fs::read(receipt)?)?;
-        if run.schema != 1 || run.run_id.len() != 32 || run.source_revision.len() != 40 {
-            return Err("invalid scanner/heal execution identity".into());
-        }
-        let built = compiled_test_identity();
-        for key in ["source_revision", "dirty", "lock_blob", "features"] {
-            assert_eq!(built[key], run.test_build[key], "compiled test identity differs for {key}");
-        }
-        assert_eq!(file_sha256(binary)?, run.binary.sha256, "server binary must match the run receipt");
-        assert_eq!(
-            file_sha256(&std::env::current_exe()?)?,
-            run.test_binary.sha256,
-            "test executable must match the run receipt"
-        );
-        if directory.join(case.oracle).exists() {
-            return Err("scanner/heal oracle already exists; create a new execution receipt".into());
-        }
-        Ok(Some(RestartEvidenceContext { directory, run, case }))
-    }
-
-    fn compiled_test_identity() -> serde_json::Value {
-        serde_json::json!({
-            "source_revision": env!("RUSTFS_E2E_BUILD_COMMIT"),
-            "dirty": env!("RUSTFS_E2E_BUILD_DIRTY") != "false",
-            "lock_blob": env!("RUSTFS_E2E_BUILD_LOCK"),
-            "features": env!("RUSTFS_E2E_BUILD_FEATURES"),
-            "target": env!("RUSTFS_E2E_BUILD_TARGET"),
-            "profile": env!("RUSTFS_E2E_BUILD_PROFILE"),
-            "rustflags_hex": env!("RUSTFS_E2E_BUILD_RUSTFLAGS_HEX"),
-        })
-    }
-
     struct TcpPortBlackhole {
         port: u16,
         comment: String,
@@ -940,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;
@@ -968,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());
             }
@@ -1005,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?
@@ -1015,7 +936,7 @@ mod tests {
     async fn test_cluster_root_heal_recovers_remote_shards_after_background_target_crash()
     -> Result<(), Box<dyn Error + Send + Sync>> {
         timeout(
-            Duration::from_secs(420),
+            Duration::from_secs(720),
             run_cluster_root_heal_interruption(InterruptionScenario::BackgroundTargetCrash),
         )
         .await?
@@ -1025,7 +946,7 @@ mod tests {
     async fn test_cluster_root_heal_recovers_ec84_shards_after_background_target_restart()
     -> Result<(), Box<dyn Error + Send + Sync>> {
         timeout(
-            Duration::from_secs(420),
+            Duration::from_secs(720),
             run_cluster_root_heal_interruption(InterruptionScenario::BackgroundTargetRestartEc84),
         )
         .await?
@@ -1035,7 +956,7 @@ mod tests {
     async fn test_cluster_root_heal_recovers_ec84_shards_after_background_target_crash()
     -> Result<(), Box<dyn Error + Send + Sync>> {
         timeout(
-            Duration::from_secs(420),
+            Duration::from_secs(720),
             run_cluster_root_heal_interruption(InterruptionScenario::BackgroundTargetCrashEc84),
         )
         .await?
@@ -1045,7 +966,7 @@ mod tests {
     async fn test_cluster_root_heal_recovers_remote_shards_after_coordinator_restart() -> Result<(), Box<dyn Error + Send + Sync>>
     {
         timeout(
-            Duration::from_secs(420),
+            Duration::from_secs(720),
             run_cluster_root_heal_interruption(InterruptionScenario::BackgroundCoordinatorRestart),
         )
         .await?
@@ -1152,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()?;
@@ -1210,6 +1140,8 @@ mod tests {
                 attempt_count += 1;
                 continue;
             }
+            let shard_census =
+                wait_for_complete_physical_shard_on_disk(&replaced_disk, bucket, &key, None, Duration::from_secs(10)).await?;
             assert!(
                 shard_census.is_complete(),
                 "node 1 should hold a complete baseline shard for {key}: {shard_census:?}"
@@ -1430,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()
@@ -1567,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)?;
@@ -1589,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"
@@ -1792,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/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

@@ -957,6 +957,10 @@ impl ErasureSetHealer {
             });
         }
 
+        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;
@@ -2029,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):
@@ -2054,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>)>>,
@@ -2155,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))))
                 }
@@ -2168,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,
@@ -2713,6 +2733,111 @@ mod resume_loop_tests {
         drop(checkpoint);
     }
 
+    #[tokio::test]
+    async fn replacement_pool_metadata_non_owner_completes_but_missing_owner_retries() {
+        for owns_pool_metadata in [false, true] {
+            let env = make_env_with_targets(vec!["replacement-a".to_string()]).await;
+            let replacement_task_id = ResumeUtils::generate_task_id();
+            let set_index = usize::from(!owns_pool_metadata);
+            let set_disk_id = format!("pool_0_set_{set_index}");
+            ResumeManager::new_replacement_intent(
+                env.healer.disk.clone(),
+                replacement_task_id.clone(),
+                set_disk_id.clone(),
+                vec!["b".to_string()],
+                vec!["replacement-a".to_string()],
+                vec![crate::heal::resume::ReplacementTargetIdentity {
+                    endpoint: "replacement-a".to_string(),
+                    canonical_path: "/mnt/replacement-a".to_string(),
+                    physical_device_ids: vec!["device-a".to_string()],
+                    filesystem_identity: "1:2:3".to_string(),
+                }],
+            )
+            .await
+            .expect("replacement intent should persist");
+            env.storage
+                .pool_metadata_not_applicable
+                .store(!owns_pool_metadata, Ordering::SeqCst);
+            env.storage.set_outcome(POOL_META_NAME, None, HealOutcome::FileNotFound);
+            let healer = ErasureSetHealer::new(
+                env.storage.clone(),
+                Arc::new(RwLock::new(HealProgress::new())),
+                CancellationToken::new(),
+                env.healer.disk.clone(),
+                HealOpts {
+                    pool: Some(0),
+                    set: Some(set_index),
+                    ..Default::default()
+                },
+                HealRequestSource::AutoHeal,
+            )
+            .with_replacement_targets(vec!["replacement-a".to_string()], Some(replacement_task_id.clone()));
+
+            let result = healer.heal_erasure_set(&["b".to_string()], &set_disk_id).await;
+            let state = ResumeManager::load_replacement_intent(env.healer.disk.clone(), &replacement_task_id)
+                .await
+                .expect("replacement state must remain until marker cleanup")
+                .get_state()
+                .await;
+            if owns_pool_metadata {
+                let error = result.expect_err("missing metadata in the owner set must keep replacement incomplete");
+                assert!(error.to_string().contains("Replacement erasure set heal incomplete"));
+                assert!(!state.completed);
+                assert_eq!(state.replacement_phase, crate::heal::resume::ReplacementPhase::Intent);
+                assert_eq!(state.retry_count, 1);
+                assert_eq!(env.storage.calls(), vec![(POOL_META_NAME.to_string(), None)]);
+            } else {
+                result.expect("a non-owner set must complete without a pool metadata replica");
+                assert!(state.completed);
+                assert_eq!(state.replacement_phase, crate::heal::resume::ReplacementPhase::Verified);
+                assert_eq!(state.retry_count, 0);
+                assert!(env.storage.calls().is_empty(), "non-owner sets must not attempt pool metadata repair");
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn replacement_pool_metadata_unknown_scope_cannot_complete() {
+        let env = make_env_with_targets(vec!["replacement-a".to_string()]).await;
+        let healer = ErasureSetHealer::new(
+            env.storage.clone(),
+            Arc::new(RwLock::new(HealProgress::new())),
+            CancellationToken::new(),
+            env.healer.disk.clone(),
+            HealOpts {
+                pool: Some(0),
+                set: Some(0),
+                ..Default::default()
+            },
+            HealRequestSource::AutoHeal,
+        )
+        .with_replacement_targets(vec!["replacement-a".to_string()], Some("generation-a".to_string()));
+        env.storage.fail_pool_metadata_scope.store(true, Ordering::SeqCst);
+        env.storage
+            .set_result(POOL_META_NAME, None, replacement_target_ok_result("replacement-a", POOL_META_NAME));
+        let mut processed_objects = 0;
+        let mut successful_objects = 0;
+        let mut failed_objects = 0;
+        let mut skipped_objects = 0;
+        let error = healer
+            .heal_replacement_pool_metadata(
+                "pool_0_set_0",
+                &mut super::ErasureSetPassCounters {
+                    processed_objects: &mut processed_objects,
+                    successful_objects: &mut successful_objects,
+                    failed_objects: &mut failed_objects,
+                    skipped_objects: &mut skipped_objects,
+                },
+                &env.resume,
+                &env.checkpoint,
+            )
+            .await
+            .expect_err("unknown metadata placement must keep replacement incomplete");
+        assert!(error.to_string().contains("injected pool metadata scope failure"));
+        assert!(env.storage.calls().is_empty());
+        assert_eq!((processed_objects, successful_objects, failed_objects, skipped_objects), (0, 0, 0, 0));
+    }
+
     #[tokio::test]
     async fn replacement_pool_metadata_readback_failure_schedules_retry() {
         let env = make_env_with_targets(vec!["replacement-a".to_string()]).await;

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/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/task/heal_object.rs

@@ -162,11 +162,6 @@ impl HealTask {
             pool: self.options.pool_index,
             set: self.options.set_index,
         };
-        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;
-
         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?;

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

@@ -1339,6 +1339,10 @@ 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>,
@@ -1482,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");
@@ -1495,6 +1499,114 @@ 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]
@@ -1844,6 +1956,14 @@ 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()));
         }

crates/heal/tests/mrf_pipeline_test.rs

@@ -626,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)
@@ -673,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)
@@ -713,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)
@@ -770,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() {
@@ -787,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"
@@ -795,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| {
@@ -811,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)]
@@ -840,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"
@@ -848,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| {
@@ -864,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)]
@@ -894,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));
@@ -911,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 |

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