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-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()"])); + }); + } + } + }); +} 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:");