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)

.config/e2e-repl-nightly-selection.txt

@@ -1 +1 @@
-sha256=95c8adc016bbc0df9fb2afa24a108bcdf6567ec4d0518725a6cae301593ab556
+sha256=0fe8408874ccec3620262a9812d67920ddd72dc9edf0e36e0d0aed3f8bad026e

crates/scanner/src/scanner.rs

@@ -988,6 +988,17 @@ pub async fn init_scanner_with_recovery(
         return None;
     }
     Some(tokio::spawn(async move {
+        if let Err(error) = run_scanner_usage_recovery_intents_for_startup(ctx.clone(), storeapi.clone()).await {
+            warn!(
+                target: "rustfs::scanner",
+                event = EVENT_SCANNER_PERSIST_STATE,
+                component = LOG_COMPONENT_SCANNER,
+                subsystem = LOG_SUBSYSTEM_RUNTIME,
+                state = "recovery_intent_startup_discovery_failed",
+                error = %error,
+                "Disabled scanner recovery intent startup discovery failed"
+            );
+        }
         if let Err(error) = resume_scanner_cycle_cleanup(ctx, storeapi).await {
             warn!(
                 target: "rustfs::scanner",

crates/scanner/src/scanner/tests/recovery_control.rs

@@ -457,7 +457,7 @@ async fn scanner_recovery_intent_startup_rejects_corrupt_pending_record() {
 
 #[tokio::test]
 #[serial]
-async fn scanner_recovery_intent_disabled_startup_preserves_non_terminal_intent() {
+async fn scanner_recovery_intent_disabled_startup_executes_persisted_non_terminal_intent() {
     let (_dir, store) = setup_scanner_cycle_store().await;
     let record = match accept_scanner_usage_recovery_intent(
         store.clone(),
@@ -473,12 +473,12 @@ async fn scanner_recovery_intent_disabled_startup_preserves_non_terminal_intent(
     let restarted = restart_scanner_cycle_store_from(&store).await;
     run_disabled_startup(CancellationToken::new(), restarted.clone()).await;
 
-    let preserved = get_scanner_usage_recovery_intent(restarted, &record.intent_id)
+    let completed = get_scanner_usage_recovery_intent(restarted, &record.intent_id)
         .await
-        .expect("startup-skipped intent should read")
-        .expect("startup-skipped intent should remain durable");
-    assert_eq!(preserved.state, "accepted");
-    assert_eq!(preserved.intent_id, record.intent_id);
+        .expect("startup-executed intent should read")
+        .expect("startup-executed intent should remain durable");
+    assert_eq!(completed.state, "completed");
+    assert_eq!(completed.intent_id, record.intent_id);
 }
 
 #[tokio::test]