feat(helm): add TLSRoute passthrough support for gateway api (#6169)

Add an optional TLS passthrough listener to the Gateway API support. When gatewayApi.listeners.tls.enabled is true, the Gateway gets a TLS listener with tls.mode: Passthrough and a TLSRoute is rendered to the RustFS service so TLS terminates at the backend (end-to-end encryption).

Refs rustfs/rustfs#3862.

crates/data-usage/src/data_usage.rs

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

crates/ecstore/src/store/mod.rs

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

crates/scanner/src/data_usage_define.rs

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

crates/scanner/src/lib.rs

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

crates/scanner/src/prefix_usage.rs

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

crates/scanner/src/scanner_io.rs

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

helm/README.md

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

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

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

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

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

helm/rustfs/values.yaml

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

rustfs/src/admin/handlers/mod.rs

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

rustfs/src/admin/handlers/site_replication.rs

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

rustfs/src/admin/handlers/system.rs

@@ -1158,10 +1158,10 @@ impl Operation for RuntimeCapabilitiesHandler {
     }
 }
 
-/// Authorization gate for GET datausageinfo: any-of the dedicated admin action
-/// OR the bucket listing action. Pinned by a unit test so the gate cannot
-/// silently narrow or widen (rustfs/backlog#1306).
-fn data_usage_info_gate_actions() -> Vec<Action> {
+/// Authorization gate for GET datausageinfo (and prefix usage): any-of the
+/// dedicated admin action OR the bucket listing action. Pinned by a unit test
+/// so the gate cannot silently narrow or widen (rustfs/backlog#1306).
+pub(crate) fn data_usage_info_gate_actions() -> Vec<Action> {
     vec![
         Action::AdminAction(AdminAction::DataUsageInfoAdminAction),
         Action::S3Action(S3Action::ListBucketAction),

rustfs/src/admin/handlers/usage_prefix.rs

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

rustfs/src/admin/mod.rs

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

rustfs/src/admin/route_policy.rs

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

rustfs/src/admin/route_registration_test.rs

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