app/docs/code/workers.md

# Remote workers

The engine runs where the automations are. The GPU is somewhere else, the
Raspberry Pi with the relays is in a shed, and neither of them is on the same
network as the other.

A **worker** is a process that runs the code of nodes marked for it. It dials
*out* to the engine over one authenticated websocket, so nothing on that
machine has to be reachable, and nothing has to expose the engine's state
backend across hosts, which it never should.

## Install and attach

```sh
pip install fluksio-worker

fluksio-worker \
  --url wss://api.${DOMAIN}/api/v1/workers/attach \
  --token "$FLUKSIO_WORKER_TOKEN" \
  --labels gpu,cuda12 \
  --python /opt/torch-venv/bin/python
```

`fluksio-worker` is its own distribution: the agent, the node runner, and
`websockets`. Nothing of the engine, so a GPU box does not install a database
driver in order to run a training step. An engine host already has it, and
`fluksio worker …` is the same program.

| Option | Default | What it does |
|---|---|---|
| `--url` | **required** | `wss://…/api/v1/workers/attach` |
| `--token` | `$FLUKSIO_WORKER_TOKEN` | the credential, minted on the engine |
| `--name` | this host's name | sent along; the worker list shows the token's name |
| `--labels` | the token's | comma-separated; must match the labels the token was minted with |
| `--python` | this interpreter | the interpreter node code runs on |
| `--parallel` | `1` | how many node calls it will take at once |
| `--artifact-url` | derived from `--url` | where the artifact store is, if not beside the socket |
| `--cpus` | what the job or the machine has | cores to advertise |
| `--gpus` | what the job says, else counted | GPUs to advertise |
| `--ram-mb` | what the job or the machine has | memory to advertise, in MB |
| `--max-idle` | never | stop after this many seconds with nothing running |
| `--insecure` | off | allow `ws://` and `http://` to an engine on another machine |

The worker refuses a plain `ws://` or `http://` address unless the engine is on
this machine (`localhost`, `127.0.0.0/8`, `::1`): its token and every secret a
node is handed would cross the network in clear. `--insecure` is for a network
you trust, such as an engine on the same LAN without TLS. Node code gets this
host's environment less anything named like a credential (`*TOKEN*`,
`*SECRET*`, `*PASSWORD*`), the same as on the engine; hand a node a credential
as a [secret](../interface/operations.md#secrets) instead.

`--python` is the important one. It is how this machine keeps its own wheels
(the CUDA build, the vendor SDK, the thing that will not install anywhere else)
without the engine ever installing them or knowing about them.

### What it says it has

A worker reports its inventory when it attaches (cores, GPUs and memory) and
the engine schedules against it: a node asking for two cores and a GPU goes to
a machine that has them free, not merely to one carrying the right label.

Cores and memory are read off the machine, or off the batch job that started
this worker (`SLURM_CPUS_ON_NODE`, `SLURM_MEM_PER_NODE`). **What the job says
it was given always wins for GPUs** (`SLURM_GPUS_ON_NODE`, `SLURM_JOB_GPUS`,
or `FLUKSIO_WORKER_GPUS`): a node with eight cards may have granted this job
one, and advertising eight would be a lie the scheduler acts on. With nothing
said, NVIDIA's device nodes are counted, the same as the engine does for its
own machine, and `--gpus` overrides either. No vendor tool is asked, which is
what keeps the one dependency from becoming two. A worker that reports nothing
still attaches and is scheduled by its label alone, as every worker was before
any of them reported anything.

The engine tells each call what it may use: thread caps, and the devices it
may see. The worker starts a process per call, so it applies them at the only
moment a numerical library still reads them: before the import.

`--max-idle` is for a worker something else started for one job, a batch
scheduler say. It exits when nothing has run for that long, so the allocation
goes back rather than idling until its walltime.

## Mint the token

On the engine, as a superuser:

```sh
curl -X POST $FLUKSIO/workers/tokens -H "Authorization: Bearer $TOKEN" \
  -H 'Content-Type: application/json' \
  -d '{"name": "gpu-dev", "labels": ["gpu", "cuda12"]}'
```

Shown once, valid for a year, since a worker is a machine somebody sets up and
leaves running. The name and labels are signed into it: labels decide which
nodes, with their source and secrets, are sent to a worker, so a worker cannot
choose its own. It shows up under the token's name, and a worker whose
`--labels` differ from the token's is refused.

```sh
curl -s $FLUKSIO/workers/tokens -H "Authorization: Bearer $TOKEN" | jq
curl -X POST $FLUKSIO/workers/tokens/$JTI/revoke -H "Authorization: Bearer $TOKEN"
```

The first lists what was minted, by `jti` (the tokens themselves are not
kept); the second refuses one from then on, here and at the artifact
endpoint, and drops a worker attached with it. Rotating the keypair agent
tokens are signed with still revokes every worker at once. A token minted
before tokens carried their labels is refused: mint a new one.

??? note "A host where pip is not an option"

    The two files work copied into one directory and run with `python agent.py
    …`. The engine serves the runner itself at `GET /api/v1/workers/runtime`.
    It is the same module its own local workers run, deliberately standard
    library only.

## Send a node to it

A node declares the label of the machine it needs:

```json
{
  "id": "train",
  "device": "gpu",
  "device_policy": "require",
  "timeout": 7200
}
```

| `device_policy` | Behaviour when nothing carrying the label is attached |
|---|---|
| `require` (default) | the run stays `queued` and says what it is waiting for |
| `prefer` | it runs on the engine instead |

`prefer` is what makes a flow work before the GPU box exists. `require` is what
you want once it does.

!!! note "Not in the panel yet"

    `device` and `device_policy`, which machine a node's code runs on, are set
    through the API rather than the panel, with `PUT /flows/{name}`.

## What follows from this

- **The node's source travels with every call.** Nothing has to be deployed to
  the worker, and changing a node's code takes effect on the next execution.
- **A node bound to a device is compiled on that machine.** A node importing
  `torch` is correct on the GPU box and a missing module on the engine, so
  checking it here would fail something that is fine.
- **`import fluksio` inside a node is the worker's own reporter.** `emit`,
  `save_artifact`, `load_artifact`, installed before your code runs, so an
  installed `fluksio` package on that box never shadows it.
- **Cancelling a run kills what it is executing**, there or here, and leaves
  other runs of the same node alone.
- **If the worker disappears mid-call**, the run fails in seconds with
  `worker went away mid-call` rather than waiting out its timeout.
- **A worker sends a heartbeat every ten seconds while it executes**, so a long
  node is distinguishable from a dead socket. Ninety seconds of nothing at all,
  not even a heartbeat, fails the call as gone. A heartbeat says the *agent* is
  alive and nothing about the node, so it never satisfies a node's own timeout:
  one set to thirty seconds fires after thirty seconds of the node reporting
  nothing, wherever it runs.

## Artifacts across machines

An artifact reference names content by its hash, not a location, so it stays
valid wherever the store is reachable from. A worker that shares the engine's
filesystem writes to it directly; one that does not fetches and uploads over
HTTP, using the artifact endpoint beside the socket it already has. Either way
your node code is the same two calls.

A fetch is cached on the worker by digest, since content addressing means an
entry is never stale. Nothing expires on its own, so the cache is bounded by
size and the oldest fall out: `FLUKSIO_ARTIFACT_CACHE` says where it lives
(default a directory in the temporary directory) and
`FLUKSIO_ARTIFACT_CACHE_BYTES` how much it holds (default 1 GiB). Worth raising
where a worker reads the same large inputs repeatedly, and worth leaving alone
where it reads a media stream — those are chunks nothing asks for twice.

## Seeing what is attached

```sh
curl -s $FLUKSIO/workers -H "Authorization: Bearer $TOKEN" | jq
```

Name, labels, how many calls it will take at once, how many are in flight, when
it attached, when it was last seen, its Python version, a digest of its
environment, and what it says it has: cores, GPUs and memory.

## Upgrading

The engine and the worker speak a version-matched protocol, and a worker
announcing anything else is refused rather than half-understood. Protocol 2,
the one that carries inventory, is `fluksio-worker` 0.2.0. An older agent is
told so on the socket and stops, rather than retrying against an engine that
will never accept it; `pip install -U fluksio-worker` on that host is the whole
upgrade. Nothing changed in the runner served at `GET /api/v1/workers/runtime`,
so a host that copies its two files copies the same one as before.

A token minted before tokens carried their labels and a `jti` is refused. Mint
each existing worker host a new one with `POST /workers/tokens`, naming the
labels it should carry, and swap it in for `FLUKSIO_WORKER_TOKEN`.

**Upgrade the engine before its workers.** A worker sends its token as an
`Authorization` header on the handshake, where it stays out of access logs, and
an engine from before that reads it only from the URL: it refuses the newer
worker at the handshake. The engine still takes the old `?token=` for one
release, so older workers keep attaching to a newer engine meanwhile.

## Machines from a batch scheduler

A cluster is not a machine that attaches and stays; it is a queue somebody else
owns. So Fluksio does not submit *nodes* to Slurm. It submits a job whose payload
is an ordinary worker dialling back in, and from there everything works the way
it already does: the same protocol, the same artifacts, the same cancellation.

Write the clusters into `provisioners.json` beside the flows:

```json
[{
  "type": "slurm",
  "name": "hpc",
  "login": "me@login.cluster",
  "ssh_key": "/secrets/hpc_ed25519",
  "engine_url": "wss://api.${DOMAIN}/api/v1/workers/attach",
  "max_idle_s": 300,
  "provision_timeout_s": 900,
  "profiles": [{
    "name": "gpu-small",
    "cpus": 8, "gpus": 1, "ram_mb": 65536,
    "labels": ["gpu"],
    "sbatch": ["--partition=gpu", "--gres=gpu:1", "--time=04:00:00"],
    "prerun": ["module load cuda/12", "source ~/venvs/flux/bin/activate"]
  }]
}]
```

A **profile** is what the scheduler is asked for, where a flavor is what a node
asks for. They are separate on purpose, and agree when you set them up to.

When a node needs a machine nothing attached can give, and a profile fits, the
engine `sbatch`es one over the system `ssh`, and the run waits meanwhile, saying
so. `prerun` owns the environment: a `module load`, or a venv with
`fluksio-worker` already in it. The generated script runs no `pip install`.
It hands the worker a token of its own, good for 24 hours, as
`FLUKSIO_WORKER_TOKEN` rather than `--token`, so other users of a shared
compute node cannot read it from `ps`.

One outstanding request per profile, however often it is asked for. A job that
never attaches within `provision_timeout_s` is `scancel`led, as is anything
outstanding when the engine stops. `--max-idle` is what ends the job at the
other end, so an allocation goes back rather than idling to its walltime.

Both take `0` for "no limit": `provision_timeout_s: 0` waits for as long as the
queue does, which is what a cluster that queues overnight needs, and
`max_idle_s: 0` keeps the machine for the job's whole walltime.

`GET /workers/resources` reports what is outstanding and what last went wrong.

!!! note "It needs a route out"

    A compute node must be able to open a connection to the engine. That is
    true of most clusters and false of air-gapped ones; there is no staging
    path today.

## What a worker is not

It is not a second engine. Subscriptions, schedules, webhooks, the dashboards
and the run queue all stay in one process, which keeps a message having one
definition and a cron tick happening once. A worker executes node bodies.

Running two engines against one data directory is not supported. Distribute
work with workers.

## See also

- [Runs: pipelines that finish](../concepts/runs.md#running-a-node-somewhere-else)
- [Writing node code](nodes.md)
- [Getting started: data science](../getting-started/data-science.md)