From a8219bfeb4a508d316f2794377c71f70ff17a6ed 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/2] 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 e5bd12bd46ab8ada99e9fccd005cbca3421cd0e0 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/2] 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 9a3f91671d..ee97abdb56 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1043,6 +1043,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);