bullmq - v6.0.3
    Preparing search index...

    Class RedisQueueBackend

    IQueueBackend

    Database-agnostic contract describing every high-level operation that the Queue, Worker and Job classes need in order to function. The goal of this interface is to express the queue semantics ("move job to active", "extend lock", "promote job", …) independently of the underlying datastore.

    Built-in implementations currently include the Redis adapter (RedisQueueBackend) and the PostgreSQL adapter. Both fulfil the same operations over different datastores without requiring any change to Queue, Worker or Job.

    The method names and signatures intentionally mirror the existing RedisQueueBackend class so that the Redis adapter is a near drop-in implementation. Operations that used to be performed via direct datastore commands scattered across the three classes (queue metadata, job getters, the blocking "wait for next job" primitive, …) have been promoted into this interface so that the three classes never need to talk to the datastore directly.

    Low-level, Redis-specific helpers (Lua KEYS/ARGV builders, error-code mapping, runCommand, …) are deliberately not part of this contract. They remain private implementation details of the Redis adapter.

    The interface intentionally exposes no connection or transaction type: a concrete adapter owns its connection(s). For example, the Redis adapter is built from a context that provides an IRedisClient (plus a dedicated blocking client for IQueueBackend.waitForJob), so callers never thread a connection or transaction through an operation.

    Hierarchy

    • EventEmitter
      • RedisQueueBackend

    Implements

    Index
    blockingConnection?: RedisConnection
    closing: Promise<void>

    Resolves once a close has been initiated. Owned by the backend (it owns the underlying connection(s)).

    connection: RedisConnection
    moveToFinishedKeys: string[]
    • Atomically inserts a whole flow (tree) of jobs that may span multiple queues, returning one [error, idOrCode] tuple per entry in the same order they were provided. For the Redis adapter this is a single MULTI transaction; another backend would use a single SQL transaction.

      Each entry is self-describing (it carries its own queue prefix and queueName), so the operation does not need to be bound to a single queue's key map.

      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.

    • 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.

    • Change delay of a delayed job.

      Reschedules a delayed job by setting a new delay from the current time. For example, calling changeDelay(5000) will reschedule the job to execute 5000 milliseconds (5 seconds) from now, regardless of the original delay.

      Parameters

      • jobId: string

        the ID of the job to change the delay for.

      • delay: number

        milliseconds from now when the job should be processed.

      Returns Promise<void>

      delay in milliseconds.

      JobNotExist This exception is thrown if jobId is missing.

      JobNotInState This exception is thrown if job is not in delayed state.

    • Changes the priority (and optionally lifo) of a waiting job.

      Parameters

      • jobId: string
      • priority: number = 0
      • lifo: boolean = false

      Returns Promise<void>

    • Remove jobs in a specific state.

      Parameters

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

      Returns Promise<string[]>

      Id jobs from the deleted records.

    • Clears a job's logs, optionally keeping the most recent keepLogs rows.

      Parameters

      • jobId: string
      • OptionalkeepLogs: number

      Returns Promise<void>

    • Builds the Redis client name ("<prefix>:<base64(queue)><suffix>"), used for CLIENT SETNAME and worker/queue discovery via CLIENT LIST.

      Parameters

      • suffix: string = ''

      Returns string

    • Closes the backend and its underlying connection(s).

      The dedicated blocking connection (if any) is closed first so that an in-flight blocking command (e.g. bzpopmin) is interrupted before the main connection is closed.

      Parameters

      • force: boolean = false

      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.

    • Parameters

      • __namedParameters: {
            code: number;
            command: string;
            jobId?: string;
            parentKey?: string;
            state?: string;
        }

      Returns Error

    • Returns a sibling backend bound to a different queue that shares this backend's connection(s). Used by FlowProducer to operate on the many queues that a flow may span over a single connection. The sibling does not own the connection, so its close/disconnect are no-ops.

      Parameters

      • queueName: string
      • Optionalprefix: string

      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[] }>

    • Fetches job ids and their job hashes for the provided states in a single script, skipping ids whose job hash is missing (for example the deprecated wait list marker or jobs removed after their id was read). Each returned entry is a [jobId, jobHashFields] tuple grouped per requested type.

      Parameters

      • types: JobType[]
      • start: number = 0
      • end: number = -1
      • asc: boolean = false

      Returns Promise<[string, 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
      • returnValue: boolean = false

      Returns Promise<number | [number, string]>

    • Checks whether a job with the given id is present in the provided queue state.

      Parameters

      • state: string
      • jobId: string

      Returns Promise<boolean>

    • Moves a job back from Active to Wait. This script is used when a job has been manually rate limited and needs to be moved back to wait from active status.

      Parameters

      • jobId: string

        Job id

      • token: string = '0'

      Returns Promise<any>

    • Looks for unlocked jobs in the active queue.

      The job was being worked on, but the worker process died and it failed to renew the lock. We call these jobs 'stalled'. This is the most common case. We resolve these by moving them back to wait to be re-processed. To prevent jobs from cycling endlessly between active and wait, (e.g. if the job handler keeps crashing), we limit the number stalled job recoveries to settings.maxStalledCount.

      Returns Promise<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.

    • 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 = false

      Returns (string | number | boolean | Buffer<ArrayBufferLike>)[]

    • Parameters

      • jobId: string
      • timestamp: number
      • token: string
      • delay: number
      • opts: MoveToDelayedOpts = {}

      Returns (string | number | Buffer<ArrayBufferLike>)[]

    • 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.

    • Type Parameters

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

      Parameters

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

      Returns (string | number | boolean | Buffer<ArrayBufferLike>)[]

    • Parameters

      • jobId: string
      • args: (string | number | boolean | Buffer<ArrayBufferLike>)[]

      Returns Promise<any[]>

    • Move parent job to waiting-children state.

      Parameters

      Returns Promise<boolean>

      true if job is successfully moved, false if there are pending dependencies.

      JobNotExist This exception is thrown if jobId is missing.

      JobLockNotExist This exception is thrown if job lock is missing.

      JobNotInState This exception is thrown if job is not in active state.

    • 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.

    • Paginate a set or hash keys.

      Parameters

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

        options to define the pagination behaviour

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

    • Parses a Redis flow child key ("<prefix>:<queue>:<id>") into its components. 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 the child→parent dependency for a not-yet-finished child.

      Parameters

      • jobId: string
      • parentKey: string

      Returns Promise<boolean>

      true if the dependency existed and was removed.

    • Removes orphaned job keys that exist in the datastore but are not referenced by any queue state set.

      Parameters

      • count: number = 1000
      • limit: number = 0

      Returns Promise<number>

      The total number of orphaned jobs removed.

    • Attempts to reprocess a job

      Type Parameters

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

      Parameters

      • job: MinimalJob<T, R, N>

        The job to reprocess

      • state: "failed" | "completed"

        The expected job state. If the job is not found on the provided state, then it's not reprocessed. Supported states: 'failed', 'completed'

      • opts: RetryOptions = {}

      Returns Promise<void>

      A promise that resolves when the job has been successfully moved to the wait queue.

      Will throw an error with a code property indicating the failure reason:

      • code 0: Job does not exist
      • code -1: Job is currently locked and can't be retried
      • code -2: Job was not found in the expected set
    • Sets a human-readable name on the underlying connection (CLIENT SETNAME). Unsupported-command and shutdown errors are swallowed.

      Parameters

      • name: string

      Returns Promise<void>

    • Blocks (up to blockTimeout seconds) until the queue signals that a new job may be available, returning the next "block-until" timestamp.

      For the Redis adapter this is a BZPOPMIN on the marker sorted set using the adapter's own dedicated blocking connection; other adapters may implement it via LISTEN/NOTIFY, change-data-capture or polling.

      Parameters

      • blockTimeout: number

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

      The marker member/score on success, or null on timeout.