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

US06-04: Plan and Execute Safe Restores

Browse Source
This commit is contained in:
domverse
2026-08-16 19:48:35 +02:00
parent 439eb9971e
commit 0982ebda70
10 changed files with 1294 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

89
photo_pipeline/api/routes/archives.py
View File

@@ -1,4 +1,4 @@
"""Archive location, preflight, and plan API (US06-01, US06-02).
"""Archive location, preflight, plan, and restore API (US06-01, US06-02, US06-04).
Registering a location writes a marker onto the medium; preflight is a command
rather than a read, because it probes the destination, hashes the scope, and issues
@@ -13,10 +13,11 @@ from fastapi import APIRouter, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from photo_pipeline.jobs.domain_handlers import ARCHIVE_LOCK, ARCHIVE_PLAN
from photo_pipeline.jobs.domain_handlers import ARCHIVE_LOCK, ARCHIVE_PLAN, RESTORE_PLAN
from photo_pipeline.services.archives import ArchiveError, ArchiveService
from photo_pipeline.services.archive_transfer import ArchiveTransferService
from photo_pipeline.services.jobs import JobBlocked, JobService
from photo_pipeline.services.restores import RestoreService
router = APIRouter(tags=["archives"])
@@ -42,10 +43,24 @@ class CreatePlanRequest(PreflightRequest):
token: str
class RestoreRequest(BaseModel):
location_id: str
# ``None`` means every asset archived at this location.
asset_ids: list[str] | None = None
class CreateRestoreRequest(RestoreRequest):
token: str
def _service(request: Request) -> ArchiveService:
return ArchiveService(request.app.state.session_factory, config=request.app.state.config)
def _restores(request: Request) -> RestoreService:
return RestoreService(request.app.state.session_factory, config=request.app.state.config)
def _transfers(request: Request) -> ArchiveTransferService:
return ArchiveTransferService(
request.app.state.session_factory, config=request.app.state.config
@@ -138,6 +153,76 @@ def apply_plan(plan_id: str, request: Request):
return {"plan_id": plan_id, "job": job}
@router.post("/restore-preflight")
def restore_preflight(body: RestoreRequest, request: Request):
"""Validate restoring archived assets back into the library. Nothing moves."""
try:
return _restores(request).preflight(body.location_id, body.asset_ids)
except ArchiveError as error:
return _error(error)
@router.post("/restore-plans", status_code=201)
def create_restore_plan(body: CreateRestoreRequest, request: Request):
try:
return _restores(request).create(body.location_id, body.asset_ids, token=body.token)
except ArchiveError as error:
return _error(error)
@router.get("/restore-plans")
def list_restore_plans(request: Request) -> dict:
return {"plans": _restores(request).list()}
@router.get("/restore-plans/{plan_id}")
def get_restore_plan(plan_id: str, request: Request):
plan = _restores(request).get(plan_id)
if plan is None:
return _error(ArchiveError("unknown_plan", f"unknown restore plan {plan_id}"))
return plan
@router.post("/restore-plans/{plan_id}/apply")
def apply_restore_plan(plan_id: str, request: Request):
"""Queue the restore on the archiver lane — the same single lane as archiving,
because both move the same originals."""
service = _restores(request)
plan = service.get(plan_id)
if plan is None:
return _error(ArchiveError("unknown_plan", f"unknown restore plan {plan_id}"))
unresolved = [row for row in service.journal.incomplete() if row["plan_id"] != plan_id]
if unresolved:
return _error(
ArchiveError(
"archive_pending",
f"an unresolved archive operation ({unresolved[0]['id']}) must be recovered",
)
)
try:
job = JobService(request.app.state.session_factory).enqueue(
RESTORE_PLAN,
lock=ARCHIVE_LOCK,
idempotency_key=f"restore:{plan_id}:{plan['version']}",
items=[plan_id],
)
except JobBlocked as error:
return JSONResponse(
status_code=409, content={"error": {"code": error.code, "message": str(error)}}
)
return {"plan_id": plan_id, "job": job}
@router.get("/restore-recovery")
def restore_recovery_status(request: Request) -> dict:
return _restores(request).recovery_status()
@router.post("/restore-recovery/resolve")
def resolve_restore_recovery(request: Request) -> dict:
return _restores(request).recover()
@router.get("/archive-recovery")
def recovery_status(request: Request) -> dict:
"""What an interrupted transfer left behind, straight from journal + disk."""

22
photo_pipeline/jobs/domain_handlers.py
View File

@@ -1,9 +1,9 @@
"""Domain job handlers: safety scoring, content analysis, uploads, archive
transfers (US02-06, US05-02, US06-02).
transfers, restores (US02-06, US05-02, US06-02, US06-04).
Importing this module registers the ``safety_score``, ``analysis``,
``upload_batch``, and ``archive_plan`` job types so the generic worker can run them
per item. Each handler delegates to its service, which owns the real work and the
``upload_batch``, ``archive_plan``, and ``restore_plan`` job types so the generic
worker can run them per item. Each handler delegates to its service, which owns the real work and the
privacy gate. Handlers are idempotent: re-scoring or re-analyzing one asset is safe
after an interrupted attempt, an upload batch refuses to re-run an attempt whose
outcome is unknown, and an archive plan skips items it already completed.
@@ -20,6 +20,7 @@ SAFETY_SCORE = "safety_score"
ANALYSIS = "analysis"
UPLOAD_BATCH = "upload_batch"
ARCHIVE_PLAN = "archive_plan"
RESTORE_PLAN = "restore_plan"
# Both mutate the library's metadata/derived state; one at a time (concept §one job).
LIBRARY_WRITE_LOCK = "library_write"
# The uploader lane: one album batch at a time (concept §16).
@@ -70,7 +71,22 @@ def _archive_plan_item(plan_id: str, ctx: JobContext) -> None:
raise RuntimeError(f"archive plan {plan_id}: {result['failed']} item(s) failed")
def _restore_plan_item(plan_id: str, ctx: JobContext) -> None:
"""One item = one restore plan. A restore removes nothing, so an item failure
simply leaves that asset archived (US06-04)."""
from photo_pipeline.config import Config
from photo_pipeline.services.restores import RestoreService
config = ctx.config if ctx.config is not None else Config.from_env()
result = RestoreService(ctx.session_factory, config=config).apply(
plan_id, worker_id=ctx.worker_id
)
if result["failed"]:
raise RuntimeError(f"restore plan {plan_id}: {result['failed']} item(s) failed")
register(SAFETY_SCORE, _safety_score_item)
register(ANALYSIS, _analysis_item)
register(UPLOAD_BATCH, _upload_batch_item)
register(ARCHIVE_PLAN, _archive_plan_item)
register(RESTORE_PLAN, _restore_plan_item)

5
photo_pipeline/models/archives.py
View File

@@ -66,6 +66,8 @@ class ArchivePlan(Base):
# The preflight token this plan was approved against; re-verified before apply.
token: Mapped[str] = mapped_column(String, nullable=False)
albums: Mapped[str | None] = mapped_column(String) # JSON array
# archive | restore — the same journal read in the opposite direction (US06-04).
direction: Mapped[str] = mapped_column(String, nullable=False, default="archive")
# planned | applying | complete | failed
state: Mapped[str] = mapped_column(String, nullable=False, default="planned")
@@ -100,6 +102,9 @@ class ArchiveOperation(Base):
album: Mapped[str] = mapped_column(String, nullable=False)
asset_id: Mapped[str] = mapped_column(ForeignKey("assets.id"), nullable=False, index=True)
# archive: library → medium. restore: medium → library (US06-04). ``source_path``
# and ``destination_path`` always mean "from" and "to" for this direction.
direction: Mapped[str] = mapped_column(String, nullable=False, default="archive")
source_path: Mapped[str] = mapped_column(String, nullable=False)
destination_path: Mapped[str] = mapped_column(String, nullable=False)
# Relative to the location root, because the medium can be mounted elsewhere.

4
photo_pipeline/models/assets.py
View File

@@ -43,6 +43,10 @@ class Asset(Base):
# link is written and read by the archive service (US06-02).
archive_location_id: Mapped[str | None] = mapped_column(String)
archive_path: Mapped[str | None] = mapped_column(String)
# Set when the archived copy was proven to hold bytes other than the recorded
# ones (US06-04). Restore refuses such an asset instead of accepting a different
# file; cleared as soon as a verification matches again.
archive_divergent_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
# Duplicate canonical link: NULL when the asset is itself canonical or undecided.
canonical_asset_id: Mapped[str | None] = mapped_column(ForeignKey("assets.id"))

48
photo_pipeline/services/archive_journal.py
View File

@@ -15,6 +15,11 @@ planned → transferring → verified → removing → complete
↘ ↘ ↘ failed
```
A restore (US06-04) uses the same rows with ``direction='restore'``: it copies from
the medium back into the library and removes nothing, so it goes ``verified →
complete`` directly. ``source_path``/``destination_path`` always mean "from"/"to",
which is why the evidence table below needs no direction of its own.
- ``transferring`` — intent recorded; a temporary copy may exist, the destination
may or may not have been published. Nothing has been removed.
- ``verified`` — the archived bytes exist at their final path, hash exactly as
@@ -73,10 +78,22 @@ ALLOWED_TRANSITIONS = {
ArchiveState.FAILED: {ArchiveState.PLANNED, ArchiveState.TRANSFERRING},
}
# A restore removes nothing, so it has no ``removing`` step: a verified published
# copy is the whole job (US06-04). Keeping this as a separate table means the
# archive direction still cannot reach ``complete`` without going through removal.
RESTORE_TRANSITIONS = {
**ALLOWED_TRANSITIONS,
ArchiveState.VERIFIED: {ArchiveState.COMPLETE, ArchiveState.FAILED},
}
TERMINAL_STATES = frozenset({ArchiveState.COMPLETE})
# States where this item may already have touched the filesystem.
UNSAFE_STATES = frozenset({ArchiveState.TRANSFERRING, ArchiveState.VERIFIED, ArchiveState.REMOVING})
# Which way the bytes move. Same rows, same evidence table, opposite direction.
ARCHIVE = "archive"
RESTORE = "restore"
RESUMABLE = "resumable"
FORWARD = "forward"
MANUAL = "manual"
@@ -94,8 +111,9 @@ class JournalConflict(JournalError):
"""Fencing check failed; a newer owner has taken over this operation."""
def can_transition(current: str, target: str) -> bool:
return target in ALLOWED_TRANSITIONS.get(current, set())
def can_transition(current: str, target: str, direction: str = ARCHIVE) -> bool:
table = RESTORE_TRANSITIONS if direction == RESTORE else ALLOWED_TRANSITIONS
return target in table.get(current, set())
def _now() -> datetime:
@@ -119,7 +137,7 @@ class ArchiveJournal:
if row.journal_state in TERMINAL_STATES:
raise InvalidTransition(f"{row.journal_state} is terminal")
if row.journal_state != ArchiveState.TRANSFERRING and not can_transition(
row.journal_state, ArchiveState.TRANSFERRING
row.journal_state, ArchiveState.TRANSFERRING, row.direction
):
raise InvalidTransition(f"{row.journal_state} -> {ArchiveState.TRANSFERRING}")
if row.journal_state != ArchiveState.TRANSFERRING:
@@ -160,7 +178,7 @@ class ArchiveJournal:
if row.journal_state == target:
session.commit()
return _operation_dict(row) # idempotent
if not can_transition(row.journal_state, target):
if not can_transition(row.journal_state, target, row.direction):
raise InvalidTransition(f"{row.journal_state} -> {target}")
row.journal_state = target
@@ -192,16 +210,18 @@ class ArchiveJournal:
)
return [_operation_dict(row) for row in rows]
def incomplete(self) -> list[dict]:
def incomplete(self, *, direction: str | None = None) -> list[dict]:
"""Every operation left in a non-terminal, non-planned state — the work a
restart has to reason about."""
restart has to reason about. Without ``direction`` this spans archives and
restores, because either one half-done blocks the other."""
with self._session_factory() as session:
stmt = select(ArchiveOperation).where(
ArchiveOperation.journal_state.not_in([*TERMINAL_STATES, ArchiveState.PLANNED])
)
if direction is not None:
stmt = stmt.where(ArchiveOperation.direction == direction)
rows = session.scalars(
select(ArchiveOperation)
.where(
ArchiveOperation.journal_state.not_in([*TERMINAL_STATES, ArchiveState.PLANNED])
)
.order_by(ArchiveOperation.plan_id, ArchiveOperation.sequence)
stmt.order_by(ArchiveOperation.plan_id, ArchiveOperation.sequence)
)
return [_operation_dict(row) for row in rows]
@@ -231,6 +251,7 @@ class ArchiveJournal:
return {
"operation_id": operation_id,
"plan_id": row["plan_id"],
"direction": row["direction"],
"album": row["album"],
"asset_id": row["asset_id"],
"source_path": row["source_path"],
@@ -243,8 +264,8 @@ class ArchiveJournal:
"destination_matches": destination_matches,
}
def classify_all(self) -> list[dict]:
return [self.classify(row["id"]) for row in self.incomplete()]
def classify_all(self, *, direction: str | None = None) -> list[dict]:
return [self.classify(row["id"]) for row in self.incomplete(direction=direction)]
def blocks_mutation(self) -> bool:
"""True when any item may have the library half-archived."""
@@ -317,6 +338,7 @@ def _operation_dict(row: ArchiveOperation) -> dict:
return {
"id": row.id,
"plan_id": row.plan_id,
"direction": row.direction,
"sequence": row.sequence,
"album": row.album,
"asset_id": row.asset_id,

64
photo_pipeline/services/archive_transfer.py
View File

@@ -54,6 +54,7 @@ from sqlalchemy.orm import sessionmaker
from photo_pipeline.config import Config
from photo_pipeline.models import ArchiveLocation, ArchiveOperation, ArchivePlan, Asset, AssetPath
from photo_pipeline.services.archive_journal import (
ARCHIVE,
MANUAL,
RESUMABLE,
ArchiveJournal,
@@ -115,6 +116,7 @@ class ArchiveTransferService:
location_id=location_id,
token=token,
albums=json.dumps(albums) if albums is not None else None,
direction=ARCHIVE,
state="planned",
schema_version=MANIFEST_VERSION,
asset_count=preflight["totals"]["assets"],
@@ -132,6 +134,7 @@ class ArchiveTransferService:
ArchiveOperation(
id=str(uuid.uuid4()),
plan_id=plan_id,
direction=ARCHIVE,
sequence=sequence,
album=album["album"],
asset_id=asset["asset_id"],
@@ -161,7 +164,11 @@ class ArchiveTransferService:
def list(self) -> list[dict]:
with self._session_factory() as session:
rows = session.scalars(select(ArchivePlan).order_by(ArchivePlan.created_at))
rows = session.scalars(
select(ArchivePlan)
.where(ArchivePlan.direction == ARCHIVE)
.order_by(ArchivePlan.created_at)
)
return [_plan_dict(row) for row in rows]
# ── apply ─────────────────────────────────────────────────────────────────
@@ -262,7 +269,7 @@ class ArchiveTransferService:
if same_filesystem:
os.rename(source, destination)
else:
self._copy_and_publish(operation, source, destination)
copy_verify_publish(source, destination, operation["expected_sha256"])
_fsync_dir(destination.parent)
# 4. The published file is the archive only once it hashes as recorded.
@@ -282,27 +289,6 @@ class ArchiveTransferService:
# 5. Only now may the active source go.
self._finish(self.journal.get(operation["id"]), location, token=token, worker_id=worker_id)
def _copy_and_publish(self, operation: dict, source: Path, destination: Path) -> None:
"""Cross-filesystem: copy to a temporary file beside the destination, prove
its bytes, then publish it atomically. The source is still untouched."""
temp = destination.with_name(f"{TEMP_PREFIX}{uuid.uuid4().hex}{TEMP_SUFFIX}")
try:
with open(source, "rb") as src, open(temp, "wb") as out:
shutil.copyfileobj(src, out, 1024 * 1024)
out.flush()
os.fsync(out.fileno())
if sha256_file(temp) != operation["expected_sha256"]:
raise PreconditionFailed("copy_mismatch", f"{source} copied with wrong bytes")
if destination.exists():
raise PreconditionFailed(
"destination_exists", f"{destination} appeared during the transfer"
)
# ponytail: rename after an exists() check. The archiver lane is single
# and local; use O_EXCL/link-based publish if a second writer ever exists.
os.rename(temp, destination)
finally:
temp.unlink(missing_ok=True)
def _finish(self, operation: dict, location: dict, *, token: int, worker_id: str) -> None:
"""Drive an item whose archive copy is durable through removal and
bookkeeping. Every step is idempotent, so recovery may replay it."""
@@ -457,7 +443,7 @@ class ArchiveTransferService:
"""
results = {"resumed": 0, "completed": 0, "manual": 0}
touched: set[str] = set()
for verdict in self.journal.classify_all():
for verdict in self.journal.classify_all(direction=ARCHIVE):
operation = self.journal.get(verdict["operation_id"])
touched.add(operation["plan_id"])
token = (operation["fencing_token"] or 0) + 1
@@ -484,7 +470,7 @@ class ArchiveTransferService:
return results
def recovery_status(self) -> dict:
verdicts = self.journal.classify_all()
verdicts = self.journal.classify_all(direction=ARCHIVE)
return {
"operations": verdicts,
"manual": [v for v in verdicts if v["classification"] == MANUAL],
@@ -528,6 +514,33 @@ class ArchiveTransferService:
# ── module helpers ───────────────────────────────────────────────────────────
def copy_verify_publish(source: Path, destination: Path, expected_sha256: str) -> None:
"""Copy to a temporary file beside the destination, prove its bytes, then publish
it atomically. The source is never touched, so a failure costs nothing.
Shared by archiving (library → medium) and restoring (medium → library, US06-04):
both need the same promise that a published file is either complete and correct
or not there at all.
"""
temp = destination.with_name(f"{TEMP_PREFIX}{uuid.uuid4().hex}{TEMP_SUFFIX}")
try:
with open(source, "rb") as src, open(temp, "wb") as out:
shutil.copyfileobj(src, out, 1024 * 1024)
out.flush()
os.fsync(out.fileno())
if sha256_file(temp) != expected_sha256:
raise PreconditionFailed("copy_mismatch", f"{source} copied with wrong bytes")
if destination.exists():
raise PreconditionFailed(
"destination_exists", f"{destination} appeared during the transfer"
)
# ponytail: rename after an exists() check. The archiver lane is single and
# local; use O_EXCL/link-based publish if a second writer ever exists.
os.rename(temp, destination)
finally:
temp.unlink(missing_ok=True)
def _same_filesystem(source: Path, destination_dir: Path) -> bool:
"""Proven at run time from the actual devices, never from the plan's preview."""
try:
@@ -618,6 +631,7 @@ def _plan_dict(plan: ArchivePlan) -> dict:
"id": plan.id,
"location_id": plan.location_id,
"token": plan.token,
"direction": plan.direction,
"albums": json.loads(plan.albums) if plan.albums else None,
"state": plan.state,
"schema_version": plan.schema_version,

648
photo_pipeline/services/restores.py Normal file
View File

@@ -0,0 +1,648 @@
"""RestoreService — plan and execute safe restores (US06-04).
Restore is archiving read backwards, with one decisive difference: it removes
nothing. The archived copy stays on its medium, so every failure mode here costs
at most a discarded temporary file. What restore must never do is *lose identity*
— the asset that comes back is the same asset, with its duplicate decision, safety
review, analysis, and upload history intact — or *overwrite* something in the
active library.
Preflight proves, per concept §9 "Restore":
- the recorded medium is mounted and is the right one (marker ``media_id``);
- every selected asset is archived, its archive copy exists, and it hashes to
exactly the bytes the database recorded — a mismatch is ``divergent`` and is
refused, never silently accepted as "the file";
- the destination lies inside the library, outside ``_IGNORE/``, and is free; a
taken path is answered with a collision-free name, never an overwrite;
- the library filesystem has room for the scope plus the configured reserve;
- no rename, archive, or restore lease is holding the lane.
Blocker codes: ``no_library_root``, ``location_offline``, ``wrong_volume``,
``unsafe_destination``, ``library_not_writable``, ``insufficient_capacity``,
``lock_conflict``, ``rename_pending``, ``archive_pending``, ``empty_scope``,
``not_archived``, ``archive_missing``, ``bytes_changed``.
Per item the sequence is:
```
journal.begin (transferring) ← intent persisted BEFORE any disk change
recheck: medium, hash, free destination, asset still archived
copy to a temporary file beside the destination, fsync, hash it back
atomically publish into the library
journal → verified
current_path = destination, availability = active, path occurrence opened
journal → complete
```
Like archiving, the confirmation token is derived from the report, so a changed
scope, a swapped medium, or a destination that filled up invalidates it.
"""
from __future__ import annotations
import hashlib
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.config import Config
from photo_pipeline.jobs.domain_handlers import ARCHIVE_LOCK, LIBRARY_WRITE_LOCK, UPLOAD_LOCK
from photo_pipeline.models import ArchiveLocation, ArchiveOperation, ArchivePlan, Asset, AssetPath
from photo_pipeline.path_policy import PathPolicyError, is_excluded, normalize_root, resolve_within
from photo_pipeline.services import availability
from photo_pipeline.services.archive_journal import (
MANUAL,
RESTORE,
RESUMABLE,
ArchiveJournal,
ArchiveState,
)
from photo_pipeline.services.archive_transfer import (
_clean_temp_files,
_fsync_dir,
_plan_dict,
copy_verify_publish,
)
from photo_pipeline.services.archives import ArchiveError
from photo_pipeline.services.hashing import sha256_file
from photo_pipeline.services.jobs import JobService
from photo_pipeline.services.rename_apply import PreconditionFailed, maybe_fault
from photo_pipeline.services.rename_journal import RenameJournal
PREFLIGHT_VERSION = 1
TOKEN_PREFIX = f"r{PREFLIGHT_VERSION}"
# What a restored file is called when its original name is taken. The suffix is
# visible on purpose: a restore that quietly reuses a name is indistinguishable
# from an overwrite.
RESTORED_SUFFIX = "restored"
LOCKS = (LIBRARY_WRITE_LOCK, UPLOAD_LOCK, ARCHIVE_LOCK)
APPLYABLE_PLAN_STATES = frozenset({"planned", "applying", "failed", "complete"})
def _now() -> datetime:
return datetime.now(timezone.utc)
def _issue(code: str, message: str) -> dict:
return {"code": code, "message": message}
class RestoreService:
def __init__(self, session_factory: sessionmaker, *, config: Config) -> None:
self._session_factory = session_factory
self._config = config
self._roots = tuple(normalize_root(root) for root in config.library_roots)
self.journal = ArchiveJournal(session_factory)
# ── preflight ─────────────────────────────────────────────────────────────
def preflight(self, location_id: str, asset_ids: list[str] | None = None) -> dict:
"""Validate a restore scope and issue its token. Nothing is written."""
with self._session_factory() as session:
location = session.get(ArchiveLocation, location_id)
if location is None:
raise ArchiveError("unknown_location", f"unknown archive location {location_id!r}")
root = Path(location.root)
online = availability.location_online(location)
marker = availability.read_marker(root)
report = {
"schema_version": PREFLIGHT_VERSION,
"location": {
"id": location.id,
"name": location.name,
"root": str(root),
"media_id": location.media_id,
"state": _location_state(root, marker, location.media_id),
},
"blockers": [],
}
items = self._items(session, location, asset_ids, reachable=online)
report["blockers"] += self._destination_blockers(report["location"]["state"], root)
report["blockers"] += self._lock_blockers()
report["items"] = items
report["totals"] = {
"assets": len(items),
"blocked": sum(1 for item in items if item["blockers"]),
"bytes": sum(item["byte_size"] or 0 for item in items),
}
report["capacity"] = self._capacity(report["totals"]["bytes"])
if not report["capacity"]["sufficient"]:
report["blockers"].append(
_issue(
"insufficient_capacity",
f"{report['totals']['bytes']} B plus a "
f"{self._config.archive_free_space_reserve_bytes} B reserve do not fit in "
f"{report['capacity']['free_bytes']} B of free space",
)
)
if not items:
report["blockers"].append(
_issue("empty_scope", "no archived assets are in the selected scope")
)
report["state"] = (
"ready"
if not report["blockers"] and not report["totals"]["blocked"]
else "blocked"
)
report["token"] = _token(report)
report["generated_at"] = _now().isoformat()
return report
def verify_token(self, token: str, location_id: str, asset_ids: list[str] | None = None) -> bool:
return bool(token) and token == self.preflight(location_id, asset_ids)["token"]
def _items(
self, session, location: ArchiveLocation, asset_ids: list[str] | None, *, reachable: bool
) -> list[dict]:
stmt = select(Asset).where(Asset.archive_location_id == location.id)
if asset_ids is None:
# A restored asset keeps its archive link; the default scope is only what
# is still archived, so restoring twice is an empty scope, not a blocker.
stmt = stmt.where(Asset.availability_state.in_(availability.ARCHIVED))
else:
stmt = stmt.where(Asset.id.in_(asset_ids))
assets = list(session.scalars(stmt.order_by(Asset.archive_path)))
if asset_ids is not None:
unknown = sorted(set(asset_ids) - {asset.id for asset in assets})
if unknown:
raise ArchiveError(
"unknown_asset", f"not archived at this location: {', '.join(unknown)}"
)
taken: set[str] = set()
return [self._item(asset, location, reachable=reachable, taken=taken) for asset in assets]
def _item(self, asset: Asset, location: ArchiveLocation, *, reachable: bool, taken: set) -> dict:
source = Path(location.root) / (asset.archive_path or "")
blockers: list[dict] = []
archive_sha256 = None
if asset.availability_state not in availability.ARCHIVED:
blockers.append(
_issue("not_archived", f"asset {asset.id} is {asset.availability_state}")
)
if reachable:
if not source.exists():
blockers.append(_issue("archive_missing", f"{source} is not on the medium"))
else:
archive_sha256 = sha256_file(source)
if asset.current_sha256 and archive_sha256 != asset.current_sha256:
blockers.append(
_issue(
"bytes_changed",
f"{source} holds bytes that are not the recorded ones; "
"the archived copy is divergent",
)
)
destination, destination_blockers = self._destination(asset, taken)
blockers += destination_blockers
if destination is not None:
taken.add(str(destination))
return {
"asset_id": asset.id,
"archive_path": asset.archive_path,
"source_path": str(source),
"destination_path": str(destination) if destination else None,
"expected_sha256": asset.current_sha256,
"archive_sha256": archive_sha256,
"byte_size": asset.byte_size,
"availability_state": asset.availability_state,
"blockers": blockers,
}
def _destination(self, asset: Asset, taken: set) -> tuple[Path | None, list[dict]]:
"""A free path inside the library that mirrors the archived layout.
Restoring onto an existing file is never an option, so a taken name is
answered with ``name (restored).ext`` — visible, ordinary, and impossible to
confuse with an overwrite.
"""
if not self._roots:
return None, [_issue("no_library_root", "no library root is configured")]
root = self._roots[0]
try:
candidate = resolve_within(root, root / (asset.archive_path or ""))
except PathPolicyError as error:
return None, [_issue("unsafe_destination", str(error))]
if is_excluded(candidate):
return None, [
_issue("unsafe_destination", f"{candidate} is inside an excluded (_IGNORE/) tree")
]
return _free_path(candidate, taken), []
def _destination_blockers(self, state: str, root: Path) -> list[dict]:
blockers: list[dict] = []
if not self._roots:
blockers.append(_issue("no_library_root", "no library root is configured"))
elif not os.access(self._roots[0], os.W_OK):
blockers.append(
_issue("library_not_writable", f"{self._roots[0]} is not writable")
)
if state == "offline":
blockers.append(
_issue("location_offline", f"the archive medium is not mounted at {root}")
)
elif state == "wrong_volume":
blockers.append(_issue("wrong_volume", f"{root} holds a different archive medium"))
return blockers
def _lock_blockers(self) -> list[dict]:
blockers: list[dict] = []
jobs = JobService(self._session_factory)
for lock in LOCKS:
held = jobs.blockers(lock)
if held:
blockers.append(
_issue("lock_conflict", f"the {lock} lane is busy: job {held[0]['id']}")
)
if RenameJournal(self._session_factory).blocks_mutation():
blockers.append(
_issue("rename_pending", "an unresolved rename must be recovered before restoring")
)
if self.journal.blocks_mutation():
blockers.append(
_issue(
"archive_pending",
"an unresolved archive or restore must be recovered before restoring",
)
)
return blockers
def _capacity(self, required: int) -> dict:
reserve = self._config.archive_free_space_reserve_bytes
free = shutil.disk_usage(self._roots[0]).free if self._roots else None
return {
"required_bytes": required,
"reserve_bytes": reserve,
"free_bytes": free,
"sufficient": free is not None and free >= required + reserve,
}
# ── plans ─────────────────────────────────────────────────────────────────
def create(self, location_id: str, asset_ids: list[str] | None = None, *, token: str) -> dict:
preflight = self.preflight(location_id, asset_ids)
if not token or token != preflight["token"]:
raise ArchiveError("stale_token", "the restore preflight changed since it was approved")
if preflight["state"] != "ready":
codes = ", ".join(sorted({issue["code"] for issue in preflight["blockers"]})) or "-"
blocked = sorted(
{issue["code"] for item in preflight["items"] for issue in item["blockers"]}
)
raise ArchiveError(
"blocked", f"the restore scope is blocked: {', '.join(blocked) or codes}"
)
plan_id = str(uuid.uuid4())
with self._session_factory() as session:
session.add(
ArchivePlan(
id=plan_id,
location_id=location_id,
token=token,
albums=json.dumps(asset_ids) if asset_ids is not None else None,
direction=RESTORE,
state="planned",
schema_version=PREFLIGHT_VERSION,
asset_count=preflight["totals"]["assets"],
byte_size=preflight["totals"]["bytes"],
)
)
session.flush()
for sequence, item in enumerate(preflight["items"]):
session.add(
ArchiveOperation(
id=str(uuid.uuid4()),
plan_id=plan_id,
direction=RESTORE,
sequence=sequence,
album=Path(item["archive_path"]).parent.name or "(root)",
asset_id=item["asset_id"],
source_path=item["source_path"],
destination_path=item["destination_path"],
archive_path=item["archive_path"],
expected_sha256=item["expected_sha256"],
byte_size=item["byte_size"],
journal_state=ArchiveState.PLANNED,
)
)
session.commit()
return self.get(plan_id)
def get(self, plan_id: str) -> dict | None:
with self._session_factory() as session:
plan = session.get(ArchivePlan, plan_id)
if plan is None or plan.direction != RESTORE:
return None
report = _plan_dict(plan)
report["operations"] = self.journal.operations(plan_id)
return report
def list(self) -> list[dict]:
with self._session_factory() as session:
rows = session.scalars(
select(ArchivePlan)
.where(ArchivePlan.direction == RESTORE)
.order_by(ArchivePlan.created_at)
)
return [_plan_dict(row) for row in rows]
# ── apply ─────────────────────────────────────────────────────────────────
def apply(
self, plan_id: str, *, expected_version: int | None = None, worker_id: str = "restore"
) -> dict:
plan = self._require_plan(plan_id)
if expected_version is not None and plan["version"] != expected_version:
raise ArchiveError(
"stale_plan",
f"plan {plan_id} is at version {plan['version']}, expected {expected_version}",
)
if plan["state"] not in APPLYABLE_PLAN_STATES:
raise ArchiveError("invalid_state", f"plan {plan_id} is {plan['state']}")
blocking = [row for row in self.journal.incomplete() if row["plan_id"] != plan_id]
if blocking:
raise ArchiveError(
"archive_pending",
f"another archive operation is unresolved ({blocking[0]['id']}); recover it first",
)
token = self._claim_plan(plan_id)
location = self._location(plan["location_id"])
restored = failed = skipped = 0
for operation in self.journal.operations(plan_id):
if operation["journal_state"] == ArchiveState.COMPLETE:
skipped += 1
continue
try:
if operation["journal_state"] == ArchiveState.VERIFIED:
self._finish(operation, token=token)
else:
self._restore_one(operation, location, token=token, worker_id=worker_id)
restored += 1
except PreconditionFailed as error:
self._fail(operation, token, error.code, str(error))
failed += 1
except Exception as error: # unexpected: record and stop touching disk
self._fail(operation, token, "restore_error", str(error))
failed += 1
state = self.journal.sync_plan_state(plan_id)
return {
"plan_id": plan_id,
"restored": restored,
"failed": failed,
"skipped": skipped,
"state": state,
}
def _restore_one(self, operation: dict, location: dict, *, token: int, worker_id: str) -> None:
source = Path(operation["source_path"])
destination = Path(operation["destination_path"])
# 1. Intent first; from here a crash is resolvable from journal + disk.
self.journal.begin(operation["id"], worker_id=worker_id, fencing_token=token)
maybe_fault(ArchiveState.TRANSFERRING)
# 2. Recheck against the medium and the library as they are right now.
self._recheck(operation, source, destination, location)
destination.parent.mkdir(parents=True, exist_ok=True)
# 3. Always copy: the archived original stays on its medium.
copy_verify_publish(source, destination, operation["expected_sha256"])
_fsync_dir(destination.parent)
if sha256_file(destination) != operation["expected_sha256"]:
raise PreconditionFailed(
"restore_mismatch", f"{destination} does not hold the expected bytes"
)
self.journal.transition(operation["id"], ArchiveState.VERIFIED, fencing_token=token)
maybe_fault(ArchiveState.VERIFIED)
self._finish(self.journal.get(operation["id"]), token=token)
def _finish(self, operation: dict, *, token: int) -> None:
"""Publish the restored file to the database. Idempotent, so recovery may
replay it after a crash between the copy and the bookkeeping."""
destination = Path(operation["destination_path"])
if not destination.exists() or sha256_file(destination) != operation["expected_sha256"]:
raise PreconditionFailed(
"restore_unverified", f"{destination} is not a verified restored copy"
)
self._record_restored(operation, destination)
self.journal.transition(operation["id"], ArchiveState.COMPLETE, fencing_token=token)
maybe_fault(ArchiveState.COMPLETE)
def _recheck(self, operation: dict, source: Path, destination: Path, location: dict) -> None:
root = Path(location["root"])
if not root.is_dir() or not (root / availability.MARKER_NAME).exists():
raise PreconditionFailed("location_offline", f"{root} is not the archive medium")
if not source.exists():
raise PreconditionFailed("archive_missing", f"{source} is not on the medium")
if source.is_symlink() or destination.is_symlink():
raise PreconditionFailed("symlink", "refusing to restore through a symlink")
if destination.exists():
# Never overwrite: the plan's free path was taken since it was made.
raise PreconditionFailed(
"destination_exists", f"destination {destination} is occupied"
)
if not self._inside_library(destination):
raise PreconditionFailed(
"destination_escape", f"{destination} is outside the library roots"
)
if sha256_file(source) != operation["expected_sha256"]:
self._mark_divergent(operation["asset_id"])
raise PreconditionFailed(
"bytes_changed", f"{source} changed since the plan was approved"
)
with self._session_factory() as session:
asset = session.get(Asset, operation["asset_id"])
if asset is None or asset.availability_state not in availability.ARCHIVED:
raise PreconditionFailed(
"not_archived", f"asset {operation['asset_id']} is no longer archived"
)
def _inside_library(self, destination: Path) -> bool:
for root in self._roots:
try:
resolve_within(root, destination)
return True
except PathPolicyError:
continue
return False
# ── database ──────────────────────────────────────────────────────────────
def _record_restored(self, operation: dict, destination: Path) -> None:
"""The bytes are back in the library: open the new active occurrence and set
availability. Identity, decisions, and history are untouched — that is the
entire point of restoring rather than re-importing."""
now = _now()
with self._session_factory() as session:
asset = session.get(Asset, operation["asset_id"])
if asset is None:
raise PreconditionFailed(
"asset_missing", f"asset {operation['asset_id']} no longer exists"
)
# A restored asset may be returning to a path it once held, so only an
# *open* occurrence counts as already registered — that is what keeps
# recovery idempotent without collapsing the path history.
recorded = session.scalar(
select(AssetPath).where(
AssetPath.asset_id == asset.id,
AssetPath.path == str(destination),
AssetPath.valid_until.is_(None),
)
)
if recorded is None: # idempotent: recovery may replay this
session.add(
AssetPath(
asset_id=asset.id,
path=str(destination),
valid_from=now,
reason="restore",
)
)
asset.current_path = str(destination)
asset.availability_state = availability.ACTIVE
asset.missing_at = None
# The archive copy stays where it is; keeping the link means a restored
# asset still knows which medium holds its archived bytes.
asset.archive_divergent_at = None
asset.state_version += 1
asset.updated_at = now
session.commit()
def _mark_divergent(self, asset_id: str) -> None:
"""Record that the archived copy is not the recorded file. Durable, because
the next restore attempt must not rediscover this from scratch."""
with self._session_factory() as session:
asset = session.get(Asset, asset_id)
if asset is None:
return
asset.archive_divergent_at = _now()
asset.state_version += 1
session.commit()
# ── recovery ──────────────────────────────────────────────────────────────
def recover(self, *, worker_id: str = "restore-recovery") -> dict:
"""Resolve every incomplete restore from journal + disk evidence.
A restore never removed anything, so ``resumable`` simply discards the
temporary debris and re-plans the item; ``forward`` finishes the bookkeeping
for a published file; ``manual`` is left untouched and keeps blocking.
"""
results = {"resumed": 0, "completed": 0, "manual": 0}
touched: set[str] = set()
for verdict in self.journal.classify_all(direction=RESTORE):
operation = self.journal.get(verdict["operation_id"])
touched.add(operation["plan_id"])
token = (operation["fencing_token"] or 0) + 1
if verdict["classification"] == MANUAL:
results["manual"] += 1
continue
if verdict["classification"] == RESUMABLE:
_clean_temp_files(Path(operation["destination_path"]).parent)
self.journal.transition(operation["id"], ArchiveState.PLANNED, fencing_token=token)
results["resumed"] += 1
continue
try:
self._finish(operation, token=token)
results["completed"] += 1
except PreconditionFailed as error:
self._fail(operation, token, error.code, str(error))
results["manual"] += 1
for plan_id in touched:
self.journal.sync_plan_state(plan_id)
return results
def recovery_status(self) -> dict:
verdicts = self.journal.classify_all(direction=RESTORE)
return {
"operations": verdicts,
"manual": [v for v in verdicts if v["classification"] == MANUAL],
"blocks_mutation": self.journal.blocks_mutation(),
}
# ── helpers ───────────────────────────────────────────────────────────────
def _fail(self, operation: dict, token: int, code: str, message: str) -> None:
self.journal.transition(
operation["id"], ArchiveState.FAILED, fencing_token=token, error=(code, message)
)
def _require_plan(self, plan_id: str) -> dict:
plan = self.get(plan_id)
if plan is None:
raise ArchiveError("unknown_plan", f"unknown restore plan {plan_id!r}")
return plan
def _location(self, location_id: str) -> dict:
with self._session_factory() as session:
location = session.get(ArchiveLocation, location_id)
if location is None:
raise ArchiveError("unknown_location", f"unknown archive location {location_id!r}")
return {"id": location.id, "root": location.root, "media_id": location.media_id}
def _claim_plan(self, plan_id: str) -> int:
with self._session_factory() as session:
plan = session.get(ArchivePlan, plan_id)
plan.version += 1
plan.state = "applying"
plan.updated_at = _now()
token = plan.version
session.commit()
return token
# ── module helpers ───────────────────────────────────────────────────────────
def _location_state(root: Path, marker: dict | None, media_id: str) -> str:
if not root.is_dir() or marker is None:
return "offline"
return "online" if marker.get("media_id") == media_id else "wrong_volume"
def _free_path(candidate: Path, taken: set) -> Path:
"""``a.jpg`` → ``a (restored).jpg`` → ``a (restored 2).jpg`` …
``taken`` holds the destinations already claimed by earlier items of the same
plan, so two restores in one scope cannot plan the same path.
"""
if not candidate.exists() and str(candidate) not in taken:
return candidate
stem, suffix = candidate.stem, candidate.suffix
attempt = 1
while True:
label = RESTORED_SUFFIX if attempt == 1 else f"{RESTORED_SUFFIX} {attempt}"
alternative = candidate.with_name(f"{stem} ({label}){suffix}")
if not alternative.exists() and str(alternative) not in taken:
return alternative
attempt += 1
def _token(report: dict) -> str:
"""Digest of everything the report asserts about the scope and the medium.
Free space is excluded: it drifts constantly without changing what a restore
would do, and the capacity verdict itself is part of the digest.
"""
payload = {key: value for key, value in report.items() if key not in ("generated_at", "token")}
payload["capacity"] = {
key: value for key, value in payload["capacity"].items() if key != "free_bytes"
}
digest = hashlib.sha256(
json.dumps(payload, sort_keys=True, ensure_ascii=False, default=str).encode("utf-8")
).hexdigest()
return f"{TOKEN_PREFIX}:{digest}"