feat(transcode): write transcodes to the derived tier, with negative rows
Step 7 of docs/plan/derived-blobs.md, write path first — the plan is explicit that fixing it before the import means transcode_import only has to handle history, not a moving target. ImageTranscodeService now reads and writes storage.content_derived_blobs under kind='transcode', keyed by the BLAKE3 of the SOURCE content. The hash is threaded in from file_retrieval_service, which already holds it as dto.content_hash; hashing here would be a BLAKE3 over the whole file on every request. Callers without one (external mounts) keep the local cache untouched, which is what the service did before this tier existed. Negative verdicts become rows rather than zero-byte .skip files. A transcode that came out larger is deterministic in the content, so it is worth remembering; the row survives moka eviction, a restart, and the deletion of .transcoded/, none of which the marker does. Only that verdict is persisted — a timeout or a read error returns Err and is recorded nowhere, because a momentary failure written here would mark a perfectly transcodable image hopeless with nothing to retry it. Representation is a NULL blob_hash, per the plan: a sentinel hash would stop blob_hash naming a real Blob and every consumer would need to learn the exception. A CHECK keeps blob_hash and content_type NULL together — a type without bytes describes nothing, bytes without a type cannot be served. Two consumers had to be corrected for NULLs first, both of which would have broken on the first negative row ever written: * satellites_consistency reported them as derived_dangling_blob at data_loss severity. SQL comparison against NULL is NULL, so EXISTS was false and a row correctly pointing at nothing read as an artifact that had gone missing. * blob_reference_sources::list_referenced_blobs decodes blob_hash into String, so the first NULL would have failed the decode and taken the whole enumeration down. It would also have been wrong if it decoded — a negative row holds no reference, which is why the counting forms (WHERE blob_hash = <hash>) already exclude it for free. lookup_derived returns a three-way answer because Option collapses the two cases a caller deciding whether to spend a decode most needs apart: never attempted, versus attempted and known not worth it. DedupService is attached after construction via a OnceLock. DI builds the transcode service ~240 lines before DedupService exists, and the retrieval path that needs it is wired earlier still, so a constructor argument would mean reordering more than this is worth. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Edouard Vanbelle
@@ -0,0 +1,63 @@
|
||||
-- Negative rows in `storage.content_derived_blobs`.
|
||||
--
|
||||
-- Some derivations can only be known to be useless by doing the whole
|
||||
-- expensive job. `ImageTranscodeService` learns that WebP is not smaller
|
||||
-- than the original by decoding and re-encoding the whole image; a
|
||||
-- thumbnail renderer learns a source is undecodable, or over the
|
||||
-- 50-megapixel ceiling, only by attempting it. Recomputing that verdict
|
||||
-- on every request is the same cost as computing it the first time.
|
||||
--
|
||||
-- Today those verdicts live in RAM (moka's zero-weight empty-Bytes
|
||||
-- convention) and, for transcodes, as zero-byte `.skip` files on local
|
||||
-- disk. Both vanish: moka evicts, and the local disk is exactly what
|
||||
-- this plan is deleting. So the verdict is stored here, next to the
|
||||
-- positive derivations, as a row whose derived Blob is NULL.
|
||||
--
|
||||
-- ## Why NULL rather than a sentinel hash
|
||||
--
|
||||
-- A reserved hash was considered and rejected. It would stop
|
||||
-- `blob_hash` naming a real Blob, and every consumer — the refcount
|
||||
-- recompute in `manifests_consistency`, `dedup_gc`'s reap predicate,
|
||||
-- `satellites_consistency`'s dangling check — would need to learn the
|
||||
-- exception or silently mis-handle it. NULL is already the SQL way to
|
||||
-- say "no Blob", and those consumers all join on `blob_hash`, so a NULL
|
||||
-- drops out of the join instead of matching something fictional.
|
||||
--
|
||||
-- ## The CHECK matters
|
||||
--
|
||||
-- A row with a `blob_hash` but no `content_type` is unserveable; a row
|
||||
-- with a `content_type` but no `blob_hash` claims a type for bytes that
|
||||
-- do not exist. Both are bugs that would surface far from their cause,
|
||||
-- so the pair moves together or not at all.
|
||||
--
|
||||
-- ## What must NOT become a negative row
|
||||
--
|
||||
-- Only failures that are DETERMINISTIC IN THE CONTENT. A transcode that
|
||||
-- was not smaller, or a source that cannot be decoded, will fail the
|
||||
-- same way forever — those are worth remembering. A generation timeout,
|
||||
-- a closed semaphore, an I/O error reading the source Blob are
|
||||
-- properties of the moment, not the content; persisting one marks a
|
||||
-- perfectly good image as underivable permanently, and nothing ever
|
||||
-- retries it. The asymmetry sets the default: a wrongly-cached
|
||||
-- transient is silent and forever, a missing negative merely costs
|
||||
-- repeated work. When in doubt, do not write the row.
|
||||
|
||||
ALTER TABLE storage.content_derived_blobs
|
||||
ALTER COLUMN blob_hash DROP NOT NULL,
|
||||
ALTER COLUMN content_type DROP NOT NULL;
|
||||
|
||||
ALTER TABLE storage.content_derived_blobs
|
||||
DROP CONSTRAINT IF EXISTS content_derived_blobs_positive_or_negative;
|
||||
|
||||
ALTER TABLE storage.content_derived_blobs
|
||||
ADD CONSTRAINT content_derived_blobs_positive_or_negative
|
||||
CHECK (
|
||||
(blob_hash IS NOT NULL AND content_type IS NOT NULL)
|
||||
OR (blob_hash IS NULL AND content_type IS NULL)
|
||||
);
|
||||
|
||||
COMMENT ON COLUMN storage.content_derived_blobs.blob_hash IS
|
||||
'The derived Blob, or NULL for a NEGATIVE row: the derivation was attempted and is known not to be worth storing (transcode came out larger, source undecodable, source over the decode ceiling). Reference HOLDER when present — bumps chunk_manifests.ref_count via DedupService::add_reference. Only content-deterministic failures may be recorded as negatives; transient ones (timeout, semaphore, I/O) must not, or a momentary failure becomes permanent.';
|
||||
|
||||
COMMENT ON COLUMN storage.content_derived_blobs.content_type IS
|
||||
'MIME type of the derived Blob. NULL exactly when blob_hash is NULL — the CHECK keeps the pair together, since a type without bytes describes nothing and bytes without a type cannot be served.';
|
||||
@@ -34,6 +34,29 @@ pub struct DerivedBlobRef {
|
||||
pub content_type: String,
|
||||
}
|
||||
|
||||
/// What the derived tier knows about one `(source_hash, kind, variant)`.
|
||||
///
|
||||
/// Three answers, not two. `Option<DerivedBlobRef>` could only say
|
||||
/// "have it" or "don't", which collapses the two cases that matter most
|
||||
/// to a caller deciding whether to spend a decode:
|
||||
///
|
||||
/// * [`Missing`](Self::Missing) — never attempted. Derive it.
|
||||
/// * [`NotDerivable`](Self::NotDerivable) — attempted, and the attempt
|
||||
/// is known to be a waste for this content: the transcode came out
|
||||
/// larger than the original, the source cannot be decoded, the source
|
||||
/// is over the decode ceiling. Serve the original and do not retry.
|
||||
/// * [`Found`](Self::Found) — here are the bytes.
|
||||
///
|
||||
/// Only failures that are deterministic in the CONTENT may be recorded
|
||||
/// as `NotDerivable`. A timeout or an I/O error is a property of the
|
||||
/// moment; persisting one would mark a good image underivable forever.
|
||||
#[derive(Debug, Clone)]
|
||||
pub enum DerivedLookup {
|
||||
Missing,
|
||||
NotDerivable,
|
||||
Found(DerivedBlobRef),
|
||||
}
|
||||
|
||||
/// Result of a deduplication store operation.
|
||||
#[derive(Debug, Clone)]
|
||||
pub enum DedupResultDto {
|
||||
|
||||
@@ -85,9 +85,15 @@ pub trait ImageTranscodePort: Send + Sync + 'static {
|
||||
/// Returns `(content, mime_type, was_transcoded)`.
|
||||
/// If transcoding is not beneficial (output larger than input), returns the
|
||||
/// original content with `was_transcoded = false`.
|
||||
///
|
||||
/// `source_hash` is the BLAKE3 of the original content, which is how the
|
||||
/// durable derived tier is keyed. `None` restricts the implementation to
|
||||
/// its local cache — correct for callers with no hash (external mounts),
|
||||
/// and the behaviour of every caller before that tier existed.
|
||||
async fn get_transcoded(
|
||||
&self,
|
||||
file_id: &str,
|
||||
source_hash: Option<&str>,
|
||||
original_content: Bytes,
|
||||
original_mime: &str,
|
||||
target_format: OutputFormat,
|
||||
|
||||
@@ -274,9 +274,15 @@ impl FileRetrievalService {
|
||||
}
|
||||
|
||||
/// Try to transcode image content to WebP and return transcoded variant.
|
||||
///
|
||||
/// `source_hash` keys the durable derived tier. The caller has it as
|
||||
/// `dto.content_hash`; passing it rather than hashing here matters,
|
||||
/// because hashing would be a BLAKE3 over the whole file on every
|
||||
/// request that reaches this path.
|
||||
async fn try_transcode(
|
||||
&self,
|
||||
id: &str,
|
||||
source_hash: Option<&str>,
|
||||
content: &Bytes,
|
||||
mime: &str,
|
||||
file_size: u64,
|
||||
@@ -291,7 +297,7 @@ impl FileRetrievalService {
|
||||
}
|
||||
let format = OutputFormat::WebP;
|
||||
match transcode
|
||||
.get_transcoded(id, content.clone(), mime, format)
|
||||
.get_transcoded(id, source_hash, content.clone(), mime, format)
|
||||
.await
|
||||
{
|
||||
Ok((transcoded, webp_mime, true)) => {
|
||||
@@ -364,7 +370,16 @@ impl FileRetrievalService {
|
||||
|
||||
if do_transcode
|
||||
&& let Some((t, m)) = self
|
||||
.try_transcode(id, &content_bytes, &mime_type, file_size, true)
|
||||
.try_transcode(
|
||||
id,
|
||||
// Empty on the hash-less stub DTOs (external mounts),
|
||||
// which the content-keyed tier cannot serve anyway.
|
||||
Some(&*dto.content_hash).filter(|h: &&str| !h.is_empty()),
|
||||
&content_bytes,
|
||||
&mime_type,
|
||||
file_size,
|
||||
true,
|
||||
)
|
||||
.await
|
||||
{
|
||||
return Ok((
|
||||
|
||||
@@ -451,6 +451,17 @@ impl AppServiceFactory {
|
||||
);
|
||||
dedup_service.initialize().await?;
|
||||
|
||||
// Hand the transcode service its derived tier.
|
||||
//
|
||||
// Deferred rather than injected at construction because that happens
|
||||
// ~240 lines above this, before `DedupService` exists, and the
|
||||
// retrieval path that needs the transcode service is wired earlier
|
||||
// still. Reordering DI to make the dependency a constructor argument
|
||||
// would move more than it is worth; the service treats a missing
|
||||
// handle as "local cache only", which is exactly its pre-derived-tier
|
||||
// behaviour.
|
||||
image_transcode_service.attach_dedup(dedup_service.clone());
|
||||
|
||||
// One-time background migration: re-chunk pre-CDC whole-file blobs
|
||||
// into chunk manifests so Range reads (and, with encryption, partial
|
||||
// decrypts) stop paying for the entire blob. No-op once converged.
|
||||
|
||||
@@ -392,6 +392,15 @@ impl BlobReferenceSource for ContentDerivedReferenceSource {
|
||||
// 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.
|
||||
//
|
||||
// `IS NOT NULL` is load-bearing, not defensive. A NEGATIVE row —
|
||||
// "this content is not worth transcoding" — carries a NULL
|
||||
// blob_hash, and this query decodes into `String`, so the first one
|
||||
// ever written would fail the decode and take the whole enumeration
|
||||
// down. It would also be wrong if it decoded: a negative row holds
|
||||
// no reference on any Blob, which is exactly why the counting forms
|
||||
// above (`WHERE blob_hash = <hash>`) already exclude it for free —
|
||||
// NULL equals nothing.
|
||||
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}"))
|
||||
@@ -401,7 +410,8 @@ impl BlobReferenceSource for ContentDerivedReferenceSource {
|
||||
|
||||
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)
|
||||
WHERE blob_hash IS NOT NULL
|
||||
AND ($1::text IS NULL OR blob_hash > $1)
|
||||
ORDER BY blob_hash
|
||||
LIMIT $2",
|
||||
)
|
||||
|
||||
@@ -778,7 +778,33 @@ impl DedupService {
|
||||
kind: &str,
|
||||
variant: &str,
|
||||
) -> Option<crate::application::ports::dedup_ports::DerivedBlobRef> {
|
||||
sqlx::query_as::<_, (String, String)>(
|
||||
match self.lookup_derived(source_hash, kind, variant).await {
|
||||
crate::application::ports::dedup_ports::DerivedLookup::Found(r) => Some(r),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Full three-way answer: no row, a negative verdict, or the blob.
|
||||
///
|
||||
/// Callers deciding whether to spend a decode want the middle case,
|
||||
/// which [`Self::find_derived_blob`] cannot express — it folds
|
||||
/// "never attempted" and "attempted, not worth it" into the same
|
||||
/// `None`, and a caller acting on that repeats the expensive work
|
||||
/// forever. Use this wherever the derivation is costly; use
|
||||
/// `find_derived_blob` when you only need the bytes.
|
||||
///
|
||||
/// A query error reads as `Missing`, deliberately: a database blip
|
||||
/// should cost a redundant render, never a wrong "not derivable"
|
||||
/// that suppresses a derivation the content can support.
|
||||
pub async fn lookup_derived(
|
||||
&self,
|
||||
source_hash: &str,
|
||||
kind: &str,
|
||||
variant: &str,
|
||||
) -> crate::application::ports::dedup_ports::DerivedLookup {
|
||||
use crate::application::ports::dedup_ports::{DerivedBlobRef, DerivedLookup};
|
||||
|
||||
let row = sqlx::query_as::<_, (Option<String>, Option<String>)>(
|
||||
"SELECT blob_hash, content_type FROM storage.content_derived_blobs
|
||||
WHERE source_hash = $1 AND kind = $2 AND variant = $3",
|
||||
)
|
||||
@@ -788,13 +814,59 @@ impl DedupService {
|
||||
.fetch_optional(self.pool.as_ref())
|
||||
.await
|
||||
.ok()
|
||||
.flatten()
|
||||
.map(|(blob_hash, content_type)| {
|
||||
crate::application::ports::dedup_ports::DerivedBlobRef {
|
||||
.flatten();
|
||||
|
||||
match row {
|
||||
None => DerivedLookup::Missing,
|
||||
// The CHECK constraint keeps blob_hash and content_type NULL
|
||||
// together, so one NULL is the whole negative row.
|
||||
Some((None, _)) | Some((_, None)) => DerivedLookup::NotDerivable,
|
||||
Some((Some(blob_hash), Some(content_type))) => DerivedLookup::Found(DerivedBlobRef {
|
||||
blob_hash,
|
||||
content_type,
|
||||
}),
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
/// Record that this derivation is not worth attempting again.
|
||||
///
|
||||
/// For outcomes that are deterministic in the source content — a
|
||||
/// transcode that came out larger, a source that will not decode, a
|
||||
/// source over the decode ceiling. **Never** for a timeout, a closed
|
||||
/// semaphore, or an I/O error: those are properties of the moment,
|
||||
/// and a row written for one marks good content underivable forever
|
||||
/// with nothing to retry it.
|
||||
///
|
||||
/// Takes no reference on any Blob — there is no derived Blob to hold
|
||||
/// one. The row is dependent on its source and is reaped with it,
|
||||
/// same as a positive row.
|
||||
///
|
||||
/// Guarded by the same source-exists check as `store_derived_blob`:
|
||||
/// a row whose source has already been reaped is a permanent leak of
|
||||
/// a mapping nothing will ever clean up.
|
||||
pub async fn store_derived_negative(
|
||||
&self,
|
||||
source_hash: &str,
|
||||
kind: &str,
|
||||
variant: &str,
|
||||
) -> Result<(), DomainError> {
|
||||
sqlx::query(
|
||||
"INSERT INTO storage.content_derived_blobs
|
||||
(source_hash, kind, variant, blob_hash, content_type)
|
||||
SELECT $1, $2, $3, NULL, NULL
|
||||
WHERE EXISTS (SELECT 1 FROM storage.chunk_manifests WHERE file_hash = $1)
|
||||
OR EXISTS (SELECT 1 FROM storage.blobs WHERE hash = $1)
|
||||
ON CONFLICT (source_hash, kind, variant) DO NOTHING",
|
||||
)
|
||||
.bind(source_hash)
|
||||
.bind(kind)
|
||||
.bind(variant)
|
||||
.execute(self.pool.as_ref())
|
||||
.await
|
||||
.map_err(|e| {
|
||||
DomainError::internal_error("Dedup", format!("store_derived_negative: {e}"))
|
||||
})?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// The registry backing the reap predicate.
|
||||
|
||||
@@ -147,6 +147,20 @@ pub struct ImageTranscodeService {
|
||||
memory_cache: moka::future::Cache<String, Bytes>,
|
||||
/// Lock-free statistics
|
||||
stats: Arc<AtomicTranscodeStats>,
|
||||
/// The derived tier, attached after construction.
|
||||
///
|
||||
/// A constructor parameter would be cleaner but does not fit: DI builds
|
||||
/// this service before `DedupService` exists, and reordering is worse
|
||||
/// than a one-shot — the transcode service is needed by the retrieval
|
||||
/// path, which is wired early. `ThumbnailService` met the same wall and
|
||||
/// took a per-call parameter instead; that does not work here because
|
||||
/// the caller (`FileRetrievalService`) holds no dedup handle either, so
|
||||
/// threading one through would push the dependency into a service that
|
||||
/// has no other use for it.
|
||||
///
|
||||
/// `OnceLock` rather than a `Mutex`: set exactly once at boot, read on
|
||||
/// every request, never replaced.
|
||||
dedup: OnceLock<Arc<crate::infrastructure::services::dedup_service::DedupService>>,
|
||||
}
|
||||
|
||||
impl ImageTranscodeService {
|
||||
@@ -176,10 +190,37 @@ impl ImageTranscodeService {
|
||||
cache_dir,
|
||||
memory_cache,
|
||||
stats: Arc::new(AtomicTranscodeStats::default()),
|
||||
dedup: OnceLock::new(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Initialize the service (create cache directories)
|
||||
/// Attach the derived tier. Called once from DI, after `DedupService`
|
||||
/// exists. Until then — and in tests that never call it — the service
|
||||
/// behaves exactly as before, reading and writing only its local cache.
|
||||
pub fn attach_dedup(
|
||||
&self,
|
||||
dedup: Arc<crate::infrastructure::services::dedup_service::DedupService>,
|
||||
) {
|
||||
if self.dedup.set(dedup).is_err() {
|
||||
tracing::warn!(
|
||||
target: "oxicloud::transcode",
|
||||
"attach_dedup called twice — the first handle is kept"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// The `content_derived_blobs.kind` for everything this service writes.
|
||||
const DERIVED_KIND: &'static str = "transcode";
|
||||
|
||||
/// Initialize the service.
|
||||
///
|
||||
/// Still creates the local cache directories, because this service DOES
|
||||
/// still write them — unlike `ThumbnailService`, whose sidecar writes
|
||||
/// are gone. When the transcode write path moves fully to the derived
|
||||
/// tier, these two `create_dir_all` calls have to go at the same time:
|
||||
/// leaving them would recreate the tree on every boot and make the
|
||||
/// absence that `transcode_import` works toward unreachable, which is
|
||||
/// exactly the bug that kept `.thumbnails/` alive across restarts.
|
||||
pub async fn initialize(&self) -> std::io::Result<()> {
|
||||
fs::create_dir_all(&self.cache_dir).await?;
|
||||
fs::create_dir_all(self.cache_dir.join("webp")).await?;
|
||||
@@ -213,13 +254,28 @@ impl ImageTranscodeService {
|
||||
///
|
||||
/// Accepts `Bytes` (ref-counted) so callers avoid copying the buffer.
|
||||
/// Cloning `Bytes` is O(1) — only an atomic increment.
|
||||
///
|
||||
/// `source_hash` is the BLAKE3 of the ORIGINAL content — the key the
|
||||
/// derived tier uses. `None` falls back to the local cache alone, which
|
||||
/// is what happens for callers that have no hash (external mounts) and
|
||||
/// what the whole service did before the derived tier existed.
|
||||
///
|
||||
/// It is a parameter rather than something computed here on purpose:
|
||||
/// hashing `original_content` per request would be a BLAKE3 over the
|
||||
/// whole file on every GET, and the caller already has the value.
|
||||
pub async fn get_transcoded(
|
||||
&self,
|
||||
file_id: &str,
|
||||
source_hash: Option<&str>,
|
||||
original_content: Bytes,
|
||||
original_mime: &str,
|
||||
target_format: OutputFormat,
|
||||
) -> Result<(Bytes, String, bool), String> {
|
||||
// Keyed by file_id, not content hash. Deliberate: this cache is
|
||||
// per-request-path and short-lived (10 min TTL), and the file id is
|
||||
// what the caller has in hand on every request. The DURABLE tier
|
||||
// below is content-keyed, which is where dedup across identical
|
||||
// files actually pays.
|
||||
let cache_key = format!("{}:{}", file_id, target_format.extension());
|
||||
|
||||
// ── 1. Fast path: moka memory cache (lock-free read) ──
|
||||
@@ -248,7 +304,13 @@ impl ImageTranscodeService {
|
||||
let cached = self
|
||||
.memory_cache
|
||||
.try_get_with(cache_key, async {
|
||||
self.compute_transcode(file_id, original_for_loader, original_mime, target_format)
|
||||
self.compute_transcode(
|
||||
file_id,
|
||||
source_hash,
|
||||
original_for_loader,
|
||||
original_mime,
|
||||
target_format,
|
||||
)
|
||||
.await
|
||||
})
|
||||
.await
|
||||
@@ -271,11 +333,62 @@ impl ImageTranscodeService {
|
||||
async fn compute_transcode(
|
||||
&self,
|
||||
file_id: &str,
|
||||
source_hash: Option<&str>,
|
||||
original_content: Bytes,
|
||||
original_mime: &str,
|
||||
target_format: OutputFormat,
|
||||
) -> Result<Bytes, String> {
|
||||
// ── Disk cache (async fs) ──
|
||||
// ── Derived tier, ahead of the local cache ──
|
||||
//
|
||||
// Content-keyed, so it is shared across every file with these bytes
|
||||
// and survives both a restart and a backend migration — neither of
|
||||
// which the local `.transcoded/` tree does. Read first for the same
|
||||
// reason the thumbnail read-order flip put it first: the local tree
|
||||
// is the legacy tier being drained, and a fallback that is consulted
|
||||
// first never stops being load-bearing.
|
||||
let derived = match (source_hash, self.dedup.get()) {
|
||||
(Some(hash), Some(dedup)) => Some((hash, dedup)),
|
||||
_ => None,
|
||||
};
|
||||
if let Some((hash, dedup)) = derived {
|
||||
use crate::application::ports::dedup_ports::DerivedLookup;
|
||||
match dedup
|
||||
.lookup_derived(hash, Self::DERIVED_KIND, target_format.extension())
|
||||
.await
|
||||
{
|
||||
DerivedLookup::Found(r) => match dedup.read_blob_bytes(&r.blob_hash).await {
|
||||
Ok(bytes) if !bytes.is_empty() => {
|
||||
self.stats.disk_hits.fetch_add(1, Ordering::Relaxed);
|
||||
tracing::debug!("🧱 Transcode derived tier HIT: {}", file_id);
|
||||
return Ok(bytes);
|
||||
}
|
||||
// The row promised bytes that are gone or empty. Fall
|
||||
// through and re-derive rather than serving nothing —
|
||||
// a transcode is a pure function of its source, so this
|
||||
// is recoverable by construction. `satellites_consistency`
|
||||
// reports the dangling row separately.
|
||||
_ => tracing::warn!(
|
||||
target: "oxicloud::transcode",
|
||||
source_hash = %hash,
|
||||
blob_hash = %r.blob_hash,
|
||||
"derived transcode row points at unreadable bytes; re-deriving"
|
||||
),
|
||||
},
|
||||
// Known not worth transcoding for this content. This is the
|
||||
// whole point of persisting the verdict: without it every GET
|
||||
// repeats a full decode + encode to throw the result away.
|
||||
DerivedLookup::NotDerivable => {
|
||||
self.stats.disk_hits.fetch_add(1, Ordering::Relaxed);
|
||||
tracing::debug!("🧱 Transcode negative derived row HIT: {}", file_id);
|
||||
return Ok(Bytes::new());
|
||||
}
|
||||
DerivedLookup::Missing => {}
|
||||
}
|
||||
}
|
||||
|
||||
// ── Legacy local cache (async fs) ──
|
||||
//
|
||||
// Drained by `transcode_import`; kept as a fallback until it is gone.
|
||||
let cache_path = self.get_cache_path(file_id, target_format);
|
||||
if tokio::fs::try_exists(&cache_path).await.unwrap_or(false) {
|
||||
match fs::read(&cache_path).await {
|
||||
@@ -326,10 +439,38 @@ impl ImageTranscodeService {
|
||||
original_size,
|
||||
transcoded_size
|
||||
);
|
||||
// Remember the negative verdict so the next GET doesn't repeat the
|
||||
// decode + encode: the caller caches the empty-Bytes sentinel (TTL)
|
||||
// and we drop a zero-byte marker on disk (survives restarts;
|
||||
// removed by `invalidate` when the file changes).
|
||||
// Remember the verdict so the next GET does not repeat the decode
|
||||
// + encode. The caller caches the empty-Bytes sentinel in RAM
|
||||
// (10 min TTL); this row is what makes it survive eviction, a
|
||||
// restart, and a move to another instance.
|
||||
//
|
||||
// Safe to persist because it is deterministic in the CONTENT:
|
||||
// these exact bytes will always re-encode larger. A timeout or a
|
||||
// read error would not be — those return `Err` above and are
|
||||
// deliberately not recorded, since a momentary failure written
|
||||
// here would mark a perfectly transcodable image as hopeless
|
||||
// with nothing to ever retry it.
|
||||
match derived {
|
||||
Some((hash, dedup)) => {
|
||||
if let Err(e) = dedup
|
||||
.store_derived_negative(hash, Self::DERIVED_KIND, target_format.extension())
|
||||
.await
|
||||
{
|
||||
tracing::warn!(
|
||||
target: "oxicloud::transcode",
|
||||
source_hash = %hash,
|
||||
error = %e,
|
||||
"failed to persist negative transcode verdict; it will be recomputed"
|
||||
);
|
||||
}
|
||||
}
|
||||
// Hash-less callers still get the zero-byte marker, for the
|
||||
// same reason they still get the local cache write: the
|
||||
// content-keyed tier cannot hold a verdict for content it
|
||||
// cannot name. Dropping this would make every external-mount
|
||||
// GET of a non-shrinking image re-decode once moka's TTL
|
||||
// expires.
|
||||
None => {
|
||||
let marker = self.get_skip_marker_path(file_id, target_format);
|
||||
tokio::spawn(async move {
|
||||
if let Some(parent) = marker.parent() {
|
||||
@@ -339,12 +480,50 @@ impl ImageTranscodeService {
|
||||
tracing::warn!("Failed to persist transcode skip marker: {}", e);
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
return Ok(Bytes::new());
|
||||
}
|
||||
|
||||
let saved = original_size - transcoded_size;
|
||||
|
||||
// ── Persist to disk cache (fire-and-forget) ──
|
||||
// ── Persist ──
|
||||
//
|
||||
// Derived tier when we have a source hash, local cache otherwise.
|
||||
// Not both: writing the sidecar too would mean `transcode_import`
|
||||
// chases a tail that keeps being refilled, which is the trap the
|
||||
// thumbnail migration hit — four render paths wrote the sidecar and
|
||||
// one wrote the row, so the tail never emptied.
|
||||
//
|
||||
// The local write survives only for hash-less callers (external
|
||||
// mounts), which the derived tier cannot serve at all. When those
|
||||
// gain a hash this branch goes, and `initialize`'s `create_dir_all`
|
||||
// calls go with it.
|
||||
match derived {
|
||||
Some((hash, dedup)) => {
|
||||
let dedup = dedup.clone();
|
||||
let hash = hash.to_string();
|
||||
let variant = target_format.extension().to_string();
|
||||
let mime = target_format.mime_type().to_string();
|
||||
let bytes = transcoded_bytes.clone();
|
||||
// Fire-and-forget, as the disk write was: the bytes are
|
||||
// already on their way to the client, and a storage hiccup
|
||||
// should cost a re-derive later rather than this response.
|
||||
tokio::spawn(async move {
|
||||
if let Err(e) = dedup
|
||||
.store_derived_blob(&hash, Self::DERIVED_KIND, &variant, &mime, bytes)
|
||||
.await
|
||||
{
|
||||
tracing::warn!(
|
||||
target: "oxicloud::transcode",
|
||||
source_hash = %hash,
|
||||
error = %e,
|
||||
"failed to store derived transcode; it will be recomputed"
|
||||
);
|
||||
}
|
||||
});
|
||||
}
|
||||
None => {
|
||||
let cache_path_clone = cache_path.clone();
|
||||
let transcoded_for_disk = transcoded_bytes.clone();
|
||||
tokio::spawn(async move {
|
||||
@@ -355,6 +534,8 @@ impl ImageTranscodeService {
|
||||
tracing::warn!("Failed to cache transcoded image: {}", e);
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
// ── Update stats (lock-free atomics) ──
|
||||
self.stats.transcodes.fetch_add(1, Ordering::Relaxed);
|
||||
@@ -476,12 +657,14 @@ impl ImageTranscodePort for ImageTranscodeService {
|
||||
async fn get_transcoded(
|
||||
&self,
|
||||
file_id: &str,
|
||||
source_hash: Option<&str>,
|
||||
original_content: Bytes,
|
||||
original_mime: &str,
|
||||
target_format: PortOutputFormat,
|
||||
) -> Result<(Bytes, String, bool), DomainError> {
|
||||
self.get_transcoded(
|
||||
file_id,
|
||||
source_hash,
|
||||
original_content,
|
||||
original_mime,
|
||||
target_format.into(),
|
||||
|
||||
@@ -92,7 +92,11 @@ struct DerivedRow {
|
||||
source_hash: String,
|
||||
kind: String,
|
||||
variant: String,
|
||||
blob_hash: String,
|
||||
/// `None` on a NEGATIVE row — the derivation was attempted and is
|
||||
/// known not to be worth storing for this content (a transcode that
|
||||
/// came out larger, an undecodable source). Those rows point at
|
||||
/// nothing on purpose and must not be read as dangling.
|
||||
blob_hash: Option<String>,
|
||||
source_exists: bool,
|
||||
artifact_exists: bool,
|
||||
}
|
||||
@@ -132,8 +136,16 @@ impl SatellitesConsistencyCheck {
|
||||
"SELECT d.source_hash, d.kind, d.variant, d.blob_hash, ",
|
||||
blob_exists!("d.source_hash"),
|
||||
" AS source_exists, ",
|
||||
// A NEGATIVE row (NULL blob_hash) has no artifact BY DESIGN, so it
|
||||
// counts as satisfied. Without this it reads as dangling: SQL
|
||||
// comparison against NULL is NULL, so `EXISTS` is false, and every
|
||||
// "this content is not worth transcoding" verdict would be reported
|
||||
// as `data_loss`. The check has to be here rather than in the Rust
|
||||
// arm below, so the column means "this row is in the state it
|
||||
// should be" for both row shapes.
|
||||
"(d.blob_hash IS NULL OR ",
|
||||
blob_exists!("d.blob_hash"),
|
||||
" AS artifact_exists
|
||||
") AS artifact_exists
|
||||
FROM storage.content_derived_blobs d
|
||||
WHERE ($1::text IS NULL
|
||||
OR (d.source_hash, d.kind, d.variant) > ($1::text, $2::text, $3::text))
|
||||
|
||||
Reference in New Issue
Block a user