Skip to main content

Remote Activity Workers

The Remote Activity Worker package (@bratsos/workflow-engine-host-remote) allows you to run heavy or specialized workflow stages (e.g. video transcoding, ffmpeg processing, or deep LLM inference) on separate, ephemeral worker machines.

Under this model, the worker nodes hold zero standing credentials (no database connection, and no root object-storage credentials). The worker leases tasks from the orchestrator, writes large binary outputs directly to object storage via presigned URLs, and reports execution results back to the central orchestrator using HTTP.

Architecture Overview​

Wiring Models​

Use defineRemoteStage to wrap your heavy stage definition. When reached, the orchestrator suspends the stage, registers it on the broker, and waits for a worker to execute it and report back. This releases the core job lease so the orchestrator isn't locked while waiting.

import { defineRemoteStage } from "@bratsos/workflow-engine-host-remote";
import { defineWorkflow } from "@bratsos/workflow-engine";

const workflow = defineWorkflow({ ... })
  .pipe(defineRemoteStage(heavyVideoStage, oTransport, {
    pollIntervalMs: 5_000,
    maxWaitMs: 3_600_000, // 1 hour timeout
    stageCodeVersion: "v1", // Must match the version on the workers
  }))
  .pipe(cleanupStage)
  .build();

2. ActivityExecutor Port (Short Tasks / Custom Routing)​

Alternatively, you can route execution dynamically using the kernel's ActivityExecutor port. The default executor is createLocalExecutor(), but you can wrap it with createRoutingExecutor to delegate select stage IDs to remote hosts:

import { createRemoteExecutor } from "@bratsos/workflow-engine-host-remote";
import { createRoutingExecutor, createLocalExecutor } from "@bratsos/workflow-engine/kernel";

const kernel = createKernel({
  ...,
  executor: createRoutingExecutor({
    remote: createRemoteExecutor(oTransport),
    remoteStageIds: ["heavy-transcode", "batch-embedding"],
  }),
});

Configuring the Worker Fleet​

Workers run in their own process, loading only @bratsos/workflow-engine-host-remote and the specific stage code (no Prisma client, no direct DB dependencies).

import { createActivityWorker, createHttpWorkerTransport } from "@bratsos/workflow-engine-host-remote";
import { transcodeStage } from "./stages/transcode";

const transport = createHttpWorkerTransport({
  baseUrl: "https://orchestrator-broker.internal",
  authToken: process.env.BROKER_AUTH_TOKEN,
});

const worker = createActivityWorker({
  workerId: "worker-gpu-1",
  stageIds: ["heavy-transcode"],
  stageCodeVersion: "v1",
  registry: new Map([["heavy-transcode", transcodeStage]]),
  transport,

  // Optional (v0.11+)
  onError: (error, { consecutiveFailures }) => {
    // Called when the poll/execute loop encounters errors (e.g. auth, lease fencing)
    console.error(`Worker error: ${error.message}. Consecutive failures: ${consecutiveFailures}`);
  },
  maxBackoffMs: 30_000, // Caps backoff delay between consecutive failures (default: 30s)
});

worker.start();

Configuring the Orchestrator (Broker)​

The orchestrator serves the HTTP endpoint that workers connect to.

import { Broker, InMemoryBrokerStore, createBrokerHttpServer, createS3Presigner } from "@bratsos/workflow-engine-host-remote";

const presigner = createS3Presigner({ 
  bucket: "my-artifacts", 
  region: "us-east-1" 
});

const broker = new Broker({
  store: new InMemoryBrokerStore(),
  presigner,
  stageCodeVersion: "v1",
  clock: { now: () => new Date() },
});

// Start the HTTP listener
const server = createBrokerHttpServer({ 
  broker, 
  objectStore: presigner, 
  authToken: process.env.BROKER_AUTH_TOKEN 
});
server.listen(3000);

Hosting on Other Platforms​

createBrokerHttpServer is a convenience wrapper around handleBrokerRequest, a platform-agnostic handler with no node:http dependency — it takes a plain IncomingRequest and returns a plain HandlerResponse. Call it directly to host the broker on any Request/Response-based platform — a Cloudflare Worker, Deno, Bun, an AWS Lambda function URL, or a framework like Express, Fastify, or Hono — by adapting that platform's request and response to and from the handler's shape.

For example, wiring it into a Cloudflare Worker fetch handler (reusing broker and presigner from above):

import { handleBrokerRequest, type BrokerServerDeps } from "@bratsos/workflow-engine-host-remote";

const deps: BrokerServerDeps = { broker, objectStore: presigner, authToken: process.env.BROKER_AUTH_TOKEN };

export default {
  async fetch(req: Request): Promise<Response> {
    const url = new URL(req.url);
    const result = await handleBrokerRequest(deps, {
      method: req.method,
      path: url.pathname + url.search,
      headers: Object.fromEntries(req.headers),
      body: await req.json().catch(() => null), // binary /blob PUTs: populate rawBody instead
    });
    const body = result.binary ?? (result.status === 204 ? null : JSON.stringify(result.json ?? null));
    return new Response(body, { status: result.status });
  },
};

Durability, Fencing, and Safety​

Durable Reports​

To prevent duplicate execution when the orchestrator restarts, workers write their results to a deterministic key in object storage before calling the HTTP /report API. On restart, the proxy stage reads this object storage reference and recovers the completed state automatically.

Fencing & Cancellation (v0.11+)​

When a workflow is cancelled on the orchestrator, active jobs are marked as cancelled. During the worker's heartbeat poll, the broker detects the cancellation and sends a cancel signal to the worker.

  • The worker ignores the output presigning and skips reporting to object storage, avoiding errors from executing reports on expired/fenced leases.

Version-Lock Safety​

Bumping stageCodeVersion acts as a deploy barrier. If a task was suspended using version v1 but a worker attempts to fetch it with version v2, the broker rejects the lease, preventing code mismatch errors.

What the worker's stage context carries​

The activity worker builds the stage context by hand, so a stage that runs remotely gets the full StageContext contract with these limits:

  • ctx.step is present but has no ledger: every ctx.step.* call throws StepLedgerNotConfiguredError. Durable steps run on the orchestrator's stages, not on remote workers.
  • ctx.ai and ctx.aiLogger throw AIServicesNotConfiguredError until the remote host wires services; create an AIHelper directly in the stage if it needs one.
  • ctx.abortSignal never fires — the kernel's heartbeat-driven abort does not cross the worker boundary. Cancellation reaches the worker through the broker's lease fencing described above.