diff --git a/crates/buzz-media/src/upload.rs b/crates/buzz-media/src/upload.rs index 7ec784bb1a..0db9dfe0a4 100644 --- a/crates/buzz-media/src/upload.rs +++ b/crates/buzz-media/src/upload.rs @@ -516,23 +516,44 @@ async fn prepare_image_metadata( config: &MediaConfig, input: MetadataInput, ) -> Result { - let body_ref = input.body.clone(); - let mime_ref = input.mime.clone(); - let ext_ref = input.ext.clone(); - let sha256_ref = input.sha256.clone(); - let cfg_ref = config.clone(); - let (mut meta, thumbs) = tokio::task::spawn_blocking(move || { - generate_image_metadata_sync(&cfg_ref, &sha256_ref, &body_ref, &mime_ref, &ext_ref) + let mut meta = store_image_thumbnails( + storage, + config, + &input.sha256, + input.body, + &input.mime, + &input.ext, + ) + .await?; + meta.uploaded_at = input.uploaded_at; + Ok(meta) +} + +/// Generate an image's metadata and store every thumbnail variant. +/// +/// Shared by the upload path and on-demand regeneration of missing thumbnails. +pub async fn store_image_thumbnails( + storage: &MediaStorage, + config: &MediaConfig, + sha256: &str, + body: Bytes, + mime: &str, + ext: &str, +) -> Result { + let (cfg_ref, sha256_ref, mime_ref, ext_ref) = ( + config.clone(), + sha256.to_owned(), + mime.to_owned(), + ext.to_owned(), + ); + let (meta, thumbs) = tokio::task::spawn_blocking(move || { + generate_image_metadata_sync(&cfg_ref, &sha256_ref, &body, &mime_ref, &ext_ref) }) .await .map_err(|_| MediaError::Internal)??; - meta.uploaded_at = input.uploaded_at; - for (thumb, bytes) in THUMBS.into_iter().zip(thumbs.into_iter().flatten()) { - storage - .put(&thumb.key(&input.sha256), &bytes, thumb.mime) - .await?; + storage.put(&thumb.key(sha256), &bytes, thumb.mime).await?; } Ok(meta) diff --git a/crates/buzz-relay/src/api/media.rs b/crates/buzz-relay/src/api/media.rs index 0e9d5aaa3a..17f992b9a8 100644 --- a/crates/buzz-relay/src/api/media.rs +++ b/crates/buzz-relay/src/api/media.rs @@ -982,9 +982,7 @@ pub(crate) async fn serve_blob_for_tenant( match single_range { None => { // Full response — 200 OK. Stream from S3 — never loads full blob into RAM. - let total = state - .media_storage - .head_with_metadata(&key) + let total = head_or_regenerate_thumb(state, tenant, sha256_ext, &key) .await? .ok_or(MediaError::NotFound)? .size; @@ -1004,9 +1002,7 @@ pub(crate) async fn serve_blob_for_tenant( } Some(range_str) => { // S3-native single-range response, capped to bound request memory. - let total = state - .media_storage - .head_with_metadata(&key) + let total = head_or_regenerate_thumb(state, tenant, sha256_ext, &key) .await? .ok_or(MediaError::NotFound)? .size; @@ -1259,7 +1255,7 @@ pub(crate) async fn head_blob( }; let key = resolve_s3_key(&state.media_storage, &tenant, &sha256_ext).await?; - match state.media_storage.head_with_metadata(&key).await? { + match head_or_regenerate_thumb(&state, &tenant, &sha256_ext, &key).await? { Some(meta) => { let size_str = meta.size.to_string(); Ok(( @@ -1277,6 +1273,73 @@ pub(crate) async fn head_blob( } } +/// Concurrent on-demand thumbnail regenerations per relay process. Each holds +/// a full-size decode in memory, so excess requests get a plain 404 instead of +/// waiting. +static THUMB_REGEN: tokio::sync::Semaphore = tokio::sync::Semaphore::const_new(2); + +/// `head_with_metadata`, first regenerating a missing thumbnail variant from +/// its image original (uploaded before the variant existed, or a failed write). +/// +/// Callers have already passed the tenant and sidecar gates for `sha256_ext`. +/// Regeneration failure is logged and reads as absent, never an error. +async fn head_or_regenerate_thumb( + state: &AppState, + tenant: &TenantContext, + sha256_ext: &str, + key: &str, +) -> Result, MediaError> { + let head = state.media_storage.head_with_metadata(key).await?; + let Some((sha256, _)) = Thumb::parse(sha256_ext).filter(|_| head.is_none()) else { + return Ok(head); + }; + match regenerate_thumbs(state, tenant, sha256).await { + Ok(true) => state.media_storage.head_with_metadata(key).await, + Ok(false) => Ok(None), + Err(e) => { + tracing::warn!(sha256, error = %e, "thumbnail regeneration failed"); + Ok(None) + } + } +} + +/// Rebuild every thumbnail of the tenant's image `sha256`. `Ok(false)` means +/// there was nothing to do: not an image, no original, too large, or busy. +async fn regenerate_thumbs( + state: &AppState, + tenant: &TenantContext, + sha256: &str, +) -> Result { + let storage = &state.media_storage; + let meta = storage.get_sidecar(tenant, sha256).await?; + if !meta.mime_type.starts_with("image/") || !is_safe_ext(&meta.ext) { + return Ok(false); + } + let key = format!("{sha256}.{}", meta.ext); + let max = state.config.media.max_image_bytes; + if !storage + .head_with_metadata(&key) + .await? + .is_some_and(|h| h.size <= max) + { + return Ok(false); + } + let Ok(_permit) = THUMB_REGEN.try_acquire() else { + return Ok(false); + }; + let body = bytes::Bytes::from(storage.get(&key).await?); + buzz_media::upload::store_image_thumbnails( + storage, + &state.config.media, + sha256, + body, + &meta.mime_type, + &meta.ext, + ) + .await?; + Ok(true) +} + /// Resolve the S3 key from a URL path segment. /// /// - `sha256.ext` → used as-is (already validated by `validate_media_path`) @@ -4681,5 +4744,186 @@ mod tests { drop(permit_b); drop(permit_a); } + + /// Thumbnail regeneration needs real object storage, so these also + /// need MinIO. They share one image and the process-wide regeneration + /// permits, so they run one at a time. + mod external_infra_minio_tests { + use super::*; + + static SERIAL: std::sync::Mutex<()> = std::sync::Mutex::new(()); + + /// A valid 4x4 PNG with no ancillary chunks. + const PNG: &[u8] = &[ + 0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a, 0x00, 0x00, 0x00, 0x0d, 0x49, 0x48, + 0x44, 0x52, 0x00, 0x00, 0x00, 0x04, 0x00, 0x00, 0x00, 0x04, 0x08, 0x02, 0x00, 0x00, + 0x00, 0x26, 0x93, 0x09, 0x29, 0x00, 0x00, 0x00, 0x10, 0x49, 0x44, 0x41, 0x54, 0x78, + 0x9c, 0x63, 0x38, 0x21, 0x27, 0x07, 0x47, 0x0c, 0xc4, 0x71, 0x00, 0xb1, 0x63, 0x10, + 0x41, 0x3b, 0x4b, 0x63, 0x82, 0x00, 0x00, 0x00, 0x00, 0x49, 0x45, 0x4e, 0x44, 0xae, + 0x42, 0x60, 0x82, + ]; + + struct Fixture { + rt: tokio::runtime::Runtime, + state: Arc, + host: String, + sha: String, + } + + impl Fixture { + /// Upload [`PNG`] through the real route, apply `configure`, and delete its thumbnails. + fn new(configure: impl FnOnce(&mut crate::config::Config)) -> Self { + let (rt, mut state, host) = media_fixture(media_off_test_state()); + let sha = sha256_hex(PNG); + let proof = blossom_upload_auth_value(&Keys::generate(), &host, &sha); + let (status, _, body) = upload_request_with( + &rt, + &state, + "/media/upload", + &host, + (Some(&sha), PNG), + &[&proof], + None, + ); + assert_eq!(status, StatusCode::OK, "upload: {body:?}"); + mutate_state(&mut state, configure); + let fixture = Self { + rt, + state, + host, + sha, + }; + fixture.delete_thumbs(); + fixture + } + + fn thumb_key(&self, ext: &str) -> String { + format!("{}.thumb.{ext}", self.sha) + } + + fn delete_thumbs(&self) { + for ext in ["webp", "jpg"] { + self.rt + .block_on(self.state.media_storage.delete(&self.thumb_key(ext))) + .expect("delete thumbnail"); + } + } + + fn stored(&self, key: &str) -> bool { + self.rt + .block_on(self.state.media_storage.head(key)) + .expect("head") + } + + fn read( + &self, + method: &str, + ext: &str, + ) -> (StatusCode, axum::http::HeaderMap, bytes::Bytes) { + let mut headers = axum::http::HeaderMap::new(); + headers.insert( + axum::http::header::AUTHORIZATION, + blossom_get_auth_value(&Keys::generate(), &self.host, &self.sha) + .parse() + .expect("valid header"), + ); + self.rt.block_on(media_oneshot( + Arc::clone(&self.state), + method, + &format!("/media/{}", self.thumb_key(ext)), + &self.host, + headers, + b"", + )) + } + } + + /// A thumbnail variant missing from storage while its image original + /// exists is rebuilt and served, by GET and by HEAD, for each variant. + #[test] + #[ignore = "requires Postgres and MinIO"] + fn missing_thumbnail_is_regenerated_from_image_original() { + let _serial = SERIAL.lock().unwrap_or_else(|e| e.into_inner()); + let f = Fixture::new(|_| {}); + for (method, ext, mime) in [ + ("GET", "webp", "image/webp"), + ("GET", "jpg", "image/jpeg"), + ("HEAD", "webp", "image/webp"), + ("HEAD", "jpg", "image/jpeg"), + ] { + f.delete_thumbs(); + let (status, headers, body) = f.read(method, ext); + assert_eq!(status, StatusCode::OK, "{method} {ext}: {body:?}"); + assert_eq!(headers["content-type"], mime, "{method} {ext}"); + if method == "GET" { + assert!(!body.is_empty(), "{ext} body"); + } + for ext in ["webp", "jpg"] { + assert!(f.stored(&f.thumb_key(ext)), "{method} {ext}: {ext} stored"); + } + } + } + + /// Regeneration stays a 404 and writes nothing when the sidecar says the + /// blob is not an image, the original is gone, the original exceeds + /// `max_image_bytes`, or every regeneration permit is taken. + #[test] + #[ignore = "requires Postgres and MinIO"] + fn missing_thumbnail_is_not_regenerated_when_unsafe() { + let _serial = SERIAL.lock().unwrap_or_else(|e| e.into_inner()); + let assert_absent = |f: &Fixture, case: &str| { + for ext in ["webp", "jpg"] { + let (status, _, body) = f.read("GET", ext); + assert_eq!(status, StatusCode::NOT_FOUND, "{case} {ext}: {body:?}"); + assert!(!f.stored(&f.thumb_key(ext)), "{case}: {ext} written"); + } + }; + + // Not an image: a text blob under its own hash. + let f = Fixture::new(|_| {}); + let text = format!("not an image {}", uuid::Uuid::new_v4()).into_bytes(); + let sha = sha256_hex(&text); + let tenant = + f.rt.block_on(crate::tenant::bind_community(&f.state.db, &f.host)) + .expect("bind community"); + f.rt.block_on(async { + let storage = &f.state.media_storage; + storage + .put(&format!("{sha}.txt"), &text, "text/plain") + .await + .expect("put text"); + let meta = buzz_media::BlobMeta { + ext: "txt".into(), + mime_type: "text/plain".into(), + size: text.len() as u64, + ..Default::default() + }; + storage + .put_sidecar(&tenant, &sha, &meta) + .await + .expect("put sidecar"); + }); + let text_fixture = Fixture { sha, ..f }; + assert_absent(&text_fixture, "non-image"); + + // Original missing: sidecar still says image. + let f = Fixture::new(|_| {}); + f.rt.block_on(f.state.media_storage.delete(&format!("{}.png", f.sha))) + .expect("delete original"); + assert_absent(&f, "missing original"); + + // Original over the size cap. + let f = Fixture::new(|c| c.media.max_image_bytes = PNG.len() as u64 - 1); + assert_absent(&f, "oversize"); + + // No free permit, then a free one. + let f = Fixture::new(|_| {}); + let held: Vec<_> = std::iter::from_fn(|| THUMB_REGEN.try_acquire().ok()).collect(); + assert!(!held.is_empty(), "fixture must hold a regeneration permit"); + assert_absent(&f, "busy"); + drop(held); + assert_eq!(f.read("GET", "webp").0, StatusCode::OK, "after release"); + } + } } }