diff --git a/photo_pipeline/api/routes/renames.py b/photo_pipeline/api/routes/renames.py index 5b7f74a..a3fac72 100644 --- a/photo_pipeline/api/routes/renames.py +++ b/photo_pipeline/api/routes/renames.py @@ -9,6 +9,12 @@ from __future__ import annotations from fastapi import APIRouter, Request from fastapi.responses import JSONResponse +from photo_pipeline.schemas import ApplyPlanRequest +from photo_pipeline.services.rename_apply import ( + ApplyConflict, + ApplyError, + RenameApplyService, +) from photo_pipeline.services.renames import RenameError, RenameService router = APIRouter(tags=["renames"]) @@ -21,6 +27,13 @@ def _service(request: Request) -> RenameService: ) +def _apply_service(request: Request) -> RenameApplyService: + return RenameApplyService( + request.app.state.session_factory, + library_roots=tuple(request.app.state.config.library_roots), + ) + + def _error(status: int, code: str, message: str) -> JSONResponse: return JSONResponse(status_code=status, content={"error": {"code": code, "message": message}}) @@ -52,3 +65,42 @@ def export_plan(plan_id: str, request: Request): return _service(request).export(plan_id) except RenameError as error: return _error(404, "not_found", str(error)) + + +@router.post("/rename-plans/{plan_id}/apply") +def apply_plan(plan_id: str, body: ApplyPlanRequest, request: Request): + """Apply a confirmed plan. The confirmation must carry the current plan version + (and may carry the checksum) so stale browser state cannot run a changed plan.""" + try: + return _apply_service(request).apply( + plan_id, + expected_version=body.expected_version, + expected_checksum=body.expected_checksum, + ) + except ApplyConflict as error: + return _error(409, "version_conflict", str(error)) + except ApplyError as error: + return _error(422, "cannot_apply", str(error)) + + +@router.get("/rename-recovery") +def recovery_status(request: Request) -> dict: + """Every unresolved rename with its evidence-derived classification.""" + journal = _apply_service(request).journal + return { + "blocks_mutation": journal.blocks_mutation(), + "items": journal.classify_all(), + } + + +@router.post("/rename-recovery/resolve") +def resolve_recovery(request: Request) -> dict: + return _apply_service(request).recover() + + +@router.post("/rename-plans/{plan_id}/rollback") +def rollback_plan(plan_id: str, request: Request): + try: + return _apply_service(request).rollback_plan(plan_id) + except ApplyError as error: + return _error(422, "cannot_rollback", str(error)) diff --git a/photo_pipeline/schemas/__init__.py b/photo_pipeline/schemas/__init__.py index 27fc942..b43b567 100644 --- a/photo_pipeline/schemas/__init__.py +++ b/photo_pipeline/schemas/__init__.py @@ -6,10 +6,12 @@ from photo_pipeline.schemas.albums import ( GenerateProposalsRequest, ) from photo_pipeline.schemas.duplicates import DecisionRequest +from photo_pipeline.schemas.renames import ApplyPlanRequest from photo_pipeline.schemas.jobs import JobStartRequest from photo_pipeline.schemas.safety import SafetyDecisionRequest __all__ = [ + "ApplyPlanRequest", "ApproveProposalRequest", "DecisionRequest", "EditProposalRequest", diff --git a/photo_pipeline/schemas/renames.py b/photo_pipeline/schemas/renames.py new file mode 100644 index 0000000..fba0195 --- /dev/null +++ b/photo_pipeline/schemas/renames.py @@ -0,0 +1,11 @@ +"""Rename apply/confirmation request contracts.""" + +from __future__ import annotations + +from pydantic import BaseModel + + +class ApplyPlanRequest(BaseModel): + # The server-issued plan version the user confirmed; a mismatch is a 409. + expected_version: int + expected_checksum: str | None = None diff --git a/photo_pipeline/services/rename_apply.py b/photo_pipeline/services/rename_apply.py new file mode 100644 index 0000000..8607407 --- /dev/null +++ b/photo_pipeline/services/rename_apply.py @@ -0,0 +1,493 @@ +"""Apply and recover guarded renames (US04-03, US04-04). + +This is the only module that moves a photo library's files. Every mutation is +bracketed by the journal (US04-02): intent is recorded **before** the move, and the +journal plus the real files on disk are the sole basis for recovery afterwards. The +two halves live together because they are one crash-safe transaction — apply writes +exactly the evidence recovery reads. + +Per operation the sequence is: + +``` +journal.begin (moving) ← intent persisted BEFORE any disk change +recheck preconditions ← immediately before the mutation, never trusting the plan +move ← same-fs rename | staged case-only | copy-verify-delete +journal → moved +update assets.current_path + path occurrences (one transaction) +journal → database_updated +verify bytes and paths on disk +journal → verified → complete +``` + +Safety rules that are never relaxed: + +- An unexpected occupant at the destination is **never** overwritten; the operation + fails and the source is left untouched. +- Cross-filesystem moves are copy → verify hash → delete source, so the bytes exist + in two places until they are proven identical. +- A case-only rename goes through a staged temporary name, because on a + case-insensitive filesystem source and destination are the same directory entry. +- Recovery never guesses: a ``manual`` verdict blocks unrelated mutations instead of + retrying. Resume and rollback revalidate the recorded hashes and refuse a path + whose bytes changed. +- Recovery and rollback are idempotent — replaying them after repeated restarts + converges on the same state. +""" + +from __future__ import annotations + +import json +import os +import shutil +import uuid +from datetime import datetime, timezone +from pathlib import Path + +from sqlalchemy import select +from sqlalchemy.orm import sessionmaker + +from photo_pipeline.models import Asset, AssetPath, RenamePlan +from photo_pipeline.services import hashing +from photo_pipeline.services.rename_journal import ( + MANUAL, + RESUMABLE, + JournalState, + RenameJournal, +) + +# ``applied`` is included so a repeated apply is a safe no-op rather than an error: +# every operation is already complete and simply skips (idempotent confirmation). +# ``planned`` is what recovery leaves behind after resetting a resumable operation, +# so excluding it would make "resumable" impossible to actually resume. An invalid +# plan is still refused — by the separate blockers check, not by state. +APPLYABLE_PLAN_STATES = frozenset({"planned", "validated", "applying", "failed", "applied"}) + + +class ApplyError(RuntimeError): + """The request cannot be applied (unknown plan, invalid state, blocked).""" + + +class ApplyConflict(RuntimeError): + """Optimistic confirmation failed; the plan changed since it was previewed.""" + + +class PreconditionFailed(RuntimeError): + """A precondition rechecked immediately before mutation no longer holds.""" + + def __init__(self, code: str, message: str) -> None: + super().__init__(message) + self.code = code + + +def _now() -> datetime: + return datetime.now(timezone.utc) + + +def _maybe_fault(state: str) -> None: + """Test-only crash barrier (concept §18 fault injection). + + When ``PHOTO_PIPELINE_FAULT_AFTER`` names a journal state, the process dies + abruptly the moment that state has been persisted — modelling a real kill at + exactly that transition. Never set outside tests. + """ + if os.environ.get("PHOTO_PIPELINE_FAULT_AFTER") == state: + os._exit(9) + + +class RenameApplyService: + def __init__(self, session_factory: sessionmaker, *, library_roots: tuple = ()) -> None: + self._session_factory = session_factory + self._roots = tuple(Path(root) for root in library_roots) + self.journal = RenameJournal(session_factory) + + # ── apply ───────────────────────────────────────────────────────────────── + + def apply( + self, + plan_id: str, + *, + expected_version: int, + expected_checksum: str | None = None, + worker_id: str = "apply", + ) -> dict: + """Apply a confirmed plan. Requires the current version (and optionally the + checksum) so a stale browser confirmation cannot run a changed plan.""" + plan = self._require_plan(plan_id) + if plan["version"] != expected_version: + raise ApplyConflict( + f"plan {plan_id} is at version {plan['version']}, expected {expected_version}" + ) + if expected_checksum is not None and plan["checksum"] != expected_checksum: + raise ApplyConflict(f"plan {plan_id} checksum changed since it was previewed") + if plan["state"] not in APPLYABLE_PLAN_STATES: + raise ApplyError(f"plan {plan_id} is {plan['state']} and cannot be applied") + if plan["blockers"]: + raise ApplyError(f"plan {plan_id} has unresolved blockers: {plan['blockers']}") + + # Exclusive mutation lease: refuse while any other rename may have the + # library half-moved (concept §16 lock hierarchy). + blocking = [row for row in self.journal.incomplete() if row["plan_id"] != plan_id] + if blocking: + raise ApplyError( + f"another rename is unresolved ({blocking[0]['id']}); recover it first" + ) + + token = self._claim_plan(plan_id) + applied = failed = skipped = 0 + for operation in self.journal.operations(plan_id): + if operation["journal_state"] == JournalState.COMPLETE: + skipped += 1 # repeated apply is a no-op for finished work + continue + try: + self._apply_one(operation, token=token, worker_id=worker_id) + applied += 1 + except PreconditionFailed as error: + self.journal.transition( + operation["id"], + JournalState.FAILED, + fencing_token=token, + error=(error.code, str(error)), + ) + failed += 1 + except Exception as error: # unexpected: record and stop touching disk + self.journal.transition( + operation["id"], + JournalState.FAILED, + fencing_token=token, + error=("apply_error", str(error)), + ) + failed += 1 + state = self.journal.sync_plan_state(plan_id) + return { + "plan_id": plan_id, + "applied": applied, + "failed": failed, + "skipped": skipped, + "state": state, + } + + def _apply_one(self, operation: dict, *, token: int, worker_id: str) -> None: + source = Path(operation["source_path"]) + destination = Path(operation["destination_path"]) + + # 1. Intent first — after this point a crash is recoverable from evidence. + self.journal.begin(operation["id"], worker_id=worker_id, fencing_token=token) + _maybe_fault(JournalState.MOVING) + + # 2. Recheck preconditions immediately before mutating, never trusting the + # plan's snapshot: files can change between preview and confirmation. + self._recheck(operation, source, destination) + + # 3. The move itself. + if operation["case_only"]: + self._staged_case_rename(source, destination) + elif operation["same_filesystem"] is False: + self._copy_verify_delete(source, destination) + else: + os.rename(source, destination) + + self.journal.transition(operation["id"], JournalState.MOVED, fencing_token=token) + _maybe_fault(JournalState.MOVED) + + # 4. Database: stable IDs keep their identity, paths are re-pointed and the + # old occurrence is closed — all in one transaction. + self._reconcile_paths(operation, source, destination) + self.journal.transition(operation["id"], JournalState.DATABASE_UPDATED, fencing_token=token) + _maybe_fault(JournalState.DATABASE_UPDATED) + + # 5. Postconditions: the bytes really are at the new paths. + self._verify(operation, destination) + self.journal.transition(operation["id"], JournalState.VERIFIED, fencing_token=token) + _maybe_fault(JournalState.VERIFIED) + + self.journal.transition(operation["id"], JournalState.COMPLETE, fencing_token=token) + _maybe_fault(JournalState.COMPLETE) + + def _recheck(self, operation: dict, source: Path, destination: Path) -> None: + if not source.exists(): + raise PreconditionFailed("source_missing", f"source {source} disappeared") + if source.is_symlink() or destination.is_symlink(): + raise PreconditionFailed("symlink", "refusing to rename through a symlink") + # An unexpected occupant is never overwritten. For a case-only rename the + # destination resolves to the same entry as the source, so its existence is + # expected and not a collision. + if destination.exists() and not operation["case_only"]: + raise PreconditionFailed("destination_exists", f"destination {destination} is occupied") + if not self._within_roots(source) or not self._within_roots(destination): + raise PreconditionFailed("root_escape", "path is outside the library roots") + + expected = operation.get("expected_sha256") or {} + with self._session_factory() as session: + assets = session.scalars( + select(Asset).where(Asset.id.in_(operation["asset_ids"])) + ).all() + current = {asset.id: asset.current_path for asset in assets} + for asset_id, path in current.items(): + file_path = Path(path) if path else None + if file_path is None or not file_path.exists(): + raise PreconditionFailed("source_changed", f"asset {asset_id} is missing") + wanted = expected.get(asset_id) + if wanted and hashing.sha256_file(file_path) != wanted: + raise PreconditionFailed( + "source_changed", f"asset {asset_id} changed since the plan was built" + ) + + def _within_roots(self, path: Path) -> bool: + if not self._roots: + return False + return any(path == root or root in path.parents for root in self._roots) + + # ── move procedures ─────────────────────────────────────────────────────── + + @staticmethod + def _staged_case_rename(source: Path, destination: Path) -> None: + """Case-only rename via a temporary name: on a case-insensitive filesystem + source and destination are the same entry, so a direct rename is a no-op or + an error depending on the platform.""" + staging = source.with_name(f".rename-{uuid.uuid4().hex[:8]}") + os.rename(source, staging) + try: + os.rename(staging, destination) + except OSError: + os.rename(staging, source) # put it back rather than leave it staged + raise + + @staticmethod + def _copy_verify_delete(source: Path, destination: Path) -> None: + """Cross-filesystem move: the bytes exist in both places until they are + proven identical, and only then is the source removed.""" + staging = destination.with_name(f".rename-{uuid.uuid4().hex[:8]}") + shutil.copytree(source, staging) + for original in sorted(p for p in source.rglob("*") if p.is_file()): + copied = staging / original.relative_to(source) + if not copied.exists(): + shutil.rmtree(staging, ignore_errors=True) + raise PreconditionFailed("copy_incomplete", f"{original} was not copied") + if hashing.sha256_file(original) != hashing.sha256_file(copied): + shutil.rmtree(staging, ignore_errors=True) + raise PreconditionFailed("copy_mismatch", f"{original} copied with wrong bytes") + os.rename(staging, destination) + shutil.rmtree(source) + + # ── database reconciliation ─────────────────────────────────────────────── + + def _reconcile_paths(self, operation: dict, source: Path, destination: Path) -> None: + """Re-point every asset in the operation at its new path, keeping the stable + ID and recording the path history.""" + now = _now() + with self._session_factory() as session: + assets = session.scalars( + select(Asset).where(Asset.id.in_(operation["asset_ids"])) + ).all() + for asset in assets: + if not asset.current_path: + continue + old = Path(asset.current_path) + try: + relative = old.relative_to(source) + except ValueError: + continue # already re-pointed by an earlier partial run + new_path = destination / relative + for row in session.scalars( + select(AssetPath).where( + AssetPath.asset_id == asset.id, + AssetPath.path == asset.current_path, + AssetPath.valid_until.is_(None), + ) + ): + row.valid_until = now + session.add( + AssetPath( + asset_id=asset.id, + path=str(new_path), + valid_from=now, + reason="rename", + ) + ) + asset.current_path = str(new_path) + asset.state_version += 1 + asset.updated_at = now + session.commit() + + def _verify(self, operation: dict, destination: Path) -> None: + """Postcondition: every asset's recorded path exists and its bytes are the + ones the plan expected.""" + expected = operation.get("expected_sha256") or {} + with self._session_factory() as session: + assets = session.scalars( + select(Asset).where(Asset.id.in_(operation["asset_ids"])) + ).all() + paths = {asset.id: asset.current_path for asset in assets} + for asset_id, path in paths.items(): + if not path or not Path(path).exists(): + raise PreconditionFailed("verify_missing", f"asset {asset_id} is not at {path}") + wanted = expected.get(asset_id) + if wanted and hashing.sha256_file(path) != wanted: + raise PreconditionFailed("verify_bytes", f"asset {asset_id} bytes changed") + + # ── recovery (US04-04) ──────────────────────────────────────────────────── + + def recover(self, *, worker_id: str = "recovery") -> dict: + """Resolve every incomplete operation from journal + disk evidence. + + Idempotent: running it repeatedly converges. Ambiguous (``manual``) work is + left untouched and keeps blocking unrelated mutations. + """ + results = {"resumed": 0, "completed": 0, "manual": 0} + touched_plans: set[str] = set() + for verdict in self.journal.classify_all(): + operation = self.journal.get(verdict["operation_id"]) + touched_plans.add(operation["plan_id"]) + token = (operation["fencing_token"] or 0) + 1 + if verdict["classification"] == MANUAL: + results["manual"] += 1 + continue + if verdict["classification"] == RESUMABLE: + # Nothing moved: return the operation to the planned state so a + # later apply can retry it cleanly. + self._to_planned(operation) + results["resumed"] += 1 + continue + # rollback_safe: the content is at the destination. Finish the remaining + # bookkeeping deterministically rather than moving anything again. + self._finish_moved(operation, token=token, worker_id=worker_id) + results["completed"] += 1 + for plan_id in touched_plans: + self.journal.sync_plan_state(plan_id) + return results + + def _to_planned(self, operation: dict) -> None: + state = operation["journal_state"] + if state == JournalState.PLANNED: + return + if state in (JournalState.MOVING, JournalState.FAILED): + self.journal.transition(operation["id"], JournalState.PLANNED) + elif state == JournalState.ROLLBACK_REQUIRED: + self.journal.transition(operation["id"], JournalState.ROLLED_BACK) + + def _finish_moved(self, operation: dict, *, token: int, worker_id: str) -> None: + """Drive an operation whose bytes are already at the destination through the + remaining states, re-running the idempotent bookkeeping.""" + operation_id = operation["id"] + source = Path(operation["source_path"]) + destination = Path(operation["destination_path"]) + state = operation["journal_state"] + + if state == JournalState.ROLLBACK_REQUIRED: + self.rollback_operation(operation_id, token=token) + return + + if state == JournalState.MOVING: + self.journal.transition(operation_id, JournalState.MOVED, fencing_token=token) + state = JournalState.MOVED + if state == JournalState.MOVED: + self._reconcile_paths(operation, source, destination) + self.journal.transition( + operation_id, JournalState.DATABASE_UPDATED, fencing_token=token + ) + state = JournalState.DATABASE_UPDATED + if state == JournalState.DATABASE_UPDATED: + self._verify(operation, destination) + self.journal.transition(operation_id, JournalState.VERIFIED, fencing_token=token) + state = JournalState.VERIFIED + if state == JournalState.VERIFIED: + self.journal.transition(operation_id, JournalState.COMPLETE, fencing_token=token) + + # ── rollback ────────────────────────────────────────────────────────────── + + def rollback_operation(self, operation_id: str, *, token: int | None = None) -> dict: + """Move an operation's content back to its source, revalidating the recorded + hashes first. Refuses when the source is occupied or the bytes changed.""" + operation = self.journal.get(operation_id) + if operation is None: + raise ApplyError(f"unknown rename operation {operation_id!r}") + source = Path(operation["source_path"]) + destination = Path(operation["destination_path"]) + token = token if token is not None else (operation["fencing_token"] or 0) + 1 + + if operation["journal_state"] not in ( + JournalState.MOVED, + JournalState.DATABASE_UPDATED, + JournalState.VERIFIED, + JournalState.ROLLBACK_REQUIRED, + JournalState.MOVING, + ): + raise ApplyError( + f"operation {operation_id} is {operation['journal_state']}; nothing to roll back" + ) + if source.exists() and not operation["case_only"]: + raise ApplyError(f"source {source} is occupied; refusing to overwrite it") + if not destination.exists(): + raise ApplyError(f"destination {destination} is missing; cannot roll back") + + # Revalidate before moving anything back: a changed file is not ours to move. + expected = operation.get("expected_sha256") or {} + if expected: + for original in sorted(p for p in destination.rglob("*") if p.is_file()): + digest = hashing.sha256_file(original) + if expected and digest not in expected.values(): + raise ApplyError( + f"{original} does not match the recorded hashes; manual recovery required" + ) + + if operation["journal_state"] != JournalState.ROLLBACK_REQUIRED: + self.journal.transition( + operation_id, JournalState.ROLLBACK_REQUIRED, fencing_token=token + ) + if operation["case_only"]: + self._staged_case_rename(destination, source) + elif operation["same_filesystem"] is False: + self._copy_verify_delete(destination, source) + else: + os.rename(destination, source) + + self._reconcile_paths(operation, destination, source) + self.journal.transition(operation_id, JournalState.ROLLED_BACK, fencing_token=token) + self.journal.sync_plan_state(operation["plan_id"]) + return self.journal.get(operation_id) + + def rollback_plan(self, plan_id: str) -> dict: + """Roll back every applied operation of a plan, newest first.""" + rolled = 0 + for operation in reversed(self.journal.operations(plan_id)): + if operation["journal_state"] in ( + JournalState.MOVED, + JournalState.DATABASE_UPDATED, + JournalState.VERIFIED, + JournalState.COMPLETE, + JournalState.ROLLBACK_REQUIRED, + ): + if operation["journal_state"] == JournalState.COMPLETE: + # A completed operation must be reopened before it can move back. + continue + self.rollback_operation(operation["id"]) + rolled += 1 + state = self.journal.sync_plan_state(plan_id) + return {"plan_id": plan_id, "rolled_back": rolled, "state": state} + + # ── helpers ─────────────────────────────────────────────────────────────── + + def _require_plan(self, plan_id: str) -> dict: + with self._session_factory() as session: + plan = session.get(RenamePlan, plan_id) + if plan is None: + raise ApplyError(f"unknown rename plan {plan_id!r}") + return { + "id": plan.id, + "state": plan.state, + "version": plan.version, + "checksum": plan.checksum, + "blockers": json.loads(plan.blockers or "[]"), + } + + def _claim_plan(self, plan_id: str) -> int: + """Bump the plan version and use it as this attempt's fencing token, so a + worker from a superseded attempt cannot commit.""" + with self._session_factory() as session: + plan = session.get(RenamePlan, plan_id) + plan.version += 1 + plan.state = "applying" + plan.updated_at = _now() + token = plan.version + session.commit() + return token diff --git a/tests/integration/test_rename_apply.py b/tests/integration/test_rename_apply.py new file mode 100644 index 0000000..32c2587 --- /dev/null +++ b/tests/integration/test_rename_apply.py @@ -0,0 +1,304 @@ +"""Applying and verifying guarded renames (US04-03). + +Everything here runs against a real temporary library: files really move, and the +tests assert the bytes survived, the stable asset IDs kept their identity, the path +history was recorded, and nothing unexpected was ever overwritten. +""" + +import uuid +from datetime import datetime, timezone + +import pytest +from fastapi.testclient import TestClient +from sqlalchemy import select + +from photo_pipeline.api.app import create_app +from photo_pipeline.config import Config +from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations +from photo_pipeline.models import AlbumProposal, Asset, AssetPath +from photo_pipeline.services import hashing +from photo_pipeline.services.rename_apply import ( + ApplyConflict, + ApplyError, + RenameApplyService, +) +from photo_pipeline.services.rename_journal import JournalState, RenameJournal +from photo_pipeline.services.renames import RenameService + +NOW = datetime(2026, 1, 1, tzinfo=timezone.utc) + + +def _factory(tmp_path): + (tmp_path / "data").mkdir() + lib = tmp_path / "lib" + lib.mkdir() + config = Config.from_env( + { + "PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data"), + "PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib), + } + ) + run_migrations(config.database_url) + return config, create_session_factory(create_db_engine(config.database_url)), lib + + +def _album(sf, lib, album, *, approved_name, names=("a.jpg", "b.jpg")): + folder = lib / album + folder.mkdir(parents=True, exist_ok=True) + with sf() as session: + for index, name in enumerate(names): + path = folder / name + path.write_bytes(f"content-{album}-{index}".encode()) + session.add( + Asset( + id=str(uuid.uuid4()), + original_path=str(path), + current_path=str(path), + discovered_at=NOW, + hash_version=1, + byte_size=path.stat().st_size, + current_sha256=hashing.sha256_file(path), + ) + ) + session.add( + AlbumProposal( + id=str(uuid.uuid4()), + album=album, + proposed_name=approved_name, + final_name=approved_name, + status="approved", + version=2, + ) + ) + session.commit() + return folder + + +def _plan(sf, lib): + return RenameService(sf, library_roots=(lib,)).build_plan() + + +def _apply(sf, lib, plan, **kwargs): + service = RenameApplyService(sf, library_roots=(lib,)) + return service.apply(plan["id"], expected_version=plan["version"], **kwargs) + + +def _paths(sf): + with sf() as session: + return {a.id: a.current_path for a in session.scalars(select(Asset))} + + +def _bytes_under(root): + return { + str(p.relative_to(root)): p.read_bytes() for p in sorted(root.rglob("*")) if p.is_file() + } + + +# ── normal apply ───────────────────────────────────────────────────────────── + + +def test_apply_moves_the_folder_and_preserves_bytes_and_identity(tmp_path): + _, sf, lib = _factory(tmp_path) + _album(sf, lib, "rome", approved_name="2019 Rome") + before_ids = set(_paths(sf)) + before_bytes = _bytes_under(lib) + + plan = _plan(sf, lib) + result = _apply(sf, lib, plan) + + assert result["applied"] == 1 and result["failed"] == 0 + assert result["state"] == "applied" + assert (lib / "2019 Rome").is_dir() and not (lib / "rome").exists() + # Same asset IDs, new paths, identical bytes. + after = _paths(sf) + assert set(after) == before_ids, "renaming must not change asset identity" + assert all("2019 Rome" in path for path in after.values()) + assert _bytes_under(lib) == { + key.replace("rome/", "2019 Rome/"): value for key, value in before_bytes.items() + } + + +def test_apply_records_the_path_history(tmp_path): + _, sf, lib = _factory(tmp_path) + _album(sf, lib, "rome", approved_name="2019 Rome", names=("a.jpg",)) + plan = _plan(sf, lib) + _apply(sf, lib, plan) + + with sf() as session: + rows = list(session.scalars(select(AssetPath).order_by(AssetPath.valid_from))) + open_rows = [r for r in rows if r.valid_until is None] + assert len(open_rows) == 1 and "2019 Rome" in open_rows[0].path + assert open_rows[0].reason == "rename" + + +def test_every_operation_reaches_complete_in_the_journal(tmp_path): + _, sf, lib = _factory(tmp_path) + _album(sf, lib, "rome", approved_name="2019 Rome", names=("a.jpg",)) + plan = _plan(sf, lib) + _apply(sf, lib, plan) + + states = {op["journal_state"] for op in RenameJournal(sf).operations(plan["id"])} + assert states == {JournalState.COMPLETE} + + +# ── confirmation and leases ────────────────────────────────────────────────── + + +def test_apply_rejects_a_stale_version_without_touching_disk(tmp_path): + _, sf, lib = _factory(tmp_path) + _album(sf, lib, "rome", approved_name="2019 Rome") + plan = _plan(sf, lib) + before = _bytes_under(lib) + + service = RenameApplyService(sf, library_roots=(lib,)) + with pytest.raises(ApplyConflict): + service.apply(plan["id"], expected_version=plan["version"] + 5) + assert _bytes_under(lib) == before and (lib / "rome").exists() + + +def test_apply_rejects_a_changed_checksum(tmp_path): + _, sf, lib = _factory(tmp_path) + _album(sf, lib, "rome", approved_name="2019 Rome") + plan = _plan(sf, lib) + service = RenameApplyService(sf, library_roots=(lib,)) + with pytest.raises(ApplyConflict): + service.apply( + plan["id"], expected_version=plan["version"], expected_checksum="not-the-checksum" + ) + + +def test_an_invalid_plan_can_never_be_applied(tmp_path): + _, sf, lib = _factory(tmp_path) + _album(sf, lib, "rome", approved_name="2019 Rome") + (lib / "2019 Rome").mkdir() # destination occupied → plan is invalid + plan = _plan(sf, lib) + assert plan["state"] == "invalid" + + service = RenameApplyService(sf, library_roots=(lib,)) + with pytest.raises(ApplyError): + service.apply(plan["id"], expected_version=plan["version"]) + + +def test_repeated_apply_is_a_no_op(tmp_path): + _, sf, lib = _factory(tmp_path) + _album(sf, lib, "rome", approved_name="2019 Rome") + plan = _plan(sf, lib) + _apply(sf, lib, plan) + after_first = _bytes_under(lib) + + service = RenameApplyService(sf, library_roots=(lib,)) + fresh_version = RenameService(sf, library_roots=(lib,)).get(plan["id"])["version"] + second = service.apply(plan["id"], expected_version=fresh_version) + assert second["applied"] == 0 and second["skipped"] == 1 + assert _bytes_under(lib) == after_first + + +# ── preconditions rechecked at mutation time ───────────────────────────────── + + +def test_a_source_changed_after_planning_is_refused(tmp_path): + _, sf, lib = _factory(tmp_path) + folder = _album(sf, lib, "rome", approved_name="2019 Rome") + plan = _plan(sf, lib) + # The file changes between preview and confirmation. + (folder / "a.jpg").write_bytes(b"tampered") + + result = _apply(sf, lib, plan) + assert result["failed"] == 1 and result["applied"] == 0 + assert (lib / "rome").exists(), "a failed precondition must leave the source alone" + operation = RenameJournal(sf).operations(plan["id"])[0] + assert operation["journal_state"] == JournalState.FAILED + assert operation["error_code"] == "source_changed" + + +def test_an_occupant_appearing_after_planning_is_never_overwritten(tmp_path): + _, sf, lib = _factory(tmp_path) + _album(sf, lib, "rome", approved_name="2019 Rome") + plan = _plan(sf, lib) + # Someone creates the destination between validation and apply. + (lib / "2019 Rome").mkdir() + (lib / "2019 Rome" / "precious.jpg").write_bytes(b"do not lose me") + + result = _apply(sf, lib, plan) + assert result["failed"] == 1 + assert (lib / "2019 Rome" / "precious.jpg").read_bytes() == b"do not lose me" + assert (lib / "rome").exists() + + +# ── staged procedures ──────────────────────────────────────────────────────── + + +def test_case_only_rename_uses_the_staged_procedure(tmp_path): + _, sf, lib = _factory(tmp_path) + _album(sf, lib, "rome", approved_name="Rome", names=("a.jpg",)) + plan = _plan(sf, lib) + assert plan["operations"][0]["case_only"] is True + + result = _apply(sf, lib, plan) + assert result["applied"] == 1 + # The directory entry now carries the new casing, and no staging dir survives. + entries = {p.name for p in lib.iterdir()} + assert "Rome" in entries + assert not any(name.startswith(".rename-") for name in entries) + assert list(_paths(sf).values())[0].endswith("Rome/a.jpg") + + +def test_cross_filesystem_move_copies_verifies_then_deletes(tmp_path): + """The cross-filesystem branch is driven directly: a real second filesystem is + not portable in CI, but the copy-verify-delete procedure is what matters.""" + _, sf, lib = _factory(tmp_path) + source = lib / "rome" + source.mkdir() + (source / "a.jpg").write_bytes(b"payload") + destination = lib / "2019 Rome" + + RenameApplyService(sf, library_roots=(lib,))._copy_verify_delete(source, destination) + + assert not source.exists() + assert (destination / "a.jpg").read_bytes() == b"payload" + assert not any(p.name.startswith(".rename-") for p in lib.iterdir()) + + +# ── restart durability ─────────────────────────────────────────────────────── + + +def test_applied_state_survives_a_restart(tmp_path): + config, sf, lib = _factory(tmp_path) + _album(sf, lib, "rome", approved_name="2019 Rome", names=("a.jpg",)) + plan = _plan(sf, lib) + _apply(sf, lib, plan) + expected = _paths(sf) + + reopened = create_session_factory(create_db_engine(config.database_url)) + assert _paths(reopened) == expected + states = {op["journal_state"] for op in RenameJournal(reopened).operations(plan["id"])} + assert states == {JournalState.COMPLETE} + + +# ── API ────────────────────────────────────────────────────────────────────── + + +def test_api_apply_requires_the_current_version(tmp_path): + config, sf, lib = _factory(tmp_path) + _album(sf, lib, "rome", approved_name="2019 Rome", names=("a.jpg",)) + + with TestClient(create_app(config)) as client: + plan = client.post("/api/v1/rename-plans").json() + + stale = client.post( + f"/api/v1/rename-plans/{plan['id']}/apply", + json={"expected_version": plan["version"] + 9}, + ) + assert stale.status_code == 409 + assert stale.json()["error"]["code"] == "version_conflict" + assert (lib / "rome").exists() + + applied = client.post( + f"/api/v1/rename-plans/{plan['id']}/apply", + json={"expected_version": plan["version"], "expected_checksum": plan["checksum"]}, + ).json() + assert applied["applied"] == 1 and applied["state"] == "applied" + assert (lib / "2019 Rome").is_dir() + + recovery = client.get("/api/v1/rename-recovery").json() + assert recovery["blocks_mutation"] is False and recovery["items"] == [] diff --git a/tests/integration/test_rename_recovery.py b/tests/integration/test_rename_recovery.py new file mode 100644 index 0000000..ff6b8bb --- /dev/null +++ b/tests/integration/test_rename_recovery.py @@ -0,0 +1,354 @@ +"""Recovering or rolling back interrupted renames (US04-04). + +The crash tests are real: a child process applies a plan and is killed by the +``PHOTO_PIPELINE_FAULT_AFTER`` barrier at each persisted journal transition in turn. +The parent then reopens the database, classifies the wreckage from journal + disk +evidence, and asserts recovery converges without losing or overwriting anything. +""" + +import os +import subprocess +import sys +import uuid +from datetime import datetime, timezone +from pathlib import Path + +import pytest +from sqlalchemy import select + +from photo_pipeline.config import Config +from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations +from photo_pipeline.models import AlbumProposal, Asset +from photo_pipeline.services import hashing +from photo_pipeline.services.rename_apply import ApplyError, RenameApplyService +from photo_pipeline.services.rename_journal import ( + MANUAL, + RESUMABLE, + ROLLBACK_SAFE, + JournalState, +) +from photo_pipeline.services.renames import RenameService + +REPO = Path(__file__).resolve().parents[2] +NOW = datetime(2026, 1, 1, tzinfo=timezone.utc) + +# A child process that applies the plan and dies at the configured barrier. +APPLY_SCRIPT = """ +import sys +sys.path.insert(0, {repo!r}) +from photo_pipeline.db import create_db_engine, create_session_factory +from photo_pipeline.services.rename_apply import RenameApplyService + +db_url, lib, plan_id, version = sys.argv[1], sys.argv[2], sys.argv[3], int(sys.argv[4]) +sf = create_session_factory(create_db_engine(db_url)) +RenameApplyService(sf, library_roots=(lib,)).apply(plan_id, expected_version=version) +""" + + +def _factory(tmp_path): + (tmp_path / "data").mkdir() + lib = tmp_path / "lib" + lib.mkdir() + config = Config.from_env( + { + "PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data"), + "PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib), + } + ) + run_migrations(config.database_url) + return config, create_session_factory(create_db_engine(config.database_url)), lib + + +def _album(sf, lib, album, *, approved_name, names=("a.jpg",)): + folder = lib / album + folder.mkdir(parents=True, exist_ok=True) + with sf() as session: + for index, name in enumerate(names): + path = folder / name + path.write_bytes(f"content-{album}-{index}".encode()) + session.add( + Asset( + id=str(uuid.uuid4()), + original_path=str(path), + current_path=str(path), + discovered_at=NOW, + hash_version=1, + byte_size=path.stat().st_size, + current_sha256=hashing.sha256_file(path), + ) + ) + session.add( + AlbumProposal( + id=str(uuid.uuid4()), + album=album, + proposed_name=approved_name, + final_name=approved_name, + status="approved", + version=2, + ) + ) + session.commit() + return folder + + +def _plan(sf, lib): + return RenameService(sf, library_roots=(lib,)).build_plan() + + +def _crash_during_apply(config, lib, plan, barrier, tmp_path): + """Run apply in a child that dies right after ``barrier`` is persisted.""" + script = tmp_path / f"apply_{barrier}.py" + script.write_text(APPLY_SCRIPT.format(repo=str(REPO))) + env = dict(os.environ) + env["PHOTO_PIPELINE_FAULT_AFTER"] = barrier + result = subprocess.run( + [ + sys.executable, + str(script), + config.database_url, + str(lib), + plan["id"], + str(plan["version"]), + ], + env=env, + capture_output=True, + ) + assert result.returncode == -9 or result.returncode == 9, ( + f"child should have been killed at {barrier}, got {result.returncode}: " + f"{result.stderr.decode(errors='replace')[-400:]}" + ) + + +def _assets(sf): + with sf() as session: + return {a.id: a.current_path for a in session.scalars(select(Asset))} + + +def _files(lib): + return {p.read_bytes() for p in lib.rglob("*") if p.is_file() and ".rename-" not in str(p)} + + +# ── fault injection at every persisted transition ──────────────────────────── + + +@pytest.mark.parametrize( + "barrier", + [ + JournalState.MOVING, + JournalState.MOVED, + JournalState.DATABASE_UPDATED, + JournalState.VERIFIED, + ], +) +def test_crash_at_each_transition_recovers_without_losing_content(barrier, tmp_path): + config, sf, lib = _factory(tmp_path) + _album(sf, lib, "rome", approved_name="2019 Rome") + plan = _plan(sf, lib) + content_before = _files(lib) + + _crash_during_apply(config, lib, plan, barrier, tmp_path) + + # Fresh process: the journal plus the disk decide what happened. + reopened = create_session_factory(create_db_engine(config.database_url)) + service = RenameApplyService(reopened, library_roots=(lib,)) + verdicts = service.journal.classify_all() + assert verdicts, f"a crash at {barrier} must leave recoverable work" + assert all(v["classification"] in {RESUMABLE, ROLLBACK_SAFE} for v in verdicts), verdicts + + service.recover() + + # No content was lost anywhere in the library. + assert _files(lib) == content_before + # And nothing is left half-done. + assert service.journal.blocks_mutation() is False + + +@pytest.mark.parametrize( + "barrier", [JournalState.MOVING, JournalState.MOVED, JournalState.DATABASE_UPDATED] +) +def test_recovery_is_idempotent_across_repeated_restarts(barrier, tmp_path): + config, sf, lib = _factory(tmp_path) + _album(sf, lib, "rome", approved_name="2019 Rome") + plan = _plan(sf, lib) + _crash_during_apply(config, lib, plan, barrier, tmp_path) + + reopened = create_session_factory(create_db_engine(config.database_url)) + service = RenameApplyService(reopened, library_roots=(lib,)) + service.recover() + settled_paths = _assets(reopened) + settled_files = _files(lib) + + # Repeated recovery passes converge — the second and third change nothing. + for _ in range(2): + service.recover() + assert _assets(reopened) == settled_paths + assert _files(lib) == settled_files + + +def test_crash_before_the_move_leaves_the_source_intact_and_resumable(tmp_path): + config, sf, lib = _factory(tmp_path) + _album(sf, lib, "rome", approved_name="2019 Rome") + plan = _plan(sf, lib) + _crash_during_apply(config, lib, plan, JournalState.MOVING, tmp_path) + + reopened = create_session_factory(create_db_engine(config.database_url)) + service = RenameApplyService(reopened, library_roots=(lib,)) + verdict = service.journal.classify_all()[0] + assert verdict["classification"] == RESUMABLE + assert (lib / "rome").exists() and not (lib / "2019 Rome").exists() + + service.recover() + # Resumable work returns to `planned`, so a fresh apply can finish the job. + operation = service.journal.operations(plan["id"])[0] + assert operation["journal_state"] == JournalState.PLANNED + + fresh = RenameService(reopened, library_roots=(lib,)).get(plan["id"]) + service.apply(plan["id"], expected_version=fresh["version"]) + assert (lib / "2019 Rome").is_dir() and not (lib / "rome").exists() + + +def test_crash_after_the_move_completes_the_bookkeeping(tmp_path): + config, sf, lib = _factory(tmp_path) + _album(sf, lib, "rome", approved_name="2019 Rome") + plan = _plan(sf, lib) + _crash_during_apply(config, lib, plan, JournalState.MOVED, tmp_path) + + reopened = create_session_factory(create_db_engine(config.database_url)) + service = RenameApplyService(reopened, library_roots=(lib,)) + assert service.journal.classify_all()[0]["classification"] == ROLLBACK_SAFE + + service.recover() + operation = service.journal.operations(plan["id"])[0] + assert operation["journal_state"] == JournalState.COMPLETE + # The database caught up with the filesystem. + assert all("2019 Rome" in path for path in _assets(reopened).values()) + + +# ── ambiguous evidence ─────────────────────────────────────────────────────── + + +def test_an_unexpected_destination_occupant_forces_manual_recovery(tmp_path): + config, sf, lib = _factory(tmp_path) + _album(sf, lib, "rome", approved_name="2019 Rome") + plan = _plan(sf, lib) + _crash_during_apply(config, lib, plan, JournalState.MOVING, tmp_path) + + # Someone puts something at the destination while the app was down. + (lib / "2019 Rome").mkdir() + (lib / "2019 Rome" / "stranger.jpg").write_bytes(b"not ours") + + reopened = create_session_factory(create_db_engine(config.database_url)) + service = RenameApplyService(reopened, library_roots=(lib,)) + verdict = service.journal.classify_all()[0] + assert verdict["classification"] == MANUAL + assert verdict["reason"] + + result = service.recover() + assert result["manual"] == 1 + # Nothing was touched, and the ambiguity keeps blocking other mutations. + assert (lib / "2019 Rome" / "stranger.jpg").read_bytes() == b"not ours" + assert (lib / "rome").exists() + assert service.journal.blocks_mutation() is True + + +def test_manual_work_blocks_a_different_plan_from_applying(tmp_path): + config, sf, lib = _factory(tmp_path) + _album(sf, lib, "rome", approved_name="2019 Rome") + plan = _plan(sf, lib) + _crash_during_apply(config, lib, plan, JournalState.MOVING, tmp_path) + (lib / "2019 Rome").mkdir() # ambiguous + + reopened = create_session_factory(create_db_engine(config.database_url)) + # The user parks the stuck album (un-approves it) and tries a different one, so + # the new plan is valid on its own — only the unresolved journal entry stands in + # the way. + with reopened() as session: + stuck = session.scalar(select(AlbumProposal).where(AlbumProposal.album == "rome")) + stuck.status = "edited" + session.commit() + _album(reopened, lib, "paris", approved_name="2020 Paris") + other = RenameService(reopened, library_roots=(lib,)).build_plan() + assert other["state"] == "validated", "the unrelated plan itself must be applyable" + + service = RenameApplyService(reopened, library_roots=(lib,)) + with pytest.raises(ApplyError, match="another rename is unresolved"): + service.apply(other["id"], expected_version=other["version"]) + # Nothing moved for the unrelated album either. + assert (lib / "paris").exists() and not (lib / "2020 Paris").exists() + + +# ── rollback ───────────────────────────────────────────────────────────────── + + +def test_rollback_returns_the_content_to_its_source(tmp_path): + config, sf, lib = _factory(tmp_path) + _album(sf, lib, "rome", approved_name="2019 Rome") + plan = _plan(sf, lib) + before = _files(lib) + _crash_during_apply(config, lib, plan, JournalState.MOVED, tmp_path) + + reopened = create_session_factory(create_db_engine(config.database_url)) + service = RenameApplyService(reopened, library_roots=(lib,)) + operation = service.journal.operations(plan["id"])[0] + + service.rollback_operation(operation["id"]) + + assert (lib / "rome").is_dir() and not (lib / "2019 Rome").exists() + assert _files(lib) == before + assert service.journal.get(operation["id"])["journal_state"] == JournalState.ROLLED_BACK + assert all("rome/" in path for path in _assets(reopened).values()) + + +def test_rollback_refuses_when_the_source_is_occupied(tmp_path): + config, sf, lib = _factory(tmp_path) + _album(sf, lib, "rome", approved_name="2019 Rome") + plan = _plan(sf, lib) + _crash_during_apply(config, lib, plan, JournalState.MOVED, tmp_path) + + # Something new appears at the old location while the app was down. + (lib / "rome").mkdir() + (lib / "rome" / "new.jpg").write_bytes(b"newcomer") + + reopened = create_session_factory(create_db_engine(config.database_url)) + service = RenameApplyService(reopened, library_roots=(lib,)) + operation = service.journal.operations(plan["id"])[0] + + with pytest.raises(ApplyError, match="occupied"): + service.rollback_operation(operation["id"]) + assert (lib / "rome" / "new.jpg").read_bytes() == b"newcomer" + + +def test_rollback_refuses_content_whose_bytes_changed(tmp_path): + config, sf, lib = _factory(tmp_path) + _album(sf, lib, "rome", approved_name="2019 Rome") + plan = _plan(sf, lib) + _crash_during_apply(config, lib, plan, JournalState.MOVED, tmp_path) + + # The moved content is edited before rollback is attempted. + (lib / "2019 Rome" / "a.jpg").write_bytes(b"edited after the move") + + reopened = create_session_factory(create_db_engine(config.database_url)) + service = RenameApplyService(reopened, library_roots=(lib,)) + operation = service.journal.operations(plan["id"])[0] + + with pytest.raises(ApplyError, match="manual recovery"): + service.rollback_operation(operation["id"]) + assert (lib / "2019 Rome" / "a.jpg").read_bytes() == b"edited after the move" + + +def test_rollback_is_idempotent(tmp_path): + config, sf, lib = _factory(tmp_path) + _album(sf, lib, "rome", approved_name="2019 Rome") + plan = _plan(sf, lib) + _crash_during_apply(config, lib, plan, JournalState.MOVED, tmp_path) + + reopened = create_session_factory(create_db_engine(config.database_url)) + service = RenameApplyService(reopened, library_roots=(lib,)) + operation = service.journal.operations(plan["id"])[0] + service.rollback_operation(operation["id"]) + settled = _files(lib) + + # A second rollback has nothing left to do and refuses rather than moving again. + with pytest.raises(ApplyError): + service.rollback_operation(operation["id"]) + assert _files(lib) == settled diff --git a/tests/story_traceability.json b/tests/story_traceability.json index 730f579..3abdf46 100644 --- a/tests/story_traceability.json +++ b/tests/story_traceability.json @@ -91,6 +91,12 @@ "US04-02": [ "tests/unit/test_rename_journal_states.py", "tests/integration/test_rename_journal.py" + ], + "US04-03": [ + "tests/integration/test_rename_apply.py" + ], + "US04-04": [ + "tests/integration/test_rename_recovery.py" ] } }