Skip to content

refactor(lifecycle): one module owns all job/run status transitions #166

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 4 commits into from
Jul 10, 2026
Merged

refactor(lifecycle): one module owns all job/run status transitions #166

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
886eff0
docs(roadmap): capture architecture-deepening candidates from design …
seandavi Jul 10, 2026
00c74b2
refactor(lifecycle): one module owns all job/run status transitions
seandavi Jul 10, 2026
908ab5e
refactor(lifecycle): honor the guarded/idempotent contract (Copilot r…
seandavi Jul 10, 2026
67193df
refactor(lifecycle): guard claim on pending + tighten mark_running (C…
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
42 changes: 40 additions & 2 deletions docs/roadmap.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,8 @@
# nextflow_telemetry — Work Plan / Roadmap

_Last updated: 2026-07-02. Author: planning pass over the open-issue backlog after a
telemetry-hardening + 100k-scale UI evaluation session._
_Last updated: 2026-07-10. Author: planning pass over the open-issue backlog after a
telemetry-hardening + 100k-scale UI evaluation session; 2026-07-10 added the Architecture
deepening section from a `/improve-codebase-architecture` design review._

The project has grown organically: orchestration works well, but the **data model
underneath it is muddled** (study identity, completion semantics, sample metadata),
Expand Down Expand Up @@ -139,6 +140,43 @@ Tracker: **#94**. Rollout: ship API with enforcement **off**, roll daemons, then
- **#48** Temporal investigation — **recommend decline/park**: solves orchestration,
not the visibility gap we actually have; re-introduces heavy stateful infra.

## Architecture deepening (design review, 2026-07-10)

A `/improve-codebase-architecture` pass over the server, ETL, nf-client, and frontend
looked for **deep-module** opportunities (small interface over a lot of behaviour, tested
through that interface) rather than features. Distinct lens from the epics above: these are
internal refactors, most **not** tracked as issues. The through-line: a load-bearing concept
(job lifecycle, membership, "is this alive?", the `trace` shape) is re-derived inline at each
call site and kept consistent by "keep in sync with…" comments instead of by code. Each
candidate gives the concept one owning module. Full write-up was a scratch HTML review (not
in-repo); the table is the durable record.

Vocabulary (from `/codebase-design`): **module · interface · depth · seam · leverage ·
locality**. Strength: Strong / Worth exploring / Speculative.

| # | Deepening | Strength | Status | Seam / files | Cross-ref |
|---|-----------|----------|--------|--------------|-----------|
| 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) |
| 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 |
| 7 | Make the ETL spec actually drive ingest | Worth exploring | open | `specs.py` bypassed (qc special-case); `lake.SCHEMAS` hand-synced to parser keys; cli re-loops | ETL #153–158 |
| 8 | Submitter seam for HPC modes (nf-client) | Strong | open | if/elif over mode twice; `submit_*` signatures disagree; `lsf` dead branch | #68 (sacct behind slurm adapter) |
| 9 | Shared `useFetch` (loading/error) hook | Strong | open | fetch+loading+error copied in all 8 pages; errors swallowed → permanent spinner | #118, #121 |
| 10 | One status→colour/label map + real union | Strong | open | ~60 inline ternaries; `classification`/`status` typed as bare `string` | #121 |
| 11 | "Run log artifacts" module | Worth exploring | open | `ON CONFLICT uq_task_log` upsert ×3; run-level logs smuggled into task-keyed table | #120, #62, #93 |
| 12 | Delete phantom `RunEvent` variants (YAGNI) | Strong · deletion | open | `wrapper_log`/`slurm_state`/`workflow_oncomplete` have zero producers | #62/#66 (re-add when real) |

**Solid ground (deep already — don't touch):** `WorkflowService` write side (one-active-version
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.

## Suggested sequencing

1. **Foundation (now):** A design docs (`sample-metadata-design.md`; confirm
Expand Down
73 changes: 17 additions & 56 deletions src/nextflow_telemetry/routers/admin.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,11 +4,13 @@
import datetime

from fastapi import APIRouter, HTTPException
from sqlalchemy import func, select, update
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncEngine

from ..db import daemon_agents_tbl, dead_letter_tbl, jobs_tbl, samples_tbl, workflow_runs_tbl, workflows_tbl
from ..services.reconcile import ReconcileService, sweep_run_incomplete
from ..services import lifecycle
from ..services.lifecycle import RUN_TERMINAL_STATUSES, JobStatus, RunStatus
from ..services.reconcile import ReconcileService

# Keep in sync with routers/daemons.ACTIVE_THRESHOLD — a daemon is "active" if
# its last heartbeat is within this window.
Expand Down Expand Up @@ -52,21 +54,10 @@ async def reconcile_jobs():
)
async def reset_running(workflow_pk: int):
async with engine.begin() as conn:
result = await conn.execute(
update(jobs_tbl)
.where(
jobs_tbl.c.workflow_pk == workflow_pk,
jobs_tbl.c.status.in_(["running", "failed"]),
)
.values(
status="pending",
run_name=None,
retry_count=0,
failed_at=None,
failure_reason=None,
)
count = await lifecycle.reset_jobs_to_pending(
conn, workflow_pk, [JobStatus.running, JobStatus.failed]
)
return {"reset": result.rowcount}
return {"reset": count}

@router.post(
"/close-run",
Expand All @@ -83,28 +74,17 @@ async def reset_running(workflow_pk: int):
async def close_run(run_name: str):
now = datetime.datetime.now(datetime.timezone.utc)
async with engine.begin() as conn:
# Check the run exists and get its current status
row = (await conn.execute(
select(workflow_runs_tbl.c.status)
.where(workflow_runs_tbl.c.run_name == run_name)
)).first()
prior_status = await lifecycle.close_run(conn, run_name, RunStatus.completed, now)

if not row:
if prior_status is None:
raise HTTPException(
status_code=404,
detail=f"No workflow run with name '{run_name}'",
)

already_closed = row[0] in ("completed", "failed", "expired")

if not already_closed:
await conn.execute(
update(workflow_runs_tbl)
.where(workflow_runs_tbl.c.run_name == run_name)
.values(status="completed", completed_at=now)
)
already_closed = prior_status in RUN_TERMINAL_STATUSES

swept = await sweep_run_incomplete(conn, run_name, now)
swept = await lifecycle.sweep_incomplete(conn, run_name, now)

return {
"run_name": run_name,
Expand Down Expand Up @@ -141,12 +121,8 @@ async def expire_stale_runs(older_than_hours: float = 2.0):
)).scalars().all()

for run_name in stale:
await conn.execute(
update(workflow_runs_tbl)
.where(workflow_runs_tbl.c.run_name == run_name)
.values(status="completed", completed_at=now)
)
total_swept += await sweep_run_incomplete(conn, run_name, now)
await lifecycle.close_run(conn, run_name, RunStatus.completed, now)
total_swept += await lifecycle.sweep_incomplete(conn, run_name, now)

return {"stale_runs_closed": len(stale), "jobs_swept": total_swept}

Expand Down Expand Up @@ -191,12 +167,8 @@ async def heartbeat_watchdog(stale_after_minutes: float = _HEARTBEAT_STALE_MINUT
)).all()

for r in stale:
await conn.execute(
update(workflow_runs_tbl)
.where(workflow_runs_tbl.c.run_name == r.run_name)
.values(status="failed", completed_at=now, slurm_reason=reason)
)
jobs = await sweep_run_incomplete(conn, r.run_name, now)
await lifecycle.close_run(conn, r.run_name, RunStatus.failed, now, reason=reason)
jobs = await lifecycle.sweep_incomplete(conn, r.run_name, now)
total_swept += jobs
swept_runs.append({
"run_name": r.run_name,
Expand Down Expand Up @@ -225,7 +197,6 @@ async def heartbeat_watchdog(stale_after_minutes: float = _HEARTBEAT_STALE_MINUT
async def requeue_dead_letter():
now = datetime.datetime.now(datetime.timezone.utc)
async with engine.begin() as conn:
from sqlalchemy import select
rows = (await conn.execute(
select(dead_letter_tbl.c.id, dead_letter_tbl.c.job_id)
.where(dead_letter_tbl.c.resolved_at.is_(None))
Expand All @@ -237,19 +208,9 @@ async def requeue_dead_letter():
job_ids = [r.job_id for r in rows]
dlq_ids = [r.id for r in rows]

await conn.execute(
update(jobs_tbl)
.where(jobs_tbl.c.id.in_(job_ids))
.values(status="pending", retry_count=0, run_name=None,
failed_at=None, failure_reason=None)
)
await conn.execute(
update(dead_letter_tbl)
.where(dead_letter_tbl.c.id.in_(dlq_ids))
.values(resolved_at=now)
)
count = await lifecycle.requeue_dead_letter(conn, job_ids, dlq_ids, now)

return {"requeued": len(job_ids)}
return {"requeued": count}

@router.get(
"/dispatchability",
Expand Down
84 changes: 18 additions & 66 deletions src/nextflow_telemetry/routers/dispatch.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,10 +9,11 @@
from typing import Any

from pydantic import BaseModel, Field
from sqlalchemy import select, update
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncEngine

from ..db import jobs_tbl, samples_tbl, workflow_runs_tbl, workflows_tbl
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]
Expand Down Expand Up @@ -149,22 +150,17 @@ async def dispatch_batch(req: DispatchBatchRequest):
repository_url = rows[0]["repository_url"]
revision = rows[0]["revision"]

await conn.execute(
workflow_runs_tbl.insert().values(
run_name=run_name,
workflow_id=first_wf_id,
workflow_version=first_wf_ver,
workflow_pk=workflow_pk,
revision=revision,
status="claimed",
claimed_at=now,
)
)

await conn.execute(
update(jobs_tbl)
.where(jobs_tbl.c.id.in_(job_ids))
.values(run_name=run_name, status="claimed")
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
Expand Down Expand Up @@ -213,38 +209,13 @@ async def dispatch_batch(req: DispatchBatchRequest):
async def report_submitted(req: SubmittedRequest):
now = datetime.now(timezone.utc)
async with engine.begin() as conn:
result = await conn.execute(
update(workflow_runs_tbl)
.where(
workflow_runs_tbl.c.run_name == req.run_name,
workflow_runs_tbl.c.status == "claimed",
)
.values(
status="submitted",
submitted_at=now,
executor_job_id=req.executor_job_id,
)
.returning(workflow_runs_tbl.c.run_name)
)
if not result.fetchone():
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",
)

# Advance jobs from `claimed` to `submitted` — distinct from
# `running`, which is set only when the weblog `started` event
# arrives. The gap between the two is the scheduler queue wait,
# which the dashboard surfaces separately.
await conn.execute(
update(jobs_tbl)
.where(
jobs_tbl.c.run_name == req.run_name,
jobs_tbl.c.status == "claimed",
)
.values(status="submitted")
)

return {"run_name": req.run_name, "status": "submitted"}

@router.post(
Expand All @@ -261,27 +232,8 @@ 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:
result = await conn.execute(
update(workflow_runs_tbl)
.where(
workflow_runs_tbl.c.status == "claimed",
workflow_runs_tbl.c.claimed_at < cutoff,
)
.values(status="expired")
.returning(workflow_runs_tbl.c.run_name)
)
expired_run_names = [r[0] for r in result.fetchall()]

if expired_run_names:
await conn.execute(
update(jobs_tbl)
.where(
jobs_tbl.c.run_name.in_(expired_run_names),
jobs_tbl.c.status == "claimed",
)
.values(status="pending", run_name=None)
)
count = await lifecycle.requeue_expired(conn, cutoff)

return {"requeued_runs": len(expired_run_names)}
return {"requeued_runs": count}

return router
17 changes: 7 additions & 10 deletions src/nextflow_telemetry/routers/runs.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@
WrapperStartedEvent,
)
from ..db import task_logs_tbl, telemetry_tbl, workflow_runs_tbl, jobs_tbl, task_executions_tbl
from ..services import lifecycle


# A non-terminal run with no heartbeat for longer than this is "stalled" —
Expand Down Expand Up @@ -430,16 +431,12 @@ async def _apply_summary_update(conn, run_name: str, parsed: RunEvent, now: date
query telemetry_tbl directly to recover the event stream.
"""
if isinstance(parsed, WrapperStartedEvent):
# Mark the run as "submitted" only if it's still in the earlier "claimed" state.
# We don't want to roll status back from running/completed.
await conn.execute(
update(workflow_runs_tbl)
.where(
workflow_runs_tbl.c.run_name == run_name,
workflow_runs_tbl.c.status == "claimed",
)
.values(status="submitted", submitted_at=now)
)
# Mark the run (+ its jobs) as "submitted" only if still "claimed" —
# a defensive fallback for the case where this event races ahead of
# POST /dispatch/submitted. Folds into the same claimed->submitted
# transition dispatch.py's report_submitted uses, so a normal-order
# run (already submitted by the time this arrives) is a no-op here.
await lifecycle.mark_submitted(conn, run_name, executor_job_id=None, now=now)
return

values: dict = {}
Expand Down
Loading
Loading