fix: quarantine failed product cleanups in failed_cleanup state

backend/service/cleanup_service.py

@@ -126,7 +126,9 @@ class CleanupService:
                 if product_dir.exists():
                     shutil.rmtree(product_dir)
             except Exception:
-                self.product_store.release_product_purge(product.product_id)
+                # rmtree may have removed only part of the archive. Quarantine
+                # the record for manual inspection instead of retrying it.
+                self.product_store.fail_product_purge(product.product_id)
                 raise
             self.product_store.finalize_product_purge(product.product_id, now=now_utc)
             report.deleted_product_ids.append(product.product_id)

backend/service/product_archive_service.py

@@ -21,7 +21,7 @@ from .schemas import ProductVersionRecord
 
 
 class ProductPendingCleanupError(ValueError):
-    """Raised when a soft-deleted product is changed before restoration."""
+    """Raised when a pending or failed-cleanup product is changed before recovery."""
 
 
 class ProductArchiveService:
@@ -83,7 +83,7 @@ class ProductArchiveService:
         if not isinstance(document, dict):
             raise ValueError("document must be a JSON object")
         product = self.store.get_product(product_id)
-        if product.status == "pending_cleanup":
+        if product.status in ("pending_cleanup", "failed_cleanup"):
             raise ProductPendingCleanupError("product is pending cleanup")
         if product.status != "active":
             raise ValueError("product is not active")

backend/service/product_archive_store.py

@@ -318,7 +318,9 @@ class ProductArchiveStore:
             raise ValueError(f"unknown version_id for product: {version_id}")
 
     def mark_pending_cleanup(self, product_id: str, now: datetime) -> ProductRecord:
-        self.get_product(product_id)
+        product = self.get_product(product_id)
+        if product.status not in ("active", "failed_cleanup"):
+            raise ValueError("product cannot be marked pending cleanup")
         purge_after = now + timedelta(days=30)
         self._execute(
             "UPDATE products SET status = ?, purge_after = ?, updated_at = ? WHERE product_id = ?",
@@ -355,8 +357,8 @@ class ProductArchiveStore:
             )
             conn.execute(
                 """UPDATE cleanup_records SET status = ?, completed_at = ?
-                   WHERE product_id = ? AND status = ?""",
-                ("restored", restored_at.isoformat(), product_id, "pending_cleanup"),
+                   WHERE product_id = ? AND status IN ('pending_cleanup', 'failed_cleanup')""",
+                ("restored", restored_at.isoformat(), product_id),
             )
         return self.get_product(product_id)
 
@@ -405,13 +407,29 @@ class ProductArchiveStore:
                 return None
             return self._product_from_row(row)
 
-    def release_product_purge(self, product_id: str) -> None:
-        """Return a failed physical deletion claim to its pending state."""
-        self._execute(
-            """UPDATE cleanup_records SET status = 'pending_cleanup'
+    def fail_product_purge(self, product_id: str) -> ProductRecord:
+        """Quarantine a purge whose physical deletion failed mid-way."""
+        with self._lock, self._connect() as conn:
+            conn.execute("BEGIN IMMEDIATE")
+            claim = conn.execute(
+                """SELECT 1 FROM cleanup_records
+                   WHERE product_id = ? AND status = 'purging' LIMIT 1""",
+                (product_id,),
+            ).fetchone()
+            if claim is None:
+                raise ValueError("product purge is not claimed")
+            updated = conn.execute(
+                """UPDATE cleanup_records SET status = 'failed_cleanup'
                    WHERE product_id = ? AND status = 'purging'""",
                 (product_id,),
             )
+            if updated.rowcount != 1:
+                raise ValueError("product purge is not claimed")
+            conn.execute(
+                "UPDATE products SET status = 'failed_cleanup', updated_at = ? WHERE product_id = ?",
+                (_now().isoformat(), product_id),
+            )
+        return self.get_product(product_id)
 
     def finalize_product_purge(self, product_id: str, now: datetime) -> ProductRecord:
         """Atomically mark a claimed product purged after file removal."""

backend/service/schemas.py

@@ -43,7 +43,7 @@ class ProductRecord(BaseModel):
     name: str
     sku: str = ""
     specification: str = ""
-    status: Literal["active", "pending_cleanup", "purged"] = "active"
+    status: Literal["active", "pending_cleanup", "failed_cleanup", "purged"] = "active"
     created_at: datetime
     updated_at: datetime
     purge_after: datetime | None = None

backend/tests/test_cleanup_service.py

@@ -197,7 +197,9 @@ def test_atomic_purge_claim_blocks_restore_before_product_files_are_deleted(
     assert service.product_store.get_product(product.product_id).status == "purged"
 
 
-def test_failed_product_file_deletion_releases_claim_back_to_pending(tmp_path, monkeypatch):
+def test_failed_product_file_deletion_enters_failed_cleanup_and_is_not_retried(
+    tmp_path, monkeypatch
+):
     service = make_cleanup_service(tmp_path)
     product = create_pending_product(service, purge_after=NOW)
 
@@ -209,11 +211,47 @@ def test_failed_product_file_deletion_releases_claim_back_to_pending(tmp_path, m
     with pytest.raises(OSError, match="disk failure"):
         service.apply(now=NOW)
 
-    assert service.product_store.get_product(product.product_id).status == "pending_cleanup"
-    assert service.product_store._fetchone(
+    record = service.product_store.get_product(product.product_id)
+    cleanup = service.product_store._fetchone(
         "SELECT status FROM cleanup_records WHERE product_id = ?", (product.product_id,)
-    )["status"] == "pending_cleanup"
-    assert service.product_store.due_product_cleanups(NOW)[0].product_id == product.product_id
+    )["status"]
+
+    assert record.status == "failed_cleanup"
+    assert record.purge_after == NOW
+    assert cleanup == "failed_cleanup"
+    assert service.product_store.due_product_cleanups(NOW) == []
+    assert (service.product_store.root.parent / product.product_id).exists()
+
+
+def test_failed_cleanup_product_can_be_restored_or_requeued(tmp_path, monkeypatch):
+    service = make_cleanup_service(tmp_path)
+
+    def _fail_purge(purge_after: datetime) -> str:
+        product = create_pending_product(service, purge_after=purge_after)
+        monkeypatch.setattr(
+            "service.cleanup_service.shutil.rmtree",
+            lambda *_args, **_kwargs: (_ for _ in ()).throw(OSError("disk failure")),
+        )
+        with pytest.raises(OSError):
+            service.apply(now=purge_after)
+        assert service.product_store.get_product(product.product_id).status == "failed_cleanup"
+        return product.product_id
+
+    failed = _fail_purge(NOW)
+    restored = service.product_store.restore_product(failed, now=NOW + timedelta(days=1))
+    assert restored.status == "active"
+    assert restored.purge_after is None
+
+    failed_requeue = _fail_purge(NOW)
+    requeued = service.product_store.mark_pending_cleanup(
+        failed_requeue, now=NOW + timedelta(days=1)
+    )
+    assert requeued.status == "pending_cleanup"
+    assert requeued.purge_after == NOW + timedelta(days=31)
+    assert service.product_store.due_product_cleanups(NOW + timedelta(days=5)) == []
+    assert service.product_store.due_product_cleanups(NOW + timedelta(days=31)) == [
+        requeued
+    ]
 
 
 def test_apply_rechecks_archive_reference_immediately_before_job_deletion(tmp_path):

backend/tests/test_product_archive.py

@@ -3,6 +3,7 @@ import hashlib
 import sqlite3
 import sys
 import shutil
+from datetime import datetime
 from io import BytesIO
 from pathlib import Path
 from types import SimpleNamespace
@@ -327,6 +328,34 @@ def test_product_routes_require_orders_auth(product_archive_client):
     assert [response.status_code for response in requests] == [403] * 6
 
 
+def test_archive_version_rejects_failed_cleanup_product_and_status_is_visible(
+    product_archive_client,
+):
+    product = create_product(product_archive_client)
+    deleted = product_archive_client.delete(
+        f"/api/products/{product['product_id']}", headers=orders_auth_header()
+    ).json()
+    store = product_archive_client.product_store
+    purge_after = datetime.fromisoformat(deleted["purge_after"])
+    store.claim_product_purge(deleted["product_id"], purge_after)
+    store.fail_product_purge(deleted["product_id"])
+
+    listed = product_archive_client.get(
+        f"/api/products/{deleted['product_id']}", headers=orders_auth_header()
+    ).json()
+    assert listed["status"] == "failed_cleanup"
+
+    response = archive_product_version(
+        product_archive_client, deleted["product_id"], {"elements": []}
+    )
+    assert response.status_code == 409
+
+    restored = product_archive_client.post(
+        f"/api/products/{deleted['product_id']}/restore", headers=orders_auth_header()
+    ).json()
+    assert restored["status"] == "active"
+
+
 def test_archive_failure_during_final_move_removes_staging_files(product_archive_client, prepared_wordcloud_job, monkeypatch):
     product = create_product(product_archive_client)
     service = service_app._product_archive_service()

backend/tests/test_product_archive_store.py

@@ -111,6 +111,49 @@ def test_restore_rejects_product_after_purge_has_deleted_its_files(tmp_path):
     assert store.get_product(product.product_id).status == "purged"
 
 
+def test_fail_product_purge_requires_an_active_claim_and_preserves_purge_after(tmp_path):
+    store = ProductArchiveStore(tmp_path / "service_products")
+    product = store.upsert_product(ProductInput("manual", None, "失败清理笔盒"))
+    store.mark_pending_cleanup(product.product_id, now=NOW)
+
+    with pytest.raises(ValueError, match="claim"):
+        store.fail_product_purge(product.product_id)
+
+    due = NOW + timedelta(days=30)
+    store.claim_product_purge(product.product_id, due)
+    failed = store.fail_product_purge(product.product_id)
+
+    assert failed.status == "failed_cleanup"
+    assert failed.purge_after == due
+    cleanup = store._fetchone(
+        "SELECT status FROM cleanup_records WHERE product_id = ?", (product.product_id,)
+    )["status"]
+    assert cleanup == "failed_cleanup"
+    assert store.due_product_cleanups(due) == []
+    with pytest.raises(ValueError):
+        store.purge_product(product.product_id, now=due + timedelta(days=1))
+
+
+def test_mark_pending_cleanup_accepts_failed_product_and_rejects_purged(tmp_path):
+    store = ProductArchiveStore(tmp_path / "service_products")
+    product = store.upsert_product(ProductInput("manual", None, "重新排队笔盒"))
+    store.mark_pending_cleanup(product.product_id, now=NOW)
+    store.claim_product_purge(product.product_id, NOW + timedelta(days=30))
+    store.fail_product_purge(product.product_id)
+
+    requeued = store.mark_pending_cleanup(product.product_id, now=NOW + timedelta(days=1))
+    assert requeued.status == "pending_cleanup"
+    assert requeued.purge_after == NOW + timedelta(days=31)
+
+    purged_product = store.upsert_product(ProductInput("manual", None, "已删除笔盒"))
+    store.mark_pending_cleanup(purged_product.product_id, now=NOW)
+    store.purge_product(purged_product.product_id, now=NOW + timedelta(days=30))
+    with pytest.raises(ValueError, match="cannot"):
+        store.mark_pending_cleanup(
+            purged_product.product_id, now=NOW + timedelta(days=31)
+        )
+
+
 def test_version_bound_operations_reject_unknown_or_mismatched_versions(tmp_path):
     store = ProductArchiveStore(tmp_path / "service_products")
     first = store.upsert_product(ProductInput("manual", None, "第一个笔盒"))