Advanced Features

Shared Client (Connection Reuse)
Job Schedulers (Repeatable / Cron Jobs)
LIFO Mode
Job TTL
Pluggable Serializers
Ordering and Group Concurrency
Custom Job IDs
Deduplication
Token Bucket Rate Limiting
Global Concurrency
Global Rate Limiting
Job Revocation (Cooperative Cancellation)
Transparent Compression
Retries and Backoff
Dead Letter Queues
Fallback Chains
Dual-axis Rate Limiting (RPM + TPM)
Per-job Lock Duration
Vector Search Index Management
Request Timeout

Shared Client

By default, each glide-mq component creates its own GLIDE client (one TCP connection). You can optionally inject a shared client to reduce connection count.

Default behavior (dedicated connections)

typescript

const connection = { addresses: [{ host: 'localhost', port: 6379 }] };

const queue = new Queue('jobs', { connection }); // 1 connection
const flow = new FlowProducer({ connection }); // 1 connection
const worker = new Worker('jobs', handler, { connection }); // 2 connections (command + blocking)
const events = new QueueEvents('jobs', { connection }); // 1 connection
// Total: 5 TCP connections

Shared client (opt-in)

typescript

import { GlideClient } from '@glidemq/speedkey';

const client = await GlideClient.createClient({ addresses: [{ host: 'localhost' }] });
const connection = { addresses: [{ host: 'localhost' }] };

const queue = new Queue('jobs', { client });
const flow = new FlowProducer({ client });
const worker = new Worker('jobs', handler, { connection, commandClient: client });
const events = new QueueEvents('jobs', { connection });
// Total: 2 TCP connections (shared + Worker's blocking client)

Queue, FlowProducer, and Worker's command client all perform non-blocking operations (FCALL, HGET, ZADD, etc.) and can safely share a single GLIDE client. GLIDE's Rust core multiplexes commands over one TCP connection with up to 1000 concurrent in-flight requests.

Worker's blocking client (XREADGROUP BLOCK) and QueueEvents (XREAD BLOCK) tie up the connection's read loop. These always get their own dedicated connection - you cannot inject a shared client into them.

QueueEvents will throw if you try to pass a client:

typescript

// Throws: "QueueEvents does not accept an injected `client`"
new QueueEvents('jobs', { connection, client } as any);

Tradeoffs

	Dedicated (default)	Shared
Connections	N+2 per setup (1 per Queue/FlowProducer + 2 per Worker + 1 per QueueEvents)	2 (shared + blocking)
Throughput	Baseline	Same or slightly better (fewer NAPI wake callbacks)
Latency	Baseline	Same (p50/p95/p99 identical in benchmarks)
Isolation	Each component has its own connection - failures are independent	All components sharing a client are affected by a disconnect
Reconnection	Each component reconnects independently	Worker emits error if shared client is unreachable - you manage reconnection
Lifecycle	Component creates and closes its own client	You create the client, you close it. `close()` on a component does not destroy the shared client.
Simplicity	Pass `connection` - done	Must create client upfront, pass it around, close in correct order
Memory	Slightly higher (N client objects + Rust state machines)	Lower (1 client object shared)

When to use shared

Many Queue instances pointing to different queue names (e.g., multi-tenant routing)
Queue + FlowProducer on the same process - saves 1 connection
Connection count is a concern (cloud Valkey with connection limits)

When to stick with dedicated

Simple setup with one Queue and one Worker - the default is fine
You want full isolation between components
You don't want to manage client lifecycle manually

Constraints

Worker always requires connection even when commandClient is provided, because the blocking client must be auto-created.
Don't close the shared client while components are alive. Close components first, then the client.
Don't mutate shared client state externally (e.g., SELECT to change database).
commandClient and client are aliases on Worker - provide one or the other, not both.

Close order

typescript

// Correct: close components first, then shared client
await queue.close(); // detaches from shared client (does not close it)
await worker.close(); // closes only the auto-created blocking client
await flow.close(); // detaches from shared client
client.close(); // now safe - no components using it

Producer with an external client

Producer also supports external client injection. When opts.client is provided the Producer borrows the connection without taking ownership - close() will not destroy it. This is the recommended pattern for serverless environments where the connection lifecycle must align with the request lifecycle:

typescript

import { GlideClient } from '@glidemq/speedkey';
import { Producer } from 'glide-mq';

export async function handler(event) {
  const client = await GlideClient.createClient({ addresses: [{ host: process.env.VALKEY_HOST }] });
  const producer = new Producer('jobs', { client });

  for (const job of event.jobs) {
    await producer.add(job.name, job.data);
  }

  // producer.close() does NOT close the client when client was injected
  await producer.close();
  client.close(); // caller owns lifecycle
}

For connection reuse across warm invocations, use ServerlessPool instead - see docs/SERVERLESS.md.

`inflightRequestsLimit`

GLIDE defaults to 1000 concurrent in-flight requests per client. For high-concurrency setups, you can tune this:

typescript

const connection = {
  addresses: [{ host: 'localhost' }],
  inflightRequestsLimit: 2000,
};

At Worker concurrency=50, peak inflight is ~55 commands. The 1000 default supports up to ~950 concurrent job activations across all components sharing one client.

Job Schedulers

Use upsertJobScheduler to define repeatable jobs driven by a cron expression or a fixed interval. Schedulers survive worker restarts - the next run time is stored in Valkey.

typescript

const queue = new Queue('tasks', { connection });

// Cron: run "daily-report" every day at 08:00 UTC
await queue.upsertJobScheduler(
  'daily-report',
  { pattern: '0 8 * * *' },
  { name: 'generate-report', data: { type: 'daily' } },
);

// Bound a scheduler to a campaign window and stop after 36 runs
await queue.upsertJobScheduler(
  'black-friday-deals',
  {
    pattern: '0 */2 * * *',
    startDate: new Date('2026-11-28T00:00:00Z'),
    endDate: new Date('2026-12-01T00:00:00Z'),
    limit: 36,
  },
  { name: 'promote-deal', data: { campaign: 'black-friday' } },
);

// Interval: run "cleanup" every 5 minutes
await queue.upsertJobScheduler(
  'cleanup',
  { every: 5 * 60 * 1_000 }, // ms
  { name: 'cleanup-old-records', data: {} },
);

// List all registered schedulers
const schedulers = await queue.getRepeatableJobs();

// Remove a scheduler (does not cancel jobs already in flight)
await queue.removeJobScheduler('cleanup');

Cron syntax

Patterns use the standard 5 fields, minute hour day-of-month month day-of-week, or 6 fields with a leading second. Each field accepts *, numbers, ranges (1-5), steps (*/15, 10-40/10, 5/15 = from 5 to the end of the field) and lists (1,15). Patterns run in UTC unless tz is set. The syntax is a superset of cron-parser 4.9, the parser BullMQ uses; a test compares the two over 44 patterns in 4 timezones.

Names: JAN-DEC in the month field, SUN-SAT in the day-of-week field, case-insensitive, in values, ranges, steps and lists: 0 9 * * MON-FRI, 0 0 1 JAN-MAR/2 *.
Day-of-week: 0-7, where both 0 and 7 are Sunday (5-7 is Friday to Sunday).
?: in either day field, the same as *.
L: in day-of-month, L is the last day of the month and LW the last weekday (Mon-Fri). In day-of-week, 5L or FRIL is the last Friday of the month.
W: in day-of-month, 15W is the weekday nearest the 15th without leaving the month (Saturday moves to Friday, Sunday to Monday; 1W on a Saturday runs Monday the 3rd; 31W skips 30-day months). cron-parser does not accept W; Quartz does.
#: in day-of-week, 2#1 or TUE#1 is the first Tuesday of the month, n from 1 to 5. Modifiers can be listed (1#2,5L), which cron-parser rejects.
Seconds: */10 * * * * * matches every 10 seconds and 30 0 9 * * * matches 09:00:30. nextCronOccurrence honors the field to the second, but the scheduler creates jobs on its promotion tick (promotionInterval, default 5000 ms), at most one per scheduler per tick: a job runs within one tick of its cron time, and a period shorter than the tick (* * * * * *) produces one job per tick, not one per second. Lower promotionInterval for finer granularity.
Errors: a malformed field throws at upsertJobScheduler (Invalid cron token: 5foo, Cron value out of bounds: 60, Cron range reversed: 10-5). A pattern that has no occurrence in the next 10 years (0 0 30 2 *) throws No cron match found within 10 years.
Day-of-month and day-of-week: when both fields are restricted, a day matches if either field matches. 0 0 1 * 1 fires on every 1st of the month and on every Monday. When one of them is unrestricted, only the other one decides. A field is unrestricted when it is *, ? or */1 (day-of-month also when it covers 1-31). An explicit 0-6 day-of-week counts as restricted. This matches cron-parser (BullMQ). Vixie cron differs only for stepped wildcards such as */2, which it treats as unrestricted. Releases up to 0.15.5 required both fields to match.
Daylight saving time (with tz): follows vixie cron. A pattern whose minute or hour field contains * (*/15 * * * *, 0 * * * *) is a wildcard pattern and runs on elapsed time: when clocks fall back it fires in both instances of the repeated hour. Any other pattern is fixed-time (30 1 * * *, 0,30 1-3 * * *) and fires once: at the earlier instant when clocks fall back, and at the first instant after the gap when clocks spring forward (30 2 * * * in America/New_York runs at 03:00 EDT on the transition day; several skipped times coalesce into that one run). Wildcard patterns skip the missing times. Releases up to 0.15.5 lost part of the repeated hour and skipped fixed times inside the gap until the next day.

Repeat-after-complete mode

repeatAfterComplete schedules the next job only after the current one completes (or terminally fails). Unlike every, which fires at fixed intervals regardless of processing time, repeatAfterComplete ensures no overlap between successive runs.

typescript

// Poll a sensor every 5 seconds after the previous poll finishes
await queue.upsertJobScheduler(
  'sensor-poll',
  {
    repeatAfterComplete: 5000, // 5s after previous job completes
  },
  { name: 'poll', data: { sensor: 'temp-1' } },
);

This mode is useful for:

Polling - avoid stacking requests when the upstream is slow.
Sequential pipelines - each step must finish before the next begins.
Adaptive intervals - combine with a custom processor that adjusts repeatAfterComplete via upsertJobScheduler based on results.

Upserting a repeatAfterComplete scheduler while its job is running (for example from inside the processor) keeps waiting for that job: the next run is scheduled when it completes, using the new interval, and iterationCount is kept unless tz, startDate or endDate changed. It never starts a second, overlapping chain. To force an immediate run, remove the scheduler and upsert it again.

Switching an existing every or pattern scheduler to repeatAfterComplete does not fire at once: the first run stays at the old mode's nextRun (or a later startDate). Every scheduler entry records the job its last tick fired (inflightJobId). If that job is still running when the held nextRun passes, the tick parks the entry (nextRun 0) and the first repeatAfterComplete run is scheduled repeatAfterComplete ms after that job completes or fails terminally, so the two never overlap. Only the recorded job advances a parked entry; a job from an earlier chain that finishes late is ignored.

repeatAfterComplete is mutually exclusive with pattern and every. Bounded options (startDate, endDate, limit) work normally with this mode.

Bounded schedulers

All three scheduler modes (pattern, every, repeatAfterComplete) support bounding via startDate, endDate, and limit:

Option	Type	Effect
`startDate`	`Date \| number`	Defer the first eligible run until this time.
`endDate`	`Date \| number`	Auto-remove the scheduler when the next scheduled time would exceed this date.
`limit`	`number`	Auto-remove the scheduler after creating this many jobs.

typescript

// Run a cron job during a specific campaign window, max 36 runs
await queue.upsertJobScheduler(
  'black-friday-deals',
  {
    pattern: '0 */2 * * *',
    startDate: new Date('2026-11-28T00:00:00Z'),
    endDate: new Date('2026-12-01T00:00:00Z'),
    limit: 36,
  },
  { name: 'promote-deal', data: { campaign: 'black-friday' } },
);

// Interval with a delayed start and a hard stop after 100 iterations
await queue.upsertJobScheduler(
  'warmup-cache',
  {
    every: 30_000,
    startDate: Date.now() + 60_000, // first run delayed 1 minute
    endDate: new Date('2026-12-31'), // stop scheduling after this date
    limit: 100, // auto-remove after 100 runs
  },
  { name: 'warmup', data: { region: 'us-east' } },
);

getJobScheduler() / getRepeatableJobs() expose the stored bounds together with iterationCount so you can inspect how many runs have already fired.

The internal Scheduler class fires a promotion loop that converts due scheduler entries into real jobs, then re-registers the next occurrence.

The template opts accept the same job options as Queue.add except delay, deduplication, parent and jobId, and upsertJobScheduler validates them the same way, so an invalid template is rejected at upsert. Ordering keys, group concurrency and rate limits, token buckets and cost apply to every scheduled job. jobId is rejected: each run gets a generated id, and a fixed id would drop every run after the first as a duplicate. delay, deduplication and parent are rejected too, since the tick never applies them.

LIFO Mode

Set lifo: true in JobOptions to process jobs in last-in-first-out order. The most recently added job is picked up first.

typescript

await queue.add('render', { frame: 100 }, { lifo: true });
await queue.add('render', { frame: 101 }, { lifo: true });
await queue.add('render', { frame: 102 }, { lifo: true });
// Processing order: 102, 101, 100

Ordering precedence

Workers check sources in this order: priority > LIFO > FIFO. Priority jobs (those with priority > 0) are always fetched first. Among non-priority jobs, LIFO jobs are fetched before FIFO jobs sitting in the stream.

Constraints

Cannot combine with ordering.key. Throws at enqueue time:
```
Error: lifo and ordering.key cannot be used together
```
LIFO jobs are stored in a dedicated Valkey LIST (glide:{queueName}:lifo), separate from the main stream. This means LIFO and FIFO jobs in the same queue coexist - LIFO jobs are drained first.
Under concurrency > 1, multiple LIFO jobs may run in parallel; strict reverse ordering is only guaranteed with concurrency: 1.
Works with all job types: delayed jobs return to the LIFO list after their delay expires, and schedulers can produce LIFO jobs via the template opts.

See also: Adding jobs for the full JobOptions reference.

Job TTL

Set ttl in JobOptions to auto-expire jobs that are not processed within a time window. The value is in milliseconds.

typescript

// Expire if not processed within 30 seconds
await queue.add(
  'time-sensitive',
  { alert: 'server-down' },
  {
    ttl: 30_000,
  },
);

// TTL works with delayed jobs  -  the clock starts at creation time
await queue.add(
  'offer',
  { code: 'FLASH50' },
  {
    delay: 5_000,
    ttl: 60_000, // must be processed within 60s of creation, not of becoming active
  },
);

// TTL works with priority jobs
await queue.add('urgent', data, {
  priority: 1,
  ttl: 10_000,
});

When a job's TTL elapses, it is failed with the reason 'expired' during the next activation check. Jobs that are already active are not interrupted - TTL is checked at fetch time, not mid-processing. Use timeout in JobOptions to limit active processing time.

See also: Adding jobs for other per-job options.

Pluggable Serializers

By default, glide-mq uses JSON.stringify / JSON.parse for job data, return values, and progress payloads. You can replace this with any synchronous serializer.

The `Serializer` interface

typescript

import type { Serializer } from 'glide-mq';

interface Serializer {
  /** Serialize a value to a string for storage in Valkey. */
  serialize(data: unknown): string;
  /** Deserialize a string from Valkey back to a value. */
  deserialize(raw: string): unknown;
}

Both methods must be synchronous. If serialize throws, the job is treated as a processor failure (in Worker). A scheduler template whose data cannot be serialized, or exceeds the 1 MB limit, is rejected by upsertJobScheduler; a stored template that still fails at run time skips that run, reports the error through the worker error event and moves on to the next occurrence.

Example: MessagePack serializer

typescript

import { Queue, Worker } from 'glide-mq';
import { encode, decode } from '@msgpack/msgpack';

const msgpackSerializer: Serializer = {
  serialize: (data) => Buffer.from(encode(data)).toString('base64'),
  deserialize: (raw) => decode(Buffer.from(raw, 'base64')),
};

const queue = new Queue('tasks', {
  connection,
  serializer: msgpackSerializer,
});

const worker = new Worker('tasks', processor, {
  connection,
  serializer: msgpackSerializer, // must match the Queue
});

What is serialized

The serializer is applied to:

data - the job payload passed to queue.add().
returnvalue - the value returned by the processor.
progress - the value passed to job.updateProgress().

Consistency requirement

The same serializer must be configured on every Queue, Worker, and FlowProducer instance that operates on the same queue. A mismatch causes silent data corruption - the consumer will see {} and the job's deserializationFailed flag will be true.

Default export

The built-in JSON serializer is exported for use in conditional logic or testing:

typescript

import { JSON_SERIALIZER } from 'glide-mq';

const serializer = process.env.USE_MSGPACK === '1' ? msgpackSerializer : JSON_SERIALIZER;

const queue = new Queue('tasks', { connection, serializer });

See also: Worker and Queue for where serializer appears in options.

Ordering and Group Concurrency

Sequential processing (concurrency=1)

Add ordering.key to a job to guarantee that all jobs with the same key are processed one at a time, in the order they were added.

typescript

// All jobs with ordering.key = 'user:42' are processed sequentially
await queue.add(
  'process-payment',
  { userId: 42, amount: 100 },
  {
    ordering: { key: 'user:42' },
  },
);
await queue.add(
  'send-receipt',
  { userId: 42 },
  {
    ordering: { key: 'user:42' },
  },
);

Group concurrency (concurrency > 1)

Set ordering.concurrency to allow up to N jobs per key to run in parallel across all workers:

typescript

// Max 3 concurrent jobs for tenant-42, regardless of worker count
await queue.add('process', data, {
  ordering: { key: 'tenant-42', concurrency: 3 },
});

Jobs exceeding the group limit are parked in a per-group wait list and automatically released when a slot opens.

typescript

// Multi-tenant isolation: each client gets max 2 concurrent jobs
for (const job of jobs) {
  await queue.add('task', job.data, {
    ordering: { key: `client-${job.clientId}`, concurrency: 2 },
  });
}

Per-group rate limiting

Limit how many jobs per ordering key can start within a time window, independent of concurrency:

typescript

// Max 10 jobs per 60 seconds for each tenant
await queue.add('sync', data, {
  ordering: {
    key: `tenant-${tenantId}`,
    concurrency: 3,
    rateLimit: { max: 10, duration: 60_000 },
  },
});

When both concurrency and rateLimit are set, both gates apply - a job must have a free concurrency slot and remaining rate capacity to start. Jobs that hit the rate limit are parked in a scheduler-managed promotion queue and released when the window resets.

Promotion latency: rate-limited jobs are promoted by the scheduler loop. Worst-case latency is one promotionInterval (default 5 s). Lower promotionInterval on the worker if tighter latency is needed.
Retried jobs consume rate slots - a retried job counts against the rate window like any new job.

Token bucket rate limiting

Use ordering.tokenBucket to enforce cost-based rate limiting per ordering key. Unlike the sliding window (rateLimit), which counts jobs, the token bucket assigns a cost to each job and deducts from a refilling bucket:

typescript

// Each API call costs 1 token (default), bulk exports cost 10
await queue.add('api-call', data, {
  ordering: {
    key: `tenant-${tenantId}`,
    concurrency: 5,
    tokenBucket: { capacity: 100, refillRate: 10 }, // 100 tokens max, 10 tokens/s
  },
  cost: 1,
});

await queue.add('bulk-export', data, {
  ordering: {
    key: `tenant-${tenantId}`,
    concurrency: 5,
    tokenBucket: { capacity: 100, refillRate: 10 },
  },
  cost: 10, // consumes 10 tokens
});

How it works: tokens refill at refillRate tokens per second up to capacity. When a job is activated, its cost is deducted from the bucket. If insufficient tokens remain, the job is parked and promoted once enough tokens have refilled. Internally, tokens are tracked as millitokens (1 token = 1000 millitokens) for sub-integer precision.

Check order: when both concurrency, token bucket, and sliding window are configured, the gates are checked in order: concurrency -> token bucket -> sliding window. All applicable limits must pass. Strict FIFO is maintained - jobs never skip ahead of earlier jobs in the same group.

Cost validation: a job with cost greater than capacity is rejected at enqueue time. If a previously valid job becomes invalid (e.g., capacity was lowered), it is failed at activation with cost exceeds token bucket capacity. A worker with deadLetterQueue adds a DLQ copy whether the failure happens in moveToActive or while completeAndFetchNext / failAndFetchNext fetch the next job or promote the group. A job failed by the scheduler tick's rate-limited group promotion gets no DLQ copy (no worker owns it).

Differences from sliding window (rateLimit):

	Sliding window (`rateLimit`)	Token bucket (`tokenBucket`)
Unit	Job count	Weighted cost per job
Config	`{ max, duration }`	`{ capacity, refillRate }`
Default cost	1 job	`cost: 1` token
Refill	Window resets after `duration` ms	Continuous refill at `refillRate`/s
Use case	"Max N jobs per window"	"Max N units of work per second"

Promotion latency: same as sliding window - worst-case one promotionInterval (default 5 s).
Composition: token bucket composes with concurrency, sliding window, and global rate limits. All gates are enforced.

Notes

Jobs with different ordering keys (or no ordering key) are processed concurrently as normal.
Ordering keys are limited to 256 characters.
concurrency=1 (or omitted) preserves strict FIFO ordering per key.
concurrency > 1 caps parallelism but does not guarantee FIFO within the group.
Group concurrency and global concurrency (setGlobalConcurrency) compose: both limits are enforced.
Per-group rate limiting, token bucket, group concurrency, and global concurrency all compose: all applicable limits are enforced.
Group slots are released on job complete, fail, retry, DLQ move, and stall recovery.

Custom Job IDs

By default glide-mq assigns a monotonically increasing integer ID to each job. You can supply your own ID via opts.jobId to get deterministic, idempotent job creation:

typescript

// Deterministic job: safe to call multiple times
const job = await queue.add(
  'send-email',
  { to: 'user@example.com' },
  {
    jobId: 'email-user-42',
  },
);
// job is null if a job with this ID already exists (silent skip)

Constraints

Max 256 characters.
Must not contain control characters (U+0000-U+001F, U+007F), curly braces ({, }), or colons (:).
Violating either constraint throws synchronously before the network call.

Duplicate behaviour by surface

Surface	Behaviour on duplicate ID
`Queue.add`	Returns `null` (silent skip)
`Queue.addBulk`	Silently omits the duplicate from the returned array
`FlowProducer.add`	Throws `Duplicate job ID in flow`
`TestQueue.add`	Returns `null` (mirrors production)

FlowProducer.add checks every custom ID in a level before writing it, so the level holding the duplicate is not created. Nested sub-flows are separate calls made first, so sub-flows created before the failing level stay in place.

Interaction with deduplication

opts.jobId and opts.deduplication are independent mechanisms. When both are set the deduplication check runs first; if the job is deduplicated, the custom ID is never stored. If the dedup check passes, the custom ID collision check runs next.

Deduplication

Prevent duplicate jobs from entering the queue using deduplication.id. Three modes are supported:

Mode	Behaviour
`simple`	Skip the new job while the job holding the ID is not yet completed or failed. `ttl` is ignored.
`throttle`	Skip the new job for `ttl` ms after the job that took the ID was added. Without `ttl`, nothing is skipped.
`debounce`	Replace the job holding the ID if it is still `delayed` or `prioritized`; skip if it is waiting or active. `ttl` is ignored.

typescript

// Simple: skip while a job with this ID is queued or active; the ID frees once it completes or fails
await queue.add(
  'send-welcome',
  { userId: 99 },
  {
    deduplication: { id: 'welcome-99', mode: 'simple' },
  },
);

// Throttle: at most one "sync" job per 10 s
await queue.add(
  'sync',
  { region: 'eu' },
  {
    deduplication: { id: 'sync-eu', mode: 'throttle', ttl: 10_000 },
  },
);

// Debounce: each add within 500 ms replaces the pending delayed job, so only the last one runs
await queue.add(
  'search',
  { query: 'hello' },
  {
    delay: 500,
    deduplication: { id: 'search-user-1', mode: 'debounce' },
  },
);

Debounce needs a delay (or a priority job not yet promoted). Once the job is waiting or active, later adds with the same ID are skipped, and once it completes or fails the next add starts a new job.

queue.add() returns null when a job is skipped by deduplication.

Global Concurrency

Limit the total number of concurrently active jobs across all workers sharing a queue, regardless of per-worker concurrency settings.

typescript

const queue = new Queue('tasks', { connection });

// Allow at most 20 active jobs across all workers at once
await queue.setGlobalConcurrency(20);

// Remove the limit
await queue.setGlobalConcurrency(0);

Workers read the limit from queue metadata on each scheduler tick. Priority and LIFO jobs are popped with an atomic check (glidemq_rpopAndReserve). Stream jobs are gated twice: glidemq_checkConcurrency before XREADGROUP keeps a worker from reading when the queue is full, and glidemq_moveToActive enforces the cap at activation. A stream claim counts in the consumer group's pending list as soon as XREADGROUP returns it, so activation ranks the pending claims by entry id: the oldest globalConcurrency - listActive claims keep their slots, a newer claim gets GLOBAL_FULL. The worker keeps that claim in the pending list (so it keeps its place in the order) and retries the activation with a capped backoff (20 ms doubling to 250 ms, woken early by a completion in the same worker) until it is admitted, for at most half the shorter of lockDuration and stalledInterval; only after that bound, or in batch mode, is the entry handed back (glidemq_deferActive: XACK, re-added to the stream as waiting, which costs it its position). close() and pause() hand a held claim back at once. completeAndFetchNext does not claim the next stream job while pending claims plus list claims already fill the cap, so a held claim is not overtaken by a chain. Workers polling at the same time therefore never run more than globalConcurrency jobs and drain a burst in FIFO order. The rank check runs only when the pending count exceeds the free slots.

Global Rate Limiting

Cap the total job throughput across all workers sharing a queue. The config is stored in the Valkey meta hash and picked up dynamically by workers within one scheduler tick.

typescript

const queue = new Queue('tasks', { connection });

// Max 500 jobs per minute across all workers
await queue.setGlobalRateLimit({ max: 500, duration: 60_000 });

// Read current config
const limit = await queue.getGlobalRateLimit();
// { max: 500, duration: 60000 } or null if not set

// Remove the limit
await queue.removeGlobalRateLimit();

While a global rate limit is set, it replaces WorkerOptions.limiter on every worker, even if the worker limiter is stricter. Removing it restores the worker limiter.
Changes are picked up by workers within one scheduler tick (no restart needed).

Job Revocation

Cooperatively cancel a job that is waiting, delayed, or currently being processed.

typescript

const job = await queue.add('long-task', { input: 'data' });

// Later...
const result = await queue.revoke(job.id);
// 'revoked'   -  job was waiting/delayed and is now in the failed set
// 'flagged'   -  job is active; the worker will abort it cooperatively
// 'not_found' -  job does not exist

In your processor, use job.abortSignal to react to revocation:

typescript

const worker = new Worker(
  'tasks',
  async (job) => {
    for (const chunk of largeDataset) {
      if (job.abortSignal?.aborted) {
        throw new Error('Job revoked');
      }
      await processChunk(chunk);
    }
    return { done: true };
  },
  { connection },
);

job.abortSignal is an AbortSignal. You can pass it directly to fetch, axios, or any AbortSignal-aware API.

Transparent Compression

Enable gzip compression at the queue level. Workers decompress automatically - no changes required in processors.

typescript

const queue = new Queue('tasks', {
  connection,
  compression: 'gzip',
});

// Payload is gzip-compressed before storing in Valkey
await queue.add('process-large', { report: '... 15 KB of data ...' });
// Stored size: ~300 bytes (98% savings on repetitive data)

Job schedulers follow the Queue that upserts them: upsertJobScheduler on a gzip Queue records compression: 'gzip' on the entry, and every run stores its template data compressed.

Payload size limit: job data must be ≤ 1 MB after serialisation but before compression. Larger payloads throw immediately:

Error: Job data exceeds maximum size (1234567 bytes > 1MB).
       Use smaller payloads or store large data externally.

Store large blobs in S3/GCS/object storage and pass a reference URL in the job data instead.

Retries and Backoff

Configure retry behaviour per job via attempts and backoff:

typescript

await queue.add('send-email', data, {
  attempts: 5,
  backoff: { type: 'exponential', delay: 1_000 },
  // delay sequence: 1s, 2s, 4s, 8s (capped at attempts)
});

// Fixed delay
await queue.add('webhook', data, {
  attempts: 3,
  backoff: { type: 'fixed', delay: 2_000 },
});

// Exponential with jitter (avoids thundering herd)
await queue.add('poll', data, {
  attempts: 10,
  backoff: { type: 'exponential', delay: 500, jitter: 0.1 },
});

// Custom strategy  -  register on the Worker
const worker = new Worker('tasks', processor, {
  connection,
  backoffStrategies: {
    'rate-limited': (attemptsMade, err) => {
      // Respect Retry-After header
      if (err.retryAfter) return err.retryAfter * 1_000;
      return attemptsMade * 3_000;
    },
  },
});

await queue.add('api-call', data, {
  attempts: 5,
  backoff: { type: 'rate-limited', delay: 0 },
});

When attempts is exhausted the job moves to the failed state. If the worker has a DLQ configured, a copy also goes to the DLQ.

Dead Letter Queues

Copy permanently failed jobs to a separate queue for later inspection. Configure it on the Worker; deadLetterQueue on a Queue only tells getDeadLetterJobs() which queue to read.

typescript

const worker = new Worker('tasks', processor, {
  connection,
  deadLetterQueue: { name: 'tasks-dlq' },
});

// Inspect DLQ contents
const dlqQueue = new Queue('tasks-dlq', { connection });
const failedJobs = await dlqQueue.getJobs('waiting');

// Or use the convenience method on the original queue
const dlqJobs = await queue.getDeadLetterJobs(0, 49);

A job is copied when it fails terminally, which the job's own attempts option decides; deadLetterQueue.maxRetries is not read (deprecated, removed in the next major version).

The DLQ entry is a best-effort copy. The original job stays in the failed state of its own queue (subject to removeOnFail), and if writing the copy fails the worker emits error and moves on. The entry is added to the DLQ queue as a new waiting job named like the original, whose data is a JSON envelope: { originalQueue, originalJobId, data, failedReason, attemptsMade }. A worker on the DLQ queue would process these entries.

Because the entry is waiting, not failed, Job.retry() on it throws. To retry, call retry() on the original failed job (queue.getJob(originalJobId)), or re-add the envelope's data to the original queue.

Fallback Chains

Configure ordered fallback models/providers on a per-job basis. On each retryable failure, the worker advances to the next entry in the chain.

job.fallbackIndex starts at 0. currentFallback returns undefined (use your default model).
On the first retry failure, glidemq_fail sets fallbackIndex to 1. currentFallback returns fallbacks[0].
Once fallbackIndex passes the end of the array, currentFallback returns undefined (back to your default model).
Each entry supports metadata for provider-specific parameters.

See USAGE.md for usage examples.

Dual-axis Rate Limiting (RPM + TPM)

The existing limiter option on WorkerOptions caps requests per time window (RPM). The tokenLimiter option adds a parallel token-per-minute (TPM) limit. Both compose - the worker pauses when either limit is hit.

Jobs report tokens via job.reportTokens(count) or via job.reportUsage() (which auto-extracts totalTokens).

Scope options

Scope	Where tracked	When to use
queue	glide:{queueName}:tpm hash	Multi-worker, strict global limit
worker	In-memory counter	Single worker, zero-latency checks
both (default)	Local first, then Valkey	Fast local check avoids Valkey when under limit

See USAGE.md for configuration examples.

Per-job Lock Duration

By default, all jobs share the worker-level lockDuration (default: 30000ms). Override per job via opts.lockDuration. The per-job value is stored in the job opts JSON field and read by glidemq_reclaimStalled and glidemq_reclaimStalledListJobs at stall-recovery time.

Constraints: must be a positive integer; values below 5000ms risk false stall detection under load.

See USAGE.md for examples.

Vector Search Index Management

queue.createJobIndex() creates a Valkey Search index over job hashes. The index enables both full-text search and vector similarity queries.

Base fields (name, state, timestamp, priority) are always included. Users add custom fields and a vector field via the fields and vectorField options.

Distance metrics

Metric	Score interpretation	Use case
COSINE	0 = identical, 2 = opposite (lower = better)	Text embeddings, semantic similarity
L2	0 = identical (lower = better)	Image features, spatial data
IP	Higher = more similar	Normalized embeddings, recommendation

See USAGE.md for full API and examples.

Request Timeout

Override the default 500ms command timeout for operations that may take longer:

typescript

const queue = new Queue('my-queue', {
  connection: {
    addresses: [{ host: 'localhost', port: 6379 }],
    requestTimeout: 5000, // 5 seconds for FT.CREATE, FUNCTION LOAD
  },
});

The default (500ms) is sufficient for most operations. Increase for createJobIndex() on databases with many existing keys, or FUNCTION LOAD with large libraries.

Advanced Features ​

Table of Contents ​

Shared Client ​

Default behavior (dedicated connections) ​

Shared client (opt-in) ​

What can share ​

What cannot share ​

Tradeoffs ​

When to use shared ​

When to stick with dedicated ​

Constraints ​

Close order ​

Producer with an external client ​

inflightRequestsLimit ​

Job Schedulers ​

Cron syntax ​

Repeat-after-complete mode ​

Bounded schedulers ​

LIFO Mode ​

Ordering precedence ​

Constraints ​

Job TTL ​

Pluggable Serializers ​

The Serializer interface ​

Example: MessagePack serializer ​

What is serialized ​

Consistency requirement ​

Default export ​

Ordering and Group Concurrency ​

Sequential processing (concurrency=1) ​

Group concurrency (concurrency > 1) ​

Per-group rate limiting ​

Token bucket rate limiting ​

Notes ​

Custom Job IDs ​

Deduplication ​

Global Concurrency ​

Global Rate Limiting ​

Job Revocation ​

Transparent Compression ​

Retries and Backoff ​

Dead Letter Queues ​

Fallback Chains ​

Dual-axis Rate Limiting (RPM + TPM) ​

Scope options ​

Per-job Lock Duration ​

Vector Search Index Management ​

Distance metrics ​

Request Timeout ​

Advanced Features

Table of Contents

Shared Client

Default behavior (dedicated connections)

Shared client (opt-in)

What can share

What cannot share

Tradeoffs

When to use shared

When to stick with dedicated

Constraints

Close order

Producer with an external client

`inflightRequestsLimit`

Job Schedulers

Cron syntax

Repeat-after-complete mode

Bounded schedulers

LIFO Mode

Ordering precedence

Constraints

Job TTL

Pluggable Serializers

The `Serializer` interface

Example: MessagePack serializer

What is serialized

Consistency requirement

Default export

Ordering and Group Concurrency

Sequential processing (concurrency=1)

Group concurrency (concurrency > 1)

Per-group rate limiting

Token bucket rate limiting

Notes

Custom Job IDs

Deduplication

Global Concurrency

Global Rate Limiting

Job Revocation

Transparent Compression

Retries and Backoff

Dead Letter Queues

Fallback Chains

Dual-axis Rate Limiting (RPM + TPM)

Scope options

Per-job Lock Duration

Vector Search Index Management

Distance metrics

Request Timeout