refactor(storage): drive blobs_consistency refcount from the registry
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) <noreply@anthropic.com>
This commit is contained in:
Edouard Vanbelle
+21
-1
@@ -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)
|
||||
|
||||
@@ -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<dyn BlobStorageBackend>,
|
||||
storage_entries: Vec<NamedStorageEntry>,
|
||||
storage_path_fallback: PathBuf,
|
||||
reference_registry: Arc<BlobReferenceRegistry>,
|
||||
) -> 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<BlobRow> = 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<BlobRow> = 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::<sqlx::Postgres>::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());
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<BlobReferenceRegistry> {
|
||||
self.reference_registry.clone()
|
||||
}
|
||||
|
||||
/// Registers the blob lifecycle dispatcher (thumbnail cleanup, …).
|
||||
pub fn with_blob_lifecycle(mut self, lifecycle: Arc<BlobLifecycleService>) -> Self {
|
||||
self.blob_lifecycle = Some(lifecycle);
|
||||
|
||||
Reference in New Issue
Block a user