diff --git a/src/js/internal/sql/query.ts b/src/js/internal/sql/query.ts index e7452be5fe3b..97985469fcf1 100644 --- a/src/js/internal/sql/query.ts +++ b/src/js/internal/sql/query.ts @@ -128,35 +128,9 @@ class Query> extends PublicPromise { } 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() { diff --git a/src/js/internal/sql/shared.ts b/src/js/internal/sql/shared.ts index 7dc59a8b0290..45ebce7ea968 100644 --- a/src/js/internal/sql/shared.ts +++ b/src/js/internal/sql/shared.ts @@ -1240,7 +1240,8 @@ abstract class BaseSQLAdapter 2 ** 31 || timeout < 0 || timeout !== timeout) { throw $ERR_INVALID_ARG_VALUE("options.timeout", timeout, "must be a non-negative integer less than 2^31"); diff --git a/test/js/sql/sql-close-pending-connection.test.ts b/test/js/sql/sql-close-pending-connection.test.ts index df35ab760a25..7cdd89d3b19c 100644 --- a/test/js/sql/sql-close-pending-connection.test.ts +++ b/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 }; + +async function pgReadyServer(onCommand?: (socket: Socket, type: number) => void): Promise { + const received = Promise.withResolvers(); + 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 { + const received = Promise.withResolvers(); + 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(); + } + }); +} diff --git a/test/js/sql/sql-mysql.test.ts b/test/js/sql/sql-mysql.test.ts index ebbfa996a7a3..6c8aad527d6c 100644 --- a/test/js/sql/sql-mysql.test.ts +++ b/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 })); diff --git a/test/js/sql/sql.test.ts b/test/js/sql/sql.test.ts index 75ce11d3b130..0de7e68cd727 100644 --- a/test/js/sql/sql.test.ts +++ b/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 })); diff --git a/test/js/sql/sqlite-sql.test.ts b/test/js/sql/sqlite-sql.test.ts index b86a49018104..d990fe7dd2d3 100644 --- a/test/js/sql/sqlite-sql.test.ts +++ b/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:");