diff --git a/src/infrastructure/services/dedup_service.rs b/src/infrastructure/services/dedup_service.rs index 38cf180f..ae013741 100644 --- a/src/infrastructure/services/dedup_service.rs +++ b/src/infrastructure/services/dedup_service.rs @@ -557,6 +557,74 @@ impl DedupService { self } + /// Store a server-derived artifact and record the mapping from the + /// content it was derived from. + /// + /// One call does the whole contract, so no caller has to remember the + /// accounting: + /// + /// 1. writes the bytes through the normal CDC path — derived blobs get + /// the same backend, encryption, migration and rotation as any other + /// content, and `store_from_stream` takes exactly one reference; + /// 2. records `(source_hash, kind, variant) -> blob_hash`; + /// 3. **releases that reference if the mapping already existed**, because + /// the row that would justify it is not ours — two instances racing + /// to render the same thumbnail must leave `ref_count` at 1, not 2. + /// + /// `bytes` is expected to be small (a thumbnail is 3-90 KB, below + /// `CDC_MIN_CHUNK`, so this is a single chunk). See + /// `docs/plan/derived-blobs.md`. + /// + /// Returns the derived blob hash. + pub async fn store_derived_blob( + &self, + source_hash: &str, + kind: &str, + variant: &str, + content_type: &str, + bytes: Bytes, + ) -> Result { + let stored = self + .store_from_stream( + stream::once(async move { Ok::(bytes) }), + Some(content_type.to_string()), + ) + .await?; + let derived_hash = stored.hash().to_string(); + + let inserted = sqlx::query( + "INSERT INTO storage.content_derived_blobs + (source_hash, kind, variant, blob_hash, content_type) + VALUES ($1, $2, $3, $4, $5) + ON CONFLICT (source_hash, kind, variant) DO NOTHING", + ) + .bind(source_hash) + .bind(kind) + .bind(variant) + .bind(&derived_hash) + .bind(content_type) + .execute(self.pool.as_ref()) + .await + .map_err(|e| DomainError::internal_error("Dedup", format!("record derived blob: {e}")))? + .rows_affected(); + + if inserted == 0 { + // Someone else already mapped this variant. Our reference has no + // row behind it; leaving it would inflate ref_count on every + // re-render and pin the blob forever. + if let Err(e) = self.remove_reference(&derived_hash).await { + tracing::warn!( + target: "oxicloud::dedup", + error = %e, + "failed to release duplicate derived-blob reference for {}", + &derived_hash[..derived_hash.len().min(12)], + ); + } + } + + Ok(derived_hash) + } + /// The registry backing the reap predicate. /// /// Exposed so `blobs_consistency` recomputes refcounts from the *same* diff --git a/src/infrastructure/services/thumbnail_service.rs b/src/infrastructure/services/thumbnail_service.rs index 09b36a32..1a6c0ced 100644 --- a/src/infrastructure/services/thumbnail_service.rs +++ b/src/infrastructure/services/thumbnail_service.rs @@ -1115,7 +1115,7 @@ impl ThumbnailService { } }; - self.render_and_persist_all_webp(&file_id, &blob_hash, original_data) + self.render_and_persist_all_webp(&file_id, &blob_hash, original_data, Some(&dedup)) .await; tracing::info!("✅ Background thumbnail generation complete: {}", file_id); @@ -1126,7 +1126,21 @@ impl ThumbnailService { /// blob_hash (disk `{hash}.webp` + moka). Shared by the image upload path and /// the video path (which passes the extracted frame as the source), so both /// produce identical, dedup-able, content-negotiable thumbnails. - async fn render_and_persist_all_webp(&self, file_id: &str, blob_hash: &str, source: Bytes) { + /// `dedup` is `Some` on every path that has a handle, which is every + /// eager background path. When present each rendered size is ALSO stored + /// as a derived blob and recorded in `storage.content_derived_blobs`. + /// + /// The sidecar write is deliberately kept: this slice fills the table + /// while reads still come from disk, so a rollback at any point leaves + /// working thumbnails and the table can be inspected against real data + /// before anything depends on it. See `docs/plan/derived-blobs.md`. + async fn render_and_persist_all_webp( + &self, + file_id: &str, + blob_hash: &str, + source: Bytes, + dedup: Option<&DedupService>, + ) { let results = tokio::task::spawn_blocking(move || { Self::render_all_thumbnails_from_data(source.as_ref(), ThumbnailFormat::Webp) }) @@ -1151,15 +1165,39 @@ impl ThumbnailService { } if let Err(e) = fs::write(&thumb_path, &bytes).await { tracing::warn!("Failed to save thumbnail {} {:?}: {}", file_id, size, e); - } else { - let cache_key = ThumbnailCacheKey { - file_id: file_id.to_string(), - size, - format: ThumbnailFormat::Webp, - }; - self.cache.insert(cache_key, bytes).await; - tracing::debug!("✅ Generated thumbnail: {} {:?}", file_id, size); + continue; } + + // Tier-3 copy. Best-effort and logged: a failure here must not + // cost the user their thumbnail, which is already on disk and in + // the cache. `derived_import` sweeps anything missed. + if let Some(dedup) = dedup + && let Err(e) = dedup + .store_derived_blob( + blob_hash, + "thumbnail", + size.dir_name(), + "image/webp", + bytes.clone(), + ) + .await + { + tracing::warn!( + target: "oxicloud::dedup", + error = %e, + "failed to record derived blob for {} {:?}", + file_id, + size, + ); + } + + let cache_key = ThumbnailCacheKey { + file_id: file_id.to_string(), + size, + format: ThumbnailFormat::Webp, + }; + self.cache.insert(cache_key, bytes).await; + tracing::debug!("✅ Generated thumbnail: {} {:?}", file_id, size); } } @@ -1251,7 +1289,7 @@ impl ThumbnailService { Ok(p) => p, Err(_) => return, }; - self.render_and_persist_all_webp(&file_id, &blob_hash, frame) + self.render_and_persist_all_webp(&file_id, &blob_hash, frame, Some(&dedup)) .await; tracing::info!("✅ Video thumbnail generation complete: {}", file_id); });