test(heal): cover MRF idle checkpoint cleanup (#7497)

Add a runtime cleanup regression test that publishes both retained replay and runtime committed checkpoints, writes scoped and legacy journals, and verifies idle cleanup removes every recovery anchor from the registered local disks.

Co-authored-by: zhi22915 <qiuzgang@gmail.com>

crates/e2e_test/src/distributed/heal_test.rs

@@ -46,6 +46,18 @@ struct ScannerHealEvidenceContext {
     run: Value,
 }
 
+struct ScannerHealEvidencePayload<'a> {
+    dist: &'a DistCluster,
+    bucket: &'a str,
+    expected: &'a [ExpectedShard],
+    outage_key: &'a str,
+    outage_body: &'a [u8],
+    replaced_drive: &'a Path,
+    pid_before: u32,
+    pid_after: u32,
+    node_listings: Vec<Vec<String>>,
+}
+
 fn file_sha256(path: &Path) -> TestResult<String> {
     let mut file = std::fs::File::open(path)?;
     let mut digest = Sha256::new();
@@ -130,23 +142,12 @@ fn assert_ec84_geometry(census: &VersionShardCensus, key: &str) -> TestResult {
     Ok(())
 }
 
-async fn write_scanner_heal_evidence(
-    context: ScannerHealEvidenceContext,
-    dist: &DistCluster,
-    bucket: &str,
-    expected: &[ExpectedShard],
-    outage_key: &str,
-    outage_body: &[u8],
-    replaced_drive: &Path,
-    pid_before: u32,
-    pid_after: u32,
-    node_listings: Vec<Vec<String>>,
-) -> TestResult {
-    let verifier = dist.client(0)?;
+async fn write_scanner_heal_evidence(context: ScannerHealEvidenceContext, payload: ScannerHealEvidencePayload<'_>) -> TestResult {
+    let verifier = payload.dist.client(0)?;
     let mut objects = Vec::new();
-    for item in expected {
-        let actual = get_object_bytes(&verifier, bucket, &item.key).await?;
-        let physical = census_object_version_on_disk(replaced_drive, bucket, &item.key, None)?;
+    for item in payload.expected {
+        let actual = get_object_bytes(&verifier, payload.bucket, &item.key).await?;
+        let physical = census_object_version_on_disk(payload.replaced_drive, payload.bucket, &item.key, None)?;
         objects.push(serde_json::json!({
             "key": item.key,
             "version_id": null,
@@ -158,14 +159,14 @@ async fn write_scanner_heal_evidence(
             "physical": physical,
         }));
     }
-    let actual = get_object_bytes(&verifier, bucket, outage_key).await?;
-    let physical = census_object_version_on_disk(replaced_drive, bucket, outage_key, None)?;
+    let actual = get_object_bytes(&verifier, payload.bucket, payload.outage_key).await?;
+    let physical = census_object_version_on_disk(payload.replaced_drive, payload.bucket, payload.outage_key, None)?;
     objects.push(serde_json::json!({
-        "key": outage_key,
+        "key": payload.outage_key,
         "version_id": null,
-        "expected_bytes": outage_body.len(),
+        "expected_bytes": payload.outage_body.len(),
         "actual_bytes": actual.len(),
-        "expected_sha256": sha256_hex(outage_body),
+        "expected_sha256": sha256_hex(payload.outage_body),
         "actual_sha256": sha256_hex(&actual),
         "expected_physical": null,
         "physical": physical,
@@ -181,11 +182,11 @@ async fn write_scanner_heal_evidence(
         "binary_sha256": string_field(&context.run, "binary.sha256")?,
         "test_binary_sha256": string_field(&context.run, "test_binary.sha256")?,
         "topology": {"nodes": EC84_NODE_COUNT, "drives_per_node": EC84_DRIVES_PER_NODE},
-        "pid_before": pid_before,
-        "pid_after": pid_after,
+        "pid_before": payload.pid_before,
+        "pid_after": payload.pid_after,
         "unclean_shutdown_marker": false,
         "objects": objects,
-        "node_listings": node_listings,
+        "node_listings": payload.node_listings,
     });
     let data = serde_json::to_vec(&evidence)?;
     if data.len() > 1024 * 1024 {
@@ -347,15 +348,17 @@ async fn three_node_four_drive_ec8_4_root_heal_rebuilds_replaced_drive_after_res
     if let Some(context) = evidence_context {
         write_scanner_heal_evidence(
             context,
-            &dist,
-            &bucket,
-            &expected,
-            outage_key,
-            &outage_body,
-            &replaced_drive,
-            target_pid_before,
-            target_pid_after,
-            node_listings,
+            ScannerHealEvidencePayload {
+                dist: &dist,
+                bucket: &bucket,
+                expected: &expected,
+                outage_key,
+                outage_body: &outage_body,
+                replaced_drive: &replaced_drive,
+                pid_before: target_pid_before,
+                pid_after: target_pid_after,
+                node_listings,
+            },
         )
         .await?;
     }

crates/heal/src/heal/mrf_queue.rs

@@ -521,6 +521,8 @@ async fn submit_mrf_heal_request(manager: &HealManager, intent: &MrfIntent) -> c
 struct MrfRuntime {
     queue: MrfQueue,
     config: MrfConsumerConfig,
+    checkpoint_owner: Uuid,
+    next_checkpoint_sequence: u64,
     new_since_flush: usize,
     /// True while the in-memory pending set has changed since the last
     /// journal flush (push, pop, or an attempts bump that alters the encoded
@@ -537,6 +539,12 @@ 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 replay source to remove after the retained replay
+    /// responsibilities are discharged. `None` means the runtime only needs
+    /// the legacy journal cleanup path for snapshots it wrote itself.
+    replay_cleanup: Option<ReplayCleanup>,
+    /// Last committed checkpoint published by this runtime flush path.
+    runtime_checkpoint: Option<(Uuid, u64)>,
     /// Earliest instant a full-admission retry may proceed.
     backoff_until: Option<tokio::time::Instant>,
 }
@@ -560,6 +568,34 @@ impl MrfRuntime {
 
     async fn flush(&mut self) {
         let (authoritative, legacy) = self.snapshot();
+        let (committed_persisted, committed_on_disk) = if authoritative.is_empty() {
+            (true, false)
+        } else {
+            match snapshot::publish_committed_snapshot(
+                &journal_disks().await,
+                self.checkpoint_owner,
+                self.next_checkpoint_sequence,
+                &authoritative,
+                self.config.journal_max_bytes,
+            )
+            .await
+            {
+                Ok(publication) => {
+                    self.runtime_checkpoint = Some((publication.owner, publication.sequence));
+                    self.next_checkpoint_sequence = publication.sequence.saturating_add(1);
+                    (true, true)
+                }
+                Err(err) => {
+                    tracing::warn!(
+                        target: "rustfs::heal::mrf",
+                        error = %err,
+                        sequence = self.next_checkpoint_sequence,
+                        "MRF committed checkpoint publish failed; retaining previous replay anchor"
+                    );
+                    (false, false)
+                }
+            }
+        };
         let authoritative_persisted = write_journal(MRF_SCOPED_JOURNAL_PATH, &authoritative).await;
         if !authoritative.is_empty() {
             counter!("rustfs_heal_mrf_journal_fsync_total").increment(1);
@@ -570,10 +606,10 @@ impl MrfRuntime {
         // old reader from observing a newer epoch that a new reader cannot
         // see when the canonical write is unavailable.
         let legacy_persisted = authoritative_persisted && write_journal(MRF_JOURNAL_PATH, &legacy).await;
-        // Keep dirty until both the authoritative snapshot and its
-        // compatibility mirror have been accepted; otherwise a one-sided
-        // failure would never retry the missing file.
-        let persisted = authoritative_persisted && legacy_persisted;
+        // Keep dirty until the committed checkpoint, authoritative snapshot,
+        // and compatibility mirror have all been accepted; otherwise a
+        // one-sided failure would never retry the missing recovery anchor.
+        let persisted = committed_persisted && authoritative_persisted && legacy_persisted;
         self.new_since_flush = 0;
         // Keep the dirty flag when every disk write failed: a clean backlog
         // would otherwise never rewrite, losing the periodic persist retry a
@@ -581,7 +617,7 @@ impl MrfRuntime {
         if persisted {
             self.dirty = false;
         }
-        self.journal_on_disk |= authoritative_persisted || legacy_persisted;
+        self.journal_on_disk |= committed_on_disk || authoritative_persisted || legacy_persisted;
     }
 
     /// Drain pending intents into the heal manager until it is full, the
@@ -639,6 +675,45 @@ impl MrfRuntime {
         self.retain_replay_journal || !self.durable_replay_anchors.is_empty()
     }
 
+    fn replay_cleanup_to_delete(&self) -> Option<ReplayCleanup> {
+        if self.journal_on_disk && !self.retained_replay_journal() {
+            Some(self.replay_cleanup.unwrap_or(ReplayCleanup::Legacy))
+        } else {
+            None
+        }
+    }
+
+    async fn delete_idle_recovery_anchors(&mut self) -> bool {
+        let runtime_deleted = match self.runtime_checkpoint {
+            Some((owner, sequence)) => {
+                match snapshot::delete_committed_snapshots_through(owner, sequence, self.config.journal_max_bytes).await {
+                    Ok(deleted) => deleted,
+                    Err(err) => {
+                        tracing::warn!(
+                            target: "rustfs::heal::mrf",
+                            error = %err,
+                            sequence,
+                            "MRF runtime checkpoint cleanup failed"
+                        );
+                        false
+                    }
+                }
+            }
+            None => true,
+        };
+        let replay_deleted = match self.replay_cleanup_to_delete() {
+            Some(cleanup) => delete_replay_source(cleanup, self.config.journal_max_bytes).await,
+            None => true,
+        };
+        if runtime_deleted && replay_deleted {
+            self.runtime_checkpoint = None;
+            self.replay_cleanup = None;
+            true
+        } else {
+            false
+        }
+    }
+
     fn discharge_durable_replay_anchors(&mut self) {
         if self.durable_replay_anchors.is_empty() {
             return;
@@ -708,6 +783,8 @@ struct ReplayOutcome {
     journal_on_disk: bool,
     retain_journal_for_replay: bool,
     durable_replay_anchors: Vec<MrfDurableRepairAnchor>,
+    cleanup: Option<ReplayCleanup>,
+    next_checkpoint_sequence: u64,
 }
 
 fn replay_must_retain_journal(
@@ -719,7 +796,7 @@ fn replay_must_retain_journal(
     rearm_incomplete || pending_depth > 0 || accepted_without_durable_anchor || durable_replay_anchors > 0
 }
 
-#[derive(Clone, Copy)]
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
 enum ReplayCleanup {
     Legacy,
     Committed { owner: Uuid, sequence: u64 },
@@ -793,6 +870,8 @@ async fn replay_into(
                 journal_on_disk: false,
                 retain_journal_for_replay: false,
                 durable_replay_anchors: Vec::new(),
+                cleanup: None,
+                next_checkpoint_sequence: 1,
             };
         }
         Err(err) => {
@@ -806,10 +885,16 @@ async fn replay_into(
                 journal_on_disk: true,
                 retain_journal_for_replay: true,
                 durable_replay_anchors: Vec::new(),
+                cleanup: None,
+                next_checkpoint_sequence: 1,
             };
         }
     };
     let cleanup = source.cleanup;
+    let next_checkpoint_sequence = match cleanup {
+        ReplayCleanup::Legacy => 1,
+        ReplayCleanup::Committed { sequence, .. } => sequence.saturating_add(1),
+    };
     let data = source.data;
     let (decoded, truncated) = decode_journal(&data);
     let replayed = decoded.len();
@@ -913,6 +998,8 @@ async fn replay_into(
         journal_on_disk,
         retain_journal_for_replay,
         durable_replay_anchors,
+        cleanup: journal_on_disk.then_some(cleanup),
+        next_checkpoint_sequence,
     }
 }
 
@@ -923,11 +1010,15 @@ async fn run_mrf_consumer(manager: Arc<HealManager>, mut receiver: mpsc::Receive
     let mut runtime = MrfRuntime {
         queue: MrfQueue::new(config.queue_capacity, config.journal_max_bytes),
         config: config.clone(),
+        checkpoint_owner: Uuid::new_v4(),
+        next_checkpoint_sequence: 1,
         new_since_flush: 0,
         dirty: false,
         journal_on_disk: false,
         retain_replay_journal: false,
         durable_replay_anchors: Vec::new(),
+        replay_cleanup: None,
+        runtime_checkpoint: None,
         backoff_until: None,
     };
 
@@ -937,6 +1028,8 @@ 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.replay_cleanup = replay.cleanup;
+    runtime.next_checkpoint_sequence = replay.next_checkpoint_sequence;
     // 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.
@@ -1001,7 +1094,7 @@ async fn run_mrf_consumer(manager: Arc<HealManager>, mut receiver: mpsc::Receive
                     TickAction::DeleteJournal => {
                         // All replayed intents have either been accepted,
                         // merged, or replaced by a pending successor snapshot.
-                        if delete_journals().await {
+                        if runtime.delete_idle_recovery_anchors().await {
                             runtime.journal_on_disk = false;
                             gauge!("rustfs_heal_mrf_journal_bytes").set(0.0);
                         }
@@ -1045,6 +1138,7 @@ fn tick_action(dirty: bool, depth: usize, journal_on_disk: bool, retain_replay_j
 mod tests {
     use super::*;
     use rustfs_common::mrf_channel::{MrfIntent, MrfKind, MrfVerifiedRepairDisposition, MrfVerifiedRepairEvent};
+    use serial_test::serial;
     use std::sync::Arc as StdArc;
 
     fn intent(bucket: &str, object: &str, attempts: u8) -> MrfIntent {
@@ -1060,6 +1154,12 @@ mod tests {
         }
     }
 
+    fn encoded_payload(intent: &MrfIntent) -> Vec<u8> {
+        let mut payload = Vec::new();
+        assert!(encode_intent(intent, &mut payload), "fixture intent must encode");
+        payload
+    }
+
     #[test]
     fn tick_action_table() {
         use TickAction::*;
@@ -1117,14 +1217,23 @@ mod tests {
         let bucket_incarnation_id = uuid::Uuid::new_v4();
         let anchor = rustfs_common::mrf_channel::MrfDurableRepairAnchor::from_intent(&intent, bucket_incarnation_id)
             .expect("fresh replay lease and bucket incarnation build a durable anchor");
+        let cleanup_owner = uuid::Uuid::new_v4();
+        let cleanup = ReplayCleanup::Committed {
+            owner: cleanup_owner,
+            sequence: 17,
+        };
         let mut runtime = MrfRuntime {
             queue: MrfQueue::new(2, usize::MAX),
             config: MrfConsumerConfig::default(),
+            checkpoint_owner: Uuid::new_v4(),
+            next_checkpoint_sequence: 1,
             new_since_flush: 0,
             dirty: false,
             journal_on_disk: true,
             retain_replay_journal: false,
             durable_replay_anchors: vec![anchor],
+            replay_cleanup: Some(cleanup),
+            runtime_checkpoint: None,
             backoff_until: None,
         };
         rustfs_common::mrf_channel::note_mrf_verified_repair(MrfVerifiedRepairEvent {
@@ -1142,14 +1251,108 @@ mod tests {
             runtime.retained_replay_journal(),
             "anchor must retain the startup journal before proof is consumed"
         );
+        assert_eq!(
+            runtime.replay_cleanup_to_delete(),
+            None,
+            "the committed replay source must not be reclaimed before the exact proof"
+        );
         runtime.discharge_durable_replay_anchors();
         assert!(
             !runtime.retained_replay_journal(),
             "matching verified proof discharges the durable replay anchor"
         );
+        assert_eq!(
+            runtime.replay_cleanup_to_delete(),
+            Some(cleanup),
+            "proof discharge must preserve the committed owner/sequence cleanup target"
+        );
         rustfs_common::mrf_channel::release_mrf_intent(&intent);
     }
 
+    #[test]
+    fn runtime_cleanup_defaults_to_legacy_for_runtime_written_journals() {
+        let runtime = MrfRuntime {
+            queue: MrfQueue::new(2, usize::MAX),
+            config: MrfConsumerConfig::default(),
+            checkpoint_owner: Uuid::new_v4(),
+            next_checkpoint_sequence: 1,
+            new_since_flush: 0,
+            dirty: false,
+            journal_on_disk: true,
+            retain_replay_journal: false,
+            durable_replay_anchors: Vec::new(),
+            replay_cleanup: None,
+            runtime_checkpoint: None,
+            backoff_until: None,
+        };
+
+        assert_eq!(
+            runtime.replay_cleanup_to_delete(),
+            Some(ReplayCleanup::Legacy),
+            "journals written by the runtime still use the legacy cleanup path"
+        );
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    #[serial]
+    async fn runtime_idle_cleanup_deletes_runtime_and_replay_recovery_anchors() {
+        let _env = rustfs_test_utils::TestECStoreEnv::builder()
+            .prefix("rustfs_mrf_runtime_idle_cleanup")
+            .build()
+            .await;
+        let disks = journal_disks().await;
+        assert!(!disks.is_empty(), "test environment must register local disks");
+
+        let replay_owner = Uuid::new_v4();
+        let runtime_owner = Uuid::new_v4();
+        let config = MrfConsumerConfig::default();
+        let journal_max_bytes = config.journal_max_bytes;
+        let replay_payload = encoded_payload(&intent("cleanup-bucket", "replay-object", 0));
+        let runtime_payload = encoded_payload(&intent("cleanup-bucket", "runtime-object", 0));
+        snapshot::publish_committed_snapshot(&disks, replay_owner, 7, &replay_payload, journal_max_bytes)
+            .await
+            .expect("publish retained replay checkpoint");
+        snapshot::publish_committed_snapshot(&disks, runtime_owner, 8, &runtime_payload, journal_max_bytes)
+            .await
+            .expect("publish runtime checkpoint");
+        assert!(write_journal(MRF_SCOPED_JOURNAL_PATH, &runtime_payload).await);
+        assert!(write_journal(MRF_JOURNAL_PATH, &runtime_payload).await);
+
+        let mut runtime = MrfRuntime {
+            queue: MrfQueue::new(2, usize::MAX),
+            config,
+            checkpoint_owner: Uuid::new_v4(),
+            next_checkpoint_sequence: 9,
+            new_since_flush: 0,
+            dirty: false,
+            journal_on_disk: true,
+            retain_replay_journal: false,
+            durable_replay_anchors: Vec::new(),
+            replay_cleanup: Some(ReplayCleanup::Committed {
+                owner: replay_owner,
+                sequence: 7,
+            }),
+            runtime_checkpoint: Some((runtime_owner, 8)),
+            backoff_until: None,
+        };
+
+        assert!(
+            runtime.delete_idle_recovery_anchors().await,
+            "idle cleanup should remove both runtime and replay recovery anchors"
+        );
+        assert_eq!(runtime.replay_cleanup, None);
+        assert_eq!(runtime.runtime_checkpoint, None);
+        assert!(
+            snapshot::inspect_local_committed_snapshot(journal_max_bytes)
+                .await
+                .expect("inspect committed checkpoints after cleanup")
+                .is_none(),
+            "both committed checkpoint generations must be gone after idle cleanup"
+        );
+        assert_eq!(read_journal(MRF_SCOPED_JOURNAL_PATH).await, None);
+        assert_eq!(read_journal(MRF_JOURNAL_PATH).await, None);
+    }
+
     #[test]
     fn durable_replay_acquires_a_fresh_lease_before_manager_admission() {
         let unique = uuid::Uuid::new_v4();

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

@@ -22,8 +22,9 @@
 //! responsibility through a newer durable snapshot or a verified repair proof.
 //! An unreadable commit path cannot prove that only legacy data exists. This
 //! explicit inspection API fails closed and never mutates recovery anchors.
-//! It is wired into the replay reader before writer activation, but the writer
-//! remains gated on ownership-aware handoff.
+//! The live consumer writes committed checkpoints alongside the scoped and
+//! legacy journal mirrors; cleanup remains gated by replay ownership and exact
+//! verified repair proof handoff.
 //! One surviving committed replica supports process restart recovery only;
 //! this reader does not establish a replication quorum or a power-loss policy.
 
@@ -356,8 +357,8 @@ fn validate_reusable_manifest_slot(existing: Option<&[u8]>, sequence: u64, paylo
 /// The writer is a narrow production primitive for the ownership-aware MRF
 /// handoff: it validates the whole journal payload, preserves the previous
 /// committed slot, and publishes the manifest only after the successor payload
-/// reaches the same disk. It does not delete legacy journals, tombstone older
-/// anchors, or activate the live consumer.
+/// reaches the same disk. It does not delete legacy journals or tombstone older
+/// anchors by itself; the consumer decides cleanup after replay handoff.
 pub async fn publish_committed_snapshot(
     disks: &[EcstoreDiskStore],
     owner: Uuid,
@@ -371,7 +372,10 @@ pub async fn publish_committed_snapshot(
     if owner.is_nil() || sequence == 0 || sequence == u64::MAX {
         return Err(SnapshotError::Corrupt);
     }
-    if payload.len() > limit || decode_journal(payload).1 != 0 {
+    if payload.len() > limit {
+        return Err(SnapshotError::TooLarge);
+    }
+    if decode_journal(payload).1 != 0 {
         return Err(SnapshotError::Corrupt);
     }
     let current = read_committed(disks, limit).await?;
@@ -555,13 +559,13 @@ pub async fn inspect_local_committed_snapshot(max_bytes: usize) -> Result<Option
     read_committed(&super::journal_disks().await, max_bytes).await
 }
 
-/// Remove committed manifests from `owner` whose sequence is no newer than
+/// Remove committed checkpoints from `owner` whose sequence is no newer than
 /// `committed_through`.
 ///
-/// Payload files are intentionally left as orphans after their manifest is
-/// removed. Readers cannot discover a payload without its matching manifest,
-/// and deleting manifests first prevents an older retained slot from becoming
-/// visible again after the newest replay has been fully discharged.
+/// Cleanup is manifest-first so readers cannot rediscover an older payload
+/// after the newest replay has been fully discharged. The payload is removed
+/// only after the manifest and body were revalidated as one complete committed
+/// checkpoint; damaged, future, mismatched, or foreign-owner slots are retained.
 pub async fn delete_committed_snapshots_through(
     owner: Uuid,
     committed_through: u64,
@@ -580,11 +584,10 @@ async fn delete_committed_snapshots_through_on(
     if disks.is_empty() {
         return Err(SnapshotError::NoWritableReplica);
     }
-    let mut any_changed = false;
     let mut first_error = None;
     for disk in disks {
-        for path in MANIFEST_PATHS {
-            let existing = match read_bounded(disk, path, MANIFEST_LEN).await {
+        for (manifest_path, payload_path) in MANIFEST_PATHS.into_iter().zip(PAYLOAD_PATHS) {
+            let manifest_bytes = match read_bounded(disk, manifest_path, MANIFEST_LEN).await {
                 Ok(Some(existing)) => existing,
                 Ok(None) => continue,
                 Err(error) => {
@@ -594,7 +597,7 @@ async fn delete_committed_snapshots_through_on(
                     continue;
                 }
             };
-            let manifest = match Manifest::decode(&existing, max_bytes) {
+            let manifest = match Manifest::decode(&manifest_bytes, max_bytes) {
                 Ok(manifest) => manifest,
                 Err(error) => {
                     if first_error.is_none() {
@@ -606,16 +609,58 @@ async fn delete_committed_snapshots_through_on(
             if manifest.owner != owner || manifest.sequence > committed_through {
                 continue;
             }
+            let payload_bytes = match read_bounded(disk, payload_path, manifest.payload_len).await {
+                Ok(Some(payload)) => payload,
+                Ok(None) => {
+                    if first_error.is_none() {
+                        first_error = Some(SnapshotError::Corrupt);
+                    }
+                    continue;
+                }
+                Err(error) => {
+                    if first_error.is_none() {
+                        first_error = Some(error);
+                    }
+                    continue;
+                }
+            };
+            if let Err(error) = CommittedSnapshot::decode(0, &manifest_bytes, payload_bytes.clone(), max_bytes) {
+                if first_error.is_none() {
+                    first_error = Some(error);
+                }
+                continue;
+            }
             match EcstoreDiskAPI::compare_and_update_file(
                 disk.as_ref(),
                 RUSTFS_META_BUCKET,
-                path,
-                Some(EcstoreDiskBytes::from(existing)),
+                manifest_path,
+                Some(EcstoreDiskBytes::copy_from_slice(&manifest_bytes)),
                 None,
             )
             .await
             {
-                Ok(EcstoreConditionalFileUpdate::Updated) => any_changed = true,
+                Ok(EcstoreConditionalFileUpdate::Updated) => {
+                    match EcstoreDiskAPI::compare_and_update_file(
+                        disk.as_ref(),
+                        RUSTFS_META_BUCKET,
+                        payload_path,
+                        Some(EcstoreDiskBytes::copy_from_slice(&payload_bytes)),
+                        None,
+                    )
+                    .await
+                    {
+                        Ok(
+                            EcstoreConditionalFileUpdate::Updated
+                            | EcstoreConditionalFileUpdate::Missing
+                            | EcstoreConditionalFileUpdate::Mismatch,
+                        ) => {}
+                        Err(error) => {
+                            if first_error.is_none() {
+                                first_error = Some(SnapshotError::Disk(error));
+                            }
+                        }
+                    }
+                }
                 Ok(EcstoreConditionalFileUpdate::Missing | EcstoreConditionalFileUpdate::Mismatch) => {}
                 Err(error) => {
                     if first_error.is_none() {
@@ -625,13 +670,7 @@ async fn delete_committed_snapshots_through_on(
             }
         }
     }
-    if any_changed {
-        Ok(true)
-    } else if let Some(error) = first_error {
-        Err(error)
-    } else {
-        Ok(true)
-    }
+    if let Some(error) = first_error { Err(error) } else { Ok(true) }
 }
 
 async fn read_recovery_snapshot(disks: &[EcstoreDiskStore], limit: usize) -> Result<Option<RecoverySnapshot>, SnapshotError> {
@@ -1101,6 +1140,45 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn committed_snapshot_writer_capacity_failure_preserves_previous_anchor() {
+        let root = TempDir::new().expect("test directory");
+        let store = disk(&root, "disk").await;
+        let owner = Uuid::new_v4();
+        let old = payload("old");
+        let next = payload("next");
+        commit(&store, 0, owner, 1, &old).await;
+
+        let result = publish_committed_snapshot(std::slice::from_ref(&store), owner, 2, &next, next.len() - 1).await;
+
+        assert!(
+            matches!(result, Err(SnapshotError::TooLarge)),
+            "capacity failure must be reported separately from corruption: {result:?}"
+        );
+        let reopened = disk(&root, "disk").await;
+        let recovered = read_committed(std::slice::from_ref(&reopened), 4096)
+            .await
+            .expect("read previous committed snapshot")
+            .expect("old anchor remains committed");
+        assert_eq!(recovered.sequence(), 1);
+        assert_eq!(recovered.slot(), 0);
+        assert_eq!(recovered.payload(), old.as_slice());
+        assert_eq!(
+            EcstoreDiskAPI::read_all(reopened.as_ref(), RUSTFS_META_BUCKET, MANIFEST_PATHS[0])
+                .await
+                .expect("old manifest retained")
+                .as_ref(),
+            manifest(owner, 1, &old).as_slice()
+        );
+        assert!(
+            matches!(
+                EcstoreDiskAPI::read_all(reopened.as_ref(), RUSTFS_META_BUCKET, PAYLOAD_PATHS[1]).await,
+                Err(EcstoreDiskError::FileNotFound | EcstoreDiskError::VolumeNotFound)
+            ),
+            "oversized successor payload must not be staged"
+        );
+    }
+
     #[tokio::test]
     async fn committed_snapshot_writer_does_not_overwrite_damaged_inactive_manifest() {
         let root = TempDir::new().expect("test directory");
@@ -1355,7 +1433,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn committed_cleanup_removes_only_manifests_at_or_below_sequence() {
+    async fn committed_cleanup_removes_complete_checkpoint_at_or_below_sequence() {
         let root = TempDir::new().expect("test directory");
         let disk = disk(&root, "disk").await;
         let owner = Uuid::new_v4();
@@ -1377,12 +1455,19 @@ mod tests {
             ),
             "old manifest is gone, so the old payload cannot become visible again"
         );
+        assert!(
+            matches!(
+                EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, PAYLOAD_PATHS[0]).await,
+                Err(EcstoreDiskError::FileNotFound | EcstoreDiskError::VolumeNotFound)
+            ),
+            "old payload should be reclaimed after its manifest is removed"
+        );
         assert_eq!(
-            EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, PAYLOAD_PATHS[0])
+            EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, PAYLOAD_PATHS[1])
                 .await
-                .expect("old payload orphan may remain")
+                .expect("newer payload retained")
                 .as_ref(),
-            older
+            newer
         );
         let recovered = read_committed(std::slice::from_ref(&disk), 4096)
             .await
@@ -1392,6 +1477,40 @@ mod tests {
         assert_eq!(recovered.payload(), newer);
     }
 
+    #[tokio::test]
+    async fn committed_cleanup_retains_manifest_when_payload_identity_mismatches() {
+        let root = TempDir::new().expect("test directory");
+        let disk = disk(&root, "disk").await;
+        let owner = Uuid::new_v4();
+        let declared = payload("declared");
+        let actual = payload("actual");
+        install(&disk, MANIFEST_PATHS[0], &manifest(owner, 3, &declared)).await;
+        install(&disk, PAYLOAD_PATHS[0], &actual).await;
+
+        assert!(
+            matches!(
+                delete_committed_snapshots_through_on(std::slice::from_ref(&disk), owner, 3, 4096).await,
+                Err(SnapshotError::Corrupt)
+            ),
+            "cleanup must fail closed when the committed body no longer matches its manifest"
+        );
+
+        assert_eq!(
+            EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, MANIFEST_PATHS[0])
+                .await
+                .expect("mismatched manifest retained")
+                .as_ref(),
+            manifest(owner, 3, &declared)
+        );
+        assert_eq!(
+            EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, PAYLOAD_PATHS[0])
+                .await
+                .expect("mismatched payload retained")
+                .as_ref(),
+            actual
+        );
+    }
+
     #[tokio::test]
     async fn committed_cleanup_preserves_other_owner_manifests_within_sequence_window() {
         let root = TempDir::new().expect("test directory");
@@ -1417,6 +1536,13 @@ mod tests {
             ),
             "the replay owner's manifest is reclaimed"
         );
+        assert!(
+            matches!(
+                EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, PAYLOAD_PATHS[0]).await,
+                Err(EcstoreDiskError::FileNotFound | EcstoreDiskError::VolumeNotFound)
+            ),
+            "the replay owner's payload is reclaimed"
+        );
         assert_eq!(
             EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, MANIFEST_PATHS[1])
                 .await

crates/heal/tests/mrf_pipeline_test.rs

@@ -51,6 +51,8 @@ const JOURNAL_REL: &str = "buckets/.heal/mrf/journal.bin";
 const SCOPED_JOURNAL_REL: &str = "buckets/.heal/mrf/journal-scoped.bin";
 const COMMITTED_PAYLOAD_REL: &str = ".heal-mrf-snapshot.0.bin";
 const COMMITTED_MANIFEST_REL: &str = ".heal-mrf-commit.0.bin";
+const COMMITTED_PAYLOAD_RELS: [&str; 2] = [".heal-mrf-snapshot.0.bin", ".heal-mrf-snapshot.1.bin"];
+const COMMITTED_MANIFEST_RELS: [&str; 2] = [".heal-mrf-commit.0.bin", ".heal-mrf-commit.1.bin"];
 const COMMITTED_MAGIC: &[u8; 8] = b"RFMRFC01";
 const COMMITTED_MANIFEST_LEN: usize = 8 + 1 + 16 + 8 + 8 + 32 + 32;
 
@@ -235,6 +237,44 @@ fn journal_matches_on_all_disks(disk_paths: &[PathBuf], relative_path: &str, exp
         .all(|path| std::fs::read(path.join(META_BUCKET).join(relative_path)).is_ok_and(|actual| actual == expected))
 }
 
+fn committed_checkpoint_matches_on_all_disks(disk_paths: &[PathBuf], sequence: u64, expected_payload: &[u8]) -> bool {
+    disk_paths.iter().all(|path| {
+        let root = path.join(META_BUCKET);
+        COMMITTED_PAYLOAD_RELS
+            .into_iter()
+            .zip(COMMITTED_MANIFEST_RELS)
+            .any(|(payload_rel, manifest_rel)| {
+                let Ok(payload) = std::fs::read(root.join(payload_rel)) else {
+                    return false;
+                };
+                if payload != expected_payload {
+                    return false;
+                }
+                let Ok(manifest) = std::fs::read(root.join(manifest_rel)) else {
+                    return false;
+                };
+                if manifest.len() != COMMITTED_MANIFEST_LEN || &manifest[..8] != COMMITTED_MAGIC || manifest[8] != 1 {
+                    return false;
+                }
+                let Ok(recorded_sequence) = <[u8; 8]>::try_from(&manifest[25..33]).map(u64::from_le_bytes) else {
+                    return false;
+                };
+                let Ok(recorded_len) = <[u8; 8]>::try_from(&manifest[33..41]).map(u64::from_le_bytes) else {
+                    return false;
+                };
+                let Ok(expected_len) = u64::try_from(expected_payload.len()) else {
+                    return false;
+                };
+                if recorded_sequence != sequence || recorded_len != expected_len {
+                    return false;
+                }
+                let payload_digest: [u8; 32] = Sha256::digest(expected_payload).into();
+                let manifest_digest: [u8; 32] = Sha256::digest(&manifest[..COMMITTED_MANIFEST_LEN - 32]).into();
+                payload_digest.as_slice() == &manifest[41..73] && manifest_digest.as_slice() == &manifest[73..]
+            })
+    })
+}
+
 async fn wait_until<F, Fut>(deadline: Duration, mut probe: F) -> bool
 where
     F: FnMut() -> Fut,
@@ -629,13 +669,14 @@ fn mrf_successor_flush_child_process_fixture() {
         let expected_successor = journal_record(1, "successor-bucket", "second-object", None, 2);
         let flushed = wait_until(Duration::from_secs(10), || async {
             manager.operations_snapshot().await.queued_by_source.mrf == 1
+                && committed_checkpoint_matches_on_all_disks(&disk_paths, 2, &expected_successor)
                 && journal_matches_on_all_disks(&disk_paths, SCOPED_JOURNAL_REL, &expected_successor)
                 && journal_matches_on_all_disks(&disk_paths, JOURNAL_REL, &expected_successor)
         })
         .await;
         assert!(
             flushed,
-            "child process must publish the pending successor snapshot before the delete phase"
+            "child process must publish the committed pending successor before the delete phase"
         );
     });
     std::process::exit(78);
@@ -676,13 +717,14 @@ fn mrf_successor_flush_waiting_child_process_fixture() {
         let expected_successor = journal_record(1, "service-kill-bucket", "second-object", None, 2);
         let flushed = wait_until(Duration::from_secs(10), || async {
             manager.operations_snapshot().await.queued_by_source.mrf == 1
+                && committed_checkpoint_matches_on_all_disks(&disk_paths, 2, &expected_successor)
                 && journal_matches_on_all_disks(&disk_paths, SCOPED_JOURNAL_REL, &expected_successor)
                 && journal_matches_on_all_disks(&disk_paths, JOURNAL_REL, &expected_successor)
         })
         .await;
         assert!(
             flushed,
-            "child process must publish the pending successor snapshot before it can be killed"
+            "child process must publish the committed pending successor before it can be killed"
         );
         std::fs::write(&ready_path, b"ready").expect("write ready marker");
         loop {

docs/operations/scanner-benchmark-runbook.md

@@ -39,8 +39,8 @@ The `scanner` and `heal` subsystems are served by `GetConfigKVHandler` (`rustfs/
 The `--abba` mode runs five independent scenario cells: `cold-hot`, `fresh-hot`,
 `multi-hot-new`, `running-heal`, and `mrf-replay`. Each scenario runs at least
 three A1/B1/B2/A2 groups for both baseline/candidate with background work on,
-and candidate-only background off/on. A measured leg lasts at least 900
-seconds; the minimum matrix contains 120 legs (30 hours before setup/oracles).
+and candidate-only background off/on. A measured release leg lasts at least 7200
+seconds; the minimum matrix contains 120 legs (240 hours before setup/oracles).
 The existing `performance-ab.yml` supplies the pattern for immutable build
 provenance and failure propagation, but its short Warp workload is not this
 scanner gate. No scheduled workflow starts this matrix automatically.
@@ -63,7 +63,7 @@ The manifest has the following JSON contract (all fields are required):
 | Field | Value |
 |---|---|
 | `schema`, `evidence` | `1`, and `measured` or `synthetic`. |
-| `rounds`, `duration_seconds`, `min_free_bytes` | 3..10 groups, 900..86400 seconds for measured runs, and the independently estimated free-space reservation in bytes. Synthetic runs may use 1 second. |
+| `rounds`, `duration_seconds`, `min_free_bytes` | 3..10 groups, 7200..86400 seconds for measured release runs, and the independently estimated free-space reservation in bytes. Synthetic runs may use 1 second. |
 | `baseline`, `candidate` | Each contains executable `binary`, full 40-character `revision`, and verified `sha256`. The runner rehashes binaries before every leg. |
 | `fixed` | `config_sha256`, `dataset_sha256`, `release_flags`, `durability`, `disk_type`, `cache_state`, `load_command`, `resource_isolation`, `topology` (`EC8+4`), and positive `offered_load_ops`. Hashes use 64 lowercase hexadecimal characters. |
 | `release_evidence` | Required for `measured` runs. It binds the 3x4 EC8+4 topology, multi-pool/multi-set coverage, per-node metrics endpoints, same-window distributed sampling, process restart and crash-restart fault modes, mixed-version reader/writer/rollback participation, and allocation/flamegraph/RSS/save-frequency profile artifact requirements. Synthetic runs do not need this field and still cannot approve release evidence. |

docs/testing/ci-gates.md

@@ -288,9 +288,13 @@ measured durations, P3's pressure run needs at least two hours, and P1 needs a
 symbolized profile summary with resolved samples. Every G14 field and every
 performance gate's fields must also share one `measurement_window_id`, so EC8+4,
 multi-set/multi-pool, ABBA, throughput, and profiling artifacts cannot be
-stitched together from unrelated runs. Missing, synthetic, stale, tampered,
-undersized, or topology-mismatched evidence returns a compact blocked or invalid
-JSON result and a nonzero exit.
+stitched together from unrelated runs. P1 `profile_evidence` must bind every
+required profile artifact kind (`allocation-profile`, `flamegraph`,
+`rss-samples`, and `save-frequency`) with a relative path, artifact format,
+non-empty file, matching SHA256, and the same measurement window when a
+per-artifact window is declared. Missing, synthetic, stale, tampered, undersized,
+or topology-mismatched evidence returns a compact blocked or invalid JSON result
+and a nonzero exit.
 
 This command validates the evidence package; it does not create evidence. A
 handwritten JSON file, a synthetic harness pass, a single focused case, or a

scripts/check_test_wiring.py

@@ -23,7 +23,16 @@ try:
 except ModuleNotFoundError:
     import tomli as tomllib
 
-from scanner_abba import MAX_JSON_BYTES, digest, number, read_json, require, sha, write_json
+from scanner_abba import (
+    MAX_JSON_BYTES,
+    RELEASE_PROFILE_ARTIFACTS,
+    digest,
+    number,
+    read_json,
+    require,
+    sha,
+    write_json,
+)
 
 
 ROOT = Path(__file__).resolve().parents[1]
@@ -80,12 +89,12 @@ SCANNER_HEAL_RELEASE_MIXED_VERSION_ROLES = {
     ("R-L", "migration_gap_evidence"): "migration-gap",
     ("R-L", "crash_safe_source_retirement_evidence"): "crash-safe-source-retirement",
 }
-SCANNER_HEAL_RELEASE_PROFILE_ARTIFACTS = (
-    "allocation-profile",
-    "flamegraph",
-    "rss-samples",
-    "save-frequency",
-)
+SCANNER_HEAL_RELEASE_MRF_DURABLE_REPLAY_FIELDS = {
+    ("G07", "mrf_responsibility_oracle"),
+    ("G07", "commit_boundary_crash_matrix"),
+    ("P4", "mrf_replay_cost_measurement"),
+    ("P4", "retained_responsibility_evidence"),
+}
 SCHEDULED_ALERT_WORKFLOWS = tuple(
     item["workflow"]
     for item in json.loads((ROOT / ".github/scheduled-validations.json").read_text())
@@ -1324,6 +1333,7 @@ def release_bundle_artifact_path(bundle_path: Path, raw_path: object, gate: str,
     resolved = (bundle_path.parent / path).resolve()
     require(resolved.is_relative_to(bundle_path.parent.resolve()), f"{gate}.{field} artifact path escapes bundle directory")
     require(resolved.is_file(), f"{gate}.{field} artifact is missing")
+    require(resolved.stat().st_size > 0, f"{gate}.{field} artifact is empty")
     return resolved
 
 
@@ -1371,7 +1381,23 @@ def validate_release_bundle_artifact(bundle_path: Path, source_revision: str, ga
         crash_points = evidence.get("crash_points")
         require(isinstance(crash_points, list) and crash_points,
                 f"{gate}.{field} requires crash-boundary evidence")
+    if (gate, field) in SCANNER_HEAL_RELEASE_MRF_DURABLE_REPLAY_FIELDS:
+        evidence_integer(evidence.get("replayed_records"), f"{gate}.{field}.replayed_records", 1, 2**63 - 1)
+        require(evidence.get("responsibility_anchor_retained") is True,
+                f"{gate}.{field} requires retained MRF responsibility anchors")
+        require(evidence.get("successor_snapshot_published") is True,
+                f"{gate}.{field} requires successor snapshot publication evidence")
     if gate == "G14":
+        if field == "same_window_field_evidence":
+            required_fields = set(SCANNER_HEAL_RELEASE_BUNDLE_REQUIRED_EVIDENCE_FIELDS["G14"]) - {field}
+            same_window_fields = evidence.get("same_window_fields")
+            require(isinstance(same_window_fields, list) and
+                    len(set(same_window_fields)) == len(same_window_fields) and
+                    all(isinstance(item, str) and item in required_fields for item in same_window_fields),
+                    "G14.same_window_field_evidence requires named G14 field coverage")
+            missing_same_window_fields = sorted(required_fields - set(same_window_fields))
+            require(not missing_same_window_fields,
+                    "G14.same_window_field_evidence missing fields: " + ", ".join(missing_same_window_fields))
         if field == "ec8_4_evidence":
             topology = evidence.get("topology")
             require(isinstance(topology, dict), "G14.ec8_4_evidence missing topology")
@@ -1387,14 +1413,28 @@ def validate_release_bundle_artifact(bundle_path: Path, source_revision: str, ga
     if field == "profile_evidence":
         evidence_integer(evidence.get("resolved_samples"), f"{gate}.{field}.resolved_samples", 1, 2**63 - 1)
         profile_artifacts = evidence.get("profile_artifacts")
-        require(isinstance(profile_artifacts, list) and
-                len(set(profile_artifacts)) == len(profile_artifacts) and
-                all(isinstance(artifact, str) and re.fullmatch(r"[a-z0-9][a-z0-9-]{1,63}", artifact) is not None
-                    for artifact in profile_artifacts),
-                f"{gate}.{field} requires named profile artifacts")
-        missing_profile_artifacts = sorted(set(SCANNER_HEAL_RELEASE_PROFILE_ARTIFACTS) - set(profile_artifacts))
-        require(not missing_profile_artifacts,
-                f"{gate}.{field} missing profile artifacts: {', '.join(missing_profile_artifacts)}")
+        require(isinstance(profile_artifacts, dict), f"{gate}.{field} missing profile artifacts")
+        missing_artifacts = sorted(set(RELEASE_PROFILE_ARTIFACTS) - set(profile_artifacts))
+        require(not missing_artifacts,
+                f"{gate}.{field} missing profile artifacts: {', '.join(missing_artifacts)}")
+        unknown_artifacts = sorted(set(profile_artifacts) - set(RELEASE_PROFILE_ARTIFACTS))
+        require(not unknown_artifacts,
+                f"{gate}.{field} unknown profile artifacts: {', '.join(unknown_artifacts)}")
+        for artifact_kind in RELEASE_PROFILE_ARTIFACTS:
+            item = profile_artifacts[artifact_kind]
+            require(isinstance(item, dict), f"{gate}.{field}.{artifact_kind} must be an object")
+            artifact_field = f"{field}.{artifact_kind}"
+            artifact_path = release_bundle_artifact_path(bundle_path, item.get("artifact"), gate, artifact_field)
+            require(sha(item.get("sha256")) and digest(artifact_path) == item["sha256"],
+                    f"{gate}.{artifact_field} artifact hash mismatch")
+            evidence_string(item.get("artifact_format"), f"{gate}.{artifact_field}.artifact_format",
+                            r"[A-Za-z0-9][A-Za-z0-9._+:-]{1,63}")
+            if "measurement_window_id" in item:
+                require(item["measurement_window_id"] == window_id,
+                        f"{gate}.{artifact_field} measurement window mismatch")
+            if "resolved_samples" in item:
+                evidence_integer(item.get("resolved_samples"), f"{gate}.{artifact_field}.resolved_samples",
+                                 0, 2**63 - 1)
     return window_id
 
 
@@ -1754,15 +1794,34 @@ class SelfTests(unittest.TestCase):
                     evidence["mixed_version_role"] = SCANNER_HEAL_RELEASE_MIXED_VERSION_ROLES[(gate, field)]
                 if gate in ("G04", "G07", "R-E", "R-L"):
                     evidence["crash_points"] = ["before-commit"]
+                if (gate, field) in SCANNER_HEAL_RELEASE_MRF_DURABLE_REPLAY_FIELDS:
+                    evidence["replayed_records"] = 2
+                    evidence["responsibility_anchor_retained"] = True
+                    evidence["successor_snapshot_published"] = True
                 if gate == "G14" and field == "ec8_4_evidence":
                     evidence["topology"] = {"erasure": "EC8+4", "nodes": 3, "drives_per_node": 4}
+                if gate == "G14" and field == "same_window_field_evidence":
+                    evidence["same_window_fields"] = [
+                        item
+                        for item in SCANNER_HEAL_RELEASE_BUNDLE_REQUIRED_EVIDENCE_FIELDS["G14"]
+                        if item != "same_window_field_evidence"
+                    ]
                 if gate == "G14" and field == "multi_set_evidence":
                     evidence["sets"] = 2
                 if gate == "G14" and field == "multi_pool_evidence":
                     evidence["pools"] = 2
                 if field == "profile_evidence":
                     evidence["resolved_samples"] = 1
-                    evidence["profile_artifacts"] = list(SCANNER_HEAL_RELEASE_PROFILE_ARTIFACTS)
+                    artifacts = {}
+                    for artifact_kind in RELEASE_PROFILE_ARTIFACTS:
+                        artifact = artifact_dir / f"{gate}-{field}-{artifact_kind}.json"
+                        write_json(artifact, {"gate": gate, "field": field, "artifact": artifact_kind})
+                        artifacts[artifact_kind] = {
+                            "artifact": artifact.relative_to(bundle_dir).as_posix(),
+                            "sha256": digest(artifact),
+                            "artifact_format": "json",
+                        }
+                    evidence["profile_artifacts"] = artifacts
                 fields[field] = evidence
             gates[gate] = {
                 "status": "pass",
@@ -1786,7 +1845,7 @@ class SelfTests(unittest.TestCase):
             self.assertEqual(status["pending_lanes"], [])
 
     def test_scanner_heal_release_bundle_rejects_synthetic_or_missing_fields(self) -> None:
-        for fault in ("synthetic", "missing-field", "hash"):
+        for fault in ("synthetic", "missing-field", "hash", "empty-artifact"):
             with self.subTest(fault=fault), tempfile.TemporaryDirectory() as tmp:
                 root, bundle = self.scanner_heal_release_bundle_fixture(Path(tmp))
                 data = read_json(bundle)
@@ -1794,9 +1853,12 @@ class SelfTests(unittest.TestCase):
                     data["evidence"] = "synthetic"
                 elif fault == "missing-field":
                     del data["gates"]["G09"]["evidence_fields"]["rollback_payload_evidence"]
-                else:
+                elif fault == "hash":
                     artifact = bundle.parent / data["gates"]["G01"]["evidence_fields"]["root_authority_evidence"]["artifact"]
                     artifact.write_text(artifact.read_text(encoding="utf-8") + "\n", encoding="utf-8")
+                else:
+                    artifact = bundle.parent / data["gates"]["G01"]["evidence_fields"]["root_authority_evidence"]["artifact"]
+                    artifact.write_text("", encoding="utf-8")
                 write_json(bundle, data)
 
                 with mock.patch("subprocess.check_output", return_value="b" * 40):
@@ -1816,10 +1878,35 @@ class SelfTests(unittest.TestCase):
             ("missing-duration", "P1", "cold_walk_share_measurement", lambda item: item.pop("duration_seconds"), "duration_seconds"),
             ("duration", "P3", "two_hour_pressure_measurement", lambda item: item.update({"duration_seconds": 7199}), "two hours"),
             ("profile", "P1", "profile_evidence", lambda item: item.pop("resolved_samples"), "resolved_samples"),
-            ("profile-artifact", "P1", "profile_evidence", lambda item: item.update({"profile_artifacts": ["allocation-profile", "rss-samples", "save-frequency"]}), "profile artifacts"),
-            ("duplicate-profile-artifact", "P1", "profile_evidence", lambda item: item.update({"profile_artifacts": ["allocation-profile", "allocation-profile", "flamegraph", "rss-samples", "save-frequency"]}), "named profile artifacts"),
+            (
+                "profile-artifact",
+                "P1",
+                "profile_evidence",
+                lambda item: item["profile_artifacts"].pop("flamegraph"),
+                "missing profile artifacts",
+            ),
+            (
+                "profile-artifact-hash",
+                "P1",
+                "profile_evidence",
+                lambda item: item["profile_artifacts"]["rss-samples"].update({"sha256": "0" * 64}),
+                "artifact hash mismatch",
+            ),
+            (
+                "profile-artifact-window",
+                "P1",
+                "profile_evidence",
+                lambda item: item["profile_artifacts"]["save-frequency"].update(
+                    {"measurement_window_id": "p1-different-window"}
+                ),
+                "measurement window mismatch",
+            ),
             ("versions", "G09", "mixed_version_reader_evidence", lambda item: item.update({"versions": [1, 2]}), "mixed-version"),
             ("stale-versions", "G09", "mixed_version_writer_evidence", lambda item: item.update({"versions": ["a" * 40, "c" * 40]}), "tested source revision"),
+            ("mrf-records", "G07", "mrf_responsibility_oracle", lambda item: item.pop("replayed_records"), "replayed_records"),
+            ("mrf-anchor", "G07", "commit_boundary_crash_matrix", lambda item: item.update({"responsibility_anchor_retained": False}), "retained MRF responsibility anchors"),
+            ("mrf-successor", "P4", "retained_responsibility_evidence", lambda item: item.pop("successor_snapshot_published"), "successor snapshot"),
+            ("same-window-fields", "G14", "same_window_field_evidence", lambda item: item.update({"same_window_fields": ["ec8_4_evidence", "multi_set_evidence"]}), "missing fields"),
         ):
             with self.subTest(fault=fault), tempfile.TemporaryDirectory() as tmp:
                 root, bundle = self.scanner_heal_release_bundle_fixture(Path(tmp))

scripts/run_scanner_heal_evidence_case.sh

@@ -74,6 +74,40 @@ print("test(/^" + re.escape(case["name"]) + "$/)")
 PY
 }
 
+runtime_profile_for() {
+    local case_id="$1"
+    case "$case_id" in
+        background-target-crash|background-target-restart)
+            echo "background-4x1"
+            ;;
+        background-target-crash-ec8-4|background-target-restart-ec8-4)
+            echo "background-ec8-4"
+            ;;
+        ec84-target-drive-restart)
+            echo "distributed-ec8-4"
+            ;;
+        *)
+            echo "default"
+            ;;
+    esac
+}
+
+apply_runtime_profile() {
+    local case_id="$1"
+    case "$(runtime_profile_for "$case_id")" in
+        background-4x1)
+            export RUSTFS_HEAL_CHAOS_OBJECT_COUNT="${RUSTFS_HEAL_CHAOS_OBJECT_COUNT:-64}"
+            export RUSTFS_HEAL_CHAOS_OBJECT_SIZE_BYTES="${RUSTFS_HEAL_CHAOS_OBJECT_SIZE_BYTES:-16777216}"
+            export RUSTFS_HEAL_CHAOS_PARTIAL_TIMEOUT_SECS="${RUSTFS_HEAL_CHAOS_PARTIAL_TIMEOUT_SECS:-120}"
+            ;;
+        background-ec8-4)
+            export RUSTFS_HEAL_CHAOS_OBJECT_COUNT="${RUSTFS_HEAL_CHAOS_OBJECT_COUNT:-32}"
+            export RUSTFS_HEAL_CHAOS_OBJECT_SIZE_BYTES="${RUSTFS_HEAL_CHAOS_OBJECT_SIZE_BYTES:-8388608}"
+            export RUSTFS_HEAL_CHAOS_PARTIAL_TIMEOUT_SECS="${RUSTFS_HEAL_CHAOS_PARTIAL_TIMEOUT_SECS:-180}"
+            ;;
+    esac
+}
+
 test_binary_from_listing() {
     local listing="$1"
     local case_id="$2"
@@ -130,9 +164,11 @@ run_self_test() {
         local expected_filter expected_profile plan
         expected_filter="$(test_filter_for "$case_id")"
         expected_profile="$(case_field "$case_id" lane)"
+        expected_runtime_profile="$(runtime_profile_for "$case_id")"
         plan="$("$0" --case "$case_id" --plan-only)"
         if [[ "$plan" != *"case=$case_id"* ]] ||
             [[ "$plan" != *"profile=$expected_profile"* ]] ||
+            [[ "$plan" != *"runtime_profile=$expected_runtime_profile"* ]] ||
             [[ "$plan" != *"filter=$expected_filter"* ]] ||
             [[ "$plan" != *"run_dir=$ROOT/target/scanner-heal-evidence/$case_id-"* ]]; then
             echo "self-test failed: registry case plan mismatch for $case_id" >&2
@@ -185,13 +221,7 @@ TEST_FILTER="$(test_filter_for "$CASE_ID")"
 if [[ -z "$PROFILE" ]]; then
     PROFILE="$(case_field "$CASE_ID" lane)"
 fi
-case "$CASE_ID" in
-  background-target-crash|background-target-restart)
-    export RUSTFS_HEAL_CHAOS_OBJECT_COUNT="${RUSTFS_HEAL_CHAOS_OBJECT_COUNT:-64}"
-    export RUSTFS_HEAL_CHAOS_OBJECT_SIZE_BYTES="${RUSTFS_HEAL_CHAOS_OBJECT_SIZE_BYTES:-16777216}"
-    export RUSTFS_HEAL_CHAOS_PARTIAL_TIMEOUT_SECS="${RUSTFS_HEAL_CHAOS_PARTIAL_TIMEOUT_SECS:-120}"
-    ;;
-esac
+apply_runtime_profile "$CASE_ID"
 if [[ -z "$RUN_DIR" ]]; then
     RUN_DIR="$ROOT/target/scanner-heal-evidence/${CASE_ID}-$(date -u +%Y%m%dT%H%M%SZ)"
 elif [[ "$RUN_DIR" != /* ]]; then
@@ -201,6 +231,7 @@ fi
 if [[ "$PLAN_ONLY" == 1 ]]; then
     echo "case=$CASE_ID"
     echo "profile=$PROFILE"
+    echo "runtime_profile=$(runtime_profile_for "$CASE_ID")"
     echo "filter=$TEST_FILTER"
     echo "run_dir=$RUN_DIR"
     exit 0

scripts/scanner_abba.py

@@ -35,6 +35,7 @@ RELEASE_PROFILE_ARTIFACTS = (
     "rss-samples",
     "save-frequency",
 )
+MIN_MEASURED_RELEASE_DURATION_SECONDS = 7200
 RELEASE_FAULT_MODES = (
     "process-restart",
     "process-crash-restart",
@@ -130,7 +131,7 @@ def validate_manifest(manifest):
     number(fixed.get("offered_load_ops"), "offered load", 1)
     require(type(manifest.get("rounds")) is int and 3 <= manifest["rounds"] <= 10,
             "rounds must be 3..10")
-    minimum = 900 if manifest["evidence"] == "measured" else 1
+    minimum = MIN_MEASURED_RELEASE_DURATION_SECONDS if manifest["evidence"] == "measured" else 1
     require(type(manifest.get("duration_seconds")) is int and
             minimum <= manifest["duration_seconds"] <= 86400, "invalid duration_seconds")
     number(manifest.get("min_free_bytes"), "min_free_bytes", 1)
@@ -173,6 +174,18 @@ def release_evidence_true(value, name):
     require(value is True, f"missing release_evidence.{name}")
 
 
+def release_evidence_exact_strings(value, expected, name):
+    require(isinstance(value, list) and all(isinstance(item, str) and item.strip() for item in value),
+            f"invalid release_evidence.{name}")
+    observed = set(value)
+    require(len(observed) == len(value), f"duplicate release_evidence.{name}")
+    missing = sorted(set(expected) - observed)
+    require(not missing, f"missing release_evidence.{name}: {', '.join(missing)}")
+    unknown = sorted(observed - set(expected))
+    require(not unknown, f"unknown release_evidence.{name}: {', '.join(unknown)}")
+    return value
+
+
 def validate_release_evidence_manifest(manifest):
     if manifest["evidence"] != "measured":
         return
@@ -210,17 +223,17 @@ def validate_release_evidence_manifest(manifest):
 
     crash = evidence.get("crash_restart")
     require(isinstance(crash, dict), "missing release_evidence.crash_restart")
-    fault_modes = crash.get("fault_modes")
-    require(
-        isinstance(fault_modes, list)
-        and all(mode in fault_modes for mode in RELEASE_FAULT_MODES)
-        and all(isinstance(mode, str) and mode.strip() for mode in fault_modes),
-        "missing release_evidence.crash_restart.fault_modes",
-    )
+    release_evidence_exact_strings(crash.get("fault_modes"), RELEASE_FAULT_MODES, "crash_restart.fault_modes")
     release_evidence_true(crash.get("unclean_shutdown_marker"), "crash_restart.unclean_shutdown_marker")
 
     mixed = evidence.get("mixed_version")
     require(isinstance(mixed, dict), "missing release_evidence.mixed_version")
+    baseline_revision = manifest["baseline"]["revision"]
+    candidate_revision = manifest["candidate"]["revision"]
+    require(baseline_revision != candidate_revision,
+            "release_evidence.mixed_version requires distinct baseline and candidate revisions")
+    require(manifest["baseline"]["sha256"] != manifest["candidate"]["sha256"],
+            "release_evidence.mixed_version requires distinct baseline and candidate binaries")
     revisions = mixed.get("participating_revisions")
     require(
         isinstance(revisions, list)
@@ -229,20 +242,15 @@ def validate_release_evidence_manifest(manifest):
                 for revision in revisions),
         "invalid release_evidence.mixed_version.participating_revisions",
     )
-    for revision in (manifest["baseline"]["revision"], manifest["candidate"]["revision"]):
+    for revision in (baseline_revision, candidate_revision):
         require(revision in revisions, "release_evidence.mixed_version omits tested build revision")
     for key in ("reader", "writer", "rollback_payload"):
         require(mixed.get(key) is True, f"missing release_evidence.mixed_version.{key}")
 
     profile = evidence.get("profile")
     require(isinstance(profile, dict), "missing release_evidence.profile")
-    artifacts = profile.get("required_artifacts")
-    require(
-        isinstance(artifacts, list)
-        and all(item in artifacts for item in RELEASE_PROFILE_ARTIFACTS)
-        and all(isinstance(item, str) and item.strip() for item in artifacts),
-        "missing release_evidence.profile.required_artifacts",
-    )
+    release_evidence_exact_strings(profile.get("required_artifacts"), RELEASE_PROFILE_ARTIFACTS,
+                                   "profile.required_artifacts")
     for key in ("collector_config_sha256", "profiler_config_sha256"):
         require(sha(profile.get(key)), f"invalid release_evidence.profile.{key}")
 

scripts/summarize_scanner_heal_perf.py

@@ -12,7 +12,12 @@ from pathlib import Path
 import sys
 from typing import Any
 
-from scanner_abba import LEGS, SCENARIOS, validate_release_evidence_manifest
+from scanner_abba import (
+    LEGS,
+    MIN_MEASURED_RELEASE_DURATION_SECONDS,
+    SCENARIOS,
+    validate_release_evidence_manifest,
+)
 
 MAX_JSON_BYTES = 1024 * 1024
 CACHE_COST_PREFIX = "CACHE_COST "
@@ -124,6 +129,9 @@ def require_measured_comparison_evidence(comparison: dict[str, Any], index: int)
 
 def require_complete_abba_matrix(manifest: dict[str, Any], report: dict[str, Any], comparisons: list[dict[str, Any]]) -> None:
     require(report.get("evidence") == manifest.get("evidence"), "manifest/report evidence mismatch")
+    require(type(manifest.get("duration_seconds")) is int and
+            manifest["duration_seconds"] >= MIN_MEASURED_RELEASE_DURATION_SECONDS,
+            "measured ABBA duration_seconds requires at least two hours")
     rounds = manifest.get("rounds")
     require(type(rounds) is int and 3 <= rounds <= 10, "invalid manifest.rounds")
     expected_cells = len(SCENARIOS) * 2 * rounds * len(LEGS)

scripts/test_scanner_abba.py

@@ -167,8 +167,15 @@ class ScannerAbbaTest(unittest.TestCase):
 
     def measured_manifest(self):
         manifest = copy.deepcopy(self.manifest)
-        manifest.update(evidence="measured", duration_seconds=900)
-        manifest["candidate"]["revision"] = "b" * 40
+        manifest.update(evidence="measured", duration_seconds=harness.MIN_MEASURED_RELEASE_DURATION_SECONDS)
+        candidate_binary = self.root / "candidate-python"
+        candidate_binary.write_bytes(self.binary.read_bytes() + b"\n")
+        candidate_binary.chmod(0o755)
+        manifest["candidate"] = {
+            "binary": str(candidate_binary),
+            "sha256": harness.digest(candidate_binary),
+            "revision": "b" * 40,
+        }
         manifest["release_evidence"] = {
             "topology": {
                 "nodes": 3,
@@ -527,7 +534,7 @@ class ScannerAbbaTest(unittest.TestCase):
         self.manifest["evidence"] = "measured"
         with self.assertRaisesRegex(ValueError, "duration_seconds"):
             harness.validate_manifest(self.manifest)
-        self.manifest["duration_seconds"] = 900
+        self.manifest["duration_seconds"] = harness.MIN_MEASURED_RELEASE_DURATION_SECONDS
         self.manifest["rounds"] = 2
         with self.assertRaisesRegex(ValueError, "rounds"):
             harness.validate_manifest(self.manifest)
@@ -548,6 +555,12 @@ class ScannerAbbaTest(unittest.TestCase):
             "missing crash": lambda manifest: manifest["release_evidence"]["crash_restart"].update(
                 fault_modes=["process-restart"],
             ),
+            "unknown crash": lambda manifest: manifest["release_evidence"]["crash_restart"].update(
+                fault_modes=["process-restart", "process-crash-restart", "kernel-panic"],
+            ),
+            "duplicate crash": lambda manifest: manifest["release_evidence"]["crash_restart"].update(
+                fault_modes=["process-restart", "process-restart", "process-crash-restart"],
+            ),
             "clean crash marker": lambda manifest: manifest["release_evidence"]["crash_restart"].update(
                 unclean_shutdown_marker=False,
             ),
@@ -555,9 +568,24 @@ class ScannerAbbaTest(unittest.TestCase):
             "missing candidate": lambda manifest: manifest["release_evidence"]["mixed_version"].update(
                 participating_revisions=["a" * 40, "c" * 40],
             ),
+            "same mixed revision": lambda manifest: manifest["candidate"].update(
+                revision=manifest["baseline"]["revision"],
+            ),
+            "same mixed binary": lambda manifest: manifest["candidate"].update(
+                binary=manifest["baseline"]["binary"],
+                sha256=manifest["baseline"]["sha256"],
+            ),
             "missing profile": lambda manifest: manifest["release_evidence"]["profile"].update(
                 required_artifacts=["allocation-profile", "flamegraph", "rss-samples"],
             ),
+            "unknown profile": lambda manifest: manifest["release_evidence"]["profile"].update(
+                required_artifacts=["allocation-profile", "flamegraph", "rss-samples", "save-frequency", "heapdump"],
+            ),
+            "duplicate profile": lambda manifest: manifest["release_evidence"]["profile"].update(
+                required_artifacts=[
+                    "allocation-profile", "flamegraph", "rss-samples", "save-frequency", "flamegraph",
+                ],
+            ),
             "bad profile hash": lambda manifest: manifest["release_evidence"]["profile"].update(
                 profiler_config_sha256="not-a-sha",
             ),

scripts/test_summarize_scanner_heal_perf.py

@@ -37,6 +37,7 @@ class ScannerHealPerfSummaryTest(unittest.TestCase):
             "schema": 1,
             "evidence": "measured",
             "rounds": 3,
+            "duration_seconds": summary.MIN_MEASURED_RELEASE_DURATION_SECONDS,
             "fixed": {
                 "config_sha256": "1" * 64,
                 "dataset_sha256": "2" * 64,
@@ -225,6 +226,19 @@ class ScannerHealPerfSummaryTest(unittest.TestCase):
                 with self.assertRaisesRegex(ValueError, "ABBA matrix|manifest/report evidence|comparison"):
                     summary.build_summary(args)
 
+    def test_passing_measured_report_requires_two_hour_window(self):
+        self.manifest["duration_seconds"] = summary.MIN_MEASURED_RELEASE_DURATION_SECONDS - 1
+        self.write_inputs()
+        args = type("Args", (), {
+            "abba_dir": self.abba,
+            "cache_cost_log": None,
+            "require_cache_cost": False,
+            "json_out": None,
+            "markdown_out": None,
+        })
+        with self.assertRaisesRegex(ValueError, "two hours"):
+            summary.build_summary(args)
+
     def test_passing_abba_report_requires_w10_w11_evidence(self):
         for fault in ("missing", "pressure", "lock", "attempt", "length", "range"):
             with self.subTest(fault=fault):