External/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-08-18 06:50:31 +02:00
Code Issues Packages Projects Releases Wiki Activity

Remove storage sweep object cap

Browse Source 
Co-authored-by: bb-expert <51cde9442e46eac81c83ec71552708c3b0cb96a88ff8e1581d8fb4de281afbf3@buzz.block.builderlab.xyz>
Signed-off-by: bb-expert <51cde9442e46eac81c83ec71552708c3b0cb96a88ff8e1581d8fb4de281afbf3@buzz.block.builderlab.xyz>
This commit is contained in:
coder 1
2026-08-17 13:24:28 -04:00
co-authored by bb-expert
parent f716eef437
commit 368eb782c8
4 changed files with 70 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/buzz-media/src/bucket_index.rs
+61 -43
View File
@@ -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<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);
}
@@ -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<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;
@@ -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<F, Fut>(
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 { .. })));
}
}
crates/buzz-media/src/storage.rs
+2 -2
View File
@@ -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>,
crates/buzz-relay/src/main.rs
+1 -2
View File
@@ -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 }
})
crates/buzz-relay/src/storage_sweep.rs
+6 -17
View File
@@ -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::<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())
@@ -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;