refactor(dispatch): extract DispatchService from the HTTP handlers#167
Merged
Conversation
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
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 <noreply@anthropic.com>
There was a problem hiding this comment.
Pull request overview
This pull request refactors dispatch orchestration by extracting the concurrency-critical claim/submit/requeue logic out of the FastAPI router into a dedicated DispatchService, aligning dispatch with the repo’s “thin router + service layer” architecture and keeping lifecycle state transitions centralized in services/lifecycle.py.
Changes:
- Added
DispatchService(services/dispatch.py) to own batch-claim orchestration, submitted confirmation, and expired-claim requeue, returning HTTP-agnostic dataclasses/bool/int. - Simplified
routers/dispatch.pyinto a thin adapter that calls the service and mapsNone/Falseto204/404without changing response shapes. - Added direct seam tests for the new service (
tests/test_dispatch_service.py) and updated the roadmap doc to mark lifecycle (#166) done and DispatchService as next.
Reviewed changes
Copilot reviewed 4 out of 4 changed files in this pull request and generated 1 comment.
| File | Description |
|---|---|
| tests/test_dispatch_service.py | Adds unit-style service seam tests for claim/submitted/requeue behavior against testcontainers Postgres. |
| src/nextflow_telemetry/services/dispatch.py | Introduces DispatchService and dataclass return types; encapsulates pick-then-lock query and lifecycle calls. |
| src/nextflow_telemetry/routers/dispatch.py | Refactors router handlers to delegate orchestration to DispatchService and keep HTTP mapping logic in the router. |
| docs/roadmap.md | Updates roadmap status notes to reflect lifecycle extraction completion and DispatchService as the next step. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+188
to
+196
| 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) | ||
| ) |
…pilot) 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 <noreply@anthropic.com>
seandavi
added a commit
that referenced
this pull request
Jul 10, 2026
…emon (#168) * docs(roadmap): mark deepening candidate #4 (DispatchService) done (#167) Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * refactor(nf-client): scheduler Submitter seam; de-duplicate submit/daemon The SLURM/PBS submission path (build the same Jinja context, render the template, dispatch via sbatch/qsub with retries) was copied verbatim across cli.py's submit() and daemon(). Extract it behind one seam in submission.py: - build_submission_context(batch, cfg, sample_ids) — the context dict, once. - SCHEDULER_SUBMITTERS registry {slurm, pbs} + submit_to_scheduler() entry point; lsf is deliberately absent (raises UnwiredSchedulerError). - Two named exceptions (TemplatePathMissingError, UnwiredSchedulerError) let each caller keep its own control flow: submit() exits nonzero for both; daemon() exits for a missing template but skips-and-continues for an unwired mode. A genuine submission failure is NOT wrapped — it propagates as the SubprocessError/OSError submit_with_retry already raised. local mode is untouched — it's deliberately different per caller (non-blocking Popen in submit(); blocking subprocess.run in daemon() to serialise runs). Behaviour-preserving. One faithfulness fix over a naive extraction: the daemon's submission-failure catch is narrowed from `except Exception` to `except (SubprocessError, OSError)` so a template/render error (a config bug) still stops the daemon as before, rather than being swallowed into an infinite skip loop mislabelled "submission failed after retries". New tests/test_submitters.py (10 tests): context keys, both dispatch commands, export_none toggle, the two can't-try signals, and failure-propagates-as-is. nf_client suite 60 passed; server suite 177 unaffected. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * refactor(nf-client): sync comments + narrow template_path type (Copilot) Copilot review on #168: - Update the seam's module comment and the propagation test's docstring to reflect the daemon's narrowed `except (SubprocessError, OSError)` catch (they still described the old bare `except Exception`). - submit_to_scheduler: bind `template_path` locally and check `is None` so the Optional is narrowed to Path before render_submission_script (clearer for type-checkers than the truthiness check). nf_client suite still 60 passed. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What
Pulls the dispatch orchestration out of
routers/dispatch.pyinto a newservices/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. Roadmap "Architecture deepening" candidate #4, built directly onlifecycle.claim()from #166.Design (agreed in the #3 grilling)
The seam was drawn during #3: lifecycle owns the state mutation; DispatchService owns the pick-then-lock and response assembly.
claim_batch()— the two-phaseSELECT … FOR UPDATE SKIP LOCKEDpick-then-lock (issue perf: tighten dispatch SELECT FOR UPDATE to a single (workflow_id, version) batch #74),run_nameminting,lifecycle.claim(), and the separate sample-metadata fetch — moved verbatim. Returns a plainClaimedBatch/ClaimedJobdataclass orNone.bool/int. The router mapsNone→204,False→404.report_submitted() -> bool,requeue_expired() -> int;CLAIM_TTL_MINUTESand theuuid7version-guard move into the service.services/lifecycle.py(refactor(lifecycle): one module owns all job/run status transitions #166).Router drops to a thin call-and-map; Pydantic models stay inline (repo convention); all OpenAPI strings unchanged.
Behaviour
Behaviour-preserving — the concurrency-critical pick-then-lock SQL is byte-identical, and the 204/404 conditions and response shapes are unchanged.
Tests
New
tests/test_dispatch_service.pydrives the service directly at the seam (claim-then-empty,report_submittedtrue/false/unknown,requeue_expiredrecycling a stale claim). Existingtest_integration.py/test_e2e.pydispatch coverage is unchanged and green.just typecheck: clean (60 files)just test: 177 passed(Also includes a roadmap doc update marking #3 done / #4 next.)
🤖 Generated with Claude Code