fix(heal): retain MRF replay journal until retry anchor

Keep the startup journal on disk when replay cannot fully re-arm or the heal manager refuses a replayed intent with Full/QueueFull. The next live snapshot can still advance the journal after the retry anchor is durable again.

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

crates/heal/src/heal/mrf_queue.rs

@@ -186,6 +186,11 @@ impl MrfQueue {
         MrfQueuePushResult::Enqueued
     }
 
+    fn raise_limits_for_replay(&mut self, intents: usize, bytes: usize) {
+        self.capacity = self.capacity.max(self.pending.len().saturating_add(intents));
+        self.byte_budget = self.byte_budget.max(self.bytes.saturating_add(bytes));
+    }
+
     /// Bool compatibility adapter: only a newly executable queue item is
     /// reported as accepted; a coalesced duplicate is not durable admission.
     #[cfg(test)]
@@ -655,9 +660,10 @@ pub fn spawn_mrf_consumer(manager: Arc<HealManager>) {
 
 /// Replay the durable journal into a fresh pending queue and submit whatever
 /// it armed. Returns the number of intact intents replayed. Duplicates are
-/// merged by the manager's dedup key; the journal file is removed once read
+/// merged by the manager's dedup key; the journal is retained whenever replay
-/// (torn tails truncate via the per-record CRC). Public for integration tests;
+/// cannot fully hand off a successor in-memory snapshot (torn tails truncate
-/// the live consumer invokes this through [`replay_into`] at startup.
+/// via the per-record CRC). Public for integration tests; the live consumer
+/// invokes this through [`replay_into`] at startup.
 pub async fn replay_journal_once(manager: &Arc<HealManager>) -> usize {
     let config = MrfConsumerConfig::default();
     let mut queue = MrfQueue::new(config.queue_capacity, config.journal_max_bytes);
@@ -670,7 +676,13 @@ struct ReplayOutcome {
     journal_on_disk: bool,
 }
 
-/// Shared replay core: read + decode + re-arm + delete, then drain what fits.
+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
+/// startup journal is removed only after every replayed record has either
+/// reached the manager or been proven redundant inside the in-memory queue.
 async fn replay_into(
     manager: &Arc<HealManager>,
     queue: &mut MrfQueue,
@@ -703,13 +715,26 @@ async fn replay_into(
     }
     counter!("rustfs_heal_mrf_replayed_total").increment(u64::try_from(intents.len()).unwrap_or(u64::MAX));
     let replayed = intents.len();
+    let replay_bytes = intents
+        .iter()
+        .fold(0usize, |total, intent| total.saturating_add(intent.estimated_bytes()));
+    // The decoded journal is already resident in memory. Allow the startup
+    // queue to arm that full bounded snapshot so a later flush can become the
+    // successor anchor instead of overwriting the old journal with only a
+    // prefix.
+    queue.raise_limits_for_replay(intents.len(), replay_bytes);
+    let mut rearm_incomplete = false;
     for intent in intents {
         let result = queue.try_push_typed(intent.clone());
-        if !matches!(result, MrfQueuePushResult::Enqueued) {
+        match result {
+            MrfQueuePushResult::Enqueued => {}
+            MrfQueuePushResult::Coalesced => rustfs_common::mrf_channel::release_mrf_intent(&intent),
+            MrfQueuePushResult::Rejected => {
+                rearm_incomplete = true;
                 rustfs_common::mrf_channel::release_mrf_intent(&intent);
             }
         }
-    let journal_on_disk = !delete_journals().await;
+    }
 
     // Drain the replayed intents immediately; whatever the manager refuses
     // stays armed in `queue` for the consumer's retry loop.
@@ -729,6 +754,11 @@ async fn replay_into(
             }
         }
     }
+    let journal_on_disk = if replay_must_retain_journal(rearm_incomplete, queue.depth()) {
+        true
+    } else {
+        !delete_journals().await
+    };
     ReplayOutcome {
         replayed,
         journal_on_disk,
@@ -748,13 +778,13 @@ async fn run_mrf_consumer(manager: Arc<HealManager>, mut receiver: mpsc::Receive
         backoff_until: None,
     };
 
-    // Replay: read the journal, re-arm intents (duplicates are merged by the
+    // Replay reads the journal and re-arms intents. The startup journal stays
-    // manager's dedup key), then drop the file so the next flush starts clean.
+    // 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;
-    // The replay deleted the journal file; anything still pending (e.g. the
+    // Anything still pending (e.g. the manager was full and backoff armed)
-    // manager was full and backoff armed) must be re-persisted by the next
+    // must be re-persisted by the next flush before replay can delete the
-    // flush or a crash before it would lose those intents.
+    // startup anchor.
     runtime.dirty = runtime.queue.depth() > 0;
 
     let mut flush_tick = tokio::time::interval(runtime.config.flush_interval);
@@ -890,6 +920,38 @@ mod tests {
         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),
+            "a rejected replay record still needs its disk anchor"
+        );
+        assert!(
+            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),
+            "only a fully consumed replay snapshot may be deleted"
+        );
+    }
+
+    #[test]
+    fn replay_can_arm_more_records_than_live_queue_budget() {
+        let mut queue = MrfQueue::new(1, intent("bucket", "object-0", 0).estimated_bytes());
+        let intents = vec![intent("bucket", "object-0", 0), intent("bucket", "object-1", 0)];
+        let bytes = intents
+            .iter()
+            .fold(0usize, |total, intent| total.saturating_add(intent.estimated_bytes()));
+
+        queue.raise_limits_for_replay(intents.len(), bytes);
+
+        for intent in intents {
+            assert_eq!(queue.try_push_typed(intent), MrfQueuePushResult::Enqueued);
+        }
+        assert_eq!(queue.depth(), 2);
+    }
+
     #[test]
     fn queue_enforces_count_and_byte_ceilings() {
         let mut queue = MrfQueue::new(2, usize::MAX);

crates/heal/tests/mrf_pipeline_test.rs

@@ -60,6 +60,29 @@ fn make_manager(storage: Arc<dyn HealStorageAPI>) -> Arc<HealManager> {
     ))
 }
 
+async fn register_local_disks(disk_paths: &[std::path::PathBuf], cmd_line: &str) {
+    let mut endpoints: Vec<Endpoint> = disk_paths
+        .iter()
+        .map(|p| Endpoint::try_from(p.to_string_lossy().as_ref()).expect("endpoint from disk path"))
+        .collect();
+    for (i, endpoint) in endpoints.iter_mut().enumerate() {
+        endpoint.set_pool_index(0);
+        endpoint.set_set_index(0);
+        endpoint.set_disk_index(i);
+    }
+    let pool = PoolEndpoints {
+        legacy: false,
+        set_count: 1,
+        drives_per_set: endpoints.len(),
+        endpoints: Endpoints::from(endpoints),
+        cmd_line: cmd_line.to_string(),
+        platform: String::new(),
+    };
+    init_local_disks(EndpointServerPools::from(vec![pool]))
+        .await
+        .expect("local disks should register");
+}
+
 /// Encode one journal record independently of the implementation, so a format
 /// drift between writer and this fixture fails loudly here.
 fn journal_record(kind: u8, bucket: &str, object: &str, version: Option<[u8; 16]>, attempts: u8) -> Vec<u8> {
@@ -151,26 +174,7 @@ async fn journal_replay_arms_intents_and_deletes_the_file() {
 
     // The journal reader resolves disks through the process-local disk map;
     // register the environment's disks the same way server startup does.
-    let mut endpoints: Vec<Endpoint> = disk_paths
+    register_local_disks(&disk_paths, "mrf-test").await;
-        .iter()
-        .map(|p| Endpoint::try_from(p.to_string_lossy().as_ref()).expect("endpoint from disk path"))
-        .collect();
-    for (i, endpoint) in endpoints.iter_mut().enumerate() {
-        endpoint.set_pool_index(0);
-        endpoint.set_set_index(0);
-        endpoint.set_disk_index(i);
-    }
-    let pool = PoolEndpoints {
-        legacy: false,
-        set_count: 1,
-        drives_per_set: endpoints.len(),
-        endpoints: Endpoints::from(endpoints),
-        cmd_line: "mrf-test".to_string(),
-        platform: String::new(),
-    };
-    init_local_disks(EndpointServerPools::from(vec![pool]))
-        .await
-        .expect("local disks should register");
 
     let mut journal = journal_record(1, "replay-bucket", "replay-object", Some([9u8; 16]), 0);
     journal.extend(journal_record(3, "replay-bucket", "partial-object", None, 1));
@@ -213,26 +217,7 @@ async fn journal_replay_arms_intents_and_deletes_the_file() {
 #[serial]
 async fn authoritative_journal_is_not_merged_with_legacy_mirror() {
     let (disk_paths, storage) = heal_env().await;
-    let mut endpoints: Vec<Endpoint> = disk_paths
+    register_local_disks(&disk_paths, "mrf-authoritative-test").await;
-        .iter()
-        .map(|p| Endpoint::try_from(p.to_string_lossy().as_ref()).expect("endpoint from disk path"))
-        .collect();
-    for (i, endpoint) in endpoints.iter_mut().enumerate() {
-        endpoint.set_pool_index(0);
-        endpoint.set_set_index(0);
-        endpoint.set_disk_index(i);
-    }
-    let pool = PoolEndpoints {
-        legacy: false,
-        set_count: 1,
-        drives_per_set: endpoints.len(),
-        endpoints: Endpoints::from(endpoints),
-        cmd_line: "mrf-authoritative-test".to_string(),
-        platform: String::new(),
-    };
-    init_local_disks(EndpointServerPools::from(vec![pool]))
-        .await
-        .expect("local disks should register");
 
     let authoritative = journal_record(1, "authoritative-bucket", "authoritative-object", None, 0);
     let legacy = journal_record(1, "legacy-bucket", "legacy-object", None, 0);
@@ -264,3 +249,67 @@ async fn authoritative_journal_is_not_merged_with_legacy_mirror() {
             && !Path::new(path).join(META_BUCKET).join(SCOPED_JOURNAL_REL).exists()
     }));
 }
+
+/// If replay reaches a full heal-manager queue, the old journal remains the
+/// durable restart anchor until a later consumer flush publishes the pending
+/// successor snapshot.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+#[serial]
+async fn journal_replay_retains_file_when_manager_is_full() {
+    let (disk_paths, storage) = heal_env().await;
+    register_local_disks(&disk_paths, "mrf-full-replay-test").await;
+
+    let mut journal = journal_record(1, "full-bucket", "first-object", None, 0);
+    journal.extend(journal_record(1, "full-bucket", "second-object", None, 0));
+    write_journal_path_to_disks(&disk_paths, SCOPED_JOURNAL_REL, &journal);
+    write_journal_path_to_disks(&disk_paths, JOURNAL_REL, &journal);
+
+    let manager = Arc::new(HealManager::new(
+        storage.clone(),
+        Some(HealConfig {
+            queue_size: 1,
+            heal_interval: Duration::from_secs(3600),
+            enable_auto_heal: false,
+            ..Default::default()
+        }),
+    ));
+    let replayed = mrf_queue::replay_journal_once(&manager).await;
+    assert_eq!(replayed, 2, "both records must be decoded before manager admission");
+    assert_eq!(
+        manager.operations_snapshot().await.queued_by_source.mrf,
+        1,
+        "only the first record can enter a one-slot manager queue"
+    );
+    assert!(
+        disk_paths
+            .iter()
+            .all(|path| Path::new(path).join(META_BUCKET).join(SCOPED_JOURNAL_REL).exists()),
+        "replay must keep the authoritative journal when a later record is pending retry"
+    );
+
+    let restarted = Arc::new(HealManager::new(
+        storage,
+        Some(HealConfig {
+            queue_size: 1,
+            heal_interval: Duration::from_secs(3600),
+            enable_auto_heal: false,
+            ..Default::default()
+        }),
+    ));
+    let replayed_after_restart = mrf_queue::replay_journal_once(&restarted).await;
+    assert_eq!(
+        replayed_after_restart, 2,
+        "retained startup journal must replay again after a process restart"
+    );
+    assert_eq!(
+        restarted.operations_snapshot().await.queued_by_source.mrf,
+        1,
+        "the restart sees the same bounded admission state instead of a lost tail"
+    );
+    assert!(
+        disk_paths
+            .iter()
+            .all(|path| Path::new(path).join(META_BUCKET).join(SCOPED_JOURNAL_REL).exists()),
+        "the anchor remains until a successor snapshot can safely replace it"
+    );
+}