Jobs and workflows

Durable background work, optionally multi-step. Postgres queue by default. idempotencyKey is required by the type. Drivers swap without touching job code.

As of 2026-08. Stable API — semver from here (Upgrading).

Two factories return a job rather than a ninth primitive, so everything on this page applies to both: backfill() (Migrations and backfills) and scrape() (Scraping).

The canonical shape

// job
export const onboardOrg = job({
  input: t.object({ orgId: t.uuid }),
  tenant: ({ orgId }) => orgId,               // the org this run acts as
  idempotencyKey: ({ orgId }) => `onboard:${orgId}`,   // REQUIRED by the type
  retry: { attempts: 5, backoff: 'exponential' },
  async run({ input, step, ctx }) {
    const org = await step.run('provision', () => ctx.orgs.provision(input.orgId));
    await step.run('welcome-email', () => ctx.mail.send(welcomeEmail, org));
    await step.sleep('3d');
    await step.run('nudge', () => ctx.mail.send(nudgeEmail, org));
  },
});

The fluent surface

Every projection is a method on the job — onboardOrg.enqueue({ orgId }), never enqueueJob(onboardOrg, input) — and every declared field is lifted onto it. A job has no .def.

One enqueue implementation. <job>.enqueue, <job>.as and a task's own enqueue() all resolve the same ambient facade; the only other calls into a driver's enqueue are the outbox relay and the scheduler's occurrence dispatch. So "does this join the transaction?" has one answer for every enqueue an app writes, and there is no second path to forget about.

run is on the handle and is still not yours to call. The worker's executeJob is the one execution path, and it owns the attempt counter, the step store, the timeout and the lease — none of which a direct call carries. That is also why .as() queues: on an action .as() runs the mutation as that actor, on a job it enqueues as that actor. Same word, and the difference is the primitive's execution surface, not an inconsistency.

describe().steps is empty by design. Step names are chosen inside run() at execution time, so they are not statically knowable — the steps a run actually recorded come from the run itself, via x jobs show <id> --json. x.manifest.json, the /_x jobs panel and the MCP dev server read one list, and that list is a map over each handle's own describe() — so the list and a single job can never disagree. name is the export name, stamped by defineApi({ jobs: [postJobs] }) — the same call that names actions and queries. A module nobody hands over keeps job()'s positional anonymous-job-<n>, on the queue row and in x.manifest.json. A definition carrying its own name: keeps it: the name is a durable queue key, so queued and dead-lettered rows survive a renamed export.

Transactional outbox by default

<job>.enqueue inside an action writes the job row in the same transaction as the business write — the handle resolves the ambient jobs facade, the one ctx.jobs names. Commit is the enqueue.

async handle({ input, ctx }) {
  const post = await ctx.posts.publish(input.postId);              // INSERT/UPDATE
  if (input.notify) await notifySubscribers.enqueue({ postId: post.id, orgId: post.orgId });  // same tx
  return post;
}

Rolled back → the job never existed. Committed → durably queued. No window in between, no compensating-action code for an agent to forget.

The relay publishes a claimed batch in two statements — driver.enqueueMany, then store.markPublishedMany, where the row loop spends two per row: 8,564–9,505 rows/s at the default batchSize of 500, against 4,259–5,052 at the 100 it was (10,000 staged rows, three runs each, Postgres, 2026-11-03, a shared box). 500 because a pass is three statements whatever its size: its longest statement was 45 ms, and a dead relay strands 500 rows for its 30 s lease instead of 100. A batch that fails anywhere falls back to the row loop for that batch, which publishes in order up to the failing row and releases the rest: the order guarantee and the poison row named in jobs.outbox.publish-failed are unchanged. JobDriver.enqueueMany is required of every driver — Postgres, Redis and memory each answer a batch in one round trip; OutboxStore.markPublishedMany is optional, and a store without it keeps the row loop. A batch republished after a crash before its mark adds no row — every row is published under the id it was staged with.

External brokers are not exempted: the outbox table stays the transactional record and a relay moves committed rows onto the broker. At-least-once delivery is preserved; the atomicity is not negotiable.

Durable steps

The step is the retry unit, not the job. A failure in nudge re-enters run, replays provision and welcome-email from storage in microseconds, and retries only nudge. That is why an onboarding flow can retry on day 3 without re-provisioning or re-emailing.

Compensation

Charge, reserve, ship: the third fails for good, the first two are undone. try { … } catch { refund } is wrong — it refunds on attempt 1 of a retryable failure, before the retry that ships.

import { job, t } from '@ultimat3/jobs';

type Order = { readonly orgId: string; readonly orderId: string };
declare function charge(order: Order): Promise<{ id: string }>;
declare function refund(chargeId: string): Promise<void>;
declare function reserve(order: Order): Promise<{ id: string }>;
declare function release(reservationId: string): Promise<void>;
declare function ship(order: Order): Promise<void>;

export const checkout = job({
  input: t.object({ orgId: t.uuid, orderId: t.uuid }),
  tenant: ({ orgId }) => orgId,
  idempotencyKey: ({ orderId }) => `checkout:${orderId}`,
  compensates: true,
  retry: { attempts: 3 },
  run: async ({ input, step }) => {
    await step.run('charge', () => charge(input), { compensate: (c) => refund(c.id) });
    await step.run('reserve', () => reserve(input), { compensate: (r) => release(r.id) });
    await step.run('ship', () => ship(input));
  },
});

The deadline cancels

A job's timeout aborts ctx.signal before it fails the attempt. The order is the whole point: failing the attempt re-queues the job, another worker claims it within milliseconds, and a body still running past that moment is a second copy of one job writing into the same run.

run: async ({ input, ctx, step }) => {
  const res = await fetch(url, { signal: ctx.signal });      // stops at the deadline
  throwIfAborted(ctx);                                       // or check it in a long loop
  await step.run('save', (signal) => save(res, { signal })); // the step's own ceiling too
},

ctx.signal is the same seam an action reads, composed with the caller's own — there is nothing jobs-specific to learn, and a job whose caller went away is cancelled for that reason too.

A cancelled body writes nothing (As of 2026-11-02). Once ctx.signal is aborted with a coded reason (X_ABORTED, X_DRAINING, X_JOB_TIMEOUT, X_JOB_SLOT_LOST, X_JOB_LEASE_LOST), a transaction, an entity write and an enqueue rethrow that reason. The step record stays fenced as before and a drained step still lands. Wrap work that must land in runToCompletion(fn).

A finished attempt is not a cancelled one (As of 2026-11-07). A body that completed, and whose ack landed, leaves ctx.signal un-aborted: no other attempt owns the run, so a timer it armed — a log sink's flush, an end() timeout, a deferred enqueue — still writes. A step it never awaited is refused (X_ABORTED, "had already ended"). A failed or suspended attempt, or a completed one whose ack did not land, aborts with X_ABORTED as before. Work meant to outlive the attempt whatever its end has ONE way: runToCompletion(() => work) from @ultimat3/core.

SIGTERM fires it too (As of 2026-09-07). The moment the process starts draining, the worker aborts ctx.signal on every job it holds with X_DRAINING — naming the worker and the signal — and then gives the body the drain's budget to unwind in. A body that stops is interrupted, not failed: the job goes straight back to the ready bucket with the attempt uncounted, so attempts: 1 survives a rollout and the worker replacing this one claims it at once. The verdict is read off the signal, not off whatever the body threw, so an app's own error for a child the shutdown killed is an interruption too. A body that ignores the signal is waited on to the deadline and abandoned there, as before; a manual worker.stop() aborts nothing and waits for its work — until a SIGTERM lands on it, at which point the teardown already waiting adopts the shutdown's deadline and every held job hears the abort.

Idempotency is in the type signature

idempotencyKey: ({ orgId }) => `onboard:${orgId}`,   // REQUIRED by the type

Omitting it is a compile error, not a lint warning. At-least-once is the only honest guarantee any queue provides, so every handler must be replay-safe — and "remember to add a key" is exactly the instruction an agent drops under pressure. A required field converts a runtime duplicate-charge incident into a red squiggle.

Durable business state lives in your tables, never only in the queue payload. A payload is a pointer, not a record. If the queue is drained, replaced, or migrated to another driver, the business must still be reconstructible from Postgres alone. So { orgId }, not { org: {...30 fields} }.

Concurrency, rate limits, queues, retry

Declared per job, held across the fleet.

export const syncCrm = job({
  input: t.object({ orgId: t.uuid, accountId: t.uuid }),
  tenant: ({ orgId }) => orgId,
  idempotencyKey: ({ accountId }) => `crm-sync:${accountId}`,
  concurrency: { key: ({ accountId }) => accountId, limit: 1, whenBusy: 'fail' },
  retry: { attempts: 3 },
  queue: 'integrations',
  async run({ input, step, ctx, finalAttempt }) { /* ... */ },
});

A job over a tenant, queue or global cap, or over a 'wait' concurrency cap, is deferred, never dropped — it stays queued with a later runAt. whenBusy: 'fail' is the one declared exception. 'wait' holds the cap and states no order: a later enqueue can start first.

A lock — withLock()

withLock(key, fn, { ttl, wait?, poll?, store?, signal? }): one holder fleet-wide, held as one slot lock:<key> in the job driver's LeaseStore (x_job_leases on Postgres) — no table of its own, no Redis. fn receives { key, token, signal }. As of 2026-11-09.

import { withLock } from '@ultimat3/jobs';

declare function sendDigest(orgId: string, options: { signal: AbortSignal }): Promise<void>;
declare const orgId: string;

await withLock(`digest:${orgId}`, ({ signal }) => sendDigest(orgId, { signal }), {
  ttl: '2m',
  wait: 0,
});

One run per key

Refused where it is written:

x jobs show <id> --json reports the key a run counts under as concurrencyKey; the manifest's job row carries concurrency: { limit, keyed, whenBusy }.

Consumers — consumer()

Reading a queue, a change stream or a vendor's cursor. A factory over job() plus a supervising task(), never a ninth primitive. It assembles what a hand-rolled consumer has to remember, and none of it can be left out:

import { consumer } from '@ultimat3/jobs';

interface VendorOrder { readonly id: string; readonly cursor: string }
declare const vendor: {
  stream(account: string, cursor: string | null, signal: AbortSignal): AsyncIterable<VendorOrder>;
  behind(account: string, cursor: string | null): Promise<number>;
};
declare const accounts: { connected(): Promise<readonly string[]> };
declare const orders: { upsertAll(rows: readonly VendorOrder[]): Promise<void> };
declare const cursors: {
  of(key: string): Promise<string | null>;
  save(key: string, next: string): Promise<void>;
};

export const syncOrders = consumer<VendorOrder, string>({
  name: 'sync-orders',                                   // REQUIRED: a durable queue key
  tenant: ({ key }) => key,                              // or 'none'
  keys: () => accounts.connected(),                      // what should be running: one, or one per account
  open: ({ key, cursor, signal }) => vendor.stream(key, cursor, signal), // the app's AsyncIterable
  handle: async ({ items }) => {
    await orders.upsertAll(items);
    return items[items.length - 1]?.cursor ?? '';       // the cursor past this batch
  },
  cursor: { load: (key) => cursors.of(key), save: (key, next) => cursors.save(key, next) },
  batch: 100,
  rearm: '1m',
  lag: ({ key, cursor }) => vendor.behind(key, cursor),
});

Delivery is at least once: a drained, failed or crashed batch is delivered again, so handle must be idempotent per item — item ids are the app's. keys runs under tenant: 'none': a tenant-scoped read there needs the app's own crossTenant(reason, fn). Transports ship none: any broker is the app's dependency behind open.

The final attempt

run({ finalAttempt }) is true when a failure of this attempt will not be retried for want of attempts — the same comparison the runner dead-letters on, so a body releases what a retry would have reused without re-deriving it from retry.attempts.

Operating a queue

Every operator capability is a member of JobIntrospection (driver.introspect), implemented by the Postgres and the memory driver alike — x jobs, /_x and the jobs dashboard every defineAdmin() serves at /admin/jobs (Admin dashboard → The jobs dashboard) read nothing else. A hand-written driver implements every member; TS2741 names the one it lacks.

export const syncAccount = job({
  input: t.object({ accountId: t.uuid, orgId: t.uuid }),
  tenant: ({ orgId }) => orgId,
  idempotencyKey: ({ accountId }) => `sync:${accountId}`,
  retry: { attempts: 3 },
  async run({ input, step, progress, finalAttempt }) {
    const rows = await step.run('load', () => loadRows(input.accountId));
    for (const [index, row] of rows.entries()) {
      await step.run(`sync:${row.id}`, () => syncRow(row));
      progress(index + 1, rows.length);
    }
    return { rows: rows.length };
  },
  // ONE hook for every ending. `settled.result` is what `run` returned, typed from it.
  async onSettled(settled) {
    if (settled.outcome === 'completed') {
      await markAccountSynced(settled.input.accountId, settled.result.rows);
    } else if (settled.input !== undefined) {
      await markAccountBroken(settled.input.accountId, settled.code);
    }
  },
});

Idle cost and pickup latency

idlePollMaxMs defaults to 2 s, and to 5 s while the wake is proven.

The wake. A worker pod holds one LISTEN session (startQueueWake, started by the boot). The enqueue statement and the outbox stage carry a pg_notify — transactional, so it is delivered when the row commits and never when it rolls back — and an idle worker or relay passes at once.

Exports and imports — exportRows() and importRows()

A bulk file in either direction is a job. exportRows() streams a paged read to object storage, one object per page; importRows() reads an object back a page at a time into the app's tables. Both are factories over job(), so both get .enqueue(), retries, the dead-letter path, x jobs show and a manifest row. Long form, with the export's rules: the @ultimat3/jobs README.

import { importRows, t } from '@ultimat3/jobs';
import { disk } from '@ultimat3/storage';

const Order = t.object({ id: t.uuid, total: t.number.int() });
declare const db: {
  readonly orders: {
    upsertAll(rows: readonly object[], args: { onConflict: readonly string[] }): Promise<unknown>;
  };
};

export const importOrders = importRows({
  name: 'orders.import',
  input: t.object({ orgId: t.uuid, key: t.string }),
  tenant: ({ orgId }) => orgId,
  source: () => disk('imports'), // ImportSource: stat + get(key, { range })
  key: ({ key }) => key,
  format: 'csv',
  columns: ['id', 'total'],
  schema: Order, // any Standard Schema; csv text is coerced as an HTTP form is
  handle: ({ rows, input }) =>
    db.orders.upsertAll(rows.map((r) => ({ ...r, orgId: input.orgId })), {
      onConflict: ['orgId', 'id'],
    }),
  errors: { sink: () => disk('imports'), prefix: ({ orgId, key }) => `${orgId}/errors/${key}` },
});

One person's data — subjectExport() and subjectErase()

Declare the person once; the framework finds every table holding something of theirs. defineSubject(entity, { driver, key, tables, targets?, columns? }) is a plain value both job factories take. The reach is walked from the foreign keys already declared, through every table the erase deletes, so a table added next month is X_SUBJECT_TABLE_UNDECLARED on x verify until it has a treatment. Opt in: no defineSubject, no check.

import type { Driver, EntityCore } from '@ultimat3/entity';
import { defineSubject, subjectErase, subjectExport, t } from '@ultimat3/jobs';
import { disk } from '@ultimat3/storage';

declare const members: EntityCore;
declare const driver: Driver;

export const memberSubject = defineSubject<{ orgId: string; memberId: string }>(members, {
  driver, // the app's own: tenancy, row observers and cache tags hold
  key: (input) => input.memberId,
  tables: {
    members: { anonymize: { email: 'erased@example.invalid', name: '' } },
    posts: { retain: 'published work belongs to the org' },
    comments: 'delete',
    likes: 'delete',
  },
});

export const eraseMember = subjectErase({
  name: 'members.erase',
  input: t.object({ orgId: t.uuid, memberId: t.uuid }),
  tenant: ({ orgId }) => orgId,
  subject: memberSubject,
});

export const exportMember = subjectExport({
  name: 'members.export',
  input: t.object({ orgId: t.uuid, memberId: t.uuid }),
  tenant: ({ orgId }) => orgId,
  subject: memberSubject,
  prefix: ({ orgId, memberId }) => `org/${orgId}/subjects/${memberId}`,
  format: 'ndjson',
  sink: () => disk(),
});

x exercise proves the declaration: subjectProof seeds every table with canaries, erases one generated person, and searches every stored row. Their value outside a retain is X_SUBJECT_ERASE_INCOMPLETE, naming the table and column.

personalData() coverage

Rule: .personalData(kind) on a column (@ultimat3/entity) adds NO second export or erase path — subjectExport/subjectErase stay the one. It is the proof that the declared subjects cover every marked column. As of 2026-11-09.

Enforced twice: a subject run refuses X_PERSONAL_DATA_UNERASED before it touches a row, and x exercise lists X_PERSONAL_DATA_UNERASED / X_PERSONAL_DATA_UNREACHED as findings. With an auditSink installed, subjectErase writes one allowed record per completed erase (primitive: 'action', surface: 'job', input: { subject, run } — the subject's entity and the run id, never the key, which an app may make an email) and one failed record per attempt that threw; a sink that refuses fails the run. At least once: a sink dedupes on the record's idempotencyKey, <job>:<runId>.

Rollups and materialized views

Two factories over job(), two jobs, As of 2026-11. A time window over append-mostly rows is rollup(), recomputing only recent buckets into the app's own table. An arbitrary join is a materialized view, refreshed whole by refreshView(). Neither is scheduled: the app's task enqueues it.

import { bigint, database, entity, integer, newId, timestamp, uuid } from '@ultimat3/entity';
import { refreshView, rollup } from '@ultimat3/jobs';

const likes = entity('likes', {
  columns: { id: uuid().primaryKey(), orgId: uuid().tenant(), createdAt: timestamp() },
});
const likesPerDay = entity('likes_per_day', {
  columns: { id: uuid().primaryKey(), orgId: uuid().tenant(), day: timestamp(), likes: integer() },
  indexes: [{ on: ['orgId', 'day'], unique: true }],
});
const postTotals = entity('post_totals', {
  view: { sql: 'select org_id, post_id, count(*) as total from likes group by 1, 2', materialized: true },
  columns: { orgId: uuid().tenant(), postId: uuid().unique(), total: bigint() },
  primaryKey: ['postId'],
});
const db = database({ likes, likesPerDay });
const ORG = '00000000-0000-7000-8000-000000000001';

export const dailyLikes = rollup({
  name: 'daily-likes',
  into: likesPerDay,                 // the app's entity, unique index over `key`
  table: () => db.likesPerDay,
  key: ['orgId', 'day'],
  lookback: '2d',                    // how late a row may arrive and still count
  bucket: { unit: 'day', timeZone: 'UTC' }, // one row of `into`: a series unit, an IANA zone
  tenant: (input) => input.orgId ?? ORG,
  source: ({ since, bucket, input }) =>
    db.likes
      .where({ orgId: input.orgId ?? ORG })
      .andWhere('createdAt', 'gte', since) // a bucket's first instant, never mid-bucket
      .series({ on: 'createdAt', ...bucket, value: 'count' }),
  map: (point, { input }) => ({
    id: newId(),
    orgId: input.orgId ?? ORG,
    day: point.bucket,
    likes: point.value,
  }),
});

export const refreshPostTotals = refreshView({
  name: 'refresh-post-totals',
  view: postTotals,
  concurrently: true,
});

Rebuilding a search index — reindex()

reindex({ name, entity, source, tenant, index? }) is a backfill() in two phases, one durable step per page. Rows: each row's searchDocumentOf(entity, row) goes to the searchIndex port (default: the installed one). Sweep: every document the index lists whose row source no longer reads — deleted while the index was down, moved to another tenant — is removed, tombstoned one millisecond above the listed version, so an id written again meanwhile keeps its newer document. source is the WHOLE table: a narrowed source empties the index of the rest. tenant: 'none' sweeps every tenant; a declared tenant, its own. Run it after swapping Postgres' built-in index for Meilisearch or Typesense, or after an index lost data. An entity with no .searchable() column is refused at declaration (X_INVARIANT). As of 2026-11-10.

Partition maintenance — partitions()

partitions({ name?, cron?, tz?, clock? }) keeps every entity({ partitionBy }) table ready: one durable step per partitioned entity runs @ultimat3/db's maintainPartitions — creation under a 5 s lock_timeout (PARTITION_LOCK_TIMEOUT_MS; a 55P03 is X_DB_LOCK_TIMEOUT, retryable, so a long report query delays the run rather than queueing every write behind ACCESS EXCLUSIVE on the parent), then each expired partition retired by detach … concurrently (outside any transaction, as Postgres requires) and dropped. It creates ahead partitions past the current one and retires those retain expired. A job plus a .schedule task() (<name>.schedule, default '17 3 * * *' UTC, daily), as consumer() carries .supervisor. now is the job's clock, so a test moves it; a second run at one instant changes nothing. Report: { tables: [{ table, created, dropped }] }. Moving older rows in: enqueue it once with { from: '<the oldest row’s ISO instant>' } — that run creates every partition back to it and retires nothing, and its idempotency key is its own, so it never folds into the schedule's — then run the backfill(). As of 2026-11-09.

import { partitions } from '@ultimat3/jobs';

export const maintainPartitions = partitions(); // list it in defineApi({ jobs })

Outbound webhooks — webhook()

An outbound webhook is a job: webhook() is a factory over job() that delivers one event to one endpoint. It brings .enqueue(), retry with core's backoff, the dead-letter path, x jobs show and a manifest row; it adds signing, a timestamped signature, disable-after-N consecutive failures and a ledger row per attempt (packages/jobs/src/webhook.ts). It never owns the event taxonomy, which endpoints exist, or what a payload means — the fan-out is the app's own loop over its own subscription table. Do not hand-write a delivery job.

import { memoryWebhookLedger, webhook } from '@ultimat3/jobs';

export const deliver = webhook({
  name: 'partner.webhooks',          // required: a durable queue key
  tenant: ({ orgId }) => orgId,      // the org that owns the endpoint — or 'none'
  endpoint: ({ endpointId }) => db.endpoints.byId(endpointId), // read per attempt, never checkpointed
  event: ({ eventId }) => db.events.byId(eventId),             // { topic, body }
  ledger: memoryWebhookLedger(),     // dev only — a WebhookLedger over your own table in production
  disableAfter: 10,                  // DEFAULT_WEBHOOK_DISABLE_AFTER
});

for (const endpoint of await subscribersOf('orders.paid')) {
  await deliver.enqueue({ endpointId: endpoint.id, eventId: event.id, orgId: endpoint.orgId });
}

Full contract, every rule with its reason: packages/jobs/README.md § Outbound webhooks.

Inbound webhooks

An inbound webhook is a job too: webhookReceiver() is a factory over job(), the twin of webhook(), plus the plain route that feeds it. The route reads the exact bytes, asks your verify, enqueues the payload verify read out of them, and only then answers 200 { ok: true, eventId, deduped }. handle runs as the job's body and receives that payload: it never re-verifies, so a retry past a vendor's 5-minute window is not refused. No vendor verifier ships: verify is your function, and verifyBodyHmac from @ultimat3/http is the one helper (packages/jobs/src/webhook-receiver.ts). It is the one door for a vendor's webhook. A pinned action (http.path) answers 400/422 from its schema before the signature is checked, and refuses XML, octet-stream and no content type.

// apps/web/app/billing/jobs/billing-events.ts — declaring it is its registration
import { verifyBodyHmac } from '@ultimat3/http';
import { t, webhookReceiver } from '@ultimat3/jobs';

const Event = t.object({ id: t.string, type: t.string });

export const billingEvents = webhookReceiver({
  name: 'billing.events',            // required: a durable queue key, and the route's webhook.<name>
  path: '/hooks/billing/:account',   // absolute, may hold :param; never under /api or /_x
  tenant: 'none',                    // or ({ orgId }) => orgId
  verify: (inbound) => {             // { bytes, headers, params, url, request() } — nothing parsed
    const { text } = verifyBodyHmac({
      bytes: inbound.bytes,
      secret: [Bun.env['BILLING_SECRET'] ?? '', Bun.env['BILLING_SECRET_OLD'] ?? ''],
      given: inbound.headers.get('x-billing-signature'),
      encoding: 'hex',               // or 'base64'
      prefix: 'sha256=',
    });
    const event = Event.parse(JSON.parse(text ?? ''), 'billing event'); // catch both: see below
    return { eventId: event.id, payload: event };   // the id from the SIGNED bytes — or { answer }
  },
  handle: async ({ eventId, payload }) => { /* the work; at least once per eventId */ },
});

List it in defineApi({ jobs }) like any job. Both boots mount every declared receiver's routes beside runtime.routes — no second list, and one also named in runtime.ts is mounted once.

Worked example: the reference app's apps/web/app/webhooks/jobs/billing-events-receiver.ts. Codes: X_WEBHOOK_ENDPOINT_UNKNOWN, X_WEBHOOK_ENDPOINT_INVALID, X_WEBHOOK_ENDPOINT_DISABLED, X_WEBHOOK_EVENT_UNKNOWN, X_WEBHOOK_EVENT_INVALID, X_WEBHOOK_DELIVERY_FAILED, X_WEBHOOK_DELIVERY_THROTTLED, X_WEBHOOK_DELIVERY_REJECTED (Error codes).

Double-entry postings — ledger()

ledger({ name, tenant, book }) is a factory over job(). Input: { key, entries: [{ account, amount: Money }], orgId? }. The run refuses anything that is not balanced Money before a line is written, then writes every line in ONE durable step through the app's LedgerBook. No provider, account plan or currency policy ships (axiom 8). As of 2026-11-09.

A provider's payment webhook is a webhookReceiver() whose handle enqueues the posting keyed by the provider's id: the receiver dedupes the delivery, the ledger dedupes the posting.

import type { LedgerBook } from '@ultimat3/jobs';
import { ledger, webhookReceiver } from '@ultimat3/jobs';

declare const book: LedgerBook; // entityLedgerBook({ driver, entity: ledgerLines }) in an app
declare function verifyPayment(bytes: Uint8Array, signature: string | null): {
  id: string;
  amountMinor: number;
  currency: string;
};

export const postPayment = ledger({ name: 'payments.post', tenant: 'none', book: () => book });

export const paymentEvents = webhookReceiver({
  name: 'payments.events',
  path: '/hooks/payments',
  tenant: 'none',
  verify: (inbound) => {
    const event = verifyPayment(inbound.bytes, inbound.headers.get('x-signature'));
    return { eventId: event.id, payload: event };
  },
  handle: async ({ eventId, payload, step }) => {
    const { amountMinor, currency } = payload as { amountMinor: number; currency: string };
    await step.run('post', () =>
      postPayment.enqueue({
        key: eventId,
        entries: [
          { account: 'cash', amount: { minor: amountMinor, currency } },
          { account: 'revenue', amount: { minor: -amountMinor, currency } },
        ],
      }),
    );
  },
});

Driver interface

One interface. Job code never changes. Step persistence hangs off the same object (steps), so it is identical on every implementation. It is the jobs adapter port: the interface is on @ultimat3/jobs/adapter, memoryJobDriver is the driver to copy, behavesLike(jobDriverConformance, () => yourDriver()) proves one, and jobs: in apps/<app>/runtime.ts wires it — Adapters. The bus step.waitForEvent resumes from (eventBus) and the store behind the outbox (outboxStore) are two more ports, swapped the same way and no other: Adapters, Adapters.

export interface JobDriver {
  readonly name: string;
  /** Step persistence lives with the queue: one store, one transaction boundary. */
  readonly steps: StepStore;
  enqueue(request: EnqueueRequest): Promise<EnqueueResult>;
  claim(options: ClaimOptions): Promise<readonly ClaimedJob[]>;
  /** Both settles are fenced on the CLAIMER and answer whether they landed. */
  ack(jobId: string, by: { workerId: string; durationMs?: number }): Promise<boolean>;
  nack(jobId: string, options: NackOptions): Promise<boolean>;
  heartbeat(jobId: string, options: HeartbeatOptions): Promise<boolean>;
  stats(): Promise<readonly QueueStats[]>;
  /** The `x_backfills` ledger, when the driver ships one. `postgres` and `memory` do. */
  readonly backfills?: BackfillLedger;
  /** Fleet-wide slot counting — the only thing that can enforce `concurrency` across replicas. */
  readonly leases?: LeaseStore;
  readonly introspect?: JobIntrospection;
  close?(): Promise<void>;
}

Three of the four optional members degrade rather than refuse: no introspect is x jobs list with nothing to list, no backfills is a backfill() pass that runs with no bookkeeping, and no close is a driver holding nothing to hand back.

leases is the one that refuses. The in-process limiter is a fast path over one heap and is multiplied by the replica count, so a driver with no LeaseStore can only hold concurrency.limit per process. jobWorker().start() therefore throws X_JOB_CONCURRENCY_UNENFORCEABLE, naming every registered job that declared concurrency, rather than logging a cap it cannot keep. postgres ships one; a driver you write yourself needs one before any job in the tree may declare concurrency.

Two implementations ship, As of 25.0.0: postgresJobDriver() and memoryJobDriver(). Redis is not a jobs driver, and neither is NATS: 25.0.0 deleted both all-throw stubs (createNatsDriver and the Redis one), whose every method raised X_NOT_IMPLEMENTED — an app could typecheck against a driver that could not enqueue a job. NATS stays the realtime fanout transport.

There is no jobs.driver, and 5.0.0 is where it went. It accepted 'postgres' | 'redis' | 'nats' and had no reader anywhere — boot always built postgresJobDriver, stated in packages/jobs/src/driver.ts's own header. So setting it to redis changed nothing at all and you silently got Postgres, which is the more dangerous behaviour because nothing reports it. From 5.0.0 to 24.x a stale key still rode through untyped; 25.0.0 refuses it at defineConfig, naming the driver seam. Deleting it from app.config.ts is the whole of the upgrade.

The seam that does work is jobs: in export const runtime = { … } (apps/<app>/runtime.ts; 31.0.0 deleted setJobDriver) — swap the driver, zero job-code change, which is what the interface buys; a test swaps one for itself with await swapAdapter('jobs', driver) from @ultimat3/testing. Every swappable backend works this way: Adapters. redisJobDriver() is the second durable driver (As of 2026-10, #710): jobs: redisJobDriver() on core's redisClient() — while Redis is down an enqueue is X_DRIVER_UNAVAILABLE at once, never a wait — imported from @ultimat3/jobs/redis, held to the same conformance suite as Postgres, without the operator surface (x jobs show, retry, cancel), backfill bookkeeping or fleet-wide concurrency — packages/jobs/README.md has the table. There is no NATS jobs driver.

x jobs drain is planned, As of 25.0.0. It exits X_NOT_IMPLEMENTED before the queue boots, pointing at x jobs list --json: with Postgres the one durable driver, there is nowhere to drain to. Its two former --to values, redis and nats, were all-throw stubs — until 2026-10 a drain leased the whole pending batch for five minutes, failed every enqueue and nacked it back, nothing moved, and no worker could claim those jobs meanwhile. 25.0.0 deleted the stubs and the unreachable drain body behind the planned answer. A drain that returns with a real second driver is new code, and owes the lesson of --to memory, a target until 2026-09: memoryJobDriver() is a Map inside the command's own process, so the drain enqueued each job into it, acked the durable row off the source, printed ok: true, and lost every copy when the command exited.

So there is no cross-driver migration procedure — and none is needed while postgres is the only driver that runs.

Dead letter

Draining a worker mid-job does not finish the current step (As of 2026-09-07). On SIGTERM the worker aborts each held job's ctx.signal with X_DRAINING; a body that reads it — outbound() from @ultimat3/core/outbound (it reads ctx.signal itself), fetch(url, { signal: ctx.signal }), throwIfAborted(ctx) between steps — is interrupted: the attempt is uncounted, the lease is released, and the job goes straight back to the ready bucket for the worker replacing this one. That worker resumes from the last step that completed — persisted steps replay from storage, the step cut short runs again from its start, and past the abort step.run refuses to write (X_ABORTED), so a half-finished step is never recorded as done. A body that ignores the signal runs until the drain deadline and is abandoned there: its lease lapses, the queue redelivers it, and the attempt is counted. Only a manual worker.stop() waits for the work it holds. Plan a deploy for the interruption, not for the step: keep steps short and idempotent, and pass ctx.signal to every outbound call.

Observability

Every command supports --json. See CLI reference.

A lease that lapses is reported, never swallowed. A worker renews the visibility window every heartbeatIntervalMs (default a third of the window) for as long as a job runs. One failed renewal is jobs.heartbeat.failed at warn — the window still has room for the next. A whole window with none landing is jobs.lease.lost at error plus one point on job_leases_lost_total{queue}: the queue is now free to hand that job to another worker while this one is still running it, which is at-least-once becoming exactly-twice. Alert on any non-zero rate. The window is measured from the last renewal that landed, on the worker's own clock, so a heartbeat that hangs is caught the same as one that rejects.

Errors

Full index: Error codes. Verbatim error shapes live in each package's src/errors.ts.

Testing

x test job — cloned DB + frozen clock.

// job test — the step guarantee, not the happy path
test('onboardOrg retries only the failed step', async ({ seed, clock, mail }) => {
  const { org } = await seed('fresh-org');
  mail.failOnce(nudgeEmail);
  await runJobs(onboardOrg, { orgId: org.id });
  clock.advance('3d');
  const trace = await runJobs.drain();
  expect(trace.steps.provision.executions).toBe(1);       // replayed from storage
  expect(trace.steps['nudge'].executions).toBe(2);        // only this one retried
});

Asserted by the runner: step replay, idempotency-key dedupe, retry/backoff, concurrency and rate limits, outbox atomicity on rollback. clock.advance drives step.sleep — never assert on wall-clock time. See Testing.

What a crash does to a job is asked by crashProof(job, input) — killed at every step boundary, a double effect is X_JOB_CRASH_DIVERGED — and what a race does by simulateJobs({ jobs, workers, invariant }), seeded. See Testing.

Rules

• Never assume a job runs once. Assume at-least-once. A backfill()'s handle is the same rule one level down: it runs before its checkpoint lands, so an attempt cancelled between the two hands that page to the next one — write through upsertAll, updateWhere or a statement whose second run changes nothing.
• At-least-once has two JS-shaped edges. A body that ignores ctx.signal keeps running past a lost fleet slot, so concurrency is exceeded until it returns. A synchronous step that blocks the event loop longer than the 30 s visibility timeout starves the heartbeat: the lease lapses and another worker runs the job again. Await I/O, honour signal, and split CPU work into steps.
• Never put durable business state only in the payload.
• Never do slow work inline in an action — enqueue a job.
• A job never renders, redirects, or reads headers. Actor and tenant come from ctx.
• One step.run per externally-visible side effect. A step that does two things cannot be retried.
• Cron never contains a handler body — that is a scheduled task enqueuing a job.