fix: reconcile and expose failed product cleanup

backend/service/product_archive_store.py

@@ -23,6 +23,9 @@ def _now() -> datetime:
     return datetime.now(timezone.utc)
 
 
+PURGE_CLAIM_TIMEOUT = timedelta(hours=1)
+
+
 class ProductPurgeInProgressError(ValueError):
     """Raised when restore loses the atomic race to physical purge."""
 
@@ -124,6 +127,7 @@ class ProductArchiveStore:
             self._ensure_column(conn, "product_wordcloud_archives", "source_job_id", "TEXT NOT NULL DEFAULT ''")
             self._ensure_column(conn, "product_wordcloud_archives", "source_asset_id", "TEXT NOT NULL DEFAULT ''")
             self._ensure_column(conn, "product_wordcloud_archives", "db_checksum", "TEXT NOT NULL DEFAULT ''")
+            self._ensure_column(conn, "cleanup_records", "claimed_at", "TEXT")
 
     @staticmethod
     def _ensure_column(conn: sqlite3.Connection, table: str, column: str, declaration: str) -> None:
@@ -399,16 +403,17 @@ class ProductArchiveStore:
             if cleanup is None:
                 return None
             updated = conn.execute(
-                """UPDATE cleanup_records SET status = 'purging'
+                """UPDATE cleanup_records SET status = 'purging', claimed_at = ?
                    WHERE cleanup_id = ? AND status = 'pending_cleanup'""",
-                (cleanup["cleanup_id"],),
+                (now.isoformat(), cleanup["cleanup_id"]),
             )
             if updated.rowcount != 1:
                 return None
             return self._product_from_row(row)
 
-    def fail_product_purge(self, product_id: str) -> ProductRecord:
+    def fail_product_purge(self, product_id: str, now: datetime | None = None) -> ProductRecord:
         """Quarantine a purge whose physical deletion failed mid-way."""
+        failed_at = now or _now()
         with self._lock, self._connect() as conn:
             conn.execute("BEGIN IMMEDIATE")
             claim = conn.execute(
@@ -427,10 +432,58 @@ class ProductArchiveStore:
                 raise ValueError("product purge is not claimed")
             conn.execute(
                 "UPDATE products SET status = 'failed_cleanup', updated_at = ? WHERE product_id = ?",
-                (_now().isoformat(), product_id),
+                (failed_at.isoformat(), product_id),
             )
         return self.get_product(product_id)
 
+    def reconcile_stalled_purges(
+        self,
+        now: datetime,
+        timeout: timedelta = PURGE_CLAIM_TIMEOUT,
+    ) -> list[ProductRecord]:
+        """Move abandoned purge claims to failed_cleanup for manual inspection."""
+        cutoff = now - timeout
+        with self._lock, self._connect() as conn:
+            conn.execute("BEGIN IMMEDIATE")
+            rows = conn.execute(
+                """SELECT product_id FROM cleanup_records
+                   WHERE status = 'purging'
+                     AND claimed_at IS NOT NULL
+                     AND claimed_at <= ?""",
+                (cutoff.isoformat(),),
+            ).fetchall()
+            affected: list[str] = []
+            for row in rows:
+                product_id = row["product_id"]
+                product_row = conn.execute(
+                    """SELECT 1 FROM products WHERE product_id = ? AND status = 'pending_cleanup'""",
+                    (product_id,),
+                ).fetchone()
+                if product_row is None:
+                    continue
+                updated = conn.execute(
+                    """UPDATE cleanup_records SET status = 'failed_cleanup'
+                       WHERE product_id = ? AND status = 'purging' AND claimed_at <= ?""",
+                    (product_id, cutoff.isoformat()),
+                )
+                if updated.rowcount != 1:
+                    continue
+                conn.execute(
+                    """UPDATE products SET status = 'failed_cleanup', updated_at = ?
+                       WHERE product_id = ? AND status = 'pending_cleanup'""",
+                    (now.isoformat(), product_id),
+                )
+                affected.append(product_id)
+        return [self.get_product(product_id) for product_id in affected]
+
+    def failed_product_cleanups(self) -> list[ProductRecord]:
+        rows = self._fetchall(
+            """SELECT * FROM products WHERE status = ?
+               ORDER BY purge_after, product_id""",
+            ("failed_cleanup",),
+        )
+        return [self._product_from_row(row) for row in rows]
+
     def finalize_product_purge(self, product_id: str, now: datetime) -> ProductRecord:
         """Atomically mark a claimed product purged after file removal."""
         with self._lock, self._connect() as conn: