test(1306): pin usage serialization, snapshot cache invalidation, and listing send classification (#5000)

test(1306): pin Some(0) usage serialization, snapshot cache invalidation, and listing send classification

Follow-up test hardening for the merged admin-usage-snapshot work
(#4979/#4980/#4981/#4982, rustfs/backlog#1306). Tests only; no production
behavior change.

- B-1 madmin: pin that a scanned-but-empty bucket (Some(0)) serializes usage
  stats as zeros, staying distinct from the no-snapshot (None) omitted case.
- B-2 gating: revert detector proving save_data_usage_in_backend invalidates
  the 30s snapshot cache so a fresh save is visible to the next cached read.
- A-1 list_objects: pin that a successful gather_results send is never
  misclassified as ConsumerGone (correct state + err sentinel delivered), and
  document the wrapper Err arm invariant. A full wrapper-level producer-error
  integration test is deferred as it needs the fake-disk list harness.

crates/ecstore/src/store/list_objects.rs

@@ -4054,6 +4054,13 @@ impl ECStore {
                     // The explicit cancel is idempotent and avoids relying on
                     // the wrapper's drop-guard ordering to stop the producer.
                     Ok(GatherResultsState::ConsumerGone) => cancel.cancel(),
+                    // Invariant (rustfs/backlog#1306): gather_results maps a
+                    // consumer disconnect to Ok(ConsumerGone) and has no
+                    // fallible pre-send path, so this arm is currently
+                    // unreachable. It is kept as a guard: it stays correct only
+                    // while a *real* gather_results error would still surface as
+                    // an error here. Real producer/listing errors take the
+                    // separate job1 -> err_tx -> err_rx path below, unaffected.
                     Err(err) => {
                         log_list_path_worker_error("store", "gather_results", &job2_context, &err);
                         let _ = err_tx2.send(Arc::new(err));
@@ -5319,6 +5326,13 @@ impl Sets {
                     // The explicit cancel is idempotent and avoids relying on
                     // the wrapper's drop-guard ordering to stop the producer.
                     Ok(GatherResultsState::ConsumerGone) => cancel.cancel(),
+                    // Invariant (rustfs/backlog#1306): gather_results maps a
+                    // consumer disconnect to Ok(ConsumerGone) and has no
+                    // fallible pre-send path, so this arm is currently
+                    // unreachable. It is kept as a guard: it stays correct only
+                    // while a *real* gather_results error would still surface as
+                    // an error here. Real producer/listing errors take the
+                    // separate job1 -> err_tx -> err_rx path below, unaffected.
                     Err(err) => {
                         log_list_path_worker_error("sets", "gather_results", &job2_context, &err);
                         let _ = err_tx2.send(Arc::new(err));
@@ -6279,6 +6293,13 @@ impl SetDisks {
                     // The explicit cancel is idempotent and avoids relying on
                     // the wrapper's drop-guard ordering to stop the producer.
                     Ok(GatherResultsState::ConsumerGone) => cancel.cancel(),
+                    // Invariant (rustfs/backlog#1306): gather_results maps a
+                    // consumer disconnect to Ok(ConsumerGone) and has no
+                    // fallible pre-send path, so this arm is currently
+                    // unreachable. It is kept as a guard: it stays correct only
+                    // while a *real* gather_results error would still surface as
+                    // an error here. Real producer/listing errors take the
+                    // separate job1 -> err_tx -> err_rx path below, unaffected.
                     Err(err) => {
                         log_list_path_worker_error("set_disks", "gather_results", &job2_context, &err);
                         let _ = err_tx2.send(Arc::new(err));
@@ -7378,6 +7399,94 @@ mod test {
         assert!(!cancel.is_cancelled());
     }
 
+    /// A-1 guard (rustfs/backlog#1306): pin that a *successful* send is never
+    /// misclassified as `ConsumerGone`. With the receiver alive, gather_results
+    /// must return the real drain state and deliver the payload carrying the
+    /// state-specific `err` sentinel the wrappers depend on (limit reached =>
+    /// err None so the wrapper knows the disk may hold more; input closed =>
+    /// err Some so the wrapper knows the disk was fully drained). Together with
+    /// the receiver-dropped ConsumerGone tests above, this locks in
+    /// "ConsumerGone is returned only on send failure": a broken send-failure
+    /// check that classified a successful send as ConsumerGone would drop the
+    /// delivered payload (and its err sentinel) and trip this test.
+    #[tokio::test]
+    async fn list_path_gather_results_send_success_delivers_state_specific_err_sentinel() {
+        // Limit reached with the receiver alive: LimitReached + err None.
+        {
+            let (entry_tx, entry_rx) = mpsc::channel(4);
+            let (result_tx, mut result_rx) = mpsc::channel(1);
+            let cancel = CancellationToken::new();
+            entry_tx
+                .send(test_meta_entry("obj-a"))
+                .await
+                .expect("test entry should be queued");
+
+            let handle = tokio::spawn(gather_results(
+                cancel.clone(),
+                ListPathOptions {
+                    bucket: "bucket".to_owned(),
+                    limit: 1,
+                    incl_deleted: true,
+                    ..Default::default()
+                },
+                entry_rx,
+                result_tx,
+            ));
+
+            let result = timeout(Duration::from_secs(1), result_rx.recv())
+                .await
+                .expect("limited result should be sent promptly")
+                .expect("limited result should be present");
+            assert_eq!(result.entries.expect("limit payload delivers entries").entries().len(), 1);
+            assert!(result.err.is_none(), "limit-reached payload must carry no err sentinel");
+
+            let state = timeout(Duration::from_secs(1), handle)
+                .await
+                .expect("gather_results should finish")
+                .expect("gather_results task should not panic")
+                .expect("a successful send must not surface as an error");
+            assert_eq!(state, GatherResultsState::LimitReached);
+        }
+
+        // Input closed with the receiver alive: InputClosed + err Some (eof sentinel).
+        {
+            let (entry_tx, entry_rx) = mpsc::channel(4);
+            let (result_tx, mut result_rx) = mpsc::channel(1);
+            let cancel = CancellationToken::new();
+            entry_tx
+                .send(test_meta_entry("obj-a"))
+                .await
+                .expect("test entry should be queued");
+            drop(entry_tx);
+
+            let handle = tokio::spawn(gather_results(
+                cancel.clone(),
+                ListPathOptions {
+                    bucket: "bucket".to_owned(),
+                    limit: 8,
+                    incl_deleted: true,
+                    ..Default::default()
+                },
+                entry_rx,
+                result_tx,
+            ));
+
+            let result = timeout(Duration::from_secs(1), result_rx.recv())
+                .await
+                .expect("eof result should be sent promptly")
+                .expect("eof result should be present");
+            assert_eq!(result.entries.expect("eof payload delivers entries").entries().len(), 1);
+            assert!(result.err.is_some(), "eof payload must carry the sentinel err the wrapper relies on");
+
+            let state = timeout(Duration::from_secs(1), handle)
+                .await
+                .expect("gather_results should finish")
+                .expect("gather_results task should not panic")
+                .expect("a successful send must not surface as an error");
+            assert_eq!(state, GatherResultsState::InputClosed);
+        }
+    }
+
     // The subscriber installed via `set_default` is thread-local, so this test
     // must run on the current-thread runtime; a multi-thread runtime would make
     // the log-absence assertion vacuous.

crates/madmin/src/user.rs

@@ -771,6 +771,33 @@ mod tests {
         assert_eq!(value["object_versions_histogram"]["SINGLE_VERSION"], 7);
     }
 
+    /// Wire pin (rustfs/backlog#1306): a scanned-but-empty bucket carries
+    /// Some(0) usage stats that must serialize as zeros, staying distinct from
+    /// the no-snapshot (None) case that is omitted. Guards against replacing
+    /// `Option::is_none` with a predicate that also skips zero values.
+    #[test]
+    fn bucket_access_info_serializes_zero_usage_stats() {
+        let info = BucketAccessInfo {
+            name: "empty-snapshot".to_string(),
+            size: Some(0),
+            objects: Some(0),
+            object_sizes_histogram: Some(HashMap::new()),
+            object_versions_histogram: Some(HashMap::new()),
+            ..Default::default()
+        };
+
+        let value = serde_json::to_value(&info).unwrap();
+        let obj = value.as_object().unwrap();
+        assert!(obj.contains_key("size"));
+        assert!(obj.contains_key("objects"));
+        assert!(obj.contains_key("object_sizes_histogram"));
+        assert!(obj.contains_key("object_versions_histogram"));
+        assert_eq!(value.get("size"), Some(&serde_json::json!(0)));
+        assert_eq!(value["objects"], 0);
+        assert_eq!(value["object_sizes_histogram"], serde_json::json!({}));
+        assert_eq!(value["object_versions_histogram"], serde_json::json!({}));
+    }
+
     #[test]
     fn test_account_status_try_from_invalid() {
         let result = AccountStatus::try_from("invalid");

rustfs/src/app/data_usage_snapshot_gating_test.rs

@@ -29,13 +29,14 @@ use super::storage_api::test::StoragePutObjReader as PutObjReader;
 use super::storage_api::test::contract::bucket::{BucketOperations, MakeBucketOptions};
 use super::storage_api::test::contract::object::ObjectIO as _;
 use super::storage_api::test::data_usage::{
-    compute_bucket_usage, live_bucket_usage_computations, record_bucket_object_write_memory, store_data_usage_in_backend,
+    compute_bucket_usage, live_bucket_usage_computations, load_data_usage_from_backend_cached, record_bucket_object_write_memory,
+    store_data_usage_in_backend,
 };
 use crate::app::admin_usecase::DefaultAdminUsecase;
 use rustfs_data_usage::{BucketUsageInfo, DataUsageInfo};
 use serial_test::serial;
 use std::collections::HashMap;
-use std::time::SystemTime;
+use std::time::{Duration, SystemTime};
 use uuid::Uuid;
 
 const SEEDED_BUCKET_SIZE: u64 = 123_456;
@@ -61,6 +62,26 @@ fn seeded_data_usage_info(bucket: &str, last_update: SystemTime) -> DataUsageInf
     info
 }
 
+fn sized_data_usage_info(bucket: &str, size: u64, last_update: SystemTime) -> DataUsageInfo {
+    let usage = BucketUsageInfo {
+        size,
+        objects_count: 1,
+        versions_count: 1,
+        ..Default::default()
+    };
+
+    let mut info = DataUsageInfo {
+        last_update: Some(last_update),
+        buckets_count: 1,
+        objects_total_count: 1,
+        objects_total_size: size,
+        ..Default::default()
+    };
+    info.buckets_usage = HashMap::from([(bucket.to_string(), usage)]);
+    info.bucket_sizes = HashMap::from([(bucket.to_string(), size)]);
+    info
+}
+
 #[tokio::test]
 #[serial]
 async fn data_usage_endpoint_serves_snapshot_without_live_listing() {
@@ -134,6 +155,60 @@ async fn data_usage_endpoint_serves_snapshot_without_live_listing() {
     assert_eq!(overlay_usage.size, 512);
 }
 
+/// Revert detector for the snapshot-cache invalidation in
+/// `save_data_usage_in_backend` (rustfs/backlog#1306): a fresh scanner save must
+/// be visible to the very next cached read, not deferred until the 30s TTL
+/// expires. Removing the `*data_usage_snapshot_cache().write().await = None`
+/// invalidation makes the post-save cached read return the stale warmed value
+/// and trips this test.
+#[tokio::test]
+#[serial]
+async fn save_data_usage_invalidates_snapshot_cache() {
+    let ecstore = shared_gating_ecstore().await;
+    let bucket = format!("usage-invalidate-{}", Uuid::new_v4());
+
+    // Future-date both snapshots beyond the stale-guard's 5min tolerance so the
+    // persists win deterministically over whatever snapshot a sibling serial
+    // test may have left behind, regardless of test execution order.
+    let warm_at = SystemTime::now() + Duration::from_secs(600);
+    let changed_at = warm_at + Duration::from_secs(1);
+
+    const WARM_SIZE: u64 = 111_000;
+    const CHANGED_SIZE: u64 = 222_000;
+
+    // Persist an initial snapshot and warm the cache from it.
+    store_data_usage_in_backend(sized_data_usage_info(&bucket, WARM_SIZE, warm_at), ecstore.clone())
+        .await
+        .expect("persist initial snapshot");
+    let warmed = load_data_usage_from_backend_cached(ecstore.clone())
+        .await
+        .expect("warm cached snapshot");
+    assert_eq!(
+        warmed.buckets_usage.get(&bucket).map(|usage| usage.size),
+        Some(WARM_SIZE),
+        "cache must be warmed with the initial snapshot before the changed save"
+    );
+
+    // Persist a changed snapshot; the save must invalidate the warmed cache.
+    store_data_usage_in_backend(sized_data_usage_info(&bucket, CHANGED_SIZE, changed_at), ecstore.clone())
+        .await
+        .expect("persist changed snapshot");
+
+    let reloaded = load_data_usage_from_backend_cached(ecstore.clone())
+        .await
+        .expect("reload cached snapshot after save");
+    assert_eq!(
+        reloaded.buckets_usage.get(&bucket).map(|usage| usage.size),
+        Some(CHANGED_SIZE),
+        "cached read after save must observe the new snapshot, not the stale warmed value"
+    );
+    assert_eq!(
+        reloaded.last_update,
+        Some(changed_at),
+        "cached read after save must report the new snapshot timestamp"
+    );
+}
+
 /// Wire pin for the no-snapshot response shape (rustfs/backlog#1306): a
 /// default `DataUsageInfo` must keep serializing `last_update` as `null` with
 /// empty bucket maps, so "no snapshot yet" stays distinguishable from real