Remote workers design

Internal plan, restricted to project members. Back to the internal index.

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)

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)

Review this design; decide the open questions below.
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.
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.
Placement: pools/requires, free-slot selection, local-only plugin flag and validation.
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.
Operations: UI list of workers, per-worker load, max-concurrent accounting that counts remote slots, native-image support for the worker profile.
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

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).
Packaging: worker is the same boot jar with SPRING_PROFILES_ACTIVE=worker (recommended) or a separate slim artifact.
Secrets: resolve on the control plane or on the worker.
First plugins to allow remotely: suggest PROCESS_EXECUTOR, HTTP_REQUEST, SQL_QUERY, LLM_CALL, MCP_TOOL_CALL, SCRIPT.
Payload limit for a frame before claim-check is required.

Get started

Concepts

Building workflows

Plugin reference

Running and controlling

Interfaces

Operations

Reference

Remote workers design

Today (what the design builds on)

Goal and non-goals

Options considered

Architecture (option A)

Semantics to preserve (and how)

Security

Phases (each its own reviewed PR)

Risks

Decisions needed

​Today (what the design builds on)

​Goal and non-goals

​Options considered

​Architecture (option A)

​Semantics to preserve (and how)

​Security

​Phases (each its own reviewed PR)

​Risks

​Decisions needed

Today (what the design builds on)

Goal and non-goals

Options considered

Architecture (option A)

Semantics to preserve (and how)

Security

Phases (each its own reviewed PR)

Risks

Decisions needed