| 1 | import { assert, it } from "@effect/vitest" |
| 2 | import { Effect, Fiber, Layer, Schedule, Schema } from "effect" |
| 3 | import { PersistedQueue } from "effect/persistence" |
| 4 | import { TestClock } from "effect/testing" |
| 5 | import * as PersistedQueueTest from "./PersistedQueueTest.ts" |
| 6 | |
| 7 | PersistedQueueTest.suite("memory", PersistedQueue.layerStoreMemory) |
| 8 | |
| 9 | const Item = Schema.Struct({ |
| 10 | n: Schema.BigInt |
| 11 | }) |
| 12 | |
| 13 | it.layer( |
| 14 | PersistedQueue.layer.pipe(Layer.provideMerge(PersistedQueue.layerStoreMemory)) |
| 15 | )("PersistedQueue.retrySchedule", (it) => { |
| 16 | it.effect("derives growing delays from the persisted attempt count", () => |
| 17 | Effect.gen(function*() { |
| 18 | const queue = yield* PersistedQueue.make({ |
| 19 | name: "retry-schedule-state", |
| 20 | schema: Item, |
| 21 | retrySchedule: Schedule.exponential(1000) |
| 22 | }) |
| 23 | const attempts: Array<number> = [] |
| 24 | const takeFail = queue.take((_val, metadata) => { |
| 25 | attempts.push(metadata.attempts) |
| 26 | return Effect.fail("boom") |
| 27 | }).pipe(Effect.flip) |
| 28 | |
| 29 | yield* queue.offer({ n: 1n }) |
| 30 | |
| 31 | // attempt 1 at t=0 schedules the retry for t=1s |
| 32 | yield* takeFail |
| 33 | |
| 34 | const fiber = yield* takeFail.pipe(Effect.forkScoped) |
| 35 | yield* TestClock.adjust(500) |
| 36 | assert.isUndefined(fiber.pollUnsafe()) |
| 37 | yield* TestClock.adjust(500) |
| 38 | assert.strictEqual(yield* Fiber.join(fiber), "boom") |
| 39 | |
| 40 | // attempt 2 failed at t=1s: the schedule must not restart at 1s, the |
| 41 | // next delay is 2s |
| 42 | const fiber2 = yield* takeFail.pipe(Effect.forkScoped) |
| 43 | yield* TestClock.adjust(1500) |
| 44 | assert.isUndefined(fiber2.pollUnsafe()) |
| 45 | yield* TestClock.adjust(600) |
| 46 | assert.strictEqual(yield* Fiber.join(fiber2), "boom") |
| 47 | |
| 48 | assert.deepStrictEqual(attempts, [1, 2, 3]) |
| 49 | })) |
| 50 | }) |
| 51 | |
| 52 | it.layer( |
| 53 | PersistedQueue.layer.pipe(Layer.provideMerge(PersistedQueue.layerStoreMemory)) |
| 54 | )("PersistedQueue.layerCleanup", (it) => { |
| 55 | it.effect("prunes completed elements on the configured interval", () => |
| 56 | Effect.gen(function*() { |
| 57 | const queue = yield* PersistedQueue.make({ name: "cleanup-layer", schema: Item }) |
| 58 | yield* queue.offer({ n: 1n }, { id: "cleanup-layer-id" }) |
| 59 | yield* queue.take(Effect.succeed) |
| 60 | |
| 61 | yield* Layer.build(PersistedQueue.layerCleanup({ |
| 62 | interval: "1 minute", |
| 63 | timeToLive: "1 hour" |
| 64 | })) |
| 65 | |
| 66 | // within the ttl the id stays deduplicated |
| 67 | yield* TestClock.adjust("2 minutes") |
| 68 | yield* queue.offer({ n: 2n }, { id: "cleanup-layer-id" }) |
| 69 | const fiber = yield* queue.take(Effect.succeed).pipe(Effect.forkScoped) |
| 70 | yield* TestClock.adjust(1000) |
| 71 | assert.isUndefined(fiber.pollUnsafe()) |
| 72 | |
| 73 | // after the ttl a cleanup run prunes the entry and the id can be reused |
| 74 | yield* TestClock.adjust("2 hours") |
| 75 | yield* queue.offer({ n: 3n }, { id: "cleanup-layer-id" }) |
| 76 | yield* TestClock.adjust(1000) |
| 77 | assert.deepStrictEqual(yield* Fiber.join(fiber), { n: 3n }) |
| 78 | })) |
| 79 | }) |