diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 532defbd8d..114b51ee7f 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -1745,84 +1745,26 @@ async fn run_usage_metrics_tick( 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. +/// Read worker storage snapshots only on the existing leader metrics tick. async fn run_storage_sweep_tick( state: &AppState, emission_scope: &EmissionScope, host_map: &HashMap, ) { - static SWEEP_CONFIG: std::sync::OnceLock = + static MODE: std::sync::OnceLock = std::sync::OnceLock::new(); - // 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); - match config.mode { - storage_sweep::StorageMetricsMode::Disabled => return, - storage_sweep::StorageMetricsMode::Inline => { - 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::StorageMetricsMode::Snapshot => { - match state.db.load_storage_accounting_snapshot().await { - Ok(Some(stored)) => match serde_json::from_value(stored.snapshot) { - Ok(snapshot) => { - let duration = std::time::Duration::from_millis( - u64::try_from(stored.duration_ms).unwrap_or_default(), - ); - let max_objects = u64::try_from(stored.max_objects).unwrap_or_default(); - storage_sweep::cache_persisted_snapshot( - &state.storage_sweep, - snapshot, - stored.completed_at, - duration, - max_objects, - ) - .await; - } - Err(error) => { - warn!(error = %error, "stored storage snapshot is invalid"); - storage_sweep::record_persisted_snapshot_load_failure(&state.storage_sweep) - .await; - } - }, - Ok(None) => { - storage_sweep::record_persisted_snapshot_load_failure(&state.storage_sweep) - .await; - } - Err(error) => { - warn!(error = %error, "failed to load stored storage snapshot"); - storage_sweep::record_persisted_snapshot_load_failure(&state.storage_sweep) - .await; - } - } - } + let mode = *MODE.get_or_init(storage_sweep::StorageMetricsMode::from_env); + if let Err(error) = storage_sweep::run_storage_metrics_tick( + &state.db, + &state.storage_sweep, + mode, + host_map, + |id| emission_scope.allows(id), + ) + .await + { + warn!(error = %error, "failed to load stored storage snapshot; retrying next usage tick"); } - - storage_sweep::emit_storage_metrics(&state.storage_sweep, config.mode, host_map, |id| { - emission_scope.allows(id) - }) - .await; } /// Emit the database-derived usage snapshot from the stable leader only. diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index 98864df49e..cbfe7b7b30 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -703,7 +703,7 @@ pub struct AppState { pub audit_tx: Option>, /// Media storage client (S3/MinIO). pub media_storage: Arc, - /// Single-flight + cache state for the hourly S3 storage sweep. See + /// Cached worker snapshot and storage metric emission bookkeeping. 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>, diff --git a/crates/buzz-relay/src/storage_sweep.rs b/crates/buzz-relay/src/storage_sweep.rs index 0e807bca1f..94c1938376 100644 --- a/crates/buzz-relay/src/storage_sweep.rs +++ b/crates/buzz-relay/src/storage_sweep.rs @@ -1,99 +1,48 @@ -//! Hourly S3 storage sweep: single-flight background task, cached snapshot, -//! re-emitted every usage-metrics tick. +//! Publish storage metrics from the latest complete worker snapshot. //! -//! 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. -//! - In inline mode, 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. +//! The leader checks PostgreSQL on every usage tick. An environment without a +//! snapshot emits no storage gauges; the first worker publication is picked up +//! without restarting or reconfiguring the relay. S3 listing belongs only to +//! the worker. Failed reads retain the last good totals with load health and +//! the original snapshot age, never a fabricated fresh measurement. use std::collections::{HashMap, HashSet}; -use std::future::Future; -use std::time::{Duration, Instant}; +use std::time::Duration; use chrono::{DateTime, Utc}; use tokio::sync::Mutex; -use tokio::task::JoinHandle; use uuid::Uuid; -use buzz_media::{BucketSnapshot, SweepError}; +use buzz_db::Db; +use buzz_media::BucketSnapshot; -/// Where relay storage gauges get their snapshot. +/// Storage metrics are read from PostgreSQL unless explicitly disabled. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum StorageMetricsMode { - /// Preserve the existing in-process S3 sweep. - Inline, /// Read the newest completed snapshot persisted by `buzz-admin`. Snapshot, - /// Emit no storage-family metrics. + /// Neither read snapshots nor emit storage-family metrics. Disabled, } -/// 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, - /// Snapshot source and kill switch. - pub mode: StorageMetricsMode, -} - -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 - /// `BUZZ_STORAGE_METRICS` (`inline`, `external`, or `off`). Unset keeps - /// the legacy inline behavior; unknown values fail closed. +impl StorageMetricsMode { + /// Read `BUZZ_STORAGE_METRICS`. Unset enables the database reader. + /// Legacy `inline`/`on` values also select the reader so existing charts + /// need no coordinated mode change. `off` and unknown values disable it. 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 raw_mode = std::env::var("BUZZ_STORAGE_METRICS") - .ok() - .map(|v| v.trim().to_ascii_lowercase()); - let mode = parse_storage_metrics_mode(raw_mode.as_deref()); - Self { - interval: Duration::from_secs(interval_secs), - timeout: Duration::from_secs(timeout_secs), - max_objects, - mode, - } + let value = std::env::var("BUZZ_STORAGE_METRICS").ok(); + parse_storage_metrics_mode(value.as_deref()) } } fn parse_storage_metrics_mode(value: Option<&str>) -> StorageMetricsMode { - match value { + match value.map(str::trim).map(str::to_ascii_lowercase).as_deref() { Some("off") => StorageMetricsMode::Disabled, - Some("snapshot") | Some("external") => StorageMetricsMode::Snapshot, - Some("inline") | Some("on") | None => StorageMetricsMode::Inline, + Some("inline") => { + tracing::warn!("BUZZ_STORAGE_METRICS=inline now reads worker snapshots; relay S3 sweeps are retired"); + StorageMetricsMode::Snapshot + } + Some("snapshot" | "external" | "on") | None => StorageMetricsMode::Snapshot, Some(other) => { tracing::error!( value = other, @@ -104,29 +53,16 @@ fn parse_storage_metrics_mode(value: Option<&str>) -> StorageMetricsMode { } } -/// 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, -} +// Bound the whole read, including pool acquisition and the database query. +const SNAPSHOT_READ_TIMEOUT: Duration = Duration::from_secs(5); -/// The most recent successfully computed snapshot. +/// The most recent successfully decoded worker snapshot. #[derive(Debug, Clone)] struct CachedSnapshot { data: BucketSnapshot, - completed_at: Instant, completed_at_wall: DateTime, duration: Duration, - max_objects: Option, -} - -/// What a spawned sweep task hands back to the tick that harvests it. -struct SweepAttempt { - result: Result, - duration: Duration, + max_objects: u64, } /// Key tracking which per-community series were emitted in the previous tick, @@ -158,267 +94,119 @@ impl StorageEmittedKey { } } -/// 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`). +/// Cached worker data and emission bookkeeping, shared across usage ticks. #[derive(Default)] pub struct StorageSweepState { - in_flight: Option>, cached: Option, - last_attempt: Option, - failures_total: u64, - /// Whether the latest external-mode read yielded a decodable snapshot. + // None means no worker snapshot exists. A failed read is distinct from + // absence and remains observable even before the first successful read. persisted_load_ok: Option, - /// 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`). +/// Load and emit one snapshot on the existing leader-only usage cadence. /// -/// 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: 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, - 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, - }, +/// No row is a normal inactive state. Database/decoding failures preserve the +/// last good cache, emit failed-load health, and propagate for caller logging; +/// the next usage tick retries. This path never lists S3 or delays startup. +pub async fn run_storage_metrics_tick( + db: &Db, + state: &Mutex, + mode: StorageMetricsMode, + host_map: &HashMap, + allows: impl Fn(&Uuid) -> bool, +) -> anyhow::Result<()> { + if mode == StorageMetricsMode::Disabled { + return Ok(()); } + let result = refresh_persisted_snapshot(db, state).await; + if result.is_err() { + record_persisted_snapshot_load_failure(state).await; + } + emit_storage_metrics(state, mode, host_map, allows).await; + result } -/// 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( +async fn refresh_persisted_snapshot( + db: &Db, 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(); - 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(), - completed_at_wall: Utc::now(), - duration: attempt.duration, - max_objects: None, - }); - } - 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, - 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); +) -> anyhow::Result<()> { + let stored = tokio::time::timeout(SNAPSHOT_READ_TIMEOUT, db.load_storage_accounting_snapshot()) + .await??; + let Some(stored) = stored else { + // A removed row also stops refreshing old metric series. They expire + // through the exporter's idle timeout; absence never means zero bytes. + let mut state = state.lock().await; + state.cached = None; + state.persisted_load_ok = None; + // Keep the emitted keys until a valid snapshot returns, so its first + // emission can clear old community labels even before idle eviction. + return Ok(()); + }; + let snapshot = serde_json::from_value(stored.snapshot)?; + let duration = Duration::from_millis(u64::try_from(stored.duration_ms)?); + let max_objects = u64::try_from(stored.max_objects)?; + anyhow::ensure!( + max_objects > 0, + "stored storage snapshot has a zero object cap" + ); + cache_persisted_snapshot(state, snapshot, stored.completed_at, duration, max_objects).await; + Ok(()) } /// Replace the cache with a complete snapshot loaded from durable storage. -pub async fn cache_persisted_snapshot( +async fn cache_persisted_snapshot( state: &Mutex, snapshot: BucketSnapshot, completed_at: DateTime, duration: Duration, max_objects: u64, ) { - let age = Utc::now() - .signed_duration_since(completed_at) - .to_std() - .unwrap_or_default(); let mut state = state.lock().await; state.cached = Some(CachedSnapshot { data: snapshot, - completed_at: Instant::now().checked_sub(age).unwrap_or_else(Instant::now), completed_at_wall: completed_at, duration, - max_objects: Some(max_objects), + max_objects, }); state.persisted_load_ok = Some(true); } -/// Record that the latest persisted-snapshot read did not yield a usable row. -pub async fn record_persisted_snapshot_load_failure(state: &Mutex) { +async fn record_persisted_snapshot_load_failure(state: &Mutex) { state.lock().await.persisted_load_ok = Some(false); } -/// Emit the mode-appropriate storage gauges on each leader metrics tick. -pub async fn emit_storage_metrics( +async fn emit_storage_metrics( state: &Mutex, mode: StorageMetricsMode, host_map: &HashMap, allows: impl Fn(&Uuid) -> bool, ) { - match mode { - StorageMetricsMode::Inline => emit_inline_storage_metrics(state, host_map, allows).await, - StorageMetricsMode::Snapshot => { - emit_persisted_storage_metrics(state, host_map, allows).await - } - StorageMetricsMode::Disabled => {} + if mode == StorageMetricsMode::Disabled { + return; } -} - -/// Emit inline-sweep health and storage gauges from the cached snapshot. Never -/// called from the spawned sweep task itself, so a sweep that completes after -/// this pod loses leadership parks its snapshot without 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. -/// -/// 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. -async fn emit_inline_storage_metrics( - state: &Mutex, - host_map: &HashMap, - allows: impl Fn(&Uuid) -> bool, -) { 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. - // 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()); - } - - emit_cached_storage_metrics(&mut state, host_map, allows, CachedMetricSource::Inline); -} - -/// Emit storage gauges loaded from the durable worker snapshot. -async fn emit_persisted_storage_metrics( - state: &Mutex, - host_map: &HashMap, - allows: impl Fn(&Uuid) -> bool, -) { - let mut state = state.lock().await; - let load_ok = state.persisted_load_ok.unwrap_or(false); + let Some(load_ok) = state.persisted_load_ok else { + return; + }; metrics::gauge!("buzz_storage_snapshot_load_ok").set(if load_ok { 1.0 } else { 0.0 }); - - emit_cached_storage_metrics(&mut state, host_map, allows, CachedMetricSource::Persisted); -} - -#[derive(Clone, Copy)] -enum CachedMetricSource { - Inline, - Persisted, + emit_cached_storage_metrics(&mut state, host_map, allows); } fn emit_cached_storage_metrics( state: &mut StorageSweepState, host_map: &HashMap, allows: impl Fn(&Uuid) -> bool, - source: CachedMetricSource, ) { - // 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; }; let age = Utc::now() .signed_duration_since(cached.completed_at_wall) .to_std() .unwrap_or_default(); - match source { - CachedMetricSource::Inline => { - metrics::gauge!("buzz_storage_sweep_age_seconds").set(age.as_secs_f64()); - } - CachedMetricSource::Persisted => { - metrics::gauge!("buzz_storage_snapshot_age_seconds").set(age.as_secs_f64()); - metrics::gauge!("buzz_storage_snapshot_duration_seconds") - .set(cached.duration.as_secs_f64()); - } - } + metrics::gauge!("buzz_storage_snapshot_age_seconds").set(age.as_secs_f64()); + metrics::gauge!("buzz_storage_snapshot_duration_seconds").set(cached.duration.as_secs_f64()); let snapshot = &cached.data; metrics::gauge!("buzz_total_storage_bytes", "kind" => "physical") @@ -437,11 +225,9 @@ fn emit_cached_storage_metrics( 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); - if let (CachedMetricSource::Persisted, Some(max_objects)) = (source, cached.max_objects) { - metrics::gauge!("buzz_storage_snapshot_max_objects").set(max_objects as f64); - metrics::gauge!("buzz_storage_snapshot_cap_utilization") - .set(snapshot.physical_objects as f64 / max_objects as f64); - } + metrics::gauge!("buzz_storage_snapshot_max_objects").set(cached.max_objects as f64); + metrics::gauge!("buzz_storage_snapshot_cap_utilization") + .set(snapshot.physical_objects as f64 / cached.max_objects as f64); let mut current = HashSet::new(); let mut unmapped_bytes = 0u64; @@ -476,11 +262,12 @@ fn emit_cached_storage_metrics( state.previously_emitted = current; } +#[cfg(test)] +mod postgres_tests; + #[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}; @@ -498,465 +285,38 @@ mod tests { } } - // --- 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", + fn default_and_legacy_chart_values_use_worker_snapshots() { + for value in [ + None, + Some("inline"), + Some("on"), + Some("snapshot"), + Some("external"), + Some(" INLINE "), ] { - 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_eq!(config.mode, StorageMetricsMode::Inline); - } - - #[test] - fn config_mode_accepts_snapshot_and_fails_closed_on_unknown_values() { - assert_eq!( - parse_storage_metrics_mode(Some("off")), - StorageMetricsMode::Disabled - ); - assert_eq!( - parse_storage_metrics_mode(Some("snapshot")), - StorageMetricsMode::Snapshot - ); - assert_eq!( - parse_storage_metrics_mode(Some("external")), - StorageMetricsMode::Snapshot - ); - assert_eq!( - parse_storage_metrics_mode(Some("on")), - StorageMetricsMode::Inline - ); - assert_eq!( - parse_storage_metrics_mode(Some("inline")), - StorageMetricsMode::Inline - ); - assert_eq!( - parse_storage_metrics_mode(Some("anything-else")), - StorageMetricsMode::Disabled - ); - assert_eq!(parse_storage_metrics_mode(None), StorageMetricsMode::Inline); - } - - // --- 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(), - completed_at_wall: Utc::now(), - duration: Duration::from_secs(1), - max_objects: None, - }); - 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, - completed_at_wall: Utc::now(), - duration: Duration::from_secs(1), - max_objects: None, - }); - 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 + parse_storage_metrics_mode(value), + StorageMetricsMode::Snapshot, + "{value:?}" ); } - - // 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); + for value in [Some("off"), Some(" OFF "), Some("unknown"), Some("")] { + assert_eq!( + parse_storage_metrics_mode(value), + StorageMetricsMode::Disabled, + "{value:?}" + ); } - 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()); - } - - // --- 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, - StorageMetricsMode::Inline, - &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, - StorageMetricsMode::Inline, - &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, - StorageMetricsMode::Inline, - &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 { recorder .snapshotter() .snapshot() .into_vec() .into_iter() + .filter(|(key, _, _, _)| key.key().name().contains("_storage_")) .filter_map(|(key, _, _, value)| match value { DebugValue::Gauge(v) => Some((key.key().name().to_owned(), v.into_inner())), _ => None, @@ -965,25 +325,65 @@ mod tests { } #[tokio::test] - async fn cold_cache_emits_only_health_gauges() { + async fn missing_snapshot_emits_nothing() { 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, - StorageMetricsMode::Inline, - &host_map, + StorageMetricsMode::Snapshot, + &HashMap::new(), |_| true, )); }); + assert!(gauge_snapshot(&recorder).is_empty()); + } - let values = gauge_snapshot(&recorder); - assert_eq!(values.get("buzz_storage_sweep_ok"), 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")); + #[tokio::test] + async fn disabled_reader_never_queries_a_closed_pool() { + let pool = sqlx::postgres::PgPoolOptions::new() + .connect_lazy("postgres://localhost/unused") + .unwrap(); + pool.close().await; + let db = Db::from_pool(pool); + let state = Mutex::new(StorageSweepState::default()); + let recorder = DebuggingRecorder::new(); + let result = metrics::with_local_recorder(&recorder, || { + futures::executor::block_on(run_storage_metrics_tick( + &db, + &state, + StorageMetricsMode::Disabled, + &HashMap::new(), + |_| true, + )) + }); + assert!(result.is_ok()); + assert!(gauge_snapshot(&recorder).is_empty()); + } + + #[tokio::test] + async fn failed_first_read_is_an_error_instead_of_worker_absence() { + let pool = sqlx::postgres::PgPoolOptions::new() + .connect_lazy("postgres://localhost/unused") + .unwrap(); + pool.close().await; + let db = Db::from_pool(pool); + let state = Mutex::new(StorageSweepState::default()); + let recorder = DebuggingRecorder::new(); + let result = metrics::with_local_recorder(&recorder, || { + futures::executor::block_on(run_storage_metrics_tick( + &db, + &state, + StorageMetricsMode::Snapshot, + &HashMap::new(), + |_| true, + )) + }); + assert!(result.is_err()); + assert_eq!( + gauge_snapshot(&recorder), + HashMap::from([("buzz_storage_snapshot_load_ok".into(), 0.0)]) + ); } #[tokio::test] @@ -1025,15 +425,11 @@ mod tests { let state = Mutex::new(StorageSweepState { cached: Some(CachedSnapshot { data: snapshot, - completed_at: Instant::now(), completed_at_wall: Utc::now(), duration: Duration::from_millis(500), - max_objects: None, - }), - last_attempt: Some(LastAttempt { - ok: true, - duration: Duration::from_millis(500), + max_objects: 1_000, }), + persisted_load_ok: Some(true), ..Default::default() }); @@ -1045,14 +441,14 @@ mod tests { metrics::with_local_recorder(&recorder, || { futures::executor::block_on(emit_storage_metrics( &state, - StorageMetricsMode::Inline, + StorageMetricsMode::Snapshot, &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_storage_snapshot_load_ok"), Some(&1.0)); assert_eq!(values.get("buzz_total_storage_bytes"), Some(&137.0)); assert_eq!( values.get("buzz_storage_unmapped_community_bytes"), @@ -1140,15 +536,11 @@ mod tests { let state = Mutex::new(StorageSweepState { cached: Some(CachedSnapshot { data: make_snapshot(true, 20), - completed_at: Instant::now(), completed_at_wall: Utc::now(), duration: Duration::from_millis(100), - max_objects: None, - }), - last_attempt: Some(LastAttempt { - ok: true, - duration: Duration::from_millis(100), + max_objects: 1_000, }), + persisted_load_ok: Some(true), ..Default::default() }); @@ -1162,7 +554,7 @@ mod tests { metrics::with_local_recorder(&recorder, || { futures::executor::block_on(emit_storage_metrics( &state, - StorageMetricsMode::Inline, + StorageMetricsMode::Snapshot, &host_map_1, |_| true, )); @@ -1192,10 +584,9 @@ mod tests { let mut guard = state.lock().await; guard.cached = Some(CachedSnapshot { data: make_snapshot(false, 20), - completed_at: Instant::now(), completed_at_wall: Utc::now(), duration: Duration::from_millis(100), - max_objects: None, + max_objects: 1_000, }); } let mut host_map_2 = HashMap::new(); @@ -1205,7 +596,7 @@ mod tests { metrics::with_local_recorder(&recorder, || { futures::executor::block_on(emit_storage_metrics( &state, - StorageMetricsMode::Inline, + StorageMetricsMode::Snapshot, &host_map_2, |id| *id != community_c, // (c) scope-excluded )); diff --git a/crates/buzz-relay/src/storage_sweep/postgres_tests.rs b/crates/buzz-relay/src/storage_sweep/postgres_tests.rs new file mode 100644 index 0000000000..efe5a097b9 --- /dev/null +++ b/crates/buzz-relay/src/storage_sweep/postgres_tests.rs @@ -0,0 +1,151 @@ +//! Exercise the same database-to-metrics tick called by the relay leader. + +use super::*; +use buzz_media::CommunityStorage; +use metrics_util::debugging::{DebugValue, DebuggingRecorder}; + +fn observe_tick( + db: &Db, + state: &Mutex, + mode: StorageMetricsMode, + hosts: &HashMap, +) -> (anyhow::Result<()>, HashMap) { + let recorder = DebuggingRecorder::new(); + let result = metrics::with_local_recorder(&recorder, || { + futures::executor::block_on(run_storage_metrics_tick(db, state, mode, hosts, |_| true)) + }); + let values = recorder + .snapshotter() + .snapshot() + .into_vec() + .into_iter() + .filter(|(key, _, _, _)| key.key().name().contains("_storage_")) + .filter_map(|(key, _, _, value)| match value { + DebugValue::Gauge(value) => Some((key.key().name().to_owned(), value.into_inner())), + _ => None, + }) + .collect(); + (result, values) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[ignore = "requires Postgres"] +async fn worker_publication_activates_reader_without_restart_or_mode_change() { + // The PostgreSQL lane supplies a separate desired-state database per test. + // Local callers must likewise supply an explicitly isolated database. + let url = std::env::var("BUZZ_TEST_DATABASE_URL").expect("isolated BUZZ_TEST_DATABASE_URL"); + let pool = sqlx::postgres::PgPoolOptions::new() + .max_connections(4) + .connect(&url) + .await + .expect("connect isolated database"); + let db = Db::from_pool(pool.clone()); + let community = Uuid::new_v4(); + let hosts = HashMap::from([(community, "storage.example.test".to_owned())]); + + for configured in [None, Some("inline"), Some("external")] { + sqlx::query("DELETE FROM storage_accounting_snapshots") + .execute(&pool) + .await + .expect("clear isolated snapshot"); + let state = Mutex::new(StorageSweepState::default()); + let mode = parse_storage_metrics_mode(configured); + let (result, values) = observe_tick(&db, &state, mode, &hosts); + result.expect("no worker is normal"); + assert!( + values.is_empty(), + "no row must not emit zeros or failed-load health" + ); + + let mut worker = db.try_lock_storage_accounting().await.unwrap().unwrap(); + for bytes in [100, 250] { + let snapshot = BucketSnapshot { + physical_bytes: bytes, + physical_objects: 1, + logical_bytes: bytes, + logical_objects: 1, + per_community: HashMap::from([(community, CommunityStorage { bytes, objects: 1 })]), + ..Default::default() + }; + worker + .save_snapshot( + &serde_json::to_value(snapshot).unwrap(), + 12, + 1_000, + "test-worker", + ) + .await + .expect("publish through production worker session"); + let (result, values) = observe_tick(&db, &state, mode, &hosts); + result.expect("same running reader picks up worker publication"); + assert_eq!(values["buzz_community_storage_bytes"], bytes as f64); + assert_eq!(values["buzz_total_storage_bytes"], bytes as f64); + assert_eq!(values["buzz_storage_snapshot_load_ok"], 1.0); + assert!(!values + .keys() + .any(|key| key.starts_with("buzz_storage_sweep_"))); + } + // Corrupt data must not be mistaken for a worker that is not installed. + worker + .save_snapshot( + &serde_json::json!({"invalid": true}), + 12, + 1_000, + "bad-worker", + ) + .await + .unwrap(); + let (result, values) = observe_tick(&db, &state, mode, &hosts); + assert!(result.is_err()); + assert_eq!(values["buzz_storage_snapshot_load_ok"], 0.0); + assert_eq!(values["buzz_community_storage_bytes"], 250.0); + + // A blocked table read is bounded, preserves the cache, and can recover. + let mut blocker = pool.begin().await.unwrap(); + sqlx::query("LOCK TABLE storage_accounting_snapshots IN ACCESS EXCLUSIVE MODE") + .execute(&mut *blocker) + .await + .unwrap(); + let started = std::time::Instant::now(); + let (result, values) = observe_tick(&db, &state, mode, &hosts); + assert!(result.is_err()); + assert!(started.elapsed() < Duration::from_secs(8)); + assert_eq!(values["buzz_storage_snapshot_load_ok"], 0.0); + assert_eq!(values["buzz_community_storage_bytes"], 250.0); + blocker.rollback().await.unwrap(); + + sqlx::query("DELETE FROM storage_accounting_snapshots") + .execute(&pool) + .await + .unwrap(); + let (result, values) = observe_tick(&db, &state, mode, &hosts); + result.expect("removed worker snapshot is inactive"); + assert!( + values.is_empty(), + "do not refresh stale totals or report zero storage" + ); + + worker + .save_snapshot( + &serde_json::to_value(BucketSnapshot::default()).unwrap(), + 1, + 100, + "restarted-worker", + ) + .await + .unwrap(); + let (result, values) = observe_tick(&db, &state, mode, &hosts); + result.expect("reader reactivates after removal and transient failure"); + assert_eq!(values["buzz_storage_snapshot_load_ok"], 1.0); + assert_eq!( + values["buzz_total_storage_bytes"], 0.0, + "a real empty snapshot may report zero" + ); + assert_eq!( + values["buzz_community_storage_bytes"], 0.0, + "clear the previous community when publication resumes" + ); + drop(worker); + } + pool.close().await; +} diff --git a/deploy/charts/buzz/tests/storage_accounting_test.yaml b/deploy/charts/buzz/tests/storage_accounting_test.yaml index 916e3028d4..d37d8aefe9 100644 --- a/deploy/charts/buzz/tests/storage_accounting_test.yaml +++ b/deploy/charts/buzz/tests/storage_accounting_test.yaml @@ -4,7 +4,7 @@ templates: - templates/deployment.yaml - templates/secret-chart.yaml tests: - - it: renders no worker by default + - it: enables the snapshot reader without requiring a worker set: relayUrl: wss://buzz.example.com ownerPubkey: "0000000000000000000000000000000000000000000000000000000000000000" @@ -17,6 +17,12 @@ tests: - hasDocuments: count: 0 template: templates/storage-accounting-cronjob.yaml + - contains: + path: spec.template.spec.containers[0].env + content: + name: BUZZ_STORAGE_METRICS + value: external + template: templates/deployment.yaml - it: renders a non-overlapping worker with an isolated cap set: relayUrl: wss://buzz.example.com diff --git a/deploy/charts/buzz/values.yaml b/deploy/charts/buzz/values.yaml index 5e41f22f5f..496c32e979 100644 --- a/deploy/charts/buzz/values.yaml +++ b/deploy/charts/buzz/values.yaml @@ -329,19 +329,22 @@ externalRedis: # the endpoint + autogenerated credentials automatically. # # Storage metrics are controlled by storageAccounting.relayMode below: -# - inline (default) keeps the relay-local hourly bucket sweep. -# - external reads the latest complete snapshot written by the CronJob. +# - external (default) reads the latest complete snapshot written by a worker. +# No row means no storage gauges; a later worker publication is picked up on +# the next usage tick without restarting or reconfiguring the relay. +# - inline is a legacy alias for external in current relay images. +# A directly configured BUZZ_STORAGE_METRICS=on also selects the reader. # - off disables storage metrics in the relay. -# storageAccounting.enabled controls the CronJob independently, so operators -# can stage the worker before switching the relay to external mode. Inline and -# worker scans require `s3:ListBucket` on the bucket ARN itself (bucket-level; +# storageAccounting.enabled controls the CronJob independently of the reader. +# Only worker scans require `s3:ListBucket` on the bucket ARN itself (bucket-level; # distinct from the object-level GetObject/PutObject/DeleteObject permissions # already required for media). Without it the scan fails AccessDenied and no # other media functionality is affected. External mode emits # buzz_storage_snapshot_{age_seconds,duration_seconds,max_objects, # cap_utilization,load_ok}; it does not infer worker-attempt health from the -# last successful snapshot. Monitor CronJob/Job status for worker failures, -# including deadline expiry and OOM termination. +# last successful snapshot. Old totals retain their original age; a missing +# row stops refreshing all storage series. Monitor snapshot age and CronJob/Job +# status for worker failures, including deadline expiry and OOM termination. # # Whole-community deletion (`buzz-admin deletions ...`) permanently removes # tenant-owned object versions through the v5 deletion path. In addition to the @@ -351,12 +354,8 @@ externalRedis: # never-versioned buckets, which S3 exposes through `ListObjectVersions` with the # `null` version id. # -# Note: inline-only 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. +# Snapshot reads run on the usage tick (default 300 s), only on its elected +# leader. A read is bounded to five seconds and never falls back to S3. s3: endpoint: "" bucket: "buzz-media" @@ -370,8 +369,8 @@ s3: accessKey: "" secretKey: "" -# Isolated S3 accounting worker. Disabled by default, so chart upgrades retain -# the existing relay-local sweep until an operator enables and sizes the job. +# Isolated S3 accounting worker. Disabled by default. The relay is ready to +# consume snapshots whenever an operator enables a worker in this environment. storageAccounting: enabled: false schedule: "0 3 * * *" @@ -379,7 +378,7 @@ storageAccounting: activeDeadlineSeconds: 3600 successfulJobsHistoryLimit: 1 failedJobsHistoryLimit: 3 - relayMode: inline # inline | external | off + relayMode: external # external | off; inline is a legacy alias serviceAccountName: "" # defaults to the main Buzz service account podLabels: {} podAnnotations: diff --git a/docs/storage-accounting.md b/docs/storage-accounting.md new file mode 100644 index 0000000000..1f695b8cc8 --- /dev/null +++ b/docs/storage-accounting.md @@ -0,0 +1,56 @@ +# Storage accounting + +`buzz-admin storage-snapshot` lists S3, calculates physical and per-community +logical usage, and atomically replaces the completed snapshot in PostgreSQL. +The relay reads that snapshot and exposes the existing storage gauges on its +metrics endpoint for collection by Datadog or another metrics collector. + +## Independent rollout + +Deploy the snapshot-reading relay version through the usual release process. +It is ready before a worker exists: no database row means no storage gauges. +Add a scheduled worker to each environment when ready. Its first successful +publication becomes visible on the next leader usage tick, without restarting +the relay or changing its mode. The default tick interval is 300 seconds. + +The snapshot table is part of the normal database schema, independent of worker +installation. A missing table or inaccessible database is an error, not evidence +that the worker is absent. Snapshot reads have a five-second budget, including +connection acquisition, and retry on the next usage tick. They do not run on +the relay startup path and never fall back to an S3 scan. + +`BUZZ_STORAGE_METRICS` accepts: + +| Value | Behavior | +|---|---| +| Unset, `external`, `snapshot`, `on` | Read completed worker snapshots. | +| `inline` | Legacy alias for the reader; logs a migration warning once. | +| `off` | Skip storage reads and metrics. | +| Anything else | Log a configuration error and disable storage metrics. | + +Relay-local scans have been removed. Existing chart values that set `inline` +need no coordinated configuration change. Environments without a worker will +stop producing storage totals when this relay version arrives. The former +`BUZZ_STORAGE_SWEEP_*` settings no longer apply; size and bound the worker +instead. An intentional `off` override still requires an operator to enable it. + +## Freshness and failures + +The reader publishes the last completed totals together with +`buzz_storage_snapshot_age_seconds`. Re-reading an old row never resets its +completion time. Alert on age using a threshold appropriate for the worker's +schedule and allowed run time; for a daily worker, allow more than 24 hours. +`buzz_storage_snapshot_load_ok=1` means the row was read and decoded, not that +the worker's most recent attempt succeeded. Check Job failures separately. + +A failed or invalid read logs an error, sets load health to zero, and retains +the last good totals with their original age. With no prior good snapshot, +only failed-load health is emitted. A successful read that finds no row clears +the cache and stops refreshing storage series. Existing exported series expire +through the metrics exporter's idle timeout; absence is never reported as zero +bytes. A valid empty snapshot can report zero bytes. + +The reader preserves leader-only emission, community scope filtering, and +cleanup of series for renamed or removed communities. It does not need S3 +listing permission. The worker still needs that permission and its database +configuration.