414 lines
17 KiB
Python
414 lines
17 KiB
Python
"""Online backups, verification, retention, and restore drills (US07-05).
|
|
|
|
A backup taken by copying a live SQLite file is not a backup: with WAL enabled the
|
|
file on disk is missing every committed page still in the write-ahead log, and a
|
|
writer mid-transaction makes the copy inconsistent. So every backup here goes
|
|
through SQLite's online backup API, which takes a consistent snapshot of a database
|
|
that is still being used (concept §3).
|
|
|
|
A backup directory holds exactly two things:
|
|
|
|
photo_pipeline.db the snapshot
|
|
manifest.json what it is, what it came from, and how to check it
|
|
|
|
The manifest is what makes the snapshot restorable by someone who was not there
|
|
when it was taken: the schema revision, the snapshot's SHA-256, the row counts it
|
|
should still have, the archive locations whose media the library depends on, and
|
|
which configuration values were set — **names and non-secret values only**. A
|
|
secret is recorded as "configured", never as its value, so a manifest can be
|
|
attached to a bug report.
|
|
|
|
Restore never writes into a live installation: it refuses a target that already
|
|
holds a database, because the one thing worse than a lost library is a half-merged
|
|
one. The drill is documented in README ("Backup and recovery").
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import shutil
|
|
import sqlite3
|
|
from contextlib import closing
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
from sqlalchemy import text
|
|
|
|
from photo_pipeline.config import Config
|
|
from photo_pipeline.db import create_db_engine, create_session_factory
|
|
|
|
SCHEMA_VERSION = 1
|
|
DB_NAME = "photo_pipeline.db"
|
|
MANIFEST_NAME = "manifest.json"
|
|
# How many backups the retention helper keeps by default. Small on purpose: a
|
|
# backup is a snapshot of state that is itself recoverable from the library, and
|
|
# the disk it lives on is the same one the low-disk warning watches.
|
|
DEFAULT_KEEP = 7
|
|
# Tables whose row counts are worth proving after a restore. Not the whole schema —
|
|
# these are the ones whose loss would be silent.
|
|
COUNTED_TABLES = (
|
|
"assets",
|
|
"asset_paths",
|
|
"safety_reviews",
|
|
"analysis_results",
|
|
"exif_projections",
|
|
"upload_batches",
|
|
"upload_items",
|
|
"archive_locations",
|
|
"archive_plans",
|
|
"archive_operations",
|
|
"rename_plans",
|
|
"rename_operations",
|
|
)
|
|
|
|
|
|
class BackupError(RuntimeError):
|
|
"""The backup could not be created, read, verified, or restored."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class VerifyResult:
|
|
ok: bool
|
|
issues: tuple[str, ...] = ()
|
|
revision: str | None = None
|
|
counts: dict | None = None
|
|
|
|
def as_dict(self) -> dict:
|
|
return {
|
|
"ok": self.ok,
|
|
"issues": list(self.issues),
|
|
"revision": self.revision,
|
|
"counts": self.counts,
|
|
}
|
|
|
|
|
|
def _now() -> datetime:
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
def sha256_file(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as handle:
|
|
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def _revision(database: Path) -> str | None:
|
|
with closing(sqlite3.connect(database)) as connection:
|
|
try:
|
|
row = connection.execute("SELECT version_num FROM alembic_version").fetchone()
|
|
except sqlite3.Error:
|
|
return None
|
|
return row[0] if row else None
|
|
|
|
|
|
def _counts(database: Path) -> dict:
|
|
counts: dict[str, int] = {}
|
|
with closing(sqlite3.connect(database)) as connection:
|
|
for table in COUNTED_TABLES:
|
|
try:
|
|
counts[table] = connection.execute(f"SELECT count(*) FROM {table}").fetchone()[0]
|
|
except sqlite3.Error:
|
|
continue # a table this revision does not have yet
|
|
return counts
|
|
|
|
|
|
def _integrity(database: Path) -> tuple[str, list[str]]:
|
|
"""``PRAGMA integrity_check`` plus ``foreign_key_check`` — structure and links.
|
|
|
|
Structural soundness is not referential soundness: a database can pass
|
|
``integrity_check`` and still hold an upload item pointing at an asset that
|
|
is gone.
|
|
"""
|
|
issues: list[str] = []
|
|
with closing(sqlite3.connect(database)) as connection:
|
|
try:
|
|
result = connection.execute("PRAGMA integrity_check").fetchone()[0]
|
|
if result != "ok":
|
|
issues.append(f"integrity_check: {result}")
|
|
violations = connection.execute("PRAGMA foreign_key_check").fetchall()
|
|
if violations:
|
|
issues.append(f"foreign_key_check: {len(violations)} violation(s)")
|
|
except sqlite3.DatabaseError as error:
|
|
issues.append(f"unreadable: {error}")
|
|
return "error", issues
|
|
return "ok" if not issues else "damaged", issues
|
|
|
|
|
|
def configuration_references(config: Config) -> dict:
|
|
"""Which configuration a restore has to reproduce — never the secrets themselves.
|
|
|
|
Paths and URLs are recorded because a restore into a fresh root has to be told
|
|
where the library and the Immich server were; API keys are recorded as
|
|
``configured`` so an operator knows one is required without the manifest ever
|
|
carrying it.
|
|
"""
|
|
return {
|
|
"data_dir": str(config.data_dir),
|
|
"database_path": str(config.database_path),
|
|
"library_roots": [str(root) for root in config.library_roots],
|
|
"thumbnail_cache_dir": str(config.thumbnail_cache_dir),
|
|
"immich_server_url": config.immich_server_url,
|
|
"immich_go_binary": config.immich_go_binary,
|
|
"secrets": {
|
|
"immich_api_key": "configured" if config.immich_api_key else "unset",
|
|
"vision_api_key": "configured" if config.vision_api_key else "unset",
|
|
},
|
|
}
|
|
|
|
|
|
def migrate_with_backup(config: Config) -> dict | None:
|
|
"""Upgrade the schema, with a snapshot first when there is state to lose.
|
|
|
|
A migration is the one routine operation that can damage every record at once,
|
|
and Alembic's own transaction does not cover SQLite DDL reliably. So a pending
|
|
upgrade is preceded by an online backup, and a failed upgrade names it in the
|
|
error: recovery is "restore that directory", not "reconstruct the library".
|
|
Returns the manifest of the backup it took, or ``None`` when none was needed.
|
|
"""
|
|
import logging
|
|
|
|
from photo_pipeline.db import run_migrations
|
|
|
|
service = BackupService(config)
|
|
manifest = service.pre_migration() if service.migration_pending() else None
|
|
try:
|
|
run_migrations(config.database_url)
|
|
except Exception:
|
|
if manifest is not None:
|
|
logging.getLogger(__name__).error(
|
|
"migration failed; restore the pre-migration backup at %s",
|
|
service.root / manifest["name"],
|
|
)
|
|
raise
|
|
return manifest
|
|
|
|
|
|
class BackupService:
|
|
def __init__(self, config: Config) -> None:
|
|
self._config = config
|
|
|
|
@property
|
|
def root(self) -> Path:
|
|
return self._config.data_dir / "backups"
|
|
|
|
# ── create ────────────────────────────────────────────────────────────────
|
|
|
|
def create(self, *, reason: str = "manual", keep: int | None = DEFAULT_KEEP) -> dict:
|
|
"""Take an online snapshot and describe it. Returns the manifest."""
|
|
source = self._config.database_path
|
|
if not source.exists():
|
|
raise BackupError(f"no database at {source}")
|
|
|
|
stamp = _now().strftime("%Y%m%dT%H%M%SZ")
|
|
safe_reason = "".join(c for c in reason if c.isalnum() or c in "-_") or "manual"
|
|
directory = self.root / f"{stamp}-{safe_reason}"
|
|
if directory.exists(): # same second, same reason
|
|
directory = self.root / f"{stamp}-{safe_reason}-{len(list(self.root.iterdir()))}"
|
|
directory.mkdir(parents=True)
|
|
|
|
target = directory / DB_NAME
|
|
try:
|
|
with closing(sqlite3.connect(source)) as src, closing(sqlite3.connect(target)) as dst:
|
|
src.backup(dst) # the online backup API, not a file copy
|
|
except (sqlite3.Error, OSError) as error:
|
|
shutil.rmtree(directory, ignore_errors=True)
|
|
raise BackupError(f"backup failed: {error}") from error
|
|
|
|
state, issues = _integrity(target)
|
|
manifest = {
|
|
"schema_version": SCHEMA_VERSION,
|
|
"name": directory.name,
|
|
"created_at": _now().isoformat(),
|
|
"reason": reason,
|
|
"revision": _revision(target),
|
|
"database": {
|
|
"name": DB_NAME,
|
|
"bytes": target.stat().st_size,
|
|
"sha256": sha256_file(target),
|
|
"integrity": state,
|
|
"issues": issues,
|
|
},
|
|
"counts": _counts(target),
|
|
"archive_locations": self._archive_locations(),
|
|
"configuration": configuration_references(self._config),
|
|
"retention": {
|
|
"keep": keep,
|
|
"guidance": (
|
|
"Keep the newest snapshot on a different disk than data_dir, and one "
|
|
"off-site copy per archive medium. A backup only covers the database: "
|
|
"the photos themselves live in the library and archive locations named "
|
|
"above, which need their own copies."
|
|
),
|
|
},
|
|
}
|
|
(directory / MANIFEST_NAME).write_text(json.dumps(manifest, indent=2))
|
|
if keep is not None:
|
|
manifest["pruned"] = self.prune(keep=keep)
|
|
return manifest
|
|
|
|
def migration_pending(self) -> bool:
|
|
"""True when the database exists and is not at the revision this code wants."""
|
|
from photo_pipeline.db import current_revision, head_revision
|
|
|
|
if not self._config.database_path.exists():
|
|
return False
|
|
return current_revision(self._config.database_url) != head_revision()
|
|
|
|
def pre_migration(self) -> dict | None:
|
|
"""Snapshot before a schema change, when there is something to lose.
|
|
|
|
Returns ``None`` when the database does not exist yet (a fresh install has
|
|
no state a failed migration could damage).
|
|
"""
|
|
if not self._config.database_path.exists():
|
|
return None
|
|
return self.create(reason="pre-migration")
|
|
|
|
def _archive_locations(self) -> list[dict]:
|
|
"""The media the library's archived originals live on.
|
|
|
|
A restored database still points at these; if they are not restored too,
|
|
the pictures are gone even though every record survived.
|
|
"""
|
|
engine = create_db_engine(self._config.database_url)
|
|
try:
|
|
factory = create_session_factory(engine)
|
|
with factory() as session:
|
|
rows = session.execute(
|
|
text("SELECT id, name, root, media_id, state FROM archive_locations")
|
|
).mappings().all()
|
|
except Exception:
|
|
return []
|
|
finally:
|
|
engine.dispose()
|
|
return [
|
|
{
|
|
"id": row["id"],
|
|
"name": row["name"],
|
|
"root": row["root"],
|
|
"media_id": row["media_id"],
|
|
"last_state": row["state"],
|
|
"mounted": Path(row["root"]).is_dir(),
|
|
}
|
|
for row in rows
|
|
]
|
|
|
|
# ── inspect ───────────────────────────────────────────────────────────────
|
|
|
|
def list(self) -> list[dict]:
|
|
"""Every backup, newest first, with what is known about it."""
|
|
if not self.root.is_dir():
|
|
return []
|
|
entries = []
|
|
for directory in sorted(self.root.iterdir(), reverse=True):
|
|
if not directory.is_dir():
|
|
continue
|
|
manifest = self.manifest(directory)
|
|
database = directory / DB_NAME
|
|
entries.append(
|
|
{
|
|
"name": directory.name,
|
|
"path": str(directory),
|
|
"created_at": (manifest or {}).get("created_at"),
|
|
"reason": (manifest or {}).get("reason"),
|
|
"revision": (manifest or {}).get("revision"),
|
|
"bytes": database.stat().st_size if database.exists() else 0,
|
|
"complete": bool(manifest) and database.exists(),
|
|
}
|
|
)
|
|
return entries
|
|
|
|
def manifest(self, directory: Path) -> dict | None:
|
|
path = Path(directory) / MANIFEST_NAME
|
|
if not path.exists():
|
|
return None
|
|
try:
|
|
return json.loads(path.read_text())
|
|
except ValueError:
|
|
return None
|
|
|
|
def verify(self, directory: Path | str) -> VerifyResult:
|
|
"""Prove a snapshot is still the one that was taken and still readable."""
|
|
directory = Path(directory)
|
|
if not directory.is_dir():
|
|
return VerifyResult(False, (f"no backup at {directory}",))
|
|
manifest = self.manifest(directory)
|
|
if manifest is None:
|
|
return VerifyResult(False, ("manifest is missing or unreadable",))
|
|
database = directory / manifest["database"]["name"]
|
|
if not database.exists():
|
|
return VerifyResult(False, ("the snapshot file is missing",), manifest.get("revision"))
|
|
|
|
issues: list[str] = []
|
|
if sha256_file(database) != manifest["database"]["sha256"]:
|
|
# Bit rot, a truncated copy, or an edited snapshot: all three mean the
|
|
# bytes are not the ones that were verified when the backup was made.
|
|
issues.append("sha256 does not match the manifest")
|
|
state, structural = _integrity(database)
|
|
issues.extend(structural)
|
|
counts = _counts(database) if state != "error" else None
|
|
if counts is not None and manifest.get("counts") and counts != manifest["counts"]:
|
|
issues.append(f"row counts changed: {manifest['counts']} -> {counts}")
|
|
return VerifyResult(not issues, tuple(issues), manifest.get("revision"), counts)
|
|
|
|
# ── retention ─────────────────────────────────────────────────────────────
|
|
|
|
def prune(self, *, keep: int = DEFAULT_KEEP) -> list[str]:
|
|
"""Delete the oldest backups beyond ``keep``. Never deletes the newest one."""
|
|
if keep < 1:
|
|
raise BackupError("retention must keep at least one backup")
|
|
removed = []
|
|
for entry in self.list()[keep:]:
|
|
shutil.rmtree(entry["path"], ignore_errors=True)
|
|
removed.append(entry["name"])
|
|
return removed
|
|
|
|
# ── restore ───────────────────────────────────────────────────────────────
|
|
|
|
def restore(self, directory: Path | str, target_data_dir: Path | str) -> dict:
|
|
"""Restore a verified snapshot into a **fresh** data directory.
|
|
|
|
Refuses a target that already holds a database. Restoring on top of a live
|
|
installation would merge two histories that disagree about which files were
|
|
renamed, uploaded, and archived — the one failure this whole story exists to
|
|
prevent. Recovering in place is: stop everything, move the old data
|
|
directory aside, restore into a new one.
|
|
"""
|
|
directory = Path(directory)
|
|
result = self.verify(directory)
|
|
if not result.ok:
|
|
raise BackupError(f"refusing to restore an unverified backup: {result.issues}")
|
|
|
|
target = Path(target_data_dir)
|
|
target.mkdir(parents=True, exist_ok=True)
|
|
destination = target / DB_NAME
|
|
if destination.exists():
|
|
raise BackupError(
|
|
f"{destination} already exists; restore into a fresh data directory"
|
|
)
|
|
shutil.copy2(directory / DB_NAME, destination)
|
|
# The write-ahead log of the *source* installation must not travel with a
|
|
# snapshot: the backup API already folded every committed page into it.
|
|
for leftover in (target / f"{DB_NAME}-wal", target / f"{DB_NAME}-shm"):
|
|
leftover.unlink(missing_ok=True)
|
|
|
|
restored = _integrity(destination)
|
|
return {
|
|
"backup": directory.name,
|
|
"restored_to": str(destination),
|
|
"revision": result.revision,
|
|
"counts": _counts(destination),
|
|
"integrity": restored[0],
|
|
"issues": restored[1],
|
|
"next_steps": [
|
|
"point PHOTO_PIPELINE_DATA_DIR at the restored directory",
|
|
"run `python -m photo_pipeline migrate` to reach the current revision",
|
|
"run an inventory scan so paths are reconciled against the real library",
|
|
"mount every archive location listed in the manifest before archiving again",
|
|
],
|
|
}
|