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