diff --git a/crates/ecstore/src/disk/os.rs b/crates/ecstore/src/disk/os.rs index 8e475d297..f64902130 100644 --- a/crates/ecstore/src/disk/os.rs +++ b/crates/ecstore/src/disk/os.rs @@ -1756,7 +1756,7 @@ impl DiskBlockingLane { struct DiskBlockingQueueState { jobs: VecDeque>, active_workers: usize, - #[cfg(any(test, feature = "test-util"))] + #[cfg(any(test, feature = "test-util", feature = "e2e-test-hooks"))] paused_workers: usize, } @@ -1786,7 +1786,7 @@ struct DiskBlockingWorkerGuard { impl Drop for DiskBlockingWorkerGuard { fn drop(&mut self) { - #[cfg(any(test, feature = "test-util"))] + #[cfg(any(test, feature = "test-util", feature = "e2e-test-hooks"))] if self.state != DiskBlockingWorkerState::Submitted { CURRENT_DISK_BLOCKING_QUEUE.with(|slot| { let _ = slot.borrow_mut().take(); @@ -1845,11 +1845,11 @@ impl DiskBlockingQueue { #[inline] fn effective_max_workers(&self, state: &DiskBlockingQueueState) -> usize { - #[cfg(any(test, feature = "test-util"))] + #[cfg(any(test, feature = "test-util", feature = "e2e-test-hooks"))] { MAX_NAMESPACE_WORKERS_PER_DISK.saturating_add(state.paused_workers) } - #[cfg(not(any(test, feature = "test-util")))] + #[cfg(not(any(test, feature = "test-util", feature = "e2e-test-hooks")))] { let _ = state; MAX_NAMESPACE_WORKERS_PER_DISK @@ -1901,7 +1901,7 @@ impl DiskBlockingQueue { }; let worker = move || { let mut guard = guard; - #[cfg(any(test, feature = "test-util"))] + #[cfg(any(test, feature = "test-util", feature = "e2e-test-hooks"))] CURRENT_DISK_BLOCKING_QUEUE.with(|slot| { *slot.borrow_mut() = Some((Arc::clone(&queue), lane)); }); @@ -1998,18 +1998,18 @@ impl DiskBlockingQueue { } } -#[cfg(any(test, feature = "test-util"))] +#[cfg(any(test, feature = "test-util", feature = "e2e-test-hooks"))] thread_local! { static CURRENT_DISK_BLOCKING_QUEUE: std::cell::RefCell, DiskBlockingLane)>> = const { std::cell::RefCell::new(None) }; } -#[cfg(any(test, feature = "test-util"))] +#[cfg(any(test, feature = "test-util", feature = "e2e-test-hooks"))] pub(crate) struct BlockingTestHookGuard { queue: Option<(Arc, DiskBlockingLane)>, } -#[cfg(any(test, feature = "test-util"))] +#[cfg(any(test, feature = "test-util", feature = "e2e-test-hooks"))] impl Drop for BlockingTestHookGuard { fn drop(&mut self) { if let Some((queue, lane)) = self.queue.take() { @@ -2019,7 +2019,7 @@ impl Drop for BlockingTestHookGuard { } } -#[cfg(any(test, feature = "test-util"))] +#[cfg(any(test, feature = "test-util", feature = "e2e-test-hooks"))] pub(crate) fn enter_blocking_test_hook() -> BlockingTestHookGuard { let queue = CURRENT_DISK_BLOCKING_QUEUE.with(|slot| slot.borrow().clone()); if let Some((ref queue, lane)) = queue { @@ -3588,6 +3588,8 @@ impl PhysicalRenameTestBarrier { } fn wait(path: &Path) -> io::Result<()> { + // The synthetic pause retains physical ownership while unrelated disk work progresses. + let _pause = enter_blocking_test_hook(); std::fs::write(path.with_extension("arrived"), b"physical worker entered")?; let deadline = Instant::now() + Duration::from_secs(120); while !path.with_extension("release").try_exists()? { @@ -6081,6 +6083,84 @@ mod tests { use tempfile::tempdir; use tracing_subscriber::fmt::MakeWriter; + #[cfg(feature = "e2e-test-hooks")] + #[tokio::test] + async fn physical_rename_barrier_allows_control_work_without_releasing_cancelled_owner() { + struct ReleaseBarrier(PathBuf); + + impl Drop for ReleaseBarrier { + fn drop(&mut self) { + let _ = std::fs::write(self.0.with_extension("release"), b"release physical worker"); + } + } + + let temp = tempdir().expect("physical barrier fixture"); + let root = temp.path().canonicalize().expect("canonical disk root"); + let queue = register_disk_blocking_queue(&root, &root); + let source = root.join("source"); + let destination = root.join("destination"); + std::fs::write(&source, b"physical payload").expect("staged payload"); + let metadata = root.join("control-metadata"); + std::fs::write(&metadata, b"control").expect("existing control metadata"); + let barrier = root.join(".rustfs.sys/e2e-physical-rename-barrier"); + std::fs::create_dir_all(barrier.parent().expect("barrier parent")).expect("barrier directory"); + std::fs::write(&barrier, destination.to_str().expect("destination path")).expect("arm physical barrier"); + let release = ReleaseBarrier(barrier.clone()); + let lease = acquire_namespace_mutation_lease(&destination).await; + assert!( + Arc::ptr_eq(&lease.blocking_queue, &queue), + "control and physical work share the disk queue" + ); + let owner = Arc::downgrade(&lease); + let publication_root = PublicationRoot::new(&root).expect("pinned disk root"); + let worker_destination = destination.clone(); + let waiter = tokio::spawn(async move { + reliable_rename_inner_with_lease(source, worker_destination, root, publication_root, false, lease).await + }); + tokio::time::timeout(Duration::from_secs(5), async { + while !barrier.with_extension("arrived").exists() { + tokio::task::yield_now().await; + } + }) + .await + .expect("physical worker enters the real E2E barrier"); + waiter.abort(); + assert!(waiter.await.expect_err("cancel async rename waiter").is_cancelled()); + assert!(barrier.with_extension("waiter-dropped").exists(), "receiver observes waiter cancellation"); + + let control_owner = owner.clone(); + let control_destination = destination.clone(); + let control = tokio::time::timeout( + Duration::from_secs(5), + queue.run(move || { + Ok(( + std::fs::read(metadata)?, + control_owner.upgrade().is_some(), + disk_namespace_mutation_lock(&control_destination).try_write_owned().is_err(), + control_destination.exists(), + )) + }), + ) + .await; + // Drain even when the progress oracle fails, so a regression does not strand a worker. + drop(release); + tokio::time::timeout(Duration::from_secs(5), async { + while owner.upgrade().is_some() || !destination.exists() { + tokio::task::yield_now().await; + } + }) + .await + .expect("physical completion releases the namespace owner"); + let (bytes, retained, locked, published) = control + .expect("control work progresses during the pause") + .expect("control read"); + assert_eq!(bytes, b"control"); + assert!(retained && locked, "cancelled waiter must leave the physical namespace lease owned"); + assert!(!published, "control work must run before physical publication"); + assert!(disk_namespace_mutation_lock(&destination).try_write_owned().is_ok()); + assert_eq!(std::fs::read(destination).expect("completed physical publication"), b"physical payload"); + } + #[tokio::test] async fn rename_staging_leases_are_ordered_and_deduplicated() { let root = tempdir().expect("namespace fixture");