import { Queue } from "bullmq";
import IORedis from "ioredis";
export const connection = new IORedis(process.env.REDIS_URL ?? "redis://localhost:6379", {
maxRetriesPerRequest: null,
});
export interface EmailJob {
to: string;
template: string;
vars: Record<string, unknown>;
}
export const emailQueue = new Queue<EmailJob>("email", {
connection,
defaultJobOptions: {
attempts: 5,
backoff: { type: "exponential", delay: 1000 },
removeOnComplete: { age: 3600, count: 1000 },
removeOnFail: { age: 24 * 3600 },
},
});
export const deadLetterQueue = new Queue("email:dead-letter", {
connection,
defaultJobOptions: { removeOnComplete: false, removeOnFail: false },
});
import { Worker, UnrecoverableError, Job } from "bullmq";
import { connection, deadLetterQueue, EmailJob } from "./queues";
import { sendEmail, InvalidRecipientError } from "./mailer";
const worker = new Worker<EmailJob>(
"email",
async (job: Job<EmailJob>) => {
try {
// keyed on job.id so retried deliveries stay idempotent
return await sendEmail(job.id!, job.data);
} catch (err) {
if (err instanceof InvalidRecipientError) {
throw new UnrecoverableError(`bad recipient: ${err.message}`);
}
throw err;
}
},
{ connection, concurrency: 10 },
);
worker.on("failed", async (job, err) => {
if (!job) return;
const exhausted = job.attemptsMade >= (job.opts.attempts ?? 1);
const permanent = err instanceof UnrecoverableError;
if (!exhausted && !permanent) return;
await deadLetterQueue.add("dead", {
originalId: job.id,
payload: job.data,
failedReason: err.message,
attemptsMade: job.attemptsMade,
});
});
worker.on("error", (err) => console.error("worker error", err));
export default worker;
import { Worker, Job } from "bullmq";
import { connection } from "./queues";
import { alertOncall, archiveFailure } from "./triage";
interface DeadJob {
originalId?: string;
payload: unknown;
failedReason: string;
attemptsMade: number;
}
const dlqWorker = new Worker<DeadJob>(
"email:dead-letter",
async (job: Job<DeadJob>) => {
const { originalId, failedReason, payload } = job.data;
await archiveFailure({ originalId, failedReason, payload });
if (failedReason.startsWith("bad recipient")) return;
await alertOncall(`email job ${originalId} dead-lettered: ${failedReason}`);
},
{ connection, concurrency: 2 },
);
dlqWorker.on("completed", (job) =>
console.log(`triaged dead job ${job.data.originalId}`),
);
export default dlqWorker;
This snippet shows how a resilient job pipeline is built with BullMQ, where transient failures are retried with backoff and permanently failed jobs are routed to a dedicated dead-letter queue rather than silently discarded.
In queues.ts, a single shared IORedis connection is created with maxRetriesPerRequest: null, which BullMQ requires for blocking commands used by workers. Two Queue instances are defined: the primary emailQueue and a separate deadLetterQueue. The defaultJobOptions encode the retry policy at enqueue time — attempts: 5 combined with an exponential backoff starting at one second means BullMQ automatically re-schedules a failed job with increasing delay (roughly 1s, 2s, 4s, and so on). removeOnComplete keeps Redis from growing unbounded by trimming succeeded jobs, while removeOnFail is left generous so failures remain inspectable.
The Worker in emailWorker.ts is the consumer. Its processor function does the real work and is deliberately allowed to throw: throwing is how a job signals failure to BullMQ, which then applies the configured backoff. The code distinguishes recoverable errors from permanent ones by throwing a plain Error for retryable cases and an UnrecoverableError for cases that should never be retried — the latter short-circuits the remaining attempts immediately. The concurrency option bounds how many jobs run in parallel per worker process.
The crucial piece is the failed event handler. BullMQ retries internally, but it does not move exhausted jobs anywhere by default. The handler checks job.attemptsMade against job.opts.attempts so it only fires the dead-letter logic once all retries are truly spent. At that point it re-enqueues the original payload onto deadLetterQueue, preserving the failedReason and originalId for later triage or manual replay.
Processing should be idempotent because a job can run more than once across retries; the sendEmail call keys on job.id so a duplicate delivery attempt is safe. dlqWorker.ts demonstrates the consuming side of the dead-letter queue — typically low-concurrency, used for alerting, logging to durable storage, or human review. This pattern is worth reaching for whenever work touches unreliable external systems and losing a job is unacceptable.
Related snips
export interface RetryOptions {
retries: number;
baseMs: number;
maxMs: number;
signal?: AbortSignal;
onRetry?: (attempt: number, delay: number, err: unknown) => void;Exponential backoff with jitter for retries
package dbutil
import (
"context"
"github.com/jackc/pgconn"Retry Postgres serialization failures with bounded attempts
require "csv"
class PeopleCsvStream
include Enumerable
HEADERS = %w[id full_name email signed_up_at plan].freezeResilient CSV Export as a Streamed Response
export type Settled<R> =
| { status: 'fulfilled'; value: R }
| { status: 'rejected'; reason: unknown };
export interface ConcurrencyOptions {
limit: number;Simple concurrency limiter for batch operations
json.array! @posts do |post|
json.cache! ['v1', post], expires_in: 1.hour do
json.id post.id
json.title post.title
json.excerpt post.excerpt
json.published_at post.published_atFragment caching for expensive JSON serialization
import { randomBytes, createHash } from "crypto";
import jwt from "jsonwebtoken";
import { RefreshTokenStore } from "./store";
const ACCESS_SECRET = process.env.ACCESS_SECRET!;
const ACCESS_TTL = "15m";JWT access + refresh token rotation (conceptual)
Share this code
Here's the card — post it anywhere.
