diff --git a/backend/service/cleanup_service.py b/backend/service/cleanup_service.py index 0abb95d..57b0749 100644 --- a/backend/service/cleanup_service.py +++ b/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) diff --git a/backend/service/product_archive_service.py b/backend/service/product_archive_service.py index 5a0a7ed..21396c9 100644 --- a/backend/service/product_archive_service.py +++ b/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") diff --git a/backend/service/product_archive_store.py b/backend/service/product_archive_store.py index 7a2caae..3975e79 100644 --- a/backend/service/product_archive_store.py +++ b/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' - WHERE product_id = ? AND status = 'purging'""", - (product_id,), - ) + 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.""" diff --git a/backend/service/schemas.py b/backend/service/schemas.py index 91e5ee8..7e81cdc 100644 --- a/backend/service/schemas.py +++ b/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 diff --git a/backend/tests/test_cleanup_service.py b/backend/tests/test_cleanup_service.py index 7a6873e..ce30490 100644 --- a/backend/tests/test_cleanup_service.py +++ b/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): diff --git a/backend/tests/test_product_archive.py b/backend/tests/test_product_archive.py index edb11ba..ce5d566 100644 --- a/backend/tests/test_product_archive.py +++ b/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() diff --git a/backend/tests/test_product_archive_store.py b/backend/tests/test_product_archive_store.py index 8ed9aac..b40b3d5 100644 --- a/backend/tests/test_product_archive_store.py +++ b/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, "第一个笔盒"))