@shivaedev/jobs

Run PostgreSQL jobs through the same Effect services as the rest of your application. A queue has one payload Schema for enqueueing and handling, and an Effect Layer owns the pg-boss client, workers and schedules.

Why you want this

A background worker lives outside the request that queued it. Without a shared boundary, the producer writes one payload shape, the worker assumes another, and every worker rebuilds its dependencies and shutdown plumbing. This package puts those decisions in one place:

import { defineQueue } from "@shivaedev/jobs/server/definition.ts";
import { Effect, Schema } from "effect";
import { Mail } from "./services.ts";

const SendEmail = defineQueue({
	name: "send-email",
	schema: Schema.Struct({ userId: Schema.NumberFromString }),
});

const emailWorker = SendEmail.handle(({ userId }) => Effect.flatMap(Mail, (mail) => mail.send(userId)));

Mail is the application's Effect service, defined in the guide below. The worker receives a number even though the durable payload stores a string. The same Schema encodes the producer's payload and decodes the worker's input; malformed stored data fails before the handler runs. Provide the worker's services when the jobs Layer starts, and the handler uses that captured context when pg-boss calls it later.

Using it

How to think about it

A queue definition names a pg-boss queue and the Schema of its payload. Application code uses the Schema's decoded type; pg-boss receives its encoded type. Both must be objects. Defining a queue describes the contract; queue.handle(handler) pairs it with the Effect that performs the work.

A schedule definition describes a queue with a cron trigger. schedule.run(effect) pairs it with an Effect that needs no payload. Queue workers and scheduled workers are registrations: the values you put in the jobs Layer's jobs list.

The application creates one jobs service with makePgBoss(identifier, queueNames?). Its Layer reads the shared PgBoss service, registers the job list and captures the services those workers need. One pgBossLayer owns the client and pool for the application; feature registrations drain before that Layer closes its client. Callers use the jobs service to enqueue a decoded payload, look up or retry a job, read queue health and live counts, or list failed and dead-lettered jobs.

The client, inspection reads and registered queue names are Layer-provided services. Core operations yield those services from the environment and keep no dependencies in initializer state. Jobs.layer(...) and the standalone PgBossInspection.layer provide their internal services together; applications need no additional Layers. Client operations emit PgBossOperations.* and PgBossInspectionReads.* spans. Enqueueing captures the caller’s span before the storage operation, so a worker still links to the request or job that queued it.

Use a distinct literal identifier for each jobs service. The service and its Layer keep that identifier in their types, so providing a Layer for a different identifier leaves the required jobs service unprovided at compile time.

The work separates naturally: declare the application services once, define a queue and its worker for each feature, provide the complete registration list at the runtime boundary, and enqueue from ordinary application Effects.

Contribute workers to application modules

jobModule from @shivaedev/jobs/server/jobModule.ts returns queues and workers fields for an AppModule in @shivaedev/server. Its queue declarations come from the same registrations its worker Layer runs:

import { Effect } from "effect";
import { jobModule } from "@shivaedev/jobs/server/jobModule.ts";
import { Jobs, SendEmail, Mail } from "./app.ts";

const jobs = jobModule(Jobs.layer, {
	jobs: [
		SendEmail.handle(({ userId }) =>
			Effect.gen(function* () {
				const mail = yield* Mail;
				yield* mail.send(userId);
			}),
		),
	],
});

Spread these fields into the owning module. Keep the producer's jobs Layer in its services or ports. The application builds appModules once and passes its result to appHost or workerHost; both hosts provide the module's services and ports to their work. Queue definitions without handlers and scheduled workers can also appear in the registration list.

Share queue names with the browser

Keep names and payload Schemas in a browser-safe module such as queues.ts:

import { Schema } from "effect";

export const emailQueue = { name: "send-email", schema: Schema.Struct({ userId: Schema.Number }) } as const;
export const cleanupSchedule = { name: "cleanup-mail" } as const;
export const queueNames = [emailQueue.name, cleanupSchedule.name] as const;

The client imports queueNames for its queue filter, or passes emailQueue to the admin jobs page. Server definitions reference the same declarations:

import { defineQueue } from "@shivaedev/jobs/server/definition.ts";
import { defineSchedule } from "@shivaedev/jobs/server/defineSchedule.ts";
import { makePgBoss } from "@shivaedev/jobs/server/service.ts";
import { cleanupSchedule, emailQueue, queueNames } from "./queues.ts";

export const SendEmail = defineQueue({ ...emailQueue, queue: { retryLimit: 5 } });
export const Cleanup = defineSchedule({ ...cleanupSchedule, cron: "17 3 * * *" });
export const Jobs = makePgBoss("@app/Jobs", queueNames);

With the second argument, every Jobs.layer({ jobs }) checks its registrations against the declared names before registering queues, workers or schedules. A server name absent from the declaration or a declared name absent from the server fails startup with PgBossQueueNamesError from @shivaedev/jobs/server/error.ts; its missing and unexpected arrays identify both differences. Order does not matter. Producer queue definitions, queue workers and scheduled workers all contribute their queue names. An empty declaration accepts only an empty registration list. Without the second argument, the service keeps ordinary registration behavior.

Declare names per jobs service, so producer and worker Layers share a declaration when they register the same queues. Keep worker-only schedules in their own service when producers do not register them. Import only the browser-safe declaration module from client code; defineQueue, defineSchedule and makePgBoss stay on the server.

Label queues and identify jobs

Queue and schedule definitions accept an optional label. A queue's subject declares display columns by naming fields in its decoded payload Schema:

const SendEmail = defineQueue({
	name: "send-email",
	label: "Email delivery",
	schema: Schema.Struct({ userId: Schema.NumberFromString }),
	subject: [{ field: "userId", label: "Recipient" }],
});

subject fields are checked against the Schema's decoded keys. Keep labels and columns beside the name and Schema in the browser-safe declaration when using the admin Jobs view, then spread that declaration into defineQueue. The browser-safe types QueueDisplay and SubjectColumn<PayloadSchema> come from @shivaedev/jobs/display.ts. Pass the declarations to the admin page and view to show their labels and subjects. Subject columns read the masked inspection payload, so native Redacted values and sensitive keys stay masked in failed and dead-letter lists. Dead-letter copies stay with the source queue and use its columns. Definitions without labels display their internal names.

Once per database: install the schema

For an application on the app host of @shivaedev/server, register pgBossSchema beside the migration loader. The host's migrate mode installs or upgrades it after committing the application ledger. Its check and wait modes include pg-boss readiness in the same check as the numbered migrations.

import { pgBossSchema } from "@shivaedev/jobs/server/schema.ts";
import { appDatabase } from "@shivaedev/server/appDatabase.ts";

const database = appDatabase({
	migrations: {
		loader,
		components: [pgBossSchema({ schema: "app_jobs" })],
		mode: "migrate",
	},
});

The component uses installPgBossSchema as its only install and upgrade path. It is pending when the schema is missing, its stored version is below the installed pg-boss package's required schema version, or its dead-letter resolution trigger is missing. Version errors name the schema and both versions (missing for an absent version); a missing trigger is named in its error. Start every producer and worker with the same schema and migrate: false. Use host mode "wait" for services that must wait for a deployment's migration process; no additional jobs-layer wait is needed.

With server's appDatabase({ migrations: { mode: "wait", release, ... } }), each jobs Layer automatically reads sql's optional WorkerRelease Context service. When the database belongs to another release, it calls pg-boss offWork for every registered worker queue, lets active jobs finish, logs Workers paused for another database release once, and idles. Producers and enqueueing continue. When its release owns the marker again and old workers have drained, it registers workers through the normal path and logs Workers resumed for database release. Equality permits rollbacks; no version ordering or payload compatibility policy is added. Without that service, the Layer registers its workers at startup and never pauses them.

A paused worker keeps its process registered with sql's worker release while active jobs drain, so the owning release waits for it; sql's worker release owns the heartbeat and how long a crashed worker can block. Closing the jobs Layer's scope waits for the native pg-boss stop before releasing the services the Layer captured. Choose the three bounds together: each queue's expireInSeconds, a stop: { graceful: true, timeout } in milliseconds that covers it, and a host termination grace beyond that timeout. For example, a queue with expireInSeconds: 60 pairs with stop: { graceful: true, timeout: 65_000 } under a 75-second termination grace.

Worker registrations drain concurrently when their Layer closes. stop.timeout bounds each drain in milliseconds, defaulting to 30 seconds. Work that finishes within that budget completes normally. At the timeout, the adapter logs a warning naming the queue and interrupts its in-flight Effect fibers before releasing their services. stop.graceful: false interrupts without waiting for drain. Keep handlers interruptible, as described under batch expiry. The drain budget does not bound PostgreSQL disconnect acknowledgements; shutdown still waits for the pool to close.

installPgBossSchema installs or upgrades pg-boss through the provided Effect SQL client without starting workers or schedules. Run it outside an application transaction, in a release step or the fixtures Effect of makePostgresTestLayer, so stories inherit the schema from the migrated template:

import { installPgBossSchema } from "@shivaedev/jobs/server/schema.ts";

const jobsFixture = installPgBossSchema({ schema: "app_jobs" });

The Effect requires SqlClient and fails with PgBossError whose operation is "install". The schema defaults to "pgboss". Repeating installation preserves existing jobs. Upgrades apply pg-boss's migration SQL and concurrent index builds, including partition indexes, before returning. Pending index commands are recorded in pg-boss's BAM table in the same transaction as the schema version. A retry resumes unfinished commands and repairs invalid indexes left by failed builds, even when the schema version has already committed.

After installation, pass the same schema and migrate: false to pgBossLayer to start producers and workers without schema installation or migration. Startup fails if the schema is missing. Ordinary Layer startup retains pg-boss's default migration behavior when migrate is omitted.

Once per application: the jobs service

Define the application's jobs service and worker dependencies in services.ts. The application's #mail alias points to its mail service folder:

import { makePgBoss } from "@shivaedev/jobs/server/service.ts";
import { Mail } from "#mail/service.ts";

export { Mail };
export const Jobs = makePgBoss("@app/Jobs");
export const MailLive = Mail.layer;

Jobs is the service that callers yield. Mail is a separate domain service that workers use. Its service.ts declares its requirements once and inventories operations imported from sibling files:

import { Effect } from "effect";
import { defineService } from "@shivaedev/service/defineService.ts";
import { removeExpired } from "./removeExpired.ts";
import { send } from "./send.ts";

export const Mail = defineService({
	id: "@app/Mail",
	requires: [],
	initialize: Effect.void,
	methods: () => ({ removeExpired, send }),
});

send.ts holds one traced operation:

import { Effect } from "effect";

export const send = Effect.fn("Mail.send")((userId: number) => Effect.log({ event: "send_email", userId }));

removeExpired.ts holds the other operation:

import { Effect } from "effect";

export const removeExpired = Effect.fn("Mail.removeExpired")(() => Effect.log({ event: "remove_expired_mail" }));

The example logs domain calls; the application supplies its mail implementation. An operation that needs another service yields it from the environment, and requires declares that service on Mail. Its initializer holds only owned lifetime state. The jobs package does not need a provider-specific interface.

Registration collections retain their workers' handler, payload-decoding and scheduled-effect service requirements in Jobs.layer. Use satisfies readonly JobRegistration[] to check a reusable collection while preserving its inferred requirements, or annotate it as readonly JobRegistration<Services>[] with the union of services it needs. A plain JobRegistration[] leaves requirements unknown. Queue workers require both their queue and handler; scheduled workers require both their schedule and effect. Incomplete registrations are TypeScript errors. Producer-only queue definitions need no handler or decoding services at startup.

Once per feature: the payload and handler

Define the feature's registrations in jobs.ts:

import { defineQueue } from "@shivaedev/jobs/server/definition.ts";
import { defineSchedule } from "@shivaedev/jobs/server/defineSchedule.ts";
import { Effect, Schema } from "effect";
import { Mail } from "./services.ts";

export const SendEmail = defineQueue({
	name: "send-email",
	schema: Schema.Struct({ userId: Schema.NumberFromString }),
	queue: { retryLimit: 5 },
});

export const emailWorker = SendEmail.handle(({ userId }, job) =>
	Effect.gen(function* () {
		const mail = yield* Mail;
		yield* mail.send(userId);
		yield* Effect.log({ jobId: job.id });
	}),
);

export const Cleanup = defineSchedule({
	name: "cleanup-mail",
	cron: "17 3 * * *",
});

export const cleanupWorker = Cleanup.run(Effect.flatMap(Mail, (mail) => mail.removeExpired()));

The queue takes { userId: number } in application code and stores { userId: string }. Its handler receives the decoded number and a job context, including the job's id, retryCount and deferralCount. Both counts start at zero. retryCount is the native retry count; deferralCount counts accepted RetryAt deferrals before this attempt, independently of native retries. Keep the queue definition available to producers; the runtime registers the worker created by handle.

The schedule runs a payload-free Effect. This schedule registers its cron expression with tz: "UTC". The package creates a <name>-dlq dead-letter queue for both kinds of worker and supplies retryLimit: 3 and retryBackoff: true by default. The email queue above overrides the retry limit to five. These are pg-boss registration settings; pg-boss runs the retry and scheduling engine.

Only a Layer whose jobs list contains a schedule registration such as cleanupWorker registers that cron. A request runtime that declares only queues never runs scheduled Effects. A development process that starts the worker runs its schedules locally too; leave schedule registrations out of that worker when the app should not run them in development.

The Layer requires the workers' Effect services. Leaving Mail unprovided leaves that requirement in the resulting Effect's type. A primitive Schema such as Schema.String is rejected as a queue definition because the durable contract must be an object.

Retire queues and schedules

Declare obsolete queue names in the worker Layer:

const live = Jobs.layer({ jobs: [sendEmailWorker, cleanupWorker], retired: ["old-mail", "old-cleanup"] });

Startup stops local workers for each retired name and deletes its queue through pg-boss, including its schedules and queued jobs, before registering current workers. Missing queues can be retired again on subsequent starts. With WorkerRelease, retirement waits until this deployment holds the database release, including Layers with no current jobs. Names still registered by the Layer, and their dead-letter queue names, are rejected before any deletion.

Dead-letter queues remain unless explicitly included in retired; list a source queue before its dead-letter queue when retiring both. Put retirement in the worker deployment that owns those queues and remove the old registrations from other running deployments. Custom clients need the native deleteQueue capability for retirement; PgBossClient also exposes optional native unschedule for removing a single schedule while keeping its queue.

Backlog throughput and concurrency

Queue and schedule workers default to batchSize: 10 and burstWhenBatchFull: true. pg-boss fetches again immediately after a full batch completes, until a fetch returns fewer than ten jobs. The adapter runs handlers sequentially within each batch, and pg-boss defaults localConcurrency to one. Set native localConcurrency to run multiple batches at once:

const Compare = defineQueue({
	name: "compare",
	schema: Schema.Struct({ id: Schema.Number }),
	queue: { notify: true },
	worker: { batchSize: 2, localConcurrency: 5 },
});

This worker runs up to five handlers at once. worker accepts jobs' WorkerOptions, including batchSize, localConcurrency, pollingIntervalSeconds, notifyPollingIntervalSeconds and the burst options. Explicit options override the adapter defaults. pg-boss ignores burstWhenBatchFull at batchSize: 1; keep a batch size of at least two for immediate backlog draining. Each job is settled separately through pg-boss: success completes that job, and a typed failure, payload decoding failure or defect fails only that job with its own error. A failed job does not spend the retries of other jobs in its batch or rerun their successful handlers. Interruption stops the batch and leaves the interrupted and remaining jobs to pg-boss expiry; handlers must still tolerate repeated side effects across retries and interrupted attempts.

Apps do not need to shorten the polling backstop to drain a backlog. With notify: true and an active LISTEN/NOTIFY listener, pg-boss wakes workers on notifications and defaults to a 30-second backstop (notifyPollingIntervalSeconds). Without active notifications its base poll defaults to two seconds (pollingIntervalSeconds). Bursting takes precedence over both intervals. After an adapter worker fails or defers a job, it reads pg-boss's stored retry time and schedules a wake for every native worker in its registration at that time, so an idle sibling can retry while another worker is busy. The wake accounts for differences between the runtime and PostgreSQL clocks. The wake, including its retry-time read, belongs to the worker Layer's scope and is canceled when that scope closes. Wake failures are logged with redacted diagnostics and do not fail the batch. Override the backstop when the application needs a different bound for other work that becomes runnable without a notification, such as delayed jobs or retries initiated outside the adapter worker.

Defer until a provider is ready

Fail a queue handler or scheduled Effect with RetryAt when a provider gives a time to try again:

import { DateTime, Effect } from "effect";
import { RetryAt } from "@shivaedev/jobs/server/definition.ts";

const waitForQuota = Effect.fail(
	new RetryAt({
		at: DateTime.makeUnsafe("2030-01-01T00:01:00Z"),
		reason: "Provider quota resets",
	}),
);

The same job returns to waiting with startAfter set to at; its ID, singleton key, payload and retry count stay unchanged. A past or current at is eligible on the next fetch. Deferral does not dead-letter the job or call onError. Its job span ends OK with a job.deferred event whose at is an ISO UTC string and whose reason is the supplied string. The adapter wakes workers in the same registration when the deferral becomes due, including queues using notify: true, so they can resume the job without waiting for the polling backstop. Stately jobs sharing a singleton key can defer together and still run one at a time. While waiting, their native job metadata reports the adapter's effect-stately-deferred policy; a database trigger restores stately when pg-boss claims them, preserving its active-job exclusion.

Set queue: { maxDeferrals: 20 } on a queue or schedule to change the default limit of ten deferrals per job. The limit must be a nonnegative safe integer; zero disables deferral. Once the limit is reached, RetryAt follows the normal failure and retry path. The adapter stores the count in its own table in the pg-boss schema, preserving it across worker restarts, ordinary failures, expiry and heartbeat timeouts. Queue handlers read this table before every attempt, including handlers that never defer. Automatic migrations and installPgBossSchema create this table and the stately claim trigger; use installPgBossSchema or the pgBossSchema migration component before starting workers with migrate: false.

A queue handler can use job.deferralCount to grow its RetryAt backoff without an application column. The count survives worker restarts and native retries; a deferral leaves job.retryCount unchanged.

One active job per entity

Put related job kinds in one tagged-union queue and choose the entity key in the definition:

const PullRequestJobs = defineQueue({
	name: "pull-request",
	schema: Schema.Union([
		Schema.TaggedStruct("Refresh", { pullRequestId: Schema.NumberFromString }),
		Schema.TaggedStruct("Reconcile", { pullRequestId: Schema.NumberFromString }),
	]),
	queue: { policy: "singleton" },
	key: ["pullRequestId"],
	worker: { batchSize: 1, localConcurrency: 3 },
});

const refresh = Effect.flatMap(Jobs, (jobs) => jobs.enqueue(PullRequestJobs, { _tag: "Refresh", pullRequestId: 42 }));

queue.policy accepts pg-boss's policy literals and rejects typos. The singleton policy requires key: ["field", ...]. Each part names a required scalar field in the decoded payload schema, so pullRequestId contributes its numeric value. enqueue supplies pg-boss's singletonKey automatically; callers cannot supply a conflicting key in send options.

PostgreSQL guarantees one active job per key across workers, including different job kinds in this queue. Different keys can run in parallel when workers have capacity. A job waiting to retry frees its key for another job; exhausting retries sends it to the dead-letter queue and leaves the key available.

Singleton guarantees one active job per key, not order. Ordering stays the application's job. Keep handlers idempotent and write state only when the source is newer than what is stored, so an older event or retry cannot overwrite newer state.

Deterministic job IDs

Declare the payload fields that identify a job:

const Revisions = defineQueue({
	name: "revisions",
	schema: Schema.Struct({ tenant: Schema.String, revision: Schema.NumberFromString, note: Schema.String }),
	id: ["tenant", "revision"],
	key: ["tenant"],
	queue: { policy: "singleton" },
});

id derives a deterministic UUID from the queue name and its ordered payload parts. Enqueuing the same parts again returns None while pg-boss retains that job, even if other fields change. Different queue names have separate identities. Here, different revisions create separate jobs while sharing the tenant's singleton key. Callers cannot pass an id or singletonKey that the queue owns.

Both declarations accept fields whose decoded values are strings, numbers, booleans or null. Parts preserve value types and field boundaries, including separators in strings. An empty list gives every payload the same identity within the queue. Omit id to keep pg-boss's generated IDs or caller-supplied IDs. Identity is separate from tracing metadata.

Size batches against expiry

pg-boss marks every job in a fetched batch active together. Each job's expiry window starts at its stored startedOn, including time waiting for earlier handlers. Size each queue's batch so its sequential work fits within that window; use a small batch for long-running work and keep the default for short work:

const Ingest = defineQueue({
	name: "ingest",
	schema: Schema.Struct({ sourceId: Schema.String }),
	queue: { expireInSeconds: 300 },
	worker: { batchSize: 1 },
});

const IndexTitle = defineQueue({
	name: "index-title",
	schema: Schema.Struct({ titleId: Schema.String }),
});

Ingest fetches one job at a time; IndexTitle keeps the default batch of ten. Configure the expiry for the work's expected duration. Increasing localConcurrency adds batches; it does not extend any job's window.

The adapter enforces each queue or scheduled job's deadline using pg-boss's startedOn and expireInSeconds. It reads PostgreSQL time once per claimed batch and translates the remaining expiry to the runtime Clock, so a different runtime date does not change the budget. It skips jobs whose window has already passed and interrupts a running Effect at its deadline. Expired jobs remain for pg-boss's expiry and retry engine. Interruption stops the rest of that batch, as described above.

An interrupted batch waits for pg-boss to release its original claims before returning successfully. Settlement reads use exponential backoff from 25 ms to 500 ms between reads, with a bound of one second after the latest fetched job's deadline, including a read that never resolves. If settlement cannot be observed within that bound, the callback fails with PgBossError (operation: "settle", with a redacted TimeoutError and its readable tag and message) and logs the error through the worker's captured Effect context.

Queue handlers can yield job.remainingBudget, an Effect.Effect<Duration.Duration> that recomputes the time left at each read and clamps it to zero. Use it to size a downstream operation's timeout:

import { Duration, Effect } from "effect";

const ingestWorker = Ingest.handle(({ sourceId }, job) =>
	Effect.gen(function* () {
		const budget = yield* job.remainingBudget;
		yield* Effect.log({ sourceId, remainingMilliseconds: Duration.toMillis(budget) });
	}),
);

Keep handler work interruptible and use cancellation-aware Effect integrations for external operations. A deadline interrupts the Effect fiber; it cannot stop an uninterruptible region or an external side effect that ignores cancellation. Handlers must tolerate retries.

Once per runtime: provide the workers

Build the jobs Layer in runtime.ts:

import { Layer, type Redacted } from "effect";
import { pgBossLayer } from "@shivaedev/jobs/server/service.ts";
import { cleanupWorker, emailWorker } from "./jobs.ts";
import { Jobs, MailLive } from "./services.ts";

export const jobsLayer = (databaseUrl: Redacted.Redacted<string>) =>
	Jobs.layer({ jobs: [emailWorker, cleanupWorker] }).pipe(
		Layer.provide(MailLive),
		Layer.provide(pgBossLayer({ connectionString: databaseUrl, schema: "app_jobs" })),
	);

The redacted connection string is unwrapped only at the pg-boss adapter. Put connection, pool size, client factory, schema, migration and stop settings in pgBossLayer. Provide the same Layer to the merged registration, worker and producer services so native Layer memoization starts one client. jobs is the complete list of registrations for this Layer. The Layer captures MailLive for both the queue handler and the scheduled Effect. A client startup or queue registration failure fails Layer acquisition with PgBossError or PgBossQueueOptionsError and closes its registrations. A name declaration mismatch fails with PgBossQueueNamesError. The owning app Layer releases the client. Registration creates a queue again, up to six times with a short backoff, when PostgreSQL rolls the creation back as a deadlock victim (SQLSTATE 40P01), which a partitioned queue can meet when other queues in the schema have traffic. A shared client stays running while another scope still owns a reference.

pgBossLayer defaults its pool to maxLifetimeSeconds: 600 and idleTimeoutMillis: 60_000. Override either with the native pool option; zero disables that timeout. Expired connections close when released, and subsequent jobs operations acquire a replacement. An application-supplied db owns its own connection lifecycle.

Keep the Layer in the application's worker runtime scope. A short scope is useful for an acquisition or health check; a worker process needs its scope to remain open while it handles jobs.

Set clientCacheKey on pgBossLayer to share one reference-counted client across Layer builds with equal pg-boss constructor settings. Reusing a key with different settings fails acquisition with PgBossClientOptionsError before constructing another client. Its key, options and message identify the key and differing option names without exposing connection values. Nested options are compared by value.

import { Effect, type Redacted } from "effect";
import { jobsLayer } from "./runtime.ts";
import { Jobs } from "./services.ts";

const inspectJobs = (databaseUrl: Redacted.Redacted<string>) =>
	Effect.scoped(
		Effect.gen(function* () {
			const jobs = yield* Jobs;
			return yield* jobs.health();
		}).pipe(Effect.provide(jobsLayer(databaseUrl))),
	);

This Effect starts the jobs Layer, reads health and closes the scope. The client stops when that scope closes.

In application code: enqueue

import { Effect } from "effect";
import { SendEmail } from "./jobs.ts";
import { Jobs } from "./services.ts";

export const enqueueEmail = (userId: number) =>
	Effect.gen(function* () {
		const jobs = yield* Jobs;
		return yield* jobs.enqueue(SendEmail, { userId });
	});

Provide Jobs through the application's existing runtime. enqueue takes the decoded payload, so passing a string userId is a type error. It encodes the number to a string before calling pg-boss. The result type is Option.Option<string>: a job identifier becomes Some, and a null send result becomes None.

PgBossPayloadError names a payload boundary failure; malformed durable input is rejected before the worker's domain code runs.

Trace jobs

Each queue or scheduled job attempt runs in one root consumer span named for its queue. The span encloses payload decoding and the whole handler, so top-level Effect.fn steps are children of the job span. Its attributes include job.id, job.queue, job.name, the one-based job.attempt and pg-boss's zero-based job.retry_count. A failed handler ends the span with its failure.

enqueue captures the current Effect span, including an external parent span, when present. It stores that context in a typed envelope in pg-boss's existing job data; the adapter unwraps the envelope before Schema decoding in the handler and getJob, so the application's payload Schema stays unchanged. Jobs sent without a span keep their ordinary encoded payload, and workers also accept those unwrapped jobs. Direct pg-boss data readers see the envelope for traced sends.

The consumer span links to the sender rather than sharing its parent trace, following OpenTelemetry messaging conventions for asynchronous processing. Each retry gets its own job trace. Provide the application's Effect tracer when building the worker Layer; the worker uses that captured tracer.

Read health

import { Effect } from "effect";
import { Jobs } from "./services.ts";

export const readJobCounts = Effect.gen(function* () {
	const jobs = yield* Jobs;
	const health = yield* jobs.health();
	return {
		active: health.activeTotal,
		deadLettered: health.deadLetteredTotal,
		failed: health.failedTotal,
		queued: health.queuedTotal,
		ready: health.readyTotal,
	};
});

health exposes pg-boss cached queue counts as data. It stays cheap and follows the pg-boss monitor interval; it does not establish that a queue has just drained. Its total fields include the registered queue's active, failed, queued and ready counts, plus the dead-letter queue's queued count. Decide in the application which values affect an HTTP health response or an alert.

health([{ name: "cleanup-mail" }]) reads only the supplied queues, in order, even when another runtime registered them. Omitting the argument keeps the Layer's registered names. Admin pages use this to include worker-only schedules without registering those schedules in the request runtime.

Inspect stored work

PgBossInspection from @shivaedev/jobs/server/inspection.ts exposes the shared read service used by admin pages and the testing controller. Provide PgBossInspection.layer with your pgBossLayer, or obtain the same reads from a configured jobs runtime with yield* jobs.inspection().

storedJobs(queue, options?) reads native stored metadata and encoded payloads. jobs(queue, options?) decodes stored payloads through the supplied queue Schema and includes native state, startAfter, expireInSeconds, startedOn and keepUntil. These reads include unredacted payloads; keep them inside trusted server code. The admin-facing getJob, failedJobs and deadLetteredJobs reads retain their payload redaction. deferred(queue) reads stored RetryAt reasons and requested times, while snapshot(queues) lists unfinished work with future RetryAt attempts marked deferred. counts, health, queue, schedules, schemaVersion and busy serve queue diagnostics and controller readouts.

changes(queue?) acquires a scoped PostgreSQL listener and returns revision and waitForChange(after). Capture its revision before reading, then wait against that revision to avoid missing a change between the read and the wait. The signal wakes for committed job changes from any client, optionally filtered to one queue. Local deferrals advance the revision before the worker continues its batch, without waiting for PostgreSQL notifications. It is a hint to read again, not a durable event log. The caller owns the listener's scope.

Automatic migration and installPgBossSchema install the queue-change trigger. Applications using migrate: false run installPgBossSchema before using change signals. Custom clients need getDb().listen to subscribe.

Administer queues and schedules

runScheduleNow(schedule) enqueues one immediate run using the registered schedule's queue, payload and options, with startAfter: 0. Pass the schedule definition, including its scheduleOptions.key when keyed. The worker processes it as a normal job, and the cron registration stays unchanged. The result is Option<string>, containing the new run's ID when pg-boss accepts it. A missing registration fails with PgBossNotFoundError.

const jobs = yield* Jobs;
const runId = yield* jobs.runScheduleNow(Cleanup);

unschedule(schedule) removes the registration named by the schedule definition and its scheduleOptions.key (the empty key by default). Other keys for that queue stay registered. A missing registration fails with PgBossNotFoundError, carrying operation: "unschedule", queue and key.

deleteQueue(queue) stops local workers for that name and deletes the queue through the owned client. Other queues stay in place. A missing queue fails with PgBossNotFoundError with operation: "deleteQueue" and queue; a native refusal, such as deleting a dead-letter queue still in use, fails with PgBossError.

refreshQueueStats(queue?) requests a native pg-boss supervisor pass for one queue, or all queues when omitted, so health can read refreshed counts. The pass also performs native queue maintenance and respects pg-boss's monitor intervals and backoff; it does not promise a fresh snapshot on every call. Use counts for an immediate live read.

Read live counts and retry a job

import { Effect, Option } from "effect";
import { SendEmail } from "./jobs.ts";
import { Jobs } from "./services.ts";

export const inspectEmail = (id: string) =>
	Effect.gen(function* () {
		const jobs = yield* Jobs;
		const job = yield* jobs.getJob(SendEmail, id);
		if (Option.isSome(job) && job.value.state === "failed") {
			yield* jobs.retry(SendEmail, id);
		}
		const counts = yield* jobs.counts(SendEmail);
		return counts.queuedCount === 0 && counts.activeCount === 0;
	});

counts(queue) reads the job table directly on each call. queuedCount includes created and retry-waiting jobs, including future jobs; activeCount counts running jobs, failedCount counts retained failures, and totalCount includes retained terminal jobs. A queue has drained when queued and active counts are both zero. These counts are a snapshot, so a producer can enqueue more work after the query returns.

getJob(queue, id) returns None when the ID is absent, or Some jobs-owned Job<unknown> metadata with masked inspection data: unknown. The queue's Schema decodes the stored payload before the errors package's redactor masks native Redacted values and sensitive keys, including nested objects and arrays. Invalid stored data fails with PgBossPayloadError, and the Effect requires the Schema's decoding services. retry(queue, id) reuses the original ID: it makes a failed job eligible for another attempt and pulls a retry-waiting job forward to run now. Missing IDs and jobs in other states are left unchanged. Its dead-letter copies stay listed while the retry is pending or fails again; a successful completion removes those copies in the same database transaction.

List failed and dead-lettered jobs

import { Effect, Option } from "effect";
import { SendEmail } from "./jobs.ts";
import { Jobs } from "./services.ts";

export const failedEmails = (cursor?: string) =>
	Effect.gen(function* () {
		const jobs = yield* Jobs;
		const failed = yield* jobs.failedJobs(SendEmail, { cursor, limit: 20 });
		const deadLettered = yield* jobs.deadLetteredJobs(SendEmail, { limit: 20 });
		return {
			deadLettered: deadLettered.jobs.map((job) => ({ id: job.id, original: Option.getOrNull(job.sourceId), error: job.sourceOutput })),
			failed: failed.jobs.map((job) => ({ id: job.id, attempt: job.attempt, error: job.output, data: job.data })),
			next: Option.getOrUndefined(failed.next),
		};
	});

health and counts say how many jobs failed; these two listings say which ones and why. failedJobs(queue, options) lists the queue's jobs in the failed state, the ones that used up their retries. deadLetteredJobs(queue, options) lists every job in the queue's dead-letter queue. Dead-letter entries live in the -dlq queue: each is a new job with its own ID that pg-boss copied there when the original failed for good. Both listings read pg-boss's job table on each call, through the jobs Layer's own pg-boss connection and in its configured schema, so applications never query pg-boss tables themselves. A custom client needs the native getDb capability for them. Automatic migration and installPgBossSchema install the completion trigger that clears resolved copies, including copies whose retained source already completed. Applications using migrate: false must run installPgBossSchema when upgrading.

Each page holds at most limit jobs, newest first by creation time; a dead-letter copy is created when its original fails for good. limit must be a positive integer. When more jobs remain, next is Some opaque cursor; pass it back as cursor for the following page, and stop when next is None. Paging continues from the last job it returned, so jobs that fail while you page appear in a fresh listing instead of shifting later pages, and no job appears twice. A malformed cursor or limit fails with SchemaError before pg-boss is called.

An entry carries the job's id, queue, native state, retryCount, retryLimit and attempt, the one-based number of its last started attempt (zero for a job that never started). createdOn is a UTC DateTime; startedOn and completedOn are Options of UTC DateTimes. output is JobFailure | null, decoded through JobFailure from @shivaedev/jobs/JobFailure.ts. Handler failures store only { _tag: string, message: string }, with the message redacted through @shivaedev/errors; stack traces, cause chains and other error fields are omitted. null means pg-boss stored no output. Direct failures and native pg-boss failures wrapped in { value: failure } share the same sanitized readout in details and listings. Failures without a tag decode with _tag: "Error"; decoding also redacts messages from older stored failures. Stored output that does not match the Schema fails the listing with PgBossPayloadError. data: unknown is decoded by the queue's JSON codec and masked through the errors package's redactor, including traced payloads; invalid stored data fails the listing with PgBossPayloadError, as getJob does, and the Effect requires the Schema's decoding services. A dead-letter entry adds sourceQueue, the source queue name without the dead-letter suffix, sourceId, the original job's ID as an Option, and sourceOutput, the original's decoded JobFailure | null. The copy has not run, so its own output stays null and its attempt is zero.

readJobFailure(queue, output) from @shivaedev/jobs/server/readJobFailure.ts exposes this readout for adapters. It returns an Effect with JobFailure | null and fails with PgBossPayloadError when stored output cannot be decoded.

Pass a failed entry's id to retry(queue, id) to run that job again under its original ID; once it completes it leaves the failed listing and its copies leave the dead-letter listing. Dead-letter IDs belong to the -dlq queue, so retry(queue, id) does not find them and leaves them unchanged.

Queue settings on startup

queue accepts pg-boss Queue options except name and deadLetter, which the definition and adapter own, plus the adapter’s maxDeferrals limit. For prompt delivery, set queue: { notify: true } and enable useListenNotify: true in the worker Layer's constructor options.

Startup creates missing queues and reconciles declared settings on existing queues through updateQueue. Unspecified settings keep their stored values; the adapter always supplies its default retry settings and dead-letter destination. Changing an explicitly declared policy or partition on an existing queue fails startup with PgBossQueueOptionsError; its queue, options and message identify the conflict. Queue setting changes apply to subsequently enqueued jobs. Schedule queues follow the same registration path.

Share a client between Layer builds

import { Layer, type Redacted } from "effect";
import { pgBossLayer } from "@shivaedev/jobs/server/service.ts";
import { cleanupWorker, emailWorker } from "./jobs.ts";
import { Jobs, MailLive } from "./services.ts";

const sharedJobsLayer = (databaseUrl: Redacted.Redacted<string>) =>
	Jobs.layer({ jobs: [emailWorker, cleanupWorker] }).pipe(
		Layer.provide(MailLive),
		Layer.provide(pgBossLayer({ connectionString: databaseUrl, schema: "app_jobs", clientCacheKey: "app-jobs" })),
	);

Overlapping Layer builds with the same clientCacheKey share one started client. The later build replaces the registered worker, and the final scope release stops the shared client. A later build after that release starts a new client. Within one Layer graph, merged feature services share the provided app client without a cache key. Separate app Layer builds create separate clients without a key. The application chooses whether to pass a key; give each distinct constructor configuration its own key.

Separate producers and workers

Pass a queue definition directly in the jobs Layer to create its queue and dead-letter queue without starting a consumer:

import { SendEmail } from "./jobs.ts";
import { Jobs } from "./services.ts";

export const producerLayer = Jobs.layer({ jobs: [SendEmail] });

Provide the app pgBossLayer to this Layer in the request runtime and call jobs.enqueue(SendEmail, { userId }) as above. A separate worker process registers SendEmail.handle(...) against the same database. Producer queues use the same retry options and appear in health; only worker registrations capture handler services.

Enqueue follow-ups from a handler

A handler can yield the same Jobs service that owns it. The Layer supplies that service from its scoped client; other handler and Schema service requirements remain required inputs to the Layer.

const Receipt = defineQueue({
	name: "receipt",
	schema: Schema.Struct({ userId: Schema.Number }),
});

const sendAndQueueReceipt = SendEmail.handle(({ userId }) =>
	Effect.gen(function* () {
		const mail = yield* Mail;
		yield* mail.send(userId);
		const jobs = yield* Jobs;
		yield* jobs.enqueue(Receipt, { userId });
	}),
);

const live = Jobs.layer({ jobs: [sendAndQueueReceipt, Receipt] }).pipe(Layer.provide(MailLive));

Include every target queue in jobs, as either a worker or a producer-only definition. Startup creates all queues before starting consumers. The example registers Receipt for another process to consume. A follow-up target can also have a worker in the same Layer. Scheduled effects can yield Jobs too. Keep one jobs Layer in the process's native Layer graph; that Layer owns one client for enqueueing and handling.

Test through the queue contract

Use the controller from test-kits in your application story kit.

API

Import the module that defines the name; the package has no root entry.

| Module | Public names | | --------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | @shivaedev/jobs/display.ts | QueueDisplay, SubjectColumn | | @shivaedev/jobs/server/definition.ts | defineQueue, jobContext, RetryAt; DefineQueueOptions, DefineScheduleOptions, JobContext, JobPayloadSchema, JobRegistration, PayloadParts, QueueDefinition, QueueWorker, RegistrationRequirements, ScheduleDefinition, ScheduledWorker | | @shivaedev/jobs/server/service.ts | makePgBoss, pgBossLayer; PgBossDefinition, PgBossLayerOptions, PgBossOptions, PgBossService | | @shivaedev/jobs/server/PgBoss.ts | PgBoss: the shared client Context service. | | @shivaedev/jobs/server/schema.ts | installPgBossSchema, pgBossSchema; PgBossSchemaOptions | | @shivaedev/jobs/server/health.ts | deadLetterQueueName; DeadLetteredJob, JobPage, JobPageOptions, JobsHealth, ListedJob, QueueCounts, QueueHealth | | @shivaedev/jobs/server/error.ts | PgBossError, PgBossNotFoundError, PgBossPayloadError, PgBossQueueNamesError, PgBossQueueOptionsError, PgBossClientOptionsError, toPgBossError; PgBossOperation | | @shivaedev/jobs/server/queueOptions.ts | QueueOptions, QueuePolicy, RetryOptions | | @shivaedev/jobs/server/workerOptions.ts | WorkerOptions, GroupConcurrencyConfig | | @shivaedev/jobs/server/enqueueOptions.ts | EnqueueOptions, QueueEnqueueOptions, ScheduleOptions | | @shivaedev/jobs/server/job.ts | Job, JobState, WorkerJob | | @shivaedev/jobs/server/clientOptions.ts | ClientOptions, StopOptions, Clock, ReindexOptions, IndexBloatOptions | | @shivaedev/jobs/server/ClientDatabase.ts | ClientDatabase | | @shivaedev/jobs/server/engineClient.ts | EngineClient, EngineSchedule | | @shivaedev/jobs/server/adapters/client.ts | PgBossClient, PgBossClientFactory | | @shivaedev/jobs/server/inspection.ts | PgBossInspection; PgBossInspectionService | | @shivaedev/jobs/server/defineSchedule.ts | defineSchedule | | @shivaedev/jobs/server/adapters/defaultClientFactory.ts | defaultClientFactory |

Jobs declares its public types independently of pg-boss. The option fields keep their familiar names: queue uses QueueOptions, including literal QueuePolicy values and maxDeferrals; worker uses WorkerOptions; a schedule's schedule uses ScheduleOptions; enqueueing uses EnqueueOptions. A queue also requires name and schema; a schedule requires name and cron.

Queue and scheduled workers reject perJobResults: true and transactional: true during Layer acquisition with a registration PgBossError; its redacted original names the unsupported options. Omit these options or set them to false. The adapter settles each job itself and does not expose pg-boss's transaction client to handlers.

pgBossLayer(options) accepts PgBossOptions, built from jobs-owned ClientOptions with a redacted connection string and lifecycle settings. Jobs.layer(options) accepts registrations, retired names and onError. Custom clients must supply native complete and fail capabilities to register queue or schedule workers; missing settlement capabilities fail Layer acquisition with a registration PgBossError. Workers also require getDb() to read the PostgreSQL clock when a batch arrives. Producer-only clients do not need these capabilities. Custom clients can also supply the optional native capabilities findJobs, retry, update, getDb and updateQueue for lookup, retry, live counts and listings, deferral and reconciliation; administration uses optional deleteQueue, getSchedule, getSchedules, unschedule and supervise capabilities. A requested operation fails with PgBossError if its required capability is absent.

onError handles pg-boss client error events using the services captured by the Layer. The Layer interrupts pending callbacks and waits for them to end before release completes. Only the owning jobs service is supplied automatically; a handler using another jobs definition still requires that definition's Layer.

Worker registration always requests includeMetadata: true, including when worker supplies includeMetadata: false; custom clients must return pg-boss's startedOn metadata. Direct callers use jobContext(job, deadline), where deadline is in runtime Clock milliseconds: measure the remaining expiry against PostgreSQL time, then add it to the runtime Clock time. Registered workers receive that translation automatically.

PgBossError has operation, optional queue and a Redacted original. Its message names the operation and the queue. For settle, it includes the original failure's tag and message through the same redaction as stored job failures. Other operations show only the original error's class and code, such as a PostgreSQL SQLSTATE, because the original message can quote payloads or connection strings. A registration failure names the queue it was preparing. PgBossPayloadError has direction, queue and a Redacted original. Those fields distinguish the pg-boss operation from the payload boundary.

Install, setup and limits

pnpm add @shivaedev/jobs @effect/sql-pg effect

The package requires Node.js 24 or later and PostgreSQL. effect, pg-boss and @effect/sql-pg are required peers. The PostgreSQL compiler builds paged listing queries for the existing pg-boss connection; it opens no additional pool. pnpm installs pg-boss as the jobs package’s peer; apps do not list pg-boss themselves.

connectionString configures pg-boss's own pg Pool, separately from the application's Effect SQL pool, and pg-boss's driver decides how its sslmode behaves. Set the TLS mode in that URL; the framework database connection security guide compares the drivers and the application's connection paths.

Provide the database connection and job registrations at the application runtime boundary. Application services own the domain work and provider calls; pg-boss owns queue execution. When SqlClient is provided and a SQL transaction is active, enqueue uses pg-boss’s per-call db option on that transaction’s connection. The application write and queued jobs commit or roll back together. The jobs client and SQL client must use the same PostgreSQL database. Outside a transaction, enqueueing keeps using the jobs client as usual. To keep enqueueing independent of an application transaction, move it after the transaction returns and handle its failure separately.

Secret payload fields

Use native Schema.Redacted or Schema.RedactedFromValue for fields such as one-time links. Jobs uses Effect's JSON codec at the durable boundary, so PostgreSQL stores the real value and the worker receives a Redacted value it can unwrap with Redacted.value. This masks inspection data; it does not encrypt queue storage.

const Confirmation = defineQueue({
	name: "confirmation",
	schema: Schema.Struct({ href: Schema.Redacted(Schema.String) }),
});
const sendConfirmation = Confirmation.handle(({ href }) => sendMail(Redacted.value(href)));

getJob, failedJobs and deadLetteredJobs expose masked diagnostic data, without recoverable Redacted wrappers. Worker log messages and annotations use the errors package's redaction policy too. Keep secrets wrapped when logging or constructing errors; unwrapping a value into arbitrary text removes its sensitivity marker. The existing errors redactor bounds diagnostic depth, collection size and text length.