US04-02: Journal Rename State and Preconditions (#67)
This commit was merged in pull request #67.
This commit is contained in:
346
photo_pipeline/services/rename_journal.py
Normal file
346
photo_pipeline/services/rename_journal.py
Normal file
@@ -0,0 +1,346 @@
|
||||
"""Rename journal — the durable, crash-safe record of every rename transition
|
||||
(US04-02).
|
||||
|
||||
The filesystem and SQLite cannot share a transaction, so the journal is the only
|
||||
thing that makes a rename recoverable: **intent is written before the mutation it
|
||||
describes**, and the recorded state plus the real filesystem evidence is what a later
|
||||
restart reasons about. This module owns the state machine and the durable writes; it
|
||||
performs no filesystem mutation itself (that is US04-03).
|
||||
|
||||
Per-operation state machine (concept §7):
|
||||
|
||||
```
|
||||
planned → moving → moved → database_updated → verified → complete
|
||||
↘ failed ↘ rollback_required → rolled_back
|
||||
```
|
||||
|
||||
``moving`` is the dangerous window: it means "we have committed to touching the disk
|
||||
and may already have done so". Anything found in ``moving`` after a restart needs
|
||||
evidence from the filesystem, never a blind retry.
|
||||
|
||||
Terminal transitions are idempotent: re-completing an already-``complete`` operation
|
||||
is a no-op rather than an error, so a recovery pass may run repeatedly. Ownership uses
|
||||
the same fencing-token discipline as durable jobs — a worker whose token has been
|
||||
superseded cannot commit.
|
||||
|
||||
Startup classification (``classify``) inspects the journal together with the current
|
||||
source/destination on disk and labels each incomplete operation:
|
||||
|
||||
- ``resumable`` — the move has not happened; the source is intact and the
|
||||
destination is free, so applying again is safe.
|
||||
- ``rollback_safe`` — the move happened but the database was not updated; the
|
||||
destination holds the expected content and the source is gone,
|
||||
so the operation can be completed or reversed deterministically.
|
||||
- ``manual`` — the evidence is ambiguous or contradicts the journal (both or
|
||||
neither path present, or an unexpected occupant). Nothing is
|
||||
guessed; the operation blocks until a human decides.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
from photo_pipeline.models import RenameOperation, RenamePlan
|
||||
|
||||
|
||||
class JournalState:
|
||||
PLANNED = "planned"
|
||||
MOVING = "moving"
|
||||
MOVED = "moved"
|
||||
DATABASE_UPDATED = "database_updated"
|
||||
VERIFIED = "verified"
|
||||
COMPLETE = "complete"
|
||||
FAILED = "failed"
|
||||
ROLLBACK_REQUIRED = "rollback_required"
|
||||
ROLLED_BACK = "rolled_back"
|
||||
|
||||
|
||||
ALLOWED_TRANSITIONS = {
|
||||
JournalState.PLANNED: {JournalState.MOVING, JournalState.FAILED},
|
||||
# From `moving` the outcome is genuinely unknown until evidence is gathered,
|
||||
# so it may resolve forward, back to planned (proven untouched), or to failure.
|
||||
JournalState.MOVING: {
|
||||
JournalState.MOVED,
|
||||
JournalState.PLANNED,
|
||||
JournalState.FAILED,
|
||||
JournalState.ROLLBACK_REQUIRED,
|
||||
},
|
||||
JournalState.MOVED: {JournalState.DATABASE_UPDATED, JournalState.ROLLBACK_REQUIRED},
|
||||
JournalState.DATABASE_UPDATED: {JournalState.VERIFIED, JournalState.ROLLBACK_REQUIRED},
|
||||
JournalState.VERIFIED: {JournalState.COMPLETE, JournalState.ROLLBACK_REQUIRED},
|
||||
JournalState.COMPLETE: set(),
|
||||
JournalState.FAILED: {JournalState.PLANNED, JournalState.ROLLBACK_REQUIRED},
|
||||
JournalState.ROLLBACK_REQUIRED: {JournalState.ROLLED_BACK, JournalState.FAILED},
|
||||
JournalState.ROLLED_BACK: set(),
|
||||
}
|
||||
|
||||
TERMINAL_STATES = frozenset({JournalState.COMPLETE, JournalState.ROLLED_BACK})
|
||||
# States where the disk may already have been touched by this operation.
|
||||
UNSAFE_STATES = frozenset({JournalState.MOVING, JournalState.MOVED, JournalState.DATABASE_UPDATED})
|
||||
|
||||
RESUMABLE = "resumable"
|
||||
ROLLBACK_SAFE = "rollback_safe"
|
||||
MANUAL = "manual"
|
||||
|
||||
|
||||
def can_transition(current: str, target: str) -> bool:
|
||||
return target in ALLOWED_TRANSITIONS.get(current, set())
|
||||
|
||||
|
||||
class JournalError(RuntimeError):
|
||||
pass
|
||||
|
||||
|
||||
class InvalidTransition(JournalError):
|
||||
pass
|
||||
|
||||
|
||||
class JournalConflict(JournalError):
|
||||
"""Fencing check failed; a newer owner has taken over this operation."""
|
||||
|
||||
|
||||
def _now() -> datetime:
|
||||
return datetime.now(timezone.utc)
|
||||
|
||||
|
||||
class RenameJournal:
|
||||
def __init__(self, session_factory: sessionmaker) -> None:
|
||||
self._session_factory = session_factory
|
||||
|
||||
# ── intent ────────────────────────────────────────────────────────────────
|
||||
|
||||
def begin(self, operation_id: str, *, worker_id: str, fencing_token: int) -> dict:
|
||||
"""Record the intent to mutate **before** touching the filesystem.
|
||||
|
||||
Claims the operation for ``worker_id`` with ``fencing_token`` and moves it to
|
||||
``moving``. A token lower than the one already recorded loses the race and
|
||||
raises ``JournalConflict`` without changing anything.
|
||||
"""
|
||||
with self._session_factory() as session:
|
||||
row = self._require(session, operation_id)
|
||||
if row.fencing_token is not None and fencing_token < row.fencing_token:
|
||||
raise JournalConflict(
|
||||
f"stale fencing token {fencing_token} (current {row.fencing_token})"
|
||||
)
|
||||
if row.journal_state in TERMINAL_STATES:
|
||||
raise InvalidTransition(f"{row.journal_state} is terminal")
|
||||
if row.journal_state != JournalState.MOVING and not can_transition(
|
||||
row.journal_state, JournalState.MOVING
|
||||
):
|
||||
raise InvalidTransition(f"{row.journal_state} -> {JournalState.MOVING}")
|
||||
if row.journal_state != JournalState.MOVING:
|
||||
row.attempt_count += 1
|
||||
row.journal_state = JournalState.MOVING
|
||||
row.worker_id = worker_id
|
||||
row.fencing_token = fencing_token
|
||||
row.error_code = row.error_message = None
|
||||
row.updated_at = _now()
|
||||
session.commit()
|
||||
return _operation_dict(row)
|
||||
|
||||
# ── transitions ───────────────────────────────────────────────────────────
|
||||
|
||||
def transition(
|
||||
self,
|
||||
operation_id: str,
|
||||
target: str,
|
||||
*,
|
||||
fencing_token: int | None = None,
|
||||
error: tuple[str, str] | None = None,
|
||||
) -> dict:
|
||||
"""Move one operation to ``target``, enforcing the state machine.
|
||||
|
||||
Re-entering the state an operation already holds is a no-op, which is what
|
||||
makes recovery idempotent across repeated restarts.
|
||||
"""
|
||||
with self._session_factory() as session:
|
||||
row = self._require(session, operation_id)
|
||||
if fencing_token is not None and row.fencing_token is not None:
|
||||
if fencing_token < row.fencing_token:
|
||||
raise JournalConflict(
|
||||
f"stale fencing token {fencing_token} (current {row.fencing_token})"
|
||||
)
|
||||
if row.journal_state == target:
|
||||
return _operation_dict(row) # idempotent
|
||||
if not can_transition(row.journal_state, target):
|
||||
raise InvalidTransition(f"{row.journal_state} -> {target}")
|
||||
|
||||
row.journal_state = target
|
||||
row.updated_at = _now()
|
||||
if target == JournalState.VERIFIED:
|
||||
row.verified_at = _now()
|
||||
if error:
|
||||
row.error_code, row.error_message = error[0], error[1][:500]
|
||||
elif target not in (JournalState.FAILED, JournalState.ROLLBACK_REQUIRED):
|
||||
row.error_code = row.error_message = None
|
||||
session.commit()
|
||||
return _operation_dict(row)
|
||||
|
||||
# ── reads ─────────────────────────────────────────────────────────────────
|
||||
|
||||
def get(self, operation_id: str) -> dict | None:
|
||||
with self._session_factory() as session:
|
||||
row = session.get(RenameOperation, operation_id)
|
||||
return _operation_dict(row) if row else None
|
||||
|
||||
def operations(self, plan_id: str) -> list[dict]:
|
||||
with self._session_factory() as session:
|
||||
rows = session.scalars(
|
||||
select(RenameOperation)
|
||||
.where(RenameOperation.plan_id == plan_id)
|
||||
.order_by(RenameOperation.sequence)
|
||||
)
|
||||
return [_operation_dict(row) for row in rows]
|
||||
|
||||
def incomplete(self) -> list[dict]:
|
||||
"""Every operation left in a non-terminal, non-planned state — the work a
|
||||
restart has to reason about."""
|
||||
with self._session_factory() as session:
|
||||
rows = session.scalars(
|
||||
select(RenameOperation)
|
||||
.where(
|
||||
RenameOperation.journal_state.not_in([*TERMINAL_STATES, JournalState.PLANNED])
|
||||
)
|
||||
.order_by(RenameOperation.plan_id, RenameOperation.sequence)
|
||||
)
|
||||
return [_operation_dict(row) for row in rows]
|
||||
|
||||
# ── startup classification ────────────────────────────────────────────────
|
||||
|
||||
def classify(self, operation_id: str) -> dict:
|
||||
"""Classify one incomplete operation from the journal plus disk evidence.
|
||||
|
||||
Returns ``{operation_id, journal_state, classification, reason, source_exists,
|
||||
destination_exists}``. Never mutates anything — deciding is US04-04's job.
|
||||
"""
|
||||
row = self.get(operation_id)
|
||||
if row is None:
|
||||
raise JournalError(f"unknown rename operation {operation_id!r}")
|
||||
source = Path(row["source_path"])
|
||||
destination = Path(row["destination_path"])
|
||||
source_exists = source.exists()
|
||||
destination_exists = destination.exists()
|
||||
|
||||
classification, reason = _classify(row["journal_state"], source_exists, destination_exists)
|
||||
return {
|
||||
"operation_id": operation_id,
|
||||
"journal_state": row["journal_state"],
|
||||
"classification": classification,
|
||||
"reason": reason,
|
||||
"source_exists": source_exists,
|
||||
"destination_exists": destination_exists,
|
||||
}
|
||||
|
||||
def classify_all(self) -> list[dict]:
|
||||
return [self.classify(row["id"]) for row in self.incomplete()]
|
||||
|
||||
def blocks_mutation(self) -> bool:
|
||||
"""True when any operation is in a state where the library may be
|
||||
half-renamed. Unrelated mutations must not start until it is resolved."""
|
||||
return any(row["journal_state"] in UNSAFE_STATES for row in self.incomplete())
|
||||
|
||||
# ── plan-level ────────────────────────────────────────────────────────────
|
||||
|
||||
def plan_state(self, plan_id: str) -> str:
|
||||
"""Derive the plan's state from its operations, so the summary can never
|
||||
disagree with the journal."""
|
||||
states = {row["journal_state"] for row in self.operations(plan_id)}
|
||||
if not states:
|
||||
return "planned"
|
||||
if states <= {JournalState.COMPLETE}:
|
||||
return "applied"
|
||||
if states <= {JournalState.ROLLED_BACK, JournalState.COMPLETE}:
|
||||
return "rolled_back"
|
||||
if states & {JournalState.FAILED, JournalState.ROLLBACK_REQUIRED}:
|
||||
return "failed"
|
||||
if states & UNSAFE_STATES:
|
||||
return "applying"
|
||||
return "planned"
|
||||
|
||||
def sync_plan_state(self, plan_id: str) -> str:
|
||||
state = self.plan_state(plan_id)
|
||||
with self._session_factory() as session:
|
||||
plan = session.get(RenamePlan, plan_id)
|
||||
if plan is None:
|
||||
raise JournalError(f"unknown rename plan {plan_id!r}")
|
||||
if plan.state != state:
|
||||
plan.state = state
|
||||
plan.version += 1
|
||||
plan.updated_at = _now()
|
||||
if state == "applied" and plan.applied_at is None:
|
||||
plan.applied_at = _now()
|
||||
session.commit()
|
||||
return state
|
||||
|
||||
@staticmethod
|
||||
def _require(session, operation_id: str) -> RenameOperation:
|
||||
row = session.get(RenameOperation, operation_id)
|
||||
if row is None:
|
||||
raise JournalError(f"unknown rename operation {operation_id!r}")
|
||||
return row
|
||||
|
||||
|
||||
def _classify(state: str, source_exists: bool, destination_exists: bool) -> tuple[str, str]:
|
||||
"""The evidence table. Kept a pure function so every combination is testable."""
|
||||
if state == JournalState.MOVING:
|
||||
# Intent was recorded; the move may or may not have happened.
|
||||
if source_exists and not destination_exists:
|
||||
return RESUMABLE, "source intact and destination free; the move did not happen"
|
||||
if destination_exists and not source_exists:
|
||||
return ROLLBACK_SAFE, "the move completed before the journal was updated"
|
||||
if source_exists and destination_exists:
|
||||
return MANUAL, "both source and destination exist; cannot tell them apart safely"
|
||||
return MANUAL, "neither source nor destination exists"
|
||||
|
||||
if state in (JournalState.MOVED, JournalState.DATABASE_UPDATED, JournalState.VERIFIED):
|
||||
if destination_exists and not source_exists:
|
||||
return (
|
||||
ROLLBACK_SAFE,
|
||||
"destination holds the content; finish or reverse deterministically",
|
||||
)
|
||||
if source_exists and not destination_exists:
|
||||
return MANUAL, f"journal says {state} but the content is still at the source"
|
||||
if source_exists and destination_exists:
|
||||
return MANUAL, "both paths exist after a recorded move"
|
||||
return MANUAL, "the recorded destination is missing"
|
||||
|
||||
if state == JournalState.ROLLBACK_REQUIRED:
|
||||
if source_exists and not destination_exists:
|
||||
return RESUMABLE, "already back at the source; the rollback is effectively done"
|
||||
if destination_exists and not source_exists:
|
||||
return ROLLBACK_SAFE, "content is at the destination and can be moved back"
|
||||
return MANUAL, "ambiguous paths for a rollback"
|
||||
|
||||
if state == JournalState.FAILED:
|
||||
if source_exists and not destination_exists:
|
||||
return RESUMABLE, "the failure left the source untouched"
|
||||
return MANUAL, "a failed operation with unexpected paths on disk"
|
||||
|
||||
return MANUAL, f"unhandled journal state {state}"
|
||||
|
||||
|
||||
def _operation_dict(row: RenameOperation) -> dict:
|
||||
return {
|
||||
"id": row.id,
|
||||
"plan_id": row.plan_id,
|
||||
"sequence": row.sequence,
|
||||
"album": row.album,
|
||||
"source_path": row.source_path,
|
||||
"destination_path": row.destination_path,
|
||||
"asset_ids": json.loads(row.asset_ids or "[]"),
|
||||
"expected_sha256": json.loads(row.expected_sha256 or "{}"),
|
||||
"case_only": row.case_only,
|
||||
"same_filesystem": row.same_filesystem,
|
||||
"journal_state": row.journal_state,
|
||||
"attempt_count": row.attempt_count,
|
||||
"fencing_token": row.fencing_token,
|
||||
"worker_id": row.worker_id,
|
||||
"verified_at": row.verified_at.isoformat() if row.verified_at else None,
|
||||
"error_code": row.error_code,
|
||||
"error_message": row.error_message,
|
||||
}
|
||||
Reference in New Issue
Block a user