From 368eb782c8f111750a0628c852fbb063522c5df7 Mon Sep 17 00:00:00 2001 From: coder 1 Date: Mon, 17 Aug 2026 13:08:28 -0400 Subject: [PATCH] Remove storage sweep object cap Co-authored-by: bb-expert <51cde9442e46eac81c83ec71552708c3b0cb96a88ff8e1581d8fb4de281afbf3@buzz.block.builderlab.xyz> Signed-off-by: bb-expert <51cde9442e46eac81c83ec71552708c3b0cb96a88ff8e1581d8fb4de281afbf3@buzz.block.builderlab.xyz> --- crates/buzz-media/src/bucket_index.rs | 104 +++++++++++++++---------- crates/buzz-media/src/storage.rs | 4 +- crates/buzz-relay/src/main.rs | 3 +- crates/buzz-relay/src/storage_sweep.rs | 23 ++---- 4 files changed, 70 insertions(+), 64 deletions(-) diff --git a/crates/buzz-media/src/bucket_index.rs b/crates/buzz-media/src/bucket_index.rs index 6c78c2e0a..9ad74a6aa 100644 --- a/crates/buzz-media/src/bucket_index.rs +++ b/crates/buzz-media/src/bucket_index.rs @@ -3,7 +3,7 @@ //! //! This module has **zero S3 I/O** — [`classify_key`] and [`BucketAggregate`] //! operate on plain `(key, size)` pairs, and [`fold_bucket_listing`] takes a -//! caller-supplied page-fetching closure so the pagination/cap logic is +//! caller-supplied page-fetching closure so continuation-token pagination is //! testable against synthetic listings. The relay wires a real //! [`crate::storage::MediaStorage::list_page`] closure at the call site (see //! `buzz-relay`'s storage sweep task). @@ -339,9 +339,11 @@ impl BucketAggregate { /// sweep, keep the old snapshot" to the caller — never a partial one. #[derive(Debug, thiserror::Error)] pub enum SweepError { - /// Cumulative listed-object count exceeded `cap` mid-listing. - #[error("object cap exceeded: {seen} listed objects > cap {cap}")] - CapExceeded { seen: u64, cap: u64 }, + /// The deletion taxonomy sweep exceeded its fleet safety cap. Storage + /// metrics sweeps do not have an object cap; they rely on paginated LIST + /// responses plus the caller's whole-sweep timeout. + #[error("taxonomy object cap exceeded: {seen} listed objects > cap {cap}")] + TaxonomyObjectCap { seen: u64, cap: u64 }, /// The page source (S3, or a test double) failed. #[error("storage error during listing: {0}")] Storage(#[from] MediaError), @@ -367,33 +369,25 @@ pub struct Page { pub is_truncated: bool, } -/// Fold an entire paginated bucket listing, checking the object cap BEFORE -/// folding each page and never retaining the full listing — only the -/// bounded per-sha/per-binding aggregate state. +/// Fold an entire paginated bucket listing without retaining the full listing +/// — only the per-sha/per-binding aggregate state. Each call consumes one S3 +/// response page and follows continuation tokens until the listing completes; +/// callers bound stalled or pathological listings with a whole-sweep timeout. /// /// `fetch_page` is called with `None` for the first page and the previous /// page's continuation token thereafter; production callers close over a /// [`crate::storage::MediaStorage`], tests close over canned [`Page`]s. -pub async fn fold_bucket_listing( - cap: u64, - mut fetch_page: F, -) -> Result +pub async fn fold_bucket_listing(mut fetch_page: F) -> Result where F: FnMut(Option) -> Fut, Fut: Future>, { let mut aggregate = BucketAggregate::default(); let mut continuation_token = None; - let mut seen: u64 = 0; loop { let page = fetch_page(continuation_token.take()).await?; - seen += page.objects.len() as u64; - if seen > cap { - return Err(SweepError::CapExceeded { seen, cap }); - } - for (key, size) in &page.objects { aggregate.fold(key, *size); } @@ -649,11 +643,11 @@ mod tests { assert_eq!(snap.unknown_key_objects, 0); } - // --- fold_bucket_listing (pagination + cap) --- + // --- fold_bucket_listing (pagination) --- #[tokio::test] async fn empty_bucket_listing_yields_zero_snapshot() { - let snapshot = fold_bucket_listing(100, |_token| async { + let snapshot = fold_bucket_listing(|_token| async { Ok(Page { objects: vec![], next_continuation_token: None, @@ -682,7 +676,7 @@ mod tests { }, ])); - let snapshot = fold_bucket_listing(100, { + let snapshot = fold_bucket_listing({ let pages = std::sync::Arc::clone(&pages); move |token| { let pages = std::sync::Arc::clone(&pages); @@ -705,32 +699,55 @@ mod tests { } #[tokio::test] - async fn cap_breach_mid_listing_fails_the_sweep_before_folding_the_page() { - let objects: Vec<(String, u64)> = (0..5).map(|i| (format!("obj-{i}"), 1)).collect(); - let result = fold_bucket_listing(3, move |_token| { - let objects = objects.clone(); - async move { - Ok(Page { - objects, - next_continuation_token: None, - is_truncated: false, - }) + async fn paginated_listing_has_no_cumulative_object_cap() { + let old_default_cap = 1_000_000u64; + let page_size = 1_000usize; + let total_pages_past_old_cap = old_default_cap as usize / page_size + 1; + let calls = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); + + let snapshot = fold_bucket_listing({ + let calls = std::sync::Arc::clone(&calls); + move |token| { + let calls = std::sync::Arc::clone(&calls); + async move { + let page_index = calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + if page_index == 0 { + assert!(token.is_none()); + } else { + assert_eq!( + token.as_deref(), + Some(format!("page-{page_index}").as_str()) + ); + } + + Ok(Page { + objects: std::iter::repeat_with(|| ("unknown".to_string(), 1)) + .take(page_size) + .collect(), + next_continuation_token: (page_index + 1 < total_pages_past_old_cap) + .then(|| format!("page-{}", page_index + 1)), + is_truncated: page_index + 1 < total_pages_past_old_cap, + }) + } } }) - .await; + .await + .expect("listing past the former storage-sweep cap must succeed"); - match result { - Err(SweepError::CapExceeded { seen, cap }) => { - assert_eq!(seen, 5); - assert_eq!(cap, 3); - } - other => panic!("expected CapExceeded, got {other:?}"), - } + let expected_objects = total_pages_past_old_cap as u64 * page_size as u64; + assert!(expected_objects > old_default_cap); + assert_eq!(snapshot.physical_objects, expected_objects); + assert_eq!(snapshot.physical_bytes, expected_objects); + assert_eq!(snapshot.unknown_key_objects, expected_objects); + assert_eq!( + calls.load(std::sync::atomic::Ordering::SeqCst), + total_pages_past_old_cap + ); } #[tokio::test] async fn storage_error_propagates_from_page_source() { - let result: Result = fold_bucket_listing(10, |_token| async { + let result: Result = fold_bucket_listing(|_token| async { Err(MediaError::StorageError("boom".to_string())) }) .await; @@ -739,7 +756,7 @@ mod tests { #[tokio::test] async fn truncated_page_with_no_continuation_token_fails_the_sweep() { - let result = fold_bucket_listing(100, |_token| async { + let result = fold_bucket_listing(|_token| async { Ok(Page { objects: vec![("some-key".to_string(), 1)], next_continuation_token: None, @@ -811,7 +828,8 @@ pub struct TaxonomySweepOutcome { /// does not understand is unsafe — but that is a *fleet* invariant, not a /// per-request one. This sweep records it once; deletion stages then gate on /// a recent clean sweep instead of re-listing the whole bucket per request. -/// Same pagination/cap contract as [`fold_bucket_listing`]; memory is +/// Same continuation-token pagination contract as [`fold_bucket_listing`]; +/// this fleet safety sweep still enforces its explicit cap, and memory is /// bounded by `sample_limit`, never the listing size. pub async fn sweep_bucket_taxonomy( cap: u64, @@ -828,7 +846,7 @@ where let page = fetch_page(continuation_token.take()).await?; outcome.listed_objects += page.objects.len() as u64; if outcome.listed_objects > cap { - return Err(SweepError::CapExceeded { + return Err(SweepError::TaxonomyObjectCap { seen: outcome.listed_objects, cap, }); @@ -1005,6 +1023,6 @@ mod deletion_taxonomy_tests { }) }) .await; - assert!(matches!(result, Err(SweepError::CapExceeded { .. }))); + assert!(matches!(result, Err(SweepError::TaxonomyObjectCap { .. }))); } } diff --git a/crates/buzz-media/src/storage.rs b/crates/buzz-media/src/storage.rs index 0f0aa7af6..56f717c7a 100644 --- a/crates/buzz-media/src/storage.rs +++ b/crates/buzz-media/src/storage.rs @@ -286,8 +286,8 @@ impl MediaStorage { /// [`crate::bucket_index::Page`] shape the pure fold consumes. /// /// `max_keys` bounds one HTTP response, not the sweep's total object - /// cap — the caller (`fold_bucket_listing`) enforces the cumulative cap - /// across pages. + /// count. Callers keep page memory bounded by following continuation + /// tokens one page at a time. pub async fn list_page( &self, continuation_token: Option, diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 3584e1849..41cf2526d 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -1652,13 +1652,12 @@ async fn run_storage_sweep_tick( } let media_storage = Arc::clone(&state.media_storage); - let max_objects = config.max_objects; storage_sweep::maybe_spawn_sweep( &state.storage_sweep, config.interval, config.timeout, async move { - buzz_media::fold_bucket_listing(max_objects, move |token| { + buzz_media::fold_bucket_listing(move |token| { let media_storage = Arc::clone(&media_storage); async move { media_storage.list_page(token, 1000).await } }) diff --git a/crates/buzz-relay/src/storage_sweep.rs b/crates/buzz-relay/src/storage_sweep.rs index eccadcd83..7d54a7365 100644 --- a/crates/buzz-relay/src/storage_sweep.rs +++ b/crates/buzz-relay/src/storage_sweep.rs @@ -37,9 +37,6 @@ pub struct StorageSweepConfig { /// independent — the usage tick never awaits the sweep, so this bounds /// how long a stalled attempt occupies the single in-flight slot. pub timeout: Duration, - /// Cumulative listed-object cap; a listing that exceeds it fails the - /// attempt (old snapshot kept) rather than growing memory unbounded. - pub max_objects: u64, /// Kill switch. `false` ⇒ no sweep ever spawns and no storage-family /// gauge (including the health gauges) is ever emitted — a relay whose /// deployment lacks `s3:ListBucket` can turn the whole feature off. @@ -48,8 +45,7 @@ pub struct StorageSweepConfig { impl StorageSweepConfig { /// Reads `BUZZ_STORAGE_SWEEP_INTERVAL_SECS` (default 3600, floor 60), - /// `BUZZ_STORAGE_SWEEP_TIMEOUT_SECS` (default 120), - /// `BUZZ_STORAGE_SWEEP_MAX_OBJECTS` (default 1_000_000), and the + /// `BUZZ_STORAGE_SWEEP_TIMEOUT_SECS` (default 120), and the /// `BUZZ_STORAGE_METRICS` kill switch (`off` ⇒ disabled, anything else /// including unset ⇒ enabled). pub fn from_env() -> Self { @@ -62,10 +58,6 @@ impl StorageSweepConfig { .ok() .and_then(|v| v.parse::().ok()) .unwrap_or(120); - let max_objects = std::env::var("BUZZ_STORAGE_SWEEP_MAX_OBJECTS") - .ok() - .and_then(|v| v.parse::().ok()) - .unwrap_or(1_000_000); let enabled = std::env::var("BUZZ_STORAGE_METRICS") .ok() .map(|v| v.trim().to_ascii_lowercase()) @@ -74,7 +66,6 @@ impl StorageSweepConfig { Self { interval: Duration::from_secs(interval_secs), timeout: Duration::from_secs(timeout_secs), - max_objects, enabled, } } @@ -155,8 +146,8 @@ pub struct StorageSweepState { /// permanently failing sweep (e.g. missing `s3:ListBucket`) will retry on /// every usage tick (default 300 s), not at the sweep-interval cadence. /// This is intentional: a permission failure (the common persistent case) -/// costs a single cheap LIST call per retry, other failures (timeout, cap, -/// malformed page) are bounded by the sweep's own timeout and object caps, +/// costs a single cheap LIST call per retry, other failures (timeout or +/// malformed page) are bounded by the sweep's own timeout, /// and tick-cadence retry means the sweep self-heals as soon as the /// underlying cause is fixed. The tick cadence is documented in values.yaml. fn should_spawn( @@ -387,7 +378,6 @@ mod tests { for key in [ "BUZZ_STORAGE_SWEEP_INTERVAL_SECS", "BUZZ_STORAGE_SWEEP_TIMEOUT_SECS", - "BUZZ_STORAGE_SWEEP_MAX_OBJECTS", "BUZZ_STORAGE_METRICS", ] { if std::env::var(key).is_ok() { @@ -397,7 +387,6 @@ mod tests { let config = StorageSweepConfig::from_env(); assert_eq!(config.interval, Duration::from_secs(3600)); assert_eq!(config.timeout, Duration::from_secs(120)); - assert_eq!(config.max_objects, 1_000_000); assert!(config.enabled); } @@ -513,7 +502,7 @@ mod tests { &state, Duration::from_secs(3600), Duration::from_secs(5), - async { Err(SweepError::CapExceeded { seen: 5, cap: 1 }) }, + async { Err(SweepError::MalformedPage) }, ) .await; tokio::task::yield_now().await; @@ -564,7 +553,7 @@ mod tests { &state, Duration::from_secs(3600), Duration::from_secs(5), - async { Err(SweepError::CapExceeded { seen: 5, cap: 1 }) }, + async { Err(SweepError::MalformedPage) }, ) .await; tokio::task::yield_now().await; @@ -572,7 +561,7 @@ mod tests { &state, Duration::from_secs(3600), Duration::from_secs(5), - async { Err(SweepError::CapExceeded { seen: 5, cap: 1 }) }, + async { Err(SweepError::MalformedPage) }, ) .await;