From be2e454d9d386ddf862c2dbf48a312b036001b9b Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Fri, 17 Jul 2026 08:31:41 +0800 Subject: [PATCH] test(ecstore): rename/commit fan-out pause barrier + background-task introspection (backlog#1325 block 2) (#4936) test(ecstore): add rename/commit fan-out pause barrier and background-task introspection Second white-box test-infra block for https://github.com/rustfs/backlog/issues/1325 (the first block landed the per-disk call counters in PR#4914). Adds a `#[cfg(test)]` awaitable pause barrier plus in-flight background-task introspection to the rename/commit fan-out in `crates/ecstore/src/set_disk/core/io_primitives.rs`, following the same dual-cfg seam style as the existing `disk_call_counters` and `cleanup_fault_injection` seams. A test arms a barrier for `(object, disk_index, phase)`; the matching spawned fan-out task parks at its checkpoint until the test releases it, and the test awaits the pause through a deterministic `tokio::sync::Notify` handshake (no sleeps). A separate object-keyed task tracker reports how many rename/cleanup background disk tasks are still in flight, so a test can assert "a background disk write is still running" while paused and "no background disk write remains" once the fan-out drains. Both mechanisms live in one process-global registry keyed by object name, so concurrent tests using distinct object names stay isolated. Barriers are placed on the real `rename_data` fan-out (phase `rename`) and the `commit_rename_data_dir` old-data-dir cleanup fan-out (phase `cleanup`). In production the barrier compiles to an immediately-ready `#[inline(always)]` no-op future and the task guard to `()`, so the fan-out control-flow shape and behavior are unchanged; only the `#[cfg(test)]` variants touch the registry. Coordinator lock-holding is asserted by the test at the store/coordinator layer via the guard it already holds; io_primitives has no handle to that namespace lock. Cross-process/black-box fault injection (toxiproxy, blackhole peers, 2-pool) remains a later cluster-harness block. Serves the barrier-style white-box acceptances of #1312 (commit fencing: abort at the first-disk rename barrier, assert no background disk write remains after release), #1319, and #1313. Three demo tests drive the real fan-out functions and double as regression guards: neutralizing the barrier seam makes the pause await time out, and neutralizing the task guard pins the in-flight count at zero, so reverting either seam fails the demos. --- .../src/set_disk/core/io_primitives.rs | 417 ++++++++++++++++++ 1 file changed, 417 insertions(+) diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs index cc5bb432a..2764bbf44 100644 --- a/crates/ecstore/src/set_disk/core/io_primitives.rs +++ b/crates/ecstore/src/set_disk/core/io_primitives.rs @@ -2626,6 +2626,11 @@ impl SetDisks { let dst_bucket = dst_bucket.clone(); futures.push(tokio::spawn(async move { + // Test-only introspection guard: counts this task as in-flight for + // the whole body. Compiles to `()` in production (no behavior). + #[allow(clippy::let_unit_value)] + let _fanout_task_guard = Self::rename_fanout_task_guard(&dst_object); + if file_info.erasure.index == 0 { file_info.erasure.index = i + 1; } @@ -2634,6 +2639,10 @@ impl SetDisks { return Err(DiskError::FileCorrupt); } + // Test-only awaitable pause point right before the disk rename. + // A no-op immediately-ready future in production. + Self::rename_fanout_barrier(&dst_object, i, rename_fanout_barrier_phase::RENAME).await; + if let Some(disk) = disk { disk.rename_data(&src_bucket, &src_object, file_info, &dst_bucket, &dst_object) .await @@ -2923,6 +2932,12 @@ impl SetDisks { let disk = disk.clone(); let object_for_fault = object_for_fault.clone(); tokio::spawn(async move { + // Test-only introspection guard + awaitable pause point for the + // old-data-dir cleanup fan-out. Both compile away in production. + #[allow(clippy::let_unit_value)] + let _fanout_task_guard = Self::rename_fanout_task_guard(&object_for_fault); + Self::rename_fanout_barrier(&object_for_fault, idx, rename_fanout_barrier_phase::CLEANUP).await; + if let Some(err) = Self::cleanup_injected_error(&object_for_fault, idx) { return Some(err); } @@ -2978,6 +2993,41 @@ impl SetDisks { #[inline(always)] fn record_read_version_call(_object: &str, _disk_index: usize) {} + /// Test-only awaitable pause point for the rename/commit fan-out (backlog#1325, + /// serving the barrier-style acceptances of #1312 / #1319 / #1313). `phase` is + /// [`rename_fanout_barrier::PHASE_RENAME`] or `PHASE_CLEANUP`. When a test has + /// armed a barrier for `object` at this `(disk_index, phase)`, the spawned + /// fan-out task blocks here until the test releases it, so the test can await + /// the pause point and then introspect in-flight background disk work. In + /// production this awaits an immediately-ready no-op future — no yield, no + /// registry access, no behavior change. + #[cfg(test)] + #[inline] + async fn rename_fanout_barrier(object: &str, disk_index: usize, phase: &'static str) { + rename_fanout_barrier::checkpoint(object, disk_index, phase).await; + } + + #[cfg(not(test))] + #[inline(always)] + #[allow(clippy::unused_async)] + async fn rename_fanout_barrier(_object: &str, _disk_index: usize, _phase: &'static str) {} + + /// Test-only RAII counter for one in-flight rename/commit fan-out task + /// (backlog#1325). Held for the whole spawned task body so a test observing + /// `object` can assert "background disk writes are still running" while the + /// fan-out is paused and "no background disk writes remain" once it drains — + /// the exact signal #1312 needs after a lock release. In production this + /// returns `()` and touches no registry. + #[cfg(test)] + #[inline] + fn rename_fanout_task_guard(object: &str) -> rename_fanout_barrier::TaskGuard { + rename_fanout_barrier::task_guard(object) + } + + #[cfg(not(test))] + #[inline(always)] + fn rename_fanout_task_guard(_object: &str) {} + /// Report a post-commit old-data-dir cleanup receipt: emit metrics, warn on /// residue/below-quorum, and — on residue — enqueue an object heal. /// @@ -4071,6 +4121,242 @@ pub(in crate::set_disk) mod disk_call_counters { } } +/// Fan-out phase labels for the rename/commit barrier seam (backlog#1325). +/// +/// These are referenced from the production fan-out call sites (as inert string +/// literals passed to a no-op seam), so — unlike the `#[cfg(test)]` barrier +/// registry itself — they are defined unconditionally. In production builds the +/// seam ignores them entirely. +pub(in crate::set_disk) mod rename_fanout_barrier_phase { + /// The per-disk `rename_data` phase of the write-commit fan-out. + pub const RENAME: &str = "rename"; + /// The per-disk old-data-dir cleanup phase of the commit fan-out. + pub const CLEANUP: &str = "cleanup"; +} + +/// Test-only awaitable pause barrier + background-task introspection for the +/// rename/commit fan-out (backlog#1325, the second facility block after the +/// per-disk call counters). Serves the barrier-style white-box acceptances of +/// #1312 (commit fencing: "abort at the first-disk rename barrier, assert the +/// coordinator still holds the lock, assert no background disk write remains +/// after release"), #1319, and #1313. +/// +/// Two independent, object-keyed mechanisms share one process-global registry: +/// +/// 1. **Awaitable pause barrier.** A test [`arm`]s a barrier for `(object, +/// disk_index, phase)`. The matching spawned fan-out task blocks at its +/// [`checkpoint`] until the test releases it. The test awaits the pause via +/// [`BarrierHandle::wait_until_paused`] (a deterministic `Notify` handshake — +/// no sleeps) and resumes it via [`BarrierHandle::release`]. At most one +/// barrier is armed per object at a time, matching the single-scope style of +/// `disk_call_counters`. +/// +/// 2. **Background-task introspection.** A test [`observe_tasks`] for `object`; +/// each instrumented fan-out task then holds a [`TaskGuard`] for its whole +/// body, so [`TaskTrackerScope::running`] reports how many rename/cleanup +/// background disk tasks are still in flight. This is the concrete "is there +/// still a background disk write?" signal #1312 asserts after a lock release. +/// +/// Both are keyed by object and only accumulate for armed/observed objects, so +/// concurrent tests using distinct object names stay fully isolated. The whole +/// module is `#[cfg(test)]` and never compiles into production; the fan-out call +/// sites reach it only through the `#[cfg(not(test))]` no-op seams on `SetDisks`. +/// +/// Scope of this block (white-box, in-process only): coordinator lock-holding is +/// asserted by the test at the store/coordinator layer via the guard it already +/// holds — `io_primitives` has no handle to that namespace lock, and fabricating +/// a lock-state registry here would be a look-alike rather than the real lock. +/// Cross-process/black-box fault injection (toxiproxy, blackhole peers, 2-pool) +/// is a later cluster-harness block, not this one. +#[cfg(test)] +pub(in crate::set_disk) mod rename_fanout_barrier { + use std::collections::HashMap; + use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; + use std::sync::{Arc, Mutex, OnceLock}; + use tokio::sync::Notify; + + pub use super::rename_fanout_barrier_phase::{CLEANUP as PHASE_CLEANUP, RENAME as PHASE_RENAME}; + + /// One armed barrier: the fan-out task matching `(disk_index, phase)` pauses. + struct Armed { + disk_index: usize, + phase: &'static str, + /// Signalled (task -> test) when the target task reaches the checkpoint. + arrived: Arc, + /// Signalled (test -> task) to release the paused task. + release: Arc, + /// Set once the target task is parked at the checkpoint. + paused: Arc, + } + + #[derive(Default)] + struct Registry { + /// object -> armed barrier (at most one per object). + armed: HashMap, + /// object -> live in-flight fan-out task count, only for observed objects. + observed: HashMap>, + } + + fn registry() -> &'static Mutex { + static REG: OnceLock> = OnceLock::new(); + REG.get_or_init(|| Mutex::new(Registry::default())) + } + + fn lock() -> std::sync::MutexGuard<'static, Registry> { + registry().lock().expect("rename fan-out barrier registry poisoned") + } + + /// RAII handle for one armed barrier. Dropping it disarms the barrier and + /// releases any still-parked task, so a panicking or forgetful test can never + /// leave a spawned fan-out task wedged. + #[must_use] + pub struct BarrierHandle { + object: String, + arrived: Arc, + release: Arc, + paused: Arc, + } + + /// Arm a barrier: the fan-out task for `object` at `(disk_index, phase)` will + /// pause at its checkpoint until the returned handle is released or dropped. + pub fn arm(object: &str, disk_index: usize, phase: &'static str) -> BarrierHandle { + let arrived = Arc::new(Notify::new()); + let release = Arc::new(Notify::new()); + let paused = Arc::new(AtomicBool::new(false)); + lock().armed.insert( + object.to_string(), + Armed { + disk_index, + phase, + arrived: arrived.clone(), + release: release.clone(), + paused: paused.clone(), + }, + ); + BarrierHandle { + object: object.to_string(), + arrived, + release, + paused, + } + } + + impl BarrierHandle { + /// Await until the armed fan-out task has parked at the checkpoint. Uses a + /// stored-permit `Notify`, so this is race-free regardless of whether the + /// task reaches the checkpoint before or after this call — no sleeps. + pub async fn wait_until_paused(&self) { + if self.paused.load(Ordering::SeqCst) { + return; + } + self.arrived.notified().await; + } + + /// Whether the target task is currently parked at the checkpoint. + pub fn is_paused(&self) -> bool { + self.paused.load(Ordering::SeqCst) + } + + /// Release the parked task so the fan-out can proceed. + pub fn release(&self) { + self.release.notify_one(); + } + } + + impl Drop for BarrierHandle { + fn drop(&mut self) { + // Unblock any task still parked at the checkpoint before disarming, so + // a dropped handle can never wedge a spawned fan-out task. + self.release.notify_one(); + lock().armed.remove(&self.object); + } + } + + /// Seam entry point invoked from inside each spawned fan-out task. A no-op + /// unless a barrier is armed for exactly this `(object, disk_index, phase)`. + /// The registry mutex is released before awaiting, so it is never held across + /// the pause. + pub(in crate::set_disk) async fn checkpoint(object: &str, disk_index: usize, phase: &'static str) { + let hooks = { + let reg = lock(); + match reg.armed.get(object) { + Some(a) if a.disk_index == disk_index && a.phase == phase => { + Some((a.arrived.clone(), a.release.clone(), a.paused.clone())) + } + _ => None, + } + }; + if let Some((arrived, release, paused)) = hooks { + paused.store(true, Ordering::SeqCst); + arrived.notify_one(); + release.notified().await; + } + } + + /// RAII scope that observes in-flight fan-out task counts for a single + /// `object`. Counts accrue only while the scope is alive; it clears its own + /// entry on drop. + #[must_use] + pub struct TaskTrackerScope { + object: String, + } + + /// Begin observing in-flight rename/cleanup fan-out task counts for `object`. + pub fn observe_tasks(object: &str) -> TaskTrackerScope { + lock() + .observed + .entry(object.to_string()) + .or_insert_with(|| Arc::new(AtomicUsize::new(0))); + TaskTrackerScope { + object: object.to_string(), + } + } + + impl TaskTrackerScope { + /// Number of instrumented fan-out tasks for the observed object that are + /// currently in flight (task body entered, guard not yet dropped). + pub fn running(&self) -> usize { + lock() + .observed + .get(&self.object) + .map(|c| c.load(Ordering::SeqCst)) + .unwrap_or(0) + } + } + + impl Drop for TaskTrackerScope { + fn drop(&mut self) { + lock().observed.remove(&self.object); + } + } + + /// RAII guard held for the lifetime of one spawned fan-out task. Increments + /// the observed counter on creation (if the object is observed) and decrements + /// it on drop. Holding the `Arc` keeps the decrement sound even if the scope + /// is dropped while a task is still draining. + #[must_use] + pub struct TaskGuard { + counter: Option>, + } + + /// Create a task guard for `object`. A no-op guard unless `object` is observed. + pub(in crate::set_disk) fn task_guard(object: &str) -> TaskGuard { + let counter = lock().observed.get(object).cloned(); + if let Some(c) = &counter { + c.fetch_add(1, Ordering::SeqCst); + } + TaskGuard { counter } + } + + impl Drop for TaskGuard { + fn drop(&mut self) { + if let Some(c) = &self.counter { + c.fetch_sub(1, Ordering::SeqCst); + } + } + } +} + #[cfg(test)] mod tests { use super::*; @@ -4294,6 +4580,137 @@ mod tests { drop(dirs); } + /// Bound for the pause handshake. This is a hang-guard, not a timing + /// dependency: under a working barrier `wait_until_paused` returns via the + /// `Notify` handshake far below this bound regardless of IO pressure, so the + /// assertions never depend on the value. It only turns a neutralized-barrier + /// hang into a deterministic failure instead of an infinite wait. + const BARRIER_PAUSE_GUARD: std::time::Duration = std::time::Duration::from_secs(10); + + fn rename_barrier_fileinfos(object: &str, count: usize) -> Vec { + (0..count).map(|_| metadata_test_fileinfo(object)).collect() + } + + /// Demo / regression guard for the backlog#1325 rename fan-out pause barrier + /// and background-task introspection. Serves the barrier-style acceptance of + /// #1312 ("assert no background disk write remains after release"). + /// + /// It drives the real `SetDisks::rename_data` fan-out: a barrier is armed at + /// the first disk's `rename` phase, and the test awaits that pause point, + /// asserts a background disk task is still in flight, releases it, and then + /// asserts the in-flight count drains to zero once the fan-out completes. + /// + /// Neutralizing the barrier seam (`rename_fanout_barrier` -> immediate no-op) + /// makes `wait_until_paused` never wake, so the guarded await elapses and the + /// test fails. Neutralizing the task guard (`rename_fanout_task_guard` -> + /// `()`) pins `running()` at zero, so the "still in flight" assertion fails. + #[tokio::test] + async fn rename_fanout_barrier_pauses_and_reports_running_background_tasks() { + const DISKS: usize = 4; + let bucket = "rename-barrier-bucket"; + let object = "rename-barrier-object"; + let (dirs, disks) = call_counter_local_disks(bucket, DISKS).await; + let file_infos = rename_barrier_fileinfos(object, DISKS); + + let tracker = rename_fanout_barrier::observe_tasks(object); + assert_eq!(tracker.running(), 0, "no fan-out tasks before rename starts"); + + let barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME); + + // Run the real rename fan-out concurrently with the control flow so we can + // introspect while it is parked at the first disk's rename checkpoint. + let rename_fut = SetDisks::rename_data(&disks, bucket, object, &file_infos, bucket, object, DISKS - 1); + let control_fut = async { + tokio::time::timeout(BARRIER_PAUSE_GUARD, barrier.wait_until_paused()) + .await + .expect("fan-out must reach the armed rename barrier"); + assert!(barrier.is_paused(), "target task must be parked at the checkpoint"); + assert!(tracker.running() >= 1, "a background rename task must still be in flight while paused"); + barrier.release(); + }; + let (_rename_res, ()) = tokio::join!(rename_fut, control_fut); + + // The fan-out has fully joined -> every task guard has dropped. + assert_eq!(tracker.running(), 0, "no background rename task may remain once the fan-out has drained"); + + drop(dirs); + } + + /// Demo / regression guard for the barrier on the commit (old-data-dir) + /// cleanup fan-out. Serves the same #1312/#1319 "no background disk write + /// after release" shape, on the reclamation path that runs *after* a write is + /// ACKed — the classic detached background delete #1312 fences against. + /// + /// It stages a real `object/` on two disks and drives the real + /// `commit_rename_data_dir` fan-out, pausing the first disk's cleanup delete. + #[tokio::test] + async fn commit_cleanup_fanout_barrier_pauses_background_deletes() { + let bucket = "cleanup-barrier-bucket"; + let object = "cleanup-barrier-object"; + let old_data_dir = "11111111-1111-1111-1111-111111111111"; + let committed_data_dir = "22222222-2222-2222-2222-222222222222"; + let path = format!("{object}/{old_data_dir}/part.1"); + let (_dir1, disk1) = read_multiple_test_disk(bucket, &[(&path, b"one".as_slice())]).await; + let (_dir2, disk2) = read_multiple_test_disk(bucket, &[(&path, b"two".as_slice())]).await; + let set = io_primitives_test_set(vec![Some(disk1.clone()), Some(disk2.clone())], 1).await; + let disks = [Some(disk1.clone()), Some(disk2.clone())]; + + let tracker = rename_fanout_barrier::observe_tasks(object); + let barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_CLEANUP); + + let cleanup_fut = set.commit_rename_data_dir(&disks, bucket, object, old_data_dir, committed_data_dir, 2); + let control_fut = async { + tokio::time::timeout(BARRIER_PAUSE_GUARD, barrier.wait_until_paused()) + .await + .expect("cleanup fan-out must reach the armed cleanup barrier"); + assert!(tracker.running() >= 1, "a background cleanup delete must still be in flight while paused"); + barrier.release(); + }; + let (cleanup, ()) = tokio::join!(cleanup_fut, control_fut); + + assert_eq!(tracker.running(), 0, "no background cleanup task may remain after drain"); + assert_eq!(cleanup.attempted, 2); + assert_eq!(cleanup.reclaimed, 2, "the real old-data-dir must still be reclaimed after release"); + + drop((disk1, disk2)); + } + + /// Isolation guard: an armed barrier / observed object only affects its own + /// object. A fan-out for a different (unobserved, unarmed) object must not be + /// paused and must not accrue any tracked task count — so concurrent tests + /// using distinct object names never interfere. + #[tokio::test] + async fn barrier_and_task_tracker_isolate_by_object() { + const DISKS: usize = 3; + let bucket = "barrier-iso-bucket"; + let observed_object = "iso-observed-object"; + let other_object = "iso-other-object"; + let (dirs, disks) = call_counter_local_disks(bucket, DISKS).await; + + // Observe + arm one object, then run the fan-out for a *different* object. + let tracker = rename_fanout_barrier::observe_tasks(observed_object); + let _barrier = rename_fanout_barrier::arm(observed_object, 0, rename_fanout_barrier::PHASE_RENAME); + + let file_infos = rename_barrier_fileinfos(other_object, DISKS); + // This must run to completion without ever pausing (no barrier for it) and + // must not touch the observed object's counter. A hang here (e.g. if the + // barrier ignored the object key) would surface as a timeout. + let _ = tokio::time::timeout( + BARRIER_PAUSE_GUARD, + SetDisks::rename_data(&disks, bucket, other_object, &file_infos, bucket, other_object, DISKS - 1), + ) + .await + .expect("fan-out for an unarmed object must not pause"); + + assert_eq!( + tracker.running(), + 0, + "another object's fan-out must not accrue tracked tasks for the observed object" + ); + + drop(dirs); + } + #[tokio::test] async fn multipart_codec_streaming_reader_zero_buffer_is_noop() { let reader = tokio::io::BufReader::new(Cursor::new(b"payload".to_vec()));