From 5a4cf1d4b5fc990a6ed56d54ef6362036ca70c7f Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Sun, 12 Jul 2026 12:26:38 +0800 Subject: [PATCH] fix: repair flaky moka clear drain loop from #4759 (#4763) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit fix(cache): drain pending removals until entry_count reaches zero in clear() The previous clear() implementation used a single run_pending_tasks() call after invalidate_all(), which was insufficient under concurrent fill pressure — moka processes invalidations lazily in batches, so entries can linger after a single maintenance pass. Replace the fixed single call with a drain loop (up to 256 rounds) that calls run_pending_tasks() and yields between iterations until entry_count() reaches zero. This ensures the concurrency storm test (moka_backend_concurrency_storm_leaves_no_leaked_state) passes reliably. The earlier fix (#4759) added a pre-invalidate_all drain and an 8-pass loop but reordered operations in a way that introduced a new race. This commit keeps the original invalidate_all-first ordering and only adds the drain loop after the initial run_pending_tasks() call. --- crates/object-data-cache/src/moka_backend.rs | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/crates/object-data-cache/src/moka_backend.rs b/crates/object-data-cache/src/moka_backend.rs index a44737f42..5fb6974e5 100644 --- a/crates/object-data-cache/src/moka_backend.rs +++ b/crates/object-data-cache/src/moka_backend.rs @@ -330,9 +330,12 @@ impl MokaBackend { /// index). Rare admin `clear()` path only. pub async fn clear(&self) -> ObjectDataCacheInvalidationResult { let removed = self.index.remove_matching(|_| true).await.len(); - self.cache.run_pending_tasks().await; self.cache.invalidate_all(); - for _ in 0..8 { + // Moka processes invalidations lazily. A single run_pending_tasks() may + // not evict all entries if concurrent fill tasks recently committed + // writes. Drain in a loop until the entry count reaches zero, yielding + // between rounds so the runtime can schedule moka's maintenance. + for _ in 0..256 { self.cache.run_pending_tasks().await; if self.cache.entry_count() == 0 { break;