From a0219398cb632907d4813d5473ecb27f6ee45b98 Mon Sep 17 00:00:00 2001 From: npub128x7j3pwgm4vs8yra3c42fcgcwcvh94g3luwzkqa376du2q6l0esqcrwch <51cde9442e46eac81c83ec71552708c3b0cb96a88ff8e1581d8fb4de281afbf3@buzz.block.builderlab.xyz> Date: Fri, 31 Jul 2026 17:09:01 -0400 Subject: [PATCH 1/8] feat(media): add sharded compatibility reads Co-authored-by: npub128x7j3pwgm4vs8yra3c42fcgcwcvh94g3luwzkqa376du2q6l0esqcrwch <51cde9442e46eac81c83ec71552708c3b0cb96a88ff8e1581d8fb4de281afbf3@buzz.block.builderlab.xyz> Signed-off-by: npub128x7j3pwgm4vs8yra3c42fcgcwcvh94g3luwzkqa376du2q6l0esqcrwch <51cde9442e46eac81c83ec71552708c3b0cb96a88ff8e1581d8fb4de281afbf3@buzz.block.builderlab.xyz> --- crates/buzz-media/src/keys.rs | 190 ++++++++++++++++++++++++ crates/buzz-media/src/lib.rs | 5 + crates/buzz-media/src/storage.rs | 21 +++ crates/buzz-relay/src/api/media.rs | 18 ++- crates/buzz-relay/src/handlers/imeta.rs | 34 ++--- 5 files changed, 241 insertions(+), 27 deletions(-) create mode 100644 crates/buzz-media/src/keys.rs diff --git a/crates/buzz-media/src/keys.rs b/crates/buzz-media/src/keys.rs new file mode 100644 index 0000000000..695d818cea --- /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: `m/<2>/<2>//.`. +pub fn sharded_blob_key( + community: CommunityId, + sha256: &str, + ext: &str, +) -> Result { + let filename = legacy_blob_key(sha256, ext)?; + Ok(format!( + "m/{}/{}/{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: `m/<2>/<2>//.thumb.jpg`. +pub fn sharded_thumb_key(community: CommunityId, sha256: &str) -> Result { + let filename = legacy_thumb_key(sha256)?; + Ok(format!( + "m/{}/{}/{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!("m/ab/cd/{community}/{SHA}.jpg") + ); + assert_eq!( + sharded_thumb_key(community, SHA).unwrap(), + format!("m/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!("m/ab/cd/{community}/{SHA}.png"), + legacy: format!("{SHA}.png"), + } + ); + assert_eq!( + read_candidates(&ctx, &format!("{SHA}.thumb.jpg")).unwrap(), + MediaReadCandidates { + sharded: format!("m/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..1a9a25b89c 100644 --- a/crates/buzz-media/src/lib.rs +++ b/crates/buzz-media/src/lib.rs @@ -6,6 +6,7 @@ pub mod auth; pub mod bucket_index; pub mod config; pub mod error; +pub mod keys; pub mod storage; pub mod thumbnail; pub mod types; @@ -19,6 +20,10 @@ pub use bucket_index::{ }; pub use config::{MediaConfig, 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/storage.rs b/crates/buzz-media/src/storage.rs index cbf980201f..18c718b8ca 100644 --- a/crates/buzz-media/src/storage.rs +++ b/crates/buzz-media/src/storage.rs @@ -177,6 +177,27 @@ 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 = + crate::keys::read_candidates(ctx, payload_name).map_err(|_| MediaError::NotFound)?; + match self.head_with_metadata(&candidates.sharded).await? { + Some(_) => Ok(candidates.sharded), + None => match self.head_with_metadata(&candidates.legacy).await? { + Some(_) => Ok(candidates.legacy), + None => Err(MediaError::NotFound), + }, + } + } + /// Build the community-scoped sidecar key for a given sha256 (bare hash). /// /// Raw media bytes remain shared content-addressed CAS (`{sha}.{ext}`), but 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/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(()) From 786f4c9d2bee6908698bfd4568efe16bae81c04b Mon Sep 17 00:00:00 2001 From: npub128x7j3pwgm4vs8yra3c42fcgcwcvh94g3luwzkqa376du2q6l0esqcrwch <51cde9442e46eac81c83ec71552708c3b0cb96a88ff8e1581d8fb4de281afbf3@buzz.block.builderlab.xyz> Date: Fri, 31 Jul 2026 19:37:20 -0400 Subject: [PATCH 2/8] feat(media): observe sharded migration reads Classify sharded payload keys in storage sweeps, preserve physical totals, and deduplicate logical legacy/sharded copies. Export read resolution, fallback, and duplicate-layout metrics. Co-authored-by: npub128x7j3pwgm4vs8yra3c42fcgcwcvh94g3luwzkqa376du2q6l0esqcrwch <51cde9442e46eac81c83ec71552708c3b0cb96a88ff8e1581d8fb4de281afbf3@buzz.block.builderlab.xyz> Signed-off-by: npub128x7j3pwgm4vs8yra3c42fcgcwcvh94g3luwzkqa376du2q6l0esqcrwch <51cde9442e46eac81c83ec71552708c3b0cb96a88ff8e1581d8fb4de281afbf3@buzz.block.builderlab.xyz> --- Cargo.lock | 1 + crates/buzz-media/Cargo.toml | 1 + crates/buzz-media/src/bucket_index.rs | 256 ++++++++++++++++++++++--- crates/buzz-media/src/storage.rs | 66 ++++++- crates/buzz-relay/src/storage_sweep.rs | 4 + 5 files changed, 291 insertions(+), 37 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 937ead564a..4fa7227e67 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1063,6 +1063,7 @@ dependencies = [ "image", "imagesize", "infer", + "metrics", "mp4", "nostr", "rust-s3", diff --git a/crates/buzz-media/Cargo.toml b/crates/buzz-media/Cargo.toml index 530ce69c90..58fa5f03a8 100644 --- a/crates/buzz-media/Cargo.toml +++ b/crates/buzz-media/Cargo.toml @@ -32,6 +32,7 @@ tempfile = "3" tokio-util = { version = "0.7", features = ["io"] } futures-util = "0.3" futures-core = "0.3" +metrics = { workspace = true } [dev-dependencies] tokio = { workspace = true, features = ["test-util"] } diff --git a/crates/buzz-media/src/bucket_index.rs b/crates/buzz-media/src/bucket_index.rs index bb83dc517f..38d8614f05 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 `m/{hh}/{hh}/{community-uuid}/{sha256}.thumb.jpg` | +//! | blob | `{sha256}.{ext}` or `m/{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() } +/// `m/{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()? != "m" { + 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!("m/ab/ab/{c}/{s}.png")), + KeyClass::Blob { + community: Some(c), + sha256: s.clone(), + ext: "png".to_string(), + } + ); + assert_eq!( + classify_key(&format!("m/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!("m/ff/ab/{c}/{s}.png"), + format!("m/ab/ff/{c}/{s}.png"), + format!("m/a/ab/{c}/{s}.png"), + format!("m/ab/ab/not-a-uuid/{s}.png"), + format!("m/ab/ab/{c}/{s}.png/extra"), + format!("m/ab/ab/{c}/{}.png", s.to_uppercase()), + format!("m/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!("m/ab/ab/{c}/{s}.jpg"), 100); + agg.fold(&format!("{s}.thumb.jpg"), 20); + agg.fold(&format!("m/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!("m/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/storage.rs b/crates/buzz-media/src/storage.rs index 18c718b8ca..b2278a191d 100644 --- a/crates/buzz-media/src/storage.rs +++ b/crates/buzz-media/src/storage.rs @@ -187,14 +187,64 @@ impl MediaStorage { ctx: &TenantContext, payload_name: &str, ) -> Result { - let candidates = - crate::keys::read_candidates(ctx, payload_name).map_err(|_| MediaError::NotFound)?; - match self.head_with_metadata(&candidates.sharded).await? { - Some(_) => Ok(candidates.sharded), - None => match self.head_with_metadata(&candidates.legacy).await? { - Some(_) => Ok(candidates.legacy), - None => Err(MediaError::NotFound), - }, + 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) => { + 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) + } } } 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); From 49bf37d94c507fe5571e99e797cd1077e5ff10fb Mon Sep 17 00:00:00 2001 From: npub128x7j3pwgm4vs8yra3c42fcgcwcvh94g3luwzkqa376du2q6l0esqcrwch <51cde9442e46eac81c83ec71552708c3b0cb96a88ff8e1581d8fb4de281afbf3@buzz.block.builderlab.xyz> Date: Fri, 31 Jul 2026 17:17:07 -0400 Subject: [PATCH 3/8] feat(media): include payload keys in upload records Co-authored-by: npub128x7j3pwgm4vs8yra3c42fcgcwcvh94g3luwzkqa376du2q6l0esqcrwch <51cde9442e46eac81c83ec71552708c3b0cb96a88ff8e1581d8fb4de281afbf3@buzz.block.builderlab.xyz> Signed-off-by: npub128x7j3pwgm4vs8yra3c42fcgcwcvh94g3luwzkqa376du2q6l0esqcrwch <51cde9442e46eac81c83ec71552708c3b0cb96a88ff8e1581d8fb4de281afbf3@buzz.block.builderlab.xyz> --- crates/buzz-media/src/upload_record.rs | 15 ++++++++++++++- 1 file changed, 14 insertions(+), 1 deletion(-) diff --git a/crates/buzz-media/src/upload_record.rs b/crates/buzz-media/src/upload_record.rs index 42f57fbd26..1d13b8c7ad 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. @@ -154,6 +158,10 @@ pub async fn record_upload_event( event_id: event_id.clone(), sha256: facts.sha256.to_string(), ext: facts.ext.to_string(), + blob_key: Some( + crate::keys::legacy_blob_key(facts.sha256, facts.ext) + .map_err(|e| crate::error::MediaError::StorageError(e.to_string()))?, + ), mime_type: facts.mime.to_string(), size: facts.size, uploaded_at: facts.uploaded_at, @@ -288,6 +296,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 +312,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 +326,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 +341,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 +360,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] From 9d8a7acf46ab69030a723b55560c12869a998679 Mon Sep 17 00:00:00 2001 From: npub128x7j3pwgm4vs8yra3c42fcgcwcvh94g3luwzkqa376du2q6l0esqcrwch <51cde9442e46eac81c83ec71552708c3b0cb96a88ff8e1581d8fb4de281afbf3@buzz.block.builderlab.xyz> Date: Fri, 31 Jul 2026 17:22:24 -0400 Subject: [PATCH 4/8] feat(media): add configurable sharded writes Co-authored-by: npub128x7j3pwgm4vs8yra3c42fcgcwcvh94g3luwzkqa376du2q6l0esqcrwch <51cde9442e46eac81c83ec71552708c3b0cb96a88ff8e1581d8fb4de281afbf3@buzz.block.builderlab.xyz> Signed-off-by: npub128x7j3pwgm4vs8yra3c42fcgcwcvh94g3luwzkqa376du2q6l0esqcrwch <51cde9442e46eac81c83ec71552708c3b0cb96a88ff8e1581d8fb4de281afbf3@buzz.block.builderlab.xyz> --- crates/buzz-media/src/config.rs | 43 ++++++++- crates/buzz-media/src/lib.rs | 2 +- crates/buzz-media/src/storage.rs | 91 ++++++++++++++++++- crates/buzz-media/src/upload.rs | 41 ++++++--- crates/buzz-media/src/upload_record.rs | 7 +- crates/buzz-media/src/validation.rs | 1 + crates/buzz-media/tests/static_creds_minio.rs | 1 + crates/buzz-relay/src/config.rs | 10 ++ .../src/commands/media_snapshot_png.rs | 1 + 9 files changed, 178 insertions(+), 19 deletions(-) diff --git a/crates/buzz-media/src/config.rs b/crates/buzz-media/src/config.rs index 3c70e4afe1..31c6ade133 100644 --- a/crates/buzz-media/src/config.rs +++ b/crates/buzz-media/src/config.rs @@ -33,6 +33,34 @@ impl FromStr for S3AddressingStyle { } } +/// Payload object-key layout used for new media writes. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum MediaKeyLayout { + /// Write only the pre-migration flat key. + #[default] + Legacy, + /// Write the sharded key first, then a flat compatibility copy. + Dual, + /// Write only the hash-leading, community-scoped key. + Sharded, +} + +impl FromStr for MediaKeyLayout { + type Err = String; + + fn from_str(value: &str) -> Result { + match value { + "legacy" => Ok(Self::Legacy), + "dual" => Ok(Self::Dual), + "sharded" => Ok(Self::Sharded), + _ => Err(format!( + "BUZZ_MEDIA_KEY_LAYOUT must be 'legacy', 'dual', or 'sharded', got {value:?}" + )), + } + } +} + fn default_max_video_bytes() -> u64 { 524_288_000 // 500 MB } @@ -67,6 +95,9 @@ pub struct MediaConfig { /// S3 URL addressing style. Defaults to path style for MinIO compatibility. #[serde(default)] pub s3_addressing_style: S3AddressingStyle, + /// Object-key layout for new media payload writes. Defaults to legacy. + #[serde(default)] + pub key_layout: MediaKeyLayout, /// 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 +190,7 @@ impl MediaConfig { #[cfg(test)] mod tests { - use super::{MediaConfig, S3AddressingStyle}; + use super::{MediaConfig, MediaKeyLayout, S3AddressingStyle}; use std::str::FromStr; fn valid_config() -> MediaConfig { @@ -170,6 +201,7 @@ mod tests { s3_bucket: "buzz-media".to_string(), s3_region: "us-east-1".to_string(), s3_addressing_style: S3AddressingStyle::Path, + key_layout: MediaKeyLayout::Legacy, max_image_bytes: 1, max_gif_bytes: 1, max_video_bytes: 1, @@ -210,6 +242,15 @@ mod tests { } } + #[test] + fn media_key_layout_parses_and_defaults_to_legacy() { + assert_eq!(MediaKeyLayout::default(), MediaKeyLayout::Legacy); + assert_eq!("legacy".parse(), Ok(MediaKeyLayout::Legacy)); + assert_eq!("dual".parse(), Ok(MediaKeyLayout::Dual)); + assert_eq!("sharded".parse(), Ok(MediaKeyLayout::Sharded)); + assert!("new".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/lib.rs b/crates/buzz-media/src/lib.rs index 1a9a25b89c..9ddaf920e5 100644 --- a/crates/buzz-media/src/lib.rs +++ b/crates/buzz-media/src/lib.rs @@ -18,7 +18,7 @@ pub use bucket_index::{ classify_key, fold_bucket_listing, BucketAggregate, BucketSnapshot, CommunityStorage, KeyClass, Page, SweepError, }; -pub use config::{MediaConfig, S3AddressingStyle}; +pub use config::{MediaConfig, MediaKeyLayout, S3AddressingStyle}; pub use error::MediaError; pub use keys::{ legacy_blob_key, legacy_thumb_key, read_candidates, sharded_blob_key, sharded_thumb_key, diff --git a/crates/buzz-media/src/storage.rs b/crates/buzz-media/src/storage.rs index b2278a191d..7e61224ac6 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, MediaKeyLayout, S3AddressingStyle}; use crate::error::MediaError; use bytes::Bytes; use s3::creds::Credentials; @@ -18,6 +18,7 @@ pub type ByteStream = Pin, + key_layout: MediaKeyLayout, } impl MediaStorage { @@ -66,7 +67,10 @@ impl MediaStorage { S3AddressingStyle::Path => bucket.with_path_style(), S3AddressingStyle::Virtual => bucket, }; - Ok(Self { bucket }) + Ok(Self { + bucket, + key_layout: config.key_layout, + }) } /// Store an object from a byte slice. @@ -104,6 +108,88 @@ 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.key_layout { + MediaKeyLayout::Legacy => self.put(&legacy, bytes, content_type).await?, + MediaKeyLayout::Dual => { + self.put(&sharded, bytes, content_type).await?; + self.put(&legacy, bytes, content_type).await?; + } + MediaKeyLayout::Sharded => self.put(&sharded, bytes, content_type).await?, + } + Ok(match self.key_layout { + MediaKeyLayout::Legacy => legacy, + MediaKeyLayout::Dual | MediaKeyLayout::Sharded => 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.key_layout { + MediaKeyLayout::Legacy => self.put_file(&legacy, path, content_type).await?, + MediaKeyLayout::Dual => { + self.put_file(&sharded, path, content_type).await?; + self.put_file(&legacy, path, content_type).await?; + } + MediaKeyLayout::Sharded => self.put_file(&sharded, path, content_type).await?, + } + Ok(match self.key_layout { + MediaKeyLayout::Legacy => legacy, + MediaKeyLayout::Dual | MediaKeyLayout::Sharded => 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.key_layout { + MediaKeyLayout::Legacy => self.put(&legacy, bytes, "image/jpeg").await?, + MediaKeyLayout::Dual => { + self.put(&sharded, bytes, "image/jpeg").await?; + self.put(&legacy, bytes, "image/jpeg").await?; + } + MediaKeyLayout::Sharded => self.put(&sharded, bytes, "image/jpeg").await?, + } + Ok(match self.key_layout { + MediaKeyLayout::Legacy => legacy, + MediaKeyLayout::Dual | MediaKeyLayout::Sharded => sharded, + }) + } + /// Retrieve an object's bytes. pub async fn get(&self, key: &str) -> Result, MediaError> { match self.bucket.get_object(key).await { @@ -360,6 +446,7 @@ mod tests { s3_bucket: "buzz-media".to_string(), s3_region: "us-west-2".to_string(), s3_addressing_style: S3AddressingStyle::Path, + key_layout: crate::config::MediaKeyLayout::Legacy, 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..e8a12c6517 100644 --- a/crates/buzz-media/src/upload.rs +++ b/crates/buzz-media/src/upload.rs @@ -88,14 +88,19 @@ 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.resolve_read_key(ctx, &legacy_key).await { + Ok(key) => Some(key), + Err(MediaError::NotFound) => None, + 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 +115,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 +144,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 +176,7 @@ where UploadEventFacts { sha256: &sha256, ext: &ext, + blob_key: &blob_key, mime: &mime, size: body.len() as u64, uploaded_at, @@ -226,7 +235,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 +432,18 @@ 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.resolve_read_key(ctx, &legacy_key).await { + Ok(key) => Some(key), + Err(MediaError::NotFound) => None, + 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 +456,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 +478,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 +505,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 +531,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 +548,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 +589,7 @@ mod tests { s3_bucket: String::new(), s3_region: "us-east-1".to_string(), s3_addressing_style: crate::config::S3AddressingStyle::Path, + key_layout: crate::config::MediaKeyLayout::Legacy, 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 1d13b8c7ad..7156c871bf 100644 --- a/crates/buzz-media/src/upload_record.rs +++ b/crates/buzz-media/src/upload_record.rs @@ -125,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. @@ -158,10 +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( - crate::keys::legacy_blob_key(facts.sha256, facts.ext) - .map_err(|e| crate::error::MediaError::StorageError(e.to_string()))?, - ), + blob_key: Some(facts.blob_key.to_string()), mime_type: facts.mime.to_string(), size: facts.size, uploaded_at: facts.uploaded_at, diff --git a/crates/buzz-media/src/validation.rs b/crates/buzz-media/src/validation.rs index f1387fc9d6..dd1a745f8d 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, + key_layout: crate::config::MediaKeyLayout::Legacy, 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..547dcbc197 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"), + key_layout: buzz_media::MediaKeyLayout::Legacy, 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/config.rs b/crates/buzz-relay/src/config.rs index 85a0ca2efe..98aa5acd8d 100644 --- a/crates/buzz-relay/src/config.rs +++ b/crates/buzz-relay/src/config.rs @@ -675,6 +675,15 @@ impl Config { )); } }; + let media_key_layout = match std::env::var("BUZZ_MEDIA_KEY_LAYOUT") { + Ok(value) => value.parse().map_err(ConfigError::InvalidValue)?, + Err(std::env::VarError::NotPresent) => buzz_media::MediaKeyLayout::default(), + Err(std::env::VarError::NotUnicode(_)) => { + return Err(ConfigError::InvalidValue( + "BUZZ_MEDIA_KEY_LAYOUT 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, + key_layout: media_key_layout, max_image_bytes: std::env::var("BUZZ_MAX_IMAGE_BYTES") .ok() .and_then(|v| v.parse().ok()) diff --git a/desktop/src-tauri/src/commands/media_snapshot_png.rs b/desktop/src-tauri/src/commands/media_snapshot_png.rs index bcaec6a592..97dc2f33c4 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, + key_layout: buzz_media_pkg::MediaKeyLayout::Legacy, max_image_bytes: 50 * 1024 * 1024, max_gif_bytes: 10 * 1024 * 1024, max_video_bytes: 524_288_000, From ff676210e8a59f28b3d959cd2fa657a7ac30248a Mon Sep 17 00:00:00 2001 From: npub128x7j3pwgm4vs8yra3c42fcgcwcvh94g3luwzkqa376du2q6l0esqcrwch <51cde9442e46eac81c83ec71552708c3b0cb96a88ff8e1581d8fb4de281afbf3@buzz.block.builderlab.xyz> Date: Fri, 31 Jul 2026 19:46:17 -0400 Subject: [PATCH 5/8] docs(media): document staged key migration Keep self-hosted upgrades on legacy writes by default, expose the write-layout gate through Helm and Compose, and document the explicit legacy-to-dual-to-sharded migration sequence. Co-authored-by: npub128x7j3pwgm4vs8yra3c42fcgcwcvh94g3luwzkqa376du2q6l0esqcrwch <51cde9442e46eac81c83ec71552708c3b0cb96a88ff8e1581d8fb4de281afbf3@buzz.block.builderlab.xyz> Signed-off-by: npub128x7j3pwgm4vs8yra3c42fcgcwcvh94g3luwzkqa376du2q6l0esqcrwch <51cde9442e46eac81c83ec71552708c3b0cb96a88ff8e1581d8fb4de281afbf3@buzz.block.builderlab.xyz> --- .env.example | 7 +++++++ deploy/charts/buzz/README.md | 19 +++++++++++++++++++ deploy/charts/buzz/templates/deployment.yaml | 1 + deploy/charts/buzz/values.schema.json | 1 + deploy/charts/buzz/values.yaml | 3 +++ deploy/compose/.env.example | 3 +++ 6 files changed, 34 insertions(+) diff --git a/.env.example b/.env.example index b9bfcada0e..a26b1bb2ae 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 legacy, so upgrades preserve +# existing flat-only writes and do not duplicate objects automatically. +# Stages: legacy (flat only) -> dual (sharded + flat rollback copy) -> sharded. +# Deploy compatibility readers everywhere and backfill/reconcile objects before +# advancing; do not roll sharded writers back to a version without sharded reads. +# BUZZ_MEDIA_KEY_LAYOUT=legacy + # ----------------------------------------------------------------------------- # Media Upload Admission # ----------------------------------------------------------------------------- diff --git a/deploy/charts/buzz/README.md b/deploy/charts/buzz/README.md index b2778df28b..b1d06c5e86 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_KEY_LAYOUT` (or Helm +`relay.mediaKeyLayout`). The supported stages are: + +1. `legacy` (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`: 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`: 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` writers back to a Buzz version that predates sharded reads. +Returning from `sharded` to `dual` 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 diff --git a/deploy/charts/buzz/templates/deployment.yaml b/deploy/charts/buzz/templates/deployment.yaml index 67a93138c5..8fb1a7154d 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_KEY_LAYOUT, value: {{ .Values.relay.mediaKeyLayout | 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..a644d33f26 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" }, + "mediaKeyLayout": { "type": "string", "enum": ["legacy", "dual", "sharded"] }, "allowNipOaAuth": { "type": "boolean" }, "huddleAudioAvailable": { "type": ["boolean", "null"], diff --git a/deploy/charts/buzz/values.yaml b/deploy/charts/buzz/values.yaml index 810f8a9658..b6d39cae34 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: legacy -> dual -> sharded. The upgrade-safe + # default is legacy, so installing a new release never starts duplicate writes. + mediaKeyLayout: legacy allowNipOaAuth: true pubkeyAllowlist: false corsOrigins: [] diff --git a/deploy/compose/.env.example b/deploy/compose/.env.example index f6ab4fcab9..eef3f7fb63 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: legacy writes only. Migrate explicitly in stages: +# legacy -> dual -> sharded, after compatibility readers and backfill are ready. +BUZZ_MEDIA_KEY_LAYOUT=legacy # Optional host ports. Base compose publishes the relay directly on BUZZ_HTTP_PORT. BUZZ_HTTP_PORT=3000 From 6089cf0d4b246081a465aa13f2094dc025c32901 Mon Sep 17 00:00:00 2001 From: npub128x7j3pwgm4vs8yra3c42fcgcwcvh94g3luwzkqa376du2q6l0esqcrwch <51cde9442e46eac81c83ec71552708c3b0cb96a88ff8e1581d8fb4de281afbf3@buzz.block.builderlab.xyz> Date: Mon, 3 Aug 2026 10:22:26 -0400 Subject: [PATCH 6/8] feat(media): add phased layout migration Consolidate read and write policy behind one upgrade-safe phase and ship guarded maintenance binaries for backfill and legacy cleanup. Co-authored-by: npub128x7j3pwgm4vs8yra3c42fcgcwcvh94g3luwzkqa376du2q6l0esqcrwch <51cde9442e46eac81c83ec71552708c3b0cb96a88ff8e1581d8fb4de281afbf3@buzz.block.builderlab.xyz> Signed-off-by: npub128x7j3pwgm4vs8yra3c42fcgcwcvh94g3luwzkqa376du2q6l0esqcrwch <51cde9442e46eac81c83ec71552708c3b0cb96a88ff8e1581d8fb4de281afbf3@buzz.block.builderlab.xyz> --- .env.example | 6 +- Cargo.lock | 3 + Dockerfile | 12 +- crates/buzz-media/Cargo.toml | 3 + .../src/bin/buzz-media-layout-backfill.rs | 94 +++++++++++++ .../bin/buzz-media-layout-delete-legacy.rs | 116 ++++++++++++++++ .../src/bin/media_layout_common/mod.rs | 60 ++++++++ crates/buzz-media/src/config.rs | 72 ++++++---- crates/buzz-media/src/lib.rs | 3 +- crates/buzz-media/src/migration.rs | 113 +++++++++++++++ crates/buzz-media/src/storage.rs | 131 ++++++++++++++---- crates/buzz-media/src/upload.rs | 12 +- crates/buzz-media/src/validation.rs | 2 +- crates/buzz-media/tests/static_creds_minio.rs | 2 +- crates/buzz-relay/src/config.rs | 8 +- deploy/charts/buzz/README.md | 51 ++++++- deploy/charts/buzz/templates/deployment.yaml | 2 +- deploy/charts/buzz/values.schema.json | 2 +- deploy/charts/buzz/values.yaml | 6 +- deploy/compose/.env.example | 6 +- .../examples/media-layout-backfill-job.yaml | 25 ++++ .../media-layout-delete-legacy-job.yaml | 27 ++++ .../src/commands/media_snapshot_png.rs | 2 +- 23 files changed, 674 insertions(+), 84 deletions(-) create mode 100644 crates/buzz-media/src/bin/buzz-media-layout-backfill.rs create mode 100644 crates/buzz-media/src/bin/buzz-media-layout-delete-legacy.rs create mode 100644 crates/buzz-media/src/bin/media_layout_common/mod.rs create mode 100644 crates/buzz-media/src/migration.rs create mode 100644 deploy/kubernetes/examples/media-layout-backfill-job.yaml create mode 100644 deploy/kubernetes/examples/media-layout-delete-legacy-job.yaml diff --git a/.env.example b/.env.example index a26b1bb2ae..d962d565c8 100644 --- a/.env.example +++ b/.env.example @@ -95,12 +95,12 @@ BUZZ_S3_BUCKET=buzz-media BUZZ_S3_REGION=us-east-1 BUZZ_S3_ADDRESSING_STYLE=path -# Media payload write migration. Unset defaults to legacy, so upgrades preserve +# 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: legacy (flat only) -> dual (sharded + flat rollback copy) -> sharded. +# 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_KEY_LAYOUT=legacy +# BUZZ_MEDIA_MIGRATION_PHASE=dual-read-legacy-write # ----------------------------------------------------------------------------- # Media Upload Admission diff --git a/Cargo.lock b/Cargo.lock index 4fa7227e67..c62b0c057e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1052,11 +1052,13 @@ dependencies = [ name = "buzz-media" version = "0.1.0" dependencies = [ + "anyhow", "axum", "blurhash", "buzz-core", "bytes", "chrono", + "clap", "futures-core", "futures-util", "hex", @@ -1075,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 58fa5f03a8..78db7217db 100644 --- a/crates/buzz-media/Cargo.toml +++ b/crates/buzz-media/Cargo.toml @@ -33,6 +33,9 @@ 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/config.rs b/crates/buzz-media/src/config.rs index 31c6ade133..dd5ff814da 100644 --- a/crates/buzz-media/src/config.rs +++ b/crates/buzz-media/src/config.rs @@ -33,29 +33,41 @@ impl FromStr for S3AddressingStyle { } } -/// Payload object-key layout used for new media writes. +/// Relay media migration phase controlling both payload reads and writes. #[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Deserialize)] -#[serde(rename_all = "lowercase")] -pub enum MediaKeyLayout { - /// Write only the pre-migration flat key. +pub enum MediaMigrationPhase { + /// Compatibility default: prefer sharded reads with legacy fallback, but + /// write only legacy keys. Upgrades therefore do not begin double-writing. #[default] - Legacy, - /// Write the sharded key first, then a flat compatibility copy. - Dual, - /// Write only the hash-leading, community-scoped key. - Sharded, + #[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 MediaKeyLayout { +impl FromStr for MediaMigrationPhase { type Err = String; fn from_str(value: &str) -> Result { match value { - "legacy" => Ok(Self::Legacy), - "dual" => Ok(Self::Dual), - "sharded" => Ok(Self::Sharded), + "dual-read-legacy-write" => Ok(Self::DualReadLegacyWrite), + "dual-read-dual-write" => Ok(Self::DualReadDualWrite), + "sharded-only" => Ok(Self::ShardedOnly), _ => Err(format!( - "BUZZ_MEDIA_KEY_LAYOUT must be 'legacy', 'dual', or 'sharded', got {value:?}" + "BUZZ_MEDIA_MIGRATION_PHASE must be 'dual-read-legacy-write', \ + 'dual-read-dual-write', or 'sharded-only', got {value:?}" )), } } @@ -95,9 +107,10 @@ pub struct MediaConfig { /// S3 URL addressing style. Defaults to path style for MinIO compatibility. #[serde(default)] pub s3_addressing_style: S3AddressingStyle, - /// Object-key layout for new media payload writes. Defaults to legacy. + /// Relay phase controlling media payload reads and writes. Defaults to + /// dual reads with legacy-only writes for upgrade safety. #[serde(default)] - pub key_layout: MediaKeyLayout, + 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. @@ -190,7 +203,7 @@ impl MediaConfig { #[cfg(test)] mod tests { - use super::{MediaConfig, MediaKeyLayout, S3AddressingStyle}; + use super::{MediaConfig, MediaMigrationPhase, S3AddressingStyle}; use std::str::FromStr; fn valid_config() -> MediaConfig { @@ -201,7 +214,7 @@ mod tests { s3_bucket: "buzz-media".to_string(), s3_region: "us-east-1".to_string(), s3_addressing_style: S3AddressingStyle::Path, - key_layout: MediaKeyLayout::Legacy, + migration_phase: MediaMigrationPhase::DualReadLegacyWrite, max_image_bytes: 1, max_gif_bytes: 1, max_video_bytes: 1, @@ -243,12 +256,23 @@ mod tests { } #[test] - fn media_key_layout_parses_and_defaults_to_legacy() { - assert_eq!(MediaKeyLayout::default(), MediaKeyLayout::Legacy); - assert_eq!("legacy".parse(), Ok(MediaKeyLayout::Legacy)); - assert_eq!("dual".parse(), Ok(MediaKeyLayout::Dual)); - assert_eq!("sharded".parse(), Ok(MediaKeyLayout::Sharded)); - assert!("new".parse::().is_err()); + 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] diff --git a/crates/buzz-media/src/lib.rs b/crates/buzz-media/src/lib.rs index 9ddaf920e5..6b0e93f7d1 100644 --- a/crates/buzz-media/src/lib.rs +++ b/crates/buzz-media/src/lib.rs @@ -7,6 +7,7 @@ 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; @@ -18,7 +19,7 @@ pub use bucket_index::{ classify_key, fold_bucket_listing, BucketAggregate, BucketSnapshot, CommunityStorage, KeyClass, Page, SweepError, }; -pub use config::{MediaConfig, MediaKeyLayout, 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, diff --git a/crates/buzz-media/src/migration.rs b/crates/buzz-media/src/migration.rs new file mode 100644 index 0000000000..7cc8a7bf35 --- /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!("m/{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 7e61224ac6..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, MediaKeyLayout, S3AddressingStyle}; +use crate::config::{MediaConfig, MediaMigrationPhase, S3AddressingStyle}; use crate::error::MediaError; use bytes::Bytes; use s3::creds::Credentials; @@ -18,7 +18,7 @@ pub type ByteStream = Pin, - key_layout: MediaKeyLayout, + migration_phase: MediaMigrationPhase, } impl MediaStorage { @@ -69,7 +69,7 @@ impl MediaStorage { }; Ok(Self { bucket, - key_layout: config.key_layout, + migration_phase: config.migration_phase, }) } @@ -124,17 +124,19 @@ impl MediaStorage { .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.key_layout { - MediaKeyLayout::Legacy => self.put(&legacy, bytes, content_type).await?, - MediaKeyLayout::Dual => { + 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?; } - MediaKeyLayout::Sharded => self.put(&sharded, bytes, content_type).await?, + MediaMigrationPhase::ShardedOnly => self.put(&sharded, bytes, content_type).await?, } - Ok(match self.key_layout { - MediaKeyLayout::Legacy => legacy, - MediaKeyLayout::Dual | MediaKeyLayout::Sharded => sharded, + Ok(match self.migration_phase { + MediaMigrationPhase::DualReadLegacyWrite => legacy, + MediaMigrationPhase::DualReadDualWrite | MediaMigrationPhase::ShardedOnly => sharded, }) } @@ -151,17 +153,19 @@ impl MediaStorage { .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.key_layout { - MediaKeyLayout::Legacy => self.put_file(&legacy, path, content_type).await?, - MediaKeyLayout::Dual => { + 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?; } - MediaKeyLayout::Sharded => self.put_file(&sharded, path, content_type).await?, + MediaMigrationPhase::ShardedOnly => self.put_file(&sharded, path, content_type).await?, } - Ok(match self.key_layout { - MediaKeyLayout::Legacy => legacy, - MediaKeyLayout::Dual | MediaKeyLayout::Sharded => sharded, + Ok(match self.migration_phase { + MediaMigrationPhase::DualReadLegacyWrite => legacy, + MediaMigrationPhase::DualReadDualWrite | MediaMigrationPhase::ShardedOnly => sharded, }) } @@ -176,20 +180,51 @@ impl MediaStorage { .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.key_layout { - MediaKeyLayout::Legacy => self.put(&legacy, bytes, "image/jpeg").await?, - MediaKeyLayout::Dual => { + 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?; } - MediaKeyLayout::Sharded => self.put(&sharded, bytes, "image/jpeg").await?, + MediaMigrationPhase::ShardedOnly => self.put(&sharded, bytes, "image/jpeg").await?, } - Ok(match self.key_layout { - MediaKeyLayout::Legacy => legacy, - MediaKeyLayout::Dual | MediaKeyLayout::Sharded => sharded, + 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 { @@ -294,6 +329,14 @@ impl MediaStorage { .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 { @@ -334,6 +377,44 @@ impl MediaStorage { } } + /// 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 @@ -446,7 +527,7 @@ mod tests { s3_bucket: "buzz-media".to_string(), s3_region: "us-west-2".to_string(), s3_addressing_style: S3AddressingStyle::Path, - key_layout: crate::config::MediaKeyLayout::Legacy, + 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 e8a12c6517..e2bc039a7b 100644 --- a/crates/buzz-media/src/upload.rs +++ b/crates/buzz-media/src/upload.rs @@ -95,9 +95,8 @@ where // 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 existing_blob_key = match storage.resolve_read_key(ctx, &legacy_key).await { - Ok(key) => Some(key), - Err(MediaError::NotFound) => None, + 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() { @@ -438,9 +437,8 @@ pub async fn process_video_upload( // --- 5. Idempotency check --- let sidecar_exists = storage.head(&meta_key).await?; - let existing_blob_key = match storage.resolve_read_key(ctx, &legacy_key).await { - Ok(key) => Some(key), - Err(MediaError::NotFound) => None, + 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() { @@ -589,7 +587,7 @@ mod tests { s3_bucket: String::new(), s3_region: "us-east-1".to_string(), s3_addressing_style: crate::config::S3AddressingStyle::Path, - key_layout: crate::config::MediaKeyLayout::Legacy, + 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/validation.rs b/crates/buzz-media/src/validation.rs index dd1a745f8d..cc2ec9c940 100644 --- a/crates/buzz-media/src/validation.rs +++ b/crates/buzz-media/src/validation.rs @@ -950,7 +950,7 @@ mod tests { s3_bucket: String::new(), s3_region: "us-east-1".to_string(), s3_addressing_style: crate::config::S3AddressingStyle::Path, - key_layout: crate::config::MediaKeyLayout::Legacy, + 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 547dcbc197..5135c2776a 100644 --- a/crates/buzz-media/tests/static_creds_minio.rs +++ b/crates/buzz-media/tests/static_creds_minio.rs @@ -35,7 +35,7 @@ fn minio_config() -> MediaConfig { .unwrap_or_else(|_| "path".to_string()) .parse() .expect("BUZZ_S3_ADDRESSING_STYLE must be path or virtual"), - key_layout: buzz_media::MediaKeyLayout::Legacy, + 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/config.rs b/crates/buzz-relay/src/config.rs index 98aa5acd8d..346f4518fb 100644 --- a/crates/buzz-relay/src/config.rs +++ b/crates/buzz-relay/src/config.rs @@ -675,12 +675,12 @@ impl Config { )); } }; - let media_key_layout = match std::env::var("BUZZ_MEDIA_KEY_LAYOUT") { + 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::MediaKeyLayout::default(), + Err(std::env::VarError::NotPresent) => buzz_media::MediaMigrationPhase::default(), Err(std::env::VarError::NotUnicode(_)) => { return Err(ConfigError::InvalidValue( - "BUZZ_MEDIA_KEY_LAYOUT must be valid Unicode".to_string(), + "BUZZ_MEDIA_MIGRATION_PHASE must be valid Unicode".to_string(), )); } }; @@ -696,7 +696,7 @@ impl Config { .or_else(|_| std::env::var("AWS_REGION")) .unwrap_or_else(|_| "us-east-1".to_string()), s3_addressing_style, - key_layout: media_key_layout, + 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/deploy/charts/buzz/README.md b/deploy/charts/buzz/README.md index b1d06c5e86..a368b9fe66 100644 --- a/deploy/charts/buzz/README.md +++ b/deploy/charts/buzz/README.md @@ -97,20 +97,20 @@ 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_KEY_LAYOUT` (or Helm -`relay.mediaKeyLayout`). The supported stages are: +explicitly advances `BUZZ_MEDIA_MIGRATION_PHASE` (or Helm +`relay.mediaMigrationPhase`). The supported stages are: -1. `legacy` (default): write only existing flat keys. Deploy this release first +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`: after all readers are compatible, write the sharded key and a flat +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`: after backfill/reconciliation and a full rollback window, stop +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` writers back to a Buzz version that predates sharded reads. -Returning from `sharded` to `dual` does not retroactively recreate legacy copies; +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 @@ -278,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 8fb1a7154d..225703dab3 100644 --- a/deploy/charts/buzz/templates/deployment.yaml +++ b/deploy/charts/buzz/templates/deployment.yaml @@ -131,7 +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_KEY_LAYOUT, value: {{ .Values.relay.mediaKeyLayout | 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 a644d33f26..57db55ae1a 100644 --- a/deploy/charts/buzz/values.schema.json +++ b/deploy/charts/buzz/values.schema.json @@ -62,7 +62,7 @@ "requireAuthToken": { "type": "boolean" }, "requireRelayMembership": { "type": "boolean" }, "requireMediaGetAuth": { "type": "boolean" }, - "mediaKeyLayout": { "type": "string", "enum": ["legacy", "dual", "sharded"] }, + "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 b6d39cae34..062a019c26 100644 --- a/deploy/charts/buzz/values.yaml +++ b/deploy/charts/buzz/values.yaml @@ -113,9 +113,9 @@ relay: # local development or fully public communities — desktop, mobile, and CLI # clients all attach read auth. requireMediaGetAuth: true - # Media payload write migration: legacy -> dual -> sharded. The upgrade-safe - # default is legacy, so installing a new release never starts duplicate writes. - mediaKeyLayout: legacy + # 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 eef3f7fb63..74f8d13149 100644 --- a/deploy/compose/.env.example +++ b/deploy/compose/.env.example @@ -35,9 +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: legacy writes only. Migrate explicitly in stages: -# legacy -> dual -> sharded, after compatibility readers and backfill are ready. -BUZZ_MEDIA_KEY_LAYOUT=legacy +# 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/src/commands/media_snapshot_png.rs b/desktop/src-tauri/src/commands/media_snapshot_png.rs index 97dc2f33c4..139a92c061 100644 --- a/desktop/src-tauri/src/commands/media_snapshot_png.rs +++ b/desktop/src-tauri/src/commands/media_snapshot_png.rs @@ -205,7 +205,7 @@ mod tests { s3_bucket: String::new(), s3_region: "us-east-1".to_string(), s3_addressing_style: buzz_media_pkg::S3AddressingStyle::Path, - key_layout: buzz_media_pkg::MediaKeyLayout::Legacy, + 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, From 4be3093c55a3b9d8292a7d20482e8e7664effe55 Mon Sep 17 00:00:00 2001 From: npub128x7j3pwgm4vs8yra3c42fcgcwcvh94g3luwzkqa376du2q6l0esqcrwch <51cde9442e46eac81c83ec71552708c3b0cb96a88ff8e1581d8fb4de281afbf3@buzz.block.builderlab.xyz> Date: Mon, 3 Aug 2026 10:27:12 -0400 Subject: [PATCH 7/8] chore(desktop): update media dependency lock Co-authored-by: npub128x7j3pwgm4vs8yra3c42fcgcwcvh94g3luwzkqa376du2q6l0esqcrwch <51cde9442e46eac81c83ec71552708c3b0cb96a88ff8e1581d8fb4de281afbf3@buzz.block.builderlab.xyz> Signed-off-by: npub128x7j3pwgm4vs8yra3c42fcgcwcvh94g3luwzkqa376du2q6l0esqcrwch <51cde9442e46eac81c83ec71552708c3b0cb96a88ff8e1581d8fb4de281afbf3@buzz.block.builderlab.xyz> --- desktop/src-tauri/Cargo.lock | 23 +++++++++++++++++++++++ 1 file changed, 23 insertions(+) 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" From 7bfd8853f0879ff93fded2e0f0d4d188f93524b1 Mon Sep 17 00:00:00 2001 From: npub128x7j3pwgm4vs8yra3c42fcgcwcvh94g3luwzkqa376du2q6l0esqcrwch <51cde9442e46eac81c83ec71552708c3b0cb96a88ff8e1581d8fb4de281afbf3@buzz.block.builderlab.xyz> Date: Mon, 3 Aug 2026 11:36:51 -0400 Subject: [PATCH 8/8] fix(media): use explicit media key prefix Co-authored-by: npub128x7j3pwgm4vs8yra3c42fcgcwcvh94g3luwzkqa376du2q6l0esqcrwch <51cde9442e46eac81c83ec71552708c3b0cb96a88ff8e1581d8fb4de281afbf3@buzz.block.builderlab.xyz> Signed-off-by: npub128x7j3pwgm4vs8yra3c42fcgcwcvh94g3luwzkqa376du2q6l0esqcrwch <51cde9442e46eac81c83ec71552708c3b0cb96a88ff8e1581d8fb4de281afbf3@buzz.block.builderlab.xyz> --- crates/buzz-media/src/bucket_index.rs | 32 +++++++++++++-------------- crates/buzz-media/src/keys.rs | 16 +++++++------- crates/buzz-media/src/migration.rs | 2 +- 3 files changed, 25 insertions(+), 25 deletions(-) diff --git a/crates/buzz-media/src/bucket_index.rs b/crates/buzz-media/src/bucket_index.rs index 38d8614f05..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` or `m/{hh}/{hh}/{community-uuid}/{sha256}.thumb.jpg` | -//! | blob | `{sha256}.{ext}` or `m/{hh}/{hh}/{community-uuid}/{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 | @@ -155,11 +155,11 @@ fn parse_canonical_uuid(s: &str) -> Option { Uuid::parse_str(s).ok() } -/// `m/{sha[0:2]}/{sha[2:4]}/{community}/{filename}`. The filename's digest +/// `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()? != "m" { + if segments.next()? != "media" { return None; } let shard_1 = segments.next()?; @@ -586,7 +586,7 @@ mod tests { let s = sha(0xab); let c = community(10); assert_eq!( - classify_key(&format!("m/ab/ab/{c}/{s}.png")), + classify_key(&format!("media/ab/ab/{c}/{s}.png")), KeyClass::Blob { community: Some(c), sha256: s.clone(), @@ -594,7 +594,7 @@ mod tests { } ); assert_eq!( - classify_key(&format!("m/ab/ab/{c}/{s}.thumb.jpg")), + classify_key(&format!("media/ab/ab/{c}/{s}.thumb.jpg")), KeyClass::Thumb { community: Some(c), sha256: s, @@ -607,13 +607,13 @@ mod tests { let s = sha(0xab); let c = community(11); for key in [ - format!("m/ff/ab/{c}/{s}.png"), - format!("m/ab/ff/{c}/{s}.png"), - format!("m/a/ab/{c}/{s}.png"), - format!("m/ab/ab/not-a-uuid/{s}.png"), - format!("m/ab/ab/{c}/{s}.png/extra"), - format!("m/ab/ab/{c}/{}.png", s.to_uppercase()), - format!("m/ab/ab/{c}/{s}.tar.gz"), + 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}"); } @@ -725,9 +725,9 @@ mod tests { let c = community(12); let mut agg = BucketAggregate::default(); agg.fold(&format!("{s}.jpg"), 100); - agg.fold(&format!("m/ab/ab/{c}/{s}.jpg"), 100); + agg.fold(&format!("media/ab/ab/{c}/{s}.jpg"), 100); agg.fold(&format!("{s}.thumb.jpg"), 20); - agg.fold(&format!("m/ab/ab/{c}/{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(); @@ -748,7 +748,7 @@ mod tests { let sharded_community = community(13); let other_community = community(14); let mut agg = BucketAggregate::default(); - agg.fold(&format!("m/cd/cd/{sharded_community}/{s}.jpg"), 200); + 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); diff --git a/crates/buzz-media/src/keys.rs b/crates/buzz-media/src/keys.rs index 695d818cea..de604684a5 100644 --- a/crates/buzz-media/src/keys.rs +++ b/crates/buzz-media/src/keys.rs @@ -60,7 +60,7 @@ pub fn legacy_blob_key(sha256: &str, ext: &str) -> Result Ok(format!("{sha256}.{ext}")) } -/// Hash-leading blob key: `m/<2>/<2>//.`. +/// Hash-leading blob key: `media/<2>/<2>//.`. pub fn sharded_blob_key( community: CommunityId, sha256: &str, @@ -68,7 +68,7 @@ pub fn sharded_blob_key( ) -> Result { let filename = legacy_blob_key(sha256, ext)?; Ok(format!( - "m/{}/{}/{community}/{filename}", + "media/{}/{}/{community}/{filename}", &sha256[..2], &sha256[2..4] )) @@ -80,11 +80,11 @@ pub fn legacy_thumb_key(sha256: &str) -> Result { Ok(format!("{sha256}.thumb.jpg")) } -/// Hash-leading thumbnail key: `m/<2>/<2>//.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!( - "m/{}/{}/{community}/{filename}", + "media/{}/{}/{community}/{filename}", &sha256[..2], &sha256[2..4] )) @@ -135,11 +135,11 @@ mod tests { assert_eq!( sharded_blob_key(community, SHA, "jpg").unwrap(), - format!("m/ab/cd/{community}/{SHA}.jpg") + format!("media/ab/cd/{community}/{SHA}.jpg") ); assert_eq!( sharded_thumb_key(community, SHA).unwrap(), - format!("m/ab/cd/{community}/{SHA}.thumb.jpg") + format!("media/ab/cd/{community}/{SHA}.thumb.jpg") ); assert_ne!( sharded_blob_key(community, SHA, "jpg").unwrap(), @@ -155,14 +155,14 @@ mod tests { assert_eq!( read_candidates(&ctx, &format!("{SHA}.png")).unwrap(), MediaReadCandidates { - sharded: format!("m/ab/cd/{community}/{SHA}.png"), + 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!("m/ab/cd/{community}/{SHA}.thumb.jpg"), + sharded: format!("media/ab/cd/{community}/{SHA}.thumb.jpg"), legacy: format!("{SHA}.thumb.jpg"), } ); diff --git a/crates/buzz-media/src/migration.rs b/crates/buzz-media/src/migration.rs index 7cc8a7bf35..a47b8be0c1 100644 --- a/crates/buzz-media/src/migration.rs +++ b/crates/buzz-media/src/migration.rs @@ -91,7 +91,7 @@ mod tests { format!("_meta/{community}/{SHA}.json/extra"), format!("_meta/{}/{SHA}.json", community.simple()), format!("_meta/{community}/ABC.json"), - format!("m/{community}/{SHA}.json"), + format!("media/{community}/{SHA}.json"), ] { assert_eq!(parse_sidecar_key(&invalid), None, "{invalid}"); }