Skip to content

fix(media): remove storage sweep object cap #6143

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
bradseiler wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix(media): remove storage sweep object cap #6143

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
104 changes: 61 additions & 43 deletions crates/buzz-media/src/bucket_index.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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).
Expand Down Expand Up @@ -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),
Expand All @@ -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<F, Fut>(
cap: u64,
mut fetch_page: F,
) -> Result<BucketSnapshot, SweepError>
pub async fn fold_bucket_listing<F, Fut>(mut fetch_page: F) -> Result<BucketSnapshot, SweepError>
where
F: FnMut(Option<String>) -> Fut,
Fut: Future<Output = Result<Page, MediaError>>,
{
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);
}
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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);
Expand All @@ -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<BucketSnapshot, SweepError> = fold_bucket_listing(10, |_token| async {
let result: Result<BucketSnapshot, SweepError> = fold_bucket_listing(|_token| async {
Err(MediaError::StorageError("boom".to_string()))
})
.await;
Expand All @@ -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,
Expand Down Expand Up @@ -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<F, Fut>(
cap: u64,
Expand All @@ -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,
});
Expand Down Expand Up @@ -1005,6 +1023,6 @@ mod deletion_taxonomy_tests {
})
})
.await;
assert!(matches!(result, Err(SweepError::CapExceeded { .. })));
assert!(matches!(result, Err(SweepError::TaxonomyObjectCap { .. })));
}
}
4 changes: 2 additions & 2 deletions crates/buzz-media/src/storage.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<String>,
Expand Down
3 changes: 1 addition & 2 deletions crates/buzz-relay/src/main.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 }
})
Expand Down
23 changes: 6 additions & 17 deletions crates/buzz-relay/src/storage_sweep.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand All @@ -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 {
Expand All @@ -62,10 +58,6 @@ impl StorageSweepConfig {
.ok()
.and_then(|v| v.parse::<u64>().ok())
.unwrap_or(120);
let max_objects = std::env::var("BUZZ_STORAGE_SWEEP_MAX_OBJECTS")
.ok()
.and_then(|v| v.parse::<u64>().ok())
.unwrap_or(1_000_000);
let enabled = std::env::var("BUZZ_STORAGE_METRICS")
.ok()
.map(|v| v.trim().to_ascii_lowercase())
Expand All @@ -74,7 +66,6 @@ impl StorageSweepConfig {
Self {
interval: Duration::from_secs(interval_secs),
timeout: Duration::from_secs(timeout_secs),
max_objects,
enabled,
}
}
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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() {
Expand All @@ -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);
}

Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -564,15 +553,15 @@ 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;
maybe_spawn_sweep(
&state,
Duration::from_secs(3600),
Duration::from_secs(5),
async { Err(SweepError::CapExceeded { seen: 5, cap: 1 }) },
async { Err(SweepError::MalformedPage) },
)
.await;

Expand Down
Loading