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.
Accessors
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 name: string;
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 and hashes with worker locks, periodically refreshes locks while items are being processed, 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 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 }) => 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/experimental/PersistedQueue/PersistedQueueError": "~@effect/experimental/PersistedQueue/PersistedQueueError";
}Layers
Provides PersistedQueueFactory using the current PersistedQueueStore.
Signature
declare const layer: Layer.Layer<PersistedQueueFactory, never, PersistedQueueStore>;layerStoreMemory
Provides an in-memory PersistedQueueStore.
Details
The store is process-local and volatile; failed takes are requeued until the configured maximum attempts is reached.
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 until the maximum attempts is reached.
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>,
options?: {
readonly maxAttempts?: number;
},
) => Effect<XA, SchemaError | 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 name: string;
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.
Signature
declare class PersistedQueueStore extends Shape<
"effect/persistence/PersistedQueue/PersistedQueueStore",
{
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 }) => 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: ErrorTypeId;ErrorTypeId type
Type-level identifier used to brand PersistedQueueError values.
Signature
type ErrorTypeId = "~@effect/experimental/PersistedQueue/PersistedQueueError";Runtime type identifier for PersistedQueue values.
Signature
declare const TypeId: TypeId;Type-level identifier used to brand PersistedQueue values.
Signature
type TypeId = "~effect/persistence/PersistedQueue";
Accesses
PersistedQueueFactoryto create a named persisted queue for a schema.