From da5c2b4503c57ac22c6ed9c2b0b3806a58885a1b Mon Sep 17 00:00:00 2001 From: obroccolio Date: Sat, 12 Sep 2026 13:55:31 +0800 Subject: [PATCH] feat: add product archive retention cleanup --- backend/service/app.py | 112 +++++++-- backend/service/cleanup_service.py | 143 ++++++++++++ backend/service/storage_metrics.py | 41 +++- backend/tests/test_cleanup_service.py | 315 ++++++++++++++++++++++++++ docker-compose.yml | 8 + 5 files changed, 595 insertions(+), 24 deletions(-) create mode 100644 backend/service/cleanup_service.py create mode 100644 backend/tests/test_cleanup_service.py diff --git a/backend/service/app.py b/backend/service/app.py index e6e7176..1837842 100644 --- a/backend/service/app.py +++ b/backend/service/app.py @@ -12,6 +12,7 @@ import threading import time import uuid import zipfile +from contextlib import asynccontextmanager from datetime import datetime, timezone from pathlib import Path from typing import Optional @@ -23,6 +24,7 @@ from PIL import Image, ImageDraw, ImageFont from core import config as wc_config from core.fonts import get_cached_font +from .cleanup_service import CleanupReport, CleanupService from .job_manager import JobManager from .line_spacing import analyze_svg_line_spacing_file from .log_config import get_logger @@ -67,7 +69,7 @@ FONTS_DIR = PROJECT_ROOT / "service_fonts" DESIGN_TEMPLATES_DIR = PROJECT_ROOT / "service_design_templates" METADATA_DIR = PROJECT_ROOT / "service_metadata" ORDERS_DIR = PROJECT_ROOT / "service_orders" -PRODUCT_ARCHIVES_DIR = PROJECT_ROOT / "service_product_archives" +PRODUCT_ARCHIVES_DIR = PROJECT_ROOT / "service_products" ASSETS_DIR.mkdir(parents=True, exist_ok=True) PROJECTS_DIR.mkdir(parents=True, exist_ok=True) FONTS_DIR.mkdir(parents=True, exist_ok=True) @@ -81,8 +83,19 @@ manager = JobManager(metadata_store) storage = Storage(WORKSPACE_DIR) runner = JobRunner(PROJECT_ROOT, manager) product_archive_store = ProductArchiveStore(PRODUCT_ARCHIVES_DIR / "metadata") +cleanup_service = CleanupService(storage, metadata_store, product_archive_store) -app = FastAPI(title="WordCloud Test Service", version="0.1.0") + +@asynccontextmanager +async def _app_lifespan(_app: FastAPI): + _start_cleanup_scheduler() + try: + yield + finally: + _stop_cleanup_scheduler() + + +app = FastAPI(title="WordCloud Test Service", version="0.1.0", lifespan=_app_lifespan) app.add_middleware( CORSMiddleware, @@ -162,26 +175,27 @@ def health() -> dict: @app.get("/api/maintenance/storage-summary") def storage_summary() -> dict: - referenced_jobs: set[str] = set() + asset_referenced_jobs: set[str] = set() for d in _list_dirs(ASSETS_DIR): meta = _read_asset_meta(d) job_id = meta.get("job_id") or "" if job_id: - referenced_jobs.add(job_id) - stale = storage.stale_job_dirs( - referenced_job_ids=referenced_jobs, - exclude_job_ids=metadata_store.job_ids(), - max_age_days=0, - ) + asset_referenced_jobs.add(job_id) + report = cleanup_service.preview(datetime.now(timezone.utc)) + archive_protected_jobs = cleanup_service.archive_protected_job_ids() return { "job_dir_count": len(storage.list_job_ids()), - "referenced_job_ids": len(referenced_jobs), - "stale_job_count": len(stale), - "reclaimable_bytes": sum(item["size_bytes"] for item in stale), + "referenced_job_ids": len(asset_referenced_jobs), + "asset_referenced_job_ids": sorted(asset_referenced_jobs), + "archive_protected_job_ids": sorted(archive_protected_jobs), + "archive_protected_job_count": len(archive_protected_jobs), + "temporary_job_ids": [item.job_id for item in report.temporary_jobs], + "stale_job_count": len(report.temporary_jobs), + "reclaimable_bytes": report.reclaimable_bytes, "metadata_db_bytes": metadata_store.summarize()["db_size_bytes"], "jobs_in_db": metadata_store.summarize()["jobs"], "events_in_db": metadata_store.summarize()["events"], - "dry_run_only": True, + "dry_run_only": not _cleanup_apply_enabled(), } @@ -1023,6 +1037,78 @@ def _require_orders_auth(request: Request) -> None: raise HTTPException(status_code=403, detail="需要登录") # noqa: S105 +def _cleanup_apply_enabled() -> bool: + return os.environ.get("CLEANUP_APPLY_ENABLED", "false").strip().lower() == "true" + + +def _run_scheduled_cleanup_once(now: datetime | None = None) -> CleanupReport: + run_at = now or datetime.now(timezone.utc) + apply_enabled = _cleanup_apply_enabled() + report = cleanup_service.apply(run_at) if apply_enabled else cleanup_service.preview(run_at) + log.info( + "cleanup cycle apply=%s temporary_jobs=%d pending_products=%d reclaimable_bytes=%d " + "reminder_job_ids=%s deleted_job_ids=%s deleted_product_ids=%s", + apply_enabled, + len(report.temporary_jobs), + len(report.pending_products), + report.reclaimable_bytes, + report.reminder_job_ids, + report.deleted_job_ids, + report.deleted_product_ids, + ) + return report + + +_cleanup_scheduler_stop = threading.Event() +_cleanup_scheduler_thread: threading.Thread | None = None + + +def _cleanup_scheduler_loop() -> None: + while not _cleanup_scheduler_stop.wait(24 * 60 * 60): + try: + _run_scheduled_cleanup_once() + except Exception: + log.exception("scheduled cleanup cycle failed") + + +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 + _cleanup_scheduler_stop.clear() + _cleanup_scheduler_thread = threading.Thread( + target=_cleanup_scheduler_loop, + name="retention-cleanup", + daemon=True, + ) + _cleanup_scheduler_thread.start() + + +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) + _cleanup_scheduler_thread = None + + +@app.get("/api/maintenance/cleanup-candidates", response_model=CleanupReport) +def cleanup_candidates(request: Request) -> CleanupReport: + _require_orders_auth(request) + return cleanup_service.preview(datetime.now(timezone.utc)) + + +@app.post("/api/maintenance/cleanup-run", response_model=CleanupReport) +def cleanup_run(request: Request, payload: dict) -> CleanupReport: + _require_orders_auth(request) + if payload.get("confirm") is not True: + raise HTTPException(status_code=400, detail="confirm must be true") + if not _cleanup_apply_enabled(): + raise HTTPException(status_code=409, detail="cleanup apply is disabled") + return cleanup_service.apply(datetime.now(timezone.utc)) + + @app.post("/api/login") async def login(request: Request) -> dict: try: diff --git a/backend/service/cleanup_service.py b/backend/service/cleanup_service.py new file mode 100644 index 0000000..5cfc1c6 --- /dev/null +++ b/backend/service/cleanup_service.py @@ -0,0 +1,143 @@ +"""Deterministic retention preview and physical cleanup for jobs and products.""" + +from __future__ import annotations + +import shutil +from datetime import datetime, timedelta, timezone +from pathlib import Path + +from pydantic import BaseModel, Field + +from .metadata_store import MetadataStore +from .product_archive_store import ProductArchiveStore +from .schemas import ProductRecord +from .storage import Storage + + +TEMPORARY_RETENTION = timedelta(days=30) +REMINDER_AGE = timedelta(days=23) + + +class TemporaryJob(BaseModel): + job_id: str + created_at: datetime + size_bytes: int + + +class CleanupReport(BaseModel): + temporary_jobs: list[TemporaryJob] = Field(default_factory=list) + pending_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) + deleted_product_ids: list[str] = Field(default_factory=list) + + +def _as_utc(value: datetime) -> datetime: + if value.tzinfo is None: + return value.replace(tzinfo=timezone.utc) + return value.astimezone(timezone.utc) + + +class CleanupService: + """Calculate candidates first and apply only those still eligible.""" + + def __init__( + self, + storage: Storage, + metadata_store: MetadataStore, + product_store: ProductArchiveStore, + ) -> None: + self.storage = storage + self.metadata_store = metadata_store + self.product_store = product_store + + def archive_protected_job_ids(self) -> set[str]: + rows = self.product_store._fetchall( + """SELECT DISTINCT source_job_id + FROM product_wordcloud_archives + WHERE source_job_id != ''""" + ) + return {str(row["source_job_id"]) for row in rows} + + def _is_archive_protected(self, job_id: str) -> bool: + row = self.product_store._fetchone( + """SELECT 1 FROM product_wordcloud_archives + WHERE source_job_id = ? LIMIT 1""", + (job_id,), + ) + return row is not None + + def preview(self, now: datetime) -> CleanupReport: + now_utc = _as_utc(now) + protected = self.archive_protected_job_ids() + temporary_jobs: list[TemporaryJob] = [] + reminder_job_ids: list[str] = [] + + for job in self.metadata_store.load_jobs(): + if job.status != "success" or not job.artifacts.get("db") or job.job_id in protected: + continue + age = now_utc - _as_utc(job.created_at) + if age >= TEMPORARY_RETENTION: + temporary_jobs.append( + TemporaryJob( + job_id=job.job_id, + created_at=job.created_at, + size_bytes=self.storage.job_dir_size(job.job_id), + ) + ) + elif age >= REMINDER_AGE: + reminder_job_ids.append(job.job_id) + + temporary_jobs.sort(key=lambda item: (item.created_at, item.job_id)) + reminder_job_ids.sort() + pending_products = self.product_store.due_product_cleanups(now_utc) + product_bytes = sum( + self._directory_size(self.product_store.root.parent / product.product_id) + for product in pending_products + ) + return CleanupReport( + temporary_jobs=temporary_jobs, + pending_products=pending_products, + reclaimable_bytes=sum(item.size_bytes for item in temporary_jobs) + product_bytes, + reminder_job_ids=reminder_job_ids, + ) + + def apply(self, now: datetime) -> CleanupReport: + now_utc = _as_utc(now) + report = self.preview(now_utc) + + for item in report.temporary_jobs: + # An archive can be committed after preview. Protect it at the last + # possible point before removing the original workspace. + if self._is_archive_protected(item.job_id): + continue + self.storage.remove_job_dir(item.job_id) + self.metadata_store.delete_job(item.job_id) + 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 + ): + 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) + report.deleted_product_ids.append(product.product_id) + + return report + + @staticmethod + def _directory_size(root: Path) -> int: + if not root.exists(): + return 0 + return sum(path.stat().st_size for path in root.rglob("*") if path.is_file()) diff --git a/backend/service/storage_metrics.py b/backend/service/storage_metrics.py index 7a9628a..b904797 100644 --- a/backend/service/storage_metrics.py +++ b/backend/service/storage_metrics.py @@ -9,10 +9,13 @@ from __future__ import annotations import argparse import json +import os from datetime import datetime, timezone from pathlib import Path +from .cleanup_service import CleanupService from .metadata_store import MetadataStore +from .product_archive_store import ProductArchiveStore from .storage import Storage @@ -21,6 +24,7 @@ PROJECT_ROOT = BACKEND_ROOT WORKSPACE_DIR = PROJECT_ROOT / "service_workspace" ASSETS_DIR = PROJECT_ROOT / "service_assets" METADATA_DIR = PROJECT_ROOT / "service_metadata" +PRODUCTS_DIR = PROJECT_ROOT / "service_products" def read_asset_meta(path: Path) -> dict: @@ -41,32 +45,40 @@ def referenced_job_ids() -> set[str]: def main() -> None: parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("--max-age-days", type=float, default=0) + 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("--json", type=Path, default=None, help="Write JSON report") args = parser.parse_args() store = MetadataStore(METADATA_DIR / "app.db") storage = Storage(WORKSPACE_DIR) + product_store = ProductArchiveStore(PRODUCTS_DIR / "metadata") + cleanup = CleanupService(storage, store, product_store) known_job_ids = store.job_ids() if store.db_path.exists() else set() - referenced = referenced_job_ids() - stale = storage.stale_job_dirs( - referenced_job_ids=referenced, + asset_referenced = referenced_job_ids() + archive_protected = cleanup.archive_protected_job_ids() + orphaned = storage.stale_job_dirs( + referenced_job_ids=archive_protected, exclude_job_ids=known_job_ids, max_age_days=args.max_age_days, ) - - reclaimable = sum(item["size_bytes"] for item in stale) + cleanup_report = cleanup.preview(datetime.now(timezone.utc)) + reclaimable = cleanup_report.reclaimable_bytes + sum(item["size_bytes"] for item in orphaned) report = { "scanned_at": datetime.now(timezone.utc).isoformat(), "job_dir_count": len(storage.list_job_ids()), - "referenced_job_ids": len(referenced), + "referenced_job_ids": len(asset_referenced), + "asset_referenced_job_ids": sorted(asset_referenced), + "archive_protected_job_ids": sorted(archive_protected), "known_job_ids": len(known_job_ids), - "stale_job_count": len(stale), + "temporary_jobs": [item.model_dump(mode="json") for item in cleanup_report.temporary_jobs], + "pending_product_ids": [item.product_id for item in cleanup_report.pending_products], + "reminder_job_ids": cleanup_report.reminder_job_ids, + "stale_job_count": len(orphaned), "reclaimable_bytes": reclaimable, "max_age_days": args.max_age_days, "apply": args.apply, - "stale_jobs": stale[:200], + "stale_jobs": orphaned[:200], } print(json.dumps(report, ensure_ascii=False, indent=2)) @@ -74,11 +86,18 @@ def main() -> None: args.json.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8") if args.apply: + 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 stale: + for item in orphaned: freed += storage.remove_job_dir(item["job_id"]) store.delete_job(item["job_id"]) - print(f"freed_bytes={freed}") + print( + "freed_bytes=" + f"{freed} deleted_job_ids={applied.deleted_job_ids} " + f"deleted_product_ids={applied.deleted_product_ids}" + ) if __name__ == "__main__": diff --git a/backend/tests/test_cleanup_service.py b/backend/tests/test_cleanup_service.py new file mode 100644 index 0000000..eb0be14 --- /dev/null +++ b/backend/tests/test_cleanup_service.py @@ -0,0 +1,315 @@ +from __future__ import annotations + +import json +import os +import sys +from datetime import datetime, timedelta, timezone +from pathlib import Path + +from fastapi.testclient import TestClient + + +BACKEND_DIR = Path(__file__).resolve().parents[1] +if str(BACKEND_DIR) not in sys.path: + sys.path.insert(0, str(BACKEND_DIR)) + +from service import app as service_app # noqa: E402 +from service.cleanup_service import CleanupService # noqa: E402 +from service.metadata_store import MetadataStore # noqa: E402 +from service.product_archive_store import ProductArchiveStore # noqa: E402 +from service.schemas import JobStatus, ProductInput # noqa: E402 +from service.storage import Storage # noqa: E402 +from service import storage_metrics # noqa: E402 + + +NOW = datetime(2026, 9, 12, 12, 0, tzinfo=timezone.utc) + + +def make_cleanup_service(tmp_path: Path) -> CleanupService: + return CleanupService( + Storage(tmp_path / "workspace"), + MetadataStore(tmp_path / "metadata" / "app.db"), + ProductArchiveStore(tmp_path / "products" / "metadata"), + ) + + +def create_success_job( + service: CleanupService, + *, + age_days: float, + archived: bool = False, + create_folder: bool = True, +) -> str: + job_id = f"{len(service.metadata_store.job_ids()) + 1:032x}" + created_at = NOW - timedelta(days=age_days) + db_path = service.storage.job_root(job_id) / "output" / "word_locations.sqlite" + if create_folder: + db_path.parent.mkdir(parents=True) + db_path.write_bytes(b"db-content") + service.metadata_store.upsert_job( + JobStatus( + job_id=job_id, + status="success", + stage="done", + progress_percent=100, + message="done", + artifacts={"db": str(db_path)}, + created_at=created_at, + updated_at=created_at, + ) + ) + if archived: + product = service.product_store.upsert_product(ProductInput(name="Archived", source="manual")) + service.product_store.write_archive_version( + product.product_id, + version_id=f"ver_{job_id}", + metadata={}, + preview_path="", + archives=[ + { + "archive_path": "snapshot.sqlite", + "source_job_id": job_id, + "source_asset_id": "asset-1", + "db_checksum": "checksum", + } + ], + now=NOW, + ) + return job_id + + +def create_pending_product( + service: CleanupService, + *, + purge_after: datetime, + create_folder: bool = True, +): + product = service.product_store.upsert_product(ProductInput(name="Pending", source="manual")) + marked_at = purge_after - timedelta(days=30) + product = service.product_store.mark_pending_cleanup(product.product_id, now=marked_at) + if create_folder: + product_dir = service.product_store.root.parent / product.product_id + product_dir.mkdir(parents=True) + (product_dir / "preview.png").write_bytes(b"product-bytes") + return product + + +def archive_job(service: CleanupService, job_id: str) -> None: + product = service.product_store.upsert_product(ProductInput(name="Late archive", source="manual")) + service.product_store.write_archive_version( + product.product_id, + version_id=f"ver_late_{job_id}", + metadata={}, + preview_path="", + archives=[ + { + "archive_path": "snapshot.sqlite", + "source_job_id": job_id, + "source_asset_id": "asset-late", + "db_checksum": "checksum", + } + ], + now=NOW, + ) + + +def auth_header() -> dict[str, str]: + return {"Authorization": f"Bearer {service_app._orders_token()}"} + + +def test_cleanup_skips_archived_job_and_marks_job_at_23_day_threshold_for_reminder(tmp_path): + service = make_cleanup_service(tmp_path) + archived_job = create_success_job(service, age_days=31, archived=True) + remind_job = create_success_job(service, age_days=23) + almost_remind_job = create_success_job(service, age_days=23 - (1 / 86400)) + + report = service.preview(now=NOW) + + assert archived_job not in {item.job_id for item in report.temporary_jobs} + assert remind_job in report.reminder_job_ids + assert almost_remind_job not in report.reminder_job_ids + + +def test_apply_removes_only_due_unarchived_job_and_due_product(tmp_path): + service = make_cleanup_service(tmp_path) + due_job = create_success_job(service, age_days=30) + young_job = create_success_job(service, age_days=29) + product = create_pending_product(service, purge_after=NOW) + + report = service.apply(now=NOW) + + assert report.deleted_job_ids == [due_job] + assert not service.storage.job_root(due_job).exists() + assert service.storage.job_root(young_job).exists() + assert due_job not in service.metadata_store.job_ids() + assert product.product_id in report.deleted_product_ids + assert not (service.product_store.root.parent / product.product_id).exists() + assert service.product_store.get_product(product.product_id).status == "purged" + + +def test_apply_treats_missing_job_and_product_folders_as_idempotent_success(tmp_path): + service = make_cleanup_service(tmp_path) + due_job = create_success_job(service, age_days=30, create_folder=False) + product = create_pending_product(service, purge_after=NOW, create_folder=False) + + report = service.apply(now=NOW) + + assert report.deleted_job_ids == [due_job] + assert report.deleted_product_ids == [product.product_id] + assert due_job not in service.metadata_store.job_ids() + assert service.product_store.get_product(product.product_id).status == "purged" + + +def test_restored_product_is_not_deleted_when_original_purge_time_arrives(tmp_path): + service = make_cleanup_service(tmp_path) + product = create_pending_product(service, purge_after=NOW) + product_dir = service.product_store.root.parent / product.product_id + service.product_store.restore_product(product.product_id, now=NOW - timedelta(days=1)) + + report = service.apply(now=NOW) + + assert product.product_id not in report.deleted_product_ids + assert product_dir.exists() + assert service.product_store.get_product(product.product_id).status == "active" + + +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) + original_preview = service.preview + + def preview_then_archive(now: datetime): + report = original_preview(now) + archive_job(service, due_job) + return report + + service.preview = preview_then_archive # type: ignore[method-assign] + + report = service.apply(now=NOW) + + assert due_job not in report.deleted_job_ids + assert service.storage.job_root(due_job).exists() + assert due_job in service.metadata_store.job_ids() + + +def test_cleanup_routes_require_admin_auth_and_confirmed_enabled_apply(tmp_path, monkeypatch): + service = make_cleanup_service(tmp_path) + due_job = create_success_job(service, age_days=3650) + monkeypatch.setattr(service_app, "cleanup_service", service) + monkeypatch.setenv("CLEANUP_APPLY_ENABLED", "false") + client = TestClient(service_app.app) + + assert client.get("/api/maintenance/cleanup-candidates").status_code == 403 + assert client.post("/api/maintenance/cleanup-run", json={"confirm": True}).status_code == 403 + assert client.post( + "/api/maintenance/cleanup-run", json={"confirm": False}, headers=auth_header() + ).status_code == 400 + disabled = client.post( + "/api/maintenance/cleanup-run", json={"confirm": True}, headers=auth_header() + ) + assert disabled.status_code == 409 + assert service.storage.job_root(due_job).exists() + + monkeypatch.setenv("CLEANUP_APPLY_ENABLED", "true") + preview = client.get("/api/maintenance/cleanup-candidates", headers=auth_header()) + applied = client.post( + "/api/maintenance/cleanup-run", json={"confirm": True}, headers=auth_header() + ) + + assert preview.status_code == 200 + assert preview.json()["temporary_jobs"][0]["job_id"] == due_job + assert applied.status_code == 200 + assert applied.json()["deleted_job_ids"] == [due_job] + assert not service.storage.job_root(due_job).exists() + + +def test_scheduled_cleanup_is_preview_only_until_apply_flag_is_enabled(tmp_path, monkeypatch): + service = make_cleanup_service(tmp_path) + first_job = create_success_job(service, age_days=31) + monkeypatch.setattr(service_app, "cleanup_service", service) + monkeypatch.setenv("CLEANUP_APPLY_ENABLED", "false") + + service_app._run_scheduled_cleanup_once(now=NOW) + + assert service.storage.job_root(first_job).exists() + + second_job = create_success_job(service, age_days=31) + monkeypatch.setenv("CLEANUP_APPLY_ENABLED", "true") + service_app._run_scheduled_cleanup_once(now=NOW) + + assert not service.storage.job_root(first_job).exists() + assert not service.storage.job_root(second_job).exists() + + +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) + asset_job = create_success_job(service, age_days=3650) + assets_dir = tmp_path / "assets" + asset_dir = assets_dir / "as" / "asset-old" + asset_dir.mkdir(parents=True) + (asset_dir / "meta.json").write_text( + '{"asset_id":"asset-old","job_id":"' + asset_job + '"}', encoding="utf-8" + ) + monkeypatch.setattr(service_app, "storage", service.storage) + monkeypatch.setattr(service_app, "metadata_store", service.metadata_store) + monkeypatch.setattr(service_app, "product_archive_store", service.product_store) + monkeypatch.setattr(service_app, "cleanup_service", service) + monkeypatch.setattr(service_app, "ASSETS_DIR", assets_dir) + + summary = service_app.storage_summary() + + assert summary["archive_protected_job_ids"] == [archived_job] + assert summary["asset_referenced_job_ids"] == [asset_job] + assert summary["temporary_job_ids"] == [asset_job] + + +def test_storage_metrics_protects_archives_but_not_old_asset_references( + tmp_path, monkeypatch, capsys +): + workspace = tmp_path / "workspace" + assets = tmp_path / "assets" + metadata = tmp_path / "metadata" + products = tmp_path / "products" + storage = Storage(workspace) + archived_job = "a" * 32 + asset_job = "b" * 32 + for job_id in (archived_job, asset_job): + root = storage.job_root(job_id) + root.mkdir(parents=True) + (root / "payload.bin").write_bytes(b"payload") + old_timestamp = (NOW - timedelta(days=31)).timestamp() + os.utime(root, (old_timestamp, old_timestamp)) + + asset_dir = assets / "as" / "asset-old" + asset_dir.mkdir(parents=True) + (asset_dir / "meta.json").write_text( + json.dumps({"asset_id": "asset-old", "job_id": asset_job}), encoding="utf-8" + ) + product_store = ProductArchiveStore(products / "metadata") + product = product_store.upsert_product(ProductInput(name="Archive", source="manual")) + product_store.write_archive_version( + product.product_id, + version_id="ver_archive", + metadata={}, + preview_path="", + archives=[{ + "archive_path": "snapshot.sqlite", + "source_job_id": archived_job, + "source_asset_id": "asset-archive", + "db_checksum": "checksum", + }], + now=NOW, + ) + monkeypatch.setattr(storage_metrics, "WORKSPACE_DIR", workspace) + monkeypatch.setattr(storage_metrics, "ASSETS_DIR", assets) + monkeypatch.setattr(storage_metrics, "METADATA_DIR", metadata) + monkeypatch.setattr(storage_metrics, "PRODUCTS_DIR", products, raising=False) + monkeypatch.setattr(sys, "argv", ["storage_metrics", "--max-age-days", "30"]) + + storage_metrics.main() + report = json.loads(capsys.readouterr().out) + + 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] diff --git a/docker-compose.yml b/docker-compose.yml index 412df5d..032868f 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -7,12 +7,17 @@ services: container_name: wordcloud-backend ports: - "8000:8000" + environment: + CLEANUP_APPLY_ENABLED: "${CLEANUP_APPLY_ENABLED:-false}" volumes: - wordcloud_workspace:/app/service_workspace - wordcloud_assets:/app/service_assets - wordcloud_projects:/app/service_projects - wordcloud_design_templates:/app/service_design_templates - wordcloud_fonts:/app/service_fonts + - wordcloud_products:/app/service_products + - wordcloud_metadata:/app/service_metadata + - wordcloud_orders:/app/service_orders restart: unless-stopped healthcheck: test: ["CMD", "curl", "-f", "http://localhost:8000/api/health"] @@ -40,3 +45,6 @@ volumes: wordcloud_projects: wordcloud_design_templates: wordcloud_fonts: + wordcloud_products: + wordcloud_metadata: + wordcloud_orders: