sql: drain queries awaited in the same tick as close()/end()

src/js/internal/sql/query.ts

@@ -128,35 +128,9 @@ class Query<T, Handle extends BaseQueryHandle<any>> extends PublicPromise<T> {
   }
 
   async #runAsync() {
-    const { [_handler]: handler, [_queryStatus]: status } = this;
-
-    if (
-      status &
-      (SQLQueryStatus.executed | SQLQueryStatus.error | SQLQueryStatus.cancelled | SQLQueryStatus.invalidHandle)
-    ) {
-      return;
-    }
-
-    if (this[_flags] & SQLQueryFlags.notTagged) {
-      this.reject(this[_adapter].notTaggedCallError());
-      return;
-    }
-
-    this[_queryStatus] |= SQLQueryStatus.executed;
-    const handle = this.#getQueryHandle();
-
-    if (!handle) {
-      return this;
-    }
-
-    await Promise.$resolve();
-
-    try {
-      return handler(this, handle);
-    } catch (err) {
-      this[_queryStatus] |= SQLQueryStatus.error;
-      this.reject(err as Error);
-    }
+    // Enqueue synchronously (same as execute()) so a same-tick close() sees
+    // the query as pending; #run()'s executed-status guard handles re-entry.
+    return this.#run();
   }
 
   get active() {

src/js/internal/sql/shared.ts

@@ -1240,7 +1240,8 @@ abstract class BaseSQLAdapter<PooledConnection extends BasePooledConnection, Con
     }
 
     let timeout = options?.timeout;
-    if (timeout) {
+    // Presence, not truthiness: `timeout: 0` means close now, undefined/null mean drain with no timer.
+    if (timeout != null) {
       timeout = Number(timeout);
       if (timeout > 2 ** 31 || timeout < 0 || timeout !== timeout) {
         throw $ERR_INVALID_ARG_VALUE("options.timeout", timeout, "must be a non-negative integer less than 2^31");

test/js/sql/sql-close-pending-connection.test.ts

@@ -18,7 +18,21 @@
 
 import { SQL } from "bun";
 import { expect, test } from "bun:test";
-import { neverAnsweringServer } from "./wire-frames";
+import type { Server, Socket } from "node:net";
+import {
+  listeningServer,
+  mysqlHandshakeV10,
+  mysqlOkPacket,
+  mysqlReadPackets,
+  mysqlTextResultSet,
+  neverAnsweringServer,
+  pgAuthenticationOk,
+  pgCommandComplete,
+  pgDataRow,
+  pgReadFrontendMessages,
+  pgReadyForQuery,
+  pgRowDescription,
+} from "./wire-frames";
 
 const drivers = [
   ["postgres", "postgres://postgres@", "ERR_POSTGRES_CONNECTION_CLOSED"],
@@ -101,3 +115,125 @@ test("pool scans tolerate unassigned connection slots during pool start", async
     server.close();
   }
 });
+
+// https://github.com/oven-sh/bun/issues/32038
+//
+// close({ timeout }) used to be gated on `if (timeout)`, so the documented
+// `close({ timeout: 0 })` ("close now") fell into the graceful-drain branch and
+// waited for in-flight queries forever. The tests above sidestep that with the
+// truthy string "0"; these use the number. `timeout: null` must still mean
+// "no timeout" (drain), not "timeout of 0".
+//
+// Each mock completes the handshake and hands the first command it receives to
+// `onCommand`; by default it never answers, leaving the query in flight.
+
+type CommandMock = { port: number; server: Server; commandReceived: Promise<void> };
+
+async function pgReadyServer(onCommand?: (socket: Socket, type: number) => void): Promise<CommandMock> {
+  const received = Promise.withResolvers<void>();
+  const { port, server } = await listeningServer(socket => {
+    let startup = true;
+    let buffered = Buffer.alloc(0);
+    socket.on("data", chunk => {
+      if (startup) {
+        startup = false;
+        socket.write(Buffer.concat([pgAuthenticationOk(), pgReadyForQuery()]));
+        return;
+      }
+      buffered = pgReadFrontendMessages(Buffer.concat([buffered, chunk]), type => {
+        onCommand?.(socket, type);
+        received.resolve();
+      });
+    });
+    socket.on("error", () => {});
+  });
+  return { port, server, commandReceived: received.promise };
+}
+
+async function mysqlReadyServer(
+  onCommand?: (socket: Socket, seq: number, payload: Buffer) => void,
+): Promise<CommandMock> {
+  const received = Promise.withResolvers<void>();
+  const { port, server } = await listeningServer(socket => {
+    let buffered = Buffer.alloc(0);
+    let authed = false;
+    socket.write(mysqlHandshakeV10());
+    socket.on("data", chunk => {
+      buffered = mysqlReadPackets(Buffer.concat([buffered, chunk]), (seq, payload) => {
+        if (!authed) {
+          authed = true;
+          socket.write(mysqlOkPacket(seq + 1));
+          return;
+        }
+        onCommand?.(socket, seq, payload);
+        received.resolve();
+      });
+    });
+    socket.on("error", () => {});
+  });
+  return { port, server, commandReceived: received.promise };
+}
+
+// Answers a simple-protocol `select 1 as x` with one text row once `respond()` is called.
+const drainableMocks = {
+  async postgres() {
+    let respond!: () => void;
+    const mock = await pgReadyServer((socket, type) => {
+      if (type !== 0x51 /* Query */) return;
+      respond = () =>
+        socket.write(
+          Buffer.concat([
+            pgRowDescription([{ name: "x", typeOid: 25 }]),
+            pgDataRow([Buffer.from("1")]),
+            pgCommandComplete("SELECT 1"),
+            pgReadyForQuery(),
+          ]),
+        );
+    });
+    return { ...mock, respond: () => respond() };
+  },
+  async mysql() {
+    let respond!: () => void;
+    const mock = await mysqlReadyServer((socket, seq, payload) => {
+      if (payload[0] !== 0x03 /* COM_QUERY */) return;
+      respond = () => socket.write(mysqlTextResultSet(seq + 1, [{ name: "x", type: 0xfd }], [["1"]]));
+    });
+    return { ...mock, respond: () => respond() };
+  },
+} as const;
+
+const silentMocks = {
+  postgres: () => pgReadyServer(),
+  mysql: () => mysqlReadyServer(),
+} as const;
+
+for (const [name, scheme, closedCode] of drivers) {
+  test(`${name}: close({ timeout: 0 }) force-closes with a query in flight`, async () => {
+    const { port, server, commandReceived } = await silentMocks[name]();
+    try {
+      const sql = new SQL({ url: `${scheme}127.0.0.1:${port}/db`, max: 1 });
+      const queryError = sql`SELECT 1`.catch(e => e);
+      // the server has the query and will never answer it
+      await commandReceived;
+      await sql.close({ timeout: 0 });
+      expect((await queryError).code).toBe(closedCode);
+    } finally {
+      server.close();
+    }
+  });
+
+  test(`${name}: close({ timeout: null }) still waits for the query in flight`, async () => {
+    const { port, server, commandReceived, respond } = await drainableMocks[name]();
+    try {
+      const sql = new SQL({ url: `${scheme}127.0.0.1:${port}/db`, max: 1 });
+      const rows = sql`select 1 as x`.simple().then(r => r);
+      await commandReceived;
+      const closing = sql.close({ timeout: null });
+      respond();
+      expect(await rows).toEqual([{ x: "1" }]);
+      await closing;
+    } finally {
+      server.close();
+    }
+  });
+}

test/js/sql/sql-mysql.test.ts

@@ -1002,6 +1002,16 @@ if (isDockerEnabled()) {
           return expect(await promise).toEqual([{ x: 0 }]);
         });
 
+        // Same contract as the .execute() case above, but via .then(): the query
+        // was handed to the pool one microtask late, so a same-tick end() ran
+        // first and rejected it with ERR_MYSQL_CONNECTION_CLOSED.
+        test("Connection end does not cancel a query awaited in the same tick", async () => {
+          const sql = new SQL({ ...getOptions(), max: 1 });
+          await sql`select 1 as x`;
+          const [rows] = await Promise.all([sql`select 1 as x`.then(r => r), sql.end()]);
+          expect(rows).toEqual([{ x: 1 }]);
+        });
+
         test("Connection destroyed", async () => {
           const sql = new SQL(getOptions());
           process.nextTick(() => sql.end({ timeout: 0 }));

test/js/sql/sql.test.ts

@@ -1805,6 +1805,23 @@ if (isDockerEnabled()) {
       return expect(await promise).toEqual([{ x: "" }]);
     });
 
+    // Same contract as the .execute() case above, but via .then(): the query
+    // was handed to the pool one microtask late, so a same-tick end() ran
+    // first and rejected it with ERR_POSTGRES_CONNECTION_CLOSED.
+    test("Connection end does not cancel a query awaited in the same tick", async () => {
+      const sql = postgres({ ...options, max: 1 });
+      await sql`select 1 as x`;
+      const [rows] = await Promise.all([sql`select 1 as x`.then(r => r), sql.end()]);
+      expect(rows).toEqual([{ x: 1 }]);
+    });
+
+    test("Connection end with a timeout does not cancel a query awaited in the same tick", async () => {
+      const sql = postgres({ ...options, max: 1 });
+      await sql`select 1 as x`;
+      const [rows] = await Promise.all([sql`select 1 as x`.then(r => r), sql.end({ timeout: 5 })]);
+      expect(rows).toEqual([{ x: 1 }]);
+    });
+
     test("Connection destroyed", async () => {
       const sql = postgres(options);
       process.nextTick(() => sql.end({ timeout: 0 }));

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

@@ -1970,6 +1970,16 @@ describe("Connection management", () => {
     }
   });
 
+  // Query.then() used to defer the pool hand-off by one microtask, so a
+  // close() in the same synchronous block ran first, saw zero pending
+  // queries, and rejected the already-awaited query with "Connection closed".
+  test("close() drains a query awaited in the same tick", async () => {
+    const sql = new SQL("sqlite://:memory:");
+    await sql`SELECT 1 AS x`;
+    const [rows] = await Promise.all([sql`SELECT 42 AS x`.then(r => r), sql.close()]);
+    expect(rows).toEqual([{ x: 42 }]);
+  });
+
   test("reserve throws for SQLite", async () => {
     const sql = new SQL("sqlite://:memory:");