Timeout Jobs

Implement timeouts for long-running jobs using AbortController

BullMQ does not provide a specific mechanism to timeout jobs, however this can be accomplished in many cases with a custom timeout code in the worker’s process function.

Basic Timeout Pattern

The basic concept is to set up a timeout callback that will abort the job processing, and throw an UnrecoverableError to avoid retries:

const worker = new Worker('foo', async job => {
  let controller = new AbortController();
  const timer = setTimeout(() => controller.abort(), job.data.timeout);
    
  try {
    await doSomethingAbortable(controller.signal);
  } catch(err) {
     if (err.name == "AbortError") {
      throw new UnrecoverableError("Timeout");
    } else {
      throw err;
    }
  } finally {
    clearTimeout(timer);
  }
});

Note how we specified the timeout as a property of the job’s data, in case we want to have different timeouts depending on the job. You could also use a fixed constant timeout for all jobs.

Timeout with Fetch API

When working with HTTP requests, you can use the native AbortController with the Fetch API:

const worker = new Worker("foo", async (job) => { 
  let controller = new AbortController();
  const timer = setTimeout(() => controller.abort(), job.data.timeout);
  
  try {
    let response = await fetch("/slowserver.com", {
      signal: controller.signal,
    }); 
    const result = await response.text();
  } catch (err) {
    if (err.name == "AbortError") {
      throw new UnrecoverableError("Timeout");
    } else {
      throw err;
    }
  } finally {
    clearTimeout(timer);
  }
});

Note that abort will even cause the async call to response.text() to also throw an Abort exception.

Timeout with Retries

If you want timeouts to trigger retries instead of immediate failure, throw a regular Error instead of UnrecoverableError:

const worker = new Worker('foo', async job => {
  let controller = new AbortController();
  const timer = setTimeout(() => controller.abort(), job.data.timeout);
    
  try {
    await doSomethingAbortable(controller.signal);
  } catch(err) {
     if (err.name == "AbortError") {
      // This will trigger a retry with backoff
      throw new Error(`Job timed out after ${job.data.timeout}ms`);
    } else {
      throw err;
    }
  } finally {
    clearTimeout(timer);
  }
});

Adding Jobs with Timeout

await queue.add(
  'process-video',
  { 
    videoUrl: 'https://example.com/video.mp4',
    timeout: 30000 // 30 seconds
  },
  {
    attempts: 3,
    backoff: {
      type: 'exponential',
      delay: 5000,
    },
  },
);

Summary

While it is possible to implement timeout in your jobs, the mechanism to do it may vary depending on the type of asynchronous operations your job is performing. In many cases, using AbortController in combination with a setTimeout is more than enough.

For sandboxed processors, see Timeout for Sandboxed Processors for special considerations.

Related Resources

Cancelling Jobs

Learn about job cancellation with AbortSignal

UnrecoverableError

API reference for UnrecoverableError