domverse/photoanalyzer
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects 1 Releases Wiki Activity

US04-03: Apply and Verify Guarded Renames (#68)

Browse Source
This commit was merged in pull request #68.
This commit is contained in:
domverse
2026-08-16 11:27:08 +02:00
parent daaa4ca67b
commit 7103cb84bf
7 changed files with 1222 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

52
photo_pipeline/api/routes/renames.py
View File

@@ -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))

2
photo_pipeline/schemas/__init__.py
View File

@@ -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",

11
photo_pipeline/schemas/renames.py Normal file
View File

@@ -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

493
photo_pipeline/services/rename_apply.py Normal file
View File

@@ -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

304
tests/integration/test_rename_apply.py Normal file
View File

@@ -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"] == []

354
tests/integration/test_rename_recovery.py Normal file
View File

@@ -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

6
tests/story_traceability.json
View File

@@ -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"
]
}
}