fix(blobs): invalidate the file_id->blob_hash cache on content swaps
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
This commit is contained in:
@@ -380,6 +380,9 @@ impl AppServiceFactory {
|
|||||||
db_pool.clone(),
|
db_pool.clone(),
|
||||||
core.dedup_service.clone(),
|
core.dedup_service.clone(),
|
||||||
folder_repo_concrete.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
|
// I18n repository — file-system backed, gated by the locale
|
||||||
|
|||||||
@@ -59,8 +59,14 @@ pub struct FileBlobReadRepository {
|
|||||||
dedup: Arc<DedupService>,
|
dedup: Arc<DedupService>,
|
||||||
/// Lock-free cache: file_id → blob_hash.
|
/// Lock-free cache: file_id → blob_hash.
|
||||||
/// Populated by `get_file()` and `resolve_blob_hash()` (slow path).
|
/// Populated by `get_file()` and `resolve_blob_hash()` (slow path).
|
||||||
/// Entries persist until TTI expiry (30 s idle) or capacity eviction —
|
/// Entries persist until TTI expiry (30 s idle) or capacity eviction.
|
||||||
/// safe because blob_hash is content-addressed and never mutated.
|
/// 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<String, String>,
|
hash_cache: Cache<String, String>,
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -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<String, String> {
|
||||||
|
self.hash_cache.clone()
|
||||||
|
}
|
||||||
|
|
||||||
/// Returns the user_id (owner) for a given file ID.
|
/// Returns the user_id (owner) for a given file ID.
|
||||||
/// Mirrors `FolderDbRepository::get_folder_user_id`.
|
/// Mirrors `FolderDbRepository::get_folder_user_id`.
|
||||||
/// Used by the AuthorizationEngine for owner short-circuit.
|
/// 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,
|
/// subsequent reads for the same file (e.g. Range Requests on a video,
|
||||||
/// thumbnail + download, browser re-fetch) hit the cache instead of PG.
|
/// thumbnail + download, browser re-fetch) hit the cache instead of PG.
|
||||||
///
|
///
|
||||||
/// This is safe because `blob_hash` is content-addressed (SHA-256)
|
/// Staleness safety: content updates remap the file to a new hash in
|
||||||
/// and never mutated — if the file's content changes, a new row with a
|
/// place — the write repository invalidates this cache (shared via
|
||||||
/// new `blob_hash` is created.
|
/// [`Self::blob_hash_cache`]) right after every swap/delete commits.
|
||||||
async fn resolve_blob_hash(&self, file_id: &str) -> Result<String, DomainError> {
|
async fn resolve_blob_hash(&self, file_id: &str) -> Result<String, DomainError> {
|
||||||
// Fast path: cached (lock-free read, refreshes TTI automatically)
|
// Fast path: cached (lock-free read, refreshes TTI automatically)
|
||||||
if let Some(hash) = self.hash_cache.get(file_id) {
|
if let Some(hash) = self.hash_cache.get(file_id) {
|
||||||
|
|||||||
@@ -7,6 +7,7 @@
|
|||||||
//! File paths are resolved by querying the materialized `storage.folders.path`
|
//! File paths are resolved by querying the materialized `storage.folders.path`
|
||||||
//! column (O(1) per lookup), so no recursive CTEs are needed.
|
//! column (O(1) per lookup), so no recursive CTEs are needed.
|
||||||
|
|
||||||
|
use moka::sync::Cache;
|
||||||
use sqlx::PgPool;
|
use sqlx::PgPool;
|
||||||
use std::path::PathBuf;
|
use std::path::PathBuf;
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
@@ -26,6 +27,10 @@ pub struct FileBlobWriteRepository {
|
|||||||
pool: Arc<PgPool>,
|
pool: Arc<PgPool>,
|
||||||
dedup: Arc<DedupService>,
|
dedup: Arc<DedupService>,
|
||||||
folder_repo: Arc<FolderDbRepository>,
|
folder_repo: Arc<FolderDbRepository>,
|
||||||
|
/// 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<String, String>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl FileBlobWriteRepository {
|
impl FileBlobWriteRepository {
|
||||||
@@ -33,11 +38,13 @@ impl FileBlobWriteRepository {
|
|||||||
pool: Arc<PgPool>,
|
pool: Arc<PgPool>,
|
||||||
dedup: Arc<DedupService>,
|
dedup: Arc<DedupService>,
|
||||||
folder_repo: Arc<FolderDbRepository>,
|
folder_repo: Arc<FolderDbRepository>,
|
||||||
|
hash_cache: Cache<String, String>,
|
||||||
) -> Self {
|
) -> Self {
|
||||||
Self {
|
Self {
|
||||||
pool,
|
pool,
|
||||||
dedup,
|
dedup,
|
||||||
folder_repo,
|
folder_repo,
|
||||||
|
hash_cache,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -54,6 +61,7 @@ impl FileBlobWriteRepository {
|
|||||||
),
|
),
|
||||||
dedup: Arc::new(DedupService::new_stub()),
|
dedup: Arc::new(DedupService::new_stub()),
|
||||||
folder_repo: Arc::new(super::folder_db_repository::FolderDbRepository::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));
|
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(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -521,8 +531,14 @@ impl FileWritePort for FileBlobWriteRepository {
|
|||||||
.await?;
|
.await?;
|
||||||
let new_hash = dedup_result.hash().to_string();
|
let new_hash = dedup_result.hash().to_string();
|
||||||
|
|
||||||
self.swap_blob_hash(file_id, &new_hash, size as i64, modified_at)
|
let swapped = self
|
||||||
.await
|
.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(
|
async fn register_file_deferred(
|
||||||
|
|||||||
Reference in New Issue
Block a user