From 0e09cb81ff3c067f204615215ec70a542811aa32 Mon Sep 17 00:00:00 2001 From: Edouard Vanbelle Date: Sun, 30 Aug 2026 15:06:29 +0200 Subject: [PATCH] feat(transcode): transcode_import drains .transcoded/, re-keying as it goes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The twin of thumb_derived_import, with the difference that shapes the whole job: the legacy tree is keyed by FILE (`{file_id}.webp`) while the destination is keyed by CONTENT. Thumbnail sidecars were already named by blob hash, so importing them was a move; every entry here has to be resolved through storage.files first. That re-keying is the point rather than bookkeeping. A sandbox with five .skip markers had three of them naming the same image, so the file-keyed tree stored one verdict three times. After the import it is one row, and any future upload of those bytes inherits it instead of paying for the decision again. Both artifact kinds are claimed by one walk: `{id}.webp` becomes a derived Blob, `{id}.webp.skip` becomes a negative row. They share a source file and a cursor, so splitting them into two passes would be two chances for the pair to disagree about what had been handled. `.skip` is matched BEFORE `.webp` — the shorter suffix matches a marker too, and getting that backwards would read a zero-byte file and store it as the transcode of its source, then serve it to clients. There is a test. Entries whose file is gone cannot be re-keyed at all, so they are reported and, under repair, deleted: unimportable by definition, and a run that keeps rediscovering them never reports zero, so the gate for removing the directory never opens. Deletion reuses verify_and_unlink, so a cached transcode is removed only after its stored replacement reads back byte-identical. Directory removal follows the same rule as .thumbnails/ — delete, and only rename aside if a non-cache file is in the way. The batch checkpoint is a shared helper rather than inline at both exits of the loop body. The first draft duplicated it and dropped the future on the skip path without awaiting: the entry counted toward the batch, the cursor never advanced, and a resumed run would have rewalked everything it had already handled. Co-Authored-By: Claude Opus 5 (1M context) --- src/common/di.rs | 18 + src/infrastructure/services/mod.rs | 1 + .../services/transcode_import_service.rs | 560 ++++++++++++++++++ 3 files changed, 579 insertions(+) create mode 100644 src/infrastructure/services/transcode_import_service.rs diff --git a/src/common/di.rs b/src/common/di.rs index 73e98ee1..8daa9324 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -1505,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/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/transcode_import_service.rs b/src/infrastructure/services/transcode_import_service.rs new file mode 100644 index 00000000..83c29d38 --- /dev/null +++ b/src/infrastructure/services/transcode_import_service.rs @@ -0,0 +1,560 @@ +//! 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 { + // Daily, matching the thumbnail imports: idempotent and resumable, + // so periodic is safe, and a migration nobody remembers to trigger + // never finishes. The tick imports but does not delete — `repair` + // defaults false. + registry + .register_recoverable_job( + self.clone(), + provider.clone(), + Some(std::time::Duration::from_secs(24 * 3600)), + ) + .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; + + 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; + if let Some(failure) = checkpoint_if_due(store, &name, &mut since_checkpoint).await { + return failure; + } + } + + // 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); + } +}