Inconsistent ID Return Behavior with BullMQ Workers in Node.js

[Ekeu]

I'm experiencing an inconsistent behavior with BullMQ where the ID returned from a createPost job sometimes returns null when the 'deletePost' job is commented out in my Node.js application. When I comment the 'deletePost' job, it works as expected, returning the created post ID every time a request is sent.

Here is my controller class

export class PostControllers {
  static async createPost(req: Request, res: Response) {
    const postData = {
      ...req.body,
      user: req.currentUser?.userId,
      authUser: req.currentUser?.authUser,
    };

    //...

    postMQ.addJob("create-post", { value: { ...postData, _id: post._id } });

    return res.status(StatusCodes.CREATED).json({
      success: true,
      message: "Post created successfully",
    });
  }

  static async deletePost(req: Request, res: Response) {
    const { postId } = req.params;

    //....

    postMQ.addJob("delete-post", { postId, userId: req.currentUser?.userId });

    return res.status(StatusCodes.OK).json({
      success: true,
      message: "Post deleted successfully",
    });
  }

}

Here is my PostMQ class

class PostMQ extends BaseMQ {
  constructor() {
    super("post-mq");
    this.processJob("create-post", PostWorker.createPost, { concurrency: 5 });
    this.processJob("delete-post", PostWorker.deletePost, { concurrency: 5 });
  }

  async addJob(jobName: string, data: IPostJob): Promise<void> {
    await this.queue.add(jobName, data);
  }
}

export const postMQ = new PostMQ();

The worker class

export class PostWorker {
  static async createPost(job: Job<IPostJob>): Promise<INewPostJobResponse> {
    console.log("POST SERVICE CALLED TO CREATE POST", post);
    const post = await PostServices.createPost(job.data.value as INewPost);
    return {
      _id: post._id.toString(),
    };
  }

  static async deletePost(job: Job<IPostJob>): Promise<void> {
    await PostServices.deletePost(job.data.postId as string, job.data.userId as string);
  }
}

This is the BaseMQ parent class

export abstract class BaseMQ {
  protected queue: Queue;
  private workers: Map<string, Worker>;
  private queueEvents: QueueEvents;
  private static queues: BullMQAdapter[] = [];
  static router: any;
  static serverAdapter: ExpressAdapter = new ExpressAdapter().setBasePath(/* Path */);

  private readonly REDIS_CONNECTION = {
    host: "localhost",
    port: 6379,
  };

  constructor(name: string) {
    this.queue = new Queue(name, {
      connection: this.REDIS_CONNECTION,
      defaultJobOptions: {
        attempts: 3,
        backoff: {
          type: "fixed",
          delay: 5000,
        },
      },
    });
    this.workers = new Map();
    this.queueEvents = new QueueEvents(name, { connection: this.REDIS_CONNECTION });
    BaseMQ.queues.push(new BullMQAdapter(this.queue));

    createBullBoard({
      queues: BaseMQ.queues,
      serverAdapter: BaseMQ.serverAdapter,
    });

    logger.info(`[BullMQ]: ${name} queue created`);

    // ...queue events
  }

  protected async addJob(jobName: string, data: TJobData): Promise<void> {
    const job = await this.queue.add(jobName, data);
    logger.info(`[BullMQ - ${this.queue.name}]: Job with ID '${job.id}' has been added to the queue.`);
  }

  protected processJob(jobName: string, callback: Processor, options?: WorkerOptions): void {
    const worker = new Worker(this.queue.name, callback, { connection: this.REDIS_CONNECTION, ...options });
    this.workers.set(jobName, worker);
    logger.info(`[BullMQ - ${this.queue.name}]: Worker for job '${jobName}' has been started.`);

    worker.on("completed", (job: Job, result) => {
      logger.info(`[JobMQ - ${job.name}]: Completed job with data: `, {
        data: job.asJSON().data,
        id: job.id,
        result,
      });
    });

    worker.on("failed", (job: Job | undefined, error: Error) => {
      logger.error(`[JobMQ - ${job?.name || "Error"}]: Failed job with data: `, {
        data: job?.asJSON().data ?? "No data",
        id: job?.id ?? "No ID",
        error,
      });
    });
  }
}

On the below images you can see that for the same request some times the create post service is called and some times it isn't. What's weird and that I can't figure out is that this only happens if the delete worker is added. That is when I call the process job function like this

this.processJob("delete-post", PostWorker.deletePost, { concurrency: 5 });

I'm still on a learning curve of BullMQ 🙏🏽

I'm using the following versions:

• Node.js: (v18.18.2)
• BullMQ: (4.13.2)
• Redis: (4.6.10)

Has anyone encountered a similar issue or can provide insight into what might be causing this behavior?

Thanks

[Ekeu]

I figured out I'm doing it wrong after reading the docs again. In fact I added 2 workers to my queue. And when I add a job the first worker available processes the job. Hence, the reason why sometimes I get an Id sometimes I don't.

But is it possible to tell a worker to process jobs with a specific job name? 😅