From 64829c704d9ab00863e09ed43e48c4d4ac028e60 Mon Sep 17 00:00:00 2001 From: houseme Date: Mon, 7 Sep 2026 12:54:38 +0800 Subject: [PATCH] fix(heal): retain MRF replay journal until retry anchor (#7340) 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: zhi22915 --- crates/heal/src/heal/mrf_queue.rs | 86 ++++++++++++++--- crates/heal/tests/mrf_pipeline_test.rs | 129 +++++++++++++++++-------- 2 files changed, 163 insertions(+), 52 deletions(-) diff --git a/crates/heal/src/heal/mrf_queue.rs b/crates/heal/src/heal/mrf_queue.rs index 8ebb461ff..5d6e86562 100644 --- a/crates/heal/src/heal/mrf_queue.rs +++ b/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) { /// 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 -/// (torn tails truncate via the per-record CRC). Public for integration tests; -/// the live consumer invokes this through [`replay_into`] at startup. +/// merged by the manager's dedup key; the journal is retained whenever replay +/// cannot fully hand off a successor in-memory snapshot (torn tails truncate +/// 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) -> 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, 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) { - rustfs_common::mrf_channel::release_mrf_intent(&intent); + 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, mut receiver: mpsc::Receive backoff_until: None, }; - // Replay: read the journal, re-arm intents (duplicates are merged by the - // manager's dedup key), then drop the file so the next flush starts clean. + // Replay reads the journal and re-arms intents. The startup journal stays + // 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 - // manager was full and backoff armed) must be re-persisted by the next - // flush or a crash before it would lose those intents. + // Anything still pending (e.g. the manager was full and backoff armed) + // must be re-persisted by the next flush before replay can delete the + // 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); diff --git a/crates/heal/tests/mrf_pipeline_test.rs b/crates/heal/tests/mrf_pipeline_test.rs index a1b9ee502..fdcf7848c 100644 --- a/crates/heal/tests/mrf_pipeline_test.rs +++ b/crates/heal/tests/mrf_pipeline_test.rs @@ -60,6 +60,29 @@ fn make_manager(storage: Arc) -> Arc { )) } +async fn register_local_disks(disk_paths: &[std::path::PathBuf], cmd_line: &str) { + let mut endpoints: Vec = 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 { @@ -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 = 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: "mrf-test".to_string(), - platform: String::new(), - }; - init_local_disks(EndpointServerPools::from(vec![pool])) - .await - .expect("local disks should register"); + register_local_disks(&disk_paths, "mrf-test").await; 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 = 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: "mrf-authoritative-test".to_string(), - platform: String::new(), - }; - init_local_disks(EndpointServerPools::from(vec![pool])) - .await - .expect("local disks should register"); + register_local_disks(&disk_paths, "mrf-authoritative-test").await; 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" + ); +}