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/operations/scanner-benchmark-runbook.md

@@ -125,6 +125,11 @@ and `metrics`. All metrics must be finite nonnegative numbers: `p99_ms`,
 `throughput_ops`, `rss_bytes`, `cpu_seconds`, `iops`, `rpc_count`,
 `cache_clone_bytes`, `encode_bytes`, `save_bytes`, `oldest_age_seconds`,
 `walk_objects`, `cold_walk_objects`, `healed_objects`, `errors`, and `requests`.
+The adapter also reports the measurement-window delta of
+`rustfs_heal_mainline_throttle_total{source="admin",result="delayed"}` as
+`heal_mainline_throttle_delayed`; a cumulative process-lifetime value is not a
+valid input.
+The clone, encode, and save byte fields are also deltas from the same window.
 Requests, throughput, and p99 must be positive; errors must be zero. Repair
 counts must match the manifest when background work is on. Keep underlying
 request samples, counter reset checks, profiler captures, and per-node telemetry
@@ -132,6 +137,19 @@ in the cell artifact directory; aggregate values alone do not establish their
 measurement provenance. Missing production instrumentation is a pending gate,
 not permission to report a fabricated zero.
 
+Each comparison records a `w22` section with clone, encode, and save bytes per
+walked object, clone/encode and save/encode byte ratios, and candidate changes.
+These are traffic amplification indicators, not allocation attribution or an
+fsync profile. The `running-heal` build comparison also records a `w10`
+section. `status=observed` requires sampled high foreground pressure, at least
+one admin pacing delay in the same window, and an improvement in either
+foreground p99 or throughput. `no_measured_benefit` means pacing ran but neither
+foreground metric improved; `pending` means the run did not prove that pacing
+engaged; `inconclusive` means ABBA repeatability failed. Baseline and candidate
+delay counts are both retained so an operator can reject unrelated or
+process-lifetime counter contamination. Correct repair oracles and the existing
+regression limits still apply in every case.
+
 For P2, `measure.convergence` contains booleans `writes_stopped`,
 `last_mutation_observed`, `first_complete_publication`; numeric
 `last_mutation_time`, `last_mutation_observed_time`, `writes_stopped_time`, `window_start`, `window_end`,

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/scanner_abba.py

@@ -23,6 +23,7 @@ METRICS = (
     "cache_clone_bytes", "encode_bytes", "save_bytes", "oldest_age_seconds",
     "walk_objects", "cold_walk_objects", "healed_objects", "errors", "requests",
     "foreground_pressure_samples", "foreground_pressure_high_samples",
+    "heal_mainline_throttle_delayed",
     "heal_lock_wait_p99_ms", "heal_attempts", "heal_attempt_failures",
     "heal_retry_attempts",
 )
@@ -59,6 +60,12 @@ def relative_change(current, baseline, name):
     return ratio(current, baseline, name) - Decimal("1")
 
 
+def relative_change_or_none(current, baseline, name):
+    if decimal_number(baseline, f"{name} baseline") == 0:
+        return None
+    return relative_change(current, baseline, name)
+
+
 def repeatability_change(first, second, name):
     first = decimal_number(first, name)
     second = decimal_number(second, name)
@@ -283,6 +290,62 @@ def pressure_high_ratio(metrics):
                  metrics["foreground_pressure_samples"], "foreground pressure high samples")
 
 
+def scanner_cache_cost(metrics):
+    walked = decimal_number(metrics["walk_objects"], "walk_objects")
+    encoded = decimal_number(metrics["encode_bytes"], "encode_bytes")
+    return {
+        "clone_bytes_per_walk_object": None if walked == 0 else float(ratio(metrics["cache_clone_bytes"], walked, "clone bytes per walk object")),
+        "encode_bytes_per_walk_object": None if walked == 0 else float(ratio(encoded, walked, "encode bytes per walk object")),
+        "save_bytes_per_walk_object": None if walked == 0 else float(ratio(metrics["save_bytes"], walked, "save bytes per walk object")),
+        "clone_to_encode_byte_ratio": None if encoded == 0 else float(ratio(metrics["cache_clone_bytes"], encoded, "clone to encode bytes")),
+        "save_to_encode_byte_amplification": None if encoded == 0 else float(ratio(metrics["save_bytes"], encoded, "save to encode bytes")),
+    }
+
+
+def scanner_cache_cost_change(candidate, baseline):
+    changes = {}
+    for key in ("cache_clone_bytes", "encode_bytes", "save_bytes"):
+        change = relative_change_or_none(candidate[key], baseline[key], key)
+        changes[f"{key}_change"] = None if change is None else float(change)
+    return changes
+
+
+def running_heal_pacing(group, baseline, candidate, p99, throughput, noisy):
+    if group[0]["scenario"] != "running-heal" or group[0]["comparison"] != "build":
+        return None
+    baseline_seconds = sum(decimal_number(cell["result"]["elapsed_seconds"], "elapsed_seconds") for cell in (group[0], group[3])) / 2
+    candidate_seconds = sum(decimal_number(cell["result"]["elapsed_seconds"], "elapsed_seconds") for cell in (group[1], group[2])) / 2
+    baseline_rate = ratio(baseline["heal_attempts"], baseline_seconds, "baseline heal attempt rate")
+    candidate_rate = ratio(candidate["heal_attempts"], candidate_seconds, "candidate heal attempt rate")
+    rate_change = relative_change_or_none(candidate_rate, baseline_rate, "heal attempt rate")
+    candidate_high_ratio = pressure_high_ratio(candidate)
+    baseline_delayed = decimal_number(baseline["heal_mainline_throttle_delayed"], "baseline pacing delays")
+    delayed = decimal_number(candidate["heal_mainline_throttle_delayed"], "candidate pacing delays")
+    pacing_observed = candidate_high_ratio > 0 and delayed > 0
+    foreground_improved = p99 < 0 or throughput > 0
+    status = (
+        "inconclusive"
+        if noisy
+        else "observed"
+        if pacing_observed and foreground_improved
+        else "no_measured_benefit"
+        if pacing_observed
+        else "pending"
+    )
+    return {
+        "status": status,
+        "pacing_observed": pacing_observed,
+        "candidate_pressure_high_ratio": float(candidate_high_ratio),
+        "baseline_delay_events": float(baseline_delayed),
+        "candidate_delay_events": float(delayed),
+        "baseline_heal_attempts_per_second": float(baseline_rate),
+        "candidate_heal_attempts_per_second": float(candidate_rate),
+        "heal_attempt_rate_change": None if rate_change is None else float(rate_change),
+        "foreground_p99_change": float(p99),
+        "foreground_throughput_change": float(throughput),
+    }
+
+
 def convergence(result):
     window = result.get("convergence")
     if not window or window.get("writes_stopped") is not True or window.get("last_mutation_observed") is not True or window.get("first_complete_publication") is not True:
@@ -342,6 +405,7 @@ def evaluate(cells):
         candidate_attempt_costs = [
             value for cell, value in zip(group, attempt_costs) if cell["leg"].startswith("B") and value is not None
         ]
+        w10 = running_heal_pacing(group, a, b, p99, throughput, noise)
         inconclusive |= noise or p2_pending
         if not noise and not passed:
             failed = True
@@ -352,6 +416,12 @@ def evaluate(cells):
                             "thresholds": {key: float(value) for key, value in thresholds.items()},
                             "p1": p1, "p2_max_work_multiple": float(P2_WORK_MULTIPLE_LIMIT),
                             "p2_post_stop_work_multiples": p2_report,
+                            "w22": {
+                                "baseline": scanner_cache_cost(a),
+                                "candidate": scanner_cache_cost(b),
+                                "candidate_vs_baseline": scanner_cache_cost_change(b, a),
+                            },
+                            "w10": w10,
                             "w10_w11": {
                                 "foreground_pressure_high_sample_ratios": [
                                     float(pressure_high_ratio(cell["result"]["metrics"])) for cell in group

scripts/test_scanner_abba.py

@@ -94,6 +94,8 @@ def fake_adapter():
                 result["metrics"].update(walk_objects=100, cold_walk_objects=0)
         elif fault == "missing-metric":
             del result["metrics"]["save_bytes"]
+        elif fault == "missing-pacing-metric":
+            del result["metrics"]["heal_mainline_throttle_delayed"]
         elif fault == "incomplete-repair":
             result["metrics"]["healed_objects"] = 0
         elif fault == "zero-pressure-samples":
@@ -102,6 +104,12 @@ def fake_adapter():
             result["metrics"]["foreground_pressure_high_samples"] = result["metrics"]["foreground_pressure_samples"] + 1
         elif fault == "attempt-accounting":
             result["metrics"]["heal_attempt_failures"] = result["metrics"]["heal_attempts"] + 1
+        elif fault == "pacing-benefit" and request["scenario"] == "running-heal" \
+                and request["comparison"] == "build" and request["leg"].startswith("B"):
+            result["metrics"].update(p99_ms=9, heal_mainline_throttle_delayed=5)
+        elif fault == "pacing-pending" and request["scenario"] == "running-heal" \
+                and request["comparison"] == "build" and request["leg"].startswith("B"):
+            result["metrics"]["heal_mainline_throttle_delayed"] = 0
     harness.write_json(Path(output_path), result)
     return 0
 
@@ -278,6 +286,31 @@ class ScannerAbbaTest(unittest.TestCase):
                     self.assertEqual({r["leg"] for r in legs}, set(harness.LEGS))
         self.assertTrue(all(c["p2_max_work_multiple"] == 1.2 for c in report["comparisons"]))
         for comparison in report["comparisons"]:
+            w22 = comparison["w22"]
+            self.assertEqual(w22["baseline"]["save_to_encode_byte_amplification"], 1.0)
+            self.assertEqual(w22["candidate"]["clone_to_encode_byte_ratio"], 1.0)
+            self.assertEqual(
+                w22["candidate_vs_baseline"],
+                {"cache_clone_bytes_change": 0.0, "encode_bytes_change": 0.0, "save_bytes_change": 0.0},
+            )
+            if comparison["scenario"] == "running-heal" and comparison["comparison"] == "build":
+                self.assertEqual(
+                    comparison["w10"],
+                    {
+                        "status": "no_measured_benefit",
+                        "pacing_observed": True,
+                        "candidate_pressure_high_ratio": 1.0,
+                        "baseline_delay_events": 10.0,
+                        "candidate_delay_events": 10.0,
+                        "baseline_heal_attempts_per_second": 10.0,
+                        "candidate_heal_attempts_per_second": 10.0,
+                        "heal_attempt_rate_change": 0.0,
+                        "foreground_p99_change": 0.0,
+                        "foreground_throughput_change": 0.0,
+                    },
+                )
+            else:
+                self.assertIsNone(comparison["w10"])
             w10_w11 = comparison["w10_w11"]
             self.assertEqual(w10_w11["foreground_pressure_high_sample_ratios"], [1.0, 1.0, 1.0, 1.0])
             self.assertEqual(w10_w11["heal_lock_wait_p99_ms"], [10, 10, 10, 10])
@@ -288,7 +321,8 @@ class ScannerAbbaTest(unittest.TestCase):
     def test_fail_closed_adapter_and_data_errors(self):
         for fault in ("measure-exit", "oracle-exit", "missing-oracle", "oracle-mismatch", "zero-samples",
                       "zero-requests", "request-errors", "load-drift", "missing-metric", "incomplete-repair",
-                      "zero-pressure-samples", "pressure-sample-order", "attempt-accounting"):
+                      "zero-pressure-samples", "pressure-sample-order", "attempt-accounting",
+                      "missing-pacing-metric"):
             with self.subTest(fault=fault), tempfile.TemporaryDirectory() as directory:
                 self.root = Path(directory)
                 with self.assertRaises((ValueError, OSError, subprocess.SubprocessError)):
@@ -302,6 +336,36 @@ class ScannerAbbaTest(unittest.TestCase):
             self.assertEqual(self.run_harness("noise"), 3)
         self.assertEqual(harness.read_json(self.root / "out/report.json")["status"], "inconclusive")
 
+    def test_noisy_running_heal_does_not_claim_pacing_benefit(self):
+        with patch.object(harness, "SCENARIOS", ("running-heal",)):
+            self.assertEqual(self.run_harness("noise"), 3)
+        comparisons = harness.read_json(self.root / "out/report.json")["comparisons"]
+        build = next(comparison for comparison in comparisons if comparison["comparison"] == "build")
+        self.assertEqual(build["w10"]["status"], "inconclusive")
+
+    def test_idle_cache_window_reports_unavailable_ratios(self):
+        metrics = dict.fromkeys(harness.METRICS, 0)
+        self.assertEqual(
+            harness.scanner_cache_cost(metrics),
+            {
+                "clone_bytes_per_walk_object": None,
+                "encode_bytes_per_walk_object": None,
+                "save_bytes_per_walk_object": None,
+                "clone_to_encode_byte_ratio": None,
+                "save_to_encode_byte_amplification": None,
+            },
+        )
+
+    def test_running_heal_pacing_status_requires_engagement_and_benefit(self):
+        for fault, expected in (("pacing-benefit", "observed"), ("pacing-pending", "pending")):
+            with self.subTest(fault=fault), tempfile.TemporaryDirectory() as directory:
+                self.root = Path(directory)
+                with patch.object(harness, "SCENARIOS", ("running-heal",)):
+                    self.assertEqual(self.run_harness(fault), 0)
+                comparisons = harness.read_json(self.root / "out/report.json")["comparisons"]
+                build = next(comparison for comparison in comparisons if comparison["comparison"] == "build")
+                self.assertEqual(build["w10"]["status"], expected)
+
     def test_missing_first_publication_is_inconclusive(self):
         with patch.object(harness, "SCENARIOS", ("cold-hot",)):
             self.assertEqual(self.run_harness("no-publication"), 3)