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

US06-04: Plan and Execute Safe Restores (#80)

Browse Source
This commit was merged in pull request #80.
This commit is contained in:
domverse
2026-08-16 19:49:21 +02:00
parent 439eb9971e
commit fb8567284d
10 changed files with 1294 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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