Submit and monitor Slurm jobs with iris-compute jobs

iris-compute is the CLI for the compute bounded context: it registers a TaskSpec/TaskRun pair per submission, dispatches it to Slurm via SlurmJobDispatcher, and tracks its lifecycle in Redis (RedisTaskRunRepository). It exists so that submitting and monitoring cluster jobs from projects/ scripts goes through the same domain model that a future API/UI will drive — the CLI commands are thin wrappers over reusable use cases in iris.applications.compute.use_cases.

This guide walks through the jobs sub-app end to end using a real two-step genomics pipeline as the workload, so you can verify a fresh Slurm + Redis environment works before wiring it into anything bigger.

Prerequisites

uv pip install -e ".[cli,events,genomics]"

• Redis reachable at IRIS_REDIS_URL (default redis://localhost:6379/0). Any reachable instance works for a smoke test — this does not have to be the same Redis used for the events streams.
• Slurm binaries (sbatch, squeue, sacct, scancel) on PATH — true by default on any cluster login node.
• A GENCODE annotation, HGNC JSON dump, the WES Zarr cohort store, and a reachable VEP Flight server / MCPS AF tables — the same data dependencies projects/recgenes/ already uses. Default paths point at the cluster's cephfs locations; override with flags if yours differ.

export IRIS_REDIS_URL=redis://localhost:6379/0
export IRIS_COMPUTE_SUBMIT_DIR=~/.iris/compute/slurm   # rendered .sbatch scripts, one per run
export IRIS_COMPUTE_LOG_DIR=~/.iris/compute/slurm/logs # defaults to $IRIS_COMPUTE_SUBMIT_DIR/logs

The example workload

projects/examples/gene_variant_extraction.py has two subcommands, deliberately split so they can run as two dependent Slurm jobs:

1. make-bed — resolves a gene symbol to its canonical transcript's exons (+ splice buffer) using the resolve_gene_regions use case (GENCODE + HGNC alias fallback), then writes them to a BED file with BedRegionExporter.
2. extract — reads that BED back with BedRegionReader, opens the WES Zarr cohort store with the existing PerIndividualAltCountProcessor genotype processor, runs VariantExtractionPipeline with VepFlightAnnotator
3. McpsAFAnnotator, and exports the annotated variants to TSV via TsvVariantExporter.

dbSNP rsIDs need no separate annotator: ZarrVariantStore already reads them from the source VCF's ID field into VariantNomenclature.dbsnp_id, exposed as the dbsnp column in the TSV output.

Try it locally first, outside Slurm, to confirm paths and connectivity:

uv run python projects/examples/gene_variant_extraction.py make-bed \
    --gene BRCA1 --output-bed /tmp/BRCA1.exons.bed

uv run python projects/examples/gene_variant_extraction.py extract \
    --bed /tmp/BRCA1.exons.bed --output /tmp/BRCA1.variants.tsv

Submit the first job

iris-compute jobs submit brca1_bed \
    "uv run python projects/examples/gene_variant_extraction.py make-bed --gene BRCA1 --output-bed results/BRCA1.exons.bed" \
    --workdir "$(pwd)" \
    --cpus 2 --mem-mb 4000 --time-s 600

✓ run_44a36f211a69440c8bfc0335bcd52077 → status=SUBMITTED
  slurm_job_id=424242

The command line is a single string parsed with shlex, exactly like typing it in a shell — this avoids ambiguity between iris-compute's own flags and the payload's flags. submit maps flags onto the domain's scheduler-neutral ResourceRequest / DispatchHints, which SlurmJobDispatcher then translates into sbatch options:

jobs submit flag	Domain field	Slurm option
--cpus	ResourceRequest.cpus_per_rank	--cpus-per-task
--mem-mb	ResourceRequest.memory_mb	--mem
--time-s	ResourceRequest.time_limit_s	--time
--nodes	ResourceRequest.nodes	--nodes
--gpus	ResourceRequest.gpus	--gres=gpu:N
--queue	DispatchHints.queue	--partition
--project	DispatchHints.project	--account
--quality	DispatchHints.quality	--qos
--nodelist	DispatchHints.extra (raw escape hatch)	--nodelist
--env KEY=VALUE (repeatable)	CommandSpec.env	export KEY=VALUE in the rendered script

Save the printed run_id — every other jobs command addresses a job by it, not by the Slurm job id.

Check on it

iris-compute jobs status run_44a36f211a69440c8bfc0335bcd52077

run_id:   run_44a36f211a69440c8bfc0335bcd52077
task:     brca1_bed (task_1eebc66eddf840af94e2b0352789ded3)
status:   SUBMITTED
slurm_id: 424242

status is whatever was last written to Redis — it does not query Slurm itself. Slurm has no push notifications, so state only changes on the next jobs poll (see below). List every run known to IRIS, optionally filtered:

iris-compute jobs list
iris-compute jobs list --status RUNNING

run_44a36f2...  SUBMITTED           slurm=424242    brca1_bed

Sync state with poll

iris-compute jobs poll

→ TaskRunStarted: run_id=run_44a36f211a69440c8bfc0335bcd52077

poll fetches every run that is not yet terminal, calls SlurmJobDispatcher.poll() (squeue, falling back to sacct for jobs that already left the queue), and folds the resulting JobObservation into a domain event (TaskQueued / TaskRunStarted / TaskRunSucceeded / TaskRunFailed / …). Run it again once the job finishes and status moves to SUCCEEDED or FAILED. Since nothing pushes changes to IRIS, poll needs to be invoked periodically — a cron entry works well for a first environment test:

* * * * * IRIS_REDIS_URL=redis://localhost:6379/0 iris-compute jobs poll >> ~/.iris/compute/poll.log 2>&1

Chain the second job with --depends-on

Once make-bed reaches SUCCEEDED, submit extract so it depends on it:

iris-compute jobs submit brca1_extract \
    "uv run python projects/examples/gene_variant_extraction.py extract --bed results/BRCA1.exons.bed --output results/BRCA1.variants.tsv --workers 4" \
    --workdir "$(pwd)" \
    --cpus 4 --mem-mb 16000 --time-s 3600 \
    --depends-on run_44a36f211a69440c8bfc0335bcd52077

--depends-on (repeatable) resolves the given run_ids to their Slurm job ids and passes --dependency=afterany:<job_id>[:<job_id>...] to sbatch — the second job won't start until the first one finishes, regardless of whether it succeeded (afterany, the default; override with --dependency-condition afterok to require success). This is the domain-CLI equivalent of the --array=0-3%1 throttling used in projects/recgenes/*.slurm to keep concurrent jobs from saturating shared I/O (Zarr, dbNSFP, VEP Flight) — except each step is now its own tracked TaskRun instead of an opaque array index.

A dependency must already have a Slurm job id (i.e. have been submitted) or submit fails fast instead of producing a broken sbatch --dependency:

iris-compute jobs submit brca1_extract "..." --depends-on run_does_not_exist

✗ run desconocido: run_does_not_exist

Cancel a job

iris-compute jobs cancel run_44a36f211a69440c8bfc0335bcd52077

✓ scancel enviado para slurm_job_id=424242

cancel only issues scancel — the tracked status still shows whatever it was before cancellation until the next jobs poll picks up Slurm's CANCELLED state and applies it.

Environment variables

Variable	Used by	Default
IRIS_REDIS_URL	build_repository()	redis://localhost:6379/0
IRIS_COMPUTE_SUBMIT_DIR	build_dispatcher()	~/.iris/compute/slurm
IRIS_COMPUTE_LOG_DIR	build_dispatcher()	$IRIS_COMPUTE_SUBMIT_DIR/logs

Troubleshooting

• sbatch failed with exit=... — submit_dir couldn't be created, or the cluster rejected the resource request (check --mem-mb/--time-s against the partition's limits). The rendered script is left under IRIS_COMPUTE_SUBMIT_DIR/<run_id>.sbatch for inspection either way.
• jobs status stays SUBMITTED forever — nothing has called jobs poll yet; it is not automatic.
• Connection refused from Redis — check IRIS_REDIS_URL and that the Redis process is reachable from the node running the CLI (the submitting shell, not the compute node the job itself runs on).

Reference

iris.applications.compute.use_cases.submit_task

Use case: build a TaskSpec from flat parameters and dispatch it to Slurm.

Sequencing (--depends-on in the CLI) is expressed as a Slurm --dependency=<condition>:<job_id>[:<job_id>...] via the dispatcher's raw escape hatch (DispatchHints.extra, see :meth:SlurmJobDispatcher._build_sbatch_args <iris.adapters.driven.dispatchers.slurm.SlurmJobDispatcher._build_sbatch_args>) rather than by extending the domain, which does not model dependency-aware dispatch. Consequently a dependency must already have been submitted (i.e. carry an external_ref with a Slurm job id) before it can be depended on — mirroring how sbatch --dependency itself requires an existing job id.

DependencyNotSubmittedError

Bases: RuntimeError

Raised when a depends_on run does not exist or has no Slurm job id yet.

submit_task

submit_task(
    *,
    repo,
    dispatcher,
    name,
    argv,
    workdir=None,
    env=None,
    cpus_per_rank=1,
    memory_mb=None,
    time_limit_s=None,
    nodes=None,
    gpus=0,
    queue=None,
    project=None,
    quality=None,
    nodelist=None,
    depends_on=None,
    dependency_condition="afterany",
)

Register a :class:TaskSpec, persist a fresh :class:TaskRun, and dispatch it.

Returns the run as left by :class:TaskDispatchRequestedHandler: SUBMITTED with external_ref carrying the Slurm job id.

RAISES	DESCRIPTION
DependencyNotSubmittedError	If any depends_on run is unknown or was never submitted to Slurm.

Source code in src/iris/applications/compute/use_cases/submit_task.py

def submit_task(
    *,
    repo: TaskRunRepository,
    dispatcher: JobDispatcher,
    name: str,
    argv: list[str],
    workdir: str | None = None,
    env: dict[str, str] | None = None,
    cpus_per_rank: int = 1,
    memory_mb: int | None = None,
    time_limit_s: int | None = None,
    nodes: int | None = None,
    gpus: int = 0,
    queue: str | None = None,
    project: str | None = None,
    quality: str | None = None,
    nodelist: str | None = None,
    depends_on: list[str] | None = None,
    dependency_condition: str = "afterany",
) -> TaskRun:
    """Register a :class:`TaskSpec`, persist a fresh :class:`TaskRun`, and dispatch it.

    Returns the run as left by :class:`TaskDispatchRequestedHandler`: ``SUBMITTED``
    with ``external_ref`` carrying the Slurm job id.

    Raises:
        DependencyNotSubmittedError: If any ``depends_on`` run is unknown or was
            never submitted to Slurm.
    """
    extra: list[tuple[str, str]] = []
    if nodelist:
        extra.append(("slurm.raw.nodelist", nodelist))
    if depends_on:
        job_ids = _resolve_dependency_job_ids(repo, depends_on)
        extra.append(("slurm.raw.dependency", f"{dependency_condition}:{':'.join(job_ids)}"))

    spec = TaskSpec(
        name=name,
        command=CommandSpec(argv=argv, workdir=workdir, env=env or {}),
        resources=ResourceRequest(
            cpus_per_rank=cpus_per_rank,
            memory_mb=memory_mb,
            time_limit_s=time_limit_s,
            nodes=nodes,
            gpus=gpus,
        ),
        hints=DispatchHints(queue=queue, project=project, quality=quality, extra=tuple(extra)),
        depends_on=tuple(depends_on or ()),
    )
    run = TaskRun(spec=spec)
    repo.save(run)

    handler = TaskDispatchRequestedHandler(repo=repo, dispatcher=dispatcher)
    handler.handle(TaskDispatchRequested(task_id=spec.task_id, run_id=run.run_id))

    return repo.get(run.run_id) or run

iris.applications.compute.use_cases.poll_tasks

Use case: sync every non-terminal run against Slurm's live state.

Slurm has no push notifications, so this is meant to be invoked periodically (cron, systemd timer, or a poll loop in a future API/UI process).

poll_tasks

poll_tasks(*, repo, dispatcher)

Poll Slurm for every tracked, non-terminal run and apply the resulting events.

Requires :class:RedisTaskRunRepository rather than the generic TaskRunRepository port: enumerating runs by status is a query-side concern the domain port does not declare (it only knows point lookups).

Source code in src/iris/applications/compute/use_cases/poll_tasks.py

def poll_tasks(*, repo: RedisTaskRunRepository, dispatcher: JobDispatcher) -> list[DomainEvent]:
    """Poll Slurm for every tracked, non-terminal run and apply the resulting events.

    Requires :class:`RedisTaskRunRepository` rather than the generic
    ``TaskRunRepository`` port: enumerating runs by status is a query-side
    concern the domain port does not declare (it only knows point lookups).
    """
    run_ids: set[str] = set()
    for status in _NON_TERMINAL_STATUSES:
        run_ids.update(repo.find_ids_by_status(status.value))

    runs = [repo.get(run_id) for run_id in run_ids]
    refs = [run.external_ref for run in runs if run is not None and run.external_ref is not None]
    if not refs:
        return []

    handler = JobObservationHandler(repo=repo)
    events: list[DomainEvent] = []
    for observation in dispatcher.poll(refs):
        events.extend(handler.handle(observation))
    return events

iris.applications.compute.use_cases.list_runs

Use case: list known TaskRuns, optionally filtered by status.

list_runs

list_runs(*, repo, status=None)

Return known runs, filtered by status (e.g. "RUNNING") when given.

Source code in src/iris/applications/compute/use_cases/list_runs.py

def list_runs(*, repo: RedisTaskRunRepository, status: str | None = None) -> list[TaskRun]:
    """Return known runs, filtered by ``status`` (e.g. ``"RUNNING"``) when given."""
    run_ids = repo.find_ids_by_status(status) if status else repo.find_all_ids()
    runs = (repo.get(run_id) for run_id in run_ids)
    return [run for run in runs if run is not None]