feat(storage): add content_derived_blobs table + reference source
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) <noreply@anthropic.com>
This commit is contained in:
Edouard Vanbelle
@@ -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.';
|
||||
+1
-18
@@ -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?;
|
||||
|
||||
|
||||
@@ -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<String> {
|
||||
}
|
||||
}
|
||||
|
||||
/// 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<String> {
|
||||
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<String> {
|
||||
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<PgPool>) -> 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<Uuid, DomainError> {
|
||||
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<PgPool>,
|
||||
}
|
||||
|
||||
impl ContentDerivedReferenceSource {
|
||||
pub fn new(pool: Arc<PgPool>) -> 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<String> {
|
||||
content_derived_ref_sql(level, outer_hash_expr)
|
||||
}
|
||||
|
||||
fn ref_exists_sql(&self, level: RefLevel, outer_hash_expr: &str) -> Option<String> {
|
||||
content_derived_exists_sql(level, outer_hash_expr)
|
||||
}
|
||||
|
||||
async fn count_references(&self, blob_hash: &str) -> Result<u64, DomainError> {
|
||||
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<Vec<u8>>,
|
||||
limit: usize,
|
||||
) -> Result<(Vec<String>, Option<Vec<u8>>), 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<String> = 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::*;
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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<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
|
||||
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"#;
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user