From e0efaed5491b48e02052586dc5c45598785566b9 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 19 Jun 2026 09:37:51 +0000 Subject: [PATCH 1/2] Harden blob GC and supervise the content-index worker Deduplication GC (garbage_collect, Phase 2): - Add an orphan grace period before a ref_count=0 blob's backing file is physically deleted, mirroring git's gc.pruneExpire. New storage.blobs.orphaned_at records when a blob last reached ref_count 0; the delete trigger and every decrement / 0-ref insert path stamp it, every re-reference clears it. - Cross-check that no manifest lists the chunk and no file points at the blob before deleting it (mirrors Phase 1's file check), so a stale ref_count can only delay collection, never delete live content. - Unlink the backing files with bounded parallel fan-out. Together these close a TOCTOU where a concurrent upload of identical content could re-reference a chunk in the window between the GC row delete committing and the backing file being unlinked. Individual file deletes still reclaim eagerly; only bulk empty-trash and the periodic sweep observe the grace window. Trash: match ErrorKind::NotFound instead of substring-matching the error message when treating an already-deleted item as success. Content-index worker: supervise the drain loop and restart it with backoff after a panic, instead of letting a panic silently freeze the search index while the dirty queue grows unbounded. Adds migration 20260802000000_blob_gc_grace.sql and an integration test covering the grace window and reference cross-checks. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_0172rsVwzTwD216R9HXT2aU4 --- migrations/20260802000000_blob_gc_grace.sql | 47 ++++ src/application/services/trash_service.rs | 13 +- src/infrastructure/services/dedup_service.rs | 205 ++++++++++++++++-- .../search_index/content_index_worker.rs | 87 +++++--- 4 files changed, 300 insertions(+), 52 deletions(-) create mode 100644 migrations/20260802000000_blob_gc_grace.sql diff --git a/migrations/20260802000000_blob_gc_grace.sql b/migrations/20260802000000_blob_gc_grace.sql new file mode 100644 index 00000000..daa93dcf --- /dev/null +++ b/migrations/20260802000000_blob_gc_grace.sql @@ -0,0 +1,47 @@ +-- Garbage-collection safety for orphaned blobs. +-- +-- The dedup GC deletes a blob row (committed) and then unlinks the backing +-- file. A concurrent uploader of identical content can re-reference a chunk in +-- that window. Two mechanisms make the sweep safe: +-- (a) garbage_collect() never collects a blob still referenced by a manifest +-- (chunk) or a file (legacy whole-file blob) — cross-checks backed by +-- idx_chunk_manifests_chunk_hashes_gin and idx_files_blob_hash. A stale +-- ref_count = 0 on live content can then only delay collection, never +-- delete it. +-- (b) garbage_collect() never collects a blob that became unreferenced only +-- moments ago — the grace period below, mirroring git's gc.pruneExpire, +-- so a writer about to pin a just-orphaned chunk cannot race the sweep. +-- +-- `orphaned_at` records when ref_count last reached 0. NULL means the row is +-- referenced (ref_count > 0) or predates this column. + +ALTER TABLE storage.blobs ADD COLUMN IF NOT EXISTS orphaned_at TIMESTAMPTZ; + +-- Existing orphans start their grace window now, so applying this migration +-- never triggers an immediate sweep of content a writer might still be racing. +UPDATE storage.blobs + SET orphaned_at = now() + WHERE ref_count <= 0 AND orphaned_at IS NULL; + +-- GC scan index: orphan rows ordered by when they became collectible. Replaces +-- the old ref_count-only partial index (the GC now also filters on orphaned_at). +DROP INDEX IF EXISTS storage.idx_blobs_orphaned; +CREATE INDEX IF NOT EXISTS idx_blobs_gc_eligible + ON storage.blobs (orphaned_at) WHERE ref_count = 0; + +-- Stamp orphaned_at when a file delete drops a blob's ref_count to 0, so the +-- grace window starts at the moment of orphaning. No-op for multi-chunk files +-- whose file_hash is not itself a storage.blobs row. +CREATE OR REPLACE FUNCTION storage.decrement_blob_ref() +RETURNS trigger AS $$ +BEGIN + UPDATE storage.blobs + SET ref_count = GREATEST(ref_count - 1, 0), + orphaned_at = CASE WHEN GREATEST(ref_count - 1, 0) = 0 THEN now() ELSE orphaned_at END + WHERE hash = OLD.blob_hash; + RETURN OLD; +END; +$$ LANGUAGE plpgsql; + +COMMENT ON COLUMN storage.blobs.orphaned_at IS + 'When ref_count last reached 0; GC waits a grace period past this before deleting (NULL = referenced or pre-migration)'; diff --git a/src/application/services/trash_service.rs b/src/application/services/trash_service.rs index 935f294e..589f4ae9 100644 --- a/src/application/services/trash_service.rs +++ b/src/application/services/trash_service.rs @@ -568,9 +568,11 @@ impl TrashUseCase for TrashService { } } Err(e) => { - // Check if the file is not found - in that case, we can continue - // because we still want to remove the item from the trash index - if format!("{}", e).contains("not found") { + // File already gone — still remove the trash index + // entry. Match on the typed error kind, not the + // message text, so a reworded message can't + // silently turn this into a hard failure. + if e.kind == ErrorKind::NotFound { info!( "File not found, may already have been deleted: {}", file_id @@ -608,8 +610,9 @@ impl TrashUseCase for TrashService { info!("Successfully deleted folder permanently: {}", folder_id); } Err(e) => { - // Check if the folder is not found - in that case, we can continue - if format!("{}", e).contains("not found") { + // Folder already gone — still remove the trash + // index entry. Typed-kind match (see file branch). + if e.kind == ErrorKind::NotFound { info!( "Folder not found, may already have been deleted: {}", folder_id diff --git a/src/infrastructure/services/dedup_service.rs b/src/infrastructure/services/dedup_service.rs index 9115463f..1219d093 100644 --- a/src/infrastructure/services/dedup_service.rs +++ b/src/infrastructure/services/dedup_service.rs @@ -161,7 +161,9 @@ impl IngestGuard { ) { if !pinned.is_empty() && let Err(e) = sqlx::query( - "UPDATE storage.blobs SET ref_count = GREATEST(ref_count - 1, 0) + "UPDATE storage.blobs + SET ref_count = GREATEST(ref_count - 1, 0), + orphaned_at = CASE WHEN GREATEST(ref_count - 1, 0) = 0 THEN now() ELSE orphaned_at END WHERE hash = ANY($1)", ) .bind(&pinned) @@ -190,8 +192,8 @@ impl IngestGuard { ); } if let Err(e) = sqlx::query( - "INSERT INTO storage.blobs (hash, size, ref_count) - SELECT h, s, 0 FROM UNNEST($1::text[], $2::bigint[]) AS t(h, s) + "INSERT INTO storage.blobs (hash, size, ref_count, orphaned_at) + SELECT h, s, 0, now() FROM UNNEST($1::text[], $2::bigint[]) AS t(h, s) ON CONFLICT (hash) DO NOTHING", ) .bind(&hashes) @@ -380,6 +382,16 @@ impl DedupService { /// ~9 MiB regardless of file size. const FLUSH_MAX_BYTES: usize = 8 * 1024 * 1024; + /// Grace period (seconds) a blob must stay orphaned (`ref_count = 0`) + /// before [`garbage_collect`](Self::garbage_collect) may physically delete + /// it. Mirrors git's `gc.pruneExpire`: content that became unreferenced + /// only moments ago is never reaped, so a concurrent uploader about to pin + /// a just-orphaned chunk — or a delta-upload client that registered loose + /// chunks at `ref_count = 0` and is about to commit their manifest — cannot + /// race the sweep. Must comfortably exceed the longest plausible gap + /// between registering a chunk and referencing it (any in-flight upload). + const GC_ORPHAN_GRACE_SECS: i64 = 60 * 60; // 1 hour + /// Store content with CDC deduplication, straight from a byte stream — /// the single write path for every upload surface (REST multipart, /// WebDAV PUT, NextCloud PUT, chunked-upload assembly, WOPI PutFile). @@ -738,8 +750,8 @@ impl DedupService { let sizes: Vec = new_rows.iter().map(|(_, s)| *s).collect(); self.backend.sync_blobs(&hashes).await?; sqlx::query( - "INSERT INTO storage.blobs (hash, size, ref_count) - SELECT h, s, 0 FROM UNNEST($1::text[], $2::bigint[]) AS t(h, s) + "INSERT INTO storage.blobs (hash, size, ref_count, orphaned_at) + SELECT h, s, 0, now() FROM UNNEST($1::text[], $2::bigint[]) AS t(h, s) ON CONFLICT (hash) DO NOTHING", ) .bind(&hashes) @@ -923,7 +935,7 @@ impl DedupService { "INSERT INTO storage.blobs (hash, size, ref_count) SELECT h, s, 1 FROM UNNEST($1::text[], $2::bigint[]) AS t(h, s) ON CONFLICT (hash) DO UPDATE - SET ref_count = storage.blobs.ref_count + 1", + SET ref_count = storage.blobs.ref_count + 1, orphaned_at = NULL", ) .bind(&new_hashes) .bind(&new_sizes) @@ -971,7 +983,7 @@ impl DedupService { // session's reference NOW; hashes not returned don't exist and are // ours to write. let pinned: HashSet = sqlx::query_scalar::<_, String>( - "UPDATE storage.blobs SET ref_count = ref_count + 1 + "UPDATE storage.blobs SET ref_count = ref_count + 1, orphaned_at = NULL WHERE hash = ANY($1) RETURNING hash", ) @@ -1123,7 +1135,9 @@ impl DedupService { // Legacy blob let rows_affected = - sqlx::query("UPDATE storage.blobs SET ref_count = ref_count + 1 WHERE hash = $1") + sqlx::query( + "UPDATE storage.blobs SET ref_count = ref_count + 1, orphaned_at = NULL WHERE hash = $1", + ) .bind(hash) .execute(self.pool.as_ref()) .await @@ -1810,9 +1824,13 @@ impl DedupService { /// Garbage collect orphaned manifests and blobs. /// - /// Phase 1: Delete manifests with ref_count = 0, then decrement - /// chunk ref_counts for their chunks. - /// Phase 2: Delete blobs (chunks + legacy) with ref_count = 0. + /// Phase 1: Delete manifests with ref_count = 0 (or no referencing file), + /// then decrement chunk ref_counts for their chunks. + /// Phase 2: Delete blobs (chunks + legacy) that are unreferenced + /// (ref_count = 0), no longer listed by any manifest or file, and have + /// been orphaned for at least [`GC_ORPHAN_GRACE_SECS`](Self::GC_ORPHAN_GRACE_SECS). + /// The grace window and reference cross-checks together make the sweep safe + /// against a concurrent uploader re-referencing a just-orphaned chunk. pub async fn garbage_collect(&self) -> Result<(u64, u64), DomainError> { const BATCH_SIZE: i64 = 500; @@ -1855,9 +1873,12 @@ impl DedupService { // single-chunk file case where the PG file-delete trigger already // decremented blobs.ref_count (because file_hash == chunk_hash); // without the clamp this would underflow the CHECK constraint. + // Stamp orphaned_at so chunks freed here get the same GC grace + // window as any other newly-orphaned blob. sqlx::query( "UPDATE storage.blobs - SET ref_count = GREATEST(ref_count - 1, 0) + SET ref_count = GREATEST(ref_count - 1, 0), + orphaned_at = CASE WHEN GREATEST(ref_count - 1, 0) = 0 THEN now() ELSE orphaned_at END WHERE hash = ANY($1)", ) .bind(chunk_hashes) @@ -1880,17 +1901,44 @@ impl DedupService { } // ── Phase 2: GC orphaned blobs/chunks ──────────────────── + // A blob row is collectible only when ALL of these hold: + // • ref_count <= 0, AND + // • it has been orphaned for at least GC_ORPHAN_GRACE_SECS (or has a + // NULL orphaned_at — a pre-migration row or a path that never + // stamped it; those are safe to take immediately), AND + // • no manifest still lists it as a chunk, AND + // • no file still points at it directly (legacy whole-file blob). + // + // The two NOT EXISTS guards mirror Phase 1's file cross-check: a stale + // ref_count = 0 on still-referenced content can then only delay + // collection, never delete live bytes. The grace window keeps a + // concurrent uploader that is about to pin a just-orphaned chunk from + // racing the row-delete → file-unlink gap (see GC_ORPHAN_GRACE_SECS). + // The ctid snapshot already protects against a pin that commits DURING + // the DELETE (the pin rewrites the row's ctid, so it drops out of the + // set); grace covers the remaining post-commit unlink window. loop { let batch: Vec<(String, i64)> = sqlx::query_as( "DELETE FROM storage.blobs WHERE ctid = ANY( - SELECT ctid FROM storage.blobs - WHERE ref_count <= 0 + SELECT b.ctid FROM storage.blobs b + WHERE b.ref_count <= 0 + AND (b.orphaned_at IS NULL + OR b.orphaned_at < now() - ($2::int * interval '1 second')) + AND NOT EXISTS ( + SELECT 1 FROM storage.chunk_manifests m + WHERE m.chunk_hashes @> ARRAY[b.hash] + ) + AND NOT EXISTS ( + SELECT 1 FROM storage.files f + WHERE f.blob_hash = b.hash + ) LIMIT $1 ) RETURNING hash, size", ) .bind(BATCH_SIZE) + .bind(Self::GC_ORPHAN_GRACE_SECS as i32) .fetch_all(self.maintenance_pool.as_ref()) .await .map_err(|e| DomainError::internal_error("Dedup", format!("GC blobs: {e}")))?; @@ -1898,15 +1946,33 @@ impl DedupService { if batch.is_empty() { break; } + let n = batch.len(); - for (hash, size) in &batch { - if let Err(e) = self.backend.delete_blob(hash).await { - tracing::warn!("Failed to delete orphan blob {hash}: {e}"); - } + // The rows are already gone, so a concurrent re-upload of identical + // content recreates both row and file (durability before + // visibility); the grace window above keeps that race vanishingly + // narrow. Unlink the backing files with bounded fan-out so a large + // sweep doesn't serialise on a slow (e.g. S3) backend. + let backend = self.backend.clone(); + let deleted: Vec<(String, i64)> = stream::iter(batch) + .map(|(hash, size)| { + let backend = backend.clone(); + async move { + if let Err(e) = backend.delete_blob(&hash).await { + tracing::warn!("Failed to delete orphan blob {hash}: {e}"); + } + (hash, size) + } + }) + .buffer_unordered(Self::CHUNK_UPLOAD_CONCURRENCY) + .collect() + .await; + + for (hash, size) in &deleted { self.fire_blob_hooks(hash); total_bytes += *size as u64; } - total_deleted += batch.len() as u64; + total_deleted += n as u64; tokio::task::yield_now().await; } @@ -2256,7 +2322,9 @@ impl DedupService { return; } if let Err(e) = sqlx::query( - "UPDATE storage.blobs SET ref_count = GREATEST(ref_count - 1, 0) + "UPDATE storage.blobs + SET ref_count = GREATEST(ref_count - 1, 0), + orphaned_at = CASE WHEN GREATEST(ref_count - 1, 0) = 0 THEN now() ELSE orphaned_at END WHERE hash = ANY($1)", ) .bind(chunk_hashes) @@ -3299,6 +3367,103 @@ mod delta_upload_integration_tests { cleanup(&pool, &file_hash, file_id, &[fresh_hash]).await; } + // ── Garbage collection: grace window + reference cross-checks ─ + #[tokio::test] + async fn garbage_collect_honours_grace_window_and_references() { + let pool = test_pool().await; + let dir = TempDir::new().unwrap(); + let svc = local_svc(&pool, &dir).await; + let user = seed_user(&pool).await; + + // (A) An aged orphan (orphaned well past the grace window) with no + // references → must be collected (row + backing file). + // (B) A freshly orphaned blob (orphaned_at = now()) → must survive: a + // concurrent uploader could still be about to pin it. + let aged = blake3::hash(format!("aged-{}", Uuid::new_v4()).as_bytes()) + .to_hex() + .to_string(); + let fresh = blake3::hash(format!("fresh-{}", Uuid::new_v4()).as_bytes()) + .to_hex() + .to_string(); + for h in [&aged, &fresh] { + svc.backend() + .put_blob_from_bytes_unsynced(h, Bytes::from_static(b"xyz")) + .await + .expect("write blob"); + } + svc.backend() + .sync_blobs(&[aged.clone(), fresh.clone()]) + .await + .expect("sync"); + sqlx::query( + "INSERT INTO storage.blobs (hash, size, ref_count, orphaned_at) VALUES + ($1, 3, 0, now() - interval '2 hours'), + ($2, 3, 0, now())", + ) + .bind(&aged) + .bind(&fresh) + .execute(pool.as_ref()) + .await + .expect("seed orphans"); + + // (C) A chunk still listed by a live file's manifest, but whose + // blobs.ref_count has drifted to 0 and aged past the grace window. + // The manifest cross-check must keep it (and its bytes) alive — a + // stale ref_count must never delete referenced content. + let data = content(3 * 1024 * 1024, 71); + let (file_hash, owned_chunks, file_id) = + seed_owned_content(&svc, &pool, user, &data, "gc").await; + let referenced = owned_chunks[0].clone(); + sqlx::query( + "UPDATE storage.blobs + SET ref_count = 0, orphaned_at = now() - interval '2 hours' + WHERE hash = $1", + ) + .bind(&referenced) + .execute(pool.as_ref()) + .await + .expect("drift referenced chunk"); + + let (deleted, _bytes) = svc.garbage_collect().await.expect("gc"); + assert!(deleted >= 1, "the aged orphan must be collected"); + + // Aged orphan fully gone. + assert!( + blob_ref(&pool, &aged).await.is_none(), + "aged orphan row removed" + ); + assert!( + !svc.backend().blob_exists(&aged).await.unwrap(), + "aged orphan file unlinked" + ); + // Fresh orphan preserved by the grace window. + assert_eq!( + blob_ref(&pool, &fresh).await, + Some(0), + "fresh orphan survives the grace window" + ); + assert!( + svc.backend().blob_exists(&fresh).await.unwrap(), + "fresh orphan bytes kept" + ); + // Referenced chunk preserved by the manifest cross-check despite ref 0. + assert_eq!( + blob_ref(&pool, &referenced).await, + Some(0), + "referenced chunk row kept" + ); + assert!( + svc.backend().blob_exists(&referenced).await.unwrap(), + "referenced chunk bytes kept" + ); + + let _ = sqlx::query("DELETE FROM storage.blobs WHERE hash = ANY($1)") + .bind(vec![aged, fresh]) + .execute(pool.as_ref()) + .await; + cleanup(&pool, &file_hash, file_id, &[]).await; + } + // ── Verification read ──────────────────────────────────────── #[tokio::test] async fn hash_chunk_sequence_recomputes_and_validates_sizes() { diff --git a/src/infrastructure/services/search_index/content_index_worker.rs b/src/infrastructure/services/search_index/content_index_worker.rs index bcc74781..74d3649f 100644 --- a/src/infrastructure/services/search_index/content_index_worker.rs +++ b/src/infrastructure/services/search_index/content_index_worker.rs @@ -56,6 +56,11 @@ const PREVIEW_BYTES: usize = 16 * 1024; /// 1.5 s interval). const ORPHAN_SWEEP_TICKS: u64 = 2400; +/// Backoff before the supervisor restarts the drain loop after an abnormal +/// exit (a panic). Long enough that a tight crash-loop can't busy-spin, short +/// enough that indexing resumes promptly. +const WORKER_RESTART_BACKOFF_SECS: u64 = 5; + pub struct ContentIndexWorker { maintenance_pool: Arc, dedup: Arc, @@ -86,49 +91,77 @@ impl ContentIndexWorker { } } - /// Spawn the indexing loop. Fire-and-forget: the loop logs and survives - /// every error (an exited loop would silently freeze the index while the - /// queue grows), and the first drain runs immediately to absorb rows left - /// over from a previous run or the migration backfill. + /// Spawn the indexing loop, supervised. The drain loop logs and survives + /// every *operational* error (a failed drain just retries next tick), but a + /// panic in the loop body would otherwise kill the task and silently freeze + /// the index while the dirty queue grows unbounded. The supervisor restarts + /// the loop after a panic (with backoff) so indexing self-heals. The first + /// drain runs immediately to absorb rows left over from a previous run or + /// the migration backfill. #[instrument(skip(self))] pub fn start(self, needs_reseed: bool) { info!( "Starting content-index worker (every {}ms, batch {}, reseed: {})", self.interval_ms, DRAIN_BATCH, needs_reseed ); + let worker = Arc::new(self); tokio::spawn(async move { - if let Err(e) = self.prepare(needs_reseed).await { + // Reseed/version cleanup runs once, not on every restart. + if let Err(e) = worker.prepare(needs_reseed).await { error!("Content-index prepare failed (continuing with queue as-is): {e}"); } - let mut ticker = - tokio::time::interval(std::time::Duration::from_millis(self.interval_ms)); - ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); - let mut ticks: u64 = 0; + // run_loop() never returns under normal operation, so any exit is + // abnormal: a panic surfaces as a JoinError; a plain return would + // be a logic bug. Either way, log loudly and restart. loop { - ticker.tick().await; - for _ in 0..MAX_BATCHES_PER_TICK { - match self.drain_once().await { - Ok(0) => break, - Ok(drained) => { - debug!("Content-index drain: processed {drained} queue row(s)"); - if drained < DRAIN_BATCH as usize { - break; - } - } - Err(e) => { - error!("Content-index drain failed (queue preserved, will retry): {e}"); + let w = worker.clone(); + match tokio::spawn(async move { w.run_loop().await }).await { + Ok(()) => error!( + "Content-index drain loop returned unexpectedly; \ + restarting in {WORKER_RESTART_BACKOFF_SECS}s" + ), + Err(e) if e.is_panic() => error!( + "Content-index drain loop panicked ({e}); \ + restarting in {WORKER_RESTART_BACKOFF_SECS}s" + ), + Err(_) => return, // task cancelled — runtime shutting down + } + tokio::time::sleep(std::time::Duration::from_secs(WORKER_RESTART_BACKOFF_SECS)) + .await; + } + }); + } + + /// The perpetual drain loop. Extracted from [`start`](Self::start) so the + /// supervisor can run it in a child task and restart it after a panic. + async fn run_loop(&self) { + let mut ticker = tokio::time::interval(std::time::Duration::from_millis(self.interval_ms)); + ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + let mut ticks: u64 = 0; + loop { + ticker.tick().await; + for _ in 0..MAX_BATCHES_PER_TICK { + match self.drain_once().await { + Ok(0) => break, + Ok(drained) => { + debug!("Content-index drain: processed {drained} queue row(s)"); + if drained < DRAIN_BATCH as usize { break; } } - } - - ticks += 1; - if ticks.is_multiple_of(ORPHAN_SWEEP_TICKS) { - self.sweep_orphaned_text().await; + Err(e) => { + error!("Content-index drain failed (queue preserved, will retry): {e}"); + break; + } } } - }); + + ticks += 1; + if ticks.is_multiple_of(ORPHAN_SWEEP_TICKS) { + self.sweep_orphaned_text().await; + } + } } /// Spawn the discard-only janitor used when content search is DISABLED: From 9a6b174a30ec710224797fe8a68a08534d6aa7fa Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 19 Jun 2026 09:58:51 +0000 Subject: [PATCH 2/2] Defer CDC chunk reclamation in manifest dereference to GC MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit remove_manifest_reference unlinked a chunk's backing file right after the row-delete committed — the same TOCTOU the GC grace window was added to close: a concurrent upload of identical content can re-reference (pin) the chunk in the gap between commit and unlink, after which the deferred unlink strands a referenced chunk with no bytes. Route physical chunk reclamation through the single grace-protected path: on last reference, delete the manifest and decrement its chunks (stamping orphaned_at on the ones that reach 0), but leave the chunk rows and files for garbage_collect() to reclaim once orphaned past the grace window. The manifest deletion and its blob-keyed thumbnail hook stay eager. remove_legacy_reference and cleanup_if_orphaned's legacy path are left as eager deletes on purpose: a legacy whole-file hash can never be re-created by an ingest (uploads are always CDC now), so there is no writer to race — the existing "row gone ⇒ no resurrection" reasoning holds for them. Adds an integration test asserting a CDC manifest dereference leaves chunks orphaned-but-present, then reclaimed by a post-grace GC. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_0172rsVwzTwD216R9HXT2aU4 --- src/application/ports/dedup_ports.rs | 5 +- src/infrastructure/services/dedup_service.rs | 132 ++++++++++++++----- 2 files changed, 105 insertions(+), 32 deletions(-) diff --git a/src/application/ports/dedup_ports.rs b/src/application/ports/dedup_ports.rs index 23ddc962..36deed77 100644 --- a/src/application/ports/dedup_ports.rs +++ b/src/application/ports/dedup_ports.rs @@ -112,7 +112,10 @@ pub trait DedupPort: Send + Sync + 'static { /// Remove a reference from a blob. /// - /// Returns `true` if the blob was deleted (ref_count reached 0). + /// Returns `true` if the last reference was removed (the content is now + /// unreferenced). For CDC content the now-orphaned chunks are reclaimed + /// later by garbage collection rather than unlinked inline; legacy + /// whole-file blobs are still freed eagerly. async fn remove_reference(&self, hash: &str) -> Result; /// Calculate BLAKE3 hash of a file (streaming). diff --git a/src/infrastructure/services/dedup_service.rs b/src/infrastructure/services/dedup_service.rs index 1219d093..9e2c93f6 100644 --- a/src/infrastructure/services/dedup_service.rs +++ b/src/infrastructure/services/dedup_service.rs @@ -1164,9 +1164,13 @@ impl DedupService { /// /// For CDC manifests: decrements manifest ref_count. When it reaches 0 /// the manifest is deleted and all chunk ref_counts are decremented; - /// chunks that reach 0 are deleted from both PG and the blob backend. + /// chunks that reach 0 are left for [`garbage_collect`](Self::garbage_collect) + /// to reclaim once they have been orphaned past the grace window — unlinking + /// them here would race a concurrent upload re-referencing the same chunk. /// - /// For legacy blobs: uses a single TX with `SELECT … FOR UPDATE`. + /// For legacy blobs: uses a single TX with `SELECT … FOR UPDATE`. A legacy + /// whole-file hash can never be re-created by an ingest (uploads are always + /// CDC now), so its file is unlinked eagerly — there is no writer to race. pub async fn remove_reference(&self, hash: &str) -> Result { // ── CDC manifest path ──────────────────────────────────── let manifest = sqlx::query_as::<_, (i32, Vec)>( @@ -1187,7 +1191,12 @@ impl DedupService { self.remove_legacy_reference(hash).await } - /// Remove a manifest reference. Handles chunk cleanup when last ref is removed. + /// Remove a manifest reference. When the last reference is removed the + /// manifest is deleted and its chunks are dereferenced, but the chunk files + /// are NOT unlinked here: a chunk hash can be re-uploaded concurrently, so + /// unlinking right after the commit would race that re-reference (the same + /// TOCTOU the GC grace window guards). Newly-orphaned chunks are stamped and + /// reclaimed by [`garbage_collect`](Self::garbage_collect). async fn remove_manifest_reference( &self, file_hash: &str, @@ -1213,7 +1222,7 @@ impl DedupService { }; if current_rc <= 1 { - // Last reference — delete manifest and decrement chunks + // Last reference — delete the manifest and dereference its chunks. sqlx::query("DELETE FROM storage.chunk_manifests WHERE file_hash = $1") .bind(file_hash) .execute(&mut *tx) @@ -1222,45 +1231,36 @@ impl DedupService { DomainError::internal_error("Dedup", format!("Delete manifest: {}", e)) })?; - // Batch decrement chunk ref_counts - sqlx::query("UPDATE storage.blobs SET ref_count = ref_count - 1 WHERE hash = ANY($1)") - .bind(chunk_hashes) - .execute(&mut *tx) - .await - .map_err(|e| { - DomainError::internal_error("Dedup", format!("Decrement chunks: {}", e)) - })?; - - // Find chunks that reached 0 - let zero_chunks: Vec = sqlx::query_scalar( - "DELETE FROM storage.blobs WHERE hash = ANY($1) AND ref_count <= 0 RETURNING hash", + // Decrement chunk ref_counts and stamp orphaned_at on the ones that + // reach 0. We deliberately do NOT delete the chunk rows or unlink + // their files here: a chunk hash can be re-uploaded concurrently, so + // unlinking right after this commit would race that re-reference + // (the TOCTOU the grace window guards). garbage_collect() reclaims + // them safely once orphaned past the grace window. GREATEST clamps + // the single-chunk case where the PG file-delete trigger already + // decremented the row (file_hash == chunk_hash). + sqlx::query( + "UPDATE storage.blobs + SET ref_count = GREATEST(ref_count - 1, 0), + orphaned_at = CASE WHEN GREATEST(ref_count - 1, 0) = 0 THEN now() ELSE orphaned_at END + WHERE hash = ANY($1)", ) .bind(chunk_hashes) - .fetch_all(&mut *tx) + .execute(&mut *tx) .await - .map_err(|e| { - DomainError::internal_error("Dedup", format!("Delete zero chunks: {}", e)) - })?; + .map_err(|e| DomainError::internal_error("Dedup", format!("Decrement chunks: {}", e)))?; tx.commit() .await .map_err(|e| DomainError::internal_error("Dedup", format!("Commit: {}", e)))?; - // Delete blob files AFTER commit - for chunk_hash in &zero_chunks { - if let Err(e) = self.backend.delete_blob(chunk_hash).await { - tracing::warn!("Failed to delete chunk blob {}: {}", chunk_hash, e); - } - } - - // Bug 4 fix: notify hooks — e.g. thumbnail cleanup keyed by file_hash + // File content is gone — drop its blob-keyed thumbnails now. self.fire_blob_hooks(file_hash); tracing::info!( - "MANIFEST DELETED: {} ({} chunks, {} orphan chunks removed)", + "MANIFEST DELETED: {} ({} chunks dereferenced; orphans reclaimed by GC)", &file_hash[..12], - chunk_hashes.len(), - zero_chunks.len() + chunk_hashes.len() ); Ok(true) } else { @@ -3464,6 +3464,76 @@ mod delta_upload_integration_tests { cleanup(&pool, &file_hash, file_id, &[]).await; } + // ── Manifest dereference defers chunk reclamation to GC ────── + #[tokio::test] + async fn manifest_dereference_defers_chunk_reclamation_to_gc() { + let pool = test_pool().await; + let dir = TempDir::new().unwrap(); + let svc = local_svc(&pool, &dir).await; + let user = seed_user(&pool).await; + + // Single-owner multi-chunk CDC file → its chunks are uniquely owned. + let data = content(3 * 1024 * 1024, 91); + let (file_hash, chunks, file_id) = + seed_owned_content(&svc, &pool, user, &data, "deref").await; + assert!(chunks.len() >= 3, "3 MiB must split into ≥3 chunks"); + + // The delete_file_permanently sequence: drop the file row (PG trigger) + // then dereference the manifest. + sqlx::query("DELETE FROM storage.files WHERE id = $1") + .bind(file_id) + .execute(pool.as_ref()) + .await + .expect("delete file row"); + assert!( + svc.remove_reference(&file_hash).await.expect("deref"), + "last reference removed" + ); + + // Manifest is gone immediately… + let manifest_rc: Option = sqlx::query_scalar( + "SELECT ref_count FROM storage.chunk_manifests WHERE file_hash = $1", + ) + .bind(&file_hash) + .fetch_optional(pool.as_ref()) + .await + .expect("manifest query"); + assert!(manifest_rc.is_none(), "manifest deleted"); + + // …but the chunk rows + bytes survive at ref_count 0: no inline unlink + // that could race a concurrent re-upload of the same chunk. + for c in &chunks { + assert_eq!( + blob_ref(&pool, c).await, + Some(0), + "chunk dereferenced, not yet deleted" + ); + assert!( + svc.backend().blob_exists(c).await.unwrap(), + "chunk bytes kept until GC reclaims them" + ); + } + + // Age the orphans past the grace window; GC then reclaims rows + files. + sqlx::query( + "UPDATE storage.blobs SET orphaned_at = now() - interval '2 hours' WHERE hash = ANY($1)", + ) + .bind(&chunks) + .execute(pool.as_ref()) + .await + .expect("age orphans"); + svc.garbage_collect().await.expect("gc"); + for c in &chunks { + assert!(blob_ref(&pool, c).await.is_none(), "chunk row reclaimed"); + assert!( + !svc.backend().blob_exists(c).await.unwrap(), + "chunk file reclaimed" + ); + } + + cleanup(&pool, &file_hash, file_id, &[]).await; + } + // ── Verification read ──────────────────────────────────────── #[tokio::test] async fn hash_chunk_sequence_recomputes_and_validates_sizes() {