diff --git a/docs/config/env.md b/docs/config/env.md index 7ddee4f7..0d1e02a8 100644 --- a/docs/config/env.md +++ b/docs/config/env.md @@ -18,7 +18,7 @@ Most runtime variables use the `OXICLOUD_` prefix. A few build-time or allocator | `OXICLOUD_CHUNK_DIR` | `{STORAGE_PATH}/.uploads` | Root directory for chunked-upload sessions (REST + NextCloud). Direct (non-chunked) uploads stream straight into the blob store and need no spool directory. Placement guidance: see [Storage Fine Tuning](./storage-fine-tuning.md). | | `OXICLOUD_REUSE_PORT` | `false` | Enable `SO_REUSEPORT` so multiple processes can share the same port. **Disabled by default** — a second accidental instance will fail with "address already in use". Enable only for deliberate multi-worker setups (process supervisor, rolling restart). Not supported on Windows. | | `OXICLOUD_METRICS_LISTEN` | (unset) | Prometheus `/metrics` listener address (e.g. `127.0.0.1:9090`, IPv6 allowed as `[::1]:9090`). **Unset = disabled**: no `/metrics` endpoint is bound and no metrics recorder is installed (zero runtime cost). When set, a separate HTTP listener on this address serves the text-format scrape. **Deliberately NOT merged into the main API** — no auth, CSRF, or DPoP layer in front. Bind to loopback or a private interface unless you intend to expose metrics publicly. Starter counters: `oxicloud_dpop_verify_failed_total{reason}`, `oxicloud_dpop_proof_missing_total`, `oxicloud_dpop_header_missing_on_bound_session_total`, `oxicloud_dpop_replay_detected_total`, `oxicloud_dpop_nonce_challenges_issued_total`. | -| `OXICLOUD_STARTUP_JOBS` | `thumb_derived_import?repair=true,thumb_attached_import?repair=true` | Background jobs dispatched once at boot, comma-separated, each `name` or `name?flag=true` using the same syntax as `POST /api/admin/jobs/{name}/trigger`. Flags: `force`, `deep`, `repair`, `storage`. **The default migrates thumbnails out of the legacy `.thumbnails/` directory and deletes the originals**, so the migration completes without anyone triggering it from the admin panel; each sidecar is read back through the normal stack before it is unlinked, and every deletion is audited. An explicit value **replaces** the default; set it empty (`OXICLOUD_STARTUP_JOBS=`) to disable startup jobs, or to `thumb_derived_import,thumb_attached_import` to import without deleting. **Non-blocking** — readiness never waits on a job; entries run sequentially in the background. **Fail-fast** — an unknown job name or flag panics at boot, because a silently-dropped entry means a migration that never runs. A run interrupted by a restart resumes from its cursor on the next boot, so a long migration finishes across restarts. Safe to leave at the default: the jobs are idempotent, and once drained a run does nothing. See [Thumbnail Migration](./thumbnail-migration.md) for the upgrade runbook. | +| `OXICLOUD_STARTUP_JOBS` | `thumb_derived_import?repair=true,thumb_attached_import?repair=true,transcode_import?repair=true` | Background jobs dispatched once at boot, comma-separated, each `name` or `name?flag=true` using the same syntax as `POST /api/admin/jobs/{name}/trigger`. Flags: `force`, `deep`, `repair`, `storage`. **The default migrates thumbnails out of the legacy `.thumbnails/` directory and deletes the originals**, so the migration completes without anyone triggering it from the admin panel; each sidecar is read back through the normal stack before it is unlinked, and every deletion is audited. An explicit value **replaces** the default; set it empty (`OXICLOUD_STARTUP_JOBS=`) to disable startup jobs, or to `thumb_derived_import,thumb_attached_import` to import without deleting. **Non-blocking** — readiness never waits on a job; entries run sequentially in the background. **Fail-fast** — an unknown job name or flag panics at boot, because a silently-dropped entry means a migration that never runs. A run interrupted by a restart resumes from its cursor on the next boot, so a long migration finishes across restarts. Safe to leave at the default: the jobs are idempotent, and once drained a run does nothing. See [Thumbnail Migration](./thumbnail-migration.md) for the upgrade runbook. | ## Database diff --git a/docs/plan/derived-blobs.md b/docs/plan/derived-blobs.md index d3133f94..705be5d3 100644 --- a/docs/plan/derived-blobs.md +++ b/docs/plan/derived-blobs.md @@ -1409,7 +1409,25 @@ hardcoded SQL). New sources bolt on independently. after, step 5 lands at scale; per-row HEADs do not survive a 4× row count. 7. **`ImageTranscodeService`** — same shape, `kind = 'transcode'`, - no new table. **Scoped 2026-08-27, not started.** The service does + no new table. **Done 2026-08-30**, validated on a snapshot restore + against S3: 13 cached transcodes imported, 5 `.skip` markers + collapsing to 3 negative rows (three of them named the same content), + `file_gone` exercised with a synthetic id, and every deletion + preceded by a byte-for-byte readback. + + Two things landed beyond the three pieces below. The memory cache is + now keyed by content as well, not just the durable tier — it was + still `{file_id}:{ext}`, so identical content held two RAM entries + and the second file missed. And `run_or_resume` binds a run to the + flags it started with, so a paused `?repair=true` import can no + longer resume as import-only. + + Still local-disk, deliberately: hash-less callers (external mounts) + have no content identity, so they keep `.transcoded/` and it will + not disappear on installs that have them. Absence is only expected + elsewhere. + + *Original scoping, for the record.* The service does not currently write the derived tier at all, which is the same gap `persist_rendered` closed for thumbnails, and it must be closed before `transcode_import` can converge. diff --git a/docs/plan/job-registry.md b/docs/plan/job-registry.md index b9845953..dda8a884 100644 --- a/docs/plan/job-registry.md +++ b/docs/plan/job-registry.md @@ -746,7 +746,7 @@ optionally with the same query syntax the admin trigger URL uses. **The default is both migration jobs, in repair mode:** ``` -OXICLOUD_STARTUP_JOBS=thumb_derived_import?repair=true,thumb_attached_import?repair=true +OXICLOUD_STARTUP_JOBS=thumb_derived_import?repair=true,thumb_attached_import?repair=true,transcode_import?repair=true ``` An explicit value replaces that list; an empty value disables startup diff --git a/example.env b/example.env index 9fa80904..69de5230 100644 --- a/example.env +++ b/example.env @@ -62,7 +62,7 @@ OXICLOUD_SERVER_HOST=0.0.0.0 # storage). # # DEFAULT (applied when this variable is unset): -# thumb_derived_import?repair=true,thumb_attached_import?repair=true +# thumb_derived_import?repair=true,thumb_attached_import?repair=true,transcode_import?repair=true # # Those two migrate thumbnails out of the legacy .thumbnails/ directory # into blob storage and then delete the originals, so the migration diff --git a/migrations/20261023000000_content_derived_blobs_negative_rows.sql b/migrations/20261023000000_content_derived_blobs_negative_rows.sql new file mode 100644 index 00000000..b119ade7 --- /dev/null +++ b/migrations/20261023000000_content_derived_blobs_negative_rows.sql @@ -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.'; diff --git a/src/application/ports/dedup_ports.rs b/src/application/ports/dedup_ports.rs index 046ca45c..21910c8a 100644 --- a/src/application/ports/dedup_ports.rs +++ b/src/application/ports/dedup_ports.rs @@ -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` 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 { diff --git a/src/application/ports/transcode_ports.rs b/src/application/ports/transcode_ports.rs index 1e7ed3b9..f727e8b0 100644 --- a/src/application/ports/transcode_ports.rs +++ b/src/application/ports/transcode_ports.rs @@ -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, diff --git a/src/application/services/file_retrieval_service.rs b/src/application/services/file_retrieval_service.rs index ac253662..6bd37ea9 100644 --- a/src/application/services/file_retrieval_service.rs +++ b/src/application/services/file_retrieval_service.rs @@ -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(( diff --git a/src/common/config.rs b/src/common/config.rs index 31f4ac94..67001e03 100644 --- a/src/common/config.rs +++ b/src/common/config.rs @@ -2401,8 +2401,15 @@ fn parse_startup_job(raw: &str) -> Result { /// Set `OXICLOUD_STARTUP_JOBS=` (empty) to disable startup jobs /// entirely; any explicit value replaces this list rather than adding /// to it. -const DEFAULT_STARTUP_JOBS: &str = - "thumb_derived_import?repair=true,thumb_attached_import?repair=true"; +/// `transcode_import` joins them for the same reason and on the same +/// terms. Its artifacts are the most disposable of the three — a +/// transcode is a pure function of its source, so anything deleted in +/// error is recomputed on the next request — and its `.skip` markers +/// collapse to one row per distinct content, which is the saving that +/// only happens once the import runs. +const DEFAULT_STARTUP_JOBS: &str = "thumb_derived_import?repair=true,\ + thumb_attached_import?repair=true,\ + transcode_import?repair=true"; /// Parse the whole `OXICLOUD_STARTUP_JOBS` value. Empty → no startup /// jobs (an explicit opt-out); unset → [`DEFAULT_STARTUP_JOBS`]. @@ -3999,7 +4006,14 @@ mod tests { fn default_startup_jobs_drain_both_thumbnail_tiers() { let jobs = AppConfig::default().startup_jobs; let names: Vec<&str> = jobs.iter().map(|j| j.name.as_str()).collect(); - assert_eq!(names, ["thumb_derived_import", "thumb_attached_import"]); + assert_eq!( + names, + [ + "thumb_derived_import", + "thumb_attached_import", + "transcode_import" + ] + ); assert!(jobs.iter().all(|j| j.args.repair)); assert!(jobs.iter().all(|j| !j.args.deep && !j.args.force)); } diff --git a/src/common/di.rs b/src/common/di.rs index f6ea7345..8daa9324 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -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. @@ -1494,6 +1505,24 @@ impl AppServiceFactory { .register_recoverable_job(&core.job_registry, &job_store_provider_dyn) .await; + // Step 7 migration tenant: drains `.transcoded/` the same way the + // thumbnail imports drain `.thumbnails/`, with one extra step — + // the legacy tree is keyed by FILE (`{file_id}.webp`) while the + // destination is keyed by CONTENT, so every entry is re-keyed + // through `storage.files` on the way in. Entries naming the same + // content collapse into one row, which is the saving this migration + // exists for; a sandbox with five `.skip` markers had three of them + // describing one image. + let _ = Arc::new( + crate::infrastructure::services::transcode_import_service::TranscodeImport::new( + std::path::Path::new(&self.storage_path).join(".transcoded"), + core.dedup_service.clone(), + maintenance_pool.clone(), + ), + ) + .register_recoverable_job(&core.job_registry, &job_store_provider_dyn) + .await; + // Both satellite tables, checked for mappings whose Blob is gone. // Nothing else can: a row whose SOURCE was reaped still holds a valid // reference to a real artifact with a correct refcount, so every diff --git a/src/infrastructure/repositories/pg/blob_reference_sources.rs b/src/infrastructure/repositories/pg/blob_reference_sources.rs index d273d2b2..e9534c82 100644 --- a/src/infrastructure/repositories/pg/blob_reference_sources.rs +++ b/src/infrastructure/repositories/pg/blob_reference_sources.rs @@ -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 = `) already exclude it for free — + // NULL equals nothing. let after: Option = 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", ) diff --git a/src/infrastructure/scheduler/recoverable.rs b/src/infrastructure/scheduler/recoverable.rs index e5546682..99df9b55 100644 --- a/src/infrastructure/scheduler/recoverable.rs +++ b/src/infrastructure/scheduler/recoverable.rs @@ -199,6 +199,63 @@ impl RunOutcome { } } +/// Write `JobRunArgs` to `params` on a Fresh run, or read them back on a +/// Resumed one. +/// +/// Returns the args the handler should actually use. On resume that is +/// whatever the original run recorded, NOT what the resuming caller +/// passed — see the call site in [`run_or_resume`] for why changing mode +/// mid-run is refused. +/// +/// Every flag is stored as a string, matching the `params` convention the +/// progress fields already use, and each is read back independently: a run +/// paused before this existed simply has no keys, and each missing one +/// falls back to `false` / `None`. That is the safe direction — a resumed +/// legacy run under-acts rather than deleting under a flag nobody gave it. +async fn persist_or_restore_args( + store: &dyn JobStore, + args: &JobRunArgs, + is_fresh: bool, +) -> Result { + const FLAGS: [&str; 3] = ["force", "deep", "repair"]; + + if is_fresh { + for (key, value) in FLAGS.iter().zip([args.force, args.deep, args.repair]) { + let v = if value { "true" } else { "false" }; + store + .set_string_param(key, v) + .await + .map_err(|e| format!("persist `{key}` to params: {e}"))?; + } + // `storage` is absent rather than empty when unset, so a run that + // did not scope itself does not grow a key claiming it did. + if let Some(name) = &args.storage { + store + .set_string_param("storage", name) + .await + .map_err(|e| format!("persist `storage` to params: {e}"))?; + } + return Ok(args.clone()); + } + + let mut restored = JobRunArgs::default(); + for (key, slot) in FLAGS.iter().zip([ + &mut restored.force, + &mut restored.deep, + &mut restored.repair, + ]) { + *slot = match store.get_string_param(key).await { + Ok(v) => v.as_deref() == Some("true"), + Err(e) => return Err(format!("read `{key}` from params: {e}")), + }; + } + restored.storage = store + .get_string_param("storage") + .await + .map_err(|e| format!("read `storage` from params: {e}"))?; + Ok(restored) +} + // ─── Traits — implementor + port ──────────────────────────────────────────── /// The implementor-facing contract for a long-running, restart-tolerant @@ -797,10 +854,44 @@ pub async fn run_or_resume( } } + // Bind the run to the flags it started with. + // + // A Fresh run records its `JobRunArgs` in `params`; a Resumed run reads + // them back and runs with THOSE, ignoring whatever the resuming caller + // passed. Two reasons, and the engine is the only place both are + // guaranteed: + // + // **A resumed run must not change mode.** Handlers read `args` on every + // call, so a paused `?repair=true` import resumed by a plain trigger + // silently continued as import-only — the deletion half never finished + // and nothing said so. The same held for `?deep=true`: a paused bit-rot + // scan resumed shallow while still reporting as the run that started + // deep. Fixing it per-handler meant every job remembering, and three of + // them did not. + // + // **The run row should say what it did.** For a destructive job, "did + // this run delete anything?" is answerable only from `params`, and that + // is what an operator reads afterwards. + // + // Deliberately NOT overridable on resume. Adding `?repair=true` to a + // resume would apply it to the remaining entries only, producing a run + // that half-deleted — the honest way to change your mind is to cancel + // and start fresh. + let args = match persist_or_restore_args(&*store, args, is_fresh).await { + Ok(effective) => effective, + Err(e) => { + // Fail the run rather than guess. Proceeding would mean acting + // under flags nothing recorded, which for the jobs that delete + // is the one thing worth refusing. + log_terminal_write_err("mark_failed", run_id, store.mark_failed(&e).await); + return JobOutcome::err(e); + } + }; + // Dispatch. Terminal writes to `jobs.recoverable_runs` happen // here (NOT in the handler) so the row always ends in a state // that matches what the handler returned. - let outcome = job.run_resumable(&*store, args, resume_cursor).await; + let outcome = job.run_resumable(&*store, &args, resume_cursor).await; // Fetch the terminal run summary so we can surface aggregate // stats (finding_count, scanned_count) on the outer JobOutcome diff --git a/src/infrastructure/services/dedup_service.rs b/src/infrastructure/services/dedup_service.rs index 4ae12f37..b089a1f6 100644 --- a/src/infrastructure/services/dedup_service.rs +++ b/src/infrastructure/services/dedup_service.rs @@ -778,7 +778,33 @@ impl DedupService { kind: &str, variant: &str, ) -> Option { - 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, Option)>( "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. diff --git a/src/infrastructure/services/image_transcode_service.rs b/src/infrastructure/services/image_transcode_service.rs index 4501f746..6b2e6a62 100644 --- a/src/infrastructure/services/image_transcode_service.rs +++ b/src/infrastructure/services/image_transcode_service.rs @@ -16,7 +16,7 @@ use bytes::Bytes; use image::ImageFormat; use std::path::{Path, PathBuf}; -use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::{Arc, OnceLock}; use tokio::fs; @@ -112,6 +112,16 @@ struct AtomicTranscodeStats { transcodes: AtomicU64, bytes_saved: AtomicU64, transcode_errors: AtomicU64, + /// Decodes + encodes that produced something LARGER than the original. + /// + /// Counted separately because `transcodes` means "work that paid off" + /// — it is incremented only on the success path, alongside + /// `bytes_saved`. Without this counter the most expensive failure mode + /// is invisible: the full decode and re-encode of a multi-megapixel + /// image, repeated for every file sharing that content, producing + /// nothing. That is precisely the cost the persisted negative verdict + /// exists to eliminate, so it needs to be measurable before and after. + not_beneficial: AtomicU64, } /// Snapshot of transcoding statistics @@ -122,6 +132,7 @@ pub struct TranscodeStats { pub transcodes: u64, pub bytes_saved: u64, pub transcode_errors: u64, + pub not_beneficial: u64, } impl AtomicTranscodeStats { @@ -132,6 +143,7 @@ impl AtomicTranscodeStats { transcodes: self.transcodes.load(Ordering::Relaxed), bytes_saved: self.bytes_saved.load(Ordering::Relaxed), transcode_errors: self.transcode_errors.load(Ordering::Relaxed), + not_beneficial: self.not_beneficial.load(Ordering::Relaxed), } } } @@ -147,6 +159,29 @@ pub struct ImageTranscodeService { memory_cache: moka::future::Cache, /// Lock-free statistics stats: Arc, + /// 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>, + /// Whether `.transcoded/` still exists, probed once by + /// [`Self::initialize`]. `false` short-circuits the local-cache reads + /// without a syscall. + /// + /// Starts `true` so a service constructed without `initialize` (tests) + /// behaves as before. Failing open is the safe direction: the wrong + /// value costs syscalls, the opposite would hide cached entries that + /// are still there. + legacy_cache: AtomicBool, } impl ImageTranscodeService { @@ -176,21 +211,110 @@ impl ImageTranscodeService { cache_dir, memory_cache, stats: Arc::new(AtomicTranscodeStats::default()), + dedup: OnceLock::new(), + legacy_cache: AtomicBool::new(true), } } - /// 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, + ) { + 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"; + + /// Memory-cache key: by CONTENT when the caller supplied a hash, by + /// file id only when it could not. + /// + /// Transcoding is a pure function of the source bytes, so file keying + /// was always the wrong axis for this cache — it just predated the + /// content-keyed tier. Two files with identical content held two + /// entries for identical bytes, and the second file missed RAM and + /// paid a DB lookup plus a blob read to fetch what was already in + /// memory under another key. + /// + /// The `c:` / `f:` prefixes keep the two namespaces disjoint. A + /// 64-hex hash and a UUID cannot collide in practice, but relying on + /// "in practice" for a cache key is how a file ends up served another + /// file's bytes. + /// + /// Same shape as `ThumbnailCacheKey`'s `content` / `external` split, + /// for the same reason: hash-less callers (external mounts) have no + /// content identity to key on, so they keep the per-file entry. + fn cache_key(source_hash: Option<&str>, file_id: &str, format: OutputFormat) -> String { + match source_hash { + Some(hash) => format!("c:{}:{}", hash, format.extension()), + None => format!("f:{}:{}", file_id, format.extension()), + } + } + + /// 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?; + // Probes, does NOT create. + // + // Creating the tree at boot is what kept `.thumbnails/` alive across + // restarts: the import removed it, the next boot put it back, and + // the absence the read path gates on was unreachable by + // construction. The write path below already calls `create_dir_all` + // on the parent before writing, so nothing needs it created eagerly + // — the only thing eager creation achieved was defeating the drain. + // + // One `stat` on the root, cached for the process lifetime. It can + // only be stale in the harmless direction: a drain completing + // mid-life leaves the flag true until restart, costing the same + // failed opens as before. It never goes false while entries remain, + // because only `transcode_import` removes the tree and it removes + // the whole thing at once. + let present = fs::metadata(&self.cache_dir).await.is_ok(); + self.legacy_cache.store(present, Ordering::Relaxed); + tracing::info!( - "🖼️ Image transcode service initialized (rayon pool: {} threads, cache dir: {:?})", + "🖼️ Image transcode service initialized (rayon pool: {} threads)", transcode_thread_count(), - self.cache_dir ); + if present { + tracing::info!( + target: "oxicloud::transcode", + event = "transcode.legacy_cache_present", + path = ?self.cache_dir, + "legacy transcode cache present — reads fall back to it. Run \ + transcode_import with ?repair=true to drain it." + ); + } Ok(()) } + /// Whether the legacy local cache is worth touching. + /// + /// Unlike the thumbnail tiers this may legitimately never reach `false`: + /// callers with no content hash (external mounts) cannot use the + /// content-keyed tier at all, so they still read and write here. On an + /// install without such mounts the directory drains once and stays + /// gone; on one with them it persists, and that is correct rather than + /// a stalled migration. + fn legacy_cache_active(&self) -> bool { + self.legacy_cache.load(Ordering::Relaxed) + } + /// Check if a mime type can be transcoded. /// /// JPEG is deliberately excluded: the `image` crate's WebP encoder is @@ -213,14 +337,24 @@ 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> { - let cache_key = format!("{}:{}", file_id, target_format.extension()); + let cache_key = Self::cache_key(source_hash, file_id, target_format); // ── 1. Fast path: moka memory cache (lock-free read) ── // An empty-Bytes entry is the negative sentinel: "transcoding this @@ -248,8 +382,14 @@ 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) - .await + self.compute_transcode( + file_id, + source_hash, + original_for_loader, + original_mime, + target_format, + ) + .await }) .await // try_get_with shares one `Arc` across waiters; DomainError @@ -271,13 +411,64 @@ impl ImageTranscodeService { async fn compute_transcode( &self, file_id: &str, + source_hash: Option<&str>, original_content: Bytes, original_mime: &str, target_format: OutputFormat, ) -> Result { - // ── 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) { + if self.legacy_cache_active() && tokio::fs::try_exists(&cache_path).await.unwrap_or(false) { match fs::read(&cache_path).await { Ok(data) => { self.stats.disk_hits.fetch_add(1, Ordering::Relaxed); @@ -292,7 +483,8 @@ impl ImageTranscodeService { // ── Negative verdict persisted on disk (survives restarts) ── let skip_marker = self.get_skip_marker_path(file_id, target_format); - if tokio::fs::try_exists(&skip_marker).await.unwrap_or(false) { + if self.legacy_cache_active() && tokio::fs::try_exists(&skip_marker).await.unwrap_or(false) + { self.stats.disk_hits.fetch_add(1, Ordering::Relaxed); tracing::debug!("💾 Transcode negative disk marker HIT: {}", file_id); return Ok(Bytes::new()); @@ -320,41 +512,127 @@ impl ImageTranscodeService { let transcoded_size = transcoded_bytes.len(); if transcoded_size >= original_size { + // Counted here, not with `transcodes` — the work happened but + // paid nothing, and conflating the two would hide the cost this + // whole negative-verdict mechanism exists to stop paying. + self.stats.not_beneficial.fetch_add(1, Ordering::Relaxed); tracing::debug!( "⚠️ Transcode not beneficial for {}: {} -> {} bytes", file_id, 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). - let marker = self.get_skip_marker_path(file_id, target_format); - tokio::spawn(async move { - if let Some(parent) = marker.parent() { - let _ = fs::create_dir_all(parent).await; + // 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" + ); + } } - if let Err(e) = fs::write(&marker, b"").await { - tracing::warn!("Failed to persist transcode skip marker: {}", e); + // 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() { + let _ = fs::create_dir_all(parent).await; + } + if let Err(e) = fs::write(&marker, b"").await { + 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) ── - let cache_path_clone = cache_path.clone(); - let transcoded_for_disk = transcoded_bytes.clone(); - tokio::spawn(async move { - if let Some(parent) = cache_path_clone.parent() { - let _ = fs::create_dir_all(parent).await; + // ── 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(); + // Awaited, NOT spawned. + // + // Fire-and-forget looked free — the bytes are already on + // their way to the client — but it raced its own purpose. A + // second request for the SAME content arriving before the + // spawn lands finds no row, re-runs the whole decode + + // encode, and stores the identical blob again. The point of + // keying by content is that identical content is derived + // once; a write that has not landed yet cannot deliver + // that, and the window is milliseconds wide precisely when + // it matters most (a page loading many images at once). + // + // Caught by `transcode_cache.hurl`, which asserts the second + // distinct file with identical bytes does not re-transcode + // — it had been passing on timing luck. + // + // The cost is bounded: this path has just spent a full + // decode and re-encode, so one blob write is marginal + // beside it, and it only runs on a genuine miss. + 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" + ); + } } - if let Err(e) = fs::write(&cache_path_clone, &transcoded_for_disk).await { - tracing::warn!("Failed to cache transcoded image: {}", e); + None => { + let cache_path_clone = cache_path.clone(); + let transcoded_for_disk = transcoded_bytes.clone(); + tokio::spawn(async move { + if let Some(parent) = cache_path_clone.parent() { + let _ = fs::create_dir_all(parent).await; + } + if let Err(e) = fs::write(&cache_path_clone, &transcoded_for_disk).await { + tracing::warn!("Failed to cache transcoded image: {}", e); + } + }); } - }); + } // ── Update stats (lock-free atomics) ── self.stats.transcodes.fetch_add(1, Ordering::Relaxed); @@ -392,7 +670,18 @@ impl ImageTranscodeService { /// Invalidate cached transcodes for a file pub async fn invalidate(&self, file_id: &str) { - let cache_key = format!("{}:{}", file_id, OutputFormat::WebP.extension()); + // Only the FILE-keyed entry, deliberately. + // + // Content-keyed entries must not be dropped here: this file's + // content changing says nothing about the other files sharing the + // old bytes, and evicting theirs would make one user's edit cost + // everyone else a re-transcode. They need no eviction anyway — + // new content is a new hash, so the old key is simply never + // consulted again, and moka's TTL reclaims it. + // + // What remains here is the fallback entry for hash-less callers, + // plus the legacy on-disk pair, which are genuinely per-file. + let cache_key = Self::cache_key(None, file_id, OutputFormat::WebP); self.memory_cache.invalidate(&cache_key).await; let cache_path = self.get_cache_path(file_id, OutputFormat::WebP); @@ -425,6 +714,65 @@ impl ImageTranscodeService { // ─── CPU-bound transcoding (runs on rayon, never on Tokio) ─────────────────── +#[cfg(test)] +mod fixture_premise { + //! Pins the property `tests/api/transcode_cache.hurl` is built on: one + //! fixture WebP shrinks, one it does not. + //! + //! The negative half was hard to come by and the reason is worth + //! recording. Synthetic images do not reproduce it — flat colour goes + //! 4780 → 186 bytes, a gradient 24852 → 102, and even uniform RGBA + //! noise still loses by ~242 bytes at any size, a margin that is + //! constant in absolute terms and so never flips. + //! + //! Two things have to be true at once, and only real content does + //! both. The encoder here is the `image` crate's own minimal VP8L + //! writer, not libwebp — it does none of libwebp's search over + //! predictors, colour transforms and Huffman groups — so it only wins + //! where redundancy is extreme enough that any encoder finds it. And + //! the original has to be near PNG-optimal, which a screenshot from a + //! real capture tool is: a 2× Retina UI is long identical runs, flat + //! panels and sharp edges, exactly what PNG's scanline filters plus + //! zlib were designed around. + //! + //! So the negative verdict this service persists is partly a property + //! of THIS encoder, not of the content. Swapping in libwebp would + //! likely flip most of these to positive and leave the stored negative + //! rows stale — an encoder change has to purge them. + + use super::*; + + fn webp_len(path: &str) -> (usize, usize) { + let png = std::fs::read(path).expect("fixture present"); + let webp = + transcode_image_blocking(&Bytes::from(png.clone()), "image/png", OutputFormat::WebP) + .expect("fixture decodes"); + (png.len(), webp.len()) + } + + /// If this ever fails, the hurl scenario's negative half has silently + /// become a second positive test — it would still pass while checking + /// nothing it was written to check. + #[test] + fn screenshot_fixture_is_a_genuine_negative() { + let (png, webp) = webp_len("tests/fixtures/negative-cache-transcode.png"); + assert!( + webp >= png, + "negative-cache-transcode.png no longer defeats the WebP encoder: \ + png={png} webp={webp}" + ); + } + + #[test] + fn flat_colour_fixture_is_a_genuine_positive() { + let (png, webp) = webp_len("tests/fixtures/red-image.png"); + assert!( + webp < png, + "red-image.png stopped shrinking: png={png} webp={webp}" + ); + } +} + /// Perform actual image transcoding. This is a pure CPU function — safe to call /// from `rayon::spawn` or `spawn_blocking`. fn transcode_image_blocking( @@ -476,12 +824,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(), diff --git a/src/infrastructure/services/mod.rs b/src/infrastructure/services/mod.rs index 4b75ac4c..b872a0bb 100644 --- a/src/infrastructure/services/mod.rs +++ b/src/infrastructure/services/mod.rs @@ -64,6 +64,7 @@ pub mod thumb_derived_import_service; pub mod thumbnail_service; #[cfg(test)] mod thumbnail_service_test; +pub mod transcode_import_service; pub mod trash_cleanup_service; pub mod tree_etag_flush_service; pub mod webdav_dead_property_store; diff --git a/src/infrastructure/services/satellites_consistency_service.rs b/src/infrastructure/services/satellites_consistency_service.rs index f54baf97..99259263 100644 --- a/src/infrastructure/services/satellites_consistency_service.rs +++ b/src/infrastructure/services/satellites_consistency_service.rs @@ -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, 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)) diff --git a/src/infrastructure/services/thumb_attached_import_service.rs b/src/infrastructure/services/thumb_attached_import_service.rs index 14748df6..92917d55 100644 --- a/src/infrastructure/services/thumb_attached_import_service.rs +++ b/src/infrastructure/services/thumb_attached_import_service.rs @@ -476,6 +476,18 @@ impl RecoverableJobHandler for ThumbAttachedImport { } } + // Flush the tail — see the derived twin. The loop only checkpoints + // on a full batch, so the remainder went uncounted: a 105-file run + // reported `scanned_count: 100`, and a run shorter than one batch + // reported zero and left the progress bar at zero throughout. + if since_checkpoint > 0 + && let Err(e) = store.checkpoint(Vec::new(), since_checkpoint as u64).await + { + return RunOutcome::Failed { + message: format!("final checkpoint: {e}"), + }; + } + // Both jobs attempt the teardown, and it no-ops unless the tree is // drained of files EITHER of them claims. Without this, whichever // job runs last leaves an empty `.thumbnails/` behind until the diff --git a/src/infrastructure/services/thumb_derived_import_service.rs b/src/infrastructure/services/thumb_derived_import_service.rs index 490f272a..37e67200 100644 --- a/src/infrastructure/services/thumb_derived_import_service.rs +++ b/src/infrastructure/services/thumb_derived_import_service.rs @@ -734,6 +734,19 @@ impl RecoverableJobHandler for ThumbDerivedImport { } } + // Flush the tail. The loop only checkpoints on a full batch, so the + // remainder after the last one was never counted — a run of fewer + // than BATCH_SIZE files reported `scanned_count: 0` against a known + // total and left the admin progress bar at zero for its whole life. + // Same fix in both imports and in transcode_import. + if since_checkpoint > 0 + && let Err(e) = store.checkpoint(Vec::new(), since_checkpoint as u64).await + { + return RunOutcome::Failed { + message: format!("final checkpoint: {e}"), + }; + } + // Remove the size directories once genuinely empty, because ABSENCE // is what step 10e gates the fallback removal on — not emptiness. // Empty is momentary: an on-demand render can repopulate it the next diff --git a/src/infrastructure/services/transcode_import_service.rs b/src/infrastructure/services/transcode_import_service.rs new file mode 100644 index 00000000..e1464c54 --- /dev/null +++ b/src/infrastructure/services/transcode_import_service.rs @@ -0,0 +1,584 @@ +//! Step 7 migration tenant: drains `.transcoded/` into the derived tier. +//! +//! The twin of `thumb_derived_import`, with one difference that shapes the +//! whole job: **the legacy tree is keyed by file, the destination by +//! content.** Thumbnail sidecars were already named by blob hash, so their +//! import was a move. These are named `{file_id}.webp`, so every entry has +//! to be re-keyed through `storage.files` before it can be stored. +//! +//! That re-keying is not bookkeeping — it is the point. A sandbox with five +//! `.skip` markers had three of them naming the same content, so the +//! file-keyed tree held three copies of one verdict. After the import that +//! is a single row, and any future upload of those bytes inherits it +//! instead of paying for the decision again. +//! +//! ### Two artifact kinds, one walk +//! +//! * `{file_id}.webp` — a cached transcode. Imported as a derived Blob. +//! * `{file_id}.webp.skip` — a zero-byte marker meaning "WebP came out +//! larger for this file". Imported as a NEGATIVE row (NULL `blob_hash`), +//! so the verdict survives the deletion of the directory holding it. +//! +//! Both are claimed by the same walk because they share a source file and +//! a cursor; splitting them would mean two passes over one directory and +//! two chances for the pair to disagree about what has been handled. +//! +//! ### What is deliberately not imported +//! +//! Entries whose file is gone. The destination row is keyed by content +//! hash, which is resolved *through* `storage.files` — no file, no hash, +//! nothing to key by. Under `repair` these are deleted, because they are +//! unimportable by definition and a run that keeps rediscovering them +//! never reports zero, so the gate for removing the directory never opens. + +use std::path::PathBuf; +use std::sync::Arc; + +use async_trait::async_trait; +use bytes::Bytes; +use sqlx::PgPool; +use tokio::fs; + +use crate::infrastructure::scheduler::{ + JobRegistry, JobRunArgs, JobStore, JobStoreProvider, Mutates, RecoverableJobHandler, + RunOutcome, RunStatus, record_or_log, +}; +use crate::infrastructure::services::dedup_service::DedupService; +use crate::infrastructure::services::thumb_derived_import_service::audit_sidecar_deleted; + +pub const TRANSCODE_IMPORT_JOB_NAME: &str = "transcode_import"; + +/// `content_derived_blobs.kind` for everything this job writes. Must match +/// `ImageTranscodeService::DERIVED_KIND`, or the import would file its rows +/// where the read path does not look. +const DERIVED_KIND: &str = "transcode"; + +/// The only variant this job handles. `.transcoded/` has exactly one +/// subdirectory today; a second output format would add a directory and a +/// variant together, and this walk would grow a loop rather than a branch. +const VARIANT_WEBP: &str = "webp"; + +/// Files handled between checkpoints. Each is a read plus, at most, a blob +/// write — deliberately smaller than a pure-DB sweep's page. +const BATCH_SIZE: usize = 100; + +pub struct TranscodeImport { + /// `{storage_path}/.transcoded`, matching `ImageTranscodeService::new`. + transcoded_root: PathBuf, + dedup: Arc, + /// Needed for the re-keying: file id → content hash. The thumbnail + /// imports have no equivalent because their sidecars were already + /// content-named. + pool: Arc, +} + +impl TranscodeImport { + pub fn new(transcoded_root: PathBuf, dedup: Arc, pool: Arc) -> Self { + Self { + transcoded_root, + dedup, + pool, + } + } + + pub async fn register_recoverable_job( + self: Arc, + registry: &JobRegistry, + provider: &Arc, + ) -> Arc { + // On-demand, matching the thumbnail imports. The boot run in repair + // mode IS the migration: nothing writes a hash-keyed entry here any + // more, so the tail cannot grow after startup, and a periodic tick + // could not finish the job anyway because ticks never pass + // `repair`. Once drained it would be a `read_dir` returning + // nothing, every day, forever. + registry + .register_recoverable_job(self.clone(), provider.clone(), None) + .await; + self + } + + /// The `webp/` subdirectory, where both artifact kinds live. + fn variant_dir(&self) -> PathBuf { + self.transcoded_root.join(VARIANT_WEBP) + } + + /// Sorted entry names, so the cursor totally orders the traversal. + /// + /// Returns both `{id}.webp` and `{id}.webp.skip`; the caller decides + /// which is which. Anything else is ignored rather than reported — the + /// directory is a local cache and has never promised to hold only our + /// files. + async fn entry_names(dir: &std::path::Path) -> Vec { + let Ok(mut entries) = fs::read_dir(dir).await else { + return Vec::new(); + }; + let mut names = Vec::new(); + while let Ok(Some(entry)) = entries.next_entry().await { + if let Some(name) = entry.file_name().to_str() + && parse_entry(name).is_some() + { + names.push(name.to_string()); + } + } + names.sort(); + names + } + + /// Resolve a file id to the BLAKE3 of its content. + /// + /// `None` means the file is gone — which is the unimportable case, not + /// an error: the destination is keyed by content, and a deleted file + /// has no content to key by. + async fn content_hash_of(&self, file_id: &str) -> Option { + sqlx::query_as::<_, (String,)>( + "SELECT blob_hash FROM storage.files WHERE id = $1::uuid AND blob_hash IS NOT NULL", + ) + .bind(file_id) + .fetch_optional(self.pool.as_ref()) + .await + .ok() + .flatten() + .map(|(h,)| h) + } +} + +/// Checkpoint once a batch has accumulated, resetting the counter. +/// +/// Returns `Some(RunOutcome::Failed)` when the store write fails, for the +/// caller to return. Shared by both exits of the loop body so the cursor +/// advances identically whether an entry was imported or skipped — the +/// alternative is two copies of this, which is how the first draft ended +/// up dropping a future and silently never checkpointing on one path. +async fn checkpoint_if_due( + store: &dyn JobStore, + name: &str, + since_checkpoint: &mut usize, +) -> Option { + if *since_checkpoint < BATCH_SIZE { + return None; + } + let scanned = *since_checkpoint as u64; + *since_checkpoint = 0; + match store.checkpoint(name.as_bytes().to_vec(), scanned).await { + Ok(()) => None, + Err(e) => Some(RunOutcome::Failed { + message: format!("checkpoint: {e}"), + }), + } +} + +/// What a `.transcoded/webp/` entry names. +/// +/// Returns the file id and whether it is a negative marker. Anything not +/// matching either shape is not ours. +fn parse_entry(name: &str) -> Option<(&str, bool)> { + if let Some(id) = name.strip_suffix(".webp.skip") { + return (!id.is_empty()).then_some((id, true)); + } + if let Some(id) = name.strip_suffix(".webp") { + return (!id.is_empty()).then_some((id, false)); + } + None +} + +#[async_trait] +impl RecoverableJobHandler for TranscodeImport { + fn name(&self) -> &str { + TRANSCODE_IMPORT_JOB_NAME + } + + fn description(&self) -> &'static str { + "Migrates cached WebP transcodes out of the legacy .transcoded/ \ + directory into content-addressed blob storage, re-keying each one \ + from its file id to its content hash. Entries for identical \ + content collapse into a single row, so the same image cached under \ + several files stops being stored several times. Zero-byte .skip \ + markers become negative rows, preserving the verdict that a file \ + is not worth transcoding." + } + + fn mutates(&self) -> Mutates { + Mutates::Always + } + + fn repair_description(&self) -> Option<&'static str> { + Some( + "Also DELETES each cached transcode once its replacement has \ + been read back and compared byte for byte, and removes the \ + directory when empty. Entries whose file no longer exists are \ + deleted without a readback — they cannot be re-keyed and \ + nothing can reference them again. Irreversible, though a \ + transcode is a pure function of its source: anything deleted \ + in error is recomputed on the next request.", + ) + } + + async fn count_total(&self) -> Option { + Some(Self::entry_names(&self.variant_dir()).await.len() as u64) + } + + async fn run_resumable( + &self, + store: &dyn JobStore, + args: &JobRunArgs, + resume_cursor: Option>, + ) -> RunOutcome { + // Cursor is the entry name. One directory, sorted, so the name + // alone totally orders the walk — unlike the thumbnail imports, + // which need `{size}/{name}` to span three directories. + let cursor: Option = match resume_cursor { + None => None, + Some(b) if b.is_empty() => None, + Some(b) => match String::from_utf8(b) { + Ok(s) => Some(s), + Err(e) => { + return RunOutcome::Failed { + message: format!("invalid cursor: not valid UTF-8: {e}"), + }; + } + }, + }; + + let delete_imported = args.repair; + let dir = self.variant_dir(); + + let mut imported = 0u64; + let mut negatives = 0u64; + let mut already = 0u64; + let mut file_gone = 0u64; + let mut deleted = 0u64; + let mut unverified = 0u64; + let mut failed = 0u64; + let mut since_checkpoint = 0usize; + // Last entry visited, for the tail flush below. + let mut last_name: Option = None; + + for name in Self::entry_names(&dir).await { + if let Some(c) = &cursor + && name.as_str() <= c.as_str() + { + continue; + } + + match store.status().await { + Ok(RunStatus::CancelRequested) => { + return RunOutcome::Paused { + cursor: name.into_bytes(), + }; + } + Ok(_) => {} + Err(e) => { + return RunOutcome::Failed { + message: format!("status poll: {e}"), + }; + } + } + + let Some((file_id, is_negative)) = parse_entry(&name) else { + continue; + }; + let path = dir.join(&name); + + // Re-key. This is the step the thumbnail imports do not have. + let Some(source_hash) = self.content_hash_of(file_id).await else { + file_gone += 1; + let mut removed = false; + if delete_imported && fs::remove_file(&path).await.is_ok() { + deleted += 1; + removed = true; + audit_sidecar_deleted( + TRANSCODE_IMPORT_JOB_NAME, + "file_gone", + file_id, + "-", + &path, + ); + } + record_or_log( + store, + TRANSCODE_IMPORT_JOB_NAME, + "transcode_file_gone", + "anomaly", + None, + serde_json::json!({ + "path": name, + "file_id": file_id, + "deleted": removed, + "note": "no storage.files row, so the entry cannot be re-keyed to a \ + content hash; unimportable and unreachable", + }), + ) + .await; + // Falls through to the shared checkpoint at the end of the + // loop rather than duplicating it here. An earlier draft + // did duplicate it and dropped the future without + // awaiting — the entry counted toward the batch, the + // cursor never advanced, and a resumed run would have + // rewalked everything already handled. + since_checkpoint += 1; + if let Some(failure) = checkpoint_if_due(store, &name, &mut since_checkpoint).await + { + return failure; + } + continue; + }; + + if is_negative { + // A verdict, not bytes. `store_derived_negative` is + // ON CONFLICT DO NOTHING, so the three markers that named + // one piece of content in the sandbox collapse here rather + // than fighting over the row. + match self + .dedup + .store_derived_negative(&source_hash, DERIVED_KIND, VARIANT_WEBP) + .await + { + Ok(()) => { + negatives += 1; + if delete_imported && fs::remove_file(&path).await.is_ok() { + deleted += 1; + audit_sidecar_deleted( + TRANSCODE_IMPORT_JOB_NAME, + "negative_imported", + file_id, + "-", + &path, + ); + } + } + Err(e) => { + failed += 1; + tracing::warn!( + target: "oxicloud::dedup", + event = "transcode_import.negative_failed", + file_id = %file_id, + source_hash = %source_hash, + error = %e, + "failed to record negative transcode row; entry kept" + ); + } + } + } else if self + .dedup + .find_derived_blob(&source_hash, DERIVED_KIND, VARIANT_WEBP) + .await + .is_some() + { + // Already imported — by an earlier run, or by another file + // sharing this content. Checked BEFORE storing so a re-run + // does not release and retake the reference. + already += 1; + if delete_imported { + let existing = self + .dedup + .find_derived_blob(&source_hash, DERIVED_KIND, VARIANT_WEBP) + .await; + if let Some(r) = existing { + if crate::infrastructure::services::thumb_derived_import_service::ThumbDerivedImport::verify_and_unlink( + &self.dedup, + TRANSCODE_IMPORT_JOB_NAME, + &source_hash, + &r.blob_hash, + &path, + ) + .await + { + deleted += 1; + } else { + unverified += 1; + record_or_log( + store, + TRANSCODE_IMPORT_JOB_NAME, + "transcode_delete_unverified", + "anomaly", + None, + serde_json::json!({ + "path": name, + "file_id": file_id, + "source_hash": source_hash, + "note": "stored transcode did not read back identical; \ + cached copy kept", + }), + ) + .await; + } + } + } + } else { + match fs::read(&path).await { + Ok(data) => { + match self + .dedup + .store_derived_blob( + &source_hash, + DERIVED_KIND, + VARIANT_WEBP, + "image/webp", + Bytes::from(data), + ) + .await + { + Ok(stored_hash) => { + imported += 1; + if delete_imported { + if crate::infrastructure::services::thumb_derived_import_service::ThumbDerivedImport::verify_and_unlink( + &self.dedup, + TRANSCODE_IMPORT_JOB_NAME, + &source_hash, + &stored_hash, + &path, + ) + .await + { + deleted += 1; + } else { + unverified += 1; + } + } + } + Err(e) => { + failed += 1; + tracing::warn!( + target: "oxicloud::dedup", + event = "transcode_import.store_failed", + file_id = %file_id, + source_hash = %source_hash, + error = %e, + "failed to store derived transcode; entry kept" + ); + } + } + } + Err(e) => { + failed += 1; + tracing::warn!( + target: "oxicloud::dedup", + event = "transcode_import.read_failed", + path = %path.display(), + error = %e, + "failed to read cached transcode; entry kept" + ); + } + } + } + + since_checkpoint += 1; + last_name = Some(name.clone()); + if let Some(failure) = checkpoint_if_due(store, &name, &mut since_checkpoint).await { + return failure; + } + } + + // Flush the tail. + // + // `checkpoint_if_due` only fires on a full batch, so a run shorter + // than BATCH_SIZE never checkpointed at all and reported + // `scanned_count: 0` against a known `total_rows` — the admin + // progress bar sat at zero through the whole run and finished + // there. Longer runs were wrong too, just less visibly: the + // remainder after the last full batch was never counted. + // + // Cursor-wise this is a no-op — the walk is finished, so nothing + // will resume from it — but the scanned delta is what the progress + // display reads, and it has to include the last partial batch. + if since_checkpoint > 0 + && let Some(name) = last_name + && let Err(e) = store + .checkpoint(name.into_bytes(), since_checkpoint as u64) + .await + { + return RunOutcome::Failed { + message: format!("final checkpoint: {e}"), + }; + } + + // Remove the tree once drained. Deletion first, rename only if a + // non-cache file is in the way — same rule as `.thumbnails/`, and + // for the same reason: absence is what the read path tests, and a + // stray `.DS_Store` must not keep the fallback alive forever. + if delete_imported { + let _ = fs::remove_dir(&dir).await; + match fs::remove_dir(&self.transcoded_root).await { + Ok(()) => tracing::info!( + target: "oxicloud::dedup", + event = "transcode_import.root_removed", + run_id = %store.run_id(), + path = %self.transcoded_root.display(), + "🧹 legacy transcode directory removed" + ), + Err(_) => { + let parked = self.transcoded_root.with_file_name(".transcoded.migrated"); + match fs::rename(&self.transcoded_root, &parked).await { + Ok(()) => tracing::info!( + target: "oxicloud::dedup", + event = "transcode_import.root_parked", + run_id = %store.run_id(), + to = %parked.display(), + "🧹 legacy transcode directory could not be removed (a non-cache \ + file remains) — moved aside instead" + ), + Err(e) => tracing::warn!( + target: "oxicloud::dedup", + event = "transcode_import.root_kept", + run_id = %store.run_id(), + reason = %e, + "legacy transcode directory neither removed nor moved aside" + ), + } + } + } + } + + tracing::info!( + target: "oxicloud::dedup", + event = "transcode_import.completed", + run_id = %store.run_id(), + imported = imported, + negatives = negatives, + already_present = already, + file_gone = file_gone, + deleted = deleted, + unverified = unverified, + failed = failed, + "transcode_import: {imported} imported, {negatives} negative verdict(s), \ + {already} already present, {file_gone} unimportable, {deleted} deleted, \ + {unverified} kept unverified, {failed} failed" + ); + + RunOutcome::completed_with(serde_json::json!({ + "imported": imported, + "negatives": negatives, + "already_present": already, + "file_gone": file_gone, + "deleted": deleted, + "unverified": unverified, + "failed": failed, + })) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// The walk claims both artifact kinds and nothing else. + /// + /// `.webp.skip` must be tested BEFORE `.webp`, or the shorter suffix + /// matches first and every marker imports as if it were a cached + /// transcode — reading a zero-byte file and storing it as the + /// transcode of its source, which would then be served to clients. + #[test] + fn entry_names_are_parsed_by_longest_suffix_first() { + let id = "3f2b1c00-1111-2222-3333-444455556666"; + + assert_eq!(parse_entry(&format!("{id}.webp")), Some((id, false))); + assert_eq!(parse_entry(&format!("{id}.webp.skip")), Some((id, true))); + + // Not ours: no id, wrong extension, or a bare marker. + assert_eq!(parse_entry(".webp"), None); + assert_eq!(parse_entry(".webp.skip"), None); + assert_eq!(parse_entry(&format!("{id}.jpg")), None); + assert_eq!(parse_entry(".DS_Store"), None); + } +} diff --git a/src/interfaces/api/handlers/admin_handler.rs b/src/interfaces/api/handlers/admin_handler.rs index 4e184929..455f672f 100644 --- a/src/interfaces/api/handlers/admin_handler.rs +++ b/src/interfaces/api/handlers/admin_handler.rs @@ -159,6 +159,11 @@ pub fn admin_routes() -> Router> { // `/blob/{hash}`) stay at `/api/dedup/*`. .route("/dedup/stats", get(get_stats)) .route("/dedup/recalculate", post(recalculate_stats)) + // Transcode effectiveness. Nothing exposed these before, so there + // was no way to tell a served-from-cache response from one that + // re-ran the decode + encode — not from the outside, and not from + // a test either. + .route("/transcode/stats", get(get_transcode_stats)) // SMTP diagnostics .route("/smtp/info", get(get_smtp_info)) .route("/smtp/test", post(send_smtp_test)) @@ -2502,6 +2507,51 @@ pub async fn delete_drive_admin( /// /// Production endpoint, always on. Read-only, so no audit line — /// the standard admin-middleware auth check is enough. +/// `GET /api/admin/transcode/stats` — WebP transcode effectiveness. +/// +/// The four counters distinguish where a response came from, which is +/// otherwise invisible: `transcodes` is work actually done, while +/// `cache_hits` (in-memory, keyed by file id) and `disk_hits` (the +/// durable content-keyed tier, plus the legacy local cache) are work +/// avoided. A rising `transcodes` against a flat `disk_hits` means the +/// derived tier is not being consulted — which is exactly the +/// regression a migration can introduce silently. +/// +/// `bytes_saved` counts only successful transcodes; images the encoder +/// could not shrink contribute nothing to it and are remembered as +/// negative rows instead. +/// +/// Read-only, so no audit line — the admin middleware gate is enough. +#[utoipa::path( + get, + path = "/api/admin/transcode/stats", + responses( + (status = 200, description = "Transcode statistics"), + (status = 401, description = "Unauthorized"), + (status = 403, description = "Admin required"), + ), + security(("bearerAuth" = [])), + tag = "admin" +)] +pub async fn get_transcode_stats(State(state): State>) -> impl IntoResponse { + let s = state.core.image_transcode_service.get_stats().await; + ( + StatusCode::OK, + Json(serde_json::json!({ + "cache_hits": s.cache_hits, + "disk_hits": s.disk_hits, + "transcodes": s.transcodes, + "bytes_saved": s.bytes_saved, + "transcode_errors": s.transcode_errors, + // Decodes that produced something larger. Work done for no + // gain — the thing the stored negative verdict prevents + // repeating, and invisible before this counter existed. + "not_beneficial": s.not_beneficial, + })), + ) + .into_response() +} + #[utoipa::path( get, path = "/api/admin/jobs", diff --git a/tests/api/run.sh b/tests/api/run.sh index 75d89c34..1c483ea1 100755 --- a/tests/api/run.sh +++ b/tests/api/run.sh @@ -169,6 +169,8 @@ hurl --variables-file "$API_DIR/test.env" --file-root "$REPO_ROOT/tests" --test "$API_DIR/derived_blob_copy.hurl" \ "$API_DIR/thumbnail_etag_content_keyed.hurl" \ "$API_DIR/attached_thumbnail_copy.hurl" \ + "$API_DIR/transcode_cache.hurl" \ + "$API_DIR/transcode_import.hurl" \ "$API_DIR/dedup_admin_gate.hurl" \ "$API_DIR/admin_jobs.hurl" \ "$API_DIR/recoverable_jobs.hurl" \ diff --git a/tests/api/transcode_cache.hurl b/tests/api/transcode_cache.hurl new file mode 100644 index 00000000..466a21f6 --- /dev/null +++ b/tests/api/transcode_cache.hurl @@ -0,0 +1,364 @@ +# ============================================================= +# OxiCloud – Transcode caching: positive and negative +# +# Pins that a WebP transcode is computed ONCE per distinct content and +# then answered from the derived tier, in both directions: +# +# * positive — WebP is smaller, so the bytes are stored and reused +# * negative — WebP came out larger, so the VERDICT is stored and the +# decode + encode is not repeated +# +# ## Why this can assert what the thumbnail tests could not +# +# `derived_blob_copy.hurl` documents that thumbnail tier selection is +# invisible over HTTP: stored blob, RAM cache and a fresh re-render all +# return identical bytes. Transcodes are the same — but +# `GET /api/admin/transcode/stats` now exposes the counters, so "was +# this computed or served" becomes observable from outside the process. +# `transcodes` is work done; `cache_hits` (RAM, keyed by file id) and +# `disk_hits` (the durable content-keyed tier) are work avoided. +# +# ## Why each case uploads the same bytes twice +# +# Re-fetching the SAME file proves only that a cache exists. Uploading +# identical content as a SECOND file gives a different file id but the +# same content hash, which is the case that distinguishes content keying +# from file keying — and both caches here are now content-keyed. +# +# So the assertion is that the second file costs NO transcode, and is +# served from RAM. Under the previous file-keyed memory cache it was a +# guaranteed RAM miss: two entries for identical bytes, and a DB lookup +# plus a blob read to fetch what was already in memory under another key. +# +# What this can no longer isolate is the DURABLE tier, because the RAM +# cache now answers first for anything within one process lifetime. +# That tier is covered instead by `satellites_consistency` (every row +# points at a live blob) and by a restart, which hurl cannot perform. +# +# ## Fixtures +# +# `red-image.png` shrinks (4780 → 186 bytes). `negative-cache-transcode.png` +# does not — a real screenshot, which is the only thing that defeats this +# encoder; synthetic images all come out positive. Both properties are +# pinned by `image_transcode_service::fixture_premise`, so if an encoder +# bump ever flips one, that unit test fails loudly instead of this +# scenario quietly testing nothing. +# +# Prerequisites: setup.hurl must have run (admin user exists). +# +# Run: +# hurl --variables-file tests/api/test.env --file-root tests \ +# --test tests/api/transcode_cache.hurl +# ============================================================= + + +# ───────────────────────────────────────────────────────────── +# Step 1 – Login +# ───────────────────────────────────────────────────────────── +POST {{base_url}}/api/auth/login +Content-Type: application/json +{ + "username": "{{username}}", + "password": "{{password}}" +} + +HTTP 200 +[Captures] +token: jsonpath "$.access_token" + + +# ───────────────────────────────────────────────────────────── +# Step 2 – Working folder +# ───────────────────────────────────────────────────────────── +POST {{base_url}}/api/folders +Authorization: Bearer {{token}} +Content-Type: application/json +{ + "name": "hurl-transcode-cache" +} + +HTTP 201 +[Captures] +folder_id: jsonpath "$.id" + + +# ───────────────────────────────────────────────────────────── +# Step 2b – A second folder, for the duplicate uploads. +# +# Re-uploading the same filename into the SAME folder overwrites the +# existing file and returns its id, so both halves of a "two distinct +# files, one content" pair would be the same row and the test would +# assert nothing. A second folder keeps the name free. +# ───────────────────────────────────────────────────────────── +POST {{base_url}}/api/folders +Authorization: Bearer {{token}} +Content-Type: application/json +{ + "name": "hurl-transcode-cache-dup" +} + +HTTP 201 +[Captures] +folder_dup_id: jsonpath "$.id" + + +# ───────────────────────────────────────────────────────────── +# Step 3 – Baseline counters. +# +# Absolute values are meaningless here — earlier scenarios in the same +# run transcode images too. Everything below is asserted as a DELTA +# from this point, which is also why this file must not assume it runs +# first. +# ───────────────────────────────────────────────────────────── +GET {{base_url}}/api/admin/transcode/stats +Authorization: Bearer {{token}} + +HTTP 200 +[Captures] +base_transcodes: jsonpath "$.transcodes" +base_not_beneficial: jsonpath "$.not_beneficial" + + +# ═════════════════════════════════════════════════════════════ +# POSITIVE CASE — WebP is smaller +# ═════════════════════════════════════════════════════════════ + +# ───────────────────────────────────────────────────────────── +# Step 4 – Upload a shrinkable image +# ───────────────────────────────────────────────────────────── +POST {{base_url}}/api/files/upload +Authorization: Bearer {{token}} +[MultipartFormData] +folder_id: {{folder_id}} +file: file,fixtures/red-image.png; image/png + +HTTP 201 +[Captures] +pos_a_id: jsonpath "$.id" +pos_hash: jsonpath "$.content_hash" + + +# ───────────────────────────────────────────────────────────── +# Step 5 – Fetch it as a WebP-capable client. +# +# `Accept: image/webp` is what selects the transcode path; +# `BrowserCapabilities::from_accept_header` looks for exactly this. +# ───────────────────────────────────────────────────────────── +GET {{base_url}}/api/files/{{pos_a_id}} +Authorization: Bearer {{token}} +Accept: image/webp,image/png,*/* + +HTTP 200 +[Asserts] +header "Content-Type" contains "image/webp" + + +# ───────────────────────────────────────────────────────────── +# Step 6 – That was work actually done, not a cache hit. +# +# Captured rather than computed: hurl has no arithmetic in predicates, +# and pinning the exact value here is stronger anyway — the later steps +# assert equality against it, so any transcode from any source shows up. +# ───────────────────────────────────────────────────────────── +GET {{base_url}}/api/admin/transcode/stats +Authorization: Bearer {{token}} + +HTTP 200 +[Captures] +after_positive: jsonpath "$.transcodes" +after_positive_disk: jsonpath "$.disk_hits" +after_positive_ram: jsonpath "$.cache_hits" +[Asserts] +jsonpath "$.not_beneficial" == {{base_not_beneficial}} + + +# ───────────────────────────────────────────────────────────── +# Step 7 – The SAME bytes uploaded as a second, distinct file. +# +# Same content hash, different file id. Asserting the hash matches is +# what makes the next step meaningful: if these two files did not share +# content, a second transcode would be correct rather than a regression. +# ───────────────────────────────────────────────────────────── +POST {{base_url}}/api/files/upload +Authorization: Bearer {{token}} +[MultipartFormData] +folder_id: {{folder_dup_id}} +file: file,fixtures/red-image.png; image/png + +HTTP 201 +[Captures] +pos_b_id: jsonpath "$.id" +[Asserts] +jsonpath "$.content_hash" == "{{pos_hash}}" +jsonpath "$.id" != "{{pos_a_id}}" + + +# ───────────────────────────────────────────────────────────── +# Step 8 – Fetching the second file still yields WebP. +# ───────────────────────────────────────────────────────────── +GET {{base_url}}/api/files/{{pos_b_id}} +Authorization: Bearer {{token}} +Accept: image/webp,image/png,*/* + +HTTP 200 +[Asserts] +header "Content-Type" contains "image/webp" + + +# ───────────────────────────────────────────────────────────── +# Step 9 – …WITHOUT a second transcode. +# +# The memory cache could not have served this: it is keyed by file id +# and this is a different file. Only the content-keyed derived tier +# answers here, which is the whole point of keying derivations by +# content rather than by file. +# ───────────────────────────────────────────────────────────── +GET {{base_url}}/api/admin/transcode/stats +Authorization: Bearer {{token}} + +HTTP 200 +[Asserts] +# The property that matters, whichever tier answered: identical content is +# transcoded ONCE, however many files carry it. +jsonpath "$.transcodes" == {{after_positive}} +jsonpath "$.not_beneficial" == {{base_not_beneficial}} +# Served from RAM, and that is the point of the memory cache being keyed by +# CONTENT rather than by file id. Under file keying this second file was a +# guaranteed RAM miss that fell through to a DB lookup plus a blob read to +# fetch bytes already in memory under another key — `disk_hits` moved and +# `cache_hits` did not. Now it is the other way round. +jsonpath "$.cache_hits" != {{after_positive_ram}} +jsonpath "$.disk_hits" == {{after_positive_disk}} + + +# ═════════════════════════════════════════════════════════════ +# NEGATIVE CASE — WebP comes out larger +# ═════════════════════════════════════════════════════════════ + +# ───────────────────────────────────────────────────────────── +# Step 10 – Upload an image the encoder cannot shrink +# ───────────────────────────────────────────────────────────── +POST {{base_url}}/api/files/upload +Authorization: Bearer {{token}} +[MultipartFormData] +folder_id: {{folder_id}} +file: file,fixtures/negative-cache-transcode.png; image/png + +HTTP 201 +[Captures] +neg_a_id: jsonpath "$.id" +neg_hash: jsonpath "$.content_hash" + + +# ───────────────────────────────────────────────────────────── +# Step 11 – A WebP-capable client gets the ORIGINAL back. +# +# Not a failure: transcoding to something larger would cost the client +# bandwidth, so the service serves the PNG and remembers why. +# ───────────────────────────────────────────────────────────── +GET {{base_url}}/api/files/{{neg_a_id}} +Authorization: Bearer {{token}} +Accept: image/webp,image/png,*/* + +HTTP 200 +[Asserts] +header "Content-Type" contains "image/png" + + +# ───────────────────────────────────────────────────────────── +# Step 12 – The attempt still cost one decode + encode. +# ───────────────────────────────────────────────────────────── +GET {{base_url}}/api/admin/transcode/stats +Authorization: Bearer {{token}} + +HTTP 200 +[Captures] +after_negative: jsonpath "$.not_beneficial" +[Asserts] +# The decode + encode ran and produced nothing usable, which is counted +# separately from `transcodes` — that only counts work that paid off. +jsonpath "$.transcodes" == {{after_positive}} + + +# ───────────────────────────────────────────────────────────── +# Step 13 – Same bytes again, as a distinct file. +# ───────────────────────────────────────────────────────────── +POST {{base_url}}/api/files/upload +Authorization: Bearer {{token}} +[MultipartFormData] +folder_id: {{folder_dup_id}} +file: file,fixtures/negative-cache-transcode.png; image/png + +HTTP 201 +[Captures] +neg_b_id: jsonpath "$.id" +[Asserts] +jsonpath "$.content_hash" == "{{neg_hash}}" +jsonpath "$.id" != "{{neg_a_id}}" + + +# ───────────────────────────────────────────────────────────── +# Step 14 – Original again, as expected. +# ───────────────────────────────────────────────────────────── +GET {{base_url}}/api/files/{{neg_b_id}} +Authorization: Bearer {{token}} +Accept: image/webp,image/png,*/* + +HTTP 200 +[Asserts] +header "Content-Type" contains "image/png" + + +# ───────────────────────────────────────────────────────────── +# Step 15 – …and the verdict was NOT recomputed. +# +# This is the assertion the negative row exists for. Without it the +# server re-runs a full decode + encode of a half-megabyte screenshot on +# every request for every file sharing that content, only to throw the +# result away each time. A counter that moved here would mean the +# negative row was not written, not read, or not keyed by content. +# ───────────────────────────────────────────────────────────── +GET {{base_url}}/api/admin/transcode/stats +Authorization: Bearer {{token}} + +HTTP 200 +[Asserts] +jsonpath "$.not_beneficial" == {{after_negative}} +jsonpath "$.transcodes" == {{after_positive}} + + +# ───────────────────────────────────────────────────────────── +# Step 16 – Teardown. Hurl files share one database, so a folder left +# behind changes what later scenarios see. +# ───────────────────────────────────────────────────────────── +DELETE {{base_url}}/api/folders/{{folder_id}} +Authorization: Bearer {{token}} + +HTTP 204 + + +DELETE {{base_url}}/api/folders/{{folder_dup_id}} +Authorization: Bearer {{token}} + +HTTP 204 + + +GET {{base_url}}/api/trash/resources +Authorization: Bearer {{token}} + +HTTP 200 +[Captures] +trash_id: jsonpath "$.items[?(@.resource.id == '{{folder_id}}')].resource.id" +trash_dup_id: jsonpath "$.items[?(@.resource.id == '{{folder_dup_id}}')].resource.id" + + +DELETE {{base_url}}/api/trash/{{trash_id}} +Authorization: Bearer {{token}} + +HTTP 200 + + +DELETE {{base_url}}/api/trash/{{trash_dup_id}} +Authorization: Bearer {{token}} + +HTTP 200 diff --git a/tests/api/transcode_import.hurl b/tests/api/transcode_import.hurl new file mode 100644 index 00000000..9ac9d0ea --- /dev/null +++ b/tests/api/transcode_import.hurl @@ -0,0 +1,148 @@ +# ============================================================= +# OxiCloud – transcode_import: registration, contract, empty-tree run +# +# ## What this covers, and what it deliberately does not +# +# It covers the job's SURFACE: that it is registered, that it declares +# the metadata the admin panel switches on, and that a run against an +# already-drained tree completes cleanly with zeroed counters rather +# than erroring or reporting phantom work. +# +# It does NOT cover the re-keying — the part that matters most, where +# `{file_id}.webp` is resolved through `storage.files` to a content +# hash, and several entries naming the same content collapse into one +# row. That needs `.transcoded/webp/` entries on disk before the run, +# and nothing reachable over HTTP can create them: since the write path +# moved to the derived tier, only callers with no content hash +# (external mounts) still write there, and hurl cannot place files in +# the server's storage directory either. +# +# So the migration itself is validated by a snapshot restore against a +# populated `.transcoded/`, the same way the thumbnail migration was — +# an exercise that found three bugs the test suite did not. This file +# guards the contract that surrounds it; it is not a substitute. +# +# The empty-tree assertions are still worth having. A drained tree is +# the steady state of every install after the first boot, so this is +# what the job does on the overwhelming majority of its runs, and +# "does nothing, quietly, forever" is a real property — the thumbnail +# equivalent warned on every boot for exactly this case until it was +# caught by eye. +# +# Prerequisites: setup.hurl must have run (admin user exists). +# +# Run: +# hurl --variables-file tests/api/test.env --file-root tests \ +# --test tests/api/transcode_import.hurl +# ============================================================= + + +# ───────────────────────────────────────────────────────────── +# Step 1 – Login +# ───────────────────────────────────────────────────────────── +POST {{base_url}}/api/auth/login +Content-Type: application/json +{ + "username": "{{username}}", + "password": "{{password}}" +} + +HTTP 200 +[Captures] +token: jsonpath "$.access_token" + + +# ───────────────────────────────────────────────────────────── +# Step 2 – The job is registered, and declares itself correctly. +# +# The admin panel keys its read-only badge and its repair toggle off +# these three fields, so a job that stops declaring them degrades the +# UI silently: `mutates` wrong means a destructive job renders as safe, +# and a missing `repair_description` removes the only way to reach the +# deletion from the interface. +# +# `always` because a plain run inserts rows and writes blobs — the +# repair flag adds deletion on top, which is why the two are +# independent rather than one implying the other. +# ───────────────────────────────────────────────────────────── +GET {{base_url}}/api/admin/jobs +Authorization: Bearer {{token}} + +HTTP 200 +[Asserts] +jsonpath "$[*].name" contains "transcode_import" +jsonpath "$[?(@.name=='transcode_import')].mutates" == "always" +jsonpath "$[?(@.name=='transcode_import')].recoverable" == true + + +# ───────────────────────────────────────────────────────────── +# Step 3 – A discovery-only run over a drained tree. +# +# No `?repair=true`: the default must import without deleting, which is +# what the daily-tick and the no-silent-auto-repair rule both depend +# on. Counters are asserted to exact zeros rather than "exists" — a +# non-zero here would mean the job found work in a directory the API +# cannot have put anything into, which is the shape a re-keying bug +# would take. +# ───────────────────────────────────────────────────────────── +POST {{base_url}}/api/admin/jobs/transcode_import/trigger +Authorization: Bearer {{token}} + +HTTP 200 +[Asserts] +jsonpath "$.ok" == true +jsonpath "$.outcome.outcome" == "ok" +jsonpath "$.outcome.extra.completed" == true +jsonpath "$.outcome.extra.extra_stats.imported" == 0 +jsonpath "$.outcome.extra.extra_stats.negatives" == 0 +jsonpath "$.outcome.extra.extra_stats.file_gone" == 0 +jsonpath "$.outcome.extra.extra_stats.deleted" == 0 +jsonpath "$.outcome.extra.extra_stats.failed" == 0 +jsonpath "$.outcome.extra.extra_stats.unverified" == 0 +# A clean run raises no findings. `file_gone` entries are recorded as +# anomalies, so a non-zero count here would contradict the zero above. +jsonpath "$.outcome.extra.finding_count" == 0 + + +# ───────────────────────────────────────────────────────────── +# Step 4 – The same run under `?repair=true`. +# +# On a drained tree this must be indistinguishable from the run above: +# nothing to import, nothing to delete, and — the part worth pinning — +# no complaint about the directory being absent. The thumbnail teardown +# warned `could not be removed / No such file or directory` on every +# boot after its migration finished, warning about success forever. +# Absence is the end state, not a failure, and the job has to treat it +# that way because this is what every run after the first looks like. +# ───────────────────────────────────────────────────────────── +POST {{base_url}}/api/admin/jobs/transcode_import/trigger?repair=true +Authorization: Bearer {{token}} + +HTTP 200 +[Asserts] +jsonpath "$.ok" == true +jsonpath "$.outcome.outcome" == "ok" +jsonpath "$.outcome.extra.extra_stats.imported" == 0 +jsonpath "$.outcome.extra.extra_stats.negatives" == 0 +jsonpath "$.outcome.extra.extra_stats.deleted" == 0 +jsonpath "$.outcome.extra.extra_stats.failed" == 0 +jsonpath "$.outcome.extra.finding_count" == 0 + + +# ───────────────────────────────────────────────────────────── +# Step 5 – Idempotence. +# +# The job is registered as a daily tick AND named in +# OXICLOUD_STARTUP_JOBS, so on a long-lived install it runs +# unattended, repeatedly, forever. Re-running must stay a no-op — +# anything that accumulated across runs would accumulate unattended. +# ───────────────────────────────────────────────────────────── +POST {{base_url}}/api/admin/jobs/transcode_import/trigger +Authorization: Bearer {{token}} + +HTTP 200 +[Asserts] +jsonpath "$.outcome.outcome" == "ok" +jsonpath "$.outcome.extra.extra_stats.imported" == 0 +jsonpath "$.outcome.extra.extra_stats.negatives" == 0 +jsonpath "$.outcome.extra.finding_count" == 0 diff --git a/tests/fixtures/negative-cache-transcode.png b/tests/fixtures/negative-cache-transcode.png new file mode 100644 index 00000000..e8e64dc8 Binary files /dev/null and b/tests/fixtures/negative-cache-transcode.png differ