bullmq - v6.0.3
    Preparing search index...

    Class PostgresQueueBackend

    PostgreSQL implementation of IQueueBackend.

    Fulfils the same database-agnostic contract as RedisQueueBackend, but backed by a PostgreSQL database: queue operations are expressed as SQL / PL/pgSQL functions (created by the migrations), job state lives in a single job table keyed by (queue, id) with a state column and partial indexes, claiming uses FOR UPDATE SKIP LOCKED, and the blocking "wait for job" primitive uses LISTEN/NOTIFY.

    The class owns its PostgresConnection; the high-level classes (Queue, Worker, FlowProducer) depend only on IQueueBackend and never touch a pg client directly.

    Hierarchy

    • EventEmitter
      • PostgresQueueBackend

    Implements

    Index
    closing: Promise<void>

    Truthy once IQueueBackend.close has begun (resolves when the close completes). Used by the worker to decide whether it is still safe to issue datastore operations (e.g. completing the current job) while the higher-level instance is shutting down.

    connection: PostgresConnection
    • Atomically inserts a flow (tree) of jobs that may span multiple queues, returning one [error, idOrCode] tuple per entry, in the same order they were provided. Each entry is self-describing (it carries its own queue prefix/queueName), so the operation is not bound to a single queue.

      For the Redis adapter this is a single MULTI; a SQL backend would use a single transaction.

      Parameters

      • entries: {
            jobData: JobJson;
            jobId: string;
            parentKeyOpts: ParentKeyOpts;
            prefix: string;
            queueName: string;
        }[]

      Returns Promise<[Error, string | number][]>

    • Adds a single job to the queue, routing it to the correct initial state (wait / delayed / prioritized / waiting-children) based on its options.

      The backend uses its own connection — callers never pass one in.

      Parameters

      Returns Promise<string>

    • Adds many jobs to the queue in a single, efficient operation.

      How the insert is batched (a Redis pipeline, a single multi-row SQL INSERT, a transaction, …) is entirely an implementation detail of the backend; the contract only requires that all jobs are added and their ids returned in order.

      Parameters

      Returns Promise<string[]>

      The generated ids, in the same order as entries.

    • Registers a job scheduler and enqueues its next delayed iteration.

      Two job-option bags are involved, with deliberately different roles:

      • templateOpts — the scheduler's template options, stored once and reused as the basis for every future iteration produced by the scheduler.
      • delayedJobOpts — the fully-resolved options for the single delayed job created right now: the template plus this iteration's jobId, delay, repeat.offset/count, etc.

      Parameters

      Returns Promise<[string, number]>

      A tuple of [jobId, delay] for the next iteration.

    • Appends a row to a job's log, optionally trimming old entries.

      Parameters

      • jobId: string
      • logRow: string
      • OptionalkeepLogs: number

      Returns Promise<number>

      The total number of log entries.

    • Removes jobs in a given state that are older than timestamp.

      Parameters

      • state: string
      • timestamp: number
      • limit: number = 0

      Returns Promise<string[]>

      The ids of the removed jobs.

    • Closes the backend and its underlying connection(s), waiting for any in-flight work to settle.

      Parameters

      • force: boolean = false

        When true, forcibly tears down the connection(s) without waiting for in-flight (e.g. blocking) commands to finish.

      Returns Promise<void>

    • Interrupts the backend's in-flight blocking wait (so a worker can stop or recover). No-op for backends without a dedicated blocking connection.

      Parameters

      • _wait: boolean = true

      Returns Promise<void>

    • Extends the lock of several active jobs at once.

      Parameters

      • jobIds: string[]
      • tokens: string[]
      • duration: number

      Returns Promise<string[]>

      The ids of the jobs whose lock could not be extended.

    • Returns a sibling backend bound to a different queue (by name) that shares this backend's underlying connection(s).

      This is used by FlowProducer, which spans multiple queues over a single connection: every node in a flow needs datastore operations scoped to its own queue, but they must all reuse the same connection. The returned backend has an independent identity (its operations target the given queue) but does not own the connection, so closing it is a no-op on the shared connection.

      Parameters

      • queueName: string

        The queue the sibling backend should operate on.

      • Optional_prefix: string

        Optional key prefix for the target queue. Flows may span queues under different prefixes, so when omitted the backend's own prefix is used.

      Returns IQueueBackend

    • Returns the raw worker/client list(s) for the queue's datastore. For the Redis adapter this is CLIENT LIST (one string per cluster node, or a single string otherwise). Backends with no notion of connected clients may return an empty array.

      Returns Promise<string[]>

    • Returns a job's children dependencies (processed/unprocessed/ignored/failed).

      Parameters

      Returns Promise<
          {
              failed?: string[];
              ignored?: Record<string, any>;
              nextFailedCursor?: number;
              nextIgnoredCursor?: number;
              nextProcessedCursor?: number;
              nextUnprocessedCursor?: number;
              processed?: Record<string, any>;
              unprocessed?: string[];
          },
      >

    • Returns a page of a job's logs together with the total log count.

      Parameters

      • jobId: string
      • start: number
      • end: number
      • asc: boolean

      Returns Promise<{ count: number; logs: string[] }>

    • Returns a range of scheduler keys with their next-run scores, flattened as [key, score, key, score, …].

      Parameters

      • start: number
      • end: number
      • asc: boolean

      Returns Promise<string[]>

    • Parameters

      • type: "failed" | "completed"
      • start: number = 0
      • end: number = -1

      Returns Promise<[string[], string[], number]>

    • Returns whether a job has finished and (optionally) its result.

      Parameters

      • jobId: string
      • OptionalreturnValue: boolean

      Returns Promise<number | [number, string]>

    • Atomically moves the next eligible job from wait/prioritized to active, returning its data (or the delay/rate-limit signals when none is ready).

      Parameters

      • token: string
      • Optionalname: string

      Returns Promise<any[]>

    • Moves an active job to the completed state and, optionally, fetches the next job to process.

      Type Parameters

      • T = any
      • R = any
      • N extends string = string

      Parameters

      • job: MinimalJob<T, R, N>
      • returnValue: R
      • removeOnComplete: number | boolean | KeepJobs
      • token: string
      • fetchNext: boolean

      Returns Promise<{ finishedOn: number; result: void | any[] }>

      The next job data tuple when fetchNext is set, plus the finishedOn timestamp that was recorded.

    • Moves an active job to the failed state and, optionally, fetches the next job to process.

      Type Parameters

      • T = any
      • R = any
      • N extends string = string

      Parameters

      • job: MinimalJob<T, R, N>
      • failedReason: string
      • removeOnFail: number | boolean | KeepJobs
      • token: string
      • fetchNext: boolean
      • OptionalfieldsToUpdate: Record<string, any>

      Returns Promise<{ finishedOn: number; result: void | any[] }>

      The next job data tuple when fetchNext is set, plus the finishedOn timestamp that was recorded.

    • Irreversibly destroys the queue and all of its contents.

      Parameters

      • opts: { count: number; force: boolean }

      Returns Promise<number>

      A cursor; 0 when obliteration is complete.

    • Paginates a datastore set or hash, optionally fetching the jobs themselves.

      Parameters

      • key: string
      • opts: { end: number; fetchJobs?: boolean; start: number }

      Returns Promise<
          {
              cursor: string;
              items: { err?: string; id: string; v?: any }[];
              jobs?: JobJson[];
              total: number;
          },
      >

    • Parses a PostgreSQL flow child key ("<queue>:<id>") into its components. There is no keyspace prefix, so prefix is always empty. Inverse of toKey.

      Parameters

      • key: string

      Returns { id: string; prefix: string; queueName: string }

    • Publishes a custom event to the queue's event stream.

      Parameters

      • fields: Record<string, string | number>
      • _maxEvents: number

      Returns Promise<string>

      The id of the appended event entry.

    • Blocks (up to blockTimeout ms) reading the queue's event stream for entries newer than id, returning the raw stream entries (or a falsy value on timeout). For the Redis adapter this is an XREAD ... BLOCK.

      Parameters

      • id: string
      • blockTimeout: number

      Returns Promise<[string, EntryRaw[]][]>

    • Removes a job and (optionally) its children.

      Parameters

      • jobId: string
      • removeChildren: boolean

      Returns Promise<number>

      1 if removed, 0 if it (or a dependency) was locked.

    • Removes orphaned job hashes (job data present but not referenced by any state set). This is a Redis keyspace-maintenance concern: on PostgreSQL a job is a single relational row inserted transactionally with its state, so orphans cannot exist and there is nothing to remove. Always returns 0.

      Parameters

      • Optional_count: number
      • Optional_limit: number

      Returns Promise<number>

    • Sets a human-readable name on the underlying connection (for observability). No-op for backends that have no such concept.

      Parameters

      • name: string

      Returns Promise<void>

    • Builds a namespaced identifier of the given type ("<queue>:<type>"), used e.g. for flow dependency identifiers. No prefix is involved.

      Parameters

      • type: string

      Returns string

    • Blocks (up to blockTimeout seconds) until a job for this queue may be available, via LISTEN/NOTIFY. Producers notify the shared bullmq_jobs channel with the queue name as payload (in add_job), so a producer in any process wakes a blocked worker immediately. Returns a marker (score 0 = "check now") or null on timeout. The Redis backend implements this with BZPOPMIN.

      Parameters

      • blockTimeout: number

      Returns Promise<{ member: string; score: number }>