refactor(data-usage): own SizeSummary once, with the scanner's semantics (#6237)

`SizeSummary` and `ReplTargetSizeSummary` existed in both `rustfs-data-usage` and `rustfs-scanner`, and the two copies had drifted three ways: four size fields were `usize` in one and `i64` in the other, only the scanner's carried `tier_stats`, and — the difference that matters — the scanner's `add` saturated while the data-usage copy used plain `+=`, which panics on overflow in a debug build and wraps in a release one.

The data-usage copy is now the only definition and takes the scanner's shape and semantics, since that is the side a test already pinned (`MAX + 1 == MAX`). An equivalent saturation test now guards it in its new home. The scanner re-exports both types alongside the ones it already re-exported.

`DataUsageEntry::add_sizes` and `BucketUsageInfo::add_size_summary` are removed. Both took a `SizeSummary` and had no callers anywhere — they were the duplicate fold paths, and `apply_scanner_size_summary` is now the only one.

`actions_accounting` stays in the scanner as the `ScannerSizeSummaryExt` extension trait: it needs `ObjectInfo`, which sits above `rustfs-data-usage`, and an inherent impl on a foreign type is not allowed. The three call sites are unchanged.

Refs backlog#1828

.github/workflows/package.yml

@@ -522,10 +522,16 @@ jobs:
             for f in "$DEB_FILE" "$RPM_FILE"; do
               if [[ -n "$f" && -f "$f" ]]; then
                 base="$(basename "$f")"
-                # Remove any stale entry, then append the fresh digest
+                # GitHub stores release asset names with '~' normalized to '.'
+                # (e.g. rustfs_1.0.0~rc.2_amd64.deb is stored as
+                # rustfs_1.0.0.rc.2_amd64.deb), so checksum entries must
+                # reference the name as stored on the release.
+                github_base="${base//\~/.}"
+                # Remove any stale entry (both naming variants), then append
                 grep -Fv -- "$base" "$checksum_file" > "${checksum_file}.tmp" || true
-                mv "${checksum_file}.tmp" "$checksum_file"
-                (cd "$(dirname "$f")" && "$checksum_cmd" -- "$base") >> "$checksum_file"
+                grep -Fv -- "$github_base" "${checksum_file}.tmp" > "${checksum_file}.tmp2" || true
+                mv "${checksum_file}.tmp2" "$checksum_file"
+                (cd "$(dirname "$f")" && "$checksum_cmd" -- "$github_base") >> "$checksum_file"
               fi
             done
 

crates/data-usage/src/data_usage.rs

@@ -317,15 +317,15 @@ pub struct SizeSummary {
     /// Number of delete markers
     pub delete_markers: usize,
     /// Replicated size
-    pub replicated_size: usize,
+    pub replicated_size: i64,
     /// Replicated count
     pub replicated_count: usize,
     /// Pending size
-    pub pending_size: usize,
+    pub pending_size: i64,
     /// Failed size
-    pub failed_size: usize,
+    pub failed_size: i64,
     /// Replica size
-    pub replica_size: usize,
+    pub replica_size: i64,
     /// Replica count
     pub replica_count: usize,
     /// Pending count
@@ -334,19 +334,21 @@ pub struct SizeSummary {
     pub failed_count: usize,
     /// Replication target stats
     pub repl_target_stats: HashMap<String, ReplTargetSizeSummary>,
+    /// Per-tier accounting, keyed by storage class or remote tier name
+    pub tier_stats: HashMap<String, TierStats>,
 }
 
 /// Replication target size summary
 #[derive(Debug, Default, Clone)]
 pub struct ReplTargetSizeSummary {
     /// Replicated size
-    pub replicated_size: usize,
+    pub replicated_size: i64,
     /// Replicated count
     pub replicated_count: usize,
     /// Pending size
-    pub pending_size: usize,
+    pub pending_size: i64,
     /// Failed size
-    pub failed_size: usize,
+    pub failed_size: i64,
     /// Pending count
     pub pending_count: usize,
     /// Failed count
@@ -710,28 +712,6 @@ impl DataUsageEntry {
         self.children.insert(hash.key());
     }
 
-    pub fn add_sizes(&mut self, summary: &SizeSummary) {
-        self.size += summary.total_size;
-        self.versions += summary.versions;
-        self.delete_markers += summary.delete_markers;
-        self.obj_sizes.add(summary.total_size as u64);
-        self.obj_versions.add(summary.versions as u64);
-
-        let replication_stats = self.replication_stats.get_or_insert_with(ReplicationAllStats::default);
-        replication_stats.replica_size += summary.replica_size as u64;
-        replication_stats.replica_count += summary.replica_count as u64;
-
-        for (arn, st) in &summary.repl_target_stats {
-            let tgt_stat = replication_stats.targets.entry(arn.to_string()).or_default();
-            tgt_stat.pending_size += st.pending_size as u64;
-            tgt_stat.failed_size += st.failed_size as u64;
-            tgt_stat.replicated_size += st.replicated_size as u64;
-            tgt_stat.replicated_count += st.replicated_count as u64;
-            tgt_stat.failed_count += st.failed_count as u64;
-            tgt_stat.pending_count += st.pending_count as u64;
-        }
-    }
-
     pub fn merge(&mut self, other: &DataUsageEntry) {
         self.objects += other.objects;
         self.versions += other.versions;
@@ -1722,14 +1702,6 @@ impl BucketUsageInfo {
     }
 
     /// Add size summary to this bucket usage
-    pub fn add_size_summary(&mut self, summary: &SizeSummary) {
-        self.size += summary.total_size as u64;
-        self.versions_count += summary.versions as u64;
-        self.delete_markers_count += summary.delete_markers as u64;
-        self.replica_size += summary.replica_size as u64;
-        self.replica_count += summary.replica_count as u64;
-    }
-
     /// Merge another BucketUsageInfo into this one
     pub fn merge(&mut self, other: &BucketUsageInfo) {
         self.size += other.size;
@@ -1775,29 +1747,32 @@ impl SizeSummary {
         Self::default()
     }
 
-    /// Add another SizeSummary to this one
+    /// Add another SizeSummary to this one.
+    ///
+    /// Saturating throughout: a scan that overflows a counter should report the
+    /// ceiling rather than panic in a debug build or wrap in a release one.
     pub fn add(&mut self, other: &SizeSummary) {
-        self.total_size += other.total_size;
-        self.versions += other.versions;
-        self.delete_markers += other.delete_markers;
-        self.replicated_size += other.replicated_size;
-        self.replicated_count += other.replicated_count;
-        self.pending_size += other.pending_size;
-        self.failed_size += other.failed_size;
-        self.replica_size += other.replica_size;
-        self.replica_count += other.replica_count;
-        self.pending_count += other.pending_count;
-        self.failed_count += other.failed_count;
+        self.total_size = self.total_size.saturating_add(other.total_size);
+        self.versions = self.versions.saturating_add(other.versions);
+        self.delete_markers = self.delete_markers.saturating_add(other.delete_markers);
+        self.replicated_size = self.replicated_size.saturating_add(other.replicated_size);
+        self.replicated_count = self.replicated_count.saturating_add(other.replicated_count);
+        self.pending_size = self.pending_size.saturating_add(other.pending_size);
+        self.failed_size = self.failed_size.saturating_add(other.failed_size);
+        self.replica_size = self.replica_size.saturating_add(other.replica_size);
+        self.replica_count = self.replica_count.saturating_add(other.replica_count);
+        self.pending_count = self.pending_count.saturating_add(other.pending_count);
+        self.failed_count = self.failed_count.saturating_add(other.failed_count);
 
         // Merge replication target stats
         for (target, stats) in &other.repl_target_stats {
             let entry = self.repl_target_stats.entry(target.clone()).or_default();
-            entry.replicated_size += stats.replicated_size;
-            entry.replicated_count += stats.replicated_count;
-            entry.pending_size += stats.pending_size;
-            entry.failed_size += stats.failed_size;
-            entry.pending_count += stats.pending_count;
-            entry.failed_count += stats.failed_count;
+            entry.replicated_size = entry.replicated_size.saturating_add(stats.replicated_size);
+            entry.replicated_count = entry.replicated_count.saturating_add(stats.replicated_count);
+            entry.pending_size = entry.pending_size.saturating_add(stats.pending_size);
+            entry.failed_size = entry.failed_size.saturating_add(stats.failed_size);
+            entry.pending_count = entry.pending_count.saturating_add(stats.pending_count);
+            entry.failed_count = entry.failed_count.saturating_add(stats.failed_count);
         }
     }
 }
@@ -2343,6 +2318,64 @@ mod tests {
         assert_eq!(usage1.versions_count, 15);
     }
 
+    #[test]
+    fn size_summary_add_saturates_instead_of_overflowing() {
+        // The scanner folds one summary per object into a per-prefix total, so a
+        // counter at its ceiling must stay there rather than panic in a debug
+        // build or wrap in a release one (backlog#1828).
+        let mut summary = SizeSummary {
+            total_size: usize::MAX,
+            versions: usize::MAX,
+            replicated_size: i64::MAX,
+            pending_size: i64::MAX,
+            failed_size: i64::MAX,
+            replica_size: i64::MAX,
+            ..Default::default()
+        };
+        summary.repl_target_stats.insert(
+            "arn".to_string(),
+            ReplTargetSizeSummary {
+                replicated_size: i64::MAX,
+                pending_size: i64::MAX,
+                failed_size: i64::MAX,
+                ..Default::default()
+            },
+        );
+
+        let mut increment = SizeSummary {
+            total_size: 1,
+            versions: 1,
+            replicated_size: 1,
+            pending_size: 1,
+            failed_size: 1,
+            replica_size: 1,
+            ..Default::default()
+        };
+        increment.repl_target_stats.insert(
+            "arn".to_string(),
+            ReplTargetSizeSummary {
+                replicated_size: 1,
+                pending_size: 1,
+                failed_size: 1,
+                ..Default::default()
+            },
+        );
+
+        summary.add(&increment);
+
+        assert_eq!(summary.total_size, usize::MAX);
+        assert_eq!(summary.versions, usize::MAX);
+        assert_eq!(summary.replicated_size, i64::MAX);
+        assert_eq!(summary.pending_size, i64::MAX);
+        assert_eq!(summary.failed_size, i64::MAX);
+        assert_eq!(summary.replica_size, i64::MAX);
+
+        let target = summary.repl_target_stats.get("arn").expect("target survives the merge");
+        assert_eq!(target.replicated_size, i64::MAX);
+        assert_eq!(target.pending_size, i64::MAX);
+        assert_eq!(target.failed_size, i64::MAX);
+    }
+
     #[test]
     fn test_size_summary_add() {
         let mut summary1 = SizeSummary::new();

crates/e2e_test/src/multipart_auth_test.rs

@@ -17,6 +17,7 @@
 use crate::common::{RustFSTestEnvironment, init_logging, local_http_client};
 use async_compression::tokio::write::{BzEncoder, XzEncoder};
 use aws_sdk_s3::error::{ProvideErrorMetadata, SdkError};
+use aws_sdk_s3::operation::head_object::HeadObjectOutput;
 use aws_sdk_s3::primitives::ByteStream;
 use aws_sdk_s3::types::{
     ServerSideEncryption, ServerSideEncryptionByDefault, ServerSideEncryptionConfiguration, ServerSideEncryptionRule,
@@ -348,6 +349,71 @@ async fn run_post_object_policy_case(
     Ok(())
 }
 
+/// One accepted POST Object upload driven end-to-end (backlog#1838): starts a
+/// fresh server, allows anonymous PutObject on `bucket`, posts an anonymous
+/// POST Object form whose policy carries `policy_conditions` and whose form
+/// carries `form_field` on top of the mandatory key+policy fields, then asserts
+/// 204 with an empty body, that `read_stored` observes the submitted value on
+/// the stored object, and that the object body round-tripped unchanged.
+/// `case` prefixes every assertion message so a failing table row is
+/// identifiable at a glance.
+#[allow(clippy::too_many_arguments)]
+async fn run_post_object_accept_case(
+    bucket: &str,
+    object_key: &str,
+    policy_conditions: Vec<serde_json::Value>,
+    form_field: (&str, &str),
+    file_mime: &str,
+    file_body: &[u8],
+    read_stored: fn(&HeadObjectOutput) -> Option<&str>,
+    case: &str,
+) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    let mut env = RustFSTestEnvironment::new().await?;
+    env.start_rustfs_server(vec![]).await?;
+
+    let admin_client = env.create_s3_client();
+    admin_client.create_bucket().bucket(bucket).send().await?;
+    allow_anonymous_put_object(&admin_client, bucket).await?;
+
+    let policy = encode_post_policy(policy_conditions);
+
+    let (field_name, field_value) = form_field;
+    let post_form = reqwest::multipart::Form::new()
+        .text("key", object_key.to_string())
+        .text("policy", policy)
+        .text(field_name.to_string(), field_value.to_string())
+        .part(
+            "file",
+            reqwest::multipart::Part::bytes(file_body.to_vec())
+                .file_name("upload.txt")
+                .mime_str(file_mime)?,
+        );
+
+    let post_resp = local_http_client()
+        .post(format!("{}/{}", env.url, bucket))
+        .multipart(post_form)
+        .send()
+        .await?;
+
+    let status = post_resp.status();
+    let response_body = post_resp.text().await?;
+
+    assert_eq!(status, reqwest::StatusCode::NO_CONTENT, "[{case}] unexpected status");
+    assert!(
+        response_body.is_empty(),
+        "[{case}] 204 response should not contain a body, got: {response_body}"
+    );
+
+    let head = admin_client.head_object().bucket(bucket).key(object_key).send().await?;
+    assert_eq!(read_stored(&head), Some(field_value), "[{case}] stored {field_name} mismatch");
+
+    let get_out = admin_client.get_object().bucket(bucket).key(object_key).send().await?;
+    let uploaded = get_out.body.collect().await?.into_bytes();
+    assert_eq!(uploaded.as_ref(), file_body, "[{case}] uploaded body mismatch");
+
+    Ok(())
+}
+
 /// Table-driven fold of the nine `*_missing_from_policy_conditions` POST
 /// Object tests (backlog#1838 PR1). Every row keeps its original test's exact
 /// bucket, key, form field, file body, and expected error strings; the shared
@@ -1534,59 +1600,6 @@ async fn test_anonymous_post_object_accepts_sse_s3_missing_from_policy_condition
     Ok(())
 }
 
-#[tokio::test]
-async fn test_anonymous_post_object_accepts_storage_class_exact_policy_match()
--> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-    init_logging();
-
-    let mut env = RustFSTestEnvironment::new().await?;
-    env.start_rustfs_server(vec![]).await?;
-
-    let bucket = "anon-post-storage-class";
-    let object_key = "post-storage-class-object.txt";
-    let expected_body = b"post-storage-class-body".to_vec();
-    let storage_class = "REDUCED_REDUNDANCY";
-
-    let admin_client = env.create_s3_client();
-    admin_client.create_bucket().bucket(bucket).send().await?;
-    allow_anonymous_put_object(&admin_client, bucket).await?;
-
-    let policy = encode_post_policy(vec![
-        serde_json::json!({ "bucket": bucket }),
-        serde_json::json!({ "key": object_key }),
-        serde_json::json!({ "x-amz-storage-class": storage_class }),
-        serde_json::json!(["content-length-range", 0, 1024]),
-    ]);
-
-    let post_form = reqwest::multipart::Form::new()
-        .text("key", object_key.to_string())
-        .text("policy", policy)
-        .text("x-amz-storage-class", storage_class)
-        .part(
-            "file",
-            reqwest::multipart::Part::bytes(expected_body.clone())
-                .file_name("upload.txt")
-                .mime_str("text/plain")?,
-        );
-
-    let post_resp = local_http_client()
-        .post(format!("{}/{}", env.url, bucket))
-        .multipart(post_form)
-        .send()
-        .await?;
-
-    assert_eq!(post_resp.status(), reqwest::StatusCode::NO_CONTENT);
-
-    let head = admin_client.head_object().bucket(bucket).key(object_key).send().await?;
-    assert_eq!(head.storage_class().map(|value| value.as_str()), Some(storage_class));
-
-    let uploaded = admin_client.get_object().bucket(bucket).key(object_key).send().await?;
-    let uploaded = uploaded.body.collect().await?.into_bytes();
-    assert_eq!(uploaded.as_ref(), expected_body.as_slice());
-
-    Ok(())
-}
-
 #[tokio::test]
 async fn test_anonymous_post_object_rejects_storage_class_missing_from_policy_conditions()
 -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
@@ -2584,512 +2597,182 @@ async fn test_anonymous_post_object_rejects_success_action_redirect_missing_from
     Ok(())
 }
 
+/// Table-driven fold of the eleven accepted POST Object form-field tests
+/// (backlog#1838 PR4). Every row keeps its original test's exact bucket, key,
+/// form field, submitted value, policy condition, file MIME type, and file
+/// body; the shared shape is: the policy covers the field (exact condition or
+/// `starts-with` prefix), the form submits it, the upload returns 204 with an
+/// empty body, and the stored object echoes the submitted value back.
 #[tokio::test]
-async fn test_anonymous_post_object_accepts_metadata_field_covered_by_starts_with()
+async fn test_anonymous_post_object_accepts_fields_covered_by_policy_conditions()
 -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     init_logging();
 
-    let mut env = RustFSTestEnvironment::new().await?;
-    env.start_rustfs_server(vec![]).await?;
+    // (case, bucket, object_key, field, submitted value, `starts-with` prefix
+    // (`None` pins the field to an exact policy condition), file part MIME type,
+    // file body, stored-value accessor)
+    type Case = (
+        &'static str,
+        &'static str,
+        &'static str,
+        &'static str,
+        &'static str,
+        Option<&'static str>,
+        &'static str,
+        &'static [u8],
+        fn(&HeadObjectOutput) -> Option<&str>,
+    );
+    let cases: &[Case] = &[
+        (
+            "storage-class",
+            "anon-post-storage-class",
+            "post-storage-class-object.txt",
+            "x-amz-storage-class",
+            "REDUCED_REDUNDANCY",
+            None,
+            "text/plain",
+            b"post-storage-class-body",
+            |head: &HeadObjectOutput| head.storage_class().map(|value| value.as_str()),
+        ),
+        (
+            "metadata-starts-with",
+            "anon-post-policy-meta-accept",
+            "uploads/meta-object.txt",
+            "x-amz-meta-project",
+            "alpha-demo",
+            Some("alpha-"),
+            "text/plain",
+            b"post-policy-meta-body",
+            |head: &HeadObjectOutput| head.metadata().and_then(|meta| meta.get("project")).map(String::as_str),
+        ),
+        (
+            "content-type",
+            "anon-post-policy-content-type-accept",
+            "uploads/content-type-accept.txt",
+            "Content-Type",
+            "text/plain",
+            None,
+            "text/plain",
+            b"post-policy-content-type-accept",
+            |head: &HeadObjectOutput| head.content_type(),
+        ),
+        (
+            "content-type-starts-with",
+            "anon-post-policy-content-type-accept",
+            "uploads/content-type-object.txt",
+            "Content-Type",
+            "image/png",
+            Some("image/"),
+            "image/png",
+            b"post-policy-content-type-body",
+            |head: &HeadObjectOutput| head.content_type(),
+        ),
+        (
+            "content-disposition",
+            "anon-post-policy-disposition-accept",
+            "uploads/disposition-object.txt",
+            "Content-Disposition",
+            "attachment; filename=\"upload.txt\"",
+            None,
+            "text/plain",
+            b"post-policy-disposition-body",
+            |head: &HeadObjectOutput| head.content_disposition(),
+        ),
+        (
+            "cache-control",
+            "anon-post-policy-cache-control-accept",
+            "uploads/cache-control-object.txt",
+            "Cache-Control",
+            "max-age=60",
+            None,
+            "text/plain",
+            b"post-policy-cache-control-body",
+            |head: &HeadObjectOutput| head.cache_control(),
+        ),
+        (
+            "content-language",
+            "anon-post-policy-content-language-accept",
+            "uploads/content-language-object.txt",
+            "Content-Language",
+            "en-US",
+            None,
+            "text/plain",
+            b"post-policy-content-language-body",
+            |head: &HeadObjectOutput| head.content_language(),
+        ),
+        (
+            "content-encoding",
+            "anon-post-policy-content-encoding-accept",
+            "uploads/content-encoding-object.txt",
+            "Content-Encoding",
+            "gzip",
+            None,
+            "text/plain",
+            b"post-policy-content-encoding-body",
+            |head: &HeadObjectOutput| head.content_encoding(),
+        ),
+        (
+            "website-redirect-location",
+            "anon-post-policy-website-redirect-accept",
+            "uploads/website-redirect-object.txt",
+            "x-amz-website-redirect-location",
+            "/docs/landing.html",
+            None,
+            "text/plain",
+            b"post-policy-website-redirect-body",
+            |head: &HeadObjectOutput| head.website_redirect_location(),
+        ),
+        (
+            "expires",
+            "anon-post-policy-expires-accept",
+            "uploads/expires-object.txt",
+            "Expires",
+            "Wed, 21 Oct 2037 07:28:00 GMT",
+            None,
+            "text/plain",
+            b"post-policy-expires-body",
+            |head: &HeadObjectOutput| head.expires_string(),
+        ),
+        (
+            "metadata-exact",
+            "anon-post-policy-meta-exact-accept",
+            "uploads/meta-exact-accept-object.txt",
+            "x-amz-meta-project",
+            "alpha-demo",
+            None,
+            "text/plain",
+            b"post-policy-meta-exact-body",
+            |head: &HeadObjectOutput| head.metadata().and_then(|meta| meta.get("project")).map(String::as_str),
+        ),
+    ];
 
-    let bucket = "anon-post-policy-meta-accept";
-    let object_key = "uploads/meta-object.txt";
-    let metadata_value = "alpha-demo";
-    let expected_body = b"post-policy-meta-body".to_vec();
+    for (case, bucket, object_key, field, value, starts_with_prefix, file_mime, file_body, read_stored) in cases {
+        let condition = match starts_with_prefix {
+            Some(prefix) => serde_json::json!(["starts-with", format!("${field}"), prefix]),
+            None => {
+                let mut exact = serde_json::Map::new();
+                exact.insert((*field).to_string(), serde_json::Value::String((*value).to_string()));
+                serde_json::Value::Object(exact)
+            }
+        };
 
-    let admin_client = env.create_s3_client();
-    admin_client.create_bucket().bucket(bucket).send().await?;
-    allow_anonymous_put_object(&admin_client, bucket).await?;
-
-    let policy = encode_post_policy(vec![
-        serde_json::json!({ "bucket": bucket }),
-        serde_json::json!({ "key": object_key }),
-        serde_json::json!(["starts-with", "$x-amz-meta-project", "alpha-"]),
-        serde_json::json!(["content-length-range", 0, 1024]),
-    ]);
-
-    let post_form = reqwest::multipart::Form::new()
-        .text("key", object_key.to_string())
-        .text("policy", policy)
-        .text("x-amz-meta-project", metadata_value)
-        .part(
-            "file",
-            reqwest::multipart::Part::bytes(expected_body.clone())
-                .file_name("upload.txt")
-                .mime_str("text/plain")?,
-        );
-
-    let post_resp = local_http_client()
-        .post(format!("{}/{}", env.url, bucket))
-        .multipart(post_form)
-        .send()
+        run_post_object_accept_case(
+            bucket,
+            object_key,
+            vec![
+                serde_json::json!({ "bucket": bucket }),
+                serde_json::json!({ "key": object_key }),
+                condition,
+                serde_json::json!(["content-length-range", 0, 1024]),
+            ],
+            (field, value),
+            file_mime,
+            file_body,
+            *read_stored,
+            case,
+        )
         .await?;
-
-    let status = post_resp.status();
-    let response_body = post_resp.text().await?;
-
-    assert_eq!(status, reqwest::StatusCode::NO_CONTENT);
-    assert!(response_body.is_empty(), "204 response should not contain a body, got: {response_body}");
-
-    let head = admin_client.head_object().bucket(bucket).key(object_key).send().await?;
-    let metadata = head.metadata().expect("head_object should expose uploaded metadata");
-    assert_eq!(metadata.get("project").map(String::as_str), Some(metadata_value));
-
-    Ok(())
-}
-
-#[tokio::test]
-async fn test_anonymous_post_object_accepts_content_type_field_exact_policy_match()
--> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-    init_logging();
-
-    let mut env = RustFSTestEnvironment::new().await?;
-    env.start_rustfs_server(vec![]).await?;
-
-    let bucket = "anon-post-policy-content-type-accept";
-    let object_key = "uploads/content-type-accept.txt";
-    let content_type = "text/plain";
-    let expected_body = b"post-policy-content-type-accept".to_vec();
-
-    let admin_client = env.create_s3_client();
-    admin_client.create_bucket().bucket(bucket).send().await?;
-    allow_anonymous_put_object(&admin_client, bucket).await?;
-
-    let policy = encode_post_policy(vec![
-        serde_json::json!({ "bucket": bucket }),
-        serde_json::json!({ "key": object_key }),
-        serde_json::json!({ "Content-Type": content_type }),
-        serde_json::json!(["content-length-range", 0, 1024]),
-    ]);
-
-    let post_form = reqwest::multipart::Form::new()
-        .text("key", object_key.to_string())
-        .text("policy", policy)
-        .text("Content-Type", content_type)
-        .part(
-            "file",
-            reqwest::multipart::Part::bytes(expected_body.clone())
-                .file_name("upload.txt")
-                .mime_str(content_type)?,
-        );
-
-    let post_resp = local_http_client()
-        .post(format!("{}/{}", env.url, bucket))
-        .multipart(post_form)
-        .send()
-        .await?;
-
-    let status = post_resp.status();
-    let response_body = post_resp.text().await?;
-
-    assert_eq!(status, reqwest::StatusCode::NO_CONTENT);
-    assert!(response_body.is_empty(), "204 response should not contain a body, got: {response_body}");
-
-    let head = admin_client.head_object().bucket(bucket).key(object_key).send().await?;
-    assert_eq!(head.content_type(), Some(content_type));
-
-    let get_out = admin_client.get_object().bucket(bucket).key(object_key).send().await?;
-    let uploaded = get_out.body.collect().await?.into_bytes();
-    assert_eq!(uploaded.as_ref(), expected_body.as_slice());
-
-    Ok(())
-}
-
-#[tokio::test]
-async fn test_anonymous_post_object_accepts_content_type_field_covered_by_starts_with()
--> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-    init_logging();
-
-    let mut env = RustFSTestEnvironment::new().await?;
-    env.start_rustfs_server(vec![]).await?;
-
-    let bucket = "anon-post-policy-content-type-accept";
-    let object_key = "uploads/content-type-object.txt";
-    let content_type = "image/png";
-    let expected_body = b"post-policy-content-type-body".to_vec();
-
-    let admin_client = env.create_s3_client();
-    admin_client.create_bucket().bucket(bucket).send().await?;
-    allow_anonymous_put_object(&admin_client, bucket).await?;
-
-    let policy = encode_post_policy(vec![
-        serde_json::json!({ "bucket": bucket }),
-        serde_json::json!({ "key": object_key }),
-        serde_json::json!(["starts-with", "$Content-Type", "image/"]),
-        serde_json::json!(["content-length-range", 0, 1024]),
-    ]);
-
-    let post_form = reqwest::multipart::Form::new()
-        .text("key", object_key.to_string())
-        .text("policy", policy)
-        .text("Content-Type", content_type)
-        .part(
-            "file",
-            reqwest::multipart::Part::bytes(expected_body.clone())
-                .file_name("upload.txt")
-                .mime_str(content_type)?,
-        );
-
-    let post_resp = local_http_client()
-        .post(format!("{}/{}", env.url, bucket))
-        .multipart(post_form)
-        .send()
-        .await?;
-
-    let status = post_resp.status();
-    let response_body = post_resp.text().await?;
-
-    assert_eq!(status, reqwest::StatusCode::NO_CONTENT);
-    assert!(response_body.is_empty(), "204 response should not contain a body, got: {response_body}");
-
-    let head = admin_client.head_object().bucket(bucket).key(object_key).send().await?;
-    assert_eq!(head.content_type(), Some(content_type));
-
-    let get_out = admin_client.get_object().bucket(bucket).key(object_key).send().await?;
-    let uploaded = get_out.body.collect().await?.into_bytes();
-    assert_eq!(uploaded.as_ref(), expected_body.as_slice());
-
-    Ok(())
-}
-
-#[tokio::test]
-async fn test_anonymous_post_object_accepts_content_disposition_field_exact_policy_match()
--> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-    init_logging();
-
-    let mut env = RustFSTestEnvironment::new().await?;
-    env.start_rustfs_server(vec![]).await?;
-
-    let bucket = "anon-post-policy-disposition-accept";
-    let object_key = "uploads/disposition-object.txt";
-    let content_disposition = "attachment; filename=\"upload.txt\"";
-    let expected_body = b"post-policy-disposition-body".to_vec();
-
-    let admin_client = env.create_s3_client();
-    admin_client.create_bucket().bucket(bucket).send().await?;
-    allow_anonymous_put_object(&admin_client, bucket).await?;
-
-    let policy = encode_post_policy(vec![
-        serde_json::json!({ "bucket": bucket }),
-        serde_json::json!({ "key": object_key }),
-        serde_json::json!({ "Content-Disposition": content_disposition }),
-        serde_json::json!(["content-length-range", 0, 1024]),
-    ]);
-
-    let post_form = reqwest::multipart::Form::new()
-        .text("key", object_key.to_string())
-        .text("policy", policy)
-        .text("Content-Disposition", content_disposition)
-        .part(
-            "file",
-            reqwest::multipart::Part::bytes(expected_body.clone())
-                .file_name("upload.txt")
-                .mime_str("text/plain")?,
-        );
-
-    let post_resp = local_http_client()
-        .post(format!("{}/{}", env.url, bucket))
-        .multipart(post_form)
-        .send()
-        .await?;
-
-    let status = post_resp.status();
-    let response_body = post_resp.text().await?;
-
-    assert_eq!(status, reqwest::StatusCode::NO_CONTENT);
-    assert!(response_body.is_empty(), "204 response should not contain a body, got: {response_body}");
-
-    let head = admin_client.head_object().bucket(bucket).key(object_key).send().await?;
-    assert_eq!(head.content_disposition(), Some(content_disposition));
-
-    let get_out = admin_client.get_object().bucket(bucket).key(object_key).send().await?;
-    let uploaded = get_out.body.collect().await?.into_bytes();
-    assert_eq!(uploaded.as_ref(), expected_body.as_slice());
-
-    Ok(())
-}
-
-#[tokio::test]
-async fn test_anonymous_post_object_accepts_cache_control_field_exact_policy_match()
--> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-    init_logging();
-
-    let mut env = RustFSTestEnvironment::new().await?;
-    env.start_rustfs_server(vec![]).await?;
-
-    let bucket = "anon-post-policy-cache-control-accept";
-    let object_key = "uploads/cache-control-object.txt";
-    let cache_control = "max-age=60";
-    let expected_body = b"post-policy-cache-control-body".to_vec();
-
-    let admin_client = env.create_s3_client();
-    admin_client.create_bucket().bucket(bucket).send().await?;
-    allow_anonymous_put_object(&admin_client, bucket).await?;
-
-    let policy = encode_post_policy(vec![
-        serde_json::json!({ "bucket": bucket }),
-        serde_json::json!({ "key": object_key }),
-        serde_json::json!({ "Cache-Control": cache_control }),
-        serde_json::json!(["content-length-range", 0, 1024]),
-    ]);
-
-    let post_form = reqwest::multipart::Form::new()
-        .text("key", object_key.to_string())
-        .text("policy", policy)
-        .text("Cache-Control", cache_control)
-        .part(
-            "file",
-            reqwest::multipart::Part::bytes(expected_body.clone())
-                .file_name("upload.txt")
-                .mime_str("text/plain")?,
-        );
-
-    let post_resp = local_http_client()
-        .post(format!("{}/{}", env.url, bucket))
-        .multipart(post_form)
-        .send()
-        .await?;
-
-    let status = post_resp.status();
-    let response_body = post_resp.text().await?;
-
-    assert_eq!(status, reqwest::StatusCode::NO_CONTENT);
-    assert!(response_body.is_empty(), "204 response should not contain a body, got: {response_body}");
-
-    let head = admin_client.head_object().bucket(bucket).key(object_key).send().await?;
-    assert_eq!(head.cache_control(), Some(cache_control));
-
-    let get_out = admin_client.get_object().bucket(bucket).key(object_key).send().await?;
-    let uploaded = get_out.body.collect().await?.into_bytes();
-    assert_eq!(uploaded.as_ref(), expected_body.as_slice());
-
-    Ok(())
-}
-
-#[tokio::test]
-async fn test_anonymous_post_object_accepts_content_language_field_exact_policy_match()
--> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-    init_logging();
-
-    let mut env = RustFSTestEnvironment::new().await?;
-    env.start_rustfs_server(vec![]).await?;
-
-    let bucket = "anon-post-policy-content-language-accept";
-    let object_key = "uploads/content-language-object.txt";
-    let content_language = "en-US";
-    let expected_body = b"post-policy-content-language-body".to_vec();
-
-    let admin_client = env.create_s3_client();
-    admin_client.create_bucket().bucket(bucket).send().await?;
-    allow_anonymous_put_object(&admin_client, bucket).await?;
-
-    let policy = encode_post_policy(vec![
-        serde_json::json!({ "bucket": bucket }),
-        serde_json::json!({ "key": object_key }),
-        serde_json::json!({ "Content-Language": content_language }),
-        serde_json::json!(["content-length-range", 0, 1024]),
-    ]);
-
-    let post_form = reqwest::multipart::Form::new()
-        .text("key", object_key.to_string())
-        .text("policy", policy)
-        .text("Content-Language", content_language)
-        .part(
-            "file",
-            reqwest::multipart::Part::bytes(expected_body.clone())
-                .file_name("upload.txt")
-                .mime_str("text/plain")?,
-        );
-
-    let post_resp = local_http_client()
-        .post(format!("{}/{}", env.url, bucket))
-        .multipart(post_form)
-        .send()
-        .await?;
-
-    let status = post_resp.status();
-    let response_body = post_resp.text().await?;
-
-    assert_eq!(status, reqwest::StatusCode::NO_CONTENT);
-    assert!(response_body.is_empty(), "204 response should not contain a body, got: {response_body}");
-
-    let head = admin_client.head_object().bucket(bucket).key(object_key).send().await?;
-    assert_eq!(head.content_language(), Some(content_language));
-
-    let get_out = admin_client.get_object().bucket(bucket).key(object_key).send().await?;
-    let uploaded = get_out.body.collect().await?.into_bytes();
-    assert_eq!(uploaded.as_ref(), expected_body.as_slice());
-
-    Ok(())
-}
-
-#[tokio::test]
-async fn test_anonymous_post_object_accepts_content_encoding_field_exact_policy_match()
--> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-    init_logging();
-
-    let mut env = RustFSTestEnvironment::new().await?;
-    env.start_rustfs_server(vec![]).await?;
-
-    let bucket = "anon-post-policy-content-encoding-accept";
-    let object_key = "uploads/content-encoding-object.txt";
-    let content_encoding = "gzip";
-    let expected_body = b"post-policy-content-encoding-body".to_vec();
-
-    let admin_client = env.create_s3_client();
-    admin_client.create_bucket().bucket(bucket).send().await?;
-    allow_anonymous_put_object(&admin_client, bucket).await?;
-
-    let policy = encode_post_policy(vec![
-        serde_json::json!({ "bucket": bucket }),
-        serde_json::json!({ "key": object_key }),
-        serde_json::json!({ "Content-Encoding": content_encoding }),
-        serde_json::json!(["content-length-range", 0, 1024]),
-    ]);
-
-    let post_form = reqwest::multipart::Form::new()
-        .text("key", object_key.to_string())
-        .text("policy", policy)
-        .text("Content-Encoding", content_encoding)
-        .part(
-            "file",
-            reqwest::multipart::Part::bytes(expected_body.clone())
-                .file_name("upload.txt")
-                .mime_str("text/plain")?,
-        );
-
-    let post_resp = local_http_client()
-        .post(format!("{}/{}", env.url, bucket))
-        .multipart(post_form)
-        .send()
-        .await?;
-
-    let status = post_resp.status();
-    let response_body = post_resp.text().await?;
-
-    assert_eq!(status, reqwest::StatusCode::NO_CONTENT);
-    assert!(response_body.is_empty(), "204 response should not contain a body, got: {response_body}");
-
-    let head = admin_client.head_object().bucket(bucket).key(object_key).send().await?;
-    assert_eq!(head.content_encoding(), Some(content_encoding));
-
-    let get_out = admin_client.get_object().bucket(bucket).key(object_key).send().await?;
-    let uploaded = get_out.body.collect().await?.into_bytes();
-    assert_eq!(uploaded.as_ref(), expected_body.as_slice());
-
-    Ok(())
-}
-
-#[tokio::test]
-async fn test_anonymous_post_object_accepts_website_redirect_location_exact_policy_match()
--> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-    init_logging();
-
-    let mut env = RustFSTestEnvironment::new().await?;
-    env.start_rustfs_server(vec![]).await?;
-
-    let bucket = "anon-post-policy-website-redirect-accept";
-    let object_key = "uploads/website-redirect-object.txt";
-    let website_redirect_location = "/docs/landing.html";
-    let expected_body = b"post-policy-website-redirect-body".to_vec();
-
-    let admin_client = env.create_s3_client();
-    admin_client.create_bucket().bucket(bucket).send().await?;
-    allow_anonymous_put_object(&admin_client, bucket).await?;
-
-    let policy = encode_post_policy(vec![
-        serde_json::json!({ "bucket": bucket }),
-        serde_json::json!({ "key": object_key }),
-        serde_json::json!({ "x-amz-website-redirect-location": website_redirect_location }),
-        serde_json::json!(["content-length-range", 0, 1024]),
-    ]);
-
-    let post_form = reqwest::multipart::Form::new()
-        .text("key", object_key.to_string())
-        .text("policy", policy)
-        .text("x-amz-website-redirect-location", website_redirect_location)
-        .part(
-            "file",
-            reqwest::multipart::Part::bytes(expected_body.clone())
-                .file_name("upload.txt")
-                .mime_str("text/plain")?,
-        );
-
-    let post_resp = local_http_client()
-        .post(format!("{}/{}", env.url, bucket))
-        .multipart(post_form)
-        .send()
-        .await?;
-
-    let status = post_resp.status();
-    let response_body = post_resp.text().await?;
-
-    assert_eq!(status, reqwest::StatusCode::NO_CONTENT);
-    assert!(response_body.is_empty(), "204 response should not contain a body, got: {response_body}");
-
-    let head = admin_client.head_object().bucket(bucket).key(object_key).send().await?;
-    assert_eq!(head.website_redirect_location(), Some(website_redirect_location));
-
-    let get_out = admin_client.get_object().bucket(bucket).key(object_key).send().await?;
-    let uploaded = get_out.body.collect().await?.into_bytes();
-    assert_eq!(uploaded.as_ref(), expected_body.as_slice());
-
-    Ok(())
-}
-
-#[tokio::test]
-async fn test_anonymous_post_object_accepts_expires_field_exact_policy_match()
--> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-    init_logging();
-
-    let mut env = RustFSTestEnvironment::new().await?;
-    env.start_rustfs_server(vec![]).await?;
-
-    let bucket = "anon-post-policy-expires-accept";
-    let object_key = "uploads/expires-object.txt";
-    let expires = "Wed, 21 Oct 2037 07:28:00 GMT";
-    let expected_body = b"post-policy-expires-body".to_vec();
-
-    let admin_client = env.create_s3_client();
-    admin_client.create_bucket().bucket(bucket).send().await?;
-    allow_anonymous_put_object(&admin_client, bucket).await?;
-
-    let policy = encode_post_policy(vec![
-        serde_json::json!({ "bucket": bucket }),
-        serde_json::json!({ "key": object_key }),
-        serde_json::json!({ "Expires": expires }),
-        serde_json::json!(["content-length-range", 0, 1024]),
-    ]);
-
-    let post_form = reqwest::multipart::Form::new()
-        .text("key", object_key.to_string())
-        .text("policy", policy)
-        .text("Expires", expires)
-        .part(
-            "file",
-            reqwest::multipart::Part::bytes(expected_body.clone())
-                .file_name("upload.txt")
-                .mime_str("text/plain")?,
-        );
-
-    let post_resp = local_http_client()
-        .post(format!("{}/{}", env.url, bucket))
-        .multipart(post_form)
-        .send()
-        .await?;
-
-    let status = post_resp.status();
-    let response_body = post_resp.text().await?;
-
-    assert_eq!(status, reqwest::StatusCode::NO_CONTENT);
-    assert!(response_body.is_empty(), "204 response should not contain a body, got: {response_body}");
-
-    let head = admin_client.head_object().bucket(bucket).key(object_key).send().await?;
-    assert_eq!(head.expires_string(), Some(expires));
-
-    let get_out = admin_client.get_object().bucket(bucket).key(object_key).send().await?;
-    let uploaded = get_out.body.collect().await?.into_bytes();
-    assert_eq!(uploaded.as_ref(), expected_body.as_slice());
+    }
 
     Ok(())
 }
@@ -3440,64 +3123,6 @@ async fn test_anonymous_post_object_accepts_tagging_field_exact_policy_match()
     Ok(())
 }
 
-#[tokio::test]
-async fn test_anonymous_post_object_accepts_metadata_field_exact_policy_match()
--> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-    init_logging();
-
-    let mut env = RustFSTestEnvironment::new().await?;
-    env.start_rustfs_server(vec![]).await?;
-
-    let bucket = "anon-post-policy-meta-exact-accept";
-    let object_key = "uploads/meta-exact-accept-object.txt";
-    let metadata_value = "alpha-demo";
-    let expected_body = b"post-policy-meta-exact-body".to_vec();
-
-    let admin_client = env.create_s3_client();
-    admin_client.create_bucket().bucket(bucket).send().await?;
-    allow_anonymous_put_object(&admin_client, bucket).await?;
-
-    let policy = encode_post_policy(vec![
-        serde_json::json!({ "bucket": bucket }),
-        serde_json::json!({ "key": object_key }),
-        serde_json::json!({ "x-amz-meta-project": metadata_value }),
-        serde_json::json!(["content-length-range", 0, 1024]),
-    ]);
-
-    let post_form = reqwest::multipart::Form::new()
-        .text("key", object_key.to_string())
-        .text("policy", policy)
-        .text("x-amz-meta-project", metadata_value)
-        .part(
-            "file",
-            reqwest::multipart::Part::bytes(expected_body.clone())
-                .file_name("upload.txt")
-                .mime_str("text/plain")?,
-        );
-
-    let post_resp = local_http_client()
-        .post(format!("{}/{}", env.url, bucket))
-        .multipart(post_form)
-        .send()
-        .await?;
-
-    let status = post_resp.status();
-    let response_body = post_resp.text().await?;
-
-    assert_eq!(status, reqwest::StatusCode::NO_CONTENT);
-    assert!(response_body.is_empty(), "204 response should not contain a body, got: {response_body}");
-
-    let head = admin_client.head_object().bucket(bucket).key(object_key).send().await?;
-    let metadata = head.metadata().expect("head_object should expose uploaded metadata");
-    assert_eq!(metadata.get("project").map(String::as_str), Some(metadata_value));
-
-    let get_out = admin_client.get_object().bucket(bucket).key(object_key).send().await?;
-    let uploaded = get_out.body.collect().await?.into_bytes();
-    assert_eq!(uploaded.as_ref(), expected_body.as_slice());
-
-    Ok(())
-}
-
 #[tokio::test]
 async fn test_anonymous_post_object_allows_x_ignore_fields_outside_policy_conditions()
 -> Result<(), Box<dyn std::error::Error + Send + Sync>> {

crates/ecstore/src/data_usage/mod.rs

@@ -19,7 +19,7 @@ pub mod local_snapshot;
 use crate::storage_api_contracts::{
     bucket::{BucketOperations as _, BucketOptions},
     list::{ListOperations as _, StorageListObjectVersionsInfo},
-    object::{EcstoreObjectIO, HTTPPreconditions, ObjectIO as _, ObjectOperations as _},
+    object::{EcstoreObjectIO, HTTPPreconditions, ObjectOperations as _},
 };
 use crate::{
     bucket::{metadata_sys::get_replication_config, versioning::VersioningApi as _, versioning_sys::BucketVersioningSys},
@@ -2009,80 +2009,102 @@ pub async fn apply_bucket_usage_memory_overlay(data_usage_info: &mut DataUsageIn
 }
 
 // Helper functions for DataUsageCache operations
-pub async fn load_data_usage_cache(store: &crate::set_disk::SetDisks, name: &str) -> crate::error::Result<DataUsageCache> {
-    use crate::disk::{BUCKET_META_PREFIX, RUSTFS_META_BUCKET};
+
+/// How many times `load_data_usage_cache` tries a read that failed for a
+/// transient reason before giving up.
+const DATA_USAGE_CACHE_LOAD_ATTEMPTS: usize = 5;
+const DATA_USAGE_CACHE_LOAD_BASE_DELAY: std::time::Duration = std::time::Duration::from_millis(100);
+const DATA_USAGE_CACHE_LOAD_MAX_DELAY: std::time::Duration = std::time::Duration::from_millis(1_000);
+
+/// Result of one attempt at reading a data-usage cache object.
+enum DataUsageCacheRead {
+    Loaded(DataUsageCache),
+    /// The object is definitively not there, so retrying cannot turn the read
+    /// into a hit.
+    Absent,
+}
+
+/// True when the error means the cache object does not exist, as opposed to a
+/// transient failure that is worth another attempt.
+///
+/// `SetDisks::get_object_reader` runs its failures through `to_object_err`,
+/// which rewrites `FileNotFound` to `ObjectNotFound` and `VolumeNotFound` to
+/// `BucketNotFound`, so those are the variants that actually arrive here. The
+/// raw pair is matched too because callers reading through a different layer
+/// can still surface it.
+fn is_data_usage_cache_absent(err: &Error) -> bool {
+    matches!(
+        err,
+        Error::FileNotFound | Error::VolumeNotFound | Error::ObjectNotFound(..) | Error::BucketNotFound(..)
+    )
+}
+
+async fn read_data_usage_cache_object<S>(store: &S, key: &str) -> crate::error::Result<DataUsageCacheRead>
+where
+    S: EcstoreObjectIO,
+{
+    use crate::disk::RUSTFS_META_BUCKET;
     use crate::object_api::ObjectOptions;
     use http::HeaderMap;
-    use rand::RngExt;
-    use std::path::Path;
-    use std::time::Duration;
-    use tokio::time::sleep;
 
-    let mut d = DataUsageCache::default();
-    let mut retries = 0;
-    while retries < 5 {
-        let path = Path::new(BUCKET_META_PREFIX).join(name);
-        match store
-            .get_object_reader(
-                RUSTFS_META_BUCKET,
-                path.to_str().unwrap(),
-                None,
-                HeaderMap::new(),
-                &ObjectOptions {
-                    no_lock: true,
-                    ..Default::default()
-                },
-            )
-            .await
-        {
-            Ok(mut reader) => {
-                if let Ok(info) = DataUsageCache::unmarshal(&reader.read_all().await?) {
-                    d = info
-                }
-                break;
-            }
-            Err(err) => match err {
-                Error::FileNotFound | Error::VolumeNotFound => {
-                    match store
-                        .get_object_reader(
-                            RUSTFS_META_BUCKET,
-                            name,
-                            None,
-                            HeaderMap::new(),
-                            &ObjectOptions {
-                                no_lock: true,
-                                ..Default::default()
-                            },
-                        )
-                        .await
-                    {
-                        Ok(mut reader) => {
-                            if let Ok(info) = DataUsageCache::unmarshal(&reader.read_all().await?) {
-                                d = info
-                            }
-                            break;
-                        }
-                        Err(_) => match err {
-                            Error::FileNotFound | Error::VolumeNotFound => {
-                                break;
-                            }
-                            _ => {}
-                        },
-                    }
-                }
-                _ => {
-                    break;
-                }
+    match store
+        .get_object_reader(
+            RUSTFS_META_BUCKET,
+            key,
+            None,
+            HeaderMap::new(),
+            &ObjectOptions {
+                no_lock: true,
+                ..Default::default()
             },
-        }
-        retries += 1;
-        let dur = {
-            let mut rng = rand::rng();
-            rng.random_range(0..1_000)
-        };
-        sleep(Duration::from_millis(dur)).await;
+        )
+        .await
+    {
+        // A cache object that fails to decode is treated as absent rather than
+        // as an error: a corrupt cache should not stall the caller, and the
+        // next scanner pass rewrites it.
+        Ok(mut reader) => Ok(DataUsageCache::unmarshal(&reader.read_all().await?)
+            .map(DataUsageCacheRead::Loaded)
+            .unwrap_or(DataUsageCacheRead::Absent)),
+        Err(err) if is_data_usage_cache_absent(&err) => Ok(DataUsageCacheRead::Absent),
+        Err(err) => Err(err),
     }
-    Ok(d)
+}
+
+/// Load a data-usage cache, preferring the prefixed key and falling back to the
+/// legacy unprefixed one.
+///
+/// A cache that is absent under both keys yields an empty cache; a transient
+/// read failure is retried with capped, jittered backoff and surfaces as an
+/// error once the attempts are exhausted.
+pub(crate) async fn load_data_usage_cache<S>(store: &S, name: &str) -> crate::error::Result<DataUsageCache>
+where
+    S: EcstoreObjectIO,
+{
+    use crate::disk::BUCKET_META_PREFIX;
+    use std::path::Path;
+
+    let prefixed = Path::new(BUCKET_META_PREFIX).join(name);
+    let prefixed = prefixed
+        .to_str()
+        .ok_or_else(|| Error::other("data usage cache path is not valid UTF-8"))?
+        .to_owned();
+
+    rustfs_utils::retry::retry_with_backoff(
+        || async {
+            match read_data_usage_cache_object(store, &prefixed).await? {
+                DataUsageCacheRead::Loaded(cache) => Ok(cache),
+                DataUsageCacheRead::Absent => match read_data_usage_cache_object(store, name).await? {
+                    DataUsageCacheRead::Loaded(cache) => Ok(cache),
+                    DataUsageCacheRead::Absent => Ok(DataUsageCache::default()),
+                },
+            }
+        },
+        DATA_USAGE_CACHE_LOAD_ATTEMPTS,
+        DATA_USAGE_CACHE_LOAD_BASE_DELAY,
+        DATA_USAGE_CACHE_LOAD_MAX_DELAY,
+    )
+    .await
 }
 
 /// Persist the current in-memory compression total to the backend.
@@ -2220,6 +2242,7 @@ pub async fn init_compression_total_memory_from_backend(store: Arc<ECStore>) {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::storage_api_contracts::object::ObjectIO as _;
     use rustfs_data_usage::BucketUsageInfo;
     use rustfs_lock::{LocalClient, LockRequest, LockType, NamespaceLock, ObjectKey};
     use serial_test::serial;
@@ -2452,6 +2475,148 @@ mod tests {
         }
     }
 
+    /// Minimal ObjectIO backing `load_data_usage_cache` tests: records the keys
+    /// read and fails the first N reads with a transient (non-absence) error.
+    #[derive(Debug, Default)]
+    struct UsageCacheReadStore {
+        transient_failures: Mutex<usize>,
+        reads: Mutex<Vec<String>>,
+    }
+
+    impl UsageCacheReadStore {
+        fn failing_first(n: usize) -> Self {
+            Self {
+                transient_failures: Mutex::new(n),
+                reads: Mutex::new(Vec::new()),
+            }
+        }
+
+        async fn read_keys(&self) -> Vec<String> {
+            self.reads.lock().await.clone()
+        }
+    }
+
+    #[async_trait::async_trait]
+    impl crate::storage_api_contracts::object::ObjectIO for UsageCacheReadStore {
+        type Error = Error;
+        type RangeSpec = crate::storage_api_contracts::range::HTTPRangeSpec;
+        type HeaderMap = http::HeaderMap;
+        type ObjectOptions = ObjectOptions;
+        type ObjectInfo = ObjectInfo;
+        type GetObjectReader = crate::object_api::GetObjectReader;
+        type PutObjectReader = PutObjReader;
+
+        async fn get_object_reader(
+            &self,
+            _bucket: &str,
+            object: &str,
+            _range: Option<Self::RangeSpec>,
+            _h: Self::HeaderMap,
+            _opts: &Self::ObjectOptions,
+        ) -> Result<Self::GetObjectReader, Self::Error> {
+            self.reads.lock().await.push(object.to_string());
+            let mut remaining = self.transient_failures.lock().await;
+            if *remaining > 0 {
+                *remaining -= 1;
+                return Err(Error::other("transient read failure"));
+            }
+            // `SetDisks::get_object_reader` reports a missing object through
+            // `to_object_err`, so the absence that reaches the caller is
+            // `ObjectNotFound`, not the raw `FileNotFound`.
+            Err(Error::ObjectNotFound(RUSTFS_META_BUCKET.to_string(), object.to_string()))
+        }
+
+        async fn put_object(
+            &self,
+            _bucket: &str,
+            _object: &str,
+            _data: &mut Self::PutObjectReader,
+            _opts: &Self::ObjectOptions,
+        ) -> Result<Self::ObjectInfo, Self::Error> {
+            unimplemented!("load_data_usage_cache never writes")
+        }
+    }
+
+    fn prefixed_usage_key(name: &str) -> String {
+        std::path::Path::new(crate::disk::BUCKET_META_PREFIX)
+            .join(name)
+            .to_str()
+            .expect("utf-8 path")
+            .to_string()
+    }
+
+    #[test]
+    fn data_usage_cache_absence_covers_the_variants_that_actually_arrive() {
+        // `to_object_err` rewrites the raw storage variants before they reach
+        // `load_data_usage_cache`; classifying only the raw pair would treat a
+        // missing cache as a transient failure and retry it.
+        assert!(is_data_usage_cache_absent(&Error::ObjectNotFound(
+            "bucket".to_string(),
+            "object".to_string()
+        )));
+        assert!(is_data_usage_cache_absent(&Error::BucketNotFound("bucket".to_string())));
+        assert!(is_data_usage_cache_absent(&Error::FileNotFound));
+        assert!(is_data_usage_cache_absent(&Error::VolumeNotFound));
+
+        assert!(!is_data_usage_cache_absent(&Error::other("transient read failure")));
+        assert!(!is_data_usage_cache_absent(&Error::DiskNotFound));
+    }
+
+    #[tokio::test]
+    async fn load_data_usage_cache_treats_absence_as_an_empty_cache_without_retrying() {
+        let name = "usage-cache";
+        let store = UsageCacheReadStore::default();
+
+        let cache = load_data_usage_cache(&store, name).await.expect("absence is not an error");
+
+        assert!(cache.cache.is_empty());
+        assert_eq!(
+            store.read_keys().await,
+            vec![prefixed_usage_key(name), name.to_string()],
+            "the prefixed key is tried first, then the legacy one, and neither absence is retried"
+        );
+    }
+
+    #[tokio::test]
+    async fn load_data_usage_cache_retries_a_transient_failure() {
+        let name = "usage-cache";
+        // Two transient failures, then the object reads as absent.
+        let store = UsageCacheReadStore::failing_first(2);
+
+        let cache = load_data_usage_cache(&store, name)
+            .await
+            .expect("retry should reach the absent read");
+
+        assert!(cache.cache.is_empty());
+        assert_eq!(
+            store.read_keys().await,
+            vec![
+                prefixed_usage_key(name),
+                prefixed_usage_key(name),
+                prefixed_usage_key(name),
+                name.to_string(),
+            ],
+            "a transient failure retries the prefixed read rather than falling through"
+        );
+    }
+
+    #[tokio::test]
+    async fn load_data_usage_cache_surfaces_a_persistent_failure() {
+        let name = "usage-cache";
+        let store = UsageCacheReadStore::failing_first(usize::MAX);
+
+        let err = load_data_usage_cache(&store, name)
+            .await
+            .expect_err("an exhausted retry must not be reported as an empty cache");
+
+        assert!(err.to_string().contains("transient read failure"));
+        assert_eq!(
+            store.read_keys().await.len(),
+            DATA_USAGE_CACHE_LOAD_ATTEMPTS,
+            "every attempt is used before giving up"
+        );
+    }
+
     async fn clear_usage_memory_cache_for_test() {
         memory_cache().write().await.clear();
         *cache_updating().write().await = false;

crates/ecstore/src/diagnostics/admin_server_info.rs

@@ -679,7 +679,7 @@ async fn get_pools_info(all_disks: &[Disk]) -> Result<HashMap<i32, HashMap<i32,
         if erasure_set.id == 0 {
             erasure_set.id = d.set_index;
             match load_data_usage_cache(
-                &store.pools[d.pool_index as usize].disk_set[d.set_index as usize].clone(),
+                store.pools[d.pool_index as usize].disk_set[d.set_index as usize].as_ref(),
                 DATA_USAGE_CACHE_NAME,
             )
             .await

crates/heal/src/heal/replacement_readiness.rs

@@ -157,4 +157,219 @@ mod tests {
         )
         .await;
     }
+
+    #[cfg(target_os = "linux")]
+    mod linux_privileged_tests {
+        use super::*;
+        use std::error::Error;
+        use std::path::Path;
+        use std::process::Command;
+
+        const ENABLE_ENV: &str = "RUSTFS_PRIVILEGED_MOUNT_READINESS_TESTS";
+        const NAMESPACE_ENV: &str = "RUSTFS_PRIVILEGED_MOUNT_READINESS_TESTS_IN_NAMESPACE";
+        const MOUNT_SIZE: &str = "size=32m,mode=0700";
+
+        struct MountGuard {
+            mounts: Vec<std::path::PathBuf>,
+        }
+
+        impl MountGuard {
+            fn new() -> Result<Self, Box<dyn Error + Send + Sync>> {
+                run_command("mount", &["--make-rprivate", "/"])?;
+                Ok(Self { mounts: Vec::new() })
+            }
+
+            fn mount_tmpfs(&mut self, target: &Path, label: &str) -> Result<(), Box<dyn Error + Send + Sync>> {
+                mount_tmpfs(target, label)?;
+                self.mounts.push(target.to_path_buf());
+                Ok(())
+            }
+
+            fn mount_bind(&mut self, source: &Path, target: &Path) -> Result<(), Box<dyn Error + Send + Sync>> {
+                mount_bind(source, target)?;
+                self.mounts.push(target.to_path_buf());
+                Ok(())
+            }
+        }
+
+        impl Drop for MountGuard {
+            fn drop(&mut self) {
+                for mount in self.mounts.iter().rev() {
+                    let _ = detach_mount(mount);
+                }
+            }
+        }
+
+        fn run_command(program: &str, args: &[&str]) -> Result<(), Box<dyn Error + Send + Sync>> {
+            let output = Command::new(program).args(args).output()?;
+            if output.status.success() {
+                return Ok(());
+            }
+            Err(format!(
+                "{program} {} failed with status {}: stdout={} stderr={}",
+                args.join(" "),
+                output.status,
+                String::from_utf8_lossy(&output.stdout),
+                String::from_utf8_lossy(&output.stderr)
+            )
+            .into())
+        }
+
+        fn path_to_string(path: &Path, label: &str) -> Result<String, Box<dyn Error + Send + Sync>> {
+            path.to_str()
+                .map(str::to_owned)
+                .ok_or_else(|| format!("{label} path is not UTF-8: {path:?}").into())
+        }
+
+        fn mount_tmpfs(target: &Path, label: &str) -> Result<(), Box<dyn Error + Send + Sync>> {
+            let target = path_to_string(target, "tmpfs target")?;
+            run_command("mount", &["-t", "tmpfs", "-o", MOUNT_SIZE, label, &target])
+        }
+
+        fn mount_bind(source: &Path, target: &Path) -> Result<(), Box<dyn Error + Send + Sync>> {
+            let source = path_to_string(source, "bind source")?;
+            let target = path_to_string(target, "bind target")?;
+            run_command("mount", &["--bind", &source, &target])
+        }
+
+        fn detach_mount(target: &Path) -> Result<(), Box<dyn Error + Send + Sync>> {
+            let target = path_to_string(target, "umount target")?;
+            run_command("umount", &[&target])
+        }
+
+        fn privileged_enabled() -> Result<bool, Box<dyn Error + Send + Sync>> {
+            let enabled = std::env::var(ENABLE_ENV)
+                .ok()
+                .is_some_and(|value| matches!(value.as_str(), "1" | "true" | "TRUE" | "yes" | "YES"));
+            if !enabled {
+                return Ok(false);
+            }
+            Ok(true)
+        }
+
+        fn run_current_test_in_mount_namespace() -> Result<(), Box<dyn Error + Send + Sync>> {
+            let test_name = std::thread::current()
+                .name()
+                .ok_or("privileged mount readiness test thread is unnamed")?
+                .to_owned();
+            let test_binary = std::env::current_exe()?;
+            let status = Command::new("unshare")
+                .arg("--mount")
+                .arg("--propagation")
+                .arg("private")
+                .arg("--")
+                .arg(test_binary)
+                .arg("--exact")
+                .arg(test_name)
+                .arg("--ignored")
+                .arg("--nocapture")
+                .env(NAMESPACE_ENV, "1")
+                .status()?;
+            if status.success() {
+                return Ok(());
+            }
+            Err(format!("{ENABLE_ENV}=1 requires Linux root or CAP_SYS_ADMIN; unshare exited with status {status}").into())
+        }
+
+        fn run_privileged_mount_test<F, Fut>(test: F) -> Result<(), Box<dyn Error + Send + Sync>>
+        where
+            F: FnOnce(MountGuard) -> Fut + Send + 'static,
+            Fut: std::future::Future<Output = Result<(), Box<dyn Error + Send + Sync>>> + 'static,
+        {
+            if !privileged_enabled()? {
+                return Ok(());
+            }
+            if std::env::var_os(NAMESPACE_ENV).is_none() {
+                return run_current_test_in_mount_namespace();
+            }
+
+            let guard = MountGuard::new()?;
+            let runtime = tokio::runtime::Builder::new_current_thread().enable_all().build()?;
+            runtime.block_on(test(guard))
+        }
+
+        #[test]
+        #[ignore = "requires Linux root/CAP_SYS_ADMIN and RUSTFS_PRIVILEGED_MOUNT_READINESS_TESTS=1"]
+        fn auto_replacement_readiness_accepts_an_independent_mount() -> Result<(), Box<dyn Error + Send + Sync>> {
+            run_privileged_mount_test(|mut mounts| async move {
+                let temp = TempDir::new().expect("temporary replacement roots should be created");
+                let target = temp.path().join("target");
+                let sibling = temp.path().join("sibling");
+                std::fs::create_dir(&target).expect("target mountpoint should be created");
+                std::fs::create_dir(&sibling).expect("sibling mountpoint should be created");
+                mounts.mount_tmpfs(&target, "rustfs-readiness-target")?;
+                mounts.mount_tmpfs(&sibling, "rustfs-readiness-sibling")?;
+
+                let target_endpoint = Endpoint::try_from(target.to_string_lossy().as_ref())?;
+                let sibling_endpoint = Endpoint::try_from(sibling.to_string_lossy().as_ref())?;
+                let target_disk = new_disk(
+                    &target_endpoint,
+                    &DiskOption {
+                        cleanup: false,
+                        health_check: false,
+                    },
+                )
+                .await?;
+                let sibling_disk = new_disk(
+                    &sibling_endpoint,
+                    &DiskOption {
+                        cleanup: false,
+                        health_check: false,
+                    },
+                )
+                .await?;
+
+                let identity = auto_replacement_target_identity(&target_disk, &[target_disk.clone(), sibling_disk.clone()]).await;
+                assert!(
+                    identity.is_some(),
+                    "a separately mounted replacement target with no sibling device overlap must be admitted"
+                );
+                Ok(())
+            })
+        }
+
+        #[test]
+        #[ignore = "requires Linux root/CAP_SYS_ADMIN and RUSTFS_PRIVILEGED_MOUNT_READINESS_TESTS=1"]
+        fn auto_replacement_readiness_rejects_a_same_device_sibling_bind_mount() -> Result<(), Box<dyn Error + Send + Sync>> {
+            run_privileged_mount_test(|mut mounts| async move {
+                let temp = TempDir::new().expect("temporary replacement roots should be created");
+                let source = temp.path().join("source");
+                let target = temp.path().join("target");
+                let sibling = temp.path().join("sibling");
+                std::fs::create_dir(&source).expect("source mountpoint should be created");
+                std::fs::create_dir(&target).expect("target mountpoint should be created");
+                std::fs::create_dir(&sibling).expect("sibling mountpoint should be created");
+                mounts.mount_tmpfs(&source, "rustfs-readiness-shared-source")?;
+                mounts.mount_bind(&source, &target)?;
+                mounts.mount_bind(&source, &sibling)?;
+
+                let target_endpoint = Endpoint::try_from(target.to_string_lossy().as_ref())?;
+                let sibling_endpoint = Endpoint::try_from(sibling.to_string_lossy().as_ref())?;
+                let target_disk = new_disk(
+                    &target_endpoint,
+                    &DiskOption {
+                        cleanup: false,
+                        health_check: false,
+                    },
+                )
+                .await?;
+                let sibling_disk = new_disk(
+                    &sibling_endpoint,
+                    &DiskOption {
+                        cleanup: false,
+                        health_check: false,
+                    },
+                )
+                .await?;
+
+                assert!(
+                    auto_replacement_target_identity(&target_disk, &[target_disk.clone(), sibling_disk.clone()])
+                        .await
+                        .is_none(),
+                    "replacement readiness must reject a target sharing its physical device with a sibling endpoint"
+                );
+                Ok(())
+            })
+        }
+    }
 }

crates/io-metrics/src/internode_metrics.rs

@@ -916,7 +916,7 @@ fn cluster_peer_health_keys() -> Vec<String> {
 mod tests {
     use super::*;
     use metrics::with_local_recorder;
-    use metrics_util::debugging::DebuggingRecorder;
+    use metrics_util::debugging::{DebugValue, DebuggingRecorder};
     use std::collections::{HashMap, HashSet};
 
     #[test]
@@ -1308,11 +1308,47 @@ mod tests {
     }
 
     #[test]
-    fn msgpack_json_fallback_counter_records_without_panicking() {
-        // Smoke test: the counter accepts both directions and a static message label.
+    fn msgpack_json_fallback_counter_separates_the_two_directions() {
+        // Previously a smoke test that asserted nothing; the counter carries no
+        // in-struct total, so the emission itself is what has to be checked
+        // (rustfs/backlog#1836).
+        let recorder = DebuggingRecorder::new();
+        let snapshotter = recorder.snapshotter();
         let metrics = InternodeMetrics::default();
-        metrics.record_msgpack_json_fallback(INTERNODE_MSGPACK_DIRECTION_REQUEST, "FileInfo");
-        metrics.record_msgpack_json_fallback(INTERNODE_MSGPACK_DIRECTION_RESPONSE, "RawFileInfo");
+
+        metrics::with_local_recorder(&recorder, || {
+            metrics.record_msgpack_json_fallback(INTERNODE_MSGPACK_DIRECTION_REQUEST, "FileInfo");
+            metrics.record_msgpack_json_fallback(INTERNODE_MSGPACK_DIRECTION_RESPONSE, "RawFileInfo");
+        });
+
+        let observed: Vec<(String, String, u64)> = snapshotter
+            .snapshot()
+            .into_vec()
+            .into_iter()
+            .filter(|(composite, _, _, _)| composite.key().name() == INTERNODE_MSGPACK_JSON_FALLBACK_TOTAL)
+            .map(|(composite, _, _, value)| {
+                let labels: HashMap<String, String> = composite
+                    .key()
+                    .labels()
+                    .map(|label| (label.key().to_string(), label.value().to_string()))
+                    .collect();
+                let count = match value {
+                    DebugValue::Counter(count) => count,
+                    other => panic!("fallback total must be a counter, got {other:?}"),
+                };
+                (
+                    labels.get(DIRECTION_LABEL).cloned().unwrap_or_default(),
+                    labels.get(MESSAGE_LABEL).cloned().unwrap_or_default(),
+                    count,
+                )
+            })
+            .collect();
+
+        // Each direction/message pair is its own series, so a regression that
+        // dropped a label would collapse these into one row.
+        assert_eq!(observed.len(), 2, "each direction must land in its own series: {observed:?}");
+        assert!(observed.contains(&(INTERNODE_MSGPACK_DIRECTION_REQUEST.to_string(), "FileInfo".to_string(), 1)));
+        assert!(observed.contains(&(INTERNODE_MSGPACK_DIRECTION_RESPONSE.to_string(), "RawFileInfo".to_string(), 1)));
     }
 
     #[test]

crates/io-metrics/src/list_objects_metrics.rs

@@ -403,73 +403,132 @@ pub fn record_list_objects_local_read_dir(observation: ListObjectsLocalReadDirOb
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::set_get_stage_metrics_enabled;
+    use crate::tests::{METRICS_FLAG_LOCK, counter_total, emitted_names, histogram_samples};
+    use metrics_util::debugging::DebuggingRecorder;
 
-    #[test]
-    fn record_gather_observation_handles_empty_page() {
-        init_list_objects_metrics();
-        record_list_objects_gather(ListObjectsGatherObservation {
-            source: LIST_OBJECTS_SOURCE_WALKER,
-            outcome: LIST_OBJECTS_GATHER_OUTCOME_INPUT_CLOSED,
-            limit: 1001,
-            scanned_entries: 42,
-            returned_entries: 0,
-            duration_ms: 3.5,
-            has_prefix: true,
-            has_delimiter: false,
-            has_marker: true,
+    /// Run `body` against a local recorder with stage metrics on, and return the
+    /// snapshot rows.
+    ///
+    /// Enabling the flag is the point: every `record_*` here returns early when
+    /// `get_stage_metrics_enabled()` is false, which defaults to false. The
+    /// previous versions of these tests never set it, so they exercised nothing
+    /// but the early return (rustfs/backlog#1836).
+    fn recorded(body: impl FnOnce()) -> Vec<crate::tests::MetricRow> {
+        let _guard = METRICS_FLAG_LOCK.lock().unwrap_or_else(|e| e.into_inner());
+        let recorder = DebuggingRecorder::new();
+        let snapshotter = recorder.snapshotter();
+        metrics::with_local_recorder(&recorder, || {
+            init_list_objects_metrics();
+            set_get_stage_metrics_enabled(true);
+            body();
+            set_get_stage_metrics_enabled(false);
         });
+        snapshotter.snapshot().into_vec()
     }
 
     #[test]
-    fn record_merge_observation_accepts_zero_quorum() {
-        init_list_objects_metrics();
-        record_list_objects_merge(LIST_OBJECTS_SOURCE_WALKER, 4, 0);
+    fn gather_scan_amplification_falls_back_to_the_scan_count_on_an_empty_page() {
+        let rows = recorded(|| {
+            record_list_objects_gather(ListObjectsGatherObservation {
+                source: LIST_OBJECTS_SOURCE_WALKER,
+                outcome: LIST_OBJECTS_GATHER_OUTCOME_INPUT_CLOSED,
+                limit: 1001,
+                scanned_entries: 42,
+                returned_entries: 0,
+                duration_ms: 3.5,
+                has_prefix: true,
+                has_delimiter: false,
+                has_marker: true,
+            })
+        });
+
+        assert_eq!(counter_total(&rows, LIST_OBJECTS_GATHER_TOTAL), Some(1));
+        // Zero returned entries must not divide: the amplification reports the
+        // scan count itself rather than an infinity or a NaN.
+        assert_eq!(histogram_samples(&rows, LIST_OBJECTS_GATHER_SCAN_AMPLIFICATION), vec![42.0]);
+        assert_eq!(histogram_samples(&rows, LIST_OBJECTS_GATHER_FILTERED_ENTRIES), vec![42.0]);
+        assert_eq!(histogram_samples(&rows, LIST_OBJECTS_GATHER_RETURNED_ENTRIES), vec![0.0]);
+        assert_eq!(histogram_samples(&rows, LIST_OBJECTS_GATHER_DURATION_MS), vec![3.5]);
     }
 
     #[test]
-    fn record_index_fallback_observation_accepts_reason() {
-        init_list_objects_metrics();
-        record_list_objects_index_fallback("index_key_only", "unsupported_request");
+    fn merge_records_a_zero_read_quorum_rather_than_skipping_it() {
+        let rows = recorded(|| record_list_objects_merge(LIST_OBJECTS_SOURCE_WALKER, 4, 0));
+
+        assert_eq!(histogram_samples(&rows, LIST_OBJECTS_MERGE_FAN_IN), vec![4.0]);
+        assert_eq!(histogram_samples(&rows, LIST_OBJECTS_MERGE_READ_QUORUM), vec![0.0]);
     }
 
     #[test]
-    fn record_index_attempt_and_served_observations_accept_counts() {
-        init_list_objects_metrics();
-        record_list_objects_index_attempt("index_key_only", "walker_key_only", true, true, false);
-        record_list_objects_index_served(ListObjectsIndexPageObservation {
-            source: "index_key_only",
-            provider: "walker_key_only",
-            candidate_keys: 1000,
-            live_verify_attempts: 700,
-            live_verify_hits: 650,
-            live_verify_misses: 50,
-            returned_objects: 600,
-            returned_prefixes: 10,
-            is_truncated: true,
+    fn index_fallback_counts_once_per_reason() {
+        let rows = recorded(|| {
+            record_list_objects_index_fallback("index_key_only", "unsupported_request");
+            record_list_objects_index_fallback("index_key_only", "unsupported_request");
         });
-        record_list_objects_index_live_verify_failure("index_key_only", "read_error");
+
+        assert_eq!(counter_total(&rows, LIST_OBJECTS_INDEX_FALLBACK_TOTAL), Some(2));
     }
 
     #[test]
-    fn record_local_read_dir_observation_accepts_whole_directory_counts() {
-        init_list_objects_metrics();
-        record_list_objects_local_read_dir(ListObjectsLocalReadDirObservation {
-            outcome: LIST_OBJECTS_LOCAL_READ_DIR_OUTCOME_OK,
-            requested_count: -1,
-            returned_entries: 4096,
-            duration_ms: 12.5,
-            is_root: true,
-            has_filter_prefix: false,
-            has_forward: false,
+    fn index_serving_reports_verification_amplification_against_returned_objects() {
+        let rows = recorded(|| {
+            record_list_objects_index_attempt("index_key_only", "walker_key_only", true, true, false);
+            record_list_objects_index_served(ListObjectsIndexPageObservation {
+                source: "index_key_only",
+                provider: "walker_key_only",
+                candidate_keys: 1000,
+                live_verify_attempts: 700,
+                live_verify_hits: 650,
+                live_verify_misses: 50,
+                returned_objects: 600,
+                returned_prefixes: 10,
+                is_truncated: true,
+            });
+            record_list_objects_index_live_verify_failure("index_key_only", "read_error");
         });
-        record_list_objects_local_read_dir(ListObjectsLocalReadDirObservation {
-            outcome: LIST_OBJECTS_LOCAL_READ_DIR_OUTCOME_ERROR,
-            requested_count: -1,
-            returned_entries: 0,
-            duration_ms: 5000.0,
-            is_root: true,
-            has_filter_prefix: false,
-            has_forward: true,
+
+        assert_eq!(counter_total(&rows, LIST_OBJECTS_INDEX_ATTEMPT_TOTAL), Some(1));
+        assert_eq!(counter_total(&rows, LIST_OBJECTS_INDEX_SERVED_TOTAL), Some(1));
+        assert_eq!(counter_total(&rows, LIST_OBJECTS_INDEX_LIVE_VERIFY_FAILURE_TOTAL), Some(1));
+        assert_eq!(histogram_samples(&rows, LIST_OBJECTS_INDEX_CANDIDATE_KEYS), vec![1000.0]);
+        assert_eq!(histogram_samples(&rows, LIST_OBJECTS_INDEX_LIVE_VERIFY_HITS), vec![650.0]);
+        assert_eq!(histogram_samples(&rows, LIST_OBJECTS_INDEX_LIVE_VERIFY_MISSES), vec![50.0]);
+        assert_eq!(
+            histogram_samples(&rows, LIST_OBJECTS_INDEX_VERIFICATION_IO_AMPLIFICATION),
+            vec![700.0 / 600.0]
+        );
+    }
+
+    #[test]
+    fn local_read_dir_passes_the_whole_directory_sentinel_through_as_the_limit() {
+        let rows = recorded(|| {
+            record_list_objects_local_read_dir(ListObjectsLocalReadDirObservation {
+                outcome: LIST_OBJECTS_LOCAL_READ_DIR_OUTCOME_OK,
+                requested_count: -1,
+                returned_entries: 4096,
+                duration_ms: 12.5,
+                is_root: true,
+                has_filter_prefix: false,
+                has_forward: false,
+            });
+            record_list_objects_local_read_dir(ListObjectsLocalReadDirObservation {
+                outcome: LIST_OBJECTS_LOCAL_READ_DIR_OUTCOME_ERROR,
+                requested_count: -1,
+                returned_entries: 0,
+                duration_ms: 5000.0,
+                is_root: true,
+                has_filter_prefix: false,
+                has_forward: true,
+            });
         });
+
+        assert_eq!(counter_total(&rows, LIST_OBJECTS_LOCAL_READ_DIR_TOTAL), Some(2));
+        // `-1` is the "read the whole directory" sentinel and must reach the
+        // limit histogram unchanged rather than being clamped to zero.
+        assert_eq!(histogram_samples(&rows, LIST_OBJECTS_LOCAL_READ_DIR_LIMIT), vec![-1.0, -1.0]);
+        assert_eq!(histogram_samples(&rows, LIST_OBJECTS_LOCAL_READ_DIR_ENTRIES), vec![0.0, 4096.0]);
+        assert_eq!(histogram_samples(&rows, LIST_OBJECTS_LOCAL_READ_DIR_DURATION_MS), vec![12.5, 5000.0]);
+        assert!(emitted_names(&rows).contains(LIST_OBJECTS_LOCAL_READ_DIR_TOTAL));
     }
 }

crates/io-metrics/src/sampler/process.rs

@@ -188,18 +188,40 @@ mod tests {
     }
 
     #[test]
-    fn process_snapshots_are_collectable() {
-        let _ = snapshot_process_resource();
-        let _ = snapshot_process_system();
-        let _ = snapshot_process_resource_and_system();
+    fn combined_snapshot_agrees_with_the_individual_ones_on_per_process_facts() {
+        // Previously three discarded calls that asserted nothing. The values that
+        // move (cpu, memory) cannot be compared across calls, but the facts that
+        // identify the process must not differ by which entry point produced them
+        // (rustfs/backlog#1836).
+        let system = snapshot_process_system();
+        let (_, combined_system) = snapshot_process_resource_and_system();
+
+        assert_eq!(
+            system.start_time_seconds, combined_system.start_time_seconds,
+            "both entry points describe this process, so its start time cannot differ"
+        );
+        assert_eq!(
+            system.file_descriptor_limit_total, combined_system.file_descriptor_limit_total,
+            "the descriptor limit is a property of the process, not of the call"
+        );
+        assert_eq!(
+            system.status_value, combined_system.status_value,
+            "the status enum and its numeric projection must stay in step"
+        );
+        assert_eq!(combined_system.status_value, combined_system.status as i64);
     }
 
     #[test]
-    fn independent_samplers_are_collectable() {
+    fn independent_samplers_observe_the_same_process() {
         let mut sampler_a = ProcessSampler::new();
         let mut sampler_b = ProcessSampler::new();
 
-        let _ = snapshot_process_resource_and_system_with(&mut sampler_a);
-        let _ = snapshot_process_resource_and_system_with(&mut sampler_b);
+        let (_, system_a) = snapshot_process_resource_and_system_with(&mut sampler_a);
+        let (_, system_b) = snapshot_process_resource_and_system_with(&mut sampler_b);
+
+        // Two samplers hold separate sysinfo state; they must still agree on the
+        // process they are both looking at rather than each inventing a value.
+        assert_eq!(system_a.start_time_seconds, system_b.start_time_seconds);
+        assert_eq!(system_a.file_descriptor_limit_total, system_b.file_descriptor_limit_total);
     }
 }

crates/kms/src/lib.rs

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

crates/scanner/src/data_usage_define.rs

@@ -29,7 +29,7 @@ use rustfs_config::ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS;
 pub use rustfs_data_usage::{
     AllTierStats, BucketTargetUsageInfo, BucketUsageInfo, DATA_USAGE_OBJECT_NAME, DATA_USAGE_OBSERVED_OBJECT_NAME,
     DataUsageEntry, DataUsageHash, DataUsageHashMap, DataUsageInfo, LEGACY_DATA_USAGE_OBJECT_NAME, PrefixUsageEntry,
-    PrefixUsageQuery, PrefixUsageSummary, TierStats, hash_path, prefix_usage_in_cache,
+    PrefixUsageQuery, PrefixUsageSummary, ReplTargetSizeSummary, SizeSummary, TierStats, hash_path, prefix_usage_in_cache,
 };
 use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf};
 use tokio::time::{Duration, Instant, sleep, timeout};
@@ -188,38 +188,18 @@ pub static BACKGROUND_HEAL_INFO_PATH: LazyLock<String> =
 
 const MAX_DATA_USAGE_CACHE_DEPTH: usize = 1024;
 
-/// Size summary for a single object or group of objects
-#[derive(Debug, Default, Clone)]
-pub struct SizeSummary {
-    /// Total size
-    pub total_size: usize,
-    /// Number of versions
-    pub versions: usize,
-    /// Number of delete markers
-    pub delete_markers: usize,
-    /// Replicated size
-    pub replicated_size: i64,
-    /// Replicated count
-    pub replicated_count: usize,
-    /// Pending size
-    pub pending_size: i64,
-    /// Failed size
-    pub failed_size: i64,
-    /// Replica size
-    pub replica_size: i64,
-    /// Replica count
-    pub replica_count: usize,
-    /// Pending count
-    pub pending_count: usize,
-    /// Failed count
-    pub failed_count: usize,
-    /// Replication target stats
-    pub repl_target_stats: HashMap<String, ReplTargetSizeSummary>,
-    pub tier_stats: HashMap<String, TierStats>,
+/// Scanner-side accounting on the shared [`SizeSummary`].
+///
+/// The type itself lives in `rustfs-data-usage`, which sits below the storage
+/// layer and cannot see `ObjectInfo`, so this stays an extension trait rather
+/// than an inherent method (backlog#1828).
+pub trait ScannerSizeSummaryExt {
+    /// Fold one object's contribution into the summary, including its tier.
+    fn actions_accounting(&mut self, oi: &ObjectInfo, size: i64, actual_size: i64);
 }
 
-impl SizeSummary {
-    pub fn actions_accounting(&mut self, oi: &ObjectInfo, size: i64, actual_size: i64) {
+impl ScannerSizeSummaryExt for SizeSummary {
+    fn actions_accounting(&mut self, oi: &ObjectInfo, size: i64, actual_size: i64) {
         if oi.delete_marker {
             self.delete_markers = self.delete_markers.saturating_add(1);
             return;
@@ -251,23 +231,6 @@ impl SizeSummary {
     }
 }
 
-/// Replication target size summary
-#[derive(Debug, Default, Clone)]
-pub struct ReplTargetSizeSummary {
-    /// Replicated size
-    pub replicated_size: i64,
-    /// Replicated count
-    pub replicated_count: usize,
-    /// Pending size
-    pub pending_size: i64,
-    /// Failed size
-    pub failed_size: i64,
-    /// Pending count
-    pub pending_count: usize,
-    /// Failed count
-    pub failed_count: usize,
-}
-
 // ===== Cache-related data structures =====
 
 #[derive(Clone, Copy, Debug, Serialize, Deserialize, PartialEq, Eq)]
@@ -1544,39 +1507,6 @@ pub trait DataUsageCacheStorage {
     async fn save(&self, name: &str) -> Result<(), Box<dyn std::error::Error + Send + Sync>>;
 }
 
-impl SizeSummary {
-    /// Create a new SizeSummary
-    pub fn new() -> Self {
-        Self::default()
-    }
-
-    /// Add another SizeSummary to this one
-    pub fn add(&mut self, other: &SizeSummary) {
-        self.total_size = self.total_size.saturating_add(other.total_size);
-        self.versions = self.versions.saturating_add(other.versions);
-        self.delete_markers = self.delete_markers.saturating_add(other.delete_markers);
-        self.replicated_size = self.replicated_size.saturating_add(other.replicated_size);
-        self.replicated_count = self.replicated_count.saturating_add(other.replicated_count);
-        self.pending_size = self.pending_size.saturating_add(other.pending_size);
-        self.failed_size = self.failed_size.saturating_add(other.failed_size);
-        self.replica_size = self.replica_size.saturating_add(other.replica_size);
-        self.replica_count = self.replica_count.saturating_add(other.replica_count);
-        self.pending_count = self.pending_count.saturating_add(other.pending_count);
-        self.failed_count = self.failed_count.saturating_add(other.failed_count);
-
-        // Merge replication target stats
-        for (target, stats) in &other.repl_target_stats {
-            let entry = self.repl_target_stats.entry(target.clone()).or_default();
-            entry.replicated_size = entry.replicated_size.saturating_add(stats.replicated_size);
-            entry.replicated_count = entry.replicated_count.saturating_add(stats.replicated_count);
-            entry.pending_size = entry.pending_size.saturating_add(stats.pending_size);
-            entry.failed_size = entry.failed_size.saturating_add(stats.failed_size);
-            entry.pending_count = entry.pending_count.saturating_add(stats.pending_count);
-            entry.failed_count = entry.failed_count.saturating_add(stats.failed_count);
-        }
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;

crates/scanner/src/scanner_folder.rs

@@ -21,7 +21,7 @@ use std::time::{Duration, Instant, SystemTime};
 use crate::ReplTargetSizeSummary;
 use crate::data_usage_define::{
     DATA_USAGE_SCAN_CHECKPOINT_VERSION, DataUsageCache, DataUsageEntry, DataUsageHash, DataUsageHashMap, DataUsageScanCheckpoint,
-    DataUsageScanCheckpointReason, PendingScannerHeal, PendingScannerHealKind, SizeSummary, hash_path,
+    DataUsageScanCheckpointReason, PendingScannerHeal, PendingScannerHealKind, ScannerSizeSummaryExt, SizeSummary, hash_path,
 };
 use crate::error::ScannerError;
 use crate::runtime_config::{

docs/testing/e2e-suite-inventory.md

@@ -63,7 +63,7 @@
 | list_objects_v2_metadata_extension_test | 1 |  |
 | list_objects_v2_pagination_test | 12 | ✅ |
 | mc_mirror_small_bucket_test | 1 |  |
-| multipart_auth_test | 85 |  |
+| multipart_auth_test | 75 |  |
 | multipart_storage_class_test | 3 | ✅ |
 | namespace_lock_quorum_test | 2 |  |
 | negative_sigv4_test | 6 | ✅ |

rustfs/src/server/layer.rs

@@ -747,11 +747,12 @@ pub struct S3ErrorMessageCompatService<S> {
     inner: S,
 }
 
-impl<S, RestBody, GrpcBody> Service<HttpRequest<Incoming>> for S3ErrorMessageCompatService<S>
+impl<S, ReqBody, RestBody, GrpcBody> Service<HttpRequest<ReqBody>> for S3ErrorMessageCompatService<S>
 where
-    S: Service<HttpRequest<Incoming>, Response = Response<HybridBody<RestBody, GrpcBody>>> + Clone + Send + 'static,
+    S: Service<HttpRequest<ReqBody>, Response = Response<HybridBody<RestBody, GrpcBody>>> + Clone + Send + 'static,
     S::Future: Send + 'static,
     S::Error: Send + 'static,
+    ReqBody: Send + 'static,
     RestBody: Body<Data = Bytes> + From<Bytes> + Send + 'static,
     RestBody::Error: Into<S::Error> + Send + 'static,
     GrpcBody: Send + 'static,
@@ -764,28 +765,27 @@ where
         self.inner.poll_ready(cx)
     }
 
-    fn call(&mut self, req: HttpRequest<Incoming>) -> Self::Future {
+    fn call(&mut self, req: HttpRequest<ReqBody>) -> Self::Future {
         let is_sts_query =
             req.method() == Method::POST && req.uri().path() == "/" && req.extensions().get::<StsQueryRequest>().is_some();
         let mut inner = self.inner.clone();
 
         Box::pin(async move {
             let response = inner.call(req).await?;
+            if is_sts_query || response.status() != StatusCode::FORBIDDEN || !is_xml_response(response.headers()) {
+                return Ok(response);
+            }
+
             let (parts, body) = response.into_parts();
-            let should_fix = !is_sts_query && parts.status == StatusCode::FORBIDDEN && is_xml_response(&parts.headers);
 
             let response = match body {
                 HybridBody::Rest { rest_body } => {
-                    if !should_fix {
-                        Response::from_parts(parts, HybridBody::Rest { rest_body })
-                    } else {
-                        let (rest_body, changed) = fix_s3_error_message_in_xml(rest_body).await.map_err(Into::into)?;
-                        let mut parts = parts;
-                        if changed {
-                            parts.headers.remove(http::header::CONTENT_LENGTH);
-                        }
-                        Response::from_parts(parts, HybridBody::Rest { rest_body })
+                    let (rest_body, changed) = fix_s3_error_message_in_xml(rest_body).await.map_err(Into::into)?;
+                    let mut parts = parts;
+                    if changed {
+                        parts.headers.remove(http::header::CONTENT_LENGTH);
                     }
+                    Response::from_parts(parts, HybridBody::Rest { rest_body })
                 }
                 HybridBody::Grpc { grpc_body } => Response::from_parts(parts, HybridBody::Grpc { grpc_body }),
             };
@@ -886,11 +886,12 @@ pub struct IcebergRestErrorCompatService<S> {
     inner: S,
 }
 
-impl<S, RestBody, GrpcBody> Service<HttpRequest<Incoming>> for IcebergRestErrorCompatService<S>
+impl<S, ReqBody, RestBody, GrpcBody> Service<HttpRequest<ReqBody>> for IcebergRestErrorCompatService<S>
 where
-    S: Service<HttpRequest<Incoming>, Response = Response<HybridBody<RestBody, GrpcBody>>> + Clone + Send + 'static,
+    S: Service<HttpRequest<ReqBody>, Response = Response<HybridBody<RestBody, GrpcBody>>> + Clone + Send + 'static,
     S::Future: Send + 'static,
     S::Error: Send + 'static,
+    ReqBody: Send + 'static,
     RestBody: Body<Data = Bytes> + From<Bytes> + Send + 'static,
     RestBody::Error: Into<S::Error> + Send + 'static,
     GrpcBody: Send + 'static,
@@ -903,18 +904,21 @@ where
         self.inner.poll_ready(cx)
     }
 
-    fn call(&mut self, req: HttpRequest<Incoming>) -> Self::Future {
+    fn call(&mut self, req: HttpRequest<ReqBody>) -> Self::Future {
         let catalog_path =
             (req.method() != Method::HEAD && is_table_catalog_path(req.uri().path())).then(|| req.uri().path().to_string());
         let mut inner = self.inner.clone();
 
         Box::pin(async move {
             let response = inner.call(req).await?;
+            if catalog_path.is_none() || response.status().is_success() || !is_xml_response(response.headers()) {
+                return Ok(response);
+            }
+
             let (parts, body) = response.into_parts();
-            let should_convert = catalog_path.is_some() && !parts.status.is_success() && is_xml_response(&parts.headers);
 
             let response = match body {
-                HybridBody::Rest { rest_body } if should_convert => {
+                HybridBody::Rest { rest_body } => {
                     let (rest_body, converted_status) = convert_iceberg_error_in_xml(
                         rest_body,
                         parts.status,
@@ -932,7 +936,6 @@ where
                     }
                     Response::from_parts(parts, HybridBody::Rest { rest_body })
                 }
-                HybridBody::Rest { rest_body } => Response::from_parts(parts, HybridBody::Rest { rest_body }),
                 HybridBody::Grpc { grpc_body } => Response::from_parts(parts, HybridBody::Grpc { grpc_body }),
             };
 
@@ -1045,11 +1048,12 @@ pub struct ObjectAttributesEtagFixService<S> {
     inner: S,
 }
 
-impl<S, RestBody, GrpcBody> Service<HttpRequest<Incoming>> for ObjectAttributesEtagFixService<S>
+impl<S, ReqBody, RestBody, GrpcBody> Service<HttpRequest<ReqBody>> for ObjectAttributesEtagFixService<S>
 where
-    S: Service<HttpRequest<Incoming>, Response = Response<HybridBody<RestBody, GrpcBody>>> + Clone + Send + 'static,
+    S: Service<HttpRequest<ReqBody>, Response = Response<HybridBody<RestBody, GrpcBody>>> + Clone + Send + 'static,
     S::Future: Send + 'static,
     S::Error: Send + 'static,
+    ReqBody: Send + 'static,
     RestBody: Body<Data = Bytes> + From<Bytes> + Send + 'static,
     RestBody::Error: Into<S::Error> + Send + 'static,
     GrpcBody: Send + 'static,
@@ -1062,27 +1066,26 @@ where
         self.inner.poll_ready(cx)
     }
 
-    fn call(&mut self, req: HttpRequest<Incoming>) -> Self::Future {
+    fn call(&mut self, req: HttpRequest<ReqBody>) -> Self::Future {
         let is_target = is_object_attributes_request(&req);
         let mut inner = self.inner.clone();
 
         Box::pin(async move {
             let response = inner.call(req).await?;
+            if !is_target || !response.status().is_success() || !is_xml_response(response.headers()) {
+                return Ok(response);
+            }
+
             let (parts, body) = response.into_parts();
-            let should_fix = is_target && parts.status.is_success() && is_xml_response(&parts.headers);
 
             let response = match body {
                 HybridBody::Rest { rest_body } => {
-                    if !should_fix {
-                        Response::from_parts(parts, HybridBody::Rest { rest_body })
-                    } else {
-                        let rest_body = fix_object_attributes_etag_in_xml(rest_body).await.map_err(Into::into)?;
+                    let rest_body = fix_object_attributes_etag_in_xml(rest_body).await.map_err(Into::into)?;
 
-                        let mut parts = parts;
-                        parts.headers.remove(http::header::CONTENT_LENGTH);
+                    let mut parts = parts;
+                    parts.headers.remove(http::header::CONTENT_LENGTH);
 
-                        Response::from_parts(parts, HybridBody::Rest { rest_body })
-                    }
+                    Response::from_parts(parts, HybridBody::Rest { rest_body })
                 }
                 HybridBody::Grpc { grpc_body } => Response::from_parts(parts, HybridBody::Grpc { grpc_body }),
             };
@@ -1144,12 +1147,11 @@ where
 
         Box::pin(async move {
             let response = inner.call(req).await?;
-            let (mut parts, body) = response.into_parts();
-
-            if !is_bodyless_status(parts.status) {
-                return Ok(Response::from_parts(parts, body));
+            if !is_bodyless_status(response.status()) {
+                return Ok(response);
             }
 
+            let (mut parts, body) = response.into_parts();
             let response = match body {
                 HybridBody::Rest { .. } => {
                     parts.headers.remove(http::header::CONTENT_LENGTH);
@@ -1802,7 +1804,7 @@ fn strip_quotes_from_first_etag(xml: String) -> String {
     fixed
 }
 
-fn is_object_attributes_request(req: &HttpRequest<Incoming>) -> bool {
+fn is_object_attributes_request<B>(req: &HttpRequest<B>) -> bool {
     if req.method() != Method::GET {
         return false;
     }
@@ -1967,11 +1969,12 @@ fn apply_bucket_cors_result(response_headers: &mut HeaderMap, bucket_cors_header
     }
 }
 
-impl<S, ResBody> Service<HttpRequest<Incoming>> for ConditionalCorsService<S>
+impl<S, ReqBody, ResBody> Service<HttpRequest<ReqBody>> for ConditionalCorsService<S>
 where
-    S: Service<HttpRequest<Incoming>, Response = Response<ResBody>> + Clone + Send + 'static,
+    S: Service<HttpRequest<ReqBody>, Response = Response<ResBody>> + Clone + Send + 'static,
     S::Future: Send + 'static,
     S::Error: Into<Box<dyn std::error::Error + Send + Sync>> + Send + 'static,
+    ReqBody: Send + 'static,
     ResBody: Default + Send + 'static,
 {
     type Response = Response<ResBody>;
@@ -1982,7 +1985,14 @@ where
         self.inner.poll_ready(cx).map_err(Into::into)
     }
 
-    fn call(&mut self, req: HttpRequest<Incoming>) -> Self::Future {
+    fn call(&mut self, req: HttpRequest<ReqBody>) -> Self::Future {
+        let is_options = req.method() == Method::OPTIONS;
+        let has_origin = req.headers().contains_key(cors::standard::ORIGIN);
+        if !is_options && !has_origin {
+            let mut inner = self.inner.clone();
+            return Box::pin(async move { inner.call(req).await.map_err(Into::into) });
+        }
+
         let path = req.uri().path().to_string();
         let method = req.method().clone();
         let request_headers = req.headers().clone();
@@ -1990,7 +2000,7 @@ where
         let is_s3 = ConditionalCorsLayer::is_s3_path(&path);
         let is_root = path == "/";
 
-        if method == Method::OPTIONS {
+        if is_options {
             let has_acrm = request_headers.contains_key(cors::request::ACCESS_CONTROL_REQUEST_METHOD);
 
             if is_root {
@@ -2192,6 +2202,7 @@ mod tests {
     use futures::future::{Ready, ready};
     use http::Request;
     use http_body_util::BodyExt;
+    use http_body_util::Empty;
     use http_body_util::Full;
     use opentelemetry::global;
     use opentelemetry_sdk::propagation::TraceContextPropagator;
@@ -3783,6 +3794,188 @@ mod tests {
         assert_eq!(bytes, input);
     }
 
+    #[derive(Clone)]
+    struct FixedHybridResponse {
+        status: StatusCode,
+        body: Bytes,
+        content_type: &'static str,
+    }
+
+    impl<B: Send + 'static> Service<Request<B>> for FixedHybridResponse {
+        type Response = Response<HybridBody<Full<Bytes>, Empty<Bytes>>>;
+        type Error = Infallible;
+        type Future = Ready<Result<Self::Response, Self::Error>>;
+
+        fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+            Poll::Ready(Ok(()))
+        }
+
+        fn call(&mut self, _req: Request<B>) -> Self::Future {
+            let body = self.body.clone();
+            ready(Ok(Response::builder()
+                .status(self.status)
+                .header(http::header::CONTENT_TYPE, self.content_type)
+                .header(http::header::CONTENT_LENGTH, body.len().to_string())
+                .body(HybridBody::Rest {
+                    rest_body: Full::from(body),
+                })
+                .expect("fixed hybrid response")))
+        }
+    }
+
+    async fn collect_hybrid_response(
+        response: Response<HybridBody<Full<Bytes>, Empty<Bytes>>>,
+    ) -> (StatusCode, HeaderMap, String) {
+        let status = response.status();
+        let headers = response.headers().clone();
+        let body = BodyExt::collect(response.into_body())
+            .await
+            .expect("collect hybrid body")
+            .to_bytes();
+        (
+            status,
+            headers,
+            String::from_utf8(body.to_vec()).expect("hybrid response body should be UTF-8"),
+        )
+    }
+
+    #[tokio::test]
+    async fn s3_error_message_compat_fixes_regular_forbidden_xml() {
+        let body = Bytes::from_static(b"<Error><Code>SignatureDoesNotMatch</Code></Error>");
+        let mut service = S3ErrorMessageCompatLayer.layer(FixedHybridResponse {
+            status: StatusCode::FORBIDDEN,
+            body,
+            content_type: "application/xml",
+        });
+        let request = Request::builder()
+            .method(Method::GET)
+            .uri("/bucket/object")
+            .body(())
+            .expect("request");
+
+        let response = service.call(request).await.expect("service response");
+        let (status, headers, body) = collect_hybrid_response(response).await;
+
+        assert_eq!(status, StatusCode::FORBIDDEN);
+        assert!(headers.get(http::header::CONTENT_LENGTH).is_none());
+        assert!(body.contains("<Message>"));
+    }
+
+    #[tokio::test]
+    async fn s3_error_message_compat_leaves_sts_query_response_unchanged() {
+        let input = Bytes::from_static(b"<Error><Code>SignatureDoesNotMatch</Code></Error>");
+        let mut service = S3ErrorMessageCompatLayer.layer(FixedHybridResponse {
+            status: StatusCode::FORBIDDEN,
+            body: input.clone(),
+            content_type: "application/xml",
+        });
+        let mut request = Request::builder().method(Method::POST).uri("/").body(()).expect("request");
+        request.extensions_mut().insert(StsQueryRequest);
+
+        let response = service.call(request).await.expect("service response");
+        let (_status, headers, body) = collect_hybrid_response(response).await;
+
+        let expected_len = input.len().to_string();
+        assert_eq!(
+            headers
+                .get(http::header::CONTENT_LENGTH)
+                .and_then(|value| value.to_str().ok()),
+            Some(expected_len.as_str())
+        );
+        assert_eq!(body.as_bytes(), input.as_ref());
+    }
+
+    #[tokio::test]
+    async fn iceberg_rest_error_compat_converts_catalog_xml_errors() {
+        let mut service = IcebergRestErrorCompatLayer.layer(FixedHybridResponse {
+            status: StatusCode::NOT_FOUND,
+            body: Bytes::from_static(b"<Error><Code>NoSuchTableException</Code><Message>missing</Message></Error>"),
+            content_type: "application/xml",
+        });
+        let request = Request::builder()
+            .method(Method::GET)
+            .uri("/iceberg/v1/warehouse/namespaces/ns/tables/events")
+            .body(())
+            .expect("request");
+
+        let response = service.call(request).await.expect("service response");
+        let (status, headers, body) = collect_hybrid_response(response).await;
+
+        assert_eq!(status, StatusCode::NOT_FOUND);
+        assert_eq!(headers.get(http::header::CONTENT_TYPE).unwrap(), "application/json");
+        assert!(headers.get(http::header::CONTENT_LENGTH).is_none());
+        assert!(body.contains("\"type\":\"NoSuchTableException\""));
+    }
+
+    #[tokio::test]
+    async fn iceberg_rest_error_compat_leaves_non_catalog_errors_unchanged() {
+        let input = Bytes::from_static(b"<Error><Code>NoSuchKey</Code><Message>missing</Message></Error>");
+        let mut service = IcebergRestErrorCompatLayer.layer(FixedHybridResponse {
+            status: StatusCode::NOT_FOUND,
+            body: input.clone(),
+            content_type: "application/xml",
+        });
+        let request = Request::builder()
+            .method(Method::GET)
+            .uri("/bucket/object")
+            .body(())
+            .expect("request");
+
+        let response = service.call(request).await.expect("service response");
+        let (status, headers, body) = collect_hybrid_response(response).await;
+
+        assert_eq!(status, StatusCode::NOT_FOUND);
+        assert_eq!(headers.get(http::header::CONTENT_TYPE).unwrap(), "application/xml");
+        assert_eq!(body.as_bytes(), input.as_ref());
+    }
+
+    #[tokio::test]
+    async fn object_attributes_etag_fix_rewrites_target_response() {
+        let mut service = ObjectAttributesEtagFixLayer.layer(FixedHybridResponse {
+            status: StatusCode::OK,
+            body: Bytes::from_static(b"<GetObjectAttributesOutput><ETag>\"abc\"</ETag></GetObjectAttributesOutput>"),
+            content_type: "application/xml",
+        });
+        let request = Request::builder()
+            .method(Method::GET)
+            .uri("/bucket/object?attributes")
+            .body(())
+            .expect("request");
+
+        let response = service.call(request).await.expect("service response");
+        let (_status, headers, body) = collect_hybrid_response(response).await;
+
+        assert!(headers.get(http::header::CONTENT_LENGTH).is_none());
+        assert!(body.contains("<ETag>abc</ETag>"));
+    }
+
+    #[tokio::test]
+    async fn object_attributes_etag_fix_leaves_regular_get_unchanged() {
+        let input = Bytes::from_static(b"<GetObjectAttributesOutput><ETag>\"abc\"</ETag></GetObjectAttributesOutput>");
+        let mut service = ObjectAttributesEtagFixLayer.layer(FixedHybridResponse {
+            status: StatusCode::OK,
+            body: input.clone(),
+            content_type: "application/xml",
+        });
+        let request = Request::builder()
+            .method(Method::GET)
+            .uri("/bucket/object")
+            .body(())
+            .expect("request");
+
+        let response = service.call(request).await.expect("service response");
+        let (_status, headers, body) = collect_hybrid_response(response).await;
+
+        let expected_len = input.len().to_string();
+        assert_eq!(
+            headers
+                .get(http::header::CONTENT_LENGTH)
+                .and_then(|value| value.to_str().ok()),
+            Some(expected_len.as_str())
+        );
+        assert_eq!(body.as_bytes(), input.as_ref());
+    }
+
     #[derive(Clone)]
     struct FixedStsResponse {
         status: StatusCode,
@@ -4270,6 +4463,63 @@ mod tests {
         });
     }
 
+    #[derive(Clone)]
+    struct CorsOkService;
+
+    impl<B> Service<Request<B>> for CorsOkService {
+        type Response = Response<Empty<Bytes>>;
+        type Error = Infallible;
+        type Future = Ready<Result<Self::Response, Self::Error>>;
+
+        fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+            Poll::Ready(Ok(()))
+        }
+
+        fn call(&mut self, _req: Request<B>) -> Self::Future {
+            ready(Ok(Response::builder()
+                .status(StatusCode::OK)
+                .body(Empty::new())
+                .expect("response")))
+        }
+    }
+
+    #[tokio::test]
+    async fn conditional_cors_passthrough_without_origin() {
+        let layer = ConditionalCorsLayer {
+            cors_origins: Some("*".to_string()),
+        };
+        let mut service = layer.layer(CorsOkService);
+        let request = Request::builder()
+            .method(Method::GET)
+            .uri("/bucket/object")
+            .body(())
+            .expect("request");
+
+        let response = service.call(request).await.expect("response");
+
+        assert_eq!(response.status(), StatusCode::OK);
+        assert!(response.headers().get(cors::response::ACCESS_CONTROL_ALLOW_ORIGIN).is_none());
+    }
+
+    #[tokio::test]
+    async fn conditional_cors_applies_origin_headers() {
+        let layer = ConditionalCorsLayer {
+            cors_origins: Some("*".to_string()),
+        };
+        let mut service = layer.layer(CorsOkService);
+        let request = Request::builder()
+            .method(Method::GET)
+            .uri("/bucket/object")
+            .header(cors::standard::ORIGIN, "https://example.com")
+            .body(())
+            .expect("request");
+
+        let response = service.call(request).await.expect("response");
+
+        assert_eq!(response.status(), StatusCode::OK);
+        assert_eq!(response.headers().get(cors::response::ACCESS_CONTROL_ALLOW_ORIGIN).unwrap(), "*");
+    }
+
     #[test]
     fn request_context_layer_populates_context_without_mutating_signed_headers() {
         let mut service = RequestContextLayer.layer(CaptureService);

rustfs/src/storage/minio_generated_read_test.rs

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

rustfs/src/storage/sse.rs

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