From 8e19557074b9d36e126af75aae9bf71affbed091 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 15 Jun 2026 10:02:52 +0000 Subject: [PATCH 1/6] perf(http): exclude file downloads from the global compression layer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The global tower-http CompressionLayer compressed every response whose Content-Type was not in the already-compressed exclusion list — including large text-ish file downloads (.csv/.log/.sql/.json/source). That (a) burned CPU re-encoding multi-GB bodies on the request path with no cached result, and (b) made tower-http strip Content-Length and Accept-Ranges, breaking byte-range seek and download resume. Add a NotForDownloads predicate that skips compression for any response carrying Content-Disposition (every download surface: REST file, share, folder/zip, batch-zip, inline previews). API JSON and static assets never set Content-Disposition, so they stay compressed. Verified with the real tower-http layer + this exact predicate (64 MiB text/plain download): - Full download regains Content-Length + Accept-Ranges (were stripped); /api/data stays brotli-compressed (fix is surgical). - CPU: 2.7-3.6x less per download sequential; 6-7x less under 8-way concurrency. - TTFB: 44ms->1ms (gzip), 110ms->1ms (brotli). - Delivered content throughput: 2.4-2.8x higher. Tradeoff: genuinely-compressible downloads now send more bytes on the wire; reclaim via compress-at-rest if it ever matters. https://claude.ai/code/session_01DCszkkU11LYxMEUWr4setK --- src/main.rs | 24 +++++++++++++++++++++++- 1 file changed, 23 insertions(+), 1 deletion(-) 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)); } From f70dffeaf5b9a071e6d0228856f2956f025649c2 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 15 Jun 2026 10:56:33 +0000 Subject: [PATCH 2/6] perf(dedup): backend-aware chunk read-ahead for CDC reassembly MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit read_blob_stream / read_blob_range_stream reassembled a CDC file by fetching its chunks with `buffered(1)` — strictly sequential, so the next chunk's backend fetch (a file `open` locally; a full request round-trip on S3/Azure) only started after the current chunk was fully drained. A benchmark of the exact pipeline (stream::iter(chunks).map(get).buffered(K) .try_flatten()) showed a blind `buffered(4)` is the WRONG fix: on a local disk it is neutral on a warm page cache and ~37% SLOWER cold, because concurrent opens turn one sequential read into several competing random-I/O streams over content-addressed (scattered) chunk files. The win is entirely on remote backends, where per-chunk request latency dominates and overlapping fetches hide it (≈ linear in K). So the read-ahead depth is now a backend hint, not a constant: - BlobStorageBackend::read_prefetch() default 1 (sequential; safe for local). - S3 / Azure override to 8 (overlap GETs to hide TTFB). - cached / encrypted / retry / migration delegate to the backend that serves the bytes. - Both CDC read paths use `self.backend.read_prefetch().max(1)`. Net: local backend unchanged (no regression); remote reassembly ~4-8x faster. Ordered `buffered` (not buffer_unordered) keeps chunks in sequence. Bench (per-chunk fetch-latency model): buffered(1)->(4)/(8) = x3.9 / x7.8 @1ms, x4.0 / x8.1 @5ms, x4.0 / x8.0 @20ms. Local warm: 230ms@1 vs 227ms@4 (noise); local cold: 425ms@1 vs 585ms@4 (why local stays at 1). https://claude.ai/code/session_01DCszkkU11LYxMEUWr4setK --- src/application/ports/blob_storage_ports.rs | 14 ++++++++++++++ src/infrastructure/services/azure_blob_backend.rs | 5 +++++ src/infrastructure/services/cached_blob_backend.rs | 6 ++++++ src/infrastructure/services/dedup_service.rs | 13 +++++++++---- .../services/encrypted_blob_backend.rs | 5 +++++ .../services/migration_blob_backend.rs | 6 ++++++ src/infrastructure/services/retry_blob_backend.rs | 5 +++++ src/infrastructure/services/s3_blob_backend.rs | 5 +++++ 8 files changed, 55 insertions(+), 4 deletions(-) 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/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 } From a9ce071a518ed790b9158929257d6eed4ff3939b Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 15 Jun 2026 11:43:15 +0000 Subject: [PATCH 3/6] perf(cache): key the file content cache by blob hash, not file id MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit FileContentCache (moka, 512 MiB) was keyed by the file UUID, so content that the CDC store already deduplicates to ONE blob on disk was cached once PER FILE in RAM: N files sharing a blob held N copies, all counting against the 512 MiB cap. With effective dedup the cache filled with duplicates and thrashed. Key it by the blob hash instead (already on FileDto::content_hash): - The in-RAM cache now benefits from dedup like the disk does — each distinct blob is cached once and shared across every file/user that references it, so a download by user A warms the cache for user B's identical content. - Content is immutable by hash, so entries never go stale; the existing invalidate(file_id) calls become harmless no-ops (a UUID never matches a hash key) and can be removed in a later cleanup. - ETag is now the immutable content hash (strong validator). - Guarded: a hash-less stub DTO disables caching for that request rather than colliding every such file on the empty key. Response Content-Type still comes from the DTO, not the cache, so keying does not affect the served MIME (verified). Benchmark (real moka, exact 512 MiB/weight config, 400 files x 4 MiB = 1600 MiB working set, 4000 uniform-random accesses): dedup 1x : file_id 35.8% hit / 2568 reads vs hash 35.6% / 2577 (no dedup -> no change; control) dedup 5x : file_id 35.8% hit / 2570 reads vs hash 98.0% / 80 (32x fewer disk reads, RAM 512->320 MiB) dedup 20x: file_id 35.1% hit / 2596 reads vs hash 99.5% / 20 (130x fewer disk reads, RAM 512->80 MiB) The win scales with the dedup ratio; with no dedup it is a no-op. https://claude.ai/code/session_01DCszkkU11LYxMEUWr4setK --- .../services/file_retrieval_service.rs | 25 +++++++++++++------ 1 file changed, 17 insertions(+), 8 deletions(-) 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; } From ecbb4ec8344470d5ba68ef288ebe75a5f066104e Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 15 Jun 2026 11:54:10 +0000 Subject: [PATCH 4/6] perf(db): drop two never-used indexes on storage.files MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit storage.files carried 13 indexes; every INSERT/DELETE/rename maintains all of them. Two are never chosen by the planner for any query the app issues — verified statically (query text) AND empirically on a 50k-row table via EXPLAIN + pg_stat_user_indexes over the real query shapes (idx_scan = 0): - idx_files_name_search (user_id, name text_pattern_ops): file-name search is `name ILIKE '%term%'` (served by the GIN trgm index); text_pattern_ops can serve neither ILIKE, a leading-% substring, nor default-collation ORDER BY. The one exact `name = $1` lookup is `WHERE folder_id=$1 AND name=$2`, served by the UNIQUE (folder_id, name, user_id) index. - idx_files_category_order (category_order): only emitted as a derived type_order alias inside the folders⊎files UNION-ALL listing; the ORDER BY runs post-UNION, so a single-column files index can't presort it. The real listing uses idx_files_folder_id + a top-N sort. Benchmark (50k single-row inserts, all triggers active): ~6% faster (WITH: 10.46/10.71s; WITHOUT: 9.83/10.07s — every WITHOUT run beat every WITH run) plus less disk and WAL on every file mutation. No query regression: the planner never used these indexes. Reversible. idx_folders_path (path text_pattern_ops) is intentionally KEPT — it serves exact `WHERE path = $1` equality lookups. https://claude.ai/code/session_01DCszkkU11LYxMEUWr4setK --- .../20260702000000_drop_dead_file_indexes.sql | 35 +++++++++++++++++++ 1 file changed, 35 insertions(+) create mode 100644 migrations/20260702000000_drop_dead_file_indexes.sql 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; From d69873297a92a9c17e5763fed511b268f5f3592c Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 15 Jun 2026 12:06:48 +0000 Subject: [PATCH 5/6] perf(db): collapse the per-upload 3 DB round-trips into one CTE MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit save_file_with_blob_impl did three sequential DB round-trips per upload: resolve_user_id (SELECT folders.user_id), the INSERT, then lookup_folder_path (SELECT folders.path) — the first and third re-reading the same folders row. Replace them with a single statement: a `parent` CTE reads the folder once and the INSERT derives user_id from it and returns the path via the CTE, in one round-trip. An empty CTE (folder vanished between ingest and insert) inserts zero rows and now surfaces as a clean NotFound instead of a generic owner error. Deadlock retry, blob-ref compensation, and the 23505 (duplicate name) mapping are preserved; owner resolution + insert are now atomic (no TOCTOU). What it does and doesn't buy (benchmarked, honest): - Server CPU: UNCHANGED. A server-side 50k loop is identical (13.6s vs 13.6s) — the two extra folder reads are cached point lookups, negligible against the INSERT + per-statement triggers + 13 indexes. - Client-observed latency: 2 fewer client<->DB round-trips per upload. At the measured ~189us/round-trip on localhost that's ~0.38ms/upload; on a networked DB (RTT 0.5-1ms) ~1-2ms/upload. - Connection pool: the metadata phase holds a pooled connection for 1 round-trip instead of 3, freeing it ~3x sooner under upload concurrency. So this is a latency + connection-utilization win (and an atomicity/cleanup), not a server-CPU win. register_file_deferred keeps its own path. https://claude.ai/code/session_01DCszkkU11LYxMEUWr4setK --- .../pg/file_blob_write_repository.rs | 81 ++++++++++++------- 1 file changed, 52 insertions(+), 29 deletions(-) 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) + ) } } From b5b80549ea03ffaf27f82c0f9a3ffda1fcdaf158 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 15 Jun 2026 12:27:22 +0000 Subject: [PATCH 6/6] perf(storage): incremental per-upload usage update (O(1)) instead of full SUM MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit After every upload, maybe_update_storage_usage spawned a full `SUM(size) OVER all the user's non-trashed files` to refresh auth.users.storage_used_bytes — O(N) in the user's file count per upload, i.e. O(N²) for a bulk upload. (The covering index makes it index-only but still scans N rows.) Replace it with an O(1) incremental `storage_used_bytes += size`, keyed by the file's owner_id (dropping the brittle "My Folder - " path-parsing hack). Deletes/trash never decremented this value — they already rely on the periodic reconciliation sweep — so the model is unchanged: the sweep remains the correctness backstop for every mutation, and the counter is clamped at 0. Both stay fire-and-forget on a background task, off the upload's latency path. Benchmarked (per-call, vs the user's existing file count N): N=1k: full-SUM 202us vs incremental 123us N=10k: full-SUM 1185us vs incremental 113us (10x) N=50k: full-SUM 5397us vs incremental 114us (47x — incremental is flat O(1)) Bulk upload of 10k files (insert + usage update each): full-SUM (O(N²)) 10.37s -> incremental (O(N)) 4.89s (>2x, diverges with scale) https://claude.ai/code/session_01DCszkkU11LYxMEUWr4setK --- .../services/file_upload_service.rs | 66 ++++++++----------- .../services/storage_usage_service.rs | 26 ++++++++ 2 files changed, 54 insertions(+), 38 deletions(-) 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. ///