DurableQueue

Durable workflow queues delegate work to persisted background workers and resume the waiting workflow with the worker result.

A workflow calls process to encode a payload, offer it to a named PersistedQueue, attach a DurableDeferred token, and suspend. A worker created with makeWorker or worker takes the item, runs the handler, and records the handler's Exit through that token so the original workflow can continue with the typed success or error.

Delivery is at-least-once: a crash between handler success and the acknowledgement redelivers the item, so handlers must be idempotent. When an item exhausts its persisted queue attempts it is dead-lettered: the DurableDeferred never resolves and the workflow stays parked until the item is requeued out of band, while the id-based de-duplication prevents replays from resurrecting the failed item. Deployments should run PersistedQueue.layerCleanup in one instance to prune old completed items.

7 exports Added in v4.0.0 Source

Constructors

make

Added in v4.0.0 Source

Creates a DurableQueue that waits for persisted items to finish processing using a DurableDeferred.

Signature

declare function make<Payload extends Top | Fields, Success extends Top = Void, Error extends Top = Never>(options: {
  readonly error?: Error;
  readonly idempotencyKey: (payload: Payload extends Fields ? View<Payload, "Type", TypeOptionalKeys<Payload>, TypeMutableKeys<Payload>> : Payload["Type"]) => string;
  readonly name: string;
  readonly payload: Payload;
  readonly success?: Success;
}): DurableQueue<Payload extends Fields ? Struct<Payload> : Payload, Success, Error>

Example

(Defining a durable queue with workers)

import { Effect, Layer, Schema } from "effect"
import { DurableQueue, Workflow } from "effect/unstable/workflow"

// Define a DurableQueue that can be used to derive workers and offer items for
// processing.
const ApiQueue = DurableQueue.make({
  name: "ApiQueue",
  payload: {
    id: Schema.String
  },
  success: Schema.Void,
  error: Schema.Never,
  idempotencyKey(payload) {
    return payload.id
  }
})

const MyWorkflow = Workflow.make("MyWorkflow", {
  payload: {
    id: Schema.String
  },
  idempotencyKey: ({ id }) => id
})

const MyWorkflowLayer = MyWorkflow.toLayer(
  Effect.fnUntraced(function*() {
    // The workflow suspends until a worker completes this queue item.
    yield* DurableQueue.process(ApiQueue, { id: "api-call-1" })
    return "Workflow succeeded!"
  })
)

const processed: Array<string> = []
const processApiCall = ({ id }: { readonly id: string }) => Effect.sync(() => processed.push(id))

// Construct the worker layer without starting background workers in this example.
const ApiWorker = DurableQueue.worker(ApiQueue, processApiCall, {
  concurrency: 5
})

const program = Effect.gen(function*() {
  // Exercise the finite handler directly instead of running a queue worker.
  yield* processApiCall({ id: "api-call-1" })
  return [Layer.isLayer(MyWorkflowLayer), Layer.isLayer(ApiWorker), processed] as const
})

await Effect.runPromise(program) // => [true, true, ["api-call-1"]]

Models

DurableQueue interface

Added in v4.0.0 Source

Durable workflow queue definition containing a payload schema, idempotency key, and deferred used to await worker results.

Signature

interface DurableQueue<Payload extends Schema.Top, Success extends Schema.Top = Schema.Void, Error extends Schema.Top = Schema.Never> {
  readonly "~effect/workflow/DurableQueue": "~effect/workflow/DurableQueue";
  readonly deferred: DurableDeferred<Success, Error>;
  readonly idempotencyKey: (payload: Payload["Type"]) => string;
  readonly name: string;
  readonly payloadSchema: Payload;
}

Running

process

Added in v4.0.0 Source

Adds an item to the queue and wait for a worker to process it.

Signature

declare const process: <Payload extends Schema.Top, Success extends Schema.Top, Error extends Schema.Top>(self: DurableQueue<Payload, Success, Error>, payload: Payload["~type.make.in"], options?: {
  readonly retrySchedule?: Schedule.Schedule<any, PersistedQueue.PersistedQueueError>;
}) => Effect.Effect<Success["Type"], Error["Type"], WorkflowEngine | WorkflowInstance | PersistedQueue.PersistedQueueFactory | Payload["EncodingServices"] | Payload["DecodingServices"] | Success["DecodingServices"] | Error["DecodingServices"]>

Type IDs

TypeId

Added in v4.0.0 Source

Runtime identifier attached to DurableQueue values.

Signature

declare const TypeId: "~effect/workflow/DurableQueue"

TypeId type

Added in v4.0.0 Source

Type-level identifier used to recognize DurableQueue values.

Signature

type TypeId = "~effect/workflow/DurableQueue"

Workers

makeWorker

Added in v4.0.0 Source

Create a worker effect that processes items from the durable queue.

Signature

declare const makeWorker: <Payload extends Schema.Top, Success extends Schema.Top, Error extends Schema.Top, R>(self: DurableQueue<Payload, Success, Error>, f: (payload: Payload["Type"]) => Effect.Effect<Success["Type"], Error["Type"], R>, options?: {
  readonly concurrency?: number;
}) => Effect.Effect<never, never, WorkflowEngine | PersistedQueue.PersistedQueueFactory | R | Payload["EncodingServices"] | Payload["DecodingServices"] | Success["EncodingServices"] | Error["EncodingServices"]>

worker

Added in v4.0.0 Source

Create a layer that runs workers for the durable queue.

Signature

declare const worker: <Payload extends Schema.Top, Success extends Schema.Top, Error extends Schema.Top, R>(self: DurableQueue<Payload, Success, Error>, f: (payload: Payload["Type"]) => Effect.Effect<Success["Type"], Error["Type"], R>, options?: {
  readonly concurrency?: number;
}) => Layer.Layer<never, never, WorkflowEngine | PersistedQueue.PersistedQueueFactory | R | Payload["EncodingServices"] | Payload["DecodingServices"] | Success["EncodingServices"] | Error["EncodingServices"]>