Manually process specific job

[Niek]

I am trying to find a way to manually process a specific job. I am aware of the documentation at https://docs.bullmq.io/patterns/manually-fetching-jobs - but the documented use-case is not suitable to process a specific job (getNextJob() returns a random job).
The code I thought of is:

const job = await queue.getJob(id)
job && job.moveToCompleted('returndata', 'token', false)

But this throws the error: Error: Missing lock for job <ID>. finished

I haven't found a way to skip the job lock or manually set it.

Any ideas?

[Niek]

Some other things I have tried:

• Use worker.nextJobFromJobData({}, id} to fetch a specific job - this works fine, but I still can't call moveToCompleted() without getting the lock error
• Use job.extendLock('token', 60000) to try to manually set a lock on the job. Unfortunately moveToCompleted() still throws with the lock error
• Use const job = await worker.moveToActive('token', id) to get the job. This works fine, but calling job.moveToCompleted() results in the error: Error: Job <ID> is not in the active state. finished
• Use worker.processJob(job, 'token') - this results in errors because the timeManager is not initialised: TypeError: Cannot read properties of undefined (reading 'setTimer')

[Niek]

My workaround now is to use a separate queue per job - this way, I can simply call getNextJob() and fetch the only job in there. This seems like a lot of overhead though - so any ideas how to fix it better would be appreciated.

[roggervalf]

Hi @Niek, you can pass '0' as token, so the check for the token will be skipped, but you also should delete the lock after moving your job to completed, you can use your redis client to call del(lockKey)

[Niek]

@roggervalf with string 0 as token I get the error Error: Job <ID> is not in the active state. finished when calling moveToCompleted().
Is there any support planned for manually processing specific jobs?

[manast]

This use case deviates too much from the spirit of this library since it is not a queue anymore when you are just picking up arbitrary jobs from the queue. Maybe if you give more context about what you are trying to do we can suggest a better approach while using the queue as intented.

[Niek]

Makes sense @manast!

My use-case is:

• Queue A gets jobs injected by Producer, a bunch of workers are processing these jobs
• The workers need to pass back some information to the Producer once they reach a certain point (finishing the job is not an option, as they need to proceed after this)
• updateProgress() works fine for this, but delivery of the progress messages to Producer is (apparently) non-guaranteed
• My idea was to have Queue B which creates a separate job, and have the Producer await the job with the same ID in Queue B

Maybe I'm overthinking this - is there another way to solve this elegantly but with guaranteed delivery of the progress?

[manast]

So let's say the Producer has a REST API or similar that can be used to send information. Then I would divide the jobs into two parts, or more, you can use flows for this. The first job does whatever it needs to do and ends by adding a job to the "step 2" queue, in this queue the job posts the information to the producer and posts a job to the "step 3" queue, where the job does whatever it needs to do in step 3.

[AntiMoron]

I did it in an ugly way... @manast How to save this code...

    const config = this.getQueueMiddleware(queueName).getConfig()?.[queueName];
    const redisConfig = config?.redis;
    const worker = new Worker(queueName, null, {
      connection: {
        host: redisConfig?.host,
        password: redisConfig?.password,
        port: redisConfig?.port,
        db: redisConfig?.db,
        tls: redisConfig?.tls,
      },
      lockDuration: 1000 * 60 * 3, // 3min lock duration
    });
    const token = getUuid();
    // get a job
    try {
      const job = (await worker.getNextJob(token)) as Job;
      if (!job) {
        return;
      }
      // Extend the lock, make sure the job is not lost
      await job.extendLock(token, 60 * 1000 * 5);
      let succeeded = false;
      let errorMessage = '';
      try {
        const ret = handler(job);
        if (ret instanceof Promise) {
          try {
            await ret;
            succeeded = true;
          } catch (err) {
            this.logger.error(err);
            errorMessage = (err as Error)?.message;
          }
        } else {
          succeeded = true;
        }
      } catch (err) {
        this.logger.error(err);
        errorMessage = (err as Error)?.message;
      }
      if (succeeded) {
        try {
          await job.moveToCompleted('Success', token, false);
        } catch (err) {
          // <--------------ugly one----------------->
          this.logger.info('lock expired not found.');
          await this.app.redis.lrem(
            'bull:staticReport:active',
            1,
            `${job?.id}`,
          );
        }
      } else {
        await job.moveToFailed(
          new Error(`Consume Queue <${queueName}> error: \n` + errorMessage),
          token,
          false,
        );
      }
    } catch (err) {
      this.logger.error(err);
    } finally {
      await worker.close();
    }

[LBC000]

// bullMQUtils.ts
import { delay, Job, Queue, Worker } from "bullmq";

/**
    // 调用手动队列处理函数
    const token = nanoid();
    const { queue: manuallyQueue, worker } = processManualQueue(
      "manuallyQueue",
      {
        connection: connection,
        concurrency: 5,
        delayTime: 3000,
        successCallback: async (data) => {
          if (data.video == 3) return false;
          return true;
        },
      },
      token
    ); 
 */
export function processManualQueue(queueName, options, token) {
  const { connection, concurrency, delayTime, successCallback } = options;
  const queue = new Queue(queueName, { connection });
  const worker = new Worker(queueName, null, {
    connection,
    concurrency: concurrency || 5,
  });

  workerRun();

  async function workerRun() {
    let job;

    while (true) {
      await delay(delayTime || 3000);

      let jobData = null;
      let jobId;
      let success;

      if (job) {
        success = await successCallback(job.data);
        // console.log("处理手动作业", job.data, "状态", success);
        if (success) {
          [jobData, jobId] = await job.moveToCompleted(
            "some return value",
            token
          );
        } else {
          await job.moveToFailed(new Error("some error message"), token);
        }

        if (jobData) {
          job = Job.fromJSON(worker, jobData, jobId);
        } else {
          job = null;
        }
      } else {
        if (!job) {
          job = await worker.getNextJob(token);
        }
      }
    }
  }

  return { queue, worker };
}

// 调用手动队列处理函数
const token = nanoid();
const { queue: manuallyQueue, worker } = processManualQueue(
  "manuallyQueue",
  {
    connection: connection,
    concurrency: 5,
    delayTime: 3000,
    successCallback: async (data) => {
      if (data.video == 3) return false;
      return true;
    },
  },
  token
);

// 手动作业, 失败重试
  async manuallyQueue() {
    // console.log('触发了-test_queue_1');
    // manuallyQueue.add("jobName", { video: 1 });
    // manuallyQueue.add("jobName", { video: 2 });

    for (let i = 1; i < 6; i++) {
      manuallyQueue.add(
        "jobName",
        { video: i },
        {
          attempts: 3, // 重试3次
          backoff: {
            type: "exponential",
            delay: 3000, // 1秒后重试
          },
        }
      );
    }
  },