Building a Task Queue From Scratch With Node.js and Redis
Build a production-ready task queue step by step using Node.js and Redis, covering reliable delivery, retries, dead letter queues, and concurrency control.
Every application eventually needs background processing. Email notifications, image resizing, report generation, webhook delivery—these tasks don't belong in your request-response cycle. A task queue decouples producers from consumers, enabling reliable asynchronous processing.
While libraries like BullMQ exist, building a queue from scratch teaches you the primitives that make distributed job processing work. You'll understand exactly what happens when a job fails, how concurrency is managed, and why certain design decisions matter.
The Core Queue Structure
A task queue needs three things: a way to enqueue jobs, a way to dequeue them reliably, and a way to track their state. Redis provides all the building blocks.
import { createClient, RedisClientType } from "redis";
interface Job<T = unknown> {
id: string;
queue: string;
payload: T;
attempts: number;
maxAttempts: number;
createdAt: number;
processAfter: number;
}
class TaskQueue {
private client: RedisClientType;
private prefix: string;
constructor(client: RedisClientType, prefix: string = "tq") {
this.client = client;
this.prefix = prefix;
}
private key(queue: string, suffix: string): string {
return `${this.prefix}:${queue}:${suffix}`;
}
async enqueue<T>(
queue: string,
payload: T,
options: { delay?: number; maxAttempts?: number } = {}
): Promise<string> {
const id = crypto.randomUUID();
const now = Date.now();
const job: Job<T> = {
id,
queue,
payload,
attempts: 0,
maxAttempts: options.maxAttempts ?? 3,
createdAt: now,
processAfter: now + (options.delay ?? 0),
};
const multi = this.client.multi();
// Store job data
multi.set(
this.key(queue, `job:${id}`),
JSON.stringify(job)
);
// Add to waiting sorted set (scored by processAfter)
multi.zAdd(this.key(queue, "waiting"), {
score: job.processAfter,
value: id,
});
await multi.exec();
return id;
}
}Using a sorted set for the waiting queue enables delayed jobs naturally—jobs are scored by their processAfter timestamp, so we only pick up jobs whose score is at or before the current time.
Reliable Job Consumption
The critical challenge in any queue is ensuring exactly-once processing. If a worker crashes mid-job, the job must not be lost. Redis's ZPOPMIN combined with a processing set provides this guarantee.
// ❌ Unreliable: job lost if worker crashes after pop
async function unsafeDequeue(client: RedisClientType, queue: string) {
const result = await client.zPopMin(queue);
// If process crashes here, job is gone forever
return result;
}// ✅ Reliable: job tracked in processing set
class TaskQueue {
// ... previous code
async dequeue(queue: string): Promise<Job | null> {
const now = Date.now();
const waitingKey = this.key(queue, "waiting");
const processingKey = this.key(queue, "processing");
// Atomically move job from waiting to processing
const script = `
local result = redis.call('ZRANGEBYSCORE', KEYS[1], '-inf', ARGV[1], 'LIMIT', 0, 1)
if #result == 0 then return nil end
local jobId = result[1]
redis.call('ZREM', KEYS[1], jobId)
redis.call('ZADD', KEYS[2], ARGV[2], jobId)
return jobId
`;
const jobId = await this.client.eval(script, {
keys: [waitingKey, processingKey],
arguments: [String(now), String(now)],
});
if (!jobId) return null;
const jobData = await this.client.get(
this.key(queue, `job:${jobId}`)
);
if (!jobData) return null;
return JSON.parse(jobData) as Job;
}
async acknowledge(queue: string, jobId: string): Promise<void> {
const multi = this.client.multi();
multi.zRem(this.key(queue, "processing"), jobId);
multi.del(this.key(queue, `job:${jobId}`));
multi.incr(this.key(queue, "stats:completed"));
await multi.exec();
}
}The Lua script runs atomically in Redis—no other client can interfere between checking the waiting set and moving the job to processing. This eliminates the race condition where two workers could grab the same job.
Retry Logic and Dead Letter Queues
Jobs fail. Networks time out, external APIs return errors, data is malformed. A robust queue retries transient failures and routes persistent failures to a dead letter queue for investigation.
class TaskQueue {
// ... previous code
async fail(
queue: string,
jobId: string,
error: string
): Promise<"retried" | "dead-lettered"> {
const jobData = await this.client.get(
this.key(queue, `job:${jobId}`)
);
if (!jobData) throw new Error(`Job ${jobId} not found`);
const job: Job = JSON.parse(jobData);
job.attempts += 1;
// Remove from processing set
await this.client.zRem(this.key(queue, "processing"), jobId);
if (job.attempts < job.maxAttempts) {
// Exponential backoff: 1s, 4s, 9s, 16s...
const delay = Math.pow(job.attempts, 2) * 1000;
job.processAfter = Date.now() + delay;
const multi = this.client.multi();
multi.set(
this.key(queue, `job:${jobId}`),
JSON.stringify(job)
);
multi.zAdd(this.key(queue, "waiting"), {
score: job.processAfter,
value: jobId,
});
await multi.exec();
return "retried";
}
// Max attempts exceeded: dead letter queue
const multi = this.client.multi();
multi.lPush(
this.key(queue, "dead"),
JSON.stringify({ ...job, error, failedAt: Date.now() })
);
multi.del(this.key(queue, `job:${jobId}`));
multi.incr(this.key(queue, "stats:dead-lettered"));
await multi.exec();
return "dead-lettered";
}
}Exponential backoff with the formula attempts² × 1000ms gives increasingly longer delays: 1 second, 4 seconds, 9 seconds. This prevents retry storms when a downstream service is struggling and gives it time to recover.
Building the Worker Loop
A worker continuously polls for jobs, processes them, and handles success or failure. The polling interval balances responsiveness against Redis load.
type JobHandler<T = unknown> = (payload: T) => Promise<void>;
class Worker {
private queue: TaskQueue;
private queueName: string;
private handler: JobHandler;
private running: boolean = false;
private concurrency: number;
private activeJobs: number = 0;
constructor(
queue: TaskQueue,
queueName: string,
handler: JobHandler,
concurrency: number = 5
) {
this.queue = queue;
this.queueName = queueName;
this.handler = handler;
this.concurrency = concurrency;
}
async start(): Promise<void> {
this.running = true;
console.log(
`Worker started for queue "${this.queueName}" ` +
`(concurrency: ${this.concurrency})`
);
while (this.running) {
if (this.activeJobs >= this.concurrency) {
await this.sleep(100);
continue;
}
const job = await this.queue.dequeue(this.queueName);
if (!job) {
await this.sleep(1000); // No jobs available, wait
continue;
}
this.activeJobs++;
this.processJob(job).finally(() => {
this.activeJobs--;
});
}
}
private async processJob(job: Job): Promise<void> {
try {
await this.handler(job.payload);
await this.queue.acknowledge(this.queueName, job.id);
} catch (error) {
const message =
error instanceof Error ? error.message : "Unknown error";
const result = await this.queue.fail(
this.queueName,
job.id,
message
);
console.warn(
`Job ${job.id} failed (${result}): ${message}`
);
}
}
stop(): void {
this.running = false;
}
private sleep(ms: number): Promise<void> {
return new Promise(resolve => setTimeout(resolve, ms));
}
}The concurrency control ensures the worker processes multiple jobs simultaneously without overwhelming the system. Each job runs independently—a slow job doesn't block others.
Stalled Job Recovery
If a worker crashes, its jobs remain in the processing set indefinitely. A separate recovery process must detect stalled jobs and re-enqueue them.
class TaskQueue {
// ... previous code
async recoverStalledJobs(
queue: string,
stallTimeout: number = 30000
): Promise<number> {
const processingKey = this.key(queue, "processing");
const cutoff = Date.now() - stallTimeout;
// Find jobs that have been processing longer than stallTimeout
const stalledIds = await this.client.zRangeByScore(
processingKey,
"-inf",
String(cutoff)
);
let recovered = 0;
for (const jobId of stalledIds) {
const jobData = await this.client.get(
this.key(queue, `job:${jobId}`)
);
if (!jobData) {
// Job data missing, just clean up
await this.client.zRem(processingKey, jobId);
continue;
}
const job: Job = JSON.parse(jobData);
job.attempts += 1;
if (job.attempts >= job.maxAttempts) {
// Exceeded retries, dead letter it
const multi = this.client.multi();
multi.zRem(processingKey, jobId);
multi.lPush(
this.key(queue, "dead"),
JSON.stringify({
...job,
error: "Stalled and exceeded max attempts",
failedAt: Date.now(),
})
);
multi.del(this.key(queue, `job:${jobId}`));
await multi.exec();
} else {
// Re-enqueue for retry
const multi = this.client.multi();
multi.zRem(processingKey, jobId);
multi.set(
this.key(queue, `job:${jobId}`),
JSON.stringify(job)
);
multi.zAdd(this.key(queue, "waiting"), {
score: Date.now(),
value: jobId,
});
await multi.exec();
recovered++;
}
}
return recovered;
}
}Run the recovery sweep on a timer—every 30 seconds is reasonable for most applications. The stall timeout should be longer than your longest expected job duration to avoid recovering jobs that are still actively processing.
Putting It All Together
Here's how producers and consumers interact through the queue in a real application scenario.
async function main() {
const redis = createClient({ url: "redis://localhost:6379" });
await redis.connect();
const taskQueue = new TaskQueue(redis);
// Producer: enqueue email jobs
await taskQueue.enqueue("emails", {
to: "user@example.com",
subject: "Welcome",
template: "onboarding",
});
await taskQueue.enqueue(
"emails",
{
to: "admin@example.com",
subject: "Daily Report",
template: "report",
},
{ delay: 60000 } // Delay 1 minute
);
// Consumer: process email jobs
const worker = new Worker(
taskQueue,
"emails",
async (payload) => {
const emailPayload = payload as {
to: string;
subject: string;
template: string;
};
console.log(`Sending email to ${emailPayload.to}`);
// await sendEmail(emailPayload);
},
3 // Process 3 emails concurrently
);
// Start stall recovery on interval
setInterval(() => {
taskQueue.recoverStalledJobs("emails").then(count => {
if (count > 0) console.log(`Recovered ${count} stalled jobs`);
});
}, 30000);
await worker.start();
}Key Takeaways
Building a task queue from scratch reveals the complexity hiding behind simple-looking job processing APIs. The core challenges aren't about queuing data—they're about reliability guarantees. Atomic job transitions between states prevent duplication. Exponential backoff prevents retry storms. Dead letter queues prevent silent data loss. Stall recovery prevents jobs from vanishing when workers crash.
Understanding these primitives makes you a better consumer of production queue libraries, and it gives you the foundation to debug issues that inevitably arise in distributed job processing systems.
