Skip to main content

Workers - config

Psychic workers leverages BullMQ under the hood to drive the background job systems. The configuration largely revolves around arguments which would be directly provided to bullmq, but with some nuance, which is both due to the nature of the system being wrapped, as well as the fact that we desired a simpler configuration path for those not looking to leverage every possible bell and whistle, and really just looking to set up something simple.

Configuration lives in conf/initializers/workers.ts, which is auto-loaded and registered as a plugin — see the installation guide for the file's full shape.

Two configuration modes​

workersApp.set('background', { ... }) accepts one of two mutually exclusive shapes, and the type system enforces the split:

  • Simple (workstream) mode — Psychic builds the queues and workers from a workstream description. This is what the CLI provisions by default, and what the rest of this guide (and the Named workstreams guide) documents. Services route to a queue via backgroundJobConfig.workstream. Selected by omitting nativeBullMQ.
  • Native BullMQ mode — you hand Psychic raw BullMQ queue and worker options per queue, and Psychic does little beyond constructing them and wiring the job handler. Services route via backgroundJobConfig.queue and groupId instead of workstream. Selected by supplying nativeBullMQ — even nativeBullMQ: {} selects it.

pnpm psy sync generates the routing union (workstream vs. queue) from whichever mode is configured, so the mode decides which key a service can use. priority is legal in both modes.

Most apps only need simple mode. Reach for native mode when you need per-queue BullMQ options simple mode doesn't expose — a distinct Redis instance or cluster node per queue, or BullMQ Pro group settings — see Native BullMQ mode below.

Types​

Psychic scans your worker config whenever you run pnpm psy sync, capturing the workstream names you have configured for use when configuring your app. In order for these types to flow throughout your system, we recommend you add a few files to your file system.

If you selected "yes" to background workers during the cli prompt, these files will already exist in your system.

// app/services/ApplicationBackgroundedService.ts

import { BaseBackgroundedService } from '@rvoh/psychic-workers'
import psychicTypes from '../../types/psychic'

export default class ApplicationBackgroundedService extends BaseBackgroundedService {
  public get psychicTypes() {
    return psychicTypes
  }
}
// app/services/ApplicationScheduledService.ts

import { BaseScheduledService } from '@rvoh/psychic-workers'
import psychicTypes from '../../types/psychic'

export default class ApplicationScheduledService extends BaseScheduledService {
  public get psychicTypes() {
    return psychicTypes
  }
}

You can now inherit from each of these classes in your application, allowing the types to flow gracefully through and protect you from making any mistakes:

Workstreams​

Psychic superimposes a new concept, called a workstream, on top of BullMQ's existing system, which enables one to easily grasp, configure, and express their background systems. In essence, you can think of a workstream as a group, containing a queue and a set of workers to work off that queue:

Many applications will only need one workstream, but multiple workstreams are extremely useful when, for example, you have background jobs responsible for hitting external APIs which are rate limited. Having a background system that doesn't respect rate limits can lead to real chaos in your systems, so it is useful to partition those out into their own workstreams. Once the work is isolated, a named workstream's rateLimit bounds how many of its jobs start per time window — see Rate limiting.

To set up workstreams in Psychic, all you need to do is add a basic workstream setup to your conf/initializers/workers.ts file:

// conf/initializers/workers.ts

function initializeWorkers(workersApp: PsychicAppWorkers) {
  workersApp.set('background', {
    defaultWorkstream: {
      workerCount: 1,
      concurrency: 10,
    },

    namedWorkstreams: [
      {
        name: 'NamedWorkstream',
        workerCount: 1,
      },

      // Rate limited workstream
      // {
      //   name: 'RateLimitedWorkstream',
      //   workerCount: 1,
      //   concurrency: 10,
      //   rateLimit: {
      //     max: 20,
      //     duration: 1000,
      //   },
      // },
    ],

    providers: {
      Queue,
      Worker,
    },

    defaultBullMQQueueOptions: {
      defaultJobOptions: {
        removeOnComplete: 1000,
        removeOnFail: 20000,
        // 524,288,000 ms (~6.1 days) using algorithm:
        // "2 ^ (attempts - 1) * delay"
        attempts: 20,
        backoff: {
          type: 'exponential',
          delay: 1000,
        },
      },
    },

    // Any instance can push onto the queue. This producer (non-blocking) connection
    // sets `enableOfflineQueue: false` so `queue.add()` fails fast when Redis is down
    // instead of buffering jobs in memory that vanish on restart. BullMQ recommends
    // disabling the offline queue on the Queue while leaving it on for Workers.
    // https://docs.bullmq.io/patterns/failing-fast-when-redis-is-down
    defaultQueueConnection: AppEnv.isProduction
      ? new Cluster(
          [
            {
              host: AppEnv.string('BG_JOBS_REDIS_HOST'),
              port: AppEnv.integer('BG_JOBS_REDIS_PORT', { optional: true }) || 6379,
            },
          ],
          {
            slotsRefreshTimeout: 10000,
            dnsLookup: (address, callback) => callback(null, address),
            redisOptions: {
              username: AppEnv.string('BG_JOBS_REDIS_USERNAME'),
              password: AppEnv.string('BG_JOBS_REDIS_PASSWORD'),
              tls: {},
            },
            clusterRetryStrategy: (times: number) => Math.max(Math.min(Math.exp(times), 20000), 1000),
            enableOfflineQueue: false,
          }
        )
      : new Redis({
          host: AppEnv.string('BG_JOBS_REDIS_HOST', { optional: true }) || 'localhost',
          port: AppEnv.integer('BG_JOBS_REDIS_PORT', { optional: true }) || 6379,
          username: AppEnv.string('BG_JOBS_REDIS_USERNAME', { optional: true }),
          password: AppEnv.string('BG_JOBS_REDIS_PASSWORD', { optional: true }),
          // tls:  {},
          retryStrategy: (times: number) => Math.max(Math.min(Math.exp(times), 20000), 1000),
          enableOfflineQueue: false,
        }),

    // Only establish the worker Redis connection if on an instance that does the work.
    // This consumer (blocking) connection sets `maxRetriesPerRequest: null` — required by
    // BullMQ, whose Worker/QueueEvents use blocking commands (BLPOP/BRPOPLPUSH) on a
    // duplicated connection and throw unless it is null. Do not change it, and do NOT copy
    // `null` to non-blocking connections (the queue above, the websockets adapter), which
    // should fail fast. https://docs.bullmq.io/guide/connections
    defaultWorkerConnection: !AppEnv.boolean('WORKER_SERVICE')
      ? undefined
      : AppEnv.isProduction
      ? new Cluster(
          [
            {
              host: AppEnv.string('BG_JOBS_REDIS_HOST'),
              port: AppEnv.integer('BG_JOBS_REDIS_PORT', { optional: true }) || 6379,
            },
          ],
          {
            slotsRefreshTimeout: 15000,
            dnsLookup: (address, callback) => callback(null, address),
            redisOptions: {
              username: AppEnv.string('BG_JOBS_REDIS_USERNAME'),
              password: AppEnv.string('BG_JOBS_REDIS_PASSWORD'),
              tls: {},
              maxRetriesPerRequest: null,
            },
            clusterRetryStrategy: (times: number) => Math.max(Math.min(Math.exp(times), 20000), 1000),
          }
        )
      : new Redis({
          host: AppEnv.string('BG_JOBS_REDIS_HOST', { optional: true }) || 'localhost',
          port: AppEnv.integer('BG_JOBS_REDIS_PORT', { optional: true }) || 6379,
          username: AppEnv.string('BG_JOBS_REDIS_USERNAME', { optional: true }),
          password: AppEnv.string('BG_JOBS_REDIS_PASSWORD', { optional: true }),
          // tls:  {},
          maxRetriesPerRequest: null,
          retryStrategy: (times: number) => Math.max(Math.min(Math.exp(times), 20000), 1000),
        }),
  })
}

At first glance, this configuration might seem overwhelming to you, but most of this is just basic connection details for the queue and worker redis connections. Since bullmq runs on redis, you must provide redis connections for the queue and worker to leverage. Once the redis connections are set up, the only remaining thing to do is add a configuration for the default worker. Here is that section, isolated for you:

    defaultWorkstream: {
      workerCount: 1,
      concurrency: 10,
    },

With this done, any service in your application extending the ApplicationBackgroundedService or ApplicationScheduledService classes will automatically send their jobs to the queue belonging to the default workstream.

The two options are not what their names suggest:

  • workerCount is how many BullMQ Worker objects this process builds. They are constructed in the same Node process, so each one opens its own blocking Redis connection and all of them share the one event loop.
  • concurrency is how many fetched jobs a single Worker runs at once.

Their product bounds how many jobs are in flight in the process, so size it against the database pool rather than against the CPU. Raising workerCount cannot make CPU-bound JavaScript use additional cores — every inline handler still runs on the same event loop — so CPU parallelism comes from running more worker processes. In particular, do not derive it from os.cpus().length: inside a container that reports the host's topology rather than the task's CPU allocation, so it inflates the in-flight bound while buying no parallelism at all. See BullMQ's parallelism and concurrency guide for the underlying model.

defaultBullMQWorkerOptions — the worker-side counterpart to defaultBullMQQueueOptions — also belongs in this block. In simple mode, a concurrency or connection placed inside it is always overwritten by the workstream's own value, so set concurrency on the workstream itself rather than here.

Redis TLS​

Pass tls: {} (or a full tls.ConnectionOptions object) to the ioredis constructor in both defaultQueueConnection and defaultWorkerConnection to opt into TLS:

new Redis({
  host: AppEnv.string('BG_JOBS_REDIS_HOST'),
  port: AppEnv.integer('BG_JOBS_REDIS_PORT'),
  tls: {},
})

See the ioredis README and Node's tls.connect docs for the option shape and the connection-trust defaults.

Process-level error semantics​

background.work() installs the worker process's uncaughtException and unhandledRejection handlers itself: each logs at error level, runs a bounded (~15s) graceful shutdown — workers:shutdown hooks, then close DB connections, close workers, and quit Redis — and finally calls process.exit(1) so an orchestrator restarts the process. SIGTERM/SIGINT go through the same bounded shutdown, exiting 0 on a clean stop and 1 on a failed or timed-out one. This has a few consequences:

  • Do not add your own process.on('uncaughtException' | 'unhandledRejection') handler in the worker entry point — a second exiting handler would race the framework's. This is a deliberate asymmetry with the web and websocket entry points, which do install their own handlers.
  • workers:shutdown is the only lifecycle hook on the workers plugin — there is no job:failed / workers:error hook, by design (workers is a thin wrapper over BullMQ, and Psychic doesn't duplicate BullMQ's own event surface). Wire job-failure observability directly on the BullMQ objects once background.work() creates them: background.workers.forEach(w => w.on('failed' | 'error' | 'stalled', ...)) and background.queues.forEach(q => q.on('error', ...)). To feed a fatal-error pipeline (Sentry, Datadog, etc.), read from logging or the orchestrator's restart signal — there is no fatal-error hook.
  • A job whose class no longer resolves by global name fails loud, throwing a typed NoClassForSpecifiedGlobalName (exported from @rvoh/psychic-workers's errors module) and landing in BullMQ's failed set. This is the designed signal to clean up a stale repeating scheduler after renaming or removing a backgroundable class. A model-instance job whose record was legitimately deleted still completes quietly — that case is find-and-return, not a resolution failure.

Redis is a trust boundary equal to the app process. Dispatch reads { globalName, method, args } off the job payload and invokes the resolved class method with no allow-list — anyone who can write to the jobs Redis instance can run any registered method with full application privileges. Lock down write access to the jobs Redis instance accordingly.

Native BullMQ mode​

Native mode is for apps that need to hand BullMQ its own options per queue — a distinct Redis instance or cluster node per queue, or BullMQ Pro group settings the workstream shape above doesn't express. Queues are declared by name, and workers are declared separately against those names:

workersApp.set('background', {
  defaultQueueConnection: bookingRedis,
  defaultWorkerConnection: !AppEnv.boolean('WORKER_SERVICE') ? undefined : bookingWorkerRedis,

  nativeBullMQ: {
    defaultQueueOptions: {
      defaultJobOptions: { attempts: 20, backoff: { type: 'exponential', delay: 1000 } },
    },
    defaultWorkerCount: 1,
    defaultWorkerOptions: { concurrency: 10 },

    namedQueueOptions: {
      BookingNotifications: {
        queueConnection: notificationsRedis,
        workerConnection: notificationsWorkerRedis,
      },
    },

    namedQueueWorkers: {
      BookingNotifications: { workerCount: 1, concurrency: 10 },
    },
  },
})

Run pnpm psy sync after changing the queue names, then route a service to one with queue instead of workstream:

export class BookingNotificationService extends ApplicationBackgroundedService {
  public static get backgroundJobConfig(): BackgroundJobConfig<ApplicationBackgroundedService> {
    return { queue: 'BookingNotifications' }
  }
}

A few sharp edges specific to this mode:

  • A queue in namedQueueOptions with no matching key in namedQueueWorkers gets zero workers. The queue is created and accepts jobs; nothing ever works them. There is no warning and no error — jobs simply accumulate in Redis. Give every named queue an entry in both maps.
  • A named queue's defaultJobOptions replaces the app-wide bag rather than merging with it. The two option objects are combined with one shallow spread, so a queue that sets defaultJobOptions: { attempts: 3 } drops the app-wide backoff, removeOnComplete, and removeOnFail entirely. Restate every key you still want.
  • There is no concurrency default here. Simple mode always writes the workstream's concurrency (10 when the workstream omits it), which is why it overrides defaultBullMQWorkerOptions. Native mode writes none, so BullMQ's own default applies unless you set it.
  • A named queue's worker connection goes on its namedQueueOptions entry, not its namedQueueWorkers entry. A connection set on the worker entry typechecks (it's a plain WorkerOptions key) but is ignored. Each connection falls back to nativeBullMQ.defaultQueueOptions/defaultWorkerOptions and then to the app-wide default. Because pnpm psy sync connects to background to generate types, a missing queue connection surfaces at sync time, not only at boot.

See the BullMQ Queue and Worker option references for the full shape of QueueOptions and WorkerOptions — native mode passes these through largely unmodified.

Transitional workstreams​

Moving background work to a different Redis instance strands whatever the old one still holds — repoint the app's connections and nothing is attached to work those jobs. transitionalWorkstreams describes the legacy topology alongside the current one (the same defaultWorkstream / namedWorkstreams shape, with its own connections) so Psychic builds queues and workers for both:

workersApp.set('background', {
  defaultQueueConnection: bookingRedis,
  defaultWorkerConnection: bookingRedis,
  namedWorkstreams: [{ name: 'BookingReminders' }],

  transitionalWorkstreams: {
    defaultQueueConnection: legacyRedis,
    defaultWorkerConnection: legacyRedis,
    namedWorkstreams: [{ name: 'BookingReminders' }],
  },
})

Workers attach to the legacy queues and work them down, while enqueueing reaches only the top-level workstreams, so every new job lands on the new instance. The old side can only drain, never be added to, which is what makes the cutover finish. Delete the transitionalWorkstreams key once those queues are empty.

Named workstreams​

If your app demands multiple queues, you will want to leverage multiple workstreams. This will enable you to segment off groups of workers to work on one queue, and one group to work on the other. To do this, you can add a named workstream to the conf/initializers/workers.ts file:

    defaultWorkstream: {
      workerCount: 1,
      concurrency: 10,
    },
    namedWorkstreams: [
      {
        name: 'FileImport',
        workerCount: 1,
      },
    ]

With a named workstream added to your config, re-run pnpm psy sync to compile the latest psychic types. Once done, you will want to also adjust the background configuration for the services you wish to tap into this queue:

export default class FileImporter extends ApplicationBackgroundedService {
  public static get backgroundJobConfig(): BackgroundJobConfig<ApplicationBackgroundedService> {
    return { priority: 'not_urgent', workstream: 'FileImport' }
  }

  ...
}

Inspecting queues outside a booted server​

background.connect() is wired to the server:init:after-routes hook. In a pnpm console session, a one-off script, or any other process that initializes the Psychic app without starting the server, nothing has connected, so background.queues is an empty array — no error, no warning, and an inspection or maintenance script reports success having done nothing.

Connect explicitly first. connect() defaults to activateWorkers: false, so it opens the producer connections and builds the Queue objects without turning the process into a worker:

import { background } from '@rvoh/psychic-workers'

background.connect()

for (const queue of background.queues) {
  console.log(queue.name, await queue.getJobCounts())
}

This applies to reading background.queues; enqueueing and scheduling connect on their own. Read the queue name off the Queue object, as above, rather than hardcoding it or deriving it from Background.defaultQueueName.

test invocation​

By default, anything you push to the background will be immediately invoked in tests. You may find this behavior to be undesirable, preferring to actually push background work to bullmq in your app. In these cases, we provide special utilities that you can use during tests to help you manually simulate the role of a worker, enabling you to manually run jobs whenever you see fit.

To enable this behavior, you must set the testInvocation to manual, like so:

workersApp.set('testInvocation', 'manual')

Now you will be able to manually work off your queues using the WorkerTestUtils, exported from this package. For more information on testing with workers, see the testing guides.