Background Jobs Designer

Design reliable background job processing with retries and monitoring.

Queue Integration

BullMQ (Node.js):

import { Queue, Worker } from "bullmq";

const emailQueue = new Queue("email", { connection: { host: "localhost", port: 6379 }, });

// Add job await emailQueue.add( "send-welcome", { userId: "123", email: "user@example.com", }, { attempts: 3, backoff: { type: "exponential", delay: 2000 }, } );

Celery (Python):

from celery import Celery

app = Celery('tasks', broker='redis://localhost:6379')

@app.task(bind=True, max_retries=3) def send_email(self, user_id, email): try: # Send email pass except Exception as exc: raise self.retry(exc=exc, countdown=60)

Job Definitions

export interface Job { id: string; type: string; payload: unknown; attempts: number; maxAttempts: number; createdAt: Date; processedAt?: Date; failedAt?: Date; error?: string; }

export const JOB_TYPES = { SEND_EMAIL: "send-email", PROCESS_PAYMENT: "process-payment", GENERATE_REPORT: "generate-report", SYNC_DATA: "sync-data", } as const;

Retry Strategy

// Exponential backoff const RETRY_CONFIG = { maxAttempts: 5, delays: [ 1000, // 1 second 5000, // 5 seconds 30000, // 30 seconds 300000, // 5 minutes 1800000, // 30 minutes ], };

// Worker with retry const worker = new Worker("email", async (job) => { try { await sendEmail(job.data); } catch (error) { if (job.attemptsMade < RETRY_CONFIG.maxAttempts) { throw error; // Will retry } await handleFailedJob(job, error); } });

Idempotent Jobs

// Track processed jobs export const processJob = async (job: Job) => { // Check if already processed const processed = await db.query( "SELECT 1 FROM processed_jobs WHERE job_id = $1", [job.id] );

if (processed.rows.length > 0) { console.log("Job already processed"); return; // Idempotent }

await db.transaction(async (trx) => { // Mark as processed await trx("processed_jobs").insert({ job_id: job.id });

// Do work
await performWork(job, trx);

}); };

Monitoring

// Job events worker.on("completed", (job) => { metrics.increment("jobs.completed", { type: job.name }); });

worker.on("failed", (job, err) => { metrics.increment("jobs.failed", { type: job.name }); logger.error("Job failed", { jobId: job.id, error: err }); });

worker.on("stalled", (jobId) => { metrics.increment("jobs.stalled"); logger.warn("Job stalled", { jobId }); });

Best Practices

• Jobs should be idempotent

• Use exponential backoff for retries

• Set reasonable timeouts

• Monitor queue depth

• Dead letter queue for failed jobs

• Log job start/completion

• Graceful shutdown handling

Output Checklist

• Queue setup (Redis/RabbitMQ)

• Job type definitions

• Retry policy with backoff

• Idempotency tracking

• Error handling

• Monitoring/metrics

• Dead letter queue

• Graceful shutdown