fix(quota): keep reads available without snapshot-wide copies (#8928)

* fix(quota): keep quota reads available during scans

Use optional scanner usage for read-only quota statistics across all
quota configurations. Preserve strict legacy write admission and durable
reservation enforcement while keeping unknown usage distinct from zero.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* perf(usage): avoid snapshot copies for single-bucket reads

Project owned snapshots, scalar bucket sizes and refresh-only results
through the same TTL-bounded cache loader. Move loaded snapshots into
the cache without changing refresh serialization or generation fencing.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

---------

Co-authored-by: heihutu <heihutu@gmail.com>
Co-authored-by: zhi22915 <qiuzgang@gmail.com>

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

@@ -235,6 +235,8 @@ impl QuotaChecker {
         Ok(updated_at)
     }
 
+    /// Return quota configuration with optional scanner usage, independently
+    /// of the usage requirements enforced by write admission.
     pub async fn get_quota_stats(&self, bucket: &str) -> Result<(BucketQuota, Option<u64>), QuotaError> {
         // If bucket doesn't exist, return ConfigNotFound error
         if !self.bucket_exists(bucket).await {
@@ -244,18 +246,13 @@ impl QuotaChecker {
         }
 
         let quota = self.get_quota_config(bucket).await?;
-        let current_usage = if quota.uses_durable_reservations() {
-            self.report_only_usage(bucket).await
-        } else {
-            Some(self.get_real_time_usage(bucket).await?)
-        };
+        let current_usage = self.report_only_usage(bucket).await;
 
         Ok((quota, current_usage))
     }
 
-    /// Scanner usage for a bucket the reservation ledger admits is reporting
-    /// input only, so unavailable memory usage is reported as "unknown" and counted
-    /// instead of becoming an error the write path would fail on.
+    /// Best-effort usage for statistics and durable admission reports.
+    /// Missing scanner coverage is unknown, not an error or a zero-sized bucket.
     async fn report_only_usage(&self, bucket: &str) -> Option<u64> {
         // The plain lookup: a report must not wait for an early snapshot reload.
         let usage = get_bucket_usage_memory(bucket).await;
@@ -267,7 +264,7 @@ impl QuotaChecker {
                 subsystem = LOG_SUBSYSTEM_QUOTA,
                 state = "usage_unknown",
                 bucket = %bucket,
-                "durable quota usage report degraded"
+                "quota usage report degraded"
             );
         }
         usage
@@ -742,8 +739,12 @@ mod tests {
             1,
             "fail-closed admission takes the early-reload lookup"
         );
-        assert!(matches!(checker.get_quota_stats(&legacy).await, Err(QuotaError::UsageUnavailable { .. })));
-        assert_eq!(crate::data_usage::reload_unknown_bucket_usage_calls_for_test(&legacy), 2);
+        let (_, usage) = checker
+            .get_quota_stats(&legacy)
+            .await
+            .expect("legacy quota statistics are reporting only");
+        assert_eq!(usage, None);
+        assert_eq!(crate::data_usage::reload_unknown_bucket_usage_calls_for_test(&legacy), 1);
         assert_eq!(
             crate::data_usage::reload_unknown_bucket_usage_calls_for_test(&bucket),
             0,

crates/ecstore/src/bucket/quota/checker/tests/reporting_retirement.rs

@@ -245,3 +245,86 @@ async fn quota_without_limit_keeps_the_zero_usage_io_fast_path() {
     assert_eq!(crate::data_usage::reload_unknown_bucket_usage_calls_for_test(&bucket), 0);
     stop_reporting_fixture(&store).await;
 }
+
+#[tokio::test]
+#[serial]
+async fn quota_statistics_remain_available_without_a_complete_scanner_snapshot() {
+    use crate::storage_api_contracts::bucket::{BucketOperations as _, MakeBucketOptions};
+
+    for configuration in ["unconfigured", "unlimited", "snapshot"] {
+        for source in ["missing", "legacy", "incomplete", "observed", "complete", "empty"] {
+            let (_dirs, store, _) = isolated_durable_quota_store().await;
+            let bucket = format!("quota-statistics-{}", Uuid::new_v4().simple());
+            match configuration {
+                "unconfigured" => store
+                    .make_bucket(&bucket, &MakeBucketOptions::default())
+                    .await
+                    .expect("create bucket without a quota configuration"),
+                "unlimited" => {
+                    create_bucket_with_quota_json(
+                        &store,
+                        &bucket,
+                        serde_json::to_vec(&BucketQuota::new(None)).expect("unlimited quota encoding"),
+                    )
+                    .await;
+                }
+                "snapshot" => create_bucket_with_quota_json(&store, &bucket, legacy_quota_json(4096)).await,
+                _ => unreachable!(),
+            }
+            // An unregistered checker keeps legacy migration out of the
+            // assertion: reporting must work before a protocol upgrade.
+            let metadata_sys = BucketMetadataSys::new(store.clone());
+            let (metadata, _) = metadata_sys.get_config(&bucket).await.expect("load existing bucket metadata");
+            metadata_sys.set(bucket.clone(), metadata).await;
+            let checker = QuotaChecker::new(Arc::new(RwLock::new(metadata_sys)));
+            let (snapshot_source, size) = if source == "empty" { ("complete", 0) } else { (source, 1000) };
+            install_reporting_snapshot(store.clone(), &bucket, snapshot_source, size).await;
+            crate::data_usage::invalidate_data_usage_snapshot_cache().await;
+
+            let statistics = checker.get_quota_stats(&bucket).await;
+            let early_reloads = crate::data_usage::reload_unknown_bucket_usage_calls_for_test(&bucket);
+            let missing_bucket = format!("missing-{}", Uuid::new_v4().simple());
+            let missing_statistics = checker.get_quota_stats(&missing_bucket).await;
+            crate::data_usage::prepare_bucket_usage_for_namespace_change(&bucket, None)
+                .await
+                .expect("fixture cache cleanup");
+            crate::data_usage::invalidate_data_usage_snapshot_cache().await;
+            stop_reporting_fixture(&store).await;
+
+            let (quota, usage) = statistics.unwrap_or_else(|error| {
+                panic!("configuration={configuration}, source={source}: quota statistics must remain available: {error}")
+            });
+            assert_eq!(quota.quota, (configuration == "snapshot").then_some(4096));
+            assert_eq!(
+                usage,
+                (snapshot_source == "complete").then_some(size),
+                "configuration={configuration}, source={source}: only known usage is numeric"
+            );
+            assert_eq!(early_reloads, 0, "read-only statistics must not force an early snapshot reload");
+            assert!(
+                matches!(missing_statistics, Err(QuotaError::ConfigNotFound { bucket }) if bucket == missing_bucket),
+                "reporting must still reject a missing bucket"
+            );
+        }
+    }
+}
+
+#[tokio::test]
+async fn quota_statistics_preserve_configuration_errors() {
+    let (_dirs, store, _) = isolated_durable_quota_store().await;
+    let bucket = format!("quota-invalid-{}", Uuid::new_v4().simple());
+    create_bucket_with_quota_json(&store, &bucket, legacy_quota_json(4096)).await;
+    let metadata_sys = BucketMetadataSys::new(store.clone());
+    let (mut metadata, _) = metadata_sys.get_config(&bucket).await.expect("load existing bucket metadata");
+    Arc::make_mut(&mut metadata).quota_config_json = b"{not-json".to_vec();
+    metadata_sys.set(bucket.clone(), metadata).await;
+    let checker = QuotaChecker::new(Arc::new(RwLock::new(metadata_sys)));
+
+    let statistics = checker.get_quota_stats(&bucket).await;
+    stop_reporting_fixture(&store).await;
+
+    assert!(
+        matches!(statistics, Err(QuotaError::InvalidConfig { .. })),
+        "unavailable reporting must not hide malformed quota configuration"
+    );
+}

crates/ecstore/src/data_usage/mod.rs

@@ -216,7 +216,7 @@ pub async fn lookup_degraded_bucket_usage_baseline(store: Arc<ECStore>, bucket:
     // Stale or empty cache: refresh through the TTL-bounded loader. A failed
     // refresh carries the previous baseline forward, so quota admission keeps
     // its last grounded values through a backend read outage.
-    let _ = load_data_usage_from_backend_cached(store).await;
+    let _ = project_data_usage_from_backend_cached(store, |_| ()).await;
     let cache = data_usage_snapshot_cache().read().await;
     cache
         .as_ref()
@@ -237,31 +237,34 @@ struct CachedDataUsageSnapshot {
 }
 
 impl CachedDataUsageSnapshot {
-    fn result(&self) -> Result<DataUsageInfo, Error> {
+    fn result<T>(&self, project: impl FnOnce(&DataUsageInfo) -> T) -> Result<T, Error> {
         self.info
-            .clone()
+            .as_ref()
+            .map(project)
             .ok_or_else(|| Error::other("data usage snapshot load recently failed"))
     }
 }
 
-fn fresh_cached_data_usage_snapshot(
+fn fresh_cached_data_usage_snapshot<T>(
     cache: &Option<CachedDataUsageSnapshot>,
     now: tokio::time::Instant,
     ttl: Duration,
-) -> Option<Result<DataUsageInfo, Error>> {
+    project: impl FnOnce(&DataUsageInfo) -> T,
+) -> Option<Result<T, Error>> {
     cache
         .as_ref()
         .filter(|cached| now.duration_since(cached.loaded_at) < ttl)
-        .map(CachedDataUsageSnapshot::result)
+        .map(|cached| cached.result(project))
 }
 
-fn cache_data_usage_snapshot_result(
+fn cache_data_usage_snapshot_result<T>(
     cache: &mut Option<CachedDataUsageSnapshot>,
     result: Result<LoadedUsageBaseline, Error>,
     loaded_at: tokio::time::Instant,
     refresh_generation: u64,
     current_generation: u64,
-) -> Option<Result<DataUsageInfo, Error>> {
+    project: impl FnOnce(&DataUsageInfo) -> T,
+) -> Option<Result<T, Error>> {
     if current_generation != refresh_generation {
         return None;
     }
@@ -280,12 +283,13 @@ fn cache_data_usage_snapshot_result(
                     degraded_baseline.entry(bucket.clone()).or_insert(*size);
                 }
             }
+            let projected = project(&info);
             *cache = Some(CachedDataUsageSnapshot {
-                info: Some(info.clone()),
+                info: Some(info),
                 loaded_at,
                 degraded_baseline,
             });
-            Ok(info)
+            Ok(projected)
         }
         Err(e) => {
             // Keep the previous baseline through a failed refresh: quota
@@ -1486,12 +1490,21 @@ async fn normalize_loaded_data_usage(
 /// `save_data_usage_in_backend` invalidates the cache so a fresh scanner save
 /// is visible immediately in the saving process.
 pub async fn load_data_usage_from_backend_cached(store: Arc<ECStore>) -> Result<DataUsageInfo, Error> {
+    project_data_usage_from_backend_cached(store, DataUsageInfo::clone).await
+}
+
+// Project under the cache guard so scalar readers never copy the snapshot's
+// bucket maps. The owned API shares the same TTL, refresh and invalidation path.
+async fn project_data_usage_from_backend_cached<T>(
+    store: Arc<ECStore>,
+    project: impl Fn(&DataUsageInfo) -> T,
+) -> Result<T, Error> {
     let ttl = Duration::from_secs(DATA_USAGE_CACHE_TTL_SECS);
 
     loop {
         {
             let cache = data_usage_snapshot_cache().read().await;
-            if let Some(result) = fresh_cached_data_usage_snapshot(&cache, tokio::time::Instant::now(), ttl) {
+            if let Some(result) = fresh_cached_data_usage_snapshot(&cache, tokio::time::Instant::now(), ttl, &project) {
                 return result;
             }
         }
@@ -1503,7 +1516,7 @@ pub async fn load_data_usage_from_backend_cached(store: Arc<ECStore>) -> Result<
             .await;
         {
             let cache = data_usage_snapshot_cache().read().await;
-            if let Some(result) = fresh_cached_data_usage_snapshot(&cache, tokio::time::Instant::now(), ttl) {
+            if let Some(result) = fresh_cached_data_usage_snapshot(&cache, tokio::time::Instant::now(), ttl, &project) {
                 return result;
             }
         }
@@ -1512,9 +1525,14 @@ pub async fn load_data_usage_from_backend_cached(store: Arc<ECStore>) -> Result<
         let result = load_data_usage_from_backend_with_baseline(store.clone()).await;
         let loaded_at = tokio::time::Instant::now();
         let mut cache = data_usage_snapshot_cache().write().await;
-        if let Some(result) =
-            cache_data_usage_snapshot_result(&mut cache, result, loaded_at, refresh_generation, data_usage_snapshot_generation())
-        {
+        if let Some(result) = cache_data_usage_snapshot_result(
+            &mut cache,
+            result,
+            loaded_at,
+            refresh_generation,
+            data_usage_snapshot_generation(),
+            &project,
+        ) {
             return result;
         }
         drop(cache);
@@ -1532,7 +1550,8 @@ pub async fn load_admin_data_usage_from_backend_cached(store: Arc<ECStore>) -> R
     loop {
         {
             let cache = admin_data_usage_snapshot_cache().read().await;
-            if let Some(result) = fresh_cached_data_usage_snapshot(&cache, tokio::time::Instant::now(), ttl) {
+            if let Some(result) = fresh_cached_data_usage_snapshot(&cache, tokio::time::Instant::now(), ttl, DataUsageInfo::clone)
+            {
                 return result;
             }
         }
@@ -1543,7 +1562,8 @@ pub async fn load_admin_data_usage_from_backend_cached(store: Arc<ECStore>) -> R
             .await;
         {
             let cache = admin_data_usage_snapshot_cache().read().await;
-            if let Some(result) = fresh_cached_data_usage_snapshot(&cache, tokio::time::Instant::now(), ttl) {
+            if let Some(result) = fresh_cached_data_usage_snapshot(&cache, tokio::time::Instant::now(), ttl, DataUsageInfo::clone)
+            {
                 return result;
             }
         }
@@ -1564,6 +1584,7 @@ pub async fn load_admin_data_usage_from_backend_cached(store: Arc<ECStore>) -> R
             loaded_at,
             refresh_generation,
             admin_data_usage_snapshot_generation(),
+            DataUsageInfo::clone,
         ) {
             return result;
         }
@@ -2277,8 +2298,10 @@ async fn get_persisted_bucket_usage(bucket: &str) -> Option<u64> {
 }
 
 async fn persisted_bucket_usage(store: Arc<ECStore>, bucket: &str) -> Option<u64> {
-    let info = load_data_usage_from_backend_cached(store).await.ok()?;
-    info.buckets_usage.get(bucket).map(|usage| usage.size)
+    project_data_usage_from_backend_cached(store, |info| info.buckets_usage.get(bucket).map(|usage| usage.size))
+        .await
+        .ok()
+        .flatten()
 }
 
 async fn reload_persisted_bucket_usage(store: Arc<ECStore>, bucket: &str) -> Option<u64> {
@@ -2375,6 +2398,7 @@ where
         tokio::time::Instant::now(),
         refresh_generation,
         data_usage_snapshot_generation(),
+        |_| (),
     )
     .is_some()
 }
@@ -2891,6 +2915,8 @@ pub async fn init_compression_total_memory_from_backend(store: Arc<ECStore>) {
 
 #[cfg(test)]
 mod tests {
+    mod snapshot_projection;
+
     use super::*;
     use crate::layout::endpoints::EndpointServerPools;
     use crate::runtime::instance::InstanceContext;
@@ -4572,15 +4598,29 @@ mod tests {
             loaded_at,
             refresh_generation,
             data_usage_snapshot_generation(),
+            DataUsageInfo::clone,
         )
         .expect("an uninterrupted refresh should populate the cache");
         assert!(matches!(first, Err(Error::ErasureReadQuorum)));
 
-        let cached = fresh_cached_data_usage_snapshot(&cache, loaded_at + Duration::from_secs(1), Duration::from_secs(30))
-            .expect("the failed load must remain cached within the TTL");
+        let cached = fresh_cached_data_usage_snapshot(
+            &cache,
+            loaded_at + Duration::from_secs(1),
+            Duration::from_secs(30),
+            DataUsageInfo::clone,
+        )
+        .expect("the failed load must remain cached within the TTL");
         assert!(cached.is_err());
 
-        assert!(fresh_cached_data_usage_snapshot(&cache, loaded_at + Duration::from_secs(30), Duration::from_secs(30)).is_none());
+        assert!(
+            fresh_cached_data_usage_snapshot(
+                &cache,
+                loaded_at + Duration::from_secs(30),
+                Duration::from_secs(30),
+                DataUsageInfo::clone
+            )
+            .is_none()
+        );
     }
 
     #[test]
@@ -4601,14 +4641,20 @@ mod tests {
             loaded_at,
             refresh_generation,
             data_usage_snapshot_generation(),
+            DataUsageInfo::clone,
         )
         .expect("an uninterrupted refresh should populate the cache")
         .expect("successful load must be returned");
         assert_snapshot_bucket(&first, "bucket");
 
-        let cached = fresh_cached_data_usage_snapshot(&cache, loaded_at + Duration::from_secs(1), Duration::from_secs(30))
-            .expect("successful load must remain cached within the TTL")
-            .expect("cached success must remain successful");
+        let cached = fresh_cached_data_usage_snapshot(
+            &cache,
+            loaded_at + Duration::from_secs(1),
+            Duration::from_secs(30),
+            DataUsageInfo::clone,
+        )
+        .expect("successful load must remain cached within the TTL")
+        .expect("cached success must remain successful");
         assert_snapshot_bucket(&cached, "bucket");
     }
 
@@ -4633,6 +4679,7 @@ mod tests {
             loaded_at,
             refresh_generation,
             data_usage_snapshot_generation(),
+            DataUsageInfo::clone,
         )
         .expect("an uninterrupted refresh should populate the cache")
         .expect("successful load must be returned");
@@ -4666,6 +4713,7 @@ mod tests {
             loaded_at,
             refresh_generation,
             data_usage_snapshot_generation(),
+            DataUsageInfo::clone,
         );
 
         assert!(stale_result.is_none());

crates/ecstore/src/data_usage/tests/snapshot_projection.rs

@@ -0,0 +1,191 @@
+// 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.
+
+use super::*;
+
+#[test]
+fn scalar_projection_borrows_snapshot_and_preserves_unknown_and_zero() {
+    let loaded_at = tokio::time::Instant::now();
+    let mut info = data_usage_info_for_test("known", 1, 42, SystemTime::UNIX_EPOCH);
+    info.buckets_usage.insert("empty".to_string(), BucketUsageInfo::default());
+    let cache = Some(CachedDataUsageSnapshot {
+        info: Some(info),
+        loaded_at,
+        degraded_baseline: HashMap::new(),
+    });
+    let stored = cache
+        .as_ref()
+        .expect("cached snapshot")
+        .info
+        .as_ref()
+        .expect("known snapshot");
+    let ttl = Duration::from_secs(30);
+
+    for (bucket, expected) in [("known", Some(42)), ("empty", Some(0)), ("missing", None)] {
+        let value = fresh_cached_data_usage_snapshot(&cache, loaded_at, ttl, |info| {
+            assert!(std::ptr::eq(info, stored), "projection must borrow the cached snapshot");
+            info.buckets_usage.get(bucket).map(|usage| usage.size)
+        })
+        .expect("fresh snapshot")
+        .expect("successful snapshot");
+        let owned = fresh_cached_data_usage_snapshot(&cache, loaded_at, ttl, DataUsageInfo::clone)
+            .expect("fresh snapshot")
+            .expect("successful snapshot");
+        assert_eq!(value, expected);
+        assert_eq!(value, owned.buckets_usage.get(bucket).map(|usage| usage.size));
+    }
+}
+
+#[test]
+fn projection_is_skipped_for_missing_failed_expired_or_invalidated_snapshots() {
+    let loaded_at = tokio::time::Instant::now();
+    let ttl = Duration::from_secs(30);
+    let never_project = |_: &DataUsageInfo| -> u64 { panic!("unusable snapshots must not be projected") };
+    assert!(fresh_cached_data_usage_snapshot(&None, loaded_at, ttl, never_project).is_none());
+
+    let mut cache = None;
+    let failure = cache_data_usage_snapshot_result(&mut cache, Err(Error::ErasureReadQuorum), loaded_at, 1, 1, never_project)
+        .expect("matching generation");
+    assert!(matches!(failure, Err(Error::ErasureReadQuorum)));
+    assert!(
+        fresh_cached_data_usage_snapshot(&cache, loaded_at, ttl, never_project)
+            .expect("negative cache remains fresh")
+            .is_err()
+    );
+
+    cache = Some(CachedDataUsageSnapshot {
+        info: Some(data_usage_info_for_test("known", 1, 42, SystemTime::UNIX_EPOCH)),
+        loaded_at,
+        degraded_baseline: HashMap::new(),
+    });
+    assert!(fresh_cached_data_usage_snapshot(&cache, loaded_at + ttl, ttl, never_project).is_none());
+    cache = None;
+    let rejected = cache_data_usage_snapshot_result(
+        &mut cache,
+        Ok(LoadedUsageBaseline {
+            info: data_usage_info_for_test("known", 1, 42, SystemTime::UNIX_EPOCH),
+            degraded_baseline: HashMap::new(),
+            observed_unavailable: false,
+        }),
+        loaded_at,
+        1,
+        2,
+        never_project,
+    );
+    assert!(rejected.is_none());
+    assert!(cache.is_none(), "rejected generation must not republish stale data");
+}
+
+#[test]
+fn projected_publication_moves_bucket_maps_into_cache() {
+    let info = data_usage_info_for_test("known", 1, 42, SystemTime::UNIX_EPOCH);
+    let original_key = info.buckets_usage.get_key_value("known").expect("source bucket").0.as_ptr();
+    let mut cache = None;
+    let usage = cache_data_usage_snapshot_result(
+        &mut cache,
+        Ok(LoadedUsageBaseline {
+            info,
+            degraded_baseline: HashMap::new(),
+            observed_unavailable: false,
+        }),
+        tokio::time::Instant::now(),
+        1,
+        1,
+        |info| {
+            assert_eq!(
+                info.buckets_usage
+                    .get_key_value("known")
+                    .expect("projected bucket")
+                    .0
+                    .as_ptr(),
+                original_key
+            );
+            info.buckets_usage.get("known").map(|usage| usage.size)
+        },
+    )
+    .expect("matching generation")
+    .expect("successful publication");
+    assert_eq!(usage, Some(42));
+    let cached = cache
+        .as_ref()
+        .expect("published cache")
+        .info
+        .as_ref()
+        .expect("published snapshot");
+    assert_eq!(
+        cached.buckets_usage.get_key_value("known").expect("cached bucket").0.as_ptr(),
+        original_key
+    );
+}
+
+#[tokio::test]
+#[serial]
+async fn persisted_scalar_projection_shares_cache_refresh_and_invalidation() {
+    use crate::bucket::metadata_sys::test_support::isolated_store_over_temp_disks;
+    use crate::config::com::save_config;
+
+    let (_dirs, store) = isolated_store_over_temp_disks().await;
+    let mut info = data_usage_info_for_test("known", 1, 42, SystemTime::UNIX_EPOCH);
+    info.buckets_usage.insert("empty".to_string(), BucketUsageInfo::default());
+    info.bucket_sizes.insert("empty".to_string(), 0);
+    info.buckets_count = 2;
+    save_config(
+        store.clone(),
+        DATA_USAGE_OBJ_NAME_PATH.as_str(),
+        serde_json::to_vec(&info).expect("encode snapshot"),
+    )
+    .await
+    .expect("persist snapshot");
+    invalidate_data_usage_snapshot_cache().await;
+
+    for (bucket, expected) in [("known", Some(42)), ("empty", Some(0)), ("missing", None)] {
+        assert_eq!(persisted_bucket_usage(store.clone(), bucket).await, expected);
+        let owned = load_data_usage_from_backend_cached(store.clone())
+            .await
+            .expect("owned cached snapshot");
+        assert_eq!(expected, owned.buckets_usage.get(bucket).map(|usage| usage.size));
+    }
+    info.buckets_usage.get_mut("known").expect("known bucket").size = 84;
+    info.bucket_sizes.insert("known".to_string(), 84);
+    save_config(
+        store.clone(),
+        DATA_USAGE_OBJ_NAME_PATH.as_str(),
+        serde_json::to_vec(&info).expect("encode new snapshot"),
+    )
+    .await
+    .expect("persist new snapshot without invalidating cache");
+    assert_eq!(
+        persisted_bucket_usage(store.clone(), "known").await,
+        Some(42),
+        "warm reads reuse the TTL cache"
+    );
+    invalidate_data_usage_snapshot_cache().await;
+    assert_eq!(
+        persisted_bucket_usage(store.clone(), "known").await,
+        Some(84),
+        "invalidation exposes the new snapshot"
+    );
+    invalidate_data_usage_snapshot_cache().await;
+    if let Some(cancel) = store.ctx.background_cancel_token() {
+        cancel.cancel();
+    }
+    for pool in &store.pools {
+        for set in &pool.disk_set {
+            let disks = set.disks.read().await.clone();
+            for disk in disks.iter().flatten() {
+                disk.close().await.expect("stop fixture disk monitoring");
+            }
+        }
+    }
+}

docs/architecture/scanner-usage-authority-decision.md

@@ -112,12 +112,13 @@ covers the separate `reload_unknown_bucket_usage` trigger only.
 This consumer removal needs no format change, writer activation or quota
 migration. Among D3-capable nodes, an older node can report the old floor while
 a newer node reports unknown; both keep ledger admission and return a successful
-best-effort report. Snapshot-protocol buckets (`reservation_protocol = None`)
-still take `get_real_time_usage`, including early refresh, the legacy and
-observed degraded baseline, and fail-closed admission when no grounded value
-exists. Earlier nodes that still make scanner availability a durable-write
-veto must first reach the D3 baseline; this change does not prove that earlier
-fleet safe.
+best-effort report. Read-only `get_quota_stats` uses optional reporting for
+all buckets, including snapshot-protocol and quota-free buckets. Snapshot-protocol
+admission (`reservation_protocol = None`) still takes `get_real_time_usage`,
+including early refresh, the legacy and observed degraded baseline, and
+fails closed when no grounded value exists. Earlier nodes that still make
+scanner availability a durable-write veto must first reach the D3 baseline;
+this change does not prove that earlier fleet safe.
 
 Rollback to a D3 reader before this consumer removal restores floor-based
 optional reporting. No scanner snapshot, backup, quota record or reservation

docs/operations/durable-quota-reservation-v2.md

@@ -177,8 +177,35 @@ For a bucket on protocol v1 or v2 the request-path quota check and the admin
 quota endpoints read scanner usage for reporting only. When no usage is
 available, `GET /rustfs/admin/v3/quota-stats/{bucket}` and
 `POST /rustfs/admin/v3/quota-check/{bucket}` answer `200` with
-`current_usage: null` (and no `remaining_quota` or `usage_percentage`) instead
-of `503`, and object writes are admitted by the ledger alone. Each such
+`current_usage: null` and `remaining_quota: null` instead of `503`;
+quota statistics also return `usage_percentage: null`. Object writes are
+admitted by the ledger alone. Each such
 lookup logs a `quota_usage_report` event at debug level and increments
 `rustfs_quota_usage_report_unavailable_total`. A bucket without
-`reservation_protocol` keeps the previous behaviour until it is migrated.
+`reservation_protocol` keeps the previous admission behaviour until it is migrated.
+
+Read-only `GET /rustfs/admin/v3/quota/{bucket}` and
+`GET /rustfs/admin/v3/quota-stats/{bucket}` use optional usage reporting for
+all buckets, including buckets without a quota and snapshot-protocol quotas.
+They return the configured limit while a full scan is still pending:
+`size: null` on the quota response, or `current_usage: null`,
+`remaining_quota: null` and `usage_percentage: null` on statistics when no
+usage is available. Known usage, including zero, remains numeric. Missing
+buckets and configuration errors remain errors. The MinIO-compatible
+`GET /rustfs/admin/v3/get-bucket-quota?bucket=...` continues returning the
+configured quota in its existing format.
+
+These reads do not force an early scanner reload or consult a degraded floor;
+the shared usage lookup retains its normal cache/backend refresh behavior.
+They do not publish intermediate bucket-scan results, so finishing a bucket
+walk does not guarantee immediately updated usage. Legacy write admission
+and quota-check previews retain their existing usage requirements.
+
+Distributed single-bucket usage reads project the requested size directly from
+the cached snapshot. They share the full-snapshot loader's TTL, negative cache,
+serialized refresh and invalidation generation, but do not clone other buckets
+or their histograms. Refresh-only consumers also avoid copying the snapshot.
+The full-snapshot API still returns an owned copy. Cache hits therefore perform
+one bucket lookup; a cold or invalidated cache still requires the existing
+backend load. This does not change scanner publication latency or the local
+mutation cache's refresh policy.

rustfs/src/admin/handlers/quota.rs

@@ -190,8 +190,7 @@ pub struct BucketQuotaResponse {
 pub struct BucketQuotaStats {
     pub bucket: String,
     pub quota_limit: Option<u64>,
-    /// `null` when the bucket is admitted by the reservation ledger and no
-    /// scanner usage is available; never reported as zero.
+    /// `null` when no scanner usage is available; never reported as zero.
     pub current_usage: Option<u64>,
     pub remaining_quota: Option<u64>,
     pub usage_percentage: Option<f64>,
@@ -1120,4 +1119,28 @@ mod tests {
         let json: serde_json::Value = serde_json::to_value(&response).unwrap();
         assert!(json["size"].is_null(), "unknown usage must not be reported as zero: {json}");
     }
+
+    #[test]
+    fn quota_responses_keep_known_zero_usage_numeric() {
+        let quota = BucketQuotaResponse {
+            bucket: "test-bucket".to_string(),
+            quota: Some(2048),
+            size: Some(0),
+            quota_type: default_quota_type(),
+        };
+        let stats = BucketQuotaStats {
+            bucket: "test-bucket".to_string(),
+            quota_limit: Some(2048),
+            current_usage: Some(0),
+            remaining_quota: Some(2048),
+            usage_percentage: Some(0.0),
+        };
+
+        let quota = serde_json::to_value(&quota).expect("quota response should serialize");
+        let stats = serde_json::to_value(&stats).expect("quota statistics should serialize");
+        assert_eq!(quota["size"], 0);
+        assert_eq!(stats["current_usage"], 0);
+        assert_eq!(stats["remaining_quota"], 2048);
+        assert_eq!(stats["usage_percentage"], 0.0);
+    }
 }