fix(postgres): reject work that caught a failed statement instead of committing nothing

lib/interfaces/sql-executor.interface.ts

@@ -114,7 +114,9 @@ export interface SqlExecutor<D extends SqlDialect = SqlDialect> {
   execute?(text: string, params?: readonly unknown[]): Promise<SqlExecuteResult>;
   /**
    * 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<T>(work: (transaction: SqlTransaction) => Promise<T>, options?: SqlTransactionOptions): Promise<T>;
   /**

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 {

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<T>(work: (transaction: SqlTransaction) => Promise<T>, options: SqlTransactionOptions = {}): Promise<T> {
     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);
     }

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<T>(work: (transaction: SqlTransaction) => Promise<T>, options: SqlTransactionOptions = {}): Promise<T> {
+    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<T>(work: (transaction: SqlTransaction) => Promise<T>, options: SqlTransactionOptions = {}): Promise<T> {
+  async transaction<T>(work: (transaction: SqlTransaction) => Promise<T>, options: SqlTransactionOptions = {}): Promise<T> {
+    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<T> = { result: T; broken?: Error } | { error: unknown; broken?: Error };
 
-async function runTransaction<T>(client: Pick<PgClientLike, 'query'>, work: (transaction: SqlTransaction) => Promise<T>, options: SqlTransactionOptions): Promise<Outcome<T>> {
+/** 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<T>(client: Pick<PgClientLike, 'query'> & TransactionStatusLike, work: (transaction: SqlTransaction) => Promise<T>, begin: string): Promise<Outcome<T>> {
   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<T>(client: Pick<PgClientLike, 'query'>, 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') {

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<T>(work: (transaction: SqlTransaction) => Promise<T>, options: SqlTransactionOptions = {}): Promise<T> {
-    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)] } : {}),
     });

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<T>(work: (transaction: SqlTransaction) => Promise<T>, transaction: SqlTransaction): Promise<T> {
+  let failure: { error: unknown } | undefined;
+  const watch = <R>(statement: Promise<R>): Promise<R> =>
+    statement.then(
+      (value) => {
+        failure = undefined;
+        return value;
+      },
+      (error: unknown) => {
+        failure = { error };
+        throw error;
+      },
+    );
+
+  const execute = transaction.execute?.bind(transaction);
+  const result = await work({
+    query: <R extends object>(text: string, params?: readonly unknown[]) => watch(transaction.query<R>(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;
+}

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<T>(work: (transaction: SqlTransaction) => Promise<T>, options: SqlTransactionOptions = {}): Promise<T> {
-    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);
   }
 

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');
     },
 

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/);