From fee6d5df50c43b21bca23a6472c0ae35811428ae Mon Sep 17 00:00:00 2001 From: houseme Date: Wed, 19 Aug 2026 01:31:08 +0800 Subject: [PATCH] feat(ecstore): add dst dir fsync group commit Co-Authored-By: heihutu --- .config/nextest.toml | 11 + crates/ecstore/src/disk/local.rs | 72 ++- crates/ecstore/src/disk/os.rs | 762 ++++++++++++++++++++++++++++++- 3 files changed, 841 insertions(+), 4 deletions(-) diff --git a/.config/nextest.toml b/.config/nextest.toml index ade76a58c..4dd2d1d0e 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -87,6 +87,13 @@ test-group = 'ecstore-serial-flaky' filter = 'package(rustfs-ecstore) & (test(bucket_delete_waits_for_config_mutation_fence) | test(stale_config_request_cannot_mutate_a_recreated_bucket) | test(disk_incarnation_read_detects_stale_cache_until_peer_reload) | test(lifecycle_expiry_fails_closed_on_corrupt_object_lock_metadata) | test(expiry_configs_are_resolved_from_the_owning_store))' test-group = 'ecstore-serial-flaky' +# Serialize the default-off dst-dir fsync group-commit tests. They use +# process-global test hooks/registry to deterministically freeze fsync batches; +# no retries, just one at a time under nextest too. +[[profile.default.overrides]] +filter = 'package(rustfs-ecstore) & test(dst_dir_fsync_group_commit)' +test-group = 'ecstore-serial-flaky' + # Serialize the 4-disk reliability / degraded-read e2e tests (see the # e2e-reliability test-group note above). The matching ci-profile override is at # the end of the file, after [profile.ci] is declared. @@ -188,6 +195,10 @@ test-group = 'ecstore-serial-flaky' filter = 'package(rustfs-ecstore) & (test(bucket_delete_waits_for_config_mutation_fence) | test(stale_config_request_cannot_mutate_a_recreated_bucket) | test(disk_incarnation_read_detects_stale_cache_until_peer_reload) | test(lifecycle_expiry_fails_closed_on_corrupt_object_lock_metadata) | test(expiry_configs_are_resolved_from_the_owning_store))' test-group = 'ecstore-serial-flaky' +[[profile.ci.overrides]] +filter = 'package(rustfs-ecstore) & test(dst_dir_fsync_group_commit)' +test-group = 'ecstore-serial-flaky' + # --------------------------------------------------------------------------- # e2e-smoke profile — PR smoke subset of the e2e_test crate (backlog#1149 ci-4) # --------------------------------------------------------------------------- diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index 273ada7d7..285a12ba1 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/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. diff --git a/crates/ecstore/src/disk/os.rs b/crates/ecstore/src/disk/os.rs index decb36964..71ce809cd 100644 --- a/crates/ecstore/src/disk/os.rs +++ b/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> = Mutex::new(Vec::new()); static LIMITED: Mutex> = Mutex::new(Vec::new()); + static GROUPED: Mutex> = Mutex::new(Vec::new()); static BEFORE_LIMITED: std::sync::LazyLock>> = std::sync::LazyLock::new(|| Mutex::new(HashMap::new())); + static BEFORE_GROUP_BATCH: std::sync::LazyLock>> = + std::sync::LazyLock::new(|| Mutex::new(HashMap::new())); + static AFTER_GROUP_ENQUEUE: std::sync::LazyLock>> = + std::sync::LazyLock::new(|| Mutex::new(HashMap::new())); + static BEFORE_GROUPED: std::sync::LazyLock>> = + std::sync::LazyLock::new(|| Mutex::new(HashMap::new())); + static GROUPED_FAILURES: std::sync::LazyLock>> = + std::sync::LazyLock::new(|| Mutex::new(HashMap::new())); fn record_path(paths: &Mutex>, 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(entries: &Mutex>, dir: &Path, description: &str) -> Option { + 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(®istered_canonical) + }) + }) + .cloned(); + matching_key.and_then(|key| entries.remove(&key)) + } + + fn remove_hook(hooks: &Mutex>, dir: &Path, description: &str) -> Option { + 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 { + 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 { + 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) -> 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 = 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> = RwLock::new(None); + static SERIAL: Mutex<()> = Mutex::new(()); + + pub(crate) fn get() -> Option { + *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 { + 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, +} + +impl OpenedDstDirFsyncGroup { + fn open(dir: &Path) -> io::Result { + 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, +} + +#[derive(Clone)] +struct SharedDstDirFsyncError { + kind: io::ErrorKind, + message: Arc, +} + +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, + inner: Mutex, +} + +#[derive(Default)] +struct DstDirFsyncGroupInner { + worker_running: bool, + pending: VecDeque, +} + +#[derive(Default)] +struct DstDirFsyncGroupCommit { + inner: Mutex, +} + +#[derive(Default)] +struct DstDirFsyncGroupCommitInner { + groups: HashMap>, + total_waiters: usize, +} + +static DST_DIR_FSYNC_GROUP_COMMIT: LazyLock = 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, Option>)> { + 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) { + 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, Option>)> { + 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) { + loop { + #[cfg(test)] + fsync_dir_recorder::run_before_group_batch(&group.dir); + tokio::task::yield_now().await; + let batch: Vec = { + 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, 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) -> 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, 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");