From f658e557519046e4afda81657285fffe16fbcd04 Mon Sep 17 00:00:00 2001 From: Edouard Vanbelle Date: Sun, 23 Aug 2026 16:13:27 +0200 Subject: [PATCH] refactor(storage): drive blobs_consistency refcount from the registry MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Completes step 1 of docs/plan/derived-blobs.md. The chunk-level `actual_ref_count` recompute was two correlated subqueries written inline; it now sums the registered reference sources instead, so `blobs_consistency` and `dedup_gc` answer "what references this hash" from one place. If they ever diverged the sweep would bless counts the collector disagrees with — and the collector wins, destructively. No behaviour change: the generated expression is the same legacy-files term (guarded by NOT EXISTS) plus the same manifests-citing-this-chunk term, and a golden test pins the whole statement byte-for-byte. Built once at construction, like the reap statement, so the sweep runs a fixed query per page rather than assembling SQL inside the loop. The builder refuses an empty registry rather than emitting a query where every blob looks unreferenced and the entire table reports refcount_mismatch; there is a test. DI now constructs one registry and hands the same instance to both consumers — `DedupService::reference_registry()` is what `BlobsConsistencyCheck` receives, so agreement is structural rather than a convention someone has to maintain. The long comment explaining the single-chunk double-count trap moved from the query site to the builder's doc comment, where the NOT EXISTS guard it describes actually lives. fmt, clippy --all-features --all-targets and the 17 affected unit tests all clean. Co-Authored-By: Claude Opus 5 (1M context) --- src/common/di.rs | 22 ++- .../services/blobs_consistency_service.rs | 182 +++++++++++++----- src/infrastructure/services/dedup_service.rs | 10 + 3 files changed, 161 insertions(+), 53 deletions(-) diff --git a/src/common/di.rs b/src/common/di.rs index 9f3fc5b7..88d35ede 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -440,6 +440,22 @@ impl AppServiceFactory { // `blob_backend` into DedupService. let blob_backend_for_consistency = blob_backend.clone(); + // Every table holding blob references. Built ONCE and shared by the + // GC reap predicate and the consistency recompute so the two cannot + // disagree about what "referenced" means — a disagreement reaps live + // content. New blob-owning tables register here. + // See docs/plan/derived-blobs.md. + let blob_reference_registry = { + use crate::infrastructure::repositories::pg::blob_reference_sources::{ + ChunksReferenceSource, FilesReferenceSource, + }; + let mut registry = + crate::application::ports::blob_reference_ports::BlobReferenceRegistry::new(); + registry.register(Arc::new(FilesReferenceSource::new(db_pool.clone()))); + registry.register(Arc::new(ChunksReferenceSource::new(db_pool.clone()))); + Arc::new(registry) + }; + // Deduplication service — PRIMARY blob storage engine (PostgreSQL-backed index) let dedup_service = Arc::new( crate::infrastructure::services::dedup_service::DedupService::new( @@ -447,7 +463,8 @@ impl AppServiceFactory { db_pool.clone(), maintenance_pool.clone(), ) - .with_blob_lifecycle(blob_lifecycle), + .with_blob_lifecycle(blob_lifecycle) + .with_reference_registry(blob_reference_registry.clone()), ); dedup_service.initialize().await?; @@ -1493,6 +1510,9 @@ impl AppServiceFactory { core.blob_backend.clone(), core.config.storage_entries.clone(), self.storage_path.clone(), + // Same registry instance GC reaps from — see + // DedupService::reference_registry. + core.dedup_service.reference_registry(), ), ) .register_recoverable_job(&core.job_registry, &job_store_provider_dyn) diff --git a/src/infrastructure/services/blobs_consistency_service.rs b/src/infrastructure/services/blobs_consistency_service.rs index 5f0c3a0e..fed22da9 100644 --- a/src/infrastructure/services/blobs_consistency_service.rs +++ b/src/infrastructure/services/blobs_consistency_service.rs @@ -63,6 +63,7 @@ use async_trait::async_trait; use chrono::{DateTime, Duration, Utc}; use sqlx::PgPool; +use crate::application::ports::blob_reference_ports::{BlobReferenceRegistry, RefLevel}; use crate::application::ports::blob_storage_ports::BlobStorageBackend; use crate::common::config::NamedStorageEntry; use crate::infrastructure::scheduler::{ @@ -114,6 +115,64 @@ pub struct BlobsConsistencyCheck { /// fallback for a Local target entry with no `_ROOT_DIR`. Same /// fallback rule the boot path uses. storage_path_fallback: PathBuf, + /// The chunk-level page query, assembled once from the blob-reference + /// registry so this recompute and `dedup_gc` agree on what "referenced" + /// means. Built at construction rather than per page so the sweep runs a + /// fixed statement — same reasoning as `DedupService::manifest_reap_sql`. + /// See `docs/plan/derived-blobs.md`. + chunk_page_sql: String, +} + +/// The chunk-level page query, with `actual_ref_count` summed from the +/// registered reference sources. +/// +/// `storage.blobs.ref_count` semantics — the invariant `dedup_service` +/// actually maintains: +/// +/// ```text +/// ref_count = (number of chunk_manifests whose chunk_hashes[] contains +/// this hash) +/// + (number of files.blob_hash pointing at this hash on the +/// LEGACY whole-file path — files with NO manifest for their +/// blob_hash) +/// ``` +/// +/// Naively `COUNT(files) + COUNT(manifests referring)` double-counts +/// single-chunk CDC files: where a file's whole-file hash equals its lone +/// chunk's hash (anything under one CDC chunk), the file appears BOTH in +/// `files.blob_hash` and in the manifest's `chunk_hashes[]`. The +/// `NOT EXISTS` guard inside `FilesReferenceSource`'s chunk-level fragment +/// excludes CDC-path files from the legacy term so the two don't overlap. +/// +/// The GIN index on `chunk_hashes` (migration +/// `20260628000000_delta_upload_gin_index`) keeps the `= ANY(chunk_hashes)` +/// probe cheap. +/// +/// # Panics +/// +/// If no source contributes at [`RefLevel::Chunk`] — a wiring bug that +/// would make every blob look unreferenced and flag the whole table as +/// `refcount_mismatch`. +fn chunk_page_sql(registry: &BlobReferenceRegistry) -> String { + let expected = registry.ref_count_expr(RefLevel::Chunk, "b.hash"); + assert!( + expected != "0", + "no chunk-level blob reference source registered: every blob would \ + appear unreferenced" + ); + + format!( + "SELECT + b.hash AS hash, + b.size AS size, + b.ref_count AS ref_count, + b.created_at AS created_at, + ({expected})::bigint AS actual_ref_count + FROM storage.blobs b + WHERE ($1::text IS NULL OR b.hash > $1) + ORDER BY b.hash + LIMIT $2" + ) } impl BlobsConsistencyCheck { @@ -122,12 +181,14 @@ impl BlobsConsistencyCheck { backend: Arc, storage_entries: Vec, storage_path_fallback: PathBuf, + reference_registry: Arc, ) -> Self { Self { pool, backend, storage_entries, storage_path_fallback, + chunk_page_sql: chunk_page_sql(&reference_registry), } } @@ -364,58 +425,14 @@ impl RecoverableJobHandler for BlobsConsistencyCheck { } } - // Fetch the next batch. Per-row `actual_ref_count` - // computed inline via correlated subqueries — one for - // legacy whole-file references (`files.blob_hash`), one - // for CDC chunk references (`chunk_manifests.chunk_hashes`). - // GIN index on `chunk_hashes` (migration - // 20260628000000_delta_upload_gin_index) makes the - // `= ANY(chunk_hashes)` probe cheap. - // `storage.blobs.ref_count` semantics — what the invariant - // dedup_service maintains actually is: - // - // ref_count = (number of chunk_manifests whose - // chunk_hashes[] contains this hash) - // + (number of files.blob_hash pointing at - // this hash on the LEGACY whole-file path - // — i.e. files with NO manifest for their - // blob_hash) - // - // Naively `COUNT(files) + COUNT(manifests referring)` - // double-counts single-chunk CDC files: for a file whose - // whole-file hash == its single chunk's hash (any file - // small enough to fit in one CDC chunk — under ~256 KB - // average), the file appears BOTH in `files.blob_hash` - // AND in the manifest's `chunk_hashes[]`. The `NOT - // EXISTS` clause below excludes CDC-path files from the - // legacy count so the two terms don't overlap. - let rows: Vec = match sqlx::query_as( - r#" - SELECT - b.hash AS hash, - b.size AS size, - b.ref_count AS ref_count, - b.created_at AS created_at, - ( - (SELECT COUNT(*) FROM storage.files f - WHERE f.blob_hash = b.hash - AND NOT EXISTS ( - SELECT 1 FROM storage.chunk_manifests m - WHERE m.file_hash = f.blob_hash - )) - + (SELECT COUNT(*) FROM storage.chunk_manifests m - WHERE b.hash = ANY(m.chunk_hashes)) - )::bigint AS actual_ref_count - FROM storage.blobs b - WHERE ($1::text IS NULL OR b.hash > $1) - ORDER BY b.hash - LIMIT $2 - "#, - ) - .bind(cursor.as_deref()) - .bind(BATCH_SIZE) - .fetch_all(self.pool.as_ref()) - .await + // Fetch the next batch. `actual_ref_count` is summed from the + // registered reference sources — see `chunk_page_sql`, which + // documents the invariant and the single-chunk double-count trap. + let rows: Vec = match sqlx::query_as(&self.chunk_page_sql) + .bind(cursor.as_deref()) + .bind(BATCH_SIZE) + .fetch_all(self.pool.as_ref()) + .await { Ok(r) => r, Err(e) => { @@ -683,3 +700,64 @@ async fn recompute_hash( Ok(hasher.finalize().to_hex().to_string()) } + +#[cfg(test)] +mod tests { + use super::*; + use crate::infrastructure::repositories::pg::blob_reference_sources::{ + ChunksReferenceSource, FilesReferenceSource, + }; + + fn default_registry() -> BlobReferenceRegistry { + let pool = Arc::new( + sqlx::pool::PoolOptions::::new() + .connect_lazy("postgres://invalid/invalid") + .expect("lazy pool never connects"), + ); + let mut registry = BlobReferenceRegistry::new(); + registry.register(Arc::new(FilesReferenceSource::new(pool.clone()))); + registry.register(Arc::new(ChunksReferenceSource::new(pool))); + registry + } + + /// Golden test for the chunk-level recompute. Pins the statement + /// byte-for-byte because it is assembled from the registry rather than + /// written as a literal — the reviewer should read the SQL here. + /// + /// This expression must stay equal to what the query computed before the + /// registry existed: the legacy-files term guarded by `NOT EXISTS`, plus + /// the manifests-citing-this-chunk term. If a change makes those two + /// overlap, every single-chunk CDC file is counted twice and the whole + /// table reports `refcount_mismatch`. + #[tokio::test] + async fn chunk_page_statement_is_stable() { + let sql = chunk_page_sql(&default_registry()); + let expected = r#"SELECT + b.hash AS hash, + b.size AS size, + b.ref_count AS ref_count, + b.created_at AS created_at, + ((SELECT COUNT(*) 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 + )) + + (SELECT COUNT(*) FROM storage.chunk_manifests cnt_m + WHERE b.hash = ANY(cnt_m.chunk_hashes)))::bigint AS actual_ref_count + FROM storage.blobs b + WHERE ($1::text IS NULL OR b.hash > $1) + ORDER BY b.hash + LIMIT $2"#; + assert_eq!(sql, expected, "chunk page statement changed:\n{sql}"); + } + + /// With no chunk-level source every blob would look unreferenced and the + /// sweep would report the entire table as `refcount_mismatch`. Refuse to + /// build the statement instead. + #[test] + #[should_panic(expected = "no chunk-level blob reference source")] + fn empty_registry_refuses_to_build_page_statement() { + let _ = chunk_page_sql(&BlobReferenceRegistry::new()); + } +} diff --git a/src/infrastructure/services/dedup_service.rs b/src/infrastructure/services/dedup_service.rs index 3f7b6a62..cc1b68db 100644 --- a/src/infrastructure/services/dedup_service.rs +++ b/src/infrastructure/services/dedup_service.rs @@ -558,6 +558,16 @@ impl DedupService { self } + /// The registry backing the reap predicate. + /// + /// Exposed so `blobs_consistency` recomputes refcounts from the *same* + /// source set GC reaps from. If the two ever diverged, the sweep would + /// bless counts the collector disagrees with — and the collector wins, + /// destructively. + pub fn reference_registry(&self) -> Arc { + self.reference_registry.clone() + } + /// Registers the blob lifecycle dispatcher (thumbnail cleanup, …). pub fn with_blob_lifecycle(mut self, lifecycle: Arc) -> Self { self.blob_lifecycle = Some(lifecycle);