test(e2e): pin bounded physical reads for compressed multipart range GETs

Byte-exactness tests stay green if the compressed range seek regresses into
decoding from byte zero: the returned bytes are still correct and only the
read amplification explodes. Assert the cost side as well.

The observation reuses rustfs_io_get_object_shard_read_observed_bytes_total,
already emitted per shard read by the erasure layer, so no production code is
instrumented. The OTLP collector learns to accumulate a second counter, keyed
by its path/role/outcome labels rather than by data-point position, which is
not stable across exports.

Two failure modes the assertions guard against:

- With RUSTFS_OBS_METER_INTERVAL=1, treating one unchanged sample as settled
  measures a delta of zero, because the range read's counter has not been
  exported yet. Settling now requires several consecutive equal samples.
- An upper bound alone passes vacuously on a zero delta, so a lower bound
  turns "measured nothing" into a failure instead of a green run.

Refs rustfs/rustfs#5957, backlog#1848.

Cargo.lock

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

_typos.toml

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

crates/data-usage/src/data_usage.rs

@@ -870,157 +870,6 @@ 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`)
@@ -1148,21 +997,6 @@ 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;
@@ -2064,126 +1898,6 @@ 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/e2e_test/src/inline_fast_path_cluster_test.rs

@@ -67,6 +67,9 @@ type MetricValues = Arc<Mutex<BTreeMap<String, MetricPointVersions>>>;
 
 const KIB: usize = 1024;
 const READER_PATH_COUNTER: &str = "rustfs_io_get_object_reader_path_by_size_total";
+/// Physical bytes the erasure layer pulled from disk, emitted per shard read by
+/// `crates/ecstore/src/erasure/coding/decode.rs`.
+const SHARD_READ_BYTES_COUNTER: &str = "rustfs_io_get_object_shard_read_observed_bytes_total";
 const MSGPACK_JSON_DECODE_COUNTER: &str = "rustfs_system_network_internode_msgpack_json_decode_total";
 const MSGPACK_JSON_FALLBACK_COUNTER: &str = "rustfs_system_network_internode_msgpack_json_fallback_total";
 const MSGPACK_JSON_DECODE_ERROR_COUNTER: &str = "rustfs_system_network_internode_msgpack_json_decode_error_total";
@@ -146,6 +149,7 @@ struct OtlpMetricCollector {
     decode_values: MetricValues,
     fallback_values: MetricValues,
     decode_error_values: MetricValues,
+    shard_read_values: MetricValues,
     task: JoinHandle<()>,
 }
 
@@ -157,10 +161,12 @@ impl OtlpMetricCollector {
         let decode_values = Arc::new(Mutex::new(BTreeMap::new()));
         let fallback_values = Arc::new(Mutex::new(BTreeMap::new()));
         let decode_error_values = Arc::new(Mutex::new(BTreeMap::new()));
+        let shard_read_values = Arc::new(Mutex::new(BTreeMap::new()));
         let task_values = values.clone();
         let task_decode_values = decode_values.clone();
         let task_fallback_values = fallback_values.clone();
         let task_decode_error_values = decode_error_values.clone();
+        let task_shard_read_values = shard_read_values.clone();
         let task = tokio::spawn(async move {
             loop {
                 let Ok((stream, _)) = listener.accept().await else {
@@ -170,6 +176,7 @@ impl OtlpMetricCollector {
                 let decode_values = task_decode_values.clone();
                 let fallback_values = task_fallback_values.clone();
                 let decode_error_values = task_decode_error_values.clone();
+                let shard_read_values = task_shard_read_values.clone();
                 tokio::spawn(async move {
                     let _ = hyper::server::conn::http1::Builder::new()
                         .serve_connection(
@@ -181,6 +188,7 @@ impl OtlpMetricCollector {
                                     decode_values.clone(),
                                     fallback_values.clone(),
                                     decode_error_values.clone(),
+                                    shard_read_values.clone(),
                                 )
                             }),
                         )
@@ -194,10 +202,48 @@ impl OtlpMetricCollector {
             decode_values,
             fallback_values,
             decode_error_values,
+            shard_read_values,
             task,
         })
     }
 
+    /// Total physical bytes read from disk across every shard-read label set.
+    async fn shard_read_bytes_total(&self) -> u64 {
+        self.shard_read_values
+            .lock()
+            .await
+            .values()
+            .map(|versions| versions.values().map(|(_, value)| *value).sum::<u64>())
+            .sum()
+    }
+
+    /// Waits until the shard-read counter stops advancing so a measurement window
+    /// is not polluted by exports still in flight.
+    ///
+    /// Requires several consecutive equal samples spanning more than one export
+    /// interval (`RUSTFS_OBS_METER_INTERVAL=1`): a single unchanged sample only
+    /// proves the latest export has not landed yet, which silently reads as "no
+    /// disk reads happened" and makes any upper-bound assertion vacuous.
+    async fn wait_for_shard_read_bytes_to_settle(&self) -> TestResult<u64> {
+        const REQUIRED_STABLE_SAMPLES: usize = 5;
+        let mut last = self.shard_read_bytes_total().await;
+        let mut stable = 0;
+        for _ in 0..60 {
+            sleep(Duration::from_millis(500)).await;
+            let current = self.shard_read_bytes_total().await;
+            if current == last {
+                stable += 1;
+                if stable >= REQUIRED_STABLE_SAMPLES {
+                    return Ok(current);
+                }
+            } else {
+                stable = 0;
+                last = current;
+            }
+        }
+        Err("timed out waiting for shard-read byte counter to settle".into())
+    }
+
     async fn reader_path_total(&self, path: &str, object_class: &str, size_bucket: &str) -> u64 {
         self.reader_path_values(path, object_class, size_bucket).await.values().sum()
     }
@@ -321,6 +367,7 @@ async fn handle_metric_export(
     decode_values: MetricValues,
     fallback_values: MetricValues,
     decode_error_values: MetricValues,
+    shard_read_values: MetricValues,
 ) -> Result<Response<Full<Bytes>>, Infallible> {
     if request.uri().path() != "/v1/metrics" {
         return Ok(response(StatusCode::NOT_FOUND));
@@ -354,7 +401,9 @@ async fn handle_metric_export(
             let mut decode_values = decode_values.lock().await;
             let mut fallback_values = fallback_values.lock().await;
             let mut decode_error_values = decode_error_values.lock().await;
+            let mut shard_read_values = shard_read_values.lock().await;
             record_reader_path_metrics(&export, &mut values);
+            record_shard_read_bytes_metrics(&export, &mut shard_read_values);
             record_msgpack_decode_metrics(&export, &mut decode_values);
             record_msgpack_fallback_metrics(&export, &mut fallback_values);
             record_msgpack_decode_error_metrics(&export, &mut decode_error_values);
@@ -375,6 +424,50 @@ fn reader_path_metric_key(path: &str, object_class: &str, size_bucket: &str) ->
     format!("{path}\u{1f}{object_class}\u{1f}{size_bucket}")
 }
 
+/// Accumulates `SHARD_READ_BYTES_COUNTER` across all label sets. Only the total
+/// matters: it is the number of physical bytes the erasure layer actually pulled
+/// from disk, which is what separates a bounded per-part read from a decode of
+/// the whole object.
+fn record_shard_read_bytes_metrics(export: &ExportMetricsServiceRequest, values: &mut BTreeMap<String, MetricPointVersions>) {
+    for resource_metrics in &export.resource_metrics {
+        for scope_metrics in &resource_metrics.scope_metrics {
+            for metric in &scope_metrics.metrics {
+                if metric.name != SHARD_READ_BYTES_COUNTER {
+                    continue;
+                }
+                let Some(metric::Data::Sum(sum)) = &metric.data else {
+                    continue;
+                };
+                for point in &sum.data_points {
+                    let Some(number_data_point::Value::AsInt(value)) = point.value.as_ref() else {
+                        continue;
+                    };
+                    let value = u64::try_from(*value).unwrap_or_default();
+                    // Keyed by labels, not by position: point order within an export
+                    // is not guaranteed stable, so an index key would alias distinct
+                    // series across batches.
+                    let key = format!(
+                        "{}\u{1f}{}\u{1f}{}",
+                        attribute_string(&point.attributes, "path").unwrap_or_default(),
+                        attribute_string(&point.attributes, "role").unwrap_or_default(),
+                        attribute_string(&point.attributes, "outcome").unwrap_or_default(),
+                    );
+                    values
+                        .entry(key)
+                        .or_default()
+                        .entry(point.start_time_unix_nano)
+                        .and_modify(|current| {
+                            if point.time_unix_nano >= current.0 {
+                                *current = (point.time_unix_nano, value);
+                            }
+                        })
+                        .or_insert((point.time_unix_nano, value));
+                }
+            }
+        }
+    }
+}
+
 fn record_reader_path_metrics(export: &ExportMetricsServiceRequest, values: &mut BTreeMap<String, MetricPointVersions>) {
     for resource_metrics in &export.resource_metrics {
         for scope_metrics in &resource_metrics.scope_metrics {
@@ -1864,6 +1957,86 @@ async fn four_node_multipart_disk_compression_roundtrip() -> TestResult {
     Ok(())
 }
 
+/// A tail range over a compressed multipart object must read only the physical
+/// data it needs, not decode the object from byte zero.
+///
+/// The byte-exactness tests around this one stay green even if the seek path
+/// regresses into decoding from the start of the object: the bytes returned are
+/// still correct, only the read amplification explodes. This asserts the cost
+/// side, using `SHARD_READ_BYTES_COUNTER` — already emitted per shard read by the
+/// erasure layer, so no production code is instrumented for the test.
+///
+/// `get_compressed_offsets` skips whole preceding parts by their stored size and
+/// then seeks inside the covering part via its compression index, so a bounded
+/// read costs on the order of the covering part's block size against a ~5 MiB
+/// object.
+#[tokio::test]
+#[serial]
+async fn four_node_compressed_multipart_tail_range_reads_are_bounded() -> TestResult {
+    init_logging();
+
+    let collector = OtlpMetricCollector::start().await?;
+    let mut cluster = RustFSTestClusterEnvironment::new(4).await?;
+    configure_reader_metric_cluster(&mut cluster, &collector);
+    cluster.set_env("RUSTFS_COMPRESSION_ENABLED", "true");
+    cluster.set_env("RUSTFS_COMPRESSION_MULTIPART_ENABLED", "true");
+    cluster.start().await?;
+
+    let bucket = "inline-multipart-compression-tail-range";
+    cluster.create_test_bucket(bucket).await?;
+    let client = cluster.create_s3_client(0)?;
+    let key = "multipart/tail-range.txt";
+    let (body, _second_part, etag) = put_two_part_multipart(&client, bucket, key).await?;
+
+    // Establish that the object really took the compressed read path; otherwise a
+    // small delta below would only prove compression never happened.
+    assert_reader_path(
+        &collector,
+        &client,
+        ReaderPathExpectation::for_class(ReaderObject::new(bucket, key, &body, etag.as_deref(), None), LEGACY_DUPLEX, COMPRESSED),
+    )
+    .await?;
+
+    let baseline = collector.wait_for_shard_read_bytes_to_settle().await?;
+
+    let tail_len = 4 * KIB;
+    let start = body.len() - tail_len;
+    let end = body.len() - 1;
+    let range = client
+        .get_object()
+        .bucket(bucket)
+        .key(key)
+        .range(format!("bytes={start}-{end}"))
+        .send()
+        .await?;
+    let tail = range.body.collect().await?.into_bytes();
+    assert_eq!(tail.as_ref(), &body[start..], "tail range returned wrong bytes");
+
+    let after = collector.wait_for_shard_read_bytes_to_settle().await?;
+    let read_bytes = after.saturating_sub(baseline);
+
+    // A zero delta means the window caught nothing — an unexported counter, or a
+    // read served without touching the erasure layer — which would make the upper
+    // bound vacuously true. Fail instead of passing blind.
+    assert!(
+        read_bytes > 0,
+        "no shard reads observed for the tail range; the budget assertion below would be vacuous"
+    );
+
+    // Part 1 alone is MPU_PART_1_SIZE, so a whole-object decode cannot come in
+    // under it. Half the logical size leaves generous headroom for erasure padding
+    // and unrelated background reads while still failing loudly on a full decode.
+    let budget = (body.len() / 2) as u64;
+    assert!(
+        read_bytes < budget,
+        "tail range read {read_bytes} physical bytes for a {tail_len}-byte range (budget {budget}, object {} bytes): \
+         the read is not bounded to the covering part",
+        body.len()
+    );
+
+    Ok(())
+}
+
 #[tokio::test]
 #[serial]
 async fn four_node_mixed_msgpack_compat_mode_preserves_fallback_controls() -> TestResult {

crates/ecstore/src/api/mod.rs

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

crates/ecstore/src/bucket/bucket_target_sys.rs

@@ -1549,8 +1549,8 @@ impl Default for PutObjectOptions {
     }
 }
 
+#[allow(dead_code)]
 impl PutObjectOptions {
-    #[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
     fn set_match_etag(&mut self, etag: &str) {
         if etag == "*" {
             self.custom_header
@@ -1561,7 +1561,6 @@ impl PutObjectOptions {
         }
     }
 
-    #[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
     fn set_match_etag_except(&mut self, etag: &str) {
         if etag == "*" {
             self.custom_header
@@ -1697,7 +1696,6 @@ impl PutObjectOptions {
         header
     }
 
-    #[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
     fn validate(&self, _c: Arc<TargetClient>) -> Result<(), std::io::Error> {
         //if self.checksum.is_set() {
         /*if !self.trailing_header_support {

crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs

@@ -456,23 +456,16 @@ impl<'a> LifecycleExpiryTrace<'a> {
     }
 }
 
+#[allow(dead_code)]
 impl ExpiryStats {
     pub fn missed_tasks(&self) -> i64 {
         self.missed_expiry_tasks.load(Ordering::SeqCst)
     }
 
-    #[allow(
-        dead_code,
-        reason = "asserted by this file's tests; the lib target cannot see test-only consumers (backlog#1823)"
-    )]
     fn missed_free_vers_tasks(&self) -> i64 {
         self.missed_freevers_tasks.load(Ordering::SeqCst)
     }
 
-    #[allow(
-        dead_code,
-        reason = "asserted by this file's tests; the lib target cannot see test-only consumers (backlog#1823)"
-    )]
     fn missed_tier_journal_tasks(&self) -> i64 {
         self.missed_tier_journal_tasks.load(Ordering::SeqCst)
     }

crates/ecstore/src/bucket/lifecycle/tier_last_day_stats.rs

@@ -80,10 +80,7 @@ impl LastDayTierStats {
         }
     }
 
-    #[allow(
-        dead_code,
-        reason = "asserted by this file's tests; the lib target cannot see test-only consumers (backlog#1823)"
-    )]
+    #[allow(dead_code)]
     fn merge(&self, m: LastDayTierStats) -> LastDayTierStats {
         let mut cl = self.clone();
         let mut cm = m;

crates/ecstore/src/bucket/lifecycle/tier_sweeper.rs

@@ -177,10 +177,9 @@ fn should_record_remote_delete_failure(err: &std::io::Error) -> bool {
 }
 
 #[derive(Default)]
+#[allow(dead_code)]
 struct ObjSweeper {
-    #[allow(dead_code, reason = "written but never read back (backlog#1823)")]
     object: String,
-    #[allow(dead_code, reason = "written but never read back (backlog#1823)")]
     bucket: String,
     version_id: Option<Uuid>,
     versioned: bool,
@@ -192,9 +191,9 @@ struct ObjSweeper {
     remote_object: String,
 }
 
+#[allow(dead_code)]
 impl ObjSweeper {
     #[allow(clippy::new_ret_no_self)]
-    #[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
     pub async fn new(bucket: &str, object: &str) -> Result<Self, std::io::Error> {
         Ok(Self {
             object: object.into(),
@@ -203,20 +202,17 @@ impl ObjSweeper {
         })
     }
 
-    #[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
     pub fn with_version(&mut self, vid: Option<Uuid>) -> &Self {
         self.version_id = vid.clone();
         self
     }
 
-    #[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
     pub fn with_versioning(&mut self, versioned: bool, suspended: bool) -> &Self {
         self.versioned = versioned;
         self.suspended = suspended;
         self
     }
 
-    #[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
     pub fn get_opts(&self) -> lifecycle::ObjectOpts {
         let mut opts = ObjectOpts {
             version_id: self.version_id.clone(),
@@ -230,7 +226,6 @@ impl ObjSweeper {
         opts
     }
 
-    #[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
     pub fn set_transition_state(&mut self, info: TransitionedObject) {
         self.transition_tier = info.tier;
         self.transition_status = info.status;
@@ -271,7 +266,6 @@ impl ObjSweeper {
         None
     }
 
-    #[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
     pub async fn sweep(&self, api: Arc<ECStore>) {
         let Some(je) = self.should_remove_remote_object() else {
             return;

crates/ecstore/src/bucket/quota/mod.rs

@@ -312,7 +312,9 @@ mod tests {
         }
         #[derive(Deserialize)]
         struct LegacyBucketQuota {
+            #[allow(dead_code)]
             quota: Option<u64>,
+            #[allow(dead_code)]
             quota_type: LegacyQuotaType,
         }
         let legacy = serde_json::from_slice::<LegacyBucketQuota>(&json)

crates/ecstore/src/client/api_get_object.rs

@@ -95,6 +95,7 @@ impl TransitionClient {
 }
 
 #[derive(Default)]
+#[allow(dead_code)]
 pub struct GetRequest {
     pub buffer: Vec<u8>,
     pub offset: i64,
@@ -106,12 +107,11 @@ pub struct GetRequest {
     pub setting_object_info: bool,
 }
 
+#[allow(dead_code)]
 pub struct GetResponse {
     pub size: i64,
     //pub error:       error,
-    #[allow(dead_code, reason = "written but never read back (backlog#1823)")]
     pub did_read: bool,
-    #[allow(dead_code, reason = "written but never read back (backlog#1823)")]
     pub object_info: ObjectInfo,
 }
 

crates/ecstore/src/client/api_get_options.rs

@@ -27,6 +27,7 @@ use tracing::warn;
 use crate::client::api_error_response::err_invalid_argument;
 
 #[derive(Default)]
+#[allow(dead_code)]
 pub struct AdvancedGetOptions {
     pub replication_delete_marker: bool,
     pub is_replication_ready_for_delete_marker: bool,

crates/ecstore/src/client/api_list.rs

@@ -360,6 +360,7 @@ impl TransitionClient {
 }
 
 #[derive(Default)]
+#[allow(dead_code)]
 pub struct ListObjectsOptions {
     reverse_versions: bool,
     with_versions: bool,

crates/ecstore/src/client/api_put_object.rs

@@ -137,8 +137,8 @@ impl Default for PutObjectOptions {
     }
 }
 
+#[allow(dead_code)]
 impl PutObjectOptions {
-    #[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
     fn set_match_etag(&mut self, etag: &str) {
         if etag == "*" {
             self.custom_header.insert("If-Match", HeaderValue::from_static("*"));
@@ -149,7 +149,6 @@ impl PutObjectOptions {
         }
     }
 
-    #[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
     fn set_match_etag_except(&mut self, etag: &str) {
         if etag == "*" {
             self.custom_header.insert("If-None-Match", HeaderValue::from_static("*"));
@@ -260,7 +259,6 @@ impl PutObjectOptions {
         header
     }
 
-    #[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
     fn validate(&self, c: TransitionClient) -> Result<(), std::io::Error> {
         //if self.checksum.is_set() {
         /*if !self.trailing_header_support {

crates/ecstore/src/client/api_remove.rs

@@ -55,6 +55,7 @@ pub struct RemoveBucketOptions {
 const DELETE_RESPONSE_PREVIEW_LEN: usize = 1024;
 
 #[derive(Debug)]
+#[allow(dead_code)]
 pub struct AdvancedRemoveOptions {
     pub replication_delete_marker: bool,
     pub replication_status: ReplicationStatus,
@@ -464,10 +465,10 @@ impl TransitionClient {
 }
 
 #[derive(Debug, Default)]
+#[allow(dead_code)]
 pub struct RemoveObjectError {
-    #[allow(dead_code, reason = "written but never read back (backlog#1823)")]
     object_name: String,
-    #[allow(dead_code, reason = "written but never read back (backlog#1823)")]
+    #[allow(dead_code)]
     version_id: String,
     err: Option<std::io::Error>,
 }

crates/ecstore/src/client/checksum.rs

@@ -372,8 +372,8 @@ pub struct Checksum {
     computed: bool,
 }
 
+#[allow(dead_code)]
 impl Checksum {
-    #[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
     fn new(t: ChecksumMode, b: &[u8]) -> Checksum {
         if t.is_set() && b.len() == t.raw_byte_len() {
             return Checksum {
@@ -385,7 +385,7 @@ impl Checksum {
         Checksum::default()
     }
 
-    #[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
+    #[allow(dead_code)]
     fn new_checksum_string(t: ChecksumMode, s: &str) -> Result<Checksum, std::io::Error> {
         let b = match base64_decode(s.as_bytes()) {
             Ok(b) => b,
@@ -412,7 +412,7 @@ impl Checksum {
         base64_encode(&self.r)
     }
 
-    #[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
+    #[allow(dead_code)]
     fn raw(&self) -> Option<Vec<u8>> {
         if !self.is_set() {
             return None;

crates/ecstore/src/client/object_api_utils.rs

@@ -37,17 +37,16 @@ pub struct PutObjReader {
     //pub sealMD5Fn: SealMD5CurrFn,
 }
 
+#[allow(dead_code)]
 impl PutObjReader {
     pub fn new(reader: HashReader) -> Self {
         Self { reader }
     }
 
-    #[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
     fn md5_current_hex_string(&self) -> String {
         self.reader.checksum().map(|v| v.encoded).unwrap_or_default()
     }
 
-    #[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
     fn with_encryption(&mut self, enc_reader: HashReader) -> Result<(), std::io::Error> {
         self.reader = enc_reader;
 

crates/ecstore/src/config/audit.rs

@@ -39,6 +39,7 @@ use rustfs_config::{
 };
 use std::sync::LazyLock;
 
+#[allow(dead_code)]
 #[allow(clippy::declare_interior_mutable_const)]
 /// Default KVS for audit webhook settings.
 pub static DEFAULT_AUDIT_WEBHOOK_KVS: LazyLock<KVS> = LazyLock::new(|| {
@@ -116,6 +117,7 @@ pub static DEFAULT_AUDIT_WEBHOOK_KVS: LazyLock<KVS> = LazyLock::new(|| {
     ])
 });
 
+#[allow(dead_code)]
 #[allow(clippy::declare_interior_mutable_const)]
 /// Default KVS for audit MQTT settings.
 pub static DEFAULT_AUDIT_MQTT_KVS: LazyLock<KVS> = LazyLock::new(|| {
@@ -373,6 +375,7 @@ pub static DEFAULT_AUDIT_NATS_KVS: LazyLock<KVS> = LazyLock::new(|| {
     ])
 });
 
+#[allow(dead_code)]
 pub static DEFAULT_AUDIT_PULSAR_KVS: LazyLock<KVS> = LazyLock::new(|| {
     KVS(vec![
         KV {

crates/ecstore/src/config/heal.rs

@@ -12,9 +12,12 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use crate::error::{Error, Result};
 use rustfs_config::server_config::{KV, KVS};
 use rustfs_config::{DEFAULT_HEAL_BITROT_CYCLE_SECS, HEAL_BITROT_CYCLE};
+use rustfs_utils::string::parse_bool;
 use std::sync::LazyLock;
+use std::time::Duration;
 
 pub static DEFAULT_KVS: LazyLock<KVS> = LazyLock::new(|| {
     KVS(vec![KV {
@@ -23,3 +26,59 @@ pub static DEFAULT_KVS: LazyLock<KVS> = LazyLock::new(|| {
         hidden_if_empty: false,
     }])
 });
+
+#[derive(Debug, Default)]
+pub struct Config {
+    pub bitrot: String,
+    pub sleep: Duration,
+    pub io_count: usize,
+    pub drive_workers: usize,
+    pub cache: Duration,
+}
+
+impl Config {
+    pub fn bitrot_scan_cycle(&self) -> Duration {
+        self.cache
+    }
+
+    pub fn get_workers(&self) -> usize {
+        self.drive_workers
+    }
+
+    pub fn update(&mut self, nopts: &Config) {
+        self.bitrot = nopts.bitrot.clone();
+        self.io_count = nopts.io_count;
+        self.sleep = nopts.sleep;
+        self.drive_workers = nopts.drive_workers;
+    }
+}
+
+const RUSTFS_BITROT_CYCLE_IN_MONTHS: u64 = 1;
+
+fn parse_bitrot_config(s: &str) -> Result<Duration> {
+    match parse_bool(s) {
+        Ok(enabled) => {
+            if enabled {
+                Ok(Duration::from_secs_f64(0.0))
+            } else {
+                Ok(Duration::from_secs_f64(-1.0))
+            }
+        }
+        Err(_) => {
+            if !s.ends_with("m") {
+                return Err(Error::other("unknown format"));
+            }
+
+            match s.trim_end_matches('m').parse::<u64>() {
+                Ok(months) => {
+                    if months < RUSTFS_BITROT_CYCLE_IN_MONTHS {
+                        return Err(Error::other(format!("minimum bitrot cycle is {RUSTFS_BITROT_CYCLE_IN_MONTHS} month(s)")));
+                    }
+
+                    Ok(Duration::from_secs(months * 30 * 24 * 60))
+                }
+                Err(err) => Err(Error::other(err)),
+            }
+        }
+    }
+}

crates/ecstore/src/config/mod.rs

@@ -16,6 +16,7 @@
 
 mod audit;
 pub mod com;
+#[allow(dead_code)]
 pub mod heal;
 mod notify;
 mod oidc;

crates/ecstore/src/core/pools.rs

@@ -1996,11 +1996,11 @@ impl PoolMeta {
         Ok(false)
     }
 
+    #[allow(dead_code)]
     pub fn validate(&self, pools: Vec<Arc<Sets>>) -> Result<bool> {
         struct PoolInfo {
             position: usize,
             completed: bool,
-            #[allow(dead_code, reason = "written but never read back (backlog#1823)")]
             decom_started: bool,
         }
 
@@ -4958,19 +4958,13 @@ fn is_disk_online_state(state: &str) -> bool {
 }
 
 #[deprecated(since = "0.1.0", note = "Use fallback_total_capacity_dedup instead")]
-#[allow(
-    dead_code,
-    reason = "superseded by the replacement named in the comment at pools.rs:5071 (backlog#1823)"
-)]
+#[allow(dead_code)]
 fn fallback_total_capacity(disks: &[rustfs_madmin::Disk]) -> usize {
     fallback_total_capacity_dedup(disks)
 }
 
 #[deprecated(since = "0.1.0", note = "Use fallback_free_capacity_dedup instead")]
-#[allow(
-    dead_code,
-    reason = "superseded by the replacement named in the comment at pools.rs:5071 (backlog#1823)"
-)]
+#[allow(dead_code)]
 fn fallback_free_capacity(disks: &[rustfs_madmin::Disk]) -> usize {
     fallback_free_capacity_dedup(disks)
 }

crates/ecstore/src/disk/local.rs

@@ -6562,7 +6562,7 @@ impl LocalDisk {
         Ok(f)
     }
 
-    #[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
+    #[allow(dead_code)]
     fn get_metrics(&self) -> DiskMetrics {
         DiskMetrics::default()
     }

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

@@ -820,263 +820,10 @@ 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_kat_check, bitrot_self_test,
-        bitrot_self_test_payload, bitrot_shard_file_size, bitrot_verify, write_all_vectored,
+        BitrotReader, BitrotWriter, BitrotWriterWrapper, CustomWriter, bitrot_shard_file_size, bitrot_verify, write_all_vectored,
     };
     use super::{MAX_RETAINED_CHUNKS_PER_BLOCK, ShardChunkRead, ShardSource};
     use bytes::Bytes;
@@ -1343,32 +1090,6 @@ 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();
@@ -1468,7 +1189,7 @@ mod tests {
         let last = corrupt.len() - 1;
         corrupt[last] ^= 0x80;
         let err = bitrot_verify(
-            std::io::Cursor::new(corrupt),
+            Cursor::new(corrupt),
             super::bitrot_shard_file_size(data.len(), shard_size, algo.clone()),
             data.len(),
             algo,
@@ -1561,7 +1282,7 @@ mod tests {
 
     #[tokio::test]
     async fn bitrot_reader_rejects_output_buffers_larger_than_shard_size() {
-        let mut reader = BitrotReader::new(std::io::Cursor::new(Vec::<u8>::new()), 4, HashAlgorithm::None, false);
+        let mut reader = BitrotReader::new(Cursor::new(Vec::<u8>::new()), 4, HashAlgorithm::None, false);
         let mut out = [0u8; 5];
         let err = reader
             .read(&mut out)
@@ -1686,7 +1407,7 @@ mod tests {
             (HashAlgorithm::HighwayHash256, true),
         ] {
             let label = format!("{algo:?}");
-            let writer = std::io::Cursor::new(Vec::<u8>::new());
+            let writer = 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();
@@ -1771,7 +1492,7 @@ mod tests {
     }
 
     async fn encode_one_block(payload: &[u8], shard_size: usize, algo: HashAlgorithm) -> Vec<u8> {
-        let mut w = BitrotWriter::new(std::io::Cursor::new(Vec::<u8>::new()), shard_size, algo);
+        let mut w = BitrotWriter::new(Cursor::new(Vec::<u8>::new()), shard_size, algo);
         w.write(payload).await.unwrap();
         w.into_inner().into_inner()
     }
@@ -1879,7 +1600,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(std::io::Cursor::new(Vec::<u8>::new()), shard_size, algo.clone());
+                let mut w = BitrotWriter::new(Cursor::new(Vec::<u8>::new()), shard_size, algo.clone());
                 for chunk in payload.chunks(shard_size) {
                     w.write(chunk).await.unwrap();
                 }
@@ -1953,14 +1674,14 @@ mod tests {
             w.write(&data).await.expect("write shard");
 
             let mut via_read = vec![0u8; SHARD];
-            let n1 = BitrotReader::new(std::io::Cursor::new(encoded.clone()), SHARD, algo.clone(), false)
+            let n1 = BitrotReader::new(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(std::io::Cursor::new(encoded), SHARD, algo.clone(), false)
+            let n2 = BitrotReader::new(Cursor::new(encoded), SHARD, algo.clone(), false)
                 .read_appending(&mut via_append, SHARD)
                 .await
                 .expect("read_appending");
@@ -1985,7 +1706,7 @@ mod tests {
             encoded.truncate(encoded.len() - 1);
 
             let mut out: Vec<u8> = Vec::with_capacity(SHARD);
-            let err = BitrotReader::new(std::io::Cursor::new(encoded), SHARD, algo.clone(), false)
+            let err = BitrotReader::new(Cursor::new(encoded), SHARD, algo.clone(), false)
                 .read_appending(&mut out, SHARD)
                 .await
                 .expect_err("a truncated shard must not succeed");
@@ -2011,7 +1732,7 @@ mod tests {
         encoded[last] ^= 0xff;
 
         let mut out: Vec<u8> = Vec::with_capacity(SHARD);
-        let err = BitrotReader::new(std::io::Cursor::new(encoded), SHARD, algo, false)
+        let err = BitrotReader::new(Cursor::new(encoded), SHARD, algo, false)
             .read_appending(&mut out, SHARD)
             .await
             .expect_err("a corrupt shard must not verify");
@@ -2123,7 +1844,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 = std::io::Cursor::new(encoded.clone());
+        let mut streamed = Cursor::new(encoded.clone());
         assert!(
             ShardSource::try_take_block(&mut streamed, 8).is_none(),
             "a non-Bytes source must stay on the streaming path"
@@ -2151,7 +1872,7 @@ mod tests {
         );
 
         let mut via_stream: Vec<u8> = Vec::with_capacity(SHARD);
-        BitrotReader::new(std::io::Cursor::new(encoded), SHARD, algo, false)
+        BitrotReader::new(Cursor::new(encoded), SHARD, algo, false)
             .read_appending(&mut via_stream, SHARD)
             .await
             .expect("streaming read");

crates/ecstore/src/services/rebalance/types.rs

@@ -132,6 +132,7 @@ impl RebalanceStopPropagationRecord {
     }
 }
 
+#[allow(dead_code)]
 #[derive(Debug, Clone, Default)]
 pub struct DiskStat {
     pub total_space: u64,

crates/ecstore/src/services/tier/tier_config.rs

@@ -16,16 +16,8 @@ use serde::{Deserialize, Serialize};
 use std::{fmt::Display, io};
 use tracing::info;
 
-#[allow(
-    dead_code,
-    reason = "tier config wire version stamped by the parity constructors below (backlog#1823)"
-)]
 const C_TIER_CONFIG_VER: &str = "v1";
 
-#[allow(
-    dead_code,
-    reason = "tier-name validation message reached only from the parity constructors below (backlog#1823)"
-)]
 const ERR_TIER_NAME_EMPTY: &str = "remote tier name empty";
 const WASABI_US_EAST_ENDPOINT: &str = "https://s3.wasabisys.com";
 const WASABI_ALTERNATIVE_ENDPOINTS: &[(&str, &str)] = &[
@@ -272,6 +264,7 @@ impl Clone for TierConfig {
     }
 }
 
+#[allow(dead_code)]
 impl TierConfig {
     pub(crate) fn clone_with_credentials(&self) -> Self {
         Self {
@@ -291,7 +284,6 @@ impl TierConfig {
         }
     }
 
-    #[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
     fn endpoint(&self) -> String {
         match self.tier_type {
             TierType::S3 => self.s3.as_ref().map(|s| s.endpoint.clone()).unwrap_or_default(),
@@ -311,7 +303,6 @@ impl TierConfig {
         }
     }
 
-    #[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
     fn bucket(&self) -> String {
         match self.tier_type {
             TierType::S3 => self.s3.as_ref().map(|s| s.bucket.clone()).unwrap_or_default(),
@@ -331,7 +322,6 @@ impl TierConfig {
         }
     }
 
-    #[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
     fn prefix(&self) -> String {
         match self.tier_type {
             TierType::S3 => self.s3.as_ref().map(|s| s.prefix.clone()).unwrap_or_default(),
@@ -351,7 +341,6 @@ impl TierConfig {
         }
     }
 
-    #[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
     fn region(&self) -> String {
         match self.tier_type {
             TierType::S3 => self.s3.as_ref().map(|s| s.region.clone()).unwrap_or_default(),
@@ -468,7 +457,7 @@ impl TierWasabi {
 }
 
 impl TierS3 {
-    #[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
+    #[allow(dead_code)]
     fn create<F>(
         name: &str,
         access_key: &str,
@@ -539,7 +528,7 @@ pub struct TierMinIO {
 }
 
 impl TierMinIO {
-    #[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
+    #[allow(dead_code)]
     fn create<F>(
         name: &str,
         endpoint: &str,

crates/ecstore/src/services/tier/tier_gen.rs

@@ -14,6 +14,7 @@
 
 use crate::services::tier::tier::TierConfigMgr;
 
+#[allow(dead_code)]
 impl TierConfigMgr {
     pub fn msg_size(&self) -> usize {
         100

crates/ecstore/src/store/mod.rs

@@ -216,16 +216,6 @@ 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/kms/src/api_types.rs

@@ -293,15 +293,6 @@ 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)]
@@ -328,17 +319,6 @@ 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,
@@ -519,7 +499,6 @@ 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,
@@ -534,7 +513,6 @@ 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,
@@ -923,42 +901,6 @@ 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,7 +550,6 @@ 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,97 +326,6 @@ 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).
 ///
@@ -555,9 +464,6 @@ 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,
@@ -580,9 +486,6 @@ 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 {
@@ -596,11 +499,6 @@ 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()));
@@ -653,10 +551,6 @@ impl VaultCredentialPolicy {
                 refresh_safety_window_secs: Some(secs),
                 ..
             }
-            | VaultAuthMethod::Kubernetes {
-                refresh_safety_window_secs: Some(secs),
-                ..
-            }
             | VaultAuthMethod::TokenFile {
                 refresh_safety_window_secs: Some(secs),
                 ..
@@ -690,25 +584,15 @@ 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.and_then(|lease| self.issued_at.checked_add(lease.ttl))
+        self.lease.map(|lease| self.issued_at + 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.and_then(|lease| self.issued_at.checked_add(lease.ttl / 2))
+        self.lease.map(|lease| self.issued_at + lease.ttl / 2)
     }
 }
 
@@ -778,7 +662,7 @@ impl VaultCredentialProvider {
         let handle = self.current.load_full();
         if let Some(expires_at) = handle.expires_at() {
             let now = Instant::now();
-            if self.inside_safety_window(now, expires_at) {
+            if now + self.policy.safety_window >= 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
@@ -788,18 +672,6 @@ 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
@@ -811,7 +683,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());
-                self.inside_safety_window(now, expires_at)
+                now + self.policy.safety_window >= expires_at
             }
             // A generation without an expiry has no remaining TTL to report
             // and can never lapse, so it can never fail closed either.
@@ -988,7 +860,7 @@ impl Drop for CredentialTaskHandle {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::config::{DEFAULT_VAULT_KUBERNETES_MOUNT, REDACTED_SECRET};
+    use crate::config::REDACTED_SECRET;
     use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
 
     const TEST_TOKEN: &str = "vault-token-debug-leak-canary";
@@ -999,7 +871,6 @@ 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,
         }
     }
 
@@ -1186,143 +1057,6 @@ 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,7 +415,6 @@ 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,10 +450,6 @@ 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,10 +25,6 @@ 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";
@@ -39,9 +35,6 @@ 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;
@@ -52,9 +45,6 @@ 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.
 ///
@@ -94,14 +84,6 @@ 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"),
@@ -508,23 +490,6 @@ 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.
@@ -555,16 +520,6 @@ 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 {
@@ -593,20 +548,6 @@ 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,
@@ -1087,12 +1028,50 @@ impl KmsConfig {
                 });
             }
             KmsBackend::VaultKv2 => {
-                config.backend_config =
-                    BackendConfig::VaultKv2(Box::new(vault_kv2_config_from_env(VaultCliOverrides::default())?));
+                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),
+                }));
             }
             KmsBackend::VaultTransit => {
-                config.backend_config =
-                    BackendConfig::VaultTransit(Box::new(vault_transit_config_from_env(VaultCliOverrides::default())?));
+                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),
+                }));
             }
             KmsBackend::Static => {
                 // Read from file first, then fall back to direct env var
@@ -1223,78 +1202,6 @@ 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;
@@ -1302,59 +1209,27 @@ fn vault_address_from_env(override_value: Option<&str>) -> String {
 /// (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.
-///
-/// `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);
-
+fn vault_auth_method_from_env() -> Result<VaultAuthMethod> {
     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.
-        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"
-                )));
-            }
+        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"
+            )));
         }
         return Ok(VaultAuthMethod::token_file(PathBuf::from(token_file)));
     }
 
-    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 {
+    let Some(role_id) = get_env_opt_str(ENV_KMS_VAULT_APPROLE_ROLE_ID) else {
         return Ok(VaultAuthMethod::Token {
-            token: token.unwrap_or_else(|| "dev-token".to_string()),
+            token: get_env_str("RUSTFS_KMS_VAULT_TOKEN", "dev-token"),
         });
     };
 
@@ -1398,22 +1273,6 @@ 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,
@@ -2117,106 +1976,6 @@ 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,11 +37,7 @@ 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"] }
@@ -53,4 +49,3 @@ doctest = false
 
 [dev-dependencies]
 rmp-serde.workspace = true
-tokio = { workspace = true, features = ["macros", "rt-multi-thread", "net"] }

crates/madmin/src/client.rs

@@ -1,851 +0,0 @@
-// 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,7 +12,6 @@
 // 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;
@@ -26,7 +25,6 @@ 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, Clone, Serialize, Deserialize, Default)]
+#[derive(Debug, 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, Clone, Serialize, Deserialize, Default)]
+#[derive(Debug, 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, Clone, Serialize, Deserialize, Default)]
+#[derive(Debug, Serialize, Deserialize, Default)]
 pub struct SRIAMItem {
     #[serde(default)]
     pub r#type: String,

crates/scanner/src/data_usage_define.rs

@@ -28,8 +28,7 @@ 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, PrefixUsageEntry,
-    PrefixUsageQuery, PrefixUsageSummary, TierStats, hash_path, prefix_usage_in_cache,
+    DataUsageEntry, DataUsageHash, DataUsageHashMap, DataUsageInfo, LEGACY_DATA_USAGE_OBJECT_NAME, TierStats, hash_path,
 };
 use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf};
 use tokio::time::{Duration, Instant, sleep, timeout};
@@ -431,13 +430,6 @@ 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,7 +53,6 @@ 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;
@@ -65,7 +64,6 @@ 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

@@ -1,349 +0,0 @@
-// 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,10 +231,6 @@ 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();
 }
 

docs/operations/kms-backend-security.md

@@ -2,7 +2,7 @@
 
 RustFS ships several KMS backends. They differ not only in deployment effort but in **where master key material lives and who can read it**. Pick a backend based on the confidentiality boundary you need, not on the name alone.
 
-For how the Vault backends authenticate (static token, AppRole, Kubernetes, Vault Agent token file) and how credential refresh and the fail-closed window behave, see the [Vault KMS authentication runbook](vault-kms-authentication.md). For what may be claimed about the cryptographic implementations themselves, see [Cryptographic compliance positioning](kms-cryptographic-compliance.md). For which RustFS identities may manage or use a given key, see [Per-key KMS authorization](kms-per-key-authorization.md). If you are migrating from MinIO, read [Migrating from MinIO: encrypted objects do not carry over](#migrating-from-minio-encrypted-objects-do-not-carry-over) first.
+For how the Vault backends authenticate (static token, AppRole, Vault Agent token file) and how credential refresh and the fail-closed window behave, see the [Vault KMS authentication runbook](vault-kms-authentication.md). For what may be claimed about the cryptographic implementations themselves, see [Cryptographic compliance positioning](kms-cryptographic-compliance.md). For which RustFS identities may manage or use a given key, see [Per-key KMS authorization](kms-per-key-authorization.md). If you are migrating from MinIO, read [Migrating from MinIO: encrypted objects do not carry over](#migrating-from-minio-encrypted-objects-do-not-carry-over) first.
 
 ## Backend comparison
 

docs/operations/vault-kms-authentication.md

@@ -8,12 +8,9 @@ This runbook covers how the RustFS Vault KMS backends (KV2 and Transit) authenti
 | --- | --- | --- | --- | --- |
 | Static token | `Token` | Whatever the operator provisioned; RustFS never renews it | None | Development; short-lived experiments |
 | AppRole | `AppRole` | Lease-bound token obtained by login; renewed by RustFS | Renew at half TTL, re-login on failure | Production without a Vault Agent sidecar |
-| Kubernetes | `Kubernetes` | Lease-bound token obtained by login; renewed by RustFS | Renew at half TTL, re-login on failure | Production on Kubernetes, with no credential to distribute |
 | Agent token file | `TokenFile` | Owned by Vault Agent; RustFS only re-reads the sink file | File re-read once per poll interval | Production with a Vault Agent (or equivalent) managing auth |
 
-Exactly one method must be configured. Setting `RUSTFS_KMS_VAULT_TOKEN_FILE` together with any other method, or `RUSTFS_KMS_VAULT_KUBERNETES_ROLE` together with `RUSTFS_KMS_VAULT_APPROLE_ROLE_ID`, is rejected at startup with a configuration error, because the effective identity would be ambiguous. A leftover `RUSTFS_KMS_VAULT_TOKEN` alongside a configured login method is tolerated and ignored, so a stale variable cannot silently downgrade the identity.
-
-All of these are read the same way whether the service is started with `RUSTFS_KMS_ENABLE=true` or configured later through `POST /rustfs/admin/v3/kms/configure`.
+Exactly one method must be configured. Setting `RUSTFS_KMS_VAULT_TOKEN_FILE` together with `RUSTFS_KMS_VAULT_APPROLE_ROLE_ID` or an explicit `RUSTFS_KMS_VAULT_TOKEN` is rejected at startup with a configuration error, because the effective identity would be ambiguous.
 
 The default `dev-token` fallback for `RUSTFS_KMS_VAULT_TOKEN` is rejected outside explicit development mode (`RUSTFS_KMS_ALLOW_INSECURE_DEV_DEFAULTS=true`), as are plain-HTTP Vault addresses and disabled TLS verification.
 
@@ -59,44 +56,7 @@ Deliver the SecretID out of band — a secrets-manager-mounted file, an init-con
 
 The secret_id file is re-read on every login attempt, so rotating the SecretID is a two-step operation with no restart: generate a new SecretID (`vault write -f auth/approle/role/rustfs-kms/secret-id`), atomically replace the file, then revoke the old SecretID accessor. The already-issued token keeps renewing; the new SecretID is only needed at the next full re-login.
 
-An empty or missing secret_id file fails the login attempt immediately (no Vault round trip). At startup the error is fatal — provider construction fails and the process exits — so a file missing at boot is recovered by restarting the process, not by an in-process retry. Once RustFS is running, the same failure is retried on the normal refresh cadence, so repairing the file mid-run heals the backend without a restart.
-
-## Kubernetes authentication
-
-On Kubernetes this is the method to prefer: the pod's own ServiceAccount is the identity, so there is no credential to distribute, rotate, or leak into a Secret.
-
-### Vault-side setup
-
-```shell
-vault auth enable kubernetes
-
-vault write auth/kubernetes/config \
-    kubernetes_host="https://$KUBERNETES_SERVICE_HOST:$KUBERNETES_SERVICE_PORT"
-
-vault write auth/kubernetes/role/rustfs \
-    bound_service_account_names=rustfs \
-    bound_service_account_namespaces=rustfs \
-    token_policies=rustfs-kms \
-    token_ttl=1h
-```
-
-As with AppRole, keep `token_ttl` comfortably above the RustFS per-attempt timeout (default 30s).
-
-### RustFS configuration
-
-```shell
-RUSTFS_KMS_BACKEND=vault-transit            # or "vault" for the KV2 backend
-RUSTFS_KMS_VAULT_ADDRESS=https://vault.vault.svc.cluster.local:8200
-RUSTFS_KMS_VAULT_KUBERNETES_ROLE=rustfs
-# Optional, defaults to "kubernetes":
-# RUSTFS_KMS_VAULT_KUBERNETES_MOUNT=kubernetes
-# Optional, defaults to the kubelet's projected token path:
-# RUSTFS_KMS_VAULT_KUBERNETES_JWT_PATH=/var/run/secrets/kubernetes.io/serviceaccount/token
-```
-
-RustFS logs in at startup and renews the token at half its TTL, falling back to a fresh login exactly as AppRole does. The ServiceAccount token is re-read from disk on every login rather than cached, so a projected token the kubelet rotates is picked up without a restart.
-
-A missing or empty token file fails the login attempt immediately (no Vault round trip). At startup the error is fatal — provider construction fails and the process exits — so a token projected late during a slow pod start is recovered by the pod restart loop, not by an in-process retry. Once RustFS is running, a token file that goes missing or turns empty is retried on the normal refresh cadence and heals the backend on its own.
+An empty or missing secret_id file fails the login attempt immediately (no Vault round trip) and is retried on the normal refresh cadence, so repairing the file heals the backend without a restart.
 
 ## Vault Agent token file
 
@@ -141,13 +101,13 @@ If the agent stops refreshing the file that is fine — RustFS re-reads the same
 
 ## Fail-closed window
 
-For lease-bound credentials (AppRole and Kubernetes tokens, token files), `current()` refuses to hand out a token that is within the safety window of its expiry and has not been refreshed. Requests then fail with `KMS credentials unavailable: ...` instead of being sent with a token that could lapse mid-flight and fail unpredictably on the Vault side.
+For lease-bound credentials (AppRole tokens, token files), `current()` refuses to hand out a token that is within the safety window of its expiry and has not been refreshed. Requests then fail with `KMS credentials unavailable: ...` instead of being sent with a token that could lapse mid-flight and fail unpredictably on the Vault side.
 
 - Default window: one per-attempt timeout (`RUSTFS_KMS_TIMEOUT_SECS`, default 30s) — a request issued now can legitimately stay in flight that long, so the token must outlive it.
-- Override: `refresh_safety_window_secs` on the `AppRole`, `Kubernetes` or `TokenFile` auth configuration.
+- Override: `refresh_safety_window_secs` on the `AppRole` or `TokenFile` auth configuration.
 - Static tokens never trip the window: they carry no lease and are assumed valid until Vault says otherwise.
 
-The window is a symptom threshold, not the fault itself: by the time it trips, refresh has been failing for roughly half the token TTL (AppRole, Kubernetes) or two poll intervals (token file).
+The window is a symptom threshold, not the fault itself: by the time it trips, refresh has been failing for roughly half the token TTL (AppRole) or two poll intervals (token file).
 
 ### Troubleshooting
 
@@ -157,8 +117,6 @@ The window is a symptom threshold, not the fault itself: by the time it trips, r
 | Renewal succeeded but re-login later fails | `Vault token renewal failed; falling back to a fresh login` followed by login errors | SecretID expired/revoked or AppRole role changed; rotate the secret_id file |
 | Token file mode error at startup or during polls | `has insecure permissions` in the error | Fix the sink `mode` (0600) and the file owner; the next poll heals the provider |
 | Token file missing/empty errors | `Failed to read Vault token file` / `token file ... is empty` | Vault Agent down or sink misconfigured; restart the agent, the next poll heals the provider |
-| Kubernetes login fails with a permission error | `Vault Kubernetes login failed` | The pod's ServiceAccount is not in the role's `bound_service_account_names`/`_namespaces`, or `auth/kubernetes/config` names the wrong API server |
-| Kubernetes ServiceAccount token errors | `Failed to read Kubernetes ServiceAccount token` / `ServiceAccount token ... is empty` | The token is not projected into the pod (check `automountServiceAccountToken` and the volume mount); the next refresh cycle heals the provider |
-| Startup fails immediately with a configuration error naming two env vars | — | Two auth methods configured at once; keep exactly one of token, AppRole, Kubernetes, token file |
+| Startup fails immediately with a configuration error naming two env vars | — | Two auth methods configured at once; keep exactly one of token, AppRole, token file |
 
 When diagnosing, confirm three clocks/lifetimes in order: the Vault token TTL (`vault token lookup` with the token's accessor), the RustFS refresh cadence (half TTL or the poll interval), and the fail-closed window. The renewal task logs every failed cycle, so a silent gap in warnings combined with `CredentialsUnavailable` errors points at the process clock or a paused runtime rather than Vault.

rustfs/src/admin/handlers/kms_backup.rs

@@ -286,7 +286,6 @@ fn auth_method_kind(auth: &VaultAuthMethod) -> String {
     match auth {
         VaultAuthMethod::Token { .. } => "token",
         VaultAuthMethod::AppRole { .. } => "approle",
-        VaultAuthMethod::Kubernetes { .. } => "kubernetes",
         VaultAuthMethod::TokenFile { .. } => "token-file",
     }
     .to_string()
@@ -485,10 +484,7 @@ fn business_trust_root_secrets(config: &KmsConfig) -> Vec<Zeroizing<String>> {
             secrets.push(Zeroizing::new(role_id.clone()));
             secrets.push(Zeroizing::new(secret_id.clone()));
         }
-        // Kubernetes and TokenFile hold no inline plaintext credential: the
-        // ServiceAccount token and the agent-managed token live in files, and
-        // the role names a Vault binding rather than half a credential pair.
-        VaultAuthMethod::Kubernetes { .. } | VaultAuthMethod::TokenFile { .. } => {}
+        VaultAuthMethod::TokenFile { .. } => {}
     };
 
     match &config.backend_config {

rustfs/src/admin/handlers/mod.rs

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

rustfs/src/admin/handlers/system.rs

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

rustfs/src/admin/handlers/usage_prefix.rs

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

rustfs/src/admin/mod.rs

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

rustfs/src/admin/route_policy.rs

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

rustfs/src/admin/route_registration_test.rs

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

rustfs/src/bitrot_selftest.rs

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

rustfs/src/init.rs

@@ -304,37 +304,30 @@ fn build_local_kms_config(cfg: &config::Config) -> std::io::Result<rustfs_kms::c
     Ok(kms_config)
 }
 
-/// Collect the Vault settings the command line owns.
-///
-/// Everything else — auth method, namespace, TLS, KV mount and metadata paths —
-/// is resolved from the environment by the KMS crate, so this path and
-/// [`rustfs_kms::config::KmsConfig::from_env`] cannot drift apart. The address
-/// stays required here so a missing one is still named instead of silently
-/// falling back to the crate's localhost default.
-fn vault_cli_overrides<'a>(
-    cfg: &'a config::Config,
-    backend_name: &str,
-) -> std::io::Result<rustfs_kms::config::VaultCliOverrides<'a>> {
-    let address = cfg
-        .kms_vault_address
-        .as_deref()
-        .ok_or_else(|| Error::other(format!("Vault address is required for {backend_name} backend")))?;
-
-    Ok(rustfs_kms::config::VaultCliOverrides {
-        address: Some(address),
-        token: cfg.kms_vault_token.as_deref(),
-        mount_path: cfg.kms_vault_mount_path.as_deref(),
-    })
-}
-
 /// Build KMS configuration for Vault backend
 fn build_vault_kms_config(cfg: &config::Config) -> std::io::Result<rustfs_kms::config::KmsConfig> {
-    let backend_config = rustfs_kms::config::vault_kv2_config_from_env(vault_cli_overrides(cfg, "vault")?)
-        .map_err(|e| Error::other(format!("Vault KMS configuration failed: {e}")))?;
+    let vault_address = cfg
+        .kms_vault_address
+        .as_ref()
+        .ok_or_else(|| Error::other("Vault address is required for vault backend"))?;
+    let vault_token = cfg
+        .kms_vault_token
+        .as_ref()
+        .ok_or_else(|| Error::other("Vault token is required for vault backend"))?;
 
     let kms_config = rustfs_kms::config::KmsConfig {
         backend: rustfs_kms::config::KmsBackend::VaultKv2,
-        backend_config: rustfs_kms::config::BackendConfig::VaultKv2(Box::new(backend_config)),
+        backend_config: rustfs_kms::config::BackendConfig::VaultKv2(Box::new(rustfs_kms::config::VaultConfig {
+            address: vault_address.clone(),
+            auth_method: rustfs_kms::config::VaultAuthMethod::Token {
+                token: vault_token.clone(),
+            },
+            namespace: None,
+            mount_path: cfg.kms_vault_mount_path.clone().unwrap_or_else(|| "transit".to_string()),
+            kv_mount: "secret".to_string(),
+            key_path_prefix: "rustfs/kms/keys".to_string(),
+            tls: None,
+        })),
         allow_insecure_dev_defaults: cfg.kms_allow_insecure_dev_defaults,
         allow_immediate_deletion: rustfs_kms::config::allow_immediate_deletion_from_env(),
         default_key_id: cfg.kms_default_key_id.clone(),
@@ -351,12 +344,26 @@ fn build_vault_kms_config(cfg: &config::Config) -> std::io::Result<rustfs_kms::c
 
 /// Build KMS configuration for Vault Transit backend
 fn build_vault_transit_kms_config(cfg: &config::Config) -> std::io::Result<rustfs_kms::config::KmsConfig> {
-    let backend_config = rustfs_kms::config::vault_transit_config_from_env(vault_cli_overrides(cfg, "vault-transit")?)
-        .map_err(|e| Error::other(format!("Vault Transit KMS configuration failed: {e}")))?;
+    let vault_address = cfg
+        .kms_vault_address
+        .as_ref()
+        .ok_or_else(|| Error::other("Vault address is required for vault-transit backend"))?;
+    let vault_token = cfg
+        .kms_vault_token
+        .as_ref()
+        .ok_or_else(|| Error::other("Vault token is required for vault-transit backend"))?;
 
     let kms_config = rustfs_kms::config::KmsConfig {
         backend: rustfs_kms::config::KmsBackend::VaultTransit,
-        backend_config: rustfs_kms::config::BackendConfig::VaultTransit(Box::new(backend_config)),
+        backend_config: rustfs_kms::config::BackendConfig::VaultTransit(Box::new(rustfs_kms::config::VaultTransitConfig {
+            address: vault_address.clone(),
+            auth_method: rustfs_kms::config::VaultAuthMethod::Token {
+                token: vault_token.clone(),
+            },
+            namespace: None,
+            mount_path: cfg.kms_vault_mount_path.clone().unwrap_or_else(|| "transit".to_string()),
+            ..rustfs_kms::config::VaultTransitConfig::default()
+        })),
         allow_insecure_dev_defaults: cfg.kms_allow_insecure_dev_defaults,
         allow_immediate_deletion: rustfs_kms::config::allow_immediate_deletion_from_env(),
         default_key_id: cfg.kms_default_key_id.clone(),
@@ -1398,10 +1405,7 @@ pub async fn init_sftp_system() -> Result<Option<ShutdownHandle>, Box<dyn std::e
 
 #[cfg(test)]
 mod tests {
-    use super::{
-        build_aws_kms_config, build_vault_kms_config, build_vault_transit_kms_config, notification_config_to_event_rules,
-        resolve_buffer_profile_config,
-    };
+    use super::{build_aws_kms_config, notification_config_to_event_rules, resolve_buffer_profile_config};
     use crate::config::{BufferConfig, WorkloadProfile};
     use rustfs_config::KI_B;
     use rustfs_s3_types::EventName;
@@ -1495,151 +1499,6 @@ mod tests {
         assert!(err.to_string().contains("Invalid ARN"), "unexpected error: {err}");
     }
 
-    fn vault_kms_test_config(backend: &str) -> crate::config::Config {
-        let mut config = crate::config::Config::new("127.0.0.1:9000", vec!["/tmp/rustfs-vault-kms".to_string()]);
-        config.kms_enable = true;
-        config.kms_backend = backend.to_string();
-        config.kms_vault_address = Some("https://vault.example.com:8200".to_string());
-        config
-    }
-
-    /// The Vault auth method and the settings the CLI has no flag for come from
-    /// the environment, so startup and `KmsConfig::from_env` cannot disagree.
-    /// Regression: startup used to hardcode token auth and require a token,
-    /// which made every non-token method unreachable through `RUSTFS_KMS_ENABLE`.
-    #[test]
-    fn build_vault_transit_kms_config_resolves_auth_and_mounts_from_env() {
-        let config = temp_env::with_vars(
-            [
-                ("RUSTFS_KMS_VAULT_TOKEN", None),
-                ("RUSTFS_KMS_VAULT_TOKEN_FILE", None),
-                ("RUSTFS_KMS_VAULT_KUBERNETES_ROLE", None),
-                ("RUSTFS_KMS_VAULT_APPROLE_ROLE_ID", Some("env-role-id")),
-                ("RUSTFS_KMS_VAULT_APPROLE_SECRET_ID", Some("env-secret-id")),
-                ("RUSTFS_KMS_VAULT_APPROLE_SECRET_ID_FILE", None),
-                ("RUSTFS_KMS_VAULT_NAMESPACE", Some("team-a")),
-                ("RUSTFS_KMS_VAULT_TRANSIT_METADATA_KV_MOUNT", Some("rustfs-kv")),
-            ],
-            || {
-                build_vault_transit_kms_config(&vault_kms_test_config("vault-transit"))
-                    .expect("vault transit KMS configuration should build")
-            },
-        );
-
-        let vault = config.vault_transit_config().expect("vault transit backend config");
-        let rustfs_kms::config::VaultAuthMethod::AppRole { role_id, secret_id, .. } = &vault.auth_method else {
-            panic!("approle in the environment must select AppRole auth, got {:?}", vault.auth_method);
-        };
-        assert_eq!(role_id, "env-role-id");
-        assert_eq!(secret_id, "env-secret-id");
-        assert_eq!(vault.namespace.as_deref(), Some("team-a"));
-        assert_eq!(vault.metadata_kv_mount, "rustfs-kv");
-    }
-
-    /// Kubernetes auth needs no credential in the environment at all: the role
-    /// selects it and the pod's projected ServiceAccount token supplies the rest.
-    #[test]
-    fn build_vault_transit_kms_config_selects_kubernetes_auth() {
-        let config = temp_env::with_vars(
-            [
-                ("RUSTFS_KMS_VAULT_TOKEN", None),
-                ("RUSTFS_KMS_VAULT_TOKEN_FILE", None),
-                ("RUSTFS_KMS_VAULT_APPROLE_ROLE_ID", None),
-                ("RUSTFS_KMS_VAULT_KUBERNETES_ROLE", Some("rustfs")),
-                ("RUSTFS_KMS_VAULT_KUBERNETES_MOUNT", None),
-                ("RUSTFS_KMS_VAULT_KUBERNETES_JWT_PATH", None),
-            ],
-            || {
-                build_vault_transit_kms_config(&vault_kms_test_config("vault-transit"))
-                    .expect("vault transit KMS configuration should build")
-            },
-        );
-
-        let vault = config.vault_transit_config().expect("vault transit backend config");
-        let rustfs_kms::config::VaultAuthMethod::Kubernetes {
-            role, mount, jwt_path, ..
-        } = &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, rustfs_kms::config::DEFAULT_VAULT_KUBERNETES_MOUNT);
-        assert_eq!(jwt_path, std::path::Path::new(rustfs_kms::config::DEFAULT_VAULT_KUBERNETES_JWT_PATH));
-    }
-
-    /// Two credential sources leave the effective identity ambiguous, so
-    /// startup refuses rather than picking one.
-    #[test]
-    fn build_vault_kms_config_refuses_two_auth_methods() {
-        temp_env::with_vars(
-            [
-                ("RUSTFS_KMS_VAULT_TOKEN", None),
-                ("RUSTFS_KMS_VAULT_TOKEN_FILE", Some("/run/vault-agent/token")),
-                ("RUSTFS_KMS_VAULT_APPROLE_ROLE_ID", None),
-                ("RUSTFS_KMS_VAULT_KUBERNETES_ROLE", Some("rustfs")),
-            ],
-            || {
-                let error = build_vault_kms_config(&vault_kms_test_config("vault"))
-                    .expect_err("two Vault auth methods must not start the server");
-                assert!(error.to_string().contains("exactly one"), "unexpected error: {error}");
-            },
-        );
-    }
-
-    /// The KV2 backend has its own builder, so the key-location settings have
-    /// to be proven separately from the Transit one: pointing at the wrong KV
-    /// mount or prefix makes existing keys look absent.
-    #[test]
-    fn build_vault_kms_config_resolves_kv_mount_and_prefix_from_env() {
-        let config = temp_env::with_vars(
-            [
-                ("RUSTFS_KMS_VAULT_TOKEN", Some("a-real-token")),
-                ("RUSTFS_KMS_VAULT_TOKEN_FILE", None),
-                ("RUSTFS_KMS_VAULT_APPROLE_ROLE_ID", None),
-                ("RUSTFS_KMS_VAULT_KUBERNETES_ROLE", None),
-                ("RUSTFS_KMS_VAULT_KV_MOUNT", Some("rustfs-kv")),
-                ("RUSTFS_KMS_VAULT_KEY_PREFIX", Some("tenant/keys")),
-            ],
-            || build_vault_kms_config(&vault_kms_test_config("vault")).expect("vault KV2 KMS configuration should build"),
-        );
-
-        let vault = config.vault_config().expect("vault kv2 backend config");
-        assert_eq!(vault.kv_mount, "rustfs-kv");
-        assert_eq!(vault.key_path_prefix, "tenant/keys");
-    }
-
-    /// Skipping TLS verification was silently dropped on this path before, so
-    /// an operator who asked for it still got a verified connection. Now that it
-    /// is honoured it must fail closed without the development opt-in, rather
-    /// than quietly downgrading the Vault connection.
-    #[test]
-    fn build_vault_transit_kms_config_refuses_skip_tls_verify_without_opt_in() {
-        let vars = [
-            ("RUSTFS_KMS_VAULT_TOKEN", Some("a-real-token")),
-            ("RUSTFS_KMS_VAULT_TOKEN_FILE", None),
-            ("RUSTFS_KMS_VAULT_APPROLE_ROLE_ID", None),
-            ("RUSTFS_KMS_VAULT_KUBERNETES_ROLE", None),
-            ("RUSTFS_KMS_VAULT_SKIP_TLS_VERIFY", Some("true")),
-        ];
-
-        temp_env::with_vars(vars, || {
-            let error = build_vault_transit_kms_config(&vault_kms_test_config("vault-transit"))
-                .expect_err("skipping TLS verification must not start the server");
-            assert!(error.to_string().contains("TLS"), "unexpected error: {error}");
-        });
-
-        temp_env::with_vars(vars, || {
-            let mut cfg = vault_kms_test_config("vault-transit");
-            cfg.kms_allow_insecure_dev_defaults = true;
-            let config = build_vault_transit_kms_config(&cfg).expect("the development opt-in should accept skip-verify");
-            let vault = config.vault_transit_config().expect("vault transit backend config");
-            assert!(vault.tls.as_ref().is_some_and(|tls| tls.skip_verify));
-        });
-    }
-
     fn aws_kms_test_config() -> crate::config::Config {
         let mut config = crate::config::Config::new("127.0.0.1:9000", vec!["/tmp/rustfs-aws-kms".to_string()]);
         config.kms_enable = true;

rustfs/src/lib.rs

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

rustfs/src/module_switches.rs

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

rustfs/src/startup_background.rs

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

rustfs/src/storage/storage_api.rs

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

rustfs/src/storage_api.rs

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