From 8d1287767cd67126e1a733bc925800e53b8538c9 Mon Sep 17 00:00:00 2001 From: Sean Davis Date: Fri, 10 Jul 2026 15:46:06 -0600 Subject: [PATCH 1/3] docs(roadmap): mark deepening candidate #3 (job-lifecycle) done (#166) Co-Authored-By: Claude Fable 5 --- docs/roadmap.md | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/docs/roadmap.md b/docs/roadmap.md index 1895c56..4d16b77 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -158,7 +158,7 @@ locality**. Strength: Strong / Worth exploring / Speculative. |---|-----------|----------|--------|--------------|-----------| | 1 | Study-membership module — one write seam | Strong | **✓ DONE #164/#165** | `add_to_collection()` in `services/collection.py`; retired `metadata.cohort` | Epic A, ADR-0005 | | 2 | Shared active-version scope predicate | Worth exploring | open | `_workflow_scope` (cohort) re-inlined in admin.stats, process_metrics; completion% redefined 3× | **#116**, Epic B | -| 3 | Job-lifecycle module (enums + legal transitions) | Strong | open | status literals across ~10 files; "close+sweep" ×4 (dispatch/runs/admin/telemetry/reconcile) | — (new) | +| 3 | Job-lifecycle module (enums + legal transitions) | Strong | **✓ DONE #166** | `services/lifecycle.py` — free fns take `conn`; JobStatus/RunStatus enums; guarded/idempotent; carve-outs documented | — | | 4 | DispatchService — pull the claim out of the handler | Strong | open | 120-line pick-then-lock inline in `dispatch.py` HTTP handler; 3 wiring conventions | pairs w/ #3 | | 5 | Liveness module — one "is this alive?" | Strong | open · **latent bug** | stalled re-derived 6× / 5 columns; read path calls `submitted` runs stalled, watchdog won't reap them | **#115**, ADR-0002 | | 6 | One reader for the `trace` JSONB | Worth exploring | open | decoded 3 ways (`.get` / `->>`); FAILED/ABORTED predicate copied ~17× | #62 | @@ -173,9 +173,10 @@ locality**. Strength: Strong / Worth exploring / Speculative. invariant + auto-retire), `process_metrics` service, ETL `engine.py`+`specs.py` split, `log.py` JSONFormatter, `lib/api.ts`+`types.ts` network seam, `_CaptureBuffer`. -**Where to start:** #3 job-lifecycle module for leverage (pairs with #4); #5 liveness for the one -real correctness bug (ties into #115 already shipped); #9 + #12 as the fastest relief for "the UI -feels confusing." #2, #6, #7 fold naturally into work already scheduled under Epic B / ETL. +**Where to start:** #3 done (#166) — #4 DispatchService is next (builds on `lifecycle.claim()`); +#5 liveness for the one real correctness bug (ties into #115 already shipped); #9 + #12 as the +fastest relief for "the UI feels confusing." #2, #6, #7 fold naturally into work already scheduled +under Epic B / ETL. ## Suggested sequencing From 61b45182fd7ce32f72cde8418b177c31f4663aa8 Mon Sep 17 00:00:00 2001 From: Sean Davis Date: Fri, 10 Jul 2026 15:52:26 -0600 Subject: [PATCH 2/3] refactor(dispatch): extract DispatchService from the HTTP handlers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pulls the dispatch orchestration out of routers/dispatch.py into a new services/dispatch.py (@dataclass DispatchService(engine)), following the service-injected-router convention (services/sample.py + routers/samples.py) — the one clean wiring the design review praised. Deepening candidate #4. - claim_batch(): the two-phase pick-then-lock (SELECT ... FOR UPDATE SKIP LOCKED, issue #74), run_name minting, lifecycle.claim(), and sample-metadata fetch — moved verbatim. Returns a plain ClaimedBatch/ClaimedJob dataclass or None; the service is HTTP/Pydantic-agnostic (no FastAPI imports). - report_submitted() -> bool, requeue_expired() -> int; CLAIM_TTL_MINUTES and the uuid7 version-guard move here too. - The state writes still go through services/lifecycle.py (the claim seam from #166); this owns only the orchestration around them. Router drops to a thin call-and-map: parse request -> service -> map None to 204 / False to 404. Pydantic models stay inline (repo convention); all OpenAPI strings unchanged. Behaviour-preserving. New tests/test_dispatch_service.py drives the service directly at the seam (claim-then-empty, submitted true/false, requeue-expired). typecheck clean; full suite 177 passed (existing dispatch integration/e2e unchanged). Co-Authored-By: Claude Fable 5 --- src/nextflow_telemetry/routers/dispatch.py | 160 +++------------ src/nextflow_telemetry/services/dispatch.py | 202 +++++++++++++++++++ tests/test_dispatch_service.py | 207 ++++++++++++++++++++ 3 files changed, 432 insertions(+), 137 deletions(-) create mode 100644 src/nextflow_telemetry/services/dispatch.py create mode 100644 tests/test_dispatch_service.py diff --git a/src/nextflow_telemetry/routers/dispatch.py b/src/nextflow_telemetry/routers/dispatch.py index 5b066c5..73f7f08 100644 --- a/src/nextflow_telemetry/routers/dispatch.py +++ b/src/nextflow_telemetry/routers/dispatch.py @@ -1,26 +1,14 @@ """Dispatch router — client-facing endpoints for claiming and reporting jobs.""" from __future__ import annotations -import sys -from datetime import datetime, timedelta, timezone +from typing import Any from fastapi import APIRouter, HTTPException from fastapi.responses import Response -from typing import Any - from pydantic import BaseModel, Field -from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncEngine -from ..db import jobs_tbl, samples_tbl, workflows_tbl -from ..services import lifecycle - -if sys.version_info >= (3, 13): - from uuid import uuid7 as _uuid7 # type: ignore[attr-defined] -else: - from uuid_extensions import uuid7 as _uuid7 - -CLAIM_TTL_MINUTES = 5 +from ..services.dispatch import DispatchService class DispatchBatchRequest(BaseModel): @@ -62,6 +50,7 @@ class SubmittedRequest(BaseModel): def create_dispatch_router(engine: AsyncEngine) -> APIRouter: router = APIRouter(prefix="/dispatch", tags=["dispatch"]) + svc = DispatchService(engine=engine) @router.post( "/batch", @@ -78,121 +67,24 @@ def create_dispatch_router(engine: AsyncEngine) -> APIRouter: ), ) async def dispatch_batch(req: DispatchBatchRequest): - now = datetime.now(timezone.utc) - - async with engine.begin() as conn: - # Two-step pick-then-lock: first decide *which* (workflow_id, - # workflow_version) batch to claim, then take a row-level lock - # only on jobs in that batch. Issue #74: a single FOR UPDATE - # SKIP LOCKED LIMIT N could lock rows across multiple workflows - # that we'd then narrow away in Python — those locks blocked - # other dispatchers needlessly. - pick_q = ( - select(jobs_tbl.c.workflow_id, jobs_tbl.c.workflow_version) - .join(workflows_tbl, jobs_tbl.c.workflow_pk == workflows_tbl.c.id) - .where( - jobs_tbl.c.status == "pending", - workflows_tbl.c.status == "active", - ) - ) - if req.workflow_id: - pick_q = pick_q.where(jobs_tbl.c.workflow_id.in_(req.workflow_id)) - if req.workflow_version: - pick_q = pick_q.where(jobs_tbl.c.workflow_version == req.workflow_version) - pick_q = ( - pick_q.order_by( - jobs_tbl.c.workflow_id, - jobs_tbl.c.workflow_version, - jobs_tbl.c.created_at, - ) - .limit(1) - # Skip rows another dispatcher is already claiming. Without - # this, a competing transaction holding a lock on the - # otherwise-first pending job would cause every subsequent - # caller to pick that workflow, fail step 2 (FOR UPDATE - # SKIP LOCKED returns 0 rows because all of that workflow's - # rows are locked too), and 204 → spin. SKIP LOCKED on - # pick steers us to a workflow that *has* claimable rows. - .with_for_update(of=jobs_tbl, skip_locked=True) - ) - - pick_row = (await conn.execute(pick_q)).mappings().first() - if pick_row is None: - return Response(status_code=204) - first_wf_id = pick_row["workflow_id"] - first_wf_ver = pick_row["workflow_version"] - - # Step 2: lock and claim jobs in that single batch only. - # If a competing dispatcher swept through between step 1 and - # step 2, this returns nothing and the caller retries — the - # next pick may resolve to a different workflow. - claim_q = ( - select(jobs_tbl, workflows_tbl) - .join(workflows_tbl, jobs_tbl.c.workflow_pk == workflows_tbl.c.id) - .where( - jobs_tbl.c.status == "pending", - workflows_tbl.c.status == "active", - jobs_tbl.c.workflow_id == first_wf_id, - jobs_tbl.c.workflow_version == first_wf_ver, - ) - .order_by(jobs_tbl.c.created_at) - .limit(req.limit) - .with_for_update(of=jobs_tbl, skip_locked=True) - ) - rows = (await conn.execute(claim_q)).mappings().all() - - if not rows: - return Response(status_code=204) - - job_ids = [r["id"] for r in rows] - run_name = "r" + str(_uuid7()) - workflow_pk = rows[0]["workflow_pk"] - repository_url = rows[0]["repository_url"] - revision = rows[0]["revision"] - - await lifecycle.claim( - conn, - job_ids, - run_name, - { - "workflow_id": first_wf_id, - "workflow_version": first_wf_ver, - "workflow_pk": workflow_pk, - "revision": revision, - "claimed_at": now, - }, - ) - - # Fetch sample fields in a separate query to avoid JOIN conflicts - # with the FOR UPDATE SKIP LOCKED above. - sample_ids = [r["sample_id"] for r in rows] - meta_result = await conn.execute( - select( - samples_tbl.c.sample_id, - samples_tbl.c.ncbi_accession, - samples_tbl.c.metadata_, - ) - .where(samples_tbl.c.sample_id.in_(sample_ids)) - ) - sample_map: dict[str, Any] = { - r["sample_id"]: r - for r in meta_result.mappings().all() - } + result = await svc.claim_batch(req.limit, req.workflow_id, req.workflow_version) + if result is None: + return Response(status_code=204) return DispatchBatchResponse( - run_name=run_name, - workflow_id=first_wf_id, - workflow_version=first_wf_ver, - workflow_pk=workflow_pk, - repository_url=repository_url, - revision=revision, + run_name=result.run_name, + workflow_id=result.workflow_id, + workflow_version=result.workflow_version, + workflow_pk=result.workflow_pk, + repository_url=result.repository_url, + revision=result.revision, jobs=[ DispatchedJob( - sample_id=r["sample_id"], - ncbi_accession=sample_map.get(r["sample_id"], {}).get("ncbi_accession"), - metadata=sample_map.get(r["sample_id"], {}).get("metadata_") or {}, + sample_id=j.sample_id, + ncbi_accession=j.ncbi_accession, + metadata=j.metadata, ) - for r in rows + for j in result.jobs ], ) @@ -207,14 +99,12 @@ async def dispatch_batch(req: DispatchBatchRequest): ), ) async def report_submitted(req: SubmittedRequest): - now = datetime.now(timezone.utc) - async with engine.begin() as conn: - ok = await lifecycle.mark_submitted(conn, req.run_name, req.executor_job_id, now) - if not ok: - raise HTTPException( - status_code=404, - detail=f"No claimed run with name '{req.run_name}' found", - ) + ok = await svc.report_submitted(req.run_name, req.executor_job_id) + if not ok: + raise HTTPException( + status_code=404, + detail=f"No claimed run with name '{req.run_name}' found", + ) return {"run_name": req.run_name, "status": "submitted"} @@ -230,10 +120,6 @@ async def report_submitted(req: SubmittedRequest): ), ) async def requeue_expired(): - cutoff = datetime.now(timezone.utc) - timedelta(minutes=CLAIM_TTL_MINUTES) - async with engine.begin() as conn: - count = await lifecycle.requeue_expired(conn, cutoff) - - return {"requeued_runs": count} + return {"requeued_runs": await svc.requeue_expired()} return router diff --git a/src/nextflow_telemetry/services/dispatch.py b/src/nextflow_telemetry/services/dispatch.py new file mode 100644 index 0000000..e395d3f --- /dev/null +++ b/src/nextflow_telemetry/services/dispatch.py @@ -0,0 +1,202 @@ +"""Dispatch orchestration — claiming, submitted-confirmation, requeue. + +Pulled out of routers/dispatch.py so the router stays a thin HTTP/Pydantic +adapter (repo convention, see services/sample.py + routers/samples.py). The +STATE writes still go through services/lifecycle.py — this module owns the +orchestration around those calls (the pick-then-lock query, run_name +minting, sample-metadata fetch, TTL policy), not the transitions themselves. + +HTTP/Pydantic-agnostic: returns plain dataclasses/bool/int, never an +HTTPException or a Pydantic model. +""" +from __future__ import annotations + +import sys +from dataclasses import dataclass, field +from datetime import datetime, timedelta, timezone +from typing import Any + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncEngine + +from ..db import jobs_tbl, samples_tbl, workflows_tbl +from . import lifecycle + +if sys.version_info >= (3, 13): + from uuid import uuid7 as _uuid7 # type: ignore[attr-defined] +else: + from uuid_extensions import uuid7 as _uuid7 + +CLAIM_TTL_MINUTES = 5 + + +@dataclass +class ClaimedJob: + """A single job included in a claimed batch.""" + sample_id: str + ncbi_accession: str | None = None + metadata: dict[str, Any] = field(default_factory=dict) + + +@dataclass +class ClaimedBatch: + """Result of DispatchService.claim_batch — everything the caller needs + to build a `nextflow run` command and report it back on the wire.""" + run_name: str + workflow_id: str + workflow_version: str + workflow_pk: int + repository_url: str + revision: str + jobs: list[ClaimedJob] + + +@dataclass +class DispatchService: + engine: AsyncEngine + + async def claim_batch( + self, + limit: int, + workflow_id: list[str] | None = None, + workflow_version: str | None = None, + ) -> ClaimedBatch | None: + """Atomically claim a batch of pending jobs for one workflow version. + + Returns None when there are no pending jobs matching the filter + (the caller maps that to HTTP 204). + """ + now = datetime.now(timezone.utc) + + async with self.engine.begin() as conn: + # Two-step pick-then-lock: first decide *which* (workflow_id, + # workflow_version) batch to claim, then take a row-level lock + # only on jobs in that batch. Issue #74: a single FOR UPDATE + # SKIP LOCKED LIMIT N could lock rows across multiple workflows + # that we'd then narrow away in Python — those locks blocked + # other dispatchers needlessly. + pick_q = ( + select(jobs_tbl.c.workflow_id, jobs_tbl.c.workflow_version) + .join(workflows_tbl, jobs_tbl.c.workflow_pk == workflows_tbl.c.id) + .where( + jobs_tbl.c.status == "pending", + workflows_tbl.c.status == "active", + ) + ) + if workflow_id: + pick_q = pick_q.where(jobs_tbl.c.workflow_id.in_(workflow_id)) + if workflow_version: + pick_q = pick_q.where(jobs_tbl.c.workflow_version == workflow_version) + pick_q = ( + pick_q.order_by( + jobs_tbl.c.workflow_id, + jobs_tbl.c.workflow_version, + jobs_tbl.c.created_at, + ) + .limit(1) + # Skip rows another dispatcher is already claiming. Without + # this, a competing transaction holding a lock on the + # otherwise-first pending job would cause every subsequent + # caller to pick that workflow, fail step 2 (FOR UPDATE + # SKIP LOCKED returns 0 rows because all of that workflow's + # rows are locked too), and 204 → spin. SKIP LOCKED on + # pick steers us to a workflow that *has* claimable rows. + .with_for_update(of=jobs_tbl, skip_locked=True) + ) + + pick_row = (await conn.execute(pick_q)).mappings().first() + if pick_row is None: + return None + first_wf_id = pick_row["workflow_id"] + first_wf_ver = pick_row["workflow_version"] + + # Step 2: lock and claim jobs in that single batch only. + # If a competing dispatcher swept through between step 1 and + # step 2, this returns nothing and the caller retries — the + # next pick may resolve to a different workflow. + claim_q = ( + select(jobs_tbl, workflows_tbl) + .join(workflows_tbl, jobs_tbl.c.workflow_pk == workflows_tbl.c.id) + .where( + jobs_tbl.c.status == "pending", + workflows_tbl.c.status == "active", + jobs_tbl.c.workflow_id == first_wf_id, + jobs_tbl.c.workflow_version == first_wf_ver, + ) + .order_by(jobs_tbl.c.created_at) + .limit(limit) + .with_for_update(of=jobs_tbl, skip_locked=True) + ) + rows = (await conn.execute(claim_q)).mappings().all() + + if not rows: + return None + + job_ids = [r["id"] for r in rows] + run_name = "r" + str(_uuid7()) + workflow_pk = rows[0]["workflow_pk"] + repository_url = rows[0]["repository_url"] + revision = rows[0]["revision"] + + await lifecycle.claim( + conn, + job_ids, + run_name, + { + "workflow_id": first_wf_id, + "workflow_version": first_wf_ver, + "workflow_pk": workflow_pk, + "revision": revision, + "claimed_at": now, + }, + ) + + # Fetch sample fields in a separate query to avoid JOIN conflicts + # with the FOR UPDATE SKIP LOCKED above. + sample_ids = [r["sample_id"] for r in rows] + meta_result = await conn.execute( + select( + samples_tbl.c.sample_id, + samples_tbl.c.ncbi_accession, + samples_tbl.c.metadata_, + ) + .where(samples_tbl.c.sample_id.in_(sample_ids)) + ) + sample_map: dict[str, Any] = { + r["sample_id"]: r + for r in meta_result.mappings().all() + } + + return ClaimedBatch( + run_name=run_name, + workflow_id=first_wf_id, + workflow_version=first_wf_ver, + workflow_pk=workflow_pk, + repository_url=repository_url, + revision=revision, + jobs=[ + ClaimedJob( + sample_id=r["sample_id"], + ncbi_accession=sample_map.get(r["sample_id"], {}).get("ncbi_accession"), + metadata=sample_map.get(r["sample_id"], {}).get("metadata_") or {}, + ) + for r in rows + ], + ) + + async def report_submitted(self, run_name: str, executor_job_id: str | None) -> bool: + """Confirm a run has been submitted to the executor. + + Returns False if the run isn't found or isn't in `claimed` state (the + caller maps that to HTTP 404). + """ + now = datetime.now(timezone.utc) + async with self.engine.begin() as conn: + return await lifecycle.mark_submitted(conn, run_name, executor_job_id, now) + + async def requeue_expired(self) -> int: + """Expire stale `claimed` runs (older than CLAIM_TTL_MINUTES) and + reset their jobs back to `pending`. Returns the count of runs expired.""" + cutoff = datetime.now(timezone.utc) - timedelta(minutes=CLAIM_TTL_MINUTES) + async with self.engine.begin() as conn: + return await lifecycle.requeue_expired(conn, cutoff) diff --git a/tests/test_dispatch_service.py b/tests/test_dispatch_service.py new file mode 100644 index 0000000..c65db69 --- /dev/null +++ b/tests/test_dispatch_service.py @@ -0,0 +1,207 @@ +"""Tests for services/dispatch.py — DispatchService. + +Same testcontainers Postgres fixture pattern as test_lifecycle.py: drives +the service directly (no FastAPI TestClient) against a fresh AsyncEngine. +test_integration.py / test_e2e.py already cover the same behaviour through +the HTTP layer — these are the behaviour-preservation proof for the service +extraction itself. +""" +from __future__ import annotations + +import asyncio +import uuid +from datetime import datetime, timedelta, timezone + +from sqlalchemy import insert, select +from sqlalchemy.ext.asyncio import create_async_engine + +from nextflow_telemetry.db import ( + jobs_tbl, + samples_tbl, + workflow_runs_tbl, + workflows_tbl, +) +from nextflow_telemetry.services.dispatch import DispatchService + + +def _run(coro): + return asyncio.run(coro) + + +async def _seed_workflow(engine, *, max_retries: int = 3) -> tuple[int, str]: + now = datetime.now(timezone.utc) + wf_id = f"ds-wf-{uuid.uuid4().hex[:8]}" + async with engine.begin() as conn: + pk = ( + await conn.execute( + insert(workflows_tbl) + .returning(workflows_tbl.c.id) + .values( + workflow_id=wf_id, + version="1.0.0", + repository_url="https://example.org/repo", + revision="abc123", + max_retries=max_retries, + status="active", + created_at=now, + updated_at=now, + ) + ) + ).scalar_one() + return pk, wf_id + + +async def _seed_sample(engine, *, ncbi_accession: str = "SRR000001") -> str: + now = datetime.now(timezone.utc) + sample_id = f"SRR-ds-{uuid.uuid4().hex[:8]}" + async with engine.begin() as conn: + await conn.execute( + insert(samples_tbl).values( + sample_id=sample_id, + ncbi_accession=ncbi_accession, + metadata_={"foo": "bar"}, + created_at=now, + updated_at=now, + ) + ) + return sample_id + + +async def _seed_job( + engine, *, workflow_pk: int, workflow_id: str, sample_id: str, status: str = "pending" +) -> int: + now = datetime.now(timezone.utc) + async with engine.begin() as conn: + job_id = ( + await conn.execute( + insert(jobs_tbl) + .returning(jobs_tbl.c.id) + .values( + sample_id=sample_id, + workflow_pk=workflow_pk, + workflow_id=workflow_id, + workflow_version="1.0.0", + status=status, + created_at=now, + ) + ) + ).scalar_one() + return job_id + + +async def _get_job(engine, job_id: int) -> dict: + async with engine.connect() as conn: + row = ( + await conn.execute(select(jobs_tbl).where(jobs_tbl.c.id == job_id)) + ).mappings().first() + return dict(row) + + +def test_claim_batch_claims_pending_jobs_then_none_when_empty(db_asyncpg_url): + async def go(): + engine = create_async_engine(db_asyncpg_url) + try: + svc = DispatchService(engine=engine) + wf_pk, wf_id = await _seed_workflow(engine) + sample_a = await _seed_sample(engine) + sample_b = await _seed_sample(engine) + job_a = await _seed_job(engine, workflow_pk=wf_pk, workflow_id=wf_id, sample_id=sample_a) + job_b = await _seed_job(engine, workflow_pk=wf_pk, workflow_id=wf_id, sample_id=sample_b) + + batch = await svc.claim_batch(limit=10, workflow_id=[wf_id]) + assert batch is not None + assert batch.run_name.startswith("r") + assert batch.workflow_id == wf_id + assert batch.workflow_version == "1.0.0" + assert batch.workflow_pk == wf_pk + assert batch.repository_url == "https://example.org/repo" + assert batch.revision == "abc123" + assert {j.sample_id for j in batch.jobs} == {sample_a, sample_b} + for j in batch.jobs: + assert j.ncbi_accession == "SRR000001" + assert j.metadata == {"foo": "bar"} + + for job_id in (job_a, job_b): + job = await _get_job(engine, job_id) + assert job["status"] == "claimed" + assert job["run_name"] == batch.run_name + + # Nothing left pending for this workflow. + empty = await svc.claim_batch(limit=10, workflow_id=[wf_id]) + assert empty is None + finally: + await engine.dispose() + + _run(go()) + + +def test_report_submitted_true_then_false(db_asyncpg_url): + async def go(): + engine = create_async_engine(db_asyncpg_url) + try: + svc = DispatchService(engine=engine) + wf_pk, wf_id = await _seed_workflow(engine) + sample_id = await _seed_sample(engine) + await _seed_job(engine, workflow_pk=wf_pk, workflow_id=wf_id, sample_id=sample_id) + + batch = await svc.claim_batch(limit=10, workflow_id=[wf_id]) + assert batch is not None + + ok = await svc.report_submitted(batch.run_name, "SLURM_1") + assert ok is True + + # Already submitted — second confirmation is a no-op / False. + ok_again = await svc.report_submitted(batch.run_name, "SLURM_1") + assert ok_again is False + + # Unknown run name. + unknown = await svc.report_submitted("no-such-run", None) + assert unknown is False + finally: + await engine.dispose() + + _run(go()) + + +def test_requeue_expired_recycles_stale_claims(db_asyncpg_url): + async def go(): + engine = create_async_engine(db_asyncpg_url) + try: + svc = DispatchService(engine=engine) + wf_pk, wf_id = await _seed_workflow(engine) + sample_id = await _seed_sample(engine) + + stale_run_name = f"r-stale-{uuid.uuid4().hex[:8]}" + stale_claimed_at = datetime.now(timezone.utc) - timedelta(minutes=10) + async with engine.begin() as conn: + await conn.execute( + insert(workflow_runs_tbl).values( + run_name=stale_run_name, + workflow_id=wf_id, + workflow_version="1.0.0", + workflow_pk=wf_pk, + revision="abc123", + status="claimed", + claimed_at=stale_claimed_at, + ) + ) + job_id = await _seed_job( + engine, workflow_pk=wf_pk, workflow_id=wf_id, sample_id=sample_id, status="claimed" + ) + async with engine.begin() as conn: + await conn.execute( + jobs_tbl.update() + .where(jobs_tbl.c.id == job_id) + .values(run_name=stale_run_name) + ) + + count = await svc.requeue_expired() + assert count == 1 + + job = await _get_job(engine, job_id) + assert job["status"] == "pending" + assert job["run_name"] is None + finally: + await engine.dispose() + + _run(go()) From d4bf5600af6d0a6745fb558e475cb2afd0dd7da0 Mon Sep 17 00:00:00 2001 From: Sean Davis Date: Fri, 10 Jul 2026 16:02:47 -0600 Subject: [PATCH 3/3] test(dispatch): seed claimed job with run_name in one transaction (Copilot) Copilot review nit on #167: the requeue-expired test seeded a claimed job with run_name=NULL then set it in a second transaction. Seed it in one insert via a new run_name param on the _seed_job helper. Co-Authored-By: Claude Fable 5 --- tests/test_dispatch_service.py | 13 +++++-------- 1 file changed, 5 insertions(+), 8 deletions(-) diff --git a/tests/test_dispatch_service.py b/tests/test_dispatch_service.py index c65db69..5796b0a 100644 --- a/tests/test_dispatch_service.py +++ b/tests/test_dispatch_service.py @@ -68,7 +68,8 @@ async def _seed_sample(engine, *, ncbi_accession: str = "SRR000001") -> str: async def _seed_job( - engine, *, workflow_pk: int, workflow_id: str, sample_id: str, status: str = "pending" + engine, *, workflow_pk: int, workflow_id: str, sample_id: str, + status: str = "pending", run_name: str | None = None, ) -> int: now = datetime.now(timezone.utc) async with engine.begin() as conn: @@ -82,6 +83,7 @@ async def _seed_job( workflow_id=workflow_id, workflow_version="1.0.0", status=status, + run_name=run_name, created_at=now, ) ) @@ -186,14 +188,9 @@ async def go(): ) ) job_id = await _seed_job( - engine, workflow_pk=wf_pk, workflow_id=wf_id, sample_id=sample_id, status="claimed" + engine, workflow_pk=wf_pk, workflow_id=wf_id, sample_id=sample_id, + status="claimed", run_name=stale_run_name, ) - async with engine.begin() as conn: - await conn.execute( - jobs_tbl.update() - .where(jobs_tbl.c.id == job_id) - .values(run_name=stale_run_name) - ) count = await svc.requeue_expired() assert count == 1