From b67f8ebfb68326378881abc1f5162eef6544306d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Kamil=20My=C5=9Bliwiec?= Date: Mon, 19 Oct 2026 11:35:03 +0200 Subject: [PATCH 1/2] fix(postgres): reject work that caught a failed statement instead of committing nothing PostgreSQL ends a transaction at its first failed statement and answers its COMMIT with a ROLLBACK, no error. So a store's work that caught a statement's error (an insert-if-absent catching 23505, say) and resolved made transaction() resolve as if committed, while every write in it was gone; on every PostgreSQL executor. Each now runs the work through runTransactionWork(): when the work's last statement failed, one statement more tells whether PostgreSQL ended the transaction, and the transaction then rolls back and rejects, with the statement's error as the cause. Work that recovered with ROLLBACK TO SAVEPOINT commits as before, and the normal path costs nothing. The contract's transactions case covers both, for every executor that runs it. fromPg(): an unknown isolation level is refused before the transaction takes a connection (it waited for one, and the pool then destroyed it), and a COMMIT that fails with an ERROR (a serialization failure, a deferred constraint) keeps a connection that one more statement shows is idle, instead of reconnecting for every such failure. --- lib/interfaces/sql-executor.interface.ts | 4 ++- lib/postgres/executors/drizzle.executor.ts | 3 +- lib/postgres/executors/kysely.executor.ts | 3 +- lib/postgres/executors/pg.executor.ts | 29 +++++++++++----- lib/postgres/executors/prisma.executor.ts | 3 +- lib/postgres/executors/transaction-work.ts | 40 ++++++++++++++++++++++ lib/postgres/executors/typeorm.executor.ts | 3 +- lib/testing/postgres-executor.cases.ts | 23 ++++++++++++- 8 files changed, 94 insertions(+), 14 deletions(-) create mode 100644 lib/postgres/executors/transaction-work.ts 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 a48d1ea..0bfa929 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 { @@ -68,8 +69,9 @@ class PgPoolExecutor implements SqlExecutor<'postgres'> { } async transaction(work: (transaction: SqlTransaction) => Promise, options: SqlTransactionOptions = {}): Promise { + const begin = beginSql(options); const client = await this.pool.connect(); - const outcome = await runTransaction(client, work, options); + const outcome = await runTransaction(client, work, begin); // A connection whose transaction can't be ended cleanly goes away rather than back to the pool. client.release(outcome.broken); @@ -98,9 +100,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; } @@ -126,16 +129,21 @@ class PgClientExecutor implements SqlExecutor<'postgres'> { type Outcome = { result: T; broken?: Error } | { error: unknown; broken?: Error }; -async function runTransaction(client: PgClientLike, work: (transaction: SqlTransaction) => Promise, options: SqlTransactionOptions): Promise> { +/** 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: PgClientLike, 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'); @@ -149,8 +157,13 @@ async function runTransaction(client: PgClientLike, work: (transaction: SqlTr 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 as { getTransactionStatus?(): string | null }).getTransactionStatus?.() ?? 'I') === 'I', + () => false, + ); + return idle ? { error } : { error, broken: asError(error) }; } } 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..df5b866 100644 --- a/lib/testing/postgres-executor.cases.ts +++ b/lib/testing/postgres-executor.cases.ts @@ -70,7 +70,28 @@ export const POSTGRES_CASES: DialectCases = { 'a transaction whose statement fails', ); 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 committed transaction and three rolled back'); + + 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'); }, From 1d5b907cdd3b25f70a32f3013195b799a6b2ac26 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Kamil=20My=C5=9Bliwiec?= Date: Mon, 19 Oct 2026 11:39:45 +0200 Subject: [PATCH 2/2] test(testing): check the contract's rows before its new case, and that it fails an executor committing caught failures --- lib/testing/postgres-executor.cases.ts | 3 ++- tests/testing/sql-executor-contract.spec.ts | 10 ++++++++-- 2 files changed, 10 insertions(+), 3 deletions(-) diff --git a/lib/testing/postgres-executor.cases.ts b/lib/testing/postgres-executor.cases.ts index df5b866..56281e7 100644 --- a/lib/testing/postgres-executor.cases.ts +++ b/lib/testing/postgres-executor.cases.ts @@ -70,6 +70,7 @@ export const POSTGRES_CASES: DialectCases = { 'a transaction whose statement fails', ); 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. @@ -82,7 +83,7 @@ export const POSTGRES_CASES: DialectCases = { '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 committed transaction and three rolled back'); + 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'); 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/);