1import { PgPool } from "@effect/sql-pg"
2import { assert, describe, it } from "@effect/vitest"
3import { Effect } from "effect"
4import { Duplex } from "node:stream"
5
6const backendMessage = (tag: string, payload: Buffer): Buffer => {
7 const length = Buffer.allocUnsafe(4)
8 length.writeInt32BE(payload.length + 4)
9 return Buffer.concat([Buffer.from(tag), length, payload])
10}
11
12const ready = Buffer.concat([
13 backendMessage("R", Buffer.alloc(4)),
14 backendMessage("K", Buffer.alloc(8)),
15 backendMessage("Z", Buffer.from("I"))
16])
17
18const waitFor = (predicate: () => boolean) =>
19 Effect.promise(async () => {
20 for (let i = 0; i < 100 && !predicate(); i++) {
21 await new Promise((resolve) => setTimeout(resolve, 10))
22 }
23 assert.ok(predicate(), "background acquisition did not start")
24 })
25
26describe("PgPool failed startup", () => {
27 it.live(
28 "retries a stale background acquisition rather than failing an unrelated checkout",
29 () =>
30 Effect.scoped(Effect.gen(function*() {
31 let attempts = 0
32 let backgroundStarted = false
33 const pool = yield* PgPool.make({
34 username: "test",
35 connectTimeout: "250 millis",
36 minConnections: 2,
37 maxConnections: 2,
38 stream: () => {
39 attempts++
40 const attempt = attempts
41 let startup = true
42 const socket = new Duplex({
43 read() {},
44 write(_chunk: Buffer, _encoding, callback) {
45 if (startup) {
46 startup = false
47 if (attempt === 2) backgroundStarted = true
48 else queueMicrotask(() => socket.push(ready))
49 }
50 callback()
51 }
52 })
53 return socket
54 }
55 })
56
57 // Keep the healthy session leased so the failure becomes a placeholder.
58 const healthy = yield* pool.get
59 yield* waitFor(() => backgroundStarted)
60 yield* Effect.promise(() => new Promise((resolve) => setTimeout(resolve, 400)))
61
62 const replacement = yield* pool.get
63 assert.notStrictEqual(replacement, healthy)
64 assert.strictEqual(attempts, 3)
65 })),
66 10_000
67 )
68
69 it.live(
70 "reports a caller-owned connect failure instead of retrying indefinitely",
71 () =>
72 Effect.scoped(Effect.gen(function*() {
73 let attempts = 0
74 const pool = yield* PgPool.make({
75 username: "test",
76 maxConnections: 1,
77 stream: () => {
78 attempts++
79 const socket = new Duplex({ read() {}, write() {} })
80 queueMicrotask(() => socket.destroy())
81 return socket
82 }
83 })
84 const result = yield* Effect.result(pool.get)
85 assert.strictEqual(result._tag, "Failure")
86 if (result._tag === "Failure") {
87 assert.strictEqual(result.failure.reason._tag, "ConnectionError")
88 assert.strictEqual(result.failure.reason.operation, "connect")
89 }
90 assert.strictEqual(attempts, 1)
91 })),
92 10_000
93 )
94})