Skip to content
Closed
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions docs/runtime/sql.mdx
Original file line number Diff line number Diff line change
Expand Up @@ -978,6 +978,8 @@ await sql.close({ timeout: 5 }); // wait 5 seconds and close all connections fro
await sql.close({ timeout: 0 }); // close all connections from the pool immediately
```

`timeout` is in seconds. With `timeout: 0`, or when the timeout expires, queries that are still in flight reject with the adapter's `ERR_*_CONNECTION_CLOSED` error.

---

## Reserved Connections
Expand Down
3 changes: 2 additions & 1 deletion src/js/internal/sql/shared.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1339,7 +1339,8 @@ abstract class BaseSQLAdapter<PooledConnection extends BasePooledConnection, Con
}
Comment thread
robobun marked this conversation as resolved.
Resolved

let timeout = options?.timeout;
const hasTimeout = !!timeout;
// Presence, not truthiness: `timeout: 0` means close now, undefined/null mean drain with no timer.
const hasTimeout = timeout != null;
Comment thread
robobun marked this conversation as resolved.
Resolved
Comment thread
robobun marked this conversation as resolved.
Resolved
if (hasTimeout) {
timeout = Number(timeout);
if (timeout > 2 ** 31 || timeout < 0 || timeout !== timeout) {
Comment thread
robobun marked this conversation as resolved.
Resolved
Expand Down
144 changes: 137 additions & 7 deletions test/js/sql/sql-close-pending-connection.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@

// https://github.com/oven-sh/bun/issues/32095
//
// A forced pool close (`close({ timeout: "0" })`) must resolve even when a
// A forced pool close (`close({ timeout: 0 })`) must resolve even when a
// pool connection has been accepted at the TCP level but the database
// handshake has not completed yet (a database that is still starting up).
// Previously the pending queries were rejected but the promise returned by
Expand All @@ -18,7 +18,22 @@

import { SQL } from "bun";
import { expect, mock, test } from "bun:test";
import { listeningServer, neverAnsweringServer, pgAuthenticationOk, pgReadyForQuery } from "./wire-frames";
import type { Server, Socket } from "node:net";
import {
listeningServer,
mysqlAckSessionSetup,
mysqlHandshakeV10,
mysqlOkPacket,
mysqlReadPackets,
mysqlTextResultSet,
neverAnsweringServer,
pgAuthenticationOk,
pgCommandComplete,
pgDataRow,
pgReadFrontendMessages,
pgReadyForQuery,
pgRowDescription,
} from "./wire-frames";

const drivers = [
["postgres", "postgres://postgres@", "ERR_POSTGRES_CONNECTION_CLOSED"],
Expand All @@ -34,7 +49,7 @@ for (const [name, scheme, closedCode] of drivers) {
// the server holds the connection open without ever completing the
// handshake, so the pool connection stays mid-handshake from here on
await accepted;
await sql.close({ timeout: "0" });
await sql.close({ timeout: 0 });
expect((await queryError).code).toBe(closedCode);
} finally {
server.close();
Expand All @@ -61,7 +76,7 @@ for (const [name, scheme, closedCode] of drivers) {
});
const queryError = sql`SELECT 1`.catch(e => e);
await accepted;
await sql.close({ timeout: "0" });
await sql.close({ timeout: 0 });
expect((await queryError).code).toBe(closedCode);
expect(onconnect).not.toHaveBeenCalled();
expect(onclose).not.toHaveBeenCalled();
Expand All @@ -77,7 +92,7 @@ for (const [name, scheme, closedCode] of drivers) {
const connectError = sql.connect().catch(e => e);
// close in the same tick: the pool slot exists but its native handle
// has not been assigned yet
await sql.close({ timeout: "0" });
await sql.close({ timeout: 0 });
expect((await connectError).code).toBe(closedCode);
} finally {
server.close();
Expand Down Expand Up @@ -125,7 +140,7 @@ test("postgres: close() mid-reconnect does not fire onclose for the unfinished c
// a new query redials the closed slot, then close() lands mid-handshake
const queryError = sql`SELECT 1`.catch(e => e);
await secondAccepted.promise;
await sql.close({ timeout: "0" });
await sql.close({ timeout: 0 });
expect((await queryError).code).toBe("ERR_POSTGRES_CONNECTION_CLOSED");
expect(onconnect).toHaveBeenCalledTimes(1);
expect(onclose).toHaveBeenCalledTimes(1);
Expand Down Expand Up @@ -175,7 +190,122 @@ test("pool scans tolerate unassigned connection slots during pool start", async
expect(errors).toEqual([]);
} finally {
// force an immediate close even with waiters queued
await sql.close({ timeout: "0" });
await sql.close({ timeout: 0 });
server.close();
}
});

// https://github.com/oven-sh/bun/issues/32038
//
// Each mock completes the handshake, holds the first query, and answers it with one text row when `respond()` is
// called. A close() that waits lets that answer resolve the query. A forced close() has already rejected it. Both
// outcomes settle, so a close() that wrongly waits fails an assertion instead of hanging.

Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
type HeldQueryMock = { port: number; server: Server; commandReceived: Promise<void>; respond: () => void };

// After the command arrives `received` is settled, so the reset caused by a forced close() is ignored.
function failUntilCommand(socket: Socket, received: PromiseWithResolvers<void>) {
socket.on("error", received.reject);
socket.on("close", () => received.reject(new Error("the client disconnected before it sent a command")));
}

const heldQueryMocks = {
async postgres(): Promise<HeldQueryMock> {
const received = Promise.withResolvers<void>();
let respond = () => {};
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 => {
if (type !== 0x51 /* Query */) return;
respond = () =>
socket.write(
Buffer.concat([
pgRowDescription([{ name: "x", typeOid: 25 }]),
pgDataRow([Buffer.from("1")]),
pgCommandComplete("SELECT 1"),
pgReadyForQuery(),
]),
);
received.resolve();
});
});
failUntilCommand(socket, received);
});
return { port, server, commandReceived: received.promise, respond: () => respond() };
Comment thread
robobun marked this conversation as resolved.
Resolved
},
async mysql(): Promise<HeldQueryMock> {
const received = Promise.withResolvers<void>();
let respond = () => {};
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;
}
if (mysqlAckSessionSetup(socket, payload)) return;
if (payload[0] !== 0x03 /* COM_QUERY */) return;
respond = () => socket.write(mysqlTextResultSet(seq + 1, [{ name: "x", type: 0xfd }], [["1"]]));
received.resolve();
});
});
failUntilCommand(socket, received);
});
return { port, server, commandReceived: received.promise, respond: () => respond() };
},
} as const;

// `false`, `""`, `"0"` and `null` are outside the declared `number` type. JS callers can still pass them.
const timeoutSpellings = [
["0", 0, "forced"],
['"0"', "0", "forced"],
["false", false, "forced"],
['""', "", "forced"],
["undefined", undefined, "drained"],
["null", null, "drained"],
["NaN", NaN, "invalid"],
["-1", -1, "invalid"],
] as const;

for (const [name, scheme, closedCode] of drivers) {
const rows = { rows: [{ x: "1" }] };
const outcomes = {
forced: { query: { code: closedCode }, close: "resolved" },
drained: { query: rows, close: "resolved" },
// The option is rejected before the pool is marked closed, so the query still completes.
invalid: { query: rows, close: "ERR_INVALID_ARG_VALUE" },
};

for (const [label, timeout, outcome] of timeoutSpellings) {
test(`${name}: close({ timeout: ${label} }) with a query in flight is ${outcome}`, async () => {
const { port, server, commandReceived, respond } = await heldQueryMocks[name]();
try {
await using sql = new SQL({ url: `${scheme}127.0.0.1:${port}/db`, max: 1 });
const query = sql`select 1 as x`.simple().then(
rows => ({ rows }),
e => ({ code: e.code }),
);
await commandReceived;
const close = sql.close({ timeout: timeout as any }).then(
() => "resolved",
e => e.code,
);
respond();
expect({ query: await query, close: await close }).toEqual(outcomes[outcome]);
} finally {
server.close();
}
});
}
}
2 changes: 1 addition & 1 deletion test/js/sql/sql-onconnect-onclose-throw.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -266,7 +266,7 @@ const sql = new SQL({
});
const queryError = sql.unsafe("SELECT 1").catch(err => err);
await accepted;
await sql.close({ timeout: "0" });
await sql.close({ timeout: 0 });
console.log("closed");
console.log("query rejected:", (await queryError).code);
process.exit(0);
Expand Down
Loading