From fb6172d7c3dd7ac29702996fce287b343713078a Mon Sep 17 00:00:00 2001 From: Matin Gathani Date: Tue, 14 Apr 2026 19:05:15 -0700 Subject: [PATCH 1/5] fix(client-engine-runtime): serialize join children inside transactions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When `update` (or any write) with `include` relations is executed inside a transaction, the `join` node in the query interpreter fires a `Promise.all` over all child relation fetches. Each child eventually calls `queryRaw` on `context.queryable`, which — inside a transaction — is a single `pg.PoolClient` (not a pool). Concurrent `client.query()` calls on a single pg Client trigger: "Calling client.query() when the client is already executing a query is deprecated and will be removed in pg@9.0." This becomes a hard error in pg@9.0. Fix: detect whether `context.queryable` is a `Transaction` (via the `commit` property on the `Transaction` interface) and, if so, fetch the join children sequentially instead of in parallel. When using a connection pool the existing parallel behaviour is preserved. Fixes #29407 --- .../src/interpreter/query-interpreter.ts | 34 +++++++++++++++---- 1 file changed, 27 insertions(+), 7 deletions(-) diff --git a/packages/client-engine-runtime/src/interpreter/query-interpreter.ts b/packages/client-engine-runtime/src/interpreter/query-interpreter.ts index f63fd9cb271c..2955d67fc96d 100644 --- a/packages/client-engine-runtime/src/interpreter/query-interpreter.ts +++ b/packages/client-engine-runtime/src/interpreter/query-interpreter.ts @@ -1,4 +1,4 @@ -import { ConnectionInfo, SqlQuery, SqlQueryable, SqlResultSet } from '@prisma/driver-adapter-utils' +import { ConnectionInfo, SqlQuery, SqlQueryable, SqlResultSet, Transaction } from '@prisma/driver-adapter-utils' import type { SqlCommenterPlugin, SqlCommenterQueryInfo } from '@prisma/sqlcommenter' import { klona } from 'klona' @@ -240,12 +240,28 @@ export class QueryInterpreter { return { value: null, lastInsertId } } - const children = await Promise.all( - node.args.children.map(async (joinExpr) => ({ - joinExpr, - childRecords: (await this.interpretNode(joinExpr.child, context)).value, - })), - ) + // When running inside a transaction the queryable is a single database connection that + // cannot handle concurrent queries. Serialise child fetches to avoid the pg deprecation + // warning "Calling client.query() when the client is already executing a query", which + // becomes a hard error in pg@9.0. When using a pool (no transaction), concurrent fetches + // are safe and remain parallel for performance. + let children: { joinExpr: JoinExpression; childRecords: Value }[] + if (isTransaction(context.queryable)) { + children = [] + for (const joinExpr of node.args.children) { + children.push({ + joinExpr, + childRecords: (await this.interpretNode(joinExpr.child, context)).value, + }) + } + } else { + children = await Promise.all( + node.args.children.map(async (joinExpr) => ({ + joinExpr, + childRecords: (await this.interpretNode(joinExpr.child, context)).value, + })), + ) + } return { value: attachChildrenToParents(parent, children, node.args.canAssumeStrictEquality), lastInsertId } } @@ -604,3 +620,7 @@ function evaluateProcessingParameters( function cloneObject(value: T): DeepUnreadonly { return klona(value) as DeepUnreadonly } + +function isTransaction(queryable: SqlQueryable): queryable is Transaction { + return 'commit' in queryable +} From f7ee940297fd9baaf533de29b839a5fac9942d10 Mon Sep 17 00:00:00 2001 From: Matin Gathani Date: Wed, 15 Apr 2026 17:22:42 -0700 Subject: [PATCH 2/5] fix(adapter-pg): serialize concurrent transaction queries with a mutex pg.PoolClient does not support concurrent client.query() calls on a single connection. Inside a transaction all relation-join child queries were fired via Promise.all, hitting this limitation and triggering the pg deprecation warning (hard error in pg@9.0). Move the fix to the adapter layer, matching the pattern already used in adapter-mssql and adapter-planetscale: PgTransaction overrides performIO and gates each call behind a Mutex from async-mutex, so concurrent callers are serialised at the connection level without touching the runtime. Revert the query-interpreter.ts change from the previous commit; the runtime no longer needs to be aware of whether a queryable is a transaction. Fixes #29407 --- packages/adapter-pg/package.json | 1 + packages/adapter-pg/src/pg.ts | 16 ++++++++- .../src/interpreter/query-interpreter.ts | 34 ++++--------------- pnpm-lock.yaml | 15 ++++---- 4 files changed, 32 insertions(+), 34 deletions(-) diff --git a/packages/adapter-pg/package.json b/packages/adapter-pg/package.json index 18c99696018f..2440e6b6b993 100644 --- a/packages/adapter-pg/package.json +++ b/packages/adapter-pg/package.json @@ -37,6 +37,7 @@ "sideEffects": false, "dependencies": { "@prisma/driver-adapter-utils": "workspace:*", + "async-mutex": "0.5.0", "pg": "^8.16.3", "postgres-array": "3.0.4", "@types/pg": "^8.16.0" diff --git a/packages/adapter-pg/src/pg.ts b/packages/adapter-pg/src/pg.ts index e0e26c55ff45..064b78771a1f 100644 --- a/packages/adapter-pg/src/pg.ts +++ b/packages/adapter-pg/src/pg.ts @@ -13,6 +13,7 @@ import type { TransactionOptions, } from '@prisma/driver-adapter-utils' import { Debug, DriverAdapterError } from '@prisma/driver-adapter-utils' +import { Mutex } from 'async-mutex' // @ts-ignore: this is used to avoid the `Module '"/node_modules/@types/pg/index"' has no default export.` error. import pg from 'pg' @@ -98,7 +99,7 @@ class PgQueryable implements SqlQ * Should the query fail due to a connection error, the connection is * marked as unhealthy. */ - private async performIO(query: SqlQuery): Promise> { + protected async performIO(query: SqlQuery): Promise> { const { sql, args } = query const values = args.map((arg, i) => mapArg(arg, query.argTypes[i])) @@ -135,6 +136,10 @@ class PgQueryable implements SqlQ } class PgTransaction extends PgQueryable implements Transaction { + // pg.PoolClient does not support concurrent queries on the same connection, + // so we serialize all performIO calls with a mutex. + #mutex = new Mutex() + constructor( client: pg.PoolClient, readonly options: TransactionOptions, @@ -144,6 +149,15 @@ class PgTransaction extends PgQueryable implements Transactio super(client, pgOptions) } + protected async performIO(query: SqlQuery): Promise> { + const release = await this.#mutex.acquire() + try { + return await super.performIO(query) + } finally { + release() + } + } + async commit(): Promise { debug(`[js::commit]`) diff --git a/packages/client-engine-runtime/src/interpreter/query-interpreter.ts b/packages/client-engine-runtime/src/interpreter/query-interpreter.ts index 2955d67fc96d..f63fd9cb271c 100644 --- a/packages/client-engine-runtime/src/interpreter/query-interpreter.ts +++ b/packages/client-engine-runtime/src/interpreter/query-interpreter.ts @@ -1,4 +1,4 @@ -import { ConnectionInfo, SqlQuery, SqlQueryable, SqlResultSet, Transaction } from '@prisma/driver-adapter-utils' +import { ConnectionInfo, SqlQuery, SqlQueryable, SqlResultSet } from '@prisma/driver-adapter-utils' import type { SqlCommenterPlugin, SqlCommenterQueryInfo } from '@prisma/sqlcommenter' import { klona } from 'klona' @@ -240,28 +240,12 @@ export class QueryInterpreter { return { value: null, lastInsertId } } - // When running inside a transaction the queryable is a single database connection that - // cannot handle concurrent queries. Serialise child fetches to avoid the pg deprecation - // warning "Calling client.query() when the client is already executing a query", which - // becomes a hard error in pg@9.0. When using a pool (no transaction), concurrent fetches - // are safe and remain parallel for performance. - let children: { joinExpr: JoinExpression; childRecords: Value }[] - if (isTransaction(context.queryable)) { - children = [] - for (const joinExpr of node.args.children) { - children.push({ - joinExpr, - childRecords: (await this.interpretNode(joinExpr.child, context)).value, - }) - } - } else { - children = await Promise.all( - node.args.children.map(async (joinExpr) => ({ - joinExpr, - childRecords: (await this.interpretNode(joinExpr.child, context)).value, - })), - ) - } + const children = await Promise.all( + node.args.children.map(async (joinExpr) => ({ + joinExpr, + childRecords: (await this.interpretNode(joinExpr.child, context)).value, + })), + ) return { value: attachChildrenToParents(parent, children, node.args.canAssumeStrictEquality), lastInsertId } } @@ -620,7 +604,3 @@ function evaluateProcessingParameters( function cloneObject(value: T): DeepUnreadonly { return klona(value) as DeepUnreadonly } - -function isTransaction(queryable: SqlQueryable): queryable is Transaction { - return 'commit' in queryable -} diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index a64018eaf999..c310af08c7ae 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -308,6 +308,9 @@ importers: '@types/pg': specifier: ^8.16.0 version: 8.20.0 + async-mutex: + specifier: 0.5.0 + version: 0.5.0 pg: specifier: ^8.16.3 version: 8.16.3 @@ -1901,7 +1904,7 @@ importers: dependencies: '@ark/attest': specifier: 0.48.2 - version: 0.48.2(typescript@5.8.2) + version: 0.48.2(typescript@5.4.5) '@prisma/client': specifier: workspace:* version: link:../client @@ -8804,16 +8807,16 @@ snapshots: '@antfu/ni@0.21.12': {} - '@ark/attest@0.48.2(typescript@5.8.2)': + '@ark/attest@0.48.2(typescript@5.4.5)': dependencies: '@ark/fs': 0.46.0 '@ark/util': 0.46.0 '@prettier/sync': 0.5.5(prettier@3.5.3) '@typescript/analyze-trace': 0.10.1 - '@typescript/vfs': 1.6.1(typescript@5.8.2) + '@typescript/vfs': 1.6.1(typescript@5.4.5) arktype: 2.1.20 prettier: 3.5.3 - typescript: 5.8.2 + typescript: 5.4.5 transitivePeerDependencies: - supports-color @@ -10878,7 +10881,7 @@ snapshots: '@swc-node/sourcemap-support@0.5.1': dependencies: source-map-support: 0.5.21 - tslib: 2.6.3 + tslib: 2.8.1 '@swc/core-darwin-arm64@1.11.5': optional: true @@ -11310,7 +11313,7 @@ snapshots: treeify: 1.1.0 yargs: 16.2.0 - '@typescript/vfs@1.6.1(typescript@5.8.2)': + '@typescript/vfs@1.6.1(typescript@5.4.5)': dependencies: debug: 4.4.3 typescript: 5.8.2 From 0be956369e477d854f8fee1e551386c2ec0bdad3 Mon Sep 17 00:00:00 2001 From: Matin Gathani Date: Thu, 16 Apr 2026 22:45:41 -0700 Subject: [PATCH 3/5] refactor(adapter-pg): simplify mutex to runExclusive per CodeRabbit MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace the manual acquire/try/finally/release pattern in PgTransaction.performIO with async-mutex's built-in `runExclusive`, which handles acquire and release internally. Semantically identical to the previous implementation — serialises all performIO calls on the underlying pg.PoolClient — but removes boilerplate and avoids any risk of a missed release on unexpected throws. Addresses CodeRabbit suggestion on PR #29468. --- packages/adapter-pg/src/pg.ts | 7 +------ 1 file changed, 1 insertion(+), 6 deletions(-) diff --git a/packages/adapter-pg/src/pg.ts b/packages/adapter-pg/src/pg.ts index 064b78771a1f..4a8aaf478672 100644 --- a/packages/adapter-pg/src/pg.ts +++ b/packages/adapter-pg/src/pg.ts @@ -150,12 +150,7 @@ class PgTransaction extends PgQueryable implements Transactio } protected async performIO(query: SqlQuery): Promise> { - const release = await this.#mutex.acquire() - try { - return await super.performIO(query) - } finally { - release() - } + return this.#mutex.runExclusive(() => super.performIO(query)) } async commit(): Promise { From 91ab9904dce2c0629903e3cef44946e855ab45b2 Mon Sep 17 00:00:00 2001 From: Alexey Orlenko's AI Agent Date: Fri, 24 Jul 2026 00:34:53 +0200 Subject: [PATCH 4/5] test(adapter-pg): cover transaction query serialization and mutex release on error --- packages/adapter-pg/src/__tests__/pg.test.ts | 58 ++++++++++++++++++++ 1 file changed, 58 insertions(+) diff --git a/packages/adapter-pg/src/__tests__/pg.test.ts b/packages/adapter-pg/src/__tests__/pg.test.ts index 451005039372..7c02f75ec2e0 100644 --- a/packages/adapter-pg/src/__tests__/pg.test.ts +++ b/packages/adapter-pg/src/__tests__/pg.test.ts @@ -134,6 +134,64 @@ describe('PrismaPgAdapterFactory', () => { await adapter.dispose() }) + it('should serialize concurrent queries within a transaction', async () => { + const config: pg.PoolConfig = { user: 'test', password: 'test', database: 'test', port: 5432, host: 'localhost' } + const factory = new PrismaPgAdapterFactory(config) + const adapter = await factory.connect() + + let inFlight = 0 + let maxInFlight = 0 + const mockConnection = { + on: vi.fn(), + removeListener: vi.fn(), + query: vi.fn(async () => { + inFlight++ + maxInFlight = Math.max(maxInFlight, inFlight) + await new Promise((resolve) => setTimeout(resolve, 10)) + inFlight-- + return { rows: [], fields: [], rowCount: 0 } + }), + release: vi.fn(), + listenerCount: vi.fn().mockReturnValue(0), + } + adapter['client'].connect = vi.fn().mockResolvedValue(mockConnection) + + const transaction = await adapter.startTransaction() + const query: SqlQuery = { sql: 'SELECT 1', args: [], argTypes: [] } + await Promise.all([transaction.queryRaw(query), transaction.queryRaw(query), transaction.queryRaw(query)]) + + expect(maxInFlight).toBe(1) + expect(mockConnection.query).toHaveBeenCalledTimes(4) // BEGIN + 3 queries + + await transaction.commit() + await adapter.dispose() + }) + + it('should release the transaction mutex when a query fails', async () => { + const config: pg.PoolConfig = { user: 'test', password: 'test', database: 'test', port: 5432, host: 'localhost' } + const factory = new PrismaPgAdapterFactory(config) + const adapter = await factory.connect() + + const mockConnection = { + on: vi.fn(), + removeListener: vi.fn(), + query: vi.fn().mockResolvedValue({ rows: [], fields: [], rowCount: 0 }), + release: vi.fn(), + listenerCount: vi.fn().mockReturnValue(0), + } + adapter['client'].connect = vi.fn().mockResolvedValue(mockConnection) + + const transaction = await adapter.startTransaction() + mockConnection.query.mockRejectedValueOnce(new Error('boom')) + + const query: SqlQuery = { sql: 'SELECT 1', args: [], argTypes: [] } + await expect(transaction.queryRaw(query)).rejects.toThrow() + await expect(transaction.queryRaw(query)).resolves.toBeDefined() + + await transaction.rollback() + await adapter.dispose() + }) + it('should not pass name when statement name generator is not provided', async () => { const factory = new PrismaPgAdapterFactory('postgresql://test:test@localhost/test') const adapter = await factory.connect() From 0f4ad1d6b52b5537d871e51fc5c1e374de598f9e Mon Sep 17 00:00:00 2001 From: Alexey Orlenko's AI Agent Date: Fri, 24 Jul 2026 00:38:07 +0200 Subject: [PATCH 5/5] chore(adapter-pg): add async-mutex to lockfile --- pnpm-lock.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 92c76104b4cf..7f46d3c62985 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -11151,7 +11151,7 @@ snapshots: '@swc-node/sourcemap-support@0.5.1': dependencies: source-map-support: 0.5.21 - tslib: 2.8.1 + tslib: 2.6.3 '@swc/core-darwin-arm64@1.11.5': optional: true