Managing Job Queues in Bull While Preventing Log Clutter from Unprocessed Jobs

Introduction

If you’ve landed here, you’re likely familiar with the intricacies of working with job queues in Bull. This isn’t an introductory article about Bull; it’s a deep dive into orchestrating a specific kind of job queue—queues that need to process jobs one at a time and manage failed or unnecessary jobs gracefully.

Our use case is an email processing system. The system needs to prepare and send out email batches. The need for sequential processing comes from the requirement to handle these tasks one at a time to avoid sending multiple emails to the same recipient or mixing up batches.

Let’s walk through how to set this up.

Queue Configuration

First, let’s set up the Bull queue with specific options to limit the processing to one job at a time.

import Bull from "bull";

const queueOptions = {
  redis: { host: 'REDIS_HOST', port: 'REDIS_PORT' },
  limiter: {
    max: 1,
    duration: 1000,
  },
};

const emailQueue = new Bull("EmailQueue", queueOptions);

Notice the limiter configuration with max: 1. This ensures that only one job is processed at a time.

Registering the Processor

Here’s an abridged version of our processor code, with the essential parts for understanding:

emailQueue.process(async (job, done) => {
  try {
    const latestEmailBatch = await getEmailBatch();
    if (!latestEmailBatch) {
      throw new Error("No email batch found");  // This line makes the job fail
    }
    
    // Processing logic
    // ...
    
    done(null, 'Job done successfully');
  } catch (err) {
    done(err);
  }
});

Notice the line that throws an error? This will cause the job to go into the “failed” state, and later on, we will specifically delete such jobs.

Handling Failed and Old Jobs

The Importance of Timers

The intervals for deleting failed and old jobs are based on your requirements. In this example, the setInterval runs every 10 minutes. You may adjust this based on how quickly your jobs are being added and processed.

Cleaning Up Failed Jobs with Specific Reasons

To delete jobs that failed because there was no work to do, use the deleteAllFailedJobsWithMessage function. This is crucial to prevent your Redis storage from bloating up with irrelevant data.

async function deleteAllFailedJobsWithMessage(messageToFind: string) {
  const jobs = await emailQueue.getJobs(["failed"]);
  
  for (const job of jobs) {
    if (
      job.failedReason &&
      job.failedReason.toLowerCase().includes(messageToFind.toLowerCase())
    ) {
      await job.remove();
    }
  }
}

Removing All Old Jobs

Sometimes, you may want to remove jobs that are older than a specific timeframe, irrespective of whether they’ve succeeded or failed. The deleteAllOldJobs function serves this purpose.

async function deleteAllOldJobs() {
  const currentTime = new Date().getTime();
  const timeThreshold = 24 * 60 * 60 * 1000; // 24 hours
  
  const jobs = await emailQueue.getJobs(["completed", "failed", "delayed", "active", "waiting"]);
  
  for (const job of jobs) {
    if (currentTime - job.timestamp > timeThreshold) {
      await job.remove();
    }
  }
}

Scheduling Cleanup Tasks

Finally, let’s schedule these cleanup tasks to run at intervals. This ensures your Redis storage stays clean and manageable.

setInterval(async () => {
  await deleteAllOldJobs();
  await deleteAllFailedJobsWithMessage("No email batch found");
}, 10 * 60 * 1000);  // Every 10 minutes

Conclusion

Balancing between efficiency and resource consumption, we’ve orchestrated a Bull job queue tailored for sequential email processing. We’ve also implemented mechanisms to clean up failed and old jobs, maintaining a clean and efficient queue.

While some might argue that this violates the principle of idempotency, the fact remains: Sometimes, you have to do what you have to do. The art of programming often involves crafting solutions that fit the problem at hand, even if they deviate from the ideal theoretical model. And in cases like this, it’s not just about getting the job done—it’s about doing it well.