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/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 30b064dd..92ec9215 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) } @@ -356,7 +356,7 @@ impl FileUploadUseCase for FileUploadService { })?; 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); } @@ -394,7 +394,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) } 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 4bc7ba04..224b89bc 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(), } } @@ -505,6 +513,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(()) } @@ -524,8 +534,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( 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)) } 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) => {