background-jobs-designer

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: "[email protected]",
  },
  {
    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);
  });
};