fix: address confirmed release validation regressions

.config/e2e-full-selection.txt

@@ -1,2 +1,2 @@
-sha256-darwin=f0c78fdb93471575d9a64c5c46eae6c806bdd0bc10a6e33d7fb574aabd8db5a3
-sha256-linux=03ed7016cab672de9320e31375a0358eceacb4408b0e79cf063614fa7c878b87
+sha256-darwin=71d04825c143b85334802dbd2d67df6f08006f18dc2835df85bc21ea85eca02a
+sha256-linux=baee6b0c8b38a5be139d5583bcd4fd645a27659a70b2ca8f81646d760c299395

.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'

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,7 +13,9 @@
 // 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::chaos::{
+    VersionShardCensus, census_object_version_on_disk, signed_admin_post, wait_for_complete_physical_shard_on_disk,
+};
 use crate::common::init_logging;
 use aws_sdk_s3::Client;
 use aws_sdk_s3::primitives::ByteStream;
@@ -110,7 +112,8 @@ 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)?;
     }
 

crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs

@@ -16,7 +16,10 @@
 
 #[cfg(test)]
 mod tests {
-    use crate::chaos::{VersionShardCensus, census_object_version_on_disk, sha256_hex, signed_admin_post};
+    use crate::chaos::{
+        VersionShardCensus, census_object_version_on_disk, sha256_hex, signed_admin_post,
+        wait_for_complete_physical_shard_on_disk,
+    };
     use crate::common::{
         ClusterTopology, FAST_DATA_USAGE_SCANNER_ENV, RustFSTestClusterEnvironment, RustFSTestEnvironment, admin_request,
         init_logging, rustfs_binary_path,
@@ -1210,6 +1213,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:?}"

crates/e2e_test/src/inline_fast_path_cluster_test.rs

@@ -2234,7 +2234,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 +2244,7 @@ async fn four_node_manual_transition_distributed_admission_conflict_reports_stat
             .send()
             .await?;
     }
+    put_lifecycle_with_transition_retry(&hot_client, &bucket, &tier_name).await?;
 
     let (node0, node1) = tokio::join!(
         start_manual_transition_job_on_node(&hot, 0, &bucket, prefix, &tier_name, false, 64),

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/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/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,7 +43,8 @@ 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` |
@@ -53,7 +54,7 @@ Promotion rule: never promote a report-only lane to required from one green run.
 | 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

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