Agenda repeatEvery('10 seconds') fails processing remaining jobs

[james-lukensow]

I'm using Agenda to process the status of an external system that is processing user-generated content. Few background before I display the snippet code that is failing.

A user creates a request, which in part, creates an Agenda job to repeat every 10 seconds until the job competition is complete. The typical processing time is around 20 - 50 seconds for each job to finish. Under no load, or minimal jobs in the queue, everything works fine, but when I batch insert 15 or more items (15 unique jobs created), it will randomly stop the processing of the remaining jobs.

The goal is for each user submission created, we have to start fetching from a 3rd party API, the status of the image transformation. If after 30 minutes, the transformation still has not been completed, then I just want to remove the job.

If the transformation is successful, then I update the associated model and remove the job.

If the transformation is unsuccessful, then I update the associated model as "failed" and remove the job.

Controller

const job = agenda.create('processingStatus', response);
job.repeatEvery('10 seconds');
await job.save();

Agenda Job

const jobName = 'processingStatus';

agenda.define(jobName, async (job, done) => {
  const { attrs: { data: { _id, uuid } } } = job;

  // Fetch the current status of the image
  const response = await getImageStatus({ uuid });

  const createdAt = new Date(job.attrs.data.createdAt);
  const now = new Date();

  const diffInMinutes = differenceInMinutes(
    now,
    createdAt,
  );

  logger.info(`Checking job for completition, _id: ${job.attrs._id}`);
  logger.info(`Checking job duration since creation: ${diffInMinutes}`);

  // The response was successful and photo transformed
  if (response && response.finished_at && response.path) {
    const { path } = response;
    const name = `${uuid}/image.jpg`;

    // Copy over the S3 asset
    await copyS3Obect({ name, path });

    const data = {
      status: 'created',
      s3Path: name,
    };

    // Update the document with the s3 url
    await Photo.findOneAndUpdate(
      {
        _id,
      },
      data,
    );

    const diffInSeconds = differenceInSeconds(
      new Date(),
      createdAt,
    );

    logger.info(`Job ${job.attrs._id} finished in ${diffInSeconds} seconds`);

    await job.remove();
  }

  // Failed creating the photo transformation, remove the job and update the model
  if (response && response.failed_at) {
    const data = {
      status: 'failed',
    };

    // Update the document (failed)
    await Photo.findOneAndUpdate(
      {
        _id,
      },
      data,
    );

    await job.remove();
  }

  // The photo failed to created within 30 minutes, remove the job and update the model
  if (diffInMinutes > 30) {
    logger.error('Photo not created within 30 minutes');
    const data = {
      status: 'failed',
    };

    // Update the document (failed)
    await Photo.findOneAndUpdate(
      {
        _id,
      },
      data,
    );

    await job.remove();
  }

  done();
})