fix(ecstore): tidy dst dir fsync group open

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

.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)
 # ---------------------------------------------------------------------------

.github/workflows/package.yml

@@ -290,6 +290,12 @@ jobs:
           Homepage: https://rustfs.com
           EOF
 
+          # Declare /etc/default/rustfs as a conffile so dpkg preserves user
+          # modifications on upgrade instead of silently overwriting them.
+          cat > "${PKG_DIR}/DEBIAN/conffiles" << 'CONFFILES'
+          /etc/default/rustfs
+          CONFFILES
+
           cat > "${PKG_DIR}/DEBIAN/postinst" << 'POSTINST'
           #!/bin/bash
           set -e

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,374 @@ 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 metadata = std::fs::metadata(&canonical_path)?;
+            let key = DstDirFsyncGroupKey::from_metadata(canonical_path, metadata)?;
+            let dir = key.canonical_path.clone();
+            Ok(Self { key, dir })
+        }
+    }
+}
+
+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 +5223,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");

crates/kms/src/lib.rs

@@ -255,10 +255,6 @@ pub use cache::KmsCacheStats;
 pub use config::*;
 pub use deletion_worker::DeletionReferenceChecker;
 pub use encryption::is_data_key_envelope;
-// Re-exported so the object layer binds encryption context exactly the way the
-// KMS backends do. A second canonicalization is how the object layer once
-// serialized a HashMap directly while the Static backend already sorted keys.
-pub use encryption::context_aad;
 pub use error::{KmsError, KmsUnavailableError, Result};
 pub use key_impact::{KeyImpactReport, KeyReference, KeyReferenceKind, ReferenceCompleteness, ReferenceCoverage, ReferenceScope};
 pub use manager::KmsManager;

rustfs/src/storage/minio_generated_read_test.rs

@@ -4,7 +4,7 @@ use std::fs;
 use std::io::Cursor;
 use std::path::{Path, PathBuf};
 
-use super::sse::{SseObjectEncryptionResolver, reset_sse_dek_provider};
+use super::sse::SseObjectEncryptionResolver;
 use super::storage_api::ecstore_test_support::{
     DiskAPI as _, DiskOption, Endpoint, Erasure, GetObjectReader, ObjectInfo, ObjectOptions, create_bitrot_reader, new_disk,
 };
@@ -131,13 +131,6 @@ async fn load_fixture_reader_input(case_id: &str) -> (ObjectInfo, Vec<u8>, Strin
 async fn read_fixture_plaintext(encrypted: Vec<u8>, object_info: ObjectInfo, kms_key_b64: String) -> Result<Vec<u8>, String> {
     let object_size = object_info.size;
 
-    // The DEK provider is cached process-wide once built, so without this reset
-    // a case that ran earlier in the same binary keeps serving its master key to
-    // every later case — which silently turned the wrong-key negative below into
-    // a test that could not fail. Reset before each read so the provider is
-    // built from the key this case actually configured.
-    reset_sse_dek_provider();
-
     async_with_vars(
         [
             ("__RUSTFS_SSE_SIMPLE_CMK", Some(kms_key_b64)),

rustfs/src/storage/sse.rs

@@ -1460,16 +1460,6 @@ fn managed_sse_domain(sse_type: SSEType) -> &'static str {
     }
 }
 
-/// The public `x-amz-server-side-encryption` value a managed scheme reports.
-fn managed_sse_public_header(sse_type: SSEType) -> &'static str {
-    match sse_type {
-        SSEType::SseKms => ServerSideEncryption::AWS_KMS,
-        // SSE-C never reaches the managed path; reporting AES256 keeps this
-        // total without inventing a third public value.
-        SSEType::SseS3 | SSEType::SseC => ServerSideEncryption::AES256,
-    }
-}
-
 fn canonical_kms_bucket_path(bucket: &str, key: &str) -> String {
     path_join_buf(&[bucket, key])
 }
@@ -2455,42 +2445,20 @@ async fn apply_managed_decryption_material_inner(
 ) -> Result<Option<DecryptionMaterial>, ApiError> {
     #[cfg(not(feature = "rio-v2"))]
     let _ = (bucket, key);
-    if !contains_managed_encryption_metadata(metadata) {
+    if !contains_managed_encryption_metadata(metadata) || !metadata.contains_key("x-amz-server-side-encryption") {
         return Ok(None);
     }
 
-    let encryption_type = match metadata.get("x-amz-server-side-encryption").map(String::as_str) {
-        Some(ServerSideEncryption::AWS_KMS) => SSEType::SseKms,
-        Some(_) => SSEType::SseS3,
-        // MinIO never persists the public scheme header: `crypto.S3.CreateMetadata`
-        // writes only the `X-Minio-Internal-*` family and the public header is
-        // synthesized onto the response by `DecryptObjectInfo`. Requiring it here
-        // is what made every MinIO-encrypted object unreadable (backlog#1638).
-        //
-        // Inferring from the sealed-key slot is self-consistent by construction:
-        // the slot decides which header the unseal reads AND which domain string
-        // the sealing key is derived under, so a scheme that disagrees with the
-        // slot cannot silently derive a wrong key — it finds no key at all.
-        // Inferring from the KMS key id would NOT be safe: MinIO writes
-        // `-S3-Kms-Key-Id` on SSE-S3 objects too.
-        #[cfg(feature = "rio-v2")]
-        None => match infer_minio_managed_sse_type(metadata) {
-            Some(sse_type) => sse_type,
-            // Still fail-closed, and deliberately not an error raised here: the
-            // read plan independently classifies the object as encrypted from
-            // its markers and refuses to serve it without material, so an
-            // object whose scheme cannot be established never degrades into a
-            // plaintext read.
-            None => return Ok(None),
-        },
-        // Without the rio-v2 reader there is no MinIO-format read path to serve
-        // such an object with, so it stays on the fail-closed branch.
-        #[cfg(not(feature = "rio-v2"))]
-        None => return Ok(None),
-    };
-
+    // Safe: presence is guaranteed by the contains_key check above.
+    let server_side_encryption = metadata.get("x-amz-server-side-encryption").cloned().unwrap_or_default();
     let normalized_metadata = normalize_managed_metadata(metadata, Some(recode_minio_kms_context));
 
+    let encryption_type = match server_side_encryption.as_str() {
+        ServerSideEncryption::AES256 => SSEType::SseS3,
+        ServerSideEncryption::AWS_KMS => SSEType::SseKms,
+        _ => SSEType::SseS3,
+    };
+
     // Extract KMS key ID from metadata (optional, used for provider context)
     let kms_key_id = normalized_metadata
         .get(INTERNAL_ENCRYPTION_KEY_ID_HEADER)
@@ -2588,19 +2556,8 @@ async fn apply_managed_decryption_material_inner(
     } else {
         get_local_sse_dek_provider().await?
     };
-    // A MinIO sealed key alone does not mean MinIO wrote the object: RustFS's own
-    // writer fills MinIO's metadata slots too, while still storing a RustFS
-    // envelope in them, so neither the slot nor the header name distinguishes the
-    // two. The data key's own shape does. RustFS envelopes are strictly-parsed
-    // JSON; MinIO's builtin-KMS ciphertext is opaque bytes that match neither, so
-    // recognizing RustFS positively — and treating only the remainder as MinIO —
-    // keeps a RustFS envelope from ever reaching MinIO's decoder.
     #[cfg(feature = "rio-v2")]
-    let decrypted_data_key = if minio_sealed_key.is_some() && !is_rustfs_managed_data_key(&encrypted_data_key) {
-        provider
-            .decrypt_minio_sse_dek(&encrypted_data_key, &kms_key_id, &object_context)
-            .await
-    } else if is_legacy_rustfs_managed_metadata(&normalized_metadata) {
+    let decrypted_data_key = if is_legacy_rustfs_managed_metadata(&normalized_metadata) {
         provider
             .decrypt_legacy_sse_dek(&encrypted_data_key, &kms_key_id, &object_context)
             .await
@@ -2635,11 +2592,7 @@ async fn apply_managed_decryption_material_inner(
 
     Ok(Some(DecryptionMaterial {
         sse_type: encryption_type,
-        // Synthesized from the resolved scheme rather than read back from
-        // metadata: a MinIO-written object has no stored scheme header, which is
-        // exactly why the gate above had to infer it. MinIO synthesizes the same
-        // header onto its own responses.
-        server_side_encryption: ServerSideEncryption::from(managed_sse_public_header(encryption_type).to_string()),
+        server_side_encryption: ServerSideEncryption::from(server_side_encryption),
         kms_key_id: Some(SSEKMSKeyId::from(kms_key_id)),
         algorithm,
         customer_key_md5: None,
@@ -2706,29 +2659,6 @@ pub trait SseDekProvider: Send + Sync {
     ) -> Result<[u8; 32], ApiError> {
         self.decrypt_sse_dek(encrypted_dek, kms_key_id, context).await
     }
-
-    /// Unwrap a data key that MinIO's builtin KMS sealed.
-    ///
-    /// A separate entry point rather than a shape sniff inside
-    /// [`Self::decrypt_sse_dek`]: the caller already knows the object carries a
-    /// MinIO sealed key, and MinIO's raw ciphertext is unstructured bytes that
-    /// no parser can reliably tell apart from anything else. Routing on the
-    /// caller's knowledge keeps a RustFS envelope from ever reaching MinIO's
-    /// decoder, and vice versa.
-    ///
-    /// Defaults to refusing: only a provider holding the MinIO master secret
-    /// can serve these, and a provider that cannot must fail rather than fall
-    /// back to a decoder that would misread the bytes.
-    async fn decrypt_minio_sse_dek(
-        &self,
-        _encrypted_dek: &[u8],
-        _kms_key_id: &str,
-        _context: &ObjectEncryptionContext,
-    ) -> Result<[u8; 32], ApiError> {
-        Err(ApiError::from(StorageError::other(
-            "This KMS provider cannot unwrap a data key sealed by MinIO's builtin KMS",
-        )))
-    }
 }
 
 // ============================================================================
@@ -2867,163 +2797,6 @@ pub(crate) struct LocalSseDekProvider {
 
 const LOCAL_SSE_DEK_FORMAT_VERSION: u8 = 1;
 
-#[cfg(feature = "rio-v2")]
-/// Returns true when a managed-SSE data key is one RustFS itself wrote.
-///
-/// Both RustFS envelope shapes are strict JSON — the KMS envelope
-/// ([`rustfs_kms::is_data_key_envelope`]) and the local provider's
-/// [`LocalSseDekEnvelope`], whose `deny_unknown_fields` keeps it from accepting
-/// anything else. Recognition is deliberately positive: an unrecognized payload
-/// is left to MinIO's decoder rather than guessed at, and neither decoder is
-/// ever handed the other's format.
-fn is_rustfs_managed_data_key(encrypted_dek: &[u8]) -> bool {
-    if rustfs_kms::is_data_key_envelope(encrypted_dek) {
-        return true;
-    }
-    std::str::from_utf8(encrypted_dek)
-        .ok()
-        .is_some_and(|text| serde_json::from_str::<LocalSseDekEnvelope<'_>>(text).is_ok())
-}
-
-#[cfg(feature = "rio-v2")]
-/// Associated data MinIO binds when sealing a data key.
-///
-/// MinIO passes the object's encryption context as the AEAD's associated data,
-/// serialized as canonical JSON with sorted keys — the same canonicalization
-/// [`rustfs_kms::context_aad`] performs, which is why the context RustFS
-/// already rebuilds for the read can be reused verbatim. For SSE-S3 that
-/// context is `{bucket: "bucket/object"}`; for SSE-KMS it is whatever the
-/// request supplied, recovered from the stored MinIO context header.
-fn minio_kms_associated_data(context: &ObjectEncryptionContext) -> Result<Vec<u8>, ApiError> {
-    let mut ctx = context.encryption_context.clone();
-    ctx.entry(context.bucket.clone())
-        .or_insert_with(|| canonical_kms_bucket_path(&context.bucket, &context.object_key));
-    rustfs_kms::context_aad(&ctx)
-        .map_err(|e| ApiError::from(StorageError::other(format!("Failed to canonicalize MinIO KMS context: {e}"))))
-}
-
-#[cfg(feature = "rio-v2")]
-/// MinIO's builtin-KMS ciphertext in its JSON encoding.
-///
-/// Deliberately its own type rather than a relaxation of
-/// [`LocalSseDekEnvelope`]: widening that envelope's `deny_unknown_fields`
-/// to admit this shape would also admit malformed RustFS envelopes, which
-/// backlog#1567 requires to keep failing closed.
-#[derive(Debug, Deserialize)]
-#[serde(deny_unknown_fields)]
-struct MinioKmsCiphertextJson {
-    aead: String,
-    #[allow(
-        dead_code,
-        reason = "present in MinIO's encoding; the key is identified by metadata instead"
-    )]
-    #[serde(default)]
-    id: String,
-    iv: String,
-    nonce: String,
-    bytes: String,
-}
-
-/// Bytes of trailing randomness every MinIO builtin-KMS ciphertext carries:
-/// a 16-byte IV followed by a 12-byte nonce, *after* the sealed bytes.
-#[cfg(feature = "rio-v2")]
-const MINIO_KMS_RANDOM_LEN: usize = 28;
-#[cfg(feature = "rio-v2")]
-const MINIO_KMS_IV_LEN: usize = 16;
-
-#[cfg(feature = "rio-v2")]
-const MINIO_KMS_AEAD_AES_GCM: &str = "AES-256-GCM-HMAC-SHA-256";
-#[cfg(feature = "rio-v2")]
-const MINIO_KMS_AEAD_CHACHA20: &str = "ChaCha20Poly1305";
-
-#[cfg(feature = "rio-v2")]
-/// Unwrap a data key sealed by MinIO's builtin (static-secret) KMS.
-///
-/// The wire format is `sealed_bytes || iv[16] || nonce[12]` — the randomness
-/// trails the ciphertext rather than leading it, and MinIO's own decoder
-/// normalizes its legacy JSON encoding into exactly that byte order before
-/// opening it (`internal/kms/secret-key.go`, `parseCiphertext`). A raw
-/// (non-JSON) ciphertext is AES-256-GCM by definition there; the JSON form
-/// names its algorithm.
-///
-/// The sealing key is derived per ciphertext rather than being the master key:
-/// `HMAC-SHA256(master, iv)` for AES-256-GCM, `HChaCha20(master, iv)` for
-/// ChaCha20-Poly1305. The encryption context is bound as associated data.
-fn decrypt_minio_kms_data_key(encrypted_dek: &[u8], master_key: &[u8; 32], aad: &[u8]) -> Result<[u8; 32], ApiError> {
-    let (body, algorithm) = match std::str::from_utf8(encrypted_dek) {
-        // MinIO only treats a payload as JSON when it both starts and ends like
-        // an object, and falls back to the raw layout when it does not parse —
-        // mirrored here so a ciphertext that merely looks like JSON is not
-        // rejected outright.
-        Ok(text)
-            if text.starts_with('{')
-                && text.ends_with('}')
-                && let Ok(json) = serde_json::from_str::<MinioKmsCiphertextJson>(text) =>
-        {
-            let decode = |what: &str, value: &str| -> Result<Vec<u8>, ApiError> {
-                BASE64_STANDARD
-                    .decode(value)
-                    .map_err(|e| ApiError::from(StorageError::other(format!("Invalid MinIO KMS {what}: {e}"))))
-            };
-            let mut body = decode("ciphertext", &json.bytes)?;
-            body.extend_from_slice(&decode("iv", &json.iv)?);
-            body.extend_from_slice(&decode("nonce", &json.nonce)?);
-            (body, json.aead)
-        }
-        _ => (encrypted_dek.to_vec(), MINIO_KMS_AEAD_AES_GCM.to_string()),
-    };
-
-    if body.len() <= MINIO_KMS_RANDOM_LEN {
-        return Err(ApiError::from(StorageError::other(
-            "MinIO KMS ciphertext is too short to carry its IV and nonce",
-        )));
-    }
-    let (sealed, random) = body.split_at(body.len() - MINIO_KMS_RANDOM_LEN);
-    let (iv, nonce) = random.split_at(MINIO_KMS_IV_LEN);
-
-    let plaintext = match algorithm.as_str() {
-        MINIO_KMS_AEAD_AES_GCM => {
-            use aes_gcm::{Aes256Gcm, KeyInit, aead::Aead};
-            let mut mac = HmacSha256::new_from_slice(master_key)
-                .map_err(|_| ApiError::from(StorageError::other("MinIO KMS sealing key derivation failed")))?;
-            mac.update(iv);
-            let sealing_key: [u8; 32] = mac.finalize().into_bytes().into();
-            let cipher = Aes256Gcm::new_from_slice(&sealing_key)
-                .map_err(|_| ApiError::from(StorageError::other("MinIO KMS sealing key is not a valid AES-256 key")))?;
-            let nonce = aes_gcm::Nonce::try_from(nonce)
-                .map_err(|_| ApiError::from(StorageError::other("MinIO KMS nonce is not 12 bytes")))?;
-            cipher.decrypt(&nonce, aes_gcm::aead::Payload { msg: sealed, aad })
-        }
-        MINIO_KMS_AEAD_CHACHA20 => {
-            use chacha20poly1305::{KeyInit, XChaCha20Poly1305, aead::Aead};
-            // MinIO derives this branch's key with HChaCha20 over the 16-byte
-            // IV, which is exactly XChaCha20-Poly1305's own construction, so the
-            // extended-nonce cipher does the derivation rather than hand-rolling it.
-            let mut extended = Vec::with_capacity(MINIO_KMS_IV_LEN + nonce.len());
-            extended.extend_from_slice(iv);
-            extended.extend_from_slice(nonce);
-            let cipher = XChaCha20Poly1305::new_from_slice(master_key)
-                .map_err(|_| ApiError::from(StorageError::other("MinIO KMS master key is not a valid ChaCha20 key")))?;
-            let nonce = chacha20poly1305::XNonce::try_from(extended.as_slice())
-                .map_err(|_| ApiError::from(StorageError::other("MinIO KMS extended nonce is not 24 bytes")))?;
-            cipher.decrypt(&nonce, chacha20poly1305::aead::Payload { msg: sealed, aad })
-        }
-        other => {
-            return Err(ApiError::from(StorageError::other(format!(
-                "Unsupported MinIO KMS AEAD algorithm: {other}"
-            ))));
-        }
-    }
-    // An AEAD failure here is authentication, not a decode slip: a wrong master
-    // key, a tampered ciphertext, and an encryption context that does not match
-    // what sealed it all land here and must all fail closed.
-    .map_err(|_| ApiError::from(StorageError::other("MinIO KMS data key failed authentication")))?;
-
-    plaintext.try_into().map_err(|value: Vec<u8>| {
-        ApiError::from(StorageError::other(format!("MinIO KMS data key must be 32 bytes, got {}", value.len())))
-    })
-}
-
 #[derive(Debug, Deserialize, Serialize)]
 #[serde(deny_unknown_fields)]
 struct LocalSseDekEnvelope<'a> {
@@ -3240,17 +3013,6 @@ impl SseDekProvider for LocalSseDekProvider {
         let dek = Self::decrypt_dek(encrypted_dek_str, self.master_key)?;
         Ok(dek)
     }
-
-    #[cfg(feature = "rio-v2")]
-    async fn decrypt_minio_sse_dek(
-        &self,
-        encrypted_dek: &[u8],
-        _kms_key_id: &str,
-        context: &ObjectEncryptionContext,
-    ) -> Result<[u8; 32], ApiError> {
-        let aad = minio_kms_associated_data(context)?;
-        decrypt_minio_kms_data_key(encrypted_dek, &self.master_key, &aad)
-    }
 }
 
 // ============================================================================
@@ -3439,23 +3201,6 @@ fn is_legacy_rustfs_managed_metadata(metadata: &HashMap<String, String>) -> bool
         && !metadata.contains_key(MINIO_INTERNAL_ENCRYPTION_KMS_SEALED_KEY_HEADER)
 }
 
-#[cfg(feature = "rio-v2")]
-#[cfg(feature = "rio-v2")]
-/// Infer the managed SSE scheme from the MinIO sealed-key slot that is present.
-///
-/// Returns `None` when no managed MinIO slot is present, which keeps callers on
-/// their fail-closed path. SSE-C is not a managed scheme and is handled by the
-/// SSE-C read path, so its slot is not considered here.
-fn infer_minio_managed_sse_type(metadata: &HashMap<String, String>) -> Option<SSEType> {
-    if metadata.contains_key(MINIO_INTERNAL_ENCRYPTION_S3_SEALED_KEY_HEADER) {
-        Some(SSEType::SseS3)
-    } else if metadata.contains_key(MINIO_INTERNAL_ENCRYPTION_KMS_SEALED_KEY_HEADER) {
-        Some(SSEType::SseKms)
-    } else {
-        None
-    }
-}
-
 #[cfg(feature = "rio-v2")]
 fn parse_minio_managed_sealed_key(
     metadata: &HashMap<String, String>,