When the Queue Is the Problem
The four ways queues fail — a backlog that cannot drain, a crash loop, work lost to a restart, and having to choose what not to do.
process.on("SIGTERM", drain) // stop receiving, finish in-flight work, then exitQueues fail in a small number of recognisable ways, and the right response to each is different.
A backlog that will never drain
If arrivals exceed what consumers can serve, and that is the steady state rather than a spike, patience does not help. The options are more consumers, a faster handler, or less work: drop what has expired, stop accepting at the edge, or degrade the job. Messages that carry a deadline let a worker discard what is already useless instead of paying full price for it.
A consumer in a crash loop
A handler that kills the process never acknowledges anything, so every message it had leased is redelivered, including the one that killed it. Throughput goes to zero while the queue looks busy and the pods look healthy between restarts. Bound the receives with a dead-letter queue, catch failures per message rather than per process, and keep a way to pause one queue's consumption without a deploy.
Work lost to a restart
Deployments, spot reclaims and OOM kills happen mid-job. On SIGTERM: stop receiving, let in-flight work finish, then exit — with a grace period longer than the longest job. Work is then redelivered rather than lost. The exception is a consumer that acknowledges before doing the work; that one loses jobs silently on every restart.
When you cannot do everything
Separate queues per priority beat a priority field in the body, because a single queue is consumed in roughly arrival order and the urgent job waits behind the bulk import.
Example
let draining = false;
const inFlight = new Set();
// SIGTERM is the deploy, the spot reclaim and the scale-in. The contract: stop
// receiving new work, finish what is held, then exit.
process.on("SIGTERM", async () => {
draining = true;
log.info({ inFlight: inFlight.size }, "draining");
await Promise.allSettled([...inFlight]); // [...set] is shorthand for Array.from(set)
process.exit(0);
});
export async function consume() {
while (!draining) {
const message = await receiveOne(); // 20s long poll
if (!message) continue;
const work = (async () => {
try {
await processJob(JSON.parse(message.Body));
// Delete last. Deleting on receive turns every restart into lost work.
await sqs.send(new DeleteMessageCommand({
QueueUrl,
ReceiptHandle: message.ReceiptHandle,
}));
} catch (err) {
// Catch per message. An uncaught throw here takes the process down and
// redelivers every message this worker was holding.
log.error({ err, messageId: message.MessageId }, "job failed");
} finally {
inFlight.delete(work);
}
})();
inFlight.add(work);
}
}
// In Kubernetes, terminationGracePeriodSeconds must exceed the longest job, or the
// kubelet follows SIGTERM with SIGKILL mid-work. The message is then redelivered
// once its visibility timeout expires - survivable, but only because the delete had
// not happened yet.When to use it
- Decide between adding consumers and shedding load when a backlog is growing faster than it drains.
- Stop a deploy from losing half-finished jobs on every rolling restart.
- Explain zero throughput on a queue whose depth is rising and whose pods look healthy.
- Keep interactive jobs responsive while a 200,000-row import is running.
More examples
A deadline in the message
The cheapest form of load shedding, and the only one that gets faster the worse things are. It also prevents the second-order failure where a two-hour backlog of notifications is finally delivered and users get a burst of stale alerts. Add notAfter at the producer wherever the work has a natural expiry; the consumer change is five lines and needs no coordination.
// Shedding starts with knowing what is already pointless. A push notification for
// a match that finished an hour ago should be dropped, not delivered late.
const job = JSON.parse(message.Body);
if (job.notAfter && Date.parse(job.notAfter) < Date.now()) {
metrics.increment("jobs.expired", { kind: job.kind });
// Deleted, not failed. An expired job must not consume the dead-letter queue or
// page anybody - it did exactly what it was told to do.
await sqs.send(new DeleteMessageCommand({
QueueUrl,
ReceiptHandle: message.ReceiptHandle,
}));
return;
}
// Producers set notAfter from the meaning of the work, not from a global default:
// live-score push -> now + 2 minutes
// password reset mail -> now + 1 hour
// invoice PDF -> no deadline; it must happen eventually
// During a backlog, a queue of mostly expired work then drains at the speed of a
// JSON parse rather than the speed of the database.Pausing consumption without a deploy
The control you want to already have at 03:00, when the choice is between a pause and a rollback. The fail-open default in the switch matters: a kill switch whose own lookup failure stops all consumption is a new outage. The retention warning is the trap, because a paused queue looks like a safely preserved backlog right up to the moment the oldest messages age out.
import { Queue } from "bullmq";
// A crash loop or a dependency outage needs a stop button that is not a code
// change. In BullMQ, pausing is a queue operation: workers finish their current
// jobs and stop taking new ones, while producers keep adding.
const queue = new Queue("thumbnails", { connection });
await queue.pause();
// ... deploy the fix, let the dependency recover ...
await queue.resume();
// With SQS the equivalent is operational rather than an API call: scale the
// consumer deployment to zero, or read a kill switch between receives.
async function shouldConsume() {
const { rows: [flag] } = await pool.query(
"SELECT enabled FROM consumer_switches WHERE queue = $1",
["thumbnails"]
);
return flag?.enabled !== false; // fail open: a missing row must not stop work
}
// Either way, check the retention period before pausing for long. SQS keeps
// messages for 4 days by default and 14 at most; a pause longer than that discards
// the work silently.Priority by queue, not by field
The standard fix for "everything is slow because someone started an import", and the two comments are the parts that go wrong. Short-polling the high-priority queue keeps empty-receive costs down while still favouring it; dedicating some workers to each class is what stops strict priority turning into permanent starvation of the bulk queue. Two queues is almost always enough — a five-level scheme is five backlogs to monitor.
// A priority field in the body cannot help: a queue is consumed in roughly arrival
// order, so the interactive job sits behind 200,000 rows of CSV import. Separate
// queues, polled in order, can.
async function receiveByPriority() {
const queues = [INTERACTIVE_QUEUE_URL, BULK_QUEUE_URL];
for (const [i, QueueUrl] of queues.entries()) {
const isLast = i === queues.length - 1;
const out = await sqs.send(new ReceiveMessageCommand({
QueueUrl,
MaxNumberOfMessages: 10,
// Long poll only the last queue, or waiting 20s on an empty urgent queue
// starves the bulk one.
WaitTimeSeconds: isLast ? 20 : 0,
}));
if (out.Messages?.length) return { QueueUrl, messages: out.Messages };
}
return null;
}
// Strict priority can starve the bulk queue entirely, so reserve capacity as well
// as order: a worker pool dedicated to interactive work cannot be starved by a
// bulk backlog, and a pool dedicated to bulk work still makes progress during one.
Discussion