gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-07 20:46:11 +00:00
Code Issues Packages Projects Releases Wiki Activity

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

Browse Source 
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 <qiuzgang@gmail.com>
This commit is contained in:
houseme
2026-09-07 12:54:38 +08:00
committed by GitHub
parent 17ba30f648
commit 64829c704d
2 changed files with 163 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/heal/src/heal/mrf_queue.rs
+74 -12
View File
@@ -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
/// (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<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) {
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<HealManager>, 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);
crates/heal/tests/mrf_pipeline_test.rs
+89 -40
View File
@@ -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
.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<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: "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"
);
}