From 028ba6a6752907ae05f6c397957e0587b7ffdf89 Mon Sep 17 00:00:00 2001 From: Ramakrishna Chilaka <49393831+RamakrishnaChilaka@users.noreply.github.com> Date: Sat, 11 Jul 2026 23:06:24 +0530 Subject: [PATCH] perf(ecstore): parallelize multipart shard syncing (#4734) Bound large shard-directory syncs per disk and process while preserving small-directory and durability behavior. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- crates/ecstore/src/disk/local.rs | 89 ++++- crates/ecstore/src/disk/os.rs | 540 ++++++++++++++++++++++++++++++- 2 files changed, 621 insertions(+), 8 deletions(-) diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index 44ee8a6a4..94a7c0686 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -67,7 +67,7 @@ use tokio::fs::{self, File}; #[cfg(not(unix))] use tokio::io::AsyncReadExt; use tokio::io::{AsyncRead, AsyncSeekExt, AsyncWrite, AsyncWriteExt, ErrorKind, ReadBuf}; -use tokio::sync::{Notify, RwLock}; +use tokio::sync::{Notify, RwLock, Semaphore}; use tokio::time::{Instant, Sleep, interval_at, timeout}; use tracing::{debug, error, info, warn}; use uuid::Uuid; @@ -3639,6 +3639,7 @@ pub struct LocalDisk { startup_cleanup_notify: Arc, exit_signal: Option>, io_backend: Arc, + file_sync_permits: Arc, } impl Drop for LocalDisk { @@ -3876,6 +3877,7 @@ impl LocalDisk { startup_cleanup_notify, exit_signal: None, io_backend: build_local_io_backend(root.clone()), + file_sync_permits: os::disk_file_sync_limiter(&root), }; let (info, _root) = get_disk_info(root.clone()).await.inspect_err(|err| { log_startup_disk_error("get_disk_info", &root, err); @@ -6780,7 +6782,7 @@ impl DiskAPI for LocalDisk { let shard_sync = async { if durability.syncs_data_shards() && let Some((src_data_path, _)) = has_data_dir_path.as_ref() - && let Err(err) = os::sync_dir_files(src_data_path).await + && let Err(err) = os::sync_dir_files_with_limiter(src_data_path, self.file_sync_permits.clone()).await && err.kind() != ErrorKind::NotFound { return Err::<(), DiskError>(to_file_error(err).into()); @@ -9005,6 +9007,89 @@ mod test { ); } + #[tokio::test] + #[serial_test::serial(file_sync_probe)] + #[allow(clippy::await_holding_lock)] + async fn test_rename_data_shares_file_sync_limit_across_one_disk() { + use tempfile::tempdir; + + let _mode = durability_mode_override::set(DurabilityMode::Strict); + let dir = tempdir().expect("temp dir should be created"); + let endpoint = Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse"); + let disk = Arc::new(LocalDisk::new(&endpoint, false).await.expect("local disk should be created")); + let bucket = "shared-sync-limit-bucket"; + ensure_test_volume(&disk, bucket).await; + ensure_test_volume(&disk, RUSTFS_META_TMP_BUCKET).await; + + let first_data_dir = Uuid::from_u128(1); + let second_data_dir = Uuid::from_u128(2); + for (tmp_object, data_dir) in [("tmp-first", first_data_dir), ("tmp-second", second_data_dir)] { + let tmp_data_dir = dir + .path() + .join(RUSTFS_META_TMP_BUCKET) + .join(tmp_object) + .join(data_dir.to_string()); + fs::create_dir_all(&tmp_data_dir) + .await + .expect("staged data dir should be created"); + for part in 1..=os::MAX_PARALLEL_FILE_SYNCS { + fs::write(tmp_data_dir.join(format!("part.{part}")), b"shard") + .await + .expect("staged shard should be written"); + } + } + + let _probe = os::file_sync_probe::set_blocking(dir.path()); + let first = { + let disk = disk.clone(); + tokio::spawn(async move { + let fi = test_file_info("first-object", Uuid::from_u128(11), Some(first_data_dir), None); + disk.rename_data(RUSTFS_META_TMP_BUCKET, "tmp-first", fi, bucket, "first-object") + .await + }) + }; + let second = { + let disk = disk.clone(); + tokio::spawn(async move { + let fi = test_file_info("second-object", Uuid::from_u128(12), Some(second_data_dir), None); + disk.rename_data(RUSTFS_META_TMP_BUCKET, "tmp-second", fi, bucket, "second-object") + .await + }) + }; + os::file_sync_probe::wait_for_active(os::MAX_PARALLEL_FILE_SYNCS).await; + + assert_eq!( + disk.file_sync_permits.available_permits(), + 0, + "concurrent rename_data calls must share the LocalDisk sync limit" + ); + assert_eq!( + os::file_sync_probe::peak(), + os::MAX_PARALLEL_FILE_SYNCS, + "one disk must not exceed its shared file-sync capacity" + ); + let reconnected = LocalDisk::new(&endpoint, false).await.expect("local disk should reconnect"); + assert!( + Arc::ptr_eq(&disk.file_sync_permits, &reconnected.file_sync_permits), + "reconnecting the same disk must preserve its file-sync limiter" + ); + assert_eq!( + reconnected.file_sync_permits.available_permits(), + 0, + "reconnected disk must inherit the outstanding sync budget" + ); + + os::file_sync_probe::release(); + first + .await + .expect("first rename_data task should join") + .expect("first rename_data should commit"); + second + .await + .expect("second rename_data task should join") + .expect("second rename_data should commit"); + } + #[tokio::test] #[allow(clippy::await_holding_lock)] async fn test_write_all_meta_skips_tmp_parent_dir_fsync_but_fsyncs_dst_parent() { diff --git a/crates/ecstore/src/disk/os.rs b/crates/ecstore/src/disk/os.rs index 268148337..ab4aa80a7 100644 --- a/crates/ecstore/src/disk/os.rs +++ b/crates/ecstore/src/disk/os.rs @@ -15,12 +15,17 @@ use crate::disk::error::DiskError; use crate::disk::error::Result; use crate::disk::error_conv::to_file_error; +use futures::TryStreamExt; +use parking_lot::Mutex; use rustfs_utils::path::SLASH_SEPARATOR; use std::{ + collections::HashMap, io, - path::{Component, Path}, + path::{Component, Path, PathBuf}, + sync::{Arc, LazyLock, Weak}, }; use tokio::fs; +use tokio::sync::{OwnedSemaphorePermit, Semaphore, SemaphorePermit}; use tracing::warn; /// Check path length according to OS limits. @@ -105,23 +110,290 @@ pub async fn fsync_dir(dir: impl AsRef) -> io::Result<()> { tokio::task::spawn_blocking(move || fsync_dir_std(dir)).await? } +// Small object directories are cheaper to flush in one blocking task. Multipart +// directories fan out only once enough files can amortize per-task scheduling. +const PARALLEL_FILE_SYNC_THRESHOLD: usize = 16; +pub(crate) const MAX_PARALLEL_FILE_SYNCS: usize = 16; +// Scale aggregate fan-out for wider nodes while reserving at least half of the +// configured Tokio blocking pool for unrelated filesystem work. +const MIN_GLOBAL_FILE_SYNCS: usize = 64; +const MAX_GLOBAL_FILE_SYNCS: usize = 512; +#[cfg(test)] +const TEST_GLOBAL_FILE_SYNCS: usize = 64; + +static FILE_SYNC_PERMITS: LazyLock = LazyLock::new(|| Semaphore::new(global_file_sync_limit())); +static DISK_FILE_SYNC_LIMITERS: LazyLock>>> = LazyLock::new(|| Mutex::new(HashMap::new())); + +fn default_global_file_sync_limit(cpu_count: usize, max_blocking_threads: usize) -> usize { + let cpu_scaled = cpu_count + .max(1) + .saturating_mul(MAX_PARALLEL_FILE_SYNCS) + .clamp(MIN_GLOBAL_FILE_SYNCS, MAX_GLOBAL_FILE_SYNCS); + cpu_scaled.min((max_blocking_threads.max(1) / 2).max(1)) +} + +#[cfg(not(test))] +fn global_file_sync_limit() -> usize { + let max_blocking_threads = + rustfs_utils::get_env_usize(rustfs_config::ENV_MAX_BLOCKING_THREADS, rustfs_config::DEFAULT_MAX_BLOCKING_THREADS); + default_global_file_sync_limit(num_cpus::get(), max_blocking_threads) +} + +#[cfg(test)] +fn global_file_sync_limit() -> usize { + TEST_GLOBAL_FILE_SYNCS +} + +/// Reuse a disk's limiter across reconnects while detached sync calls still hold it. +pub(crate) fn disk_file_sync_limiter(root: &Path) -> Arc { + let mut limiters = DISK_FILE_SYNC_LIMITERS.lock(); + limiters.retain(|_, limiter| limiter.strong_count() > 0); + if let Some(limiter) = limiters.get(root).and_then(Weak::upgrade) { + return limiter; + } + + let limiter = Arc::new(Semaphore::new(MAX_PARALLEL_FILE_SYNCS)); + limiters.insert(root.to_path_buf(), Arc::downgrade(&limiter)); + limiter +} + +/// Always acquire the per-disk permit before the process-wide permit. Keeping +/// this order uniform prevents one slow disk from reserving global capacity +/// while it waits for its own concurrency slot. +async fn acquire_file_sync_permits(disk_permits: Arc) -> io::Result<(OwnedSemaphorePermit, SemaphorePermit<'static>)> { + let disk_permit = disk_permits + .acquire_owned() + .await + .map_err(|_| io::Error::other("disk file sync concurrency limiter closed"))?; + let global_permit = FILE_SYNC_PERMITS + .acquire() + .await + .map_err(|_| io::Error::other("global file sync concurrency limiter closed"))?; + Ok((disk_permit, global_permit)) +} + +/// Keep the per-disk permit with the blocking syscall so cancellation cannot +/// amplify work on a wedged disk. The global permit stays with the async waiter, +/// allowing healthy disks to make progress after a timed-out request is dropped. +async fn run_file_sync_blocking(disk_permits: Arc, work: F) -> io::Result +where + T: Send + 'static, + F: FnOnce() -> io::Result + Send + 'static, +{ + let (disk_permit, global_permit) = acquire_file_sync_permits(disk_permits).await?; + let result = tokio::task::spawn_blocking(move || { + let _disk_permit = disk_permit; + work() + }) + .await; + drop(global_permit); + result? +} + +#[cfg(test)] +pub(crate) mod file_sync_probe { + use std::path::{Path, PathBuf}; + use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; + use std::sync::{Condvar, Mutex, RwLock}; + use tokio::sync::Notify; + + static ROOT: RwLock> = RwLock::new(None); + static BLOCK_MUTEX: Mutex<()> = Mutex::new(()); + static BLOCK_CONDVAR: Condvar = Condvar::new(); + static ACTIVE_CHANGED: Notify = Notify::const_new(); + static ACTIVE: AtomicUsize = AtomicUsize::new(0); + static PEAK: AtomicUsize = AtomicUsize::new(0); + static ATTEMPTS: AtomicUsize = AtomicUsize::new(0); + static FAIL_ON_ATTEMPT: AtomicUsize = AtomicUsize::new(usize::MAX); + static BLOCK: AtomicBool = AtomicBool::new(false); + + pub(crate) struct ProbeGuard; + + pub(super) struct ActiveGuard { + fail: bool, + } + + impl ActiveGuard { + pub(super) fn should_fail(&self) -> bool { + self.fail + } + } + + impl Drop for ActiveGuard { + fn drop(&mut self) { + ACTIVE.fetch_sub(1, Ordering::SeqCst); + ACTIVE_CHANGED.notify_waiters(); + } + } + + impl Drop for ProbeGuard { + fn drop(&mut self) { + release(); + FAIL_ON_ATTEMPT.store(usize::MAX, Ordering::SeqCst); + *ROOT.write().expect("file sync probe lock poisoned") = None; + } + } + + fn configure(root: &Path, fail_on_attempt: Option, block: bool) -> ProbeGuard { + ACTIVE.store(0, Ordering::SeqCst); + PEAK.store(0, Ordering::SeqCst); + ATTEMPTS.store(0, Ordering::SeqCst); + FAIL_ON_ATTEMPT.store(fail_on_attempt.unwrap_or(usize::MAX), Ordering::SeqCst); + { + let _guard = BLOCK_MUTEX.lock().expect("file sync probe blocker poisoned"); + BLOCK.store(block, Ordering::SeqCst); + } + *ROOT.write().expect("file sync probe lock poisoned") = Some(root.to_path_buf()); + ProbeGuard + } + + pub(super) fn set(root: &Path) -> ProbeGuard { + configure(root, None, false) + } + + pub(super) fn set_failing(root: &Path) -> ProbeGuard { + configure(root, Some(1), false) + } + + pub(super) fn set_failing_blocking(root: &Path) -> ProbeGuard { + configure(root, Some(1), true) + } + + pub(crate) fn set_blocking(root: &Path) -> ProbeGuard { + configure(root, None, true) + } + + pub(super) fn enter(path: &Path) -> Option { + let enabled = ROOT + .read() + .expect("file sync probe lock poisoned") + .as_ref() + .is_some_and(|root| path.starts_with(root)); + if !enabled { + return None; + } + + let attempt = ATTEMPTS.fetch_add(1, Ordering::SeqCst) + 1; + let active = ACTIVE.fetch_add(1, Ordering::SeqCst) + 1; + PEAK.fetch_max(active, Ordering::SeqCst); + ACTIVE_CHANGED.notify_waiters(); + let fail = attempt == FAIL_ON_ATTEMPT.load(Ordering::SeqCst); + if !fail { + let guard = BLOCK_MUTEX.lock().expect("file sync probe blocker poisoned"); + drop( + BLOCK_CONDVAR + .wait_while(guard, |_| BLOCK.load(Ordering::SeqCst)) + .expect("file sync probe blocker poisoned"), + ); + } + Some(ActiveGuard { fail }) + } + + pub(crate) fn peak() -> usize { + PEAK.load(Ordering::SeqCst) + } + + pub(super) fn attempts() -> usize { + ATTEMPTS.load(Ordering::SeqCst) + } + + pub(crate) async fn wait_for_active(target: usize) { + loop { + let changed = ACTIVE_CHANGED.notified(); + if ACTIVE.load(Ordering::SeqCst) >= target { + return; + } + changed.await; + } + } + + pub(super) async fn wait_for_idle() { + loop { + let changed = ACTIVE_CHANGED.notified(); + if ACTIVE.load(Ordering::SeqCst) == 0 { + return; + } + changed.await; + } + } + + pub(crate) fn release() { + let _guard = BLOCK_MUTEX.lock().expect("file sync probe blocker poisoned"); + BLOCK.store(false, Ordering::SeqCst); + BLOCK_CONDVAR.notify_all(); + } +} + +fn sync_file(path: &Path) -> io::Result<()> { + #[cfg(test)] + let _probe = file_sync_probe::enter(path); + #[cfg(test)] + if _probe.as_ref().is_some_and(file_sync_probe::ActiveGuard::should_fail) { + return Err(io::Error::other("injected file sync failure")); + } + std::fs::File::open(path)?.sync_data() +} + +fn sync_files(paths: &[PathBuf]) -> io::Result<()> { + for path in paths { + sync_file(path)?; + } + Ok(()) +} + +fn regular_files(dir: &Path) -> io::Result> { + let mut files = Vec::with_capacity(PARALLEL_FILE_SYNC_THRESHOLD); + for entry in std::fs::read_dir(dir)? { + let entry = entry?; + if entry.file_type()?.is_file() { + files.push(entry.path()); + } + } + Ok(files) +} + /// Fdatasync every regular file directly inside `dir`, then fsync the directory -/// itself. Used at commit points so erasure shard files written through the page -/// cache are durable before their directory is renamed into its final location. +/// itself. pub fn sync_dir_files_std(dir: impl AsRef) -> io::Result<()> { for entry in std::fs::read_dir(dir.as_ref())? { let entry = entry?; if entry.file_type()?.is_file() { - std::fs::File::open(entry.path())?.sync_data()?; + sync_file(&entry.path())?; } } fsync_dir_std(dir) } -/// Async wrapper around [`sync_dir_files_std`]; runs the blocking syncs off the runtime. +/// Async wrapper around [`sync_dir_files_std`]. Large directories flush files +/// concurrently, bounded both per directory and process-wide. pub async fn sync_dir_files(dir: impl AsRef) -> io::Result<()> { + sync_dir_files_with_limiter(dir, Arc::new(Semaphore::new(MAX_PARALLEL_FILE_SYNCS))).await +} + +pub(crate) async fn sync_dir_files_with_limiter(dir: impl AsRef, disk_permits: Arc) -> io::Result<()> { let dir = dir.as_ref().to_path_buf(); - tokio::task::spawn_blocking(move || sync_dir_files_std(dir)).await? + let scan_dir = dir.clone(); + let files = run_file_sync_blocking(disk_permits.clone(), move || { + let files = regular_files(&scan_dir)?; + if files.len() < PARALLEL_FILE_SYNC_THRESHOLD { + sync_files(&files)?; + fsync_dir_std(scan_dir)?; + return Ok(None); + } + Ok::<_, io::Error>(Some(files)) + }) + .await?; + + let Some(files) = files else { + return Ok(()); + }; + futures::stream::iter(files.into_iter().map(Ok::<_, io::Error>)) + .try_for_each_concurrent(MAX_PARALLEL_FILE_SYNCS, |path| { + let disk_permits = disk_permits.clone(); + async move { run_file_sync_blocking(disk_permits, move || sync_file(&path)).await } + }) + .await?; + run_file_sync_blocking(disk_permits, move || fsync_dir_std(dir)).await } /// Check if the given disk path is the root disk. @@ -352,6 +624,19 @@ mod tests { use super::*; use tempfile::tempdir; + fn file_sync_limiter() -> Arc { + Arc::new(Semaphore::new(MAX_PARALLEL_FILE_SYNCS)) + } + + #[test] + fn global_file_sync_limit_scales_and_preserves_blocking_capacity() { + assert_eq!(default_global_file_sync_limit(1, 1024), MIN_GLOBAL_FILE_SYNCS); + assert_eq!(default_global_file_sync_limit(16, 1024), 256); + assert_eq!(default_global_file_sync_limit(64, 1024), MAX_GLOBAL_FILE_SYNCS); + assert_eq!(default_global_file_sync_limit(64, 128), 64); + assert_eq!(default_global_file_sync_limit(0, 0), 1); + } + #[tokio::test] async fn rename_all_missing_source_returns_file_not_found() { let temp_dir = tempdir().expect("create temp dir"); @@ -419,19 +704,262 @@ mod tests { } #[tokio::test] + #[serial_test::serial(file_sync_probe)] async fn sync_dir_files_syncs_regular_files_and_dir() { let temp_dir = tempdir().expect("create temp dir"); std::fs::write(temp_dir.path().join("part.1"), b"shard-one").expect("write part.1"); std::fs::write(temp_dir.path().join("part.2"), b"shard-two").expect("write part.2"); std::fs::create_dir(temp_dir.path().join("subdir")).expect("create subdir"); + let _probe = file_sync_probe::set(temp_dir.path()); sync_dir_files(temp_dir.path()).await.expect("sync dir files must succeed"); + + assert_eq!(std::fs::read(temp_dir.path().join("part.1")).expect("read part.1"), b"shard-one"); + assert!( + fsync_dir_recorder::was_fsynced(temp_dir.path()), + "successful sequential sync must fsync the directory" + ); } #[tokio::test] + #[serial_test::serial(file_sync_probe)] + async fn sync_dir_files_parallelizes_large_directories() { + let temp_dir = tempdir().expect("create temp dir"); + for index in 0..PARALLEL_FILE_SYNC_THRESHOLD { + std::fs::write(temp_dir.path().join(format!("part.{index}")), b"shard").expect("write part"); + } + let _probe = file_sync_probe::set_blocking(temp_dir.path()); + let path = temp_dir.path().to_path_buf(); + let task = tokio::spawn(async move { sync_dir_files_with_limiter(path, file_sync_limiter()).await }); + file_sync_probe::wait_for_active(MAX_PARALLEL_FILE_SYNCS).await; + + assert!(file_sync_probe::peak() > 1, "large directories must sync more than one file concurrently"); + assert!( + file_sync_probe::peak() <= MAX_PARALLEL_FILE_SYNCS.min(TEST_GLOBAL_FILE_SYNCS), + "file sync concurrency must remain bounded" + ); + file_sync_probe::release(); + task.await + .expect("join parallel file sync") + .expect("parallel file sync must succeed"); + assert!( + fsync_dir_recorder::was_fsynced(temp_dir.path()), + "successful parallel sync must fsync the directory" + ); + } + + #[tokio::test] + #[serial_test::serial(file_sync_probe)] + async fn sync_dir_files_keeps_small_directories_sequential() { + let temp_dir = tempdir().expect("create temp dir"); + for index in 0..(PARALLEL_FILE_SYNC_THRESHOLD - 1) { + std::fs::write(temp_dir.path().join(format!("part.{index}")), b"shard").expect("write part"); + } + let _probe = file_sync_probe::set_blocking(temp_dir.path()); + let path = temp_dir.path().to_path_buf(); + let task = tokio::spawn(async move { sync_dir_files_with_limiter(path, file_sync_limiter()).await }); + file_sync_probe::wait_for_active(1).await; + + assert_eq!(file_sync_probe::peak(), 1, "small directories must avoid parallel task overhead"); + file_sync_probe::release(); + task.await + .expect("join sequential file sync") + .expect("sequential file sync must succeed"); + } + + #[tokio::test] + #[serial_test::serial(file_sync_probe)] + async fn sync_dir_files_bounds_concurrency_across_directories() { + let temp_dir = tempdir().expect("create temp dir"); + let directory_count = TEST_GLOBAL_FILE_SYNCS / MAX_PARALLEL_FILE_SYNCS + 1; + let mut directories = Vec::with_capacity(directory_count); + for directory_index in 0..directory_count { + let directory = temp_dir.path().join(format!("disk.{directory_index}")); + std::fs::create_dir(&directory).expect("create disk directory"); + for file_index in 0..PARALLEL_FILE_SYNC_THRESHOLD { + std::fs::write(directory.join(format!("part.{file_index}")), b"shard").expect("write part"); + } + directories.push(directory); + } + let _probe = file_sync_probe::set_blocking(temp_dir.path()); + let task = tokio::spawn(async move { + futures::future::join_all( + directories + .iter() + .map(|directory| sync_dir_files_with_limiter(directory, file_sync_limiter())), + ) + .await + }); + file_sync_probe::wait_for_active(TEST_GLOBAL_FILE_SYNCS).await; + + assert!( + file_sync_probe::peak() > MAX_PARALLEL_FILE_SYNCS, + "independent directories should share the global sync capacity" + ); + assert!( + file_sync_probe::peak() <= TEST_GLOBAL_FILE_SYNCS, + "aggregate file sync concurrency must remain process-bounded" + ); + file_sync_probe::release(); + let results = task.await.expect("join cross-directory file syncs"); + assert!(results.iter().all(std::result::Result::is_ok), "all directory syncs must succeed"); + } + + #[tokio::test] + #[serial_test::serial(file_sync_probe)] + async fn sync_dir_files_bounds_concurrency_per_disk() { + let temp_dir = tempdir().expect("create temp dir"); + let mut directories = Vec::with_capacity(2); + for directory_index in 0..2 { + let directory = temp_dir.path().join(format!("disk.{directory_index}")); + std::fs::create_dir(&directory).expect("create disk directory"); + for file_index in 0..PARALLEL_FILE_SYNC_THRESHOLD { + std::fs::write(directory.join(format!("part.{file_index}")), b"shard").expect("write part"); + } + directories.push(directory); + } + let _probe = file_sync_probe::set_blocking(temp_dir.path()); + let disk_permits = file_sync_limiter(); + let task = tokio::spawn(async move { + futures::future::join_all( + directories + .iter() + .map(|directory| sync_dir_files_with_limiter(directory, disk_permits.clone())), + ) + .await + }); + file_sync_probe::wait_for_active(MAX_PARALLEL_FILE_SYNCS).await; + + assert!(file_sync_probe::peak() > 1, "one disk should sync multiple files concurrently"); + assert!( + file_sync_probe::peak() <= MAX_PARALLEL_FILE_SYNCS, + "one disk must not exceed its own sync capacity" + ); + file_sync_probe::release(); + let results = task.await.expect("join per-disk file syncs"); + assert!(results.iter().all(std::result::Result::is_ok), "all directory syncs must succeed"); + } + + #[tokio::test] + #[serial_test::serial(file_sync_probe)] + async fn sync_dir_files_acquires_disk_capacity_before_global_capacity() { + let global_reservation = FILE_SYNC_PERMITS + .acquire_many(TEST_GLOBAL_FILE_SYNCS as u32) + .await + .expect("global file sync limiter must remain open"); + let disk_permits = Arc::new(Semaphore::new(1)); + let mut acquisition = Box::pin(acquire_file_sync_permits(disk_permits.clone())); + + assert!(futures::poll!(&mut acquisition).is_pending()); + assert_eq!( + disk_permits.available_permits(), + 0, + "a waiter blocked on global capacity must already hold its disk permit" + ); + + drop(acquisition); + assert_eq!(disk_permits.available_permits(), 1, "cancelling the waiter must return its disk permit"); + drop(global_reservation); + } + + #[tokio::test] + #[serial_test::serial(file_sync_probe)] + async fn sync_dir_files_does_not_fsync_dir_after_sequential_file_failure() { + let temp_dir = tempdir().expect("create temp dir"); + std::fs::write(temp_dir.path().join("part.1"), b"shard").expect("write part"); + let _probe = file_sync_probe::set_failing(temp_dir.path()); + + let err = sync_dir_files_with_limiter(temp_dir.path(), file_sync_limiter()) + .await + .expect_err("file sync failure must propagate"); + + assert_eq!(err.kind(), io::ErrorKind::Other); + assert!( + !fsync_dir_recorder::was_fsynced(temp_dir.path()), + "directory must not be fsynced after a file sync failure" + ); + } + + #[tokio::test] + #[serial_test::serial(file_sync_probe)] + async fn sync_dir_files_parallel_failure_stops_new_work_and_skips_dir_fsync() { + let temp_dir = tempdir().expect("create temp dir"); + let file_count = TEST_GLOBAL_FILE_SYNCS * 16; + for index in 0..file_count { + std::fs::write(temp_dir.path().join(format!("part.{index}")), b"shard").expect("write part"); + } + let _probe = file_sync_probe::set_failing_blocking(temp_dir.path()); + + let err = sync_dir_files_with_limiter(temp_dir.path(), file_sync_limiter()) + .await + .expect_err("parallel file sync failure must propagate"); + + assert_eq!(err.kind(), io::ErrorKind::Other); + assert!( + file_sync_probe::attempts() <= MAX_PARALLEL_FILE_SYNCS, + "parallel sync must stop scheduling files after the first failure" + ); + assert!( + !fsync_dir_recorder::was_fsynced(temp_dir.path()), + "directory must not be fsynced after a parallel file sync failure" + ); + file_sync_probe::release(); + file_sync_probe::wait_for_idle().await; + } + + #[tokio::test] + #[serial_test::serial(file_sync_probe)] + async fn sync_dir_files_cancellation_isolates_global_capacity_from_stuck_disk_work() { + let temp_dir = tempdir().expect("create temp dir"); + for index in 0..PARALLEL_FILE_SYNC_THRESHOLD { + std::fs::write(temp_dir.path().join(format!("part.{index}")), b"shard").expect("write part"); + } + let _probe = file_sync_probe::set_blocking(temp_dir.path()); + let disk_permits = file_sync_limiter(); + let initial_disk_permits = disk_permits.available_permits(); + let global_reservation = FILE_SYNC_PERMITS + .acquire_many((TEST_GLOBAL_FILE_SYNCS - MAX_PARALLEL_FILE_SYNCS) as u32) + .await + .expect("global file sync limiter must remain open"); + let path = temp_dir.path().to_path_buf(); + let task = tokio::spawn({ + let disk_permits = disk_permits.clone(); + async move { sync_dir_files_with_limiter(path, disk_permits).await } + }); + file_sync_probe::wait_for_active(MAX_PARALLEL_FILE_SYNCS).await; + + task.abort(); + let join_err = task.await.expect_err("file sync task must be cancelled"); + + assert!(join_err.is_cancelled(), "task abort must cancel the outer file sync future"); + assert_eq!( + disk_permits.available_permits(), + 0, + "detached blocking syncs must retain their per-disk permits" + ); + let returned_global_permits = FILE_SYNC_PERMITS + .try_acquire_many(MAX_PARALLEL_FILE_SYNCS as u32) + .expect("cancelled work must return global capacity for healthy disks"); + file_sync_probe::release(); + file_sync_probe::wait_for_idle().await; + let returned_disk_permits = tokio::time::timeout( + std::time::Duration::from_secs(30), + disk_permits.clone().acquire_many_owned(initial_disk_permits as u32), + ) + .await + .expect("blocking syncs must return their per-disk permits") + .expect("disk file sync limiter must remain open"); + drop(returned_disk_permits); + drop(returned_global_permits); + drop(global_reservation); + } + + #[tokio::test] + #[serial_test::serial(file_sync_probe)] async fn sync_dir_files_missing_dir_returns_not_found() { let temp_dir = tempdir().expect("create temp dir"); let missing = temp_dir.path().join("missing"); + let _probe = file_sync_probe::set(temp_dir.path()); let err = sync_dir_files(&missing).await.expect_err("missing dir must fail"); assert_eq!(err.kind(), io::ErrorKind::NotFound);