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);
 

crates/heal/src/heal/task/tests.rs

@@ -183,72 +183,6 @@ mod canonical_outcome {
         assert_eq!((progress.objects_scanned, progress.objects_healed, progress.objects_failed), (2, 1, 1));
     }
 
-    #[tokio::test(start_paused = true)]
-    async fn admin_cluster_lock_timeout_exhaustion_keeps_progress_and_retry_outcome() {
-        let storage = Arc::new(MockStorage::default());
-        storage.heal_object_outcomes.lock().expect("outcomes").insert(
-            "object-a".to_string(),
-            (0..4).map(|_| MockHealObjectOutcome::RetryableLockTimeout).collect(),
-        );
-        let mut request = HealRequest::new(
-            HealType::Cluster,
-            HealOptions {
-                recursive: true,
-                timeout: None,
-                ..Default::default()
-            },
-            HealPriority::Normal,
-        );
-        request.source = HealRequestSource::Admin;
-        let task = HealTask::from_request(request, storage.clone());
-
-        let err = task
-            .execute()
-            .await
-            .expect_err("legacy adapter still returns the batch failure detail");
-        assert!(
-            err.to_string()
-                .contains("Lock error: Lock acquisition timeout for resource 'object-a' after 5s"),
-            "lock timeout must remain actionable in the retained failure detail: {err}"
-        );
-
-        let outcome = task.get_outcome().await;
-        assert_eq!(outcome.execution, HealExecutionOutcome::CompletedWithErrors);
-        assert_eq!(outcome.coverage, HealTraversalCoverage::Complete);
-        assert_eq!(
-            (
-                outcome.counters.processed,
-                outcome.counters.failed,
-                outcome.counters.unknown,
-                outcome.counters.attempt_failures
-            ),
-            (2, 1, 1, 4)
-        );
-        let failed = outcome
-            .objects
-            .iter()
-            .find(|item| item.identity.object == "object-a")
-            .expect("lock-contended object outcome");
-        assert_eq!(failed.disposition, HealObjectDisposition::Failed(HealFailureClass::RetryExhausted));
-        assert!(
-            failed
-                .detail
-                .as_deref()
-                .is_some_and(|detail| detail.contains("Lock error: Lock acquisition timeout for resource 'object-a' after 5s")),
-            "exhausted lock detail stays observable"
-        );
-
-        let progress = task.get_progress().await;
-        assert_eq!((progress.objects_scanned, progress.objects_healed, progress.objects_failed), (2, 1, 1));
-        let (legacy_summary, legacy_detail) = outcome.legacy_status("finished", None);
-        assert_eq!(legacy_summary, "stopped");
-        assert_eq!(legacy_detail.as_deref(), Some("heal traversal completed with errors: 1 failed objects"));
-        assert_eq!(
-            storage.heal_object_calls.lock().expect("object calls").as_slice(),
-            ["object-a", "object-b", "object-a", "object-a", "object-a"]
-        );
-    }
-
     #[tokio::test(start_paused = true)]
     async fn retry_success_counts_one_terminal_outcome() {
         let storage = Arc::new(MockStorage::default());
@@ -1359,7 +1293,6 @@ fn replacement_identity(
 
 enum MockHealObjectOutcome {
     RetryableLock,
-    RetryableLockTimeout,
     OkWithOtherError(&'static str),
     ErrOther(&'static str),
     DanglingGraceDeferred,
@@ -1503,13 +1436,6 @@ impl HealStorageAPI for MockStorage {
                         owner: "competing-writer".to_string(),
                     }))),
                 )),
-                MockHealObjectOutcome::RetryableLockTimeout => Ok((
-                    HealResultItem::default(),
-                    Some(Error::Storage(EcstoreError::Lock(rustfs_lock::LockError::Timeout {
-                        resource: object.to_string(),
-                        timeout: Duration::from_secs(5),
-                    }))),
-                )),
                 MockHealObjectOutcome::RetryableSlowDown => {
                     Ok((HealResultItem::default(), Some(Error::Storage(EcstoreError::SlowDown))))
                 }
@@ -1542,13 +1468,6 @@ impl HealStorageAPI for MockStorage {
                         owner: "competing-writer".to_string(),
                     }))),
                 )),
-                MockHealObjectOutcome::RetryableLockTimeout => Ok((
-                    HealResultItem::default(),
-                    Some(Error::Storage(EcstoreError::Lock(rustfs_lock::LockError::Timeout {
-                        resource: object.to_string(),
-                        timeout: Duration::from_secs(5),
-                    }))),
-                )),
                 MockHealObjectOutcome::RetryableSlowDown => {
                     Ok((HealResultItem::default(), Some(Error::Storage(EcstoreError::SlowDown))))
                 }

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)]