Skip to content
Effect 4.0 is here. One ecosystem. Zero dependencies.

PersistedQueue unstable

Stores schema-encoded queue work in persistent storage.

A PersistedQueue<A> keeps JSON-encoded values in a named queue and lets workers take one value at a time inside a scoped processing window. It is useful for durable handoffs, background jobs, outbox-style integrations, and work that should retry across fibers, process restarts, or multiple workers. This module includes a queue factory, store service, id-based de-duplication, retry handling, and in-memory, Redis, and SQL-backed store layers.

Delivery is at-least-once: a crash between handler success and the acknowledgement redelivers the element, so handlers must be idempotent.

17 exports Since v4.0.0 Source

Accessors

make

unstable
Since v4.0.0

Accesses PersistedQueueFactory to create a named persisted queue for a schema.

Details

maxAttempts defaults to 10. retrySchedule controls the delay before a failed element becomes visible again, and defaults to an exponential delay starting at 1 second and capped at 5 minutes.

The schedule's state is the element's persisted attempt count. On each failure the schedule is replayed up to the current attempt, so delays keep progressing even when consecutive retries run in different processes. The schedule input is the attempt number.

Replay simulates elapsed time from the sum of the computed delays. Attempt-driven schedules are therefore exact, while wall-clock-anchored schedules observe this idealized time. In particular, Schedule.upTo({ duration }) caps the summed delays rather than real time since the original failure.

Signature

declare function make<S extends Constraint>(options: {
  readonly maxAttempts?: number;
  readonly name: string;
  readonly retrySchedule?: Schedule<any, number, never, never>;
  readonly schema: S;
}): Effect<PersistedQueue<S["Type"], S["EncodingServices"] | S["DecodingServices"]>, never, PersistedQueueFactory>

Constructors

makeFactory

unstable
Since v4.0.0

Creates a PersistedQueueFactory from the current PersistedQueueStore.

Details

Values are encoded and decoded with the supplied schema, automatically assigned an id when needed, and acknowledged or retried according to the take handler's exit.

Signature

declare const makeFactory: Effect<{
  readonly make: <S extends Constraint>(options: {
    readonly maxAttempts?: number;
    readonly name: string;
    readonly retrySchedule?: Schedule<any, number, never, never>;
    readonly schema: S;
  }) => Effect<PersistedQueue<S["Type"], S["EncodingServices"] | S["DecodingServices"]>>;
}, never, PersistedQueueStore>

makeStoreRedis

unstable
Since v4.0.0

Creates a Redis-backed PersistedQueueStore.

Details

The store uses Redis lists, hashes, and sorted sets with worker locks, periodically refreshes locks while items are being processed, delays retried items with the queue's retry schedule, and moves exhausted items to a failed queue. Delayed offers share the retry sorted set: they are scored with the offering process's clock and released by the polling worker's clock, so clock skew between hosts shifts their delivery.

Signature

declare const makeStoreRedis: (...args: [options?: {
  readonly lockExpiration?: Input;
  readonly lockRefreshInterval?: Input;
  readonly pollInterval?: Input;
  readonly prefix?: string;
}]) => Effect<{
  readonly cleanup: (options: {
    readonly failedTimeToLive: Duration | undefined;
    readonly timeToLive: Duration;
  }) => Effect<void, PersistedQueueError>;
  readonly offer: (options: {
    readonly delay?: Duration;
    readonly element: unknown;
    readonly id: string;
    readonly isCustomId: boolean;
    readonly name: string;
  }) => Effect<void, PersistedQueueError>;
  readonly take: (options: {
    readonly maxAttempts: number;
    readonly name: string;
    readonly retryDelay: (attempts: number) => Effect<Duration>;
  }) => Effect<{
    readonly attempts: number;
    readonly element: unknown;
    readonly id: string;
  }, PersistedQueueError, Scope>;
}, never, Scope | Redis>

makeStoreSql

unstable
Since v4.0.0

Creates a SQL-backed PersistedQueueStore.

Details

The store creates the queue table and indexes, acquires rows with per-worker locks, refreshes active locks while scoped takes are running, and retries or completes rows according to the processing exit.

Offer and retry delays are measured with the database clock and rounded up to whole seconds.

Signature

declare const makeStoreSql: (options?: {
  readonly lockExpiration?: Duration.Input;
  readonly lockRefreshInterval?: Duration.Input;
  readonly pollInterval?: Duration.Input;
  readonly tableName?: string;
}) => Effect.Effect<PersistedQueueStore["Service"], SqlError, SqlClient.SqlClient | Scope.Scope>

Errors

PersistedQueueError

unstable
Since v4.0.0

Error raised by persisted queue store operations.

Signature

declare class PersistedQueueError extends {
  readonly _tag: "PersistedQueueError";
  readonly cause?: unknown;
  readonly message: string;
} & YieldableError<this> {
  constructor(...args: [props: {
    readonly _tag?: "PersistedQueueError";
    readonly cause?: unknown;
    readonly message: string;
  }, options?: MakeOptions]);
  readonly "~effect/persistence/PersistedQueue/PersistedQueueError": "~effect/persistence/PersistedQueue/PersistedQueueError";
}

Layers

layer

unstable
Since v4.0.0

Provides PersistedQueueFactory using the current PersistedQueueStore.

Signature

declare const layer: Layer.Layer<PersistedQueueFactory, never, PersistedQueueStore>

layerCleanup

unstable
Since v4.0.0

Runs PersistedQueueStore.cleanup on a schedule.

Details

Completed elements are retained for timeToLive (default 30 days) so offer de-duplication keeps working across replays, then removed. Failed elements are the dead-letter record and are kept forever unless failedTimeToLive is set.

Run this layer in one instance of a deployment rather than on every worker; racing instances are harmless since deletes are idempotent, but the work is redundant.

Signature

declare function layerCleanup(options?: {
  readonly failedTimeToLive?: Input;
  readonly interval?: Input;
  readonly timeToLive?: Input;
}): Layer<never, never, PersistedQueueStore>

layerStoreMemory

unstable
Since v4.0.0

Provides an in-memory PersistedQueueStore.

Details

The store is process-local and volatile; failed takes are requeued with the queue's retry schedule until the configured maximum attempts is reached, after which the element is marked as failed.

Delayed elements are scanned on every take loop, so the store is meant for development and tests rather than large delayed backlogs.

Signature

declare const layerStoreMemory: Layer.Layer<PersistedQueueStore>

layerStoreRedis

unstable
Since v4.0.0

Provides a Redis-backed PersistedQueueStore using makeStoreRedis.

Signature

declare const layerStoreRedis: (options?: {
  readonly lockExpiration?: Duration.Input;
  readonly lockRefreshInterval?: Duration.Input;
  readonly pollInterval?: Duration.Input;
  readonly prefix?: string;
}) => Layer.Layer<PersistedQueueStore, never, Redis.Redis>

layerStoreSql

unstable
Since v4.0.0

Provides a SQL-backed PersistedQueueStore using makeStoreSql.

Signature

declare const layerStoreSql: (options?: {
  readonly lockExpiration?: Duration.Input;
  readonly lockRefreshInterval?: Duration.Input;
  readonly pollInterval?: Duration.Input;
  readonly tableName?: string;
}) => Layer.Layer<PersistedQueueStore, SqlError, SqlClient.SqlClient>

Models

PersistedQueue

interface unstable
Since v4.0.0

Persistent queue of schema-encoded values.

Details

offer enqueues values by id, and take processes one value at a time, marking it complete on success or retrying it with the queue's retry schedule until the maximum attempts is reached, after which it is marked as failed.

Delivery is at-least-once: a crash between handler success and the acknowledgement redelivers the element, so handlers must be idempotent.

Signature

interface PersistedQueue<in out A, out R = never> {
  readonly "~effect/persistence/PersistedQueue": "~effect/persistence/PersistedQueue";
  readonly offer: (value: A, options?: {
    readonly delay?: Input;
    readonly id?: string;
  }) => Effect<string, SchemaError | PersistedQueueError, R>;
  readonly take: <XA, XE, XR>(f: (value: A, metadata: {
    readonly attempts: number;
    readonly id: string;
  }) => Effect<XA, XE, XR>) => Effect<XA, PersistedQueueError | XE, R | XR>;
}

Services

PersistedQueueFactory

unstable
Since v4.0.0

Service for constructing named PersistedQueue instances from schemas.

Signature

declare class PersistedQueueFactory extends Shape<"effect/persistence/PersistedQueue/PersistedQueueFactory", {
  readonly make: <S extends Constraint>(options: {
    readonly maxAttempts?: number;
    readonly name: string;
    readonly retrySchedule?: Schedule<any, number, never, never>;
    readonly schema: S;
  }) => Effect<PersistedQueue<S["Type"], S["EncodingServices"] | S["DecodingServices"]>>;
}, this> {
  constructor(_: never);
}

PersistedQueueStore

unstable
Since v4.0.0

Defines the low-level backing store service used by PersistedQueue.

When to use

Use to provide the persistence backend that stores queued elements, scoped takes, retry attempts, and acknowledgements.

Details

The store persists offered elements and returns taken elements in a scope so the finalizer can complete or retry them based on the processing exit.

An offered delay is always finite and positive, and the element must not be taken before it elapses. It is undefined for immediate elements.

Claiming an element counts an attempt, so the attempts returned by take is 1-based. When the take scope closes with a success the element is marked completed; a failure retries it according to retryDelay or marks it failed once maxAttempts is exhausted; an interruption releases it without counting the attempt.

Signature

declare class PersistedQueueStore extends Shape<"effect/persistence/PersistedQueue/PersistedQueueStore", {
  readonly cleanup: (options: {
    readonly failedTimeToLive: Duration | undefined;
    readonly timeToLive: Duration;
  }) => Effect<void, PersistedQueueError>;
  readonly offer: (options: {
    readonly delay?: Duration;
    readonly element: unknown;
    readonly id: string;
    readonly isCustomId: boolean;
    readonly name: string;
  }) => Effect<void, PersistedQueueError>;
  readonly take: (options: {
    readonly maxAttempts: number;
    readonly name: string;
    readonly retryDelay: (attempts: number) => Effect<Duration>;
  }) => Effect<{
    readonly attempts: number;
    readonly element: unknown;
    readonly id: string;
  }, PersistedQueueError, Scope>;
}, this> {
  constructor(_: never);
}

Type IDs

ErrorTypeId

unstable
Since v4.0.0

Runtime type identifier for PersistedQueueError.

Signature

declare const ErrorTypeId: ErrorTypeId

ErrorTypeId

type unstable
Since v4.0.0

Type-level identifier used to brand PersistedQueueError values.

Signature

type ErrorTypeId = "~effect/persistence/PersistedQueue/PersistedQueueError"

TypeId

unstable
Since v4.0.0

Runtime type identifier for PersistedQueue values.

Signature

declare const TypeId: TypeId

TypeId

type unstable
Since v4.0.0

Type-level identifier used to brand PersistedQueue values.

Signature

type TypeId = "~effect/persistence/PersistedQueue"