Job Queues
Record slow work as a job, answer the request now, and let a separate worker process do it with retries.
await queue.add('send-welcome', { userId }, { attempts: 3, backoff: { type: 'exponential', delay: 1000 } })Some work does not belong inside a request. Resizing an upload, sending an email, building a month-end report: the caller waits, the connection times out, and their retry starts the work a second time. A job queue lets the handler write down what needs doing, answer straight away, and leave the doing to something else.
Three parts
The producer is your request handler. It adds a job, which is a small message naming the work and its inputs, then returns 202 Accepted. The broker is the store that holds jobs until someone takes them: Redis, Amazon SQS, or a table in your own database. The consumer, almost always called a worker, is a separate process that pulls jobs and runs them. A process is one running copy of a program, so the worker is started by its own command and needs its own copy of the environment variables, the named values the operating system hands a process, to reach the same database.
What survives a restart
Only what the broker wrote down. An array inside your web process is not a queue: restart and the jobs are gone. A broker keeps the job, hands it to another worker when yours dies, retries a failure after a growing delay, which is called backoff, and after the final attempt moves it to a dead-letter queue, a side list of jobs that never succeeded so that someone can look at them.
Keep payloads small: ids, not file contents. The job may run minutes later, so it should re-read the row. Note that { userId } is shorthand for { userId: userId } and the two are the same object.
Example
When to use it
- Send a welcome email without making signup wait for the mail server.
- Generate a PDF invoice and tell the browser to poll for it.
- Resize an uploaded image into four sizes after returning the upload id.
- Retry a flaky third-party call with backoff instead of failing the request.
More examples
Producer and worker with BullMQ
Three files and two running processes, which is the whole shape of a queued system. The handler returns 202 because the work has been accepted rather than finished. The worker is deployed and scaled on its own, and concurrency decides how many jobs it runs at once.
// queue.js - shared by both processes
import { Queue } from 'bullmq';
export const emails = new Queue('emails', {
connection: { host: process.env.REDIS_HOST, port: Number(process.env.REDIS_PORT) },
});
// server.js - the request handler only records the work
import { emails } from './queue.js';
export async function signup(req, res) {
const user = await createUser(req.body);
await emails.add(
'send-welcome',
{ userId: user.id }, // ids, not the body
{ attempts: 3, backoff: { type: 'exponential', delay: 1000 } }
);
res.writeHead(202, { 'content-type': 'application/json' });
res.end(JSON.stringify({ id: user.id })); // answer now
}
// worker.js - a SEPARATE process, started with: node worker.js
import { Worker } from 'bullmq';
new Worker(
'emails',
async (job) => {
const user = await loadUser(job.data.userId); // re-read: the row may have changed
await sendWelcome(user);
},
{ connection: { host: process.env.REDIS_HOST, port: Number(process.env.REDIS_PORT) }, concurrency: 5 }
);Your own database as the broker
Jobs in the same database as your data means one backup, one transaction, and no new service to run. FOR UPDATE SKIP LOCKED is what makes it safe for several workers. Worth knowing before you add a broker for a few thousand jobs a day.
-- A jobs table is enough when you already run Postgres and do not want Redis.
CREATE TABLE jobs (
id bigserial PRIMARY KEY,
name text NOT NULL,
payload jsonb NOT NULL,
run_after timestamptz NOT NULL DEFAULT now(),
attempts int NOT NULL DEFAULT 0,
locked_at timestamptz
);
-- One worker claims one job. SKIP LOCKED makes a second worker take the
-- next row instead of waiting, so workers never hand each other the same job.
WITH next AS (
SELECT id FROM jobs
WHERE locked_at IS NULL AND run_after <= now()
ORDER BY id
FOR UPDATE SKIP LOCKED
LIMIT 1
)
UPDATE jobs SET locked_at = now(), attempts = attempts + 1
WHERE id IN (SELECT id FROM next)
RETURNING id, name, payload;Large input stays out of the job
Brokers are built for small messages; many cap them in the low hundreds of kilobytes, and a base64 payload is a third larger than the file. Passing a storage key keeps the queue fast and lets a retry re-read the same input.
// Wrong: the whole upload travels through the broker and sits in memory.
await uploads.add('resize', { filename, bytes: buffer.toString('base64') });
// Right: store the bytes where they belong and pass a reference.
const objectKey = `uploads/${user.id}/${filename}`;
await saveToObjectStorage(objectKey, buffer);
await uploads.add('resize', { objectKey, userId: user.id, width: 800 });
// The worker fetches the bytes itself when it gets to the job.
Discussion