cjw/Oxicloud
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Merge pull request #707 from EdouardVanbelle/test/gc-reap-refcount-authority

Browse Source
This commit is contained in:
Dionisio Pozo
2026-09-05 04:13:07 +02:00
committed by GitHub
parent 9597aaa424 2ee78f28c1
commit beeaf40576
3 changed files with 644 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/plan/derived-blobs.md
+24 -2
View File
@@ -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 <no registered source references it>`. 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
src/infrastructure/repositories/pg/blob_reference_sources.rs
+25
View File
@@ -76,6 +76,27 @@ fn files_exists_sql(level: RefLevel, outer_hash_expr: &str) -> Option<String> {
}
}
/// 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<String> {
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<String> {
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<String> {
chunks_exists_sql(level, outer_hash_expr)
}
async fn count_references(&self, blob_hash: &str) -> Result<u64, DomainError> {
let n: i64 = sqlx::query_scalar(
"SELECT COUNT(*) FROM storage.chunk_manifests WHERE $1 = ANY(chunk_hashes)",
src/infrastructure/services/dedup_service.rs
+595 -67
View File
@@ -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 <unreferenced>`. 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(&registry),
blob_reap_sql: blob_reap_sql(&registry),
}
}
@@ -553,6 +654,7 @@ impl DedupService {
/// entirely — see `docs/plan/derived-blobs.md`.
pub fn with_reference_registry(mut self, registry: Arc<BlobReferenceRegistry>) -> Self {
self.manifest_reap_sql = manifest_reap_sql(&registry);
self.blob_reap_sql = blob_reap_sql(&registry);
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 <unreferenced>`, 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<String> {
self.ref_exists_sql(level, outer)
}
fn ref_exists_sql(&self, level: RefLevel, outer: &str) -> Option<String> {
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<u64, DomainError> {
Ok(0)
}
async fn list_referenced_blobs(
&self,
_cursor: Option<Vec<u8>>,
_limit: usize,
) -> Result<(Vec<String>, Option<Vec<u8>>), DomainError> {
Ok((Vec::new(), None))
}
}
let mut registry = BlobReferenceRegistry::new();
registry.register(Arc::new(FakeChunkSource));
let sql = blob_reap_sql(&registry);
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<PgPool> {
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 <no registered source references it>` and does not mention
// `ref_count` at all.
//
// It used to read `ref_count <= 0 OR <unreferenced>`. 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<PgPool> {
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::<Uuid, _>("drive_id"))
.expect("storage.drives must be seeded (init-test-schema.sh)")
}
async fn local_svc(pool: &Arc<PgPool>, 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<u8> {
let mut data: Vec<u8> = (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<String>, 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<String> = 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"
);
}
}