diff --git a/Justfile b/Justfile index 00c4fecc2f9..80af8226db2 100644 --- a/Justfile +++ b/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. diff --git a/crates/buzz-admin/src/main.rs b/crates/buzz-admin/src/main.rs index 19a3b1d9d48..bfb256e311c 100644 --- a/crates/buzz-admin/src/main.rs +++ b/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 { 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 { } } +async fn cmd_storage_snapshot(max_objects: u64) -> Result { + 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( + fold: FoldFuture, + persist: Persist, +) -> Result<(BucketSnapshot, i64)> +where + FoldFuture: Future>, + Persist: FnOnce(serde_json::Value, i64) -> PersistFuture, + PersistFuture: Future>, +{ + 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 { + 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::() + .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 { 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::(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); + } +} diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 6f8d0ffb3d4..802db4376bb 100644 --- a/crates/buzz-db/src/lib.rs +++ b/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; diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index 59015125042..6c80b430f10 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/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(), 44); + assert_eq!(migrations.len(), 45); 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[44].version, 45); + assert!(migrations[44] + .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[])"), diff --git a/crates/buzz-db/src/store/mod.rs b/crates/buzz-db/src/store/mod.rs index 1fa1273eb0f..79739373d0f 100644 --- a/crates/buzz-db/src/store/mod.rs +++ b/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. diff --git a/crates/buzz-db/src/store/storage_accounting.rs b/crates/buzz-db/src/store/storage_accounting.rs new file mode 100644 index 00000000000..73c39bea39c --- /dev/null +++ b/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, + /// 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> { + 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> { + let mut connection = + observability::acquire_writer(&self.pool, observability::WriterOperation::Maintenance) + .await?; + let row = sqlx::query_as::<_, (serde_json::Value, DateTime, 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; + } +} diff --git a/crates/buzz-media/src/bucket_index.rs b/crates/buzz-media/src/bucket_index.rs index 6c78c2e0a16..86a3eff7919 100644 --- a/crates/buzz-media/src/bucket_index.rs +++ b/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] diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 206f0329c0e..2d8c0eae315 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -1763,26 +1763,54 @@ 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 { - return; + 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"); + } + }, + Ok(None) => {} + Err(error) => { + warn!(error = %error, "failed to load stored storage snapshot"); + } + } + } } - let media_storage = Arc::clone(&state.media_storage); - let max_objects = config.max_objects; - storage_sweep::maybe_spawn_sweep( - &state.storage_sweep, - config.interval, - config.timeout, - async move { - buzz_media::fold_bucket_listing(max_objects, move |token| { - let media_storage = Arc::clone(&media_storage); - async move { media_storage.list_page(token, 1000).await } - }) - .await - }, - ) - .await; - storage_sweep::emit_storage_metrics(&state.storage_sweep, host_map, |id| { emission_scope.allows(id) }) diff --git a/crates/buzz-relay/src/storage_sweep.rs b/crates/buzz-relay/src/storage_sweep.rs index eccadcd835d..e7ad7f92b02 100644 --- a/crates/buzz-relay/src/storage_sweep.rs +++ b/crates/buzz-relay/src/storage_sweep.rs @@ -21,12 +21,24 @@ 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 - /// gauge (including the health gauges) is ever emitted — a relay whose - /// deployment lacks `s3:ListBucket` can turn the whole feature off. - pub enabled: bool, + /// 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 the - /// `BUZZ_STORAGE_METRICS` kill switch (`off` ⇒ disabled, anything else - /// including unset ⇒ enabled). + /// `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. 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::().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()) - .as_deref() - != Some("off"); + .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, - 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 } } } @@ -94,6 +118,8 @@ struct LastAttempt { struct CachedSnapshot { data: BucketSnapshot, completed_at: Instant, + completed_at_wall: DateTime, + max_objects: Option, } /// What a spawned sweep task hands back to the tick that harvests it. @@ -214,6 +240,8 @@ pub async fn maybe_spawn_sweep( // Stamped at harvest, not sweep completion — exported // age/cadence may lag by ≤1 usage tick. completed_at: Instant::now(), + completed_at_wall: Utc::now(), + max_objects: None, }); } Err(err) => { @@ -258,6 +286,28 @@ pub async fn maybe_spawn_sweep( state.in_flight = Some(handle); } +/// Replace the cache with a complete snapshot loaded from durable storage. +pub 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, + max_objects: Some(max_objects), + }); + state.last_attempt = Some(LastAttempt { ok: true, duration }); +} + /// Emit the storage-family gauges from the cached snapshot. Call every usage /// tick, leader-only (mirrors `emit_db_usage_metrics`'s leadership gate) — /// never from the spawned sweep task itself, so a sweep that completes after @@ -302,8 +352,11 @@ pub async fn emit_storage_metrics( } return; }; - metrics::gauge!("buzz_storage_sweep_age_seconds") - .set(cached.completed_at.elapsed().as_secs_f64()); + let age = Utc::now() + .signed_duration_since(cached.completed_at_wall) + .to_std() + .unwrap_or_default(); + metrics::gauge!("buzz_storage_sweep_age_seconds").set(age.as_secs_f64()); let snapshot = &cached.data; metrics::gauge!("buzz_total_storage_bytes", "kind" => "physical") @@ -322,6 +375,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 Some(max_objects) = cached.max_objects { + metrics::gauge!("buzz_storage_sweep_max_objects").set(max_objects as f64); + metrics::gauge!("buzz_storage_sweep_cap_utilization") + .set(snapshot.physical_objects as f64 / max_objects as f64); + } let mut current = HashSet::new(); let mut unmapped_bytes = 0u64; @@ -398,20 +456,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() { - assert!(!parse_enabled(Some("off"))); - assert!(!parse_enabled(Some("OFF"))); - assert!(parse_enabled(Some("on"))); - assert!(parse_enabled(Some("anything-else"))); - assert!(parse_enabled(None)); - } - - fn parse_enabled(value: Option<&str>) -> bool { - value.map(str::trim).map(str::to_ascii_lowercase).as_deref() != Some("off") + 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 --- @@ -437,6 +511,8 @@ mod tests { let cached = Some(CachedSnapshot { data: BucketSnapshot::default(), completed_at: Instant::now(), + completed_at_wall: Utc::now(), + max_objects: None, }); assert!(should_spawn( &cached, @@ -456,6 +532,8 @@ mod tests { let cached = Some(CachedSnapshot { data: BucketSnapshot::default(), completed_at: now, + completed_at_wall: Utc::now(), + max_objects: None, }); assert!(!should_spawn( &cached, @@ -864,6 +942,8 @@ mod tests { cached: Some(CachedSnapshot { data: snapshot, completed_at: Instant::now(), + completed_at_wall: Utc::now(), + max_objects: None, }), last_attempt: Some(LastAttempt { ok: true, @@ -973,6 +1053,8 @@ mod tests { cached: Some(CachedSnapshot { data: make_snapshot(true, 20), completed_at: Instant::now(), + completed_at_wall: Utc::now(), + max_objects: None, }), last_attempt: Some(LastAttempt { ok: true, @@ -1017,6 +1099,8 @@ mod tests { guard.cached = Some(CachedSnapshot { data: make_snapshot(false, 20), completed_at: Instant::now(), + completed_at_wall: Utc::now(), + max_objects: None, }); } let mut host_map_2 = HashMap::new(); @@ -1087,4 +1171,35 @@ mod tests { "(c) scope removal: host.c objects must be zeroed" ); } + + #[tokio::test] + async fn persisted_snapshot_emits_worker_freshness_and_cap() { + 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, &host_map, |_| true)); + }); + let values = gauge_snapshot(&recorder); + assert_eq!(values.get("buzz_storage_sweep_ok"), Some(&1.0)); + assert_eq!( + values.get("buzz_storage_sweep_duration_seconds"), + Some(&12.0) + ); + assert_eq!(values.get("buzz_storage_sweep_max_objects"), Some(&100.0)); + assert_eq!( + values.get("buzz_storage_sweep_cap_utilization"), + Some(&0.25) + ); + assert!(values["buzz_storage_sweep_age_seconds"] >= 30.0); + } } diff --git a/deploy/charts/buzz/Chart.yaml b/deploy/charts/buzz/Chart.yaml index 49a6fafd192..e26a2815fff 100644 --- a/deploy/charts/buzz/Chart.yaml +++ b/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 diff --git a/deploy/charts/buzz/templates/_helpers.tpl b/deploy/charts/buzz/templates/_helpers.tpl index 13efe0d1fd6..e5ec9182d68 100644 --- a/deploy/charts/buzz/templates/_helpers.tpl +++ b/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. diff --git a/deploy/charts/buzz/templates/deployment.yaml b/deploy/charts/buzz/templates/deployment.yaml index 319ec7f1594..59f52f759e7 100644 --- a/deploy/charts/buzz/templates/deployment.yaml +++ b/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 diff --git a/deploy/charts/buzz/templates/storage-accounting-cronjob.yaml b/deploy/charts/buzz/templates/storage-accounting-cronjob.yaml new file mode 100644 index 00000000000..deaa816888e --- /dev/null +++ b/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 }} diff --git a/deploy/charts/buzz/tests/storage_accounting_test.yaml b/deploy/charts/buzz/tests/storage_accounting_test.yaml new file mode 100644 index 00000000000..916e3028d4e --- /dev/null +++ b/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 diff --git a/deploy/charts/buzz/values.schema.json b/deploy/charts/buzz/values.schema.json index aaab848fd33..ea5db6398dd 100644 --- a/deploy/charts/buzz/values.schema.json +++ b/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, diff --git a/deploy/charts/buzz/values.yaml b/deploy/charts/buzz/values.yaml index 6c57a5c8ac9..7dab9ff61cf 100644 --- a/deploy/charts/buzz/values.yaml +++ b/deploy/charts/buzz/values.yaml @@ -328,13 +328,17 @@ 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): -# the credentials above must additionally grant `s3:ListBucket` on the bucket -# ARN itself (bucket-level; distinct from the object-level GetObject/ -# PutObject/DeleteObject perms already required for media). Without it the -# first sweep fails AccessDenied and buzz_storage_sweep_ok stays 0 — no other -# media functionality is affected. Set BUZZ_STORAGE_METRICS=off to disable -# the sweep entirely on a deployment that can't grant it. +# 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. +# - 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; +# 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. The configured-cap and cap-utilization +# gauges are external-only because inline snapshots do not persist their cap. # # Whole-community deletion (`buzz-admin deletions ...`) permanently removes # tenant-owned object versions through the v5 deletion path. In addition to the @@ -363,6 +367,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: diff --git a/migrations/0045_storage_accounting_snapshots.sql b/migrations/0045_storage_accounting_snapshots.sql new file mode 100644 index 00000000000..363071b51a1 --- /dev/null +++ b/migrations/0045_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'); diff --git a/schema/schema.sql b/schema/schema.sql index 09508125622..18eb0382faf 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -1893,3 +1893,17 @@ 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'); diff --git a/scripts/run-tests.sh b/scripts/run-tests.sh index 6a14fe0cfee..8395e9d0580 100755 --- a/scripts/run-tests.sh +++ b/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) --------------------------------