Global only one job at a time

[vsamofal]

I know if you do have multiple workers it's not possible to archive it, it is stated clearly in docs. But what if I will be checking for active jobs, and if there are more then one active jobs, I will be just skipping this execution run.

Does anybody see any race condition problem with it? as for me it should be fine

And if it's ok, it is worth adding it as a pattern to the docs?
and I didn't find easy way to get active jobs by id, nothing in docs, probably it's not possible, or possible with some custom redis request

Here is an example implementation for this:

@Processor('fake-job-queue', {
  concurrency: 50,
})
export class FakeJob extends WorkerHost {
  public callCounter: number = 0;

  constructor(@InjectQueue('fake-job-queue') private queue: Queue) {
    super();
  }

  override async process(job: Job<FakeJobData, number>): Promise<FakeJobData> {
    const hasActiveJobs = await this.hasOtherActiveJobsById(job);

    if (hasActiveJobs) {
      return job.data;
    }

    this.callCounter++;

    await wait(job.data.executeForMillis);

    return job.data;
  }

  private async hasOtherActiveJobsById(
    job: Job<FakeJobData, number>,
  ): Promise<boolean> {
    const currentJobId = job.opts.repeat?.jobId ?? job.opts.jobId;
    const BATCH_SIZE = 100;

    let activeJobsCount = 0;
    for (let i = 0; ; i++) {
      const activeJobs = await this.queue.getJobs(
        ['active'],
        i * BATCH_SIZE,
        BATCH_SIZE,
        true,
      );

      for (const activeJob of activeJobs) {
        const jobOptions = activeJob.opts;
        if (
          jobOptions.repeat?.jobId === currentJobId ||
          jobOptions.jobId === currentJobId
        ) {
          activeJobsCount++;
          if (activeJobsCount > 1) {
            return true;
          }
        }
      }

      if (activeJobs.length < BATCH_SIZE) {
        return false;
      }
    }
  }
}

and test:

  it('should execute only one job at a time with overlap and concurrency bigger then 1', async () => {
    await fakeJobQueue.add(
      'fake-job-name',
      { executeForMillis: 3000 },
      {
        jobId,
        repeat: {
          every: 1000,
        },
      },
    );

    await wait(5000);

    expect(fakeJob.callCounter).toBe(2);
  });

[manast]

Wouldn't be simpler to just check if the active count is larger or equal than 2 and if so skip the processing of this job?

[vsamofal]

@manast thanks for the immediate response!

I'm creating a custom id, because I need to have one job per tenant data source (organisation) that uses an app.

that's why I can't just check for all, I need to verify that there is no other job by this id in this queue is running

and also I have a repeatable job, that runs once per day, and the user may have the ability to run a job right now from UI, and in case time overlaps, I don't want these jobs to run in parallel, and also bull is doing a great job by itself for throttling

the only concern is a race condition, there is a possibility I guess that both jobs will be skipped.

if two jobs will instantiate in the same millisecond (that is unlikely, but may happen) and both will call getActive and both will see 2 jobs.

[vsamofal]

to fix this, I probably can leverage job.updateJson, and add a lock ... but not sure yet