From 4142d2b84e1c3230fbe6c4b2d1e994721c409092 Mon Sep 17 00:00:00 2001 From: ravarora2 <130506156+ravarora2@users.noreply.github.com> Date: Wed, 23 Sep 2026 14:39:39 -0400 Subject: [PATCH] Use worker snapshots for relay storage metrics (#7845) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary Make the relay read completed storage-accounting snapshots from PostgreSQL by default. An environment with no snapshot emits no storage metrics. When a worker publishes its first result, the relay picks it up on the next successful leader usage tick, without a restart or configuration change. This changes how storage usage is measured and reported. **S3 remains the object store. PostgreSQL stores the completed counts, byte totals, community breakdown, and calculation metadata—not the objects themselves.** ### Why move relay storage accounting entirely to database reads? Today the relay supports two sources: its own S3 scan (`inline`, the default) and the worker's saved snapshot (`external`). That requires operators to deploy a worker and then coordinate a separate relay-mode change in each environment. The worker already owns the S3 listing and calculation. It runs in a separate process with its own object cap, memory limit, and deadline. Keeping a second scan inside the serving relay retains the resource pressure and failure modes that this isolation is meant to remove. PostgreSQL provides a durable handoff. The worker replaces one completed snapshot atomically after a successful calculation. The relay reads that row and exposes its values through the existing metrics endpoint. A relay restart or leadership change can load the last completed result without scanning the bucket again. There is deliberately no S3 fallback when a snapshot is absent or a read fails. Such a fallback would put the expensive scan back in the relay precisely when a worker is absent or unhealthy. It would also leave two calculation paths with different resource limits and freshness behavior. ### What happens if no worker exists? **A worker that has never published is a normal, successful inactive state.** The snapshot table must exist through the normal database schema, but it may contain zero rows. The reader returns success, emits no storage totals or failed-load gauge, and checks again on the next usage tick. It does not exit the relay, start a worker, or list S3. The relay checks for a completed database row; it does not inspect Kubernetes Jobs or CronJobs. Worker presence and snapshot presence are therefore different: | Situation | Storage-reader behavior | |---|---| | No worker and no snapshot row | Return success, emit no storage metrics, and query again next tick. | | Worker starts later and publishes its first row | Read and emit it on the next successful leader usage tick; no restart or mode change. | | Worker is removed or fails, but its last valid row remains | Continue reporting the last completed totals with their increasing age. | | Query succeeds but a previously present row has been removed | Clear the cache and stop refreshing storage series; existing exporter series expire through the idle timeout. | | Database read fails, times out, or returns an invalid snapshot | Log a warning, set `buzz_storage_snapshot_load_ok=0`, retain any last-good cached totals, and retry next tick. | | Snapshot table is missing | Treat this as a schema error, not normal worker absence; follow the failed-read behavior above. | | `BUZZ_STORAGE_METRICS=off` | Skip the snapshot query and all storage metric emission. | No row means “not measured,” not zero bytes. A valid completed snapshot of an empty bucket can report zero. If a read fails before any good snapshot has been cached, only failed-load health is emitted. These reads happen in the existing background metrics task, not the relay startup path. A storage-read failure does not request process exit or change readiness directly. A wider database outage can still affect the relay's existing database-dependent operations and readiness. ### Read, parse, and emit path ```text Worker: S3 listing -> BucketSnapshot -> JSON -> PostgreSQL Relay: PostgreSQL -> JSON -> BucketSnapshot -> existing metric gauges Collector: relay /metrics endpoint -> Datadog ``` 1. `main.rs::run_storage_sweep_tick` calls the reader from the existing leader-only usage task. The default interval remains 300 seconds. 2. `storage_sweep.rs::run_storage_metrics_tick` coordinates the read, cache update, failure health, and metric emission. 3. `refresh_persisted_snapshot` calls the existing `Db::load_storage_accounting_snapshot()` method. A five-second timeout covers pool acquisition and the query. It uses the relay's existing pool. 4. The existing query in `buzz-db/src/store/storage_accounting.rs` reads `snapshot`, `completed_at`, `duration_ms`, `max_objects`, and `code_sha` from the singleton row. 5. `serde_json::from_value(stored.snapshot)` decodes the JSON into the existing `buzz_media::BucketSnapshot` type. Rust infers that type from `cache_persisted_snapshot`'s argument. The generated `Deserialize` implementation handles all fields, including the UUID-keyed `per_community` map. 6. `emit_cached_storage_metrics` sets the existing physical/logical totals and community byte/object gauges. The existing Prometheus exporter exposes them for Datadog collection; this code does not send snapshot JSON to Datadog. The worker's `BucketSnapshot`/`CommunityStorage` structures, JSON field names, SQL publication method, and database schema are unchanged. The old external-mode JSON conversion moves out of `main.rs` into the reader helper. No new format or hand-written field parser is needed. Duration and object-cap metadata are validated before replacing the cache. The five-second relay read timeout is separate from the worker's 30-second startup acquisition budget introduced in #7770. This PR does not change the worker's pool or startup logic. ### Configuration and code changes | Configuration | Before | After | |---|---|---| | Unset, `inline`, or `on` | Relay-local S3 scan | Read completed database snapshots. `inline` logs one migration warning. | | `external` or `snapshot` | Read completed database snapshots | Continue reading snapshots. | | `off` | Disabled | Disabled. | | Unknown or empty value | Disabled with a configuration error | Same. | Existing `inline` settings intentionally become reader aliases. Merely changing the default would leave older deployments on their explicit `inline` setting and preserve the need for coordinated configuration PRs. Remove relay-local scan scheduling, in-flight scan state, scan configuration, and obsolete scan-attempt tests and health metrics. Preserve the physical/logical total and per-community metric names, leader-only emission, community scope filtering, and cleanup of old community labels. The shared relay-state field remains in place; its comment now describes a cached worker result. The Helm chart defaults to `relayMode: external` while leaving `storageAccounting.enabled: false`. Its rendering test verifies that the reader is configured even when no worker is created. The reader regression tests cover missing rows, later publication, updates, invalid data, timeout, removal, reactivation, and the explicit off switch. `docs/storage-accounting.md` describes this operating model. ### Freshness and rollout impact `buzz_storage_snapshot_age_seconds` uses the worker's original completion timestamp. Re-reading an old row never makes it fresh. `buzz_storage_snapshot_load_ok=1` means the row was read and decoded; it does not prove that the worker's latest attempt succeeded. Monitor snapshot age against the worker's schedule and allowed runtime, and monitor Job failures separately. The old relay scan-attempt health gauges are retired because the relay no longer runs those scans. Deploy this relay version through the usual release process. Once it reaches an environment, workers can be enabled independently. Already-released charts that still set `inline` work with the new binary; future chart releases default explicitly to `external`. An intentional `off` override still requires an operator to enable the reader. **Environments with no completed snapshot lose relay-calculated storage totals after this upgrade.** That is the intended tradeoff for removing scans from the serving process. Environments with a prior snapshot continue showing its last completed totals and age. This PR does not install workers in additional environments or deploy the new relay image. ### Related issue Related to #4601. Follows the worker startup fix in #7770. The separate fleet-usage collector work in #7176 overlaps this code and may need reconciliation when it merges. ### Testing Built and ran the relay locally against isolated PostgreSQL and Redis with the legacy `inline` setting, five-second usage ticks, and a 15-second metric idle timeout. The same process was exercised through these steps: 1. Start with an empty snapshot table: relay becomes ready and exposes no storage series. 2. Publish a snapshot containing 123 bytes: the community gauge becomes 123. 3. Replace it with 456 bytes: the gauge becomes 456. 4. Remove the row: storage series expire without fabricating a zero-byte measurement. 5. Publish 789 bytes: metrics reactivate without restarting the relay; readiness remains healthy. The local S3 endpoint recorded zero requests. The unrelated Git conformance startup probe was disabled for this isolated reader check; the test covers storage accounting, not every other use of S3 in the relay. No new relay image was deployed to staging or production during this test. The broad local integration run also hit an existing `p0_pool_acquisitions_use_typed_operation_pairs_without_other` failure in `buzz-db/tests/observability_source.rs`. The same source-only test fails on unmodified `main` at `0ef7a2222`; this PR changes neither the test nor its failing source. The full local `just ci` run stopped at mobile native-asset setup: the `objective_c` build hook received no SDK path from `xcrun`. Mobile tests did not run. Generated with Codex Signed-off-by: Ravneet Arora --- crates/buzz-relay/src/main.rs | 84 +- crates/buzz-relay/src/state.rs | 2 +- crates/buzz-relay/src/storage_sweep.rs | 945 ++++-------------- .../src/storage_sweep/postgres_tests.rs | 151 +++ .../buzz/tests/storage_accounting_test.yaml | 8 +- deploy/charts/buzz/values.yaml | 31 +- docs/storage-accounting.md | 56 ++ 7 files changed, 411 insertions(+), 866 deletions(-) create mode 100644 crates/buzz-relay/src/storage_sweep/postgres_tests.rs create mode 100644 docs/storage-accounting.md 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.