@beignet/provider-jobs-bullmq

Runtime: Beignet requires Node.js 22.12 or newer. Bun is optional.

BullMQ-backed JobDispatcherPort provider for Beignet applications.

The provider installs the app-facing ctx.ports.jobs dispatcher for durable Redis-backed background jobs using BullMQ. Worker execution is explicit: call createBullMQJobWorker(...) from a worker process, CLI entrypoint, or provider-owned route. Do not start workers from provider lifecycle hooks in serverless apps.

createBullMQJobsProvider(...) returns the stable BullMQJobsProvider type. BullMQJobsConfig describes its validated config; the Zod schema remains internal.

Install

bun add @beignet/provider-jobs-bullmq @beignet/core bullmq
Copy

Setup

import { definePorts } from "@beignet/core/ports";
import { createNextServer, createNextServerLoader } from "@beignet/next";
import { createBullMQJobsProvider } from "@beignet/provider-jobs-bullmq";
import { routes } from "@/server/routes";

// Set environment variables:
// BULLMQ_REDIS_URL=redis://localhost:6379/0
// BULLMQ_QUEUE_NAME=beignet-jobs
// BULLMQ_PREFIX=beignet

const initialPorts = definePorts({});

export const getServer = createNextServerLoader(() =>
  createNextServer({
    ports: initialPorts,
    providers: [createBullMQJobsProvider()],
    context: ({ ports }) => ({ ports }),
    routes,
  }),
);
Copy

Dispatch jobs

Define jobs with @beignet/core/jobs, then dispatch through the stable jobs port:

import { retry } from "@beignet/core/jobs";
import { z } from "zod";
import { defineJob } from "@/lib/jobs";

export const SendInviteEmailJob = defineJob("mail.invite.send", {
  payload: z.object({
    inviteId: z.string(),
    email: z.string().email(),
  }),
  retry: retry.fixed({
    attempts: 3,
    delay: "30s",
  }),
  timeout: "30s",
  async handle({ payload, ctx }) {
    await ctx.ports.mailer.send({
      to: payload.email,
      subject: "You were invited",
      text: `Invite id: ${payload.inviteId}`,
    });
  },
});

await ctx.ports.jobs.dispatch(SendInviteEmailJob, {
  inviteId: invite.id,
  email: invite.email,
});
Copy

The dispatcher validates the payload before enqueueing. BullMQ stores the Beignet job name as the BullMQ job name. When tracing is active, Beignet wraps the original, validated payload in a versioned transport envelope; the worker unwraps it before its execution-time payload parse and continues the producer trace. This preserves early rejection while applying transforming schemas exactly once before the handler. Legacy jobs keep their raw payload shape and still execute. Malformed trace metadata is ignored without dropping the payload. Code using the raw BullMQJob escape hatch should treat job.data as provider transport data rather than application payload.

createBullMQJobWorker(...) enforces Beignet timeout declarations around registered handlers. A timeout throws JobTimeoutError, aborts the handler's signal, and then flows through the same retry/dead-letter classification as other handler failures. BullMQ still owns process recovery and the concrete failed-job set.

createBullMQJobWorker(...) also accepts hooks. Worker hooks wrap job-local hooks around every handler attempt, run inside the job timeout, and fail through the same retry classification path as handler errors.

Unique jobs

BullMQ dispatchers are ordinary Beignet job dispatchers, so apps can wrap them with createUniqueJobDispatcher(...) from @beignet/core/jobs when a job's unique declaration should suppress duplicate dispatches through ctx.ports.locks:

import { createUniqueJobDispatcher } from "@beignet/core/jobs";
import type { LocksPort } from "@beignet/core/locks";
import {
  createBullMQJobDispatcher,
  type BullMQQueue,
} from "@beignet/provider-jobs-bullmq";

export function createJobsPort(args: {
  queue: BullMQQueue;
  locks: LocksPort;
}) {
  const bullmqJobs = createBullMQJobDispatcher(args.queue);

  return createUniqueJobDispatcher({
    jobs: bullmqJobs,
    locks: args.locks,
  });
}
Copy

The uniqueness lease prevents duplicate enqueue attempts for the configured TTL. BullMQ workers are still at-least-once, so handlers must remain idempotent for retries and process failures.

Run a worker

Workers consume an explicit registry of job definitions. This keeps production workers honest: a queued job that is not registered in the worker fails without retrying instead of silently no-oping.

// server/workers/jobs.ts
import { createBullMQJobWorker } from "@beignet/provider-jobs-bullmq";
import { SendInviteEmailJob } from "@/features/invitations/jobs";
import { getServer } from "@/server";

const server = await getServer();

export const jobsWorker = createBullMQJobWorker({
  queueName: process.env.BULLMQ_QUEUE_NAME ?? "beignet-jobs",
  redisUrl: process.env.BULLMQ_REDIS_URL ?? "redis://localhost:6379/0",
  prefix: process.env.BULLMQ_PREFIX ?? "beignet",
  jobs: [SendInviteEmailJob],
  ctx: () => server.createServiceContext(),
  instrumentation: server.ports,
  workerOptions: {
    concurrency: 5,
  },
  errorReporter: ({ ctx }) => ctx.ports.errorReporter,
  infrastructureErrorReporter: server.ports.errorReporter,
});

const health = await jobsWorker.checkHealth({ timeoutMs: 5_000 });
if (!health.ok) {
  throw new Error(health.error.message);
}

const shutdownDeadline = 25_000;
async function shutdown() {
  const deadline = setTimeout(() => process.exit(1), shutdownDeadline);
  deadline.unref();

  try {
    await jobsWorker.close();
  } finally {
    await server.stop();
    clearTimeout(deadline);
  }
}

process.once("SIGTERM", () => {
  void shutdown().catch((error) => {
    console.error("BullMQ worker shutdown failed", error);
    process.exit(1);
  });
});
Copy

close() delegates to BullMQ's graceful shutdown behavior: the worker stops claiming new jobs and waits for active jobs to finish. Pass { force: true } only when the host cannot wait; unfinished work may later be recovered as stalled. Keep the deployment's hard shutdown deadline outside the provider so the app can stop its database, telemetry, and other resources in order.

When the instrumentation target exposes ports.tracing, the worker starts the job span before evaluating the lazy ctx factory. Service contexts created by that factory therefore inherit the real active job span. Register the app-owned OpenTelemetry SDK in the standalone worker process before initializing the server. The dispatcher captures the active span through the same instrumentation target and carries it across Redis to the worker.

Configuration

The provider reads configuration from environment variables with the BULLMQ_ prefix:

Use createBullMQJobsProvider({ connection }) when you want to provide BullMQ connection options directly instead of parsing BULLMQ_REDIS_URL. createBullMQJobWorker(...) accepts the same Redis URL through redisUrl and defaults its Redis key prefix to "beignet" to match the provider default.

URL-created queue producers fail quickly during Redis outages with the offline queue disabled and one retry per request. URL-created workers use BullMQ's persistent consumer profile with maxRetriesPerRequest: null. Explicit connection and workerOptions.connection objects remain app-owned and are passed through unchanged. Queue connection readiness waits at most 5000ms by default, and failed-startup cleanup is bounded separately by the same timeout; override both bounds with startupTimeoutMs. startupTimeoutMs and health-check timeoutMs values must be integers from 1 through 2,147,483,647 milliseconds, the JavaScript timer range.

Redis URL query parameters are forwarded using IORedis URL semantics. For example, ?family=6&connectTimeout=2500 selects IPv6 and sets the connection timeout. Beignet keeps its URL-created connection safety profile authoritative: producer URLs cannot override enableOfflineQueue: false or maxRetriesPerRequest: 1, and worker URLs cannot override maxRetriesPerRequest: null. Pass an explicit connection object for options that require booleans, nested objects, or callbacks. BullMQ does not support IORedis keyPrefix; Beignet rejects it in Redis URLs and uses the provider or worker prefix option for queue key namespacing instead.

Finalized-job retention

Beignet bounds BullMQ's finalized-job sets by default:

• completed jobs: up to 1000 jobs for 24 hours
• failed jobs: up to 5000 jobs for 7 days

Override those limits with retention, or pass false when defaultJobOptions or per-job jobOptions own removal:

createBullMQJobsProvider({
  retention: {
    completed: { age: 60 * 60, count: 500 },
    failed: { age: 14 * 24 * 60 * 60, count: 20_000 },
  },
});
Copy

BullMQ applies removal lazily when another job finalizes. Removing a job also removes its BullMQ job-ID deduplication history, so do not use retention as the only idempotency boundary for external side effects.

beignet doctor --strict checks that installed BullMQ jobs providers are registered in server/providers.ts and that BULLMQ_REDIS_URL is present in app env examples or config when the env-backed provider is used.

Ports

jobs: JobDispatcherPort

The jobs port is the recommended application API:

await ctx.ports.jobs.dispatch(SendInviteEmailJob, {
  inviteId: invite.id,
  email: invite.email,
});
Copy

bullMQJobs: BullMQJobsPort

The bullMQJobs port is an escape hatch for BullMQ-specific operations:

const counts = await ctx.ports.bullMQJobs.queue.getJobCounts();
Copy

It also exposes a small readiness helper for app-owned health endpoints:

const health = await ctx.ports.bullMQJobs.checkHealth({
  timeoutMs: 2_000,
});

if (!health.ok) {
  return Response.json({ ok: false, jobs: health }, { status: 503 });
}
Copy

checkHealth(...) verifies that the queue connection becomes ready and returns selected BullMQ queue counts. By default it includes waiting, active, delayed, and failed counts. It returns { ok: false } with a short error summary instead of throwing so health routes can map dependency failure to a 503 response.

Use the escape hatch only for operations the stable Beignet jobs port intentionally does not model.

Worker handles expose their own bounded readiness check:

const health = await jobsWorker.checkHealth({ timeoutMs: 2_000 });
Copy

It returns healthy only when the worker is connected, running, and not paused. Queue backlog belongs in metrics and incident views, not readiness status.

Retry semantics

Beignet job attempts means maximum total attempts, including the first try. The BullMQ provider maps:

The provider fails fast for Beignet retry fields BullMQ's built-ins cannot honor exactly: maxDelay, a custom exponential factor, and boolean jitter. createBullMQJobWorker(...) validates every registered job's retry policy at worker creation, and the dispatcher validates again at dispatch, so an unsupported policy fails at construction instead of at first dispatch. Use the Beignet outbox when the app needs Beignet-owned retry delay calculation, custom jitter, or durable dead-letter rows in the application database.

Beignet job timeouts are enforced by createBullMQJobWorker(...), not by the enqueue call. Pass the handler's signal into cancellable provider calls where possible; JavaScript cannot forcibly stop work that ignores the signal.

BullMQ jobs are at-least-once. Put idempotency inside handlers when duplicate execution would create a side effect:

import {
  createIdempotencyFingerprint,
  runIdempotently,
} from "@beignet/core/idempotency";

await runIdempotently(ctx.ports.idempotency, {
  namespace: "mail.invite.send",
  key: payload.inviteId,
  fingerprint: await createIdempotencyFingerprint(payload),
  ttlSec: 60 * 60 * 24,
  run: () => ctx.ports.mailer.send(message),
});
Copy

Devtools and error reporting

When @beignet/devtools or a provider instrumentation port is available, the dispatcher records enqueue attempts as scheduled or failed. Worker helpers record started, completed, retryScheduled, and deadLettered lifecycle events under the jobs watcher.

For direct BullMQ jobs, deadLettered in Beignet instrumentation means the worker classified the failure as terminal and BullMQ should not retry it again. The provider marks non-retryable failures and schema-invalid payloads with BullMQ's UnrecoverableError; BullMQ then owns the concrete failed-job state.

Pass errorReporter to createBullMQJobWorker(...) to capture job failures through ErrorReporterPort. Retryable attempts remain visible as retryScheduled instrumentation and do not create incident reports. Terminal handler failures, payload validation failures, and unknown job names report once. A resolver such as ({ ctx }) => ctx.ports.errorReporter is used after app context exists. Reporter failures are swallowed so BullMQ remains the source of truth for retry and failure state.

Pass infrastructureErrorReporter as the context-free fallback for validation, context-construction, and unknown-job failures when errorReporter is a resolver. It also captures worker-level Redis errors, stalled jobs, and shutdown failures. Beignet selects the primary reporter when it can resolve one and does not intentionally fan a failure out to both reporters.

Failure behavior

The env-backed provider throws during startup when BULLMQ_REDIS_URL is missing or the queue connection does not become ready within the bounded startup timeout. Dispatch validates the Beignet job payload before enqueueing and throws validation or BullMQ errors rather than waiting indefinitely for an offline URL-created producer connection. Workers classify schema-invalid and non-retryable failures as terminal so BullMQ records them as failed jobs without further retries. Timed-out handlers throw JobTimeoutError; retry policies decide whether the timeout retries or becomes terminal.

Local and tests

Use createInlineJobDispatcher(...) or a recording dispatcher in use-case tests. Use BullMQ with a local Redis instance for provider and worker tests that need queue, retry, or shutdown behavior.

Deployment notes

Run workers from explicit long-running processes, containers, or platform worker services. Do not start BullMQ workers from Beignet provider lifecycle hooks or serverless request handlers. Use ctx.ports.bullMQJobs.checkHealth(...) for producer readiness and jobsWorker.checkHealth(...) for worker readiness.

Production Redis must use a non-evicting memory policy and appropriate persistence. BullMQ stores job payloads in Redis, so enqueue identifiers and lookup keys instead of secrets or unnecessary personal data.

Direct jobs vs outbox-backed jobs

Use direct BullMQ jobs when enqueueing after the application workflow commits is acceptable, or when the job is triggered by a workflow that already has its own durability boundary.

Use the Beignet outbox when the job enqueue must commit atomically with a database transaction. The outbox stores durable rows, applies Beignet-owned retry classification and dead-letter state, then dispatches through whatever ctx.ports.jobs provider is installed.