From e4ae5da4404a74c789bd0e53ff95df44998a624d Mon Sep 17 00:00:00 2001 From: domverse Date: Sun, 16 Aug 2026 13:06:37 +0200 Subject: [PATCH] US05-02: Orchestrate Album Upload Batches (#72) --- migrations/versions/0008_upload_batches.py | 91 ++++ photo_pipeline/__main__.py | 2 +- photo_pipeline/api/app.py | 4 + photo_pipeline/api/routes/uploads.py | 86 ++- photo_pipeline/integrations/immich_go.py | 119 +++++ photo_pipeline/jobs/domain_handlers.py | 30 +- photo_pipeline/jobs/handlers.py | 3 + photo_pipeline/jobs/worker.py | 6 +- photo_pipeline/models/__init__.py | 3 + photo_pipeline/models/uploads.py | 90 ++++ photo_pipeline/services/hashing.py | 11 +- photo_pipeline/services/upload_batches.py | 424 +++++++++++++++ tests/integration/test_upload_batches.py | 588 +++++++++++++++++++++ tests/story_traceability.json | 3 + 14 files changed, 1446 insertions(+), 14 deletions(-) create mode 100644 migrations/versions/0008_upload_batches.py create mode 100644 photo_pipeline/models/uploads.py create mode 100644 photo_pipeline/services/upload_batches.py create mode 100644 tests/integration/test_upload_batches.py diff --git a/migrations/versions/0008_upload_batches.py b/migrations/versions/0008_upload_batches.py new file mode 100644 index 0000000..ef7d4c9 --- /dev/null +++ b/migrations/versions/0008_upload_batches.py @@ -0,0 +1,91 @@ +"""Upload batches and their items (US05-02). + +Revision ID: 0008_upload_batches +Revises: 0007_rename_plans +Create Date: 2026-08-16 + +One row per approved album folder handed to immich-go, plus the per-asset +pre-upload hashes that later stories verify the result against. +""" + +import sqlalchemy as sa +from alembic import op + +revision = "0008_upload_batches" +down_revision = "0007_rename_plans" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "upload_batches", + sa.Column("id", sa.String(), primary_key=True), + sa.Column("album", sa.String(), nullable=False), + sa.Column("folder", sa.String(), nullable=False), + sa.Column("album_name", sa.String(), nullable=False), + # planned | running | cancelling | cancelled | succeeded | failed + # | unknown_requires_verification + sa.Column("state", sa.String(), nullable=False, server_default="planned"), + # Preflight token this batch was approved against; re-checked before every + # attempt, so changed bytes or decisions cannot be uploaded silently. + sa.Column("preflight_token", sa.String(), nullable=False), + sa.Column("allow_partial", sa.Boolean(), nullable=False, server_default="0"), + sa.Column("command", sa.String(), nullable=True), # JSON array, redacted + sa.Column("uploader_version", sa.String(), nullable=True), + sa.Column("asset_count", sa.Integer(), nullable=False, server_default="0"), + sa.Column("attempt_count", sa.Integer(), nullable=False, server_default="0"), + # Bumped on every claim and used as the fencing token. + sa.Column("version", sa.Integer(), nullable=False, server_default="1"), + sa.Column("worker_id", sa.String(), nullable=True), + sa.Column("report_path", sa.String(), nullable=True), + sa.Column("report_bytes", sa.Integer(), nullable=True), + sa.Column("report_truncated", sa.Boolean(), nullable=False, server_default="0"), + sa.Column("exit_code", sa.Integer(), nullable=True), + sa.Column("error_code", sa.String(), nullable=True), + sa.Column("error_message", sa.String(), nullable=True), + sa.Column("started_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + nullable=False, + server_default=sa.text("CURRENT_TIMESTAMP"), + ), + sa.Column( + "updated_at", + sa.DateTime(timezone=True), + nullable=False, + server_default=sa.text("CURRENT_TIMESTAMP"), + ), + ) + op.create_index("ix_upload_batches_album", "upload_batches", ["album"]) + op.create_index("ix_upload_batches_state", "upload_batches", ["state"]) + + op.create_table( + "upload_items", + sa.Column( + "batch_id", + sa.String(), + sa.ForeignKey("upload_batches.id", ondelete="CASCADE"), + primary_key=True, + ), + sa.Column("asset_id", sa.String(), primary_key=True), + sa.Column("path", sa.String(), nullable=False), + # The bytes as they were when the batch was created; SHA-1 is what Immich + # uses to recognise a file it already holds. + sa.Column("sha256", sa.String(), nullable=True), + sa.Column("sha1", sa.String(), nullable=True), + sa.Column("state", sa.String(), nullable=False, server_default="pending"), + sa.Column( + "updated_at", + sa.DateTime(timezone=True), + nullable=False, + server_default=sa.text("CURRENT_TIMESTAMP"), + ), + ) + + +def downgrade() -> None: + op.drop_table("upload_items") + op.drop_table("upload_batches") diff --git a/photo_pipeline/__main__.py b/photo_pipeline/__main__.py index db68a53..3165505 100644 --- a/photo_pipeline/__main__.py +++ b/photo_pipeline/__main__.py @@ -36,7 +36,7 @@ def main(argv: Sequence[str] | None = None) -> int: run_migrations(config.database_url) engine = create_db_engine(config.database_url) - Worker(create_session_factory(engine), worker_id=args.id).run_forever() + Worker(create_session_factory(engine), worker_id=args.id, config=config).run_forever() return 0 import uvicorn diff --git a/photo_pipeline/api/app.py b/photo_pipeline/api/app.py index 90d711e..ccac17f 100644 --- a/photo_pipeline/api/app.py +++ b/photo_pipeline/api/app.py @@ -34,6 +34,7 @@ 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.logging import configure_logging +from photo_pipeline.services.upload_batches import UploadBatchService FRONTEND_DIR = Path(__file__).resolve().parents[2] / "frontend" @@ -50,6 +51,9 @@ def create_app(config: Config | None = None) -> FastAPI: app.state.config = config app.state.engine = engine app.state.session_factory = create_session_factory(engine) + # An upload whose process died left no outcome behind; resolve it now so the + # uploader lane is free and the uncertain batch is visible (US05-02). + UploadBatchService(app.state.session_factory, config=config).recover() try: yield finally: diff --git a/photo_pipeline/api/routes/uploads.py b/photo_pipeline/api/routes/uploads.py index 8290295..8803ce9 100644 --- a/photo_pipeline/api/routes/uploads.py +++ b/photo_pipeline/api/routes/uploads.py @@ -1,8 +1,9 @@ -"""Upload preflight API (US05-01). +"""Upload preflight and batch API (US05-01, US05-02). Preflight is a command, not a resource read: it contacts the Immich server, hashes -the current bytes, and issues a token. It uploads nothing — starting a batch is -US05-02, so there is deliberately no start endpoint here. +the current bytes, and issues a token. Creating a batch requires that token, and +starting one enqueues a durable job on the single ``upload`` lane — the API never +runs the uploader in the request thread. """ from __future__ import annotations @@ -11,6 +12,13 @@ from fastapi import APIRouter, Request from fastapi.responses import JSONResponse from pydantic import BaseModel +from photo_pipeline.jobs.domain_handlers import UPLOAD_BATCH, UPLOAD_LOCK +from photo_pipeline.services.jobs import JobBlocked, JobService +from photo_pipeline.services.upload_batches import ( + BatchConflict, + BatchError, + UploadBatchService, +) from photo_pipeline.services.uploads import UploadError, UploadService router = APIRouter(tags=["uploads"]) @@ -23,17 +31,83 @@ class PreflightRequest(BaseModel): allow_partial: bool = False +class CreateBatchRequest(PreflightRequest): + # The token of the preflight the user approved; a stale one is refused. + token: str + + def _service(request: Request) -> UploadService: return UploadService(request.app.state.session_factory, config=request.app.state.config) +def _batches(request: Request) -> UploadBatchService: + return UploadBatchService(request.app.state.session_factory, config=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.post("/upload-preflight") def preflight(request: Request, body: PreflightRequest | None = None): body = body or PreflightRequest() try: return _service(request).preflight(body.albums, allow_partial=body.allow_partial) except UploadError as error: - return JSONResponse( - status_code=422, - content={"error": {"code": "unknown_album", "message": str(error)}}, + return _error(422, "unknown_album", str(error)) + + +@router.post("/upload-batches", status_code=201) +def create_batches(body: CreateBatchRequest, request: Request): + """Turn an approved preflight into one durable batch per album.""" + try: + return {"batches": _batches(request).create( + body.albums, token=body.token, allow_partial=body.allow_partial + )} + except UploadError as error: + return _error(422, "unknown_album", str(error)) + except BatchConflict as error: + return _error(409, error.code, str(error)) + except BatchError as error: + return _error(422, error.code, str(error)) + + +@router.get("/upload-batches") +def list_batches(request: Request) -> dict: + return {"batches": _batches(request).list()} + + +@router.get("/upload-batches/{batch_id}") +def get_batch(batch_id: str, request: Request): + batch = _batches(request).get(batch_id) + if batch is None: + return _error(404, "not_found", f"unknown upload batch {batch_id}") + return batch + + +@router.post("/upload-batches/{batch_id}/start") +def start_batch(batch_id: str, request: Request): + """Queue the batch on the uploader lane. The worker performs the upload.""" + service = _batches(request) + batch = service.get(batch_id) + if batch is None: + return _error(404, "not_found", f"unknown upload batch {batch_id}") + try: + job = JobService(request.app.state.session_factory).enqueue( + UPLOAD_BATCH, + lock=UPLOAD_LOCK, + # One queued attempt per batch attempt: a double-clicked start reuses it. + idempotency_key=f"upload:{batch_id}:{batch['attempt_count']}", + items=[batch_id], ) + except JobBlocked as error: + return _error(409, error.code, str(error)) + return {"batch_id": batch_id, "job": job} + + +@router.post("/upload-batches/{batch_id}/cancel") +def cancel_batch(batch_id: str, request: Request): + try: + return _batches(request).cancel(batch_id) + except BatchError as error: + return _error(404 if error.code == "not_found" else 409, error.code, str(error)) diff --git a/photo_pipeline/integrations/immich_go.py b/photo_pipeline/integrations/immich_go.py index 39081e8..589dc44 100644 --- a/photo_pipeline/integrations/immich_go.py +++ b/photo_pipeline/integrations/immich_go.py @@ -9,20 +9,42 @@ the command that would actually run. Server reachability uses ``/api/server/ping`` through stdlib ``urllib`` — the app has no HTTP client dependency and this is one request. + +:func:`run_upload` is the only place the uploader is actually executed. It never +uses a shell (the argument list goes straight to ``execve``, so no path or album +name can be interpreted), streams the report to a file with a byte cap so a chatty +or looping uploader cannot fill the disk, scrubs the API key out of anything the +process echoes back, and polls a cancellation callback so a running upload can be +stopped without killing the worker. """ from __future__ import annotations import json +import os import shutil +import signal import subprocess +import threading +import time import urllib.error import urllib.request +from collections.abc import Callable from pathlib import Path REDACTED = "***" PING_PATH = "/api/server/ping" PING_TIMEOUT_SECONDS = 5.0 +# Reports are kept in full up to this size; beyond it the tail is dropped and the +# result is flagged truncated rather than growing without bound (concept §17). +MAX_REPORT_BYTES = 4_000_000 +UPLOAD_TIMEOUT_SECONDS = 6 * 60 * 60 +POLL_SECONDS = 0.05 +# Grace period between asking the uploader to stop and killing it. +TERMINATE_GRACE_SECONDS = 10.0 +# How long to wait for the last output after the process is gone. A child that +# outlived its parent can still hold the pipe; the report is not worth hanging for. +DRAIN_SECONDS = 2.0 def find_binary(binary: str = "immich-go") -> str | None: @@ -92,6 +114,103 @@ def redact(command: list[str]) -> list[str]: return [f"--api-key={REDACTED}" if arg.startswith("--api-key=") else arg for arg in command] +def run_upload( + command: list[str], + *, + report_path: Path | str, + secret: str | None = None, + max_report_bytes: int | None = None, # resolved at call time; see MAX_REPORT_BYTES + cancelled: Callable[[], bool] | None = None, + timeout: float = UPLOAD_TIMEOUT_SECONDS, +) -> dict: + """Run one upload and return its outcome. + + ``{"exit_code", "cancelled", "timed_out", "report_path", "report_bytes", + "report_truncated"}``. Output is streamed to ``report_path`` with ``secret`` + masked and the file capped at ``max_report_bytes``; the pipe keeps being drained + after the cap so the child never blocks on a full buffer. ``cancelled`` is polled + while the process runs: when it returns true the uploader is asked to stop, then + killed if it does not. + """ + report_path = Path(report_path) + report_path.parent.mkdir(parents=True, exist_ok=True) + max_report_bytes = MAX_REPORT_BYTES if max_report_bytes is None else max_report_bytes + needle = (secret or "").encode() or None + written = 0 + truncated = False + + process = subprocess.Popen( # noqa: S603 — argv list, never a shell string + command, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + shell=False, + # Own process group: stopping the upload must stop whatever the uploader + # spawned too, not leave orphans holding the pipe open. + start_new_session=True, + ) + + def _drain() -> None: + nonlocal written, truncated + with open(report_path, "wb") as report: + for line in process.stdout: # line granularity keeps the mask reliable + if needle: + line = line.replace(needle, REDACTED.encode()) + if written >= max_report_bytes: + truncated = True + continue # keep draining the pipe, stop growing the file + room = max_report_bytes - written + report.write(line[:room]) + written += min(len(line), room) + truncated = truncated or len(line) > room + report.flush() + + reader = threading.Thread(target=_drain, daemon=True) + reader.start() + + stopped = timed_out = False + deadline = time.monotonic() + timeout + while process.poll() is None: + if cancelled is not None and cancelled(): + stopped = True + elif time.monotonic() >= deadline: + timed_out = True + if stopped or timed_out: + _stop(process) + break + time.sleep(POLL_SECONDS) + + exit_code = process.wait() + reader.join(timeout=DRAIN_SECONDS) + if process.stdout is not None: + process.stdout.close() + return { + "exit_code": exit_code, + "cancelled": stopped, + "timed_out": timed_out, + "report_path": str(report_path), + "report_bytes": written, + "report_truncated": truncated, + } + + +def _stop(process: subprocess.Popen) -> None: + """Ask the uploader's whole process group to stop, then kill what remains.""" + _signal_group(process, signal.SIGTERM) + try: + process.wait(timeout=TERMINATE_GRACE_SECONDS) + except subprocess.TimeoutExpired: + _signal_group(process, signal.SIGKILL) + process.wait() + + +def _signal_group(process: subprocess.Popen, sig: int) -> None: + try: + os.killpg(os.getpgid(process.pid), sig) + except (ProcessLookupError, PermissionError, OSError): + # No group (already reaped, or a platform without them): signal the child. + process.send_signal(sig) + + def preview_command( *, binary: str, server_url: str, album_name: str, folder: Path | str ) -> list[str]: diff --git a/photo_pipeline/jobs/domain_handlers.py b/photo_pipeline/jobs/domain_handlers.py index 6aa5252..d73936c 100644 --- a/photo_pipeline/jobs/domain_handlers.py +++ b/photo_pipeline/jobs/domain_handlers.py @@ -1,10 +1,10 @@ -"""Domain job handlers: safety scoring and content analysis (US02-06). +"""Domain job handlers: safety scoring, content analysis, uploads (US02-06, US05-02). -Importing this module registers the ``safety_score`` and ``analysis`` job types so -the generic worker can run them per item (one item = one ``asset_id``). Each handler +Importing this module registers the ``safety_score``, ``analysis``, and +``upload_batch`` job types so the generic worker can run them per item. Each handler delegates to its service, which owns the real work and the privacy gate. Handlers are idempotent: re-scoring or re-analyzing one asset is safe after an interrupted -attempt. +attempt, and an upload batch refuses to re-run an attempt whose outcome is unknown. Providers/models are the service defaults here (real NsfwModel / vision provider); tests exercise the services directly with injected fakes rather than the worker. @@ -12,12 +12,15 @@ tests exercise the services directly with injected fakes rather than the worker. from __future__ import annotations -from photo_pipeline.jobs.handlers import JobContext, register +from photo_pipeline.jobs.handlers import Cancelled, JobContext, register SAFETY_SCORE = "safety_score" ANALYSIS = "analysis" +UPLOAD_BATCH = "upload_batch" # Both mutate the library's metadata/derived state; one at a time (concept §one job). LIBRARY_WRITE_LOCK = "library_write" +# The uploader lane: one album batch at a time (concept §16). +UPLOAD_LOCK = "upload" def _safety_score_item(asset_id: str, ctx: JobContext) -> None: @@ -32,5 +35,22 @@ def _analysis_item(asset_id: str, ctx: JobContext) -> None: AnalysisService(ctx.session_factory).run([asset_id]) +def _upload_batch_item(batch_id: str, ctx: JobContext) -> None: + """One item = one album batch. The upload itself is long and external, so the + handler hands the job's cancellation check to the service, which stops the + uploader and leaves a resumable batch.""" + from photo_pipeline.config import Config + from photo_pipeline.services.upload_batches import BatchState, UploadBatchService + + config = ctx.config if ctx.config is not None else Config.from_env() + service = UploadBatchService(ctx.session_factory, config=config) + batch = service.run(batch_id, worker_id=ctx.worker_id, cancelled=ctx.cancelled) + if batch["state"] == BatchState.CANCELLED: + raise Cancelled(f"upload batch {batch_id} was cancelled") + if batch["state"] in (BatchState.FAILED, BatchState.UNKNOWN): + raise RuntimeError(f"upload batch {batch_id} is {batch['state']}: {batch['error_code']}") + + register(SAFETY_SCORE, _safety_score_item) register(ANALYSIS, _analysis_item) +register(UPLOAD_BATCH, _upload_batch_item) diff --git a/photo_pipeline/jobs/handlers.py b/photo_pipeline/jobs/handlers.py index 3532e42..110a00f 100644 --- a/photo_pipeline/jobs/handlers.py +++ b/photo_pipeline/jobs/handlers.py @@ -34,6 +34,9 @@ class JobContext: fencing_token: int service: "JobService" session_factory: object | None = None + # Handlers that talk to an external tool (uploads) need the typed configuration; + # the worker passes its own so a test stack is never read from the environment. + config: object | None = None def cancelled(self) -> bool: from photo_pipeline.services.jobs import JobState diff --git a/photo_pipeline/jobs/worker.py b/photo_pipeline/jobs/worker.py index e70106e..68a6b9b 100644 --- a/photo_pipeline/jobs/worker.py +++ b/photo_pipeline/jobs/worker.py @@ -30,6 +30,7 @@ class Worker: *, job_types: Sequence[str] | None = None, lease_seconds: int = 60, + config: object | None = None, ) -> None: self._session_factory = session_factory self.service = JobService(session_factory) @@ -37,6 +38,7 @@ class Worker: self.worker_id = worker_id self.job_types = list(job_types if job_types is not None else self.handlers.keys()) self.lease_seconds = lease_seconds + self.config = config def run_once(self) -> str | None: """Recover stragglers, then claim and fully process one job. Returns its id.""" @@ -51,7 +53,9 @@ class Worker: def _process(self, job_id: str, job_type: str, token: int) -> None: handler = self.handlers[job_type] - ctx = JobContext(job_id, self.worker_id, token, self.service, self._session_factory) + ctx = JobContext( + job_id, self.worker_id, token, self.service, self._session_factory, self.config + ) self._reset_interrupted_items(job_id, token) cancelled = False diff --git a/photo_pipeline/models/__init__.py b/photo_pipeline/models/__init__.py index 983b437..09196e3 100644 --- a/photo_pipeline/models/__init__.py +++ b/photo_pipeline/models/__init__.py @@ -14,6 +14,7 @@ from photo_pipeline.models.duplicates import ( from photo_pipeline.models.jobs import Job, JobEvent, JobItem from photo_pipeline.models.renames import RenameOperation, RenamePlan from photo_pipeline.models.thumbnails import Thumbnail +from photo_pipeline.models.uploads import UploadBatch, UploadItem from photo_pipeline.models.workflow import AnalysisResult, SafetyReview __all__ = [ @@ -29,6 +30,8 @@ __all__ = [ "RenamePlan", "RenameOperation", "Thumbnail", + "UploadBatch", + "UploadItem", "SafetyReview", "AnalysisResult", ] diff --git a/photo_pipeline/models/uploads.py b/photo_pipeline/models/uploads.py new file mode 100644 index 0000000..821397c --- /dev/null +++ b/photo_pipeline/models/uploads.py @@ -0,0 +1,90 @@ +"""Upload batch persistence (US05-02). + +One batch is one approved album folder handed to ``immich-go``. It is the durable +record of an irreversible external action, so it stores everything needed to answer +"which exact bytes did we send, with which command, and how did it end?" after a +crash: the preflight token that authorised it, the redacted command, the uploader +version, the per-asset pre-upload hashes, every attempt, and where the raw report +was written. + +Item rows keep both digests: SHA-256 is the app's byte identity and SHA-1 is what +Immich/immich-go use to recognise a file it already has (concept §8). +""" + +from __future__ import annotations + +from datetime import datetime + +from sqlalchemy import ( + Boolean, + DateTime, + ForeignKey, + Integer, + String, + func, +) +from sqlalchemy.orm import Mapped, mapped_column + +from photo_pipeline.db import Base + + +class UploadBatch(Base): + __tablename__ = "upload_batches" + + id: Mapped[str] = mapped_column(String, primary_key=True) + album: Mapped[str] = mapped_column(String, nullable=False, index=True) + folder: Mapped[str] = mapped_column(String, nullable=False) + album_name: Mapped[str] = mapped_column(String, nullable=False) + # planned | running | cancelling | cancelled | succeeded | failed + # | unknown_requires_verification + state: Mapped[str] = mapped_column(String, nullable=False, default="planned") + # The preflight token this batch was approved against; re-checked before every + # attempt so changed bytes or decisions cannot be uploaded silently. + preflight_token: Mapped[str] = mapped_column(String, nullable=False) + allow_partial: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + + command: Mapped[str | None] = mapped_column(String) # JSON array, redacted + uploader_version: Mapped[str | None] = mapped_column(String) + asset_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + + attempt_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + # Bumped on every claim and used as the fencing token, so a superseded attempt + # cannot commit its outcome. + version: Mapped[int] = mapped_column(Integer, nullable=False, default=1) + worker_id: Mapped[str | None] = mapped_column(String) + + report_path: Mapped[str | None] = mapped_column(String) + report_bytes: Mapped[int | None] = mapped_column(Integer) + report_truncated: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + exit_code: Mapped[int | None] = mapped_column(Integer) + error_code: Mapped[str | None] = mapped_column(String) + error_message: Mapped[str | None] = mapped_column(String) + + started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now() + ) + + +class UploadItem(Base): + __tablename__ = "upload_items" + + batch_id: Mapped[str] = mapped_column( + ForeignKey("upload_batches.id", ondelete="CASCADE"), primary_key=True + ) + asset_id: Mapped[str] = mapped_column(String, primary_key=True) + path: Mapped[str] = mapped_column(String, nullable=False) + # Hashes of the bytes as they were when the batch was created. + sha256: Mapped[str | None] = mapped_column(String) + sha1: Mapped[str | None] = mapped_column(String) + # pending | sent | failed — the per-asset upload *result* is parsed from the + # report in US05-03 and verified in US05-04; ``sent`` only means the batch + # process exited successfully. + state: Mapped[str] = mapped_column(String, nullable=False, default="pending") + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now() + ) diff --git a/photo_pipeline/services/hashing.py b/photo_pipeline/services/hashing.py index 79526c5..9718683 100644 --- a/photo_pipeline/services/hashing.py +++ b/photo_pipeline/services/hashing.py @@ -24,7 +24,16 @@ _CHUNK = 1 << 20 def sha256_file(path: Path | str) -> str: - digest = hashlib.sha256() + return _digest_file(path, hashlib.sha256()) + + +def sha1_file(path: Path | str) -> str: + """SHA-1 of the file bytes. Not an identity hash here — it is the checksum + Immich/immich-go use to recognise an asset they already hold (concept §8).""" + return _digest_file(path, hashlib.sha1()) + + +def _digest_file(path: Path | str, digest) -> str: with open(path, "rb") as handle: for chunk in iter(lambda: handle.read(_CHUNK), b""): digest.update(chunk) diff --git a/photo_pipeline/services/upload_batches.py b/photo_pipeline/services/upload_batches.py new file mode 100644 index 0000000..3d97ad4 --- /dev/null +++ b/photo_pipeline/services/upload_batches.py @@ -0,0 +1,424 @@ +"""UploadBatchService — one approved album at a time through immich-go (US05-02). + +Preflight (US05-01) proves a scope is safe and issues a token; this service turns +that approval into a durable batch and runs it. Upload is the one stage the app +cannot undo, so the discipline is: + +- **the approval is re-proved, not remembered.** Before every attempt the batch's + preflight token is recomputed from the current library state. Bytes edited after + approval, a withdrawn safety decision, or a server that stopped answering all + produce a different token and the attempt is refused, never run "optimistically". +- **one lane.** A batch can only start while no other batch is running; the album + scope of a batch never widens after creation (concept §16 uploader lane). +- **cancellation is cooperative and durable.** ``cancel`` writes ``cancelling``; + the running attempt observes it through the database, stops the uploader, and + records ``cancelled``. A cancelled batch is re-runnable from its own boundary. +- **an interrupted attempt is uncertain, not failed.** Immich may have accepted + files the app never saw a report for, so ``recover`` marks a batch whose worker + vanished ``unknown_requires_verification`` (concept §15) instead of retrying it + blindly. Resolving that is US05-04. + +Per-asset upload *results* are deliberately not interpreted here — parsing the +report is US05-03. What this story guarantees is that the report exists, is bounded, +carries no credential, and that the batch's state is an honest description of what +the process did. +""" + +from __future__ import annotations + +import json +import uuid +from datetime import datetime, timezone +from pathlib import Path + +from sqlalchemy import select, update +from sqlalchemy.orm import sessionmaker + +from photo_pipeline.config import Config +from photo_pipeline.integrations import immich_go +from photo_pipeline.models import UploadBatch, UploadItem +from photo_pipeline.services.hashing import sha1_file +from photo_pipeline.services.uploads import UploadService + + +class BatchState: + PLANNED = "planned" + RUNNING = "running" + CANCELLING = "cancelling" + CANCELLED = "cancelled" + SUCCEEDED = "succeeded" + FAILED = "failed" + # The attempt died without a parsed outcome: the server may hold the files. + UNKNOWN = "unknown_requires_verification" + + +# States that occupy the single uploader lane. +LANE_STATES = frozenset({BatchState.RUNNING, BatchState.CANCELLING}) +# States a batch may be (re)started from. ``unknown_requires_verification`` is not +# among them: an uncertain upload must be verified (US05-04), never blindly retried. +RUNNABLE_STATES = frozenset({BatchState.PLANNED, BatchState.FAILED, BatchState.CANCELLED}) +# States whose batch is still the live one for its album. +OPEN_STATES = frozenset({BatchState.PLANNED, BatchState.RUNNING, BatchState.CANCELLING}) + +class ItemState: + PENDING = "pending" + # ``sent`` means the batch process exited cleanly, not that Immich confirmed the + # asset — the per-item outcome comes from the report in US05-03/US05-04. + SENT = "sent" + FAILED = "failed" + + +class BatchError(RuntimeError): + """The request cannot be carried out (unknown batch, wrong state, blocked).""" + + def __init__(self, code: str, message: str) -> None: + super().__init__(message) + self.code = code + + +class BatchConflict(BatchError): + """The lane is busy or the approval is stale — retry after resolving it.""" + + +def _now() -> datetime: + return datetime.now(timezone.utc) + + +class UploadBatchService: + def __init__(self, session_factory: sessionmaker, *, config: Config) -> None: + self._session_factory = session_factory + self._config = config + self._preflight = UploadService(session_factory, config=config) + + # ── creation ────────────────────────────────────────────────────────────── + + def create( + self, albums: list[str] | None = None, *, token: str, allow_partial: bool = False + ) -> list[dict]: + """Turn a *ready* preflight into one durable batch per album. + + ``token`` must be the token of the current preflight for the same scope and + policy; anything else means the browser is acting on a stale preview. + """ + report = self._preflight.preflight(albums, allow_partial=allow_partial) + if token != report["token"]: + raise BatchConflict( + "stale_preflight", "the preflight token does not describe the current state" + ) + if report["state"] != "ready": + raise BatchError("not_ready", "the scope has unresolved upload blockers") + + created: list[dict] = [] + for album in report["albums"]: + existing = self._open_batch_for(album["album"]) + if existing is not None: + created.append(existing) # idempotent: one open batch per album + continue + created.append(self._create_one(album, token=token, allow_partial=allow_partial)) + return created + + def _create_one(self, album: dict, *, token: str, allow_partial: bool) -> dict: + eligible = [asset for asset in album["assets"] if not asset["blockers"]] + batch_id = str(uuid.uuid4()) + command = immich_go.preview_command( + binary=self._config.immich_go_binary, + server_url=self._config.immich_server_url, + album_name=album["album_name"], + folder=album["folder"], + ) + with self._session_factory() as session: + session.add( + UploadBatch( + id=batch_id, + album=album["album"], + folder=album["folder"], + album_name=album["album_name"], + state=BatchState.PLANNED, + preflight_token=token, + allow_partial=allow_partial, + command=json.dumps(command), + uploader_version=immich_go.version(self._config.immich_go_binary), + asset_count=len(eligible), + ) + ) + for asset in eligible: + session.add( + UploadItem( + batch_id=batch_id, + asset_id=asset["asset_id"], + path=asset["current_path"], + sha256=asset["current_sha256"], + sha1=sha1_file(asset["current_path"]), + state=ItemState.PENDING, + ) + ) + session.commit() + return self.get(batch_id) + + # ── running ─────────────────────────────────────────────────────────────── + + def run(self, batch_id: str, *, worker_id: str = "uploader", cancelled=None) -> dict: + """Run one batch to completion. Blocks for the duration of the upload.""" + batch = self._require(batch_id) + if batch["state"] not in RUNNABLE_STATES: + code = ( + "requires_verification" + if batch["state"] == BatchState.UNKNOWN + else "not_runnable" + ) + raise BatchError(code, f"batch {batch_id} is {batch['state']}") + busy = [row for row in self.list() if row["id"] != batch_id and row["state"] in LANE_STATES] + if busy: + raise BatchConflict("lane_busy", f"upload batch {busy[0]['id']} is still running") + + # The approval is re-proved here, immediately before the irreversible act. + if not self._preflight.verify_token( + batch["preflight_token"], [batch["album"]], allow_partial=batch["allow_partial"] + ): + self._finish( + batch_id, + token=batch["version"], + state=BatchState.FAILED, + error=("stale_preflight", "the library changed after this batch was approved"), + ) + raise BatchConflict( + "stale_preflight", "the library changed after this batch was approved" + ) + + token = self._claim(batch_id, worker_id=worker_id) + attempt = self.get(batch_id)["attempt_count"] + report_path = ( + Path(self._config.data_dir) / "uploads" / f"{batch_id}-attempt-{attempt}.log" + ) + key = self._config.immich_api_key.get_secret_value() if self._config.immich_api_key else "" + command = immich_go.build_command( + binary=self._config.immich_go_binary, + server_url=self._config.immich_server_url, + api_key=key, + album_name=batch["album_name"], + folder=batch["folder"], + ) + + def _stop_requested() -> bool: + if cancelled is not None and cancelled(): + return True + current = self.get(batch_id) + return current is None or current["state"] == BatchState.CANCELLING + + try: + result = immich_go.run_upload( + command, + report_path=report_path, + secret=key or None, + cancelled=_stop_requested, + ) + except OSError as error: # uploader vanished between preflight and exec + self._finish( + batch_id, + token=token, + state=BatchState.FAILED, + error=("uploader_failed", str(error)), + ) + return self.get(batch_id) + + if result["cancelled"]: + state, error, item_state = BatchState.CANCELLED, None, None + elif result["timed_out"]: + # Killed mid-flight: the server may already hold some of the files. + state = BatchState.UNKNOWN + error = ("timeout", "the uploader exceeded its time limit and was stopped") + item_state = None + elif result["exit_code"] == 0: + state, error, item_state = BatchState.SUCCEEDED, None, ItemState.SENT + else: + state = BatchState.FAILED + error = ("uploader_failed", f"immich-go exited with {result['exit_code']}") + item_state = ItemState.FAILED + + self._finish(batch_id, token=token, state=state, error=error, result=result) + if item_state: + self._set_items(batch_id, item_state) + return self.get(batch_id) + + def cancel(self, batch_id: str) -> dict: + """Request a stop. A running attempt drains; a planned batch stops outright.""" + batch = self._require(batch_id) + if batch["state"] == BatchState.PLANNED: + target = BatchState.CANCELLED + elif batch["state"] == BatchState.RUNNING: + target = BatchState.CANCELLING + else: + raise BatchError("not_cancellable", f"batch {batch_id} is {batch['state']}") + with self._session_factory() as session: + row = session.get(UploadBatch, batch_id) + row.state = target + row.updated_at = _now() + if target == BatchState.CANCELLED: + row.finished_at = _now() + session.commit() + return self.get(batch_id) + + # ── recovery ────────────────────────────────────────────────────────────── + + def recover(self) -> dict: + """Resolve batches whose attempt died with the process. + + A batch that never started is left ``planned`` and simply runs later. One + that was mid-upload cannot be classified from local state — immich-go may + have transferred everything before the crash — so it becomes + ``unknown_requires_verification`` and releases the lane rather than being + retried or declared failed. + """ + interrupted = 0 + with self._session_factory() as session: + for row in session.scalars( + select(UploadBatch).where(UploadBatch.state.in_(LANE_STATES)) + ): + row.state = BatchState.UNKNOWN + row.error_code = "interrupted" + row.error_message = "the uploader process ended without a recorded outcome" + row.finished_at = _now() + row.updated_at = _now() + row.version += 1 + interrupted += 1 + session.commit() + return {"interrupted": interrupted} + + # ── reads ───────────────────────────────────────────────────────────────── + + def get(self, batch_id: str) -> dict | None: + with self._session_factory() as session: + row = session.get(UploadBatch, batch_id) + if row is None: + return None + items = list( + session.scalars( + select(UploadItem) + .where(UploadItem.batch_id == batch_id) + .order_by(UploadItem.path) + ) + ) + return _batch_dict(row, items) + + def list(self) -> list[dict]: + with self._session_factory() as session: + rows = list(session.scalars(select(UploadBatch).order_by(UploadBatch.created_at))) + return [_batch_dict(row, []) for row in rows] + + def report(self, batch_id: str) -> str: + """The raw uploader output kept for this batch, or ``""`` when there is none.""" + batch = self._require(batch_id) + path = Path(batch["report_path"]) if batch["report_path"] else None + if path is None or not path.exists(): + return "" + return path.read_text(errors="replace") + + # ── internals ───────────────────────────────────────────────────────────── + + def _open_batch_for(self, album: str) -> dict | None: + with self._session_factory() as session: + row = session.scalar( + select(UploadBatch).where( + UploadBatch.album == album, UploadBatch.state.in_(OPEN_STATES) + ) + ) + return self.get(row.id) if row else None + + def _claim(self, batch_id: str, *, worker_id: str) -> int: + """Take ownership: bump the version (the fencing token) and start an attempt.""" + with self._session_factory() as session: + row = session.get(UploadBatch, batch_id) + row.version += 1 + row.attempt_count += 1 + row.state = BatchState.RUNNING + row.worker_id = worker_id + row.error_code = row.error_message = None + row.started_at = _now() + row.finished_at = None + row.updated_at = _now() + token = row.version + session.commit() + return token + + def _finish( + self, + batch_id: str, + *, + token: int, + state: str, + error: tuple[str, str] | None = None, + result: dict | None = None, + ) -> None: + """Record the outcome, but only for the attempt that still owns the batch.""" + values = { + "state": state, + "finished_at": _now(), + "updated_at": _now(), + "error_code": error[0] if error else None, + "error_message": error[1][:500] if error else None, + } + if result is not None: + values |= { + "exit_code": result["exit_code"], + "report_path": result["report_path"], + "report_bytes": result["report_bytes"], + "report_truncated": result["report_truncated"], + } + with self._session_factory() as session: + session.execute( + update(UploadBatch) + .where(UploadBatch.id == batch_id, UploadBatch.version == token) + .values(**values) + ) + session.commit() + + def _set_items(self, batch_id: str, state: str) -> None: + with self._session_factory() as session: + session.execute( + update(UploadItem) + .where(UploadItem.batch_id == batch_id) + .values(state=state, updated_at=_now()) + ) + session.commit() + + def _require(self, batch_id: str) -> dict: + batch = self.get(batch_id) + if batch is None: + raise BatchError("not_found", f"unknown upload batch {batch_id!r}") + return batch + + +def _batch_dict(row: UploadBatch, items: list[UploadItem]) -> dict: + return { + "id": row.id, + "album": row.album, + "folder": row.folder, + "album_name": row.album_name, + "state": row.state, + "preflight_token": row.preflight_token, + "allow_partial": row.allow_partial, + "command": json.loads(row.command or "[]"), + "uploader_version": row.uploader_version, + "asset_count": row.asset_count, + "attempt_count": row.attempt_count, + "version": row.version, + "worker_id": row.worker_id, + "report_path": row.report_path, + "report_bytes": row.report_bytes, + "report_truncated": row.report_truncated, + "exit_code": row.exit_code, + "error_code": row.error_code, + "error_message": row.error_message, + "started_at": row.started_at.isoformat() if row.started_at else None, + "finished_at": row.finished_at.isoformat() if row.finished_at else None, + "items": [ + { + "asset_id": item.asset_id, + "path": item.path, + "sha256": item.sha256, + "sha1": item.sha1, + "state": item.state, + } + for item in items + ], + } diff --git a/tests/integration/test_upload_batches.py b/tests/integration/test_upload_batches.py new file mode 100644 index 0000000..0fdbf64 --- /dev/null +++ b/tests/integration/test_upload_batches.py @@ -0,0 +1,588 @@ +"""Upload batch orchestration (US05-02). + +The uploader is a real executable on disk driven through the real integration +adapter and ``subprocess`` — never a mock — so argument construction, output +bounding, credential privacy, and killing a running process are exercised the way +production does them. Each fake uploader records the argv it was given into a +side-channel file, which is what the argument/album-isolation assertions read. +""" + +import json +import os +import signal +import stat +import threading +import time +import uuid +from datetime import datetime, timezone +from http.server import BaseHTTPRequestHandler, HTTPServer +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.jobs.domain_handlers import UPLOAD_BATCH +from photo_pipeline.jobs.worker import Worker +from photo_pipeline.models import AnalysisResult, Asset, SafetyReview, UploadBatch +from photo_pipeline.services.hashing import sha1_file, sha256_file +from photo_pipeline.services.jobs import JobService, JobState +from photo_pipeline.services.upload_batches import ( + BatchConflict, + BatchError, + BatchState, + ItemState, + UploadBatchService, +) +from photo_pipeline.services.uploads import UploadService + +NOW = datetime(2026, 1, 1, tzinfo=timezone.utc) +SENTINEL_KEY = "immich-sentinel-9f3a2b" +UPLOADER_VERSION = "immich-go 0.21.0" + + +# ── fake external boundary ─────────────────────────────────────────────────── + + +class _PingHandler(BaseHTTPRequestHandler): + def do_GET(self): # noqa: N802 (BaseHTTPRequestHandler API) + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.end_headers() + self.wfile.write(b'{"res":"pong"}') + + def log_message(self, *args): + pass + + +@pytest.fixture +def immich_server(): + server = HTTPServer(("127.0.0.1", 0), _PingHandler) + threading.Thread(target=server.serve_forever, daemon=True).start() + yield f"http://127.0.0.1:{server.server_port}" + server.shutdown() + server.server_close() + + +def _uploader(tmp_path, body: str = "", *, name="immich-go"): + """A real executable standing in for immich-go. + + ``--version`` answers like the real tool; any other invocation appends its + complete argv to ``.argv`` and then runs ``body``. + """ + path = tmp_path / name + argv_log = tmp_path / f"{name}.argv" + path.write_text( + "#!/bin/sh\n" + f'if [ "$1" = "--version" ]; then echo "{UPLOADER_VERSION}"; exit 0; fi\n' + f'printf "%s\\n" "$*" >> "{argv_log}"\n' + f"{body}\n" + ) + path.chmod(path.stat().st_mode | stat.S_IEXEC | stat.S_IXGRP | stat.S_IXOTH) + return path + + +def _argv(tmp_path, name="immich-go") -> list[str]: + log = tmp_path / f"{name}.argv" + return log.read_text().splitlines() if log.exists() else [] + + +# ── environment ────────────────────────────────────────────────────────────── + + +def _env(tmp_path, server_url, *, uploader=None): + (tmp_path / "data").mkdir(exist_ok=True) + lib = tmp_path / "lib" + lib.mkdir(exist_ok=True) + config = Config.from_env( + { + "PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data"), + "PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib), + "PHOTO_PIPELINE_IMMICH_SERVER_URL": server_url, + "PHOTO_PIPELINE_IMMICH_API_KEY": SENTINEL_KEY, + "PHOTO_PIPELINE_IMMICH_GO_BINARY": str( + uploader if uploader is not None else _uploader(tmp_path, "exit 0") + ), + } + ) + run_migrations(config.database_url) + return config, create_session_factory(create_db_engine(config.database_url)), lib + + +def _album(sf, lib, album="rome", names=("a.jpg", "b.jpg")): + folder = lib / album + folder.mkdir(parents=True, exist_ok=True) + with sf() as session: + for name in names: + path = folder / name + path.write_bytes(f"{album}/{name}".encode() * 16) + asset_id = str(uuid.uuid4()) + session.add( + Asset( + id=asset_id, + original_path=str(path), + current_path=str(path), + discovered_at=NOW, + hash_version=1, + byte_size=path.stat().st_size, + current_sha256=sha256_file(path), + ) + ) + session.add( + SafetyReview( + id=str(uuid.uuid4()), + asset_id=asset_id, + decision="sfw", + exif_verified_at=NOW, + ) + ) + session.add( + AnalysisResult(asset_id=asset_id, status="analyzed", exif_written_at=NOW) + ) + session.commit() + return folder + + +def _service(sf, config): + return UploadBatchService(sf, config=config) + + +def _approved(sf, config, albums=None, **kwargs): + """Create batches from a fresh, ready preflight.""" + report = UploadService(sf, config=config).preflight(albums, **kwargs) + assert report["state"] == "ready", report["blockers"] + return _service(sf, config).create(albums, token=report["token"], **kwargs) + + +# ── creation ───────────────────────────────────────────────────────────────── + + +def test_batch_records_album_assets_hashes_and_command(tmp_path, immich_server): + config, sf, lib = _env(tmp_path, immich_server) + folder = _album(sf, lib) + + (batch,) = _approved(sf, config) + + assert batch["state"] == BatchState.PLANNED + assert batch["album"] == "rome" and batch["folder"] == str(folder) + assert batch["asset_count"] == 2 and len(batch["items"]) == 2 + assert batch["uploader_version"] == UPLOADER_VERSION + assert "--album-name=rome" in batch["command"] + assert "--api-key=***" in batch["command"] + for item in batch["items"]: + assert item["sha256"] == sha256_file(item["path"]) + assert item["sha1"] == sha1_file(item["path"]) + assert item["state"] == ItemState.PENDING + + +def test_one_batch_per_album_and_scope_is_isolated(tmp_path, immich_server): + config, sf, lib = _env(tmp_path, immich_server) + _album(sf, lib, "rome") + _album(sf, lib, "paris", names=("c.jpg",)) + + batches = _approved(sf, config, ["paris"]) + + assert [b["album"] for b in batches] == ["paris"] + assert [b["asset_count"] for b in batches] == [1] + + +def test_creating_twice_reuses_the_open_batch(tmp_path, immich_server): + config, sf, lib = _env(tmp_path, immich_server) + _album(sf, lib) + + first = _approved(sf, config)[0] + second = _approved(sf, config)[0] + + assert first["id"] == second["id"] + + +def test_stale_preflight_token_cannot_create_a_batch(tmp_path, immich_server): + config, sf, lib = _env(tmp_path, immich_server) + folder = _album(sf, lib) + token = UploadService(sf, config=config).preflight()["token"] + (folder / "a.jpg").write_bytes(b"edited after approval") + + with pytest.raises(BatchConflict) as error: + _service(sf, config).create(token=token) + + assert error.value.code == "stale_preflight" + + +def test_blocked_scope_cannot_create_a_batch(tmp_path, immich_server): + config, sf, lib = _env(tmp_path, immich_server) + folder = _album(sf, lib) + (folder / "a.jpg").write_bytes(b"edited before approval") + report = UploadService(sf, config=config).preflight() + + with pytest.raises(BatchError) as error: + _service(sf, config).create(token=report["token"]) + + assert error.value.code == "not_ready" + + +# ── running the uploader ───────────────────────────────────────────────────── + + +def test_run_invokes_the_uploader_with_the_batch_album_and_folder(tmp_path, immich_server): + config, sf, lib = _env(tmp_path, immich_server) + folder = _album(sf, lib) + (batch,) = _approved(sf, config) + + result = _service(sf, config).run(batch["id"]) + + assert result["state"] == BatchState.SUCCEEDED and result["exit_code"] == 0 + (invocation,) = _argv(tmp_path) + assert "upload from-folder" in invocation + assert "--album-name=rome" in invocation + assert str(folder) in invocation + assert all(item["state"] == ItemState.SENT for item in result["items"]) + assert result["attempt_count"] == 1 and result["started_at"] and result["finished_at"] + + +def test_other_albums_are_never_passed_to_the_uploader(tmp_path, immich_server): + config, sf, lib = _env(tmp_path, immich_server) + _album(sf, lib, "rome") + paris = _album(sf, lib, "paris", names=("c.jpg",)) + (batch,) = _approved(sf, config, ["rome"]) + + _service(sf, config).run(batch["id"]) + + (invocation,) = _argv(tmp_path) + assert str(paris) not in invocation and "paris" not in invocation + + +def test_album_names_are_arguments_not_shell_text(tmp_path, immich_server): + """A folder whose name contains shell metacharacters must reach the uploader + verbatim; nothing may be interpreted (no shell is involved).""" + config, sf, lib = _env(tmp_path, immich_server) + hostile = "rome; touch pwned" + _album(sf, lib, hostile, names=("a.jpg",)) + (batch,) = _approved(sf, config) + + result = _service(sf, config).run(batch["id"]) + + assert result["state"] == BatchState.SUCCEEDED + assert f"--album-name={hostile}" in _argv(tmp_path)[0] + assert not (Path.cwd() / "pwned").exists() and not (lib / "pwned").exists() + + +def test_uploader_failure_is_recorded_and_items_are_not_sent(tmp_path, immich_server): + config, sf, lib = _env( + tmp_path, immich_server, uploader=_uploader(tmp_path, "echo 'boom' >&2; exit 3") + ) + _album(sf, lib) + (batch,) = _approved(sf, config) + + result = _service(sf, config).run(batch["id"]) + + assert result["state"] == BatchState.FAILED and result["exit_code"] == 3 + assert result["error_code"] == "uploader_failed" + assert all(item["state"] == ItemState.FAILED for item in result["items"]) + assert "boom" in _service(sf, config).report(batch["id"]) + + +def test_a_failed_batch_can_be_retried_as_a_new_attempt(tmp_path, immich_server): + config, sf, lib = _env(tmp_path, immich_server, uploader=_uploader(tmp_path, "exit 3")) + _album(sf, lib) + (batch,) = _approved(sf, config) + service = _service(sf, config) + service.run(batch["id"]) + + retried = service.run(batch["id"]) + + assert retried["attempt_count"] == 2 + assert len(_argv(tmp_path)) == 2 + + +def test_bytes_edited_after_approval_block_the_upload(tmp_path, immich_server): + config, sf, lib = _env(tmp_path, immich_server) + folder = _album(sf, lib) + (batch,) = _approved(sf, config) + (folder / "a.jpg").write_bytes(b"edited after approval") + + with pytest.raises(BatchConflict) as error: + _service(sf, config).run(batch["id"]) + + assert error.value.code == "stale_preflight" + assert _argv(tmp_path) == [], "the uploader must not run on changed bytes" + assert _service(sf, config).get(batch["id"])["state"] == BatchState.FAILED + + +# ── credential privacy ─────────────────────────────────────────────────────── + + +def test_the_api_key_reaches_the_uploader_but_never_the_record(tmp_path, immich_server): + config, sf, lib = _env(tmp_path, immich_server) + _album(sf, lib) + (batch,) = _approved(sf, config) + + result = _service(sf, config).run(batch["id"]) + + assert f"--api-key={SENTINEL_KEY}" in _argv(tmp_path)[0], "the real key must be passed" + assert SENTINEL_KEY not in json.dumps(result, default=str) + + +def test_a_key_echoed_by_the_uploader_is_scrubbed_from_the_report(tmp_path, immich_server): + config, sf, lib = _env( + tmp_path, immich_server, uploader=_uploader(tmp_path, 'echo "using key $4"; exit 0') + ) + _album(sf, lib) + (batch,) = _approved(sf, config) + service = _service(sf, config) + + service.run(batch["id"]) + + report = service.report(batch["id"]) + assert SENTINEL_KEY not in report and "***" in report + + +# ── bounded output ─────────────────────────────────────────────────────────── + + +def test_a_chatty_uploader_cannot_grow_the_report_without_bound(tmp_path, immich_server): + config, sf, lib = _env( + tmp_path, + immich_server, + uploader=_uploader(tmp_path, "i=0; while [ $i -lt 2000 ]; do echo line-$i; i=$((i+1)); done"), + ) + _album(sf, lib) + (batch,) = _approved(sf, config) + + from photo_pipeline.integrations import immich_go + + service = _service(sf, config) + original = immich_go.MAX_REPORT_BYTES + immich_go.MAX_REPORT_BYTES = 500 # a cap small enough to hit in one test + try: + result = service.run(batch["id"]) + finally: + immich_go.MAX_REPORT_BYTES = original + + assert result["state"] == BatchState.SUCCEEDED, "the uploader still finished normally" + assert result["report_truncated"] is True + assert result["report_bytes"] <= 500 + assert Path(result["report_path"]).stat().st_size <= 500 + + +# ── one lane ───────────────────────────────────────────────────────────────── + + +def test_a_second_batch_cannot_run_while_one_is_running(tmp_path, immich_server): + config, sf, lib = _env(tmp_path, immich_server) + _album(sf, lib, "rome") + _album(sf, lib, "paris", names=("c.jpg",)) + rome, paris = sorted(_approved(sf, config), key=lambda b: b["album"] != "rome") + with sf() as session: # a batch already occupying the lane + session.get(UploadBatch, rome["id"]).state = BatchState.RUNNING + session.commit() + + with pytest.raises(BatchConflict) as error: + _service(sf, config).run(paris["id"]) + + assert error.value.code == "lane_busy" + assert _argv(tmp_path) == [] + + +def test_the_job_lock_refuses_a_second_queued_upload(tmp_path, immich_server): + config, sf, lib = _env(tmp_path, immich_server) + _album(sf, lib, "rome") + _album(sf, lib, "paris", names=("c.jpg",)) + batches = _approved(sf, config) + + with TestClient(create_app(config)) as client: + first = client.post(f"/api/v1/upload-batches/{batches[0]['id']}/start") + second = client.post(f"/api/v1/upload-batches/{batches[1]['id']}/start") + + assert first.status_code == 200 + assert second.status_code == 409 and second.json()["error"]["code"] == "lock_held" + + +# ── cancellation ───────────────────────────────────────────────────────────── + + +def test_cancelling_a_running_batch_stops_the_uploader(tmp_path, immich_server): + """The uploader sleeps; cancelling flips the batch to ``cancelling`` and the + running attempt must terminate the process and record ``cancelled``.""" + config, sf, lib = _env( + tmp_path, immich_server, uploader=_uploader(tmp_path, 'echo started; sleep 30; exit 0') + ) + _album(sf, lib) + (batch,) = _approved(sf, config) + service = _service(sf, config) + outcome = {} + + def _run(): + outcome["batch"] = service.run(batch["id"]) + + runner = threading.Thread(target=_run) + started = time.monotonic() + runner.start() + while service.get(batch["id"])["state"] != BatchState.RUNNING: + assert time.monotonic() - started < 30, "the attempt never started" + time.sleep(0.02) + + service.cancel(batch["id"]) + runner.join(timeout=30) + + assert not runner.is_alive(), "cancellation must not wait for the uploader's own timeout" + assert outcome["batch"]["state"] == BatchState.CANCELLED + assert all(item["state"] == ItemState.PENDING for item in outcome["batch"]["items"]) + + +def test_cancelling_a_planned_batch_never_starts_the_uploader(tmp_path, immich_server): + config, sf, lib = _env(tmp_path, immich_server) + _album(sf, lib) + (batch,) = _approved(sf, config) + service = _service(sf, config) + + cancelled = service.cancel(batch["id"]) + + assert cancelled["state"] == BatchState.CANCELLED + assert _argv(tmp_path) == [] + + +def test_a_cancelled_batch_can_be_run_again(tmp_path, immich_server): + config, sf, lib = _env(tmp_path, immich_server) + _album(sf, lib) + (batch,) = _approved(sf, config) + service = _service(sf, config) + service.cancel(batch["id"]) + + result = service.run(batch["id"]) + + assert result["state"] == BatchState.SUCCEEDED and result["attempt_count"] == 1 + + +# ── restart / recovery ─────────────────────────────────────────────────────── + + +def test_an_interrupted_attempt_becomes_uncertain_not_failed(tmp_path, immich_server): + """The process died mid-upload: Immich may hold the files, so the batch requires + verification (US05-04) instead of a blind retry, and the lane is released.""" + config, sf, lib = _env(tmp_path, immich_server) + _album(sf, lib) + (batch,) = _approved(sf, config) + with sf() as session: # what a killed worker leaves behind + session.get(UploadBatch, batch["id"]).state = BatchState.RUNNING + session.commit() + service = _service(sf, config) + + assert service.recover() == {"interrupted": 1} + + recovered = service.get(batch["id"]) + assert recovered["state"] == BatchState.UNKNOWN + assert recovered["error_code"] == "interrupted" + with pytest.raises(BatchError) as error: + service.run(batch["id"]) + assert error.value.code == "requires_verification" + assert _argv(tmp_path) == [] + + +def test_recovery_leaves_a_planned_batch_runnable(tmp_path, immich_server): + config, sf, lib = _env(tmp_path, immich_server) + _album(sf, lib) + (batch,) = _approved(sf, config) + service = _service(sf, config) + + assert service.recover() == {"interrupted": 0} + + assert service.run(batch["id"])["state"] == BatchState.SUCCEEDED + + +def test_application_startup_recovers_an_interrupted_batch(tmp_path, immich_server): + config, sf, lib = _env(tmp_path, immich_server) + _album(sf, lib) + (batch,) = _approved(sf, config) + with sf() as session: + session.get(UploadBatch, batch["id"]).state = BatchState.RUNNING + session.commit() + + with TestClient(create_app(config)) as client: + state = client.get(f"/api/v1/upload-batches/{batch['id']}").json()["state"] + + assert state == BatchState.UNKNOWN + + +def test_a_killed_uploader_leaves_an_uncertain_batch(tmp_path, immich_server): + """A real SIGKILL of the uploader process, not a simulated state write.""" + config, sf, lib = _env( + tmp_path, + immich_server, + uploader=_uploader(tmp_path, 'echo "pid $$"; sleep 30; exit 0'), + ) + _album(sf, lib) + (batch,) = _approved(sf, config) + service = _service(sf, config) + outcome = {} + runner = threading.Thread(target=lambda: outcome.update(batch=service.run(batch["id"]))) + runner.start() + + report = tmp_path / "data" / "uploads" + deadline = time.monotonic() + 30 + pid = None + while pid is None: + assert time.monotonic() < deadline, "the uploader never announced itself" + for log in report.glob("*.log"): + text = log.read_text() + if text.startswith("pid "): + pid = int(text.split()[1]) + time.sleep(0.02) + os.kill(pid, signal.SIGKILL) + runner.join(timeout=30) + + # The process is gone with a non-zero status and no parsed report: the attempt + # failed locally, and a restart classifies it honestly. + assert outcome["batch"]["state"] in (BatchState.FAILED, BatchState.UNKNOWN) + assert outcome["batch"]["exit_code"] != 0 + + +# ── worker integration ─────────────────────────────────────────────────────── + + +def test_the_worker_runs_a_queued_batch_on_the_upload_lane(tmp_path, immich_server): + config, sf, lib = _env(tmp_path, immich_server) + _album(sf, lib) + (batch,) = _approved(sf, config) + + with TestClient(create_app(config)) as client: + response = client.post(f"/api/v1/upload-batches/{batch['id']}/start") + job_id = response.json()["job"]["id"] + Worker(sf, worker_id="uploader-1", job_types=[UPLOAD_BATCH], config=config).run_once() + + assert JobService(sf).get(job_id)["state"] == JobState.SUCCEEDED + assert _service(sf, config).get(batch["id"])["state"] == BatchState.SUCCEEDED + assert len(_argv(tmp_path)) == 1 + + +# ── API surface ────────────────────────────────────────────────────────────── + + +def test_api_creates_lists_and_reads_batches_without_secrets(tmp_path, immich_server): + config, sf, lib = _env(tmp_path, immich_server) + _album(sf, lib) + + with TestClient(create_app(config)) as client: + token = client.post("/api/v1/upload-preflight", json={}).json()["token"] + created = client.post("/api/v1/upload-batches", json={"token": token}) + listed = client.get("/api/v1/upload-batches") + batch_id = created.json()["batches"][0]["id"] + fetched = client.get(f"/api/v1/upload-batches/{batch_id}") + + assert created.status_code == 201 + assert listed.json()["batches"][0]["id"] == batch_id + assert fetched.json()["state"] == BatchState.PLANNED + assert SENTINEL_KEY not in created.text + listed.text + fetched.text + + +def test_api_rejects_a_stale_token_and_an_unknown_batch(tmp_path, immich_server): + config, sf, lib = _env(tmp_path, immich_server) + _album(sf, lib) + + with TestClient(create_app(config)) as client: + stale = client.post("/api/v1/upload-batches", json={"token": "v1:not-the-token"}) + missing = client.get("/api/v1/upload-batches/does-not-exist") + + assert stale.status_code == 409 and stale.json()["error"]["code"] == "stale_preflight" + assert missing.status_code == 404 diff --git a/tests/story_traceability.json b/tests/story_traceability.json index 51312c9..b0f8301 100644 --- a/tests/story_traceability.json +++ b/tests/story_traceability.json @@ -106,6 +106,9 @@ ], "US05-01": [ "tests/integration/test_upload_preflight.py" + ], + "US05-02": [ + "tests/integration/test_upload_batches.py" ] } }