diff --git a/.env.example b/.env.example index b9bfcada0e..d962d565c8 100644 --- a/.env.example +++ b/.env.example @@ -95,6 +95,13 @@ BUZZ_S3_BUCKET=buzz-media BUZZ_S3_REGION=us-east-1 BUZZ_S3_ADDRESSING_STYLE=path +# Media payload write migration. Unset defaults to dual-read-legacy-write, so upgrades preserve +# existing flat-only writes and do not duplicate objects automatically. +# Stages: dual-read-legacy-write -> dual-read-dual-write -> sharded-only. +# Deploy compatibility readers everywhere and backfill/reconcile objects before +# advancing; do not roll sharded writers back to a version without sharded reads. +# BUZZ_MEDIA_MIGRATION_PHASE=dual-read-legacy-write + # ----------------------------------------------------------------------------- # Media Upload Admission # ----------------------------------------------------------------------------- diff --git a/Cargo.lock b/Cargo.lock index 937ead564a..c62b0c057e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1052,17 +1052,20 @@ dependencies = [ name = "buzz-media" version = "0.1.0" dependencies = [ + "anyhow", "axum", "blurhash", "buzz-core", "bytes", "chrono", + "clap", "futures-core", "futures-util", "hex", "image", "imagesize", "infer", + "metrics", "mp4", "nostr", "rust-s3", @@ -1074,6 +1077,7 @@ dependencies = [ "tokio", "tokio-util", "tracing", + "tracing-subscriber", "ulid", "uuid", ] diff --git a/Dockerfile b/Dockerfile index d883ac6b01..b750db4736 100644 --- a/Dockerfile +++ b/Dockerfile @@ -69,14 +69,18 @@ RUN cargo chef cook --release --recipe-path recipe.json COPY . . RUN cargo build --release --locked -p buzz-relay --bin buzz-relay \ -p buzz-admin --bin buzz-admin \ - -p buzz-pair-relay --bin buzz-pair-relay + -p buzz-pair-relay --bin buzz-pair-relay \ + -p buzz-media --bin buzz-media-layout-backfill \ + --bin buzz-media-layout-delete-legacy # Derive the normal release binaries from the same optimized ELF files as the # debug image so the two variants cannot drift at code-generation time. FROM builder AS stripped-binaries RUN strip target/release/buzz-relay \ && strip target/release/buzz-admin \ - && strip target/release/buzz-pair-relay + && strip target/release/buzz-pair-relay \ + && strip target/release/buzz-media-layout-backfill \ + && strip target/release/buzz-media-layout-delete-legacy # ─── Stage 4: web bundle (pnpm + vite) ────────────────────────────────────── # Independent of the Rust layers so a CSS change doesn't bust Rust cache and @@ -169,6 +173,8 @@ FROM runtime-base AS runtime-debug COPY --from=builder /build/target/release/buzz-relay /usr/local/bin/buzz-relay COPY --from=builder /build/target/release/buzz-admin /usr/local/bin/buzz-admin COPY --from=builder /build/target/release/buzz-pair-relay /usr/local/bin/buzz-pair-relay +COPY --from=builder /build/target/release/buzz-media-layout-backfill /usr/local/bin/buzz-media-layout-backfill +COPY --from=builder /build/target/release/buzz-media-layout-delete-legacy /usr/local/bin/buzz-media-layout-delete-legacy # Keep the stripped runtime as the final/default Dockerfile target so existing # `docker build .` callers and release tags retain their current behavior. @@ -176,3 +182,5 @@ FROM runtime-base AS runtime COPY --from=stripped-binaries /build/target/release/buzz-relay /usr/local/bin/buzz-relay COPY --from=stripped-binaries /build/target/release/buzz-admin /usr/local/bin/buzz-admin COPY --from=stripped-binaries /build/target/release/buzz-pair-relay /usr/local/bin/buzz-pair-relay +COPY --from=stripped-binaries /build/target/release/buzz-media-layout-backfill /usr/local/bin/buzz-media-layout-backfill +COPY --from=stripped-binaries /build/target/release/buzz-media-layout-delete-legacy /usr/local/bin/buzz-media-layout-delete-legacy diff --git a/crates/buzz-media/Cargo.toml b/crates/buzz-media/Cargo.toml index 530ce69c90..78db7217db 100644 --- a/crates/buzz-media/Cargo.toml +++ b/crates/buzz-media/Cargo.toml @@ -32,6 +32,10 @@ tempfile = "3" tokio-util = { version = "0.7", features = ["io"] } futures-util = "0.3" futures-core = "0.3" +metrics = { workspace = true } +clap = { version = "4", features = ["derive", "env"] } +anyhow = { workspace = true } +tracing-subscriber = { workspace = true } [dev-dependencies] tokio = { workspace = true, features = ["test-util"] } diff --git a/crates/buzz-media/src/bin/buzz-media-layout-backfill.rs b/crates/buzz-media/src/bin/buzz-media-layout-backfill.rs new file mode 100644 index 0000000000..5033ebd7ec --- /dev/null +++ b/crates/buzz-media/src/bin/buzz-media-layout-backfill.rs @@ -0,0 +1,94 @@ +//! Idempotently copy legacy media payloads into the sharded layout. + +use anyhow::{Context, Result}; +use buzz_media::migration::{objects_for_sidecar, parse_sidecar_key, RequestPacer}; +use clap::Parser; + +mod media_layout_common; +use media_layout_common::CommonArgs; + +#[derive(Debug, Parser)] +#[command(name = "buzz-media-layout-backfill")] +struct Args { + #[command(flatten)] + common: CommonArgs, + /// Report actions without copying objects. + #[arg(long, env = "BUZZ_MEDIA_MIGRATION_DRY_RUN", default_value_t = false)] + dry_run: bool, +} + +#[tokio::main] +async fn main() -> Result<()> { + tracing_subscriber::fmt().with_target(false).init(); + let args = Args::parse(); + let storage = args.common.storage()?; + let mut pacer = RequestPacer::new(args.common.requests_per_second); + let mut continuation = None; + let mut start_after = args.common.start_after.clone(); + let mut processed = 0_u64; + let mut copied = 0_u64; + let mut skipped = 0_u64; + let mut checkpoint = start_after.clone(); + + loop { + pacer.wait().await; + let page = storage + .list_prefix_page( + "_meta/", + continuation.take(), + start_after.take(), + args.common.page_size, + ) + .await + .context("list media sidecars")?; + for (sidecar_key, _) in page.objects { + let Some((community, sha)) = parse_sidecar_key(&sidecar_key) else { + tracing::warn!(key = %sidecar_key, "skipping malformed sidecar key"); + continue; + }; + pacer.wait().await; + let bytes = storage + .get(&sidecar_key) + .await + .context("read media sidecar")?; + let meta = serde_json::from_slice(&bytes).context("parse media sidecar")?; + for object in objects_for_sidecar(community, sha, &meta)? { + processed += 1; + pacer.wait().await; + if storage.head(&object.sharded).await? { + skipped += 1; + continue; + } + pacer.wait().await; + let source_exists = storage.head(&object.legacy).await?; + if !source_exists { + anyhow::bail!( + "legacy source missing: {} (checkpoint: {sidecar_key})", + object.legacy + ); + } + if args.dry_run { + tracing::info!(source = %object.legacy, destination = %object.sharded, "would copy"); + } else { + pacer.wait().await; + storage.copy(&object.legacy, &object.sharded).await?; + pacer.wait().await; + if !storage.head(&object.sharded).await? { + anyhow::bail!("destination verification failed: {}", object.sharded); + } + copied += 1; + } + } + checkpoint = Some(sidecar_key); + } + if !page.is_truncated { + break; + } + continuation = page.next_continuation_token; + if continuation.is_none() { + anyhow::bail!("truncated S3 listing returned no continuation token"); + } + } + tracing::info!(processed, copied, skipped, dry_run = args.dry_run, checkpoint = ?checkpoint, "backfill complete"); + Ok(()) +} diff --git a/crates/buzz-media/src/bin/buzz-media-layout-delete-legacy.rs b/crates/buzz-media/src/bin/buzz-media-layout-delete-legacy.rs new file mode 100644 index 0000000000..872f70d4f2 --- /dev/null +++ b/crates/buzz-media/src/bin/buzz-media-layout-delete-legacy.rs @@ -0,0 +1,116 @@ +//! Delete verified legacy media payloads after migration reconciliation. + +use anyhow::{Context, Result}; +use buzz_media::migration::{objects_for_sidecar, parse_sidecar_key, RequestPacer}; +use buzz_media::MediaStorage; +use clap::Parser; + +mod media_layout_common; +use media_layout_common::CommonArgs; + +const CONFIRMATION: &str = "delete-verified-legacy-media"; + +#[derive(Debug, Parser)] +#[command(name = "buzz-media-layout-delete-legacy")] +struct Args { + #[command(flatten)] + common: CommonArgs, + /// Preview is the safe default. Set false only after reconciliation. + #[arg(long, env = "BUZZ_MEDIA_MIGRATION_DRY_RUN", default_value_t = true, action = clap::ArgAction::Set)] + dry_run: bool, + /// Required with --dry-run=false to guard accidental destructive Jobs. + #[arg(long, env = "BUZZ_MEDIA_DELETE_CONFIRM")] + confirm: Option, +} + +/// Scan all selected sidecars before deleting anything. This two-pass design is +/// important because one flat legacy CAS key can serve multiple communities: +/// every community destination must exist before the shared source is removed. +async fn scan( + storage: &MediaStorage, + common: &CommonArgs, + pacer: &mut RequestPacer, + delete: bool, + dry_run: bool, +) -> Result<(u64, u64, Option)> { + let mut continuation = None; + let mut start_after = common.start_after.clone(); + let mut changed = 0_u64; + let mut skipped = 0_u64; + let mut checkpoint = start_after.clone(); + loop { + pacer.wait().await; + let page = storage + .list_prefix_page( + "_meta/", + continuation.take(), + start_after.take(), + common.page_size, + ) + .await + .context("list media sidecars")?; + for (sidecar_key, _) in page.objects { + let Some((community, sha)) = parse_sidecar_key(&sidecar_key) else { + tracing::warn!(key = %sidecar_key, "skipping malformed sidecar key"); + continue; + }; + pacer.wait().await; + let bytes = storage + .get(&sidecar_key) + .await + .context("read media sidecar")?; + let meta = serde_json::from_slice(&bytes).context("parse media sidecar")?; + for object in objects_for_sidecar(community, sha, &meta)? { + pacer.wait().await; + if !storage.head(&object.sharded).await? { + anyhow::bail!( + "refusing deletion: sharded destination missing: {} (checkpoint: {sidecar_key})", + object.sharded + ); + } + if !delete { + continue; + } + pacer.wait().await; + if !storage.head(&object.legacy).await? { + skipped += 1; + continue; + } + if dry_run { + tracing::info!(key = %object.legacy, "would delete verified legacy object"); + } else { + pacer.wait().await; + storage.delete(&object.legacy).await?; + changed += 1; + } + } + checkpoint = Some(sidecar_key); + } + if !page.is_truncated { + break; + } + continuation = page.next_continuation_token; + if continuation.is_none() { + anyhow::bail!("truncated S3 listing returned no continuation token"); + } + } + Ok((changed, skipped, checkpoint)) +} + +#[tokio::main] +async fn main() -> Result<()> { + tracing_subscriber::fmt().with_target(false).init(); + let args = Args::parse(); + if !args.dry_run && args.confirm.as_deref() != Some(CONFIRMATION) { + anyhow::bail!("destructive mode requires --confirm={CONFIRMATION}"); + } + let storage = args.common.storage()?; + let mut pacer = RequestPacer::new(args.common.requests_per_second); + let (_, _, verified_checkpoint) = + scan(&storage, &args.common, &mut pacer, false, args.dry_run).await?; + tracing::info!(checkpoint = ?verified_checkpoint, "all selected sharded destinations verified; beginning deletion pass"); + let (deleted, skipped, checkpoint) = + scan(&storage, &args.common, &mut pacer, true, args.dry_run).await?; + tracing::info!(deleted, skipped, dry_run = args.dry_run, checkpoint = ?checkpoint, "legacy deletion complete"); + Ok(()) +} diff --git a/crates/buzz-media/src/bin/media_layout_common/mod.rs b/crates/buzz-media/src/bin/media_layout_common/mod.rs new file mode 100644 index 0000000000..208df67f9f --- /dev/null +++ b/crates/buzz-media/src/bin/media_layout_common/mod.rs @@ -0,0 +1,60 @@ +use anyhow::{bail, Context, Result}; +use buzz_media::{MediaConfig, MediaMigrationPhase, MediaStorage, S3AddressingStyle}; +use clap::Args; + +#[derive(Debug, Args)] +pub struct CommonArgs { + #[arg(long, env = "BUZZ_S3_ENDPOINT")] + pub s3_endpoint: String, + #[arg(long, env = "BUZZ_S3_BUCKET")] + pub s3_bucket: String, + #[arg(long, env = "BUZZ_S3_REGION", default_value = "us-east-1")] + pub s3_region: String, + #[arg(long, env = "BUZZ_S3_ACCESS_KEY", default_value = "")] + pub s3_access_key: String, + #[arg(long, env = "BUZZ_S3_SECRET_KEY", default_value = "")] + pub s3_secret_key: String, + #[arg(long, env = "BUZZ_S3_ADDRESSING_STYLE", default_value = "path")] + pub s3_addressing_style: S3AddressingStyle, + /// Maximum S3 requests per second, including listing and verification. + #[arg( + long, + env = "BUZZ_MEDIA_MIGRATION_REQUESTS_PER_SECOND", + default_value_t = 25 + )] + pub requests_per_second: u32, + /// Resume after this sidecar key. The final log line prints the next value. + #[arg(long, env = "BUZZ_MEDIA_MIGRATION_START_AFTER")] + pub start_after: Option, + #[arg(long, env = "BUZZ_MEDIA_MIGRATION_PAGE_SIZE", default_value_t = 100)] + pub page_size: usize, +} + +impl CommonArgs { + pub fn storage(&self) -> Result { + if self.requests_per_second == 0 { + bail!("requests-per-second must be greater than zero"); + } + if !(1..=1000).contains(&self.page_size) { + bail!("page-size must be between 1 and 1000"); + } + MediaStorage::new(&MediaConfig { + s3_endpoint: self.s3_endpoint.clone(), + s3_access_key: self.s3_access_key.clone(), + s3_secret_key: self.s3_secret_key.clone(), + s3_bucket: self.s3_bucket.clone(), + s3_region: self.s3_region.clone(), + s3_addressing_style: self.s3_addressing_style, + migration_phase: MediaMigrationPhase::DualReadLegacyWrite, + max_image_bytes: 1, + max_gif_bytes: 1, + max_video_bytes: 1, + max_file_bytes: 1, + public_base_url: "http://localhost/media".into(), + upload_records_enabled: false, + upload_ip_header: None, + upload_port_header: None, + }) + .context("create media S3 client") + } +} diff --git a/crates/buzz-media/src/bucket_index.rs b/crates/buzz-media/src/bucket_index.rs index bb83dc517f..2807371fd8 100644 --- a/crates/buzz-media/src/bucket_index.rs +++ b/crates/buzz-media/src/bucket_index.rs @@ -13,8 +13,8 @@ //! //! | Class | Shape | //! |---|---| -//! | thumb | `{sha256}.thumb.jpg` | -//! | blob | `{sha256}.{ext}` (ext: 1-8 mixed-case alphanumeric) | +//! | thumb | `{sha256}.thumb.jpg` or `media/{hh}/{hh}/{community-uuid}/{sha256}.thumb.jpg` | +//! | blob | `{sha256}.{ext}` or `media/{hh}/{hh}/{community-uuid}/{sha256}.{ext}` (ext: 1-8 mixed-case alphanumeric) | //! | sidecar | `_meta/{community-uuid}/{sha256}.json` | //! | auxiliary | `_uploads/{community-uuid}/{sha256}/{ulid}.json` | //! | unknown | everything else | @@ -32,10 +32,17 @@ use crate::error::MediaError; /// `Auxiliary`, so visibility gauges stay loud instead of silently wrong. #[derive(Debug, Clone, PartialEq, Eq)] pub enum KeyClass { - /// `{sha256}.thumb.jpg` — attributed to the blob's sha. - Thumb { sha256: String }, - /// `{sha256}.{ext}` — physical bytes, logical join key. - Blob { sha256: String, ext: String }, + /// Legacy or sharded thumbnail, attributed to the blob's sha. + Thumb { + community: Option, + sha256: String, + }, + /// Legacy or sharded blob; sharded keys carry direct community attribution. + Blob { + community: Option, + sha256: String, + ext: String, + }, /// `_meta/{community}/{sha256}.json` — the (community, sha) binding. Sidecar { community: Uuid, sha256: String }, /// `_uploads/{community}/{sha256}/{event_id}.json` — fleet physical only. @@ -52,11 +59,34 @@ pub enum KeyClass { /// shape of the blob pattern's segment count), then blob, sidecar, /// auxiliary, and finally unknown. See module docs for the exact shapes. pub fn classify_key(key: &str) -> KeyClass { + if let Some((community, filename)) = parse_sharded_prefix(key) { + if let Some(parsed_sha) = parse_thumb_key(filename) { + return KeyClass::Thumb { + community: Some(community), + sha256: parsed_sha, + }; + } + if let Some((parsed_sha, ext)) = parse_blob_key(filename) { + return KeyClass::Blob { + community: Some(community), + sha256: parsed_sha, + ext, + }; + } + return KeyClass::Unknown; + } if let Some(sha256) = parse_thumb_key(key) { - return KeyClass::Thumb { sha256 }; + return KeyClass::Thumb { + community: None, + sha256, + }; } if let Some((sha256, ext)) = parse_blob_key(key) { - return KeyClass::Blob { sha256, ext }; + return KeyClass::Blob { + community: None, + sha256, + ext, + }; } if let Some((community, sha256)) = parse_sidecar_key(key) { return KeyClass::Sidecar { community, sha256 }; @@ -125,6 +155,27 @@ fn parse_canonical_uuid(s: &str) -> Option { Uuid::parse_str(s).ok() } +/// `media/{sha[0:2]}/{sha[2:4]}/{community}/{filename}`. The filename's digest +/// must agree with both shard segments; malformed migration keys stay unknown. +fn parse_sharded_prefix(key: &str) -> Option<(Uuid, &str)> { + let mut segments = key.split('/'); + if segments.next()? != "media" { + return None; + } + let shard_1 = segments.next()?; + let shard_2 = segments.next()?; + let community = parse_canonical_uuid(segments.next()?)?; + let filename = segments.next()?; + if segments.next().is_some() || shard_1.len() != 2 || shard_2.len() != 2 { + return None; + } + let sha256 = filename.split('.').next()?; + if !is_sha256(sha256) || shard_1 != &sha256[..2] || shard_2 != &sha256[2..4] { + return None; + } + Some((community, filename)) +} + /// `{sha256}.thumb.jpg` fn parse_thumb_key(key: &str) -> Option { let mut parts = key.split('.'); @@ -221,6 +272,10 @@ pub struct BucketSnapshot { pub multi_variant_shas: u64, /// Total bytes of ALL blob variants belonging to anomalous shas. pub multi_variant_bytes: u64, + /// Logical blob variants present in both legacy and sharded layouts. + pub duplicate_layout_variants: u64, + /// Physical bytes across both layouts for duplicate logical variants. + pub duplicate_layout_bytes: u64, pub unknown_key_bytes: u64, pub unknown_key_objects: u64, } @@ -228,14 +283,44 @@ pub struct BucketSnapshot { /// Pure, incremental fold over classified bucket keys. Never retains a full /// object listing — only per-sha/per-binding running totals, bounded by the /// number of distinct shas and sidecar bindings actually present. +#[derive(Debug, Default)] +struct LayoutCopies { + legacy: Option, + sharded: HashMap, +} + +impl LayoutCopies { + fn insert(&mut self, community: Option, size: u64) { + match community { + Some(community) => { + self.sharded.insert(community, size); + } + None => { + self.legacy = Some(size); + } + } + } + + fn physical_bytes(&self) -> u64 { + self.legacy.unwrap_or(0) + self.sharded.values().sum::() + } + + fn logical_bytes(&self, community: Uuid) -> u64 { + self.sharded + .get(&community) + .copied() + .or(self.legacy) + .unwrap_or(0) + } +} + #[derive(Debug, Default)] pub struct BucketAggregate { - /// sha -> bytes of every blob variant seen for that sha (D-EXT: multiple - /// entries is the multi-variant anomaly). - blob_variant_bytes: HashMap>, - /// sha -> thumb bytes. At most one thumb key per sha, so a plain insert - /// is correct (no accumulation needed). - thumb_bytes: HashMap, + /// (sha, ext) -> physical copies by layout. Layout copies are one logical + /// variant and must not double bill during migration. + blob_variants: HashMap<(String, String), LayoutCopies>, + /// sha -> physical thumbnail copies by layout. + thumb_copies: HashMap, /// (community, sha) -> sidecar object's own byte size (informational; /// not part of logical bytes). sidecar_bindings: HashMap<(Uuid, String), u64>, @@ -251,14 +336,21 @@ impl BucketAggregate { self.physical_objects += 1; self.physical_bytes += size; match classify_key(key) { - KeyClass::Thumb { sha256 } => { - self.thumb_bytes.insert(sha256, size); - } - KeyClass::Blob { sha256, .. } => { - self.blob_variant_bytes + KeyClass::Thumb { community, sha256 } => { + self.thumb_copies .entry(sha256) .or_default() - .push(size); + .insert(community, size); + } + KeyClass::Blob { + community, + sha256, + ext, + } => { + self.blob_variants + .entry((sha256, ext)) + .or_default() + .insert(community, size); } KeyClass::Sidecar { community, sha256 } => { self.sidecar_bindings.insert((community, sha256), size); @@ -285,13 +377,28 @@ impl BucketAggregate { let mut multi_variant_bytes = 0u64; let mut orphan_blob_count = 0u64; let mut orphan_blob_bytes = 0u64; - for (sha256, variants) in &self.blob_variant_bytes { - let variant_bytes: u64 = variants.iter().sum(); + let mut duplicate_layout_variants = 0u64; + let mut duplicate_layout_bytes = 0u64; + for copies in self.blob_variants.values() { + if copies.legacy.is_some() && !copies.sharded.is_empty() { + duplicate_layout_variants += 1; + duplicate_layout_bytes += copies.physical_bytes(); + } + } + let mut variants_by_sha: HashMap<&str, Vec<&LayoutCopies>> = HashMap::new(); + for ((sha256, _), copies) in &self.blob_variants { + variants_by_sha + .entry(sha256.as_str()) + .or_default() + .push(copies); + } + for (sha256, variants) in &variants_by_sha { + let variant_bytes: u64 = variants.iter().map(|copies| copies.physical_bytes()).sum(); if variants.len() > 1 { multi_variant_shas += 1; multi_variant_bytes += variant_bytes; } - if !bound_shas.contains(sha256.as_str()) { + if !bound_shas.contains(*sha256) { orphan_blob_count += 1; orphan_blob_bytes += variant_bytes; } @@ -300,17 +407,25 @@ impl BucketAggregate { let orphan_sidecar_count = self .sidecar_bindings .keys() - .filter(|(_, sha256)| !self.blob_variant_bytes.contains_key(sha256)) + .filter(|(_, sha256)| !variants_by_sha.contains_key(sha256.as_str())) .count() as u64; let mut per_community: HashMap = HashMap::new(); for (community, sha256) in self.sidecar_bindings.keys() { - let blob_bytes: u64 = self - .blob_variant_bytes + let blob_bytes: u64 = variants_by_sha + .get(sha256.as_str()) + .map(|variants| { + variants + .iter() + .map(|copies| copies.logical_bytes(*community)) + .sum() + }) + .unwrap_or(0); + let thumb_bytes = self + .thumb_copies .get(sha256) - .map(|v| v.iter().sum()) + .map(|copies| copies.logical_bytes(*community)) .unwrap_or(0); - let thumb_bytes = self.thumb_bytes.get(sha256).copied().unwrap_or(0); let entry = per_community.entry(*community).or_default(); entry.bytes += blob_bytes + thumb_bytes; entry.objects += 1; @@ -329,6 +444,8 @@ impl BucketAggregate { orphan_sidecar_count, multi_variant_shas, multi_variant_bytes, + duplicate_layout_variants, + duplicate_layout_bytes, unknown_key_bytes: self.unknown_bytes, unknown_key_objects: self.unknown_objects, } @@ -429,7 +546,10 @@ mod tests { let s = sha(0xaa); assert_eq!( classify_key(&format!("{s}.thumb.jpg")), - KeyClass::Thumb { sha256: s } + KeyClass::Thumb { + community: None, + sha256: s, + } ); } @@ -439,6 +559,7 @@ mod tests { assert_eq!( classify_key(&format!("{s}.png")), KeyClass::Blob { + community: None, sha256: s, ext: "png".to_string() } @@ -453,12 +574,51 @@ mod tests { assert_eq!( classify_key(&format!("{s}.Z")), KeyClass::Blob { + community: None, sha256: s, ext: "Z".to_string() } ); } + #[test] + fn classifies_sharded_blob_and_thumb_keys_with_community() { + let s = sha(0xab); + let c = community(10); + assert_eq!( + classify_key(&format!("media/ab/ab/{c}/{s}.png")), + KeyClass::Blob { + community: Some(c), + sha256: s.clone(), + ext: "png".to_string(), + } + ); + assert_eq!( + classify_key(&format!("media/ab/ab/{c}/{s}.thumb.jpg")), + KeyClass::Thumb { + community: Some(c), + sha256: s, + } + ); + } + + #[test] + fn malformed_sharded_keys_are_unknown() { + let s = sha(0xab); + let c = community(11); + for key in [ + format!("media/ff/ab/{c}/{s}.png"), + format!("media/ab/ff/{c}/{s}.png"), + format!("media/a/ab/{c}/{s}.png"), + format!("media/ab/ab/not-a-uuid/{s}.png"), + format!("media/ab/ab/{c}/{s}.png/extra"), + format!("media/ab/ab/{c}/{}.png", s.to_uppercase()), + format!("media/ab/ab/{c}/{s}.tar.gz"), + ] { + assert_eq!(classify_key(&key), KeyClass::Unknown, "key: {key}"); + } + } + #[test] fn classifies_sidecar_key() { let s = sha(0xdd); @@ -559,6 +719,44 @@ mod tests { assert_eq!(snap.per_community[&c].objects, 1); } + #[test] + fn dual_layout_copies_count_physically_but_dedupe_logical_usage() { + let s = sha(0xab); + let c = community(12); + let mut agg = BucketAggregate::default(); + agg.fold(&format!("{s}.jpg"), 100); + agg.fold(&format!("media/ab/ab/{c}/{s}.jpg"), 100); + agg.fold(&format!("{s}.thumb.jpg"), 20); + agg.fold(&format!("media/ab/ab/{c}/{s}.thumb.jpg"), 20); + agg.fold(&format!("_meta/{c}/{s}.json"), 10); + + let snap = agg.finish(); + assert_eq!(snap.physical_objects, 5); + assert_eq!(snap.physical_bytes, 250); + assert_eq!(snap.logical_objects, 1); + assert_eq!(snap.logical_bytes, 120); + assert_eq!(snap.per_community[&c].bytes, 120); + assert_eq!(snap.multi_variant_shas, 0); + assert_eq!(snap.duplicate_layout_variants, 1); + assert_eq!(snap.duplicate_layout_bytes, 200); + assert_eq!(snap.unknown_key_objects, 0); + } + + #[test] + fn sharded_copy_is_attributed_only_to_its_community() { + let s = sha(0xcd); + let sharded_community = community(13); + let other_community = community(14); + let mut agg = BucketAggregate::default(); + agg.fold(&format!("media/cd/cd/{sharded_community}/{s}.jpg"), 200); + agg.fold(&format!("_meta/{sharded_community}/{s}.json"), 10); + agg.fold(&format!("_meta/{other_community}/{s}.json"), 10); + + let snap = agg.finish(); + assert_eq!(snap.per_community[&sharded_community].bytes, 200); + assert_eq!(snap.per_community[&other_community].bytes, 0); + } + #[test] fn orphan_blob_has_no_sidecar_binding() { let s = sha(0x44); diff --git a/crates/buzz-media/src/config.rs b/crates/buzz-media/src/config.rs index 3c70e4afe1..dd5ff814da 100644 --- a/crates/buzz-media/src/config.rs +++ b/crates/buzz-media/src/config.rs @@ -33,6 +33,46 @@ impl FromStr for S3AddressingStyle { } } +/// Relay media migration phase controlling both payload reads and writes. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Deserialize)] +pub enum MediaMigrationPhase { + /// Compatibility default: prefer sharded reads with legacy fallback, but + /// write only legacy keys. Upgrades therefore do not begin double-writing. + #[default] + #[serde(rename = "dual-read-legacy-write")] + DualReadLegacyWrite, + /// Migration phase: prefer sharded reads with legacy fallback and write + /// both layouts (sharded first, then legacy). + #[serde(rename = "dual-read-dual-write")] + DualReadDualWrite, + /// Final phase: read and write only hash-sharded keys. + #[serde(rename = "sharded-only")] + ShardedOnly, +} + +impl MediaMigrationPhase { + /// Whether reads may fall back to the legacy key. + pub const fn reads_legacy(self) -> bool { + !matches!(self, Self::ShardedOnly) + } +} + +impl FromStr for MediaMigrationPhase { + type Err = String; + + fn from_str(value: &str) -> Result { + match value { + "dual-read-legacy-write" => Ok(Self::DualReadLegacyWrite), + "dual-read-dual-write" => Ok(Self::DualReadDualWrite), + "sharded-only" => Ok(Self::ShardedOnly), + _ => Err(format!( + "BUZZ_MEDIA_MIGRATION_PHASE must be 'dual-read-legacy-write', \ + 'dual-read-dual-write', or 'sharded-only', got {value:?}" + )), + } + } +} + fn default_max_video_bytes() -> u64 { 524_288_000 // 500 MB } @@ -67,6 +107,10 @@ pub struct MediaConfig { /// S3 URL addressing style. Defaults to path style for MinIO compatibility. #[serde(default)] pub s3_addressing_style: S3AddressingStyle, + /// Relay phase controlling media payload reads and writes. Defaults to + /// dual reads with legacy-only writes for upgrade safety. + #[serde(default)] + pub migration_phase: MediaMigrationPhase, /// Maximum upload size for images (bytes). Default: 50 MB. pub max_image_bytes: u64, /// Maximum upload size for animated GIFs (bytes). Default: 10 MB. @@ -159,7 +203,7 @@ impl MediaConfig { #[cfg(test)] mod tests { - use super::{MediaConfig, S3AddressingStyle}; + use super::{MediaConfig, MediaMigrationPhase, S3AddressingStyle}; use std::str::FromStr; fn valid_config() -> MediaConfig { @@ -170,6 +214,7 @@ mod tests { s3_bucket: "buzz-media".to_string(), s3_region: "us-east-1".to_string(), s3_addressing_style: S3AddressingStyle::Path, + migration_phase: MediaMigrationPhase::DualReadLegacyWrite, max_image_bytes: 1, max_gif_bytes: 1, max_video_bytes: 1, @@ -210,6 +255,26 @@ mod tests { } } + #[test] + fn media_migration_phase_parses_and_has_upgrade_safe_default() { + assert_eq!( + MediaMigrationPhase::default(), + MediaMigrationPhase::DualReadLegacyWrite + ); + assert_eq!( + "dual-read-legacy-write".parse(), + Ok(MediaMigrationPhase::DualReadLegacyWrite) + ); + assert_eq!( + "dual-read-dual-write".parse(), + Ok(MediaMigrationPhase::DualReadDualWrite) + ); + assert_eq!("sharded-only".parse(), Ok(MediaMigrationPhase::ShardedOnly)); + for invalid in ["legacy", "dual", "sharded", "new"] { + assert!(invalid.parse::().is_err()); + } + } + #[test] fn upload_record_knobs_default_off_and_validate() { assert!(valid_config().validate().is_ok()); diff --git a/crates/buzz-media/src/keys.rs b/crates/buzz-media/src/keys.rs new file mode 100644 index 0000000000..de604684a5 --- /dev/null +++ b/crates/buzz-media/src/keys.rs @@ -0,0 +1,190 @@ +//! Deterministic object-key derivation for media payloads. +//! +//! Public Blossom URLs stay flat (`/media/.`), while S3 payloads use +//! hash-leading shards so aggregate request traffic is distributed before the +//! community segment. Legacy keys remain read candidates during migration. + +use buzz_core::tenant::{CommunityId, TenantContext}; + +/// Invalid data supplied to media object-key construction. +#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] +pub enum MediaKeyError { + /// SHA-256 must be exactly 64 lowercase hexadecimal characters. + #[error("invalid SHA-256 digest")] + InvalidSha256, + /// Extensions are canonical lowercase alphanumeric tokens of 1-8 bytes. + #[error("invalid media extension")] + InvalidExtension, + /// Only `.` and `.thumb.jpg` payload names are accepted. + #[error("invalid media payload name")] + InvalidPayloadName, +} + +/// Ordered object keys for compatibility reads. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct MediaReadCandidates { + /// Hash-sharded, community-scoped key tried first. + pub sharded: String, + /// Flat pre-migration key tried only when `sharded` is not found. + pub legacy: String, +} + +fn validate_sha256(sha256: &str) -> Result<(), MediaKeyError> { + if sha256.len() == 64 + && sha256 + .bytes() + .all(|byte| byte.is_ascii_digit() || matches!(byte, b'a'..=b'f')) + { + Ok(()) + } else { + Err(MediaKeyError::InvalidSha256) + } +} + +fn validate_extension(ext: &str) -> Result<(), MediaKeyError> { + if (1..=8).contains(&ext.len()) + && ext + .bytes() + .all(|byte| byte.is_ascii_digit() || byte.is_ascii_lowercase()) + { + Ok(()) + } else { + Err(MediaKeyError::InvalidExtension) + } +} + +/// Flat pre-migration blob key: `.`. +pub fn legacy_blob_key(sha256: &str, ext: &str) -> Result { + validate_sha256(sha256)?; + validate_extension(ext)?; + Ok(format!("{sha256}.{ext}")) +} + +/// Hash-leading blob key: `media/<2>/<2>//.`. +pub fn sharded_blob_key( + community: CommunityId, + sha256: &str, + ext: &str, +) -> Result { + let filename = legacy_blob_key(sha256, ext)?; + Ok(format!( + "media/{}/{}/{community}/{filename}", + &sha256[..2], + &sha256[2..4] + )) +} + +/// Flat pre-migration thumbnail key: `.thumb.jpg`. +pub fn legacy_thumb_key(sha256: &str) -> Result { + validate_sha256(sha256)?; + Ok(format!("{sha256}.thumb.jpg")) +} + +/// Hash-leading thumbnail key: `media/<2>/<2>//.thumb.jpg`. +pub fn sharded_thumb_key(community: CommunityId, sha256: &str) -> Result { + let filename = legacy_thumb_key(sha256)?; + Ok(format!( + "media/{}/{}/{community}/{filename}", + &sha256[..2], + &sha256[2..4] + )) +} + +/// Build new-first, legacy-fallback candidates from a validated public payload name. +/// +/// The community always comes from the server-resolved tenant context; callers +/// cannot supply it through a URL, sidecar, or upload record. +pub fn read_candidates( + ctx: &TenantContext, + payload_name: &str, +) -> Result { + if let Some(sha256) = payload_name.strip_suffix(".thumb.jpg") { + return Ok(MediaReadCandidates { + sharded: sharded_thumb_key(ctx.community(), sha256)?, + legacy: legacy_thumb_key(sha256)?, + }); + } + + let (sha256, ext) = payload_name + .split_once('.') + .ok_or(MediaKeyError::InvalidPayloadName)?; + if ext.contains('.') { + return Err(MediaKeyError::InvalidPayloadName); + } + Ok(MediaReadCandidates { + sharded: sharded_blob_key(ctx.community(), sha256, ext)?, + legacy: legacy_blob_key(sha256, ext)?, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use uuid::Uuid; + + const SHA: &str = "abcdef0123456789abcdef0123456789abcdef0123456789abcdef0123456789"; + + fn tenant(n: u128) -> TenantContext { + TenantContext::resolved(CommunityId::from_uuid(Uuid::from_u128(n)), "media.example") + } + + #[test] + fn derives_hash_leading_community_scoped_blob_and_thumb_keys() { + let ctx = tenant(1); + let community = ctx.community(); + + assert_eq!( + sharded_blob_key(community, SHA, "jpg").unwrap(), + format!("media/ab/cd/{community}/{SHA}.jpg") + ); + assert_eq!( + sharded_thumb_key(community, SHA).unwrap(), + format!("media/ab/cd/{community}/{SHA}.thumb.jpg") + ); + assert_ne!( + sharded_blob_key(community, SHA, "jpg").unwrap(), + sharded_blob_key(tenant(2).community(), SHA, "jpg").unwrap() + ); + } + + #[test] + fn orders_sharded_before_legacy_for_blobs_and_thumbnails() { + let ctx = tenant(1); + let community = ctx.community(); + + assert_eq!( + read_candidates(&ctx, &format!("{SHA}.png")).unwrap(), + MediaReadCandidates { + sharded: format!("media/ab/cd/{community}/{SHA}.png"), + legacy: format!("{SHA}.png"), + } + ); + assert_eq!( + read_candidates(&ctx, &format!("{SHA}.thumb.jpg")).unwrap(), + MediaReadCandidates { + sharded: format!("media/ab/cd/{community}/{SHA}.thumb.jpg"), + legacy: format!("{SHA}.thumb.jpg"), + } + ); + } + + #[test] + fn rejects_noncanonical_or_ambiguous_inputs() { + for sha in ["abc", &"A".repeat(64), &"g".repeat(64)] { + assert_eq!( + legacy_blob_key(sha, "jpg"), + Err(MediaKeyError::InvalidSha256) + ); + } + for ext in ["", "JPG", "tar.gz", "toolongext", "../jpg"] { + assert_eq!( + legacy_blob_key(SHA, ext), + Err(MediaKeyError::InvalidExtension) + ); + } + assert_eq!( + read_candidates(&tenant(1), SHA), + Err(MediaKeyError::InvalidPayloadName) + ); + } +} diff --git a/crates/buzz-media/src/lib.rs b/crates/buzz-media/src/lib.rs index 67896d4ef2..6b0e93f7d1 100644 --- a/crates/buzz-media/src/lib.rs +++ b/crates/buzz-media/src/lib.rs @@ -6,6 +6,8 @@ pub mod auth; pub mod bucket_index; pub mod config; pub mod error; +pub mod keys; +pub mod migration; pub mod storage; pub mod thumbnail; pub mod types; @@ -17,8 +19,12 @@ pub use bucket_index::{ classify_key, fold_bucket_listing, BucketAggregate, BucketSnapshot, CommunityStorage, KeyClass, Page, SweepError, }; -pub use config::{MediaConfig, S3AddressingStyle}; +pub use config::{MediaConfig, MediaMigrationPhase, S3AddressingStyle}; pub use error::MediaError; +pub use keys::{ + legacy_blob_key, legacy_thumb_key, read_candidates, sharded_blob_key, sharded_thumb_key, + MediaKeyError, MediaReadCandidates, +}; pub use storage::{BlobHeadMeta, BlobMeta, ByteStream, MediaStorage}; pub use types::BlobDescriptor; pub use upload::{process_file_upload, process_upload, process_video_upload}; diff --git a/crates/buzz-media/src/migration.rs b/crates/buzz-media/src/migration.rs new file mode 100644 index 0000000000..a47b8be0c1 --- /dev/null +++ b/crates/buzz-media/src/migration.rs @@ -0,0 +1,113 @@ +//! Shared primitives for media-layout maintenance binaries. + +use std::time::Duration; + +use crate::{legacy_blob_key, legacy_thumb_key, sharded_blob_key, sharded_thumb_key, BlobMeta}; +use buzz_core::tenant::CommunityId; + +/// Operation selected by a maintenance binary. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum MigrationOperation { + Backfill, + DeleteLegacy, +} + +/// Derived payload pair for one community sidecar. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct MigrationObject { + pub legacy: String, + pub sharded: String, +} + +/// Parse `_meta//.json`. +pub fn parse_sidecar_key(key: &str) -> Option<(CommunityId, &str)> { + let rest = key.strip_prefix("_meta/")?; + let (community, filename) = rest.split_once('/')?; + if filename.contains('/') { + return None; + } + let sha = filename.strip_suffix(".json")?; + let uuid = uuid::Uuid::parse_str(community).ok()?; + if uuid.hyphenated().to_string() != community { + return None; + } + // Reuse key validation without inventing a second SHA validator. + legacy_thumb_key(sha).ok()?; + Some((CommunityId::from_uuid(uuid), sha)) +} + +/// Derive payload and optional thumbnail pairs represented by a sidecar. +pub fn objects_for_sidecar( + community: CommunityId, + sha: &str, + meta: &BlobMeta, +) -> Result, crate::MediaKeyError> { + let mut objects = vec![MigrationObject { + legacy: legacy_blob_key(sha, &meta.ext)?, + sharded: sharded_blob_key(community, sha, &meta.ext)?, + }]; + if !meta.thumb_url.is_empty() { + objects.push(MigrationObject { + legacy: legacy_thumb_key(sha)?, + sharded: sharded_thumb_key(community, sha)?, + }); + } + Ok(objects) +} + +/// Simple globally paced request limiter. One permit is consumed for every S3 +/// request, so HEAD+copy/delete work stays below the configured request rate. +pub struct RequestPacer { + interval: tokio::time::Interval, +} + +impl RequestPacer { + pub fn new(requests_per_second: u32) -> Self { + let period = Duration::from_secs_f64(1.0 / f64::from(requests_per_second)); + let mut interval = tokio::time::interval(period); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + Self { interval } + } + + pub async fn wait(&mut self) { + self.interval.tick().await; + } +} + +#[cfg(test)] +mod tests { + use super::*; + + const SHA: &str = "abcdef0123456789abcdef0123456789abcdef0123456789abcdef0123456789"; + + #[test] + fn sidecar_parser_is_strict() { + let community = uuid::Uuid::from_u128(1); + assert_eq!( + parse_sidecar_key(&format!("_meta/{community}/{SHA}.json")), + Some((CommunityId::from_uuid(community), SHA)) + ); + for invalid in [ + format!("_meta/{community}/{SHA}.json/extra"), + format!("_meta/{}/{SHA}.json", community.simple()), + format!("_meta/{community}/ABC.json"), + format!("media/{community}/{SHA}.json"), + ] { + assert_eq!(parse_sidecar_key(&invalid), None, "{invalid}"); + } + } + + #[test] + fn derives_blob_and_only_existing_thumbnail_variant() { + let community = CommunityId::from_uuid(uuid::Uuid::from_u128(1)); + let mut meta = BlobMeta { + ext: "jpg".into(), + ..BlobMeta::default() + }; + assert_eq!(objects_for_sidecar(community, SHA, &meta).unwrap().len(), 1); + meta.thumb_url = "https://media.example/thumb".into(); + let objects = objects_for_sidecar(community, SHA, &meta).unwrap(); + assert_eq!(objects.len(), 2); + assert!(objects[1].legacy.ends_with(".thumb.jpg")); + } +} diff --git a/crates/buzz-media/src/storage.rs b/crates/buzz-media/src/storage.rs index cbf980201f..bbb15672c6 100644 --- a/crates/buzz-media/src/storage.rs +++ b/crates/buzz-media/src/storage.rs @@ -5,7 +5,7 @@ use std::pin::Pin; use buzz_core::tenant::{CommunityId, TenantContext}; -use crate::config::{MediaConfig, S3AddressingStyle}; +use crate::config::{MediaConfig, MediaMigrationPhase, S3AddressingStyle}; use crate::error::MediaError; use bytes::Bytes; use s3::creds::Credentials; @@ -18,6 +18,7 @@ pub type ByteStream = Pin, + migration_phase: MediaMigrationPhase, } impl MediaStorage { @@ -66,7 +67,10 @@ impl MediaStorage { S3AddressingStyle::Path => bucket.with_path_style(), S3AddressingStyle::Virtual => bucket, }; - Ok(Self { bucket }) + Ok(Self { + bucket, + migration_phase: config.migration_phase, + }) } /// Store an object from a byte slice. @@ -104,6 +108,123 @@ impl MediaStorage { Ok(()) } + /// Store a media payload according to the configured migration layout. + /// + /// Dual mode writes the sharded primary first and the legacy compatibility + /// copy second. Callers publish sidecars only after this returns success. + pub async fn put_payload( + &self, + ctx: &TenantContext, + sha256: &str, + ext: &str, + bytes: &[u8], + content_type: &str, + ) -> Result { + let legacy = crate::keys::legacy_blob_key(sha256, ext) + .map_err(|e| MediaError::StorageError(e.to_string()))?; + let sharded = crate::keys::sharded_blob_key(ctx.community(), sha256, ext) + .map_err(|e| MediaError::StorageError(e.to_string()))?; + match self.migration_phase { + MediaMigrationPhase::DualReadLegacyWrite => { + self.put(&legacy, bytes, content_type).await? + } + MediaMigrationPhase::DualReadDualWrite => { + self.put(&sharded, bytes, content_type).await?; + self.put(&legacy, bytes, content_type).await?; + } + MediaMigrationPhase::ShardedOnly => self.put(&sharded, bytes, content_type).await?, + } + Ok(match self.migration_phase { + MediaMigrationPhase::DualReadLegacyWrite => legacy, + MediaMigrationPhase::DualReadDualWrite | MediaMigrationPhase::ShardedOnly => sharded, + }) + } + + /// Stream a media payload from disk according to the configured layout. + pub async fn put_payload_file( + &self, + ctx: &TenantContext, + sha256: &str, + ext: &str, + path: &Path, + content_type: &str, + ) -> Result { + let legacy = crate::keys::legacy_blob_key(sha256, ext) + .map_err(|e| MediaError::StorageError(e.to_string()))?; + let sharded = crate::keys::sharded_blob_key(ctx.community(), sha256, ext) + .map_err(|e| MediaError::StorageError(e.to_string()))?; + match self.migration_phase { + MediaMigrationPhase::DualReadLegacyWrite => { + self.put_file(&legacy, path, content_type).await? + } + MediaMigrationPhase::DualReadDualWrite => { + self.put_file(&sharded, path, content_type).await?; + self.put_file(&legacy, path, content_type).await?; + } + MediaMigrationPhase::ShardedOnly => self.put_file(&sharded, path, content_type).await?, + } + Ok(match self.migration_phase { + MediaMigrationPhase::DualReadLegacyWrite => legacy, + MediaMigrationPhase::DualReadDualWrite | MediaMigrationPhase::ShardedOnly => sharded, + }) + } + + /// Store a thumbnail according to the configured migration layout. + pub async fn put_thumbnail( + &self, + ctx: &TenantContext, + sha256: &str, + bytes: &[u8], + ) -> Result { + let legacy = crate::keys::legacy_thumb_key(sha256) + .map_err(|e| MediaError::StorageError(e.to_string()))?; + let sharded = crate::keys::sharded_thumb_key(ctx.community(), sha256) + .map_err(|e| MediaError::StorageError(e.to_string()))?; + match self.migration_phase { + MediaMigrationPhase::DualReadLegacyWrite => { + self.put(&legacy, bytes, "image/jpeg").await? + } + MediaMigrationPhase::DualReadDualWrite => { + self.put(&sharded, bytes, "image/jpeg").await?; + self.put(&legacy, bytes, "image/jpeg").await?; + } + MediaMigrationPhase::ShardedOnly => self.put(&sharded, bytes, "image/jpeg").await?, + } + Ok(match self.migration_phase { + MediaMigrationPhase::DualReadLegacyWrite => legacy, + MediaMigrationPhase::DualReadDualWrite | MediaMigrationPhase::ShardedOnly => sharded, + }) + } + + /// Return the authoritative key only when every layout required by the + /// current write phase already exists. This prevents a re-upload from + /// skipping the missing compatibility copy after a phase change. + pub async fn existing_write_key( + &self, + ctx: &TenantContext, + payload_name: &str, + ) -> Result, MediaError> { + let candidates = + crate::keys::read_candidates(ctx, payload_name).map_err(|_| MediaError::NotFound)?; + match self.migration_phase { + MediaMigrationPhase::DualReadLegacyWrite => self + .head(&candidates.legacy) + .await + .map(|exists| exists.then_some(candidates.legacy)), + MediaMigrationPhase::DualReadDualWrite => { + if self.head(&candidates.sharded).await? && self.head(&candidates.legacy).await? { + Ok(Some(candidates.sharded)) + } else { + Ok(None) + } + } + MediaMigrationPhase::ShardedOnly => self + .head(&candidates.sharded) + .await + .map(|exists| exists.then_some(candidates.sharded)), + } + } + /// Retrieve an object's bytes. pub async fn get(&self, key: &str) -> Result, MediaError> { match self.bucket.get_object(key).await { @@ -177,6 +298,123 @@ impl MediaStorage { } } + /// Resolve a media payload to its new-first, legacy-fallback object key. + /// + /// Only an actual not-found result advances to the legacy candidate. Any + /// authorization, transport, throttling, or service error is returned so + /// the compatibility path cannot mask an unhealthy object store. + pub async fn resolve_read_key( + &self, + ctx: &TenantContext, + payload_name: &str, + ) -> Result { + let candidates = match crate::keys::read_candidates(ctx, payload_name) { + Ok(candidates) => candidates, + Err(_) => { + metrics::counter!( + "buzz_media_s3_read_resolutions_total", + "result" => "missing" + ) + .increment(1); + return Err(MediaError::NotFound); + } + }; + + match self.head_with_metadata(&candidates.sharded).await { + Ok(Some(_)) => { + metrics::counter!( + "buzz_media_s3_read_resolutions_total", + "result" => "sharded" + ) + .increment(1); + Ok(candidates.sharded) + } + Ok(None) if !self.migration_phase.reads_legacy() => { + metrics::counter!( + "buzz_media_s3_read_resolutions_total", + "result" => "missing" + ) + .increment(1); + Err(MediaError::NotFound) + } + Ok(None) => { + metrics::counter!("buzz_media_s3_read_fallbacks_total").increment(1); + match self.head_with_metadata(&candidates.legacy).await { + Ok(Some(_)) => { + metrics::counter!( + "buzz_media_s3_read_resolutions_total", + "result" => "legacy" + ) + .increment(1); + Ok(candidates.legacy) + } + Ok(None) => { + metrics::counter!( + "buzz_media_s3_read_resolutions_total", + "result" => "missing" + ) + .increment(1); + Err(MediaError::NotFound) + } + Err(error) => { + metrics::counter!( + "buzz_media_s3_read_resolutions_total", + "result" => "storage_error" + ) + .increment(1); + Err(error) + } + } + } + Err(error) => { + metrics::counter!( + "buzz_media_s3_read_resolutions_total", + "result" => "storage_error" + ) + .increment(1); + Err(error) + } + } + } + + /// Copy an object within the media bucket using an S3 server-side copy. + pub async fn copy(&self, source: &str, destination: &str) -> Result<(), MediaError> { + self.bucket + .copy_object_internal(source, destination) + .await + .map_err(|e| MediaError::StorageError(e.to_string()))?; + Ok(()) + } + + /// List one bounded page under a prefix for maintenance tools. + pub async fn list_prefix_page( + &self, + prefix: &str, + continuation_token: Option, + start_after: Option, + max_keys: usize, + ) -> Result { + let (result, _status) = self + .bucket + .list_page( + prefix.to_string(), + None, + continuation_token, + start_after, + Some(max_keys), + ) + .await?; + Ok(crate::bucket_index::Page { + objects: result + .contents + .into_iter() + .map(|obj| (obj.key, obj.size)) + .collect(), + next_continuation_token: result.next_continuation_token, + is_truncated: result.is_truncated, + }) + } + /// Build the community-scoped sidecar key for a given sha256 (bare hash). /// /// Raw media bytes remain shared content-addressed CAS (`{sha}.{ext}`), but @@ -289,6 +527,7 @@ mod tests { s3_bucket: "buzz-media".to_string(), s3_region: "us-west-2".to_string(), s3_addressing_style: S3AddressingStyle::Path, + migration_phase: crate::config::MediaMigrationPhase::DualReadLegacyWrite, max_image_bytes: 50 * 1024 * 1024, max_gif_bytes: 10 * 1024 * 1024, max_video_bytes: 524_288_000, diff --git a/crates/buzz-media/src/upload.rs b/crates/buzz-media/src/upload.rs index 524b033280..e2bc039a7b 100644 --- a/crates/buzz-media/src/upload.rs +++ b/crates/buzz-media/src/upload.rs @@ -88,14 +88,18 @@ where .await .map_err(|_| MediaError::Internal)??; - let key = format!("{sha256}.{ext}"); + let legacy_key = crate::keys::legacy_blob_key(&sha256, &ext) + .map_err(|e| MediaError::StorageError(e.to_string()))?; let meta_key = MediaStorage::ctx_sidecar_key(ctx, &sha256); // Idempotent: short-circuit only if BOTH sidecar and blob exist. If the // sidecar exists but the blob is missing, fall through to re-upload. let sidecar_exists = storage.head(&meta_key).await?; - let blob_exists = storage.head(&key).await?; - if sidecar_exists && blob_exists { + let existing_blob_key = match storage.existing_write_key(ctx, &legacy_key).await { + Ok(key) => key, + Err(error) => return Err(error), + }; + if sidecar_exists && existing_blob_key.is_some() { let meta = storage.get_sidecar(ctx, &sha256).await?; // A re-upload of known bytes is still a distinct upload *event*: no // blob PUT happens, so without this record the uploader would be @@ -110,6 +114,7 @@ where UploadEventFacts { sha256: &sha256, ext: &ext, + blob_key: existing_blob_key.as_deref().expect("checked above"), mime: &mime, size: body.len() as u64, uploaded_at: chrono::Utc::now().timestamp(), @@ -138,7 +143,9 @@ where // content-addressed and bounded by the upload size limit, so the storage // cost is negligible. A V2 background GC job can sweep blobs with no // matching sidecar after a grace period. - storage.put(&key, &body, &mime).await?; + let blob_key = storage + .put_payload(ctx, &sha256, &ext, &body, &mime) + .await?; let meta = match prepare_metadata(MetadataInput { sha256: sha256.clone(), @@ -168,6 +175,7 @@ where UploadEventFacts { sha256: &sha256, ext: &ext, + blob_key: &blob_key, mime: &mime, size: body.len() as u64, uploaded_at, @@ -226,7 +234,7 @@ pub async fn process_upload( let ext = mime_to_ext(&mime).to_string(); Ok((mime, ext)) }, - |input| async move { prepare_image_metadata(storage, config, input).await }, + |input| async move { prepare_image_metadata(storage, config, ctx, input).await }, ) .await } @@ -423,13 +431,17 @@ pub async fn process_video_upload( .map_err(|_| MediaError::Internal)??; let ext = "mp4"; - let key = format!("{sha256_hex}.{ext}"); + let legacy_key = crate::keys::legacy_blob_key(&sha256_hex, ext) + .map_err(|e| MediaError::StorageError(e.to_string()))?; let meta_key = MediaStorage::ctx_sidecar_key(ctx, &sha256_hex); // --- 5. Idempotency check --- let sidecar_exists = storage.head(&meta_key).await?; - let blob_exists = storage.head(&key).await?; - if sidecar_exists && blob_exists { + let existing_blob_key = match storage.existing_write_key(ctx, &legacy_key).await { + Ok(key) => key, + Err(error) => return Err(error), + }; + if sidecar_exists && existing_blob_key.is_some() { let meta = storage.get_sidecar(ctx, &sha256_hex).await?; // Re-upload of known bytes: still a distinct upload event — see the // buffered path's short-circuit for the rationale. @@ -442,6 +454,7 @@ pub async fn process_video_upload( UploadEventFacts { sha256: &sha256_hex, ext, + blob_key: existing_blob_key.as_deref().expect("checked above"), mime: &mime, size: file_size, uploaded_at: chrono::Utc::now().timestamp(), @@ -463,7 +476,9 @@ pub async fn process_video_upload( let uploaded_at = chrono::Utc::now().timestamp(); // --- 6. Stream blob from temp file to S3 --- - storage.put_file(&key, &tmp_path, &mime).await?; + let blob_key = storage + .put_payload_file(ctx, &sha256_hex, ext, &tmp_path, &mime) + .await?; drop(tmp); // Free temp file disk space immediately after S3 upload. // --- 7. Build metadata (no thumbnail for video — desktop handles that) --- @@ -488,6 +503,7 @@ pub async fn process_video_upload( UploadEventFacts { sha256: &sha256_hex, ext, + blob_key: &blob_key, mime: &mime, size: file_size, uploaded_at, @@ -513,6 +529,7 @@ pub async fn process_video_upload( async fn prepare_image_metadata( storage: &MediaStorage, config: &MediaConfig, + ctx: &TenantContext, input: MetadataInput, ) -> Result { let body_ref = input.body.clone(); @@ -529,8 +546,7 @@ async fn prepare_image_metadata( meta.uploaded_at = input.uploaded_at; if let Some(ref tb) = thumb_bytes { - let thumb_key = format!("{}.thumb.jpg", input.sha256); - storage.put(&thumb_key, tb, "image/jpeg").await?; + storage.put_thumbnail(ctx, &input.sha256, tb).await?; } Ok(meta) @@ -571,6 +587,7 @@ mod tests { s3_bucket: String::new(), s3_region: "us-east-1".to_string(), s3_addressing_style: crate::config::S3AddressingStyle::Path, + migration_phase: crate::config::MediaMigrationPhase::DualReadLegacyWrite, max_image_bytes: 50 * 1024 * 1024, max_gif_bytes: 10 * 1024 * 1024, max_video_bytes: 524_288_000, diff --git a/crates/buzz-media/src/upload_record.rs b/crates/buzz-media/src/upload_record.rs index 42f57fbd26..7156c871bf 100644 --- a/crates/buzz-media/src/upload_record.rs +++ b/crates/buzz-media/src/upload_record.rs @@ -60,8 +60,12 @@ pub struct UploadRecord { pub event_id: String, /// Content hash of the uploaded bytes (64 lowercase hex chars). pub sha256: String, - /// Canonical extension — consumers derive the blob key `{sha256}.{ext}`. + /// Canonical extension; retained for backward-compatible consumers. pub ext: String, + /// Exact payload object key. New consumers prefer this over deriving a + /// legacy flat key from `sha256` and `ext`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub blob_key: Option, /// Sniffed MIME type of the uploaded bytes. pub mime_type: String, /// Size of the uploaded bytes. @@ -121,6 +125,8 @@ pub struct UploadEventFacts<'a> { pub sha256: &'a str, /// Canonical extension. pub ext: &'a str, + /// Exact object key containing the payload bytes. + pub blob_key: &'a str, /// Sniffed MIME type. pub mime: &'a str, /// Uploaded byte size. @@ -154,6 +160,7 @@ pub async fn record_upload_event( event_id: event_id.clone(), sha256: facts.sha256.to_string(), ext: facts.ext.to_string(), + blob_key: Some(facts.blob_key.to_string()), mime_type: facts.mime.to_string(), size: facts.size, uploaded_at: facts.uploaded_at, @@ -288,6 +295,7 @@ mod tests { event_id: "01J9W3TEST".into(), sha256: "b".repeat(64), ext: "png".into(), + blob_key: Some(format!("{}.png", "b".repeat(64))), mime_type: "image/png".into(), size: 12345, uploaded_at: 1_783_358_352, @@ -303,6 +311,7 @@ mod tests { assert_eq!(json["version"], 1); assert_eq!(json["ext"], "png"); assert_eq!(json["mime_type"], "image/png"); + assert_eq!(json["blob_key"], format!("{}.png", "b".repeat(64))); assert_eq!(json["size"], 12345); assert_eq!(json["ip"], "203.0.113.7"); assert_eq!(json["port"], 51234); @@ -316,6 +325,7 @@ mod tests { event_id: "01J9W3TEST".into(), sha256: "b".repeat(64), ext: "mp4".into(), + blob_key: None, mime_type: "video/mp4".into(), size: 1, uploaded_at: 0, @@ -330,6 +340,7 @@ mod tests { let json = serde_json::to_value(&record).unwrap(); // Omitted, not null — the consumer contract. assert!(json.get("uploader_name").is_none()); + assert!(json.get("blob_key").is_none()); assert!(json.get("ip").is_none()); assert!(json.get("port").is_none()); } @@ -348,6 +359,7 @@ mod tests { let record: UploadRecord = serde_json::from_str(json).unwrap(); assert_eq!(record.version, 1); assert_eq!(record.ip, None); + assert_eq!(record.blob_key, None); } #[test] diff --git a/crates/buzz-media/src/validation.rs b/crates/buzz-media/src/validation.rs index f1387fc9d6..cc2ec9c940 100644 --- a/crates/buzz-media/src/validation.rs +++ b/crates/buzz-media/src/validation.rs @@ -950,6 +950,7 @@ mod tests { s3_bucket: String::new(), s3_region: "us-east-1".to_string(), s3_addressing_style: crate::config::S3AddressingStyle::Path, + migration_phase: crate::config::MediaMigrationPhase::DualReadLegacyWrite, max_image_bytes: 50 * 1024 * 1024, max_gif_bytes: 10 * 1024 * 1024, max_video_bytes: 524_288_000, diff --git a/crates/buzz-media/tests/static_creds_minio.rs b/crates/buzz-media/tests/static_creds_minio.rs index 4c8c10702c..5135c2776a 100644 --- a/crates/buzz-media/tests/static_creds_minio.rs +++ b/crates/buzz-media/tests/static_creds_minio.rs @@ -35,6 +35,7 @@ fn minio_config() -> MediaConfig { .unwrap_or_else(|_| "path".to_string()) .parse() .expect("BUZZ_S3_ADDRESSING_STYLE must be path or virtual"), + migration_phase: buzz_media::MediaMigrationPhase::DualReadLegacyWrite, max_image_bytes: 50 * 1024 * 1024, max_gif_bytes: 10 * 1024 * 1024, max_video_bytes: 524_288_000, diff --git a/crates/buzz-relay/src/api/media.rs b/crates/buzz-relay/src/api/media.rs index fa0401bc26..12e32bcade 100644 --- a/crates/buzz-relay/src/api/media.rs +++ b/crates/buzz-relay/src/api/media.rs @@ -666,7 +666,13 @@ pub(crate) async fn serve_blob_for_tenant( "attachment" }; - let key = resolve_s3_key(&state.media_storage, tenant, sha256_ext).await?; + let key = state + .media_storage + .resolve_read_key( + tenant, + &resolve_payload_name(&state.media_storage, tenant, sha256_ext).await?, + ) + .await?; // Parse optional Range header. let range_header = req_headers @@ -835,7 +841,13 @@ pub async fn head_blob( sidecar_mime }; - let key = resolve_s3_key(&state.media_storage, &tenant, &sha256_ext).await?; + let key = state + .media_storage + .resolve_read_key( + &tenant, + &resolve_payload_name(&state.media_storage, &tenant, &sha256_ext).await?, + ) + .await?; match state.media_storage.head_with_metadata(&key).await? { Some(meta) => { let size_str = meta.size.to_string(); @@ -861,7 +873,7 @@ pub async fn head_blob( /// /// Sidecar-derived extensions are validated as safe tokens to prevent /// object-key confusion if sidecar data is ever tampered with. -async fn resolve_s3_key( +async fn resolve_payload_name( storage: &buzz_media::MediaStorage, tenant: &TenantContext, sha256_ext: &str, diff --git a/crates/buzz-relay/src/config.rs b/crates/buzz-relay/src/config.rs index 85a0ca2efe..346f4518fb 100644 --- a/crates/buzz-relay/src/config.rs +++ b/crates/buzz-relay/src/config.rs @@ -675,6 +675,15 @@ impl Config { )); } }; + let media_migration_phase = match std::env::var("BUZZ_MEDIA_MIGRATION_PHASE") { + Ok(value) => value.parse().map_err(ConfigError::InvalidValue)?, + Err(std::env::VarError::NotPresent) => buzz_media::MediaMigrationPhase::default(), + Err(std::env::VarError::NotUnicode(_)) => { + return Err(ConfigError::InvalidValue( + "BUZZ_MEDIA_MIGRATION_PHASE must be valid Unicode".to_string(), + )); + } + }; let media = buzz_media::MediaConfig { s3_endpoint: std::env::var("BUZZ_S3_ENDPOINT") .unwrap_or_else(|_| "http://localhost:9000".to_string()), @@ -687,6 +696,7 @@ impl Config { .or_else(|_| std::env::var("AWS_REGION")) .unwrap_or_else(|_| "us-east-1".to_string()), s3_addressing_style, + migration_phase: media_migration_phase, max_image_bytes: std::env::var("BUZZ_MAX_IMAGE_BYTES") .ok() .and_then(|v| v.parse().ok()) diff --git a/crates/buzz-relay/src/handlers/imeta.rs b/crates/buzz-relay/src/handlers/imeta.rs index b75060ce6f..fcabb73b3d 100644 --- a/crates/buzz-relay/src/handlers/imeta.rs +++ b/crates/buzz-relay/src/handlers/imeta.rs @@ -244,15 +244,12 @@ pub async fn verify_imeta_blobs( .await .map_err(|_| format!("imeta references nonexistent blob: {x_value}"))?; - // 2. HEAD the actual blob object - let blob_key = format!("{x_value}.{}", sidecar.ext); - let blob_exists = storage - .head(&blob_key) + // 2. Resolve the actual blob object across sharded and legacy layouts. + let blob_name = format!("{x_value}.{}", sidecar.ext); + storage + .resolve_read_key(ctx, &blob_name) .await .map_err(|e| format!("storage error checking blob {x_value}: {e}"))?; - if !blob_exists { - return Err(format!("imeta blob object missing in storage: {x_value}")); - } // 3. Cross-check claimed metadata against sidecar. if !m_value.is_empty() && sidecar.mime_type != m_value { @@ -275,18 +272,12 @@ pub async fn verify_imeta_blobs( } } - // 4. If thumb is claimed, HEAD the thumbnail object too. + // 4. If thumb is claimed, resolve the thumbnail object too. if !thumb_value.is_empty() { - let thumb_key = format!("{x_value}.thumb.jpg"); - let thumb_exists = storage - .head(&thumb_key) + storage + .resolve_read_key(ctx, &format!("{x_value}.thumb.jpg")) .await .map_err(|e| format!("storage error checking thumbnail: {e}"))?; - if !thumb_exists { - return Err(format!( - "imeta thumb references missing thumbnail: {x_value}" - )); - } } // 5. If image (poster frame) is claimed, verify sidecar + blob. @@ -316,16 +307,11 @@ pub async fn verify_imeta_blobs( } } - let img_key = format!("{img_hash}.{}", img_sidecar.ext); - let img_exists = storage - .head(&img_key) + let img_name = format!("{img_hash}.{}", img_sidecar.ext); + storage + .resolve_read_key(ctx, &img_name) .await .map_err(|e| format!("storage error checking poster image: {e}"))?; - if !img_exists { - return Err(format!( - "imeta image references missing poster frame: {img_hash}" - )); - } } } Ok(()) diff --git a/crates/buzz-relay/src/storage_sweep.rs b/crates/buzz-relay/src/storage_sweep.rs index eccadcd835..01141cc6c2 100644 --- a/crates/buzz-relay/src/storage_sweep.rs +++ b/crates/buzz-relay/src/storage_sweep.rs @@ -320,6 +320,10 @@ pub async fn emit_storage_metrics( metrics::gauge!("buzz_storage_orphan_sidecars").set(snapshot.orphan_sidecar_count as f64); metrics::gauge!("buzz_storage_multi_variant_shas").set(snapshot.multi_variant_shas as f64); metrics::gauge!("buzz_storage_multi_variant_bytes").set(snapshot.multi_variant_bytes as f64); + metrics::gauge!("buzz_storage_duplicate_layout_variants") + .set(snapshot.duplicate_layout_variants as f64); + metrics::gauge!("buzz_storage_duplicate_layout_bytes") + .set(snapshot.duplicate_layout_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); diff --git a/deploy/charts/buzz/README.md b/deploy/charts/buzz/README.md index b2778df28b..a368b9fe66 100644 --- a/deploy/charts/buzz/README.md +++ b/deploy/charts/buzz/README.md @@ -94,6 +94,25 @@ disables that probe through `relay.extraEnv`, `/_readiness` does not test object storage; configuration is still parsed strictly, but reachability and addressing errors surface on the first storage operation. +## Media object-key migration + +Media writes stay on the flat legacy layout after an upgrade unless the operator +explicitly advances `BUZZ_MEDIA_MIGRATION_PHASE` (or Helm +`relay.mediaMigrationPhase`). The supported stages are: + +1. `dual-read-legacy-write` (default): write only existing flat keys. Deploy this release first + so every relay can read both layouts; upgrading alone does **not** double-write. +2. `dual-read-dual-write`: after all readers are compatible, write the sharded key and a flat + rollback copy. Monitor `buzz_media_s3_read_resolutions_total`, + `buzz_media_s3_read_fallbacks_total`, and the storage duplicate-layout gauges. +3. `sharded-only`: after backfill/reconciliation and a full rollback window, stop + writing flat copies. Keep compatibility readers deployed while legacy objects + are migrated and verified. + +Do not roll `sharded-only` writers back to a Buzz version that predates sharded reads. +Returning from `sharded-only` to `dual-read-dual-write` does not retroactively recreate legacy copies; +run and verify the backfill before relying on old-version rollback. + ## Relay Pod extensions The chart exposes narrow extension points for init containers, volumes, relay @@ -259,3 +278,40 @@ helm unittest . helm dependency build . ct lint --config ../../../ct.yaml --charts . ``` + +### Backfill and legacy cleanup Jobs + +The relay image includes `/usr/local/bin/buzz-media-layout-backfill` and +`/usr/local/bin/buzz-media-layout-delete-legacy`; both use the relay's +`BUZZ_S3_*` settings or IRSA credential chain. Runnable Kubernetes Job examples +are in [`deploy/kubernetes/examples/`](../../kubernetes/examples/). The tools: + +- list canonical `_meta//.json` sidecars in bounded pages; +- default to 25 total S3 requests/second, configurable with + `BUZZ_MEDIA_MIGRATION_REQUESTS_PER_SECOND`; +- print the last completed sidecar as `checkpoint`; restart with + `BUZZ_MEDIA_MIGRATION_START_AFTER=` if a Job fails; +- are idempotent: backfill skips an existing destination, and deletion skips an + absent legacy source; +- fail closed on malformed metadata or a missing source/destination. + +For an **existing deployment**: + +1. Upgrade with the default `dual-read-legacy-write`; confirm all relay pods can + read both layouts. No new duplicate writes start in this phase. +2. Select `dual-read-dual-write` and observe successful writes and storage + telemetry through a rollback window. +3. Run the backfill Job. Re-run from any logged checkpoint as needed, then run + it again to a clean `copied=0` result. Reconcile storage sweep unknown keys, + duplicate gauges, migration failures, and legacy fallback traffic. +4. Select `sharded-only` only after every supported rollback version understands + sharded keys and reconciliation finds no missing sharded destination. +5. Run the deletion Job with its default `dry-run=true`, review the output, and + retain a recovery window/S3 versions. Only then set `dry-run=false` and + `BUZZ_MEDIA_DELETE_CONFIRM=delete-verified-legacy-media`. The tool checks the + corresponding sharded object immediately before every deletion. + +For a **new empty deployment**, the chart and Compose defaults deliberately use +`dual-read-legacy-write`, matching relay startup defaults. An operator who has +verified the bucket has no legacy media may set `sharded-only` before accepting +the first upload; no backfill or deletion Job is then needed. diff --git a/deploy/charts/buzz/templates/deployment.yaml b/deploy/charts/buzz/templates/deployment.yaml index 67a93138c5..225703dab3 100644 --- a/deploy/charts/buzz/templates/deployment.yaml +++ b/deploy/charts/buzz/templates/deployment.yaml @@ -131,6 +131,7 @@ spec: - { name: BUZZ_REQUIRE_AUTH_TOKEN, value: {{ .Values.relay.requireAuthToken | quote }} } - { name: BUZZ_REQUIRE_RELAY_MEMBERSHIP, value: {{ .Values.relay.requireRelayMembership | quote }} } - { name: BUZZ_REQUIRE_MEDIA_GET_AUTH, value: {{ .Values.relay.requireMediaGetAuth | quote }} } + - { name: BUZZ_MEDIA_MIGRATION_PHASE, value: {{ .Values.relay.mediaMigrationPhase | quote }} } - { name: BUZZ_ALLOW_NIP_OA_AUTH, value: {{ .Values.relay.allowNipOaAuth | quote }} } - { name: BUZZ_PUBKEY_ALLOWLIST, value: {{ .Values.relay.pubkeyAllowlist | quote }} } {{- if .Values.relay.corsOrigins }} diff --git a/deploy/charts/buzz/values.schema.json b/deploy/charts/buzz/values.schema.json index 9cb6a02c9b..57db55ae1a 100644 --- a/deploy/charts/buzz/values.schema.json +++ b/deploy/charts/buzz/values.schema.json @@ -62,6 +62,7 @@ "requireAuthToken": { "type": "boolean" }, "requireRelayMembership": { "type": "boolean" }, "requireMediaGetAuth": { "type": "boolean" }, + "mediaMigrationPhase": { "type": "string", "enum": ["dual-read-legacy-write", "dual-read-dual-write", "sharded-only"] }, "allowNipOaAuth": { "type": "boolean" }, "huddleAudioAvailable": { "type": ["boolean", "null"], diff --git a/deploy/charts/buzz/values.yaml b/deploy/charts/buzz/values.yaml index 810f8a9658..062a019c26 100644 --- a/deploy/charts/buzz/values.yaml +++ b/deploy/charts/buzz/values.yaml @@ -113,6 +113,9 @@ relay: # local development or fully public communities — desktop, mobile, and CLI # clients all attach read auth. requireMediaGetAuth: true + # Media payload write migration: dual-read-legacy-write -> dual-read-dual-write -> sharded-only. The upgrade-safe + # default is dual-read-legacy-write, so installing a new release never starts duplicate writes. + mediaMigrationPhase: dual-read-legacy-write allowNipOaAuth: true pubkeyAllowlist: false corsOrigins: [] diff --git a/deploy/compose/.env.example b/deploy/compose/.env.example index f6ab4fcab9..74f8d13149 100644 --- a/deploy/compose/.env.example +++ b/deploy/compose/.env.example @@ -35,6 +35,9 @@ BUZZ_S3_SECRET_KEY=CHANGE_ME_RANDOM_SECRET_KEY BUZZ_S3_BUCKET=buzz-media # Bundled MinIO uses path-style URLs; deploy/compose/compose.yml pins this. BUZZ_S3_ADDRESSING_STYLE=path +# Safe upgrade default: dual reads and legacy writes. Migrate explicitly in stages: +# dual-read-legacy-write -> dual-read-dual-write -> sharded-only, after compatibility readers and backfill are ready. +BUZZ_MEDIA_MIGRATION_PHASE=dual-read-legacy-write # Optional host ports. Base compose publishes the relay directly on BUZZ_HTTP_PORT. BUZZ_HTTP_PORT=3000 diff --git a/deploy/kubernetes/examples/media-layout-backfill-job.yaml b/deploy/kubernetes/examples/media-layout-backfill-job.yaml new file mode 100644 index 0000000000..51c213078b --- /dev/null +++ b/deploy/kubernetes/examples/media-layout-backfill-job.yaml @@ -0,0 +1,25 @@ +# Copy this example, pin the same image tag as the relay, and supply its S3 +# environment/IRSA. Re-running is safe: existing sharded objects are skipped. +apiVersion: batch/v1 +kind: Job +metadata: + name: buzz-media-layout-backfill +spec: + backoffLimit: 3 + template: + spec: + restartPolicy: Never + serviceAccountName: buzz-relay + containers: + - name: backfill + image: ghcr.io/block/buzz:REPLACE_WITH_RELAY_TAG + command: ["/usr/local/bin/buzz-media-layout-backfill"] + envFrom: + - secretRef: + name: buzz-relay-env + env: + - name: BUZZ_MEDIA_MIGRATION_REQUESTS_PER_SECOND + value: "25" + # Set true for a no-write inventory pass. + - name: BUZZ_MEDIA_MIGRATION_DRY_RUN + value: "false" diff --git a/deploy/kubernetes/examples/media-layout-delete-legacy-job.yaml b/deploy/kubernetes/examples/media-layout-delete-legacy-job.yaml new file mode 100644 index 0000000000..20c871051b --- /dev/null +++ b/deploy/kubernetes/examples/media-layout-delete-legacy-job.yaml @@ -0,0 +1,27 @@ +# Run in dry-run mode first. Destructive mode verifies every sharded destination +# before deleting its legacy source and requires the explicit confirmation value. +apiVersion: batch/v1 +kind: Job +metadata: + name: buzz-media-layout-delete-legacy +spec: + backoffLimit: 1 + template: + spec: + restartPolicy: Never + serviceAccountName: buzz-relay + containers: + - name: delete-legacy + image: ghcr.io/block/buzz:REPLACE_WITH_RELAY_TAG + command: ["/usr/local/bin/buzz-media-layout-delete-legacy"] + envFrom: + - secretRef: + name: buzz-relay-env + env: + - name: BUZZ_MEDIA_MIGRATION_REQUESTS_PER_SECOND + value: "25" + - name: BUZZ_MEDIA_MIGRATION_DRY_RUN + value: "true" + # For destructive mode set dry-run to false and uncomment: + # - name: BUZZ_MEDIA_DELETE_CONFIRM + # value: delete-verified-legacy-media diff --git a/desktop/src-tauri/Cargo.lock b/desktop/src-tauri/Cargo.lock index 9feecbee01..bf10d200ed 100644 --- a/desktop/src-tauri/Cargo.lock +++ b/desktop/src-tauri/Cargo.lock @@ -1128,17 +1128,20 @@ dependencies = [ name = "buzz-media" version = "0.1.0" dependencies = [ + "anyhow", "axum", "blurhash", "buzz-core", "bytes", "chrono", + "clap", "futures-core", "futures-util", "hex", "image", "imagesize", "infer", + "metrics", "mp4", "nostr", "rust-s3", @@ -1150,6 +1153,7 @@ dependencies = [ "tokio", "tokio-util", "tracing", + "tracing-subscriber", "ulid", "uuid", ] @@ -5506,6 +5510,16 @@ dependencies = [ "tracing", ] +[[package]] +name = "metrics" +version = "0.24.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "89550ee9f79e88fef3119de263694973a8adb26c21d75322164fb8c493039fe2" +dependencies = [ + "portable-atomic", + "rapidhash", +] + [[package]] name = "miette" version = "7.6.0" @@ -8057,6 +8071,15 @@ dependencies = [ "rand_core 0.9.5", ] +[[package]] +name = "rapidhash" +version = "4.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5da7e78a036ce858e8d55b7e7dc8ba3a88b78350fd2155d3591bbd966b58589e" +dependencies = [ + "rustversion", +] + [[package]] name = "ratatui" version = "0.30.2" diff --git a/desktop/src-tauri/src/commands/media_snapshot_png.rs b/desktop/src-tauri/src/commands/media_snapshot_png.rs index bcaec6a592..139a92c061 100644 --- a/desktop/src-tauri/src/commands/media_snapshot_png.rs +++ b/desktop/src-tauri/src/commands/media_snapshot_png.rs @@ -205,6 +205,7 @@ mod tests { s3_bucket: String::new(), s3_region: "us-east-1".to_string(), s3_addressing_style: buzz_media_pkg::S3AddressingStyle::Path, + migration_phase: buzz_media_pkg::MediaMigrationPhase::DualReadLegacyWrite, max_image_bytes: 50 * 1024 * 1024, max_gif_bytes: 10 * 1024 * 1024, max_video_bytes: 524_288_000,