diff --git a/src/js/bun/sql.ts b/src/js/bun/sql.ts index c3a74582a25..066d4984c1a 100644 --- a/src/js/bun/sql.ts +++ b/src/js/bun/sql.ts @@ -27,6 +27,13 @@ interface TransactionState { queries: Set>; } +function cannotAcceptQueries(state: TransactionState) { + return ( + state.connectionState & ReservedConnectionState.closed || + !(state.connectionState & ReservedConnectionState.acceptQueries) + ); +} + function adapterFromOptions(options: Bun.SQL.__internal.DefinedOptions) { switch (options.adapter) { case "postgres": @@ -259,10 +266,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 (cannotAcceptQueries(state)) { return Promise.$reject(pool.connectionClosedError()); } if ($isArray(strings)) { @@ -278,13 +282,22 @@ const SQL: typeof Bun.SQL = function SQL( } reserved_sql.unsafe = (string, args = []) => { + if (cannotAcceptQueries(state)) { + return Promise.$reject(pool.connectionClosedError()); + } return unsafeQueryFromTransaction(string, args, pooledConnection, state.queries); }; reserved_sql.file = async (path: string, args = []) => { + if (cannotAcceptQueries(state)) { + return Promise.$reject(pool.connectionClosedError()); + } return await Bun.file(path) .text() .then(text => { + if (cannotAcceptQueries(state)) { + return Promise.$reject(pool.connectionClosedError()); + } return unsafeQueryFromTransaction(text, args, pooledConnection, state.queries); }); }; @@ -343,10 +356,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 (cannotAcceptQueries(state)) { return Promise.$reject(pool.connectionClosedError()); } let callback = fn; @@ -380,10 +390,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 (cannotAcceptQueries(state)) { return Promise.$resolve(undefined); } state.connectionState &= ~ReservedConnectionState.acceptQueries; @@ -426,10 +433,7 @@ const SQL: typeof Bun.SQL = function SQL( return Promise.$resolve(undefined); }; reserved_sql.release = () => { - if ( - state.connectionState & ReservedConnectionState.closed || - !(state.connectionState & ReservedConnectionState.acceptQueries) - ) { + if (cannotAcceptQueries(state)) { return Promise.$reject(pool.connectionClosedError()); } // just release the connection back to the pool @@ -561,10 +565,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 (cannotAcceptQueries(state)) { return Promise.$reject(pool.connectionClosedError()); } if ($isArray(strings)) { @@ -579,12 +580,21 @@ const SQL: typeof Bun.SQL = function SQL( return queryFromTransaction(strings, values, pooledConnection, state.queries); } transaction_sql.unsafe = (string, args = []) => { + if (cannotAcceptQueries(state)) { + return Promise.$reject(pool.connectionClosedError()); + } return unsafeQueryFromTransaction(string, args, pooledConnection, state.queries); }; transaction_sql.file = async (path: string, args = []) => { + if (cannotAcceptQueries(state)) { + return Promise.$reject(pool.connectionClosedError()); + } return await Bun.file(path) .text() .then(text => { + if (cannotAcceptQueries(state)) { + return Promise.$reject(pool.connectionClosedError()); + } return unsafeQueryFromTransaction(text, args, pooledConnection, state.queries); }); }; @@ -645,10 +655,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 (cannotAcceptQueries(state)) { return Promise.$resolve(undefined); } state.connectionState &= ~ReservedConnectionState.acceptQueries; @@ -730,10 +737,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 (cannotAcceptQueries(state)) { throw pool.connectionClosedError(); } diff --git a/test/js/sql/sql-transaction-closed-handle.test.ts b/test/js/sql/sql-transaction-closed-handle.test.ts new file mode 100644 index 00000000000..a39da698f66 --- /dev/null +++ b/test/js/sql/sql-transaction-closed-handle.test.ts @@ -0,0 +1,42 @@ +import { SQL } from "bun"; +import { expect, test } from "bun:test"; +import { tempDirWithFiles } from "harness"; +import { join } from "node:path"; + +// tx.unsafe()/tx.file() must reject once sql.begin() has settled, the same way the +// tagged-template call and tx.savepoint() already do. Before the fix they reached the +// pooled connection the handle no longer owns and the write was durable. +test("tx.unsafe() and tx.file() reject after begin() settles", async () => { + const dir = tempDirWithFiles("sql-tx-closed", { + "q.sql": `INSERT INTO accounts VALUES (98, 0)`, + }); + const sql = new SQL("sqlite://:memory:"); + try { + await sql`CREATE TABLE accounts (id INTEGER PRIMARY KEY, balance REAL)`; + let leaked: any; + await sql.begin(async tx => { + leaked = tx; + await tx.unsafe(`INSERT INTO accounts VALUES (10, 0)`); + }); + const outcome = async (p: Promise) => + p.then( + () => "fulfilled", + (e: any) => `rejected: ${e.message}`, + ); + expect({ + template: await outcome(leaked`SELECT 1`), + savepoint: await outcome(Promise.resolve().then(() => leaked.savepoint(async () => {}))), + unsafe: await outcome(leaked.unsafe(`INSERT INTO accounts VALUES (99, 0)`)), + file: await outcome(leaked.file(join(dir, "q.sql"))), + }).toEqual({ + template: "rejected: Connection closed", + savepoint: "rejected: Connection closed", + unsafe: "rejected: Connection closed", + file: "rejected: Connection closed", + }); + const rows = await sql`SELECT id FROM accounts ORDER BY id`; + expect(rows).toEqual([{ id: 10 }]); + } finally { + await sql.close(); + } +}); diff --git a/test/js/sql/sql.test.ts b/test/js/sql/sql.test.ts index 7818fecb5ca..5e802b0f76e 100644 --- a/test/js/sql/sql.test.ts +++ b/test/js/sql/sql.test.ts @@ -3997,6 +3997,49 @@ CREATE TABLE ${table_name} ( expect(xs.map(x => x.x).join("")).toBe("123"); }); + test("reserved.unsafe() and reserved.file() reject after release()", async () => { + await using sql = postgres({ ...options, max: 1 }); + const reserved = await sql.reserve(); + await reserved.release(); + const outcome = async (p: Promise) => + p.then( + () => "fulfilled", + (e: any) => `rejected: ${e.code}`, + ); + expect({ + template: await outcome(reserved`select 1 as x`), + unsafe: await outcome(reserved.unsafe("select 1 as x")), + file: await outcome(reserved.file(rel("select.sql"))), + }).toEqual({ + template: "rejected: ERR_POSTGRES_CONNECTION_CLOSED", + unsafe: "rejected: ERR_POSTGRES_CONNECTION_CLOSED", + file: "rejected: ERR_POSTGRES_CONNECTION_CLOSED", + }); + }); + + test("tx.unsafe() and tx.file() reject after begin() settles", async () => { + await using sql = postgres({ ...options, max: 1 }); + let leaked: any; + await sql.begin(async tx => { + leaked = tx; + await tx.unsafe("select 1 as x"); + }); + const outcome = async (p: Promise) => + p.then( + () => "fulfilled", + (e: any) => `rejected: ${e.code}`, + ); + expect({ + template: await outcome(leaked`select 1 as x`), + unsafe: await outcome(leaked.unsafe("select 1 as x")), + file: await outcome(leaked.file(rel("select.sql"))), + }).toEqual({ + template: "rejected: ERR_POSTGRES_CONNECTION_CLOSED", + unsafe: "rejected: ERR_POSTGRES_CONNECTION_CLOSED", + file: "rejected: ERR_POSTGRES_CONNECTION_CLOSED", + }); + }); + test("keeps process alive when it should", async () => { const file = path.posix.join(__dirname, "sql-fixture-ref.ts"); const result = await $`DATABASE_URL=${process.env.DATABASE_URL} ${bunExe()} ${file}`;