feat(storage): audit every sidecar deletion, and reclaim orphaned uploads

Two changes to the import jobs' destructive path.

thumb_attached_import now deletes orphaned sidecars under `repair`,
matching the dead-source case on the derived side. An `ext-` file whose
owner is gone is unimportable — the FK on file_id would reject the row —
so leaving it means it is rediscovered every run, the tail never empties
and step 10e's gate never opens. Safe despite these being the
non-regenerable bytes: the preview is keyed to a file_id that no longer
exists, so nothing can reference it again. Unrecoverable and unreachable
are different things, and this is both.

And every deletion is now audited. A one-way migration removing
user-visible files should leave a trail that outlives the run history:
findings are per-run and get purged, whereas target: "audit" is
separable and retained. If a preview later turns out to be missing, this
is the only record saying the migration removed it and when.

`owner` carries the id the file belonged to — source_hash for
content-keyed, file_id for uploaded — because that is where an
investigation starts, and the raw logs cannot supply it: NEW BLOB names
the hash of the STORED BYTES, a different value from the sidecar's own
name, which is why grepping one against the other finds nothing.

reason is a stable key: `imported` (replaced by a verified blob),
`source_gone`, `orphaned`. The first lives inside verify_and_unlink so a
verified deletion cannot be logged inconsistently; the other two are
explicit, since those paths have nothing to verify against.

src/infrastructure/services/thumb_attached_import_service.rs

@@ -262,6 +262,8 @@ impl RecoverableJobHandler for ThumbAttachedImport {
                         let path = self.thumbnails_root.join(&dir_name).join(&name);
                         if ThumbDerivedImport::verify_and_unlink(
                             &self.dedup,
+                            THUMB_ATTACHED_IMPORT_JOB_NAME,
+                            &file_id_str,
                             &existing.blob_hash,
                             &path,
                         )
@@ -286,24 +288,59 @@ impl RecoverableJobHandler for ThumbAttachedImport {
                         }
                     }
                 } else if !self.file_exists(file_id).await {
-                    // The file is gone; the sidecar outlived it. Reported
+                    // The file is gone, so this sidecar is unimportable: the
-                    // rather than deleted — this job imports, it does not
+                    // FK on `file_id` would reject the row. Mirrors the
-                    // reclaim, and a destructive default on a migration is
+                    // dead-source case in thumb_derived_import.
-                    // exactly what `no silent auto-repair` forbids.
+                    //
+                    // Reported by default — a destructive default on a
+                    // migration is what no-silent-auto-repair forbids — and
+                    // deleted under `repair`, because otherwise it is
+                    // rediscovered on every run, the tail never empties, and
+                    // step 10e's gate never opens.
+                    //
+                    // Safe to delete despite these being the non-regenerable
+                    // bytes: the preview is keyed to a `file_id` that no
+                    // longer exists, so nothing can ever reference it again.
+                    // Unrecoverable and unreachable are different things, and
+                    // this is both.
+                    //
+                    // No readback before unlinking, unlike the imported path:
+                    // there is no row and no blob to read back, and nothing to
+                    // regenerate from either.
                     orphaned += 1;
-                    record_or_log(
+                    if delete_imported {
-                        store,
+                        let path = self.thumbnails_root.join(&dir_name).join(&name);
-                        THUMB_ATTACHED_IMPORT_JOB_NAME,
+                        if fs::remove_file(&path).await.is_ok() {
-                        "attached_sidecar_orphan",
+                            deleted += 1;
-                        "anomaly",
+                            // Explicit: nothing to verify against, so this
-                        None,
+                            // bypasses verify_and_unlink. Worth auditing
-                        serde_json::json!({
+                            // loudest of all — these bytes were
-                            "path":    position,
+                            // user-supplied and cannot be regenerated, even
-                            "file_id": file_id_str,
+                            // though the file that owned them is gone.
-                            "note":    "no storage.files row; sidecar left in place for the operator",
+                            crate::infrastructure::services::thumb_derived_import_service::audit_sidecar_deleted(
-                        }),
+                                THUMB_ATTACHED_IMPORT_JOB_NAME,
-                    )
+                                "orphaned",
-                    .await;
+                                &file_id_str,
+                                "-",
+                                &path,
+                            );
+                        }
+                    } else {
+                        record_or_log(
+                            store,
+                            THUMB_ATTACHED_IMPORT_JOB_NAME,
+                            "attached_sidecar_orphan",
+                            "anomaly",
+                            None,
+                            serde_json::json!({
+                                "path":    position,
+                                "file_id": file_id_str,
+                                "note":    "no storage.files row; unimportable, and deleted on a \
+                                            repair run since nothing can reference it again",
+                            }),
+                        )
+                        .await;
+                    }
                 } else {
                     let path = self.thumbnails_root.join(&dir_name).join(&name);
                     match fs::read(&path).await {
@@ -328,6 +365,8 @@ impl RecoverableJobHandler for ThumbAttachedImport {
                                     if delete_imported {
                                         if ThumbDerivedImport::verify_and_unlink(
                                             &self.dedup,
+                                            THUMB_ATTACHED_IMPORT_JOB_NAME,
+                                            &file_id_str,
                                             &attached_hash,
                                             &path,
                                         )

src/infrastructure/services/thumb_derived_import_service.rs

@@ -48,6 +48,41 @@ use crate::infrastructure::services::dedup_service::DedupService;
 
 pub const THUMB_DERIVED_IMPORT_JOB_NAME: &str = "thumb_derived_import";
 
+/// Record a sidecar deletion on the audit channel.
+///
+/// Both import jobs delete user-visible files during a one-way migration, so
+/// the trail has to survive the run history: findings are per-run and get
+/// purged, whereas `target: "audit"` is separable and retained. If a preview
+/// later turns out to be missing, this is the only record that says the
+/// migration removed it, when, and on whose behalf.
+///
+/// `owner` is the id the file belonged to — a `source_hash` for content-keyed
+/// sidecars, a `file_id` for uploaded ones. That is the field an
+/// investigation starts from, and the raw `NEW BLOB` logs cannot supply it:
+/// they name the hash of the stored bytes, which is a different value from
+/// the sidecar's own name.
+///
+/// `reason` is a stable machine-readable key, per the convention: `imported`
+/// (replaced by a verified blob), `source_gone`, `orphaned`.
+pub(crate) fn audit_sidecar_deleted(
+    job: &str,
+    reason: &str,
+    owner: &str,
+    blob_hash: &str,
+    path: &std::path::Path,
+) {
+    tracing::info!(
+        target: "audit",
+        event = "thumbnail.sidecar_deleted",
+        reason = reason,
+        job = job,
+        owner = owner,
+        blob_hash = blob_hash,
+        path = %path.display(),
+        "👮🏻‍♂️ migration deleted a thumbnail sidecar ({reason})",
+    );
+}
+
 /// 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;
@@ -142,6 +177,8 @@ impl ThumbDerivedImport {
     /// and two copies of that rule would be two chances to weaken one.
     pub(crate) async fn verify_and_unlink(
         dedup: &DedupService,
+        job: &str,
+        owner: &str,
         stored_hash: &str,
         path: &std::path::Path,
     ) -> bool {
@@ -154,7 +191,11 @@ impl ThumbDerivedImport {
         if stored.is_empty() || stored.len() as u64 != meta.len() {
             return false;
         }
-        fs::remove_file(path).await.is_ok()
+        if fs::remove_file(path).await.is_err() {
+            return false;
+        }
+        audit_sidecar_deleted(job, "imported", owner, stored_hash, path);
+        true
     }
 
     /// Sorted sidecar filenames for one size directory.
@@ -299,7 +340,15 @@ impl RecoverableJobHandler for ThumbDerivedImport {
                     already += 1;
                     if delete_imported {
                         let path = self.thumbnails_root.join(dir_name).join(&name);
-                        if Self::verify_and_unlink(&self.dedup, &existing.blob_hash, &path).await {
+                        if Self::verify_and_unlink(
+                            &self.dedup,
+                            THUMB_DERIVED_IMPORT_JOB_NAME,
+                            hash,
+                            &existing.blob_hash,
+                            &path,
+                        )
+                        .await
+                        {
                             deleted += 1;
                         } else {
                             unverified += 1;
@@ -345,6 +394,16 @@ impl RecoverableJobHandler for ThumbDerivedImport {
                         let path = self.thumbnails_root.join(dir_name).join(&name);
                         if fs::remove_file(&path).await.is_ok() {
                             deleted += 1;
+                            // Audited explicitly: this unlink bypasses
+                            // verify_and_unlink, which has nothing to verify
+                            // against here.
+                            audit_sidecar_deleted(
+                                THUMB_DERIVED_IMPORT_JOB_NAME,
+                                "source_gone",
+                                hash,
+                                "-",
+                                &path,
+                            );
                         }
                     } else {
                         record_or_log(
@@ -382,6 +441,8 @@ impl RecoverableJobHandler for ThumbDerivedImport {
                                     if delete_imported {
                                         if Self::verify_and_unlink(
                                             &self.dedup,
+                                            THUMB_DERIVED_IMPORT_JOB_NAME,
+                                            hash,
                                             &derived_hash,
                                             &path,
                                         )