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/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/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 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,