From 1dcbed2588286c5e1d5004a5825de412ddb09ab2 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sun, 27 Sep 2026 09:19:16 +0000 Subject: [PATCH 1/4] sql: start a query in the call of then(), catch(), finally() and run() These four methods started a query through #runAsync(), which waited one promise job before it gave the query to the pool. A close() in the same tick counted the queries of the pool before that job, found none, and closed. The query was then rejected with ERR_*_CONNECTION_CLOSED and its statement was never sent. The four methods now start the query through #run(), as execute() does. The query reaches the pool in the call, so close() waits for it. sql.commitDistributed() and sql.rollbackDistributed() start their query in the call too. --- docs/runtime/sql.mdx | 8 + packages/bun-types/sql.d.ts | 6 + src/js/bun/sql.ts | 6 +- src/js/internal/sql/query.ts | 51 +-- .../sql/sql-close-pending-connection.test.ts | 269 ++++++++++++++++ test/js/sql/sqlite-sql.test.ts | 298 +++++++++++++++++- 6 files changed, 587 insertions(+), 51 deletions(-) diff --git a/docs/runtime/sql.mdx b/docs/runtime/sql.mdx index caf4609f3e4e..695e8646c1e3 100644 --- a/docs/runtime/sql.mdx +++ b/docs/runtime/sql.mdx @@ -980,6 +980,14 @@ await sql.close({ timeout: 0 }); // close all connections from the pool immediat `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. +`sql.close()` waits for the queries of the pool that started before the call. A query starts in the call of `.then()`, `.catch()`, `.finally()` or `.execute()`. `await query` and `Promise.all([query])` call `.then()` in a later microtask. When `close()` follows them in the same tick, it comes before the start, and the query rejects with the adapter's `ERR_*_CONNECTION_CLOSED` error. The same applies to a library that awaits the query in an async function of its own. + +```ts +sql`INSERT INTO users ${sql(alice)}`.catch(report); // started, close() waits for it +const inserted = (async () => await sql`INSERT INTO users ${sql(bob)}`)(); // not started yet +await sql.close(); // `inserted` rejects +``` + --- ## Reserved Connections diff --git a/packages/bun-types/sql.d.ts b/packages/bun-types/sql.d.ts index 86c7ec2fb7fa..46f3c965011b 100644 --- a/packages/bun-types/sql.d.ts +++ b/packages/bun-types/sql.d.ts @@ -683,6 +683,12 @@ declare module "bun" { * Closes the database connection. With `timeout: 0` it closes * immediately; with no timeout it waits for all queries to finish first. * + * It waits for the queries of the pool that started before the call. A + * query starts in the call of `.then()`, `.catch()`, `.finally()` or + * `.execute()`. `await query` calls `.then()` in a later microtask, so + * a `close()` that follows it in the same tick comes before the start, + * and the query rejects. + * * @param options Optional `timeout` in seconds * * @example diff --git a/src/js/bun/sql.ts b/src/js/bun/sql.ts index 17e3f5ed7aa8..68fb87a9d72d 100644 --- a/src/js/bun/sql.ts +++ b/src/js/bun/sql.ts @@ -993,7 +993,8 @@ const SQL = function SQL( } const sqlQuery = pool.getRollbackDistributedSQL(name); - return await sql.unsafe(sqlQuery); + // then() starts the query in this call. `await` alone starts it one promise job later. + return await sql.unsafe(sqlQuery).then(); }; sql.commitDistributed = async function (name: string) { @@ -1006,7 +1007,8 @@ const SQL = function SQL( } const sqlQuery = pool.getCommitDistributedSQL(name); - return await sql.unsafe(sqlQuery); + // then() starts the query in this call. `await` alone starts it one promise job later. + return await sql.unsafe(sqlQuery).then(); }; sql.beginDistributed = (name: string, fn: TransactionCallback) => { diff --git a/src/js/internal/sql/query.ts b/src/js/internal/sql/query.ts index b037f80a2308..c7595c4b947f 100644 --- a/src/js/internal/sql/query.ts +++ b/src/js/internal/sql/query.ts @@ -128,38 +128,6 @@ 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); - } - } - get active() { return (this[_queryStatus] & SQLQueryStatus.active) != 0; } @@ -240,7 +208,7 @@ class Query> extends PublicPromise { throw this[_adapter].notTaggedCallError(); } - await this.#runAsync(); + this.#run(); return this; } @@ -271,23 +239,12 @@ class Query> extends PublicPromise { return this; } - #runAsyncAndCatch() { - const runPromise = this.#runAsync(); - - if ($isPromise(runPromise) && runPromise !== this) { - runPromise.catch(() => { - // Error is already handled via this.reject() in #runAsync - // This catch is just to prevent unhandled rejection warnings - }); - } - } - then( onfulfilled?: ((value: T) => TResult1 | PromiseLike) | null, onrejected?: ((reason: any) => TResult2 | PromiseLike) | null, ): Promise; then() { - this.#runAsyncAndCatch(); + this.#run(); const result = super.$then.$apply(this, arguments); @@ -305,7 +262,7 @@ class Query> extends PublicPromise { throw this[_adapter].notTaggedCallError(); } - this.#runAsyncAndCatch(); + this.#run(); const result = super.catch.$apply(this, arguments); $pokePromiseAsHandled(result); @@ -318,7 +275,7 @@ class Query> extends PublicPromise { throw this[_adapter].notTaggedCallError(); } - this.#runAsyncAndCatch(); + this.#run(); return super.finally.$apply(this, arguments); } diff --git a/test/js/sql/sql-close-pending-connection.test.ts b/test/js/sql/sql-close-pending-connection.test.ts index 1686134e5906..2f442f706bb5 100644 --- a/test/js/sql/sql-close-pending-connection.test.ts +++ b/test/js/sql/sql-close-pending-connection.test.ts @@ -18,8 +18,10 @@ import { SQL } from "bun"; import { expect, mock, test } from "bun:test"; +import { describeWithContainer } from "harness"; import type { Server, Socket } from "node:net"; import { + closedPort, listeningServer, mysqlAckSessionSetup, mysqlHandshakeV10, @@ -296,3 +298,270 @@ for (const [name, scheme, closedCode] of drivers) { }); } } + +// https://github.com/oven-sh/bun/issues/43887 +// +// A query reaches the pool in the call that starts it, so a close() that follows in the same tick waits for it. +// Only a mock can never answer or refuse the connection, so these cases are here. The other cases use a real +// server, below. +for (const [name, scheme, closedCode] of drivers) { + const started = (sql: SQL) => + sql`select 1 as x`.then( + rows => ({ rows: [...rows] }), + e => ({ code: e.code }), + ); + + test(`${name}: close() waits no longer than connectionTimeout for a query that started before it`, async () => { + const { port, server, accepted } = await neverAnsweringServer(); + try { + const sql = new SQL({ url: `${scheme}127.0.0.1:${port}/db`, max: 1, connectionTimeout: 0.05 }); + const query = started(sql); + await sql.close(); + expect(await query).toEqual({ code: closedCode.replace("CONNECTION_CLOSED", "CONNECTION_TIMEOUT") }); + await accepted; + } finally { + server.close(); + } + }); + + test(`${name}: a refused connection rejects a query that started before close()`, async () => { + const sql = new SQL({ url: `${scheme}127.0.0.1:${await closedPort()}/db`, max: 1 }); + const query = started(sql); + await sql.close(); + expect(await query).toEqual({ code: closedCode.replace("CONNECTION_CLOSED", "CONNECTION_REFUSED") }); + }); +} + +// These tests use a real server. Each test has a table of its own, and a second pool reads that table after the +// close: a row shows that the server got the statement. +const settle = (promise: Promise) => + promise.then( + value => ({ value }), + e => ({ code: e.code }), + ); + +const starts = { + "then()": query => query.then(rows => rows), + "catch()": query => query.catch(e => Promise.reject(e)), + "finally()": query => query.finally(() => {}), + "run()": query => query.run(), + "execute()": query => query.execute(), +}; +// These call then() from a later promise job, so a close() in the same tick comes before the start. +const laterStarts = { + "`await` in a function that nobody awaits": query => (async () => await query)(), + "Promise.all()": query => Promise.all([query]), + "Promise.resolve()": query => Promise.resolve(query), +}; +const closes = { + "close({ timeout: 5 })": sql => sql.close({ timeout: 5 }), + "end()": sql => sql.end(), + "[Symbol.asyncDispose]()": sql => sql[Symbol.asyncDispose](), +}; + +const servers = [ + { adapter: "postgres", image: "postgres_plain", user: "bun_sql_test", serial: "SERIAL PRIMARY KEY" }, + { adapter: "mysql", image: "mysql_plain", user: "root", serial: "INT AUTO_INCREMENT PRIMARY KEY" }, +] as const; + +for (const { adapter, image, user, serial } of servers) { + const code = (name: string) => `ERR_${adapter.toUpperCase()}_${name}`; + + describeWithContainer(`${adapter}: close() and the queries that started before it`, { image }, container => { + // `warm: false` leaves the pool without a connection. + async function openTable({ warm = true, ...options }: Bun.SQL.Options & { warm?: boolean } = {}) { + await container.ready; + const url = `${adapter}://${user}@${container.host}:${container.port}/bun_sql_test`; + const reader = new SQL({ url, max: 1 }); + const table = "close_" + crypto.randomUUID().replaceAll("-", ""); + await reader.unsafe(`CREATE TABLE ${table} (id ${serial}, x INT)`); + const sql = new SQL({ url, max: 1, ...options }); + if (warm) await sql.connect(); + return { + sql, + table, + insert: (x: number, on: SQL = sql) => on`INSERT INTO ${sql(table)} (x) VALUES (${x})`, + rows: async () => (await reader.unsafe(`SELECT x FROM ${table} ORDER BY id`)).map(row => row.x), + async [Symbol.asyncDispose]() { + await sql.close({ timeout: 0 }); + await reader.unsafe(`DROP TABLE ${table}`); + await reader.close(); + }, + }; + } + + for (const [name, start] of Object.entries(starts)) { + test(`close() waits for a query that ${name} started`, async () => { + await using table = await openTable(); + const inserted = settle(start(table.insert(1)).then(() => "done")); + await table.sql.close(); + expect({ query: await inserted, rows: await table.rows() }).toEqual({ query: { value: "done" }, rows: [1] }); + }); + } + + for (const [name, close] of Object.entries(closes)) { + test(`${name} waits for a query that started before it`, async () => { + await using table = await openTable(); + const inserted = settle(table.insert(1).then(() => "done")); + await close(table.sql); + expect({ query: await inserted, rows: await table.rows() }).toEqual({ query: { value: "done" }, rows: [1] }); + }); + } + + for (const [name, { rows, start }] of Object.entries({ + "sql.unsafe()": { rows: [1], start: ({ sql, table }) => sql.unsafe(`INSERT INTO ${table} (x) VALUES (1)`) }, + "two statements in simple()": { + rows: [1, 2], + start: ({ sql, table }) => + sql.unsafe(`INSERT INTO ${table} (x) VALUES (1); INSERT INTO ${table} (x) VALUES (2)`).simple(), + }, + })) { + test(`close() waits for ${name}`, async () => { + await using table = await openTable(); + const inserted = settle(start(table).then(() => "done")); + await table.sql.close(); + expect({ query: await inserted, rows: await table.rows() }).toEqual({ query: { value: "done" }, rows }); + }); + } + + test("an `await using` block runs the 20 inserts that it did not await", async () => { + await using table = await openTable(); + const errors: unknown[] = []; + { + await using _ = table.sql; + for (let x = 0; x < 20; x++) table.insert(x).catch(e => errors.push(e.code)); + } + expect({ errors, rows: await table.rows() }).toEqual({ + errors: [], + rows: Array.from({ length: 20 }, (_, x) => x), + }); + }); + + test("queries reach the server in the order of their starts", async () => { + await using table = await openTable(); + const expected = Array.from({ length: 50 }, (_, x) => x); + const names = Object.keys(starts); + const inserted = expected.map(x => settle(starts[names[x % names.length]](table.insert(x)).then(() => x))); + await table.sql.close(); + expect({ queries: await Promise.all(inserted), rows: await table.rows() }).toEqual({ + queries: expected.map(value => ({ value })), + rows: expected, + }); + }); + + test("two close() calls in the same tick wait for a started query", async () => { + await using table = await openTable(); + const inserted = settle(table.insert(1).then(() => "done")); + await Promise.all([table.sql.close(), table.sql.close()]); + expect({ query: await inserted, rows: await table.rows() }).toEqual({ query: { value: "done" }, rows: [1] }); + }); + + test("close() waits for a query that started on a pool with no connection", async () => { + await using table = await openTable({ warm: false }); + const inserted = settle(table.insert(1).then(() => "done")); + await table.sql.close(); + expect({ query: await inserted, rows: await table.rows() }).toEqual({ query: { value: "done" }, rows: [1] }); + }); + + for (const [name, statement] of [ + ["commitDistributed", "COMMIT"], + ["rollbackDistributed", "ROLLBACK"], + ] as const) { + // No such transaction exists, so the error of the server shows that the server got the statement. + test(`close() waits for the ${statement} of sql.${name}() that was called before it`, async () => { + await using table = await openTable(); + const finished = settle(table.sql[name]("close_no_such_transaction")); + await table.sql.close(); + expect(await finished).toEqual({ code: code("SERVER_ERROR") }); + }); + } + + test("close() waits for a started query when a transaction has the only connection", async () => { + await using table = await openTable(); + const order: string[] = []; + const open = Promise.withResolvers(); + const gate = Promise.withResolvers(); + const transaction = table.sql + .begin(async tx => { + await table.insert(1, tx); + open.resolve(); + await gate.promise; + await table.insert(2, tx); + }) + .finally(() => order.push("transaction")); + await open.promise; + const inserted = settle(table.insert(3).then(() => "done")).finally(() => order.push("query")); + const closed = table.sql.close().finally(() => order.push("close()")); + gate.resolve(); + await Promise.all([transaction, closed]); + expect({ query: await inserted, rows: await table.rows(), order }).toEqual({ + query: { value: "done" }, + rows: [1, 2, 3], + order: ["transaction", "query", "close()"], + }); + }); + + test("close() waits for a started query when a reservation has the only connection", async () => { + await using table = await openTable(); + const order: string[] = []; + const reserved = await table.sql.reserve(); + await table.insert(1, reserved); + const inserted = settle(table.insert(3).then(() => "done")).finally(() => order.push("query")); + const closed = table.sql.close().finally(() => order.push("close()")); + await table.insert(2, reserved); + reserved.release(); + await closed; + expect({ query: await inserted, rows: await table.rows(), order }).toEqual({ + query: { value: "done" }, + rows: [1, 2, 3], + order: ["query", "close()"], + }); + }); + + test("a query that starts after close() is rejected and never sent", async () => { + await using table = await openTable(); + const early = settle(table.insert(1).then(() => "done")); + const closed = table.sql.close(); + const late = [settle(table.insert(2).execute()), settle(table.insert(3).then(rows => rows))]; + await closed; + expect({ early: await early, late: await Promise.all(late), rows: await table.rows() }).toEqual({ + early: { value: "done" }, + late: [{ code: code("CONNECTION_CLOSED") }, { code: code("CONNECTION_CLOSED") }], + rows: [1], + }); + }); + + for (const [name, start] of Object.entries(laterStarts)) { + test(`a query that only ${name} started is rejected and never sent`, async () => { + await using table = await openTable(); + const inserted = settle(start(table.insert(1))); + await table.sql.close(); + expect({ query: await inserted, rows: await table.rows() }).toEqual({ + query: { code: code("CONNECTION_CLOSED") }, + rows: [], + }); + }); + } + + // What cancel() does to a query that started is not in these tests. The way that the query started must not + // change it. + for (const warm of [true, false]) { + for (const [name, start] of Object.entries(starts)) { + if (name === "execute()") continue; + const pool = warm ? "a pool with a connection" : "a pool with no connection"; + test(`cancel() in the tick of ${name} does what it does in the tick of execute(), on ${pool}`, async () => { + async function cancelAfter(start: (query: any) => Promise) { + await using table = await openTable({ warm }); + const query = table.insert(1); + const settled = settle(start(query).then(() => "done")); + query.cancel(); + const outcome = await settled; + await table.sql.close(); + return { query: outcome, rows: await table.rows() }; + } + expect(await cancelAfter(start)).toEqual(await cancelAfter(starts["execute()"])); + }); + } + } + }); +} diff --git a/test/js/sql/sqlite-sql.test.ts b/test/js/sql/sqlite-sql.test.ts index ed0567c186fb..bad9c1cf308f 100644 --- a/test/js/sql/sqlite-sql.test.ts +++ b/test/js/sql/sqlite-sql.test.ts @@ -1,7 +1,7 @@ import { randomUUIDv7, SQL } from "bun"; import { Database } from "bun:sqlite"; -import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, mock, test } from "bun:test"; -import { isDebug, tempDir } from "harness"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, mock, spyOn, test } from "bun:test"; +import { bunEnv, bunExe, isDebug, tempDir } from "harness"; import { existsSync } from "node:fs"; import { rm, stat } from "node:fs/promises"; import { join } from "node:path"; @@ -2080,6 +2080,168 @@ describe("Connection management", () => { } }); + // https://github.com/oven-sh/bun/issues/43887 + describe("close() and the queries that started before it", () => { + const settle = (promise: Promise) => + promise.then( + rows => ({ rows }), + e => ({ code: e.code }), + ); + + function rowsIn(filename: string) { + const db = new Database(filename, { readonly: true }); + try { + return db.query("SELECT x FROM t ORDER BY rowid").values().flat(); + } finally { + db.close(); + } + } + + async function openTable(dir: { toString(): string }) { + const filename = join(String(dir), "t.db"); + const sql = new SQL({ adapter: "sqlite", filename }); + await sql`CREATE TABLE t (x INTEGER)`; + return { sql, filename }; + } + + const starts = { + "then()": query => query.then(rows => rows), + "catch()": query => query.catch(e => Promise.reject(e)), + "finally()": query => query.finally(() => {}), + "run()": query => query.run(), + "execute()": query => query.execute(), + }; + const closes = { + "close({ timeout: 5 })": sql => sql.close({ timeout: 5 }), + "end()": sql => sql.end(), + "[Symbol.asyncDispose]()": sql => sql[Symbol.asyncDispose](), + }; + const statements = { + "with a parameter": { rows: [1], start: sql => sql`INSERT INTO t VALUES (${1})` }, + "that returns rows": { rows: [1], start: sql => sql`INSERT INTO t VALUES (1) RETURNING x` }, + "in a string of two statements": { + rows: [1, 2], + start: sql => sql.unsafe("INSERT INTO t VALUES (1); INSERT INTO t VALUES (2)"), + }, + }; + + for (const [name, start] of Object.entries(starts)) { + test(`a query that ${name} started runs when close() follows in the same tick`, async () => { + using dir = tempDir("sqlite-close", {}); + const { sql, filename } = await openTable(dir); + const inserted = settle(start(sql`INSERT INTO t VALUES (1)`).then(() => "done")); + await sql.close(); + expect({ query: await inserted, rows: rowsIn(filename) }).toEqual({ query: { rows: "done" }, rows: [1] }); + }); + } + + for (const [name, close] of Object.entries(closes)) { + test(`a query that started runs when ${name} follows in the same tick`, async () => { + using dir = tempDir("sqlite-close", {}); + const { sql, filename } = await openTable(dir); + const inserted = settle(sql`INSERT INTO t VALUES (1)`.then(() => "done")); + await close(sql); + expect({ query: await inserted, rows: rowsIn(filename) }).toEqual({ query: { rows: "done" }, rows: [1] }); + }); + } + + for (const [name, { rows, start }] of Object.entries(statements)) { + test(`a statement ${name} runs when close() follows in the same tick`, async () => { + using dir = tempDir("sqlite-close", {}); + const { sql, filename } = await openTable(dir); + const inserted = settle(start(sql).then(() => "done")); + await sql.close(); + expect({ query: await inserted, rows: rowsIn(filename) }).toEqual({ query: { rows: "done" }, rows }); + }); + } + + test("an `await using` block runs the 20 inserts that it did not await", async () => { + using dir = tempDir("sqlite-close", {}); + const filename = join(String(dir), "t.db"); + const errors: unknown[] = []; + { + await using db = new SQL({ adapter: "sqlite", filename }); + await db`CREATE TABLE IF NOT EXISTS t (x)`; + for (let i = 0; i < 20; i++) db`INSERT INTO t VALUES (${i})`.catch(e => errors.push(e.code)); + } + expect({ errors, rows: rowsIn(filename) }).toEqual({ errors: [], rows: Array.from({ length: 20 }, (_, i) => i) }); + }); + + for (const exit of ["return", "throw"] as const) { + test(`an \`await using\` block left by ${exit} runs the queries that started in it`, async () => { + using dir = tempDir("sqlite-close", {}); + const filename = join(String(dir), "t.db"); + const inserted: Promise[] = []; + async function block() { + await using sql = new SQL({ adapter: "sqlite", filename }); + await sql`CREATE TABLE t (x INTEGER)`; + for (let x = 0; x < 3; x++) inserted.push(settle(sql`INSERT INTO t VALUES (${x})`.then(() => x))); + if (exit === "return") return; + throw new Error("leave the block"); + } + await block().catch(() => {}); + expect({ queries: await Promise.all(inserted), rows: rowsIn(filename) }).toEqual({ + queries: [{ rows: 0 }, { rows: 1 }, { rows: 2 }], + rows: [0, 1, 2], + }); + }); + } + + test("two close() calls in the same tick leave a started query to run once", async () => { + using dir = tempDir("sqlite-close", {}); + const { sql, filename } = await openTable(dir); + const inserted = settle(sql`INSERT INTO t VALUES (1)`.then(() => "done")); + await Promise.all([sql.close(), sql.close()]); + expect({ query: await inserted, rows: rowsIn(filename) }).toEqual({ query: { rows: "done" }, rows: [1] }); + }); + + test("the COMMIT of a transaction that the caller made with BEGIN runs when close() follows", async () => { + using dir = tempDir("sqlite-close", {}); + const { sql, filename } = await openTable(dir); + await sql`BEGIN`; + await sql`INSERT INTO t VALUES (1)`; + const committed = settle(sql`COMMIT`.then(() => "done")); + await sql.close(); + expect({ commit: await committed, rows: rowsIn(filename) }).toEqual({ commit: { rows: "done" }, rows: [1] }); + }); + + test("a query that starts after close() is rejected and does not run", async () => { + using dir = tempDir("sqlite-close", {}); + const { sql, filename } = await openTable(dir); + const early = settle(sql`INSERT INTO t VALUES (1)`.then(() => "done")); + const closed = sql.close(); + const late = [ + settle(sql`INSERT INTO t VALUES (2)`.execute()), + settle(sql`INSERT INTO t VALUES (3)`.then(rows => rows)), + ]; + await closed; + expect({ early: await early, late: await Promise.all(late), rows: rowsIn(filename) }).toEqual({ + early: { rows: "done" }, + late: [{ code: "ERR_SQLITE_CONNECTION_CLOSED" }, { code: "ERR_SQLITE_CONNECTION_CLOSED" }], + rows: [1], + }); + }); + + // `await`, Promise.all() and Promise.resolve() call then() from a later promise job, so a close() in the same + // tick comes before the start. + for (const [name, start] of Object.entries({ + "`await` in a function that nobody awaits": query => (async () => await query)(), + "Promise.all()": query => Promise.all([query]), + "Promise.resolve()": query => Promise.resolve(query), + })) { + test(`a query that only ${name} started is rejected`, async () => { + using dir = tempDir("sqlite-close", {}); + const { sql, filename } = await openTable(dir); + const inserted = settle(start(sql`INSERT INTO t VALUES (1)`)); + await sql.close(); + expect({ query: await inserted, rows: rowsIn(filename) }).toEqual({ + query: { code: "ERR_SQLITE_CONNECTION_CLOSED" }, + rows: [], + }); + }); + } + }); + test("reserve throws for SQLite", async () => { const sql = new SQL("sqlite://:memory:"); @@ -2119,6 +2281,138 @@ describe("Connection management", () => { }); }); +// then(), catch(), finally() and run() start a query as execute() does: the query reaches the database in the call. +describe("Query start", () => { + const starts = { + "then()": query => query.then(rows => rows), + "catch()": query => query.catch(e => Promise.reject(e)), + "finally()": query => query.finally(() => {}), + "run()": query => query.run(), + "execute()": query => query.execute(), + }; + + for (const [name, start] of Object.entries(starts)) { + test(`${name} gives the query to the database in the call`, async () => { + const sql = new SQL("sqlite://:memory:"); + const order: string[] = []; + const prepare = Database.prototype.prepare; + const spy = spyOn(Database.prototype, "prepare").mockImplementation(function (this: Database, ...args: any) { + order.push("database"); + return prepare.apply(this, args); + } as typeof prepare); + try { + const started = start(sql`SELECT 1 AS x`); + order.push("the call returned"); + await started; + expect(order).toEqual(["database", "the call returned"]); + } finally { + spy.mockRestore(); + await sql.close(); + } + }); + } + + test("queries reach the database in the order of their starts", async () => { + const sql = new SQL("sqlite://:memory:"); + try { + await sql`CREATE TABLE t (id INTEGER PRIMARY KEY AUTOINCREMENT, x INTEGER)`; + const expected = Array.from({ length: 200 }, (_, x) => x); + const names = Object.keys(starts); + await Promise.all(expected.map(x => starts[names[x % names.length]](sql`INSERT INTO t (x) VALUES (${x})`))); + expect((await sql`SELECT x FROM t ORDER BY id`).map(row => row.x)).toEqual(expected); + } finally { + await sql.close(); + } + }); + + // What cancel() does to a query that started is not in this test. The way that the query started must not change it. + for (const [name, start] of Object.entries(starts)) { + if (name === "execute()") continue; + test(`cancel() in the tick of ${name} does what it does in the tick of execute()`, async () => { + async function cancelAfter(start: (query: any) => Promise) { + const sql = new SQL("sqlite://:memory:"); + try { + await sql`CREATE TABLE t (x INTEGER)`; + const query = sql`INSERT INTO t VALUES (1)`; + const settled = start(query).then( + () => "resolved", + e => e.code, + ); + query.cancel(); + return { query: await settled, rows: (await sql`SELECT x FROM t`).length }; + } finally { + await sql.close(); + } + } + expect(await cancelAfter(start)).toEqual(await cancelAfter(starts["execute()"])); + }); + } + + // Runs in a child process: bun:test turns an unhandled rejection into a test failure. + test("a query that fails is reported as unhandled only when the caller gave it no rejection handler", async () => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + ` + const reported = []; + process.on("unhandledRejection", err => reported.push(err.code)); + const starts = { + "then(f)": query => query.then(() => {}), + "then(f, g)": query => query.then(() => {}, () => {}), + "then()": query => query.then(), + "catch(g)": query => query.catch(() => {}), + "catch()": query => query.catch(), + "finally(f)": query => query.finally(() => {}), + "run()": query => query.run(), + "execute()": query => query.execute(), + "await": query => void (async () => await query)(), + }; + const failures = { + "the statement fails": sql => sql\`SELECT * FROM missing\`, + "the pool is closed": sql => (sql.close(), sql\`SELECT 1\`), + }; + const events = {}; + for (const [failure, fail] of Object.entries(failures)) { + events[failure] = {}; + for (const [name, start] of Object.entries(starts)) { + const sql = new Bun.SQL("sqlite://:memory:"); + reported.length = 0; + start(fail(sql)); + // Unhandled rejections are reported after the promise jobs of the turn that made them. + await new Promise(resolve => setImmediate(resolve)); + events[failure][name] = [...reported]; + await sql.close(); + } + } + console.log(JSON.stringify(events)); + `, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + const eventsFor = (code: string) => ({ + "then(f)": [code], + "then(f, g)": [], + "then()": [code], + "catch(g)": [], + "catch()": [], + "finally(f)": [], + "run()": [code], + "execute()": [code], + "await": [code], + }); + expect(stderr).toBe(""); + expect(JSON.parse(stdout)).toEqual({ + "the statement fails": eventsFor("SQLITE_ERROR"), + "the pool is closed": eventsFor("ERR_SQLITE_CONNECTION_CLOSED"), + }); + expect(exitCode).toBe(0); + }); +}); + describe("Performance & Edge Cases", () => { test("handles large datasets", async () => { const sql = new SQL("sqlite://:memory:"); From 10f824b017ccdb4a0c1604048bf04423a9c88f39 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Mon, 28 Sep 2026 22:39:17 +0000 Subject: [PATCH 2/4] test(sql): put the tests of close() with a server in a file of their own The PostgreSQL and MySQL tests are now in sql-close-started-queries.test.ts. The tests of a block share one table. Each test writes rows with an id of its own. sql-close-pending-connection.test.ts is as on the base again. --- .../sql/sql-close-pending-connection.test.ts | 269 ----------------- test/js/sql/sql-close-started-queries.test.ts | 284 ++++++++++++++++++ 2 files changed, 284 insertions(+), 269 deletions(-) create mode 100644 test/js/sql/sql-close-started-queries.test.ts diff --git a/test/js/sql/sql-close-pending-connection.test.ts b/test/js/sql/sql-close-pending-connection.test.ts index 2f442f706bb5..1686134e5906 100644 --- a/test/js/sql/sql-close-pending-connection.test.ts +++ b/test/js/sql/sql-close-pending-connection.test.ts @@ -18,10 +18,8 @@ import { SQL } from "bun"; import { expect, mock, test } from "bun:test"; -import { describeWithContainer } from "harness"; import type { Server, Socket } from "node:net"; import { - closedPort, listeningServer, mysqlAckSessionSetup, mysqlHandshakeV10, @@ -298,270 +296,3 @@ for (const [name, scheme, closedCode] of drivers) { }); } } - -// https://github.com/oven-sh/bun/issues/43887 -// -// A query reaches the pool in the call that starts it, so a close() that follows in the same tick waits for it. -// Only a mock can never answer or refuse the connection, so these cases are here. The other cases use a real -// server, below. -for (const [name, scheme, closedCode] of drivers) { - const started = (sql: SQL) => - sql`select 1 as x`.then( - rows => ({ rows: [...rows] }), - e => ({ code: e.code }), - ); - - test(`${name}: close() waits no longer than connectionTimeout for a query that started before it`, async () => { - const { port, server, accepted } = await neverAnsweringServer(); - try { - const sql = new SQL({ url: `${scheme}127.0.0.1:${port}/db`, max: 1, connectionTimeout: 0.05 }); - const query = started(sql); - await sql.close(); - expect(await query).toEqual({ code: closedCode.replace("CONNECTION_CLOSED", "CONNECTION_TIMEOUT") }); - await accepted; - } finally { - server.close(); - } - }); - - test(`${name}: a refused connection rejects a query that started before close()`, async () => { - const sql = new SQL({ url: `${scheme}127.0.0.1:${await closedPort()}/db`, max: 1 }); - const query = started(sql); - await sql.close(); - expect(await query).toEqual({ code: closedCode.replace("CONNECTION_CLOSED", "CONNECTION_REFUSED") }); - }); -} - -// These tests use a real server. Each test has a table of its own, and a second pool reads that table after the -// close: a row shows that the server got the statement. -const settle = (promise: Promise) => - promise.then( - value => ({ value }), - e => ({ code: e.code }), - ); - -const starts = { - "then()": query => query.then(rows => rows), - "catch()": query => query.catch(e => Promise.reject(e)), - "finally()": query => query.finally(() => {}), - "run()": query => query.run(), - "execute()": query => query.execute(), -}; -// These call then() from a later promise job, so a close() in the same tick comes before the start. -const laterStarts = { - "`await` in a function that nobody awaits": query => (async () => await query)(), - "Promise.all()": query => Promise.all([query]), - "Promise.resolve()": query => Promise.resolve(query), -}; -const closes = { - "close({ timeout: 5 })": sql => sql.close({ timeout: 5 }), - "end()": sql => sql.end(), - "[Symbol.asyncDispose]()": sql => sql[Symbol.asyncDispose](), -}; - -const servers = [ - { adapter: "postgres", image: "postgres_plain", user: "bun_sql_test", serial: "SERIAL PRIMARY KEY" }, - { adapter: "mysql", image: "mysql_plain", user: "root", serial: "INT AUTO_INCREMENT PRIMARY KEY" }, -] as const; - -for (const { adapter, image, user, serial } of servers) { - const code = (name: string) => `ERR_${adapter.toUpperCase()}_${name}`; - - describeWithContainer(`${adapter}: close() and the queries that started before it`, { image }, container => { - // `warm: false` leaves the pool without a connection. - async function openTable({ warm = true, ...options }: Bun.SQL.Options & { warm?: boolean } = {}) { - await container.ready; - const url = `${adapter}://${user}@${container.host}:${container.port}/bun_sql_test`; - const reader = new SQL({ url, max: 1 }); - const table = "close_" + crypto.randomUUID().replaceAll("-", ""); - await reader.unsafe(`CREATE TABLE ${table} (id ${serial}, x INT)`); - const sql = new SQL({ url, max: 1, ...options }); - if (warm) await sql.connect(); - return { - sql, - table, - insert: (x: number, on: SQL = sql) => on`INSERT INTO ${sql(table)} (x) VALUES (${x})`, - rows: async () => (await reader.unsafe(`SELECT x FROM ${table} ORDER BY id`)).map(row => row.x), - async [Symbol.asyncDispose]() { - await sql.close({ timeout: 0 }); - await reader.unsafe(`DROP TABLE ${table}`); - await reader.close(); - }, - }; - } - - for (const [name, start] of Object.entries(starts)) { - test(`close() waits for a query that ${name} started`, async () => { - await using table = await openTable(); - const inserted = settle(start(table.insert(1)).then(() => "done")); - await table.sql.close(); - expect({ query: await inserted, rows: await table.rows() }).toEqual({ query: { value: "done" }, rows: [1] }); - }); - } - - for (const [name, close] of Object.entries(closes)) { - test(`${name} waits for a query that started before it`, async () => { - await using table = await openTable(); - const inserted = settle(table.insert(1).then(() => "done")); - await close(table.sql); - expect({ query: await inserted, rows: await table.rows() }).toEqual({ query: { value: "done" }, rows: [1] }); - }); - } - - for (const [name, { rows, start }] of Object.entries({ - "sql.unsafe()": { rows: [1], start: ({ sql, table }) => sql.unsafe(`INSERT INTO ${table} (x) VALUES (1)`) }, - "two statements in simple()": { - rows: [1, 2], - start: ({ sql, table }) => - sql.unsafe(`INSERT INTO ${table} (x) VALUES (1); INSERT INTO ${table} (x) VALUES (2)`).simple(), - }, - })) { - test(`close() waits for ${name}`, async () => { - await using table = await openTable(); - const inserted = settle(start(table).then(() => "done")); - await table.sql.close(); - expect({ query: await inserted, rows: await table.rows() }).toEqual({ query: { value: "done" }, rows }); - }); - } - - test("an `await using` block runs the 20 inserts that it did not await", async () => { - await using table = await openTable(); - const errors: unknown[] = []; - { - await using _ = table.sql; - for (let x = 0; x < 20; x++) table.insert(x).catch(e => errors.push(e.code)); - } - expect({ errors, rows: await table.rows() }).toEqual({ - errors: [], - rows: Array.from({ length: 20 }, (_, x) => x), - }); - }); - - test("queries reach the server in the order of their starts", async () => { - await using table = await openTable(); - const expected = Array.from({ length: 50 }, (_, x) => x); - const names = Object.keys(starts); - const inserted = expected.map(x => settle(starts[names[x % names.length]](table.insert(x)).then(() => x))); - await table.sql.close(); - expect({ queries: await Promise.all(inserted), rows: await table.rows() }).toEqual({ - queries: expected.map(value => ({ value })), - rows: expected, - }); - }); - - test("two close() calls in the same tick wait for a started query", async () => { - await using table = await openTable(); - const inserted = settle(table.insert(1).then(() => "done")); - await Promise.all([table.sql.close(), table.sql.close()]); - expect({ query: await inserted, rows: await table.rows() }).toEqual({ query: { value: "done" }, rows: [1] }); - }); - - test("close() waits for a query that started on a pool with no connection", async () => { - await using table = await openTable({ warm: false }); - const inserted = settle(table.insert(1).then(() => "done")); - await table.sql.close(); - expect({ query: await inserted, rows: await table.rows() }).toEqual({ query: { value: "done" }, rows: [1] }); - }); - - for (const [name, statement] of [ - ["commitDistributed", "COMMIT"], - ["rollbackDistributed", "ROLLBACK"], - ] as const) { - // No such transaction exists, so the error of the server shows that the server got the statement. - test(`close() waits for the ${statement} of sql.${name}() that was called before it`, async () => { - await using table = await openTable(); - const finished = settle(table.sql[name]("close_no_such_transaction")); - await table.sql.close(); - expect(await finished).toEqual({ code: code("SERVER_ERROR") }); - }); - } - - test("close() waits for a started query when a transaction has the only connection", async () => { - await using table = await openTable(); - const order: string[] = []; - const open = Promise.withResolvers(); - const gate = Promise.withResolvers(); - const transaction = table.sql - .begin(async tx => { - await table.insert(1, tx); - open.resolve(); - await gate.promise; - await table.insert(2, tx); - }) - .finally(() => order.push("transaction")); - await open.promise; - const inserted = settle(table.insert(3).then(() => "done")).finally(() => order.push("query")); - const closed = table.sql.close().finally(() => order.push("close()")); - gate.resolve(); - await Promise.all([transaction, closed]); - expect({ query: await inserted, rows: await table.rows(), order }).toEqual({ - query: { value: "done" }, - rows: [1, 2, 3], - order: ["transaction", "query", "close()"], - }); - }); - - test("close() waits for a started query when a reservation has the only connection", async () => { - await using table = await openTable(); - const order: string[] = []; - const reserved = await table.sql.reserve(); - await table.insert(1, reserved); - const inserted = settle(table.insert(3).then(() => "done")).finally(() => order.push("query")); - const closed = table.sql.close().finally(() => order.push("close()")); - await table.insert(2, reserved); - reserved.release(); - await closed; - expect({ query: await inserted, rows: await table.rows(), order }).toEqual({ - query: { value: "done" }, - rows: [1, 2, 3], - order: ["query", "close()"], - }); - }); - - test("a query that starts after close() is rejected and never sent", async () => { - await using table = await openTable(); - const early = settle(table.insert(1).then(() => "done")); - const closed = table.sql.close(); - const late = [settle(table.insert(2).execute()), settle(table.insert(3).then(rows => rows))]; - await closed; - expect({ early: await early, late: await Promise.all(late), rows: await table.rows() }).toEqual({ - early: { value: "done" }, - late: [{ code: code("CONNECTION_CLOSED") }, { code: code("CONNECTION_CLOSED") }], - rows: [1], - }); - }); - - for (const [name, start] of Object.entries(laterStarts)) { - test(`a query that only ${name} started is rejected and never sent`, async () => { - await using table = await openTable(); - const inserted = settle(start(table.insert(1))); - await table.sql.close(); - expect({ query: await inserted, rows: await table.rows() }).toEqual({ - query: { code: code("CONNECTION_CLOSED") }, - rows: [], - }); - }); - } - - // What cancel() does to a query that started is not in these tests. The way that the query started must not - // change it. - for (const warm of [true, false]) { - for (const [name, start] of Object.entries(starts)) { - if (name === "execute()") continue; - const pool = warm ? "a pool with a connection" : "a pool with no connection"; - test(`cancel() in the tick of ${name} does what it does in the tick of execute(), on ${pool}`, async () => { - async function cancelAfter(start: (query: any) => Promise) { - await using table = await openTable({ warm }); - const query = table.insert(1); - const settled = settle(start(query).then(() => "done")); - query.cancel(); - const outcome = await settled; - await table.sql.close(); - return { query: outcome, rows: await table.rows() }; - } - expect(await cancelAfter(start)).toEqual(await cancelAfter(starts["execute()"])); - }); - } - } - }); -} diff --git a/test/js/sql/sql-close-started-queries.test.ts b/test/js/sql/sql-close-started-queries.test.ts new file mode 100644 index 000000000000..a56064ae2f35 --- /dev/null +++ b/test/js/sql/sql-close-started-queries.test.ts @@ -0,0 +1,284 @@ +// https://github.com/oven-sh/bun/issues/43887 +// +// A query reaches the pool in the call that starts it: in the call of then(), catch(), finally(), run() or +// execute(). So a close() that follows in the same tick waits for the query. +// +// Most tests use a real server. Each test writes rows with an id of its own, and a second pool reads them after +// the close: a row shows that the server got the statement. +import { SQL } from "bun"; +import { afterAll, beforeAll, expect, test } from "bun:test"; +import { describeWithContainer } from "harness"; +import { closedPort, neverAnsweringServer } from "./wire-frames"; + +// Fault-injection tests: a server that never completes the handshake, and a port that refuses the connection. A +// healthy server does neither. +const drivers = [ + ["postgres", "postgres://postgres@", "ERR_POSTGRES_CONNECTION_CLOSED"], + ["mysql", "mysql://root@", "ERR_MYSQL_CONNECTION_CLOSED"], +] as const; + +for (const [name, scheme, closedCode] of drivers) { + const started = (sql: SQL) => + sql`select 1 as x`.then( + rows => ({ rows: [...rows] }), + e => ({ code: e.code }), + ); + + test(`${name}: close() waits no longer than connectionTimeout for a query that started before it`, async () => { + const { port, server, accepted } = await neverAnsweringServer(); + try { + const sql = new SQL({ url: `${scheme}127.0.0.1:${port}/db`, max: 1, connectionTimeout: 0.05 }); + const query = started(sql); + await sql.close(); + expect(await query).toEqual({ code: closedCode.replace("CONNECTION_CLOSED", "CONNECTION_TIMEOUT") }); + await accepted; + } finally { + server.close(); + } + }); + + test(`${name}: a refused connection rejects a query that started before close()`, async () => { + const sql = new SQL({ url: `${scheme}127.0.0.1:${await closedPort()}/db`, max: 1 }); + const query = started(sql); + await sql.close(); + expect(await query).toEqual({ code: closedCode.replace("CONNECTION_CLOSED", "CONNECTION_REFUSED") }); + }); +} + +const settle = (promise: Promise) => + promise.then( + value => ({ value }), + e => ({ code: e.code }), + ); + +const starts = { + "then()": query => query.then(rows => rows), + "catch()": query => query.catch(e => Promise.reject(e)), + "finally()": query => query.finally(() => {}), + "run()": query => query.run(), + "execute()": query => query.execute(), +}; +// These call then() from a later promise job, so a close() in the same tick comes before the start. +const laterStarts = { + "`await` in a function that nobody awaits": query => (async () => await query)(), + "Promise.all()": query => Promise.all([query]), + "Promise.resolve()": query => Promise.resolve(query), +}; +const closes = { + "close({ timeout: 5 })": sql => sql.close({ timeout: 5 }), + "end()": sql => sql.end(), + "[Symbol.asyncDispose]()": sql => sql[Symbol.asyncDispose](), +}; + +const servers = [ + { adapter: "postgres", image: "postgres_plain", user: "bun_sql_test", serial: "SERIAL PRIMARY KEY" }, + { adapter: "mysql", image: "mysql_plain", user: "root", serial: "INT AUTO_INCREMENT PRIMARY KEY" }, +] as const; + +for (const { adapter, image, user, serial } of servers) { + const code = (name: string) => `ERR_${adapter.toUpperCase()}_${name}`; + + describeWithContainer(`${adapter}: close() and the queries that started before it`, { image }, container => { + const url = () => `${adapter}://${user}@${container.host}:${container.port}/bun_sql_test`; + const tableName = "close_" + crypto.randomUUID().replaceAll("-", ""); + let reader: SQL; + // One table for all the tests of the block. The hooks have the time that a busy server can need for DDL. + beforeAll(async () => { + await container.ready; + reader = new SQL({ url: url(), max: 1 }); + await reader.unsafe(`CREATE TABLE ${tableName} (id ${serial}, test CHAR(36), x INT)`); + }, 60_000); + afterAll(async () => { + await reader.unsafe(`DROP TABLE ${tableName}`); + await reader.close(); + }, 60_000); + + // `warm: false` leaves the pool without a connection. + async function openTable({ warm = true, ...options }: Bun.SQL.Options & { warm?: boolean } = {}) { + const test = crypto.randomUUID(); + const sql = new SQL({ url: url(), max: 1, ...options }); + if (warm) await sql.connect(); + return { + sql, + insert: (x: number, on: SQL = sql) => on`INSERT INTO ${sql(tableName)} (test, x) VALUES (${test}, ${x})`, + insertText: (x: number) => `INSERT INTO ${tableName} (test, x) VALUES ('${test}', ${x})`, + rows: async () => + (await reader`SELECT x FROM ${reader(tableName)} WHERE test = ${test} ORDER BY id`).map(row => row.x), + [Symbol.asyncDispose]: () => sql.close({ timeout: 0 }), + }; + } + + for (const [name, start] of Object.entries(starts)) { + test(`close() waits for a query that ${name} started`, async () => { + await using table = await openTable(); + const inserted = settle(start(table.insert(1)).then(() => "done")); + await table.sql.close(); + expect({ query: await inserted, rows: await table.rows() }).toEqual({ query: { value: "done" }, rows: [1] }); + }); + } + + for (const [name, close] of Object.entries(closes)) { + test(`${name} waits for a query that started before it`, async () => { + await using table = await openTable(); + const inserted = settle(table.insert(1).then(() => "done")); + await close(table.sql); + expect({ query: await inserted, rows: await table.rows() }).toEqual({ query: { value: "done" }, rows: [1] }); + }); + } + + for (const [name, { rows, start }] of Object.entries({ + "sql.unsafe()": { rows: [1], start: ({ sql, insertText }) => sql.unsafe(insertText(1)) }, + "two statements in simple()": { + rows: [1, 2], + start: ({ sql, insertText }) => sql.unsafe(`${insertText(1)}; ${insertText(2)}`).simple(), + }, + })) { + test(`close() waits for ${name}`, async () => { + await using table = await openTable(); + const inserted = settle(start(table).then(() => "done")); + await table.sql.close(); + expect({ query: await inserted, rows: await table.rows() }).toEqual({ query: { value: "done" }, rows }); + }); + } + + test("an `await using` block runs the 20 inserts that it did not await", async () => { + await using table = await openTable(); + const errors: unknown[] = []; + { + await using _ = table.sql; + for (let x = 0; x < 20; x++) table.insert(x).catch(e => errors.push(e.code)); + } + expect({ errors, rows: await table.rows() }).toEqual({ + errors: [], + rows: Array.from({ length: 20 }, (_, x) => x), + }); + }); + + test("queries reach the server in the order of their starts", async () => { + await using table = await openTable(); + const expected = Array.from({ length: 50 }, (_, x) => x); + const names = Object.keys(starts); + const inserted = expected.map(x => settle(starts[names[x % names.length]](table.insert(x)).then(() => x))); + await table.sql.close(); + expect({ queries: await Promise.all(inserted), rows: await table.rows() }).toEqual({ + queries: expected.map(value => ({ value })), + rows: expected, + }); + }); + + test("two close() calls in the same tick wait for a started query", async () => { + await using table = await openTable(); + const inserted = settle(table.insert(1).then(() => "done")); + await Promise.all([table.sql.close(), table.sql.close()]); + expect({ query: await inserted, rows: await table.rows() }).toEqual({ query: { value: "done" }, rows: [1] }); + }); + + test("close() waits for a query that started on a pool with no connection", async () => { + await using table = await openTable({ warm: false }); + const inserted = settle(table.insert(1).then(() => "done")); + await table.sql.close(); + expect({ query: await inserted, rows: await table.rows() }).toEqual({ query: { value: "done" }, rows: [1] }); + }); + + for (const [name, statement] of [ + ["commitDistributed", "COMMIT"], + ["rollbackDistributed", "ROLLBACK"], + ] as const) { + // No such transaction exists, so the error of the server shows that the server got the statement. + test(`close() waits for the ${statement} of sql.${name}() that was called before it`, async () => { + await using table = await openTable(); + const finished = settle(table.sql[name]("close_no_such_transaction")); + await table.sql.close(); + expect(await finished).toEqual({ code: code("SERVER_ERROR") }); + }); + } + + test("close() waits for a started query when a transaction has the only connection", async () => { + await using table = await openTable(); + const order: string[] = []; + const open = Promise.withResolvers(); + const gate = Promise.withResolvers(); + const transaction = table.sql + .begin(async tx => { + await table.insert(1, tx); + open.resolve(); + await gate.promise; + await table.insert(2, tx); + }) + .finally(() => order.push("transaction")); + await open.promise; + const inserted = settle(table.insert(3).then(() => "done")).finally(() => order.push("query")); + const closed = table.sql.close().finally(() => order.push("close()")); + gate.resolve(); + await Promise.all([transaction, closed]); + expect({ query: await inserted, rows: await table.rows(), order }).toEqual({ + query: { value: "done" }, + rows: [1, 2, 3], + order: ["transaction", "query", "close()"], + }); + }); + + test("close() waits for a started query when a reservation has the only connection", async () => { + await using table = await openTable(); + const order: string[] = []; + const reserved = await table.sql.reserve(); + await table.insert(1, reserved); + const inserted = settle(table.insert(3).then(() => "done")).finally(() => order.push("query")); + const closed = table.sql.close().finally(() => order.push("close()")); + await table.insert(2, reserved); + reserved.release(); + await closed; + expect({ query: await inserted, rows: await table.rows(), order }).toEqual({ + query: { value: "done" }, + rows: [1, 2, 3], + order: ["query", "close()"], + }); + }); + + test("a query that starts after close() is rejected and never sent", async () => { + await using table = await openTable(); + const early = settle(table.insert(1).then(() => "done")); + const closed = table.sql.close(); + const late = [settle(table.insert(2).execute()), settle(table.insert(3).then(rows => rows))]; + await closed; + expect({ early: await early, late: await Promise.all(late), rows: await table.rows() }).toEqual({ + early: { value: "done" }, + late: [{ code: code("CONNECTION_CLOSED") }, { code: code("CONNECTION_CLOSED") }], + rows: [1], + }); + }); + + for (const [name, start] of Object.entries(laterStarts)) { + test(`a query that only ${name} started is rejected and never sent`, async () => { + await using table = await openTable(); + const inserted = settle(start(table.insert(1))); + await table.sql.close(); + expect({ query: await inserted, rows: await table.rows() }).toEqual({ + query: { code: code("CONNECTION_CLOSED") }, + rows: [], + }); + }); + } + + // What cancel() does to a query that started is not in these tests. The way that the query started must not + // change it. + for (const warm of [true, false]) { + for (const [name, start] of Object.entries(starts)) { + if (name === "execute()") continue; + const pool = warm ? "a pool with a connection" : "a pool with no connection"; + test(`cancel() in the tick of ${name} does what it does in the tick of execute(), on ${pool}`, async () => { + async function cancelAfter(start: (query: any) => Promise) { + await using table = await openTable({ warm }); + const query = table.insert(1); + const settled = settle(start(query).then(() => "done")); + query.cancel(); + const outcome = await settled; + await table.sql.close(); + return { query: outcome, rows: await table.rows() }; + } + expect(await cancelAfter(start)).toEqual(await cancelAfter(starts["execute()"])); + }); + } + } + }); +} From 6194643e3bc9c1a41a66c3702900ba02c62b6492 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Mon, 28 Sep 2026 23:22:33 +0000 Subject: [PATCH 3/4] ci: retrigger From 30bcbed25d6d00763f6f9ae8327e92829986bc53 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Tue, 29 Sep 2026 00:25:53 +0000 Subject: [PATCH 4/4] test: run the checks again