US04-03: Apply and Verify Guarded Renames (#68)
This commit was merged in pull request #68.
This commit is contained in:
@@ -9,6 +9,12 @@ from __future__ import annotations
|
|||||||
from fastapi import APIRouter, Request
|
from fastapi import APIRouter, Request
|
||||||
from fastapi.responses import JSONResponse
|
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
|
from photo_pipeline.services.renames import RenameError, RenameService
|
||||||
|
|
||||||
router = APIRouter(tags=["renames"])
|
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:
|
def _error(status: int, code: str, message: str) -> JSONResponse:
|
||||||
return JSONResponse(status_code=status, content={"error": {"code": code, "message": message}})
|
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)
|
return _service(request).export(plan_id)
|
||||||
except RenameError as error:
|
except RenameError as error:
|
||||||
return _error(404, "not_found", str(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))
|
||||||
|
|||||||
@@ -6,10 +6,12 @@ from photo_pipeline.schemas.albums import (
|
|||||||
GenerateProposalsRequest,
|
GenerateProposalsRequest,
|
||||||
)
|
)
|
||||||
from photo_pipeline.schemas.duplicates import DecisionRequest
|
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.jobs import JobStartRequest
|
||||||
from photo_pipeline.schemas.safety import SafetyDecisionRequest
|
from photo_pipeline.schemas.safety import SafetyDecisionRequest
|
||||||
|
|
||||||
__all__ = [
|
__all__ = [
|
||||||
|
"ApplyPlanRequest",
|
||||||
"ApproveProposalRequest",
|
"ApproveProposalRequest",
|
||||||
"DecisionRequest",
|
"DecisionRequest",
|
||||||
"EditProposalRequest",
|
"EditProposalRequest",
|
||||||
|
|||||||
11
photo_pipeline/schemas/renames.py
Normal file
11
photo_pipeline/schemas/renames.py
Normal 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
493
photo_pipeline/services/rename_apply.py
Normal 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
304
tests/integration/test_rename_apply.py
Normal 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
354
tests/integration/test_rename_recovery.py
Normal 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
|
||||||
@@ -91,6 +91,12 @@
|
|||||||
"US04-02": [
|
"US04-02": [
|
||||||
"tests/unit/test_rename_journal_states.py",
|
"tests/unit/test_rename_journal_states.py",
|
||||||
"tests/integration/test_rename_journal.py"
|
"tests/integration/test_rename_journal.py"
|
||||||
|
],
|
||||||
|
"US04-03": [
|
||||||
|
"tests/integration/test_rename_apply.py"
|
||||||
|
],
|
||||||
|
"US04-04": [
|
||||||
|
"tests/integration/test_rename_recovery.py"
|
||||||
]
|
]
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user