diff --git a/.changeset/eff-428-pg-transaction-permit.md b/.changeset/eff-428-pg-transaction-permit.md new file mode 100644 index 00000000000..be6a9a20e9f --- /dev/null +++ b/.changeset/eff-428-pg-transaction-permit.md @@ -0,0 +1,5 @@ +--- +"@effect/sql-pg": patch +--- + +Hold the shared PostgreSQL client permit for the full transaction lifetime. diff --git a/packages/sql/pg/src/PgClient.ts b/packages/sql/pg/src/PgClient.ts index 4e33ab77a8d..9003a925aec 100644 --- a/packages/sql/pg/src/PgClient.ts +++ b/packages/sql/pg/src/PgClient.ts @@ -522,6 +522,17 @@ export const fromClient = Effect.fnUntraced(function*( ) const connection = makeConection(client) const acquirer = semaphore.withPermit(Effect.succeed(connection)) + const transactionAcquirer = Effect.uninterruptibleMask((restore) => { + const fiber = Fiber.getCurrent()! + const scope = Context.getUnsafe(fiber.context, Scope.Scope) + return Effect.as( + Effect.tap( + restore(semaphore.take(1)), + () => Scope.addFinalizer(scope, semaphore.release(1)) + ), + connection + ) + }) const config: PgClientConfig = { ...options, @@ -535,7 +546,7 @@ export const fromClient = Effect.fnUntraced(function*( return yield* makeWith({ acquirer, - transactionAcquirer: acquirer, + transactionAcquirer, listenAcquirer: streamClient, config, spanAttributes: options.spanAttributes, diff --git a/packages/sql/pg/test/Client.integration.test.ts b/packages/sql/pg/test/Client.integration.test.ts index 78b4ed2c111..4db2866fffd 100644 --- a/packages/sql/pg/test/Client.integration.test.ts +++ b/packages/sql/pg/test/Client.integration.test.ts @@ -1,7 +1,8 @@ import { PgClient } from "@effect/sql-pg" import { assert, expect, it } from "@effect/vitest" -import { Deferred, Effect, Fiber, Redacted, Stream, String } from "effect" +import { Deferred, Effect, Fiber, Option, Redacted, Stream, String } from "effect" import { TestClock } from "effect/testing" +import * as Reactivity from "effect/unstable/reactivity/Reactivity" import { SqlClient } from "effect/unstable/sql" import * as Statement from "effect/unstable/sql/Statement" import { parse as parsePgConnectionString } from "pg-connection-string" @@ -413,3 +414,51 @@ it.layer(PgContainer.layerClientSingleConnection, { timeout: "30 seconds" })("Pg expect(Array.from(payloads)).toEqual(["payload"]) }).pipe(TestClock.withLive), 20_000) }) + +it.effect("serializes transactions that share one pg.Client", () => + Effect.gen(function*() { + const secondBegin = yield* Deferred.make() + const firstBodyStarted = yield* Deferred.make() + const releaseFirstBody = yield* Deferred.make() + let beginCalls = 0 + const pg = { + host: "localhost", + port: 5432, + database: "postgres", + user: "postgres", + password: undefined, + ssl: false, + on() {}, + off() {}, + query(sql: string, _params: ReadonlyArray, callback: (error: null, result: unknown) => void) { + if (sql === "BEGIN" && ++beginCalls === 2) { + Deferred.doneUnsafe(secondBegin, Effect.void) + } + callback(null, { rows: [] }) + } + } + + const sql = yield* PgClient.fromClient({ + acquire: Effect.succeed(pg as any), + acquireForStream: false + }) + const first = yield* sql.withTransaction(Effect.gen(function*() { + yield* Deferred.succeed(firstBodyStarted, undefined) + yield* Deferred.await(releaseFirstBody) + })).pipe(Effect.forkScoped) + yield* Deferred.await(firstBodyStarted) + const second = yield* sql.withTransaction(Effect.void).pipe(Effect.forkScoped) + + const overlap = yield* Deferred.await(secondBegin).pipe( + Effect.timeoutOption("100 millis"), + TestClock.withLive + ) + yield* Deferred.succeed(releaseFirstBody, undefined) + yield* Fiber.join(first) + yield* Fiber.join(second) + + assert.isTrue(Option.isNone(overlap)) + }).pipe( + Effect.scoped, + Effect.provide(Reactivity.layer) + ))