Glide Mq Migrate Bullmq

Migrate from BullMQ to glide-mq

The glide-mq API is intentionally similar to BullMQ. Most changes are connection format and imports.

When to Apply

Use this skill when:

• Replacing BullMQ with glide-mq in an existing project
• Converting BullMQ Queue/Worker/FlowProducer code
• Updating connection configuration from ioredis to valkey-glide format
• Comparing API differences between BullMQ and glide-mq

Prerequisites

• Node.js 20+
• Valkey 7.0+ or Redis 7.0+ (both supported)
• TypeScript 5+ recommended

Install

npm remove bullmq
npm install glide-mq

// Before
import { Queue, Worker, Job, QueueEvents, FlowProducer } from 'bullmq';

// After
import { Queue, Worker, Job, QueueEvents, FlowProducer } from 'glide-mq';

---

Breaking changes

---

Step-by-step conversion

1. Connection config (the biggest change)

// BEFORE (BullMQ)
const connection = { host: 'localhost', port: 6379 };

// AFTER (glide-mq)
const connection = { addresses: [{ host: 'localhost', port: 6379 }] };

For TLS + password + cluster, see references/connection-mapping.md.

2. Queue.add - identical API

// BEFORE
const queue = new Queue('tasks', { connection });
await queue.add('send-email', { to: 'user@example.com' });

// AFTER - only the connection changes
const queue = new Queue('tasks', { connection });
await queue.add('send-email', { to: 'user@example.com' });

3. Worker - identical API, different connection

// BEFORE
const worker = new Worker('tasks', async (job) => {
  await sendEmail(job.data.to);
}, { connection: { host: 'localhost', port: 6379 }, concurrency: 10 });

// AFTER
const worker = new Worker('tasks', async (job) => {
  await sendEmail(job.data.to);
}, { connection: { addresses: [{ host: 'localhost', port: 6379 }] }, concurrency: 10 });

4. FlowProducer - identical API

// Both - same usage, only connection format differs
const flow = new FlowProducer({ connection });
await flow.add({
  name: 'parent',
  queueName: 'tasks',
  data: { step: 'final' },       // NOTE: data is required in glide-mq
  children: [
    { name: 'child-1', queueName: 'tasks', data: { step: '1' } },
    { name: 'child-2', queueName: 'tasks', data: { step: '2' } },
  ],
});

5. QueueEvents - identical API

// Both - same, only connection format differs
const qe = new QueueEvents('tasks', { connection });
qe.on('completed', ({ jobId }) => console.log(jobId, 'done'));
qe.on('failed', ({ jobId, failedReason }) => console.error(jobId, failedReason));

Note: some BullMQ events are not yet emitted. See Current gaps.

6. Graceful shutdown

// BullMQ
await worker.close();
await queue.close();

// glide-mq - identical
await worker.close();
await queue.close();

7. UnrecoverableError - identical

// Both
import { UnrecoverableError } from 'glide-mq'; // was 'bullmq'

throw new UnrecoverableError('permanent failure');

8. Scheduling (repeatable jobs)

// BEFORE - opts.repeat (deprecated in BullMQ v5)
await queue.add('report', data, {
  repeat: { pattern: '0 9 * * *', tz: 'America/New_York' },
});

// AFTER - upsertJobScheduler
await queue.upsertJobScheduler(
  'report',
  { pattern: '0 9 * * *', tz: 'America/New_York' },
  { name: 'report', data: { v: 1 } },
);

9. Custom backoff strategies

// BEFORE
const worker = new Worker('q', processor, {
  connection,
  settings: {
    backoffStrategy: (attemptsMade, type, delay, err) => {
      if (type === 'jitter') return delay + Math.random() * delay;
      return delay * attemptsMade;
    },
  },
});

// AFTER
const worker = new Worker('q', processor, {
  connection,
  backoffStrategies: {
    jitter: (attemptsMade, err) => 1000 + Math.random() * 1000,
    linear: (attemptsMade, err) => 1000 * attemptsMade,
  },
});

10. defaultJobOptions removal

// BEFORE
const queue = new Queue('tasks', {
  connection,
  defaultJobOptions: { attempts: 3, backoff: { type: 'exponential', delay: 1000 } },
});

// AFTER - wrap add() with your defaults
const DEFAULTS = { attempts: 3, backoff: { type: 'exponential', delay: 1000 } } as const;
const add = (name: string, data: unknown, opts?: JobOptions) =>
  queue.add(name, data, { ...DEFAULTS, ...opts });

11. getJobs with multiple types

// BEFORE
const jobs = await queue.getJobs(['waiting', 'active'], 0, 99);

// AFTER
const [waiting, active] = await Promise.all([
  queue.getJobs('waiting', 0, 99),
  queue.getJobs('active', 0, 99),
]);
const jobs = [...waiting, ...active];

12. job.waitUntilFinished

// BEFORE
const qe = new QueueEvents('tasks', { connection });
const result = await job.waitUntilFinished(qe, 30000);

// AFTER - no QueueEvents needed
const result = await job.waitUntilFinished(500, 30000);
// args: pollIntervalMs (default 500), timeoutMs (default 30000)

13. BullMQ Pro groups to ordering keys

// BEFORE (BullMQ Pro)
await queue.add('job', data, {
  group: { id: 'tenant-123', limit: { max: 2, duration: 0 } },
});

// AFTER (glide-mq, open source)
await queue.add('job', data, {
  ordering: { key: 'tenant-123', concurrency: 2 },
});

---

What's new in glide-mq (not in BullMQ)

See references/new-features.md for detailed documentation.

---

Current gaps

---

Performance comparison

AWS ElastiCache Valkey 8.2 (r7g.large), TLS enabled, same-region EC2 client.

Most production deployments run c=5 to c=20, where glide-mq's 1-RTT architecture pays off the most.

---

Migration checklist

- [ ] Replace `bullmq` with `glide-mq` in package.json
- [ ] Update all imports from 'bullmq' to 'glide-mq'
- [ ] Convert connection configs: { host, port } -> { addresses: [{ host, port }] }
- [ ] Convert TLS: tls: {} -> useTLS: true
- [ ] Convert password: password -> credentials: { password }
- [ ] Replace opts.repeat with queue.upsertJobScheduler()
- [ ] Replace settings.backoffStrategy with backoffStrategies map
- [ ] Remove QueueScheduler instantiation (not needed)
- [ ] Remove defaultJobOptions from QueueOptions; apply per job or via wrapper
- [ ] Replace queue.getJobs([...types]) with per-type calls
- [ ] Update worker.on('active') handlers: (job, jobId) not (job, prev)
- [ ] Replace job.waitUntilFinished(queueEvents, ttl) with (pollMs, timeoutMs)
- [ ] Check QueueEvents listeners for removed events (waiting, active, delayed, drained)
- [ ] Replace group.id (BullMQ Pro) with ordering.key
- [ ] Run test suite: npm test
- [ ] Confirm queue counts: await queue.getJobCounts()
- [ ] Confirm no jobs stuck in active state
- [ ] Smoke-test QueueEvents or SSE listeners if the app exposes them
- [ ] Confirm workers, queues, and connections close cleanly

---

Troubleshooting

Full Documentation

• Migration Guide
• New Features Reference
• Connection Mapping Reference

Connection config mapping: BullMQ to glide-mq

BullMQ uses ioredis's flat connection format. glide-mq uses valkey-glide's structured format with an addresses array. This is the most common source of migration errors.

---

Basic (standalone)

// BullMQ
const connection = { host: 'localhost', port: 6379 };

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

---

TLS

// BullMQ
const connection = {
  host: 'my-server.example.com',
  port: 6380,
  tls: {},
};

// glide-mq
const connection = {
  addresses: [{ host: 'my-server.example.com', port: 6380 }],
  useTLS: true,
};

Note: BullMQ uses an empty tls: {} object (or with TLS options). glide-mq uses a boolean useTLS: true.

---

Password authentication

// BullMQ
const connection = {
  host: 'my-server.example.com',
  port: 6379,
  password: 'secret',
};

// glide-mq
const connection = {
  addresses: [{ host: 'my-server.example.com', port: 6379 }],
  credentials: { password: 'secret' },
};

---

Username + password (ACL auth)

// BullMQ
const connection = {
  host: 'my-server.example.com',
  port: 6379,
  username: 'myuser',
  password: 'secret',
};

// glide-mq
const connection = {
  addresses: [{ host: 'my-server.example.com', port: 6379 }],
  credentials: { username: 'myuser', password: 'secret' },
};

---

TLS + password + cluster

// BullMQ
const connection = {
  host: 'my-cluster.cache.amazonaws.com',
  port: 6379,
  tls: {},
  password: 'secret',
};
// BullMQ auto-detects cluster mode in some configurations, or you use natMap

// glide-mq
const connection = {
  addresses: [{ host: 'my-cluster.cache.amazonaws.com', port: 6379 }],
  useTLS: true,
  credentials: { password: 'secret' },
  clusterMode: true,
};

Key difference: glide-mq requires explicit clusterMode: true for Redis Cluster / ElastiCache cluster / MemoryDB.

---

IAM authentication (AWS ElastiCache / MemoryDB)

BullMQ has no equivalent. This is glide-mq only.

// glide-mq only
const connection = {
  addresses: [{ host: 'my-cluster.cache.amazonaws.com', port: 6379 }],
  useTLS: true,
  clusterMode: true,
  credentials: {
    type: 'iam',
    serviceType: 'elasticache',   // or 'memorydb'
    region: 'us-east-1',
    userId: 'my-iam-user',
    clusterName: 'my-cluster',
  },
};

No credential rotation needed - the client handles IAM token refresh automatically.

---

AZ-affinity routing (cluster only)

BullMQ has no equivalent. Reduces cross-AZ network cost and latency.

// glide-mq only
const connection = {
  addresses: [{ host: 'cluster.cache.amazonaws.com', port: 6379 }],
  clusterMode: true,
  useTLS: true,
  readFrom: 'AZAffinity',
  clientAz: 'us-east-1a',
};

---

Multiple seed nodes (cluster)

// BullMQ - typically one host, or uses natMap for discovery
const connection = { host: 'node-1.example.com', port: 6379 };

// glide-mq - pass multiple seed addresses for cluster discovery
const connection = {
  addresses: [
    { host: 'node-1.example.com', port: 6379 },
    { host: 'node-2.example.com', port: 6379 },
    { host: 'node-3.example.com', port: 6379 },
  ],
  clusterMode: true,
};

---

Option mapping table

---

Common mistakes

1. Forgetting the array wrapper: { addresses: { host, port } } will fail. It must be { addresses: [{ host, port }] } - note the square brackets.

2. Using `tls: {}` instead of `useTLS: true`: valkey-glide does not accept a TLS options object. Pass the boolean flag.

3. Omitting `clusterMode: true`: Unlike ioredis which can auto-detect cluster mode, valkey-glide requires you to explicitly opt in.

4. Using `password` at top level: Must be credentials: { password }, not password directly.

glide-mq features not available in BullMQ

These features have no BullMQ equivalent. They are available after migrating to glide-mq.

---

Per-key ordering

Guarantees sequential execution per key across all workers, regardless of worker concurrency. Jobs with the same ordering.key run one at a time in enqueue order. Jobs with different keys run in parallel.

await queue.add('sync', data, {
  ordering: { key: 'tenant-123' },
});

Replaces BullMQ Pro's group.id feature (which requires a Pro license).

Group concurrency

Allow N parallel jobs per key instead of strict serialization:

await queue.add('sync', data, {
  ordering: { key: 'tenant-123', concurrency: 3 },
});

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

Per-group rate limiting

Cap throughput per ordering key:

await queue.add('sync', data, {
  ordering: {
    key: 'tenant-123',
    concurrency: 3,
    rateLimit: { max: 10, duration: 60_000 },
  },
});

Rate-limited jobs are promoted by the scheduler loop (latency up to promotionInterval, default 5 s).

Cost-based token bucket

Assign a cost to each job and deduct from a refilling bucket per key:

await queue.add('heavy-job', data, {
  ordering: {
    key: 'tenant-123',
    tokenBucket: { capacity: 100, refillRate: 10 },
  },
  cost: 25,  // this job consumes 25 tokens
});

---

Global rate limiting

Queue-wide rate limit stored in Valkey, dynamically picked up by all workers:

await queue.setGlobalRateLimit({ max: 500, duration: 60_000 });

const limit = await queue.getGlobalRateLimit(); // { max, duration } or null
await queue.removeGlobalRateLimit();

When both global rate limit and WorkerOptions.limiter are set, the stricter limit wins.

---

Dead letter queue

First-class DLQ support configured at the queue level:

const queue = new Queue('tasks', {
  connection,
  deadLetterQueue: {
    name: 'tasks-dlq',
    maxRetries: 3,
  },
});

// Retrieve DLQ jobs:
const dlqQueue = new Queue('tasks-dlq', { connection });
const dlqJobs = await dlqQueue.getDeadLetterJobs();

BullMQ has no native DLQ - failed jobs stay in the failed state.

---

Job revocation

Cancel an in-flight job from outside the worker:

await queue.revoke(jobId);

The processor must cooperate via job.abortSignal:

const worker = new Worker('q', async (job) => {
  for (const chunk of data) {
    if (job.abortSignal?.aborted) return;
    await processChunk(chunk);
  }
}, { connection });

---

Transparent compression

Gzip compression of all job payloads, transparent to application code:

const queue = new Queue('tasks', {
  connection,
  compression: 'gzip',
});
// No changes needed in worker or job code

98% payload reduction on 15 KB JSON payloads (15 KB -> 331 bytes).

---

AZ-affinity routing

Pin worker reads to replicas in your availability zone to reduce cross-AZ network cost:

const connection = {
  addresses: [{ host: 'cluster.cache.amazonaws.com', port: 6379 }],
  clusterMode: true,
  readFrom: 'AZAffinity',
  clientAz: 'us-east-1a',
};

---

IAM authentication

Native AWS ElastiCache and MemoryDB IAM auth with automatic token refresh:

const connection = {
  addresses: [{ host: 'my-cluster.cache.amazonaws.com', port: 6379 }],
  useTLS: true,
  clusterMode: true,
  credentials: {
    type: 'iam',
    serviceType: 'elasticache',
    region: 'us-east-1',
    userId: 'my-iam-user',
    clusterName: 'my-cluster',
  },
};

---

In-memory test mode

Test queue logic without a running Valkey/Redis instance:

import { TestQueue, TestWorker } from 'glide-mq/testing';

const queue = new TestQueue<{ email: string }, { sent: boolean }>('tasks');
const worker = new TestWorker(queue, async (job) => {
  return { sent: true };
});

await queue.add('send-email', { email: 'user@example.com' });
await new Promise(r => setTimeout(r, 10));

const jobs = await queue.getJobs('completed');

BullMQ has no equivalent. Typically requires ioredis-mock or a real Redis instance.

---

Broadcast / BroadcastWorker

Pub/sub fan-out where every connected BroadcastWorker receives every message. Supports per-subscriber retries for reliable delivery:

import { Broadcast, BroadcastWorker } from 'glide-mq';

const broadcast = new Broadcast('notifications', { connection });
const bw = new BroadcastWorker('notifications', async (message) => {
  console.log('Received:', message);
}, { connection, subscription: 'my-group' });

await broadcast.publish('alerts', { type: 'alert', text: 'Server restarting' });

---

Batch processing

Process multiple jobs in a single processor invocation:

const worker = new Worker('q', async (jobs) => {
  // jobs is an array when batch mode is enabled
  const results = await bulkProcess(jobs.map(j => j.data));
  return results; // per-job results array
}, {
  connection,
  batch: { size: 50, timeout: 1000 },
});

---

DAG workflows

Arbitrary directed acyclic graphs where a job can depend on multiple parents (BullMQ only supports trees - one parent per job):

import { FlowProducer, dag } from 'glide-mq';

// Option 1: dag() helper - standalone, creates its own FlowProducer
const jobs = await dag([
  { name: 'fetch-a', queueName: 'tasks', data: { source: 'a' } },
  { name: 'fetch-b', queueName: 'tasks', data: { source: 'b' } },
  { name: 'aggregate', queueName: 'tasks', data: {}, deps: ['fetch-a', 'fetch-b'] },
], connection);

// Option 2: FlowProducer.addDAG() - when you manage the FlowProducer
const flow = new FlowProducer({ connection });
const jobs2 = await flow.addDAG({
  nodes: [
    { name: 'fetch-a', queueName: 'tasks', data: { source: 'a' } },
    { name: 'fetch-b', queueName: 'tasks', data: { source: 'b' } },
    { name: 'aggregate', queueName: 'tasks', data: {}, deps: ['fetch-a', 'fetch-b'] },
  ],
});
await flow.close();

---

Workflow helpers

Higher-level orchestration built on FlowProducer:

import { chain, group, chord } from 'glide-mq';

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

// chain: sequential pipeline
await chain('tasks', [
  { name: 'step-1', data: {} },
  { name: 'step-2', data: {} },
  { name: 'step-3', data: {} },
], connection);

// group: parallel fan-out, synthetic parent waits for all
await group('tasks', [
  { name: 'shard-1', data: {} },
  { name: 'shard-2', data: {} },
], connection);

// chord: group then callback
await chord('tasks', [
  { name: 'task-1', data: {} },
  { name: 'task-2', data: {} },
], { name: 'aggregate', data: {} }, connection);

---

Step jobs

Multi-step state machines using job.moveToDelayed() with an optional step token:

const worker = new Worker('q', async (job) => {
  const step = job.data.__step ?? 'init';

  switch (step) {
    case 'init':
      await doInit(job.data);
      await job.moveToDelayed(Date.now(), 'process');
      return;
    case 'process':
      await doProcess(job.data);
      await job.moveToDelayed(Date.now(), 'finalize');
      return;
    case 'finalize':
      return doFinalize(job.data);
  }
}, { connection });

BullMQ's moveToDelayed has no step parameter.

---

addAndWait (request-reply)

Synchronous RPC pattern - enqueue a job and wait for its result:

const result = await queue.addAndWait('compute', { input: 42 }, {
  waitTimeout: 30_000,
});
console.log(result); // the job's return value

---

Pluggable serializers

Use MessagePack, Protobuf, or any custom format instead of JSON:

import msgpack from 'msgpack-lite';

const queue = new Queue('tasks', {
  connection,
  serializer: {
    serialize: (data) => msgpack.encode(data),
    deserialize: (buffer) => msgpack.decode(buffer),
  },
});

---

Job TTL

Auto-expire jobs after a given duration:

await queue.add('ephemeral', data, {
  ttl: 60_000,  // job fails if not completed within 60 seconds
});

---

repeatAfterComplete

Scheduler mode that enqueues the next job only after the previous one completes, guaranteeing no overlap:

await queue.upsertJobScheduler(
  'sequential-poll',
  { repeatAfterComplete: 5000 },
  { name: 'poll', data: {} },
);

---

LIFO mode

Last-in-first-out processing - newest jobs are processed first:

await queue.add('urgent', data, { lifo: true });

Priority and delayed jobs take precedence over LIFO. Cannot be combined with ordering keys.

Note: LIFO + globalConcurrency has a crash limitation. If a worker is killed hard (SIGKILL, OOM) while processing a LIFO job, the list-active counter is not decremented. Reset with: DEL glide:{queueName}:list-active.

---

Job search

Search over job data fields:

const results = await queue.searchJobs({
  // search options
});

---

excludeData

Lightweight job listings without payload data:

const jobs = await queue.getJobs('waiting', 0, 99, { excludeData: true });
// jobs[0].data is undefined - useful for dashboard listings of large-payload queues

---

globalConcurrency on WorkerOptions

Set queue-wide concurrency cap at worker startup (shorthand for queue.setGlobalConcurrency()):

const worker = new Worker('q', processor, {
  connection,
  concurrency: 10,
  globalConcurrency: 50,  // queue-wide cap across all workers
});

---

Deduplication modes

Beyond BullMQ's simple deduplication, glide-mq adds explicit modes:

await queue.add('job', data, {
  deduplication: {
    id: 'my-dedup-key',
    ttl: 60_000,
    mode: 'simple',    // drop if exists (default)
    // mode: 'throttle' - drop duplicates within window
    // mode: 'debounce' - reset window on each add
  },
});

---

Backoff jitter

Spread retries under load with a jitter field:

await queue.add('job', data, {
  attempts: 5,
  backoff: { type: 'exponential', delay: 1000, jitter: 0.25 }, // +/- 25% random jitter
});

---

AI-Native Primitives

The following features are purpose-built for LLM/AI orchestration pipelines. None of them exist in BullMQ.

Usage Metadata (job.reportUsage)

Track model, tokens, cost, and latency per job. Persisted to the job hash and emitted as a 'usage' event.

const worker = new Worker('inference', async (job) => {
  const result = await callLLM(job.data);
  await job.reportUsage({
    model: 'gpt-5.4',
    provider: 'openai',
    tokens: { input: result.promptTokens, output: result.completionTokens },
    costs: { total: 0.003 },
    costUnit: 'usd',
    latencyMs: 800,
  });
  return result.content;
}, { connection });

Token Streaming (job.stream / queue.readStream)

Stream LLM output tokens in real-time via per-job Valkey Streams.

// Worker side
const worker = new Worker('chat', async (job) => {
  for await (const chunk of llmStream) {
    await job.stream({ token: chunk.text });
  }
  return { done: true };
}, { connection });

// Consumer side
const entries = await queue.readStream(jobId, { block: 5000 });

Suspend / Resume (Human-in-the-Loop)

Pause a job to wait for external approval, then resume with signals.

// Suspend in processor
await job.suspend({ reason: 'Needs review', timeout: 86_400_000 });

// Resume externally
await queue.signal(jobId, 'approve', { reviewer: 'alice' });

// On resume, job.signals contains all received signals

Budget Middleware (Flow-Level Caps)

Cap total tokens and/or cost across all jobs in a flow.

await flow.add(flowTree, {
  budget: { maxTotalTokens: 50_000, maxTotalCost: 0.50, costUnit: 'usd', onExceeded: 'fail' },
});

const budget = await queue.getFlowBudget(parentJobId);

Fallback Chains

Ordered model/provider alternatives tried on retryable failure.

await queue.add('inference', { prompt: '...' }, {
  attempts: 4,
  fallbacks: [
    { model: 'gpt-5.4', provider: 'openai' },
    { model: 'claude-sonnet-4-20250514', provider: 'anthropic' },
    { model: 'llama-3-70b', provider: 'groq' },
  ],
});

// Worker reads job.currentFallback for the active model/provider

Dual-Axis Rate Limiting (RPM + TPM)

Rate-limit by both requests and tokens per minute for LLM API compliance.

const worker = new Worker('inference', processor, {
  connection,
  limiter: { max: 60, duration: 60_000 },           // RPM
  tokenLimiter: { maxTokens: 100_000, duration: 60_000 },  // TPM
});

// Report tokens in processor
await job.reportTokens(totalTokens);

Flow Usage Aggregation

Aggregate AI usage across all jobs in a flow.

const usage = await queue.getFlowUsage(parentJobId);
// { tokens, totalTokens, costs, totalCost, costUnit, jobCount, models }

Vector Search (Valkey Search)

Create search indexes and run KNN vector similarity queries over job hashes.

await queue.createJobIndex({
  vectorField: { name: 'embedding', dimensions: 1536 },
});

const job = await queue.add('document', { text: 'Hello world' });
if (job) {
  await job.storeVector('embedding', queryEmbedding);
}

const results = await queue.vectorSearch(queryEmbedding, {
  k: 10,
  filter: '@state:{completed}',
});
// results: { job, score }[]

await queue.dropJobIndex();

Requires valkey-search module on the server (standalone mode).