Background jobs (queue-agnostic)

DaloyJS ships a background-job interface. Work that must outlive the HTTP request (and maybe the process) becomes a named handler plus a JSON payload, persisted behind the JobStore SPI and run by a leased worker with bounded retries. It is the durable counterpart to in-process cron, with zero runtime dependencies.

SPI is short for service provider interface, the mirror image of an API. An API is the surface you call; an SPI is the surface you implement so the framework calls you. DaloyJS defines JobStore and ships one implementation (MemoryJobStore); your Redis, Postgres, or SQS adapter is the other. Every other term here that reads like queue jargon (lease, fencing, full jitter, dead letter, redrive, outbox) is defined in the glossary at the end of the page.

• Idempotent enqueue. A (queue, idempotencyKey) pair is unique. Retried producers (a client that re-POSTs, a cron tick firing on 8 replicas) collapse into one job instead of eight side effects.
• At-least-once worker. Atomic claims with leases and heartbeats, retries with full-jitter backoff, per-attempt timeouts, and a dead-letter state for poison jobs. Graceful shutdown drains in-flight work.
• Bring your own durability. MemoryJobStore covers tests and single-process dev. Production plugs Redis, Postgres, or a cloud queue in through the same SPI, as application code. DaloyJS never takes a storage dependency.

The companion post Background Jobs After the HTTP Response covers the design argument. This page is the reference.

Quick start

Mount jobs with app.useJobs(). That creates the queue, optionally starts an in-process worker, and registers the graceful-shutdown drain. In-flight jobs get the grace period, then their AbortSignal fires and they fail back to the queue for another worker to claim.

import { createApp, MemoryJobStore, jobIdempotencyKey } from "@daloyjs/core";

const app = createApp();

app.useJobs({
  // Tests / single-process dev. Production: your Redis/Postgres JobStore adapter.
  store: new MemoryJobStore(),
  handlers: {
    "email.welcome": async ({ job, signal }) => {
      const { to, locale } = job.payload as { to: string; locale: string };
      await sendWelcomeEmail(to, locale, { signal }); // your mailer
    },
  },
  startWorker: true, // this process also claims and runs jobs
});

app.post("/users", contract, async (ctx) => {
  const user = await db.insertUser(ctx.body); // commit FIRST, then enqueue
  await app.jobs!.enqueue({
    name: "email.welcome",
    payload: { userId: user.id, to: user.email, locale: user.locale },
    tenant: ctx.state.tenant,
    idempotencyKey: jobIdempotencyKey({
      tenant: ctx.state.tenant,
      name: "email.welcome",
      key: user.id, // a duplicate of this request never double-sends
    }),
  });
  return { status: 201 as const, body: user };
});

Without an App (scripts, dedicated worker binaries, tests), drive the primitives directly. worker.runOnce() claims and settles one job, so tests stay deterministic without timers or Redis:

import {
  createJobQueue,
  createJobWorker,
  MemoryJobStore,
} from "@daloyjs/core/jobs";

const queue = createJobQueue({ store: new MemoryJobStore() });
const worker = createJobWorker({
  queue,
  handlers: {
    "email.welcome": async ({ job }) => {
      const { to } = job.payload as { to: string };
      await sendWelcomeEmail(to, "en");
    },
  },
});

const { job, duplicate } = await queue.enqueue({
  name: "email.welcome",
  payload: { to: "ada@example.com" },
});

await worker.runOnce(); // true: claimed, ran, completed
(await queue.get(job.id))?.status; // "completed"

A runnable end-to-end version of this flow (dedupe, a transient failure that succeeds on retry, graceful stop) ships in the repo as examples/jobs-basic.ts.

When to use

Jobs are for work that must survive the HTTP request, or the process. If the work fits in the request, do not enqueue.

Does the client need the result to build the HTTP response?
  YES -> do it in the handler (maybe resilientFetch). Not a job.
  NO  -> would losing this work on process crash / deploy be unacceptable?
          NO  -> in-process is fine (handler fire-and-forget or app.cron).
          YES -> can it be one named function + JSON payload, retried as a whole?
                  NO  -> workflow engine (Temporal / Inngest / Eve). Daloy stays the API.
                  YES -> JOBS (this feature).

• Now, in this request. Handler.
• Now, in this process, on a clock. app.cron()
• Eventually, even if this process dies. jobs.enqueue
• Same function must pause for hours and resume. Temporal, Inngest, or Eve. Daloy stays the HTTP API.

A job is worth it when most of these hold:

1. The HTTP response can succeed without the side effect having finished (email, webhook fan-out, thumbnail, search index, PDF, analytics).
2. The side effect may fail transiently (SMTP 421, Stripe 429, a 503 from Azure OpenAI) and should retry with backoff.
3. A deploy, OOM, or rolling update must not drop the work.
4. The unit of work is one handler, or you are willing to enqueue the next job at the end of this one (a job chain).
5. Duplicate runs are safe because you pass an idempotency key through to the downstream API (delivery is at-least-once).
6. The runtime is mixed: an API on Lambda/Workers cannot finish a two-minute PDF, but a Node worker pool can.

When not to use

Where it runs

The queue is producer-only and needs no timers, so any runtime can enqueue. The worker is a poll loop, so it belongs where a long-lived process exists.

A. Local dev / CI (always Memory)

One Node process: MemoryJobStore + createJobWorker + runOnce() in tests, or startWorker: true while developing. Use this for unit and integration tests. The store dies with the process, so it is not production-durable.

B. Single VPS / one replica (small prod)

One long-lived Node process serves HTTP and runs the worker (useJobs({ startWorker: true })). That is fine when side effects are light and brief deploy downtime is acceptable, if the store is Redis or Postgres so jobs survive the restart. Do not run Memory in production. useJobs logs a warning. strictProduction: true refuses to boot. Prefer cronEnqueue over cron for side effects, so a second process cannot double-send.

C. Kubernetes (AKS)

Deployment api: N pods, enqueue only (startWorker: false). Deployment worker: M pods, claim and run (startWorker: true, no public ingress). The store is Azure Cache for Redis or Azure Database for PostgreSQL through a JobStore adapter in your repo. Core does not ship that adapter. Ingress exposes only api. Workers need egress to the store plus SMTP, Stripe, or OpenAI. Set terminationGracePeriodSeconds above the worker's stop(graceMs) so SIGTERM drains instead of killing in-flight jobs.

D. Serverless API + always-on worker

Vercel, Lambda, or Cloudflare handlers enqueue to a remote store, then the isolate dies. A Node container (Container Apps, a VM, Fly.io) runs the worker. Do not set startWorker: true on Lambda or Workers in v1. Isolates are the wrong place for a poll loop.

E. The queue is SQS / Service Bus already

A JobStore adapter maps put → send message, claim → receive + visibility timeout, complete → delete, fail → the broker's native retry and dead-letter queue (DLQ). If Azure Functions or another consumer already drains the queue, DaloyJS can be producer-only. The worker is optional.

F. Multi-tenant SaaS

Every enqueue sets tenant and builds its key with jobIdempotencyKey({ tenant, name, key }), so two tenants cannot collide on the same natural key. Share one queue: "mail" with the tenant field on the record, or partition at the broker with a per-tenant queue name when isolation demands it. Jobs are not HTTP, so pass ctx.state.tenant explicitly. Nothing reads it for you.

Deployment/api     replicas: N   startWorker: false   PORT 3000 (public ingress)
Deployment/worker  replicas: M   startWorker: true    no public Service
Secret             DATABASE_URL / REDIS_URL (Key Vault -> env, both deployments)
Clock              ONE K8s CronJob -> authenticated POST that enqueues,
                   or cronEnqueue on the api pods (duplicate ticks collapse
                   into one job via the per-slot idempotency key)

Delivery is at-least-once

A job is delivered at least once. The worker claims a record with an atomic lease (lockedBy + leaseUntil), runs your handler, then marks it complete. If the process dies between “handler succeeded” and “complete persisted”, the lease expires and another worker runs the handler again. Two enqueues with the same idempotency key and a deep-equal payload return one job (duplicate: true). The same key with a different payload throws JobIdempotencyConflictError.

Handlers must be idempotent. For side effects that move money or send messages, pass a key through to the downstream API so a duplicate run is a no-op there. With Stripe, that is the Idempotency-Key header. The Daloy job key and the Stripe key should be the same natural value.

import { JobFatalError } from "@daloyjs/core/jobs";

const worker = createJobWorker({
  queue,
  handlers: {
    // You already returned 201 for the order; capture happens here.
    "payments.capture": async ({ job, signal }) => {
      const { orderId } = job.payload as { orderId: string };
      const order = await db.getOrder(orderId);
      if (!order) throw new JobFatalError("order vanished: " + orderId); // no retry
      await stripe.paymentIntents.create(
        { amount: order.totalCents, currency: order.currency, confirm: true },
        { idempotencyKey: orderId }, // Stripe dedupes retries of THIS job
      );
      // Forward ctx.signal to any fetch-based I/O so timeouts unwind promptly.
    },
  },
});

Long-running handlers get a heartbeat: the worker extends the lease every leaseMs / 3 automatically, and ctx.heartbeat() extends it manually. If the lease is lost (it expired, another worker claimed the job, and this worker's writes are now fenced off), the handler's signal aborts so it stops touching the job. Keep leaseMs above your slowest expected attempt so the automatic heartbeat can keep the lease.

Job status machine

delayed  --(runAt <= now)----------> queued
queued   --claim--------------------> running
running  --complete-----------------> completed  (terminal)
running  --fail, attempts < max-----> delayed (runAt = now + backoff) or queued
running  --fail, attempts >= max----> dead       (terminal)
running  --JobFatalError------------> dead       (terminal)
running  --lease expired------------> queued     (attempts +1: poison handlers
                                                  cannot loop forever uncounted)
queued / delayed / running --cancel-> cancelled  (terminal)
completed / dead / cancelled: no transitions

Terminal records are not deleted on the spot. completed/cancelled are retained 24h and dead 7d (for inspection) by default, swept lazily on mutating operations. A dead job keeps its lastError message. Wire onDead to your alerting.

API reference

Everything lives at @daloyjs/core/jobs and is re-exported from @daloyjs/core.

createJobQueue(options)

queue.enqueue(options)

Also on the queue: queue.get(id) reads one job, queue.cancel(id) cancels a non-terminal job.

createJobWorker(options)

Worker methods: start() begins the poll loop, stop(graceMs?) drains in-flight jobs then aborts stragglers (they fail back to the queue), runOnce() settles one job (tests), getState() reports { running, inFlight, workerId }.

MemoryJobStore

A correct, full implementation of the SPI for tests and single-process apps. It is not durable across processes and is invisible to other replicas. useJobs warns when it sees this store with production config. Payloads are serialized and re-parsed with the prototype-pollution-safe parser on every read, and returned as deep copies, so callers cannot mutate store state. Options: capacity (10,000 jobs, overflow sweeps expired terminal records then throws store_full without evicting queued work), retentionMs (24h for completed/cancelled), deadRetentionMs (7d), and an injectable now. Adds list(filter) and dump() for test inspection.

Errors

cron vs cronEnqueue vs jobs

cronEnqueue registers a scheduler task whose tick enqueues a job instead of running the side effect in-process:

app.cronEnqueue(
  { name: "nightly-reconcile", cron: "0 2 * * *" },
  { name: "ops.reconcile", payload: {} },
);
// Every replica may tick. The derived idempotency key
// "cron:nightly-reconcile:<schedule slot>" is identical on all of them,
// so 8 ticks collapse into 1 job and exactly 1 worker runs it.
// Requires app.useJobs() first; it throws store_required at registration
// otherwise, not silently at the first tick.

Sweeping this isolate's MemoryResponseCacheStore must stay app.cron(), because other replicas have their own memory. A job would run on one random worker and sweep the wrong process's cache.

Production stores and the JobStore SPI

All durability lives behind JobStore. DaloyJS ships the SPI and the Memory implementation. Redis, Postgres, SQS, and similar backends are adapters in your repository, so the core keeps zero runtime dependencies and your queue choice stays yours. Each method owes this atomicity:

A Redis adapter is application code. The sketch below is docs-only (implement JobStore, DaloyJS does not ship Redis):

// src/jobs/redis-store.ts - application code, NOT shipped by DaloyJS.
// Illustrative sketch: bring your own redis client and own the atomics.
import type { EnqueueResult, Job, JobStore } from "@daloyjs/core/jobs";

export class RedisJobStore implements JobStore {
  constructor(private redis: RedisClient) {}

  async put(job: Job, fingerprint: string | null): Promise<EnqueueResult> {
    // Idempotency: SET queue:{q}:idem:{key} -> job.id NX.
    // On a hit, load the winner and compare fingerprints (conflict throws).
    // On a miss, HSET job:{id} and ZADD queue:{q} (score = runAt, priority).
    // One Lua script keeps the pair atomic.
    // ...
  }

  async claim(queue: string, workerId: string, now: number): Promise<Job | null> {
    // Lua: ZRANGEBYSCORE queue:{q} up to now -> HSET job:{id} status=running,
    // lockedBy=workerId, leaseUntil=now+leaseMs, attempts+1 -> ZREM.
    // The SET-lock equivalent: SET job:{id}:lock workerId PX leaseMs NX.
    // Also requeue records whose leaseUntil < now (lease-expired crashes).
    // ...
  }

  // heartbeat: extend PX only when the lock value is still workerId.
  // complete / fail / cancel: compare-lock, update hash, publish nothing.
  // get: HGETALL, then lazily requeue when leaseUntil < now.
}

SQS / Service Bus adapters map even more directly: put is send-message, claim is receive with a visibility timeout (the broker's name for a lease), complete is delete, and fail is the native redrive policy that moves the message to a dead-letter queue.

Recipes

The recipes below cover the cases people hit first. Each names the job, the payload shape, the idempotency key, and the fatal-vs-retry split. The full list of fourteen is at the end of this section.

1. Welcome email after signup

Shown end to end in the quick start. SMTP must not delay the 201. Payload is { userId, to, locale }, never the password. Key per user id. Retry SMTP 4xx and timeouts. Throw JobFatalError on an unknown user or a permanent bounce.

2. Stripe (or any provider) webhook → 202 + job

Providers demand a fast 2xx and will disable endpoints that time out. Verify the signature, enqueue, and return 202. Entitlements and invoice work run in the worker.

app.post("/webhooks/stripe", contract, async (ctx) => {
  const event = verifyStripeSignature(ctx); // your existing check, first
  await app.jobs!.enqueue({
    name: "billing.stripe_event",
    // The id and type, not the megabyte event body: the handler re-fetches
    // or reads your stored copy.
    payload: { providerEventId: event.id, type: event.type },
    idempotencyKey: jobIdempotencyKey({
      name: "billing.stripe_event",
      key: event.id, // Stripe event ids are already unique: provider retries
                     // of the webhook collapse into one job.
    }),
  });
  return { status: 202 as const, body: { received: true } };
});
// Handler: retry downstream 503s; JobFatalError on event types you refuse.

9. LLM / Azure OpenAI batch → 202 + job id

A two-minute summarization cannot hold an API request, and cannot run on a Workers isolate at all. Return 202 with the job id and let the client poll a status route you write. DaloyJS mounts no /jobs HTTP API.

app.post("/tickets/:id/summary", contract, async (ctx) => {
  const { job } = await app.jobs!.enqueue({
    name: "ai.summarize_ticket",
    payload: { ticketId: ctx.params.id }, // handler loads the text itself
    maxAttempts: 3,
    timeoutMs: 120_000,
    idempotencyKey: jobIdempotencyKey({
      name: "ai.summarize_ticket",
      key: ctx.params.id,
    }),
  });
  return { status: 202 as const, body: { jobId: job.id } };
});

// Handler: call Azure OpenAI with your credential + ctx.signal.
// Retry 429/5xx; JobFatalError on a 400 (bad prompt / policy).
// Status route (yours): GET /jobs/:id -> queue.get(id) -> { status, result }.

12. Idempotent payment capture off the request

Shown in delivery semantics. The order already returned 201, capture runs as a job, and the handler sends Stripe Idempotency-Key: orderId so retries of the job are no-ops at Stripe. The Daloy job key and the Stripe key are different systems. Set them to the same natural value so you never confuse the two.

7. Nightly reconciliation, once, cluster-wide

Shown in cron vs cronEnqueue: cronEnqueue turns every replica's 02:00 tick into one idempotent enqueue. The key includes the calendar slot, so tonight's run never dedupes against tomorrow's.

8. Process-local cache sweep

This stays on app.cron(). It sweeps this process's memory, and the other replicas sweep their own. Enqueueing this work would sweep one random worker's cache and leave the API pods dirty.

All fourteen instances

1. Welcome email after POST /users: payload { userId, to, locale }. Key per user. Fatal on permanent bounce.
2. Provider webhook processing: signature at HTTP, 202 immediately. Payload { providerEventId, type }. Key = provider event id.
3. Outbound webhook that must live 24h: webhook.deliver with { url, eventType, bodyId }. The handler calls createWebhookSender() once. JobFatalError when the sender dead-letters so you do not retry forever in two systems.
4. Search reindex: search.index_article with { articleId }. Key articleId + version (or last-write-wins in the handler). delayMs: 500 coalesces bursts.
5. Thumbnail / PDF / image variant: media.derive with { blobUrl, ownerId, variant }, a URL on blob storage, never bytes. Fatal on unsupported MIME.
6. Fan-out notifications: one domain event enqueues N jobs (notify.email, notify.sms) or one notify.fanout that enqueues children. That is a job chain.
7. Nightly tenant reconciliation: cronEnqueue on 8 replicas, instead of app.cron().
8. Process-local cache sweep: app.cron()
9. LLM batch off the API: 202 + job id. The client polls your status route.
10. Graph / Entra invite user: entra.invite with { userId }. Your 201 is the local user row. Fatal on 404 user deleted.
11. Data export (GDPR dump): the worker writes the zip to blob storage, then enqueues email.export_ready. That is a job chain.
12. Idempotent payment capture: Stripe Idempotency-Key set to the same natural key as the job.
13. Test suite: Memory + runOnce() + fake timers. No Redis in CI.
14. MCP / agent tool that would time out the client: return a job id fast, then let the agent poll.

Runtime matrix

Security model

Config-time throws: concurrency < 1, maxAttempts < 1, payloadMaxBytes < 1, a missing store, cronEnqueue before useJobs, a handler name outside the charset, or a duplicate define. Boot warnings: Memory in production, startWorker without handlers, concurrency > 32.

Jobs next to cron, idempotency, and webhooks

Anti-patterns

1. Enqueue before the DB commit. The job runs, the row is missing, and you get spurious retries. Commit first, then enqueue (or use a transactional outbox: insert the order and the outbox row in one SQL transaction, and let a drainer call JobStore.put).
2. File bytes in the payload. Payloads cap at 64 KiB. Store the blob, enqueue the URL.
3. startWorker: true on every API replica and a worker deployment, without idempotency keys. That is the duplicate-send scenario. Pick one topology per queue, and always set keys.
4. Catch-all handler (handlers[job.name] = dynamicImport). Forbidden. A store record must never pick the code that runs it. The registry is frozen at construction for this reason.
5. Jobs as a distributed cron lock without keys. “Only one nightly run in the cluster” works because of the per-slot idempotency key. The queue alone does not provide that uniqueness. Without a key there is no uniqueness guarantee.
6. Treating completed as exactly-once evidence for money movement. A completed job proves the handler finished at least once. Money needs the downstream idempotency key (or a ledger transaction you own).
7. MemoryJobStore behind a load balancer. Each replica gets its own private queue. Use a shared store, or accept that jobs are process-local.

Glossary

Queues borrow vocabulary from three unrelated worlds: Java service loading, distributed locking, and cloud message brokers. A reader who knows one of the three still trips on the other two. Every term this page uses is defined below, in the sense DaloyJS means it.

The interface

Delivery, retries, and failure

Ownership and lifetime

Patterns named on this page

Not in v1

• First-party @daloyjs/jobs-postgres / Redis adapters as separate packages with their own dependencies
• A signed job-completion HTTP callback helper (auth + idempotency)
• Recurring job definitions beyond cronEnqueue
• Job metrics / OTLP export
• Cloudflare waitUntil run-once semantics
• Workflow engines as store backends, and batch enqueue

Until those land, implement JobStore in your repo and keep the handler registry explicit. The companion blog post covers the design argument. examples/jobs-basic.ts is the runnable reference.