US07-04: Prove Concurrency and Crash Recovery

photo_pipeline/services/analysis.py

@@ -78,6 +78,11 @@ class AnalysisService:
             latest[review.asset_id] = review.decision
         return {aid for aid, decision in latest.items() if decision == SFW}
 
+    def _is_still_sfw(self, asset_id: str) -> bool:
+        """Re-read the current safety decision straight from the database."""
+        with self._session_factory() as session:
+            return asset_id in self._sfw_asset_ids(session)
+
     def eligible_asset_ids(self) -> list[str]:
         """Confirmed-SFW canonical active assets without a completed analysis."""
         with self._session_factory() as session:
@@ -161,6 +166,22 @@ class AnalysisService:
                 self._store(asset_id, status="error", result=None, error=str(error), tokens=0, raw="")
                 errors += 1
                 continue
+            # Third gate, after the call: a provider request takes seconds, and the
+            # reviewer may have flipped this asset to NSFW while it was in flight.
+            # The result describes an asset that is no longer analysable, so it is
+            # discarded — not stored, and above all not written into its EXIF
+            # (concept §18 scenario 7, US07-04).
+            if not self._is_still_sfw(asset_id):
+                self._store(
+                    asset_id,
+                    status="skipped_nsfw",
+                    result=None,
+                    error="the safety decision changed while analysis was in flight",
+                    tokens=0,
+                    raw="",
+                )
+                skipped += 1
+                continue
             self._store(
                 asset_id,
                 status="analyzed",

photo_pipeline/services/archive_transfer.py

@@ -52,6 +52,7 @@ from sqlalchemy import select
 from sqlalchemy.orm import sessionmaker
 
 from photo_pipeline.config import Config
+from photo_pipeline.faults import maybe_fault
 from photo_pipeline.models import ArchiveLocation, ArchiveOperation, ArchivePlan, Asset, AssetPath
 from photo_pipeline.services.archive_journal import (
     ARCHIVE,
@@ -63,7 +64,7 @@ from photo_pipeline.services.archive_journal import (
 from photo_pipeline.services.archives import MARKER_NAME, ArchiveError, ArchiveService
 from photo_pipeline.services.duplicates import DuplicateService
 from photo_pipeline.services.hashing import sha256_file
-from photo_pipeline.services.rename_apply import PreconditionFailed, maybe_fault
+from photo_pipeline.services.rename_apply import PreconditionFailed
 from photo_pipeline.services.thumbnails import ThumbnailService
 
 # The per-medium manifest: one JSON line per archived file, appended and fsynced

photo_pipeline/services/exif_checkpoint.py

@@ -25,6 +25,7 @@ import uuid
 from dataclasses import dataclass
 from datetime import datetime, timezone
 
+from photo_pipeline.faults import EXIF_WRITTEN, maybe_fault
 from photo_pipeline.integrations import exiftool
 from photo_pipeline.models import ExifProjection
 from photo_pipeline.services import hashing
@@ -120,6 +121,10 @@ def run(
     if not exiftool.apply_keywords(path, add=add, remove=remove):
         return CheckpointResult(FAILED, reason="write_failed")
 
+    # The file on disk has changed; nothing about it is recorded yet. A crash here
+    # is the worst case for metadata, so it is a fault control point (US07-04).
+    maybe_fault(EXIF_WRITTEN)
+
     after = exiftool.read_all(path)
     if after is None:
         return CheckpointResult(FAILED, reason="readback_unreadable")

photo_pipeline/services/jobs.py

@@ -236,17 +236,32 @@ class JobService:
                 raise InvalidTransition(f"{job.state} -> {to_state}")
             if worker_id is not None and job.lease_owner not in (None, worker_id):
                 raise JobConflict(f"job {job_id} owned by {job.lease_owner}, not {worker_id}")
-            job.state = to_state
-            job.version += 1
-            job.updated_at = now
+
+            # Compare-and-set on the version this decision was made against. Without
+            # it a transition validated against a row that has since been claimed,
+            # cancelled, or finished would overwrite that newer state (concept §16
+            # database rule 6) — a cancel racing a claim used to un-claim a running
+            # job and leave the worker finalizing a job it no longer owned.
+            values = {
+                "state": to_state,
+                "version": job.version + 1,
+                "updated_at": now,
+            }
             if error:
-                job.error_code, job.error_message = error
+                values["error_code"], values["error_message"] = error
             if to_state in TERMINAL_STATES:
-                job.finished_at = now
-                job.lease_owner = None
-                job.lease_expires_at = None
+                values.update(finished_at=now, lease_owner=None, lease_expires_at=None)
+            result = session.execute(
+                update(Job).where(Job.id == job_id, Job.version == job.version).values(**values)
+            )
+            if result.rowcount != 1:
+                session.rollback()
+                raise JobConflict(
+                    f"job {job_id} changed while transitioning to {to_state}; retry"
+                )
             self._event(session, job_id, f"state:{to_state}", error[1] if error else None)
             session.commit()
+            session.expire_all()  # the core UPDATE bypassed the identity map
             return self._snapshot(session, job_id)
 
     def cancel(self, job_id: str) -> dict:

photo_pipeline/services/rename_apply.py

@@ -46,9 +46,11 @@ from pathlib import Path
 from sqlalchemy import select
 from sqlalchemy.orm import sessionmaker
 
+from photo_pipeline.faults import maybe_fault
 from photo_pipeline.models import Asset, AssetPath, RenamePlan
 from photo_pipeline.services import hashing
 from photo_pipeline.services.rename_journal import (
+    ALLOWED_TRANSITIONS,
     MANUAL,
     RESUMABLE,
     JournalState,
@@ -83,18 +85,6 @@ def _now() -> datetime:
     return datetime.now(timezone.utc)
 
 
-def maybe_fault(state: str) -> None:
-    """Test-only crash barrier (concept §18 fault injection).
-
-    When ``PHOTO_PIPELINE_FAULT_AFTER`` names a journal state, the process dies
-    abruptly the moment that state has been persisted — modelling a real kill at
-    exactly that transition. Never set outside tests. Shared with the archive
-    transfer journal (US06-02), which uses the same env var and its own state names.
-    """
-    if os.environ.get("PHOTO_PIPELINE_FAULT_AFTER") == state:
-        os._exit(9)
-
-
 class RenameApplyService:
     def __init__(self, session_factory: sessionmaker, *, library_roots: tuple = ()) -> None:
         self._session_factory = session_factory
@@ -143,20 +133,10 @@ class RenameApplyService:
                 self._apply_one(operation, token=token, worker_id=worker_id)
                 applied += 1
             except PreconditionFailed as error:
-                self.journal.transition(
-                    operation["id"],
-                    JournalState.FAILED,
-                    fencing_token=token,
-                    error=(error.code, str(error)),
-                )
+                self._record_failure(operation["id"], token, error.code, str(error))
                 failed += 1
             except Exception as error:  # unexpected: record and stop touching disk
-                self.journal.transition(
-                    operation["id"],
-                    JournalState.FAILED,
-                    fencing_token=token,
-                    error=("apply_error", str(error)),
-                )
+                self._record_failure(operation["id"], token, "apply_error", str(error))
                 failed += 1
         state = self.journal.sync_plan_state(plan_id)
         return {
@@ -167,6 +147,26 @@ class RenameApplyService:
             "state": state,
         }
 
+    def _record_failure(self, operation_id: str, token: int, code: str, message: str) -> None:
+        """Record a failed operation in a state its journal can actually reach.
+
+        ``failed`` only makes sense while nothing has moved. Once the folder is at
+        its destination — a postcondition failure such as bytes edited during the
+        move — the operation is not "failed and forgotten": the disk changed and
+        the database followed, so it becomes ``rollback_required`` and waits for a
+        human (US07-04). Guessing an unreachable transition used to raise out of
+        ``apply`` and lose the record entirely.
+        """
+        current = self.journal.get(operation_id)["journal_state"]
+        target = (
+            JournalState.FAILED
+            if JournalState.FAILED in ALLOWED_TRANSITIONS.get(current, set())
+            else JournalState.ROLLBACK_REQUIRED
+        )
+        self.journal.transition(
+            operation_id, target, fencing_token=token, error=(code, message)
+        )
+
     def _apply_one(self, operation: dict, *, token: int, worker_id: str) -> None:
         source = Path(operation["source_path"])
         destination = Path(operation["destination_path"])

photo_pipeline/services/rename_journal.py

@@ -81,7 +81,17 @@ ALLOWED_TRANSITIONS = {
 
 TERMINAL_STATES = frozenset({JournalState.COMPLETE, JournalState.ROLLED_BACK})
 # States where the disk may already have been touched by this operation.
-UNSAFE_STATES = frozenset({JournalState.MOVING, JournalState.MOVED, JournalState.DATABASE_UPDATED})
+# ``rollback_required`` belongs here too (US07-04): the move happened and someone
+# has to decide what to do about it, so the library is not in a state another
+# mutation may build on.
+UNSAFE_STATES = frozenset(
+    {
+        JournalState.MOVING,
+        JournalState.MOVED,
+        JournalState.DATABASE_UPDATED,
+        JournalState.ROLLBACK_REQUIRED,
+    }
+)
 
 RESUMABLE = "resumable"
 ROLLBACK_SAFE = "rollback_safe"

photo_pipeline/services/restores.py

@@ -53,6 +53,7 @@ from sqlalchemy import select
 from sqlalchemy.orm import sessionmaker
 
 from photo_pipeline.config import Config
+from photo_pipeline.faults import maybe_fault
 from photo_pipeline.jobs.domain_handlers import ARCHIVE_LOCK, LIBRARY_WRITE_LOCK, UPLOAD_LOCK
 from photo_pipeline.models import ArchiveLocation, ArchiveOperation, ArchivePlan, Asset, AssetPath
 from photo_pipeline.path_policy import PathPolicyError, is_excluded, normalize_root, resolve_within
@@ -73,7 +74,7 @@ from photo_pipeline.services.archive_transfer import (
 from photo_pipeline.services.archives import ArchiveError
 from photo_pipeline.services.hashing import sha256_file
 from photo_pipeline.services.jobs import JobService
-from photo_pipeline.services.rename_apply import PreconditionFailed, maybe_fault
+from photo_pipeline.services.rename_apply import PreconditionFailed
 from photo_pipeline.services.rename_journal import RenameJournal
 
 PREFLIGHT_VERSION = 1

photo_pipeline/services/upload_batches.py

@@ -37,6 +37,7 @@ from sqlalchemy import select, update
 from sqlalchemy.orm import sessionmaker
 
 from photo_pipeline.config import Config
+from photo_pipeline.faults import UPLOAD_ACCEPTED, maybe_fault
 from photo_pipeline.integrations import immich_go
 from photo_pipeline.models import UploadBatch, UploadItem
 from photo_pipeline.services.hashing import sha1_file
@@ -236,6 +237,10 @@ class UploadBatchService:
             error = ("uploader_failed", f"immich-go exited with {result['exit_code']}")
             item_state = ItemState.FAILED
 
+        # The uploader is done and Immich may already hold every file, but nothing
+        # about that is durable yet — the control point for "accepted, outcome not
+        # recorded" (US07-04). Recovery must answer ``unknown_requires_verification``.
+        maybe_fault(UPLOAD_ACCEPTED)
         self._finish(batch_id, token=token, state=state, error=error, result=result)
         if item_state:
             self._set_items(batch_id, item_state)