From 93c89ef13201211a32a39e4bcbd85284e188ab6f Mon Sep 17 00:00:00 2001 From: overtrue Date: Sun, 6 Sep 2026 03:20:33 +0800 Subject: [PATCH] test(ecstore): expose stale fsync group cleanup --- crates/ecstore/src/disk/os.rs | 114 ++++++++++++++++++++++++++++++++++ 1 file changed, 114 insertions(+) diff --git a/crates/ecstore/src/disk/os.rs b/crates/ecstore/src/disk/os.rs index f25ce50ba..e1e7044c6 100644 --- a/crates/ecstore/src/disk/os.rs +++ b/crates/ecstore/src/disk/os.rs @@ -6479,6 +6479,120 @@ mod tests { ); } + #[cfg(unix)] + #[test] + fn stale_idle_group_cleanup_preserves_successor_registration() { + let temp_dir = tempdir().expect("fixture directory"); + let registry = DstDirFsyncGroupCommit::default(); + let (mut first_rx, first_worker) = registry.enqueue_for_test(temp_dir.path()).expect("enqueue first worker"); + let old_group = first_worker.expect("first waiter starts a worker"); + // W1 has completed its batch and marked G idle, but has not cleaned G up. + let first_waiter = old_group.inner.lock().pending.pop_front().expect("first batch waiter"); + registry.complete_batch(1); + old_group.inner.lock().worker_running = false; + + let (mut second_rx, second_worker) = registry.enqueue_for_test(temp_dir.path()).expect("enqueue second worker"); + let reused_group = second_worker.expect("idle G starts another worker"); + assert!(Arc::ptr_eq(&old_group, &reused_group)); + let second_waiter = reused_group.inner.lock().pending.pop_front().expect("second batch waiter"); + registry.complete_batch(1); + reused_group.inner.lock().worker_running = false; + registry.remove_idle_group(&reused_group); + assert_eq!(registry.counts_for_test(), (0, 0), "normal idle cleanup must remove G"); + assert!(second_waiter.result_tx.send(Ok(())).is_ok()); + assert!(second_rx.try_recv().expect("second worker reports completion").is_ok()); + + let (mut successor_rx, successor_worker) = registry.enqueue_for_test(temp_dir.path()).expect("enqueue successor"); + let successor = successor_worker.expect("successor starts a new group"); + assert!(!Arc::ptr_eq(&old_group, &successor)); + assert_eq!(registry.counts_for_test(), (1, 1)); + // W1 resumes with its old Arc after W2 removed G and W3 installed G2. + registry.remove_idle_group(&old_group); + assert!(first_waiter.result_tx.send(Ok(())).is_ok()); + assert!(first_rx.try_recv().expect("first worker reports completion").is_ok()); + assert!( + registry + .inner + .lock() + .groups + .get(&successor.key) + .is_some_and(|registered| Arc::ptr_eq(registered, &successor)), + "stale cleanup must retain the exact successor Arc" + ); + assert_eq!(registry.counts_for_test(), (1, 1)); + assert!(successor.inner.lock().worker_running); + assert_eq!(successor.inner.lock().pending.len(), 1); + assert!(matches!(successor_rx.try_recv(), Err(oneshot::error::TryRecvError::Empty))); + + let (_joined_rx, new_worker) = registry.enqueue_for_test(temp_dir.path()).expect("join successor"); + assert!(new_worker.is_none(), "a later waiter must join G2 instead of creating G3"); + assert_eq!(successor.inner.lock().pending.len(), 2); + assert_eq!(registry.counts_for_test(), (1, 2)); + } + + #[cfg(unix)] + #[tokio::test] + #[serial_test::serial(dst_dir_fsync_group_commit)] + async fn stale_idle_cleanup_then_unpolled_worker_drop_releases_waiter_budget() { + wait_for_dst_dir_fsync_group_commit_idle().await; + let temp_dir = tempdir().expect("fixture directory"); + let (old_rx, old_worker) = DST_DIR_FSYNC_GROUP_COMMIT + .enqueue_for_test(temp_dir.path()) + .expect("enqueue old group"); + let old_group = old_worker.expect("old group starts a worker"); + tokio::time::timeout(Duration::from_secs(5), run_dst_dir_fsync_group_worker(old_group.clone())) + .await + .expect("old worker must finish its actual fsync"); + assert!(old_rx.await.expect("old worker reports completion").is_ok()); + assert!(fsync_dir_recorder::was_fsynced(temp_dir.path())); + assert_eq!(fsync_dir_recorder::grouped_batch_sizes(temp_dir.path()), vec![1]); + assert_eq!(DST_DIR_FSYNC_GROUP_COMMIT.counts_for_test(), (0, 0)); + + let ctx = Arc::new(crate::runtime::instance::InstanceContext::new()); + let owner = ctx.begin_namespace_commit(); + let probe = Arc::downgrade(&owner); + let generation = ctx.namespace_commit_generation(); + let (rx, successor_worker) = DST_DIR_FSYNC_GROUP_COMMIT + .enqueue_opened( + OpenedDstDirFsyncGroup::open(temp_dir.path()).expect("open successor directory"), + Some(owner), + ) + .expect("enqueue successor owner"); + let successor = successor_worker.expect("successor starts a new group"); + assert!(!Arc::ptr_eq(&old_group, &successor)); + let worker = run_dst_dir_fsync_group_worker(successor.clone()); + // The stale Arc represents W1 resuming after another worker removed G. + DST_DIR_FSYNC_GROUP_COMMIT.remove_idle_group(&old_group); + assert!(ctx.namespace_commits_pending()); + assert!(probe.upgrade().is_some()); + drop(worker); + let channel_closed = tokio::time::timeout(Duration::from_secs(5), rx) + .await + .expect("dropping the unpolled worker must release its channel") + .is_err(); + let counts_after_drop = DST_DIR_FSYNC_GROUP_COMMIT.counts_for_test(); + let owner_released = probe.upgrade().is_none(); + let namespace_pending = ctx.namespace_commits_pending(); + let generation_after_drop = ctx.namespace_commit_generation(); + let successor_pending = successor.inner.lock().pending.len(); + let worker_running = successor.inner.lock().worker_running; + // Preserve the observed result before cleanup, so a RED run cannot leak + // its phantom count into unrelated tests in the same process. + clear_dst_dir_fsync_group_commit_for_test(); + assert!(channel_closed); + assert!(owner_released); + assert!(!namespace_pending); + assert!(generation_after_drop > generation); + assert_eq!(successor_pending, 0); + assert!(!worker_running); + assert_eq!( + fsync_dir_recorder::grouped_batch_sizes(temp_dir.path()), + vec![1], + "dropping the successor before its first poll must not dispatch another fsync" + ); + assert_eq!(counts_after_drop, (0, 0), "stale cleanup must not strand a phantom waiter"); + } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] #[serial_test::serial(dst_dir_fsync_group_commit)] async fn dst_dir_fsync_group_commit_cancellation_releases_waiter_state() {