Bring your own container

Any OCI image that follows the unit contract below can run as a job. Its first units run as a calibration on up to two GPU classes, the time they take sets the price per unit, and the rest of the job runs at that price.

The job spec

name: checksums
image: ghcr.io/acme/checksum@sha256:<64 hex digits>
command: ["python", "/app/worker.py"]
input: files.jsonl             # or `units: 5000` when the units need no input
resources:
  gpu: { min_vram_gb: 16 }     # or `class: rtx4090`; leave it out for any GPU
  cpu: 4
  memory_gb: 16
  scratch_gb: 50
unit:
  max_seconds: 120             # a unit still running then is stopped and tried again
  max_attempts: 3
price:
  max_per_unit: 0.002          # optional: a calibration quote above this rejects the job; a later requote above it pauses the job until you accept or cancel
network: egress-only           # or none
checkpoint: auto               # or off
pool: public                   # or verified, platform, own

Only name, image, command and one of input or units are required.

image is a lowercase registry reference pinned by digest, <registry>/<path>@sha256:<64 hex digits>, so every node runs exactly the image you submitted. Nodes pull it without credentials, so it must be public (on GHCR a new package is private until you change its visibility).
command runs as the container’s command, after the image’s ENTRYPOINT if it has one.
input is a manifest with one JSON object per line, one unit per line (with the CLI, a local file, which it uploads). A line may carry its own unit_id: 1 to 64 letters, digits, ., _ or -, starting with a letter or digit. Otherwise units are numbered by line, 000000, 000001, …. A line is at most 65,536 characters, a manifest at most 32 MiB, a job at most a million units. units: N instead gives N units that need no input, each as {"unit_id": "000042", "index": 42}. Your worker reads the lines’ fields as it likes, and a line’s url, if it has one, may be any string your worker understands, with two exceptions: the key of a file your org uploaded (what unitfabric upload printed, see uploading inputs, or POST /v1/uploads returned: "kind": "input" takes a file of up to 4 GiB) reaches your worker as a file, which the node downloads before your worker sees the line and names in the line’s path, relative to /in, instead of its url (a line that has a path of its own keeps the url, as an https address), and another org’s upload is refused, however its key is spelled (422, naming the line).
resources is what each lease needs: gpu.min_vram_gb or gpu.class narrows the GPU, and the job’s CPU and memory become the container’s limits. The defaults are any GPU, 4 CPUs, 16 GB of memory and 50 GB of scratch disk. scratch_gb bounds what the lease keeps on the node for your worker: what it writes under /out and inside its own container, and the inputs of the units it has begun (a unit’s results are cleared from the node once it is reported). When that reaches it, the oldest running unit is failed and its files are deleted. Inputs the node fetches ahead of your worker, for units it has not begun, come on top, up to a quarter of it.
unit.max_seconds is 300 by default (1 to 86,400); unit.max_attempts is 3 by default (1 to 10). What fails an attempt is listed under guarantees.
network: egress-only, the default, sends traffic out through our gateway (see network), so the job runs on our machines alone; none gives the container no network.
checkpoint: auto, the default, allows checkpointing, which is best effort today: interrupted units usually start again from the beginning, and a unit is billed once either way (see checkpoint and restore). With off they always start again. A worker that uses CUDA IPC or unified memory cannot be checkpointed: give it off.
pool says whose machines may run the job: public, the default, verified, platform or own (see pools). A machine another organization supplies runs your image only if its owner accepts other organizations’ images.

The unit contract

The container is a long-running worker. The agent mounts /in (read-only) and /out (writable by any uid).

/in/units.jsonl: the units in this lease, one JSON object per line, each with unit_id plus the input fields from the manifest line. Appended to as the lease grows; the worker tails it (a torn final line is read again on the next pass).
Before working on a unit the worker writes /out/<unit_id>/started.json ({"unit_id": ..., "started_at": <unix seconds>}). It finishes the unit by writing /out/<unit_id>/done.json ({"unit_id": ..., "artifacts": ["image.png"], "seconds": 1.23}) once every named artifact exists as a regular file in that directory, or /out/<unit_id>/error.json ({"unit_id": ..., "error": "...", "trace": "...", "seconds": ...}) when the unit cannot be done. Both are written atomically (temporary file, then rename). The agent uploads the artifacts to output/<unit_id>/<lease_id>/ (the key the lease’s upload URLs carry, so a lost lease’s late upload can never overwrite a billed unit’s output), records each sha256, and reports the unit; an error.json is reported as a failed attempt with its message and trace, without restarting the worker.
A unit is attempted once per lease workspace: on start, and after a restart, the worker skips units that already have done.json, error.json or started.json. The agent fails a unit that has started.json but neither result after unit.max_seconds, restarts the worker, and the control plane retries the unit elsewhere up to max_attempts; a unit is never leased again to the lease that failed it.
Environment: GRID_LEASE_ID, GRID_JOB_ID, GRID_GPU_VRAM_GB, GRID_UNIT_MAX_SECONDS, and OMP_NUM_THREADS (the job’s CPU count). None of it names the node; the job’s network address carries the node’s index in our fleet, which identifies the node to us alone. On a machine another organization supplies, GRID_JOB_ID is an alias of the job’s id, a new one for each lease. Hostname is the lease id. The worker exits promptly on SIGTERM.
The container runs with no Linux capabilities, no new privileges, the job’s CPU and memory as cgroup limits (no swap beyond them) and a pids ceiling. A worker that needs chown, su or gosu will fail; run as the image’s user and write only under /out.

A minimal worker

gridworker.py, the reference implementation below, does everything the contract asks; a worker supplies only process(unit, workdir), which reads the unit’s fields, writes its artifacts into workdir and returns their names. This one downloads each unit’s url and stores its SHA-256:

import gridworker  # first: it handles SIGTERM from the first second
import hashlib
from pathlib import Path


def process(unit: dict, workdir: Path) -> list[str]:
    src = gridworker.fetch_input(unit, gridworker.IN, workdir)  # downloads unit["url"]
    with src.open("rb") as f:
        digest = hashlib.file_digest(f, "sha256").hexdigest()  # reads in chunks, never the whole file at once
    (workdir / "sha256.txt").write_text(digest + "\n")
    return ["sha256.txt"]


if __name__ == "__main__":
    gridworker.main(process)

FROM docker.io/library/python:3.12-slim
RUN pip install --no-cache-dir httpx
WORKDIR /app
COPY gridworker.py worker.py ./
CMD ["python", "/app/worker.py"]

Build it, push it to a public registry and use the digest the registry assigned (Podman’s local digest differs):

podman build -t ghcr.io/acme/checksum:1 .
podman push --digestfile digest.txt ghcr.io/acme/checksum:1
cat digest.txt        # sha256:...: the job spec's image is ghcr.io/acme/checksum@sha256:...

files.jsonl then lists one file per line, {"url": "https://example.com/files/0001.bin"}, and each unit’s results are sha256.txt and done.json.

Uploading inputs

A file that is not on the web goes up with unitfabric upload, at most 4 GiB of it, and the CLI prints its key:

$ unitfabric upload files/0001.bin
r2://grid/uploads/org_ists4trnmppr/up_he3rt7aicyig/0001.bin

A manifest line names the file by that key, and your worker gets the file itself: the node downloads it before your worker sees the line, which then names it in path, relative to /in, instead of url (gridworker.fetch_input reads either):

{"url": "r2://grid/uploads/org_ists4trnmppr/up_he3rt7aicyig/0001.bin"}

The key goes to stdout alone, so a loop can write the whole manifest, one upload per line, stopping at the first upload that fails (its one-line error on stderr, exit status 1); unitfabric submit then uploads the manifest itself, as the job’s input:

(for f in files/*.bin; do key=$(unitfabric upload "$f") || exit 1; echo "{\"url\": \"$key\"}"; done) > files.jsonl

Only your org’s uploads can be named: the key of another org’s upload, however it is spelled, is refused when you submit (422, naming the line).

gridworker.py

The reference implementation of the unit contract, used by every template. Import it before anything heavy: it installs the SIGTERM handler as it is imported. fetch_input() downloads a unit’s url (http or https, at most GRID_MAX_INPUT_BYTES, 4 GiB by default, within half the unit’s time) and cpu_quota() tells a worker how many threads its CPU limit affords.

"""The Grid unit contract (spec 6.2), implemented once for every template image.

A template module defines `process(unit: dict, workdir: Path) -> list[str]`: it reads the unit's input fields, writes
its artifacts into workdir, and returns their filenames. This module does everything else: tail /in/units.jsonl,
mark a unit started, skip units already started or finished, fetch inputs, time the call, check the artifacts,
write done.json or error.json atomically, remove downloaded inputs, keep going, and exit promptly on SIGTERM.
`cpu_quota()` tells a template how many threads its CPU quota affords.
"""
from __future__ import annotations

import json
import os
import re
import shutil
import signal
import sys
import threading
import time
import traceback
from pathlib import Path
from urllib.parse import urlparse

import httpx

IN = Path(os.environ.get("GRID_IN", "/in"))
OUT = Path(os.environ.get("GRID_OUT", "/out"))
UNIT_MAX_SECONDS = float(os.environ.get("GRID_UNIT_MAX_SECONDS", "0") or 0)
MAX_INPUT_BYTES = int(os.environ.get("GRID_MAX_INPUT_BYTES", str(4 << 30)))     # one unit's downloaded input
PLAIN_NAME = re.compile(r"[^/\\\x00]{1,255}")                                   # one path component (fullmatch)
RESERVED = ("started.json", "done.json", "error.json")                            # the library's files in a unit directory
CPU_MAX = Path("/sys/fs/cgroup/cpu.max")                                          # cgroup v2: "<quota> <period>" or "max <period>"

if threading.current_thread() is threading.main_thread():
    # Installed at import, not in main(): a template's module-level model load (tens of seconds) runs before main(),
    # and a stop during it must not wait out podman's 10-second SIGKILL. PID 1 has no default handler.
    signal.signal(signal.SIGTERM, lambda s, f: os._exit(128 + s))


def log(msg: str) -> None:
    print(msg, file=sys.stderr, flush=True)


_logged: set[str] = set()


def _log_once(msg: str) -> None:
    if msg not in _logged:
        _logged.add(msg)
        log(msg)


def _plain_name(n) -> bool:
    """One path component, valid UTF-8, at most 255 bytes: what a unit id or an artifact name may be. A name that
    os.fsdecode built from non-UTF-8 bytes carries surrogates, which neither JSON nor the agent's upload can carry."""
    if not isinstance(n, str) or not PLAIN_NAME.fullmatch(n) or n in (".", ".."):
        return False
    try:
        return len(n.encode("utf-8")) <= 255
    except UnicodeEncodeError:
        return False


def _text(s: str) -> str:
    """Valid UTF-8 for error.json: a message built from a non-UTF-8 file name would be refused downstream."""
    return s.encode("utf-8", "replace").decode()


def _describe(e: BaseException) -> str:
    """'ValueError: message', even when str(e) itself raises (the traceback module substitutes a placeholder)."""
    return "".join(traceback.format_exception_only(e)).strip()


def cpu_quota() -> int:
    """CPUs this container may use, for sizing a thread pool: the cgroup v2 quota (cpu.max), rounded up; where cpu.max
    cannot be read (cgroup v1), OMP_NUM_THREADS, which the agent sets from the same --cpus figure; 0 when neither says
    (no quota), and the caller keeps its library's default. Libraries size their pools to the host's cores, not the
    quota: x265 made a 32-thread pool in a 4-CPU quota and ran 1.76x slower than with 4."""
    try:
        quota, period = CPU_MAX.read_text().split()
        return -(-int(quota) // int(period)) if quota != "max" else 0
    except (OSError, ValueError, ZeroDivisionError):
        pass
    try:
        return max(0, int(os.environ.get("OMP_NUM_THREADS", "")))
    except ValueError:
        return 0


def fetch_input(unit: dict, inp: Path, workdir: Path) -> Path:
    """A local path for the unit's input: `path` is relative to /in (and must stay inside it), where the agent puts an
    upload of the job's org it fetched on the host, bounded by the same size cap as a download; `url` is downloaded into
    workdir with a size cap and an overall deadline (half the unit's time budget; each read is bounded by the same budget,
    capped at 60 s, so a server that stops sending cannot stall past it), and removed again on any failure."""
    if unit.get("path"):
        p = (inp / str(unit["path"])).resolve()
        if not p.is_relative_to(inp.resolve()):
            raise ValueError(f"input path escapes /in: {unit['path']}")
        if not p.is_file():
            raise FileNotFoundError(f"input path not found: {unit['path']}")
        if p.stat().st_size > MAX_INPUT_BYTES:
            raise ValueError(f"input is {p.stat().st_size} bytes; the limit is {MAX_INPUT_BYTES}")
        return p
    url = unit.get("url")
    if not url:
        raise ValueError("unit needs a `path` or a `url`")
    if urlparse(str(url)).scheme not in ("http", "https"):
        raise ValueError(f"input url must be http(s): {url}")
    name = Path(urlparse(url).path).name or "input"
    dst = workdir / ("input-" + name)
    if not UNIT_MAX_SECONDS:
        _log_once("GRID_UNIT_MAX_SECONDS is not set: downloads get the default 600 s budget")
    budget = UNIT_MAX_SECONDS / 2 if UNIT_MAX_SECONDS else 600
    read_timeout = min(60.0, budget)
    deadline = time.monotonic() + budget
    try:
        with httpx.stream("GET", url, timeout=httpx.Timeout(read_timeout), follow_redirects=True) as r:
            r.raise_for_status()
            declared = int(r.headers.get("content-length") or 0)
            if declared > MAX_INPUT_BYTES:
                raise ValueError(f"input is {declared} bytes; the limit is {MAX_INPUT_BYTES}")
            total = 0
            with dst.open("wb") as f:
                for chunk in r.iter_bytes():
                    total += len(chunk)
                    if total > MAX_INPUT_BYTES:
                        raise ValueError(f"input exceeds {MAX_INPUT_BYTES} bytes")
                    if time.monotonic() > deadline:
                        raise TimeoutError("input download exceeded the unit's time budget")
                    f.write(chunk)
    except httpx.TimeoutException as e:
        dst.unlink(missing_ok=True)
        raise TimeoutError(f"input download timed out ({type(e).__name__} after {read_timeout:g} s without data)") from None
    except BaseException:
        dst.unlink(missing_ok=True)
        raise
    return dst


class Manifest:
    """Tails /in/units.jsonl from where it left off. Bytes after the last newline (the agent may be mid-append) wait
    for the next pass; lines that do not parse or have no proper unit_id are skipped (the first few are logged, then
    every thousandth). A file shorter than the offset was truncated and rewritten, which the agent never does: it is
    read again from the start, and the markers in /out keep finished units from running twice."""

    def __init__(self, path: Path):
        self.path, self.offset, self.partial, self.skipped = path, 0, b"", 0

    def new_units(self):
        if not self.path.exists():
            return
        with self.path.open("rb") as f:
            size = os.fstat(f.fileno()).st_size
            if size < self.offset:
                log(f"units.jsonl shrank from {self.offset} to {size} bytes: reading it from the start")
                self.offset, self.partial = 0, b""
            f.seek(self.offset)
            data = self.partial + f.read()
            self.offset = f.tell()
        lines = data.split(b"\n")
        self.partial = lines.pop()
        for line in lines:
            if not line.strip():
                continue
            try:
                unit = json.loads(line)
            except (ValueError, RecursionError):
                unit = None
            if isinstance(unit, dict) and _plain_name(unit.get("unit_id")):
                yield unit
            else:
                self.skipped += 1
                if self.skipped <= 3 or self.skipped % 1000 == 0:
                    log(f"units.jsonl: skipped an unusable line ({self.skipped} so far): {line[:120]!r}")


def _write_json(path: Path, obj: dict) -> None:
    """Atomic: the agent reads done.json as soon as it appears, so it must never see a half-written file."""
    tmp = path.with_name(path.name + ".tmp")
    tmp.write_text(json.dumps(obj))
    os.replace(tmp, path)


def _names(result) -> list:
    """process()'s return value as a list. A string is refused (it is one name, not a list of them); any other iterable,
    a generator included, is consumed here, while the unit's downloaded inputs still exist."""
    if isinstance(result, (str, bytes)):
        raise TypeError("process() must return a list of artifact names, not a string")
    return list(result)


def _check_artifacts(workdir: Path, names: list) -> list[str]:
    """Every name a template returns must be a plain file name of a regular file it wrote: a unit is billed only when
    its output is stored, so a claimed-but-missing artifact is an error, not a done unit. The library's own names are
    refused: the three markers, `*.tmp` (its atomic-write temporaries) and `input-*` (downloaded inputs, removed once
    process() returns)."""
    out: list[str] = []
    for n in names:
        if not _plain_name(n) or n in RESERVED or n.endswith(".tmp") or n.startswith("input-") or n in out:
            raise ValueError(f"bad artifact name {n!r}")
        p = workdir / n
        if p.is_symlink() or not p.is_file():
            raise FileNotFoundError(f"artifact {n} was not written")
        out.append(n)
    return out


def _finished(workdir: Path) -> bool:
    return any((workdir / f).exists() for f in RESERVED)


def serve(process, inp: Path = IN, out: Path = OUT, poll_s: float = 0.5, stop: threading.Event | None = None) -> None:
    manifest = Manifest(inp / "units.jsonl")
    while not (stop and stop.is_set()):
        for unit in manifest.new_units():
            uid = unit["unit_id"]
            workdir = out / uid
            if _finished(workdir):                    # done, failed, or begun by a previous worker process: never again
                continue
            t0 = time.monotonic()
            try:
                workdir.mkdir(parents=True, exist_ok=True)
                _write_json(workdir / "started.json", {"unit_id": uid, "started_at": time.time()})
                try:
                    names = _names(process(unit, workdir))    # a generator runs to its end here, before its inputs go
                finally:
                    cleanup_inputs(workdir)               # downloaded inputs go whether the unit succeeded or not
                artifacts = _check_artifacts(workdir, names)
                _write_json(workdir / "done.json", {"unit_id": uid, "artifacts": artifacts, "seconds": round(time.monotonic() - t0, 3)})
            except (Exception, SystemExit) as e:      # a bad unit must not kill the worker, not even through sys.exit
                error = _text(_describe(e))
                try:
                    _write_json(workdir / "error.json", {"unit_id": uid, "error": error[:2000], "trace": _text(traceback.format_exc())[-4000:],
                                                         "seconds": round(time.monotonic() - t0, 3)})
                except OSError as write_error:        # a full disk: say so; the agent's per-unit timer handles the unit
                    log(f"unit {uid}: cannot write error.json: {write_error}")
                log(f"unit {uid} failed: {error}")
        time.sleep(poll_s)


def cleanup_inputs(workdir: Path) -> None:
    """Remove downloaded inputs (`input-*`). serve() calls it after every process() call, so a template need not."""
    for p in workdir.glob("input-*"):
        try:
            if p.is_dir() and not p.is_symlink():
                shutil.rmtree(p, ignore_errors=True)
            else:
                p.unlink()
        except OSError as e:
            log(f"cannot remove {p}: {e}")


def main(process) -> None:
    serve(process)

Calibration and the quote

A job with your own image has no list price. Its first units are its calibration:

The first node of a GPU class to ask for work takes 20 of the job’s units (all of them, if the job has fewer) as a calibration lease. At most two GPU classes calibrate at a time, and at most two are measured. Calibration needs a node with a GPU.
The node measures each calibration unit’s time itself, as the time between consecutive completions; the first also carries the worker’s start-up. A class’s rate comes from the median of those times, which, from three units up, leaves the start-up out, each counted at no more than unit.max_seconds. A median under 0.05 s is not a measurement, and the class tries again: give each unit enough work to take longer than that.
The quote is three times that rate’s cost per unit, the calibrating node’s hourly cost divided by the units it completes in an hour, and never below $0.00001.
The first class to finish quotes the job, and it runs at that price on the nodes of the classes it was calibrated on. If the second class then measures cheaper, the quote falls to its price, and units already billed at the higher one are refunded the difference: every unit of the job costs the better class’s quote.
price.max_per_unit caps the quote: a quote above it rejects the job, once no other class is still measuring.
At its quote the whole job must fit your credit that is not committed to other open jobs, as a template job must when it is submitted; otherwise the job is rejected. Submitting needs a positive balance, and nothing is reserved before the quote.
A class whose calibration lease ended without a rate (its node was lost, say) can try again. After six calibration leases without a quote the job is rejected.
The calibration units are units of your job: their results are kept and they are billed at the quote. Results are released only once the job has an agreed price: while it is calibrating, and for good if it is rejected or cancelled before its quote, asking for them answers 409. A rejected job bills nothing, its calibration units included, and its remaining units fail with the reason (rejected: quote above max_per_unit, for example).
The job’s run leases check the calibration: the control plane times your units there itself. A lease’s unit time is the larger of two measures: from its first result to its last, over the units after the first, and, for a lease that did at least half a batch (about five minutes of work at the calibrated rate), from the lease’s start to its last result, over all of its units, so results held back and released together still show how long the units took. A shorter lease is mostly your worker’s start-up, so only the time between its results counts. A lease’s start is when its node started your worker’s container, as the node reports it, so pulling the image does not count; a start-up inside your container (loading a model) does, since from outside it cannot be told from work: on short leases results released together are not caught, and such a start-up longer than a lease’s own work (about ten minutes for a whole batch) makes honest units look more than twice as slow as they are. Once three leases of a class have run, if the median unit takes more than twice as long as calibration measured, or the class’s leases taken together do (all their time, leases that reported nothing included, each less the start-up half a batch allows, over every unit they reported: work done on one lease and reported on the next shows there), the class takes the longer of the two as its unit time and the job is quoted again the same way. A higher quote within price.max_per_unit (or with no cap) that your uncommitted credit covers applies to the units billed from then on. Otherwise the job is paused, with the new quote in requote_usd: its leases end and its units wait until you accept the new quote with POST /v1/jobs/{job_id}/accept or cancel the job. Units billed before keep their price.
No node takes up a paused job until you decide, and nothing of it is billed at the new price. unitfabric job and unitfabric jobs show the price it ran at, the new one and why it waits for you (it is above the job’s max_per_unit, or your uncommitted credit does not cover it), and so does its page in the dashboard; through the API, quoted_price_usd is the price so far, requote_usd the new one and pause_reason why it waits, above_max_per_unit or insufficient_credit. To go on at the new price, accept it with unitfabric accept <job id>, the dashboard’s Accept new price or POST /v1/jobs/{job_id}/accept: your uncommitted credit must cover the difference on every unit of the job (402 otherwise; top up and accept again). An accept names the price it agrees to ({"requote_usd": "0.015000"} in the route’s body), and a job whose new price is another by then is refused (409, the proposed price changed: <the new one>) with nothing accepted. The dashboard names the price its page showed you, and unitfabric accept <job id> --price <p> the one unitfabric job showed you (its hint prints that line, ready to paste), so you never agree to a price you were not shown; without --price, unitfabric accept names the one it reads just before accepting, the price waiting when you run it. To stop, cancel the job. Units billed before keep their price either way.

unitfabric job <id> shows calibrating until the quote, then the price per unit.