chore: adjudicate the last 18 bare dead_code allows in the library crates (#6265)

* chore: adjudicate the last 18 bare dead_code allows in the library crates

Finishes backlog#1823 step 10 outside `rustfs/src` and `protocols`: config, s3select-query, common, madmin, heal, ecstore, signer and notify. Stripped first, then clippy asked which the compiler actually missed — 8 of the 18 were inert.

Seven items are deleted, each checked by grep as well as by clippy:

- `common/last_minute.rs`'s private `TimedAction` (with its impl) and `SizeCategory` (with its `Display` impl). The file's public surface — `AccElem`, `LastMinuteLatency` — stays; ecstore consumes it.
- `s3select-query`'s three `with_*` builders. `DefaultLogicalOptimizer::with_optimizer_rules` looks used, but the call in the same file is `SessionStateBuilder::with_optimizer_rules` from DataFusion; the local methods have no callers.
- `heal/manager.rs`'s `contains_key`. Its six apparent references are all `HashMap::contains_key`.

Three keep their code:

- `heal/storage.rs`'s `Test` variant is constructed by the `#[cfg(test)] test()` helper, which the lib target cannot see, so it takes a reasoned allow.
- `signer`'s `STREAMING_PAYLOAD_HDR` and `try_build_chunk_string_to_sign` gain the `_` prefix instead. That file already marks deliberately-unheld code that way — `_STREAMING_TRAILER_HDR`, `_PAYLOAD_CHUNK_SIZE`, and `_try_build_chunk_signature`, which is the only caller of that function. Following the existing convention removes the allow without an attribute.

`protocols` keeps its four; that crate needs `--features swift,sftp` to compile fully and is verified differently. The four `#![allow(dead_code)]` in `e2e_test` are module-root blankets in test-support files, which belong to steps 1-5 rather than step 10.

Refs backlog#1823

* chore(e2e_test): adjudicate the two dead_code allows the lib test target still needs

`cargo clippy --all-targets` compiles e2e_test's lib test target, which the earlier pass did not cover, so these two removals only surfaced in CI.

test_large_multipart_upload's allow was load-bearing: its call site in test_local_kms_multipart_upload is commented out behind "TODO: Re-enable after fixing streaming encryption issues with large files". The allow comes back with the reason string this batch uses everywhere else, so the next reader sees why it is parked instead of deleting a test we intend to run again.

TestDefinition.category was the opposite: written at all six definitions, read nowhere, and its enum's impl block is empty. The live copy of that type is crates/e2e_test/src/kms/test_runner.rs, which has an as_str; the policy copy is a vestige of it. Dropping the field, the enum, and the constructor parameter leaves the runner unchanged — it dispatches on name and filters on is_critical.

Verification: cargo clippy --all-targets -- -D warnings (workspace, the CI command) and cargo fmt --all --check both pass.

---------

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

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/common/src/last_minute.rs

@@ -13,82 +13,6 @@
 // limitations under the License.
 
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
-
-#[allow(dead_code)]
-#[derive(Debug, Default)]
-struct TimedAction {
-    count: u64,
-    acc_time: u64,
-    min_time: Option<u64>,
-    max_time: Option<u64>,
-    bytes: u64,
-}
-
-#[allow(dead_code)]
-impl TimedAction {
-    // Avg returns the average time spent on the action.
-    pub fn avg(&self) -> Option<Duration> {
-        if self.count == 0 {
-            return None;
-        }
-        Some(Duration::from_nanos(self.acc_time / self.count))
-    }
-
-    // AvgBytes returns the average bytes processed.
-    pub fn avg_bytes(&self) -> u64 {
-        if self.count == 0 {
-            return 0;
-        }
-        self.bytes / self.count
-    }
-
-    // Merge other into t.
-    pub fn merge(&mut self, other: TimedAction) {
-        self.count += other.count;
-        self.acc_time += other.acc_time;
-        self.bytes += other.bytes;
-
-        if self.count == 0 {
-            self.min_time = other.min_time;
-        }
-        if let Some(other_min) = other.min_time {
-            self.min_time = self.min_time.map_or(Some(other_min), |min| Some(min.min(other_min)));
-        }
-
-        self.max_time = self
-            .max_time
-            .map_or(other.max_time, |max| Some(max.max(other.max_time.unwrap_or(0))));
-    }
-}
-
-#[allow(dead_code)]
-#[derive(Debug)]
-enum SizeCategory {
-    SizeLessThan1KiB = 0,
-    SizeLessThan1MiB,
-    SizeLessThan10MiB,
-    SizeLessThan100MiB,
-    SizeLessThan1GiB,
-    SizeGreaterThan1GiB,
-    // Add new entries here
-    SizeLastElemMarker,
-}
-
-impl std::fmt::Display for SizeCategory {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        let s = match *self {
-            SizeCategory::SizeLessThan1KiB => "SizeLessThan1KiB",
-            SizeCategory::SizeLessThan1MiB => "SizeLessThan1MiB",
-            SizeCategory::SizeLessThan10MiB => "SizeLessThan10MiB",
-            SizeCategory::SizeLessThan100MiB => "SizeLessThan100MiB",
-            SizeCategory::SizeLessThan1GiB => "SizeLessThan1GiB",
-            SizeCategory::SizeGreaterThan1GiB => "SizeGreaterThan1GiB",
-            SizeCategory::SizeLastElemMarker => "SizeLastElemMarker",
-        };
-        write!(f, "{s}")
-    }
-}
-
 #[derive(Clone, Debug, Default, Copy)]
 pub struct AccElem {
     pub total: u64,

crates/common/src/mrf_channel.rs

@@ -148,6 +148,62 @@ fn unix_now_ms() -> u64 {
         .unwrap_or(0)
 }
 
+/// A repair the MRF consumer landed, fanned out so retry ledgers can drop
+/// entries the journal no longer tracks (backlog#1894 axis B). The payload
+/// mirrors the intent identity so consumers match without re-parsing.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct MrfRepairedEvent {
+    pub bucket: Arc<str>,
+    pub object: Arc<str>,
+    pub version_id: Option<[u8; 16]>,
+}
+
+/// Bound on the repaired-event backlog. Notices are best-effort hints; when
+/// the ring is full the oldest are dropped and the affected ledger entries
+/// simply expire through their own attempts/age limits.
+const MRF_REPAIRED_EVENT_CAP: usize = 4096;
+
+static MRF_REPAIRED_EVENTS: OnceLock<std::sync::Mutex<std::collections::VecDeque<MrfRepairedEvent>>> = OnceLock::new();
+
+/// Record that the MRF consumer landed a repair. Never blocks: the critical
+/// section is a deque push under a std mutex.
+pub fn note_mrf_repaired(bucket: &str, object: &str, version_id: Option<[u8; 16]>) {
+    let registry = MRF_REPAIRED_EVENTS.get_or_init(|| std::sync::Mutex::new(std::collections::VecDeque::new()));
+    let Ok(mut events) = registry.lock() else {
+        return;
+    };
+    if events.len() >= MRF_REPAIRED_EVENT_CAP {
+        events.pop_front();
+    }
+    events.push_back(MrfRepairedEvent {
+        bucket: Arc::from(bucket),
+        object: Arc::from(object),
+        version_id,
+    });
+}
+
+/// Take the repair notices recorded for `bucket`, leaving other buckets'
+/// notices in place for their own scanners.
+pub fn take_mrf_repaired_events_for(bucket: &str) -> Vec<MrfRepairedEvent> {
+    let Some(registry) = MRF_REPAIRED_EVENTS.get() else {
+        return Vec::new();
+    };
+    let Ok(mut events) = registry.lock() else {
+        return Vec::new();
+    };
+    let mut taken = Vec::new();
+    let mut retained = std::collections::VecDeque::with_capacity(events.len());
+    while let Some(event) = events.pop_front() {
+        if event.bucket.as_ref() == bucket {
+            taken.push(event);
+        } else {
+            retained.push_back(event);
+        }
+    }
+    *events = retained;
+    taken
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -200,4 +256,32 @@ mod tests {
         assert!(!try_send_mrf_intent(MrfKind::MetadataCorruption, "b", "o", None));
         set_mrf_delivery_enabled(true);
     }
+
+    #[test]
+    fn repaired_events_take_is_bucket_scoped_and_cap_bounded() {
+        // Distinct buckets keep their notices until their own scanner takes
+        // them; a take for one bucket leaves the others' notices in place.
+        note_mrf_repaired("bucket-a", "object-1", None);
+        note_mrf_repaired("bucket-b", "object-2", None);
+        note_mrf_repaired("bucket-a", "object-3", None);
+
+        let taken_a = take_mrf_repaired_events_for("bucket-a");
+        assert_eq!(taken_a.len(), 2);
+        assert_eq!(taken_a[0].object.as_ref(), "object-1");
+        assert_eq!(taken_a[1].object.as_ref(), "object-3");
+        assert!(take_mrf_repaired_events_for("bucket-a").is_empty(), "take is destructive per bucket");
+
+        let taken_b = take_mrf_repaired_events_for("bucket-b");
+        assert_eq!(taken_b.len(), 1);
+        assert_eq!(taken_b[0].object.as_ref(), "object-2");
+
+        // Cap bound: flooding the ring drops the oldest notices rather than
+        // growing unbounded.
+        for i in 0..=(MRF_REPAIRED_EVENT_CAP + 8) {
+            note_mrf_repaired("flood-bucket", &format!("object-{i}"), None);
+        }
+        let flooded = take_mrf_repaired_events_for("flood-bucket");
+        assert_eq!(flooded.len(), MRF_REPAIRED_EVENT_CAP);
+        assert_eq!(flooded[0].object.as_ref(), "object-9", "the oldest notices past the cap are dropped");
+    }
 }

crates/config/src/notify/mod.rs

@@ -92,15 +92,11 @@ pub const NOTIFY_SUB_SYSTEMS: &[&str] = &[
 pub const NOTIFY_KAFKA_SUB_SYS: &str = "notify_kafka";
 pub const NOTIFY_MQTT_SUB_SYS: &str = "notify_mqtt";
 pub const NOTIFY_MYSQL_SUB_SYS: &str = "notify_mysql";
-#[allow(dead_code)]
 pub const NOTIFY_NATS_SUB_SYS: &str = "notify_nats";
-#[allow(dead_code)]
 pub const NOTIFY_NSQ_SUB_SYS: &str = "notify_nsq";
-#[allow(dead_code)]
 pub const NOTIFY_ES_SUB_SYS: &str = "notify_elasticsearch";
 pub const NOTIFY_AMQP_SUB_SYS: &str = "notify_amqp";
 pub const NOTIFY_POSTGRES_SUB_SYS: &str = "notify_postgres";
-#[allow(dead_code)]
 pub const NOTIFY_REDIS_SUB_SYS: &str = "notify_redis";
 pub const NOTIFY_REDIS_DEFAULT_CHANNEL: &str = "rustfs_notify_channel";
 pub const NOTIFY_PULSAR_SUB_SYS: &str = "notify_pulsar";

crates/e2e_test/src/kms/kms_local_test.rs

@@ -647,7 +647,10 @@ async fn test_multipart_upload_with_sse_c(
 }
 
 /// Test large multipart upload to verify streaming encryption works correctly
-#[allow(dead_code)]
+#[allow(
+    dead_code,
+    reason = "parked behind the TODO in test_local_kms_multipart_upload until streaming encryption is fixed for large files (backlog#1823)"
+)]
 async fn test_large_multipart_upload(
     s3_client: &aws_sdk_s3::Client,
     bucket: &str,

crates/e2e_test/src/policy/test_runner.rs

@@ -19,32 +19,17 @@ use std::time::Instant;
 use tokio::time::{Duration, sleep};
 use tracing::{error, info};
 
-/// Core test categories
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub enum TestCategory {
-    SingleValue,
-    MultiValue,
-    Concatenation,
-    Nested,
-    DenyScenarios,
-}
-
-impl TestCategory {}
-
 /// Test case definition
 #[derive(Debug, Clone)]
 pub struct TestDefinition {
     pub name: String,
-    #[allow(dead_code)]
-    pub category: TestCategory,
     pub is_critical: bool,
 }
 
 impl TestDefinition {
-    pub fn new(name: impl Into<String>, category: TestCategory, is_critical: bool) -> Self {
+    pub fn new(name: impl Into<String>, is_critical: bool) -> Self {
         Self {
             name: name.into(),
-            category,
             is_critical,
         }
     }
@@ -92,12 +77,12 @@ impl PolicyTestSuite {
     /// Create default test suite
     pub fn new() -> Self {
         let tests = vec![
-            TestDefinition::new("test_aws_policy_variables_single_value", TestCategory::SingleValue, true),
+            TestDefinition::new("test_aws_policy_variables_single_value", true),
-            TestDefinition::new("test_aws_policy_variables_multi_value", TestCategory::MultiValue, true),
+            TestDefinition::new("test_aws_policy_variables_multi_value", true),
-            TestDefinition::new("test_aws_policy_variables_concatenation", TestCategory::Concatenation, true),
+            TestDefinition::new("test_aws_policy_variables_concatenation", true),
-            TestDefinition::new("test_aws_policy_variables_nested", TestCategory::Nested, true),
+            TestDefinition::new("test_aws_policy_variables_nested", true),
-            TestDefinition::new("test_aws_policy_variables_deny", TestCategory::DenyScenarios, true),
+            TestDefinition::new("test_aws_policy_variables_deny", true),
-            TestDefinition::new("test_aws_policy_variables_sts", TestCategory::SingleValue, true),
+            TestDefinition::new("test_aws_policy_variables_sts", true),
         ];
 
         Self {

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 {
+/// A data-movement stage failure that keeps the error it wrapped.
-    Error::other(format!("{op_label}: {stage} failed for {bucket}/{object}: {err}"))
+///
+/// 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/ecstore/src/set_disk/core/io_primitives.rs

@@ -1095,6 +1095,14 @@ pub(in crate::set_disk) struct ReadRepairHealSubmission<'a> {
     pub(in crate::set_disk) set_index: usize,
     pub(in crate::set_disk) part_number: Option<usize>,
     pub(in crate::set_disk) reason: &'static str,
+    /// Durable MRF journal intent to file alongside the read-repair request
+    /// (backlog#1894 axis A): the intent kind plus its native `Uuid`
+    /// version id (the submission's string form stays display-only). Bound
+    /// to the reservation — the intent is only delivered when this sighting
+    /// wins the dedup TTL, so a burst of reads failing on the same object
+    /// books exactly one journal record instead of one per retry. `None`
+    /// keeps the historical no-intent behavior.
+    pub(in crate::set_disk) mrf_intent: Option<(rustfs_common::mrf_channel::MrfKind, Option<uuid::Uuid>)>,
 }
 
 pub(in crate::set_disk) fn send_read_repair_heal_request(
@@ -1126,6 +1134,7 @@ pub(in crate::set_disk) async fn submit_read_repair_heal(
             set_index,
             part_number,
             reason,
+            mrf_intent: None,
         },
         send_read_repair_heal_request,
     )
@@ -1144,6 +1153,7 @@ pub(in crate::set_disk) async fn submit_read_repair_heal_with_submitter(
         set_index,
         part_number,
         reason,
+        mrf_intent,
     } = submission;
 
     let Some(dedup_key) = reserve_read_repair_heal(bucket, object, version_id, pool_index, set_index).await else {
@@ -1155,6 +1165,12 @@ pub(in crate::set_disk) async fn submit_read_repair_heal_with_submitter(
         return;
     };
 
+    // Reservation won: this sighting owns the repair records for the object,
+    // including the durable journal intent when the caller asked for one.
+    if let Some((kind, version_uuid)) = mrf_intent {
+        rustfs_common::mrf_channel::try_send_mrf_intent(kind, bucket, object, version_uuid);
+    }
+
     let mut request = rustfs_common::heal_channel::create_heal_request_with_options(
         bucket.to_string(),
         Some(object.to_string()),
@@ -8710,6 +8726,42 @@ mod tests {
         assert_eq!(responses[0].error, Error::ErasureReadQuorum.to_string());
     }
 
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn mrf_intent_is_filed_once_per_read_repair_reservation() {
+        // Serial: owns the process-global MRF channel for this test binary
+        // (same key as the other channel-owning tests above).
+        let bucket = format!("mrf-intent-bucket-{}", Uuid::new_v4());
+        let object = format!("object-{}", Uuid::new_v4());
+        let mut receiver = rustfs_common::mrf_channel::init_mrf_channel().expect("first channel init in this binary");
+        rustfs_common::mrf_channel::set_mrf_delivery_enabled(true);
+
+        fn intent_submission<'a>(bucket: &'a str, object: &'a str) -> ReadRepairHealSubmission<'a> {
+            ReadRepairHealSubmission {
+                bucket,
+                object,
+                version_id: None,
+                pool_index: 9,
+                set_index: 9,
+                part_number: Some(1),
+                reason: "decode_error",
+                mrf_intent: Some((rustfs_common::mrf_channel::MrfKind::DecodeFailure, None)),
+            }
+        }
+
+        // First sighting wins the reservation: the journal intent is filed
+        // synchronously before the admission task is spawned.
+        submit_read_repair_heal_with_submitter(intent_submission(&bucket, &object), accepted_read_repair_submitter).await;
+        let first = receiver.try_recv().expect("first sighting must file exactly one MRF intent");
+        assert_eq!(*first.bucket, bucket);
+        assert_eq!(*first.object, object);
+
+        // Second sighting within the dedup TTL is a duplicate: no request, no
+        // second journal record.
+        submit_read_repair_heal_with_submitter(intent_submission(&bucket, &object), accepted_read_repair_submitter).await;
+        assert!(receiver.try_recv().is_err(), "duplicate sighting must not file another MRF intent");
+    }
+
     #[tokio::test]
     async fn reserve_read_repair_heal_dedupes_by_object_version_and_set() {
         let object = format!("object-{}", Uuid::new_v4());
@@ -8818,6 +8870,7 @@ mod tests {
                 set_index: 2,
                 part_number: Some(1),
                 reason: "test",
+                mrf_intent: None,
             },
             failed_read_repair_submitter,
         )
@@ -8846,6 +8899,7 @@ mod tests {
                 set_index: 3,
                 part_number: Some(2),
                 reason: "test",
+                mrf_intent: None,
             },
             dropped_read_repair_submitter,
         )
@@ -8874,6 +8928,7 @@ mod tests {
                 set_index: 4,
                 part_number: None,
                 reason: "test",
+                mrf_intent: None,
             },
             accepted_read_repair_submitter,
         )

crates/ecstore/src/set_disk/read.rs

@@ -1077,23 +1077,23 @@ impl SetDisks {
                             "Recoverable decode error triggered read repair"
                         );
                         let version_id = fi.version_id.as_ref().map(ToString::to_string);
-                        // MRF journal intent: keeps a durable Urgent ECDecode
+                        // Single-flight (backlog#1894 axis A): the durable
-                        // request alive across restarts even when the in-memory
+                        // MRF intent (Urgent ECDecode across restarts, HS-01)
-                        // read-repair request is dropped or lost (HS-01).
+                        // is bound to the read-repair reservation, so only the
-                        rustfs_common::mrf_channel::try_send_mrf_intent(
+                        // first sighting within the dedup TTL books a journal
-                            rustfs_common::mrf_channel::MrfKind::DecodeFailure,
+                        // record instead of one per retried read.
-                            bucket,
+                        submit_read_repair_heal_with_submitter(
-                            object,
+                            ReadRepairHealSubmission {
-                            fi.version_id,
+                                bucket,
-                        );
+                                object,
-                        submit_read_repair_heal(
+                                version_id: version_id.as_deref(),
-                            bucket,
+                                pool_index,
-                            object,
+                                set_index,
-                            version_id.as_deref(),
+                                part_number: Some(part_number),
-                            pool_index,
+                                reason: "decode_error",
-                            set_index,
+                                mrf_intent: Some((rustfs_common::mrf_channel::MrfKind::DecodeFailure, fi.version_id)),
-                            Some(part_number),
+                            },
-                            "decode_error",
+                            send_read_repair_heal_request,
                         )
                         .await;
                         has_err = false;
@@ -2577,6 +2577,7 @@ mod metadata_cache_tests {
                 set_index: 0,
                 part_number: Some(1),
                 reason: "missing_shards",
+                mrf_intent: None,
             },
             slow_read_repair_submitter,
         )
@@ -2611,6 +2612,7 @@ mod metadata_cache_tests {
                 set_index: 0,
                 part_number: Some(1),
                 reason: "missing_shards",
+                mrf_intent: None,
             },
             dropped_read_repair_submitter,
         )
@@ -2647,6 +2649,7 @@ mod metadata_cache_tests {
                 set_index: 0,
                 part_number: Some(1),
                 reason: "missing_shards",
+                mrf_intent: None,
             },
             capture_read_repair_submitter,
         )

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;
@@ -597,14 +597,6 @@ impl PriorityHealQueue {
             }
         }
     }
-
-    /// Check if a request with the same key already exists in the queue
-    #[allow(dead_code)]
-    fn contains_key(&self, request: &HealRequest) -> bool {
-        let key = Self::make_dedup_key(request);
-        self.dedup_keys.contains_key(&key)
-    }
-
     /// Check if an erasure set heal request for a specific set_disk_id exists
     fn contains_erasure_set(&self, set_disk_id: &str) -> bool {
         let key = format!("erasure_set:{set_disk_id}");
@@ -1027,8 +1019,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.
+    /// Recently completed heal statuses retained for status queries. Values
-    completed_heals: Arc<Mutex<HashMap<String, CompletedHealStatus>>>,
+    /// 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 +1045,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 +2165,79 @@ impl HealManager {
     }
 
     /// Get task status
-    pub async fn get_task_status(&self, task_id: &str) -> Result<HealTaskStatus> {
+    /// Ordered task-state lookup shared by every status/report query. The
-        let canonical_task_id = self.canonical_task_id(task_id).await;
+    /// 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) {
+            if let Some(task) = active_heals
-                return Ok(task.get_status().await);
+                .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) {
+            if let Some(retrying) = retrying_heals
-                return Ok(retrying.status());
+                .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;
+            let queue = self.heal_queue.lock().await;
-            prune_completed_heal_statuses(&mut completed_heals);
+            let queued = match heal_path {
-            if let Some(completed) = completed_heals.get(&canonical_task_id)
+                Some(path) => queue.contains_request_id_matching_path(canonical_task_id, path),
-                && completed_status_is_retrying(&completed.status)
+                None => queue.contains_request_id(canonical_task_id),
-            {
+            };
-                return Ok(completed.status.clone());
+            if queued {
+                return TaskStateLookup::Queued;
             }
         }
 
-        let queue = self.heal_queue.lock().await;
+        match terminal_completed {
-        if queue.contains_request_id(&canonical_task_id) {
+            Some(completed) => TaskStateLookup::Completed(completed),
-            return Ok(HealTaskStatus::Pending);
+            None => TaskStateLookup::NotFound,
         }
-        drop(queue);
+    }
 
-        let mut completed_heals = self.completed_heals.lock().await;
+    pub async fn get_task_status(&self, task_id: &str) -> Result<HealTaskStatus> {
-        prune_completed_heal_statuses(&mut completed_heals);
+        let canonical_task_id = self.canonical_task_id(task_id).await;
-        if let Some(completed) = completed_heals.get(&canonical_task_id) {
+        match self.lookup_task_state(&canonical_task_id, None).await {
-            return Ok(completed.status.clone());
+            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 +2249,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;
-        {
+        match self.lookup_task_state(&canonical_task_id, None).await {
-            let active_heals = self.active_heals.lock().await;
+            TaskStateLookup::Active(task) => Ok(active_task_report(&task, since).await),
-            if let Some(task) = active_heals.get(&canonical_task_id) {
+            TaskStateLookup::Retrying(status) => Ok(empty_task_report(status)),
-                return Ok(active_task_report(task, since).await);
+            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 +2272,20 @@ impl HealManager {
         since: Option<u64>,
     ) -> Result<HealTaskReport> {
         let canonical_task_id = self.canonical_task_id(task_id).await;
-        {
+        match self.lookup_task_state(&canonical_task_id, Some(heal_path)).await {
-            let active_heals = self.active_heals.lock().await;
+            TaskStateLookup::Active(task) => Ok(active_task_report(&task, since).await),
-            if let Some(task) = active_heals.get(&canonical_task_id)
+            TaskStateLookup::Retrying(status) => Ok(empty_task_report(status)),
-                && heal_type_matches_path(&task.heal_type, heal_path)
+            TaskStateLookup::Completed(completed) => Ok(completed_task_report(&completed, since)),
-            {
+            TaskStateLookup::Queued => Ok(empty_task_report(HealTaskStatus::Pending)),
-                return Ok(active_task_report(task, since).await);
+            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 +2295,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;
-        {
+        match self.lookup_task_state(&canonical_task_id, Some(heal_path)).await {
-            let active_heals = self.active_heals.lock().await;
+            TaskStateLookup::Active(task) => Ok(task.get_status().await),
-            if let Some(task) = active_heals.get(&canonical_task_id)
+            TaskStateLookup::Retrying(status) => Ok(status),
-                && heal_type_matches_path(&task.heal_type, heal_path)
+            TaskStateLookup::Completed(completed) => Ok(completed.status.clone()),
-            {
+            TaskStateLookup::Queued => Ok(HealTaskStatus::Pending),
-                return Ok(task.get_status().await);
+            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 +3431,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,
+                            seqed_items,
-                            next_seq: final_window.next_seq,
+                            next_seq,
-                            min_seq: final_window.min_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 +3739,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 +5206,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 +5915,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 +5963,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 +5992,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(),
+                seqed_items: vec![(
-                next_seq: 0,
+                    1,
-                min_seq: 0,
+                    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 +6024,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/mrf_queue.rs

@@ -25,9 +25,12 @@
 //! set, rewritten on a group-commit cadence (every flush interval or flush
 //! threshold new intents). A rewrite is atomic at the record level only — a
 //! torn tail simply truncates during replay because every record carries its
-//! own CRC32. Losing the last flush window (≤500 ms) is acceptable: replayed
+//! own CRC32. Losing the last flush window (≤500 ms) is acceptable because
-//! duplicates are merged by the manager's dedup key, and read-repair remains
+//! every producer keeps its own safety net: read-repair re-detects on the
-//! the safety net.
+//! next failing read, and the scanner's corrupt-metadata branch leaves a
+//! pending-ledger entry behind even when its MRF intent is accepted
+//! (backlog#1894 axis A), so a lost intent is retried by the ledger rather
+//! than waiting for the failed-object TTL to re-scan the path.
 
 use super::{DiskStore, HealDiskExt as _, local_disk_map_read};
 use crate::heal::manager::HealManager;
@@ -409,8 +412,13 @@ impl MrfRuntime {
             let request = build_heal_request(&intent);
             match manager.submit_heal_request(request).await {
                 // Accepted intents leave the pending set; the next flush persists the
-                // smaller snapshot, which is the journal's compaction.
+                // smaller snapshot, which is the journal's compaction. Fan out a
-                Ok(HealAdmissionResult::Accepted) | Ok(HealAdmissionResult::Merged) => {}
+                // best-effort repaired notice so retry ledgers (the scanner's
+                // pending-heal oracle) can drop entries whose repair the manager
+                // now owns (backlog#1894 axis B).
+                Ok(HealAdmissionResult::Accepted) | Ok(HealAdmissionResult::Merged) => {
+                    rustfs_common::mrf_channel::note_mrf_repaired(&intent.bucket, &intent.object, intent.version_id);
+                }
                 Ok(HealAdmissionResult::Full) | Ok(HealAdmissionResult::Dropped(HealAdmissionDropReason::QueueFull)) => {
                     intent.attempts = intent.attempts.saturating_add(1);
                     if intent.attempts >= MRF_MAX_ATTEMPTS {
@@ -514,7 +522,9 @@ async fn replay_into(
         while let Some(mut intent) = queue.pop_front() {
             let request = build_heal_request(&intent);
             match manager.submit_heal_request(request).await {
-                Ok(HealAdmissionResult::Accepted) | Ok(HealAdmissionResult::Merged) => {}
+                Ok(HealAdmissionResult::Accepted) | Ok(HealAdmissionResult::Merged) => {
+                    rustfs_common::mrf_channel::note_mrf_repaired(&intent.bucket, &intent.object, intent.version_id);
+                }
                 Ok(HealAdmissionResult::Full) | Ok(HealAdmissionResult::Dropped(HealAdmissionDropReason::QueueFull)) => {
                     intent.attempts = intent.attempts.saturating_add(1);
                     if intent.attempts < MRF_MAX_ATTEMPTS {

crates/heal/src/heal/storage.rs

@@ -42,7 +42,10 @@ pub struct HealLifecycleExpiryContext {
 
 enum HealLifecycleExpiryContextInner {
     Ecstore(EcstoreHealLifecycleExpiryContext),
-    #[allow(dead_code)]
+    #[allow(
+        dead_code,
+        reason = "constructed by the #[cfg(test)] `test()` helper; the lib target cannot see test-only consumers (backlog#1823)"
+    )]
     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);
+                .store(result_items.front().map_or(seq, |(oldest, _)| *oldest), Ordering::Relaxed);
-            result_items.push((seq, result));
+            result_items.push_back((seq, result));
             self.result_items_truncated.store(true, Ordering::Relaxed);
         }
     }

crates/heal/tests/mrf_pipeline_test.rs

@@ -127,6 +127,20 @@ async fn decode_failure_intent_maps_to_urgent_mrf_heal_request() {
         "MRF intent must reach the manager queue as an Urgent request (snapshot: {:?})",
         manager.operations_snapshot().await
     );
+
+    // Axis B (backlog#1894): the accepted dispatch must also fan out a
+    // repaired notice for the intent's bucket so the scanner ledger can drop
+    // its retry entry for the same target. Polled: the queue observation
+    // above can land between the manager push and the consumer's notice.
+    let noticed = wait_until(Duration::from_secs(10), || async {
+        !mrf_channel::take_mrf_repaired_events_for("mrf-bucket").is_empty()
+    })
+    .await;
+    assert!(noticed, "accepted intent must fan out a repaired notice");
+    assert!(
+        mrf_channel::take_mrf_repaired_events_for("mrf-bucket").is_empty(),
+        "notice take is destructive"
+    );
 }
 
 /// A journal left behind by a previous process must be replayed into the

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/madmin/src/service_commands.rs

@@ -19,7 +19,6 @@ use hyper::Uri;
 use crate::{trace::TraceType, utils::parse_duration};
 
 #[derive(Debug, Default)]
-#[allow(dead_code)]
 pub struct ServiceTraceOpts {
     s3: bool,
     internal: bool,
@@ -41,7 +40,6 @@ pub struct ServiceTraceOpts {
     threshold: Duration,
 }
 
-#[allow(dead_code)]
 impl ServiceTraceOpts {
     pub fn trace_types(&self) -> TraceType {
         let mut tt = TraceType::default();

crates/notify/examples/base.rs

@@ -15,7 +15,6 @@
 use std::io::IsTerminal;
 use tracing_subscriber::{EnvFilter, fmt, prelude::*, util::SubscriberInitExt};
 
-#[allow(dead_code)]
 fn main() {
     init_logger(LogLevel::Info);
     tracing::info!("Tracing logger initialized with Info level");

crates/s3select-query/src/sql/logical/optimizer.rs

@@ -46,15 +46,6 @@ pub struct DefaultLogicalOptimizer {
     analyzer: AnalyzerRef,
     rules: Vec<Arc<dyn OptimizerRule + Send + Sync>>,
 }
-
-impl DefaultLogicalOptimizer {
-    #[allow(dead_code)]
-    fn with_optimizer_rules(mut self, rules: Vec<Arc<dyn OptimizerRule + Send + Sync>>) -> Self {
-        self.rules = rules;
-        self
-    }
-}
-
 impl Default for DefaultLogicalOptimizer {
     fn default() -> Self {
         let analyzer = Arc::new(DefaultAnalyzer::default());

crates/s3select-query/src/sql/physical/planner.rs

@@ -36,21 +36,9 @@ pub struct DefaultPhysicalPlanner {
     ext_physical_optimizer_rules: Vec<Arc<dyn PhysicalOptimizerRule + Send + Sync>>,
 }
 
-impl DefaultPhysicalPlanner {
+impl DefaultPhysicalPlanner {}
-    #[allow(dead_code)]
-    fn with_physical_transform_rules(mut self, rules: Vec<Arc<dyn ExtensionPlanner + Send + Sync>>) -> Self {
-        self.ext_physical_transform_rules = rules;
-        self
-    }
-}
 
-impl DefaultPhysicalPlanner {
+impl DefaultPhysicalPlanner {}
-    #[allow(dead_code)]
-    fn with_optimizer_rules(mut self, rules: Vec<Arc<dyn PhysicalOptimizerRule + Send + Sync>>) -> Self {
-        self.ext_physical_optimizer_rules = rules;
-        self
-    }
-}
 
 impl Default for DefaultPhysicalPlanner {
     fn default() -> Self {

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> {
+/// How long the scanner caches the runtime tier-name list before re-reading
-    ecstore_get_global_tier_config_mgr().read().await.list_tiers()
+/// 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,
+    Disk, DiskError, DiskInfoOptions, Evaluator, Event, LcEventSrc, ListPathRawOptions, ObjectOpts, ReplicationConfig,
-    ReplicationConfig, ReplicationHealObject, ReplicationQueueAdmission, ReplicationStatusType, STORAGE_FORMAT_FILE,
+    ReplicationHealObject, ReplicationQueueAdmission, ReplicationStatusType, STORAGE_FORMAT_FILE, ScannerDiskExt as _,
-    ScannerDiskExt as _, ScannerLifecycleConfigExt as _, ScannerVersioningConfigExt as _, StorageError, apply_expiry_rule,
+    ScannerLifecycleConfigExt as _, ScannerVersioningConfigExt as _, StorageError, apply_expiry_rule, apply_transition_rule,
-    apply_transition_rule, enqueue_runtime_newer_noncurrent, is_reserved_or_invalid_bucket, list_path_raw, path2_bucket_object,
+    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,
 };
@@ -645,6 +645,32 @@ enum GetSizeFailureAction {
     HealMetadata { object: String },
 }
 
+/// How the corrupt-metadata branch records the repair after attempting an
+/// MRF intent (backlog#1894 axis A).
+#[derive(Debug, PartialEq, Eq)]
+enum CorruptMetadataRecording {
+    /// Intent accepted: the MRF consumer owns the repair (High Metadata
+    /// heal, durable after the journal's group-commit flush), so the
+    /// immediate heal request is skipped — the manager would otherwise book
+    /// two tasks for one target. A pending-ledger entry stays behind as the
+    /// backstop for what the journal cannot cover on its own (a crash inside
+    /// the flush window, or the consumer exhausting its admission attempts);
+    /// the repaired-notice fanout (axis B) drops the entry once the repair
+    /// lands.
+    LedgerOnly,
+    /// Intent rejected (feature disabled, channel uninitialized, or full):
+    /// the historical immediate heal request plus the ledger entry.
+    ImmediateAndLedger,
+}
+
+fn corrupt_metadata_recording(mrf_accepted: bool) -> CorruptMetadataRecording {
+    if mrf_accepted {
+        CorruptMetadataRecording::LedgerOnly
+    } else {
+        CorruptMetadataRecording::ImmediateAndLedger
+    }
+}
+
 fn build_bucket_heal_request(bucket: String, priority: HealChannelPriority) -> HealChannelRequest {
     HealChannelRequest {
         bucket,
@@ -700,6 +726,16 @@ fn pending_scanner_heal_identity(entry: &PendingScannerHeal) -> (u8, &str, Optio
     (kind, entry.bucket.as_str(), entry.object.as_deref(), entry.version_id.as_deref())
 }
 
+/// Decode an MRF repaired-notice version id for ledger matching. A nil UUID
+/// means "no value" per the repo-wide defensive-UUID invariant, so it maps
+/// to `None` and matches unversioned ledger entries only.
+fn mrf_repaired_version_id(version_id: Option<[u8; 16]>) -> Option<String> {
+    version_id
+        .map(uuid::Uuid::from_bytes)
+        .filter(|uuid| !uuid.is_nil())
+        .map(|uuid| uuid.to_string())
+}
+
 fn sort_pending_scanner_heals_for_retry(entries: &mut [PendingScannerHeal]) {
     entries.sort_by(|a, b| {
         a.last_attempt
@@ -934,6 +970,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 +995,8 @@ impl ScannerItem {
             "Scanner lifecycle evaluation started"
         );
 
-        let versioning_config = match BucketVersioningSys::get(&self.bucket).await {
+        // `versioning_config` is resolved once per object by the caller
-            Ok(versioning_config) => versioning_config,
+        // (`get_size`) and handed in; only `prefix_enabled` is consulted here.
-            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()
-            }
-        };
 
         let Some(lifecycle) = self.lifecycle.as_ref() else {
             let mut cumulative_size = 0;
@@ -1402,6 +1426,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);
@@ -1639,6 +1668,32 @@ impl FolderScanner {
         }
     }
 
+    /// Batched variant of [`Self::clear_pending_scanner_heal`] for repaired
+    /// notices (backlog#1894 axis B): one retain pass and one ledger sync
+    /// for the whole notice set, so a mass-recovery first sweep cannot turn
+    /// into thousands of full-table clones on the scan task. Only Object
+    /// entries match — bucket-level heals are never the MRF consumer's work.
+    fn clear_pending_scanner_heals_for_repaired(&mut self, events: &[rustfs_common::mrf_channel::MrfRepairedEvent]) {
+        // Pre-resolve the notice version strings once; each ledger entry then
+        // compares against plain Option<&str>.
+        let targets: Vec<(&str, &str, Option<String>)> = events
+            .iter()
+            .map(|event| (event.bucket.as_ref(), event.object.as_ref(), mrf_repaired_version_id(event.version_id)))
+            .collect();
+        let before = self.new_cache.info.pending_heals.len();
+        self.new_cache.info.pending_heals.retain(|entry| {
+            entry.kind != PendingScannerHealKind::Object
+                || !targets.iter().any(|(bucket, object, version)| {
+                    entry.bucket.as_str() == *bucket
+                        && entry.object.as_deref() == Some(*object)
+                        && entry.version_id.as_deref() == version.as_deref()
+                })
+        });
+        if self.new_cache.info.pending_heals.len() != before {
+            self.sync_pending_heals();
+        }
+    }
+
     fn record_pending_scanner_heal(
         &mut self,
         kind: PendingScannerHealKind,
@@ -1977,6 +2032,14 @@ impl FolderScanner {
         }
 
         let bucket = self.new_cache.info.name.clone();
+        // Backlog#1894 axis B: repairs the MRF consumer landed hand the
+        // manager the heal task, so the matching pending-ledger entries are
+        // retried nowhere — drop them here. Best-effort: a lost notice just
+        // leaves the entry to expire through its own attempts/age limits.
+        let repaired = rustfs_common::mrf_channel::take_mrf_repaired_events_for(&bucket);
+        if !repaired.is_empty() {
+            self.clear_pending_scanner_heals_for_repaired(&repaired);
+        }
         for pending in pending_scanner_heal_retry_candidates(&self.new_cache.info.pending_heals, &bucket) {
             if !self.should_heal().await {
                 break;
@@ -2441,29 +2504,46 @@ impl FolderScanner {
                         }
 
                         if let GetSizeFailureAction::HealMetadata { object } = failure_action {
-                            // MRF journal intent: durable High-priority Metadata
+                            // Single-flight (backlog#1894 axis A) — the
-                            // heal across restarts (HS-01); the scanner heal
+                            // recording mode and its guarantees are pinned by
-                            // request below stays as the immediate path.
+                            // corrupt_metadata_recording below.
-                            rustfs_common::mrf_channel::try_send_mrf_intent(
+                            let mrf_accepted = rustfs_common::mrf_channel::try_send_mrf_intent(
                                 rustfs_common::mrf_channel::MrfKind::MetadataCorruption,
                                 &item.bucket,
                                 &object,
                                 None,
                             );
-                            self.send_required_scanner_heal_request(
+                            match corrupt_metadata_recording(mrf_accepted) {
-                                PendingScannerHealKind::Object,
+                                CorruptMetadataRecording::LedgerOnly => {
-                                item.bucket.clone(),
+                                    // Recorded as Full (retry-later): admission
-                                Some(object.clone()),
+                                    // for this target happens in the MRF
-                                None,
+                                    // consumer, not in the manager's queue here.
-                                build_object_heal_request(
+                                    self.update_pending_scanner_heal_after_admission(
-                                    item.bucket.clone(),
+                                        PendingScannerHealKind::Object,
-                                    object.clone(),
+                                        &item.bucket,
-                                    None,
+                                        Some(&object),
-                                    self.scan_mode,
+                                        None,
-                                    HealChannelPriority::High,
+                                        self.scan_mode,
-                                ),
+                                        HealAdmissionResult::Full,
-                            )
+                                    );
-                            .await?;
+                                }
+                                CorruptMetadataRecording::ImmediateAndLedger => {
+                                    self.send_required_scanner_heal_request(
+                                        PendingScannerHealKind::Object,
+                                        item.bucket.clone(),
+                                        Some(object.clone()),
+                                        None,
+                                        build_object_heal_request(
+                                            item.bucket.clone(),
+                                            object.clone(),
+                                            None,
+                                            self.scan_mode,
+                                            HealChannelPriority::High,
+                                        ),
+                                    )
+                                    .await?;
+                                }
+                            }
                         }
 
                         timer.sleep().await;
@@ -3358,6 +3438,17 @@ mod tests {
         assert_eq!(EVENT_SCANNER_BIG_PREFIX, EventName::ScannerBigPrefix.to_string());
     }
 
+    /// Single-flight decision for the corrupt-metadata branch (backlog#1894
+    /// axis A): an accepted MRF intent must drop the immediate heal request
+    /// (the consumer files one; the manager would double-book) while a
+    /// rejected one must keep it — in both cases a ledger entry remains, so
+    /// the backstop survives regardless of delivery.
+    #[test]
+    fn corrupt_metadata_recording_maps_delivery_to_backstop() {
+        assert_eq!(corrupt_metadata_recording(true), CorruptMetadataRecording::LedgerOnly);
+        assert_eq!(corrupt_metadata_recording(false), CorruptMetadataRecording::ImmediateAndLedger);
+    }
+
     fn cooldown_map_len() -> usize {
         SCANNER_ALERT_EMISSION_COOLDOWN
             .lock()
@@ -4331,6 +4422,86 @@ mod tests {
         }
     }
 
+    /// The nil-UUID branch of the defensive-UUID invariant: a nil version in
+    /// a repaired notice means "no value" and must match unversioned ledger
+    /// entries only.
+    #[test]
+    fn test_mrf_repaired_version_id_maps_nil_to_none() {
+        assert_eq!(mrf_repaired_version_id(None), None);
+        assert_eq!(mrf_repaired_version_id(Some([0u8; 16])), None);
+        let uuid = Uuid::new_v4();
+        assert_eq!(mrf_repaired_version_id(Some(*uuid.as_bytes())), Some(uuid.to_string()));
+    }
+
+    /// Full wiring of backlog#1894 axis B: notes taken for the scanned bucket
+    /// clear exactly the matching Object ledger entries — bucket-level
+    /// entries, other buckets' entries, and version-mismatched entries
+    /// survive; a real (non-nil) version matches only the same version.
+    #[tokio::test]
+    async fn test_mrf_repaired_notices_clear_matching_ledger_entries() {
+        use rustfs_common::mrf_channel::note_mrf_repaired;
+
+        let (mut scanner, temp_dir) = build_test_scanner().await;
+        let _guard = TestGuard::new(u64::MAX, usize::MAX, &mut scanner, temp_dir);
+        scanner.new_cache.info.name = "bucket".to_string();
+        scanner.update_cache.info.name = "bucket".to_string();
+        scanner.heal_object_select = 1;
+
+        let version = Uuid::new_v4().to_string();
+        scanner.new_cache.info.pending_heals = vec![
+            pending_heal(PendingScannerHealKind::Object, "bucket", Some("object-a"), None, 1, 1),
+            pending_heal(PendingScannerHealKind::Object, "bucket", Some("object-b"), Some(&version), 1, 1),
+            pending_heal(PendingScannerHealKind::Object, "bucket", Some("object-c"), None, 1, 1),
+            pending_heal(
+                PendingScannerHealKind::Object,
+                "bucket",
+                Some("object-c"),
+                Some("00000000-0000-0000-0000-000000000001"),
+                1,
+                1,
+            ),
+            pending_heal(PendingScannerHealKind::Bucket, "bucket", None, None, 1, 1),
+            pending_heal(PendingScannerHealKind::Object, "other-bucket", Some("object-a"), None, 1, 1),
+        ];
+
+        note_mrf_repaired("bucket", "object-a", None);
+        note_mrf_repaired("bucket", "object-b", Some(*Uuid::parse_str(&version).unwrap().as_bytes()));
+        // A nil-UUID notice for object-c means "no value": it clears the
+        // unversioned entry but must not touch the versioned one.
+        note_mrf_repaired("bucket", "object-c", Some([0u8; 16]));
+        // A notice for a target the ledger does not track must be a no-op.
+        note_mrf_repaired("bucket", "object-untracked", None);
+
+        scanner
+            .retry_pending_scanner_heals()
+            .await
+            .expect("retry pass should succeed");
+
+        let survivors: Vec<(PendingScannerHealKind, &str, Option<&str>, Option<&str>)> = scanner
+            .new_cache
+            .info
+            .pending_heals
+            .iter()
+            .map(|entry| (entry.kind, entry.bucket.as_str(), entry.object.as_deref(), entry.version_id.as_deref()))
+            .collect();
+        // Cleared: object-a (no version), object-b (exact version match), and
+        // object-c's unversioned entry (the nil branch matched no-version
+        // only — the versioned object-c entry survives).
+        assert_eq!(
+            survivors,
+            vec![
+                (
+                    PendingScannerHealKind::Object,
+                    "bucket",
+                    Some("object-c"),
+                    Some("00000000-0000-0000-0000-000000000001")
+                ),
+                (PendingScannerHealKind::Bucket, "bucket", None, None),
+                (PendingScannerHealKind::Object, "other-bucket", Some("object-a"), None),
+            ]
+        );
+    }
+
     #[test]
     fn test_pending_heal_reconstructs_bucket_request() {
         let pending = pending_heal(PendingScannerHealKind::Bucket, "bucket", None, None, 1, 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)
+        // Single versioning lookup per object, shared with `apply_actions`
-            .await
+        // (which used to query it a second time). On failure keep the
-            .map(|v| v.versioned(&item.object_path()))
+        // historical fallback: default configuration (versioned = false) plus
-            .unwrap_or(false);
+        // 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;
+        // Tier names come from the process-wide TTL cache; seeding from them
-
+        // replaces the per-object clone of every full TierConfig.
-        for tier in tiers.iter() {
+        let tier_names = runtime_tier_names().await;
-            size_summary.tier_stats.insert(tier.name.clone(), TierStats::default());
+        size_summary.tier_stats = tier_stats_template(&tier_names);
-        }
-        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());
-        }
 
         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,
+        EcstoreSetDisks, EcstoreStorageError, EcstoreStore, EcstoreVersioningApi, ScannerReplicationHealObject,
-        ScannerReplicationHealObject, ScannerReplicationHealResult, ScannerReplicationQueueAdmission, ecstore_apply_expiry_rule,
+        ScannerReplicationHealResult, ScannerReplicationQueueAdmission, ecstore_apply_expiry_rule, ecstore_apply_transition_rule,
-        ecstore_apply_transition_rule, ecstore_expiry_state_handle, ecstore_get_global_tier_config_mgr,
+        ecstore_expiry_state_handle, ecstore_get_global_tier_config_mgr, ecstore_get_lifecycle_config,
-        ecstore_get_lifecycle_config, ecstore_get_object_lock_config, ecstore_get_replication_config,
+        ecstore_get_object_lock_config, ecstore_get_replication_config, ecstore_invalidate_admin_data_usage_snapshot_cache,
-        ecstore_invalidate_admin_data_usage_snapshot_cache, ecstore_invalidate_data_usage_snapshot_cache, ecstore_is_erasure,
+        ecstore_invalidate_data_usage_snapshot_cache, ecstore_is_erasure, ecstore_is_erasure_sd,
-        ecstore_is_erasure_sd, ecstore_is_reserved_or_invalid_bucket, ecstore_list_path_raw,
+        ecstore_is_reserved_or_invalid_bucket, ecstore_list_path_raw, ecstore_object_opts_from_object_info,
-        ecstore_object_opts_from_object_info, ecstore_path2_bucket_object, ecstore_path2_bucket_object_with_base_path,
+        ecstore_path2_bucket_object, ecstore_path2_bucket_object_with_base_path, ecstore_read_config,
-        ecstore_read_config, ecstore_replace_bucket_usage_memory_from_info, ecstore_resolve_object_store_handle,
+        ecstore_replace_bucket_usage_memory_from_info, ecstore_resolve_object_store_handle, ecstore_save_config,
-        ecstore_save_config, ecstore_send_event, scanner_replication_config_for_lifecycle_eval,
+        ecstore_send_event, scanner_replication_config_for_lifecycle_eval,
     };
 
     #[cfg(test)]

crates/signer/src/request_signature_streaming.rs

@@ -22,7 +22,7 @@ use s3s::Body;
 
 const STREAMING_SIGN_ALGORITHM: &str = "STREAMING-AWS4-HMAC-SHA256-PAYLOAD";
 const STREAMING_SIGN_TRAILER_ALGORITHM: &str = "STREAMING-AWS4-HMAC-SHA256-PAYLOAD-TRAILER";
-const STREAMING_PAYLOAD_HDR: &str = "AWS4-HMAC-SHA256-PAYLOAD";
+const _STREAMING_PAYLOAD_HDR: &str = "AWS4-HMAC-SHA256-PAYLOAD";
 const _STREAMING_TRAILER_HDR: &str = "AWS4-HMAC-SHA256-TRAILER";
 const _PAYLOAD_CHUNK_SIZE: i64 = 64 * 1024;
 const _CHUNK_SIGCONST_LEN: i64 = 17;
@@ -51,15 +51,14 @@ fn streaming_fail(request: request::Request<Body>, error: SignV4Error) -> Stream
     Err(Box::new(StreamingSignFailure { request, error }))
 }
 
-#[allow(dead_code)]
+fn _try_build_chunk_string_to_sign(
-fn try_build_chunk_string_to_sign(
     t: OffsetDateTime,
     region: &str,
     previous_sig: &str,
     chunk_check_sum: &str,
 ) -> Result<String, SignV4Error> {
     let mut string_to_sign_parts = <Vec<String>>::new();
-    string_to_sign_parts.push(STREAMING_PAYLOAD_HDR.to_string());
+    string_to_sign_parts.push(_STREAMING_PAYLOAD_HDR.to_string());
     let format = format_description!("[year][month][day]T[hour][minute][second]Z");
     string_to_sign_parts.push(
         t.format(&format)
@@ -79,7 +78,7 @@ fn _try_build_chunk_signature(
     previous_signature: &str,
     secret_access_key: &str,
 ) -> Result<String, SignV4Error> {
-    let chunk_string_to_sign = try_build_chunk_string_to_sign(req_time, region, previous_signature, chunk_check_sum)?;
+    let chunk_string_to_sign = _try_build_chunk_string_to_sign(req_time, region, previous_signature, chunk_check_sum)?;
     let signing_key = get_signing_key(secret_access_key, region, req_time, SERVICE_TYPE_S3);
     Ok(get_signature(signing_key, &chunk_string_to_sign))
 }

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) =
+    authorize_admin_request(req, vec![Action::AdminAction(action)]).await?;
-        check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
+    Ok(())
-    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
 }
 
 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;
-use crate::server::{ADMIN_PREFIX, RemoteAddr};
 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) =
+    authorize_admin_request(req, vec![Action::AdminAction(action)]).await?;
-        check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
+    Ok(())
-    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
 }
 
 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,
+    ADMIN_PREFIX, is_notify_module_enabled, refresh_notify_module_enabled, refresh_persisted_module_switches_from_store,
-    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) =
+    authorize_admin_request(req, vec![Action::AdminAction(action)]).await?;
-        check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
+    Ok(())
-    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
 }
 
 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) =
+    authorize_admin_request(req, vec![Action::AdminAction(AdminAction::GetMetricsAction)]).await?;
-        check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
+    Ok(())
-    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
 }
 
 #[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) =
+    authorize_admin_request(req, vec![Action::AdminAction(action)]).await?;
-        check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
+    Ok(())
-
-    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
 }
 
 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::admin::{auth::authorize_admin_request, router::Operation};
-use crate::auth::{check_key_valid, get_session_token};
-use crate::server::RemoteAddr;
 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) =
+    authorize_admin_request(req, vec![Action::AdminAction(AdminAction::ProfilingAdminAction)]).await?;
-        check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
+    Ok(())
-    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
 }
 
 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;
-use crate::server::{ADMIN_PREFIX, RemoteAddr};
 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) =
+    authorize_admin_request(req, vec![Action::AdminAction(action)]).await?;
-        check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
+    Ok(())
-    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
 }
 
 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) =
+    authorize_admin_request(req, vec![Action::AdminAction(AdminAction::ServerInfoAdminAction)]).await
-        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)
 }
 
 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;
-use crate::server::{ADMIN_PREFIX, RemoteAddr};
 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 {
+        // The shared gate reports the same `InvalidRequest` "get cred failed" this
-            return Err(s3_error!(InvalidRequest, "get cred failed"));
+        // 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 (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?;
 
         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/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