Pattern for accumulating job data for batch process

[SaucyJack]

We have a queue we will use to POST batches of data to a service endpoint. Due to volume and constraints on the receiving server, we cannot use a standard queue worker, but rather need to manually process queue jobs. Something like the pattern found here, except in a loop: https://github.com/OptimalBits/bull/blob/develop/PATTERNS.md#manually-fetching-jobs.

Each job in the queue can contain multiple items, in varying numbers, that get accumulated into a batch. When the batch reaches a certain amount, we POST it, then start accumulating job items into a new batch.

For example, say we POST in batches of 100 items:

Job 1: 24 items
Job 2: 33 items
Job 3: 12 items
Job 4: 34 items -----> POST batch of 103 items
Job 5: 22 items (new batch)
etc.

Ideally, if a batch failed to POST for any reason, the associated queue jobs would get failed so that we can retry them. But, since we have to use moveToCompleted to fetch the next job (after the first iteration), I can't keep the jobs and fail them if there is an error posting a batch.

Is there a better pattern I can use to do this?

[manast]

Ideally, if a batch failed to POST for any reason, the associated queue jobs would get failed so that we can retry them. But, since we have to use moveToCompleted to fetch the next job (after the first iteration), I can't keep the jobs and fail them if there is an error posting a batch.

You can pass "true" as last argument of "moveToComplete" if you do not need to fetch the next job. Is that enough to solve your use case?

[SaucyJack]

It might be enough. Could I then use a combination of getNextJob and moveToCompleted to manually process jobs? Something like:

while (1) {
  let job = await queue.getNextJob();
  if (job) {
    // add items to current batch, process POST if necessary, etc.
    await job.moveToCompleted('finished processing', true, true);
  } else {
    break;
  }
}

Also, could I store jobs in an array so that I could fail a series of jobs, something like (apologies, this is not "good" code, just trying to get an idea across):

while (1) {
  let job = await queue.getNextJob();
  if (job) {
    currentJobs.push(job);
    try {
      // add items to current batch, process POST if necessary, etc.
    } catch (err) {
      for (failJob in currentJobs) {
        await failJob.moveToFailed();
      }
    }
  } else {
    break;
  }
}

Thanks for your response!

[manast]

You know there is flow support in BullMQ right? https://docs.bullmq.io/guide/flows maybe it is helpful in your case.

[SaucyJack]

Flows are interesting, but I don't think they would solve this problem. Our jobs aren't really hierarchical, we just have to "batch" the job data up so we don't end up sending tons of service requests, and we will run the process as a cron scheduled task as opposed to an "always up" queue worker.

I was able to use your previous answer to develop a working solution. In case anyone else needs it, here is the basic solution:

async function doWork() {
  let queue
  try {
    let currentBatch = [], currentJobs = [];
    let totalItems = 0;
    while (1) {
      try {
        const job = await queue.getNextJob();
        if (!job) {
          // no more jobs for now
          if (currentBatch.length > 0) {
            // clean up last batch
            await processBatch(currentBatch);
            await completeJobs(currentJobs);
          }
          return totalItems;
        }
        // add the items to the current batch (and job to the current set of jobs)
        currentBatch.push(...job.data);
        currentJobs.push(job);
        // if we have reached the batch limit...
        if (currentBatch.length > BATCH_LIMIT) {
          // process events
         await processBatch(currentBatch);
         // reset the arrays and complete the jobs
         currentBatch = [];
         await completeJobs(currentJobs);
         currentJobs = [];
        }
      } catch (err) {
        // log the error
        console.log(err);
        // fail the jobs in the current batch to be retried later
        for (const failJob of currentJobs) {
          await failJob.moveToFailed(error, true);
        }
        currentBatch = [];
        currentJobs = [];
      }
    }
  } finally {
    if (queue) {
      await queue.close();
    }
  }
}

async function completeJobs(jobs) {
  for (const doneJob of jobs) {
    // IMPORTANT! - must provide second 'true' argument to not fetch next job
    await doneJob.moveToComplete('Successfully processed', true, true);
  }
}

I'm still testing it but I think it is going to work fine. Thanks!