From 020726c718d5a7a445fab0d3fbd47aa10f9b0a5c Mon Sep 17 00:00:00 2001 From: domverse Date: Wed, 15 Jul 2026 22:46:45 +0200 Subject: [PATCH] US02-02: Persist and Coordinate Durable Jobs --- migrations/versions/0004_durable_jobs.py | 85 ++++++ photo_pipeline/models/__init__.py | 4 + photo_pipeline/models/jobs.py | 71 +++++ photo_pipeline/services/jobs.py | 359 +++++++++++++++++++++++ tests/integration/test_jobs.py | 147 ++++++++++ tests/story_traceability.json | 4 + tests/unit/test_job_state_machine.py | 51 ++++ 7 files changed, 721 insertions(+) create mode 100644 migrations/versions/0004_durable_jobs.py create mode 100644 photo_pipeline/models/jobs.py create mode 100644 photo_pipeline/services/jobs.py create mode 100644 tests/integration/test_jobs.py create mode 100644 tests/unit/test_job_state_machine.py diff --git a/migrations/versions/0004_durable_jobs.py b/migrations/versions/0004_durable_jobs.py new file mode 100644 index 0000000..de9bdfb --- /dev/null +++ b/migrations/versions/0004_durable_jobs.py @@ -0,0 +1,85 @@ +"""Durable jobs: jobs, job_items, job_events. + +Revision ID: 0004_durable_jobs +Revises: 0003_thumbnails +Create Date: 2026-07-15 +""" + +import sqlalchemy as sa +from alembic import op + +revision = "0004_durable_jobs" +down_revision = "0003_thumbnails" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "jobs", + sa.Column("id", sa.String(), primary_key=True), + sa.Column("job_type", sa.String(), nullable=False), + sa.Column("state", sa.String(), nullable=False), + sa.Column("lock_key", sa.String(), nullable=True), + sa.Column("idempotency_key", sa.String(), nullable=True, unique=True), + sa.Column("lease_owner", sa.String(), nullable=True), + sa.Column("lease_expires_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("heartbeat_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("fencing_token", sa.Integer(), nullable=False, server_default="0"), + sa.Column("error_code", sa.String(), nullable=True), + sa.Column("error_message", sa.String(), 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"), + ), + sa.Column("started_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("version", sa.Integer(), nullable=False, server_default="1"), + ) + op.create_index("ix_jobs_state", "jobs", ["state"]) + op.create_index("ix_jobs_lock_key", "jobs", ["lock_key"]) + + op.create_table( + "job_items", + sa.Column("job_id", sa.String(), sa.ForeignKey("jobs.id"), primary_key=True), + sa.Column("item_key", sa.String(), primary_key=True), + sa.Column("state", sa.String(), nullable=False), + sa.Column("attempt_count", sa.Integer(), nullable=False, server_default="0"), + sa.Column("error_code", sa.String(), nullable=True), + sa.Column("error_message", sa.String(), nullable=True), + sa.Column( + "updated_at", + sa.DateTime(timezone=True), + nullable=False, + server_default=sa.text("CURRENT_TIMESTAMP"), + ), + ) + + op.create_table( + "job_events", + sa.Column("id", sa.String(), primary_key=True), + sa.Column("job_id", sa.String(), sa.ForeignKey("jobs.id"), nullable=False), + sa.Column("event_type", sa.String(), nullable=False), + sa.Column("message", sa.String(), nullable=True), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + nullable=False, + server_default=sa.text("CURRENT_TIMESTAMP"), + ), + ) + op.create_index("ix_job_events_job_id", "job_events", ["job_id"]) + + +def downgrade() -> None: + op.drop_table("job_events") + op.drop_table("job_items") + op.drop_table("jobs") diff --git a/photo_pipeline/models/__init__.py b/photo_pipeline/models/__init__.py index 4d4e911..392a45e 100644 --- a/photo_pipeline/models/__init__.py +++ b/photo_pipeline/models/__init__.py @@ -10,6 +10,7 @@ from photo_pipeline.models.duplicates import ( DuplicateMember, DuplicateNegativeLink, ) +from photo_pipeline.models.jobs import Job, JobEvent, JobItem from photo_pipeline.models.thumbnails import Thumbnail __all__ = [ @@ -18,5 +19,8 @@ __all__ = [ "DuplicateCluster", "DuplicateMember", "DuplicateNegativeLink", + "Job", + "JobItem", + "JobEvent", "Thumbnail", ] diff --git a/photo_pipeline/models/jobs.py b/photo_pipeline/models/jobs.py new file mode 100644 index 0000000..a84def7 --- /dev/null +++ b/photo_pipeline/models/jobs.py @@ -0,0 +1,71 @@ +"""Durable jobs, their per-item progress, and their event history. + +A job is a unit of long-running work whose lifecycle survives browser, API, and +worker interruption. Progress is derived from ``job_items`` (the durable source of +truth), never from browser memory. ``job_events`` is an append-only activity log. +Leases (owner + expiry + heartbeat) and a monotonic ``fencing_token`` let a +recovered worker be fenced out after a newer worker takes ownership. +""" + +from __future__ import annotations + +from datetime import datetime + +from sqlalchemy import DateTime, ForeignKey, Integer, String, func +from sqlalchemy.orm import Mapped, mapped_column + +from photo_pipeline.db import Base + + +class Job(Base): + __tablename__ = "jobs" + + id: Mapped[str] = mapped_column(String, primary_key=True) + job_type: Mapped[str] = mapped_column(String, nullable=False) + state: Mapped[str] = mapped_column(String, nullable=False) + lock_key: Mapped[str | None] = mapped_column(String) + idempotency_key: Mapped[str | None] = mapped_column(String, unique=True) + + lease_owner: Mapped[str | None] = mapped_column(String) + lease_expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + heartbeat_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + fencing_token: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + + error_code: Mapped[str | None] = mapped_column(String) + error_message: Mapped[str | None] = mapped_column(String) + + 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() + ) + started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + version: Mapped[int] = mapped_column(Integer, nullable=False, default=1) + + +class JobItem(Base): + __tablename__ = "job_items" + + job_id: Mapped[str] = mapped_column(ForeignKey("jobs.id"), primary_key=True) + item_key: Mapped[str] = mapped_column(String, primary_key=True) + state: Mapped[str] = mapped_column(String, nullable=False) + attempt_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + error_code: Mapped[str | None] = mapped_column(String) + error_message: Mapped[str | None] = mapped_column(String) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now() + ) + + +class JobEvent(Base): + __tablename__ = "job_events" + + id: Mapped[str] = mapped_column(String, primary_key=True) + job_id: Mapped[str] = mapped_column(ForeignKey("jobs.id"), nullable=False) + event_type: Mapped[str] = mapped_column(String, nullable=False) + message: Mapped[str | None] = mapped_column(String) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) diff --git a/photo_pipeline/services/jobs.py b/photo_pipeline/services/jobs.py new file mode 100644 index 0000000..f750ba9 --- /dev/null +++ b/photo_pipeline/services/jobs.py @@ -0,0 +1,359 @@ +"""JobService — durable job coordinator (no execution; the worker lands in US02-03). + +Responsibilities: enqueue with idempotency keys and one-mutating-job locks, hand a +queued job to a worker atomically (compare-and-set claim with a fencing token), +track heartbeats/leases, validate every state transition, derive progress from +durable item state, request cancellation, and recover jobs whose worker vanished. + +State machine (concept §3): + + queued ─▶ running ─▶ succeeded + │ │ ╲──▶ failed ─▶ retry_queued ─▶ running + │ ╲──▶ cancelling ─▶ cancelled + ╰──▶ cancelled + +Terminal states (succeeded, cancelled) have no outgoing transition, so a duplicate +terminal transition is rejected. +""" + +from __future__ import annotations + +import uuid +from collections import Counter +from datetime import datetime, timedelta, timezone +from collections.abc import Iterable, Sequence + +from sqlalchemy import select, update +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import sessionmaker + +from photo_pipeline.models import Job, JobEvent, JobItem + + +class JobState: + QUEUED = "queued" + RUNNING = "running" + CANCELLING = "cancelling" + CANCELLED = "cancelled" + SUCCEEDED = "succeeded" + FAILED = "failed" + RETRY_QUEUED = "retry_queued" + + +class ItemState: + QUEUED = "queued" + RUNNING = "running" + SUCCEEDED = "succeeded" + FAILED = "failed" + CANCELLED = "cancelled" + + +# Jobs that occupy a lock or are eligible to run are "active" for lock/one-job checks. +ACTIVE_STATES = frozenset( + {JobState.QUEUED, JobState.RUNNING, JobState.CANCELLING, JobState.RETRY_QUEUED} +) +TERMINAL_STATES = frozenset({JobState.SUCCEEDED, JobState.CANCELLED}) +CLAIMABLE_STATES = frozenset({JobState.QUEUED, JobState.RETRY_QUEUED}) + +ALLOWED_TRANSITIONS = { + JobState.QUEUED: {JobState.RUNNING, JobState.CANCELLED, JobState.CANCELLING}, + JobState.RUNNING: {JobState.SUCCEEDED, JobState.FAILED, JobState.CANCELLING}, + JobState.CANCELLING: {JobState.CANCELLED, JobState.SUCCEEDED, JobState.FAILED}, + JobState.FAILED: {JobState.RETRY_QUEUED}, + JobState.RETRY_QUEUED: {JobState.RUNNING, JobState.CANCELLED, JobState.CANCELLING}, + JobState.SUCCEEDED: set(), + JobState.CANCELLED: set(), +} + +ITEM_TRANSITIONS = { + ItemState.QUEUED: {ItemState.RUNNING, ItemState.CANCELLED}, + ItemState.RUNNING: {ItemState.SUCCEEDED, ItemState.FAILED, ItemState.CANCELLED}, + ItemState.FAILED: {ItemState.QUEUED, ItemState.RUNNING}, + ItemState.SUCCEEDED: set(), + ItemState.CANCELLED: set(), +} + + +def can_transition(current: str, target: str) -> bool: + return target in ALLOWED_TRANSITIONS.get(current, set()) + + +class JobError(RuntimeError): + pass + + +class JobBlocked(JobError): + """Enqueue refused by a prerequisite or the one-mutating-job lock policy.""" + + def __init__(self, code: str, detail: str = "") -> None: + super().__init__(f"{code}: {detail}" if detail else code) + self.code = code + self.detail = detail + + +class InvalidTransition(JobError): + pass + + +class JobConflict(JobError): + """Optimistic/fencing check failed; another owner or version won the race.""" + + +def _now() -> datetime: + return datetime.now(timezone.utc) + + +class JobService: + def __init__(self, session_factory: sessionmaker) -> None: + self._session_factory = session_factory + + # ── enqueue ──────────────────────────────────────────────────────────────── + def enqueue( + self, + job_type: str, + *, + lock: str | None = None, + idempotency_key: str | None = None, + items: Iterable[str] = (), + ) -> dict: + with self._session_factory() as session: + if idempotency_key: + existing = session.scalar( + select(Job).where(Job.idempotency_key == idempotency_key) + ) + if existing: + return self._snapshot(session, existing.id) + if lock: + blocker = session.scalar( + select(Job).where(Job.lock_key == lock, Job.state.in_(ACTIVE_STATES)) + ) + if blocker: + raise JobBlocked("lock_held", f"{lock} held by job {blocker.id}") + + job = Job( + id=str(uuid.uuid4()), + job_type=job_type, + state=JobState.QUEUED, + lock_key=lock, + idempotency_key=idempotency_key, + fencing_token=0, + version=1, + ) + session.add(job) + for item_key in items: + session.add( + JobItem(job_id=job.id, item_key=item_key, state=ItemState.QUEUED) + ) + self._event(session, job.id, "queued") + try: + session.commit() + except IntegrityError: + # Concurrent enqueue with the same idempotency key won the insert. + session.rollback() + existing = session.scalar( + select(Job).where(Job.idempotency_key == idempotency_key) + ) + if existing: + return self._snapshot(session, existing.id) + raise + return self._snapshot(session, job.id) + + # ── claim / lease ────────────────────────────────────────────────────────── + def claim( + self, job_types: Sequence[str], worker_id: str, *, lease_seconds: int = 60 + ) -> dict | None: + now = _now() + expires = now + timedelta(seconds=lease_seconds) + with self._session_factory() as session: + candidates = ( + session.execute( + select(Job) + .where(Job.state.in_(CLAIMABLE_STATES), Job.job_type.in_(list(job_types))) + .order_by(Job.created_at) + ) + .scalars() + .all() + ) + for job in candidates: + # Compare-and-set: only the worker whose UPDATE matches the observed + # state wins; concurrent claimers see rowcount 0 and try the next job. + result = session.execute( + update(Job) + .where(Job.id == job.id, Job.state == job.state) + .values( + state=JobState.RUNNING, + lease_owner=worker_id, + lease_expires_at=expires, + heartbeat_at=now, + started_at=job.started_at or now, + fencing_token=Job.fencing_token + 1, + version=Job.version + 1, + updated_at=now, + ) + ) + if result.rowcount == 1: + self._event(session, job.id, "claimed", worker_id) + session.commit() + return self._snapshot(session, job.id) + return None + + def heartbeat(self, job_id: str, worker_id: str, *, lease_seconds: int = 60) -> bool: + now = _now() + with self._session_factory() as session: + result = session.execute( + update(Job) + .where( + Job.id == job_id, + Job.lease_owner == worker_id, + Job.state == JobState.RUNNING, + ) + .values(heartbeat_at=now, lease_expires_at=now + timedelta(seconds=lease_seconds)) + ) + session.commit() + return result.rowcount == 1 + + # ── transitions ──────────────────────────────────────────────────────────── + def transition( + self, + job_id: str, + to_state: str, + *, + worker_id: str | None = None, + error: tuple[str, str] | None = None, + ) -> dict: + now = _now() + with self._session_factory() as session: + job = session.get(Job, job_id) + if job is None: + raise JobError(f"unknown job {job_id}") + if not can_transition(job.state, to_state): + 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 + if error: + job.error_code, job.error_message = error + if to_state in TERMINAL_STATES: + job.finished_at = now + job.lease_owner = None + job.lease_expires_at = None + self._event(session, job_id, f"state:{to_state}", error[1] if error else None) + session.commit() + return self._snapshot(session, job_id) + + def cancel(self, job_id: str) -> dict: + with self._session_factory() as session: + job = session.get(Job, job_id) + if job is None: + raise JobError(f"unknown job {job_id}") + # queued work can be cancelled outright; running work drains via cancelling. + target = ( + JobState.CANCELLED + if job.state in (JobState.QUEUED, JobState.RETRY_QUEUED) + else JobState.CANCELLING + ) + return self.transition(job_id, target) + + # ── items / progress ─────────────────────────────────────────────────────── + def set_item( + self, job_id: str, item_key: str, state: str, *, error: tuple[str, str] | None = None + ) -> None: + now = _now() + with self._session_factory() as session: + item = session.get(JobItem, {"job_id": job_id, "item_key": item_key}) + if item is None: + raise JobError(f"unknown item {job_id}/{item_key}") + if state != item.state and state not in ITEM_TRANSITIONS.get(item.state, set()): + raise InvalidTransition(f"item {item.state} -> {state}") + if state == ItemState.RUNNING and item.state != ItemState.RUNNING: + item.attempt_count += 1 + item.state = state + item.updated_at = now + if error: + item.error_code, item.error_message = error + session.commit() + + def progress(self, job_id: str) -> dict: + with self._session_factory() as session: + states = session.execute( + select(JobItem.state).where(JobItem.job_id == job_id) + ).scalars().all() + counts = Counter(states) + total = len(states) + done = counts.get(ItemState.SUCCEEDED, 0) + counts.get(ItemState.FAILED, 0) + counts.get( + ItemState.CANCELLED, 0 + ) + return {"total": total, "done": done, "by_state": dict(counts)} + + # ── recovery ─────────────────────────────────────────────────────────────── + def recover_stale(self, *, now: datetime | None = None) -> list[str]: + """Return running jobs whose lease expired to ``retry_queued`` for safe resume.""" + now = now or _now() + recovered: list[str] = [] + with self._session_factory() as session: + stale = ( + session.execute( + select(Job).where( + Job.state == JobState.RUNNING, Job.lease_expires_at < now + ) + ) + .scalars() + .all() + ) + for job in stale: + job.state = JobState.RETRY_QUEUED + job.lease_owner = None + job.lease_expires_at = None + job.version += 1 + job.updated_at = now + self._event(session, job.id, "interrupted", "lease expired") + recovered.append(job.id) + session.commit() + return recovered + + def blockers(self, lock: str) -> list[dict]: + with self._session_factory() as session: + held = ( + session.execute( + select(Job).where(Job.lock_key == lock, Job.state.in_(ACTIVE_STATES)) + ) + .scalars() + .all() + ) + return [{"id": j.id, "state": j.state, "job_type": j.job_type} for j in held] + + def get(self, job_id: str) -> dict | None: + with self._session_factory() as session: + if session.get(Job, job_id) is None: + return None + return self._snapshot(session, job_id) + + def events(self, job_id: str) -> list[dict]: + with self._session_factory() as session: + rows = session.execute( + select(JobEvent).where(JobEvent.job_id == job_id).order_by(JobEvent.created_at) + ).scalars() + return [{"type": e.event_type, "message": e.message} for e in rows] + + # ── helpers ────────────────────────────────────────────────────────────────── + @staticmethod + def _event(session, job_id: str, event_type: str, message: str | None = None) -> None: + session.add( + JobEvent(id=str(uuid.uuid4()), job_id=job_id, event_type=event_type, message=message) + ) + + def _snapshot(self, session, job_id: str) -> dict: + job = session.get(Job, job_id) + return { + "id": job.id, + "job_type": job.job_type, + "state": job.state, + "lock_key": job.lock_key, + "idempotency_key": job.idempotency_key, + "lease_owner": job.lease_owner, + "fencing_token": job.fencing_token, + "version": job.version, + "error_code": job.error_code, + } diff --git a/tests/integration/test_jobs.py b/tests/integration/test_jobs.py new file mode 100644 index 0000000..c9d2051 --- /dev/null +++ b/tests/integration/test_jobs.py @@ -0,0 +1,147 @@ +"""Durable job coordinator: idempotent submission, atomic claim, locks, progress +from durable item state, cancellation, stale recovery, and restart persistence.""" + +from concurrent.futures import ThreadPoolExecutor +from datetime import timedelta + +import pytest + +from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations +from photo_pipeline.services.jobs import ( + ItemState, + JobBlocked, + JobService, + JobState, + _now, +) + + +@pytest.fixture +def db_url(tmp_path): + url = f"sqlite:///{tmp_path / 'jobs.db'}" + run_migrations(url) + return url + + +@pytest.fixture +def make_factory(): + engines = [] + + def _make(url): + engine = create_db_engine(url) + engines.append(engine) + return create_session_factory(engine) + + yield _make + for engine in engines: + engine.dispose() + + +@pytest.fixture +def jobs(db_url, make_factory): + return JobService(make_factory(db_url)) + + +def test_enqueue_persists_items_and_event(jobs): + job = jobs.enqueue("scan", items=["a", "b", "c"]) + assert job["state"] == JobState.QUEUED + assert jobs.progress(job["id"]) == { + "total": 3, + "done": 0, + "by_state": {ItemState.QUEUED: 3}, + } + assert any(e["type"] == "queued" for e in jobs.events(job["id"])) + + +def test_idempotency_key_returns_same_job(jobs): + a = jobs.enqueue("scan", idempotency_key="key-1") + b = jobs.enqueue("scan", idempotency_key="key-1") + assert a["id"] == b["id"] + + +def test_concurrent_idempotent_submission_creates_one_job(db_url, make_factory): + service = JobService(make_factory(db_url)) + with ThreadPoolExecutor(max_workers=8) as pool: + ids = list( + pool.map(lambda _: service.enqueue("scan", idempotency_key="race")["id"], range(8)) + ) + assert len(set(ids)) == 1 + + +def test_one_mutating_job_lock_blocks(jobs): + jobs.enqueue("scan", lock="library_write") + with pytest.raises(JobBlocked) as excinfo: + jobs.enqueue("rename", lock="library_write") + assert excinfo.value.code == "lock_held" + assert jobs.blockers("library_write") + + +def test_lock_frees_after_terminal(jobs): + first = jobs.enqueue("scan", lock="library_write") + jobs.claim(["scan"], "w1") + jobs.transition(first["id"], JobState.SUCCEEDED, worker_id="w1") + # Lock is free now; a new mutating job enqueues fine. + assert jobs.enqueue("rename", lock="library_write")["state"] == JobState.QUEUED + + +def test_atomic_claim_gives_job_to_one_worker(db_url, make_factory): + service = JobService(make_factory(db_url)) + service.enqueue("scan") + with ThreadPoolExecutor(max_workers=4) as pool: + claims = list(pool.map(lambda i: service.claim(["scan"], f"w{i}"), range(4))) + winners = [c for c in claims if c is not None] + assert len(winners) == 1 + assert winners[0]["state"] == JobState.RUNNING + assert winners[0]["fencing_token"] == 1 + + +def test_progress_derives_from_item_state(jobs): + job = jobs.enqueue("scan", items=["a", "b"]) + jobs.set_item(job["id"], "a", ItemState.RUNNING) + jobs.set_item(job["id"], "a", ItemState.SUCCEEDED) + jobs.set_item(job["id"], "b", ItemState.RUNNING) + jobs.set_item(job["id"], "b", ItemState.FAILED, error=("boom", "kaboom")) + progress = jobs.progress(job["id"]) + assert progress["total"] == 2 and progress["done"] == 2 + assert progress["by_state"] == {ItemState.SUCCEEDED: 1, ItemState.FAILED: 1} + + +def test_cancel_running_job_drains_via_cancelling(jobs): + job = jobs.enqueue("scan") + jobs.claim(["scan"], "w1") + after = jobs.cancel(job["id"]) + assert after["state"] == JobState.CANCELLING + + +def test_cancel_queued_job_is_immediate(jobs): + job = jobs.enqueue("scan") + assert jobs.cancel(job["id"])["state"] == JobState.CANCELLED + + +def test_recover_stale_requeues_expired_leases(jobs): + job = jobs.enqueue("scan") + jobs.claim(["scan"], "w1", lease_seconds=30) + recovered = jobs.recover_stale(now=_now() + timedelta(hours=1)) + assert job["id"] in recovered + assert jobs.get(job["id"])["state"] == JobState.RETRY_QUEUED + assert jobs.get(job["id"])["lease_owner"] is None + + +def test_heartbeat_only_from_lease_owner(jobs): + job = jobs.enqueue("scan") + jobs.claim(["scan"], "w1") + assert jobs.heartbeat(job["id"], "w1") is True + assert jobs.heartbeat(job["id"], "intruder") is False + + +def test_state_durable_across_restart(db_url, make_factory): + first = JobService(make_factory(db_url)) + job = first.enqueue("scan", items=["a", "b"]) + first.claim(["scan"], "w1") + first.set_item(job["id"], "a", ItemState.RUNNING) + first.set_item(job["id"], "a", ItemState.SUCCEEDED) + + # Fresh service/engine against the same database. + reopened = JobService(make_factory(db_url)) + assert reopened.get(job["id"])["state"] == JobState.RUNNING + assert reopened.progress(job["id"])["done"] == 1 diff --git a/tests/story_traceability.json b/tests/story_traceability.json index 852912c..c75c468 100644 --- a/tests/story_traceability.json +++ b/tests/story_traceability.json @@ -44,6 +44,10 @@ "tests/unit/test_safety.py", "tests/unit/test_design_tokens.py", "tests/integration/test_safety_parity.py" + ], + "US02-02": [ + "tests/unit/test_job_state_machine.py", + "tests/integration/test_jobs.py" ] } } diff --git a/tests/unit/test_job_state_machine.py b/tests/unit/test_job_state_machine.py new file mode 100644 index 0000000..d3c40d9 --- /dev/null +++ b/tests/unit/test_job_state_machine.py @@ -0,0 +1,51 @@ +"""Job/item state-machine rules: only declared transitions are legal, and terminal +states are truly terminal (no duplicate terminal transition).""" + +import pytest + +from photo_pipeline.services.jobs import ( + ALLOWED_TRANSITIONS, + ITEM_TRANSITIONS, + TERMINAL_STATES, + ItemState, + JobState, + can_transition, +) + + +def test_happy_path_transitions_are_allowed(): + assert can_transition(JobState.QUEUED, JobState.RUNNING) + assert can_transition(JobState.RUNNING, JobState.SUCCEEDED) + assert can_transition(JobState.RUNNING, JobState.FAILED) + assert can_transition(JobState.FAILED, JobState.RETRY_QUEUED) + assert can_transition(JobState.RETRY_QUEUED, JobState.RUNNING) + assert can_transition(JobState.RUNNING, JobState.CANCELLING) + assert can_transition(JobState.CANCELLING, JobState.CANCELLED) + + +@pytest.mark.parametrize( + "current,target", + [ + (JobState.QUEUED, JobState.SUCCEEDED), # can't skip running + (JobState.SUCCEEDED, JobState.RUNNING), # terminal, no revive + (JobState.CANCELLED, JobState.RUNNING), # terminal + (JobState.RUNNING, JobState.QUEUED), # no backward + ], +) +def test_invalid_transitions_are_rejected(current, target): + assert not can_transition(current, target) + + +def test_terminal_states_have_no_outgoing_transition(): + for state in TERMINAL_STATES: + assert ALLOWED_TRANSITIONS[state] == set() + + +def test_duplicate_terminal_transition_is_rejected(): + assert not can_transition(JobState.SUCCEEDED, JobState.SUCCEEDED) + assert not can_transition(JobState.CANCELLED, JobState.CANCELLED) + + +def test_item_states_are_consistent(): + assert ItemState.SUCCEEDED in ITEM_TRANSITIONS[ItemState.RUNNING] + assert ITEM_TRANSITIONS[ItemState.SUCCEEDED] == set() # terminal item