Conductor API¶

The Conductor manages task definitions, database connections, and task invocation.

Conductor.create()¶

Create a new Conductor instance:

import { Conductor } from "pgconductor-js";
import { TaskSchemas } from "pgconductor-js";

const conductor = Conductor.create({
  sql,                  // postgres.js SQL instance (or use connectionString)
  tasks,                // TaskSchemas from defineTask
  context,              // Custom context object
  middleware?,          // Optional per-attempt middleware
  logger?,              // Optional custom logger
  events?,              // Optional event definitions
  metadata?,            // Optional Standard Schema for execution metadata
  clock?,               // Optional time source, for tests
  telemetry?,           // Set to false to disable OpenTelemetry tracing
});

Parameters:

sql: postgres.js SQL instance (mutually exclusive with connectionString)
connectionString: Postgres connection string (mutually exclusive with sql)
tasks: TaskSchemas.fromSchema([...]) array of task definitions
context: Object passed to task handlers as ctx.{property} (see Custom Context)
middleware: (optional) Functions that run around every attempt of a non-batch task and can add to ctx (see Middleware)
logger: (optional) Custom logger implementation
events: (optional) Array of event definitions from defineEvent() for typed custom events
metadata: (optional) Standard Schema that types and validates execution metadata (see Execution Metadata)
clock: (optional) Object with now(): Date that workers use instead of the system clock for cron schedules and time windows, still corrected to database time. Tests pass createTestClock(sql) (see Testing)
telemetry: (optional) false disables OpenTelemetry spans and trace context propagation. Enabled by default, using the globally registered provider

Connection options:

// Option 1: Use existing postgres.js instance
const conductor = Conductor.create({
  sql: postgres(connectionString),
  tasks,
  context,
});

// Option 2: Provide connection string
const conductor = Conductor.create({
  connectionString: "postgres://localhost:5432/mydb",
  tasks,
  context,
});

With connectionString, the Conductor owns the connection pool. Call conductor.close() when you are done with it. A sql instance you pass in stays yours to close.

conductor.createTask()¶

Register a task handler:

const task = conductor.createTask(
  taskRef,              // { name, queue?, maxAttempts?, timeoutMs?, window?, removeOnComplete?, removeOnFail? }
  triggers,             // Trigger config or array of configs
  handler               // async (event, ctx) => result
);

taskRef:

{
  name: string;                           // Task name (required)
  queue?: string;                         // Queue name (default: "default")
  maxAttempts?: number;                   // Max retry attempts (default: 3)
  timeoutMs?: number;                     // Fail an attempt that runs longer (default: none)
  concurrency?: number;                   // Max concurrent executions (default: unlimited)
  window?: [string, string];              // Time window (e.g., ["09:00", "17:00"])
  removeOnComplete?: boolean | { days: number };  // Retention policy (true: at the next maintenance run)
  removeOnFail?: boolean | { days: number };      // Retention policy
  batch?: { size: number; timeoutMs: number };  // Batch processing config
  deadLetter?: { queue: string; task?: Task };    // Final-failure destination
}

triggers:

Single trigger:

{ invocable: true }
{ cron: "0 9 * * *", name: "schedule-name" }

Multiple triggers:

[
  { invocable: true },
  { cron: "0 9 * * *", name: "morning" },
  { cron: "0 17 * * *", name: "evening" },
]

handler:

async (event, ctx) => {
  // event.name: "pgconductor.invoke" | <schedule-name>
  // For invocable: event.payload contains the task payload
  // For cron: event.name is the schedule name (e.g., "hourly", "daily")
  // ctx: Task context with step(), sleep(), invoke(), logger, etc.

  return result; // Optional return value
}

conductor.invoke()¶

Invoke a task:

const executionId = await conductor.invoke(
  { name: "task-name", queue?: "queue-name" },
  payload,
  options?
);

Single invocation:

await conductor.invoke(
  { name: "send-email" },
  { to: "user@example.com", subject: "Hello" },
  {
    dedupe_key?: string,
    priority?: number,
    run_at?: Date,
    throttle?: { seconds: number },
    debounce?: { seconds: number },
    metadata?: Metadata,  // Execution metadata, at most 8 KB
  }
);

Batch invocation:

await conductor.invoke(
  { name: "process-item" },
  [
    { payload: { id: 1 }, dedupe_key: "item-1" },
    { payload: { id: 2 }, dedupe_key: "item-2" },
    { payload: { id: 3 }, priority: 10 },
  ]
);

conductor.createWorker()¶

Create a custom worker for a queue:

const worker = conductor.createWorker({
  queue: "queue-name",
  tasks: [task1, task2],
  config: {
    concurrency?: number,
    pollIntervalMs?: number,
    flushIntervalMs?: number,
    fetchBatchSize?: number,
  },
});

conductor.ensureInstalled()¶

Initialize database schema:

await conductor.ensureInstalled();

Called automatically by Orchestrator.start(), but useful for manual schema setup. Safe to call from several processes at once. Throws if the database schema is newer than this version of Postgres Conductor.

conductor.emit()¶

Emit a typed custom event:

await conductor.emit(
  "event-name",
  { /* event payload */ },
  options?: { metadata?: Metadata; dedupe_key?: string }
);

Returns the event ID. Events trigger tasks with matching event triggers. Triggered executions inherit the event's metadata.

dedupe_key deduplicates the emit: repeating it for the same event name emits nothing and returns the original event ID. Keys are scoped per event name and remembered for 1–2 days after the event was dispatched. See Deduplicating Emits.

See Task Triggers for details on event-triggered tasks.

conductor.cancel()¶

Cancel an execution:

const cancelled = await conductor.cancel(
  executionId,
  options?: { reason?: string }
);

Returns true if cancelled, false if already completed.

Executions it waits on with ctx.invokeAndWait() and executions it started with ctx.invoke(..., { cancelWithParent: true }) are cancelled with it. See Cancellation.

conductor.signal()¶

Send a signal to an execution. Returns false if the execution is completed, failed, cancelled or unknown. Sending a signal id again stores it once. See Signals.

await conductor.signal(executionId, { name: "input", id: requestId, payload: { text: "hi" } });

conductor.signalWithStart()¶

Signal the non-terminal execution of a task for a key, or start one that receives the signal first. Returns the execution ID.

const executionId = await conductor.signalWithStart({
  task: "session",
  queue: "default", // optional
  key: "thread-1",
  signal: { name: "input", id: requestId, payload },
  input: {}, // used only when it starts an execution (default: {})
  priority: 0, // optional
  metadata: {}, // optional
});

conductor.getExecution()¶

Read an execution by ID:

const execution = await conductor.getExecution(executionId);

// pass the task to type the result
const execution = await conductor.getExecution({ name: "send-email" }, executionId);

Returns null if the ID is unknown or the execution was removed by its retention policy. Otherwise:

{
  id: string;
  taskKey: string;
  queue: string;
  status: "pending" | "running" | "completed" | "failed" | "cancelled";
  result: TResult | null;     // return value, set when completed
  error: string | null;       // last attempt's error, or the cancellation reason
  attempts: number;
  createdAt: Date;
  completedAt: Date | null;
  failedAt: Date | null;
}

pending covers executions that wait for their first attempt, a retry, a sleep, a child or an event. cancelled is reported once the cancellation has settled; until then a running execution stays running.

conductor.waitForResult()¶

Poll an execution until it settles and return its result:

const result = await conductor.waitForResult(
  { name: "send-email" },     // optional, types the result
  executionId,
  {
    timeout?: DurationInput,  // milliseconds or a string such as "30s"; rejects with a TimeoutError
    pollIntervalMs?: number,  // default: 1000
    signal?: AbortSignal,     // rejects with the abort reason
  },
);

Resolves with the return value of a completed execution. The timeout also cancels a read in flight, so a blocked database cannot delay the rejection, and timeout: 0 rejects at once. Rejects with the error message if the execution failed or was cancelled, and rejects if the execution does not exist. The result is only available while the execution exists: with removeOnComplete: true the execution is removed when it completes, so waitForResult rejects with a not found error.

Inside a task, use ctx.waitForResult() instead: it suspends the task rather than polling and returns failures and cancellations as values.

conductor.close()¶

Close the connection pool created from connectionString:

await orchestrator.stop();
await conductor.close();

Stopping an orchestrator leaves the connection open, so the conductor can keep invoking tasks and the orchestrator can be started again. Close the conductor last. If you passed in a sql instance, close() does nothing; call sql.end() yourself.

What's Next?¶

Task Context API - Available methods in task handlers
Orchestrator API - Manage workers and task execution
Defining Tasks and Events - Define tasks with Zod schemas