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,7 +9906,6 @@ dependencies = [
  "regex",
  "rmp",
  "rmp-serde",
- "rustfs-config",
  "rustfs-utils",
  "s3s",
  "serde",

crates/config/src/constants/api.rs

@@ -100,10 +100,5 @@ 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 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;
+/// Default for `RUSTFS_API_OBJECT_MAX_VERSIONS`.
+pub const DEFAULT_API_OBJECT_MAX_VERSIONS: u64 = 9_223_372_036_854_775_807;

crates/filemeta/Cargo.toml

@@ -44,7 +44,6 @@ 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,8 +70,12 @@ 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.
-pub const DEFAULT_OBJECT_MAX_VERSIONS: usize = rustfs_config::DEFAULT_API_OBJECT_MAX_VERSIONS;
+/// 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
+};
 
 static OBJECT_MAX_VERSIONS: AtomicUsize = AtomicUsize::new(DEFAULT_OBJECT_MAX_VERSIONS);
 

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

@@ -183,6 +183,72 @@ 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());
@@ -1293,6 +1359,7 @@ fn replacement_identity(
 
 enum MockHealObjectOutcome {
     RetryableLock,
+    RetryableLockTimeout,
     OkWithOtherError(&'static str),
     ErrOther(&'static str),
     DanglingGraceDeferred,
@@ -1436,6 +1503,13 @@ 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))))
                 }
@@ -1468,6 +1542,13 @@ 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::{DEFAULT_API_OBJECT_MAX_VERSIONS, ENV_API_OBJECT_MAX_VERSIONS};
+use rustfs_config::ENV_API_OBJECT_MAX_VERSIONS;
 use rustfs_utils::EnvParseOutcome;
 use std::io::{Error, Result};
 
@@ -32,8 +32,13 @@ 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 => DEFAULT_API_OBJECT_MAX_VERSIONS,
-        EnvParseOutcome::Invalid => return Err(object_max_versions_config_error()),
+        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::Parsed(value) => object_max_versions_limit_from_u64(value)?,
     };
 
@@ -42,20 +47,18 @@ fn init_object_max_versions_config() -> Result<()> {
 
 fn object_max_versions_limit_from_u64(value: u64) -> Result<usize> {
     if value == 0 {
-        return Err(object_max_versions_config_error());
+        return 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}"
-    ))
+    usize::try_from(value).map_err(|_| {
+        Error::other(format!(
+            "{ENV_API_OBJECT_MAX_VERSIONS} must be a positive integer no greater than {}",
+            usize::MAX
+        ))
+    })
 }
 
 #[cfg(test)]