From cad0fd9b2f53b37abbe05825bd056d68acf5ab87 Mon Sep 17 00:00:00 2001 From: cxymds Date: Tue, 4 Aug 2026 06:01:39 +0800 Subject: [PATCH] fix(replication): retain MRF failures during recovery (#5667) * fix(replication): retain MRF failures during recovery * fix(replication): harden MRF recovery retries * fix(replication): preserve MRF recovery durability * test(replication): assert MRF append entries explicitly * fix(replication): detect committed MRF append retries * fix(replication): tighten MRF persister recovery * fix(replication): drain overflow after recovery shrink * test(replication): avoid repeated MRF recovery decode * test(replication): cover recovery overflow suffix * fix(replication): gate MRF recovery flushes * test(replication): satisfy MRF clippy checks * fix(replication): drain closed MRF channels --------- Co-authored-by: overtrue --- .../bucket/replication/replication_pool.rs | 1148 +++++++++++++++-- 1 file changed, 1043 insertions(+), 105 deletions(-) diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index a6eca120c..cb6546c76 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/crates/ecstore/src/bucket/replication/replication_pool.rs @@ -51,6 +51,7 @@ use super::replication_versioning_boundary::ReplicationVersioningStore; use super::runtime_boundary as runtime_sources; use futures_util::stream::{self, StreamExt}; use metrics::{counter, histogram}; +use rustfs_utils::hash::HashAlgorithm; use rustfs_utils::http::{SUFFIX_REPLICATION_TIMESTAMP, get_str}; use std::collections::HashMap; use std::collections::hash_map::Entry; @@ -83,6 +84,9 @@ const DELETE_BATCH_ADMISSION_CONCURRENCY: usize = 16; const METRIC_DELETE_BATCH_ITEMS_TOTAL: &str = "rustfs_replication_delete_batch_items_total"; const METRIC_DELETE_BATCH_SIZE: &str = "rustfs_replication_delete_batch_size"; const MRF_CORRUPT_FILE_PREFIX: &str = "config/replication/mrf.corrupt"; +const MRF_PENDING_CAP: usize = 200_000; +const MRF_RETRY_INITIAL_DELAY: Duration = Duration::from_millis(100); +const MRF_RETRY_MAX_DELAY: Duration = Duration::from_secs(5); #[derive(Debug, Default)] pub struct DurableMrfBacklog { @@ -214,6 +218,45 @@ impl DurableMrfBacklogTracker { } } +fn durable_mrf_backlog_tracker_from_entries(entries: &[MrfReplicateEntry]) -> DurableMrfBacklogTracker { + let mut tracker = DurableMrfBacklogTracker { + available: true, + ..Default::default() + }; + for entry in entries { + tracker.add_entry(entry); + } + tracker +} + +fn move_staged_mrf_entries(pending: &mut Vec, staged: &mut Vec) -> Vec { + let pending_capacity = MRF_PENDING_CAP.saturating_sub(pending.len()); + let mut staged_entries = std::mem::take(staged); + let capped_batch = staged_entries.split_off(pending_capacity.min(staged_entries.len())); + pending.append(&mut staged_entries); + capped_batch +} + +#[derive(Debug)] +struct PendingMrfAppend { + digest: [u8; 32], + entry_count: usize, +} + +fn mrf_payload_digest(data: &[u8]) -> [u8; 32] { + let encoded = HashAlgorithm::SHA256.hash_encode(data); + let mut digest = [0; 32]; + digest.copy_from_slice(encoded.as_ref()); + digest +} + +fn add_durable_mrf_suffix(tracker: &mut DurableMrfBacklogTracker, entries: &[MrfReplicateEntry], count: usize) { + let start = entries.len().saturating_sub(count); + for entry in &entries[start..] { + tracker.add_entry(entry); + } +} + #[derive(Debug, Clone, Default)] struct MrfBacklogObservabilityTracker { buckets: HashMap, @@ -1411,33 +1454,49 @@ impl ReplicationPool { let recovery_result = self.mrf_recovery_result.clone(); let handle = tokio::spawn(async move { + let mut staged = Vec::new(); + let mut staging_closed = false; + let retry_timer = tokio::time::sleep(Duration::ZERO); + tokio::pin!(retry_timer); let mut pending = loop { - match ReplicationConfigStore::read(storage.clone(), ReplicationMetadataStore::MRF_REPLICATION_FILE).await { - Ok(data) => match decode_mrf_file(&data) { - Ok(entries) => break entries, - Err(error) => { - warn!( - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_REPLICATION, - error = %error, - "Failed to seed MRF persister from the startup recovery file; retrying without overwriting it" - ); - tokio::time::sleep(Duration::from_millis(100)).await; - } + tokio::select! { + entry = rx.recv(), if staged.len() < MRF_PENDING_CAP && !staging_closed => match entry { + Some(entry) => { + observe_mrf_pending(&entry); + staged.push(entry); + }, + None => staging_closed = true, }, - Err(EcstoreError::ConfigNotFound) => break Vec::new(), - Err(error) => { - warn!( - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_REPLICATION, - error = %error, - "Failed to read the startup MRF backlog for persister seeding; retrying without overwriting it" - ); - tokio::time::sleep(Duration::from_millis(100)).await; + _ = &mut retry_timer => { + match ReplicationConfigStore::read(storage.clone(), ReplicationMetadataStore::MRF_REPLICATION_FILE).await { + Ok(data) => match decode_mrf_file(&data) { + Ok(entries) => break entries, + Err(error) => { + warn!( + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION, + error = %error, + "Failed to seed MRF persister from the startup recovery file; retrying without overwriting it" + ); + } + }, + Err(EcstoreError::ConfigNotFound) => break Vec::new(), + Err(error) => { + warn!( + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION, + error = %error, + "Failed to read the startup MRF backlog for persister seeding; retrying without overwriting it" + ); + } + } + retry_timer.as_mut().reset(tokio::time::Instant::now() + MRF_RETRY_INITIAL_DELAY); } } }; let initial_pending_len = pending.len(); + let mut capped_batch = move_staged_mrf_entries(&mut pending, &mut staged); + let pending_staged_len = pending.len().saturating_sub(initial_pending_len); // The on-disk MRF file is a restart-recovery backstop: entries are // only replayed (and the file cleared) at startup, never during the // run. So the file must hold the *cumulative* set of overflow entries @@ -1446,22 +1505,23 @@ impl ReplicationPool { // let the next flush overwrite the file and drop everything written // earlier (backlog#859 / #799 B10). Bounded by `MRF_PENDING_CAP` so a // sustained failure storm can't grow it without limit. - const MRF_PENDING_CAP: usize = 200_000; let mut durable_tracker = DurableMrfBacklogTracker { available: true, ..Default::default() }; - for entry in &pending { + for entry in &pending[..initial_pending_len] { durable_tracker.add_entry(entry); } if initial_pending_len > 0 { set_durable_mrf_backlog_snapshot(durable_tracker.clone().into_snapshot()); } - let mut new_entries_pending_stats = 0usize; - let mut new_entries_pending_observability = 0usize; - let mut dirty = false; + let mut new_entries_pending_stats = pending_staged_len; + let mut new_entries_pending_observability = pending_staged_len; + let mut dirty = pending_staged_len > 0; let mut capped = initial_pending_len >= MRF_PENDING_CAP; let mut recovery_applied = false; + let mut channel_closed = false; + let mut capped_payload = None; if capped { warn!( component = LOG_COMPONENT_ECSTORE, @@ -1477,9 +1537,16 @@ impl ReplicationPool { interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); loop { - if pending.len() >= MRF_PENDING_CAP && recovery_applied { + if !channel_closed && rx.is_closed() && rx.is_empty() { + channel_closed = true; + } + if recovery_applied && channel_closed && !dirty && capped_batch.is_empty() { + break; + } + if recovery_applied && (pending.len() >= MRF_PENDING_CAP || !capped_batch.is_empty()) { if dirty { if let Some(duration_millis) = flush_mrf_to_disk(&pending, &storage).await { + add_durable_mrf_suffix(&mut durable_tracker, &pending, new_entries_pending_stats); set_durable_mrf_backlog_snapshot(durable_tracker.clone().into_snapshot()); let observe_start = pending.len().saturating_sub(new_entries_pending_observability); observe_mrf_pending_flushed(&pending[observe_start..], duration_millis); @@ -1507,36 +1574,36 @@ impl ReplicationPool { "MRF pending backlog reached capacity — applying backpressure" ); } - let mut batch = Vec::new(); - while let Ok(entry) = rx.try_recv() { - batch.push(entry); - } - if !batch.is_empty() { - for entry in &batch { - durable_tracker.add_entry(entry); + if capped_batch.is_empty() { + while let Ok(entry) = rx.try_recv() { + capped_batch.push(entry); + } + for entry in &capped_batch { observe_mrf_pending(entry); } - if !dirty && let Some(duration_millis) = append_mrf_entries_to_disk(&batch, &storage).await { - set_durable_mrf_backlog_snapshot(durable_tracker.clone().into_snapshot()); - observe_mrf_pending_flushed(&batch, duration_millis); - dec_mrf_entries(stats.as_ref(), &batch); - } else { - new_entries_pending_stats += batch.len(); - new_entries_pending_observability += batch.len(); - pending.extend(batch); - dirty = true; - } } - if rx.is_closed() && rx.is_empty() && !dirty { + if !capped_batch.is_empty() + && let Some(duration_millis) = + append_mrf_entries_to_disk(&capped_batch, &storage, &mut capped_payload, &pending).await + { + for entry in &capped_batch { + durable_tracker.add_entry(entry); + } + set_durable_mrf_backlog_snapshot(durable_tracker.clone().into_snapshot()); + observe_mrf_pending_flushed(&capped_batch, duration_millis); + dec_mrf_entries(stats.as_ref(), &capped_batch); + capped_batch.clear(); + capped_payload = None; + } + if channel_closed && capped_batch.is_empty() && !dirty { break; } interval.tick().await; continue; } tokio::select! { - entry = rx.recv(), if pending.len() < MRF_PENDING_CAP => match entry { + entry = rx.recv(), if (!channel_closed || !rx.is_empty()) && pending.len() < MRF_PENDING_CAP => match entry { Some(e) => { - durable_tracker.add_entry(&e); observe_mrf_pending(&e); pending.push(e); new_entries_pending_stats += 1; @@ -1546,9 +1613,11 @@ impl ReplicationPool { // since the last write (measured against the flushed // set, not the absolute length, so a large backlog is // not rewritten on every single add). - if new_entries_pending_stats >= 1000 + if recovery_applied + && new_entries_pending_stats >= 1000 && let Some(duration_millis) = flush_mrf_to_disk(&pending, &storage).await { + add_durable_mrf_suffix(&mut durable_tracker, &pending, new_entries_pending_stats); set_durable_mrf_backlog_snapshot(durable_tracker.clone().into_snapshot()); let observe_start = pending.len().saturating_sub(new_entries_pending_observability); observe_mrf_pending_flushed(&pending[observe_start..], duration_millis); @@ -1560,17 +1629,7 @@ impl ReplicationPool { } } None => { - // Channel closed (pool shutting down) — final flush. - if dirty && let Some(duration_millis) = flush_mrf_to_disk(&pending, &storage).await { - set_durable_mrf_backlog_snapshot(durable_tracker.clone().into_snapshot()); - let observe_start = pending.len().saturating_sub(new_entries_pending_observability); - observe_mrf_pending_flushed(&pending[observe_start..], duration_millis); - if new_entries_pending_stats > 0 { - let stats_start = pending.len().saturating_sub(new_entries_pending_stats); - dec_mrf_entries(stats.as_ref(), &pending[stats_start..]); - } - } - break; + channel_closed = true; } }, _ = recovery_complete.notified(), if !recovery_applied => { @@ -1579,18 +1638,25 @@ impl ReplicationPool { let new_entries = pending.split_off(initial_pending_len.min(pending.len())); pending = retry_entries; pending.extend(new_entries); - durable_tracker = DurableMrfBacklogTracker { - available: true, - ..Default::default() - }; - for entry in &pending { - durable_tracker.add_entry(entry); + let pending_capacity = MRF_PENDING_CAP.saturating_sub(pending.len()); + let moved_count = pending_capacity.min(capped_batch.len()); + if moved_count > 0 { + pending.extend(capped_batch.drain(..moved_count)); + new_entries_pending_stats += moved_count; + new_entries_pending_observability += moved_count; } + durable_tracker = durable_mrf_backlog_tracker_from_entries( + &pending[..pending.len().saturating_sub(new_entries_pending_stats)], + ); dirty = true; } }, _ = interval.tick() => { - if dirty && let Some(duration_millis) = flush_mrf_to_disk(&pending, &storage).await { + if recovery_applied + && dirty + && let Some(duration_millis) = flush_mrf_to_disk(&pending, &storage).await + { + add_durable_mrf_suffix(&mut durable_tracker, &pending, new_entries_pending_stats); set_durable_mrf_backlog_snapshot(durable_tracker.clone().into_snapshot()); let observe_start = pending.len().saturating_sub(new_entries_pending_observability); observe_mrf_pending_flushed(&pending[observe_start..], duration_millis); @@ -2075,35 +2141,100 @@ async fn queue_mrf_save_entry( async fn quarantine_mrf_file(storage: &Arc, data: &[u8]) { let quarantine_file = format!("{MRF_CORRUPT_FILE_PREFIX}.{}.bin", OffsetDateTime::now_utc().unix_timestamp_nanos()); - match ReplicationConfigStore::save(storage.clone(), &quarantine_file, data.to_vec()).await { - Ok(()) => { - warn!( + let payload = data.to_vec(); + let mut retry_delay = MRF_RETRY_INITIAL_DELAY; + loop { + match ReplicationConfigStore::save(storage.clone(), &quarantine_file, payload.clone()).await { + Ok(()) => { + warn!( + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION, + file = %quarantine_file, + "Quarantined corrupt MRF recovery file" + ); + break; + } + Err(error) => warn!( component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_REPLICATION, file = %quarantine_file, - "Quarantined corrupt MRF recovery file" - ); - // Clear the active path only after the quarantine copy succeeds. An empty - // config is treated as absent, preventing the same corrupt bytes from being - // quarantined again on every restart while preserving the forensic copy. - if let Err(error) = - ReplicationConfigStore::save(storage.clone(), ReplicationMetadataStore::MRF_REPLICATION_FILE, Vec::new()).await - { + error = %error, + "Failed to quarantine corrupt MRF recovery file; retrying without overwriting it" + ), + } + tokio::time::sleep(retry_delay).await; + retry_delay = retry_delay.saturating_mul(2).min(MRF_RETRY_MAX_DELAY); + } + + // Clear the active path only if it still contains the bytes that were + // quarantined. The write lock closes the read/clear race with another node + // replacing the active generation while this node retries quarantine. + retry_delay = MRF_RETRY_INITIAL_DELAY; + loop { + let lock = match storage + .new_ns_lock( + ReplicationMetadataStore::rustfs_meta_bucket(), + ReplicationMetadataStore::MRF_REPLICATION_FILE, + ) + .await + { + Ok(lock) => lock, + Err(error) => { warn!( component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_REPLICATION, error = %error, - "Failed to clear the corrupt MRF recovery path after quarantine" + "Failed to acquire the MRF recovery lock before clearing quarantine source; retrying" ); + tokio::time::sleep(retry_delay).await; + retry_delay = retry_delay.saturating_mul(2).min(MRF_RETRY_MAX_DELAY); + continue; } + }; + let _guard = match lock.get_write_lock(ReplicationLockTiming::acquire_timeout()).await { + Ok(guard) => guard, + Err(error) => { + warn!( + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION, + error = %error, + "Failed to acquire the MRF recovery lock before clearing quarantine source; retrying" + ); + tokio::time::sleep(retry_delay).await; + retry_delay = retry_delay.saturating_mul(2).min(MRF_RETRY_MAX_DELAY); + continue; + } + }; + match ReplicationConfigStore::read_no_lock(storage.clone(), ReplicationMetadataStore::MRF_REPLICATION_FILE).await { + Err(EcstoreError::ConfigNotFound) => return, + Ok(current) if current != data => return, + Ok(_) => { + match ReplicationConfigStore::save_no_lock( + storage.clone(), + ReplicationMetadataStore::MRF_REPLICATION_FILE, + Vec::new(), + ) + .await + { + Ok(()) => return, + Err(error) => warn!( + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION, + error = %error, + "Failed to clear the corrupt MRF recovery path after quarantine; retrying" + ), + } + } + Err(error) => warn!( + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION, + error = %error, + "Failed to verify the corrupt MRF recovery path before clearing; retrying" + ), } - Err(error) => warn!( - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_REPLICATION, - file = %quarantine_file, - error = %error, - "Failed to quarantine corrupt MRF recovery file; original was preserved" - ), + drop(_guard); + tokio::time::sleep(retry_delay).await; + retry_delay = retry_delay.saturating_mul(2).min(MRF_RETRY_MAX_DELAY); } } @@ -2156,39 +2287,193 @@ async fn flush_mrf_to_disk(entries: &[MrfReplicateEntry] } } -async fn append_mrf_entries_to_disk( +async fn recover_corrupt_mrf_generation( + corrupt_generation: &[u8], + entries_to_append: &[MrfReplicateEntry], + known_pending: &[MrfReplicateEntry], + storage: &Arc, + pending_payload: &mut Option, + started: Instant, +) -> Option { + let quarantine_file = format!("{MRF_CORRUPT_FILE_PREFIX}.{}.bin", OffsetDateTime::now_utc().unix_timestamp_nanos()); + if let Err(error) = ReplicationConfigStore::save_no_lock(storage.clone(), &quarantine_file, corrupt_generation.to_vec()).await + { + warn!( + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION, + file = %quarantine_file, + error = %error, + "Failed to quarantine a corrupt active MRF generation before rebuilding it" + ); + return None; + } + + let mut entries = Vec::with_capacity(known_pending.len().saturating_add(entries_to_append.len())); + entries.extend_from_slice(known_pending); + entries.extend_from_slice(entries_to_append); + let data = match encode_mrf_file(&entries) { + Ok(data) => data, + Err(error) => { + observe_mrf_flush_failure(0); + warn!( + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION, + count = entries.len(), + error = %error, + "Failed to rebuild the active MRF generation after quarantining corruption" + ); + return None; + } + }; + *pending_payload = Some(PendingMrfAppend { + digest: mrf_payload_digest(&data), + entry_count: entries.len(), + }); + if let Err(error) = + ReplicationConfigStore::save_no_lock(storage.clone(), ReplicationMetadataStore::MRF_REPLICATION_FILE, data).await + { + observe_mrf_flush_failure(duration_millis_u64(started.elapsed())); + warn!( + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION, + count = entries.len(), + error = %error, + "Failed to replace the active MRF generation after quarantining corruption" + ); + return None; + } + Some(duration_millis_u64(started.elapsed())) +} + +async fn append_mrf_entries_to_disk( entries_to_append: &[MrfReplicateEntry], storage: &Arc, + pending_payload: &mut Option, + known_pending: &[MrfReplicateEntry], ) -> Option { if entries_to_append.is_empty() { return Some(0); } - let mut entries = match ReplicationConfigStore::read(storage.clone(), ReplicationMetadataStore::MRF_REPLICATION_FILE).await { - Ok(data) => match decode_mrf_file(&data) { - Ok(entries) => entries, - Err(error) => { - warn!( - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_REPLICATION, - error = %error, - "Failed to decode MRF backlog before appending a capped entry" - ); - return None; - } - }, - Err(EcstoreError::ConfigNotFound) => Vec::new(), + let started = Instant::now(); + let lock = match storage + .new_ns_lock( + ReplicationMetadataStore::rustfs_meta_bucket(), + ReplicationMetadataStore::MRF_REPLICATION_FILE, + ) + .await + { + Ok(lock) => lock, Err(error) => { warn!( component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_REPLICATION, error = %error, - "Failed to read MRF backlog before appending a capped entry" + "Failed to acquire the MRF lock before appending a capped entry" ); return None; } }; + let _guard = match lock.get_write_lock(ReplicationLockTiming::acquire_timeout()).await { + Ok(guard) => guard, + Err(error) => { + warn!( + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION, + error = %error, + "Failed to acquire the MRF write lock before appending a capped entry" + ); + return None; + } + }; + + let current = + match ReplicationConfigStore::read_no_lock(storage.clone(), ReplicationMetadataStore::MRF_REPLICATION_FILE).await { + Ok(data) => data, + Err(EcstoreError::ConfigNotFound) => Vec::new(), + Err(error) => { + warn!( + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION, + error = %error, + "Failed to read MRF backlog before appending a capped entry" + ); + return None; + } + }; + + if pending_payload + .as_ref() + .is_some_and(|pending| pending.digest == mrf_payload_digest(¤t)) + { + return Some(duration_millis_u64(started.elapsed())); + } + + let mut entries = match decode_mrf_file(¤t) { + Ok(entries) => entries, + Err(_) if current.is_empty() => Vec::new(), + Err(error) => { + warn!( + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION, + error = %error, + "Failed to decode MRF backlog before appending a capped entry" + ); + return recover_corrupt_mrf_generation(¤t, entries_to_append, known_pending, storage, pending_payload, started) + .await; + } + }; + if let Some(pending) = pending_payload.as_ref() + && entries.len() >= pending.entry_count + { + match encode_mrf_file(&entries[..pending.entry_count]) { + Ok(prefix) if mrf_payload_digest(&prefix) == pending.digest => return Some(duration_millis_u64(started.elapsed())), + Ok(_) => {} + Err(error) => { + observe_mrf_flush_failure(0); + warn!( + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION, + error = %error, + "Failed to verify a capped MRF append after an ambiguous save error" + ); + return None; + } + } + } entries.extend_from_slice(entries_to_append); - flush_mrf_to_disk(&entries, storage).await + let data = match encode_mrf_file(&entries) { + Ok(data) => data, + Err(error) => { + observe_mrf_flush_failure(0); + warn!( + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION, + count = entries.len(), + error = %error, + "Failed to encode capped MRF entries for disk append" + ); + return None; + } + }; + *pending_payload = Some(PendingMrfAppend { + digest: mrf_payload_digest(&data), + entry_count: entries.len(), + }); + if let Err(error) = + ReplicationConfigStore::save_no_lock(storage.clone(), ReplicationMetadataStore::MRF_REPLICATION_FILE, data).await + { + let duration_millis = duration_millis_u64(started.elapsed()); + observe_mrf_flush_failure(duration_millis); + warn!( + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION, + count = entries.len(), + error = %error, + "Failed to append capped MRF entries to disk" + ); + return None; + } + Some(duration_millis_u64(started.elapsed())) } fn duration_millis_u64(duration: std::time::Duration) -> u64 { @@ -2637,9 +2922,12 @@ mod tests { lock_manager: Arc, first_read_started: Notify, delay_first_read: AtomicBool, + hold_first_read: AtomicBool, + allow_first_read: Notify, read_count: AtomicUsize, write_count: AtomicUsize, fail_next_write: AtomicBool, + fail_after_write: AtomicBool, block_next_write: AtomicBool, write_started: Notify, allow_write: Notify, @@ -2693,7 +2981,11 @@ mod tests { let read_index = self.shared.read_count.fetch_add(1, Ordering::SeqCst); if read_index == 0 && self.shared.delay_first_read.load(Ordering::SeqCst) { self.shared.first_read_started.notify_waiters(); - tokio::time::sleep(Duration::from_millis(1_500)).await; + if self.shared.hold_first_read.load(Ordering::SeqCst) { + self.shared.allow_first_read.notified().await; + } else { + tokio::time::sleep(Duration::from_millis(1_500)).await; + } } let data = self @@ -2777,9 +3069,14 @@ mod tests { .lock() .expect("test writes lock should not be poisoned") .push((object.to_string(), encoded.clone())); - *self.shared.data.lock().expect("test data lock should not be poisoned") = encoded; - self.shared.etag_revision.fetch_add(1, Ordering::SeqCst); + if !object.starts_with(MRF_CORRUPT_FILE_PREFIX) { + *self.shared.data.lock().expect("test data lock should not be poisoned") = encoded; + self.shared.etag_revision.fetch_add(1, Ordering::SeqCst); + } self.shared.write_count.fetch_add(1, Ordering::SeqCst); + if self.shared.fail_after_write.swap(false, Ordering::SeqCst) { + return Err(EcstoreError::Unexpected); + } Ok(ObjectInfo::default()) } } @@ -3240,9 +3537,12 @@ mod tests { lock_manager: Arc::new(rustfs_lock::GlobalLockManager::new()), first_read_started: Notify::new(), delay_first_read: AtomicBool::new(false), + hold_first_read: AtomicBool::new(false), + allow_first_read: Notify::new(), read_count: AtomicUsize::new(0), write_count: AtomicUsize::new(0), fail_next_write: AtomicBool::new(false), + fail_after_write: AtomicBool::new(false), block_next_write: AtomicBool::new(false), write_started: Notify::new(), allow_write: Notify::new(), @@ -3892,9 +4192,12 @@ mod tests { lock_manager: Arc::new(rustfs_lock::GlobalLockManager::new()), first_read_started: Notify::new(), delay_first_read: AtomicBool::new(true), + hold_first_read: AtomicBool::new(false), + allow_first_read: Notify::new(), read_count: AtomicUsize::new(0), write_count: AtomicUsize::new(0), fail_next_write: AtomicBool::new(false), + fail_after_write: AtomicBool::new(false), block_next_write: AtomicBool::new(false), write_started: Notify::new(), allow_write: Notify::new(), @@ -4023,6 +4326,275 @@ mod tests { assert!(marker.1.is_empty(), "the active MRF path should be marked absent"); } + #[tokio::test] + async fn corrupt_mrf_quarantine_preserves_a_concurrently_replaced_generation() { + let shared = empty_resync_shared_state(); + let corrupt = vec![0xde, 0xad, 0xbe, 0xef]; + let replacement = encode_mrf_file(&[MrfReplicateEntry { + bucket: "mrf-replacement".to_string(), + object: "new-generation".to_string(), + op: MrfOpKind::Object, + ..Default::default() + }]) + .expect("replacement MRF generation should encode"); + *shared.data.lock().expect("test data lock should not be poisoned") = corrupt; + shared.block_next_write.store(true, Ordering::SeqCst); + let pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("mrf-corrupt-race", shared.clone()))).await; + let write_started = shared.write_started.notified(); + + pool.start_mrf_processor().await; + tokio::time::timeout(Duration::from_secs(2), write_started) + .await + .expect("quarantine write should block before its payload is persisted"); + *shared.data.lock().expect("test data lock should not be poisoned") = replacement.clone(); + shared.allow_write.notify_one(); + + let handle = pool + .task_handles + .lock() + .await + .pop() + .expect("MRF processor task should be registered"); + handle.await.expect("MRF processor should not panic"); + + assert_eq!( + *shared.data.lock().expect("test data lock should not be poisoned"), + replacement, + "quarantine cleanup must not clear a newer active MRF generation" + ); + assert!( + !shared + .writes + .lock() + .expect("test writes lock should not be poisoned") + .iter() + .any(|(file, data)| file == ReplicationMetadataStore::MRF_REPLICATION_FILE && data.is_empty()), + "the newer active generation must not be replaced with the empty marker" + ); + } + + #[tokio::test] + async fn corrupt_mrf_quarantine_retries_without_blocking_new_failures() { + temp_env::async_with_vars([("RUSTFS_REPL_MRF_FLUSH_INTERVAL_MS", Some("10"))], async { + let shared = empty_resync_shared_state(); + shared.fail_next_write.store(true, Ordering::SeqCst); + *shared.data.lock().expect("test data lock should not be poisoned") = vec![0xde, 0xad, 0xbe, 0xef]; + let pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("mrf-corrupt-retry", shared.clone()))).await; + + pool.start_mrf_persister().await; + pool.start_mrf_processor().await; + pool.mrf_save_tx + .send(MrfReplicateEntry { + bucket: "mrf-corrupt-retry".to_string(), + object: "new-failure".to_string(), + op: MrfOpKind::Object, + ..Default::default() + }) + .await + .expect("new failure should be staged during corrupt-file recovery"); + + pool.mrf_save_tx + .send(MrfReplicateEntry { + bucket: "mrf-corrupt-retry".to_string(), + object: "second-failure".to_string(), + op: MrfOpKind::Object, + ..Default::default() + }) + .await + .expect("persister should drain the first staged failure before recovery completes"); + let processor_handle = pool + .task_handles + .lock() + .await + .pop() + .expect("MRF processor task should be registered"); + processor_handle.await.expect("MRF processor should retry quarantine writes"); + + tokio::time::timeout(Duration::from_secs(2), async { + loop { + let quarantine_complete = { + let writes = shared.writes.lock().expect("test writes lock should not be poisoned"); + let quarantined = writes + .iter() + .any(|(file, data)| file.starts_with(MRF_CORRUPT_FILE_PREFIX) && data == &[0xde, 0xad, 0xbe, 0xef]); + let cleared = writes + .iter() + .any(|(file, data)| file == ReplicationMetadataStore::MRF_REPLICATION_FILE && data.is_empty()); + quarantined && cleared + }; + if quarantine_complete { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("quarantine should retry after the injected write failure and clear the active path"); + + tokio::time::timeout(Duration::from_secs(2), async { + loop { + let data = shared.data.lock().expect("test data lock should not be poisoned").clone(); + if decode_mrf_file(&data).is_ok_and(|entries| entries.iter().any(|entry| entry.object == "new-failure")) { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("staged failures should flush after quarantine recovery"); + + let persister_handle = pool + .task_handles + .lock() + .await + .pop() + .expect("MRF persister task should be registered"); + persister_handle.abort(); + }) + .await; + } + + #[tokio::test] + async fn mrf_startup_staging_does_not_publish_entries_before_flush() { + temp_env::async_with_vars([("RUSTFS_REPL_MRF_FLUSH_INTERVAL_MS", Some("10"))], async { + let shared = empty_resync_shared_state(); + *shared.data.lock().expect("test data lock should not be poisoned") = encode_mrf_file(&[MrfReplicateEntry { + bucket: "mrf-durable-seed".to_string(), + object: "seed".to_string(), + size: 7, + op: MrfOpKind::Object, + ..Default::default() + }]) + .expect("seed MRF backlog should encode"); + shared.delay_first_read.store(true, Ordering::SeqCst); + shared.block_next_write.store(true, Ordering::SeqCst); + let pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("mrf-durable-staged", shared.clone()))).await; + let read_started = shared.first_read_started.notified(); + let write_started = shared.write_started.notified(); + + pool.start_mrf_persister().await; + tokio::time::timeout(Duration::from_secs(2), read_started) + .await + .expect("startup MRF read should be delayed"); + pool.mrf_save_tx + .send(MrfReplicateEntry { + bucket: "mrf-not-yet-durable".to_string(), + object: "staged".to_string(), + size: 11, + op: MrfOpKind::Object, + ..Default::default() + }) + .await + .expect("staged MRF entry should be accepted"); + tokio::time::sleep(Duration::from_millis(100)).await; + assert_eq!( + shared.write_count.load(Ordering::SeqCst), + 0, + "staged entries must not flush before startup recovery is applied" + ); + *pool.mrf_recovery_result.lock().await = Some(Vec::new()); + pool.mrf_recovery_complete.notify_one(); + tokio::time::timeout(Duration::from_secs(3), write_started) + .await + .expect("first flush should block after startup recovery is applied"); + + let snapshot = durable_mrf_backlog_summary_snapshot(); + assert_eq!( + snapshot.buckets.iter().find(|bucket| bucket.bucket == "mrf-durable-seed"), + Some(&DurableMrfBucketBacklog { + bucket: "mrf-durable-seed".to_string(), + count: 1, + bytes: 7, + }) + ); + assert!( + snapshot.buckets.iter().all(|bucket| bucket.bucket != "mrf-not-yet-durable"), + "staged entries must not appear in durable metrics before the first successful flush" + ); + + let handle = pool + .task_handles + .lock() + .await + .pop() + .expect("MRF persister task should be registered"); + handle.abort(); + let _ = handle.await; + set_durable_mrf_backlog_snapshot(DurableMrfBacklogSnapshot::default()); + }) + .await; + } + + #[tokio::test] + async fn mrf_persister_does_not_eager_flush_before_recovery_snapshot() { + temp_env::async_with_vars([("RUSTFS_REPL_MRF_FLUSH_INTERVAL_MS", Some("10"))], async { + let shared = empty_resync_shared_state(); + *shared.data.lock().expect("test data lock should not be poisoned") = + encode_mrf_file(&[MrfReplicateEntry::default()]).expect("seed MRF backlog should encode"); + shared.delay_first_read.store(true, Ordering::SeqCst); + shared.hold_first_read.store(true, Ordering::SeqCst); + let pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("mrf-recovery-gate", shared.clone()))).await; + + pool.start_mrf_processor().await; + tokio::time::timeout(Duration::from_secs(2), shared.first_read_started.notified()) + .await + .expect("processor startup read should be delayed"); + pool.start_mrf_persister().await; + tokio::time::timeout(Duration::from_secs(2), async { + loop { + if shared.read_count.load(Ordering::SeqCst) >= 2 { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("persister should finish its startup snapshot while processor is delayed"); + + for index in 0..1000 { + pool.mrf_save_tx + .send(MrfReplicateEntry { + bucket: "mrf-recovery-gate".to_string(), + object: format!("staged-{index}"), + op: MrfOpKind::Object, + ..Default::default() + }) + .await + .expect("failure should be accepted before recovery completes"); + } + assert_eq!( + shared.write_count.load(Ordering::SeqCst), + 0, + "the eager 1,000-entry threshold must not flush before recovery snapshot completion" + ); + + shared.allow_first_read.notify_one(); + let processor_handle = pool.task_handles.lock().await.remove(0); + processor_handle.await.expect("processor recovery should complete"); + + tokio::time::timeout(Duration::from_secs(3), async { + loop { + let data = shared.data.lock().expect("test data lock should not be poisoned").clone(); + if decode_mrf_file(&data).is_ok_and(|entries| entries.len() == 1001) { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("staged failures should flush after recovery snapshot completion"); + + let persister_handle = pool + .task_handles + .lock() + .await + .pop() + .expect("MRF persister task should be registered"); + persister_handle.abort(); + }) + .await; + } + #[tokio::test] async fn mrf_persister_seeds_retained_startup_entries() { assert!( @@ -4094,6 +4666,372 @@ mod tests { .await; } + #[tokio::test] + async fn mrf_capped_append_retries_and_retains_existing_backlog() { + assert!( + runtime_sources::replication_pool().is_none(), + "test requires the runtime replication pool to be unavailable" + ); + temp_env::async_with_vars([("RUSTFS_REPL_MRF_FLUSH_INTERVAL_MS", Some("10"))], async { + let shared = empty_resync_shared_state(); + let retained = (0..MRF_PENDING_CAP) + .map(|index| MrfReplicateEntry { + bucket: "mrf-capped-retry".to_string(), + object: format!("retained-{index}"), + op: MrfOpKind::Object, + ..Default::default() + }) + .collect::>(); + *shared.data.lock().expect("test data lock should not be poisoned") = + encode_mrf_file(&retained).expect("MRF backlog should encode"); + let pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("mrf-capped-retry", shared.clone()))).await; + + pool.start_mrf_persister().await; + pool.start_mrf_processor().await; + let processor_handle = pool + .task_handles + .lock() + .await + .pop() + .expect("MRF processor task should be registered"); + processor_handle.await.expect("MRF processor should not panic"); + + tokio::time::timeout(Duration::from_secs(30), async { + loop { + if shared.write_count.load(Ordering::SeqCst) > 0 { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("startup backlog should be flushed before appending a capped batch"); + shared.fail_next_write.store(true, Ordering::SeqCst); + pool.mrf_save_tx + .send(MrfReplicateEntry { + bucket: "mrf-capped-retry".to_string(), + object: "new-capped-failure".to_string(), + op: MrfOpKind::Object, + ..Default::default() + }) + .await + .expect("new capped failure should be accepted"); + + tokio::time::timeout(Duration::from_secs(30), async { + loop { + let data = shared.data.lock().expect("test data lock should not be poisoned").clone(); + if decode_mrf_file(&data).is_ok_and(|entries| { + entries.len() == MRF_PENDING_CAP + 1 + && entries.first().is_some_and(|entry| entry.object == "retained-0") + && entries.last().is_some_and(|entry| entry.object == "new-capped-failure") + }) { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("a failed capped append should be retried without dropping either batch"); + + let persister_handle = pool + .task_handles + .lock() + .await + .pop() + .expect("MRF persister task should be registered"); + persister_handle.abort(); + }) + .await; + } + + #[tokio::test] + async fn mrf_capped_append_recognizes_a_post_commit_error_after_a_concurrent_append() { + let shared = empty_resync_shared_state(); + let initial = MrfReplicateEntry { + bucket: "mrf-append-idempotency".to_string(), + object: "retained".to_string(), + op: MrfOpKind::Object, + ..Default::default() + }; + let appended = MrfReplicateEntry { + object: "new-failure".to_string(), + ..initial.clone() + }; + *shared.data.lock().expect("test data lock should not be poisoned") = + encode_mrf_file(std::slice::from_ref(&initial)).expect("initial MRF backlog should encode"); + shared.fail_after_write.store(true, Ordering::SeqCst); + let storage = Arc::new(LoadResyncNodeStore::new("mrf-append-idempotency", shared.clone())); + let mut pending_payload = None; + + assert_eq!( + append_mrf_entries_to_disk(std::slice::from_ref(&appended), &storage, &mut pending_payload, &[]).await, + None, + "the injected post-commit error should leave the capped batch pending" + ); + let concurrent = MrfReplicateEntry { + object: "concurrent-failure".to_string(), + ..initial.clone() + }; + let mut concurrent_payload = None; + assert!( + append_mrf_entries_to_disk(std::slice::from_ref(&concurrent), &storage, &mut concurrent_payload, &[]) + .await + .is_some(), + "a concurrent node should be able to append after the ambiguous save" + ); + assert!( + append_mrf_entries_to_disk(std::slice::from_ref(&appended), &storage, &mut pending_payload, &[]) + .await + .is_some(), + "retry should recognize the already-committed payload" + ); + + let entries = decode_mrf_file(&shared.data.lock().expect("test data lock should not be poisoned")) + .expect("persisted MRF backlog should decode"); + assert_eq!(entries.len(), 3); + assert_eq!(entries[0].bucket, initial.bucket); + assert_eq!(entries[0].object, initial.object); + assert_eq!(entries[1].bucket, appended.bucket); + assert_eq!(entries[1].object, appended.object); + assert_eq!(entries[2].bucket, concurrent.bucket); + assert_eq!(entries[2].object, concurrent.object); + assert_eq!( + shared.write_count.load(Ordering::SeqCst), + 2, + "retry must not write the appended MRF batch twice" + ); + } + + #[tokio::test] + async fn mrf_capped_append_recovers_a_late_corrupt_generation() { + let shared = empty_resync_shared_state(); + let retained = MrfReplicateEntry { + bucket: "mrf-late-corruption".to_string(), + object: "retained".to_string(), + op: MrfOpKind::Object, + ..Default::default() + }; + let appended = MrfReplicateEntry { + bucket: retained.bucket.clone(), + object: "appended-after-corruption".to_string(), + op: MrfOpKind::Object, + ..Default::default() + }; + *shared.data.lock().expect("test data lock should not be poisoned") = vec![0xde, 0xad, 0xbe, 0xef]; + let storage = Arc::new(LoadResyncNodeStore::new("mrf-late-corruption", shared.clone())); + let mut pending_payload = None; + + assert!( + append_mrf_entries_to_disk( + std::slice::from_ref(&appended), + &storage, + &mut pending_payload, + std::slice::from_ref(&retained), + ) + .await + .is_some(), + "a corrupt active generation should be quarantined and rebuilt" + ); + + let writes = shared.writes.lock().expect("test writes lock should not be poisoned"); + assert!( + writes + .iter() + .any(|(file, data)| file.starts_with(MRF_CORRUPT_FILE_PREFIX) && data == &[0xde, 0xad, 0xbe, 0xef]), + "the corrupt active generation should be retained in quarantine" + ); + let recovered = decode_mrf_file(&shared.data.lock().expect("test data lock should not be poisoned")) + .expect("the active MRF generation should be rebuilt"); + assert_eq!(recovered.len(), 2); + assert_eq!(recovered[0].object, retained.object); + assert_eq!(recovered[1].object, appended.object); + } + + #[tokio::test] + async fn mrf_persister_appends_startup_staging_overflow_after_recovery() { + assert!( + runtime_sources::replication_pool().is_none(), + "test requires the runtime replication pool to be unavailable" + ); + temp_env::async_with_vars([("RUSTFS_REPL_MRF_FLUSH_INTERVAL_MS", Some("10"))], async { + let shared = empty_resync_shared_state(); + let retained = vec![MrfReplicateEntry::default(); MRF_PENDING_CAP]; + *shared.data.lock().expect("test data lock should not be poisoned") = + encode_mrf_file(&retained).expect("full startup MRF backlog should encode"); + shared.delay_first_read.store(true, Ordering::SeqCst); + let pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("mrf-staged-overflow", shared.clone()))).await; + let read_started = shared.first_read_started.notified(); + + pool.start_mrf_persister().await; + tokio::time::timeout(Duration::from_secs(2), read_started) + .await + .expect("persister startup read should be delayed while staging the overflow entry"); + pool.mrf_save_tx + .send(MrfReplicateEntry { + bucket: "mrf-staged-overflow".to_string(), + object: "staged-overflow".to_string(), + op: MrfOpKind::Object, + ..Default::default() + }) + .await + .expect("overflow entry should be staged before startup recovery completes"); + *pool.mrf_recovery_result.lock().await = Some(Vec::new()); + pool.mrf_recovery_complete.notify_one(); + + tokio::time::timeout(Duration::from_secs(30), async { + loop { + let data = shared.data.lock().expect("test data lock should not be poisoned").clone(); + if decode_mrf_file(&data).is_ok_and(|entries| { + entries.len() == MRF_PENDING_CAP + 1 + && entries.last().is_some_and(|entry| entry.object == "staged-overflow") + }) { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("persister should append staged overflow through its capped recovery path"); + + let handle = pool + .task_handles + .lock() + .await + .pop() + .expect("MRF persister task should be registered"); + handle.abort(); + }) + .await; + } + + #[tokio::test] + async fn mrf_recovery_shrink_refills_pending_and_flushes_before_shutdown() { + assert!( + runtime_sources::replication_pool().is_none(), + "test requires the runtime replication pool to be unavailable" + ); + temp_env::async_with_vars([("RUSTFS_REPL_MRF_FLUSH_INTERVAL_MS", Some("60000"))], async { + let shared = empty_resync_shared_state(); + let retained = vec![MrfReplicateEntry::default(); MRF_PENDING_CAP]; + *shared.data.lock().expect("test data lock should not be poisoned") = + encode_mrf_file(&retained).expect("full startup MRF backlog should encode"); + shared.delay_first_read.store(true, Ordering::SeqCst); + shared.block_next_write.store(true, Ordering::SeqCst); + let pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("mrf-recovery-shrink", shared.clone()))).await; + let read_started = shared.first_read_started.notified(); + let write_started = shared.write_started.notified(); + + pool.start_mrf_persister().await; + tokio::time::timeout(Duration::from_secs(2), read_started) + .await + .expect("startup MRF read should be delayed"); + for object in ["staged-overflow-1", "staged-overflow-2", "staged-overflow-3"] { + pool.mrf_save_tx + .send(MrfReplicateEntry { + bucket: "mrf-recovery-shrink".to_string(), + object: object.to_string(), + op: MrfOpKind::Object, + ..Default::default() + }) + .await + .expect("overflow entry should be staged before recovery completes"); + } + *pool.mrf_recovery_result.lock().await = Some(vec![MrfReplicateEntry::default(); MRF_PENDING_CAP - 2]); + pool.mrf_recovery_complete.notify_one(); + + tokio::time::timeout(Duration::from_secs(5), write_started) + .await + .expect("recovery shrink should trigger the normal pending flush before shutdown"); + assert!( + decode_mrf_file(&shared.data.lock().expect("test data lock should not be poisoned")) + .expect("the blocked write should leave the old backlog readable") + .iter() + .all(|entry| !entry.object.starts_with("staged-overflow-")), + "the staged overflow must remain pending until the flush completes" + ); + shared.allow_write.notify_one(); + + tokio::time::timeout(Duration::from_secs(30), async { + loop { + if shared.write_count.load(Ordering::SeqCst) >= 2 { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("the recovered pending prefix should persist before the task is stopped"); + + let persisted = shared + .writes + .lock() + .expect("test writes lock should not be poisoned") + .iter() + .rev() + .find(|(file, _)| file == ReplicationMetadataStore::MRF_REPLICATION_FILE) + .map(|(_, data)| decode_mrf_file(data).expect("persisted MRF data should decode")) + .expect("the capped suffix should be persisted after the pending flush"); + assert_eq!(persisted.len(), MRF_PENDING_CAP + 1); + assert_eq!(persisted[MRF_PENDING_CAP - 1].object, "staged-overflow-1"); + assert_eq!(persisted[MRF_PENDING_CAP].object, "staged-overflow-2"); + assert_eq!(persisted.last().expect("capped suffix should be present").object, "staged-overflow-3"); + + let handle = pool + .task_handles + .lock() + .await + .pop() + .expect("MRF persister task should be registered"); + handle.abort(); + }) + .await; + } + + #[test] + fn mrf_durable_tracker_flushes_only_the_new_suffix() { + let retained = MrfReplicateEntry { + bucket: "mrf-tracker-suffix".to_string(), + size: 3, + ..Default::default() + }; + let appended = MrfReplicateEntry { + bucket: retained.bucket.clone(), + size: 5, + ..Default::default() + }; + let pending = [retained.clone(), appended]; + let mut tracker = durable_mrf_backlog_tracker_from_entries(std::slice::from_ref(&retained)); + + add_durable_mrf_suffix(&mut tracker, &pending, 1); + + let snapshot = tracker.into_snapshot(); + assert_eq!(snapshot.summary.buckets.len(), 1); + assert_eq!(snapshot.summary.buckets[0].count, 2); + assert_eq!(snapshot.summary.buckets[0].bytes, 8); + } + + #[test] + fn move_staged_mrf_entries_releases_staging_capacity() { + let mut pending = vec![MrfReplicateEntry::default(); MRF_PENDING_CAP - 1]; + let mut staged = Vec::with_capacity(2); + staged.push(MrfReplicateEntry { + object: "pending-entry".to_string(), + ..Default::default() + }); + staged.push(MrfReplicateEntry { + object: "capped-entry".to_string(), + ..Default::default() + }); + + let capped_batch = move_staged_mrf_entries(&mut pending, &mut staged); + + assert_eq!(pending.len(), MRF_PENDING_CAP); + assert_eq!(pending.last().expect("pending prefix should be filled").object, "pending-entry"); + assert_eq!(capped_batch.len(), 1); + assert_eq!(capped_batch[0].object, "capped-entry"); + assert_eq!(staged.capacity(), 0, "the staging allocation should be released after the move"); + } + #[tokio::test] async fn mrf_delete_replay_result_is_retained_when_runtime_pool_is_unavailable() { assert!(