1/**
2 * Stores schema-encoded queue work in persistent storage.
3 *
4 * A `PersistedQueue<A>` keeps JSON-encoded values in a named queue and lets
5 * workers take one value at a time inside a scoped processing window. It is
6 * useful for durable handoffs, background jobs, outbox-style integrations, and
7 * work that should retry across fibers, process restarts, or multiple workers.
8 * This module includes a queue factory, store service, id-based de-duplication,
9 * retry handling, and in-memory, Redis, and SQL-backed store layers.
10 *
11 * Delivery is at-least-once: a crash between handler success and the
12 * acknowledgement redelivers the element, so handlers must be idempotent.
13 *
14 * @stability unstable
15 * @since 4.0.0
16 */
17import * as Cause from "../Cause.ts"
18import * as Clock from "../Clock.ts"
19import * as Context from "../Context.ts"
20import * as Duration from "../Duration.ts"
21import * as Effect from "../Effect.ts"
22import { flow } from "../Function.ts"
23import { sqlCleanupBatchSize } from "../internal/persistence.ts"
24import * as Latch from "../Latch.ts"
25import * as Layer from "../Layer.ts"
26import * as MutableRef from "../MutableRef.ts"
27import * as Predicate from "../Predicate.ts"
28import * as Pull from "../Pull.ts"
29import * as Queue from "../Queue.ts"
30import * as RcMap from "../RcMap.ts"
31import * as Schedule from "../Schedule.ts"
32import * as Schema from "../Schema.ts"
33import * as Scope from "../Scope.ts"
34import * as Migrator from "../sql/Migrator.ts"
35import * as SqlClient from "../sql/SqlClient.ts"
36import type { SqlError } from "../sql/SqlError.ts"
37import * as Redis from "./Redis.ts"
38
39/**
40 * Runtime type identifier for `PersistedQueue` values.
41 *
42 * @stability unstable
43 * @category type IDs
44 * @since 4.0.0
45 */
46export const TypeId: TypeId = "~effect/persistence/PersistedQueue"
47
48/**
49 * Type-level identifier used to brand `PersistedQueue` values.
50 *
51 * @stability unstable
52 * @category type IDs
53 * @since 4.0.0
54 */
55export type TypeId = "~effect/persistence/PersistedQueue"
56
57/**
58 * Persistent queue of schema-encoded values.
59 *
60 * **Details**
61 *
62 * `offer` enqueues values by id, and `take` processes one value at a time,
63 * marking it complete on success or retrying it with the queue's retry
64 * schedule until the maximum attempts is reached, after which it is marked as
65 * failed.
66 *
67 * Delivery is at-least-once: a crash between handler success and the
68 * acknowledgement redelivers the element, so handlers must be idempotent.
69 *
70 * @stability unstable
71 * @category models
72 * @since 4.0.0
73 */
74export interface PersistedQueue<in out A, out R = never> {
75 readonly [TypeId]: TypeId
76
77 /**
78 * Adds an element to the queue and returns the id of the enqueued element.
79 *
80 * **Details**
81 *
82 * If an element with the same id already exists in the queue, it will not be
83 * added again. De-duplication survives completion until the id is removed by
84 * `layerCleanup`.
85 */
86 readonly offer: (value: A, options?: {
87 readonly id: string | undefined
88 }) => Effect.Effect<string, PersistedQueueError | Schema.SchemaError, R>
89
90 /**
91 * Takes an element from the queue, waiting until one is available when the
92 * queue is empty.
93 *
94 * **Details**
95 *
96 * If the returned effect succeeds, the element is marked as processed;
97 * otherwise it will be retried with the queue's retry schedule until the
98 * maximum attempts is reached, after which it is marked as failed.
99 *
100 * An attempt is counted when the element is claimed, so `attempts` in the
101 * handler metadata is 1-based ("this is attempt 3"). A handler crash that
102 * takes down the process still consumes an attempt.
103 *
104 * Elements that fail to decode with the queue's schema are marked as failed
105 * immediately and the next element is taken instead, so schema decode errors
106 * never surface from `take`.
107 */
108 readonly take: <XA, XE, XR>(
109 f: (value: A, metadata: {
110 readonly id: string
111 readonly attempts: number
112 }) => Effect.Effect<XA, XE, XR>
113 ) => Effect.Effect<XA, XE | PersistedQueueError, R | XR>
114}
115
116/**
117 * Service for constructing named `PersistedQueue` instances from schemas.
118 *
119 * @stability unstable
120 * @category services
121 * @since 4.0.0
122 */
123export class PersistedQueueFactory extends Context.Service<
124 PersistedQueueFactory,
125 {
126 readonly make: <S extends Schema.Constraint>(options: {
127 readonly name: string
128 readonly schema: S
129 readonly maxAttempts?: number | undefined
130 readonly retrySchedule?: Schedule.Schedule<any, number> | undefined
131 }) => Effect.Effect<PersistedQueue<S["Type"], S["EncodingServices"] | S["DecodingServices"]>>
132 }
133>()("effect/persistence/PersistedQueue/PersistedQueueFactory") {}
134
135/**
136 * Accesses `PersistedQueueFactory` to create a named persisted queue for a
137 * schema.
138 *
139 * **Details**
140 *
141 * `maxAttempts` defaults to 10. `retrySchedule` controls the delay before a
142 * failed element becomes visible again, and defaults to an exponential delay
143 * starting at 1 second and capped at 5 minutes.
144 *
145 * The schedule's state is the element's persisted attempt count. On each
146 * failure the schedule is replayed up to the current attempt, so delays keep
147 * progressing even when consecutive retries run in different processes. The
148 * schedule input is the attempt number.
149 *
150 * Replay simulates elapsed time from the sum of the computed delays.
151 * Attempt-driven schedules are therefore exact, while wall-clock-anchored
152 * schedules observe this idealized time. In particular,
153 * `Schedule.upTo({ duration })` caps the summed delays rather than real time
154 * since the original failure.
155 *
156 * @stability unstable
157 * @category accessors
158 * @since 4.0.0
159 */
160export const make = <S extends Schema.Constraint>(options: {
161 readonly name: string
162 readonly schema: S
163 readonly maxAttempts?: number | undefined
164 readonly retrySchedule?: Schedule.Schedule<any, number> | undefined
165}): Effect.Effect<
166 PersistedQueue<S["Type"], S["EncodingServices"] | S["DecodingServices"]>,
167 never,
168 PersistedQueueFactory
169> => PersistedQueueFactory.use((factory) => factory.make(options))
170
171const defaultRetrySchedule = Schedule.min([
172 Schedule.exponential("1 second"),
173 Schedule.spaced("5 minutes")
174])
175
176// The persisted attempt count is the schedule state: consecutive retries of an
177// element can run in different processes, so a fresh schedule step is replayed
178// up to the given attempt on every call. Delays therefore depend only on the
179// attempt count, with elapsed time simulated from the summed delays.
180const retryDelay = Effect.fnUntraced(function*(
181 schedule: Schedule.Schedule<any, number>,
182 attempts: number
183): Effect.fn.Return<Duration.Duration> {
184 const step = yield* Schedule.toStep(schedule)
185 let now = 0
186 let delay = Duration.zero
187 for (let i = 0; i < attempts; i++) {
188 const result = yield* Pull.catchDone(step(now, i + 1), () => Effect.undefined)
189 if (result === undefined) {
190 // the schedule is done, keep using its final delay
191 break
192 }
193 delay = result[1]
194 now += Duration.toMillis(delay)
195 }
196 return delay
197})
198
199/**
200 * Creates a `PersistedQueueFactory` from the current `PersistedQueueStore`.
201 *
202 * **Details**
203 *
204 * Values are encoded and decoded with the supplied schema, automatically
205 * assigned an id when needed, and acknowledged or retried according to the
206 * `take` handler's exit.
207 *
208 * @stability unstable
209 * @category constructors
210 * @since 4.0.0
211 */
212export const makeFactory = Effect.gen(function*() {
213 const store = yield* PersistedQueueStore
214
215 return PersistedQueueFactory.of({
216 make<S extends Schema.Constraint>(options: {
217 readonly name: string
218 readonly schema: S
219 readonly maxAttempts?: number | undefined
220 readonly retrySchedule?: Schedule.Schedule<any, number> | undefined
221 }) {
222 const jsonSchema = Schema.toCodecJson(options.schema)
223 const encodeUnknown = Schema.encodeUnknownEffect(jsonSchema)
224 const decodeUnknown = Schema.decodeUnknownEffect(jsonSchema)
225 const retrySchedule = options.retrySchedule ?? defaultRetrySchedule
226 const takeOptions = {
227 name: options.name,
228 maxAttempts: options.maxAttempts ?? 10,
229 retryDelay: (attempts: number) => retryDelay(retrySchedule, attempts)
230 }
231
232 return Effect.succeed<PersistedQueue<S["Type"], S["EncodingServices"] | S["DecodingServices"]>>({
233 [TypeId]: TypeId,
234 offer: (value, opts) =>
235 Effect.flatMap(
236 encodeUnknown(value),
237 (element) => {
238 const id = opts?.id ?? crypto.randomUUID()
239 return Effect.as(
240 store.offer({
241 name: options.name,
242 id,
243 element,
244 isCustomId: opts?.id !== undefined
245 }),
246 id
247 )
248 }
249 ),
250 take: <XA, XE, XR>(
251 f: (value: S["Type"], metadata: {
252 readonly id: string
253 readonly attempts: number
254 }) => Effect.Effect<XA, XE, XR>
255 ) => {
256 const loop: Effect.Effect<XA, any, any> = Effect.scopedWith((scope) =>
257 store.take(takeOptions).pipe(
258 Scope.provide(scope),
259 Effect.flatMap((item) =>
260 decodeUnknown(item.element).pipe(
261 Effect.catchCause((cause): Effect.Effect<never, DeadLetter> =>
262 Cause.hasInterruptsOnly(cause)
263 ? Effect.failCause(cause as Cause.Cause<never>)
264 : Effect.fail(new DeadLetter(cause))
265 ),
266 Effect.flatMap((value) => f(value as S["Type"], { id: item.id, attempts: item.attempts }))
267 )
268 )
269 )
270 ).pipe(
271 Effect.catchIf(
272 isDeadLetter,
273 (): Effect.Effect<XA, any, any> => loop
274 )
275 )
276 return loop as Effect.Effect<
277 XA,
278 XE | PersistedQueueError,
279 XR | S["EncodingServices"] | S["DecodingServices"]
280 >
281 }
282 })
283 }
284 })
285})
286
287/**
288 * Provides `PersistedQueueFactory` using the current `PersistedQueueStore`.
289 *
290 * @stability unstable
291 * @category layers
292 * @since 4.0.0
293 */
294export const layer: Layer.Layer<
295 PersistedQueueFactory,
296 never,
297 PersistedQueueStore
298> = Layer.effect(PersistedQueueFactory, makeFactory)
299
300/**
301 * Runs `PersistedQueueStore.cleanup` on a schedule.
302 *
303 * **Details**
304 *
305 * Completed elements are retained for `timeToLive` (default 30 days) so
306 * offer de-duplication keeps working across replays, then removed. Failed
307 * elements are the dead-letter record and are kept forever unless
308 * `failedTimeToLive` is set.
309 *
310 * Run this layer in one instance of a deployment rather than on every worker;
311 * racing instances are harmless since deletes are idempotent, but the work is
312 * redundant.
313 *
314 * @stability unstable
315 * @category layers
316 * @since 4.0.0
317 */
318export const layerCleanup = (options?: {
319 readonly interval?: Duration.Input | undefined
320 readonly timeToLive?: Duration.Input | undefined
321 readonly failedTimeToLive?: Duration.Input | undefined
322}): Layer.Layer<never, never, PersistedQueueStore> =>
323 Layer.effectDiscard(Effect.gen(function*() {
324 const store = yield* PersistedQueueStore
325 const cleanupOptions = {
326 timeToLive: Duration.fromInputUnsafe(options?.timeToLive ?? Duration.days(30)),
327 failedTimeToLive: options?.failedTimeToLive === undefined
328 ? undefined
329 : Duration.fromInputUnsafe(options.failedTimeToLive)
330 }
331 yield* store.cleanup(cleanupOptions).pipe(
332 Effect.catchCause((cause) => Effect.logWarning("Failed to clean up persisted queue", cause)),
333 Effect.repeat(Schedule.spaced(options?.interval ?? Duration.hours(1))),
334 Effect.interruptible,
335 Effect.forkScoped,
336 Effect.annotateLogs({
337 module: "effect/persistence/PersistedQueue",
338 fiber: "cleanup"
339 })
340 )
341 }))
342
343/**
344 * Runtime type identifier for `PersistedQueueError`.
345 *
346 * @stability unstable
347 * @category type IDs
348 * @since 4.0.0
349 */
350export const ErrorTypeId: ErrorTypeId = "~effect/persistence/PersistedQueue/PersistedQueueError"
351
352/**
353 * Type-level identifier used to brand `PersistedQueueError` values.
354 *
355 * @stability unstable
356 * @category type IDs
357 * @since 4.0.0
358 */
359export type ErrorTypeId = "~effect/persistence/PersistedQueue/PersistedQueueError"
360
361/**
362 * Error raised by persisted queue store operations.
363 *
364 * @stability unstable
365 * @category errors
366 * @since 4.0.0
367 */
368export class PersistedQueueError extends Schema.Error<PersistedQueueError>(
369 "effect/persistence/PersistedQueue/PersistedQueueError"
370)({
371 _tag: Schema.tag("PersistedQueueError"),
372 message: Schema.String,
373 cause: Schema.optional(Schema.Defect())
374}) {
375 /**
376 * Marks this value as a persisted queue error for runtime guards.
377 *
378 * @stability unstable
379 * @since 4.0.0
380 */
381 readonly [ErrorTypeId]: ErrorTypeId = ErrorTypeId
382}
383
384// Local per-queue state shared between offer, take, and the mailbox lookup in
385// the Redis and SQL stores: take records maxAttempts here before the mailbox
386// is created, and offer opens the nudge latch so local pollers skip the poll
387// interval.
388type QueueState = {
389 maxAttempts: number
390 readonly nudge: Latch.Latch
391}
392
393const makeQueueStates = (): {
394 readonly peek: (name: string) => QueueState | undefined
395 readonly get: (name: string) => QueueState
396} => {
397 const states = new Map<string, QueueState>()
398 return {
399 peek: (name) => states.get(name),
400 get: (name) => {
401 let state = states.get(name)
402 if (state === undefined) {
403 // the placeholder is overwritten by take before any mailbox reads it,
404 // and is kept int32-safe for SQL parameters
405 state = { maxAttempts: 2147483647, nudge: Latch.makeUnsafe(false) }
406 states.set(name, state)
407 }
408 return state
409 }
410 }
411}
412
413const makeAckRetrySchedule = (lockExpiration: Duration.Input) =>
414 Schedule.min([
415 Schedule.exponential(200, 1.5),
416 Schedule.spaced(5000)
417 ]).pipe(Schedule.upTo({ duration: lockExpiration }))
418
419// Internal signal used by the factory to tell a store that a taken element
420// cannot ever be processed (its stored payload fails to decode) and must be
421// dead-lettered instead of retried.
422class DeadLetter {
423 readonly _tag = "~effect/persistence/PersistedQueue/DeadLetter"
424 readonly cause: Cause.Cause<unknown>
425 constructor(cause: Cause.Cause<unknown>) {
426 this.cause = cause
427 }
428}
429
430const isDeadLetter = (u: unknown): u is DeadLetter =>
431 Predicate.isTagged(u, "~effect/persistence/PersistedQueue/DeadLetter")
432
433const deadLetterFromCause = (cause: Cause.Cause<unknown>): DeadLetter | undefined => {
434 for (const reason of cause.reasons) {
435 if (Cause.isFailReason(reason) && isDeadLetter(reason.error)) {
436 return reason.error
437 }
438 }
439 return undefined
440}
441
442/**
443 * Defines the low-level backing store service used by `PersistedQueue`.
444 *
445 * **When to use**
446 *
447 * Use to provide the persistence backend that stores queued elements, scoped
448 * takes, retry attempts, and acknowledgements.
449 *
450 * **Details**
451 *
452 * The store persists offered elements and returns taken elements in a scope so
453 * the finalizer can complete or retry them based on the processing exit.
454 *
455 * Claiming an element counts an attempt, so the `attempts` returned by `take`
456 * is 1-based. When the take scope closes with a success the element is marked
457 * completed; a failure retries it according to `retryDelay` or marks it failed
458 * once `maxAttempts` is exhausted; an interruption releases it without
459 * counting the attempt.
460 *
461 * @stability unstable
462 * @category services
463 * @since 4.0.0
464 */
465export class PersistedQueueStore extends Context.Service<
466 PersistedQueueStore,
467 {
468 readonly offer: (
469 options: {
470 readonly name: string
471 readonly id: string
472 readonly element: unknown
473 readonly isCustomId: boolean
474 }
475 ) => Effect.Effect<void, PersistedQueueError>
476
477 readonly take: (options: {
478 readonly name: string
479 readonly maxAttempts: number
480 readonly retryDelay: (attempts: number) => Effect.Effect<Duration.Duration>
481 }) => Effect.Effect<
482 {
483 readonly id: string
484 readonly attempts: number
485 readonly element: unknown
486 },
487 PersistedQueueError,
488 Scope.Scope
489 >
490
491 /**
492 * Removes completed elements older than `timeToLive`, together with their
493 * de-duplication records. Failed elements are removed only when
494 * `failedTimeToLive` is provided.
495 */
496 readonly cleanup: (options: {
497 readonly timeToLive: Duration.Duration
498 readonly failedTimeToLive: Duration.Duration | undefined
499 }) => Effect.Effect<void, PersistedQueueError>
500 }
501>()("effect/persistence/PersistedQueue/PersistedQueueStore") {}
502
503/**
504 * Provides an in-memory `PersistedQueueStore`.
505 *
506 * **Details**
507 *
508 * The store is process-local and volatile; failed takes are requeued with the
509 * queue's retry schedule until the configured maximum attempts is reached,
510 * after which the element is marked as failed.
511 *
512 * @stability unstable
513 * @category layers
514 * @since 4.0.0
515 */
516export const layerStoreMemory: Layer.Layer<
517 PersistedQueueStore
518> = Layer.effect(
519 PersistedQueueStore,
520 Effect.gen(function*() {
521 const clock = yield* Clock.Clock
522 type Entry = {
523 readonly id: string
524 readonly element: unknown
525 attempts: number
526 state: "pending" | "processing" | "completed" | "failed"
527 visibleAt: number
528 stateChangedAt: number
529 }
530 const queues = new Map<string, {
531 latch: Latch.Latch
532 // all entries ever offered, for de-duplication; `pending` holds the
533 // deliverable subset so take does not scan the completed backlog
534 entries: Map<string, Entry>
535 pending: Set<Entry>
536 }>()
537 const getOrCreateQueue = (name: string) => {
538 let queue = queues.get(name)
539 if (!queue) {
540 queue = {
541 latch: Latch.makeUnsafe(false),
542 entries: new Map(),
543 pending: new Set()
544 }
545 queues.set(name, queue)
546 }
547 return queue
548 }
549
550 return PersistedQueueStore.of({
551 offer: (options) =>
552 Effect.sync(() => {
553 const now = clock.currentTimeMillisUnsafe()
554 const queue = getOrCreateQueue(options.name)
555 if (queue.entries.has(options.id)) return
556 const entry: Entry = {
557 id: options.id,
558 element: options.element,
559 attempts: 0,
560 state: "pending",
561 visibleAt: now,
562 stateChangedAt: now
563 }
564 queue.entries.set(options.id, entry)
565 queue.pending.add(entry)
566 queue.latch.openUnsafe()
567 }),
568 take: Effect.fnUntraced(function*(options) {
569 const queue = getOrCreateQueue(options.name)
570 while (true) {
571 // close before scanning so a mutation after the scan reopens the latch
572 queue.latch.closeUnsafe()
573 const now = clock.currentTimeMillisUnsafe()
574 let item: Entry | undefined
575 let nextVisibleAt = Infinity
576 for (const entry of queue.pending) {
577 if (entry.attempts >= options.maxAttempts) continue
578 if (entry.visibleAt <= now) {
579 item = entry
580 break
581 }
582 nextVisibleAt = Math.min(nextVisibleAt, entry.visibleAt)
583 }
584 if (item === undefined) {
585 yield* nextVisibleAt === Infinity
586 ? queue.latch.await
587 : Effect.race(queue.latch.await, Effect.sleep(Duration.millis(nextVisibleAt - now)))
588 continue
589 }
590 const entry = item
591 entry.state = "processing"
592 entry.attempts += 1
593 queue.pending.delete(entry)
594 queue.latch.openUnsafe()
595 yield* Effect.addFinalizer(
596 Effect.fnUntraced(function*(exit) {
597 const now = clock.currentTimeMillisUnsafe()
598 if (exit._tag === "Success") {
599 entry.state = "completed"
600 entry.stateChangedAt = now
601 return
602 }
603 const deadLetter = deadLetterFromCause(exit.cause)
604 if (deadLetter !== undefined) {
605 entry.state = "failed"
606 entry.stateChangedAt = now
607 return
608 }
609 if (Cause.hasInterruptsOnly(exit.cause)) {
610 entry.attempts -= 1
611 } else if (entry.attempts >= options.maxAttempts) {
612 entry.state = "failed"
613 entry.stateChangedAt = now
614 return
615 } else {
616 entry.visibleAt = now + Duration.toMillis(yield* options.retryDelay(entry.attempts))
617 }
618 entry.state = "pending"
619 queue.pending.add(entry)
620 queue.latch.openUnsafe()
621 })
622 )
623 return { id: entry.id, attempts: entry.attempts, element: entry.element }
624 }
625 }),
626 cleanup: (options) =>
627 Effect.sync(() => {
628 const now = clock.currentTimeMillisUnsafe()
629 const completedCutoff = now - Duration.toMillis(options.timeToLive)
630 const failedCutoff = options.failedTimeToLive === undefined
631 ? undefined
632 : now - Duration.toMillis(options.failedTimeToLive)
633 for (const queue of queues.values()) {
634 for (const [id, entry] of queue.entries) {
635 const cutoff = entry.state === "completed"
636 ? completedCutoff
637 : entry.state === "failed"
638 ? failedCutoff
639 : undefined
640 if (cutoff !== undefined && entry.stateChangedAt <= cutoff) {
641 queue.entries.delete(id)
642 }
643 }
644 }
645 })
646 })
647 })
648)
649
650/**
651 * Creates a Redis-backed `PersistedQueueStore`.
652 *
653 * **Details**
654 *
655 * The store uses Redis lists, hashes, and sorted sets with worker locks,
656 * periodically refreshes locks while items are being processed, delays retried
657 * items with the queue's retry schedule, and moves exhausted items to a failed
658 * queue.
659 *
660 * @stability unstable
661 * @category constructors
662 * @since 4.0.0
663 */
664export const makeStoreRedis = Effect.fnUntraced(function*(
665 options?: {
666 readonly prefix?: string | undefined
667 readonly pollInterval?: Duration.Input | undefined
668 readonly lockRefreshInterval?: Duration.Input | undefined
669 readonly lockExpiration?: Duration.Input | undefined
670 }
671) {
672 const redis = yield* Redis.Redis
673 const clock = yield* Clock.Clock
674
675 const pollInterval = Duration.max(
676 options?.pollInterval ? Duration.fromInputUnsafe(options.pollInterval) : Duration.seconds(1),
677 Duration.millis(1)
678 )
679 const lockRefreshMillis = Math.max(
680 options?.lockRefreshInterval
681 ? Duration.toMillis(Duration.fromInputUnsafe(options.lockRefreshInterval))
682 : 30_000,
683 1
684 )
685 const lockExpirationMillis = Math.max(
686 options?.lockExpiration
687 ? Duration.toMillis(Duration.fromInputUnsafe(options.lockExpiration))
688 : 90_000,
689 1
690 )
691 const prefix = options?.prefix ?? "effectq:"
692 const keyLock = (id: string) => `${prefix}${id}:lock`
693 const keysFor = (name: string) => ({
694 queue: `${prefix}${name}`,
695 pending: `${prefix}${name}:pending`,
696 failed: `${prefix}${name}:failed`,
697 delayed: `${prefix}${name}:delayed`,
698 attempts: `${prefix}${name}:attempts`,
699 ids: `${prefix}${name}:ids`
700 })
701 const workerId = crypto.randomUUID()
702
703 const ackRetrySchedule = makeAckRetrySchedule(lockExpirationMillis)
704
705 type Element = {
706 readonly id: string
707 readonly element: unknown
708 readonly attempts: number
709 // the raw wire payload, so requeue and retry do not re-stringify it
710 readonly payload: string
711 }
712
713 const requeue = redis.eval(requeueRedis)
714 const complete = redis.eval(completeRedis)
715 const failed = redis.eval(failedRedis)
716 const retry = redis.eval(retryRedis)
717 const resetQueue = redis.eval(resetQueueRedis)
718 const offer = redis.eval(offerRedis)
719 const take = redis.eval(takeRedis)
720 const expireAll = redis.eval(expireAllRedis)
721 const trimFailed = redis.eval(trimFailedRedis)
722
723 const queueStates = makeQueueStates()
724
725 const queues = yield* RcMap.make({
726 lookup: Effect.fnUntraced(function*(name: string) {
727 const keys = keysFor(name)
728 const queue = yield* Queue.make<Element>()
729 const takers = MutableRef.make(0)
730 const pollLatch = Latch.makeUnsafe()
731 const takenLatch = Latch.makeUnsafe()
732 const state = queueStates.get(name)
733
734 yield* Effect.addFinalizer(() =>
735 Effect.orDie(
736 Effect.flatMap(
737 Queue.clear(queue),
738 (elements) =>
739 Effect.forEach(elements, (element) =>
740 requeue(
741 keys.queue,
742 keys.pending,
743 keyLock(element.id),
744 keys.attempts,
745 element.id,
746 element.payload
747 ), { concurrency: "unbounded", discard: true })
748 )
749 )
750 )
751
752 yield* Effect.suspend(() =>
753 resetQueue(
754 keys.queue,
755 keys.pending,
756 keys.attempts,
757 keys.failed,
758 keys.ids,
759 prefix,
760 state.maxAttempts,
761 clock.currentTimeMillisUnsafe()
762 )
763 ).pipe(
764 Effect.andThen(Effect.sleep(lockRefreshMillis)),
765 Effect.forever,
766 Effect.forkScoped
767 )
768
769 const poll = (size: number) =>
770 Effect.suspend(() =>
771 take(
772 keys.queue,
773 keys.pending,
774 keys.delayed,
775 keys.attempts,
776 prefix,
777 workerId,
778 size,
779 lockExpirationMillis,
780 clock.currentTimeMillisUnsafe()
781 )
782 )
783
784 yield* Effect.gen(function*() {
785 while (true) {
786 yield* pollLatch.await
787 yield* Effect.yieldNow
788 state.nudge.closeUnsafe()
789 const results = takers.current === 0 ? null : yield* poll(takers.current)
790 if (results === null || results.length === 0) {
791 yield* Effect.race(Effect.sleep(pollInterval), state.nudge.await)
792 continue
793 }
794 takenLatch.closeUnsafe()
795 const elements: Array<Element> = []
796 for (let i = 0; i < results.length; i += 2) {
797 const payload = results[i] as string
798 const parsed = JSON.parse(payload)
799 elements.push({
800 id: parsed.id,
801 element: parsed.element,
802 attempts: Number(results[i + 1]),
803 payload
804 })
805 }
806 yield* Queue.offerAll(queue, elements)
807 yield* takenLatch.await
808 yield* Effect.yieldNow
809 }
810 }).pipe(
811 Effect.tapCause(Effect.logWarning),
812 Effect.sandbox,
813 Effect.retry(Schedule.spaced(500)),
814 Effect.forkScoped,
815 Effect.interruptible
816 )
817
818 return { queue, takers, pollLatch, takenLatch } as const
819 }),
820 idleTimeToLive: Duration.seconds(30)
821 })
822
823 const activeLockKeys = new Set<string>()
824
825 yield* Effect.gen(function*() {
826 while (true) {
827 yield* Effect.sleep(lockRefreshMillis)
828 yield* Effect.ignore(expireAll(Array.from(activeLockKeys), lockExpirationMillis))
829 }
830 }).pipe(
831 Effect.forkScoped,
832 Effect.interruptible,
833 Effect.annotateLogs({
834 module: "effect/persistence/PersistedQueue",
835 fiber: "refreshLocks"
836 })
837 )
838
839 const scanKeys = (pattern: string) =>
840 Effect.gen(function*() {
841 const keys: Array<string> = []
842 let cursor = "0"
843 do {
844 const [next, batch] = yield* redis.send<[string, Array<string>]>(
845 "SCAN",
846 cursor,
847 "MATCH",
848 pattern,
849 "COUNT",
850 "100"
851 )
852 cursor = next
853 for (const key of batch) keys.push(key)
854 } while (cursor !== "0")
855 return keys
856 })
857
858 return PersistedQueueStore.of({
859 offer: ({ element, id, isCustomId, name }) => {
860 const keys = keysFor(name)
861 const payload = JSON.stringify({ id, element })
862 return (isCustomId
863 ? offer(keys.queue, keys.ids, id, payload)
864 : redis.send("RPUSH", keys.queue, payload)).pipe(
865 Effect.mapError(({ cause }) =>
866 new PersistedQueueError({
867 message: "Failed to offer element to persisted queue",
868 cause
869 })
870 ),
871 Effect.tap(() =>
872 Effect.sync(() => {
873 queueStates.peek(name)?.nudge.openUnsafe()
874 })
875 )
876 )
877 },
878 take: (options) =>
879 Effect.uninterruptibleMask((restore) => {
880 queueStates.get(options.name).maxAttempts = options.maxAttempts
881 return RcMap.get(queues, options.name).pipe(
882 Effect.flatMap(({ pollLatch, queue, takenLatch, takers }) => {
883 takers.current++
884 if (takers.current === 1) {
885 pollLatch.openUnsafe()
886 }
887 // onExit so the decrement also runs when a waiting take is
888 // interrupted, otherwise the poller keeps fetching for a phantom
889 // taker
890 return Effect.onExit(restore(Queue.take(queue)), () =>
891 Effect.sync(() => {
892 takers.current--
893 if (takers.current === 0) {
894 pollLatch.closeUnsafe()
895 takenLatch.openUnsafe()
896 } else if (Queue.sizeUnsafe(queue) === 0) {
897 takenLatch.openUnsafe()
898 }
899 }))
900 }),
901 Effect.scoped,
902 Effect.tap((element) => {
903 const keys = keysFor(options.name)
904 const lock = keyLock(element.id)
905 activeLockKeys.add(lock)
906 const ack = (effect: Effect.Effect<unknown, Redis.RedisError>) =>
907 effect.pipe(
908 Effect.retry(ackRetrySchedule),
909 Effect.orDie,
910 Effect.ensuring(Effect.sync(() => activeLockKeys.delete(lock)))
911 )
912 return Effect.addFinalizer((exit) =>
913 Effect.suspend(() => {
914 const now = clock.currentTimeMillisUnsafe()
915 if (exit._tag === "Success") {
916 return ack(complete(keys.pending, lock, keys.attempts, keys.ids, element.id, now))
917 }
918 const dead = deadLetterFromCause(exit.cause)
919 const failElement = (cause: Cause.Cause<unknown>) =>
920 ack(failed(
921 keys.pending,
922 lock,
923 keys.failed,
924 keys.attempts,
925 keys.ids,
926 element.id,
927 JSON.stringify({
928 id: element.id,
929 element: element.element,
930 attempts: element.attempts,
931 lastFailure: Cause.pretty(cause),
932 failedAt: now
933 })
934 ))
935 if (dead !== undefined) {
936 return failElement(dead.cause)
937 }
938 if (Cause.hasInterruptsOnly(exit.cause)) {
939 return ack(requeue(keys.queue, keys.pending, lock, keys.attempts, element.id, element.payload))
940 }
941 if (element.attempts >= options.maxAttempts) {
942 return failElement(exit.cause)
943 }
944 return Effect.flatMap(options.retryDelay(element.attempts), (delay) =>
945 ack(retry(
946 keys.pending,
947 lock,
948 keys.delayed,
949 element.id,
950 element.payload,
951 now + Duration.toMillis(delay)
952 )))
953 })
954 )
955 })
956 )
957 }),
958 cleanup: ({ failedTimeToLive, timeToLive }) =>
959 Effect.gen(function*() {
960 const now = clock.currentTimeMillisUnsafe()
961 const idsKeys = yield* scanKeys(`${prefix}*:ids`)
962 const cutoff = now - Duration.toMillis(timeToLive)
963 yield* Effect.forEach(
964 idsKeys,
965 (key) => redis.send("ZREMRANGEBYSCORE", key, "-inf", `(${cutoff}`),
966 { concurrency: 16, discard: true }
967 )
968 if (failedTimeToLive !== undefined) {
969 const failedCutoff = now - Duration.toMillis(failedTimeToLive)
970 const failedKeys = yield* scanKeys(`${prefix}*:failed`)
971 yield* Effect.forEach(
972 failedKeys,
973 (key) =>
974 trimFailed(key, `${key.slice(0, -":failed".length)}:ids`, failedCutoff).pipe(
975 // each call trims at most one batch, so drain until done
976 Effect.repeat({ while: (removed) => removed >= trimFailedBatchSize })
977 ),
978 { concurrency: 16, discard: true }
979 )
980 }
981 }).pipe(
982 Effect.mapError(({ cause }) =>
983 new PersistedQueueError({
984 message: "Failed to clean up persisted queue",
985 cause
986 })
987 )
988 )
989 })
990})
991
992const offerRedis = Redis.script(
993 (...args: [keyQueue: string, keyIds: string, id: string, payload: string]) => args,
994 {
995 lua: `
996local key_queue = KEYS[1]
997local key_ids = KEYS[2]
998local id = ARGV[1]
999local payload = ARGV[2]
1000
1001-- park the dedupe entry outside the timeToLive trim range until the element
1002-- completes, so unprocessed elements never lose dedupe protection
1003local result = redis.call("ZADD", key_ids, "NX", "+inf", id)
1004if result == 1 then
1005 redis.call("RPUSH", key_queue, payload)
1006end
1007`,
1008 numberOfKeys: 2
1009 }
1010)
1011
1012const resetQueueRedis = Redis.script(
1013 (
1014 ...args: [
1015 keyQueue: string,
1016 keyPending: string,
1017 keyAttempts: string,
1018 keyFailed: string,
1019 keyIds: string,
1020 prefix: string,
1021 maxAttempts: number,
1022 now: number
1023 ]
1024 ) => args,
1025 {
1026 lua: `
1027local key_queue = KEYS[1]
1028local key_pending = KEYS[2]
1029local key_attempts = KEYS[3]
1030local key_failed = KEYS[4]
1031local key_ids = KEYS[5]
1032local prefix = ARGV[1]
1033local max_attempts = tonumber(ARGV[2])
1034local now = ARGV[3]
1035
1036local entries = redis.call("HGETALL", key_pending)
1037for i = 1, #entries, 2 do
1038 local id = entries[i]
1039 local payload = entries[i + 1]
1040 local lock_key = prefix .. id .. ":lock"
1041 local exists = redis.call("EXISTS", lock_key)
1042 if exists == 0 then
1043 local attempts = tonumber(redis.call("HGET", key_attempts, id) or "0")
1044 if attempts >= max_attempts then
1045 -- compose the failed record by hand so the element payload does not go
1046 -- through a lossy cjson round-trip
1047 local record = string.sub(payload, 1, -2) .. ',"attempts":' .. attempts ..
1048 ',"lastFailure":"Lock expired after final attempt","failedAt":' .. now .. '}'
1049 redis.call("RPUSH", key_failed, record)
1050 redis.call("HDEL", key_attempts, id)
1051 -- failed ids keep their dedupe entry until failedTimeToLive removes the
1052 -- dead-letter record, so park them outside the timeToLive trim range
1053 redis.call("ZADD", key_ids, "XX", "+inf", id)
1054 else
1055 redis.call("RPUSH", key_queue, payload)
1056 end
1057 redis.call("HDEL", key_pending, id)
1058 end
1059end
1060`,
1061 numberOfKeys: 5
1062 }
1063)
1064
1065const requeueRedis = Redis.script(
1066 (
1067 ...args: [keyQueue: string, keyPending: string, keyLock: string, keyAttempts: string, id: string, payload: string]
1068 ) => args,
1069 {
1070 lua: `
1071local key_queue = KEYS[1]
1072local key_pending = KEYS[2]
1073local key_lock = KEYS[3]
1074local key_attempts = KEYS[4]
1075local id = ARGV[1]
1076local payload = ARGV[2]
1077
1078redis.call("DEL", key_lock)
1079redis.call("HDEL", key_pending, id)
1080local attempts = redis.call("HINCRBY", key_attempts, id, -1)
1081if attempts <= 0 then
1082 redis.call("HDEL", key_attempts, id)
1083end
1084redis.call("RPUSH", key_queue, payload)
1085`,
1086 numberOfKeys: 4
1087 }
1088)
1089
1090const completeRedis = Redis.script(
1091 (...args: [keyPending: string, keyLock: string, keyAttempts: string, keyIds: string, id: string, now: number]) =>
1092 args,
1093 {
1094 lua: `
1095local key_pending = KEYS[1]
1096local key_lock = KEYS[2]
1097local key_attempts = KEYS[3]
1098local key_ids = KEYS[4]
1099local id = ARGV[1]
1100local now = ARGV[2]
1101
1102redis.call("DEL", key_lock)
1103redis.call("HDEL", key_pending, id)
1104redis.call("HDEL", key_attempts, id)
1105redis.call("ZADD", key_ids, "XX", now, id)
1106`,
1107 numberOfKeys: 4
1108 }
1109)
1110
1111const retryRedis = Redis.script(
1112 (
1113 ...args: [keyPending: string, keyLock: string, keyDelayed: string, id: string, payload: string, visibleAt: number]
1114 ) => args,
1115 {
1116 lua: `
1117local key_pending = KEYS[1]
1118local key_lock = KEYS[2]
1119local key_delayed = KEYS[3]
1120local id = ARGV[1]
1121local payload = ARGV[2]
1122local visible_at = ARGV[3]
1123
1124redis.call("DEL", key_lock)
1125redis.call("HDEL", key_pending, id)
1126redis.call("ZADD", key_delayed, visible_at, payload)
1127`,
1128 numberOfKeys: 3
1129 }
1130)
1131
1132const failedRedis = Redis.script(
1133 (
1134 ...args: [
1135 keyPending: string,
1136 keyLock: string,
1137 keyFailed: string,
1138 keyAttempts: string,
1139 keyIds: string,
1140 id: string,
1141 payload: string
1142 ]
1143 ) => args,
1144 {
1145 lua: `
1146local key_pending = KEYS[1]
1147local key_lock = KEYS[2]
1148local key_failed = KEYS[3]
1149local key_attempts = KEYS[4]
1150local key_ids = KEYS[5]
1151local id = ARGV[1]
1152local payload = ARGV[2]
1153
1154redis.call("DEL", key_lock)
1155redis.call("HDEL", key_pending, id)
1156redis.call("HDEL", key_attempts, id)
1157redis.call("RPUSH", key_failed, payload)
1158-- failed ids keep their dedupe entry until failedTimeToLive removes the
1159-- dead-letter record, so park them outside the timeToLive trim range
1160redis.call("ZADD", key_ids, "XX", "+inf", id)
1161`,
1162 numberOfKeys: 5
1163 }
1164)
1165
1166const takeRedis = Redis.script(
1167 (
1168 ...args: [
1169 keyQueue: string,
1170 keyPending: string,
1171 keyDelayed: string,
1172 keyAttempts: string,
1173 prefix: string,
1174 workerId: string,
1175 batchSize: number,
1176 pttl: number,
1177 now: number
1178 ]
1179 ) => args,
1180 {
1181 lua: `
1182local key_queue = KEYS[1]
1183local key_pending = KEYS[2]
1184local key_delayed = KEYS[3]
1185local key_attempts = KEYS[4]
1186local prefix = ARGV[1]
1187local worker_id = ARGV[2]
1188local batch_size = tonumber(ARGV[3])
1189local pttl = ARGV[4]
1190local now = ARGV[5]
1191
1192local due = redis.call("ZRANGEBYSCORE", key_delayed, "-inf", now, "LIMIT", 0, 100)
1193if #due > 0 then
1194 for i, payload in ipairs(due) do
1195 redis.call("RPUSH", key_queue, payload)
1196 end
1197 redis.call("ZREM", key_delayed, unpack(due))
1198end
1199
1200local payloads = redis.call("LPOP", key_queue, batch_size)
1201if not payloads then
1202 return nil
1203end
1204
1205local result = {}
1206for i, payload in ipairs(payloads) do
1207 local id = cjson.decode(payload).id
1208 local key_lock = prefix .. id .. ":lock"
1209 redis.call("SET", key_lock, worker_id, "PX", pttl)
1210 redis.call("HSET", key_pending, id, payload)
1211 local attempts = redis.call("HINCRBY", key_attempts, id, 1)
1212 result[i * 2 - 1] = payload
1213 result[i * 2] = attempts
1214end
1215
1216return result
1217`,
1218 numberOfKeys: 4
1219 }
1220).withReturnType<ReadonlyArray<string | number> | null>()
1221
1222const expireAllRedis = Redis.script(
1223 (keys: ReadonlyArray<string>, ttl: number) => [...keys, ttl],
1224 {
1225 numberOfKeys: (keys) => keys.length,
1226 lua: `
1227local ttl = ARGV[1]
1228for i, key in ipairs(KEYS) do
1229 redis.call("PEXPIRE", key, ttl)
1230end
1231`
1232 }
1233)
1234
1235const trimFailedBatchSize = 1000
1236
1237const trimFailedRedis = Redis.script(
1238 (...args: [keyFailed: string, keyIds: string, cutoff: number]) => args,
1239 {
1240 lua: `
1241local key_failed = KEYS[1]
1242local key_ids = KEYS[2]
1243local cutoff = tonumber(ARGV[1])
1244local removed = 0
1245
1246while removed < ${trimFailedBatchSize} do
1247 local head = redis.call("LINDEX", key_failed, 0)
1248 if not head then break end
1249 local ok, decoded = pcall(cjson.decode, head)
1250 if ok then
1251 local failed_at = tonumber(decoded.failedAt)
1252 if failed_at ~= nil and failed_at >= cutoff then break end
1253 redis.call("LPOP", key_failed)
1254 if decoded.id ~= nil then
1255 redis.call("ZREM", key_ids, decoded.id)
1256 end
1257 else
1258 -- a corrupt head would otherwise block trimming of the whole list
1259 redis.call("LPOP", key_failed)
1260 end
1261 removed = removed + 1
1262end
1263
1264return removed
1265`,
1266 numberOfKeys: 2
1267 }
1268).withReturnType<number>()
1269
1270/**
1271 * Provides a Redis-backed `PersistedQueueStore` using `makeStoreRedis`.
1272 *
1273 * @stability unstable
1274 * @category layers
1275 * @since 4.0.0
1276 */
1277export const layerStoreRedis: (
1278 options?: {
1279 readonly prefix?: string | undefined
1280 readonly pollInterval?: Duration.Input | undefined
1281 readonly lockRefreshInterval?: Duration.Input | undefined
1282 readonly lockExpiration?: Duration.Input | undefined
1283 } | undefined
1284) => Layer.Layer<
1285 PersistedQueueStore,
1286 never,
1287 Redis.Redis
1288> = flow(makeStoreRedis, Layer.effect(PersistedQueueStore))
1289
1290/**
1291 * Creates a SQL-backed `PersistedQueueStore`.
1292 *
1293 * **Details**
1294 *
1295 * The store creates the queue table and indexes, acquires rows with
1296 * per-worker locks, refreshes active locks while scoped takes are running, and
1297 * retries or completes rows according to the processing exit.
1298 *
1299 * @stability unstable
1300 * @category constructors
1301 * @since 4.0.0
1302 */
1303export const makeStoreSql: (
1304 options?: {
1305 readonly tableName?: string | undefined
1306 readonly pollInterval?: Duration.Input | undefined
1307 readonly lockRefreshInterval?: Duration.Input | undefined
1308 readonly lockExpiration?: Duration.Input | undefined
1309 } | undefined
1310) => Effect.Effect<
1311 PersistedQueueStore["Service"],
1312 SqlError,
1313 SqlClient.SqlClient | Scope.Scope
1314> = Effect.fnUntraced(function*(options) {
1315 const sql = (yield* SqlClient.SqlClient).withoutTransforms()
1316 const tableName = options?.tableName ?? "effect_queue"
1317 const tableNameSql = sql(tableName)
1318 const pollInterval = Duration.max(
1319 options?.pollInterval ? Duration.fromInputUnsafe(options.pollInterval) : Duration.millis(1000),
1320 Duration.millis(1)
1321 )
1322 const lockRefreshInterval = Duration.max(
1323 options?.lockRefreshInterval ? Duration.fromInputUnsafe(options.lockRefreshInterval) : Duration.seconds(30),
1324 Duration.millis(1)
1325 )
1326 const lockExpiration = Duration.max(
1327 options?.lockExpiration ? Duration.fromInputUnsafe(options.lockExpiration) : Duration.minutes(2),
1328 Duration.millis(1)
1329 )
1330 const workerId = crypto.randomUUID()
1331
1332 const ackRetrySchedule = makeAckRetrySchedule(lockExpiration)
1333
1334 yield* Effect.orDie(
1335 Migrator.make({})({
1336 loader: sqlMigrations(tableName),
1337 table: `${tableName}_migrations`
1338 })
1339 )
1340
1341 const sqlNow = sql.onDialectOrElse({
1342 // GETDATE() rounds to 1/300s and can land in the future, hiding freshly
1343 // written visible_at values from the poll query
1344 mssql: () => sql.literal("SYSDATETIME()"),
1345 mysql: () => sql.literal("NOW()"),
1346 pg: () => sql.literal("NOW()"),
1347 // sqlite
1348 orElse: () => sql.literal("CURRENT_TIMESTAMP")
1349 })
1350
1351 // `seconds` is a whole number, possibly negative
1352 const secondsOffset = (seconds: number) => {
1353 const s = sql.literal(seconds.toString())
1354 return sql.onDialectOrElse({
1355 pg: () => sql`${sqlNow} + INTERVAL '${s} seconds'`,
1356 mysql: () => sql`DATE_ADD(${sqlNow}, INTERVAL ${s} SECOND)`,
1357 mssql: () => sql`DATEADD(SECOND, ${s}, ${sqlNow})`,
1358 orElse: () => sql`datetime(${sqlNow}, '${s} seconds')`
1359 })
1360 }
1361 const secondsAgo = (seconds: number) => secondsOffset(-Math.max(Math.ceil(seconds), 0))
1362 const secondsFromNow = (seconds: number) => secondsOffset(Math.max(Math.ceil(seconds), 0))
1363 const expiresAt = secondsAgo(Duration.toSeconds(lockExpiration))
1364
1365 const offer = sql.onDialectOrElse({
1366 pg: () => (id: string, name: string, element: string) =>
1367 sql`
1368 INSERT INTO ${tableNameSql} (id, queue_name, element, state, attempts, visible_at, created_at, updated_at)
1369 VALUES (${id}, ${name}, ${element}, 'pending', 0, ${sqlNow}, ${sqlNow}, ${sqlNow})
1370 ON CONFLICT (id, queue_name) DO NOTHING
1371 `,
1372 mysql: () => (id: string, name: string, element: string) =>
1373 sql`
1374 INSERT IGNORE INTO ${tableNameSql} (id, queue_name, element, state, attempts, visible_at, created_at, updated_at)
1375 VALUES (${id}, ${name}, ${element}, 'pending', 0, ${sqlNow}, ${sqlNow}, ${sqlNow})
1376 `,
1377 mssql: () => (id: string, name: string, element: string) =>
1378 sql`
1379 MERGE ${tableNameSql} WITH (HOLDLOCK) AS target
1380 USING (SELECT ${id} AS id, ${name} AS queue_name) AS source
1381 ON target.id = source.id AND target.queue_name = source.queue_name
1382 WHEN NOT MATCHED THEN
1383 INSERT (id, queue_name, element, state, attempts, visible_at, created_at, updated_at)
1384 VALUES (source.id, source.queue_name, ${element}, 'pending', 0, ${sqlNow}, ${sqlNow}, ${sqlNow});
1385 `,
1386 // sqlite
1387 orElse: () => (id: string, name: string, element: string) =>
1388 sql`
1389 INSERT OR IGNORE INTO ${tableNameSql} (id, queue_name, element, state, attempts, visible_at, created_at, updated_at)
1390 VALUES (${id}, ${name}, ${element}, 'pending', 0, ${sqlNow}, ${sqlNow}, ${sqlNow})
1391 `
1392 })
1393
1394 const wrapString = sql.onDialectOrElse({
1395 mssql: () => (s: string) => `N'${s}'`,
1396 orElse: () => (s: string) => `'${s}'`
1397 })
1398 const stringLiteral = (s: string) => sql.literal(wrapString(s))
1399
1400 const workerIdSql = stringLiteral(workerId)
1401 const elementIds = new Set<number | string>()
1402 const refreshLocks: Effect.Effect<void, SqlError> = Effect.suspend((): Effect.Effect<void, SqlError> => {
1403 if (elementIds.size === 0) return Effect.void
1404 const ids = Array.from(elementIds)
1405 return sql`
1406 UPDATE ${tableNameSql}
1407 SET acquired_at = ${sqlNow}
1408 WHERE sequence IN (${sql.literal(ids.join(","))})
1409 AND acquired_by = ${workerIdSql}
1410 `
1411 })
1412 const ack = (
1413 statement: Effect.Effect<unknown, SqlError>,
1414 sequences: ReadonlyArray<number | string>
1415 ): Effect.Effect<void> =>
1416 statement.pipe(
1417 Effect.retry(ackRetrySchedule),
1418 Effect.orDie,
1419 Effect.ensuring(Effect.sync(() => {
1420 for (const sequence of sequences) {
1421 elementIds.delete(sequence)
1422 }
1423 })),
1424 Effect.asVoid
1425 )
1426 const complete = (sequence: number | string) =>
1427 ack(
1428 sql`
1429 UPDATE ${tableNameSql}
1430 SET acquired_at = NULL, acquired_by = NULL, updated_at = ${sqlNow}, state = 'completed'
1431 WHERE sequence = ${sequence}
1432 AND acquired_by = ${workerIdSql}
1433 `,
1434 [sequence]
1435 )
1436 const fail = (sequence: number | string, cause: Cause.Cause<any>) =>
1437 ack(
1438 sql`
1439 UPDATE ${tableNameSql}
1440 SET acquired_at = NULL, acquired_by = NULL, updated_at = ${sqlNow}, state = 'failed', last_failure = ${
1441 Cause.pretty(cause)
1442 }
1443 WHERE sequence = ${sequence}
1444 AND acquired_by = ${workerIdSql}
1445 `,
1446 [sequence]
1447 )
1448 const retry = (sequence: number | string, delay: Duration.Duration, cause: Cause.Cause<any>) =>
1449 ack(
1450 sql`
1451 UPDATE ${tableNameSql}
1452 SET acquired_at = NULL, acquired_by = NULL, updated_at = ${sqlNow}, visible_at = ${
1453 secondsFromNow(Duration.toSeconds(delay))
1454 }, last_failure = ${Cause.pretty(cause)}
1455 WHERE sequence = ${sequence}
1456 AND acquired_by = ${workerIdSql}
1457 `,
1458 [sequence]
1459 )
1460 const interrupt = (ids: Array<number | string>) =>
1461 ack(
1462 sql`
1463 UPDATE ${tableNameSql}
1464 SET acquired_at = NULL, acquired_by = NULL, attempts = attempts - 1
1465 WHERE sequence IN (${sql.literal(ids.join(","))})
1466 AND acquired_by = ${workerIdSql}
1467 `,
1468 ids
1469 )
1470
1471 yield* refreshLocks.pipe(
1472 Effect.tapCause(Effect.logWarning),
1473 Effect.retry(Schedule.spaced(500)),
1474 Effect.schedule(Schedule.fixed(lockRefreshInterval)),
1475 Effect.annotateLogs({
1476 package: "@effect/sql",
1477 module: "SqlPersistedQueue",
1478 fiber: "refreshLocks"
1479 }),
1480 Effect.forkScoped
1481 )
1482
1483 type Element = {
1484 readonly id: string
1485 readonly sequence: number | string
1486 readonly element: string
1487 attempts: number
1488 }
1489 const queueStates = makeQueueStates()
1490 const mailboxes = yield* RcMap.make({
1491 lookup: Effect.fnUntraced(function*(name: string) {
1492 const state = queueStates.get(name)
1493 const maxAttempts = state.maxAttempts
1494 const queue = yield* Queue.make<Element>()
1495 const takers = MutableRef.make(0)
1496 const pollLatch = Latch.makeUnsafe()
1497 const takenLatch = Latch.makeUnsafe()
1498 const nudge = state.nudge
1499
1500 yield* Effect.addFinalizer(() =>
1501 Effect.flatMap(Queue.clear(queue), (elements) => {
1502 if (elements.length === 0) return Effect.void
1503 return interrupt(Array.from(elements, (e) => e.sequence))
1504 })
1505 )
1506
1507 // flip exhausted rows whose lock expired (worker crashed on the final
1508 // attempt) to failed, since no finalizer will ever run for them
1509 yield* sql`
1510 UPDATE ${tableNameSql}
1511 SET state = 'failed', acquired_at = NULL, acquired_by = NULL, updated_at = ${sqlNow},
1512 last_failure = COALESCE(last_failure, 'Lock expired after final attempt')
1513 WHERE queue_name = ${name}
1514 AND state = 'pending'
1515 AND attempts >= ${maxAttempts}
1516 AND (acquired_at IS NULL OR acquired_at < ${expiresAt})
1517 `.pipe(
1518 Effect.tapCause(Effect.logWarning),
1519 Effect.ignore,
1520 Effect.schedule(Schedule.spaced(lockRefreshInterval)),
1521 Effect.forkScoped,
1522 Effect.interruptible
1523 )
1524
1525 const poll = sql.onDialectOrElse({
1526 pg: () => (size: number) =>
1527 sql<Element>`
1528 WITH cte AS (
1529 UPDATE ${tableNameSql}
1530 SET acquired_at = ${sqlNow}, acquired_by = ${workerIdSql}, attempts = attempts + 1
1531 WHERE sequence IN (
1532 SELECT sequence FROM ${tableNameSql}
1533 WHERE queue_name = ${name}
1534 AND state = 'pending'
1535 AND attempts < ${maxAttempts}
1536 AND visible_at <= ${sqlNow}
1537 AND (acquired_at IS NULL OR acquired_at < ${expiresAt})
1538 ORDER BY visible_at ASC, sequence ASC
1539 FOR UPDATE SKIP LOCKED
1540 LIMIT ${sql.literal(size.toString())}
1541 )
1542 RETURNING sequence, id, element, attempts, visible_at
1543 )
1544 SELECT sequence, id, element, attempts FROM cte
1545 ORDER BY visible_at ASC, sequence ASC
1546 `,
1547 mysql: () => (size: number) =>
1548 sql<Element>`
1549 SELECT sequence, id, element, attempts FROM ${tableNameSql} q
1550 WHERE queue_name = ${name}
1551 AND state = 'pending'
1552 AND attempts < ${maxAttempts}
1553 AND visible_at <= ${sqlNow}
1554 AND (acquired_at IS NULL OR acquired_at < ${expiresAt})
1555 ORDER BY visible_at ASC, sequence ASC
1556 LIMIT ${sql.literal(size.toString())}
1557 FOR UPDATE SKIP LOCKED
1558 `.pipe(
1559 Effect.tap((rows) => {
1560 if (rows.length === 0) return Effect.void
1561 return sql`
1562 UPDATE ${tableNameSql}
1563 SET acquired_at = ${sqlNow}, acquired_by = ${workerIdSql}, attempts = attempts + 1
1564 WHERE sequence IN (${sql.literal(rows.map((r) => r.sequence).join(","))})
1565 `.unprepared
1566 }),
1567 Effect.map((rows) => {
1568 for (const row of rows) {
1569 row.attempts = Number(row.attempts) + 1
1570 }
1571 return rows
1572 }),
1573 sql.withTransaction
1574 ),
1575 mssql: () => (size: number) =>
1576 sql<Element>`
1577 WITH cte AS (
1578 SELECT TOP ${sql.literal(size.toString())} sequence FROM ${tableNameSql}
1579 WHERE queue_name = ${name}
1580 AND state = 'pending'
1581 AND attempts < ${maxAttempts}
1582 AND visible_at <= ${sqlNow}
1583 AND (acquired_at IS NULL OR acquired_at < ${expiresAt})
1584 ORDER BY visible_at ASC, sequence ASC
1585 )
1586 UPDATE q
1587 SET acquired_at = ${sqlNow}, acquired_by = ${workerIdSql}, attempts = q.attempts + 1
1588 OUTPUT inserted.sequence, inserted.id, inserted.element, inserted.attempts
1589 FROM ${tableNameSql} AS q
1590 INNER JOIN cte ON q.sequence = cte.sequence
1591 `,
1592 // sqlite
1593 orElse: () => (size: number) =>
1594 sql<Element>`
1595 UPDATE ${tableNameSql}
1596 SET acquired_at = ${sqlNow}, acquired_by = ${workerIdSql}, attempts = attempts + 1
1597 WHERE sequence IN (
1598 SELECT sequence FROM ${tableNameSql}
1599 WHERE queue_name = ${name}
1600 AND state = 'pending'
1601 AND attempts < ${maxAttempts}
1602 AND visible_at <= ${sqlNow}
1603 AND (acquired_at IS NULL OR acquired_at < ${expiresAt})
1604 ORDER BY visible_at ASC, sequence ASC
1605 LIMIT ${sql.literal(size.toString())}
1606 )
1607 RETURNING sequence, id, element, attempts
1608 `
1609 })
1610
1611 yield* Effect.gen(function*() {
1612 while (true) {
1613 yield* pollLatch.await
1614 yield* Effect.yieldNow
1615 nudge.closeUnsafe()
1616 const results = takers.current === 0 ? [] : yield* poll(takers.current)
1617 if (results.length === 0) {
1618 yield* Effect.race(Effect.sleep(pollInterval), nudge.await)
1619 continue
1620 }
1621 takenLatch.closeUnsafe()
1622 for (const row of results) {
1623 row.attempts = Number(row.attempts)
1624 elementIds.add(row.sequence)
1625 }
1626 yield* Queue.offerAll(queue, results)
1627 yield* takenLatch.await
1628 yield* Effect.yieldNow
1629 }
1630 }).pipe(
1631 Effect.tapCause(Effect.logWarning),
1632 Effect.sandbox,
1633 Effect.retry(Schedule.spaced(500)),
1634 Effect.forkScoped
1635 )
1636
1637 return { queue, takers, pollLatch, takenLatch } as const
1638 }),
1639 idleTimeToLive: Duration.seconds(30)
1640 })
1641
1642 const cleanupBatch = sql.onDialectOrElse({
1643 pg: () => (state: string, seconds: number) =>
1644 sql<{ readonly count: number }>`
1645 WITH deleted_entries AS (
1646 DELETE FROM ${tableNameSql}
1647 WHERE sequence IN (
1648 SELECT sequence FROM ${tableNameSql}
1649 WHERE state = ${state} AND updated_at <= ${secondsAgo(seconds)}
1650 LIMIT ${sql.literal(String(sqlCleanupBatchSize))}
1651 )
1652 RETURNING 1
1653 )
1654 SELECT COUNT(*)::INT AS count FROM deleted_entries
1655 `.pipe(Effect.map((rows) => rows[0].count)),
1656 mysql: () =>
1657 Effect.fnUntraced(
1658 function*(state: string, seconds: number) {
1659 const connection = yield* sql.reserve
1660 const [statement, parameters] = sql`
1661 DELETE FROM ${tableNameSql}
1662 WHERE state = ${state} AND updated_at <= ${secondsAgo(seconds)}
1663 LIMIT ${sql.literal(String(sqlCleanupBatchSize))}
1664 `.compile()
1665 yield* connection.execute(statement, parameters, undefined)
1666 const rows = yield* connection.executeValues("SELECT ROW_COUNT()", [])
1667 return Number(rows[0][0])
1668 },
1669 Effect.scoped
1670 ),
1671 mssql: () => (state: string, seconds: number) =>
1672 sql<{ readonly sequence: number }>`
1673 DELETE TOP (${sql.literal(String(sqlCleanupBatchSize))}) FROM ${tableNameSql}
1674 OUTPUT DELETED.sequence
1675 WHERE state = ${state} AND updated_at <= ${secondsAgo(seconds)}
1676 `.pipe(Effect.map((rows) => rows.length)),
1677 // sqlite
1678 orElse: () => (state: string, seconds: number) =>
1679 sql<{ readonly deleted: number }>`
1680 DELETE FROM ${tableNameSql}
1681 WHERE sequence IN (
1682 SELECT sequence FROM ${tableNameSql}
1683 WHERE state = ${state} AND updated_at <= ${secondsAgo(seconds)}
1684 LIMIT ${sql.literal(String(sqlCleanupBatchSize))}
1685 )
1686 RETURNING 1 AS deleted
1687 `.pipe(Effect.map((rows) => rows.length))
1688 })
1689
1690 const cleanupState = (state: string, timeToLive: Duration.Duration) =>
1691 cleanupBatch(state, Duration.toSeconds(timeToLive)).pipe(
1692 Effect.repeat({
1693 while: (deletedCount) => deletedCount === sqlCleanupBatchSize,
1694 schedule: Schedule.spaced(Duration.millis(10))
1695 })
1696 )
1697
1698 return PersistedQueueStore.of({
1699 offer: ({ element, id, name }) =>
1700 Effect.catchCause(Effect.suspend(() => offer(id, name, JSON.stringify(element))), (cause) =>
1701 Effect.fail(
1702 new PersistedQueueError({
1703 message: "Failed to offer element to persisted queue",
1704 cause
1705 })
1706 )).pipe(
1707 Effect.tap(() =>
1708 Effect.sync(() => {
1709 queueStates.peek(name)?.nudge.openUnsafe()
1710 })
1711 )
1712 ),
1713 take: (options) => {
1714 queueStates.get(options.name).maxAttempts = options.maxAttempts
1715 const loop: Effect.Effect<
1716 {
1717 readonly id: string
1718 readonly attempts: number
1719 readonly element: unknown
1720 },
1721 PersistedQueueError,
1722 Scope.Scope
1723 > = Effect.uninterruptibleMask((restore) =>
1724 RcMap.get(mailboxes, options.name).pipe(
1725 Effect.flatMap(({ pollLatch, queue, takenLatch, takers }) => {
1726 takers.current++
1727 if (takers.current === 1) {
1728 pollLatch.openUnsafe()
1729 }
1730 // onExit so the decrement also runs when a waiting take is
1731 // interrupted, otherwise the poller keeps fetching for a phantom
1732 // taker
1733 return Effect.onExit(restore(Queue.take(queue)), () =>
1734 Effect.sync(() => {
1735 takers.current--
1736 if (takers.current === 0) {
1737 pollLatch.closeUnsafe()
1738 takenLatch.openUnsafe()
1739 } else if (Queue.sizeUnsafe(queue) === 0) {
1740 takenLatch.openUnsafe()
1741 }
1742 }))
1743 }),
1744 Effect.scoped,
1745 restore,
1746 Effect.flatMap((element) => {
1747 let parsed: unknown
1748 try {
1749 parsed = JSON.parse(element.element)
1750 } catch (defect) {
1751 // a row that cannot be parsed will never succeed, so dead-letter
1752 // it and take the next element
1753 return Effect.andThen(fail(element.sequence, Cause.die(defect)), loop)
1754 }
1755 return Effect.as(
1756 Effect.addFinalizer((exit) => {
1757 if (exit._tag === "Success") {
1758 return complete(element.sequence)
1759 }
1760 const dead = deadLetterFromCause(exit.cause)
1761 if (dead !== undefined) {
1762 return fail(element.sequence, dead.cause)
1763 }
1764 if (Cause.hasInterruptsOnly(exit.cause)) {
1765 return interrupt([element.sequence])
1766 }
1767 if (element.attempts >= options.maxAttempts) {
1768 return fail(element.sequence, exit.cause)
1769 }
1770 return Effect.flatMap(
1771 options.retryDelay(element.attempts),
1772 (delay) => retry(element.sequence, delay, exit.cause)
1773 )
1774 }),
1775 { id: element.id, attempts: element.attempts, element: parsed }
1776 )
1777 })
1778 )
1779 )
1780 return loop
1781 },
1782 cleanup: ({ failedTimeToLive, timeToLive }) =>
1783 Effect.gen(function*() {
1784 yield* cleanupState("completed", timeToLive)
1785 if (failedTimeToLive !== undefined) {
1786 yield* cleanupState("failed", failedTimeToLive)
1787 }
1788 }).pipe(
1789 Effect.catchCause((cause) =>
1790 Effect.fail(
1791 new PersistedQueueError({
1792 message: "Failed to clean up persisted queue",
1793 cause
1794 })
1795 )
1796 )
1797 )
1798 })
1799})
1800
1801const sqlMigrations = (tableName: string) =>
1802 Migrator.fromRecord({
1803 "0001_create_table": Effect.gen(function*() {
1804 const sql = (yield* SqlClient.SqlClient).withoutTransforms()
1805 const tableNameSql = sql(tableName)
1806
1807 yield* sql.onDialectOrElse({
1808 mysql: () =>
1809 sql`CREATE TABLE IF NOT EXISTS ${tableNameSql} (
1810 sequence BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
1811 id VARCHAR(36) NOT NULL,
1812 queue_name VARCHAR(100) NOT NULL,
1813 element TEXT NOT NULL,
1814 completed BOOLEAN NOT NULL,
1815 attempts INT NOT NULL DEFAULT 0,
1816 last_failure TEXT NULL,
1817 acquired_at DATETIME NULL,
1818 acquired_by VARCHAR(36) NULL,
1819 created_at DATETIME NOT NULL,
1820 updated_at DATETIME NOT NULL
1821 )`,
1822 pg: () =>
1823 sql`CREATE TABLE IF NOT EXISTS ${tableNameSql} (
1824 sequence SERIAL PRIMARY KEY,
1825 id VARCHAR(36) NOT NULL,
1826 queue_name VARCHAR(100) NOT NULL,
1827 element TEXT NOT NULL,
1828 completed BOOLEAN NOT NULL,
1829 attempts INTEGER NOT NULL DEFAULT 0,
1830 last_failure TEXT NULL,
1831 acquired_at TIMESTAMP NULL,
1832 acquired_by UUID NULL,
1833 created_at TIMESTAMP NOT NULL,
1834 updated_at TIMESTAMP NOT NULL
1835 )`,
1836 mssql: () =>
1837 sql`IF NOT EXISTS (SELECT * FROM sysobjects WHERE name=${tableName} AND xtype='U')
1838 CREATE TABLE ${tableNameSql} (
1839 sequence INT IDENTITY(1,1) PRIMARY KEY,
1840 id NVARCHAR(36) NOT NULL,
1841 queue_name NVARCHAR(100) NOT NULL,
1842 element NVARCHAR(MAX) NOT NULL,
1843 completed BIT NOT NULL,
1844 attempts INT NOT NULL DEFAULT 0,
1845 last_failure NVARCHAR(MAX) NULL,
1846 acquired_at DATETIME2 NULL,
1847 acquired_by UNIQUEIDENTIFIER NULL,
1848 created_at DATETIME2 NOT NULL,
1849 updated_at DATETIME2 NOT NULL
1850 )`,
1851 // sqlite
1852 orElse: () =>
1853 sql`CREATE TABLE IF NOT EXISTS ${tableNameSql} (
1854 sequence INTEGER PRIMARY KEY AUTOINCREMENT,
1855 id TEXT NOT NULL,
1856 queue_name TEXT NOT NULL,
1857 element TEXT NOT NULL,
1858 completed BOOLEAN NOT NULL,
1859 attempts INTEGER NOT NULL DEFAULT 0,
1860 last_failure TEXT NULL,
1861 acquired_at DATETIME NULL,
1862 acquired_by TEXT NULL,
1863 created_at DATETIME NOT NULL,
1864 updated_at DATETIME NOT NULL
1865 )`
1866 })
1867
1868 yield* sql.onDialectOrElse({
1869 mssql: () =>
1870 sql`IF NOT EXISTS (SELECT * FROM sys.indexes WHERE name = N'idx_${tableName}_id')
1871 CREATE UNIQUE INDEX ${sql(`idx_${tableName}_id`)} ON ${tableNameSql} (id, queue_name)`,
1872 mysql: () =>
1873 sql`CREATE UNIQUE INDEX ${sql(`idx_${tableName}_id`)} ON ${tableNameSql} (id, queue_name)`.pipe(
1874 Effect.ignore
1875 ),
1876 orElse: () =>
1877 sql`CREATE UNIQUE INDEX IF NOT EXISTS ${sql(`idx_${tableName}_id`)} ON ${tableNameSql} (id, queue_name)`
1878 })
1879
1880 yield* sql.onDialectOrElse({
1881 mssql: () =>
1882 sql`IF NOT EXISTS (SELECT * FROM sys.indexes WHERE name = N'idx_${tableName}_take')
1883 CREATE INDEX ${
1884 sql(`idx_${tableName}_take`)
1885 } ON ${tableNameSql} (queue_name, completed, attempts, acquired_at)`,
1886 mysql: () =>
1887 sql`CREATE INDEX ${
1888 sql(`idx_${tableName}_take`)
1889 } ON ${tableNameSql} (queue_name, completed, attempts, acquired_at)`
1890 .pipe(Effect.ignore),
1891 orElse: () =>
1892 sql`CREATE INDEX IF NOT EXISTS ${
1893 sql(`idx_${tableName}_take`)
1894 } ON ${tableNameSql} (queue_name, completed, attempts, acquired_at)`
1895 })
1896
1897 yield* sql.onDialectOrElse({
1898 mssql: () =>
1899 sql`IF NOT EXISTS (SELECT * FROM sys.indexes WHERE name = N'idx_${tableName}_update')
1900 CREATE INDEX ${sql(`idx_${tableName}_update`)} ON ${tableNameSql} (sequence, acquired_by)`,
1901 mysql: () =>
1902 sql`CREATE INDEX ${sql(`idx_${tableName}_update`)} ON ${tableNameSql} (sequence, acquired_by)`.pipe(
1903 Effect.ignore
1904 ),
1905 orElse: () =>
1906 sql`CREATE INDEX IF NOT EXISTS ${sql(`idx_${tableName}_update`)} ON ${tableNameSql} (sequence, acquired_by)`
1907 })
1908 }),
1909 "0002_upgrade_schema": Effect.gen(function*() {
1910 const sql = (yield* SqlClient.SqlClient).withoutTransforms()
1911 const tableNameSql = sql(tableName)
1912 const takeIndex = sql(`idx_${tableName}_take`)
1913
1914 yield* sql.onDialectOrElse({
1915 pg: () =>
1916 Effect.gen(function*() {
1917 yield* sql`DROP INDEX IF EXISTS ${takeIndex}`
1918 yield* sql`ALTER TABLE ${tableNameSql}
1919 ADD COLUMN state VARCHAR(10),
1920 ADD COLUMN visible_at TIMESTAMP`
1921 yield* sql`UPDATE ${tableNameSql}
1922 SET state = CASE WHEN completed THEN 'completed' ELSE 'pending' END,
1923 visible_at = updated_at`
1924 yield* sql`ALTER TABLE ${tableNameSql}
1925 ALTER COLUMN sequence TYPE BIGINT,
1926 ALTER COLUMN id TYPE VARCHAR(255),
1927 ALTER COLUMN queue_name TYPE VARCHAR(255),
1928 ALTER COLUMN state SET NOT NULL,
1929 ALTER COLUMN visible_at SET NOT NULL,
1930 DROP COLUMN completed`
1931 }),
1932 mysql: () =>
1933 Effect.gen(function*() {
1934 yield* sql`DROP INDEX ${takeIndex} ON ${tableNameSql}`
1935 yield* sql`ALTER TABLE ${tableNameSql}
1936 MODIFY COLUMN id VARCHAR(255) NOT NULL,
1937 MODIFY COLUMN queue_name VARCHAR(255) NOT NULL,
1938 MODIFY COLUMN element MEDIUMTEXT NOT NULL,
1939 MODIFY COLUMN last_failure MEDIUMTEXT NULL,
1940 ADD COLUMN state VARCHAR(10) NULL,
1941 ADD COLUMN visible_at DATETIME NULL`
1942 yield* sql`UPDATE ${tableNameSql}
1943 SET state = CASE WHEN completed THEN 'completed' ELSE 'pending' END,
1944 visible_at = updated_at`
1945 yield* sql`ALTER TABLE ${tableNameSql}
1946 MODIFY COLUMN state VARCHAR(10) NOT NULL,
1947 MODIFY COLUMN visible_at DATETIME NOT NULL,
1948 DROP COLUMN completed`
1949 }),
1950 mssql: () =>
1951 Effect.gen(function*() {
1952 const upgradedTableName = `${tableName}_upgrade`
1953 const upgradedTable = sql(upgradedTableName)
1954 yield* sql`CREATE TABLE ${upgradedTable} (
1955 sequence BIGINT IDENTITY(1,1) PRIMARY KEY,
1956 id NVARCHAR(255) NOT NULL,
1957 queue_name NVARCHAR(255) NOT NULL,
1958 element NVARCHAR(MAX) NOT NULL,
1959 state NVARCHAR(10) NOT NULL,
1960 attempts INT NOT NULL DEFAULT 0,
1961 last_failure NVARCHAR(MAX) NULL,
1962 visible_at DATETIME2 NOT NULL,
1963 acquired_at DATETIME2 NULL,
1964 acquired_by UNIQUEIDENTIFIER NULL,
1965 created_at DATETIME2 NOT NULL,
1966 updated_at DATETIME2 NOT NULL
1967 )`
1968 yield* sql`SET IDENTITY_INSERT ${upgradedTable} ON;
1969 INSERT INTO ${upgradedTable}
1970 (sequence, id, queue_name, element, state, attempts, last_failure, visible_at,
1971 acquired_at, acquired_by, created_at, updated_at)
1972 SELECT sequence, id, queue_name, element,
1973 CASE WHEN completed = 1 THEN 'completed' ELSE 'pending' END,
1974 attempts, last_failure, updated_at, acquired_at, acquired_by, created_at, updated_at
1975 FROM ${tableNameSql};
1976 SET IDENTITY_INSERT ${upgradedTable} OFF`
1977 yield* sql`DROP TABLE ${tableNameSql}`
1978 yield* sql`EXEC sp_rename ${upgradedTableName}, ${tableName}`
1979 yield* sql`CREATE UNIQUE INDEX ${sql(`idx_${tableName}_id`)} ON ${tableNameSql} (id, queue_name)`
1980 yield* sql`CREATE INDEX ${sql(`idx_${tableName}_update`)} ON ${tableNameSql} (sequence, acquired_by)`
1981 }),
1982 // sqlite rebuilds the table because altering or dropping constrained
1983 // columns is not portable across supported SQLite versions.
1984 orElse: () =>
1985 Effect.gen(function*() {
1986 const upgradedTableName = `${tableName}_upgrade`
1987 const upgradedTable = sql(upgradedTableName)
1988 yield* sql`DROP INDEX IF EXISTS ${takeIndex}`
1989 yield* sql`CREATE TABLE ${upgradedTable} (
1990 sequence INTEGER PRIMARY KEY AUTOINCREMENT,
1991 id TEXT NOT NULL,
1992 queue_name TEXT NOT NULL,
1993 element TEXT NOT NULL,
1994 state TEXT NOT NULL,
1995 attempts INTEGER NOT NULL DEFAULT 0,
1996 last_failure TEXT NULL,
1997 visible_at DATETIME NOT NULL,
1998 acquired_at DATETIME NULL,
1999 acquired_by TEXT NULL,
2000 created_at DATETIME NOT NULL,
2001 updated_at DATETIME NOT NULL
2002 )`
2003 yield* sql`INSERT INTO ${upgradedTable}
2004 (sequence, id, queue_name, element, state, attempts, last_failure, visible_at,
2005 acquired_at, acquired_by, created_at, updated_at)
2006 SELECT sequence, id, queue_name, element,
2007 CASE WHEN completed THEN 'completed' ELSE 'pending' END,
2008 attempts, last_failure, updated_at, acquired_at, acquired_by, created_at, updated_at
2009 FROM ${tableNameSql}`
2010 yield* sql`DROP TABLE ${tableNameSql}`
2011 yield* sql`ALTER TABLE ${upgradedTable} RENAME TO ${tableNameSql}`
2012 yield* sql`CREATE UNIQUE INDEX ${sql(`idx_${tableName}_id`)} ON ${tableNameSql} (id, queue_name)`
2013 yield* sql`CREATE INDEX ${sql(`idx_${tableName}_update`)} ON ${tableNameSql} (sequence, acquired_by)`
2014 })
2015 })
2016
2017 // partial index where supported, so pollers never scan completed rows
2018 yield* sql.onDialectOrElse({
2019 mssql: () =>
2020 sql`CREATE INDEX ${takeIndex} ON ${tableNameSql} (queue_name, visible_at)
2021 WHERE state = 'pending'`,
2022 mysql: () => sql`CREATE INDEX ${takeIndex} ON ${tableNameSql} (queue_name, state, visible_at)`,
2023 pg: () =>
2024 sql`CREATE INDEX ${takeIndex} ON ${tableNameSql} (queue_name, visible_at)
2025 WHERE state = 'pending'`,
2026 orElse: () =>
2027 sql`CREATE INDEX ${takeIndex} ON ${tableNameSql} (queue_name, visible_at)
2028 WHERE state = 'pending'`
2029 })
2030 })
2031 })
2032
2033/**
2034 * Provides a SQL-backed `PersistedQueueStore` using `makeStoreSql`.
2035 *
2036 * @stability unstable
2037 * @category layers
2038 * @since 4.0.0
2039 */
2040export const layerStoreSql: (
2041 options?: {
2042 readonly tableName?: string | undefined
2043 readonly pollInterval?: Duration.Input | undefined
2044 readonly lockRefreshInterval?: Duration.Input | undefined
2045 readonly lockExpiration?: Duration.Input | undefined
2046 } | undefined
2047) => Layer.Layer<
2048 PersistedQueueStore,
2049 SqlError,
2050 SqlClient.SqlClient
2051> = flow(makeStoreSql, Layer.effect(PersistedQueueStore))