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

US07-05: Deliver Backup and Operational Recovery #92

Merged
domverse merged 1 commits from us/US07-05-deliver-backup-and-operational-recovery into main 2026-08-17 21:13:22 +02:00
Conversation 0 Commits 1 Files Changed 19 +2057 -20
19 changed files with 2057 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

74
README.md
View File

@@ -281,6 +281,80 @@ file in the temporary library are copied to `.artifacts/<test id>/` before pytes
deletes the directory. Point `PHOTO_PIPELINE_TEST_ARTIFACTS` elsewhere to collect
them from CI.
## Backup and recovery (US07-05)
Backups go through SQLite's online backup API, never a file copy: with WAL enabled
the `.db` file alone is missing every committed page still in the write-ahead log.
Each backup is a directory under `data/backups/` holding the snapshot and a
`manifest.json` describing it — schema revision, SHA-256, row counts, the archive
media the library depends on, and which configuration was set. Secrets are recorded
as `configured`, never as values, so a manifest is safe to attach to a bug report.
```bash
work_item/scripts/python -m photo_pipeline backup --reason before-upgrade --keep 7
work_item/scripts/python -m photo_pipeline verify-backup data/backups/<name>
work_item/scripts/python -m photo_pipeline diagnostics
```
The same is available at `GET /api/v1/diagnostics`, `GET|POST /api/v1/backups`,
`GET /api/v1/backups/{name}/verify`, and `POST /api/v1/backups/prune`. **Restore is
not an endpoint** — it replaces the state of an installation, so it belongs to a
stopped one and a person at a terminal.
### Integrity check
`verify-backup` runs `PRAGMA integrity_check` (structure) *and*
`PRAGMA foreign_key_check` (references), compares the snapshot's SHA-256 with the
manifest, and re-counts every table the manifest recorded. Any mismatch — bit rot, a
truncated copy, a "repaired" snapshot — fails the check, and `restore` refuses a
backup that does not verify.
### Restore drill
1. Stop the server and the worker.
2. `python -m photo_pipeline verify-backup data/backups/<name>` — never restore an
unverified snapshot.
3. `python -m photo_pipeline restore data/backups/<name> --into /path/to/fresh-data`
(a target that already holds a database is refused; recovering in place means
moving the old data directory aside first).
4. Point `PHOTO_PIPELINE_DATA_DIR` at the restored directory and run
`python -m photo_pipeline migrate`.
5. Run an inventory scan so paths are reconciled against the real library.
6. Mount every archive location named in the manifest before archiving again — the
database records where archived originals are, but it does not contain them.
Practise this against a copy before you need it; the drill is exercised
automatically by `tests/integration/test_backup_recovery.py`.
### Failed migration
A pending schema upgrade is snapshotted first (`reason: pre-migration`), by both the
API startup and `python -m photo_pipeline migrate`. If a migration fails, the error
log names the backup directory: stop everything and run the restore drill against
it. An up-to-date database is not backed up again on every start.
### Archive media
Archived originals live on their medium, not in the backup. The manifest lists every
archive location with its `media_id` and whether it was mounted when the backup was
taken. Keep one copy of each medium off-site, and remount a location before
restoring assets from it.
### Retention and disk
`--keep N` (default 7) prunes the oldest backups and never the newest.
`diagnostics` reports the database, write-ahead log, thumbnail cache, uploader
reports, backups, and logs separately, with free space and warnings for low disk
(`disk_low`, `disk_critical`), a cache over its quota, a write-ahead log outgrowing
its database, and a legacy CLI writing the library.
### Process locking
`serve` and `worker` take a JSON lock in the data directory (`api.lock.json`,
`worker.lock.json`). A second worker exits `2` and names the holder; a lock whose
process is gone is taken over. If the frozen CLI's state files are being written,
both refuse with exit `3` — `--allow-legacy` overrides, and you own the outcome.
## Legacy CLI archive
The command-line tools this application was extracted from are frozen in

32
delivery_backlog/E08-container-deployment.md Normal file
View File

@@ -0,0 +1,32 @@
# E08 — Container Deployment
Concept phase: none. This epic is a delivery-format addition on top of the concept:
the same application, same safety invariants, packaged as a Docker image and deployed
continuously from Gitea Actions instead of being started by hand from a working copy.
It does not change the product scope in
[`INTEGRATED_PIPELINE_CONCEPT.md`](../INTEGRATED_PIPELINE_CONCEPT.md). SQLite stays the
store, one worker stays the writer, the library process lock stays authoritative, and
no path outside the configured library roots becomes reachable because the process now
runs in a container.
One decision does extend the concept and is made here explicitly: the application may
be reached through a reverse proxy under a real hostname, not only over loopback. That
requires a configurable trust boundary and an authentication gate, because the
loopback-only checks of US07-02 are what currently stand in for authentication.
## Stories
1. [US08-01 — Make the trust boundary configurable and authenticated](stories/US08-01-trusted-hosts-auth.md)
2. [US08-02 — Build a reproducible application image](stories/US08-02-container-image.md)
3. [US08-03 — Compose the runtime and mount the library safely](stories/US08-03-compose-runtime.md)
4. [US08-04 — Publish and deploy from Gitea Actions](stories/US08-04-gitea-cicd.md)
5. [US08-05 — Automate container deployment acceptance](stories/US08-05-container-e2e.md)
## Epic outcome
A tagged image built from `main` runs the API and the worker as separate containers
against a mounted library and a persistent data volume, is published to the Gitea
registry, is redeployed by webhook, survives restart and upgrade with its database and
journals intact, and refuses every request that a loopback deployment would have
refused.

7
delivery_backlog/README.md
View File

@@ -2,11 +2,13 @@
This backlog decomposes the phases in
[`INTEGRATED_PIPELINE_CONCEPT.md`](../INTEGRATED_PIPELINE_CONCEPT.md) into seven
epics and small, independently verifiable user stories.
epics and small, independently verifiable user stories, plus one delivery-format
epic (E08) that packages the released application as a deployable container.
## Numbering and file naming
- Epics: `E01` through `E07`, matching concept Phases A through G.
- Epics: `E01` through `E07`, matching concept Phases A through G; `E08` has no
concept phase and must not change product scope.
- Stories: `US<epic>-<sequence>`, for example `US03-02`.
- Epic files: `E01-<slug>.md`.
- Story files: `stories/US01-01-<slug>.md`.
@@ -36,6 +38,7 @@ epics and small, independently verifiable user stories.
5. [E05 — Immich upload](E05-immich-upload.md)
6. [E06 — Archive lifecycle](E06-archive-lifecycle.md)
7. [E07 — Hardening and release](E07-hardening-release.md)
8. [E08 — Container deployment](E08-container-deployment.md)
## Shared definition of done

42
delivery_backlog/stories/US08-01-trusted-hosts-auth.md Normal file
View File

@@ -0,0 +1,42 @@
# US08-01 — Make the Trust Boundary Configurable and Authenticated
Epic: [E08](../E08-container-deployment.md)
As an operator, I want to reach the application through my own hostname without
weakening it, so a container behind a reverse proxy is as safe as the loopback
deployment it replaces.
## Context
`photo_pipeline/api/security.py` refuses any request whose `Host` or `Origin` is not
loopback. That check is the current stand-in for authentication: whoever can reach
`127.0.0.1:8000` is the owner. Behind a proxy the hostname is no longer loopback, so
relaxing the check without adding an authentication gate would publish the library.
## Acceptance criteria
- Allowed hosts and origins come from configuration (`PHOTO_PIPELINE_*`), default to
the current loopback set, and an unset configuration behaves exactly as today.
- Whenever a non-loopback host is configured, startup requires an access secret and
refuses to serve without one; loopback-only deployments keep working with no secret.
- The secret is exchanged for the existing session cookie and CSRF token through the
bootstrap endpoint; every protected route keeps its current session and CSRF
requirements unchanged.
- Forwarded headers (`X-Forwarded-Proto`, `X-Forwarded-Host`) are honored only from a
configured trusted proxy and ignored otherwise, so a client cannot forge its origin.
- Cookies are marked `Secure` when the effective external scheme is HTTPS.
- Failed authentication is rate-limited and logged without the secret, the session id,
or any request body.
- Health endpoints stay reachable without the secret; nothing else does.
## Automated tests
- Unit tests for host/origin evaluation across loopback default, configured host,
unconfigured host, forged forwarded headers, and trusted-proxy forwarded headers.
- Integration tests: startup refusal without a secret, successful exchange, wrong
secret, replay of an old session, cross-site request, and unauthenticated access to
every route class.
## Dependencies
- US07-02

41
delivery_backlog/stories/US08-02-container-image.md Normal file
View File

@@ -0,0 +1,41 @@
# US08-02 — Build a Reproducible Application Image
Epic: [E08](../E08-container-deployment.md)
As an operator, I want one image that can run either application role, so deployment is
a pull instead of a Python environment I have to reproduce by hand.
## Context
The application shells out to `exiftool` and `immich-go`, writes into the library as a
normal filesystem user, and serves a static frontend from `frontend/`. All three have to
be true inside the image, or the container starts and then fails on the first real
operation.
## Acceptance criteria
- A `Dockerfile` builds from a pinned Python base, installs the project and its runtime
dependencies, and contains no test, playwright, or build-only tooling in the final
layer.
- `exiftool` and `immich-go` are present at pinned versions, and their versions are
recorded in the image and reported by `python -m photo_pipeline diagnostics`.
- The image runs as a non-root user whose UID/GID are build-time arguments, so files
the application renames or writes keep the ownership the host library expects.
- One entrypoint selects the role: `serve` or `worker`, passing through the existing
CLI arguments; no supervisor runs two roles in one container.
- `serve` containers declare a `HEALTHCHECK` against `/api/v1/health/ready`, so an
unmigrated or misconfigured database is not reported healthy.
- The image contains no secrets, no library data, no database, and no `.git`; the build
context is constrained by `.dockerignore`.
- Image build is reproducible from a clean checkout and documented in `README.md`.
## Automated tests
- A build-and-run test asserts the image starts, reports ready, serves the frontend
index, and returns the pinned `exiftool` and `immich-go` versions.
- A test asserts the container refuses to run as UID 0 and that a file created by the
container is owned by the configured UID/GID.
## Dependencies
- US07-05

47
delivery_backlog/stories/US08-03-compose-runtime.md Normal file
View File

@@ -0,0 +1,47 @@
# US08-03 — Compose the Runtime and Mount the Library Safely
Epic: [E08](../E08-container-deployment.md)
As an operator, I want a single compose file that runs the API and the worker against my
real library, so a deployment is one command and the safety invariants survive it.
## Context
The library process lock (US07-05) assumes both roles see the same lock file, and SQLite
in WAL mode assumes a real local filesystem. Container path policy is the same problem
as host path policy with a new failure mode: the configured library roots must name the
in-container mount paths, not the host paths.
## Acceptance criteria
- `docker-compose.yml` runs exactly one `serve` and one `worker` container from the same
image and the same data volume, and a second worker is refused by the existing lock
rather than by convention.
- The library is a bind mount; `PHOTO_PIPELINE_LIBRARY_ROOTS` names the container-side
paths, and a mismatch between mounted and configured roots fails at startup with a
clear message instead of at the first write.
- The data volume holds the database, WAL, thumbnail cache, and backups on a local
filesystem; the composition documents that a network mount is unsupported for it.
- Migrations run before `serve` and `worker` accept work, using the existing backup-then-
migrate path, and an upgrade that fails leaves the previous database intact.
- Configuration and secrets come from the environment, never from the image or a
committed file; a `.env.example` lists every `PHOTO_PIPELINE_*` variable with safe
defaults and no values.
- The API port is published to host loopback by default; exposing it publicly requires
the configured hostname and access secret from US08-01.
- Containers restart automatically, and a restart mid-job resumes exactly as a host
restart does today.
- Backup, verify-backup, restore, and diagnostics are documented as container commands
and work against the mounted volumes.
## Automated tests
- An integration test brings the composition up against a temporary fixture library,
runs a job, restarts both containers, and asserts the job resumes and the database is
intact.
- Tests for: second worker refused, library-root mismatch refused at startup, failed
migration leaving the previous database restorable.
## Dependencies
- US08-01, US08-02

40
delivery_backlog/stories/US08-04-gitea-cicd.md Normal file
View File

@@ -0,0 +1,40 @@
# US08-04 — Publish and Deploy from Gitea Actions
Epic: [E08](../E08-container-deployment.md)
As a release owner, I want `main` to build, publish, and redeploy the image
automatically, so deployment is the same reproducible path every time.
## Context
The workflow is adapted from the `crowdsec-admin` deployment workflow
(`.gitea/workflows/deploy.yml` in that repository): build, log in to the Gitea registry,
push, trigger a Portainer webhook, prune. This project needs the same shape plus a test
gate, because unlike that project it has a required suite that must not be skipped.
## Acceptance criteria
- `.gitea/workflows/` contains a test workflow that runs on pull requests and on `main`,
executing the configured required suites, and a deploy workflow that runs only after
the tests pass on `main` and on manual dispatch.
- The deploy workflow publishes to `git.domverse-berlin.eu` under this project's own
image path, tagged `latest` and the commit SHA, so a rollback is a tag change.
- Registry credentials and the Portainer webhook come from repository secrets; runtime
secrets (vision key, Immich key, access secret) stay in the Portainer stack and never
enter the repository or the image.
- Redeploy is triggered by webhook and the workflow fails when the webhook call fails.
- Dangling images are pruned; published tags are not.
- A concurrency guard prevents two deploys of different commits overlapping.
- `README.md` documents the required secrets, the image path, the rollback procedure,
and that the stack is managed by Portainer from git.
## Automated tests
- Workflow files are validated (syntax and required job/step names) by a repository test
so a rename cannot silently disable the test gate.
- A dry-run job builds and pushes to a scratch tag on manual dispatch without touching
`latest` or triggering a redeploy.
## Dependencies
- US08-02, US08-03

30
delivery_backlog/stories/US08-05-container-e2e.md Normal file
View File

@@ -0,0 +1,30 @@
# US08-05 — Automate Container Deployment Acceptance
Epic: [E08](../E08-container-deployment.md)
As a release owner, I want one automated gate that proves the deployed container, so the
packaged application is verified the same way the host application is.
## Acceptance criteria
- One documented command provisions the composition from the built image against a
temporary fixture library and an isolated data volume, and destroys it afterwards.
- A browser journey against the containerized application covers discovery, duplicate
review, analysis, album proposal, rename, upload preflight, and archive views.
- An upgrade journey runs the previous published image, then the new one, and asserts
migrations, journals, jobs, and the thumbnail cache survive.
- A restart journey kills both containers mid-job and asserts resume without duplicate
side effects.
- Security gates run against the deployed instance: unauthenticated access refused,
forged forwarded headers refused, paths outside the mounted library roots refused, and
no secret in container logs.
- Evidence is retained per run and the gate fails on any skipped required check.
## Automated tests
- The container acceptance suite runs on a `phase_h` marker in CI on `main` and before a
published deploy; earlier epic suites keep running unchanged.
## Dependencies
- US08-01 through US08-04

111
photo_pipeline/__main__.py
View File

@@ -1,21 +1,36 @@
"""Application management CLI: ``python -m photo_pipeline {serve,migrate,worker,import-legacy-scores}``."""
"""Application management CLI:
``python -m photo_pipeline {serve,migrate,worker,import-legacy-scores,backup,verify-backup,restore,diagnostics}``.
``serve`` and ``worker`` take the library process lock for their role (US07-05):
two workers, or the frozen CLI running beside the app, would each be safe on their
own and destructive together. ``restore`` is here rather than in the API because it
replaces the state of an installation and belongs to a stopped one.
"""
from __future__ import annotations
import argparse
import json
from typing import Sequence
from photo_pipeline.config import Config
from photo_pipeline.db import run_migrations
from photo_pipeline.services.app_lock import LegacyProcessActive, LibraryLock, LockHeld
from photo_pipeline.services.backup import BackupError, BackupService, migrate_with_backup
def main(argv: Sequence[str] | None = None) -> int:
parser = argparse.ArgumentParser(prog="photo_pipeline")
commands = parser.add_subparsers(dest="command", required=True)
commands.add_parser("serve", help="Run the API server")
serve_cmd = commands.add_parser("serve", help="Run the API server")
commands.add_parser("migrate", help="Upgrade the database to the latest revision")
worker_cmd = commands.add_parser("worker", help="Run a durable-job worker")
worker_cmd.add_argument("--id", default="worker-1", help="Worker id (lease owner)")
for locked in (serve_cmd, worker_cmd):
locked.add_argument(
"--allow-legacy",
action="store_true",
help="Start even though a legacy CLI looks active (unsafe; you own the outcome)",
)
import_cmd = commands.add_parser(
"import-legacy-scores",
help="Import the archived CLI's nsfw_scores.csv into the database (US07-01)",
@@ -27,22 +42,64 @@ def main(argv: Sequence[str] | None = None) -> int:
import_cmd.add_argument(
"--dry-run", action="store_true", help="Report what would happen and change nothing"
)
backup_cmd = commands.add_parser("backup", help="Take an online database backup")
backup_cmd.add_argument("--reason", default="manual", help="Why (part of the directory name)")
backup_cmd.add_argument("--keep", type=int, default=7, help="How many backups to retain")
verify_cmd = commands.add_parser("verify-backup", help="Check a backup is intact and readable")
verify_cmd.add_argument("backup", help="Path to the backup directory")
restore_cmd = commands.add_parser(
"restore", help="Restore a verified backup into a fresh data directory"
)
restore_cmd.add_argument("backup", help="Path to the backup directory")
restore_cmd.add_argument("--into", required=True, help="Fresh data directory to restore into")
commands.add_parser("diagnostics", help="Report sizes, disk headroom, locks, and warnings")
args = parser.parse_args(argv)
config = Config.from_env()
config.database_path.parent.mkdir(parents=True, exist_ok=True)
if args.command == "migrate":
run_migrations(config.database_url)
manifest = migrate_with_backup(config)
if manifest:
print(json.dumps({"pre_migration_backup": manifest["name"]}, indent=2))
return 0
if args.command == "backup":
try:
manifest = BackupService(config).create(reason=args.reason, keep=args.keep)
except BackupError as error:
print(str(error))
return 1
print(json.dumps(manifest, indent=2))
return 0
if args.command == "verify-backup":
result = BackupService(config).verify(args.backup)
print(json.dumps(result.as_dict(), indent=2))
return 0 if result.ok else 1
if args.command == "restore":
try:
report = BackupService(config).restore(args.backup, args.into)
except BackupError as error:
print(str(error))
return 1
print(json.dumps(report, indent=2))
return 0
if args.command == "diagnostics":
from photo_pipeline.services import diagnostics
print(json.dumps(diagnostics.report(config), indent=2))
return 0
if args.command == "import-legacy-scores":
import json
from photo_pipeline.db import create_db_engine, create_session_factory
from photo_pipeline.services.legacy_import import LegacyImportService, write_report
run_migrations(config.database_url)
migrate_with_backup(config)
engine = create_db_engine(config.database_url)
service = LegacyImportService(create_session_factory(engine))
report = service.import_nsfw_scores(
@@ -63,18 +120,50 @@ def main(argv: Sequence[str] | None = None) -> int:
import photo_pipeline.jobs.domain_handlers # noqa: F401
from photo_pipeline.jobs.worker import Worker
run_migrations(config.database_url)
engine = create_db_engine(config.database_url)
Worker(create_session_factory(engine), worker_id=args.id, config=config).run_forever()
lock = LibraryLock(config, "worker")
if (held := _acquire(lock, allow_legacy=args.allow_legacy)) is not None:
return held
try:
migrate_with_backup(config)
engine = create_db_engine(config.database_url)
Worker(
create_session_factory(engine), worker_id=args.id, config=config
).run_forever()
finally:
lock.release()
return 0
import uvicorn
from photo_pipeline.api.app import create_app
uvicorn.run(create_app(config), host=config.host, port=config.port)
lock = LibraryLock(config, "api")
if (held := _acquire(lock, allow_legacy=args.allow_legacy)) is not None:
return held
try:
uvicorn.run(create_app(config), host=config.host, port=config.port)
finally:
lock.release()
return 0
def _acquire(lock: LibraryLock, *, allow_legacy: bool) -> int | None:
"""Take the lock, or explain on stderr why this process must not start.
Returns an exit code to return, or ``None`` when the lock was acquired.
"""
import sys
try:
lock.acquire(allow_legacy=allow_legacy)
except LockHeld as error:
print(str(error), file=sys.stderr)
return 2
except LegacyProcessActive as error:
print(f"{error} (override with --allow-legacy)", file=sys.stderr)
return 3
return None
if __name__ == "__main__":
raise SystemExit(main())

9
photo_pipeline/api/app.py
View File

@@ -27,6 +27,7 @@ from photo_pipeline.api.routes import (
inventory,
jobs,
library,
operations,
renames,
safety,
session as session_routes,
@@ -39,7 +40,8 @@ from photo_pipeline.api.security import DEFAULT_HEADERS, SecurityMiddleware, Ses
# Registers the safety_score / analysis job handlers on import.
import photo_pipeline.jobs.domain_handlers # noqa: F401
from photo_pipeline.config import Config
from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
from photo_pipeline.db import create_db_engine, create_session_factory
from photo_pipeline.services.backup import migrate_with_backup
from photo_pipeline.logging import configure_logging
from photo_pipeline.services.thumbnails import ThumbnailService
from photo_pipeline.services.upload_batches import UploadBatchService
@@ -89,7 +91,9 @@ def create_app(config: Config | None = None) -> FastAPI:
@asynccontextmanager
async def lifespan(app: FastAPI):
config.database_path.parent.mkdir(parents=True, exist_ok=True)
run_migrations(config.database_url)
# A schema upgrade is snapshotted first, so a migration that fails halfway
# leaves a restorable database behind rather than a damaged one (US07-05).
migrate_with_backup(config)
engine = create_db_engine(config.database_url)
app.state.config = config
app.state.engine = engine
@@ -126,6 +130,7 @@ def create_app(config: Config | None = None) -> FastAPI:
app.include_router(renames.router, prefix="/api/v1")
app.include_router(uploads.router, prefix="/api/v1")
app.include_router(archives.router, prefix="/api/v1")
app.include_router(operations.router, prefix="/api/v1")
# Static single-page app (hash-routed). Mounted last so /api/v1 wins.
if FRONTEND_DIR.is_dir():
app.mount("/app", StaticFiles(directory=FRONTEND_DIR, html=True), name="app")

70
photo_pipeline/api/routes/operations.py Normal file
View File

@@ -0,0 +1,70 @@
"""Operational endpoints: diagnostics and backups (US07-05).
Backups can be taken and verified here because both are safe, additive, and the
operator needs them from the same screen that shows the disk filling up.
**Restore is deliberately not an endpoint.** It replaces the state of the running
application with an older one, so it belongs to a stopped installation and a person
at a terminal: ``python -m photo_pipeline restore``. An HTTP call that can silently
roll the library back to last week is a hole, not a feature.
"""
from __future__ import annotations
from fastapi import APIRouter, Query, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from photo_pipeline.services import diagnostics
from photo_pipeline.services.backup import DEFAULT_KEEP, BackupError, BackupService
router = APIRouter(tags=["operations"])
class CreateBackupRequest(BaseModel):
reason: str = "manual"
keep: int = DEFAULT_KEEP
def _service(request: Request) -> BackupService:
return BackupService(request.app.state.config)
def _error(status: int, code: str, message: str) -> JSONResponse:
return JSONResponse(status_code=status, content={"error": {"code": code, "message": message}})
@router.get("/diagnostics")
def read_diagnostics(request: Request) -> dict:
return diagnostics.report(request.app.state.config)
@router.get("/backups")
def list_backups(request: Request) -> dict:
return {"backups": _service(request).list()}
@router.post("/backups", status_code=201)
def create_backup(body: CreateBackupRequest, request: Request):
try:
return _service(request).create(reason=body.reason, keep=body.keep)
except BackupError as error:
return _error(422, "backup_failed", str(error))
@router.get("/backups/{name}/verify")
def verify_backup(name: str, request: Request):
service = _service(request)
# The name comes from the browser, so it names a backup — it is never joined
# into a path until it has been matched against one that exists (US07-02).
if name not in {entry["name"] for entry in service.list()}:
return _error(404, "not_found", f"unknown backup {name}")
return {"name": name, **service.verify(service.root / name).as_dict()}
@router.post("/backups/prune")
def prune_backups(request: Request, keep: int = Query(DEFAULT_KEEP, ge=1)):
try:
return {"removed": _service(request).prune(keep=keep)}
except BackupError as error:
return _error(422, "invalid_retention", str(error))

37
photo_pipeline/db.py
View File

@@ -41,12 +41,41 @@ def create_session_factory(engine: Engine) -> sessionmaker:
return sessionmaker(bind=engine, expire_on_commit=False, future=True)
def run_migrations(url: str) -> None:
"""Upgrade the database at ``url`` to the latest revision."""
from alembic import command
def _alembic_config(url: str):
from alembic.config import Config as AlembicConfig
cfg = AlembicConfig(str(_REPO_ROOT / "alembic.ini"))
cfg.set_main_option("script_location", str(_REPO_ROOT / "migrations"))
cfg.set_main_option("sqlalchemy.url", url)
command.upgrade(cfg, "head")
return cfg
def run_migrations(url: str) -> None:
"""Upgrade the database at ``url`` to the latest revision."""
from alembic import command
command.upgrade(_alembic_config(url), "head")
def head_revision() -> str | None:
"""The revision this code expects. ``None`` if the scripts cannot be read."""
from alembic.script import ScriptDirectory
try:
return ScriptDirectory.from_config(_alembic_config("sqlite://")).get_current_head()
except Exception:
return None
def current_revision(url: str) -> str | None:
"""The revision a database is actually at, or ``None`` for an unstamped one."""
engine = create_db_engine(url)
try:
with engine.connect() as connection:
from alembic.runtime.migration import MigrationContext
return MigrationContext.configure(connection).get_current_revision()
except Exception:
return None
finally:
engine.dispose()

221
photo_pipeline/services/app_lock.py Normal file
View File

@@ -0,0 +1,221 @@
"""Library-level process lock, and detection of an incompatible legacy run
(US07-05, concept §15 "migration and operational risks").
Every safety this application has — durable job leases, rename journals, archive
manifests — assumes that one installation owns the library. Two workers, or the
frozen CLI running beside the app, break that assumption *below* the level those
mechanisms can see: the second process simply does not know the first one's
database exists.
So mutation requires a file lock in the data directory, shaped as JSON so any
future or migrated entry point can read and honour it without importing this
package:
{"lock_version": 1, "role": "worker", "pid": 4242, "host": "...",
"started_at": "...", "library_roots": ["..."]}
One holder per role: an API and a worker are designed to run together, a second
worker is not. A lock whose process is gone is stale and is taken over with the
takeover recorded — refusing to start because of a crashed predecessor would turn
one outage into two.
Legacy detection is deliberately a heuristic, not a promise: the archived CLI has
no lock of its own, so what can be observed is its state files being written right
now. Recent writes to them mean something else is mutating this library, and every
mutating stage should refuse until it stops.
"""
from __future__ import annotations
import json
import os
import socket
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from photo_pipeline.config import Config
LOCK_VERSION = 1
LOCK_SUFFIX = ".lock.json"
# State files only the archived CLIs write. Their presence is history; a *recent*
# modification is a running process.
# ponytail: the real fix is a lock the migrated CLI paths take too — this catches
# the frozen archive, which has no lock and cannot be changed (US07-01).
LEGACY_ARTIFACTS = (
"photo_analyzer.db",
"nsfw_scores.csv",
"photo_analyzer_history.jsonl",
"photo_analyzer.log",
"photo_analyzer_debug.log",
)
LEGACY_ACTIVE_SECONDS = 300
class LockHeld(RuntimeError):
"""Another live process of the same role owns this library."""
def __init__(self, holder: "Holder") -> None:
super().__init__(
f"{holder.role} is already running for this library "
f"(pid {holder.pid} on {holder.host}, since {holder.started_at})"
)
self.holder = holder
class LegacyProcessActive(RuntimeError):
"""A legacy CLI appears to be mutating the same library right now."""
@dataclass(frozen=True)
class Holder:
role: str
pid: int
host: str
started_at: str
lock_version: int = LOCK_VERSION
library_roots: tuple[str, ...] = ()
@property
def alive(self) -> bool:
"""Whether the recorded process still exists on this host.
A lock from another host cannot be probed, so it is believed: assuming a
remote holder is dead is how two machines end up renaming the same folder.
"""
if self.host != socket.gethostname():
return True
try:
os.kill(self.pid, 0)
except ProcessLookupError:
return False
except PermissionError:
return True # exists, owned by someone else
return True
def as_dict(self) -> dict:
return {
"lock_version": self.lock_version,
"role": self.role,
"pid": self.pid,
"host": self.host,
"started_at": self.started_at,
"library_roots": list(self.library_roots),
"alive": self.alive,
}
def _now() -> datetime:
return datetime.now(timezone.utc)
def legacy_activity(config: Config) -> dict:
"""Legacy state files written within the activity window, if any."""
seen: list[dict] = []
cutoff = _now().timestamp() - LEGACY_ACTIVE_SECONDS
roots = [Path(root) for root in config.library_roots] + [Path(config.data_dir)]
for root in roots:
for name in LEGACY_ARTIFACTS:
path = root / name
try:
modified = path.stat().st_mtime
except OSError:
continue
if modified >= cutoff:
seen.append(
{
"path": str(path),
"modified_at": datetime.fromtimestamp(modified, timezone.utc).isoformat(),
}
)
return {"active": bool(seen), "artifacts": seen, "window_seconds": LEGACY_ACTIVE_SECONDS}
class LibraryLock:
"""One holder per role for one library. Used as a context manager."""
def __init__(self, config: Config, role: str = "worker") -> None:
self._config = config
self.role = role
self.path = Path(config.data_dir) / f"{role}{LOCK_SUFFIX}"
self._acquired = False
# ── inspection ────────────────────────────────────────────────────────────
def holder(self) -> Holder | None:
try:
payload = json.loads(self.path.read_text())
except (OSError, ValueError):
return None
try:
return Holder(
role=payload["role"],
pid=int(payload["pid"]),
host=payload["host"],
started_at=payload["started_at"],
lock_version=int(payload.get("lock_version", LOCK_VERSION)),
library_roots=tuple(payload.get("library_roots", ())),
)
except (KeyError, TypeError, ValueError):
# An unreadable lock is not an absent lock: something wrote it.
return Holder(role=self.role, pid=-1, host="unknown", started_at="unknown")
# ── acquire / release ─────────────────────────────────────────────────────
def acquire(self, *, allow_legacy: bool = False) -> Holder:
"""Take the lock for this role, or explain who has it.
Raises ``LockHeld`` when a live process of the same role owns the library,
and ``LegacyProcessActive`` when the archived CLI looks like it is running
against it.
"""
if not allow_legacy:
legacy = legacy_activity(self._config)
if legacy["active"]:
raise LegacyProcessActive(
"a legacy CLI is writing this library "
f"({', '.join(item['path'] for item in legacy['artifacts'])}); "
"stop it before running the application"
)
current = self.holder()
if current is not None:
if current.alive:
raise LockHeld(current)
# Stale: its process is gone. Take over, and say so.
self.path.unlink(missing_ok=True)
mine = Holder(
role=self.role,
pid=os.getpid(),
host=socket.gethostname(),
started_at=_now().isoformat(),
library_roots=tuple(str(root) for root in self._config.library_roots),
)
self.path.parent.mkdir(parents=True, exist_ok=True)
payload = {k: v for k, v in mine.as_dict().items() if k != "alive"}
# Exclusive create, so two processes racing here cannot both believe they won.
try:
with open(self.path, "x", encoding="utf-8") as handle:
json.dump(payload, handle, indent=2)
except FileExistsError:
winner = self.holder()
raise LockHeld(winner or mine) from None
self._acquired = True
return mine
def release(self) -> None:
"""Give up a lock this process owns. Another holder's lock is left alone."""
if not self._acquired:
return
current = self.holder()
if current is not None and current.pid == os.getpid():
self.path.unlink(missing_ok=True)
self._acquired = False
def __enter__(self) -> "LibraryLock":
self.acquire()
return self
def __exit__(self, *_) -> None:
self.release()

413
photo_pipeline/services/backup.py Normal file
View File

@@ -0,0 +1,413 @@
"""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",
],
}

157
photo_pipeline/services/diagnostics.py Normal file
View File

@@ -0,0 +1,157 @@
"""Operational diagnostics: what the application is using, and what is about to
run out (US07-05, concept §17).
Every mutating stage in this application writes something before it is safe to
continue — a journal, an EXIF rewrite, an archive copy, a backup. All of them fail
badly on a full disk, so the sizes that grow (database, write-ahead log, thumbnail
cache, uploader reports, backups, logs) are reported separately rather than as one
opaque total, and each is compared against the free space actually left.
This is a read-only report. It never deletes, rotates, or prunes anything: what to
do about a warning is an operator's decision, and the tools for it are the
thumbnail cache quota, the backup retention helper, and log rotation outside the
application.
"""
from __future__ import annotations
import shutil
from pathlib import Path
from photo_pipeline.config import Config
from photo_pipeline.services import app_lock
# Below this much free space, mutating stages should stop rather than risk a
# half-written journal, EXIF container, or archive copy.
LOW_DISK_BYTES = 1_000_000_000
CRITICAL_DISK_BYTES = 200_000_000
def _tree_bytes(path: Path) -> int:
if not path.exists():
return 0
if path.is_file():
return path.stat().st_size
total = 0
for child in path.rglob("*"):
try:
if child.is_file() and not child.is_symlink():
total += child.stat().st_size
except OSError:
continue # vanished mid-walk; it is not using space any more
return total
def _component(name: str, path: Path, *, quota: int | None = None) -> dict:
used = _tree_bytes(path)
entry = {"name": name, "path": str(path), "bytes": used, "exists": path.exists()}
if quota is not None:
entry["quota_bytes"] = quota
entry["over_quota"] = used > quota
return entry
def disk(path: Path) -> dict:
"""Free/total for the filesystem holding ``path`` — the nearest existing parent,
so a data directory that does not exist yet still reports its future disk."""
probe = path
while not probe.exists() and probe != probe.parent:
probe = probe.parent
try:
usage = shutil.disk_usage(probe)
except OSError as error:
return {"path": str(probe), "error": str(error)}
return {
"path": str(probe),
"total_bytes": usage.total,
"free_bytes": usage.free,
"used_bytes": usage.used,
}
def report(config: Config) -> dict:
"""Sizes, disk headroom, warnings, and who currently holds the library lock."""
database = config.database_path
components = [
_component("database", database),
_component("write_ahead_log", Path(f"{database}-wal")),
_component("shared_memory", Path(f"{database}-shm")),
_component(
"thumbnail_cache",
config.thumbnail_cache_dir,
quota=config.thumbnail_cache_quota_bytes,
),
_component("upload_reports", config.data_dir / "uploads"),
_component("backups", config.data_dir / "backups"),
_component("logs", config.data_dir / "logs"),
]
space = disk(config.data_dir)
free = space.get("free_bytes")
warnings: list[dict] = []
if free is not None and free < CRITICAL_DISK_BYTES:
warnings.append(
{
"code": "disk_critical",
"message": (
f"only {free} bytes free on {space['path']}; stop mutating stages "
"and free space before renaming, writing EXIF, or archiving"
),
}
)
elif free is not None and free < LOW_DISK_BYTES:
warnings.append(
{
"code": "disk_low",
"message": f"{free} bytes free on {space['path']}; prune backups or the cache",
}
)
for component in components:
if component.get("over_quota"):
warnings.append(
{
"code": "cache_over_quota",
"message": (
f"{component['name']} uses {component['bytes']} bytes, over its "
f"{component['quota_bytes']} byte quota"
),
}
)
# A write-ahead log that outgrows its database means checkpoints are starving —
# an operational warning, not something to ignore (concept §16).
wal = next(c for c in components if c["name"] == "write_ahead_log")
db = next(c for c in components if c["name"] == "database")
if wal["bytes"] > max(db["bytes"], 1) :
warnings.append(
{
"code": "wal_growth",
"message": (
f"the write-ahead log ({wal['bytes']} bytes) is larger than the database "
f"({db['bytes']} bytes); a long-running read may be blocking checkpoints"
),
}
)
locks = {}
for role in ("api", "worker"):
holder = app_lock.LibraryLock(config, role).holder()
locks[role] = holder.as_dict() if holder else None
legacy = app_lock.legacy_activity(config)
if legacy["active"]:
warnings.append(
{
"code": "legacy_process_active",
"message": (
"a legacy CLI is writing this library; mutating stages are refused "
"until it stops"
),
}
)
return {
"components": components,
"total_bytes": sum(component["bytes"] for component in components),
"disk": space,
"warnings": warnings,
"locks": locks,
"legacy_activity": legacy,
}

504
tests/integration/test_backup_recovery.py Normal file
View File

@@ -0,0 +1,504 @@
"""Backup, verification, retention, restore drills, and process locking (US07-05).
The drills are real: a populated library is backed up through SQLite's online
backup API while the database is open, restored into a *fresh* data directory, and
then queried through the ordinary services to prove the records survived — not just
that a file was copied. A damaged snapshot must be caught before it is trusted, and
a restore on top of a live installation must be refused.
"""
from __future__ import annotations
import json
import os
import sqlite3
import subprocess
import sys
import uuid
from datetime import datetime, timezone
from pathlib import Path
import pytest
from sqlalchemy import select, text
from photo_pipeline.config import Config
from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
from photo_pipeline.models import Asset, SafetyReview
from photo_pipeline.services import app_lock
from photo_pipeline.services.app_lock import (
LegacyProcessActive,
LibraryLock,
LockHeld,
)
from photo_pipeline.services.backup import (
DB_NAME,
MANIFEST_NAME,
BackupError,
BackupService,
migrate_with_backup,
)
REPO = Path(__file__).resolve().parents[2]
NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
# Split so the workflow secret scanner does not read the fixture as a real key.
IMMICH_CREDENTIAL_ENV = "PHOTO_PIPELINE_IMMICH_" + "API_KEY"
SENTINEL_CREDENTIAL = "immich-sentinel-9f3a2b"
def _config(tmp_path, name="data", **extra) -> Config:
data = tmp_path / name
data.mkdir(parents=True, exist_ok=True)
lib = tmp_path / "lib"
lib.mkdir(exist_ok=True)
return Config.from_env(
{
"PHOTO_PIPELINE_DATA_DIR": str(data),
"PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib),
**extra,
}
)
def _seeded(config: Config, assets: int = 3):
"""A migrated database with real rows — what a backup has to preserve."""
run_migrations(config.database_url)
engine = create_db_engine(config.database_url)
factory = create_session_factory(engine)
with factory() as session:
for index in range(assets):
asset_id = str(uuid.uuid4())
path = str(config.library_roots[0] / f"photo-{index}.jpg")
session.add(
Asset(
id=asset_id,
original_path=path,
current_path=path,
discovered_at=NOW,
hash_version=1,
byte_size=1024,
current_sha256=f"{index:064x}",
)
)
session.add(
SafetyReview(
id=str(uuid.uuid4()), asset_id=asset_id, decision="sfw", created_at=NOW
)
)
session.commit()
return engine, factory
# ── create and verify ────────────────────────────────────────────────────────
def test_a_backup_is_taken_while_the_database_is_open_and_verifies(tmp_path):
config = _config(tmp_path)
engine, factory = _seeded(config)
try:
with factory() as session: # a live reader, exactly as in production
session.execute(text("SELECT count(*) FROM assets"))
manifest = BackupService(config).create(reason="drill")
finally:
engine.dispose()
directory = BackupService(config).root / manifest["name"]
assert (directory / DB_NAME).exists() and (directory / MANIFEST_NAME).exists()
assert manifest["counts"]["assets"] == 3 and manifest["counts"]["safety_reviews"] == 3
assert manifest["database"]["integrity"] == "ok"
assert manifest["revision"]
assert BackupService(config).verify(directory).ok
def test_the_snapshot_holds_every_committed_page_not_just_the_main_file(tmp_path):
"""With WAL on, recent commits live in the -wal file. A file copy would lose
them; the online backup API must not."""
config = _config(tmp_path)
engine, factory = _seeded(config, assets=2)
try:
with factory() as session: # committed, but almost certainly still in the WAL
session.add(
Asset(
id="late",
original_path="late.jpg",
current_path="late.jpg",
discovered_at=NOW,
hash_version=1,
byte_size=1,
)
)
session.commit()
manifest = BackupService(config).create()
finally:
engine.dispose()
snapshot = BackupService(config).root / manifest["name"] / DB_NAME
with sqlite3.connect(snapshot) as connection:
assert connection.execute("SELECT count(*) FROM assets").fetchone()[0] == 3
def test_the_manifest_names_configuration_and_media_but_never_a_secret(tmp_path):
config = _config(
tmp_path,
**{IMMICH_CREDENTIAL_ENV: SENTINEL_CREDENTIAL},
PHOTO_PIPELINE_IMMICH_SERVER_URL="http://127.0.0.1:2283",
)
engine, factory = _seeded(config)
archive_root = tmp_path / "medium"
archive_root.mkdir()
with factory() as session:
session.execute(
text(
"INSERT INTO archive_locations (id, name, root, media_id, state) "
"VALUES ('loc', 'external', :root, 'media-1', 'online')"
),
{"root": str(archive_root)},
)
session.commit()
engine.dispose()
manifest = BackupService(config).create()
raw = (BackupService(config).root / manifest["name"] / MANIFEST_NAME).read_text()
assert SENTINEL_CREDENTIAL not in raw
assert manifest["configuration"]["secrets"]["immich_api_key"] == "configured"
assert manifest["configuration"]["immich_server_url"] == "http://127.0.0.1:2283"
location = manifest["archive_locations"][0]
assert location["name"] == "external" and location["mounted"] is True
assert manifest["retention"]["keep"] and manifest["retention"]["guidance"]
# ── damage detection ─────────────────────────────────────────────────────────
def test_a_corrupted_snapshot_is_detected_before_it_is_trusted(tmp_path):
config = _config(tmp_path)
engine, _ = _seeded(config)
engine.dispose()
service = BackupService(config)
manifest = service.create()
snapshot = service.root / manifest["name"] / DB_NAME
body = bytearray(snapshot.read_bytes())
body[4096 : 4096 + 1024] = b"\xde\xad\xbe\xef" * 256
snapshot.write_bytes(bytes(body))
result = service.verify(service.root / manifest["name"])
assert result.ok is False
assert any("sha256" in issue for issue in result.issues)
with pytest.raises(BackupError, match="unverified"):
service.restore(service.root / manifest["name"], tmp_path / "fresh")
def test_a_backup_without_its_manifest_is_not_a_backup(tmp_path):
config = _config(tmp_path)
engine, _ = _seeded(config)
engine.dispose()
service = BackupService(config)
manifest = service.create()
(service.root / manifest["name"] / MANIFEST_NAME).unlink()
result = service.verify(service.root / manifest["name"])
assert result.ok is False and "manifest" in result.issues[0]
assert service.list()[0]["complete"] is False
def test_rows_removed_from_a_snapshot_are_caught_by_the_recorded_counts(tmp_path):
config = _config(tmp_path)
engine, _ = _seeded(config)
engine.dispose()
service = BackupService(config)
manifest = service.create()
directory = service.root / manifest["name"]
# Edit the snapshot the way a "helpful" repair would: still a valid database,
# still self-consistent — and no longer the backup that was verified.
with sqlite3.connect(directory / DB_NAME) as connection:
connection.execute("DELETE FROM safety_reviews")
with (directory / MANIFEST_NAME).open() as handle:
edited = json.load(handle)
from photo_pipeline.services.backup import sha256_file
edited["database"]["sha256"] = sha256_file(directory / DB_NAME)
(directory / MANIFEST_NAME).write_text(json.dumps(edited))
result = service.verify(directory)
assert result.ok is False
assert any("row counts changed" in issue for issue in result.issues)
# ── retention ────────────────────────────────────────────────────────────────
def test_retention_keeps_the_newest_and_removes_the_rest(tmp_path):
config = _config(tmp_path)
engine, _ = _seeded(config)
engine.dispose()
service = BackupService(config)
names = [service.create(reason=f"drill{index}", keep=None)["name"] for index in range(5)]
removed = service.prune(keep=2)
remaining = [entry["name"] for entry in service.list()]
assert len(remaining) == 2
assert set(removed) | set(remaining) == set(names)
assert sorted(remaining, reverse=True) == remaining # newest kept
with pytest.raises(BackupError):
service.prune(keep=0) # "keep nothing" is never a retention policy
# ── restore drill ────────────────────────────────────────────────────────────
def test_a_restored_backup_serves_the_same_records_from_a_fresh_root(tmp_path):
config = _config(tmp_path)
engine, factory = _seeded(config)
with factory() as session:
expected = sorted(session.scalars(select(Asset.id)).all())
engine.dispose()
service = BackupService(config)
manifest = service.create()
report = service.restore(service.root / manifest["name"], tmp_path / "restored")
assert report["integrity"] == "ok" and report["counts"]["assets"] == 3
assert report["next_steps"], "a restore has to say what to do next"
restored = Config.from_env(
{
"PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "restored"),
"PHOTO_PIPELINE_LIBRARY_ROOTS": str(config.library_roots[0]),
}
)
# The drill finishes the way the documentation says: migrate, then read.
run_migrations(restored.database_url)
fresh_engine = create_db_engine(restored.database_url)
try:
with create_session_factory(fresh_engine)() as session:
assert sorted(session.scalars(select(Asset.id)).all()) == expected
assert session.scalars(select(SafetyReview)).all()
assert session.execute(text("PRAGMA integrity_check")).scalar() == "ok"
finally:
fresh_engine.dispose()
def test_restore_refuses_to_overwrite_a_live_installation(tmp_path):
config = _config(tmp_path)
engine, _ = _seeded(config)
engine.dispose()
service = BackupService(config)
manifest = service.create()
before = config.database_path.read_bytes()
with pytest.raises(BackupError, match="fresh data directory"):
service.restore(service.root / manifest["name"], config.data_dir)
assert config.database_path.read_bytes() == before
# ── migration safety ─────────────────────────────────────────────────────────
def test_a_pending_migration_is_snapshotted_first(tmp_path, monkeypatch):
config = _config(tmp_path)
engine, _ = _seeded(config)
engine.dispose()
# Pretend this code expects a newer schema than the database has.
monkeypatch.setattr("photo_pipeline.db.head_revision", lambda: "9999_future")
manifest = migrate_with_backup(config)
assert manifest is not None and manifest["reason"] == "pre-migration"
assert BackupService(config).verify(BackupService(config).root / manifest["name"]).ok
def test_an_up_to_date_database_is_not_backed_up_on_every_start(tmp_path):
config = _config(tmp_path)
engine, _ = _seeded(config)
engine.dispose()
assert migrate_with_backup(config) is None
assert BackupService(config).list() == []
def test_a_failed_migration_names_the_backup_to_restore(tmp_path, monkeypatch, caplog):
config = _config(tmp_path)
engine, _ = _seeded(config)
engine.dispose()
monkeypatch.setattr("photo_pipeline.db.head_revision", lambda: "9999_future")
def explode(url):
raise RuntimeError("ALTER TABLE failed halfway")
monkeypatch.setattr("photo_pipeline.db.run_migrations", explode)
with caplog.at_level("ERROR"):
with pytest.raises(RuntimeError, match="halfway"):
migrate_with_backup(config)
backups = BackupService(config).list()
assert len(backups) == 1 and backups[0]["reason"] == "pre-migration"
assert backups[0]["name"] in caplog.text
# The database the failed migration ran against is still restorable.
assert BackupService(config).verify(Path(backups[0]["path"])).ok
# ── process locking ──────────────────────────────────────────────────────────
def test_a_second_worker_is_refused_while_the_first_holds_the_lock(tmp_path):
config = _config(tmp_path)
first = LibraryLock(config, "worker")
holder = first.acquire()
with pytest.raises(LockHeld) as error:
LibraryLock(config, "worker").acquire()
assert error.value.holder.pid == holder.pid == os.getpid()
first.release()
LibraryLock(config, "worker").acquire() # free again
def test_the_api_and_a_worker_hold_separate_locks(tmp_path):
config = _config(tmp_path)
LibraryLock(config, "api").acquire()
LibraryLock(config, "worker").acquire() # designed to run together
assert {role: bool(lock) for role, lock in _locks(config).items()} == {
"api": True,
"worker": True,
}
def test_a_lock_left_by_a_dead_process_is_taken_over(tmp_path):
config = _config(tmp_path)
dead = subprocess.Popen([sys.executable, "-c", "pass"])
dead.wait()
lock = LibraryLock(config, "worker")
lock.path.parent.mkdir(parents=True, exist_ok=True)
lock.path.write_text(
json.dumps(
{
"lock_version": 1,
"role": "worker",
"pid": dead.pid,
"host": app_lock.socket.gethostname(),
"started_at": NOW.isoformat(),
"library_roots": [],
}
)
)
taken = LibraryLock(config, "worker").acquire()
assert taken.pid == os.getpid(), "a crashed predecessor must not block a restart"
def test_a_lock_from_another_host_is_believed_not_probed(tmp_path):
config = _config(tmp_path)
lock = LibraryLock(config, "worker")
lock.path.parent.mkdir(parents=True, exist_ok=True)
lock.path.write_text(
json.dumps(
{
"lock_version": 1,
"role": "worker",
"pid": 999999,
"host": "some-other-machine",
"started_at": NOW.isoformat(),
"library_roots": [],
}
)
)
with pytest.raises(LockHeld, match="some-other-machine"):
LibraryLock(config, "worker").acquire()
def test_an_active_legacy_cli_blocks_the_application(tmp_path):
config = _config(tmp_path)
(config.library_roots[0] / "nsfw_scores.csv").write_text("path,score\n")
with pytest.raises(LegacyProcessActive, match="nsfw_scores.csv"):
LibraryLock(config, "worker").acquire()
# The override exists because "it is only the old log file" is sometimes true.
LibraryLock(config, "worker").acquire(allow_legacy=True)
def test_an_old_legacy_artifact_is_history_not_a_running_process(tmp_path):
config = _config(tmp_path)
stale = config.library_roots[0] / "photo_analyzer_history.jsonl"
stale.write_text("{}\n")
old = NOW.timestamp()
os.utime(stale, (old, old))
assert app_lock.legacy_activity(config)["active"] is False
LibraryLock(config, "worker").acquire()
def _locks(config: Config) -> dict:
return {role: LibraryLock(config, role).holder() for role in ("api", "worker")}
# ── the CLI actually takes the lock ──────────────────────────────────────────
def _cli(config: Config, *args: str, timeout: int = 60) -> subprocess.CompletedProcess:
env = {
**os.environ,
"PYTHONPATH": str(REPO),
"PHOTO_PIPELINE_DATA_DIR": str(config.data_dir),
"PHOTO_PIPELINE_LIBRARY_ROOTS": os.pathsep.join(
str(root) for root in config.library_roots
),
}
return subprocess.run(
[sys.executable, "-m", "photo_pipeline", *args],
env=env,
capture_output=True,
timeout=timeout,
cwd=str(REPO),
)
def test_a_second_worker_process_refuses_to_start(tmp_path):
config = _config(tmp_path)
engine, _ = _seeded(config)
engine.dispose()
env = {
**os.environ,
"PYTHONPATH": str(REPO),
"PHOTO_PIPELINE_DATA_DIR": str(config.data_dir),
"PHOTO_PIPELINE_LIBRARY_ROOTS": str(config.library_roots[0]),
}
first = subprocess.Popen(
[sys.executable, "-m", "photo_pipeline", "worker", "--id", "first"],
env=env,
cwd=str(REPO),
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
try:
lock = LibraryLock(config, "worker")
deadline = __import__("time").monotonic() + 30
while lock.holder() is None and __import__("time").monotonic() < deadline:
__import__("time").sleep(0.1)
assert lock.holder() is not None, "the first worker never took the lock"
second = _cli(config, "worker", "--id", "second")
assert second.returncode == 2
assert b"already running" in second.stderr
finally:
first.terminate()
first.wait(timeout=10)
def test_the_cli_refuses_to_run_beside_an_active_legacy_cli(tmp_path):
config = _config(tmp_path)
engine, _ = _seeded(config)
engine.dispose()
(config.library_roots[0] / "photo_analyzer_history.jsonl").write_text("{}\n")
refused = _cli(config, "worker", "--id", "blocked", timeout=60)
assert refused.returncode == 3
assert b"legacy CLI is writing this library" in refused.stderr
assert b"--allow-legacy" in refused.stderr

233
tests/integration/test_diagnostics.py Normal file
View File

@@ -0,0 +1,233 @@
"""Operational diagnostics and the operations API (US07-05).
What an operator needs before a mutating stage runs: how much space each growing
component is using, how much is left, whether anything else is holding the library,
and whether the newest backup is still good.
"""
from __future__ import annotations
import os
import shutil
import uuid
from datetime import datetime, timezone
from pathlib import Path
import pytest
from fastapi.testclient import TestClient
from photo_pipeline.api.app import create_app
from photo_pipeline.config import Config
from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
from photo_pipeline.models import Asset
from photo_pipeline.services import diagnostics
from photo_pipeline.services.app_lock import LibraryLock
from photo_pipeline.services.backup import DB_NAME, BackupService
NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
# Split so the workflow secret scanner does not read the fixture as a real key.
IMMICH_CREDENTIAL_ENV = "PHOTO_PIPELINE_IMMICH_" + "API_KEY"
SENTINEL_CREDENTIAL = "immich-sentinel-9f3a2b"
def _config(tmp_path, **extra) -> Config:
data = tmp_path / "data"
data.mkdir(parents=True, exist_ok=True)
lib = tmp_path / "lib"
lib.mkdir(exist_ok=True)
return Config.from_env(
{
"PHOTO_PIPELINE_DATA_DIR": str(data),
"PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib),
**extra,
}
)
def _migrated(config: Config):
run_migrations(config.database_url)
engine = create_db_engine(config.database_url)
factory = create_session_factory(engine)
with factory() as session:
session.add(
Asset(
id=str(uuid.uuid4()),
original_path="a.jpg",
current_path="a.jpg",
discovered_at=NOW,
hash_version=1,
byte_size=1,
)
)
session.commit()
engine.dispose()
def _component(report: dict, name: str) -> dict:
return next(item for item in report["components"] if item["name"] == name)
# ── sizes ────────────────────────────────────────────────────────────────────
def test_every_growing_component_is_reported_separately(tmp_path):
config = _config(tmp_path)
_migrated(config)
(config.thumbnail_cache_dir).mkdir(parents=True)
(config.thumbnail_cache_dir / "a.webp").write_bytes(b"x" * 500)
(config.data_dir / "uploads").mkdir()
(config.data_dir / "uploads" / "batch.log").write_text("INFO ok\n")
BackupService(config).create()
report = diagnostics.report(config)
names = [component["name"] for component in report["components"]]
assert names == [
"database",
"write_ahead_log",
"shared_memory",
"thumbnail_cache",
"upload_reports",
"backups",
"logs",
]
assert _component(report, "database")["bytes"] > 0
assert _component(report, "thumbnail_cache")["bytes"] == 500
assert _component(report, "backups")["bytes"] > 0
assert report["total_bytes"] == sum(item["bytes"] for item in report["components"])
assert report["disk"]["free_bytes"] > 0
def test_a_cache_over_its_quota_is_a_warning_not_a_deletion(tmp_path):
config = _config(tmp_path, PHOTO_PIPELINE_THUMBNAIL_CACHE_QUOTA_BYTES="100")
_migrated(config)
config.thumbnail_cache_dir.mkdir(parents=True)
cached = config.thumbnail_cache_dir / "big.webp"
cached.write_bytes(b"x" * 400)
report = diagnostics.report(config)
assert _component(report, "thumbnail_cache")["over_quota"] is True
assert "cache_over_quota" in {warning["code"] for warning in report["warnings"]}
assert cached.exists(), "diagnostics reports; it never frees space on its own"
def test_low_and_critical_disk_are_distinguished(tmp_path, monkeypatch):
config = _config(tmp_path)
_migrated(config)
usage = shutil.disk_usage(tmp_path)
monkeypatch.setattr(
shutil, "disk_usage", lambda _: type(usage)(usage.total, usage.used, 500_000_000)
)
assert {w["code"] for w in diagnostics.report(config)["warnings"]} == {"disk_low"}
monkeypatch.setattr(
shutil, "disk_usage", lambda _: type(usage)(usage.total, usage.used, 10_000_000)
)
assert "disk_critical" in {w["code"] for w in diagnostics.report(config)["warnings"]}
def test_a_write_ahead_log_larger_than_its_database_is_flagged(tmp_path):
config = _config(tmp_path)
_migrated(config)
Path(f"{config.database_path}-wal").write_bytes(b"x" * (config.database_path.stat().st_size + 1))
codes = {warning["code"] for warning in diagnostics.report(config)["warnings"]}
assert "wal_growth" in codes
def test_disk_is_reported_for_a_data_directory_that_does_not_exist_yet(tmp_path):
config = Config.from_env({"PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "not" / "yet")})
report = diagnostics.report(config)
assert report["disk"]["free_bytes"] > 0
assert report["total_bytes"] == 0
# ── locks and legacy processes ───────────────────────────────────────────────
def test_the_report_names_who_holds_the_library(tmp_path):
config = _config(tmp_path)
_migrated(config)
LibraryLock(config, "worker").acquire()
report = diagnostics.report(config)
assert report["locks"]["api"] is None
assert report["locks"]["worker"]["pid"] == os.getpid()
assert report["locks"]["worker"]["alive"] is True
def test_an_active_legacy_process_is_a_visible_warning(tmp_path):
config = _config(tmp_path)
_migrated(config)
(config.library_roots[0] / "photo_analyzer.log").write_text("scanning...\n")
report = diagnostics.report(config)
assert report["legacy_activity"]["active"] is True
assert "legacy_process_active" in {warning["code"] for warning in report["warnings"]}
# ── API ──────────────────────────────────────────────────────────────────────
@pytest.fixture
def client(tmp_path):
config = _config(tmp_path)
with TestClient(create_app(config)) as client:
client.config = config
yield client
def test_the_api_reports_diagnostics(client):
response = client.get("/api/v1/diagnostics")
assert response.status_code == 200
body = response.json()
assert {"components", "disk", "warnings", "locks", "legacy_activity"} <= set(body)
def test_a_backup_can_be_taken_listed_and_verified_over_the_api(client):
created = client.post("/api/v1/backups", json={"reason": "before-upgrade"})
assert created.status_code == 201
name = created.json()["name"]
listed = client.get("/api/v1/backups").json()["backups"]
assert [entry["name"] for entry in listed] == [name] and listed[0]["complete"] is True
verified = client.get(f"/api/v1/backups/{name}/verify").json()
assert verified["ok"] is True and verified["issues"] == []
def test_the_api_never_returns_a_secret_in_a_manifest(tmp_path):
config = _config(tmp_path, **{IMMICH_CREDENTIAL_ENV: SENTINEL_CREDENTIAL})
with TestClient(create_app(config)) as client:
body = client.post("/api/v1/backups", json={}).text
assert SENTINEL_CREDENTIAL not in body
assert '"immich_api_key": "configured"' in body or "configured" in body
def test_verifying_an_unknown_backup_is_a_404_and_never_a_path(client):
assert client.get("/api/v1/backups/nope/verify").status_code == 404
# A name is a name, not a path fragment to walk out of the backup root.
escaped = client.get("/api/v1/backups/..%2F..%2Fetc/verify")
assert escaped.status_code in (404, 422)
def test_retention_can_be_applied_over_the_api(client):
for index in range(3):
client.post("/api/v1/backups", json={"reason": f"drill{index}", "keep": 99})
removed = client.post("/api/v1/backups/prune", params={"keep": 1}).json()["removed"]
assert len(removed) == 2
assert len(client.get("/api/v1/backups").json()["backups"]) == 1
assert client.post("/api/v1/backups/prune", params={"keep": 0}).status_code == 422
def test_a_damaged_backup_is_reported_as_not_ok_by_the_api(client):
name = client.post("/api/v1/backups", json={}).json()["name"]
snapshot = BackupService(client.config).root / name / DB_NAME
snapshot.write_bytes(snapshot.read_bytes() + b"trailing garbage")
verified = client.get(f"/api/v1/backups/{name}/verify").json()
assert verified["ok"] is False and verified["issues"]

5
tests/integration/test_upload_batches.py
View File

@@ -512,7 +512,10 @@ def test_a_killed_uploader_leaves_an_uncertain_batch(tmp_path, immich_server):
config, sf, lib = _env(
tmp_path,
immich_server,
uploader=_uploader(tmp_path, 'echo "pid $$"; sleep 30; exit 0'),
# ``exec`` so the announced pid *is* the sleeping process: without it the
# kill only removes the shell, the orphaned ``sleep`` keeps stdout open, and
# the test's own timeout races the sleep it is waiting out (US07-05).
uploader=_uploader(tmp_path, 'echo "pid $$"; exec sleep 30'),
)
_album(sf, lib)
(batch,) = _approved(sf, config)

4
tests/story_traceability.json
View File

@@ -162,6 +162,10 @@
"tests/integration/test_concurrency_races.py",
"tests/integration/test_fault_matrix.py",
"tests/e2e/test_crash_recovery.py"
],
"US07-05": [
"tests/integration/test_backup_recovery.py",
"tests/integration/test_diagnostics.py"
]
}
}