diff --git a/src/infrastructure/services/thumbnail_service.rs b/src/infrastructure/services/thumbnail_service.rs index 23924f1c..afe9702a 100644 --- a/src/infrastructure/services/thumbnail_service.rs +++ b/src/infrastructure/services/thumbnail_service.rs @@ -258,7 +258,10 @@ impl ThumbnailService { /// Get a thumbnail from raw image bytes, generating it if needed. /// /// This is the storage-model-safe entrypoint for CDC/manifest-backed - /// blobs where no single local source file exists on disk. + /// blobs where no single local source file exists on disk. Prefer + /// [`Self::get_thumbnail_from_blob`] on request paths β€” it defers the + /// full blob read until a decode permit is held, so a stampede of + /// cache misses cannot stack one source image per request in RAM. pub async fn get_thumbnail_from_bytes( &self, file_id: &str, @@ -287,30 +290,12 @@ impl ThumbnailService { return Bytes::from(data); } - tracing::info!("🎨 Generating thumbnail: {} {:?}", file_id_owned, size); - match Self::generate_thumbnail_from_data( - original_data, - size, - self.generation_timeout, - ) - .await - { - Ok(bytes) => { - if let Some(parent) = thumb_path.parent() { - let _ = fs::create_dir_all(parent).await; - } - let _ = fs::write(&thumb_path, &bytes).await; - bytes - } - Err(e) => { - tracing::warn!( - "Thumbnail generation failed for {} {:?}: {e}", - file_id_owned, - size - ); - Bytes::new() - } - } + let Ok(_permit) = self.decode_semaphore.acquire().await else { + tracing::warn!("Decode semaphore closed, skipping {}", file_id_owned); + return Bytes::new(); + }; + self.generate_and_persist(&file_id_owned, &thumb_path, size, original_data) + .await }) .await; @@ -325,6 +310,106 @@ impl ThumbnailService { Ok(bytes) } + /// Get a thumbnail for a content-addressed blob, generating it if needed. + /// + /// Request-path entrypoint: on a memory+disk cache miss the source blob + /// is read **after** a decode permit is acquired, so peak RAM under a + /// thumbnail stampede is `permits Γ— image size` instead of + /// `in-flight requests Γ— image size`. moka's per-key init additionally + /// collapses concurrent requests for the same thumbnail into one read. + pub async fn get_thumbnail_from_blob( + &self, + file_id: &str, + blob_hash: &str, + size: ThumbnailSize, + dedup: Arc, + ) -> Result { + let cache_key = ThumbnailCacheKey { + file_id: file_id.to_string(), + size, + }; + + let thumb_path = self.get_thumbnail_path(blob_hash, size); + let file_id_owned = file_id.to_string(); + let blob_hash_owned = blob_hash.to_string(); + + let entry = self + .cache + .entry(cache_key) + .or_insert_with(async move { + if let Ok(data) = fs::read(&thumb_path).await { + tracing::debug!( + "πŸ’Ύ Thumbnail loaded from disk: {} {:?}", + file_id_owned, + size + ); + return Bytes::from(data); + } + + let Ok(_permit) = self.decode_semaphore.acquire().await else { + tracing::warn!("Decode semaphore closed, skipping {}", file_id_owned); + return Bytes::new(); + }; + let original_data = match dedup.read_blob_bytes(&blob_hash_owned).await { + Ok(bytes) => bytes, + Err(e) => { + tracing::warn!( + "Failed to read blob for thumbnail {} {:?}: {e}", + file_id_owned, + size + ); + return Bytes::new(); + } + }; + self.generate_and_persist(&file_id_owned, &thumb_path, size, original_data) + .await + }) + .await; + + let bytes = entry.into_value(); + if bytes.is_empty() { + return Err(ThumbnailError::ImageError( + "Thumbnail generation failed".to_string(), + )); + } + + tracing::debug!("πŸ”₯ Thumbnail served: {} {:?}", file_id, size); + Ok(bytes) + } + + /// Decode `original_data` into one thumbnail size, persist it to its + /// blob-keyed disk path, and return the encoded bytes β€” empty `Bytes` + /// on failure (moka's zero-weight negative-entry convention). + /// + /// Callers must hold a `decode_semaphore` permit. + async fn generate_and_persist( + &self, + file_id: &str, + thumb_path: &Path, + size: ThumbnailSize, + original_data: Bytes, + ) -> Bytes { + tracing::info!("🎨 Generating thumbnail: {} {:?}", file_id, size); + match Self::generate_thumbnail_from_data(original_data, size, self.generation_timeout).await + { + Ok(bytes) => { + if let Some(parent) = thumb_path.parent() { + let _ = fs::create_dir_all(parent).await; + } + let _ = fs::write(&thumb_path, &bytes).await; + bytes + } + Err(e) => { + tracing::warn!( + "Thumbnail generation failed for {} {:?}: {e}", + file_id, + size + ); + Bytes::new() + } + } + } + /// Try to serve a thumbnail from cache only (memory β†’ disk). /// /// Unlike `get_thumbnail`, this does **not** generate a new thumbnail. @@ -823,15 +908,16 @@ impl ThumbnailService { }); } - /// Generate all thumbnail sizes in the background from raw image bytes. + /// Generate all thumbnail sizes in the background for a content-addressed + /// blob (CDC/manifest-safe β€” no physical source file required). /// - /// This is compatible with CDC/manifest-backed blobs because it does not - /// require a single physical source file on disk. - pub fn generate_all_sizes_background_from_bytes( + /// The source blob is read **after** the decode permit is acquired, so N + /// concurrent uploads queue as N small tasks, not N full images in RAM: + /// peak memory is `permits Γ— image size` regardless of upload concurrency. + pub fn generate_all_sizes_background_from_blob( self: Arc, file_id: String, blob_hash: String, - original_data: Bytes, dedup: Arc, ) { tokio::spawn(async move { @@ -889,6 +975,20 @@ impl ThumbnailService { } }; + // Read the source only now that a permit bounds how many of + // these full-image buffers can exist at once. + let original_data = match dedup.read_blob_bytes(&blob_hash).await { + Ok(bytes) => bytes, + Err(e) => { + tracing::warn!( + "Failed to read blob for thumbnail generation {}: {}", + file_id, + e + ); + return; + } + }; + let results = tokio::task::spawn_blocking(move || { Self::render_all_thumbnails_from_data(original_data.as_ref()) }) @@ -1013,12 +1113,13 @@ impl crate::application::ports::file_lifecycle::FileLifecycleHook for ThumbnailR if !is_new_blob || !ThumbnailService::is_supported_image(content_type) { return; } - Self::spawn_thumbnail_generation( - self.thumbnail.clone(), - self.dedup.clone(), - file_id.to_string(), - blob_hash.to_string(), - ); + self.thumbnail + .clone() + .generate_all_sizes_background_from_blob( + file_id.to_string(), + blob_hash.to_string(), + self.dedup.clone(), + ); } fn on_file_copied( @@ -1047,7 +1148,7 @@ impl crate::application::ports::file_lifecycle::FileLifecycleHook for ThumbnailR e ); } - Self::spawn_thumbnail_generation(thumbnail, dedup, file_id, blob_hash); + thumbnail.generate_all_sizes_background_from_blob(file_id, blob_hash, dedup); }); } @@ -1066,30 +1167,6 @@ impl crate::application::ports::file_lifecycle::FileLifecycleHook for ThumbnailR // to avoid a circular Arc: DedupServiceβ†’BlobLifecycleServiceβ†’ThumbnailRefreshHookβ†’DedupService. // ThumbnailService does not hold DedupService so no cycle exists. -impl ThumbnailRefreshHook { - fn spawn_thumbnail_generation( - ts: Arc, - ds: Arc, - file_id: String, - hash: String, - ) { - tokio::spawn(async move { - match ds.read_blob_bytes(&hash).await { - Ok(bytes) => { - ts.generate_all_sizes_background_from_bytes(file_id, hash, bytes, ds.clone()); - } - Err(e) => { - tracing::warn!( - "Failed to read blob for thumbnail generation {}: {}", - file_id, - e - ); - } - } - }); - } -} - // ─── BlobLifecycleHook ─────────────────────────────────────────────────────── impl crate::application::ports::blob_lifecycle::BlobLifecycleHook for ThumbnailService { diff --git a/src/interfaces/api/handlers/file_handler.rs b/src/interfaces/api/handlers/file_handler.rs index 1162dd2b..d80cd494 100644 --- a/src/interfaces/api/handlers/file_handler.rs +++ b/src/interfaces/api/handlers/file_handler.rs @@ -442,19 +442,13 @@ impl FileHandler { .into_response(); } - let original_bytes = match state.core.dedup_service.read_blob_bytes(&blob_hash).await { - Ok(bytes) => bytes, - Err(err) => { - return AppError::internal_error(format!( - "Failed to load source image for thumbnail generation: {}", - err - )) - .into_response(); - } - }; - match thumbnail_service - .get_thumbnail_from_bytes(&id, &blob_hash, thumb_size.into(), original_bytes) + .get_thumbnail_from_blob( + &id, + &blob_hash, + thumb_size.into(), + state.core.dedup_service.clone(), + ) .await { Ok(data) => Response::builder() diff --git a/src/interfaces/nextcloud/preview_handler.rs b/src/interfaces/nextcloud/preview_handler.rs index 82028caf..9fb9ac30 100644 --- a/src/interfaces/nextcloud/preview_handler.rs +++ b/src/interfaces/nextcloud/preview_handler.rs @@ -157,26 +157,18 @@ pub async fn handle_preview( .unwrap(); } - let original_bytes = match state.core.dedup_service.read_blob_bytes(&blob_hash).await { - Ok(bytes) => bytes, - Err(err) => { - tracing::error!( - "Failed to load source image for preview {}: {}", - object_id, - err - ); - return Response::builder() - .status(StatusCode::INTERNAL_SERVER_ERROR) - .body(Body::from("Failed to load preview source")) - .unwrap(); - } - }; - - // Generate/get thumbnail + // Generate/get thumbnail β€” the blob is read inside the service once a + // decode permit is held, so preview stampedes cannot stack source + // images in RAM. match state .core .thumbnail_service - .get_thumbnail_from_bytes(&object_id, &blob_hash, thumb_size.into(), original_bytes) + .get_thumbnail_from_blob( + &object_id, + &blob_hash, + thumb_size.into(), + state.core.dedup_service.clone(), + ) .await { Ok(data) => {