1import { PgConnection, PgPool } from "@effect/sql-pg"
2import { assert, it } from "@effect/vitest"
3import { Effect, Fiber, Queue, Redacted, Stream } from "effect"
4import * as Net from "node:net"
5import { Duplex } from "node:stream"
6import { PgContainer } from "./utils.ts"
7
8// `it.effect` runs under the TestClock, so poll loops sleep in real time.
9const realSleep = Effect.promise(() => new Promise((resolve) => setTimeout(resolve, 10)))
10const activeTimeoutMs = 10_000
11const cancellationTestTimeout = 20_000
12
13const poolConfig = Effect.gen(function*() {
14 const container = yield* PgContainer
15 return { url: Redacted.make(container.getConnectionUri()) }
16})
17
18const waitUntilActive = (observer: PgConnection.PgConnection, pid: number) =>
19 Effect.gen(function*() {
20 const startedAt = Date.now()
21 while (Date.now() - startedAt < activeTimeoutMs) {
22 const active = yield* observer.query(
23 "SELECT count(*)::int4 AS active FROM pg_stat_activity WHERE pid = $1 AND state = 'active'",
24 [pid]
25 )
26 if (active.rows[0].active === 1) return
27 yield* realSleep
28 }
29 return yield* Effect.fail(new Error(`PostgreSQL backend ${pid} did not become active within 10 seconds`))
30 })
31
32const cancelRequestCode = 80877102
33
34/** Holds `CancelRequest` messages until the test releases them. */
35class LateCancelProxy extends Duplex {
36 private readonly backend: Net.Socket
37 private cancelDeferred = false
38
39 constructor(host: string, port: number, private readonly gate: CancelRequestGate) {
40 super()
41 this.backend = Net.connect({ host, port, noDelay: true })
42 this.backend.on("data", (chunk: Buffer) => {
43 if (!this.push(chunk)) this.backend.pause()
44 })
45 this.backend.on("end", () => this.push(null))
46 this.backend.on("error", (error) => this.destroy(error))
47 this.backend.on("close", () => {
48 if (this.cancelDeferred) this.gate.markDelivered()
49 else this.destroy()
50 })
51 }
52
53 override _read(): void {
54 this.backend.resume()
55 }
56
57 override _write(chunk: Buffer, _encoding: BufferEncoding, callback: (error?: Error | null) => void): void {
58 if (chunk.length === 16 && chunk.readInt32BE(0) === 16 && chunk.readInt32BE(4) === cancelRequestCode) {
59 this.cancelDeferred = true
60 const backend = this.backend
61 this.gate.intercept(() => backend.end(chunk))
62 callback()
63 this.destroy()
64 return
65 }
66 this.backend.write(chunk, callback)
67 }
68
69 override _final(callback: (error?: Error | null) => void): void {
70 this.backend.end(callback)
71 }
72
73 override _destroy(error: Error | null, callback: (error?: Error | null) => void): void {
74 if (!this.cancelDeferred) this.backend.destroy()
75 callback(error)
76 }
77}
78
79class CancelRequestGate {
80 private readonly interceptedPromise: Promise<void>
81 private readonly deliveredPromise: Promise<void>
82 private resolveIntercepted!: () => void
83 private resolveDelivered!: () => void
84 private forward: (() => void) | undefined
85 private released = false
86
87 constructor() {
88 this.interceptedPromise = new Promise((resolve) => {
89 this.resolveIntercepted = resolve
90 })
91 this.deliveredPromise = new Promise((resolve) => {
92 this.resolveDelivered = resolve
93 })
94 }
95
96 get intercepted() {
97 return Effect.promise(() => this.interceptedPromise).pipe(Effect.timeout("5 seconds"))
98 }
99
100 get delivered() {
101 return Effect.promise(() => this.deliveredPromise).pipe(Effect.timeout("5 seconds"))
102 }
103
104 intercept(forward: () => void): void {
105 this.forward = forward
106 this.resolveIntercepted()
107 if (this.released) this.release()
108 }
109
110 release(): void {
111 this.released = true
112 const forward = this.forward
113 this.forward = undefined
114 forward?.()
115 }
116
117 markDelivered(): void {
118 this.resolveDelivered()
119 }
120}
121
122const lateCancelPoolConfig = (gate: CancelRequestGate) =>
123 Effect.gen(function*() {
124 const container = yield* PgContainer
125 const host = container.getHost()
126 const port = container.getMappedPort(5432)
127 return {
128 ...(yield* poolConfig),
129 maxConnections: 1,
130 stream: () => new LateCancelProxy(host, port, gate)
131 }
132 })
133
134const blockedStream = `
135 SELECT n FROM generate_series(1, 1000) AS g(n)
136 UNION ALL
137 SELECT 1001::int4 AS n
138 FROM (SELECT pg_advisory_xact_lock($1::int4)) AS blocked
139`
140
141it.layer(PgContainer.layer, { timeout: "30 seconds", concurrent: false })("PgPool", (it) => {
142 it.effect("reuses checked out connections", () =>
143 Effect.gen(function*() {
144 const pool = yield* PgPool.make({ ...(yield* poolConfig), maxConnections: 1 })
145 const checkout = Effect.scoped(Effect.gen(function*() {
146 const connection = yield* pool.get
147 const result = yield* connection.query("SELECT $1::int4 AS one", [1])
148 assert.deepStrictEqual(result.rows, [{ one: 1 }])
149 return connection.processId
150 }))
151 const first = yield* checkout
152 const second = yield* checkout
153 assert.strictEqual(first, second)
154 }))
155
156 it.effect("uses a connection once when its TTL is zero", () =>
157 Effect.gen(function*() {
158 const pool = yield* PgPool.make({ ...(yield* poolConfig), connectionTTL: 0, maxConnections: 1 })
159 const checkout = Effect.scoped(Effect.map(pool.get, (connection) => connection.processId)).pipe(
160 Effect.timeout("1 second")
161 )
162 const first = yield* checkout
163 const second = yield* checkout
164 assert.notStrictEqual(first, second)
165 }))
166
167 it.effect("streams rows incrementally and cancels on early abort", () =>
168 Effect.gen(function*() {
169 const pool = yield* PgPool.make(yield* poolConfig)
170 const connection = yield* pool.get
171 const rows = yield* Stream.runCollect(
172 connection.stream("SELECT n FROM generate_series(1, $1::int4) AS g(n)", [1000])
173 )
174 assert.strictEqual(rows.length, 1000)
175 assert.deepStrictEqual(rows[0], { n: 1 })
176 assert.deepStrictEqual(rows[999], { n: 1000 })
177
178 const aborted = yield* Stream.runCollect(
179 connection.stream("SELECT n FROM generate_series(1, 5000) AS g(n)").pipe(Stream.take(5))
180 )
181 assert.deepStrictEqual(aborted, [{ n: 1 }, { n: 2 }, { n: 3 }, { n: 4 }, { n: 5 }])
182
183 const result = yield* connection.query("SELECT pg_backend_pid()::int4 AS pid, $1::text AS after", ["ok"])
184 assert.deepStrictEqual(result.rows, [{ pid: connection.processId, after: "ok" }])
185 }), cancellationTestTimeout)
186
187 it.effect("acquires a multiplexed listener after registration", () =>
188 Effect.gen(function*() {
189 const pool = yield* PgPool.make({ ...(yield* poolConfig), maxConnections: 2, multiplex: true })
190 const listener = yield* pool.reserve
191 const notifications = yield* listener.listen("test_channel")
192 const notifier = yield* pool.get
193 assert.notStrictEqual(listener.processId, notifier.processId)
194 yield* notifier.query("SELECT pg_notify($1, $2)", ["test_channel", "hello"])
195 const notification = yield* Queue.take(notifications)
196 assert.strictEqual(notification.channel, "test_channel")
197 assert.strictEqual(notification.payload, "hello")
198 assert.strictEqual(notification.processId, notifier.processId)
199 }))
200
201 it.effect("reports listener registration failure during acquisition", () =>
202 Effect.gen(function*() {
203 const pool = yield* PgPool.make({ ...(yield* poolConfig), maxConnections: 1 })
204 const connection = yield* pool.get
205 yield* connection.query("BEGIN")
206 yield* Effect.flip(connection.query("SELECT * FROM effect_missing_relation"))
207
208 const error = yield* Effect.flip(connection.listen("test_channel"))
209 assert.strictEqual(error._tag, "SqlError")
210
211 // A failed acquisition must release the connection pin.
212 yield* connection.query("ROLLBACK")
213 const result = yield* connection.query("SELECT 1 AS one")
214 assert.deepStrictEqual(result.rows, [{ one: 1 }])
215 }))
216
217 it.effect("preserves the server error when a listener backend is terminated", () =>
218 Effect.gen(function*() {
219 const config = yield* poolConfig
220 const pool = yield* PgPool.make({ ...config, maxConnections: 1, multiplex: true })
221 const listener = yield* pool.reserve
222 const notifications = yield* listener.listen("terminated_listener")
223 const consumer = yield* Effect.forkScoped(Queue.take(notifications))
224 const terminator = yield* PgConnection.make(config)
225
226 yield* terminator.query("SELECT pg_terminate_backend($1)", [listener.processId])
227 const error = yield* Effect.flip(Fiber.join(consumer))
228 assert.strictEqual(error._tag, "SqlError")
229 assert.propertyVal(error.reason.cause, "code", "57P01")
230 assert.strictEqual(yield* Effect.flip(listener.query("SELECT 1")), error)
231 }))
232
233 it.effect("returns a multiplexed reservation to shared circulation", () =>
234 Effect.gen(function*() {
235 const pool = yield* PgPool.make({ ...(yield* poolConfig), maxConnections: 1, multiplex: true })
236 const reserve = Effect.scoped(Effect.map(pool.reserve, (connection) => connection.processId))
237 const get = Effect.scoped(Effect.map(pool.get, (connection) => connection.processId))
238
239 const first = yield* reserve
240 assert.strictEqual(yield* get, first)
241 assert.strictEqual(yield* reserve, first)
242 assert.strictEqual(yield* get, first)
243 }))
244
245 it.effect("interrupt cancels an in-flight query", () =>
246 Effect.gen(function*() {
247 const config = yield* poolConfig
248 const pool = yield* PgPool.make({ ...config, maxConnections: 1 })
249 const blocker = yield* PgConnection.make(config)
250 const observer = yield* PgConnection.make(config)
251 const first = yield* Effect.scoped(Effect.gen(function*() {
252 const connection = yield* pool.get
253 yield* blocker.query("BEGIN")
254 yield* blocker.query("SELECT pg_advisory_xact_lock($1::int4)", [connection.processId])
255 const fiber = yield* Effect.forkScoped(
256 connection.query("SELECT pg_advisory_xact_lock($1::int4)", [connection.processId])
257 )
258 yield* waitUntilActive(observer, connection.processId)
259 yield* connection.interrupt
260 const error = yield* Effect.flip(Fiber.join(fiber))
261 assert.strictEqual(error._tag, "SqlError")
262 assert.strictEqual(error.reason._tag, "StatementTimeoutError")
263 yield* blocker.query("ROLLBACK")
264 return connection.processId
265 }))
266
267 const second = yield* Effect.scoped(Effect.map(pool.get, (connection) => connection.processId))
268 assert.strictEqual(second, first)
269 }), cancellationTestTimeout)
270
271 it.effect("replaces a pooled session after an unconfirmed query cancel", () =>
272 Effect.gen(function*() {
273 const config = yield* poolConfig
274 const gate = new CancelRequestGate()
275 const pool = yield* PgPool.make(yield* lateCancelPoolConfig(gate))
276 const blocker = yield* PgConnection.make(config)
277 const observer = yield* PgConnection.make(config)
278
279 yield* Effect.gen(function*() {
280 const first = yield* Effect.scoped(Effect.gen(function*() {
281 const connection = yield* pool.get
282 yield* blocker.query("BEGIN")
283 yield* blocker.query("SELECT pg_advisory_xact_lock($1::int4)", [connection.processId])
284 const query = yield* Effect.forkScoped(
285 connection.query("SELECT pg_advisory_xact_lock($1::int4)", [connection.processId])
286 )
287 yield* waitUntilActive(observer, connection.processId)
288 const interruption = yield* Effect.forkScoped(Fiber.interrupt(query))
289 yield* gate.intercepted
290 yield* blocker.query("COMMIT")
291 yield* Fiber.join(interruption)
292 return connection.processId
293 }))
294
295 const second = yield* Effect.scoped(Effect.map(pool.get, (connection) => connection.processId))
296 assert.notStrictEqual(second, first)
297 gate.release()
298 yield* gate.delivered
299 }).pipe(Effect.ensuring(Effect.sync(() => gate.release())))
300 }), cancellationTestTimeout)
301
302 it.effect("replaces a session before pool.use after an unconfirmed query cancel", () =>
303 Effect.gen(function*() {
304 const config = yield* poolConfig
305 const gate = new CancelRequestGate()
306 const pool = yield* PgPool.make(yield* lateCancelPoolConfig(gate))
307 const blocker = yield* PgConnection.make(config)
308 const observer = yield* PgConnection.make(config)
309
310 yield* Effect.gen(function*() {
311 const first = yield* Effect.scoped(Effect.gen(function*() {
312 const connection = yield* pool.get
313 yield* blocker.query("BEGIN")
314 yield* blocker.query("SELECT pg_advisory_xact_lock($1::int4)", [connection.processId])
315 const query = yield* Effect.forkScoped(
316 connection.query("SELECT pg_advisory_xact_lock($1::int4)", [connection.processId])
317 )
318 yield* waitUntilActive(observer, connection.processId)
319 const interruption = yield* Effect.forkScoped(Fiber.interrupt(query))
320 yield* gate.intercepted
321 yield* blocker.query("COMMIT")
322 yield* Fiber.join(interruption)
323 return connection.processId
324 }))
325
326 const second = yield* pool.use((connection) => Effect.succeed(connection.processId))
327 assert.notStrictEqual(second, first)
328 gate.release()
329 yield* gate.delivered
330 }).pipe(Effect.ensuring(Effect.sync(() => gate.release())))
331 }), cancellationTestTimeout)
332
333 it.effect("protects an active pool.use query from a late cancel", () =>
334 Effect.gen(function*() {
335 const config = yield* poolConfig
336 const gate = new CancelRequestGate()
337 const pool = yield* PgPool.make(yield* lateCancelPoolConfig(gate))
338 const blocker = yield* PgConnection.make(config)
339 const observer = yield* PgConnection.make(config)
340 const started = yield* Queue.unbounded<number>()
341 const lockAcquired = yield* Queue.unbounded<void>()
342
343 yield* Effect.gen(function*() {
344 yield* Effect.scoped(Effect.gen(function*() {
345 const connection = yield* pool.get
346 yield* blocker.query("BEGIN")
347 yield* blocker.query("SELECT pg_advisory_xact_lock($1::int4)", [connection.processId])
348 const query = yield* Effect.forkScoped(
349 connection.query("SELECT pg_advisory_xact_lock($1::int4)", [connection.processId])
350 )
351 yield* waitUntilActive(observer, connection.processId)
352 const interruption = yield* Effect.forkScoped(Fiber.interrupt(query))
353 yield* gate.intercepted
354 yield* blocker.query("COMMIT")
355 yield* Fiber.join(interruption)
356 }))
357
358 const followUp = yield* Effect.forkScoped(
359 pool.use((connection) =>
360 Effect.gen(function*() {
361 yield* Queue.offer(started, connection.processId)
362 yield* Queue.take(lockAcquired)
363 return yield* connection.query(
364 "SELECT 1 AS after FROM pg_advisory_xact_lock($1::int4)",
365 [connection.processId]
366 )
367 })
368 )
369 )
370 const followUpPid = yield* Queue.take(started)
371 yield* blocker.query("BEGIN")
372 yield* blocker.query("SELECT pg_advisory_xact_lock($1::int4)", [followUpPid])
373 yield* Queue.offer(lockAcquired, undefined)
374 yield* waitUntilActive(observer, followUpPid)
375 gate.release()
376 yield* gate.delivered
377 yield* blocker.query("COMMIT")
378 const result = yield* Fiber.join(followUp)
379 assert.deepStrictEqual(result.rows, [{ after: 1 }])
380 }).pipe(
381 Effect.ensuring(
382 Effect.sync(() => gate.release()).pipe(
383 Effect.andThen(blocker.query("ROLLBACK")),
384 Effect.ignore
385 )
386 )
387 )
388 }), cancellationTestTimeout)
389
390 it.effect("retires an idle interrupted session before pool.use", () =>
391 Effect.gen(function*() {
392 const config = yield* poolConfig
393 const gate = new CancelRequestGate()
394 const pool = yield* PgPool.make(yield* lateCancelPoolConfig(gate))
395 const observer = yield* PgConnection.make(config)
396 const started = yield* Queue.unbounded<number>()
397
398 yield* Effect.gen(function*() {
399 const first = yield* Effect.scoped(Effect.gen(function*() {
400 const connection = yield* pool.get
401 yield* connection.interrupt
402 yield* gate.intercepted
403 return connection.processId
404 }))
405
406 const followUp = yield* Effect.forkScoped(
407 pool.use((connection) =>
408 Effect.gen(function*() {
409 yield* Queue.offer(started, connection.processId)
410 return yield* connection.query("SELECT 1 AS after FROM pg_sleep(1)")
411 })
412 )
413 )
414 const followUpPid = yield* Queue.take(started)
415 assert.notStrictEqual(followUpPid, first)
416 yield* waitUntilActive(observer, followUpPid)
417 gate.release()
418 yield* gate.delivered
419 const result = yield* Fiber.join(followUp)
420 assert.deepStrictEqual(result.rows, [{ after: 1 }])
421 }).pipe(Effect.ensuring(Effect.sync(() => gate.release())))
422 }), cancellationTestTimeout)
423
424 it.effect("replaces a pooled session after an unconfirmed stream cancel", () =>
425 Effect.gen(function*() {
426 const config = yield* poolConfig
427 const gate = new CancelRequestGate()
428 const pool = yield* PgPool.make(yield* lateCancelPoolConfig(gate))
429 const blocker = yield* PgConnection.make(config)
430
431 yield* Effect.gen(function*() {
432 const first = yield* Effect.scoped(Effect.gen(function*() {
433 const connection = yield* pool.get
434 yield* blocker.query("BEGIN")
435 yield* blocker.query("SELECT pg_advisory_xact_lock($1::int4)", [connection.processId])
436 const stream = yield* Effect.forkScoped(
437 connection.stream(blockedStream, [connection.processId]).pipe(Stream.take(1), Stream.runCollect)
438 )
439 yield* gate.intercepted
440 yield* blocker.query("COMMIT")
441 const rows = yield* Fiber.join(stream)
442 assert.deepStrictEqual(rows, [{ n: 1 }])
443 return connection.processId
444 }))
445
446 yield* Effect.scoped(Effect.gen(function*() {
447 const replacement = yield* pool.get
448 assert.notStrictEqual(replacement.processId, first)
449 const query = yield* Effect.forkScoped(replacement.query("SELECT 1 AS after FROM pg_sleep(1)"))
450 yield* waitUntilActive(blocker, replacement.processId)
451 gate.release()
452 yield* gate.delivered
453 const result = yield* Fiber.join(query)
454 assert.deepStrictEqual(result.rows, [{ after: 1 }])
455 }))
456 }).pipe(Effect.ensuring(Effect.sync(() => gate.release())))
457 }), cancellationTestTimeout)
458
459 it.effect("keeps a session after a confirmed stream cancel", () =>
460 Effect.gen(function*() {
461 const config = yield* poolConfig
462 const pool = yield* PgPool.make({ ...config, maxConnections: 1 })
463 const blocker = yield* PgConnection.make(config)
464 const connection = yield* pool.get
465 yield* blocker.query("BEGIN")
466 yield* blocker.query("SELECT pg_advisory_xact_lock($1::int4)", [connection.processId])
467 const rows = yield* connection.stream(blockedStream, [connection.processId]).pipe(
468 Stream.take(1),
469 Stream.runCollect,
470 Effect.timeout("5 seconds")
471 )
472 assert.deepStrictEqual(rows, [{ n: 1 }])
473 const result = yield* connection.query("SELECT pg_backend_pid()::int4 AS pid")
474 assert.deepStrictEqual(result.rows, [{ pid: connection.processId }])
475 yield* blocker.query("ROLLBACK")
476 }), cancellationTestTimeout)
477
478 it.effect("keeps a held session after an unconfirmed query cancel", () =>
479 Effect.gen(function*() {
480 const config = yield* poolConfig
481 const gate = new CancelRequestGate()
482 const pool = yield* PgPool.make(yield* lateCancelPoolConfig(gate))
483 const blocker = yield* PgConnection.make(config)
484 const observer = yield* PgConnection.make(config)
485
486 yield* Effect.gen(function*() {
487 const connection = yield* pool.get
488 yield* blocker.query("BEGIN")
489 yield* blocker.query("SELECT pg_advisory_xact_lock($1::int4)", [connection.processId])
490 const query = yield* Effect.forkScoped(
491 connection.query("SELECT pg_advisory_xact_lock($1::int4)", [connection.processId])
492 )
493 yield* waitUntilActive(observer, connection.processId)
494 const interruption = yield* Effect.forkScoped(Fiber.interrupt(query))
495 yield* gate.intercepted
496 yield* blocker.query("COMMIT")
497 yield* Fiber.join(interruption)
498
499 const beforeDelivery = yield* connection.query("SELECT pg_backend_pid()::int4 AS pid")
500 assert.deepStrictEqual(beforeDelivery.rows, [{ pid: connection.processId }])
501 gate.release()
502 yield* gate.delivered
503 const afterDelivery = yield* connection.query("SELECT pg_backend_pid()::int4 AS pid")
504 assert.deepStrictEqual(afterDelivery.rows, [{ pid: connection.processId }])
505 }).pipe(Effect.ensuring(Effect.sync(() => gate.release())))
506 }), cancellationTestTimeout)
507
508 it.effect("replaces connections that die", () =>
509 Effect.gen(function*() {
510 const config = yield* poolConfig
511 const pool = yield* PgPool.make({ ...config, maxConnections: 1 })
512 const first = yield* Effect.scoped(Effect.map(pool.get, (connection) => connection.processId))
513 const terminator = yield* PgConnection.make(config)
514 yield* terminator.query("SELECT pg_terminate_backend($1)", [first])
515 // The pool notices the dead connection asynchronously; retry until it
516 // hands out a healthy replacement.
517 while (true) {
518 const replaced = yield* Effect.scoped(Effect.gen(function*() {
519 const connection = yield* pool.get
520 const alive = yield* Effect.isSuccess(connection.query("SELECT 1 AS one"))
521 return alive && connection.processId !== first ? connection.processId : undefined
522 }))
523 if (replaced !== undefined) {
524 assert.notStrictEqual(replaced, first)
525 break
526 }
527 yield* realSleep
528 }
529 }))
530 it.effect("defaults multiplex concurrency to 32", () =>
531 Effect.gen(function*() {
532 const pool = yield* PgPool.make({ ...(yield* poolConfig), maxConnections: 4, multiplex: true })
533 const processIds = yield* Effect.all(
534 Array.from({ length: 32 }, () =>
535 Effect.scoped(Effect.flatMap(pool.get, (connection) =>
536 Effect.as(
537 connection.query("SELECT pg_sleep(0.05)"),
538 connection.processId
539 )))),
540 { concurrency: "unbounded" }
541 )
542 assert.strictEqual(new Set(processIds).size, 1)
543 }))
544 it.effect("invalidates a reserved connection", () =>
545 Effect.gen(function*() {
546 const pool = yield* PgPool.make({ ...(yield* poolConfig), maxConnections: 2, multiplex: true })
547 const reserved = yield* pool.reserve
548 yield* pool.invalidate(reserved)
549 const replacement = yield* pool.get
550 assert.notStrictEqual(replacement.processId, reserved.processId)
551 }))
552
553 it.effect("keeps a multiplexed stream's connection to itself", () =>
554 Effect.gen(function*() {
555 const pool = yield* PgPool.make({ ...(yield* poolConfig), maxConnections: 2, multiplex: true })
556 const connection = yield* pool.get
557 // A stream pins its connection for its lifetime. A checkout that landed
558 // on the same one would wait behind a stream only it could drain.
559 const doubled = yield* connection.stream("SELECT generate_series(1, 5) AS n").pipe(
560 Stream.mapEffect((row) =>
561 Effect.flatMap(pool.get, (other) => other.query("SELECT $1::int4 AS d", [(row as any).n * 2]))
562 ),
563 Stream.map((result) => (result.rows[0] as any).d),
564 Stream.runCollect
565 )
566 assert.deepStrictEqual(Array.from(doubled), [2, 4, 6, 8, 10])
567 }))
568 it.effect("admits a waiting checkout when a reserved connection dies", () =>
569 Effect.gen(function*() {
570 const pool = yield* PgPool.make({ ...(yield* poolConfig), maxConnections: 1, multiplex: true })
571 const reserved = yield* pool.reserve
572 const waiter = yield* Effect.forkScoped(
573 Effect.scoped(Effect.flatMap(pool.get, (connection) => connection.query("SELECT 1 AS ok")))
574 )
575 yield* realSleep
576
577 // The reservation is still held when its connection dies, so nothing
578 // returns the connection to the pool. The waiter behind it would queue
579 // for a connection that is never coming back.
580 yield* Effect.ignore(reserved.query("SELECT pg_terminate_backend(pg_backend_pid())"))
581
582 const result = yield* Fiber.join(waiter)
583 assert.deepStrictEqual(result.rows, [{ ok: 1 }])
584 }))
585 it.effect("returns a borrowed session on success, failure, and interruption", () =>
586 Effect.gen(function*() {
587 // A pool of one: a lease that is not returned hangs the next borrow.
588 const pool = yield* PgPool.make({ ...(yield* poolConfig), maxConnections: 1 })
589 const borrow = <A, E>(effect: (connection: PgConnection.PgConnection) => Effect.Effect<A, E>) => pool.use(effect)
590
591 assert.deepStrictEqual(
592 (yield* borrow((connection) => connection.query("SELECT 1 AS ok"))).rows,
593 [{ ok: 1 }]
594 )
595
596 const failed = yield* Effect.result(borrow((connection) => connection.query("SELECT * FROM nope")))
597 assert.strictEqual(failed._tag, "Failure")
598
599 const running = yield* Effect.forkScoped(borrow((connection) => connection.query("SELECT pg_sleep(30)")))
600 yield* realSleep
601 yield* Fiber.interrupt(running)
602
603 assert.deepStrictEqual(
604 (yield* borrow((connection) => connection.query("SELECT 2 AS ok"))).rows,
605 [{ ok: 2 }]
606 )
607 }), cancellationTestTimeout)
608
609 it.effect("replaces a fatal session before the next pool.use query", () =>
610 Effect.gen(function*() {
611 const pool = yield* PgPool.make({ ...(yield* poolConfig), maxConnections: 1 })
612 const first = yield* pool.use((connection) => Effect.succeed(connection.processId))
613
614 // Decoding this interval fails fatally in the connection's data handler.
615 // The next borrow must not see the dead session, even in the same fiber.
616 const error = yield* Effect.flip(pool.use((connection) => connection.query("SELECT interval '-1 day' AS v")))
617 assert.strictEqual(error._tag, "SqlError")
618 assert.strictEqual(error.reason.message, "PgConnection: Failed to decode row")
619 const next = yield* pool.use((connection) =>
620 Effect.map(connection.query("SELECT 1::int4 AS one"), (result) => ({
621 pid: connection.processId,
622 rows: result.rows
623 }))
624 )
625 assert.notStrictEqual(next.pid, first)
626 assert.deepStrictEqual(next.rows, [{ one: 1 }])
627 }))
628
629 it.effect("borrows around a connection that has to be replaced", () =>
630 Effect.gen(function*() {
631 const pool = yield* PgPool.make({ ...(yield* poolConfig), maxConnections: 1 })
632 const first = yield* pool.use((connection) => Effect.as(connection.query("SELECT 1 AS ok"), connection.processId))
633 // Killing the only session leaves it queued dead, which is exactly the
634 // case the fast path declines to take. The backend answers with an error
635 // and closes afterwards, so wait for the close to land before borrowing
636 // again - both paths would race it otherwise.
637 yield* Effect.ignore(
638 pool.use((connection) => connection.query("SELECT pg_terminate_backend(pg_backend_pid())"))
639 )
640 for (let i = 0; i < 50; i++) yield* realSleep
641 const second = yield* pool.use((connection) =>
642 Effect.as(connection.query("SELECT 2 AS ok"), connection.processId)
643 )
644 assert.notStrictEqual(second, first)
645 }), 20_000)
646 it.effect("honours a configured multiplex concurrency", () =>
647 Effect.gen(function*() {
648 // Two statements to a connection means eight of them need four
649 // connections, where the default for a pool this size would fit them on
650 // one.
651 const pool = yield* PgPool.make({
652 ...(yield* poolConfig),
653 maxConnections: 4,
654 multiplex: true,
655 multiplexConcurrency: 2
656 })
657 const processIds = yield* Effect.all(
658 Array.from({ length: 8 }, () =>
659 Effect.scoped(Effect.flatMap(pool.get, (connection) =>
660 Effect.as(connection.query("SELECT pg_sleep(0.05)"), connection.processId)))),
661 { concurrency: "unbounded" }
662 )
663 assert.strictEqual(new Set(processIds).size, 4)
664 }))
665})