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

@@ -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<Notify>,
     exit_signal: Option<tokio::sync::broadcast::Sender<()>>,
     io_backend: Arc<dyn LocalIoBackend>,
+    file_sync_permits: Arc<Semaphore>,
 }
 
 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() {

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<Path>) -> 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<Semaphore> = LazyLock::new(|| Semaphore::new(global_file_sync_limit()));
+static DISK_FILE_SYNC_LIMITERS: LazyLock<Mutex<HashMap<PathBuf, Weak<Semaphore>>>> = 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<Semaphore> {
+    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<Semaphore>) -> 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<T, F>(disk_permits: Arc<Semaphore>, work: F) -> io::Result<T>
+where
+    T: Send + 'static,
+    F: FnOnce() -> io::Result<T> + 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<Option<PathBuf>> = 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<usize>, 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<ActiveGuard> {
+        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<Vec<PathBuf>> {
+    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
+/// itself.
-/// cache are durable before their directory is renamed into its final location.
 pub fn sync_dir_files_std(dir: impl AsRef<Path>) -> 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<Path>) -> 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<Path>, disk_permits: Arc<Semaphore>) -> 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<Semaphore> {
+        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);