Durable jobs with better-effect-mq

better-effect-mq adds storage-neutral durable queue capabilities to better-effect. It provides immutable Job definitions, producer operations, a Worker supervisor, and a JobStore Service contract. It does not open a database connection or choose a persistence technology.

At-least-once delivery and optional capabilities

A successful settlement under the current lease is not normally delivered again. If a worker performs a side effect and crashes before settlement is persisted, the job can be delivered again. This is at-least-once, not exactly-once processing. Optional parent/child composition, controls, events, schedules, and outbox publishing are additive capabilities supplied by the core package or an adapter; they do not provide a distributed transaction.

Design handlers so repeating the external operation is safe. Use the Job ID or a business idempotency key with the downstream API, and make the downstream write conditional or deduplicated where possible. A lease token fences an old worker from settling a newer delivery; it cannot undo a side effect that already happened before a crash.

Persisted records are JSON-safe DTOs. They contain identities, state, counters, leases, encoded payloads and bounded failure envelopes—not live Error, Result, Service or connection objects.

For parent/child stores or a new adapter, use the advanced storage adapter guide.

Define a queue and jobs

A queue definition is an inert namespace. A Job definition is immutable and versioned by the literal queue, name and positive integer version. Defining a Job does not register a handler, resolve a Service or open storage. For a boundary payload, use a Zod 4 schema through better-effect-schema and the MQ Codec.standardSchema bridge:

import * as z from 'zod'
import { Codec, JobEncodeFailure, Queue, Retry } from 'better-effect-mq'
import { Schema as CoreSchema } from 'better-effect-schema'
import { Schema } from 'better-effect-schema/zod'

const SendEmailPayloadSchema = z.object({
  messageId: z.string().min(1),
  recipient: z.email()
})
class SendEmailPayload extends Schema.Class<SendEmailPayload>('app/SendEmailPayload')(
  SendEmailPayloadSchema
) {}

const SendEmailPayloadCodec = Codec.standardSchema({
  schema: SendEmailPayload,
  encode: (value) =>
    CoreSchema.encode(SendEmailPayload, value).mapError(
      (error) => new JobEncodeFailure({ message: error.message, code: 'schema-encode' })
    )
})

const Emails = Queue.define('emails')

const SendEmail = Emails.job('send-email', {
  version: 1,
  payload: SendEmailPayloadCodec,
  result: Codec.standardSchema({ schema: z.object({ accepted: z.boolean() }) }),
  failure: Codec.standardSchema({ schema: z.object({ code: z.string().min(1) }) }),
  defaults: {
    attempts: 5,
    backoff: Retry.exponential({
      initialDelayMs: 1_000,
      factor: 2,
      maxDelayMs: 60_000,
      maxAttempts: 5
    }),
    timeoutMs: 30_000
  },
  idempotencyKey: ({ messageId }) => messageId,
  retryable: ({ code }) => code !== 'recipient-blocked'
})

payload is the input/value boundary for the payload codec. result and failure codecs are optional; without them, the corresponding persisted result or typed failure is not available. defaults.attempts includes the first execution. If a retry policy supplies maxAttempts, it must agree with the attempt default. Codec.standardSchema accepts native Zod 4 schemas and the schemas exposed by better-effect-schema; it decodes persisted JSON before the handler runs.

Use Job.define when the direct form is more convenient. This continuation reuses the schema-backed payload codec from the canonical SendEmail descriptor in Defining Jobs, rather than redeclaring the schema here. When the persisted contract changes, create a new local class and codec as shown on that page.

import { Job, JobRegistry } from 'better-effect-mq'

const SendEmailV2 = Job.define('send-email', {
  queue: Emails,
  version: 2,
  payload: SendEmailPayloadCodec
})
const Jobs = JobRegistry.make([SendEmail, SendEmailV2] as const)

The queue/name/version identity is what storage and claims use. Function names are not identity, and the package does not provide a global handler registry. Use the trusted JSON escape hatch only when the value is already trusted and JSON-safe; keep provider-backed schemas as the default for boundary payloads and results.

Produce jobs

A Job is also a typed producer. Its methods are better-effect programs: run them inside a Runtime that provides the Job's bound JobStore token. enqueue encodes the payload, runs the metadata/idempotency callbacks and returns a branded JobId.

The following continuation uses the canonical SendEmail descriptor from Defining Jobs.

import { Effect, Layer, Runtime } from 'better-effect'
import { ClockLive } from 'better-effect/standard-services'
import { Result } from 'better-result'
import { MemoryJobStore } from 'better-effect-mq'

const runtime = await Runtime.make(Layer.merge(MemoryJobStore.layer, ClockLive))

const id = await runtime.run(() =>
  Effect.gen(async function* () {
    const jobId = yield* SendEmail.enqueue({
      messageId: 'message-123',
      recipient: 'ada@example.test'
    })
    return Result.ok(jobId)
  })
)

await runtime.dispose()

Production callers typically use a durable Layer instead of MemoryJobStore.layer. enqueueMany accepts payloads or { payload, options } entries, preserves input order, and processes bounded chunks; a store failure after an earlier chunk can therefore leave a partial batch. Deterministic IDs or idempotency keys make a producer retry safe.

Other Job operations include poll, attempts, awaitResult, execute, cancel, promote and retry. execute means enqueue then poll; it is not an exactly-once RPC. Aborting awaitResult stops waiting but does not cancel a persisted Job.

MemoryJobStore for tests

MemoryJobStore is an isolated in-process reference driver. It is useful for unit tests, examples and disposable processes, but is not durable, shared between processes or a distributed coordination mechanism. Each MemoryJobStore.layer resolution creates fresh store state.

import { Runtime } from 'better-effect'
import { MemoryJobStore } from 'better-effect-mq'

let sequence = 0
const testRuntime = await Runtime.make(
  MemoryJobStore.layerWith({
    idGenerator: () => `test-${sequence++}`
  })
)

// Run producer and worker programs against testRuntime.
await testRuntime.dispose()

For deterministic tests, clock controls the epoch milliseconds used by store requests and idGenerator controls generated Job IDs and lease tokens. Use MemoryJobStore.make(options) when a Layer is not needed. A named token can use MemoryJobStore.layerFor(token, options).

The runner-neutral better-effect-mq/testing entrypoint also exports jobStoreContract. It supplies scenarios for enqueue identity, ordering, claims, leases, settlements, administration, listing and wake semantics. The adapter provides makeRuntime, setup/reset hooks and declared capabilities; register each returned scenario with the test runner of your choice.

For focused application tests, TestJobStore keeps a real isolated MemoryJobStore together with ClockTest, IdGeneratorTest, a Layer, and a recorded observer. Its enqueued, enqueuedPayloads, job, attempts, and counts helpers use public store operations and the definition's exact codec; claim, settle, and release accept explicit lease tokens. Named stores use TestJobStore.makeFor(namedStore, options). This is a harness around the real store contract, not a second worker or a private state mutator.

The following continuation uses the canonical SendEmail descriptor from Defining Jobs and observes it with TestJobStore.

import { ClockTest, IdGeneratorTest } from 'better-effect/standard-services'
import { TestJobStore } from 'better-effect-mq/testing'

const testStore = TestJobStore.make({
  clock: new ClockTest(0),
  ids: IdGeneratorTest.from((index) => `test-${index}`)
})
const observedSendEmail = testStore.observe(SendEmail)
// Provide testStore.layer and testStore.clock to a real Runtime/TestRuntime.
void observedSendEmail

Run and stop a Worker

Workers run handlers over a named Worker.service(tag).layer(factory) in the same Layer graph as the application, so the Runtime owns startup, quiesce and release.

The following continuation uses the canonical SendEmail descriptor from Defining Jobs and wires a handler through a Layer-owned Worker.

import { Effect, Layer, Runtime } from 'better-effect'
import { Result } from 'better-result'
import { Worker } from 'better-effect-mq'

const SendEmailHandler = Worker.handle(SendEmail, (payload) =>
  Effect.fn(async function* () {
    // Put the idempotent external send here. Use the Job ID or messageId as
    // the downstream idempotency key.
    return Result.ok({ accepted: payload.recipient.length > 0 })
  })
)

const AppWorker = Worker.service('@app/EmailWorker')
const AppWorkerLive = AppWorker.layer(() => ({
  handlers: [SendEmailHandler],
  concurrency: 8
}))
const application = Layer.complete(Layer.merge(AppLive, AppWorkerLive))
const runtime = await Runtime.make(application)
await runtime.warmup()

const workerResult = await runtime.run(() =>
  Effect.gen(async function* () {
    return Result.ok(yield* AppWorker)
  })
)

Worker.stop() remains idempotent for an acquired handle when an application needs to quiesce one Worker early. Normal application shutdown should dispose the owning Runtime, which quiesces, drains and releases its Layer-owned Workers.

Retry, backoff and timeout

Retry.never(), Retry.fixed, Retry.linear and Retry.exponential create validated immutable policies. Backoff is persisted as data; Retry.custom is evaluated synchronously in the Worker and is not delegated to a store.

Typed failures are retried only when the Job's retryable predicate returns true. Mark an object failure with Job.unrecoverable(failure) when it must not retry. Defects retry by default; set retryDefects: false to disable that. Decode and encode failures are terminal. A timeout aborts the attempt cooperatively and is persisted as a timeout failure; a Promise that ignores its signal cannot be forcibly killed.

const WorkerLive = AppWorker.layer(() => ({
  handlers: [SendEmailHandler],
  retryDefects: false,
  leaseDurationMs: 30_000,
  heartbeatIntervalMs: 10_000,
  stalledIntervalMs: 30_000,
  maxStalledCount: 1,
  shutdown: {
    gracePeriodMs: 5_000,
    abortAfterGracePeriod: true
  }
}))

The lease must remain valid while the attempt runs. Workers heartbeat active leases, recover expired leases with recoverStalled, and fence late results with the lease token. Shutdown and lease loss are cooperative; external work still needs idempotency and its own fencing strategy.

Errors and settlement semantics

A handler's nominal Result.ok becomes a completed settlement. A nominal Result.err becomes a typed failed settlement, and a thrown or rejected handler becomes a defect settlement. Results and failure values are encoded by the Job codecs before persistence. onJobFailure is best effort and runs only after an applied terminal or retry settlement; onError reports supervisor errors without stopping other claim loops.

The store applies settlement as one fenced state transition and attempt-ledger write. A lost response can be retried with the same Job ID and lease token without duplicating the attempt ledger; a stale or expired token cannot settle a newer claim. Release, retry and stalled recovery are explicit transitions that preserve the delivery history needed for operations.

The storage adapter boundary

JobStore is the only storage seam. It is a yieldable Service whose JobStore.Contract contains atomic operations such as enqueue, claim, settle, release, heartbeat, recoverStalled, awaitWake, inspection and administration. Every operation returns a completed better-effect Result or a PromiseLike of one, so an adapter can perform asynchronous I/O without changing the consumer API.

An adapter supplies its structural contract through a Layer:

import { Layer } from 'better-effect'
import { JobStore } from 'better-effect-mq'

const DurableStore = JobStore.named('durable')

declare const adapter: JobStore.Contract
const DurableStoreLayer = Layer.succeed(DurableStore, DurableStore.of(adapter))

Bind a Job to that named store with store: DurableStore in its definition or with bindJob(job, DurableStore). The core only sees the Service contract and stable JSON-safe records. The adapter owns transactions, persistence, indexes, wake notifications and backend-specific translation while preserving claim ordering, lease fencing, time validation and atomic settlement semantics. Capability flags are performance hints, not permission to weaken correctness.

Do not put driver types, SQL/Redis keys or connection objects in Job records, codecs or core APIs. The same boundary lets MemoryJobStore, a database adapter and a test double implement the same store contract without changing producer or worker code.