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.

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>;

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"]
>;