diff --git a/alembic/versions/0099_library_placement_run.py b/alembic/versions/0099_library_placement_run.py new file mode 100644 index 0000000..8907a7f --- /dev/null +++ b/alembic/versions/0099_library_placement_run.py @@ -0,0 +1,99 @@ +"""library_placement_run — the placement reconciler's plan/apply/undo ledger. + +Milestone #421 step 3. The survey (#4245) measured 33,789 ImageRecord rows +sitting outside their artist's canonical directory, across 56 artists. This +table holds one run of the sweep that trues them up: the plan, what it did, +and where every file came from. + +## Why the moves live in a table rather than a log line + +`ImageRecord.path` is the only pointer at the bytes, so a move rewrites the +row. Once that write lands, the previous location exists nowhere — unless it +was recorded first. `moves` is that record, which is what makes a 33,789-file +operation something the operator can undo per artist after looking at the +result, rather than a one-way door. + +An `applied` row is therefore HISTORY, not state (lesson #4226). Any future +retention on this table may prune `ready`, `cancelled` and `error` runs; an +`applied` one is only disposable once someone decides undo is no longer +wanted. That is deliberately not a timer's decision, and no pruning is added +here. + +Revision ID: 0099 +Revises: 0098 +Create Date: 2026-09-21 + +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects import postgresql + +revision: str = "0099" +down_revision: Union[str, None] = "0098" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "library_placement_run", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column( + "status", sa.String(length=16), server_default="running", + nullable=False, + ), + # SET NULL, not CASCADE: deleting an artist must not destroy the + # record of where their files were moved. + sa.Column("artist_id", sa.Integer(), nullable=True), + sa.Column( + "started_at", sa.DateTime(timezone=True), + server_default=sa.text("now()"), nullable=False, + ), + sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True), + sa.Column( + "planned_count", sa.Integer(), server_default="0", nullable=False, + ), + sa.Column( + "moved_count", sa.Integer(), server_default="0", nullable=False, + ), + sa.Column( + "refused_count", sa.Integer(), server_default="0", nullable=False, + ), + sa.Column( + "moves", postgresql.JSONB(astext_type=sa.Text()), + server_default=sa.text("'[]'::jsonb"), nullable=False, + ), + sa.Column( + "refusals", postgresql.JSONB(astext_type=sa.Text()), + server_default=sa.text("'[]'::jsonb"), nullable=False, + ), + sa.Column("error", sa.Text(), nullable=True), + sa.ForeignKeyConstraint( + ["artist_id"], ["artist.id"], + name="fk_library_placement_run_artist_id", ondelete="SET NULL", + ), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index( + "ix_library_placement_run_status", "library_placement_run", ["status"], + ) + op.create_index( + "ix_library_placement_run_artist_id", "library_placement_run", + ["artist_id"], + ) + + +def downgrade() -> None: + # Dropping this table destroys the only record of where moved files came + # from. That is correct for a downgrade — the code that reads it is going + # away too — but it is worth saying out loud rather than discovering. + op.drop_index( + "ix_library_placement_run_artist_id", + table_name="library_placement_run", + ) + op.drop_index( + "ix_library_placement_run_status", table_name="library_placement_run", + ) + op.drop_table("library_placement_run") diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index 9e085f4..b9bcf4a 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -22,6 +22,7 @@ from .import_batch import ImportBatch from .import_settings import ImportSettings from .import_task import ImportTask from .library_audit_run import LibraryAuditRun +from .library_placement_run import LibraryPlacementRun from .membership_sync import MembershipSync from .ml_settings import MLSettings from .patreon_failed_media import PatreonFailedMedia @@ -85,6 +86,7 @@ __all__ = [ "ImportTask", "ImportSettings", "LibraryAuditRun", + "LibraryPlacementRun", "MembershipSync", "MLSettings", "HeadAutoApplyRun", diff --git a/backend/app/models/library_placement_run.py b/backend/app/models/library_placement_run.py new file mode 100644 index 0000000..da1fbfc --- /dev/null +++ b/backend/app/models/library_placement_run.py @@ -0,0 +1,99 @@ +"""LibraryPlacementRun — one run of the placement reconciler (milestone #421). + +The library is keyed on the Artist row's `slug`, one directory per artist. +Every writer agrees on that now (`utils.paths.canonical_subdir`, task #4244), +but ~33,789 rows were written under older rules and sit in some other +artist's directory. This row is a run of the sweep that trues them up. + +State machine, mirroring LibraryAuditRun: + + running -> ready -> applied -> reverted + \\-> cancelled + (any) -> error + +## The `moves` column does three jobs + +`moves` is the plan: `[{"image_id": 1, "from": "...", "to": "..."}, ...]`. + + 1. **Preview.** It is what the operator reads before agreeing. + 2. **Apply.** The apply executes THIS list rather than re-deriving the set, + so the preview cannot describe a different set from the apply. That is + rule 93's guarantee reached the way LibraryAuditRun reaches it — the + plan is materialised, not recomputed. + 3. **Revert.** `from` is retained, so a batch that looks wrong in the + gallery goes back where it came from. + +## An applied run IS the undo ledger — it must never be pruned + +This is the trap lesson #4226 names: a record that answers both "what is the +current plan" and "what happened" gets deleted by whatever forgets the first. +A `ready` run is disposable state. An `applied` run is HISTORY, and it is the +only record of where 33,789 files used to be — delete it and the moves become +irreversible. + +No pruning exists for this table today, and that is deliberate. If retention +is ever added here, it may prune `ready`, `cancelled` and `error` runs; an +`applied` run is only safe to drop once someone decides the moves are settled +and undo is no longer wanted, which is an operator decision and not a +timer's. +""" + +from datetime import datetime +from typing import Any + +from sqlalchemy import DateTime, ForeignKey, Integer, String, Text, func, text +from sqlalchemy.dialects.postgresql import JSONB +from sqlalchemy.orm import Mapped, mapped_column + +from .base import Base + + +class LibraryPlacementRun(Base): + __tablename__ = "library_placement_run" + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + status: Mapped[str] = mapped_column( + String(16), nullable=False, default="running", index=True, + server_default="running", + ) + # running | ready | applied | reverted | cancelled | error + + # Scope. NULL = the whole library; set = one artist, which is how this is + # meant to be used — do one artist, look at it in the gallery, continue or + # revert. ondelete SET NULL rather than CASCADE: deleting an artist must + # not destroy the record of where their files were moved. + artist_id: Mapped[int | None] = mapped_column( + ForeignKey("artist.id", ondelete="SET NULL"), nullable=True, index=True, + ) + + started_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now(), + ) + finished_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True, + ) + + planned_count: Mapped[int] = mapped_column( + Integer, nullable=False, default=0, server_default="0", + ) + moved_count: Mapped[int] = mapped_column( + Integer, nullable=False, default=0, server_default="0", + ) + refused_count: Mapped[int] = mapped_column( + Integer, nullable=False, default=0, server_default="0", + ) + + # [{"image_id": int, "from": str, "to": str}, ...] — see the module + # docstring. This is the plan, the audit trail and the undo, in that order + # of appearance and in one place. + moves: Mapped[list[dict[str, Any]]] = mapped_column( + JSONB, nullable=False, default=list, server_default=text("'[]'::jsonb"), + ) + # [{"image_id": int, "reason": str}, ...] — rows the apply declined to + # touch, with why. A refusal is an expected outcome, not an error: the + # world moves between plan and apply, and every gate fails closed. + refusals: Mapped[list[dict[str, Any]]] = mapped_column( + JSONB, nullable=False, default=list, server_default=text("'[]'::jsonb"), + ) + + error: Mapped[str | None] = mapped_column(Text, nullable=True) diff --git a/backend/app/services/library_layout.py b/backend/app/services/library_layout.py index f1c2eeb..dc3961b 100644 --- a/backend/app/services/library_layout.py +++ b/backend/app/services/library_layout.py @@ -16,17 +16,23 @@ writing their own — the house shape for rule 93, snippet #3087. A preview that computes its set differently from the apply is a preview that can lie, and here the apply RENAMES the operator's art. -Nothing in this module writes. It reads rows, and it stats files when asked. +## What writes and what does not + +`survey_layout` and `plan_placement` are read-only — they report and they +record a plan. `apply_run` and `revert_run` are the only functions here that +rename a file or rewrite a row, and each does both for one row at a time, +updating the row only after its rename lands. """ from __future__ import annotations from dataclasses import dataclass, field +from datetime import UTC, datetime from pathlib import Path from sqlalchemy import func, select from sqlalchemy.orm import Session -from ..models import Artist, ImageRecord +from ..models import Artist, ImageRecord, LibraryPlacementRun from ..utils.paths import canonical_subdir # Top-level directories under the images root that are STORES, not artists. @@ -229,3 +235,167 @@ def survey_layout( report.unmovable += layout.unmovable return report + + +# --- the reconciler: plan -> apply -> revert (#4246) ------------------------- +# +# The three verbs share one materialised plan rather than each deriving its +# own set. `plan_placement` writes `LibraryPlacementRun.moves`; `apply_run` +# executes THAT list; `revert_run` walks it backwards. A preview that can +# disagree with its apply is the failure this shape exists to prevent, and +# here the apply renames the operator's art. +# +# Every step fails CLOSED. The world moves between planning and applying — +# a download lands, a supersede rewrites a path, a file is deleted — so the +# apply re-checks each row against what the plan recorded and declines the +# ones that moved on, instead of trusting a plan that may be minutes old. + + +def plan_placement( + session: Session, images_root: Path, *, artist_id: int | None = None, +) -> LibraryPlacementRun: + """Build (and persist) the move plan. Touches no files. + + `artist_id` scopes the run to one artist, which is how this is meant to be + used: do one, look at the gallery, then continue or revert. None plans the + whole library. + """ + stmt = select(Artist).order_by(Artist.slug) + if artist_id is not None: + stmt = stmt.where(Artist.id == artist_id) + artists = session.execute(stmt).scalars().all() + + candidates: list[dict] = [] + wanted: dict[str, int] = {} + for artist in artists: + rows = session.execute( + select(ImageRecord.id, ImageRecord.path) + .where(*_misplaced_conditions(images_root, artist.id, artist.slug)) + ).all() + for row_id, path in rows: + dest = destination_for(path, images_root, artist.slug) + if dest is None or dest.exists(): + continue + key = str(dest) + candidates.append({"image_id": row_id, "from": path, "to": key}) + wanted[key] = wanted.get(key, 0) + 1 + + # Two rows wanting one destination: plan NEITHER. Which of them "wins" is + # not this sweep's call, and planning one of them would silently pick a + # winner by iteration order. Counting first and filtering after is what + # makes that true — claiming as we go would quietly keep whichever came + # first. + moves = [m for m in candidates if wanted[m["to"]] == 1] + + run = LibraryPlacementRun( + status="ready", artist_id=artist_id, moves=moves, + planned_count=len(moves), + ) + session.add(run) + session.flush() + return run + + +def _move_one(src: Path, dest: Path) -> str | None: + """Rename `src` to `dest`. Returns a refusal reason, or None on success. + + A rename within one filesystem, so no copy and no free space needed. The + destination check is not a race-free guarantee — nothing here is — but it + turns the common case of "something already landed there" into a refusal + instead of a silent overwrite. + """ + if not src.exists(): + return "source missing" + if dest.exists(): + return "destination occupied" + try: + dest.parent.mkdir(parents=True, exist_ok=True) + src.rename(dest) + except OSError as exc: + return f"rename failed: {exc}" + return None + + +def apply_run( + session: Session, run: LibraryPlacementRun, +) -> LibraryPlacementRun: + """Execute a `ready` run's stored plan: file and row together, per row. + + The row is updated ONLY after its rename lands, so a refused or failed + 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. + """ + 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: + record = session.get(ImageRecord, move["image_id"]) + if record is None: + refusals.append({"image_id": move["image_id"], "reason": "row gone"}) + continue + if record.path != move["from"]: + # Something rewrote this row since the plan was built — a + # supersede, or an earlier run. The plan is stale for it. + refusals.append({ + "image_id": move["image_id"], "reason": "row moved since planning", + }) + continue + reason = _move_one(Path(move["from"]), Path(move["to"])) + if reason is not None: + refusals.append({"image_id": move["image_id"], "reason": reason}) + continue + record.path = move["to"] + moved.append(move) + + run.moves = moved + run.refusals = refusals + run.moved_count = len(moved) + run.refused_count = len(refusals) + run.status = "applied" + run.finished_at = datetime.now(UTC) + session.flush() + return run + + +def revert_run( + session: Session, run: LibraryPlacementRun, +) -> LibraryPlacementRun: + """Put an applied run's files back where they came from. + + This is why `from` is retained. It is the answer to "do one artist, look + at it, and undo if it reads wrong" — which is a cheaper way to settle + whether artist_id or the folder held the truth (#4257) than arguing it + from a sample. + + 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. + """ + if run.status != "applied": + raise ValueError(f"run {run.id} is {run.status}, not applied") + + refusals: list[dict] = [] + reverted = 0 + for move in run.moves: + record = session.get(ImageRecord, move["image_id"]) + if record is None or record.path != move["to"]: + refusals.append({ + "image_id": move["image_id"], "reason": "row changed since apply", + }) + continue + reason = _move_one(Path(move["to"]), Path(move["from"])) + if reason is not None: + refusals.append({"image_id": move["image_id"], "reason": reason}) + continue + record.path = move["from"] + reverted += 1 + + run.refusals = refusals + run.refused_count = len(refusals) + run.moved_count = run.moved_count - reverted + run.status = "reverted" + run.finished_at = datetime.now(UTC) + session.flush() + return run diff --git a/tests/test_library_layout.py b/tests/test_library_layout.py index 52cd88d..1594082 100644 --- a/tests/test_library_layout.py +++ b/tests/test_library_layout.py @@ -213,3 +213,191 @@ def test_survey_is_read_only(db_sync, tmp_path): assert db_sync.execute( select(ImageRecord.path).where(ImageRecord.id == rec.id) ).scalar_one() == before + + +# --- plan / apply / revert (#4246) ------------------------------------------ + + +def _staged(db, tmp_path, slug, stray, name="x.png", n=100): + """An artist with one file sitting in `stray`'s directory.""" + artist = _artist(db, slug.title(), slug) + src = tmp_path / stray / name + src.parent.mkdir(parents=True, exist_ok=True) + src.write_bytes(b"pixels") + rec = _image(db, str(src), artist, n) + return artist, rec, src + + +@pytest.mark.integration +def test_plan_records_where_each_file_came_from(db_sync, tmp_path): + from backend.app.services.library_layout import plan_placement + + _, rec, src = _staged(db_sync, tmp_path, "conto", "Conto", n=20) + run = plan_placement(db_sync, tmp_path) + + assert run.status == "ready" + assert run.planned_count == 1 + assert run.moves == [{ + "image_id": rec.id, + "from": str(src), + "to": str(tmp_path / "conto" / "x.png"), + }] + # Planning touches nothing. + assert src.exists() + assert db_sync.get(ImageRecord, rec.id).path == str(src) + + +@pytest.mark.integration +def test_apply_moves_file_and_row_together(db_sync, tmp_path): + from backend.app.services.library_layout import apply_run, plan_placement + + _, rec, src = _staged(db_sync, tmp_path, "conto", "Conto", n=21) + run = apply_run(db_sync, plan_placement(db_sync, tmp_path)) + + dest = tmp_path / "conto" / "x.png" + assert run.status == "applied" + assert run.moved_count == 1 and run.refused_count == 0 + assert dest.exists() and not src.exists() + db_sync.expire_all() + assert db_sync.get(ImageRecord, rec.id).path == str(dest) + + +@pytest.mark.integration +def test_revert_puts_it_back(db_sync, tmp_path): + """The whole reason `from` is retained: do one artist, look, undo.""" + from backend.app.services.library_layout import ( + apply_run, + plan_placement, + revert_run, + ) + + _, rec, src = _staged(db_sync, tmp_path, "conto", "Conto", n=22) + run = revert_run(db_sync, apply_run(db_sync, plan_placement(db_sync, tmp_path))) + + assert run.status == "reverted" + assert src.exists() + assert not (tmp_path / "conto" / "x.png").exists() + db_sync.expire_all() + assert db_sync.get(ImageRecord, rec.id).path == str(src) + + +@pytest.mark.integration +def test_apply_refuses_a_row_that_moved_since_planning(db_sync, tmp_path): + """A supersede or an earlier run can rewrite a path between plan and + apply. The stale entry is declined, not forced.""" + from backend.app.services.library_layout import apply_run, plan_placement + + _, rec, src = _staged(db_sync, tmp_path, "conto", "Conto", n=23) + run = plan_placement(db_sync, tmp_path) + + elsewhere = tmp_path / "conto" / "already-here.png" + elsewhere.parent.mkdir(parents=True, exist_ok=True) + elsewhere.write_bytes(b"pixels") + rec.path = str(elsewhere) + db_sync.flush() + + run = apply_run(db_sync, run) + + assert run.moved_count == 0 and run.refused_count == 1 + assert run.refusals[0]["reason"] == "row moved since planning" + assert src.exists() # untouched + + +@pytest.mark.integration +def test_apply_never_overwrites_an_occupied_destination(db_sync, tmp_path): + from backend.app.services.library_layout import apply_run, plan_placement + + _, rec, src = _staged(db_sync, tmp_path, "conto", "Conto", n=24) + run = plan_placement(db_sync, tmp_path) + + squatter = tmp_path / "conto" / "x.png" + squatter.parent.mkdir(parents=True, exist_ok=True) + squatter.write_bytes(b"someone else") + + run = apply_run(db_sync, run) + + assert run.refused_count == 1 + assert run.refusals[0]["reason"] == "destination occupied" + assert squatter.read_bytes() == b"someone else" + db_sync.expire_all() + assert db_sync.get(ImageRecord, rec.id).path == str(src) + + +@pytest.mark.integration +def test_apply_leaves_the_row_alone_when_the_source_is_gone(db_sync, tmp_path): + from backend.app.services.library_layout import apply_run, plan_placement + + _, rec, src = _staged(db_sync, tmp_path, "conto", "Conto", n=25) + run = plan_placement(db_sync, tmp_path) + src.unlink() + + run = apply_run(db_sync, run) + + assert run.refusals[0]["reason"] == "source missing" + db_sync.expire_all() + # The row still points at the missing file rather than at a file that + # was never created — a broken row is recoverable, a lying one is not. + assert db_sync.get(ImageRecord, rec.id).path == str(src) + + +@pytest.mark.integration +def test_plan_scopes_to_one_artist(db_sync, tmp_path): + """Per-artist scope is what makes this incremental instead of one + irreversible sweep.""" + from backend.app.services.library_layout import plan_placement + + conto, _, _ = _staged(db_sync, tmp_path, "conto", "Conto", n=26) + _staged(db_sync, tmp_path, "maewix", "Maewix", name="y.png", n=27) + + run = plan_placement(db_sync, tmp_path, artist_id=conto.id) + + assert run.planned_count == 1 + assert run.artist_id == conto.id + assert "Conto" in run.moves[0]["from"] + + +@pytest.mark.integration +def test_plan_skips_both_rows_when_two_want_one_destination(db_sync, tmp_path): + """Which of two colliding rows 'wins' is not this sweep's call.""" + from backend.app.services.library_layout import plan_placement + + artist = _artist(db_sync, "Sticky", "sticky") + for stray, n in (("StickySpoodge", 28), ("Stickyspoodge", 29)): + p = tmp_path / stray / "dup.png" + p.parent.mkdir(parents=True, exist_ok=True) + p.write_bytes(b"pixels") + _image(db_sync, str(p), artist, n) + + run = plan_placement(db_sync, tmp_path) + + assert run.planned_count == 0 + + +@pytest.mark.integration +def test_thumbnails_do_not_move(db_sync, tmp_path): + """Thumbs are sha-addressed (`thumbs//.jpg`), not path-keyed, so + a placement move must not touch them. Pinned so nobody 'fixes' it.""" + from backend.app.services.library_layout import apply_run, plan_placement + + artist, rec, _ = _staged(db_sync, tmp_path, "conto", "Conto", n=30) + thumb = tmp_path / "thumbs" / "ab" / "abc.jpg" + thumb.parent.mkdir(parents=True, exist_ok=True) + thumb.write_bytes(b"thumb") + rec.thumbnail_path = str(thumb) + db_sync.flush() + + apply_run(db_sync, plan_placement(db_sync, tmp_path)) + + db_sync.expire_all() + assert thumb.exists() + assert db_sync.get(ImageRecord, rec.id).thumbnail_path == str(thumb) + + +@pytest.mark.integration +def test_apply_refuses_a_run_that_is_not_ready(db_sync, tmp_path): + from backend.app.services.library_layout import apply_run, plan_placement + + _staged(db_sync, tmp_path, "conto", "Conto", n=31) + run = apply_run(db_sync, plan_placement(db_sync, tmp_path)) + with pytest.raises(ValueError): + apply_run(db_sync, run)