fix(heal): cleanup consumed MRF replay journals

Do not retain Accepted or Merged replay intents as startup anchors after they have been handed to the heal manager. Only refused or still-pending replay records keep the journal on disk until a successor snapshot can persist them.

This keeps successor snapshots limited to the pending queue, which lets successful replay remove both authoritative and legacy journal paths and restores the crash-boundary tests around successor flush.

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

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>
(cherry picked from commit d5b8f49c9d)

crates/heal/src/heal/mrf_queue.rs

@@ -516,7 +516,6 @@ async fn submit_mrf_heal_request(manager: &HealManager, intent: &MrfIntent) -> c
 
 struct MrfRuntime {
     queue: MrfQueue,
-    retained_replay_intents: Vec<MrfIntent>,
     config: MrfConsumerConfig,
     new_since_flush: usize,
     /// True while the in-memory pending set has changed since the last
@@ -536,7 +535,7 @@ impl MrfRuntime {
     fn snapshot(&self) -> (Vec<u8>, Vec<u8>) {
         let mut authoritative = Vec::new();
         let mut legacy = Vec::new();
-        for intent in self.retained_replay_intents.iter().chain(self.queue.intents()) {
+        for intent in self.queue.intents() {
             let scoped_identity =
                 !matches!(intent.kind, rustfs_common::mrf_channel::MrfKind::MetadataCorruption) && intent.scope.is_some();
             if !encode_intent(intent, &mut authoritative) {
@@ -674,11 +673,10 @@ pub async fn replay_journal_once(manager: &Arc<HealManager>) -> usize {
 struct ReplayOutcome {
     replayed: usize,
     journal_on_disk: bool,
-    retained_replay_intents: Vec<MrfIntent>,
 }
 
-fn replay_must_retain_journal(rearm_incomplete: bool, pending_depth: usize, retained_replay_depth: usize) -> bool {
-    rearm_incomplete || pending_depth > 0 || retained_replay_depth > 0
+fn replay_must_retain_journal(rearm_incomplete: bool, pending_depth: usize) -> bool {
+    rearm_incomplete || pending_depth > 0
 }
 
 /// Shared replay core: read + decode + re-arm, then drain what fits. The
@@ -700,7 +698,6 @@ async fn replay_into(
                 return ReplayOutcome {
                     replayed: 0,
                     journal_on_disk: false,
-                    retained_replay_intents: Vec::new(),
                 };
             }
         },
@@ -740,13 +737,10 @@ async fn replay_into(
 
     // Drain the replayed intents immediately; whatever the manager refuses
     // stays armed in `queue` for the consumer's retry loop.
-    let mut retained_replay_intents = Vec::new();
     if backoff_until.is_none() {
         while let Some(mut intent) = queue.pop_front() {
             match submit_mrf_heal_request(manager, &intent).await {
-                Ok(HealAdmissionResult::Accepted) | Ok(HealAdmissionResult::Merged) => {
-                    retained_replay_intents.push(intent);
-                }
+                Ok(HealAdmissionResult::Accepted) | Ok(HealAdmissionResult::Merged) => {}
                 Ok(HealAdmissionResult::Full) | Ok(HealAdmissionResult::Dropped(HealAdmissionDropReason::QueueFull)) => {
                     intent.attempts = intent.attempts.saturating_add(1);
                     if intent.attempts < MRF_MAX_ATTEMPTS {
@@ -775,7 +769,7 @@ async fn replay_into(
             }
         }
     }
-    let journal_on_disk = if replay_must_retain_journal(rearm_incomplete, queue.depth(), retained_replay_intents.len()) {
+    let journal_on_disk = if replay_must_retain_journal(rearm_incomplete, queue.depth()) {
         true
     } else {
         !delete_journals().await
@@ -783,7 +777,6 @@ async fn replay_into(
     ReplayOutcome {
         replayed,
         journal_on_disk,
-        retained_replay_intents,
     }
 }
 
@@ -793,7 +786,6 @@ async fn run_mrf_consumer(manager: Arc<HealManager>, mut receiver: mpsc::Receive
     let config = MrfConsumerConfig::default();
     let mut runtime = MrfRuntime {
         queue: MrfQueue::new(config.queue_capacity, config.journal_max_bytes),
-        retained_replay_intents: Vec::new(),
         config: config.clone(),
         new_since_flush: 0,
         dirty: false,
@@ -805,7 +797,6 @@ async fn run_mrf_consumer(manager: Arc<HealManager>, mut receiver: mpsc::Receive
     // on disk whenever any replayed intent still needs a successor snapshot.
     let replay = replay_into(&manager, &mut runtime.queue, &mut runtime.backoff_until).await;
     runtime.journal_on_disk = replay.journal_on_disk;
-    runtime.retained_replay_intents = replay.retained_replay_intents;
     // Anything still pending (e.g. the manager was full and backoff armed)
     // must be re-persisted by the next flush before replay can delete the
     // startup anchor.
@@ -823,7 +814,7 @@ async fn run_mrf_consumer(manager: Arc<HealManager>, mut receiver: mpsc::Receive
                     // provably current AND idle (a dirty or pending state
                     // gets one last persist attempt, matching the shutdown
                     // retry the unconditional flush used to provide).
-                    if runtime.dirty || runtime.queue.depth() > 0 || !runtime.retained_replay_intents.is_empty() {
+                    if runtime.dirty || runtime.queue.depth() > 0 {
                         runtime.flush().await;
                     }
                     tracing::info!(
@@ -852,7 +843,6 @@ async fn run_mrf_consumer(manager: Arc<HealManager>, mut receiver: mpsc::Receive
                 match tick_action(
                     runtime.dirty,
                     runtime.queue.depth(),
-                    runtime.retained_replay_intents.len(),
                     runtime.journal_on_disk,
                 ) {
                     TickAction::Flush => {
@@ -867,8 +857,8 @@ async fn run_mrf_consumer(manager: Arc<HealManager>, mut receiver: mpsc::Receive
                         runtime.dispatch(manager.as_ref()).await;
                     }
                     TickAction::DeleteJournal => {
-                        // Only remove a stale journal after every replayed
-                        // intent has a durable successor proof.
+                        // All replayed intents have either been accepted,
+                        // merged, or replaced by a pending successor snapshot.
                         if delete_journals().await {
                             runtime.journal_on_disk = false;
                             gauge!("rustfs_heal_mrf_journal_bytes").set(0.0);
@@ -897,13 +887,11 @@ enum TickAction {
     Idle,
 }
 
-fn tick_action(dirty: bool, depth: usize, retained_replay_depth: usize, journal_on_disk: bool) -> TickAction {
+fn tick_action(dirty: bool, depth: usize, journal_on_disk: bool) -> TickAction {
     if dirty {
         TickAction::Flush
     } else if depth > 0 {
         TickAction::Retry
-    } else if retained_replay_depth > 0 {
-        TickAction::Idle
     } else if journal_on_disk {
         TickAction::DeleteJournal
     } else {
@@ -936,73 +924,34 @@ mod tests {
 
         // Dirty dominates: a changed pending set flushes even when idle
         // otherwise.
-        assert!(matches!(tick_action(true, 0, 0, false), Flush));
-        assert!(matches!(tick_action(true, 3, 0, true), Flush));
+        assert!(matches!(tick_action(true, 0, false), Flush));
+        assert!(matches!(tick_action(true, 3, true), Flush));
 
         // Clean backlog: no rewrite, but keep draining so an expired
         // admission backoff retries on time.
-        assert!(matches!(tick_action(false, 1, 0, false), Retry));
-        assert!(matches!(tick_action(false, 2, 0, true), Retry));
-
-        // Replayed records accepted by the manager are still restart anchors
-        // until a durable successor proof can tombstone them.
-        assert!(matches!(tick_action(false, 0, 1, true), Idle));
+        assert!(matches!(tick_action(false, 1, false), Retry));
+        assert!(matches!(tick_action(false, 2, true), Retry));
 
         // Quiescent with a stale journal file on disk: remove it.
-        assert!(matches!(tick_action(false, 0, 0, true), DeleteJournal));
+        assert!(matches!(tick_action(false, 0, true), DeleteJournal));
 
         // Fully quiescent: nothing to do.
-        assert!(matches!(tick_action(false, 0, 0, false), Idle));
+        assert!(matches!(tick_action(false, 0, false), Idle));
     }
 
     #[test]
     fn replay_cleanup_retains_journal_for_unarmed_or_refused_records() {
         assert!(
-            replay_must_retain_journal(true, 0, 0),
+            replay_must_retain_journal(true, 0),
             "a rejected replay record still needs its disk anchor"
         );
         assert!(
-            replay_must_retain_journal(false, 1, 0),
+            replay_must_retain_journal(false, 1),
             "a Full admission retry must keep the startup journal until the next snapshot"
         );
         assert!(
-            replay_must_retain_journal(false, 0, 1),
-            "an accepted replay record still needs a durable successor before cleanup"
-        );
-        assert!(
-            !replay_must_retain_journal(false, 0, 0),
-            "only a fully consumed replay snapshot with no retained anchors may be deleted"
-        );
-    }
-
-    #[test]
-    fn retained_replay_anchor_remains_in_successor_snapshot() {
-        let retained = intent("accepted-replay", "object", 0);
-        let mut runtime = MrfRuntime {
-            queue: MrfQueue::new(8, 8192),
-            retained_replay_intents: vec![retained.clone()],
-            config: MrfConsumerConfig::default(),
-            new_since_flush: 0,
-            dirty: false,
-            journal_on_disk: true,
-            backoff_until: None,
-        };
-        assert_eq!(
-            runtime.queue.try_push_typed(intent("new-pending", "object", 0)),
-            MrfQueuePushResult::Enqueued
-        );
-
-        let (authoritative, legacy) = runtime.snapshot();
-        let (decoded, truncated) = decode_journal(&authoritative);
-        let (legacy_decoded, legacy_truncated) = decode_journal(&legacy);
-
-        assert_eq!(truncated, 0);
-        assert_eq!(legacy_truncated, 0);
-        assert_eq!(decoded.len(), 2);
-        assert_eq!(legacy_decoded.len(), 2);
-        assert!(
-            decoded.iter().any(|intent| intent.bucket == retained.bucket),
-            "accepted replay anchor must remain crash-replayable"
+            !replay_must_retain_journal(false, 0),
+            "only a fully consumed replay snapshot may be deleted"
         );
     }
 

docs/testing/ci-gates.md

@@ -228,6 +228,19 @@ these. The external `rustfs/auto-testing` functional workflows propagate suite
 failures. Their workflow status does not establish this registry's required
 case coverage, build provenance, or object-level oracles.
 
+For automation, `--check-scanner-heal-release "$RUN_DIR"` emits one compact
+JSON decision and exits nonzero while blocked. `verified_cases` contains only
+cases that pass the complete receipt, build provenance, nextest/JUnit and real
+oracle checks; `rejected_cases` names registered cases that do not, and
+`pending_gates` names the unimplemented release requirements. Approval requires
+every registered case to verify, `pending_gates` to be empty, and a future
+registry schema capable of representing the complete release matrix. Schema 1
+is deliberately marked `release_schema_capable: false`: it models only the
+single-version, unversioned-object restart/crash cases and cannot represent
+mixed-version, rollback, EC8+4 or performance evidence. A focused run,
+synthetic harness, compile-only result, skipped/retried test, ordinary CI
+success, or removal of pending text therefore cannot become a release approval.
+
 Run parser/receipt regressions with
 `scripts/python_bin.sh scripts/check_test_wiring.py --self-test`. Those fixtures
 validate the checker only and produce no runtime or performance evidence.

rustfs/src/error.rs

@@ -496,9 +496,9 @@ impl From<StorageError> for ApiError {
 
         let message = if matches!(&err, StorageError::QuotaExceeded { .. }) {
             err.to_string()
-        } else if matches!(&err, StorageError::MaxVersionsExceeded) {
-            ApiError::error_code_to_message(&code)
-        } else if code == S3ErrorCode::InternalError && matches!(&err, StorageError::Io(_)) {
+        } else if matches!(&err, StorageError::MaxVersionsExceeded)
+            || (code == S3ErrorCode::InternalError && matches!(&err, StorageError::Io(_)))
+        {
             ApiError::error_code_to_message(&code)
         } else if code == S3ErrorCode::InternalError {
             err.to_string()

scripts/check_test_wiring.py

@@ -1094,6 +1094,38 @@ def check_scanner_heal_evidence(root: Path, directory: Path, case_id: str) -> li
         return [f"scanner/heal evidence rejected: {error}"]
 
 
+def scanner_heal_release_status(root: Path, directory: Path) -> dict[str, object]:
+    """Return a compact release decision without weakening case validation."""
+    registry = read_json(root / ".config/scanner-heal-required-tests.json")
+    evidence_integer(registry.get("schema"), "registry schema", 1, 1)
+    cases = registry.get("cases")
+    require(isinstance(cases, dict) and cases, "invalid scanner/heal registry")
+    pending = registry.get("release_pending")
+    require(isinstance(pending, dict), "invalid scanner/heal release requirements")
+    for gate, reason in pending.items():
+        require(isinstance(gate, str) and re.fullmatch(r"[A-Z][A-Z0-9-]*", gate) is not None,
+                "invalid scanner/heal release gate")
+        require(isinstance(reason, str) and reason.strip(), f"missing release requirement for {gate}")
+
+    verified_cases = []
+    rejected_cases = []
+    for case_id in sorted(cases):
+        if check_scanner_heal_evidence(root, directory, case_id):
+            rejected_cases.append(case_id)
+        else:
+            verified_cases.append(case_id)
+
+    return {
+        "schema": 1,
+        "decision": "blocked",
+        "release_approved": False,
+        "release_schema_capable": False,
+        "verified_cases": verified_cases,
+        "rejected_cases": rejected_cases,
+        "pending_gates": sorted(pending),
+    }
+
+
 def validate(root: Path) -> list[str]:
     errors: list[str] = []
     errors.extend(check_core_fixtures(root))
@@ -1311,6 +1343,61 @@ class SelfTests(unittest.TestCase):
             self.assertTrue(any(error.startswith("pending R-D:") for error in errors))
             self.assertTrue(any(error.startswith("pending R-L:") for error in errors))
 
+            status = scanner_heal_release_status(root, run_dir)
+            self.assertEqual(status["decision"], "blocked")
+            self.assertFalse(status["release_approved"])
+            self.assertEqual(status["rejected_cases"], [])
+            self.assertEqual(len(status["pending_gates"]), 21)
+
+    def test_scanner_heal_case_only_schema_cannot_approve_release(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root, run_dir = self.scanner_heal_fixture(Path(tmp))
+            registry = read_json(root / ".config/scanner-heal-required-tests.json")
+            registry["release_pending"] = {}
+            write_json(root / ".config/scanner-heal-required-tests.json", registry)
+
+            status = scanner_heal_release_status(root, run_dir)
+            self.assertEqual(status["decision"], "blocked")
+            self.assertFalse(status["release_approved"])
+            self.assertFalse(status["release_schema_capable"])
+            self.assertEqual(status["rejected_cases"], [])
+            self.assertEqual(status["pending_gates"], [])
+
+    def test_scanner_heal_release_status_rejects_synthetic_case(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root, run_dir = self.scanner_heal_fixture(Path(tmp))
+            registry = read_json(root / ".config/scanner-heal-required-tests.json")
+            registry["release_pending"] = {}
+            write_json(root / ".config/scanner-heal-required-tests.json", registry)
+            path = run_dir / "background-target-crash.json"
+            oracle = read_json(path)
+            oracle["evidence"] = "synthetic"
+            write_json(path, oracle)
+            (run_dir / "execution.json").unlink()
+            finish_scanner_heal_receipt(run_dir, 0, root)
+
+            status = scanner_heal_release_status(root, run_dir)
+            self.assertEqual(status["decision"], "blocked")
+            self.assertFalse(status["release_approved"])
+            self.assertEqual(status["rejected_cases"], ["background-target-crash"])
+            self.assertEqual(status["pending_gates"], [])
+
+    def test_scanner_heal_release_status_rejects_focused_case_run(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root, run_dir = self.scanner_heal_fixture(Path(tmp))
+            registry = read_json(root / ".config/scanner-heal-required-tests.json")
+            registry["release_pending"] = {}
+            write_json(root / ".config/scanner-heal-required-tests.json", registry)
+            (run_dir / "background-target-crash.json").unlink()
+            (run_dir / "execution.json").unlink()
+            finish_scanner_heal_receipt(run_dir, 0, root)
+
+            status = scanner_heal_release_status(root, run_dir)
+            self.assertEqual(status["decision"], "blocked")
+            self.assertFalse(status["release_approved"])
+            self.assertEqual(status["verified_cases"], ["background-target-restart"])
+            self.assertEqual(status["rejected_cases"], ["background-target-crash"])
+
     def test_scanner_heal_finish_collects_oracles_from_registry(self) -> None:
         with tempfile.TemporaryDirectory() as tmp:
             root, run_dir = self.scanner_heal_fixture(Path(tmp))
@@ -2158,7 +2245,8 @@ def main() -> int:
     if sys.argv[1:] == ["--self-test"]:
         suite = unittest.defaultTestLoader.loadTestsFromTestCase(SelfTests)
         return 0 if unittest.TextTestRunner(verbosity=2).run(suite).wasSuccessful() else 1
-    if sys.argv[1:2] in (["--begin-scanner-heal"], ["--finish-scanner-heal"], ["--check-scanner-heal"]):
+    if sys.argv[1:2] in (["--begin-scanner-heal"], ["--finish-scanner-heal"], ["--check-scanner-heal"],
+                         ["--check-scanner-heal-release"]):
         try:
             if len(sys.argv) == 5 and sys.argv[1] == "--begin-scanner-heal":
                 begin_scanner_heal_receipt(ROOT, Path(sys.argv[2]), Path(sys.argv[3]), Path(sys.argv[4]))
@@ -2173,7 +2261,16 @@ def main() -> int:
                 if not errors:
                     print(f"Case evidence verified: {sys.argv[3]}; this does not approve release")
                 return 1 if errors else 0
-            raise ValueError("expected --begin-scanner-heal DIR BINARY TEST_BINARY, --finish-scanner-heal DIR EXIT, or --check-scanner-heal DIR CASE|release")
+            if len(sys.argv) == 3 and sys.argv[1] == "--check-scanner-heal-release":
+                try:
+                    status = scanner_heal_release_status(ROOT, Path(sys.argv[2]))
+                except (OSError, KeyError, TypeError, ValueError, ET.ParseError) as error:
+                    print(json.dumps({"schema": 1, "decision": "invalid", "release_approved": False,
+                                      "error": str(error)}, sort_keys=True, separators=(",", ":")))
+                    return 2
+                print(json.dumps(status, sort_keys=True, separators=(",", ":")))
+                return 0 if status["release_approved"] else 1
+            raise ValueError("expected --begin-scanner-heal DIR BINARY TEST_BINARY, --finish-scanner-heal DIR EXIT, --check-scanner-heal DIR CASE|release, or --check-scanner-heal-release DIR")
         except (OSError, KeyError, TypeError, ValueError, subprocess.SubprocessError) as error:
             print(f"ERROR: {error}", file=sys.stderr)
             return 1

scripts/run_scanner_heal_evidence_case.sh

@@ -90,15 +90,22 @@ PY
 
 release_gate_must_remain_blocked() {
     local run_dir="$1"
-    local output="$run_dir/release-check.txt"
-    if "$PYTHON_BIN" "$ROOT/scripts/check_test_wiring.py" --check-scanner-heal "$run_dir" release >"$output" 2>&1; then
+    local output="$run_dir/release-status.json"
+    if "$PYTHON_BIN" "$ROOT/scripts/check_test_wiring.py" --check-scanner-heal-release "$run_dir" >"$output"; then
         echo "release gate unexpectedly approved a single Scanner/Heal evidence run" >&2
         return 1
     fi
-    if ! grep -Eq 'required test not selected:|pending [A-Z0-9-]+:' "$output"; then
-        echo "release gate did not explain why the Scanner/Heal release remains blocked" >&2
-        return 1
-    fi
+    "$PYTHON_BIN" - "$output" <<'PY'
+import json
+import pathlib
+import sys
+
+status = json.loads(pathlib.Path(sys.argv[1]).read_text())
+if status.get("decision") != "blocked" or status.get("release_approved") is not False:
+    raise SystemExit("release status did not record a blocked decision")
+if status.get("release_schema_capable") is not False:
+    raise SystemExit("case-only evidence schema unexpectedly became release-capable")
+PY
 }
 
 run_self_test() {
@@ -242,5 +249,5 @@ fi
 "$PYTHON_BIN" "$ROOT/scripts/check_test_wiring.py" --check-scanner-heal "$RUN_DIR" "$CASE_ID"
 release_gate_must_remain_blocked "$RUN_DIR"
 echo "Scanner/Heal evidence case verified: $CASE_ID"
-echo "Release gate remains blocked; details: $RUN_DIR/release-check.txt"
+echo "Release gate remains blocked; status: $RUN_DIR/release-status.json"
 echo "Evidence directory: $RUN_DIR"