Shipplane
  • English

      • Route Tasks Across Workers

      Worker routing lets an orchestrator select from multiple workers based on 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 worker with the fewest active jobs and uses registration order to break ties.
      • 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 development and test workflows, not as a security sandbox.

      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 for workers connected through worker-thread ports. The pooled gateway listens for worker.ready and worker.heartbeat messages, updates the directory, and dispatches 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

      Newly registered workers start as 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.