sql: reject a transaction or reserved query that first runs after the handle closed

docs/runtime/sql.mdx

@@ -1002,6 +1002,8 @@ try {
 } // Automatically released
 ```
 
+Await each query before you release the connection. A query is lazy and runs when it is awaited. A query that first runs after `release()` rejects with `ERR_POSTGRES_CONNECTION_CLOSED` (or the MySQL equivalent), because the connection can already belong to another caller. `return reserved\`SELECT 1\``without`await` from a block that releases the handle is such a query.
+
 When every connection is busy, `reserve()` waits until one frees up. Pass an `AbortSignal` to stop waiting: the promise rejects with `signal.reason` and no connection is taken from the pool. Aborting after the promise resolved has no effect, so a connection you received must still be released.
 
 ```ts

src/js/bun/sql.ts

@@ -176,8 +176,9 @@ const SQL: typeof Bun.SQL = function SQL(
     transactionQueries.delete(query);
   }
 
-  function queryFromTransactionHandler(transactionQueries, query, handle, err) {
+  function queryFromTransactionHandler(state: TransactionState, query, handle, err) {
     const pooledConnection = this;
+    const transactionQueries = state.queries;
     if (err) {
       transactionQueries.delete(query);
       return query.reject(err);
@@ -189,6 +190,14 @@ const SQL: typeof Bun.SQL = function SQL(
       return query.reject(pool.queryCancelledError());
     }
 
+    // A query is lazy: it runs when it is first awaited. The handle can be
+    // committed, rolled back or released by then, and the connection can
+    // already belong to another caller.
+    if (state.connectionState & ReservedConnectionState.closed) {
+      transactionQueries.delete(query);
+      return query.reject(pool.connectionClosedError());
+    }
+
     query.finally(onTransactionQueryDisconnected.bind(transactionQueries, query));
 
     try {
@@ -207,7 +216,7 @@ const SQL: typeof Bun.SQL = function SQL(
     strings: string | TemplateStringsArray | import("internal/sql/shared.ts").SQLHelper<any> | Query<any, any>,
     values: any[],
     pooledConnection: PooledPostgresConnection,
-    transactionQueries: Set<Query<any, any>>,
+    state: TransactionState,
   ) {
     try {
       const query = new Query(
@@ -216,11 +225,11 @@ const SQL: typeof Bun.SQL = function SQL(
         connectionInfo.bigint
           ? SQLQueryFlags.allowUnsafeTransaction | SQLQueryFlags.bigint
           : SQLQueryFlags.allowUnsafeTransaction,
-        queryFromTransactionHandler.bind(pooledConnection, transactionQueries),
+        queryFromTransactionHandler.bind(pooledConnection, state),
         pool,
       );
 
-      transactionQueries.add(query);
+      state.queries.add(query);
       return query;
     } catch (err) {
       return Promise.$reject(err);
@@ -242,17 +251,17 @@ const SQL: typeof Bun.SQL = function SQL(
     strings: string | TemplateStringsArray | import("internal/sql/shared.ts").SQLHelper<any> | Query<any, any>,
     values: any[],
     pooledConnection: PooledPostgresConnection,
-    transactionQueries: Set<Query<any, any>>,
+    state: TransactionState,
   ) {
     try {
       const query = new Query(
         strings,
         values,
         unsafeTransactionQueryFlags(values),
-        queryFromTransactionHandler.bind(pooledConnection, transactionQueries),
+        queryFromTransactionHandler.bind(pooledConnection, state),
         pool,
       );
-      transactionQueries.add(query);
+      state.queries.add(query);
       return query;
     } catch (err) {
       return Promise.$reject(err);
@@ -271,7 +280,7 @@ const SQL: typeof Bun.SQL = function SQL(
     values: any[],
   ) {
     if (acceptsQueries(state)) {
-      return unsafeQueryFromTransaction(strings, values, pooledConnection, state.queries);
+      return unsafeQueryFromTransaction(strings, values, pooledConnection, state);
     }
     // Still a lazy Query, so .values() and use as a fragment work. It rejects when it runs.
     return new Query(strings, values, unsafeTransactionQueryFlags(values), rejectConnectionClosed, pool);
@@ -407,7 +416,7 @@ const SQL: typeof Bun.SQL = function SQL(
         return new SQLHelper([strings], values);
       }
       // we use the same code path as the transaction sql
-      return queryFromTransaction(strings, values, pooledConnection, state.queries);
+      return queryFromTransaction(strings, values, pooledConnection, state);
     }
 
     reserved_sql.unsafe = (string, args = []) => unsafeQueryFromHandle(state, pooledConnection, string, args);
@@ -678,7 +687,7 @@ const SQL: typeof Bun.SQL = function SQL(
       if (state.connectionState & ReservedConnectionState.closed) {
         return Promise.$reject(pool.connectionClosedError());
       }
-      return unsafeQueryFromTransaction(string, [], pooledConnection, state.queries);
+      return unsafeQueryFromTransaction(string, [], pooledConnection, state);
     }
     function transaction_sql(
       strings: string | TemplateStringsArray | import("internal/sql/shared.ts").SQLHelper<any> | Query<any, any>,
@@ -696,7 +705,7 @@ const SQL: typeof Bun.SQL = function SQL(
         return new SQLHelper([strings], values);
       }
 
-      return queryFromTransaction(strings, values, pooledConnection, state.queries);
+      return queryFromTransaction(strings, values, pooledConnection, state);
     }
     transaction_sql.unsafe = (string, args = []) => unsafeQueryFromHandle(state, pooledConnection, string, args);
     transaction_sql.file = (path: string, args = []) => fileQueryFromHandle(state, pooledConnection, path, args);

test/js/sql/sql-pool-transaction-isolation.test.ts

@@ -400,6 +400,144 @@ describe.each(adapters)("$adapter", ({ adapter, mockServer, beginCommand }) => {
     }
   });
 
+  // A query is lazy: it runs when it is first awaited. One created on a transaction or
+  // a reservation and first awaited after that handle settled must not run, because
+  // the pool's only connection can already be inside somebody else's transaction.
+  // unsafe() without parameters, because the mock servers answer simple queries only.
+  test("a query first awaited after its transaction settled rejects instead of running on the next transaction", async () => {
+    const received: Received[] = [];
+    const { port, server } = await mockServer(received);
+    const sql = new SQL(options(port));
+    const closedCode = adapter === "postgres" ? "ERR_POSTGRES_CONNECTION_CLOSED" : "ERR_MYSQL_CONNECTION_CLOSED";
+    try {
+      let afterRollback!: Promise<unknown>;
+      let afterCommit!: Promise<unknown>;
+      const err = await sql
+        .begin(async tx => {
+          afterRollback = tx.unsafe("SELECT 'lazy from T1'");
+          throw new Error("t1-app-error");
+        })
+        .catch(e => e);
+      expect(err.message).toBe("t1-app-error");
+      await sql.begin(async tx => {
+        await tx.unsafe("SELECT 'T2a'");
+        afterCommit = tx.unsafe("SELECT 'lazy from T2'");
+      });
+
+      const t3 = sql.begin(async tx => {
+        await tx.unsafe("SELECT 'T3a'");
+        expect(
+          await afterRollback.then(
+            () => null,
+            e => e.code,
+          ),
+        ).toBe(closedCode);
+        expect(
+          await afterCommit.then(
+            () => null,
+            e => e.code,
+          ),
+        ).toBe(closedCode);
+        await tx.unsafe("SELECT 'T3b'");
+        return "t3";
+      });
+      expect(await t3).toBe("t3");
+
+      expect(received).toEqual([
+        { conn: 0, sql: beginCommand },
+        { conn: 0, sql: "ROLLBACK" },
+        { conn: 0, sql: beginCommand },
+        { conn: 0, sql: "SELECT 'T2a'" },
+        { conn: 0, sql: "COMMIT" },
+        { conn: 0, sql: beginCommand },
+        { conn: 0, sql: "SELECT 'T3a'" },
+        { conn: 0, sql: "SELECT 'T3b'" },
+        { conn: 0, sql: "COMMIT" },
+      ]);
+    } finally {
+      await sql.close({ timeout: 0 }).catch(() => {});
+      await new Promise<void>(r => server.close(() => r()));
+    }
+  });
+
+  test("a query first awaited after its reservation was released rejects instead of running on the next transaction", async () => {
+    const received: Received[] = [];
+    const { port, server } = await mockServer(received);
+    const sql = new SQL(options(port));
+    const closedCode = adapter === "postgres" ? "ERR_POSTGRES_CONNECTION_CLOSED" : "ERR_MYSQL_CONNECTION_CLOSED";
+    try {
+      const reserved = await sql.reserve();
+      await reserved.unsafe("SELECT 'R1'");
+      const afterRelease = reserved.unsafe("SELECT 'lazy from R'");
+      reserved.release();
+
+      const t1 = await sql.begin(async tx => {
+        await tx.unsafe("SELECT 'T1a'");
+        expect(
+          await afterRelease.then(
+            () => null,
+            e => e.code,
+          ),
+        ).toBe(closedCode);
+        await tx.unsafe("SELECT 'T1b'");
+        return "t1";
+      });
+      expect(t1).toBe("t1");
+
+      expect(received).toEqual([
+        { conn: 0, sql: "SELECT 'R1'" },
+        { conn: 0, sql: beginCommand },
+        { conn: 0, sql: "SELECT 'T1a'" },
+        { conn: 0, sql: "SELECT 'T1b'" },
+        { conn: 0, sql: "COMMIT" },
+      ]);
+    } finally {
+      await sql.close({ timeout: 0 }).catch(() => {});
+      await new Promise<void>(r => server.close(() => r()));
+    }
+  });
+
+  // Behaviour change: a query returned without await from a block that releases the
+  // handle first runs after release(). It used to run on the released connection.
+  test.each([
+    [
+      "a using block",
+      async (sql: Bun.SQL) => {
+        using reserved = await sql.reserve();
+        return reserved.unsafe("SELECT 'returned without await'");
+      },
+    ],
+    [
+      "a finally block that calls release()",
+      async (sql: Bun.SQL) => {
+        const reserved = await sql.reserve();
+        try {
+          return reserved.unsafe("SELECT 'returned without await'");
+        } finally {
+          reserved.release();
+        }
+      },
+    ],
+  ])("a query returned without await from %s rejects and sends nothing", async (_name, run) => {
+    const received: Received[] = [];
+    const { port, server } = await mockServer(received);
+    const sql = new SQL(options(port));
+    const closedCode = adapter === "postgres" ? "ERR_POSTGRES_CONNECTION_CLOSED" : "ERR_MYSQL_CONNECTION_CLOSED";
+    try {
+      expect(
+        await run(sql).then(
+          () => null,
+          e => e.code,
+        ),
+      ).toBe(closedCode);
+      await sql.unsafe("SELECT 'barrier'");
+      expect(received).toEqual([{ conn: 0, sql: "SELECT 'barrier'" }]);
+    } finally {
+      await sql.close({ timeout: 0 }).catch(() => {});
+      await new Promise<void>(r => server.close(() => r()));
+    }
+  });
+
   // Runs in a child process: bun:test would turn any unhandled rejection into a test
   // failure, and the second half of this contract is that one rejection IS reported.
   test("a rejected reserved begin() is reported as unhandled only when the caller ignores it", async () => {

test/js/sql/sqlite-sql.test.ts

@@ -1209,6 +1209,43 @@ describe("Transactions", () => {
     expect(accounts[1].balance).toBe(500);
   });
 
+  // A query is lazy. One created inside the callback but first awaited after the
+  // transaction settled must not run outside the transaction.
+  test.each([
+    ["tagged template", (tx: Bun.TransactionSQL) => tx`UPDATE accounts SET balance = 0 WHERE id = 1`],
+    ["unsafe()", (tx: Bun.TransactionSQL) => tx.unsafe("UPDATE accounts SET balance = 0 WHERE id = 1")],
+  ])("a %s query first awaited after the transaction settled rejects", async (_name, make) => {
+    let afterRollback!: Promise<unknown>;
+    let afterCommit!: Promise<unknown>;
+
+    const err = await sql
+      .begin(async tx => {
+        afterRollback = make(tx);
+        throw new Error("roll back");
+      })
+      .catch(e => e);
+    expect(err.message).toBe("roll back");
+    expect(
+      await afterRollback.then(
+        () => null,
+        e => e.code,
+      ),
+    ).toBe("ERR_SQLITE_CONNECTION_CLOSED");
+
+    await sql.begin(async tx => {
+      afterCommit = make(tx);
+    });
+    expect(
+      await afterCommit.then(
+        () => null,
+        e => e.code,
+      ),
+    ).toBe("ERR_SQLITE_CONNECTION_CLOSED");
+
+    const accounts = await sql`SELECT balance FROM accounts WHERE id = 1`;
+    expect(accounts).toEqual([{ balance: 1000 }]);
+  });
+
   test("nested transactions (savepoints)", async () => {
     await sql.begin(async tx => {
       await tx`UPDATE accounts SET balance = balance - 100 WHERE id = 1`;