Defining Jobs

A job pairs a unique name with a handle(payload, ctx) function. defineJob registers it in a process-wide registry and returns a typed QueueJob you dispatch from anywhere.

src/app/invoices/jobs/send-invoice.job.ts

import { defineJob } from "@warlock.js/queue";

export const sendInvoice = defineJob({
  name: "invoices.send", // unique across the app; duplicate names replace the handler
  queue: "invoices", // default: config.defaultQueue ("default")
  attempts: 5, // total tries including the first; default 1 (no retry)
  backoff: { type: "exponential", delay: 2000 }, // or a plain number = fixed delay
  removeOnComplete: true, // retention for completed jobs; default { age: "24h", count: 1000 }
  removeOnFail: 100, // keep the newest 100 failed jobs; default { age: "7d", count: 5000 }
  async handle(payload: { invoiceId: string }, ctx) {
    await ctx.log(`attempt ${ctx.attempt} of ${ctx.maxAttempts}`);
    await ctx.progress(50);
    // throw to fail this attempt; it retries until attempts run out
    return { sent: true }; // stored as the job result
  },
});

The module must be imported by whatever process runs the workers — in a Warlock app, anything under src/app loaded at boot, since queueConnector() starts in the late phase after that import.

Retention

Finished jobs are pruned out of the box, so Redis or the SQLite file does not grow with every job ever run:

Option	Default	false
removeOnComplete	{ age: "24h", count: 1000 }	keeps every completed job
removeOnFail	{ age: "7d", count: 5000 }	keeps every failed job

A retention is true (remove at once), false (keep all), a number (keep the newest N), or { age?, count? }: remove jobs finished longer than age ago (milliseconds or a string like "12h") and keep at most the newest count. Set it per job or app-wide in queue.defaultJobOptions:

src/config/queue.ts

defaultJobOptions: {
  removeOnComplete: { age: "6h", count: 500 },
  removeOnFail: false, // keep every failure for inspection
},

On BullMQ, age-based pruning runs when another job of the same kind finishes.

defineJob validates at call time: an empty name, a missing handle, or a non-integer attempts < 1 throws InvalidJobDefinitionError immediately, not at dispatch time.

Job context

The handler’s second argument:

Field	Type	Notes
id	string	The job id
name	string	The name given to defineJob
queue	string	The queue this job runs on
attempt	number	Current attempt, starting at 1
maxAttempts	number	Total attempts allowed
progress(value)	Promise<void>	Record progress — a number or a JSON object
log(line)	Promise<void>	Append a line to the job’s log

Carry request context into jobs

Register a context once during application setup to carry the current tenant from dispatch into the worker handler.

import { defineQueueContext, setQueueContext } from "@warlock.js/queue";

setQueueContext(
  defineQueueContext({
    capture: () => tenantStorage.getStore(), // runs inside dispatch()
    restore: (tenant, run) => tenantStorage.run(tenant, run), // wraps handle()
  }),
);

• undefined from capture carries nothing; the value must be JSON-safe. Read it in a handler as ctx.context (unknown).
• Jobs dispatched from a restored handler capture the same context. Set context: false in defineJob() to opt one job out; such a job never runs inside a restored context.
• The context comes only from what capture returned at dispatch. A payload that looks like the queue’s internal envelope (for example a request body passed straight to dispatch()) is delivered to the handler as plain data and never sets the context.
• Throw UnrecoverableJobError from a handler or restore to fail a job without retries.

Dispatch

const { id } = await sendInvoice.dispatch({ invoiceId: "42" });

await sendInvoice.dispatch(
  { invoiceId: "43" },
  {
    delay: "10m", // a number of ms, or "500ms" | "30s" | "10m" | "2h" | "1d"
    priority: 1, // 1 runs first; larger numbers later; omitted runs ahead of any prioritized job
    jobId: "invoice:43", // dispatching again with a still-live id is a no-op — idempotent dispatch
    attempts: 3, // overrides defineJob's attempts for this call only
    backoff: { type: "fixed", delay: 500 },
    timeout: "30s", // overrides defineJob's per-attempt time limit — see Retry and failed jobs
  },
);

Option precedence for attempts / backoff / timeout: dispatch options > defineJob > queue.defaultJobOptions.

Read a job

const job = await sendInvoice.find(id);
// JobSnapshot | undefined:
// { id, name, queue, state, payload, progress, attemptsMade, result, failedReason, createdAt, finishedAt }

state is one of "waiting" | "delayed" | "prioritized" | "active" | "completed" | "failed" | "waiting-children" | "unknown".

Test what was dispatched

In the test environment @warlock.js/core/tests installs queueFake: dispatches are recorded instead of pushed to Redis (no broker needed), and find() answers from the record.

import { queueFake } from "@warlock.js/core/tests";

await placeOrder(order);

queueFake.assertPushed(sendInvoice, { invoiceId: order.invoiceId });
queueFake.assertPushedCount(1);

Jobs are recorded, not run: call the handler directly to test what it does. Opt out with tests.fakes: { queue: false } in src/config/tests.ts. Other fakes can use the same seam: setDispatchInterceptor({ dispatch, find? }) replaces the broker for every job in the process until it is set back to undefined.

Rules

• A dispatched job whose name has no handler registered in the worker process fails immediately, with no retries.
• Payloads are stored as JSON in Redis: pass ids, not model instances.
• Defining the same name again replaces the handler for that name — this is what keeps dev-server reloads working — so keep names unique in production code.