Isolate S3 storage metrics from the relay (#7543)

## Why

**This PR moves the S3 storage scan out of the serving relay and into a
separate batch job.** The job saves a complete result in PostgreSQL. The
relay reads that result, exports the existing storage totals, and
reports the snapshot's freshness and database-read health.

```text
Today:         relay -> scan S3 -> calculate totals -> Datadog metrics
After rollout: CronJob -> scan S3 -> PostgreSQL snapshot -> relay -> Datadog metrics
```

The code in this PR is the new pipeline. It adds the worker, database
snapshot, relay read mode, and Helm CronJob template. It does **not**
enable the pipeline in production. The chart keeps the CronJob disabled,
and the relay keeps its current inline mode by default.
[squareup/builderbot-platform-core-infrastructure#244](https://github.com/squareup/builderbot-platform-core-infrastructure/pull/244)
will set the production limits, enable the job, and switch the relay to
the saved snapshot after this PR merges.

The relay currently lists the media bucket once an hour and calculates
storage totals inside the serving process. The production bucket now
exceeds the relay's one million object cap, so each sweep fails and the
last good per-community values remain stale. The Datadog dashboard can
no longer show current storage use by community.

Raising that cap inside the serving relay would move a larger,
memory-heavy scan into the process that owns live WebSocket connections.
The scan retains state for unique blob hashes and community bindings, so
its memory use grows with the bucket. A large scan could raise relay
memory use, trigger restarts, and drop user connections.

The goal is to restore current metrics without putting a
production-scale bucket scan in the relay's failure domain. No media
objects move or change. Only the process that calculates the totals and
the handoff to the relay change.

## What

- Add `buzz-admin storage-snapshot`, a run-once worker that scans S3 and
saves one complete accounting snapshot in PostgreSQL.
- Add `external` relay mode to read the saved snapshot. Keep the current
`inline` mode as the default, and add `off` as a fail-closed option.
- Add an opt-in Helm CronJob for the worker. The 10 million object cap
exists only on this Job.
- Keep the current fleet and per-community gauges, and add snapshot age,
generation duration, configured cap, cap utilization, and database-load
health gauges.
- Add an additive migration for the latest complete snapshot and its run
metadata.

## Overall plan

1. Merge this PR to add the worker, database handoff, relay modes, and
disabled-by-default chart support. This step does not change production
behavior.
2. Publish the relay image and Helm chart `0.1.9` from `main`.
3. Update
[squareup/builderbot-platform-core-infrastructure#244](https://github.com/squareup/builderbot-platform-core-infrastructure/pull/244)
to pin those final artifacts and rerender its manifests.
4. Use the infrastructure PR to deploy the CronJob with a 10 million
object cap, a one-hour deadline, and a 20 GiB memory limit.
5. Start one manual Job during a staffed window and verify its snapshot,
duration, memory use, relay stability, and Datadog output.
6. Keep the daily schedule and `external` relay mode only after that
gate passes, then add alerts for stale snapshots, snapshot-load failure,
failed jobs, cap pressure, duration, and worker memory.

The code and infrastructure remain separate so this PR can ship the
mechanism without silently enabling a production-scale scan. The
infrastructure PR owns the production settings and rollout gate.

## How

### Complete snapshot handoff

The worker takes a deployment-wide PostgreSQL advisory lock before it
lists S3. This prevents scheduled and manual workers from overlapping,
even if Kubernetes starts both.

The final database replacement runs through that same lock-owning
session. If the session dies during the scan, publication fails instead
of letting the stale worker overwrite its successor. A worker that no
longer owns the lock cannot publish through another pooled connection.

The worker reads S3 in pages of 1,000 objects and folds each page into
physical totals, logical per-community usage, and anomaly counts. It
stops before folding a page that would exceed `--max-objects` and emits
structured progress every 100,000 objects.

The worker replaces the singleton database row only after the full scan
succeeds. A listing error, malformed page, cap breach, serialization
error, lost lock session, or database error cannot publish a partial
snapshot. The last completed snapshot stays available. Each stored
result includes its completion time, duration, cap, and image or source
revision.

PostgreSQL is the handoff between the batch worker and the relay. The
worker writes one complete deployment-wide snapshot, and the relay reads
that snapshot on its normal leader metrics tick. No new network service
or worker endpoint is required.

### Relay behavior

`BUZZ_STORAGE_METRICS=inline` keeps the existing in-process sweep and
its one million object default. `external` loads the newest completed
worker snapshot on each leader metrics tick. `off` emits no
storage-family metrics. An unknown value fails closed to `off`.

External mode republishes the existing fleet, per-community, and anomaly
metric names, so those current queries remain valid. Snapshot handoff
health uses a separate family: `buzz_storage_snapshot_age_seconds`,
`buzz_storage_snapshot_duration_seconds`,
`buzz_storage_snapshot_max_objects`,
`buzz_storage_snapshot_cap_utilization`, and
`buzz_storage_snapshot_load_ok`.

The `buzz_storage_sweep_*` gauges remain inline-only because they
describe a scan attempted by the relay process. External mode does not
synthesize `sweep_ok=1` from the last successful database row: an old
row proves only that an earlier worker completed. Kubernetes CronJob/Job
status is authoritative for the latest worker result, including non-zero
exit, deadline expiry, and OOM termination.

The relay keeps the last complete snapshot available when a worker run
fails. Snapshot age shows that no replacement arrived on schedule, while
`buzz_storage_snapshot_load_ok` distinguishes staleness from a missing
row, database read failure, or invalid payload. If a community
disappears, changes host, or leaves the configured emission scope, the
relay zeroes the old labeled series instead of leaving stale values
behind.

Per-community bytes are logical referenced media usage. If two
communities reference the same blob, each community receives the blob's
bytes. Fleet physical totals still count each S3 object once. This PR
preserves those existing metric semantics.

### Deployment boundary

The CronJob is disabled by default. When enabled, it uses
`concurrencyPolicy: Forbid`, no retry, a configurable active deadline,
bounded history, explicit CPU and memory resources, a read-only root
filesystem, and no Istio sidecar by default.

The worker receives only `DATABASE_URL`, S3 settings, optional S3
credential references, and the code revision. It does not receive the
relay private key, Git hook HMAC secret, Redis URL, or the full relay
secret as environment variables. Operators may reuse the Buzz service
account or select a dedicated one.

### Non-goals

- This PR does not enable the worker or `external` mode in production.
- This PR does not change media uploads, object ownership, retention, or
deletion.
- This PR does not change the names or meaning of the existing fleet,
per-community, or anomaly storage gauges.
- This PR does not add a lease service or allow multiple workers to
publish concurrently.

## Risk

The application and chart defaults do not enable the new worker or raise
the serving relay's cap. The database change is an additive
deployment-global singleton table. The relay's default remains `inline`,
so merging this PR alone does not change the active metrics path.

The main rollout risk is worker memory use because the fold retains
state for distinct blob hashes and community bindings. The first
production-sized run must confirm peak memory and duration before the
schedule is trusted. The 10 million cap and one-hour deadline bound the
first deployment, but they do not replace measurement.

External mode also needs stale-snapshot, snapshot-load, and job-failure
monitors because the relay intentionally keeps publishing the last
complete result after a failed job or read error. This behavior avoids
gaps and partial data, while snapshot age and load health tell operators
when the result or PostgreSQL handoff needs attention. Job status
remains necessary for immediate failures and termination reasons that a
killed process cannot record itself.

## Testing

No production S3 scan was run from this PR.

Automated regressions verify that external mode keeps stale last-good
totals visible without emitting inline attempt-health gauges, reports
database-load failure while retaining the cache, and leaves inline
telemetry unchanged. Mutation checks also prove those tests fail if
external mode is routed through inline emission or a failed load is
reported as healthy.

After deployment, create one manual Job from the CronJob during a
staffed window. Require a completed Job, exactly one new snapshot, fewer
objects than the configured cap, acceptable duration and peak memory, no
relay restart regression, restored Datadog points, and no OpenMetrics
sample truncation.

## Next steps

1. Complete review and merge this PR.
2. Publish chart `0.1.9` and the final main-branch relay image.
3. Replace the provisional pins in
[squareup/builderbot-platform-core-infrastructure#244](https://github.com/squareup/builderbot-platform-core-infrastructure/pull/244),
rerender, and require clean CI.
4. Run the staffed manual Job gate and record object count, peak memory,
duration, snapshot revision, and relay stability.
5. Confirm the relay reads the new snapshot in `external` mode and the
existing `block.buzz_relay.*` storage series update again.
6. Enable the daily 03:17 UTC schedule and add dashboard panels and
monitors for snapshot age, snapshot-load health, Job failures, cap
pressure, duration, and worker memory.

If production must roll back to an older relay image, set
`BUZZ_STORAGE_METRICS=off` first. Older images do not understand
`external` and would resume the failing inline scan.

## Bigger picture

This is the application half of the storage-metrics repair. The paired
infrastructure PR owns the production resource limits, schedule,
artifact pins, secret boundary, and monitored rollout. Keeping those
changes separate lets this PR remain safe by default while still
providing a complete path to current per-community S3 totals.

Related issue: N/A. No matching issue or pull request exists in
`block/buzz`.

Generated with Codex

---------

Signed-off-by: Ravneet Arora <rarora@squareup.com>
Co-authored-by: Codex <noreply@openai.com>

Justfile

@@ -385,6 +385,12 @@ test-unit:
         # #[ignore]d, so --lib runs only the infra-free set. Without this gate a
         # stray file in migrations/ or a broken lint ships green.
         cargo nextest run -p buzz-db --lib
+        # Storage accounting crosses three crates whose focused regression
+        # suites are otherwise absent from the infra-free unit lane.
+        cargo nextest run -p buzz-media --lib \
+            -E 'test(=bucket_index::tests::bucket_snapshot_json_round_trip_preserves_community_keys)'
+        cargo nextest run -p buzz-admin \
+            -E 'test(=storage_snapshot_tests::failed_fold_never_invokes_snapshot_persistence)'
         # Multi-tenant conformance gate (buzz-conformance): the independent
         # replay checker + golden fixtures. No infra — pure in-process trace
         # replay — so it belongs in the unit job. Run all targets (lib + the
@@ -451,7 +457,7 @@ test-unit:
         # the ~30s sqlx acquire timeout, so they do not belong in the infra-free
         # unit job either.
         cargo nextest run -p buzz-relay --lib \
-            -E '(test(/^api::admin::/) - test(=api::admin::tests::disabled_mode_allows_unauthenticated_requests_on_the_admin_host) - test(=api::admin::tests::nip98_mode_unrostered_signer_does_not_consume_a_replay_slot)) + test(/^handlers::channel_authz::/) + test(/^handlers::moderation_authz::/) + test(/^handlers::side_effects::tests::/)'
+            -E '(test(/^api::admin::/) - test(=api::admin::tests::disabled_mode_allows_unauthenticated_requests_on_the_admin_host) - test(=api::admin::tests::nip98_mode_unrostered_signer_does_not_consume_a_replay_slot)) + test(/^handlers::channel_authz::/) + test(/^handlers::moderation_authz::/) + test(/^handlers::side_effects::tests::/) + test(/^storage_sweep::tests::/)'
         # ACP author-gate and queue tests protect the trust boundary between
         # relay events and agent prompts. They are infra-free; ignored lifecycle
         # tests remain excluded and run in their dedicated integration lanes.

crates/buzz-admin/src/main.rs

@@ -22,12 +22,16 @@
 
 mod deletions;
 
+use std::future::Future;
+use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::Arc;
+use std::time::Instant;
 
 use anyhow::Result;
 use buzz_core::kind::KIND_NIP43_MEMBERSHIP_LIST;
 use buzz_core::tenant::{relay_url_authority, TenantContext};
 use buzz_db::{Db, DbConfig};
+use buzz_media::{BucketSnapshot, MediaConfig, MediaStorage, S3AddressingStyle, SweepError};
 use buzz_pubsub::{EventTopic, PubSubManager};
 use clap::{Parser, Subcommand};
 use nostr::{EventBuilder, Keys, Kind, Tag};
@@ -78,6 +82,12 @@ enum Command {
     GenerateKey,
     /// Run pending database migrations.
     Migrate,
+    /// Compute one complete S3 storage snapshot and persist it for relay readers.
+    StorageSnapshot {
+        /// Abort before folding a page that would exceed this object count.
+        #[arg(long, default_value_t = 10_000_000)]
+        max_objects: u64,
+    },
     /// Inspect deployment-wide Buzz product feedback.
     ProductFeedback {
         #[command(subcommand)]
@@ -154,6 +164,7 @@ async fn run(cli: Cli) -> Result<i32> {
             println!("Database migrations complete.");
             Ok(0)
         }
+        Command::StorageSnapshot { max_objects } => cmd_storage_snapshot(max_objects).await,
         Command::AddMember { pubkey, role } => cmd_add_member(pubkey, role).await,
         Command::RemoveMember { pubkey, role } => cmd_remove_member(pubkey, role).await,
         Command::ListMembers => cmd_list_members().await,
@@ -168,6 +179,140 @@ async fn run(cli: Cli) -> Result<i32> {
     }
 }
 
+async fn cmd_storage_snapshot(max_objects: u64) -> Result<i32> {
+    let max_objects_db = i64::try_from(max_objects)
+        .map_err(|_| anyhow::anyhow!("--max-objects must be at most {}", i64::MAX))?;
+    if max_objects == 0 {
+        return Err(anyhow::anyhow!("--max-objects must be greater than zero"));
+    }
+
+    let db = connect_db().await?;
+    let mut leader = db.try_lock_storage_accounting().await?.ok_or_else(|| {
+        anyhow::anyhow!("another storage-snapshot worker already holds the lease")
+    })?;
+    let storage = Arc::new(MediaStorage::new(&storage_config_from_env()?)?);
+    let code_sha =
+        std::env::var("BUZZ_STORAGE_SNAPSHOT_CODE_SHA").unwrap_or_else(|_| "unknown".to_string());
+    if code_sha.is_empty() || code_sha.len() > 128 {
+        return Err(anyhow::anyhow!(
+            "BUZZ_STORAGE_SNAPSHOT_CODE_SHA must contain 1 to 128 bytes"
+        ));
+    }
+
+    println!(
+        "{}",
+        serde_json::json!({
+            "event": "storage_snapshot_started",
+            "max_objects": max_objects,
+            "code_sha": code_sha,
+        })
+    );
+    let run_started = Instant::now();
+    let listed_objects = Arc::new(AtomicU64::new(0));
+    let fold = buzz_media::fold_bucket_listing(max_objects, move |token| {
+        let storage = Arc::clone(&storage);
+        let listed_objects = Arc::clone(&listed_objects);
+        async move {
+            let page = storage.list_page(token, 1000).await?;
+            let page_objects = u64::try_from(page.objects.len()).unwrap_or(u64::MAX);
+            let before = listed_objects.fetch_add(page_objects, Ordering::Relaxed);
+            let after = before.saturating_add(page_objects);
+            if before / 100_000 != after / 100_000 {
+                println!(
+                    "{}",
+                    serde_json::json!({
+                        "event": "storage_snapshot_progress",
+                        "listed_objects": after,
+                        "max_objects": max_objects,
+                    })
+                );
+            }
+            Ok(page)
+        }
+    });
+    let snapshot_code_sha = code_sha.clone();
+    let persisted = persist_completed_fold(fold, move |encoded, duration_ms| async move {
+        leader
+            .save_snapshot(&encoded, duration_ms, max_objects_db, &snapshot_code_sha)
+            .await?;
+        Ok(())
+    })
+    .await;
+    let (snapshot, duration_ms) = match persisted {
+        Ok(completed) => completed,
+        Err(error) => {
+            println!(
+                "{}",
+                serde_json::json!({
+                    "event": "storage_snapshot_failed",
+                    "duration_ms": i64::try_from(run_started.elapsed().as_millis()).unwrap_or(i64::MAX),
+                    "max_objects": max_objects,
+                    "code_sha": code_sha,
+                    "error": error.to_string(),
+                })
+            );
+            return Err(error);
+        }
+    };
+    println!(
+        "{}",
+        serde_json::json!({
+            "event": "storage_snapshot_completed",
+            "duration_ms": duration_ms,
+            "listed_objects": snapshot.physical_objects,
+            "listed_bytes": snapshot.physical_bytes,
+            "logical_objects": snapshot.logical_objects,
+            "logical_bytes": snapshot.logical_bytes,
+            "max_objects": max_objects,
+            "code_sha": code_sha,
+        })
+    );
+    Ok(0)
+}
+
+async fn persist_completed_fold<FoldFuture, Persist, PersistFuture>(
+    fold: FoldFuture,
+    persist: Persist,
+) -> Result<(BucketSnapshot, i64)>
+where
+    FoldFuture: Future<Output = std::result::Result<BucketSnapshot, SweepError>>,
+    Persist: FnOnce(serde_json::Value, i64) -> PersistFuture,
+    PersistFuture: Future<Output = Result<()>>,
+{
+    let started = Instant::now();
+    let snapshot = fold.await?;
+    let duration_ms = i64::try_from(started.elapsed().as_millis()).unwrap_or(i64::MAX);
+    let encoded = serde_json::to_value(&snapshot)?;
+    persist(encoded, duration_ms).await?;
+    Ok((snapshot, duration_ms))
+}
+
+fn storage_config_from_env() -> Result<MediaConfig> {
+    let required = |name: &str| {
+        std::env::var(name).map_err(|_| anyhow::anyhow!("{name} must be set for storage-snapshot"))
+    };
+    let addressing_style = std::env::var("BUZZ_S3_ADDRESSING_STYLE")
+        .unwrap_or_else(|_| "path".to_string())
+        .parse::<S3AddressingStyle>()
+        .map_err(anyhow::Error::msg)?;
+    Ok(MediaConfig {
+        s3_endpoint: std::env::var("BUZZ_S3_ENDPOINT").unwrap_or_default(),
+        s3_access_key: std::env::var("BUZZ_S3_ACCESS_KEY").unwrap_or_default(),
+        s3_secret_key: std::env::var("BUZZ_S3_SECRET_KEY").unwrap_or_default(),
+        s3_bucket: required("BUZZ_S3_BUCKET")?,
+        s3_region: std::env::var("BUZZ_S3_REGION").unwrap_or_else(|_| "us-east-1".to_string()),
+        s3_addressing_style: addressing_style,
+        max_image_bytes: 1,
+        max_gif_bytes: 1,
+        max_video_bytes: 1,
+        max_file_bytes: 1,
+        public_base_url: "http://storage-snapshot.invalid/media".to_string(),
+        upload_records_enabled: false,
+        upload_ip_header: None,
+        upload_port_header: None,
+    })
+}
+
 async fn cmd_add_member(pubkey_arg: String, role: String) -> Result<i32> {
     if let Err(msg) = validate_role(&role) {
         eprintln!("error: {msg}");
@@ -631,3 +776,27 @@ async fn reconcile_channels(
     );
     Ok(())
 }
+
+#[cfg(test)]
+mod storage_snapshot_tests {
+    use std::sync::atomic::{AtomicUsize, Ordering};
+
+    use super::*;
+
+    #[tokio::test]
+    async fn failed_fold_never_invokes_snapshot_persistence() {
+        let persist_calls = Arc::new(AtomicUsize::new(0));
+        let observed_calls = Arc::clone(&persist_calls);
+        let result = persist_completed_fold(
+            async { Err::<BucketSnapshot, _>(SweepError::MalformedPage) },
+            move |_, _| async move {
+                observed_calls.fetch_add(1, Ordering::SeqCst);
+                Ok(())
+            },
+        )
+        .await;
+
+        assert!(result.is_err());
+        assert_eq!(persist_calls.load(Ordering::SeqCst), 0);
+    }
+}

crates/buzz-db/src/lib.rs

@@ -50,7 +50,7 @@ pub use store::{
     admin_moderation, allowlist, api_token, archived_identities, channel, channel_members,
     community, deletion, dm, event, feed, git_repo, moderation, partition, product_feedback, push,
     reaction, relay_admin_actions, relay_invite, relay_members, relay_operators, reminder,
-    replaceable, thread, usage, user, workflow,
+    replaceable, storage_accounting, thread, usage, user, workflow,
 };
 
 pub use allowlist::AllowlistEntry;

crates/buzz-db/src/runtime/migration.rs

@@ -489,6 +489,7 @@ mod postgres_tests {
             "relay_admin_actions",
             "relay_admin_outbox",
             "relay_operator_audit",
+            "storage_accounting_snapshots",
         ] {
             if normalized[insert_pos..].contains(&format!("'{value}'")) {
                 globals.insert(value.to_owned());
@@ -702,7 +703,7 @@ mod postgres_tests {
         let mut migrations: Vec<_> = MIGRATOR.iter().collect();
         migrations.sort_by_key(|migration| migration.version);
 
-        assert_eq!(migrations.len(), 45);
+        assert_eq!(migrations.len(), 46);
         assert_eq!(migrations[0].version, 1);
         assert_eq!(&*migrations[0].description, "initial schema");
         assert!(migrations[0]
@@ -1282,6 +1283,11 @@ mod postgres_tests {
         // The restored exclusion function must NOT list any NIP-FI relation.
         assert!(!ledger_removal.contains("'authorization_operation_receipts'"));
         assert!(!ledger_removal.contains("'identity_bindings'"));
+        assert_eq!(migrations[45].version, 46);
+        assert!(migrations[45]
+            .sql
+            .as_str()
+            .contains("CREATE TABLE storage_accounting_snapshots"));
         // schema.sql exclusion list must match the restored (pre-0041) body.
         assert!(
             desired_schema.contains("'rate_limit_violations'\n    ]::TEXT[])"),

crates/buzz-db/src/store/mod.rs

@@ -46,6 +46,8 @@ pub mod relay_operators;
 pub mod reminder;
 /// Replaceable-event persistence and coordinate locking.
 pub mod replaceable;
+/// Durable completed snapshots from the isolated media-storage worker.
+pub mod storage_accounting;
 /// Thread metadata persistence.
 pub mod thread;
 /// Per-community usage rollup queries for Prometheus gauges.

crates/buzz-db/src/store/storage_accounting.rs

@@ -0,0 +1,257 @@
+//! Durable handoff for isolated media-storage accounting.
+
+use buzz_datastore_tracing::datastore_span;
+use chrono::{DateTime, Utc};
+use sqlx::postgres::PgConnection;
+
+use crate::{observability, Db, Result};
+
+/// Deployment-global advisory lock for the run-once storage worker.
+pub const STORAGE_ACCOUNTING_LOCK_KEY: i64 = 0x4255_5a5a_5354_4f52;
+
+/// Owns the detached PostgreSQL session that excludes overlapping workers.
+pub struct StorageAccountingLeader {
+    connection: PgConnection,
+}
+
+/// The newest complete snapshot stored by the worker.
+#[derive(Debug, Clone)]
+pub struct StoredStorageSnapshot {
+    /// Serialized `buzz_media::BucketSnapshot`.
+    pub snapshot: serde_json::Value,
+    /// Database commit time for freshness reporting.
+    pub completed_at: DateTime<Utc>,
+    /// Wall-clock duration of the successful S3 fold.
+    pub duration_ms: i64,
+    /// Object ceiling configured for that run.
+    pub max_objects: i64,
+    /// Image or source revision supplied by the worker.
+    pub code_sha: String,
+}
+
+impl StorageAccountingLeader {
+    /// Atomically replace the singleton through the lock-owning session.
+    #[datastore_span(name = "save_storage_accounting_snapshot", system = "postgresql")]
+    pub async fn save_snapshot(
+        &mut self,
+        snapshot: &serde_json::Value,
+        duration_ms: i64,
+        max_objects: i64,
+        code_sha: &str,
+    ) -> Result<()> {
+        sqlx::query(
+            "INSERT INTO storage_accounting_snapshots \
+             (singleton, snapshot, completed_at, duration_ms, max_objects, code_sha) \
+             VALUES (TRUE, $1, transaction_timestamp(), $2, $3, $4) \
+             ON CONFLICT (singleton) DO UPDATE SET \
+               snapshot = EXCLUDED.snapshot, \
+               completed_at = EXCLUDED.completed_at, \
+               duration_ms = EXCLUDED.duration_ms, \
+               max_objects = EXCLUDED.max_objects, \
+               code_sha = EXCLUDED.code_sha",
+        )
+        .bind(snapshot)
+        .bind(duration_ms)
+        .bind(max_objects)
+        .bind(code_sha)
+        .execute(&mut self.connection)
+        .await?;
+        Ok(())
+    }
+}
+
+impl Db {
+    /// Try to acquire the deployment-global storage-worker lease.
+    #[datastore_span(name = "try_lock_storage_accounting", system = "postgresql")]
+    pub async fn try_lock_storage_accounting(&self) -> Result<Option<StorageAccountingLeader>> {
+        let mut connection = observability::acquire_writer_with_legacy_metrics(
+            &self.pool,
+            observability::WriterOperation::Maintenance,
+        )
+        .await?;
+        let acquired = sqlx::query_scalar::<_, bool>("SELECT pg_try_advisory_lock($1)")
+            .bind(STORAGE_ACCOUNTING_LOCK_KEY)
+            .fetch_one(&mut *connection)
+            .await?;
+        Ok(acquired.then(|| StorageAccountingLeader {
+            connection: connection.detach(),
+        }))
+    }
+
+    /// Load the newest complete worker snapshot from the writer.
+    #[datastore_span(name = "load_storage_accounting_snapshot", system = "postgresql")]
+    pub async fn load_storage_accounting_snapshot(&self) -> Result<Option<StoredStorageSnapshot>> {
+        let mut connection =
+            observability::acquire_writer(&self.pool, observability::WriterOperation::Maintenance)
+                .await?;
+        let row = sqlx::query_as::<_, (serde_json::Value, DateTime<Utc>, i64, i64, String)>(
+            "SELECT snapshot, completed_at, duration_ms, max_objects, code_sha \
+             FROM storage_accounting_snapshots WHERE singleton = TRUE",
+        )
+        .fetch_optional(&mut *connection)
+        .await?;
+        Ok(row.map(
+            |(snapshot, completed_at, duration_ms, max_objects, code_sha)| StoredStorageSnapshot {
+                snapshot,
+                completed_at,
+                duration_ms,
+                max_objects,
+                code_sha,
+            },
+        ))
+    }
+}
+
+#[cfg(test)]
+mod postgres_tests {
+    use super::*;
+    use sqlx::postgres::PgPoolOptions;
+    use sqlx::PgPool;
+    use uuid::Uuid;
+
+    async fn create_scratch_db(admin: &PgPool) -> (PgPool, String) {
+        let name = format!("storage_accounting_{}", Uuid::new_v4().simple());
+        sqlx::query(sqlx::AssertSqlSafe(format!("CREATE DATABASE {name}")))
+            .execute(admin)
+            .await
+            .expect("create scratch db");
+        let base = crate::test_support::database_url();
+        let idx = base.rfind('/').expect("db url has path");
+        let pool = PgPool::connect(&format!("{}/{name}", &base[..idx]))
+            .await
+            .expect("connect scratch db");
+        crate::migration::run_migrations(&pool)
+            .await
+            .expect("migrate scratch db");
+        (pool, name)
+    }
+
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn complete_snapshot_replaces_atomically_and_worker_lock_excludes_overlap() {
+        let admin = PgPoolOptions::new()
+            .max_connections(1)
+            .connect(&crate::test_support::database_url())
+            .await
+            .expect("connect admin");
+        let (pool, name) = create_scratch_db(&admin).await;
+        let first = Db::from_pool(pool.clone());
+        let second = Db::from_pool(pool.clone());
+
+        let mut leader = first
+            .try_lock_storage_accounting()
+            .await
+            .expect("first lock")
+            .expect("first worker owns lock");
+        assert!(
+            second
+                .try_lock_storage_accounting()
+                .await
+                .expect("second lock")
+                .is_none(),
+            "overlapping worker must not start"
+        );
+
+        leader
+            .save_snapshot(&serde_json::json!({"version": 1}), 10, 100, "a")
+            .await
+            .expect("save first snapshot");
+        leader
+            .save_snapshot(&serde_json::json!({"version": 2}), 20, 200, "b")
+            .await
+            .expect("replace snapshot");
+        let stored = second
+            .load_storage_accounting_snapshot()
+            .await
+            .expect("load snapshot")
+            .expect("snapshot exists");
+        assert_eq!(stored.snapshot, serde_json::json!({"version": 2}));
+        assert_eq!(stored.duration_ms, 20);
+        assert_eq!(stored.max_objects, 200);
+        assert_eq!(stored.code_sha, "b");
+
+        drop(leader);
+        drop(first);
+        drop(second);
+        pool.close().await;
+        let _ = sqlx::query(sqlx::AssertSqlSafe(format!(
+            "DROP DATABASE IF EXISTS {name} WITH (FORCE)"
+        )))
+        .execute(&admin)
+        .await;
+    }
+
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn lost_lock_session_cannot_overwrite_successor_snapshot() {
+        let admin = PgPoolOptions::new()
+            .max_connections(1)
+            .connect(&crate::test_support::database_url())
+            .await
+            .expect("connect admin");
+        let (pool, name) = create_scratch_db(&admin).await;
+        let first = Db::from_pool(pool.clone());
+        let second = Db::from_pool(pool.clone());
+
+        let mut stale_leader = first
+            .try_lock_storage_accounting()
+            .await
+            .expect("first lock")
+            .expect("first worker owns lock");
+        let stale_backend_pid = sqlx::query_scalar::<_, i32>("SELECT pg_backend_pid()")
+            .fetch_one(&mut stale_leader.connection)
+            .await
+            .expect("load first worker backend pid");
+        let terminated = sqlx::query_scalar::<_, bool>("SELECT pg_terminate_backend($1)")
+            .bind(stale_backend_pid)
+            .fetch_one(&admin)
+            .await
+            .expect("terminate first worker backend");
+        assert!(terminated, "first worker backend must terminate");
+
+        let mut fresh_leader = tokio::time::timeout(std::time::Duration::from_secs(5), async {
+            loop {
+                if let Some(leader) = second
+                    .try_lock_storage_accounting()
+                    .await
+                    .expect("successor lock attempt")
+                {
+                    break leader;
+                }
+                tokio::time::sleep(std::time::Duration::from_millis(25)).await;
+            }
+        })
+        .await
+        .expect("successor acquires released lock");
+
+        fresh_leader
+            .save_snapshot(&serde_json::json!({"worker": "fresh"}), 20, 200, "b")
+            .await
+            .expect("successor publishes fresh snapshot");
+        stale_leader
+            .save_snapshot(&serde_json::json!({"worker": "stale"}), 30, 100, "a")
+            .await
+            .expect_err("worker that lost its lock session cannot publish");
+
+        let stored = second
+            .load_storage_accounting_snapshot()
+            .await
+            .expect("load snapshot")
+            .expect("snapshot exists");
+        assert_eq!(stored.snapshot, serde_json::json!({"worker": "fresh"}));
+        assert_eq!(stored.duration_ms, 20);
+        assert_eq!(stored.max_objects, 200);
+        assert_eq!(stored.code_sha, "b");
+
+        drop(stale_leader);
+        drop(fresh_leader);
+        drop(first);
+        drop(second);
+        pool.close().await;
+        let _ = sqlx::query(sqlx::AssertSqlSafe(format!(
+            "DROP DATABASE IF EXISTS {name} WITH (FORCE)"
+        )))
+        .execute(&admin)
+        .await;
+    }
+}

crates/buzz-media/src/bucket_index.rs

@@ -193,7 +193,7 @@ fn parse_auxiliary_key(key: &str) -> Option<(Uuid, String, String)> {
 }
 
 /// Per-community logical storage: bytes and object count of bound shas.
-#[derive(Debug, Clone, Default, PartialEq, Eq)]
+#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
 pub struct CommunityStorage {
     pub bytes: u64,
     pub objects: u64,
@@ -202,7 +202,7 @@ pub struct CommunityStorage {
 /// The full computed sweep result: fleet physical/logical totals,
 /// per-community logical breakdown, and anomaly/visibility gauges. Pure
 /// data — no I/O, cheap to clone into a cached snapshot.
-#[derive(Debug, Clone, Default, PartialEq, Eq)]
+#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
 pub struct BucketSnapshot {
     /// Every listed object, every class (kind=physical).
     pub physical_bytes: u64,
@@ -422,6 +422,27 @@ mod tests {
         Uuid::from_u128(n)
     }
 
+    #[test]
+    fn bucket_snapshot_json_round_trip_preserves_community_keys() {
+        let community = community(7);
+        let mut snapshot = BucketSnapshot {
+            physical_objects: 3,
+            ..BucketSnapshot::default()
+        };
+        snapshot.per_community.insert(
+            community,
+            CommunityStorage {
+                bytes: 42,
+                objects: 1,
+            },
+        );
+
+        let encoded = serde_json::to_value(&snapshot).expect("serialize snapshot");
+        let decoded: BucketSnapshot =
+            serde_json::from_value(encoded).expect("deserialize snapshot");
+        assert_eq!(decoded, snapshot);
+    }
+
     // --- classify_key ---
 
     #[test]

crates/buzz-relay/src/main.rs

@@ -1764,27 +1764,62 @@ async fn run_storage_sweep_tick(
     // leader tick, and stable for the process lifetime (env is immutable).
     // Keeping it here avoids widening Config/AppState for a single consumer.
     let config = *SWEEP_CONFIG.get_or_init(storage_sweep::StorageSweepConfig::from_env);
-    if !config.enabled {
+    match config.mode {
-        return;
+        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 media_storage = Arc::clone(&state.media_storage);
+    storage_sweep::emit_storage_metrics(&state.storage_sweep, config.mode, host_map, |id| {
-    let max_objects = config.max_objects;
-    storage_sweep::maybe_spawn_sweep(
-        &state.storage_sweep,
-        config.interval,
-        config.timeout,
-        async move {
-            buzz_media::fold_bucket_listing(max_objects, move |token| {
-                let media_storage = Arc::clone(&media_storage);
-                async move { media_storage.list_page(token, 1000).await }
-            })
-            .await
-        },
-    )
-    .await;
-
-    storage_sweep::emit_storage_metrics(&state.storage_sweep, host_map, |id| {
         emission_scope.allows(id)
     })
     .await;

crates/buzz-relay/src/storage_sweep.rs

@@ -12,21 +12,33 @@
 //!   re-publishes the cached snapshot regardless of whether a sweep is
 //!   currently running — DB-derived gauges keep their configured cadence,
 //!   storage gauges lag by at most one tick after a sweep completes.
-//! - A cold cache (no sweep has ever succeeded) publishes health gauges only
+//! - In inline mode, a cold cache (no sweep has ever succeeded) publishes
-//!   (`sweep_ok=0`); a warm cache re-publishes the last good snapshot even
+//!   health gauges only (`sweep_ok=0`); a warm cache re-publishes the last good
-//!   while the newest attempt is failing, so a transient S3 blip never blanks
+//!   snapshot even while the newest attempt is failing, so a transient S3 blip
-//!   the dashboards.
+//!   never blanks the dashboards.
 
 use std::collections::{HashMap, HashSet};
 use std::future::Future;
 use std::time::{Duration, Instant};
 
+use chrono::{DateTime, Utc};
 use tokio::sync::Mutex;
 use tokio::task::JoinHandle;
 use uuid::Uuid;
 
 use buzz_media::{BucketSnapshot, SweepError};
 
+/// Where relay storage gauges get their snapshot.
+#[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.
+    Disabled,
+}
+
 /// Sweep knobs, read once at boot. See `PLANS/S3_STORAGE_METRICS_PLAN.md` F7.
 #[derive(Debug, Clone, Copy)]
 pub struct StorageSweepConfig {
@@ -40,18 +52,16 @@ pub struct StorageSweepConfig {
     /// Cumulative listed-object cap; a listing that exceeds it fails the
     /// attempt (old snapshot kept) rather than growing memory unbounded.
     pub max_objects: u64,
-    /// Kill switch. `false` ⇒ no sweep ever spawns and no storage-family
+    /// Snapshot source and kill switch.
-    /// gauge (including the health gauges) is ever emitted — a relay whose
+    pub mode: StorageMetricsMode,
-    /// deployment lacks `s3:ListBucket` can turn the whole feature off.
-    pub enabled: bool,
 }
 
 impl StorageSweepConfig {
     /// Reads `BUZZ_STORAGE_SWEEP_INTERVAL_SECS` (default 3600, floor 60),
     /// `BUZZ_STORAGE_SWEEP_TIMEOUT_SECS` (default 120),
-    /// `BUZZ_STORAGE_SWEEP_MAX_OBJECTS` (default 1_000_000), and the
+    /// `BUZZ_STORAGE_SWEEP_MAX_OBJECTS` (default 1_000_000), and
-    /// `BUZZ_STORAGE_METRICS` kill switch (`off` ⇒ disabled, anything else
+    /// `BUZZ_STORAGE_METRICS` (`inline`, `external`, or `off`). Unset keeps
-    /// including unset ⇒ enabled).
+    /// the legacy inline behavior; unknown values fail closed.
     pub fn from_env() -> Self {
         let interval_secs = std::env::var("BUZZ_STORAGE_SWEEP_INTERVAL_SECS")
             .ok()
@@ -66,16 +76,30 @@ impl StorageSweepConfig {
             .ok()
             .and_then(|v| v.parse::<u64>().ok())
             .unwrap_or(1_000_000);
-        let enabled = std::env::var("BUZZ_STORAGE_METRICS")
+        let raw_mode = std::env::var("BUZZ_STORAGE_METRICS")
             .ok()
-            .map(|v| v.trim().to_ascii_lowercase())
+            .map(|v| v.trim().to_ascii_lowercase());
-            .as_deref()
+        let mode = parse_storage_metrics_mode(raw_mode.as_deref());
-            != Some("off");
         Self {
             interval: Duration::from_secs(interval_secs),
             timeout: Duration::from_secs(timeout_secs),
             max_objects,
-            enabled,
+            mode,
+        }
+    }
+}
+
+fn parse_storage_metrics_mode(value: Option<&str>) -> StorageMetricsMode {
+    match value {
+        Some("off") => StorageMetricsMode::Disabled,
+        Some("snapshot") | Some("external") => StorageMetricsMode::Snapshot,
+        Some("inline") | Some("on") | None => StorageMetricsMode::Inline,
+        Some(other) => {
+            tracing::error!(
+                value = other,
+                "invalid BUZZ_STORAGE_METRICS value; storage metrics disabled"
+            );
+            StorageMetricsMode::Disabled
         }
     }
 }
@@ -89,11 +113,14 @@ struct LastAttempt {
     duration: Duration,
 }
 
-/// The most recent *successful* sweep, cut as one coherent snapshot.
+/// The most recent successfully computed snapshot.
 #[derive(Debug, Clone)]
 struct CachedSnapshot {
     data: BucketSnapshot,
     completed_at: Instant,
+    completed_at_wall: DateTime<Utc>,
+    duration: Duration,
+    max_objects: Option<u64>,
 }
 
 /// What a spawned sweep task hands back to the tick that harvests it.
@@ -141,6 +168,8 @@ pub struct StorageSweepState {
     cached: Option<CachedSnapshot>,
     last_attempt: Option<LastAttempt>,
     failures_total: u64,
+    /// Whether the latest external-mode read yielded a decodable snapshot.
+    persisted_load_ok: Option<bool>,
     /// 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<StorageEmittedKey>,
@@ -214,6 +243,9 @@ pub async fn maybe_spawn_sweep<Fut>(
                             // 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) => {
@@ -258,10 +290,53 @@ pub async fn maybe_spawn_sweep<Fut>(
     state.in_flight = Some(handle);
 }
 
-/// Emit the storage-family gauges from the cached snapshot. Call every usage
+/// Replace the cache with a complete snapshot loaded from durable storage.
-/// tick, leader-only (mirrors `emit_db_usage_metrics`'s leadership gate) —
+pub async fn cache_persisted_snapshot(
-/// never from the spawned sweep task itself, so a sweep that completes after
+    state: &Mutex<StorageSweepState>,
-/// this pod loses leadership parks its snapshot without ever publishing it.
+    snapshot: BucketSnapshot,
+    completed_at: DateTime<Utc>,
+    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),
+    });
+    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<StorageSweepState>) {
+    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(
+    state: &Mutex<StorageSweepState>,
+    mode: StorageMetricsMode,
+    host_map: &HashMap<Uuid, String>,
+    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 => {}
+    }
+}
+
+/// 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
@@ -276,7 +351,7 @@ pub async fn maybe_spawn_sweep<Fut>(
 /// would linger at its last nonzero value until the recorder's idle eviction
 /// fires (≥3 ticks), producing a transient double-count against the
 /// `buzz_storage_unmapped_community_bytes` gauge.
-pub async fn emit_storage_metrics(
+async fn emit_inline_storage_metrics(
     state: &Mutex<StorageSweepState>,
     host_map: &HashMap<Uuid, String>,
     allows: impl Fn(&Uuid) -> bool,
@@ -293,6 +368,34 @@ pub async fn emit_storage_metrics(
         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<StorageSweepState>,
+    host_map: &HashMap<Uuid, String>,
+    allows: impl Fn(&Uuid) -> bool,
+) {
+    let mut state = state.lock().await;
+    let load_ok = state.persisted_load_ok.unwrap_or(false);
+    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,
+}
+
+fn emit_cached_storage_metrics(
+    state: &mut StorageSweepState,
+    host_map: &HashMap<Uuid, String>,
+    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.
@@ -302,8 +405,20 @@ pub async fn emit_storage_metrics(
         }
         return;
     };
-    metrics::gauge!("buzz_storage_sweep_age_seconds")
+    let age = Utc::now()
-        .set(cached.completed_at.elapsed().as_secs_f64());
+        .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());
+        }
+    }
 
     let snapshot = &cached.data;
     metrics::gauge!("buzz_total_storage_bytes", "kind" => "physical")
@@ -322,6 +437,11 @@ pub async fn emit_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);
+    }
 
     let mut current = HashSet::new();
     let mut unmapped_bytes = 0u64;
@@ -398,20 +518,36 @@ mod tests {
         assert_eq!(config.interval, Duration::from_secs(3600));
         assert_eq!(config.timeout, Duration::from_secs(120));
         assert_eq!(config.max_objects, 1_000_000);
-        assert!(config.enabled);
+        assert_eq!(config.mode, StorageMetricsMode::Inline);
     }
 
     #[test]
-    fn config_kill_switch_only_off_disables() {
+    fn config_mode_accepts_snapshot_and_fails_closed_on_unknown_values() {
-        assert!(!parse_enabled(Some("off")));
+        assert_eq!(
-        assert!(!parse_enabled(Some("OFF")));
+            parse_storage_metrics_mode(Some("off")),
-        assert!(parse_enabled(Some("on")));
+            StorageMetricsMode::Disabled
-        assert!(parse_enabled(Some("anything-else")));
+        );
-        assert!(parse_enabled(None));
+        assert_eq!(
-    }
+            parse_storage_metrics_mode(Some("snapshot")),
-
+            StorageMetricsMode::Snapshot
-    fn parse_enabled(value: Option<&str>) -> bool {
+        );
-        value.map(str::trim).map(str::to_ascii_lowercase).as_deref() != Some("off")
+        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 ---
@@ -437,6 +573,9 @@ mod tests {
         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,
@@ -456,6 +595,9 @@ mod tests {
         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,
@@ -711,7 +853,12 @@ mod tests {
         let recorder = DebuggingRecorder::new();
         let host_map = HashMap::new();
         metrics::with_local_recorder(&recorder, || {
-            futures::executor::block_on(emit_storage_metrics(&state, &host_map, |_| true));
+            futures::executor::block_on(emit_storage_metrics(
+                &state,
+                StorageMetricsMode::Inline,
+                &host_map,
+                |_| true,
+            ));
         });
 
         let values = gauge_snapshot(&recorder);
@@ -764,7 +911,12 @@ mod tests {
             let recorder = DebuggingRecorder::new();
             let host_map = HashMap::new();
             metrics::with_local_recorder(&recorder, || {
-                futures::executor::block_on(emit_storage_metrics(&state, &host_map, |_| true));
+                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));
@@ -785,7 +937,12 @@ mod tests {
         let recorder = DebuggingRecorder::new();
         let host_map = HashMap::new();
         metrics::with_local_recorder(&recorder, || {
-            futures::executor::block_on(emit_storage_metrics(&state, &host_map, |_| true));
+            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));
@@ -814,7 +971,12 @@ mod tests {
         let host_map = HashMap::new();
 
         metrics::with_local_recorder(&recorder, || {
-            futures::executor::block_on(emit_storage_metrics(&state, &host_map, |_| true));
+            futures::executor::block_on(emit_storage_metrics(
+                &state,
+                StorageMetricsMode::Inline,
+                &host_map,
+                |_| true,
+            ));
         });
 
         let values = gauge_snapshot(&recorder);
@@ -864,6 +1026,9 @@ mod tests {
             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,
@@ -878,9 +1043,12 @@ mod tests {
 
         let recorder = DebuggingRecorder::new();
         metrics::with_local_recorder(&recorder, || {
-            futures::executor::block_on(emit_storage_metrics(&state, &host_map, |id| {
+            futures::executor::block_on(emit_storage_metrics(
-                *id != excluded
+                &state,
-            }));
+                StorageMetricsMode::Inline,
+                &host_map,
+                |id| *id != excluded,
+            ));
         });
 
         let values = gauge_snapshot(&recorder);
@@ -973,6 +1141,9 @@ mod tests {
             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,
@@ -989,7 +1160,12 @@ mod tests {
         host_map_1.insert(community_b, "host.old".to_string());
         host_map_1.insert(community_c, "host.c".to_string());
         metrics::with_local_recorder(&recorder, || {
-            futures::executor::block_on(emit_storage_metrics(&state, &host_map_1, |_| true));
+            futures::executor::block_on(emit_storage_metrics(
+                &state,
+                StorageMetricsMode::Inline,
+                &host_map_1,
+                |_| true,
+            ));
         });
         {
             let labeled = labeled_community_gauges(&recorder);
@@ -1017,6 +1193,9 @@ mod tests {
             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,
             });
         }
         let mut host_map_2 = HashMap::new();
@@ -1026,6 +1205,7 @@ mod tests {
         metrics::with_local_recorder(&recorder, || {
             futures::executor::block_on(emit_storage_metrics(
                 &state,
+                StorageMetricsMode::Inline,
                 &host_map_2,
                 |id| *id != community_c, // (c) scope-excluded
             ));
@@ -1087,4 +1267,112 @@ mod tests {
             "(c) scope removal: host.c objects must be zeroed"
         );
     }
+
+    #[tokio::test]
+    async fn persisted_snapshot_emits_snapshot_health_without_attempt_health() {
+        let community = Uuid::from_u128(77);
+        let state = Mutex::new(StorageSweepState::default());
+        cache_persisted_snapshot(
+            &state,
+            snapshot_with(community, 100, 25),
+            Utc::now() - chrono::Duration::seconds(30),
+            Duration::from_secs(12),
+            100,
+        )
+        .await;
+        let host_map = HashMap::from([(community, "example.test".to_string())]);
+        let recorder = DebuggingRecorder::new();
+        metrics::with_local_recorder(&recorder, || {
+            futures::executor::block_on(emit_storage_metrics(
+                &state,
+                StorageMetricsMode::Snapshot,
+                &host_map,
+                |_| true,
+            ));
+        });
+        let values = gauge_snapshot(&recorder);
+        assert_eq!(values.get("buzz_storage_snapshot_load_ok"), Some(&1.0));
+        assert_eq!(
+            values.get("buzz_storage_snapshot_duration_seconds"),
+            Some(&12.0)
+        );
+        assert_eq!(
+            values.get("buzz_storage_snapshot_max_objects"),
+            Some(&100.0)
+        );
+        assert_eq!(
+            values.get("buzz_storage_snapshot_cap_utilization"),
+            Some(&0.25)
+        );
+        assert!(values["buzz_storage_snapshot_age_seconds"] >= 30.0);
+        assert!(!values.contains_key("buzz_storage_sweep_ok"));
+        assert!(!values.contains_key("buzz_storage_sweep_failures"));
+        assert!(!values.contains_key("buzz_storage_sweep_duration_seconds"));
+        assert!(!values.contains_key("buzz_storage_sweep_age_seconds"));
+    }
+
+    #[tokio::test]
+    async fn stale_persisted_snapshot_remains_visible_without_claiming_attempt_success() {
+        let community = Uuid::from_u128(78);
+        let state = Mutex::new(StorageSweepState::default());
+        cache_persisted_snapshot(
+            &state,
+            snapshot_with(community, 200, 50),
+            Utc::now() - chrono::Duration::hours(48),
+            Duration::from_secs(20),
+            1_000,
+        )
+        .await;
+
+        let host_map = HashMap::from([(community, "stale.example.test".to_string())]);
+        let recorder = DebuggingRecorder::new();
+        metrics::with_local_recorder(&recorder, || {
+            futures::executor::block_on(emit_storage_metrics(
+                &state,
+                StorageMetricsMode::Snapshot,
+                &host_map,
+                |_| true,
+            ));
+        });
+
+        let values = gauge_snapshot(&recorder);
+        assert_eq!(values.get("buzz_storage_snapshot_load_ok"), Some(&1.0));
+        assert_eq!(values.get("buzz_total_storage_bytes"), Some(&200.0));
+        assert!(values["buzz_storage_snapshot_age_seconds"] >= 48.0 * 60.0 * 60.0);
+        assert!(!values.contains_key("buzz_storage_sweep_ok"));
+        assert!(!values.contains_key("buzz_storage_sweep_failures"));
+    }
+
+    #[tokio::test]
+    async fn failed_persisted_load_keeps_last_good_totals_and_reports_unhealthy_handoff() {
+        let community = Uuid::from_u128(79);
+        let state = Mutex::new(StorageSweepState::default());
+        cache_persisted_snapshot(
+            &state,
+            snapshot_with(community, 300, 75),
+            Utc::now() - chrono::Duration::minutes(5),
+            Duration::from_secs(30),
+            1_000,
+        )
+        .await;
+        record_persisted_snapshot_load_failure(&state).await;
+
+        let host_map = HashMap::from([(community, "cached.example.test".to_string())]);
+        let recorder = DebuggingRecorder::new();
+        metrics::with_local_recorder(&recorder, || {
+            futures::executor::block_on(emit_storage_metrics(
+                &state,
+                StorageMetricsMode::Snapshot,
+                &host_map,
+                |_| true,
+            ));
+        });
+
+        let values = gauge_snapshot(&recorder);
+        assert_eq!(values.get("buzz_storage_snapshot_load_ok"), Some(&0.0));
+        assert_eq!(values.get("buzz_total_storage_bytes"), Some(&300.0));
+        assert!(values["buzz_storage_snapshot_age_seconds"] >= 5.0 * 60.0);
+        assert!(!values.contains_key("buzz_storage_sweep_ok"));
+        assert!(!values.contains_key("buzz_storage_sweep_failures"));
+    }
 }

deploy/charts/buzz/Chart.yaml

@@ -7,7 +7,7 @@ description: |
   PostgreSQL and Redis. Configurable for single-node evaluation
   (subcharts on) and HA production (external services, existingSecret).
 type: application
-version: 0.1.8
+version: 0.1.9
 appVersion: "0.1.0"
 home: https://github.com/block/buzz
 sources:
@@ -24,7 +24,7 @@ maintainers:
 annotations:
   artifacthub.io/changes: |
     - kind: added
-      description: Optional immutable relay image digest pinning with backwards-compatible tag fallback.
+      description: Optional isolated storage-accounting CronJob with durable relay snapshots.
   artifacthub.io/license: Apache-2.0
 
 # Optional eval-only subcharts. Production deploys disable both and point

deploy/charts/buzz/templates/_helpers.tpl

@@ -61,6 +61,14 @@ app.kubernetes.io/component: relay
 {{- end -}}
 {{- end -}}
 
+{{- define "buzz.imageRevision" -}}
+{{- if .Values.image.digest -}}
+{{- .Values.image.digest -}}
+{{- else -}}
+{{- default .Chart.AppVersion .Values.image.tag -}}
+{{- end -}}
+{{- end -}}
+
 {{/*
 Name of the chart-managed Secret holding relay-identity material and any
 chart-composed connection strings.

deploy/charts/buzz/templates/deployment.yaml

@@ -172,6 +172,7 @@ spec:
             - { name: BUZZ_S3_BUCKET, value: {{ .Values.s3.bucket | quote }} }
             - { name: BUZZ_S3_REGION, value: {{ .Values.s3.region | quote }} }
             - { name: BUZZ_S3_ADDRESSING_STYLE, value: {{ .Values.s3.addressingStyle | quote }} }
+            - { name: BUZZ_STORAGE_METRICS, value: {{ .Values.storageAccounting.relayMode | quote }} }
 
             # ── Secrets (from chart-managed or existing) ─────────────
             - name: BUZZ_RELAY_PRIVATE_KEY

deploy/charts/buzz/templates/storage-accounting-cronjob.yaml

@@ -0,0 +1,74 @@
+{{- if .Values.storageAccounting.enabled }}
+apiVersion: batch/v1
+kind: CronJob
+metadata:
+  name: {{ include "buzz.fullname" . }}-storage-accounting
+  labels:
+    {{- include "buzz.labels" . | nindent 4 }}
+    app.kubernetes.io/component: storage-accounting
+spec:
+  schedule: {{ .Values.storageAccounting.schedule | quote }}
+  concurrencyPolicy: Forbid
+  successfulJobsHistoryLimit: {{ .Values.storageAccounting.successfulJobsHistoryLimit }}
+  failedJobsHistoryLimit: {{ .Values.storageAccounting.failedJobsHistoryLimit }}
+  jobTemplate:
+    spec:
+      activeDeadlineSeconds: {{ .Values.storageAccounting.activeDeadlineSeconds }}
+      backoffLimit: 0
+      template:
+        metadata:
+          labels:
+            {{- include "buzz.selectorLabels" . | nindent 12 }}
+            app.kubernetes.io/component: storage-accounting
+            {{- with .Values.storageAccounting.podLabels }}
+            {{- toYaml . | nindent 12 }}
+            {{- end }}
+          annotations:
+            {{- toYaml .Values.storageAccounting.podAnnotations | nindent 12 }}
+        spec:
+          restartPolicy: Never
+          serviceAccountName: {{ default (include "buzz.serviceAccountName" .) .Values.storageAccounting.serviceAccountName }}
+          securityContext:
+            {{- toYaml .Values.relay.securityContext | nindent 12 }}
+          {{- with .Values.image.pullSecrets }}
+          imagePullSecrets:
+            {{- toYaml . | nindent 12 }}
+          {{- end }}
+          containers:
+            - name: storage-accounting
+              image: {{ include "buzz.image" . }}
+              imagePullPolicy: {{ .Values.image.pullPolicy }}
+              securityContext:
+                {{- omit .Values.relay.containerSecurityContext "readOnlyRootFilesystem" | toYaml | nindent 16 }}
+                readOnlyRootFilesystem: true
+              command: ["/usr/local/bin/buzz-admin"]
+              args: ["storage-snapshot", "--max-objects", {{ .Values.storageAccounting.maxObjects | int64 | quote }}]
+              env:
+                {{- $s3Endpoint := include "buzz.s3Endpoint" . }}
+                {{- if $s3Endpoint }}
+                - { name: BUZZ_S3_ENDPOINT, value: {{ $s3Endpoint | quote }} }
+                {{- end }}
+                - { name: BUZZ_S3_BUCKET, value: {{ .Values.s3.bucket | quote }} }
+                - { name: BUZZ_S3_REGION, value: {{ .Values.s3.region | quote }} }
+                - { name: BUZZ_S3_ADDRESSING_STYLE, value: {{ .Values.s3.addressingStyle | quote }} }
+                - { name: BUZZ_STORAGE_SNAPSHOT_CODE_SHA, value: {{ include "buzz.imageRevision" . | quote }} }
+                - name: DATABASE_URL
+                  valueFrom:
+                    secretKeyRef:
+                      name: {{ include "buzz.envSecretName" . }}
+                      key: DATABASE_URL
+                - name: BUZZ_S3_ACCESS_KEY
+                  valueFrom:
+                    secretKeyRef:
+                      name: {{ include "buzz.envSecretName" . }}
+                      key: BUZZ_S3_ACCESS_KEY
+                      optional: true
+                - name: BUZZ_S3_SECRET_KEY
+                  valueFrom:
+                    secretKeyRef:
+                      name: {{ include "buzz.envSecretName" . }}
+                      key: BUZZ_S3_SECRET_KEY
+                      optional: true
+              resources:
+                {{- toYaml .Values.storageAccounting.resources | nindent 16 }}
+{{- end }}

deploy/charts/buzz/tests/storage_accounting_test.yaml

@@ -0,0 +1,139 @@
+suite: isolated storage accounting
+templates:
+  - templates/storage-accounting-cronjob.yaml
+  - templates/deployment.yaml
+  - templates/secret-chart.yaml
+tests:
+  - it: renders no worker by default
+    set:
+      relayUrl: wss://buzz.example.com
+      ownerPubkey: "0000000000000000000000000000000000000000000000000000000000000000"
+      externalPostgresql.url: postgres://u:p@h:5432/d
+      externalRedis.url: redis://h:6379
+      s3.endpoint: https://s3.example.com
+      s3.accessKey: test
+      s3.secretKey: test
+    asserts:
+      - hasDocuments:
+          count: 0
+        template: templates/storage-accounting-cronjob.yaml
+  - it: renders a non-overlapping worker with an isolated cap
+    set:
+      relayUrl: wss://buzz.example.com
+      ownerPubkey: "0000000000000000000000000000000000000000000000000000000000000000"
+      externalPostgresql.url: postgres://u:p@h:5432/d
+      externalRedis.url: redis://h:6379
+      s3.endpoint: https://s3.example.com
+      s3.bucket: buzz-media-example
+      s3.region: us-west-2
+      s3.addressingStyle: virtual
+      s3.accessKey: test
+      s3.secretKey: test
+      storageAccounting:
+        enabled: true
+        relayMode: external
+        maxObjects: 10000000
+        activeDeadlineSeconds: 3600
+        serviceAccountName: buzz-storage-accounting
+        podLabels:
+          tags.datadoghq.com/env: production
+          tags.datadoghq.com/service: buzz-storage-accounting
+          tags.datadoghq.com/version: test-sha
+    asserts:
+      - hasDocuments:
+          count: 1
+        template: templates/storage-accounting-cronjob.yaml
+      - equal:
+          path: kind
+          value: CronJob
+        template: templates/storage-accounting-cronjob.yaml
+      - equal:
+          path: spec.concurrencyPolicy
+          value: Forbid
+        template: templates/storage-accounting-cronjob.yaml
+      - equal:
+          path: spec.jobTemplate.spec.backoffLimit
+          value: 0
+        template: templates/storage-accounting-cronjob.yaml
+      - equal:
+          path: spec.jobTemplate.spec.activeDeadlineSeconds
+          value: 3600
+        template: templates/storage-accounting-cronjob.yaml
+      - equal:
+          path: spec.jobTemplate.spec.template.spec.restartPolicy
+          value: Never
+        template: templates/storage-accounting-cronjob.yaml
+      - equal:
+          path: spec.jobTemplate.spec.template.spec.serviceAccountName
+          value: buzz-storage-accounting
+        template: templates/storage-accounting-cronjob.yaml
+      - equal:
+          path: spec.jobTemplate.spec.template.spec.containers[0].securityContext.readOnlyRootFilesystem
+          value: true
+        template: templates/storage-accounting-cronjob.yaml
+      - equal:
+          path: spec.jobTemplate.spec.template.metadata.annotations["sidecar.istio.io/inject"]
+          value: "false"
+        template: templates/storage-accounting-cronjob.yaml
+      - equal:
+          path: spec.jobTemplate.spec.template.metadata.labels["tags.datadoghq.com/service"]
+          value: buzz-storage-accounting
+        template: templates/storage-accounting-cronjob.yaml
+      - equal:
+          path: spec.jobTemplate.spec.template.metadata.labels["tags.datadoghq.com/env"]
+          value: production
+        template: templates/storage-accounting-cronjob.yaml
+      - equal:
+          path: spec.jobTemplate.spec.template.metadata.labels["tags.datadoghq.com/version"]
+          value: test-sha
+        template: templates/storage-accounting-cronjob.yaml
+      - equal:
+          path: spec.jobTemplate.spec.template.spec.containers[0].args
+          value: ["storage-snapshot", "--max-objects", "10000000"]
+        template: templates/storage-accounting-cronjob.yaml
+      - equal:
+          path: spec.jobTemplate.spec.template.spec.containers[0].resources.requests.memory
+          value: 20Gi
+        template: templates/storage-accounting-cronjob.yaml
+      - equal:
+          path: spec.jobTemplate.spec.template.spec.containers[0].resources.limits.memory
+          value: 20Gi
+        template: templates/storage-accounting-cronjob.yaml
+      - contains:
+          path: spec.jobTemplate.spec.template.spec.containers[0].env
+          content:
+            name: BUZZ_STORAGE_SNAPSHOT_CODE_SHA
+            value: 0.1.0
+        template: templates/storage-accounting-cronjob.yaml
+      - contains:
+          path: spec.jobTemplate.spec.template.spec.containers[0].env
+          content:
+            name: BUZZ_S3_BUCKET
+            value: buzz-media-example
+        template: templates/storage-accounting-cronjob.yaml
+      - lengthEqual:
+          path: spec.jobTemplate.spec.template.spec.containers[0].env
+          count: 8
+        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
+      - notContains:
+          path: spec.template.spec.containers[0].env
+          content:
+            name: BUZZ_STORAGE_SWEEP_MAX_OBJECTS
+            value: "10000000"
+        template: templates/deployment.yaml
+      - notContains:
+          path: spec.jobTemplate.spec.template.spec.containers[0].env
+          content:
+            name: BUZZ_RELAY_PRIVATE_KEY
+        template: templates/storage-accounting-cronjob.yaml
+      - notContains:
+          path: spec.jobTemplate.spec.template.spec.containers[0].env
+          content:
+            name: BUZZ_GIT_HOOK_HMAC_SECRET
+        template: templates/storage-accounting-cronjob.yaml

deploy/charts/buzz/values.schema.json

@@ -217,6 +217,23 @@
         "secretKey": { "type": "string" }
       }
     },
+    "storageAccounting": {
+      "type": "object",
+      "additionalProperties": false,
+      "properties": {
+        "enabled": { "type": "boolean" },
+        "schedule": { "type": "string", "minLength": 1 },
+        "maxObjects": { "type": "integer", "minimum": 1 },
+        "activeDeadlineSeconds": { "type": "integer", "minimum": 1 },
+        "successfulJobsHistoryLimit": { "type": "integer", "minimum": 0 },
+        "failedJobsHistoryLimit": { "type": "integer", "minimum": 0 },
+        "relayMode": { "type": "string", "enum": ["inline", "external", "off"] },
+        "serviceAccountName": { "type": "string" },
+        "podLabels": { "type": "object" },
+        "podAnnotations": { "type": "object" },
+        "resources": { "type": "object" }
+      }
+    },
     "minio": {
       "type": "object",
       "additionalProperties": false,

deploy/charts/buzz/values.yaml

@@ -328,13 +328,20 @@ externalRedis:
 # MinIO Deployment, creates the bucket via a post-install Job, and composes
 # the endpoint + autogenerated credentials automatically.
 #
-# Storage metrics (hourly bucket sweep, BUZZ_STORAGE_METRICS — see env docs):
+# Storage metrics are controlled by storageAccounting.relayMode below:
-# the credentials above must additionally grant `s3:ListBucket` on the bucket
+# - inline (default) keeps the relay-local hourly bucket sweep.
-# ARN itself (bucket-level; distinct from the object-level GetObject/
+# - external reads the latest complete snapshot written by the CronJob.
-# PutObject/DeleteObject perms already required for media). Without it the
+# - off disables storage metrics in the relay.
-# first sweep fails AccessDenied and buzz_storage_sweep_ok stays 0 — no other
+# storageAccounting.enabled controls the CronJob independently, so operators
-# media functionality is affected. Set BUZZ_STORAGE_METRICS=off to disable
+# can stage the worker before switching the relay to external mode. Inline and
-# the sweep entirely on a deployment that can't grant it.
+# 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.
 #
 # Whole-community deletion (`buzz-admin deletions ...`) permanently removes
 # tenant-owned object versions through the v5 deletion path. In addition to the
@@ -344,8 +351,8 @@ externalRedis:
 # never-versioned buckets, which S3 exposes through `ListObjectVersions` with the
 # `null` version id.
 #
-# Note: buzz_storage_sweep_failures is a process-local gauge — on leader
+# Note: inline-only buzz_storage_sweep_failures is a process-local gauge — on
-# failover it resets to the new leader's local count, not a global total.
+# 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
@@ -363,6 +370,28 @@ 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.
+storageAccounting:
+  enabled: false
+  schedule: "0 3 * * *"
+  maxObjects: 10000000
+  activeDeadlineSeconds: 3600
+  successfulJobsHistoryLimit: 1
+  failedJobsHistoryLimit: 3
+  relayMode: inline              # inline | external | off
+  serviceAccountName: ""         # defaults to the main Buzz service account
+  podLabels: {}
+  podAnnotations:
+    sidecar.istio.io/inject: "false"
+  resources:
+    requests:
+      cpu: "1"
+      memory: 20Gi
+    limits:
+      cpu: "1"
+      memory: 20Gi
+
 # In-cluster MinIO for the quickstart profile only. Production deploys leave
 # this disabled and use s3.* (or secrets.existingSecret) against managed S3.
 minio:

migrations/0046_storage_accounting_snapshots.sql

@@ -0,0 +1,15 @@
+-- Last complete media-storage accounting result produced by the isolated
+-- buzz-admin worker. The worker replaces this singleton row only after the
+-- full S3 listing and fold succeed, so relay readers never observe a partial
+-- snapshot.
+CREATE TABLE storage_accounting_snapshots (
+    singleton BOOLEAN PRIMARY KEY DEFAULT TRUE CHECK (singleton),
+    snapshot JSONB NOT NULL CHECK (jsonb_typeof(snapshot) = 'object'),
+    completed_at TIMESTAMPTZ NOT NULL DEFAULT transaction_timestamp(),
+    duration_ms BIGINT NOT NULL CHECK (duration_ms >= 0),
+    max_objects BIGINT NOT NULL CHECK (max_objects > 0),
+    code_sha TEXT NOT NULL CHECK (octet_length(code_sha) BETWEEN 1 AND 128)
+);
+
+INSERT INTO _operator_global_tables (table_name, reason) VALUES
+    ('storage_accounting_snapshots', 'deployment-global completed media accounting handoff');

schema/schema.sql

@@ -1895,3 +1895,18 @@ CREATE INDEX idx_relay_operator_audit_target
 
 INSERT INTO _operator_global_tables (table_name, reason) VALUES
     ('relay_operator_audit', 'deployment-global append-only roster mutation audit trail; no community_id intentionally');
+
+-- ── Storage accounting snapshot ─────────────────────────────────────────────
+-- Deployment-global singleton produced by the isolated S3 accounting worker.
+
+CREATE TABLE storage_accounting_snapshots (
+    singleton BOOLEAN PRIMARY KEY DEFAULT TRUE CHECK (singleton),
+    snapshot JSONB NOT NULL CHECK (jsonb_typeof(snapshot) = 'object'),
+    completed_at TIMESTAMPTZ NOT NULL DEFAULT transaction_timestamp(),
+    duration_ms BIGINT NOT NULL CHECK (duration_ms >= 0),
+    max_objects BIGINT NOT NULL CHECK (max_objects > 0),
+    code_sha TEXT NOT NULL CHECK (octet_length(code_sha) BETWEEN 1 AND 128)
+);
+
+INSERT INTO _operator_global_tables (table_name, reason) VALUES
+    ('storage_accounting_snapshots', 'deployment-global completed media accounting handoff');

scripts/run-tests.sh

@@ -104,6 +104,12 @@ run_unit_tests() {
   run_test_step "buzz-db unit tests" \
     cargo test -p buzz-db --lib -- --nocapture
 
+  run_test_step "buzz-media storage snapshot serialization test" \
+    cargo test -p buzz-media --lib bucket_index::tests::bucket_snapshot_json_round_trip_preserves_community_keys -- --exact --nocapture
+
+  run_test_step "buzz-admin completed snapshot persistence test" \
+    cargo test -p buzz-admin storage_snapshot_tests::failed_fold_never_invokes_snapshot_persistence -- --exact --nocapture
+
   # Multi-tenant conformance gate: independent replay checker + golden
   # fixtures (buzz-conformance). Pure in-process trace replay, no infra.
   run_test_step "buzz-conformance tests" \
@@ -142,6 +148,9 @@ run_unit_tests() {
 
   run_test_step "buzz-relay side-effects helper tests" \
     cargo test -p buzz-relay --lib handlers::side_effects::tests:: -- --nocapture
+
+  run_test_step "buzz-relay storage snapshot tests" \
+    cargo test -p buzz-relay --lib storage_sweep::tests:: -- --nocapture
 }
 
 # ---- DB / integration tests (infra required) --------------------------------