Skip to content

refactor(dispatch): extract DispatchService from the HTTP handlers #167

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
seandavi merged 3 commits into from
Jul 10, 2026
Merged

refactor(dispatch): extract DispatchService from the HTTP handlers #167

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
8d12877
docs(roadmap): mark deepening candidate #3 (job-lifecycle) done (#166)
seandavi Jul 10, 2026
61b4518
refactor(dispatch): extract DispatchService from the HTTP handlers
seandavi Jul 10, 2026
d4bf560
test(dispatch): seed claimed job with run_name in one transaction (Co…
seandavi Jul 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 5 additions & 4 deletions docs/roadmap.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 |
Expand All @@ -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

Expand Down
160 changes: 23 additions & 137 deletions src/nextflow_telemetry/routers/dispatch.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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):
Expand Down Expand Up @@ -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",
Expand All @@ -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
],
)

Expand All @@ -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"}

Expand All @@ -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
Loading