diff --git a/src/common/di.rs b/src/common/di.rs index 4630da69..a31c0980 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -1476,6 +1476,24 @@ impl AppServiceFactory { .register_recoverable_job(&core.job_registry, &job_store_provider_dyn) .await; + // Step 10 migration tenant: backfills `content_derived_blobs` from + // the on-disk thumbnail sidecars that predate it. Idempotent, so it + // is safe to trigger repeatedly — Phase 3 (deleting the sidecars) is + // gated on a run reporting zero imported. Registered unconditionally + // rather than behind a flag: a migration nobody can find is a + // migration nobody runs. + // + // `.thumbnails` lives under the storage path, matching + // `ThumbnailService::new(&self.storage_path, …)` above. + let _ = Arc::new( + crate::infrastructure::services::thumb_derived_import_service::ThumbDerivedImport::new( + std::path::Path::new(&self.storage_path).join(".thumbnails"), + core.dedup_service.clone(), + ), + ) + .register_recoverable_job(&core.job_registry, &job_store_provider_dyn) + .await; + // Third recoverable-run tenant. Iterates `storage.files` // and reports parent-folder-trashed cascade misses, // `missing_blob` (data-loss indicator — file references diff --git a/src/infrastructure/services/mod.rs b/src/infrastructure/services/mod.rs index 2fbb00ae..8ff128d6 100644 --- a/src/infrastructure/services/mod.rs +++ b/src/infrastructure/services/mod.rs @@ -57,6 +57,7 @@ pub mod session_liveness_gauges; pub mod share_unlock_cookie; pub mod smtp_email_sender; pub mod swappable_blob_backend; +pub mod thumb_derived_import_service; pub mod thumbnail_service; #[cfg(test)] mod thumbnail_service_test; diff --git a/src/infrastructure/services/thumb_derived_import_service.rs b/src/infrastructure/services/thumb_derived_import_service.rs new file mode 100644 index 00000000..c01f7e31 --- /dev/null +++ b/src/infrastructure/services/thumb_derived_import_service.rs @@ -0,0 +1,313 @@ +//! `thumb_derived_import` — backfill `storage.content_derived_blobs` from the +//! on-disk thumbnail sidecars that predate it. +//! +//! Step 10 of `docs/plan/derived-blobs.md`. Every server-rendered thumbnail +//! written before `content_derived_blobs` existed lives only as +//! `{thumbnails_root}/{size}/{hash}.webp`. That is local-disk state: another +//! instance cannot see it, a backend migration does not carry it, and no +//! consistency job covers it. This job moves those bytes into the blob store +//! and records the mapping, after which the derived tier can become +//! authoritative and the sidecar can be deleted. +//! +//! **Thumbnails only, and that is permanent.** The table also holds +//! `kind = 'transcode'`, but transcoding lands *after* this migration, so +//! transcodes are born into the table and never pass through a sidecar era. +//! This job will not grow a transcode arm. +//! +//! ### Idempotent by construction +//! +//! Each file is skipped when a row already exists for its +//! `(source_hash, 'thumbnail', variant)`, and `store_derived_blob` is +//! `ON CONFLICT DO NOTHING` with a release-on-conflict underneath, so a +//! re-run cannot inflate refcounts. Re-running is the expected operator +//! behaviour — Phase 3 (deleting the sidecars) is gated on a run reporting +//! zero imported. +//! +//! ### Multi-instance caveat +//! +//! Sidecars are local. Running this on one instance migrates only that +//! instance's files, so Phase 3 must be gated on *every* instance reporting +//! an empty tail. The run history does not aggregate across instances; that +//! remains an operator responsibility. + +use std::path::PathBuf; +use std::sync::Arc; + +use async_trait::async_trait; +use bytes::Bytes; +use tokio::fs; + +use crate::application::ports::thumbnail_ports::ThumbnailSize; +use crate::infrastructure::scheduler::{ + JobRegistry, JobRunArgs, JobStore, JobStoreProvider, RecoverableJobHandler, RunOutcome, + RunStatus, record_or_log, +}; +use crate::infrastructure::services::dedup_service::DedupService; + +pub const THUMB_DERIVED_IMPORT_JOB_NAME: &str = "thumb_derived_import"; + +/// Files handled between checkpoints. Each one is a read plus (at most) a +/// blob write, so this is deliberately smaller than a pure-DB sweep's page. +const BATCH_SIZE: usize = 100; + +pub struct ThumbDerivedImport { + thumbnails_root: PathBuf, + dedup: Arc, +} + +impl ThumbDerivedImport { + pub fn new(thumbnails_root: PathBuf, dedup: Arc) -> Self { + Self { + thumbnails_root, + dedup, + } + } + + pub async fn register_recoverable_job( + self: Arc, + registry: &JobRegistry, + provider: &Arc, + ) -> Arc { + registry + .register_recoverable_job(self.clone(), provider.clone(), None) + .await; + self + } + + /// The hash a sidecar filename names, or `None` when the file is not one. + /// + /// Strict, and deliberately rejects `ext-{file_id}.jpg`: those are + /// user-supplied, file-keyed bytes. Importing them here would content-key + /// them and share one user's uploaded preview onto every file with + /// identical content — the poisoning `file_attached_blobs` exists to + /// prevent. They belong to `thumb_attached_import`. + fn hash_from_sidecar_name(name: &str) -> Option<&str> { + let stem = name.strip_suffix(".webp")?; + if stem.len() != 64 || !stem.chars().all(|c| c.is_ascii_hexdigit()) { + return None; + } + Some(stem) + } + + /// Sorted sidecar filenames for one size directory. + /// + /// Sorted so the cursor is meaningful: resume skips everything at or + /// before it, which only works over a stable order. + async fn sidecar_names(&self, size: ThumbnailSize) -> Vec { + let dir = self.thumbnails_root.join(size.dir_name()); + 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() + && Self::hash_from_sidecar_name(name).is_some() + { + names.push(name.to_string()); + } + } + names.sort(); + names + } +} + +#[async_trait] +impl RecoverableJobHandler for ThumbDerivedImport { + fn name(&self) -> &str { + THUMB_DERIVED_IMPORT_JOB_NAME + } + + async fn count_total(&self) -> Option { + let mut total = 0u64; + for size in ThumbnailSize::all() { + total += self.sidecar_names(*size).await.len() as u64; + } + Some(total) + } + + async fn run_resumable( + &self, + store: &dyn JobStore, + _args: &JobRunArgs, + resume_cursor: Option>, + ) -> RunOutcome { + // Cursor is `{size_dir}/{filename}` — the last file completed. Sizes + // are walked in `ThumbnailSize::all()` order, and names are sorted + // within each, so the pair totally orders the walk. + 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 mut imported = 0u64; + let mut already = 0u64; + let mut failed = 0u64; + let mut since_checkpoint = 0usize; + let variant_of = |s: ThumbnailSize| s.dir_name().to_string(); + + for size in ThumbnailSize::all() { + let dir_name = variant_of(*size); + for name in self.sidecar_names(*size).await { + let position = format!("{dir_name}/{name}"); + + // Resume: everything at or before the cursor is done. + if let Some(c) = &cursor + && position.as_str() <= c.as_str() + { + continue; + } + + match store.status().await { + Ok(RunStatus::CancelRequested) => { + return RunOutcome::Paused { + cursor: position.into_bytes(), + }; + } + Ok(_) => {} + Err(e) => { + return RunOutcome::Failed { + message: format!("status poll: {e}"), + }; + } + } + + let Some(hash) = Self::hash_from_sidecar_name(&name) else { + continue; + }; + + // Already mapped — the common case on a re-run, and the + // reason this job is safe to trigger repeatedly. + if self + .dedup + .find_derived_blob(hash, "thumbnail", &dir_name) + .await + .is_some() + { + already += 1; + } else { + let path = self.thumbnails_root.join(&dir_name).join(&name); + match fs::read(&path).await { + Ok(data) => { + match self + .dedup + .store_derived_blob( + hash, + "thumbnail", + &dir_name, + "image/webp", + Bytes::from(data), + ) + .await + { + Ok(_) => imported += 1, + Err(e) => { + failed += 1; + record_or_log( + store, + THUMB_DERIVED_IMPORT_JOB_NAME, + "thumbnail_import_failed", + "anomaly", + None, + serde_json::json!({ + "path": position, + "hash": hash, + "error": format!("{e}"), + "note": "sidecar left in place; safe to re-run", + }), + ) + .await; + } + } + } + Err(e) => { + // Unreadable, or removed between listing and read + // (a concurrent GC unlink). Neither is fatal. + failed += 1; + record_or_log( + store, + THUMB_DERIVED_IMPORT_JOB_NAME, + "thumbnail_unreadable", + "anomaly", + None, + serde_json::json!({ + "path": position, + "error": format!("{e}"), + }), + ) + .await; + } + } + } + + since_checkpoint += 1; + if since_checkpoint >= BATCH_SIZE { + if let Err(e) = store + .checkpoint(position.clone().into_bytes(), since_checkpoint as u64) + .await + { + return RunOutcome::Failed { + message: format!("checkpoint: {e}"), + }; + } + since_checkpoint = 0; + } + } + } + + tracing::info!( + target: "oxicloud::dedup", + event = "thumb_derived_import.completed", + run_id = %store.run_id(), + imported = imported, + already_present = already, + failed = failed, + "thumb_derived_import: {imported} imported, {already} already present, {failed} failed" + ); + + RunOutcome::completed() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + const H: &str = "0a1b2c3d4e5f60718293a4b5c6d7e8f90a1b2c3d4e5f60718293a4b5c6d7e8f9"; + + #[test] + fn accepts_a_canonical_sidecar_name() { + assert_eq!( + ThumbDerivedImport::hash_from_sidecar_name(&format!("{H}.webp")), + Some(H) + ); + } + + /// `ext-` files are user-supplied and file-keyed. Importing one here + /// would content-key it and share it across every file with identical + /// content — the exact poisoning the table split prevents. + #[test] + fn rejects_external_and_malformed_names() { + for name in [ + format!("ext-{H}.jpg"), + "ext-3f2b1c00-0000-0000-0000-000000000000.jpg".to_string(), + format!("{H}.jpg"), + format!("{}.webp", &H[..63]), + H.to_string(), + "junk.webp".to_string(), + ] { + assert_eq!( + ThumbDerivedImport::hash_from_sidecar_name(&name), + None, + "must not be imported as a derived thumbnail: {name}" + ); + } + } +}