Merge branch 'main' into overtrue/backlog-1823-step10-batch3

crates/checksums/src/http.rs

@@ -38,7 +38,10 @@ pub const XXHASH_3_HEADER_NAME: &str = "x-amz-checksum-xxhash3";
 pub const XXHASH_64_HEADER_NAME: &str = "x-amz-checksum-xxhash64";
 pub const XXHASH_128_HEADER_NAME: &str = "x-amz-checksum-xxhash128";
 
-#[allow(dead_code)]
+#[allow(
+    dead_code,
+    reason = "Content-MD5 wire name, resolved by header_name() below and asserted by this crate's tests (backlog#1823)"
+)]
 pub(crate) static MD5_HEADER_NAME: &str = "content-md5";
 
 pub const CHECKSUM_ALGORITHMS_IN_PRIORITY_ORDER: [&str; 5] =

crates/checksums/src/lib.rs

@@ -476,13 +476,19 @@ impl Checksum for Xxhash64 {
     }
 }
 
-#[allow(dead_code)]
 #[derive(Debug, Default)]
+#[allow(
+    dead_code,
+    reason = "Content-MD5 is not a ChecksumAlgorithm variant and has no arm in into_impl: S3 carries it as its own header, separate from the x-amz-checksum-* family. This impl exists so the two paths share the Checksum trait, and is asserted by this crate's tests (backlog#1823)"
+)]
 struct Md5 {
     hasher: md5::Md5,
 }
 
-#[allow(dead_code)]
+#[allow(
+    dead_code,
+    reason = "Content-MD5 is not a ChecksumAlgorithm variant and has no arm in into_impl: S3 carries it as its own header, separate from the x-amz-checksum-* family. This impl exists so the two paths share the Checksum trait, and is asserted by this crate's tests (backlog#1823)"
+)]
 impl Md5 {
     fn update(&mut self, bytes: &[u8]) {
         use md5::Digest;

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/keystone/src/client.rs

@@ -31,7 +31,10 @@ pub struct KeystoneClient {
     admin_password: Option<String>,
     admin_project: Option<String>,
     admin_domain: String,
-    #[allow(dead_code)]
+    #[allow(
+        dead_code,
+        reason = "TLS verification flag parsed from config; the reqwest client is built before it is consulted, so nothing reads it back (backlog#1823)"
+    )]
     verify_ssl: bool,
     /// Request timeout applied to the underlying HTTP client.
     timeout: std::time::Duration,

crates/keystone/src/identity.rs

@@ -20,7 +20,10 @@ use tracing::{debug, info};
 
 /// Maps Keystone identities to RustFS concepts
 pub struct KeystoneIdentityMapper {
-    #[allow(dead_code)]
+    #[allow(
+        dead_code,
+        reason = "keeps the Keystone client alive for the mapper's lifetime; the mapping paths do not call through it yet (backlog#1823)"
+    )]
     client: Arc<KeystoneClient>,
     role_policy_map: HashMap<String, String>,
     enable_tenant_prefix: bool,

crates/kms/src/api_types.rs

@@ -293,6 +293,15 @@ enum StrictVaultAuthMethod {
         #[serde(default)]
         refresh_safety_window_secs: Option<u64>,
     },
+    Kubernetes {
+        role: String,
+        #[serde(default)]
+        mount: Option<String>,
+        #[serde(default)]
+        jwt_path: Option<std::path::PathBuf>,
+        #[serde(default)]
+        refresh_safety_window_secs: Option<u64>,
+    },
     TokenFile {
         path: std::path::PathBuf,
         #[serde(default)]
@@ -319,6 +328,17 @@ impl From<StrictVaultAuthMethod> for VaultAuthMethod {
                 mount: mount.unwrap_or_else(|| crate::config::DEFAULT_VAULT_APPROLE_MOUNT.to_string()),
                 refresh_safety_window_secs,
             },
+            StrictVaultAuthMethod::Kubernetes {
+                role,
+                mount,
+                jwt_path,
+                refresh_safety_window_secs,
+            } => Self::Kubernetes {
+                role,
+                mount: mount.unwrap_or_else(|| crate::config::DEFAULT_VAULT_KUBERNETES_MOUNT.to_string()),
+                jwt_path: jwt_path.unwrap_or_else(|| std::path::PathBuf::from(crate::config::DEFAULT_VAULT_KUBERNETES_JWT_PATH)),
+                refresh_safety_window_secs,
+            },
             StrictVaultAuthMethod::TokenFile {
                 path,
                 poll_interval_secs,
@@ -499,6 +519,7 @@ impl From<&KmsConfig> for KmsConfigSummary {
                 auth_method_type: match &vault_config.auth_method {
                     VaultAuthMethod::Token { .. } => "token".to_string(),
                     VaultAuthMethod::AppRole { .. } => "approle".to_string(),
+                    VaultAuthMethod::Kubernetes { .. } => "kubernetes".to_string(),
                     VaultAuthMethod::TokenFile { .. } => "token_file".to_string(),
                 },
                 has_stored_credentials: true,
@@ -513,6 +534,7 @@ impl From<&KmsConfig> for KmsConfigSummary {
                 auth_method_type: match &vault_config.auth_method {
                     VaultAuthMethod::Token { .. } => "token".to_string(),
                     VaultAuthMethod::AppRole { .. } => "approle".to_string(),
+                    VaultAuthMethod::Kubernetes { .. } => "kubernetes".to_string(),
                     VaultAuthMethod::TokenFile { .. } => "token_file".to_string(),
                 },
                 has_stored_credentials: true,
@@ -901,6 +923,42 @@ mod tests {
         assert!(request.to_kms_config().validate().is_ok());
     }
 
+    /// The admin API reaches Kubernetes auth with the role alone; the mount and
+    /// the projected token path fall back to the cluster defaults, so a Tenant
+    /// manifest carries no credential and no cluster-specific paths.
+    #[test]
+    fn test_deserialize_vault_configure_request_accepts_kubernetes_auth() {
+        let raw = serde_json::json!({
+            "backend_type": "vault-transit",
+            "address": "https://vault.example.com:8200",
+            "mount_path": "rustfs",
+            "auth_method": { "Kubernetes": { "role": "rustfs" } }
+        });
+
+        let request: ConfigureKmsRequest = serde_json::from_value(raw).expect("kubernetes auth should deserialize");
+        let config = request.to_kms_config();
+        config.validate().expect("kubernetes auth must validate");
+
+        let vault = config.vault_transit_config().expect("vault transit backend config");
+        let VaultAuthMethod::Kubernetes {
+            role, mount, jwt_path, ..
+        } = &vault.auth_method
+        else {
+            panic!("expected Kubernetes auth, got {:?}", vault.auth_method);
+        };
+        assert_eq!(role, "rustfs");
+        assert_eq!(mount, crate::config::DEFAULT_VAULT_KUBERNETES_MOUNT);
+        assert_eq!(jwt_path, std::path::Path::new(crate::config::DEFAULT_VAULT_KUBERNETES_JWT_PATH));
+
+        let unknown_field = serde_json::json!({
+            "backend_type": "vault-transit",
+            "address": "https://vault.example.com:8200",
+            "auth_method": { "Kubernetes": { "role": "rustfs", "service_account": "rustfs" } }
+        });
+        serde_json::from_value::<ConfigureKmsRequest>(unknown_field)
+            .expect_err("an unknown auth field must be rejected rather than silently dropped");
+    }
+
     #[test]
     fn test_deserialize_aws_configure_request_accepts_type_aliases() {
         for backend_type in ["AWS", "AwsKms", "aws", "aws-kms", "aws_kms"] {

crates/kms/src/backends/vault.rs

@@ -550,6 +550,7 @@ impl VaultKmsClient {
             address: config.address.clone(),
             namespace: config.namespace.clone(),
             attempt_timeout: kms_config.effective_timeout(),
+            skip_tls_verify: config.tls.as_ref().is_some_and(|tls| tls.skip_verify),
         };
         let source = token_source_for(&config.auth_method, &settings)?;
         let policy = VaultCredentialPolicy::from_kms_config(

crates/kms/src/backends/vault_credentials.rs

@@ -326,6 +326,97 @@ impl fmt::Debug for AppRoleLogin {
     }
 }
 
+/// Token source for [`VaultAuthMethod::Kubernetes`]: exchanges the pod's
+/// projected ServiceAccount token for a lease-bound Vault token.
+///
+/// The JWT is re-read on every login because the kubelet rotates a projected
+/// token well inside the pod's lifetime; caching it would strand the source on
+/// an expired assertion once the current Vault token can no longer be renewed.
+///
+/// Unlike [`TokenFileSource`], the file mode is not checked: the kubelet owns
+/// the projected token and mounts it world-readable by default, so rejecting
+/// group/other bits would refuse every standard pod rather than catch a
+/// deployment error.
+pub(crate) struct KubernetesLogin {
+    /// Unauthenticated client used only for the login exchange.
+    login_client: VaultClient,
+    mount: String,
+    role: String,
+    jwt_path: PathBuf,
+}
+
+impl KubernetesLogin {
+    pub(crate) fn new(settings: &VaultConnectionSettings, mount: String, role: String, jwt_path: PathBuf) -> Result<Self> {
+        Ok(Self {
+            login_client: settings.build_login_client()?,
+            mount,
+            role,
+            jwt_path,
+        })
+    }
+
+    /// Read the ServiceAccount token for one login attempt.
+    ///
+    /// Mirrors [`AppRoleLogin::resolve_secret_id`]: a read failure is fatal for
+    /// the attempt but the refresh loop keeps retrying, so a token the kubelet
+    /// has not projected yet heals the source without a restart.
+    async fn resolve_jwt(&self) -> AttemptResult<SecretString> {
+        let mut raw = tokio::fs::read_to_string(&self.jwt_path)
+            .await
+            .map_err(|error| AttemptError {
+                class: ErrorClass::Fatal,
+                error: KmsError::configuration_error(format!(
+                    "Failed to read Kubernetes ServiceAccount token {}: {error}",
+                    self.jwt_path.display()
+                )),
+            })?;
+        let trimmed = raw.trim();
+        if trimmed.is_empty() {
+            raw.zeroize();
+            return Err(AttemptError {
+                class: ErrorClass::Fatal,
+                error: KmsError::configuration_error(format!(
+                    "Kubernetes ServiceAccount token {} is empty",
+                    self.jwt_path.display()
+                )),
+            });
+        }
+        let jwt = SecretString::new(trimmed.to_string());
+        raw.zeroize();
+        Ok(jwt)
+    }
+}
+
+#[async_trait]
+impl TokenSource for KubernetesLogin {
+    async fn acquire(&self) -> AttemptResult<TokenLease> {
+        let jwt = self.resolve_jwt().await?;
+        let auth = vaultrs::auth::kubernetes::login(&self.login_client, &self.mount, &self.role, jwt.expose())
+            .await
+            .map_err(|error| attempt_error("Kubernetes login", error))?;
+        Ok(TokenLease::from_auth(auth))
+    }
+
+    async fn renew(&self, client: &VaultClient) -> AttemptResult<TokenLease> {
+        let auth = vaultrs::token::renew_self(client, None)
+            .await
+            .map_err(|error| attempt_error("token renewal", error))?;
+        Ok(TokenLease::from_auth(auth))
+    }
+}
+
+impl fmt::Debug for KubernetesLogin {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        // The login client embeds Vault client settings and must stay out of
+        // Debug output; the role name is not a secret, and the JWT is never held.
+        f.debug_struct("KubernetesLogin")
+            .field("mount", &self.mount)
+            .field("role", &self.role)
+            .field("jwt_path", &self.jwt_path)
+            .finish_non_exhaustive()
+    }
+}
+
 /// Token source for [`VaultAuthMethod::TokenFile`]: reads an agent-managed
 /// token file (for example a Vault Agent auto-auth sink).
 ///
@@ -464,6 +555,9 @@ pub(crate) fn token_source_for(
             secret_id.clone(),
             secret_id_file.clone(),
         )?)),
+        VaultAuthMethod::Kubernetes {
+            role, mount, jwt_path, ..
+        } => Ok(Box::new(KubernetesLogin::new(settings, mount.clone(), role.clone(), jwt_path.clone())?)),
         VaultAuthMethod::TokenFile {
             path,
             poll_interval_secs,
@@ -486,6 +580,9 @@ pub(crate) struct VaultConnectionSettings {
     pub(crate) namespace: Option<String>,
     /// Per-attempt HTTP timeout applied to the underlying reqwest client.
     pub(crate) attempt_timeout: Duration,
+    /// Whether to accept an unverified Vault server certificate. Gated on
+    /// `allow_insecure_dev_defaults` by `KmsConfig::validate`.
+    pub(crate) skip_tls_verify: bool,
 }
 
 impl VaultConnectionSettings {
@@ -499,6 +596,11 @@ impl VaultConnectionSettings {
         // operation-level retry policy.
         settings_builder.timeout(Some(self.attempt_timeout));
         settings_builder.token(token);
+        // Always set explicitly: left unset, vaultrs derives this from its own
+        // VAULT_SKIP_VERIFY variable, so a stray value in the environment would
+        // disable certificate verification behind the KMS configuration and its
+        // insecure-defaults gate.
+        settings_builder.verify(!self.skip_tls_verify);
 
         if let Some(namespace) = &self.namespace {
             settings_builder.namespace(Some(namespace.clone()));
@@ -551,6 +653,10 @@ impl VaultCredentialPolicy {
                 refresh_safety_window_secs: Some(secs),
                 ..
             }
+            | VaultAuthMethod::Kubernetes {
+                refresh_safety_window_secs: Some(secs),
+                ..
+            }
             | VaultAuthMethod::TokenFile {
                 refresh_safety_window_secs: Some(secs),
                 ..
@@ -584,15 +690,25 @@ pub(crate) struct VaultClientHandle {
 
 impl VaultClientHandle {
     /// Absolute expiry of this generation's token.
+    ///
+    /// `lease.ttl` is built from the `lease_duration` the Vault server sent, so
+    /// a value too large to add to `issued_at` would panic on the bare `+`. A
+    /// TTL that cannot be represented is indistinguishable from no expiry, so it
+    /// collapses to `None` — the same answer already given for the zero-lease
+    /// tokens Vault issues, which keeps the token in use and still fully
+    /// validated by Vault on every call.
     fn expires_at(&self) -> Option<Instant> {
-        self.lease.map(|lease| self.issued_at + lease.ttl)
+        self.lease.and_then(|lease| self.issued_at.checked_add(lease.ttl))
     }
 
     /// When the renewal task should refresh this generation: half the TTL,
     /// leaving the second half as budget for retries before the fail-closed
     /// window is reached.
+    ///
+    /// Unrepresentable TTLs collapse to `None` as in [`Self::expires_at`],
+    /// leaving a token that never expires with nothing to renew.
     fn renew_at(&self) -> Option<Instant> {
-        self.lease.map(|lease| self.issued_at + lease.ttl / 2)
+        self.lease.and_then(|lease| self.issued_at.checked_add(lease.ttl / 2))
     }
 }
 
@@ -662,7 +778,7 @@ impl VaultCredentialProvider {
         let handle = self.current.load_full();
         if let Some(expires_at) = handle.expires_at() {
             let now = Instant::now();
-            if now + self.policy.safety_window >= expires_at {
+            if self.inside_safety_window(now, expires_at) {
                 return Err(KmsError::credentials_unavailable(format!(
                     "Vault token (generation {}) is within {:?} of expiry and has not been refreshed; refusing to use it",
                     handle.generation, self.policy.safety_window
@@ -672,6 +788,18 @@ impl VaultCredentialProvider {
         Ok(handle)
     }
 
+    /// Whether the token expiring at `expires_at` is close enough to refuse.
+    ///
+    /// `safety_window` reaches here from persisted configuration, so it is not
+    /// guaranteed to have passed this version's validation: a window too large
+    /// to add to the current instant would panic on the bare `+`. Such a window
+    /// means every token is always inside it, so saturating to "refuse" is both
+    /// the fail-closed answer and the one the arithmetic was reaching for.
+    fn inside_safety_window(&self, now: Instant, expires_at: Instant) -> bool {
+        now.checked_add(self.policy.safety_window)
+            .is_none_or(|deadline| deadline >= expires_at)
+    }
+
     /// Publish the credential gauges for the generation currently installed.
     ///
     /// The fail-closed gauge re-evaluates the very gate
@@ -683,7 +811,7 @@ impl VaultCredentialProvider {
         let fail_closed = match handle.expires_at() {
             Some(expires_at) => {
                 metrics::gauge!(METRIC_TOKEN_TTL_SECONDS).set(expires_at.saturating_duration_since(now).as_secs_f64());
-                now + self.policy.safety_window >= expires_at
+                self.inside_safety_window(now, expires_at)
             }
             // A generation without an expiry has no remaining TTL to report
             // and can never lapse, so it can never fail closed either.
@@ -860,7 +988,7 @@ impl Drop for CredentialTaskHandle {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::config::REDACTED_SECRET;
+    use crate::config::{DEFAULT_VAULT_KUBERNETES_MOUNT, REDACTED_SECRET};
     use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
 
     const TEST_TOKEN: &str = "vault-token-debug-leak-canary";
@@ -871,6 +999,7 @@ mod tests {
             address: "http://127.0.0.1:8200".to_string(),
             namespace: Some("team-namespace".to_string()),
             attempt_timeout: Duration::from_secs(30),
+            skip_tls_verify: false,
         }
     }
 
@@ -1057,6 +1186,143 @@ mod tests {
         assert!(format!("{source:?}").contains("AppRoleLogin"));
     }
 
+    #[tokio::test]
+    async fn test_kubernetes_auth_method_maps_to_login_source() {
+        let settings = test_settings();
+        let source = token_source_for(&VaultAuthMethod::kubernetes("rustfs".to_string()), &settings)
+            .expect("kubernetes auth must map to a login source");
+
+        assert!(format!("{source:?}").contains("KubernetesLogin"));
+    }
+
+    /// `refresh_safety_window_secs` is operator-supplied and reaches the request
+    /// path from persisted configuration, so the fail-closed comparison must
+    /// survive a window too large to add to the current instant. Before the
+    /// checked arithmetic this panicked with "overflow when adding duration to
+    /// instant" on the first request after a lease-bearing login.
+    #[tokio::test]
+    async fn test_current_refuses_rather_than_panics_on_an_unrepresentable_safety_window() {
+        let (provider, _state) = scripted_provider(
+            Duration::from_secs(60),
+            true,
+            test_policy(Duration::from_secs(u64::MAX), Duration::from_secs(5)),
+        )
+        .await;
+
+        let error = provider
+            .current()
+            .expect_err("a window wider than any lease must refuse the token");
+        assert!(
+            matches!(error, KmsError::CredentialsUnavailable { .. }),
+            "expected CredentialsUnavailable, got {error:?}"
+        );
+    }
+
+    /// `lease_duration` is a bare u64 straight off the Vault response and forms
+    /// the other side of the same comparison, so an absurd one must not panic
+    /// either. It is indistinguishable from a non-expiring token, which is how
+    /// the zero-lease case already behaves.
+    #[tokio::test]
+    async fn test_an_unrepresentable_lease_is_treated_as_non_expiring() {
+        let (provider, _state) = scripted_provider(
+            Duration::from_secs(u64::MAX),
+            true,
+            test_policy(Duration::from_secs(30), Duration::from_secs(5)),
+        )
+        .await;
+
+        provider
+            .current()
+            .expect("a token whose expiry cannot be represented must stay usable");
+    }
+
+    /// The configured flag has to reach the HTTP client, not just the config
+    /// struct: every generation (authenticated and login) builds its own client,
+    /// and a Vault with a self-signed certificate fails the handshake unless
+    /// each one carries the setting.
+    #[test]
+    fn test_skip_tls_verify_reaches_every_vault_client_generation() {
+        for skip_tls_verify in [false, true] {
+            let settings = VaultConnectionSettings {
+                address: "https://vault.example.com:8200".to_string(),
+                namespace: None,
+                attempt_timeout: Duration::from_secs(30),
+                skip_tls_verify,
+            };
+
+            let authenticated = settings.build_client(TEST_TOKEN).expect("authenticated client must build");
+            assert_eq!(authenticated.settings.verify, !skip_tls_verify);
+
+            let login = settings.build_login_client().expect("login client must build");
+            assert_eq!(login.settings.verify, !skip_tls_verify);
+        }
+    }
+
+    /// vaultrs derives `verify` from its own VAULT_SKIP_VERIFY variable when the
+    /// builder leaves it unset, which would disable certificate verification
+    /// without passing the KMS insecure-defaults gate.
+    #[test]
+    fn test_vaultrs_skip_verify_env_cannot_override_the_configured_setting() {
+        temp_env::with_var("VAULT_SKIP_VERIFY", Some("true"), || {
+            let client = test_settings().build_client(TEST_TOKEN).expect("client must build");
+            assert!(
+                client.settings.verify,
+                "a stray VAULT_SKIP_VERIFY must not disable verification behind the KMS configuration"
+            );
+        });
+    }
+
+    /// The projected token is read fresh per login attempt and trimmed, so a
+    /// kubelet rotation is picked up without a restart and a trailing newline
+    /// does not corrupt the assertion sent to Vault.
+    #[tokio::test]
+    async fn test_kubernetes_login_rereads_and_trims_the_service_account_token() {
+        let dir = tempfile::tempdir().expect("temp dir");
+        let path = dir.path().join("token");
+        tokio::fs::write(&path, "  first-jwt\n").await.expect("write token");
+
+        let login = KubernetesLogin::new(
+            &test_settings(),
+            DEFAULT_VAULT_KUBERNETES_MOUNT.to_string(),
+            "rustfs".to_string(),
+            path.clone(),
+        )
+        .expect("login source must build");
+
+        assert_eq!(login.resolve_jwt().await.expect("first read").expose(), "first-jwt");
+
+        tokio::fs::write(&path, "rotated-jwt").await.expect("rotate token");
+        assert_eq!(
+            login.resolve_jwt().await.expect("second read").expose(),
+            "rotated-jwt",
+            "a rotated projected token must be picked up without a restart"
+        );
+    }
+
+    /// The ServiceAccount token is re-read per attempt, so an unreadable or
+    /// empty one fails that attempt without reaching Vault; the refresh loop
+    /// keeps retrying, which is what lets a late projection heal the source.
+    #[tokio::test]
+    async fn test_kubernetes_login_rejects_an_unusable_service_account_token() {
+        let dir = tempfile::tempdir().expect("temp dir");
+        let missing = dir.path().join("absent-token");
+        let empty = dir.path().join("empty-token");
+        tokio::fs::write(&empty, "  \n").await.expect("write empty token");
+
+        for (path, expected) in [(missing, "Failed to read"), (empty, "is empty")] {
+            let login =
+                KubernetesLogin::new(&test_settings(), DEFAULT_VAULT_KUBERNETES_MOUNT.to_string(), "rustfs".to_string(), path)
+                    .expect("login source must build");
+
+            let error = login
+                .acquire()
+                .await
+                .expect_err("an unusable ServiceAccount token must fail the attempt");
+            assert!(matches!(error.class, ErrorClass::Fatal));
+            assert!(error.error.to_string().contains(expected), "got {}", error.error);
+        }
+    }
+
     #[tokio::test(start_paused = true)]
     async fn test_renewal_task_renews_at_half_ttl() {
         let (provider, state) = scripted_provider(

crates/kms/src/backends/vault_transit.rs

@@ -415,6 +415,7 @@ impl VaultTransitKmsClient {
             address: config.address.clone(),
             namespace: config.namespace.clone(),
             attempt_timeout: kms_config.effective_timeout(),
+            skip_tls_verify: config.tls.as_ref().is_some_and(|tls| tls.skip_verify),
         };
         let source = token_source_for(&config.auth_method, &settings)?;
         let policy = VaultCredentialPolicy::from_kms_config(

crates/kms/src/backup/vault_restore.rs

@@ -450,6 +450,10 @@ impl VaultRestoreClient {
             address: target.address.clone(),
             namespace: target.namespace.clone(),
             attempt_timeout: kms_config.effective_timeout(),
+            // A restore target carries no TLS settings, so certificates are
+            // always verified: recovery is the last path that should accept an
+            // unauthenticated Vault.
+            skip_tls_verify: false,
         };
         let source = token_source_for(&target.auth_method, &settings)?;
         let policy = VaultCredentialPolicy::from_kms_config(

crates/kms/src/config.rs

@@ -25,6 +25,10 @@ use url::Url;
 
 pub const ENV_KMS_ALLOW_INSECURE_DEV_DEFAULTS: &str = "RUSTFS_KMS_ALLOW_INSECURE_DEV_DEFAULTS";
 pub const ENV_KMS_ALLOW_IMMEDIATE_DELETION: &str = "RUSTFS_KMS_ALLOW_IMMEDIATE_DELETION";
+pub const ENV_KMS_VAULT_ADDRESS: &str = "RUSTFS_KMS_VAULT_ADDRESS";
+pub const ENV_KMS_VAULT_TOKEN: &str = "RUSTFS_KMS_VAULT_TOKEN";
+pub const ENV_KMS_VAULT_NAMESPACE: &str = "RUSTFS_KMS_VAULT_NAMESPACE";
+pub const ENV_KMS_VAULT_MOUNT_PATH: &str = "RUSTFS_KMS_VAULT_MOUNT_PATH";
 pub const ENV_KMS_VAULT_SKIP_TLS_VERIFY: &str = "RUSTFS_KMS_VAULT_SKIP_TLS_VERIFY";
 pub const ENV_KMS_VAULT_TRANSIT_METADATA_KV_MOUNT: &str = "RUSTFS_KMS_VAULT_TRANSIT_METADATA_KV_MOUNT";
 pub const ENV_KMS_VAULT_TRANSIT_METADATA_PREFIX: &str = "RUSTFS_KMS_VAULT_TRANSIT_METADATA_PREFIX";
@@ -35,6 +39,9 @@ pub const ENV_KMS_VAULT_APPROLE_SECRET_ID: &str = "RUSTFS_KMS_VAULT_APPROLE_SECR
 pub const ENV_KMS_VAULT_APPROLE_SECRET_ID_FILE: &str = "RUSTFS_KMS_VAULT_APPROLE_SECRET_ID_FILE";
 pub const ENV_KMS_VAULT_APPROLE_MOUNT: &str = "RUSTFS_KMS_VAULT_APPROLE_MOUNT";
 pub const ENV_KMS_VAULT_TOKEN_FILE: &str = "RUSTFS_KMS_VAULT_TOKEN_FILE";
+pub const ENV_KMS_VAULT_KUBERNETES_ROLE: &str = "RUSTFS_KMS_VAULT_KUBERNETES_ROLE";
+pub const ENV_KMS_VAULT_KUBERNETES_MOUNT: &str = "RUSTFS_KMS_VAULT_KUBERNETES_MOUNT";
+pub const ENV_KMS_VAULT_KUBERNETES_JWT_PATH: &str = "RUSTFS_KMS_VAULT_KUBERNETES_JWT_PATH";
 pub const ENV_KMS_AWS_REGION: &str = "RUSTFS_KMS_AWS_REGION";
 pub const ENV_KMS_AWS_ENDPOINT_URL: &str = "RUSTFS_KMS_AWS_ENDPOINT_URL";
 /// Age in whole seconds beyond which a key is reported as due for rotation;
@@ -45,6 +52,9 @@ pub const ENV_KMS_ROTATION_MAX_WRAPS: &str = "RUSTFS_KMS_ROTATION_MAX_WRAPS";
 pub const DEFAULT_VAULT_TRANSIT_METADATA_KV_MOUNT: &str = "secret";
 pub const DEFAULT_VAULT_TRANSIT_METADATA_KEY_PREFIX: &str = "rustfs/kms/transit-metadata";
 pub const DEFAULT_VAULT_APPROLE_MOUNT: &str = "approle";
+pub const DEFAULT_VAULT_KUBERNETES_MOUNT: &str = "kubernetes";
+/// Where the kubelet projects a pod's ServiceAccount token by default.
+pub const DEFAULT_VAULT_KUBERNETES_JWT_PATH: &str = "/var/run/secrets/kubernetes.io/serviceaccount/token";
 
 /// Upper bound applied to `KmsConfig::timeout` when deriving backend behavior.
 ///
@@ -84,6 +94,14 @@ fn default_vault_approle_mount() -> String {
     DEFAULT_VAULT_APPROLE_MOUNT.to_string()
 }
 
+fn default_vault_kubernetes_mount() -> String {
+    DEFAULT_VAULT_KUBERNETES_MOUNT.to_string()
+}
+
+fn default_vault_kubernetes_jwt_path() -> PathBuf {
+    PathBuf::from(DEFAULT_VAULT_KUBERNETES_JWT_PATH)
+}
+
 pub const KMS_CONFIG_REDACTION_RULES: &[RedactionRule] = &[
     RedactionRule::new("kms.local.master_key", RedactionLevel::Secret, "local backend key encryption material"),
     RedactionRule::new("kms.vault.token", RedactionLevel::Secret, "vault authentication token"),
@@ -490,6 +508,23 @@ pub enum VaultAuthMethod {
         #[serde(default)]
         refresh_safety_window_secs: Option<u64>,
     },
+    /// Kubernetes authentication: the pod's ServiceAccount token is exchanged
+    /// for a lease-bound Vault token that is renewed in the background.
+    Kubernetes {
+        /// Vault role bound to this ServiceAccount.
+        role: String,
+        /// Kubernetes auth engine mount path.
+        #[serde(default = "default_vault_kubernetes_mount")]
+        mount: String,
+        /// Projected ServiceAccount token to present. Re-read on every login so
+        /// a token the kubelet rotates is picked up without a restart.
+        #[serde(default = "default_vault_kubernetes_jwt_path")]
+        jwt_path: PathBuf,
+        /// Fail-closed margin in seconds, as on `AppRole`. Defaults to the
+        /// per-attempt timeout.
+        #[serde(default)]
+        refresh_safety_window_secs: Option<u64>,
+    },
     /// Agent-managed token file (for example a Vault Agent auto-auth sink):
     /// the token is read from `path` and re-read periodically so a token
     /// rotated by the agent is picked up without a restart.
@@ -520,6 +555,16 @@ impl VaultAuthMethod {
         }
     }
 
+    /// Kubernetes authentication with the default mount and projected token path.
+    pub fn kubernetes(role: String) -> Self {
+        Self::Kubernetes {
+            role,
+            mount: default_vault_kubernetes_mount(),
+            jwt_path: default_vault_kubernetes_jwt_path(),
+            refresh_safety_window_secs: None,
+        }
+    }
+
     /// Agent-managed token file with the default poll interval.
     pub fn token_file(path: PathBuf) -> Self {
         Self::TokenFile {
@@ -548,6 +593,20 @@ impl fmt::Debug for VaultAuthMethod {
                 .field("mount", mount)
                 .field("refresh_safety_window_secs", refresh_safety_window_secs)
                 .finish(),
+            // No redaction: the role and mount name a Vault binding, and the
+            // ServiceAccount token itself is never held on this type.
+            Self::Kubernetes {
+                role,
+                mount,
+                jwt_path,
+                refresh_safety_window_secs,
+            } => f
+                .debug_struct("Kubernetes")
+                .field("role", role)
+                .field("mount", mount)
+                .field("jwt_path", jwt_path)
+                .field("refresh_safety_window_secs", refresh_safety_window_secs)
+                .finish(),
             Self::TokenFile {
                 path,
                 poll_interval_secs,
@@ -1028,50 +1087,12 @@ impl KmsConfig {
                 });
             }
             KmsBackend::VaultKv2 => {
-                let address = get_env_str("RUSTFS_KMS_VAULT_ADDRESS", "http://localhost:8200");
-                let auth_method = vault_auth_method_from_env()?;
-                let skip_tls_verify = get_env_bool(ENV_KMS_VAULT_SKIP_TLS_VERIFY, false);
-
-                let mount_path = match get_env_opt_str("RUSTFS_KMS_VAULT_MOUNT_PATH") {
-                    Some(path) => {
-                        tracing::warn!(
-                            "RUSTFS_KMS_VAULT_MOUNT_PATH is deprecated for the Vault KV2 backend: it never calls the Transit engine and the value is stored but unused"
-                        );
-                        path
-                    }
-                    None => default_vault_kv2_mount_path(),
-                };
-
-                config.backend_config = BackendConfig::VaultKv2(Box::new(VaultConfig {
-                    address,
-                    auth_method,
-                    namespace: get_env_opt_str("RUSTFS_KMS_VAULT_NAMESPACE"),
-                    mount_path,
-                    kv_mount: get_env_str("RUSTFS_KMS_VAULT_KV_MOUNT", "secret"),
-                    key_path_prefix: get_env_str("RUSTFS_KMS_VAULT_KEY_PREFIX", "rustfs/kms/keys"),
-                    tls: vault_tls_config(skip_tls_verify),
-                }));
+                config.backend_config =
+                    BackendConfig::VaultKv2(Box::new(vault_kv2_config_from_env(VaultCliOverrides::default())?));
             }
             KmsBackend::VaultTransit => {
-                let address = get_env_str("RUSTFS_KMS_VAULT_ADDRESS", "http://localhost:8200");
-                let auth_method = vault_auth_method_from_env()?;
-                let skip_tls_verify = get_env_bool(ENV_KMS_VAULT_SKIP_TLS_VERIFY, false);
-
-                config.backend_config = BackendConfig::VaultTransit(Box::new(VaultTransitConfig {
-                    address,
-                    auth_method,
-                    namespace: get_env_opt_str("RUSTFS_KMS_VAULT_NAMESPACE"),
-                    mount_path: get_env_str("RUSTFS_KMS_VAULT_MOUNT_PATH", "transit"),
-                    metadata_kv_mount: get_env_str(
-                        ENV_KMS_VAULT_TRANSIT_METADATA_KV_MOUNT,
-                        DEFAULT_VAULT_TRANSIT_METADATA_KV_MOUNT,
-                    ),
-                    metadata_key_prefix: get_env_str(
-                        ENV_KMS_VAULT_TRANSIT_METADATA_PREFIX,
-                        DEFAULT_VAULT_TRANSIT_METADATA_KEY_PREFIX,
-                    ),
-                    tls: vault_tls_config(skip_tls_verify),
-                }));
+                config.backend_config =
+                    BackendConfig::VaultTransit(Box::new(vault_transit_config_from_env(VaultCliOverrides::default())?));
             }
             KmsBackend::Static => {
                 // Read from file first, then fall back to direct env var
@@ -1202,6 +1223,78 @@ fn is_under_temp_dir(path: &Path) -> bool {
     path.starts_with(std::env::temp_dir())
 }
 
+/// Command-line values that take precedence over the matching environment
+/// variables when assembling a Vault backend configuration.
+///
+/// Every field has a `RUSTFS_KMS_VAULT_*` equivalent that the CLI layer already
+/// reads, so these are only set when the operator passed an explicit flag.
+///
+/// Deliberately not `Debug`: `token` holds the raw Vault token, and the
+/// redacting `Debug` impls elsewhere in this module exist because a derived one
+/// would print it. Denying the derive makes a future `{overrides:?}` a compile
+/// error instead of a leak.
+#[derive(Default, Clone, Copy)]
+pub struct VaultCliOverrides<'a> {
+    pub address: Option<&'a str>,
+    pub token: Option<&'a str>,
+    pub mount_path: Option<&'a str>,
+}
+
+/// Assemble the Vault KV2 backend configuration from the environment.
+///
+/// Shared by [`KmsConfig::from_env`] and the server's command-line startup path
+/// so both resolve the same auth method, namespace, TLS and mount settings.
+pub fn vault_kv2_config_from_env(overrides: VaultCliOverrides<'_>) -> Result<VaultConfig> {
+    let mount_path = match overrides
+        .mount_path
+        .map(str::to_string)
+        .or_else(|| get_env_opt_str(ENV_KMS_VAULT_MOUNT_PATH))
+    {
+        Some(path) => {
+            tracing::warn!(
+                "RUSTFS_KMS_VAULT_MOUNT_PATH is deprecated for the Vault KV2 backend: it never calls the Transit engine and the value is stored but unused"
+            );
+            path
+        }
+        None => default_vault_kv2_mount_path(),
+    };
+
+    Ok(VaultConfig {
+        address: vault_address_from_env(overrides.address),
+        auth_method: vault_auth_method_from_env(overrides.token)?,
+        namespace: get_env_opt_str(ENV_KMS_VAULT_NAMESPACE),
+        mount_path,
+        kv_mount: get_env_str("RUSTFS_KMS_VAULT_KV_MOUNT", "secret"),
+        key_path_prefix: get_env_str("RUSTFS_KMS_VAULT_KEY_PREFIX", "rustfs/kms/keys"),
+        tls: vault_tls_config(get_env_bool(ENV_KMS_VAULT_SKIP_TLS_VERIFY, false)),
+    })
+}
+
+/// Assemble the Vault Transit backend configuration from the environment.
+///
+/// Companion to [`vault_kv2_config_from_env`]; see there for why both entry
+/// points share it.
+pub fn vault_transit_config_from_env(overrides: VaultCliOverrides<'_>) -> Result<VaultTransitConfig> {
+    Ok(VaultTransitConfig {
+        address: vault_address_from_env(overrides.address),
+        auth_method: vault_auth_method_from_env(overrides.token)?,
+        namespace: get_env_opt_str(ENV_KMS_VAULT_NAMESPACE),
+        mount_path: overrides
+            .mount_path
+            .map(str::to_string)
+            .unwrap_or_else(|| get_env_str(ENV_KMS_VAULT_MOUNT_PATH, "transit")),
+        metadata_kv_mount: get_env_str(ENV_KMS_VAULT_TRANSIT_METADATA_KV_MOUNT, DEFAULT_VAULT_TRANSIT_METADATA_KV_MOUNT),
+        metadata_key_prefix: get_env_str(ENV_KMS_VAULT_TRANSIT_METADATA_PREFIX, DEFAULT_VAULT_TRANSIT_METADATA_KEY_PREFIX),
+        tls: vault_tls_config(get_env_bool(ENV_KMS_VAULT_SKIP_TLS_VERIFY, false)),
+    })
+}
+
+fn vault_address_from_env(override_value: Option<&str>) -> String {
+    override_value
+        .map(str::to_string)
+        .unwrap_or_else(|| get_env_str(ENV_KMS_VAULT_ADDRESS, "http://localhost:8200"))
+}
+
 /// Resolve the Vault auth method from environment variables.
 ///
 /// Setting `RUSTFS_KMS_VAULT_APPROLE_ROLE_ID` selects AppRole authentication;
@@ -1209,27 +1302,59 @@ fn is_under_temp_dir(path: &Path) -> bool {
 /// (re-read on every login, mirroring the `RUSTFS_KMS_STATIC_SECRET_KEY_FILE`
 /// precedent) or inline from `RUSTFS_KMS_VAULT_APPROLE_SECRET_ID`, with the
 /// file taking precedence. Without a role id the legacy token flow applies.
-fn vault_auth_method_from_env() -> Result<VaultAuthMethod> {
+///
+/// `RUSTFS_KMS_VAULT_KUBERNETES_ROLE` selects Kubernetes authentication, which
+/// presents the pod's projected ServiceAccount token.
+///
+/// `token_override` carries a token supplied on the command line; it stands in
+/// for `RUSTFS_KMS_VAULT_TOKEN` everywhere below, including the conflict checks,
+/// so a flag and the variable it mirrors select the same method.
+fn vault_auth_method_from_env(token_override: Option<&str>) -> Result<VaultAuthMethod> {
+    let token = token_override
+        .map(str::to_string)
+        .or_else(|| get_env_opt_str(ENV_KMS_VAULT_TOKEN));
+    let role_id = get_env_opt_str(ENV_KMS_VAULT_APPROLE_ROLE_ID);
+    let kubernetes_role = get_env_opt_str(ENV_KMS_VAULT_KUBERNETES_ROLE);
+
     if let Some(token_file) = get_env_opt_str(ENV_KMS_VAULT_TOKEN_FILE) {
         // A token file names one authoritative credential source; combining it
         // with another one would leave the effective identity ambiguous, so
         // that is a configuration error rather than a precedence rule.
-        if get_env_opt_str(ENV_KMS_VAULT_APPROLE_ROLE_ID).is_some() {
-            return Err(KmsError::configuration_error(format!(
-                "{ENV_KMS_VAULT_TOKEN_FILE} cannot be combined with {ENV_KMS_VAULT_APPROLE_ROLE_ID}; configure exactly one Vault auth method"
-            )));
-        }
-        if get_env_opt_str("RUSTFS_KMS_VAULT_TOKEN").is_some() {
-            return Err(KmsError::configuration_error(format!(
-                "{ENV_KMS_VAULT_TOKEN_FILE} cannot be combined with RUSTFS_KMS_VAULT_TOKEN; configure exactly one Vault auth method"
-            )));
+        for (name, configured) in [
+            (ENV_KMS_VAULT_APPROLE_ROLE_ID, role_id.is_some()),
+            (ENV_KMS_VAULT_KUBERNETES_ROLE, kubernetes_role.is_some()),
+            (ENV_KMS_VAULT_TOKEN, token.is_some()),
+        ] {
+            if configured {
+                return Err(KmsError::configuration_error(format!(
+                    "{ENV_KMS_VAULT_TOKEN_FILE} cannot be combined with {name}; configure exactly one Vault auth method"
+                )));
+            }
         }
         return Ok(VaultAuthMethod::token_file(PathBuf::from(token_file)));
     }
 
-    let Some(role_id) = get_env_opt_str(ENV_KMS_VAULT_APPROLE_ROLE_ID) else {
+    if let Some(role) = kubernetes_role {
+        // Unlike a leftover static token, a second login method is never a
+        // stale remnant: both were configured deliberately and neither can be
+        // ranked over the other.
+        if role_id.is_some() {
+            return Err(KmsError::configuration_error(format!(
+                "{ENV_KMS_VAULT_KUBERNETES_ROLE} cannot be combined with {ENV_KMS_VAULT_APPROLE_ROLE_ID}; configure exactly one Vault auth method"
+            )));
+        }
+        return Ok(VaultAuthMethod::Kubernetes {
+            role,
+            mount: get_env_str(ENV_KMS_VAULT_KUBERNETES_MOUNT, DEFAULT_VAULT_KUBERNETES_MOUNT),
+            jwt_path: get_env_opt_str(ENV_KMS_VAULT_KUBERNETES_JWT_PATH)
+                .map_or_else(default_vault_kubernetes_jwt_path, PathBuf::from),
+            refresh_safety_window_secs: None,
+        });
+    }
+
+    let Some(role_id) = role_id else {
         return Ok(VaultAuthMethod::Token {
-            token: get_env_str("RUSTFS_KMS_VAULT_TOKEN", "dev-token"),
+            token: token.unwrap_or_else(|| "dev-token".to_string()),
         });
     };
 
@@ -1273,6 +1398,22 @@ fn validate_vault_auth_method(backend_name: &str, auth_method: &VaultAuthMethod)
             }
             Ok(())
         }
+        VaultAuthMethod::Kubernetes {
+            role, mount, jwt_path, ..
+        } => {
+            if role.is_empty() {
+                return Err(KmsError::configuration_error(format!("{backend_name} Kubernetes role cannot be empty")));
+            }
+            if mount.is_empty() {
+                return Err(KmsError::configuration_error(format!("{backend_name} Kubernetes mount cannot be empty")));
+            }
+            if jwt_path.as_os_str().is_empty() {
+                return Err(KmsError::configuration_error(format!(
+                    "{backend_name} Kubernetes ServiceAccount token path cannot be empty"
+                )));
+            }
+            Ok(())
+        }
         VaultAuthMethod::TokenFile {
             path,
             poll_interval_secs,
@@ -1976,6 +2117,106 @@ mod tests {
             .expect("well-formed token file auth must validate");
     }
 
+    /// A Kubernetes role alone configures the method: the credential is the
+    /// pod's projected ServiceAccount token, so nothing secret is in the
+    /// environment and the mount and token path fall back to the cluster
+    /// defaults.
+    #[test]
+    fn test_from_env_selects_kubernetes() {
+        with_vars(
+            vec![
+                ("RUSTFS_KMS_BACKEND", Some("vault-transit")),
+                (ENV_KMS_VAULT_ADDRESS, Some("https://vault.example.com")),
+                (ENV_KMS_VAULT_KUBERNETES_ROLE, Some("rustfs")),
+                (ENV_KMS_VAULT_KUBERNETES_MOUNT, None),
+                (ENV_KMS_VAULT_KUBERNETES_JWT_PATH, None),
+                (ENV_KMS_VAULT_TOKEN, None),
+                (ENV_KMS_VAULT_TOKEN_FILE, None),
+                (ENV_KMS_VAULT_APPROLE_ROLE_ID, None),
+            ],
+            || {
+                let config = KmsConfig::from_env().expect("kms config should load from env");
+                let vault = config.vault_transit_config().expect("vault transit backend config");
+                let VaultAuthMethod::Kubernetes {
+                    role,
+                    mount,
+                    jwt_path,
+                    refresh_safety_window_secs,
+                } = &vault.auth_method
+                else {
+                    panic!(
+                        "a kubernetes role in the environment must select Kubernetes auth, got {:?}",
+                        vault.auth_method
+                    );
+                };
+                assert_eq!(role, "rustfs");
+                assert_eq!(mount, DEFAULT_VAULT_KUBERNETES_MOUNT);
+                assert_eq!(jwt_path, Path::new(DEFAULT_VAULT_KUBERNETES_JWT_PATH));
+                assert_eq!(refresh_safety_window_secs, &None);
+            },
+        );
+    }
+
+    #[test]
+    fn test_from_env_kubernetes_is_mutually_exclusive_with_other_auth() {
+        with_vars(
+            vec![
+                ("RUSTFS_KMS_BACKEND", Some("vault-transit")),
+                (ENV_KMS_VAULT_KUBERNETES_ROLE, Some("rustfs")),
+                (ENV_KMS_VAULT_APPROLE_ROLE_ID, Some("env-role-id")),
+                (ENV_KMS_VAULT_TOKEN, None),
+                (ENV_KMS_VAULT_TOKEN_FILE, None),
+            ],
+            || {
+                let error = KmsConfig::from_env().expect_err("kubernetes combined with approle must be rejected");
+                assert!(error.to_string().contains(ENV_KMS_VAULT_KUBERNETES_ROLE));
+                assert!(error.to_string().contains(ENV_KMS_VAULT_APPROLE_ROLE_ID));
+            },
+        );
+    }
+
+    #[test]
+    fn test_validate_rejects_bad_kubernetes_settings() {
+        let vault_config = |auth_method: VaultAuthMethod| KmsConfig {
+            backend: KmsBackend::VaultTransit,
+            backend_config: BackendConfig::VaultTransit(Box::new(VaultTransitConfig {
+                address: "https://vault.example.com:8200".to_string(),
+                auth_method,
+                ..Default::default()
+            })),
+            ..Default::default()
+        };
+
+        let error = vault_config(VaultAuthMethod::kubernetes(String::new()))
+            .validate()
+            .expect_err("an empty kubernetes role must be rejected");
+        assert!(error.to_string().contains("role"), "got {error}");
+
+        let error = vault_config(VaultAuthMethod::Kubernetes {
+            role: "rustfs".to_string(),
+            mount: String::new(),
+            jwt_path: PathBuf::from(DEFAULT_VAULT_KUBERNETES_JWT_PATH),
+            refresh_safety_window_secs: None,
+        })
+        .validate()
+        .expect_err("an empty kubernetes mount must be rejected");
+        assert!(error.to_string().contains("mount"), "got {error}");
+
+        let error = vault_config(VaultAuthMethod::Kubernetes {
+            role: "rustfs".to_string(),
+            mount: DEFAULT_VAULT_KUBERNETES_MOUNT.to_string(),
+            jwt_path: PathBuf::new(),
+            refresh_safety_window_secs: None,
+        })
+        .validate()
+        .expect_err("an empty ServiceAccount token path must be rejected");
+        assert!(error.to_string().contains("token path"), "got {error}");
+
+        vault_config(VaultAuthMethod::kubernetes("rustfs".to_string()))
+            .validate()
+            .expect("well-formed kubernetes auth must validate");
+    }
+
     /// Every KV2 read, write and listing is routed through `kv_mount`, so an
     /// empty one names a path no Vault engine answers. The Transit backend
     /// already rejects its own empty mounts; this closes the same gap on the

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

@@ -258,7 +258,7 @@ pub struct SRLDAPUser {
     pub api_version: Option<String>,
 }
 
-#[derive(Debug, Serialize, Deserialize, Default)]
+#[derive(Debug, Clone, Serialize, Deserialize, Default)]
 pub struct SRIAMUser {
     #[serde(rename = "accessKey", default)]
     pub access_key: String,
@@ -270,7 +270,7 @@ pub struct SRIAMUser {
     pub api_version: Option<String>,
 }
 
-#[derive(Debug, Serialize, Deserialize, Default)]
+#[derive(Debug, Clone, Serialize, Deserialize, Default)]
 pub struct SRGroupInfo {
     #[serde(rename = "updateReq", default)]
     pub update_req: GroupAddRemove,
@@ -346,7 +346,7 @@ pub struct SRCredInfo {
     pub api_version: Option<String>,
 }
 
-#[derive(Debug, Serialize, Deserialize, Default)]
+#[derive(Debug, Clone, Serialize, Deserialize, Default)]
 pub struct SRIAMItem {
     #[serde(default)]
     pub r#type: String,

crates/notify/src/rules/config.rs

@@ -40,7 +40,10 @@ impl RuleEvents for RuleView {
 #[derive(Debug)]
 struct CompiledRules {
     // Keep RulesMap (can be used later if you want to make more complex judgments during the snapshot reading phase)
-    #[allow(dead_code)]
+    #[allow(
+        dead_code,
+        reason = "speculative retention: the comment above keeps it for richer snapshot-time judgements that no code performs yet (backlog#1823)"
+    )]
     rules_map: RulesMap,
     // for RulesContainer::iter_rules
     rule_views: Vec<RuleView>,

crates/notify/src/rules/rules_map.rs

@@ -187,7 +187,6 @@ impl RulesMap {
     /// # Parameters
     /// * `event_name` - The EventName from which to remove the rule.
     /// * `pattern` - The pattern of the rule to be removed.
-    #[allow(dead_code)]
     pub fn remove_rule(&mut self, event_name: &EventName, pattern: &str) {
         let mut remove_event = false;
 
@@ -209,7 +208,6 @@ impl RulesMap {
     ///
     /// # Parameters
     /// * `event_names` - A slice of EventNames to be removed.
-    #[allow(dead_code)]
     pub fn remove_rules(&mut self, event_names: &[EventName]) {
         for event_name in event_names {
             self.map.remove(event_name);
@@ -223,7 +221,6 @@ impl RulesMap {
     /// * `event_name` - The EventName to update.
     /// * `pattern` - The pattern of the rule to be updated.
     /// * `target_id` - The TargetID to be added.
-    #[allow(dead_code)]
     pub fn update_rule(&mut self, event_name: EventName, pattern: String, target_id: TargetID) {
         self.map.entry(event_name).or_default().add(pattern, target_id);
         self.total_events_mask |= event_name.mask(); // Update only the relevant bitmask

crates/notify/src/rules/target_id_set.rs

@@ -18,12 +18,6 @@ use rustfs_targets::arn::TargetID;
 /// TargetIDSet - A collection representation of TargetID.
 pub type TargetIdSet = HashSet<TargetID>;
 
-/// Provides a Go-like method for TargetIdSet (can be implemented as trait if needed)
-#[allow(dead_code)]
-pub(crate) fn new_target_id_set(target_ids: Vec<TargetID>) -> TargetIdSet {
-    target_ids.into_iter().collect()
-}
-
 // HashSet has built-in clone, union, difference and other operations.
 // But the Go version of the method returns a new Set, and the HashSet method is usually iterator or modify itself.
 // If you need to exactly match Go's API style, you can add wrapper functions.

crates/obs/src/metrics/config.rs

@@ -17,7 +17,6 @@ use std::time::Duration;
 /// Environment variable key for the global default metrics interval (seconds).
 pub const ENV_DEFAULT_METRICS_INTERVAL: &str = "RUSTFS_METRICS_DEFAULT_INTERVAL_SEC";
 /// Default interval for metrics collection if not specified otherwise.
-#[allow(dead_code)]
 pub const DEFAULT_METRICS_INTERVAL: Duration = Duration::from_secs(60);
 
 /// Environment variable key for cluster metrics interval (seconds).

crates/obs/src/metrics/report.rs

@@ -145,21 +145,18 @@ impl PrometheusMetric {
     }
 
     #[inline]
-    #[allow(dead_code)]
     pub fn with_label(mut self, key: &'static str, value: impl Into<Cow<'static, str>>) -> Self {
         self.labels.push((key, value.into()));
         self
     }
 
     #[inline]
-    #[allow(dead_code)]
     pub fn with_label_owned(mut self, key: &'static str, value: String) -> Self {
         self.labels.push((key, Cow::Owned(value)));
         self
     }
 
     #[inline]
-    #[allow(dead_code)]
     pub fn with_labels(mut self, labels: Vec<(&'static str, Cow<'static, str>)>) -> Self {
         self.labels = labels;
         self

crates/obs/src/metrics/schema/entry/descriptor.rs

@@ -16,7 +16,6 @@ use crate::{MetricName, MetricNamespace, MetricSubsystem, MetricType};
 use std::collections::HashSet;
 
 /// MetricDescriptor - Metric descriptors
-#[allow(dead_code)]
 #[derive(Debug, Clone)]
 pub struct MetricDescriptor {
     pub name: MetricName,
@@ -52,7 +51,6 @@ impl MetricDescriptor {
     }
 
     /// Get the full metric name in Prometheus style: <namespace>_<subsystem>_<name>
-    #[allow(dead_code)]
     pub fn get_full_metric_name(&self) -> String {
         let namespace = self.namespace.as_str();
         let formatted_subsystem = self.subsystem.as_str();
@@ -61,7 +59,6 @@ impl MetricDescriptor {
     }
 
     /// check whether the label is in the label set
-    #[allow(dead_code)]
     pub fn has_label(&mut self, label: &str) -> bool {
         self.get_label_set().contains(label)
     }

crates/obs/src/metrics/schema/entry/metric_name.rs

@@ -13,7 +13,6 @@
 // limitations under the License.
 
 /// The metric name is the individual name of the metric
-#[allow(dead_code)]
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum MetricName {
     // The generic metric name
@@ -443,7 +442,6 @@ pub enum MetricName {
 }
 
 impl MetricName {
-    #[allow(dead_code)]
     pub fn as_str(&self) -> String {
         match self {
             Self::AuthTotal => "auth_total".to_string(),

crates/obs/src/metrics/schema/entry/metric_type.rs

@@ -13,7 +13,6 @@
 // limitations under the License.
 
 /// MetricType - Indicates the type of indicator
-#[allow(dead_code)]
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum MetricType {
     Counter,
@@ -23,7 +22,6 @@ pub enum MetricType {
 
 impl MetricType {
     /// convert the metric type to a string representation
-    #[allow(dead_code)]
     pub fn as_str(&self) -> &'static str {
         match self {
             Self::Counter => "counter",
@@ -34,7 +32,6 @@ impl MetricType {
 
     /// Convert the metric type to the Prometheus value type
     /// In a Rust implementation, this might return the corresponding Prometheus Rust client type
-    #[allow(dead_code)]
     pub fn as_prom(&self) -> &'static str {
         match self {
             Self::Counter => "counter.",

crates/obs/src/metrics/schema/entry/mod.rs

@@ -56,7 +56,6 @@ pub fn new_gauge_md(
 }
 
 /// create a new histogram indicator descriptor
-#[allow(dead_code)]
 pub fn new_histogram_md(
     name: impl Into<MetricName>,
     help: impl Into<String>,

crates/obs/src/metrics/schema/entry/namespace.rs

@@ -19,7 +19,6 @@ pub enum MetricNamespace {
 }
 
 impl MetricNamespace {
-    #[allow(dead_code)]
     pub fn as_str(&self) -> &'static str {
         match self {
             Self::RustFS => "rustfs",

crates/obs/src/metrics/schema/entry/path_utils.rs

@@ -14,7 +14,6 @@
 
 /// Format the path to the metric name format
 /// Replace '/' and '-' with '_'
-#[allow(dead_code)]
 pub fn format_path_to_metric_name(path: &str) -> String {
     path.trim_start_matches('/').replace(['/', '-'], "_")
 }

crates/obs/src/metrics/schema/entry/subsystem.rs

@@ -102,7 +102,6 @@ impl MetricSubsystem {
     }
 
     /// Get the formatted metric name format string
-    #[allow(dead_code)]
     pub fn as_str(&self) -> String {
         format_path_to_metric_name(self.path())
     }
@@ -151,7 +150,6 @@ impl MetricSubsystem {
     }
 
     /// A convenient way to create custom subsystems directly
-    #[allow(dead_code)]
     pub fn new(path: impl Into<String>) -> Self {
         Self::Custom(path.into())
     }
@@ -176,7 +174,6 @@ impl std::fmt::Display for MetricSubsystem {
     }
 }
 
-#[allow(dead_code)]
 pub mod subsystems {
     use super::MetricSubsystem;
 

crates/obs/src/telemetry/rolling.rs

@@ -38,7 +38,10 @@ pub enum Rotation {
     Minutely,
     Hourly,
     Daily,
-    #[allow(dead_code)]
+    #[allow(
+        dead_code,
+        reason = "constructed only by this file's rolling-appender tests; the lib target cannot see them (backlog#1823)"
+    )]
     Never,
 }
 

crates/policy/src/policy/function.rs

@@ -219,10 +219,6 @@ impl PartialEq for Functions {
     }
 }
 
-#[derive(Clone, Serialize, Deserialize)]
-#[allow(dead_code)]
-pub struct Value;
-
 #[cfg(test)]
 mod tests {
     use crate::policy::Functions;

crates/policy/src/policy/utils/wildcard.rs

@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#[allow(dead_code)]
 pub fn is_simple_match<P, N>(pattern: P, name: N) -> bool
 where
     P: AsRef<str>,
@@ -29,7 +28,10 @@ where
     inner_match(pattern, name, false)
 }
 
-#[allow(dead_code)]
+#[allow(
+    dead_code,
+    reason = "prefix-matcher asserted by this file's tests; no production caller yet (backlog#1823)"
+)]
 pub fn is_match_as_pattern_prefix<P, N>(pattern: P, text: N) -> bool
 where
     P: AsRef<str>,

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/trusted-proxies/src/cloud/metadata/aws.rs

@@ -27,6 +27,10 @@ use crate::CloudMetadataFetcher;
 #[derive(Debug, Clone)]
 pub struct AwsMetadataFetcher {
     client: Client,
+    #[allow(
+        dead_code,
+        reason = "IMDS endpoint retained beside the client it configures; requests build their own URLs (backlog#1823)"
+    )]
     metadata_endpoint: String,
 }
 
@@ -46,55 +50,6 @@ impl AwsMetadataFetcher {
             metadata_endpoint: "http://169.254.169.254".to_string(),
         }
     }
-
-    /// Retrieves an IMDSv2 token for secure metadata access.
-    #[allow(dead_code)]
-    async fn get_metadata_token(&self) -> Result<String, AppError> {
-        let url = format!("{}/latest/api/token", self.metadata_endpoint);
-
-        match self
-            .client
-            .put(&url)
-            .header("X-aws-ec2-metadata-token-ttl-seconds", "21600")
-            .send()
-            .await
-        {
-            Ok(response) => {
-                if response.status().is_success() {
-                    let token = response
-                        .text()
-                        .await
-                        .map_err(|e| AppError::cloud(format!("Failed to read IMDSv2 token: {}", e)))?;
-                    Ok(token)
-                } else {
-                    debug!(
-                        event = "trusted_proxies.cloud_metadata",
-                        component = "trusted_proxies",
-                        subsystem = "aws_metadata",
-                        provider = "aws",
-                        operation = "imdsv2_token",
-                        result = "http_error",
-                        status = %response.status(),
-                        "trusted proxy cloud metadata request failed"
-                    );
-                    Err(AppError::cloud("Failed to obtain IMDSv2 token"))
-                }
-            }
-            Err(e) => {
-                debug!(
-                    event = "trusted_proxies.cloud_metadata",
-                    component = "trusted_proxies",
-                    subsystem = "aws_metadata",
-                    provider = "aws",
-                    operation = "imdsv2_token",
-                    result = "request_failed",
-                    error = %e,
-                    "trusted proxy cloud metadata request failed"
-                );
-                Err(AppError::cloud(format!("IMDSv2 request failed: {}", e)))
-            }
-        }
-    }
 }
 
 #[async_trait]

crates/trusted-proxies/src/config/env.rs

@@ -68,7 +68,6 @@ pub fn is_env_set(key: &str) -> bool {
 }
 
 /// Returns a list of all proxy-related environment variables and their current values.
-#[allow(dead_code)]
 pub fn get_all_proxy_env_vars() -> Vec<(String, String)> {
     let vars = [
         ENV_TRUSTED_PROXY_ENABLED,

crates/utils/src/io.rs

@@ -68,7 +68,6 @@ pub async fn read_full_or_eof<R: AsyncRead + Send + Sync + Unpin>(
 
 /// Read exactly buf.len() bytes into buf, or return an error if EOF is reached before any bytes are read.
 /// Like Go's io.ReadFull.
-#[allow(dead_code)]
 pub async fn read_full<R: AsyncRead + Send + Sync + Unpin>(reader: R, buf: &mut [u8]) -> std::io::Result<usize> {
     match read_full_or_eof(reader, buf).await? {
         Some(n) => Ok(n),

crates/utils/src/net.rs

@@ -431,7 +431,6 @@ pub fn parse_and_resolve_address(addr_str: &str) -> std::io::Result<SocketAddr>
     Ok(resolved_addr)
 }
 
-#[allow(dead_code)]
 pub fn bytes_stream<S, E>(stream: S, content_length: usize) -> impl Stream<Item = Result<Bytes, E>> + Send + 'static
 where
     S: Stream<Item = Result<Bytes, E>> + Send + 'static,

crates/utils/src/os/fs_type.rs

@@ -16,7 +16,6 @@
 ///
 /// The table follows Linux `include/uapi/linux/magic.h`; filesystem magic
 /// values without a stable Linux uapi source stay `UNKNOWN`.
-#[allow(dead_code)]
 pub(crate) fn get_fs_type(fs_type: u64) -> &'static str {
     // Magic numbers for various filesystems.
     match fs_type {

crates/utils/src/path.rs

@@ -70,7 +70,6 @@ pub fn is_dir_object(object: &str) -> bool {
 ///
 /// If the object name ends with `GLOBAL_DIR_SUFFIX`, it is replaced with a slash.
 /// Otherwise, the name is returned as is.
-#[allow(dead_code)]
 pub fn decode_dir_object(object: &str) -> String {
     if has_suffix(object, GLOBAL_DIR_SUFFIX) {
         format!("{}{}", object.trim_end_matches(GLOBAL_DIR_SUFFIX), SLASH_SEPARATOR)