fix(get): name the failing object on mid-stream GET body failures

A GET body that ends short of its committed Content-Length breaks every
downstream copier. Replication, site replication and `rclone sync` all read
locally and PUT remotely, so a truncated source read surfaces as an
`unexpected EOF` on the destination's PUT and an `Io error: error reading a
body from connection` 500 on the receiving server. Issue #4784 stalled for a
month because the source side reported none of it:

* `GetObjectReaderStream`'s short-read and read-error arms only fed a metric.
  Their log lines sat behind the `tracing-chunk-debug` cargo feature, which is
  not in the default feature set and therefore is not compiled into any
  released binary.
* `GetObjectStreamingReader` did log mid-stream failures, but discarded the
  bucket and key its constructor was already handed, leaving only a request_id
  that cannot be resolved back to an object after the request is over.
* Those lines were `warn!`, while DEFAULT_LOG_LEVEL is `error`, so a default
  deployment filtered them out even where they existed.

Keep the object identity on both readers and name it in every stream-body
failure; log the reader-stream short-read arm unconditionally; and raise the
two states that mean "the server cannot deliver the length it already
committed" - short_eof and read_failed - to `error!`. Stall timeouts, slow
first bytes and client-side drops stay at `warn!`.

The reader-stream read-error arm stays feature-gated on purpose: every
production body wraps a GetObjectStreamingReader, which already reports that
same error once with the object identity, so a second unconditional line per
failed GET would read as two distinct faults.

Tests: unit coverage asserts the captured event fields rather than just the
returned error, and a new e2e reproduces the fault over the S3 API against a
beyond-quorum damaged object, asserting the evidence at the default `error`
log level. `get_object_reader_stream_errors_on_short_eof` becomes serial: it
drives the same log callsite as the capture test, and tracing caches callsite
interest process-wide, so running it concurrently on a subscriber-less thread
re-cached that callsite as "never interested" and blinded the capture.

Cargo.lock

@@ -9278,7 +9278,6 @@ dependencies = [
  "jiff",
  "metrics",
  "rmp-serde",
- "s3s",
  "serde",
  "serde_json",
  "smallvec",

crates/common/Cargo.toml

@@ -44,7 +44,6 @@ metrics = { workspace = true }
 serde = { workspace = true, features = ["derive"] }
 smallvec = { workspace = true }
 rmp-serde = { workspace = true }
-s3s = { workspace = true, features = ["minio"] }
 tracing = { workspace = true }
 
 [dev-dependencies]

crates/common/src/heal_channel.rs

@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use s3s::dto::{BucketLifecycleConfiguration, ExpirationStatus, LifecycleRule, ReplicationConfiguration, ReplicationRuleStatus};
 use serde::{Deserialize, Serialize};
 use std::{
     fmt::{self, Display},
@@ -633,104 +632,6 @@ pub fn create_heal_response(
     }
 }
 
-fn lc_get_prefix(rule: &LifecycleRule) -> String {
-    if let Some(p) = &rule.prefix {
-        return p.to_string();
-    } else if let Some(filter) = &rule.filter {
-        if let Some(p) = &filter.prefix {
-            return p.to_string();
-        } else if let Some(and) = &filter.and
-            && let Some(p) = &and.prefix
-        {
-            return p.to_string();
-        }
-    }
-
-    "".into()
-}
-
-pub fn lc_has_active_rules(config: &BucketLifecycleConfiguration, prefix: &str) -> bool {
-    if config.rules.is_empty() {
-        return false;
-    }
-
-    for rule in config.rules.iter() {
-        if rule.status == ExpirationStatus::from_static(ExpirationStatus::DISABLED) {
-            continue;
-        }
-        let rule_prefix = lc_get_prefix(rule);
-        if !prefix.is_empty() && !rule_prefix.is_empty() && !prefix.starts_with(&rule_prefix) && !rule_prefix.starts_with(prefix)
-        {
-            continue;
-        }
-
-        if let Some(e) = &rule.noncurrent_version_expiration {
-            if e.noncurrent_days.is_some() {
-                return true;
-            }
-            if let Some(true) = e.newer_noncurrent_versions.map(|d| d > 0) {
-                return true;
-            }
-        }
-
-        if rule.noncurrent_version_transitions.is_some() {
-            return true;
-        }
-        if let Some(true) = rule.expiration.as_ref().map(|e| e.date.is_some()) {
-            return true;
-        }
-
-        if let Some(true) = rule.expiration.as_ref().map(|e| e.days.is_some()) {
-            return true;
-        }
-
-        if let Some(Some(true)) = rule.expiration.as_ref().map(|e| e.expired_object_delete_marker) {
-            return true;
-        }
-
-        if let Some(true) = rule.transitions.as_ref().map(|t| !t.is_empty()) {
-            return true;
-        }
-
-        if rule.transitions.is_some() {
-            return true;
-        }
-    }
-    false
-}
-
-pub fn rep_has_active_rules(config: &ReplicationConfiguration, prefix: &str, recursive: bool) -> bool {
-    if config.rules.is_empty() {
-        return false;
-    }
-
-    for rule in config.rules.iter() {
-        if rule
-            .status
-            .eq(&ReplicationRuleStatus::from_static(ReplicationRuleStatus::DISABLED))
-        {
-            continue;
-        }
-        if !prefix.is_empty()
-            && let Some(filter) = &rule.filter
-            && let Some(r_prefix) = &filter.prefix
-            && !r_prefix.is_empty()
-        {
-            // incoming prefix must be in rule prefix
-            if !recursive && !prefix.starts_with(r_prefix) {
-                continue;
-            }
-            // If recursive, we can skip this rule if it doesn't match the tested prefix or level below prefix
-            // does not match
-            if recursive && !r_prefix.starts_with(prefix) && !prefix.starts_with(r_prefix) {
-                continue;
-            }
-        }
-        return true;
-    }
-    false
-}
-
 pub async fn send_heal_disk(set_disk_id: String, priority: Option<HealChannelPriority>) -> Result<(), String> {
     let req = HealChannelRequest {
         id: Uuid::new_v4().to_string(),

crates/e2e_test/src/get_stream_failure_observability_test.rs

@@ -0,0 +1,250 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! E2E proof that a mid-stream GET failure is *reportable* — rustfs#4784.
+//!
+//! The functional invariant (a beyond-quorum read must fail rather than return
+//! a clean short body) is already covered by
+//! `degraded_read_eof_regression_test`. This suite covers the half that issue
+//! #4784 got stuck on for a month: whether an operator can tell, from the
+//! source server's log alone, that a GET failed mid-body and **which object**
+//! it failed on.
+//!
+//! The reporter saw only downstream symptoms — `rclone` reporting
+//! `unexpected EOF` on its PUT, and the receiving RustFS logging
+//! `Io error: error reading a body from connection` with a 500. In a cross-remote
+//! `rclone sync`, the source GET body *is* the destination PUT body, so a source
+//! read that ends short of its committed `Content-Length` surfaces as a PUT
+//! failure on the far side. Built-in replication and site replication have the
+//! same shape (read locally, PUT remotely), which is why every transport in that
+//! report failed the same way.
+//!
+//! The source side, meanwhile, said nothing:
+//!   * `GetObjectReaderStream`'s short-read and read-error arms only incremented
+//!     a metric; their log lines sat behind the `tracing-chunk-debug` cargo
+//!     feature, which is not in the default feature set and therefore is not
+//!     compiled into any released binary.
+//!   * `GetObjectStreamingReader` did log mid-stream failures, but only under a
+//!     `request_id` — with no bucket or object name, a failure could not be
+//!     traced back to the object that caused it.
+//!   * Those lines were `warn!`, while `DEFAULT_LOG_LEVEL` is `error`, so a
+//!     default deployment filtered them out anyway.
+//!
+//! This test reproduces the source-side fault against a real server over the S3
+//! API and asserts the operator-visible evidence, at the **default** log level.
+
+#[cfg(test)]
+mod tests {
+    use crate::chaos::DiskFaultHarness;
+    use crate::common::init_logging;
+    use aws_sdk_s3::Client;
+    use aws_sdk_s3::primitives::ByteStream;
+    use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart};
+    use serial_test::serial;
+    use std::error::Error;
+    use tokio::time::{Duration, timeout};
+    use tracing::info;
+
+    type TestResult = Result<(), Box<dyn Error + Send + Sync>>;
+
+    const MIB: usize = 1024 * 1024;
+    const OP_TIMEOUT: Duration = Duration::from_secs(90);
+
+    /// The structured event name every GET body failure is tagged with.
+    const STREAM_BODY_EVENT: &str = "get_object_stream_body";
+
+    fn payload(len: usize, seed: u8) -> Vec<u8> {
+        (0..len)
+            .map(|i| (i as u64).wrapping_mul(2654435761).wrapping_add(seed as u64) as u8)
+            .collect()
+    }
+
+    /// Upload a multipart object so the data lands in real `part.*` shard files
+    /// rather than being inlined into `xl.meta` (inlined objects cannot be
+    /// corrupted shard-wise, and never exercise the streaming read path).
+    async fn put_multipart(
+        client: &Client,
+        bucket: &str,
+        key: &str,
+        parts: Vec<Vec<u8>>,
+    ) -> Result<usize, Box<dyn Error + Send + Sync>> {
+        let total_len = parts.iter().map(Vec::len).sum();
+
+        let create = client.create_multipart_upload().bucket(bucket).key(key).send().await?;
+        let upload_id = create.upload_id().ok_or("missing upload id")?.to_string();
+
+        let mut completed = Vec::with_capacity(parts.len());
+        for (index, part_body) in parts.into_iter().enumerate() {
+            let part_number = (index + 1) as i32;
+            let uploaded = timeout(
+                OP_TIMEOUT,
+                client
+                    .upload_part()
+                    .bucket(bucket)
+                    .key(key)
+                    .upload_id(&upload_id)
+                    .part_number(part_number)
+                    .body(ByteStream::from(part_body))
+                    .send(),
+            )
+            .await
+            .map_err(|_| format!("upload_part {part_number} timed out"))??;
+            completed.push(
+                CompletedPart::builder()
+                    .part_number(part_number)
+                    .e_tag(uploaded.e_tag().ok_or("missing part etag")?)
+                    .build(),
+            );
+        }
+
+        timeout(
+            OP_TIMEOUT,
+            client
+                .complete_multipart_upload()
+                .bucket(bucket)
+                .key(key)
+                .upload_id(&upload_id)
+                .multipart_upload(CompletedMultipartUpload::builder().set_parts(Some(completed)).build())
+                .send(),
+        )
+        .await
+        .map_err(|_| "complete_multipart_upload timed out")??;
+
+        Ok(total_len)
+    }
+
+    /// rustfs#4784: reproduce the source-side fault the reporter kept hitting —
+    /// a GET that commits `200` + a full `Content-Length` and then cannot finish
+    /// the body — and assert the server log names the object, at the log level a
+    /// default deployment actually runs with.
+    #[tokio::test]
+    #[serial]
+    async fn midstream_get_failure_is_logged_with_the_object_at_default_log_level() -> TestResult {
+        init_logging();
+        info!("rustfs#4784: a mid-stream GET failure must name its object in the source log");
+
+        let mut harness = DiskFaultHarness::new(4).await?;
+
+        // Capture the child's stdout so the test can read what an operator would.
+        let log_path = format!("{}/server.log", harness.env.temp_dir);
+        harness.env.capture_log_path = Some(log_path.clone());
+
+        // Reproduce a DEFAULT deployment's logging, not the e2e harness's
+        // permissive `rustfs=info`: `DEFAULT_LOG_LEVEL` is `error`. Before the
+        // #4784 fix these failures were `warn!`, so a default deployment
+        // filtered them out entirely — which is why the reporter's source logs
+        // were empty. extra_env is applied after the harness's own RUST_LOG, so
+        // this wins.
+        harness.set_env("RUST_LOG", "error");
+        harness.set_env("RUSTFS_OBS_LOGGER_LEVEL", "error");
+
+        harness.start_server().await?;
+        let client = harness.env.create_s3_client();
+
+        let bucket = "issue4784-source-read";
+        client.create_bucket().bucket(bucket).send().await?;
+
+        // Named after the reporter's restic index objects, which is where they
+        // saw the failures.
+        let key = "index/3b18542ab3af4c3d03f804c7a24173e7836ef7fa447b5d1e9d634f975cc51611";
+        let expected_len = put_multipart(
+            &client,
+            bucket,
+            key,
+            vec![payload(5 * MIB, 71), payload(5 * MIB, 72), payload(5 * MIB, 73)],
+        )
+        .await?;
+
+        // Baseline: the object reads back completely before any corruption.
+        let baseline = timeout(OP_TIMEOUT, client.get_object().bucket(bucket).key(key).send())
+            .await
+            .map_err(|_| "baseline GET timed out")??
+            .body
+            .collect()
+            .await?;
+        assert_eq!(baseline.into_bytes().len(), expected_len, "baseline GET must return the whole object");
+
+        // Corrupt three of four shards in a 2+2 set: below the 2-shard read
+        // quorum. The corruption sits mid-file, so block 0 still reads clean —
+        // the server commits 200 + the full Content-Length and only then cannot
+        // reconstruct. That is the mid-stream window the reporter's downstream
+        // saw as `unexpected EOF`.
+        harness.corrupt_object_shard(0, bucket, key)?;
+        harness.corrupt_object_shard(1, bucket, key)?;
+        harness.corrupt_object_shard(2, bucket, key)?;
+
+        let response = timeout(OP_TIMEOUT, client.get_object().bucket(bucket).key(key).send())
+            .await
+            .map_err(|_| "degraded GET timed out")?;
+
+        // Either outcome is functionally correct (that invariant belongs to
+        // degraded_read_eof_regression_test); this suite only needs the read to
+        // have failed so there is something to report.
+        let delivered = match response {
+            Err(err) => {
+                info!("degraded GET failed before the body: {err}");
+                None
+            }
+            Ok(response) => match response.body.collect().await {
+                Ok(aggregated) => Some(aggregated.into_bytes().len()),
+                Err(err) => {
+                    info!("degraded GET failed mid-body as expected: {err}");
+                    None
+                }
+            },
+        };
+        assert_ne!(
+            delivered,
+            Some(expected_len),
+            "the beyond-quorum read unexpectedly succeeded; this suite needs a failed read to have something to report"
+        );
+
+        // Give the child a moment to flush its stdout.
+        tokio::time::sleep(Duration::from_millis(500)).await;
+        let logged = std::fs::read_to_string(&log_path)?;
+
+        let failure_lines: Vec<&str> = logged.lines().filter(|line| line.contains(STREAM_BODY_EVENT)).collect();
+
+        assert!(
+            !failure_lines.is_empty(),
+            "a mid-stream GET failure produced no `{STREAM_BODY_EVENT}` line at the default log level. \
+             This is the #4784 blind spot: the failure was only counted in a metric, or logged below \
+             `error` and filtered out. Captured log:\n{logged}"
+        );
+
+        // The identity is the whole point: a request_id alone cannot be resolved
+        // back to an object once the request is over.
+        assert!(
+            failure_lines.iter().any(|line| line.contains(key)),
+            "no `{STREAM_BODY_EVENT}` line named the failing object `{key}`, so the report is still \
+             unactionable. Lines seen:\n{}",
+            failure_lines.join("\n")
+        );
+        assert!(
+            failure_lines.iter().any(|line| line.contains(bucket)),
+            "no `{STREAM_BODY_EVENT}` line named the failing bucket `{bucket}`. Lines seen:\n{}",
+            failure_lines.join("\n")
+        );
+
+        info!(
+            "source-side evidence now present: {} stream-body failure line(s) naming the object",
+            failure_lines.len()
+        );
+        for line in &failure_lines {
+            info!("operator-visible evidence: {line}");
+        }
+
+        Ok(())
+    }
+}

crates/e2e_test/src/lib.rs

@@ -48,6 +48,11 @@ mod replacement_privileged_e2e_test;
 #[cfg(test)]
 mod degraded_read_eof_regression_test;
 
+// rustfs#4784: a mid-stream GET failure must be reportable from the source
+// server's log alone — naming the object, at the default log level.
+#[cfg(test)]
+mod get_stream_failure_observability_test;
+
 // backlog#1183: GET codec-streaming fast path must be byte/header identical to
 // the legacy duplex path before its rollout gates can be flipped on by default.
 #[cfg(test)]

crates/ecstore/src/core/pools.rs

@@ -1041,7 +1041,13 @@ fn should_count_decommission_version_complete(ignore: bool, cleanup_ignored: boo
 fn is_decommission_copy_cleanup_safe_error(err: &Error) -> bool {
     // DataMovementOverwriteErr only means source and destination pool resolved to
     // the same pool. Without a target equivalence check it is not cleanup-safe.
-    is_err_object_not_found(err) || is_err_version_not_found(err)
+    if is_err_object_not_found(err) || is_err_version_not_found(err) {
+        return true;
+    }
+
+    // A not-found surfacing from inside a data-movement stage is the same
+    // condition once the wrapper is unwrapped (backlog#1827 T2).
+    crate::data_movement::data_movement_stage_source(err).is_some_and(is_decommission_copy_cleanup_safe_error)
 }
 
 fn is_decommission_target_capacity_error(err: &Error) -> bool {
@@ -1049,6 +1055,13 @@ fn is_decommission_target_capacity_error(err: &Error) -> bool {
         return true;
     }
 
+    // A stage failure keeps the error it wrapped, so classify by type rather
+    // than by the rendered message (backlog#1827 T2). The substring fallback
+    // stays for errors that reached here through some other wrapper.
+    if let Some(source) = crate::data_movement::data_movement_stage_source(err) {
+        return is_decommission_target_capacity_error(source);
+    }
+
     let message = err.to_string();
     let disk_full = Error::DiskFull.to_string();
     let storage_full = Error::StorageFull.to_string();
@@ -4427,6 +4440,36 @@ mod tests {
         assert!(is_decommission_target_capacity_error(&Error::StorageFull));
     }
 
+    /// The decommission loop classifies errors that came back through a
+    /// data-movement stage wrapper. Before backlog#1827 T2 the wrapper flattened
+    /// everything into `Error::other(String)`, so these two classifiers had to
+    /// match on rendered text; now the wrapped error is recoverable by type.
+    #[test]
+    fn decommission_classifiers_see_through_a_stage_wrapper() {
+        let wrap = |inner: Error| {
+            crate::data_movement::data_movement_stage_error_for_test(
+                "decommission_object",
+                "put_object",
+                "bucket-a",
+                "object-a",
+                inner,
+            )
+        };
+
+        // Capacity: the target pool filling up must still stop the loop.
+        assert!(is_decommission_target_capacity_error(&wrap(Error::DiskFull)));
+        assert!(is_decommission_target_capacity_error(&wrap(Error::StorageFull)));
+        assert!(!is_decommission_target_capacity_error(&wrap(Error::SlowDown)));
+
+        // Cleanup safety: a not-found surfacing from inside a stage is the same
+        // condition as one surfacing directly, so the source entry stays
+        // eligible for cleanup.
+        let not_found = Error::ObjectNotFound("bucket-a".to_string(), "object-a".to_string());
+        assert!(is_decommission_copy_cleanup_safe_error(&not_found));
+        assert!(is_decommission_copy_cleanup_safe_error(&wrap(not_found)));
+        assert!(!is_decommission_copy_cleanup_safe_error(&wrap(Error::SlowDown)));
+    }
+
     #[test]
     fn decommission_target_capacity_error_accepts_wrapped_capacity_errors() {
         let disk_full = Error::other(format!("decommission_object: put_object failed for bucket/object: {}", Error::DiskFull));

crates/ecstore/src/data_movement/mod.rs

@@ -471,8 +471,60 @@ fn resolve_data_movement_abort_result(
     ))
 }
 
-fn data_movement_stage_error(op_label: &str, stage: &str, bucket: &str, object: &str, err: impl std::fmt::Display) -> Error {
-    Error::other(format!("{op_label}: {stage} failed for {bucket}/{object}: {err}"))
+/// A data-movement stage failure that keeps the error it wrapped.
+///
+/// The rendered message is byte-identical to the `format!` this replaced, so
+/// logs and any message-matching callers are unaffected. What changes is that
+/// the original error stays reachable through `source()`, which is what lets
+/// the decommission loop classify by type instead of by substring
+/// (backlog#1827 T2).
+#[derive(Debug)]
+struct DataMovementStageError {
+    rendered: String,
+    source: Box<dyn std::error::Error + Send + Sync>,
+}
+
+impl std::fmt::Display for DataMovementStageError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str(&self.rendered)
+    }
+}
+
+impl std::error::Error for DataMovementStageError {
+    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+        Some(self.source.as_ref())
+    }
+}
+
+fn data_movement_stage_error<E>(op_label: &str, stage: &str, bucket: &str, object: &str, err: E) -> Error
+where
+    E: std::error::Error + Send + Sync + 'static,
+{
+    let rendered = format!("{op_label}: {stage} failed for {bucket}/{object}: {err}");
+    Error::other(DataMovementStageError {
+        rendered,
+        source: Box::new(err),
+    })
+}
+
+#[cfg(test)]
+pub(crate) fn data_movement_stage_error_for_test(op_label: &str, stage: &str, bucket: &str, object: &str, err: Error) -> Error {
+    data_movement_stage_error(op_label, stage, bucket, object, err)
+}
+
+/// Recover the error a [`data_movement_stage_error`] wrapped, if this is one.
+///
+/// `Error::other` boxes through `std::io::Error`, so the chain is
+/// `StorageError::Io` -> `DataMovementStageError` -> the original error.
+pub(crate) fn data_movement_stage_source(err: &Error) -> Option<&Error> {
+    let Error::Io(io_err) = err else {
+        return None;
+    };
+    io_err
+        .get_ref()?
+        .downcast_ref::<DataMovementStageError>()?
+        .source
+        .downcast_ref::<Error>()
 }
 
 fn schedule_data_movement_multipart_abort_cleanup(
@@ -1865,6 +1917,40 @@ mod tests {
         assert!(message.contains(Error::SlowDown.to_string().as_str()));
     }
 
+    #[test]
+    fn stage_error_renders_exactly_as_the_format_it_replaced() {
+        // The wrapper gained a source; its message must not have moved, or log
+        // scrapers and any message-matching caller would break (backlog#1827 T2).
+        // `Error::other` renders through `StorageError::Io`, which prefixes
+        // "Io error: " — that was true of the `format!` this replaced too, so
+        // the full string is what must stay stable.
+        let err = data_movement_stage_error("rebalance_object", "put_object", "bucket-a", "object-a", Error::SlowDown);
+        assert_eq!(
+            err.to_string(),
+            format!("Io error: rebalance_object: put_object failed for bucket-a/object-a: {}", Error::SlowDown)
+        );
+        assert_eq!(
+            err.to_string(),
+            Error::other(format!("rebalance_object: put_object failed for bucket-a/object-a: {}", Error::SlowDown)).to_string()
+        );
+    }
+
+    #[test]
+    fn stage_error_keeps_the_wrapped_error_recoverable() {
+        for original in [Error::DiskFull, Error::StorageFull, Error::FileNotFound, Error::SlowDown] {
+            let wrapped =
+                data_movement_stage_error("decommission_object", "put_object", "bucket-a", "object-a", original.clone());
+            let recovered = data_movement_stage_source(&wrapped).expect("the wrapped error must be recoverable");
+            assert_eq!(recovered.to_string(), original.to_string());
+        }
+    }
+
+    #[test]
+    fn stage_source_ignores_errors_it_did_not_wrap() {
+        assert!(data_movement_stage_source(&Error::DiskFull).is_none());
+        assert!(data_movement_stage_source(&Error::other("plain io error")).is_none());
+    }
+
     #[test]
     fn test_data_movement_part_stage_error_includes_stage_object_and_part() {
         let err =

crates/heal/src/heal/manager.rs

@@ -227,11 +227,11 @@ impl ForegroundPressure {
 struct CompletedHealStatus {
     heal_type: HealType,
     status: HealTaskStatus,
-    result_items: Vec<HealResultItem>,
     result_items_truncated: bool,
     completed_at: SystemTime,
     /// Sequence-stamped retained window, archived with the completion so
     /// incremental consumers keep their cursor across the transition (HS-06).
+    /// The un-stamped legacy view is derived from it on demand.
     seqed_items: Vec<(u64, HealResultItem)>,
     next_seq: u64,
     min_seq: u64,
@@ -293,7 +293,7 @@ fn empty_task_report(status: HealTaskStatus) -> HealTaskReport {
 fn completed_task_report(completed: &CompletedHealStatus, since: Option<u64>) -> HealTaskReport {
     let mut lagged = false;
     let result_items = match since {
-        None => completed.result_items.clone(),
+        None => completed.seqed_items.iter().map(|(_, item)| item.clone()).collect(),
         Some(cursor) => {
             if cursor + 1 < completed.min_seq {
                 lagged = true;
@@ -1027,8 +1027,10 @@ pub struct HealManager {
     active_heals: Arc<Mutex<HashMap<String, Arc<HealTask>>>>,
     /// Heal queue (priority-based)
     heal_queue: Arc<Mutex<PriorityHealQueue>>,
-    /// Recently completed heal statuses retained for status queries.
-    completed_heals: Arc<Mutex<HashMap<String, CompletedHealStatus>>>,
+    /// Recently completed heal statuses retained for status queries. Values
+    /// are shared so the lookup helper can hand a completed entry to a
+    /// caller without cloning the retained result window.
+    completed_heals: Arc<Mutex<HashMap<String, Arc<CompletedHealStatus>>>>,
     /// Client tokens merged into an existing task id.
     task_aliases: Arc<Mutex<HashMap<String, HealTaskAlias>>>,
     /// Heal tasks waiting for a retry backoff to expire.
@@ -1051,10 +1053,21 @@ pub struct HealManager {
     workload_provider: Option<WorkloadSnapshotProviderRef>,
 }
 
+/// Where a task-id lookup resolved. The variants carry the resolved state
+/// so both the status and the report adapters can consume one shared
+/// cascade without re-locking.
+enum TaskStateLookup {
+    Active(Arc<HealTask>),
+    Retrying(HealTaskStatus),
+    Completed(Arc<CompletedHealStatus>),
+    Queued,
+    NotFound,
+}
+
 struct HealQueueContext<'a> {
     heal_queue: &'a Arc<Mutex<PriorityHealQueue>>,
     active_heals: &'a Arc<Mutex<HashMap<String, Arc<HealTask>>>>,
-    completed_heals: &'a Arc<Mutex<HashMap<String, CompletedHealStatus>>>,
+    completed_heals: &'a Arc<Mutex<HashMap<String, Arc<CompletedHealStatus>>>>,
     retrying_heals: &'a Arc<Mutex<HashMap<String, RetryingHeal>>>,
     replacement_recovery_anchors: &'a Arc<std::sync::Mutex<HashMap<String, String>>>,
     config: &'a Arc<RwLock<HealConfig>>,
@@ -2160,47 +2173,79 @@ impl HealManager {
     }
 
     /// Get task status
-    pub async fn get_task_status(&self, task_id: &str) -> Result<HealTaskStatus> {
-        let canonical_task_id = self.canonical_task_id(task_id).await;
+    /// Ordered task-state lookup shared by every status/report query. The
+    /// map precedence mirrors the historical per-method cascades exactly:
+    /// active, then retrying, then completed — where a completed entry in a
+    /// retrying state outranks the queue so a retrying task reports
+    /// Retrying, never Pending — then the queue, and finally a terminal
+    /// completed entry. `heal_path` additionally constrains the map matches
+    /// the way the `*_for_path` variants always have.
+    async fn lookup_task_state(&self, canonical_task_id: &str, heal_path: Option<&str>) -> TaskStateLookup {
+        let matches_path = |heal_type: &HealType| heal_path.is_none_or(|path| heal_type_matches_path(heal_type, path));
+
         {
             let active_heals = self.active_heals.lock().await;
-            if let Some(task) = active_heals.get(&canonical_task_id) {
-                return Ok(task.get_status().await);
+            if let Some(task) = active_heals
+                .get(canonical_task_id)
+                .filter(|task| matches_path(&task.heal_type))
+            {
+                return TaskStateLookup::Active(Arc::clone(task));
             }
         }
 
         {
             let retrying_heals = self.retrying_heals.lock().await;
-            if let Some(retrying) = retrying_heals.get(&canonical_task_id) {
-                return Ok(retrying.status());
+            if let Some(retrying) = retrying_heals
+                .get(canonical_task_id)
+                .filter(|retrying| matches_path(&retrying.request.heal_type))
+            {
+                return TaskStateLookup::Retrying(retrying.status());
+            }
+        }
+
+        // One completed-map pass (single lock + prune): a retrying completion
+        // returns immediately; a terminal completion is held back until the
+        // queue has been checked, so queued work outranks it.
+        let mut terminal_completed: Option<Arc<CompletedHealStatus>> = None;
+        {
+            let mut completed_heals = self.completed_heals.lock().await;
+            prune_completed_heal_statuses(&mut completed_heals);
+            if let Some(completed) = completed_heals.get(canonical_task_id).filter(|c| matches_path(&c.heal_type)) {
+                if completed_status_is_retrying(&completed.status) {
+                    return TaskStateLookup::Completed(Arc::clone(completed));
+                }
+                terminal_completed = Some(Arc::clone(completed));
             }
         }
 
         {
-            let mut completed_heals = self.completed_heals.lock().await;
-            prune_completed_heal_statuses(&mut completed_heals);
-            if let Some(completed) = completed_heals.get(&canonical_task_id)
-                && completed_status_is_retrying(&completed.status)
-            {
-                return Ok(completed.status.clone());
+            let queue = self.heal_queue.lock().await;
+            let queued = match heal_path {
+                Some(path) => queue.contains_request_id_matching_path(canonical_task_id, path),
+                None => queue.contains_request_id(canonical_task_id),
+            };
+            if queued {
+                return TaskStateLookup::Queued;
             }
         }
 
-        let queue = self.heal_queue.lock().await;
-        if queue.contains_request_id(&canonical_task_id) {
-            return Ok(HealTaskStatus::Pending);
+        match terminal_completed {
+            Some(completed) => TaskStateLookup::Completed(completed),
+            None => TaskStateLookup::NotFound,
         }
-        drop(queue);
+    }
 
-        let mut completed_heals = self.completed_heals.lock().await;
-        prune_completed_heal_statuses(&mut completed_heals);
-        if let Some(completed) = completed_heals.get(&canonical_task_id) {
-            return Ok(completed.status.clone());
+    pub async fn get_task_status(&self, task_id: &str) -> Result<HealTaskStatus> {
+        let canonical_task_id = self.canonical_task_id(task_id).await;
+        match self.lookup_task_state(&canonical_task_id, None).await {
+            TaskStateLookup::Active(task) => Ok(task.get_status().await),
+            TaskStateLookup::Retrying(status) => Ok(status),
+            TaskStateLookup::Completed(completed) => Ok(completed.status.clone()),
+            TaskStateLookup::Queued => Ok(HealTaskStatus::Pending),
+            TaskStateLookup::NotFound => Err(Error::TaskNotFound {
+                task_id: task_id.to_string(),
+            }),
         }
-
-        Err(Error::TaskNotFound {
-            task_id: task_id.to_string(),
-        })
     }
 
     pub async fn get_task_report(&self, task_id: &str) -> Result<HealTaskReport> {
@@ -2212,46 +2257,15 @@ impl HealManager {
     /// full-snapshot semantics.
     pub async fn get_task_report_since(&self, task_id: &str, since: Option<u64>) -> Result<HealTaskReport> {
         let canonical_task_id = self.canonical_task_id(task_id).await;
-        {
-            let active_heals = self.active_heals.lock().await;
-            if let Some(task) = active_heals.get(&canonical_task_id) {
-                return Ok(active_task_report(task, since).await);
-            }
+        match self.lookup_task_state(&canonical_task_id, None).await {
+            TaskStateLookup::Active(task) => Ok(active_task_report(&task, since).await),
+            TaskStateLookup::Retrying(status) => Ok(empty_task_report(status)),
+            TaskStateLookup::Completed(completed) => Ok(completed_task_report(&completed, since)),
+            TaskStateLookup::Queued => Ok(empty_task_report(HealTaskStatus::Pending)),
+            TaskStateLookup::NotFound => Err(Error::TaskNotFound {
+                task_id: task_id.to_string(),
+            }),
         }
-
-        {
-            let retrying_heals = self.retrying_heals.lock().await;
-            if let Some(retrying) = retrying_heals.get(&canonical_task_id) {
-                return Ok(empty_task_report(retrying.status()));
-            }
-        }
-
-        {
-            let mut completed_heals = self.completed_heals.lock().await;
-            prune_completed_heal_statuses(&mut completed_heals);
-            if let Some(completed) = completed_heals.get(&canonical_task_id)
-                && completed_status_is_retrying(&completed.status)
-            {
-                return Ok(completed_task_report(completed, since));
-            }
-        }
-
-        {
-            let queue = self.heal_queue.lock().await;
-            if queue.contains_request_id(&canonical_task_id) {
-                return Ok(empty_task_report(HealTaskStatus::Pending));
-            }
-        }
-
-        let mut completed_heals = self.completed_heals.lock().await;
-        prune_completed_heal_statuses(&mut completed_heals);
-        if let Some(completed) = completed_heals.get(&canonical_task_id) {
-            return Ok(completed_task_report(completed, since));
-        }
-
-        Err(Error::TaskNotFound {
-            task_id: task_id.to_string(),
-        })
     }
 
     pub async fn get_task_report_for_path(&self, heal_path: &str, task_id: &str) -> Result<HealTaskReport> {
@@ -2266,59 +2280,20 @@ impl HealManager {
         since: Option<u64>,
     ) -> Result<HealTaskReport> {
         let canonical_task_id = self.canonical_task_id(task_id).await;
-        {
-            let active_heals = self.active_heals.lock().await;
-            if let Some(task) = active_heals.get(&canonical_task_id)
-                && heal_type_matches_path(&task.heal_type, heal_path)
-            {
-                return Ok(active_task_report(task, since).await);
+        match self.lookup_task_state(&canonical_task_id, Some(heal_path)).await {
+            TaskStateLookup::Active(task) => Ok(active_task_report(&task, since).await),
+            TaskStateLookup::Retrying(status) => Ok(empty_task_report(status)),
+            TaskStateLookup::Completed(completed) => Ok(completed_task_report(&completed, since)),
+            TaskStateLookup::Queued => Ok(empty_task_report(HealTaskStatus::Pending)),
+            TaskStateLookup::NotFound => {
+                if self.path_has_task(heal_path).await {
+                    return Err(Error::InvalidClientToken);
+                }
+                Err(Error::TaskNotFound {
+                    task_id: task_id.to_string(),
+                })
             }
         }
-
-        {
-            let retrying_heals = self.retrying_heals.lock().await;
-            if let Some(retrying) = retrying_heals.get(&canonical_task_id)
-                && heal_type_matches_path(&retrying.request.heal_type, heal_path)
-            {
-                return Ok(empty_task_report(retrying.status()));
-            }
-        }
-
-        {
-            let mut completed_heals = self.completed_heals.lock().await;
-            prune_completed_heal_statuses(&mut completed_heals);
-            if let Some(completed) = completed_heals.get(&canonical_task_id)
-                && heal_type_matches_path(&completed.heal_type, heal_path)
-                && completed_status_is_retrying(&completed.status)
-            {
-                return Ok(completed_task_report(completed, since));
-            }
-        }
-
-        {
-            let queue = self.heal_queue.lock().await;
-            if queue.contains_request_id_matching_path(&canonical_task_id, heal_path) {
-                return Ok(empty_task_report(HealTaskStatus::Pending));
-            }
-        }
-
-        {
-            let mut completed_heals = self.completed_heals.lock().await;
-            prune_completed_heal_statuses(&mut completed_heals);
-            if let Some(completed) = completed_heals.get(&canonical_task_id)
-                && heal_type_matches_path(&completed.heal_type, heal_path)
-            {
-                return Ok(completed_task_report(completed, since));
-            }
-        }
-
-        if self.path_has_task(heal_path).await {
-            return Err(Error::InvalidClientToken);
-        }
-
-        Err(Error::TaskNotFound {
-            task_id: task_id.to_string(),
-        })
     }
 
     /// Get task status for a path-bound client token.
@@ -2328,59 +2303,20 @@ impl HealManager {
     /// recently completed task, a different token is invalid for that path.
     pub async fn get_task_status_for_path(&self, heal_path: &str, task_id: &str) -> Result<HealTaskStatus> {
         let canonical_task_id = self.canonical_task_id(task_id).await;
-        {
-            let active_heals = self.active_heals.lock().await;
-            if let Some(task) = active_heals.get(&canonical_task_id)
-                && heal_type_matches_path(&task.heal_type, heal_path)
-            {
-                return Ok(task.get_status().await);
+        match self.lookup_task_state(&canonical_task_id, Some(heal_path)).await {
+            TaskStateLookup::Active(task) => Ok(task.get_status().await),
+            TaskStateLookup::Retrying(status) => Ok(status),
+            TaskStateLookup::Completed(completed) => Ok(completed.status.clone()),
+            TaskStateLookup::Queued => Ok(HealTaskStatus::Pending),
+            TaskStateLookup::NotFound => {
+                if self.path_has_task(heal_path).await {
+                    return Err(Error::InvalidClientToken);
+                }
+                Err(Error::TaskNotFound {
+                    task_id: task_id.to_string(),
+                })
             }
         }
-
-        {
-            let retrying_heals = self.retrying_heals.lock().await;
-            if let Some(retrying) = retrying_heals.get(&canonical_task_id)
-                && heal_type_matches_path(&retrying.request.heal_type, heal_path)
-            {
-                return Ok(retrying.status());
-            }
-        }
-
-        {
-            let mut completed_heals = self.completed_heals.lock().await;
-            prune_completed_heal_statuses(&mut completed_heals);
-            if let Some(completed) = completed_heals.get(&canonical_task_id)
-                && heal_type_matches_path(&completed.heal_type, heal_path)
-                && completed_status_is_retrying(&completed.status)
-            {
-                return Ok(completed.status.clone());
-            }
-        }
-
-        {
-            let queue = self.heal_queue.lock().await;
-            if queue.contains_request_id_matching_path(&canonical_task_id, heal_path) {
-                return Ok(HealTaskStatus::Pending);
-            }
-        }
-
-        {
-            let mut completed_heals = self.completed_heals.lock().await;
-            prune_completed_heal_statuses(&mut completed_heals);
-            if let Some(completed) = completed_heals.get(&canonical_task_id)
-                && heal_type_matches_path(&completed.heal_type, heal_path)
-            {
-                return Ok(completed.status.clone());
-            }
-        }
-
-        if self.path_has_task(heal_path).await {
-            return Err(Error::InvalidClientToken);
-        }
-
-        Err(Error::TaskNotFound {
-            task_id: task_id.to_string(),
-        })
     }
 
     async fn path_has_task(&self, heal_path: &str) -> bool {
@@ -3503,20 +3439,23 @@ impl HealManager {
                             completed_task.get_status().await
                         };
                         let completed_progress = completed_task.get_progress().await;
-                        let final_window = completed_task.get_result_items_since(None).await;
+                        // Single snapshot of the retained window: the task is
+                        // finished and already off the active map, so there is
+                        // no concurrent writer to race with.
+                        let seqed_items = completed_task.get_seqed_result_items().await;
+                        let (next_seq, min_seq) = completed_task.result_seq_cursors();
                         let completed_status_entry = CompletedHealStatus {
                             heal_type: completed_task.heal_type.clone(),
                             status: completed_status.clone(),
-                            result_items: final_window.items.clone(),
                             result_items_truncated: completed_task.result_items_truncated(),
                             completed_at: SystemTime::now(),
-                            seqed_items: completed_task.get_seqed_result_items().await,
-                            next_seq: final_window.next_seq,
-                            min_seq: final_window.min_seq,
+                            seqed_items,
+                            next_seq,
+                            min_seq,
                         };
                         let mut completed_heals_guard = completed_heals_clone.lock().await;
                         prune_completed_heal_statuses(&mut completed_heals_guard);
-                        completed_heals_guard.insert(task_id.clone(), completed_status_entry);
+                        completed_heals_guard.insert(task_id.clone(), Arc::new(completed_status_entry));
                         // update statistics
                         let mut stats = statistics_clone.write().await;
                         match completed_status {
@@ -3808,7 +3747,7 @@ fn heal_request_set_key_for_task(task: &HealTask) -> Option<String> {
     }
 }
 
-fn prune_completed_heal_statuses(completed_heals: &mut HashMap<String, CompletedHealStatus>) {
+fn prune_completed_heal_statuses(completed_heals: &mut HashMap<String, Arc<CompletedHealStatus>>) {
     let Ok(now) = SystemTime::now().duration_since(SystemTime::UNIX_EPOCH) else {
         return;
     };
@@ -5275,19 +5214,18 @@ mod tests {
         );
         manager.completed_heals.lock().await.insert(
             task_id,
-            CompletedHealStatus {
+            Arc::new(CompletedHealStatus {
                 heal_type: request.heal_type,
                 status: HealTaskStatus::Retrying {
                     error: "Lock acquisition timeout".to_string(),
                     retry_attempt: request.retry_attempts,
                 },
-                result_items: Vec::new(),
                 result_items_truncated: false,
                 seqed_items: Vec::new(),
                 next_seq: 0,
                 min_seq: 0,
                 completed_at: SystemTime::now(),
-            },
+            }),
         );
         cancel_token
     }
@@ -5985,6 +5923,47 @@ mod tests {
         assert!(report.result_items.is_empty());
     }
 
+    #[tokio::test]
+    async fn test_retrying_completion_outranks_the_queue_for_the_same_id() {
+        let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
+        let manager = HealManager::new(storage, None);
+
+        // A completed entry recorded in a Retrying state for a task whose
+        // request is also (still) queued under the same id: the retrying
+        // completion must win the lookup, or the task would read back as
+        // Pending while it is actually waiting out a retry backoff.
+        let request = HealRequest::object("bucket".to_string(), "object".to_string(), None);
+        let task_id = request.id.clone();
+        manager.completed_heals.lock().await.insert(
+            task_id.clone(),
+            Arc::new(CompletedHealStatus {
+                heal_type: request.heal_type.clone(),
+                status: HealTaskStatus::Retrying {
+                    error: "transient disk failure".to_string(),
+                    retry_attempt: 1,
+                },
+                result_items_truncated: false,
+                seqed_items: Vec::new(),
+                next_seq: 0,
+                min_seq: 0,
+                completed_at: SystemTime::now(),
+            }),
+        );
+        manager.heal_queue.lock().await.push(HealRequest {
+            id: task_id.clone(),
+            heal_type: request.heal_type,
+            ..request
+        });
+
+        assert_eq!(
+            manager.get_task_status(&task_id).await.expect("task must resolve"),
+            HealTaskStatus::Retrying {
+                error: "transient disk failure".to_string(),
+                retry_attempt: 1
+            }
+        );
+    }
+
     #[tokio::test]
     async fn test_get_task_status_reads_recent_completed_status() {
         let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
@@ -5992,18 +5971,17 @@ mod tests {
 
         manager.completed_heals.lock().await.insert(
             "completed-token".to_string(),
-            CompletedHealStatus {
+            Arc::new(CompletedHealStatus {
                 heal_type: HealType::Bucket {
                     bucket: "bucket".to_string(),
                 },
                 status: HealTaskStatus::Completed,
-                result_items: Vec::new(),
                 result_items_truncated: false,
                 seqed_items: Vec::new(),
                 next_seq: 0,
                 min_seq: 0,
                 completed_at: SystemTime::now(),
-            },
+            }),
         );
 
         assert_eq!(
@@ -6022,25 +6000,27 @@ mod tests {
 
         manager.completed_heals.lock().await.insert(
             "completed-token".to_string(),
-            CompletedHealStatus {
+            Arc::new(CompletedHealStatus {
                 heal_type: HealType::Object {
                     bucket: "bucket".to_string(),
                     object: "object".to_string(),
                     version_id: None,
                 },
                 status: HealTaskStatus::Completed,
-                result_items: vec![HealResultItem {
-                    bucket: "bucket".to_string(),
-                    object: "object".to_string(),
-                    object_size: 1024,
-                    ..Default::default()
-                }],
                 result_items_truncated: true,
-                seqed_items: Vec::new(),
-                next_seq: 0,
-                min_seq: 0,
+                seqed_items: vec![(
+                    1,
+                    HealResultItem {
+                        bucket: "bucket".to_string(),
+                        object: "object".to_string(),
+                        object_size: 1024,
+                        ..Default::default()
+                    },
+                )],
+                next_seq: 2,
+                min_seq: 1,
                 completed_at: SystemTime::now(),
-            },
+            }),
         );
 
         let report = manager
@@ -6052,6 +6032,10 @@ mod tests {
         assert_eq!(report.status, HealTaskStatus::Completed);
         assert_eq!(report.result_items.len(), 1);
         assert_eq!(report.result_items[0].object_size, 1024);
+        // The archived cursors pass through to the report so an incremental
+        // consumer can resume against the next expected sequence.
+        assert_eq!(report.next_seq, 2);
+        assert_eq!(report.min_seq, 1);
     }
 
     #[tokio::test]

crates/heal/src/heal/task.rs

@@ -29,6 +29,7 @@ use rustfs_madmin::heal_commands::HealResultItem;
 use rustfs_utils::path::SLASH_SEPARATOR;
 use serde::{Deserialize, Serialize};
 use std::{
+    collections::VecDeque,
     future::Future,
     sync::{
         Arc,
@@ -384,7 +385,7 @@ pub struct HealTask {
     /// monotonically increasing sequence number for incremental consumption
     /// (the client passes the last seen seq back and receives only newer
     /// items; see `get_result_items_since`).
-    pub result_items: Arc<RwLock<Vec<(u64, HealResultItem)>>>,
+    pub result_items: Arc<RwLock<VecDeque<(u64, HealResultItem)>>>,
     /// Next sequence number to assign; starts at 1.
     next_item_seq: Arc<AtomicU64>,
     /// Sequence number of the oldest item still inside the retention window;
@@ -440,7 +441,7 @@ impl HealTask {
             replacement_resume_endpoint: None,
             status: Arc::new(RwLock::new(HealTaskStatus::Pending)),
             progress: Arc::new(RwLock::new(HealProgress::new())),
-            result_items: Arc::new(RwLock::new(Vec::new())),
+            result_items: Arc::new(RwLock::new(VecDeque::with_capacity(MAX_RETAINED_HEAL_RESULT_ITEMS))),
             next_item_seq: Arc::new(AtomicU64::new(1)),
             min_available_seq: Arc::new(AtomicU64::new(1)),
             result_items_truncated: Arc::new(AtomicBool::new(false)),
@@ -931,7 +932,14 @@ impl HealTask {
     /// Sequence-stamped retained window, used when archiving a completed
     /// task so incremental cursors survive the transition (HS-06).
     pub async fn get_seqed_result_items(&self) -> Vec<(u64, HealResultItem)> {
-        self.result_items.read().await.clone()
+        self.result_items.read().await.iter().cloned().collect::<Vec<_>>()
+    }
+
+    /// Sequence cursors of the retained window (next to assign, oldest
+    /// retained) — the same pair `get_result_items_since` reports, without
+    /// copying the items. Used when archiving a finished task.
+    pub fn result_seq_cursors(&self) -> (u64, u64) {
+        (self.next_item_seq.load(Ordering::Relaxed), self.min_available_seq.load(Ordering::Relaxed))
     }
 
     /// Incremental result window (HS-06): `since = None` returns the full
@@ -974,14 +982,14 @@ impl HealTask {
         let seq = self.next_item_seq.fetch_add(1, Ordering::Relaxed);
         let mut result_items = self.result_items.write().await;
         if result_items.len() < MAX_RETAINED_HEAL_RESULT_ITEMS {
-            result_items.push((seq, result));
+            result_items.push_back((seq, result));
         } else {
             // Slide the window: the oldest item leaves and the cursor for the
             // oldest still-available item moves forward with it.
-            result_items.remove(0);
+            result_items.pop_front();
             self.min_available_seq
-                .store(result_items.first().map_or(seq, |(oldest, _)| *oldest), Ordering::Relaxed);
-            result_items.push((seq, result));
+                .store(result_items.front().map_or(seq, |(oldest, _)| *oldest), Ordering::Relaxed);
+            result_items.push_back((seq, result));
             self.result_items_truncated.store(true, Ordering::Relaxed);
         }
     }

crates/madmin/src/client.rs

@@ -368,9 +368,10 @@ impl AdminClient {
         }
     }
 
-    /// Cluster-aggregated background heal status.
+    /// Cluster-aggregated background heal status. The route is registered
+    /// POST-only on the server, so this must not go out as a GET.
     pub async fn background_heal_status(&self) -> Result<BackgroundHealStatus, AdminClientError> {
-        self.get_json("/v3/background-heal/status").await
+        self.post_json("/v3/background-heal/status", &[], Vec::new()).await
     }
 
     /// Data scanner status (enabled state, freshness, runtime config).
@@ -698,6 +699,21 @@ mod tests {
         assert!(!request.query.contains("clientToken"));
     }
 
+    #[tokio::test]
+    async fn background_heal_status_posts_to_the_registered_route() {
+        let body = r#"{"state":"idle","healQueueLength":0,"healActiveTasks":0,"clusterStatusComplete":true}"#;
+        let server = TestServer::spawn(body, 200).await;
+        let client = AdminClient::new(&format!("http://{}", server.addr), "ak", "sk").unwrap();
+
+        let status = client.background_heal_status().await.expect("status decodes");
+        assert_eq!(status.state, "idle");
+        let request = server.recorded();
+        // The server registers this route POST-only; a GET here answers 405.
+        assert_eq!(request.method, "POST");
+        assert_eq!(request.path, "/rustfs/admin/v3/background-heal/status");
+        assert_eq!(request.query, "");
+    }
+
     #[tokio::test]
     async fn http_error_status_maps_to_a_typed_error_with_body() {
         let server = TestServer::spawn(r#"{"code":"AccessDenied","message":"denied"}"#, 403).await;

crates/scanner/src/lib.rs

@@ -26,6 +26,8 @@ use http::HeaderMap;
 use rustfs_config::server_config::{Config as ServerConfig, get_global_server_config as config_get_global_server_config};
 use std::path::PathBuf;
 use std::sync::Arc;
+use std::sync::RwLock;
+use std::time::{Duration, Instant};
 use storage_api::owner::{
     ECSTORE_BUCKET_META_PREFIX, ECSTORE_RUSTFS_META_BUCKET, ECSTORE_STORAGE_FORMAT_FILE, ECSTORE_STORAGECLASS_RRS,
     ECSTORE_STORAGECLASS_STANDARD, ECSTORE_TRANSITION_COMPLETE, EcstoreBucketTargetSys, EcstoreBucketVersioningSys, EcstoreDisk,
@@ -33,7 +35,7 @@ use storage_api::owner::{
     EcstoreDiskResult, EcstoreErrorType, EcstoreEvaluator, EcstoreEvent, EcstoreLcEventSrc, EcstoreLifecycle,
     EcstoreListPathRawOptions, EcstoreNsScannerOpenRequest, EcstoreObjectOpts, EcstoreReplicationConfigurationExt,
     EcstoreReplicationScannerBridge, EcstoreResultType, EcstoreScanGuard, EcstoreSetDisks, EcstoreStorageError, EcstoreStore,
-    EcstoreTierConfig, EcstoreVersioningApi, HTTPPreconditions, HTTPRangeSpec, ObjectIO, ObjectOperations, ObjectToDelete,
+    EcstoreVersioningApi, HTTPPreconditions, HTTPRangeSpec, ObjectIO, ObjectOperations, ObjectToDelete,
     ScannerReplicationHealObject, ScannerReplicationHealResult, ScannerReplicationQueueAdmission, ecstore_apply_expiry_rule,
     ecstore_apply_transition_rule, ecstore_expiry_state_handle, ecstore_get_global_tier_config_mgr, ecstore_get_lifecycle_config,
     ecstore_get_object_lock_config, ecstore_get_replication_config, ecstore_invalidate_admin_data_usage_snapshot_cache,
@@ -363,8 +365,46 @@ pub(crate) fn resolve_scanner_server_config() -> Option<ServerConfig> {
     config_get_global_server_config()
 }
 
-pub(crate) async fn list_runtime_tiers() -> Vec<EcstoreTierConfig> {
-    ecstore_get_global_tier_config_mgr().read().await.list_tiers()
+/// How long the scanner caches the runtime tier-name list before re-reading
+/// the tier configuration manager.
+const TIER_NAME_CACHE_TTL: Duration = Duration::from_secs(30);
+
+/// Process-wide TTL cache of runtime tier names.
+///
+/// The scan hot path only needs tier *names* to seed `SizeSummary::tier_stats`
+/// per object, but every `list_tiers()` call clones each full `TierConfig`
+/// (endpoints, credentials, prefixes) from the global manager. Caching just
+/// the names keeps the per-object cost at an `Arc` clone.
+///
+/// Staleness bounds: a newly added tier starts showing up in scans at most
+/// `TIER_NAME_CACHE_TTL` later; a removed tier can leave an all-zero
+/// `TierStats` seed behind for one cache generation, which merges harmlessly
+/// by key in per-object accounting and disappears on the next refresh.
+static TIER_NAME_CACHE: RwLock<Option<(Instant, Arc<[String]>)>> = RwLock::new(None);
+
+/// Tier names currently registered in the tier configuration, cached for
+/// `TIER_NAME_CACHE_TTL`.
+pub(crate) async fn runtime_tier_names() -> Arc<[String]> {
+    {
+        let cached = TIER_NAME_CACHE.read().unwrap_or_else(|err| err.into_inner()).clone();
+        if let Some((refreshed_at, names)) = cached
+            && refreshed_at.elapsed() < TIER_NAME_CACHE_TTL
+        {
+            return names;
+        }
+    }
+
+    let tiers = ecstore_get_global_tier_config_mgr().read().await.list_tiers();
+    let names: Arc<[String]> = tiers.iter().map(|tier| tier.name.clone()).collect::<Vec<_>>().into();
+    *TIER_NAME_CACHE.write().unwrap_or_else(|err| err.into_inner()) = Some((Instant::now(), Arc::clone(&names)));
+    names
+}
+
+/// Test-only cache reset; the production cache has no invalidation hook
+/// because the TTL is its only refresh path.
+#[cfg(test)]
+fn reset_tier_name_cache_for_test() {
+    *TIER_NAME_CACHE.write().unwrap_or_else(|err| err.into_inner()) = None;
 }
 
 pub(crate) async fn enqueue_runtime_free_version(oi: ScannerObjectInfo) {
@@ -561,6 +601,20 @@ mod tests {
     use super::*;
     use serial_test::serial;
 
+    #[tokio::test]
+    #[serial]
+    async fn runtime_tier_names_serves_cached_arc_within_ttl() {
+        reset_tier_name_cache_for_test();
+        // The tier config manager is unconfigured in unit tests, so the
+        // first call populates the cache from an empty tier list...
+        let first = runtime_tier_names().await;
+        assert!(first.is_empty());
+        // ...and a second call within the TTL must return the cached Arc
+        // (pointer-equal) without re-reading the manager.
+        let second = runtime_tier_names().await;
+        assert!(Arc::ptr_eq(&first, &second));
+    }
+
     #[test]
     #[serial]
     fn foreground_read_guard_tracks_stream_lifetime() {

crates/scanner/src/scanner_folder.rs

@@ -45,7 +45,7 @@ use rustfs_common::metrics::{
 use rustfs_common::trace_bus::{TraceEvent, TraceFunc, TraceKind, trace_emit, trace_subscriber_count};
 use rustfs_filemeta::{MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams};
 use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf};
-use s3s::dto::{BucketLifecycleConfiguration, ObjectLockConfiguration};
+use s3s::dto::{BucketLifecycleConfiguration, ObjectLockConfiguration, VersioningConfiguration};
 use time::OffsetDateTime;
 use tokio::select;
 use tokio::sync::mpsc;
@@ -53,10 +53,10 @@ use tokio_util::sync::CancellationToken;
 use tracing::{debug, error, warn};
 
 use crate::{
-    BucketVersioningSys, Disk, DiskError, DiskInfoOptions, Evaluator, Event, LcEventSrc, ListPathRawOptions, ObjectOpts,
-    ReplicationConfig, ReplicationHealObject, ReplicationQueueAdmission, ReplicationStatusType, STORAGE_FORMAT_FILE,
-    ScannerDiskExt as _, ScannerLifecycleConfigExt as _, ScannerVersioningConfigExt as _, StorageError, apply_expiry_rule,
-    apply_transition_rule, enqueue_runtime_newer_noncurrent, is_reserved_or_invalid_bucket, list_path_raw, path2_bucket_object,
+    Disk, DiskError, DiskInfoOptions, Evaluator, Event, LcEventSrc, ListPathRawOptions, ObjectOpts, ReplicationConfig,
+    ReplicationHealObject, ReplicationQueueAdmission, ReplicationStatusType, STORAGE_FORMAT_FILE, ScannerDiskExt as _,
+    ScannerLifecycleConfigExt as _, ScannerVersioningConfigExt as _, StorageError, apply_expiry_rule, apply_transition_rule,
+    enqueue_runtime_newer_noncurrent, is_reserved_or_invalid_bucket, list_path_raw, path2_bucket_object,
     path2_bucket_object_with_base_path, queue_replication_heal, scanner_is_erasure,
     scanner_replication_config_for_lifecycle_eval,
 };
@@ -934,6 +934,7 @@ impl ScannerItem {
         &mut self,
         object_infos: Vec<ObjectInfo>,
         lock_retention: Option<Arc<ObjectLockConfiguration>>,
+        versioning_config: VersioningConfiguration,
         size_summary: &mut SizeSummary,
     ) {
         if object_infos.is_empty() {
@@ -958,21 +959,8 @@ impl ScannerItem {
             "Scanner lifecycle evaluation started"
         );
 
-        let versioning_config = match BucketVersioningSys::get(&self.bucket).await {
-            Ok(versioning_config) => versioning_config,
-            Err(_) => {
-                warn!(
-                    target: "rustfs::scanner::folder",
-                    event = EVENT_SCANNER_LIFECYCLE_ACTION,
-                    component = LOG_COMPONENT_SCANNER,
-                    subsystem = LOG_SUBSYSTEM_LIFECYCLE,
-                    bucket = %self.bucket,
-                    state = "versioning_lookup_failed_defaulting",
-                    "Scanner lifecycle action falling back to default bucket versioning"
-                );
-                Default::default()
-            }
-        };
+        // `versioning_config` is resolved once per object by the caller
+        // (`get_size`) and handed in; only `prefix_enabled` is consulted here.
 
         let Some(lifecycle) = self.lifecycle.as_ref() else {
             let mut cumulative_size = 0;
@@ -1402,6 +1390,11 @@ impl ScannerItem {
     fn alert_excessive_versions(&self, remaining_versions: usize, cumulative_size: i64) {
         ensure_scanner_alert_metrics_registered();
         let (too_many_versions, too_large_versions) = should_alert_excessive_versions(remaining_versions, cumulative_size);
+        // Threshold check first so healthy objects never pay for the
+        // object-path allocation below.
+        if !too_many_versions && !too_large_versions {
+            return;
+        }
         let object_path = self.object_path();
         if too_many_versions {
             global_metrics().record_scanner_source_executed(ScannerWorkSource::Alerts, 1);

crates/scanner/src/scanner_io.rs

@@ -32,7 +32,9 @@ use rustfs_data_usage::{BucketTargetUsageInfo, BucketUsageInfo};
 use rustfs_filemeta::FileMeta;
 use rustfs_lock::{LockError, NamespaceLockGuard};
 use rustfs_utils::path::path_join_buf;
-use s3s::dto::{BucketLifecycleConfiguration, ObjectLockConfiguration, ObjectLockEnabled, ReplicationConfiguration};
+use s3s::dto::{
+    BucketLifecycleConfiguration, ObjectLockConfiguration, ObjectLockEnabled, ReplicationConfiguration, VersioningConfiguration,
+};
 use sha2::{Digest as _, Sha256};
 use std::collections::{HashMap, HashSet};
 use std::future::Future;
@@ -55,7 +57,7 @@ use crate::{
     BucketTargetSys, BucketVersioningSys, Disk, DiskError, ECStore, EcstoreError as Error, EcstoreResult as Result,
     RUSTFS_META_BUCKET, ReplicationConfig, STORAGE_FORMAT_FILE, ScannerDiskExt as _, ScannerLifecycleConfigExt as _,
     ScannerReplicationConfigExt as _, ScannerVersioningConfigExt as _, SetDisks, StorageError, enqueue_runtime_free_version,
-    get_lifecycle_config, get_object_lock_config, get_replication_config, list_runtime_tiers, storageclass,
+    get_lifecycle_config, get_object_lock_config, get_replication_config, runtime_tier_names, storageclass,
 };
 
 pub(crate) const SCANNER_SKIP_FILE_ERROR: &str = "skip file";
@@ -63,6 +65,11 @@ pub(crate) const SCANNER_METADATA_CORRUPT_ERROR: &str = "scanner metadata corrup
 pub(crate) const SCANNER_METADATA_TRANSIENT_ERROR: &str = "scanner metadata transient";
 const LOG_COMPONENT_SCANNER: &str = "scanner";
 const LOG_SUBSYSTEM_IO: &str = "io";
+// Mirrors `scanner_folder.rs` so the versioning-lookup fallback warn keeps its
+// historical `rustfs::scanner::folder` lifecycle event identity after the
+// lookup moved into `get_size`.
+const LOG_SUBSYSTEM_LIFECYCLE: &str = "lifecycle";
+const EVENT_SCANNER_LIFECYCLE_ACTION: &str = "scanner_lifecycle_action";
 const EVENT_SCANNER_DISK_BUCKET_STATE: &str = "scanner_disk_bucket_state";
 const EVENT_SCANNER_DATA_USAGE_STREAM: &str = "scanner_data_usage_stream";
 const EVENT_SCANNER_CACHE_PERSIST_STATE: &str = "scanner_cache_persist_state";
@@ -3822,6 +3829,24 @@ impl ScannerIOCache for SetDisks {
     }
 }
 
+/// Seed [`SizeSummary::tier_stats`] from the cached tier-name list.
+///
+/// Preserves the original seeding semantics: with no tiers configured the map
+/// stays completely empty (STANDARD/RRS are not seeded either); otherwise the
+/// standard storage classes are seeded alongside every configured tier so
+/// per-object accounting always finds its tier key.
+fn tier_stats_template(tier_names: &[String]) -> HashMap<String, TierStats> {
+    let mut tier_stats = HashMap::with_capacity(tier_names.len() + 2);
+    for tier_name in tier_names {
+        tier_stats.insert(tier_name.clone(), TierStats::default());
+    }
+    if !tier_stats.is_empty() {
+        tier_stats.insert(storageclass::STANDARD.to_string(), TierStats::default());
+        tier_stats.insert(storageclass::RRS.to_string(), TierStats::default());
+    }
+    tier_stats
+}
+
 #[async_trait::async_trait]
 impl ScannerIODisk for Disk {
     async fn get_size(&self, mut item: ScannerItem) -> Result<SizeSummary> {
@@ -3861,10 +3886,26 @@ impl ScannerIODisk for Disk {
             }
         };
 
-        let versioned = BucketVersioningSys::get(&item.bucket)
-            .await
-            .map(|v| v.versioned(&item.object_path()))
-            .unwrap_or(false);
+        // Single versioning lookup per object, shared with `apply_actions`
+        // (which used to query it a second time). On failure keep the
+        // historical fallback: default configuration (versioned = false) plus
+        // the warn that `apply_actions` used to emit.
+        let versioning_config = match BucketVersioningSys::get(&item.bucket).await {
+            Ok(versioning_config) => versioning_config,
+            Err(_) => {
+                warn!(
+                    target: "rustfs::scanner::folder",
+                    event = EVENT_SCANNER_LIFECYCLE_ACTION,
+                    component = LOG_COMPONENT_SCANNER,
+                    subsystem = LOG_SUBSYSTEM_LIFECYCLE,
+                    bucket = %item.bucket,
+                    state = "versioning_lookup_failed_defaulting",
+                    "Scanner lifecycle action falling back to default bucket versioning"
+                );
+                VersioningConfiguration::default()
+            }
+        };
+        let versioned = versioning_config.versioned(&item.object_path());
 
         let object_infos = fivs
             .versions
@@ -3879,19 +3920,10 @@ impl ScannerIODisk for Disk {
 
         let mut size_summary = SizeSummary::default();
 
-        let tiers = list_runtime_tiers().await;
-
-        for tier in tiers.iter() {
-            size_summary.tier_stats.insert(tier.name.clone(), TierStats::default());
-        }
-        if !size_summary.tier_stats.is_empty() {
-            size_summary
-                .tier_stats
-                .insert(storageclass::STANDARD.to_string(), TierStats::default());
-            size_summary
-                .tier_stats
-                .insert(storageclass::RRS.to_string(), TierStats::default());
-        }
+        // Tier names come from the process-wide TTL cache; seeding from them
+        // replaces the per-object clone of every full TierConfig.
+        let tier_names = runtime_tier_names().await;
+        size_summary.tier_stats = tier_stats_template(&tier_names);
 
         let lock_config = object_lock_config_for_scanner_item(&item).await;
 
@@ -3901,7 +3933,8 @@ impl ScannerIODisk for Disk {
         // `object_infos`.
         global_metrics().record_scanner_versions_scanned(object_infos.len() as u64);
 
-        item.apply_actions(object_infos, lock_config, &mut size_summary).await;
+        item.apply_actions(object_infos, lock_config, versioning_config, &mut size_summary)
+            .await;
 
         if !free_version_infos.is_empty() {
             for oi in free_version_infos {
@@ -4968,6 +5001,23 @@ mod tests {
         assert!(is_xl_meta_path("/data/bucket/object/xl.meta"));
     }
 
+    #[test]
+    fn tier_stats_template_seeds_tiers_and_standard_classes() {
+        let template = tier_stats_template(&["WARM".to_string(), "COLD".to_string()]);
+
+        assert_eq!(template.len(), 4);
+        for tier in ["WARM", "COLD", storageclass::STANDARD, storageclass::RRS] {
+            assert_eq!(template.get(tier), Some(&TierStats::default()), "missing seed for tier {tier}");
+        }
+    }
+
+    #[test]
+    fn tier_stats_template_stays_empty_without_tiers() {
+        let template = tier_stats_template(&[]);
+
+        assert!(template.is_empty());
+    }
+
     #[tokio::test]
     async fn get_size_treats_missing_metadata_as_skip_file() {
         let temp_dir = std::env::temp_dir().join(format!("rustfs-scanner-missing-meta-{}", Uuid::new_v4()));

crates/scanner/src/storage_api.rs

@@ -99,7 +99,6 @@ pub(crate) use rustfs_ecstore::api::set_disk::SetDisks as EcstoreSetDisks;
 pub(crate) use rustfs_ecstore::api::storage::ECStore as EcstoreStore;
 #[cfg(test)]
 pub(crate) use rustfs_ecstore::api::storage::init_local_disks_with_instance_ctx as ecstore_init_local_disks_with_instance_ctx;
-pub(crate) use rustfs_ecstore::api::tier::tier_config::TierConfig as EcstoreTierConfig;
 use rustfs_storage_api as storage_contracts;
 
 pub(crate) mod owner {
@@ -114,15 +113,15 @@ pub(crate) mod owner {
         EcstoreDiskLocation, EcstoreDiskResult, EcstoreErrorType, EcstoreEvaluator, EcstoreEvent, EcstoreEventArgs,
         EcstoreLcEventSrc, EcstoreLifecycle, EcstoreListPathRawOptions, EcstoreNsScannerOpenRequest, EcstoreObjectOpts,
         EcstoreReplicationConfigurationExt, EcstoreReplicationScannerBridge, EcstoreResultType, EcstoreScanGuard,
-        EcstoreSetDisks, EcstoreStorageError, EcstoreStore, EcstoreTierConfig, EcstoreVersioningApi,
-        ScannerReplicationHealObject, ScannerReplicationHealResult, ScannerReplicationQueueAdmission, ecstore_apply_expiry_rule,
-        ecstore_apply_transition_rule, ecstore_expiry_state_handle, ecstore_get_global_tier_config_mgr,
-        ecstore_get_lifecycle_config, ecstore_get_object_lock_config, ecstore_get_replication_config,
-        ecstore_invalidate_admin_data_usage_snapshot_cache, ecstore_invalidate_data_usage_snapshot_cache, ecstore_is_erasure,
-        ecstore_is_erasure_sd, ecstore_is_reserved_or_invalid_bucket, ecstore_list_path_raw,
-        ecstore_object_opts_from_object_info, ecstore_path2_bucket_object, ecstore_path2_bucket_object_with_base_path,
-        ecstore_read_config, ecstore_replace_bucket_usage_memory_from_info, ecstore_resolve_object_store_handle,
-        ecstore_save_config, ecstore_send_event, scanner_replication_config_for_lifecycle_eval,
+        EcstoreSetDisks, EcstoreStorageError, EcstoreStore, EcstoreVersioningApi, ScannerReplicationHealObject,
+        ScannerReplicationHealResult, ScannerReplicationQueueAdmission, ecstore_apply_expiry_rule, ecstore_apply_transition_rule,
+        ecstore_expiry_state_handle, ecstore_get_global_tier_config_mgr, ecstore_get_lifecycle_config,
+        ecstore_get_object_lock_config, ecstore_get_replication_config, ecstore_invalidate_admin_data_usage_snapshot_cache,
+        ecstore_invalidate_data_usage_snapshot_cache, ecstore_is_erasure, ecstore_is_erasure_sd,
+        ecstore_is_reserved_or_invalid_bucket, ecstore_list_path_raw, ecstore_object_opts_from_object_info,
+        ecstore_path2_bucket_object, ecstore_path2_bucket_object_with_base_path, ecstore_read_config,
+        ecstore_replace_bucket_usage_memory_from_info, ecstore_resolve_object_store_handle, ecstore_save_config,
+        ecstore_send_event, scanner_replication_config_for_lifecycle_eval,
     };
 
     #[cfg(test)]

rustfs/src/admin/handlers/audit.rs

@@ -13,7 +13,7 @@
 // limitations under the License.
 
 use crate::admin::{
-    auth::validate_admin_request,
+    auth::authorize_admin_request,
     handlers::audit_runtime_config::{load_server_config_from_store, update_audit_config_and_reload},
     handlers::target_descriptor::{
         AdminTargetSpec, EndpointKey, RuntimeHealthStatus, TargetEndpointSource, admin_target_spec_from_builtin,
@@ -23,9 +23,8 @@ use crate::admin::{
     },
     router::{AdminOperation, Operation, S3Router},
 };
-use crate::auth::{check_key_valid, get_session_token};
 use crate::server::{
-    ADMIN_PREFIX, RemoteAddr, is_audit_module_enabled, refresh_audit_module_enabled, refresh_persisted_module_switches_from_store,
+    ADMIN_PREFIX, is_audit_module_enabled, refresh_audit_module_enabled, refresh_persisted_module_switches_from_store,
 };
 use http::StatusCode;
 use hyper::Method;
@@ -213,14 +212,14 @@ fn audit_target_specs() -> &'static [AdminTargetSpec] {
     &AUDIT_TARGET_SPECS
 }
 
+/// The pre-check keeps these endpoints' historical missing-credentials message;
+/// the shared gate reports "get cred failed".
 async fn authorize_audit_admin_request(req: &S3Request<Body>, action: AdminAction) -> S3Result<()> {
-    let Some(input_cred) = &req.credentials else {
+    if req.credentials.is_none() {
         return Err(s3_error!(InvalidRequest, "credentials not found"));
-    };
-    let (cred, owner) =
-        check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
-    let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
-    validate_admin_request(&req.headers, &cred, owner, false, vec![Action::AdminAction(action)], remote_addr).await
+    }
+    authorize_admin_request(req, vec![Action::AdminAction(action)]).await?;
+    Ok(())
 }
 
 fn audit_target_mutation_block_reason(config: &Config, target_type: &str, target_name: &str) -> S3Result<Option<String>> {
@@ -824,6 +823,30 @@ mod tests {
         });
     }
 
+    /// These endpoints authorize through the shared admin gate, which reports
+    /// "get cred failed" for a credential-less request. The pre-check keeps the
+    /// message they have always returned (rustfs/backlog#1829).
+    #[tokio::test]
+    async fn audit_target_gate_keeps_its_missing_credentials_message() {
+        let req = S3Request {
+            input: Body::from(String::new()),
+            method: Method::PUT,
+            uri: http::Uri::from_static("/rustfs/admin/v3/audit/target"),
+            headers: http::HeaderMap::new(),
+            extensions: http::Extensions::new(),
+            credentials: None,
+            region: None,
+            service: None,
+            trailing_headers: None,
+        };
+
+        let err = authorize_audit_admin_request(&req, AdminAction::SetBucketTargetAction)
+            .await
+            .expect_err("a request without credentials must be rejected");
+        assert_eq!(err.code(), &s3s::S3ErrorCode::InvalidRequest);
+        assert_eq!(err.message(), Some("credentials not found"));
+    }
+
     #[test]
     fn audit_target_handlers_require_admin_authorization_contract() {
         let src = include_str!("audit.rs");

rustfs/src/admin/handlers/diagnostics.rs

@@ -23,11 +23,10 @@
 //! backing infrastructure (in-process log ring buffer, cross-node object
 //! speedtest harness).
 
-use crate::admin::auth::validate_admin_request;
+use crate::admin::auth::authorize_admin_request;
 use crate::admin::router::{AdminOperation, Operation, S3Router};
 use crate::admin::storage_api::access::spawn_traced;
-use crate::auth::{check_key_valid, get_session_token};
-use crate::server::{ADMIN_PREFIX, RemoteAddr};
+use crate::server::ADMIN_PREFIX;
 use crate::storage::storage_api::get_global_lock_clients;
 use bytes::Bytes;
 use futures::{Stream, StreamExt, future::join_all};
@@ -133,16 +132,15 @@ pub fn register_diagnostics_route(r: &mut S3Router<AdminOperation>) -> std::io::
 // Shared auth helper
 // ---------------------------------------------------------------------------
 
+/// The pre-check keeps these endpoints' historical `AccessDenied` missing-credentials
+/// response; the shared gate reports `InvalidRequest` "get cred failed".
 async fn authorize(req: &S3Request<Body>, action: AdminAction) -> S3Result<()> {
-    let Some(input_cred) = req.credentials.as_ref() else {
+    if req.credentials.is_none() {
         return Err(s3_error!(AccessDenied, "Signature is required"));
-    };
+    }
 
-    let (cred, owner) =
-        check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
-    let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
-
-    validate_admin_request(&req.headers, &cred, owner, false, vec![Action::AdminAction(action)], remote_addr).await
+    authorize_admin_request(req, vec![Action::AdminAction(action)]).await?;
+    Ok(())
 }
 
 fn json_response<T: Serialize>(status: StatusCode, value: &T) -> S3Result<S3Response<(StatusCode, Body)>> {
@@ -1078,6 +1076,22 @@ mod tests {
         }
     }
 
+    /// These endpoints authorize through the shared admin gate, which rejects a
+    /// credential-less request with `InvalidRequest` "get cred failed". The
+    /// pre-check keeps the `AccessDenied` response they have always returned
+    /// (rustfs/backlog#1829).
+    #[tokio::test]
+    async fn diagnostics_gate_keeps_its_missing_credentials_response() {
+        let err = authorize(
+            &build_request(Method::GET, "/rustfs/admin/v3/top/locks"),
+            AdminAction::ServerInfoAdminAction,
+        )
+        .await
+        .expect_err("a request without credentials must be rejected");
+        assert_eq!(err.code(), &S3ErrorCode::AccessDenied);
+        assert_eq!(err.message(), Some("Signature is required"));
+    }
+
     #[tokio::test]
     async fn top_locks_handler_rejects_missing_credentials() {
         let err = TopLocksHandler {}

rustfs/src/admin/handlers/durability.rs

@@ -28,6 +28,7 @@ use crate::admin::storage_api::bucket::metadata::BUCKET_DURABILITY_CONFIG;
 use crate::admin::storage_api::bucket::metadata_sys;
 use crate::auth::{check_key_valid, get_session_token};
 use crate::server::ADMIN_PREFIX;
+use crate::server::RemoteAddr;
 use hyper::{Method, StatusCode};
 use matchit::Params;
 use rustfs_policy::policy::action::{Action, AdminAction};
@@ -126,13 +127,14 @@ async fn authenticate_admin(req: &S3Request<Body>) -> S3Result<()> {
 
     let (cred, owner) = check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &cred.access_key).await?;
 
+    let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
     validate_admin_request(
         &req.headers,
         &cred,
         owner,
         false,
         vec![Action::AdminAction(AdminAction::ConfigUpdateAdminAction)],
-        None,
+        remote_addr,
     )
     .await?;
 

rustfs/src/admin/handlers/event.rs

@@ -13,7 +13,7 @@
 // limitations under the License.
 
 use crate::admin::{
-    auth::validate_admin_request,
+    auth::authorize_admin_request,
     handlers::notify_runtime_access::{get_notification_system, load_notification_config_snapshot},
     handlers::supervise_admin_mutation,
     handlers::target_descriptor::{
@@ -26,10 +26,8 @@ use crate::admin::{
     runtime_sources::{AppContext, app_context_from_req},
     service::config::{preflight_dynamic_config_reload_for_context, signal_dynamic_config_reload_checked_for_context},
 };
-use crate::auth::{check_key_valid, get_session_token};
 use crate::server::{
-    ADMIN_PREFIX, RemoteAddr, is_notify_module_enabled, refresh_notify_module_enabled,
-    refresh_persisted_module_switches_from_store,
+    ADMIN_PREFIX, is_notify_module_enabled, refresh_notify_module_enabled, refresh_persisted_module_switches_from_store,
 };
 use http::StatusCode;
 use hyper::Method;
@@ -264,14 +262,14 @@ fn notification_target_specs() -> &'static [AdminTargetSpec] {
 
 // --- Helper Functions ---
 
+/// The pre-check keeps these endpoints' historical missing-credentials message;
+/// the shared gate reports "get cred failed".
 async fn authorize_notification_admin_request(req: &S3Request<Body>, action: AdminAction) -> S3Result<()> {
-    let Some(input_cred) = &req.credentials else {
+    if req.credentials.is_none() {
         return Err(s3_error!(InvalidRequest, "credentials not found"));
-    };
-    let (cred, owner) =
-        check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
-    let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
-    validate_admin_request(&req.headers, &cred, owner, false, vec![Action::AdminAction(action)], remote_addr).await
+    }
+    authorize_admin_request(req, vec![Action::AdminAction(action)]).await?;
+    Ok(())
 }
 
 fn target_mutation_block_reason(config: &Config, target_type: &str, target_name: &str) -> S3Result<Option<String>> {
@@ -987,6 +985,30 @@ mod tests {
         );
     }
 
+    /// These endpoints authorize through the shared admin gate, which reports
+    /// "get cred failed" for a credential-less request. The pre-check keeps the
+    /// message they have always returned (rustfs/backlog#1829).
+    #[tokio::test]
+    async fn notification_target_gate_keeps_its_missing_credentials_message() {
+        let req = S3Request {
+            input: Body::from(String::new()),
+            method: Method::PUT,
+            uri: http::Uri::from_static("/rustfs/admin/v3/notification/target"),
+            headers: http::HeaderMap::new(),
+            extensions: http::Extensions::new(),
+            credentials: None,
+            region: None,
+            service: None,
+            trailing_headers: None,
+        };
+
+        let err = authorize_notification_admin_request(&req, AdminAction::SetBucketTargetAction)
+            .await
+            .expect_err("a request without credentials must be rejected");
+        assert_eq!(err.code(), &s3s::S3ErrorCode::InvalidRequest);
+        assert_eq!(err.message(), Some("credentials not found"));
+    }
+
     #[test]
     fn notification_target_handlers_require_admin_authorization_contract() {
         let src = include_str!("event.rs");

rustfs/src/admin/handlers/metrics.rs

@@ -18,12 +18,10 @@
 //! keeping the response format explicitly NDJSON. It is not a Prometheus text
 //! exposition endpoint.
 
-use crate::admin::auth::validate_admin_request;
+use crate::admin::auth::authorize_admin_request;
 use crate::admin::router::Operation;
 use crate::admin::storage_api::access::spawn_traced;
 use crate::admin::storage_api::metrics::{CollectMetricsOpts, MetricType, collect_local_metrics};
-use crate::auth::{check_key_valid, get_session_token};
-use crate::server::RemoteAddr;
 use bytes::Bytes;
 use futures::{Stream, StreamExt};
 use http::{HeaderMap, HeaderValue, Uri};
@@ -182,24 +180,15 @@ impl ByteStream for MetricsStream {}
 
 pub struct MetricsHandler {}
 
+/// The pre-check keeps this endpoint's historical `AccessDenied` missing-credentials
+/// response; the shared gate reports `InvalidRequest` "get cred failed".
 async fn authorize_metrics_request(req: &S3Request<Body>) -> S3Result<()> {
-    let Some(input_cred) = req.credentials.as_ref() else {
+    if req.credentials.is_none() {
         return Err(s3_error!(AccessDenied, "Signature is required"));
-    };
+    }
 
-    let (cred, owner) =
-        check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
-    let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
-
-    validate_admin_request(
-        &req.headers,
-        &cred,
-        owner,
-        false,
-        vec![Action::AdminAction(AdminAction::GetMetricsAction)],
-        remote_addr,
-    )
-    .await
+    authorize_admin_request(req, vec![Action::AdminAction(AdminAction::GetMetricsAction)]).await?;
+    Ok(())
 }
 
 #[async_trait::async_trait]

rustfs/src/admin/handlers/module_switch.rs

@@ -17,14 +17,13 @@ use crate::admin::service::config::{
     preflight_dynamic_config_reload_for_context, signal_dynamic_config_reload_checked_for_context,
 };
 use crate::admin::{
-    auth::validate_admin_request,
+    auth::authorize_admin_request,
     handlers::supervise_admin_mutation,
     router::{AdminOperation, Operation, S3Router},
 };
-use crate::auth::{check_key_valid, get_session_token};
 use crate::server::{
     ADMIN_PREFIX, MODULE_SWITCHES_SIGNAL_SUBSYSTEM, ModuleSwitchSnapshot, ModuleSwitchSource, PersistedModuleSwitches,
-    RemoteAddr, apply_audit_module_switch_for_context, current_module_switch_snapshot, mark_event_notifier_reconciled,
+    apply_audit_module_switch_for_context, current_module_switch_snapshot, mark_event_notifier_reconciled,
     mark_event_notifier_unreconciled, refresh_audit_module_enabled, refresh_notify_module_enabled,
     refresh_persisted_module_switches_from, refresh_persisted_module_switches_from_store, save_persisted_module_switches_to,
     validate_module_switch_update,
@@ -114,23 +113,15 @@ fn build_response<T: Serialize>(
     Ok(S3Response::with_headers((status, Body::from(data)), header))
 }
 
+/// The pre-check keeps these endpoints' historical missing-credentials message;
+/// the shared gate reports "get cred failed".
 async fn authorize_module_switch_request(req: &S3Request<Body>, action: AdminAction) -> S3Result<()> {
-    let Some(input_cred) = &req.credentials else {
+    if req.credentials.is_none() {
         return Err(s3_error!(InvalidRequest, "authentication required"));
-    };
+    }
 
-    let (cred, owner) =
-        check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
-
-    validate_admin_request(
-        &req.headers,
-        &cred,
-        owner,
-        false,
-        vec![Action::AdminAction(action)],
-        req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
-    )
-    .await
+    authorize_admin_request(req, vec![Action::AdminAction(action)]).await?;
+    Ok(())
 }
 
 async fn refresh_module_switch_snapshot() -> S3Result<ModuleSwitchSnapshot> {
@@ -269,6 +260,30 @@ impl Operation for UpdateModuleSwitchesHandler {
 mod tests {
     use super::{ModuleSwitchDiscovery, ModuleSwitchSource, ModuleSwitchesResponse};
 
+    /// These endpoints authorize through the shared admin gate, which reports
+    /// "get cred failed" for a credential-less request. The pre-check keeps the
+    /// message they have always returned (rustfs/backlog#1829).
+    #[tokio::test]
+    async fn module_switch_gate_keeps_its_missing_credentials_message() {
+        let req = s3s::S3Request {
+            input: s3s::Body::from(String::new()),
+            method: http::Method::GET,
+            uri: http::Uri::from_static("/rustfs/admin/v3/module-switches"),
+            headers: http::HeaderMap::new(),
+            extensions: http::Extensions::new(),
+            credentials: None,
+            region: None,
+            service: None,
+            trailing_headers: None,
+        };
+
+        let err = super::authorize_module_switch_request(&req, rustfs_policy::policy::action::AdminAction::ServerInfoAdminAction)
+            .await
+            .expect_err("a request without credentials must be rejected");
+        assert_eq!(err.code(), &s3s::S3ErrorCode::InvalidRequest);
+        assert_eq!(err.message(), Some("authentication required"));
+    }
+
     #[test]
     fn module_switch_handlers_require_admin_authorization_contract() {
         let src = include_str!("module_switch.rs");

rustfs/src/admin/handlers/profile.rs

@@ -12,33 +12,22 @@
 //  See the License for the specific language governing permissions and
 //  limitations under the License.
 
-use crate::admin::{auth::validate_admin_request, router::Operation};
-use crate::auth::{check_key_valid, get_session_token};
-use crate::server::RemoteAddr;
+use crate::admin::{auth::authorize_admin_request, router::Operation};
 use http::StatusCode;
 use matchit::Params;
 use rustfs_policy::policy::action::{Action, AdminAction};
 use s3s::{Body, S3Request, S3Response, S3Result, s3_error};
 use tracing::info;
 
+/// The pre-check keeps these endpoints' historical `AccessDenied` missing-credentials
+/// response; the shared gate reports `InvalidRequest` "get cred failed".
 pub(super) async fn authorize_profile_request(req: &S3Request<Body>) -> S3Result<()> {
-    let Some(input_cred) = req.credentials.as_ref() else {
+    if req.credentials.is_none() {
         return Err(s3_error!(AccessDenied, "Signature is required"));
-    };
+    }
 
-    let (cred, owner) =
-        check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
-    let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
-
-    validate_admin_request(
-        &req.headers,
-        &cred,
-        owner,
-        false,
-        vec![Action::AdminAction(AdminAction::ProfilingAdminAction)],
-        remote_addr,
-    )
-    .await
+    authorize_admin_request(req, vec![Action::AdminAction(AdminAction::ProfilingAdminAction)]).await?;
+    Ok(())
 }
 
 pub(super) fn profile_not_implemented_response(message: String) -> S3Response<(StatusCode, Body)> {

rustfs/src/admin/handlers/profile_admin.rs

@@ -13,11 +13,10 @@
 // limitations under the License.
 
 use super::profile::{authorize_profile_request, profile_not_implemented_response};
-use crate::admin::auth::validate_admin_request;
+use crate::admin::auth::authorize_admin_request;
 use crate::admin::router::{AdminOperation, Operation, S3Router};
 use crate::admin::storage_api::access::spawn_traced;
-use crate::auth::{check_key_valid, get_session_token};
-use crate::server::{ADMIN_PREFIX, RemoteAddr};
+use crate::server::ADMIN_PREFIX;
 use bytes::Bytes;
 use futures::{Stream, StreamExt};
 use http::{HeaderMap, HeaderValue};
@@ -89,14 +88,14 @@ pub fn register_profiling_route(r: &mut S3Router<AdminOperation>) -> std::io::Re
 }
 
 /// Authorize a request against a single admin action (profiling or trace).
+/// The pre-check keeps these endpoints' historical `AccessDenied` missing-credentials
+/// response; the shared gate reports `InvalidRequest` "get cred failed".
 async fn authorize_action(req: &S3Request<Body>, action: AdminAction) -> S3Result<()> {
-    let Some(input_cred) = req.credentials.as_ref() else {
+    if req.credentials.is_none() {
         return Err(s3_error!(AccessDenied, "Signature is required"));
-    };
-    let (cred, owner) =
-        check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
-    let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
-    validate_admin_request(&req.headers, &cred, owner, false, vec![Action::AdminAction(action)], remote_addr).await
+    }
+    authorize_admin_request(req, vec![Action::AdminAction(action)]).await?;
+    Ok(())
 }
 
 pub struct ProfileHandler {}
@@ -530,7 +529,7 @@ fn trace_value_string(value: &TraceVal) -> String {
 mod tests {
     use super::{
         ProfileControlHandler, ProfileHandler, ProfileStatusHandler, ProfilingDownloadHandler, ProfilingStartHandler,
-        TraceHandler, TraceKindFilter, TraceStreamFilter, TraceWireRecord,
+        TraceHandler, TraceKindFilter, TraceStreamFilter, TraceWireRecord, authorize_action,
     };
     use crate::admin::router::Operation;
     use http::{Extensions, HeaderMap, Uri};
@@ -539,6 +538,7 @@ mod tests {
     use rustfs_common::trace_bus::{TraceEvent, TraceFunc, TraceKind};
     use rustfs_madmin::service_commands::ServiceTraceOpts;
     use rustfs_madmin::trace::TraceType;
+    use rustfs_policy::policy::action::AdminAction;
     use s3s::{Body, S3ErrorCode, S3Request, S3Result};
     use std::time::{Duration, UNIX_EPOCH};
 
@@ -563,6 +563,22 @@ mod tests {
         TraceStreamFilter::from_request(&uri, &opts)
     }
 
+    /// The profiling/trace endpoints authorize through the shared admin gate, which
+    /// rejects a credential-less request with `InvalidRequest` "get cred failed". The
+    /// pre-check keeps the `AccessDenied` response they have always returned
+    /// (rustfs/backlog#1829).
+    #[tokio::test]
+    async fn profile_admin_gate_keeps_its_missing_credentials_response() {
+        let err = authorize_action(
+            &build_profile_request("/rustfs/admin/v3/profiling/start"),
+            AdminAction::ProfilingAdminAction,
+        )
+        .await
+        .expect_err("a request without credentials must be rejected");
+        assert_eq!(err.code(), &S3ErrorCode::AccessDenied);
+        assert_eq!(err.message(), Some("Signature is required"));
+    }
+
     #[tokio::test]
     async fn profile_handler_rejects_missing_credentials() {
         let result = ProfileHandler {}

rustfs/src/admin/handlers/quota.rs

@@ -24,6 +24,7 @@ use crate::admin::storage_api::bucket::quota::{BucketQuota, QuotaError, QuotaOpe
 use crate::auth::{check_key_valid, get_session_token};
 use crate::error::ApiError;
 use crate::server::ADMIN_PREFIX;
+use crate::server::RemoteAddr;
 use hyper::{Method, StatusCode};
 use matchit::Params;
 use rustfs_madmin::{SITE_REPL_API_VERSION, SRBucketMeta};
@@ -264,13 +265,14 @@ impl Operation for SetBucketQuotaHandler {
         let (cred, owner) =
             check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &cred.access_key).await?;
 
+        let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
         validate_admin_request(
             &req.headers,
             &cred,
             owner,
             false,
             vec![Action::AdminAction(AdminAction::SetBucketQuotaAdminAction)],
-            None,
+            remote_addr,
         )
         .await?;
 
@@ -393,13 +395,14 @@ impl Operation for GetBucketQuotaHandler {
         if bucket.is_empty() {
             return Err(s3_error!(InvalidRequest, "bucket name is required"));
         }
+        let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
         validate_admin_request_with_bucket(
             &req.headers,
             &cred,
             owner,
             false,
             vec![Action::S3Action(S3Action::GetBucketQuotaAction)],
-            None,
+            remote_addr,
             &bucket,
         )
         .await?;
@@ -461,13 +464,14 @@ impl Operation for ClearBucketQuotaHandler {
         let (cred, owner) =
             check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &cred.access_key).await?;
 
+        let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
         validate_admin_request(
             &req.headers,
             &cred,
             owner,
             false,
             vec![Action::AdminAction(AdminAction::SetBucketQuotaAdminAction)],
-            None,
+            remote_addr,
         )
         .await?;
 
@@ -577,13 +581,14 @@ impl Operation for GetBucketQuotaStatsHandler {
             return Err(s3_error!(InvalidRequest, "bucket name is required"));
         }
 
+        let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
         validate_admin_request_with_bucket(
             &req.headers,
             &cred,
             owner,
             false,
             vec![Action::S3Action(S3Action::GetBucketQuotaAction)],
-            None,
+            remote_addr,
             &bucket,
         )
         .await?;
@@ -649,13 +654,14 @@ impl Operation for CheckBucketQuotaHandler {
             return Err(s3_error!(InvalidRequest, "bucket name is required"));
         }
 
+        let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
         validate_admin_request_with_bucket(
             &req.headers,
             &cred,
             owner,
             false,
             vec![Action::S3Action(S3Action::GetBucketQuotaAction)],
-            None,
+            remote_addr,
             &bucket,
         )
         .await?;

rustfs/src/admin/handlers/scanner.rs

@@ -12,12 +12,11 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::admin::auth::validate_admin_request;
+use crate::admin::auth::authorize_admin_request;
 use crate::admin::router::{AdminOperation, Operation, S3Router};
 use crate::admin::runtime_sources::current_scanner_metrics_report;
-use crate::auth::{check_key_valid, get_session_token};
 use crate::module_switches::{ENV_SCANNER_ENABLED, scanner_enabled_from_env};
-use crate::server::{ADMIN_PREFIX, RemoteAddr};
+use crate::server::ADMIN_PREFIX;
 use chrono::Utc;
 use http::{HeaderMap, HeaderValue};
 use hyper::{Method, StatusCode};
@@ -154,29 +153,14 @@ pub fn register_scanner_route(r: &mut S3Router<AdminOperation>) -> std::io::Resu
     Ok(())
 }
 
+/// The pre-check keeps these endpoints' historical missing-credentials message;
+/// the shared gate reports "get cred failed".
 async fn validate_scanner_status_request(req: &S3Request<Body>) -> S3Result<Credentials> {
-    let Some(input_cred) = req.credentials.as_ref() else {
+    if req.credentials.is_none() {
         return Err(s3_error!(InvalidRequest, "missing credentials"));
-    };
+    }
 
-    let (cred, owner) =
-        check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
-
-    let remote_addr = req
-        .extensions
-        .get::<Option<RemoteAddr>>()
-        .and_then(|opt| opt.map(|addr| addr.0));
-    validate_admin_request(
-        &req.headers,
-        &cred,
-        owner,
-        false,
-        vec![Action::AdminAction(AdminAction::ServerInfoAdminAction)],
-        remote_addr,
-    )
-    .await?;
-
-    Ok(cred)
+    authorize_admin_request(req, vec![Action::AdminAction(AdminAction::ServerInfoAdminAction)]).await
 }
 
 fn json_response(body: Vec<u8>) -> S3Result<S3Response<(StatusCode, Body)>> {
@@ -229,6 +213,30 @@ impl Operation for IlmExpiryStatusHandler {
 mod tests {
     use super::*;
 
+    /// These endpoints authorize through the shared admin gate, which reports
+    /// "get cred failed" for a credential-less request. The pre-check keeps the
+    /// message they have always returned (rustfs/backlog#1829).
+    #[tokio::test]
+    async fn scanner_status_gate_keeps_its_missing_credentials_message() {
+        let req = S3Request {
+            input: Body::from(String::new()),
+            method: Method::GET,
+            uri: http::Uri::from_static("/rustfs/admin/v3/scanner/status"),
+            headers: HeaderMap::new(),
+            extensions: http::Extensions::new(),
+            credentials: None,
+            region: None,
+            service: None,
+            trailing_headers: None,
+        };
+
+        let err = validate_scanner_status_request(&req)
+            .await
+            .expect_err("a request without credentials must be rejected");
+        assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
+        assert_eq!(err.message(), Some("missing credentials"));
+    }
+
     #[test]
     fn scanner_disabled_reason_reports_startup_env_key() {
         assert_eq!(scanner_disabled_reason(true), None);

rustfs/src/admin/handlers/usage_prefix.rs

@@ -19,11 +19,10 @@
 //! usage caches, with a one-level sub-prefix breakdown — the data console
 //! buckets view MinIO serves from `loadPrefixUsageFromBackend`.
 
-use crate::admin::auth::validate_admin_request;
+use crate::admin::auth::authorize_admin_request;
 use crate::admin::handlers::system::data_usage_info_gate_actions;
 use crate::admin::router::{AdminOperation, Operation, S3Router};
-use crate::auth::{check_key_valid, get_session_token};
-use crate::server::{ADMIN_PREFIX, RemoteAddr};
+use crate::server::ADMIN_PREFIX;
 use http::{HeaderMap, HeaderValue, StatusCode};
 use hyper::Method;
 use matchit::Params;
@@ -70,15 +69,10 @@ fn parse_usage_prefix_query(query: Option<&str>) -> S3Result<(String, usize)> {
 #[async_trait::async_trait]
 impl Operation for BucketPrefixUsageHandler {
     async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
-        let Some(input_cred) = req.credentials else {
-            return Err(s3_error!(InvalidRequest, "get cred failed"));
-        };
-
-        let (cred, owner) =
-            check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
-
-        let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
-        validate_admin_request(&req.headers, &cred, owner, false, data_usage_info_gate_actions(), remote_addr).await?;
+        // The shared gate reports the same `InvalidRequest` "get cred failed" this
+        // handler has always returned for a credential-less request, so it needs no
+        // message-preserving pre-check.
+        authorize_admin_request(&req, data_usage_info_gate_actions()).await?;
 
         let bucket = params.get("bucket").unwrap_or_default().to_string();
         if bucket.is_empty() {
@@ -104,13 +98,40 @@ impl Operation for BucketPrefixUsageHandler {
 
 #[cfg(test)]
 mod tests {
-    use super::{DEFAULT_MAX_ENTRIES, MAX_ENTRIES_LIMIT, parse_usage_prefix_query};
+    use super::{BucketPrefixUsageHandler, DEFAULT_MAX_ENTRIES, MAX_ENTRIES_LIMIT, parse_usage_prefix_query};
+    use crate::admin::router::Operation;
     use s3s::S3Error;
 
     fn query(raw: &str) -> Result<(String, usize), S3Error> {
         parse_usage_prefix_query(Some(raw))
     }
 
+    /// This endpoint authorizes through the shared admin gate, whose
+    /// credential-less rejection is the same `InvalidRequest` "get cred failed"
+    /// the handler returned inline before (rustfs/backlog#1829), so no
+    /// message-preserving pre-check is needed here.
+    #[tokio::test]
+    async fn prefix_usage_handler_keeps_its_missing_credentials_message() {
+        let req = s3s::S3Request {
+            input: s3s::Body::from(String::new()),
+            method: http::Method::GET,
+            uri: http::Uri::from_static("/rustfs/admin/v3/usage/bucket"),
+            headers: http::HeaderMap::new(),
+            extensions: http::Extensions::new(),
+            credentials: None,
+            region: None,
+            service: None,
+            trailing_headers: None,
+        };
+
+        let err = BucketPrefixUsageHandler {}
+            .call(req, matchit::Params::new())
+            .await
+            .expect_err("a request without credentials must be rejected");
+        assert_eq!(err.code(), &s3s::S3ErrorCode::InvalidRequest);
+        assert_eq!(err.message(), Some("get cred failed"));
+    }
+
     #[test]
     fn defaults_apply_when_no_query_is_given() {
         assert_eq!(parse_usage_prefix_query(None).unwrap(), (String::new(), DEFAULT_MAX_ENTRIES));

rustfs/src/app/object_usecase.rs

@@ -1056,9 +1056,22 @@ pin_project! {
         remaining: usize,
         emitted: usize,
         expected: usize,
+        // Diagnostic-only identity for the body this stream is serving. Unset in
+        // unit tests that drive the stream over a bare reader; every production
+        // body carries it via `with_diagnostics`.
+        diagnostics: GetObjectReaderStreamDiagnostics,
     }
 }
 
+/// Object identity carried alongside a streaming GET body purely so a
+/// mid-stream failure names the object it happened on.
+#[derive(Clone, Default)]
+struct GetObjectReaderStreamDiagnostics {
+    bucket: String,
+    object: String,
+    request_id: String,
+}
+
 impl MemoryTrackedBytesStream {
     fn new(
         bytes: Bytes,
@@ -1107,8 +1120,19 @@ where
             remaining,
             emitted: 0,
             expected: remaining,
+            diagnostics: GetObjectReaderStreamDiagnostics::default(),
         }
     }
+
+    /// Attach the object identity a failed body should be reported against.
+    fn with_diagnostics(mut self, bucket: &str, object: &str, request_id: &str) -> Self {
+        self.diagnostics = GetObjectReaderStreamDiagnostics {
+            bucket: bucket.to_string(),
+            object: object.to_string(),
+            request_id: request_id.to_string(),
+        };
+        self
+    }
 }
 
 impl futures::Stream for MemoryTrackedBytesStream {
@@ -1569,12 +1593,29 @@ where
                     *this.emitted,
                     *this.remaining,
                 );
-                #[cfg(feature = "tracing-chunk-debug")]
-                tracing::error!(
-                    emitted = *this.emitted,
+                // The inner GetObjectStreamingReader is what normally reports a
+                // short body, so reaching this arm means the reader signalled a
+                // clean EOF while this layer still owed bytes against an
+                // already-committed Content-Length. That disagreement is a data
+                // plane fault, not chunk noise: log it unconditionally so the
+                // truncated object is named in the operator's log rather than
+                // only in a metric counter (issue #4784).
+                error!(
+                    event = EVENT_GET_OBJECT_STREAM_BODY,
+                    component = LOG_COMPONENT_APP,
+                    subsystem = LOG_SUBSYSTEM_OBJECT,
+                    bucket = %this.diagnostics.bucket,
+                    object = %this.diagnostics.object,
+                    request_id = %this.diagnostics.request_id,
+                    size_bucket = get_object_stream_size_bucket(*this.expected),
                     expected = *this.expected,
+                    emitted = *this.emitted,
+                    remaining = *this.remaining,
+                    strategy = this.strategy,
+                    buffer_source = this.buffer_source,
+                    state = "reader_stream_short_eof",
                     error = %err,
-                    "GetObject ReaderStream ended before expected length"
+                    "GetObject reader stream ended before the committed content length"
                 );
                 Poll::Ready(Some(Err(Box::new(err) as S3StdError)))
             }
@@ -1590,10 +1631,17 @@ where
                     *this.emitted,
                     *this.remaining,
                 );
+                // Deliberately not logged at warn here: every production body
+                // wraps a GetObjectStreamingReader, and that layer already
+                // reports this same error once with `state = "read_failed"` and
+                // the object identity. A second unconditional line per failed
+                // GET would read as two distinct faults. The chunk-debug build
+                // still gets this layer's view of the same error.
                 #[cfg(feature = "tracing-chunk-debug")]
                 tracing::error!(
                     emitted = *this.emitted,
                     expected = *this.expected,
+                    error_class = error_class,
                     error = %err,
                     "GetObject ReaderStream returned error"
                 );
@@ -1646,8 +1694,12 @@ where
 
 struct GetObjectStreamingReader<R> {
     inner: Option<R>,
-    // request_id + optional content_range are only used for diagnostic correlation and
-    // failure bucketing; they do not alter stream behavior.
+    // bucket/object + request_id + optional content_range are only used for diagnostic
+    // correlation and failure bucketing; they do not alter stream behavior. The object
+    // identity is what turns a mid-stream failure into an actionable report: a request_id
+    // alone cannot tell an operator which object reads short (issue #4784).
+    bucket: String,
+    object: String,
     request_id: String,
     content_range: Option<String>,
     expected: usize,
@@ -1666,8 +1718,8 @@ impl<R> GetObjectStreamingReader<R> {
     #[allow(clippy::too_many_arguments)]
     fn new(
         inner: R,
-        _bucket: &str,
-        _key: &str,
+        bucket: &str,
+        key: &str,
         request_id: &str,
         content_range: Option<String>,
         expected: usize,
@@ -1677,6 +1729,8 @@ impl<R> GetObjectStreamingReader<R> {
     ) -> Self {
         Self {
             inner: Some(inner),
+            bucket: bucket.to_string(),
+            object: key.to_string(),
             request_id: request_id.to_string(),
             content_range,
             expected,
@@ -1817,6 +1871,8 @@ impl<R> GetObjectStreamingReader<R> {
                 event = EVENT_GET_OBJECT_STREAM_BODY,
                 component = LOG_COMPONENT_APP,
                 subsystem = LOG_SUBSYSTEM_OBJECT,
+                bucket = %self.bucket,
+                object = %self.object,
                 request_id = %self.request_id,
                 range = %self.content_range.as_deref().unwrap_or("full"),
                 size_bucket = get_object_stream_size_bucket(self.expected),
@@ -1853,6 +1909,8 @@ impl<R: AsyncRead + Unpin> AsyncRead for GetObjectStreamingReader<R> {
                             event = EVENT_GET_OBJECT_STREAM_BODY,
                             component = LOG_COMPONENT_APP,
                             subsystem = LOG_SUBSYSTEM_OBJECT,
+                            bucket = %self.bucket,
+                            object = %self.object,
                             request_id = %self.request_id,
                             range = %self.content_range.as_deref().unwrap_or("full"),
                             size_bucket = get_object_stream_size_bucket(self.expected),
@@ -1871,10 +1929,12 @@ impl<R: AsyncRead + Unpin> AsyncRead for GetObjectStreamingReader<R> {
                         self.timer = None;
                         let failure_reason = Self::classify_read_error(&error);
                         self.finish_err();
-                        warn!(
+                        error!(
                             event = EVENT_GET_OBJECT_STREAM_BODY,
                             component = LOG_COMPONENT_APP,
                             subsystem = LOG_SUBSYSTEM_OBJECT,
+                            bucket = %self.bucket,
+                            object = %self.object,
                             request_id = %self.request_id,
                             range = %self.content_range.as_deref().unwrap_or("full"),
                             size_bucket = get_object_stream_size_bucket(self.expected),
@@ -1916,6 +1976,8 @@ impl<R: AsyncRead + Unpin> AsyncRead for GetObjectStreamingReader<R> {
                                         event = EVENT_GET_OBJECT_STREAM_BODY,
                                         component = LOG_COMPONENT_APP,
                                         subsystem = LOG_SUBSYSTEM_OBJECT,
+                                        bucket = %self.bucket,
+                                        object = %self.object,
                                         request_id = %self.request_id,
                                         range = %self.content_range.as_deref().unwrap_or("full"),
                                         size_bucket = get_object_stream_size_bucket(self.expected),
@@ -1956,10 +2018,12 @@ impl<R: AsyncRead + Unpin> AsyncRead for GetObjectStreamingReader<R> {
                             self.begin_resume(error);
                             continue;
                         }
-                        warn!(
+                        error!(
                             event = EVENT_GET_OBJECT_STREAM_BODY,
                             component = LOG_COMPONENT_APP,
                             subsystem = LOG_SUBSYSTEM_OBJECT,
+                            bucket = %self.bucket,
+                            object = %self.object,
                             request_id = %self.request_id,
                             range = %self.content_range.as_deref().unwrap_or("full"),
                             size_bucket = get_object_stream_size_bucket(self.expected),
@@ -1990,10 +2054,12 @@ impl<R: AsyncRead + Unpin> AsyncRead for GetObjectStreamingReader<R> {
                     let failure_reason = Self::classify_read_error(&err);
                     self.timer = None;
                     self.finish_err();
-                    warn!(
+                    error!(
                         event = EVENT_GET_OBJECT_STREAM_BODY,
                         component = LOG_COMPONENT_APP,
                         subsystem = LOG_SUBSYSTEM_OBJECT,
+                        bucket = %self.bucket,
+                        object = %self.object,
                         request_id = %self.request_id,
                         range = %self.content_range.as_deref().unwrap_or("full"),
                         size_bucket = get_object_stream_size_bucket(self.expected),
@@ -2029,6 +2095,8 @@ impl<R> Drop for GetObjectStreamingReader<R> {
             event = EVENT_GET_OBJECT_STREAM_BODY,
             component = LOG_COMPONENT_APP,
             subsystem = LOG_SUBSYSTEM_OBJECT,
+            bucket = %self.bucket,
+            object = %self.object,
             request_id = %self.request_id,
             range = %self.content_range.as_deref().unwrap_or("full"),
             size_bucket = get_object_stream_size_bucket(self.expected),
@@ -4303,7 +4371,8 @@ impl DefaultObjectUsecase {
             lifecycle,
             resume,
         );
-        let stream = GetObjectReaderStream::new(reader, stream_buffer_size, expected, stream_strategy.as_str(), buffer_source);
+        let stream = GetObjectReaderStream::new(reader, stream_buffer_size, expected, stream_strategy.as_str(), buffer_source)
+            .with_diagnostics(bucket, key, request_id);
         let blob = StreamingBlob::new(stream);
         if let Some(handoff_start) = handoff_start {
             rustfs_io_metrics::record_get_object_response_handoff(
@@ -16326,7 +16395,12 @@ mod tests {
         assert_eq!(body, vec![b'a'; 65]);
     }
 
+    // Serial with the capture test below: both drive the same short-EOF log
+    // callsite, and `tracing` caches callsite interest process-wide. Running
+    // this one concurrently on a thread with no subscriber re-caches that
+    // callsite as "never interested" and blinds the capture.
     #[tokio::test]
+    #[serial_test::serial]
     async fn get_object_reader_stream_errors_on_short_eof() {
         let stream = GetObjectReaderStream::new(
             std::io::Cursor::new(b"he".to_vec()),
@@ -16349,6 +16423,134 @@ mod tests {
         );
     }
 
+    /// Collects the structured fields of every event emitted while installed,
+    /// so a test can assert what an operator would actually read in the log
+    /// rather than only that an error value was returned.
+    type CapturedFieldMap = std::collections::HashMap<String, String>;
+    type CapturedEventLog = Arc<Mutex<Vec<CapturedFieldMap>>>;
+
+    struct CapturedEvents(CapturedEventLog);
+
+    struct CapturedFields(CapturedFieldMap);
+
+    impl tracing::field::Visit for CapturedFields {
+        fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
+            self.0.insert(field.name().to_string(), format!("{value:?}"));
+        }
+
+        fn record_str(&mut self, field: &tracing::field::Field, value: &str) {
+            self.0.insert(field.name().to_string(), value.to_string());
+        }
+    }
+
+    impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> for CapturedEvents {
+        fn on_event(&self, event: &tracing::Event<'_>, _ctx: tracing_subscriber::layer::Context<'_, S>) {
+            let mut fields = CapturedFields(CapturedFieldMap::new());
+            event.record(&mut fields);
+            self.0.lock().expect("captured events should not poison").push(fields.0);
+        }
+    }
+
+    fn capture_events() -> (CapturedEventLog, tracing::subscriber::DefaultGuard) {
+        use tracing_subscriber::{Registry, prelude::*};
+
+        let captured = Arc::new(Mutex::new(Vec::new()));
+        let subscriber = Registry::default().with(CapturedEvents(Arc::clone(&captured)));
+        let guard = tracing::subscriber::set_default(subscriber);
+        // `tracing` caches per-callsite interest process-wide, so a subscriber
+        // installed by a test running in parallel can leave the log sites below
+        // cached as "never interested" and this capture would silently see
+        // nothing. Force the callsites to re-ask the subscriber we just
+        // installed.
+        tracing::callsite::rebuild_interest_cache();
+        (captured, guard)
+    }
+
+    fn find_stream_body_event(captured: &CapturedEventLog, state: &str) -> CapturedFieldMap {
+        let events = captured.lock().expect("captured events should not poison");
+        events
+            .iter()
+            .find(|fields| fields.get("state").is_some_and(|value| value == state))
+            .unwrap_or_else(|| {
+                panic!(
+                    "a `{state}` streaming body failure must be logged, not only counted in a metric. \
+                     Captured {} event(s): {:?}",
+                    events.len(),
+                    events
+                )
+            })
+            .clone()
+    }
+
+    /// rustfs#4784: a GET body that ends short of its committed Content-Length
+    /// is the fault that breaks every downstream copier (replication, site
+    /// replication, `rclone sync`), yet this layer only fed a metric counter —
+    /// its log line was compiled out unless the `tracing-chunk-debug` feature
+    /// was on, so operators saw nothing on the source side.
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn get_object_reader_stream_short_eof_names_the_object() {
+        let (captured, _guard) = capture_events();
+
+        let stream = GetObjectReaderStream::new(
+            std::io::Cursor::new(b"he".to_vec()),
+            64,
+            5,
+            GetObjectStreamStrategy::Standard.as_str(),
+            GET_READER_STREAM_BUFFER_SOURCE_SELECTED,
+        )
+        .with_diagnostics("restic-paperless", "index/41b5a4c2344edb90", "req-reader-stream-short-eof");
+
+        stream
+            .collect::<Vec<_>>()
+            .await
+            .into_iter()
+            .collect::<Result<Vec<_>, _>>()
+            .expect_err("short reader should fail the streaming body");
+
+        let event = find_stream_body_event(&captured, "reader_stream_short_eof");
+        assert_eq!(event.get("bucket").map(String::as_str), Some("restic-paperless"));
+        assert_eq!(event.get("object").map(String::as_str), Some("index/41b5a4c2344edb90"));
+        assert_eq!(event.get("request_id").map(String::as_str), Some("req-reader-stream-short-eof"));
+        assert_eq!(event.get("expected").map(String::as_str), Some("5"));
+        assert_eq!(event.get("emitted").map(String::as_str), Some("2"));
+        assert_eq!(event.get("remaining").map(String::as_str), Some("3"));
+    }
+
+    /// The inner reader already logged mid-stream failures, but only under a
+    /// request_id — which cannot be resolved back to an object once the request
+    /// is gone. Without the identity the report in #4784 was unactionable.
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn get_object_streaming_reader_short_eof_names_the_object() {
+        use tokio::io::AsyncReadExt;
+
+        let (captured, _guard) = capture_events();
+
+        let mut reader = GetObjectStreamingReader::new(
+            std::io::Cursor::new(b"short".to_vec()),
+            "restic-paperless",
+            "index/41b5a4c2344edb90",
+            "req-streaming-short-eof",
+            None,
+            10,
+            Duration::ZERO,
+            GetObjectBodyLifecycle::tracked(GetObjectGuard::new()),
+            None,
+        );
+
+        let mut out = Vec::new();
+        reader
+            .read_to_end(&mut out)
+            .await
+            .expect_err("short body under a larger Content-Length must fail the stream");
+
+        let event = find_stream_body_event(&captured, "short_eof");
+        assert_eq!(event.get("bucket").map(String::as_str), Some("restic-paperless"));
+        assert_eq!(event.get("object").map(String::as_str), Some("index/41b5a4c2344edb90"));
+        assert_eq!(event.get("request_id").map(String::as_str), Some("req-streaming-short-eof"));
+    }
+
     #[test]
     fn get_object_stream_failure_labels_are_low_cardinality() {
         assert_eq!(get_object_stream_failure_reason("short_eof"), GET_STREAMING_BODY_FAILURE_REASON_SHORT_EOF);

rustfs/src/auth.rs

@@ -2210,6 +2210,89 @@ mod tests_policy {
         assert!(!policy.is_allowed(&args_fail).await, "IAM Policy should deny non-matching IP");
     }
 
+    /// The failure this issue is about: when `remote_addr` is dropped the
+    /// `aws:SourceIp` key never reaches the condition map, and `AddrFunc::evaluate`
+    /// returns `false` for an absent key. That flips two policy shapes in
+    /// opposite directions, and only one of them looks like a failure
+    /// (rustfs/backlog#1885).
+    #[tokio::test]
+    async fn source_ip_policies_break_in_both_directions_when_the_key_is_missing() {
+        let allow_from_office = |effect: &str| {
+            format!(
+                r#"{{
+                    "Version": "2012-10-17",
+                    "Statement": [
+                        {{"Effect": "Allow", "Action": ["admin:ConfigUpdate"], "Resource": ["arn:aws:s3:::*"]}},
+                        {{
+                            "Effect": "{effect}",
+                            "Action": ["admin:ConfigUpdate"],
+                            "Resource": ["arn:aws:s3:::*"],
+                            "Condition": {{"IpAddress": {{"aws:SourceIp": "192.168.1.0/24"}}}}
+                        }}
+                    ]
+                }}"#
+            )
+        };
+
+        let claims = HashMap::new();
+        let groups = None;
+        let mut with_ip = HashMap::new();
+        with_ip.insert("SourceIp".to_string(), vec!["192.168.1.10".to_string()]);
+        let without_ip: HashMap<String, Vec<String>> = HashMap::new();
+
+        let args_with_ip = Args {
+            account: "test-account",
+            groups: &groups,
+            action: Action::AdminAction(rustfs_policy::policy::action::AdminAction::ConfigUpdateAdminAction),
+            bucket: "",
+            conditions: &with_ip,
+            is_owner: false,
+            object: "",
+            claims: &claims,
+            deny_only: false,
+        };
+        let args_without_ip = Args {
+            conditions: &without_ip,
+            ..args_with_ip
+        };
+
+        // Deny + blacklist: the bypass shape. With the key present the deny
+        // matches and the request is refused; drop the key and the deny stops
+        // matching, so a source that policy means to block gets through.
+        let deny_policy: Policy = serde_json::from_str(&allow_from_office("Deny")).expect("deny policy parses");
+        assert!(
+            !deny_policy.is_allowed(&args_with_ip).await,
+            "a blacklisted source must be refused while aws:SourceIp is present"
+        );
+        assert!(
+            deny_policy.is_allowed(&args_without_ip).await,
+            "dropping remote_addr makes the Deny statement unreachable — this is the bypass"
+        );
+
+        // Allow + whitelist: the availability shape, and the only one an
+        // operator would notice, which is why the bypass above went unseen.
+        let allow_policy: Policy = serde_json::from_str(
+            r#"{
+                "Version": "2012-10-17",
+                "Statement": [{
+                    "Effect": "Allow",
+                    "Action": ["admin:ConfigUpdate"],
+                    "Resource": ["arn:aws:s3:::*"],
+                    "Condition": {"IpAddress": {"aws:SourceIp": "192.168.1.0/24"}}
+                }]
+            }"#,
+        )
+        .expect("allow policy parses");
+        assert!(
+            allow_policy.is_allowed(&args_with_ip).await,
+            "a whitelisted source must be allowed while aws:SourceIp is present"
+        );
+        assert!(
+            !allow_policy.is_allowed(&args_without_ip).await,
+            "dropping remote_addr locks out a legitimate admin"
+        );
+    }
+
     #[tokio::test]
     async fn test_bucket_policy_source_ip() {
         let policy_json = r#"{

scripts/find_assertless_tests.py

@@ -81,6 +81,78 @@ FN_LINE = re.compile(r"^\s*(?:pub\s+)?(?:async\s+)?fn\s+([a-zA-Z0-9_]+)")
 
 
 
+# A char literal is 'x' or '\n'; a lone `'` is a lifetime (`&'a str`), and
+# consuming to the next quote on one would swallow the rest of the line.
+CHAR_LITERAL = re.compile(r"'(?:[^'\\]|\\.)'")
+RAW_STRING_OPEN = re.compile(r'r(#*)"')
+
+
+class LiteralStripper:
+    """Blanks out literals and comments so brace matching sees only code.
+
+    Carries state across lines: Rust string literals — the JSON and `r#"..."#`
+    fixtures these tests are full of — routinely span lines, and a per-line
+    scanner falls out of phase on the first one. A `{` inside a string would
+    otherwise unbalance the count and truncate a test body before its
+    assertions.
+    """
+
+    def __init__(self) -> None:
+        self.in_string = False
+        self.raw_hashes = None  # None when the open string is not raw
+
+    def feed(self, line: str) -> str:
+        out = []
+        i = 0
+        n = len(line)
+        while i < n:
+            if self.in_string:
+                if self.raw_hashes is not None:
+                    close = '"' + "#" * self.raw_hashes
+                    idx = line.find(close, i)
+                    if idx == -1:
+                        return "".join(out)
+                    i = idx + len(close)
+                    self.in_string = False
+                    self.raw_hashes = None
+                    continue
+                if line[i] == "\\":
+                    i += 2
+                    continue
+                if line[i] == '"':
+                    self.in_string = False
+                    i += 1
+                    continue
+                i += 1
+                continue
+
+            ch = line[i]
+            if ch == "/" and i + 1 < n and line[i + 1] == "/":
+                break
+            m = RAW_STRING_OPEN.match(line, i)
+            if m:
+                self.in_string = True
+                self.raw_hashes = len(m.group(1))
+                i = m.end()
+                continue
+            if ch == '"':
+                self.in_string = True
+                self.raw_hashes = None
+                i += 1
+                continue
+            if ch == "'":
+                cm = CHAR_LITERAL.match(line, i)
+                if cm:
+                    i = cm.end()
+                    continue
+                out.append(ch)
+                i += 1
+                continue
+            out.append(ch)
+            i += 1
+        return "".join(out)
+
+
 def extract_body(text: str) -> str:
     """Return what is between the outermost braces of a scanned function."""
     start = text.find("{")
@@ -121,8 +193,9 @@ def scan_file(path: Path):
         begun = False
         body = []
         k = j
+        stripper = LiteralStripper()
         while k < len(lines):
-            for ch in lines[k]:
+            for ch in stripper.feed(lines[k]):
                 if ch == "{":
                     depth += 1
                     begun = True