From c3b853abd214985c043844be9db9560bc81900ef Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 10 Jun 2026 12:44:13 +0000 Subject: [PATCH 1/4] fix(lifecycle): pass blob content_hash to file lifecycle hooks, not etag MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The five FileLifecycleHook call sites handed dto.etag (format "{blob_hash[..16]}-{modified_at}") to the blob_hash parameter. Blob lookups keyed by that value can never resolve, so background thumbnail generation and audio metadata extraction silently failed on every upload/copy/update, pushing all thumbnail work onto the request path. Pass dto.content_hash — the raw full BLAKE3 hash already carried by FileDto — instead. https://claude.ai/code/session_01QxwJDHqQhbMkHK333QtMme --- src/application/services/file_management_service.rs | 2 +- src/application/services/file_upload_service.rs | 8 ++++---- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/src/application/services/file_management_service.rs b/src/application/services/file_management_service.rs index 453a3a9c..70e5b327 100644 --- a/src/application/services/file_management_service.rs +++ b/src/application/services/file_management_service.rs @@ -151,7 +151,7 @@ impl FileManagementService { let dto = FileDto::from(copied_file); if let Some(hook) = &self.file_lifecycle_hook { - hook.on_file_copied(&dto.id, &dto.etag, &dto.mime_type, file_id); + hook.on_file_copied(&dto.id, &dto.content_hash, &dto.mime_type, file_id); } Ok(dto) } diff --git a/src/application/services/file_upload_service.rs b/src/application/services/file_upload_service.rs index 9490f7e6..20e18898 100644 --- a/src/application/services/file_upload_service.rs +++ b/src/application/services/file_upload_service.rs @@ -177,7 +177,7 @@ impl FileUploadUseCase for FileUploadService { ); self.maybe_update_storage_usage(&dto); if let Some(hook) = &self.file_lifecycle_hook { - hook.on_file_created(&dto.id, &dto.etag, &dto.mime_type, is_new_blob); + hook.on_file_created(&dto.id, &dto.content_hash, &dto.mime_type, is_new_blob); } Ok(dto) } @@ -260,7 +260,7 @@ impl FileUploadUseCase for FileUploadService { let dto = FileDto::from(file); self.maybe_update_storage_usage(&dto); if let Some(hook) = &self.file_lifecycle_hook { - hook.on_file_created(&dto.id, &dto.etag, &dto.mime_type, is_new_blob); + hook.on_file_created(&dto.id, &dto.content_hash, &dto.mime_type, is_new_blob); } Ok(dto) } @@ -337,7 +337,7 @@ impl FileUploadUseCase for FileUploadService { let updated = file_read.get_file(&file_id).await?; let dto = FileDto::from(updated); if let Some(hook) = &self.file_lifecycle_hook { - hook.on_file_updated(&file_id, &dto.etag, content_type); + hook.on_file_updated(&file_id, &dto.content_hash, content_type); } return Ok(dto); } @@ -375,7 +375,7 @@ impl FileUploadUseCase for FileUploadService { .await?; let dto = FileDto::from(created); if let Some(hook) = &self.file_lifecycle_hook { - hook.on_file_created(&dto.id, &dto.etag, content_type, is_new_blob); + hook.on_file_created(&dto.id, &dto.content_hash, content_type, is_new_blob); } Ok(dto) } From 4200209d4a48124c2672136e6f0fc9a3ad451262 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 10 Jun 2026 13:22:27 +0000 Subject: [PATCH 2/4] perf(db): statement-level tree-ETag triggers; batched trash purge MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace the per-row tree_modified_at bump triggers with AFTER ... FOR EACH STATEMENT triggers using transition tables: each DML statement now pays one bump covering the distinct ancestor chains of all affected rows (locked in id order so concurrent bumps over overlapping chains cannot deadlock), instead of one chain UPDATE per affected row. Value-based change detection replaces UPDATE OF column lists (PG forbids those with transition tables), so EXIF media_sort_date syncs and no-op updates no longer bump at all, and file moves now invalidate the source chain as well as the destination. Also fix a latent bug surfaced while testing this: the descendant path/lpath cascade (trg_folders_cascade_path) was declared AFTER UPDATE OF path, lpath, but rename/move statements SET name/parent_id, and BEFORE-trigger rewrites do not count for UPDATE OF — the cascade never fired, leaving every descendant folder with a stale path/lpath after any rename or move. The migration canonically repairs existing trees and re-creates the cascade on the columns the app actually writes. delete_expired_bulk now deletes in LIMIT-ed batches (1000 files / 100 folders per round, each its own implicit transaction) ordered by trashed_at and served by new partial indexes over trashed rows, so retention purges no longer hold one unbounded transaction. Verified against a local PG16 cluster: all migrations apply from scratch, the new one re-runs idempotently, and a 16-case behavioral battery passes (chain bumps for insert/update/delete/move/trash, EXIF invisibility, no-op invisibility, cascade repair, FK-cascade depth guards, batched purge shape and partial-index plan). https://claude.ai/code/session_01QxwJDHqQhbMkHK333QtMme --- ...626000000_tree_etag_statement_triggers.sql | 318 ++++++++++++++++++ .../repositories/pg/trash_db_repository.rs | 92 +++-- 2 files changed, 379 insertions(+), 31 deletions(-) create mode 100644 migrations/20260626000000_tree_etag_statement_triggers.sql diff --git a/migrations/20260626000000_tree_etag_statement_triggers.sql b/migrations/20260626000000_tree_etag_statement_triggers.sql new file mode 100644 index 00000000..d1657315 --- /dev/null +++ b/migrations/20260626000000_tree_etag_statement_triggers.sql @@ -0,0 +1,318 @@ +-- Statement-level rewrite of the folder-tree ETag bump triggers. +-- +-- The per-row triggers from `20260625000002_folder_tree_modified_at` +-- executed one ancestor-chain UPDATE (`lpath @> target`) for EVERY +-- affected file/folder row: +-- * one upload → 1 lpath SELECT + O(depth) folder updates, +-- fired AGAIN by the EXIF media_sort_date +-- sync UPDATE (double bump per image); +-- * emptying a 50k trash → 50k trigger executions × chain UPDATEs +-- inside one transaction (WAL storm, long +-- lock hold); +-- * concurrent uploads under one root serialize on the root folder +-- row and lock overlapping chains in arbitrary per-row order +-- (deadlock-prone). +-- +-- This migration replaces them with AFTER … FOR EACH STATEMENT triggers +-- using transition tables (PG ≥ 10): each DML statement pays exactly ONE +-- bump covering the distinct ancestor chains of all affected rows, and +-- ancestor rows are locked in deterministic id order. +-- +-- Deliberate semantic deltas vs the per-row version: +-- * File moves now bump the OLD parent chain too. The per-row trigger +-- used COALESCE(NEW.folder_id, OLD.folder_id), so a move-out never +-- changed the source folder's ETag and sync clients watching the +-- source never saw the file disappear without a deep re-walk. +-- * UPDATEs that only touch storage.files.media_sort_date (the EXIF +-- denormalisation sync) no longer bump: the column is not visible +-- to DAV clients. This removes the double bump per image upload. +-- * Bumps fired from inside another trigger's DML (pg_trigger_depth +-- > 1: FK cascades of user/folder deletion, the lpath cascade +-- rewrite) are skipped on the FILE side as well — the folder-side +-- trigger of the outermost statement already covers the surviving +-- ancestors. The per-row file trigger had no such guard and burned +-- bumps on rows that were themselves being deleted. +-- * Ancestor rows already stamped with this transaction's NOW() are +-- skipped — pure WAL saving, the stored value would be identical. + +-- ── File side: INSERT / DELETE ─────────────────────────────────────── +-- Both triggers alias their transition table to `changed_rows`, so one +-- function body serves both events (PL/pgSQL resolves the name against +-- the tuplestore registered by whichever trigger fired). +CREATE OR REPLACE FUNCTION storage.bump_tree_from_files_stmt() +RETURNS TRIGGER LANGUAGE plpgsql AS $$ +BEGIN + IF pg_trigger_depth() > 1 THEN + RETURN NULL; + END IF; + + WITH targets AS ( + -- Distinct parent folders of all rows in this statement. + -- Root-level files (folder_id IS NULL) have no ancestors; a + -- vanished parent row simply drops out of the JOIN. + SELECT DISTINCT fo.lpath + FROM (SELECT DISTINCT folder_id + FROM changed_rows + WHERE folder_id IS NOT NULL) c + JOIN storage.folders fo ON fo.id = c.folder_id + ), + victims AS ( + -- `lpath @> target` = the parent folder itself plus every + -- ancestor up to the root (GiST-indexed). Lock in id order so + -- concurrent bumps over overlapping chains cannot deadlock. + SELECT f.id + FROM storage.folders f + WHERE EXISTS (SELECT 1 FROM targets t WHERE f.lpath @> t.lpath) + AND f.tree_modified_at IS DISTINCT FROM NOW() + ORDER BY f.id + FOR UPDATE + ) + UPDATE storage.folders f + SET tree_modified_at = NOW() + FROM victims v + WHERE f.id = v.id; + + RETURN NULL; +END; +$$; + +-- ── File side: UPDATE ──────────────────────────────────────────────── +-- Bumps the union of OLD and NEW parent chains so a move invalidates +-- both the source and the destination collection ETags. +-- +-- PostgreSQL forbids `AFTER UPDATE OF ` together with transition +-- tables ("transition tables cannot be specified for triggers with +-- column lists"), so the DAV-visibility filter lives inside the +-- function instead: a row only counts when one of the observable +-- columns actually changed value. This is strictly better than a +-- column list — the EXIF media_sort_date sync and no-op UPDATEs +-- (same value re-written) no longer bump anything. +CREATE OR REPLACE FUNCTION storage.bump_tree_from_files_stmt_upd() +RETURNS TRIGGER LANGUAGE plpgsql AS $$ +BEGIN + IF pg_trigger_depth() > 1 THEN + RETURN NULL; + END IF; + + WITH changed AS ( + SELECT o.folder_id AS old_folder_id, n.folder_id AS new_folder_id + FROM old_rows o + JOIN new_rows n USING (id) + WHERE (o.name, o.folder_id, o.blob_hash, o.size, + o.mime_type, o.is_trashed, o.updated_at) + IS DISTINCT FROM + (n.name, n.folder_id, n.blob_hash, n.size, + n.mime_type, n.is_trashed, n.updated_at) + ), + targets AS ( + SELECT DISTINCT fo.lpath + FROM (SELECT old_folder_id AS folder_id + FROM changed WHERE old_folder_id IS NOT NULL + UNION + SELECT new_folder_id + FROM changed WHERE new_folder_id IS NOT NULL) c + JOIN storage.folders fo ON fo.id = c.folder_id + ), + victims AS ( + SELECT f.id + FROM storage.folders f + WHERE EXISTS (SELECT 1 FROM targets t WHERE f.lpath @> t.lpath) + AND f.tree_modified_at IS DISTINCT FROM NOW() + ORDER BY f.id + FOR UPDATE + ) + UPDATE storage.folders f + SET tree_modified_at = NOW() + FROM victims v + WHERE f.id = v.id; + + RETURN NULL; +END; +$$; + +-- ── Folder side: INSERT / DELETE ───────────────────────────────────── +-- Strict ancestors only (`lpath <> target` preserves the per-row +-- version's self-exclusion: a folder's own create/delete/rename does +-- not bump its own tree_modified_at, only its ancestors'). +CREATE OR REPLACE FUNCTION storage.bump_tree_from_folders_stmt() +RETURNS TRIGGER LANGUAGE plpgsql AS $$ +BEGIN + IF pg_trigger_depth() > 1 THEN + RETURN NULL; + END IF; + + WITH targets AS ( + SELECT DISTINCT lpath FROM changed_rows WHERE lpath IS NOT NULL + ), + victims AS ( + SELECT f.id + FROM storage.folders f + WHERE EXISTS (SELECT 1 FROM targets t + WHERE f.lpath @> t.lpath AND f.lpath <> t.lpath) + AND f.tree_modified_at IS DISTINCT FROM NOW() + ORDER BY f.id + FOR UPDATE + ) + UPDATE storage.folders f + SET tree_modified_at = NOW() + FROM victims v + WHERE f.id = v.id; + + RETURN NULL; +END; +$$; + +-- ── Folder side: UPDATE ────────────────────────────────────────────── +-- Union of OLD and NEW lpaths: a move bumps the chain it left and the +-- chain it joined. Same value-based filter as the file side (column +-- lists are incompatible with transition tables); the descendant +-- lpath rewrites done by `trg_folders_cascade_path` run at trigger +-- depth 2 and are stopped by the depth guard, and they change none of +-- the compared columns anyway — exactly one bump per move statement. +CREATE OR REPLACE FUNCTION storage.bump_tree_from_folders_stmt_upd() +RETURNS TRIGGER LANGUAGE plpgsql AS $$ +BEGIN + IF pg_trigger_depth() > 1 THEN + RETURN NULL; + END IF; + + WITH changed AS ( + SELECT o.lpath AS old_lpath, n.lpath AS new_lpath + FROM old_rows o + JOIN new_rows n USING (id) + WHERE (o.name, o.parent_id, o.is_trashed, o.updated_at) + IS DISTINCT FROM + (n.name, n.parent_id, n.is_trashed, n.updated_at) + ), + targets AS ( + SELECT DISTINCT lpath + FROM (SELECT old_lpath AS lpath + FROM changed WHERE old_lpath IS NOT NULL + UNION + SELECT new_lpath + FROM changed WHERE new_lpath IS NOT NULL) c + ), + victims AS ( + SELECT f.id + FROM storage.folders f + WHERE EXISTS (SELECT 1 FROM targets t + WHERE f.lpath @> t.lpath AND f.lpath <> t.lpath) + AND f.tree_modified_at IS DISTINCT FROM NOW() + ORDER BY f.id + FOR UPDATE + ) + UPDATE storage.folders f + SET tree_modified_at = NOW() + FROM victims v + WHERE f.id = v.id; + + RETURN NULL; +END; +$$; + +-- ── Swap the triggers ──────────────────────────────────────────────── +-- PG 13 compatibility: DROP-then-CREATE (no CREATE OR REPLACE TRIGGER). +DROP TRIGGER IF EXISTS files_bump_folder_tree_etag ON storage.files; +DROP TRIGGER IF EXISTS folders_bump_folder_tree_etag ON storage.folders; + +DROP TRIGGER IF EXISTS files_bump_tree_etag_ins ON storage.files; +CREATE TRIGGER files_bump_tree_etag_ins + AFTER INSERT ON storage.files + REFERENCING NEW TABLE AS changed_rows + FOR EACH STATEMENT EXECUTE FUNCTION storage.bump_tree_from_files_stmt(); + +DROP TRIGGER IF EXISTS files_bump_tree_etag_del ON storage.files; +CREATE TRIGGER files_bump_tree_etag_del + AFTER DELETE ON storage.files + REFERENCING OLD TABLE AS changed_rows + FOR EACH STATEMENT EXECUTE FUNCTION storage.bump_tree_from_files_stmt(); + +-- No column list (incompatible with transition tables): the function +-- compares the DAV-observable columns (rename / move / content swap / +-- trash / restore / mtime touch) per row and ignores everything else, +-- including the EXIF media_sort_date sync. +DROP TRIGGER IF EXISTS files_bump_tree_etag_upd ON storage.files; +CREATE TRIGGER files_bump_tree_etag_upd + AFTER UPDATE ON storage.files + REFERENCING OLD TABLE AS old_rows NEW TABLE AS new_rows + FOR EACH STATEMENT EXECUTE FUNCTION storage.bump_tree_from_files_stmt_upd(); + +DROP TRIGGER IF EXISTS folders_bump_tree_etag_ins ON storage.folders; +CREATE TRIGGER folders_bump_tree_etag_ins + AFTER INSERT ON storage.folders + REFERENCING NEW TABLE AS changed_rows + FOR EACH STATEMENT EXECUTE FUNCTION storage.bump_tree_from_folders_stmt(); + +DROP TRIGGER IF EXISTS folders_bump_tree_etag_del ON storage.folders; +CREATE TRIGGER folders_bump_tree_etag_del + AFTER DELETE ON storage.folders + REFERENCING OLD TABLE AS changed_rows + FOR EACH STATEMENT EXECUTE FUNCTION storage.bump_tree_from_folders_stmt(); + +-- The function's value filter ignores path/lpath rewrites (done only +-- by the depth-guarded `trg_folders_cascade_path` cascade) and this +-- trigger's own tree_modified_at writes — neither can re-fire a bump. +DROP TRIGGER IF EXISTS folders_bump_tree_etag_upd ON storage.folders; +CREATE TRIGGER folders_bump_tree_etag_upd + AFTER UPDATE ON storage.folders + REFERENCING OLD TABLE AS old_rows NEW TABLE AS new_rows + FOR EACH STATEMENT EXECUTE FUNCTION storage.bump_tree_from_folders_stmt_upd(); + +-- Old per-row functions are no longer referenced by any trigger. +DROP FUNCTION IF EXISTS storage.bump_folder_tree_from_file(); +DROP FUNCTION IF EXISTS storage.bump_folder_tree_from_folder(); + +-- ── Fix: descendant path/lpath cascade never fired ─────────────────── +-- `trg_folders_cascade_path` was declared `AFTER UPDATE OF path, lpath`, +-- but `UPDATE OF` only matches columns named in the statement's SET +-- clause — and the app's rename/move statements SET `name`/`parent_id` +-- (path/lpath are rewritten by the BEFORE trigger, which does not +-- count). Net effect on deployments to date: renaming or moving a +-- folder silently left every DESCENDANT folder with a stale path and +-- lpath, corrupting subtree queries (deletes, search, ACL cascade, +-- tree-ETag chains) under the old location. +-- +-- Repair order matters: drop the broken trigger, canonically rebuild +-- path/lpath for every folder from its parent chain (only stale rows +-- are written; the statement-level bump trigger above sees no +-- DAV-visible column change, so folder ETags are untouched), then +-- re-create the cascade with the column list the app actually hits. +DROP TRIGGER IF EXISTS trg_folders_cascade_path ON storage.folders; + +WITH RECURSIVE canon AS ( + SELECT id, + name::text AS path, + replace(id::text, '-', '_')::ltree AS lpath + FROM storage.folders + WHERE parent_id IS NULL + UNION ALL + SELECT f.id, + c.path || '/' || f.name, + c.lpath || replace(f.id::text, '-', '_')::ltree + FROM storage.folders f + JOIN canon c ON f.parent_id = c.id +) +UPDATE storage.folders f + SET path = c.path, lpath = c.lpath + FROM canon c + WHERE f.id = c.id + AND (f.path IS DISTINCT FROM c.path OR f.lpath IS DISTINCT FROM c.lpath); + +-- name/parent_id: what rename/move statements actually SET (the BEFORE +-- trigger has already recomputed this row's path/lpath by the time the +-- AFTER trigger compares OLD vs NEW). path/lpath stay listed for direct +-- writes. The cascade function's own pg_trigger_depth() guard still +-- stops its batch descendant rewrite from re-firing itself. +CREATE TRIGGER trg_folders_cascade_path + AFTER UPDATE OF name, parent_id, path, lpath ON storage.folders + FOR EACH ROW EXECUTE FUNCTION storage.cascade_folder_path(); + +-- ── Retention-purge support indexes ────────────────────────────────── +-- `delete_expired_bulk` now deletes in LIMIT-ed batches ordered by +-- trashed_at. These partial indexes (trashed rows only — tiny) turn +-- each batch's candidate scan into an index range scan instead of a +-- repeated sequential scan over the whole table. +CREATE INDEX IF NOT EXISTS idx_files_trash_expiry + ON storage.files (trashed_at) WHERE is_trashed; +CREATE INDEX IF NOT EXISTS idx_folders_trash_expiry + ON storage.folders (trashed_at) WHERE is_trashed; diff --git a/src/infrastructure/repositories/pg/trash_db_repository.rs b/src/infrastructure/repositories/pg/trash_db_repository.rs index ab9d3741..3f0eadc1 100644 --- a/src/infrastructure/repositories/pg/trash_db_repository.rs +++ b/src/infrastructure/repositories/pg/trash_db_repository.rs @@ -47,6 +47,40 @@ impl TrashDbRepository { } } + /// Runs a LIMIT-ed DELETE statement repeatedly until a round affects + /// fewer rows than `batch_size`, yielding to the runtime between rounds. + /// + /// `sql` must bind `$1` = cutoff timestamp and `$2` = batch size; the + /// candidate sub-select is served by the `idx_*_trash_expiry` partial + /// indexes. Each round is its own implicit transaction, so row locks, + /// WAL volume and the statement-trigger transition tables stay bounded + /// no matter how many items expired. Partial progress is fine — the + /// next retention sweep continues where this one stopped. + async fn delete_expired_batch_loop( + &self, + sql: &'static str, + cutoff: DateTime, + batch_size: i64, + ) -> Result { + let mut total: u64 = 0; + loop { + let affected = sqlx::query(sql) + .bind(cutoff) + .bind(batch_size) + .execute(self.pool.as_ref()) + .await + .map_err(|e| { + DomainError::internal_error("TrashDb", format!("bulk delete batch: {e}")) + })? + .rows_affected(); + total += affected; + if affected < batch_size as u64 { + return Ok(total); + } + tokio::task::yield_now().await; + } + } + /// Convert a trash_items view row into a TrashedItem entity. fn row_to_trashed_item( &self, @@ -180,40 +214,36 @@ impl TrashRepository for TrashDbRepository { async fn delete_expired_bulk(&self) -> Result<(u64, u64)> { let cutoff = Utc::now() - chrono::Duration::days(self.retention_days); - let mut tx = self - .pool - .begin() - .await - .map_err(|e| DomainError::internal_error("TrashDb", format!("begin tx: {e}")))?; - - // 1. Bulk-delete expired trashed files. + // 1. Bulk-delete expired trashed files in batches. // The PG trigger `trg_files_decrement_blob_ref` automatically // decrements blob ref_count for every deleted row. - let files_deleted = - sqlx::query("DELETE FROM storage.files WHERE is_trashed = TRUE AND trashed_at < $1") - .bind(cutoff) - .execute(&mut *tx) - .await - .map_err(|e| { - DomainError::internal_error("TrashDb", format!("bulk delete files: {e}")) - })? - .rows_affected(); + let files_deleted = self + .delete_expired_batch_loop( + "DELETE FROM storage.files + WHERE id IN (SELECT id FROM storage.files + WHERE is_trashed = TRUE AND trashed_at < $1 + ORDER BY trashed_at + LIMIT $2)", + cutoff, + 1_000, + ) + .await?; - // 2. Bulk-delete expired trashed folders. - // FK ON DELETE CASCADE handles descendant folders and their files. - let folders_deleted = - sqlx::query("DELETE FROM storage.folders WHERE is_trashed = TRUE AND trashed_at < $1") - .bind(cutoff) - .execute(&mut *tx) - .await - .map_err(|e| { - DomainError::internal_error("TrashDb", format!("bulk delete folders: {e}")) - })? - .rows_affected(); - - tx.commit() - .await - .map_err(|e| DomainError::internal_error("TrashDb", format!("commit tx: {e}")))?; + // 2. Bulk-delete expired trashed folders in batches. + // FK ON DELETE CASCADE handles descendant folders and their + // files, so each row can fan out to an entire subtree — hence + // the smaller batch size. + let folders_deleted = self + .delete_expired_batch_loop( + "DELETE FROM storage.folders + WHERE id IN (SELECT id FROM storage.folders + WHERE is_trashed = TRUE AND trashed_at < $1 + ORDER BY trashed_at + LIMIT $2)", + cutoff, + 100, + ) + .await?; Ok((files_deleted, folders_deleted)) } From b0f83cfa343b158fb7378af19e0bc5b117c99f3b Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 10 Jun 2026 13:36:29 +0000 Subject: [PATCH 3/4] perf(thumbnails): hold a decode permit before reading source blobs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Request-path thumbnail generation (REST get_thumbnail and NC preview) read the full source blob into RAM and decoded it with no concurrency bound — a first-view gallery of K images stacked K full-size buffers and K parallel decodes on the blocking pool. The background hook had the inverse ordering problem: it read the blob eagerly and only then queued on the decode semaphore, so N concurrent uploads held N originals in memory while waiting. All three paths now acquire the decode semaphore first and read the blob under the permit, capping peak RAM at permits x image size: - new ThumbnailService::get_thumbnail_from_blob defers the blob read into the moka init closure (cache and disk hits never touch the blob); both handlers use it and no longer pre-read. - get_thumbnail_from_bytes (direct-bytes variant) now also takes a permit before decoding; shared generate_and_persist core keeps the two entrypoints duplicate-free. - generate_all_sizes_background_from_blob (renamed from _from_bytes) checks blob existence and disk-dedup state, then acquires the permit, then reads; the redundant pre-read spawn wrapper in ThumbnailRefreshHook is gone. https://claude.ai/code/session_01QxwJDHqQhbMkHK333QtMme --- .../services/thumbnail_service.rs | 199 ++++++++++++------ src/interfaces/api/handlers/file_handler.rs | 18 +- src/interfaces/nextcloud/preview_handler.rs | 26 +-- 3 files changed, 153 insertions(+), 90 deletions(-) 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) => { From f4ce4092f0fe7917889c89ce187e19f1bd4052df Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 10 Jun 2026 14:04:36 +0000 Subject: [PATCH 4/4] fix(blobs): invalidate the file_id->blob_hash cache on content swaps MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The read repository's hash_cache assumed blob hashes were immutable per file_id, but update_file_content_from_temp remaps the SAME row to a new hash via swap_blob_hash. After a WebDAV/NC PUT overwrite, streaming downloads (>=10 MB tier, Range requests, video playback) kept resolving the OLD blob for the 30 s TTI window — and because every read refreshes the TTI, a polling client could pin the stale mapping indefinitely, eventually turning into 500s once the old blob's refcount hit zero and GC removed it. The write repository now holds a shared handle to the same moka cache (clones share storage) and invalidates the entry right after every content swap and hard delete commits — covering every present and future caller of the write port rather than one service path. Stale doc comments claiming the mapping was immutable (and SHA-256) fixed. https://claude.ai/code/session_01QxwJDHqQhbMkHK333QtMme --- src/common/di.rs | 3 +++ .../pg/file_blob_read_repository.rs | 24 +++++++++++++++---- .../pg/file_blob_write_repository.rs | 20 ++++++++++++++-- 3 files changed, 40 insertions(+), 7 deletions(-) diff --git a/src/common/di.rs b/src/common/di.rs index 8d805565..5a07193f 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -380,6 +380,9 @@ impl AppServiceFactory { db_pool.clone(), core.dedup_service.clone(), folder_repo_concrete.clone(), + // Shared blob-hash cache: the write side invalidates entries + // on content swaps/deletes so reads never serve stale blobs. + file_read_repository.blob_hash_cache(), )); // I18n repository — file-system backed, gated by the locale diff --git a/src/infrastructure/repositories/pg/file_blob_read_repository.rs b/src/infrastructure/repositories/pg/file_blob_read_repository.rs index 3077001a..35ffaf45 100644 --- a/src/infrastructure/repositories/pg/file_blob_read_repository.rs +++ b/src/infrastructure/repositories/pg/file_blob_read_repository.rs @@ -59,8 +59,14 @@ pub struct FileBlobReadRepository { dedup: Arc, /// Lock-free cache: file_id → blob_hash. /// Populated by `get_file()` and `resolve_blob_hash()` (slow path). - /// Entries persist until TTI expiry (30 s idle) or capacity eviction — - /// safe because blob_hash is content-addressed and never mutated. + /// Entries persist until TTI expiry (30 s idle) or capacity eviction. + /// Content updates DO remap a file_id to a new hash in place + /// (`swap_blob_hash`), so the write repository shares this cache (see + /// [`Self::blob_hash_cache`]) and invalidates the entry on every + /// content swap and hard delete — without that, streaming downloads + /// kept serving the previous blob for the TTI window after a PUT + /// update (or 500'd once the old blob was garbage-collected), and + /// every read refreshed the TTI, extending the window indefinitely. hash_cache: Cache, } @@ -80,6 +86,14 @@ impl FileBlobReadRepository { } } + /// Shared handle to the file_id → blob_hash cache (moka clones share + /// the underlying storage). Handed to `FileBlobWriteRepository` at DI + /// time so content swaps and hard deletes invalidate the mapping the + /// moment they commit. + pub fn blob_hash_cache(&self) -> Cache { + self.hash_cache.clone() + } + /// Returns the user_id (owner) for a given file ID. /// Mirrors `FolderDbRepository::get_folder_user_id`. /// Used by the AuthorizationEngine for owner short-circuit. @@ -157,9 +171,9 @@ impl FileBlobReadRepository { /// subsequent reads for the same file (e.g. Range Requests on a video, /// thumbnail + download, browser re-fetch) hit the cache instead of PG. /// - /// This is safe because `blob_hash` is content-addressed (SHA-256) - /// and never mutated — if the file's content changes, a new row with a - /// new `blob_hash` is created. + /// Staleness safety: content updates remap the file to a new hash in + /// place — the write repository invalidates this cache (shared via + /// [`Self::blob_hash_cache`]) right after every swap/delete commits. async fn resolve_blob_hash(&self, file_id: &str) -> Result { // Fast path: cached (lock-free read, refreshes TTI automatically) if let Some(hash) = self.hash_cache.get(file_id) { diff --git a/src/infrastructure/repositories/pg/file_blob_write_repository.rs b/src/infrastructure/repositories/pg/file_blob_write_repository.rs index 7819f939..d05f62c4 100644 --- a/src/infrastructure/repositories/pg/file_blob_write_repository.rs +++ b/src/infrastructure/repositories/pg/file_blob_write_repository.rs @@ -7,6 +7,7 @@ //! File paths are resolved by querying the materialized `storage.folders.path` //! column (O(1) per lookup), so no recursive CTEs are needed. +use moka::sync::Cache; use sqlx::PgPool; use std::path::PathBuf; use std::sync::Arc; @@ -26,6 +27,10 @@ pub struct FileBlobWriteRepository { pool: Arc, dedup: Arc, folder_repo: Arc, + /// Shared handle to `FileBlobReadRepository`'s file_id → blob_hash + /// cache. Content swaps and hard deletes invalidate the mapping here + /// so the read side can never serve a stale blob after a PUT update. + hash_cache: Cache, } impl FileBlobWriteRepository { @@ -33,11 +38,13 @@ impl FileBlobWriteRepository { pool: Arc, dedup: Arc, folder_repo: Arc, + hash_cache: Cache, ) -> Self { Self { pool, dedup, folder_repo, + hash_cache, } } @@ -54,6 +61,7 @@ impl FileBlobWriteRepository { ), dedup: Arc::new(DedupService::new_stub()), folder_repo: Arc::new(super::folder_db_repository::FolderDbRepository::new_stub()), + hash_cache: Cache::builder().max_capacity(10_000).build(), } } @@ -502,6 +510,8 @@ impl FileWritePort for FileBlobWriteRepository { return Err(DomainError::not_found("File", id)); } + // Drop the read-side file_id → blob_hash mapping for the dead row. + self.hash_cache.invalidate(id); Ok(()) } @@ -521,8 +531,14 @@ impl FileWritePort for FileBlobWriteRepository { .await?; let new_hash = dedup_result.hash().to_string(); - self.swap_blob_hash(file_id, &new_hash, size as i64, modified_at) - .await + let swapped = self + .swap_blob_hash(file_id, &new_hash, size as i64, modified_at) + .await?; + // The file now maps to a different blob — drop the read-side cache + // entry so streaming downloads cannot serve the previous content + // for the rest of its TTI window. + self.hash_cache.invalidate(file_id); + Ok(swapped) } async fn register_file_deferred(