api reference · types

Every TypeScript type, spelled out.

bunqueue is written in TypeScript and provides comprehensive type definitions. Exported types are imported from bunqueue/client; some shapes below (for example JobStateType, JobCounts, RepeatOptions, and AttemptRecord) are documented for reference but not exported.

Job Types

JobStateType

type JobStateType =
  | 'waiting' // In queue, priority = 0
  | 'prioritized' // In queue, priority > 0 (BullMQ v5)
  | 'delayed' // Waiting for delay to expire
  | 'active' // Currently being processed
  | 'completed' // Successfully finished
  | 'failed' // Failed after all retries (DLQ)
  | 'waiting-children' // Waiting for child jobs to complete (flows)
  | 'unknown'; // Job not found or invalid state

BullMQ v5 State Machine

bunqueue implements the full BullMQ v5 job state machine:

Job state machineBullMQ v5

push priority = 0

→

waiting

push priority > 0

→

prioritized

push delay > 0

→

delayed

→

waiting / prioritized delay expires

↓ pull

active

↓

completed ack

failed fail (terminal)

waiting / prioritized retry

flow dependencies

active

→

waiting-children

→

waiting children complete

Key differences from BullMQ v5:

• failed = BullMQ’s failed state. Internally stored in DLQ with metadata (reason, attempt history).
• prioritized = BullMQ’s prioritized state. Jobs with priority > 0 are in a separate logical state but share the same priority queue data structure.
• waiting-children = Parent jobs waiting for child flows to complete before becoming processable.

Job

The main job interface returned by Queue methods and passed to worker processors.

interface Job<T = unknown> {
  // ── Core Properties ──────────────────────────────────────────

  /** Unique job identifier (UUIDv7) */
  id: string;

  /** Job name/type */
  name: string;

  /** Job payload data */
  data: T;

  /** Queue name this job belongs to */
  queueName: string;

  /** Number of processing attempts made */
  attemptsMade: number;

  /** Job creation timestamp (ms since epoch) */
  timestamp: number;

  /** Current progress (0-100) */
  progress: number;

  /** Return value after successful completion */
  returnvalue?: unknown;

  /** Error message if the job failed */
  failedReason?: string;

  /** Parent job reference (if this job is part of a flow) */
  parent?: { id: string; queueQualifiedName: string };

  // ── Scheduling & Timing ──────────────────────────────────────

  /** Delay in ms before job becomes available for processing (0 for a negative delay option) */
  delay: number;

  /** Timestamp when job started processing */
  processedOn?: number;

  /** Timestamp when job finished (completed or failed) */
  finishedOn?: number;

  /** Ungrouped: higher runs sooner. Grouped: 0 runs first, then ascending. */
  priority: number;

  // ── Failure & Stall Tracking ─────────────────────────────────

  /** Stack traces from failed attempts */
  stacktrace: string[] | null;

  /** Number of times this job has been stalled */
  stalledCounter: number;

  // ── Metadata ─────────────────────────────────────────────────

  /** Parent key in format queueName:jobId */
  parentKey?: string;

  /** Original job options used when adding this job */
  opts: JobOptions;

  /** Lock token for this job (present when processing) */
  token?: string;

  /** Worker/client identifier processing this job */
  processedBy?: string;

  /** Deduplication ID (if set via jobId or deduplication option) */
  deduplicationId?: string;

  /** Repeat job key (for repeatable jobs) */
  repeatJobKey?: string;

  /** Number of times job processing has been started (includes retries) */
  attemptsStarted: number;

  // ── Core Methods ─────────────────────────────────────────────

  /**
   * Update job progress (0-100, clamped) with optional status message. It never throws
   * for the value, as in 2.9.10: NaN is 0, `'50'`/`true`/`null` are 50/1/0, other text is
   * 0 with the text as the message, and an object is 0 with its JSON as the message. The
   * Worker's `progress` event carries the value you passed.
   */
  updateProgress(progress: number | object, message?: string): Promise<void>;

  /** Add a log entry to the job */
  log(message: string): Promise<void>;

  /** Get the current state of the job */
  getState(): Promise<JobStateType>;

  /**
   * Remove this job: a waiting, delayed, prioritized or waiting-children job, or a
   * completed one (its data and result are deleted). Resolves when the job is gone
   * or does not exist; rejects when it is active or failed.
   */
  remove(): Promise<void>;

  /**
   * Retry this job. State-dispatched per BullMQ v5 contract:
   * - `failed` → requeue from DLQ (throws if not present)
   * - `active` → move to waiting (throws if move fails)
   * - `waiting`/`prioritized`/`delayed` → no-op
   * - other states → throws
   */
  retry(): Promise<void>;

  /**
   * Get the return values of all children jobs.
   * Keys are job keys (queueName:jobId), values are return values.
   */
  getChildrenValues<R = unknown>(): Promise<Record<string, R>>;

  // ── State Check Methods ──────────────────────────────────────

  /** Check if job is in waiting state */
  isWaiting(): Promise<boolean>;

  /** Check if job is currently active/processing */
  isActive(): Promise<boolean>;

  /** Check if job is delayed */
  isDelayed(): Promise<boolean>;

  /** Check if job has completed successfully */
  isCompleted(): Promise<boolean>;

  /** Check if job has failed */
  isFailed(): Promise<boolean>;

  /** Check if job is waiting for children to complete */
  isWaitingChildren(): Promise<boolean>;

  // ── Mutation Methods ─────────────────────────────────────────

  /**
   * Update the job's data payload (JSON serializable; no size limit, as in 2.9.10). Throws
   * for unserializable data and when the job cannot be updated
   * (`Job not found or cannot be updated`): the new data would otherwise be lost silently.
   */
  updateData(data: T): Promise<void>;

  /** Promote a delayed job to the waiting state; a job that is not delayed is left as is */
  promote(): Promise<void>;

  /**
   * Change the delay of a job (any finite number of ms; a negative delay makes the job
   * ready at once with a past run time, ahead of later ready jobs, as in 2.9.10). Throws for a NaN, infinite or non-numeric delay. A job that
   * can no longer be changed (gone, active or finished) is left as is, as in 2.9.10; a
   * Worker processor's own `changeDelay` throws instead, since its delivery must end.
   */
  changeDelay(delay: number): Promise<void>;

  /**
   * Change the job's priority: any finite number, for grouped jobs too (as in 2.9.10; a
   * numeric string is its number). A missing priority is 0 (`{ lifo: true }`, as in
   * BullMQ), and `lifo` is made a boolean like on add. Throws for a NaN, infinite or
   * non-numeric priority. A job that is no longer queued is left unchanged.
   */
  changePriority(opts: ChangePriorityOpts): Promise<void>;

  /**
   * Extend the job's lock duration. Returns the new duration on success, 0 if the lock
   * could not be extended (wrong token, lock expired, or no active lock). Any finite
   * duration is applied, as in 2.9.10 (`extendLock(token, 0)` resolves 0). Throws for a
   * NaN or infinite duration, and for any other broker rejection.
   */
  extendLock(token: string, duration: number): Promise<number>;

  /**
   * Clear job logs, optionally keeping the last N entries: 0 or a negative N clears all, a
   * fraction keeps its whole part, a numeric string is its number (as in 2.9.10). Throws
   * for NaN or text.
   */
  clearLogs(keepLogs?: number): Promise<void>;

  /**
   * Discard this job. Marks it to not be processed further.
   * The job will be moved to failed state (the DLQ) with reason `unknown` and no error message.
   */
  discard(): void;

  // ── Dependency Methods ───────────────────────────────────────

  /** Get job dependencies (children) with pagination */
  getDependencies(opts?: GetDependenciesOpts): Promise<JobDependencies>;

  /** Get count of job dependencies */
  getDependenciesCount(opts?: GetDependenciesOpts): Promise<JobDependenciesCount>;

  /** Get return values of failed children jobs */
  getFailedChildrenValues(): Promise<Record<string, string>>;

  /** Get ignored child failures (via ignoreDependencyOnFailure) */
  getIgnoredChildrenFailures(): Promise<Record<string, string>>;

  /** Remove this job's dependency relationship with its parent */
  removeChildDependency(): Promise<boolean>;

  /**
   * Remove the deduplication key associated with this job.
   * Returns false when this job no longer owns the key (for example, after a
   * replacement generation acquired it).
   */
  removeDeduplicationKey(): Promise<boolean>;

  /** Remove all unprocessed child jobs of this job */
  removeUnprocessedChildren(): Promise<void>;

  /**
   * Return every member of the current native processor batch.
   * Present only on jobs delivered through WorkerOptions.batch.
   */
  getBatch?(): Job<T>[];

  /**
   * Fail only this member while allowing the rest of its native batch to
   * complete. Present only on jobs delivered through WorkerOptions.batch.
   */
  setAsFailed?(error: Error): void;

  // ── Move Methods ─────────────────────────────────────────────

  /**
   * Move job to completed state.
   * @param returnValue - The return value of the job
   * @param token - Exact lock token, required when the job has a lock
   * @param fetchNext - Accepted for BullMQ signature compatibility. bunqueue
   * Workers fetch their next job through the polling loop, so this method does
   * not perform a chained fetch.
   * @returns null after the transition
   */
  moveToCompleted(returnValue: unknown, token?: string, fetchNext?: boolean): Promise<unknown>;

  /**
   * Move job to failed state.
   * @param error - The error that caused the failure
   * @param token - Exact lock token, required when the job has a lock
   * @param fetchNext - Accepted for BullMQ signature compatibility. bunqueue
   * Workers fetch their next job through the polling loop, so this method does
   * not perform a chained fetch.
   */
  moveToFailed(error: Error, token?: string, fetchNext?: boolean): Promise<void>;

  /**
   * Move job back to waiting state.
   * @param token - Exact lock token, required when the job has a lock
   * @returns true if job was moved
   */
  moveToWait(token?: string): Promise<boolean>;

  /**
   * Move job to delayed state.
   * @param timestamp - When the job should become available: a finite epoch-ms time (a
   *   past time means now); a NaN or infinite timestamp throws
   * @param token - Exact lock token, required when the job has a lock
   */
  moveToDelayed(timestamp: number, token?: string): Promise<void>;

  /**
   * Move job to waiting-children state.
   * Job will wait for all children to complete before processing.
   * @param token - Exact lock token, required when the job has a lock
   * @param opts - Options including child reference
   * @returns true if job was moved
   * Available in both embedded and TCP mode.
   */
  moveToWaitingChildren(
    token?: string,
    opts?: { child?: { id: string; queue: string } }
  ): Promise<boolean>;

  /**
   * Wait until the job has finished: completed, or failed with no retry left.
   * A failed attempt that will be retried does not end the wait.
   * @param queueEvents - QueueEvents instance to listen on, or null to wait without one
   * @param ttl - Maximum time to wait in ms. A positive number bounds the wait; 0
   *   (or any other non-positive or non-finite value) means no timeout. Omitted: no
   *   timeout with QueueEvents, 30000 without.
   * @returns The job's return value
   * @throws Error with the last attempt's failure reason; `Job <id> not found` when
   *   the job no longer exists (removed, or removed on completion with its retained
   *   result already gone), so its outcome is unknown; the timeout error when the TTL
   *   elapses first;
   *   `waitUntilFinished: the embedded engine was shut down` after shutdownManager()
   */
  waitUntilFinished(queueEvents: unknown, ttl?: number): Promise<unknown>;

  // ── Serialization Methods ────────────────────────────────────

  /** Get job as a typed JSON object */
  toJSON(): JobJson<T>;

  /** Get job as raw JSON (all values stringified) */
  asJSON(): JobJsonRaw;
}

JobJson

Typed JSON representation of a job.

interface JobJson<T = unknown> {
  id: string;
  name: string;
  data: T;
  opts: JobOptions;
  progress: number;
  delay: number;
  timestamp: number;
  attemptsMade: number;
  stacktrace: string[] | null;
  returnvalue?: unknown;
  failedReason?: string;
  finishedOn?: number;
  processedOn?: number;
  queueQualifiedName: string;
  parentKey?: string;
}

JobJsonRaw

Raw JSON representation with all values as strings.

interface JobJsonRaw {
  id: string;
  name: string;
  data: string; // JSON stringified
  opts: string; // JSON stringified
  progress: string; // JSON stringified
  delay: string;
  timestamp: string;
  attemptsMade: string;
  stacktrace: string | null; // JSON stringified
  returnvalue?: string; // JSON stringified
  failedReason?: string;
  finishedOn?: string;
  processedOn?: string;
  parentKey?: string;
}

ChangePriorityOpts

interface ChangePriorityOpts {
  /** New priority value */
  priority: number;
  /** Process in LIFO order after priority change */
  lifo?: boolean;
}

GetDependenciesOpts

interface GetDependenciesOpts {
  processed?: { cursor?: number; count?: number };
  unprocessed?: { cursor?: number; count?: number };
}

JobDependencies

interface JobDependencies {
  processed: Record<string, unknown>;
  unprocessed: string[];
  nextProcessedCursor?: number;
  nextUnprocessedCursor?: number;
}

JobDependenciesCount

interface JobDependenciesCount {
  processed: number;
  unprocessed: number;
}

JobOptions

Options when adding a job to a queue.

interface JobOptions {
  /** Ungrouped job priority (higher = processed sooner, default: 0) */
  priority?: number;

  /**
   * Delay in milliseconds before job becomes available (default: 0). A negative delay
   * makes the job ready at once with a run time in the past, so it runs ahead of later
   * ready jobs, as in 2.9.10. The maximum is `MAX_JOB_DURATION_MS` (4,320,000,000,000,000
   * ms, about 136,900 years); a longer finite delay is clamped to it.
   */
  delay?: number;

  /**
   * Maximum number of processing attempts (default: 3). `0` or less runs the job once (as
   * `1`), a fraction rounds up, and `Infinity` retries up to 2,147,483,647 times, as in
   * 2.9.10.
   */
  attempts?: number;

  /**
   * Backoff between retries. Either a delay in ms or a BackoffOptions object.
   * Default: 1000
   */
  backoff?: number | BackoffOptions;

  /** Processing timeout in milliseconds. Job fails if exceeded. */
  timeout?: number;

  /**
   * Custom job ID for idempotent/deduplication.
   * If a job with this ID already exists, the existing job is returned.
   */
  jobId?: string;

  /**
   * Remove job on completion. Boolean only: age/count retention
   * (number | KeepJobs) is not implemented and would be silently
   * ignored, so the type is narrowed to prevent the misuse.
   * Default: false
   */
  removeOnComplete?: boolean;

  /** Remove job on failure. Boolean only, see removeOnComplete. Default: false */
  removeOnFail?: boolean;

  /** Stall timeout in ms. Job is stalled if no heartbeat after this time. */
  stallTimeout?: number;

  /** Repeat configuration for recurring jobs */
  repeat?: RepeatOptions;

  /**
   * Request immediate admission persistence.
   * SQLite bypasses its write buffer; PostgreSQL admissions are already transactional.
   * Default: false (uses the SQLite buffer when that backend is selected)
   */
  durable?: boolean;

  /**
   * Parent job reference for flow dependencies.
   * When set, this job becomes a child of the specified parent.
   * The parent will wait for all children to complete before processing.
   */
  parent?: ParentOpts;

  /** Process jobs in LIFO order (newest first, default: false) */
  lifo?: boolean;

  /** Maximum stack trace lines to store on failure (default: 10) */
  stackTraceLimit?: number;

  /** Log entries to keep per job: stored in `job.opts` for BullMQ compatibility, not applied when logs are added */
  keepLogs?: number;

  /** Maximum job data size in bytes: stored in `job.opts` for BullMQ compatibility, not enforced (larger data is accepted) */
  sizeLimit?: number;

  /** Fail parent job if this child job fails (default: false) */
  failParentOnFailure?: boolean;

  /** Remove dependency relationship if this job fails (default: false) */
  removeDependencyOnFailure?: boolean;

  /** Continue parent processing even if this child fails (default: false) */
  continueParentOnFailure?: boolean;

  /** Move job to parent's failed dependencies instead of blocking parent (default: false) */
  ignoreDependencyOnFailure?: boolean;

  /** Job creation timestamp in ms (default: Date.now()) */
  timestamp?: number;

  /** Deduplication configuration */
  deduplication?: DeduplicationOptions;

  /** Debounce configuration */
  debounce?: DebounceOptions;

  /** Round-robin/FIFO job group membership */
  group?: GroupJobOptions;
}

GroupJobOptions

interface GroupJobOptions {
  /** Non-empty group identifier; safe integers are normalized to strings */
  id: string | number;

  /** Maximum pending jobs admitted atomically for this group */
  maxSize?: number;

  /** Integer from 0 to 2,097,151; lower values run first */
  priority?: number;
}

ParentOpts

interface ParentOpts {
  /** Parent job ID */
  id: string;
  /** Parent job queue name */
  queue: string;
}

BackoffOptions

interface BackoffOptions {
  /** Backoff strategy type: 'fixed', or 'exponential' (any other value runs as exponential) */
  type: 'fixed' | 'exponential';
  /**
   * Base delay in milliseconds (finite, >= 0; each retry delay is capped by maxDelay).
   * When missing at runtime, the 1000 ms default base is used.
   */
  delay: number;
  /** Upper bound for one retry delay in milliseconds (0 to 86,400,000). Default: 1 hour */
  maxDelay?: number;
}

All backoff delays include automatic jitter to prevent thundering herd:

• Exponential: ±50% jitter around the computed delay
• Fixed: ±20% jitter around the configured delay

Each retry delay is capped at maxDelay, or at 1 hour when maxDelay is not set. This prevents runaway delays at high attempt counts. maxDelay: 0 retries immediately.

// Exponential growth, but never wait more than 30 seconds between attempts
await queue.add('sync', data, {
  attempts: 10,
  backoff: { type: 'exponential', delay: 1000, maxDelay: 30_000 },
});

A maxDelay that is not a finite number between 0 and 86,400,000 (24 hours) is rejected on every path: Queue.add/addBulk in both modes, scheduler job templates, PUSH, PUSHB, HTTP push and atomic flows. Queue.getJob, Queue.getJobs, FlowProducer results, embedded add() and the jobs a Worker receives (embedded and TCP) return maxDelay in job.opts.backoff when it was set. A job added with a numeric backoff keeps the number in job.opts.backoff. The server applies the cap itself when it schedules a retry.

A processor that throws DelayedError postpones the job by its base backoff (backoff, or backoff.delay for the object form) without growth or jitter, falling back to 1000 ms when the base is 0 or negative, and capped at maxDelay (1 hour by default). The wait is never zero: maxDelay: 0 only makes failed attempts retry immediately, so a DelayedError job with maxDelay: 0 waits its base backoff, capped at 1 hour. See Postpone a job with DelayedError.

KeepJobs

Defined for BullMQ compatibility but not re-exported from bunqueue/client. Not accepted by per-job removeOnComplete/removeOnFail (those are boolean only); the equivalent shape is accepted by WorkerOptions.removeOnComplete/removeOnFail.

interface KeepJobs {
  /** Maximum age in milliseconds */
  age?: number;
  /** Maximum count of jobs to keep */
  count?: number;
}

RepeatOptions

Configuration for recurring/repeatable jobs.

interface RepeatOptions {
  /** Repeat every N milliseconds (alternative to pattern) */
  every?: number;

  /** Maximum repetitions (omit for infinite) */
  limit?: number;

  /** Cron pattern (alternative to every) */
  pattern?: string;

  /** Start date for repeat jobs */
  startDate?: Date | string | number;

  /** End date for repeat jobs */
  endDate?: Date | string | number;

  /** Timezone for cron pattern (e.g. 'America/New_York') */
  tz?: string;

  /** Execute immediately on start (default: false) */
  immediately?: boolean;

  /** Current repeat count (internal) */
  count?: number;

  /** Previous execution timestamp (internal) */
  prevMillis?: number;

  /** Offset in milliseconds */
  offset?: number;

  /** Custom job ID for repeat jobs */
  jobId?: string;
}

DeduplicationOptions

Prevent duplicate jobs from being added to the queue.

interface DeduplicationOptions {
  /** Unique deduplication ID (required) */
  id: string;

  /** TTL in milliseconds for the deduplication key */
  ttl?: number;

  /** Extend TTL when a duplicate job arrives (for debounce mode) */
  extend?: boolean;

  /** Replace job data when duplicate arrives while in delayed state */
  replace?: boolean;
}

DebounceOptions

Debounce job creation within a time window.

interface DebounceOptions {
  /** Unique debounce ID (required) */
  id: string;

  /** TTL in milliseconds for the debounce window (required) */
  ttl: number;
}

Processor

interface ProcessorContext {
  signal: AbortSignal;
}

interface ObservableLike<T> {
  subscribe(observer: {
    next(value: T): void;
    error(error: unknown): void;
    complete(): void;
  }): { unsubscribe(): void } | (() => void) | undefined;
}

type Processor<T = unknown, R = unknown> = (
  job: Job<T & FlowJobData>,
  context?: ProcessorContext
) => Promise<R> | ObservableLike<R> | R;

The Worker supplies a fresh AbortSignal for every delivery. A processing timeout, worker.cancelJob(), or worker.cancelAllJobs() aborts it. Promise processors are cooperative and must pass the signal to cancellable work or check signal.aborted; an ignored signal does not forcibly stop JavaScript. A structural Observable needs no RxJS dependency: its final next value is the job result, error fails the attempt, completion without a value fails, and abort unsubscribes it.

FlowJobData contains the optional flow-injected fields (__parentId, __parentQueue, __childrenIds, __flowParentId, __flowParentIds) that FlowProducer adds to broker-backed job.data when a job is part of a flow. They are engine-owned: flow creation rejects caller __* keys, and updateData() preserves the existing values rather than allowing them to be removed or replaced.

JobCounts

Returned by queue.getJobCounts().

interface JobCounts {
  waiting: number;
  prioritized: number;
  active: number;
  completed: number;
  failed: number;
  delayed: number;
  paused: number;
  'waiting-children': number;
}

Queue Types

QueueOptions

interface QueueOptions {
  /** Default job options applied to all jobs in this queue */
  defaultJobOptions?: JobOptions;

  /** TCP connection options (for server mode) */
  connection?: ConnectionOptions;

  /** Use embedded mode (in-process memory or SQLite, default: false) */
  embedded?: boolean;

  /**
   * SQLite path for embedded mode. A later explicit path must match the
   * process-wide manager's active database or construction throws.
   */
  dataPath?: string;

  /**
   * Auto-batching for queue.add() calls in TCP mode.
   * Buffers concurrent add() calls and sends them as a single PUSHB command.
   * Default: enabled for TCP mode, disabled for embedded mode.
   */
  autoBatch?: AutoBatchOptions;

  /**
   * Namespace prefix prepended to the queue name on the server.
   * Lets multiple environments (e.g. `dev:`, `prod:`) or tenants share
   * the same broker without their jobs, workers, cron schedulers, stats,
   * pause state, DLQ, or rate limits overlapping. `Queue.name` keeps
   * returning the logical name; only the server-side key is prefixed.
   * See the [Namespace Isolation](/guide/queue/advanced/#namespace-isolation-prefixkey)
   * guide.
   */
  prefixKey?: string;
}

AutoBatchOptions

interface AutoBatchOptions {
  /** Enable auto-batching (default: true for TCP, false for embedded; 'false'/0 also disable) */
  enabled?: boolean;
  /** Max items before auto-flush (default: 50; below 1 flushes every add, fractions round up, Infinity, NaN or a non-number = no size limit) */
  maxSize?: number;
  /** Max delay in ms before auto-flush (default: 5; a negative value, NaN, Infinity or a non-number is 0) */
  maxDelayMs?: number;
}

Jobs added with durable: true bypass the batcher and are sent as individual PUSH commands.

ConnectionOptions

interface ConnectionOptions {
  /** Server hostname (default: 'localhost'; a non-blank string) */
  host?: string;

  /** TCP port (default: 6789; a whole number from 1 to 65535, or its decimal string such as process.env.PORT) */
  port?: number;

  /** Declared but not read by the client: connections always use host/port */
  socketPath?: string;

  /** Enable TLS to the server: true (system CAs) or custom options (default: off) */
  tls?: boolean | ClientTlsOptions;

  /** Authentication token */
  token?: string;

  /**
   * Connection pool size for parallel operations.
   * Default: 4 for Queue/FlowProducer; Worker defaults to min(concurrency, 8).
   * Up to 65535; below 1 means one connection and a fraction rounds up.
   */
  poolSize?: number;

  /** Ping interval in ms for health checks (default: 30000; 0, a negative value or Infinity = disabled, otherwise >= 1) */
  pingInterval?: number;

  /** Command timeout in ms (default: 30000; >= 1, or Infinity = no client-side timeout) */
  commandTimeout?: number;

  /**
   * Consecutive command timeouts (no intervening success) before the connection
   * is concluded dead and a reconnect is forced (default: 3; 0, a negative value or NaN disables, a fraction rounds up).
   * Recovery path for a half-open socket, independent of the health-check ping.
   */
  maxCommandTimeouts?: number;

  /** Enable TCP pipelining (default: true) */
  pipelining?: boolean;

  /** Max commands in flight per connection (default: 100; > 0, a fraction rounds up, or Infinity) */
  maxInFlight?: number;
}

The host, port, token, timeout, interval and in-flight values are checked when the TCP connection is created (when the Queue, Worker, FlowProducer, QueueEvents or SandboxedWorker is constructed). A numeric string such as process.env.PORT is read as that number; otherwise a TypeError (not a number) or a RangeError names the option, e.g. TcpClient: pingInterval must be a finite number of milliseconds >= 0 or Infinity (got NaN). undefined or null keeps the default. Durations above 2^31 - 1 ms are honoured exactly.

ClientTlsOptions

interface ClientTlsOptions {
  /** Verify the server certificate (default: true). Set false for self-signed in dev. */
  rejectUnauthorized?: boolean;
  /** Path to a PEM CA certificate to trust (e.g. the self-signed server cert) */
  caFile?: string;
}

RateLimiterOptions

interface RateLimiterOptions {
  /** Maximum number of jobs to process in the duration window */
  max: number;

  /** Duration window in milliseconds */
  duration: number;

  /** Optional group key for per-group rate limiting */
  groupKey?: string;
}

Worker Types

WorkerOptions

interface WorkerOptions {
  /** Number of concurrent jobs (default: 1; > 0, a fraction rounds up, or Infinity) */
  concurrency?: number;

  /** Auto-run on creation (default: true) */
  autorun?: boolean;

  /** Heartbeat interval in ms (default: 10000; 0, a negative value or NaN disables, any other value is finite and >= 1) */
  heartbeatInterval?: number;

  /** TCP connection options (for server mode) */
  connection?: ConnectionOptions;

  /** Use embedded mode (in-process memory or SQLite, default: false) */
  embedded?: boolean;

  /**
   * SQLite path for embedded mode. A later explicit path must match the
   * process-wide manager's active database or construction throws.
   */
  dataPath?: string;

  /** Number of jobs to pull per batch (default: 10; > 0, a fraction rounds up, larger values and Infinity clamp to 1000) */
  batchSize?: number;

  /** Long poll timeout in ms when queue is empty (default: 0; a negative value or NaN is 0, larger values and Infinity clamp to 30000) */
  pollTimeout?: number;

  /**
   * Use lock-based job ownership.
   * When enabled, each pulled job gets a lock renewed via heartbeat.
   * Disable for high-throughput scenarios where stall detection is sufficient.
   * Default: true
   */
  useLocks?: boolean;

  /** Rate limiter configuration for controlling job processing rate */
  limiter?: RateLimiterOptions;

  /** Broker-authoritative job-group defaults; omitted means unlimited/disabled */
  group?: GroupWorkerOptions;

  /** Native BullMQ Pro-compatible batch processing */
  batch?: BatchWorkerOptions;

  /** Lock duration in ms (default: 30000; finite, >= 1 when useLocks). Sent to the server on pull; also used by stall detection. */
  lockDuration?: number;

  /** Max stalls before moving to failed (default: 1). Applied to stall config in embedded mode. */
  maxStalledCount?: number;

  /** Skip stalled job check, disables the stalled event subscription (default: false) */
  skipStalledCheck?: boolean;

  /** Skip lock renewal via heartbeat (default: false) */
  skipLockRenewal?: boolean;

  /** Delay in ms between polls when the queue is drained (default: 50; finite, >= 1 when pollTimeout is 0) */
  drainDelay?: number;

  /** Remove jobs on complete, applied as default for all jobs processed by this worker */
  removeOnComplete?: boolean | number | { age?: number; count?: number };

  /** Remove jobs on fail, applied as default for all jobs processed by this worker */
  removeOnFail?: boolean | number | { age?: number; count?: number };

  /**
   * Namespace prefix; must match the producing `Queue.prefixKey` to
   * consume its jobs. The Worker is registered under `prefixKey + name`
   * on the server, so two workers with the same logical queue name but
   * different prefixes never see each other's jobs. See the
   * [Namespace Isolation](/guide/queue/advanced/#namespace-isolation-prefixkey) guide.
   */
  prefixKey?: string;
}

The constructor validates the four durations and the two counts (concurrency, batchSize) before it starts a timer or contacts the broker. A numeric string is read as that number; otherwise a TypeError (not a number) or a RangeError names the option, e.g. Worker: heartbeatInterval must be a finite number of milliseconds >= 1 (got NaN). undefined or null keeps the default. Durations above 2^31 - 1 ms are honoured exactly. See Duration validation.

GroupWorkerOptions

interface GroupWorkerOptions {
  /** Maximum active jobs per group; omitted means unlimited */
  concurrency?: number;

  /** Fixed-window starts allowed independently for every group */
  limit?: { max: number; duration: number };
}

BatchWorkerOptions

interface BatchWorkerOptions {
  /** Maximum jobs in one processor invocation (1..1000) */
  size: number;

  /** Minimum members before starting (default: 1; must be <= size) */
  minSize?: number;

  /** Maximum wait for minSize in ms; omitted/0 waits indefinitely */
  timeout?: number;

  /** Keep every member in a batch on the same group ID (default: false) */
  groupAffinity?: boolean;
}

With batch, concurrency counts concurrent processor invocations, while each invocation may own up to batch.size independently leased jobs. The processor is called once with a leading job; job.getBatch() returns all members and member.setAsFailed(error) selectively fails one. Without groupAffinity, a batch that contains grouped work does not wait for minSize. With affinity, the broker and Worker keep one group ID per batch and the minimum-size wait is honored.

Worker Pro control methods

worker.cancelJob(jobId: string, reason?: string): boolean;
worker.cancelAllJobs(reason?: string): void;
worker.isJobCancelled(jobId: string): boolean;
worker.rateLimitGroup(job: Job, duration: number): Promise<void>;

Cancellation targets only deliveries active in that Worker and aborts their processor signals. rateLimitGroup requires a grouped active job, installs a broker-authoritative manual deadline, and moves that delivery back to waiting. It does not require WorkerOptions.group.limit.

Worker Events

// Worker emits these events:
worker.on('ready', () => void);
worker.on('active', (job: Job) => void);
worker.on('completed', (job: Job, result: R) => void);
worker.on('failed', (job: Job, error: Error) => void);
worker.on('progress', (job: Job | null, progress: number) => void); // the value passed to updateProgress: annotate `number | object` if you pass objects
worker.on('stalled', (jobId: string, reason: string) => void);
worker.on('cancelled', (data: { jobId: string; reason: string }) => void);
worker.on('log', (job: Job, message: string) => void);
worker.on('error', (error: Error) => void);
worker.on('drained', () => void);
worker.on('closed', () => void);

QueueEvents Types

QueueEvents

Event listener class for monitoring queue activity without processing jobs.

class QueueEvents<R = unknown, P = unknown> extends EventEmitter {
  /** Queue name being monitored */
  readonly name: string;

  constructor(name: string, options?: QueueEventsOptions);

  /** Wait until the QueueEvents instance is ready to receive events */
  waitUntilReady(): Promise<void>;

  /** Close the event listener and stop receiving events */
  close(): void;

  /** Disconnect from the event stream (alias for close) */
  disconnect(): Promise<void>;
}

interface QueueEventsOptions {
  embedded?: boolean;
  connection?: ConnectionOptions;
  /** Must match the process-wide embedded manager when one is already active. */
  dataPath?: string;
  prefixKey?: string;
}

Calling new QueueEvents(name) keeps the historical embedded default. Pass { connection } or { embedded: false, connection } to subscribe to a remote broker. The TCP subscription uses a dedicated authenticated connection and automatically re-subscribes after reconnect.

BullMQ Pro aliases

import { QueuePro, WorkerPro, QueueEventsPro } from 'bunqueue/client';
import type { JobPro } from 'bunqueue/client';

QueuePro, WorkerPro, and QueueEventsPro are aliases of the native Queue, Worker, and QueueEvents implementations; JobPro<T> aliases Job<T>. They add no wrapper state, connection, persistence path, or telemetry. The aliases make Pro-oriented migrations explicit while retaining the ordinary class names and behavior.

QueueEvents Event Payloads

Each event emitted by QueueEvents has a typed payload:

/** Emitted when a job is added to the queue */
interface WaitingEvent {
  jobId: string;
}

/** Emitted when a job begins processing */
interface ActiveEvent {
  jobId: string;
}

/** Emitted when a job completes successfully */
interface CompletedEvent<R = unknown> {
  jobId: string;
  returnvalue: R;
}

/** Emitted when a job fails */
interface FailedEvent {
  jobId: string;
  failedReason: string;
  data?: unknown;
  terminal?: boolean; // false while a retry is pending; true once the job failed for good
}

/** Emitted when job progress is updated */
interface ProgressEvent<P = unknown> {
  jobId: string;
  data: P;
}

/** Emitted when a job stalls (no heartbeat) */
interface StalledEvent {
  jobId: string;
}

/** Emitted when a job is removed from the queue */
interface RemovedEvent {
  jobId: string;
  prev: string;
}

/** Emitted when a job is moved to delayed state */
interface DelayedEvent {
  jobId: string;
  delay: number;
}

/** Emitted when a duplicate job is detected */
interface DuplicatedEvent {
  jobId: string;
}

/** Emitted when a job is retried */
interface RetriedEvent {
  jobId: string;
  prev: string;
}

/** Emitted when a job enters waiting-children state */
interface WaitingChildrenEvent {
  jobId: string;
}

/** Emitted when the queue has no more waiting jobs */
interface DrainedEvent {
  id: string;
}

QueueEvents Usage

const events = new QueueEvents('my-queue', {
  connection: { host: '127.0.0.1', port: 6789, token: process.env.BUNQUEUE_TOKEN },
});

events.on('waiting', ({ jobId }) => {
  /* ... */
});
events.on('active', ({ jobId }) => {
  /* ... */
});
events.on('completed', ({ jobId, returnvalue }) => {
  /* ... */
});
events.on('failed', ({ jobId, failedReason }) => {
  /* ... */
});
events.on('progress', ({ jobId, data }) => {
  /* ... */
});
events.on('stalled', ({ jobId }) => {
  /* ... */
});
events.on('removed', ({ jobId, prev }) => {
  /* ... */
});
events.on('delayed', ({ jobId, delay }) => {
  /* ... */
});
events.on('duplicated', ({ jobId }) => {
  /* ... */
});
events.on('retried', ({ jobId, prev }) => {
  /* ... */
});
events.on('waiting-children', ({ jobId }) => {
  /* ... */
});
events.on('drained', ({ id }) => {
  /* ... */
});
events.on('paused', () => {
  /* ... */
});
events.on('resumed', () => {
  /* ... */
});
events.on('error', (error: Error) => {
  /* ... */
});

QueueEventType

type QueueEventType =
  'waiting' | 'active' | 'completed' | 'failed' | 'progress' | 'removed' | 'drained';

FlowProducer Types

FlowProducerOptions

interface FlowProducerOptions {
  /** Use embedded mode (no server) */
  embedded?: boolean;
  /** TCP connection options */
  connection?: ConnectionOptions;
}

Note

FlowProducer extends EventEmitter (BullMQ v5 compatible). You can listen for events using .on(), .once(), etc. The close() method returns Promise<void>. closing is null while live and becomes that one stable Promise when close() or disconnect() first starts shutdown.

Result helpers are transport-aware:

getParentResult<R>(id: string): R | undefined | Promise<R | undefined>;
getParentResults<R>(ids: string[]): Map<string, R> | Promise<Map<string, R>>;

Always await them in portable code. Embedded mode preserves its synchronous return; TCP mode performs broker GetResult calls. Missing is undefined, whereas a persisted null is retained as a real completed result.

FlowOpts

Per-flow options passed as the second argument to flow.add(flowJob, opts).

interface FlowOpts {
  /**
   * Default job options per queue name.
   * Applied as defaults; per-job opts override these.
   * jobId is intentionally excluded because identity belongs to each flow node.
   */
  queuesOptions?: Record<string, Omit<Partial<JobOptions>, 'jobId'>>;
}

Set jobId on a FlowJob.opts object when a node needs a custom identity. jobId is rejected inside queuesOptions: a shared default could assign the same identity to multiple nodes and make the atomic graph ambiguous. Python follows the same rule using job_id inside each node’s opts, never inside queues_options.

Example:

await flow.add(
  {
    name: 'parent',
    queueName: 'reports',
    children: [
      { name: 'fetch', queueName: 'api', data: { url: '...' } },
      { name: 'parse', queueName: 'cpu', data: {} },
    ],
  },
  {
    queuesOptions: {
      api: { attempts: 5, backoff: 2000 }, // All jobs in 'api' queue
      cpu: { timeout: 60000 }, // All jobs in 'cpu' queue
    },
  }
);

FlowJob

A job definition within a flow. Children are processed before the parent.

interface FlowJob<T = unknown> {
  /** Job name */
  name: string;
  /** Queue name */
  queueName: string;
  /** Job data */
  data?: T;
  /** Job options */
  opts?: JobOptions;
  /** Child jobs (processed BEFORE this job) */
  children?: FlowJob<T>[];
}

JobNode

Result from adding a flow. Contains the created job and its children.

interface JobNode<T = unknown> {
  /** The created job instance */
  job: Job<T>;
  /** Child nodes (if any) */
  children?: JobNode<T>[];
}

GetFlowOpts

interface GetFlowOpts {
  /** Job ID to get the flow for */
  id: string;
  /** Queue name where the job is located */
  queueName: string;
  /** Maximum depth to traverse (default: unlimited) */
  depth?: number;
  /** Maximum number of children to fetch per level (default: unlimited) */
  maxChildren?: number;
}

SandboxedWorker Types

SandboxedWorkerOptions

interface SandboxedWorkerOptions {
  /** Path to processor file (must export default async function) */
  processor: string;

  /** Number of worker threads (default: 1; finite, rounded up, below 1 = one thread) */
  concurrency?: number;

  /** Job timeout in ms (default: 30000; 0, a negative value, NaN or Infinity = disabled) */
  timeout?: number;

  /** Max memory per worker in MB (default: 256, uses smol mode if <= 64) */
  maxMemory?: number;

  /** Max restarts before giving up (default: 10) */
  maxRestarts?: number;

  /** Auto-restart crashed workers (default: true) */
  autoRestart?: boolean;

  /** Poll interval in ms when no workers are idle (default: 10; finite, >= 1) */
  pollInterval?: number;

  /** Job heartbeat interval in ms (default: 10000 for TCP, 5000 for embedded; <= 0 disables, otherwise finite, >= 1) */
  heartbeatInterval?: number;

  /** TCP connection options (omit for embedded mode) */
  connection?: ConnectionOptions;

  /** Auto-stop after this many ms of inactivity (default: 0; 0, a negative value, NaN or Infinity = disabled) */
  idleTimeout?: number;

  /** Recycle individual idle worker processes after this many ms (default: 30000; 0, a negative value, NaN or Infinity = disabled) */
  idleRecycleMs?: number;

  /** Auto-restart the worker pool when new jobs arrive after idle shutdown (default: false) */
  autoStart?: boolean;

  /** Wait in ms between checks for new jobs while in idle-shutdown state (default: 5000; finite, >= 1) */
  autoStartPollMs?: number;
}

The constructor rejects an invalid concurrency or duration with a TypeError (not a number) or a RangeError naming the option, e.g. SandboxedWorker: pollInterval must be a finite number of milliseconds >= 1 (got NaN) or SandboxedWorker: concurrency must be a finite number of threads (got Infinity). Durations above 2^31 - 1 ms are honoured exactly.

SandboxedWorker Stats

Returned by sandboxedWorker.getStats().

{
  total: number; // Worker threads in the pool, recycled ones included (busy + idle + recycled)
  busy: number; // Currently processing
  idle: number; // Alive and available for work
  recycled: number; // Threads currently terminated by idle recycling
  restarts: number; // Total restarts across all workers
}

Stall Detection Types

StallConfig

interface StallConfig {
  /** Enable stall detection (default: true) */
  enabled?: boolean;

  /** Stall timeout in ms (default: 30000) */
  stallInterval?: number;

  /** Max stalls before moving to DLQ (default: 3) */
  maxStalls?: number;

  /** Grace period after job start in ms (default: 5000) */
  gracePeriod?: number;
}

DLQ Types

DlqConfig

interface DlqConfig {
  /** Enable auto-retry from DLQ (default: false) */
  autoRetry?: boolean;

  /** Auto-retry interval in ms (default: 3600000 = 1 hour) */
  autoRetryInterval?: number;

  /** Max auto-retries before giving up (default: 3) */
  maxAutoRetries?: number;

  /** Max age before auto-purge in ms (default: 604800000 = 7 days, null = never) */
  maxAge?: number | null;

  /** Max entries per queue (default: 10000) */
  maxEntries?: number;
}

FailureReason

type FailureReason =
  | 'explicit_fail' // Job explicitly failed via fail() or thrown error
  | 'max_attempts_exceeded' // Exceeded all retry attempts
  | 'timeout' // Job processing timed out (exceeded timeout option)
  | 'stalled' // Job stalled (no heartbeat within stallInterval)
  | 'ttl_expired' // Reserved: declared but never produced
  | 'worker_lost' // Reserved: declared but never produced
  | 'unknown'; // Catch-all for edge cases

When is ‘unknown’ used?

The unknown reason is a catch-all for rare edge cases:

• Jobs moved to the DLQ with job.discard() (or the Discard command), which records no error message
• Job data corruption during serialization
• Internal queue manager errors
• Jobs recovered from database without failure metadata
• Race conditions during shutdown

If you see many unknown failures, check logs for underlying errors.

DlqEntry

interface DlqEntry<T = unknown> {
  /** The failed job */
  job: Job<T>;

  /** When job entered DLQ (ms since epoch) */
  enteredAt: number;

  /** Last failure reason */
  reason: FailureReason;

  /** Last error message */
  error: string | null;

  /** Full attempt history */
  attempts: Array<AttemptRecord>;

  /** Number of retry attempts from DLQ */
  retryCount: number;

  /** Last retry timestamp */
  lastRetryAt: number | null;

  /** Next scheduled auto-retry (null = no auto-retry) */
  nextRetryAt: number | null;

  /** When entry expires for auto-purge (null = never) */
  expiresAt: number | null;
}

AttemptRecord

interface AttemptRecord {
  /** Attempt number (1-based) */
  attempt: number;

  /** When this attempt started (ms since epoch) */
  startedAt: number;

  /** When this attempt failed (ms since epoch) */
  failedAt: number;

  /** Failure reason for this attempt */
  reason: FailureReason;

  /** Error message if any */
  error: string | null;

  /** Duration of this attempt in ms */
  duration: number;
}

DlqFilter

interface DlqFilter {
  /** Filter by failure reason */
  reason?: FailureReason;

  /** Only entries older than this timestamp */
  olderThan?: number;

  /** Only entries newer than this timestamp */
  newerThan?: number;

  /** Only entries that can be retried */
  retriable?: boolean;

  /** Only entries that are expired */
  expired?: boolean;

  /** Limit number of results */
  limit?: number;

  /** Offset for pagination */
  offset?: number;
}

DlqStats

interface DlqStats {
  /** Total DLQ entries */
  total: number;

  /** Entries grouped by failure reason */
  byReason: Record<FailureReason, number>;

  /** Entries grouped by queue name */
  byQueue: Record<string, number>;

  /** Entries awaiting auto-retry */
  pendingRetry: number;

  /** Expired entries (awaiting cleanup) */
  expired: number;

  /** Oldest entry timestamp (null if empty) */
  oldestEntry: number | null;

  /** Newest entry timestamp (null if empty) */
  newestEntry: number | null;
}

Generic Type Helpers

bunqueue supports generic types for type-safe job data and results:

// Define typed job data
interface EmailJobData {
  to: string;
  subject: string;
  body: string;
}

interface EmailResult {
  sent: boolean;
  messageId: string;
}

// Queue with typed data
const queue = new Queue<EmailJobData>('emails');

// TypeScript enforces the data shape
await queue.add('welcome', {
  to: 'user@example.com',
  subject: 'Welcome!',
  body: 'Hello and welcome.',
});

// Worker with typed data and result
const worker = new Worker<EmailJobData, EmailResult>('emails', async (job) => {
  // job.data is typed as EmailJobData
  const { to, subject, body } = job.data;
  return { sent: true, messageId: 'msg-123' };
});

// Type error at compile time: missing required fields
await queue.add('send', { to: 'test@example.com' }); // Error!

// QueueEvents with typed result and progress
const events = new QueueEvents<EmailResult, number>('emails');
events.on('completed', ({ jobId, returnvalue }) => {
  // returnvalue is typed as EmailResult
  console.log(returnvalue.messageId);
});

Related

• Queue API - Queue usage with these types
• Worker API - Worker usage with these types
• HTTP API Reference - HTTP endpoints reference