fix(storage): stop dedup_gc reaping manifests held only by new sources

Prerequisite 0 of docs/plan/derived-blobs.md. The zero-ref manifest
sweep read:

    WHERE m.ref_count <= 0
       OR NOT EXISTS (SELECT 1 FROM storage.files f
                       WHERE f.blob_hash = m.file_hash)

That OR hardcodes "storage.files is the only thing that can reference a
manifest". A thumbnail manifest held by storage.content_derived_blobs
has ref_count = 1, so the first clause is false — but no files row names
a thumbnail's Blob hash, so NOT EXISTS is true, the OR fires, and the
manifest is deleted, its chunks dereferenced and the bytes reaped on the
next sweep. Landing content_derived_blobs before this fix would destroy
the derived tier on the first GC run.

The second clause is not merely defensive: it is the ONLY reap path for
bulk deletes (user cascade, empty_trash), where the PG trigger touches
storage.blobs but never decrements the manifest and the per-file
cleanup_if_orphaned call is skipped. So the fix has to preserve that
role, not just add tables to the NOT EXISTS. It is now the union of
every registered manifest-level source.

Assembled once, not per sweep. An earlier cut of this change put a
format! inside the DELETE, which made the most dangerous statement in
the file unreadable, un-pasteable into psql, and injection-shaped even
though every input is &'static str. The statement is now built at
construction and stored on DedupService, so:

  * the reap loop runs a fixed statement with no string work,
  * the SQL string is stable, so prepared-statement cache keys are too,
  * a golden test pins it byte-for-byte — a reviewer reads the SQL in
    the test rather than mentally evaluating the registry,
  * initialize() logs it at debug with the contributing source names,
    recovering the "paste it into psql" property the literal had.

The registry is mandatory rather than Option. An empty registry makes
"nothing references it" vacuously true for every row, so the builder
panics instead of emitting a statement that would delete every manifest
in the database; DedupService::new always registers the two built-in
sources, so that panic is unreachable by construction. There is a test
for it.

Adds ref_exists_sql to the port, defaulting to (count) > 0 and
overridden by FilesReferenceSource with a real EXISTS. Without it the
reap predicate would have traded today's short-circuiting NOT EXISTS
for a COUNT(*) = 0 that scans every referrer — a regression precisely
on heavily-deduplicated blobs, which is what GC walks most.

fmt, clippy --all-features --all-targets, and the 11 affected unit
tests all clean.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

src/application/ports/blob_reference_ports.rs

@@ -99,6 +99,19 @@ pub trait BlobReferenceSource: Send + Sync {
     /// by a request; no fragment may interpolate caller input.
     fn ref_count_sql(&self, level: RefLevel, outer_hash_expr: &str) -> Option<String>;
 
+    /// Existence form of [`Self::ref_count_sql`] — a boolean fragment, true
+    /// when this source holds at least one reference at `level`.
+    ///
+    /// Defaults to `(<count>) > 0`. Override when the source can express a
+    /// short-circuiting `EXISTS`, which the planner can stop at the first
+    /// matching row: `dedup_gc`'s reap predicate runs this per candidate
+    /// manifest, and a heavily-deduplicated blob has many referrers, so
+    /// counting all of them where existence would do is a real regression.
+    fn ref_exists_sql(&self, level: RefLevel, outer_hash_expr: &str) -> Option<String> {
+        self.ref_count_sql(level, outer_hash_expr)
+            .map(|fragment| format!("{fragment} > 0"))
+    }
+
     /// Count of references this source holds on `blob_hash`, across both
     /// levels.
     ///
@@ -171,6 +184,28 @@ impl BlobReferenceRegistry {
         }
     }
 
+    /// Predicate selecting rows that **no** registered source references at
+    /// `level` — i.e. reap candidates.
+    ///
+    /// Returns `None` when no source contributes at this level, and callers
+    /// **must** treat that as "refuse to act" rather than substituting a
+    /// default. The natural default would be the sum-equals-zero form, which
+    /// on an empty registry reduces to `0 = 0` — vacuously true for every
+    /// row, i.e. "delete everything". Returning `None` makes that
+    /// unrepresentable at the call site instead of merely discouraged.
+    pub fn no_reference_predicate(&self, level: RefLevel, outer_hash_expr: &str) -> Option<String> {
+        let fragments: Vec<String> = self
+            .sources
+            .iter()
+            .filter_map(|s| s.ref_exists_sql(level, outer_hash_expr))
+            .collect();
+
+        if fragments.is_empty() {
+            return None;
+        }
+        Some(format!("NOT ({})", fragments.join("\n        OR ")))
+    }
+
     /// Total references held on `hash` across every source.
     ///
     /// On-demand path only — see [`BlobReferenceSource::count_references`].

src/infrastructure/repositories/pg/blob_reference_sources.rs

@@ -51,6 +51,27 @@ fn files_ref_sql(level: RefLevel, outer_hash_expr: &str) -> Option<String> {
     }
 }
 
+/// Short-circuiting existence form of [`files_ref_sql`].
+///
+/// `dedup_gc` evaluates this per candidate manifest, so counting every
+/// referrer where existence would do is a real cost on a heavily-deduplicated
+/// blob. This is also the exact shape the reap predicate used before the
+/// registry existed, so wiring it in changes no plan.
+fn files_exists_sql(level: RefLevel, outer_hash_expr: &str) -> Option<String> {
+    let f = FILES_ALIAS;
+    match level {
+        RefLevel::Chunk => Some(format!(
+            "EXISTS (SELECT 1 FROM storage.files {f} \
+             WHERE {f}.blob_hash = {outer_hash_expr} \
+             AND NOT EXISTS (SELECT 1 FROM storage.chunk_manifests {MANIFEST_ALIAS} \
+             WHERE {MANIFEST_ALIAS}.file_hash = {f}.blob_hash))"
+        )),
+        RefLevel::Manifest => Some(format!(
+            "EXISTS (SELECT 1 FROM storage.files {f} WHERE {f}.blob_hash = {outer_hash_expr})"
+        )),
+    }
+}
+
 /// Fragment for [`ChunksReferenceSource`]. See [`files_ref_sql`].
 fn chunks_ref_sql(level: RefLevel, outer_hash_expr: &str) -> Option<String> {
     match level {
@@ -99,6 +120,10 @@ impl BlobReferenceSource for FilesReferenceSource {
         files_ref_sql(level, outer_hash_expr)
     }
 
+    fn ref_exists_sql(&self, level: RefLevel, outer_hash_expr: &str) -> Option<String> {
+        files_exists_sql(level, outer_hash_expr)
+    }
+
     async fn count_references(&self, blob_hash: &str) -> Result<u64, DomainError> {
         // No level split here: the question is "how many file rows name this
         // exact hash", and a hash names either a manifest or a legacy blob,

src/infrastructure/services/dedup_service.rs

@@ -55,6 +55,7 @@ use std::sync::Arc;
 use tokio_util::io::StreamReader;
 
 use crate::application::ports::blob_lifecycle::BlobLifecycleHook;
+use crate::application::ports::blob_reference_ports::{BlobReferenceRegistry, RefLevel};
 use crate::application::ports::blob_storage_ports::BlobStorageBackend;
 use crate::application::ports::dedup_ports::{
     BlobMetadataDto, DedupPort, DedupResultDto, DedupStatsDto,
@@ -424,6 +425,51 @@ async fn populate_integrity_blob_sizes<'a>(
     IntegrityBlobSizes { hashes, sizes }
 }
 
+/// 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.
+///
+/// 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`.
+///
+/// # Panics
+///
+/// If no source contributes at [`RefLevel::Manifest`]. That is a wiring bug,
+/// and it must be loud: with no source, "nothing references it" is vacuously
+/// 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.
+fn manifest_reap_sql(registry: &BlobReferenceRegistry) -> String {
+    let orphaned = registry
+        .no_reference_predicate(RefLevel::Manifest, "m.file_hash")
+        .expect(
+            "no manifest-level blob reference source registered: the reap \
+             predicate would match every manifest",
+        );
+
+    format!(
+        "DELETE FROM storage.chunk_manifests
+ WHERE ctid = ANY(
+     SELECT ctid
+       FROM storage.chunk_manifests m
+      WHERE m.ref_count <= 0
+         OR {orphaned}
+      LIMIT $1
+ )
+ RETURNING file_hash, chunk_hashes, total_size"
+    )
+}
+
 pub struct DedupService {
     /// Pluggable blob storage backend (local FS, S3, …).
     backend: Arc<dyn BlobStorageBackend>,
@@ -442,6 +488,16 @@ pub struct DedupService {
     /// seen immediately), weight-bounded (a manifest is ~72 B per chunk),
     /// short TTL so GC'd manifests age out fast (benches/MANIFEST-CACHE.md).
     manifest_cache: moka::future::Cache<String, Arc<ChunkManifest>>,
+    /// Every table that holds blob references, so GC agrees with the
+    /// consistency jobs on what "referenced" means. Defaults to the two
+    /// built-in sources; DI replaces it once more tables exist. Never
+    /// optional — an empty registry would make "nothing references it"
+    /// vacuously true and the manifest sweep would reap everything.
+    reference_registry: Arc<BlobReferenceRegistry>,
+    /// The manifest reap statement, built once from `reference_registry`.
+    /// 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,
 }
 
 impl DedupService {
@@ -455,15 +511,32 @@ impl DedupService {
         pool: Arc<PgPool>,
         maintenance_pool: Arc<PgPool>,
     ) -> Self {
+        let registry = Arc::new(Self::default_reference_registry(pool.clone()));
         Self {
             backend,
             pool,
             maintenance_pool,
             blob_lifecycle: None,
             manifest_cache: Self::build_manifest_cache(),
+            reference_registry: registry.clone(),
+            manifest_reap_sql: manifest_reap_sql(&registry),
         }
     }
 
+    /// The two sources that were implicit before the registry existed.
+    /// Keeping this as the default means every construction path — including
+    /// tests — has a manifest-level source, so the reap predicate can never
+    /// degenerate to "nothing references anything".
+    fn default_reference_registry(pool: Arc<PgPool>) -> BlobReferenceRegistry {
+        use crate::infrastructure::repositories::pg::blob_reference_sources::{
+            ChunksReferenceSource, FilesReferenceSource,
+        };
+        let mut registry = BlobReferenceRegistry::new();
+        registry.register(Arc::new(FilesReferenceSource::new(pool.clone())));
+        registry.register(Arc::new(ChunksReferenceSource::new(pool)));
+        registry
+    }
+
     /// See the `manifest_cache` field docs. Weight ≈ real heap bytes of one
     /// entry; 32 MiB cap ≈ tens of thousands of typical (sub-1 GB) files.
     fn build_manifest_cache() -> moka::future::Cache<String, Arc<ChunkManifest>> {
@@ -476,6 +549,15 @@ impl DedupService {
             .build()
     }
 
+    /// Registers the blob-reference registry used by the manifest reap
+    /// predicate. Without it `garbage_collect` skips manifest collection
+    /// 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.reference_registry = registry;
+        self
+    }
+
     /// Registers the blob lifecycle dispatcher (thumbnail cleanup, …).
     pub fn with_blob_lifecycle(mut self, lifecycle: Arc<BlobLifecycleService>) -> Self {
         self.blob_lifecycle = Some(lifecycle);
@@ -510,12 +592,15 @@ impl DedupService {
                 .connect_lazy("postgres://invalid:5432/none")
                 .unwrap(),
         );
+        let stub_registry = Arc::new(Self::default_reference_registry(stub_pool.clone()));
         Self {
             backend: Arc::new(LocalBlobBackend::new(Path::new("/tmp/oxicloud_stub_blobs"))),
             pool: stub_pool.clone(),
-            maintenance_pool: stub_pool,
+            maintenance_pool: stub_pool.clone(),
             blob_lifecycle: None,
             manifest_cache: Self::build_manifest_cache(),
+            reference_registry: stub_registry.clone(),
+            manifest_reap_sql: manifest_reap_sql(&stub_registry),
         }
     }
 
@@ -523,6 +608,21 @@ impl DedupService {
     pub async fn initialize(&self) -> Result<(), DomainError> {
         self.backend.initialize().await?;
 
+        // The reap statement is assembled from the registered reference
+        // sources, so it is not greppable in the source tree. Log it once so an
+        // operator can read — or paste into psql — exactly what GC will delete.
+        tracing::debug!(
+            target: "oxicloud::dedup",
+            sources = ?self
+                .reference_registry
+                .sources()
+                .iter()
+                .map(|s| s.source_name())
+                .collect::<Vec<_>>(),
+            "manifest reap statement:\n{}",
+            self.manifest_reap_sql,
+        );
+
         let blob_count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM storage.blobs")
             .fetch_one(self.pool.as_ref())
             .await
@@ -2558,10 +2658,18 @@ impl DedupService {
         // 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 `storage.files.blob_hash` references its file_hash
+        //   • 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).
+        //
+        // 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.
         loop {
             // Keep the historically cheap DELETE-only shape for the dominant
             // no-work sweep. Embedding it in the delete/aggregate/update CTE
@@ -2570,23 +2678,14 @@ impl DedupService {
             // update. From two onward, aggregate in-process and issue one UPDATE:
             // the measured crossover is already positive at two, while 500 and
             // 1,000 manifests improve by 60.03x and 51.16x respectively.
-            let batch: Vec<(String, Vec<String>, i64)> = sqlx::query_as(
-                "DELETE FROM storage.chunk_manifests
-                  WHERE ctid = ANY(
-                      SELECT ctid FROM storage.chunk_manifests m
-                       WHERE m.ref_count <= 0
-                          OR NOT EXISTS (
-                              SELECT 1 FROM storage.files f
-                               WHERE f.blob_hash = m.file_hash
-                          )
-                       LIMIT $1
-                  )
-                  RETURNING file_hash, chunk_hashes, total_size",
-            )
-            .bind(BATCH_SIZE)
-            .fetch_all(self.maintenance_pool.as_ref())
-            .await
-            .map_err(|e| DomainError::internal_error("Dedup", format!("GC manifests: {e}")))?;
+            // Assembled once at construction (see `manifest_reap_sql`), not
+            // per sweep: no string work in the hot path, a stable statement for
+            // prepared-statement caching, and a byte-for-byte golden test.
+            let batch: Vec<(String, Vec<String>, i64)> = sqlx::query_as(&self.manifest_reap_sql)
+                .bind(BATCH_SIZE)
+                .fetch_all(self.maintenance_pool.as_ref())
+                .await
+                .map_err(|e| DomainError::internal_error("Dedup", format!("GC manifests: {e}")))?;
 
             if batch.is_empty() {
                 break;
@@ -3267,6 +3366,41 @@ impl crate::infrastructure::scheduler::JobHandler for DedupService {
 #[cfg(test)]
 mod tests {
     use super::*;
+
+    /// Golden test for the statement `garbage_collect` runs against production
+    /// data. It is assembled from the registered reference sources rather than
+    /// written as a literal, so this pins the whole thing byte-for-byte — the
+    /// point being that a reviewer reads the SQL *here* instead of mentally
+    /// evaluating the registry.
+    ///
+    /// If this fails after adding a source, read the diff carefully: the new
+    /// 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.
+    #[tokio::test]
+    async fn manifest_reap_statement_is_stable() {
+        let sql = DedupService::new_stub().manifest_reap_sql;
+        let expected = r#"DELETE FROM storage.chunk_manifests
+ 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))
+      LIMIT $1
+ )
+ RETURNING file_hash, chunk_hashes, total_size"#;
+        assert_eq!(sql, expected, "reap statement changed:\n{sql}");
+    }
+
+    /// The reap predicate must never match a manifest that some source still
+    /// references. With an empty registry `NOT (...)` would have no operands,
+    /// so the builder refuses rather than emitting a statement that deletes
+    /// every manifest in the database.
+    #[test]
+    #[should_panic(expected = "no manifest-level blob reference source")]
+    fn empty_registry_refuses_to_build_reap_statement() {
+        let _ = manifest_reap_sql(&BlobReferenceRegistry::new());
+    }
     use std::collections::HashSet;
     use tempfile::NamedTempFile;