US07-04: Prove Concurrency and Crash Recovery (#86)

2026-08-17 11:09:29 +02:00
parent 3fa35fe21e
commit d08d19c03c
21 changed files with 1740 additions and 40 deletions
--- a/tests/integration/test_concurrency_races.py
+++ b/tests/integration/test_concurrency_races.py
@@ -0,0 +1,535 @@
+"""Randomized concurrency and race tests (US07-04, concept §16 and §18).
+
+Every test here runs several times with a *recorded* seed: the seed decides the
+jitter injected around each racing operation, it is attached to the test result
+(``race_seed``), and a failing run can be replayed exactly with
+
+    PHOTO_PIPELINE_RACE_SEED=<seed> pytest tests/integration/test_concurrency_races.py
+
+``PHOTO_PIPELINE_RACE_REPEATS`` raises the repeat count for a soak run; the
+default is small enough to belong in the ordinary suite.
+
+The assertions are invariants, not schedules — a race whose interleaving decides
+the *outcome* is fine, one whose interleaving decides whether the database still
+makes sense is not:
+
+- no work is claimed, executed, or completed twice;
+- no commit from a stale fencing token lands;
+- no file is lost, overwritten, or left with foreign content;
+- a decision that changed mid-flight is never overwritten by the older answer;
+- the database passes ``PRAGMA integrity_check`` afterwards.
+"""
+
+from __future__ import annotations
+
+import os
+import random
+import subprocess
+import sys
+import threading
+import time
+import uuid
+from collections import Counter
+from concurrent.futures import ThreadPoolExecutor
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+import numpy as np
+import pytest
+from PIL import Image
+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.faults import JOB_ITEM_DONE
+from photo_pipeline.jobs.worker import Worker
+from photo_pipeline.models import (
+    AlbumProposal,
+    AnalysisResult,
+    Asset,
+    ExifProjection,
+    SafetyReview,
+)
+from photo_pipeline.services import hashing
+from photo_pipeline.services.analysis import AnalysisService
+from photo_pipeline.services.jobs import (
+    InvalidTransition,
+    ItemState,
+    JobConflict,
+    JobService,
+    JobState,
+    _now,
+)
+from photo_pipeline.services.rename_apply import RenameApplyService
+from photo_pipeline.services.renames import RenameService
+from photo_pipeline.services.thumbnails import ThumbnailError, ThumbnailService
+
+REPO = Path(__file__).resolve().parents[2]
+NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
+REPEATS = int(os.environ.get("PHOTO_PIPELINE_RACE_REPEATS", "3"))
+MAX_JITTER_SECONDS = 0.01
+
+
+# ── seeded randomness ────────────────────────────────────────────────────────
+
+
+@pytest.fixture(params=range(REPEATS), ids=lambda run: f"run{run}")
+def rng(request, record_property):
+    """A seeded RNG whose seed is recorded on the test result.
+
+    Without a pinned seed the run is genuinely random — which is the point, a
+    fixed schedule stops finding new interleavings after the first green run —
+    so the seed is reported for replay instead.
+    """
+    pinned = os.environ.get("PHOTO_PIPELINE_RACE_SEED")
+    seed = int(pinned) + request.param if pinned else random.SystemRandom().randrange(2**32)
+    record_property("race_seed", seed)
+    print(f"race seed: {seed}")  # visible with -s and in the failure report
+    return random.Random(seed)
+
+
+def jitter(rng: random.Random) -> None:
+    """Sleep a random sliver so racing threads interleave differently each run."""
+    time.sleep(rng.uniform(0, MAX_JITTER_SECONDS))
+
+
+# ── fixtures ─────────────────────────────────────────────────────────────────
+
+
+def _config(tmp_path) -> tuple[Config, Path]:
+    (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),
+        }
+    )
+    run_migrations(config.database_url)
+    return config, lib
+
+
+@pytest.fixture
+def stack(tmp_path):
+    """Config, library root, and a factory for *independent* session factories —
+    each thread must own its session (concept §16 database rule 1)."""
+    config, lib = _config(tmp_path)
+    engines = []
+
+    def factory():
+        engine = create_db_engine(config.database_url)
+        engines.append(engine)
+        return create_session_factory(engine)
+
+    yield config, lib, factory
+    for engine in engines:
+        engine.dispose()
+
+
+def image(path: Path, seed: int) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    pixels = np.random.default_rng(seed).integers(0, 256, (64, 96, 3), dtype=np.uint8)
+    Image.fromarray(pixels).save(path, quality=90)
+
+
+def register(sf, path: Path) -> str:
+    asset_id = str(uuid.uuid4())
+    with sf() as session:
+        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=hashing.sha256_file(path),
+            )
+        )
+        session.commit()
+    return asset_id
+
+
+def album(sf, lib: Path, name: str, *, approved_name: str, count: int = 2) -> list[str]:
+    """A real folder of real images with an approved proposal renaming it."""
+    folder = lib / name
+    ids = []
+    for index in range(count):
+        path = folder / f"{name}-{index}.jpg"
+        image(path, seed=abs(hash((name, index))) % 10_000)
+        ids.append(register(sf, path))
+    with sf() as session:
+        session.add(
+            AlbumProposal(
+                id=str(uuid.uuid4()),
+                album=name,
+                proposed_name=approved_name,
+                final_name=approved_name,
+                status="approved",
+                version=2,
+            )
+        )
+        session.commit()
+    return ids
+
+
+def contents(lib: Path) -> Counter:
+    """Every file body under the library — what may never be lost or duplicated."""
+    return Counter(
+        path.read_bytes()
+        for path in lib.rglob("*")
+        if path.is_file() and ".rename-" not in path.name
+    )
+
+
+def integrity_ok(sf) -> bool:
+    with sf() as session:
+        return session.execute(text("PRAGMA integrity_check")).scalar() == "ok"
+
+
+# ── database writer pressure ─────────────────────────────────────────────────
+
+
+def test_many_writers_finish_without_losing_a_row(stack, rng):
+    """Eight lanes committing at once: SQLite has one writer, so this either works
+    through the busy timeout or loses data. Nothing may be lost."""
+    _, _, factory = stack
+    service = JobService(factory())
+    job_ids = [service.enqueue("scan", items=[f"i{n}" for n in range(4)])["id"] for n in range(8)]
+
+    def write(index: int) -> None:
+        own = JobService(factory())
+        job_id = job_ids[index]
+        token = own.claim(["scan"], f"w{index}")
+        if token is None:
+            return
+        for item in [f"i{n}" for n in range(4)]:
+            jitter(rng)
+            own.set_item(token["id"], item, ItemState.RUNNING, fencing_token=token["fencing_token"])
+            own.set_item(
+                token["id"], item, ItemState.SUCCEEDED, fencing_token=token["fencing_token"]
+            )
+        assert job_id  # the claim order is racy; every job is claimed by someone
+
+    with ThreadPoolExecutor(max_workers=8) as pool:
+        list(pool.map(write, range(8)))
+
+    for job_id in job_ids:
+        progress = service.progress(job_id)
+        assert progress["by_state"].get(ItemState.SUCCEEDED) == 4, progress
+    assert integrity_ok(factory())
+
+
+# ── worker claim races ───────────────────────────────────────────────────────
+
+
+def test_a_job_is_claimed_by_exactly_one_worker(stack, rng):
+    _, _, factory = stack
+    service = JobService(factory())
+    job_ids = {service.enqueue("scan", items=["a"])["id"] for _ in range(10)}
+
+    claimed: list[str] = []
+    lock = threading.Lock()
+
+    def claim_all(index: int) -> None:
+        own = JobService(factory())
+        while True:
+            jitter(rng)
+            job = own.claim(["scan"], f"w{index}")
+            if job is None:
+                return
+            with lock:
+                claimed.append(job["id"])
+
+    with ThreadPoolExecutor(max_workers=4) as pool:
+        list(pool.map(claim_all, range(4)))
+
+    assert sorted(claimed) == sorted(job_ids), "every job claimed exactly once"
+    assert len(set(claimed)) == len(claimed)
+
+
+def test_a_late_commit_from_an_expired_lease_is_refused(stack, rng):
+    """Lease expiry then a late write from the old owner: the fencing token, not
+    timing, decides who may commit."""
+    _, _, factory = stack
+    service = JobService(factory())
+    job = service.enqueue("scan", items=["a"])
+    old = service.claim(["scan"], "old")
+    service.recover_stale(now=_now() + timedelta(hours=1))
+    new = service.claim(["scan"], "new")
+
+    jitter(rng)
+    with pytest.raises(JobConflict):
+        service.set_item(job["id"], "a", ItemState.RUNNING, fencing_token=old["fencing_token"])
+    with pytest.raises(JobConflict):
+        service.transition(job["id"], JobState.SUCCEEDED, fencing_token=old["fencing_token"])
+
+    # The current owner finishes the same work without interference.
+    service.set_item(job["id"], "a", ItemState.RUNNING, fencing_token=new["fencing_token"])
+    service.set_item(job["id"], "a", ItemState.SUCCEEDED, fencing_token=new["fencing_token"])
+    service.transition(job["id"], JobState.SUCCEEDED, fencing_token=new["fencing_token"])
+    assert service.get(job["id"])["state"] == JobState.SUCCEEDED
+    assert integrity_ok(factory())
+
+
+def test_cancellation_arriving_at_a_random_moment_leaves_a_consistent_job(stack, rng):
+    _, _, factory = stack
+    service = JobService(factory())
+    job = service.enqueue("scan", items=[f"i{n}" for n in range(6)])
+    handled: list[str] = []
+
+    def handler(item, ctx):
+        jitter(rng)
+        handled.append(item)
+
+    def cancel() -> None:
+        try:
+            service.cancel(job["id"])
+        except (JobConflict, InvalidTransition):
+            # The request lost its race with the worker's own transition; the job
+            # simply finishes. What must never happen is a *silent* overwrite.
+            pass
+
+    canceller = threading.Timer(rng.uniform(0, 0.03), cancel)
+    canceller.start()
+    Worker(factory(), {"scan": handler}, "w1").run_once()
+    canceller.join()
+
+    state = service.get(job["id"])["state"]
+    assert state in {JobState.SUCCEEDED, JobState.CANCELLED, JobState.CANCELLING}
+    by_state = service.progress(job["id"])["by_state"]
+    assert by_state.get(ItemState.RUNNING, 0) == 0, "no item left mid-flight"
+    assert len(handled) == len(set(handled)), "no item ran twice"
+    if state == JobState.CANCELLED:
+        # Whatever did not run is resumable, never silently dropped.
+        assert by_state.get(ItemState.SUCCEEDED, 0) + by_state.get(ItemState.QUEUED, 0) == 6
+    assert integrity_ok(factory())
+
+
+KILLABLE_WORKER = """
+import sys
+sys.path.insert(0, {repo!r})
+from pathlib import Path
+from photo_pipeline.db import create_db_engine, create_session_factory
+from photo_pipeline.jobs.worker import Worker
+
+db_url, log = sys.argv[1], Path(sys.argv[2])
+
+def handler(item_key, ctx):
+    with log.open("a") as handle:
+        handle.write(item_key + "\\n")
+
+sf = create_session_factory(create_db_engine(db_url))
+Worker(sf, {{"scan": handler}}, sys.argv[3], lease_seconds=1).run_once()
+"""
+
+
+def test_a_worker_killed_at_a_random_item_resumes_exactly_once(stack, rng, tmp_path):
+    """Kill a real worker mid-batch, restart it, and assert the invariant across
+    the restart: every item ran exactly once and the job ends succeeded."""
+    config, _, factory = stack
+    service = JobService(factory())
+    items = [f"i{n}" for n in range(rng.randint(2, 5))]
+    job = service.enqueue("scan", items=items)
+    log = tmp_path / "handled.log"
+    script = tmp_path / "killable_worker.py"
+    script.write_text(KILLABLE_WORKER.format(repo=str(REPO)))
+
+    def run_worker(worker_id: str, barrier: str | None) -> None:
+        env = dict(os.environ)
+        env.pop("PHOTO_PIPELINE_FAULT_AFTER", None)
+        if barrier:
+            env["PHOTO_PIPELINE_FAULT_AFTER"] = barrier
+        subprocess.run(
+            [sys.executable, str(script), config.database_url, str(log), worker_id],
+            env=env,
+            capture_output=True,
+        )
+
+    run_worker("killable", JOB_ITEM_DONE)  # dies after its first completed item
+    assert log.read_text().split() == items[:1]
+    time.sleep(1.1)  # let the dead worker's one-second lease expire
+    jitter(rng)
+    run_worker("survivor", None)
+
+    assert service.get(job["id"])["state"] == JobState.SUCCEEDED
+    handled = log.read_text().split()
+    assert sorted(handled) == sorted(items), f"an item ran twice or not at all: {handled}"
+    assert integrity_ok(factory())
+
+
+# ── filesystem races ─────────────────────────────────────────────────────────
+
+
+def test_a_file_changed_during_apply_never_loses_content(stack, rng):
+    """The user edits a photo while its folder is being renamed. The rename may
+    win or be refused, but no body may disappear or be overwritten."""
+    config, lib, factory = stack
+    sf = factory()
+    album(sf, lib, "rome", approved_name="2019 Rome")
+    plan = RenameService(sf, library_roots=(lib,)).build_plan()
+    before = contents(lib)
+
+    edited = b"the user saved over this file"
+
+    def edit() -> None:
+        jitter(rng)
+        target = next((lib / "rome").glob("*.jpg"), None)
+        if target is None:
+            return  # the rename won the race; the folder already moved
+        try:
+            target.write_bytes(edited)
+        except OSError:
+            pass  # ...or it moved between the glob and the write
+
+    thread = threading.Thread(target=edit)
+    thread.start()
+    RenameApplyService(sf, library_roots=(lib,)).apply(plan["id"], expected_version=plan["version"])
+    thread.join()
+
+    after = contents(lib)
+    assert sum(after.values()) == sum(before.values()), "a file was lost or duplicated"
+    survived = (before - Counter({edited: 1})) & after
+    assert sum(survived.values()) >= sum(before.values()) - 1, "unrelated content was destroyed"
+    assert integrity_ok(sf)
+
+
+def test_two_folders_claiming_one_destination_never_merge(stack, rng):
+    """Two approved albums want the same name. The plan must refuse rather than
+    move one folder into the other."""
+    config, lib, factory = stack
+    sf = factory()
+    album(sf, lib, "rome-a", approved_name="2019 Rome")
+    album(sf, lib, "rome-b", approved_name="2019 Rome")
+    jitter(rng)
+
+    plan = RenameService(sf, library_roots=(lib,)).build_plan()
+    codes = {issue["code"] for op in plan["operations"] for issue in op["issues"]}
+
+    assert "duplicate_target" in codes
+    assert plan["applicable"] is False
+    assert (lib / "rome-a").is_dir() and (lib / "rome-b").is_dir()
+
+
+def test_thumbnail_requests_racing_a_rename_never_serve_a_wrong_file(stack, rng):
+    """Previews are keyed by pixels, not paths, so a rename must not make a request
+    fail loudly *or* return another asset's picture."""
+    config, lib, factory = stack
+    sf = factory()
+    asset_ids = album(sf, lib, "rome", approved_name="2019 Rome")
+    plan = RenameService(sf, library_roots=(lib,)).build_plan()
+
+    thumbnails = ThumbnailService(factory(), config)
+    expected = {aid: thumbnails.generate(aid, 256).read_bytes() for aid in asset_ids}
+    served: dict[str, set[bytes]] = {aid: set() for aid in asset_ids}
+    errors: list[str] = []
+    stop = threading.Event()
+
+    def serve() -> None:
+        own = ThumbnailService(factory(), config)
+        while not stop.is_set():
+            for asset_id in asset_ids:
+                jitter(rng)
+                try:
+                    served[asset_id].add(own.generate(asset_id, 256).read_bytes())
+                except ThumbnailError as error:
+                    errors.append(error.code)  # precise, never an unhandled crash
+
+    reader = threading.Thread(target=serve)
+    reader.start()
+    try:
+        RenameApplyService(sf, library_roots=(lib,)).apply(
+            plan["id"], expected_version=plan["version"]
+        )
+    finally:
+        stop.set()
+        reader.join()
+
+    for asset_id, bodies in served.items():
+        assert bodies <= {expected[asset_id]}, "a request served another asset's picture"
+    assert integrity_ok(sf)
+
+
+# ── stage races ──────────────────────────────────────────────────────────────
+
+
+def test_an_analysis_result_racing_a_safety_flip_is_discarded(stack, rng):
+    """Concept §18 scenario 7: the reviewer marks an asset NSFW while the provider
+    call is in flight. The answer that comes back describes an asset that may no
+    longer be analysed, so it is dropped — and no analysis EXIF is written."""
+    _, lib, factory = stack
+    sf = factory()
+    path = lib / "beach.jpg"
+    image(path, seed=7)
+    asset_id = register(sf, path)
+    with sf() as session:
+        session.add(
+            SafetyReview(id=str(uuid.uuid4()), asset_id=asset_id, decision="sfw", created_at=NOW)
+        )
+        session.commit()
+    before_sha = hashing.sha256_file(path)
+
+    class FlippingProvider:
+        """Records the call, then the reviewer's decision lands mid-flight."""
+
+        def __init__(self) -> None:
+            self.calls: list[str] = []
+
+        def analyze(self, path, *, album_hint):
+            self.calls.append(path)
+            jitter(rng)
+            with sf() as session:
+                session.add(
+                    SafetyReview(
+                        id=str(uuid.uuid4()),
+                        asset_id=asset_id,
+                        decision="nsfw",
+                        prior_decision="sfw",
+                        created_at=NOW + timedelta(minutes=1),
+                    )
+                )
+                session.commit()
+            return {"description": "a beach", "tags": ["beach", "sand"]}
+
+    provider = FlippingProvider()
+    result = AnalysisService(sf, provider=provider, library_roots=(lib,)).run([asset_id])
+
+    assert provider.calls == [str(path)], "the call was legitimate when it started"
+    assert result == {"analyzed": 0, "skipped": 1, "errors": 0}
+    with sf() as session:
+        row = session.get(AnalysisResult, asset_id)
+        assert row.status == "skipped_nsfw" and row.description is None
+        assert session.get(ExifProjection, (asset_id, "analysis")) is None
+    assert hashing.sha256_file(path) == before_sha, "the file was written after the flip"
+
+
+def test_a_safety_decision_taken_twice_at_once_keeps_one_history(stack, rng):
+    """Two windows, one asset, the same decision: the audit trail may record both
+    attempts, but the effective state must be a single coherent decision."""
+    _, lib, factory = stack
+    sf = factory()
+    path = lib / "twice.jpg"
+    image(path, seed=11)
+    asset_id = register(sf, path)
+
+    from photo_pipeline.services.safety import SafetyService
+
+    def decide(decision: str) -> None:
+        jitter(rng)
+        SafetyService(factory()).decide(asset_id, decision, write_exif=False)
+
+    with ThreadPoolExecutor(max_workers=2) as pool:
+        list(pool.map(decide, ["sfw", "sfw"]))
+
+    with sf() as session:
+        decisions = [
+            review.decision
+            for review in session.scalars(
+                select(SafetyReview)
+                .where(SafetyReview.asset_id == asset_id)
+                .order_by(SafetyReview.created_at)
+            )
+        ]
+    assert decisions and set(decisions) == {"sfw"}
+    assert integrity_ok(sf)