feat(ecstore): tune fdatasync group wait budget

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

crates/ecstore/src/disk/os.rs

@@ -23,7 +23,7 @@ use std::{
     io,
     path::{Component, Path, PathBuf},
     sync::{Arc, LazyLock, Weak},
-    time::Instant,
+    time::{Duration, Instant},
 };
 use tokio::fs;
 use tokio::sync::{
@@ -328,6 +328,9 @@ const ENV_DST_DIR_FSYNC_GROUP_COMMIT_ENABLE: &str = "RUSTFS_EXPERIMENTAL_DST_DIR
 const DEFAULT_DST_DIR_FSYNC_GROUP_COMMIT_ENABLE: bool = false;
 const ENV_FILE_FDATASYNC_GROUP_COMMIT_ENABLE: &str = "RUSTFS_EXPERIMENTAL_FILE_FDATASYNC_GROUP_COMMIT_ENABLE";
 const DEFAULT_FILE_FDATASYNC_GROUP_COMMIT_ENABLE: bool = false;
+const ENV_FILE_FDATASYNC_GROUP_COMMIT_WAIT_MICROS: &str = "RUSTFS_EXPERIMENTAL_FILE_FDATASYNC_GROUP_COMMIT_WAIT_MICROS";
+const DEFAULT_FILE_FDATASYNC_GROUP_COMMIT_WAIT_MICROS: u64 = 0;
+const MAX_FILE_FDATASYNC_GROUP_COMMIT_WAIT_MICROS: u64 = 1_000;
 #[cfg(not(test))]
 const MAX_DST_DIR_FSYNC_GROUPS: usize = 1024;
 #[cfg(test)]
@@ -354,6 +357,15 @@ static DST_DIR_FSYNC_GROUP_COMMIT_ENABLED: LazyLock<bool> = LazyLock::new(|| {
 static FILE_FDATASYNC_GROUP_COMMIT_ENABLED: LazyLock<bool> = LazyLock::new(|| {
     rustfs_utils::get_env_bool(ENV_FILE_FDATASYNC_GROUP_COMMIT_ENABLE, DEFAULT_FILE_FDATASYNC_GROUP_COMMIT_ENABLE)
 });
+static FILE_FDATASYNC_GROUP_COMMIT_WAIT: LazyLock<Duration> = LazyLock::new(|| {
+    Duration::from_micros(
+        rustfs_utils::get_env_u64(
+            ENV_FILE_FDATASYNC_GROUP_COMMIT_WAIT_MICROS,
+            DEFAULT_FILE_FDATASYNC_GROUP_COMMIT_WAIT_MICROS,
+        )
+        .min(MAX_FILE_FDATASYNC_GROUP_COMMIT_WAIT_MICROS),
+    )
+});
 
 #[cfg(test)]
 mod dst_dir_fsync_group_commit_override {
@@ -400,8 +412,10 @@ fn dst_dir_fsync_group_commit_enabled() -> bool {
 #[cfg(test)]
 mod file_fdatasync_group_commit_override {
     use std::sync::{Mutex, MutexGuard, PoisonError, RwLock};
+    use std::time::Duration;
 
     static OVERRIDE: RwLock<Option<bool>> = RwLock::new(None);
+    static WAIT_OVERRIDE: RwLock<Option<Duration>> = RwLock::new(None);
     static SERIAL: Mutex<()> = Mutex::new(());
 
     pub(crate) fn get() -> Option<bool> {
@@ -415,6 +429,7 @@ mod file_fdatasync_group_commit_override {
     impl Drop for OverrideGuard {
         fn drop(&mut self) {
             *OVERRIDE.write().unwrap_or_else(PoisonError::into_inner) = None;
+            *WAIT_OVERRIDE.write().unwrap_or_else(PoisonError::into_inner) = None;
         }
     }
 
@@ -423,6 +438,14 @@ mod file_fdatasync_group_commit_override {
         *OVERRIDE.write().unwrap_or_else(PoisonError::into_inner) = Some(enabled);
         OverrideGuard { _serial: serial }
     }
+
+    pub(crate) fn set_wait(wait: Duration) {
+        *WAIT_OVERRIDE.write().unwrap_or_else(PoisonError::into_inner) = Some(wait);
+    }
+
+    pub(crate) fn wait() -> Option<Duration> {
+        *WAIT_OVERRIDE.read().unwrap_or_else(PoisonError::into_inner)
+    }
 }
 
 #[cfg(test)]
@@ -430,6 +453,11 @@ pub(crate) fn set_file_fdatasync_group_commit_for_test(enabled: bool) -> file_fd
     file_fdatasync_group_commit_override::set(enabled)
 }
 
+#[cfg(test)]
+fn set_file_fdatasync_group_commit_wait_for_test(wait: Duration) {
+    file_fdatasync_group_commit_override::set_wait(wait);
+}
+
 fn file_fdatasync_group_commit_enabled() -> bool {
     #[cfg(test)]
     if let Some(enabled) = file_fdatasync_group_commit_override::get() {
@@ -439,6 +467,15 @@ fn file_fdatasync_group_commit_enabled() -> bool {
     *FILE_FDATASYNC_GROUP_COMMIT_ENABLED
 }
 
+fn file_fdatasync_group_commit_wait() -> Duration {
+    #[cfg(test)]
+    if let Some(wait) = file_fdatasync_group_commit_override::wait() {
+        return wait;
+    }
+
+    *FILE_FDATASYNC_GROUP_COMMIT_WAIT
+}
+
 #[derive(Clone, Eq, Hash, PartialEq)]
 struct DstDirFsyncGroupKey {
     canonical_path: PathBuf,
@@ -934,6 +971,10 @@ async fn run_file_fdatasync_group_worker(group: Arc<FileFdatasyncGroup>) {
         #[cfg(test)]
         file_sync_probe::run_before_group_batch();
         tokio::task::yield_now().await;
+        let wait = file_fdatasync_group_commit_wait();
+        if !wait.is_zero() {
+            tokio::time::sleep(wait).await;
+        }
         let (batch, batch_file_count): (Vec<FileFdatasyncWaiter>, usize) = {
             let mut group_state = group.inner.lock();
             let batch_file_count = group_state.pending_files;
@@ -6075,6 +6116,7 @@ mod tests {
         use std::sync::mpsc;
 
         let _group_commit = set_file_fdatasync_group_commit_for_test(true);
+        set_file_fdatasync_group_commit_wait_for_test(Duration::ZERO);
         clear_file_fdatasync_group_commit_for_test();
         let temp_dir = tempdir().expect("create temp dir");
         let first_dir = temp_dir.path().join("first");
@@ -6141,12 +6183,83 @@ mod tests {
         assert_eq!(file_fdatasync_group_commit_counts_for_test(), (0, 0, 0));
     }
 
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    #[serial_test::serial(file_sync_probe)]
+    async fn file_fdatasync_group_commit_wait_budget_batches_late_follower() {
+        use std::sync::mpsc;
+
+        let _group_commit = set_file_fdatasync_group_commit_for_test(true);
+        set_file_fdatasync_group_commit_wait_for_test(Duration::from_millis(50));
+        clear_file_fdatasync_group_commit_for_test();
+        let temp_dir = tempdir().expect("create temp dir");
+        let first_dir = temp_dir.path().join("first");
+        let second_dir = temp_dir.path().join("second");
+        std::fs::create_dir(&first_dir).expect("create first dir");
+        std::fs::create_dir(&second_dir).expect("create second dir");
+        std::fs::write(first_dir.join("part.1"), b"first").expect("write first part");
+        std::fs::write(second_dir.join("part.1"), b"second").expect("write second part");
+        let _probe = file_sync_probe::set_blocking(temp_dir.path());
+        let (entered_tx, entered_rx) = mpsc::channel();
+        file_sync_probe::set_before_group_batch(move || {
+            entered_tx.send(()).expect("signal first file fdatasync group worker");
+        });
+
+        let limiter = file_sync_limiter();
+        let first_limiter = limiter.clone();
+        let first_path = first_dir.clone();
+        let first = tokio::spawn(async move { sync_dir_files_with_limiter(first_path, first_limiter).await });
+        tokio::task::spawn_blocking(move || entered_rx.recv_timeout(Duration::from_secs(30)))
+            .await
+            .expect("group worker hook waiter should run")
+            .expect("first file fdatasync group worker should start");
+
+        let second_limiter = limiter.clone();
+        let second_path = second_dir.clone();
+        let second = tokio::spawn(async move { sync_dir_files_with_limiter(second_path, second_limiter).await });
+        tokio::time::timeout(Duration::from_secs(30), async {
+            loop {
+                if file_fdatasync_group_commit_counts_for_test().1 == 2 {
+                    return;
+                }
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("second waiter should enqueue during the configured wait budget");
+        file_sync_probe::wait_for_active(1).await;
+
+        assert_eq!(
+            file_sync_probe::group_batches(),
+            vec![2],
+            "configured wait budget should let a follower join the leader's batch"
+        );
+        file_sync_probe::release();
+        first
+            .await
+            .expect("join first wait-budget file sync")
+            .expect("first wait-budget file sync must succeed");
+        second
+            .await
+            .expect("join second wait-budget file sync")
+            .expect("second wait-budget file sync must succeed");
+        assert!(
+            fsync_dir_recorder::was_fsynced(&first_dir),
+            "first source directory must still be fsynced"
+        );
+        assert!(
+            fsync_dir_recorder::was_fsynced(&second_dir),
+            "second source directory must still be fsynced"
+        );
+        assert_eq!(file_fdatasync_group_commit_counts_for_test(), (0, 0, 0));
+    }
+
     #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
     #[serial_test::serial(file_sync_probe)]
     async fn file_fdatasync_group_commit_failure_fails_all_waiters_before_dir_fsync() {
         use std::sync::mpsc;
 
         let _group_commit = set_file_fdatasync_group_commit_for_test(true);
+        set_file_fdatasync_group_commit_wait_for_test(Duration::ZERO);
         clear_file_fdatasync_group_commit_for_test();
         let temp_dir = tempdir().expect("create temp dir");
         let first_dir = temp_dir.path().join("first");

rustfs/src/app/object_usecase.rs

@@ -3969,6 +3969,14 @@ fn delete_creates_delete_marker(opts: &ObjectOptions) -> bool {
     opts.version_id.is_none() && opts.versioned && !opts.version_suspended
 }
 
+/// `DeleteObjects` is idempotent. A raw filesystem `NotFound` can cross the
+/// distributed delete path instead of its usual typed missing-object error.
+fn is_delete_objects_not_found(error: &EcstoreError) -> bool {
+    is_err_object_not_found(error)
+        || is_err_version_not_found(error)
+        || matches!(error, StorageError::Io(source) if source.kind() == std::io::ErrorKind::NotFound)
+}
+
 /// Bounded concurrency for the per-object pre-delete stat fanout in
 /// `execute_delete_objects` (backlog#929 / HP-8). Keeps the metadata reads for
 /// a 1000-key batch from serializing while capping the disk fanout pressure.
@@ -4030,6 +4038,27 @@ fn delete_response_version_id(version_id: Option<Uuid>, synthetic_version_id: bo
     }
 }
 
+fn reduce_delete_objects_result<'a>(
+    object: &ObjectToDelete,
+    deleted: &'a StorageDeletedObject,
+    error: Option<&EcstoreError>,
+    synthetic_version_id: bool,
+) -> Result<&'a StorageDeletedObject, s3s::dto::Error> {
+    match error {
+        None => Ok(deleted),
+        Some(error) if is_delete_objects_not_found(error) => Ok(deleted),
+        Some(error) => {
+            let api_error = ApiError::from(error.clone());
+            Err(s3s::dto::Error {
+                code: Some(api_error.code.as_str().to_string()),
+                key: Some(object.object_name.clone()),
+                message: Some(api_error.message),
+                version_id: delete_response_version_id(object.version_id, synthetic_version_id),
+            })
+        }
+    }
+}
+
 fn resolve_put_object_extract_options(headers: &HeaderMap) -> S3Result<PutObjectExtractOptions> {
     let prefix = snowball_meta_value(headers, SNOWBALL_PREFIX_HEADER_KEYS, SNOWBALL_PREFIX_SUFFIX_LOWER)
         .map(|value| normalize_snowball_prefix(&value))
@@ -8476,39 +8505,31 @@ impl DefaultObjectUsecase {
         for (i, err) in errs.iter().enumerate() {
             let didx = object_to_delete_idx[i];
 
-            if err.is_none()
-                || err
-                    .clone()
-                    .is_some_and(|v| is_err_object_not_found(&v) || is_err_version_not_found(&v))
-            {
-                delete_results[didx].delete_object = Some(dobjs[i].clone());
-                let (versioned, version_suspended) = object_versioning[i];
-                let creates_delete_marker = object_to_delete[i].version_id.is_none() && versioned && !version_suspended;
-                if creates_delete_marker {
-                    record_bucket_delete_marker_memory(&bucket).await;
-                } else {
-                    let size = object_sizes[i].max(0) as u64;
-                    record_bucket_object_delete_memory(
-                        &bucket,
-                        size,
-                        existing_object_infos[i].is_some() && object_to_delete[i].version_id.is_none(),
-                    )
-                    .await;
+            match reduce_delete_objects_result(
+                &object_to_delete[i],
+                &dobjs[i],
+                err.as_ref(),
+                delete_results[didx].synthetic_version_id,
+            ) {
+                Ok(deleted_object) => {
+                    delete_results[didx].delete_object = Some(deleted_object.clone());
+                    let (versioned, version_suspended) = object_versioning[i];
+                    let creates_delete_marker = object_to_delete[i].version_id.is_none() && versioned && !version_suspended;
+                    if creates_delete_marker {
+                        record_bucket_delete_marker_memory(&bucket).await;
+                    } else {
+                        let size = object_sizes[i].max(0) as u64;
+                        record_bucket_object_delete_memory(
+                            &bucket,
+                            size,
+                            existing_object_infos[i].is_some() && object_to_delete[i].version_id.is_none(),
+                        )
+                        .await;
+                    }
+                }
+                Err(error) => {
+                    delete_results[didx].error = Some(error);
                 }
-                continue;
-            }
-
-            if let Some(err) = err.clone() {
-                let api_error = ApiError::from(err);
-                delete_results[didx].error = Some(s3s::dto::Error {
-                    code: Some(api_error.code.as_str().to_string()),
-                    key: Some(object_to_delete[i].object_name.clone()),
-                    message: Some(api_error.message),
-                    version_id: delete_response_version_id(
-                        object_to_delete[i].version_id,
-                        delete_results[didx].synthetic_version_id,
-                    ),
-                });
             }
         }
 
@@ -17692,6 +17713,35 @@ mod tests {
         assert_eq!(internal_version_id, None);
     }
 
+    #[test]
+    fn delete_objects_treats_raw_io_not_found_as_idempotent() {
+        assert!(is_delete_objects_not_found(&StorageError::FileNotFound));
+        assert!(is_delete_objects_not_found(&StorageError::Io(std::io::Error::from(
+            std::io::ErrorKind::NotFound,
+        ))));
+        assert!(!is_delete_objects_not_found(&StorageError::Io(std::io::Error::from(
+            std::io::ErrorKind::PermissionDenied,
+        ))));
+        assert!(!is_delete_objects_not_found(&StorageError::DiskNotFound));
+    }
+
+    #[test]
+    fn delete_objects_result_reducer_reports_raw_not_found_as_deleted() {
+        let object = ObjectToDelete {
+            object_name: "missing-key".to_string(),
+            ..Default::default()
+        };
+        let deleted = StorageDeletedObject {
+            object_name: object.object_name.clone(),
+            ..Default::default()
+        };
+        let error = StorageError::Io(std::io::Error::from(std::io::ErrorKind::NotFound));
+
+        let deleted = reduce_delete_objects_result(&object, &deleted, Some(&error), false)
+            .expect("raw not-found must produce a deleted result");
+        assert_eq!(deleted.object_name, "missing-key");
+    }
+
     #[test]
     fn recursive_force_delete_requires_administrative_or_replica_context() {
         let mut headers = HeaderMap::new();