Part 2 of 10: Reliable background jobs in Node.js, from zero.
By Sandro Dezerio, senior full-stack engineer.
The signup scenario is made up, the same one as in Part 1. The timings and the provider limit come from a simulation with stated assumptions, not from a real email provider.
The queue from Part 1 sends one email at a time. When 20 signups arrive together, the last welcome email leaves six seconds later.
The obvious fix is to send them all at once. That works until the other side says no.
This post adds job queue concurrency to the queue: a setting for how many jobs may run at the same time. It also shows how to choose that number, and how to shut the queue down without losing work that is already running.
Why one at a time is slow
I will keep the made-up assumptions from Part 1. Each email call takes 300 ms. This time 20 welcome emails arrive at the same moment, and nothing fails.
One at a time, that is 20 x 300 ms, or 6.0 seconds. For most of that time, the server is only waiting for the provider to answer. The queue is safe, but it wastes the wait.
Why all at once is worse
Now say the email provider accepts at most 5 calls in flight at the same time. A sixth call is rejected right away with a “429, too many concurrent requests” error. This limit is made up, but real services have limits like it.
If the queue starts all 20 emails together, everything finishes in 0.3 seconds. It also loses 15 of the 20 emails because the provider rejected them.
All at once looks like the fastest option, and it loses the most work. So the goal is not raw speed. It is the fastest rate the other side accepts.
Picture a small app that sends a welcome email on signup. A newsletter mentions it, and 200 people sign up within a few minutes. With one worker and 300 ms per email, the last person waits a full minute. So someone changes the limit to “as many as possible”. Now the provider answers with 429 errors, and with no retries the rejected emails are simply lost. Both versions of the queue fail, in opposite ways. This scenario is made up, like the rest of this post.
The change: a job queue concurrency limit
Here is the queue again. The rest of the class works as in Part 1, but the worker loop changed, so I show the whole file once.
/**
* Part 2: the same queue, now with a concurrency limit and a clean shutdown.
* Still in memory, still no retries. Those are Part 3 and Part 6.
*/
export type Job<T> = { id: number; data: T };
export type Handler<T> = (job: Job<T>) => Promise<void>;
export class SimpleQueue<T> {
private jobs: Job<T>[] = [];
private nextId = 1;
private active = 0; // jobs running right now
private closing = false;
private idleWaiters: Array<() => void> = [];
readonly failed: Array<{ job: Job<T>; error: unknown }> = [];
constructor(
private handler: Handler<T>,
private concurrency = 1, // how many jobs may run at the same time
) {}
/** Put a job in line and return right away. The work happens later. */
add(data: T): number {
if (this.closing) throw new Error("queue is closing");
const job = { id: this.nextId++, data };
this.jobs.push(job);
this.pump();
return job.id;
}
/** Start jobs until the limit is reached or the line is empty. */
private pump(): void {
while (!this.closing && this.active < this.concurrency && this.jobs.length > 0) {
const job = this.jobs.shift()!;
this.active++;
void this.run(job);
}
}
private async run(job: Job<T>): Promise<void> {
try {
await this.handler(job);
} catch (error) {
this.failed.push({ job, error }); // remember it, keep going
}
this.active--;
this.pump(); // a slot is free: start the next job
this.notifyIfIdle();
}
private notifyIfIdle(): void {
if (this.active === 0 && (this.closing || this.jobs.length === 0)) {
this.idleWaiters.splice(0).forEach((resolve) => resolve());
}
}
/** Resolves when every job added so far has been handled. */
onIdle(): Promise<void> {
if (this.active === 0 && this.jobs.length === 0) return Promise.resolve();
return new Promise((resolve) => this.idleWaiters.push(resolve));
}
/**
* Graceful shutdown: start nothing new, let running jobs finish,
* and return the jobs that never started so the caller can decide what to do.
*/
async close(): Promise<Job<T>[]> {
this.closing = true;
if (this.active > 0) {
await new Promise<void>((resolve) => this.idleWaiters.push(resolve));
}
return this.jobs.splice(0);
}
}There are three ideas in it:
activecounts the jobs running right now, andconcurrencyis the limit. The default is 1, so the queue behaves like Part 1 unless you ask for more.pumpstarts jobs until the limit is reached or the line is empty.- When a job ends, whether it worked or failed, it frees a slot and calls
pumpagain.
What the limit does
The simulation below runs the same burst of 20 emails with three limits. Every number depends on the made-up assumptions above.
| Concurrency | Time to finish 20 emails | Emails sent | Rejected by the provider |
|---|---|---|---|
| 1 (Part 1) | 6.0 s | 20 | 0 |
| 5 | 1.2 s | 20 | 0 |
| 20 | 0.3 s | 5 | 15 |
A limit of 5 is five times faster than one at a time and loses nothing. A limit of 20 is the fastest and loses 75% of the emails.
How to choose the number
Start from the limit of the thing you call, such as a provider or a database pool. Treat it as a ceiling, not as a target. If the provider accepts 5 calls at once, run 3 or 4, because other code may call the same provider too.
Then measure instead of guessing. Raise the number one step at a time and watch the error rate and the time to finish. Stop when errors appear or when the time stops improving.
This advice fits jobs that mostly wait for the network. Jobs that burn CPU are different, because Node.js runs your code on one event loop, and long CPU work blocks everything else. A higher limit does not help there.
You will meet the same setting in production libraries. A BullMQ worker has a concurrency option that defaults to 1 and helps with asynchronous work. Part 6 uses BullMQ, so this number will look familiar.
Shutting down without losing running work
Deployments restart your process. If jobs are running at that moment, you want them to finish, and you want to know which jobs never started.
close() does that. It stops starting new jobs, waits for the running ones, and returns the jobs that never started. After that, add throws.
In a test with a limit of 2 and 6 jobs, the 2 running jobs finish, and the other 4 come back to the caller.
Those 4 jobs still live only in memory. If the process exits, they are gone. Part 6 fixes that.
What this queue still does not do
- It never retries. The 15 rejected emails above are the kind of failure worth retrying later. The reason matters: a rate limit is worth retrying, but a mailbox that does not exist is not. Part 3 covers retries with backoff.
- It lives in memory. Part 6 moves the queue to Redis.
- It cannot run a job twice yet, but retries will change that. Part 5 covers idempotency.
Takeaways
- Running jobs one at a time is safe but slow, and running them all at once can overload what you depend on.
- Job queue concurrency controls the speed of a queue, so start from the limit of the dependency and leave some room below it.
- A queue that shuts down cleanly finishes what is running and tells you what it never started.
Try it
Save the queue from above as queue.ts. Next to it, save this as simulate.ts. It uses a virtual clock, so it prints the same numbers every time:
import { mock } from "node:test";
import { SimpleQueue } from "./queue.ts";
// Illustrative assumptions (not a real provider):
const EMAIL_MS = 300; // each email call takes 300 ms
const PROVIDER_LIMIT = 5; // the provider rejects calls when more than 5 are in flight
const BURST = 20; // 20 welcome emails arrive at the same moment
mock.timers.enable({ apis: ["setTimeout"] });
const flush = () => new Promise<void>((resolve) => setImmediate(resolve));
let now = 0;
async function advanceUntil(done: () => boolean) {
await flush();
while (!done()) {
mock.timers.tick(10);
now += 10;
await flush();
}
}
const sleep = (ms: number) => new Promise<void>((resolve) => setTimeout(resolve, ms));
async function run(concurrency: number) {
now = 0;
let inFlight = 0;
let sent = 0;
const queue = new SimpleQueue<number>(async () => {
inFlight++;
const rejected = inFlight > PROVIDER_LIMIT;
try {
if (rejected) throw new Error("429 too many concurrent requests");
await sleep(EMAIL_MS);
sent++;
} finally {
inFlight--;
}
}, concurrency);
for (let i = 0; i < BURST; i++) queue.add(i);
let idle = false;
void queue.onIdle().then(() => (idle = true));
await advanceUntil(() => idle);
return { concurrency, seconds: now / 1000, sent, failed: queue.failed.length };
}
async function main() {
const rows = [];
for (const c of [1, 5, 20]) rows.push(await run(c));
console.table(rows);
mock.timers.reset();
}
main();Run it with Node.js 22, the version I tested:
pnpm dlx tsx simulate.tsNode prints a warning that its mock timers API is experimental. You can ignore it. You should see this table:
┌─────────┬─────────────┬─────────┬──────┬────────┐
│ (index) │ concurrency │ seconds │ sent │ failed │
├─────────┼─────────────┼─────────┼──────┼────────┤
│ 0 │ 1 │ 6 │ 20 │ 0 │
│ 1 │ 5 │ 1.2 │ 20 │ 0 │
│ 2 │ 20 │ 0.3 │ 5 │ 15 │
└─────────┴─────────────┴─────────┴──────┴────────┘To see the ceiling move, change PROVIDER_LIMIT to 8 and the list [1, 5, 20] to [1, 5, 8, 20], then run it again. A limit of 8 finishes in 0.9 seconds and loses nothing, while 20 now loses 12 emails.
Next: Part 3, “How to retry failed jobs in Node.js without making it worse”
Leave a Reply