From abe449b4f272a3179af34c98e6163ab4f9e95129 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Mon, 21 Sep 2026 14:13:25 -0400 Subject: [PATCH] feat: placement reconciler tasks + API (4246, slice 3b) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three Celery tasks wrapping the 3a service, and the endpoints that drive them. Routed to `maintenance_long` alongside backups: 33k renames on NFS have no business in the quick lane where the self-healing sweeps live (the 2026-06-07 starvation). A durability bug in 3a, found by thinking about what a crash costs rather than by a failing test: `apply_run` wrote its ledger only at the end, so a worker dying at row 30,000 of 33,789 would have taken the undo information for the first 29,999 with it — and that ledger is the ONLY record of where those files came from. It now persists every 200 moves. Two things fell out of writing that: - `_persist` reassigns `run.moves`, so the loop had to snapshot the plan first rather than iterate the attribute it rewrites. - the reassignment is itself load-bearing: SQLAlchemy does not track in-place mutation of a JSONB list, so an `.append()` alone would never reach the database and the ledger would have stayed silently empty. Re-running a partially-applied plan is safe — the moved rows no longer match their `from` and refuse as "row moved since planning" — but `apply_placement` deliberately has NO autoretry: re-entering a half-applied plan should be the operator's call after reading what happened, not the queue's. Endpoints gate on run state as well as the service does, so a stray POST cannot re-apply an applied run. The list response omits `moves` (an applied whole-library run carries tens of thousands of entries); the detail endpoint includes them, because that detail IS the preview read before agreeing. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01LVjrnpQjRgHdvq95rASoiR --- backend/app/api/cleanup.py | 90 +++++++++++++- backend/app/celery_app.py | 3 + backend/app/services/library_layout.py | 37 +++++- backend/app/tasks/library_placement.py | 125 +++++++++++++++++++ tests/test_api_placement.py | 163 +++++++++++++++++++++++++ 5 files changed, 415 insertions(+), 3 deletions(-) create mode 100644 backend/app/tasks/library_placement.py create mode 100644 tests/test_api_placement.py diff --git a/backend/app/api/cleanup.py b/backend/app/api/cleanup.py index 44cf6d3..d4b2196 100644 --- a/backend/app/api/cleanup.py +++ b/backend/app/api/cleanup.py @@ -29,7 +29,7 @@ from quart import Blueprint, jsonify, request from sqlalchemy import select from ..extensions import get_session -from ..models import LibraryAuditRun +from ..models import LibraryAuditRun, LibraryPlacementRun from ..services import cleanup_service, library_layout from ._responses import error_response as _bad @@ -218,3 +218,91 @@ async def layout_survey(): ) ) return jsonify({**report.as_dict(), "checked_disk": check_disk}) + + +def _serialize_placement_run(run: LibraryPlacementRun, *, moves: bool = False) -> dict: + """`moves` is opt-in: an applied whole-library run carries tens of + thousands of entries, which is a fine thing to hold in Postgres and a + poor thing to put in every list response.""" + out = { + "id": run.id, + "status": run.status, + "artist_id": run.artist_id, + "started_at": run.started_at.isoformat() if run.started_at else None, + "finished_at": run.finished_at.isoformat() if run.finished_at else None, + "planned_count": run.planned_count, + "moved_count": run.moved_count, + "refused_count": run.refused_count, + "refusals": run.refusals or [], + "error": run.error, + } + if moves: + out["moves"] = run.moves or [] + return out + + +@cleanup_bp.route("/placement/runs", methods=["GET"]) +async def placement_runs(): + """Newest first. Without `moves`, so the list stays small.""" + try: + limit = min(int(request.args.get("limit", "25")), 100) + except ValueError: + return _bad("invalid_limit") + async with get_session() as session: + rows = (await session.execute( + select(LibraryPlacementRun) + .order_by(LibraryPlacementRun.id.desc()).limit(limit) + )).scalars().all() + return jsonify({"runs": [_serialize_placement_run(r) for r in rows]}) + + +@cleanup_bp.route("/placement/runs/", methods=["GET"]) +async def placement_run(run_id: int): + """One run WITH its moves — this is the preview the operator reads before + agreeing, and the record of what happened afterwards.""" + async with get_session() as session: + run = await session.get(LibraryPlacementRun, run_id) + if run is None: + return _bad("not_found", status=404) + return jsonify(_serialize_placement_run(run, moves=True)) + + +@cleanup_bp.route("/placement/plan", methods=["POST"]) +async def placement_plan(): + """Queue a planning run. `artist_id` scopes it to one artist, which is the + intended use: do one, look at the gallery, then continue or revert.""" + body = await request.get_json(silent=True) or {} + artist_id = body.get("artist_id") + if artist_id is not None and not isinstance(artist_id, int): + return _bad("invalid_artist_id") + from ..tasks.library_placement import plan_placement + plan_placement.delay(artist_id) + return jsonify({"status": "dispatched"}), 202 + + +@cleanup_bp.route("/placement/runs//apply", methods=["POST"]) +async def placement_apply(run_id: int): + """Execute a ready run. This renames files and rewrites rows.""" + async with get_session() as session: + run = await session.get(LibraryPlacementRun, run_id) + if run is None: + return _bad("not_found", status=404) + if run.status != "ready": + return _bad("not_ready", detail=f"run is {run.status}") + from ..tasks.library_placement import apply_placement + apply_placement.delay(run_id) + return jsonify({"status": "dispatched"}), 202 + + +@cleanup_bp.route("/placement/runs//revert", methods=["POST"]) +async def placement_revert(run_id: int): + """Put an applied run's files back. The reason the ledger is kept.""" + async with get_session() as session: + run = await session.get(LibraryPlacementRun, run_id) + if run is None: + return _bad("not_found", status=404) + if run.status != "applied": + return _bad("not_applied", detail=f"run is {run.status}") + from ..tasks.library_placement import revert_placement + revert_placement.delay(run_id) + return jsonify({"status": "dispatched"}), 202 diff --git a/backend/app/celery_app.py b/backend/app/celery_app.py index d49c8ef..5f176a2 100644 --- a/backend/app/celery_app.py +++ b/backend/app/celery_app.py @@ -35,6 +35,7 @@ def make_celery() -> Celery: "backend.app.tasks.backup", "backend.app.tasks.admin", "backend.app.tasks.library_audit", + "backend.app.tasks.library_placement", "backend.app.tasks.translation", ], ) @@ -62,6 +63,8 @@ def make_celery() -> Celery: # 2026-06-07: a 2h audit blocked vacuum/backup/normalize for hours). "backend.app.tasks.maintenance.*": {"queue": "maintenance"}, "backend.app.tasks.backup.*": {"queue": "maintenance_long"}, + # 33k renames on NFS: long lane, same as backups. + "backend.app.tasks.library_placement.*": {"queue": "maintenance_long"}, "backend.app.tasks.admin.*": {"queue": "maintenance_long"}, "backend.app.tasks.library_audit.*": {"queue": "maintenance_long"}, # Translation backfill hits the LLM (~1–6s/item) → the long lane so it diff --git a/backend/app/services/library_layout.py b/backend/app/services/library_layout.py index dc3961b..c14c711 100644 --- a/backend/app/services/library_layout.py +++ b/backend/app/services/library_layout.py @@ -317,7 +317,7 @@ def _move_one(src: Path, dest: Path) -> str | None: def apply_run( - session: Session, run: LibraryPlacementRun, + session: Session, run: LibraryPlacementRun, *, chunk: int = 0, ) -> LibraryPlacementRun: """Execute a `ready` run's stored plan: file and row together, per row. @@ -325,13 +325,38 @@ def apply_run( move can never leave `ImageRecord.path` pointing at a file that is not there. Refusals are recorded and the run continues — one row that moved on since planning is not a reason to abandon the other 33,788. + + `chunk` commits progress every N moves. Set it for any real run: the + ledger is the ONLY record of where a file came from, so a worker that + dies at row 30,000 of 33,789 must not take the undo information for the + first 29,999 with it. Left at 0 (tests, small runs) everything persists + in one go at the end. + + Re-running a partially-applied plan is safe rather than clever: the rows + already moved no longer match their `from`, so they refuse as "row moved + since planning" instead of being moved twice. """ if run.status != "ready": raise ValueError(f"run {run.id} is {run.status}, not ready") refusals: list[dict] = [] moved: list[dict] = [] - for move in run.moves: + + def _persist() -> None: + # Reassign rather than mutate: SQLAlchemy does not track in-place + # changes to a JSONB list, so an .append() alone would never reach + # the database and the ledger would silently stay empty. + run.moves = list(moved) + run.refusals = list(refusals) + run.moved_count = len(moved) + run.refused_count = len(refusals) + session.commit() + + # Snapshot the plan before iterating: `_persist` reassigns `run.moves`, + # and iterating the attribute while rewriting it would walk a list that + # changes underneath the loop. + plan = list(run.moves) + for done, move in enumerate(plan, start=1): record = session.get(ImageRecord, move["image_id"]) if record is None: refusals.append({"image_id": move["image_id"], "reason": "row gone"}) @@ -349,6 +374,8 @@ def apply_run( continue record.path = move["to"] moved.append(move) + if chunk and done % chunk == 0: + _persist() run.moves = moved run.refusals = refusals @@ -372,6 +399,12 @@ def revert_run( Refuses the same way the apply does: a file someone has since moved or replaced stays where it is, and its row is left alone. + + A revert interrupted half way is resumable by re-running it: the rows + already put back no longer sit at `to`, so they refuse rather than move + twice. Unlike the apply this needs no chunked persistence — it consumes + the ledger rather than producing it, so a crash costs progress, not + information. """ if run.status != "applied": raise ValueError(f"run {run.id} is {run.status}, not applied") diff --git a/backend/app/tasks/library_placement.py b/backend/app/tasks/library_placement.py new file mode 100644 index 0000000..510795d --- /dev/null +++ b/backend/app/tasks/library_placement.py @@ -0,0 +1,125 @@ +"""Placement reconciler tasks — plan, apply, revert (milestone #421). + +The service (`services.library_layout`) holds the decisions; this module is +only the async wrapper, matching `tasks.library_audit`: run on the +maintenance queue, mark the run `error` with a traceback if anything escapes, +and return a small summary dict so eager-mode tests can assert on it. + +Applying is a long run — 33,789 renames on the operator's library at the time +of writing — so `apply_placement` persists its ledger in chunks rather than +at the end. That ledger is the only record of where each file came from, and +a worker that dies two thirds of the way through must not take the undo +information for the first two thirds with it. +""" + +import logging +import traceback +from datetime import UTC, datetime +from pathlib import Path + +from sqlalchemy.exc import DBAPIError, OperationalError + +from ..celery_app import celery +from ..models import LibraryPlacementRun +from ..services import library_layout +from ._sync_engine import sync_session_factory as _sync_session_factory + +log = logging.getLogger(__name__) + +IMAGES_ROOT = Path("/images") + +# Commit the ledger every this many moves. Small enough that a crash loses +# seconds of work, large enough not to make a COMMIT per rename. +_APPLY_CHUNK = 200 + + +def _fail(session, run_id: int, message: str) -> None: + run = session.get(LibraryPlacementRun, run_id) + if run is not None: + run.status = "error" + run.error = message + run.finished_at = datetime.now(UTC) + session.commit() + + +@celery.task( + name="backend.app.tasks.library_placement.plan_placement", + autoretry_for=(OperationalError, DBAPIError), + retry_backoff=5, retry_backoff_max=60, retry_jitter=True, max_retries=3, + soft_time_limit=900, time_limit=1000, +) +def plan_placement(artist_id: int | None = None) -> dict: + """Build a move plan and leave it `ready` for the operator to read. + + Reads rows and stats destinations; moves nothing. + """ + SessionLocal = _sync_session_factory() + with SessionLocal() as session: + run = library_layout.plan_placement( + session, IMAGES_ROOT, artist_id=artist_id, + ) + session.commit() + return { + "run_id": run.id, "status": run.status, + "planned_count": run.planned_count, + } + + +@celery.task( + name="backend.app.tasks.library_placement.apply_placement", + soft_time_limit=7200, time_limit=7500, +) +def apply_placement(run_id: int) -> dict: + """Execute a `ready` run's stored plan. Renames files and rewrites rows. + + No autoretry: a retry would re-enter a half-applied plan on a schedule + nobody asked for. Re-running IS safe (the applied rows refuse as "row + moved since planning"), but that should be the operator's decision after + reading what happened, not the queue's. + """ + SessionLocal = _sync_session_factory() + with SessionLocal() as session: + run = session.get(LibraryPlacementRun, run_id) + if run is None: + return {"run_id": run_id, "status": "missing"} + if run.status != "ready": + return {"run_id": run_id, "status": run.status, "skipped": True} + try: + library_layout.apply_run(session, run, chunk=_APPLY_CHUNK) + session.commit() + except Exception: + log.exception("placement apply failed for run %s", run_id) + session.rollback() + _fail(session, run_id, traceback.format_exc()) + return {"run_id": run_id, "status": "error"} + return { + "run_id": run_id, "status": run.status, + "moved": run.moved_count, "refused": run.refused_count, + } + + +@celery.task( + name="backend.app.tasks.library_placement.revert_placement", + soft_time_limit=7200, time_limit=7500, +) +def revert_placement(run_id: int) -> dict: + """Put an applied run's files back where they came from.""" + SessionLocal = _sync_session_factory() + with SessionLocal() as session: + run = session.get(LibraryPlacementRun, run_id) + if run is None: + return {"run_id": run_id, "status": "missing"} + if run.status != "applied": + return {"run_id": run_id, "status": run.status, "skipped": True} + try: + library_layout.revert_run(session, run) + session.commit() + except Exception: + log.exception("placement revert failed for run %s", run_id) + session.rollback() + _fail(session, run_id, traceback.format_exc()) + return {"run_id": run_id, "status": "error"} + return { + "run_id": run_id, "status": run.status, + "refused": run.refused_count, + } diff --git a/tests/test_api_placement.py b/tests/test_api_placement.py new file mode 100644 index 0000000..4bbe5e6 --- /dev/null +++ b/tests/test_api_placement.py @@ -0,0 +1,163 @@ +"""The placement reconciler's task + API surface (milestone #421, slice 3b). + +The move logic itself is covered in tests/test_library_layout.py; this module +covers the wrapper — that the tasks are registered and routed, that the +endpoints gate on run state, and that a list response stays small. +""" + +import pytest +from sqlalchemy import select + +from backend.app.celery_app import celery +from backend.app.models import Artist, LibraryPlacementRun + +pytestmark = pytest.mark.integration + +_TASKS = ( + "backend.app.tasks.library_placement.plan_placement", + "backend.app.tasks.library_placement.apply_placement", + "backend.app.tasks.library_placement.revert_placement", +) + + +@pytest.mark.parametrize("name", _TASKS) +def test_placement_tasks_are_registered(name): + assert name in celery.tasks + + +def test_placement_runs_on_the_long_maintenance_lane(): + """33k renames must not sit in the quick lane, which is where the + self-healing sweeps live (the 2026-06-07 starvation).""" + routes = celery.conf.task_routes + assert routes["backend.app.tasks.library_placement.*"] == { + "queue": "maintenance_long" + } + + +def _run(db, status="ready", moves=None, artist_id=None): + """Adds the row; the caller awaits the flush. `db` is the ASYNC session, + so flushing here would leave an un-awaited coroutine and the row would + never reach the database.""" + run = LibraryPlacementRun( + status=status, artist_id=artist_id, moves=moves or [], + planned_count=len(moves or []), + ) + db.add(run) + return run + + +@pytest.mark.asyncio +async def test_runs_list_omits_the_moves(client, db): + """An applied whole-library run carries tens of thousands of entries. + Fine in Postgres, wrong in every list response.""" + run = LibraryPlacementRun( + status="applied", + moves=[{"image_id": 1, "from": "/images/A/x.png", "to": "/images/a/x.png"}], + planned_count=1, moved_count=1, + ) + db.add(run) + await db.flush() + + resp = await client.get("/api/cleanup/placement/runs") + assert resp.status_code == 200 + body = await resp.get_json() + assert body["runs"][0]["planned_count"] == 1 + assert "moves" not in body["runs"][0] + + +@pytest.mark.asyncio +async def test_run_detail_carries_the_moves(client, db): + """The detail IS the preview the operator reads before agreeing.""" + run = LibraryPlacementRun( + status="ready", + moves=[{"image_id": 7, "from": "/images/Conto/x.png", "to": "/images/conto/x.png"}], + planned_count=1, + ) + db.add(run) + await db.flush() + + resp = await client.get(f"/api/cleanup/placement/runs/{run.id}") + assert resp.status_code == 200 + body = await resp.get_json() + assert body["moves"][0]["from"] == "/images/Conto/x.png" + assert body["moves"][0]["to"] == "/images/conto/x.png" + + +@pytest.mark.asyncio +async def test_run_detail_404s_for_an_unknown_run(client): + resp = await client.get("/api/cleanup/placement/runs/999999") + assert resp.status_code == 404 + + +@pytest.mark.asyncio +async def test_plan_rejects_a_non_integer_artist(client): + resp = await client.post( + "/api/cleanup/placement/plan", json={"artist_id": "conto"}, + ) + assert resp.status_code == 400 + assert (await resp.get_json())["error"] == "invalid_artist_id" + + +@pytest.mark.asyncio +async def test_plan_accepts_an_artist_scope(client, db, monkeypatch): + sent = {} + from backend.app.tasks import library_placement + + monkeypatch.setattr( + library_placement.plan_placement, "delay", + lambda artist_id=None: sent.update(artist_id=artist_id), + ) + artist = Artist(name="Conto", slug="conto") + db.add(artist) + await db.flush() + + resp = await client.post( + "/api/cleanup/placement/plan", json={"artist_id": artist.id}, + ) + assert resp.status_code == 202 + assert sent["artist_id"] == artist.id + + +@pytest.mark.asyncio +async def test_apply_refuses_a_run_that_is_not_ready(client, db): + """The gate is here as well as in the service — an applied run must not + be re-applied by a stray POST.""" + run = _run(db, status="applied") + await db.flush() + + resp = await client.post(f"/api/cleanup/placement/runs/{run.id}/apply") + assert resp.status_code == 400 + assert (await resp.get_json())["error"] == "not_ready" + + +@pytest.mark.asyncio +async def test_revert_refuses_a_run_that_was_never_applied(client, db): + run = _run(db, status="ready") + await db.flush() + + resp = await client.post(f"/api/cleanup/placement/runs/{run.id}/revert") + assert resp.status_code == 400 + assert (await resp.get_json())["error"] == "not_applied" + + +@pytest.mark.asyncio +async def test_apply_dispatches_for_a_ready_run(client, db, monkeypatch): + sent = {} + from backend.app.tasks import library_placement + + monkeypatch.setattr( + library_placement.apply_placement, "delay", + lambda run_id: sent.update(run_id=run_id), + ) + run = _run(db, status="ready") + await db.flush() + + resp = await client.post(f"/api/cleanup/placement/runs/{run.id}/apply") + assert resp.status_code == 202 + assert sent["run_id"] == run.id + # Dispatch only — the endpoint must not have moved anything itself. + still = (await db.execute( + select(LibraryPlacementRun.status) + .where(LibraryPlacementRun.id == run.id) + )).scalar_one() + assert still == "ready"