PersistedQueue
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.
Accessors
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
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
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.
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 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
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.
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
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
Provides PersistedQueueFactory using the current PersistedQueueStore.
Signature
declare const layer: Layer.Layer<PersistedQueueFactory, never, PersistedQueueStore>layerCleanup
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
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.
Signature
declare const layerStoreMemory: Layer.Layer<PersistedQueueStore>layerStoreRedis
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
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
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 id: string | undefined; }) => 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
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
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.
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 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
Runtime type identifier for PersistedQueueError.
Signature
declare const ErrorTypeId: ErrorTypeIdErrorTypeId type
Type-level identifier used to brand PersistedQueueError values.
Signature
type ErrorTypeId = "~effect/persistence/PersistedQueue/PersistedQueueError"Runtime type identifier for PersistedQueue values.
Signature
declare const TypeId: TypeIdType-level identifier used to brand PersistedQueue values.
Signature
type TypeId = "~effect/persistence/PersistedQueue"