Compute adapters

Driven adapters for the compute bounded context: a Slurm job dispatcher and a Redis-backed task run repository.

SlurmJobDispatcher

Implements JobDispatcher using sbatch / squeue / sacct / scancel.

from pathlib import Path
from iris.adapters.driven.dispatchers.slurm import SlurmJobDispatcher

dispatcher = SlurmJobDispatcher(
    submit_dir=Path("/scratch/project/.dispatch"),
    log_dir=Path("/scratch/project/logs"),   # defaults to submit_dir/logs
    queue="short",                           # passed as --partition
)

Polling strategy — poll_one() queries squeue first (live jobs); if the job is not found there it falls back to sacct (accounting data for finished jobs). Both squeue and sacct output is translated to ExternalJobState via slurm_state_to_external().

Script rendering — each run gets its own .sbatch file under submit_dir named after the run_id. The script sets TASK_ID and TASK_RUN_ID as environment variables so the payload can identify itself if needed.

Escape hatch for Slurm-specific options — pass rare sbatch flags via DispatchHints.extra:

from iris.domain.compute.objects import DispatchHints

hints = DispatchHints(
    queue="gpu",
    extra={
        "slurm.exclusive": "true",        # → --exclusive
        "slurm.raw.mail-type": "END,FAIL", # → --mail-type=END,FAIL
    },
)

RedisTaskRunRepository

Implements TaskRunRepository with Redis as the backing store. Stores the full TaskRun aggregate as a Redis Hash and maintains three secondary indices:

Key pattern	Type	Purpose
iris:compute:run:{run_id}	Hash	Full serialized TaskRun
iris:compute:ref:{dispatcher}:{external_id}	String	Reverse lookup by ExternalJobRef
iris:compute:runs	Set	All known run_ids
iris:compute:runs:status:{STATUS}	Set	run_ids grouped by status

Optimistic concurrency — save() watches the run key and retries on WatchError. When check_version=True (default), a write with an older version than the stored one raises StaleTaskRunError.

from iris.adapters.driven.repositories.redis_compute import make_redis_task_run_repository

# From environment variable IRIS_REDIS_URL, or defaults to redis://localhost:6379/0
repo = make_redis_task_run_repository()

# Explicit URL + optional TTL
repo = make_redis_task_run_repository(
    url="redis://redis-host:6379/1",
    default_ttl=86_400,   # expire run keys after 24 h
)

UI-oriented reads — two extra methods not in the port are available for dashboards:

all_ids = repo.find_all_ids()
running_ids = repo.find_ids_by_status("RUNNING")

Reference

Slurm dispatcher

iris.adapters.driven.dispatchers.slurm

Slurm implementation of the compute domain's :class:JobDispatcher port.

This is the only place where Slurm-specific concepts appear: sbatch, squeue, sacct, scancel and native states such as PD, R, CD, OOM, TIMEOUT. Everything crossing back into the domain is expressed as :class:ExternalJobRef / :class:JobObservation.

JobDispatchError

Bases: RuntimeError

Raised when a Slurm command fails.

SlurmJobDispatcher

Bases: JobDispatcher

Dispatch compute jobs to Slurm via sbatch / squeue / sacct.

Renders an sbatch script per run under submit_dir and submits it with --parsable for robust job-id parsing. Polling first consults squeue (live jobs) and falls back to sacct (accounting) for finished jobs.

submit

submit(request)

Render an sbatch script and submit it, returning the job reference.

Source code in src/iris/adapters/driven/dispatchers/slurm.py

def submit(self, request: DispatchRequest) -> ExternalJobRef:
    """Render an sbatch script and submit it, returning the job reference."""
    self.submit_dir.mkdir(parents=True, exist_ok=True)

    log_dir = self.log_dir or self.submit_dir / "logs"
    log_dir.mkdir(parents=True, exist_ok=True)

    script_path = self.submit_dir / f"{request.run_id}.sbatch"
    script_path.write_text(self._render_script(request), encoding="utf-8")

    args = self._build_sbatch_args(request, script_path, log_dir)

    proc = subprocess.run(args, check=False, text=True, capture_output=True)
    if proc.returncode != 0:
        raise JobDispatchError(
            f"sbatch failed with exit={proc.returncode}: {proc.stderr.strip()}"
        )

    job_id, cluster = parse_sbatch_parsable(proc.stdout)
    return ExternalJobRef(dispatcher=self.name, external_id=job_id, cluster=cluster)

cancel

cancel(ref)

Cancel the job identified by ref via scancel.

Source code in src/iris/adapters/driven/dispatchers/slurm.py

def cancel(self, ref: ExternalJobRef) -> None:
    """Cancel the job identified by ``ref`` via ``scancel``."""
    if ref.dispatcher != self.name:
        raise ValueError(f"cannot cancel ref for dispatcher={ref.dispatcher!r}")

    proc = subprocess.run(
        [self.scancel_bin, ref.external_id], check=False, text=True, capture_output=True
    )
    if proc.returncode != 0:
        raise JobDispatchError(f"scancel failed for {ref.external_id}: {proc.stderr.strip()}")

poll

poll(refs)

Yield a :class:JobObservation for each reference.

Source code in src/iris/adapters/driven/dispatchers/slurm.py

def poll(self, refs: Iterable[ExternalJobRef]) -> Iterable[JobObservation]:
    """Yield a :class:`JobObservation` for each reference."""
    for ref in refs:
        yield self.poll_one(ref)

poll_one

poll_one(ref)

Poll a single reference, preferring squeue then sacct.

Source code in src/iris/adapters/driven/dispatchers/slurm.py

def poll_one(self, ref: ExternalJobRef) -> JobObservation:
    """Poll a single reference, preferring ``squeue`` then ``sacct``."""
    if ref.dispatcher != self.name:
        raise ValueError(f"wrong dispatcher ref: {ref.dispatcher!r}")

    live = self._query_squeue(ref)
    if live is not None:
        raw_state, reason = live
        return JobObservation(
            ref=ref,
            state=slurm_state_to_external(raw_state),
            reason=reason,
            raw_state=raw_state,
        )

    accounting = self._query_sacct(ref)
    if accounting is not None:
        raw_state, exit_code = accounting
        return JobObservation(
            ref=ref,
            state=slurm_state_to_external(raw_state),
            exit_status=parse_exit_status(exit_code),
            reason=raw_state,
            raw_state=raw_state,
        )

    return JobObservation(
        ref=ref,
        state=ExternalJobState.UNKNOWN,
        reason="not found in squeue or sacct",
        raw_state=None,
    )

format_slurm_time

format_slurm_time(seconds)

Format a duration in seconds as a Slurm --time string.

Returns D-HH:MM:SS when the duration spans one or more days, otherwise HH:MM:SS.

Source code in src/iris/adapters/driven/dispatchers/slurm.py

def format_slurm_time(seconds: int) -> str:
    """Format a duration in seconds as a Slurm ``--time`` string.

    Returns ``D-HH:MM:SS`` when the duration spans one or more days, otherwise
    ``HH:MM:SS``.
    """
    if seconds < 1:
        raise ValueError("seconds must be >= 1")

    minutes, sec = divmod(seconds, 60)
    hours, minute = divmod(minutes, 60)
    days, hour = divmod(hours, 24)

    if days:
        return f"{days}-{hour:02d}:{minute:02d}:{sec:02d}"
    return f"{hour:02d}:{minute:02d}:{sec:02d}"

parse_exit_status

parse_exit_status(value)

Parse a Slurm ExitCode field (code or code:signal).

Source code in src/iris/adapters/driven/dispatchers/slurm.py

def parse_exit_status(value: str | None) -> ExitStatus | None:
    """Parse a Slurm ``ExitCode`` field (``code`` or ``code:signal``)."""
    if not value:
        return None

    value = value.strip()
    if value in {"", "Unknown", "None"}:
        return None

    if ":" not in value:
        try:
            return ExitStatus(code=int(value), signal=None)
        except ValueError:
            return ExitStatus(message=value)

    code_s, signal_s = value.split(":", 1)
    code = int(code_s) if code_s.isdigit() else None
    signal = int(signal_s) if signal_s.isdigit() else None
    return ExitStatus(code=code, signal=signal)

parse_sbatch_parsable

parse_sbatch_parsable(stdout)

Parse sbatch --parsable output into (job_id, cluster | None).

Output is either <jobid> or <jobid>;<cluster>.

Source code in src/iris/adapters/driven/dispatchers/slurm.py

def parse_sbatch_parsable(stdout: str) -> tuple[str, str | None]:
    """Parse ``sbatch --parsable`` output into ``(job_id, cluster | None)``.

    Output is either ``<jobid>`` or ``<jobid>;<cluster>``.
    """
    first_line = stdout.strip().splitlines()[0]
    parts = first_line.split(";", 1)
    if len(parts) == 1:
        return parts[0], None
    return parts[0], parts[1]

slurm_state_to_external

slurm_state_to_external(raw_state)

Map a raw Slurm state (short or long form) to an :class:ExternalJobState.

Source code in src/iris/adapters/driven/dispatchers/slurm.py

def slurm_state_to_external(raw_state: str) -> ExternalJobState:
    """Map a raw Slurm state (short or long form) to an :class:`ExternalJobState`."""
    state = raw_state.strip().upper().split()[0]

    mapping = {
        "PD": ExternalJobState.QUEUED,
        "PENDING": ExternalJobState.QUEUED,
        "CF": ExternalJobState.QUEUED,
        "CONFIGURING": ExternalJobState.QUEUED,
        "POWER_UP_NODE": ExternalJobState.QUEUED,
        "R": ExternalJobState.RUNNING,
        "RUNNING": ExternalJobState.RUNNING,
        "CG": ExternalJobState.RUNNING,
        "COMPLETING": ExternalJobState.RUNNING,
        "S": ExternalJobState.RUNNING,
        "SUSPENDED": ExternalJobState.RUNNING,
        "CD": ExternalJobState.SUCCEEDED,
        "COMPLETED": ExternalJobState.SUCCEEDED,
        "F": ExternalJobState.FAILED,
        "FAILED": ExternalJobState.FAILED,
        "LAUNCH_FAILED": ExternalJobState.INFRA_FAILED,
        "CA": ExternalJobState.CANCELLED,
        "CANCELLED": ExternalJobState.CANCELLED,
        "TO": ExternalJobState.TIMED_OUT,
        "TIMEOUT": ExternalJobState.TIMED_OUT,
        "DL": ExternalJobState.TIMED_OUT,
        "DEADLINE": ExternalJobState.TIMED_OUT,
        "OOM": ExternalJobState.OUT_OF_MEMORY,
        "OUT_OF_MEMORY": ExternalJobState.OUT_OF_MEMORY,
        "NF": ExternalJobState.INFRA_FAILED,
        "NODE_FAIL": ExternalJobState.INFRA_FAILED,
        "BF": ExternalJobState.INFRA_FAILED,
        "BOOT_FAIL": ExternalJobState.INFRA_FAILED,
        "PR": ExternalJobState.PREEMPTED,
        "PREEMPTED": ExternalJobState.PREEMPTED,
    }

    return mapping.get(state, ExternalJobState.UNKNOWN)

Redis repository

iris.adapters.driven.repositories.redis_compute

Redis-backed implementation of the compute domain's TaskRunRepository.

Stores the current state of each TaskRun as a Redis Hash and maintains secondary indices so the observation handler can resolve an external scheduler reference back to its run, and so a UI can list/filter runs by status without scanning.

Key layout

iris:compute:run:{run_id} Hash full serialized TaskRun iris:compute:ref:{dispatcher}:{external_id} String run_id (external-ref index) iris:compute:runs Set all run_ids iris:compute:runs:status:{STATUS} Set run_ids grouped by status

This is complementary to RedisEventPublisher (Redis Streams): the repository holds queryable current state, the publisher holds the append-only event log.

StaleTaskRunError

Bases: RuntimeError

Raised when a save would overwrite a newer version of a run.

RedisTaskRunRepository

RedisTaskRunRepository(
    client, *, check_version=True, default_ttl=None
)

Bases: TaskRunRepository

Persist TaskRun aggregates in Redis with external-ref and status indices.

PARAMETER	DESCRIPTION
client	Connected redis.Redis (expects decode_responses=True). TYPE: Redis
check_version	Reject writes whose version is older than the stored one. TYPE: bool DEFAULT: True
default_ttl	Optional TTL (seconds) applied to run keys on save. TYPE: int | None DEFAULT: None

Initialize with a connected Redis client.

Source code in src/iris/adapters/driven/repositories/redis_compute.py

def __init__(
    self, client: redis.Redis, *, check_version: bool = True, default_ttl: int | None = None
) -> None:
    """Initialize with a connected Redis client."""
    self._client = client
    self._check_version = check_version
    self._default_ttl = default_ttl

get

get(identifier)

Return the run with run_id == identifier, or None.

Source code in src/iris/adapters/driven/repositories/redis_compute.py

def get(self, identifier: str) -> TaskRun | None:
    """Return the run with ``run_id == identifier``, or None."""
    fields = cast(dict[str, Any], self._client.hgetall(self._run_key(identifier)))
    return self._load(fields) if fields else None

find_by_external_ref

find_by_external_ref(ref)

Return the run whose external_ref matches ref, or None.

Source code in src/iris/adapters/driven/repositories/redis_compute.py

def find_by_external_ref(self, ref: ExternalJobRef) -> TaskRun | None:
    """Return the run whose ``external_ref`` matches ``ref``, or None."""
    run_id = cast("str | None", self._client.get(self._ref_key(ref)))
    return self.get(run_id) if run_id else None

save

save(entity)

Persist entity atomically, updating all indices.

Uses optimistic concurrency: the run key is watched and the write is retried if a concurrent modification is detected. When check_version is set, a write carrying an older version than the stored one raises :class:StaleTaskRunError.

Source code in src/iris/adapters/driven/repositories/redis_compute.py

def save(self, entity: TaskRun) -> TaskRun:
    """Persist ``entity`` atomically, updating all indices.

    Uses optimistic concurrency: the run key is watched and the write is
    retried if a concurrent modification is detected. When ``check_version``
    is set, a write carrying an older ``version`` than the stored one raises
    :class:`StaleTaskRunError`.
    """
    run_key = self._run_key(entity.run_id)

    with self._client.pipeline() as pipe:
        while True:
            try:
                pipe.watch(run_key)
                prev = cast(dict[str, str], pipe.hgetall(run_key))

                if self._check_version and prev:
                    prev_version = int(prev.get("version", "0"))
                    if entity.version < prev_version:
                        pipe.unwatch()
                        raise StaleTaskRunError(
                            f"stale save for {entity.run_id}: "
                            f"{entity.version} < {prev_version}"
                        )

                pipe.multi()
                # Reindex: drop the run from its previous status set. The
                # external-ref key is stable per (dispatcher, external_id)
                # and only set once at submission, so it needs no cleanup.
                if prev:
                    pipe.srem(self._status_key(prev["status"]), entity.run_id)

                pipe.hset(run_key, mapping=self._dump(entity))
                pipe.sadd(f"{_PREFIX}:runs", entity.run_id)
                pipe.sadd(self._status_key(entity.status.value), entity.run_id)
                if entity.external_ref is not None:
                    pipe.set(self._ref_key(entity.external_ref), entity.run_id)
                if self._default_ttl:
                    pipe.expire(run_key, self._default_ttl)

                pipe.execute()
                break
            except redis.WatchError:
                continue  # concurrent modification — retry

    return entity

find_all_ids

find_all_ids()

Return all known run_ids (for a dashboard listing).

Source code in src/iris/adapters/driven/repositories/redis_compute.py

def find_all_ids(self) -> list[str]:
    """Return all known run_ids (for a dashboard listing)."""
    return sorted(cast(set[str], self._client.smembers(f"{_PREFIX}:runs")))

find_ids_by_status

find_ids_by_status(status)

Return run_ids currently in the given status.

Source code in src/iris/adapters/driven/repositories/redis_compute.py

def find_ids_by_status(self, status: str) -> list[str]:
    """Return run_ids currently in the given status."""
    return sorted(cast(set[str], self._client.smembers(self._status_key(status))))

make_redis_task_run_repository

make_redis_task_run_repository(
    url=None, *, check_version=True, default_ttl=None
)

Construye un RedisTaskRunRepository.

PARAMETER	DESCRIPTION
url	URL de Redis. Si es None, lee IRIS_REDIS_URL del entorno. Default: redis://localhost:6379/0 TYPE: str | None DEFAULT: None
check_version	Rechaza escrituras con una versión anterior a la guardada. TYPE: bool DEFAULT: True
default_ttl	TTL en segundos para las claves de run (None = sin expiración). TYPE: int | None DEFAULT: None

Source code in src/iris/adapters/driven/repositories/redis_compute.py

def make_redis_task_run_repository(
    url: str | None = None, *, check_version: bool = True, default_ttl: int | None = None
) -> RedisTaskRunRepository:
    """Construye un RedisTaskRunRepository.

    Args:
        url: URL de Redis. Si es None, lee IRIS_REDIS_URL del entorno.
             Default: redis://localhost:6379/0
        check_version: Rechaza escrituras con una versión anterior a la guardada.
        default_ttl: TTL en segundos para las claves de run (None = sin expiración).
    """
    redis_url = url or os.getenv("IRIS_REDIS_URL", "redis://localhost:6379/0")
    client = redis.from_url(redis_url, decode_responses=True)
    return RedisTaskRunRepository(
        client=client, check_version=check_version, default_ttl=default_ttl
    )