fix(adapter-pg): serialize queries on single-connection clients

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'
@@ -153,3 +155,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 timers.setImmediate()
+      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()
+  })
+})

packages/adapter-pg/src/pg.ts

@@ -32,6 +32,15 @@ class PgQueryable<ClientT extends StdClient | TransactionClient> 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
+  // 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<void> = Promise.resolve()
+
   constructor(
     protected readonly client: ClientT,
     protected readonly pgOptions?: PrismaPgOptions,
@@ -99,6 +108,21 @@ class PgQueryable<ClientT extends StdClient | TransactionClient> implements SqlQ
    * marked as unhealthy.
    */
   private async performIO(query: SqlQuery): Promise<pg.QueryArrayResult<any>> {
+    if (!this.serializeQueries) {
+      return this.#performIO(query)
+    }
+    const previous = this.#queryLock
+    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<pg.QueryArrayResult<any>> {
     const { sql, args } = query
     const values = args.map((arg, i) => mapArg(arg, query.argTypes[i]))
 
@@ -197,6 +221,8 @@ export type UserDefinedTypeParser = (oid: number, value: unknown, adapter: SqlQu
 export type StatementNameGenerator = (query: SqlQuery) => string
 
 export class PrismaPgAdapter extends PgQueryable<StdClient> implements SqlDriverAdapter {
+  protected override readonly serializeQueries = false
+
   constructor(
     client: StdClient,
     protected readonly pgOptions?: PrismaPgOptions,

packages/client-engine-runtime/src/interpreter/query-interpreter.test.ts

@@ -1,3 +1,5 @@
+import timers from 'node:timers/promises'
+
 import {
   ColumnTypeEnum,
   SqlDriverAdapter,
@@ -95,6 +97,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 timers.setImmediate()
+      inFlight--
+      return userResultSet(1, 'Alice')
+    },
+    executeRaw: () => Promise.resolve(0),
+  }
+
+  const joinChild = (parentField: string): Extract<QueryPlanNode, { type: 'join' }>['args']['children'][number] => ({
+    child: queryNode(`SELECT * FROM ${parentField}`),
+    on: [['id', 'id']],
+    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