diff --git a/.changeset/olive-queues-persist.md b/.changeset/olive-queues-persist.md new file mode 100644 index 00000000000..ee9e644cdea --- /dev/null +++ b/.changeset/olive-queues-persist.md @@ -0,0 +1,5 @@ +--- +"effect": patch +--- + +Improve `PersistedQueue` reliability across SQL, Redis, and memory stores. Retry policy now lives on `make()`, attempts count on claim, retries follow a `Schedule`, and exhausted or undecodable elements are dead-lettered. Add retention cleanup, durable acknowledgement retries, storage schema fixes, local poll wakeups, and fixes for the memory take race and Redis dedup growth. diff --git a/packages/effect/src/unstable/persistence/PersistedQueue.ts b/packages/effect/src/unstable/persistence/PersistedQueue.ts index 8d2fc703b8c..c46c4749e36 100644 --- a/packages/effect/src/unstable/persistence/PersistedQueue.ts +++ b/packages/effect/src/unstable/persistence/PersistedQueue.ts @@ -8,20 +8,23 @@ * 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. + * * @since 4.0.0 */ -import type * as Arr from "../../Array.ts" import * as Cause from "../../Cause.ts" +import * as Clock from "../../Clock.ts" import * as Context from "../../Context.ts" -import * as Data from "../../Data.ts" import * as Duration from "../../Duration.ts" import * as Effect from "../../Effect.ts" -import * as Exit from "../../Exit.ts" import { flow } from "../../Function.ts" -import * as Iterable from "../../Iterable.ts" +import { sqlCleanupBatchSize } from "../../internal/persistence.ts" import * as Latch from "../../Latch.ts" import * as Layer from "../../Layer.ts" import * as MutableRef from "../../MutableRef.ts" +import * as Predicate from "../../Predicate.ts" +import * as Pull from "../../Pull.ts" import * as Queue from "../../Queue.ts" import * as RcMap from "../../RcMap.ts" import * as Schedule from "../../Schedule.ts" @@ -54,8 +57,12 @@ export type TypeId = "~effect/persistence/PersistedQueue" * **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. + * 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. * * @category models * @since 4.0.0 @@ -69,7 +76,8 @@ export interface PersistedQueue { * **Details** * * If an element with the same id already exists in the queue, it will not be - * added again. + * added again. De-duplication survives completion until the id is removed by + * `layerCleanup`. */ readonly offer: (value: A, options?: { readonly id: string | undefined @@ -82,18 +90,23 @@ export interface PersistedQueue { * **Details** * * If the returned effect succeeds, the element is marked as processed; - * otherwise it will be retried according to the provided options. By default, - * max attempts is set to 10. + * otherwise it will be retried with the queue's retry schedule until the + * maximum attempts is reached, after which it is marked as failed. + * + * An attempt is counted when the element is claimed, so `attempts` in the + * handler metadata is 1-based ("this is attempt 3"). A handler crash that + * takes down the process still consumes an attempt. + * + * Elements that fail to decode with the queue's schema are marked as failed + * immediately and the next element is taken instead, so schema decode errors + * never surface from `take`. */ readonly take: ( f: (value: A, metadata: { readonly id: string readonly attempts: number - }) => Effect.Effect, - options?: { - readonly maxAttempts?: number | undefined - } - ) => Effect.Effect + }) => Effect.Effect + ) => Effect.Effect } /** @@ -108,6 +121,8 @@ export class PersistedQueueFactory extends Context.Service< readonly make: (options: { readonly name: string readonly schema: S + readonly maxAttempts?: number | undefined + readonly retrySchedule?: Schedule.Schedule | undefined }) => Effect.Effect> } >()("effect/persistence/PersistedQueue/PersistedQueueFactory") {} @@ -116,18 +131,65 @@ export class PersistedQueueFactory extends Context.Service< * 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. + * * @category accessors * @since 4.0.0 */ export const make = (options: { readonly name: string readonly schema: S + readonly maxAttempts?: number | undefined + readonly retrySchedule?: Schedule.Schedule | undefined }): Effect.Effect< PersistedQueue, never, PersistedQueueFactory > => PersistedQueueFactory.use((factory) => factory.make(options)) +const defaultRetrySchedule = Schedule.min([ + Schedule.exponential("1 second"), + Schedule.spaced("5 minutes") +]) + +// The persisted attempt count is the schedule state: consecutive retries of an +// element can run in different processes, so a fresh schedule step is replayed +// up to the given attempt on every call. Delays therefore depend only on the +// attempt count, with elapsed time simulated from the summed delays. +const retryDelay = Effect.fnUntraced(function*( + schedule: Schedule.Schedule, + attempts: number +): Effect.fn.Return { + const step = yield* Schedule.toStep(schedule) + let now = 0 + let delay = Duration.zero + for (let i = 0; i < attempts; i++) { + const result = yield* Pull.catchDone(step(now, i + 1), () => Effect.undefined) + if (result === undefined) { + // the schedule is done, keep using its final delay + break + } + delay = result[1] + now += Duration.toMillis(delay) + } + return delay +}) + /** * Creates a `PersistedQueueFactory` from the current `PersistedQueueStore`. * @@ -147,10 +209,18 @@ export const makeFactory = Effect.gen(function*() { make(options: { readonly name: string readonly schema: S + readonly maxAttempts?: number | undefined + readonly retrySchedule?: Schedule.Schedule | undefined }) { const jsonSchema = Schema.toCodecJson(options.schema) const encodeUnknown = Schema.encodeUnknownEffect(jsonSchema) const decodeUnknown = Schema.decodeUnknownEffect(jsonSchema) + const retrySchedule = options.retrySchedule ?? defaultRetrySchedule + const takeOptions = { + name: options.name, + maxAttempts: options.maxAttempts ?? 10, + retryDelay: (attempts: number) => retryDelay(retrySchedule, attempts) + } return Effect.succeed>({ [TypeId]: TypeId, @@ -170,15 +240,38 @@ export const makeFactory = Effect.gen(function*() { ) } ), - take: (f, opts) => - Effect.scopedWith(Effect.fnUntraced(function*(scope) { - const item = yield* store.take({ - name: options.name, - maxAttempts: opts?.maxAttempts ?? 10 - }).pipe(Scope.provide(scope)) - const decoded = yield* decodeUnknown(item.element) - return yield* f(decoded, { id: item.id, attempts: item.attempts }) - })) + take: ( + f: (value: S["Type"], metadata: { + readonly id: string + readonly attempts: number + }) => Effect.Effect + ) => { + const loop: Effect.Effect = Effect.scopedWith((scope) => + store.take(takeOptions).pipe( + Scope.provide(scope), + Effect.flatMap((item) => + decodeUnknown(item.element).pipe( + Effect.catchCause((cause): Effect.Effect => + Cause.hasInterruptsOnly(cause) + ? Effect.failCause(cause as Cause.Cause) + : Effect.fail(new DeadLetter(cause)) + ), + Effect.flatMap((value) => f(value as S["Type"], { id: item.id, attempts: item.attempts })) + ) + ) + ) + ).pipe( + Effect.catchIf( + isDeadLetter, + (): Effect.Effect => loop + ) + ) + return loop as Effect.Effect< + XA, + XE | PersistedQueueError, + XR | S["EncodingServices"] | S["DecodingServices"] + > + } }) } }) @@ -196,6 +289,48 @@ export const layer: Layer.Layer< PersistedQueueStore > = Layer.effect(PersistedQueueFactory, makeFactory) +/** + * 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. + * + * @category layers + * @since 4.0.0 + */ +export const layerCleanup = (options?: { + readonly interval?: Duration.Input | undefined + readonly timeToLive?: Duration.Input | undefined + readonly failedTimeToLive?: Duration.Input | undefined +}): Layer.Layer => + Layer.effectDiscard(Effect.gen(function*() { + const store = yield* PersistedQueueStore + const cleanupOptions = { + timeToLive: Duration.fromInputUnsafe(options?.timeToLive ?? Duration.days(30)), + failedTimeToLive: options?.failedTimeToLive === undefined + ? undefined + : Duration.fromInputUnsafe(options.failedTimeToLive) + } + yield* store.cleanup(cleanupOptions).pipe( + Effect.catchCause((cause) => Effect.logWarning("Failed to clean up persisted queue", cause)), + Effect.repeat(Schedule.spaced(options?.interval ?? Duration.hours(1))), + Effect.interruptible, + Effect.forkScoped, + Effect.annotateLogs({ + module: "effect/persistence/PersistedQueue", + fiber: "cleanup" + }) + ) + })) + /** * Runtime type identifier for `PersistedQueueError`. * @@ -233,6 +368,64 @@ export class PersistedQueueError extends Schema.Error( readonly [ErrorTypeId]: ErrorTypeId = ErrorTypeId } +// Local per-queue state shared between offer, take, and the mailbox lookup in +// the Redis and SQL stores: take records maxAttempts here before the mailbox +// is created, and offer opens the nudge latch so local pollers skip the poll +// interval. +type QueueState = { + maxAttempts: number + readonly nudge: Latch.Latch +} + +const makeQueueStates = (): { + readonly peek: (name: string) => QueueState | undefined + readonly get: (name: string) => QueueState +} => { + const states = new Map() + return { + peek: (name) => states.get(name), + get: (name) => { + let state = states.get(name) + if (state === undefined) { + // the placeholder is overwritten by take before any mailbox reads it, + // and is kept int32-safe for SQL parameters + state = { maxAttempts: 2147483647, nudge: Latch.makeUnsafe(false) } + states.set(name, state) + } + return state + } + } +} + +const makeAckRetrySchedule = (lockExpiration: Duration.Input) => + Schedule.min([ + Schedule.exponential(200, 1.5), + Schedule.spaced(5000) + ]).pipe(Schedule.upTo({ duration: lockExpiration })) + +// Internal signal used by the factory to tell a store that a taken element +// cannot ever be processed (its stored payload fails to decode) and must be +// dead-lettered instead of retried. +class DeadLetter { + readonly _tag = "~effect/persistence/PersistedQueue/DeadLetter" + readonly cause: Cause.Cause + constructor(cause: Cause.Cause) { + this.cause = cause + } +} + +const isDeadLetter = (u: unknown): u is DeadLetter => + Predicate.isTagged(u, "~effect/persistence/PersistedQueue/DeadLetter") + +const deadLetterFromCause = (cause: Cause.Cause): DeadLetter | undefined => { + for (const reason of cause.reasons) { + if (Cause.isFailReason(reason) && isDeadLetter(reason.error)) { + return reason.error + } + } + return undefined +} + /** * Defines the low-level backing store service used by `PersistedQueue`. * @@ -246,6 +439,12 @@ export class PersistedQueueError extends Schema.Error( * 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. + * * @category services * @since 4.0.0 */ @@ -264,6 +463,7 @@ export class PersistedQueueStore extends Context.Service< readonly take: (options: { readonly name: string readonly maxAttempts: number + readonly retryDelay: (attempts: number) => Effect.Effect }) => Effect.Effect< { readonly id: string @@ -273,6 +473,16 @@ export class PersistedQueueStore extends Context.Service< PersistedQueueError, Scope.Scope > + + /** + * Removes completed elements older than `timeToLive`, together with their + * de-duplication records. Failed elements are removed only when + * `failedTimeToLive` is provided. + */ + readonly cleanup: (options: { + readonly timeToLive: Duration.Duration + readonly failedTimeToLive: Duration.Duration | undefined + }) => Effect.Effect } >()("effect/persistence/PersistedQueue/PersistedQueueStore") {} @@ -281,83 +491,156 @@ export class PersistedQueueStore extends Context.Service< * * **Details** * - * The store is process-local and volatile; failed takes are requeued until the - * configured maximum attempts is reached. + * 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. * * @category layers * @since 4.0.0 */ export const layerStoreMemory: Layer.Layer< PersistedQueueStore -> = Layer.sync(PersistedQueueStore, () => { - type Entry = { - readonly id: string - attempts: number - readonly element: unknown - } - const queues = new Map - items: Set - }>() - const getOrCreateQueue = (name: string) => { - let queue = queues.get(name) - if (!queue) { - queue = { - latch: Latch.makeUnsafe(false), - ids: new Set(), - items: new Set() +> = Layer.effect( + PersistedQueueStore, + Effect.gen(function*() { + const clock = yield* Clock.Clock + type Entry = { + readonly id: string + readonly element: unknown + attempts: number + state: "pending" | "processing" | "completed" | "failed" + visibleAt: number + stateChangedAt: number + } + const queues = new Map + pending: Set + }>() + const getOrCreateQueue = (name: string) => { + let queue = queues.get(name) + if (!queue) { + queue = { + latch: Latch.makeUnsafe(false), + entries: new Map(), + pending: new Set() + } + queues.set(name, queue) } - queues.set(name, queue) + return queue } - return queue - } - return PersistedQueueStore.of({ - offer: (options) => - Effect.sync(() => { + return PersistedQueueStore.of({ + offer: (options) => + Effect.sync(() => { + const now = clock.currentTimeMillisUnsafe() + const queue = getOrCreateQueue(options.name) + if (queue.entries.has(options.id)) return + const entry: Entry = { + id: options.id, + element: options.element, + attempts: 0, + state: "pending", + visibleAt: now, + stateChangedAt: now + } + queue.entries.set(options.id, entry) + queue.pending.add(entry) + queue.latch.openUnsafe() + }), + take: Effect.fnUntraced(function*(options) { const queue = getOrCreateQueue(options.name) - if (queue.ids.has(options.id)) return - queue.ids.add(options.id) - queue.items.add({ id: options.id, attempts: 0, element: options.element }) - queue.latch.openUnsafe() - }), - take: Effect.fnUntraced(function*(options) { - const queue = getOrCreateQueue(options.name) - while (true) { - yield* queue.latch.await - const item = Iterable.headUnsafe(queue.items) - queue.items.delete(item) - if (queue.items.size === 0) { + while (true) { + // close before scanning so a mutation after the scan reopens the latch queue.latch.closeUnsafe() - } - yield* Effect.addFinalizer((exit) => { - if (exit._tag === "Success") { - return Effect.void - } else if (!Exit.hasInterrupts(exit)) { - item.attempts += 1 + const now = clock.currentTimeMillisUnsafe() + let item: Entry | undefined + let nextVisibleAt = Infinity + for (const entry of queue.pending) { + if (entry.attempts >= options.maxAttempts) continue + if (entry.visibleAt <= now) { + item = entry + break + } + nextVisibleAt = Math.min(nextVisibleAt, entry.visibleAt) } - if (item.attempts >= options.maxAttempts) { - return Effect.void + if (item === undefined) { + yield* nextVisibleAt === Infinity + ? queue.latch.await + : Effect.race(queue.latch.await, Effect.sleep(Duration.millis(nextVisibleAt - now))) + continue } - queue.items.add(item) + const entry = item + entry.state = "processing" + entry.attempts += 1 + queue.pending.delete(entry) queue.latch.openUnsafe() - return Effect.void + yield* Effect.addFinalizer( + Effect.fnUntraced(function*(exit) { + const now = clock.currentTimeMillisUnsafe() + if (exit._tag === "Success") { + entry.state = "completed" + entry.stateChangedAt = now + return + } + const deadLetter = deadLetterFromCause(exit.cause) + if (deadLetter !== undefined) { + entry.state = "failed" + entry.stateChangedAt = now + return + } + if (Cause.hasInterruptsOnly(exit.cause)) { + entry.attempts -= 1 + } else if (entry.attempts >= options.maxAttempts) { + entry.state = "failed" + entry.stateChangedAt = now + return + } else { + entry.visibleAt = now + Duration.toMillis(yield* options.retryDelay(entry.attempts)) + } + entry.state = "pending" + queue.pending.add(entry) + queue.latch.openUnsafe() + }) + ) + return { id: entry.id, attempts: entry.attempts, element: entry.element } + } + }), + cleanup: (options) => + Effect.sync(() => { + const now = clock.currentTimeMillisUnsafe() + const completedCutoff = now - Duration.toMillis(options.timeToLive) + const failedCutoff = options.failedTimeToLive === undefined + ? undefined + : now - Duration.toMillis(options.failedTimeToLive) + for (const queue of queues.values()) { + for (const [id, entry] of queue.entries) { + const cutoff = entry.state === "completed" + ? completedCutoff + : entry.state === "failed" + ? failedCutoff + : undefined + if (cutoff !== undefined && entry.stateChangedAt <= cutoff) { + queue.entries.delete(id) + } + } + } }) - return item - } }) }) -}) +) /** * 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. + * 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. * * @category constructors * @since 4.0.0 @@ -371,6 +654,7 @@ export const makeStoreRedis = Effect.fnUntraced(function*( } ) { const redis = yield* Redis.Redis + const clock = yield* Clock.Clock const pollInterval = Duration.max( options?.pollInterval ? Duration.fromInputUnsafe(options.pollInterval) : Duration.seconds(1), @@ -389,35 +673,47 @@ export const makeStoreRedis = Effect.fnUntraced(function*( 1 ) const prefix = options?.prefix ?? "effectq:" - const keyQueue = (name: string) => `${prefix}${name}` const keyLock = (id: string) => `${prefix}${id}:lock` - const keyPending = (name: string) => `${prefix}${name}:pending` - const keyFailed = (name: string) => `${prefix}${name}:failed` + const keysFor = (name: string) => ({ + queue: `${prefix}${name}`, + pending: `${prefix}${name}:pending`, + failed: `${prefix}${name}:failed`, + delayed: `${prefix}${name}:delayed`, + attempts: `${prefix}${name}:attempts`, + ids: `${prefix}${name}:ids` + }) const workerId = crypto.randomUUID() + const ackRetrySchedule = makeAckRetrySchedule(lockExpirationMillis) + type Element = { readonly id: string readonly element: unknown - attempts: number - lastFailure?: string + readonly attempts: number + // the raw wire payload, so requeue and retry do not re-stringify it + readonly payload: string } const requeue = redis.eval(requeueRedis) const complete = redis.eval(completeRedis) const failed = redis.eval(failedRedis) + const retry = redis.eval(retryRedis) const resetQueue = redis.eval(resetQueueRedis) const offer = redis.eval(offerRedis) const take = redis.eval(takeRedis) const expireAll = redis.eval(expireAllRedis) + const trimFailed = redis.eval(trimFailedRedis) + + const queueStates = makeQueueStates() const queues = yield* RcMap.make({ lookup: Effect.fnUntraced(function*(name: string) { - const queueKey = keyQueue(name) - const pendingKey = keyPending(name) + const keys = keysFor(name) const queue = yield* Queue.make() const takers = MutableRef.make(0) const pollLatch = Latch.makeUnsafe() const takenLatch = Latch.makeUnsafe() + const state = queueStates.get(name) yield* Effect.addFinalizer(() => Effect.orDie( @@ -426,47 +722,77 @@ export const makeStoreRedis = Effect.fnUntraced(function*( (elements) => Effect.forEach(elements, (element) => requeue( - queueKey, - pendingKey, + keys.queue, + keys.pending, keyLock(element.id), + keys.attempts, element.id, - JSON.stringify(element) + element.payload ), { concurrency: "unbounded", discard: true }) ) ) ) - yield* resetQueue(queueKey, pendingKey, prefix).pipe( + yield* Effect.suspend(() => + resetQueue( + keys.queue, + keys.pending, + keys.attempts, + keys.failed, + keys.ids, + prefix, + state.maxAttempts, + clock.currentTimeMillisUnsafe() + ) + ).pipe( Effect.andThen(Effect.sleep(lockRefreshMillis)), Effect.forever, Effect.forkScoped ) const poll = (size: number) => - take( - queueKey, - pendingKey, - prefix, - workerId, - size, - lockExpirationMillis + Effect.suspend(() => + take( + keys.queue, + keys.pending, + keys.delayed, + keys.attempts, + prefix, + workerId, + size, + lockExpirationMillis, + clock.currentTimeMillisUnsafe() + ) ) yield* Effect.gen(function*() { while (true) { yield* pollLatch.await yield* Effect.yieldNow + state.nudge.closeUnsafe() const results = takers.current === 0 ? null : yield* poll(takers.current) - if (results === null) { - yield* Effect.sleep(pollInterval) + if (results === null || results.length === 0) { + yield* Effect.race(Effect.sleep(pollInterval), state.nudge.await) continue } takenLatch.closeUnsafe() - yield* Queue.offerAll(queue, results.map((json) => JSON.parse(json))) + const elements: Array = [] + for (let i = 0; i < results.length; i += 2) { + const payload = results[i] as string + const parsed = JSON.parse(payload) + elements.push({ + id: parsed.id, + element: parsed.element, + attempts: Number(results[i + 1]), + payload + }) + } + yield* Queue.offerAll(queue, elements) yield* takenLatch.await yield* Effect.yieldNow } }).pipe( + Effect.tapCause(Effect.logWarning), Effect.sandbox, Effect.retry(Schedule.spaced(500)), Effect.forkScoped, @@ -494,32 +820,58 @@ export const makeStoreRedis = Effect.fnUntraced(function*( }) ) + const scanKeys = (pattern: string) => + Effect.gen(function*() { + const keys: Array = [] + let cursor = "0" + do { + const [next, batch] = yield* redis.send<[string, Array]>( + "SCAN", + cursor, + "MATCH", + pattern, + "COUNT", + "100" + ) + cursor = next + for (const key of batch) keys.push(key) + } while (cursor !== "0") + return keys + }) + return PersistedQueueStore.of({ - offer: ({ element, id, isCustomId, name }) => - Effect.mapError( - isCustomId - ? offer( - `${prefix}${name}`, - `${prefix}${name}:ids`, - id, - JSON.stringify({ id, element, attempts: 0 }) + offer: ({ element, id, isCustomId, name }) => { + const keys = keysFor(name) + const payload = JSON.stringify({ id, element }) + return (isCustomId + ? offer(keys.queue, keys.ids, id, payload) + : redis.send("RPUSH", keys.queue, payload)).pipe( + Effect.mapError(({ cause }) => + new PersistedQueueError({ + message: "Failed to offer element to persisted queue", + cause + }) + ), + Effect.tap(() => + Effect.sync(() => { + queueStates.peek(name)?.nudge.openUnsafe() + }) ) - : redis.send("RPUSH", `${prefix}${name}`, JSON.stringify({ id, element, attempts: 0 })), - ({ cause }) => - new PersistedQueueError({ - message: "Failed to offer element to persisted queue", - cause - }) - ), + ) + }, take: (options) => - Effect.uninterruptibleMask((restore) => - RcMap.get(queues, options.name).pipe( + Effect.uninterruptibleMask((restore) => { + queueStates.get(options.name).maxAttempts = options.maxAttempts + return RcMap.get(queues, options.name).pipe( Effect.flatMap(({ pollLatch, queue, takenLatch, takers }) => { takers.current++ if (takers.current === 1) { pollLatch.openUnsafe() } - return Effect.tap(restore(Queue.take(queue)), () => + // onExit so the decrement also runs when a waiting take is + // interrupted, otherwise the poller keeps fetching for a phantom + // taker + return Effect.onExit(restore(Queue.take(queue)), () => Effect.sync(() => { takers.current-- if (takers.current === 0) { @@ -532,50 +884,89 @@ export const makeStoreRedis = Effect.fnUntraced(function*( }), Effect.scoped, Effect.tap((element) => { + const keys = keysFor(options.name) const lock = keyLock(element.id) activeLockKeys.add(lock) - return Effect.addFinalizer(Exit.match({ - onFailure: (cause) => { - activeLockKeys.delete(lock) - const nextAttempts = element.attempts + 1 - if (nextAttempts >= options.maxAttempts) { - return Effect.orDie(failed( - keyPending(options.name), + const ack = (effect: Effect.Effect) => + effect.pipe( + Effect.retry(ackRetrySchedule), + Effect.orDie, + Effect.ensuring(Effect.sync(() => activeLockKeys.delete(lock))) + ) + return Effect.addFinalizer((exit) => + Effect.suspend(() => { + const now = clock.currentTimeMillisUnsafe() + if (exit._tag === "Success") { + return ack(complete(keys.pending, lock, keys.attempts, keys.ids, element.id, now)) + } + const dead = deadLetterFromCause(exit.cause) + const failElement = (cause: Cause.Cause) => + ack(failed( + keys.pending, lock, - keyFailed(options.name), + keys.failed, + keys.attempts, + keys.ids, element.id, JSON.stringify({ - ...element, + id: element.id, + element: element.element, + attempts: element.attempts, lastFailure: Cause.pretty(cause), - attempts: nextAttempts + failedAt: now }) )) + if (dead !== undefined) { + return failElement(dead.cause) } - return Effect.orDie(requeue( - keyQueue(options.name), - keyPending(options.name), - lock, - element.id, - JSON.stringify( - Cause.hasInterruptsOnly(cause) - ? element - : { - ...element, - lastFailure: Cause.pretty(cause), - attempts: nextAttempts - } - ) - )) - }, - onSuccess: () => { - activeLockKeys.delete(lock) - return Effect.orDie(complete( - keyPending(options.name), - lock, - element.id - )) - } - })) + if (Cause.hasInterruptsOnly(exit.cause)) { + return ack(requeue(keys.queue, keys.pending, lock, keys.attempts, element.id, element.payload)) + } + if (element.attempts >= options.maxAttempts) { + return failElement(exit.cause) + } + return Effect.flatMap(options.retryDelay(element.attempts), (delay) => + ack(retry( + keys.pending, + lock, + keys.delayed, + element.id, + element.payload, + now + Duration.toMillis(delay) + ))) + }) + ) + }) + ) + }), + cleanup: ({ failedTimeToLive, timeToLive }) => + Effect.gen(function*() { + const now = clock.currentTimeMillisUnsafe() + const idsKeys = yield* scanKeys(`${prefix}*:ids`) + const cutoff = now - Duration.toMillis(timeToLive) + yield* Effect.forEach( + idsKeys, + (key) => redis.send("ZREMRANGEBYSCORE", key, "-inf", `(${cutoff}`), + { concurrency: 16, discard: true } + ) + if (failedTimeToLive !== undefined) { + const failedCutoff = now - Duration.toMillis(failedTimeToLive) + const failedKeys = yield* scanKeys(`${prefix}*:failed`) + yield* Effect.forEach( + failedKeys, + (key) => + trimFailed(key, `${key.slice(0, -":failed".length)}:ids`, failedCutoff).pipe( + // each call trims at most one batch, so drain until done + Effect.repeat({ while: (removed) => removed >= trimFailedBatchSize }) + ), + { concurrency: 16, discard: true } + ) + } + }).pipe( + Effect.mapError(({ cause }) => + new PersistedQueueError({ + message: "Failed to clean up persisted queue", + cause }) ) ) @@ -591,7 +982,9 @@ local key_ids = KEYS[2] local id = ARGV[1] local payload = ARGV[2] -local result = redis.call("SADD", key_ids, id) +-- park the dedupe entry outside the timeToLive trim range until the element +-- completes, so unprocessed elements never lose dedupe protection +local result = redis.call("ZADD", key_ids, "NX", "+inf", id) if result == 1 then redis.call("RPUSH", key_queue, payload) end @@ -601,12 +994,28 @@ end ) const resetQueueRedis = Redis.script( - (...args: [keyQueue: string, keyPending: string, prefix: string]) => args, + ( + ...args: [ + keyQueue: string, + keyPending: string, + keyAttempts: string, + keyFailed: string, + keyIds: string, + prefix: string, + maxAttempts: number, + now: number + ] + ) => args, { lua: ` local key_queue = KEYS[1] local key_pending = KEYS[2] +local key_attempts = KEYS[3] +local key_failed = KEYS[4] +local key_ids = KEYS[5] local prefix = ARGV[1] +local max_attempts = tonumber(ARGV[2]) +local now = ARGV[3] local entries = redis.call("HGETALL", key_pending) for i = 1, #entries, 2 do @@ -615,96 +1024,184 @@ for i = 1, #entries, 2 do local lock_key = prefix .. id .. ":lock" local exists = redis.call("EXISTS", lock_key) if exists == 0 then - redis.call("RPUSH", key_queue, payload) + local attempts = tonumber(redis.call("HGET", key_attempts, id) or "0") + if attempts >= max_attempts then + -- compose the failed record by hand so the element payload does not go + -- through a lossy cjson round-trip + local record = string.sub(payload, 1, -2) .. ',"attempts":' .. attempts .. + ',"lastFailure":"Lock expired after final attempt","failedAt":' .. now .. '}' + redis.call("RPUSH", key_failed, record) + redis.call("HDEL", key_attempts, id) + -- failed ids keep their dedupe entry until failedTimeToLive removes the + -- dead-letter record, so park them outside the timeToLive trim range + redis.call("ZADD", key_ids, "XX", "+inf", id) + else + redis.call("RPUSH", key_queue, payload) + end redis.call("HDEL", key_pending, id) end end `, - numberOfKeys: 2 + numberOfKeys: 5 } ) const requeueRedis = Redis.script( - (...args: [keyQueue: string, keyPending: string, keyLock: string, id: string, payload: string]) => args, + ( + ...args: [keyQueue: string, keyPending: string, keyLock: string, keyAttempts: string, id: string, payload: string] + ) => args, { lua: ` local key_queue = KEYS[1] local key_pending = KEYS[2] local key_lock = KEYS[3] +local key_attempts = KEYS[4] local id = ARGV[1] local payload = ARGV[2] redis.call("DEL", key_lock) redis.call("HDEL", key_pending, id) +local attempts = redis.call("HINCRBY", key_attempts, id, -1) +if attempts <= 0 then + redis.call("HDEL", key_attempts, id) +end redis.call("RPUSH", key_queue, payload) `, - numberOfKeys: 3 + numberOfKeys: 4 } ) const completeRedis = Redis.script( - (...args: [keyPending: string, keyLock: string, id: string]) => args, + (...args: [keyPending: string, keyLock: string, keyAttempts: string, keyIds: string, id: string, now: number]) => + args, { lua: ` local key_pending = KEYS[1] local key_lock = KEYS[2] +local key_attempts = KEYS[3] +local key_ids = KEYS[4] local id = ARGV[1] +local now = ARGV[2] redis.call("DEL", key_lock) redis.call("HDEL", key_pending, id) +redis.call("HDEL", key_attempts, id) +redis.call("ZADD", key_ids, "XX", now, id) `, - numberOfKeys: 2 + numberOfKeys: 4 + } +) + +const retryRedis = Redis.script( + ( + ...args: [keyPending: string, keyLock: string, keyDelayed: string, id: string, payload: string, visibleAt: number] + ) => args, + { + lua: ` +local key_pending = KEYS[1] +local key_lock = KEYS[2] +local key_delayed = KEYS[3] +local id = ARGV[1] +local payload = ARGV[2] +local visible_at = ARGV[3] + +redis.call("DEL", key_lock) +redis.call("HDEL", key_pending, id) +redis.call("ZADD", key_delayed, visible_at, payload) +`, + numberOfKeys: 3 } ) const failedRedis = Redis.script( - (...args: [keyPending: string, keyLock: string, keyFailed: string, id: string, payload: string]) => args, + ( + ...args: [ + keyPending: string, + keyLock: string, + keyFailed: string, + keyAttempts: string, + keyIds: string, + id: string, + payload: string + ] + ) => args, { lua: ` local key_pending = KEYS[1] local key_lock = KEYS[2] local key_failed = KEYS[3] +local key_attempts = KEYS[4] +local key_ids = KEYS[5] local id = ARGV[1] local payload = ARGV[2] redis.call("DEL", key_lock) redis.call("HDEL", key_pending, id) +redis.call("HDEL", key_attempts, id) redis.call("RPUSH", key_failed, payload) +-- failed ids keep their dedupe entry until failedTimeToLive removes the +-- dead-letter record, so park them outside the timeToLive trim range +redis.call("ZADD", key_ids, "XX", "+inf", id) `, - numberOfKeys: 3 + numberOfKeys: 5 } ) const takeRedis = Redis.script( ( - ...args: [keyQueue: string, keyPending: string, prefix: string, workerId: string, batchSize: number, pttl: number] + ...args: [ + keyQueue: string, + keyPending: string, + keyDelayed: string, + keyAttempts: string, + prefix: string, + workerId: string, + batchSize: number, + pttl: number, + now: number + ] ) => args, { lua: ` local key_queue = KEYS[1] local key_pending = KEYS[2] +local key_delayed = KEYS[3] +local key_attempts = KEYS[4] local prefix = ARGV[1] local worker_id = ARGV[2] local batch_size = tonumber(ARGV[3]) local pttl = ARGV[4] +local now = ARGV[5] + +local due = redis.call("ZRANGEBYSCORE", key_delayed, "-inf", now, "LIMIT", 0, 100) +if #due > 0 then + for i, payload in ipairs(due) do + redis.call("RPUSH", key_queue, payload) + end + redis.call("ZREM", key_delayed, unpack(due)) +end local payloads = redis.call("LPOP", key_queue, batch_size) if not payloads then return nil end +local result = {} for i, payload in ipairs(payloads) do local id = cjson.decode(payload).id local key_lock = prefix .. id .. ":lock" redis.call("SET", key_lock, worker_id, "PX", pttl) redis.call("HSET", key_pending, id, payload) + local attempts = redis.call("HINCRBY", key_attempts, id, 1) + result[i * 2 - 1] = payload + result[i * 2] = attempts end -return payloads +return result `, - numberOfKeys: 2 + numberOfKeys: 4 } -).withReturnType | null>() +).withReturnType | null>() const expireAllRedis = Redis.script( (keys: ReadonlyArray, ttl: number) => [...keys, ttl], @@ -719,6 +1216,41 @@ end } ) +const trimFailedBatchSize = 1000 + +const trimFailedRedis = Redis.script( + (...args: [keyFailed: string, keyIds: string, cutoff: number]) => args, + { + lua: ` +local key_failed = KEYS[1] +local key_ids = KEYS[2] +local cutoff = tonumber(ARGV[1]) +local removed = 0 + +while removed < ${trimFailedBatchSize} do + local head = redis.call("LINDEX", key_failed, 0) + if not head then break end + local ok, decoded = pcall(cjson.decode, head) + if ok then + local failed_at = tonumber(decoded.failedAt) + if failed_at ~= nil and failed_at >= cutoff then break end + redis.call("LPOP", key_failed) + if decoded.id ~= nil then + redis.call("ZREM", key_ids, decoded.id) + end + else + -- a corrupt head would otherwise block trimming of the whole list + redis.call("LPOP", key_failed) + end + removed = removed + 1 +end + +return removed +`, + numberOfKeys: 2 + } +).withReturnType() + /** * Provides a Redis-backed `PersistedQueueStore` using `makeStoreRedis`. * @@ -777,9 +1309,10 @@ export const makeStoreSql: ( options?.lockExpiration ? Duration.fromInputUnsafe(options.lockExpiration) : Duration.minutes(2), Duration.millis(1) ) - const lockExpirationSql = sql.literal(Math.ceil(Duration.toSeconds(lockExpiration)).toString()) const workerId = crypto.randomUUID() + const ackRetrySchedule = makeAckRetrySchedule(lockExpiration) + yield* Effect.orDie( Migrator.make({})({ loader: sqlMigrations(tableName), @@ -788,45 +1321,55 @@ export const makeStoreSql: ( ) const sqlNow = sql.onDialectOrElse({ - mssql: () => sql.literal("GETDATE()"), + // GETDATE() rounds to 1/300s and can land in the future, hiding freshly + // written visible_at values from the poll query + mssql: () => sql.literal("SYSDATETIME()"), mysql: () => sql.literal("NOW()"), pg: () => sql.literal("NOW()"), // sqlite orElse: () => sql.literal("CURRENT_TIMESTAMP") }) - const expiresAt = sql.onDialectOrElse({ - pg: () => sql`${sqlNow} - INTERVAL '${lockExpirationSql} seconds'`, - mysql: () => sql`DATE_SUB(${sqlNow}, INTERVAL ${lockExpirationSql} SECOND)`, - mssql: () => sql`DATEADD(SECOND, -${lockExpirationSql}, ${sqlNow})`, - orElse: () => sql`datetime(${sqlNow}, '-${lockExpirationSql} seconds')` - }) + // `seconds` is a whole number, possibly negative + const secondsOffset = (seconds: number) => { + const s = sql.literal(seconds.toString()) + return sql.onDialectOrElse({ + pg: () => sql`${sqlNow} + INTERVAL '${s} seconds'`, + mysql: () => sql`DATE_ADD(${sqlNow}, INTERVAL ${s} SECOND)`, + mssql: () => sql`DATEADD(SECOND, ${s}, ${sqlNow})`, + orElse: () => sql`datetime(${sqlNow}, '${s} seconds')` + }) + } + const secondsAgo = (seconds: number) => secondsOffset(-Math.max(Math.ceil(seconds), 0)) + const secondsFromNow = (seconds: number) => secondsOffset(Math.max(Math.ceil(seconds), 0)) + const expiresAt = secondsAgo(Duration.toSeconds(lockExpiration)) const offer = sql.onDialectOrElse({ pg: () => (id: string, name: string, element: string) => sql` - INSERT INTO ${tableNameSql} (id, queue_name, element, completed, attempts, created_at, updated_at) - VALUES (${id}, ${name}, ${element}, FALSE, 0, ${sqlNow}, ${sqlNow}) + INSERT INTO ${tableNameSql} (id, queue_name, element, state, attempts, visible_at, created_at, updated_at) + VALUES (${id}, ${name}, ${element}, 'pending', 0, ${sqlNow}, ${sqlNow}, ${sqlNow}) ON CONFLICT (id, queue_name) DO NOTHING `, mysql: () => (id: string, name: string, element: string) => sql` - INSERT IGNORE INTO ${tableNameSql} (id, queue_name, element, completed, attempts, created_at, updated_at) - VALUES (${id}, ${name}, ${element}, FALSE, 0, ${sqlNow}, ${sqlNow}) + INSERT IGNORE INTO ${tableNameSql} (id, queue_name, element, state, attempts, visible_at, created_at, updated_at) + VALUES (${id}, ${name}, ${element}, 'pending', 0, ${sqlNow}, ${sqlNow}, ${sqlNow}) `, mssql: () => (id: string, name: string, element: string) => sql` - IF NOT EXISTS (SELECT 1 FROM ${tableNameSql} WHERE id = ${id} AND queue_name = ${name}) - BEGIN - INSERT INTO ${tableNameSql} (id, queue_name, element, completed, attempts, created_at, updated_at) - VALUES (${id}, ${name}, ${element}, 0, 0, ${sqlNow}, ${sqlNow}) - END + MERGE ${tableNameSql} WITH (HOLDLOCK) AS target + USING (SELECT ${id} AS id, ${name} AS queue_name) AS source + ON target.id = source.id AND target.queue_name = source.queue_name + WHEN NOT MATCHED THEN + INSERT (id, queue_name, element, state, attempts, visible_at, created_at, updated_at) + VALUES (source.id, source.queue_name, ${element}, 'pending', 0, ${sqlNow}, ${sqlNow}, ${sqlNow}); `, // sqlite orElse: () => (id: string, name: string, element: string) => sql` - INSERT OR IGNORE INTO ${tableNameSql} (id, queue_name, element, completed, attempts, created_at, updated_at) - VALUES (${id}, ${name}, ${element}, FALSE, 0, ${sqlNow}, ${sqlNow}) + INSERT OR IGNORE INTO ${tableNameSql} (id, queue_name, element, state, attempts, visible_at, created_at, updated_at) + VALUES (${id}, ${name}, ${element}, 'pending', 0, ${sqlNow}, ${sqlNow}, ${sqlNow}) ` }) @@ -836,13 +1379,8 @@ export const makeStoreSql: ( }) const stringLiteral = (s: string) => sql.literal(wrapString(s)) - const sqlTrue = sql.onDialectOrElse({ - sqlite: () => sql.literal("1"), - orElse: () => sql.literal("TRUE") - }) - const workerIdSql = stringLiteral(workerId) - const elementIds = new Set() + const elementIds = new Set() const refreshLocks: Effect.Effect = Effect.suspend((): Effect.Effect => { if (elementIds.size === 0) return Effect.void const ids = Array.from(elementIds) @@ -853,55 +1391,64 @@ export const makeStoreSql: ( AND acquired_by = ${workerIdSql} ` }) - const complete = (sequence: number, attempts: number) => { - elementIds.delete(sequence) - return sql` - UPDATE ${tableNameSql} - SET acquired_at = NULL, acquired_by = NULL, updated_at = ${sqlNow}, completed = ${sqlTrue}, attempts = ${attempts} - WHERE sequence = ${sequence} - AND acquired_by = ${workerIdSql} - `.pipe( - Effect.retry({ - times: 5, - schedule: Schedule.exponential(100, 1.5) - }), - Effect.orDie + const ack = ( + statement: Effect.Effect, + sequences: ReadonlyArray + ): Effect.Effect => + statement.pipe( + Effect.retry(ackRetrySchedule), + Effect.orDie, + Effect.ensuring(Effect.sync(() => { + for (const sequence of sequences) { + elementIds.delete(sequence) + } + })), + Effect.asVoid ) - } - const retry = (sequence: number, attempts: number, cause: Cause.Cause) => { - elementIds.delete(sequence) - return sql` - UPDATE ${tableNameSql} - SET acquired_at = NULL, acquired_by = NULL, updated_at = ${sqlNow}, attempts = ${attempts}, last_failure = ${ - Cause.pretty(cause) - } - WHERE sequence = ${sequence} - AND acquired_by = ${workerIdSql} - `.pipe( - Effect.retry({ - times: 5, - schedule: Schedule.exponential(100, 1.5) - }), - Effect.orDie + const complete = (sequence: number | string) => + ack( + sql` + UPDATE ${tableNameSql} + SET acquired_at = NULL, acquired_by = NULL, updated_at = ${sqlNow}, state = 'completed' + WHERE sequence = ${sequence} + AND acquired_by = ${workerIdSql} + `, + [sequence] ) - } - const interrupt = (ids: Array) => { - for (const id of ids) { - elementIds.delete(id) - } - return sql` - UPDATE ${tableNameSql} - SET acquired_at = NULL, acquired_by = NULL - WHERE sequence IN (${sql.literal(ids.join(","))}) - AND acquired_by = ${workerIdSql} - `.pipe( - Effect.retry({ - times: 5, - schedule: Schedule.exponential(100, 1.5) - }), - Effect.orDie + const fail = (sequence: number | string, cause: Cause.Cause) => + ack( + sql` + UPDATE ${tableNameSql} + SET acquired_at = NULL, acquired_by = NULL, updated_at = ${sqlNow}, state = 'failed', last_failure = ${ + Cause.pretty(cause) + } + WHERE sequence = ${sequence} + AND acquired_by = ${workerIdSql} + `, + [sequence] + ) + const retry = (sequence: number | string, delay: Duration.Duration, cause: Cause.Cause) => + ack( + sql` + UPDATE ${tableNameSql} + SET acquired_at = NULL, acquired_by = NULL, updated_at = ${sqlNow}, visible_at = ${ + secondsFromNow(Duration.toSeconds(delay)) + }, last_failure = ${Cause.pretty(cause)} + WHERE sequence = ${sequence} + AND acquired_by = ${workerIdSql} + `, + [sequence] + ) + const interrupt = (ids: Array) => + ack( + sql` + UPDATE ${tableNameSql} + SET acquired_at = NULL, acquired_by = NULL, attempts = attempts - 1 + WHERE sequence IN (${sql.literal(ids.join(","))}) + AND acquired_by = ${workerIdSql} + `, + ids ) - } yield* refreshLocks.pipe( Effect.tapCause(Effect.logWarning), @@ -917,17 +1464,20 @@ export const makeStoreSql: ( type Element = { readonly id: string - sequence: number - readonly queue_name: string - element: string - readonly attempts: number + readonly sequence: number | string + readonly element: string + attempts: number } + const queueStates = makeQueueStates() const mailboxes = yield* RcMap.make({ - lookup: Effect.fnUntraced(function*({ maxAttempts, name }: QueueKey) { + lookup: Effect.fnUntraced(function*(name: string) { + const state = queueStates.get(name) + const maxAttempts = state.maxAttempts const queue = yield* Queue.make() const takers = MutableRef.make(0) const pollLatch = Latch.makeUnsafe() const takenLatch = Latch.makeUnsafe() + const nudge = state.nudge yield* Effect.addFinalizer(() => Effect.flatMap(Queue.clear(queue), (elements) => { @@ -936,35 +1486,55 @@ export const makeStoreSql: ( }) ) + // flip exhausted rows whose lock expired (worker crashed on the final + // attempt) to failed, since no finalizer will ever run for them + yield* sql` + UPDATE ${tableNameSql} + SET state = 'failed', acquired_at = NULL, acquired_by = NULL, updated_at = ${sqlNow}, + last_failure = COALESCE(last_failure, 'Lock expired after final attempt') + WHERE queue_name = ${name} + AND state = 'pending' + AND attempts >= ${maxAttempts} + AND (acquired_at IS NULL OR acquired_at < ${expiresAt}) + `.pipe( + Effect.tapCause(Effect.logWarning), + Effect.ignore, + Effect.schedule(Schedule.spaced(lockRefreshInterval)), + Effect.forkScoped, + Effect.interruptible + ) + const poll = sql.onDialectOrElse({ pg: () => (size: number) => sql` WITH cte AS ( UPDATE ${tableNameSql} - SET acquired_at = ${sqlNow}, acquired_by = ${workerIdSql} + SET acquired_at = ${sqlNow}, acquired_by = ${workerIdSql}, attempts = attempts + 1 WHERE sequence IN ( SELECT sequence FROM ${tableNameSql} WHERE queue_name = ${name} - AND completed = FALSE + AND state = 'pending' AND attempts < ${maxAttempts} + AND visible_at <= ${sqlNow} AND (acquired_at IS NULL OR acquired_at < ${expiresAt}) - ORDER BY updated_at ASC, sequence ASC + ORDER BY visible_at ASC, sequence ASC FOR UPDATE SKIP LOCKED LIMIT ${sql.literal(size.toString())} ) - RETURNING sequence, id, queue_name, element, attempts, updated_at + RETURNING sequence, id, element, attempts, visible_at ) - SELECT sequence, id, queue_name, element, attempts FROM cte - ORDER BY updated_at ASC, sequence ASC + SELECT sequence, id, element, attempts FROM cte + ORDER BY visible_at ASC, sequence ASC `, mysql: () => (size: number) => sql` - SELECT sequence, id, queue_name, element, attempts FROM ${tableNameSql} q + SELECT sequence, id, element, attempts FROM ${tableNameSql} q WHERE queue_name = ${name} - AND completed = FALSE + AND state = 'pending' AND attempts < ${maxAttempts} + AND visible_at <= ${sqlNow} AND (acquired_at IS NULL OR acquired_at < ${expiresAt}) - ORDER BY updated_at ASC, sequence ASC + ORDER BY visible_at ASC, sequence ASC LIMIT ${sql.literal(size.toString())} FOR UPDATE SKIP LOCKED `.pipe( @@ -972,10 +1542,16 @@ export const makeStoreSql: ( if (rows.length === 0) return Effect.void return sql` UPDATE ${tableNameSql} - SET acquired_at = ${sqlNow}, acquired_by = ${workerIdSql} + SET acquired_at = ${sqlNow}, acquired_by = ${workerIdSql}, attempts = attempts + 1 WHERE sequence IN (${sql.literal(rows.map((r) => r.sequence).join(","))}) `.unprepared }), + Effect.map((rows) => { + for (const row of rows) { + row.attempts = Number(row.attempts) + 1 + } + return rows + }), sql.withTransaction ), mssql: () => (size: number) => @@ -983,14 +1559,15 @@ export const makeStoreSql: ( WITH cte AS ( SELECT TOP ${sql.literal(size.toString())} sequence FROM ${tableNameSql} WHERE queue_name = ${name} - AND completed = 0 + AND state = 'pending' AND attempts < ${maxAttempts} + AND visible_at <= ${sqlNow} AND (acquired_at IS NULL OR acquired_at < ${expiresAt}) - ORDER BY updated_at ASC, sequence ASC + ORDER BY visible_at ASC, sequence ASC ) UPDATE q - SET acquired_at = ${sqlNow}, acquired_by = ${workerIdSql} - OUTPUT inserted.sequence, inserted.id, inserted.queue_name, inserted.element, inserted.attempts + SET acquired_at = ${sqlNow}, acquired_by = ${workerIdSql}, attempts = q.attempts + 1 + OUTPUT inserted.sequence, inserted.id, inserted.element, inserted.attempts FROM ${tableNameSql} AS q INNER JOIN cte ON q.sequence = cte.sequence `, @@ -998,17 +1575,18 @@ export const makeStoreSql: ( orElse: () => (size: number) => sql` UPDATE ${tableNameSql} - SET acquired_at = ${sqlNow}, acquired_by = ${workerIdSql} + SET acquired_at = ${sqlNow}, acquired_by = ${workerIdSql}, attempts = attempts + 1 WHERE sequence IN ( SELECT sequence FROM ${tableNameSql} WHERE queue_name = ${name} - AND completed = FALSE + AND state = 'pending' AND attempts < ${maxAttempts} + AND visible_at <= ${sqlNow} AND (acquired_at IS NULL OR acquired_at < ${expiresAt}) - ORDER BY updated_at ASC, sequence ASC + ORDER BY visible_at ASC, sequence ASC LIMIT ${sql.literal(size.toString())} ) - RETURNING sequence, id, queue_name, element, attempts + RETURNING sequence, id, element, attempts ` }) @@ -1016,15 +1594,16 @@ export const makeStoreSql: ( while (true) { yield* pollLatch.await yield* Effect.yieldNow + nudge.closeUnsafe() const results = takers.current === 0 ? [] : yield* poll(takers.current) if (results.length === 0) { - yield* Effect.sleep(pollInterval) + yield* Effect.race(Effect.sleep(pollInterval), nudge.await) continue } takenLatch.closeUnsafe() - for (let i = 0; i < results.length; i++) { - const element = results[i] - elementIds.add(element.sequence) + for (const row of results) { + row.attempts = Number(row.attempts) + elementIds.add(row.sequence) } yield* Queue.offerAll(queue, results) yield* takenLatch.await @@ -1042,6 +1621,62 @@ export const makeStoreSql: ( idleTimeToLive: Duration.seconds(30) }) + const cleanupBatch = sql.onDialectOrElse({ + pg: () => (state: string, seconds: number) => + sql<{ readonly count: number }>` + WITH deleted_entries AS ( + DELETE FROM ${tableNameSql} + WHERE sequence IN ( + SELECT sequence FROM ${tableNameSql} + WHERE state = ${state} AND updated_at <= ${secondsAgo(seconds)} + LIMIT ${sql.literal(String(sqlCleanupBatchSize))} + ) + RETURNING 1 + ) + SELECT COUNT(*)::INT AS count FROM deleted_entries + `.pipe(Effect.map((rows) => rows[0].count)), + mysql: () => + Effect.fnUntraced( + function*(state: string, seconds: number) { + const connection = yield* sql.reserve + const [statement, parameters] = sql` + DELETE FROM ${tableNameSql} + WHERE state = ${state} AND updated_at <= ${secondsAgo(seconds)} + LIMIT ${sql.literal(String(sqlCleanupBatchSize))} + `.compile() + yield* connection.execute(statement, parameters, undefined) + const rows = yield* connection.executeValues("SELECT ROW_COUNT()", []) + return Number(rows[0][0]) + }, + Effect.scoped + ), + mssql: () => (state: string, seconds: number) => + sql<{ readonly sequence: number }>` + DELETE TOP (${sql.literal(String(sqlCleanupBatchSize))}) FROM ${tableNameSql} + OUTPUT DELETED.sequence + WHERE state = ${state} AND updated_at <= ${secondsAgo(seconds)} + `.pipe(Effect.map((rows) => rows.length)), + // sqlite + orElse: () => (state: string, seconds: number) => + sql<{ readonly deleted: number }>` + DELETE FROM ${tableNameSql} + WHERE sequence IN ( + SELECT sequence FROM ${tableNameSql} + WHERE state = ${state} AND updated_at <= ${secondsAgo(seconds)} + LIMIT ${sql.literal(String(sqlCleanupBatchSize))} + ) + RETURNING 1 AS deleted + `.pipe(Effect.map((rows) => rows.length)) + }) + + const cleanupState = (state: string, timeToLive: Duration.Duration) => + cleanupBatch(state, Duration.toSeconds(timeToLive)).pipe( + Effect.repeat({ + while: (deletedCount) => deletedCount === sqlCleanupBatchSize, + schedule: Schedule.spaced(Duration.millis(10)) + }) + ) + return PersistedQueueStore.of({ offer: ({ element, id, name }) => Effect.catchCause(Effect.suspend(() => offer(id, name, JSON.stringify(element))), (cause) => @@ -1050,16 +1685,34 @@ export const makeStoreSql: ( message: "Failed to offer element to persisted queue", cause }) - )), - take: ({ maxAttempts, name }) => - Effect.uninterruptibleMask((restore) => - RcMap.get(mailboxes, new QueueKey({ name, maxAttempts })).pipe( + )).pipe( + Effect.tap(() => + Effect.sync(() => { + queueStates.peek(name)?.nudge.openUnsafe() + }) + ) + ), + take: (options) => { + queueStates.get(options.name).maxAttempts = options.maxAttempts + const loop: Effect.Effect< + { + readonly id: string + readonly attempts: number + readonly element: unknown + }, + PersistedQueueError, + Scope.Scope + > = Effect.uninterruptibleMask((restore) => + RcMap.get(mailboxes, options.name).pipe( Effect.flatMap(({ pollLatch, queue, takenLatch, takers }) => { takers.current++ if (takers.current === 1) { pollLatch.openUnsafe() } - return Effect.tap(restore(Queue.take(queue)), () => + // onExit so the decrement also runs when a waiting take is + // interrupted, otherwise the poller keeps fetching for a phantom + // taker + return Effect.onExit(restore(Queue.take(queue)), () => Effect.sync(() => { takers.current-- if (takers.current === 0) { @@ -1072,19 +1725,56 @@ export const makeStoreSql: ( }), Effect.scoped, restore, - Effect.tap((element) => - Effect.addFinalizer(Exit.match({ - onFailure: (cause) => - Cause.hasInterruptsOnly(cause) - ? interrupt([element.sequence]) - : retry(element.sequence, element.attempts + 1, cause), - onSuccess: () => complete(element.sequence, element.attempts + 1) - })) - ), - Effect.map((element) => ({ - ...element, - element: JSON.parse(element.element) - })) + Effect.flatMap((element) => { + let parsed: unknown + try { + parsed = JSON.parse(element.element) + } catch (defect) { + // a row that cannot be parsed will never succeed, so dead-letter + // it and take the next element + return Effect.andThen(fail(element.sequence, Cause.die(defect)), loop) + } + return Effect.as( + Effect.addFinalizer((exit) => { + if (exit._tag === "Success") { + return complete(element.sequence) + } + const dead = deadLetterFromCause(exit.cause) + if (dead !== undefined) { + return fail(element.sequence, dead.cause) + } + if (Cause.hasInterruptsOnly(exit.cause)) { + return interrupt([element.sequence]) + } + if (element.attempts >= options.maxAttempts) { + return fail(element.sequence, exit.cause) + } + return Effect.flatMap( + options.retryDelay(element.attempts), + (delay) => retry(element.sequence, delay, exit.cause) + ) + }), + { id: element.id, attempts: element.attempts, element: parsed } + ) + }) + ) + ) + return loop + }, + cleanup: ({ failedTimeToLive, timeToLive }) => + Effect.gen(function*() { + yield* cleanupState("completed", timeToLive) + if (failedTimeToLive !== undefined) { + yield* cleanupState("failed", failedTimeToLive) + } + }).pipe( + Effect.catchCause((cause) => + Effect.fail( + new PersistedQueueError({ + message: "Failed to clean up persisted queue", + cause + }) + ) ) ) }) @@ -1126,7 +1816,7 @@ const sqlMigrations = (tableName: string) => updated_at TIMESTAMP NOT NULL )`, mssql: () => - sql`IF NOT EXISTS (SELECT * FROM sysobjects WHERE name=${tableNameSql} AND xtype='U') + sql`IF NOT EXISTS (SELECT * FROM sysobjects WHERE name=${tableName} AND xtype='U') CREATE TABLE ${tableNameSql} ( sequence INT IDENTITY(1,1) PRIMARY KEY, id NVARCHAR(36) NOT NULL, @@ -1160,7 +1850,7 @@ const sqlMigrations = (tableName: string) => yield* sql.onDialectOrElse({ mssql: () => sql`IF NOT EXISTS (SELECT * FROM sys.indexes WHERE name = N'idx_${tableName}_id') - CREATE UNIQUE INDEX idx_${tableNameSql}_id ON ${tableNameSql} (id, queue_name)`, + CREATE UNIQUE INDEX ${sql(`idx_${tableName}_id`)} ON ${tableNameSql} (id, queue_name)`, mysql: () => sql`CREATE UNIQUE INDEX ${sql(`idx_${tableName}_id`)} ON ${tableNameSql} (id, queue_name)`.pipe( Effect.ignore @@ -1172,7 +1862,9 @@ const sqlMigrations = (tableName: string) => yield* sql.onDialectOrElse({ mssql: () => sql`IF NOT EXISTS (SELECT * FROM sys.indexes WHERE name = N'idx_${tableName}_take') - CREATE INDEX idx_${tableNameSql}_take ON ${tableNameSql} (queue_name, completed, attempts, acquired_at)`, + CREATE INDEX ${ + sql(`idx_${tableName}_take`) + } ON ${tableNameSql} (queue_name, completed, attempts, acquired_at)`, mysql: () => sql`CREATE INDEX ${ sql(`idx_${tableName}_take`) @@ -1195,14 +1887,131 @@ const sqlMigrations = (tableName: string) => orElse: () => sql`CREATE INDEX IF NOT EXISTS ${sql(`idx_${tableName}_update`)} ON ${tableNameSql} (sequence, acquired_by)` }) + }), + "0002_upgrade_schema": Effect.gen(function*() { + const sql = (yield* SqlClient.SqlClient).withoutTransforms() + const tableNameSql = sql(tableName) + const takeIndex = sql(`idx_${tableName}_take`) + + yield* sql.onDialectOrElse({ + pg: () => + Effect.gen(function*() { + yield* sql`DROP INDEX IF EXISTS ${takeIndex}` + yield* sql`ALTER TABLE ${tableNameSql} + ADD COLUMN state VARCHAR(10), + ADD COLUMN visible_at TIMESTAMP` + yield* sql`UPDATE ${tableNameSql} + SET state = CASE WHEN completed THEN 'completed' ELSE 'pending' END, + visible_at = updated_at` + yield* sql`ALTER TABLE ${tableNameSql} + ALTER COLUMN sequence TYPE BIGINT, + ALTER COLUMN id TYPE VARCHAR(255), + ALTER COLUMN queue_name TYPE VARCHAR(255), + ALTER COLUMN state SET NOT NULL, + ALTER COLUMN visible_at SET NOT NULL, + DROP COLUMN completed` + }), + mysql: () => + Effect.gen(function*() { + yield* sql`DROP INDEX ${takeIndex} ON ${tableNameSql}` + yield* sql`ALTER TABLE ${tableNameSql} + MODIFY COLUMN id VARCHAR(255) NOT NULL, + MODIFY COLUMN queue_name VARCHAR(255) NOT NULL, + MODIFY COLUMN element MEDIUMTEXT NOT NULL, + MODIFY COLUMN last_failure MEDIUMTEXT NULL, + ADD COLUMN state VARCHAR(10) NULL, + ADD COLUMN visible_at DATETIME NULL` + yield* sql`UPDATE ${tableNameSql} + SET state = CASE WHEN completed THEN 'completed' ELSE 'pending' END, + visible_at = updated_at` + yield* sql`ALTER TABLE ${tableNameSql} + MODIFY COLUMN state VARCHAR(10) NOT NULL, + MODIFY COLUMN visible_at DATETIME NOT NULL, + DROP COLUMN completed` + }), + mssql: () => + Effect.gen(function*() { + const upgradedTableName = `${tableName}_upgrade` + const upgradedTable = sql(upgradedTableName) + yield* sql`CREATE TABLE ${upgradedTable} ( + sequence BIGINT IDENTITY(1,1) PRIMARY KEY, + id NVARCHAR(255) NOT NULL, + queue_name NVARCHAR(255) NOT NULL, + element NVARCHAR(MAX) NOT NULL, + state NVARCHAR(10) NOT NULL, + attempts INT NOT NULL DEFAULT 0, + last_failure NVARCHAR(MAX) NULL, + visible_at DATETIME2 NOT NULL, + acquired_at DATETIME2 NULL, + acquired_by UNIQUEIDENTIFIER NULL, + created_at DATETIME2 NOT NULL, + updated_at DATETIME2 NOT NULL + )` + yield* sql`SET IDENTITY_INSERT ${upgradedTable} ON; + INSERT INTO ${upgradedTable} + (sequence, id, queue_name, element, state, attempts, last_failure, visible_at, + acquired_at, acquired_by, created_at, updated_at) + SELECT sequence, id, queue_name, element, + CASE WHEN completed = 1 THEN 'completed' ELSE 'pending' END, + attempts, last_failure, updated_at, acquired_at, acquired_by, created_at, updated_at + FROM ${tableNameSql}; + SET IDENTITY_INSERT ${upgradedTable} OFF` + yield* sql`DROP TABLE ${tableNameSql}` + yield* sql`EXEC sp_rename ${upgradedTableName}, ${tableName}` + yield* sql`CREATE UNIQUE INDEX ${sql(`idx_${tableName}_id`)} ON ${tableNameSql} (id, queue_name)` + yield* sql`CREATE INDEX ${sql(`idx_${tableName}_update`)} ON ${tableNameSql} (sequence, acquired_by)` + }), + // sqlite rebuilds the table because altering or dropping constrained + // columns is not portable across supported SQLite versions. + orElse: () => + Effect.gen(function*() { + const upgradedTableName = `${tableName}_upgrade` + const upgradedTable = sql(upgradedTableName) + yield* sql`DROP INDEX IF EXISTS ${takeIndex}` + yield* sql`CREATE TABLE ${upgradedTable} ( + sequence INTEGER PRIMARY KEY AUTOINCREMENT, + id TEXT NOT NULL, + queue_name TEXT NOT NULL, + element TEXT NOT NULL, + state TEXT NOT NULL, + attempts INTEGER NOT NULL DEFAULT 0, + last_failure TEXT NULL, + visible_at DATETIME NOT NULL, + acquired_at DATETIME NULL, + acquired_by TEXT NULL, + created_at DATETIME NOT NULL, + updated_at DATETIME NOT NULL + )` + yield* sql`INSERT INTO ${upgradedTable} + (sequence, id, queue_name, element, state, attempts, last_failure, visible_at, + acquired_at, acquired_by, created_at, updated_at) + SELECT sequence, id, queue_name, element, + CASE WHEN completed THEN 'completed' ELSE 'pending' END, + attempts, last_failure, updated_at, acquired_at, acquired_by, created_at, updated_at + FROM ${tableNameSql}` + yield* sql`DROP TABLE ${tableNameSql}` + yield* sql`ALTER TABLE ${upgradedTable} RENAME TO ${tableNameSql}` + yield* sql`CREATE UNIQUE INDEX ${sql(`idx_${tableName}_id`)} ON ${tableNameSql} (id, queue_name)` + yield* sql`CREATE INDEX ${sql(`idx_${tableName}_update`)} ON ${tableNameSql} (sequence, acquired_by)` + }) + }) + + // partial index where supported, so pollers never scan completed rows + yield* sql.onDialectOrElse({ + mssql: () => + sql`CREATE INDEX ${takeIndex} ON ${tableNameSql} (queue_name, visible_at) + WHERE state = 'pending'`, + mysql: () => sql`CREATE INDEX ${takeIndex} ON ${tableNameSql} (queue_name, state, visible_at)`, + pg: () => + sql`CREATE INDEX ${takeIndex} ON ${tableNameSql} (queue_name, visible_at) + WHERE state = 'pending'`, + orElse: () => + sql`CREATE INDEX ${takeIndex} ON ${tableNameSql} (queue_name, visible_at) + WHERE state = 'pending'` + }) }) }) -class QueueKey extends Data.Class<{ - readonly name: string - readonly maxAttempts: number -}> {} - /** * Provides a SQL-backed `PersistedQueueStore` using `makeStoreSql`. * diff --git a/packages/effect/src/unstable/workflow/DurableQueue.ts b/packages/effect/src/unstable/workflow/DurableQueue.ts index c30e5c59dac..1b31b83fe9f 100644 --- a/packages/effect/src/unstable/workflow/DurableQueue.ts +++ b/packages/effect/src/unstable/workflow/DurableQueue.ts @@ -8,6 +8,14 @@ * records the handler's `Exit` through that token so the original workflow can * continue with the typed success or error. * + * Delivery is at-least-once: a crash between handler success and the + * acknowledgement redelivers the item, so handlers must be idempotent. When an + * item exhausts its persisted queue attempts it is dead-lettered: the + * `DurableDeferred` never resolves and the workflow stays parked until the + * item is requeued out of band, while the id-based de-duplication prevents + * replays from resurrecting the failed item. Deployments should run + * `PersistedQueue.layerCleanup` in one instance to prune old completed items. + * * @since 4.0.0 */ import * as Effect from "../../Effect.ts" diff --git a/packages/effect/test/unstable/persistence/PersistedQueue.test.ts b/packages/effect/test/unstable/persistence/PersistedQueue.test.ts index d1065c99f2a..a7d84b76fe1 100644 --- a/packages/effect/test/unstable/persistence/PersistedQueue.test.ts +++ b/packages/effect/test/unstable/persistence/PersistedQueue.test.ts @@ -1,4 +1,79 @@ +import { assert, it } from "@effect/vitest" +import { Effect, Fiber, Layer, Schedule, Schema } from "effect" +import { TestClock } from "effect/testing" import { PersistedQueue } from "effect/unstable/persistence" import * as PersistedQueueTest from "./PersistedQueueTest.ts" PersistedQueueTest.suite("memory", PersistedQueue.layerStoreMemory) + +const Item = Schema.Struct({ + n: Schema.BigInt +}) + +it.layer( + PersistedQueue.layer.pipe(Layer.provideMerge(PersistedQueue.layerStoreMemory)) +)("PersistedQueue.retrySchedule", (it) => { + it.effect("derives growing delays from the persisted attempt count", () => + Effect.gen(function*() { + const queue = yield* PersistedQueue.make({ + name: "retry-schedule-state", + schema: Item, + retrySchedule: Schedule.exponential(1000) + }) + const attempts: Array = [] + const takeFail = queue.take((_val, metadata) => { + attempts.push(metadata.attempts) + return Effect.fail("boom") + }).pipe(Effect.flip) + + yield* queue.offer({ n: 1n }) + + // attempt 1 at t=0 schedules the retry for t=1s + yield* takeFail + + const fiber = yield* takeFail.pipe(Effect.forkScoped) + yield* TestClock.adjust(500) + assert.isUndefined(fiber.pollUnsafe()) + yield* TestClock.adjust(500) + assert.strictEqual(yield* Fiber.join(fiber), "boom") + + // attempt 2 failed at t=1s: the schedule must not restart at 1s, the + // next delay is 2s + const fiber2 = yield* takeFail.pipe(Effect.forkScoped) + yield* TestClock.adjust(1500) + assert.isUndefined(fiber2.pollUnsafe()) + yield* TestClock.adjust(600) + assert.strictEqual(yield* Fiber.join(fiber2), "boom") + + assert.deepStrictEqual(attempts, [1, 2, 3]) + })) +}) + +it.layer( + PersistedQueue.layer.pipe(Layer.provideMerge(PersistedQueue.layerStoreMemory)) +)("PersistedQueue.layerCleanup", (it) => { + it.effect("prunes completed elements on the configured interval", () => + Effect.gen(function*() { + const queue = yield* PersistedQueue.make({ name: "cleanup-layer", schema: Item }) + yield* queue.offer({ n: 1n }, { id: "cleanup-layer-id" }) + yield* queue.take(Effect.succeed) + + yield* Layer.build(PersistedQueue.layerCleanup({ + interval: "1 minute", + timeToLive: "1 hour" + })) + + // within the ttl the id stays deduplicated + yield* TestClock.adjust("2 minutes") + yield* queue.offer({ n: 2n }, { id: "cleanup-layer-id" }) + const fiber = yield* queue.take(Effect.succeed).pipe(Effect.forkScoped) + yield* TestClock.adjust(1000) + assert.isUndefined(fiber.pollUnsafe()) + + // after the ttl a cleanup run prunes the entry and the id can be reused + yield* TestClock.adjust("2 hours") + yield* queue.offer({ n: 3n }, { id: "cleanup-layer-id" }) + yield* TestClock.adjust(1000) + assert.deepStrictEqual(yield* Fiber.join(fiber), { n: 3n }) + })) +}) diff --git a/packages/effect/test/unstable/persistence/PersistedQueueTest.ts b/packages/effect/test/unstable/persistence/PersistedQueueTest.ts index c897e7f5fe1..9aa36874506 100644 --- a/packages/effect/test/unstable/persistence/PersistedQueueTest.ts +++ b/packages/effect/test/unstable/persistence/PersistedQueueTest.ts @@ -1,17 +1,38 @@ import { assert, it } from "@effect/vitest" import type { Vitest } from "@effect/vitest" -import { Effect, Fiber, Latch, Layer, Schema } from "effect" -import type { Duration } from "effect" +import { Duration, Effect, Fiber, Latch, Layer, Schedule, Schema } from "effect" import { TestClock } from "effect/testing" import { PersistedQueue } from "effect/unstable/persistence" +const Item = Schema.Struct({ + n: Schema.BigInt +}) + +// advance the virtual clock one poll cycle, give real-time stores a moment, +// then check the forked take is still waiting +const assertNotDelivered = (fiber: Fiber.Fiber) => + Effect.gen(function*() { + yield* TestClock.adjust(1000) + yield* Effect.sleep(1000).pipe(TestClock.withLive) + assert.isUndefined(fiber.pollUnsafe()) + }) + +// move both the virtual clock and real time past a 1 second ttl. The virtual +// jump is kept small: a large jump can fire the SQL clients' pool timers and +// time out in-flight connection acquisitions +const advancePastTtl = Effect.gen(function*() { + yield* TestClock.adjust("2 seconds") + yield* Effect.sleep(1500).pipe(TestClock.withLive) +}) + export const suiteWith = ( name: string, layer: Layer.Layer, testApi: Vitest.MethodsNonLive, timeout: Duration.Input = "30 seconds" -) => - testApi.layer( +) => { + const testOptions = { timeout: Duration.toMillis(timeout) } + return testApi.layer( PersistedQueue.layer.pipe( Layer.provideMerge(layer) ), @@ -25,10 +46,11 @@ export const suiteWith = ( }) yield* queue.offer({ n: 42n }) - yield* queue.take(Effect.fnUntraced(function*(value) { + yield* queue.take(Effect.fnUntraced(function*(value, metadata) { assert.strictEqual(value.n, 42n) + assert.strictEqual(metadata.attempts, 1) })) - })) + }), testOptions) it.effect("interrupt", () => Effect.gen(function*() { @@ -49,26 +71,22 @@ export const suiteWith = ( yield* latch.await - // allow some real time to pass to ensure the second take is really - // waiting - yield* TestClock.adjust(1000) - yield* Effect.sleep(1000).pipe( - TestClock.withLive - ) - assert.isUndefined(fiber2.pollUnsafe()) + // the second take really waits while the element is being processed + yield* assertNotDelivered(fiber2) yield* Fiber.interrupt(fiber) yield* TestClock.adjust(1000) assert.strictEqual((yield* Fiber.join(fiber2)).n, 42n) - })) + }), testOptions) it.effect("failure", () => Effect.gen(function*() { const queue = yield* PersistedQueue.make({ name: "test-queue-c", - schema: Item + schema: Item, + retrySchedule: Schedule.spaced(0) }) yield* queue.offer({ n: 42n }) @@ -77,11 +95,43 @@ export const suiteWith = ( assert.strictEqual(error, "boom") const value = yield* queue.take((val, { attempts }) => { - assert.strictEqual(attempts, 1) + assert.strictEqual(attempts, 2) return Effect.succeed(val) }) assert.strictEqual(value.n, 42n) - })) + }), testOptions) + + it.effect("delays retries with the retry schedule", () => + Effect.gen(function*() { + const queue = yield* PersistedQueue.make({ + name: "test-queue-retry-schedule", + schema: Item, + retrySchedule: Schedule.spaced(500) + }) + + yield* queue.offer({ n: 42n }) + + const error = yield* queue.take(() => Effect.fail("boom")).pipe(Effect.flip) + assert.strictEqual(error, "boom") + + const fiber = yield* queue.take((_val, { attempts }) => Effect.succeed(attempts)).pipe( + Effect.forkScoped + ) + + // not redelivered before the retry delay elapses + yield* TestClock.adjust(100) + yield* Effect.sleep(100).pipe(TestClock.withLive) + assert.isUndefined(fiber.pollUnsafe()) + + // give real-time backends time to pass the retry delay and poll again + for (let i = 0; i < 3; i++) { + yield* TestClock.adjust(1000) + yield* Effect.sleep(700).pipe(TestClock.withLive) + } + yield* TestClock.adjust(1000) + + assert.strictEqual(yield* Fiber.join(fiber), 2) + }), testOptions) it.effect("idempotent offer", () => Effect.gen(function*() { @@ -99,13 +149,8 @@ export const suiteWith = ( assert.strictEqual(value.n, 42n) })).pipe(Effect.forkScoped) - yield* TestClock.adjust(1000) - yield* Effect.sleep(1000).pipe( - TestClock.withLive - ) - - assert.isUndefined(fiber.pollUnsafe()) - })) + yield* assertNotDelivered(fiber) + }), testOptions) it.effect("deduplicates custom ids independently in each queue", () => Effect.gen(function*() { @@ -121,7 +166,7 @@ export const suiteWith = ( assert.isDefined(fiber.pollUnsafe()) assert.deepStrictEqual(yield* Fiber.join(fiber), { n: 2n }) - })) + }), testOptions) it.effect("does not redeliver in-flight elements", () => Effect.gen(function*() { @@ -141,81 +186,189 @@ export const suiteWith = ( // allow any periodic reset in the store to run while the element is // still being processed - yield* TestClock.adjust(1000) - yield* Effect.sleep(1000).pipe( - TestClock.withLive - ) - - assert.isUndefined(fiber2.pollUnsafe()) + yield* assertNotDelivered(fiber2) // after a successful take the element should not be delivered again yield* release.open yield* Fiber.join(fiber) - yield* TestClock.adjust(1000) - yield* Effect.sleep(1000).pipe( - TestClock.withLive - ) - - assert.isUndefined(fiber2.pollUnsafe()) + yield* assertNotDelivered(fiber2) yield* Fiber.interrupt(fiber2) - })) + }), testOptions) it.effect("stops delivering when maxAttempts is exhausted", () => Effect.gen(function*() { const queue = yield* PersistedQueue.make({ name: "test-queue-exhausted", - schema: Item + schema: Item, + maxAttempts: 1, + retrySchedule: Schedule.spaced(0) }) yield* queue.offer({ n: 42n }) - const error = yield* queue - .take(() => Effect.fail("boom"), { maxAttempts: 1 }) - .pipe(Effect.flip) - + const error = yield* queue.take(() => Effect.fail("boom")).pipe(Effect.flip) assert.strictEqual(error, "boom") - const fiber = yield* queue.take((val) => Effect.succeed(val), { maxAttempts: 1 }).pipe(Effect.forkScoped) - - yield* TestClock.adjust(1000) - yield* Effect.sleep(1000).pipe( - TestClock.withLive - ) + const fiber = yield* queue.take((val) => Effect.succeed(val)).pipe(Effect.forkScoped) - assert.isUndefined(fiber.pollUnsafe()) - })) + yield* assertNotDelivered(fiber) + }), testOptions) - it.effect("counts schema decode failures as attempts", () => + it.effect("dead-letters elements that fail to decode", () => Effect.gen(function*() { const store = yield* PersistedQueue.PersistedQueueStore const queue = yield* PersistedQueue.make({ name: "test-queue-decode-failure", - schema: Item + schema: Item, + retrySchedule: Schedule.spaced(0) }) yield* store.offer({ name: "test-queue-decode-failure", - id: crypto.randomUUID(), + id: "poison", element: { n: null }, - isCustomId: false + isCustomId: true + }) + yield* queue.offer({ n: 42n }) + + // the poison element is marked as failed and skipped + const value = yield* queue.take(Effect.succeed) + assert.deepStrictEqual(value, { n: 42n }) + + // the poison element is not delivered again + const fiber = yield* queue.take(Effect.succeed).pipe(Effect.forkScoped) + yield* assertNotDelivered(fiber) + }), testOptions) + + it.effect("cleanup removes expired completed elements", () => + Effect.gen(function*() { + const store = yield* PersistedQueue.PersistedQueueStore + const queue = yield* PersistedQueue.make({ + name: "test-queue-cleanup", + schema: Item }) - const error = yield* queue.take(Effect.succeed, { maxAttempts: 1 }).pipe(Effect.flip) - assert.isTrue(Schema.isSchemaError(error)) + yield* queue.offer({ n: 1n }, { id: "cleanup-id" }) + yield* queue.take(Effect.succeed) - const fiber = yield* queue.take(Effect.succeed, { maxAttempts: 1 }).pipe(Effect.forkScoped) + // within the ttl the dedupe entry survives, so re-offers are ignored + yield* store.cleanup({ timeToLive: Duration.days(30), failedTimeToLive: undefined }) + yield* queue.offer({ n: 2n }, { id: "cleanup-id" }) + const fiber = yield* queue.take(Effect.succeed).pipe(Effect.forkScoped) + yield* assertNotDelivered(fiber) + // after the ttl the completed element and its dedupe entry go away + yield* advancePastTtl + yield* store.cleanup({ timeToLive: Duration.seconds(1), failedTimeToLive: undefined }) + yield* queue.offer({ n: 3n }, { id: "cleanup-id" }) yield* TestClock.adjust(1000) - yield* Effect.sleep(1000).pipe(TestClock.withLive) + assert.deepStrictEqual(yield* Fiber.join(fiber), { n: 3n }) + }), testOptions) - assert.isUndefined(fiber.pollUnsafe()) - })) - }) + it.effect("cleanup removes failed elements only with failedTimeToLive", () => + Effect.gen(function*() { + const store = yield* PersistedQueue.PersistedQueueStore + const queue = yield* PersistedQueue.make({ + name: "test-queue-cleanup-failed", + schema: Item, + maxAttempts: 1, + retrySchedule: Schedule.spaced(0) + }) -const Item = Schema.Struct({ - n: Schema.BigInt -}) + yield* queue.offer({ n: 1n }, { id: "failed-cleanup-id" }) + const error = yield* queue.take(() => Effect.fail("boom")).pipe(Effect.flip) + assert.strictEqual(error, "boom") + + // without failedTimeToLive the failed element is the dead-letter + // record and is kept, so its id stays deduplicated + yield* advancePastTtl + yield* store.cleanup({ timeToLive: Duration.seconds(1), failedTimeToLive: undefined }) + yield* queue.offer({ n: 2n }, { id: "failed-cleanup-id" }) + const fiber = yield* queue.take(Effect.succeed).pipe(Effect.forkScoped) + yield* assertNotDelivered(fiber) + + // with failedTimeToLive the failed element and its dedupe entry go away + yield* store.cleanup({ timeToLive: Duration.days(30), failedTimeToLive: Duration.seconds(1) }) + yield* queue.offer({ n: 3n }, { id: "failed-cleanup-id" }) + yield* TestClock.adjust(1000) + assert.deepStrictEqual(yield* Fiber.join(fiber), { n: 3n }) + }), testOptions) + + it.effect("cleanup keeps dedupe entries for unprocessed elements", () => + Effect.gen(function*() { + const store = yield* PersistedQueue.PersistedQueueStore + const queue = yield* PersistedQueue.make({ + name: "test-queue-cleanup-pending", + schema: Item + }) + + yield* queue.offer({ n: 1n }, { id: "pending-cleanup-id" }) + + // an element older than the ttl that was never processed keeps its + // dedupe entry, so the re-offer does not enqueue a duplicate + yield* advancePastTtl + yield* store.cleanup({ timeToLive: Duration.seconds(1), failedTimeToLive: undefined }) + yield* queue.offer({ n: 2n }, { id: "pending-cleanup-id" }) + + const value = yield* queue.take(Effect.succeed) + assert.deepStrictEqual(value, { n: 1n }) + + const fiber = yield* queue.take(Effect.succeed).pipe(Effect.forkScoped) + yield* assertNotDelivered(fiber) + }), testOptions) + + it.effect("processes concurrent elements exactly once with retries", () => + Effect.gen(function*() { + const queue = yield* PersistedQueue.make({ + name: "test-queue-soak", + schema: Item, + retrySchedule: Schedule.spaced(0) + }) + + const total = 24 + const deliveries = new Map() + const succeeded = new Set() + const completed = Latch.makeUnsafe() + + yield* Effect.forEach( + Array.from({ length: total }, (_, i) => BigInt(i)), + (n) => queue.offer({ n }), + { concurrency: 8, discard: true } + ) + + const worker = queue.take(({ n }) => + Effect.suspend(() => { + const count = (deliveries.get(n) ?? 0) + 1 + deliveries.set(n, count) + // every third element fails on its first delivery + if (n % 3n === 0n && count === 1) { + return Effect.fail("transient") + } + succeeded.add(n) + if (succeeded.size === total) { + completed.openUnsafe() + } + return Effect.void + }) + ).pipe(Effect.ignore, Effect.forever) + + yield* Effect.forkScoped(worker) + yield* Effect.forkScoped(worker) + yield* Effect.forkScoped(worker) + + while (!completed.isOpen()) { + yield* TestClock.adjust(1000) + yield* Effect.sleep(250).pipe(TestClock.withLive) + } + + assert.strictEqual(succeeded.size, total) + for (const [n, count] of deliveries) { + assert.strictEqual(count, n % 3n === 0n ? 2 : 1, `deliveries for element ${n}`) + } + }), testOptions) + }) +} export const suite = (name: string, layer: Layer.Layer) => suiteWith(name, layer, it) diff --git a/packages/platform/deno/test/DenoRedis.integration.test.ts b/packages/platform/deno/test/DenoRedis.integration.test.ts index 8007b8c3fd6..9315ba5af9f 100644 --- a/packages/platform/deno/test/DenoRedis.integration.test.ts +++ b/packages/platform/deno/test/DenoRedis.integration.test.ts @@ -68,10 +68,11 @@ it.layer(PersistedQueueRedisLayer, { timeout: "30 seconds" })( const queue = yield* PersistedQueue.make({ name: queueName, - schema: RedisItem + schema: RedisItem, + maxAttempts: 1 }) const id = yield* queue.offer({ n: 42 }) - const error = yield* queue.take(() => Effect.fail("boom"), { maxAttempts: 1 }).pipe(Effect.flip) + const error = yield* queue.take(() => Effect.fail("boom")).pipe(Effect.flip) assert.strictEqual(error, "boom") const failed = yield* redis.use((client) => client.lrange(`effectq:${queueName}:failed`, 0, -1)) diff --git a/packages/platform/node/test/NodeRedis.integration.test.ts b/packages/platform/node/test/NodeRedis.integration.test.ts index 7b1b3f150ff..d2fd302ab4c 100644 --- a/packages/platform/node/test/NodeRedis.integration.test.ts +++ b/packages/platform/node/test/NodeRedis.integration.test.ts @@ -4,6 +4,7 @@ import { RedisContainer } from "@testcontainers/redis" import { Effect, Layer, Queue, Schema } from "effect" import * as PersistedCacheTest from "effect-test/unstable/persistence/PersistedCacheTest" import * as PersistedQueueTest from "effect-test/unstable/persistence/PersistedQueueTest" +import { TestClock } from "effect/testing" import { PersistedQueue, Persistence, Redis } from "effect/unstable/persistence" import { createServer } from "node:net" @@ -95,10 +96,11 @@ it.layer(PersistedQueueRedisLayer, { timeout: "30 seconds" })( const queue = yield* PersistedQueue.make({ name: queueName, - schema: RedisItem + schema: RedisItem, + maxAttempts: 1 }) const id = yield* queue.offer({ n: 42 }) - const error = yield* queue.take(() => Effect.fail("boom"), { maxAttempts: 1 }).pipe(Effect.flip) + const error = yield* queue.take(() => Effect.fail("boom")).pipe(Effect.flip) assert.strictEqual(error, "boom") const failed = yield* redis.use((client) => client.lRange(`effectq:${queueName}:failed`, 0, -1)) @@ -111,6 +113,78 @@ it.layer(PersistedQueueRedisLayer, { timeout: "30 seconds" })( const pending = yield* redis.use((client) => client.hLen(`effectq:${queueName}:pending`)) assert.strictEqual(pending, 0) })) + + it.effect("recovers elements from crashed workers", () => + Effect.gen(function*() { + const prefix = "effectq-crash:" + const store = yield* PersistedQueue.makeStoreRedis({ + prefix, + pollInterval: "50 millis", + lockRefreshInterval: "100 millis", + lockExpiration: "1 second" + }) + const factory = yield* PersistedQueue.makeFactory.pipe( + Effect.provideService(PersistedQueue.PersistedQueueStore, store) + ) + const queue = yield* factory.make({ name: "crash-recovery", schema: RedisItem }) + const redis = yield* Redis.Redis + + // simulate a worker that claimed the element and then crashed: the + // element sits in the pending hash with a consumed attempt and no lock + yield* redis.send( + "HSET", + `${prefix}crash-recovery:pending`, + "crashed", + JSON.stringify({ id: "crashed", element: { n: 1 } }) + ) + yield* redis.send("HSET", `${prefix}crash-recovery:attempts`, "crashed", "1") + + const result = yield* queue.take((value, metadata) => Effect.succeed([value.n, metadata.attempts])) + assert.deepStrictEqual(result, [1, 2]) + }).pipe(TestClock.withLive), { timeout: 20000 }) + + it.effect("dead-letters elements from workers that crashed on the final attempt", () => + Effect.gen(function*() { + const prefix = "effectq-crash-exhausted:" + const store = yield* PersistedQueue.makeStoreRedis({ + prefix, + pollInterval: "50 millis", + lockRefreshInterval: "100 millis", + lockExpiration: "1 second" + }) + const factory = yield* PersistedQueue.makeFactory.pipe( + Effect.provideService(PersistedQueue.PersistedQueueStore, store) + ) + const queue = yield* factory.make({ name: "crash-exhausted", schema: RedisItem, maxAttempts: 1 }) + const redis = yield* Redis.Redis + + // the final attempt was claimed by a worker that crashed, so no + // finalizer will ever settle this element + yield* redis.send( + "HSET", + `${prefix}crash-exhausted:pending`, + "crashed", + JSON.stringify({ id: "crashed", element: { n: 1 } }) + ) + yield* redis.send("HSET", `${prefix}crash-exhausted:attempts`, "crashed", "1") + + // an active taker runs the periodic reset that dead-letters such + // elements instead of redelivering them + const fiber = yield* queue.take(Effect.succeed).pipe(Effect.forkScoped) + yield* Effect.sleep(1000) + + const failed = yield* redis.send>("LRANGE", `${prefix}crash-exhausted:failed`, "0", "-1") + assert.strictEqual(failed.length, 1) + const failedItem = JSON.parse(failed[0]) + assert.strictEqual(failedItem.id, "crashed") + assert.deepStrictEqual(failedItem.element, { n: 1 }) + assert.strictEqual(failedItem.attempts, 1) + assert.include(failedItem.lastFailure, "Lock expired after final attempt") + + const pending = yield* redis.send("HLEN", `${prefix}crash-exhausted:pending`) + assert.strictEqual(Number(pending), 0) + assert.isUndefined(fiber.pollUnsafe()) + }).pipe(TestClock.withLive), { timeout: 20000 }) } ) diff --git a/packages/sql/mssql/test/Persistence.integration.test.ts b/packages/sql/mssql/test/Persistence.integration.test.ts index 7a39ac8fa4b..9e2f390db42 100644 --- a/packages/sql/mssql/test/Persistence.integration.test.ts +++ b/packages/sql/mssql/test/Persistence.integration.test.ts @@ -1,6 +1,7 @@ import { Layer } from "effect" import * as PersistedCacheTest from "effect-test/unstable/persistence/PersistedCacheTest" -import { Persistence } from "effect/unstable/persistence" +import * as PersistedQueueTest from "effect-test/unstable/persistence/PersistedQueueTest" +import { PersistedQueue, Persistence } from "effect/unstable/persistence" import { MssqlContainer } from "./utils.ts" PersistedCacheTest.suite( @@ -12,3 +13,8 @@ PersistedCacheTest.suite( "sql-mssql-single", Persistence.layerSql.pipe(Layer.provide(MssqlContainer.layerClient)) ) + +PersistedQueueTest.suite( + "sql-mssql", + PersistedQueue.layerStoreSql().pipe(Layer.provide(MssqlContainer.layerClient)) +) diff --git a/packages/sql/mysql2/test/Persistence.integration.test.ts b/packages/sql/mysql2/test/Persistence.integration.test.ts index 61ac5457cf0..ea8f80e4d83 100644 --- a/packages/sql/mysql2/test/Persistence.integration.test.ts +++ b/packages/sql/mysql2/test/Persistence.integration.test.ts @@ -1,5 +1,5 @@ import { assert, it } from "@effect/vitest" -import { Effect, Layer } from "effect" +import { Effect, Layer, Schema } from "effect" import * as PersistedCacheTest from "effect-test/unstable/persistence/PersistedCacheTest" import * as PersistedQueueTest from "effect-test/unstable/persistence/PersistedQueueTest" import * as SqlCleanupTest from "effect-test/unstable/persistence/SqlCleanupTest" @@ -15,6 +15,25 @@ it.layer(MysqlContainer.layerClient, { timeout: "90 seconds" })("Persistence", ( PersistedQueueTest.suiteWith("sql-mysql2", PersistedQueue.layerStoreSql(), it) + // elements are stored in a MEDIUMTEXT column, so payloads must survive the + // 64KB TEXT limit + it.effect("round-trips queue payloads larger than 64KB", () => + Effect.gen(function*() { + const store = yield* PersistedQueue.makeStoreSql({ + tableName: "effect_queue_large_payload", + pollInterval: "10 millis" + }) + const factory = yield* PersistedQueue.makeFactory.pipe( + Effect.provideService(PersistedQueue.PersistedQueueStore, store) + ) + const queue = yield* factory.make({ name: "large-payload", schema: Schema.String }) + + const payload = "x".repeat(200_000) + yield* queue.offer(payload) + const value = yield* queue.take(Effect.succeed) + assert.strictEqual(value, payload) + }).pipe(TestClock.withLive), { timeout: 30000 }) + it.effect("deletes expired entries in batches", () => Effect.gen(function*() { const sql = (yield* SqlClient.SqlClient).withoutTransforms() diff --git a/packages/sql/pg/test/Persistence.integration.test.ts b/packages/sql/pg/test/Persistence.integration.test.ts index a8b15032855..06cd0511a56 100644 --- a/packages/sql/pg/test/Persistence.integration.test.ts +++ b/packages/sql/pg/test/Persistence.integration.test.ts @@ -1,5 +1,5 @@ import { assert, it } from "@effect/vitest" -import { Effect, Exit, Fiber, Latch, Layer, Schema } from "effect" +import { Duration, Effect, Fiber, Latch, Layer, Schema } from "effect" import * as PersistedCacheTest from "effect-test/unstable/persistence/PersistedCacheTest" import * as PersistedQueueTest from "effect-test/unstable/persistence/PersistedQueueTest" import * as SqlCleanupTest from "effect-test/unstable/persistence/SqlCleanupTest" @@ -43,9 +43,14 @@ it.layer(PgContainer.layerClient, { timeout: "30 seconds" })("PersistedQueue SQL isCustomId: false }) + const takeOptions = { + name: "lock-refresh", + maxAttempts: 10, + retryDelay: () => Effect.succeed(Duration.zero) + } const acquired = Latch.makeUnsafe() const first = yield* Effect.scoped(Effect.gen(function*() { - yield* store1.take({ name: "lock-refresh", maxAttempts: 10 }) + yield* store1.take(takeOptions) yield* acquired.open return yield* Effect.never })).pipe(Effect.forkScoped) @@ -53,7 +58,7 @@ it.layer(PgContainer.layerClient, { timeout: "30 seconds" })("PersistedQueue SQL yield* acquired.await const second = yield* Effect.scoped( - store2.take({ name: "lock-refresh", maxAttempts: 10 }) + store2.take(takeOptions) ).pipe(Effect.forkScoped) yield* Effect.sleep("1500 millis") @@ -64,7 +69,7 @@ it.layer(PgContainer.layerClient, { timeout: "30 seconds" })("PersistedQueue SQL assert.deepStrictEqual(received.element, element) }).pipe(TestClock.withLive)) - it.effect("counts malformed JSON as an attempt and continues", () => + it.effect("dead-letters malformed JSON and continues", () => Effect.gen(function*() { const tableName = "effect_queue_invalid_json" const store = yield* PersistedQueue.makeStoreSql({ @@ -91,21 +96,140 @@ it.layer(PgContainer.layerClient, { timeout: "30 seconds" })("PersistedQueue SQL yield* sql`UPDATE ${table} SET element = ${"{"} WHERE id = ${poisonId}` yield* queue.offer("valid") - const malformed = yield* Effect.exit(queue.take(Effect.succeed, { maxAttempts: 1 })) - assert.isTrue(Exit.isFailure(malformed)) + // the malformed element is skipped and the next one is delivered + const value = yield* queue.take(Effect.succeed) + assert.strictEqual(value, "valid") const rows = yield* sql<{ + readonly state: string readonly attempts: number readonly last_failure: string | null - }>`SELECT attempts, last_failure FROM ${table} WHERE id = ${poisonId}` - assert.strictEqual(rows[0].attempts, 1) + }>`SELECT state, attempts, last_failure FROM ${table} WHERE id = ${poisonId}` + assert.strictEqual(rows[0].state, "failed") + assert.strictEqual(Number(rows[0].attempts), 1) assert.isNotNull(rows[0].last_failure) - - const value = yield* queue.take(Effect.succeed, { maxAttempts: 1 }) - assert.strictEqual(value, "valid") }).pipe(TestClock.withLive)) + + it.effect("processes elements exactly once across two workers", () => + Effect.gen(function*() { + const options = { + tableName: "effect_queue_two_workers", + pollInterval: "10 millis" + } as const + const makeQueue = Effect.fnUntraced(function*(store: PersistedQueue.PersistedQueueStore["Service"]) { + const factory = yield* PersistedQueue.makeFactory.pipe( + Effect.provideService(PersistedQueue.PersistedQueueStore, store) + ) + return yield* factory.make({ name: "two-workers", schema: Schema.Number }) + }) + const queue1 = yield* makeQueue(yield* PersistedQueue.makeStoreSql(options)) + const queue2 = yield* makeQueue(yield* PersistedQueue.makeStoreSql(options)) + + const total = 20 + yield* Effect.forEach( + Array.from({ length: total }, (_, i) => i), + (n) => queue1.offer(n), + { concurrency: 8, discard: true } + ) + + const seen = new Map() + const worker = (queue: typeof queue1) => + queue.take((n) => + Effect.sync(() => { + seen.set(n, (seen.get(n) ?? 0) + 1) + }) + ).pipe(Effect.forever, Effect.forkScoped) + yield* worker(queue1) + yield* worker(queue1) + yield* worker(queue2) + yield* worker(queue2) + + yield* waitFor(() => Effect.sync(() => seen.size === total)) + + assert.strictEqual(seen.size, total) + for (const [n, count] of seen) { + assert.strictEqual(count, 1, `deliveries for element ${n}`) + } + }).pipe(TestClock.withLive), { timeout: 30000 }) + + it.effect("recovers elements from crashed workers after lock expiration", () => + Effect.gen(function*() { + const tableName = "effect_queue_crash_recovery" + const store = yield* PersistedQueue.makeStoreSql({ + tableName, + pollInterval: "10 millis", + lockExpiration: "1 second" + }) + const factory = yield* PersistedQueue.makeFactory.pipe( + Effect.provideService(PersistedQueue.PersistedQueueStore, store) + ) + const queue = yield* factory.make({ name: "crash-recovery", schema: Schema.Number }) + const sql = (yield* SqlClient.SqlClient).withoutTransforms() + const table = sql(tableName) + + const id = yield* queue.offer(1) + // simulate a worker that claimed the element and then crashed: the lock + // is held by a dead worker and the claim consumed an attempt + yield* sql` + UPDATE ${table} + SET acquired_by = ${crypto.randomUUID()}, acquired_at = NOW(), attempts = 1 + WHERE id = ${id} + ` + + const attempts = yield* queue.take((_n, metadata) => Effect.succeed(metadata.attempts)) + assert.strictEqual(attempts, 2) + }).pipe(TestClock.withLive), { timeout: 20000 }) + + it.effect("dead-letters elements from workers that crashed on the final attempt", () => + Effect.gen(function*() { + const tableName = "effect_queue_crash_exhausted" + const store = yield* PersistedQueue.makeStoreSql({ + tableName, + pollInterval: "10 millis", + lockExpiration: "500 millis", + lockRefreshInterval: "200 millis" + }) + const factory = yield* PersistedQueue.makeFactory.pipe( + Effect.provideService(PersistedQueue.PersistedQueueStore, store) + ) + const queue = yield* factory.make({ name: "crash-exhausted", schema: Schema.Number, maxAttempts: 1 }) + const sql = (yield* SqlClient.SqlClient).withoutTransforms() + const table = sql(tableName) + + const id = yield* queue.offer(1) + // the final attempt was claimed by a worker that crashed, so no + // finalizer will ever settle this element + yield* sql` + UPDATE ${table} + SET acquired_by = ${crypto.randomUUID()}, acquired_at = NOW(), attempts = 1 + WHERE id = ${id} + ` + + // an active taker runs the periodic pass that flips such rows to failed + const fiber = yield* queue.take(Effect.succeed).pipe(Effect.forkScoped) + + const state = () => + sql<{ readonly state: string; readonly last_failure: string | null }>` + SELECT state, last_failure FROM ${table} WHERE id = ${id} + `.pipe(Effect.map((rows) => rows[0])) + yield* waitFor(() => state().pipe(Effect.map((row) => row.state === "failed"))) + + const row = yield* state() + assert.strictEqual(row.state, "failed") + assert.include(row.last_failure ?? "", "Lock expired after final attempt") + assert.isUndefined(fiber.pollUnsafe()) + }).pipe(TestClock.withLive), { timeout: 20000 }) }) +// polls a condition on the live clock until it holds or the rounds run out +const waitFor = (condition: () => Effect.Effect) => + Effect.gen(function*() { + for (let i = 0; i < 100; i++) { + if (yield* condition()) return + yield* Effect.sleep(100) + } + }) + it.layer(PgContainer.layerClient, { timeout: "30 seconds" })("Persistence SQL cleanup", (it) => { it.effect("deletes expired entries in batches", () => Effect.gen(function*() { diff --git a/packages/sql/pglite/test/PersistedQueue.test.ts b/packages/sql/pglite/test/PersistedQueue.test.ts index 737028902da..839e5b53e64 100644 --- a/packages/sql/pglite/test/PersistedQueue.test.ts +++ b/packages/sql/pglite/test/PersistedQueue.test.ts @@ -8,25 +8,10 @@ const ClientLayer = PgliteClient.layer({}) describe("PersistedQueue SQL migrations", () => { layer(ClientLayer, { timeout: "30 seconds" })((it) => { - it.effect("adopts an existing queue table and records the migration once", () => + it.effect("runs fresh-install migrations once", () => Effect.gen(function*() { const sql = (yield* SqlClient.SqlClient).withoutTransforms() const tableName = "persisted_queue_migration_test" - const table = sql(tableName) - - yield* sql`CREATE TABLE ${table} ( - sequence SERIAL PRIMARY KEY, - id VARCHAR(36) NOT NULL, - queue_name VARCHAR(100) NOT NULL, - element TEXT NOT NULL, - completed BOOLEAN NOT NULL, - attempts INTEGER NOT NULL DEFAULT 0, - last_failure TEXT NULL, - acquired_at TIMESTAMP NULL, - acquired_by UUID NULL, - created_at TIMESTAMP NOT NULL, - updated_at TIMESTAMP NOT NULL - )` yield* PersistedQueue.makeStoreSql({ tableName }) yield* PersistedQueue.makeStoreSql({ tableName }) @@ -35,7 +20,10 @@ describe("PersistedQueue SQL migrations", () => { readonly migration_id: number readonly name: string }>`SELECT migration_id, name FROM ${sql(`${tableName}_migrations`)} ORDER BY migration_id` - assert.deepStrictEqual(migrations, [{ migration_id: 1, name: "create_table" }]) + assert.deepStrictEqual(migrations, [ + { migration_id: 1, name: "create_table" }, + { migration_id: 2, name: "upgrade_schema" } + ]) const indexes = yield* sql<{ readonly indexname: string }>` SELECT indexname FROM pg_indexes