> ## Documentation Index
> Fetch the complete documentation index at: https://yukta.infenia.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Remote workers design

> Design for running processor nodes on remote workers: options considered, architecture, semantics, security, phases, risks and open decisions (F11).

<Note>
  Internal plan, restricted to project members. Back to the [internal index](/internal).
</Note>

Status: **phases 1 and 2 are built** (registry, leases, `GET /api/workers`, RSocket server and agent in the `workers` module, token auth; a processor node with `engine.placement` runs on a worker, with pool and free-slot choice). Failure handling beyond "the attempt fails and the engine retries", operations and control-plane HA are design only. Decision taken: scale out with **remote workers**
(a control plane schedules nodes, workers run them), not active/standby only. Needs F10 (PostgreSQL
persistence, first increment done) and benefits from F7/F9 (durable state, definition versions).

## Today (what the design builds on)

* One JVM runs everything. A workflow is a graph of `Flux` pipelines: a processor node is
  `processor.process(Flux<Message> input, config)`; terminals consume, triggers produce.
* The seam to hang remote execution on already exists: `ProcessorNodeAssemblerStrategy` builds the
  node pipeline and `NodeMessageChannel.inbound/outbound` wrap its edges. Everything around the plugin
  call is engine-side and stays on the control plane: merge of parents, pause/skip/breakpoint valves,
  retries and `onError` (`NodePolicies`), idle/element/deadline timeouts (`NodeActivity`),
  checkpoints (`StreamTopologyDecorator`), status and tracing (tracker).
* Plugins reach the engine only through the Reactor context (`executionId`, `nodeId`, heartbeat
  `Runnable`, result collector) and the `ControlSignalHandler`/`onControlSignal` path.
* Not shareable across JVMs today: in-memory `SESSION_MEMORY`/`KV_STORE`/`WAIT_FOR_SIGNAL` waits,
  `ActivePluginRegistry`, the scheduler's slots, the tracker queue.

## Goal and non-goals

Goal: run **processor nodes** on other machines, chosen by label/capacity, with the same semantics
as local nodes (back-pressure, timeouts, retries, pause/stop/skip, checkpoints, tracing), so CPU- or
IO-heavy nodes (process execution, LLM calls, SQL, HTTP, scripts) scale out and can be isolated.

Non-goals (first versions): moving triggers or terminals, running one node on several workers at
once (sharding a stream), exactly-once side effects, cross-region latency tuning, HA of the control
plane itself (separate step below).

## Options considered

| Option | What | Verdict |
| - | - | - |
| A. Remote node invocation | Control plane keeps the DAG; for a node with `engine.placement` it calls a worker over a streaming transport and splices the worker's output stream back | **Chosen.** Smallest change to the engine; reuses every policy that already wraps the plugin call |
| B. Whole runs on workers | A worker takes a complete run | Simple, but no per-node isolation, and control/pause/stop must be proxied to wherever the run is |
| C. Active/standby only | Several servers, one holds a DB lease | Does not scale out; keep as the HA step, not as the scale-out answer |

## Architecture (option A)

```text theme={"system"}
control plane                                   worker (same jar, `worker` profile)
 DAG, valves, policies, checkpoints,             plugin registry (the plugins it is allowed to run)
 tracker, scheduler, WorkerRegistry              WorkerAgent --- outbound connection ---+
   RemoteProcessor(node)  <== bidirectional stream (RSocket requestChannel) ==> RemoteNodeRunner
```

* **Transport: RSocket `requestChannel`** over TLS: one bidirectional stream per node *attempt*, with
  Reactive-Streams back-pressure end to end and cancellation that propagates (stop/timeout on the
  control plane cancels the stream; the worker disposes the plugin pipeline). The worker connects
  **out** to the control plane (no inbound ports on workers; works behind NAT).
* **Frames** on a node stream: `START{executionId, nodeId, type, config, attempt, deadlines}` →
  `MESSAGE{headers, payload}` *(both directions)* → `HEARTBEAT` (worker → control plane; calls
  `NodeActivity.touch`) → `COMPLETE` | `ERROR{class, message, retryable}`. Payloads are JSON (the
  checkpoint store already round-trips messages as plain JSON values); a size cap per frame, with a
  claim-check reference above it (decided together with the checkpoint store in PostgreSQL).
* **Control plane side:** `RemoteProcessor` is a `ProcessorPlugin` stub created by the assembler when
  the node has `engine.placement`. It picks a worker, opens the stream and returns the worker's
  output as the node's `Flux`. Because it sits where `processor.process` sits, retries, `onError`,
  timeouts, skip/pause valves, checkpoints and status keep working unchanged.
* **Worker side:** `WorkerAgent` registers (`REGISTER{workerId, pools, plugin types, capacity}`),
  sends `LEASE_HEARTBEAT` every few seconds, and runs `START` frames with the local plugin instance
  (`plugin.process(input, config)`), putting `executionId`/`nodeId`/heartbeat into the Reactor
  context exactly as the assembler does locally. `onControlSignal` (e.g. `WAIT_FOR_SIGNAL`) is
  forwarded as a `SIGNAL` frame to the stream of that node.
* **WorkerRegistry** (control plane): the set of live workers (lease with expiry), their pools and
  free slots; exposed in `GET /api/workers` and as metrics (`yukta.workers`, `yukta.worker.slots`).
  A worker that misses its lease is removed and its open streams fail with a retryable error.
* **Placement:** `engine.placement: {pool: "gpu", requires: ["LLM_CALL"]}`; default is `local`. The
  scheduler picks the live worker in the pool that has the plugin type and the most free slots.
  Nodes without `placement`, and all triggers/terminals, run locally. Plugins that keep process-local
  state (`SESSION_MEMORY`, `KV_STORE`, `WAIT_FOR_SIGNAL`, `SUB_WORKFLOW`) are marked `local-only`
  until their state is shared (F7 DB backend); validation rejects `placement` on them.

## Semantics to preserve (and how)

| Concern | How it carries over |
| - | - |
| Back-pressure | RSocket request-n credit on both directions of the channel |
| Idle timeout / heartbeat | worker `HEARTBEAT` frames call `NodeActivity.touch`; a silent worker times out like a silent local plugin |
| Retries, `onError`, failure budget | unchanged: they wrap the `RemoteProcessor` call; a lost worker is a retryable `IOException`-class error and the next attempt may choose another worker |
| Pause, skip, breakpoints | valves run on the control plane before the stream is opened and after output returns; a paused node simply stops requesting |
| Stop / cancel / drain | cancel of the stream; the worker disposes its pipeline and releases the slot |
| Checkpoints and resume | messages cross the control plane, so `StreamTopologyDecorator` records them as today; resume replays the boundary and re-runs the node on any worker |
| Tracing and metrics | trace context in `START`; the worker reports spans/metrics to the same collector, node status is still emitted by the control plane |
| Side effects | **at-least-once**: a node attempt may run twice if a worker dies after the effect and before `COMPLETE`; the message `id` and `attempt` header are passed so plugins can de-duplicate (as for retries today) |
| Secrets | `decrypted:` references are resolved on the control plane and sent over TLS in `START`, or resolved by the worker's own `SecretProvider` (setting `yukta.worker.secrets=local\|control-plane`) |

## Security

Worker authentication with mTLS (built: `yukta.workers.ssl-bundle` plus the RSocket server's SSL bundle with `client-auth=need`) or a per-worker token (built: one shared token); the control plane accepts
`REGISTER` only from authenticated workers, and the worker only runs plugin types on its own allow
list (so a compromised control plane cannot make a worker run `PROCESS_EXECUTOR`). Roles for the
REST API (F3) gain `ADMIN`-only worker endpoints.

## Phases (each its own reviewed PR)

0. Review this design; decide the open questions below.
1. **Worker registry and agent (done, token auth only; mTLS is left to the RSocket server's SSL settings):** RSocket server/client, mTLS/token, REGISTER/lease, `GET /api/workers`,
   metrics. No node runs remotely yet.
2. **Remote processor execution (done, tested with a real control plane and worker in one JVM):** `RemoteProcessor`, frames, one worker, happy path, cancel and
   heartbeat; acceptance test with two JVMs in Testcontainers/`ProcessBuilder`.
3. **Placement:** pools/requires, free-slot selection, `local-only` plugin flag and validation.
4. **Failure semantics (lost worker mid-attempt fails retryably and the retry lands on another worker: done and tested; drain on shutdown: done and tested; kill -9 chaos test with the worker in its own JVM: done):** lost worker, retry on another worker, drain on worker shutdown, chaos
   tests (kill -9 mid node), at-least-once documentation.
5. **Operations:** UI list of workers, per-worker load, `max-concurrent` accounting that counts remote
   slots, native-image support for the worker profile.
6. **Control plane HA (done: PostgreSQL lease, standby refuses starts, takeover recovers and auto-resumes; open: workers finding the new leader on their own, queued starts and waits surviving):** active/standby with a PostgreSQL lease; workers reconnect to the new leader;
   in-flight runs resume from checkpoints (needs the checkpoint store in PostgreSQL).

## Risks

* The in-memory control state (tracker, scheduler slots, signal waits) is the real single point of
  failure; phase 6 plus a PostgreSQL checkpoint store is what makes a control plane crash recoverable.
* RSocket adds a dependency and native-image reachability work; the alternative below avoids it.
* A big-payload node (files, long LLM outputs) makes the control plane a data hop; the claim-check
  reference (object store or PostgreSQL large objects) is needed before such nodes are placed.

## Decisions needed

1. **Transport:** RSocket (recommended: native back-pressure and cancel) vs gRPC bidirectional
   streaming (more tooling, same shape) vs plain WebSocket/HTTP streaming (least dependencies,
   back-pressure by hand).
2. **Packaging:** worker is the same boot jar with `SPRING_PROFILES_ACTIVE=worker` (recommended) or a
   separate slim artifact.
3. **Secrets:** resolve on the control plane or on the worker.
4. **First plugins to allow remotely:** suggest `PROCESS_EXECUTOR`, `HTTP_REQUEST`, `SQL_QUERY`,
   `LLM_CALL`, `MCP_TOOL_CALL`, `SCRIPT`.
5. **Payload limit** for a frame before claim-check is required.


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.