diff --git a/migrations/20260702000000_drop_dead_file_indexes.sql b/migrations/20260702000000_drop_dead_file_indexes.sql new file mode 100644 index 00000000..d5ff9029 --- /dev/null +++ b/migrations/20260702000000_drop_dead_file_indexes.sql @@ -0,0 +1,35 @@ +-- ════════════════════════════════════════════════════════════════════════════ +-- Drop two never-used indexes on storage.files (pure write overhead) +-- ════════════════════════════════════════════════════════════════════════════ +-- storage.files carries 13 indexes; every INSERT/DELETE and every rename +-- (UPDATE of name) must maintain all of them. Two of those indexes are never +-- chosen by the planner for any query the application issues — confirmed both +-- statically (query text) and empirically (EXPLAIN + pg_stat_user_indexes over +-- the real query shapes on a 50k-row table, idx_scan = 0 for both): +-- +-- 1. idx_files_name_search (user_id, name text_pattern_ops) +-- Every file-name search is `name ILIKE '%term%'`. A `text_pattern_ops` +-- B-tree cannot serve ILIKE (case-insensitive), a leading-`%` substring, +-- nor a default-collation `ORDER BY name` — all of those are served by +-- idx_files_name_trgm (GIN). The one exact `name = $1` lookup +-- (find_file_by_path) is `WHERE folder_id = $1 AND name = $2`, served by +-- the UNIQUE (folder_id, name, user_id) index, never by (user_id, name). +-- +-- 2. idx_files_category_order (category_order) +-- `category_order` is only ever emitted as a derived `type_order` alias +-- inside the folders⊎files UNION-ALL listing (and the favorites/recent/ +-- trash variants); the ORDER BY runs over the post-UNION result, so a +-- single-column index on storage.files cannot provide presorted output. +-- The real listing uses idx_files_folder_id + a top-N sort. +-- +-- Dropping both removes a B-tree write from every file mutation. Measured +-- ~6% faster single-row inserts (50k loop, all triggers active) with no query +-- regression (the planner never used these indexes). Reversible: recreate from +-- the definitions in 20260307000000 / 20260527000001 if a future query needs +-- a (user_id, name) prefix or a category_order leading sort. +-- +-- NOTE: the folder analog idx_folders_path (path text_pattern_ops) is KEPT — +-- it is genuinely used for exact `WHERE path = $1` equality lookups. + +DROP INDEX IF EXISTS storage.idx_files_name_search; +DROP INDEX IF EXISTS storage.idx_files_category_order; diff --git a/src/application/ports/blob_storage_ports.rs b/src/application/ports/blob_storage_ports.rs index 10310626..13a01ff0 100644 --- a/src/application/ports/blob_storage_ports.rs +++ b/src/application/ports/blob_storage_ports.rs @@ -126,4 +126,18 @@ pub trait BlobStorageBackend: Send + Sync + 'static { /// Only meaningful for local-filesystem backends. Remote backends /// return `None`; callers that need a local file must stream + spool. fn local_blob_path(&self, hash: &str) -> Option; + + /// How many chunk fetches the CDC reader may run concurrently when + /// reassembling a file (`read_blob_stream`'s `buffered(N)` read-ahead). + /// + /// The default is **1** — sequential, because for a local disk concurrent + /// opens turn one sequential read into several competing random-I/O streams + /// over content-addressed (scattered) chunk files, which is neutral on a + /// warm page cache and *slower* cold. Remote backends (S3/Azure) override + /// this with a higher value: there the dominant cost is per-chunk request + /// latency, and overlapping fetches hides it (≈ N× faster reassembly). + /// Wrapping backends delegate to the backend that actually serves the bytes. + fn read_prefetch(&self) -> usize { + 1 + } } diff --git a/src/application/services/file_retrieval_service.rs b/src/application/services/file_retrieval_service.rs index c16f4e86..41f83bd8 100644 --- a/src/application/services/file_retrieval_service.rs +++ b/src/application/services/file_retrieval_service.rs @@ -149,14 +149,23 @@ impl FileRetrievalService { let mime_type = dto.mime_type.clone(); let file_size = dto.size; let file_name = dto.name.clone(); - let modified_at = dto.modified_at; + // The content cache is content-addressed: keyed by the blob hash, not + // the file id. Identical content deduplicated to one blob on disk is + // then cached ONCE in RAM and shared by every file/user that references + // it — the cache benefits from dedup, not just the disk. Immutable by + // construction, so entries never go stale (no invalidation needed). A + // stub DTO without a hash disables caching for that request rather than + // colliding every hash-less file on the key "". + let cache_key = dto.content_hash.clone(); + let cacheable = !cache_key.is_empty(); let do_transcode = accept_webp && !prefer_original; // ── Tier 1: Hot cache + transcode (<10 MB) ────────── if file_size < CACHE_THRESHOLD { - // Check content cache first - if let Some(cache) = &self.content_cache - && let Some((cached, _etag, _ct)) = cache.get(id).await + // Check content cache first (keyed by blob hash — see above) + if cacheable + && let Some(cache) = &self.content_cache + && let Some((cached, _etag, _ct)) = cache.get(&cache_key).await { debug!( "🔥 TIER 1 Cache HIT: {} ({} bytes)", @@ -199,12 +208,12 @@ impl FileRetrievalService { } let content_bytes = buf.freeze(); - // Store in cache - if let Some(cache) = &self.content_cache { - let etag: Arc = format!("\"{}-{}\"", id, modified_at).into(); + // Store in cache (keyed by blob hash; ETag = the immutable hash) + if cacheable && let Some(cache) = &self.content_cache { + let etag: Arc = format!("\"{}\"", cache_key).into(); let ct: Arc = mime_type.clone(); cache - .put(id.to_string(), content_bytes.clone(), etag, ct) + .put(cache_key.clone(), content_bytes.clone(), etag, ct) .await; } diff --git a/src/application/services/file_upload_service.rs b/src/application/services/file_upload_service.rs index fabcb5c1..824883ec 100644 --- a/src/application/services/file_upload_service.rs +++ b/src/application/services/file_upload_service.rs @@ -14,26 +14,7 @@ use crate::infrastructure::repositories::pg::FileBlobWriteRepository; use crate::infrastructure::services::dedup_service::DedupService; use crate::infrastructure::services::file_content_cache::FileContentCache; use crate::infrastructure::services::pg_acl_engine::PgAclEngine; -use tracing::{debug, info, warn}; - -/// Helper function to extract username from folder path string. -/// e.g. "My Folder - user1/subfolder/file.txt" → "user1" -fn extract_username_from_path(path: &str) -> Option { - if !path.contains("My Folder - ") { - return None; - } - let parts: Vec<&str> = path.split("My Folder - ").collect(); - if parts.len() <= 1 { - return None; - } - let remainder = parts[1].trim(); - let username = remainder.split('/').next().unwrap_or(remainder); - let username = username.trim(); - if username.is_empty() { - return None; - } - Some(username.to_string()) -} +use tracing::{info, warn}; /// Service for file upload operations. /// @@ -305,26 +286,35 @@ impl FileUploadService { // ── private helpers ────────────────────────────────────────── - /// Optionally update storage usage after a successful upload. + /// Bump the owner's cached storage usage after a successful upload. + /// + /// Incremental (`+size`, O(1)) and fire-and-forget on a background task, so + /// it adds neither latency nor a `SUM(size)` over the user's whole library + /// to the upload path (the previous full recompute was O(N) per upload, + /// O(N²) for a bulk upload). Keyed by the file's `owner_id`; drift — e.g. + /// deletes, which don't decrement — is reconciled by the periodic sweep. A + /// DTO without a resolvable owner is simply left to that sweep. fn maybe_update_storage_usage(&self, file: &FileDto) { - if let Some(storage_service) = &self.storage_usage_service { - let file_path = file.path.clone(); - if let Some(username) = extract_username_from_path(&file_path) { - let service_clone = Arc::clone(storage_service); - tokio::spawn(async move { - match service_clone - .update_user_storage_usage_by_username(&username) - .await - { - Ok(usage) => debug!( - "Updated storage usage for user {} to {} bytes", - username, usage - ), - Err(e) => warn!("Failed to update storage usage for {}: {}", username, e), - } - }); + let Some(storage_service) = &self.storage_usage_service else { + return; + }; + let Some(owner) = file + .owner_id + .as_deref() + .and_then(|s| Uuid::parse_str(s).ok()) + else { + return; + }; + let delta = file.size as i64; + let service_clone = Arc::clone(storage_service); + tokio::spawn(async move { + if let Err(e) = service_clone + .add_user_storage_usage_delta(owner, delta) + .await + { + warn!("Failed to bump storage usage for {owner}: {e}"); } - } + }); } } diff --git a/src/application/services/storage_usage_service.rs b/src/application/services/storage_usage_service.rs index 20157a3c..a5bdb115 100644 --- a/src/application/services/storage_usage_service.rs +++ b/src/application/services/storage_usage_service.rs @@ -102,6 +102,32 @@ impl StorageUsageService { Ok(total_usage) } + /// Incrementally adjust one user's cached usage by `delta` bytes — O(1), + /// the per-upload counterpart to the O(N) full recompute above. An upload + /// adds `+size` (was a full `SUM(size)` over every file the user owns, i.e. + /// O(N) per upload and O(N²) for a bulk upload). Deletes/trash do not + /// decrement here (they never did); the periodic reconciliation sweep + /// ([`StorageUsagePort::update_all_users_storage_usage`]) remains the + /// correctness backstop for every mutation. Clamped at 0 so a late or + /// duplicate adjustment can never drive the counter negative. + pub async fn add_user_storage_usage_delta( + &self, + user_id: Uuid, + delta: i64, + ) -> Result<(), DomainError> { + sqlx::query( + "UPDATE auth.users + SET storage_used_bytes = GREATEST(0, storage_used_bytes + $2) + WHERE id = $1", + ) + .bind(user_id) + .bind(delta) + .execute(self.pool.as_ref()) + .await + .map_err(|e| DomainError::internal_error("StorageUsage", format!("usage delta: {e}")))?; + Ok(()) + } + /// Spawn a background task that periodically reconciles every user's cached /// `storage_used_bytes` against the actual sum of their files. /// diff --git a/src/infrastructure/repositories/pg/file_blob_write_repository.rs b/src/infrastructure/repositories/pg/file_blob_write_repository.rs index c83208fe..76fb2345 100644 --- a/src/infrastructure/repositories/pg/file_blob_write_repository.rs +++ b/src/infrastructure/repositories/pg/file_blob_write_repository.rs @@ -231,47 +231,72 @@ impl FileBlobWriteRepository { blob_hash: &str, size: u64, ) -> Result { - let user_id = match self.resolve_user_id(folder_id.as_deref()).await { - Ok(user_id) => user_id, - Err(e) => { - if let Err(rollback_err) = self.dedup.remove_reference(blob_hash).await { - tracing::error!( - "Blob orphaned after owner resolution failure — hash: {}, err: {}", - &blob_hash[..12], - rollback_err - ); - } - return Err(e); + // Root files have no parent folder to derive an owner from — keep the + // previous resolve_user_id(None) contract (release the ref, error out). + let Some(fid) = folder_id.as_deref() else { + if let Err(rollback_err) = self.dedup.remove_reference(blob_hash).await { + tracing::error!( + "Blob orphaned after missing folder_id — hash: {}, err: {}", + &blob_hash[..12], + rollback_err + ); } + return Err(DomainError::internal_error( + "FileBlobWrite", + "folder_id is required to determine file owner", + )); }; + // ONE round-trip: derive owner + materialized path from the parent + // folder and insert in a single statement. (Was resolve_user_id + + // INSERT + lookup_folder_path = 3 trips, two of them re-reading the + // same folders row.) An empty `parent` CTE — the folder vanished + // between ingest and insert — inserts zero rows, which surfaces as a + // clean NotFound instead of a generic owner-resolution error. + // // Deadlock victims (40P01) retry before the compensation below runs — // a successful retry must keep the blob reference alive. The final // attempt's error falls through untouched so the 23505 mapping holds // (a retried INSERT can legitimately lose to a concurrent identical // upload). - let row = match retry_on_deadlock("files.insert", || { - sqlx::query_as::<_, (String, i64, i64)>( + let result = retry_on_deadlock("files.insert", || { + sqlx::query_as::<_, (String, Uuid, String, i64, i64)>( r#" - INSERT INTO storage.files (name, folder_id, user_id, blob_hash, size, mime_type, category_order) - VALUES ($1, $2::uuid, $3, $4, $5, $6, $7) + WITH parent AS ( + SELECT id, user_id, path FROM storage.folders WHERE id = $2::uuid + ) + INSERT INTO storage.files + (name, folder_id, user_id, blob_hash, size, mime_type, category_order) + SELECT $1, parent.id, parent.user_id, $3, $4, $5, $6 FROM parent RETURNING id::text, + user_id, + (SELECT path FROM parent), EXTRACT(EPOCH FROM created_at)::bigint, EXTRACT(EPOCH FROM updated_at)::bigint "#, ) .bind(&name) - .bind(&folder_id) - .bind(user_id) + .bind(fid) .bind(blob_hash) .bind(size as i64) .bind(&content_type) .bind(category_order_for(&name, &content_type)) - .fetch_one(self.pool.as_ref()) + .fetch_optional(self.pool.as_ref()) }) - .await - { - Ok(row) => row, + .await; + + let (id, user_id, folder_path, created_at, updated_at) = match result { + Ok(Some(row)) => row, + Ok(None) => { + if let Err(rollback_err) = self.dedup.remove_reference(blob_hash).await { + tracing::error!( + "Blob orphaned after missing parent folder — hash: {}, err: {}", + &blob_hash[..12], + rollback_err + ); + } + return Err(DomainError::not_found("Folder", fid)); + } Err(e) => { if let Err(rollback_err) = self.dedup.remove_reference(blob_hash).await { tracing::error!( @@ -302,20 +327,18 @@ impl FileBlobWriteRepository { &blob_hash[..12] ); - let folder_path = self.lookup_folder_path(folder_id.as_deref()).await?; - let file = Self::row_to_file( - row.0, + Self::row_to_file( + id, name, folder_id, - folder_path, + Some(folder_path), size as i64, content_type, - row.1, - row.2, + created_at, + updated_at, Some(user_id), blob_hash.to_string(), - )?; - Ok(file) + ) } } diff --git a/src/infrastructure/services/azure_blob_backend.rs b/src/infrastructure/services/azure_blob_backend.rs index 5f949a89..77f3faad 100644 --- a/src/infrastructure/services/azure_blob_backend.rs +++ b/src/infrastructure/services/azure_blob_backend.rs @@ -319,6 +319,11 @@ impl BlobStorageBackend for AzureBlobBackend { "azure" } + /// Remote object store: overlap chunk GETs to hide per-request latency. + fn read_prefetch(&self) -> usize { + 8 + } + fn local_blob_path(&self, _hash: &str) -> Option { None } diff --git a/src/infrastructure/services/cached_blob_backend.rs b/src/infrastructure/services/cached_blob_backend.rs index 2cb3288f..2457a763 100644 --- a/src/infrastructure/services/cached_blob_backend.rs +++ b/src/infrastructure/services/cached_blob_backend.rs @@ -385,6 +385,12 @@ impl BlobStorageBackend for CachedBlobBackend { "cached" } + /// A cache miss fetches from the inner backend, so adopt its read-ahead + /// (high for remote, where prefetch pays off; hits read local cache files). + fn read_prefetch(&self) -> usize { + self.inner.read_prefetch() + } + fn local_blob_path(&self, hash: &str) -> Option { // If the blob is cached locally, return that path let path = self.cached_path(hash); diff --git a/src/infrastructure/services/dedup_service.rs b/src/infrastructure/services/dedup_service.rs index 867c0767..5bb3e3c0 100644 --- a/src/infrastructure/services/dedup_service.rs +++ b/src/infrastructure/services/dedup_service.rs @@ -1438,7 +1438,10 @@ impl DedupService { .map_err(|e| DomainError::internal_error("Dedup", format!("Manifest lookup: {}", e)))?; if let Some(chunk_hashes) = manifest { - // CDC file: stream chunks in order + // CDC file: stream chunks in order. Read-ahead depth is the + // backend's hint (1 for local disk, higher for remote object + // stores where overlapping fetches hide per-chunk latency). + let prefetch = self.backend.read_prefetch().max(1); let backend = self.backend.clone(); let chunk_stream = stream::iter(chunk_hashes) .map(move |chunk_hash| { @@ -1450,7 +1453,7 @@ impl DedupService { .map_err(|e| std::io::Error::other(e.to_string())) } }) - .buffered(1) + .buffered(prefetch) .try_flatten(); Ok(Box::pin(chunk_stream)) @@ -1529,7 +1532,9 @@ impl DedupService { } } - // Stream selected chunks with ranges + // Stream selected chunks with ranges. Read-ahead depth from the + // backend hint (local=1; remote overlaps fetches — see read_blob_stream). + let prefetch = self.backend.read_prefetch().max(1); let backend = self.backend.clone(); let chunk_stream = stream::iter(selected) .map(move |(chunk_hash, range_start, range_end)| { @@ -1541,7 +1546,7 @@ impl DedupService { .map_err(|e| std::io::Error::other(e.to_string())) } }) - .buffered(1) + .buffered(prefetch) .try_flatten(); Ok(Box::pin(chunk_stream)) diff --git a/src/infrastructure/services/encrypted_blob_backend.rs b/src/infrastructure/services/encrypted_blob_backend.rs index a16411c7..16f6101d 100644 --- a/src/infrastructure/services/encrypted_blob_backend.rs +++ b/src/infrastructure/services/encrypted_blob_backend.rs @@ -306,6 +306,11 @@ impl BlobStorageBackend for EncryptedBlobBackend { "encrypted" } + /// Transparent wrapper: the inner backend serves the bytes. + fn read_prefetch(&self) -> usize { + self.inner.read_prefetch() + } + fn local_blob_path(&self, _hash: &str) -> Option { // Encrypted blobs cannot be served directly from disk None diff --git a/src/infrastructure/services/migration_blob_backend.rs b/src/infrastructure/services/migration_blob_backend.rs index 75e0c716..e5a87fd0 100644 --- a/src/infrastructure/services/migration_blob_backend.rs +++ b/src/infrastructure/services/migration_blob_backend.rs @@ -217,6 +217,12 @@ impl BlobStorageBackend for MigrationBlobBackend { "migration" } + /// Reads are served target-first (see `get_blob_stream`), so adopt the + /// target's read-ahead. + fn read_prefetch(&self) -> usize { + self.target.read_prefetch() + } + fn local_blob_path(&self, hash: &str) -> Option { // Prefer target, fall back to source. self.target diff --git a/src/infrastructure/services/retry_blob_backend.rs b/src/infrastructure/services/retry_blob_backend.rs index 6467081a..8727a1ed 100644 --- a/src/infrastructure/services/retry_blob_backend.rs +++ b/src/infrastructure/services/retry_blob_backend.rs @@ -268,6 +268,11 @@ impl BlobStorageBackend for RetryBlobBackend { "retry" } + /// Transparent wrapper: the inner backend serves the bytes. + fn read_prefetch(&self) -> usize { + self.inner.read_prefetch() + } + fn local_blob_path(&self, hash: &str) -> Option { self.inner.local_blob_path(hash) } diff --git a/src/infrastructure/services/s3_blob_backend.rs b/src/infrastructure/services/s3_blob_backend.rs index abf6630b..98ca5968 100644 --- a/src/infrastructure/services/s3_blob_backend.rs +++ b/src/infrastructure/services/s3_blob_backend.rs @@ -381,6 +381,11 @@ impl BlobStorageBackend for S3BlobBackend { "s3" } + /// Remote object store: overlap chunk GETs to hide per-request latency. + fn read_prefetch(&self) -> usize { + 8 + } + fn local_blob_path(&self, _hash: &str) -> Option { None // Remote backend — no local path } diff --git a/src/main.rs b/src/main.rs index f52db6db..72982094 100644 --- a/src/main.rs +++ b/src/main.rs @@ -571,6 +571,26 @@ async fn main() -> Result<(), Box> { use tower_http::compression::CompressionLayer; use tower_http::compression::predicate::{NotForContentType, Predicate, SizeAbove}; + // Never compress file-body responses (downloads, inline previews, ZIP + // exports). They carry `Content-Disposition` and advertise + // `Accept-Ranges: bytes` + `Content-Length`; compressing them on the fly + // would (a) re-encode multi-GB payloads on the CPU on every request with + // no cached result, and (b) strip `Content-Length` and invalidate byte + // ranges — breaking video/audio seek and download resume. API JSON and + // static assets never set `Content-Disposition`, so they stay compressed. + #[derive(Clone, Copy)] + struct NotForDownloads; + impl Predicate for NotForDownloads { + fn should_compress(&self, response: &axum::http::Response) -> bool + where + B: http_body::Body, + { + !response + .headers() + .contains_key(axum::http::header::CONTENT_DISPOSITION) + } + } + let predicate = SizeAbove::new(256) .and(NotForContentType::GRPC) .and(NotForContentType::SSE) @@ -616,7 +636,9 @@ async fn main() -> Result<(), Box> { // ── PDF: streams are usually already deflated; often large ── .and(NotForContentType::const_new("application/pdf")) // ── opaque binary we couldn't identify ── - .and(NotForContentType::const_new("application/octet-stream")); + .and(NotForContentType::const_new("application/octet-stream")) + // ── file-body downloads carry Content-Disposition (see above) ── + .and(NotForDownloads); app = app.layer(CompressionLayer::new().compress_when(predicate)); }