Shipplane
  • English

      • @shipplane/orchestrator

      @shipplane/orchestrator dispatches tasks from an @shipplane/core workflow to workers. Core manages workflow state and scheduling while the orchestrator handles task execution.

      The current alpha exports gateways for in-memory workers and worker threads. HTTP and SSH adapters are planned but not yet available.

      Install

      npm install @shipplane/orchestrator

      Usage

      import { createTaskRegistry, workflow } from "@shipplane/core";
import { createOrchestrator } from "@shipplane/orchestrator";
import { createInMemoryOrchestratorStore } from "@shipplane/orchestrator/store";
import { createInMemoryWorkerGateway } from "@shipplane/orchestrator/worker/in-memory";

const gateway = createInMemoryWorkerGateway();
const store = createInMemoryOrchestratorStore({
  onEvent: async (event) => {
    if (event.type === "task.log") {
      console.log(`[${event.taskId}] ${event.message}`);
    }
  },
});

const handlers = createTaskRegistry().register("shell.exec", {
  run: async (ctx) => {
    const command = String(ctx.params.command);
    await ctx.log("run command", { command });
    return `ok:${command}`;
  },
});

const worker = await gateway.registerWorker({
  workerId: "local-worker",
  handlers,
});

const wf = workflow("install-tools")
  .task("install", {
    type: "shell.exec",
    params: { command: "apt-get install -y git docker.io" },
  })
  .task("verify", {
    type: "shell.exec",
    dependsOn: ["install"],
    params: { command: "git --version && docker --version" },
  });

try {
  const orchestrator = createOrchestrator({ store, workerGateway: gateway });
  const started = await orchestrator.startRun({ workflow: wf.toJSON() });

  // Optionally cancel a running workflow or one task in a managed run:
  // await orchestrator.cancelTask(started.state.runId, "install");
  // await orchestrator.cancelRun(started.state.runId);

  const completed = await orchestrator.waitForRun(started.state.runId);

  console.log(completed.state.status);
} finally {
  await worker.stop();
}

      Public API

      Main export (@shipplane/orchestrator):

      • createOrchestrator({ store, workerGateway, recovery? })
        • startRun({ workflow })
        • waitForRun(runId)
        • cancelRun(runId)
        • cancelTask(runId, taskId)
      • createDispatchingTaskRegistry({ workflow, gateway })
      • createInMemoryWorkerDirectory({ heartbeatTimeoutMs?, now? })
      • createLeastActiveWorkerRouter(directory)
      • createRoutingWorkerGateway({ directory, router, endpoints })
      • WorkerGateway, WorkerJob, WorkerResultMessage

      Subpath exports

      • @shipplane/orchestrator/store
        • createInMemoryOrchestratorStore({ deepClone?, maxEvents?, maxSnapshots?, onEvent? })
        • OrchestratorStore
      • @shipplane/orchestrator/worker
        • createWorkerRuntime({ workerId, handlers })
        • runtime.run(job, context)
        • runtime.cancel(jobId)
        • runtime.stop()
      • @shipplane/orchestrator/worker/in-memory
        • createInMemoryWorkerGateway()
        • gateway.registerWorker({ workerId, handlers })
      • @shipplane/orchestrator/worker/thread
        • createWorkerThreadGateway(worker)
        • createPooledWorkerThreadGateway({ directory, router, workers })
        • createWorkerThreadWorker({ workerId, handlers, port })

      Next steps