diff --git a/docs/runtime/sql.mdx b/docs/runtime/sql.mdx index e14b2b9b54b0..77a7c0254ce0 100644 --- a/docs/runtime/sql.mdx +++ b/docs/runtime/sql.mdx @@ -1002,6 +1002,8 @@ try { } // Automatically released ``` +Await each query before you release the connection. A query is lazy and runs when it is awaited. A query that first runs after `release()` rejects with `ERR_POSTGRES_CONNECTION_CLOSED` (or the MySQL equivalent), because the connection can already belong to another caller. `return reserved\`SELECT 1\``without`await` from a block that releases the handle is such a query. + When every connection is busy, `reserve()` waits until one frees up. Pass an `AbortSignal` to stop waiting: the promise rejects with `signal.reason` and no connection is taken from the pool. Aborting after the promise resolved has no effect, so a connection you received must still be released. ```ts diff --git a/src/js/bun/sql.ts b/src/js/bun/sql.ts index 06a64d5257d0..24594ea3c0e7 100644 --- a/src/js/bun/sql.ts +++ b/src/js/bun/sql.ts @@ -176,8 +176,9 @@ const SQL: typeof Bun.SQL = function SQL( transactionQueries.delete(query); } - function queryFromTransactionHandler(transactionQueries, query, handle, err) { + function queryFromTransactionHandler(state: TransactionState, query, handle, err) { const pooledConnection = this; + const transactionQueries = state.queries; if (err) { transactionQueries.delete(query); return query.reject(err); @@ -189,6 +190,14 @@ const SQL: typeof Bun.SQL = function SQL( return query.reject(pool.queryCancelledError()); } + // A query is lazy: it runs when it is first awaited. The handle can be + // committed, rolled back or released by then, and the connection can + // already belong to another caller. + if (state.connectionState & ReservedConnectionState.closed) { + transactionQueries.delete(query); + return query.reject(pool.connectionClosedError()); + } + query.finally(onTransactionQueryDisconnected.bind(transactionQueries, query)); try { @@ -207,7 +216,7 @@ const SQL: typeof Bun.SQL = function SQL( strings: string | TemplateStringsArray | import("internal/sql/shared.ts").SQLHelper | Query, values: any[], pooledConnection: PooledPostgresConnection, - transactionQueries: Set>, + state: TransactionState, ) { try { const query = new Query( @@ -216,11 +225,11 @@ const SQL: typeof Bun.SQL = function SQL( connectionInfo.bigint ? SQLQueryFlags.allowUnsafeTransaction | SQLQueryFlags.bigint : SQLQueryFlags.allowUnsafeTransaction, - queryFromTransactionHandler.bind(pooledConnection, transactionQueries), + queryFromTransactionHandler.bind(pooledConnection, state), pool, ); - transactionQueries.add(query); + state.queries.add(query); return query; } catch (err) { return Promise.$reject(err); @@ -242,17 +251,17 @@ const SQL: typeof Bun.SQL = function SQL( strings: string | TemplateStringsArray | import("internal/sql/shared.ts").SQLHelper | Query, values: any[], pooledConnection: PooledPostgresConnection, - transactionQueries: Set>, + state: TransactionState, ) { try { const query = new Query( strings, values, unsafeTransactionQueryFlags(values), - queryFromTransactionHandler.bind(pooledConnection, transactionQueries), + queryFromTransactionHandler.bind(pooledConnection, state), pool, ); - transactionQueries.add(query); + state.queries.add(query); return query; } catch (err) { return Promise.$reject(err); @@ -271,7 +280,7 @@ const SQL: typeof Bun.SQL = function SQL( values: any[], ) { if (acceptsQueries(state)) { - return unsafeQueryFromTransaction(strings, values, pooledConnection, state.queries); + return unsafeQueryFromTransaction(strings, values, pooledConnection, state); } // 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); @@ -407,7 +416,7 @@ const SQL: typeof Bun.SQL = function SQL( return new SQLHelper([strings], values); } // we use the same code path as the transaction sql - return queryFromTransaction(strings, values, pooledConnection, state.queries); + return queryFromTransaction(strings, values, pooledConnection, state); } reserved_sql.unsafe = (string, args = []) => unsafeQueryFromHandle(state, pooledConnection, string, args); @@ -678,7 +687,7 @@ const SQL: typeof Bun.SQL = function SQL( if (state.connectionState & ReservedConnectionState.closed) { return Promise.$reject(pool.connectionClosedError()); } - return unsafeQueryFromTransaction(string, [], pooledConnection, state.queries); + return unsafeQueryFromTransaction(string, [], pooledConnection, state); } function transaction_sql( strings: string | TemplateStringsArray | import("internal/sql/shared.ts").SQLHelper | Query, @@ -696,7 +705,7 @@ const SQL: typeof Bun.SQL = function SQL( return new SQLHelper([strings], values); } - return queryFromTransaction(strings, values, pooledConnection, state.queries); + return queryFromTransaction(strings, values, pooledConnection, state); } transaction_sql.unsafe = (string, args = []) => unsafeQueryFromHandle(state, pooledConnection, string, args); transaction_sql.file = (path: string, args = []) => fileQueryFromHandle(state, pooledConnection, path, args); diff --git a/test/js/sql/sql-pool-transaction-isolation.test.ts b/test/js/sql/sql-pool-transaction-isolation.test.ts index 5c71e48bcc04..00d343c203e4 100644 --- a/test/js/sql/sql-pool-transaction-isolation.test.ts +++ b/test/js/sql/sql-pool-transaction-isolation.test.ts @@ -400,6 +400,144 @@ describe.each(adapters)("$adapter", ({ adapter, mockServer, beginCommand }) => { } }); + // A query is lazy: it runs when it is first awaited. One created on a transaction or + // a reservation and first awaited after that handle settled must not run, because + // the pool's only connection can already be inside somebody else's transaction. + // unsafe() without parameters, because the mock servers answer simple queries only. + test("a query first awaited after its transaction settled rejects instead of running on the next transaction", async () => { + const received: Received[] = []; + const { port, server } = await mockServer(received); + const sql = new SQL(options(port)); + const closedCode = adapter === "postgres" ? "ERR_POSTGRES_CONNECTION_CLOSED" : "ERR_MYSQL_CONNECTION_CLOSED"; + try { + let afterRollback!: Promise; + let afterCommit!: Promise; + const err = await sql + .begin(async tx => { + afterRollback = tx.unsafe("SELECT 'lazy from T1'"); + throw new Error("t1-app-error"); + }) + .catch(e => e); + expect(err.message).toBe("t1-app-error"); + await sql.begin(async tx => { + await tx.unsafe("SELECT 'T2a'"); + afterCommit = tx.unsafe("SELECT 'lazy from T2'"); + }); + + const t3 = sql.begin(async tx => { + await tx.unsafe("SELECT 'T3a'"); + expect( + await afterRollback.then( + () => null, + e => e.code, + ), + ).toBe(closedCode); + expect( + await afterCommit.then( + () => null, + e => e.code, + ), + ).toBe(closedCode); + await tx.unsafe("SELECT 'T3b'"); + return "t3"; + }); + expect(await t3).toBe("t3"); + + expect(received).toEqual([ + { conn: 0, sql: beginCommand }, + { conn: 0, sql: "ROLLBACK" }, + { conn: 0, sql: beginCommand }, + { conn: 0, sql: "SELECT 'T2a'" }, + { conn: 0, sql: "COMMIT" }, + { conn: 0, sql: beginCommand }, + { conn: 0, sql: "SELECT 'T3a'" }, + { conn: 0, sql: "SELECT 'T3b'" }, + { conn: 0, sql: "COMMIT" }, + ]); + } finally { + await sql.close({ timeout: 0 }).catch(() => {}); + await new Promise(r => server.close(() => r())); + } + }); + + test("a query first awaited after its reservation was released rejects instead of running on the next transaction", async () => { + const received: Received[] = []; + const { port, server } = await mockServer(received); + const sql = new SQL(options(port)); + const closedCode = adapter === "postgres" ? "ERR_POSTGRES_CONNECTION_CLOSED" : "ERR_MYSQL_CONNECTION_CLOSED"; + try { + const reserved = await sql.reserve(); + await reserved.unsafe("SELECT 'R1'"); + const afterRelease = reserved.unsafe("SELECT 'lazy from R'"); + reserved.release(); + + const t1 = await sql.begin(async tx => { + await tx.unsafe("SELECT 'T1a'"); + expect( + await afterRelease.then( + () => null, + e => e.code, + ), + ).toBe(closedCode); + await tx.unsafe("SELECT 'T1b'"); + return "t1"; + }); + expect(t1).toBe("t1"); + + expect(received).toEqual([ + { conn: 0, sql: "SELECT 'R1'" }, + { conn: 0, sql: beginCommand }, + { conn: 0, sql: "SELECT 'T1a'" }, + { conn: 0, sql: "SELECT 'T1b'" }, + { conn: 0, sql: "COMMIT" }, + ]); + } finally { + await sql.close({ timeout: 0 }).catch(() => {}); + await new Promise(r => server.close(() => r())); + } + }); + + // Behaviour change: a query returned without await from a block that releases the + // handle first runs after release(). It used to run on the released connection. + test.each([ + [ + "a using block", + async (sql: Bun.SQL) => { + using reserved = await sql.reserve(); + return reserved.unsafe("SELECT 'returned without await'"); + }, + ], + [ + "a finally block that calls release()", + async (sql: Bun.SQL) => { + const reserved = await sql.reserve(); + try { + return reserved.unsafe("SELECT 'returned without await'"); + } finally { + reserved.release(); + } + }, + ], + ])("a query returned without await from %s rejects and sends nothing", async (_name, run) => { + const received: Received[] = []; + const { port, server } = await mockServer(received); + const sql = new SQL(options(port)); + const closedCode = adapter === "postgres" ? "ERR_POSTGRES_CONNECTION_CLOSED" : "ERR_MYSQL_CONNECTION_CLOSED"; + try { + expect( + await run(sql).then( + () => null, + e => e.code, + ), + ).toBe(closedCode); + 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())); + } + }); + // Runs in a child process: bun:test would turn any unhandled rejection into a test // failure, and the second half of this contract is that one rejection IS reported. test("a rejected reserved begin() is reported as unhandled only when the caller ignores it", async () => { diff --git a/test/js/sql/sqlite-sql.test.ts b/test/js/sql/sqlite-sql.test.ts index d73b351e3bc5..cba0fe4eb161 100644 --- a/test/js/sql/sqlite-sql.test.ts +++ b/test/js/sql/sqlite-sql.test.ts @@ -1209,6 +1209,43 @@ describe("Transactions", () => { expect(accounts[1].balance).toBe(500); }); + // A query is lazy. One created inside the callback but first awaited after the + // transaction settled must not run outside the transaction. + test.each([ + ["tagged template", (tx: Bun.TransactionSQL) => tx`UPDATE accounts SET balance = 0 WHERE id = 1`], + ["unsafe()", (tx: Bun.TransactionSQL) => tx.unsafe("UPDATE accounts SET balance = 0 WHERE id = 1")], + ])("a %s query first awaited after the transaction settled rejects", async (_name, make) => { + let afterRollback!: Promise; + let afterCommit!: Promise; + + const err = await sql + .begin(async tx => { + afterRollback = make(tx); + throw new Error("roll back"); + }) + .catch(e => e); + expect(err.message).toBe("roll back"); + expect( + await afterRollback.then( + () => null, + e => e.code, + ), + ).toBe("ERR_SQLITE_CONNECTION_CLOSED"); + + await sql.begin(async tx => { + afterCommit = make(tx); + }); + expect( + await afterCommit.then( + () => null, + e => e.code, + ), + ).toBe("ERR_SQLITE_CONNECTION_CLOSED"); + + const accounts = await sql`SELECT balance FROM accounts WHERE id = 1`; + expect(accounts).toEqual([{ balance: 1000 }]); + }); + test("nested transactions (savepoints)", async () => { await sql.begin(async tx => { await tx`UPDATE accounts SET balance = balance - 100 WHERE id = 1`;