fix(adapter-pg): release the pool client at most once per transaction

packages/adapter-pg/src/__tests__/pg.test.ts

@@ -1,7 +1,7 @@
 import { getLogs } from '@prisma/debug'
 import type { SqlQuery } from '@prisma/driver-adapter-utils'
 import pg, { DatabaseError } from 'pg'
-import { describe, expect, it, vi } from 'vitest'
+import { afterEach, describe, expect, it, vi } from 'vitest'
 
 import { PrismaPgAdapterFactory } from '../pg'
 
@@ -153,3 +153,120 @@ describe('PrismaPgAdapterFactory', () => {
     await adapter.dispose()
   })
 })
+
+describe('PgTransaction', () => {
+  // Regression tests for https://github.com/prisma/prisma/issues/29952: when an
+  // interactive transaction timeout expires while COMMIT is in flight, the query
+  // engine settles the transaction twice (the abandoned commit chain and the
+  // compensating rollback), which must not release the pooled client twice or
+  // dispatch SQL on a client the pool may have re-lent.
+  //
+  // No database is needed: pg.Client's connect/query/end are stubbed and the
+  // engine's settlement sequences are scripted directly against the adapter.
+
+  const statements: string[] = []
+  const pools: pg.Pool[] = []
+
+  afterEach(async () => {
+    await Promise.all(pools.map((pool) => pool.end()))
+    pools.length = 0
+    vi.restoreAllMocks()
+  })
+
+  async function setup(queryImpl?: (text: string) => Promise<unknown> | undefined) {
+    statements.length = 0
+
+    vi.spyOn(pg.Client.prototype, 'connect').mockImplementation(function (cb?: (err?: Error) => void) {
+      process.nextTick(() => cb?.())
+    } as never)
+    vi.spyOn(pg.Client.prototype, 'end').mockImplementation((() => Promise.resolve()) as never)
+    vi.spyOn(pg.Client.prototype, 'query').mockImplementation(((query: string | { text: string }) => {
+      const text = typeof query === 'string' ? query : query.text
+      statements.push(text)
+      return queryImpl?.(text) ?? Promise.resolve({ rowCount: 0, rows: [], fields: [] })
+    }) as never)
+
+    const pool = new pg.Pool({
+      connectionString: 'postgresql://user:pass@localhost:5432/db',
+      max: 1,
+    })
+    pools.push(pool)
+    const releases: unknown[] = []
+    pool.on('release', (err) => releases.push(err))
+    const adapter = await new PrismaPgAdapterFactory(pool).connect()
+
+    return { pool, adapter, releases }
+  }
+
+  it('should release the client exactly once when COMMIT is sent before commit()', async () => {
+    const { pool, adapter, releases } = await setup()
+    const tx = await adapter.startTransaction()
+
+    await tx.executeRaw({ sql: 'COMMIT', args: [], argTypes: [] })
+    await tx.commit()
+
+    expect(statements).toEqual(['BEGIN', 'COMMIT'])
+    expect(releases).toEqual([undefined])
+    expect(pool.idleCount).toBe(1)
+    expect(pool.totalCount).toBe(1)
+  })
+
+  it('should treat the second settlement as a no-op instead of releasing twice', async () => {
+    const { pool, adapter, releases } = await setup()
+    const tx = await adapter.startTransaction()
+
+    await tx.commit()
+    await expect(tx.rollback()).resolves.toBeUndefined()
+
+    expect(releases).toEqual([undefined])
+    expect(pool.idleCount).toBe(1)
+    expect(pool.totalCount).toBe(1)
+  })
+
+  it('should not release a client the pool has re-lent to a new owner', async () => {
+    const { pool, adapter } = await setup()
+    const tx = await adapter.startTransaction()
+
+    // The engine's compensation sequence when the transaction deadline drops an
+    // in-flight commit: rollback wins first, the pool re-lends the client, then
+    // the orphaned commit chain lands late.
+    await tx.rollback()
+    const nextOwner = await pool.connect()
+    await tx.commit()
+
+    expect(() => nextOwner.release()).not.toThrow()
+    expect(pool.idleCount).toBe(1)
+    expect(pool.totalCount).toBe(1)
+  })
+
+  it('should reject queries after settlement without dispatching SQL', async () => {
+    const { adapter } = await setup()
+    const tx = await adapter.startTransaction()
+
+    await tx.rollback()
+    statements.length = 0
+
+    await expect(tx.executeRaw({ sql: 'ROLLBACK', args: [], argTypes: [] })).rejects.toMatchObject({
+      name: 'DriverAdapterError',
+      cause: { kind: 'TransactionAlreadyClosed' },
+    })
+    await expect(tx.queryRaw({ sql: 'SELECT 1', args: [], argTypes: [] })).rejects.toMatchObject({
+      name: 'DriverAdapterError',
+      cause: { kind: 'TransactionAlreadyClosed' },
+    })
+    expect(statements).toEqual([])
+  })
+
+  it('should destroy the connection when settling with a query still in flight', async () => {
+    const { pool, adapter, releases } = await setup((text) =>
+      text === 'SELECT pending' ? new Promise(() => {}) : undefined,
+    )
+    const tx = await adapter.startTransaction()
+
+    void tx.executeRaw({ sql: 'SELECT pending', args: [], argTypes: [] })
+    await tx.rollback()
+
+    expect(releases).toEqual([expect.any(Error)])
+    expect(pool.totalCount).toBe(0)
+  })
+})

packages/adapter-pg/src/pg.ts

@@ -98,7 +98,7 @@ class PgQueryable<ClientT extends StdClient | TransactionClient> implements SqlQ
    * Should the query fail due to a connection error, the connection is
    * marked as unhealthy.
    */
-  private async performIO(query: SqlQuery): Promise<pg.QueryArrayResult<any>> {
+  protected async performIO(query: SqlQuery): Promise<pg.QueryArrayResult<any>> {
     const { sql, args } = query
     const values = args.map((arg, i) => mapArg(arg, query.argTypes[i]))
 
@@ -132,6 +132,9 @@ class PgQueryable<ClientT extends StdClient | TransactionClient> implements SqlQ
 }
 
 class PgTransaction extends PgQueryable<TransactionClient> implements Transaction {
+  private settled = false
+  private inFlightQueries = 0
+
   constructor(
     client: pg.PoolClient,
     readonly options: TransactionOptions,
@@ -141,18 +144,58 @@ class PgTransaction extends PgQueryable<TransactionClient> implements Transactio
     super(client, pgOptions)
   }
 
+  protected override async performIO(query: SqlQuery): Promise<pg.QueryArrayResult<any>> {
+    if (this.settled) {
+      throw new DriverAdapterError({
+        kind: 'TransactionAlreadyClosed',
+        cause: 'The transaction has already been committed or rolled back',
+      })
+    }
+
+    this.inFlightQueries++
+    try {
+      return await super.performIO(query)
+    } finally {
+      this.inFlightQueries--
+    }
+  }
+
+  /**
+   * Settles the transaction at most once. The query engine can settle twice when
+   * an interactive transaction timeout expires while COMMIT is in flight: the
+   * abandoned commit chain and the compensating rollback both reach the adapter
+   * (https://github.com/prisma/prisma/issues/29952). A second `release()` would
+   * corrupt pg-pool's accounting once the pool has re-lent the client, so it is
+   * skipped, and a client with statements still in flight is released with an
+   * error so the pool destroys the connection instead of re-lending a busy one.
+   */
+  private settle(): void {
+    if (this.settled) {
+      return
+    }
+    this.settled = true
+
+    try {
+      this.cleanup?.()
+    } finally {
+      if (this.inFlightQueries > 0) {
+        this.client.release(new Error('Transaction settled with statements still in flight'))
+      } else {
+        this.client.release()
+      }
+    }
+  }
+
   async commit(): Promise<void> {
     debug(`[js::commit]`)
 
-    this.cleanup?.()
-    this.client.release()
+    this.settle()
   }
 
   async rollback(): Promise<void> {
     debug(`[js::rollback]`)
 
-    this.cleanup?.()
-    this.client.release()
+    this.settle()
   }
 
   async createSavepoint(name: string): Promise<void> {