diff --git a/backend/service/app.py b/backend/service/app.py index 1837842..0428dae 100644 --- a/backend/service/app.py +++ b/backend/service/app.py @@ -30,7 +30,7 @@ from .line_spacing import analyze_svg_line_spacing_file from .log_config import get_logger from .metadata_store import MetadataStore from .product_archive_service import ProductArchiveService, ProductPendingCleanupError -from .product_archive_store import ProductArchiveStore +from .product_archive_store import ProductArchiveStore, ProductPurgeInProgressError from .runner import JobRunner from .schemas import ( Asset, @@ -1073,10 +1073,12 @@ def _cleanup_scheduler_loop() -> None: def _start_cleanup_scheduler() -> None: global _cleanup_scheduler_thread - _run_scheduled_cleanup_once() - if _cleanup_scheduler_thread is not None and _cleanup_scheduler_thread.is_alive(): - return + if _cleanup_scheduler_thread is not None: + if _cleanup_scheduler_thread.is_alive(): + return + _cleanup_scheduler_thread = None _cleanup_scheduler_stop.clear() + _run_scheduled_cleanup_once() _cleanup_scheduler_thread = threading.Thread( target=_cleanup_scheduler_loop, name="retention-cleanup", @@ -1088,8 +1090,10 @@ def _start_cleanup_scheduler() -> None: def _stop_cleanup_scheduler() -> None: global _cleanup_scheduler_thread _cleanup_scheduler_stop.set() - if _cleanup_scheduler_thread is not None: - _cleanup_scheduler_thread.join(timeout=1) + worker = _cleanup_scheduler_thread + if worker is not None: + worker.join(timeout=5) + if worker is not None and not worker.is_alive() and _cleanup_scheduler_thread is worker: _cleanup_scheduler_thread = None @@ -1240,6 +1244,8 @@ def restore_product(request: Request, product_id: str) -> ProductRecord: _get_product_or_404(product_id) try: return product_archive_store.restore_product(product_id, now=datetime.now(timezone.utc)) + except ProductPurgeInProgressError as exc: + raise HTTPException(status_code=409, detail=str(exc)) from exc except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc diff --git a/backend/service/cleanup_service.py b/backend/service/cleanup_service.py index 5cfc1c6..0abb95d 100644 --- a/backend/service/cleanup_service.py +++ b/backend/service/cleanup_service.py @@ -117,25 +117,25 @@ class CleanupService: report.deleted_job_ids.append(item.job_id) for candidate in report.pending_products: - # Re-read logical state so a concurrent restore wins over cleanup. - try: - product = self.product_store.get_product(candidate.product_id) - except ValueError: - continue - if ( - product.status != "pending_cleanup" - or product.purge_after is None - or _as_utc(product.purge_after) > now_utc - ): + product = self.product_store.claim_product_purge(candidate.product_id, now_utc) + if product is None: continue product_dir = self.product_store.root.parent / product.product_id - if product_dir.exists(): - shutil.rmtree(product_dir) - self.product_store.purge_product(product.product_id, now=now_utc) + try: + self._before_product_files_delete(product.product_id) + if product_dir.exists(): + shutil.rmtree(product_dir) + except Exception: + self.product_store.release_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) return report + def _before_product_files_delete(self, product_id: str) -> None: + """Interleaving seam between the durable claim and physical deletion.""" + @staticmethod def _directory_size(root: Path) -> int: if not root.exists(): diff --git a/backend/service/product_archive_store.py b/backend/service/product_archive_store.py index d828f75..7a2caae 100644 --- a/backend/service/product_archive_store.py +++ b/backend/service/product_archive_store.py @@ -23,6 +23,10 @@ def _now() -> datetime: return datetime.now(timezone.utc) +class ProductPurgeInProgressError(ValueError): + """Raised when restore loses the atomic race to physical purge.""" + + class ProductArchiveStore: """Owns durable product metadata and logical cleanup state.""" @@ -328,43 +332,111 @@ class ProductArchiveStore: return self.get_product(product_id) def restore_product(self, product_id: str, now: datetime | None = None) -> ProductRecord: - self.get_product(product_id) restored_at = now or _now() - self._execute( - "UPDATE products SET status = ?, purge_after = NULL, updated_at = ? WHERE product_id = ?", - ("active", restored_at.isoformat(), product_id), - ) - self._execute( - """UPDATE cleanup_records SET status = ?, completed_at = ? - WHERE product_id = ? AND status = ?""", - ("restored", restored_at.isoformat(), product_id, "pending_cleanup"), - ) + with self._lock, self._connect() as conn: + conn.execute("BEGIN IMMEDIATE") + product = conn.execute( + "SELECT product_id, status FROM products WHERE product_id = ?", (product_id,) + ).fetchone() + if product is None: + raise ValueError(f"unknown product_id: {product_id}") + if product["status"] == "purged": + raise ValueError("purged product cannot be restored") + purging = conn.execute( + """SELECT 1 FROM cleanup_records + WHERE product_id = ? AND status = 'purging' LIMIT 1""", + (product_id,), + ).fetchone() + if purging is not None: + raise ProductPurgeInProgressError("product is purging") + conn.execute( + "UPDATE products SET status = ?, purge_after = NULL, updated_at = ? WHERE product_id = ?", + ("active", restored_at.isoformat(), product_id), + ) + conn.execute( + """UPDATE cleanup_records SET status = ?, completed_at = ? + WHERE product_id = ? AND status = ?""", + ("restored", restored_at.isoformat(), product_id, "pending_cleanup"), + ) return self.get_product(product_id) def due_product_cleanups(self, now: datetime) -> list[ProductRecord]: rows = self._fetchall( - """SELECT * FROM products WHERE status = ? AND purge_after IS NOT NULL AND purge_after <= ? - ORDER BY purge_after, product_id""", + """SELECT products.* FROM products + WHERE products.status = ? + AND products.purge_after IS NOT NULL + AND products.purge_after <= ? + AND EXISTS ( + SELECT 1 FROM cleanup_records + WHERE cleanup_records.product_id = products.product_id + AND cleanup_records.status = 'pending_cleanup' + ) + ORDER BY products.purge_after, products.product_id""", ("pending_cleanup", now.isoformat()), ) return [self._product_from_row(row) for row in rows] + def claim_product_purge(self, product_id: str, now: datetime) -> ProductRecord | None: + """Atomically win the right to delete a due product's files.""" + with self._lock, self._connect() as conn: + conn.execute("BEGIN IMMEDIATE") + row = conn.execute( + "SELECT * FROM products WHERE product_id = ?", (product_id,) + ).fetchone() + if row is None: + return None + purge_after = datetime.fromisoformat(row["purge_after"]) if row["purge_after"] else None + if row["status"] != "pending_cleanup" or purge_after is None or purge_after > now: + return None + cleanup = conn.execute( + """SELECT cleanup_id FROM cleanup_records + WHERE product_id = ? AND status = 'pending_cleanup' + ORDER BY created_at DESC, cleanup_id DESC LIMIT 1""", + (product_id,), + ).fetchone() + if cleanup is None: + return None + updated = conn.execute( + """UPDATE cleanup_records SET status = 'purging' + WHERE cleanup_id = ? AND status = 'pending_cleanup'""", + (cleanup["cleanup_id"],), + ) + if updated.rowcount != 1: + 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 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: + 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") + conn.execute( + "UPDATE products SET status = ?, updated_at = ? WHERE product_id = ?", + ("purged", now.isoformat(), product_id), + ) + conn.execute( + """UPDATE cleanup_records SET status = ?, completed_at = ? + WHERE product_id = ? AND status = 'purging'""", + ("purged", now.isoformat(), product_id), + ) + return self.get_product(product_id) + def purge_product(self, product_id: str, now: datetime | None = None) -> ProductRecord: purged_at = now or _now() - product = self.get_product(product_id) - if ( - product.status != "pending_cleanup" - or product.purge_after is None - or product.purge_after > purged_at - ): + if self.claim_product_purge(product_id, purged_at) is None: raise ValueError("product is not due for purge") - self._execute( - "UPDATE products SET status = ?, updated_at = ? WHERE product_id = ?", - ("purged", purged_at.isoformat(), product_id), - ) - self._execute( - """UPDATE cleanup_records SET status = ?, completed_at = ? - WHERE product_id = ? AND status = ?""", - ("purged", purged_at.isoformat(), product_id, "pending_cleanup"), - ) - return self.get_product(product_id) + return self.finalize_product_purge(product_id, purged_at) diff --git a/backend/service/storage_metrics.py b/backend/service/storage_metrics.py index b904797..b131e34 100644 --- a/backend/service/storage_metrics.py +++ b/backend/service/storage_metrics.py @@ -1,8 +1,8 @@ #!/usr/bin/env python3 -"""Inspect and optionally clean stale job directories. +"""Inspect retention candidates and optionally clean verified managed jobs. -Default mode is a safe dry-run that reports reclaimable bytes. Pass --apply to -actually remove unreferenced job directories. +Default mode is a safe dry-run. Metadata-less workspaces are reported for +manual investigation but are never removed by ``--apply``. """ from __future__ import annotations @@ -46,7 +46,11 @@ def referenced_job_ids() -> set[str]: def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--max-age-days", type=float, default=30) - parser.add_argument("--apply", action="store_true", help="Actually delete stale job directories") + parser.add_argument( + "--apply", + action="store_true", + help="Delete only CleanupService-verified candidates; orphan workspaces remain report-only", + ) parser.add_argument("--json", type=Path, default=None, help="Write JSON report") args = parser.parse_args() @@ -89,10 +93,11 @@ def main() -> None: if os.environ.get("CLEANUP_APPLY_ENABLED", "false").strip().lower() != "true": parser.error("--apply requires CLEANUP_APPLY_ENABLED=true") applied = cleanup.apply(datetime.now(timezone.utc)) - freed = 0 - for item in orphaned: - freed += storage.remove_job_dir(item["job_id"]) - store.delete_job(item["job_id"]) + freed = sum( + item.size_bytes + for item in applied.temporary_jobs + if item.job_id in applied.deleted_job_ids + ) print( "freed_bytes=" f"{freed} deleted_job_ids={applied.deleted_job_ids} " diff --git a/backend/tests/test_cleanup_service.py b/backend/tests/test_cleanup_service.py index eb0be14..7a6873e 100644 --- a/backend/tests/test_cleanup_service.py +++ b/backend/tests/test_cleanup_service.py @@ -6,6 +6,7 @@ import sys from datetime import datetime, timedelta, timezone from pathlib import Path +import pytest from fastapi.testclient import TestClient @@ -173,6 +174,48 @@ def test_restored_product_is_not_deleted_when_original_purge_time_arrives(tmp_pa assert service.product_store.get_product(product.product_id).status == "active" +def test_atomic_purge_claim_blocks_restore_before_product_files_are_deleted( + tmp_path, monkeypatch +): + service = make_cleanup_service(tmp_path) + product = create_pending_product(service, purge_after=NOW) + restore_conflicts: list[str] = [] + + def attempt_restore(product_id: str) -> None: + with pytest.raises(ValueError, match="purging"): + service.product_store.restore_product(product_id, now=NOW) + restore_conflicts.append(product_id) + + monkeypatch.setattr( + service, "_before_product_files_delete", attempt_restore, raising=False + ) + + report = service.apply(now=NOW) + + assert restore_conflicts == [product.product_id] + assert report.deleted_product_ids == [product.product_id] + 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): + service = make_cleanup_service(tmp_path) + product = create_pending_product(service, purge_after=NOW) + + monkeypatch.setattr( + "service.cleanup_service.shutil.rmtree", + lambda *_args, **_kwargs: (_ for _ in ()).throw(OSError("disk failure")), + ) + + 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( + "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 + + 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) @@ -241,6 +284,31 @@ def test_scheduled_cleanup_is_preview_only_until_apply_flag_is_enabled(tmp_path, assert not service.storage.job_root(second_job).exists() +def test_scheduler_stop_retains_live_worker_and_restart_does_not_duplicate_cycle(monkeypatch): + class LiveThread: + def __init__(self): + self.join_calls: list[float | None] = [] + + def is_alive(self) -> bool: + return True + + def join(self, timeout=None) -> None: + self.join_calls.append(timeout) + + worker = LiveThread() + cycles: list[str] = [] + monkeypatch.setattr(service_app, "_cleanup_scheduler_thread", worker) + monkeypatch.setattr(service_app, "_run_scheduled_cleanup_once", lambda: cycles.append("run")) + + service_app._stop_cleanup_scheduler() + service_app._start_cleanup_scheduler() + + assert service_app._cleanup_scheduler_stop.is_set() + assert service_app._cleanup_scheduler_thread is worker + assert worker.join_calls + assert cycles == [] + + def test_storage_summary_separates_archive_protection_from_asset_references(tmp_path, monkeypatch): service = make_cleanup_service(tmp_path) archived_job = create_success_job(service, age_days=3650, archived=True) @@ -313,3 +381,44 @@ def test_storage_metrics_protects_archives_but_not_old_asset_references( assert report["archive_protected_job_ids"] == [archived_job] assert report["asset_referenced_job_ids"] == [asset_job] assert [item["job_id"] for item in report["stale_jobs"]] == [asset_job] + + +def test_storage_metrics_apply_never_deletes_unproven_or_failed_workspaces( + tmp_path, monkeypatch +): + workspace = tmp_path / "workspace" + metadata = tmp_path / "metadata" + storage = Storage(workspace) + metadata_store = MetadataStore(metadata / "app.db") + orphan_job = "c" * 32 + failed_job = "d" * 32 + for job_id in (orphan_job, failed_job): + root = storage.job_root(job_id) + root.mkdir(parents=True) + (root / "payload.bin").write_bytes(b"payload") + old_timestamp = (datetime.now(timezone.utc) - timedelta(days=31)).timestamp() + os.utime(root, (old_timestamp, old_timestamp)) + metadata_store.upsert_job( + JobStatus( + job_id=failed_job, + status="failed", + stage="failed", + progress_percent=100, + message="failed", + artifacts={"db": str(storage.job_root(failed_job) / "output" / "db.sqlite")}, + error="generation failed", + created_at=datetime.now(timezone.utc) - timedelta(days=31), + updated_at=datetime.now(timezone.utc) - timedelta(days=31), + ) + ) + monkeypatch.setattr(storage_metrics, "WORKSPACE_DIR", workspace) + monkeypatch.setattr(storage_metrics, "ASSETS_DIR", tmp_path / "assets") + monkeypatch.setattr(storage_metrics, "METADATA_DIR", metadata) + monkeypatch.setattr(storage_metrics, "PRODUCTS_DIR", tmp_path / "products") + monkeypatch.setattr(sys, "argv", ["storage_metrics", "--apply"]) + monkeypatch.setenv("CLEANUP_APPLY_ENABLED", "true") + + storage_metrics.main() + + assert storage.job_root(orphan_job).exists() + assert storage.job_root(failed_job).exists() diff --git a/backend/tests/test_product_archive_store.py b/backend/tests/test_product_archive_store.py index e28ea26..8ed9aac 100644 --- a/backend/tests/test_product_archive_store.py +++ b/backend/tests/test_product_archive_store.py @@ -99,6 +99,18 @@ def test_restore_removes_product_from_due_cleanup_selection(tmp_path): assert store.due_product_cleanups(now=NOW + timedelta(days=31)) == [] +def test_restore_rejects_product_after_purge_has_deleted_its_files(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.purge_product(product.product_id, now=NOW + timedelta(days=30)) + + with pytest.raises(ValueError, match="purged"): + store.restore_product(product.product_id, now=NOW + timedelta(days=31)) + + assert store.get_product(product.product_id).status == "purged" + + 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, "第一个笔盒"))