API Reference

Every public export of @warlock.js/queue. For the narrative version of any of these, follow the cross-links into the guides.

import {
  defineJob,
  queueConnector,
  queueDashboard,
  failedJobs,
  retryFailedJob,
  setQueueConfig,
  getQueueConfig,
  resetQueueConfig,
  startWorkers,
  closeQueue,
  getQueue,
  getQueueDriver,
  runningWorkers,
  toMilliseconds,
} from "@warlock.js/queue";

import type {
  QueueConfig,
  LocalQueueConfig,
  QueueDriver,
  DriverDispatch,
  QueueWorkersConfig,
  JobOptions,
  JobBackoff,
  JobRetention,
  JobContext,
  JobProgress,
  JobDefinition,
  DispatchOptions,
  DispatchedJob,
  JobState,
  JobSnapshot,
  FailedJob,
  QueueJob,
  Duration,
  DurationUnit,
  CloseQueueOptions,
  QueueConnectorOptions,
  QueueDashboardOptions,
  DashboardServer,
} from "@warlock.js/queue";

Jobs

defineJob(definition)

function defineJob<TPayload, TResult = unknown>(
  definition: JobDefinition<TPayload, TResult>,
): QueueJob<TPayload, TResult>;

Registers definition.name in a process-wide registry and returns a QueueJob. Throws InvalidJobDefinitionError for an empty name, a missing handle, or a non-integer attempts < 1. See Defining jobs.

JobDefinition<TPayload, TResult> extends JobOptions with:

Field	Type	Notes
name	string	Unique job name
queue?	string	Default: queue.defaultQueue ("default")
handle	(payload: TPayload, context: JobContext) => Promise<TResult> | TResult	Throw to fail the attempt

JobOptions (also the shape of queue.defaultJobOptions):

Field	Type	Notes
attempts?	number	Total attempts including the first. Default 1
backoff?	JobBackoff	number (fixed ms) or { type: "fixed" | "exponential", delay }
removeOnComplete?	JobRetention	boolean | number | { age?, count? } — true removes at once, false keeps all, a number keeps the newest N, age (a Duration) drops older jobs. Default { age: "24h", count: 1000 }
removeOnFail?	JobRetention	Same shape, for failed jobs. Default { age: "7d", count: 5000 }
timeout?	Duration	Longest one attempt may run. Must be greater than zero. Default: no limit
failOnTimeout?	boolean	Fail for good on timeout instead of retrying. Default false

registeredJobs() / queueOf(definition)

New in 5.31. registeredJobs() returns every definition defineJob() registered in this process; queueOf(definition) the queue it runs on (its queue, else queue.defaultQueue). Neither touches Redis. warlock inspect reads them for its jobs section.

import { queueOf, registeredJobs } from "@warlock.js/queue";

for (const job of registeredJobs()) {
  console.log(job.name, queueOf(job));
}

QueueJob<TPayload, TResult>

Member	Type	Notes
name	string	readonly
queue	string	readonly, resolved queue name
dispatch(payload, options?)	Promise<DispatchedJob>	{ id, name, queue }
find(id)	Promise<JobSnapshot<TPayload, TResult> | undefined>	undefined if the job no longer exists, or belongs to a different name

DispatchOptions

Field	Type	Notes
delay?	Duration	number (ms) or `${number}${"ms"|"s"|"m"|"h"|"d"}`
priority?	number	1 is highest; omitted runs ahead of any prioritized job
jobId?	string	Dispatching again with a still-live id is a no-op — idempotent
attempts?	number	Overrides defineJob for this dispatch
backoff?	JobBackoff	Overrides defineJob for this dispatch
timeout?	Duration	Overrides defineJob for this dispatch
failOnTimeout?	boolean	Overrides defineJob for this dispatch

Precedence: dispatch options > defineJob > queue.defaultJobOptions.

JobContext

Field	Type
id	string
name	string
queue	string
attempt	number (1-based)
maxAttempts	number
progress(value: JobProgress)	Promise<void>
log(line: string)	Promise<void>
signal?	AbortSignal

signal is present when the job has a timeout (it aborts with a JobTimeoutError), and always on local/memory workers (it also aborts on lease loss or a shutdown timeout). See Timeouts.

JobSnapshot<TPayload, TResult>

type JobSnapshot<TPayload = unknown, TResult = unknown> = {
  id: string;
  name: string;
  queue: string;
  state: JobState; // "waiting" | "delayed" | "prioritized" | "active"
  // | "completed" | "failed" | "waiting-children" | "unknown"
  payload: TPayload;
  progress: JobProgress;
  attemptsMade: number;
  result?: TResult;
  failedReason?: string;
  createdAt: Date;
  finishedAt?: Date;
};

setDispatchInterceptor(interceptor) / getDispatchInterceptor()

Replace the broker for every job in the process: with an interceptor set, dispatch() calls interceptor.dispatch({ name, queue, payload, options }) and returns its DispatchedJob, and find() calls interceptor.find?.({ name, queue, id }). Nothing connects to Redis. Pass undefined to restore the broker. It is the seam queueFake (@warlock.js/core/tests) is built on; the slot lives on globalThis, so every copy of the package in a process shares it.

Failed jobs

failedJobs(options?)

function failedJobs(options?: {
  queue?: string; // default: the default queue
  start?: number; // default 0
  end?: number; // default 99, inclusive
}): Promise<FailedJob[]>;

Newest first. FailedJob adds payload, attemptsMade, failedReason, stacktrace: string[], failedAt?, and retry(): Promise<void>.

retryFailedJob(id, options?)

function retryFailedJob(id: string, options?: { queue?: string }): Promise<void>;

Throws FailedJobNotFoundError when no FAILED job in that queue has that id.

Configuration & lifecycle

QueueConfig

type QueueConfig = {
  driver?: "bullmq" | "local" | "memory"; // default bullmq
  local?: LocalQueueConfig;
  connection?: ConnectionOptions; // any BullMQ connection option, or an ioredis instance
  prefix?: string; // namespace, default "warlock"
  defaultQueue?: string; // default "default"
  defaultJobOptions?: JobOptions;
  workers?: QueueWorkersConfig;
};

type LocalQueueConfig = {
  path?: string; // SQLite file, default .warlock/queue.sqlite
  leaseDuration?: number; // ms, default 30000, minimum 30
  pollInterval?: number; // ms, default 100, minimum 1
};

type QueueWorkersConfig = {
  enabled?: boolean; // default true
  concurrency?: number; // default 1
  shutdownTimeout?: number; // ms, default 30000
};

queueConnector(options?)

function queueConnector(options?: { config?: QueueConfig }): Connector;

For warlock.config.ts > connectors. Runs in the late lifecycle phase. start() reads config.get("queue") (or uses options.config) and starts workers unless workers.enabled === false. shutdown() calls closeQueue(). See Configuration.

setQueueConfig / getQueueConfig / resetQueueConfig

Programmatic equivalents of the config file, for scripts, tests, and dispatch-only or worker-only processes.

startWorkers()

function startWorkers(): Promise<string[]>;

Starts one worker per queue that has a registered job (and keeps starting workers for queues whose first job is defined later). Returns [] when workers.enabled is false. Calling it again while workers run only starts the missing ones. Resolves once every started worker is ready.

closeQueue(options?)

function closeQueue(options?: { timeout?: number }): Promise<void>;

Stops workers taking new jobs, waits up to timeout (default workers.shutdownTimeout, else 30000) for active jobs, then force-disconnects any that are still running and closes every queue connection. Safe to call more than once, and when nothing was started.

getQueue(name) / runningWorkers()

getQueue returns (creating on first use) the BullMQ Queue for name. runningWorkers() lists the queue names with a running worker in this process.

getQueueDriver() / QueueDriver

getQueueDriver(): QueueDriver lazily creates the configured backend. The same instance powers every job and worker until closeQueue() closes it. The built-in BullMQ, SQLite, and memory drivers implement:

interface QueueDriver {
  dispatch(request: DriverDispatch): Promise<DispatchedJob>;
  find(queue: string, id: string): Promise<JobSnapshot | undefined>;
  failed(queue: string, start: number, end: number): Promise<FailedJob[]>;
  retry(queue: string, id: string): Promise<void>;
  start(queue: string): Promise<void>;
  running(): string[];
  close(timeout: number): Promise<void>;
}

DriverDispatch is { name, queue, data, options }: data is the payload/context envelope { __warlock: { v: 2, context? }, payload } that defineJob() writes for every dispatch (tools that show raw job data, such as the dashboard, show this wrapper), options is merged job/dispatch configuration with numeric millisecond delay and retention age. Prefer defineJob() and startWorkers() for normal use: they capture context, merge defaults, register handlers, and honour process roles and workers.enabled.

connection is required by BullMQ only. getQueue() and bull-board throw with the other backends. Local/memory contexts also have signal?: AbortSignal, aborted on lease loss or a shutdown timeout. Local shutdown releases leases; crashes re-deliver after expiry. See Local and memory queues.

toMilliseconds(duration: Duration): number

Converts a Duration (number or `${number}${DurationUnit}`, where DurationUnit is "ms" | "s" | "m" | "h" | "d") to milliseconds.

Dashboard

queueDashboard(server, options?)

function queueDashboard(
  server: DashboardServer,
  options?: { basePath?: string; queues?: string[] },
): Promise<void>;

Mounts bull-board on a Fastify-compatible server (Warlock’s getHttpServer() satisfies DashboardServer). Loads @bull-board/api and @bull-board/fastify via dynamic import() only when called; a missing one throws QueueDashboardDependencyError. See Dashboard.

Notifications

@warlock.js/notifications ships its own BullMQ dispatcher, bullmqQueue(), which lazily loads this package. See Notifications and @warlock.js/notifications’s own API reference.

Errors

Error	Thrown when
QueueNotConfiguredError	The queue is used before setQueueConfig() (or the connector) supplied a QueueConfig
JobTimeoutError	An attempt outlived its timeout; carries jobName, timeout (ms) and failOnTimeout
InvalidDurationError	A malformed Duration string, e.g. "10 minutes"
InvalidJobDefinitionError	defineJob called with an empty name, no handle, or a bad attempts
FailedJobNotFoundError	retryFailedJob(id) finds no FAILED job with that id
QueueDashboardDependencyError	queueDashboard() called without @bull-board/api / @bull-board/fastify installed