Rate Limiting

Named workstreams can be rate limited by adding a rateLimit configuration to the workstream. max and duration bound how many of that workstream's jobs may start per time window:

  workersApp.set('background', {
    ...

    namedWorkstreams: [
      {
        name: 'Iterable',
        workerCount: 1,
        concurrency: 10,

        rateLimit: {
          max: 10,
          duration: 1000,
        },
      },
      {
        name: 'Twilio',
        workerCount: 1,
        concurrency: 10,

        // 20 jobs per second
        rateLimit: {
          max: 20,
          duration: 1000,
        },
      },
    ],
  })

  ...
})

rateLimit becomes the BullMQ limiter on that workstream's workers, which is an open-source BullMQ feature — no BullMQ Pro license is involved.

Reacting to a 429

A rate limit you configure is a guess at the external service's budget. When the service itself tells you to slow down — an HTTP 429 carrying a retry-after — throw RateLimitedPsychicJob to pause the whole workstream for that duration without burning one of the job's retry attempts:

import { RateLimitedPsychicJob } from '@rvoh/psychic-workers/errors'

if (response.status === 429) {
  throw new RateLimitedPsychicJob({
    pauseQueueForSeconds: Number(response.headers.get('retry-after') ?? 1),
  })
}

Because the pause doesn't consume an attempt, a service that keeps answering 429 could cycle one job through the pause indefinitely. maxStartedAttempts bounds that; set it on the global defaultBullMQWorkerOptions. Any workstream whose jobs throw the signal wants it.

Throw RateLimitedPsychicJob only from a job on a workstream whose workers carry a limiter — thrown where there is none, it surfaces as a misconfiguration rather than pausing anything.