From a7938344dd341ae4c2c67649434771a3cc0e1779 Mon Sep 17 00:00:00 2001 From: Edouard Vanbelle Date: Sun, 23 Aug 2026 19:13:19 +0200 Subject: [PATCH] feat(storage): add content_derived_blobs table + reference source MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Step 5 foundation of docs/plan/derived-blobs.md. Creates the mapping table for server-derived artifacts and registers it as a blob-reference source — deliberately BEFORE anything writes to it, which is the ordering the plan requires: dedup_gc's reap predicate has to know the table exists, or the first sweep after the first thumbnail deletes it. No writer yet, so this is inert: the table is empty and every added SQL term counts zero. The point is that the machinery is in place first. storage.content_derived_blobs maps (source_hash, kind, variant) to the derived blob_hash. The two hash columns mean different things and the migration says so at length: source_hash is a DEPENDENT pointer holding no reference (the file keeps the source alive), while blob_hash is a reference HOLDER bumping chunk_manifests.ref_count. Counting source_hash would pin every source Blob for as long as a thumbnail existed. ContentDerivedReferenceSource contributes at the manifest level only. A derived artifact's blob_hash names a Blob, never a chunk, and contributing at the chunk level would double-count — a thumbnail is almost always single-chunk, so its manifest hash equals its lone chunk's hash, the same aliasing trap the legacy-files term guards against with NOT EXISTS. There is a test for the invariant, and the chunk-level golden test passing UNCHANGED is independent confirmation. Collapses three definitions of "what references a blob" into one. Adding the source revealed that DI assembled its own registry while DedupService::new built a different default, and the two consistency test helpers built a third — so the golden tests would have pinned SQL production never runs. There is now a single `built_in_registry(pool)`; DI reads it back via DedupService::reference_registry() rather than assembling its own. The reap-predicate golden test caught the change exactly as designed, and the new branch landed inside the NOT (...) group ORed with files — so a manifest is reaped only when NEITHER source references it. A branch landing outside that group would have inverted the predicate for every other source; that is why the test pins the whole statement rather than asserting substrings. fmt, clippy --all-features --all-targets, and 15 unit tests clean. fix(migrations): order content_derived_blobs after the refcount fixes Renames 20261015000000_content_derived_blobs.sql to 20261018000000_content_derived_blobs.sql. The file was authored before the rebase onto fix/copy_folder_ref_count_issue, so its version sorted BEFORE migrations that now precede it in history: 20261016000000_copy_folder_tree_manifest_refcount.sql 20261017000000_file_delete_trigger_manifest_aware.sql 20261017000002_repair_existing_refcount_drift.sql Filename order and commit order disagreeing is the problem, not any dependency — the table is standalone and creates nothing those migrations touch. But an installation that has already applied through …17000002 would then be offered a LOWER unapplied version, which sqlx either applies out of order or rejects on its version check, and a fresh install would get an ordering no upgrade path ever produces. Reproducibility between the two is the whole point of the version prefix. Kept as its own commit rather than amending 01d90524, since interactive rebase isn't available here and rewriting mid-branch while the ref_count work is still being rebased elsewhere would churn hashes again. Worth squashing into 01d90524 at merge. No content change — pure rename, verified nothing references the old filename. Co-Authored-By: Claude Opus 5 (1M context) --- .../20261018000000_content_derived_blobs.sql | 69 ++++++++++ src/common/di.rs | 19 +-- .../repositories/pg/blob_reference_sources.rs | 129 +++++++++++++++++- .../services/blobs_consistency_service.rs | 8 +- src/infrastructure/services/dedup_service.rs | 24 ++-- .../services/manifests_consistency_service.rs | 11 +- 6 files changed, 214 insertions(+), 46 deletions(-) create mode 100644 migrations/20261018000000_content_derived_blobs.sql diff --git a/migrations/20261018000000_content_derived_blobs.sql b/migrations/20261018000000_content_derived_blobs.sql new file mode 100644 index 00000000..2baee5a2 --- /dev/null +++ b/migrations/20261018000000_content_derived_blobs.sql @@ -0,0 +1,69 @@ +-- Derived content as blobs — tier-2 refactor, step 5. +-- See `docs/plan/derived-blobs.md`. +-- +-- Maps a source Blob to the artifacts derived FROM it: thumbnails today, +-- transcodes next. Both the mapping key and the value are BLAKE3 hashes, +-- but they mean different things: +-- +-- * `source_hash` — the Blob the artifact was derived from. A +-- *dependent* reference: it keeps nothing alive (the file does), and +-- when that Blob dies these rows are deleted with it. +-- * `blob_hash` — the derived Blob itself. A reference *holder*: it +-- bumps `chunk_manifests.ref_count`, which is why +-- `ContentDerivedReferenceSource` must be registered before the first +-- row is written, or `dedup_gc` reaps the content on its next sweep. +-- +-- KEYING — the rule this table exists to enforce: +-- +-- Bytes that are a pure deterministic function of the source content +-- belong here, content-keyed, and dedupe across every file holding +-- that content. Bytes that are user-supplied or user-chosen do NOT: +-- they must be file-keyed, because content-keying them lets one user's +-- upload be served for another user's identical file. Client-uploaded +-- previews (PDF page 1, video poster frames) are the live example and +-- belong in a separate file-keyed table. +-- +-- `variant` is opaque text. New axes go INSIDE it, never into new +-- columns: 'preview-avif' beside 'preview', '720p-av1' beside '720p'. +-- That is what keeps this table from growing a column per rendering +-- parameter. +-- +-- No FK on either hash column, for the reason +-- `20260701000000_content_search_index.sql` already documents: a hash +-- resolves to either `storage.blobs` (legacy whole blob) or +-- `storage.chunk_manifests` (CDC file hash), so the reference cannot be +-- expressed as a single FK. Orphans are reclaimed by GC and reported by +-- the consistency jobs instead. +-- +-- No `size` column: the bytes are content-addressed, so their length is +-- an immutable fact the blob layer already owns via `blob_hash`. +-- `content_type` IS stored — the thumbnail handler byte-sniffs every +-- response today, and this retires that. + +CREATE TABLE IF NOT EXISTS storage.content_derived_blobs ( + source_hash VARCHAR(64) NOT NULL, + kind TEXT NOT NULL CHECK (kind IN ('thumbnail', 'transcode')), + variant TEXT NOT NULL, + blob_hash VARCHAR(64) NOT NULL, + content_type TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (source_hash, kind, variant) +); + +-- Reverse lookup: "what still references this derived Blob?" — used by +-- the manifest-level refcount recompute in `manifests_consistency` and by +-- `dedup_gc`'s reap predicate. +CREATE INDEX IF NOT EXISTS idx_content_derived_blobs_blob_hash + ON storage.content_derived_blobs (blob_hash); + +COMMENT ON TABLE storage.content_derived_blobs IS + 'Server-derived artifacts (thumbnails, transcodes) keyed by the BLAKE3 of their SOURCE content. Content-keyed on purpose: identical content shares one derivation. User-supplied bytes must NOT be stored here — see docs/plan/derived-blobs.md.'; + +COMMENT ON COLUMN storage.content_derived_blobs.source_hash IS + 'The Blob this was derived from. Dependent reference — holds no ref_count; rows are deleted when the source Blob is reaped.'; + +COMMENT ON COLUMN storage.content_derived_blobs.blob_hash IS + 'The derived Blob. Reference HOLDER — bumps chunk_manifests.ref_count via DedupService::add_reference.'; + +COMMENT ON COLUMN storage.content_derived_blobs.variant IS + 'Opaque rendering discriminator (icon | preview | large | 720p...). New axes go inside this string, never into new columns.'; diff --git a/src/common/di.rs b/src/common/di.rs index 330134ef..4630da69 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -440,22 +440,6 @@ 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( @@ -463,8 +447,7 @@ impl AppServiceFactory { db_pool.clone(), maintenance_pool.clone(), ) - .with_blob_lifecycle(blob_lifecycle) - .with_reference_registry(blob_reference_registry.clone()), + .with_blob_lifecycle(blob_lifecycle), ); dedup_service.initialize().await?; diff --git a/src/infrastructure/repositories/pg/blob_reference_sources.rs b/src/infrastructure/repositories/pg/blob_reference_sources.rs index 4424671b..ea3c1ac1 100644 --- a/src/infrastructure/repositories/pg/blob_reference_sources.rs +++ b/src/infrastructure/repositories/pg/blob_reference_sources.rs @@ -15,7 +15,9 @@ use async_trait::async_trait; use sqlx::{PgPool, Row}; use uuid::Uuid; -use crate::application::ports::blob_reference_ports::{BlobReferenceSource, RefLevel}; +use crate::application::ports::blob_reference_ports::{ + BlobReferenceRegistry, BlobReferenceSource, RefLevel, +}; use crate::domain::errors::DomainError; /// Aliases used inside the emitted fragments. @@ -26,6 +28,7 @@ use crate::domain::errors::DomainError; /// sweep and silently correlate against itself. const FILES_ALIAS: &str = "cnt_f"; const MANIFEST_ALIAS: &str = "cnt_m"; +const DERIVED_ALIAS: &str = "cnt_d"; /// Fragment for [`FilesReferenceSource`], as a free function so the SQL /// shape can be tested without constructing a pool — it is a property of @@ -87,6 +90,51 @@ fn chunks_ref_sql(level: RefLevel, outer_hash_expr: &str) -> Option { } } +/// Fragment for [`ContentDerivedReferenceSource`]. +/// +/// **Manifest level only.** A derived artifact's `blob_hash` names a Blob +/// (its own manifest), never a chunk. Contributing at the chunk level would +/// double-count, because a thumbnail is almost always single-chunk and its +/// manifest hash therefore equals its lone chunk's hash. +fn content_derived_ref_sql(level: RefLevel, outer_hash_expr: &str) -> Option { + match level { + RefLevel::Chunk => None, + RefLevel::Manifest => Some(format!( + "(SELECT COUNT(*) FROM storage.content_derived_blobs {DERIVED_ALIAS} \ + WHERE {DERIVED_ALIAS}.blob_hash = {outer_hash_expr})" + )), + } +} + +/// Short-circuiting existence form, used by `dedup_gc`'s reap predicate. +fn content_derived_exists_sql(level: RefLevel, outer_hash_expr: &str) -> Option { + match level { + RefLevel::Chunk => None, + RefLevel::Manifest => Some(format!( + "EXISTS (SELECT 1 FROM storage.content_derived_blobs {DERIVED_ALIAS} \ + WHERE {DERIVED_ALIAS}.blob_hash = {outer_hash_expr})" + )), + } +} + +/// Every built-in blob-reference source, in one place. +/// +/// THE definition of "what references a blob". `DedupService::new` uses it +/// as its construction default and hands it to the consistency jobs via +/// `reference_registry()`, so GC and the sweeps cannot disagree — and the +/// golden tests that pin the generated SQL exercise the same set production +/// runs, rather than a test-local approximation of it. +pub fn built_in_registry(pool: Arc) -> BlobReferenceRegistry { + let mut registry = BlobReferenceRegistry::new(); + registry.register(Arc::new(FilesReferenceSource::new(pool.clone()))); + registry.register(Arc::new(ChunksReferenceSource::new(pool.clone()))); + // Registered before anything writes a derived blob: dedup_gc's reap + // predicate must already know this table exists, or the first sweep + // after the first thumbnail deletes it. + registry.register(Arc::new(ContentDerivedReferenceSource::new(pool))); + registry +} + // ─── storage.files ─────────────────────────────────────────────────────── /// References held by `storage.files.blob_hash`. @@ -261,6 +309,85 @@ fn decode_uuid_cursor(bytes: &[u8]) -> Result { Ok(Uuid::from_bytes(raw)) } +// ─── storage.content_derived_blobs ─────────────────────────────────────── + +/// References held by `storage.content_derived_blobs.blob_hash` — the +/// DERIVED artifact, not the source it came from. +/// +/// **`source_hash` is deliberately not a reference.** It is a dependent +/// pointer: the source Blob is kept alive by the file that owns it, and when +/// that Blob is reaped these rows go with it. Counting `source_hash` here +/// would pin every source Blob for as long as a thumbnail existed. +pub struct ContentDerivedReferenceSource { + pool: Arc, +} + +impl ContentDerivedReferenceSource { + pub fn new(pool: Arc) -> Self { + Self { pool } + } +} + +#[async_trait] +impl BlobReferenceSource for ContentDerivedReferenceSource { + fn source_name(&self) -> &'static str { + "content_derived" + } + + fn ref_count_sql(&self, level: RefLevel, outer_hash_expr: &str) -> Option { + content_derived_ref_sql(level, outer_hash_expr) + } + + fn ref_exists_sql(&self, level: RefLevel, outer_hash_expr: &str) -> Option { + content_derived_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.content_derived_blobs WHERE blob_hash = $1", + ) + .bind(blob_hash) + .fetch_one(self.pool.as_ref()) + .await + .map_err(|e| DomainError::internal_error("BlobRefSource", format!("derived count: {e}")))?; + Ok(n.max(0) as u64) + } + + async fn list_referenced_blobs( + &self, + cursor: Option>, + limit: usize, + ) -> Result<(Vec, Option>), DomainError> { + // Paged by `blob_hash` itself — unlike files it IS the value we + // return, and DISTINCT keeps a Blob shared by several variants from + // appearing more than once per page. + let after: Option = match cursor { + Some(bytes) => Some(String::from_utf8(bytes).map_err(|e| { + DomainError::internal_error("BlobRefSource", format!("bad derived cursor: {e}")) + })?), + None => None, + }; + + let rows: Vec<(String,)> = sqlx::query_as( + "SELECT DISTINCT blob_hash FROM storage.content_derived_blobs + WHERE ($1::text IS NULL OR blob_hash > $1) + ORDER BY blob_hash + LIMIT $2", + ) + .bind(after) + .bind(limit as i64) + .fetch_all(self.pool.as_ref()) + .await + .map_err(|e| DomainError::internal_error("BlobRefSource", format!("derived page: {e}")))?; + + let next = rows + .last() + .map(|(h,)| h.clone().into_bytes()) + .filter(|_| rows.len() == limit); + Ok((rows.into_iter().map(|(h,)| h).collect(), next)) + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/src/infrastructure/services/blobs_consistency_service.rs b/src/infrastructure/services/blobs_consistency_service.rs index a2eae950..bab26707 100644 --- a/src/infrastructure/services/blobs_consistency_service.rs +++ b/src/infrastructure/services/blobs_consistency_service.rs @@ -816,9 +816,6 @@ async fn recompute_hash( #[cfg(test)] mod tests { use super::*; - use crate::infrastructure::repositories::pg::blob_reference_sources::{ - ChunksReferenceSource, FilesReferenceSource, - }; fn default_registry() -> BlobReferenceRegistry { let pool = Arc::new( @@ -826,10 +823,7 @@ mod tests { .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 + crate::infrastructure::repositories::pg::blob_reference_sources::built_in_registry(pool) } /// Golden test for the chunk-level recompute. Pins the statement diff --git a/src/infrastructure/services/dedup_service.rs b/src/infrastructure/services/dedup_service.rs index 243f037e..38cf180f 100644 --- a/src/infrastructure/services/dedup_service.rs +++ b/src/infrastructure/services/dedup_service.rs @@ -523,18 +523,17 @@ impl DedupService { } } - /// 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". + /// Every built-in blob-reference source, in one place. + /// + /// This is THE definition of "what references a blob" — DI does not + /// assemble its own, it reads this one back via + /// [`Self::reference_registry`] and hands it to the consistency jobs, so + /// GC and the sweeps cannot disagree. Keeping it as the construction + /// default also means every 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) -> 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 + crate::infrastructure::repositories::pg::blob_reference_sources::built_in_registry(pool) } /// See the `manifest_cache` field docs. Weight ≈ real heap bytes of one @@ -3388,7 +3387,8 @@ mod tests { 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)) + OR 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)) LIMIT $1 ) RETURNING file_hash, chunk_hashes, total_size"#; diff --git a/src/infrastructure/services/manifests_consistency_service.rs b/src/infrastructure/services/manifests_consistency_service.rs index 3ee5b016..3c2d1248 100644 --- a/src/infrastructure/services/manifests_consistency_service.rs +++ b/src/infrastructure/services/manifests_consistency_service.rs @@ -404,9 +404,6 @@ impl RecoverableJobHandler for ManifestsConsistencyCheck { #[cfg(test)] mod tests { use super::*; - use crate::infrastructure::repositories::pg::blob_reference_sources::{ - ChunksReferenceSource, FilesReferenceSource, - }; fn default_registry() -> BlobReferenceRegistry { let pool = Arc::new( @@ -414,10 +411,7 @@ mod tests { .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 + crate::infrastructure::repositories::pg::blob_reference_sources::built_in_registry(pool) } /// Golden test — the statement is assembled from the registry, so pin it @@ -437,7 +431,8 @@ mod tests { m.total_size AS total_size, m.chunk_count AS chunk_count, ((SELECT COUNT(*) FROM storage.files cnt_f - WHERE cnt_f.blob_hash = m.file_hash))::bigint AS actual_ref_count + WHERE cnt_f.blob_hash = m.file_hash) + + (SELECT COUNT(*) FROM storage.content_derived_blobs cnt_d WHERE cnt_d.blob_hash = m.file_hash))::bigint AS actual_ref_count FROM storage.chunk_manifests m WHERE ($1::text IS NULL OR m.file_hash > $1) ORDER BY m.file_hash