feat(ecstore): add dst dir fsync group commit

Co-Authored-By: heihutu <heihutu@gmail.com>

crates/ecstore/src/disk/local.rs

@@ -9405,7 +9405,7 @@ impl DiskAPI for LocalDisk {
                 && let Some(parent) = dst_file_path.parent()
             {
                 let fsync_started = rustfs_io_metrics::put_stage_timer();
-                if let Err(err) = os::fsync_dir(parent).await {
+                if let Err(err) = os::fsync_dst_dir_group_commit(parent).await {
                     rustfs_io_metrics::record_put_object_stage_duration_from(
                         rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_DST_DIR_FSYNC,
                         fsync_started,
@@ -13022,6 +13022,76 @@ mod test {
         );
     }
 
+    #[tokio::test]
+    #[serial_test::serial(dst_dir_fsync_group_commit)]
+    async fn rename_data_non_inline_uses_dst_dir_fsync_group_commit_when_enabled() {
+        let _group_commit = os::set_dst_dir_fsync_group_commit_for_test(true);
+        let bucket = "grouped-dst-fsync-bucket";
+        let object = "dir/object";
+        let (disk, _dir) = commit_new_object(DurabilityMode::Strict, bucket, object).await;
+        let dst_meta_parent = disk
+            .get_object_path(bucket, &format!("{object}/{STORAGE_FORMAT_FILE}"))
+            .expect("dst meta path should resolve")
+            .parent()
+            .expect("dst meta should have a parent")
+            .to_path_buf();
+
+        assert_eq!(
+            os::fsync_dir_recorder::grouped_batch_sizes(&dst_meta_parent),
+            vec![1],
+            "enabled non-inline rename_data must route the dst parent fsync through the group commit coordinator"
+        );
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(dst_dir_fsync_group_commit)]
+    async fn rename_data_non_inline_dst_dir_fsync_group_commit_failure_rolls_back_fresh_put() {
+        use tempfile::tempdir;
+
+        let _group_commit = os::set_dst_dir_fsync_group_commit_for_test(true);
+        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 = LocalDisk::new(&endpoint, false).await.expect("local disk should be created");
+        let bucket = "grouped-dst-fsync-failure-bucket";
+        let object = "dir/object";
+        let tmp_object = "tmp-grouped-dst-fsync-failure";
+        let version_id = Uuid::parse_str("99999999-9999-9999-9999-999999999999").expect("version id should parse");
+        let new_data_dir = Uuid::parse_str("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa").expect("data dir should parse");
+        ensure_test_volume(&disk, bucket).await;
+        ensure_test_volume(&disk, RUSTFS_META_TMP_BUCKET).await;
+
+        let tmp_data_dir = dir
+            .path()
+            .join(RUSTFS_META_TMP_BUCKET)
+            .join(tmp_object)
+            .join(new_data_dir.to_string());
+        fs::create_dir_all(&tmp_data_dir)
+            .await
+            .expect("new tmp data dir should be created");
+        fs::write(tmp_data_dir.join("part.1"), b"new-data")
+            .await
+            .expect("new tmp data should be written");
+        let dst_meta_parent = dir.path().join(bucket).join(object);
+        os::fsync_dir_recorder::set_grouped_failure(&dst_meta_parent, io::ErrorKind::PermissionDenied);
+
+        let new_fi = test_file_info(object, version_id, Some(new_data_dir), None);
+        let err = disk
+            .rename_data(RUSTFS_META_TMP_BUCKET, tmp_object, new_fi, bucket, object)
+            .await
+            .expect_err("grouped dst dir fsync failure must fail the fresh PUT");
+
+        assert_eq!(err, DiskError::FileAccessDenied);
+        assert!(
+            !dst_meta_parent.join(STORAGE_FORMAT_FILE).exists(),
+            "fresh PUT rollback must remove the committed xl.meta after grouped dst dir fsync failure"
+        );
+        assert!(
+            !dst_meta_parent.join(new_data_dir.to_string()).exists(),
+            "fresh PUT rollback must remove the committed data dir after grouped dst dir fsync failure"
+        );
+    }
+
     // Seed a first PUT of `object` (no prior version) through the non-inline
     // rename_data path and return (disk, tempdir). The object dir and any prefix
     // dirs are created during the commit.

crates/ecstore/src/disk/os.rs

@@ -19,14 +19,14 @@ use futures::TryStreamExt;
 use parking_lot::Mutex;
 use rustfs_utils::path::SLASH_SEPARATOR;
 use std::{
-    collections::HashMap,
+    collections::{HashMap, VecDeque},
     io,
     path::{Component, Path, PathBuf},
     sync::{Arc, LazyLock, Weak},
 };
 use tokio::fs;
 use tokio::sync::{
-    Mutex as AsyncMutex, OwnedMutexGuard, OwnedRwLockReadGuard, OwnedSemaphorePermit, RwLock, Semaphore, SemaphorePermit,
+    Mutex as AsyncMutex, OwnedMutexGuard, OwnedRwLockReadGuard, OwnedSemaphorePermit, RwLock, Semaphore, SemaphorePermit, oneshot,
 };
 use tracing::warn;
 
@@ -79,6 +79,7 @@ pub fn check_path_length(path_name: &str) -> Result<()> {
 #[cfg(test)]
 pub(crate) mod fsync_dir_recorder {
     use std::collections::HashMap;
+    use std::io;
     use std::path::{Path, PathBuf};
     use std::sync::Mutex;
 
@@ -86,8 +87,17 @@ pub(crate) mod fsync_dir_recorder {
 
     static RECORDED: Mutex<Vec<PathBuf>> = Mutex::new(Vec::new());
     static LIMITED: Mutex<Vec<PathBuf>> = Mutex::new(Vec::new());
+    static GROUPED: Mutex<Vec<(PathBuf, usize)>> = Mutex::new(Vec::new());
     static BEFORE_LIMITED: std::sync::LazyLock<Mutex<HashMap<PathBuf, Hook>>> =
         std::sync::LazyLock::new(|| Mutex::new(HashMap::new()));
+    static BEFORE_GROUP_BATCH: std::sync::LazyLock<Mutex<HashMap<PathBuf, Hook>>> =
+        std::sync::LazyLock::new(|| Mutex::new(HashMap::new()));
+    static AFTER_GROUP_ENQUEUE: std::sync::LazyLock<Mutex<HashMap<PathBuf, Hook>>> =
+        std::sync::LazyLock::new(|| Mutex::new(HashMap::new()));
+    static BEFORE_GROUPED: std::sync::LazyLock<Mutex<HashMap<PathBuf, Hook>>> =
+        std::sync::LazyLock::new(|| Mutex::new(HashMap::new()));
+    static GROUPED_FAILURES: std::sync::LazyLock<Mutex<HashMap<PathBuf, io::ErrorKind>>> =
+        std::sync::LazyLock::new(|| Mutex::new(HashMap::new()));
 
     fn record_path(paths: &Mutex<Vec<PathBuf>>, path: &Path, description: &str) {
         let mut paths = paths.lock().expect(description);
@@ -106,6 +116,29 @@ pub(crate) mod fsync_dir_recorder {
             .any(|recorded| recorded == path || canonical.as_ref().is_some_and(|canonical| recorded == canonical))
     }
 
+    fn remove_path_keyed<T>(entries: &Mutex<HashMap<PathBuf, T>>, dir: &Path, description: &str) -> Option<T> {
+        let mut entries = entries.lock().expect(description);
+        if let Some(value) = entries.remove(dir) {
+            return Some(value);
+        }
+        let canonical = dir.canonicalize().ok();
+        let matching_key = entries
+            .keys()
+            .find(|registered| {
+                registered.as_path() == dir
+                    || canonical.as_ref().is_some_and(|canonical| *registered == canonical)
+                    || registered.canonicalize().ok().is_some_and(|registered_canonical| {
+                        registered_canonical == dir || canonical.as_ref() == Some(&registered_canonical)
+                    })
+            })
+            .cloned();
+        matching_key.and_then(|key| entries.remove(&key))
+    }
+
+    fn remove_hook(hooks: &Mutex<HashMap<PathBuf, Hook>>, dir: &Path, description: &str) -> Option<Hook> {
+        remove_path_keyed(hooks, dir, description)
+    }
+
     pub(crate) fn record(dir: &Path) {
         record_path(&RECORDED, dir, "fsync dir recorder");
     }
@@ -116,7 +149,7 @@ pub(crate) mod fsync_dir_recorder {
 
     pub(crate) fn record_limited(dir: &Path) {
         record_path(&LIMITED, dir, "limited fsync dir recorder");
-        let hook = BEFORE_LIMITED.lock().expect("limited fsync hook poisoned").remove(dir);
+        let hook = remove_hook(&BEFORE_LIMITED, dir, "limited fsync hook poisoned");
         if let Some(hook) = hook {
             hook();
         }
@@ -132,6 +165,78 @@ pub(crate) mod fsync_dir_recorder {
             .expect("limited fsync hook poisoned")
             .insert(dir.to_path_buf(), Box::new(hook));
     }
+
+    pub(crate) fn record_grouped(dir: &Path, batch_len: usize) {
+        let mut grouped = GROUPED.lock().expect("grouped fsync dir recorder poisoned");
+        grouped.push((dir.to_path_buf(), batch_len));
+        if let Ok(canonical) = dir.canonicalize()
+            && canonical != dir
+        {
+            grouped.push((canonical, batch_len));
+        }
+        drop(grouped);
+        let hook = remove_hook(&BEFORE_GROUPED, dir, "grouped fsync hook poisoned");
+        if let Some(hook) = hook {
+            hook();
+        }
+    }
+
+    pub(crate) fn run_before_group_batch(dir: &Path) {
+        let hook = remove_hook(&BEFORE_GROUP_BATCH, dir, "grouped fsync batch hook poisoned");
+        if let Some(hook) = hook {
+            hook();
+        }
+    }
+
+    pub(crate) fn set_before_group_batch(dir: &Path, hook: impl FnOnce() + Send + 'static) {
+        BEFORE_GROUP_BATCH
+            .lock()
+            .expect("grouped fsync batch hook poisoned")
+            .insert(dir.to_path_buf(), Box::new(hook));
+    }
+
+    pub(crate) fn run_after_group_enqueue(dir: &Path) {
+        let hook = remove_hook(&AFTER_GROUP_ENQUEUE, dir, "grouped fsync enqueue hook poisoned");
+        if let Some(hook) = hook {
+            hook();
+        }
+    }
+
+    pub(crate) fn set_after_group_enqueue(dir: &Path, hook: impl FnOnce() + Send + 'static) {
+        AFTER_GROUP_ENQUEUE
+            .lock()
+            .expect("grouped fsync enqueue hook poisoned")
+            .insert(dir.to_path_buf(), Box::new(hook));
+    }
+
+    pub(crate) fn grouped_batch_sizes(dir: &Path) -> Vec<usize> {
+        let grouped = GROUPED.lock().expect("grouped fsync dir recorder poisoned");
+        let canonical = dir.canonicalize().ok();
+        grouped
+            .iter()
+            .filter_map(|(recorded, batch_len)| {
+                (recorded == dir || canonical.as_ref().is_some_and(|canonical| recorded == canonical)).then_some(*batch_len)
+            })
+            .collect()
+    }
+
+    pub(crate) fn set_before_grouped(dir: &Path, hook: impl FnOnce() + Send + 'static) {
+        BEFORE_GROUPED
+            .lock()
+            .expect("grouped fsync hook poisoned")
+            .insert(dir.to_path_buf(), Box::new(hook));
+    }
+
+    pub(crate) fn set_grouped_failure(dir: &Path, kind: io::ErrorKind) {
+        GROUPED_FAILURES
+            .lock()
+            .expect("grouped fsync failure hook poisoned")
+            .insert(dir.to_path_buf(), kind);
+    }
+
+    pub(crate) fn take_grouped_failure(dir: &Path) -> Option<io::ErrorKind> {
+        remove_path_keyed(&GROUPED_FAILURES, dir, "grouped fsync failure hook poisoned")
+    }
 }
 
 #[cfg(all(test, windows))]
@@ -218,6 +323,375 @@ pub async fn fsync_dir(dir: impl AsRef<Path>) -> io::Result<()> {
     }
 }
 
+const ENV_DST_DIR_FSYNC_GROUP_COMMIT_ENABLE: &str = "RUSTFS_EXPERIMENTAL_DST_DIR_FSYNC_GROUP_COMMIT_ENABLE";
+const DEFAULT_DST_DIR_FSYNC_GROUP_COMMIT_ENABLE: bool = false;
+#[cfg(not(test))]
+const MAX_DST_DIR_FSYNC_GROUPS: usize = 1024;
+#[cfg(test)]
+const MAX_DST_DIR_FSYNC_GROUPS: usize = 4;
+#[cfg(not(test))]
+const MAX_DST_DIR_FSYNC_WAITERS: usize = 8192;
+#[cfg(test)]
+const MAX_DST_DIR_FSYNC_WAITERS: usize = 8;
+static DST_DIR_FSYNC_GROUP_COMMIT_ENABLED: LazyLock<bool> = LazyLock::new(|| {
+    rustfs_utils::get_env_bool(ENV_DST_DIR_FSYNC_GROUP_COMMIT_ENABLE, DEFAULT_DST_DIR_FSYNC_GROUP_COMMIT_ENABLE)
+});
+
+#[cfg(test)]
+mod dst_dir_fsync_group_commit_override {
+    use std::sync::{Mutex, MutexGuard, PoisonError, RwLock};
+
+    static OVERRIDE: RwLock<Option<bool>> = RwLock::new(None);
+    static SERIAL: Mutex<()> = Mutex::new(());
+
+    pub(crate) fn get() -> Option<bool> {
+        *OVERRIDE.read().unwrap_or_else(PoisonError::into_inner)
+    }
+
+    pub(crate) struct OverrideGuard {
+        _serial: MutexGuard<'static, ()>,
+    }
+
+    impl Drop for OverrideGuard {
+        fn drop(&mut self) {
+            *OVERRIDE.write().unwrap_or_else(PoisonError::into_inner) = None;
+        }
+    }
+
+    pub(crate) fn set(enabled: bool) -> OverrideGuard {
+        let serial = SERIAL.lock().unwrap_or_else(PoisonError::into_inner);
+        *OVERRIDE.write().unwrap_or_else(PoisonError::into_inner) = Some(enabled);
+        OverrideGuard { _serial: serial }
+    }
+}
+
+#[cfg(test)]
+pub(crate) fn set_dst_dir_fsync_group_commit_for_test(enabled: bool) -> dst_dir_fsync_group_commit_override::OverrideGuard {
+    dst_dir_fsync_group_commit_override::set(enabled)
+}
+
+fn dst_dir_fsync_group_commit_enabled() -> bool {
+    #[cfg(test)]
+    if let Some(enabled) = dst_dir_fsync_group_commit_override::get() {
+        return enabled;
+    }
+
+    *DST_DIR_FSYNC_GROUP_COMMIT_ENABLED
+}
+
+#[derive(Clone, Eq, Hash, PartialEq)]
+struct DstDirFsyncGroupKey {
+    canonical_path: PathBuf,
+    #[cfg(unix)]
+    dev: u64,
+    #[cfg(unix)]
+    ino: u64,
+}
+
+impl DstDirFsyncGroupKey {
+    fn from_metadata(canonical_path: PathBuf, metadata: std::fs::Metadata) -> io::Result<Self> {
+        if !metadata.is_dir() {
+            return Err(io::Error::new(io::ErrorKind::InvalidInput, "dst dir fsync group key must be a directory"));
+        }
+        #[cfg(unix)]
+        {
+            use std::os::unix::fs::MetadataExt;
+            Ok(Self {
+                canonical_path,
+                dev: metadata.dev(),
+                ino: metadata.ino(),
+            })
+        }
+        #[cfg(not(unix))]
+        {
+            Ok(Self { canonical_path })
+        }
+    }
+}
+
+struct OpenedDstDirFsyncGroup {
+    key: DstDirFsyncGroupKey,
+    #[cfg(any(test, not(unix)))]
+    dir: PathBuf,
+    #[cfg(unix)]
+    dir_file: Arc<std::fs::File>,
+}
+
+impl OpenedDstDirFsyncGroup {
+    fn open(dir: &Path) -> io::Result<Self> {
+        let canonical_path = dir.canonicalize()?;
+        #[cfg(unix)]
+        {
+            let file = std::fs::File::open(&canonical_path)?;
+            let key = DstDirFsyncGroupKey::from_metadata(canonical_path, file.metadata()?)?;
+            #[cfg(test)]
+            let dir = key.canonical_path.clone();
+            Ok(Self {
+                key,
+                #[cfg(test)]
+                dir,
+                dir_file: Arc::new(file),
+            })
+        }
+        #[cfg(not(unix))]
+        {
+            let key = DstDirFsyncGroupKey::from_metadata(canonical_path.clone(), std::fs::metadata(&canonical_path)?)?;
+            Ok(Self {
+                key,
+                dir: canonical_path,
+            })
+        }
+    }
+}
+
+struct DstDirFsyncWaiter {
+    result_tx: oneshot::Sender<SharedDstDirFsyncResult>,
+}
+
+#[derive(Clone)]
+struct SharedDstDirFsyncError {
+    kind: io::ErrorKind,
+    message: Arc<str>,
+}
+
+impl SharedDstDirFsyncError {
+    fn from_error(err: io::Error) -> Self {
+        Self {
+            kind: err.kind(),
+            message: Arc::from(err.to_string()),
+        }
+    }
+
+    fn into_error(self) -> io::Error {
+        io::Error::new(self.kind, self.message.to_string())
+    }
+}
+
+type SharedDstDirFsyncResult = std::result::Result<(), SharedDstDirFsyncError>;
+
+struct DstDirFsyncGroup {
+    key: DstDirFsyncGroupKey,
+    #[cfg(any(test, not(unix)))]
+    dir: PathBuf,
+    #[cfg(unix)]
+    dir_file: Arc<std::fs::File>,
+    inner: Mutex<DstDirFsyncGroupInner>,
+}
+
+#[derive(Default)]
+struct DstDirFsyncGroupInner {
+    worker_running: bool,
+    pending: VecDeque<DstDirFsyncWaiter>,
+}
+
+#[derive(Default)]
+struct DstDirFsyncGroupCommit {
+    inner: Mutex<DstDirFsyncGroupCommitInner>,
+}
+
+#[derive(Default)]
+struct DstDirFsyncGroupCommitInner {
+    groups: HashMap<DstDirFsyncGroupKey, Arc<DstDirFsyncGroup>>,
+    total_waiters: usize,
+}
+
+static DST_DIR_FSYNC_GROUP_COMMIT: LazyLock<DstDirFsyncGroupCommit> = LazyLock::new(DstDirFsyncGroupCommit::default);
+
+impl DstDirFsyncGroupCommit {
+    // Lock order: registry first, then per-group state. No path may hold a
+    // group lock while acquiring the registry lock.
+    fn enqueue_opened(
+        &self,
+        opened: OpenedDstDirFsyncGroup,
+    ) -> io::Result<(oneshot::Receiver<SharedDstDirFsyncResult>, Option<Arc<DstDirFsyncGroup>>)> {
+        let (result_tx, result_rx) = oneshot::channel();
+        let mut registry = self.inner.lock();
+        if registry.total_waiters >= MAX_DST_DIR_FSYNC_WAITERS {
+            return Err(io::Error::new(
+                io::ErrorKind::WouldBlock,
+                "dst dir fsync group commit waiter limit reached",
+            ));
+        }
+        let group = if let Some(group) = registry.groups.get(&opened.key) {
+            group.clone()
+        } else {
+            if registry.groups.len() >= MAX_DST_DIR_FSYNC_GROUPS {
+                return Err(io::Error::new(
+                    io::ErrorKind::WouldBlock,
+                    "dst dir fsync group commit active group limit reached",
+                ));
+            }
+            let group = Arc::new(DstDirFsyncGroup {
+                key: opened.key.clone(),
+                #[cfg(any(test, not(unix)))]
+                dir: opened.dir,
+                #[cfg(unix)]
+                dir_file: opened.dir_file,
+                inner: Mutex::new(DstDirFsyncGroupInner::default()),
+            });
+            registry.groups.insert(opened.key, group.clone());
+            group
+        };
+        let mut group_state = group.inner.lock();
+        group_state.pending.push_back(DstDirFsyncWaiter { result_tx });
+        let start_worker = !group_state.worker_running;
+        if start_worker {
+            group_state.worker_running = true;
+        }
+        registry.total_waiters += 1;
+        drop(group_state);
+        drop(registry);
+        #[cfg(test)]
+        fsync_dir_recorder::run_after_group_enqueue(&group.dir);
+
+        Ok((result_rx, start_worker.then_some(group)))
+    }
+
+    fn complete_batch(&self, count: usize) {
+        let mut registry = self.inner.lock();
+        registry.total_waiters = registry.total_waiters.saturating_sub(count);
+    }
+
+    fn remove_idle_group(&self, group: &Arc<DstDirFsyncGroup>) {
+        let mut registry = self.inner.lock();
+        let group_state = group.inner.lock();
+        if !group_state.worker_running && group_state.pending.is_empty() {
+            registry.groups.remove(&group.key);
+        }
+    }
+
+    #[cfg(test)]
+    fn counts_for_test(&self) -> (usize, usize) {
+        let registry = self.inner.lock();
+        (registry.groups.len(), registry.total_waiters)
+    }
+
+    #[cfg(test)]
+    fn clear_for_test(&self) {
+        let mut registry = self.inner.lock();
+        registry.groups.clear();
+        registry.total_waiters = 0;
+    }
+
+    #[cfg(test)]
+    fn enqueue_for_test(
+        &self,
+        dir: &Path,
+    ) -> io::Result<(oneshot::Receiver<SharedDstDirFsyncResult>, Option<Arc<DstDirFsyncGroup>>)> {
+        self.enqueue_opened(OpenedDstDirFsyncGroup::open(dir)?)
+    }
+}
+
+#[cfg(unix)]
+async fn fsync_open_dst_dir_group(group: &DstDirFsyncGroup) -> io::Result<()> {
+    #[cfg(test)]
+    let dir = group.dir.clone();
+    let dir_file = group.dir_file.clone();
+    tokio::task::spawn_blocking(move || {
+        #[cfg(test)]
+        {
+            if let Some(kind) = fsync_dir_recorder::take_grouped_failure(&dir) {
+                return Err(io::Error::new(kind, "injected grouped dst dir fsync failure"));
+            }
+            fsync_dir_recorder::record(&dir);
+        }
+        dir_file.sync_all()
+    })
+    .await
+    .map_err(|err| io::Error::other(format!("blocking dst dir group fsync failed: {err}")))?
+}
+
+#[cfg(not(unix))]
+async fn fsync_open_dst_dir_group(group: &DstDirFsyncGroup) -> io::Result<()> {
+    fsync_dir(&group.dir).await
+}
+
+async fn run_dst_dir_fsync_group_worker(group: Arc<DstDirFsyncGroup>) {
+    loop {
+        #[cfg(test)]
+        fsync_dir_recorder::run_before_group_batch(&group.dir);
+        tokio::task::yield_now().await;
+        let batch: Vec<DstDirFsyncWaiter> = {
+            let mut group_state = group.inner.lock();
+            group_state.pending.drain(..).collect()
+        };
+        if batch.is_empty() {
+            let mut group_state = group.inner.lock();
+            group_state.worker_running = false;
+            drop(group_state);
+            DST_DIR_FSYNC_GROUP_COMMIT.remove_idle_group(&group);
+            return;
+        }
+
+        #[cfg(test)]
+        fsync_dir_recorder::record_grouped(&group.dir, batch.len());
+        let result = fsync_open_dst_dir_group(&group)
+            .await
+            .map_err(SharedDstDirFsyncError::from_error);
+        let batch_len = batch.len();
+        DST_DIR_FSYNC_GROUP_COMMIT.complete_batch(batch_len);
+
+        let should_stop = {
+            let mut group_state = group.inner.lock();
+            if group_state.pending.is_empty() {
+                group_state.worker_running = false;
+                true
+            } else {
+                false
+            }
+        };
+        if should_stop {
+            DST_DIR_FSYNC_GROUP_COMMIT.remove_idle_group(&group);
+        }
+        for waiter in batch {
+            let _ = waiter.result_tx.send(result.clone());
+        }
+        if should_stop {
+            return;
+        }
+    }
+}
+
+async fn fsync_dst_dir_group_commit_with_enabled(dir: impl AsRef<Path>, enabled: bool) -> io::Result<()> {
+    if !enabled {
+        return fsync_dir(dir).await;
+    }
+
+    let dir = dir.as_ref().to_path_buf();
+    let opened = tokio::task::spawn_blocking(move || OpenedDstDirFsyncGroup::open(&dir))
+        .await
+        .map_err(|err| io::Error::other(format!("blocking dst dir group open failed: {err}")))??;
+    let (result_rx, worker) = DST_DIR_FSYNC_GROUP_COMMIT.enqueue_opened(opened)?;
+    if let Some(group) = worker {
+        tokio::spawn(run_dst_dir_fsync_group_worker(group));
+    }
+
+    match result_rx.await {
+        Ok(Ok(())) => Ok(()),
+        Ok(Err(err)) => Err(err.into_error()),
+        Err(_) => Err(io::Error::other("dst dir fsync group worker dropped the waiter")),
+    }
+}
+
+pub(crate) async fn fsync_dst_dir_group_commit(dir: impl AsRef<Path>) -> io::Result<()> {
+    fsync_dst_dir_group_commit_with_enabled(dir, dst_dir_fsync_group_commit_enabled()).await
+}
+
+#[cfg(test)]
+pub(crate) async fn fsync_dst_dir_group_commit_for_test(dir: impl AsRef<Path>, enabled: bool) -> io::Result<()> {
+    fsync_dst_dir_group_commit_with_enabled(dir, enabled).await
+}
+
+#[cfg(test)]
+pub(crate) fn dst_dir_fsync_group_commit_counts_for_test() -> (usize, usize) {
+    DST_DIR_FSYNC_GROUP_COMMIT.counts_for_test()
+}
+
+#[cfg(test)]
+fn clear_dst_dir_fsync_group_commit_for_test() {
+    DST_DIR_FSYNC_GROUP_COMMIT.clear_for_test();
+}
+
 // 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;
@@ -4750,6 +5224,288 @@ mod tests {
         fsync_dir(temp_dir.path()).await.expect("fsync dir must succeed");
     }
 
+    async fn wait_for_dst_dir_fsync_group_commit_idle() {
+        for _ in 0..100 {
+            if dst_dir_fsync_group_commit_counts_for_test() == (0, 0) {
+                return;
+            }
+            tokio::task::yield_now().await;
+        }
+        assert_eq!(
+            dst_dir_fsync_group_commit_counts_for_test(),
+            (0, 0),
+            "dst dir fsync group registry must release idle groups and waiters"
+        );
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    #[serial_test::serial(dst_dir_fsync_group_commit)]
+    async fn dst_dir_fsync_group_commit_default_off_uses_direct_fsync() {
+        let temp_dir = tempdir().expect("create temp dir");
+        let dir = temp_dir.path().join("object");
+        std::fs::create_dir(&dir).expect("create object dir");
+
+        fsync_dst_dir_group_commit_for_test(&dir, false)
+            .await
+            .expect("direct dst dir fsync should succeed");
+
+        assert!(fsync_dir_recorder::was_fsynced(&dir), "default-off path must still fsync the dst dir");
+        assert!(
+            fsync_dir_recorder::grouped_batch_sizes(&dir).is_empty(),
+            "default-off path must not enter the group commit coordinator"
+        );
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    #[serial_test::serial(dst_dir_fsync_group_commit)]
+    async fn dst_dir_fsync_group_commit_batches_same_directory_waiters() {
+        use std::sync::mpsc;
+        let temp_dir = tempdir().expect("create temp dir");
+        let dir = temp_dir.path().join("object");
+        std::fs::create_dir(&dir).expect("create object dir");
+        let (batch_entered_tx, batch_entered_rx) = mpsc::channel();
+        let (release_batch_tx, release_batch_rx) = mpsc::channel();
+        fsync_dir_recorder::set_before_group_batch(&dir, move || {
+            batch_entered_tx.send(()).expect("signal first worker before freezing batch");
+            release_batch_rx.recv().expect("wait until second waiter is queued");
+        });
+
+        let first_dir = dir.clone();
+        let first = tokio::spawn(async move { fsync_dst_dir_group_commit_for_test(first_dir, true).await });
+        tokio::task::spawn_blocking(move || batch_entered_rx.recv_timeout(Duration::from_secs(30)))
+            .await
+            .expect("batch hook waiter should run")
+            .expect("first worker should reach the batch hook");
+
+        let (second_enqueued_tx, second_enqueued_rx) = mpsc::channel();
+        fsync_dir_recorder::set_after_group_enqueue(&dir, move || {
+            second_enqueued_tx.send(()).expect("signal second waiter enqueue");
+        });
+        let second_dir = dir.clone();
+        let second = tokio::spawn(async move { fsync_dst_dir_group_commit_for_test(second_dir, true).await });
+        tokio::task::spawn_blocking(move || second_enqueued_rx.recv_timeout(Duration::from_secs(30)))
+            .await
+            .expect("enqueue hook waiter should run")
+            .expect("second waiter should be enqueued");
+        assert_eq!(
+            dst_dir_fsync_group_commit_counts_for_test(),
+            (1, 2),
+            "second waiter must be queued before the first batch is released"
+        );
+        release_batch_tx.send(()).expect("release first batch");
+
+        let (first_result, second_result) = tokio::time::timeout(Duration::from_secs(30), async { tokio::join!(first, second) })
+            .await
+            .expect("same-directory fsync waiters should complete");
+        first_result
+            .expect("first waiter task should not panic")
+            .expect("first waiter should observe successful fsync");
+        second_result
+            .expect("second waiter task should not panic")
+            .expect("second waiter should observe successful fsync");
+
+        assert_eq!(
+            fsync_dir_recorder::grouped_batch_sizes(&dir),
+            vec![2],
+            "two waiters queued before the batch freezes must share exactly one dst dir fsync"
+        );
+        wait_for_dst_dir_fsync_group_commit_idle().await;
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    #[serial_test::serial(dst_dir_fsync_group_commit)]
+    async fn dst_dir_fsync_group_commit_late_join_waits_for_next_fsync() {
+        use std::sync::mpsc;
+        let temp_dir = tempdir().expect("create temp dir");
+        let dir = temp_dir.path().join("object");
+        std::fs::create_dir(&dir).expect("create object dir");
+        let (fsync_entered_tx, fsync_entered_rx) = mpsc::channel();
+        let (release_fsync_tx, release_fsync_rx) = mpsc::channel();
+        fsync_dir_recorder::set_before_grouped(&dir, move || {
+            fsync_entered_tx.send(()).expect("signal first frozen batch");
+            release_fsync_rx.recv().expect("wait until late waiter is queued");
+        });
+
+        let first_dir = dir.clone();
+        let first = tokio::spawn(async move { fsync_dst_dir_group_commit_for_test(first_dir, true).await });
+        tokio::task::spawn_blocking(move || fsync_entered_rx.recv_timeout(Duration::from_secs(30)))
+            .await
+            .expect("grouped fsync hook waiter should run")
+            .expect("first batch should reach fsync");
+
+        let second_dir = dir.clone();
+        let second = tokio::spawn(async move { fsync_dst_dir_group_commit_for_test(second_dir, true).await });
+        release_fsync_tx.send(()).expect("release first fsync");
+
+        let (first_result, second_result) = tokio::time::timeout(Duration::from_secs(30), async { tokio::join!(first, second) })
+            .await
+            .expect("late waiter should complete after a second fsync");
+        first_result
+            .expect("first waiter task should not panic")
+            .expect("first waiter should observe successful fsync");
+        second_result
+            .expect("second waiter task should not panic")
+            .expect("late waiter should observe successful fsync");
+
+        assert_eq!(
+            fsync_dir_recorder::grouped_batch_sizes(&dir),
+            vec![1, 1],
+            "a waiter queued after the first batch is frozen must not be covered by the earlier fsync"
+        );
+        wait_for_dst_dir_fsync_group_commit_idle().await;
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    #[serial_test::serial(dst_dir_fsync_group_commit)]
+    async fn dst_dir_fsync_group_commit_propagates_shared_fsync_failure() {
+        let temp_dir = tempdir().expect("create temp dir");
+        let dir = temp_dir.path().join("object");
+        std::fs::create_dir(&dir).expect("create object dir");
+        fsync_dir_recorder::set_grouped_failure(&dir, io::ErrorKind::Other);
+
+        let err = fsync_dst_dir_group_commit_for_test(&dir, true)
+            .await
+            .expect_err("shared dst dir fsync failure must be returned to the waiter");
+
+        assert_eq!(err.kind(), io::ErrorKind::Other);
+        wait_for_dst_dir_fsync_group_commit_idle().await;
+    }
+
+    #[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() {
+        use std::sync::mpsc;
+        let temp_dir = tempdir().expect("create temp dir");
+        let dir = temp_dir.path().join("object");
+        std::fs::create_dir(&dir).expect("create object dir");
+        let (fsync_entered_tx, fsync_entered_rx) = mpsc::channel();
+        let (release_fsync_tx, release_fsync_rx) = mpsc::channel();
+        fsync_dir_recorder::set_before_grouped(&dir, move || {
+            fsync_entered_tx.send(()).expect("signal grouped fsync");
+            release_fsync_rx.recv().expect("wait for cancellation");
+        });
+
+        let cancelled_dir = dir.clone();
+        let cancelled = tokio::spawn(async move { fsync_dst_dir_group_commit_for_test(cancelled_dir, true).await });
+        tokio::task::spawn_blocking(move || fsync_entered_rx.recv_timeout(Duration::from_secs(30)))
+            .await
+            .expect("grouped fsync hook waiter should run")
+            .expect("first grouped fsync should start");
+        cancelled.abort();
+        assert!(
+            cancelled
+                .await
+                .expect_err("cancelled waiter task should abort")
+                .is_cancelled(),
+            "waiter cancellation must be observable"
+        );
+        release_fsync_tx.send(()).expect("release grouped fsync");
+
+        fsync_dst_dir_group_commit_for_test(&dir, true)
+            .await
+            .expect("a later waiter should not be blocked by cancelled waiter state");
+        wait_for_dst_dir_fsync_group_commit_idle().await;
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    #[serial_test::serial(dst_dir_fsync_group_commit)]
+    async fn dst_dir_fsync_group_commit_recreated_directory_gets_new_group() {
+        use std::sync::mpsc;
+        let temp_dir = tempdir().expect("create temp dir");
+        let dir = temp_dir.path().join("object");
+        std::fs::create_dir(&dir).expect("create object dir");
+        let (fsync_entered_tx, fsync_entered_rx) = mpsc::channel();
+        let (release_fsync_tx, release_fsync_rx) = mpsc::channel();
+        let dir_for_hook = dir.clone();
+        fsync_dir_recorder::set_before_grouped(&dir, move || {
+            std::fs::remove_dir(&dir_for_hook).expect("remove old object dir");
+            std::fs::create_dir(&dir_for_hook).expect("recreate object dir at the same path");
+            fsync_entered_tx.send(()).expect("signal grouped fsync");
+            release_fsync_rx.recv().expect("wait until recreated dir is enqueued");
+        });
+
+        let first_dir = dir.clone();
+        let first = tokio::spawn(async move { fsync_dst_dir_group_commit_for_test(first_dir, true).await });
+        tokio::task::spawn_blocking(move || fsync_entered_rx.recv_timeout(Duration::from_secs(30)))
+            .await
+            .expect("grouped fsync hook waiter should run")
+            .expect("first grouped fsync should start");
+
+        let (_result_rx, worker) = DST_DIR_FSYNC_GROUP_COMMIT
+            .enqueue_for_test(&dir)
+            .expect("recreated dir should enqueue separately");
+        assert!(worker.is_some(), "same path with a new inode must not join the stale in-flight group");
+        assert_eq!(
+            dst_dir_fsync_group_commit_counts_for_test().0,
+            2,
+            "old and recreated directory identities must be tracked as separate active groups"
+        );
+
+        release_fsync_tx.send(()).expect("release grouped fsync");
+        first
+            .await
+            .expect("first waiter task should not panic")
+            .expect("first stale directory fd should still fsync successfully");
+        clear_dst_dir_fsync_group_commit_for_test();
+        assert_eq!(
+            dst_dir_fsync_group_commit_counts_for_test(),
+            (0, 0),
+            "test registry cleanup must release the unstarted recreated-directory waiter"
+        );
+    }
+
+    #[test]
+    #[serial_test::serial(dst_dir_fsync_group_commit)]
+    fn dst_dir_fsync_group_commit_rejects_active_group_overflow() {
+        let temp_dir = tempdir().expect("create temp dir");
+        let mut receivers = Vec::new();
+        for index in 0..MAX_DST_DIR_FSYNC_GROUPS {
+            let dir = temp_dir.path().join(format!("object-{index}"));
+            std::fs::create_dir(&dir).expect("create object dir");
+            let (result_rx, _worker) = DST_DIR_FSYNC_GROUP_COMMIT
+                .enqueue_for_test(&dir)
+                .expect("group below cap should enqueue");
+            receivers.push(result_rx);
+        }
+        let overflow_dir = temp_dir.path().join("overflow");
+        std::fs::create_dir(&overflow_dir).expect("create overflow dir");
+
+        let err = match DST_DIR_FSYNC_GROUP_COMMIT.enqueue_for_test(&overflow_dir) {
+            Ok(_) => panic!("active group max+1 must fail closed"),
+            Err(err) => err,
+        };
+
+        assert_eq!(err.kind(), io::ErrorKind::WouldBlock);
+        clear_dst_dir_fsync_group_commit_for_test();
+        assert_eq!(dst_dir_fsync_group_commit_counts_for_test(), (0, 0));
+        drop(receivers);
+    }
+
+    #[test]
+    #[serial_test::serial(dst_dir_fsync_group_commit)]
+    fn dst_dir_fsync_group_commit_rejects_waiter_overflow() {
+        let temp_dir = tempdir().expect("create temp dir");
+        let dir = temp_dir.path().join("object");
+        std::fs::create_dir(&dir).expect("create object dir");
+        let mut receivers = Vec::new();
+        for _ in 0..MAX_DST_DIR_FSYNC_WAITERS {
+            let (result_rx, _worker) = DST_DIR_FSYNC_GROUP_COMMIT
+                .enqueue_for_test(&dir)
+                .expect("waiter below cap should enqueue");
+            receivers.push(result_rx);
+        }
+
+        let err = match DST_DIR_FSYNC_GROUP_COMMIT.enqueue_for_test(&dir) {
+            Ok(_) => panic!("waiter max+1 must fail closed"),
+            Err(err) => err,
+        };
+
+        assert_eq!(err.kind(), io::ErrorKind::WouldBlock);
+        clear_dst_dir_fsync_group_commit_for_test();
+        assert_eq!(dst_dir_fsync_group_commit_counts_for_test(), (0, 0));
+        drop(receivers);
+    }
+
     #[tokio::test]
     async fn file_sync_admission_is_reused_across_commit_barriers() {
         let temp_dir = tempdir().expect("create temp dir");