1import { assert, it } from "@effect/vitest"
2import type { Vitest } from "@effect/vitest"
3import { Duration, Effect, Fiber, Latch, Layer, Schedule, Schema } from "effect"
4import { PersistedQueue } from "effect/persistence"
5import { TestClock } from "effect/testing"
6
7const Item = Schema.Struct({
8 n: Schema.BigInt
9})
10
11// advance the virtual clock one poll cycle, give real-time stores a moment,
12// then check the forked take is still waiting
13const assertNotDelivered = <A, E>(fiber: Fiber.Fiber<A, E>) =>
14 Effect.gen(function*() {
15 yield* TestClock.adjust(1000)
16 yield* Effect.sleep(1000).pipe(TestClock.withLive)
17 assert.isUndefined(fiber.pollUnsafe())
18 })
19
20// move both the virtual clock and real time past a 1 second ttl. The virtual
21// jump is kept small: a large jump can fire the SQL clients' pool timers and
22// time out in-flight connection acquisitions
23const advancePastTtl = Effect.gen(function*() {
24 yield* TestClock.adjust("2 seconds")
25 yield* Effect.sleep(1500).pipe(TestClock.withLive)
26})
27
28export const suiteWith = <R>(
29 name: string,
30 layer: Layer.Layer<PersistedQueue.PersistedQueueStore, unknown, R>,
31 testApi: Vitest.MethodsNonLive<R>,
32 timeout: Duration.Input = "30 seconds",
33 hookTimeout: Duration.Input = timeout
34) => {
35 // Tests share and advance the same TestClock.
36 const testOptions = { timeout: Duration.toMillis(timeout) }
37 return testApi.layer(
38 PersistedQueue.layer.pipe(
39 Layer.provideMerge(layer)
40 ),
41 { timeout: hookTimeout, concurrent: false }
42 )(`PersistedQueue (${name})`, (it) => {
43 it.effect("offer + take", () =>
44 Effect.gen(function*() {
45 const queue = yield* PersistedQueue.make({
46 name: "test-queue-a",
47 schema: Item
48 })
49
50 yield* queue.offer({ n: 42n })
51 yield* queue.take(Effect.fnUntraced(function*(value, metadata) {
52 assert.strictEqual(value.n, 42n)
53 assert.strictEqual(metadata.attempts, 1)
54 }))
55 }), testOptions)
56
57 it.effect("interrupt", () =>
58 Effect.gen(function*() {
59 const queue = yield* PersistedQueue.make({
60 name: "test-queue-b",
61 schema: Item
62 })
63
64 yield* queue.offer({ n: 42n })
65
66 const latch = Latch.makeUnsafe()
67 const fiber = yield* queue.take(Effect.fnUntraced(function*(_value) {
68 yield* latch.open
69 return yield* Effect.never
70 })).pipe(Effect.forkScoped)
71
72 const fiber2 = yield* queue.take((val) => Effect.succeed(val)).pipe(Effect.forkScoped)
73
74 yield* latch.await
75
76 // the second take really waits while the element is being processed
77 yield* assertNotDelivered(fiber2)
78
79 yield* Fiber.interrupt(fiber)
80
81 yield* TestClock.adjust(1000)
82
83 assert.strictEqual((yield* Fiber.join(fiber2)).n, 42n)
84 }), testOptions)
85
86 it.effect("failure", () =>
87 Effect.gen(function*() {
88 const queue = yield* PersistedQueue.make({
89 name: "test-queue-c",
90 schema: Item,
91 retrySchedule: Schedule.spaced(0)
92 })
93
94 yield* queue.offer({ n: 42n })
95
96 const error = yield* queue.take(() => Effect.fail("boom")).pipe(Effect.flip)
97 assert.strictEqual(error, "boom")
98
99 const value = yield* queue.take((val, { attempts }) => {
100 assert.strictEqual(attempts, 2)
101 return Effect.succeed(val)
102 })
103 assert.strictEqual(value.n, 42n)
104 }), testOptions)
105
106 it.effect("delays retries with the retry schedule", () =>
107 Effect.gen(function*() {
108 const queue = yield* PersistedQueue.make({
109 name: "test-queue-retry-schedule",
110 schema: Item,
111 // MySQL and SQLite truncate retry timestamps to whole seconds.
112 // Leave room for that truncation before the real-time assertion below.
113 retrySchedule: Schedule.spaced("2 seconds")
114 })
115
116 yield* queue.offer({ n: 42n })
117
118 const error = yield* queue.take(() => Effect.fail("boom")).pipe(Effect.flip)
119 assert.strictEqual(error, "boom")
120
121 const fiber = yield* queue.take((_val, { attempts }) => Effect.succeed(attempts)).pipe(
122 Effect.forkScoped
123 )
124
125 // not redelivered before the retry delay elapses
126 yield* TestClock.adjust(100)
127 yield* Effect.sleep(100).pipe(TestClock.withLive)
128 assert.isUndefined(fiber.pollUnsafe())
129
130 // Keep advancing while the take is pending: an SQL poll can still be
131 // in flight during an adjust and miss that virtual-clock wakeup.
132 for (let i = 0; i < 8 && fiber.pollUnsafe() === undefined; i++) {
133 yield* TestClock.adjust(1000)
134 yield* Effect.sleep(700).pipe(TestClock.withLive)
135 }
136 assert.isDefined(fiber.pollUnsafe())
137 assert.strictEqual(yield* Fiber.join(fiber), 2)
138 }), testOptions)
139
140 it.effect("idempotent offer", () =>
141 Effect.gen(function*() {
142 const queue = yield* PersistedQueue.make({
143 name: "idempotent-offer",
144 schema: Item
145 })
146
147 yield* queue.offer({ n: 42n }, { id: "custom-id" })
148 yield* queue.offer({ n: 42n }, { id: "custom-id" })
149 yield* queue.take(Effect.fnUntraced(function*(value) {
150 assert.strictEqual(value.n, 42n)
151 }))
152 const fiber = yield* queue.take(Effect.fnUntraced(function*(value) {
153 assert.strictEqual(value.n, 42n)
154 })).pipe(Effect.forkScoped)
155
156 yield* assertNotDelivered(fiber)
157 }), testOptions)
158
159 it.effect("deduplicates custom ids independently in each queue", () =>
160 Effect.gen(function*() {
161 const first = yield* PersistedQueue.make({ name: "custom-id-first", schema: Item })
162 const second = yield* PersistedQueue.make({ name: "custom-id-second", schema: Item })
163
164 yield* first.offer({ n: 1n }, { id: "shared-custom-id" })
165 yield* second.offer({ n: 2n }, { id: "shared-custom-id" })
166
167 const fiber = yield* second.take(Effect.succeed).pipe(Effect.forkScoped)
168 yield* TestClock.adjust(1000)
169 yield* Effect.sleep(1000).pipe(TestClock.withLive)
170
171 assert.isDefined(fiber.pollUnsafe())
172 assert.deepStrictEqual(yield* Fiber.join(fiber), { n: 2n })
173 }), testOptions)
174
175 it.effect("does not redeliver in-flight elements", () =>
176 Effect.gen(function*() {
177 const queue = yield* PersistedQueue.make({
178 name: "test-queue-reset",
179 schema: Item
180 })
181
182 yield* queue.offer({ n: 42n })
183
184 const taken = Latch.makeUnsafe()
185 const release = Latch.makeUnsafe()
186 const fiber = yield* queue.take(() => Effect.andThen(taken.open, release.await)).pipe(Effect.forkScoped)
187 yield* taken.await
188
189 const fiber2 = yield* queue.take((val) => Effect.succeed(val)).pipe(Effect.forkScoped)
190
191 // allow any periodic reset in the store to run while the element is
192 // still being processed
193 yield* assertNotDelivered(fiber2)
194
195 // after a successful take the element should not be delivered again
196 yield* release.open
197 yield* Fiber.join(fiber)
198 yield* assertNotDelivered(fiber2)
199
200 yield* Fiber.interrupt(fiber2)
201 }), testOptions)
202
203 it.effect("stops delivering when maxAttempts is exhausted", () =>
204 Effect.gen(function*() {
205 const queue = yield* PersistedQueue.make({
206 name: "test-queue-exhausted",
207 schema: Item,
208 maxAttempts: 1,
209 retrySchedule: Schedule.spaced(0)
210 })
211
212 yield* queue.offer({ n: 42n })
213
214 const error = yield* queue.take(() => Effect.fail("boom")).pipe(Effect.flip)
215 assert.strictEqual(error, "boom")
216
217 const fiber = yield* queue.take((val) => Effect.succeed(val)).pipe(Effect.forkScoped)
218
219 yield* assertNotDelivered(fiber)
220 }), testOptions)
221
222 it.effect("dead-letters elements that fail to decode", () =>
223 Effect.gen(function*() {
224 const store = yield* PersistedQueue.PersistedQueueStore
225 const queue = yield* PersistedQueue.make({
226 name: "test-queue-decode-failure",
227 schema: Item,
228 retrySchedule: Schedule.spaced(0)
229 })
230
231 yield* store.offer({
232 name: "test-queue-decode-failure",
233 id: "poison",
234 element: { n: null },
235 isCustomId: true
236 })
237 yield* queue.offer({ n: 42n })
238
239 // the poison element is marked as failed and skipped
240 const value = yield* queue.take(Effect.succeed)
241 assert.deepStrictEqual(value, { n: 42n })
242
243 // the poison element is not delivered again
244 const fiber = yield* queue.take(Effect.succeed).pipe(Effect.forkScoped)
245 yield* assertNotDelivered(fiber)
246 }), testOptions)
247
248 it.effect("cleanup removes expired completed elements", () =>
249 Effect.gen(function*() {
250 const store = yield* PersistedQueue.PersistedQueueStore
251 const queue = yield* PersistedQueue.make({
252 name: "test-queue-cleanup",
253 schema: Item
254 })
255
256 yield* queue.offer({ n: 1n }, { id: "cleanup-id" })
257 yield* queue.take(Effect.succeed)
258
259 // within the ttl the dedupe entry survives, so re-offers are ignored
260 yield* store.cleanup({ timeToLive: Duration.days(30), failedTimeToLive: undefined })
261 yield* queue.offer({ n: 2n }, { id: "cleanup-id" })
262 const fiber = yield* queue.take(Effect.succeed).pipe(Effect.forkScoped)
263 yield* assertNotDelivered(fiber)
264
265 // after the ttl the completed element and its dedupe entry go away
266 yield* advancePastTtl
267 yield* store.cleanup({ timeToLive: Duration.seconds(1), failedTimeToLive: undefined })
268 yield* queue.offer({ n: 3n }, { id: "cleanup-id" })
269 yield* TestClock.adjust(1000)
270 assert.deepStrictEqual(yield* Fiber.join(fiber), { n: 3n })
271 }), testOptions)
272
273 it.effect("cleanup removes failed elements only with failedTimeToLive", () =>
274 Effect.gen(function*() {
275 const store = yield* PersistedQueue.PersistedQueueStore
276 const queue = yield* PersistedQueue.make({
277 name: "test-queue-cleanup-failed",
278 schema: Item,
279 maxAttempts: 1,
280 retrySchedule: Schedule.spaced(0)
281 })
282
283 yield* queue.offer({ n: 1n }, { id: "failed-cleanup-id" })
284 const error = yield* queue.take(() => Effect.fail("boom")).pipe(Effect.flip)
285 assert.strictEqual(error, "boom")
286
287 // without failedTimeToLive the failed element is the dead-letter
288 // record and is kept, so its id stays deduplicated
289 yield* advancePastTtl
290 yield* store.cleanup({ timeToLive: Duration.seconds(1), failedTimeToLive: undefined })
291 yield* queue.offer({ n: 2n }, { id: "failed-cleanup-id" })
292 const fiber = yield* queue.take(Effect.succeed).pipe(Effect.forkScoped)
293 yield* assertNotDelivered(fiber)
294
295 // with failedTimeToLive the failed element and its dedupe entry go away
296 yield* store.cleanup({ timeToLive: Duration.days(30), failedTimeToLive: Duration.seconds(1) })
297 yield* queue.offer({ n: 3n }, { id: "failed-cleanup-id" })
298 yield* TestClock.adjust(1000)
299 assert.deepStrictEqual(yield* Fiber.join(fiber), { n: 3n })
300 }), testOptions)
301
302 it.effect("cleanup keeps dedupe entries for unprocessed elements", () =>
303 Effect.gen(function*() {
304 const store = yield* PersistedQueue.PersistedQueueStore
305 const queue = yield* PersistedQueue.make({
306 name: "test-queue-cleanup-pending",
307 schema: Item
308 })
309
310 yield* queue.offer({ n: 1n }, { id: "pending-cleanup-id" })
311
312 // an element older than the ttl that was never processed keeps its
313 // dedupe entry, so the re-offer does not enqueue a duplicate
314 yield* advancePastTtl
315 yield* store.cleanup({ timeToLive: Duration.seconds(1), failedTimeToLive: undefined })
316 yield* queue.offer({ n: 2n }, { id: "pending-cleanup-id" })
317
318 const value = yield* queue.take(Effect.succeed)
319 assert.deepStrictEqual(value, { n: 1n })
320
321 const fiber = yield* queue.take(Effect.succeed).pipe(Effect.forkScoped)
322 yield* assertNotDelivered(fiber)
323 }), testOptions)
324
325 it.effect("processes concurrent elements exactly once with retries", () =>
326 Effect.gen(function*() {
327 const queue = yield* PersistedQueue.make({
328 name: "test-queue-soak",
329 schema: Item,
330 retrySchedule: Schedule.spaced(0)
331 })
332
333 const total = 24
334 const deliveries = new Map<bigint, number>()
335 const succeeded = new Set<bigint>()
336 const completed = Latch.makeUnsafe()
337
338 yield* Effect.forEach(
339 Array.from({ length: total }, (_, i) => BigInt(i)),
340 (n) => queue.offer({ n }),
341 { concurrency: 8, discard: true }
342 )
343
344 const worker = queue.take(({ n }) =>
345 Effect.suspend(() => {
346 const count = (deliveries.get(n) ?? 0) + 1
347 deliveries.set(n, count)
348 // every third element fails on its first delivery
349 if (n % 3n === 0n && count === 1) {
350 return Effect.fail("transient")
351 }
352 succeeded.add(n)
353 if (succeeded.size === total) {
354 completed.openUnsafe()
355 }
356 return Effect.void
357 })
358 ).pipe(Effect.ignore, Effect.forever)
359
360 yield* Effect.forkScoped(worker)
361 yield* Effect.forkScoped(worker)
362 yield* Effect.forkScoped(worker)
363
364 while (!completed.isOpen()) {
365 yield* TestClock.adjust(1000)
366 yield* Effect.sleep(250).pipe(TestClock.withLive)
367 }
368
369 assert.strictEqual(succeeded.size, total)
370 for (const [n, count] of deliveries) {
371 assert.strictEqual(count, n % 3n === 0n ? 2 : 1, `deliveries for element ${n}`)
372 }
373 }), testOptions)
374 })
375}
376
377export const suite = (name: string, layer: Layer.Layer<PersistedQueue.PersistedQueueStore, unknown>) =>
378 suiteWith(name, layer, it)