diff --git a/packages/adapter-pg/src/__tests__/pg.test.ts b/packages/adapter-pg/src/__tests__/pg.test.ts index 451005039372..793911b3039c 100644 --- a/packages/adapter-pg/src/__tests__/pg.test.ts +++ b/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 | 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) + }) +}) diff --git a/packages/adapter-pg/src/pg.ts b/packages/adapter-pg/src/pg.ts index 6a677bdad614..e0a9db68fe9c 100644 --- a/packages/adapter-pg/src/pg.ts +++ b/packages/adapter-pg/src/pg.ts @@ -98,7 +98,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])) @@ -132,6 +132,9 @@ class PgQueryable implements SqlQ } class PgTransaction extends PgQueryable implements Transaction { + private settled = false + private inFlightQueries = 0 + constructor( client: pg.PoolClient, readonly options: TransactionOptions, @@ -141,18 +144,58 @@ class PgTransaction extends PgQueryable implements Transactio super(client, pgOptions) } + protected override async performIO(query: SqlQuery): Promise> { + 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 { debug(`[js::commit]`) - this.cleanup?.() - this.client.release() + this.settle() } async rollback(): Promise { debug(`[js::rollback]`) - this.cleanup?.() - this.client.release() + this.settle() } async createSavepoint(name: string): Promise {