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)

Cargo.lock

@@ -9906,6 +9906,7 @@ dependencies = [
  "regex",
  "rmp",
  "rmp-serde",
+ "rustfs-config",
  "rustfs-utils",
  "s3s",
  "serde",

crates/config/src/constants/api.rs

@@ -100,5 +100,10 @@ pub const DEFAULT_API_MAX_CONNECTIONS: usize = 0;
 /// Example: RUSTFS_API_OBJECT_MAX_VERSIONS=50000
 pub const ENV_API_OBJECT_MAX_VERSIONS: &str = "RUSTFS_API_OBJECT_MAX_VERSIONS";
 
-/// Default for `RUSTFS_API_OBJECT_MAX_VERSIONS`.
-pub const DEFAULT_API_OBJECT_MAX_VERSIONS: u64 = 9_223_372_036_854_775_807;
+/// Default and maximum accepted value for `RUSTFS_API_OBJECT_MAX_VERSIONS`.
+#[cfg(target_pointer_width = "64")]
+pub const DEFAULT_API_OBJECT_MAX_VERSIONS: usize = 9_223_372_036_854_775_807;
+
+/// Default and maximum accepted value for `RUSTFS_API_OBJECT_MAX_VERSIONS`.
+#[cfg(not(target_pointer_width = "64"))]
+pub const DEFAULT_API_OBJECT_MAX_VERSIONS: usize = usize::MAX;

crates/filemeta/Cargo.toml

@@ -44,6 +44,7 @@ tokio = { workspace = true, features = ["io-util", "macros", "sync", "fs", "rt-m
 xxhash-rust = { workspace = true, features = ["xxh64", "xxh3"] }
 bytes = { workspace = true, features = ["serde"] }
 rustfs-utils = { workspace = true, features = ["hash", "http"] }
+rustfs-config = { workspace = true, features = ["constants"] }
 byteorder = { workspace = true }
 tracing.workspace = true
 thiserror.workspace = true

crates/filemeta/src/filemeta.rs

@@ -70,12 +70,8 @@ const _XL_FLAG_INLINE_DATA: u8 = 1 << 2;
 const META_DATA_READ_DEFAULT: usize = 4 << 10;
 const MSGP_UINT32_SIZE: usize = 5;
 
-/// Default max object versions per object, aligned with MinIO's default.
-pub const DEFAULT_OBJECT_MAX_VERSIONS: usize = if usize::BITS >= 64 {
-    9_223_372_036_854_775_807
-} else {
-    usize::MAX
-};
+/// Default max object versions per object.
+pub const DEFAULT_OBJECT_MAX_VERSIONS: usize = rustfs_config::DEFAULT_API_OBJECT_MAX_VERSIONS;
 
 static OBJECT_MAX_VERSIONS: AtomicUsize = AtomicUsize::new(DEFAULT_OBJECT_MAX_VERSIONS);
 

docs/operations/scanner-benchmark-runbook.md

@@ -125,11 +125,6 @@ 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
@@ -137,19 +132,6 @@ 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/startup_runtime.rs

@@ -17,7 +17,7 @@ use crate::{
     startup_runtime_hooks::{init_profiling_runtime, install_default_crypto_provider, log_startup_runtime_diagnostics},
     startup_tls_material::init_outbound_tls_material,
 };
-use rustfs_config::ENV_API_OBJECT_MAX_VERSIONS;
+use rustfs_config::{DEFAULT_API_OBJECT_MAX_VERSIONS, ENV_API_OBJECT_MAX_VERSIONS};
 use rustfs_utils::EnvParseOutcome;
 use std::io::{Error, Result};
 
@@ -32,13 +32,8 @@ pub(crate) async fn init_startup_runtime_foundation(config: &Config) -> Result<(
 
 fn init_object_max_versions_config() -> Result<()> {
     let limit = match rustfs_utils::get_env_parse_outcome::<u64>(ENV_API_OBJECT_MAX_VERSIONS) {
-        EnvParseOutcome::Absent => rustfs_filemeta::DEFAULT_OBJECT_MAX_VERSIONS,
-        EnvParseOutcome::Invalid => {
-            return Err(Error::other(format!(
-                "{ENV_API_OBJECT_MAX_VERSIONS} must be a positive integer no greater than {}",
-                usize::MAX
-            )));
-        }
+        EnvParseOutcome::Absent => DEFAULT_API_OBJECT_MAX_VERSIONS,
+        EnvParseOutcome::Invalid => return Err(object_max_versions_config_error()),
         EnvParseOutcome::Parsed(value) => object_max_versions_limit_from_u64(value)?,
     };
 
@@ -47,18 +42,20 @@ fn init_object_max_versions_config() -> Result<()> {
 
 fn object_max_versions_limit_from_u64(value: u64) -> Result<usize> {
     if value == 0 {
-        return Err(Error::other(format!(
-            "{ENV_API_OBJECT_MAX_VERSIONS} must be a positive integer no greater than {}",
-            usize::MAX
-        )));
+        return Err(object_max_versions_config_error());
     }
 
-    usize::try_from(value).map_err(|_| {
-        Error::other(format!(
-            "{ENV_API_OBJECT_MAX_VERSIONS} must be a positive integer no greater than {}",
-            usize::MAX
-        ))
-    })
+    let limit = usize::try_from(value).map_err(|_| object_max_versions_config_error())?;
+    if limit > DEFAULT_API_OBJECT_MAX_VERSIONS {
+        return Err(object_max_versions_config_error());
+    }
+    Ok(limit)
+}
+
+fn object_max_versions_config_error() -> Error {
+    Error::other(format!(
+        "{ENV_API_OBJECT_MAX_VERSIONS} must be a positive integer no greater than {DEFAULT_API_OBJECT_MAX_VERSIONS}"
+    ))
 }
 
 #[cfg(test)]

scripts/scanner_abba.py

@@ -23,7 +23,6 @@ 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",
 )
@@ -60,12 +59,6 @@ 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)
@@ -290,62 +283,6 @@ 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:
@@ -405,7 +342,6 @@ 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
@@ -416,12 +352,6 @@ 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,8 +94,6 @@ 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":
@@ -104,12 +102,6 @@ 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
 
@@ -286,31 +278,6 @@ 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])
@@ -321,8 +288,7 @@ 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",
-                      "missing-pacing-metric"):
+                      "zero-pressure-samples", "pressure-sample-order", "attempt-accounting"):
             with self.subTest(fault=fault), tempfile.TemporaryDirectory() as directory:
                 self.root = Path(directory)
                 with self.assertRaises((ValueError, OSError, subprocess.SubprocessError)):
@@ -336,36 +302,6 @@ 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)