createDurably

Creates a new Durably instance.

Signature

// Without jobs (use .register() later)
function createDurably<TLabels>(
  options: DurablyOptions<TLabels>,
): Durably<{}, TLabels>

// With jobs (1-step initialization)
function createDurably<TLabels, TJobs>(
  options: DurablyOptions<TLabels, TJobs> & { jobs: TJobs },
): Durably<TransformToHandles<TJobs, TLabels>, TLabels>

Options

interface DurablyOptions<
  TLabels extends Record<string, string> = Record<string, string>,
  TJobs extends Record<string, JobDefinition> = Record<string, never>,
> {
  dialect: Dialect
  pollingIntervalMs?: number
  maxConcurrentRuns?: number
  leaseRenewIntervalMs?: number
  leaseMs?: number
  preserveSteps?: boolean
  retainRuns?: string
  labels?: z.ZodType<TLabels>
  jobs?: TJobs
}

Option	Type	Default	Description
dialect	Dialect	required	Kysely dialect (SQLite, libSQL, or PostgreSQL)
pollingIntervalMs	number	1000	How often to check for pending jobs when the worker is idle (ms)
maxConcurrentRuns	number	1	Maximum runs the worker executes concurrently (still respects concurrencyKey exclusion in storage)
leaseRenewIntervalMs	number	5000	How often to renew the lease (ms)
leaseMs	number	30000	Lease duration — time until a job is considered stale (ms)
labels	z.ZodType	—	Zod schema for labels. Enables type-safe labels and runtime validation on trigger()
preserveSteps	boolean	false	Keep step output data on terminal runs. With false, a failed step.all() that has a successful sibling still retains its checkpoints and logs so the sibling output remains inspectable; other terminal runs clean up checkpoints. Durable attempt records are retained independently.
retainRuns	string	—	Auto-delete terminal runs older than this duration (e.g. '30d', '12h', '90m'). Throws if format is invalid.
jobs	TJobs	—	Job definitions to register. Shorthand for calling .register() after creation

Returns

Returns a Durably instance with the following methods:

init()

await durably.init(): Promise<void>

Initialize Durably: runs database migrations and starts the worker. This is the recommended way to start Durably. Equivalent to calling migrate() then start().

migrate()

await durably.migrate(): Promise<void>

Runs database migrations to create the required tables. Use init() instead for most cases.

start()

durably.start(): void

Starts the worker that processes pending jobs. Typically called after migrate(), or use init() for both.

stop()

await durably.stop(): Promise<void>

Stops the worker gracefully: no new work is claimed, pending delayed polls are cleared, and the returned promise resolves after every in-flight run and any idle-maintenance cycle started by the worker have finished.

register()

durably.register<TJobs extends Record<string, JobDefinition>>(
  jobs: TJobs
): { [K in keyof TJobs]: JobHandle }

Registers one or more job definitions and returns an object of job handles. Also populates durably.jobs with the same handles for type-safe access.

TIP

You can also pass jobs directly to createDurably() as a shorthand:

const durably = createDurably({
  dialect,
  jobs: { syncUsers: syncUsersJob, processImage: processImageJob },
})

const { syncUsers, processImage } = durably.register({
  syncUsers: syncUsersJob,
  processImage: processImageJob,
})

// Or access via durably.jobs
await durably.jobs.syncUsers.trigger({ orgId: '123' })

See defineJob for details.

on()

durably.on<E extends EventType>(
  event: E,
  handler: EventHandler<E>
): () => void

Subscribes to an event. Returns an unsubscribe function. See Events.

retrigger()

await durably.retrigger(runId: string): Promise<Run>

Retriggers a completed, failed, or cancelled run by creating a fresh run with the same input and labels (idempotency key is not carried forward). Returns the new Run object. Throws if the original input doesn't match the current job's input schema.

cancel()

await durably.cancel(runId: string): Promise<void>

Cancels a pending, leased, or waiting run. run:cancel is emitted as soon as the cancellation is written. With preserveSteps: false the run's checkpoints and logs are deleted afterwards; if that cleanup fails, cancel() still resolves and the failure is reported as worker:error with context: 'cancel-cleanup'.

waitForRun()

await durably.waitForRun(
  runId: string,
  options?: WaitForRunOptions
): Promise<Run<TLabels> & { status: 'completed'; output: unknown }>

Waits for an existing run to complete without creating a new run. Resolves only when the run reaches completed status. Throws NotFoundError if the run doesn't exist, CancelledError if cancelled, or Error if failed.

Listeners on this process settle the wait as soon as a matching terminal event is emitted. If another runtime completes, fails, or cancels the run against the same database, the wait observes the outcome by polling storage at the configured interval. Omitted pollingIntervalMs uses the instance createDurably({ pollingIntervalMs }) value (default 1000).

const run = await durably.waitForRun(runId, {
  timeout: 60000,
  onProgress: (p) => console.log(`${p.current}/${p.total}`),
  onLog: (l) => console.log(`[${l.level}] ${l.message}`),
})
console.log(run.output)

Option	Type	Description
timeout	number	Timeout in ms
pollingIntervalMs	number	Storage poll interval while the run is non-terminal; inherits instance default when omitted
onProgress	function	Called on live progress updates (no replay)
onLog	function	Called on live log entries (no replay)

deleteRun()

await durably.deleteRun(runId: string): Promise<void>

Deletes a run and its associated steps, attempts, logs, and labels.

purgeRuns()

await durably.purgeRuns(options: {
  olderThan: Date      // cutoff — terminal runs with completedAt before this are deleted
  limit?: number       // max rows to delete per call (default: 500)
}): Promise<number>

Deletes terminal runs (completed, failed, cancelled) with completedAt older than the cutoff. Returns the number of deleted runs. Associated steps, attempts, logs, and labels are cascade-deleted.

For automatic cleanup, use the retainRuns option instead (auto-purge uses a batch size of 100).

getRun()

await durably.getRun<T extends Run<TLabels> = Run<TLabels>>(runId: string): Promise<T | null>

Gets a single run by ID. Supports generic type parameter for type-safe access.

// Untyped (returns Run)
const run = await durably.getRun(runId)

// Typed (returns custom type)
type MyRun = Run & {
  input: { userId: string }
  output: { count: number } | null
}
const typedRun = await durably.getRun<MyRun>(runId)

getStepAttempts()

const attempts = await durably.getStepAttempts(runId)

Returns durable callback attempts ordered by start time, then ID. Unknown runs return []. Each attempt includes id, runId, stepName, stepIndex, leaseGeneration, metadata, startedAt, status (started, completed, or failed), nullable completedAt, nullable error, and nullable interruptionReason (lease-lost, cancelled, or unknown). An unresolved attempt does not prove that the callback began or when external work ended. Its inferred interruption reason may change until the run is terminal. Attempts survive preserveSteps: false cleanup but are deleted with the run. retainRuns applies only to terminal runs; cancel a perpetually reclaimed run to stop new attempts before retention can remove it.

metadata is a snapshot of the latest committed JSON value for that attempt. It starts as null if the step omits the metadata option; attempt.setMetadata() replaces the whole value.

getRuns()

await durably.getRuns<T extends Run<TLabels> = Run<TLabels>>(filter?: RunFilter<TLabels>): Promise<T[]>

interface RunFilter<TLabels extends Record<string, string> = Record<string, string>> {
  jobName?: string | string[]  // single or multiple job names
  status?: RunStatus | RunStatus[]  // single or multiple statuses
  labels?: Partial<TLabels>    // filter by labels (all specified must match)
  limit?: number
  offset?: number
}

Returns runs newest first: by creation time, then by ID for runs created in the same millisecond, so limit/offset pages are stable.

Gets runs with optional filtering and pagination. Supports generic type parameter.

// Filter by multiple job names
const runs = await durably.getRuns({
  jobName: ['sync-users', 'import-data'],
  status: 'completed',
})

// Typed getRuns
type MyRun = Run & {
  input: { userId: string }
  output: { count: number } | null
}
const runs = await durably.getRuns<MyRun>({ jobName: 'my-job' })

Run Type

The Run object returned by getRun() and getRuns():

interface Run<TLabels extends Record<string, string> = Record<string, string>> {
  id: string
  jobName: string
  input: unknown
  status:
    'pending' | 'leased' | 'waiting' | 'completed' | 'failed' | 'cancelled'
  idempotencyKey: string | null
  concurrencyKey: string | null
  currentStepIndex: number
  completedStepCount: number
  progress: { current: number; total?: number; message?: string } | null
  output: unknown | null
  error: string | null
  labels: TLabels
  leaseOwner: string | null
  leaseExpiresAt: string | null
  startedAt: string | null
  completedAt: string | null
  createdAt: string
  updatedAt: string
}

Field	Type	Description
id	string	Unique run ID
jobName	string	Name of the job
input	unknown	Input payload passed to the job
status	'pending' | 'leased' | 'waiting' | 'completed' | 'failed' | 'cancelled'	Current run status
idempotencyKey	string | null	Deduplication key
concurrencyKey	string | null	Concurrency group key
currentStepIndex	number	Index of the current step being executed
completedStepCount	number	Total number of completed steps
progress	{ current: number; total?: number; message?: string } | null	Latest progress report
output	unknown | null	Return value of the job (when completed)
error	string | null	Error message (when failed)
labels	TLabels (defaults to Record<string, string>)	Key/value labels for filtering (type-safe when schema provided)
leaseOwner	string | null	Worker ID that holds the lease (null when not leased)
leaseExpiresAt	string | null	ISO timestamp when the lease expires (null when not leased)
startedAt	string | null	ISO timestamp when the run started
completedAt	string | null	ISO timestamp when the run completed or failed
createdAt	string	ISO timestamp when the run was created
updatedAt	string	ISO timestamp of the last update

getJob()

durably.getJob(name: string): JobHandle | undefined

Gets a registered job by name.

subscribe()

durably.subscribe(runId: string): ReadableStream<DurablyEvent>

Subscribes to events for a specific run as a ReadableStream. The stream automatically closes when the run completes or fails.

Example

import { createDurably } from '@coji/durably'
import { createClient } from '@libsql/client'
import { LibsqlDialect } from '@libsql/kysely-libsql'

const client = createClient({ url: 'file:local.db' })
const dialect = new LibsqlDialect({ client })

const durably = createDurably({
  dialect,
  pollingIntervalMs: 1000,
  leaseRenewIntervalMs: 5000,
  leaseMs: 30000,
})

// Initialize (migrate + start)
await durably.init()

// Define and register jobs
import { defineJob } from '@coji/durably'
import { z } from 'zod'

const myJobDef = defineJob({
  name: 'my-job',
  input: z.object({ id: z.string() }),
  run: async (step, input) => {
    // ...
  },
})

const { myJob } = durably.register({ myJob: myJobDef })

// Or trigger via durably.jobs
await durably.jobs.myJob.trigger({ id: '123' })

// Clean shutdown
process.on('SIGTERM', async () => {
  await durably.stop()
})

See Also

• HTTP Handler — Expose Durably via HTTP/SSE for React clients
• defineJob — Define jobs with typed schemas
• Events — Subscribe to run and step events

Durable wait methods

await durably.signal(waitId, payload, { signalId }) // Promise<DurableWait>
await durably.getWait(waitId) // Promise<DurableWait | null>
await durably.getWaits(runId) // Promise<DurableWait[]>

These direct APIs persist and inspect external input for waits prepared by a job. signalId is required. Retry an uncertain delivery using the same ID and JSON payload. The returned receipt confirms durable acceptance, not job completion. The first signal or deadline result is immutable; a new signal at or after the deadline is rejected even before a worker sweep. Conflicting deliveries and unknown wait IDs are rejected. An identical retry of an accepted signal still returns its original receipt after the deadline or cancellation. Wait records are retained until their run is deleted. See durable external waits for replay, concurrency, cancellation, and usage constraints.

DurableWait

interface DurableWait {
  id: string
  runId: string
  name: string
  metadata: JsonValue | null
  status: 'pending' | 'resolved' | 'cancelled' | 'closed'
  payload: JsonValue | null
  signalId: string | null
  createdAt: string
  resolvedAt: string | null
  deadlineAt: string | null
  outcome: 'signal' | 'timeout' | null
  suspendedAt: string | null
  firstResumedAt: string | null
  inputWaitMs: number | null
  executionSlotWaitMs: number | null
}

A pending wait has no result. Resolved records have an immutable outcome: signal carries the accepted payload, while timeout has no signal. Cancelled and closed waits no longer accept input. createdAt is preparation time; deadlineAt is fixed on first preparation; resolvedAt is signal acceptance time or the deadline for a timeout. suspendedAt and firstResumedAt are written once when those transitions occur.

inputWaitMs measures time from suspension to signal acceptance or the deadline, floored at zero if the result wins during suspension handoff. It is zero if the result is consumed without suspension, and null while unresolved. executionSlotWaitMs measures time from the later of result finalization and suspension until the first resumed lease. It is zero when a result is consumed without suspension, and null until a suspended run resumes. A result can be finalized just before the suspension handoff; in that race, input wait is zero but slot wait can be positive. Wait records created before this migration may have unknown historical durations (null). A valid same-key lease can increase slot wait without changing input wait. These are per-wait direct-query fields, not aggregate job timings.

getWait returns null for an unknown ID, and getWaits returns an empty list for an unknown run. Signal delivery raises NotFoundError for missing IDs, ConflictError for conflicting or closed input, WaitExpiredError (a ConflictError subtype) for expired input, and ValidationError for invalid arguments. payload: null is also a valid signal: inspect outcome to distinguish it from timeout or missing input.