Design a Task Queue System

Difficulty: Intermediate

Question

Design a distributed task queue system like Celery or Bull. How would you handle job prioritization, retries, and failure recovery?

Answer

A task queue processes background jobs asynchronously, decoupling time-consuming work from user-facing requests.

Use cases: sending emails, generating reports, processing images, running ML inference, scheduled tasks.

Core components: 1. Producer: Enqueues tasks with priority and metadata 2. Queue: Ordered storage (Redis sorted sets or dedicated broker) 3. Worker Pool: Processes tasks with concurrency control 4. Scheduler: Handles delayed and recurring tasks (cron) 5. Dead Letter Queue: Captures permanently failed tasks

Key features: priority levels, delayed execution, retries with exponential backoff, job deduplication, progress tracking, rate limiting per queue.

Code examples

Task Queue with Redis (BullMQ-style)

// Job producer
class TaskQueue {
  constructor(name) {
    this.name = name;
    this.queueKey = `queue:${name}`;
  }

  async addJob(type, data, options = {}) {
    const job = {
      id: generateId(),
      type,
      data,
      priority: options.priority || 0,
      attempts: 0,
      maxAttempts: options.maxAttempts || 3,
      delay: options.delay || 0,
      createdAt: Date.now()
    };

    if (options.dedupeKey) {
      const exists = await redis.get(`dedupe:${options.dedupeKey}`);
      if (exists) return null; // Skip duplicate
      await redis.set(`dedupe:${options.dedupeKey}`, '1', 'EX', 3600);
    }

    const executeAt = Date.now() + (options.delay || 0);

    // Sorted set: score = priority (lower = higher priority) + timestamp
    const score = job.priority * 1e13 + executeAt;
    await redis.zadd(this.queueKey, score, JSON.stringify(job));

    return job;
  }
}

// Usage
const emailQueue = new TaskQueue('emails');

// Immediate job
await emailQueue.addJob('WELCOME_EMAIL', { userId: '123', email: 'user@test.com' });

// Delayed job (send in 1 hour)
await emailQueue.addJob('REMINDER_EMAIL', { userId: '123' }, { delay: 3600000 });

// High priority job
await emailQueue.addJob('OTP_EMAIL', { email: 'user@test.com', code: '1234' }, { priority: -10 });

Redis sorted sets provide natural priority ordering. Lower scores are dequeued first, so high-priority jobs use negative scores. Deduplication prevents the same job from being queued twice.

Worker Pool with Retry Logic

class Worker {
  constructor(queue, handlers, concurrency = 5) {
    this.queue = queue;
    this.handlers = handlers;
    this.concurrency = concurrency;
    this.active = 0;
  }

  async start() {
    console.log(`Worker started: ${this.queue.name} (concurrency: ${this.concurrency})`);

    while (true) {
      if (this.active >= this.concurrency) {
        await sleep(100);
        continue;
      }

      // Atomic dequeue: get and remove in one operation
      const jobData = await redis.zpopmin(this.queue.queueKey);
      if (!jobData || jobData.length === 0) {
        await sleep(1000); // No jobs, wait
        continue;
      }

      const job = JSON.parse(jobData[0]);

      // Check if delayed job is ready
      if (job.delay && Date.now() < job.createdAt + job.delay) {
        await redis.zadd(this.queue.queueKey, jobData[1], jobData[0]);
        await sleep(1000);
        continue;
      }

      this.active++;
      this.processJob(job).finally(() => this.active--);
    }
  }

  async processJob(job) {
    const handler = this.handlers[job.type];
    if (!handler) {
      console.error(`No handler for job type: ${job.type}`);
      return;
    }

    try {
      job.attempts++;
      await handler(job.data);
      console.log(`Job completed: ${job.id} (${job.type})`);
    } catch (error) {
      console.error(`Job failed: ${job.id}`, error.message);

      if (job.attempts < job.maxAttempts) {
        // Retry with exponential backoff
        const backoff = Math.pow(2, job.attempts) * 1000;
        job.delay = backoff;
        job.createdAt = Date.now();
        const score = job.priority * 1e13 + Date.now() + backoff;
        await redis.zadd(this.queue.queueKey, score, JSON.stringify(job));
        console.log(`Retrying job ${job.id} in ${backoff}ms`);
      } else {
        // Move to dead letter queue
        await redis.lpush('dlq:' + this.queue.name, JSON.stringify({
          ...job, error: error.message, failedAt: Date.now()
        }));
      }
    }
  }
}

// Start worker
const worker = new Worker(emailQueue, {
  WELCOME_EMAIL: async (data) => await sendWelcomeEmail(data),
  OTP_EMAIL: async (data) => await sendOtpEmail(data),
  REMINDER_EMAIL: async (data) => await sendReminderEmail(data)
}, 5);

Workers process jobs concurrently with a configurable limit. Failed jobs are retried with exponential backoff. After max attempts, jobs move to a dead letter queue for manual investigation.

Scheduled & Recurring Jobs

// Cron-like scheduler for recurring tasks
class Scheduler {
  constructor() {
    this.schedules = [];
  }

  // Register a recurring job
  addSchedule(cronExpression, queue, jobType, jobData) {
    this.schedules.push({
      cron: cronExpression,
      queue,
      jobType,
      jobData,
      lastRun: 0
    });
  }

  async start() {
    // Check every minute
    setInterval(async () => {
      for (const schedule of this.schedules) {
        if (this.shouldRun(schedule)) {
          await schedule.queue.addJob(
            schedule.jobType,
            schedule.jobData,
            { dedupeKey: `cron:${schedule.jobType}:${Date.now()}` }
          );
          schedule.lastRun = Date.now();
          console.log(`Scheduled job enqueued: ${schedule.jobType}`);
        }
      }
    }, 60000);
  }

  shouldRun(schedule) {
    // Simplified cron matching
    return cronMatch(schedule.cron, new Date());
  }
}

// Usage
const scheduler = new Scheduler();

// Daily report at 9 AM
scheduler.addSchedule('0 9 * * *', reportQueue, 'DAILY_REPORT', {});

// Hourly cleanup
scheduler.addSchedule('0 * * * *', maintenanceQueue, 'CLEANUP_EXPIRED', {});

// Weekly digest every Monday at 10 AM
scheduler.addSchedule('0 10 * * 1', emailQueue, 'WEEKLY_DIGEST', {});

// Monitoring dashboard data
// Track: jobs processed/failed per minute
// Track: average processing time per job type
// Track: queue depth (pending jobs)
// Track: worker utilization
// Alert: if queue depth > threshold for > 5 min

The scheduler enqueues jobs on a cron schedule. Deduplication keys prevent duplicate jobs if the scheduler runs on multiple instances. Monitor queue depth to detect processing bottlenecks.

Key points

Concepts covered

Task Queue, Background Jobs, Priority Queue, Retry Logic, Worker Pool