ReviewOS
Reload

also looking at this

stacks/bun-query-builder

fix(client): a builder's writes belong to the builder's own connection

#1065
Merged glennmichael123 wants to merge fix/builder-local-connection into main
2 files +161 -29
packages/bun-query-builder/src/client.tsmodified+49-29
@@ -2740,6 +2740,28 @@ export function createQueryBuilder<DB extends DatabaseSchema<any>>(state?: Parti
27402740 // Single boundary cast: `state.sql` is `any` (allows mock/tx injection) and
27412741 // getOrCreateBunSql() returns Bun's `SQL`; both satisfy DriverConnection. With
27422742 // `_sql` typed, the downstream `.unsafe(...)` calls no longer need casts (#1044).
2743 //
2744 // EVERY statement this builder executes must go through `_sql`, never the
2745 // module-global `bunSql`. They are the same object only for a plain builder;
2746 // the moment one is created with its own connection — `db.transaction()`,
2747 // `savepoint()`, `beginDistributed()`, `reserve()`, or an explicit
2748 // `createQueryBuilder({ sql })` — they are two different connections, and the
2749 // global is a DIFFERENT pooled session from the one the builder's own reads
2750 // and writes use.
2751 //
2752 // That is not a cosmetic distinction. A write issued on the global while the
2753 // caller is inside `db.transaction()` autocommits on its own session, so it
2754 // outlives a ROLLBACK: the transaction fails, the caller sees the error, and
2755 // the row is still there. A read issued on the global cannot see the
2756 // transaction's uncommitted rows, so read-then-decide logic acts on stale
2757 // data. With `pool: { max: 1 }` it does not even get that far — the call
2758 // blocks forever waiting for the single connection its own transaction is
2759 // holding. All three are silent; none produces a driver error.
2760 //
2761 // The exceptions are the pool-lifecycle methods (`reserve`, `close`), which
2762 // act on the pool itself rather than on a session, and the two-phase
2763 // `commitDistributed`/`rollbackDistributed`, which by design run outside the
2764 // transaction that prepared them.
27432765 const _sql: DriverConnection = (state?.sql ?? getOrCreateBunSql()) as unknown as DriverConnection
27442766 const meta = state?.meta
27452767 const schema = state?.schema
@@ -6973,7 +6995,7 @@ export function createQueryBuilder<DB extends DatabaseSchema<any>>(state?: Parti
69736995 let hash = 7
69746996 for (let i = 0; i < s.length; i++) hash = (hash * 31 + s.charCodeAt(i)) | 0
69756997 const k = typeof key === 'number' ? key : Math.abs(hash)
6976 const q = bunSql`SELECT pg_advisory_lock(${k})`
6998 const q = _sql`SELECT pg_advisory_lock(${k})`
69776999 await runWithHooks<any[]>(q, 'raw')
69787000 return
69797001 }
@@ -6981,7 +7003,7 @@ export function createQueryBuilder<DB extends DatabaseSchema<any>>(state?: Parti
69817003 // MySQL has `GET_LOCK(name, timeout)`. Wait indefinitely
69827004 // (timeout=-1) to match Postgres `pg_advisory_lock` semantics.
69837005 const lockName = `bqb:${String(key)}`
6984 const q = bunSql`SELECT GET_LOCK(${lockName}, -1) AS ok`
7006 const q = _sql`SELECT GET_LOCK(${lockName}, -1) AS ok`
69857007 await runWithHooks<any[]>(q, 'raw')
69867008 return
69877009 }
@@ -6997,7 +7019,7 @@ export function createQueryBuilder<DB extends DatabaseSchema<any>>(state?: Parti
69977019 let hash = 7
69987020 for (let i = 0; i < s.length; i++) hash = (hash * 31 + s.charCodeAt(i)) | 0
69997021 const k = typeof key === 'number' ? key : Math.abs(hash)
7000 const q = bunSql`SELECT pg_try_advisory_lock(${k}) as ok`
7022 const q = _sql`SELECT pg_try_advisory_lock(${k}) as ok`
70017023 const rows = await runWithHooks<any[]>(q, 'raw')
70027024 return Boolean(rows?.[0]?.ok)
70037025 }
@@ -7005,7 +7027,7 @@ export function createQueryBuilder<DB extends DatabaseSchema<any>>(state?: Parti
70057027 // MySQL `GET_LOCK(name, 0)` returns 1 immediately if free, 0
70067028 // if held by another connection.
70077029 const lockName = `bqb:${String(key)}`
7008 const q = bunSql`SELECT GET_LOCK(${lockName}, 0) AS ok`
7030 const q = _sql`SELECT GET_LOCK(${lockName}, 0) AS ok`
70097031 const rows = await runWithHooks<any[]>(q, 'raw')
70107032 return Number(rows?.[0]?.ok) === 1
70117033 }
@@ -7023,7 +7045,7 @@ export function createQueryBuilder<DB extends DatabaseSchema<any>>(state?: Parti
70237045 return (_sql as any).unsafe(query, params)
70247046 },
70257047 file(path: string, params?: any[]) {
7026 return (bunSql as any).file(path, params)
7048 return (_sql as any).file(path, params)
70277049 },
70287050 async reserve() {
70297051 const reserved = await (bunSql as any).reserve()
@@ -7058,7 +7080,7 @@ export function createQueryBuilder<DB extends DatabaseSchema<any>>(state?: Parti
70587080 },
70597081 async ping() {
70607082 try {
7061 const q = bunSql`SELECT 1`
7083 const q = _sql`SELECT 1`
70627084 await runWithHooks<any[]>(q, 'select')
70637085 return true
70647086 }
renamed throughout · 34 hidden lines
@@ -7242,28 +7264,28 @@ export function createQueryBuilder<DB extends DatabaseSchema<any>>(state?: Parti renamed throughout
72427264 const sqlText = isMysqlLike(config.dialect)
72437265 ? `INSERT IGNORE INTO ${tbl} (${colsSql}) VALUES ${valuesSql}`
72447266 : `INSERT INTO ${tbl} (${colsSql}) VALUES ${valuesSql} ON CONFLICT DO NOTHING`
7245 return (bunSql.unsafe(sqlText, params) as any).execute()
7267 return (_sql.unsafe(sqlText, params) as any).execute()
72467268 },
72477269 async insertGetId(table, values, idColumn = 'id' as any) {
72487270 const { colsSql, valuesSql, params } = buildInsertClause([values as Record<string, any>])
72497271 const tbl = quoteInsertIdent(String(table))
72507272 if (isMysqlLike(config.dialect)) {
72517273 // MySQL has no RETURNING — insert then read LAST_INSERT_ID().
7252 await (bunSql.unsafe(`INSERT INTO ${tbl} (${colsSql}) VALUES ${valuesSql}`, params) as any).execute()
7253 const [row] = await (bunSql.unsafe(`SELECT LAST_INSERT_ID() as id`) as any).execute()
7274 await (_sql.unsafe(`INSERT INTO ${tbl} (${colsSql}) VALUES ${valuesSql}`, params) as any).execute()
7275 const [row] = await (_sql.unsafe(`SELECT LAST_INSERT_ID() as id`) as any).execute()
72547276 return row?.id
72557277 }
72567278 if (config.dialect === 'sqlite') {
72577279 // The bun:sqlite wrapper returns { changes, lastInsertRowid } rather
72587280 // than RETURNING rows, so read the rowid directly.
7259 const res = await (bunSql.unsafe(`INSERT INTO ${tbl} (${colsSql}) VALUES ${valuesSql}`, params) as any).execute()
7281 const res = await (_sql.unsafe(`INSERT INTO ${tbl} (${colsSql}) VALUES ${valuesSql}`, params) as any).execute()
72607282 if (res?.lastInsertRowid != null)
72617283 return res.lastInsertRowid
7262 const [row] = await (bunSql.unsafe(`SELECT last_insert_rowid() as id`) as any).execute()
7284 const [row] = await (_sql.unsafe(`SELECT last_insert_rowid() as id`) as any).execute()
72637285 return row?.id
72647286 }
72657287 // Postgres supports RETURNING.
7266 const [row] = await (bunSql.unsafe(`INSERT INTO ${tbl} (${colsSql}) VALUES ${valuesSql} RETURNING ${quoteInsertIdent(String(idColumn))} as id`, params) as any).execute()
7288 const [row] = await (_sql.unsafe(`INSERT INTO ${tbl} (${colsSql}) VALUES ${valuesSql} RETURNING ${quoteInsertIdent(String(idColumn))} as id`, params) as any).execute()
72677289 return row?.id
72687290 },
72697291 async updateOrInsert(table, match, values) {
renamed throughout · 21 hidden lines
@@ -7272,18 +7294,18 @@ export function createQueryBuilder<DB extends DatabaseSchema<any>>(state?: Parti renamed throughout
72727294 let idx = 1
72737295 const whereSql = matchKeys.map(k => `${quoteInsertIdent(k)} = ${getPlaceholder(idx++)}`).join(' AND ')
72747296 const whereParams = matchKeys.map(k => (match as any)[k])
7275 const existsRows = await (bunSql.unsafe(`SELECT 1 FROM ${tbl} WHERE ${whereSql} LIMIT 1`, whereParams) as any).execute()
7297 const existsRows = await (_sql.unsafe(`SELECT 1 FROM ${tbl} WHERE ${whereSql} LIMIT 1`, whereParams) as any).execute()
72767298 if ((existsRows as any[]).length) {
72777299 const setKeys = Object.keys(values)
72787300 let i = 1
72797301 const setSql = setKeys.map(k => `${quoteInsertIdent(k)} = ${getPlaceholder(i++)}`).join(', ')
72807302 const whereSql2 = matchKeys.map(k => `${quoteInsertIdent(k)} = ${getPlaceholder(i++)}`).join(' AND ')
72817303 const params = [...setKeys.map(k => (values as any)[k]), ...matchKeys.map(k => (match as any)[k])]
7282 await (bunSql.unsafe(`UPDATE ${tbl} SET ${setSql} WHERE ${whereSql2}`, params) as any).execute()
7304 await (_sql.unsafe(`UPDATE ${tbl} SET ${setSql} WHERE ${whereSql2}`, params) as any).execute()
72837305 return true
72847306 }
72857307 const { colsSql, valuesSql, params } = buildInsertClause([{ ...match, ...values } as Record<string, any>])
7286 await (bunSql.unsafe(`INSERT INTO ${tbl} (${colsSql}) VALUES ${valuesSql}`, params) as any).execute()
7308 await (_sql.unsafe(`INSERT INTO ${tbl} (${colsSql}) VALUES ${valuesSql}`, params) as any).execute()
72877309 return true
72887310 },
72897311 async upsert(table, rows, conflictColumns, mergeColumns) {
renamed throughout · 20 hidden lines
@@ -7301,16 +7323,16 @@ export function createQueryBuilder<DB extends DatabaseSchema<any>>(state?: Parti renamed throughout
73017323 // "do nothing" form (an empty SET is a syntax error). See #1035, #1052.
73027324 if (isMysqlLike(config.dialect)) {
73037325 if (setCols.length === 0)
7304 return (bunSql.unsafe(`INSERT IGNORE INTO ${tbl} (${colsSql}) VALUES ${valuesSql}`, params) as any).execute()
7326 return (_sql.unsafe(`INSERT IGNORE INTO ${tbl} (${colsSql}) VALUES ${valuesSql}`, params) as any).execute()
73057327 const updateList = setCols.map(c => `${quoteInsertIdent(c)} = VALUES(${quoteInsertIdent(c)})`).join(', ')
7306 return (bunSql.unsafe(`${insert} ON DUPLICATE KEY UPDATE ${updateList}`, params) as any).execute()
7328 return (_sql.unsafe(`${insert} ON DUPLICATE KEY UPDATE ${updateList}`, params) as any).execute()
73077329 }
73087330
73097331 const targets = targetCols.map(quoteInsertIdent).join(', ')
73107332 if (setCols.length === 0)
7311 return (bunSql.unsafe(`${insert} ON CONFLICT (${targets}) DO NOTHING`, params) as any).execute()
7333 return (_sql.unsafe(`${insert} ON CONFLICT (${targets}) DO NOTHING`, params) as any).execute()
73127334 const updateList = setCols.map(c => `${quoteInsertIdent(c)} = EXCLUDED.${quoteInsertIdent(c)}`).join(', ')
7313 return (bunSql.unsafe(`${insert} ON CONFLICT (${targets}) DO UPDATE SET ${updateList}`, params) as any).execute()
7335 return (_sql.unsafe(`${insert} ON CONFLICT (${targets}) DO UPDATE SET ${updateList}`, params) as any).execute()
73147336 },
73157337 async save(table, values) {
73167338 const pk = meta?.primaryKeys[String(table)] ?? 'id'
@@ -7361,7 +7383,7 @@ export function createQueryBuilder<DB extends DatabaseSchema<any>>(state?: Parti
73617383 const start = Date.now()
73627384 try {
73637385 config.hooks?.onQueryStart?.({ sql: query, kind: 'raw' })
7364 const res = await (bunSql as any).unsafe(query)
7386 const res = await (_sql as any).unsafe(query)
73657387 config.hooks?.onQueryEnd?.({ sql: query, durationMs: Date.now() - start, kind: 'raw' })
73667388 return res
73677389 }
@@ -7383,11 +7405,9 @@ export function createQueryBuilder<DB extends DatabaseSchema<any>>(state?: Parti
73837405
73847406 if (config.dialect === 'postgres') {
73857407 // For PostgreSQL, use RETURNING to get the ID, then fetch the full row
7386 const q = bunSql`INSERT INTO ${bunSql(String(table))} ${bunSql(values as any)} RETURNING ${bunSql(String(pk))} as id`
7408 const q = _sql`INSERT INTO ${_sql(String(table))} ${_sql(values as any)} RETURNING ${_sql(String(pk))} as id`
73877409 const [result] = await q.execute()
73887410
7389 console.log('resultId', result)
7390
73917411 if (!result?.id) {
73927412 console.error(`create() failed to get insert ID for table ${String(table)}`)
73937413 console.error('Inserted values:', values)
@@ -7600,28 +7620,28 @@ export function createQueryBuilder<DB extends DatabaseSchema<any>>(state?: Parti
76007620 return await (this as any).create(table, { ...(match as any), ...(values as any) })
76017621 },
76027622 async count(table, column) {
7603 const col = column ? bunSql(String(column)) : bunSql`*`
7604 const q = bunSql`SELECT COUNT(${col}) as c FROM ${bunSql(String(table))}`
7623 const col = column ? _sql(String(column)) : _sql`*`
7624 const q = _sql`SELECT COUNT(${col}) as c FROM ${_sql(String(table))}`
76057625 const [row] = await (q as any).execute()
76067626 return Number((row?.c ?? 0) as any)
76077627 },
76087628 async sum(table, column) {
7609 const q = bunSql`SELECT SUM(${bunSql(String(column))}) as s FROM ${bunSql(String(table))}`
7629 const q = _sql`SELECT SUM(${_sql(String(column))}) as s FROM ${_sql(String(table))}`
76107630 const [row] = await (q as any).execute()
76117631 return Number((row?.s ?? 0) as any)
76127632 },
76137633 async avg(table, column) {
7614 const q = bunSql`SELECT AVG(${bunSql(String(column))}) as a FROM ${bunSql(String(table))}`
7634 const q = _sql`SELECT AVG(${_sql(String(column))}) as a FROM ${_sql(String(table))}`
76157635 const [row] = await (q as any).execute()
76167636 return Number((row?.a ?? 0) as any)
76177637 },
76187638 async min(table, column) {
7619 const q = bunSql`SELECT MIN(${bunSql(String(column))}) as m FROM ${bunSql(String(table))}`
7639 const q = _sql`SELECT MIN(${_sql(String(column))}) as m FROM ${_sql(String(table))}`
76207640 const [row] = await (q as any).execute()
76217641 return (row?.m as any)
76227642 },
76237643 async max(table, column) {
7624 const q = bunSql`SELECT MAX(${bunSql(String(column))}) as m FROM ${bunSql(String(table))}`
7644 const q = _sql`SELECT MAX(${_sql(String(column))}) as m FROM ${_sql(String(table))}`
76257645 const [row] = await (q as any).execute()
76267646 return (row?.m as any)
76277647 },
packages/bun-query-builder/test/client.transaction-connection.test.tsadded+112-0
@@ -0,0 +1,112 @@
1/**
2 * Every statement a builder issues must go through the builder's OWN
3 * connection (`_sql`), never the module-global `bunSql`.
4 *
5 * They are the same object for a plain builder, so this is invisible until a
6 * builder is created with a connection of its own — which is exactly what
7 * `db.transaction()` does for its callback. Then the global is a DIFFERENT
8 * pooled session, and ~29 helper methods were using it:
9 *
10 * - A write inside `db.transaction()` autocommitted on that other session and
11 * SURVIVED THE ROLLBACK. The transaction failed, the caller saw the error,
12 * and the row was still in the table. Silent, with no driver error.
13 * - A read inside the transaction could not see the transaction's own
14 * uncommitted rows, so `tx.count()` returned 0 for a row `tx` had just
15 * inserted and read-then-decide logic acted on stale data.
16 *
17 * SQLite cannot catch either one: `bunSql` and the transaction share a single
18 * `bun:sqlite` handle, so the distinction the bug turns on does not exist. It
19 * needs a real pooled driver, so this file is gated on a live Postgres and runs
20 * in a subprocess (an earlier `configureOrm()` elsewhere in the suite pins the
21 * global executor at sqlite and would mask the network path).
22 *
23 * `unsafe()` was fixed for this once already (client.ts, the comment about raw
24 * SQL escaping the transaction); these are the rest of the same family.
25 */
26
27import { mkdtempSync, rmSync, writeFileSync } from 'node:fs'
28import { tmpdir } from 'node:os'
29import { join, resolve } from 'node:path'
30import process from 'node:process'
31import { SQL } from 'bun'
32import { describe, expect, it } from 'bun:test'
33
34const PG_URL = `postgres://${process.env.USER}@localhost:5432/postgres`
35
36let pgAvailable = false
37try {
38 const probe = new SQL(PG_URL)
39 await probe.unsafe('SELECT 1')
40 await probe.end()
41 pgAvailable = true
42}
43catch {
44 pgAvailable = false
45}
46
47describe.skipIf(!pgAvailable)('builder-local connection routing', () => {
48 it('keeps transactional writes inside the transaction, and reads inside it current', () => {
49 const srcEntry = resolve(import.meta.dir, '../src/index.ts')
50 const dir = mkdtempSync(join(tmpdir(), 'qb-txconn-'))
51 const scriptPath = join(dir, 'probe.ts')
52
53 writeFileSync(scriptPath, `
54import { SQL } from 'bun'
55import { setConfig, createQueryBuilder, buildDatabaseSchema } from ${JSON.stringify(srcEntry)}
56
57const URL = ${JSON.stringify(PG_URL)}
58const raw = new SQL(URL)
59await raw.unsafe('DROP TABLE IF EXISTS _qb_txconn')
60await raw.unsafe('CREATE TABLE _qb_txconn (id serial primary key, name text)')
61
62setConfig({ dialect: 'postgres', database: { url: URL } })
63const schema = buildDatabaseSchema({
64 Widget: { name: 'Widget', table: '_qb_txconn', primaryKey: 'id', attributes: { id: { validation: { rule: {} } }, name: { validation: { rule: {} } } } },
65})
66const db = createQueryBuilder({ schema })
67
68// A write in a transaction that throws must not outlive the ROLLBACK.
69try {
70 await db.transaction(async (tx) => {
71 await tx.insertGetId('_qb_txconn', { name: 'rolled-back' })
72 throw new Error('deliberate failure')
73 })
74}
75catch {}
76const survivors = await raw.unsafe('SELECT name FROM _qb_txconn')
77
78// A read in a transaction must see that transaction's own uncommitted write.
79let seenInside = -1
80await db.transaction(async (tx) => {
81 await tx.insertGetId('_qb_txconn', { name: 'committed' })
82 seenInside = Number(await tx.count('_qb_txconn'))
83})
84const committed = await raw.unsafe('SELECT name FROM _qb_txconn')
85
86await raw.unsafe('DROP TABLE IF EXISTS _qb_txconn')
87await raw.end()
88
89console.log(JSON.stringify({
90 survivedRollback: survivors.length,
91 seenInside,
92 committed: committed.length,
93}))
94`)
95
96 const proc = Bun.spawnSync({ cmd: ['bun', scriptPath], cwd: dir, stdout: 'pipe', stderr: 'pipe' })
97 const stdout = new TextDecoder().decode(proc.stdout).trim()
98 const stderr = new TextDecoder().decode(proc.stderr).trim()
99 rmSync(dir, { recursive: true, force: true })
100
101 if (proc.exitCode !== 0)
102 throw new Error(`probe exited ${proc.exitCode}: ${stderr || stdout}`)
103
104 const result = JSON.parse(stdout.split('\n').filter(Boolean).at(-1) ?? '{}')
105
106 // Was 1 — the write autocommitted on the global connection and outlived the rollback.
107 expect(result.survivedRollback).toBe(0)
108 // Was 0 — the count ran on the global connection and could not see the tx's row.
109 expect(result.seenInside).toBe(1)
110 expect(result.committed).toBe(1)
111 })
112})