@kilnjs/jobs

Background queues for KilnJS, as a module core is given rather than one it contains. Built on BullMQ, over the Redis connection core already holds.

One file in the jobs directory is one queue. Nothing is registered, no queue is constructed, and there is no barrel to keep in step - a new queue is a new file, the way a new route is.

Install

yarn add @kilnjs/jobs bullmq

import { jobs } from "@kilnjs/jobs";

export default {
  modules: [jobs()],
} satisfies AppConfig;

Declaring it is what makes the files in src/jobs run. Leave it out and they are never loaded: nothing is queued and nothing is processed, which is what a server with no background work should cost.

The directory is this package's too - it merges jobs into core's segments, so paths.jobs exists in app.config.ts exactly as long as this package is installed. Core loads the routes and the seeds and knows nothing of queues.

What jobs() takes

Every option is a default for all of them: whatever a queue declares of its own wins, a field at a time. Each has an environment variable behind it, so a deployment can set one without a build.

jobs({
  workers: process.env.ROLE === "worker",

  concurrency: 4,
  limiter: { max: 100, duration: 1000 },

  prefix: "kiln",
  defaults: { attempts: 3, removeOnComplete: 100 },

  onFailure: (queue, job, error) => alert(`${queue} failed: ${error.message}`),
});

| | | | | --- | --- | --- | | workers | whether this process runs them | JOBS_WORKERS | | prefix | what every key in Redis sits under | JOBS_PREFIX | | concurrency | for a queue that declared none | JOBS_CONCURRENCY | | limiter | a rate limit for every worker | | | lockDuration | how long a job may run before its lock is renewed | JOBS_LOCK_DURATION_MS | | maxStalledCount | times a job may stall before it is failed | JOBS_MAX_STALLED | | defaults | job options, under each queue's own | | | onFailure onCompleted onStalled onError | the same hooks a queue declares, for all of them, given the queue's name first | |

workers is how one codebase becomes two processes. The API enqueues and runs nothing; the worker runs everything and answers no requests. Both read the same directory, so a queue cannot exist on one and not the other. It is false in a test run unless something says otherwise, which is why a suite can assert on what was enqueued.

prefix is what lets two deployments share a Redis: BullMQ's own default is bull, and two applications under it would consume each other's jobs.

A queue

// src/jobs/mail/welcome.ts
import type { Queue } from "@kilnjs/jobs";

const queue: Queue<{ userId: string }> = async ({ userId }) => {
  await sendWelcome(userId);
};

queue.concurrency = 5;

queue.onFailure = (job, error) => {
  console.warn(`Welcome mail for ${job?.data.userId} failed: ${error.message}`);
};


export default queue;

The function is what runs. Everything else it needs is on the function, not passed to a constructor somewhere else - the same shape an endpoint has, where handler.authorize sits on the handler.

It is handed the payload, and the job second. The job carries the attempt count, the id and the timestamps, and almost nothing reads them:

const queue: Queue<{ userId: string }> = async ({ userId }, job) => {
  if (job.attemptsMade > 1) console.warn(`Retrying ${userId}`);
};

The file's path is the queue's name, so this one is mail/welcome. Two queues can share a filename without sharing a queue.

| | | | --- | --- | | concurrency | how many jobs the worker takes at once. One by default | | limiter | at most max jobs per duration milliseconds | | lockDuration | how long a job may run before its lock is renewed | | maxStalledCount | times a job may stall before it is failed for good | | schedule | a cron pattern, or the milliseconds between runs | | timezone | which clock a pattern is read against. The server's by default | | defaults | applied to every job put on this queue | | onFailure | after a job threw. The job is absent when the failure was in taking it | | onCompleted | after a job returned | | onStalled | a job whose worker stopped reporting, usually killed mid-run | | onError | the worker itself, rather than one job: a lost connection, say | | title | the name, written by discovery rather than declared |

A module can add to that list by augmenting QueueExtensions, which is how something per-queue reaches a queue without this package knowing about it.

Putting a job on one

Pass the queue itself, not its name:

import { enqueueJob } from "@kilnjs/jobs";

import welcome from "@/jobs/mail/welcome";

await enqueueJob(welcome, { userId: user.id });

The payload is checked against what that queue declared, and a queue that takes none is enqueued with nothing: enqueueJob(prune). A function nothing discovered is refused rather than queued somewhere nothing reads.

Taking the function rather than a name is what makes the payload typed and a renamed queue a compile error instead of a job nothing consumes.

It is also on the server, for a route that wants it: server.jobs.enqueue, beside server.jobs.queues - the names of everything discovered.

Scheduling

queue.schedule = "0 * * * *";   // hourly, cron
queue.schedule = 60_000;        // or every minute

Upserted under one key per queue, so a restart replaces the schedule rather than leaving the old one running beside it.

A schedule that cannot mean anything is refused at boot, naming the queue: an empty pattern, an interval of zero or a negative one, and a cron expression BullMQ will not parse. Each of those otherwise fails inside BullMQ in a message with no idea which file it came from.

What a test asserts against

No worker runs in a test run, so nothing consumes what a queue is given. listEnqueuedJobs() records what was put on one, which mirrors listSentMail() in @kilnjs/mailer:

await archiveConversation(id);

expect(listEnqueuedJobs(deliverReply)).toEqual([]);

With no argument it answers every job this process enqueued, in order. clearEnqueuedJobs() drops the record, and resetQueues() does too. Nothing is recorded outside a test run, so nothing grows per job in production.

Before this, asserting that an action enqueues nothing meant reading BullMQ's own Redis keys (bull:<queue>:wait, then the job hash's data), which couples a suite to a storage layout nobody chose.

What it does for you

A failure is never silent. Every one is reported to telemetry whether or not the queue declared onFailure, so a hook is for what only that queue knows to do - not for making sure anyone hears about it. A hook that throws is reported too, and does not take the worker down.

A job in flight finishes. The module drains on shutdown, after the server has stopped accepting requests, so a worker is never killed mid-job. The drain runs on any close rather than only on a signal, and it finishes: a worker whose connection refuses to close is reported, and the rest are still released.

No queue runs in a test suite. The directory is still discovered in a test run - a file that exports no queue fails the boot rather than the next deploy - but no worker is started, because a suite whose workers picked up its own jobs would race with whatever it is asserting.

Nothing connects until it has to. Discovery imports the files and names the queues; the BullMQ queue behind one is built on first use.

Development

yarn verify   # ts:check, lint, build, exports:check, test

The tests discover a stand-in application under tests/application and reach no Redis.