From d4e3668efb1bbbb8f48c7538a3efb20ddc70aa1b Mon Sep 17 00:00:00 2001 From: Camille Barneaud <1693643+gadcam@users.noreply.github.com> Date: Wed, 12 Aug 2026 00:02:36 +0200 Subject: [PATCH 1/2] fix(adapter-pg): serialize queries on single-connection clients A pg.PoolClient is a single connection and does not support concurrent queries (deprecated in pg@8, an error in pg@9), but the query interpreter loads sibling relations concurrently, so any query with 2+ relations inside a transaction triggered the deprecation warning. Serialize performIO in PgQueryable for single-connection clients, leaving the pool path parallel. Supersedes #29468, fixes #29407. Co-Authored-By: Claude Fable 5 Signed-off-by: Camille Barneaud <1693643+gadcam@users.noreply.github.com> --- packages/adapter-pg/src/__tests__/pg.test.ts | 79 +++++++++++++++++++ packages/adapter-pg/src/pg.ts | 21 +++++ .../src/interpreter/query-interpreter.test.ts | 41 ++++++++++ 3 files changed, 141 insertions(+) diff --git a/packages/adapter-pg/src/__tests__/pg.test.ts b/packages/adapter-pg/src/__tests__/pg.test.ts index 451005039372..2de34b76a8f8 100644 --- a/packages/adapter-pg/src/__tests__/pg.test.ts +++ b/packages/adapter-pg/src/__tests__/pg.test.ts @@ -153,3 +153,82 @@ describe('PrismaPgAdapterFactory', () => { await adapter.dispose() }) }) + +describe('query serialization', () => { + const query = (sql: string): SqlQuery => ({ sql, args: [], argTypes: [] }) + const emptyResult = { rows: [], fields: [], rowCount: 0 } + + function trackingQueryMock() { + let inFlight = 0 + let maxInFlight = 0 + const started: string[] = [] + const mock = vi.fn(async ({ text }: { text: string }) => { + started.push(text) + maxInFlight = Math.max(maxInFlight, ++inFlight) + await new Promise((resolve) => setImmediate(resolve)) + inFlight-- + return emptyResult + }) + return { mock, started, maxInFlight: () => maxInFlight } + } + + async function connectedAdapter() { + const factory = new PrismaPgAdapterFactory('postgresql://test:test@localhost:5432/test') + return await factory.connect() + } + + it('serializes concurrent queries on a transaction connection', async () => { + const adapter = await connectedAdapter() + const { mock, maxInFlight } = trackingQueryMock() + const mockConnection = { on: vi.fn(), removeListener: vi.fn(), query: mock, release: vi.fn() } + adapter['client'].connect = vi.fn().mockResolvedValue(mockConnection) + + const transaction = await adapter.startTransaction() + await Promise.all([ + transaction.queryRaw(query('SELECT 1')), + transaction.queryRaw(query('SELECT 2')), + transaction.queryRaw(query('SELECT 3')), + ]) + + // A pg.PoolClient is a single connection: queries must never overlap. + expect(maxInFlight()).toBe(1) + await transaction.commit() + await adapter.dispose() + }) + + it('does not serialize queries on the pool', async () => { + const adapter = await connectedAdapter() + const { mock, maxInFlight } = trackingQueryMock() + adapter['client'].query = mock + + await Promise.all([ + adapter.queryRaw(query('SELECT 1')), + adapter.queryRaw(query('SELECT 2')), + adapter.queryRaw(query('SELECT 3')), + ]) + + // The pool handles concurrency itself; serializing here would limit the + // whole application to one query at a time. + expect(maxInFlight()).toBe(3) + await adapter.dispose() + }) + + it('keeps serializing after a failed query', async () => { + const adapter = await connectedAdapter() + const mock = vi + .fn() + .mockResolvedValueOnce(emptyResult) // BEGIN + .mockRejectedValueOnce(new Error('boom')) + .mockResolvedValue(emptyResult) + const mockConnection = { on: vi.fn(), removeListener: vi.fn(), query: mock, release: vi.fn() } + adapter['client'].connect = vi.fn().mockResolvedValue(mockConnection) + + const tx = await adapter.startTransaction() + const failing = tx.queryRaw(query('SELECT 1')) + const following = tx.queryRaw(query('SELECT 2')) + + await expect(failing).rejects.toThrow() + await expect(following).resolves.toBeDefined() + await adapter.dispose() + }) +}) diff --git a/packages/adapter-pg/src/pg.ts b/packages/adapter-pg/src/pg.ts index 6a677bdad614..0c1629090fab 100644 --- a/packages/adapter-pg/src/pg.ts +++ b/packages/adapter-pg/src/pg.ts @@ -32,6 +32,13 @@ class PgQueryable implements SqlQ readonly provider = 'postgres' readonly adapterName = packageName + // `pg.Client` and `pg.PoolClient` are single connections and don't support + // concurrent queries (deprecated in pg@8, an error in pg@9), so queries must + // be serialized. `pg.Pool` handles concurrency itself and must not be + // serialized, or the whole pool would be limited to one query at a time. + protected readonly serializeQueries: boolean = true + #queryLock: Promise = Promise.resolve() + constructor( protected readonly client: ClientT, protected readonly pgOptions?: PrismaPgOptions, @@ -99,6 +106,18 @@ class PgQueryable implements SqlQ * marked as unhealthy. */ private async performIO(query: SqlQuery): Promise> { + if (!this.serializeQueries) { + return this.#performIO(query) + } + const previous = this.#queryLock + const current = previous.then(() => this.#performIO(query)) + // Keep the lock chain alive even if the query fails; the failure still + // propagates to the caller through `current`. + this.#queryLock = current.catch(() => {}) + return current + } + + async #performIO(query: SqlQuery): Promise> { const { sql, args } = query const values = args.map((arg, i) => mapArg(arg, query.argTypes[i])) @@ -197,6 +216,8 @@ export type UserDefinedTypeParser = (oid: number, value: unknown, adapter: SqlQu export type StatementNameGenerator = (query: SqlQuery) => string export class PrismaPgAdapter extends PgQueryable implements SqlDriverAdapter { + protected override readonly serializeQueries = false + constructor( client: StdClient, protected readonly pgOptions?: PrismaPgOptions, diff --git a/packages/client-engine-runtime/src/interpreter/query-interpreter.test.ts b/packages/client-engine-runtime/src/interpreter/query-interpreter.test.ts index b47b2fcdc10a..a49d48bb4c2a 100644 --- a/packages/client-engine-runtime/src/interpreter/query-interpreter.test.ts +++ b/packages/client-engine-runtime/src/interpreter/query-interpreter.test.ts @@ -95,6 +95,47 @@ test('merges chunked query results without overflowing the stack', async () => { expect(result).toHaveLength(rowsPerLaterChunk) }) +// Loading sibling relations concurrently is intentional: adapters whose connection +// cannot run queries concurrently (e.g. a single pg connection) are responsible for +// serializing them in `performIO` (see https://github.com/prisma/prisma/issues/29407). +// This pins the interpreter side of that contract so join loading stays parallel. +test('loads join children in parallel', async () => { + let inFlight = 0 + let maxInFlight = 0 + const queryable: SqlQueryable = { + provider: 'postgres', + adapterName: 'test', + queryRaw: async () => { + maxInFlight = Math.max(maxInFlight, ++inFlight) + await new Promise((resolve) => setImmediate(resolve)) + inFlight-- + return userResultSet(1, 'Alice') + }, + executeRaw: () => Promise.resolve(0), + } + + const joinChild = (parentField: string) => ({ + child: queryNode(`SELECT * FROM ${parentField}`), + on: [['id', 'id']] as [string, string][], + parentField, + isRelationUnique: true, + }) + + const queryPlan: QueryPlanNode = { + type: 'join', + args: { + parent: queryNode('SELECT * FROM users'), + children: [joinChild('posts'), joinChild('profile'), joinChild('settings')], + canAssumeStrictEquality: true, + }, + } + + const interpreter = QueryInterpreter.forSql({ tracingHelper: noopTracingHelper }) + await interpreter.run(queryPlan, { queryable, transactionManager: { enabled: false }, scope: {} }) + + expect(maxInFlight).toBe(3) +}) + class MockTransactionAdapter implements SqlDriverAdapter { adapterName = 'mock-adapter' provider = 'postgres' as const From 9adde528713986014da2d1b0a573a116a255d4c2 Mon Sep 17 00:00:00 2001 From: Camille Barneaud <1693643+gadcam@users.noreply.github.com> Date: Thu, 13 Aug 2026 13:18:14 +0200 Subject: [PATCH 2/2] refactor(adapter-pg): make query lock resolve-only, address review nits Restructure the serialization lock so it can never carry a query's error: the lock promise only signals completion, and a failed query rejects its own caller only. Use timers.setImmediate() in tests and derive the join-child type from the query plan instead of asserting. Co-Authored-By: Claude Fable 5 Signed-off-by: Camille Barneaud <1693643+gadcam@users.noreply.github.com> --- packages/adapter-pg/src/__tests__/pg.test.ts | 4 +++- packages/adapter-pg/src/pg.ts | 17 +++++++++++------ .../src/interpreter/query-interpreter.test.ts | 8 +++++--- 3 files changed, 19 insertions(+), 10 deletions(-) diff --git a/packages/adapter-pg/src/__tests__/pg.test.ts b/packages/adapter-pg/src/__tests__/pg.test.ts index 2de34b76a8f8..754753d80c5d 100644 --- a/packages/adapter-pg/src/__tests__/pg.test.ts +++ b/packages/adapter-pg/src/__tests__/pg.test.ts @@ -1,3 +1,5 @@ +import timers from 'node:timers/promises' + import { getLogs } from '@prisma/debug' import type { SqlQuery } from '@prisma/driver-adapter-utils' import pg, { DatabaseError } from 'pg' @@ -165,7 +167,7 @@ describe('query serialization', () => { const mock = vi.fn(async ({ text }: { text: string }) => { started.push(text) maxInFlight = Math.max(maxInFlight, ++inFlight) - await new Promise((resolve) => setImmediate(resolve)) + await timers.setImmediate() inFlight-- return emptyResult }) diff --git a/packages/adapter-pg/src/pg.ts b/packages/adapter-pg/src/pg.ts index 0c1629090fab..b0d7244eeb86 100644 --- a/packages/adapter-pg/src/pg.ts +++ b/packages/adapter-pg/src/pg.ts @@ -37,7 +37,9 @@ class PgQueryable implements SqlQ // be serialized. `pg.Pool` handles concurrency itself and must not be // serialized, or the whole pool would be limited to one query at a time. protected readonly serializeQueries: boolean = true - #queryLock: Promise = Promise.resolve() + // Resolve-only lock: it tracks completion of the previous query and can + // never carry its error, so a failed query rejects only its own caller. + #queryLock: Promise = Promise.resolve() constructor( protected readonly client: ClientT, @@ -110,11 +112,14 @@ class PgQueryable implements SqlQ return this.#performIO(query) } const previous = this.#queryLock - const current = previous.then(() => this.#performIO(query)) - // Keep the lock chain alive even if the query fails; the failure still - // propagates to the caller through `current`. - this.#queryLock = current.catch(() => {}) - return current + let release!: () => void + this.#queryLock = new Promise((resolve) => (release = resolve)) + await previous + try { + return await this.#performIO(query) + } finally { + release() + } } async #performIO(query: SqlQuery): Promise> { diff --git a/packages/client-engine-runtime/src/interpreter/query-interpreter.test.ts b/packages/client-engine-runtime/src/interpreter/query-interpreter.test.ts index a49d48bb4c2a..0ee1f5713c7d 100644 --- a/packages/client-engine-runtime/src/interpreter/query-interpreter.test.ts +++ b/packages/client-engine-runtime/src/interpreter/query-interpreter.test.ts @@ -1,3 +1,5 @@ +import timers from 'node:timers/promises' + import { ColumnTypeEnum, SqlDriverAdapter, @@ -107,16 +109,16 @@ test('loads join children in parallel', async () => { adapterName: 'test', queryRaw: async () => { maxInFlight = Math.max(maxInFlight, ++inFlight) - await new Promise((resolve) => setImmediate(resolve)) + await timers.setImmediate() inFlight-- return userResultSet(1, 'Alice') }, executeRaw: () => Promise.resolve(0), } - const joinChild = (parentField: string) => ({ + const joinChild = (parentField: string): Extract['args']['children'][number] => ({ child: queryNode(`SELECT * FROM ${parentField}`), - on: [['id', 'id']] as [string, string][], + on: [['id', 'id']], parentField, isRelationUnique: true, })