diff --git a/lib/interfaces/sql-executor.interface.ts b/lib/interfaces/sql-executor.interface.ts index 7f633ee..6951dcd 100644 --- a/lib/interfaces/sql-executor.interface.ts +++ b/lib/interfaces/sql-executor.interface.ts @@ -114,7 +114,9 @@ export interface SqlExecutor { execute?(text: string, params?: readonly unknown[]): Promise; /** * Runs `work` in a transaction of its own, on one connection: commits when `work` resolves, rolls back and rethrows - * when it rejects. + * when it rejects. On PostgreSQL, which ends a transaction at its first failed statement, `work` that caught a + * statement's error and resolved without a `ROLLBACK TO SAVEPOINT` rolls back and rejects as well: nothing of it + * could commit. */ transaction(work: (transaction: SqlTransaction) => Promise, options?: SqlTransactionOptions): Promise; /** diff --git a/lib/postgres/executors/drizzle.executor.ts b/lib/postgres/executors/drizzle.executor.ts index 45953b5..8164a69 100644 --- a/lib/postgres/executors/drizzle.executor.ts +++ b/lib/postgres/executors/drizzle.executor.ts @@ -1,6 +1,7 @@ import type { SqlExecuteResult, SqlExecutor, SqlIsolationLevel, SqlTransaction, SqlTransactionOptions } from '../../interfaces/sql-executor.interface.js'; import { notATransaction } from '../../sql/not-a-transaction.js'; import { describeValue, hasMethod, isolationSql } from '../../utils/executor.util.js'; +import { runTransactionWork } from './transaction-work.js'; /** The part of a Drizzle PostgreSQL database (or its `tx`) the executor uses. */ export interface DrizzleDatabaseLike { @@ -54,7 +55,7 @@ class DrizzleExecutor implements SqlExecutor<'postgres'> { if (config) { isolationSql(config.isolationLevel); } - return this.db.transaction((tx) => work(drizzleTransaction(tx)), config); + return this.db.transaction((tx) => runTransactionWork(work, drizzleTransaction(tx)), config); } wrapTransaction(transaction: unknown): SqlTransaction { diff --git a/lib/postgres/executors/kysely.executor.ts b/lib/postgres/executors/kysely.executor.ts index 8952901..c0efebb 100644 --- a/lib/postgres/executors/kysely.executor.ts +++ b/lib/postgres/executors/kysely.executor.ts @@ -1,6 +1,7 @@ import type { SqlExecuteResult, SqlExecutor, SqlIsolationLevel, SqlTransaction, SqlTransactionOptions } from '../../interfaces/sql-executor.interface.js'; import { notATransaction } from '../../sql/not-a-transaction.js'; import { describeValue, hasMethod, isolationSql } from '../../utils/executor.util.js'; +import { runTransactionWork } from './transaction-work.js'; /** The part of a Kysely instance (or of a `Transaction`) the executor uses. */ export interface KyselyLike { @@ -59,7 +60,7 @@ class KyselyExecutor implements SqlExecutor<'postgres'> { async transaction(work: (transaction: SqlTransaction) => Promise, options: SqlTransactionOptions = {}): Promise { const builder = this.db.transaction(); - const inTransaction = (trx: KyselyLike) => work(kyselyTransaction(trx)); + const inTransaction = (trx: KyselyLike) => runTransactionWork(work, kyselyTransaction(trx)); if (!options.isolationLevel) { return builder.execute(inTransaction); } diff --git a/lib/postgres/executors/pg.executor.ts b/lib/postgres/executors/pg.executor.ts index 639b1c5..d918ac5 100644 --- a/lib/postgres/executors/pg.executor.ts +++ b/lib/postgres/executors/pg.executor.ts @@ -1,6 +1,7 @@ import type { SqlExecuteResult, SqlExecutor, SqlTransaction, SqlTransactionOptions } from '../../interfaces/sql-executor.interface.js'; import { notATransaction } from '../../sql/not-a-transaction.js'; import { describeValue, hasMethod, isMysql2Client, isolationSql } from '../../utils/executor.util.js'; +import { runTransactionWork } from './transaction-work.js'; /** The part of a node-postgres `Client` (or `PoolClient`) the executor uses. */ export interface PgClientLike { @@ -81,6 +82,7 @@ class PgPoolExecutor implements SqlExecutor<'postgres'> { } async transaction(work: (transaction: SqlTransaction) => Promise, options: SqlTransactionOptions = {}): Promise { + const begin = beginSql(options); // The pool listens to a connection's errors only while it's idle, and the client emits 'error' when the server ends // the connection (a restart, a failover), during a statement too: an 'error' nothing listens to ends the process. // The transaction listens from the tick the pool hands the connection out in, as an `await` would resume too late: @@ -102,8 +104,11 @@ class PgPoolExecutor implements SqlExecutor<'postgres'> { ); // A statement after the connection is lost fails with what ended it (the server's 25P03 or 57P01, a reset...), // rather than with pg's "Client has encountered a connection error and is not queryable". - const connection = { query: (text: string, values?: any[]) => (lost ? Promise.reject(lost) : client.query(text, values)) }; - const outcome = await runTransaction(connection, work, options); + const connection = { + query: (text: string, values?: any[]) => (lost ? Promise.reject(lost) : client.query(text, values)), + getTransactionStatus: () => (client as TransactionStatusLike).getTransactionStatus?.() ?? null, + }; + const outcome = await runTransaction(connection, work, begin); client.removeListener?.('error', onError); // A connection whose transaction can't be ended cleanly, or that failed, goes away rather than back to the pool. @@ -133,9 +138,10 @@ class PgClientExecutor implements SqlExecutor<'postgres'> { return this.exclusive(async () => affected(await this.client.query(text, [...params]))); } - transaction(work: (transaction: SqlTransaction) => Promise, options: SqlTransactionOptions = {}): Promise { + async transaction(work: (transaction: SqlTransaction) => Promise, options: SqlTransactionOptions = {}): Promise { + const begin = beginSql(options); return this.exclusive(async () => { - const outcome = await runTransaction(this.client, work, options); + const outcome = await runTransaction(this.client, work, begin); if ('error' in outcome) { throw outcome.error; } @@ -161,16 +167,24 @@ class PgClientExecutor implements SqlExecutor<'postgres'> { type Outcome = { result: T; broken?: Error } | { error: unknown; broken?: Error }; -async function runTransaction(client: Pick, work: (transaction: SqlTransaction) => Promise, options: SqlTransactionOptions): Promise> { +/** node-postgres 8.21 and later: the connection's transaction status from its last ReadyForQuery ('I', 'T' or 'E'). */ +type TransactionStatusLike = { getTransactionStatus?(): string | null }; + +/** The transaction's BEGIN, checked before it takes a connection: an unknown isolation level needs none to be refused. */ +function beginSql(options: SqlTransactionOptions): string { + return options.isolationLevel ? `BEGIN ISOLATION LEVEL ${isolationSql(options.isolationLevel)}` : 'BEGIN'; +} + +async function runTransaction(client: Pick & TransactionStatusLike, work: (transaction: SqlTransaction) => Promise, begin: string): Promise> { try { - await client.query(options.isolationLevel ? `BEGIN ISOLATION LEVEL ${isolationSql(options.isolationLevel)}` : 'BEGIN'); + await client.query(begin); } catch (error) { return { error, broken: asError(error) }; } let result: T; try { - result = await work(clientTransaction(client)); + result = await runTransactionWork(work, clientTransaction(client)); } catch (error) { try { await client.query('ROLLBACK'); @@ -184,8 +198,13 @@ async function runTransaction(client: Pick, work: (tra await client.query('COMMIT'); return { result }; } catch (error) { - // PostgreSQL ends a transaction whose COMMIT fails; a lost connection is broken too. - return { error, broken: asError(error) }; + // PostgreSQL ends a transaction whose COMMIT fails (a serialization failure, a deferred constraint): the connection + // is idle and goes back to the pool once a statement shows it is. A lost one is broken. + const idle = await client.query('SELECT 1').then( + () => (client.getTransactionStatus?.() ?? 'I') === 'I', + () => false, + ); + return idle ? { error } : { error, broken: asError(error) }; } } @@ -200,7 +219,7 @@ function pgTransaction(transaction: unknown): SqlTransaction { ); } - const client = transaction as PgClientLike & { getTransactionStatus?(): string | null }; + const client = transaction as PgClientLike & TransactionStatusLike; if (typeof client.getTransactionStatus === 'function') { const status = client.getTransactionStatus(); if (status === 'E') { diff --git a/lib/postgres/executors/prisma.executor.ts b/lib/postgres/executors/prisma.executor.ts index 7617120..c05b28c 100644 --- a/lib/postgres/executors/prisma.executor.ts +++ b/lib/postgres/executors/prisma.executor.ts @@ -2,6 +2,7 @@ import type { SqlExecuteResult, SqlExecutor, SqlTransaction, SqlTransactionOptio import { notATransaction } from '../../sql/not-a-transaction.js'; import { parseDuration, type Duration } from '../../utils/duration.util.js'; import { describeValue, hasMethod, isolationSql } from '../../utils/executor.util.js'; +import { runTransactionWork } from './transaction-work.js'; /** The part of a Prisma client (or of the transaction client `$transaction()` hands its callback) the executor uses. */ export interface PrismaClientLike { @@ -84,7 +85,7 @@ class PrismaExecutor implements SqlExecutor<'postgres'> { } async transaction(work: (transaction: SqlTransaction) => Promise, options: SqlTransactionOptions = {}): Promise { - return this.prisma.$transaction((tx: PrismaClientLike) => work(prismaTransaction(tx)), { + return this.prisma.$transaction((tx: PrismaClientLike) => runTransactionWork(work, prismaTransaction(tx)), { ...this.limits, ...(options.isolationLevel ? { isolationLevel: PRISMA_ISOLATION[isolationSql(options.isolationLevel)] } : {}), }); diff --git a/lib/postgres/executors/transaction-work.ts b/lib/postgres/executors/transaction-work.ts new file mode 100644 index 0000000..7fe8cbd --- /dev/null +++ b/lib/postgres/executors/transaction-work.ts @@ -0,0 +1,40 @@ +import type { SqlTransaction } from '../../interfaces/sql-executor.interface.js'; + +/** + * Runs a store's `work` on the transaction an executor opened for it, and rejects rather than let the executor commit + * a transaction PostgreSQL has ended: PostgreSQL fails a transaction at its first failed statement, and answers its + * COMMIT with a ROLLBACK, no error. So `work` that caught a statement's error and resolved would look committed while + * nothing of it was. A rejection makes the executor roll back. + * + * Costs nothing on the way to a commit: only `work` whose last statement failed gets one statement more, which tells a + * transaction PostgreSQL ended from a statement the driver failed before sending it (that leaves the transaction as it + * was) and from a `ROLLBACK TO SAVEPOINT` that followed a failure (the last statement didn't fail). + */ +export async function runTransactionWork(work: (transaction: SqlTransaction) => Promise, transaction: SqlTransaction): Promise { + let failure: { error: unknown } | undefined; + const watch = (statement: Promise): Promise => + statement.then( + (value) => { + failure = undefined; + return value; + }, + (error: unknown) => { + failure = { error }; + throw error; + }, + ); + + const execute = transaction.execute?.bind(transaction); + const result = await work({ + query: (text: string, params?: readonly unknown[]) => watch(transaction.query(text, params)), + ...(execute ? { execute: (text: string, params?: readonly unknown[]) => watch(execute(text, params)) } : {}), + }); + + if (failure && (await transaction.query('SELECT 1').then(() => false, () => true))) { + throw new Error( + "The transaction was rolled back: one of its statements failed and its error was caught, but PostgreSQL ends a transaction at its first failed statement. Let the statement's error reject the work, or run the statement after a SAVEPOINT and ROLLBACK TO SAVEPOINT when it fails.", + { cause: failure.error }, + ); + } + return result; +} diff --git a/lib/postgres/executors/typeorm.executor.ts b/lib/postgres/executors/typeorm.executor.ts index 07f6213..ced720d 100644 --- a/lib/postgres/executors/typeorm.executor.ts +++ b/lib/postgres/executors/typeorm.executor.ts @@ -1,6 +1,7 @@ import type { SqlExecuteResult, SqlExecutor, SqlTransaction, SqlTransactionOptions } from '../../interfaces/sql-executor.interface.js'; import { notATransaction } from '../../sql/not-a-transaction.js'; import { describeValue, hasMethod, isolationSql } from '../../utils/executor.util.js'; +import { runTransactionWork } from './transaction-work.js'; /** The part of a TypeORM `QueryRunner` the executor uses. */ export interface TypeOrmQueryRunnerLike { @@ -79,7 +80,7 @@ class TypeOrmExecutor implements SqlExecutor<'postgres'> { } async transaction(work: (transaction: SqlTransaction) => Promise, options: SqlTransactionOptions = {}): Promise { - const inTransaction = (manager: TypeOrmEntityManagerLike) => work(runnerTransaction(manager.queryRunner!)); + const inTransaction = (manager: TypeOrmEntityManagerLike) => runTransactionWork(work, runnerTransaction(manager.queryRunner!)); return options.isolationLevel ? this.dataSource.transaction(isolationSql(options.isolationLevel), inTransaction) : this.dataSource.transaction(inTransaction); } diff --git a/lib/testing/postgres-executor.cases.ts b/lib/testing/postgres-executor.cases.ts index b760494..56281e7 100644 --- a/lib/testing/postgres-executor.cases.ts +++ b/lib/testing/postgres-executor.cases.ts @@ -71,6 +71,28 @@ export const POSTGRES_CASES: DialectCases = { ); includes(messages(duplicate), 'duplicate key', 'a transaction whose statement fails'); equal(await ids(executor, table), ['kept'], 'the rows after a committed transaction and two rolled back'); + + // PostgreSQL ends a transaction at its first failed statement, and answers its COMMIT with a ROLLBACK, no error: + // work that caught the statement's error and resolved must not look committed. + const swallowed = await rejection( + executor.transaction(async (tx) => { + await tx.query(`INSERT INTO ${table} (id) VALUES ($1::text)`, ['swallowed']); + await tx.query(`INSERT INTO ${table} (id) VALUES ($1::text)`, ['kept']).catch(() => undefined); + return 'resolved'; + }), + 'a transaction whose work catches a failed statement and resolves', + ); + includes(messages(swallowed), 'duplicate key', 'a transaction whose work catches a failed statement and resolves'); + equal(await ids(executor, table), ['kept'], 'the rows after a transaction whose work caught a failed statement'); + + const recovered = await executor.transaction(async (tx) => { + await tx.query('SAVEPOINT attempt'); + await tx.query(`INSERT INTO ${table} (id) VALUES ($1::text)`, ['kept']).catch(() => tx.query('ROLLBACK TO SAVEPOINT attempt')); + await tx.query(`INSERT INTO ${table} (id) VALUES ($1::text)`, ['after-savepoint']); + return 'committed'; + }); + equal(recovered, 'committed', 'a transaction whose work rolled back to a savepoint after a failed statement'); + equal(await ids(executor, table), ['after-savepoint', 'kept'], 'the rows after a transaction that rolled back to a savepoint'); equal(await executor.transaction((tx) => tx.query("SELECT 'next'::text AS status")), [{ status: 'next' }], 'a transaction after failed ones'); }, diff --git a/tests/testing/sql-executor-contract.spec.ts b/tests/testing/sql-executor-contract.spec.ts index 17ed2b5..6c4b9a3 100644 --- a/tests/testing/sql-executor-contract.spec.ts +++ b/tests/testing/sql-executor-contract.spec.ts @@ -1,7 +1,8 @@ /** * The executor contract checks what it claims: executors that each break one rule (ignore the isolation level, run - * the work outside a transaction, join anything, share one connection without taking turns, rename columns, inline - * parameters, wrap the work's error) fail the case that guards it, and pass the others where they can. + * the work outside a transaction, commit work that caught a failed statement, join anything, share one connection + * without taking turns, rename columns, inline parameters, wrap the work's error) fail the case that guards it, and + * pass the others where they can. */ import pg from 'pg'; import { fromPg, type SqlExecutor, type SqlTransaction } from '../../lib/postgres/index.js'; @@ -80,6 +81,11 @@ describe('sqlExecutorContract()', () => { expect(await outcome(executor, CASE.connections)).not.toBeNull(); }); + it('fails an executor that commits work which caught a failed statement, as if PostgreSQL had kept its transaction', async () => { + const executor = broken({ transaction: (work) => client!.transaction((tx) => work(client!.executor.wrapTransaction(tx))) }); + expect((await outcome(executor, CASE.transactions))?.message).toMatch(/^a transaction whose work catches a failed statement and resolves: expected a rejection/); + }); + it("fails an executor that joins anything, running statements outside the application's transaction", async () => { const executor = broken({ wrapTransaction: () => client!.executor }); expect((await outcome(executor, CASE.joins))?.message).toMatch(/^the rows after the application's rollback and commit/);