Jobs

Jobs represent explicit work to do. Use a job when the code says "do this work" and one handler owns the work: send an email, process an import, sync a record, generate a report, or call a slow third-party API.

Beignet jobs are typed definitions. Dispatchers decide whether they run inline, in tests, or through a durable provider such as BullMQ or Inngest.

bun add @beignet/core

Define a job

Create the app-bound defineJob builder once in lib/jobs.ts with createJobs<AppContext>() (see app-bound builders), then define jobs in feature files:

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

export const SendWelcomeEmailJob = defineJob("mail.welcome", {
  payload: z.object({
    email: z.string().email(),
  }),
  retry: retry.exponential({
    attempts: 3,
  }),
  async handle({ payload, ctx }) {
    await ctx.ports.mailer.send({
      to: payload.email,
      subject: "Welcome",
      text: "Thanks for joining.",
    });
  },
});

The payload schema is validated before the job is dispatched and before durable worker execution calls handle(...).

Dispatch jobs

Use cases dispatch jobs through ctx.ports.jobs:

await ctx.ports.jobs.dispatch(SendWelcomeEmailJob, {
  email: user.email,
});

That port can be an inline dispatcher in local development and tests, or a durable provider in production.

Inline dispatcher

Use the inline dispatcher when the work should run immediately in the same process:

import { createInlineJobDispatcher } from "@beignet/core/jobs";

const jobs = createInlineJobDispatcher<AppContext>({
  ctx,
  onError(error, job) {
    ctx.ports.logger.error("Job failed", {
      error,
      jobName: job.name,
    });
  },
});

Inline dispatch honors the job's declared retry policy in-process: a failed handler retries with the policy's delays before the dispatch rejects. Jobs without a retry policy run once, and payload validation failures never retry. Pass sleep to replace the real backoff delays in tests, or retry: false when another layer owns execution retries for every dispatch through the dispatcher.

Retry policies run in exactly one layer. When an outbox drain delivers a job through the inline dispatcher, the drain runs the handler once per pass and reschedules failures itself using the job's policy — the inline retry loop does not stack on top. No configuration is needed; the drain detects the inline dispatcher automatically.

Job timeouts

Use timeout when a job attempt should fail after a bounded execution window:

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

export const GenerateReportJob = defineJob("reports.generate", {
  payload: z.object({
    reportId: z.string(),
  }),
  timeout: "30s",
  retry: retry.exponential({ attempts: 3 }),
  async handle({ payload, ctx, signal }) {
    await ctx.ports.reports.generate(payload.reportId, { signal });
  },
});

Timeouts apply per attempt. When the window expires, Beignet throws JobTimeoutError; retry policies then classify that timeout like any other handler failure. Inline dispatchers and outbox-backed inline drains enforce the timeout directly. createBullMQJobWorker(...) enforces it around the registered handler. createInngestJobFunction(...) maps whole-second timeouts to Inngest's timeouts.finish setting and fails fast for millisecond-precision timeouts Inngest cannot honor.

The signal argument is cooperative cancellation. Beignet aborts it when the timeout fires, but JavaScript cannot forcibly stop work that ignores the signal, so pass it into cancellable provider calls where possible.

Job hooks

Use hooks for app-owned behavior that should wrap every handler attempt: logging, tracing, tenant setup, per-job leases, or rate-limit checks. Hooks receive the same parsed payload, context, job definition, and timeout signal as the handler. Runner hooks wrap job-local hooks.

import type { JobDef, JobHook, StandardSchema } from "@beignet/core/jobs";
import { retry } from "@beignet/core/jobs";
import { z } from "zod";
import type { AppContext } from "@/app-context";
import { defineJob } from "@/lib/jobs";

export const logJobAttempts: JobHook<
  JobDef<string, StandardSchema, AppContext>,
  AppContext
> = async ({ job, ctx, attempt, maxAttempts }, next) => {
  ctx.ports.logger.info("Job attempt started", {
    jobName: job.name,
    attempt: attempt ?? null,
    maxAttempts: maxAttempts ?? null,
  });

  await next();

  ctx.ports.logger.info("Job attempt completed", {
    jobName: job.name,
    attempt: attempt ?? null,
    maxAttempts: maxAttempts ?? null,
  });
};

export const GenerateReportJob = defineJob("reports.generate", {
  payload: z.object({
    reportId: z.string(),
  }),
  timeout: "30s",
  retry: retry.exponential({ attempts: 3 }),
  hooks: [logJobAttempts],
  async handle({ payload, ctx, signal }) {
    await ctx.ports.reports.generate(payload.reportId, { signal });
  },
});

Global hooks can be installed on execution runners:

createInlineJobDispatcher<AppContext>({
  ctx,
  hooks: [logJobAttempts],
});

The same hooks option is available on createBullMQJobWorker(...) and createInngestJobFunction(...). Hooks run inside the job timeout, and errors thrown by hooks are classified by the same retry policy as handler errors. A hook may skip next() to short-circuit the handler; that counts as a successful attempt. When a runner reports attempt metadata, Beignet forwards it to hooks with the same one-based attempt convention used by retry policies. Direct job.handle(...) calls bypass hooks, so tests that need hook behavior should dispatch the job or call runJobHandler(...).

Execution lease hooks

Use createJobExecutionLeaseHook(...) when one handler attempt should run at a time for a logical job key. This is execution-time coordination: it does not replace unique, which suppresses duplicate dispatches before work is queued.

import {
  createJobExecutionLeaseHook,
  type JobDef,
  retry,
} from "@beignet/core/jobs";
import { z } from "zod";
import type { AppContext } from "@/app-context";
import { defineJob, logJobAttempts } from "@/lib/jobs";

const reportPayloadSchema = z.object({
  reportId: z.string(),
  workspaceId: z.string(),
});

const reportExecutionLease = createJobExecutionLeaseHook<
  JobDef<"reports.generate", typeof reportPayloadSchema, AppContext>,
  AppContext
>({
  locks: ({ ctx }) => ctx.ports.locks,
  key: ({ payload }) => payload.workspaceId,
  ttl: "5m",
  onUnavailable: "skip",
});

export const GenerateReportJob = defineJob("reports.generate", {
  payload: reportPayloadSchema,
  timeout: "30s",
  retry: retry.exponential({ attempts: 3 }),
  hooks: [logJobAttempts, reportExecutionLease],
  async handle({ payload, ctx, signal }) {
    await ctx.ports.reports.generate(payload.reportId, { signal });
  },
});

The helper performs one bounded LocksPort.acquire(...) call per attempt. It does not start a renewal loop or background worker, so it is safe for serverless entrypoints when locks is backed by shared storage such as the Redis locks provider. Release is best effort; ttl is the real safety boundary if the runtime is frozen or terminated before finally runs.

onUnavailable defaults to "skip", which treats an overlapping attempt as successful without running the handler. Use "throw" to raise JobExecutionLeaseUnavailableError and let the retry policy decide whether to try again, or pass a function to log and optionally throw your own error.

Unique jobs

Use unique when duplicate dispatches of the same logical job should collapse for a bounded time window. The uniqueness guard is dispatch-time coordination: it prevents another enqueue while the lease is active, but it does not replace handler idempotency for provider retries or worker crashes.

import {
  createInlineJobDispatcher,
  createUniqueJobDispatcher,
  retry,
} from "@beignet/core/jobs";
import { z } from "zod";
import type { AppContext } from "@/app-context";
import { defineJob } from "@/lib/jobs";

const syncAccountPayloadSchema = z.object({
  accountId: z.string().min(1),
});

export const SyncAccountJob = defineJob("billing.sync-account", {
  payload: syncAccountPayloadSchema,
  unique: ({ payload }) => ({
    key: payload.accountId,
    ttl: "10m",
  }),
  timeout: "30s",
  retry: retry.exponential({ attempts: 3 }),
  async handle({ payload, ctx, signal }) {
    await ctx.ports.billing.syncAccount(payload.accountId, { signal });
  },
});

export function createJobsPort(args: {
  ports: Pick<AppContext["ports"], "locks">;
  createBackgroundContext: () => AppContext;
}) {
  const baseJobs = createInlineJobDispatcher<AppContext>({
    ctx: args.createBackgroundContext,
  });

  return createUniqueJobDispatcher({
    jobs: baseJobs,
    locks: args.ports.locks,
  });
}

The concrete lock key is namespaced as jobs:unique:<job-name>:<unique-key>. A successful dispatch keeps the lease until ttl expires; a failed dispatch releases it so the caller can retry. Wrap direct BullMQ or Inngest dispatchers the same way when an app wants provider-backed dispatch plus Beignet-owned uniqueness.

Wrap root dispatchers for unique durable jobs. If you wrap a transaction-scoped outbox dispatcher, the lease is acquired before the transaction commits, so a rollback can still suppress duplicates until the TTL expires.

Durable dispatch with Inngest

Install the Inngest jobs provider when production jobs should be queued outside the request process:

bun add @beignet/provider-jobs-inngest @beignet/core inngest

import { createInngestJobsProvider } from "@beignet/provider-jobs-inngest";

export const providers = [createInngestJobsProvider()];

createInngestJobsProvider(options) sets the app name and event key in code; options override env-derived values.

The provider adapts Inngest's durable execution platform into Beignet's JobDispatcherPort. It installs ctx.ports.jobs and exposes ctx.ports.inngest.client as an escape hatch for Inngest-specific features. It does not replace Beignet domain events; use listeners to turn domain facts into Inngest-backed jobs.

Workers are defined separately from your Beignet HTTP server:

// app/api/inngest/route.ts
import { createInngestJobFunction } from "@beignet/provider-jobs-inngest";
import { serve } from "inngest/next";
import { SendWelcomeEmailJob } from "@/features/users/jobs";
import { getServer } from "@/server";

const server = await getServer();

const sendWelcomeEmail = createInngestJobFunction({
  client: server.ports.inngest.client,
  job: SendWelcomeEmailJob,
  ctx: () => server.createServiceContext(),
  instrumentation: server.ports,
});

export const { GET, POST, PUT } = serve({
  client: server.ports.inngest.client,
  functions: [sendWelcomeEmail],
});

server.createServiceContext() is app-owned through the server context blueprint. Passing server.ports as the instrumentation target lets Beignet start the job span before resolving that lazy context, so tracing, logging, and provider instrumentation share the active execution.

See Runtime recipes for how Beignet separates provider adapters from serverless-safe worker entrypoints. Direct provider jobs run through provider-owned entrypoints such as Inngest functions; outbox-backed jobs run through beignet outbox drain.

When a job defines a retry policy, the Inngest helper maps the total attempt count to Inngest's function retry setting, and createInngestJobFunction(...) fails fast if a job policy includes custom backoff, jitter, or retryIf behavior that Inngest cannot honor. Whole-second job timeouts map to Inngest timeouts.finish; sub-second timeouts fail fast because Inngest function timeouts cannot honor them exactly.

Durable workers with BullMQ

Install the BullMQ provider when production jobs should run through a Redis-backed queue that your app owns:

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

import { createBullMQJobsProvider } from "@beignet/provider-jobs-bullmq";

export const providers = [createBullMQJobsProvider()];

The provider installs ctx.ports.jobs and exposes ctx.ports.bullMQJobs.queue as an escape hatch for BullMQ-specific operations.

Run workers from an explicit worker process, CLI entrypoint, or provider-owned route:

// server/workers/jobs.ts
import { createBullMQJobWorker } from "@beignet/provider-jobs-bullmq";
import { SendWelcomeEmailJob } from "@/features/users/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: [SendWelcomeEmailJob],
  ctx: () => server.createServiceContext(),
  instrumentation: server.ports,
  workerOptions: {
    concurrency: 5,
  },
  errorReporter: ({ ctx }) => ctx.ports.errorReporter,
});

BullMQ workers are at-least-once. Put idempotency inside handlers when duplicate execution would create a side effect. Use jobsWorker.close() from process shutdown handling so BullMQ can stop claiming new jobs and wait for active jobs to finish. The worker helper defaults to the same "beignet" Redis key prefix as the provider; pass prefix when BULLMQ_PREFIX changes.

Register the app-owned OpenTelemetry SDK in this standalone process before initializing the server. Passing server.ports starts the job span before the lazy service context is created. See Observability for the shared bootstrap pattern.

The BullMQ provider maps Beignet fixed and exponential retry attempts to BullMQ attempts/backoff. retryIf is honored by createBullMQJobWorker(...) with BullMQ unrecoverable failures. Retry fields BullMQ cannot honor exactly, such as maxDelay, custom exponential factor, and boolean jitter, fail fast. createBullMQJobWorker(...) also enforces Beignet job timeouts around registered handlers and reports timeout failures through the same retry/dead-letter classification path.

Use the provider escape hatch in health or readiness routes when the deployment needs to prove the direct jobs queue is reachable:

const health = await ctx.ports.bullMQJobs.checkHealth();

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

For direct BullMQ jobs, Beignet instrumentation records terminal worker failures as deadLettered; BullMQ owns the concrete failed-job set. Use the outbox instead when the application database must contain durable retry and dead-letter rows for the side effect.

Retry policy

Use the retry helpers to make durable failure behavior explicit:

import { retry } from "@beignet/core/jobs";

class TemporaryProviderError extends Error {}

retry.none();

retry.fixed({
  attempts: 3,
  delay: "30s",
});

retry.exponential({
  attempts: 5,
  initialDelay: "10s",
  maxDelay: "10m",
  jitter: true,
  retryIf: ({ error }) => error instanceof TemporaryProviderError,
});

attempts is the maximum total attempts, including the first attempt. Use retryIf for app-owned transient/permanent error classification.

Retry vocabulary

Beignet uses the same retry language for jobs, outbox-backed delivery, and scheduled work:

Jobs and transactions

Avoid dispatching durable side effects before the database work commits. When a workflow uses Unit of Work, record a domain event during the transaction and let a listener dispatch the job after commit — see side effects after commit for the rule.

Use Outbox when the job enqueue must commit with the database write. The outbox can sit behind a transaction-scoped tx.jobs dispatcher, then a worker drains the durable row into your production job provider. beignet make job registers new feature job registries in an existing server/outbox.ts, and beignet doctor warns when a feature job is missing from the outbox registry (--fix registers it).

Retry-safe jobs

Job providers may retry handlers after process failures, timeouts, or transient errors. Put idempotency inside the job handler when the handler owns work that must not happen twice:

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

export const GenerateReportJob = defineJob("reports.generate", {
  payload: z.object({
    reportId: z.string(),
    requestedBy: z.string(),
  }),
  retry: retry.exponential({ attempts: 3 }),
  async handle({ payload, ctx }) {
    await runIdempotently(ctx.ports.idempotency, {
      namespace: "reports.generate",
      key: payload.reportId,
      scope: { actorId: payload.requestedBy },
      fingerprint: await createIdempotencyFingerprint(payload),
      ttlSec: 60 * 60 * 24,
      run: () => ctx.ports.reports.generate(payload.reportId),
    });
  },
});

Use Idempotency for the full command, webhook, and job pattern.

Testing

In use-case tests, pass a job dispatcher that records dispatches:

const dispatchedJobs: Array<{ name: string; payload: unknown }> = [];

const jobs = {
  dispatch: async (job, payload) => {
    dispatchedJobs.push({ name: job.name, payload });
  },
};

In job tests, call the job handler directly with an in-memory context:

await SendWelcomeEmailJob.handle({
  job: SendWelcomeEmailJob,
  payload: { email: "user@example.com" },
  ctx,
});

Where jobs fit

Workflow primitives gives the full decision guide for commands, events, jobs, schedules, notifications, idempotency keys, and outbox records, and the workflows overview shows the transition pattern that decides when jobs should be dispatched.