sql: reject tx.unsafe()/tx.file() on a settled transaction or released reserved handle

src/js/bun/sql.ts

@@ -27,6 +27,13 @@ interface TransactionState {
   queries: Set<Query<any, any>>;
 }
 
+function cannotAcceptQueries(state: TransactionState) {
+  return (
+    state.connectionState & ReservedConnectionState.closed ||
+    !(state.connectionState & ReservedConnectionState.acceptQueries)
+  );
+}
+
 function adapterFromOptions(options: Bun.SQL.__internal.DefinedOptions) {
   switch (options.adapter) {
     case "postgres":
@@ -259,10 +266,7 @@ const SQL: typeof Bun.SQL = function SQL(
     }
 
     function reserved_sql(strings: string | TemplateStringsArray | SQLHelper<any> | Query<any, any>, ...values: any[]) {
-      if (
-        state.connectionState & ReservedConnectionState.closed ||
-        !(state.connectionState & ReservedConnectionState.acceptQueries)
-      ) {
+      if (cannotAcceptQueries(state)) {
         return Promise.$reject(pool.connectionClosedError());
       }
       if ($isArray(strings)) {
@@ -278,13 +282,22 @@ const SQL: typeof Bun.SQL = function SQL(
     }
 
     reserved_sql.unsafe = (string, args = []) => {
+      if (cannotAcceptQueries(state)) {
+        return Promise.$reject(pool.connectionClosedError());
+      }
       return unsafeQueryFromTransaction(string, args, pooledConnection, state.queries);
     };
 
     reserved_sql.file = async (path: string, args = []) => {
+      if (cannotAcceptQueries(state)) {
+        return Promise.$reject(pool.connectionClosedError());
+      }
       return await Bun.file(path)
         .text()
         .then(text => {
+          if (cannotAcceptQueries(state)) {
+            return Promise.$reject(pool.connectionClosedError());
+          }
           return unsafeQueryFromTransaction(text, args, pooledConnection, state.queries);
         });
     };
@@ -343,10 +356,7 @@ const SQL: typeof Bun.SQL = function SQL(
     };
     reserved_sql.begin = (options_or_fn: string | TransactionCallback, fn?: TransactionCallback) => {
       // begin is allowed the difference is that we need to make sure to use the same connection and never release it
-      if (
-        state.connectionState & ReservedConnectionState.closed ||
-        !(state.connectionState & ReservedConnectionState.acceptQueries)
-      ) {
+      if (cannotAcceptQueries(state)) {
         return Promise.$reject(pool.connectionClosedError());
       }
       let callback = fn;
@@ -380,10 +390,7 @@ const SQL: typeof Bun.SQL = function SQL(
     };
     reserved_sql.close = async (options?: { timeout?: number }) => {
       const reserveQueries = state.queries;
-      if (
-        state.connectionState & ReservedConnectionState.closed ||
-        !(state.connectionState & ReservedConnectionState.acceptQueries)
-      ) {
+      if (cannotAcceptQueries(state)) {
         return Promise.$resolve(undefined);
       }
       state.connectionState &= ~ReservedConnectionState.acceptQueries;
@@ -426,10 +433,7 @@ const SQL: typeof Bun.SQL = function SQL(
       return Promise.$resolve(undefined);
     };
     reserved_sql.release = () => {
-      if (
-        state.connectionState & ReservedConnectionState.closed ||
-        !(state.connectionState & ReservedConnectionState.acceptQueries)
-      ) {
+      if (cannotAcceptQueries(state)) {
         return Promise.$reject(pool.connectionClosedError());
       }
       // just release the connection back to the pool
@@ -561,10 +565,7 @@ const SQL: typeof Bun.SQL = function SQL(
       strings: string | TemplateStringsArray | import("internal/sql/shared.ts").SQLHelper<any> | Query<any, any>,
       ...values: any[]
     ) {
-      if (
-        state.connectionState & ReservedConnectionState.closed ||
-        !(state.connectionState & ReservedConnectionState.acceptQueries)
-      ) {
+      if (cannotAcceptQueries(state)) {
         return Promise.$reject(pool.connectionClosedError());
       }
       if ($isArray(strings)) {
@@ -579,12 +580,21 @@ const SQL: typeof Bun.SQL = function SQL(
       return queryFromTransaction(strings, values, pooledConnection, state.queries);
     }
     transaction_sql.unsafe = (string, args = []) => {
+      if (cannotAcceptQueries(state)) {
+        return Promise.$reject(pool.connectionClosedError());
+      }
       return unsafeQueryFromTransaction(string, args, pooledConnection, state.queries);
     };
     transaction_sql.file = async (path: string, args = []) => {
+      if (cannotAcceptQueries(state)) {
+        return Promise.$reject(pool.connectionClosedError());
+      }
       return await Bun.file(path)
         .text()
         .then(text => {
+          if (cannotAcceptQueries(state)) {
+            return Promise.$reject(pool.connectionClosedError());
+          }
           return unsafeQueryFromTransaction(text, args, pooledConnection, state.queries);
         });
     };
@@ -645,10 +655,7 @@ const SQL: typeof Bun.SQL = function SQL(
     };
     transaction_sql.close = async function (options?: { timeout?: number }) {
       // we dont actually close the connection here, we just set the state to closed and rollback the transaction
-      if (
-        state.connectionState & ReservedConnectionState.closed ||
-        !(state.connectionState & ReservedConnectionState.acceptQueries)
-      ) {
+      if (cannotAcceptQueries(state)) {
         return Promise.$resolve(undefined);
       }
       state.connectionState &= ~ReservedConnectionState.acceptQueries;
@@ -730,10 +737,7 @@ const SQL: typeof Bun.SQL = function SQL(
       transaction_sql.savepoint = async (fn: TransactionCallback, name?: string): Promise<any> => {
         let savepoint_callback = fn;
 
-        if (
-          state.connectionState & ReservedConnectionState.closed ||
-          !(state.connectionState & ReservedConnectionState.acceptQueries)
-        ) {
+        if (cannotAcceptQueries(state)) {
           throw pool.connectionClosedError();
         }
 

test/js/sql/sql-transaction-closed-handle.test.ts

@@ -0,0 +1,42 @@
+import { SQL } from "bun";
+import { expect, test } from "bun:test";
+import { tempDirWithFiles } from "harness";
+import { join } from "node:path";
+
+// tx.unsafe()/tx.file() must reject once sql.begin() has settled, the same way the
+// tagged-template call and tx.savepoint() already do. Before the fix they reached the
+// pooled connection the handle no longer owns and the write was durable.
+test("tx.unsafe() and tx.file() reject after begin() settles", async () => {
+  const dir = tempDirWithFiles("sql-tx-closed", {
+    "q.sql": `INSERT INTO accounts VALUES (98, 0)`,
+  });
+  const sql = new SQL("sqlite://:memory:");
+  try {
+    await sql`CREATE TABLE accounts (id INTEGER PRIMARY KEY, balance REAL)`;
+    let leaked: any;
+    await sql.begin(async tx => {
+      leaked = tx;
+      await tx.unsafe(`INSERT INTO accounts VALUES (10, 0)`);
+    });
+    const outcome = async (p: Promise<any>) =>
+      p.then(
+        () => "fulfilled",
+        (e: any) => `rejected: ${e.message}`,
+      );
+    expect({
+      template: await outcome(leaked`SELECT 1`),
+      savepoint: await outcome(Promise.resolve().then(() => leaked.savepoint(async () => {}))),
+      unsafe: await outcome(leaked.unsafe(`INSERT INTO accounts VALUES (99, 0)`)),
+      file: await outcome(leaked.file(join(dir, "q.sql"))),
+    }).toEqual({
+      template: "rejected: Connection closed",
+      savepoint: "rejected: Connection closed",
+      unsafe: "rejected: Connection closed",
+      file: "rejected: Connection closed",
+    });
+    const rows = await sql`SELECT id FROM accounts ORDER BY id`;
+    expect(rows).toEqual([{ id: 10 }]);
+  } finally {
+    await sql.close();
+  }
+});

test/js/sql/sql.test.ts

@@ -3997,6 +3997,49 @@ CREATE TABLE ${table_name} (
       expect(xs.map(x => x.x).join("")).toBe("123");
     });
 
+    test("reserved.unsafe() and reserved.file() reject after release()", async () => {
+      await using sql = postgres({ ...options, max: 1 });
+      const reserved = await sql.reserve();
+      await reserved.release();
+      const outcome = async (p: Promise<any>) =>
+        p.then(
+          () => "fulfilled",
+          (e: any) => `rejected: ${e.code}`,
+        );
+      expect({
+        template: await outcome(reserved`select 1 as x`),
+        unsafe: await outcome(reserved.unsafe("select 1 as x")),
+        file: await outcome(reserved.file(rel("select.sql"))),
+      }).toEqual({
+        template: "rejected: ERR_POSTGRES_CONNECTION_CLOSED",
+        unsafe: "rejected: ERR_POSTGRES_CONNECTION_CLOSED",
+        file: "rejected: ERR_POSTGRES_CONNECTION_CLOSED",
+      });
+    });
+
+    test("tx.unsafe() and tx.file() reject after begin() settles", async () => {
+      await using sql = postgres({ ...options, max: 1 });
+      let leaked: any;
+      await sql.begin(async tx => {
+        leaked = tx;
+        await tx.unsafe("select 1 as x");
+      });
+      const outcome = async (p: Promise<any>) =>
+        p.then(
+          () => "fulfilled",
+          (e: any) => `rejected: ${e.code}`,
+        );
+      expect({
+        template: await outcome(leaked`select 1 as x`),
+        unsafe: await outcome(leaked.unsafe("select 1 as x")),
+        file: await outcome(leaked.file(rel("select.sql"))),
+      }).toEqual({
+        template: "rejected: ERR_POSTGRES_CONNECTION_CLOSED",
+        unsafe: "rejected: ERR_POSTGRES_CONNECTION_CLOSED",
+        file: "rejected: ERR_POSTGRES_CONNECTION_CLOSED",
+      });
+    });
+
     test("keeps process alive when it should", async () => {
       const file = path.posix.join(__dirname, "sql-fixture-ref.ts");
       const result = await $`DATABASE_URL=${process.env.DATABASE_URL} ${bunExe()} ${file}`;