test(crypto): replace the one-file key scan with a repo-wide guard

crates/crypto/src/license_token.rs asserted, via include_str! on its own
file, that the license-token signing key is never checked in. The scan saw
exactly one file: the key moved anywhere else passed silently, and renaming
license_token.rs stopped the guard from compiling instead of reporting.

scripts/check_embedded_secrets.sh scans every tracked and not-yet-added text
file for the same needle plus the other private-key header forms and eleven
provider credential formats, and it does not skip the paths
.github/secret_scanning.yml tells push protection to ignore. Non-secret
matches are excused by exact literal, never by path glob, and an exemption
that stops matching is reported as stale. --self-test asserts every pattern
family fires.

.config/make/lint-fmt.mak

@@ -70,6 +70,11 @@ fips-wording-check: ## Check docs and crates/kms do not over-claim crypto capabi
 	@echo "📣 Checking cryptographic capability wording guard..."
 	./scripts/check_fips_wording.sh
 
+.PHONY: embedded-secrets-check
+embedded-secrets-check: ## Check no private key material or credential literal is committed
+	@echo "🔑 Checking embedded secret material guard..."
+	./scripts/check_embedded_secrets.sh
+
 .PHONY: log-analyzer-rules-check
 log-analyzer-rules-check: core-deps ## Check log-analyzer rule anchors still exist verbatim in source
 	@echo "🩺 Checking log-analyzer rule anchors..."

.config/make/pre-commit.mak

@@ -19,13 +19,13 @@ planning-docs-check: ## Check that no planning-type documents are committed
 	./scripts/check_no_planning_docs.sh
 
 .PHONY: pre-commit
-pre-commit: fmt-check unsafe-code-check architecture-migration-check logging-guardrails-check tokio-io-uring-check extension-schema-check body-cache-whitelist-check s3s-footprint-check fips-wording-check doc-paths-check planning-docs-check quick-check ## Run fast pre-commit checks without clippy/full tests
+pre-commit: fmt-check unsafe-code-check architecture-migration-check logging-guardrails-check tokio-io-uring-check extension-schema-check body-cache-whitelist-check s3s-footprint-check fips-wording-check embedded-secrets-check doc-paths-check planning-docs-check quick-check ## Run fast pre-commit checks without clippy/full tests
 	@echo "✅ All pre-commit checks passed!"
 
 .PHONY: pre-pr
-pre-pr: fmt-check unsafe-code-check architecture-migration-check logging-guardrails-check tokio-io-uring-check extension-schema-check body-cache-whitelist-check s3s-footprint-check fips-wording-check doc-paths-check planning-docs-check log-analyzer-rules-check clippy-check test ## Run full pre-PR checks with clippy and tests
+pre-pr: fmt-check unsafe-code-check architecture-migration-check logging-guardrails-check tokio-io-uring-check extension-schema-check body-cache-whitelist-check s3s-footprint-check fips-wording-check embedded-secrets-check doc-paths-check planning-docs-check log-analyzer-rules-check clippy-check test ## Run full pre-PR checks with clippy and tests
 	@echo "✅ All pre-PR checks passed!"
 
 .PHONY: dev-check
-dev-check: fmt-check unsafe-code-check architecture-migration-check logging-guardrails-check tokio-io-uring-check extension-schema-check body-cache-whitelist-check s3s-footprint-check fips-wording-check doc-paths-check planning-docs-check quick-check ## Run fast local development checks
+dev-check: fmt-check unsafe-code-check architecture-migration-check logging-guardrails-check tokio-io-uring-check extension-schema-check body-cache-whitelist-check s3s-footprint-check fips-wording-check embedded-secrets-check doc-paths-check planning-docs-check quick-check ## Run fast local development checks
 	@echo "✅ Fast development checks passed!"

.config/make/tests.mak

@@ -34,6 +34,7 @@ script-tests: ## Run shell script tests
 	./scripts/test_exact_1mib_handoff_abba.sh
 	./scripts/test_pinned_paired_abba_bench.sh
 	./scripts/test_manual_transition_runbooks.sh
+	./scripts/check_embedded_secrets.sh --self-test
 	bash -n ./scripts/validate_object_data_cache_cold_stampede.sh
 	python3 ./scripts/check_object_data_cache_follower_samples.py --self-test
 	./scripts/validate_object_data_cache_cold_stampede.sh --self-test

.github/workflows/ci-docs-only.yml

@@ -120,6 +120,9 @@ jobs:
       - name: Check cryptographic capability wording
         run: ./scripts/check_fips_wording.sh
 
+      - name: Check no embedded secret material
+        run: ./scripts/check_embedded_secrets.sh
+
       - name: Check no planning docs committed
         run: ./scripts/check_no_planning_docs.sh
 

.github/workflows/ci.yml

@@ -155,6 +155,9 @@ jobs:
       - name: Check cryptographic capability wording
         run: ./scripts/check_fips_wording.sh
 
+      - name: Check no embedded secret material
+        run: ./scripts/check_embedded_secrets.sh
+
       - name: Check no planning docs committed
         run: ./scripts/check_no_planning_docs.sh
 

.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/crypto/src/license_token.rs

@@ -203,14 +203,6 @@ mod tests {
         assert!(result.is_err());
     }
 
-    #[test]
-    fn test_source_does_not_embed_private_key() {
-        let source = include_str!("license_token.rs");
-        let forbidden = ["BEGIN", "PRIVATE KEY"].join(" ");
-
-        assert!(!source.contains(&forbidden));
-    }
-
     #[test]
     fn test_parse_signed_license_token_rejects_invalid_token() {
         let mut rng = rand::rng();

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

@@ -2026,8 +2026,17 @@ enum DataUsageCacheRead {
 
 /// 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)
+    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>
@@ -2511,7 +2520,10 @@ mod tests {
                 *remaining -= 1;
                 return Err(Error::other("transient read failure"));
             }
-            Err(Error::FileNotFound)
+            // `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(
@@ -2533,6 +2545,23 @@ mod tests {
             .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";

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/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);

scripts/README.md

@@ -28,6 +28,7 @@ their issue closes.
 | `check_architecture_migration_rules.sh` | ci-gate | Architecture-boundary anti-regression guard | ci.yml Quick Checks; `make pre-commit` |
 | `check_body_cache_whitelist.sh` | ci-gate | Keeps the app-layer body-cache eligibility gate fail-closed | ci.yml Quick Checks |
 | `check_doc_paths.sh` | ci-gate | Fails when instruction/architecture docs reference repo paths that no longer exist | `make pre-commit` / `pre-pr` |
+| `check_embedded_secrets.sh` | ci-gate | Repo-wide scan blocking committed private key material and provider credential literals | ci.yml Quick Checks; `make pre-commit` / `pre-pr` |
 | `check_extension_schema_boundaries.sh` | ci-gate | Extension-schema crate boundary guard | ci.yml Quick Checks; `make pre-commit` |
 | `check_layer_dependencies.sh` | ci-gate | Crate-layering DAG guard (reads `layer-dependency-baseline.txt`) | ci.yml Quick Checks |
 | `check_logging_guardrails.sh` | ci-gate | Blocks legacy logging patterns from returning | `make pre-commit` / `pre-pr` |

scripts/check_embedded_secrets.sh

@@ -0,0 +1,279 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+# Guard: no private key material and no long-lived provider credential may
+# exist as a literal anywhere in the repository — see AGENTS.md "Security
+# Baseline" ("Never commit secrets, credentials, or key material") and
+# .agents/skills/security-advisory-lessons ("Do not ship hard-coded shared
+# tokens, HMAC secrets, private keys, or production test keys").
+#
+# This replaces the unit test `test_source_does_not_embed_private_key` that
+# used to live in crates/crypto/src/license_token.rs (rustfs/backlog#1884).
+# That test read its own file with include_str! and asserted the file did not
+# contain a PEM private-key header, protecting exactly one invariant: the RSA
+# key that signs license tokens must never be checked in, because verification
+# only ever needs the public key (crates/crypto/src/license_token.rs exposes
+# `parse_signed_license_token`, and rustfs/src/license.rs reads the public key
+# from RUSTFS_LICENSE_PUBLIC_KEY at runtime — no key material belongs in the
+# tree at all). Its coverage was one file: moving the key one file sideways,
+# even inside the same crate, passed silently, and renaming license_token.rs
+# stopped the guard from compiling rather than reporting anything.
+#
+# This scan covers every tracked — and every not-yet-added, non-ignored — text
+# file in the repository, so it is a strict superset of the retired assertion:
+# the same needle, everywhere, plus the algorithm variants and the credential
+# formats below.
+#
+# It deliberately does not exclude the paths .github/secret_scanning.yml tells
+# GitHub push protection to ignore (crates/e2e_test, **/tests, **/benches,
+# .docker, .vscode). Those exclusions exist because pasted test credentials are
+# expected there, which is exactly where a real key is most likely to arrive
+# unnoticed; this guard is the CI-side gate that still looks.
+#
+# Only literals are in scope. Key material injected at build time is out of
+# scope on purpose: no build.rs in the workspace embeds key material and the
+# license public key is read from the environment at startup, so an artifact
+# scan would add a release build to a compile-free check job for no reachable
+# failure mode today. Revisit if a build script ever bakes in key material.
+# Binary files are skipped (`git grep -I`), and a key stored as bare base64
+# with its header stripped is not detected — the same two blind spots the
+# retired test had.
+#
+# Every needle below is assembled around a variable so that the script's own
+# text does not match the pattern it defines (the retired test used the same
+# trick with ["BEGIN", "PRIVATE KEY"].join(" ")). That is what lets this script
+# scan itself along with everything else instead of carving out a blind spot.
+#
+# `--self-test` builds throwaway fixture repositories and asserts every pattern
+# family fires and every exemption holds; it is wired into `make script-tests`.
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+ROOT_DIR="${CHECK_EMBEDDED_SECRETS_ROOT:-$(cd "${SCRIPT_DIR}/.." && pwd)}"
+
+BEGIN_MARK="BEGIN"
+KEY_MARK="Key"
+
+# The file the retired test pinned with include_str!. It stays listed so that
+# renaming or moving it is reported here explicitly instead of quietly ending
+# the license-key invariant, which is how the test failed.
+PINNED_SOURCES=(
+    "crates/crypto/src/license_token.rs"
+)
+
+# "<name>|<extended regex>". The name is free of "|", so the first "|" splits.
+#
+# The private-key family matches the header phrase without requiring the PEM
+# dashes, so a key pasted into a JSON/YAML string, a doc block, or a Rust
+# string built without the delimiters is still caught. The credential family is
+# format-anchored — fixed prefix plus fixed-width charset — so a match is a
+# credential shape and not prose.
+PATTERNS=(
+    "PEM private key header|${BEGIN_MARK}[[:space:]]+([A-Z0-9]+[[:space:]]+)*PRIVATE KEY"
+    "PuTTY private key file|PuTTY-User-${KEY_MARK}-File"
+    "AWS access key id|(A3T[A-Z0-9]|AKIA|ASIA|ABIA|ACCA)[A-Z0-9]{16}"
+    "GitHub token|gh[pousr]_[A-Za-z0-9]{36}"
+    "GitHub fine-grained token|github_pat_[A-Za-z0-9_]{22,}"
+    "Slack token|xox[abprs]-[A-Za-z0-9-]{10,}"
+    "Stripe live key|sk_live_[0-9a-zA-Z]{20,}"
+    "Google API key|AIza[0-9A-Za-z_-]{35}"
+    "SendGrid API key|SG\.[A-Za-z0-9_-]{20,}\.[A-Za-z0-9_-]{20,}"
+    "npm access token|npm_[A-Za-z0-9]{36}"
+    "PyPI upload token|pypi-AgEIcHlwaS5vcmc[A-Za-z0-9_-]{50,}"
+)
+
+# Exact strings that carry no secret wherever they appear. A hit is excused
+# only if the line stops matching once these exact strings are removed, so a
+# line holding both an example value and a real credential still fails, and
+# editing an entry — swapping a placeholder body for real key material — makes
+# the guard fire again. Entries that stop matching anything are reported as
+# stale, so the list cannot decay into a blanket exclusion.
+#
+# 1-2: rustfs/src/admin/handlers/site_replication.rs negative fixtures for
+#      `validate_peer_connection_inner`, which must reject a private key
+#      submitted where a peer CA certificate is expected. Asserting on the
+#      rejection requires the header in the input; the key bodies are the
+#      literal word "secret".
+# 3-4: AWS's own documented example access key id from the SigV4 test vectors,
+#      which this repository pairs with the equally documented example secret
+#      key across signer, IAM, madmin, and auth tests, plus the deliberate
+#      one-character variant rustfs/src/auth.rs uses to prove key comparison
+#      distinguishes near-identical ids.
+AWS_EXAMPLE_STEM="AKIAIOSFODNN7EXAMPL"
+NON_SECRET_LITERALS=(
+    "-----${BEGIN_MARK} PRIVATE KEY-----\\nsecret\\n-----END PRIVATE KEY-----"
+    "-----${BEGIN_MARK} RSA PRIVATE KEY-----\\nsecret\\n-----END RSA PRIVATE KEY-----"
+    "${AWS_EXAMPLE_STEM}E"
+    "${AWS_EXAMPLE_STEM}F"
+)
+
+run_scan() {
+    cd "$ROOT_DIR"
+
+    # Without this, a scan run outside a work tree would make every `git grep`
+    # fail and the guard would report success having read nothing.
+    if ! git rev-parse --is-inside-work-tree >/dev/null 2>&1; then
+        printf 'Embedded secret guard failed: %s is not a git work tree, so the scan cannot enumerate files\n' \
+            "$ROOT_DIR" >&2
+        return 1
+    fi
+
+    local literal_used=()
+    local i
+    for ((i = 0; i < ${#NON_SECRET_LITERALS[@]}; i++)); do
+        literal_used[i]="0"
+    done
+
+    local status=0
+    local source
+    for source in "${PINNED_SOURCES[@]}"; do
+        if [[ ! -f "$source" ]]; then
+            printf 'Embedded secret guard failed: %s is missing; update PINNED_SOURCES in scripts/check_embedded_secrets.sh after moving it\n' \
+                "$source" >&2
+            status=1
+        fi
+    done
+
+    local entry name pattern hits grep_status hit file rest line_no text trimmed sanitized
+    for entry in "${PATTERNS[@]}"; do
+        name="${entry%%|*}"
+        pattern="${entry#*|}"
+
+        hits=""
+        grep_status=0
+        hits="$(git grep --untracked -I -n -E -e "$pattern" -- .)" || grep_status=$?
+        if [[ "$grep_status" -gt 1 ]]; then
+            printf 'Embedded secret guard failed: git grep exited %s while scanning for %s\n' "$grep_status" "$name" >&2
+            status=1
+            continue
+        fi
+
+        while IFS= read -r hit; do
+            [[ -z "$hit" ]] && continue
+            file="${hit%%:*}"
+            rest="${hit#*:}"
+            line_no="${rest%%:*}"
+            text="${rest#*:}"
+            trimmed="$(printf '%s' "$text" | sed -e 's/^[[:space:]]*//' -e 's/[[:space:]]*$//')"
+
+            sanitized="$trimmed"
+            for ((i = 0; i < ${#NON_SECRET_LITERALS[@]}; i++)); do
+                if [[ "$sanitized" == *"${NON_SECRET_LITERALS[i]}"* ]]; then
+                    sanitized="${sanitized//"${NON_SECRET_LITERALS[i]}"/}"
+                    literal_used[i]="1"
+                fi
+            done
+
+            if ! printf '%s' "$sanitized" | grep -q -E -e "$pattern"; then
+                continue
+            fi
+
+            printf 'Embedded secret guard failed: %s at %s:%s\n    %s\n' "$name" "$file" "$line_no" "$trimmed" >&2
+            status=1
+        done <<<"$hits"
+    done
+
+    for ((i = 0; i < ${#NON_SECRET_LITERALS[@]}; i++)); do
+        if [[ "${literal_used[i]}" != "1" ]]; then
+            printf 'Embedded secret guard failed: stale exemption, nothing matches it any more: %s\n' \
+                "${NON_SECRET_LITERALS[i]}" >&2
+            status=1
+        fi
+    done
+
+    if [[ "$status" -ne 0 ]]; then
+        printf '\nRemove the key material or credential above, and rotate anything that was committed even briefly.\n' >&2
+        printf 'A genuine non-secret match is excused by adding its exact text to NON_SECRET_LITERALS in scripts/check_embedded_secrets.sh with a reason.\n' >&2
+        return 1
+    fi
+
+    printf 'Embedded secret guard passed (no private key material or provider credential literal in the tree).\n'
+    return 0
+}
+
+# One synthetic value per pattern family, assembled from a filler so this
+# function does not match the patterns it exercises.
+self_test_violation_lines() {
+    local fill="QWERTYUIOPASDFGHJKLZXCVBNM0123456789"
+    printf '%s\n' \
+        "-----${BEGIN_MARK} PRIVATE KEY-----" \
+        "PuTTY-User-${KEY_MARK}-File: ssh-rsa" \
+        "AKIA${fill:0:16}" \
+        "ghp_${fill:0:36}" \
+        "github_pat_${fill:0:22}" \
+        "xoxb-${fill:0:12}" \
+        "sk_live_${fill:0:20}" \
+        "AIza${fill:0:35}" \
+        "SG.${fill:0:20}.${fill:0:20}" \
+        "npm_${fill:0:36}" \
+        "pypi-AgEIcHlwaS5vcmc${fill}${fill:0:14}"
+}
+
+self_test_fixture() {
+    local dir="$1" i
+    mkdir -p "${dir}/crates/crypto/src"
+    : >"${dir}/crates/crypto/src/license_token.rs"
+    # Every exemption must appear, or the stale-exemption check fires and the
+    # fixture would fail for a reason the case under test is not about.
+    for ((i = 0; i < ${#NON_SECRET_LITERALS[@]}; i++)); do
+        printf 'excused %s\n' "${NON_SECRET_LITERALS[i]}" >>"${dir}/excused.txt"
+    done
+    git -C "$dir" init -q
+}
+
+SELF_TEST_TMP=""
+
+self_test() {
+    SELF_TEST_TMP="$(mktemp -d)"
+    trap 'rm -rf "$SELF_TEST_TMP"' EXIT
+
+    local failures=0 out scan_status
+    local clean="${SELF_TEST_TMP}/clean" dirty="${SELF_TEST_TMP}/dirty" renamed="${SELF_TEST_TMP}/renamed"
+
+    self_test_fixture "$clean"
+    if out="$(CHECK_EMBEDDED_SECRETS_ROOT="$clean" "$0" 2>&1)"; then
+        printf 'self-test ok: clean fixture with every exemption present passes\n'
+    else
+        printf 'self-test FAILED: clean fixture should pass but reported:\n%s\n' "$out" >&2
+        failures=$((failures + 1))
+    fi
+
+    self_test_fixture "$dirty"
+    self_test_violation_lines >"${dirty}/leaked.txt"
+    scan_status=0
+    out="$(CHECK_EMBEDDED_SECRETS_ROOT="$dirty" "$0" 2>&1)" || scan_status=$?
+    if [[ "$scan_status" -eq 0 ]]; then
+        printf 'self-test FAILED: fixture holding one value per pattern family should fail\n' >&2
+        failures=$((failures + 1))
+    fi
+    local entry name
+    for entry in "${PATTERNS[@]}"; do
+        name="${entry%%|*}"
+        if ! printf '%s' "$out" | grep -q -F -- "$name"; then
+            printf 'self-test FAILED: pattern family "%s" did not fire on its own probe value\n' "$name" >&2
+            failures=$((failures + 1))
+        fi
+    done
+    [[ "$failures" -eq 0 ]] && printf 'self-test ok: all %s pattern families fire\n' "${#PATTERNS[@]}"
+
+    self_test_fixture "$renamed"
+    rm -f "${renamed}/crates/crypto/src/license_token.rs"
+    if CHECK_EMBEDDED_SECRETS_ROOT="$renamed" "$0" >/dev/null 2>&1; then
+        printf 'self-test FAILED: a moved pinned source should be reported\n' >&2
+        failures=$((failures + 1))
+    else
+        printf 'self-test ok: moving a pinned source is reported\n'
+    fi
+
+    if [[ "$failures" -ne 0 ]]; then
+        printf '%s self-test assertion(s) failed\n' "$failures" >&2
+        return 1
+    fi
+    printf 'Embedded secret guard self-test passed.\n'
+    return 0
+}
+
+if [[ "${1:-}" == "--self-test" ]]; then
+    self_test
+else
+    run_scan
+fi