External/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-08-18 06:50:31 +02:00
Code Issues Packages Projects Releases Wiki Activity

feat(media): add configurable sharded writes

Browse Source 
Co-authored-by: npub128x7j3pwgm4vs8yra3c42fcgcwcvh94g3luwzkqa376du2q6l0esqcrwch <51cde9442e46eac81c83ec71552708c3b0cb96a88ff8e1581d8fb4de281afbf3@buzz.block.builderlab.xyz>
Signed-off-by: npub128x7j3pwgm4vs8yra3c42fcgcwcvh94g3luwzkqa376du2q6l0esqcrwch <51cde9442e46eac81c83ec71552708c3b0cb96a88ff8e1581d8fb4de281afbf3@buzz.block.builderlab.xyz>
This commit is contained in:
npub128x7j3pwgm4vs8yra3c42fcgcwcvh94g3luwzkqa376du2q6l0esqcrwch
2026-07-31 19:43:51 -04:00
parent 106166b6ec
commit 83bb84fd28
9 changed files with 178 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/buzz-media/src/config.rs
+42 -1
View File
@@ -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<Self, Self::Err> {
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::<MediaKeyLayout>().is_err());
}
#[test]
fn upload_record_knobs_default_off_and_validate() {
assert!(valid_config().validate().is_ok());
crates/buzz-media/src/lib.rs
+1 -1
View File
@@ -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,
crates/buzz-media/src/storage.rs
+89 -2
View File
@@ -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<Box<dyn futures_core::Stream<Item = Result<Bytes, Medi
/// S3-compatible object storage client.
pub struct MediaStorage {
bucket: Box<Bucket>,
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<String, MediaError> {
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<String, MediaError> {
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<String, MediaError> {
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<Vec<u8>, 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,
crates/buzz-media/src/upload.rs
+30 -11
View File
@@ -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<BlobMeta, MediaError> {
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,
crates/buzz-media/src/upload_record.rs
+3 -4
View File
@@ -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,
crates/buzz-media/src/validation.rs
+1
View File
@@ -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,
crates/buzz-media/tests/static_creds_minio.rs
+1
View File
@@ -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,
crates/buzz-relay/src/config.rs
+10
View File
@@ -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())
desktop/src-tauri/src/commands/media_snapshot_png.rs
+1
View File
@@ -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,