From f33ae8e9afc0c67292d209e0c500c4f2faf239de Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Mon, 6 Jul 2026 21:34:45 +0800 Subject: [PATCH] fix(replication): keep MRF persister backlog cumulative so flushes don't drop entries (backlog#859) (#4314) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The MRF persister accumulated overflow entries in `pending`, flushed them with `flush_mrf_to_disk`, and cleared `pending` on success. But `flush_mrf_to_disk` *overwrites* the whole MRF file with exactly the entries passed. After flushing batch A (file = A) and clearing, the next flush wrote batch B and thereby overwrote the file to contain only B — and the MRF file is only replayed (and cleared) at startup, never during the run, so batch A's entries were silently lost. A crash after the B flush lost all of batch A's pending replications. Keep `pending` cumulative (the file must hold the full set of overflow entries for the run) and rewrite the whole set on each flush instead of clearing after success: - flush eagerly once 1 000 *new* entries accumulate since the last write (measured against the flushed length, so a large backlog isn't rewritten on every add), and on the 10s tick when dirty; - bound the in-memory/on-disk backlog with `MRF_PENDING_CAP` (200 000) and log once when the cap is hit rather than growing without limit. Refs backlog#799 (B10). --- .../bucket/replication/replication_pool.rs | 48 +++++++++++++++---- 1 file changed, 40 insertions(+), 8 deletions(-) diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index 88c628276..224d97f10 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/crates/ecstore/src/bucket/replication/replication_pool.rs @@ -693,9 +693,10 @@ impl ReplicationPool { /// Starts the MRF persister — ongoing background task. /// /// Drains `mrf_save_rx` (entries that overflowed the normal worker channels) and - /// writes them to the on-disk MRF file every 10 seconds or when 1 000 entries - /// accumulate. The file is overwritten (not appended) on each flush so it always - /// reflects the current pending backlog. + /// writes them to the on-disk MRF file every 10 seconds or when 1 000 new + /// entries accumulate. Each flush rewrites the whole cumulative backlog so no + /// previously-persisted (and not-yet-replayed) entry is lost; the file is only + /// consumed and cleared at startup. async fn start_mrf_persister(&self) { let Some(mut rx) = self.mrf_save_rx.lock().await.take() else { return; @@ -703,7 +704,19 @@ impl ReplicationPool { let storage = self.storage.clone(); let handle = tokio::spawn(async move { + // 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 + // written this run. `pending` is therefore kept cumulative and the + // whole set is rewritten on each flush — clearing it after a flush + // 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 pending: Vec = Vec::new(); + let mut flushed_len = 0usize; + let mut dirty = false; + let mut capped = false; let mut interval = tokio::time::interval(Duration::from_secs(10)); interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); @@ -711,22 +724,41 @@ impl ReplicationPool { tokio::select! { entry = rx.recv() => match entry { Some(e) => { + if pending.len() >= MRF_PENDING_CAP { + if !capped { + capped = true; + warn!( + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION, + cap = MRF_PENDING_CAP, + "MRF pending backlog hit cap — dropping further recovery entries for this run" + ); + } + continue; + } pending.push(e); - if pending.len() >= 1000 && flush_mrf_to_disk(&pending, &storage).await { - pending.clear(); + dirty = true; + // Flush eagerly once enough new entries have accumulated + // 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 pending.len() - flushed_len >= 1000 && flush_mrf_to_disk(&pending, &storage).await { + flushed_len = pending.len(); + dirty = false; } } None => { // Channel closed (pool shutting down) — final flush. - if !pending.is_empty() { + if dirty { flush_mrf_to_disk(&pending, &storage).await; } break; } }, _ = interval.tick() => { - if !pending.is_empty() && flush_mrf_to_disk(&pending, &storage).await { - pending.clear(); + if dirty && flush_mrf_to_disk(&pending, &storage).await { + flushed_len = pending.len(); + dirty = false; } } }