使用moveToDelayed后BullMQ抛出'Missing lock for job <jobId> failed'错误

问题描述 #

参考BullMQ分步任务模式实现代码时，抛出Missing lock for job <jobId> failed错误。以下是最小复现代码：

import { Worker, Queue } from "bullmq";

const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));

enum Step {
  Initial,
  Second,
  Finish,
}

type JobData = {
  step: Step;
};

const worker = new Worker<JobData>("reproduce-error", async (job, token) => {
  let step = job.data.step;
  while (step !== Step.Finish) {
    switch (step) {
      case Step.Initial: {
        await sleep(3 * 1000);
        await job.moveToDelayed(Date.now() + 1000, token);
        await job.update({
          step: Step.Second,
        });
        await job.updateProgress(1);
        step = Step.Second;
        break;
      }
      case Step.Second: {
        await sleep(3 * 1000);
        await job.update({
          step: Step.Finish,
        });
        await job.updateProgress(2);
        step = Step.Finish;
        return Step.Finish;
      }
      default: {
        throw new Error("invalid step");
      }
    }
  }
});

worker.on("error", (failedReason) => console.log(failedReason));

worker.on("progress", (job, progress) =>
  console.log("progress", job.data, progress)
);

const queue = new Queue<JobData>("reproduce-error");
queue.add("Reproduce error", { step: Step.Initial });

错误堆栈信息：

at Scripts.finishedErrors (/Users/sumit/Coding/indiemaker/twips/reproduce-bullmq-error/node_modules/bullmq/src/classes/scripts.ts:356:16)
    at Job.moveToFailed (/Users/sumit/Coding/indiemaker/twips/reproduce-bullmq-error/node_modules/bullmq/src/classes/job.ts:618:26)
    at processTicksAndRejections (node:internal/process/task_queues:95:5)
    at async handleFailed (/Users/sumit/Coding/indiemaker/twips/reproduce-bullmq-error/node_modules/bullmq/src/classes/worker.ts:642:11)
    at async Worker.retryIfFailed (/Users/sumit/Coding/indiemaker/twips/reproduce-bullmq-error/node_modules/bullmq/src/classes/worker.ts:788:16)
    at async Worker.run (/Users/sumit/Coding/indiemaker/twips/reproduce-bullmq-error/node_modules/bullmq/src/classes/worker.ts:385:34)

初步排查时尝试将Initial分支的break改为return，但问题依旧。

原因分析 #

错误的核心原因是：

1. 调用job.moveToDelayed后，BullMQ会立即释放当前任务的锁，因为任务已被移至延迟队列，当前Worker的处理逻辑应该终止。
2. 原代码中moveToDelayed之后还调用了job.update和job.updateProgress，这些方法需要持有任务锁才能执行，此时锁已释放，因此触发Missing lock错误。
3. 即使改成return，如果moveToDelayed之后的代码已经进入异步队列，还是会在锁释放后执行，导致报错。

解决方案 #

调整执行顺序：先更新任务数据和进度，再调用moveToDelayed，最后立即return终止当前处理。这样确保所有需要锁的操作都在锁释放前完成，且任务移至延迟队列后不再执行后续逻辑。

修改后的代码：

import { Worker, Queue } from "bullmq";

const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));

enum Step {
  Initial,
  Second,
  Finish,
}

type JobData = {
  step: Step;
};

const worker = new Worker<JobData>("reproduce-error", async (job, token) => {
  let step = job.data.step;
  while (step !== Step.Finish) {
    switch (step) {
      case Step.Initial: {
        await sleep(3 * 1000);
        // 先更新任务数据和进度（此时锁仍持有）
        await job.update({
          step: Step.Second,
        });
        await job.updateProgress(1);
        // 再移至延迟队列
        await job.moveToDelayed(Date.now() + 1000, token);
        // 立即return，终止当前Worker执行
        return;
      }
      case Step.Second: {
        await sleep(3 * 1000);
        await job.update({
          step: Step.Finish,
        });
        await job.updateProgress(2);
        step = Step.Finish;
        return Step.Finish;
      }
      default: {
        throw new Error("invalid step");
      }
    }
  }
});

worker.on("error", (failedReason) => console.log(failedReason));

worker.on("progress", (job, progress) =>
  console.log("progress", job.data, progress)
);

const queue = new Queue<JobData>("reproduce-error");
queue.add("Reproduce error", { step: Step.Initial });

关键修改点：

• 调换job.moveToDelayed与job.update、job.updateProgress的顺序，确保锁释放前完成所有需要锁的操作
• 在job.moveToDelayed后添加return，终止当前Worker的执行逻辑，避免后续代码运行

补充说明 #

分步任务的核心逻辑是：任务完成当前步骤后，更新状态并移至延迟队列，等待下一次被Worker取出执行后续步骤。因此每次处理完一个步骤（除了最终步骤），都应该终止当前执行，让任务在延迟结束后重新进入队列处理下一个步骤。

内容的提问来源于stack exchange，提问作者Sumit Ghosh