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(