Merge branch 'main' into codex/backlog-1548-protocols-workspace-lints

* fix(scanner): report active first-cycle status

* fix(scanner): publish cycle activity consistently

* test(common): satisfy Rust 1.97 waker lint

* fix(scanner): satisfy Rust 1.97 clippy

---------

Co-authored-by: houseme <housemecn@gmail.com>

.github/workflows/ci.yml

@@ -175,6 +175,33 @@ jobs:
       - name: Run nextest tests
         run: |
           mkdir -p artifacts/test-and-lint
+          # Evidence sampler for issue #5394: the post-mortem pgrep below runs
+          # only after `timeout` has already TERM'd the whole cargo process
+          # group, so it cannot name a wedged process. Sample system and
+          # process state every 60s instead; the last samples before the
+          # timeout show what was stuck (rustc, linker, build script, memory
+          # pressure, ...). The log rides along in the existing artifact.
+          (
+            while true; do
+              {
+                echo "=== $(date --utc --iso-8601=seconds)"
+                echo "--- load";  cat /proc/loadavg
+                echo "--- psi";   grep -H . /proc/pressure/* 2>/dev/null || true
+                echo "--- mem";   free -m
+                echo "--- disk";  df -h / /home/runner 2>/dev/null || df -h /
+                echo "--- top-rss"
+                ps -eo pid,ppid,stat,etime,rss,pcpu,args --sort=-rss | head -15
+                echo "--- build/test processes"
+                ps -eo pid,ppid,stat,etime,rss,pcpu,args | grep -E '[c]argo|[r]ustc|[n]extest|[c]ollect2|rust-ll[d]|[b]uild-script|deps[/]' || true
+                echo "--- d-state (uninterruptible IO)"
+                ps -eo pid,stat,etime,args | awk 'NR > 1 && $2 ~ /D/' || true
+                echo
+              } >> artifacts/test-and-lint/sampler.log 2>&1 || true
+              sleep 60
+            done
+          ) &
+          sampler_pid=$!
+          trap 'kill "${sampler_pid}" 2>/dev/null || true' EXIT
           set +e
           NEXTEST_HIDE_PROGRESS_BAR=1 timeout --verbose --signal=TERM --kill-after=30s 75m \
             cargo nextest run --profile ci --all --exclude e2e_test \
@@ -188,6 +215,9 @@ jobs:
             echo
             echo "Remaining test-related processes:"
             pgrep -af 'cargo|nextest|target/.*/deps/' || true
+            echo
+            echo "Kernel OOM / kill events:"
+            dmesg -T 2>/dev/null | grep -iE 'oom|out of memory|killed process' | tail -20 || true
           } > artifacts/test-and-lint/nextest-diagnostics.txt
           exit "${status}"
 

Cargo.lock

@@ -5645,9 +5645,9 @@ dependencies = [
 
 [[package]]
 name = "lazy-regex"
-version = "3.6.0"
+version = "3.6.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6bae91019476d3ec7147de9aa291cadb6d870abf2f3015d2da73a90325ac1496"
+checksum = "4994ba703f78b083e2f7946dac9251abd83fd43a0365f030e99b69be5b4b9ef9"
 dependencies = [
  "lazy-regex-proc_macros",
  "once_cell",
@@ -5656,9 +5656,9 @@ dependencies = [
 
 [[package]]
 name = "lazy-regex-proc_macros"
-version = "3.6.0"
+version = "3.6.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4de9c1e1439d8b7b3061b2d209809f447ca33241733d9a3c01eabf2dc8d94358"
+checksum = "fd97232314824e6dbef1918a871bb93f51070455e3715bf26e19a6d01aa977a0"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -9106,6 +9106,7 @@ dependencies = [
 name = "rustfs-ecstore"
 version = "1.0.0-beta.11"
 dependencies = [
+ "aes-gcm",
  "arc-swap",
  "async-channel",
  "async-recursion",
@@ -9121,6 +9122,7 @@ dependencies = [
  "byteorder",
  "bytes",
  "bytesize",
+ "chacha20poly1305",
  "chrono",
  "criterion",
  "enumset",
@@ -9173,6 +9175,7 @@ dependencies = [
  "rustfs-erasure-codec",
  "rustfs-filemeta",
  "rustfs-io-metrics",
+ "rustfs-kms",
  "rustfs-lifecycle",
  "rustfs-lock",
  "rustfs-madmin",
@@ -10128,6 +10131,7 @@ dependencies = [
  "bytes",
  "convert_case 0.11.0",
  "crc-fast",
+ "criterion",
  "flate2",
  "futures",
  "hex-simd",

crates/common/src/metrics.rs

@@ -1114,6 +1114,8 @@ pub struct ScannerLastMinute {
 pub struct ScannerMetricsReport {
     pub collected_at: DateTime<Utc>,
     pub current_cycle: u64,
+    #[serde(default)]
+    pub current_cycle_active: bool,
     pub current_started: DateTime<Utc>,
     pub cycles_completed_at: Vec<DateTime<Utc>>,
     pub ongoing_buckets: usize,
@@ -2051,7 +2053,7 @@ impl Metrics {
     pub fn record_scanner_transition_failed(&self, count: u64) {
         self.scanner_transition_failed.fetch_add(count, Ordering::Relaxed);
         self.record_scanner_source_failed(ScannerWorkSource::Lifecycle, count);
-        if !self.current_scan_cycle_work_active.load(Ordering::Relaxed) {
+        if !self.current_scan_cycle_work_active.load(Ordering::Acquire) {
             self.record_last_cycle_scanner_source_work(ScannerWorkSource::Lifecycle, ScannerSourceWorkUpdate::failed(count));
         }
     }
@@ -2336,6 +2338,21 @@ impl Metrics {
         *self.cycle_info.write().await = cycle;
     }
 
+    /// Publish a scanner cycle and its work-accounting baseline as one state transition.
+    pub async fn start_scan_cycle_work_with_cycle(&self, cycle: CurrentCycle) -> ScanCycleWorkSnapshot {
+        let mut current_cycle = self.cycle_info.write().await;
+        let snapshot = self.start_scan_cycle_work();
+        *current_cycle = Some(cycle);
+        snapshot
+    }
+
+    /// Publish the completed work snapshot and idle cycle state as one state transition.
+    pub async fn finish_scan_cycle_work_with_cycle(&self, start: ScanCycleWorkSnapshot, cycle: CurrentCycle) {
+        let mut current_cycle = self.cycle_info.write().await;
+        self.finish_scan_cycle_work(start);
+        *current_cycle = Some(cycle);
+    }
+
     /// Read the current cycle record.
     pub async fn get_cycle(&self) -> Option<CurrentCycle> {
         self.cycle_info.read().await.clone()
@@ -2464,7 +2481,7 @@ impl Metrics {
             &self.current_scan_cycle_replication_repair_work_start,
             &replication_repair_snapshot,
         );
-        self.current_scan_cycle_work_active.store(true, Ordering::Relaxed);
+        self.current_scan_cycle_work_active.store(true, Ordering::Release);
         snapshot
     }
 
@@ -2476,11 +2493,11 @@ impl Metrics {
         self.record_scan_cycle_work(work);
         self.record_scan_cycle_source_work(&source_work);
         self.record_scan_cycle_replication_repair_work(&replication_repair_work);
-        self.current_scan_cycle_work_active.store(false, Ordering::Relaxed);
+        self.current_scan_cycle_work_active.store(false, Ordering::Release);
     }
 
     pub fn current_scan_cycle_has_unresolved_heal_work(&self) -> bool {
-        if !self.current_scan_cycle_work_active.load(Ordering::Relaxed) {
+        if !self.current_scan_cycle_work_active.load(Ordering::Acquire) {
             return false;
         }
 
@@ -2746,13 +2763,41 @@ impl Metrics {
     pub async fn report(&self) -> ScannerMetricsReport {
         let mut m = ScannerMetricsReport::default();
 
-        let has_cycle = if let Some(cycle) = self.get_cycle().await {
-            m.current_cycle = cycle.current;
-            m.cycles_completed_at = cycle.cycle_completed;
-            m.current_started = cycle.started;
-            true
-        } else {
-            false
+        let has_cycle = {
+            let cycle = self.cycle_info.read().await;
+            let has_cycle = if let Some(cycle) = cycle.as_ref() {
+                m.current_cycle = cycle.current;
+                m.cycles_completed_at = cycle.cycle_completed.clone();
+                m.current_started = cycle.started;
+                true
+            } else {
+                false
+            };
+            m.current_cycle_active = self.current_scan_cycle_work_active.load(Ordering::Acquire);
+            if m.current_cycle_active {
+                let current_work = self.scan_cycle_work_since(self.current_scan_cycle_work_start());
+                let current_source_work = self.scanner_source_work_since(&self.current_scan_cycle_source_work_start_values());
+                let current_replication_repair_work =
+                    self.scanner_replication_repair_work_since(&self.current_scan_cycle_replication_repair_work_start_values());
+                m.current_cycle_objects_scanned = current_work.objects_scanned;
+                m.current_cycle_directories_scanned = current_work.directories_scanned;
+                m.current_cycle_bucket_drive_scans = current_work.bucket_drive_scans;
+                m.current_cycle_bucket_drive_failures = current_work.bucket_drive_failures;
+                m.current_cycle_yield_events = current_work.yield_events;
+                m.current_cycle_yield_duration_seconds = current_work.yield_duration_millis as f64 / 1000.0;
+                m.current_cycle_throttle_sleep_events = current_work.throttle_sleep_events;
+                m.current_cycle_throttle_sleep_duration_seconds = current_work.throttle_sleep_duration_millis as f64 / 1000.0;
+                m.current_cycle_ilm_actions = current_work.ilm_actions;
+                m.current_cycle_lifecycle_expiry_actions = current_work.lifecycle_expiry_actions;
+                m.current_cycle_lifecycle_transition_actions = current_work.lifecycle_transition_actions;
+                m.current_cycle_heal_objects = current_work.heal_objects;
+                m.current_cycle_replication_checks = current_work.replication_checks;
+                m.current_cycle_usage_saves = current_work.usage_saves;
+                m.current_cycle_source_work = self.scanner_source_work_snapshots(&current_source_work);
+                m.current_cycle_replication_repair =
+                    self.scanner_replication_repair_work_snapshots(&current_replication_repair_work);
+            }
+            has_cycle
         };
 
         if !has_cycle && let Some(init_time) = crate::get_global_init_time().await {
@@ -2793,28 +2838,6 @@ impl Metrics {
         m.current_disk_scan_concurrency_limit = disk_scan_concurrency_limit;
         m.current_disk_bucket_scans_queued = disk_bucket_scans_queued;
         m.current_disk_bucket_scans_active = disk_bucket_scans_active;
-        if self.current_scan_cycle_work_active.load(Ordering::Relaxed) {
-            let current_work = self.scan_cycle_work_since(self.current_scan_cycle_work_start());
-            let current_source_work = self.scanner_source_work_since(&self.current_scan_cycle_source_work_start_values());
-            let current_replication_repair_work =
-                self.scanner_replication_repair_work_since(&self.current_scan_cycle_replication_repair_work_start_values());
-            m.current_cycle_objects_scanned = current_work.objects_scanned;
-            m.current_cycle_directories_scanned = current_work.directories_scanned;
-            m.current_cycle_bucket_drive_scans = current_work.bucket_drive_scans;
-            m.current_cycle_bucket_drive_failures = current_work.bucket_drive_failures;
-            m.current_cycle_yield_events = current_work.yield_events;
-            m.current_cycle_yield_duration_seconds = current_work.yield_duration_millis as f64 / 1000.0;
-            m.current_cycle_throttle_sleep_events = current_work.throttle_sleep_events;
-            m.current_cycle_throttle_sleep_duration_seconds = current_work.throttle_sleep_duration_millis as f64 / 1000.0;
-            m.current_cycle_ilm_actions = current_work.ilm_actions;
-            m.current_cycle_lifecycle_expiry_actions = current_work.lifecycle_expiry_actions;
-            m.current_cycle_lifecycle_transition_actions = current_work.lifecycle_transition_actions;
-            m.current_cycle_heal_objects = current_work.heal_objects;
-            m.current_cycle_replication_checks = current_work.replication_checks;
-            m.current_cycle_usage_saves = current_work.usage_saves;
-            m.current_cycle_source_work = self.scanner_source_work_snapshots(&current_source_work);
-            m.current_cycle_replication_repair = self.scanner_replication_repair_work_snapshots(&current_replication_repair_work);
-        }
         let last_cycle_result = self.last_scan_cycle_result.load(Ordering::Relaxed);
         m.last_cycle_result = scan_cycle_result_label(last_cycle_result).to_string();
         m.last_cycle_result_code = last_cycle_result as u64;
@@ -4142,6 +4165,8 @@ mod tests {
 
         let report = metrics.report().await;
 
+        assert!(report.current_cycle_active);
+        assert_eq!(report.current_cycle, 0);
         assert_eq!(report.current_cycle_objects_scanned, 7);
         assert_eq!(report.current_cycle_directories_scanned, 3);
         assert_eq!(report.current_cycle_bucket_drive_scans, 2);
@@ -4158,6 +4183,8 @@ mod tests {
         metrics.finish_scan_cycle_work(start);
         let report = metrics.report().await;
 
+        assert!(!report.current_cycle_active);
+        assert_eq!(report.current_cycle, 0);
         assert_eq!(report.current_cycle_objects_scanned, 0);
         assert_eq!(report.current_cycle_directories_scanned, 0);
         assert_eq!(report.current_cycle_bucket_drive_scans, 0);
@@ -4184,6 +4211,91 @@ mod tests {
         assert_eq!(report.last_cycle_usage_saves, 2);
     }
 
+    #[tokio::test]
+    async fn scan_cycle_activity_and_cycle_state_publish_together() {
+        let metrics = Metrics::new();
+        let cycle_started = Utc::now() - chrono::Duration::seconds(5);
+        let active_cycle = CurrentCycle {
+            current: 12,
+            next: 13,
+            started: cycle_started,
+            ..Default::default()
+        };
+
+        let cycle_state = metrics.cycle_info.read().await;
+        let mut start_transition = Box::pin(metrics.start_scan_cycle_work_with_cycle(active_cycle));
+        let waker = std::task::Waker::noop();
+        let mut context = std::task::Context::from_waker(waker);
+        assert!(start_transition.as_mut().poll(&mut context).is_pending());
+        assert!(!metrics.current_scan_cycle_work_active.load(Ordering::Acquire));
+        drop(cycle_state);
+
+        let start = start_transition.await;
+        let active = metrics.report().await;
+        assert!(active.current_cycle_active);
+        assert_eq!(active.current_cycle, 12);
+        assert_eq!(active.current_started, cycle_started);
+
+        let idle_cycle = CurrentCycle {
+            current: 0,
+            next: 13,
+            started: cycle_started,
+            ..Default::default()
+        };
+        let cycle_state = metrics.cycle_info.read().await;
+        let mut finish_transition = Box::pin(metrics.finish_scan_cycle_work_with_cycle(start, idle_cycle));
+        assert!(finish_transition.as_mut().poll(&mut context).is_pending());
+        assert!(metrics.current_scan_cycle_work_active.load(Ordering::Acquire));
+        drop(cycle_state);
+
+        finish_transition.await;
+        let idle = metrics.report().await;
+        assert!(!idle.current_cycle_active);
+        assert_eq!(idle.current_cycle, 0);
+    }
+
+    #[tokio::test]
+    async fn report_keeps_cycle_identity_and_work_in_one_snapshot() {
+        let metrics = Metrics::new();
+        let cycle_ten = CurrentCycle {
+            current: 10,
+            next: 11,
+            started: Utc::now() - chrono::Duration::seconds(10),
+            ..Default::default()
+        };
+        let cycle_ten_start = metrics.start_scan_cycle_work_with_cycle(cycle_ten.clone()).await;
+        metrics.operations[Metric::ScanObject as usize].store(1, Ordering::Relaxed);
+
+        let paths = metrics.current_paths.write().await;
+        let mut report = Box::pin(metrics.report());
+        let waker = std::task::Waker::noop();
+        let mut context = std::task::Context::from_waker(waker);
+        assert!(report.as_mut().poll(&mut context).is_pending());
+
+        metrics
+            .finish_scan_cycle_work_with_cycle(cycle_ten_start, CurrentCycle { current: 0, ..cycle_ten })
+            .await;
+        let cycle_eleven_start = metrics
+            .start_scan_cycle_work_with_cycle(CurrentCycle {
+                current: 11,
+                next: 12,
+                started: Utc::now(),
+                ..Default::default()
+            })
+            .await;
+        metrics.operations[Metric::ScanObject as usize].store(101, Ordering::Relaxed);
+
+        drop(paths);
+        let snapshot = report.await;
+
+        assert_eq!(snapshot.current_cycle, 10);
+        assert_eq!(snapshot.current_cycle_objects_scanned, 1);
+
+        metrics
+            .finish_scan_cycle_work_with_cycle(cycle_eleven_start, CurrentCycle::default())
+            .await;
+    }
+
     #[tokio::test]
     async fn scanner_cycle_ilm_actions_ignore_global_ilm_work() {
         let metrics = Metrics::new();

crates/ecstore/Cargo.toml

@@ -57,6 +57,7 @@ rustfs-policy.workspace = true
 rustfs-protos.workspace = true
 rustfs-replication.workspace = true
 rustfs-lifecycle.workspace = true
+rustfs-kms.workspace = true
 rustfs-s3-types = { workspace = true }
 rustfs-data-usage.workspace = true
 rustfs-object-capacity.workspace = true
@@ -123,6 +124,8 @@ libc.workspace = true
 rustix = { workspace = true, features = ["process", "fs"] }
 rustfs-madmin.workspace = true
 reqwest = { workspace = true }
+aes-gcm = { workspace = true, features = ["rand_core"] }
+chacha20poly1305.workspace = true
 aws-sdk-s3 = { workspace = true, default-features = false, features = ["sigv4a", "default-https-client", "rt-tokio"] }
 urlencoding = { workspace = true }
 smallvec = { workspace = true, features = ["serde"] }

crates/ecstore/src/api/mod.rs

@@ -382,12 +382,10 @@ pub mod notification {
 
 pub mod object {
     pub use crate::object_api::{
-        BLOCK_SIZE_V2, ERASURE_ALGORITHM, EncryptionResolutionError, EncryptionResolutionErrorKind, GetObjectBodyCacheHook,
-        GetObjectBodyCacheHookLookup, GetObjectBodySource, GetObjectReader, ObjectEncryptionResolver, ObjectInfo,
-        ObjectMutationHook, ObjectOptions, PutObjReader, RangedDecompressReader, ReadEncryptionMaterial, ReadEncryptionMode,
-        ReadEncryptionRequest, StreamConsumer, get_object_body_cache_plaintext_len, lookup_get_object_body_cache_hook,
-        register_get_object_body_cache_hook, register_object_mutation_hook, unregister_get_object_body_cache_hook,
-        unregister_object_mutation_hook,
+        BLOCK_SIZE_V2, ERASURE_ALGORITHM, GetObjectBodyCacheHook, GetObjectBodyCacheHookLookup, GetObjectBodySource,
+        GetObjectReader, ObjectInfo, ObjectMutationHook, ObjectOptions, PutObjReader, RangedDecompressReader, StreamConsumer,
+        get_object_body_cache_plaintext_len, lookup_get_object_body_cache_hook, register_get_object_body_cache_hook,
+        register_object_mutation_hook, unregister_get_object_body_cache_hook, unregister_object_mutation_hook,
     };
     pub use crate::store::PreparedGetObjectReader;
 }

crates/ecstore/src/bucket/metadata_sys.rs

@@ -35,7 +35,10 @@ use s3s::dto::{
 };
 use std::collections::HashSet;
 use std::time::Duration;
-use std::{collections::HashMap, sync::Arc};
+use std::{
+    collections::HashMap,
+    sync::{Arc, Mutex as StdMutex, Weak},
+};
 use time::OffsetDateTime;
 use tokio::sync::{Mutex, RwLock};
 use tokio::time::sleep;
@@ -91,18 +94,17 @@ pub async fn set_bucket_metadata(bucket: String, bm: BucketMetadata) -> Result<(
     Ok(())
 }
 
-/// Peer LoadBucketMetadata entry point; see
-/// [`BucketMetadataSys::reload_from_store`] for the caching contract.
-///
-/// The outer write guard spans the disk load, mirroring [`update`]: every
-/// other cache installer holds this lock (read or write), so the snapshot
-/// read here can never land after — and roll back — a newer concurrent
-/// install, and the install-plus-registry-sync sequence stays atomic
-/// against concurrent removes and reloads.
-pub async fn reload_bucket_metadata(bucket: &str) -> Result<()> {
-    let sys = get_bucket_metadata_sys()?;
-    let lock = sys.write().await;
-    lock.reload_from_store(bucket).await
+pub async fn reload_bucket_metadata(api: Arc<ECStore>, bucket: &str) -> Result<()> {
+    if is_meta_bucketname(bucket) {
+        return Err(Error::other("errInvalidArgument"));
+    }
+    let namespace_lock = api.new_ns_lock(bucket, bucket).await?;
+    let namespace_guard = namespace_lock
+        .get_read_lock(crate::set_disk::get_lock_acquire_timeout())
+        .await?;
+    let sys = bucket_metadata_sys_of(&api.ctx)?;
+    let lock = sys.read().await;
+    lock.reload_from_store_under_namespace(bucket, &namespace_guard).await
 }
 
 /// Drop a bucket's cached metadata from the in-memory map.
@@ -159,9 +161,7 @@ async fn refresh_buckets_metadata_once(sys: Arc<RwLock<BucketMetadataSys>>) {
     let mut failed_buckets = HashSet::new();
 
     for chunk in buckets.chunks(count) {
-        let sys = sys.read().await;
-        sys.concurrent_load(chunk, &mut failed_buckets, MetadataLoadMode::Refresh)
-            .await;
+        BucketMetadataSys::concurrent_refresh_load(Arc::clone(&sys), chunk, &mut failed_buckets).await;
     }
 
     if !failed_buckets.is_empty() {
@@ -478,11 +478,42 @@ pub async fn list_bucket_targets(bucket: &str) -> Result<BucketTargets> {
 /// notification was lost; the capacity bounds memory under bogus-name floods.
 const ABSENT_BUCKET_METADATA_TTL: Duration = Duration::from_secs(30);
 const ABSENT_BUCKET_METADATA_MAX_ENTRIES: u64 = 10_000;
+const PEER_METADATA_NOT_PERSISTED: &str = "no persisted bucket metadata readable; peer cache left unchanged";
+#[derive(Debug)]
+struct MetadataPublishLockRegistry {
+    locks: StdMutex<HashMap<String, Weak<Mutex<MetadataPublishLockState>>>>,
+}
+
+#[derive(Debug)]
+struct MetadataPublishLockState {
+    bucket: String,
+    registry: Weak<MetadataPublishLockRegistry>,
+    lock: Weak<Mutex<MetadataPublishLockState>>,
+}
+
+impl Drop for MetadataPublishLockState {
+    fn drop(&mut self) {
+        let Some(registry) = self.registry.upgrade() else {
+            return;
+        };
+        let mut locks = registry.locks.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
+        if locks.get(&self.bucket).is_some_and(|current| current.ptr_eq(&self.lock)) {
+            locks.remove(&self.bucket);
+        }
+    }
+}
+
+#[derive(Debug)]
+struct MetadataPublishGuard {
+    _guard: tokio::sync::OwnedMutexGuard<MetadataPublishLockState>,
+}
 
 #[derive(Debug)]
 pub struct BucketMetadataSys {
     metadata_map: RwLock<HashMap<String, Arc<BucketMetadata>>>,
-    metadata_publish_lock: Mutex<()>,
+    /// Serializes metadata-map commits and their derived cache updates for one
+    /// bucket. Namespace locks, when present, are acquired before this lock.
+    metadata_publish_locks: Arc<MetadataPublishLockRegistry>,
     #[cfg(test)]
     lazy_load_lock_probe: std::sync::atomic::AtomicBool,
     /// Buckets recently observed to have no persisted metadata. Serving the
@@ -500,7 +531,9 @@ impl BucketMetadataSys {
     pub fn new(api: Arc<ECStore>) -> Self {
         Self {
             metadata_map: RwLock::new(HashMap::new()),
-            metadata_publish_lock: Mutex::new(()),
+            metadata_publish_locks: Arc::new(MetadataPublishLockRegistry {
+                locks: StdMutex::new(HashMap::new()),
+            }),
             #[cfg(test)]
             lazy_load_lock_probe: std::sync::atomic::AtomicBool::new(false),
             absent_metadata: moka::future::Cache::builder()
@@ -516,6 +549,65 @@ impl BucketMetadataSys {
         self.api.clone()
     }
 
+    fn metadata_publish_lock(&self, bucket: &str) -> Arc<Mutex<MetadataPublishLockState>> {
+        let mut locks = self
+            .metadata_publish_locks
+            .locks
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner());
+        locks.get(bucket).and_then(Weak::upgrade).unwrap_or_else(|| {
+            let lock = Arc::new_cyclic(|lock| {
+                Mutex::new(MetadataPublishLockState {
+                    bucket: bucket.to_string(),
+                    registry: Arc::downgrade(&self.metadata_publish_locks),
+                    lock: lock.clone(),
+                })
+            });
+            locks.insert(bucket.to_string(), Arc::downgrade(&lock));
+            lock
+        })
+    }
+
+    async fn lock_metadata_publish(
+        &self,
+        bucket: &str,
+        namespace_guard: &rustfs_lock::NamespaceLockGuard,
+        operation: &'static str,
+    ) -> Result<MetadataPublishGuard> {
+        let lock = self.metadata_publish_lock(bucket);
+        let guard = await_bucket_namespace_operation(Some(namespace_guard), bucket, operation, async {
+            Ok(MetadataPublishGuard {
+                _guard: lock.lock_owned().await,
+            })
+        })
+        .await?;
+        if namespace_guard.is_lock_lost() {
+            return Err(Error::other(format!("bucket namespace lock was lost before {operation}: {bucket}")));
+        }
+        Ok(guard)
+    }
+
+    async fn bucket_exists(
+        &self,
+        bucket: &str,
+        namespace_guard: &rustfs_lock::NamespaceLockGuard,
+        operation: &'static str,
+    ) -> Result<bool> {
+        await_bucket_namespace_operation(Some(namespace_guard), bucket, operation, async {
+            match self
+                .api
+                .peer_sys
+                .get_bucket_info(bucket, &crate::storage_api_contracts::bucket::BucketOptions::default())
+                .await
+            {
+                Ok(_) => Ok(true),
+                Err(crate::disk::error::Error::VolumeNotFound) => Ok(false),
+                Err(err) => Err(err.into()),
+            }
+        })
+        .await
+    }
+
     pub async fn init(&mut self, buckets: Vec<String>) {
         let _ = self.init_internal(buckets).await;
     }
@@ -554,55 +646,16 @@ impl BucketMetadataSys {
             let bucket = bucket.clone();
             futures.push(async move {
                 sleep(Duration::from_millis(30)).await;
-                match mode {
-                    MetadataLoadMode::Initial => {
-                        let _ = api
-                            .heal_bucket(
-                                &bucket,
-                                &HealOpts {
-                                    recreate: true,
-                                    ..Default::default()
-                                },
-                            )
-                            .await;
-                        let (bm, persisted) =
-                            load_bucket_metadata_parse_with_presence(self.api.clone(), bucket.as_str(), true).await?;
-                        if persisted {
-                            self.set(bucket, Arc::new(bm)).await;
-                        } else {
-                            let _publish_guard = self.metadata_publish_lock.lock().await;
-                            let mut map = self.metadata_map.write().await;
-                            map.entry(bucket).or_insert_with(|| Arc::new(bm));
-                        }
-                    }
-                    MetadataLoadMode::Refresh => {
-                        let expected = self.metadata_map.read().await.get(&bucket).cloned();
-                        let heal_lock = api.new_ns_lock(&bucket, &bucket).await?;
-                        let heal_guard = heal_lock.get_read_lock(crate::set_disk::get_lock_acquire_timeout()).await?;
-                        await_bucket_namespace_operation(
-                            Some(&heal_guard),
-                            &bucket,
-                            "bucket metadata refresh heal",
-                            api.heal_bucket(&bucket, &HealOpts::default()),
-                        )
-                        .await?;
-                        drop(heal_guard);
-                        let (bm, persisted) =
-                            load_bucket_metadata_parse_with_presence(self.api.clone(), bucket.as_str(), true).await?;
-                        let publish_lock = api.new_ns_lock(&bucket, &bucket).await?;
-                        let guard = publish_lock
-                            .get_read_lock(crate::set_disk::get_lock_acquire_timeout())
-                            .await?;
-                        if guard.is_lock_lost() {
-                            return Err(Error::other(format!(
-                                "bucket namespace lock was lost before bucket metadata refresh publish: {bucket}"
-                            )));
-                        }
-                        self.publish_refresh_if_unchanged(&bucket, expected.as_ref(), bm, persisted)
-                            .await;
-                    }
-                }
-                Ok::<(), Error>(())
+                let expected = match mode {
+                    MetadataLoadMode::Initial => None,
+                    MetadataLoadMode::Refresh => self.metadata_map.read().await.get(&bucket).cloned(),
+                };
+                let namespace_lock = api.new_ns_lock(&bucket, &bucket).await?;
+                let namespace_guard = namespace_lock
+                    .get_read_lock(crate::set_disk::get_lock_acquire_timeout())
+                    .await?;
+                self.load_bucket_under_namespace(&bucket, mode, expected.as_ref(), &namespace_guard)
+                    .await
             });
         }
 
@@ -621,30 +674,134 @@ impl BucketMetadataSys {
         }
     }
 
-    async fn publish_refresh_if_unchanged(
+    async fn concurrent_refresh_load(sys: Arc<RwLock<Self>>, buckets: &[String], failed_buckets: &mut HashSet<String>) {
+        let mut futures = Vec::with_capacity(buckets.len());
+        for bucket in buckets {
+            let sys = Arc::clone(&sys);
+            let bucket = bucket.clone();
+            futures.push(async move {
+                sleep(Duration::from_millis(30)).await;
+                let api = sys.read().await.api.clone();
+                let namespace_lock = api.new_ns_lock(&bucket, &bucket).await?;
+                let namespace_guard = namespace_lock
+                    .get_read_lock(crate::set_disk::get_lock_acquire_timeout())
+                    .await?;
+                let metadata_sys = sys.read().await;
+                let expected = metadata_sys.metadata_map.read().await.get(&bucket).cloned();
+                metadata_sys
+                    .load_bucket_under_namespace(&bucket, MetadataLoadMode::Refresh, expected.as_ref(), &namespace_guard)
+                    .await
+            });
+        }
+        let results = join_all(futures).await;
+        for (idx, result) in results.into_iter().enumerate() {
+            if let Err(err) = result {
+                error!("Unable to load bucket metadata, will be retried: {:?}", err);
+                if let Some(bucket) = buckets.get(idx) {
+                    failed_buckets.insert(bucket.clone());
+                }
+            }
+        }
+    }
+
+    async fn load_bucket_under_namespace(
+        &self,
+        bucket: &str,
+        mode: MetadataLoadMode,
+        expected: Option<&Arc<BucketMetadata>>,
+        namespace_guard: &rustfs_lock::NamespaceLockGuard,
+    ) -> Result<()> {
+        await_bucket_namespace_operation(
+            Some(namespace_guard),
+            bucket,
+            "bucket metadata heal",
+            self.api.heal_bucket(bucket, &HealOpts::default()),
+        )
+        .await?;
+
+        if !self
+            .bucket_exists(bucket, namespace_guard, "bucket metadata existence check")
+            .await?
+        {
+            if matches!(mode, MetadataLoadMode::Refresh) {
+                let _publish_guard = self
+                    .lock_metadata_publish(bucket, namespace_guard, "stale bucket metadata removal")
+                    .await?;
+                let removed = self.metadata_map.write().await.remove(bucket).is_some();
+                if removed {
+                    BucketTargetSys::get().delete(bucket).await;
+                    clear_bucket_durability(bucket);
+                }
+            }
+            return Ok(());
+        }
+
+        let (bm, persisted) = await_bucket_namespace_operation(
+            Some(namespace_guard),
+            bucket,
+            "bucket metadata load",
+            load_bucket_metadata_parse_with_presence(self.api.clone(), bucket, true),
+        )
+        .await?;
+        match mode {
+            MetadataLoadMode::Initial if persisted => {
+                let bm = Arc::new(bm);
+                let _publish_guard = self
+                    .lock_metadata_publish(bucket, namespace_guard, "initial bucket metadata publish")
+                    .await?;
+                self.metadata_map.write().await.insert(bucket.to_string(), Arc::clone(&bm));
+                self.absent_metadata.invalidate(bucket).await;
+                sync_bucket_target_sys(bucket, &bm).await;
+                sync_bucket_durability(bucket, &bm);
+            }
+            MetadataLoadMode::Initial => {
+                let _publish_guard = self
+                    .lock_metadata_publish(bucket, namespace_guard, "initial bucket metadata publish")
+                    .await?;
+                self.metadata_map
+                    .write()
+                    .await
+                    .entry(bucket.to_string())
+                    .or_insert_with(|| Arc::new(bm));
+            }
+            MetadataLoadMode::Refresh => {
+                self.publish_if_unchanged(bucket, expected, bm, persisted, namespace_guard)
+                    .await?;
+            }
+        }
+        Ok(())
+    }
+
+    async fn publish_if_unchanged(
         &self,
         bucket: &str,
         expected: Option<&Arc<BucketMetadata>>,
         metadata: BucketMetadata,
         persisted: bool,
-    ) {
+        namespace_guard: &rustfs_lock::NamespaceLockGuard,
+    ) -> Result<()> {
         if !persisted {
-            return;
+            return Ok(());
         }
-        let _publish_guard = self.metadata_publish_lock.lock().await;
+        let _publish_guard = self
+            .lock_metadata_publish(bucket, namespace_guard, "refreshed bucket metadata publish")
+            .await?;
         let metadata = Arc::new(metadata);
         let mut map = self.metadata_map.write().await;
-        let unchanged = expected
-            .zip(map.get(bucket))
-            .is_some_and(|(expected, current)| Arc::ptr_eq(expected, current));
+        let unchanged = match (expected, map.get(bucket)) {
+            (None, None) => true,
+            (Some(expected), Some(current)) => Arc::ptr_eq(expected, current),
+            _ => false,
+        };
         if !unchanged {
-            return;
+            return Ok(());
         }
         map.insert(bucket.to_string(), Arc::clone(&metadata));
         drop(map);
         self.absent_metadata.invalidate(bucket).await;
         sync_bucket_target_sys(bucket, &metadata).await;
         sync_bucket_durability(bucket, &metadata);
+        Ok(())
     }
 
     pub async fn get(&self, bucket: &str) -> Result<Arc<BucketMetadata>> {
@@ -662,7 +819,8 @@ impl BucketMetadataSys {
 
     pub async fn set(&self, bucket: String, bm: Arc<BucketMetadata>) {
         if !is_meta_bucketname(&bucket) {
-            let _publish_guard = self.metadata_publish_lock.lock().await;
+            let publish_lock = self.metadata_publish_lock(&bucket);
+            let _publish_guard = publish_lock.lock().await;
             let mut map = self.metadata_map.write().await;
             map.insert(bucket.clone(), bm.clone());
             drop(map);
@@ -673,43 +831,6 @@ impl BucketMetadataSys {
         }
     }
 
-    /// Reload `bucket`'s metadata from this system's own store and cache it,
-    /// refusing to treat a load miss as authoritative (the peer
-    /// LoadBucketMetadata notification path, [`reload_bucket_metadata`]).
-    ///
-    /// Only metadata actually read from persisted storage reaches the cache.
-    /// On a miss the fabricated default is discarded and an error is
-    /// returned: installing it would let a transient ConfigNotFound during
-    /// the notification overwrite a lock-enabled bucket's cached metadata
-    /// with an authoritative "no Object Lock" default, disabling the
-    /// batch-delete retention gate (`object_lock_delete_check_required`) on
-    /// this node until the next refresh. A miss is also not treated as
-    /// deletion: bucket deletion propagates through the dedicated
-    /// DeleteBucketMetadata notification ([`remove_bucket_metadata`]), which
-    /// is best-effort — a reload racing it can still re-install a just
-    /// deleted bucket's entry (pre-existing, bounded by the next delete or
-    /// restart) — but a reload miss removing entries would turn every
-    /// transient quorum dip into dropped metadata and spurious
-    /// target/durability teardown.
-    ///
-    /// The peer-visible error text is deliberately fixed: the notifying peer
-    /// matches error strings against network-failure needles
-    /// (`is_network_like_error`), so interpolating a caller-controlled
-    /// bucket name here could mark a healthy peer offline.
-    ///
-    /// Lock order: the caller holds the outer metadata-sys guard, and the
-    /// load acquires the namespace lock on the bucket's metadata config
-    /// object — the same `outer guard → meta-config namespace lock` order
-    /// `update`'s load takes; no path acquires these in reverse.
-    pub(crate) async fn reload_from_store(&self, bucket: &str) -> Result<()> {
-        let (bm, persisted) = load_bucket_metadata_parse_with_presence(self.api.clone(), bucket, true).await?;
-        if !persisted {
-            return Err(Error::other("no persisted bucket metadata readable; peer cache left unchanged"));
-        }
-        self.set(bucket.to_string(), Arc::new(bm)).await;
-        Ok(())
-    }
-
     /// Remove a bucket's cached metadata from the in-memory map.
     ///
     /// Returns `true` if an entry was present. Reserved meta buckets are ignored.
@@ -717,7 +838,8 @@ impl BucketMetadataSys {
         if is_meta_bucketname(bucket) {
             return false;
         }
-        let _publish_guard = self.metadata_publish_lock.lock().await;
+        let publish_lock = self.metadata_publish_lock(bucket);
+        let _publish_guard = publish_lock.lock().await;
         let mut map = self.metadata_map.write().await;
         let removed = map.remove(bucket).is_some();
         drop(map);
@@ -831,6 +953,49 @@ impl BucketMetadataSys {
         load_bucket_metadata(self.api.clone(), bucket).await
     }
 
+    /// Reload persisted metadata under the bucket namespace generation fence.
+    ///
+    /// A miss is never published as an authoritative default, and a snapshot
+    /// read before delete plus same-name recreation cannot replace the new
+    /// generation.
+    pub(crate) async fn reload_from_store(&self, bucket: &str) -> Result<()> {
+        if is_meta_bucketname(bucket) {
+            return Err(Error::other("errInvalidArgument"));
+        }
+
+        let namespace_lock = self.api.new_ns_lock(bucket, bucket).await?;
+        let namespace_guard = namespace_lock
+            .get_read_lock(crate::set_disk::get_lock_acquire_timeout())
+            .await?;
+        self.reload_from_store_under_namespace(bucket, &namespace_guard).await
+    }
+
+    async fn reload_from_store_under_namespace(
+        &self,
+        bucket: &str,
+        namespace_guard: &rustfs_lock::NamespaceLockGuard,
+    ) -> Result<()> {
+        let expected = self.metadata_map.read().await.get(bucket).cloned();
+        if !self
+            .bucket_exists(bucket, namespace_guard, "peer bucket metadata existence check")
+            .await?
+        {
+            return Err(Error::other(PEER_METADATA_NOT_PERSISTED));
+        }
+        let (metadata, persisted) = await_bucket_namespace_operation(
+            Some(namespace_guard),
+            bucket,
+            "peer bucket metadata load",
+            load_bucket_metadata_parse_with_presence(self.api.clone(), bucket, true),
+        )
+        .await?;
+        if !persisted {
+            return Err(Error::other(PEER_METADATA_NOT_PERSISTED));
+        }
+        self.publish_if_unchanged(bucket, expected.as_ref(), metadata, true, namespace_guard)
+            .await
+    }
+
     pub async fn get_config(&self, bucket: &str) -> Result<(Arc<BucketMetadata>, bool)> {
         let has_bm = {
             let map = self.metadata_map.read().await;
@@ -909,7 +1074,9 @@ impl BucketMetadataSys {
                         "bucket namespace lock was lost before lazy bucket metadata publish: {bucket}"
                     )));
                 }
-                let _publish_guard = self.metadata_publish_lock.lock().await;
+                let _publish_guard = self
+                    .lock_metadata_publish(bucket, &guard, "lazy bucket metadata publish")
+                    .await?;
                 let mut map = self.metadata_map.write().await;
                 if let Some(current) = map.get(bucket) {
                     return Ok((Arc::clone(current), true));
@@ -1247,6 +1414,11 @@ mod tests {
             .await
             .expect("lazily loaded persisted metadata must be cached");
         assert_eq!(cached.policy_config_json, b"persisted-marker".to_vec());
+        sys.metadata_map.write().await.clear();
+        sys.reload_from_store("absent-bucket")
+            .await
+            .expect("peer reload should publish persisted metadata into a cold cache");
+        assert_eq!(sys.get("absent-bucket").await.unwrap().policy_config_json, b"persisted-marker".to_vec());
 
         // (c) Persisted metadata left behind after physical deletion must not
         // be lazily republished as a live bucket generation.
@@ -1299,8 +1471,8 @@ mod tests {
             "a fabricated refresh default must not replace real metadata"
         );
 
-        // (f) A stale cache entry for a physically deleted bucket must not
-        // recreate the bucket during periodic refresh.
+        // (f) A stale cache entry for a physically deleted bucket must be
+        // removed without recreating the bucket during periodic refresh.
         sys.set("deleted-bucket".to_string(), Arc::new(BucketMetadata::new("deleted-bucket")))
             .await;
         let deleted_targets = vec!["deleted-bucket".to_string()];
@@ -1310,8 +1482,30 @@ mod tests {
             dirs.iter().all(|dir| !dir.path().join("deleted-bucket").exists()),
             "periodic refresh must not recreate a bucket from stale cached metadata"
         );
+        assert!(
+            sys.get("deleted-bucket").await.is_err(),
+            "periodic refresh must remove stale cached metadata"
+        );
 
-        // (g) Metadata loaded for an old bucket generation must not replace
+        // (g) Persisted metadata left behind after physical deletion must not
+        // keep the deleted generation authoritative during refresh.
+        let mut deleted_persisted = BucketMetadata::new("deleted-persisted-bucket");
+        deleted_persisted.policy_config_json = b"stale-persisted-generation".to_vec();
+        sys.persist_and_set(deleted_persisted)
+            .await
+            .expect("stale metadata should persist");
+        sys.concurrent_load(&["deleted-persisted-bucket".to_string()], &mut failed, MetadataLoadMode::Refresh)
+            .await;
+        assert!(
+            sys.get("deleted-persisted-bucket").await.is_err(),
+            "refresh must remove persisted metadata for a physically absent bucket"
+        );
+        assert!(
+            sys.reload_from_store("deleted-persisted-bucket").await.is_err(),
+            "peer reload must not publish stale metadata for an absent bucket"
+        );
+
+        // (h) Metadata loaded for an old bucket generation must not replace
         // metadata published by delete plus same-name recreation.
         let old = Arc::new(BucketMetadata::new("recreated-bucket"));
         sys.set("recreated-bucket".to_string(), Arc::clone(&old)).await;
@@ -1320,11 +1514,27 @@ mod tests {
         sys.set("recreated-bucket".to_string(), Arc::new(recreated)).await;
         let mut stale = BucketMetadata::new("recreated-bucket");
         stale.policy_config_json = b"old-generation".to_vec();
-        sys.publish_refresh_if_unchanged("recreated-bucket", Some(&old), stale, true)
-            .await;
-        assert_eq!(sys.get("recreated-bucket").await.unwrap().policy_config_json, b"new-generation".to_vec());
+        let namespace_lock = sys
+            .api
+            .new_ns_lock("recreated-bucket", "recreated-bucket")
+            .await
+            .expect("namespace lock should be created");
+        let namespace_guard = namespace_lock
+            .get_read_lock(crate::set_disk::get_lock_acquire_timeout())
+            .await
+            .expect("namespace read lock should be acquired");
+        sys.publish_if_unchanged("recreated-bucket", Some(&old), stale, true, &namespace_guard)
+            .await
+            .expect("stale refresh publish should be fenced");
+        assert_eq!(
+            sys.get("recreated-bucket")
+                .await
+                .expect("recreated bucket metadata should remain cached")
+                .policy_config_json,
+            b"new-generation".to_vec()
+        );
 
-        // (f) Refresh retains periodic healing for a partially missing bucket.
+        // (i) Refresh retains periodic healing for a partially missing bucket.
         sys.set("partial-bucket".to_string(), Arc::new(BucketMetadata::new("partial-bucket")))
             .await;
         for dir in dirs.iter().take(3) {
@@ -1334,7 +1544,22 @@ mod tests {
             .await;
         assert!(dirs.iter().all(|dir| dir.path().join("partial-bucket").is_dir()));
 
-        // (g) Initial discovery retains the historical unconditional heal.
+        // (j) A stale initial snapshot must not recreate a bucket that has
+        // disappeared from every disk.
+        let stale_initial_targets = vec!["deleted-initial-bucket".to_string()];
+        sys.concurrent_load(&stale_initial_targets, &mut failed, MetadataLoadMode::Initial)
+            .await;
+        assert!(
+            dirs.iter().all(|dir| !dir.path().join("deleted-initial-bucket").exists()),
+            "initial load must not recreate a bucket absent from every disk"
+        );
+
+        // (k) Initial discovery still heals a bucket present on part of the
+        // storage topology.
+        for dir in dirs.iter().take(3) {
+            std::fs::create_dir_all(dir.path().join("initial-bucket"))
+                .expect("partial initial bucket directory should be created");
+        }
         let initial_targets = vec!["initial-bucket".to_string()];
         sys.concurrent_load(&initial_targets, &mut failed, MetadataLoadMode::Initial)
             .await;
@@ -1344,6 +1569,46 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn metadata_publish_locks_are_isolated_per_bucket() {
+        let (_dirs, ecstore) = isolated_store_over_temp_disks().await;
+        let sys = Arc::new(BucketMetadataSys::new(ecstore));
+        let first_lock = sys.metadata_publish_lock("blocked-bucket");
+        let same_lock = sys.metadata_publish_lock("blocked-bucket");
+        assert!(Arc::ptr_eq(&first_lock, &same_lock));
+        let first_guard = first_lock.lock().await;
+        let cancelled_waiter_lock = sys.metadata_publish_lock("blocked-bucket");
+        let cancelled_waiter = tokio::spawn(async move {
+            let _guard = cancelled_waiter_lock.lock_owned().await;
+        });
+        tokio::task::yield_now().await;
+        cancelled_waiter.abort();
+        assert!(cancelled_waiter.await.unwrap_err().is_cancelled());
+        let other_bucket = "other-bucket".to_string();
+        let other_lock = sys.metadata_publish_lock(&other_bucket);
+        assert!(!Arc::ptr_eq(&first_lock, &other_lock));
+
+        timeout(
+            Duration::from_secs(1),
+            sys.set(other_bucket.clone(), Arc::new(BucketMetadata::new(&other_bucket))),
+        )
+        .await
+        .expect("one bucket publish lock must not block another bucket");
+        assert!(sys.get(&other_bucket).await.is_ok());
+
+        drop(first_guard);
+        drop(first_lock);
+        drop(same_lock);
+        drop(other_lock);
+        assert!(
+            sys.metadata_publish_locks
+                .locks
+                .lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner())
+                .is_empty()
+        );
+    }
+
     #[tokio::test]
     async fn get_bucket_policy_rejects_malformed_cached_policy() {
         let (_dirs, ecstore) = isolated_store_over_temp_disks().await;
@@ -1492,7 +1757,7 @@ mod tests {
     /// default and disable the batch-delete retention gate on this peer.
     #[tokio::test]
     async fn peer_reload_never_caches_fabricated_defaults_as_authoritative() {
-        let (_dirs, ecstore) = isolated_store_over_temp_disks().await;
+        let (dirs, ecstore) = isolated_store_over_temp_disks().await;
         let sys = BucketMetadataSys::new(ecstore.clone());
 
         // (a) Miss with no cached entry: the reload fails and installs nothing.
@@ -1530,6 +1795,9 @@ mod tests {
         let mut persisted = BucketMetadata::new("reload-bucket");
         persisted.policy_config_json = b"persisted-marker".to_vec();
         sys.persist_and_set(persisted).await.expect("metadata should persist");
+        for dir in &dirs {
+            std::fs::create_dir_all(dir.path().join("reload-bucket")).expect("physical bucket should exist before reload");
+        }
         let mut stale = BucketMetadata::new("reload-bucket");
         stale.policy_config_json = b"stale-cache-marker".to_vec();
         sys.set("reload-bucket".to_string(), Arc::new(stale)).await;

crates/ecstore/src/client/utils.rs

@@ -46,6 +46,16 @@ lazy_static! {
         m.insert("x-amz-replication-status".to_string(), true);
         m
     };
+    static ref SSE_HEADERS: HashMap<String, bool> = {
+        let mut m = HashMap::new();
+        m.insert("x-amz-server-side-encryption".to_string(), true);
+        m.insert("x-amz-server-side-encryption-aws-kms-key-id".to_string(), true);
+        m.insert("x-amz-server-side-encryption-context".to_string(), true);
+        m.insert("x-amz-server-side-encryption-customer-algorithm".to_string(), true);
+        m.insert("x-amz-server-side-encryption-customer-key".to_string(), true);
+        m.insert("x-amz-server-side-encryption-customer-key-md5".to_string(), true);
+        m
+    };
 }
 
 pub fn is_standard_query_value(qs_key: &str) -> bool {
@@ -60,12 +70,16 @@ pub fn is_standard_header(header_key: &str) -> bool {
     *SUPPORTED_HEADERS.get(&header_key.to_lowercase()).unwrap_or(&false)
 }
 
+pub fn is_sse_header(header_key: &str) -> bool {
+    *SSE_HEADERS.get(&header_key.to_lowercase()).unwrap_or(&false)
+}
+
 pub fn is_amz_header(header_key: &str) -> bool {
     let key = header_key.to_lowercase();
     key.starts_with("x-amz-meta-")
         || key.starts_with("x-amz-grant-")
         || key == "x-amz-acl"
-        || rustfs_utils::http::is_sse_header(header_key)
+        || is_sse_header(header_key)
         || key.starts_with("x-amz-checksum-")
 }
 

crates/ecstore/src/disk/local.rs

@@ -77,6 +77,8 @@ const STALE_TMP_OBJECT_EXPIRY: Duration = Duration::from_secs(24 * 60 * 60);
 const RUSTFS_META_TMP_OLD_BUCKET: &str = ".rustfs.sys/tmp-old";
 const INLINE_METADATA_ROLLBACK_DIR_XOR: u128 = 0x7275737466735f696e6c696e655f7262;
 const DELETE_MARKER_ROLLBACK_FILE: &str = "xl.meta.delete-marker.rollback";
+pub(crate) const DELETE_DATA_DIR_MARKER_PREFIX: &str = "delete-data.";
+pub(crate) const RESERVED_DELETE_DATA_DIR_MARKER_PREFIX: &str = "reserve-delete-data.";
 const STARTUP_CLEANUP_WAIT_TIMEOUT: Duration = Duration::from_secs(2);
 const ENV_BITROT_SIZE_MISMATCH_RETRY_COUNT: &str = "RUSTFS_BITROT_SIZE_MISMATCH_RETRY_COUNT";
 const ENV_BITROT_SIZE_MISMATCH_RETRY_DELAY_MS: &str = "RUSTFS_BITROT_SIZE_MISMATCH_RETRY_DELAY_MS";
@@ -243,6 +245,7 @@ async fn restore_metadata_backup(object_dir: &Path, xl_path: &Path, rollback_dir
 }
 
 async fn restore_delete_rollback(object_dir: &Path, xl_path: &Path, rollback_dir: Uuid) -> Result<()> {
+    remove_version_delete_markers(object_dir, rollback_dir).await?;
     let rollback_path = object_dir.join(rollback_dir.to_string());
     let mut staged_paths = Vec::new();
     let mut remove_new_metadata = false;
@@ -300,6 +303,31 @@ async fn restore_delete_rollback(object_dir: &Path, xl_path: &Path, rollback_dir
     }
 }
 
+async fn remove_version_delete_markers(object_dir: &Path, rollback_dir: Uuid) -> Result<()> {
+    let reserved_name = format!("{RESERVED_DELETE_DATA_DIR_MARKER_PREFIX}{rollback_dir}");
+    let committed_name = format!("{DELETE_DATA_DIR_MARKER_PREFIX}{rollback_dir}");
+    let mut entries = match fs::read_dir(object_dir).await {
+        Ok(entries) => entries,
+        Err(err) if err.kind() == ErrorKind::NotFound => return Ok(()),
+        Err(err) => return Err(to_file_error(err).into()),
+    };
+    while let Some(entry) = entries.next_entry().await.map_err(to_file_error)? {
+        if !entry.file_type().await.map_err(to_file_error)?.is_dir()
+            || !entry.file_name().to_str().is_some_and(|name| Uuid::parse_str(name).is_ok())
+        {
+            continue;
+        }
+        for marker_name in [&reserved_name, &committed_name] {
+            match fs::remove_file(entry.path().join(marker_name)).await {
+                Ok(()) => {}
+                Err(err) if err.kind() == ErrorKind::NotFound => {}
+                Err(err) => return Err(to_file_error(err).into()),
+            }
+        }
+    }
+    Ok(())
+}
+
 async fn restore_delete_rollback_after_error(
     object_dir: &Path,
     xl_path: &Path,
@@ -4917,6 +4945,7 @@ impl LocalDisk {
 
         fm.unmarshal_msg(&data)?;
         let rollback_dir = opts.old_data_dir;
+        let mut reserved_version_delete = false;
         if let Some(rollback_dir) = rollback_dir {
             write_metadata_rollback_backup(object_dir, rollback_dir, &data).await?;
         }
@@ -4930,6 +4959,18 @@ impl LocalDisk {
                         continue;
                     }
 
+                    if reserved_version_delete && let Some(rollback_dir) = rollback_dir {
+                        return Err(self
+                            .abort_reserved_version_delete(
+                                object_dir,
+                                rollback_dir,
+                                volume,
+                                path,
+                                "delete_versions_metadata_update",
+                                err,
+                            )
+                            .await);
+                    }
                     return Err(restore_delete_rollback_after_error(
                         object_dir,
                         &xlpath,
@@ -4950,6 +4991,18 @@ impl LocalDisk {
                 let dir_path = match self.get_object_path(volume, format!("{path}/{dir}").as_str()) {
                     Ok(dir_path) => dir_path,
                     Err(err) => {
+                        if reserved_version_delete && let Some(rollback_dir) = rollback_dir {
+                            return Err(self
+                                .abort_reserved_version_delete(
+                                    object_dir,
+                                    rollback_dir,
+                                    volume,
+                                    path,
+                                    "delete_versions_data_path",
+                                    err,
+                                )
+                                .await);
+                        }
                         return Err(restore_delete_rollback_after_error(
                             object_dir,
                             &xlpath,
@@ -4966,6 +5019,18 @@ impl LocalDisk {
                     let rollback_path = object_dir.join(rollback_dir.to_string());
                     if let Err(err) = fs::create_dir_all(&rollback_path).await {
                         let err: DiskError = to_file_error(err).into();
+                        if reserved_version_delete {
+                            return Err(self
+                                .abort_reserved_version_delete(
+                                    object_dir,
+                                    rollback_dir,
+                                    volume,
+                                    path,
+                                    "delete_versions_rollback_dir",
+                                    err,
+                                )
+                                .await);
+                        }
                         return Err(restore_delete_rollback_after_error(
                             object_dir,
                             &xlpath,
@@ -4977,8 +5042,26 @@ impl LocalDisk {
                         )
                         .await);
                     }
+                    let reserved = match self.reserve_version_delete(volume, path, dir, rollback_dir).await {
+                        Ok(reserved) => reserved,
+                        Err(err) => {
+                            return Err(self
+                                .abort_reserved_version_delete(
+                                    object_dir,
+                                    rollback_dir,
+                                    volume,
+                                    path,
+                                    "delete_versions_reserve_data",
+                                    err,
+                                )
+                                .await);
+                        }
+                    };
+                    reserved_version_delete |= reserved;
                     let rollback_data_path = rollback_path.join(dir.to_string());
-                    if let Err(err) = rename_all_ignore_missing_source(&dir_path, &rollback_data_path, &rollback_path).await {
+                    if !reserved
+                        && let Err(err) = rename_all_ignore_missing_source(&dir_path, &rollback_data_path, &rollback_path).await
+                    {
                         return Err(restore_delete_rollback_after_error(
                             object_dir,
                             &xlpath,
@@ -4991,6 +5074,18 @@ impl LocalDisk {
                         .await);
                     }
                     if should_fail_after_delete_data_staged(path) {
+                        if reserved_version_delete {
+                            return Err(self
+                                .abort_reserved_version_delete(
+                                    object_dir,
+                                    rollback_dir,
+                                    volume,
+                                    path,
+                                    "delete_versions_test_after_stage",
+                                    DiskError::Unexpected,
+                                )
+                                .await);
+                        }
                         return Err(restore_delete_rollback_after_error(
                             object_dir,
                             &xlpath,
@@ -5018,6 +5113,18 @@ impl LocalDisk {
         // Remove xl.meta when no versions remain
         if fm.versions.is_empty() {
             if let Err(err) = self.delete_file(&volume_dir, &xlpath, true, false).await {
+                if reserved_version_delete && let Some(rollback_dir) = rollback_dir {
+                    return Err(self
+                        .abort_reserved_version_delete(
+                            object_dir,
+                            rollback_dir,
+                            volume,
+                            path,
+                            "delete_versions_commit_delete",
+                            err,
+                        )
+                        .await);
+                }
                 return Err(restore_delete_rollback_after_error(
                     object_dir,
                     &xlpath,
@@ -5029,6 +5136,14 @@ impl LocalDisk {
                 )
                 .await);
             }
+            if reserved_version_delete
+                && let Some(rollback_dir) = rollback_dir
+                && let Err(err) = self.commit_reserved_version_delete(volume, path, rollback_dir).await
+            {
+                return Err(self
+                    .abort_reserved_version_delete(object_dir, rollback_dir, volume, path, "delete_versions_commit_intent", err)
+                    .await);
+            }
             return Ok(());
         }
 
@@ -5038,6 +5153,18 @@ impl LocalDisk {
             Ok(buf) => buf,
             Err(err) => {
                 let err: DiskError = err.into();
+                if reserved_version_delete && let Some(rollback_dir) = rollback_dir {
+                    return Err(self
+                        .abort_reserved_version_delete(
+                            object_dir,
+                            rollback_dir,
+                            volume,
+                            path,
+                            "delete_versions_metadata_encode",
+                            err,
+                        )
+                        .await);
+                }
                 return Err(restore_delete_rollback_after_error(
                     object_dir,
                     &xlpath,
@@ -5055,6 +5182,11 @@ impl LocalDisk {
             .write_all_meta(volume, format!("{path}/{STORAGE_FORMAT_FILE}").as_str(), &buf, true)
             .await
         {
+            if reserved_version_delete && let Some(rollback_dir) = rollback_dir {
+                return Err(self
+                    .abort_reserved_version_delete(object_dir, rollback_dir, volume, path, "delete_versions_commit_write", err)
+                    .await);
+            }
             return Err(restore_delete_rollback_after_error(
                 object_dir,
                 &xlpath,
@@ -5067,6 +5199,15 @@ impl LocalDisk {
             .await);
         }
 
+        if reserved_version_delete
+            && let Some(rollback_dir) = rollback_dir
+            && let Err(err) = self.commit_reserved_version_delete(volume, path, rollback_dir).await
+        {
+            return Err(self
+                .abort_reserved_version_delete(object_dir, rollback_dir, volume, path, "delete_versions_commit_intent", err)
+                .await);
+        }
+
         Ok(())
     }
 
@@ -6086,6 +6227,108 @@ fn normalize_path_components(path: impl AsRef<Path>) -> PathBuf {
     result
 }
 
+impl LocalDisk {
+    async fn reserve_version_delete(&self, volume: &str, object: &str, data_dir: Uuid, rollback_dir: Uuid) -> Result<bool> {
+        let path = format!("{object}/{data_dir}");
+        let data_path = self.get_object_path(volume, &path)?;
+        match fs::metadata(&data_path).await {
+            Ok(metadata) if metadata.is_dir() => {}
+            Ok(_) => return Ok(false),
+            Err(err) if err.kind() == ErrorKind::NotFound => return Ok(false),
+            Err(err) => return Err(to_file_error(err).into()),
+        }
+        let marker_path = data_path.join(format!("{RESERVED_DELETE_DATA_DIR_MARKER_PREFIX}{rollback_dir}"));
+        let marker = File::create(marker_path).await.map_err(to_file_error)?;
+        if effective_durability(volume).syncs_commit_metadata() {
+            marker.sync_all().await.map_err(to_file_error)?;
+            os::fsync_dir(&data_path).await.map_err(to_file_error)?;
+        }
+        Ok(true)
+    }
+
+    async fn commit_reserved_version_delete(&self, volume: &str, object: &str, rollback_dir: Uuid) -> Result<()> {
+        let object_path = self.get_object_path(volume, object)?;
+        let mut entries = match fs::read_dir(object_path).await {
+            Ok(entries) => entries,
+            Err(err) if err.kind() == ErrorKind::NotFound => return Ok(()),
+            Err(err) => return Err(to_file_error(err).into()),
+        };
+        let reserved_name = format!("{RESERVED_DELETE_DATA_DIR_MARKER_PREFIX}{rollback_dir}");
+        let committed_name = format!("{DELETE_DATA_DIR_MARKER_PREFIX}{rollback_dir}");
+        while let Some(entry) = entries.next_entry().await.map_err(to_file_error)? {
+            if !entry.file_type().await.map_err(to_file_error)?.is_dir()
+                || !entry.file_name().to_str().is_some_and(|name| Uuid::parse_str(name).is_ok())
+            {
+                continue;
+            }
+            let reserved_path = entry.path().join(&reserved_name);
+            match fs::rename(&reserved_path, entry.path().join(&committed_name)).await {
+                Ok(()) => {
+                    if effective_durability(volume).syncs_commit_metadata() {
+                        os::fsync_dir(&entry.path()).await.map_err(to_file_error)?;
+                    }
+                }
+                Err(err) if err.kind() == ErrorKind::NotFound => {}
+                Err(err) => return Err(to_file_error(err).into()),
+            }
+        }
+        Ok(())
+    }
+
+    async fn finish_version_delete(&self, volume: &str, object: &str, rollback_dir: Uuid) -> Result<bool> {
+        let object_path = self.get_object_path(volume, object)?;
+        let mut entries = match fs::read_dir(object_path).await {
+            Ok(entries) => entries,
+            Err(err) if err.kind() == ErrorKind::NotFound => return Ok(false),
+            Err(err) => return Err(to_file_error(err).into()),
+        };
+        let marker_name = format!("{DELETE_DATA_DIR_MARKER_PREFIX}{rollback_dir}");
+        let mut first_err = None;
+        let mut found = false;
+        while let Some(entry) = entries.next_entry().await.map_err(to_file_error)? {
+            let Some(data_dir) = entry.file_name().to_str().and_then(|data_dir| Uuid::parse_str(data_dir).ok()) else {
+                continue;
+            };
+            match fs::metadata(entry.path().join(&marker_name)).await {
+                Ok(metadata) if metadata.is_file() => found = true,
+                Ok(_) => continue,
+                Err(err) if err.kind() == ErrorKind::NotFound => continue,
+                Err(err) => return Err(to_file_error(err).into()),
+            }
+            if let Err(err) = self
+                .delete_data_dir(
+                    volume,
+                    &format!("{object}/{data_dir}"),
+                    DeleteOptions {
+                        recursive: true,
+                        ..Default::default()
+                    },
+                )
+                .await
+                && first_err.is_none()
+                && err != DiskError::FileNotFound
+                && err != DiskError::VolumeNotFound
+            {
+                first_err = Some(err);
+            }
+        }
+        first_err.map_or(Ok(found), Err)
+    }
+
+    async fn abort_reserved_version_delete(
+        &self,
+        object_dir: &Path,
+        rollback_dir: Uuid,
+        volume: &str,
+        object: &str,
+        stage: &'static str,
+        err: DiskError,
+    ) -> DiskError {
+        let xl_path = object_dir.join(STORAGE_FORMAT_FILE);
+        restore_delete_rollback_after_error(object_dir, &xl_path, Some(rollback_dir), volume, object, stage, err).await
+    }
+}
+
 #[async_trait::async_trait]
 impl DiskAPI for LocalDisk {
     fn to_string(&self) -> String {
@@ -6254,7 +6497,19 @@ impl DiskAPI for LocalDisk {
     #[tracing::instrument(level = "trace", skip_all)]
     async fn delete(&self, volume: &str, path: &str, opt: DeleteOptions) -> Result<()> {
         crate::hp_guard!("LocalDisk::delete");
-        self.delete_unleased(volume, path, &opt).await
+        let handled_version_delete = if opt.recursive
+            && opt.immediate
+            && let Some((object, transaction_id)) = path.rsplit_once('/')
+            && let Ok(transaction_id) = Uuid::parse_str(transaction_id)
+        {
+            self.finish_version_delete(volume, object, transaction_id).await?
+        } else {
+            false
+        };
+        match self.delete_unleased(volume, path, &opt).await {
+            Err(DiskError::FileNotFound) if handled_version_delete => Ok(()),
+            result => result,
+        }
     }
 
     #[tracing::instrument(level = "trace", skip_all)]
@@ -8252,6 +8507,7 @@ impl DiskAPI for LocalDisk {
 
         let mut meta = FileMeta::load(&buf)?;
         let old_dir = meta.delete_version(&fi)?;
+        let mut reserved_version_delete = false;
         if let Some(rollback_dir) = rollback_dir {
             write_metadata_rollback_backup(file_path.as_path(), rollback_dir, &buf).await?;
         }
@@ -8301,8 +8557,25 @@ impl DiskAPI for LocalDisk {
                     )
                     .await);
                 }
+                reserved_version_delete = match self.reserve_version_delete(volume, path, uuid, rollback_dir).await {
+                    Ok(reserved) => reserved,
+                    Err(err) => {
+                        return Err(restore_delete_rollback_after_error(
+                            file_path.as_path(),
+                            &xl_path,
+                            Some(rollback_dir),
+                            volume,
+                            path,
+                            "delete_version_reserve_data",
+                            err,
+                        )
+                        .await);
+                    }
+                };
                 let rollback_data_path = rollback_path.join(uuid.to_string());
-                if let Err(err) = rename_all_ignore_missing_source(&old_path, &rollback_data_path, &rollback_path).await {
+                if !reserved_version_delete
+                    && let Err(err) = rename_all_ignore_missing_source(&old_path, &rollback_data_path, &rollback_path).await
+                {
                     return Err(restore_delete_rollback_after_error(
                         file_path.as_path(),
                         &xl_path,
@@ -8315,6 +8588,18 @@ impl DiskAPI for LocalDisk {
                     .await);
                 }
                 if should_fail_after_delete_data_staged(path) {
+                    if reserved_version_delete {
+                        return Err(self
+                            .abort_reserved_version_delete(
+                                file_path.as_path(),
+                                rollback_dir,
+                                volume,
+                                path,
+                                "delete_version_test_after_stage",
+                                DiskError::Unexpected,
+                            )
+                            .await);
+                    }
                     return Err(restore_delete_rollback_after_error(
                         file_path.as_path(),
                         &xl_path,
@@ -8346,6 +8631,18 @@ impl DiskAPI for LocalDisk {
                 Ok(buf) => buf,
                 Err(err) => {
                     let err: DiskError = err.into();
+                    if reserved_version_delete && let Some(rollback_dir) = rollback_dir {
+                        return Err(self
+                            .abort_reserved_version_delete(
+                                file_path.as_path(),
+                                rollback_dir,
+                                volume,
+                                path,
+                                "delete_version_metadata_encode",
+                                err,
+                            )
+                            .await);
+                    }
                     return Err(restore_delete_rollback_after_error(
                         file_path.as_path(),
                         &xl_path,
@@ -8365,6 +8662,11 @@ impl DiskAPI for LocalDisk {
         };
 
         if let Err(err) = commit_result {
+            if reserved_version_delete && let Some(rollback_dir) = rollback_dir {
+                return Err(self
+                    .abort_reserved_version_delete(file_path.as_path(), rollback_dir, volume, path, "delete_version_commit", err)
+                    .await);
+            }
             return Err(restore_delete_rollback_after_error(
                 file_path.as_path(),
                 &xl_path,
@@ -8377,6 +8679,22 @@ impl DiskAPI for LocalDisk {
             .await);
         }
 
+        if reserved_version_delete
+            && let Some(rollback_dir) = rollback_dir
+            && let Err(err) = self.commit_reserved_version_delete(volume, path, rollback_dir).await
+        {
+            return Err(self
+                .abort_reserved_version_delete(
+                    file_path.as_path(),
+                    rollback_dir,
+                    volume,
+                    path,
+                    "delete_version_commit_intent",
+                    err,
+                )
+                .await);
+        }
+
         if should_fail_after_delete_commit(self.root.as_path(), path) {
             return Err(DiskError::Unexpected);
         }
@@ -11784,7 +12102,7 @@ mod test {
     }
 
     #[tokio::test]
-    async fn test_delete_version_rollback_restores_staged_data_dir() {
+    async fn test_delete_version_rollback_releases_reserved_data_dir() {
         use tempfile::tempdir;
 
         let dir = tempdir().expect("temp dir should be created");
@@ -11826,20 +12144,17 @@ mod test {
         .expect("delete should stage rollback state");
 
         assert!(!object_dir.join(STORAGE_FORMAT_FILE).exists());
-        assert!(!data_path.exists());
+        assert!(
+            data_path.exists(),
+            "the delete transaction must reserve the original data dir instead of moving it"
+        );
         assert!(
             object_dir
                 .join(rollback_dir.to_string())
                 .join(STORAGE_FORMAT_FILE_BACKUP)
                 .exists()
         );
-        assert!(
-            object_dir
-                .join(rollback_dir.to_string())
-                .join(data_dir.to_string())
-                .join("part.1")
-                .exists()
-        );
+        assert!(!object_dir.join(rollback_dir.to_string()).join(data_dir.to_string()).exists());
 
         disk.delete_version(
             bucket,
@@ -15016,6 +15331,284 @@ mod test {
         assert!(matches!(disk.read_all(volume, &first_part).await, Err(DiskError::FileNotFound)));
     }
 
+    #[tokio::test]
+    async fn delete_version_keeps_later_part_until_snapshot_release() {
+        use tempfile::tempdir;
+
+        let root_dir = tempdir().expect("temp dir should be created");
+        let endpoint = Endpoint::try_from(root_dir.path().to_string_lossy().as_ref()).expect("endpoint should parse");
+        let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created");
+        let volume = "snapshot-version-delete";
+        let object = "object";
+        let version_id = Uuid::new_v4();
+        let data_dir = Uuid::new_v4();
+        let rollback_dir = Uuid::new_v4();
+        let data_path = path_join_buf(&[object, &data_dir.to_string()]);
+        let first_part = path_join_buf(&[&data_path, "part.1"]);
+        let later_part = path_join_buf(&[&data_path, "part.2"]);
+        ensure_test_volume(&disk, volume).await;
+        disk.write_all(volume, &first_part, Bytes::from_static(b"first"))
+            .await
+            .expect("first shard should be written");
+        disk.write_all(volume, &later_part, Bytes::from_static(b"later"))
+            .await
+            .expect("later shard should be written");
+        let fi = test_file_info(object, version_id, Some(data_dir), None);
+        disk.write_all(volume, &path_join_buf(&[object, STORAGE_FORMAT_FILE]), test_meta(fi.clone()).into())
+            .await
+            .expect("metadata should be written");
+
+        let snapshot = disk
+            .acquire_snapshot_lease(volume, &data_path)
+            .await
+            .expect("snapshot lease should be acquired");
+        disk.delete_version(
+            volume,
+            object,
+            fi.clone(),
+            false,
+            DeleteOptions {
+                old_data_dir: Some(rollback_dir),
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("version delete should commit metadata");
+        disk.delete(
+            volume,
+            &format!("{object}/{rollback_dir}"),
+            DeleteOptions {
+                recursive: true,
+                immediate: true,
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("version delete should schedule physical cleanup");
+
+        assert_eq!(
+            disk.read_all(volume, &later_part)
+                .await
+                .expect("a later multipart shard must remain openable while leased"),
+            Bytes::from_static(b"later")
+        );
+        disk.release_snapshot_lease(volume, &data_path, snapshot)
+            .await
+            .expect("snapshot release should run deferred cleanup");
+        assert!(matches!(disk.read_all(volume, &first_part).await, Err(DiskError::FileNotFound)));
+    }
+
+    #[tokio::test]
+    async fn version_delete_cleanup_intent_survives_local_disk_restart() {
+        use tempfile::tempdir;
+
+        let root_dir = tempdir().expect("temp dir should be created");
+        let endpoint = Endpoint::try_from(root_dir.path().to_string_lossy().as_ref()).expect("endpoint should parse");
+        let volume = "snapshot-version-delete-restart";
+        let object = "object";
+        let data_dir = Uuid::new_v4();
+        let rollback_dir = Uuid::new_v4();
+        let data_path = path_join_buf(&[object, &data_dir.to_string()]);
+        let part = path_join_buf(&[&data_path, "part.1"]);
+        let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created");
+        ensure_test_volume(&disk, volume).await;
+        disk.write_all(volume, &part, Bytes::from_static(b"part"))
+            .await
+            .expect("shard should be written");
+        fs::create_dir_all(root_dir.path().join(volume).join(object).join(rollback_dir.to_string()))
+            .await
+            .expect("rollback directory should be created");
+        assert!(
+            disk.reserve_version_delete(volume, object, data_dir, rollback_dir)
+                .await
+                .expect("cleanup intent should be persisted")
+        );
+        disk.commit_reserved_version_delete(volume, object, rollback_dir)
+            .await
+            .expect("cleanup intent should be committed");
+        drop(disk);
+
+        let restarted = LocalDisk::new(&endpoint, false).await.expect("local disk should restart");
+        restarted
+            .delete(
+                volume,
+                &format!("{object}/{rollback_dir}"),
+                DeleteOptions {
+                    recursive: true,
+                    immediate: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("rollback cleanup should recover persisted intent");
+        assert!(matches!(restarted.read_all(volume, &part).await, Err(DiskError::FileNotFound)));
+    }
+
+    #[tokio::test]
+    async fn uuid_suffix_delete_does_not_run_version_cleanup_without_bound_marker() {
+        use tempfile::tempdir;
+
+        let root_dir = tempdir().expect("temp dir should be created");
+        let endpoint = Endpoint::try_from(root_dir.path().to_string_lossy().as_ref()).expect("endpoint should parse");
+        let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created");
+        let volume = "snapshot-non-transaction-delete";
+        let object = "object";
+        let requested_dir = Uuid::new_v4();
+        let victim_dir = Uuid::new_v4();
+        ensure_test_volume(&disk, volume).await;
+        disk.write_all(
+            volume,
+            &format!("{object}/{requested_dir}/{DELETE_DATA_DIR_MARKER_PREFIX}{victim_dir}"),
+            Bytes::new(),
+        )
+        .await
+        .expect("legacy-shaped marker should be written");
+        disk.write_all(volume, &format!("{object}/{victim_dir}/part.1"), Bytes::from_static(b"live"))
+            .await
+            .expect("victim shard should be written");
+
+        disk.delete(
+            volume,
+            &format!("{object}/{requested_dir}"),
+            DeleteOptions {
+                recursive: true,
+                immediate: true,
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("ordinary UUID directory delete should succeed");
+
+        assert_eq!(
+            disk.read_all(volume, &format!("{object}/{victim_dir}/part.1"))
+                .await
+                .expect("unbound sibling must not be deleted"),
+            Bytes::from_static(b"live")
+        );
+    }
+
+    #[tokio::test]
+    async fn version_delete_marker_is_durable_and_marker_errors_propagate() {
+        use tempfile::tempdir;
+
+        let _mode = durability_mode_override::set(DurabilityMode::Strict);
+        let root_dir = tempdir().expect("temp dir should be created");
+        let endpoint = Endpoint::try_from(root_dir.path().to_string_lossy().as_ref()).expect("endpoint should parse");
+        let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created");
+        let volume = "snapshot-marker-durability";
+        let object = "object";
+        let data_dir = Uuid::new_v4();
+        let rollback_dir = Uuid::new_v4();
+        ensure_test_volume(&disk, volume).await;
+        let data_path = disk
+            .get_object_path(volume, &format!("{object}/{data_dir}"))
+            .expect("data path should resolve");
+        fs::create_dir_all(&data_path).await.expect("data dir should be created");
+
+        assert!(
+            disk.reserve_version_delete(volume, object, data_dir, rollback_dir)
+                .await
+                .expect("reserved marker should be written")
+        );
+        assert!(
+            os::fsync_dir_recorder::was_fsynced(&data_path),
+            "strict durability must fsync the data directory after marker creation"
+        );
+        let committed_path = data_path.join(format!("{DELETE_DATA_DIR_MARKER_PREFIX}{rollback_dir}"));
+        fs::create_dir_all(&committed_path)
+            .await
+            .expect("conflicting committed marker directory should be created");
+        fs::write(committed_path.join("entry"), b"conflict")
+            .await
+            .expect("conflicting marker directory should be non-empty");
+        disk.commit_reserved_version_delete(volume, object, rollback_dir)
+            .await
+            .expect_err("marker rename failure must propagate");
+
+        let second_data_dir = Uuid::new_v4();
+        let second_rollback = Uuid::new_v4();
+        let second_path = disk
+            .get_object_path(volume, &format!("{object}/{second_data_dir}"))
+            .expect("second data path should resolve");
+        fs::create_dir_all(second_path.join(format!("{RESERVED_DELETE_DATA_DIR_MARKER_PREFIX}{second_rollback}")))
+            .await
+            .expect("reserved marker conflict directory should be created");
+        assert!(
+            disk.reserve_version_delete(volume, object, second_data_dir, second_rollback)
+                .await
+                .is_err(),
+            "marker creation failure must propagate"
+        );
+    }
+
+    #[tokio::test]
+    async fn deferred_version_delete_replays_after_restart_without_rollback_dir() {
+        use tempfile::tempdir;
+
+        let root_dir = tempdir().expect("temp dir should be created");
+        let endpoint = Endpoint::try_from(root_dir.path().to_string_lossy().as_ref()).expect("endpoint should parse");
+        let volume = "snapshot-deferred-delete-restart";
+        let object = "object";
+        let version_id = Uuid::new_v4();
+        let data_dir = Uuid::new_v4();
+        let rollback_dir = Uuid::new_v4();
+        let data_path = path_join_buf(&[object, &data_dir.to_string()]);
+        let part = path_join_buf(&[&data_path, "part.1"]);
+        let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created");
+        ensure_test_volume(&disk, volume).await;
+        disk.write_all(volume, &part, Bytes::from_static(b"part"))
+            .await
+            .expect("shard should be written");
+        let fi = test_file_info(object, version_id, Some(data_dir), None);
+        disk.write_all(volume, &path_join_buf(&[object, STORAGE_FORMAT_FILE]), test_meta(fi.clone()).into())
+            .await
+            .expect("metadata should be written");
+        let _lease = disk
+            .acquire_snapshot_lease(volume, &data_path)
+            .await
+            .expect("snapshot lease should be acquired");
+        disk.delete_version(
+            volume,
+            object,
+            fi,
+            false,
+            DeleteOptions {
+                old_data_dir: Some(rollback_dir),
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("version delete should commit");
+        disk.delete(
+            volume,
+            &format!("{object}/{rollback_dir}"),
+            DeleteOptions {
+                recursive: true,
+                immediate: true,
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("physical cleanup should be deferred");
+        assert!(disk.read_all(volume, &part).await.is_ok(), "leased data must remain");
+        drop(disk);
+
+        let restarted = LocalDisk::new(&endpoint, false).await.expect("local disk should restart");
+        restarted
+            .delete(
+                volume,
+                &format!("{object}/{rollback_dir}"),
+                DeleteOptions {
+                    recursive: true,
+                    immediate: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("committed marker should replay without rollback directory");
+        assert!(matches!(restarted.read_all(volume, &part).await, Err(DiskError::FileNotFound)));
+    }
+
     #[tokio::test]
     async fn data_dir_cleanup_without_a_lease_keeps_existing_behavior() {
         use tempfile::tempdir;

crates/ecstore/src/object_api/encryption.rs

@@ -1,80 +0,0 @@
-// Copyright 2024 RustFS Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use async_trait::async_trait;
-use http::{HeaderMap, HeaderValue};
-use std::collections::HashMap;
-use std::error::Error;
-use std::fmt::{Display, Formatter};
-
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub enum ReadEncryptionMode {
-    Direct { base_nonce: [u8; 12] },
-    Object,
-}
-
-pub struct ReadEncryptionMaterial {
-    pub key_bytes: [u8; 32],
-    pub mode: ReadEncryptionMode,
-}
-
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub enum EncryptionResolutionErrorKind {
-    InvalidRequest,
-    InvalidMetadata,
-    ServiceUnavailable,
-    DecryptionFailed,
-}
-
-#[derive(Debug)]
-pub struct EncryptionResolutionError {
-    kind: EncryptionResolutionErrorKind,
-    message: String,
-}
-
-impl EncryptionResolutionError {
-    pub fn new(kind: EncryptionResolutionErrorKind, message: impl Into<String>) -> Self {
-        Self {
-            kind,
-            message: message.into(),
-        }
-    }
-
-    pub fn kind(&self) -> EncryptionResolutionErrorKind {
-        self.kind
-    }
-}
-
-impl Display for EncryptionResolutionError {
-    fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
-        formatter.write_str(&self.message)
-    }
-}
-
-impl Error for EncryptionResolutionError {}
-
-pub struct ReadEncryptionRequest<'a> {
-    pub bucket: &'a str,
-    pub object: &'a str,
-    pub metadata: &'a HashMap<String, String>,
-    pub headers: &'a HeaderMap<HeaderValue>,
-}
-
-#[async_trait]
-pub trait ObjectEncryptionResolver: Send + Sync {
-    async fn resolve_read_material(
-        &self,
-        request: ReadEncryptionRequest<'_>,
-    ) -> Result<Option<ReadEncryptionMaterial>, EncryptionResolutionError>;
-}

crates/ecstore/src/object_api/mod.rs

@@ -84,7 +84,6 @@ pub(crate) fn legacy_encrypted_range_seek_enabled() -> bool {
 }
 
 mod body_cache_hook;
-mod encryption;
 mod hook_slot;
 mod object_mutation_hook;
 mod readers;
@@ -99,10 +98,6 @@ pub use body_cache_hook::{
 pub(crate) use body_cache_hook::{
     get_object_body_cache_hook, get_object_body_cache_hook_suppressed, without_get_object_body_cache_hook,
 };
-pub use encryption::{
-    EncryptionResolutionError, EncryptionResolutionErrorKind, ObjectEncryptionResolver, ReadEncryptionMaterial,
-    ReadEncryptionMode, ReadEncryptionRequest,
-};
 pub(crate) use object_mutation_hook::notify_object_mutation;
 pub use object_mutation_hook::{ObjectMutationHook, register_object_mutation_hook, unregister_object_mutation_hook};
 pub use readers::*;

crates/ecstore/src/object_api/types.rs

@@ -273,9 +273,29 @@ impl ObjectInfo {
     }
 
     pub fn is_encrypted(&self) -> bool {
-        self.user_defined
-            .keys()
-            .any(|key| rustfs_utils::http::is_object_encryption_marker(key))
+        // Corresponding to the logic in rustfs/src/sse.rs/encryption_material_to_metadata function
+        use rustfs_utils::http::{SSEC_ALGORITHM_HEADER, SSEC_KEY_HEADER, SSEC_KEY_MD5_HEADER};
+
+        self.user_defined.keys().any(|key| {
+            let lower = key.to_ascii_lowercase();
+            lower.starts_with("x-minio-encryption-")
+                || lower.starts_with("x-minio-internal-server-side-encryption-")
+                || matches!(
+                    lower.as_str(),
+                    "x-minio-internal-encrypted-multipart"
+                        | "x-rustfs-encryption-key"
+                        | "x-rustfs-encryption-algorithm"
+                        | "x-rustfs-encryption-iv"
+                        | "x-rustfs-encryption-key-id"
+                        | "x-rustfs-encryption-context"
+                        | "x-rustfs-encryption-tag"
+                        | "x-amz-server-side-encryption-aws-kms-key-id"
+                        | SSEC_ALGORITHM_HEADER
+                        | SSEC_KEY_HEADER
+                        | SSEC_KEY_MD5_HEADER
+                        | "x-amz-server-side-encryption"
+                )
+        })
     }
 
     /// Maximum inline size for non-versioned objects (128 KiB).
@@ -319,7 +339,26 @@ impl ObjectInfo {
     }
 
     pub fn encryption_original_size(&self) -> std::io::Result<Option<i64>> {
-        rustfs_utils::http::get_object_encryption_original_size(&self.user_defined)
+        let actual_size = rustfs_utils::http::get_str(&self.user_defined, rustfs_utils::http::SUFFIX_ACTUAL_SIZE);
+        if let Some(size_str) = self
+            .user_defined
+            .get("x-rustfs-encryption-original-size")
+            .map(String::as_str)
+            .or_else(|| {
+                self.user_defined
+                    .get("x-amz-server-side-encryption-customer-original-size")
+                    .map(String::as_str)
+            })
+            .or(actual_size.as_deref())
+            && !size_str.is_empty()
+        {
+            let size = size_str
+                .parse::<i64>()
+                .map_err(|e| std::io::Error::other(format!("Failed to parse encryption original size: {e}")))?;
+            return Ok(Some(size));
+        }
+
+        Ok(None)
     }
 
     pub fn decrypted_size(&self) -> std::io::Result<i64> {
@@ -349,6 +388,9 @@ impl ObjectInfo {
             return Ok(actual_size);
         }
 
+        // Check if object is encrypted
+        // Managed SSE stores original size in x-rustfs-encryption-original-size metadata
+        // SSE-C stores original size in x-amz-server-side-encryption-customer-original-size
         if let Some(size) = self.encryption_original_size()? {
             return Ok(size);
         }
@@ -839,19 +881,6 @@ mod tests {
         assert!(!object.is_inline_fast_path_eligible(), "transitioned objects must fall back");
     }
 
-    #[test]
-    fn minio_internal_encryption_metadata_is_not_treated_as_plaintext() {
-        let object = ObjectInfo {
-            user_defined: Arc::new(HashMap::from([(
-                "X-Minio-Internal-Server-Side-Encryption-Sealed-Key".to_string(),
-                "sealed".to_string(),
-            )])),
-            ..Default::default()
-        };
-
-        assert!(object.is_encrypted());
-    }
-
     #[test]
     fn versions_after_marker_handles_null_version_marker() {
         let first_version = Uuid::parse_str("11111111-2222-3333-4444-555555555555").unwrap();

crates/ecstore/src/runtime/instance.rs

@@ -46,7 +46,6 @@ use crate::bucket::metadata_sys::BucketMetadataSys;
 use crate::bucket::replication::{DynReplicationPool, ReplicationStats};
 use crate::disk::DiskStore;
 use crate::layout::endpoints::{EndpointServerPools, SetupType};
-use crate::object_api::ObjectEncryptionResolver;
 use crate::services::event_notification::EventNotifier;
 use crate::services::tier::tier::TierConfigMgr;
 use rustfs_lock::{GlobalLockManager, get_global_lock_manager};
@@ -160,8 +159,6 @@ pub struct InstanceContext {
     /// workers (scanner/heal/tier/lifecycle) without touching another instance.
     /// Replaces the process-global cancel-token static.
     background_cancel_token: OnceLock<CancellationToken>,
-    /// Resolves object-encryption material at the application boundary.
-    object_encryption_resolver: OnceLock<Arc<dyn ObjectEncryptionResolver>>,
     tier_delete_journal_recovery_stores: std::sync::Mutex<HashSet<Uuid>>,
     transition_transaction_recovery_stores: std::sync::Mutex<HashSet<Uuid>>,
     #[cfg(test)]
@@ -200,7 +197,6 @@ impl InstanceContext {
             local_disk_set_drives: Arc::new(RwLock::new(Vec::new())),
             bucket_metadata_sys: std::sync::Mutex::new(None),
             background_cancel_token: OnceLock::new(),
-            object_encryption_resolver: OnceLock::new(),
             tier_delete_journal_recovery_stores: std::sync::Mutex::new(HashSet::new()),
             transition_transaction_recovery_stores: std::sync::Mutex::new(HashSet::new()),
             #[cfg(test)]
@@ -213,19 +209,6 @@ impl InstanceContext {
         self.lock_manager.clone()
     }
 
-    /// Install the application-owned object-encryption resolver once.
-    pub fn set_object_encryption_resolver(
-        &self,
-        resolver: Arc<dyn ObjectEncryptionResolver>,
-    ) -> Result<(), Arc<dyn ObjectEncryptionResolver>> {
-        self.object_encryption_resolver.set(resolver)
-    }
-
-    /// Return the configured object-encryption resolver, if startup installed one.
-    pub fn object_encryption_resolver(&self) -> Option<&dyn ObjectEncryptionResolver> {
-        self.object_encryption_resolver.get().map(Arc::as_ref)
-    }
-
     /// Set this instance's S3 region.
     ///
     /// Write-once: panics on a second write, preserving the startup fail-fast

crates/ecstore/src/runtime/sources.rs

@@ -46,6 +46,7 @@ use crate::{
 use rustfs_concurrency::WorkloadAdmissionSnapshotProvider;
 use rustfs_config::server_config::{Config, get_global_server_config, set_global_server_config};
 use rustfs_io_metrics::internode_metrics::global_internode_metrics;
+use rustfs_kms::{ObjectEncryptionService, get_global_encryption_service};
 use rustfs_lock::client::LockClient;
 use s3s::dto::BucketLifecycleConfiguration;
 use s3s::region::Region;
@@ -104,6 +105,10 @@ pub(crate) fn record_erasure_write_quorum_failure(stage: &'static str, dominant_
     global_internode_metrics().record_erasure_write_quorum_failure(stage, dominant_error);
 }
 
+pub(crate) async fn object_encryption_service() -> Option<Arc<ObjectEncryptionService>> {
+    get_global_encryption_service().await
+}
+
 pub fn object_store_handle() -> Option<Arc<ECStore>> {
     resolve_object_store_handle()
 }
@@ -202,10 +207,6 @@ pub(crate) async fn ensure_boot_time() {
     GLOBAL_BOOT_TIME.get_or_init(|| async { SystemTime::now() }).await;
 }
 
-pub(crate) async fn scanner_init_time() -> Option<chrono::DateTime<chrono::Utc>> {
-    rustfs_common::get_global_init_time().await
-}
-
 pub(crate) async fn root_disk_threshold_for_erasure_disk() -> Option<u64> {
     if is_erasure_sd().await {
         None

crates/ecstore/src/services/metrics_realtime.rs

@@ -71,6 +71,7 @@ fn to_madmin_scanner_metrics(metrics: rustfs_common::metrics::ScannerMetricsRepo
     MadminScannerMetrics {
         collected_at: metrics.collected_at,
         current_cycle: metrics.current_cycle,
+        current_cycle_active: Some(metrics.current_cycle_active),
         current_started: metrics.current_started,
         cycles_completed_at: metrics.cycles_completed_at,
         ongoing_buckets: metrics.ongoing_buckets,
@@ -398,10 +399,7 @@ pub async fn collect_local_metrics(types: MetricType, opts: &CollectMetricsOpts)
 
     if types.contains(&MetricType::SCANNER) {
         debug!("start get scanner metrics");
-        let mut metrics = global_metrics().report().await;
-        if let Some(init_time) = runtime_sources::scanner_init_time().await {
-            metrics.current_started = init_time;
-        }
+        let metrics = global_metrics().report().await;
         real_time_metrics.aggregated.scanner = Some(to_madmin_scanner_metrics(metrics));
     }
 
@@ -540,7 +538,9 @@ async fn collect_local_disks_metrics(disks: &HashSet<String>) -> HashMap<String,
 #[cfg(test)]
 mod test {
     use super::*;
+    use rustfs_common::metrics::CurrentCycle;
     use rustfs_io_metrics::internode_metrics::global_internode_metrics;
+    use serial_test::serial;
     use std::time::Duration;
 
     #[test]
@@ -588,7 +588,10 @@ mod test {
 
     #[test]
     fn scanner_metrics_mapping_preserves_partial_source_status() {
+        let current_started = Utc::now() - chrono::Duration::seconds(5);
         let scanner = to_madmin_scanner_metrics(rustfs_common::metrics::ScannerMetricsReport {
+            current_cycle_active: true,
+            current_started,
             last_cycle_partial_source: "usage".to_string(),
             last_cycle_partial_source_code: 1,
             partial_cycles_by_source: vec![rustfs_common::metrics::ScannerSourceCycleSnapshot {
@@ -598,6 +601,8 @@ mod test {
             ..Default::default()
         });
 
+        assert_eq!(scanner.current_cycle_active, Some(true));
+        assert_eq!(scanner.current_started, current_started);
         assert_eq!(scanner.last_cycle_partial_source, "usage");
         assert_eq!(scanner.last_cycle_partial_source_code, 1);
         let usage = scanner
@@ -608,6 +613,39 @@ mod test {
         assert_eq!(usage.cycles, 2);
     }
 
+    #[tokio::test]
+    #[serial]
+    async fn collect_local_metrics_preserves_scanner_cycle_started_time() {
+        let previous_init_time = *rustfs_common::globals::GLOBAL_INIT_TIME.read().await;
+        let previous_cycle = global_metrics().get_cycle().await;
+        let init_time = Utc::now() - chrono::Duration::hours(1);
+        let cycle_started = Utc::now() - chrono::Duration::seconds(5);
+        *rustfs_common::globals::GLOBAL_INIT_TIME.write().await = Some(init_time);
+        let cycle = CurrentCycle {
+            current: 0,
+            next: 1,
+            started: cycle_started,
+            ..Default::default()
+        };
+        let cycle_start = global_metrics().start_scan_cycle_work_with_cycle(cycle).await;
+
+        let realtime = collect_local_metrics(MetricType::SCANNER, &CollectMetricsOpts::default()).await;
+
+        global_metrics()
+            .finish_scan_cycle_work_with_cycle(cycle_start, previous_cycle.clone().unwrap_or_default())
+            .await;
+        global_metrics().set_cycle(previous_cycle).await;
+        *rustfs_common::globals::GLOBAL_INIT_TIME.write().await = previous_init_time;
+
+        let encoded = rmp_serde::to_vec_named(&realtime).expect("realtime metrics should encode");
+        let decoded: RealtimeMetrics = rmp_serde::from_slice(&encoded).expect("realtime metrics should decode");
+        let mut aggregated = RealtimeMetrics::default();
+        aggregated.merge(decoded);
+        let scanner = aggregated.aggregated.scanner.expect("scanner metrics");
+        assert_eq!(scanner.current_cycle_active, Some(true));
+        assert_eq!(scanner.current_started, cycle_started);
+    }
+
     #[test]
     fn scanner_metrics_mapping_preserves_pacing_pressure() {
         let scanner = to_madmin_scanner_metrics(rustfs_common::metrics::ScannerMetricsReport {

crates/ecstore/src/services/tier/warm_backend_gcs.rs

@@ -19,6 +19,7 @@
 #![allow(clippy::all)]
 
 use std::collections::HashMap;
+use std::io::{Error, ErrorKind};
 use std::sync::Arc;
 
 use bytes::Bytes;
@@ -45,6 +46,19 @@ const MAX_PARTS_COUNT: i64 = 10000;
 const _MAX_PART_SIZE: i64 = 1024 * 1024 * 1024 * 5;
 const MIN_PART_SIZE: i64 = 1024 * 1024 * 128;
 
+fn parse_generation(remote_version: &str) -> Result<Option<i64>, Error> {
+    if remote_version.is_empty() {
+        return Ok(None);
+    }
+    let generation = remote_version
+        .parse::<i64>()
+        .map_err(|_| Error::new(ErrorKind::InvalidData, "GCS remote version is not a valid generation"))?;
+    if generation <= 0 {
+        return Err(Error::new(ErrorKind::InvalidData, "GCS remote version generation must be positive"));
+    }
+    Ok(Some(generation))
+}
+
 pub struct WarmBackendGCS {
     pub client: Arc<Storage>,
     pub control: Arc<StorageControl>,
@@ -105,6 +119,10 @@ impl WarmBackendGCS {
 
 #[async_trait::async_trait]
 impl WarmBackend for WarmBackendGCS {
+    fn validate_remote_version_id(&self, remote_version_id: &str) -> Result<(), std::io::Error> {
+        parse_generation(remote_version_id).map(|_| ())
+    }
+
     async fn put_with_meta(
         &self,
         object: &str,
@@ -135,6 +153,9 @@ impl WarmBackend for WarmBackendGCS {
 
     async fn get(&self, object: &str, rv: &str, opts: WarmBackendGetOpts) -> Result<ReadCloser, std::io::Error> {
         let mut req = self.client.read_object(&self.bucket, &self.get_dest(object));
+        if let Some(generation) = parse_generation(rv)? {
+            req = req.set_generation(generation);
+        }
 
         // Honor the requested byte range so Range GETs on tiered objects return the exact
         // interval instead of the whole object (matches the s3/s3sdk/rustfs warm backends).
@@ -164,13 +185,15 @@ impl WarmBackend for WarmBackendGCS {
         // gRPC v2 DeleteObject requires the bucket in resource-name form. Without this the
         // deleted tiered object was never removed from GCS (empty impl returned Ok), leaking
         // remote data forever.
-        self.control
+        let mut req = self
+            .control
             .delete_object()
             .set_bucket(format!("projects/_/buckets/{}", self.bucket))
-            .set_object(self.get_dest(object))
-            .send()
-            .await
-            .map_err(|e| std::io::Error::other(e.to_string()))?;
+            .set_object(self.get_dest(object));
+        if let Some(generation) = parse_generation(rv)? {
+            req = req.set_generation(generation);
+        }
+        req.send().await.map_err(|e| std::io::Error::other(e.to_string()))?;
         Ok(())
     }
 
@@ -191,6 +214,30 @@ impl WarmBackend for WarmBackendGCS {
     }
 }
 
+#[cfg(test)]
+mod tests {
+    use super::parse_generation;
+    use std::io::ErrorKind;
+
+    #[test]
+    fn generation_parser_preserves_exact_numeric_versions() {
+        assert_eq!(parse_generation("").expect("empty generation means no version condition"), None);
+        assert_eq!(parse_generation("1").expect("minimum generation should parse"), Some(1));
+        assert_eq!(
+            parse_generation(&i64::MAX.to_string()).expect("maximum generation should parse"),
+            Some(i64::MAX)
+        );
+    }
+
+    #[test]
+    fn generation_parser_rejects_unknown_or_non_positive_versions() {
+        for value in ["unknown", "1.0", "-1", "0", "9223372036854775808"] {
+            let err = parse_generation(value).expect_err("unknown generation must fail closed");
+            assert_eq!(err.kind(), ErrorKind::InvalidData, "{value}");
+        }
+    }
+}
+
 /*fn gcs_to_object_error(err: Error, params: Vec<String>) -> Option<Error> {
   if err == nil {
     return nil

crates/ecstore/src/set_disk/core/io_primitives.rs

@@ -46,6 +46,7 @@ use crate::diagnostics::get::{
     GetObjectFailureReason, classify_disk_error, get_stage_timer_if_enabled, record_get_object_pipeline_failure,
     record_get_object_pipeline_failure_for_path, record_get_stage_duration_if_enabled,
 };
+use crate::disk::local::DELETE_DATA_DIR_MARKER_PREFIX;
 use crate::disk::{
     DataDirDeleteStatus, OldCurrentSize, PART_TRANSACTION_NEW_META, PART_TRANSACTION_OLD_META, PART_TRANSACTION_ROLLBACK,
     PartTransactionAction, part_transaction_path,
@@ -3952,9 +3953,9 @@ impl SetDisks {
     /// * The set of referenced data dirs is the UNION of `get_data_dirs()` across
     ///   every online disk's `xl.meta`, so a dir named by *any* replica is kept.
     /// * If a disk holds the object directory but its `xl.meta` is missing or
-    ///   unparsable, the object is treated as degraded and NOTHING is removed —
-    ///   the unreadable copy could be the only one naming a live data dir, and a
-    ///   heal must run first.
+    ///   unparsable, the object is treated as degraded and unmarked data dirs are
+    ///   never removed. A data dir carrying a committed delete-transaction marker
+    ///   remains reclaimable after a downgrade/re-upgrade cleanup interruption.
     /// * Only subdirectories whose names parse as a UUID are ever considered;
     ///   removal is non-recursive-safe via a recursive delete of the full stray
     ///   data-dir path only.
@@ -3969,7 +3970,7 @@ impl SetDisks {
         // physical UUID subdirectories present on each disk. Abort on any degraded
         // copy so a healable object is never stripped of a referenced data dir.
         let mut referenced: HashSet<Uuid> = HashSet::new();
-        let mut per_disk_dirs: Vec<(usize, Vec<Uuid>)> = Vec::new();
+        let mut per_disk_dirs: Vec<(usize, Vec<(Uuid, bool)>)> = Vec::new();
         let mut healthy_metas = 0usize;
 
         for (i, disk) in disks.iter().enumerate() {
@@ -4005,6 +4006,22 @@ impl SetDisks {
                         // to the orphan-dir / dangling-object heal paths.
                         continue;
                     }
+                    let mut committed = Vec::with_capacity(physical.len());
+                    for dir in physical {
+                        let data_dir = format!("{object}/{dir}");
+                        let committed_delete = disk.list_dir("", bucket, &data_dir, 0).await.is_ok_and(|entries| {
+                            entries.iter().any(|entry| {
+                                entry
+                                    .strip_prefix(DELETE_DATA_DIR_MARKER_PREFIX)
+                                    .is_some_and(|transaction| Uuid::parse_str(transaction).is_ok())
+                            })
+                        });
+                        committed.push((dir, committed_delete));
+                    }
+                    if committed.iter().all(|(_, committed_delete)| *committed_delete) {
+                        per_disk_dirs.push((i, committed));
+                        continue;
+                    }
                     warn!(
                         target: "rustfs_ecstore::set_disk",
                         bucket, object,
@@ -4041,22 +4058,16 @@ impl SetDisks {
 
             healthy_metas += 1;
             if !physical.is_empty() {
-                per_disk_dirs.push((i, physical));
+                per_disk_dirs.push((i, physical.into_iter().map(|dir| (dir, false)).collect()));
             }
         }
 
-        // No healthy metadata anywhere: this is not a live object, so surplus dirs
-        // (if any) belong to the dangling-object heal path, not here.
-        if healthy_metas == 0 {
-            return Ok(0);
-        }
-
         // Phase 2: delete every physical data dir not referenced by the union.
         let mut removed = 0usize;
         for (i, physical) in per_disk_dirs {
             let Some(disk) = disks[i].as_ref() else { continue };
-            for dir in physical {
-                if referenced.contains(&dir) {
+            for (dir, committed_delete) in physical {
+                if referenced.contains(&dir) || (healthy_metas == 0 && !committed_delete) {
                     continue;
                 }
                 let stray = format!("{object}/{dir}");

crates/ecstore/src/set_disk/metadata.rs

@@ -505,7 +505,9 @@ impl SetDisks {
     }
 
     fn file_info_has_encryption_metadata(meta: &FileInfo) -> bool {
-        meta.metadata.keys().any(|name| http::is_object_encryption_marker(name))
+        meta.metadata
+            .keys()
+            .any(|name| http::is_encryption_metadata_key(name) || http::is_sse_header(name))
     }
 
     fn starts_with_ignore_ascii_case(value: &str, prefix: &str) -> bool {

crates/ecstore/src/set_disk/ops/object.rs

@@ -42,7 +42,6 @@ use crate::object_api::{GetObjectBodySource, get_object_body_cache_hook_suppress
 use crate::services::tier::tier::{TierConfigMgr, TierOperationLease};
 use crate::store::ECStore;
 use futures::FutureExt as _;
-use http::HeaderValue;
 use std::future::Future;
 
 fn erasure_from_file_info(fi: &FileInfo, uses_legacy: bool) -> Result<coding::Erasure> {
@@ -50,17 +49,6 @@ fn erasure_from_file_info(fi: &FileInfo, uses_legacy: bool) -> Result<coding::Er
         .map_err(Error::from)
 }
 
-async fn get_object_reader_with_context(
-    ctx: &InstanceContext,
-    reader: Box<dyn AsyncRead + Unpin + Send + Sync>,
-    range: Option<HTTPRangeSpec>,
-    object_info: &ObjectInfo,
-    opts: &ObjectOptions,
-    headers: &HeaderMap<HeaderValue>,
-) -> Result<(GetObjectReader, usize, i64)> {
-    GetObjectReader::new_with_resolver(reader, range, object_info, opts, headers, ctx.object_encryption_resolver()).await
-}
-
 /// Length of the full plaintext body when — and only when — this read's output
 /// is exactly the object's complete plaintext, so the app-layer body cache may
 /// serve it in place of the erasure read.
@@ -562,13 +550,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
                 &self.ctx.tier_config_mgr(),
             )
             .await?;
-            return Ok(finish_set_disk_read_lock(
-                gr,
-                read_lock_guard.take(),
-                lock_optimization_enabled,
-                bucket,
-                object,
-            ));
+            return Ok(finish_set_disk_read_lock(gr, read_lock_guard.take(), None, bucket, object));
         }
 
         // App-layer object data cache probe: metadata (etag/size) is resolved
@@ -695,6 +677,18 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
             return Ok(reader);
         }
 
+        let snapshot_lease = if lock_optimization_enabled {
+            match fi.data_dir.filter(|data_dir| !data_dir.is_nil()) {
+                Some(data_dir) => {
+                    let data_dir_path = format!("{object}/{data_dir}");
+                    acquire_snapshot_leases(&disks, bucket, &data_dir_path, fi.erasure.data_blocks).await
+                }
+                None => None,
+            }
+        } else {
+            None
+        };
+
         match codec_streaming_gate.decision {
             GetCodecStreamingDecision::Use => {
                 match Self::get_object_decode_reader_with_fileinfo(
@@ -719,18 +713,11 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
                             size_bucket,
                         );
                         record_get_object_reader_path_observation(GET_OBJECT_PATH_CODEC_STREAMING, object_class, size_bucket);
-                        let (mut reader, _offset, _length) =
-                            get_object_reader_with_context(&self.ctx, stream, range, &object_info, opts, &h).await?;
+                        let (mut reader, _offset, _length) = GetObjectReader::new(stream, range, &object_info, opts, &h).await?;
                         // Carry the hook probe result so the app layer skips its
                         // now-redundant lookup on the streaming miss path (ODC-16).
                         reader.body_source = body_source;
-                        return Ok(finish_set_disk_read_lock(
-                            reader,
-                            read_lock_guard.take(),
-                            lock_optimization_enabled,
-                            bucket,
-                            object,
-                        ));
+                        return Ok(finish_set_disk_read_lock(reader, read_lock_guard.take(), snapshot_lease, bucket, object));
                     }
                     core::io_primitives::GetCodecStreamingReaderBuildOutcome::Fallback(reason) => {
                         record_get_codec_streaming_gate_decision(
@@ -758,8 +745,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
         let (rd, wd) = tokio::io::duplex(duplex_buffer_size);
         debug!(bucket, object, duplex_buffer_size, "Created duplex pipe for object data transfer");
 
-        let (mut reader, offset, length) =
-            get_object_reader_with_context(&self.ctx, Box::new(rd), range, &object_info, opts, &h).await?;
+        let (mut reader, offset, length) = GetObjectReader::new(Box::new(rd), range, &object_info, opts, &h).await?;
         // Carry the hook probe result so the app layer skips its now-redundant
         // lookup on the streaming miss path (ODC-16).
         reader.body_source = body_source;
@@ -770,15 +756,22 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
         let set_index = self.set_index;
         let pool_index = self.pool_index;
         let skip_verify = opts.skip_verify_bitrot;
-        if lock_optimization_enabled {
+        let producer_snapshot_lease = snapshot_lease.clone();
+        if let Some(lease) = snapshot_lease {
             release_materialized_read_lock(&bucket, &object, read_lock_guard.take());
-            debug!(bucket, object, "Lock optimization: released read lock before streaming read");
+            reader.stream = Box::new(SnapshotLeaseReader {
+                inner: reader.stream,
+                lease: Some(lease),
+                terminal_error: false,
+            });
+            debug!(bucket, object, "Lock optimization: replaced read lock with snapshot leases");
         }
 
-        // When lock optimization is disabled, keep the read-lock guard in the
-        // task so it lives for the duration of the streaming read.
+        // The producer shares the lease lifetime with the body so cancellation
+        // cannot release the snapshot while the duplex task is still unwinding.
         tokio::spawn(async move {
             let _guard = read_lock_guard;
+            let _snapshot_lease = producer_snapshot_lease;
             let mut writer = GetObjectDownstreamWriter::new(wd);
             // Do not wrap the entire read+write pipeline in `disk_read_timeout`.
             // `get_object_with_fileinfo` also waits on `writer`, so an outer timeout
@@ -4468,61 +4461,6 @@ mod erasure_construction_tests {
     }
 }
 
-#[cfg(test)]
-mod object_encryption_resolver_wiring_tests {
-    use super::*;
-    use crate::object_api::{EncryptionResolutionError, ObjectEncryptionResolver, ReadEncryptionMaterial, ReadEncryptionRequest};
-    use std::io::Cursor;
-    use std::sync::atomic::{AtomicUsize, Ordering};
-
-    struct CountingResolver {
-        calls: AtomicUsize,
-    }
-
-    #[async_trait::async_trait]
-    impl ObjectEncryptionResolver for CountingResolver {
-        async fn resolve_read_material(
-            &self,
-            _request: ReadEncryptionRequest<'_>,
-        ) -> std::result::Result<Option<ReadEncryptionMaterial>, EncryptionResolutionError> {
-            self.calls.fetch_add(1, Ordering::Relaxed);
-            Ok(None)
-        }
-    }
-
-    #[tokio::test]
-    async fn get_object_reader_forwards_instance_resolver() {
-        let resolver = Arc::new(CountingResolver {
-            calls: AtomicUsize::new(0),
-        });
-        let ctx = InstanceContext::new();
-        assert!(
-            ctx.set_object_encryption_resolver(resolver.clone()).is_ok(),
-            "fresh context should accept resolver"
-        );
-        let object_info = ObjectInfo {
-            bucket: "bucket".to_string(),
-            name: "object".to_string(),
-            size: 1,
-            user_defined: Arc::new(HashMap::from([("x-amz-server-side-encryption".to_string(), "AES256".to_string())])),
-            ..Default::default()
-        };
-
-        let result = get_object_reader_with_context(
-            &ctx,
-            Box::new(Cursor::new(Vec::<u8>::new())),
-            None,
-            &object_info,
-            &ObjectOptions::default(),
-            &HeaderMap::new(),
-        )
-        .await;
-
-        assert!(result.is_err(), "resolver returning no material must fail closed");
-        assert_eq!(resolver.calls.load(Ordering::Relaxed), 1);
-    }
-}
-
 #[cfg(test)]
 pub(in crate::set_disk::ops) mod hermetic_set_disks_support {
     //! Shared hermetic `SetDisks` construction for the ops tests below: the

crates/ecstore/tests/README.md

@@ -2,7 +2,7 @@
 
 ## MinIO-generated encrypted fixtures
 
-`rustfs/src/storage/minio_generated_read_test.rs` validates the `bitrot -> GetObjectReader` path against raw MinIO backend data captured by
+`minio_generated_read_test.rs` validates the `bitrot -> GetObjectReader` path against raw MinIO backend data captured by
 `.\rustfs\scripts\minio_fixture_lab\lab.py`.
 
 It currently covers multipart fixtures for:
@@ -20,5 +20,5 @@ Example:
 ```powershell
 $env:RUSTFS_MINIO_FIXTURE_ROOT = '.\rustfs\tmp\minio-fixture-lab-local-key'
 $env:RUSTFS_MINIO_STATIC_KMS_KEY_B64 = '<base64-32-byte-local-minio-kms-key>'
-cargo +1.97.1 test -p rustfs --features rio-v2 storage::minio_generated_read_test --lib -- --ignored
+cargo +1.97.1 test -p rustfs-ecstore --features rio-v2 --test minio_generated_read_test -- --ignored
 ```

crates/ecstore/tests/minio_generated_read_test.rs

@@ -4,13 +4,14 @@ use std::fs;
 use std::io::Cursor;
 use std::path::{Path, PathBuf};
 
-use super::sse::SseObjectEncryptionResolver;
-use super::storage_api::ecstore_test_support::{
-    DiskAPI as _, DiskOption, Endpoint, Erasure, GetObjectReader, ObjectInfo, ObjectOptions, create_bitrot_reader, new_disk,
-};
+mod storage_api;
+
 use rustfs_filemeta::{FileInfo, FileInfoOpts, get_file_info};
 use serde::Deserialize;
 use sha2::{Digest, Sha256};
+use storage_api::minio_generated_read::{
+    DiskAPI as _, DiskOption, Endpoint, Erasure, GetObjectReader, ObjectInfo, ObjectOptions, create_bitrot_reader, new_disk,
+};
 use temp_env::async_with_vars;
 use tokio::io::{AsyncReadExt, AsyncWrite};
 
@@ -48,7 +49,7 @@ impl AsyncWrite for VecAsyncWriter {
 fn fixture_root() -> PathBuf {
     std::env::var_os("RUSTFS_MINIO_FIXTURE_ROOT")
         .map(PathBuf::from)
-        .unwrap_or_else(|| PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../crates/rio-v2/tests/fixtures/minio-generated"))
+        .unwrap_or_else(|| PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../rio-v2/tests/fixtures/minio-generated"))
 }
 
 fn case_dir(case_id: &str) -> PathBuf {
@@ -136,14 +137,12 @@ async fn read_fixture_plaintext(encrypted: Vec<u8>, object_info: ObjectInfo, kms
             ("RUSTFS_SSE_S3_MASTER_KEY", None::<String>),
         ],
         async move {
-            let resolver = SseObjectEncryptionResolver;
-            let (mut reader, offset, length) = GetObjectReader::new_with_resolver(
+            let (mut reader, offset, length) = GetObjectReader::new(
                 Box::new(Cursor::new(encrypted)),
                 None,
                 &object_info,
                 &ObjectOptions::default(),
                 &http::HeaderMap::new(),
-                Some(&resolver),
             )
             .await
             .map_err(|err| format!("construct GetObjectReader from MinIO raw fixture: {err:?}"))?;
@@ -220,12 +219,11 @@ async fn encrypted_fixture_bytes(case_dir: &Path, manifest: &ManifestRecord, fil
             readers.push(reader);
         }
 
-        let erasure = Erasure::try_new(
+        let erasure = Erasure::new(
             file_info.erasure.data_blocks,
             file_info.erasure.parity_blocks,
             file_info.erasure.block_size,
-        )
-        .expect("fixture erasure geometry");
+        );
         let mut writer = VecAsyncWriter::default();
         let (written, err) = erasure.decode(&mut writer, readers, 0, part.size, part.size).await;
         if let Some(err) = err {

crates/keystone/Cargo.toml

@@ -25,6 +25,9 @@ keywords = ["rustfs", "openstack", "keystone", "authentication", "s3"]
 categories = ["authentication", "web-programming"]
 authors.workspace = true
 
+[lints]
+workspace = true
+
 [dependencies]
 tokio = { workspace = true, features = ["rt", "sync"] }
 reqwest = { workspace = true, features = ["json"] }

crates/kms/AGENTS.md

@@ -25,25 +25,3 @@ For local KMS end-to-end tests, keep proxy bypass settings:
 NO_PROXY=127.0.0.1,localhost HTTP_PROXY= HTTPS_PROXY= http_proxy= https_proxy= \
 cargo test --package e2e_test test_local_kms_end_to_end -- --nocapture --test-threads=1
 ```
-
-## Local Key Export for SSE-S3 Migration Tests
-
-Use the read-only `local_kms_key_decrypt` example to export an AES-256 Local
-KMS key as the base64 value expected by `RUSTFS_SSE_S3_MASTER_KEY`:
-
-```bash
-export RUSTFS_KMS_LOCAL_MASTER_KEY='<local-kms-at-rest-master-key>'
-export RUSTFS_SSE_S3_MASTER_KEY="$(
-  cargo run -q -p rustfs-kms --example local_kms_key_decrypt -- \
-    /absolute/path/to/<key-id>.key
-)"
-```
-
-For a `plaintext-dev-only` Local KMS key file,
-`RUSTFS_KMS_LOCAL_MASTER_KEY` is not required.
-
-The example writes only the base64-encoded 32-byte key to stdout. Diagnostics
-go to stderr. Never paste its output into logs, shell history, issue comments,
-or committed configuration. The export path must remain read-only and must
-reuse `LocalKmsClient` decoding so current Argon2id and legacy key-file
-compatibility stay aligned with the backend.

crates/kms/examples/local_kms_key_decrypt.rs

@@ -1,112 +0,0 @@
-// Copyright 2024 RustFS Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64_STANDARD};
-use rustfs_kms::{LocalConfig, backends::local::LocalKmsClient};
-use std::io::{self, Write};
-use std::path::{Path, PathBuf};
-use zeroize::Zeroizing;
-
-const LOCAL_KMS_MASTER_KEY_ENV: &str = "RUSTFS_KMS_LOCAL_MASTER_KEY";
-
-fn usage(program: &str) -> String {
-    format!(
-        "Usage: {program} <local-kms-key-file>\n\
-         Reads {LOCAL_KMS_MASTER_KEY_ENV} when the key file is encrypted.\n\
-         Writes only the base64-encoded 32-byte key to stdout."
-    )
-}
-
-fn resolve_key_file(path: &Path) -> Result<(PathBuf, String), String> {
-    let canonical = std::fs::canonicalize(path).map_err(|error| format!("cannot open Local KMS key file: {error}"))?;
-    if canonical.extension().and_then(|extension| extension.to_str()) != Some("key") {
-        return Err("Local KMS key file must have a .key extension".to_string());
-    }
-    let key_dir = canonical
-        .parent()
-        .ok_or_else(|| "Local KMS key file must have a parent directory".to_string())?
-        .to_path_buf();
-    let key_id = canonical
-        .file_stem()
-        .and_then(|stem| stem.to_str())
-        .filter(|stem| !stem.is_empty())
-        .ok_or_else(|| "Local KMS key file name must contain a valid UTF-8 key ID".to_string())?
-        .to_string();
-    Ok((key_dir, key_id))
-}
-
-async fn run() -> Result<(), String> {
-    let mut args = std::env::args();
-    let program = args.next().unwrap_or_else(|| "local_kms_key_decrypt".to_string());
-    let Some(key_file) = args.next() else {
-        return Err(usage(&program));
-    };
-    if args.next().is_some() {
-        return Err(usage(&program));
-    }
-
-    let (key_dir, key_id) = resolve_key_file(Path::new(&key_file))?;
-    let master_key = std::env::var(LOCAL_KMS_MASTER_KEY_ENV).ok().filter(|value| !value.is_empty());
-    let client = LocalKmsClient::new_for_key_export(LocalConfig {
-        key_dir,
-        master_key,
-        file_permissions: Some(0o600),
-    })
-    .await
-    .map_err(|error| error.to_string())?;
-    let key_material = client
-        .decrypt_key_material_for_export(&key_id)
-        .await
-        .map_err(|error| error.to_string())?;
-    let encoded = Zeroizing::new(BASE64_STANDARD.encode(key_material.as_ref()));
-
-    let mut stdout = io::stdout().lock();
-    writeln!(stdout, "{}", encoded.as_str()).map_err(|error| format!("failed to write decrypted key: {error}"))
-}
-
-#[tokio::main]
-async fn main() {
-    if let Err(error) = run().await {
-        let _ = writeln!(io::stderr().lock(), "local_kms_key_decrypt: {error}");
-        std::process::exit(1);
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn resolve_key_file_extracts_directory_and_key_id() {
-        let directory = tempfile::tempdir().expect("create temporary directory");
-        let key_file = directory.path().join("migration-key.key");
-        std::fs::write(&key_file, b"{}").expect("create key file");
-
-        let (key_dir, key_id) = resolve_key_file(&key_file).expect("resolve key file");
-
-        assert_eq!(key_dir, directory.path().canonicalize().expect("canonical directory"));
-        assert_eq!(key_id, "migration-key");
-    }
-
-    #[test]
-    fn resolve_key_file_rejects_non_key_extension() {
-        let directory = tempfile::tempdir().expect("create temporary directory");
-        let key_file = directory.path().join("migration-key.json");
-        std::fs::write(&key_file, b"{}").expect("create key file");
-
-        let error = resolve_key_file(&key_file).expect_err("non-key file must be rejected");
-
-        assert!(error.contains(".key"));
-    }
-}

crates/kms/src/backends/local.rs

@@ -36,7 +36,6 @@ use std::path::{Component, Path, PathBuf};
 use std::time::Duration;
 use tokio::fs;
 use tracing::{debug, warn};
-use zeroize::Zeroizing;
 
 /// Reject key identifiers that would not name a single file directly inside the key
 /// directory.
@@ -147,45 +146,6 @@ impl LocalKmsClient {
         Ok(client)
     }
 
-    /// Open a Local KMS key directory without creating or modifying any files.
-    ///
-    /// This constructor is restricted to explicit key-export tooling. Normal
-    /// backend operation must use [`Self::new`].
-    pub async fn new_for_key_export(config: LocalConfig) -> Result<Self> {
-        if !fs::try_exists(&config.key_dir).await? {
-            return Err(KmsError::configuration_error("Local KMS key directory does not exist"));
-        }
-
-        let (master_cipher, legacy_master_cipher) = if let Some(ref master_key) = config.master_key {
-            let legacy_key = Self::derive_legacy_master_key(master_key)?;
-            let legacy_master_cipher = Aes256Gcm::new(&legacy_key);
-            let salt_path = Self::master_key_salt_path(&config);
-            let master_cipher = if fs::try_exists(&salt_path).await? {
-                let salt = fs::read(&salt_path).await?;
-                let salt: [u8; LOCAL_KMS_MASTER_KEY_SALT_LEN] = salt.try_into().map_err(|_| {
-                    KmsError::configuration_error(format!(
-                        "Local KMS master key salt at {} must be exactly {} bytes",
-                        salt_path.display(),
-                        LOCAL_KMS_MASTER_KEY_SALT_LEN
-                    ))
-                })?;
-                Aes256Gcm::new(&Self::derive_master_key(master_key, &salt)?)
-            } else {
-                Aes256Gcm::new(&legacy_key)
-            };
-            (Some(master_cipher), Some(legacy_master_cipher))
-        } else {
-            (None, None)
-        };
-
-        Ok(Self {
-            config,
-            master_cipher,
-            legacy_master_cipher,
-            dek_crypto: AesDekCrypto::new(),
-        })
-    }
-
     /// Derive a 256-bit key from the master key string using a persistent Argon2id salt.
     fn derive_master_key(master_key: &str, salt: &[u8]) -> Result<Key<Aes256Gcm>> {
         let params = Params::new(
@@ -475,20 +435,6 @@ impl LocalKmsClient {
         Ok(key_material)
     }
 
-    /// Decrypt an AES-256 Local KMS key for explicit migration tooling.
-    ///
-    /// The returned buffer is zeroized on drop. Callers must treat the value as
-    /// plaintext key material and avoid logging or persisting it.
-    pub async fn decrypt_key_material_for_export(&self, key_id: &str) -> Result<Zeroizing<[u8; 32]>> {
-        let (stored_key, key_material) = self.decode_stored_key(key_id).await?;
-        if stored_key.algorithm != "AES_256" {
-            return Err(KmsError::unsupported_algorithm(stored_key.algorithm));
-        }
-        let actual = key_material.len();
-        let key_material = key_material.try_into().map_err(|_| KmsError::invalid_key_size(32, actual))?;
-        Ok(Zeroizing::new(key_material))
-    }
-
     async fn validate_existing_keys(&self) -> Result<()> {
         let mut entries = fs::read_dir(&self.config.key_dir).await?;
         while let Some(entry) = entries.next_entry().await? {
@@ -1262,52 +1208,6 @@ mod tests {
         assert!(matches!(wrong_master_error, KmsError::CryptographicError { .. }));
     }
 
-    #[tokio::test]
-    async fn key_export_uses_existing_local_decryption_path_without_writing_files() {
-        let (client, _temp_dir) = create_test_client().await;
-        let key_id = "export-key";
-        client
-            .create_key(key_id, "AES_256", None)
-            .await
-            .expect("create encrypted key");
-        let expected = client.get_key_material(key_id).await.expect("load expected key material");
-        let salt_path = LocalKmsClient::master_key_salt_path(&client.config);
-        let salt_before = fs::read(&salt_path).await.expect("read existing salt");
-
-        let export_client = LocalKmsClient::new_for_key_export(client.config.clone())
-            .await
-            .expect("open read-only export client");
-        let exported = export_client
-            .decrypt_key_material_for_export(key_id)
-            .await
-            .expect("decrypt key for export");
-
-        assert_eq!(exported.as_ref(), expected.as_slice());
-        assert_eq!(fs::read(&salt_path).await.expect("read unchanged salt"), salt_before);
-    }
-
-    #[tokio::test]
-    async fn key_export_accepts_plaintext_dev_only_key_without_master_key() {
-        let (client, _temp_dir) = create_dev_mode_client().await;
-        let key_id = "plaintext-export-key";
-        client
-            .create_key(key_id, "AES_256", None)
-            .await
-            .expect("create plaintext-dev-only key");
-        let expected = client.get_key_material(key_id).await.expect("load expected key material");
-
-        let export_client = LocalKmsClient::new_for_key_export(client.config.clone())
-            .await
-            .expect("open read-only export client");
-        let exported = export_client
-            .decrypt_key_material_for_export(key_id)
-            .await
-            .expect("export plaintext-dev-only key");
-
-        assert_eq!(exported.as_ref(), expected.as_slice());
-        assert!(!LocalKmsClient::master_key_salt_path(&client.config).exists());
-    }
-
     #[tokio::test]
     async fn test_plaintext_dev_only_storage_is_explicit_and_loadable() {
         let (client, _temp_dir) = create_dev_mode_client().await;

crates/kms/src/lib.rs

@@ -89,7 +89,7 @@ pub use error::{KmsError, KmsUnavailableError, Result};
 pub use manager::KmsManager;
 pub use service::{DataKey, ObjectEncryptionService};
 pub use service_manager::{
-    KmsServiceManager, KmsServiceStatus, KmsStartOutcome, get_global_encryption_service, get_global_kms_service_manager,
+    KmsServiceManager, KmsServiceStatus, get_global_encryption_service, get_global_kms_service_manager,
     init_global_kms_service_manager,
 };
 pub use types::*;

crates/kms/src/service_manager.rs

@@ -21,13 +21,12 @@ use crate::manager::KmsManager;
 use crate::service::ObjectEncryptionService;
 use arc_swap::ArcSwap;
 use sha2::{Digest, Sha256};
-use std::future::Future;
 use std::sync::{
     Arc, OnceLock,
     atomic::{AtomicU64, Ordering},
 };
 use subtle::ConstantTimeEq;
-use tokio::sync::Mutex;
+use tokio::sync::{Mutex, RwLock};
 use tracing::{debug, error, info, warn};
 
 const LOG_COMPONENT_KMS: &str = "kms";
@@ -94,13 +93,6 @@ pub enum KmsServiceStatus {
     Error(String),
 }
 
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub enum KmsStartOutcome {
-    Started,
-    Restarted,
-    AlreadyRunning,
-}
-
 /// Service version information for zero-downtime reconfiguration
 #[derive(Clone)]
 struct ServiceVersion {
@@ -112,17 +104,16 @@ struct ServiceVersion {
     manager: Arc<KmsManager>,
 }
 
-#[derive(Clone)]
-struct RuntimeState {
-    config: Option<KmsConfig>,
-    status: KmsServiceStatus,
-    current_service: Option<ServiceVersion>,
-}
-
 /// Dynamic KMS service manager with versioned services for zero-downtime reconfiguration
 pub struct KmsServiceManager {
-    /// Atomically published configuration, status, and current service.
-    state: ArcSwap<RuntimeState>,
+    /// Current service version (if running)
+    /// Uses ArcSwap for atomic, lock-free service switching
+    /// This allows instant atomic updates without blocking readers
+    current_service: ArcSwap<Option<ServiceVersion>>,
+    /// Current configuration
+    config: Arc<RwLock<Option<KmsConfig>>>,
+    /// Current status
+    status: Arc<RwLock<KmsServiceStatus>>,
     /// Version counter (monotonically increasing)
     version_counter: Arc<AtomicU64>,
     /// Mutex to protect lifecycle operations (start, stop, reconfigure)
@@ -134,11 +125,9 @@ impl KmsServiceManager {
     /// Create a new KMS service manager (not configured)
     pub fn new() -> Self {
         Self {
-            state: ArcSwap::from_pointee(RuntimeState {
-                config: None,
-                status: KmsServiceStatus::NotConfigured,
-                current_service: None,
-            }),
+            current_service: ArcSwap::from_pointee(None),
+            config: Arc::new(RwLock::new(None)),
+            status: Arc::new(RwLock::new(KmsServiceStatus::NotConfigured)),
             version_counter: Arc::new(AtomicU64::new(0)),
             lifecycle_mutex: Arc::new(Mutex::new(())),
         }
@@ -146,67 +135,44 @@ impl KmsServiceManager {
 
     /// Get current service status
     pub async fn get_status(&self) -> KmsServiceStatus {
-        self.state.load().status.clone()
+        self.status.read().await.clone()
     }
 
     /// Get current configuration (if any)
     pub async fn get_config(&self) -> Option<KmsConfig> {
-        self.state.load().config.clone()
+        self.config.read().await.clone()
     }
 
     /// Get configuration for status and management responses without static key material.
     pub async fn get_redacted_config(&self) -> Option<KmsConfig> {
-        let mut config = self.state.load().config.clone()?;
-        Self::redact_config(&mut config);
-        Some(config)
-    }
-
-    /// Get status and redacted configuration from the same published snapshot.
-    pub async fn get_redacted_state(&self) -> (KmsServiceStatus, Option<KmsConfig>) {
-        let state = self.state.load();
-        let mut config = state.config.clone();
-        if let Some(config) = &mut config {
-            Self::redact_config(config);
-        }
-        (state.status.clone(), config)
-    }
-
-    fn redact_config(config: &mut KmsConfig) {
+        let mut config = self.config.read().await.clone()?;
         if let BackendConfig::Static(static_config) = &mut config.backend_config {
             use zeroize::Zeroize;
             static_config.secret_key.zeroize();
         }
+        Some(config)
     }
 
     /// Configure KMS with new configuration
     pub async fn configure(&self, new_config: KmsConfig) -> Result<()> {
-        self.configure_with_persistence(new_config, || async { Ok(()) }).await
-    }
-
-    /// Configure KMS and publish the in-memory state only after persistence succeeds.
-    ///
-    /// The persistence callback runs under the lifecycle lock and must not call
-    /// another lifecycle method on this manager.
-    pub async fn configure_with_persistence<Persist, PersistFuture>(&self, new_config: KmsConfig, persist: Persist) -> Result<()>
-    where
-        Persist: FnOnce() -> PersistFuture,
-        PersistFuture: Future<Output = Result<()>>,
-    {
-        new_config.validate()?;
         let _guard = self.lifecycle_mutex.lock().await;
-        let current = self.state.load_full();
-        validate_local_transition(current.config.as_ref(), &new_config)?;
-        if current.current_service.is_some() {
-            return Err(KmsError::configuration_error(
-                "Cannot configure KMS while it is running; use reconfigure instead",
-            ));
+        new_config.validate()?;
+        {
+            let config = self.config.read().await;
+            validate_local_transition(config.as_ref(), &new_config)?;
+        }
+
+        // Update configuration
+        {
+            let mut config = self.config.write().await;
+            *config = Some(new_config.clone());
+        }
+
+        // Update status
+        {
+            let mut status = self.status.write().await;
+            *status = KmsServiceStatus::Configured;
         }
-        persist().await?;
-        self.state.store(Arc::new(RuntimeState {
-            config: Some(new_config),
-            status: KmsServiceStatus::Configured,
-            current_service: None,
-        }));
 
         debug!(
             event = EVENT_KMS_SERVICE_STATE,
@@ -224,35 +190,19 @@ impl KmsServiceManager {
         self.start_internal().await
     }
 
-    /// Start or restart KMS with the running-state decision serialized with the lifecycle action.
-    pub async fn start_or_restart(&self, force: bool) -> Result<KmsStartOutcome> {
-        let _guard = self.lifecycle_mutex.lock().await;
-        let running = self.state.load().current_service.is_some();
-        if running && !force {
-            return Ok(KmsStartOutcome::AlreadyRunning);
-        }
-        self.start_internal().await?;
-        Ok(if running {
-            KmsStartOutcome::Restarted
-        } else {
-            KmsStartOutcome::Started
-        })
-    }
-
     /// Internal start implementation (called within lifecycle mutex)
     async fn start_internal(&self) -> Result<()> {
-        let state = self.state.load_full();
-        let config = match state.config.as_ref() {
-            Some(config) => config.clone(),
-            None => {
-                let err_msg = "Cannot start KMS: no configuration provided";
-                error!("{}", err_msg);
-                self.state.store(Arc::new(RuntimeState {
-                    config: None,
-                    status: KmsServiceStatus::Error(err_msg.to_string()),
-                    current_service: None,
-                }));
-                return Err(KmsError::configuration_error(err_msg));
+        let config = {
+            let config_guard = self.config.read().await;
+            match config_guard.as_ref() {
+                Some(config) => config.clone(),
+                None => {
+                    let err_msg = "Cannot start KMS: no configuration provided";
+                    error!("{}", err_msg);
+                    let mut status = self.status.write().await;
+                    *status = KmsServiceStatus::Error(err_msg.to_string());
+                    return Err(KmsError::configuration_error(err_msg));
+                }
             }
         };
 
@@ -265,9 +215,17 @@ impl KmsServiceManager {
             "KMS service starting"
         );
 
-        match self.create_healthy_service_version(&config).await {
+        match self.create_service_version(&config).await {
             Ok(service_version) => {
-                self.publish_running(config, service_version);
+                // Atomically update to new service version (lock-free, instant)
+                // ArcSwap::store() is a true atomic operation using CAS
+                self.current_service.store(Arc::new(Some(service_version)));
+
+                // Update status
+                {
+                    let mut status = self.status.write().await;
+                    *status = KmsServiceStatus::Running;
+                }
 
                 debug!(
                     event = EVENT_KMS_SERVICE_STATE,
@@ -281,24 +239,13 @@ impl KmsServiceManager {
             Err(e) => {
                 let err_msg = format!("Failed to create KMS backend: {e}");
                 error!("{}", err_msg);
-                if state.current_service.is_none() {
-                    self.state.store(Arc::new(RuntimeState {
-                        config: state.config.clone(),
-                        status: KmsServiceStatus::Error(err_msg.clone()),
-                        current_service: None,
-                    }));
-                }
+                let mut status = self.status.write().await;
+                *status = KmsServiceStatus::Error(err_msg.clone());
                 Err(KmsError::backend_error(&err_msg))
             }
         }
     }
 
-    /// Replace the running service without exposing a stopped interval.
-    pub async fn restart(&self) -> Result<()> {
-        let _guard = self.lifecycle_mutex.lock().await;
-        self.start_internal().await
-    }
-
     /// Stop KMS service
     ///
     /// Note: This stops accepting new operations, but existing operations using
@@ -320,16 +267,15 @@ impl KmsServiceManager {
 
         // Atomically clear current service version (lock-free, instant)
         // Note: Existing Arc references will keep the service alive until operations complete
-        let state = self.state.load_full();
-        self.state.store(Arc::new(RuntimeState {
-            config: state.config.clone(),
-            status: if state.config.is_some() {
-                KmsServiceStatus::Configured
-            } else {
-                KmsServiceStatus::NotConfigured
-            },
-            current_service: None,
-        }));
+        self.current_service.store(Arc::new(None));
+
+        // Update status (keep configuration)
+        {
+            let mut status = self.status.write().await;
+            if !matches!(*status, KmsServiceStatus::NotConfigured) {
+                *status = KmsServiceStatus::Configured;
+            }
+        }
 
         debug!(
             event = EVENT_KMS_SERVICE_STATE,
@@ -352,22 +298,6 @@ impl KmsServiceManager {
     /// This ensures zero downtime during reconfiguration, even for long-running
     /// operations like encrypting large files.
     pub async fn reconfigure(&self, new_config: KmsConfig) -> Result<()> {
-        self.reconfigure_with_persistence(new_config, || async { Ok(()) }).await
-    }
-
-    /// Reconfigure KMS after the candidate is healthy and persistence succeeds.
-    ///
-    /// The persistence callback runs under the lifecycle lock and must not call
-    /// another lifecycle method on this manager.
-    pub async fn reconfigure_with_persistence<Persist, PersistFuture>(
-        &self,
-        new_config: KmsConfig,
-        persist: Persist,
-    ) -> Result<()>
-    where
-        Persist: FnOnce() -> PersistFuture,
-        PersistFuture: Future<Output = Result<()>>,
-    {
         let _guard = self.lifecycle_mutex.lock().await;
 
         debug!(
@@ -378,18 +308,33 @@ impl KmsServiceManager {
             "KMS service reconfiguring"
         );
         new_config.validate()?;
-        validate_local_transition(self.state.load().config.as_ref(), &new_config)?;
+        {
+            let config = self.config.read().await;
+            validate_local_transition(config.as_ref(), &new_config)?;
+        }
 
         // Create new service version without stopping old one
         // This allows existing operations to continue while new operations use new service
-        match self.create_healthy_service_version(&new_config).await {
+        match self.create_service_version(&new_config).await {
             Ok(new_service_version) => {
                 // Get old version for logging (lock-free read)
-                let old_version = self.state.load().current_service.as_ref().map(|sv| sv.version);
+                let old_version = self.current_service.load().as_ref().as_ref().map(|sv| sv.version);
 
-                persist().await?;
+                {
+                    let mut config = self.config.write().await;
+                    *config = Some(new_config);
+                }
 
-                self.publish_running(new_config, new_service_version.clone());
+                // Atomically switch to new service version (lock-free, instant CAS operation)
+                // This is a true atomic operation - no waiting for locks, instant switch
+                // Old service will be dropped when no more Arc references exist
+                self.current_service.store(Arc::new(Some(new_service_version.clone())));
+
+                // Update status
+                {
+                    let mut status = self.status.write().await;
+                    *status = KmsServiceStatus::Running;
+                }
 
                 if let Some(old_ver) = old_version {
                     info!(
@@ -426,7 +371,7 @@ impl KmsServiceManager {
     /// Returns the manager from the current service version.
     /// Uses lock-free atomic load for optimal performance.
     pub async fn get_manager(&self) -> Option<Arc<KmsManager>> {
-        self.state.load().current_service.as_ref().map(|sv| sv.manager.clone())
+        self.current_service.load().as_ref().as_ref().map(|sv| sv.manager.clone())
     }
 
     /// Get encryption service (if running)
@@ -436,7 +381,7 @@ impl KmsServiceManager {
     /// This ensures new operations always use the latest service version,
     /// while existing operations continue using their Arc references.
     pub async fn get_encryption_service(&self) -> Option<Arc<ObjectEncryptionService>> {
-        self.state.load().current_service.as_ref().map(|sv| sv.service.clone())
+        self.current_service.load().as_ref().as_ref().map(|sv| sv.service.clone())
     }
 
     /// Get current service version number
@@ -444,16 +389,14 @@ impl KmsServiceManager {
     /// Useful for monitoring and debugging.
     /// Uses lock-free atomic load.
     pub async fn get_service_version(&self) -> Option<u64> {
-        self.state.load().current_service.as_ref().map(|sv| sv.version)
+        self.current_service.load().as_ref().as_ref().map(|sv| sv.version)
     }
 
     /// Health check for the KMS service
     pub async fn health_check(&self) -> Result<bool> {
-        let checked_state = self.state.load_full();
-        match checked_state.current_service.as_ref() {
-            Some(service_version) => {
-                let manager = service_version.manager.clone();
-                let checked_version = service_version.version;
+        let manager = self.get_manager().await;
+        match manager {
+            Some(manager) => {
                 // Perform health check on the backend
                 match manager.health_check().await {
                     Ok(healthy) => {
@@ -464,8 +407,9 @@ impl KmsServiceManager {
                     }
                     Err(e) => {
                         error!("KMS health check error: {}", e);
-                        let _guard = self.lifecycle_mutex.lock().await;
-                        self.mark_health_error_if_current(checked_version, &e);
+                        // Update status to error
+                        let mut status = self.status.write().await;
+                        *status = KmsServiceStatus::Error(format!("Health check failed: {e}"));
                         Err(e)
                     }
                 }
@@ -524,33 +468,6 @@ impl KmsServiceManager {
             manager: kms_manager,
         })
     }
-
-    async fn create_healthy_service_version(&self, config: &KmsConfig) -> Result<ServiceVersion> {
-        let service_version = self.create_service_version(config).await?;
-        if !service_version.manager.health_check().await? {
-            return Err(KmsError::backend_error("KMS backend health check failed"));
-        }
-        Ok(service_version)
-    }
-
-    fn publish_running(&self, config: KmsConfig, service_version: ServiceVersion) {
-        self.state.store(Arc::new(RuntimeState {
-            config: Some(config),
-            status: KmsServiceStatus::Running,
-            current_service: Some(service_version),
-        }));
-    }
-
-    fn mark_health_error_if_current(&self, checked_version: u64, error: &KmsError) {
-        let current = self.state.load_full();
-        if current.current_service.as_ref().map(|version| version.version) == Some(checked_version) {
-            self.state.store(Arc::new(RuntimeState {
-                config: current.config.clone(),
-                status: KmsServiceStatus::Error(format!("Health check failed: {error}")),
-                current_service: current.current_service.clone(),
-            }));
-        }
-    }
 }
 
 impl Default for KmsServiceManager {
@@ -583,11 +500,6 @@ pub async fn get_global_encryption_service() -> Option<Arc<ObjectEncryptionServi
 #[cfg(test)]
 mod tests {
     use super::*;
-    use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64_STANDARD};
-
-    fn static_config(key_id: &str, fill: u8) -> KmsConfig {
-        KmsConfig::static_kms(key_id.to_string(), BASE64_STANDARD.encode([fill; 32]))
-    }
 
     #[tokio::test]
     async fn configure_rejects_insecure_development_defaults_before_state_update() {
@@ -605,6 +517,8 @@ mod tests {
 
     #[tokio::test]
     async fn redacted_config_omits_static_key_material() {
+        use base64::Engine as _;
+
         let manager = KmsServiceManager::new();
         let encoded_key = base64::engine::general_purpose::STANDARD.encode([0x5au8; 32]);
         manager
@@ -619,136 +533,6 @@ mod tests {
         assert!(static_config.secret_key.is_empty());
     }
 
-    #[tokio::test]
-    async fn configure_persistence_failure_leaves_state_unchanged() {
-        let manager = KmsServiceManager::new();
-
-        let result = manager
-            .configure_with_persistence(static_config("key-a", 0x11), || async { Err(KmsError::backend_error("persist failed")) })
-            .await;
-
-        assert!(result.is_err());
-        assert_eq!(manager.get_status().await, KmsServiceStatus::NotConfigured);
-        assert!(manager.get_config().await.is_none());
-        assert!(manager.get_encryption_service().await.is_none());
-    }
-
-    #[tokio::test]
-    async fn configure_rejects_running_service_without_changing_snapshot() {
-        let manager = KmsServiceManager::new();
-        manager.configure(static_config("key-a", 0x11)).await.expect("configure");
-        manager.start().await.expect("start");
-        let version = manager.get_service_version().await;
-
-        let result = manager.configure(static_config("key-b", 0x22)).await;
-
-        assert!(result.is_err());
-        assert_eq!(manager.get_status().await, KmsServiceStatus::Running);
-        assert_eq!(manager.get_service_version().await, version);
-        assert_eq!(
-            manager.get_config().await.and_then(|config| config.default_key_id),
-            Some("key-a".to_string())
-        );
-    }
-
-    #[tokio::test]
-    async fn reconfigure_persistence_failure_keeps_old_running_snapshot() {
-        let manager = KmsServiceManager::new();
-        manager.configure(static_config("key-a", 0x11)).await.expect("configure");
-        manager.start().await.expect("start");
-        let old_version = manager.get_service_version().await;
-        let old_service = manager.get_encryption_service().await.expect("old service");
-
-        let result = manager
-            .reconfigure_with_persistence(static_config("key-b", 0x22), || async {
-                Err(KmsError::backend_error("persist failed"))
-            })
-            .await;
-
-        assert!(result.is_err());
-        assert_eq!(manager.get_status().await, KmsServiceStatus::Running);
-        assert_eq!(manager.get_service_version().await, old_version);
-        assert_eq!(
-            manager.get_config().await.and_then(|config| config.default_key_id),
-            Some("key-a".to_string())
-        );
-        assert!(Arc::ptr_eq(
-            &old_service,
-            &manager.get_encryption_service().await.expect("old service remains")
-        ));
-    }
-
-    #[tokio::test]
-    async fn reconfigure_candidate_failure_keeps_old_running_snapshot() {
-        let manager = KmsServiceManager::new();
-        manager.configure(static_config("key-a", 0x11)).await.expect("configure");
-        manager.start().await.expect("start");
-        let old_version = manager.get_service_version().await;
-        let invalid_parent = tempfile::NamedTempFile::new().expect("temporary file");
-        let invalid_config = KmsConfig::local(invalid_parent.path().join("keys")).with_insecure_development_defaults();
-
-        let result = manager.reconfigure(invalid_config).await;
-
-        assert!(result.is_err());
-        assert_eq!(manager.get_status().await, KmsServiceStatus::Running);
-        assert_eq!(manager.get_service_version().await, old_version);
-        assert_eq!(
-            manager.get_config().await.and_then(|config| config.default_key_id),
-            Some("key-a".to_string())
-        );
-    }
-
-    #[tokio::test]
-    async fn restart_never_unpublishes_the_running_service() {
-        let manager = Arc::new(KmsServiceManager::new());
-        manager.configure(static_config("key-a", 0x11)).await.expect("configure");
-        manager.start().await.expect("start");
-        let old_version = manager.get_service_version().await.expect("old version");
-        let restarting = {
-            let manager = manager.clone();
-            tokio::spawn(async move { manager.restart().await })
-        };
-
-        while !restarting.is_finished() {
-            assert!(manager.get_encryption_service().await.is_some());
-            tokio::task::yield_now().await;
-        }
-        restarting.await.expect("restart task").expect("restart");
-
-        assert!(manager.get_encryption_service().await.is_some());
-        assert!(manager.get_service_version().await.expect("new version") > old_version);
-        assert_eq!(manager.get_status().await, KmsServiceStatus::Running);
-    }
-
-    #[tokio::test]
-    async fn start_or_restart_decides_under_the_lifecycle_lock() {
-        let manager = KmsServiceManager::new();
-        manager.configure(static_config("key-a", 0x11)).await.expect("configure");
-
-        assert_eq!(manager.start_or_restart(false).await.expect("initial start"), KmsStartOutcome::Started);
-        let first_version = manager.get_service_version().await.expect("first version");
-        assert_eq!(
-            manager.start_or_restart(false).await.expect("already running"),
-            KmsStartOutcome::AlreadyRunning
-        );
-        assert_eq!(manager.get_service_version().await, Some(first_version));
-        assert_eq!(manager.start_or_restart(true).await.expect("forced restart"), KmsStartOutcome::Restarted);
-        assert!(manager.get_service_version().await.expect("restarted version") > first_version);
-    }
-
-    #[tokio::test]
-    async fn stale_health_failure_cannot_poison_new_service_status() {
-        let manager = KmsServiceManager::new();
-        manager.configure(static_config("key-a", 0x11)).await.expect("configure");
-        manager.start().await.expect("start");
-        let old_version = manager.get_service_version().await.expect("old version");
-        manager.restart().await.expect("restart");
-
-        manager.mark_health_error_if_current(old_version, &KmsError::backend_error("stale failure"));
-
-        assert_eq!(manager.get_status().await, KmsServiceStatus::Running);
-    }
-
     #[tokio::test]
     async fn forbidden_local_master_key_change_preserves_running_config_and_service() {
         use crate::types::{CreateKeyRequest, KeyUsage};

crates/madmin/src/metrics.rs

@@ -545,6 +545,8 @@ pub struct ScannerMetrics {
     pub collected_at: DateTime<Utc>,
     #[serde(rename = "current_cycle")]
     pub current_cycle: u64,
+    #[serde(rename = "current_cycle_active", default, skip_serializing_if = "Option::is_none")]
+    pub current_cycle_active: Option<bool>,
     #[serde(rename = "current_started")]
     pub current_started: DateTime<Utc>,
     #[serde(rename = "cycle_complete_times")]
@@ -718,7 +720,31 @@ pub struct ScannerMetrics {
 }
 
 impl ScannerMetrics {
+    pub fn is_current_cycle_active(&self) -> bool {
+        self.current_cycle_active.unwrap_or(self.current_cycle > 0)
+    }
+
     pub fn merge(&mut self, other: &Self) {
+        // Legacy nodes omit the activity field and use a non-zero cycle as
+        // their active signal. New nodes publish explicit first-cycle and idle
+        // states, including cycle zero.
+        let self_cycle_active = self.is_current_cycle_active();
+        let other_cycle_active = other.is_current_cycle_active();
+        let self_cycle_authority = (
+            self_cycle_active,
+            self.current_cycle,
+            self.cycles_completed_at.len(),
+            self.cycles_completed_at.as_slice(),
+            self.current_started,
+        );
+        let other_cycle_authority = (
+            other_cycle_active,
+            other.current_cycle,
+            other.cycles_completed_at.len(),
+            other.cycles_completed_at.as_slice(),
+            other.current_started,
+        );
+        let other_cycle_is_authoritative = self_cycle_authority < other_cycle_authority;
         let other_is_newer = self.collected_at < other.collected_at;
         if other_is_newer {
             self.collected_at = other.collected_at;
@@ -854,15 +880,12 @@ impl ScannerMetrics {
             self.ongoing_buckets = other.ongoing_buckets;
         }
 
-        if self.current_cycle < other.current_cycle {
+        if other_cycle_is_authoritative {
             self.current_cycle = other.current_cycle;
             self.cycles_completed_at = other.cycles_completed_at.clone();
             self.current_started = other.current_started;
         }
-
-        if other.cycles_completed_at.len() > self.cycles_completed_at.len() {
-            self.cycles_completed_at = other.cycles_completed_at.clone();
-        }
+        self.current_cycle_active = Some(self_cycle_active || other_cycle_active);
 
         if !other.life_time_ops.is_empty() && self.life_time_ops.is_empty() {
             self.life_time_ops = other.life_time_ops.clone();
@@ -931,7 +954,13 @@ impl Metrics {
         if let Some(scanner) = other.scanner.as_ref() {
             match self.scanner {
                 Some(ref mut s_scanner) => s_scanner.merge(scanner),
-                None => self.scanner = Some(scanner.clone()),
+                None => {
+                    let mut scanner = scanner.clone();
+                    if scanner.current_cycle_active.is_none() {
+                        scanner.current_cycle_active = Some(scanner.is_current_cycle_active());
+                    }
+                    self.scanner = Some(scanner);
+                }
             }
         }
 
@@ -1389,6 +1418,280 @@ pub struct Operations {
 mod tests {
     use super::*;
 
+    #[test]
+    fn scanner_metrics_serializes_cycle_active_presence() {
+        let missing_value = serde_json::to_value(ScannerMetrics::default()).expect("scanner metrics should serialize");
+        assert!(missing_value.get("current_cycle_active").is_none());
+        let missing: ScannerMetrics =
+            serde_json::from_value(missing_value).expect("older scanner metrics without cycle-active should decode");
+        assert_eq!(missing.current_cycle_active, None);
+
+        let explicit_false_value = serde_json::to_value(ScannerMetrics {
+            current_cycle_active: Some(false),
+            ..Default::default()
+        })
+        .expect("scanner metrics with explicit cycle-active should serialize");
+        assert_eq!(explicit_false_value["current_cycle_active"], serde_json::Value::Bool(false));
+        let explicit_false: ScannerMetrics =
+            serde_json::from_value(explicit_false_value).expect("explicit cycle-active should decode");
+        assert_eq!(explicit_false.current_cycle_active, Some(false));
+    }
+
+    #[test]
+    fn scanner_metrics_merge_prefers_an_active_first_cycle() {
+        let collected_at = Utc::now();
+        let idle_started = collected_at - chrono::Duration::hours(1);
+        let active_started = collected_at - chrono::Duration::seconds(5);
+        let mut scanner = ScannerMetrics {
+            collected_at,
+            current_cycle: 0,
+            current_cycle_active: Some(false),
+            current_started: idle_started,
+            ..Default::default()
+        };
+
+        scanner.merge(&ScannerMetrics {
+            collected_at: collected_at + chrono::Duration::seconds(1),
+            current_cycle: 0,
+            current_cycle_active: Some(true),
+            current_started: active_started,
+            ..Default::default()
+        });
+
+        assert_eq!(scanner.current_cycle_active, Some(true));
+        assert_eq!(scanner.current_cycle, 0);
+        assert_eq!(scanner.current_started, active_started);
+    }
+
+    #[test]
+    fn metrics_merge_preserves_explicit_active_first_cycle() {
+        let mut aggregated = Metrics::default();
+        aggregated.merge(&Metrics {
+            scanner: Some(ScannerMetrics {
+                current_cycle_active: Some(true),
+                ..Default::default()
+            }),
+            ..Default::default()
+        });
+
+        let scanner = aggregated.scanner.expect("aggregated scanner metrics");
+        assert_eq!(scanner.current_cycle_active, Some(true));
+        assert_eq!(scanner.current_cycle, 0);
+    }
+
+    #[test]
+    fn scanner_metrics_merge_preserves_legacy_nonzero_active_signal() {
+        let collected_at = Utc::now();
+        let mut scanner = ScannerMetrics {
+            collected_at,
+            ..Default::default()
+        };
+
+        scanner.merge(&ScannerMetrics {
+            collected_at: collected_at + chrono::Duration::seconds(1),
+            current_cycle: 7,
+            ..Default::default()
+        });
+
+        assert_eq!(scanner.current_cycle_active, Some(true));
+        assert_eq!(scanner.current_cycle, 7);
+    }
+
+    #[test]
+    fn metrics_merge_normalizes_first_legacy_scanner_snapshot() {
+        let legacy = Metrics {
+            scanner: Some(ScannerMetrics {
+                current_cycle: 7,
+                ..Default::default()
+            }),
+            ..Default::default()
+        };
+        let mut aggregated = Metrics::default();
+
+        aggregated.merge(&legacy);
+
+        let scanner = aggregated.scanner.expect("aggregated scanner metrics");
+        assert_eq!(scanner.current_cycle_active, Some(true));
+        assert_eq!(scanner.current_cycle, 7);
+    }
+
+    #[test]
+    fn scanner_metrics_merge_preserves_explicit_inactive_nonzero_cycle() {
+        let collected_at = Utc::now();
+        let mut scanner = ScannerMetrics::default();
+
+        scanner.merge(&ScannerMetrics {
+            collected_at,
+            current_cycle: 7,
+            current_cycle_active: Some(false),
+            ..Default::default()
+        });
+
+        assert_eq!(scanner.current_cycle_active, Some(false));
+        assert_eq!(scanner.current_cycle, 7);
+    }
+
+    #[test]
+    fn scanner_metrics_merge_cycle_active_is_order_independent() {
+        let collected_at = Utc::now();
+        let legacy_active = ScannerMetrics {
+            collected_at,
+            current_cycle: 7,
+            current_started: collected_at - chrono::Duration::seconds(10),
+            ..Default::default()
+        };
+        let explicit_idle = ScannerMetrics {
+            collected_at,
+            current_cycle: 0,
+            current_cycle_active: Some(false),
+            current_started: collected_at - chrono::Duration::hours(1),
+            ..Default::default()
+        };
+
+        let mut legacy_first = legacy_active.clone();
+        legacy_first.merge(&explicit_idle);
+        let mut legacy_second = explicit_idle.clone();
+        legacy_second.merge(&legacy_active);
+        assert_eq!(legacy_first.current_cycle_active, Some(true));
+        assert_eq!(legacy_second.current_cycle_active, Some(true));
+        assert_eq!(legacy_first.current_cycle, 7);
+        assert_eq!(legacy_second.current_cycle, 7);
+
+        let explicit_inactive_nonzero = ScannerMetrics {
+            collected_at,
+            current_cycle: 7,
+            current_cycle_active: Some(false),
+            ..Default::default()
+        };
+        let mut inactive_first = explicit_inactive_nonzero.clone();
+        inactive_first.merge(&explicit_idle);
+        let mut inactive_second = explicit_idle;
+        inactive_second.merge(&explicit_inactive_nonzero);
+        assert_eq!(inactive_first.current_cycle_active, Some(false));
+        assert_eq!(inactive_second.current_cycle_active, Some(false));
+        assert_eq!(inactive_first.current_cycle, 7);
+        assert_eq!(inactive_second.current_cycle, 7);
+    }
+
+    #[test]
+    fn scanner_metrics_merge_cycle_authority_is_order_independent() {
+        let collected_at = Utc::now();
+        let completion = collected_at - chrono::Duration::minutes(1);
+        let earlier_active = ScannerMetrics {
+            collected_at,
+            current_cycle: 7,
+            current_cycle_active: Some(true),
+            current_started: collected_at - chrono::Duration::seconds(10),
+            cycles_completed_at: vec![completion],
+            ..Default::default()
+        };
+        let later_active = ScannerMetrics {
+            collected_at: collected_at + chrono::Duration::seconds(1),
+            current_cycle: 7,
+            current_cycle_active: Some(true),
+            current_started: collected_at - chrono::Duration::seconds(5),
+            cycles_completed_at: vec![completion],
+            ..Default::default()
+        };
+
+        let mut earlier_first = earlier_active.clone();
+        earlier_first.merge(&later_active);
+        let mut later_first = later_active.clone();
+        later_first.merge(&earlier_active);
+
+        assert_eq!(earlier_first.current_started, later_active.current_started);
+        assert_eq!(later_first.current_started, later_active.current_started);
+        assert_eq!(earlier_first.cycles_completed_at, later_active.cycles_completed_at);
+        assert_eq!(later_first.cycles_completed_at, later_active.cycles_completed_at);
+
+        let stale_idle = ScannerMetrics {
+            collected_at,
+            current_cycle_active: Some(false),
+            current_started: collected_at - chrono::Duration::hours(1),
+            ..Default::default()
+        };
+        let completed_idle = ScannerMetrics {
+            collected_at: collected_at + chrono::Duration::seconds(1),
+            current_cycle_active: Some(false),
+            current_started: collected_at - chrono::Duration::seconds(5),
+            cycles_completed_at: vec![completion],
+            ..Default::default()
+        };
+
+        let mut stale_first = stale_idle.clone();
+        stale_first.merge(&completed_idle);
+        let mut completed_first = completed_idle.clone();
+        completed_first.merge(&stale_idle);
+
+        assert_eq!(stale_first.current_started, completed_idle.current_started);
+        assert_eq!(completed_first.current_started, completed_idle.current_started);
+        assert_eq!(stale_first.cycles_completed_at, completed_idle.cycles_completed_at);
+        assert_eq!(completed_first.cycles_completed_at, completed_idle.cycles_completed_at);
+    }
+
+    #[test]
+    fn scanner_metrics_merge_cycle_authority_is_associative() {
+        let collected_at = Utc::now();
+        let older_completion = collected_at - chrono::Duration::minutes(3);
+        let last_completion = collected_at - chrono::Duration::minutes(1);
+        let cycle_seven = ScannerMetrics {
+            collected_at,
+            current_cycle: 7,
+            current_cycle_active: Some(true),
+            current_started: collected_at - chrono::Duration::seconds(10),
+            cycles_completed_at: vec![older_completion, last_completion],
+            ..Default::default()
+        };
+        let cycle_eight = ScannerMetrics {
+            collected_at: collected_at + chrono::Duration::seconds(1),
+            current_cycle: 8,
+            current_cycle_active: Some(true),
+            current_started: collected_at - chrono::Duration::seconds(5),
+            cycles_completed_at: vec![last_completion],
+            ..Default::default()
+        };
+        let newer_idle = ScannerMetrics {
+            collected_at: collected_at + chrono::Duration::hours(1),
+            current_cycle_active: Some(false),
+            current_started: collected_at - chrono::Duration::hours(1),
+            ..Default::default()
+        };
+
+        let mut left_associative = cycle_seven.clone();
+        left_associative.merge(&cycle_eight);
+        left_associative.merge(&newer_idle);
+
+        let mut right_group = cycle_eight.clone();
+        right_group.merge(&newer_idle);
+        let mut right_associative = cycle_seven.clone();
+        right_associative.merge(&right_group);
+
+        assert_eq!(left_associative.current_cycle, 8);
+        assert_eq!(right_associative.current_cycle, 8);
+        assert_eq!(left_associative.current_started, cycle_eight.current_started);
+        assert_eq!(right_associative.current_started, cycle_eight.current_started);
+        assert_eq!(left_associative.cycles_completed_at, cycle_eight.cycles_completed_at);
+        assert_eq!(right_associative.cycles_completed_at, cycle_eight.cycles_completed_at);
+
+        for order in [
+            [&cycle_seven, &cycle_eight, &newer_idle],
+            [&cycle_seven, &newer_idle, &cycle_eight],
+            [&cycle_eight, &cycle_seven, &newer_idle],
+            [&cycle_eight, &newer_idle, &cycle_seven],
+            [&newer_idle, &cycle_seven, &cycle_eight],
+            [&newer_idle, &cycle_eight, &cycle_seven],
+        ] {
+            let mut merged = ScannerMetrics::default();
+            for scanner in order {
+                merged.merge(scanner);
+            }
+            assert_eq!(merged.current_cycle_active, Some(true));
+            assert_eq!(merged.current_cycle, 8);
+            assert_eq!(merged.current_started, cycle_eight.current_started);
+            assert_eq!(merged.cycles_completed_at, cycle_eight.cycles_completed_at);
+        }
+    }
+
     #[test]
     fn scanner_metrics_merge_aggregates_partial_cycles_by_source() {
         let collected_at = Utc::now();

crates/obs/src/metrics/stats_collector.rs

@@ -262,11 +262,11 @@ async fn obs_site_replication_stats() -> ReplicationStats {
 }
 
 fn current_scanner_cycle_age_seconds(
-    current_cycle: u64,
+    current_cycle_active: bool,
     current_started: chrono::DateTime<Utc>,
     now: chrono::DateTime<Utc>,
 ) -> u64 {
-    if current_cycle == 0 {
+    if !current_cycle_active {
         0
     } else {
         now.signed_duration_since(current_started).num_seconds().max(0) as u64
@@ -1090,7 +1090,7 @@ pub async fn collect_scanner_metric_stats() -> Option<ScannerStats> {
     let reference_time = metrics.cycles_completed_at.last().copied().unwrap_or(metrics.current_started);
     let last_activity_seconds = now.signed_duration_since(reference_time).num_seconds().max(0) as u64;
     let active_paths = metrics.active_scan_paths as u64;
-    let current_cycle_age_seconds = current_scanner_cycle_age_seconds(metrics.current_cycle, metrics.current_started, now);
+    let current_cycle_age_seconds = current_scanner_cycle_age_seconds(metrics.current_cycle_active, metrics.current_started, now);
     let current_scan_mode = scanner_scan_mode_code(&metrics.current_scan_mode);
     let current_cycle_age = current_cycle_age_seconds as f64;
     let last_cycle_duration = metrics.last_cycle_duration_seconds;
@@ -1464,21 +1464,21 @@ mod tests {
     fn current_scanner_cycle_age_seconds_returns_zero_when_idle() {
         let now = Utc::now();
 
-        assert_eq!(current_scanner_cycle_age_seconds(0, now - chrono::Duration::seconds(30), now), 0);
+        assert_eq!(current_scanner_cycle_age_seconds(false, now - chrono::Duration::seconds(30), now), 0);
     }
 
     #[test]
     fn current_scanner_cycle_age_seconds_clamps_future_start() {
         let now = Utc::now();
 
-        assert_eq!(current_scanner_cycle_age_seconds(4, now + chrono::Duration::seconds(30), now), 0);
+        assert_eq!(current_scanner_cycle_age_seconds(true, now + chrono::Duration::seconds(30), now), 0);
     }
 
     #[test]
-    fn current_scanner_cycle_age_seconds_reports_active_elapsed_time() {
+    fn current_scanner_cycle_age_seconds_reports_active_first_cycle_elapsed_time() {
         let now = Utc::now();
 
-        assert_eq!(current_scanner_cycle_age_seconds(4, now - chrono::Duration::seconds(45), now), 45);
+        assert_eq!(current_scanner_cycle_age_seconds(true, now - chrono::Duration::seconds(45), now), 45);
     }
 
     #[test]

crates/protocols/Cargo.toml

@@ -25,6 +25,9 @@ description = "Protocol implementations for RustFS (FTPS, SFTP, etc.)"
 keywords = ["ftp", "sftp", "protocol", "storage", "rustfs"]
 categories = ["network-programming", "filesystem"]
 
+[lints]
+workspace = true
+
 [features]
 default = []
 ftps = ["dep:libunftp", "dep:unftp-core", "dep:rustls", "dep:rustfs-tls-runtime", "dep:subtle"]

crates/protocols/src/common/dummy_storage.rs

@@ -608,7 +608,7 @@ impl StorageBackend for DummyBackend {
             inner.put_object_calls.push(PutObjectCall {
                 bucket: input.bucket.to_string(),
                 key: input.key.to_string(),
-                metadata: input.metadata.clone(),
+                metadata: input.metadata,
             });
             let stall = inner.stall_put_object;
             let entered = inner.put_object_entered.clone();
@@ -731,7 +731,7 @@ impl StorageBackend for DummyBackend {
             inner.create_multipart_calls.push(CreateMultipartCall {
                 bucket: input.bucket.to_string(),
                 key: input.key.to_string(),
-                metadata: input.metadata.clone(),
+                metadata: input.metadata,
             });
         }
         match self.inner.lock().expect("lock").create_multipart_upload.pop_front() {
@@ -749,7 +749,7 @@ impl StorageBackend for DummyBackend {
             inner.upload_part_calls.push(UploadPartCall {
                 bucket: input.bucket.to_string(),
                 key: input.key.to_string(),
-                upload_id: input.upload_id.to_string(),
+                upload_id: input.upload_id,
                 part_number: input.part_number,
                 content_length: input.content_length,
             });
@@ -787,7 +787,7 @@ impl StorageBackend for DummyBackend {
             inner.complete_multipart_calls.push(CompleteCall {
                 bucket: input.bucket.to_string(),
                 key: input.key.to_string(),
-                upload_id: input.upload_id.to_string(),
+                upload_id: input.upload_id,
                 part_count,
             });
         }
@@ -808,7 +808,7 @@ impl StorageBackend for DummyBackend {
             inner.abort_multipart_calls.push(AbortCall {
                 bucket: input.bucket.to_string(),
                 key: input.key.to_string(),
-                upload_id: input.upload_id.to_string(),
+                upload_id: input.upload_id,
             });
         }
         match self.inner.lock().expect("lock").abort_multipart_upload.pop_front() {

crates/protocols/src/ftps/driver.rs

@@ -379,9 +379,7 @@ where
             .await
             .map_err(|_| Error::new(ErrorKind::PermanentFileNotAvailable, "Access denied"))?;
 
-        let prefix_with_slash = prefix
-            .clone()
-            .map(|p| if p.ends_with('/') { p.to_string() } else { format!("{}/", p) });
+        let prefix_with_slash = prefix.clone().map(|p| if p.ends_with('/') { p } else { format!("{}/", p) });
 
         let list_input = ListObjectsV2Input::builder()
             .bucket(bucket)

crates/protocols/src/ftps/server.rs

@@ -371,7 +371,7 @@ impl UserDetailProvider for FtpsUserDetailProvider {
 
         let ftps_user = FtpsUser {
             username: principal.username.clone(),
-            name: identity.credentials.name.clone(),
+            name: identity.credentials.name,
             session_context,
         };
 

crates/protocols/src/sftp/write.rs

@@ -1495,12 +1495,12 @@ mod tests {
             let buffer_len_u64 = part_buffer_len as u64;
             let phase = match phase_variant {
                 0 => WritePhase::Buffering {
-                    part_buffer: part_buffer.clone(),
+                    part_buffer,
                 },
                 1 => WritePhase::Streaming {
                     upload_id: "UP-proptest".to_string(),
                     abort_authorized: true,
-                    part_buffer: part_buffer.clone(),
+                    part_buffer,
                     uploaded_parts: Vec::new(),
                     next_part_number,
                 },

crates/protocols/src/swift/bulk.rs

@@ -627,9 +627,7 @@ mod tests {
     #[test]
     fn test_parse_paths_with_empty_lines() {
         let body = "/container1/file1.txt\n\n/container2/file2.txt\n   \n/container1/file3.txt";
-        let paths: Vec<&str> = body.lines().filter(|line| !line.trim().is_empty()).collect();
-
-        assert_eq!(paths.len(), 3);
+        assert_eq!(body.lines().filter(|line| !line.trim().is_empty()).count(), 3);
     }
 
     /// Tests for the `extract_tar_entries` async function.

crates/protocols/src/swift/object.rs

@@ -206,10 +206,9 @@ impl ObjectKeyMapper {
     #[allow(dead_code)] // Used in: object operations
     pub fn normalize_path(object: &str) -> String {
         // Split by '/', filter out empty segments (except if it's the end)
-        let segments: Vec<&str> = object.split('/').collect();
         let has_trailing_slash = object.ends_with('/');
 
-        let normalized_segments: Vec<&str> = segments.into_iter().filter(|s| !s.is_empty()).collect();
+        let normalized_segments: Vec<&str> = object.split('/').filter(|s| !s.is_empty()).collect();
 
         let mut result = normalized_segments.join("/");
 

crates/protocols/src/webdav/driver.rs

@@ -222,7 +222,7 @@ where
                         created: modified,
                         is_dir: false,
                         etag: output.e_tag.as_ref().map(etag_to_string),
-                        content_type: output.content_type.map(|c| c.to_string()),
+                        content_type: output.content_type,
                     }) as Box<dyn DavMetaData>)
                 }
                 Err(e) => {
@@ -1185,7 +1185,7 @@ where
                             created: modified,
                             is_dir: false,
                             etag: output.e_tag.as_ref().map(etag_to_string),
-                            content_type: output.content_type.map(|c| c.to_string()),
+                            content_type: output.content_type,
                         }) as Box<dyn DavMetaData>)
                     }
                     ResolvedPath::Directory { metadata, .. } => {
@@ -1204,7 +1204,7 @@ where
                             created: modified,
                             is_dir: true,
                             etag: metadata.as_ref().and_then(|output| output.e_tag.as_ref().map(etag_to_string)),
-                            content_type: metadata.and_then(|output| output.content_type.map(|c| c.to_string())),
+                            content_type: metadata.and_then(|output| output.content_type),
                         }) as Box<dyn DavMetaData>)
                     }
                 };
@@ -2035,7 +2035,7 @@ mod tests {
             _access_key: &str,
             _secret_key: &str,
         ) -> Result<ListObjectsV2Output, Self::Error> {
-            let prefix = input.prefix.map(|p| p.to_string()).unwrap_or_default();
+            let prefix = input.prefix.unwrap_or_default();
             let mut keys: Vec<String> = self
                 .state
                 .lock()

crates/protocols/tests/swift_listing_symlink_tests.rs

@@ -436,12 +436,10 @@ fn test_symlink_to_nested_object() {
 fn test_listing_empty_container() {
     let objects: Vec<&str> = vec![];
 
-    let filtered: Vec<_> = objects.iter().collect();
-    assert_eq!(filtered.len(), 0);
+    assert!(objects.is_empty());
 
     // With prefix
-    let with_prefix: Vec<_> = objects.iter().filter(|o| o.starts_with("prefix/")).collect();
-    assert_eq!(with_prefix.len(), 0);
+    assert!(!objects.iter().any(|o| o.starts_with("prefix/")));
 }
 
 /// Test listing lexicographic ordering

crates/rio-v2/tests/minio_fixture_lab/README.md

@@ -137,7 +137,7 @@ tests read):
 ./capture_via_docker.sh
 
 RUSTFS_MINIO_STATIC_KMS_KEY_B64=IyqsU3kMFloCNup4BsZtf/rmfHVcTgznO2F25CkEH1g= \
-  cargo test -p rustfs --features rio-v2 storage::minio_generated_read_test --lib -- --ignored
+  cargo test -p rustfs-ecstore --features rio-v2 --test minio_generated_read_test -- --ignored
 ```
 
 This is exactly what the nightly `minio-interop` GitHub Actions workflow runs

crates/scanner/Cargo.toml

@@ -47,7 +47,7 @@ rmp-serde = { workspace = true }
 hmac = { workspace = true }
 sha2 = { workspace = true }
 rustfs-filemeta = { workspace = true }
-tokio-util = { workspace = true, features = ["io", "compat"] }
+tokio-util = { workspace = true, features = ["io", "compat", "rt"] }
 rustfs-ecstore = { workspace = true }
 rustfs-storage-api = { workspace = true }
 http = { workspace = true }

crates/scanner/src/scanner.rs

@@ -14,6 +14,8 @@
 
 use std::collections::BTreeMap;
 use std::future::Future;
+#[cfg(test)]
+use std::sync::Mutex as StdMutex;
 use std::sync::{Arc, LazyLock, RwLock};
 
 use crate::ScannerObjectIO;
@@ -38,8 +40,8 @@ use bytes::Bytes;
 use chrono::{DateTime, Utc};
 use rustfs_common::heal_channel::HealScanMode;
 use rustfs_common::metrics::{
-    CurrentCycle, Metric, Metrics, ScanCyclePartialReason, ScannerUsageSaveResult, ScannerWorkSource, emit_scan_cycle_complete,
-    emit_scan_cycle_partial_with_source, emit_scan_cycle_superseded, global_metrics,
+    CurrentCycle, Metric, Metrics, ScanCyclePartialReason, ScanCycleWorkSnapshot, ScannerUsageSaveResult, ScannerWorkSource,
+    emit_scan_cycle_complete, emit_scan_cycle_partial_with_source, emit_scan_cycle_superseded, global_metrics,
 };
 use rustfs_config::ScannerSpeed;
 #[cfg(test)]
@@ -50,9 +52,12 @@ use rustfs_config::{
 use rustfs_config::{ENV_SCANNER_CYCLE, ENV_SCANNER_SPEED, ENV_SCANNER_START_DELAY_SECS};
 use serde::{Deserialize, Serialize};
 use sha2::{Digest as _, Sha256};
+#[cfg(test)]
+use tokio::sync::Notify;
 use tokio::sync::mpsc;
 use tokio::time::{Duration, Instant};
 use tokio_util::sync::CancellationToken;
+use tokio_util::task::AbortOnDropHandle;
 use tracing::{debug, error, info, instrument, warn};
 
 use crate::storage_api::scan::{
@@ -93,6 +98,44 @@ const SCANNER_CYCLE_STATE_MAGIC: &[u8; 8] = b"RSCYC001";
 const SCANNER_CYCLE_STATE_HEADER_LEN: usize = 24;
 #[cfg(test)]
 const ENV_SCANNER_START_DELAY_SECS_DEPRECATED: &str = "RUSTFS_DATA_SCANNER_START_DELAY_SECS";
+#[cfg(test)]
+type ScannerCycleStatePersistTestHook = (u64, Arc<Notify>);
+#[cfg(test)]
+static SCANNER_CYCLE_STATE_PERSIST_TEST_HOOK: LazyLock<StdMutex<Option<ScannerCycleStatePersistTestHook>>> =
+    LazyLock::new(|| StdMutex::new(None));
+
+#[cfg(test)]
+struct ScannerCycleStatePersistTestHookGuard;
+
+#[cfg(test)]
+impl Drop for ScannerCycleStatePersistTestHookGuard {
+    fn drop(&mut self) {
+        *SCANNER_CYCLE_STATE_PERSIST_TEST_HOOK
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner()) = None;
+    }
+}
+
+#[cfg(test)]
+fn set_scanner_cycle_state_persist_test_hook(leader_epoch: u64, reached: Arc<Notify>) -> ScannerCycleStatePersistTestHookGuard {
+    *SCANNER_CYCLE_STATE_PERSIST_TEST_HOOK
+        .lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some((leader_epoch, reached));
+    ScannerCycleStatePersistTestHookGuard
+}
+
+#[cfg(test)]
+fn notify_scanner_cycle_state_persist_test_hook(leader_epoch: u64) {
+    let reached = SCANNER_CYCLE_STATE_PERSIST_TEST_HOOK
+        .lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner())
+        .as_ref()
+        .filter(|(expected_epoch, _)| *expected_epoch == leader_epoch)
+        .map(|(_, reached)| reached.clone());
+    if let Some(reached) = reached {
+        reached.notify_one();
+    }
+}
 
 #[derive(Debug, thiserror::Error)]
 enum ScannerCycleStateError {
@@ -1691,10 +1734,10 @@ fn data_usage_persist_timeout() -> Duration {
     DataUsageCache::persistence_timeout()
 }
 
-async fn mark_scan_cycle_idle(cycle_info: &mut CurrentCycle) {
+async fn mark_scan_cycle_idle(cycle_info: &mut CurrentCycle, cycle_metrics_guard: &mut ScannerCycleMetricsGuard) {
     cycle_info.current = 0;
     global_metrics().clear_current_scan_mode();
-    global_metrics().set_cycle(Some(cycle_info.clone())).await;
+    cycle_metrics_guard.finish(cycle_info.clone()).await;
 }
 
 fn encode_scanner_cycle_state(cycle_info: &CurrentCycle, leader_epoch: u64) -> Result<Vec<u8>, ScannerCycleStateError> {
@@ -2218,6 +2261,8 @@ async fn persist_scanner_cycle_state(
             return false;
         }
 
+        #[cfg(test)]
+        notify_scanner_cycle_state_persist_test_hook(leader_epoch);
         match save_config_with_preconditions(storeapi.clone(), &DATA_USAGE_BLOOM_NAME_PATH, buf.clone(), revision.preconditions())
             .await
         {
@@ -2340,7 +2385,6 @@ async fn persist_scanner_cycle_state(
 
                     if persisted_cycle.next >= cycle_info.next {
                         *cycle_info = persisted_cycle;
-                        global_metrics().set_cycle(Some(cycle_info.clone())).await;
                         debug!(
                             target: "rustfs::scanner",
                             event = EVENT_SCANNER_PERSIST_STATE,
@@ -2406,6 +2450,7 @@ async fn finalize_partial_scan_cycle(
     cycle_info: &mut CurrentCycle,
     revision: &mut DataUsageCacheRevision,
     leader_epoch: u64,
+    cycle_metrics_guard: &mut ScannerCycleMetricsGuard,
 ) -> bool {
     // A budget-limited cycle is deliberate pacing, not a failure. The cycle counter
     // must still advance (and persist) because per-bucket next_cycle is stamped from
@@ -2422,11 +2467,14 @@ async fn finalize_partial_scan_cycle(
             error = %err,
             "Scanner partial cycle could not advance"
         );
-        mark_scan_cycle_idle(cycle_info).await;
+        mark_scan_cycle_idle(cycle_info, cycle_metrics_guard).await;
         return false;
     }
-    mark_scan_cycle_idle(cycle_info).await;
-    persist_scanner_cycle_state(ctx, storeapi, cycle_info, revision, leader_epoch).await
+    cycle_info.current = 0;
+    global_metrics().clear_current_scan_mode();
+    let persisted = persist_scanner_cycle_state(ctx, storeapi, cycle_info, revision, leader_epoch).await;
+    cycle_metrics_guard.finish(cycle_info.clone()).await;
+    persisted
 }
 
 async fn persist_required_scanner_cycle_floor(
@@ -2436,6 +2484,7 @@ async fn persist_required_scanner_cycle_floor(
     revision: &mut DataUsageCacheRevision,
     leader_epoch: u64,
     required_cycle: u64,
+    cycle_metrics_guard: &mut ScannerCycleMetricsGuard,
 ) -> bool {
     if required_cycle <= cycle_info.current || required_cycle == u64::MAX {
         error!(
@@ -2448,13 +2497,16 @@ async fn persist_required_scanner_cycle_floor(
             state = "invalid_cache_cycle_floor",
             "Scanner cache cycle floor is invalid"
         );
-        mark_scan_cycle_idle(cycle_info).await;
+        mark_scan_cycle_idle(cycle_info, cycle_metrics_guard).await;
         return false;
     }
 
     cycle_info.next = cycle_info.next.max(required_cycle);
-    mark_scan_cycle_idle(cycle_info).await;
-    persist_scanner_cycle_state(ctx, storeapi, cycle_info, revision, leader_epoch).await
+    cycle_info.current = 0;
+    global_metrics().clear_current_scan_mode();
+    let persisted = persist_scanner_cycle_state(ctx, storeapi, cycle_info, revision, leader_epoch).await;
+    cycle_metrics_guard.finish(cycle_info.clone()).await;
+    persisted
 }
 
 async fn await_scanner_cycle_with_lock_fence<Cycle, LockLost>(
@@ -2513,7 +2565,7 @@ async fn run_data_scanner_cycle(
     let now = Instant::now();
     cycle_info.started = Utc::now();
 
-    global_metrics().set_cycle(Some(cycle_info.clone())).await;
+    let mut cycle_metrics_guard = ScannerCycleMetricsGuard::new(cycle_info.clone()).await;
 
     let mut background_heal_info = read_background_heal_info(storeapi.clone()).await;
 
@@ -2564,14 +2616,14 @@ async fn run_data_scanner_cycle(
                 "Scanner cycle could not capture the data usage persistence baseline"
             );
             emit_scan_cycle_complete(false, cycle_start.elapsed());
-            mark_scan_cycle_idle(cycle_info).await;
+            mark_scan_cycle_idle(cycle_info, &mut cycle_metrics_guard).await;
             return ScannerCycleOutcome::Failed;
         }
     };
     let (sender, receiver) = mpsc::channel::<DataUsageInfo>(1);
     let storeapi_clone = storeapi.clone();
     let ctx_clone = ctx.clone();
-    let mut usage_persist_task = tokio::spawn(async move {
+    let mut usage_persist_task = AbortOnDropHandle::new(tokio::spawn(async move {
         store_data_usage_in_backend_with_outcome_for_epoch_and_baseline(
             ctx_clone,
             storeapi_clone,
@@ -2580,10 +2632,9 @@ async fn run_data_scanner_cycle(
             Some(usage_persist_baseline),
         )
         .await
-    });
+    }));
 
     let done_cycle = Metrics::time(Metric::ScanCycle);
-    let cycle_work_start = global_metrics().start_scan_cycle_work();
     let cycle_budget = ScannerCycleBudget::new(ctx, cycle_budget_config);
     let scan_result = storeapi
         .clone()
@@ -2640,7 +2691,6 @@ async fn run_data_scanner_cycle(
         }
     };
     let unresolved_heal_work = global_metrics().current_scan_cycle_has_unresolved_heal_work();
-    global_metrics().finish_scan_cycle_work(cycle_work_start);
 
     let scan_cycle_result = match scan_result {
         Ok(result) => result,
@@ -2663,7 +2713,7 @@ async fn run_data_scanner_cycle(
             {
                 save_background_heal_info(storeapi.clone(), new_heal_info).await;
             }
-            mark_scan_cycle_idle(cycle_info).await;
+            mark_scan_cycle_idle(cycle_info, &mut cycle_metrics_guard).await;
             return ScannerCycleOutcome::Failed;
         }
     };
@@ -2678,7 +2728,7 @@ async fn run_data_scanner_cycle(
             "Scanner cycle stopped before committing cycle state"
         );
         emit_scan_cycle_complete(false, cycle_start.elapsed());
-        mark_scan_cycle_idle(cycle_info).await;
+        mark_scan_cycle_idle(cycle_info, &mut cycle_metrics_guard).await;
         return ScannerCycleOutcome::Failed;
     }
     if let Some(required_cycle) = scan_cycle_result.required_cycle_floor() {
@@ -2700,6 +2750,7 @@ async fn run_data_scanner_cycle(
             cycle_revision,
             leader_epoch,
             required_cycle,
+            &mut cycle_metrics_guard,
         )
         .await
         {
@@ -2719,7 +2770,7 @@ async fn run_data_scanner_cycle(
             "Scanner cycle completed without a durable data usage snapshot"
         );
         emit_scan_cycle_complete(false, cycle_start.elapsed());
-        mark_scan_cycle_idle(cycle_info).await;
+        mark_scan_cycle_idle(cycle_info, &mut cycle_metrics_guard).await;
         return ScannerCycleOutcome::Failed;
     }
     if budget_elapsed {
@@ -2743,7 +2794,16 @@ async fn run_data_scanner_cycle(
             scan_cycle_partial_reason(budget_reason),
             scan_cycle_partial_source(budget_reason),
         );
-        return if finalize_partial_scan_cycle(ctx, storeapi.clone(), cycle_info, cycle_revision, leader_epoch).await {
+        return if finalize_partial_scan_cycle(
+            ctx,
+            storeapi.clone(),
+            cycle_info,
+            cycle_revision,
+            leader_epoch,
+            &mut cycle_metrics_guard,
+        )
+        .await
+        {
             ScannerCycleOutcome::Partial
         } else {
             ScannerCycleOutcome::Failed
@@ -2792,7 +2852,7 @@ async fn run_data_scanner_cycle(
                 "Scanner cycle completed without a durable data usage snapshot"
             );
             emit_scan_cycle_complete(false, cycle_start.elapsed());
-            mark_scan_cycle_idle(cycle_info).await;
+            mark_scan_cycle_idle(cycle_info, &mut cycle_metrics_guard).await;
             return ScannerCycleOutcome::Failed;
         }
         ScannerCycleOutcome::Partial => {
@@ -2818,7 +2878,16 @@ async fn run_data_scanner_cycle(
                 );
             }
             emit_scan_cycle_partial_with_source(cycle_start.elapsed(), ScanCyclePartialReason::Unknown, None);
-            return if finalize_partial_scan_cycle(ctx, storeapi.clone(), cycle_info, cycle_revision, leader_epoch).await {
+            return if finalize_partial_scan_cycle(
+                ctx,
+                storeapi.clone(),
+                cycle_info,
+                cycle_revision,
+                leader_epoch,
+                &mut cycle_metrics_guard,
+            )
+            .await
+            {
                 ScannerCycleOutcome::Partial
             } else {
                 ScannerCycleOutcome::Failed
@@ -2834,7 +2903,16 @@ async fn run_data_scanner_cycle(
                 state = "superseded",
                 "Scanner cycle usage snapshot was superseded by concurrent namespace activity"
             );
-            if finalize_partial_scan_cycle(ctx, storeapi.clone(), cycle_info, cycle_revision, leader_epoch).await {
+            if finalize_partial_scan_cycle(
+                ctx,
+                storeapi.clone(),
+                cycle_info,
+                cycle_revision,
+                leader_epoch,
+                &mut cycle_metrics_guard,
+            )
+            .await
+            {
                 emit_scan_cycle_superseded(cycle_start.elapsed());
                 return ScannerCycleOutcome::Superseded;
             }
@@ -2853,7 +2931,7 @@ async fn run_data_scanner_cycle(
             error = %err,
             "Scanner completed cycle could not advance"
         );
-        mark_scan_cycle_idle(cycle_info).await;
+        mark_scan_cycle_idle(cycle_info, &mut cycle_metrics_guard).await;
         emit_scan_cycle_complete(false, cycle_start.elapsed());
         return ScannerCycleOutcome::Failed;
     }
@@ -2862,9 +2940,8 @@ async fn run_data_scanner_cycle(
     global_metrics().clear_current_scan_mode();
 
     retain_recent_cycle_completions(&mut cycle_info.cycle_completed);
-    global_metrics().set_cycle(Some(cycle_info.clone())).await;
     if !persist_scanner_cycle_state(ctx, storeapi.clone(), cycle_info, cycle_revision, leader_epoch).await {
-        mark_scan_cycle_idle(cycle_info).await;
+        cycle_metrics_guard.finish(cycle_info.clone()).await;
         emit_scan_cycle_complete(false, cycle_start.elapsed());
         return ScannerCycleOutcome::Failed;
     }
@@ -2888,9 +2965,37 @@ async fn run_data_scanner_cycle(
         "Scanner cycle completed"
     );
 
+    cycle_metrics_guard.finish(cycle_info.clone()).await;
     scanner_cycle_outcome_with_pending_maintenance(ScannerCycleOutcome::Completed, pending_maintenance_work)
 }
 
+struct ScannerCycleMetricsGuard {
+    start: Option<ScanCycleWorkSnapshot>,
+}
+
+impl ScannerCycleMetricsGuard {
+    async fn new(cycle: CurrentCycle) -> Self {
+        Self {
+            start: Some(global_metrics().start_scan_cycle_work_with_cycle(cycle).await),
+        }
+    }
+
+    async fn finish(&mut self, cycle: CurrentCycle) {
+        if let Some(start) = self.start {
+            global_metrics().finish_scan_cycle_work_with_cycle(start, cycle).await;
+            self.start = None;
+        }
+    }
+}
+
+impl Drop for ScannerCycleMetricsGuard {
+    fn drop(&mut self) {
+        if let Some(start) = self.start.take() {
+            global_metrics().finish_scan_cycle_work(start);
+        }
+    }
+}
+
 async fn record_scanner_leader_lock_lost(message: &'static str) {
     reset_scanner_cycle_schedule();
     record_scanner_leader_lock_state("lost");
@@ -3443,7 +3548,7 @@ enum DataUsagePersistTaskResult {
 
 async fn wait_for_data_usage_persist_task(
     ctx: &CancellationToken,
-    task: &mut tokio::task::JoinHandle<DataUsagePersistOutcome>,
+    task: &mut AbortOnDropHandle<DataUsagePersistOutcome>,
     timeout: Duration,
 ) -> DataUsagePersistTaskResult {
     tokio::select! {
@@ -3840,8 +3945,9 @@ mod tests {
     use super::*;
     use crate::EcstoreResult;
     use crate::{
-        ScannerGetObjectReader as GetObjectReader, ScannerObjectInfo as ObjectInfo, ScannerObjectOptions as ObjectOptions,
-        ScannerPutObjReader as PutObjReader,
+        Endpoint, EndpointServerPools, Endpoints, InstanceContext, PoolEndpoints, ScannerGetObjectReader as GetObjectReader,
+        ScannerObjectInfo as ObjectInfo, ScannerObjectOptions as ObjectOptions, ScannerPutObjReader as PutObjReader,
+        init_bucket_metadata_sys_for_scanner_tests, init_ecstore_config_for_scanner_tests, init_local_disks_with_instance_ctx,
     };
     use serial_test::serial;
     use std::collections::HashMap;
@@ -3853,6 +3959,47 @@ mod tests {
 
     const TEST_DEFAULT_SCANNER_CYCLE_SECS: u64 = 24 * 60 * 60;
 
+    async fn setup_scanner_cycle_store() -> (tempfile::TempDir, Arc<ECStore>) {
+        init_ecstore_config_for_scanner_tests();
+        let temp_dir = tempfile::tempdir().expect("scanner cycle test directory should be created");
+        let mut endpoints = Vec::new();
+        for disk_index in 0..4 {
+            let disk_path = temp_dir.path().join(format!("disk{disk_index}"));
+            tokio::fs::create_dir_all(&disk_path)
+                .await
+                .expect("scanner cycle test disk should be created");
+            let mut endpoint =
+                Endpoint::try_from(disk_path.to_str().expect("disk path should be utf8")).expect("endpoint should parse");
+            endpoint.set_pool_index(0);
+            endpoint.set_set_index(0);
+            endpoint.set_disk_index(disk_index);
+            endpoints.push(endpoint);
+        }
+        let endpoint_pools = EndpointServerPools::from(vec![PoolEndpoints {
+            legacy: false,
+            set_count: 1,
+            drives_per_set: 4,
+            endpoints: Endpoints::from(endpoints),
+            cmd_line: "scanner-cycle-metrics".to_string(),
+            platform: format!("OS: {} | Arch: {}", std::env::consts::OS, std::env::consts::ARCH),
+        }]);
+        let instance_ctx = Arc::new(InstanceContext::new());
+        init_local_disks_with_instance_ctx(&instance_ctx, endpoint_pools.clone())
+            .await
+            .expect("scanner cycle test disks should initialize");
+        let store = ECStore::new_with_instance_ctx(
+            "127.0.0.1:0".parse().expect("test address should parse"),
+            endpoint_pools,
+            CancellationToken::new(),
+            instance_ctx,
+        )
+        .await
+        .expect("scanner cycle test ECStore should initialize");
+        init_bucket_metadata_sys_for_scanner_tests(store.clone()).await;
+
+        (temp_dir, store)
+    }
+
     fn assert_run_data_scanner_signature<F, Fut>(_run: F)
     where
         F: Fn(CancellationToken, Arc<ECStore>) -> Fut,
@@ -4312,9 +4459,9 @@ mod tests {
         };
 
         global_metrics().set_current_scan_mode(HealScanMode::Deep);
-        global_metrics().set_cycle(Some(cycle_info.clone())).await;
+        let mut cycle_metrics_guard = ScannerCycleMetricsGuard::new(cycle_info.clone()).await;
 
-        mark_scan_cycle_idle(&mut cycle_info).await;
+        mark_scan_cycle_idle(&mut cycle_info, &mut cycle_metrics_guard).await;
 
         let published = global_metrics()
             .get_cycle()
@@ -4330,6 +4477,123 @@ mod tests {
         global_metrics().set_cycle(None).await;
     }
 
+    #[tokio::test]
+    #[serial]
+    async fn scanner_cycle_metrics_guard_covers_published_first_cycle_lifetime() {
+        let cycle_started = Utc::now() - chrono::Duration::seconds(5);
+        let mut cycle_info = CurrentCycle {
+            current: 0,
+            next: 1,
+            started: cycle_started,
+            ..Default::default()
+        };
+        let mut guard = ScannerCycleMetricsGuard::new(cycle_info.clone()).await;
+        let setup_report = global_metrics().report().await;
+        assert!(setup_report.current_cycle_active);
+        assert_eq!(setup_report.current_cycle, 0);
+        assert_eq!(setup_report.current_started, cycle_started);
+
+        mark_scan_cycle_idle(&mut cycle_info, &mut guard).await;
+        let idle_report = global_metrics().report().await;
+        assert!(!idle_report.current_cycle_active);
+
+        global_metrics().set_cycle(None).await;
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn scanner_cycle_metrics_guard_keeps_active_cycle_published_during_finalization() {
+        let mut cycle_info = CurrentCycle {
+            current: 12,
+            next: 13,
+            started: Utc::now(),
+            ..Default::default()
+        };
+        let mut guard = ScannerCycleMetricsGuard::new(cycle_info.clone()).await;
+
+        cycle_info.current = 0;
+        tokio::task::yield_now().await;
+        let finalizing_report = global_metrics().report().await;
+        assert!(finalizing_report.current_cycle_active);
+        assert_eq!(finalizing_report.current_cycle, 12);
+
+        guard.finish(cycle_info).await;
+        let idle_report = global_metrics().report().await;
+        assert!(!idle_report.current_cycle_active);
+        assert_eq!(idle_report.current_cycle, 0);
+
+        global_metrics().set_cycle(None).await;
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn scanner_cycle_metrics_guard_drop_clears_activity() {
+        let guard = ScannerCycleMetricsGuard::new(CurrentCycle {
+            current: 12,
+            next: 13,
+            started: Utc::now(),
+            ..Default::default()
+        })
+        .await;
+        assert!(global_metrics().report().await.current_cycle_active);
+
+        drop(guard);
+
+        assert!(!global_metrics().report().await.current_cycle_active);
+        global_metrics().set_cycle(None).await;
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn run_data_scanner_cycle_publishes_activity_for_owner_lifetime() {
+        let (_temp_dir, store) = setup_scanner_cycle_store().await;
+        let ctx = CancellationToken::new();
+        let mut cycle_info = CurrentCycle::default();
+        let mut revision = DataUsageCacheRevision::Missing;
+        let leader_epoch = u64::MAX - 1;
+        let state_persist_reached = Arc::new(Notify::new());
+        let _state_persist_hook = set_scanner_cycle_state_persist_test_hook(leader_epoch, state_persist_reached.clone());
+        let state_lock = store
+            .new_ns_lock(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_NAME_PATH.as_str())
+            .await
+            .expect("scanner cycle state lock should be created");
+        let state_guard = state_lock
+            .get_write_lock(Duration::from_secs(1))
+            .await
+            .expect("scanner cycle state lock should be acquired");
+        let mut cycle = Box::pin(run_data_scanner_cycle(&ctx, &store, &mut cycle_info, &mut revision, leader_epoch));
+        let waker = std::task::Waker::noop();
+        let mut context = std::task::Context::from_waker(waker);
+
+        assert!(cycle.as_mut().poll(&mut context).is_pending());
+        let active = global_metrics().report().await;
+        assert!(active.current_cycle_active);
+        assert_eq!(active.current_cycle, 0);
+
+        tokio::time::timeout(Duration::from_secs(30), async {
+            tokio::select! {
+                outcome = &mut cycle => panic!("scanner cycle finished before state persistence was released: {outcome:?}"),
+                _ = state_persist_reached.notified() => {}
+            }
+        })
+        .await
+        .expect("scanner cycle should reach state persistence");
+        let finalizing = global_metrics().report().await;
+        assert!(finalizing.current_cycle_active);
+
+        drop(state_guard);
+        let outcome = tokio::time::timeout(Duration::from_secs(30), cycle)
+            .await
+            .expect("scanner cycle should finish");
+        assert!(matches!(
+            outcome,
+            ScannerCycleOutcome::Completed | ScannerCycleOutcome::CompletedWithPendingMaintenance
+        ));
+        assert!(!global_metrics().report().await.current_cycle_active);
+
+        global_metrics().set_cycle(None).await;
+    }
+
     #[tokio::test]
     #[serial]
     async fn test_finalize_partial_scan_cycle_advances_and_persists_counter() {
@@ -4342,8 +4606,11 @@ mod tests {
             cycle_completed: vec![],
             started: Utc::now(),
         };
+        let mut cycle_metrics_guard = ScannerCycleMetricsGuard::new(cycle_info.clone()).await;
 
-        assert!(finalize_partial_scan_cycle(&ctx, store.clone(), &mut cycle_info, &mut revision, 1).await);
+        assert!(
+            finalize_partial_scan_cycle(&ctx, store.clone(), &mut cycle_info, &mut revision, 1, &mut cycle_metrics_guard,).await
+        );
 
         assert_eq!(cycle_info.next, 13);
         assert_eq!(cycle_info.current, 0);
@@ -4377,8 +4644,20 @@ mod tests {
             cycle_completed: vec![],
             started: Utc::now(),
         };
+        let mut cycle_metrics_guard = ScannerCycleMetricsGuard::new(cycle_info.clone()).await;
 
-        assert!(persist_required_scanner_cycle_floor(&ctx, store.clone(), &mut cycle_info, &mut revision, 7, 19).await);
+        assert!(
+            persist_required_scanner_cycle_floor(
+                &ctx,
+                store.clone(),
+                &mut cycle_info,
+                &mut revision,
+                7,
+                19,
+                &mut cycle_metrics_guard,
+            )
+            .await
+        );
         assert_eq!(cycle_info.current, 0);
         assert_eq!(cycle_info.next, 19);
 
@@ -4405,22 +4684,37 @@ mod tests {
             cycle_completed: vec![],
             started: Utc::now(),
         };
+        let mut cycle_metrics_guard = ScannerCycleMetricsGuard::new(cycle_info.clone()).await;
 
-        assert!(!persist_required_scanner_cycle_floor(&ctx, store.clone(), &mut cycle_info, &mut revision, 7, 12).await);
-        assert_eq!(cycle_info.next, 12);
-        assert_eq!(revision, DataUsageCacheRevision::Missing);
         assert!(
             !persist_required_scanner_cycle_floor(
                 &ctx,
                 store.clone(),
-                &mut CurrentCycle {
-                    current: 12,
-                    next: 12,
-                    ..Default::default()
-                },
+                &mut cycle_info,
+                &mut revision,
+                7,
+                12,
+                &mut cycle_metrics_guard,
+            )
+            .await
+        );
+        assert_eq!(cycle_info.next, 12);
+        assert_eq!(revision, DataUsageCacheRevision::Missing);
+        let mut max_cycle_info = CurrentCycle {
+            current: 12,
+            next: 12,
+            ..Default::default()
+        };
+        let mut max_cycle_metrics_guard = ScannerCycleMetricsGuard::new(max_cycle_info.clone()).await;
+        assert!(
+            !persist_required_scanner_cycle_floor(
+                &ctx,
+                store.clone(),
+                &mut max_cycle_info,
                 &mut revision,
                 7,
                 u64::MAX,
+                &mut max_cycle_metrics_guard,
             )
             .await
         );
@@ -4685,8 +4979,9 @@ mod tests {
             cycle_completed: vec![],
             started: Utc::now(),
         };
+        let mut cycle_metrics_guard = ScannerCycleMetricsGuard::new(cycle_info.clone()).await;
 
-        assert!(!finalize_partial_scan_cycle(&ctx, store, &mut cycle_info, &mut revision, 1).await);
+        assert!(!finalize_partial_scan_cycle(&ctx, store, &mut cycle_info, &mut revision, 1, &mut cycle_metrics_guard,).await);
         assert_eq!(cycle_info.next, 13);
         assert_eq!(cycle_info.current, 0);
         assert_eq!(revision, DataUsageCacheRevision::Missing);
@@ -5943,10 +6238,10 @@ mod tests {
     #[tokio::test]
     async fn data_usage_persist_wait_aborts_when_scanner_is_cancelled() {
         let ctx = CancellationToken::new();
-        let mut task = tokio::spawn(async {
+        let mut task = AbortOnDropHandle::new(tokio::spawn(async {
             std::future::pending::<()>().await;
             DataUsagePersistOutcome::Saved
-        });
+        }));
         ctx.cancel();
 
         let result = wait_for_data_usage_persist_task(&ctx, &mut task, Duration::from_secs(60)).await;
@@ -5958,10 +6253,10 @@ mod tests {
     #[tokio::test(start_paused = true)]
     async fn data_usage_persist_wait_aborts_after_timeout() {
         let ctx = CancellationToken::new();
-        let mut task = tokio::spawn(async {
+        let mut task = AbortOnDropHandle::new(tokio::spawn(async {
             std::future::pending::<()>().await;
             DataUsagePersistOutcome::Saved
-        });
+        }));
 
         let result = wait_for_data_usage_persist_task(&ctx, &mut task, Duration::from_secs(30)).await;
 

crates/utils/Cargo.toml

@@ -58,11 +58,16 @@ url = { workspace = true, optional = true }
 zstd = { workspace = true, optional = true }
 
 [dev-dependencies]
+criterion = { workspace = true, features = ["html_reports"] }
 tempfile = { workspace = true }
 tokio = { workspace = true, features = ["macros", "rt"] }
 temp-env = { workspace = true }
 proptest = "1"
 
+[[bench]]
+name = "hash_hotpath_benchmark"
+harness = false
+
 [target.'cfg(windows)'.dependencies]
 windows = { workspace = true, optional = true, features = ["Win32_Storage_FileSystem"] }
 

crates/utils/benches/hash_hotpath_benchmark.rs

@@ -0,0 +1,55 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main};
+use rustfs_utils::HashAlgorithm;
+use std::hint::black_box;
+
+fn generate_payload(size: usize) -> Vec<u8> {
+    (0..size)
+        .map(|i| u8::try_from(i % 251).expect("modulo output fits in u8"))
+        .collect()
+}
+
+fn bench_hash_hotpaths(c: &mut Criterion) {
+    let payloads = [
+        ("64KiB", generate_payload(64 * 1024)),
+        ("1MiB", generate_payload(1024 * 1024)),
+    ];
+    let algorithms = [
+        ("md5", HashAlgorithm::Md5),
+        ("sha256", HashAlgorithm::SHA256),
+        ("highwayhash256s", HashAlgorithm::HighwayHash256S),
+        ("highwayhash256s_legacy", HashAlgorithm::HighwayHash256SLegacy),
+    ];
+
+    let mut group = c.benchmark_group("hash_hotpath");
+    for (payload_name, payload) in &payloads {
+        let payload_len = u64::try_from(payload.len()).expect("benchmark payload length fits in u64");
+        group.throughput(Throughput::Bytes(payload_len));
+        for (algo_name, algorithm) in &algorithms {
+            let algorithm = algorithm.clone();
+            group.bench_with_input(BenchmarkId::new(*algo_name, payload_name), payload.as_slice(), move |b, payload| {
+                b.iter(|| {
+                    let hash = algorithm.hash_encode(black_box(payload));
+                    black_box(hash.as_ref()[0]);
+                });
+            });
+        }
+    }
+    group.finish();
+}
+
+criterion_group!(benches, bench_hash_hotpaths);
+criterion_main!(benches);

crates/utils/src/hash.rs

@@ -29,14 +29,24 @@ const LEGACY_HIGHWAY_HASH256_KEY: [u8; 32] = [
     3, 0, 0, 0, 0, 0, 0, 0, 4, 0, 0, 0, 0, 0, 0, 0, 2, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0,
 ];
 
-fn highway_key_from_bytes(bytes: &[u8; 32]) -> [u64; 4] {
-    let mut key = [0u64; 4];
-    for (i, chunk) in bytes.chunks_exact(8).enumerate() {
-        key[i] = u64::from_le_bytes(chunk.try_into().unwrap());
-    }
-    key
+const fn highway_key_from_bytes(bytes: &[u8; 32]) -> [u64; 4] {
+    [
+        u64::from_le_bytes([bytes[0], bytes[1], bytes[2], bytes[3], bytes[4], bytes[5], bytes[6], bytes[7]]),
+        u64::from_le_bytes([
+            bytes[8], bytes[9], bytes[10], bytes[11], bytes[12], bytes[13], bytes[14], bytes[15],
+        ]),
+        u64::from_le_bytes([
+            bytes[16], bytes[17], bytes[18], bytes[19], bytes[20], bytes[21], bytes[22], bytes[23],
+        ]),
+        u64::from_le_bytes([
+            bytes[24], bytes[25], bytes[26], bytes[27], bytes[28], bytes[29], bytes[30], bytes[31],
+        ]),
+    ]
 }
 
+const MAGIC_HIGHWAY_HASH256_PARSED_KEY: Key = Key(highway_key_from_bytes(&MAGIC_HIGHWAY_HASH256_KEY));
+const LEGACY_HIGHWAY_HASH256_PARSED_KEY: Key = Key(highway_key_from_bytes(&LEGACY_HIGHWAY_HASH256_KEY));
+
 #[derive(Serialize, Deserialize, Debug, PartialEq, Default, Clone, Eq, Hash)]
 /// Supported hash algorithms for bitrot protection.
 pub enum HashAlgorithm {
@@ -100,25 +110,23 @@ impl HashAlgorithm {
     /// # Returns
     /// A byte slice containing the hash of the input data
     ///
+    #[inline]
     pub fn hash_encode(&self, data: &[u8]) -> impl AsRef<[u8]> {
         match self {
             HashAlgorithm::Md5 => HashEncoded::Md5(Md5::digest(data).into()),
             HashAlgorithm::HighwayHash256 => {
-                let key = Key(highway_key_from_bytes(&MAGIC_HIGHWAY_HASH256_KEY));
-                let mut hasher = HighwayHasher::new(key);
+                let mut hasher = HighwayHasher::new(MAGIC_HIGHWAY_HASH256_PARSED_KEY);
                 hasher.append(data);
                 HashEncoded::HighwayHash256(u8x32_from_u64x4(hasher.finalize256()))
             }
             HashAlgorithm::SHA256 => HashEncoded::Sha256(Sha256::digest(data).into()),
             HashAlgorithm::HighwayHash256S => {
-                let key = Key(highway_key_from_bytes(&MAGIC_HIGHWAY_HASH256_KEY));
-                let mut hasher = HighwayHasher::new(key);
+                let mut hasher = HighwayHasher::new(MAGIC_HIGHWAY_HASH256_PARSED_KEY);
                 hasher.append(data);
                 HashEncoded::HighwayHash256S(u8x32_from_u64x4(hasher.finalize256()))
             }
             HashAlgorithm::HighwayHash256SLegacy => {
-                let key = Key(highway_key_from_bytes(&LEGACY_HIGHWAY_HASH256_KEY));
-                let mut hasher = HighwayHasher::new(key);
+                let mut hasher = HighwayHasher::new(LEGACY_HIGHWAY_HASH256_PARSED_KEY);
                 hasher.append(data);
                 HashEncoded::HighwayHash256SLegacy(u8x32_from_u64x4(hasher.finalize256()))
             }

crates/utils/src/http/header_compat.rs

@@ -25,11 +25,6 @@ const RUSTFS_PREFIX: &str = "x-rustfs-";
 const MINIO_PREFIX: &str = "x-minio-";
 const MINIO_ENCRYPTION_PREFIX: &str = "x-minio-encryption-";
 const RUSTFS_ENCRYPTION_PREFIX: &str = "x-rustfs-encryption-";
-const MINIO_INTERNAL_ENCRYPTION_PREFIX: &str = "x-minio-internal-server-side-encryption-";
-const MINIO_INTERNAL_ENCRYPTED_MULTIPART: &str = "x-minio-internal-encrypted-multipart";
-const RUSTFS_ENCRYPTION_ORIGINAL_SIZE: &str = "x-rustfs-encryption-original-size";
-const MINIO_ENCRYPTION_ORIGINAL_SIZE: &str = "x-minio-encryption-original-size";
-const SSEC_ORIGINAL_SIZE: &str = "x-amz-server-side-encryption-customer-original-size";
 
 // Suffix constants (part after x-rustfs- or x-minio-). Use with get_header/insert_header.
 pub const SUFFIX_FORCE_DELETE: &str = "force-delete";
@@ -45,49 +40,11 @@ pub const SUFFIX_SOURCE_REPLICATION_REQUEST: &str = "source-replication-request"
 pub const SUFFIX_SOURCE_REPLICATION_CHECK: &str = "source-replication-check";
 pub const SUFFIX_REPLICATION_SSEC_CRC: &str = "replication-ssec-crc";
 
-/// Returns true if the key is object-encryption metadata understood by RustFS or MinIO.
-/// Case-insensitive for metadata filtering.
+/// Returns true if the key is an internal encryption metadata key (x-rustfs-encryption-* or
+/// x-minio-encryption-*). Case-insensitive for metadata filtering.
 pub fn is_encryption_metadata_key(key: &str) -> bool {
     let lower = key.to_lowercase();
-    lower.starts_with(RUSTFS_ENCRYPTION_PREFIX)
-        || lower.starts_with(MINIO_ENCRYPTION_PREFIX)
-        || lower.starts_with(MINIO_INTERNAL_ENCRYPTION_PREFIX)
-        || lower == MINIO_INTERNAL_ENCRYPTED_MULTIPART
-}
-
-/// Returns true when a metadata key proves that object data is encrypted.
-///
-/// Original-size metadata alone is not proof: older plaintext objects can
-/// retain that compatibility field after metadata migration.
-pub fn is_object_encryption_marker(key: &str) -> bool {
-    (is_encryption_metadata_key(key)
-        && !key.eq_ignore_ascii_case(RUSTFS_ENCRYPTION_ORIGINAL_SIZE)
-        && !key.eq_ignore_ascii_case(MINIO_ENCRYPTION_ORIGINAL_SIZE))
-        || super::is_sse_header(key)
-}
-
-/// Reads the logical object size recorded by encryption metadata.
-pub fn get_object_encryption_original_size(metadata: &std::collections::HashMap<String, String>) -> std::io::Result<Option<i64>> {
-    let actual_size = super::get_str(metadata, super::SUFFIX_ACTUAL_SIZE);
-    let size = get_case_insensitive(metadata, RUSTFS_ENCRYPTION_ORIGINAL_SIZE)
-        .or_else(|| get_case_insensitive(metadata, SSEC_ORIGINAL_SIZE))
-        .or(actual_size.as_deref());
-
-    let Some(size) = size.filter(|size| !size.is_empty()) else {
-        return Ok(None);
-    };
-    size.parse::<i64>()
-        .map(Some)
-        .map_err(|error| std::io::Error::other(format!("Failed to parse encryption original size: {error}")))
-}
-
-fn get_case_insensitive<'a>(metadata: &'a std::collections::HashMap<String, String>, key: &str) -> Option<&'a str> {
-    metadata.get(key).map(String::as_str).or_else(|| {
-        metadata
-            .iter()
-            .find(|(candidate, _)| candidate.eq_ignore_ascii_case(key))
-            .map(|(_, value)| value.as_str())
-    })
+    lower.starts_with(RUSTFS_ENCRYPTION_PREFIX) || lower.starts_with(MINIO_ENCRYPTION_PREFIX)
 }
 
 fn rustfs_key(suffix: &str) -> String {
@@ -149,37 +106,10 @@ mod tests {
         assert!(is_encryption_metadata_key("x-rustfs-encryption-iv"));
         assert!(is_encryption_metadata_key("X-Rustfs-Encryption-Key"));
         assert!(is_encryption_metadata_key("x-minio-encryption-iv"));
-        assert!(is_encryption_metadata_key("X-Minio-Internal-Server-Side-Encryption-Sealed-Key"));
-        assert!(is_encryption_metadata_key("X-Minio-Internal-Encrypted-Multipart"));
         assert!(!is_encryption_metadata_key("x-amz-meta-custom"));
         assert!(!is_encryption_metadata_key("x-rustfs-internal-healing"));
     }
 
-    #[test]
-    fn object_encryption_marker_excludes_size_only_metadata() {
-        assert!(!is_object_encryption_marker(RUSTFS_ENCRYPTION_ORIGINAL_SIZE));
-        assert!(is_object_encryption_marker("X-Minio-Internal-Server-Side-Encryption-Sealed-Key"));
-        assert!(is_object_encryption_marker("x-amz-server-side-encryption"));
-    }
-
-    #[test]
-    fn object_encryption_original_size_is_case_insensitive() {
-        let metadata = std::collections::HashMap::from([(
-            "X-Amz-Server-Side-Encryption-Customer-Original-Size".to_string(),
-            "42".to_string(),
-        )]);
-        assert_eq!(get_object_encryption_original_size(&metadata).expect("valid size"), Some(42));
-    }
-
-    #[test]
-    fn object_encryption_original_size_prefers_rustfs_metadata() {
-        let metadata = std::collections::HashMap::from([
-            (SSEC_ORIGINAL_SIZE.to_string(), "21".to_string()),
-            (RUSTFS_ENCRYPTION_ORIGINAL_SIZE.to_string(), "42".to_string()),
-        ]);
-        assert_eq!(get_object_encryption_original_size(&metadata).expect("valid size"), Some(42));
-    }
-
     #[test]
     fn test_get_header() {
         let mut headers = HeaderMap::new();

docs/testing/ecstore-validation-suite-design.md

@@ -358,7 +358,7 @@ Fixture-backed tests should run when the fixture path is present:
 
 ```bash
 cargo test -p rustfs-ecstore --test legacy_bitrot_read_test -- --nocapture
-cargo test -p rustfs --features rio-v2 storage::minio_generated_read_test --lib -- --ignored --nocapture
+cargo test -p rustfs-ecstore --features rio-v2 --test minio_generated_read_test -- --ignored --nocapture
 ```
 
 ## Multi-Expert Adversarial Review Summary

rustfs/src/admin/handlers/kms_dynamic.rs

@@ -393,27 +393,36 @@ impl Operation for ConfigureKmsHandler {
         // Convert request to KmsConfig
         let kms_config = configure_request.to_kms_config();
 
-        let persisted_config = kms_config.clone();
-        let (success, message, status) = match service_manager
-            .configure_with_persistence(kms_config, || async move {
-                save_kms_config(&persisted_config)
-                    .await
-                    .map_err(|error| rustfs_kms::KmsError::backend_error(format!("Failed to persist KMS configuration: {error}")))
-            })
-            .await
-        {
+        // Configure the service
+        let (success, message, status) = match service_manager.configure(kms_config.clone()).await {
             Ok(()) => {
-                let status = service_manager.get_status().await;
-                info!(
-                    component = LOG_COMPONENT_ADMIN,
-                    subsystem = LOG_SUBSYSTEM_KMS,
-                    event = "kms_service_state",
-                    operation = "configure",
-                    state = "configured",
-                    status = ?status,
-                    "admin kms dynamic state"
-                );
-                (true, "KMS configured successfully".to_string(), status)
+                // Persist the configuration to cluster storage
+                if let Err(e) = save_kms_config(&kms_config).await {
+                    let error_msg = format!("KMS configured in memory but failed to persist: {e}");
+                    error!(
+                        component = LOG_COMPONENT_ADMIN,
+                        subsystem = LOG_SUBSYSTEM_KMS,
+                        event = "kms_service_state",
+                        operation = "configure",
+                        state = "persist_failed",
+                        error = %e,
+                        "admin kms dynamic state"
+                    );
+                    let status = service_manager.get_status().await;
+                    (false, error_msg, status)
+                } else {
+                    let status = service_manager.get_status().await;
+                    info!(
+                        component = LOG_COMPONENT_ADMIN,
+                        subsystem = LOG_SUBSYSTEM_KMS,
+                        event = "kms_service_state",
+                        operation = "configure",
+                        state = "configured",
+                        status = ?status,
+                        "admin kms dynamic state"
+                    );
+                    (true, "KMS configured successfully".to_string(), status)
+                }
             }
             Err(e) => {
                 let error_msg = format!("Failed to configure KMS: {e}");
@@ -520,61 +529,125 @@ impl Operation for StartKmsHandler {
         );
 
         let service_manager = kms_service_manager_from_context();
-        let force = start_request.force.unwrap_or(false);
-        let (success, message, status) = match service_manager.start_or_restart(force).await {
-            Ok(rustfs_kms::KmsStartOutcome::Started) => {
-                let status = service_manager.get_status().await;
-                info!(
-                    component = LOG_COMPONENT_ADMIN,
-                    subsystem = LOG_SUBSYSTEM_KMS,
-                    event = "kms_service_state",
-                    operation = "start",
-                    state = "running",
-                    status = ?status,
-                    "admin kms dynamic state"
-                );
-                (true, "KMS service started successfully".to_string(), status)
-            }
-            Ok(rustfs_kms::KmsStartOutcome::Restarted) => {
-                let status = service_manager.get_status().await;
-                info!(
-                    component = LOG_COMPONENT_ADMIN,
-                    subsystem = LOG_SUBSYSTEM_KMS,
-                    event = "kms_service_state",
-                    operation = "restart",
-                    state = "running",
-                    status = ?status,
-                    "admin kms dynamic state"
-                );
-                (true, "KMS service restarted successfully".to_string(), status)
-            }
-            Ok(rustfs_kms::KmsStartOutcome::AlreadyRunning) => {
-                let status = service_manager.get_status().await;
-                warn!(
-                    component = LOG_COMPONENT_ADMIN,
-                    subsystem = LOG_SUBSYSTEM_KMS,
-                    event = "kms_service_state",
-                    operation = "start",
-                    state = "already_running",
-                    "admin kms dynamic state"
-                );
-                (false, "KMS service is already running. Use force=true to restart.".to_string(), status)
-            }
-            Err(e) => {
-                let error_msg = format!("Failed to start or restart KMS service: {e}");
-                error!(
-                    component = LOG_COMPONENT_ADMIN,
-                    subsystem = LOG_SUBSYSTEM_KMS,
-                    event = "kms_service_state",
-                    operation = "start",
-                    state = "start_failed",
-                    error = %e,
-                    "admin kms dynamic state"
-                );
-                let status = service_manager.get_status().await;
-                (false, error_msg, status)
-            }
-        };
+
+        // Check if already running and force flag
+        let current_status = service_manager.get_status().await;
+        if matches!(current_status, KmsServiceStatus::Running) && !start_request.force.unwrap_or(false) {
+            warn!(
+                component = LOG_COMPONENT_ADMIN,
+                subsystem = LOG_SUBSYSTEM_KMS,
+                event = "kms_service_state",
+                operation = "start",
+                state = "already_running",
+                "admin kms dynamic state"
+            );
+            let response = StartKmsResponse {
+                success: false,
+                message: "KMS service is already running. Use force=true to restart.".to_string(),
+                status: current_status,
+            };
+            let json_response = match serde_json::to_string(&response) {
+                Ok(json) => json,
+                Err(e) => {
+                    error!(
+                        component = LOG_COMPONENT_ADMIN,
+                        subsystem = LOG_SUBSYSTEM_KMS,
+                        event = EVENT_ADMIN_KMS_DYNAMIC_STATE,
+                        operation = "start",
+                        result = "response_serialize_failed",
+                        error = %e,
+                        "admin kms dynamic state"
+                    );
+                    return Ok(S3Response::new((
+                        StatusCode::INTERNAL_SERVER_ERROR,
+                        Body::from("Serialization error".to_string()),
+                    )));
+                }
+            };
+            return Ok(S3Response::new((StatusCode::OK, Body::from(json_response))));
+        }
+
+        // Start the service (or restart if force=true)
+        let (success, message, status) =
+            if start_request.force.unwrap_or(false) && matches!(current_status, KmsServiceStatus::Running) {
+                // Force restart
+                match service_manager.stop().await {
+                    Ok(()) => match service_manager.start().await {
+                        Ok(()) => {
+                            let status = service_manager.get_status().await;
+                            info!(
+                                component = LOG_COMPONENT_ADMIN,
+                                subsystem = LOG_SUBSYSTEM_KMS,
+                                event = "kms_service_state",
+                                operation = "restart",
+                                state = "running",
+                                status = ?status,
+                                "admin kms dynamic state"
+                            );
+                            (true, "KMS service restarted successfully".to_string(), status)
+                        }
+                        Err(e) => {
+                            let error_msg = format!("Failed to restart KMS service: {e}");
+                            error!(
+                                component = LOG_COMPONENT_ADMIN,
+                                subsystem = LOG_SUBSYSTEM_KMS,
+                                event = "kms_service_state",
+                                operation = "restart",
+                                state = "start_failed",
+                                error = %e,
+                                "admin kms dynamic state"
+                            );
+                            let status = service_manager.get_status().await;
+                            (false, error_msg, status)
+                        }
+                    },
+                    Err(e) => {
+                        let error_msg = format!("Failed to stop KMS service for restart: {e}");
+                        error!(
+                            component = LOG_COMPONENT_ADMIN,
+                            subsystem = LOG_SUBSYSTEM_KMS,
+                            event = "kms_service_state",
+                            operation = "restart",
+                            state = "stop_failed",
+                            error = %e,
+                            "admin kms dynamic state"
+                        );
+                        let status = service_manager.get_status().await;
+                        (false, error_msg, status)
+                    }
+                }
+            } else {
+                // Normal start
+                match service_manager.start().await {
+                    Ok(()) => {
+                        let status = service_manager.get_status().await;
+                        info!(
+                            component = LOG_COMPONENT_ADMIN,
+                            subsystem = LOG_SUBSYSTEM_KMS,
+                            event = "kms_service_state",
+                            operation = "start",
+                            state = "running",
+                            status = ?status,
+                            "admin kms dynamic state"
+                        );
+                        (true, "KMS service started successfully".to_string(), status)
+                    }
+                    Err(e) => {
+                        let error_msg = format!("Failed to start KMS service: {e}");
+                        error!(
+                            component = LOG_COMPONENT_ADMIN,
+                            subsystem = LOG_SUBSYSTEM_KMS,
+                            event = "kms_service_state",
+                            operation = "start",
+                            state = "start_failed",
+                            error = %e,
+                            "admin kms dynamic state"
+                        );
+                        let status = service_manager.get_status().await;
+                        (false, error_msg, status)
+                    }
+                }
+            };
 
         let response = StartKmsResponse {
             success,
@@ -731,7 +804,8 @@ impl Operation for GetKmsStatusHandler {
 
         let service_manager = kms_service_manager_from_context();
 
-        let (status, config) = service_manager.get_redacted_state().await;
+        let status = service_manager.get_status().await;
+        let config = service_manager.get_redacted_config().await;
 
         // Get backend type and health status
         let backend_type = config.as_ref().map(|c| c.backend.clone());
@@ -864,27 +938,36 @@ impl Operation for ReconfigureKmsHandler {
         // Convert request to KmsConfig
         let kms_config = configure_request.to_kms_config();
 
-        let persisted_config = kms_config.clone();
-        let (success, message, status) = match service_manager
-            .reconfigure_with_persistence(kms_config, || async move {
-                save_kms_config(&persisted_config)
-                    .await
-                    .map_err(|error| rustfs_kms::KmsError::backend_error(format!("Failed to persist KMS configuration: {error}")))
-            })
-            .await
-        {
+        // Reconfigure the service (stops, reconfigures, and starts)
+        let (success, message, status) = match service_manager.reconfigure(kms_config.clone()).await {
             Ok(()) => {
-                let status = service_manager.get_status().await;
-                info!(
-                    component = LOG_COMPONENT_ADMIN,
-                    subsystem = LOG_SUBSYSTEM_KMS,
-                    event = "kms_service_state",
-                    operation = "reconfigure",
-                    state = "reconfigured",
-                    status = ?status,
-                    "admin kms dynamic state"
-                );
-                (true, "KMS reconfigured and restarted successfully".to_string(), status)
+                // Persist the configuration to cluster storage
+                if let Err(e) = save_kms_config(&kms_config).await {
+                    let error_msg = format!("KMS reconfigured in memory but failed to persist: {e}");
+                    error!(
+                        component = LOG_COMPONENT_ADMIN,
+                        subsystem = LOG_SUBSYSTEM_KMS,
+                        event = "kms_service_state",
+                        operation = "reconfigure",
+                        state = "persist_failed",
+                        error = %e,
+                        "admin kms dynamic state"
+                    );
+                    let status = service_manager.get_status().await;
+                    (false, error_msg, status)
+                } else {
+                    let status = service_manager.get_status().await;
+                    info!(
+                        component = LOG_COMPONENT_ADMIN,
+                        subsystem = LOG_SUBSYSTEM_KMS,
+                        event = "kms_service_state",
+                        operation = "reconfigure",
+                        state = "reconfigured",
+                        status = ?status,
+                        "admin kms dynamic state"
+                    );
+                    (true, "KMS reconfigured and restarted successfully".to_string(), status)
+                }
             }
             Err(e) => {
                 let error_msg = format!("Failed to reconfigure KMS: {e}");

rustfs/src/app/context/global.rs

@@ -18,12 +18,13 @@ use super::handles::{
     IamHandle, KmsHandle, default_action_credential_interface, default_boot_time_interface, default_bucket_metadata_interface,
     default_bucket_monitor_interface, default_buffer_config_interface, default_deployment_id_interface,
     default_endpoints_interface, default_expiry_state_interface, default_federated_identity_interface,
-    default_internode_metrics_interface, default_local_node_name_interface, default_lock_client_interface,
-    default_lock_clients_interface, default_notification_system_interface, default_notify_interface,
-    default_outbound_tls_runtime_interface, default_performance_metrics_interface, default_region_interface,
-    default_replication_pool_interface, default_replication_stats_interface, default_runtime_port_interface,
-    default_s3select_db_interface, default_scanner_metrics_interface, default_server_config_interface,
-    default_storage_class_interface, default_tier_config_interface, default_transition_state_interface,
+    default_internode_metrics_interface, default_kms_runtime_interface, default_local_node_name_interface,
+    default_lock_client_interface, default_lock_clients_interface, default_notification_system_interface,
+    default_notify_interface, default_outbound_tls_runtime_interface, default_performance_metrics_interface,
+    default_region_interface, default_replication_pool_interface, default_replication_stats_interface,
+    default_runtime_port_interface, default_s3select_db_interface, default_scanner_metrics_interface,
+    default_server_config_interface, default_storage_class_interface, default_tier_config_interface,
+    default_transition_state_interface,
 };
 use super::interfaces::{
     ActionCredentialInterface, BootTimeInterface, BucketMetadataInterface, BucketMonitorInterface, BufferConfigInterface,
@@ -79,7 +80,6 @@ pub struct AppContext {
 impl AppContext {
     pub fn new(object_store: Arc<ECStore>, iam: Arc<dyn IamInterface>, kms: Arc<dyn KmsInterface>) -> Self {
         let object_data_cache = ObjectDataCacheAdapter::from_env_or_disabled();
-        let kms_runtime = Arc::new(crate::app::context::handles::KmsRuntimeHandle::new(kms.handle()));
         // Let ecstore probe this cache inside get_object_reader, after
         // metadata resolution but before the erasure data read (backlog#802).
         crate::app::object_data_cache::register_object_data_cache_body_hook(Arc::clone(&object_data_cache));
@@ -94,7 +94,7 @@ impl AppContext {
             iam,
             federated_identity: default_federated_identity_interface(),
             kms,
-            kms_runtime,
+            kms_runtime: default_kms_runtime_interface(),
             outbound_tls_runtime: default_outbound_tls_runtime_interface(),
             notify: default_notify_interface(),
             notification_system: default_notification_system_interface(),

rustfs/src/app/context/handles.rs

@@ -128,19 +128,12 @@ impl KmsInterface for KmsHandle {
 }
 
 /// Default KMS runtime interface adapter.
-pub struct KmsRuntimeHandle {
-    kms: Option<Arc<KmsServiceManager>>,
-}
-
-impl KmsRuntimeHandle {
-    pub fn new(kms: Arc<KmsServiceManager>) -> Self {
-        Self { kms: Some(kms) }
-    }
-}
+#[derive(Default)]
+pub struct KmsRuntimeHandle;
 
 impl KmsRuntimeInterface for KmsRuntimeHandle {
     fn service_manager(&self) -> Option<Arc<KmsServiceManager>> {
-        self.kms.clone()
+        runtime_sources::kms_service_manager()
     }
 }
 
@@ -492,9 +485,7 @@ pub fn default_notification_system_interface() -> Arc<dyn NotificationSystemInte
 }
 
 pub fn default_kms_runtime_interface() -> Arc<dyn KmsRuntimeInterface> {
-    Arc::new(KmsRuntimeHandle {
-        kms: runtime_sources::kms_service_manager(),
-    })
+    Arc::new(KmsRuntimeHandle)
 }
 
 pub fn default_outbound_tls_runtime_interface() -> Arc<dyn OutboundTlsRuntimeInterface> {
@@ -616,10 +607,10 @@ pub fn default_buffer_config_interface() -> Arc<dyn BufferConfigInterface> {
 #[cfg(test)]
 mod tests {
     use super::{
-        KmsRuntimeHandle, KmsServiceManager, ServerConfigHandle, default_federated_identity_interface,
-        federated_identity_interface, publish_default_federated_identity_service, runtime_sources,
+        ServerConfigHandle, default_federated_identity_interface, federated_identity_interface,
+        publish_default_federated_identity_service, runtime_sources,
     };
-    use crate::app::context::interfaces::{KmsRuntimeInterface, ServerConfigInterface};
+    use crate::app::context::interfaces::ServerConfigInterface;
     use rustfs_config::server_config::Config;
     use rustfs_iam::{
         federation::{FederatedIdentityRegistry, FederatedIdentityService, oidc::StandardOidcAdapter},
@@ -742,19 +733,4 @@ mod tests {
             "handle B must serve its own credentials"
         );
     }
-
-    #[test]
-    fn kms_runtime_handles_keep_injected_managers_isolated() {
-        let manager_a = Arc::new(KmsServiceManager::new());
-        let manager_b = Arc::new(KmsServiceManager::new());
-        let handle_a = KmsRuntimeHandle::new(manager_a.clone());
-        let handle_b = KmsRuntimeHandle::new(manager_b.clone());
-
-        assert!(Arc::ptr_eq(&handle_a.service_manager().expect("manager A"), &manager_a));
-        assert!(Arc::ptr_eq(&handle_b.service_manager().expect("manager B"), &manager_b));
-        assert!(!Arc::ptr_eq(
-            &handle_a.service_manager().expect("manager A"),
-            &handle_b.service_manager().expect("manager B")
-        ));
-    }
 }

rustfs/src/storage/mod.rs

@@ -36,8 +36,6 @@ mod ecfs_extend;
 mod ecfs_test;
 pub(crate) mod head_prefix;
 #[cfg(test)]
-mod minio_generated_read_test;
-#[cfg(test)]
 mod multi_factor_scheduler_integration_test;
 pub(crate) mod runtime_sources;
 #[cfg(test)]

rustfs/src/storage/rpc/node_service/bucket.rs

@@ -66,14 +66,14 @@ impl NodeService {
             }));
         }
 
-        let Some(_store) = self.resolve_object_store() else {
+        let Some(store) = self.resolve_object_store() else {
             return Ok(Response::new(LoadBucketMetadataResponse {
                 success: false,
                 error_info: Some("errServerNotInitialized".to_string()),
             }));
         };
 
-        match reload_bucket_metadata(&bucket).await {
+        match reload_bucket_metadata(store, &bucket).await {
             Ok(()) => {
                 if scanner_maintenance_change {
                     rustfs_scanner::record_scanner_maintenance_change(&bucket);

rustfs/src/storage/sse.rs

@@ -70,10 +70,6 @@
 //! ```
 
 use super::StorageError;
-use super::storage_api::ecstore_object::{
-    EncryptionResolutionError, EncryptionResolutionErrorKind, ObjectEncryptionResolver, ReadEncryptionMaterial,
-    ReadEncryptionMode, ReadEncryptionRequest,
-};
 use crate::storage::storage_api::runtime_sources_consumer::runtime_sources;
 #[cfg(feature = "rio-v2")]
 use aes_gcm::aead::Payload;
@@ -159,7 +155,6 @@ use rustfs_utils::http::headers::{
 };
 use rustfs_utils::path::path_join_buf;
 use s3s::dto::{SSECustomerAlgorithm, SSECustomerKey, SSECustomerKeyMD5, SSEKMSKeyId};
-use std::borrow::Cow;
 
 // ============================================================================
 // High-Level SSE Configuration
@@ -657,23 +652,6 @@ pub(crate) fn validate_sse_headers_for_read(metadata: &HashMap<String, String>,
 }
 
 pub(crate) fn map_get_object_reader_error(err: StorageError) -> ApiError {
-    if let StorageError::Io(io_error) = &err
-        && let Some(resolution_error) = io_error
-            .get_ref()
-            .and_then(|source| source.downcast_ref::<EncryptionResolutionError>())
-    {
-        let code = match resolution_error.kind() {
-            EncryptionResolutionErrorKind::InvalidRequest => S3ErrorCode::InvalidRequest,
-            EncryptionResolutionErrorKind::ServiceUnavailable => S3ErrorCode::ServiceUnavailable,
-            _ => S3ErrorCode::InternalError,
-        };
-        return ApiError {
-            code,
-            message: resolution_error.to_string(),
-            source: Some(Box::new(err)),
-        };
-    }
-
     if let Some(message) = map_ssec_get_object_reader_error_message(&err) {
         return ApiError {
             code: S3ErrorCode::InvalidRequest,
@@ -796,117 +774,6 @@ pub enum EncryptionKeyKind {
     Object,
 }
 
-pub(crate) struct SseObjectEncryptionResolver;
-
-#[async_trait]
-impl ObjectEncryptionResolver for SseObjectEncryptionResolver {
-    async fn resolve_read_material(
-        &self,
-        request: ReadEncryptionRequest<'_>,
-    ) -> Result<Option<ReadEncryptionMaterial>, EncryptionResolutionError> {
-        let metadata = normalize_encryption_metadata_case(request.metadata)?;
-        let (customer_algorithm, customer_key, customer_key_md5) =
-            extract_ssec_params_from_headers(request.headers).map_err(map_encryption_resolution_error)?;
-        if let Some(stored_algorithm) = metadata.get("x-amz-server-side-encryption-customer-algorithm") {
-            let request_algorithm = customer_algorithm.as_ref().ok_or_else(|| {
-                map_encryption_resolution_error(ssec_invalid_request(
-                    "The object was stored using a form of Server Side Encryption. \
-                     The correct parameters must be provided to retrieve the object.",
-                ))
-            })?;
-            if stored_algorithm != request_algorithm.as_str() {
-                return Err(map_encryption_resolution_error(ssec_invalid_request(
-                    "The provided encryption parameters did not match the ones used originally to encrypt the object.",
-                )));
-            }
-        }
-        let material = sse_decryption(DecryptionRequest {
-            bucket: request.bucket,
-            key: request.object,
-            metadata: &metadata,
-            sse_customer_key: customer_key.as_ref(),
-            sse_customer_key_md5: customer_key_md5.as_ref(),
-        })
-        .await
-        .map_err(map_encryption_resolution_error)?;
-
-        Ok(material.map(|material| ReadEncryptionMaterial {
-            key_bytes: material.key_bytes,
-            mode: match material.key_kind {
-                EncryptionKeyKind::Direct => ReadEncryptionMode::Direct {
-                    base_nonce: material.base_nonce,
-                },
-                EncryptionKeyKind::Object => ReadEncryptionMode::Object,
-            },
-        }))
-    }
-}
-
-fn normalize_encryption_metadata_case(
-    metadata: &HashMap<String, String>,
-) -> Result<Cow<'_, HashMap<String, String>>, EncryptionResolutionError> {
-    const CANONICAL_KEYS: &[&str] = &[
-        "x-amz-server-side-encryption",
-        "x-amz-server-side-encryption-aws-kms-key-id",
-        "x-amz-server-side-encryption-customer-algorithm",
-        "x-amz-server-side-encryption-customer-key-md5",
-        SSEC_ORIGINAL_SIZE_HEADER,
-        INTERNAL_ENCRYPTION_KEY_ID_HEADER,
-        INTERNAL_ENCRYPTION_KEY_HEADER,
-        INTERNAL_ENCRYPTION_ALGORITHM_HEADER,
-        INTERNAL_ENCRYPTION_IV_HEADER,
-        "x-rustfs-encryption-context",
-        "x-rustfs-encryption-tag",
-        INTERNAL_ENCRYPTION_ORIGINAL_SIZE_HEADER,
-        MINIO_INTERNAL_ENCRYPTION_MULTIPART_HEADER,
-        MINIO_INTERNAL_ENCRYPTION_IV_HEADER,
-        MINIO_INTERNAL_ENCRYPTION_ALGORITHM_HEADER,
-        MINIO_INTERNAL_ENCRYPTION_SSEC_SEALED_KEY_HEADER,
-        MINIO_INTERNAL_ENCRYPTION_S3_SEALED_KEY_HEADER,
-        MINIO_INTERNAL_ENCRYPTION_KMS_SEALED_KEY_HEADER,
-        MINIO_INTERNAL_ENCRYPTION_KMS_KEY_ID_HEADER,
-        MINIO_INTERNAL_ENCRYPTION_KMS_CONTEXT_HEADER,
-    ];
-
-    let needs_normalization = metadata.keys().any(|key| {
-        CANONICAL_KEYS
-            .iter()
-            .any(|canonical| key != canonical && key.eq_ignore_ascii_case(canonical))
-    });
-    if !needs_normalization {
-        return Ok(Cow::Borrowed(metadata));
-    }
-
-    let mut normalized = metadata.clone();
-    for canonical in CANONICAL_KEYS {
-        let mut matching_values = metadata
-            .iter()
-            .filter_map(|(key, value)| key.eq_ignore_ascii_case(canonical).then_some(value));
-        let Some(value) = matching_values.next() else {
-            continue;
-        };
-        if matching_values.any(|candidate| candidate != value) {
-            return Err(EncryptionResolutionError::new(
-                EncryptionResolutionErrorKind::InvalidMetadata,
-                format!("conflicting object encryption metadata for {canonical}"),
-            ));
-        }
-        if !normalized.contains_key(*canonical) {
-            normalized.insert((*canonical).to_string(), value.clone());
-        }
-    }
-    Ok(Cow::Owned(normalized))
-}
-
-fn map_encryption_resolution_error(error: ApiError) -> EncryptionResolutionError {
-    let kind = match error.code {
-        S3ErrorCode::InvalidArgument | S3ErrorCode::InvalidRequest => EncryptionResolutionErrorKind::InvalidRequest,
-        S3ErrorCode::ServiceUnavailable => EncryptionResolutionErrorKind::ServiceUnavailable,
-        _ => EncryptionResolutionErrorKind::DecryptionFailed,
-    };
-    EncryptionResolutionError::new(kind, error.message)
-}
-
 #[derive(Debug, Clone)]
 pub struct ManagedSealedKey {
     #[cfg(feature = "rio-v2")]
@@ -2775,20 +2642,19 @@ fn ssec_invalid_request(message: &str) -> ApiError {
 mod tests {
     use super::{
         ApiError, DataKey, DecryptionRequest, EncryptionKeyKind, EncryptionMaterial, EncryptionRequest,
-        EncryptionResolutionErrorKind, INTERNAL_ENCRYPTION_ALGORITHM_HEADER, INTERNAL_ENCRYPTION_IV_HEADER,
-        INTERNAL_ENCRYPTION_KEY_HEADER, INTERNAL_ENCRYPTION_KEY_ID_HEADER, KmsSseDekProvider, KmsUnavailableError,
-        MINIO_INTERNAL_ENCRYPTION_ALGORITHM_HEADER, MINIO_INTERNAL_ENCRYPTION_IV_HEADER,
-        MINIO_INTERNAL_ENCRYPTION_KMS_CONTEXT_HEADER, MINIO_INTERNAL_ENCRYPTION_KMS_KEY_ID_HEADER,
-        MINIO_INTERNAL_ENCRYPTION_KMS_SEALED_KEY_HEADER, MINIO_INTERNAL_ENCRYPTION_MULTIPART_HEADER,
-        MINIO_INTERNAL_ENCRYPTION_S3_SEALED_KEY_HEADER, MINIO_INTERNAL_ENCRYPTION_SSEC_SEALED_KEY_HEADER,
-        ObjectEncryptionResolver, PrepareEncryptionRequest, ReadEncryptionMode, ReadEncryptionRequest, SSEC_ORIGINAL_SIZE_HEADER,
-        SSEType, SseDekProvider, SseObjectEncryptionResolver, SsecParams, StorageError, TestSseDekProvider,
-        apply_managed_decryption_material, apply_managed_encryption_material, encryption_material_to_metadata,
-        extract_server_side_encryption_from_headers, extract_ssec_params_from_headers, extract_ssekms_context_from_headers,
-        generate_ssec_nonce, is_managed_sse, kms_operation_error, map_get_object_reader_error, mark_encrypted_multipart_metadata,
-        md5_base64, normalize_managed_metadata, reset_sse_dek_provider, resolve_effective_kms_key_id, sse_decryption,
-        sse_encryption, sse_prepare_encryption, strip_managed_encryption_metadata, validate_sse_headers_for_read,
-        validate_sse_headers_for_write, validate_ssec_for_read, validate_ssec_params, verify_ssec_key_match,
+        INTERNAL_ENCRYPTION_ALGORITHM_HEADER, INTERNAL_ENCRYPTION_IV_HEADER, INTERNAL_ENCRYPTION_KEY_HEADER,
+        INTERNAL_ENCRYPTION_KEY_ID_HEADER, KmsSseDekProvider, KmsUnavailableError, MINIO_INTERNAL_ENCRYPTION_ALGORITHM_HEADER,
+        MINIO_INTERNAL_ENCRYPTION_IV_HEADER, MINIO_INTERNAL_ENCRYPTION_KMS_CONTEXT_HEADER,
+        MINIO_INTERNAL_ENCRYPTION_KMS_KEY_ID_HEADER, MINIO_INTERNAL_ENCRYPTION_KMS_SEALED_KEY_HEADER,
+        MINIO_INTERNAL_ENCRYPTION_MULTIPART_HEADER, MINIO_INTERNAL_ENCRYPTION_S3_SEALED_KEY_HEADER,
+        MINIO_INTERNAL_ENCRYPTION_SSEC_SEALED_KEY_HEADER, PrepareEncryptionRequest, SSEC_ORIGINAL_SIZE_HEADER, SSEType,
+        SseDekProvider, SsecParams, StorageError, TestSseDekProvider, apply_managed_decryption_material,
+        apply_managed_encryption_material, encryption_material_to_metadata, extract_server_side_encryption_from_headers,
+        extract_ssec_params_from_headers, extract_ssekms_context_from_headers, generate_ssec_nonce, is_managed_sse,
+        kms_operation_error, map_get_object_reader_error, mark_encrypted_multipart_metadata, md5_base64,
+        normalize_managed_metadata, reset_sse_dek_provider, resolve_effective_kms_key_id, sse_decryption, sse_encryption,
+        sse_prepare_encryption, strip_managed_encryption_metadata, validate_sse_headers_for_read, validate_sse_headers_for_write,
+        validate_ssec_for_read, validate_ssec_params, verify_ssec_key_match,
     };
     #[cfg(feature = "rio-v2")]
     use super::{
@@ -2843,154 +2709,11 @@ mod tests {
     use tokio::sync::Mutex;
 
     static SSE_TEST_LOCK: OnceLock<Mutex<()>> = OnceLock::new();
-    static SSE_TEST_KMS_KEY_DIR: OnceLock<tempfile::TempDir> = OnceLock::new();
 
     async fn lock_sse_test_state() -> tokio::sync::MutexGuard<'static, ()> {
         SSE_TEST_LOCK.get_or_init(|| Mutex::new(())).lock().await
     }
 
-    async fn configure_test_global_local_kms() -> Arc<rustfs_kms::KmsServiceManager> {
-        let key_dir = SSE_TEST_KMS_KEY_DIR.get_or_init(|| tempfile::TempDir::new().expect("create KMS key directory"));
-        let manager = rustfs_kms::init_global_kms_service_manager();
-        manager
-            .reconfigure(rustfs_kms::KmsConfig::local(key_dir.path().to_path_buf()).with_insecure_development_defaults())
-            .await
-            .expect("configure test KMS service");
-        manager
-    }
-
-    #[tokio::test]
-    async fn object_encryption_resolver_returns_ssec_read_material() {
-        let key = [0x31; 32];
-        let key_b64 = BASE64_STANDARD.encode(key);
-        let key_md5 = md5_base64(key);
-        let nonce = [0x42; 12];
-        let metadata = HashMap::from([
-            ("X-Amz-Server-Side-Encryption-Customer-Algorithm".to_string(), "AES256".to_string()),
-            ("X-Amz-Server-Side-Encryption-Customer-Key-Md5".to_string(), key_md5.clone()),
-            ("X-Rustfs-Encryption-Iv".to_string(), BASE64_STANDARD.encode(nonce)),
-        ]);
-        let mut headers = HeaderMap::new();
-        headers.insert("x-amz-server-side-encryption-customer-algorithm", HeaderValue::from_static("AES256"));
-        headers.insert(
-            "x-amz-server-side-encryption-customer-key",
-            HeaderValue::from_str(&key_b64).expect("base64 key is a valid header"),
-        );
-        headers.insert(
-            "x-amz-server-side-encryption-customer-key-md5",
-            HeaderValue::from_str(&key_md5).expect("base64 MD5 is a valid header"),
-        );
-
-        let material = SseObjectEncryptionResolver
-            .resolve_read_material(ReadEncryptionRequest {
-                bucket: "bucket",
-                object: "object",
-                metadata: &metadata,
-                headers: &headers,
-            })
-            .await
-            .expect("SSE-C material should resolve")
-            .expect("SSE-C metadata should produce material");
-
-        assert_eq!(material.key_bytes, key);
-        assert_eq!(material.mode, ReadEncryptionMode::Direct { base_nonce: nonce });
-    }
-
-    #[tokio::test]
-    async fn object_encryption_resolver_rejects_missing_or_invalid_ssec_algorithm() {
-        let key = [0x31; 32];
-        let key_b64 = BASE64_STANDARD.encode(key);
-        let key_md5 = md5_base64(key);
-        let metadata = HashMap::from([
-            ("x-amz-server-side-encryption-customer-algorithm".to_string(), "AES256".to_string()),
-            ("x-amz-server-side-encryption-customer-key-md5".to_string(), key_md5.clone()),
-        ]);
-
-        for algorithm in [None, Some("AES128")] {
-            let mut headers = HeaderMap::new();
-            if let Some(algorithm) = algorithm {
-                headers.insert("x-amz-server-side-encryption-customer-algorithm", HeaderValue::from_static(algorithm));
-            }
-            headers.insert(
-                "x-amz-server-side-encryption-customer-key",
-                HeaderValue::from_str(&key_b64).expect("base64 key is a valid header"),
-            );
-            headers.insert(
-                "x-amz-server-side-encryption-customer-key-md5",
-                HeaderValue::from_str(&key_md5).expect("base64 MD5 is a valid header"),
-            );
-
-            let result = SseObjectEncryptionResolver
-                .resolve_read_material(ReadEncryptionRequest {
-                    bucket: "bucket",
-                    object: "object",
-                    metadata: &metadata,
-                    headers: &headers,
-                })
-                .await;
-            let error = match result {
-                Err(error) => error,
-                Ok(_) => panic!("missing or invalid SSE-C algorithm must fail closed"),
-            };
-
-            assert_eq!(error.kind(), EncryptionResolutionErrorKind::InvalidRequest);
-        }
-    }
-
-    #[tokio::test]
-    async fn object_encryption_resolver_classifies_missing_ssec_key_as_invalid_request() {
-        let metadata = HashMap::from([("x-amz-server-side-encryption-customer-algorithm".to_string(), "AES256".to_string())]);
-        let result = SseObjectEncryptionResolver
-            .resolve_read_material(ReadEncryptionRequest {
-                bucket: "bucket",
-                object: "object",
-                metadata: &metadata,
-                headers: &HeaderMap::new(),
-            })
-            .await;
-        let error = match result {
-            Err(error) => error,
-            Ok(_) => panic!("missing SSE-C key must fail closed"),
-        };
-
-        assert_eq!(error.kind(), EncryptionResolutionErrorKind::InvalidRequest);
-    }
-
-    #[tokio::test]
-    async fn object_encryption_resolver_rejects_conflicting_metadata_case_variants() {
-        let metadata = HashMap::from([
-            ("x-rustfs-encryption-key".to_string(), "first".to_string()),
-            ("X-Rustfs-Encryption-Key".to_string(), "second".to_string()),
-        ]);
-        let result = SseObjectEncryptionResolver
-            .resolve_read_material(ReadEncryptionRequest {
-                bucket: "bucket",
-                object: "object",
-                metadata: &metadata,
-                headers: &HeaderMap::new(),
-            })
-            .await;
-        let error = match result {
-            Err(error) => error,
-            Ok(_) => panic!("conflicting metadata aliases must fail closed"),
-        };
-
-        assert_eq!(error.kind(), EncryptionResolutionErrorKind::InvalidMetadata);
-    }
-
-    #[test]
-    fn normalize_encryption_metadata_case_accepts_lowercase_minio_internal_keys() {
-        let lowercase_key = MINIO_INTERNAL_ENCRYPTION_S3_SEALED_KEY_HEADER.to_ascii_lowercase();
-        let metadata = HashMap::from([(lowercase_key, "sealed-key".to_string())]);
-
-        let normalized = super::normalize_encryption_metadata_case(&metadata).expect("metadata aliases should normalize");
-
-        assert_eq!(
-            normalized.get(MINIO_INTERNAL_ENCRYPTION_S3_SEALED_KEY_HEADER),
-            Some(&"sealed-key".to_string())
-        );
-    }
-
     struct UnavailableSseDekProvider;
 
     #[async_trait::async_trait]
@@ -3710,11 +3433,18 @@ mod tests {
     #[cfg(feature = "rio-v2")]
     #[tokio::test]
     async fn test_sse_kms_roundtrip_persists_and_uses_minio_context() {
+        use rustfs_kms::config::KmsConfig;
         use rustfs_kms::types::{CreateKeyRequest, KeyUsage};
+        use tempfile::TempDir;
         let _guard = lock_sse_test_state().await;
 
         reset_sse_dek_provider();
-        let manager = configure_test_global_local_kms().await;
+        let manager = rustfs_kms::init_global_kms_service_manager();
+        let temp_dir = TempDir::new().expect("temp dir");
+        manager
+            .reconfigure(KmsConfig::local(temp_dir.path().to_path_buf()).with_insecure_development_defaults())
+            .await
+            .expect("kms reconfigure should succeed");
         manager
             .get_encryption_service()
             .await
@@ -4006,19 +3736,6 @@ mod tests {
 
                 assert_eq!(decrypted.key_kind, EncryptionKeyKind::Object);
                 assert_eq!(decrypted.key_bytes, material.key_bytes);
-
-                let resolved = SseObjectEncryptionResolver
-                    .resolve_read_material(ReadEncryptionRequest {
-                        bucket: "bucket",
-                        object: "object",
-                        metadata: &metadata,
-                        headers: &HeaderMap::new(),
-                    })
-                    .await
-                    .expect("managed resolver")
-                    .expect("managed material");
-                assert_eq!(resolved.mode, ReadEncryptionMode::Object);
-                assert_eq!(resolved.key_bytes, material.key_bytes);
             },
         )
         .await;
@@ -4079,29 +3796,6 @@ mod tests {
 
         assert_eq!(decrypted.key_kind, EncryptionKeyKind::Object);
         assert_eq!(decrypted.key_bytes, material.key_bytes);
-
-        let mut headers = HeaderMap::new();
-        headers.insert("x-amz-server-side-encryption-customer-algorithm", HeaderValue::from_static("AES256"));
-        headers.insert(
-            "x-amz-server-side-encryption-customer-key",
-            HeaderValue::from_str(&customer_key).expect("customer key header"),
-        );
-        headers.insert(
-            "x-amz-server-side-encryption-customer-key-md5",
-            HeaderValue::from_str(&customer_key_md5).expect("customer key MD5 header"),
-        );
-        let resolved = SseObjectEncryptionResolver
-            .resolve_read_material(ReadEncryptionRequest {
-                bucket: "bucket",
-                object: "object",
-                metadata: &metadata,
-                headers: &headers,
-            })
-            .await
-            .expect("SSE-C resolver")
-            .expect("SSE-C material");
-        assert_eq!(resolved.mode, ReadEncryptionMode::Object);
-        assert_eq!(resolved.key_bytes, material.key_bytes);
     }
 
     #[cfg(feature = "rio-v2")]
@@ -4576,9 +4270,17 @@ mod tests {
 
     #[tokio::test]
     async fn test_managed_decryption_selects_provider_from_persisted_dek() {
+        use rustfs_kms::config::KmsConfig;
+        use tempfile::TempDir;
+
         let _guard = lock_sse_test_state().await;
         reset_sse_dek_provider();
-        let manager = configure_test_global_local_kms().await;
+        let manager = rustfs_kms::init_global_kms_service_manager();
+        let key_dir = TempDir::new().expect("create KMS key directory");
+        manager
+            .reconfigure(KmsConfig::local(key_dir.path().to_path_buf()).with_insecure_development_defaults())
+            .await
+            .expect("start test KMS service");
 
         let local_master_key = [7u8; 32];
         let local_provider = TestSseDekProvider::new_with_key(local_master_key);
@@ -4646,6 +4348,9 @@ mod tests {
     /// the local-provider cache.
     #[tokio::test]
     async fn test_kms_envelope_never_routes_to_cached_local_provider() {
+        use rustfs_kms::config::KmsConfig;
+        use tempfile::TempDir;
+
         let _guard = lock_sse_test_state().await;
         reset_sse_dek_provider();
 
@@ -4657,7 +4362,12 @@ mod tests {
             .expect("write local provider into local cache") = Some(Arc::new(TestSseDekProvider::new_with_key(local_master_key)));
 
         // 2. Start a KMS service (dynamic enable).
-        let manager = configure_test_global_local_kms().await;
+        let manager = rustfs_kms::init_global_kms_service_manager();
+        let key_dir = TempDir::new().expect("create KMS key directory");
+        manager
+            .reconfigure(KmsConfig::local(key_dir.path().to_path_buf()).with_insecure_development_defaults())
+            .await
+            .expect("start test KMS service");
 
         // 3. Construct a KMS JSON envelope — the persisted format of a KMS-wrapped DEK.
         //    is_data_key_envelope() will return true for this payload.
@@ -4748,7 +4458,7 @@ mod tests {
         use rustfs_kms::config::KmsConfig;
         let _guard = lock_sse_test_state().await;
 
-        let manager = Arc::new(rustfs_kms::KmsServiceManager::new());
+        let manager = rustfs_kms::init_global_kms_service_manager();
 
         manager
             .reconfigure(KmsConfig::static_kms("first-key".to_string(), BASE64_STANDARD.encode([0x11; 32])))
@@ -4982,15 +4692,6 @@ mod tests {
         );
     }
 
-    #[test]
-    fn test_map_get_object_reader_error_preserves_typed_service_unavailable() {
-        let resolution_error =
-            super::EncryptionResolutionError::new(EncryptionResolutionErrorKind::ServiceUnavailable, "KMS unavailable");
-        let err = map_get_object_reader_error(StorageError::other(resolution_error));
-        assert_eq!(err.code, S3ErrorCode::ServiceUnavailable);
-        assert_eq!(err.message, "KMS unavailable");
-    }
-
     #[test]
     fn test_map_get_object_reader_error_leaves_non_ssec_errors_unchanged() {
         let err = map_get_object_reader_error(StorageError::other("plain io failure"));

rustfs/src/storage/storage_api.rs

@@ -510,21 +510,12 @@ pub(crate) mod ecstore_object {
     #[cfg(test)]
     pub(crate) use rustfs_ecstore::api::object::GetObjectBodySource;
     pub(crate) use rustfs_ecstore::api::object::{
-        EncryptionResolutionError, EncryptionResolutionErrorKind, GetObjectBodyCacheHook, GetObjectBodyCacheHookLookup,
-        ObjectEncryptionResolver, ObjectMutationHook, ReadEncryptionMaterial, ReadEncryptionMode, ReadEncryptionRequest,
-        get_object_body_cache_plaintext_len, lookup_get_object_body_cache_hook, register_get_object_body_cache_hook,
-        register_object_mutation_hook, unregister_get_object_body_cache_hook, unregister_object_mutation_hook,
+        GetObjectBodyCacheHook, GetObjectBodyCacheHookLookup, ObjectMutationHook, get_object_body_cache_plaintext_len,
+        lookup_get_object_body_cache_hook, register_get_object_body_cache_hook, register_object_mutation_hook,
+        unregister_get_object_body_cache_hook, unregister_object_mutation_hook,
     };
 }
 
-#[cfg(all(test, feature = "rio-v2"))]
-pub(crate) mod ecstore_test_support {
-    pub(crate) use rustfs_ecstore::api::bitrot::create_bitrot_reader;
-    pub(crate) use rustfs_ecstore::api::disk::{DiskAPI, DiskOption, endpoint::Endpoint, new_disk};
-    pub(crate) use rustfs_ecstore::api::erasure::Erasure;
-    pub(crate) use rustfs_ecstore::api::object::{GetObjectReader, ObjectInfo, ObjectOptions};
-}
-
 pub(crate) mod ecstore_set_disk {
     pub(crate) use rustfs_ecstore::api::set_disk::{DEFAULT_READ_BUFFER_SIZE, get_lock_acquire_timeout, is_valid_storage_class};
 }
@@ -955,21 +946,13 @@ pub(crate) async fn init_local_disks(endpoint_pools: EndpointServerPools) -> Res
 /// The process-level bootstrap instance context that single-instance startup
 /// threads through the storage foundation (Phase 5 follow-up, backlog#1052).
 pub(crate) fn bootstrap_instance_ctx() -> Arc<InstanceContext> {
-    let context = ecstore_runtime::bootstrap_ctx();
-    configure_object_encryption_resolver(&context);
-    context
+    ecstore_runtime::bootstrap_ctx()
 }
 
 /// Construct a fresh per-server instance context (backlog#1052 S5): a second
 /// embedded server owns its own erasure/region/endpoint/deployment id cells.
 pub(crate) fn new_instance_ctx() -> Arc<InstanceContext> {
-    let context = Arc::new(InstanceContext::new());
-    configure_object_encryption_resolver(&context);
-    context
-}
-
-fn configure_object_encryption_resolver(context: &InstanceContext) {
-    let _ = context.set_object_encryption_resolver(Arc::new(super::sse::SseObjectEncryptionResolver));
+    Arc::new(InstanceContext::new())
 }
 
 pub(crate) fn init_lock_clients(endpoint_pools: EndpointServerPools) {
@@ -1459,8 +1442,8 @@ pub(crate) async fn set_bucket_metadata(bucket: String, bm: BucketMetadata) -> R
     ecstore_bucket::metadata_sys::set_bucket_metadata(bucket, bm).await
 }
 
-pub(crate) async fn reload_bucket_metadata(bucket: &str) -> Result<()> {
-    ecstore_bucket::metadata_sys::reload_bucket_metadata(bucket).await
+pub(crate) async fn reload_bucket_metadata(api: Arc<ECStore>, bucket: &str) -> Result<()> {
+    ecstore_bucket::metadata_sys::reload_bucket_metadata(api, bucket).await
 }
 
 pub(crate) async fn remove_bucket_metadata(bucket: &str) -> Result<bool> {
@@ -1747,7 +1730,7 @@ pub(crate) async fn init_compression_total_memory_from_backend(store: Arc<ECStor
 mod tests {
     use super::{
         apply_active_resync_intents, bucket_targets_metadata_lock_shard, ecstore_bucket, lock_bucket_targets_metadata,
-        new_instance_ctx, scanner_maintenance_config_file,
+        scanner_maintenance_config_file,
     };
     use std::time::Duration;
 
@@ -1778,16 +1761,6 @@ mod tests {
         );
     }
 
-    #[test]
-    fn fresh_instance_context_installs_object_encryption_resolver() {
-        assert!(new_instance_ctx().object_encryption_resolver().is_some());
-    }
-
-    #[test]
-    fn bootstrap_instance_context_installs_object_encryption_resolver() {
-        assert!(super::bootstrap_instance_ctx().object_encryption_resolver().is_some());
-    }
-
     #[test]
     fn scanner_maintenance_config_only_includes_scanner_owned_work() {
         assert!(scanner_maintenance_config_file(ecstore_bucket::metadata::BUCKET_LIFECYCLE_CONFIG));

scripts/run_ecstore_validation_suite.sh

@@ -317,7 +317,7 @@ write_blackbox_matrix() {
 	    printf 'quick\theal degraded erasure disk rebuild\tblack-box\tcargo test --package e2e_test heal_erasure_disk_rebuild_test -- --nocapture\tnone\t%s\n' "$e2e_status"
 	    printf 'quick\tnamespace lock quorum under EC ops\tblack-box\tcargo test --package e2e_test namespace_lock_quorum_test -- --nocapture\tnone\t%s\n' "$e2e_status"
 	    printf 'full\tlegacy bitrot read fixture restore\tfixture\tcargo test -p rustfs-ecstore --test legacy_bitrot_read_test -- --nocapture\tRUSTFS_LEGACY_TEST_ROOT,RUSTFS_LEGACY_TEST_DISK\t%s\n' "$legacy_status"
-	    printf 'full\tMinIO generated encrypted read and negative restore fixture\tfixture\tcargo test -p rustfs --features rio-v2 storage::minio_generated_read_test --lib -- --ignored --nocapture\tRUSTFS_MINIO_FIXTURE_ROOT,RUSTFS_MINIO_STATIC_KMS_KEY_B64\t%s\n' "$minio_status"
+	    printf 'full\tMinIO generated encrypted read and negative restore fixture\tfixture\tcargo test -p rustfs-ecstore --features rio-v2 --test minio_generated_read_test -- --ignored --nocapture\tRUSTFS_MINIO_FIXTURE_ROOT,RUSTFS_MINIO_STATIC_KMS_KEY_B64\t%s\n' "$minio_status"
 	    printf 'full\tS3 multipart range versioning delete subset\tblack-box\tenv TESTEXPR=\"multipart or range or versioning or delete\" DEPLOY_MODE=build MAXFAIL=0 ./scripts/s3-tests/run.sh\tnone\t%s\n' "$s3_status"
 	    printf 'destructive\tdistributed cluster concurrency\tblack-box\tcargo test --package e2e_test cluster_concurrency_test -- --nocapture\tnone\t%s\n' "$destructive_status"
 	    printf 'destructive\tstale multipart cleanup cluster\tblack-box\tcargo test --package e2e_test stale_multipart_cleanup_cluster_test -- --nocapture\tnone\t%s\n' "$destructive_status"
@@ -377,7 +377,7 @@ run_fixture_steps() {
 
 	  if fixture_available; then
 	    run_step "ecstore-minio-generated-read-fixture" \
-	      cargo test -p rustfs --features rio-v2 storage::minio_generated_read_test --lib -- --ignored --nocapture
+	      cargo test -p rustfs-ecstore --features rio-v2 --test minio_generated_read_test -- --ignored --nocapture
 	  elif [[ "$REQUIRE_FIXTURES" == "true" ]]; then
 	    echo "ERROR: $(minio_fixture_missing_reason)" >&2
 	    exit 1

scripts/run_hotpath_warp_ab.sh

@@ -337,7 +337,7 @@ measure() {
   )
   [[ -n "$COOLDOWN_SECS" ]] && args+=(--cooldown-secs "$COOLDOWN_SECS")
   [[ -n "$baseline_csv" ]] && args+=(--baseline-csv "$baseline_csv")
-  run "$ENHANCED_BENCH" "${args[@]}"
+  run "$ENHANCED_BENCH" "${args[@]}" >&2
   echo "$cell"
 }
 

scripts/test_entrypoint_credentials.sh

@@ -226,7 +226,11 @@ run_cli_expect_failure \
 
 # Non-server commands skip credential validation entirely.
 cargo_log="$TMP_DIR/cargo.log"
-if ! env -i PATH="$PATH" RUSTFS_VOLUMES="$TMP_DIR/data" RUSTFS_OBS_LOG_DIRECTORY= sh "$ENTRYPOINT" cargo --version >"$cargo_log" 2>&1; then
+toolchain_env=(PATH="$PATH")
+[ "${HOME+x}" = x ] && toolchain_env+=(HOME="$HOME")
+[ "${CARGO_HOME+x}" = x ] && toolchain_env+=(CARGO_HOME="$CARGO_HOME")
+[ "${RUSTUP_HOME+x}" = x ] && toolchain_env+=(RUSTUP_HOME="$RUSTUP_HOME")
+if ! env -i "${toolchain_env[@]}" RUSTFS_VOLUMES="$TMP_DIR/data" RUSTFS_OBS_LOG_DIRECTORY= sh "$ENTRYPOINT" cargo --version >"$cargo_log" 2>&1; then
   echo "Expected cargo passthrough to skip server credential checks" >&2
   cat "$cargo_log" >&2
   exit 1