Shipplane
  • English

      • Worker Routing

      Worker routing lets one orchestrator choose among multiple workers by task capability, active job count, and health.

      Use it when:

      • multiple workers can run the same task type
      • workers expose different task types
      • stale or unhealthy workers should be skipped
      • cancellation must be routed to the worker currently running a job

      Concepts

      • WorkerDirectory tracks worker ids, task types, health, heartbeats, and active jobs.
      • WorkerRouter selects a capable healthy worker. The default router chooses the least-active worker and keeps registration order as a tie-break.
      • RoutingWorkerGateway implements WorkerGateway, so it can be passed directly to createOrchestrator.

      In-memory workers

      The in-memory transport runs handlers in the same process. Use it for trusted dev/test flows, not sandboxing.

      import {
  createInMemoryOrchestratorStore,
  createInMemoryWorkerDirectory,
  createLeastActiveWorkerRouter,
  createOrchestrator,
  createRoutingWorkerGateway,
} from "@shipplane/orchestrator";
import { createInMemoryWorkerGateway } from "@shipplane/orchestrator/worker/in-memory";

const workerGateway1 = createInMemoryWorkerGateway();
const workerGateway2 = createInMemoryWorkerGateway();

await workerGateway1.registerWorker({ workerId: "worker-1", handlers });
await workerGateway2.registerWorker({ workerId: "worker-2", handlers });

const directory = createInMemoryWorkerDirectory();
await directory.register({ workerId: "worker-1", taskTypes: handlers.types() });
await directory.register({ workerId: "worker-2", taskTypes: handlers.types() });

const workerGateway = createRoutingWorkerGateway({
  directory,
  router: createLeastActiveWorkerRouter(directory),
  endpoints: [
    { workerId: "worker-1", gateway: workerGateway1 },
    { workerId: "worker-2", gateway: workerGateway2 },
  ],
});

const orchestrator = createOrchestrator({
  store: createInMemoryOrchestratorStore(),
  workerGateway,
});

      Worker thread pools

      Use createPooledWorkerThreadGateway when workers run behind worker thread ports. The pooled gateway listens for worker.ready and worker.heartbeat messages, updates the directory, and routes jobs through the same router.

      import {
  createInMemoryOrchestratorStore,
  createInMemoryWorkerDirectory,
  createLeastActiveWorkerRouter,
  createOrchestrator,
} from "@shipplane/orchestrator";
import { createPooledWorkerThreadGateway } from "@shipplane/orchestrator/worker/thread";

const directory = createInMemoryWorkerDirectory({
  heartbeatTimeoutMs: 5_000,
});

const workerGateway = createPooledWorkerThreadGateway({
  directory,
  router: createLeastActiveWorkerRouter(directory),
  workers: [
    { workerId: "worker-1", worker: workerThreadPort1 },
    { workerId: "worker-2", worker: workerThreadPort2 },
  ],
});

const orchestrator = createOrchestrator({
  store: createInMemoryOrchestratorStore(),
  workerGateway,
});

      Worker thread workers can send heartbeat messages by setting heartbeatIntervalMs:

      createWorkerThreadWorker({
  workerId: "worker-1",
  handlers,
  port,
  heartbeatIntervalMs: 1_000,
});

      Health and cancellation

      Registered workers start healthy. A worker becomes stale when now - lastHeartbeatAt > heartbeatTimeoutMs. The least-active router skips stale and unhealthy workers.

      During dispatch, the routing gateway records jobId -> workerId. requestCancel(jobId) uses that mapping to send cancellation to the active endpoint.