From 3b05b754bef153656d961e49bc6ae71a47d0bc54 Mon Sep 17 00:00:00 2001 From: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 Date: Fri, 17 Jul 2026 15:40:02 -0500 Subject: [PATCH 1/5] feat(media): add S3-truth per-community storage sweep Adds the hourly S3 bucket sweep that computes per-community storage metrics directly from bucket listings (source of truth), with audit-log per-user attribution planned as a follow-on overlay. - buzz-media::bucket_index: pure key classifier (thumb/blob/sidecar/ auxiliary/unknown) + incremental aggregation fold, zero I/O, unit tested against synthetic listings covering multi-variant SHAs, mixed-case extensions, orphans, unmapped communities, and malformed keys. - buzz-relay::storage_sweep: single-flight spawned sweep task with a cached snapshot re-emitted every usage-metrics tick, so the DB- derived gauges keep their configured cadence regardless of sweep duration and a sweep from a demoted leader never publishes stale gauges. - Wires StorageSweepState into AppState and the leader-only branch of the usage-metrics tick; BUZZ_STORAGE_SWEEP_* env knobs plus a BUZZ_STORAGE_METRICS kill switch. - Documents the s3:ListBucket deployment prerequisite in the Helm chart's values.yaml. See PLANS/S3_STORAGE_METRICS_PLAN.md (Rev 3) for the full design. Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- Cargo.lock | 1 + crates/buzz-media/Cargo.toml | 2 +- crates/buzz-media/src/bucket_index.rs | 737 +++++++++++++++++++++++++ crates/buzz-media/src/lib.rs | 5 + crates/buzz-media/src/storage.rs | 34 ++ crates/buzz-relay/Cargo.toml | 1 + crates/buzz-relay/src/lib.rs | 1 + crates/buzz-relay/src/main.rs | 43 ++ crates/buzz-relay/src/state.rs | 7 + crates/buzz-relay/src/storage_sweep.rs | 703 +++++++++++++++++++++++ deploy/charts/buzz/values.yaml | 8 + 11 files changed, 1541 insertions(+), 1 deletion(-) create mode 100644 crates/buzz-media/src/bucket_index.rs create mode 100644 crates/buzz-relay/src/storage_sweep.rs diff --git a/Cargo.lock b/Cargo.lock index 0afe1e2773..5faa0d9c98 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1135,6 +1135,7 @@ dependencies = [ "chrono", "dashmap", "deadpool-redis", + "futures", "futures-util", "hex", "hmac 0.13.0", diff --git a/crates/buzz-media/Cargo.toml b/crates/buzz-media/Cargo.toml index aee5fdbdf4..530ce69c90 100644 --- a/crates/buzz-media/Cargo.toml +++ b/crates/buzz-media/Cargo.toml @@ -19,6 +19,7 @@ sha2 = { workspace = true } hex = { workspace = true } chrono = { workspace = true } ulid = "1" +uuid = { workspace = true } axum = { workspace = true } s3 = { version = "0.37", package = "rust-s3", default-features = false, features = ["tokio-rustls-tls", "fail-on-err", "tags"] } infer = "0.19" @@ -34,4 +35,3 @@ futures-core = "0.3" [dev-dependencies] tokio = { workspace = true, features = ["test-util"] } -uuid = { workspace = true } diff --git a/crates/buzz-media/src/bucket_index.rs b/crates/buzz-media/src/bucket_index.rs new file mode 100644 index 0000000000..cf92291b58 --- /dev/null +++ b/crates/buzz-media/src/bucket_index.rs @@ -0,0 +1,737 @@ +//! Bucket key taxonomy classifier and pure aggregation fold for the S3-truth +//! storage sweep. +//! +//! 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 +//! 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). +//! +//! Five key classes (thumb matched first, everything unrecognized is +//! `Unknown` — never silently folded into another class): +//! +//! | Class | Shape | +//! |---|---| +//! | thumb | `{sha256}.thumb.jpg` | +//! | blob | `{sha256}.{ext}` (ext: 1-8 mixed-case alphanumeric) | +//! | sidecar | `_meta/{community-uuid}/{sha256}.json` | +//! | auxiliary | `_uploads/{community-uuid}/{sha256}/{ulid}.json` | +//! | unknown | everything else | + +use std::collections::HashMap; +use std::future::Future; + +use uuid::Uuid; + +use crate::error::MediaError; + +/// The classification of one bucket key. `Unknown` is the deliberate +/// catch-all — a malformed variant of a known prefix (e.g. a truncated +/// `_uploads/` key) falls to `Unknown` rather than being coerced into +/// `Auxiliary`, so visibility gauges stay loud instead of silently wrong. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum KeyClass { + /// `{sha256}.thumb.jpg` — attributed to the blob's sha. + Thumb { sha256: String }, + /// `{sha256}.{ext}` — physical bytes, logical join key. + Blob { sha256: String, ext: String }, + /// `_meta/{community}/{sha256}.json` — the (community, sha) binding. + Sidecar { community: Uuid, sha256: String }, + /// `_uploads/{community}/{sha256}/{event_id}.json` — fleet physical only. + Auxiliary { + community: Uuid, + sha256: String, + event_id: String, + }, + /// Anything that doesn't match one of the four strict shapes above. + Unknown, +} + +/// Classify one bucket key. Matches `thumb` first (its suffix is a superset +/// shape of the blob pattern's segment count), then blob, sidecar, +/// auxiliary, and finally unknown. See module docs for the exact shapes. +pub fn classify_key(key: &str) -> KeyClass { + if let Some(sha256) = parse_thumb_key(key) { + return KeyClass::Thumb { sha256 }; + } + if let Some((sha256, ext)) = parse_blob_key(key) { + return KeyClass::Blob { sha256, ext }; + } + if let Some((community, sha256)) = parse_sidecar_key(key) { + return KeyClass::Sidecar { community, sha256 }; + } + if let Some((community, sha256, event_id)) = parse_auxiliary_key(key) { + return KeyClass::Auxiliary { + community, + sha256, + event_id, + }; + } + KeyClass::Unknown +} + +/// A 64-char lowercase-hex SHA-256 digest, strictly. +fn is_sha256(s: &str) -> bool { + s.len() == 64 + && s.bytes() + .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b)) +} + +/// Blob extension charset: 1-8 mixed-case alphanumeric chars (F4-bis — infer +/// 0.19 emits uppercase `Z` for `application/x-compress`, which is +/// legitimate writer output that must classify as a blob, not unknown). +fn is_blob_ext(s: &str) -> bool { + !s.is_empty() && s.len() <= 8 && s.bytes().all(|b| b.is_ascii_alphanumeric()) +} + +/// Strict Crockford-base32 ULID charset check, uppercase only (matches the +/// ulid crate's `Display` output — see `upload_record.rs`'s writer). Not +/// `ulid::Ulid::from_string`, which is deliberately case-insensitive on +/// decode and would accept lowercase variants the writer never produces — +/// looser than the plan's anchored regex. +fn is_ulid_charset(s: &str) -> bool { + s.len() == 26 + && s.bytes().all(|b| { + b.is_ascii_digit() + || (b'A'..=b'H').contains(&b) + || b == b'J' + || b == b'K' + || b == b'M' + || b == b'N' + || (b'P'..=b'T').contains(&b) + || (b'V'..=b'Z').contains(&b) + }) +} + +/// Strict canonical UUID: exactly 36 chars, lowercase hex + hyphens at +/// positions 8/13/18/23. Rejects the braced/urn/no-hyphen forms +/// `Uuid::parse_str` alone would otherwise accept — every UUID this server +/// writes into a key is `Display`-formatted canonically lowercase, so +/// anything else is not a UUID we wrote and must not silently parse as one. +fn parse_canonical_uuid(s: &str) -> Option { + if s.len() != 36 { + return None; + } + for (i, b) in s.bytes().enumerate() { + let ok = match i { + 8 | 13 | 18 | 23 => b == b'-', + _ => b.is_ascii_digit() || (b'a'..=b'f').contains(&b), + }; + if !ok { + return None; + } + } + Uuid::parse_str(s).ok() +} + +/// `{sha256}.thumb.jpg` +fn parse_thumb_key(key: &str) -> Option { + let mut parts = key.split('.'); + let sha256 = parts.next()?; + let thumb = parts.next()?; + let jpg = parts.next()?; + if parts.next().is_some() || thumb != "thumb" || jpg != "jpg" || !is_sha256(sha256) { + return None; + } + Some(sha256.to_string()) +} + +/// `{sha256}.{ext}` — exactly two dot-separated segments. +fn parse_blob_key(key: &str) -> Option<(String, String)> { + let mut parts = key.split('.'); + let sha256 = parts.next()?; + let ext = parts.next()?; + if parts.next().is_some() || !is_sha256(sha256) || !is_blob_ext(ext) { + return None; + } + Some((sha256.to_string(), ext.to_string())) +} + +/// `_meta/{community}/{sha256}.json` +fn parse_sidecar_key(key: &str) -> Option<(Uuid, String)> { + let mut segments = key.split('/'); + if segments.next()? != "_meta" { + return None; + } + let community = parse_canonical_uuid(segments.next()?)?; + let last = segments.next()?; + if segments.next().is_some() { + return None; + } + let mut last_parts = last.split('.'); + let sha256 = last_parts.next()?; + let json = last_parts.next()?; + if last_parts.next().is_some() || json != "json" || !is_sha256(sha256) { + return None; + } + Some((community, sha256.to_string())) +} + +/// `_uploads/{community}/{sha256}/{event_id}.json`, `event_id` a ULID. +fn parse_auxiliary_key(key: &str) -> Option<(Uuid, String, String)> { + let mut segments = key.split('/'); + if segments.next()? != "_uploads" { + return None; + } + let community = parse_canonical_uuid(segments.next()?)?; + let sha256 = segments.next()?; + if !is_sha256(sha256) { + return None; + } + let last = segments.next()?; + if segments.next().is_some() { + return None; + } + let mut last_parts = last.split('.'); + let event_id = last_parts.next()?; + let json = last_parts.next()?; + if last_parts.next().is_some() || json != "json" || !is_ulid_charset(event_id) { + return None; + } + Some((community, sha256.to_string(), event_id.to_string())) +} + +/// Per-community logical storage: bytes and object count of bound shas. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct CommunityStorage { + pub bytes: u64, + pub objects: u64, +} + +/// The full computed sweep result: fleet physical/logical totals, +/// per-community logical breakdown, and anomaly/visibility gauges. Pure +/// data — no I/O, cheap to clone into a cached snapshot. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct BucketSnapshot { + /// Every listed object, every class (kind=physical). + pub physical_bytes: u64, + pub physical_objects: u64, + /// Sum of per-community logical bytes/objects (kind=logical). Bills a + /// blob bound to N communities N times — intentional (D-EXT/D-COUNT). + pub logical_bytes: u64, + pub logical_objects: u64, + pub per_community: HashMap, + /// Blob shas with zero sidecar binding in any community. + pub orphan_blob_bytes: u64, + pub orphan_blob_count: u64, + /// Sidecar bindings whose sha has no blob bytes at all. + pub orphan_sidecar_count: u64, + /// Shas with more than one blob variant (anomaly — see plan D-EXT). + pub multi_variant_shas: u64, + /// Total bytes of ALL blob variants belonging to anomalous shas. + pub multi_variant_bytes: u64, + pub unknown_key_bytes: u64, + pub unknown_key_objects: u64, +} + +/// Pure, incremental fold over classified bucket keys. Never retains a full +/// object listing — only per-sha/per-binding running totals, bounded by the +/// number of distinct shas and sidecar bindings actually present. +#[derive(Debug, Default)] +pub struct BucketAggregate { + /// sha -> bytes of every blob variant seen for that sha (D-EXT: multiple + /// entries is the multi-variant anomaly). + blob_variant_bytes: HashMap>, + /// sha -> thumb bytes. At most one thumb key per sha, so a plain insert + /// is correct (no accumulation needed). + thumb_bytes: HashMap, + /// (community, sha) -> sidecar object's own byte size (informational; + /// not part of logical bytes). + sidecar_bindings: HashMap<(Uuid, String), u64>, + physical_bytes: u64, + physical_objects: u64, + unknown_bytes: u64, + unknown_objects: u64, +} + +impl BucketAggregate { + /// Fold one classified `(key, size)` pair into the running aggregate. + pub fn fold(&mut self, key: &str, size: u64) { + self.physical_objects += 1; + self.physical_bytes += size; + match classify_key(key) { + KeyClass::Thumb { sha256 } => { + self.thumb_bytes.insert(sha256, size); + } + KeyClass::Blob { sha256, .. } => { + self.blob_variant_bytes + .entry(sha256) + .or_default() + .push(size); + } + KeyClass::Sidecar { community, sha256 } => { + self.sidecar_bindings.insert((community, sha256), size); + } + KeyClass::Auxiliary { .. } => { + // Fleet physical only — never enters logical/orphan math (F4). + } + KeyClass::Unknown => { + self.unknown_objects += 1; + self.unknown_bytes += size; + } + } + } + + /// Compute the final snapshot from everything folded so far. + pub fn finish(self) -> BucketSnapshot { + let bound_shas: std::collections::HashSet<&str> = self + .sidecar_bindings + .keys() + .map(|(_, sha256)| sha256.as_str()) + .collect(); + + let mut multi_variant_shas = 0u64; + let mut multi_variant_bytes = 0u64; + let mut orphan_blob_count = 0u64; + let mut orphan_blob_bytes = 0u64; + for (sha256, variants) in &self.blob_variant_bytes { + let variant_bytes: u64 = variants.iter().sum(); + if variants.len() > 1 { + multi_variant_shas += 1; + multi_variant_bytes += variant_bytes; + } + if !bound_shas.contains(sha256.as_str()) { + orphan_blob_count += 1; + orphan_blob_bytes += variant_bytes; + } + } + + let orphan_sidecar_count = self + .sidecar_bindings + .keys() + .filter(|(_, sha256)| !self.blob_variant_bytes.contains_key(sha256)) + .count() as u64; + + let mut per_community: HashMap = HashMap::new(); + for (community, sha256) in self.sidecar_bindings.keys() { + let blob_bytes: u64 = self + .blob_variant_bytes + .get(sha256) + .map(|v| v.iter().sum()) + .unwrap_or(0); + let thumb_bytes = self.thumb_bytes.get(sha256).copied().unwrap_or(0); + let entry = per_community.entry(*community).or_default(); + entry.bytes += blob_bytes + thumb_bytes; + entry.objects += 1; + } + let logical_bytes = per_community.values().map(|c| c.bytes).sum(); + let logical_objects = per_community.values().map(|c| c.objects).sum(); + + BucketSnapshot { + physical_bytes: self.physical_bytes, + physical_objects: self.physical_objects, + logical_bytes, + logical_objects, + per_community, + orphan_blob_bytes, + orphan_blob_count, + orphan_sidecar_count, + multi_variant_shas, + multi_variant_bytes, + unknown_key_bytes: self.unknown_bytes, + unknown_key_objects: self.unknown_objects, + } + } +} + +/// Failure modes for the paginated listing fold. All variants mean "failed +/// 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 page source (S3, or a test double) failed. + #[error("storage error during listing: {0}")] + Storage(#[from] MediaError), + /// The whole sweep (this fold plus any caller-side wrapping) exceeded its + /// deadline. Constructed by the relay's sweep task, which wraps this + /// entire function in `tokio::time::timeout` — kept here (not a + /// relay-local error type) so `SweepError` stays the single failure + /// currency the whole sweep pipeline reasons about. + #[error("sweep timed out after {0:?}")] + Timeout(std::time::Duration), +} + +/// One page of a bucket listing, decoupled from any S3 crate type so the +/// fold below can be driven by a synthetic closure in tests. +#[derive(Debug, Clone, Default)] +pub struct Page { + pub objects: Vec<(String, u64)>, + pub next_continuation_token: Option, + 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. +/// +/// `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 +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); + } + + if !page.is_truncated { + break; + } + match page.next_continuation_token { + Some(token) => continuation_token = Some(token), + // Defensive: truncated with no token to resume from is a + // malformed response, not an infinite-loop invitation. + None => break, + } + } + + Ok(aggregate.finish()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn sha(byte: u8) -> String { + hex::encode([byte; 32]) + } + + fn community(n: u128) -> Uuid { + Uuid::from_u128(n) + } + + // --- classify_key --- + + #[test] + fn classifies_thumb_key() { + let s = sha(0xaa); + assert_eq!( + classify_key(&format!("{s}.thumb.jpg")), + KeyClass::Thumb { sha256: s } + ); + } + + #[test] + fn classifies_blob_key_lowercase_ext() { + let s = sha(0xbb); + assert_eq!( + classify_key(&format!("{s}.png")), + KeyClass::Blob { + sha256: s, + ext: "png".to_string() + } + ); + } + + /// F4-bis: infer 0.19 emits uppercase `Z` for `application/x-compress`, + /// legitimate writer output — must classify as blob, not unknown. + #[test] + fn classifies_blob_key_uppercase_z_extension() { + let s = sha(0xcc); + assert_eq!( + classify_key(&format!("{s}.Z")), + KeyClass::Blob { + sha256: s, + ext: "Z".to_string() + } + ); + } + + #[test] + fn classifies_sidecar_key() { + let s = sha(0xdd); + let c = community(1); + assert_eq!( + classify_key(&format!("_meta/{c}/{s}.json")), + KeyClass::Sidecar { + community: c, + sha256: s + } + ); + } + + #[test] + fn classifies_auxiliary_key() { + let s = sha(0xee); + let c = community(2); + let ulid = "01ARZ3NDEKTSV4RRFFQ69G5FAV"; + assert_eq!( + classify_key(&format!("_uploads/{c}/{s}/{ulid}.json")), + KeyClass::Auxiliary { + community: c, + sha256: s, + event_id: ulid.to_string(), + } + ); + } + + #[test] + fn malformed_uploads_key_is_unknown_not_auxiliary() { + let s = sha(0xff); + let c = community(3); + // Lowercase ULID: the writer never emits this — must not silently + // pass as auxiliary; visibility (unknown) beats a wrong guess. + assert_eq!( + classify_key(&format!("_uploads/{c}/{s}/01arz3ndektsv4rrffq69g5fav.json")), + KeyClass::Unknown + ); + // Wrong-length event id. + assert_eq!( + classify_key(&format!("_uploads/{c}/{s}/TOOSHORT.json")), + KeyClass::Unknown + ); + } + + #[test] + fn malformed_sidecar_non_uuid_community_is_unknown() { + let s = sha(0x11); + assert_eq!( + classify_key(&format!("_meta/not-a-uuid/{s}.json")), + KeyClass::Unknown + ); + } + + #[test] + fn malformed_keys_fall_to_unknown() { + let s = sha(0xab); + // Uppercase hex in the sha segment. + assert_eq!( + classify_key(&format!("{}.png", s.to_uppercase())), + KeyClass::Unknown + ); + // Wrong sha length. + assert_eq!(classify_key("abc123.png"), KeyClass::Unknown); + // Extra segment. + assert_eq!(classify_key(&format!("{s}.png.bak")), KeyClass::Unknown); + // Extension too long (9 chars). + assert_eq!(classify_key(&format!("{s}.123456789")), KeyClass::Unknown); + // No extension at all. + assert_eq!(classify_key(&s), KeyClass::Unknown); + // Totally unrelated key. + assert_eq!(classify_key("README.md"), KeyClass::Unknown); + } + + // --- BucketAggregate / finish --- + + #[test] + fn empty_listing_yields_zero_snapshot() { + let snapshot = BucketAggregate::default().finish(); + assert_eq!(snapshot, BucketSnapshot::default()); + } + + #[test] + fn multi_variant_sha_is_anomalous_and_bills_the_sum() { + let s = sha(0x33); + let c = community(4); + let mut agg = BucketAggregate::default(); + agg.fold(&format!("{s}.jpg"), 100); + agg.fold(&format!("{s}.png"), 200); // same sha, second variant + agg.fold(&format!("_meta/{c}/{s}.json"), 10); + + let snap = agg.finish(); + assert_eq!(snap.multi_variant_shas, 1); + assert_eq!(snap.multi_variant_bytes, 300); + assert_eq!(snap.orphan_blob_count, 0); + // Logical bytes bill the sum of both variants (D-EXT). + assert_eq!(snap.per_community[&c].bytes, 300); + assert_eq!(snap.per_community[&c].objects, 1); + } + + #[test] + fn orphan_blob_has_no_sidecar_binding() { + let s = sha(0x44); + let mut agg = BucketAggregate::default(); + agg.fold(&format!("{s}.jpg"), 500); + + let snap = agg.finish(); + assert_eq!(snap.orphan_blob_count, 1); + assert_eq!(snap.orphan_blob_bytes, 500); + assert!(snap.per_community.is_empty()); + assert_eq!(snap.logical_bytes, 0); + } + + #[test] + fn orphan_sidecar_has_no_blob_bytes() { + let s = sha(0x55); + let c = community(5); + let mut agg = BucketAggregate::default(); + agg.fold(&format!("_meta/{c}/{s}.json"), 20); + + let snap = agg.finish(); + assert_eq!(snap.orphan_sidecar_count, 1); + // The binding still counts as a logical object per D-COUNT, but + // contributes zero bytes since there's no blob to bill. + assert_eq!(snap.per_community[&c].objects, 1); + assert_eq!(snap.per_community[&c].bytes, 0); + } + + #[test] + fn unmapped_community_binding_still_aggregates_under_its_uuid() { + // bucket_index has no DB access — "unmapped" (no matching community + // row) is a join the caller performs against per_community's keys. + // Here we only assert the raw UUID is preserved for that join. + let s = sha(0x66); + let c = community(6); + let mut agg = BucketAggregate::default(); + agg.fold(&format!("{s}.jpg"), 40); + agg.fold(&format!("_meta/{c}/{s}.json"), 5); + + let snap = agg.finish(); + assert!(snap.per_community.contains_key(&c)); + } + + #[test] + fn thumb_bytes_attribute_to_the_blobs_sha() { + let s = sha(0x77); + let c = community(7); + let mut agg = BucketAggregate::default(); + agg.fold(&format!("{s}.jpg"), 1000); + agg.fold(&format!("{s}.thumb.jpg"), 50); + agg.fold(&format!("_meta/{c}/{s}.json"), 5); + + let snap = agg.finish(); + assert_eq!(snap.per_community[&c].bytes, 1050); + assert_eq!(snap.per_community[&c].objects, 1); + // Fleet physical totals count every object separately. + assert_eq!(snap.physical_objects, 3); + assert_eq!(snap.physical_bytes, 1055); + } + + #[test] + fn unknown_keys_are_counted_but_excluded_from_logical_math() { + let mut agg = BucketAggregate::default(); + agg.fold("garbage-key", 999); + agg.fold("_meta/not-a-uuid/x.json", 1); + + let snap = agg.finish(); + assert_eq!(snap.unknown_key_objects, 2); + assert_eq!(snap.unknown_key_bytes, 1000); + assert_eq!(snap.physical_objects, 2); + assert_eq!(snap.logical_bytes, 0); + assert!(snap.per_community.is_empty()); + } + + #[test] + fn auxiliary_keys_are_physical_only() { + let s = sha(0x88); + let c = community(8); + let ulid = "01ARZ3NDEKTSV4RRFFQ69G5FAV"; + let mut agg = BucketAggregate::default(); + agg.fold(&format!("_uploads/{c}/{s}/{ulid}.json"), 30); + + let snap = agg.finish(); + assert_eq!(snap.physical_objects, 1); + assert_eq!(snap.physical_bytes, 30); + assert_eq!(snap.logical_bytes, 0); + assert!(snap.per_community.is_empty()); + assert_eq!(snap.unknown_key_objects, 0); + } + + // --- fold_bucket_listing (pagination + cap) --- + + #[tokio::test] + async fn empty_bucket_listing_yields_zero_snapshot() { + let snapshot = fold_bucket_listing(100, |_token| async { + Ok(Page { + objects: vec![], + next_continuation_token: None, + is_truncated: false, + }) + }) + .await + .expect("empty listing must not fail"); + assert_eq!(snapshot, BucketSnapshot::default()); + } + + #[tokio::test] + async fn pagination_follows_continuation_tokens_across_pages() { + let s1 = sha(0x99); + let s2 = sha(0xa1); + let pages = std::sync::Arc::new(std::sync::Mutex::new(vec![ + Page { + objects: vec![(format!("{s1}.jpg"), 10)], + next_continuation_token: Some("page-2".to_string()), + is_truncated: true, + }, + Page { + objects: vec![(format!("{s2}.jpg"), 20)], + next_continuation_token: None, + is_truncated: false, + }, + ])); + + let snapshot = fold_bucket_listing(100, { + let pages = std::sync::Arc::clone(&pages); + move |token| { + let pages = std::sync::Arc::clone(&pages); + async move { + let mut pages = pages.lock().unwrap(); + assert!( + (token.is_none() && pages.len() == 2) + || (token.as_deref() == Some("page-2") && pages.len() == 1) + ); + Ok(pages.remove(0)) + } + } + }) + .await + .expect("two-page listing must succeed"); + + assert_eq!(snapshot.physical_objects, 2); + assert_eq!(snapshot.physical_bytes, 30); + assert_eq!(snapshot.orphan_blob_count, 2); + } + + #[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, + }) + } + }) + .await; + + match result { + Err(SweepError::CapExceeded { seen, cap }) => { + assert_eq!(seen, 5); + assert_eq!(cap, 3); + } + other => panic!("expected CapExceeded, got {other:?}"), + } + } + + #[tokio::test] + async fn storage_error_propagates_from_page_source() { + let result: Result = fold_bucket_listing(10, |_token| async { + Err(MediaError::StorageError("boom".to_string())) + }) + .await; + assert!(matches!(result, Err(SweepError::Storage(_)))); + } +} diff --git a/crates/buzz-media/src/lib.rs b/crates/buzz-media/src/lib.rs index b12f2c2b78..ac05ea6d51 100644 --- a/crates/buzz-media/src/lib.rs +++ b/crates/buzz-media/src/lib.rs @@ -3,6 +3,7 @@ //! Library crate — no Axum dependency for handlers. Axum handlers live in `buzz-relay`. pub mod auth; +pub mod bucket_index; pub mod config; pub mod error; pub mod storage; @@ -12,6 +13,10 @@ pub mod upload; pub mod upload_record; pub mod validation; +pub use bucket_index::{ + classify_key, fold_bucket_listing, BucketAggregate, BucketSnapshot, CommunityStorage, KeyClass, + Page, SweepError, +}; pub use config::MediaConfig; pub use error::MediaError; pub use storage::{BlobHeadMeta, BlobMeta, ByteStream, MediaStorage}; diff --git a/crates/buzz-media/src/storage.rs b/crates/buzz-media/src/storage.rs index a469826df9..0e9809af2f 100644 --- a/crates/buzz-media/src/storage.rs +++ b/crates/buzz-media/src/storage.rs @@ -230,6 +230,40 @@ impl MediaStorage { .ok() .map(|m| m.mime_type) } + + /// One page of a full-bucket listing, for the storage sweep. Wraps + /// rust-s3's manual `list_page` (NOT the auto-paginating `list`, which + /// has no cap) and converts the result into the storage-agnostic + /// [`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. + pub async fn list_page( + &self, + continuation_token: Option, + max_keys: usize, + ) -> Result { + let (result, _status) = self + .bucket + .list_page( + String::new(), + None, + continuation_token, + None, + Some(max_keys), + ) + .await?; + Ok(crate::bucket_index::Page { + objects: result + .contents + .into_iter() + .map(|obj| (obj.key, obj.size)) + .collect(), + next_continuation_token: result.next_continuation_token, + is_truncated: result.is_truncated, + }) + } } #[cfg(test)] diff --git a/crates/buzz-relay/Cargo.toml b/crates/buzz-relay/Cargo.toml index 4ef5708fe4..0aba81ec84 100644 --- a/crates/buzz-relay/Cargo.toml +++ b/crates/buzz-relay/Cargo.toml @@ -89,3 +89,4 @@ buzz-core = { workspace = true, features = ["test-utils"] } buzz-auth = { workspace = true, features = ["dev"] } reqwest = { workspace = true } tokio-tungstenite = { workspace = true } +futures = "0.3" diff --git a/crates/buzz-relay/src/lib.rs b/crates/buzz-relay/src/lib.rs index 6271f9ff70..314adad92e 100644 --- a/crates/buzz-relay/src/lib.rs +++ b/crates/buzz-relay/src/lib.rs @@ -37,6 +37,7 @@ pub mod push_runtime; pub mod router; /// Shared application state. pub mod state; +pub mod storage_sweep; /// Subscription registry with (channel, kind) fan-out index. pub mod subscription; /// OpenTelemetry tracing initialisation (tracer provider + OTLP exporter). diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 744633baa9..019a2e613e 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -17,6 +17,7 @@ use buzz_relay::config::Config; use buzz_relay::metrics as relay_metrics; use buzz_relay::router::{build_health_router, build_router}; use buzz_relay::state::AppState; +use buzz_relay::storage_sweep; use buzz_relay::telemetry; use buzz_workflow::WorkflowEngine; use tokio_util::sync::CancellationToken; @@ -1342,11 +1343,53 @@ async fn run_usage_metrics_tick( *leader = None; return Err(error); } + run_storage_sweep_tick(state, emission_scope, &host_map).await; } Ok(()) } +/// Storage-sweep half of the leader-only tick: harvest/spawn (never awaits +/// the sweep itself) then re-emit whatever snapshot is cached. Split out of +/// [`run_usage_metrics_tick`] because it has its own always-on config +/// (independent of `EmissionScope`) and a hard kill switch — a disabled +/// sweep must never touch a single storage-family gauge, including the +/// health ones, so a relay without `s3:ListBucket` can turn the whole +/// feature off cleanly. +async fn run_storage_sweep_tick( + state: &AppState, + emission_scope: &EmissionScope, + host_map: &HashMap, +) { + static SWEEP_CONFIG: std::sync::OnceLock = + std::sync::OnceLock::new(); + let config = *SWEEP_CONFIG.get_or_init(storage_sweep::StorageSweepConfig::from_env); + if !config.enabled { + return; + } + + 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| { + let media_storage = Arc::clone(&media_storage); + async move { media_storage.list_page(token, 1000).await } + }) + .await + }, + ) + .await; + + storage_sweep::emit_storage_metrics(&state.storage_sweep, host_map, |id| { + emission_scope.allows(id) + }) + .await; +} + /// Emit the database-derived usage snapshot from the stable leader only. async fn emit_db_usage_metrics( state: &AppState, diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index bc41cdcc97..6dcfcd9a24 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -502,6 +502,10 @@ pub struct AppState { pub audit_tx: mpsc::Sender, /// Media storage client (S3/MinIO). pub media_storage: Arc, + /// Single-flight + cache state for the hourly S3 storage sweep. See + /// `storage_sweep` module docs; shared with the usage-metrics tick via + /// `Arc` the same way other cross-tick poller state lives on `AppState`. + pub storage_sweep: Arc>, /// Git object-store backend (content-addressed packs/manifests plus /// CAS-guarded manifest pointer). This is the durable git source of truth; /// see `api::git::store` and `docs/git-on-object-storage.md`. @@ -680,6 +684,9 @@ impl AppState { ), audit_tx, media_storage: Arc::new(media_storage), + storage_sweep: Arc::new(tokio::sync::Mutex::new( + crate::storage_sweep::StorageSweepState::default(), + )), git_store, audio_rooms: Arc::new(AudioRoomManager::new()), shutting_down: Arc::new(AtomicBool::new(false)), diff --git a/crates/buzz-relay/src/storage_sweep.rs b/crates/buzz-relay/src/storage_sweep.rs new file mode 100644 index 0000000000..badae8dbae --- /dev/null +++ b/crates/buzz-relay/src/storage_sweep.rs @@ -0,0 +1,703 @@ +//! Hourly S3 storage sweep: single-flight background task, cached snapshot, +//! re-emitted every usage-metrics tick. +//! +//! See `PLANS/S3_STORAGE_METRICS_PLAN.md` (Rev 3) "Sweep architecture" for +//! the full design; this module implements the relay-side half (the +//! classifier + pure fold live in `buzz_media::bucket_index`). Summary: +//! +//! - The usage tick never awaits the sweep. [`maybe_spawn_sweep`] harvests +//! any finished in-flight attempt into the cache, then spawns at most one +//! new attempt (single-flight) when the cadence/failure rule allows it. +//! - [`emit_storage_metrics`] is called every tick, leader-only, and +//! re-publishes the cached snapshot regardless of whether a sweep is +//! currently running — DB-derived gauges keep their configured cadence, +//! storage gauges lag by at most one tick after a sweep completes. +//! - A cold cache (no sweep has ever succeeded) publishes health gauges only +//! (`sweep_ok=0`); a warm cache re-publishes the last good snapshot even +//! while the newest attempt is failing, so a transient S3 blip never blanks +//! the dashboards. + +use std::collections::HashMap; +use std::future::Future; +use std::time::{Duration, Instant}; + +use tokio::sync::Mutex; +use tokio::task::JoinHandle; +use uuid::Uuid; + +use buzz_media::{BucketSnapshot, SweepError}; + +/// Sweep knobs, read once at boot. See `PLANS/S3_STORAGE_METRICS_PLAN.md` F7. +#[derive(Debug, Clone, Copy)] +pub struct StorageSweepConfig { + /// Minimum time between successful sweeps. Floored so a misconfigured + /// value can't turn this into a listing busy-loop. + pub interval: Duration, + /// `tokio::time::timeout` around one whole sweep attempt. Cadence- + /// 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. + pub enabled: bool, +} + +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_METRICS` kill switch (`off` ⇒ disabled, anything else + /// including unset ⇒ enabled). + pub fn from_env() -> Self { + let interval_secs = std::env::var("BUZZ_STORAGE_SWEEP_INTERVAL_SECS") + .ok() + .and_then(|v| v.parse::().ok()) + .unwrap_or(3600) + .max(60); + let timeout_secs = std::env::var("BUZZ_STORAGE_SWEEP_TIMEOUT_SECS") + .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()) + .as_deref() + != Some("off"); + Self { + interval: Duration::from_secs(interval_secs), + timeout: Duration::from_secs(timeout_secs), + max_objects, + enabled, + } + } +} + +/// The last completed sweep attempt's outcome and timing — kept regardless +/// of success/failure so health gauges can report on the newest attempt even +/// when [`CachedSnapshot`] still holds an older successful one. +#[derive(Debug, Clone, Copy)] +struct LastAttempt { + ok: bool, + duration: Duration, +} + +/// The most recent *successful* sweep, cut as one coherent snapshot. +#[derive(Debug, Clone)] +struct CachedSnapshot { + data: BucketSnapshot, + completed_at: Instant, +} + +/// What a spawned sweep task hands back to the tick that harvests it. +struct SweepAttempt { + result: Result, + duration: Duration, +} + +/// Single-flight + cache state for the storage sweep. One instance lives in +/// `AppState`, shared behind a `Mutex` — the same pattern this codebase uses +/// for other cross-tick poller state (e.g. the audit worker's +/// `JoinHandle` in `state::AuditShutdownHandle`). +#[derive(Default)] +pub struct StorageSweepState { + in_flight: Option>, + cached: Option, + last_attempt: Option, + failures_total: u64, +} + +/// Single-flight + cadence rule (F5/F5-bis): spawn a new sweep iff no sweep +/// is in flight AND (cold cache, OR the last attempt failed — respawn on the +/// very next tick rather than waiting a full interval, OR the cached +/// snapshot is older than `interval`). +fn should_spawn( + cached: &Option, + last_attempt: &Option, + interval: Duration, + now: Instant, +) -> bool { + match last_attempt { + None => true, + Some(attempt) if !attempt.ok => true, + Some(_) => match cached { + // A prior successful attempt without a cached snapshot can't + // happen through this module's own harvest path, but treat it + // as cold rather than panicking on a broken invariant. + None => true, + Some(snapshot) => now.duration_since(snapshot.completed_at) >= interval, + }, + } +} + +/// Harvest any finished in-flight attempt into the cache, then spawn a new +/// attempt if the single-flight + cadence rule allows it. +/// +/// Harvest and spawn share one lock acquisition and one "is there a live +/// handle" check by design — splitting them would open a race window where +/// a tick sees a freshly-emptied `in_flight` slot from a harvest that hasn't +/// yet updated `cached`/`last_attempt`, and spawns a redundant second sweep. +/// +/// `sweep_fut` is constructed by the caller on every tick but only ever +/// polled if this call decides to spawn — an unpolled async value has not +/// started its body, so building it speculatively has no side effects. +pub async fn maybe_spawn_sweep( + state: &Mutex, + interval: Duration, + timeout: Duration, + sweep_fut: Fut, +) where + Fut: Future> + Send + 'static, +{ + let mut state = state.lock().await; + + if let Some(handle) = state.in_flight.take() { + if !handle.is_finished() { + state.in_flight = Some(handle); + return; // single-flight: a sweep is already running + } + match handle.await { + Ok(attempt) => { + let ok = attempt.result.is_ok(); + if let Ok(snapshot) = attempt.result { + state.cached = Some(CachedSnapshot { + data: snapshot, + completed_at: Instant::now(), + }); + } else if !ok { + // Warm-cache-on-failure (F5): keep the previous + // successful snapshot; only the health gauges regress. + } + if !ok { + state.failures_total += 1; + } + state.last_attempt = Some(LastAttempt { + ok, + duration: attempt.duration, + }); + } + Err(join_error) => { + tracing::error!(error = %join_error, "storage sweep task panicked"); + state.failures_total += 1; + state.last_attempt = Some(LastAttempt { + ok: false, + duration: Duration::ZERO, + }); + } + } + } + + if !should_spawn(&state.cached, &state.last_attempt, interval, Instant::now()) { + return; + } + + let handle = tokio::spawn(async move { + let started = Instant::now(); + let result = tokio::time::timeout(timeout, sweep_fut) + .await + .unwrap_or(Err(SweepError::Timeout(timeout))); + SweepAttempt { + result, + duration: started.elapsed(), + } + }); + state.in_flight = Some(handle); +} + +/// Emit the storage-family gauges from the cached snapshot. Call every usage +/// tick, leader-only (mirrors `emit_db_usage_metrics`'s leadership gate) — +/// never from the spawned sweep task itself, so a sweep that completes after +/// this pod loses leadership parks its snapshot without ever publishing it. +/// +/// `host_map` resolves a community UUID to its label string for per- +/// community series; `allows` gates those series the same way +/// `EmissionScope` gates the DB-derived ones. A bound community UUID absent +/// from `host_map` is "unmapped" (sidecar references a community with no DB +/// row) and rolls into `buzz_storage_unmapped_community_bytes` instead of a +/// per-community series. +pub async fn emit_storage_metrics( + state: &Mutex, + host_map: &HashMap, + allows: impl Fn(&Uuid) -> bool, +) { + let state = state.lock().await; + + let ok = state.last_attempt.is_some_and(|a| a.ok); + metrics::gauge!("buzz_storage_sweep_ok").set(if ok { 1.0 } else { 0.0 }); + metrics::gauge!("buzz_storage_sweep_failures_total").set(state.failures_total as f64); + if let Some(attempt) = state.last_attempt { + metrics::gauge!("buzz_storage_sweep_duration_seconds").set(attempt.duration.as_secs_f64()); + } + + // Cold cache + failure (F5): no storage-family/per-community gauges yet. + let Some(cached) = &state.cached else { + return; + }; + metrics::gauge!("buzz_storage_sweep_age_seconds") + .set(cached.completed_at.elapsed().as_secs_f64()); + + let snapshot = &cached.data; + metrics::gauge!("buzz_total_storage_bytes", "kind" => "physical") + .set(snapshot.physical_bytes as f64); + metrics::gauge!("buzz_total_storage_objects", "kind" => "physical") + .set(snapshot.physical_objects as f64); + metrics::gauge!("buzz_total_storage_bytes", "kind" => "logical") + .set(snapshot.logical_bytes as f64); + metrics::gauge!("buzz_total_storage_objects", "kind" => "logical") + .set(snapshot.logical_objects as f64); + + metrics::gauge!("buzz_storage_orphan_blob_bytes").set(snapshot.orphan_blob_bytes as f64); + metrics::gauge!("buzz_storage_orphan_blobs").set(snapshot.orphan_blob_count as f64); + metrics::gauge!("buzz_storage_orphan_sidecars").set(snapshot.orphan_sidecar_count as f64); + metrics::gauge!("buzz_storage_multi_variant_shas").set(snapshot.multi_variant_shas as f64); + metrics::gauge!("buzz_storage_multi_variant_bytes").set(snapshot.multi_variant_bytes as f64); + metrics::gauge!("buzz_storage_unknown_key_bytes").set(snapshot.unknown_key_bytes as f64); + metrics::gauge!("buzz_storage_unknown_key_objects").set(snapshot.unknown_key_objects as f64); + + let mut unmapped_bytes = 0u64; + for (community_id, storage) in &snapshot.per_community { + let Some(host) = host_map.get(community_id) else { + unmapped_bytes += storage.bytes; + continue; + }; + if !allows(community_id) { + continue; + } + metrics::gauge!("buzz_community_storage_bytes", "community" => host.clone()) + .set(storage.bytes as f64); + metrics::gauge!("buzz_community_storage_objects", "community" => host.clone()) + .set(storage.objects as f64); + } + metrics::gauge!("buzz_storage_unmapped_community_bytes").set(unmapped_bytes as f64); +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::Arc; + + use buzz_media::CommunityStorage; + use metrics_util::debugging::{DebugValue, DebuggingRecorder}; + + fn snapshot_with(community: Uuid, bytes: u64, objects: u64) -> BucketSnapshot { + let mut per_community = HashMap::new(); + per_community.insert(community, CommunityStorage { bytes, objects }); + BucketSnapshot { + physical_bytes: bytes, + physical_objects: objects, + logical_bytes: bytes, + logical_objects: objects, + per_community, + ..Default::default() + } + } + + // --- StorageSweepConfig::from_env --- + + #[test] + fn config_defaults_and_floors_apply_when_env_absent() { + // No env manipulation: absent vars in the test process must resolve + // to the documented defaults. + 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() { + return; // externally forced — skip rather than assert a lie + } + } + 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); + } + + #[test] + fn config_kill_switch_only_off_disables() { + assert!(!parse_enabled(Some("off"))); + assert!(!parse_enabled(Some("OFF"))); + assert!(parse_enabled(Some("on"))); + assert!(parse_enabled(Some("anything-else"))); + assert!(parse_enabled(None)); + } + + fn parse_enabled(value: Option<&str>) -> bool { + value.map(str::trim).map(str::to_ascii_lowercase).as_deref() != Some("off") + } + + // --- should_spawn --- + + #[test] + fn should_spawn_cold_cache_and_no_attempt_yet() { + assert!(should_spawn( + &None, + &None, + Duration::from_secs(3600), + Instant::now() + )); + } + + #[test] + fn should_spawn_respawns_immediately_after_a_failed_attempt() { + let last_attempt = Some(LastAttempt { + ok: false, + duration: Duration::from_secs(1), + }); + // Cached snapshot from an earlier success is still warm and fresh, + // but the failed attempt alone must force an immediate respawn. + let cached = Some(CachedSnapshot { + data: BucketSnapshot::default(), + completed_at: Instant::now(), + }); + assert!(should_spawn( + &cached, + &last_attempt, + Duration::from_secs(3600), + Instant::now() + )); + } + + #[test] + fn should_spawn_waits_out_the_interval_after_a_success() { + let now = Instant::now(); + let last_attempt = Some(LastAttempt { + ok: true, + duration: Duration::from_secs(1), + }); + let cached = Some(CachedSnapshot { + data: BucketSnapshot::default(), + completed_at: now, + }); + assert!(!should_spawn( + &cached, + &last_attempt, + Duration::from_secs(3600), + now + )); + assert!(should_spawn( + &cached, + &last_attempt, + Duration::from_secs(3600), + now + Duration::from_secs(3600) + )); + } + + // --- maybe_spawn_sweep --- + + #[tokio::test] + async fn cold_cache_success_populates_the_cache() { + let state = Mutex::new(StorageSweepState::default()); + maybe_spawn_sweep( + &state, + Duration::from_secs(3600), + Duration::from_secs(5), + async { Ok(snapshot_with(Uuid::new_v4(), 10, 1)) }, + ) + .await; + // The spawned task needs a scheduling point to run before the next + // call can harvest it. + tokio::task::yield_now().await; + maybe_spawn_sweep( + &state, + Duration::from_secs(3600), + Duration::from_secs(5), + async { panic!("must not spawn a second attempt while nothing changed") }, + ) + .await; + + let guard = state.lock().await; + assert!(guard.cached.is_some()); + assert!(guard.last_attempt.unwrap().ok); + assert_eq!(guard.failures_total, 0); + } + + #[tokio::test] + async fn cold_cache_failure_leaves_no_snapshot_but_records_the_failure() { + let state = Mutex::new(StorageSweepState::default()); + // Each call harvests the PREVIOUS call's spawned attempt (harvest + // and spawn share one lock acquisition — see `maybe_spawn_sweep` + // doc comment), so two failures need three calls: the first spawns + // attempt 1, the second harvests attempt 1 and spawns attempt 2, + // the third harvests attempt 2. + for _ in 0..3 { + maybe_spawn_sweep( + &state, + Duration::from_secs(3600), + Duration::from_secs(5), + async { Err(SweepError::CapExceeded { seen: 5, cap: 1 }) }, + ) + .await; + tokio::task::yield_now().await; + } + + let guard = state.lock().await; + assert!(guard.cached.is_none(), "cold cache stays cold on failure"); + assert_eq!(guard.failures_total, 2); + assert!(!guard.last_attempt.unwrap().ok); + } + + #[tokio::test] + async fn warm_cache_failure_keeps_the_old_snapshot() { + let community = Uuid::new_v4(); + let state = Mutex::new(StorageSweepState::default()); + maybe_spawn_sweep( + &state, + Duration::from_secs(3600), + Duration::from_secs(5), + async move { Ok(snapshot_with(community, 42, 1)) }, + ) + .await; + tokio::task::yield_now().await; + // This call harvests the success above, then (last_attempt.ok=true, + // cache fresh) does NOT spawn again. + maybe_spawn_sweep( + &state, + Duration::from_secs(3600), + Duration::from_secs(5), + async { panic!("must not spawn: cache is warm and fresh") }, + ) + .await; + { + let guard = state.lock().await; + assert_eq!( + guard.cached.as_ref().unwrap().data.per_community[&community].bytes, + 42 + ); + } + + // Force a respawn by aging the cache past the interval, then fail it. + { + let mut guard = state.lock().await; + guard.cached.as_mut().unwrap().completed_at = + Instant::now() - Duration::from_secs(7200); + } + maybe_spawn_sweep( + &state, + Duration::from_secs(3600), + Duration::from_secs(5), + async { Err(SweepError::CapExceeded { seen: 5, cap: 1 }) }, + ) + .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 }) }, + ) + .await; + + let guard = state.lock().await; + assert_eq!( + guard.cached.as_ref().unwrap().data.per_community[&community].bytes, + 42, + "old snapshot must survive a later failed attempt" + ); + assert!(!guard.last_attempt.unwrap().ok); + assert_eq!(guard.failures_total, 1); + } + + #[tokio::test] + async fn single_flight_never_spawns_a_second_attempt_while_one_is_running() { + let started = Arc::new(AtomicUsize::new(0)); + let gate = Arc::new(tokio::sync::Notify::new()); + let state = Mutex::new(StorageSweepState::default()); + + let started_for_first = Arc::clone(&started); + let gate_for_first = Arc::clone(&gate); + maybe_spawn_sweep( + &state, + Duration::from_secs(3600), + Duration::from_secs(5), + async move { + started_for_first.fetch_add(1, Ordering::SeqCst); + gate_for_first.notified().await; + Ok(BucketSnapshot::default()) + }, + ) + .await; + tokio::task::yield_now().await; + + let started_for_second = Arc::clone(&started); + maybe_spawn_sweep( + &state, + Duration::from_secs(3600), + Duration::from_secs(5), + async move { + started_for_second.fetch_add(1, Ordering::SeqCst); + Ok(BucketSnapshot::default()) + }, + ) + .await; + + assert_eq!( + started.load(Ordering::SeqCst), + 1, + "second attempt must never have been polled while the first was in flight" + ); + gate.notify_one(); // release the first attempt so the test can end cleanly + } + + #[tokio::test(start_paused = true)] + async fn a_stalled_attempt_times_out_and_is_recorded_as_a_failure() { + let state = Mutex::new(StorageSweepState::default()); + maybe_spawn_sweep( + &state, + Duration::from_secs(3600), + Duration::from_secs(5), + async { + tokio::time::sleep(Duration::from_secs(3600)).await; + Ok(BucketSnapshot::default()) + }, + ) + .await; + + // The spawned task must be polled at least once to register its + // inner `tokio::time::timeout` deadline before the paused clock can + // be advanced past it — `advance` only fires timers that already + // exist. + tokio::task::yield_now().await; + tokio::time::advance(Duration::from_secs(6)).await; + // `advance` fires the deadline, but driving the woken task to + // completion still needs the executor to actually poll it — loop + // `yield_now` until the handle reports finished rather than + // guessing a fixed poll count. + for _ in 0..50 { + let finished = state + .lock() + .await + .in_flight + .as_ref() + .is_none_or(JoinHandle::is_finished); + if finished { + break; + } + tokio::task::yield_now().await; + } + // Harvest the timed-out attempt. + maybe_spawn_sweep( + &state, + Duration::from_secs(3600), + Duration::from_secs(5), + async { panic!("cadence not yet due — must not spawn") }, + ) + .await; + + let guard = state.lock().await; + assert_eq!(guard.failures_total, 1); + assert!(!guard.last_attempt.unwrap().ok); + assert!(guard.cached.is_none()); + } + + // --- emit_storage_metrics --- + + fn gauge_snapshot(recorder: &DebuggingRecorder) -> std::collections::HashMap { + recorder + .snapshotter() + .snapshot() + .into_vec() + .into_iter() + .filter_map(|(key, _, _, value)| match value { + DebugValue::Gauge(v) => Some((key.key().name().to_owned(), v.into_inner())), + _ => None, + }) + .collect() + } + + #[tokio::test] + async fn cold_cache_emits_only_health_gauges() { + let state = Mutex::new(StorageSweepState::default()); + let recorder = DebuggingRecorder::new(); + let host_map = HashMap::new(); + + metrics::with_local_recorder(&recorder, || { + futures::executor::block_on(emit_storage_metrics(&state, &host_map, |_| true)); + }); + + let values = gauge_snapshot(&recorder); + assert_eq!(values.get("buzz_storage_sweep_ok"), Some(&0.0)); + assert_eq!(values.get("buzz_storage_sweep_failures_total"), Some(&0.0)); + assert!(!values.contains_key("buzz_total_storage_bytes")); + assert!(!values.contains_key("buzz_storage_sweep_age_seconds")); + } + + #[tokio::test] + async fn warm_cache_emits_community_and_unmapped_totals_with_scope_gating() { + let mapped = Uuid::new_v4(); + let unmapped = Uuid::new_v4(); + let excluded = Uuid::new_v4(); + let mut per_community = HashMap::new(); + per_community.insert( + mapped, + CommunityStorage { + bytes: 100, + objects: 2, + }, + ); + per_community.insert( + unmapped, + CommunityStorage { + bytes: 30, + objects: 1, + }, + ); + per_community.insert( + excluded, + CommunityStorage { + bytes: 7, + objects: 1, + }, + ); + let snapshot = BucketSnapshot { + physical_bytes: 137, + physical_objects: 4, + logical_bytes: 137, + logical_objects: 4, + per_community, + ..Default::default() + }; + + let state = Mutex::new(StorageSweepState { + cached: Some(CachedSnapshot { + data: snapshot, + completed_at: Instant::now(), + }), + last_attempt: Some(LastAttempt { + ok: true, + duration: Duration::from_millis(500), + }), + ..Default::default() + }); + + let mut host_map = HashMap::new(); + host_map.insert(mapped, "mapped.example".to_string()); + host_map.insert(excluded, "excluded.example".to_string()); + + let recorder = DebuggingRecorder::new(); + metrics::with_local_recorder(&recorder, || { + futures::executor::block_on(emit_storage_metrics(&state, &host_map, |id| { + *id != excluded + })); + }); + + let values = gauge_snapshot(&recorder); + assert_eq!(values.get("buzz_storage_sweep_ok"), Some(&1.0)); + assert_eq!(values.get("buzz_total_storage_bytes"), Some(&137.0)); + assert_eq!( + values.get("buzz_storage_unmapped_community_bytes"), + Some(&30.0) + ); + assert_eq!(values.get("buzz_community_storage_bytes"), Some(&100.0)); + } +} diff --git a/deploy/charts/buzz/values.yaml b/deploy/charts/buzz/values.yaml index 9edb534f1b..5a14e7dc3c 100644 --- a/deploy/charts/buzz/values.yaml +++ b/deploy/charts/buzz/values.yaml @@ -272,6 +272,14 @@ externalRedis: # Quickstart (`minio.enabled: true`): the chart runs an in-cluster, eval-only # MinIO Deployment, creates the bucket via a post-install Job, and composes # the endpoint + autogenerated credentials automatically. +# +# Storage metrics (hourly bucket sweep, BUZZ_STORAGE_METRICS — see env docs): +# the credentials above must additionally grant `s3:ListBucket` on the bucket +# ARN itself (bucket-level; distinct from the object-level GetObject/ +# PutObject/DeleteObject perms already required for media). Without it the +# first sweep fails AccessDenied and buzz_storage_sweep_ok stays 0 — no other +# media functionality is affected. Set BUZZ_STORAGE_METRICS=off to disable +# the sweep entirely on a deployment that can't grant it. s3: endpoint: "" bucket: "buzz-media" From 15af8439ddbcaad07712e5a7b772e4f2cad5ed97 Mon Sep 17 00:00:00 2001 From: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 Date: Fri, 17 Jul 2026 18:06:32 -0500 Subject: [PATCH 2/5] test(buzz-relay): add Rev 3's demoted-leader and stalled-sweep-cadence tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Rev 3's Tasks section required a demoted-leader-never-emits test and a paused-time tick-cadence composite; neither existed in storage_sweep.rs despite an earlier self-report listing 'demoted-leader-never-emits' among the module's 12 tests. Adds two module-level tests that exercise the property this module owns without a DB harness: a completed-but-unharvested sweep attempt (the demoted-leader case, since harvest only happens inside maybe_spawn_sweep, which the relay only calls from the leader-only tick branch) emits zero storage gauges, and a stalled sweep across several simulated ticks emits health-only gauges on every tick, never double-spawns, then emits the real snapshot once on the first tick after completion. The DB-leader-transition half of the composite (no leader demotion across ticks) needs a live Postgres advisory lock via buzz_db::UsageMetricsLeader, which has no fixture-free construction — that half is documented as integration territory, matching the existing #[ignore = "requires Postgres"] convention used elsewhere in this crate for the same reason. Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- crates/buzz-relay/src/storage_sweep.rs | 114 +++++++++++++++++++++++++ 1 file changed, 114 insertions(+) diff --git a/crates/buzz-relay/src/storage_sweep.rs b/crates/buzz-relay/src/storage_sweep.rs index badae8dbae..adef501168 100644 --- a/crates/buzz-relay/src/storage_sweep.rs +++ b/crates/buzz-relay/src/storage_sweep.rs @@ -600,6 +600,120 @@ mod tests { assert!(guard.cached.is_none()); } + // --- Rev 3 required tests: demoted-leader-never-emits + paused-time + // tick-cadence composite --- + // + // Harvest only happens inside `maybe_spawn_sweep`, which the relay only + // calls from the leader-only branch of the usage-metrics tick + // (`run_usage_metrics_tick`, gated on `leader.is_some()`). A pod that + // loses leadership stops calling both `maybe_spawn_sweep` and + // `emit_storage_metrics` — so a sweep that finishes after demotion sits + // in `in_flight` forever un-harvested and its data is never published. + // That leader-transition itself lives behind a real Postgres advisory + // lock (`buzz_db::UsageMetricsLeader`, see `crates/buzz-db/src/lib.rs`) + // and has no fixture-free construction, so it can't be driven from this + // module; the property this module DOES own — a completed-but- + // unharvested attempt emits nothing — is what's covered below. + + #[tokio::test] + async fn a_completed_but_unharvested_sweep_never_emits_its_snapshot() { + let community = Uuid::new_v4(); + let state = Mutex::new(StorageSweepState::default()); + maybe_spawn_sweep( + &state, + Duration::from_secs(3600), + Duration::from_secs(5), + async move { Ok(snapshot_with(community, 999, 1)) }, + ) + .await; + // The spawned task finishes here, but nothing calls + // `maybe_spawn_sweep` again to harvest it — the demoted-leader case. + tokio::task::yield_now().await; + + let recorder = DebuggingRecorder::new(); + let host_map = HashMap::new(); + metrics::with_local_recorder(&recorder, || { + futures::executor::block_on(emit_storage_metrics(&state, &host_map, |_| true)); + }); + + let values = gauge_snapshot(&recorder); + assert_eq!( + values.get("buzz_storage_sweep_ok"), + Some(&0.0), + "unharvested attempt must not register as a success" + ); + assert!(!values.contains_key("buzz_total_storage_bytes")); + assert!(!values.contains_key("buzz_community_storage_bytes")); + assert!(!values.contains_key("buzz_storage_sweep_age_seconds")); + } + + #[tokio::test(start_paused = true)] + async fn stalled_sweep_across_several_ticks_emits_health_only_then_once_on_completion() { + // Rev 3's composite covers two halves: (a) DB metrics emit every + // tick with no leader demotion — that's `run_usage_metrics_tick` + + // a live leader lock, integration territory, not exercised here — + // and (b) the storage-sweep half this test drives directly: a sweep + // stalled across several simulated ticks never double-spawns, emits + // health-only gauges while stalled, then emits the real snapshot on + // the first tick after it completes. + let community = Uuid::new_v4(); + let gate = Arc::new(tokio::sync::Notify::new()); + let gate_for_sweep = Arc::clone(&gate); + let state = Mutex::new(StorageSweepState::default()); + + maybe_spawn_sweep( + &state, + Duration::from_secs(3600), + Duration::from_secs(120), + async move { + gate_for_sweep.notified().await; + Ok(snapshot_with(community, 500, 3)) + }, + ) + .await; + tokio::task::yield_now().await; + + for _ in 0..3 { + tokio::time::advance(Duration::from_secs(5)).await; + maybe_spawn_sweep( + &state, + Duration::from_secs(3600), + Duration::from_secs(120), + async { panic!("single-flight: must not spawn while one is in flight") }, + ) + .await; + + let recorder = DebuggingRecorder::new(); + let host_map = HashMap::new(); + metrics::with_local_recorder(&recorder, || { + futures::executor::block_on(emit_storage_metrics(&state, &host_map, |_| true)); + }); + let values = gauge_snapshot(&recorder); + assert_eq!(values.get("buzz_storage_sweep_ok"), Some(&0.0)); + assert!(!values.contains_key("buzz_total_storage_bytes")); + } + + // Let the stalled sweep complete; the next tick harvests it. + gate.notify_one(); + tokio::task::yield_now().await; + maybe_spawn_sweep( + &state, + Duration::from_secs(3600), + Duration::from_secs(120), + async { panic!("cadence not yet due — must not spawn a second attempt") }, + ) + .await; + + let recorder = DebuggingRecorder::new(); + let host_map = HashMap::new(); + metrics::with_local_recorder(&recorder, || { + futures::executor::block_on(emit_storage_metrics(&state, &host_map, |_| true)); + }); + let values = gauge_snapshot(&recorder); + assert_eq!(values.get("buzz_storage_sweep_ok"), Some(&1.0)); + assert_eq!(values.get("buzz_total_storage_bytes"), Some(&500.0)); + } + // --- emit_storage_metrics --- fn gauge_snapshot(recorder: &DebuggingRecorder) -> std::collections::HashMap { From d0b269efac28de8da9d089ccd599d94852cebb41 Mon Sep 17 00:00:00 2001 From: Will Pfleger Date: Mon, 20 Jul 2026 12:04:58 -0400 Subject: [PATCH 3/5] fix(media,relay): log failed sweep diagnostic and reject malformed truncated pages Failed sweep attempts discarded the SweepError without logging, making AccessDenied from a missing s3:ListBucket grant invisible except via a gauge. A truncated listing page with no continuation token silently returned a partial snapshot instead of failing. --- crates/buzz-media/src/bucket_index.rs | 24 ++++++++++++++++++--- crates/buzz-relay/src/storage_sweep.rs | 29 ++++++++++++++++---------- deploy/charts/buzz/values.yaml | 2 ++ 3 files changed, 41 insertions(+), 14 deletions(-) diff --git a/crates/buzz-media/src/bucket_index.rs b/crates/buzz-media/src/bucket_index.rs index cf92291b58..bb83dc517f 100644 --- a/crates/buzz-media/src/bucket_index.rs +++ b/crates/buzz-media/src/bucket_index.rs @@ -352,6 +352,10 @@ pub enum SweepError { /// currency the whole sweep pipeline reasons about. #[error("sweep timed out after {0:?}")] Timeout(std::time::Duration), + /// A listing page reported `is_truncated=true` but supplied no + /// continuation token — a malformed S3 response that cannot be resumed. + #[error("truncated listing page with no continuation token")] + MalformedPage, } /// One page of a bucket listing, decoupled from any S3 crate type so the @@ -399,9 +403,7 @@ where } match page.next_continuation_token { Some(token) => continuation_token = Some(token), - // Defensive: truncated with no token to resume from is a - // malformed response, not an infinite-loop invitation. - None => break, + None => return Err(SweepError::MalformedPage), } } @@ -734,4 +736,20 @@ mod tests { .await; assert!(matches!(result, Err(SweepError::Storage(_)))); } + + #[tokio::test] + async fn truncated_page_with_no_continuation_token_fails_the_sweep() { + let result = fold_bucket_listing(100, |_token| async { + Ok(Page { + objects: vec![("some-key".to_string(), 1)], + next_continuation_token: None, + is_truncated: true, + }) + }) + .await; + assert!( + matches!(result, Err(SweepError::MalformedPage)), + "truncated page without a continuation token must fail, not return partial data" + ); + } } diff --git a/crates/buzz-relay/src/storage_sweep.rs b/crates/buzz-relay/src/storage_sweep.rs index adef501168..64db96dac2 100644 --- a/crates/buzz-relay/src/storage_sweep.rs +++ b/crates/buzz-relay/src/storage_sweep.rs @@ -166,17 +166,23 @@ pub async fn maybe_spawn_sweep( match handle.await { Ok(attempt) => { let ok = attempt.result.is_ok(); - if let Ok(snapshot) = attempt.result { - state.cached = Some(CachedSnapshot { - data: snapshot, - completed_at: Instant::now(), - }); - } else if !ok { - // Warm-cache-on-failure (F5): keep the previous - // successful snapshot; only the health gauges regress. - } - if !ok { - state.failures_total += 1; + match attempt.result { + Ok(snapshot) => { + state.cached = Some(CachedSnapshot { + data: snapshot, + // Stamped at harvest, not sweep completion — exported + // age/cadence may lag by ≤1 usage tick. + completed_at: Instant::now(), + }); + } + Err(err) => { + tracing::error!( + error = %err, + "storage sweep failed; verify s3:ListBucket \ + (or MinIO list) permission is granted on the bucket" + ); + state.failures_total += 1; + } } state.last_attempt = Some(LastAttempt { ok, @@ -231,6 +237,7 @@ pub async fn emit_storage_metrics( let ok = state.last_attempt.is_some_and(|a| a.ok); metrics::gauge!("buzz_storage_sweep_ok").set(if ok { 1.0 } else { 0.0 }); + // Process-local gauge: resets/jumps on leader failover — not a global counter. metrics::gauge!("buzz_storage_sweep_failures_total").set(state.failures_total as f64); if let Some(attempt) = state.last_attempt { metrics::gauge!("buzz_storage_sweep_duration_seconds").set(attempt.duration.as_secs_f64()); diff --git a/deploy/charts/buzz/values.yaml b/deploy/charts/buzz/values.yaml index 5a14e7dc3c..5bdc6a1489 100644 --- a/deploy/charts/buzz/values.yaml +++ b/deploy/charts/buzz/values.yaml @@ -280,6 +280,8 @@ externalRedis: # first sweep fails AccessDenied and buzz_storage_sweep_ok stays 0 — no other # media functionality is affected. Set BUZZ_STORAGE_METRICS=off to disable # the sweep entirely on a deployment that can't grant it. +# Note: buzz_storage_sweep_failures_total is a process-local gauge — on leader +# failover it resets to the new leader's local count, not a global total. s3: endpoint: "" bucket: "buzz-media" From 3e08943156ab261b49ecfb25e53c5306e4a4b933 Mon Sep 17 00:00:00 2001 From: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 Date: Tue, 21 Jul 2026 16:49:31 -0700 Subject: [PATCH 4/5] fix(storage-metrics): retire stale community series, rename failures gauge, add comments MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit F-EXT1: emit_storage_metrics now tracks per-community series emitted in the previous tick via StorageEmittedKey{Bytes,Objects}(host). Any series whose community disappears (unmapped, host rename, scope exclusion) is explicitly zeroed on the next emission rather than lingering at its last nonzero value until recorder idle-eviction. Mirrors the emit_in_memory_usage_metrics pattern in main.rs. Adds regression test covering all three disappearance scenarios in a single two-emission sequence. F-EXT2: rename buzz_storage_sweep_failures_total → buzz_storage_sweep_failures to avoid the _total counter-convention confusion (nothing has shipped, so rename is safe per Will's forward-only ruling). Adds an inline comment explaining the gauge/counter distinction. Updates values.yaml doc. F-EXT3 (doc-only): adds a comment on should_spawn explaining that a failed attempt retries at tick cadence (default 300 s), not sweep-interval cadence, and why that is intentional. Adds a matching note in values.yaml. F-EXT4: adds a comment on the SWEEP_CONFIG function-local OnceLock explaining why it lives here rather than on AppState (6-call-site constructor churn explicitly avoided; first-call init is effectively boot-time for this path). Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- crates/buzz-relay/src/main.rs | 6 + crates/buzz-relay/src/storage_sweep.rs | 272 ++++++++++++++++++++++++- deploy/charts/buzz/values.yaml | 6 +- 3 files changed, 279 insertions(+), 5 deletions(-) diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index e87d4d55ab..fab0507c4c 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -1434,6 +1434,12 @@ async fn run_storage_sweep_tick( ) { static SWEEP_CONFIG: std::sync::OnceLock = std::sync::OnceLock::new(); + // SWEEP_CONFIG is a function-local OnceLock (not on AppState) by design: + // AppState is constructed once at startup with six call sites, and adding + // a StorageSweepConfig field would require touching all of them. The + // OnceLock-on-AppState pattern was explicitly rejected to avoid that churn. + // First-call init here is equivalent to boot-time for this code path + // (leader-only tick, runs within seconds of startup). let config = *SWEEP_CONFIG.get_or_init(storage_sweep::StorageSweepConfig::from_env); if !config.enabled { return; diff --git a/crates/buzz-relay/src/storage_sweep.rs b/crates/buzz-relay/src/storage_sweep.rs index 64db96dac2..f53e02f086 100644 --- a/crates/buzz-relay/src/storage_sweep.rs +++ b/crates/buzz-relay/src/storage_sweep.rs @@ -17,7 +17,7 @@ //! while the newest attempt is failing, so a transient S3 blip never blanks //! the dashboards. -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::future::Future; use std::time::{Duration, Instant}; @@ -102,6 +102,35 @@ struct SweepAttempt { duration: Duration, } +/// Key tracking which per-community series were emitted in the previous tick, +/// so series for communities that disappear from the snapshot (unmapped, host +/// renamed, or scope-excluded) are zeroed rather than left at their last +/// nonzero value until the recorder's idle-eviction kicks in. +/// +/// Carries the resolved host label (not the UUID) so a rename can still zero +/// the old series, and distinguishes bytes vs. objects because they are +/// separate Prometheus series. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub(crate) enum StorageEmittedKey { + Bytes(String), + Objects(String), +} + +impl StorageEmittedKey { + fn set(&self, value: f64) { + match self { + Self::Bytes(host) => { + metrics::gauge!("buzz_community_storage_bytes", "community" => host.clone()) + .set(value); + } + Self::Objects(host) => { + metrics::gauge!("buzz_community_storage_objects", "community" => host.clone()) + .set(value); + } + } + } +} + /// Single-flight + cache state for the storage sweep. One instance lives in /// `AppState`, shared behind a `Mutex` — the same pattern this codebase uses /// for other cross-tick poller state (e.g. the audit worker's @@ -112,12 +141,22 @@ pub struct StorageSweepState { cached: Option, last_attempt: Option, failures_total: u64, + /// Per-community series emitted on the previous tick — used to zero series + /// whose community disappears from the snapshot (see [`emit_storage_metrics`]). + previously_emitted: HashSet, } /// Single-flight + cadence rule (F5/F5-bis): spawn a new sweep iff no sweep /// is in flight AND (cold cache, OR the last attempt failed — respawn on the /// very next tick rather than waiting a full interval, OR the cached /// snapshot is older than `interval`). +/// +/// Note: a failed attempt (`!ok`) returns `true` unconditionally, so a +/// 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: the retry is a single cheap LIST call, it means the +/// sweep self-heals the moment the permission is added, and the tick cadence +/// is documented in values.yaml. fn should_spawn( cached: &Option, last_attempt: &Option, @@ -228,23 +267,37 @@ pub async fn maybe_spawn_sweep( /// from `host_map` is "unmapped" (sidecar references a community with no DB /// row) and rolls into `buzz_storage_unmapped_community_bytes` instead of a /// per-community series. +/// +/// Per-community series whose community disappears from the current snapshot +/// (unmapped, host rename, or scope exclusion) are explicitly zeroed — the +/// same pattern as `emit_in_memory_usage_metrics`. Without this, a series +/// would linger at its last nonzero value until the recorder's idle eviction +/// fires (≥3 ticks), producing a transient double-count against the +/// `buzz_storage_unmapped_community_bytes` gauge. pub async fn emit_storage_metrics( state: &Mutex, host_map: &HashMap, allows: impl Fn(&Uuid) -> bool, ) { - let state = state.lock().await; + let mut state = state.lock().await; let ok = state.last_attempt.is_some_and(|a| a.ok); metrics::gauge!("buzz_storage_sweep_ok").set(if ok { 1.0 } else { 0.0 }); // Process-local gauge: resets/jumps on leader failover — not a global counter. - metrics::gauge!("buzz_storage_sweep_failures_total").set(state.failures_total as f64); + // Named without _total suffix to avoid confusing tools that infer counter + // semantics from the _total convention (e.g. rate() in PromQL). + metrics::gauge!("buzz_storage_sweep_failures").set(state.failures_total as f64); if let Some(attempt) = state.last_attempt { metrics::gauge!("buzz_storage_sweep_duration_seconds").set(attempt.duration.as_secs_f64()); } // Cold cache + failure (F5): no storage-family/per-community gauges yet. + // Zero any previously-emitted per-community series before returning so + // they don't linger if we had a warm cache in a prior tick. let Some(cached) = &state.cached else { + for key in state.previously_emitted.drain() { + key.set(0.0); + } return; }; metrics::gauge!("buzz_storage_sweep_age_seconds") @@ -268,6 +321,7 @@ pub async fn emit_storage_metrics( metrics::gauge!("buzz_storage_unknown_key_bytes").set(snapshot.unknown_key_bytes as f64); metrics::gauge!("buzz_storage_unknown_key_objects").set(snapshot.unknown_key_objects as f64); + let mut current = HashSet::new(); let mut unmapped_bytes = 0u64; for (community_id, storage) in &snapshot.per_community { let Some(host) = host_map.get(community_id) else { @@ -281,8 +335,23 @@ pub async fn emit_storage_metrics( .set(storage.bytes as f64); metrics::gauge!("buzz_community_storage_objects", "community" => host.clone()) .set(storage.objects as f64); + current.insert(StorageEmittedKey::Bytes(host.clone())); + current.insert(StorageEmittedKey::Objects(host.clone())); } metrics::gauge!("buzz_storage_unmapped_community_bytes").set(unmapped_bytes as f64); + + // Zero series for communities that were emitted last tick but are no longer + // present in the current snapshot (community removed, host renamed, or + // scope exclusion added). + for key in state + .previously_emitted + .difference(¤t) + .cloned() + .collect::>() + { + key.set(0.0); + } + state.previously_emitted = current; } #[cfg(test)] @@ -748,7 +817,7 @@ mod tests { let values = gauge_snapshot(&recorder); assert_eq!(values.get("buzz_storage_sweep_ok"), Some(&0.0)); - assert_eq!(values.get("buzz_storage_sweep_failures_total"), Some(&0.0)); + assert_eq!(values.get("buzz_storage_sweep_failures"), Some(&0.0)); assert!(!values.contains_key("buzz_total_storage_bytes")); assert!(!values.contains_key("buzz_storage_sweep_age_seconds")); } @@ -821,4 +890,199 @@ mod tests { ); assert_eq!(values.get("buzz_community_storage_bytes"), Some(&100.0)); } + + // --- F-EXT1 regression: stale per-community series are zeroed --- + + /// Returns a map of `(metric_name, community_label_value) -> gauge_value` + /// for gauges that carry a "community" label. Used to verify per-community + /// series are zeroed rather than left stale between emissions. + fn labeled_community_gauges( + recorder: &DebuggingRecorder, + ) -> std::collections::HashMap<(String, String), f64> { + recorder + .snapshotter() + .snapshot() + .into_vec() + .into_iter() + .filter_map(|(composite_key, _, _, value)| { + let DebugValue::Gauge(v) = value else { + return None; + }; + let key = composite_key.key(); + let community = key + .labels() + .find(|l| l.key() == "community") + .map(|l| l.value().to_owned())?; + Some(((key.name().to_owned(), community), v.into_inner())) + }) + .collect() + } + + #[tokio::test] + async fn stale_per_community_series_are_zeroed_on_disappearance() { + // Three scenarios in one state machine using a single StorageSweepState: + // (a) community disappears from snapshot (mapped → no entry), + // (b) host label rename (same UUID, different host string), + // (c) scope removal (community excluded by the `allows` predicate). + // + // After the second emission, the old series from (a), (b), and (c) + // must read 0.0, not their last nonzero value. + + let community_a = Uuid::new_v4(); // (a) will disappear from snapshot + let community_b = Uuid::new_v4(); // (b) will be renamed host.old → host.new + let community_c = Uuid::new_v4(); // (c) will be scope-excluded + + let make_snapshot = |include_a: bool, b_bytes: u64| { + let mut per_community = HashMap::new(); + if include_a { + per_community.insert( + community_a, + CommunityStorage { + bytes: 10, + objects: 1, + }, + ); + } + per_community.insert( + community_b, + CommunityStorage { + bytes: b_bytes, + objects: 2, + }, + ); + per_community.insert( + community_c, + CommunityStorage { + bytes: 7, + objects: 1, + }, + ); + BucketSnapshot { + physical_bytes: 10 + b_bytes + 7, + physical_objects: 4, + logical_bytes: 10 + b_bytes + 7, + logical_objects: 4, + per_community, + ..Default::default() + } + }; + + let state = Mutex::new(StorageSweepState { + cached: Some(CachedSnapshot { + data: make_snapshot(true, 20), + completed_at: Instant::now(), + }), + last_attempt: Some(LastAttempt { + ok: true, + duration: Duration::from_millis(100), + }), + ..Default::default() + }); + + let recorder = DebuggingRecorder::new(); + + // --- Emission 1: all three communities visible --- + let mut host_map_1 = HashMap::new(); + host_map_1.insert(community_a, "host.a".to_string()); + host_map_1.insert(community_b, "host.old".to_string()); + host_map_1.insert(community_c, "host.c".to_string()); + metrics::with_local_recorder(&recorder, || { + futures::executor::block_on(emit_storage_metrics(&state, &host_map_1, |_| true)); + }); + { + let labeled = labeled_community_gauges(&recorder); + assert_eq!( + labeled.get(&( + "buzz_community_storage_bytes".to_string(), + "host.a".to_string() + )), + Some(&10.0), + "emission 1: host.a bytes should be 10" + ); + assert_eq!( + labeled.get(&( + "buzz_community_storage_bytes".to_string(), + "host.old".to_string() + )), + Some(&20.0), + "emission 1: host.old bytes should be 20" + ); + } + + // --- Emission 2: community_a gone, community_b renamed, community_c excluded --- + { + let mut guard = state.lock().await; + guard.cached = Some(CachedSnapshot { + data: make_snapshot(false, 20), + completed_at: Instant::now(), + }); + } + let mut host_map_2 = HashMap::new(); + // community_a absent from host_map → unmapped (gone from per-community series) + host_map_2.insert(community_b, "host.new".to_string()); // renamed + host_map_2.insert(community_c, "host.c".to_string()); + metrics::with_local_recorder(&recorder, || { + futures::executor::block_on(emit_storage_metrics( + &state, + &host_map_2, + |id| *id != community_c, // (c) scope-excluded + )); + }); + + let labeled = labeled_community_gauges(&recorder); + + // (a) community_a disappeared — old host.a series must be zeroed + assert_eq!( + labeled.get(&( + "buzz_community_storage_bytes".to_string(), + "host.a".to_string() + )), + Some(&0.0), + "(a) disappeared community: host.a bytes must be zeroed" + ); + assert_eq!( + labeled.get(&( + "buzz_community_storage_objects".to_string(), + "host.a".to_string() + )), + Some(&0.0), + "(a) disappeared community: host.a objects must be zeroed" + ); + + // (b) community_b renamed host.old → host.new — old series must be zeroed + assert_eq!( + labeled.get(&( + "buzz_community_storage_bytes".to_string(), + "host.old".to_string() + )), + Some(&0.0), + "(b) host rename: host.old bytes must be zeroed" + ); + assert_eq!( + labeled.get(&( + "buzz_community_storage_bytes".to_string(), + "host.new".to_string() + )), + Some(&20.0), + "(b) host rename: host.new bytes must be 20" + ); + + // (c) community_c scope-excluded — host.c series must be zeroed + assert_eq!( + labeled.get(&( + "buzz_community_storage_bytes".to_string(), + "host.c".to_string() + )), + Some(&0.0), + "(c) scope removal: host.c bytes must be zeroed" + ); + assert_eq!( + labeled.get(&( + "buzz_community_storage_objects".to_string(), + "host.c".to_string() + )), + Some(&0.0), + "(c) scope removal: host.c objects must be zeroed" + ); + } } diff --git a/deploy/charts/buzz/values.yaml b/deploy/charts/buzz/values.yaml index be3d297fc4..180afc6746 100644 --- a/deploy/charts/buzz/values.yaml +++ b/deploy/charts/buzz/values.yaml @@ -312,8 +312,12 @@ externalRedis: # first sweep fails AccessDenied and buzz_storage_sweep_ok stays 0 — no other # media functionality is affected. Set BUZZ_STORAGE_METRICS=off to disable # the sweep entirely on a deployment that can't grant it. -# Note: buzz_storage_sweep_failures_total is a process-local gauge — on leader +# Note: buzz_storage_sweep_failures is a process-local gauge — on leader # failover it resets to the new leader's local count, not a global total. +# Note: on a failed sweep attempt, the next retry fires on the next usage tick +# (default 300 s BUZZ_USAGE_METRICS_INTERVAL_SECS), not at sweep-interval +# cadence — so a permanently missing s3:ListBucket yields one cheap LIST call +# per tick until the permission is added. s3: endpoint: "" bucket: "buzz-media" From 802538226ba989dce3cf8871af8b785a60889fc6 Mon Sep 17 00:00:00 2001 From: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 Date: Tue, 21 Jul 2026 17:12:54 -0700 Subject: [PATCH 5/5] docs(storage-metrics): correct two comment inaccuracies flagged in pass 3/3 Fix 1 (storage_sweep.rs should_spawn): the prior note claimed a failing sweep costs 'a single cheap LIST call' per retry. This is only true for immediate permission failures; timeout, cap, and malformed-page failures exhaust the sweep's full paginated walk before failing. Reword to scope the cheap-call claim to the permission case and note that other failures are bounded by the sweep's own timeout and object caps. Fix 2 (main.rs SWEEP_CONFIG OnceLock): the prior comment cited six AppState constructor call sites (there are ten), argued that adding a StorageSweepConfig field would require touching all of them (Config has a single Self initializer, so it wouldn't), and claimed first-call init runs within seconds of startup (the first tick is jittered by [0, interval) and this path is leader-only). Replace with the accurate invariant: localized feature config, read once on the first leader tick, stable for the process lifetime. Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- crates/buzz-relay/src/main.rs | 10 ++++------ crates/buzz-relay/src/storage_sweep.rs | 8 +++++--- 2 files changed, 9 insertions(+), 9 deletions(-) diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index fab0507c4c..00ef7819cb 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -1434,12 +1434,10 @@ async fn run_storage_sweep_tick( ) { static SWEEP_CONFIG: std::sync::OnceLock = std::sync::OnceLock::new(); - // SWEEP_CONFIG is a function-local OnceLock (not on AppState) by design: - // AppState is constructed once at startup with six call sites, and adding - // a StorageSweepConfig field would require touching all of them. The - // OnceLock-on-AppState pattern was explicitly rejected to avoid that churn. - // First-call init here is equivalent to boot-time for this code path - // (leader-only tick, runs within seconds of startup). + // SWEEP_CONFIG is a function-local OnceLock by design: it is localized + // feature config consumed only by this code path, read once on the first + // leader tick, and stable for the process lifetime (env is immutable). + // Keeping it here avoids widening Config/AppState for a single consumer. let config = *SWEEP_CONFIG.get_or_init(storage_sweep::StorageSweepConfig::from_env); if !config.enabled { return; diff --git a/crates/buzz-relay/src/storage_sweep.rs b/crates/buzz-relay/src/storage_sweep.rs index f53e02f086..eccadcd835 100644 --- a/crates/buzz-relay/src/storage_sweep.rs +++ b/crates/buzz-relay/src/storage_sweep.rs @@ -154,9 +154,11 @@ pub struct StorageSweepState { /// Note: a failed attempt (`!ok`) returns `true` unconditionally, so a /// 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: the retry is a single cheap LIST call, it means the -/// sweep self-heals the moment the permission is added, and the tick cadence -/// is documented in values.yaml. +/// 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, +/// 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( cached: &Option, last_attempt: &Option,