diff --git a/docs/plan/derived-blobs.md b/docs/plan/derived-blobs.md index 2a09e2b0..f96b9bdf 100644 --- a/docs/plan/derived-blobs.md +++ b/docs/plan/derived-blobs.md @@ -954,8 +954,8 @@ Findings each job reports today, and where the new tables land: | 4 | manifest → chunks | chunk reaped | `chunk_missing` (files_consistency) | ✓ | | 5 | `files` → Blob | dangling | `missing_blob` (files_consistency) | ✓ | | 6 | `storage.blobs.ref_count` | recompute | `refcount_mismatch` | ✓ chunk level only | -| 7 | `chunk_manifests.ref_count` | recompute | — | ✗ gap, pre-existing | -| 8 | manifest orphan reaping | GC predicate | `OR NOT EXISTS(files)` | ⚠ **breaks — see below** | +| 7 | `chunk_manifests.ref_count` | recompute | `refcount_mismatch` (manifests_consistency) | ✓ manifest level | +| 8 | manifest orphan reaping | GC predicate | registry `NOT EXISTS` union, no `ref_count` | ✓ | | 9 | derived/attached → Blob | dangling | — | ✗ new check needed | | 10 | `content_derived_blobs.source_hash` → Blob | orphan mapping | — | ✗ new check needed | | 11 | chunk at `ref_count = 0` past grace, still present | GC lag | — | ✗ a stalled GC is silent | @@ -974,6 +974,28 @@ pre-existing hole. Row 8 is the blocker: ### ⚠ Blocker — `dedup_gc` will delete every derived blob +> **RESOLVED.** Rows 7 and 8 are both closed, and this section is kept +> because the reasoning explains why the predicate looks the way it does. +> +> * **Row 8** — the predicate is registry-driven and, since the +> `ref_count` arm was removed, contains no counter at all: +> `WHERE `. The counter could +> otherwise delete on its own, which made a reference that was never +> taken into data loss rather than a wrong number. GC phase 2 got the +> same treatment, with the registry predicate ANDed onto its existing +> guards rather than replacing them (the counting fragments are too +> narrow to be a reap guard — see `blob_reap_sql`). +> * **Row 7** — `ManifestsConsistencyCheck` +> (`manifests_consistency_service.rs`) reconciles +> `chunk_manifests.ref_count` against the same registry, so the +> counter is corrected rather than trusted. +> +> The two were indeed coupled, as predicted below — but the resolution +> inverted the dependency. Rather than the recompute making the counter +> safe to trust, the reap predicate stopped trusting it, which demotes +> counter drift from data loss to a space leak that the recompute then +> reports. + The zero-ref manifest sweep (`dedup_service.rs:2574`) is: ```sql diff --git a/src/infrastructure/repositories/pg/blob_reference_sources.rs b/src/infrastructure/repositories/pg/blob_reference_sources.rs index e9534c82..c2b2fa39 100644 --- a/src/infrastructure/repositories/pg/blob_reference_sources.rs +++ b/src/infrastructure/repositories/pg/blob_reference_sources.rs @@ -76,6 +76,27 @@ fn files_exists_sql(level: RefLevel, outer_hash_expr: &str) -> Option { } } +/// Short-circuiting existence form of [`chunks_ref_sql`]. +/// +/// Same motivation as [`files_exists_sql`], and it now matters more: this +/// fragment sits in `dedup_gc`'s **phase-2 reap guard**, evaluated per +/// candidate blob row. Without the override the trait default wraps the +/// counting form as `(SELECT COUNT(*) …) > 0`, which scans every manifest +/// listing the chunk before comparing — a heavily-deduplicated chunk is +/// exactly the case where that is most expensive and least necessary. +fn chunks_exists_sql(level: RefLevel, outer_hash_expr: &str) -> Option { + match level { + RefLevel::Chunk => { + let m = MANIFEST_ALIAS; + Some(format!( + "EXISTS (SELECT 1 FROM storage.chunk_manifests {m} \ + WHERE {outer_hash_expr} = ANY({m}.chunk_hashes))" + )) + } + RefLevel::Manifest => None, + } +} + /// Fragment for [`ChunksReferenceSource`]. See [`files_ref_sql`]. fn chunks_ref_sql(level: RefLevel, outer_hash_expr: &str) -> Option { match level { @@ -280,6 +301,10 @@ impl BlobReferenceSource for ChunksReferenceSource { chunks_ref_sql(level, outer_hash_expr) } + fn ref_exists_sql(&self, level: RefLevel, outer_hash_expr: &str) -> Option { + chunks_exists_sql(level, outer_hash_expr) + } + async fn count_references(&self, blob_hash: &str) -> Result { let n: i64 = sqlx::query_scalar( "SELECT COUNT(*) FROM storage.chunk_manifests WHERE $1 = ANY(chunk_hashes)", diff --git a/src/infrastructure/services/dedup_service.rs b/src/infrastructure/services/dedup_service.rs index bb90967a..f5faab14 100644 --- a/src/infrastructure/services/dedup_service.rs +++ b/src/infrastructure/services/dedup_service.rs @@ -427,20 +427,49 @@ async fn populate_integrity_blob_sizes<'a>( /// Build the manifest reap statement from the registered reference sources. /// -/// A manifest is collectible when either: -/// * `ref_count` reached 0 via `cleanup_if_orphaned` on the single-file -/// delete path, **or** -/// * nothing references it any more — the bulk-delete path (user cascade, -/// `empty_trash`), where the PG trigger only touches `storage.blobs` and -/// the per-file `cleanup_if_orphaned` call is skipped, so `ref_count` is -/// never decremented and the second clause is the only thing that reaps. +/// **A manifest is collectible when, and only when, no registered source +/// references it.** The reference registry is the sole authority; `ref_count` +/// does not appear in this predicate at all. /// -/// The second clause used to name `storage.files` directly, which hardcoded -/// "files is the only thing that can reference a manifest". Any new referring -/// table — thumbnails via `storage.content_derived_blobs`, previews via -/// `storage.file_attached_blobs` — would then have its manifests reaped on the -/// next sweep *despite a correct `ref_count`*: clause one false, clause two -/// true, `OR` fires, bytes gone. See `docs/plan/derived-blobs.md`. +/// # Why `ref_count` was removed from it +/// +/// This used to read `ref_count <= 0 OR `. Each arm had a +/// purpose — the single-file delete path decrements the counter via +/// `cleanup_if_orphaned`, while bulk paths (user cascade, `empty_trash`) only +/// fire the `storage.blobs` trigger and leave the counter untouched — so the +/// disjunction looked like belt and braces. +/// +/// It was the opposite. With `OR`, **either signal alone deletes**, so a +/// counter that under-reports does not merely report a wrong number: it makes +/// live content collectible, and the registry that knows better is never +/// consulted because the first arm already matched. That is not hypothetical. +/// `storage.copy_folder_tree` used to take references with +/// `UPDATE storage.blobs … WHERE hash = blob_hash`, which matches nothing for +/// a CDC file — whose `blob_hash` names a manifest, not a chunk — so it took +/// no reference at all. Copy a folder, delete the original, and the copy's +/// bytes were reaped. +/// +/// Dropping the counter arm loses no coverage, because the single-file path +/// deletes the `storage.files` row too, which makes the row unreferenced +/// anyway. And it costs no performance: under `OR`, Postgres had to evaluate +/// the `EXISTS` union for every row whose `ref_count` was above zero — which +/// on a healthy install is nearly all of them — so the expensive predicate was +/// already running unconditionally. +/// +/// What it does change: a counter stuck *high* with no referrers left is no +/// longer reaped here. That is the bulk-delete residue, and it now belongs to +/// the manifest-level refcount recompute (`docs/plan/derived-blobs.md`, +/// coverage matrix row 7) — a counter being wrong is a job for the thing that +/// reconciles counters, not for the thing that deletes data. +/// +/// The predicate is registry-driven rather than naming `storage.files` +/// directly, so a new referring table — thumbnails via +/// `storage.content_derived_blobs`, previews via +/// `storage.file_attached_blobs` — is covered by registering its source. +/// Hardcoded, each new table would have had its manifests reaped on the next +/// sweep despite a correct `ref_count`. +/// +/// Pinned by `gc_reference_authority_integration_tests`. /// /// # Panics /// @@ -449,6 +478,74 @@ async fn populate_integrity_blob_sizes<'a>( /// true for every row and this statement would delete every manifest in the /// database. `DedupService::new` always registers `FilesReferenceSource`, so /// the only way to reach this is to pass a deliberately empty registry. +/// Build the chunk/blob reap statement (GC phase 2) from the registered +/// reference sources. +/// +/// Unlike [`manifest_reap_sql`], the registry predicate here is **added to** +/// the hardcoded guards rather than replacing them. That asymmetry is +/// deliberate and the reason this was not a mechanical swap. +/// +/// `no_reference_predicate` is built from fragments designed for *counting*, +/// and `FilesReferenceSource`'s chunk-level fragment deliberately excludes +/// files whose `blob_hash` has a manifest — otherwise a single-chunk blob, +/// where the file hash and its lone chunk hash are the same BLAKE3, would be +/// counted at both levels. Correct for a recompute; too narrow for a reap +/// guard. A `storage.blobs` row keyed by a MULTI-chunk file's hash — which +/// exists transiently while `rechunk` migrates a legacy blob, and is not a +/// member of its own manifest's `chunk_hashes` — would satisfy the registry's +/// "unreferenced" test while a live `storage.files` row still points at it. +/// Swapping the guards out would have reaped it mid-migration. +/// +/// So the statement keeps `NOT EXISTS (manifest lists it as a chunk)` and +/// `NOT EXISTS (any file points at it)`, and ANDs the registry predicate on +/// top. Adding a conjunct can only ever spare more rows, never reap more, so +/// this cannot regress; what it buys is that a future source contributing at +/// [`RefLevel::Chunk`] is honoured automatically instead of being silently +/// missed — the same failure that made Phase 1's hardcoded cross-check +/// dangerous. +/// +/// Today the registry adds nothing operationally: +/// `content_derived_blobs` and `file_attached_blobs` both return `None` at +/// `RefLevel::Chunk`, so its union is exactly manifests + legacy files. The +/// point is what happens when that stops being true. +/// +/// `$1` is the batch limit, `$2` the grace window in seconds. +/// +/// # Panics +/// +/// If no source contributes at [`RefLevel::Chunk`]. Same reasoning as +/// [`manifest_reap_sql`]: a missing predicate must be loud rather than +/// silently degrading to "nothing references anything". +fn blob_reap_sql(registry: &BlobReferenceRegistry) -> String { + let unreferenced = registry + .no_reference_predicate(RefLevel::Chunk, "b.hash") + .expect( + "no chunk-level blob reference source registered: the reap \ + predicate would lose its registry cross-check", + ); + + format!( + "DELETE FROM storage.blobs + WHERE ctid = ANY( + 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::text] + ) + AND NOT EXISTS ( + SELECT 1 FROM storage.files f + WHERE f.blob_hash = b.hash + ) + AND {unreferenced} + LIMIT $1 + ) + RETURNING hash, size" + ) +} + fn manifest_reap_sql(registry: &BlobReferenceRegistry) -> String { let orphaned = registry .no_reference_predicate(RefLevel::Manifest, "m.file_hash") @@ -462,8 +559,7 @@ fn manifest_reap_sql(registry: &BlobReferenceRegistry) -> String { WHERE ctid = ANY( SELECT ctid FROM storage.chunk_manifests m - WHERE m.ref_count <= 0 - OR {orphaned} + WHERE {orphaned} LIMIT $1 ) RETURNING file_hash, chunk_hashes, total_size" @@ -498,6 +594,10 @@ pub struct DedupService { /// Kept as a field so `garbage_collect` runs a fixed statement rather /// than assembling SQL inside a delete loop — see `manifest_reap_sql`. manifest_reap_sql: String, + /// The chunk/blob reap statement (GC phase 2), same treatment — see + /// [`blob_reap_sql`], including why its registry predicate is additive + /// rather than a replacement for the hardcoded guards. + blob_reap_sql: String, } impl DedupService { @@ -520,6 +620,7 @@ impl DedupService { manifest_cache: Self::build_manifest_cache(), reference_registry: registry.clone(), manifest_reap_sql: manifest_reap_sql(®istry), + blob_reap_sql: blob_reap_sql(®istry), } } @@ -553,6 +654,7 @@ impl DedupService { /// entirely — see `docs/plan/derived-blobs.md`. pub fn with_reference_registry(mut self, registry: Arc) -> Self { self.manifest_reap_sql = manifest_reap_sql(®istry); + self.blob_reap_sql = blob_reap_sql(®istry); self.reference_registry = registry; self } @@ -1005,6 +1107,7 @@ impl DedupService { manifest_cache: Self::build_manifest_cache(), reference_registry: stub_registry.clone(), manifest_reap_sql: manifest_reap_sql(&stub_registry), + blob_reap_sql: blob_reap_sql(&stub_registry), } } @@ -3060,21 +3163,18 @@ impl DedupService { let mut total_bytes = 0u64; // ── Phase 1: GC orphaned manifests ─────────────────────── - // A manifest is collectible when: - // • ref_count has been decremented to 0 by cleanup_if_orphaned - // on the single-file-delete service path, OR - // • NO registered reference source references its file_hash - // (covers bulk-delete paths: user cascade, empty_trash — - // where the PG trigger only touches storage.blobs and the - // per-file cleanup_if_orphaned call is skipped). + // A manifest is collectible when NO registered reference source + // references its file_hash. That single condition covers both + // delete paths: the single-file service path removes the + // storage.files row, and so do the bulk paths (user cascade, + // empty_trash) — whichever decrements ref_count along the way is + // irrelevant here. // - // The second clause used to name `storage.files` directly. That - // hardcoded "files is the only thing that can reference a manifest", - // so any new referring table (thumbnails via - // storage.content_derived_blobs, …) would see its manifests reaped - // on the next sweep despite a correct ref_count — the first clause - // is false, the second true, and the OR fires. It is now the union - // of every registered source; see docs/plan/derived-blobs.md. + // ref_count is deliberately NOT part of this. It used to be, as + // `ref_count <= 0 OR `, which meant a counter that + // under-reported deleted live content without ever consulting the + // registry that knew better. See `manifest_reap_sql` for the full + // reasoning and for what moved to the refcount recompute instead. loop { // Keep the historically cheap DELETE-only shape for the dominant // no-work sweep. Embedding it in the delete/aggregate/update CTE @@ -3205,41 +3305,29 @@ impl DedupService { // 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). + // • no file still points at it directly (legacy whole-file blob), + // AND + // • no registered reference source claims it at the chunk level. // - // 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 + // The NOT EXISTS guards mean a stale ref_count = 0 on still-referenced + // content can only delay collection, never delete live bytes — unlike + // Phase 1 before `manifest_reap_sql` dropped its ref_count arm, this + // phase always had that property. The registry conjunct is additive + // (see `blob_reap_sql`): it cannot reap anything the hardcoded guards + // would have spared, it just stops a future chunk-level source from + // being missed. 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 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::text] - ) - AND NOT EXISTS ( - SELECT 1 FROM storage.files f - WHERE f.blob_hash = b.hash - ) - LIMIT $1 - ) - RETURNING hash, size", - ) - .bind(BATCH_SIZE) - .bind(grace_secs as i32) - .fetch_all(self.maintenance_pool.as_ref()) - .await - .map_err(|e| DomainError::internal_error("Dedup", format!("GC blobs: {e}")))?; + let batch: Vec<(String, i64)> = sqlx::query_as(&self.blob_reap_sql) + .bind(BATCH_SIZE) + .bind(grace_secs as i32) + .fetch_all(self.maintenance_pool.as_ref()) + .await + .map_err(|e| DomainError::internal_error("Dedup", format!("GC blobs: {e}")))?; if batch.is_empty() { break; @@ -3827,6 +3915,13 @@ mod tests { /// branch must appear inside the `NOT (...)` group, ORed with the others. /// A branch landing outside that group inverts the predicate for every /// other source and reaps live manifests. + /// + /// **`ref_count` must not reappear in this statement.** It used to be + /// there as `ref_count <= 0 OR NOT (…)`, which let a counter that + /// under-reported delete content the registry still knew was referenced. + /// If a future change reintroduces it, this test fails, and that failure + /// is the point — see `manifest_reap_sql` and + /// `gc_reference_authority_integration_tests`. #[tokio::test] async fn manifest_reap_statement_is_stable() { let sql = DedupService::new_stub().manifest_reap_sql; @@ -3834,14 +3929,18 @@ mod tests { WHERE ctid = ANY( SELECT ctid FROM storage.chunk_manifests m - WHERE m.ref_count <= 0 - OR NOT (EXISTS (SELECT 1 FROM storage.files cnt_f WHERE cnt_f.blob_hash = m.file_hash) + WHERE NOT (EXISTS (SELECT 1 FROM storage.files cnt_f WHERE cnt_f.blob_hash = m.file_hash) OR EXISTS (SELECT 1 FROM storage.content_derived_blobs cnt_d WHERE cnt_d.blob_hash = m.file_hash) OR EXISTS (SELECT 1 FROM storage.file_attached_blobs cnt_a WHERE cnt_a.blob_hash = m.file_hash)) LIMIT $1 ) RETURNING file_hash, chunk_hashes, total_size"#; assert_eq!(sql, expected, "reap statement changed:\n{sql}"); + assert!( + !sql.contains("ref_count"), + "ref_count is back in the reap predicate — the counter must not be \ + able to delete data on its own" + ); } /// The reap predicate must never match a manifest that some source still @@ -3853,6 +3952,96 @@ mod tests { fn empty_registry_refuses_to_build_reap_statement() { let _ = manifest_reap_sql(&BlobReferenceRegistry::new()); } + + #[test] + #[should_panic(expected = "no chunk-level blob reference source")] + fn empty_registry_refuses_to_build_blob_reap_statement() { + let _ = blob_reap_sql(&BlobReferenceRegistry::new()); + } + + /// Golden test for GC phase 2, same purpose as the manifest one. + /// + /// Note what this pins that the manifest statement does not: the two + /// hardcoded `NOT EXISTS` guards **and** the registry predicate, ANDed. + /// The registry fragment is not a replacement here — see `blob_reap_sql` + /// for why substituting it would reap a legacy blob row mid-rechunk. + #[tokio::test] + async fn blob_reap_statement_is_stable() { + let sql = DedupService::new_stub().blob_reap_sql; + let expected = r#"DELETE FROM storage.blobs + WHERE ctid = ANY( + 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::text] + ) + AND NOT EXISTS ( + SELECT 1 FROM storage.files f + WHERE f.blob_hash = b.hash + ) + AND NOT (EXISTS (SELECT 1 FROM storage.files cnt_f WHERE cnt_f.blob_hash = b.hash AND NOT EXISTS (SELECT 1 FROM storage.chunk_manifests cnt_m WHERE cnt_m.file_hash = cnt_f.blob_hash)) + OR EXISTS (SELECT 1 FROM storage.chunk_manifests cnt_m WHERE b.hash = ANY(cnt_m.chunk_hashes))) + LIMIT $1 + ) + RETURNING hash, size"#; + assert_eq!(sql, expected, "blob reap statement changed:\n{sql}"); + } + + /// The reason phase 2 became registry-driven at all. + /// + /// Today no source contributes at [`RefLevel::Chunk`] beyond files and + /// manifests, so the registry conjunct is operationally redundant and a + /// golden test alone would not notice if it stopped being wired up. This + /// registers a synthetic chunk-level source and asserts its fragment + /// reaches the statement — which is what stops a future + /// `content_derived_blobs`-style table from being silently missed the way + /// Phase 1's hardcoded cross-check missed them. + #[tokio::test] + async fn a_new_chunk_level_source_reaches_the_blob_reap_statement() { + use crate::application::ports::blob_reference_ports::BlobReferenceSource; + + struct FakeChunkSource; + + #[async_trait::async_trait] + impl BlobReferenceSource for FakeChunkSource { + fn source_name(&self) -> &'static str { + "fake_chunk_source" + } + fn ref_count_sql(&self, level: RefLevel, outer: &str) -> Option { + self.ref_exists_sql(level, outer) + } + fn ref_exists_sql(&self, level: RefLevel, outer: &str) -> Option { + match level { + RefLevel::Chunk => Some(format!( + "EXISTS (SELECT 1 FROM storage.zzz_fake WHERE blob_hash = {outer})" + )), + RefLevel::Manifest => None, + } + } + async fn count_references(&self, _hash: &str) -> Result { + Ok(0) + } + async fn list_referenced_blobs( + &self, + _cursor: Option>, + _limit: usize, + ) -> Result<(Vec, Option>), DomainError> { + Ok((Vec::new(), None)) + } + } + + let mut registry = BlobReferenceRegistry::new(); + registry.register(Arc::new(FakeChunkSource)); + let sql = blob_reap_sql(®istry); + + assert!( + sql.contains("storage.zzz_fake"), + "a chunk-level source must reach the phase-2 reap guard:\n{sql}" + ); + } use std::collections::HashSet; use tempfile::NamedTempFile; @@ -4686,6 +4875,30 @@ mod rechunk_integration_tests { } } +/// Serializes every integration test that runs a **global** GC sweep. +/// +/// GC sweeps the shared integration database, while each test intentionally +/// owns a different `TempDir`-backed blob store. Two sweep tests running +/// concurrently can therefore delete test A's row through test B's backend, +/// leaving A's physical blob behind and failing an assertion that has nothing +/// to do with the code under test. Production has one shared backend for the +/// swept database; serializing only these tests models that invariant. +/// +/// **Any new test that calls `garbage_collect*` must take this guard**, +/// wherever it lives in this file. It sat inside +/// `delta_upload_integration_tests` until `gc_reference_authority_integration_tests` +/// was added without it and broke +/// `garbage_collect_honours_grace_window_and_references` — a failure that +/// appeared only in the full suite and pointed at the wrong test. Hoisted to +/// module scope so the next suite finds it. +/// +/// `allow(dead_code)`: gated on a cfg flag rather than on `test`, so a plain +/// build with `--cfg integration_tests` compiles it while `#[tokio::test]` +/// drops every caller. +#[cfg(integration_tests)] +#[allow(dead_code)] +static GC_TEST_SERIALIZER: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); + // ───────────────────────────────────────────────────────────────────────────── // Integration tests for the delta-upload primitives — the entitlement and // verification rules the chunk-negotiation protocol stands on. Same gating @@ -4702,14 +4915,6 @@ mod delta_upload_integration_tests { use tempfile::TempDir; use uuid::Uuid; - // GC sweeps the shared integration database globally, while every test - // intentionally owns a different TempDir-backed blob store. Running two - // sweep tests concurrently can therefore delete test A's row through test - // B's backend, leaving A's physical blob behind. Production has one shared - // backend for the swept database; serialize only these global-sweep tests - // so the integration topology models that invariant. - static GC_TEST_SERIALIZER: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); - async fn test_pool() -> Arc { let pool = PgPoolOptions::new() .max_connections(4) @@ -5326,3 +5531,326 @@ mod delta_upload_integration_tests { cleanup(&pool, &file_hash, file_id, &[]).await; } } + +// ───────────────────────────────────────────────────────────────────────────── +// Who decides a manifest is dead: the counter, or the reference registry? +// +// **The registry, and only the registry.** `manifest_reap_sql` asks +// `WHERE ` and does not mention +// `ref_count` at all. +// +// It used to read `ref_count <= 0 OR `. Each arm covered a +// real deletion path — the single-file path decrements the counter via +// `cleanup_if_orphaned`, bulk paths (user cascade, empty_trash) only fire the +// `storage.blobs` trigger — so the disjunction looked like belt and braces. +// It was the opposite: with OR, either signal alone deletes, so a counter +// that under-reported made live content collectible and the registry that +// knew better was never consulted. +// +// Not hypothetical. `storage.copy_folder_tree` used to take references with +// `UPDATE storage.blobs … WHERE hash = blob_hash`, which matches nothing for +// a CDC file — whose `blob_hash` names a manifest, not a chunk — so it took +// no reference at all. Copy a folder, delete the original, and the copy's +// bytes were reaped. Both copy paths now go through +// `storage.add_blob_references`, but that fix relied on getting the counter +// right, and there are two implementations of the reference contract +// (`storage.add_blob_references` in SQL, `DedupService::add_reference` in +// Rust) that must agree forever. Removing the counter's authority is what +// makes a future disagreement a leak rather than data loss. +// +// The two tests pin both directions, and they are only meaningful together: +// +// * `gc_spares_a_manifest_with_a_live_referrer` — a wrong-LOW counter must +// not delete. This is the fix. +// * `gc_reaps_an_unreferenced_manifest_despite_a_high_refcount` — a +// wrong-HIGH counter must not veto. This is the coverage the removed arm +// used to provide, and dropping it must not have traded one failure for +// the other. +// +// See `docs/plan/derived-blobs.md`. Gated on `--cfg integration_tests` like +// the other PG suites. +// ───────────────────────────────────────────────────────────────────────────── +// `allow(dead_code)`: the module is gated on a cfg flag, not on `test`, so a +// plain `cargo build --cfg integration_tests` compiles the helpers while +// `#[tokio::test]` drops their only callers. Same reason the rechunk suite +// above carries it. +#[cfg(integration_tests)] +#[allow(dead_code)] +mod gc_reference_authority_integration_tests { + use super::*; + use crate::infrastructure::services::local_blob_backend::LocalBlobBackend; + use crate::integration_test_support::{ensure_clean_test_db, test_db_url}; + use sqlx::Row; + use sqlx::postgres::PgPoolOptions; + use tempfile::TempDir; + use uuid::Uuid; + + async fn test_pool() -> Arc { + let pool = PgPoolOptions::new() + .max_connections(4) + .connect(&test_db_url()) + .await + .expect("connect to test DB — run tests/common/spawn-db.sh first"); + ensure_clean_test_db(&pool).await; + Arc::new(pool) + } + + async fn seed_user(pool: &PgPool) -> Uuid { + sqlx::query("SELECT d.id AS drive_id FROM storage.drives d WHERE d.default_for_user IS NOT NULL LIMIT 1") + .fetch_one(pool) + .await + .map(|r| r.get::("drive_id")) + .expect("storage.drives must be seeded (init-test-schema.sh)") + } + + async fn local_svc(pool: &Arc, dir: &TempDir) -> DedupService { + let backend = Arc::new(LocalBlobBackend::new(&dir.path().join("blobs"))); + backend.initialize().await.expect("init backend"); + DedupService::new(backend, pool.clone(), pool.clone()) + } + + /// Unique, poorly-compressible content of `len` bytes. The random tail + /// keeps every invocation's hash distinct, so rows left behind by a + /// panicking run can never collide with the current one. + fn content(len: usize) -> Vec { + let mut data: Vec = (0..len) + .map(|i| ((i % 251) as u8).wrapping_add((i / 7919) as u8)) + .collect(); + data.extend_from_slice(Uuid::new_v4().as_bytes()); + data + } + + /// A stored CDC blob plus a live `storage.files` row referencing it. + /// + /// The file row is inserted BEFORE the store, deliberately: phase 1 of + /// `garbage_collect` reaps manifests no source references, so with the + /// opposite order a concurrent GC from another test could reap ours in + /// the window between the two statements. BLAKE3 is deterministic, so + /// the hash is known in advance and the order costs nothing. + /// + /// Returns `(file_hash, chunk_hashes, file_id)`. + async fn seed_referenced_cdc_blob( + svc: &DedupService, + pool: &PgPool, + drive_id: Uuid, + data: &[u8], + label: &str, + ) -> (String, Vec, Uuid) { + let file_hash = blake3::hash(data).to_hex().to_string(); + + let file_id: Uuid = sqlx::query_scalar( + "INSERT INTO storage.files (name, drive_id, blob_hash, size) + VALUES ($1, $2, $3, $4) RETURNING id", + ) + .bind(format!( + "rust-test-gcauth-{label}-{}", + &Uuid::new_v4().to_string()[..8] + )) + .bind(drive_id) + .bind(&file_hash) + .bind(data.len() as i64) + .fetch_one(pool) + .await + .expect("file row"); + + let source = stream::iter(vec![Ok::<_, std::io::Error>(Bytes::copy_from_slice(data))]); + let stored = svc + .store_from_stream(source, Some("application/octet-stream".into())) + .await + .expect("store"); + assert_eq!( + stored.hash(), + file_hash, + "pre-computed BLAKE3 must match CDC-store output" + ); + + let chunks: Vec = sqlx::query_scalar( + "SELECT UNNEST(chunk_hashes) FROM storage.chunk_manifests WHERE file_hash = $1", + ) + .bind(&file_hash) + .fetch_all(pool) + .await + .expect("chunks"); + + // Fixture premise. A single-chunk blob has `file_hash == chunk_hash` + // (both BLAKE3 over the same bytes), which is the aliasing case the + // reference contract carries a `NOT EXISTS` guard for. This suite is + // about the multi-chunk shape — the one the copy bug broke, where + // `blob_hash` names a manifest that `storage.blobs` has no row for — + // so assert we actually got it rather than silently testing the easy + // case if CDC parameters change. + assert!( + chunks.len() > 1, + "fixture must be multi-chunk to exercise the manifest level, got {} \ + chunk(s) for {} bytes (CDC_AVG_CHUNK = {CDC_AVG_CHUNK})", + chunks.len(), + data.len() + ); + + (file_hash, chunks, file_id) + } + + async fn manifest_exists(pool: &PgPool, file_hash: &str) -> bool { + sqlx::query_scalar::<_, i64>( + "SELECT COUNT(*) FROM storage.chunk_manifests WHERE file_hash = $1", + ) + .bind(file_hash) + .fetch_one(pool) + .await + .expect("count manifests") + > 0 + } + + /// Simulate a reference that was never taken: the file row is live, the + /// counter says nothing needs the content. Exactly the state the + /// `copy_folder_tree` bug produced, and the state any future divergence + /// between the SQL and Rust reference contracts would produce. + async fn force_zero_manifest_refcount(pool: &PgPool, file_hash: &str) { + let updated = + sqlx::query("UPDATE storage.chunk_manifests SET ref_count = 0 WHERE file_hash = $1") + .bind(file_hash) + .execute(pool) + .await + .expect("zero the manifest refcount") + .rows_affected(); + assert_eq!(updated, 1, "expected exactly one manifest for {file_hash}"); + } + + async fn cleanup(pool: &PgPool, file_hash: &str, file_id: Uuid, chunks: &[String]) { + let _ = sqlx::query("DELETE FROM storage.files WHERE id = $1") + .bind(file_id) + .execute(pool) + .await; + let _ = sqlx::query( + "DELETE FROM storage.files + WHERE blob_hash = $1 AND name LIKE 'rust-test-gcauth-%'", + ) + .bind(file_hash) + .execute(pool) + .await; + let _ = sqlx::query("DELETE FROM storage.chunk_manifests WHERE file_hash = $1") + .bind(file_hash) + .execute(pool) + .await; + let mut to_drop = chunks.to_vec(); + to_drop.push(file_hash.to_string()); + let _ = sqlx::query("DELETE FROM storage.blobs WHERE hash = ANY($1)") + .bind(&to_drop) + .execute(pool) + .await; + } + + /// The coverage that dropping the `ref_count` arm had to preserve. + /// + /// Bulk-delete paths (user cascade, `empty_trash`) remove + /// `storage.files` rows via a trigger that only touches `storage.blobs`, + /// so the manifest's counter is left **stuck high** with no referrers. + /// Under the old `OR` predicate the registry arm collected those. Now + /// that the registry is the sole authority it still does — a high counter + /// no longer keeps dead content alive, just as a zero one no longer kills + /// live content. + /// + /// This is the direction the counter can still be wrong in, and it is the + /// benign one: a leak, detected by the refcount recompute, not data loss. + #[tokio::test] + async fn gc_reaps_an_unreferenced_manifest_despite_a_high_refcount() { + let _gc_test_guard = GC_TEST_SERIALIZER.lock().await; + let pool = test_pool().await; + let drive_id = seed_user(&pool).await; + let dir = TempDir::new().expect("tempdir"); + let svc = local_svc(&pool, &dir).await; + + let data = content(2 * 1024 * 1024); + let (file_hash, chunks, file_id) = + seed_referenced_cdc_blob(&svc, &pool, drive_id, &data, "stuckhigh").await; + + // Simulate the bulk path: referrer gone, counter untouched. + sqlx::query("DELETE FROM storage.files WHERE id = $1") + .bind(file_id) + .execute(pool.as_ref()) + .await + .expect("drop the referrer"); + let bumped = + sqlx::query("UPDATE storage.chunk_manifests SET ref_count = 7 WHERE file_hash = $1") + .bind(&file_hash) + .execute(pool.as_ref()) + .await + .expect("inflate the refcount") + .rows_affected(); + assert_eq!(bumped, 1, "expected exactly one manifest for {file_hash}"); + + // Plain GC, NOT `garbage_collect_force`. Phase 1 has no time filter — + // the manifest predicate is purely "is it referenced" — so the grace + // window is irrelevant to what these tests assert. Forcing it would + // bypass the CHUNK-level grace for the whole shared test database and + // reap sibling tests' just-uploaded orphans; that is exactly how this + // suite first broke `claim_and_pin_respect_ownership_and_orphans`. + svc.garbage_collect().await.expect("gc"); + + let survived = manifest_exists(&pool, &file_hash).await; + cleanup(&pool, &file_hash, file_id, &chunks).await; + + assert!( + !survived, + "GC left a manifest nothing references, because its ref_count was \ + above zero. Removing the `ref_count <= 0` arm must not have made \ + the counter able to VETO collection either — the registry is the \ + authority in both directions." + ); + } + + /// **The contract.** A manifest with a live `storage.files` referrer + /// survives GC no matter what its counter says. + /// + /// This failed until `manifest_reap_sql` dropped its `ref_count <= 0` + /// arm. The counter was a second, independent licence to delete, so a + /// reference that was never taken — the `copy_folder_tree` bug — destroyed + /// the copy's content rather than merely mis-reporting a number. + #[tokio::test] + async fn gc_spares_a_manifest_with_a_live_referrer() { + let _gc_test_guard = GC_TEST_SERIALIZER.lock().await; + let pool = test_pool().await; + let drive_id = seed_user(&pool).await; + let dir = TempDir::new().expect("tempdir"); + let svc = local_svc(&pool, &dir).await; + + let data = content(2 * 1024 * 1024); + let (file_hash, chunks, file_id) = + seed_referenced_cdc_blob(&svc, &pool, drive_id, &data, "spare").await; + + force_zero_manifest_refcount(&pool, &file_hash).await; + + // The file row is still there — this is the whole premise, so assert + // it rather than trusting that nothing else reaped it concurrently. + let referrers: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM storage.files WHERE id = $1") + .bind(file_id) + .fetch_one(pool.as_ref()) + .await + .expect("count referrers"); + assert_eq!( + referrers, 1, + "fixture file row must still reference the blob" + ); + + // Plain GC — see the sibling test for why `force` is wrong here. + svc.garbage_collect().await.expect("gc"); + + let survived = manifest_exists(&pool, &file_hash).await; + let readable = svc.read_blob_stream(&file_hash).await.is_ok(); + cleanup(&pool, &file_hash, file_id, &chunks).await; + + assert!( + survived, + "GC reaped a manifest that storage.files still references. \ + ref_count was 0 and something let that alone decide — check \ + whether `manifest_reap_sql` has regained a `ref_count` clause. \ + FilesReferenceSource is registered and knows the row is live; it \ + must be the only authority on collectibility." + ); + assert!( + readable, + "manifest survived but its content is unreadable — chunk-level \ + reclamation followed the same zero counter" + ); + } +}