Move out of the HTTP request anything that is slow, depends on third-party services and is not needed for the response: emails and notifications, CRM calls and outgoing webhooks, reports and exports. The API puts a job on a queue and replies right away, while a separate worker process runs it and, on failure, retries with growing pauses. To keep a retry from sending a second message, make the handler idempotent, and collect jobs that ran out of attempts for separate review.
After this you will be able to
- decide which operations to move out of the HTTP request into a background queue and which to keep in the request
- enqueue a BullMQ job from an Express handler and process it in a worker with retries and exponential backoff
- protect processing from duplicates:
Idempotency-Keyin the API,jobIdon the queue and an "already sent" mark in the worker - find dead jobs with
getFailed()and put them back on the queue once the cause is fixed
Know beforehand
Verified with
Node.js 24 LTS (24.18)BullMQ 6.3ioredis 6.0Express 5.3Redis 7.4 (redis:7-alpine)TypeScript 7.0
Checked against docs: October 11, 2026
Contents
Why a long request is a problem
Take the leads API from the exercise in the Express REST API lesson: the POST /api/leads handler validates input, saves the lead and responds. The temptation is to add an email to the manager, a Telegram message and a CRM sync right there. Response time then becomes the sum of every external service's latency, and the form is only as reliable as the weakest of them.
- The user waits and clicks again, and you get a duplicate lead.
- Proxies, load balancers and the client all have timeouts: the request is cut off after part of the work is done, and nobody knows which part.
- An external service is down, so the form returns an error even though the lead is already in the database.
- Retries inside the request ("let's try three more times") multiply the wait instead of solving the problem.
Validating input, saving the lead
- Where to run it
- In the request
- Why
- You can't honestly say "lead received" without it
Email or message to a manager
- Where to run it
- In the background
- Why
- The user doesn't need to wait for delivery; the service may be down
CRM or ERP sync, outgoing webhook
- Where to run it
- In the background
- Why
- A third-party service with its own limits, outages and latency
PDF report, Excel export, image processing
- Where to run it
- In the background + job status
- Why
- Can take minutes and load the CPU
Payment confirmation
- Where to run it
- By the provider's rules
- Why
- The user needs the result, and the final status usually arrives via the provider's webhook
Queue, producer and worker
There are three roles. The producer is the code that puts a job on the queue; here it is the API handler. The queue stores jobs until they run; in BullMQ it is a set of data structures in Redis. The worker is a separate process that takes jobs one at a time or in batches and runs them. If the worker is stopped, jobs are not lost: they pile up and wait, as long as Redis persists data to disk (more on that below).
- 01Form
- 02API — producer
- 03Redis — queue
- 04Worker
- 05Notification service
A BullMQ job moves through states: waiting for a worker, active while running, then completed or failed. Between failed attempts it sits in delayed, waiting out the pause before the next retry.
The code below was checked on BullMQ 6.3. In version 6 the ioredis Redis client became an optional peer dependency: install it explicitly, or creating a queue fails with "BullMQ could not load the optional 'ioredis' package". For development, Redis is easiest to run in Docker.
npm init -y && npm pkg set type=module
npm install bullmq ioredis express
npm install -D typescript @types/express @types/node
docker run -d --rm --name kb-redis-jobs -p 6379:6379 redis:7-alpineNode.js 24 runs .ts files directly by stripping type annotations, which is why imports use the .ts extension. The compiler is only needed for checking: npx tsc --noEmit.
{
"compilerOptions": {
"target": "es2024",
"module": "nodenext",
"strict": true,
"noEmit": true,
"allowImportingTsExtensions": true,
"erasableSyntaxOnly": true,
"verbatimModuleSyntax": true,
"skipLibCheck": true
},
"include": ["src"]
}import { Redis } from 'ioredis';
const url = process.env.REDIS_URL ?? 'redis://localhost:6379';
// The producer runs inside an HTTP handler: if Redis is down,
// fail fast instead of holding the client's request.
export const producerConnection = () =>
new Redis(url, { maxRetriesPerRequest: 1, enableOfflineQueue: false });
// The worker must ride out disconnects: BullMQ requires null here.
export const workerConnection = () => new Redis(url, { maxRetriesPerRequest: null });import { Queue } from 'bullmq';
import type { Redis } from 'ioredis';
export type LeadNotification = { leadId: string; name: string; phone: string };
export const NOTIFICATIONS = 'notifications';
export function createNotificationsQueue(connection: Redis) {
return new Queue<LeadNotification>(NOTIFICATIONS, {
connection,
defaultJobOptions: {
attempts: 4, // first try + 3 retries
backoff: { type: 'exponential', delay: 1000 }, // waits 1 s, 2 s, 4 s
removeOnComplete: { count: 1000 }, // keep the last 1000 successful jobs
removeOnFail: { age: 7 * 24 * 3600 }, // keep failed jobs for a week
},
});
}The producer: the API enqueues a job and replies at once
The handler saves the lead (in process memory in this example, in a database table in a real project), enqueues a lead-created job and returns 201 without waiting for the notification.
import { randomUUID } from 'node:crypto';
import express from 'express';
import { createNotificationsQueue } from './queue.ts';
import { producerConnection } from './redis.ts';
type Lead = { id: string; name: string; phone: string; createdAt: string };
const leads = new Map<string, Lead>(); // a database table in a real project
const queue = createNotificationsQueue(producerConnection());
queue.on('error', (err) => console.error('[queue]', err.message || err.name));
// While Redis is down, queue.add waits for a reconnect: don't let the request hang.
async function withTimeout<T>(promise: Promise<T>, ms: number): Promise<T> {
let timer: NodeJS.Timeout | undefined;
const timeout = new Promise<never>((_, reject) => {
timer = setTimeout(() => reject(new Error(`timeout after ${ms} ms`)), ms);
});
promise.catch(() => {}); // a late rejection must not crash the process
try {
return await Promise.race([promise, timeout]);
} finally {
clearTimeout(timer);
}
}
const app = express();
app.use(express.json());
app.post('/api/leads', async (req, res) => {
const { name, phone } = req.body ?? {};
if (typeof name !== 'string' || typeof phone !== 'string') {
res.status(400).json({ error: 'name and phone are required' });
return;
}
// The client generates the key once per form submission:
// a repeated request creates neither a second lead nor a second job.
const id = req.get('Idempotency-Key') ?? randomUUID();
if (leads.has(id)) {
res.status(200).json({ id, duplicate: true });
return;
}
leads.set(id, { id, name, phone, createdAt: new Date().toISOString() });
try {
// jobId dedupes at the queue level: a job with an existing id is ignored
const job = queue.add('lead-created', { leadId: id, name, phone }, { jobId: `lead-${id}` });
await withTimeout(job, 2000);
} catch (err) {
// The lead is saved but the notification is not queued: make it visible in logs.
console.error('[api] enqueue failed', id, (err as Error).message);
}
res.status(201).json({ id }); // the response does not wait for the notification
});
const port = Number(process.env.PORT ?? 3000);
app.listen(port, () => console.log(`[api] http://localhost:${port}`));Idempotency-Keyis a key the client generates once per form submission. Repeating the same request (a double click, a retry after a dropped connection) returns the same lead instead of creating a new one.jobId: lead-<id>protects at the queue level: per the BullMQ docs, a job with an existing id is simply ignored. The id must not contain a colon and cannot be digits only.withTimeoutwas added after testing: while Redis is unavailable,queue.addwaits for a reconnect, and in our run the request hung for more than 30 seconds even withenableOfflineQueue: false. A 2-second limit gets the client an answer in time.
$ curl -s -X POST localhost:3000/api/leads … -w ' HTTP %{http_code}, %{time_total}s\n'
{"id":"ab6a5c91-26cb-4c2a-aac4-f5fe5842624c"} HTTP 201, 2.019723s
# api
[api] enqueue failed ab6a5c91-26cb-4c2a-aac4-f5fe5842624c timeout after 2000 msThe worker: retries, backoff and errors a retry won't fix
A worker's processor is an async function. If it returns a value, the job is completed; if it throws, the attempt has failed. With attempts: 4 and backoff: { type: 'exponential', delay: 1000 }, BullMQ waits 2^(n-1) × delay milliseconds before each retry: 1, 2 and 4 seconds. Without backoff the retry happens immediately, which is useless against a service that is down.
import { UnrecoverableError, Worker, type Job } from 'bullmq';
import { Redis } from 'ioredis';
import { NOTIFICATIONS, type LeadNotification } from './queue.ts';
import { workerConnection } from './redis.ts';
const NOTIFY_URL = process.env.NOTIFY_URL ?? 'http://localhost:4001/notify';
const store = new Redis(process.env.REDIS_URL ?? 'redis://localhost:6379'); // "already sent" marks
const log = (msg: string) => console.log(new Date().toISOString().slice(11, 19), msg);
async function notifyManager(job: Job<LeadNotification>): Promise<string> {
const doneKey = `lead-notified:${job.data.leadId}`;
// A retry after a successful send (e.g. the worker died before completing the job)
if (await store.exists(doneKey)) return 'already sent';
const res = await fetch(NOTIFY_URL, {
method: 'POST',
headers: { 'content-type': 'application/json' },
body: JSON.stringify({ text: `New lead: ${job.data.name}, ${job.data.phone}` }),
signal: AbortSignal.timeout(5000), // a hung request is an error, not an endless wait
});
// A 4xx (except 429) won't be fixed by retrying: fail now, skip the remaining attempts
if (res.status >= 400 && res.status < 500 && res.status !== 429) {
throw new UnrecoverableError(`notify rejected: HTTP ${res.status}`);
}
if (!res.ok) throw new Error(`notify unavailable: HTTP ${res.status}`);
await store.set(doneKey, '1', 'EX', 7 * 24 * 3600);
return 'sent';
}
const worker = new Worker<LeadNotification, string>(NOTIFICATIONS, notifyManager, {
connection: workerConnection(),
concurrency: 5,
});
worker.on('completed', (job, result) => {
log(`${job.id} ${result} (attempt ${job.attemptsMade})`);
});
worker.on('failed', async (job, err) => {
if (!job) return;
if ((await job.getState()) === 'failed') {
// Out of attempts or unrecoverable: a dead job that needs a human
log(`DEAD ${job.id} after ${job.attemptsMade} attempt(s): ${err.message}`);
} else {
log(`retry ${job.id} attempt ${job.attemptsMade} failed: ${err.message}`);
}
});
worker.on('error', (err) => console.error('[worker]', err.message));
async function shutdown() {
log('shutting down');
await worker.close(); // finish active jobs, take no new ones
await store.quit();
process.exit(0);
}
process.on('SIGINT', shutdown);
process.on('SIGTERM', shutdown);
log('waiting for jobs');2xx
- What the worker does
- Sets the mark, job is
completed - Why
- The notification was delivered
5xx, timeout, network error
- What the worker does
throw new Error→ retry after a pause- Why
- The failure is temporary and may clear in a few seconds
429 Too Many Requests
- What the worker does
- Retry after a pause
- Why
- Rate limit; better to honor
Retry-After(see the exercise)
Other 4xx
- What the worker does
UnrecoverableError→ straight tofailed- Why
- Bad data or a bad key: a retry gets the same answer
To test retries without a real Telegram bot or mail server, run a stand-in: it answers the first FAIL_FIRST requests with 503 and rejects a message without a phone number with 400.
// A local stand-in for the notification service (Telegram, email, CRM).
import { createServer } from 'node:http';
let failFirst = Number(process.env.FAIL_FIRST ?? 2); // how many first requests fail
createServer(async (req, res) => {
let body = '';
for await (const chunk of req) body += chunk;
if (failFirst > 0) {
failFirst--;
console.log('[notify] 503 Service Unavailable');
res.writeHead(503).end();
return;
}
const { text } = JSON.parse(body);
if (!/\d{10,}/.test(text)) {
console.log('[notify] 400 Bad Request');
res.writeHead(400).end();
return;
}
console.log(`[notify] 200 delivered: ${text}`);
res.writeHead(200).end();
}).listen(4001, () => console.log('[notify] http://localhost:4001'));FAIL_FIRST=2 node src/fake-notify.ts # terminal 1: the "notification service"
node src/worker.ts # terminal 2: the worker
node src/api.ts # terminal 3: the API on :3000$ curl -s -X POST localhost:3000/api/leads -H 'content-type: application/json' \
-H 'Idempotency-Key: 7f3c2a10' --data-binary @lead.json -w ' HTTP %{http_code}, %{time_total}s\n'
{"id":"7f3c2a10"} HTTP 201, 0.015741s
$ curl …the same command again…
{"id":"7f3c2a10","duplicate":true} HTTP 200, 0.004159s
# worker
23:24:53 waiting for jobs
23:24:55 retry lead-7f3c2a10 attempt 1 failed: notify unavailable: HTTP 503
23:24:56 retry lead-7f3c2a10 attempt 2 failed: notify unavailable: HTTP 503
23:24:58 lead-7f3c2a10 sent (attempt 3)
# fake-notify
[notify] 503 Service Unavailable
[notify] 503 Service Unavailable
[notify] 200 delivered: New lead: Aigerim, +77011234567The timestamps show the pauses: the first attempt at :55, the second a second later, the third two seconds after that. The notification service got one successful message even though the request arrived twice. Now a lead with an invalid phone:
$ curl -s -X POST localhost:3000/api/leads -H 'content-type: application/json' \
-H 'Idempotency-Key: 9b1e44d0' -d '{"name":"Test","phone":"none"}'
{"id":"9b1e44d0"}
# worker
23:25:03 DEAD lead-9b1e44d0 after 1 attempt(s): notify rejected: HTTP 400Idempotency: a retry must not send a second message
The BullMQ docs explicitly advise designing jobs with failure in mind: the outcome should be the same whether the job succeeded on the first attempt or after a retry. That is idempotency. Re-runs do happen: besides explicit retries, a job whose worker hung or crashed and failed to renew its lock is considered stalled and goes back to the queue.
- API level: the client's
Idempotency-Key, one lead per click. - Queue level:
jobId, one job per lead while the job is stored in Redis. Once the job is removed (removeOnComplete,removeOnFail), the same id can be added again. - Handler level: the
lead-notified:<id>mark, so a retry sends nothing if the notification already went out.
Let's check the third level: enqueue a job for an already-notified lead under a new jobId, as if another source had sent it.
$ node check/replay.ts # the same leadId under a new jobId
# worker
23:25:28 lead-7f3c2a10-replay already sent (attempt 1)
# fake-notify: no new requestsFor a different need, "no more than once every N seconds" (say, recomputing a summary on every new lead), BullMQ has a deduplication option with id and ttl: new jobs with the same id are ignored during the window.
Dead jobs and observability
When attempts run out or an UnrecoverableError is thrown, the job moves to the failed set. It plays the role of a dead letter queue in other brokers: these jobs won't run again on their own, so a person or a script has to step in. By default BullMQ keeps finished jobs forever, which is why removeOnComplete and removeOnFail in queue.ts are a must, or Redis will slowly fill up.
# fake-notify started with FAIL_FIRST=100: the service is "down"
23:25:09 retry lead-c5d2e9a1 attempt 1 failed: notify unavailable: HTTP 503
23:25:10 retry lead-c5d2e9a1 attempt 2 failed: notify unavailable: HTTP 503
23:25:12 retry lead-c5d2e9a1 attempt 3 failed: notify unavailable: HTTP 503
23:25:16 DEAD lead-c5d2e9a1 after 4 attempt(s): notify unavailable: HTTP 503
$ node src/dead-jobs.ts
lead-c5d2e9a1 attempts=4 reason="notify unavailable: HTTP 503"
lead-9b1e44d0 attempts=1 reason="notify rejected: HTTP 400"
failed: 2import { createNotificationsQueue } from './queue.ts';
import { producerConnection } from './redis.ts';
const connection = producerConnection();
const queue = createNotificationsQueue(connection);
const failed = await queue.getFailed();
for (const job of failed) {
console.log(`${job.id} attempts=${job.attemptsMade} reason="${job.failedReason}"`);
if (process.argv.includes('--retry')) await job.retry(); // back to the queue after a fix
}
console.log(`failed: ${failed.length}`);
await queue.close();
await connection.quit(); // we created this connection, so we close it# the service is back (FAIL_FIRST=0)
$ node src/dead-jobs.ts --retry
lead-c5d2e9a1 attempts=4 reason="notify unavailable: HTTP 503"
lead-9b1e44d0 attempts=1 reason="notify rejected: HTTP 400"
failed: 2
# worker
23:25:21 lead-c5d2e9a1 sent (attempt 5)
23:25:21 DEAD lead-9b1e44d0 after 2 attempt(s): notify rejected: HTTP 400- Redis runs with
maxmemory-policy noeviction: if Redis evicts keys, the queue breaks (this is the default in theredis:7-alpineimage, checked withCONFIG GET). - Persistence is on:
appendonly yeswithappendfsync everysec; per the Redis docs, a crash loses at most about a second of writes. errorhandlers are attached to the queue and the worker, otherwise connection errors go unhandled.- Graceful shutdown on
SIGINTandSIGTERM:worker.close()waits for active jobs. In a container,docker stopended our worker with exit code 0. - Auto-removal via
removeOnCompleteandremoveOnFailis configured. - Job
dataholds no personal data: BullMQ stores it in Redis as plain text. The example puts the name and phone in the job for clarity; in a real project pass onlyleadIdand let the worker read the rest from the database. - The number of
failedjobs (queue.getFailedCount()) is exported to metrics, with an alert on growth.
$ docker stop <container with node src/worker.ts>
23:25:56 waiting for jobs
23:25:57 shutting down
container exit code: 0Delayed and repeatable jobs; alternatives
A delayed job is a regular job with the delay option: it sits in delayed and becomes available to a worker once the pause is over. Repeatable jobs in BullMQ 6 are created through Job Schedulers: upsertJobScheduler creates a schedule or updates an existing one with the same id. The old way, the repeat option of queue.add, was removed in version 6, as the v5 migration guide describes.
import { Queue } from 'bullmq';
import { producerConnection } from './redis.ts';
const connection = producerConnection();
const reports = new Queue('reports', { connection });
// Repeatable job: a lead summary every day at 09:00 Almaty time.
// upsert: running this on every deploy updates the schedule instead of duplicating it.
await reports.upsertJobScheduler(
'daily-leads-report',
{ pattern: '0 0 9 * * *', tz: 'Asia/Almaty' },
{ name: 'daily-leads-report', data: {} },
);
// Delayed job: remind the manager if nobody picks up the lead within 30 minutes.
await reports.add('lead-reminder', { leadId: '7f3c2a10' }, { delay: 30 * 60 * 1000 });
for (const s of await reports.getJobSchedulers()) {
console.log(s.key, s.pattern, s.tz, 'next:', new Date(s.next ?? 0).toISOString());
}
console.log(await reports.getJobCounts('delayed'));
await reports.close();
await connection.quit();$ node src/schedule.ts
daily-leads-report 0 0 9 * * * Asia/Almaty next: 2026-10-12T04:00:00.000Z
{ delayed: 2 }The cron expression has six fields here; the first one is seconds. The schedule only enqueues the summary; a worker for the reports queue runs it and is written just like the notification worker.
Cron (system cron, a Kubernetes CronJob, the hosting scheduler)
- When it fits
- A periodic script: a nightly export, a cleanup
- Limitations
- No job queue, retries or failure tracking; you write all of that yourself
Managed cloud queues (Amazon SQS, Google Cloud Tasks and the like)
- When it fits
- You'd rather not run your own Redis; your infrastructure already lives in that cloud
- Limitations
- Vendor lock-in, plus their own limits and delivery guarantees: read their docs
BullMQ with the PostgreSQL backend
- When it fits
- The project has no Redis and you'd rather not add one
- Limitations
- Per the BullMQ docs, throughput is lower; Redis remains the default and most battle-tested option
setTimeout or "fire without await" inside the API process
- When it fits
- Experiments only
- Limitations
- The job is lost on restart; no retries, no tracking
Exercises
How this was verified3 · passed1 · partialchecked 11 October 2026
“Run” means the code was executed in a scratch project or container and the output in the article comes from that run. “Partial” means it was checked against stubs instead of external services. “Not run” means it needs a device, an account or a key; the steps were checked against the official documentation. The verification project files and run logs are kept in the InterLAB archive.