PersistedQueue unstable

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.

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>

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.

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>

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. A handler whose lock lapsed can still settle or renew its claim until a reset recovers the element.

Gotchas

The lock format is incompatible with older workers on the same prefix: their settlements are not fenced and can corrupt newer claims. Stop every old worker and let its locks expire before starting upgraded workers. Keep the existing Redis data rather than flushing it.

Every failure of an id re-offered with deduplicate: "untilExecuted" stays in the failed list until layerCleanup removes it, which requires failedTimeToLive. Ids that failed on an earlier effect version stay reserved until then, even for "untilExecuted" offers.

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 deduplicate?: "untilExecuted" | "untilRemoved";
    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>

Details

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

Settlements and refreshes only apply to the acquisition a handler received, so a handler whose lock lapsed can still settle or renew its claim until another poll re-claims the row.

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

An id re-offered with deduplicate: "untilExecuted" keeps only its latest failure.

Gotchas

An "untilExecuted" offer inside a transaction checks the id against that transaction's snapshot. Under an isolation level with one snapshot per transaction, such as MySQL's default repeatable read, an element that completed or failed after the first read still looks in flight, so the offer is ignored.

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>

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.

declare function layerCleanup(options?: {
  readonly failedTimeToLive?: Input;
  readonly interval?: Input;
  readonly timeToLive?: Input;
}): Layer<never, never, 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.

declare const layerStoreMemory: Layer.Layer<PersistedQueueStore>

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.

interface PersistedQueue<in out A, out R = never> {
  readonly "~effect/persistence/PersistedQueue": "~effect/persistence/PersistedQueue";
  readonly offer: (value: A, options?: {
    readonly deduplicate?: "untilExecuted" | "untilRemoved";
    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>;
}

When to use

Use when a custom PersistedQueueStore take finalizer decides whether a failed element is marked failed at once or retried.

Details

A payload that fails to decode with the queue's schema can never be processed, so PersistedQueue fails the take scope with an internal dead-letter signal and takes the next element. The store should mark the element failed whatever attempts remain, which also releases its id for a deduplicate: "untilExecuted" offer.

See

• PersistedQueueStore for the store contract that calls for this check

declare function isDeadLetterCause(cause: Cause<unknown>): boolean

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.

An offer whose id is already stored is ignored, unless deduplicate is "untilExecuted" and that element has completed or failed its final attempt. The offer is then stored as a new element after earlier offers, atomically with respect to concurrent offers of the id. undefined means "untilRemoved".

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. A failure for which isDeadLetterCause holds marks the element failed immediately, without a retry.

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 deduplicate?: "untilExecuted" | "untilRemoved";
    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);
}