diff --git a/crates/ecstore/src/store/list_objects.rs b/crates/ecstore/src/store/list_objects.rs index f0458cbe1..81b8cdf39 100644 --- a/crates/ecstore/src/store/list_objects.rs +++ b/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. diff --git a/crates/madmin/src/user.rs b/crates/madmin/src/user.rs index fe7de5bc5..b6c9ca249 100644 --- a/crates/madmin/src/user.rs +++ b/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"); diff --git a/rustfs/src/app/data_usage_snapshot_gating_test.rs b/rustfs/src/app/data_usage_snapshot_gating_test.rs index d2ab87ff1..eb6b27771 100644 --- a/rustfs/src/app/data_usage_snapshot_gating_test.rs +++ b/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