1import { assert, it } from "@effect/vitest"
2import { Duration, Effect, Fiber, Latch, Layer, Schema } from "effect"
3import * as PersistedCacheTest from "effect-test/persistence/PersistedCacheTest"
4import * as PersistedQueueTest from "effect-test/persistence/PersistedQueueTest"
5import * as SqlCleanupTest from "effect-test/persistence/SqlCleanupTest"
6import { PersistedQueue, Persistence } from "effect/persistence"
7import { SqlClient } from "effect/sql"
8import { TestClock } from "effect/testing"
9import { PgContainer } from "./utils.ts"
10
11PersistedCacheTest.suite(
12 "sql-pg-multi",
13 Persistence.layerSqlMultiTable.pipe(Layer.provide(PgContainer.layerClient))
14)
15
16PersistedCacheTest.suite(
17 "sql-pg-single",
18 Persistence.layerSql.pipe(Layer.provide(PgContainer.layerClient))
19)
20
21PersistedQueueTest.suite(
22 "sql-pg",
23 PersistedQueue.layerStoreSql().pipe(Layer.provide(PgContainer.layerClient))
24)
25
26it.layer(PgContainer.layerClient, { timeout: "30 seconds" })("PersistedQueue SQL locks", (it) => {
27 it.effect("refreshes locks for acquired elements", () =>
28 Effect.gen(function*() {
29 const options = {
30 tableName: "effect_queue_lock_refresh",
31 pollInterval: "10 millis",
32 lockRefreshInterval: "100 millis",
33 lockExpiration: "1 second"
34 } as const
35 const store1 = yield* PersistedQueue.makeStoreSql(options)
36 const store2 = yield* PersistedQueue.makeStoreSql(options)
37 const element = { message: "hello" }
38
39 yield* store1.offer({
40 name: "lock-refresh",
41 id: crypto.randomUUID(),
42 element,
43 isCustomId: false
44 })
45
46 const takeOptions = {
47 name: "lock-refresh",
48 maxAttempts: 10,
49 retryDelay: () => Effect.succeed(Duration.zero)
50 }
51 const acquired = Latch.makeUnsafe()
52 const first = yield* Effect.scoped(Effect.gen(function*() {
53 yield* store1.take(takeOptions)
54 yield* acquired.open
55 return yield* Effect.never
56 })).pipe(Effect.forkScoped)
57
58 yield* acquired.await
59
60 const second = yield* Effect.scoped(
61 store2.take(takeOptions)
62 ).pipe(Effect.forkScoped)
63
64 yield* Effect.sleep("1500 millis")
65 assert.isUndefined(second.pollUnsafe())
66
67 yield* Fiber.interrupt(first)
68 const received = yield* Fiber.join(second)
69 assert.deepStrictEqual(received.element, element)
70 }).pipe(TestClock.withLive))
71
72 it.effect("dead-letters malformed JSON and continues", () =>
73 Effect.gen(function*() {
74 const tableName = "effect_queue_invalid_json"
75 const store = yield* PersistedQueue.makeStoreSql({
76 tableName,
77 pollInterval: "10 millis"
78 })
79 const factory = yield* PersistedQueue.makeFactory.pipe(
80 Effect.provideService(PersistedQueue.PersistedQueueStore, store)
81 )
82 const queue = yield* factory.make({
83 name: "invalid-json",
84 schema: Schema.String
85 })
86 const sql = (yield* SqlClient.SqlClient).withoutTransforms()
87 const table = sql(tableName)
88 const poisonId = crypto.randomUUID()
89
90 yield* store.offer({
91 name: "invalid-json",
92 id: poisonId,
93 element: "poison",
94 isCustomId: false
95 })
96 yield* sql`UPDATE ${table} SET element = ${"{"} WHERE id = ${poisonId}`
97 yield* queue.offer("valid")
98
99 // the malformed element is skipped and the next one is delivered
100 const value = yield* queue.take(Effect.succeed)
101 assert.strictEqual(value, "valid")
102
103 const rows = yield* sql<{
104 readonly state: string
105 readonly attempts: number
106 readonly last_failure: string | null
107 }>`SELECT state, attempts, last_failure FROM ${table} WHERE id = ${poisonId}`
108 assert.strictEqual(rows[0].state, "failed")
109 assert.strictEqual(Number(rows[0].attempts), 1)
110 assert.isNotNull(rows[0].last_failure)
111 }).pipe(TestClock.withLive))
112
113 it.effect("processes elements exactly once across two workers", () =>
114 Effect.gen(function*() {
115 const options = {
116 tableName: "effect_queue_two_workers",
117 pollInterval: "10 millis"
118 } as const
119 const makeQueue = Effect.fnUntraced(function*(store: PersistedQueue.PersistedQueueStore["Service"]) {
120 const factory = yield* PersistedQueue.makeFactory.pipe(
121 Effect.provideService(PersistedQueue.PersistedQueueStore, store)
122 )
123 return yield* factory.make({ name: "two-workers", schema: Schema.Number })
124 })
125 const queue1 = yield* makeQueue(yield* PersistedQueue.makeStoreSql(options))
126 const queue2 = yield* makeQueue(yield* PersistedQueue.makeStoreSql(options))
127
128 const total = 20
129 yield* Effect.forEach(
130 Array.from({ length: total }, (_, i) => i),
131 (n) => queue1.offer(n),
132 { concurrency: 8, discard: true }
133 )
134
135 const seen = new Map<number, number>()
136 const worker = (queue: typeof queue1) =>
137 queue.take((n) =>
138 Effect.sync(() => {
139 seen.set(n, (seen.get(n) ?? 0) + 1)
140 })
141 ).pipe(Effect.forever, Effect.forkScoped)
142 yield* worker(queue1)
143 yield* worker(queue1)
144 yield* worker(queue2)
145 yield* worker(queue2)
146
147 yield* waitFor(() => Effect.sync(() => seen.size === total))
148
149 assert.strictEqual(seen.size, total)
150 for (const [n, count] of seen) {
151 assert.strictEqual(count, 1, `deliveries for element ${n}`)
152 }
153 }).pipe(TestClock.withLive), { timeout: 30000 })
154
155 it.effect("recovers elements from crashed workers after lock expiration", () =>
156 Effect.gen(function*() {
157 const tableName = "effect_queue_crash_recovery"
158 const store = yield* PersistedQueue.makeStoreSql({
159 tableName,
160 pollInterval: "10 millis",
161 lockExpiration: "1 second"
162 })
163 const factory = yield* PersistedQueue.makeFactory.pipe(
164 Effect.provideService(PersistedQueue.PersistedQueueStore, store)
165 )
166 const queue = yield* factory.make({ name: "crash-recovery", schema: Schema.Number })
167 const sql = (yield* SqlClient.SqlClient).withoutTransforms()
168 const table = sql(tableName)
169
170 const id = yield* queue.offer(1)
171 // simulate a worker that claimed the element and then crashed: the lock
172 // is held by a dead worker and the claim consumed an attempt
173 yield* sql`
174 UPDATE ${table}
175 SET acquired_by = ${crypto.randomUUID()}, acquired_at = NOW(), attempts = 1
176 WHERE id = ${id}
177 `
178
179 const attempts = yield* queue.take((_n, metadata) => Effect.succeed(metadata.attempts))
180 assert.strictEqual(attempts, 2)
181 }).pipe(TestClock.withLive), { timeout: 20000 })
182
183 it.effect("dead-letters elements from workers that crashed on the final attempt", () =>
184 Effect.gen(function*() {
185 const tableName = "effect_queue_crash_exhausted"
186 const store = yield* PersistedQueue.makeStoreSql({
187 tableName,
188 pollInterval: "10 millis",
189 lockExpiration: "500 millis",
190 lockRefreshInterval: "200 millis"
191 })
192 const factory = yield* PersistedQueue.makeFactory.pipe(
193 Effect.provideService(PersistedQueue.PersistedQueueStore, store)
194 )
195 const queue = yield* factory.make({ name: "crash-exhausted", schema: Schema.Number, maxAttempts: 1 })
196 const sql = (yield* SqlClient.SqlClient).withoutTransforms()
197 const table = sql(tableName)
198
199 const id = yield* queue.offer(1)
200 // the final attempt was claimed by a worker that crashed, so no
201 // finalizer will ever settle this element
202 yield* sql`
203 UPDATE ${table}
204 SET acquired_by = ${crypto.randomUUID()}, acquired_at = NOW(), attempts = 1
205 WHERE id = ${id}
206 `
207
208 // an active taker runs the periodic pass that flips such rows to failed
209 const fiber = yield* queue.take(Effect.succeed).pipe(Effect.forkScoped)
210
211 const state = () =>
212 sql<{ readonly state: string; readonly last_failure: string | null }>`
213 SELECT state, last_failure FROM ${table} WHERE id = ${id}
214 `.pipe(Effect.map((rows) => rows[0]))
215 yield* waitFor(() => state().pipe(Effect.map((row) => row.state === "failed")))
216
217 const row = yield* state()
218 assert.strictEqual(row.state, "failed")
219 assert.include(row.last_failure ?? "", "Lock expired after final attempt")
220 assert.isUndefined(fiber.pollUnsafe())
221 }).pipe(TestClock.withLive), { timeout: 20000 })
222})
223
224// polls a condition on the live clock until it holds or the rounds run out
225const waitFor = <E>(condition: () => Effect.Effect<boolean, E>) =>
226 Effect.gen(function*() {
227 for (let i = 0; i < 100; i++) {
228 if (yield* condition()) return
229 yield* Effect.sleep(100)
230 }
231 })
232
233it.layer(PgContainer.layerClient, { timeout: "30 seconds" })("Persistence SQL cleanup", (it) => {
234 it.effect("deletes expired entries in batches", () =>
235 Effect.gen(function*() {
236 const sql = (yield* SqlClient.SqlClient).withoutTransforms()
237 const table = sql("effect_persistence")
238 const expiredCount = sql<{ readonly count: number }>`
239 SELECT COUNT(*)::INT AS count FROM ${table} WHERE store_id = 'expired'
240 `.pipe(Effect.map((rows) => rows[0].count))
241 yield* sql`
242 CREATE TABLE ${table} (
243 store_id TEXT NOT NULL,
244 id TEXT NOT NULL,
245 value TEXT NOT NULL,
246 expires BIGINT,
247 PRIMARY KEY (store_id, id)
248 )
249 `
250
251 const entries = Array.from({ length: SqlCleanupTest.expiredEntryCount }, (_, i) => ({
252 store_id: "expired",
253 id: String(i),
254 value: "{}",
255 expires: SqlCleanupTest.expiredAtEpoch
256 }))
257 yield* sql`INSERT INTO ${table} ${sql.insert(entries)}`.unprepared
258 yield* sql`
259 INSERT INTO ${table} (store_id, id, value, expires)
260 VALUES ('live', 'live', '{}', NULL), ('live', 'future', '{}', ${BigInt(SqlCleanupTest.futureExpiresAt)})
261 `
262
263 yield* Layer.build(Persistence.layerBackingSql).pipe(TestClock.withLive)
264
265 const expired = yield* SqlCleanupTest.waitForCount(expiredCount, (count) => count === 0)
266 assert.strictEqual(expired, 0)
267 const live = yield* sql<{ readonly count: number }>`
268 SELECT COUNT(*)::INT AS count FROM ${table} WHERE store_id = 'live'
269 `
270 assert.strictEqual(live[0].count, 2)
271
272 const indexes = yield* sql<{ readonly count: number }>`
273 SELECT COUNT(*)::INT AS count FROM pg_indexes
274 WHERE tablename = 'effect_persistence'
275 AND indexname = 'effect_persistence_expires_idx'
276 `
277 assert.strictEqual(indexes[0].count, 1)
278 }), { timeout: SqlCleanupTest.testTimeout })
279})