diff --git a/src/js/bun/sql.ts b/src/js/bun/sql.ts index c26fa18a01d7..06a64d5257d0 100644 --- a/src/js/bun/sql.ts +++ b/src/js/bun/sql.ts @@ -31,6 +31,13 @@ interface TransactionState { queries: Set>; } +function acceptsQueries(state: TransactionState) { + return ( + (state.connectionState & ReservedConnectionState.closed) === 0 && + (state.connectionState & ReservedConnectionState.acceptQueries) !== 0 + ); +} + /// Bound as `this` to both callbacks of a reserve({ signal }) call, so each can /// reach the other without a per-call closure. interface ReserveAbortState { @@ -220,6 +227,17 @@ const SQL: typeof Bun.SQL = function SQL( } } + function unsafeTransactionQueryFlags(values: any[]) { + let flags = connectionInfo.bigint + ? SQLQueryFlags.allowUnsafeTransaction | SQLQueryFlags.unsafe | SQLQueryFlags.bigint + : SQLQueryFlags.allowUnsafeTransaction | SQLQueryFlags.unsafe; + + if ((values?.length ?? 0) === 0) { + flags |= SQLQueryFlags.simple; + } + return flags; + } + function unsafeQueryFromTransaction( strings: string | TemplateStringsArray | import("internal/sql/shared.ts").SQLHelper | Query, values: any[], @@ -227,17 +245,10 @@ const SQL: typeof Bun.SQL = function SQL( transactionQueries: Set>, ) { try { - let flags = connectionInfo.bigint - ? SQLQueryFlags.allowUnsafeTransaction | SQLQueryFlags.unsafe | SQLQueryFlags.bigint - : SQLQueryFlags.allowUnsafeTransaction | SQLQueryFlags.unsafe; - - if ((values?.length ?? 0) === 0) { - flags |= SQLQueryFlags.simple; - } const query = new Query( strings, values, - flags, + unsafeTransactionQueryFlags(values), queryFromTransactionHandler.bind(pooledConnection, transactionQueries), pool, ); @@ -248,6 +259,35 @@ const SQL: typeof Bun.SQL = function SQL( } } + function rejectConnectionClosed(query: Query) { + query.reject(pool.connectionClosedError()); + } + + // Not for COMMIT and ROLLBACK: close() stops accepting queries and then still has to send ROLLBACK. + function unsafeQueryFromHandle( + state: TransactionState, + pooledConnection: PooledPostgresConnection, + strings: string, + values: any[], + ) { + if (acceptsQueries(state)) { + return unsafeQueryFromTransaction(strings, values, pooledConnection, state.queries); + } + // Still a lazy Query, so .values() and use as a fragment work. It rejects when it runs. + return new Query(strings, values, unsafeTransactionQueryFlags(values), rejectConnectionClosed, pool); + } + + async function fileQueryFromHandle( + state: TransactionState, + pooledConnection: PooledPostgresConnection, + path: string, + values: any[], + ) { + const text = await Bun.file(path).text(); + // The handle can close while the file is read, so its state is read here and not before. + return unsafeQueryFromHandle(state, pooledConnection, text, values); + } + function onTransactionDisconnected(this: TransactionState, err: Error) { const reject = this.reject; this.connectionState |= ReservedConnectionState.closed; @@ -355,10 +395,7 @@ const SQL: typeof Bun.SQL = function SQL( } function reserved_sql(strings: string | TemplateStringsArray | SQLHelper | Query, ...values: any[]) { - if ( - state.connectionState & ReservedConnectionState.closed || - !(state.connectionState & ReservedConnectionState.acceptQueries) - ) { + if (!acceptsQueries(state)) { return Promise.$reject(pool.connectionClosedError()); } if ($isArray(strings)) { @@ -373,17 +410,8 @@ const SQL: typeof Bun.SQL = function SQL( return queryFromTransaction(strings, values, pooledConnection, state.queries); } - reserved_sql.unsafe = (string, args = []) => { - return unsafeQueryFromTransaction(string, args, pooledConnection, state.queries); - }; - - reserved_sql.file = async (path: string, args = []) => { - return await Bun.file(path) - .text() - .then(text => { - return unsafeQueryFromTransaction(text, args, pooledConnection, state.queries); - }); - }; + reserved_sql.unsafe = (string, args = []) => unsafeQueryFromHandle(state, pooledConnection, string, args); + reserved_sql.file = (path: string, args = []) => fileQueryFromHandle(state, pooledConnection, path, args); reserved_sql.connect = () => { if (state.connectionState & ReservedConnectionState.closed) { @@ -433,10 +461,7 @@ const SQL: typeof Bun.SQL = function SQL( }; reserved_sql.begin = (options_or_fn: string | TransactionCallback, fn?: TransactionCallback) => { // begin is allowed the difference is that we need to make sure to use the same connection and never release it - if ( - state.connectionState & ReservedConnectionState.closed || - !(state.connectionState & ReservedConnectionState.acceptQueries) - ) { + if (!acceptsQueries(state)) { return Promise.$reject(pool.connectionClosedError()); } let callback = fn; @@ -465,10 +490,7 @@ const SQL: typeof Bun.SQL = function SQL( }; reserved_sql.close = async (options?: { timeout?: number }) => { const reserveQueries = state.queries; - if ( - state.connectionState & ReservedConnectionState.closed || - !(state.connectionState & ReservedConnectionState.acceptQueries) - ) { + if (!acceptsQueries(state)) { return Promise.$resolve(undefined); } state.connectionState &= ~ReservedConnectionState.acceptQueries; @@ -662,10 +684,7 @@ const SQL: typeof Bun.SQL = function SQL( strings: string | TemplateStringsArray | import("internal/sql/shared.ts").SQLHelper | Query, ...values: any[] ) { - if ( - state.connectionState & ReservedConnectionState.closed || - !(state.connectionState & ReservedConnectionState.acceptQueries) - ) { + if (!acceptsQueries(state)) { return Promise.$reject(pool.connectionClosedError()); } if ($isArray(strings)) { @@ -679,16 +698,8 @@ const SQL: typeof Bun.SQL = function SQL( return queryFromTransaction(strings, values, pooledConnection, state.queries); } - transaction_sql.unsafe = (string, args = []) => { - return unsafeQueryFromTransaction(string, args, pooledConnection, state.queries); - }; - transaction_sql.file = async (path: string, args = []) => { - return await Bun.file(path) - .text() - .then(text => { - return unsafeQueryFromTransaction(text, args, pooledConnection, state.queries); - }); - }; + transaction_sql.unsafe = (string, args = []) => unsafeQueryFromHandle(state, pooledConnection, string, args); + transaction_sql.file = (path: string, args = []) => fileQueryFromHandle(state, pooledConnection, path, args); // reserve is allowed to be called inside transaction connection but will return a new reserved connection from the pool and will not be part of the transaction // this matchs the behavior of the postgres package transaction_sql.reserve = (options?: { signal?: AbortSignal }) => sql.reserve(options); @@ -748,10 +759,7 @@ const SQL: typeof Bun.SQL = function SQL( }; transaction_sql.close = async function (options?: { timeout?: number }) { // we dont actually close the connection here, we just set the state to closed and rollback the transaction - if ( - state.connectionState & ReservedConnectionState.closed || - !(state.connectionState & ReservedConnectionState.acceptQueries) - ) { + if (!acceptsQueries(state)) { return Promise.$resolve(undefined); } state.connectionState &= ~ReservedConnectionState.acceptQueries; @@ -833,10 +841,7 @@ const SQL: typeof Bun.SQL = function SQL( transaction_sql.savepoint = async (fn: TransactionCallback, name?: string): Promise => { let savepoint_callback = fn; - if ( - state.connectionState & ReservedConnectionState.closed || - !(state.connectionState & ReservedConnectionState.acceptQueries) - ) { + if (!acceptsQueries(state)) { throw pool.connectionClosedError(); } diff --git a/test/js/sql/postgres-listen-notify.test.ts b/test/js/sql/postgres-listen-notify.test.ts index 731ace6d4957..642f1e12a8e8 100644 --- a/test/js/sql/postgres-listen-notify.test.ts +++ b/test/js/sql/postgres-listen-notify.test.ts @@ -769,6 +769,34 @@ describe("notify()", () => { expect((await settled).map(result => result.status)).toEqual(["rejected", "rejected"]); expect(server.liveConnections).toBe(1); }); + + // notify() on a handle sends through that handle's unsafe(). + test("rejects on a released reserved handle and on a settled transaction handle, and sends nothing", async () => { + await using server = await mockServer(); + await using sql = client(server.url); + const reserved = await sql.reserve(); + await reserved.release(); + let settledTransaction!: Bun.TransactionSQL; + await sql.begin(async tx => { + settledTransaction = tx; + }); + + const rejectionCode = (promise: Promise) => + promise.then( + () => "resolved", + err => err.code ?? err.message, + ); + expect({ + reserved: await rejectionCode(reserved.notify("ch", "payload")), + transaction: await rejectionCode(settledTransaction.notify("ch", "payload")), + }).toEqual({ + reserved: "ERR_POSTGRES_CONNECTION_CLOSED", + transaction: "ERR_POSTGRES_CONNECTION_CLOSED", + }); + // The pool has one connection, so a notify that was sent would arrive before this query. + await sql.unsafe("SELECT 'barrier'"); + expect(server.queries).toEqual(["BEGIN", "COMMIT", "SELECT 'barrier'"]); + }); }); describe("in a subprocess", () => { diff --git a/test/js/sql/sql-pool-transaction-isolation.test.ts b/test/js/sql/sql-pool-transaction-isolation.test.ts index 6e20d7a26293..5c71e48bcc04 100644 --- a/test/js/sql/sql-pool-transaction-isolation.test.ts +++ b/test/js/sql/sql-pool-transaction-isolation.test.ts @@ -5,8 +5,9 @@ // Wire bytes come from ./wire-frames.ts. import { SQL } from "bun"; import { describe, expect, test } from "bun:test"; -import { bunEnv, bunExe } from "harness"; +import { bunEnv, bunExe, tempDir } from "harness"; import type net from "node:net"; +import { join } from "node:path"; import { listeningServer, mysqlAckSessionSetup, @@ -114,6 +115,12 @@ const adapters: Array<{ adapter: "postgres" | "mysql"; mockServer: MockServer; b { adapter: "mysql", mockServer: mysqlMockServer, beginCommand: "START TRANSACTION" }, ]; +const rejectionCode = (promise: Promise) => + promise.then( + () => "resolved", + err => err.code ?? err.message, + ); + // reserved.begin() / beginDistributed() calls that reject before anything is sent. const rejectedBeforeBegin = [ { @@ -129,6 +136,7 @@ const rejectedBeforeBegin = [ ]; describe.each(adapters)("$adapter", ({ adapter, mockServer, beginCommand }) => { + const closedCode = `ERR_${adapter.toUpperCase()}_CONNECTION_CLOSED`; const options = (port: number): Bun.SQL.Options => ({ adapter, hostname: "127.0.0.1", @@ -439,4 +447,183 @@ describe.each(adapters)("$adapter", ({ adapter, mockServer, beginCommand }) => { await new Promise(r => server.close(() => r())); } }); + + // With max: 1 the next transaction holds the same connection, so a statement from the + // stale handle would land inside it. + test("unsafe() and file() on a transaction handle kept after begin() settles reject and send nothing", async () => { + using dir = tempDir("sql-stale-transaction", { "stale.sql": "SELECT 'stale file'" }); + const received: Received[] = []; + const { port, server } = await mockServer(received); + const sql = new SQL(options(port)); + try { + let stale!: Bun.TransactionSQL; + await sql.begin(async tx => { + stale = tx; + await tx.unsafe("SELECT 'T1a'"); + }); + + const outcomes = await sql.begin(async tx => { + await tx.unsafe("SELECT 'T2a'"); + return { + tagged: await rejectionCode(stale`SELECT 'stale tagged'`), + unsafe: await rejectionCode(stale.unsafe("SELECT 'stale unsafe'")), + unsafeValues: await rejectionCode(stale.unsafe("SELECT 'stale values'").values()), + file: await rejectionCode(stale.file(join(String(dir), "stale.sql"))), + }; + }); + expect(outcomes).toEqual({ tagged: closedCode, unsafe: closedCode, unsafeValues: closedCode, file: closedCode }); + expect(received).toEqual([ + { conn: 0, sql: beginCommand }, + { conn: 0, sql: "SELECT 'T1a'" }, + { conn: 0, sql: "COMMIT" }, + { conn: 0, sql: beginCommand }, + { conn: 0, sql: "SELECT 'T2a'" }, + { conn: 0, sql: "COMMIT" }, + ]); + } finally { + await sql.close({ timeout: 0 }).catch(() => {}); + await new Promise(r => server.close(() => r())); + } + }); + + test("unsafe() and file() on a reserved handle reject and send nothing after release()", async () => { + using dir = tempDir("sql-stale-reserved", { "stale.sql": "SELECT 'stale file'" }); + const received: Received[] = []; + const { port, server } = await mockServer(received); + const sql = new SQL(options(port)); + try { + const reserved = await sql.reserve(); + await reserved.unsafe("SELECT 'R1'"); + await reserved.release(); + + const outcomes = await sql.begin(async tx => { + await tx.unsafe("SELECT 'T1a'"); + return { + tagged: await rejectionCode(reserved`SELECT 'stale tagged'`), + unsafe: await rejectionCode(reserved.unsafe("SELECT 'stale unsafe'")), + file: await rejectionCode(reserved.file(join(String(dir), "stale.sql"))), + // Both send their statement through reserved.unsafe(). + commitDistributed: await rejectionCode(reserved.commitDistributed("stale")), + rollbackDistributed: await rejectionCode(reserved.rollbackDistributed("stale")), + }; + }); + expect(outcomes).toEqual({ + tagged: closedCode, + unsafe: closedCode, + file: closedCode, + commitDistributed: closedCode, + rollbackDistributed: closedCode, + }); + expect(received).toEqual([ + { conn: 0, sql: "SELECT 'R1'" }, + { conn: 0, sql: beginCommand }, + { conn: 0, sql: "SELECT 'T1a'" }, + { conn: 0, sql: "COMMIT" }, + ]); + } finally { + await sql.close({ timeout: 0 }).catch(() => {}); + await new Promise(r => server.close(() => r())); + } + }); + + test("reserved.file() does not run when release() lands while the file is read", async () => { + using dir = tempDir("sql-late-reserved-file", { "late.sql": "SELECT 'late file'" }); + const received: Received[] = []; + const { port, server } = await mockServer(received); + const sql = new SQL(options(port)); + try { + const reserved = await sql.reserve(); + // release() changes the handle's state synchronously, and the read needs an event loop turn. + const late = rejectionCode(reserved.file(join(String(dir), "late.sql"))); + reserved.release(); + expect(await late).toBe(closedCode); + // The pool has one connection, so a late statement would arrive before this one. + await sql.unsafe("SELECT 'barrier'"); + expect(received).toEqual([{ conn: 0, sql: "SELECT 'barrier'" }]); + } finally { + await sql.close({ timeout: 0 }).catch(() => {}); + await new Promise(r => server.close(() => r())); + } + }); + + // close({ timeout }) stops accepting queries at once and keeps the connection until the + // pending query settles. The timeout is in seconds. + test("unsafe() and file() reject while reserved.close({ timeout }) waits for a pending query", async () => { + using dir = tempDir("sql-closing-reserved", { "late.sql": "SELECT 'late file'" }); + const received: Received[] = []; + const { port, server } = await mockServer(received); + const sql = new SQL(options(port)); + try { + const reserved = await sql.reserve(); + const pending = reserved.unsafe("SELECT 'R1'").execute(); + const closed = reserved.close({ timeout: 60 }); + const tagged = rejectionCode(reserved`SELECT 'late tagged'`); + const unsafe = rejectionCode(reserved.unsafe("SELECT 'late unsafe'")); + const file = rejectionCode(reserved.file(join(String(dir), "late.sql"))); + await pending; + await closed; + const outcomes = { tagged: await tagged, unsafe: await unsafe, file: await file }; + reserved.release(); + await sql.unsafe("SELECT 'barrier'"); + expect(outcomes).toEqual({ tagged: closedCode, unsafe: closedCode, file: closedCode }); + // Which connection carries the barrier depends on what close() does with the reserved one after the wait. + expect(received.map(statement => statement.sql)).toEqual(["SELECT 'R1'", "SELECT 'barrier'"]); + } finally { + await sql.close({ timeout: 0 }).catch(() => {}); + await new Promise(r => server.close(() => r())); + } + }); + + test("unsafe() and file() reject with CONNECTION_CLOSED after the connection drops mid-transaction", async () => { + using dir = tempDir("sql-dropped-transaction", { "after.sql": "SELECT 'after file'" }); + const received: Received[] = []; + const { port, server } = await mockServer(received); + const sql = new SQL(options(port)); + try { + // begin() rejects as soon as the connection drops, before the callback is done. + let callback!: Promise>; + const begin = rejectionCode( + sql.begin(tx => { + callback = (async () => { + await tx.unsafe("SELECT 'KILL'").catch(() => {}); + return { + tagged: await rejectionCode(tx`SELECT 'after tagged'`), + unsafe: await rejectionCode(tx.unsafe("SELECT 'after unsafe'")), + file: await rejectionCode(tx.file(join(String(dir), "after.sql"))), + }; + })(); + return callback; + }), + ); + expect(await begin).toBe(closedCode); + expect(await callback).toEqual({ tagged: closedCode, unsafe: closedCode, file: closedCode }); + expect(received).toEqual([ + { conn: 0, sql: beginCommand }, + { conn: 0, sql: "SELECT 'KILL'" }, + ]); + } finally { + await sql.close({ timeout: 0 }).catch(() => {}); + await new Promise(r => server.close(() => r())); + } + }); + + test("unsafe() and file() reject with CONNECTION_CLOSED after a reserved connection drops", async () => { + using dir = tempDir("sql-dropped-reserved", { "after.sql": "SELECT 'after file'" }); + const received: Received[] = []; + const { port, server } = await mockServer(received); + const sql = new SQL(options(port)); + try { + const reserved = await sql.reserve(); + await reserved.unsafe("SELECT 'KILL'").catch(() => {}); + expect({ + tagged: await rejectionCode(reserved`SELECT 'after tagged'`), + unsafe: await rejectionCode(reserved.unsafe("SELECT 'after unsafe'")), + file: await rejectionCode(reserved.file(join(String(dir), "after.sql"))), + }).toEqual({ tagged: closedCode, unsafe: closedCode, file: closedCode }); + expect(received).toEqual([{ conn: 0, sql: "SELECT 'KILL'" }]); + } finally { + await sql.close({ timeout: 0 }).catch(() => {}); + await new Promise(r => server.close(() => r())); + } + }); }); diff --git a/test/js/sql/sqlite-sql.test.ts b/test/js/sql/sqlite-sql.test.ts index ed0567c186fb..d73b351e3bc5 100644 --- a/test/js/sql/sqlite-sql.test.ts +++ b/test/js/sql/sqlite-sql.test.ts @@ -1254,6 +1254,69 @@ describe("Transactions", () => { const accounts = await sql`SELECT * FROM accounts WHERE id = 1`; expect(accounts[0].balance).toBe(1002); }); + + const rejectionCode = (promise: Promise) => + promise.then( + () => "resolved", + err => err.code ?? err.message, + ); + + // A statement from a settled handle runs outside the transaction: its row stays even after a rollback. + test.each([ + { settles: "commits", fails: false, ids: [1, 2, 10] }, + { settles: "rolls back", fails: true, ids: [1, 2] }, + ])("a transaction handle rejects unsafe() and file() after begin() $settles", async ({ fails, ids }) => { + using dir = tempDir("sqlite-sql-stale-tx", { "insert.sql": "INSERT INTO accounts VALUES (98, 0)" }); + let stale!: Bun.TransactionSQL; + const settled = await sql + .begin(async tx => { + stale = tx; + await tx`INSERT INTO accounts VALUES (10, 0)`; + if (fails) throw new Error("roll back"); + return "committed"; + }) + .catch(err => err.message); + expect(settled).toBe(fails ? "roll back" : "committed"); + + expect({ + tagged: await rejectionCode(stale`INSERT INTO accounts VALUES (97, 0)`), + unsafe: await rejectionCode(stale.unsafe("INSERT INTO accounts VALUES (99, 0)")), + file: await rejectionCode(stale.file(join(String(dir), "insert.sql"))), + }).toEqual({ + tagged: "ERR_SQLITE_CONNECTION_CLOSED", + unsafe: "ERR_SQLITE_CONNECTION_CLOSED", + file: "ERR_SQLITE_CONNECTION_CLOSED", + }); + expect((await sql`SELECT id FROM accounts ORDER BY id`).map(row => row.id)).toEqual(ids); + }); + + // bun:test also fails this test if the query that nothing awaits reports an unhandled rejection. + test("unsafe() on a settled transaction handle returns a lazy Query that rejects when it runs", async () => { + let stale!: Bun.TransactionSQL; + await sql.begin(async tx => { + stale = tx; + }); + + stale.unsafe("INSERT INTO accounts VALUES (99, 0)"); + expect(await sql`SELECT ${stale.unsafe("balance")} FROM accounts WHERE id = 1`).toEqual([{ balance: 1000 }]); + expect(await rejectionCode(stale.unsafe("INSERT INTO accounts VALUES (98, 0)").values())).toBe( + "ERR_SQLITE_CONNECTION_CLOSED", + ); + expect((await sql`SELECT id FROM accounts ORDER BY id`).map(row => row.id)).toEqual([1, 2]); + }); + + test("tx.file() does not run when the transaction settles while the file is read", async () => { + using dir = tempDir("sqlite-sql-late-file", { "insert.sql": "INSERT INTO accounts VALUES (99, 0)" }); + let late!: Promise; + await sql.begin(async tx => { + await tx`INSERT INTO accounts VALUES (10, 0)`; + // Not awaited. SQLite answers COMMIT in microtasks and the read needs an + // event loop turn, so the transaction settles first. + late = rejectionCode(tx.file(join(String(dir), "insert.sql"))); + }); + expect(await late).toBe("ERR_SQLITE_CONNECTION_CLOSED"); + expect((await sql`SELECT id FROM accounts ORDER BY id`).map(row => row.id)).toEqual([1, 2, 10]); + }); }); describe("SQLite-specific features", () => {