diff --git a/backend/service/app.py b/backend/service/app.py index 0428dae..eda35d3 100644 --- a/backend/service/app.py +++ b/backend/service/app.py @@ -1232,6 +1232,8 @@ def delete_product(request: Request, product_id: str) -> ProductRecord: product = _get_product_or_404(product_id) if product.status == "pending_cleanup": raise HTTPException(status_code=409, detail="product is pending cleanup") + if product.status == "failed_cleanup": + raise HTTPException(status_code=409, detail="product cleanup failed; restore it before deleting") try: return product_archive_store.mark_pending_cleanup(product_id, now=datetime.now(timezone.utc)) except ValueError as exc: diff --git a/backend/service/cleanup_service.py b/backend/service/cleanup_service.py index 57b0749..2e8bb5a 100644 --- a/backend/service/cleanup_service.py +++ b/backend/service/cleanup_service.py @@ -27,6 +27,7 @@ class TemporaryJob(BaseModel): class CleanupReport(BaseModel): temporary_jobs: list[TemporaryJob] = Field(default_factory=list) pending_products: list[ProductRecord] = Field(default_factory=list) + failed_products: list[ProductRecord] = Field(default_factory=list) reclaimable_bytes: int = 0 reminder_job_ids: list[str] = Field(default_factory=list) deleted_job_ids: list[str] = Field(default_factory=list) @@ -70,6 +71,7 @@ class CleanupService: def preview(self, now: datetime) -> CleanupReport: now_utc = _as_utc(now) + self.product_store.reconcile_stalled_purges(now_utc) protected = self.archive_protected_job_ids() temporary_jobs: list[TemporaryJob] = [] reminder_job_ids: list[str] = [] @@ -99,6 +101,7 @@ class CleanupService: return CleanupReport( temporary_jobs=temporary_jobs, pending_products=pending_products, + failed_products=self.product_store.failed_product_cleanups(), reclaimable_bytes=sum(item.size_bytes for item in temporary_jobs) + product_bytes, reminder_job_ids=reminder_job_ids, ) diff --git a/backend/service/product_archive_store.py b/backend/service/product_archive_store.py index 3975e79..d21afeb 100644 --- a/backend/service/product_archive_store.py +++ b/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: diff --git a/backend/tests/test_cleanup_service.py b/backend/tests/test_cleanup_service.py index ce30490..462d61e 100644 --- a/backend/tests/test_cleanup_service.py +++ b/backend/tests/test_cleanup_service.py @@ -2,6 +2,7 @@ from __future__ import annotations import json import os +import sqlite3 import sys from datetime import datetime, timedelta, timezone from pathlib import Path @@ -254,6 +255,62 @@ def test_failed_cleanup_product_can_be_restored_or_requeued(tmp_path, monkeypatc ] +def test_stale_purging_claim_is_reconciled_to_failed_cleanup(tmp_path, monkeypatch): + service = make_cleanup_service(tmp_path) + product = create_pending_product(service, purge_after=NOW) + store = service.product_store + + original_fail = store.fail_product_purge + calls = {"n": 0} + + def flaky_fail(product_id, now=None): + calls["n"] += 1 + if calls["n"] == 1: + raise sqlite3.OperationalError("simulated busy") + return original_fail(product_id, now=now) + + monkeypatch.setattr(store, "fail_product_purge", flaky_fail) + monkeypatch.setattr( + "service.cleanup_service.shutil.rmtree", + lambda *_args, **_kwargs: (_ for _ in ()).throw(OSError("disk failure")), + ) + + with pytest.raises(sqlite3.OperationalError, match="simulated busy"): + service.apply(now=NOW) + + stranded = store._fetchone( + "SELECT status, claimed_at FROM cleanup_records WHERE product_id = ?", + (product.product_id,), + ) + assert stranded["status"] == "purging" + assert datetime.fromisoformat(stranded["claimed_at"]) == NOW + + report = service.preview(now=NOW + timedelta(hours=2)) + + assert [item.product_id for item in report.failed_products] == [product.product_id] + assert store.get_product(product.product_id).status == "failed_cleanup" + assert store.due_product_cleanups(NOW + timedelta(days=1)) == [] + + +def test_cleanup_candidates_reports_failed_products(tmp_path, monkeypatch): + service = make_cleanup_service(tmp_path) + product = create_pending_product(service, purge_after=NOW) + store = service.product_store + store.claim_product_purge(product.product_id, NOW) + store.fail_product_purge(product.product_id, now=NOW) + monkeypatch.setattr(service_app, "cleanup_service", service, raising=False) + client = TestClient(service_app.app) + + response = client.get("/api/maintenance/cleanup-candidates", headers=auth_header()) + + assert response.status_code == 200 + data = response.json() + assert [item["product_id"] for item in data["failed_products"]] == [ + product.product_id + ] + assert data["pending_products"] == [] + + def test_apply_rechecks_archive_reference_immediately_before_job_deletion(tmp_path): service = make_cleanup_service(tmp_path) due_job = create_success_job(service, age_days=31) diff --git a/backend/tests/test_product_archive.py b/backend/tests/test_product_archive.py index ce5d566..2953b48 100644 --- a/backend/tests/test_product_archive.py +++ b/backend/tests/test_product_archive.py @@ -356,6 +356,35 @@ def test_archive_version_rejects_failed_cleanup_product_and_status_is_visible( assert restored["status"] == "active" +def test_delete_rejects_failed_cleanup_product_until_restore(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"], now=purge_after) + + rejected = product_archive_client.delete( + f"/api/products/{deleted['product_id']}", headers=orders_auth_header() + ) + assert rejected.status_code == 409 + assert store.get_product(deleted["product_id"]).status == "failed_cleanup" + + restored = product_archive_client.post( + f"/api/products/{deleted['product_id']}/restore", headers=orders_auth_header() + ) + assert restored.status_code == 200 + assert restored.json()["status"] == "active" + + soft_deleted = product_archive_client.delete( + f"/api/products/{deleted['product_id']}", headers=orders_auth_header() + ) + assert soft_deleted.status_code == 200 + assert soft_deleted.json()["status"] == "pending_cleanup" + + 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 b40b3d5..a8e33ec 100644 --- a/backend/tests/test_product_archive_store.py +++ b/backend/tests/test_product_archive_store.py @@ -9,7 +9,10 @@ BACKEND_DIR = Path(__file__).resolve().parents[1] if str(BACKEND_DIR) not in sys.path: sys.path.insert(0, str(BACKEND_DIR)) -from service.product_archive_store import ProductArchiveStore # noqa: E402 +from service.product_archive_store import ( # noqa: E402 + ProductArchiveStore, + ProductPurgeInProgressError, +) from service.schemas import ProductInput # noqa: E402 @@ -121,10 +124,11 @@ def test_fail_product_purge_requires_an_active_claim_and_preserves_purge_after(t due = NOW + timedelta(days=30) store.claim_product_purge(product.product_id, due) - failed = store.fail_product_purge(product.product_id) + failed = store.fail_product_purge(product.product_id, now=due + timedelta(hours=1)) assert failed.status == "failed_cleanup" assert failed.purge_after == due + assert failed.updated_at == due + timedelta(hours=1) cleanup = store._fetchone( "SELECT status FROM cleanup_records WHERE product_id = ?", (product.product_id,) )["status"] @@ -134,6 +138,28 @@ def test_fail_product_purge_requires_an_active_claim_and_preserves_purge_after(t store.purge_product(product.product_id, now=due + timedelta(days=1)) +def test_reconcile_stalled_purges_quarantines_only_claims_older_than_timeout(tmp_path): + store = ProductArchiveStore(tmp_path / "service_products") + stale_product = store.upsert_product(ProductInput("manual", None, "卡住清理笔盒")) + fresh_product = store.upsert_product(ProductInput("manual", None, "正常清理笔盒")) + for product in (stale_product, fresh_product): + store.mark_pending_cleanup(product.product_id, now=NOW) + store.claim_product_purge(product.product_id, NOW + timedelta(days=30)) + old_claim = (NOW - timedelta(hours=2)).isoformat() + store._execute( + "UPDATE cleanup_records SET claimed_at = ? WHERE product_id = ? AND status = 'purging'", + (old_claim, stale_product.product_id), + ) + + failed = store.reconcile_stalled_purges(now=NOW, timeout=timedelta(hours=1)) + + assert [item.product_id for item in failed] == [stale_product.product_id] + assert store.get_product(stale_product.product_id).status == "failed_cleanup" + assert store.get_product(fresh_product.product_id).status == "pending_cleanup" + with pytest.raises(ProductPurgeInProgressError): + store.restore_product(fresh_product.product_id, now=NOW + timedelta(seconds=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, "重新排队笔盒"))