test(scanner): require exact ABBA release evidence sets

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

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

crates/e2e_test/src/distributed/heal_test.rs

@@ -46,18 +46,6 @@ 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();
@@ -142,12 +130,23 @@ fn assert_ec84_geometry(census: &VersionShardCensus, key: &str) -> TestResult {
     Ok(())
 }
 
-async fn write_scanner_heal_evidence(context: ScannerHealEvidenceContext, payload: ScannerHealEvidencePayload<'_>) -> TestResult {
-    let verifier = payload.dist.client(0)?;
+async fn write_scanner_heal_evidence(
+    context: ScannerHealEvidenceContext,
+    dist: &DistCluster,
+    bucket: &str,
+    expected: &[ExpectedShard],
+    outage_key: &str,
+    outage_body: &[u8],
+    replaced_drive: &Path,
+    pid_before: u32,
+    pid_after: u32,
+    node_listings: Vec<Vec<String>>,
+) -> TestResult {
+    let verifier = dist.client(0)?;
     let mut objects = Vec::new();
-    for item in 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)?;
+    for item in expected {
+        let actual = get_object_bytes(&verifier, bucket, &item.key).await?;
+        let physical = census_object_version_on_disk(replaced_drive, bucket, &item.key, None)?;
         objects.push(serde_json::json!({
             "key": item.key,
             "version_id": null,
@@ -159,14 +158,14 @@ async fn write_scanner_heal_evidence(context: ScannerHealEvidenceContext, payloa
             "physical": physical,
         }));
     }
-    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)?;
+    let actual = get_object_bytes(&verifier, bucket, outage_key).await?;
+    let physical = census_object_version_on_disk(replaced_drive, bucket, outage_key, None)?;
     objects.push(serde_json::json!({
-        "key": payload.outage_key,
+        "key": outage_key,
         "version_id": null,
-        "expected_bytes": payload.outage_body.len(),
+        "expected_bytes": outage_body.len(),
         "actual_bytes": actual.len(),
-        "expected_sha256": sha256_hex(payload.outage_body),
+        "expected_sha256": sha256_hex(outage_body),
         "actual_sha256": sha256_hex(&actual),
         "expected_physical": null,
         "physical": physical,
@@ -182,11 +181,11 @@ async fn write_scanner_heal_evidence(context: ScannerHealEvidenceContext, payloa
         "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": payload.pid_before,
-        "pid_after": payload.pid_after,
+        "pid_before": pid_before,
+        "pid_after": pid_after,
         "unclean_shutdown_marker": false,
         "objects": objects,
-        "node_listings": payload.node_listings,
+        "node_listings": node_listings,
     });
     let data = serde_json::to_vec(&evidence)?;
     if data.len() > 1024 * 1024 {
@@ -348,17 +347,15 @@ 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,
-            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,
-            },
+            &dist,
+            &bucket,
+            &expected,
+            outage_key,
+            &outage_body,
+            &replaced_drive,
+            target_pid_before,
+            target_pid_after,
+            node_listings,
         )
         .await?;
     }

crates/heal/src/heal/mrf_queue.rs

@@ -521,8 +521,6 @@ 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
@@ -539,12 +537,6 @@ 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>,
 }
@@ -568,34 +560,6 @@ 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);
@@ -606,10 +570,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 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;
+        // 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;
         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
@@ -617,7 +581,7 @@ impl MrfRuntime {
         if persisted {
             self.dirty = false;
         }
-        self.journal_on_disk |= committed_on_disk || authoritative_persisted || legacy_persisted;
+        self.journal_on_disk |= authoritative_persisted || legacy_persisted;
     }
 
     /// Drain pending intents into the heal manager until it is full, the
@@ -675,45 +639,6 @@ 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;
@@ -783,8 +708,6 @@ 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(
@@ -796,7 +719,7 @@ fn replay_must_retain_journal(
     rearm_incomplete || pending_depth > 0 || accepted_without_durable_anchor || durable_replay_anchors > 0
 }
 
-#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+#[derive(Clone, Copy)]
 enum ReplayCleanup {
     Legacy,
     Committed { owner: Uuid, sequence: u64 },
@@ -870,8 +793,6 @@ 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) => {
@@ -885,16 +806,10 @@ 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();
@@ -998,8 +913,6 @@ async fn replay_into(
         journal_on_disk,
         retain_journal_for_replay,
         durable_replay_anchors,
-        cleanup: journal_on_disk.then_some(cleanup),
-        next_checkpoint_sequence,
     }
 }
 
@@ -1010,15 +923,11 @@ 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,
     };
 
@@ -1028,8 +937,6 @@ async fn run_mrf_consumer(manager: Arc<HealManager>, mut receiver: mpsc::Receive
     runtime.journal_on_disk = replay.journal_on_disk;
     runtime.retain_replay_journal = replay.retain_journal_for_replay;
     runtime.durable_replay_anchors = replay.durable_replay_anchors;
-    runtime.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.
@@ -1094,7 +1001,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 runtime.delete_idle_recovery_anchors().await {
+                        if delete_journals().await {
                             runtime.journal_on_disk = false;
                             gauge!("rustfs_heal_mrf_journal_bytes").set(0.0);
                         }
@@ -1138,7 +1045,6 @@ 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 {
@@ -1154,12 +1060,6 @@ 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::*;
@@ -1217,23 +1117,14 @@ 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 {
@@ -1251,108 +1142,14 @@ 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,9 +22,8 @@
 //! 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.
-//! 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.
+//! It is wired into the replay reader before writer activation, but the writer
+//! remains gated on ownership-aware handoff.
 //! One surviving committed replica supports process restart recovery only;
 //! this reader does not establish a replication quorum or a power-loss policy.
 
@@ -357,8 +356,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 or tombstone older
-/// anchors by itself; the consumer decides cleanup after replay handoff.
+/// reaches the same disk. It does not delete legacy journals, tombstone older
+/// anchors, or activate the live consumer.
 pub async fn publish_committed_snapshot(
     disks: &[EcstoreDiskStore],
     owner: Uuid,
@@ -372,10 +371,7 @@ pub async fn publish_committed_snapshot(
     if owner.is_nil() || sequence == 0 || sequence == u64::MAX {
         return Err(SnapshotError::Corrupt);
     }
-    if payload.len() > limit {
-        return Err(SnapshotError::TooLarge);
-    }
-    if decode_journal(payload).1 != 0 {
+    if payload.len() > limit || decode_journal(payload).1 != 0 {
         return Err(SnapshotError::Corrupt);
     }
     let current = read_committed(disks, limit).await?;
@@ -559,13 +555,13 @@ pub async fn inspect_local_committed_snapshot(max_bytes: usize) -> Result<Option
     read_committed(&super::journal_disks().await, max_bytes).await
 }
 
-/// Remove committed checkpoints from `owner` whose sequence is no newer than
+/// Remove committed manifests from `owner` whose sequence is no newer than
 /// `committed_through`.
 ///
-/// 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.
+/// Payload files are intentionally left as orphans after their manifest is
+/// removed. Readers cannot discover a payload without its matching manifest,
+/// and deleting manifests first prevents an older retained slot from becoming
+/// visible again after the newest replay has been fully discharged.
 pub async fn delete_committed_snapshots_through(
     owner: Uuid,
     committed_through: u64,
@@ -584,10 +580,11 @@ 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 (manifest_path, payload_path) in MANIFEST_PATHS.into_iter().zip(PAYLOAD_PATHS) {
-            let manifest_bytes = match read_bounded(disk, manifest_path, MANIFEST_LEN).await {
+        for path in MANIFEST_PATHS {
+            let existing = match read_bounded(disk, path, MANIFEST_LEN).await {
                 Ok(Some(existing)) => existing,
                 Ok(None) => continue,
                 Err(error) => {
@@ -597,7 +594,7 @@ async fn delete_committed_snapshots_through_on(
                     continue;
                 }
             };
-            let manifest = match Manifest::decode(&manifest_bytes, max_bytes) {
+            let manifest = match Manifest::decode(&existing, max_bytes) {
                 Ok(manifest) => manifest,
                 Err(error) => {
                     if first_error.is_none() {
@@ -609,58 +606,16 @@ 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,
-                manifest_path,
-                Some(EcstoreDiskBytes::copy_from_slice(&manifest_bytes)),
+                path,
+                Some(EcstoreDiskBytes::from(existing)),
                 None,
             )
             .await
             {
-                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::Updated) => any_changed = true,
                 Ok(EcstoreConditionalFileUpdate::Missing | EcstoreConditionalFileUpdate::Mismatch) => {}
                 Err(error) => {
                     if first_error.is_none() {
@@ -670,7 +625,13 @@ async fn delete_committed_snapshots_through_on(
             }
         }
     }
-    if let Some(error) = first_error { Err(error) } else { Ok(true) }
+    if any_changed {
+        Ok(true)
+    } else if let Some(error) = first_error {
+        Err(error)
+    } else {
+        Ok(true)
+    }
 }
 
 async fn read_recovery_snapshot(disks: &[EcstoreDiskStore], limit: usize) -> Result<Option<RecoverySnapshot>, SnapshotError> {
@@ -1140,45 +1101,6 @@ 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");
@@ -1433,7 +1355,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn committed_cleanup_removes_complete_checkpoint_at_or_below_sequence() {
+    async fn committed_cleanup_removes_only_manifests_at_or_below_sequence() {
         let root = TempDir::new().expect("test directory");
         let disk = disk(&root, "disk").await;
         let owner = Uuid::new_v4();
@@ -1455,19 +1377,12 @@ 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[1])
+            EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, PAYLOAD_PATHS[0])
                 .await
-                .expect("newer payload retained")
+                .expect("old payload orphan may remain")
                 .as_ref(),
-            newer
+            older
         );
         let recovered = read_committed(std::slice::from_ref(&disk), 4096)
             .await
@@ -1477,40 +1392,6 @@ 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");
@@ -1536,13 +1417,6 @@ 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,8 +51,6 @@ 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;
 
@@ -237,44 +235,6 @@ 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,
@@ -669,14 +629,13 @@ 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 committed pending successor before the delete phase"
+            "child process must publish the pending successor snapshot before the delete phase"
         );
     });
     std::process::exit(78);
@@ -717,14 +676,13 @@ 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 committed pending successor before it can be killed"
+            "child process must publish the pending successor snapshot 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 release leg lasts at least 7200
-seconds; the minimum matrix contains 120 legs (240 hours before setup/oracles).
+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).
 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, 7200..86400 seconds for measured release 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, 900..86400 seconds for measured 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,13 +288,9 @@ 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. 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.
+stitched together from unrelated runs. Missing, synthetic, stale, tampered,
+undersized, or topology-mismatched evidence returns a compact blocked or invalid
+JSON result and a nonzero exit.
 
 This command validates the evidence package; it does not create evidence. A
 handwritten JSON file, a synthetic harness pass, a single focused case, or a

scripts/check_test_wiring.py

@@ -23,16 +23,7 @@ try:
 except ModuleNotFoundError:
     import tomli as tomllib
 
-from scanner_abba import (
-    MAX_JSON_BYTES,
-    RELEASE_PROFILE_ARTIFACTS,
-    digest,
-    number,
-    read_json,
-    require,
-    sha,
-    write_json,
-)
+from scanner_abba import MAX_JSON_BYTES, digest, number, read_json, require, sha, write_json
 
 
 ROOT = Path(__file__).resolve().parents[1]
@@ -89,12 +80,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_MRF_DURABLE_REPLAY_FIELDS = {
-    ("G07", "mrf_responsibility_oracle"),
-    ("G07", "commit_boundary_crash_matrix"),
-    ("P4", "mrf_replay_cost_measurement"),
-    ("P4", "retained_responsibility_evidence"),
-}
+SCANNER_HEAL_RELEASE_PROFILE_ARTIFACTS = (
+    "allocation-profile",
+    "flamegraph",
+    "rss-samples",
+    "save-frequency",
+)
 SCHEDULED_ALERT_WORKFLOWS = tuple(
     item["workflow"]
     for item in json.loads((ROOT / ".github/scheduled-validations.json").read_text())
@@ -1333,7 +1324,6 @@ 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
 
 
@@ -1381,23 +1371,7 @@ 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")
@@ -1413,28 +1387,14 @@ 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, 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)
+        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)}")
     return window_id
 
 
@@ -1794,34 +1754,15 @@ 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
-                    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
+                    evidence["profile_artifacts"] = list(SCANNER_HEAL_RELEASE_PROFILE_ARTIFACTS)
                 fields[field] = evidence
             gates[gate] = {
                 "status": "pass",
@@ -1845,7 +1786,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", "empty-artifact"):
+        for fault in ("synthetic", "missing-field", "hash"):
             with self.subTest(fault=fault), tempfile.TemporaryDirectory() as tmp:
                 root, bundle = self.scanner_heal_release_bundle_fixture(Path(tmp))
                 data = read_json(bundle)
@@ -1853,12 +1794,9 @@ class SelfTests(unittest.TestCase):
                     data["evidence"] = "synthetic"
                 elif fault == "missing-field":
                     del data["gates"]["G09"]["evidence_fields"]["rollback_payload_evidence"]
-                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")
+                    artifact.write_text(artifact.read_text(encoding="utf-8") + "\n", encoding="utf-8")
                 write_json(bundle, data)
 
                 with mock.patch("subprocess.check_output", return_value="b" * 40):
@@ -1878,35 +1816,10 @@ 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["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",
-            ),
+            ("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"),
             ("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,40 +74,6 @@ 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"
@@ -164,11 +130,9 @@ 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
@@ -221,7 +185,13 @@ TEST_FILTER="$(test_filter_for "$CASE_ID")"
 if [[ -z "$PROFILE" ]]; then
     PROFILE="$(case_field "$CASE_ID" lane)"
 fi
-apply_runtime_profile "$CASE_ID"
+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
 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
@@ -231,7 +201,6 @@ 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,7 +35,6 @@ RELEASE_PROFILE_ARTIFACTS = (
     "rss-samples",
     "save-frequency",
 )
-MIN_MEASURED_RELEASE_DURATION_SECONDS = 7200
 RELEASE_FAULT_MODES = (
     "process-restart",
     "process-crash-restart",
@@ -131,7 +130,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 = MIN_MEASURED_RELEASE_DURATION_SECONDS if manifest["evidence"] == "measured" else 1
+    minimum = 900 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)
@@ -174,16 +173,15 @@ 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}")
+def release_evidence_exact_set(value, name, expected):
+    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
+    require(observed == set(expected), f"invalid release_evidence.{name}")
 
 
 def validate_release_evidence_manifest(manifest):
@@ -223,17 +221,11 @@ def validate_release_evidence_manifest(manifest):
 
     crash = evidence.get("crash_restart")
     require(isinstance(crash, dict), "missing release_evidence.crash_restart")
-    release_evidence_exact_strings(crash.get("fault_modes"), RELEASE_FAULT_MODES, "crash_restart.fault_modes")
+    release_evidence_exact_set(crash.get("fault_modes"), "crash_restart.fault_modes", RELEASE_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)
@@ -242,15 +234,14 @@ def validate_release_evidence_manifest(manifest):
                 for revision in revisions),
         "invalid release_evidence.mixed_version.participating_revisions",
     )
-    for revision in (baseline_revision, candidate_revision):
+    for revision in (manifest["baseline"]["revision"], manifest["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")
-    release_evidence_exact_strings(profile.get("required_artifacts"), RELEASE_PROFILE_ARTIFACTS,
-                                   "profile.required_artifacts")
+    release_evidence_exact_set(profile.get("required_artifacts"), "profile.required_artifacts", RELEASE_PROFILE_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,12 +12,7 @@ from pathlib import Path
 import sys
 from typing import Any
 
-from scanner_abba import (
-    LEGS,
-    MIN_MEASURED_RELEASE_DURATION_SECONDS,
-    SCENARIOS,
-    validate_release_evidence_manifest,
-)
+from scanner_abba import LEGS, SCENARIOS, validate_release_evidence_manifest
 
 MAX_JSON_BYTES = 1024 * 1024
 CACHE_COST_PREFIX = "CACHE_COST "
@@ -129,9 +124,6 @@ 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,15 +167,8 @@ class ScannerAbbaTest(unittest.TestCase):
 
     def measured_manifest(self):
         manifest = copy.deepcopy(self.manifest)
-        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.update(evidence="measured", duration_seconds=900)
+        manifest["candidate"]["revision"] = "b" * 40
         manifest["release_evidence"] = {
             "topology": {
                 "nodes": 3,
@@ -534,7 +527,7 @@ class ScannerAbbaTest(unittest.TestCase):
         self.manifest["evidence"] = "measured"
         with self.assertRaisesRegex(ValueError, "duration_seconds"):
             harness.validate_manifest(self.manifest)
-        self.manifest["duration_seconds"] = harness.MIN_MEASURED_RELEASE_DURATION_SECONDS
+        self.manifest["duration_seconds"] = 900
         self.manifest["rounds"] = 2
         with self.assertRaisesRegex(ValueError, "rounds"):
             harness.validate_manifest(self.manifest)
@@ -556,7 +549,7 @@ class ScannerAbbaTest(unittest.TestCase):
                 fault_modes=["process-restart"],
             ),
             "unknown crash": lambda manifest: manifest["release_evidence"]["crash_restart"].update(
-                fault_modes=["process-restart", "process-crash-restart", "kernel-panic"],
+                fault_modes=["process-restart", "process-crash-restart", "power-cycle"],
             ),
             "duplicate crash": lambda manifest: manifest["release_evidence"]["crash_restart"].update(
                 fault_modes=["process-restart", "process-restart", "process-crash-restart"],
@@ -568,23 +561,14 @@ 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"],
+                required_artifacts=["allocation-profile", "flamegraph", "rss-samples", "save-frequency", "heap-dump"],
             ),
             "duplicate profile": lambda manifest: manifest["release_evidence"]["profile"].update(
-                required_artifacts=[
-                    "allocation-profile", "flamegraph", "rss-samples", "save-frequency", "flamegraph",
-                ],
+                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,7 +37,6 @@ 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,
@@ -226,19 +225,6 @@ 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):