Class RedisQueue

Redis queue adapter implementation.

Uses Redis sorted sets (ZSET) for ready/processing/dead sets and hashes (HASH) for job payloads. Implements the claim-based reserve pattern to prevent double-dispatch.

Key structure per job name:

queue:<name>:ready - ZSET (score = availableAtMs, member = jobId)
queue:<name>:processing - ZSET (score = lease deadline on the Redis clock)
queue:<name>:claims / stalls - HASH (delivery tokens / abandoned attempts)
queue:<name>:claimseq - persistent INCR sequence shared across replicas
queue:<name>:dead - ZSET (score = deadLetterAtMs, member = jobId)
queue:<name>:jobs - HASH (field = jobId, value = JSON job)
queue:<name>:dead:jobs - HASH (dead payloads; only when a TTL is configured)

Recurring jobs:

queue:recurring:due - ZSET (score = nextRunAtMs, member = recurringId)
queue:recurring:jobs - HASH (field = recurringId, value = JSON recurring)

Since

0.1.0

Implements

QueueAdapter

Index

Constructors

constructor

new RedisQueue(options?: RedisQueueOptions): RedisQueue
Parameters
- Optionaloptions: RedisQueueOptions
  Connection, client and retention options
Returns RedisQueue

Throws
For an invalid key prefix, a non-positive lease, or a duration outside the finite timer range (0 disables only commandTimeoutMs)
- Defined in packages/queue-plugin/src/adapters/redis-queue.ts:276

Properties

`Optional`claimRenewal

claimRenewal?: {
intervalMs: number;
extend(name: string, id: string, claimToken: string): Promise<boolean>;
}

Renewed lease, present only while connected to an eval-capable client.

`Optional`depths

depths?: (name: string) => Promise<QueueDepths>

M70k (X8-4): counts this name's three states with one ZCARD each.

Absent on a client that does not expose zcard, which is why the member is an assigned property rather than a declared method — the health indicator must be able to tell "cannot report" from "nothing there". Absent again after disconnect, for the same reason isHealthy is: the closure captures the client it was installed for, so leaving it in place would answer a depth read with that dead client's rejection instead of the absence that says the adapter cannot count right now.

Type Declaration

- (name: string): Promise<QueueDepths>
- Parameters
  - name: string
    Job name
  Returns Promise<QueueDepths>
  The depth of each state

Since

0.3.0

`Optional`isHealthy

isHealthy?: () => Promise<boolean>

M70c: present only when the client exposes ping(); its absence is unknown reachability, not false (a minimal injected fake has not told us the server is dead).

Since

0.1.0

Methods

ack

ack(name: string, id: string, claimToken?: string): Promise<void>
Acknowledges a job as successfully processed.
Parameters
- name: string
  Job name
- id: string
  Job ID (stable public identity)
- OptionalclaimToken: string
  Opaque claim token returned from reserve (prevents stale settlement)
Returns Promise<void>
Resolves when acknowledged

Throws
If the adapter is not connected

Since
0.1.0
Implementation of QueueAdapter.ack
- Defined in packages/queue-plugin/src/adapters/redis-queue.ts:444

advanceRecurring

advanceRecurring(id: string, nextRunAtMs: number): Promise<void>
Advances a recurring job's next run time.
Parameters
- id: string
  Recurring job ID
- nextRunAtMs: number
  Next run timestamp
Returns Promise<void>
Resolves when advanced

Throws
If the adapter is not connected

Since
0.1.0
Implementation of QueueAdapter.advanceRecurring
- Defined in packages/queue-plugin/src/adapters/redis-queue.ts:783

connect

connect(): Promise<void>
Connects the adapter to its backend.

Returns Promise<void>
Resolves when connected

Since
0.1.0
Implementation of QueueAdapter.connect
- Defined in packages/queue-plugin/src/adapters/redis-queue.ts:290

deadLetter

deadLetter(
    name: string,
    id: string,
    nowMs: number,
    claimToken?: string,
): Promise<void>
Moves a job to the dead letter queue.
Parameters
- name: string
  Job name
- id: string
  Job ID (stable public identity)
- nowMs: number
  Current timestamp in ms (for dead-letter timestamp)
- OptionalclaimToken: string
  Opaque claim token returned from reserve (prevents stale settlement)
Returns Promise<void>
Resolves when dead-lettered

Throws
If the adapter is not connected

Since
0.1.0
Implementation of QueueAdapter.deadLetter
- Defined in packages/queue-plugin/src/adapters/redis-queue.ts:517

disconnect

disconnect(): Promise<void>
Disconnects the adapter.

Returns Promise<void>
Resolves when disconnected

Since
0.1.0
Implementation of QueueAdapter.disconnect
- Defined in packages/queue-plugin/src/adapters/redis-queue.ts:351

enqueue

enqueue<T>(job: StoredJob<T>): Promise<void>
Enqueues a job.
Type Parameters
- T
Parameters
- job: StoredJob<T>
  The job to enqueue
Returns Promise<void>
Resolves when enqueued

Throws
If the adapter is not connected

Since
0.1.0
Implementation of QueueAdapter.enqueue
- Defined in packages/queue-plugin/src/adapters/redis-queue.ts:366

fetchRecurringDue

fetchRecurringDue(nowMs: number): Promise<readonly StoredRecurring[]>
Fetches recurring jobs that are due.
Parameters
- nowMs: number
  Current timestamp in ms
Returns Promise<readonly StoredRecurring[]>
The due recurring jobs

Throws
If the adapter is not connected

Since
0.1.0
Implementation of QueueAdapter.fetchRecurringDue
- Defined in packages/queue-plugin/src/adapters/redis-queue.ts:761

isReady

isReady(): boolean
Checks if the adapter is ready/connected.

Returns boolean
true if connected

Since
0.1.0
Implementation of QueueAdapter.isReady
- Defined in packages/queue-plugin/src/adapters/redis-queue.ts:362

requeue

requeue<T>(
    name: string,
    id: string,
    availableAtMs: number,
    attempts: number,
    claimToken?: string,
): Promise<void>
Requeues a job with a new available timestamp.
Type Parameters
- T
Parameters
- name: string
  Job name
- id: string
  Job ID (stable public identity)
- availableAtMs: number
  When the job becomes available again
- attempts: number
  Updated attempt count
- OptionalclaimToken: string
  Opaque claim token returned from reserve (prevents stale settlement)
Returns Promise<void>
Resolves when requeued

Throws
If the adapter is not connected

Since
0.1.0
Implementation of QueueAdapter.requeue
- Defined in packages/queue-plugin/src/adapters/redis-queue.ts:468

reserve

reserve<T>(
    name: string,
    limit: number,
    nowMs: number,
): Promise<readonly StoredJob<T>[]>
Reserves up to limit due jobs. Redis scripts use server time; adapters without a server clock compare availableAtMs to nowMs.

CLAIMS jobs: moves them from ready set to processing set. A reserved job is not returned by subsequent reserve calls.
Type Parameters
- T
Parameters
- name: string
  Job name
- limit: number
  Maximum jobs to reserve
- nowMs: number
  Caller timestamp in ms, used when there is no server clock
Returns Promise<readonly StoredJob<T>[]>
The reserved jobs

Throws
If the adapter is not connected

Since
0.1.0
Implementation of QueueAdapter.reserve
- Defined in packages/queue-plugin/src/adapters/redis-queue.ts:388

storeRecurring

storeRecurring(rec: StoredRecurring): Promise<void>
Stores a recurring job.
Parameters
- rec: StoredRecurring
  The recurring job to store
Returns Promise<void>
Resolves when stored

Throws
If the adapter is not connected

Since
0.1.0
Implementation of QueueAdapter.storeRecurring
- Defined in packages/queue-plugin/src/adapters/redis-queue.ts:746

Class RedisQueue

Since

Implements

Index

Constructors

Properties

Methods

Constructors

constructor

Parameters

Returns RedisQueue

Throws

Properties

OptionalclaimRenewal

Optionaldepths

Type Declaration

Parameters

Returns Promise<QueueDepths>

Since

OptionalisHealthy

Since

Methods

ack

Parameters

Returns Promise<void>

Throws

Since

advanceRecurring

Parameters

Returns Promise<void>

Throws

Since

connect

Returns Promise<void>

Since

deadLetter

Parameters

Returns Promise<void>

Throws

Since

disconnect

Returns Promise<void>

Since

enqueue

Type Parameters

Parameters

Returns Promise<void>

Throws

Since

fetchRecurringDue

Parameters

Returns Promise<readonly StoredRecurring[]>

Throws

Since

isReady

Returns boolean

Since

requeue

Type Parameters

Parameters

Returns Promise<void>

Throws

Since

reserve

Type Parameters

Parameters

Returns Promise<readonly StoredJob<T>[]>

Throws

Since

storeRecurring

Parameters

Returns Promise<void>

Throws

Since

Settings

On This Page

`Optional`claimRenewal

`Optional`depths

`Optional`isHealthy