chore: adjudicate 26 bare dead_code allows across five crates

Remove every bare `#[allow(dead_code)]` in io-core, object-capacity, targets, rio, and scanner. Each allow was stripped first and clippy was then asked which ones the compiler actually missed, so the verdicts rest on the diagnostic rather than on inspection.

23 were inert: they sat on `pub fn`s inside `pub mod`s, where `dead_code` does not apply, or on scanner integration-test helpers that the tests in the same file do call.

The remaining 3 are in rio's private `compress_index` module and the code behind them is deleted rather than annotated. `remove_index_headers` is dead and also wrong — after skipping the 4-byte chunk header it matches against `S2_INDEX_TRAILER` where `S2_INDEX_HEADER` sits, so it returns `None` for every well-formed index; rio-v2 carries the correct equivalent that is actually in use. `restore_index_headers` is its unreachable counterpart, likewise duplicated live in rio-v2. `Index::reset` is a private method with no caller.

Refs backlog#1823

Cargo.lock

@@ -9825,14 +9825,19 @@ name = "rustfs-madmin"
 version = "1.0.0-rc.2"
 dependencies = [
  "hotpath",
+ "http 1.5.0",
  "humantime",
  "hyper",
  "jiff",
+ "reqwest",
  "rmp-serde",
+ "rustfs-signer",
+ "s3s",
  "serde",
  "serde_json",
  "sysinfo",
  "time",
+ "tokio",
 ]
 
 [[package]]

_typos.toml

@@ -40,6 +40,7 @@ mak = "mak"
 gae = "gae"
 GAE = "GAE"
 thr = "thr"
+mis = "mis"
 # s3-tests original test names (cannot be changed)
 nonexisted = "nonexisted"
 consts = "consts"

crates/data-usage/src/data_usage.rs

@@ -870,6 +870,157 @@ pub struct DataUsageCacheInfo {
     pub snapshot_complete: bool,
 }
 
+/// Prefix-level usage over a raw entry map — the shared core behind
+/// [`DataUsageCache::prefix_usage`], usable by any cache-shaped reader (the
+/// scanner's writer-side cache has the same map type).
+///
+/// Cache keys are cleaned literal paths (`bucket/pre/fix`), so sub-prefix
+/// names come straight off the child keys — no reverse mapping exists or is
+/// needed. A compacted prefix carries its aggregate but no children, which
+/// the `compacted` flag reports so callers can say why the breakdown is
+/// empty. `truncated` is set when the breakdown exceeded `max_entries` and
+/// was cut (largest first).
+pub fn prefix_usage_in_cache(
+    cache: &HashMap<String, DataUsageEntry>,
+    bucket: &str,
+    prefix: &str,
+    max_entries: usize,
+) -> Option<PrefixUsageQuery> {
+    let prefix = prefix.trim_matches('/');
+    let root = if prefix.is_empty() {
+        bucket.to_string()
+    } else {
+        format!("{bucket}/{prefix}")
+    };
+    let entry = cache.get(&hash_path(&root).key())?.clone();
+
+    let usage = PrefixUsageSummary::from_entry(&flatten_entry(cache, &entry, 0)?);
+
+    let child_prefix = format!("{root}/");
+    let mut sub_prefixes: Vec<PrefixUsageEntry> = entry
+        .children
+        .iter()
+        .filter_map(|child_key| {
+            let child = cache.get(child_key)?;
+            let child_flat = flatten_entry(cache, child, 1)?;
+            // Child keys are literal `bucket/pre/name` paths; a trailing
+            // slash marks a directory object and is display-only here.
+            let name = child_key
+                .strip_prefix(child_prefix.as_str())
+                .unwrap_or(child_key.as_str())
+                .trim_end_matches('/')
+                .to_string();
+            Some(PrefixUsageEntry {
+                prefix: name,
+                usage: PrefixUsageSummary::from_entry(&child_flat),
+            })
+        })
+        .collect();
+    sub_prefixes.sort_by(|left, right| {
+        right
+            .usage
+            .size
+            .cmp(&left.usage.size)
+            .then_with(|| left.prefix.cmp(&right.prefix))
+    });
+    let truncated = sub_prefixes.len() > max_entries;
+    sub_prefixes.truncate(max_entries);
+
+    Some(PrefixUsageQuery {
+        usage,
+        compacted: entry.compacted,
+        truncated,
+        sub_prefixes,
+    })
+}
+
+/// Maximum subtree depth [`flatten_entry`] will walk before declaring the
+/// cache corrupt — the same bound the scanner's checked flatten uses.
+const PREFIX_USAGE_MAX_DEPTH: usize = 1024;
+
+/// Flatten one entry's subtree into an aggregate: the free-function twin of
+/// [`DataUsageCache::flatten`], carrying the scanner checked-flatten
+/// hardening so a corrupt cache (cycles, over-deep trees, overflowing
+/// counters) yields `None` instead of unbounded recursion or wrapped totals.
+fn flatten_entry(cache: &HashMap<String, DataUsageEntry>, root: &DataUsageEntry, depth: usize) -> Option<DataUsageEntry> {
+    if depth > PREFIX_USAGE_MAX_DEPTH {
+        return None;
+    }
+    let mut flattened = DataUsageEntry::default();
+    if !flattened.checked_merge(root) {
+        return None;
+    }
+    flattened.compacted = root.compacted;
+    // The root itself is not pre-seeded: it is merged above, and a corrupt
+    // child edge pointing back at the root's own key is still terminated by
+    // the visited set on first encounter.
+    let mut visited: HashSet<&str> = HashSet::new();
+    let mut pending: Vec<(&String, usize)> = root.children.iter().map(|child| (child, depth + 1)).collect();
+    while let Some((key, child_depth)) = pending.pop() {
+        if child_depth > PREFIX_USAGE_MAX_DEPTH || !visited.insert(key.as_str()) {
+            return None;
+        }
+        let entry = cache.get(key)?;
+        if !flattened.checked_merge(entry) {
+            return None;
+        }
+        pending.extend(entry.children.iter().map(|child| (child, child_depth + 1)));
+    }
+    flattened.children.clear();
+    Some(flattened)
+}
+
+/// Flattened counters of one prefix subtree, as returned by
+/// [`DataUsageCache::prefix_usage`].
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, serde::Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct PrefixUsageSummary {
+    pub size: u64,
+    pub objects: u64,
+    pub versions: u64,
+    pub delete_markers: u64,
+}
+
+impl PrefixUsageSummary {
+    fn from_entry(entry: &DataUsageEntry) -> Self {
+        Self {
+            size: entry.size as u64,
+            objects: entry.objects as u64,
+            versions: entry.versions as u64,
+            delete_markers: entry.delete_markers as u64,
+        }
+    }
+
+    /// Add another set's counters into this one (entries are partitioned by
+    /// set, so per-set results sum).
+    pub fn merge(&mut self, other: &Self) {
+        self.size = self.size.saturating_add(other.size);
+        self.objects = self.objects.saturating_add(other.objects);
+        self.versions = self.versions.saturating_add(other.versions);
+        self.delete_markers = self.delete_markers.saturating_add(other.delete_markers);
+    }
+}
+
+/// One first-level sub-prefix row of a [`PrefixUsageQuery`].
+#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize)]
+pub struct PrefixUsageEntry {
+    pub prefix: String,
+    pub usage: PrefixUsageSummary,
+}
+
+/// Result of [`DataUsageCache::prefix_usage`].
+#[derive(Clone, Debug, Default, PartialEq, Eq, serde::Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct PrefixUsageQuery {
+    pub usage: PrefixUsageSummary,
+    /// The prefix entry was compacted by the scanner: its aggregate is valid
+    /// but no sub-prefix breakdown exists on disk.
+    pub compacted: bool,
+    /// The breakdown had more entries than `max_entries`; the largest remain.
+    pub truncated: bool,
+    pub sub_prefixes: Vec<PrefixUsageEntry>,
+}
+
 /// Read-only projection of a scanner-written `.usage-cache.bin` file.
 ///
 /// The scanner-side `DataUsageCache` (`crates/scanner/src/data_usage_define.rs`)
@@ -997,6 +1148,21 @@ impl DataUsageCache {
         }
     }
 
+    /// Prefix-level usage for one bucket subtree, plus the one-level
+    /// breakdown below it (rustfs/backlog#1872, MinIO
+    /// `loadPrefixUsageFromBackend` parity and beyond: arbitrary prefixes and
+    /// full counters instead of first-level sizes only).
+    ///
+    /// Cache keys are cleaned literal paths (`bucket/pre/fix`), so sub-prefix
+    /// names come straight off the child keys — no reverse mapping exists or
+    /// is needed. A compacted prefix carries its aggregate but no children,
+    /// which the `compacted` flag reports so callers can say why the
+    /// breakdown is empty. `truncated` is set when the breakdown exceeded
+    /// `max_entries` and was cut (largest first).
+    pub fn prefix_usage(&self, bucket: &str, prefix: &str, max_entries: usize) -> Option<PrefixUsageQuery> {
+        prefix_usage_in_cache(&self.cache, bucket, prefix, max_entries)
+    }
+
     pub fn force_compact(&mut self, limit: usize) {
         if self.cache.len() < limit {
             return;
@@ -1898,6 +2064,126 @@ mod tests {
         );
     }
 
+    /// Build a cache shaped like `bucket/{a,b/{c,d}},bucket/loose` with
+    /// distinct counters so aggregation is observable.
+    fn prefix_usage_fixture_cache() -> DataUsageCache {
+        let mut cache = DataUsageCache::default();
+        let mut insert = |path: &str, parent: &str, size: usize, objects: usize, versions: usize, delete_markers: usize| {
+            cache.replace(
+                path,
+                parent,
+                DataUsageEntry {
+                    size,
+                    objects,
+                    versions,
+                    delete_markers,
+                    ..Default::default()
+                },
+            );
+        };
+        insert("bucket", "", 0, 0, 0, 0);
+        insert("bucket/a", "bucket", 100, 1, 1, 0);
+        insert("bucket/b", "bucket", 0, 0, 0, 0);
+        insert("bucket/b/c", "bucket/b", 200, 2, 2, 1);
+        insert("bucket/b/d", "bucket/b", 40, 1, 3, 0);
+        insert("bucket/loose", "bucket", 10, 1, 1, 1);
+        cache
+    }
+
+    #[test]
+    fn prefix_usage_aggregates_bucket_root_and_one_level_below() {
+        let cache = prefix_usage_fixture_cache();
+
+        let root = cache
+            .prefix_usage("bucket", "", 100)
+            .expect("root query must find the bucket entry");
+        assert_eq!(root.usage.size, 350, "root aggregate flattens the whole subtree");
+        assert_eq!(root.usage.objects, 5);
+        assert_eq!(root.usage.versions, 7);
+        assert_eq!(root.usage.delete_markers, 2);
+        assert!(!root.compacted);
+        assert!(!root.truncated);
+        // Breakdown is one level: b (240) before a (100) before loose (10),
+        // each flattened to its own subtree total.
+        let names: Vec<(&str, u64)> = root
+            .sub_prefixes
+            .iter()
+            .map(|entry| (entry.prefix.as_str(), entry.usage.size))
+            .collect();
+        assert_eq!(names, vec![("b", 240), ("a", 100), ("loose", 10)]);
+    }
+
+    #[test]
+    fn prefix_usage_drills_into_arbitrary_prefixes() {
+        let cache = prefix_usage_fixture_cache();
+
+        let b = cache.prefix_usage("bucket", "b", 100).expect("nested prefix must resolve");
+        assert_eq!(b.usage.size, 240);
+        assert_eq!(b.usage.versions, 5);
+        let names: Vec<&str> = b.sub_prefixes.iter().map(|entry| entry.prefix.as_str()).collect();
+        assert_eq!(names, vec!["c", "d"]);
+
+        // Prefix slashes are normalized away.
+        let slashed = cache.prefix_usage("bucket", "/b/", 100).expect("slash-insensitive lookup");
+        assert_eq!(slashed.usage.size, 240);
+
+        assert!(cache.prefix_usage("bucket", "absent", 100).is_none(), "unknown prefix must be a miss");
+        assert!(cache.prefix_usage("other", "", 100).is_none(), "unknown bucket must be a miss");
+    }
+
+    #[test]
+    fn prefix_usage_reports_and_respects_truncation() {
+        let cache = prefix_usage_fixture_cache();
+        let capped = cache.prefix_usage("bucket", "", 2).expect("root query");
+        assert!(capped.truncated, "three children capped to two must flag truncation");
+        let names: Vec<&str> = capped.sub_prefixes.iter().map(|entry| entry.prefix.as_str()).collect();
+        assert_eq!(names, vec!["b", "a"], "largest prefixes survive the cut");
+    }
+
+    #[test]
+    fn prefix_usage_marks_compacted_entries() {
+        let mut cache = DataUsageCache::default();
+        cache.replace(
+            "bucket",
+            "",
+            DataUsageEntry {
+                size: 999,
+                objects: 9,
+                compacted: true,
+                ..Default::default()
+            },
+        );
+
+        let compacted = cache.prefix_usage("bucket", "", 100).expect("compacted root resolves");
+        assert!(compacted.compacted, "compaction must be visible to callers");
+        assert_eq!(compacted.usage.size, 999);
+        assert!(compacted.sub_prefixes.is_empty(), "a compacted entry carries no children");
+    }
+
+    #[test]
+    fn prefix_usage_rejects_cyclic_and_dangling_caches() {
+        // A self-referencing child (corrupt cache) must yield a miss for the
+        // whole query, not unbounded recursion.
+        let mut cache = prefix_usage_fixture_cache();
+        if let Some(entry) = cache.cache.get_mut("bucket/b") {
+            entry.children.insert("bucket/b".to_string());
+        }
+        assert!(cache.prefix_usage("bucket", "b", 100).is_none(), "a cyclic subtree must be rejected");
+        // The unaffected sibling still answers.
+        assert!(cache.prefix_usage("bucket", "a", 100).is_some());
+
+        // A child key with no entry (dangling link) is rejected rather than
+        // silently dropped: half a tree would under-report usage.
+        let mut dangling = prefix_usage_fixture_cache();
+        if let Some(entry) = dangling.cache.get_mut("bucket/b") {
+            entry.children.insert("bucket/b/ghost".to_string());
+        }
+        assert!(
+            dangling.prefix_usage("bucket", "b", 100).is_none(),
+            "a dangling child link must be rejected"
+        );
+    }
+
     #[test]
     fn hash_path_uses_portable_slash_semantics() {
         for (input, expected) in [

crates/ecstore/src/api/mod.rs

@@ -373,8 +373,8 @@ pub mod error {
 
 pub mod erasure {
     pub use crate::erasure::coding::{
-        BitrotReader, BitrotWriter, BitrotWriterWrapper, CustomWriter, Erasure, ErasureConstructionError, ReedSolomonEncoder,
-        calc_shard_size, calc_shard_size_legacy,
+        BitrotReader, BitrotSelfTestError, BitrotWriter, BitrotWriterWrapper, CustomWriter, Erasure, ErasureConstructionError,
+        ReedSolomonEncoder, bitrot_self_test, calc_shard_size, calc_shard_size_legacy,
     };
 }
 

crates/ecstore/src/erasure/coding/bitrot.rs

@@ -820,10 +820,263 @@ impl BitrotWriterWrapper {
     }
 }
 
+// --- startup bitrot self-test (rustfs/backlog#1873, MinIO bitrotSelfTest parity) ---
+//
+// A broken hash implementation (bad SIMD feature combination, platform drift, a
+// key-handling regression) fails silently: every shard reads back "corrupt",
+// heal rewrites data that was fine, and cross-platform clusters disagree about
+// which copy is healthy. The self-test below pins the algorithms the moment a
+// process starts, so a drifted build announces itself instead of quietly
+// rewriting objects. See docs/rustfs-heal-scanner-vs-minio-comprehensive-
+// analysis-2026-08-16.md §6 HS-11.
+
+/// Length of the deterministic self-test payload.
+pub const BITROT_SELF_TEST_PAYLOAD_LEN: usize = 4096;
+
+/// Known-answer digest of [`bitrot_self_test_payload`] under `HighwayHash256S`
+/// (the production default). Pinned so any platform or build where the
+/// implementation drifts fails startup instead of miss-hashing shards.
+const BITROT_SELF_TEST_KAT_HIGHWAY_HASH256S: [u8; 32] = [
+    0xb9, 0x32, 0xa2, 0xaa, 0x4a, 0xb7, 0x33, 0x6a, 0xa3, 0xca, 0x7e, 0x61, 0x9d, 0x86, 0x52, 0x14, 0x6e, 0x7f, 0xd8, 0x9e, 0xea,
+    0x08, 0xd9, 0x8c, 0x33, 0x85, 0x87, 0x19, 0x30, 0xd6, 0xed, 0x06,
+];
+
+/// Known-answer digest of the same payload under `HighwayHash256SLegacy`.
+const BITROT_SELF_TEST_KAT_HIGHWAY_HASH256S_LEGACY: [u8; 32] = [
+    0x98, 0x24, 0x71, 0x4f, 0x16, 0xbb, 0x48, 0x39, 0xed, 0x68, 0xfa, 0x63, 0x5e, 0xd9, 0x07, 0x61, 0xdf, 0x0a, 0xff, 0xcf, 0x7d,
+    0x8c, 0xa8, 0xc7, 0xc0, 0xb6, 0x6f, 0x05, 0xdb, 0xda, 0x5a, 0x22,
+];
+
+/// FIPS 180-2 test vector: SHA-256 of the ASCII string "abc". Unlike the
+/// Highway digests above this one is externally verifiable, so it guards the
+/// whole `HashAlgorithm` plumbing even for readers who distrust pinned
+/// self-computed constants.
+const BITROT_SELF_TEST_KAT_SHA256_ABC: [u8; 32] = [
+    0xba, 0x78, 0x16, 0xbf, 0x8f, 0x01, 0xcf, 0xea, 0x41, 0x41, 0x40, 0xde, 0x5d, 0xae, 0x22, 0x23, 0xb0, 0x03, 0x61, 0xa3, 0x96,
+    0x17, 0x7a, 0x9c, 0xb4, 0x10, 0xff, 0x61, 0xf2, 0x00, 0x15, 0xad,
+];
+
+/// Deterministic self-test payload: xorshift64* from a fixed seed, so every
+/// platform and every run hashes the same 4096 bytes.
+fn bitrot_self_test_payload() -> [u8; BITROT_SELF_TEST_PAYLOAD_LEN] {
+    let mut state = 0x9E37_79B9_7F4A_7C15u64;
+    let mut payload = [0u8; BITROT_SELF_TEST_PAYLOAD_LEN];
+    for byte in payload.iter_mut() {
+        state ^= state >> 12;
+        state ^= state << 25;
+        state ^= state >> 27;
+        *byte = state.wrapping_mul(0x2545_F491_4F6C_DD1D) as u8;
+    }
+    payload
+}
+
+/// Why a bitrot self-test failed.
+#[derive(Debug)]
+pub enum BitrotSelfTestError {
+    /// A known-answer digest mismatched the pinned constant.
+    KnownAnswerMismatch {
+        algorithm: &'static str,
+        got: String,
+        want: String,
+    },
+    /// A freshly encoded shard failed `bitrot_verify`.
+    RoundtripVerify { algorithm: &'static str, detail: String },
+    /// A verified roundtrip read back different bytes than were written.
+    RoundtripReadback { algorithm: &'static str },
+    /// A deliberately tampered shard was not rejected by `bitrot_verify`.
+    TamperNotRejected {
+        algorithm: &'static str,
+        tampered: &'static str,
+    },
+}
+
+impl std::fmt::Display for BitrotSelfTestError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::KnownAnswerMismatch { algorithm, got, want } => {
+                write!(f, "known-answer mismatch for {algorithm}: got {got}, want {want}")
+            }
+            Self::RoundtripVerify { algorithm, detail } => write!(f, "{algorithm} roundtrip shard failed verification: {detail}"),
+            Self::RoundtripReadback { algorithm } => write!(f, "{algorithm} roundtrip read back different bytes"),
+            Self::TamperNotRejected { algorithm, tampered } => {
+                write!(f, "{algorithm} tampered shard ({tampered}) was not rejected")
+            }
+        }
+    }
+}
+
+impl std::error::Error for BitrotSelfTestError {}
+
+fn self_test_hex(bytes: &[u8]) -> String {
+    rustfs_utils::hex(bytes)
+}
+
+// (kept as a named one-liner so every KAT failure site reads the same; the
+// underlying formatter is the shared `rustfs_utils::hex`)
+
+/// Compare a digest against its pinned constant. Split out so a test can drive
+/// it with a wrong constant and prove the mismatch path fires.
+fn bitrot_kat_check(
+    algorithm: &'static str,
+    algo: &HashAlgorithm,
+    payload: &[u8],
+    expected: &[u8; 32],
+) -> Result<(), BitrotSelfTestError> {
+    let digest = algo.hash_encode(payload);
+    let digest = digest.as_ref();
+    if digest.len() != expected.len() || digest != expected.as_slice() {
+        return Err(BitrotSelfTestError::KnownAnswerMismatch {
+            algorithm,
+            got: self_test_hex(digest),
+            want: self_test_hex(expected),
+        });
+    }
+    Ok(())
+}
+
+/// Encode `payload` with `shard_size` blocks, verify it end to end, and read
+/// every block back through `BitrotReader` comparing bytes.
+async fn bitrot_roundtrip_check(
+    algorithm: &'static str,
+    algo: HashAlgorithm,
+    payload: &[u8],
+    shard_size: usize,
+) -> Result<(), BitrotSelfTestError> {
+    let mut writer = BitrotWriter::new(std::io::Cursor::new(Vec::<u8>::new()), shard_size, algo.clone());
+    for chunk in payload.chunks(shard_size) {
+        writer
+            .write(chunk)
+            .await
+            .map_err(|err| BitrotSelfTestError::RoundtripVerify {
+                algorithm,
+                detail: format!("encode failed: {err}"),
+            })?;
+    }
+    let encoded = writer.into_inner().into_inner();
+
+    let on_disk = bitrot_shard_file_size(payload.len(), shard_size, algo.clone());
+    if encoded.len() != on_disk {
+        return Err(BitrotSelfTestError::RoundtripVerify {
+            algorithm,
+            detail: format!("encoded {} bytes, size formula says {on_disk}", encoded.len()),
+        });
+    }
+    bitrot_verify(std::io::Cursor::new(encoded.clone()), on_disk, payload.len(), algo.clone(), shard_size)
+        .await
+        .map_err(|err| BitrotSelfTestError::RoundtripVerify {
+            algorithm,
+            detail: err.to_string(),
+        })?;
+
+    let mut reader = BitrotReader::new(std::io::Cursor::new(encoded), shard_size, algo, false);
+    let mut offset = 0usize;
+    while offset < payload.len() {
+        let want = shard_size.min(payload.len() - offset);
+        let mut buf = vec![0u8; want];
+        let read = reader
+            .read(&mut buf)
+            .await
+            .map_err(|err| BitrotSelfTestError::RoundtripVerify {
+                algorithm,
+                detail: format!("read back failed at offset {offset}: {err}"),
+            })?;
+        if read != want || buf[..read] != payload[offset..offset + read] {
+            return Err(BitrotSelfTestError::RoundtripReadback { algorithm });
+        }
+        offset += read;
+    }
+    Ok(())
+}
+
+/// Flip one byte and require `bitrot_verify` to reject the result.
+async fn bitrot_tamper_check(
+    algorithm: &'static str,
+    algo: HashAlgorithm,
+    payload: &[u8],
+    shard_size: usize,
+    tampered: &'static str,
+    flip_at: usize,
+) -> Result<(), BitrotSelfTestError> {
+    let mut writer = BitrotWriter::new(std::io::Cursor::new(Vec::<u8>::new()), shard_size, algo.clone());
+    for chunk in payload.chunks(shard_size) {
+        writer.write(chunk).await.expect("self-test encode should not fail");
+    }
+    let mut corrupt = writer.into_inner().into_inner();
+    let flip_index = flip_at % corrupt.len();
+    corrupt[flip_index] ^= 0x80;
+
+    let on_disk = bitrot_shard_file_size(payload.len(), shard_size, algo.clone());
+    match bitrot_verify(std::io::Cursor::new(corrupt), on_disk, payload.len(), algo, shard_size).await {
+        // The flipped byte must be rejected as a hash mismatch specifically, not
+        // by any incidental read error: an in-memory cursor cannot fail reads,
+        // so accepting any other failure here would mask a verify path that
+        // errors out before it ever compares hashes.
+        Err(err) if err.to_string().contains("hash mismatch") => Ok(()),
+        Ok(()) => Err(BitrotSelfTestError::TamperNotRejected { algorithm, tampered }),
+        Err(err) => Err(BitrotSelfTestError::RoundtripVerify {
+            algorithm,
+            detail: format!("tampered shard rejected with an unexpected error: {err}"),
+        }),
+    }
+}
+
+/// Verify every bitrot algorithm this crate can write or verify in production:
+/// both streaming Highway variants roundtrip end to end (encode → size formula
+/// → `bitrot_verify` → read back) and reject a flipped byte in both the data
+/// and the leading hash, while all three hashed algorithms reproduce their
+/// pinned known-answer digests.
+///
+/// Runs in well under a millisecond on 4 KiB of data; callers may run it inline
+/// at startup. Pure CPU, no allocation beyond a few KiB of scratch.
+pub async fn bitrot_self_test() -> Result<(), BitrotSelfTestError> {
+    let payload = bitrot_self_test_payload();
+
+    // Externally verifiable vector first: it guards the HashAlgorithm plumbing
+    // itself, before any self-pinned constants are consulted.
+    let abc = HashAlgorithm::SHA256.hash_encode(b"abc");
+    if abc.as_ref() != BITROT_SELF_TEST_KAT_SHA256_ABC.as_slice() {
+        return Err(BitrotSelfTestError::KnownAnswerMismatch {
+            algorithm: "SHA256",
+            got: self_test_hex(abc.as_ref()),
+            want: self_test_hex(&BITROT_SELF_TEST_KAT_SHA256_ABC),
+        });
+    }
+
+    bitrot_kat_check(
+        "HighwayHash256S",
+        &HashAlgorithm::HighwayHash256S,
+        &payload,
+        &BITROT_SELF_TEST_KAT_HIGHWAY_HASH256S,
+    )?;
+    bitrot_kat_check(
+        "HighwayHash256SLegacy",
+        &HashAlgorithm::HighwayHash256SLegacy,
+        &payload,
+        &BITROT_SELF_TEST_KAT_HIGHWAY_HASH256S_LEGACY,
+    )?;
+
+    for (algorithm, algo) in [
+        ("HighwayHash256S", HashAlgorithm::HighwayHash256S),
+        ("HighwayHash256SLegacy", HashAlgorithm::HighwayHash256SLegacy),
+    ] {
+        // Full blocks plus a partial tail, exactly like a real part stripe.
+        let tail_len = 2 * 1024 + 333;
+        bitrot_roundtrip_check(algorithm, algo.clone(), &payload, 1024).await?;
+        bitrot_roundtrip_check(algorithm, algo.clone(), &payload[..tail_len], 1024).await?;
+        // One flipped byte in the final data block, one in the first leading
+        // hash: both must fail verification.
+        bitrot_tamper_check(algorithm, algo.clone(), &payload, 1024, "final data byte", payload.len() - 1).await?;
+        bitrot_tamper_check(algorithm, algo, &payload, 1024, "leading hash byte", 0).await?;
+    }
+
+    Ok(())
+}
+
 #[cfg(test)]
 mod tests {
     use super::{
-        BitrotReader, BitrotWriter, BitrotWriterWrapper, CustomWriter, bitrot_shard_file_size, bitrot_verify, write_all_vectored,
+        BitrotReader, BitrotWriter, BitrotWriterWrapper, CustomWriter, bitrot_kat_check, bitrot_self_test,
+        bitrot_self_test_payload, bitrot_shard_file_size, bitrot_verify, write_all_vectored,
     };
     use super::{MAX_RETAINED_CHUNKS_PER_BLOCK, ShardChunkRead, ShardSource};
     use bytes::Bytes;
@@ -1090,6 +1343,32 @@ mod tests {
         }
     }
 
+    #[test]
+    fn bitrot_self_test_payload_is_deterministic() {
+        // Two independent builds of the payload must agree byte for byte, or
+        // the pinned known-answer digests below would be meaningless.
+        assert_eq!(bitrot_self_test_payload(), bitrot_self_test_payload());
+    }
+
+    #[test]
+    fn bitrot_self_test_rejects_a_wrong_known_answer_digest() {
+        let payload = bitrot_self_test_payload();
+        let wrong = [0u8; 32];
+        let err = bitrot_kat_check("HighwayHash256S", &HashAlgorithm::HighwayHash256S, &payload, &wrong)
+            .expect_err("a zeroed digest must never match");
+        match err {
+            super::BitrotSelfTestError::KnownAnswerMismatch { algorithm, .. } => assert_eq!(algorithm, "HighwayHash256S"),
+            other => panic!("expected KnownAnswerMismatch, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn bitrot_self_test_passes() {
+        bitrot_self_test()
+            .await
+            .expect("the pinned digests and roundtrip checks must all pass on this platform");
+    }
+
     #[tokio::test]
     async fn vectored_test_writers_cover_fallback_flush_and_shutdown_paths() {
         let mut counting = VectoredCountingWriter::default();
@@ -1189,7 +1468,7 @@ mod tests {
         let last = corrupt.len() - 1;
         corrupt[last] ^= 0x80;
         let err = bitrot_verify(
-            Cursor::new(corrupt),
+            std::io::Cursor::new(corrupt),
             super::bitrot_shard_file_size(data.len(), shard_size, algo.clone()),
             data.len(),
             algo,
@@ -1282,7 +1561,7 @@ mod tests {
 
     #[tokio::test]
     async fn bitrot_reader_rejects_output_buffers_larger_than_shard_size() {
-        let mut reader = BitrotReader::new(Cursor::new(Vec::<u8>::new()), 4, HashAlgorithm::None, false);
+        let mut reader = BitrotReader::new(std::io::Cursor::new(Vec::<u8>::new()), 4, HashAlgorithm::None, false);
         let mut out = [0u8; 5];
         let err = reader
             .read(&mut out)
@@ -1407,7 +1686,7 @@ mod tests {
             (HashAlgorithm::HighwayHash256, true),
         ] {
             let label = format!("{algo:?}");
-            let writer = Cursor::new(Vec::<u8>::new());
+            let writer = std::io::Cursor::new(Vec::<u8>::new());
             let mut w = BitrotWriter::new(writer, shard_size, algo.clone());
             w.write(&[7u8; 16]).await.unwrap();
             let written = w.into_inner().into_inner();
@@ -1492,7 +1771,7 @@ mod tests {
     }
 
     async fn encode_one_block(payload: &[u8], shard_size: usize, algo: HashAlgorithm) -> Vec<u8> {
-        let mut w = BitrotWriter::new(Cursor::new(Vec::<u8>::new()), shard_size, algo);
+        let mut w = BitrotWriter::new(std::io::Cursor::new(Vec::<u8>::new()), shard_size, algo);
         w.write(payload).await.unwrap();
         w.into_inner().into_inner()
     }
@@ -1600,7 +1879,7 @@ mod tests {
         for algo in [HashAlgorithm::HighwayHash256S, HashAlgorithm::HighwayHash256SLegacy] {
             for &size in &[1usize, 16, 17, 32, 40, 48] {
                 let payload: Vec<u8> = (0..size).map(|i| i as u8).collect();
-                let mut w = BitrotWriter::new(Cursor::new(Vec::<u8>::new()), shard_size, algo.clone());
+                let mut w = BitrotWriter::new(std::io::Cursor::new(Vec::<u8>::new()), shard_size, algo.clone());
                 for chunk in payload.chunks(shard_size) {
                     w.write(chunk).await.unwrap();
                 }
@@ -1674,14 +1953,14 @@ mod tests {
             w.write(&data).await.expect("write shard");
 
             let mut via_read = vec![0u8; SHARD];
-            let n1 = BitrotReader::new(Cursor::new(encoded.clone()), SHARD, algo.clone(), false)
+            let n1 = BitrotReader::new(std::io::Cursor::new(encoded.clone()), SHARD, algo.clone(), false)
                 .read(&mut via_read)
                 .await
                 .expect("read");
 
             // A buffer with only capacity — no initialized bytes at all.
             let mut via_append: Vec<u8> = Vec::with_capacity(SHARD);
-            let n2 = BitrotReader::new(Cursor::new(encoded), SHARD, algo.clone(), false)
+            let n2 = BitrotReader::new(std::io::Cursor::new(encoded), SHARD, algo.clone(), false)
                 .read_appending(&mut via_append, SHARD)
                 .await
                 .expect("read_appending");
@@ -1706,7 +1985,7 @@ mod tests {
             encoded.truncate(encoded.len() - 1);
 
             let mut out: Vec<u8> = Vec::with_capacity(SHARD);
-            let err = BitrotReader::new(Cursor::new(encoded), SHARD, algo.clone(), false)
+            let err = BitrotReader::new(std::io::Cursor::new(encoded), SHARD, algo.clone(), false)
                 .read_appending(&mut out, SHARD)
                 .await
                 .expect_err("a truncated shard must not succeed");
@@ -1732,7 +2011,7 @@ mod tests {
         encoded[last] ^= 0xff;
 
         let mut out: Vec<u8> = Vec::with_capacity(SHARD);
-        let err = BitrotReader::new(Cursor::new(encoded), SHARD, algo, false)
+        let err = BitrotReader::new(std::io::Cursor::new(encoded), SHARD, algo, false)
             .read_appending(&mut out, SHARD)
             .await
             .expect_err("a corrupt shard must not verify");
@@ -1844,7 +2123,7 @@ mod tests {
             "Cursor<Bytes> must be able to hand out a block, otherwise the fast path is dead code"
         );
         assert_eq!(mem.position(), 8, "taking a block must advance like a read of the same length");
-        let mut streamed = Cursor::new(encoded.clone());
+        let mut streamed = std::io::Cursor::new(encoded.clone());
         assert!(
             ShardSource::try_take_block(&mut streamed, 8).is_none(),
             "a non-Bytes source must stay on the streaming path"
@@ -1872,7 +2151,7 @@ mod tests {
         );
 
         let mut via_stream: Vec<u8> = Vec::with_capacity(SHARD);
-        BitrotReader::new(Cursor::new(encoded), SHARD, algo, false)
+        BitrotReader::new(std::io::Cursor::new(encoded), SHARD, algo, false)
             .read_appending(&mut via_stream, SHARD)
             .await
             .expect("streaming read");

crates/ecstore/src/store/mod.rs

@@ -216,6 +216,16 @@ impl std::fmt::Debug for ECStore {
 /// These delegate to the process-global statics. No local state — the globals
 /// remain the single source of truth until the migration is complete.
 impl ECStore {
+    /// Every erasure set across all pools, pool-major order.
+    ///
+    /// Read-only queries that must consult each set's own copy of a
+    /// per-bucket object (e.g. the scanner's `.usage-cache.bin`) iterate
+    /// this instead of the hash-routed store path, which would always land
+    /// on one set (rustfs/backlog#1872).
+    pub fn all_set_disks(&self) -> Vec<Arc<crate::set_disk::SetDisks>> {
+        self.pools.iter().flat_map(|pool| pool.disk_set.iter().cloned()).collect()
+    }
+
     /// Get server configuration (delegates to global)
     pub fn get_server_config(&self) -> Option<Config> {
         runtime_sources::server_config()

crates/io-core/src/io_profile.rs

@@ -26,7 +26,6 @@ pub enum StorageMedia {
 }
 
 impl StorageMedia {
-    #[allow(dead_code)]
     pub fn as_str(&self) -> &'static str {
         match self {
             Self::Nvme => "nvme",
@@ -60,7 +59,6 @@ pub enum AccessPattern {
 }
 
 impl AccessPattern {
-    #[allow(dead_code)]
     pub fn as_str(&self) -> &'static str {
         match self {
             Self::Sequential => "sequential",
@@ -71,25 +69,21 @@ impl AccessPattern {
     }
 
     /// Check if this is a sequential access pattern.
-    #[allow(dead_code)]
     pub fn is_sequential(&self) -> bool {
         matches!(self, Self::Sequential)
     }
 
     /// Check if this is a random access pattern.
-    #[allow(dead_code)]
     pub fn is_random(&self) -> bool {
         matches!(self, Self::Random)
     }
 
     /// Check if this is a mixed access pattern.
-    #[allow(dead_code)]
     pub fn is_mixed(&self) -> bool {
         matches!(self, Self::Mixed)
     }
 
     /// Check if this pattern is unknown.
-    #[allow(dead_code)]
     pub fn is_unknown(&self) -> bool {
         matches!(self, Self::Unknown)
     }

crates/madmin/Cargo.toml

@@ -37,7 +37,11 @@ hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
 [dependencies]
 hotpath.workspace = true
 humantime.workspace = true
+http.workspace = true
 hyper = { workspace = true, features = ["http2", "http1", "server"] }
+reqwest = { workspace = true, features = ["json"] }
+rustfs-signer.workspace = true
+s3s.workspace = true
 jiff = { workspace = true, features = ["serde"] }
 serde = { workspace = true, features = ["derive"] }
 serde_json = { workspace = true, features = ["raw_value"] }
@@ -49,3 +53,4 @@ doctest = false
 
 [dev-dependencies]
 rmp-serde.workspace = true
+tokio = { workspace = true, features = ["macros", "rt-multi-thread", "net"] }

crates/madmin/src/client.rs

@@ -0,0 +1,851 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Admin API HTTP client for heal and scanner management (rustfs/backlog#1869).
+//!
+//! [`AdminClient`] speaks the `/rustfs/admin/v3` surface with S3 SigV4
+//! request signing (the same scheme the server's admin router authenticates),
+//! so `mc`-style tooling and automation can drive heal start/query/cancel and
+//! read background-heal / scanner status without hand-rolling HTTP.
+//!
+//! Wire structs in this module mirror the server-side shapes
+//! (`rustfs/src/admin/handlers/heal.rs`, `handlers/scanner.rs`,
+//! `rustfs-common/src/heal_channel.rs`), following the madmin-go model where
+//! the SDK owns its own copies and round-trip tests pin the encoding. Deeply
+//! nested status payloads that the server composes from runtime types are
+//! carried through as `serde_json::Value` and flattened maps rather than
+//! duplicated field-for-field, so the client cannot silently drift on fields
+//! it never interprets.
+
+use crate::heal_commands::HealResultItem;
+use http::Method;
+use serde::{Deserialize, Serialize, de};
+use std::time::Duration;
+
+/// Default admin API path prefix on a RustFS endpoint.
+pub const DEFAULT_ADMIN_API_PREFIX: &str = "/rustfs/admin";
+/// Default SigV4 region when the server has no explicit region configured.
+pub const DEFAULT_REGION: &str = "us-east-1";
+
+/// Scan mode for a heal request, mirroring the server's numeric-or-name wire
+/// encoding (`0` unknown/default, `1` normal, `2` deep).
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+pub enum HealScanMode {
+    /// Server default; behaves as [`HealScanMode::Normal`].
+    #[default]
+    Unknown,
+    /// Metadata-level checks only.
+    Normal,
+    /// Full bitrot verification while healing.
+    Deep,
+}
+
+impl HealScanMode {
+    fn wire_number(self) -> u8 {
+        match self {
+            Self::Unknown => 0,
+            Self::Normal => 1,
+            Self::Deep => 2,
+        }
+    }
+
+    fn from_wire_number(value: u8) -> Option<Self> {
+        match value {
+            0 => Some(Self::Unknown),
+            1 => Some(Self::Normal),
+            2 => Some(Self::Deep),
+            _ => None,
+        }
+    }
+
+    fn from_wire_name(value: &str) -> Option<Self> {
+        match value {
+            "unknown" => Some(Self::Unknown),
+            "normal" => Some(Self::Normal),
+            "deep" => Some(Self::Deep),
+            _ => None,
+        }
+    }
+}
+
+impl Serialize for HealScanMode {
+    fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
+        serializer.serialize_u8(self.wire_number())
+    }
+}
+
+impl<'de> Deserialize<'de> for HealScanMode {
+    fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
+        struct HealScanModeVisitor;
+
+        impl de::Visitor<'_> for HealScanModeVisitor {
+            type Value = HealScanMode;
+
+            fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+                formatter.write_str("a heal scan mode number or name")
+            }
+
+            fn visit_u64<E: de::Error>(self, value: u64) -> Result<Self::Value, E> {
+                u8::try_from(value)
+                    .ok()
+                    .and_then(HealScanMode::from_wire_number)
+                    .ok_or_else(|| E::custom(format!("unknown heal scan mode number: {value}")))
+            }
+
+            fn visit_str<E: de::Error>(self, value: &str) -> Result<Self::Value, E> {
+                HealScanMode::from_wire_name(value).ok_or_else(|| E::custom(format!("unknown heal scan mode name: {value}")))
+            }
+        }
+
+        deserializer.deserialize_any(HealScanModeVisitor)
+    }
+}
+
+/// Heal options for an admin heal request (mirror of the server body type).
+/// Fields default on decode: a client should tolerate a server response whose
+/// settings object omits fields it never set.
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
+pub struct HealOpts {
+    #[serde(default)]
+    pub recursive: bool,
+    #[serde(rename = "dryRun", default)]
+    pub dry_run: bool,
+    #[serde(default)]
+    pub remove: bool,
+    #[serde(default)]
+    pub recreate: bool,
+    #[serde(rename = "scanMode", default)]
+    pub scan_mode: HealScanMode,
+    #[serde(rename = "updateParity", default)]
+    pub update_parity: bool,
+    #[serde(rename = "nolock", default)]
+    pub no_lock: bool,
+    #[serde(rename = "pool", default)]
+    pub pool: Option<usize>,
+    #[serde(rename = "set", default)]
+    pub set: Option<usize>,
+}
+
+/// Successful heal start / path-scoped cancel response.
+#[derive(Debug, Clone, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct HealStartSuccess {
+    pub client_token: String,
+    pub client_address: String,
+    #[serde(default)]
+    pub start_time: String,
+}
+
+/// Heal task status response (query, cancel-with-token, start-then-poll).
+#[derive(Debug, Clone, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct HealTaskStatus {
+    /// `running` | `finished` | `stopped` | `notFound`.
+    pub summary: String,
+    /// Failure detail for stopped tasks; empty otherwise.
+    #[serde(rename = "detail", default)]
+    pub failure_detail: String,
+    #[serde(default)]
+    pub start_time: String,
+    #[serde(default)]
+    pub settings: HealOpts,
+    #[serde(default)]
+    pub items: Vec<HealResultItem>,
+    #[serde(default)]
+    pub truncated: bool,
+    /// Live progress snapshot; the exact shape is owned by the heal runtime.
+    #[serde(default)]
+    pub progress: Option<serde_json::Value>,
+}
+
+/// `POST /v3/background-heal/status` response. Known top-level fields are
+/// typed; the flattened heal info and operations matrix pass through verbatim.
+#[derive(Debug, Clone, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct BackgroundHealStatus {
+    /// `disabled` | `uninitialized` | `idle` | `active` | `degraded`.
+    pub state: String,
+    #[serde(default)]
+    pub heal_queue_length: u64,
+    #[serde(default)]
+    pub heal_active_tasks: u64,
+    #[serde(default)]
+    pub cluster_status_complete: bool,
+    #[serde(default)]
+    pub progress: Option<serde_json::Value>,
+    /// Remaining wire fields (flattened `BackgroundHealInfo` plus the
+    /// priority-by-source operations matrix), carried verbatim.
+    #[serde(flatten)]
+    pub extra: serde_json::Map<String, serde_json::Value>,
+}
+
+/// `GET /v3/scanner/status` response, typed at the fields operators branch
+/// on; everything else passes through verbatim.
+#[derive(Debug, Clone, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ScannerStatus {
+    pub enabled: bool,
+    /// `fresh` | `stale` | `unknown`; absent when the scanner never completed
+    /// a cycle.
+    #[serde(default)]
+    pub freshness: Option<ScannerFreshness>,
+    #[serde(flatten)]
+    pub extra: serde_json::Map<String, serde_json::Value>,
+}
+
+/// Freshness block of the scanner status response.
+#[derive(Debug, Clone, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ScannerFreshness {
+    /// `fresh` | `stale` | `unknown`.
+    pub state: String,
+}
+
+impl ScannerStatus {
+    /// Convenience accessor for the freshness state string.
+    pub fn freshness(&self) -> &str {
+        self.freshness
+            .as_ref()
+            .map(|freshness| freshness.state.as_str())
+            .unwrap_or("unknown")
+    }
+}
+
+/// Everything that can go wrong in an admin client call.
+#[derive(Debug)]
+pub enum AdminClientError {
+    /// The endpoint URL could not be parsed.
+    InvalidEndpoint(String),
+    /// Request build/send failed (DNS, connect, timeout, body read).
+    Transport(reqwest::Error),
+    /// The server answered a non-2xx status.
+    HttpStatus { status: u16, body: String },
+    /// The response body did not decode into the expected shape.
+    Decode { message: String },
+}
+
+impl std::fmt::Display for AdminClientError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::InvalidEndpoint(message) => write!(f, "invalid admin endpoint: {message}"),
+            Self::Transport(err) => write!(f, "admin request transport failure: {err}"),
+            Self::HttpStatus { status, body } => write!(f, "admin request failed with HTTP {status}: {body}"),
+            Self::Decode { message } => write!(f, "admin response decode failure: {message}"),
+        }
+    }
+}
+
+impl std::error::Error for AdminClientError {}
+
+impl From<reqwest::Error> for AdminClientError {
+    fn from(err: reqwest::Error) -> Self {
+        Self::Transport(err)
+    }
+}
+
+/// A signed client for a RustFS admin API.
+#[derive(Debug, Clone)]
+pub struct AdminClient {
+    endpoint: reqwest::Url,
+    access_key: String,
+    secret_key: String,
+    session_token: String,
+    region: String,
+    api_prefix: String,
+    http: reqwest::Client,
+}
+
+impl AdminClient {
+    /// Build a client for `endpoint` (e.g. `http://127.0.0.1:9000`) using root
+    /// or admin credentials. Requests are SigV4-signed with the same scheme
+    /// the server's admin router authenticates.
+    pub fn new(endpoint: &str, access_key: &str, secret_key: &str) -> Result<Self, AdminClientError> {
+        let url = reqwest::Url::parse(endpoint).map_err(|err| AdminClientError::InvalidEndpoint(err.to_string()))?;
+        if url.host_str().is_none() {
+            return Err(AdminClientError::InvalidEndpoint("endpoint has no host".to_string()));
+        }
+        let http = reqwest::Client::builder()
+            .connect_timeout(Duration::from_secs(10))
+            .timeout(Duration::from_secs(30))
+            .build()
+            .map_err(AdminClientError::Transport)?;
+        Ok(Self {
+            endpoint: url,
+            access_key: access_key.to_string(),
+            secret_key: secret_key.to_string(),
+            session_token: String::new(),
+            region: DEFAULT_REGION.to_string(),
+            api_prefix: DEFAULT_ADMIN_API_PREFIX.to_string(),
+            http,
+        })
+    }
+
+    /// Attach an STS session token (signed as `x-amz-security-token`).
+    pub fn with_session_token(mut self, session_token: impl Into<String>) -> Self {
+        self.session_token = session_token.into();
+        self
+    }
+
+    /// Override the SigV4 region (defaults to `us-east-1`, matching a
+    /// region-less RustFS deployment).
+    pub fn with_region(mut self, region: impl Into<String>) -> Self {
+        self.region = region.into();
+        self
+    }
+
+    /// Override the admin API path prefix (defaults to `/rustfs/admin`).
+    pub fn with_api_prefix(mut self, prefix: impl Into<String>) -> Self {
+        self.api_prefix = prefix.into();
+        self
+    }
+
+    /// Start a heal. `bucket` empty and `prefix` empty heals the whole
+    /// deployment (requires `recursive` or a `pool`/`set` pair in `opts`,
+    /// enforced server-side); a bucket alone heals the bucket (the server
+    /// forces `recursive` for bucket heals).
+    pub async fn heal_start(
+        &self,
+        bucket: Option<&str>,
+        prefix: Option<&str>,
+        opts: &HealOpts,
+        force_start: bool,
+    ) -> Result<HealStartSuccess, AdminClientError> {
+        let body = serde_json::to_vec(opts).map_err(|err| AdminClientError::Decode {
+            message: err.to_string(),
+        })?;
+        let mut query = Vec::new();
+        if force_start {
+            query.push(("forceStart", "true".to_string()));
+        }
+        self.post_json(&heal_path(bucket, prefix), &query, body).await
+    }
+
+    /// Query the status of the heal identified by `client_token` (the token
+    /// returned by [`Self::heal_start`]) at the path it was started on.
+    pub async fn heal_status(
+        &self,
+        bucket: Option<&str>,
+        prefix: Option<&str>,
+        client_token: &str,
+    ) -> Result<HealTaskStatus, AdminClientError> {
+        self.post_json(&heal_path(bucket, prefix), &[("clientToken", client_token.to_string())], Vec::new())
+            .await
+    }
+
+    /// Stop a heal: with a `client_token` only that task is cancelled and its
+    /// final status returned; without one, every heal task at the path is
+    /// cancelled (the server answers with a start-success-shaped receipt).
+    pub async fn heal_stop(
+        &self,
+        bucket: Option<&str>,
+        prefix: Option<&str>,
+        client_token: Option<&str>,
+    ) -> Result<HealStopOutcome, AdminClientError> {
+        let mut query = vec![("forceStop", "true".to_string())];
+        if let Some(token) = client_token {
+            query.push(("clientToken", token.to_string()));
+        }
+        match client_token {
+            Some(_) => {
+                let status: HealTaskStatus = self.post_json(&heal_path(bucket, prefix), &query, Vec::new()).await?;
+                Ok(HealStopOutcome::Stopped(status))
+            }
+            None => {
+                let success: HealStartSuccess = self.post_json(&heal_path(bucket, prefix), &query, Vec::new()).await?;
+                Ok(HealStopOutcome::PathStopped(success))
+            }
+        }
+    }
+
+    /// Cluster-aggregated background heal status.
+    pub async fn background_heal_status(&self) -> Result<BackgroundHealStatus, AdminClientError> {
+        self.get_json("/v3/background-heal/status").await
+    }
+
+    /// Data scanner status (enabled state, freshness, runtime config).
+    pub async fn scanner_status(&self) -> Result<ScannerStatus, AdminClientError> {
+        self.get_json("/v3/scanner/status").await
+    }
+
+    /// ILM expiry worker status. The payload is owned by the expiry
+    /// subsystem and still evolving; returned verbatim.
+    pub async fn ilm_expiry_status(&self) -> Result<serde_json::Value, AdminClientError> {
+        self.get_json("/v3/ilm/expiry/status").await
+    }
+
+    /// Durable replacement-recovery status (admin v4). The payload is owned
+    /// by the heal runtime; returned verbatim.
+    pub async fn replacement_recovery_status(&self) -> Result<serde_json::Value, AdminClientError> {
+        self.get_json("/v4/heal/replacement-recovery").await
+    }
+
+    /// Signed GET returning a decoded JSON body; escape hatch for endpoints
+    /// this client does not wrap yet.
+    pub async fn get_json<T: for<'de> Deserialize<'de>>(&self, path: &str) -> Result<T, AdminClientError> {
+        let url = self.url_for(path, &[])?;
+        let request = self.sign_and_build(Method::GET, url, Vec::new(), None).await?;
+        self.execute(request).await
+    }
+
+    /// Signed POST returning a decoded JSON body.
+    async fn post_json<T: for<'de> Deserialize<'de>>(
+        &self,
+        path: &str,
+        query: &[(&str, String)],
+        body: Vec<u8>,
+    ) -> Result<T, AdminClientError> {
+        let content_type = if body.is_empty() { None } else { Some("application/json") };
+        let url = self.url_for(path, query)?;
+        let request = self.sign_and_build(Method::POST, url, body, content_type).await?;
+        self.execute(request).await
+    }
+
+    fn url_for(&self, path: &str, query: &[(&str, String)]) -> Result<reqwest::Url, AdminClientError> {
+        let mut url = self
+            .endpoint
+            .join(&format!("{}{}", self.api_prefix.trim_end_matches('/'), path))
+            .map_err(|err| AdminClientError::InvalidEndpoint(err.to_string()))?;
+        if !query.is_empty() {
+            let mut pairs = url.query_pairs_mut();
+            for (key, value) in query {
+                pairs.append_pair(key, value);
+            }
+        }
+        Ok(url)
+    }
+
+    /// Build a SigV4-signed request via the same signer the server trusts,
+    /// then hand the signed headers to the HTTP client. The signature covers
+    /// method, path, query, and an unsigned-payload marker — the same shape
+    /// RustFS itself sends for peer admin calls.
+    async fn sign_and_build(
+        &self,
+        method: Method,
+        url: reqwest::Url,
+        body: Vec<u8>,
+        content_type: Option<&str>,
+    ) -> Result<reqwest::Request, AdminClientError> {
+        let authority = match (url.host_str(), url.port_or_known_default()) {
+            (Some(host), Some(port)) => format!("{host}:{port}"),
+            _ => return Err(AdminClientError::InvalidEndpoint("endpoint has no authority".to_string())),
+        };
+        let mut builder = http::Request::builder()
+            .method(method.clone())
+            .uri(url.as_str())
+            .header(http::header::HOST, &authority)
+            .header("x-amz-content-sha256", rustfs_signer::constants::UNSIGNED_PAYLOAD);
+        if let Some(content_type) = content_type {
+            builder = builder.header(http::header::CONTENT_TYPE, content_type);
+        }
+        let unsigned = builder
+            .body(s3s::Body::empty())
+            .map_err(|err| AdminClientError::InvalidEndpoint(format!("build request failed: {err}")))?;
+        let signed = rustfs_signer::sign_v4(
+            unsigned,
+            body.len() as i64,
+            &self.access_key,
+            &self.secret_key,
+            &self.session_token,
+            &self.region,
+        );
+
+        let mut request = self
+            .http
+            .request(method, url)
+            .body(body)
+            .build()
+            .map_err(AdminClientError::Transport)?;
+        let headers = request.headers_mut();
+        for (name, value) in signed.headers().iter() {
+            // HOST is owned by the HTTP client; the signed value above was
+            // built from the same URL authority, so they always agree.
+            if name == http::header::HOST {
+                continue;
+            }
+            headers.insert(name, value.clone());
+        }
+        Ok(request)
+    }
+
+    async fn execute<T: for<'de> Deserialize<'de>>(&self, request: reqwest::Request) -> Result<T, AdminClientError> {
+        let response = self.http.execute(request).await?;
+        let status = response.status();
+        let bytes = response.bytes().await?;
+        if !status.is_success() {
+            return Err(AdminClientError::HttpStatus {
+                status: status.as_u16(),
+                body: String::from_utf8_lossy(&bytes).into_owned(),
+            });
+        }
+        serde_json::from_slice(&bytes).map_err(|err| AdminClientError::Decode {
+            message: err.to_string(),
+        })
+    }
+}
+
+/// Response of [`AdminClient::heal_stop`]: cancelling a single tokened task
+/// answers with that task's status, cancelling a whole path answers with a
+/// start-success-shaped receipt.
+#[derive(Debug, Clone)]
+pub enum HealStopOutcome {
+    Stopped(HealTaskStatus),
+    PathStopped(HealStartSuccess),
+}
+
+fn heal_path(bucket: Option<&str>, prefix: Option<&str>) -> String {
+    match (bucket, prefix) {
+        (Some(bucket), Some(prefix)) if !bucket.is_empty() && !prefix.is_empty() => {
+            format!("/v3/heal/{}/{}", percent_encode_path_segment(bucket), percent_encode_path_segment(prefix))
+        }
+        (Some(bucket), Some(_)) | (Some(bucket), None) if !bucket.is_empty() => {
+            format!("/v3/heal/{}", percent_encode_path_segment(bucket))
+        }
+        _ => "/v3/heal/".to_string(),
+    }
+}
+
+/// Encode a single path segment (slashes are content, not separators, inside
+/// bucket/prefix path params).
+fn percent_encode_path_segment(segment: &str) -> String {
+    let mut out = String::with_capacity(segment.len());
+    for byte in segment.bytes() {
+        match byte {
+            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => out.push(byte as char),
+            _ => out.push_str(&format!("%{byte:02X}")),
+        }
+    }
+    out
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{
+        AdminClient, AdminClientError, BackgroundHealStatus, HealOpts, HealScanMode, HealStartSuccess, HealTaskStatus,
+        ScannerStatus, heal_path, percent_encode_path_segment,
+    };
+    use serde_json::json;
+    use std::sync::{Arc, Mutex};
+
+    #[test]
+    fn heal_paths_cover_root_bucket_and_prefix() {
+        assert_eq!(heal_path(None, None), "/v3/heal/");
+        assert_eq!(heal_path(Some(""), Some("")), "/v3/heal/");
+        assert_eq!(heal_path(Some("bucket"), None), "/v3/heal/bucket");
+        assert_eq!(heal_path(Some("bucket"), Some("pre/fix")), "/v3/heal/bucket/pre%2Ffix");
+    }
+
+    #[test]
+    fn path_segments_percent_encode_reserved_characters() {
+        assert_eq!(percent_encode_path_segment("a b"), "a%20b");
+        assert_eq!(percent_encode_path_segment("a/b"), "a%2Fb");
+        assert_eq!(percent_encode_path_segment("ü"), "%C3%BC");
+    }
+
+    #[test]
+    fn heal_opts_round_trip_through_the_server_wire_shape() {
+        let opts = HealOpts {
+            recursive: true,
+            dry_run: false,
+            remove: true,
+            recreate: false,
+            scan_mode: HealScanMode::Deep,
+            update_parity: true,
+            no_lock: false,
+            pool: Some(1),
+            set: Some(2),
+        };
+        let wire = serde_json::to_value(&opts).unwrap();
+        assert_eq!(wire["scanMode"], json!(2), "the server body decodes scanMode as a number");
+        let back: HealOpts = serde_json::from_value(wire).unwrap();
+        assert_eq!(back.scan_mode, HealScanMode::Deep);
+        assert_eq!(back.pool, Some(1));
+    }
+
+    #[test]
+    fn heal_scan_mode_accepts_both_wire_encodings() {
+        assert_eq!(serde_json::from_value::<HealScanMode>(json!(1)).unwrap(), HealScanMode::Normal);
+        assert_eq!(serde_json::from_value::<HealScanMode>(json!("deep")).unwrap(), HealScanMode::Deep);
+        assert!(serde_json::from_value::<HealScanMode>(json!(9)).is_err());
+        assert!(serde_json::from_value::<HealScanMode>(json!("sideways")).is_err());
+    }
+
+    #[test]
+    fn heal_task_status_decodes_the_server_response_shape() {
+        let raw = json!({
+            "summary": "finished",
+            "detail": "",
+            "startTime": "2026-08-17T00:00:00Z",
+            "settings": {"recursive": false, "scanMode": 1},
+            "items": [{
+                "resultId": 1, "type": "object", "bucket": "b", "object": "o", "versionId": "", "detail": "",
+                "parityBlocks": 2, "dataBlocks": 2, "diskCount": 4, "setCount": 1,
+                "before": {"drives": []}, "after": {"drives": []}, "objectSize": 128
+            }],
+            "truncated": false
+        });
+        let status: HealTaskStatus = serde_json::from_value(raw).unwrap();
+        assert_eq!(status.summary, "finished");
+        assert_eq!(status.items.len(), 1);
+        assert_eq!(status.settings.scan_mode, HealScanMode::Normal);
+        assert!(status.progress.is_none());
+    }
+
+    #[test]
+    fn background_heal_status_types_known_fields_and_passes_the_rest_through() {
+        let raw = json!({
+            "state": "active",
+            "bitrotStartTime": "t",
+            "healQueueLength": 3,
+            "healActiveTasks": 1,
+            "healOperations": {"queueLength": 3},
+            "clusterStatusComplete": true
+        });
+        let status: BackgroundHealStatus = serde_json::from_value(raw).unwrap();
+        assert_eq!(status.state, "active");
+        assert_eq!(status.heal_queue_length, 3);
+        assert!(status.cluster_status_complete);
+        assert!(status.extra.contains_key("healOperations"), "unknown nested payloads must pass through");
+    }
+
+    #[test]
+    fn scanner_status_defaults_freshness_to_unknown() {
+        let raw = json!({"enabled": true, "freshness": {"state": "stale"}, "metrics": {}});
+        let status: ScannerStatus = serde_json::from_value(raw).unwrap();
+        assert_eq!(status.freshness(), "stale");
+        let bare: ScannerStatus = serde_json::from_value(json!({"enabled": false})).unwrap();
+        assert_eq!(bare.freshness(), "unknown");
+    }
+
+    #[test]
+    fn invalid_endpoint_is_rejected_without_io() {
+        let err = AdminClient::new("not a url", "ak", "sk").unwrap_err();
+        assert!(matches!(err, AdminClientError::InvalidEndpoint(_)));
+    }
+
+    #[tokio::test]
+    async fn signed_requests_carry_sigv4_authorization_and_correct_target() {
+        let server = TestServer::spawn(r#"{"clientToken":"token-1","clientAddress":"127.0.0.1:9","startTime":"t"}"#, 200).await;
+        let client = AdminClient::new(&format!("http://{}", server.addr), "minioadmin", "minioadmin")
+            .expect("client builds against the test server");
+
+        let start: HealStartSuccess = client
+            .heal_start(
+                Some("bucket"),
+                None,
+                &HealOpts {
+                    recursive: true,
+                    ..Default::default()
+                },
+                false,
+            )
+            .await
+            .expect("signed heal start decodes");
+
+        assert_eq!(start.client_token, "token-1");
+        let request = server.recorded();
+        assert_eq!(request.method, "POST");
+        assert_eq!(request.path, "/rustfs/admin/v3/heal/bucket");
+        assert!(!request.query.contains("forceStart"), "absent flags must not be sent");
+        let auth = request.header("authorization").expect("request must be signed");
+        assert!(auth.starts_with("AWS4-HMAC-SHA256"), "SigV4 scheme, got: {auth}");
+        assert!(auth.contains("Credential=minioadmin/"), "credentials must be in the Authorization header");
+        assert_eq!(
+            request.header("x-amz-content-sha256").as_deref(),
+            Some("UNSIGNED-PAYLOAD"),
+            "the client signs the same payload marker RustFS peer calls use"
+        );
+        assert_eq!(request.header("content-type").as_deref(), Some("application/json"));
+        assert!(request.body.contains("\"recursive\":true"));
+    }
+
+    #[tokio::test]
+    async fn query_sends_client_token_on_the_same_path() {
+        let body = r#"{"summary":"running","detail":"","settings":{"recursive":false},"items":[],"truncated":false}"#;
+        let server = TestServer::spawn(body, 200).await;
+        let client = AdminClient::new(&format!("http://{}", server.addr), "ak", "sk").unwrap();
+
+        let status = client
+            .heal_status(Some("bucket"), None, "token-1")
+            .await
+            .expect("status decodes");
+        assert_eq!(status.summary, "running");
+        let request = server.recorded();
+        assert_eq!(request.path, "/rustfs/admin/v3/heal/bucket");
+        assert!(request.query.contains("clientToken=token-1"));
+        assert!(!request.query.contains("forceStop"));
+    }
+
+    #[tokio::test]
+    async fn stop_without_token_takes_the_path_cancel_branch() {
+        let server = TestServer::spawn(r#"{"clientToken":"path","clientAddress":"c","startTime":"t"}"#, 200).await;
+        let client = AdminClient::new(&format!("http://{}", server.addr), "ak", "sk").unwrap();
+
+        let outcome = client.heal_stop(Some("bucket"), None, None).await.expect("path stop decodes");
+        assert!(matches!(outcome, super::HealStopOutcome::PathStopped(_)));
+        let request = server.recorded();
+        assert!(request.query.contains("forceStop=true"));
+        assert!(!request.query.contains("clientToken"));
+    }
+
+    #[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;
+        let client = AdminClient::new(&format!("http://{}", server.addr), "ak", "sk").unwrap();
+        let err = client.scanner_status().await.unwrap_err();
+        match err {
+            AdminClientError::HttpStatus { status, body } => {
+                assert_eq!(status, 403);
+                assert!(body.contains("AccessDenied"));
+            }
+            other => panic!("expected HttpStatus, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn malformed_success_body_maps_to_a_decode_error() {
+        let server = TestServer::spawn("not json", 200).await;
+        let client = AdminClient::new(&format!("http://{}", server.addr), "ak", "sk").unwrap();
+        assert!(matches!(client.scanner_status().await.unwrap_err(), AdminClientError::Decode { .. }));
+    }
+
+    /// One recorded request, parsed off the wire with the minimum needed for
+    /// assertions: method, path, query, headers, body.
+    #[derive(Debug, Clone)]
+    struct RecordedRequest {
+        method: String,
+        path: String,
+        query: String,
+        headers: Vec<(String, String)>,
+        body: String,
+    }
+
+    impl RecordedRequest {
+        fn header(&self, name: &str) -> Option<String> {
+            self.headers
+                .iter()
+                .find(|(key, _)| key.eq_ignore_ascii_case(name))
+                .map(|(_, value)| value.clone())
+        }
+    }
+
+    /// Minimal HTTP/1.1 server: one canned response per connection, every
+    /// request recorded behind an `Arc<Mutex>`. Deliberately dependency-free —
+    /// the assertions only need the raw request bytes.
+    struct TestServer {
+        addr: std::net::SocketAddr,
+        requests: Arc<Mutex<Vec<RecordedRequest>>>,
+    }
+
+    impl TestServer {
+        async fn spawn(response_body: &'static str, status: u16) -> Self {
+            use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
+            let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
+                .await
+                .expect("bind ephemeral port");
+            let addr = listener.local_addr().expect("local addr");
+            let requests: Arc<Mutex<Vec<RecordedRequest>>> = Arc::new(Mutex::new(Vec::new()));
+
+            let recorded = requests.clone();
+            tokio::spawn(async move {
+                let reason = if status == 200 { "OK" } else { "Forbidden" };
+                let response = format!(
+                    "HTTP/1.1 {status} {reason}\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{response_body}",
+                    response_body.len()
+                );
+                // Each request is a fresh connection (connection: close); a
+                // bounded loop serves every call a test makes while letting
+                // the task exit instead of lingering for the whole process.
+                for _ in 0..16 {
+                    let Ok((mut stream, _)) = listener.accept().await else {
+                        break;
+                    };
+                    let mut buffer = Vec::with_capacity(2048);
+                    let mut chunk = [0u8; 2048];
+                    // Read headers plus content-length body, or stop on close.
+                    loop {
+                        if let Some(end) = find_header_end(&buffer) {
+                            let content_length = extract_content_length(&buffer[..end]);
+                            if buffer.len() >= end + content_length {
+                                break;
+                            }
+                        }
+                        let n = match stream.read(&mut chunk).await {
+                            Ok(0) | Err(_) => break,
+                            Ok(n) => n,
+                        };
+                        buffer.extend_from_slice(&chunk[..n]);
+                        if buffer.len() > 64 * 1024 {
+                            break;
+                        }
+                    }
+                    if let Some(request) = parse_request(&buffer) {
+                        recorded.lock().expect("recorded lock").push(request);
+                    }
+                    let _ = stream.write_all(response.as_bytes()).await;
+                    let _ = stream.shutdown().await;
+                }
+            });
+
+            Self { addr, requests }
+        }
+
+        fn recorded(&self) -> RecordedRequest {
+            self.requests
+                .lock()
+                .expect("recorded lock")
+                .last()
+                .cloned()
+                .expect("the client call must have produced one recorded request")
+        }
+    }
+
+    fn find_header_end(buffer: &[u8]) -> Option<usize> {
+        buffer.windows(4).position(|window| window == b"\r\n\r\n").map(|pos| pos + 4)
+    }
+
+    fn extract_content_length(headers: &[u8]) -> usize {
+        let text = String::from_utf8_lossy(headers).to_ascii_lowercase();
+        text.lines()
+            .find_map(|line| line.strip_prefix("content-length:"))
+            .and_then(|value| value.trim().parse().ok())
+            .unwrap_or(0)
+    }
+
+    fn parse_request(raw: &[u8]) -> Option<RecordedRequest> {
+        let end = find_header_end(raw)?;
+        let head = String::from_utf8_lossy(&raw[..end]);
+        let body = String::from_utf8_lossy(&raw[end..]).into_owned();
+        let mut lines = head.lines();
+        let request_line = lines.next()?;
+        let mut parts = request_line.split_whitespace();
+        let method = parts.next()?.to_string();
+        let target = parts.next()?.to_string();
+        let (path, query) = match target.split_once('?') {
+            Some((path, query)) => (path.to_string(), query.to_string()),
+            None => (target, String::new()),
+        };
+        let headers = lines
+            .filter_map(|line| line.split_once(':'))
+            .map(|(name, value)| (name.trim().to_string(), value.trim().to_string()))
+            .collect();
+        Some(RecordedRequest {
+            method,
+            path,
+            query,
+            headers,
+            body,
+        })
+    }
+}

crates/madmin/src/lib.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+pub mod client;
 pub mod group;
 pub mod heal_commands;
 pub mod health;
@@ -25,6 +26,7 @@ pub mod trace;
 pub mod user;
 pub mod utils;
 
+pub use client::*;
 pub use group::*;
 pub use info_commands::*;
 pub use policy::*;

crates/object-capacity/src/capacity_manager.rs

@@ -427,7 +427,6 @@ pub enum DataSource {
     /// Write triggered
     WriteTriggered,
     /// Fallback value
-    #[allow(dead_code)]
     Fallback,
 }
 
@@ -603,7 +602,6 @@ impl WriteRecord {
 
 /// Hybrid strategy configuration
 #[derive(Debug, Clone)]
-#[allow(dead_code)]
 pub struct HybridStrategyConfig {
     /// Scheduled update interval
     pub scheduled_update_interval: Duration,
@@ -998,14 +996,12 @@ impl HybridCapacityManager {
     }
 
     /// Get cache age
-    #[allow(dead_code)]
     pub async fn get_cache_age(&self) -> Option<Duration> {
         let cache = self.cache.read().await;
         cache.as_ref().map(|c| c.last_update.elapsed())
     }
 
     /// Get write frequency (writes/minute)
-    #[allow(dead_code)]
     pub async fn get_write_frequency(&self) -> usize {
         let record = &self.write_record;
         record.recent_write_count(record.monotonic_second())
@@ -1300,7 +1296,6 @@ pub fn get_capacity_manager() -> Arc<HybridCapacityManager> {
 ///     .update_capacity(CapacityUpdate::exact(1000, 0), DataSource::RealTime)
 ///     .await;
 /// ```
-#[allow(dead_code)]
 pub fn create_isolated_manager(config: HybridStrategyConfig) -> Arc<HybridCapacityManager> {
     Arc::new(HybridCapacityManager::new(config))
 }

crates/rio/src/compress_index.rs

@@ -49,7 +49,6 @@ pub struct IndexInfo {
     pub uncompressed_offset: i64,
 }
 
-#[allow(dead_code)]
 impl Index {
     pub fn new() -> Self {
         Self {
@@ -60,14 +59,6 @@ impl Index {
         }
     }
 
-    #[allow(dead_code)]
-    fn reset(&mut self, max_block: usize) {
-        self.est_block_uncomp = max_block as i64;
-        self.total_compressed = -1;
-        self.total_uncompressed = -1;
-        self.info.clear();
-    }
-
     pub fn len(&self) -> usize {
         self.info.len()
     }
@@ -511,47 +502,6 @@ fn read_varint(buf: &[u8]) -> io::Result<(i64, usize)> {
     Err(io::Error::new(io::ErrorKind::UnexpectedEof, "unexpected EOF"))
 }
 
-// Helper functions for index header manipulation
-#[allow(dead_code)]
-pub fn remove_index_headers(b: &[u8]) -> Option<&[u8]> {
-    if b.len() < 4 + S2_INDEX_TRAILER.len() {
-        return None;
-    }
-
-    // Skip size
-    let b = &b[4..];
-
-    // Check trailer
-    if !b.starts_with(S2_INDEX_TRAILER) {
-        return None;
-    }
-
-    Some(&b[S2_INDEX_TRAILER.len()..])
-}
-
-#[allow(dead_code)]
-pub fn restore_index_headers(in_data: &[u8]) -> Vec<u8> {
-    if in_data.is_empty() {
-        return Vec::new();
-    }
-
-    let mut b = Vec::with_capacity(4 + S2_INDEX_HEADER.len() + in_data.len() + S2_INDEX_TRAILER.len() + 4);
-    b.extend_from_slice(&[0x50, 0x2A, 0x4D, 0x18]);
-    b.extend_from_slice(S2_INDEX_HEADER);
-    b.extend_from_slice(in_data);
-
-    let total_size = (b.len() + 4 + S2_INDEX_TRAILER.len()) as u32;
-    b.extend_from_slice(&total_size.to_le_bytes());
-    b.extend_from_slice(S2_INDEX_TRAILER);
-
-    let chunk_len = b.len() - 4;
-    b[1] = chunk_len as u8;
-    b[2] = (chunk_len >> 8) as u8;
-    b[3] = (chunk_len >> 16) as u8;
-
-    b
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;

crates/scanner/src/data_usage_define.rs

@@ -28,7 +28,8 @@ use rustfs_common::heal_channel::HealScanMode;
 use rustfs_config::ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS;
 pub use rustfs_data_usage::{
     AllTierStats, BucketTargetUsageInfo, BucketUsageInfo, DATA_USAGE_OBJECT_NAME, DATA_USAGE_OBSERVED_OBJECT_NAME,
-    DataUsageEntry, DataUsageHash, DataUsageHashMap, DataUsageInfo, LEGACY_DATA_USAGE_OBJECT_NAME, TierStats, hash_path,
+    DataUsageEntry, DataUsageHash, DataUsageHashMap, DataUsageInfo, LEGACY_DATA_USAGE_OBJECT_NAME, PrefixUsageEntry,
+    PrefixUsageQuery, PrefixUsageSummary, TierStats, hash_path, prefix_usage_in_cache,
 };
 use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf};
 use tokio::time::{Duration, Instant, sleep, timeout};
@@ -430,6 +431,13 @@ pub(crate) enum DataUsageCachePrepareOutcome {
 }
 
 impl DataUsageCache {
+    /// Prefix-level usage query over this (writer-side) cache; see
+    /// [`prefix_usage_in_cache`] for the semantics
+    /// (rustfs/backlog#1872).
+    pub fn prefix_usage(&self, bucket: &str, prefix: &str, max_entries: usize) -> Option<PrefixUsageQuery> {
+        prefix_usage_in_cache(&self.cache, bucket, prefix, max_entries)
+    }
+
     pub(crate) fn prepare_for_scan(
         &mut self,
         name: &str,

crates/scanner/src/lib.rs

@@ -53,6 +53,7 @@ use tokio_util::sync::CancellationToken;
 
 pub mod data_usage_define;
 pub mod error;
+pub mod prefix_usage;
 mod remote_scanner;
 pub mod runtime_config;
 pub mod scanner;
@@ -64,6 +65,7 @@ pub(crate) mod storage_api;
 
 pub use data_usage_define::*;
 pub use error::ScannerError;
+pub use prefix_usage::{BucketPrefixUsageResponse, bucket_prefix_usage, invalidate_prefix_usage_cache};
 pub use remote_scanner::{
     NS_SCANNER_MAX_REQUEST_BODY_SIZE, RemoteScannerAdmission, RemoteScannerRequest, admit_remote_scanner_request,
     claim_remote_scanner_request, decode_remote_scanner_request, preflight_remote_scanner_request,

crates/scanner/src/prefix_usage.rs

@@ -0,0 +1,349 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Prefix-level bucket usage for admin/console consumers (rustfs/backlog#1872,
+//! MinIO `loadPrefixUsageFromBackend` parity).
+//!
+//! The per-bucket, per-set `.usage-cache.bin` objects already hold a
+//! path-keyed prefix tree; this module reads every set's copy through that
+//! set's own object layer (the hash-routed store path would always land on
+//! one set), aggregates the overlapping trees, and serves the result from a
+//! bounded 30-second cache. Bucket writes poke the cache through the
+//! dirty-usage hook so a fresh scan is visible immediately.
+
+use crate::data_usage_define::{DATA_USAGE_CACHE_NAME, DataUsageCache};
+use crate::error::ScannerError;
+use crate::storage_api::owner::{
+    EcstoreSetDisks, EcstoreStore, ecstore_is_reserved_or_invalid_bucket, ecstore_resolve_object_store_handle,
+};
+use futures::future::join_all;
+use rustfs_data_usage::{PrefixUsageEntry, PrefixUsageSummary};
+use std::collections::HashMap;
+use std::sync::{Arc, Mutex};
+use std::time::{Duration, SystemTime};
+use tracing::{debug, warn};
+
+const LOG_COMPONENT_SCANNER: &str = "scanner";
+const LOG_SUBSYSTEM_PREFIX_USAGE: &str = "prefix_usage";
+const EVENT_PREFIX_USAGE_CACHE_STATE: &str = "prefix_usage_cache_state";
+
+/// How long a computed breakdown stays fresh. MinIO uses the same 30s for
+/// its prefix-usage cache; bucket writes additionally invalidate on the spot.
+const CACHE_TTL: Duration = Duration::from_secs(30);
+/// Hard entry cap for the result cache; exceeded, expired entries go first
+/// and the map clears rather than growing past the bound.
+const CACHE_MAX_ENTRIES: usize = 128;
+/// Per-set cache read budget. The underlying loader retries for up to a
+/// minute per attempt on backend errors — far too long for an admin GET, so
+/// a slow set degrades to "not reporting" instead of stalling the caller.
+const PER_SET_LOAD_TIMEOUT: Duration = Duration::from_secs(5);
+
+/// Aggregated prefix-usage answer across every erasure set.
+#[derive(Clone, Debug, PartialEq, serde::Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct BucketPrefixUsageResponse {
+    pub bucket: String,
+    pub prefix: String,
+    pub usage: PrefixUsageSummary,
+    /// Every reporting set's prefix entry was compacted: the aggregate is
+    /// valid, the sub-prefix breakdown is empty on disk.
+    pub compacted: bool,
+    /// The sub-prefix breakdown is incomplete: at least one reporting set
+    /// had the prefix compacted (or absent while others found it), so its
+    /// objects cannot be attributed to a sub-prefix.
+    pub sub_prefixes_partial: bool,
+    /// The breakdown exceeded the caller's entry limit; largest remain.
+    pub truncated: bool,
+    pub sub_prefixes: Vec<PrefixUsageEntry>,
+    /// Sets whose cache held this bucket and prefix.
+    pub sets_reporting: usize,
+    pub sets_total: usize,
+    /// Newest `last_update` across reporting sets, unix seconds.
+    pub last_update_unix_secs: Option<u64>,
+}
+
+#[derive(Clone)]
+struct CachedResponse {
+    computed_at: std::time::Instant,
+    response: Arc<BucketPrefixUsageResponse>,
+}
+
+/// Cache key: (lowercased bucket, normalized prefix, max entries).
+type PrefixUsageCacheKey = (String, String, usize);
+type PrefixUsageCacheMap = Option<HashMap<PrefixUsageCacheKey, CachedResponse>>;
+
+static PREFIX_USAGE_CACHE: Mutex<PrefixUsageCacheMap> = Mutex::new(None);
+
+/// Drop cached results for `bucket` (empty string clears everything). Wired
+/// into the dirty-usage recording path so a write makes the next prefix
+/// query recompute instead of serving up to `CACHE_TTL` seconds of stale
+/// numbers.
+pub fn invalidate_prefix_usage_cache(bucket: &str) {
+    let mut guard = PREFIX_USAGE_CACHE.lock().unwrap_or_else(|poison| poison.into_inner());
+    let Some(map) = guard.as_mut() else {
+        return;
+    };
+    if bucket.is_empty() {
+        map.clear();
+        return;
+    }
+    map.retain(|(cached_bucket, ..), _| !cached_bucket.eq_ignore_ascii_case(bucket));
+}
+
+/// Query prefix usage for `bucket` (arbitrary `prefix`, empty = whole
+/// bucket), merging every erasure set's own cache copy. `max_entries` bounds
+/// the sub-prefix rows (largest first).
+pub async fn bucket_prefix_usage(
+    bucket: &str,
+    prefix: &str,
+    max_entries: usize,
+) -> Result<BucketPrefixUsageResponse, ScannerError> {
+    if ecstore_is_reserved_or_invalid_bucket(bucket, true) {
+        return Err(ScannerError::Other(format!("invalid bucket name: {bucket}")));
+    }
+    let normalized_prefix = prefix.trim_matches('/').to_string();
+    let cache_key = (bucket.to_ascii_lowercase(), normalized_prefix.clone(), max_entries);
+    if let Some(response) = lookup_cached(&cache_key) {
+        return Ok((*response).clone());
+    }
+
+    let store = ecstore_resolve_object_store_handle()
+        .ok_or_else(|| ScannerError::Other("object store is not initialized".to_string()))?;
+    let response = Arc::new(compute_prefix_usage(store, bucket, &normalized_prefix, max_entries).await);
+    store_cached(cache_key, response.clone());
+    Ok((*response).clone())
+}
+
+async fn compute_prefix_usage(
+    store: Arc<EcstoreStore>,
+    bucket: &str,
+    prefix: &str,
+    max_entries: usize,
+) -> BucketPrefixUsageResponse {
+    let sets: Vec<Arc<EcstoreSetDisks>> = store.all_set_disks();
+    let sets_total = sets.len();
+    let cache_name = format!("{bucket}/{DATA_USAGE_CACHE_NAME}");
+
+    let per_set = join_all(sets.into_iter().map(|set| {
+        let cache_name = cache_name.clone();
+        async move {
+            let mut cache = DataUsageCache::default();
+            // A set that has never scanned this bucket (or cannot be read
+            // within the budget) reports nothing — the remaining sets still
+            // produce a usable, flagged answer.
+            let loaded = match tokio::time::timeout(PER_SET_LOAD_TIMEOUT, cache.load(set, &cache_name)).await {
+                Ok(Ok(())) => cache,
+                Ok(Err(err)) => {
+                    debug!(
+                        target: "rustfs::scanner::prefix_usage",
+                        event = EVENT_PREFIX_USAGE_CACHE_STATE,
+                        component = LOG_COMPONENT_SCANNER,
+                        subsystem = LOG_SUBSYSTEM_PREFIX_USAGE,
+                        bucket = %bucket,
+                        state = "set_load_failed",
+                        error = %err,
+                        "Prefix usage set cache load failed"
+                    );
+                    return None;
+                }
+                Err(_) => {
+                    warn!(
+                        target: "rustfs::scanner::prefix_usage",
+                        event = EVENT_PREFIX_USAGE_CACHE_STATE,
+                        component = LOG_COMPONENT_SCANNER,
+                        subsystem = LOG_SUBSYSTEM_PREFIX_USAGE,
+                        bucket = %bucket,
+                        state = "set_load_timeout",
+                        "Prefix usage set cache load timed out"
+                    );
+                    return None;
+                }
+            };
+            if loaded.info.name != bucket {
+                // Empty or stale-scoped cache: this set has no data for the bucket.
+                return None;
+            }
+            let last_update = loaded.info.last_update;
+            let query = loaded.prefix_usage(bucket, prefix, max_entries);
+            Some((query, last_update))
+        }
+    }))
+    .await;
+
+    let mut usage = PrefixUsageSummary::default();
+    let mut sub_prefix_map: HashMap<String, PrefixUsageSummary> = HashMap::new();
+    let mut sets_reporting = 0usize;
+    let mut reporting_but_absent = 0usize;
+    let mut any_compacted = false;
+    let mut all_compacted = true;
+    let mut truncated = false;
+    let mut last_update: Option<SystemTime> = None;
+
+    for (query, set_last_update) in per_set.into_iter().flatten() {
+        // last_update counts every set that has scanned the bucket, even
+        // when the prefix itself is absent on that set.
+        if let Some(set_last_update) = set_last_update
+            && last_update.map(|current| set_last_update > current).unwrap_or(true)
+        {
+            last_update = Some(set_last_update);
+        }
+        let Some(query) = query else {
+            // The set knows the bucket but not this prefix: legitimate when
+            // the prefix's objects all hash to other sets, but it means the
+            // breakdown below cannot attribute that set's (zero) objects.
+            reporting_but_absent += 1;
+            continue;
+        };
+        sets_reporting += 1;
+        usage.merge(&query.usage);
+        if query.compacted {
+            any_compacted = true;
+        } else {
+            all_compacted = false;
+        }
+        truncated |= query.truncated;
+        for entry in query.sub_prefixes {
+            sub_prefix_map.entry(entry.prefix).or_default().merge(&entry.usage);
+        }
+    }
+
+    let mut sub_prefixes: Vec<PrefixUsageEntry> = sub_prefix_map
+        .into_iter()
+        .map(|(prefix, usage)| PrefixUsageEntry { prefix, usage })
+        .collect();
+    sub_prefixes.sort_by(|left, right| {
+        right
+            .usage
+            .size
+            .cmp(&left.usage.size)
+            .then_with(|| left.prefix.cmp(&right.prefix))
+    });
+    // Merged rows can exceed max_entries only when per-set truncation
+    // already flagged; enforce the caller bound on the merged view too.
+    if sub_prefixes.len() > max_entries {
+        truncated = true;
+        sub_prefixes.truncate(max_entries);
+    }
+
+    let found = sets_reporting > 0;
+    BucketPrefixUsageResponse {
+        bucket: bucket.to_string(),
+        prefix: prefix.to_string(),
+        usage,
+        compacted: found && all_compacted,
+        sub_prefixes_partial: any_compacted || reporting_but_absent > 0,
+        truncated,
+        sub_prefixes,
+        sets_reporting,
+        sets_total,
+        last_update_unix_secs: last_update
+            .and_then(|time| time.duration_since(SystemTime::UNIX_EPOCH).ok())
+            .map(|dur| dur.as_secs()),
+    }
+}
+
+fn lookup_cached(key: &(String, String, usize)) -> Option<Arc<BucketPrefixUsageResponse>> {
+    let mut guard = PREFIX_USAGE_CACHE.lock().unwrap_or_else(|poison| poison.into_inner());
+    let map = guard.as_mut()?;
+    let cached = map.get(key)?;
+    if cached.computed_at.elapsed() > CACHE_TTL {
+        map.remove(key);
+        return None;
+    }
+    Some(cached.response.clone())
+}
+
+fn store_cached(key: (String, String, usize), response: Arc<BucketPrefixUsageResponse>) {
+    let mut guard = PREFIX_USAGE_CACHE.lock().unwrap_or_else(|poison| poison.into_inner());
+    let map = guard.get_or_insert_with(HashMap::new);
+    // Bound the cache: drop expired entries first, and if the cap is still
+    // exceeded clear wholesale — the next queries recompute in milliseconds.
+    if map.len() >= CACHE_MAX_ENTRIES {
+        map.retain(|_, cached| cached.computed_at.elapsed() <= CACHE_TTL);
+        if map.len() >= CACHE_MAX_ENTRIES {
+            map.clear();
+        }
+    }
+    map.insert(
+        key,
+        CachedResponse {
+            computed_at: std::time::Instant::now(),
+            response,
+        },
+    );
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{CACHE_MAX_ENTRIES, PREFIX_USAGE_CACHE, invalidate_prefix_usage_cache, store_cached};
+    use rustfs_data_usage::PrefixUsageSummary;
+
+    fn response(bucket: &str) -> super::BucketPrefixUsageResponse {
+        super::BucketPrefixUsageResponse {
+            bucket: bucket.to_string(),
+            prefix: String::new(),
+            usage: PrefixUsageSummary::default(),
+            compacted: false,
+            sub_prefixes_partial: false,
+            truncated: false,
+            sub_prefixes: Vec::new(),
+            sets_reporting: 1,
+            sets_total: 1,
+            last_update_unix_secs: None,
+        }
+    }
+
+    fn seed(bucket: &str, prefix: &str) {
+        store_cached(
+            (bucket.to_ascii_lowercase(), prefix.to_string(), 10),
+            std::sync::Arc::new(response(bucket)),
+        );
+    }
+
+    fn contains(bucket: &str, prefix: &str) -> bool {
+        PREFIX_USAGE_CACHE
+            .lock()
+            .unwrap_or_else(|poison| poison.into_inner())
+            .as_ref()
+            .is_some_and(|map| map.contains_key(&(bucket.to_ascii_lowercase(), prefix.to_string(), 10)))
+    }
+
+    /// All cache tests run inside one test to keep the process-global map
+    /// free of cross-test ordering (the flake class this module avoids).
+    #[test]
+    fn invalidation_scopes_to_bucket_and_cache_stays_bounded() {
+        invalidate_prefix_usage_cache("");
+        seed("alpha", "x");
+        seed("beta", "y");
+
+        // Case-insensitive bucket scoping.
+        invalidate_prefix_usage_cache("ALPHA");
+        assert!(!contains("alpha", "x"));
+        assert!(contains("beta", "y"));
+
+        // Wholesale clear.
+        invalidate_prefix_usage_cache("");
+        assert!(!contains("beta", "y"));
+
+        // Hard cap: overflow clears rather than grows.
+        for index in 0..=(CACHE_MAX_ENTRIES / 2) {
+            let bucket = format!("cap-bucket-{index}");
+            seed(&bucket, "a");
+            seed(&bucket, "b");
+        }
+        let guard = PREFIX_USAGE_CACHE.lock().unwrap_or_else(|poison| poison.into_inner());
+        let map = guard.as_ref().expect("seeded");
+        assert!(map.len() <= CACHE_MAX_ENTRIES, "cache must stay bounded, got {}", map.len());
+    }
+}

crates/scanner/src/scanner_io.rs

@@ -231,6 +231,10 @@ pub fn record_dirty_usage_bucket(bucket: &str) {
         dirty_buckets.len()
     };
     global_metrics().record_scanner_dirty_usage_pending(usize_to_u64_saturated(pending_buckets));
+    // A write invalidates this bucket's prefix-usage answers on the spot so
+    // admin/console consumers never ride the full TTL after a change
+    // (rustfs/backlog#1872).
+    crate::prefix_usage::invalidate_prefix_usage_cache(bucket);
     DIRTY_USAGE_BUCKET_NOTIFY.notify_one();
 }
 

crates/scanner/tests/lifecycle_integration_test.rs

@@ -206,7 +206,6 @@ async fn setup_isolated_test_env(init_expiry: bool) -> (Vec<PathBuf>, Arc<ECStor
 }
 
 /// Test helper: Create a test bucket
-#[allow(dead_code)]
 async fn create_test_bucket(ecstore: &Arc<ECStore>, bucket_name: &str) {
     (**ecstore)
         .make_bucket(bucket_name, &Default::default())
@@ -251,7 +250,6 @@ async fn modeled_versioned_delete_opts(bucket: &str, object: &str) -> ObjectOpti
 }
 
 /// Test helper: Set bucket lifecycle configuration
-#[allow(dead_code)]
 async fn set_bucket_lifecycle(bucket_name: &str) -> Result<(), Box<dyn std::error::Error>> {
     // Create a simple lifecycle configuration XML with 0 days expiry for immediate testing
     let lifecycle_xml = r#"<?xml version="1.0" encoding="UTF-8"?>
@@ -274,7 +272,6 @@ async fn set_bucket_lifecycle(bucket_name: &str) -> Result<(), Box<dyn std::erro
 }
 
 /// Test helper: Set bucket lifecycle configuration
-#[allow(dead_code)]
 async fn set_bucket_lifecycle_deletemarker(bucket_name: &str) -> Result<(), Box<dyn std::error::Error>> {
     // Create lifecycle rule that targets delete-marker cleanup only.
     // Keep Expiration.Days unset to avoid expiring live transitioned object versions.
@@ -297,7 +294,6 @@ async fn set_bucket_lifecycle_deletemarker(bucket_name: &str) -> Result<(), Box<
     Ok(())
 }
 
-#[allow(dead_code)]
 async fn set_bucket_lifecycle_delmarker_expiration(bucket_name: &str, days: i64) -> Result<(), Box<dyn std::error::Error>> {
     let lifecycle_xml = format!(
         r#"<?xml version="1.0" encoding="UTF-8"?>
@@ -320,7 +316,6 @@ async fn set_bucket_lifecycle_delmarker_expiration(bucket_name: &str, days: i64)
     Ok(())
 }
 
-#[allow(dead_code)]
 async fn set_bucket_lifecycle_transition_with_tier(
     bucket_name: &str,
     storage_class: &str,
@@ -368,7 +363,6 @@ async fn object_exists(ecstore: &Arc<ECStore>, bucket: &str, object: &str) -> bo
 }
 
 /// Test helper: Check if object exists
-#[allow(dead_code)]
 async fn object_is_delete_marker(ecstore: &Arc<ECStore>, bucket: &str, object: &str) -> bool {
     if let Ok(oi) = (**ecstore).get_object_info(bucket, object, &ObjectOptions::default()).await {
         println!("oi: {oi:?}");
@@ -379,7 +373,6 @@ async fn object_is_delete_marker(ecstore: &Arc<ECStore>, bucket: &str, object: &
     }
 }
 
-#[allow(dead_code)]
 async fn wait_for_object_absence(ecstore: &Arc<ECStore>, bucket: &str, object: &str, timeout: Duration) -> bool {
     let deadline = tokio::time::Instant::now() + timeout;
 

crates/targets/src/net.rs

@@ -428,7 +428,6 @@ pub fn parse_url(s: &str) -> Result<ParsedURL, NetError> {
     Ok(ParsedURL(uu))
 }
 
-#[allow(dead_code)]
 pub fn parse_http_url(s: &str) -> Result<ParsedURL, NetError> {
     let u = parse_url(s)?;
     match u.0.scheme() {
@@ -437,7 +436,6 @@ pub fn parse_http_url(s: &str) -> Result<ParsedURL, NetError> {
     }
 }
 
-#[allow(dead_code)]
 pub fn is_network_or_host_down(err: &std::io::Error, expect_timeouts: bool) -> bool {
     if err.kind() == std::io::ErrorKind::TimedOut {
         return !expect_timeouts;
@@ -449,12 +447,10 @@ pub fn is_network_or_host_down(err: &std::io::Error, expect_timeouts: bool) -> b
         || err_str.contains("use of closed network connection")
 }
 
-#[allow(dead_code)]
 pub fn is_conn_reset_err(err: &std::io::Error) -> bool {
     err.to_string().contains("connection reset by peer") || matches!(err.raw_os_error(), Some(libc::ECONNRESET))
 }
 
-#[allow(dead_code)]
 pub fn is_conn_refused_err(err: &std::io::Error) -> bool {
     err.to_string().contains("connection refused") || matches!(err.raw_os_error(), Some(libc::ECONNREFUSED))
 }

helm/README.md

@@ -273,6 +273,10 @@ uer. `ClusterIssuer` or `Issuer`. |
 | gatewayApi.listeners.http.port| int | `8000` | Gateway API http listener port. |
 | gatewayApi.listeners.https.name | string | `websecure` | Gateway API https listener name. |
 | gatewayApi.listeners.https.port| int | `8443` | Gateway API https listener port. |
+| gatewayApi.listeners.tls.enabled | bool | `false` | Enable a TLS passthrough listener and generate a TLSRoute. |
+| gatewayApi.listeners.tls.name | string | `tls` | Gateway API TLS passthrough listener name. |
+| gatewayApi.listeners.tls.port | int | `443` | Gateway API TLS passthrough listener port. |
+| gatewayApi.listeners.tls.backendPort | int | `null` | Backend service port that terminates TLS; defaults to the console port. |
 | gatewayApi.hostname | string | Hostname to access RustFS via gateway api. |
 | gatewayApi.secretName | string | Secret tls to via RustFS using HTTPS. |
 | gatewayApi.existingGateway.name | string | `""` |  The existing gateway name, instead of creating a new one. |
@@ -447,6 +451,8 @@ rustfs-route   ["example.rustfs.com"]   172m
 
 Then, via RustFS instance via `https://example.rustfs.com` or `http://example.rustfs.com`.
 
+For end-to-end encryption, set `gatewayApi.listeners.tls.enabled` to `true`. The chart then adds a `TLS` listener with `tls.mode: Passthrough` to the `Gateway` and generates a `TLSRoute` that forwards the encrypted stream to the RustFS service, where TLS is terminated on the backend side. Note that backend TLS termination must be configured on RustFS itself (for example `RUSTFS_TLS_PATH` pointing to server certificates), and the installed Gateway API CRDs must include `TLSRoute`.
+
 # Uninstall
 
 Uninstalling the rustfs installation with command,

helm/rustfs/templates/gateway-api/gateway.yml

@@ -26,5 +26,15 @@ spec:
           - name: {{ include "rustfs.fullname" $ }}-tls
             kind: Secret
   {{- end }}
+  {{- if .tls.enabled }}
+    - name: {{ .tls.name }}
+      port: {{ .tls.port }}
+      protocol: TLS
+      tls:
+        mode: Passthrough
+      allowedRoutes:
+        namespaces:
+          from: Same
+  {{- end }}
   {{- end }}
 {{- end }}

helm/rustfs/templates/gateway-api/tlsroute.yml

@@ -0,0 +1,25 @@
+{{- if and .Values.gatewayApi.enabled .Values.gatewayApi.listeners.tls.enabled }}
+apiVersion: gateway.networking.k8s.io/v1
+kind: TLSRoute
+metadata:
+  name: {{ include "rustfs.fullname" . }}-tlsroute
+  namespace: {{ .Release.Namespace }}
+spec:
+  parentRefs:
+    {{- if .Values.gatewayApi.existingGateway.name }}
+    - name: {{ .Values.gatewayApi.existingGateway.name }}
+      {{- if .Values.gatewayApi.existingGateway.namespace }}
+      namespace: {{ .Values.gatewayApi.existingGateway.namespace }}
+      {{- end }}
+      sectionName: {{ .Values.gatewayApi.listeners.tls.name }}
+    {{- else }}
+    - name: {{ include "rustfs.fullname" $ }}-gateway
+      sectionName: {{ .Values.gatewayApi.listeners.tls.name }}
+    {{- end }}
+  hostnames:
+    - {{ .Values.gatewayApi.hostname }}
+  rules:
+    - backendRefs:
+        - name: {{ include "rustfs.fullname" . }}-svc
+          port: {{ .Values.gatewayApi.listeners.tls.backendPort | default .Values.service.console.port }}
+{{- end }}

helm/rustfs/values.yaml

@@ -369,6 +369,12 @@ gatewayApi:
     https:
       name: websecure
       port: 8443
+    tls: # Optional TLS passthrough listener; renders a TLSRoute so TLS terminates at the RustFS backend.
+      enabled: false
+      name: tls
+      port: 443
+      # Service port that terminates TLS on the backend; defaults to the console port.
+      backendPort: null
   hostname: example.rustfs.com
   httpToHttpsRedirect: true
   existingGateway:

rustfs/src/admin/handlers/mod.rs

@@ -64,6 +64,7 @@ mod target_descriptor;
 pub mod tier;
 pub mod tls_debug;
 pub mod trace;
+pub mod usage_prefix;
 pub mod user;
 pub mod user_iam;
 pub mod user_lifecycle;

rustfs/src/admin/handlers/site_replication.rs

@@ -6023,7 +6023,10 @@ fn edit_generation_wall_clock() -> u64 {
 /// node's clock behind the clock that fed the previous lifetime) mints
 /// below the stale mark and the origin stays fenced — but only until real
 /// time passes the previous lifetime's last allocation, because every later
-/// allocation takes the wall-clock floor again. Bounded by the skew,
+/// allocation takes the wall-clock floor again (and never longer than
+/// [`PEER_EDIT_FENCE_STALENESS_WINDOW_NANOS`]: a regression past the window
+/// leaves the mark implausibly distant and the origin runs unfenced
+/// immediately). Bounded by the skew,
 /// self-healing, and no rollback window beyond the plain counter's: a
 /// delivery applies only at or above the receiver's mark, so the one
 /// cross-lifetime interleaving that can apply stale content — a
@@ -6063,6 +6066,52 @@ fn peer_edit_fence(queries: &HashMap<String, String>) -> Option<(String, u64)> {
     Some((origin.clone(), generation))
 }
 
+/// How far below the recorded high-water mark a delivery may sit and still
+/// be fenced as stale. The distance a GENUINE superseded delivery can trail
+/// its origin's mark is small: retransmissions re-run the sender flow and
+/// mint a fresh generation (the retry queue keys on the bare path and never
+/// replays a fenced URL), so only an in-flight straggler of the losing
+/// fan-out race trails the mark, by delivery latency — minutes at the
+/// outside. A mark further above than this window cannot be explained by
+/// any genuine race, only by a forged fence (the shared service account
+/// lets any peer stamp any origin) or by a persisted clock excursion the
+/// origin has since left behind — and fencing on it would silently drop the
+/// origin's real edits, so the stale check ignores it instead.
+const PEER_EDIT_FENCE_STALENESS_WINDOW_NANOS: u64 = 24 * 60 * 60 * 1_000_000_000;
+
+/// Whether an incoming fence may be honoured, as far as this site can vouch
+/// for it. The sender's identity is unverifiable (shared service account),
+/// so the check runs over what the receiving state knows: the claimed origin
+/// must be a site this state currently replicates with — the same membership
+/// rule the load-time mark pruning applies, so every mark recorded behind
+/// this check is one a reload would keep — and not this site itself, which
+/// never delivers edits to itself. The caller IGNORES an inadmissible fence
+/// rather than failing the request: the delivery applies exactly as an
+/// unstamped (pre-fence) delivery would, no high-water mark is read or
+/// written, and the worst a forged fence achieves is forfeiting an ordering
+/// guarantee its sender was never owed. The generation itself is NOT
+/// bounded here: a genuine origin whose hybrid clock persisted a wall-clock
+/// excursion allocates arbitrarily far in the future, and refusing to
+/// record its marks would strip the ordering fence from exactly the
+/// deliveries that still race — the staleness window on the read side is
+/// what defuses forged marks instead.
+fn peer_edit_fence_is_admissible(state: &SiteReplicationState, local_deployment_id: &str, fence: &(String, u64)) -> bool {
+    let (origin, generation) = fence;
+    if origin != local_deployment_id && state.peers.contains_key(origin) {
+        return true;
+    }
+    warn!(
+        event = EVENT_ADMIN_SITE_REPLICATION_STATE,
+        component = LOG_COMPONENT_ADMIN,
+        subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
+        result = "fence_origin_not_a_remote_peer",
+        origin = %origin,
+        generation = *generation,
+        "ignoring inadmissible peer-edit fence"
+    );
+    false
+}
+
 /// True when a strictly newer edit from the same origin site already landed
 /// here. No lock on the sending side can order deliveries issued by two
 /// nodes of that site, so ordering is decided here, on the generation the
@@ -6070,11 +6119,42 @@ fn peer_edit_fence(queries: &HashMap<String, String>) -> Option<(String, u64)> {
 /// stale: one edit legitimately fans out several deliveries under a single
 /// generation (the ILM-expiry edit sends every peer's record), and a replay of
 /// an applied delivery re-applies the same edit idempotently.
+///
+/// A mark more than [`PEER_EDIT_FENCE_STALENESS_WINDOW_NANOS`] above the
+/// delivery is implausible and does NOT fence: the shared service account
+/// means any peer can stamp any origin, so a forged `u64::MAX`-scale mark
+/// would otherwise silently swallow the origin's genuine edits for good.
+/// Bounding the fence by distance instead of by an absolute ceiling keeps
+/// ordering intact wherever the origin's clock actually operates — two
+/// racing deliveries trail each other by seconds whether the hybrid clock
+/// tracks wall time or persists a long-gone excursion far ahead of it —
+/// while a mark no genuine race can explain merely downgrades the origin to
+/// unfenced (pre-fence) delivery instead of dropping its edits. (One genuine
+/// shape does land out here: a plain-counter straggler arriving after its
+/// origin's first hybrid-clock edit. It gets the same downgrade — applied
+/// unfenced — once, at upgrade time; fencing it instead would silence the
+/// mirror case, a hybrid-clock origin downgraded back to the plain counter.)
 fn peer_edit_delivery_is_stale(state: &SiteReplicationState, origin: &str, generation: u64) -> bool {
-    state
-        .applied_edit_generations
-        .get(origin)
-        .is_some_and(|applied| *applied > generation)
+    let Some(applied) = state.applied_edit_generations.get(origin) else {
+        return false;
+    };
+    if *applied <= generation {
+        return false;
+    }
+    if *applied - generation > PEER_EDIT_FENCE_STALENESS_WINDOW_NANOS {
+        warn!(
+            event = EVENT_ADMIN_SITE_REPLICATION_STATE,
+            component = LOG_COMPONENT_ADMIN,
+            subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
+            result = "fence_mark_beyond_staleness_window",
+            origin,
+            generation,
+            applied_mark = *applied,
+            "ignoring implausibly distant peer-edit high-water mark"
+        );
+        return false;
+    }
+    true
 }
 
 fn record_applied_peer_edit_generation(state: &mut SiteReplicationState, origin: &str, generation: u64) {
@@ -10698,6 +10778,11 @@ impl Operation for SRPeerEditHandler {
         let outcome = update_site_replication_state_when_changed(move |state| {
             let mut incoming = incoming;
             let local_peer = local_peer_at_endpoint(commit_endpoint, state);
+            // The fence is self-reported — the shared service account means
+            // the sender cannot be identified — so it is honoured only after
+            // the admissibility check, against the same state it will gate.
+            let commit_fence =
+                commit_fence.filter(|fence| peer_edit_fence_is_admissible(state, &local_peer.deployment_id, fence));
             // Ordering fence: the sending site allocates the generation under
             // its state-object lock, so a delivery that lost the race carries
             // a generation this site has already passed. Applying it would
@@ -13393,6 +13478,15 @@ mod tests {
             handler_block.contains("record_applied_peer_edit_generation(state, origin, *generation);"),
             "SRPeerEditHandler must record the applied generation so later stale deliveries are recognised"
         );
+        // Fence hardening: origin and generation are self-reported by a
+        // caller the shared service account cannot identify, so the handler
+        // must pass the fence through the admissibility check — against the
+        // same state the fence gates, i.e. inside the transaction — before
+        // reading or raising any high-water mark.
+        assert!(
+            handler_block.contains(".filter(|fence| peer_edit_fence_is_admissible(state, &local_peer.deployment_id, fence))"),
+            "SRPeerEditHandler must admit a fence only through peer_edit_fence_is_admissible inside the state transaction"
+        );
         // P1-15 PR2: both halves of the fence and the edit they fence share
         // ONE transaction. Checking the fence against a state read outside the
         // lock would let the check pass on one snapshot and the write land on
@@ -14769,6 +14863,121 @@ mod tests {
         assert!(peer_edit_delivery_is_stale(&state, origin, generation - 1));
     }
 
+    /// A fence is self-reported: every site authenticates peer traffic with
+    /// the same site-replicator credential, so a compromised peer can stamp
+    /// ANY origin with ANY generation. An origin the receiver does not
+    /// replicate with — or the receiver itself — is ignored and plants no
+    /// mark; a mark a compromised peer plants for a CURRENT origin cannot
+    /// silence that origin, because the staleness window refuses to fence on
+    /// a mark implausibly far above the genuine deliveries.
+    #[test]
+    fn forged_peer_edit_fences_cannot_poison_the_high_water_marks() {
+        let mut state = SiteReplicationState {
+            peers: BTreeMap::from([
+                (
+                    "site-local".to_string(),
+                    PeerInfo {
+                        deployment_id: "site-local".to_string(),
+                        ..peer("local", "https://local.example:9000")
+                    },
+                ),
+                (
+                    "site-victim".to_string(),
+                    PeerInfo {
+                        deployment_id: "site-victim".to_string(),
+                        ..peer("victim", "https://victim.example:9000")
+                    },
+                ),
+            ]),
+            ..Default::default()
+        };
+        // An origin outside the current membership is refused outright...
+        let unknown = ("site-unknown".to_string(), 4u64);
+        assert!(!peer_edit_fence_is_admissible(&state, "site-local", &unknown));
+
+        // No site delivers edits to itself: a fence claiming the receiver as
+        // its origin is forged by construction, current peer or not.
+        let own = ("site-local".to_string(), 4u64);
+        assert!(!peer_edit_fence_is_admissible(&state, "site-local", &own));
+
+        // A current remote peer's fence is admitted and works end to end.
+        let genuine = ("site-victim".to_string(), 1u64);
+        assert!(peer_edit_fence_is_admissible(&state, "site-local", &genuine));
+        assert!(!peer_edit_delivery_is_stale(&state, &genuine.0, genuine.1));
+        record_applied_peer_edit_generation(&mut state, &genuine.0, genuine.1);
+        assert_eq!(state.applied_edit_generations.get("site-victim"), Some(&1));
+
+        // A forged u64::MAX-scale mark CAN be recorded — the shared service
+        // account means the receiver cannot tell the stamp was forged — but
+        // it is inert: the victim's genuine hybrid-clock deliveries sit far
+        // more than the staleness window below it, so they keep applying
+        // instead of being silently acked-and-dropped.
+        record_applied_peer_edit_generation(&mut state, "site-victim", u64::MAX);
+        assert!(!peer_edit_delivery_is_stale(&state, "site-victim", edit_generation_wall_clock()));
+    }
+
+    /// The staleness window bounds the fence by DISTANCE from the mark, not
+    /// by an absolute clock ceiling, so ordering must hold wherever the
+    /// origin's hybrid clock actually operates. The regression that matters:
+    /// a temporary wall-clock excursion far in the future is persisted by
+    /// `next_peer_edit_generation` (`max(now, prev + 1)` never comes back
+    /// down), and two later edits g+1 then g can arrive in reverse order —
+    /// g must still be fenced, even though both generations dwarf the
+    /// receiver's clock. Conversely a mark further above a delivery than any
+    /// genuine race can explain must not fence it.
+    #[test]
+    fn peer_edit_fence_orders_a_persisted_future_clock_and_defuses_distant_marks() {
+        let mut state = SiteReplicationState {
+            peers: BTreeMap::from([(
+                "site-origin".to_string(),
+                PeerInfo {
+                    deployment_id: "site-origin".to_string(),
+                    ..peer("origin", "https://origin.example:9000")
+                },
+            )]),
+            ..Default::default()
+        };
+
+        // The origin's clock once jumped ten years ahead; the hybrid clock
+        // keeps allocating from there long after the clock was corrected.
+        let excursion = edit_generation_wall_clock() + 10 * 365 * 24 * 60 * 60 * 1_000_000_000;
+        let fence = ("site-origin".to_string(), excursion + 1);
+        assert!(peer_edit_fence_is_admissible(&state, "site-local", &fence));
+        record_applied_peer_edit_generation(&mut state, &fence.0, fence.1);
+
+        // The reverse delivery of the race: g arrives after g+1 landed.
+        // Without the fence it would commit last and roll g+1 back.
+        assert!(peer_edit_delivery_is_stale(&state, "site-origin", excursion));
+        // Equal generation (same edit's fan-out or a replay) still applies,
+        // as does the next edit.
+        assert!(!peer_edit_delivery_is_stale(&state, "site-origin", excursion + 1));
+        assert!(!peer_edit_delivery_is_stale(&state, "site-origin", excursion + 2));
+
+        // The window's exact boundary: a delivery trailing the mark by the
+        // full window is still fenced; one nanosecond further is not — that
+        // distance is no longer explicable by a genuine race, only by a
+        // forged mark or an excursion the origin has left behind.
+        let mark = fence.1;
+        // A straggler trailing by a concrete hour must still be fenced —
+        // pins the window's real magnitude, not just its symbolic boundary.
+        assert!(peer_edit_delivery_is_stale(&state, "site-origin", mark - 60 * 60 * 1_000_000_000));
+        assert!(peer_edit_delivery_is_stale(
+            &state,
+            "site-origin",
+            mark - PEER_EDIT_FENCE_STALENESS_WINDOW_NANOS
+        ));
+        assert!(!peer_edit_delivery_is_stale(
+            &state,
+            "site-origin",
+            mark - PEER_EDIT_FENCE_STALENESS_WINDOW_NANOS - 1
+        ));
+
+        // A pre-hybrid plain-counter origin trails such a mark by eons: it
+        // is not fenced (the rc.2-era downgrade case), it just runs
+        // unfenced until its counter regime catches up.
+        assert!(!peer_edit_delivery_is_stale(&state, "site-origin", 3));
+    }
+
     /// P1-15 review follow-up: a site that leaves the mesh drops below two
     /// peers, which clears its state object and restarts its generation
     /// counter at zero. A mark left over from its previous membership would

rustfs/src/admin/handlers/system.rs

@@ -417,6 +417,13 @@ struct SystemAdminDiscovery {
 struct ServerInfoResponse {
     info: InfoMessage,
     admin_discovery: SystemAdminDiscovery,
+    /// Startup bitrot algorithm self-test outcome (rustfs/backlog#1873):
+    /// `passed` (algorithms verified at boot), `failed` (a drifted hash
+    /// implementation — the process is serving with degraded integrity
+    /// checking unless `RUSTFS_BITROT_SELFTEST_STRICT` aborted it), or
+    /// `unknown` (not yet run or disabled).
+    #[serde(rename = "bitrotSelftest")]
+    bitrot_selftest: &'static str,
 }
 
 #[derive(Serialize)]
@@ -433,6 +440,14 @@ fn system_admin_discovery(usecase: &DefaultAdminUsecase) -> SystemAdminDiscovery
     }
 }
 
+fn bitrot_selftest_status_str() -> &'static str {
+    match crate::bitrot_selftest::bitrot_selftest_passed() {
+        Some(true) => "passed",
+        Some(false) => "failed",
+        None => "unknown",
+    }
+}
+
 #[async_trait::async_trait]
 impl Operation for ServerInfoHandler {
     async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
@@ -464,6 +479,7 @@ impl Operation for ServerInfoHandler {
         let response = ServerInfoResponse {
             info,
             admin_discovery: system_admin_discovery(&usecase),
+            bitrot_selftest: bitrot_selftest_status_str(),
         };
 
         let data = serde_json::to_vec(&response).map_err(|e| {
@@ -1142,10 +1158,10 @@ impl Operation for RuntimeCapabilitiesHandler {
     }
 }
 
-/// Authorization gate for GET datausageinfo: any-of the dedicated admin action
-/// OR the bucket listing action. Pinned by a unit test so the gate cannot
-/// silently narrow or widen (rustfs/backlog#1306).
-fn data_usage_info_gate_actions() -> Vec<Action> {
+/// Authorization gate for GET datausageinfo (and prefix usage): any-of the
+/// dedicated admin action OR the bucket listing action. Pinned by a unit test
+/// so the gate cannot silently narrow or widen (rustfs/backlog#1306).
+pub(crate) fn data_usage_info_gate_actions() -> Vec<Action> {
     vec![
         Action::AdminAction(AdminAction::DataUsageInfoAdminAction),
         Action::S3Action(S3Action::ListBucketAction),
@@ -1535,6 +1551,18 @@ mod tests {
         );
     }
 
+    /// The startup bitrot self-test outcome must surface in server info as one
+    /// of three closed-set strings, never an internal enum or a null
+    /// (rustfs/backlog#1873). This test pins the string mapping; whether the
+    /// process-global cell holds Some(true)/Some(false)/None is owned by
+    /// `crate::bitrot_selftest`'s own tests.
+    #[test]
+    fn bitrot_selftest_status_str_is_a_closed_set_of_operators_strings() {
+        let rendered = super::bitrot_selftest_status_str();
+        assert!(matches!(rendered, "passed" | "failed" | "unknown"));
+        assert_eq!(super::bitrot_selftest_status_str(), rendered);
+    }
+
     #[test]
     fn server_info_response_exposes_admin_discovery_paths() {
         let usecase = DefaultAdminUsecase::without_context();
@@ -1556,6 +1584,7 @@ mod tests {
                 pools: None,
             },
             admin_discovery: system_admin_discovery(&usecase),
+            bitrot_selftest: super::bitrot_selftest_status_str(),
         };
 
         let value = serde_json::to_value(response).expect("server info response should serialize");

rustfs/src/admin/handlers/usage_prefix.rs

@@ -0,0 +1,142 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Prefix-level bucket usage admin handler (rustfs/backlog#1872).
+//!
+//! `GET /rustfs/admin/v3/usage/{bucket}?prefix=&max-entries=` answers
+//! "what does this bucket / this prefix hold" from the scanner's per-set
+//! 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::handlers::system::data_usage_info_gate_actions;
+use crate::admin::router::{AdminOperation, Operation, S3Router};
+use crate::auth::{check_key_valid, get_session_token};
+use crate::server::{ADMIN_PREFIX, RemoteAddr};
+use http::{HeaderMap, HeaderValue, StatusCode};
+use hyper::Method;
+use matchit::Params;
+use s3s::header::CONTENT_TYPE;
+use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error};
+
+const JSON_CONTENT_TYPE: &str = "application/json";
+const DEFAULT_MAX_ENTRIES: usize = 1000;
+const MAX_ENTRIES_LIMIT: usize = 10_000;
+
+pub struct BucketPrefixUsageHandler {}
+
+pub fn register_usage_prefix_route(r: &mut S3Router<AdminOperation>) -> std::io::Result<()> {
+    r.insert(
+        Method::GET,
+        format!("{}{}", ADMIN_PREFIX, "/v3/usage/{bucket}").as_str(),
+        AdminOperation(&BucketPrefixUsageHandler {}),
+    )?;
+    Ok(())
+}
+
+/// Parse `prefix` and `max-entries` from the query string. Unknown keys are
+/// rejected so a typo'd parameter cannot silently change the answer's shape.
+fn parse_usage_prefix_query(query: Option<&str>) -> S3Result<(String, usize)> {
+    let mut prefix: Option<String> = None;
+    let mut max_entries: Option<usize> = None;
+    for (key, value) in url::form_urlencoded::parse(query.unwrap_or_default().as_bytes()) {
+        match key.as_ref() {
+            "prefix" => prefix = Some(value.into_owned()),
+            "max-entries" => {
+                max_entries = Some(
+                    value
+                        .parse::<usize>()
+                        .map_err(|_| s3_error!(InvalidArgument, "max-entries must be a positive integer"))?,
+                );
+            }
+            other => return Err(s3_error!(InvalidArgument, "unknown query parameter: {other}")),
+        }
+    }
+    let max_entries = max_entries.unwrap_or(DEFAULT_MAX_ENTRIES).clamp(1, MAX_ENTRIES_LIMIT);
+    Ok((prefix.unwrap_or_default(), max_entries))
+}
+
+#[async_trait::async_trait]
+impl Operation for BucketPrefixUsageHandler {
+    async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        let Some(input_cred) = req.credentials else {
+            return Err(s3_error!(InvalidRequest, "get cred failed"));
+        };
+
+        let (cred, owner) =
+            check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
+
+        let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
+        validate_admin_request(&req.headers, &cred, owner, false, data_usage_info_gate_actions(), remote_addr).await?;
+
+        let bucket = params.get("bucket").unwrap_or_default().to_string();
+        if bucket.is_empty() {
+            return Err(s3_error!(InvalidRequest, "bucket path parameter is required"));
+        }
+        let (prefix, max_entries) = parse_usage_prefix_query(req.uri.query())?;
+
+        // Authorization is bucket-scoped by the same any-of gate as the
+        // datausageinfo route; the bucket name itself is validated by the
+        // scanner layer, which rejects reserved/invalid names.
+        let response = rustfs_scanner::bucket_prefix_usage(&bucket, &prefix, max_entries)
+            .await
+            .map_err(|err| s3_error!(InvalidArgument, "{}", err))?;
+
+        let data = serde_json::to_vec(&response)
+            .map_err(|_| S3Error::with_message(S3ErrorCode::InternalError, "parse prefix usage failed"))?;
+        let mut header = HeaderMap::new();
+        header.insert(CONTENT_TYPE, HeaderValue::from_static(JSON_CONTENT_TYPE));
+
+        Ok(S3Response::with_headers((StatusCode::OK, Body::from(data)), header))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{DEFAULT_MAX_ENTRIES, MAX_ENTRIES_LIMIT, parse_usage_prefix_query};
+    use s3s::S3Error;
+
+    fn query(raw: &str) -> Result<(String, usize), S3Error> {
+        parse_usage_prefix_query(Some(raw))
+    }
+
+    #[test]
+    fn defaults_apply_when_no_query_is_given() {
+        assert_eq!(parse_usage_prefix_query(None).unwrap(), (String::new(), DEFAULT_MAX_ENTRIES));
+        assert_eq!(query("").unwrap(), (String::new(), DEFAULT_MAX_ENTRIES));
+    }
+
+    #[test]
+    fn prefix_round_trips_url_encoded_characters() {
+        let (prefix, _) = query("prefix=pre%2Ffix%20name").unwrap();
+        assert_eq!(prefix, "pre/fix name");
+    }
+
+    #[test]
+    fn max_entries_parses_and_clamps_to_documented_bounds() {
+        assert_eq!(query("max-entries=5").unwrap().1, 5);
+        assert_eq!(query("max-entries=0").unwrap().1, 1, "zero must clamp up, not mean unlimited");
+        assert_eq!(query("max-entries=99999999").unwrap().1, MAX_ENTRIES_LIMIT);
+        assert!(query("max-entries=-3").is_err());
+        assert!(query("max-entries=abc").is_err());
+    }
+
+    #[test]
+    fn unknown_parameters_are_rejected_not_ignored() {
+        assert!(
+            query("prefixes=x").is_err(),
+            "a typo'd parameter must fail the request, not widen the query"
+        );
+    }
+}

rustfs/src/admin/mod.rs

@@ -40,7 +40,8 @@ use handlers::{
     audit, batch_job, bucket_meta, cluster_snapshot, config_admin, diagnostics, durability as durability_handler, extensions,
     heal, health, idp_compat, ilm_transition, inspect_archive, kms, module_switch, object_data_cache, object_zip_download, oidc,
     plugins_catalog, plugins_instances, pools, profile_admin, quota as quota_handler, rebalance,
-    replication as replication_handler, scanner, site_replication, sts, system, table_catalog, tier, tls_debug, user,
+    replication as replication_handler, scanner, site_replication, sts, system, table_catalog, tier, tls_debug, usage_prefix,
+    user,
 };
 use router::{AdminOperation, S3Router};
 use s3s::route::S3Route;
@@ -80,6 +81,7 @@ fn register_admin_routes(r: &mut S3Router<AdminOperation>) -> std::io::Result<()
     bucket_meta::register_bucket_meta_route(r)?;
     config_admin::register_config_route(r)?;
     scanner::register_scanner_route(r)?;
+    usage_prefix::register_usage_prefix_route(r)?;
     ilm_transition::register_ilm_transition_route(r)?;
     object_data_cache::register_object_data_cache_route(r)?;
     audit::register_audit_target_route(r)?;

rustfs/src/admin/route_policy.rs

@@ -1558,6 +1558,11 @@ pub const DEFERRED_ADMIN_ROUTE_POLICIES: &[DeferredAdminRoutePolicy] = &[
         "/rustfs/admin/v3/datausageinfo",
         DeferredRoutePolicyReason::MultipleActions,
     ),
+    deferred(
+        HttpMethod::Get,
+        "/rustfs/admin/v3/usage/{bucket}",
+        DeferredRoutePolicyReason::MultipleActions,
+    ),
     deferred(
         HttpMethod::Post,
         "/rustfs/admin/v3/object-zip-downloads",

rustfs/src/admin/route_registration_test.rs

@@ -172,6 +172,7 @@ fn expected_admin_route_matrix() -> Vec<RouteMatrixEntry> {
         admin_route(Method::POST, "/v4/inspect/archive"),
         admin_route(Method::GET, "/v3/storageinfo"),
         admin_route(Method::GET, "/v3/datausageinfo"),
+        admin_route_sample(Method::GET, "/v3/usage/{bucket}", "/v3/usage/test-bucket"),
         admin_route(Method::GET, "/v3/metrics"),
         admin_route(Method::GET, "/v3/object-data-cache/stats"),
         admin_route(Method::POST, "/v3/object-data-cache/flush"),

rustfs/src/bitrot_selftest.rs

@@ -0,0 +1,181 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Startup bitrot algorithm self-test (rustfs/backlog#1873).
+//!
+//! A drifted hash implementation fails silently in production: every shard
+//! reads back "corrupt", heal rewrites healthy data, and cross-platform
+//! clusters disagree about which copy is good. [`run_startup_bitrot_self_test`]
+//! pins the algorithms once at process start — the check itself runs in well
+//! under a millisecond on 4 KiB, so it executes inline before background
+//! services come up and the result is published before the server accepts
+//! traffic.
+//!
+//! Outcome surface:
+//! - one structured `bitrot_selftest` log event (`passed`/`failed`/`skipped`),
+//! - the `rustfs_bitrot_selftest_status` gauge (1=passed, 0=failed, 2=skipped),
+//! - [`bitrot_selftest_passed`] for admin/health surfaces,
+//! - `RUSTFS_BITROT_SELFTEST_STRICT=on` turns a failure into a startup error
+//!   (MinIO `bitrotSelfTest` Fatal parity); the default only degrades the
+//!   status so a bad build cannot brick an existing fleet on upgrade.
+
+use crate::storage_api::startup::background::{BitrotSelfTestError, bitrot_self_test};
+use metrics::gauge;
+use std::future::Future;
+use std::io;
+use std::sync::atomic::{AtomicU8, Ordering};
+use std::time::Instant;
+use tracing::{debug, error, info};
+
+const LOG_COMPONENT_MAIN: &str = "main";
+const LOG_SUBSYSTEM_STARTUP: &str = "startup";
+const EVENT_BITROT_SELFTEST: &str = "bitrot_selftest";
+const METRIC_BITROT_SELFTEST_STATUS: &str = "rustfs_bitrot_selftest_status";
+
+/// Gauge values for [`METRIC_BITROT_SELFTEST_STATUS`].
+const STATUS_PASSED: f64 = 1.0;
+const STATUS_FAILED: f64 = 0.0;
+const STATUS_SKIPPED: f64 = 2.0;
+
+/// Internal cell values for [`BITROT_SELF_TEST_STATUS`].
+const STATUS_CELL_UNSET: u8 = 0;
+const STATUS_CELL_PASSED: u8 = 1;
+const STATUS_CELL_FAILED: u8 = 2;
+
+static BITROT_SELF_TEST_STATUS: AtomicU8 = AtomicU8::new(STATUS_CELL_UNSET);
+
+/// Last recorded self-test outcome: `None` before the first run, then
+/// `Some(true)` on a passing check and `Some(false)` on a failed one (a
+/// skipped check never publishes, so it cannot read as a pass). The cell is
+/// last-writer-wins rather than set-once: production runs the self-test once,
+/// and last-writer-wins keeps tests that exercise both outcomes
+/// order-independent.
+pub fn bitrot_selftest_passed() -> Option<bool> {
+    match BITROT_SELF_TEST_STATUS.load(Ordering::Acquire) {
+        STATUS_CELL_UNSET => None,
+        STATUS_CELL_PASSED => Some(true),
+        STATUS_CELL_FAILED => Some(false),
+        _ => None,
+    }
+}
+
+/// Run the bitrot self-test and publish the outcome. In strict mode a failure
+/// is returned as an error so the caller aborts startup.
+pub(crate) async fn run_startup_bitrot_self_test(enabled: bool, strict: bool) -> io::Result<()> {
+    run_startup_bitrot_self_test_with(enabled, strict, bitrot_self_test).await
+}
+
+async fn run_startup_bitrot_self_test_with<F, Fut>(enabled: bool, strict: bool, run_check: F) -> io::Result<()>
+where
+    F: FnOnce() -> Fut,
+    Fut: Future<Output = Result<(), BitrotSelfTestError>>,
+{
+    if !enabled {
+        gauge!(METRIC_BITROT_SELFTEST_STATUS).set(STATUS_SKIPPED);
+        debug!(
+            target: "rustfs::main::run",
+            event = EVENT_BITROT_SELFTEST,
+            component = LOG_COMPONENT_MAIN,
+            subsystem = LOG_SUBSYSTEM_STARTUP,
+            state = "skipped",
+            reason = "disabled",
+            "Bitrot self-test skipped"
+        );
+        return Ok(());
+    }
+
+    let started = Instant::now();
+    match run_check().await {
+        Ok(()) => {
+            BITROT_SELF_TEST_STATUS.store(STATUS_CELL_PASSED, Ordering::Release);
+            gauge!(METRIC_BITROT_SELFTEST_STATUS).set(STATUS_PASSED);
+            info!(
+                target: "rustfs::main::run",
+                event = EVENT_BITROT_SELFTEST,
+                component = LOG_COMPONENT_MAIN,
+                subsystem = LOG_SUBSYSTEM_STARTUP,
+                state = "passed",
+                duration_us = started.elapsed().as_micros() as u64,
+                "Bitrot self-test passed"
+            );
+        }
+        Err(err) => {
+            BITROT_SELF_TEST_STATUS.store(STATUS_CELL_FAILED, Ordering::Release);
+            gauge!(METRIC_BITROT_SELFTEST_STATUS).set(STATUS_FAILED);
+            error!(
+                target: "rustfs::main::run",
+                event = EVENT_BITROT_SELFTEST,
+                component = LOG_COMPONENT_MAIN,
+                subsystem = LOG_SUBSYSTEM_STARTUP,
+                state = "failed",
+                duration_us = started.elapsed().as_micros() as u64,
+                error = %err,
+                "Bitrot self-test failed"
+            );
+            if strict {
+                return Err(io::Error::other(format!("bitrot self-test failed: {err}")));
+            }
+        }
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{BITROT_SELF_TEST_STATUS, STATUS_CELL_UNSET, bitrot_selftest_passed, run_startup_bitrot_self_test_with};
+    use crate::storage_api::startup::background::BitrotSelfTestError;
+    use std::future::ready;
+    use std::sync::atomic::Ordering;
+
+    fn failing_check() -> impl Future<Output = Result<(), BitrotSelfTestError>> {
+        ready(Err(BitrotSelfTestError::RoundtripReadback {
+            algorithm: "HighwayHash256S",
+        }))
+    }
+
+    /// All scenarios run sequentially inside one test: the status cell is
+    /// process-global, so parallel per-scenario tests would race the reset and
+    /// read each other's outcomes (the exact order-dependent flake class this
+    /// module exists to avoid).
+    #[tokio::test]
+    async fn startup_self_test_publishes_outcome_and_strict_gates_abort() {
+        BITROT_SELF_TEST_STATUS.store(STATUS_CELL_UNSET, Ordering::Release);
+
+        // Skipped: publishes nothing, never fails, never aborts.
+        run_startup_bitrot_self_test_with(false, true, || async { Ok(()) })
+            .await
+            .expect("a disabled self-test must not fail even in strict mode");
+        assert_eq!(bitrot_selftest_passed(), None, "a skipped run must leave the status unset");
+
+        // Passing: publishes Some(true), never fails.
+        run_startup_bitrot_self_test_with(true, false, || async { Ok(()) })
+            .await
+            .expect("a passing check must never fail startup");
+        assert_eq!(bitrot_selftest_passed(), Some(true), "a passing run must publish Some(true)");
+
+        // Failing, non-strict: publishes Some(false) but startup continues.
+        run_startup_bitrot_self_test_with(true, false, failing_check)
+            .await
+            .expect("a failed check must not abort startup in non-strict mode");
+        assert_eq!(bitrot_selftest_passed(), Some(false), "a failing run must publish Some(false)");
+
+        // Failing, strict: startup error carries the failure and the published
+        // outcome stays a failure.
+        let err = run_startup_bitrot_self_test_with(true, true, failing_check)
+            .await
+            .expect_err("strict mode must turn a failed check into a startup error");
+        assert!(err.to_string().contains("bitrot self-test failed"));
+        assert_eq!(bitrot_selftest_passed(), Some(false));
+    }
+}

rustfs/src/lib.rs

@@ -76,6 +76,7 @@ pub mod allocator_reclaim;
 pub mod app;
 pub mod auth;
 pub mod auth_keystone;
+pub(crate) mod bitrot_selftest;
 pub mod capacity;
 pub mod cluster_snapshot;
 pub mod config;

rustfs/src/module_switches.rs

@@ -33,6 +33,8 @@ pub(crate) const ENV_SCANNER_ENABLED: &str = "RUSTFS_SCANNER_ENABLED";
 pub(crate) const ENV_SCANNER_ENABLED_DEPRECATED: &str = "RUSTFS_ENABLE_SCANNER";
 pub(crate) const ENV_HEAL_ENABLED: &str = "RUSTFS_HEAL_ENABLED";
 pub(crate) const ENV_HEAL_ENABLED_DEPRECATED: &str = "RUSTFS_ENABLE_HEAL";
+pub(crate) const ENV_BITROT_SELFTEST_ENABLE: &str = "RUSTFS_BITROT_SELFTEST_ENABLE";
+pub(crate) const ENV_BITROT_SELFTEST_STRICT: &str = "RUSTFS_BITROT_SELFTEST_STRICT";
 
 static AUDIT_MODULE_ENABLED: AtomicBool = AtomicBool::new(rustfs_config::DEFAULT_AUDIT_ENABLE);
 static NOTIFY_MODULE_ENABLED: AtomicBool = AtomicBool::new(rustfs_config::DEFAULT_NOTIFY_ENABLE);
@@ -47,6 +49,18 @@ pub(crate) fn heal_enabled_from_env() -> bool {
     get_env_bool_with_aliases(ENV_HEAL_ENABLED, &[ENV_HEAL_ENABLED_DEPRECATED], true)
 }
 
+/// Whether the startup bitrot algorithm self-test runs, defaulting to on
+/// (rustfs/backlog#1873).
+pub(crate) fn bitrot_selftest_enabled_from_env() -> bool {
+    rustfs_utils::get_env_bool(ENV_BITROT_SELFTEST_ENABLE, true)
+}
+
+/// Whether a failed bitrot self-test aborts startup instead of only logging
+/// and exposing a failed status, defaulting to off.
+pub(crate) fn bitrot_selftest_strict_from_env() -> bool {
+    rustfs_utils::get_env_bool(ENV_BITROT_SELFTEST_STRICT, false)
+}
+
 /// Last published audit-module state.
 pub fn is_audit_module_enabled() -> bool {
     AUDIT_MODULE_ENABLED.load(Ordering::Relaxed)

rustfs/src/startup_background.rs

@@ -12,7 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::module_switches::{heal_enabled_from_env, scanner_enabled_from_env};
+use crate::bitrot_selftest::run_startup_bitrot_self_test;
+use crate::module_switches::{
+    bitrot_selftest_enabled_from_env, bitrot_selftest_strict_from_env, heal_enabled_from_env, scanner_enabled_from_env,
+};
 use crate::storage_api::startup::background::{ECStore, set_workload_admission_snapshot_provider};
 use crate::workload_admission::RustFsWorkloadAdmissionSnapshotProvider;
 use rustfs_concurrency::WorkloadAdmissionSnapshotProvider;
@@ -27,6 +30,12 @@ const LOG_SUBSYSTEM_STARTUP: &str = "startup";
 const EVENT_BACKGROUND_SERVICES_CONFIGURED: &str = "background_services_configured";
 
 pub(crate) async fn init_background_service_runtime(store: Arc<ECStore>) -> Result<bool> {
+    // Pin the bitrot algorithms before anything can write or verify a shard:
+    // the check costs well under a millisecond, and in strict mode a drifted
+    // build must abort here rather than after it has touched data
+    // (rustfs/backlog#1873).
+    run_startup_bitrot_self_test(bitrot_selftest_enabled_from_env(), bitrot_selftest_strict_from_env()).await?;
+
     let _ = create_ahm_services_cancel_token();
 
     let enable_scanner = scanner_enabled_from_env();

rustfs/src/storage/storage_api.rs

@@ -569,6 +569,10 @@ pub(crate) mod ecstore_erasure {
     pub(crate) use rustfs_ecstore::api::erasure::{BitrotReader, Erasure};
 }
 
+/// Startup bitrot algorithm self-test (rustfs/backlog#1873), re-exported for
+/// the root facade's background-startup section.
+pub(crate) use rustfs_ecstore::api::erasure::{BitrotSelfTestError, bitrot_self_test};
+
 pub(crate) mod ecstore_storage {
     #[cfg(test)]
     pub(crate) use rustfs_ecstore::api::storage::init_local_disks;

rustfs/src/storage_api.rs

@@ -214,7 +214,9 @@ pub(crate) mod startup {
     }
 
     pub(crate) mod background {
-        pub(crate) use crate::storage::storage_api::{ECStore, set_workload_admission_snapshot_provider};
+        pub(crate) use crate::storage::storage_api::{
+            BitrotSelfTestError, ECStore, bitrot_self_test, set_workload_admission_snapshot_provider,
+        };
     }
 
     pub(crate) mod bucket_metadata {