EFF-749 Backport sql-pg dedicated listen connection (#6140)

Author: tim-smart

.changeset/bright-pants-thank.md

@@ -0,0 +1,5 @@
+---
+"@effect/sql-pg": patch
+---
+
+Use a dedicated PostgreSQL connection for LISTEN / UNLISTEN so active listeners do not consume a pooled query connection.

packages/sql-pg/src/PgClient.ts

@@ -311,8 +311,33 @@ const makeClient = (
     })
     const reserve = Effect.map(reserveRaw, (client) => new ConnectionImpl(client))
 
+    const onListenClientError = (_: Error) => {
+    }
+
     const listenClient = yield* RcRef.make({
-      acquire: reserveRaw
+      acquire: Effect.acquireRelease(
+        Effect.tryPromise({
+          try: async () => {
+            const client = new Pg.Client(pool.options)
+            await client.connect()
+            client.on("error", onListenClientError)
+            return client
+          },
+          catch: (cause) =>
+            new SqlError({
+              cause,
+              message: "Failed to acquire connection for listen"
+            })
+        }),
+        (client) =>
+          Effect.promise(() => {
+            client.off("error", onListenClientError)
+            return client.end()
+          }).pipe(
+            Effect.interruptible,
+            Effect.timeoutOption(1000)
+          )
+      )
     })
 
     return Object.assign(

packages/sql-pg/test/Client.test.ts

@@ -2,7 +2,7 @@ import { PgClient } from "@effect/sql-pg"
 import * as SqlClient from "@effect/sql/SqlClient"
 import * as Statement from "@effect/sql/Statement"
 import { assert, expect, it } from "@effect/vitest"
-import { Effect, Redacted, String } from "effect"
+import { Effect, Fiber, Redacted, String } from "effect"
 import * as Chunk from "effect/Chunk"
 import * as Stream from "effect/Stream"
 import * as TestServices from "effect/TestServices"
@@ -344,3 +344,37 @@ it.layer(PgContainer.ClientFromPoolLive, { timeout: "30 seconds" })("PgClient fr
       expect(sql.config.database).toEqual(parsedConfig.database)
     }))
 })
+
+it.layer(PgContainer.ClientSingleConnectionLive, { timeout: "30 seconds" })("PgClient listen", (it) => {
+  it.scoped("listen does not reserve a pool connection", () =>
+    Effect.gen(function*() {
+      const sql = yield* PgClient.PgClient
+      const channel = "pool_connection_listen"
+
+      const listenFiber = yield* sql.listen(channel).pipe(
+        Stream.take(1),
+        Stream.runCollect,
+        Effect.forkScoped
+      )
+
+      yield* Effect.sleep("250 millis")
+
+      const rows = yield* sql<{ value: number }>`SELECT 1 as value`.pipe(
+        Effect.timeoutFail({
+          duration: "3 seconds",
+          onTimeout: () => new Error("query timed out while listener was active")
+        })
+      )
+      expect(rows).toEqual([{ value: 1 }])
+
+      yield* sql`SELECT pg_notify(${channel}, ${"payload"})`
+
+      const payloads = yield* Fiber.join(listenFiber).pipe(
+        Effect.timeoutFail({
+          duration: "3 seconds",
+          onTimeout: () => new Error("listener did not receive notification in time")
+        })
+      )
+      expect(Chunk.toReadonlyArray(payloads)).toEqual(["payload"])
+    }).pipe(TestServices.provideLive))
+})

packages/sql-pg/test/utils.ts

@@ -50,4 +50,14 @@ export class PgContainer extends Effect.Service<PgContainer>()("test/PgContainer
       })
     })
   ).pipe(Layer.provide(this.Default))
+
+  static ClientSingleConnectionLive = Layer.unwrapEffect(
+    Effect.gen(function*() {
+      const container = yield* PgContainer
+      return PgClient.layer({
+        url: Redacted.make(container.getConnectionUri()),
+        maxConnections: 1
+      })
+    })
+  ).pipe(Layer.provide(this.Default))
 }