diff --git a/src/sql/postgres/PostgresRequest.zig b/src/sql/postgres/PostgresRequest.zig index f38966243231..9eb5085d045b 100644 --- a/src/sql/postgres/PostgresRequest.zig +++ b/src/sql/postgres/PostgresRequest.zig @@ -257,6 +257,65 @@ pub fn bindAndExecute( try writer.write(&protocol.Sync); } +/// Atomically sends Parse + [Describe] + Bind + Execute + Flush + Sync as a single message batch. +/// This is required for unnamed prepared statements to work correctly with connection poolers +/// like PgBouncer in transaction mode, which may reassign server connections between protocol +/// round-trips. Without this, Parse and Bind+Execute could be routed to different backend +/// connections, causing queries to execute against the wrong prepared statement. +pub fn parseAndBindAndExecute( + globalObject: *jsc.JSGlobalObject, + query: []const u8, + statement: *PostgresSQLStatement, + array_value: JSValue, + columns_value: JSValue, + include_describe: bool, + comptime Context: type, + writer: protocol.NewWriter(Context), +) AnyPostgresError!void { + const name = statement.signature.prepared_statement_name; + + // Parse + { + var q = protocol.Parse{ + .name = name, + .params = statement.signature.fields, + .query = query, + }; + try q.writeInternal(Context, writer); + debug("Parse: {f}", .{bun.fmt.quote(query)}); + } + + // Describe (needed on first execution to learn parameter/result types for caching) + if (include_describe) { + var d = protocol.Describe{ + .p = .{ + .prepared_statement = name, + }, + }; + try d.writeInternal(Context, writer); + debug("Describe: {f}", .{bun.fmt.quote(name)}); + } + + // Bind — use server-provided types if available (binary format), otherwise + // fall back to signature types (text format for unknowns). The server will + // handle text-to-type conversion based on the parameter types from Parse. + const param_fields = if (statement.parameters.len > 0) statement.parameters else statement.signature.fields; + const result_fields = statement.fields; + + try writeBind(name, bun.String.empty, globalObject, array_value, columns_value, param_fields, result_fields, Context, writer); + + // Execute + var exec = protocol.Execute{ + .p = .{ + .prepared_statement = name, + }, + }; + try exec.writeInternal(Context, writer); + + try writer.write(&protocol.Flush); + try writer.write(&protocol.Sync); +} + pub fn executeQuery( query: []const u8, comptime Context: type, diff --git a/src/sql/postgres/PostgresSQLConnection.zig b/src/sql/postgres/PostgresSQLConnection.zig index 9130e185688e..a64937767463 100644 --- a/src/sql/postgres/PostgresSQLConnection.zig +++ b/src/sql/postgres/PostgresSQLConnection.zig @@ -1090,6 +1090,16 @@ pub fn canPrepareQuery(noalias this: *const @This()) bool { return this.flags.is_ready_for_query and !this.flags.waiting_to_prepare and this.pipelined_requests == 0; } +/// Process pending requests and flush. Called from the enqueue path when +/// unnamed prepared statements with params skip writeQuery+Sync and need +/// advance() to send everything atomically on an idle connection. +pub fn advanceAndFlush(this: *PostgresSQLConnection) void { + if (!this.flags.has_backpressure and this.flags.is_ready_for_query) { + this.advance(); + this.flushData(); + } +} + fn advance(this: *PostgresSQLConnection) void { var offset: usize = 0; debug("advance", .{}); @@ -1182,24 +1192,52 @@ fn advance(this: *PostgresSQLConnection) void { const binding_value = PostgresSQLQuery.js.bindingGetCached(thisValue) orelse .zero; const columns_value = PostgresSQLQuery.js.columnsGetCached(thisValue) orelse .zero; req.flags.binary = statement.fields.len > 0; - debug("binding and executing stmt", .{}); - PostgresRequest.bindAndExecute(this.globalObject, statement, binding_value, columns_value, PostgresSQLConnection.Writer, this.writer()) catch |err| { - if (this.globalObject.tryTakeException()) |err_| { - req.onJSError(err_, this.globalObject); - } else { - req.onWriteFail(err, this.globalObject, this.getQueriesArray()); - } - if (offset == 0) { - req.deref(); - this.requests.discard(1); - } else { - // deinit later - req.status = .fail; - offset += 1; - } - debug("bind and execute failed: {s}", .{@errorName(err)}); - continue; - }; + + if (this.flags.use_unnamed_prepared_statements) { + // For unnamed prepared statements, always include Parse + // before Bind+Execute. The unnamed statement may not exist + // on the current server connection when using PgBouncer or + // other connection poolers in transaction mode. + debug("parse, bind and execute unnamed stmt", .{}); + var query_str = req.query.toUTF8(bun.default_allocator); + defer query_str.deinit(); + PostgresRequest.parseAndBindAndExecute(this.globalObject, query_str.slice(), statement, binding_value, columns_value, false, PostgresSQLConnection.Writer, this.writer()) catch |err| { + if (this.globalObject.tryTakeException()) |err_| { + req.onJSError(err_, this.globalObject); + } else { + req.onWriteFail(err, this.globalObject, this.getQueriesArray()); + } + if (offset == 0) { + req.deref(); + this.requests.discard(1); + } else { + // deinit later + req.status = .fail; + offset += 1; + } + debug("parse, bind and execute failed: {s}", .{@errorName(err)}); + continue; + }; + } else { + debug("binding and executing stmt", .{}); + PostgresRequest.bindAndExecute(this.globalObject, statement, binding_value, columns_value, PostgresSQLConnection.Writer, this.writer()) catch |err| { + if (this.globalObject.tryTakeException()) |err_| { + req.onJSError(err_, this.globalObject); + } else { + req.onWriteFail(err, this.globalObject, this.getQueriesArray()); + } + if (offset == 0) { + req.deref(); + this.requests.discard(1); + } else { + // deinit later + req.status = .fail; + offset += 1; + } + debug("bind and execute failed: {s}", .{@errorName(err)}); + continue; + }; + } this.flags.is_ready_for_query = false; req.status = .binding; @@ -1268,6 +1306,49 @@ fn advance(this: *PostgresSQLConnection) void { return; } + if (this.flags.use_unnamed_prepared_statements) { + // For unnamed prepared statements, send Parse+Describe+Bind+Execute + // atomically to prevent PgBouncer from splitting them across + // server connections. Uses signature field types for encoding + // (text format for unknowns); actual types will be cached from + // ParameterDescription for subsequent executions. + const thisValue = req.thisValue.tryGet() orelse { + bun.assertf(false, "query value was freed earlier than expected", .{}); + bun.assert(offset == 0); + req.deref(); + this.requests.discard(1); + continue; + }; + const binding_value = PostgresSQLQuery.js.bindingGetCached(thisValue) orelse .zero; + const columns_value = PostgresSQLQuery.js.columnsGetCached(thisValue) orelse .zero; + debug("parseAndBindAndExecute (unnamed, first execution)", .{}); + PostgresRequest.parseAndBindAndExecute(this.globalObject, query_str.slice(), statement, binding_value, columns_value, true, PostgresSQLConnection.Writer, this.writer()) catch |err| { + if (this.globalObject.tryTakeException()) |err_| { + req.onJSError(err_, this.globalObject); + } else { + statement.status = .failed; + statement.error_response = .{ .postgres_error = err }; + req.onWriteFail(err, this.globalObject, this.getQueriesArray()); + } + bun.assert(offset == 0); + req.deref(); + this.requests.discard(1); + debug("parseAndBindAndExecute failed: {s}", .{@errorName(err)}); + continue; + }; + this.flags.is_ready_for_query = false; + this.flags.waiting_to_prepare = true; + req.status = .binding; + statement.status = .parsing; + req.flags.pipelined = true; + this.pipelined_requests += 1; + this.flushDataAndResetTimeout(); + return; + } + + // Named prepared statements: send Parse+Describe first, wait for + // ParameterDescription, then send Bind+Execute in a second phase. + // This is safe because named statements persist on the connection. const connection_writer = this.writer(); debug("writing query", .{}); // write query and wait for it to be prepared diff --git a/src/sql/postgres/PostgresSQLQuery.zig b/src/sql/postgres/PostgresSQLQuery.zig index 9ef87c575277..535f7f16e83d 100644 --- a/src/sql/postgres/PostgresSQLQuery.zig +++ b/src/sql/postgres/PostgresSQLQuery.zig @@ -420,7 +420,9 @@ pub fn doRun(this: *PostgresSQLQuery, globalObject: *jsc.JSGlobalObject, callfra this.status = .binding; did_write = true; connection.flags.waiting_to_prepare = true; - } else { + } else if (!connection.flags.use_unnamed_prepared_statements) { + // Named prepared statements: send Parse+Describe+Sync now and wait + // for ParameterDescription before sending Bind+Execute in advance(). debug("writeQuery", .{}); PostgresRequest.writeQuery(query_str.slice(), signature.prepared_statement_name, signature.fields, PostgresSQLConnection.Writer, writer) catch |err| { @@ -450,6 +452,9 @@ pub fn doRun(this: *PostgresSQLQuery, globalObject: *jsc.JSGlobalObject, callfra did_write = true; connection.flags.waiting_to_prepare = true; } + // Unnamed prepared statements with params: skip writeQuery+Sync here. + // advance() will send Parse+Describe+Bind+Execute atomically via + // parseAndBindAndExecute(), preventing PgBouncer from splitting them. } { const stmt = bun.default_allocator.create(PostgresSQLStatement) catch { @@ -465,7 +470,7 @@ pub fn doRun(this: *PostgresSQLQuery, globalObject: *jsc.JSGlobalObject, callfra stmt.* = .{ .signature = signature, .ref_count = .initExactRefs(2), - .status = if (can_execute) .parsing else .pending, + .status = if (did_write) .parsing else .pending, }; this.statement = stmt; @@ -473,7 +478,7 @@ pub fn doRun(this: *PostgresSQLQuery, globalObject: *jsc.JSGlobalObject, callfra } else { stmt.* = .{ .signature = signature, - .status = if (can_execute) .parsing else .pending, + .status = if (did_write) .parsing else .pending, }; this.statement = stmt; } @@ -488,6 +493,9 @@ pub fn doRun(this: *PostgresSQLQuery, globalObject: *jsc.JSGlobalObject, callfra connection.flushDataAndResetTimeout(); } else { connection.resetConnectionTimeout(); + // For unnamed prepared statements with params, we skip writeQuery+Sync + // in the enqueue path and let advance() handle it atomically. + connection.advanceAndFlush(); } return .js_undefined; } diff --git a/test/js/sql/sql-prepare-false.test.ts b/test/js/sql/sql-prepare-false.test.ts new file mode 100644 index 000000000000..a98f3de2b7b1 --- /dev/null +++ b/test/js/sql/sql-prepare-false.test.ts @@ -0,0 +1,135 @@ +import { SQL } from "bun"; +import { afterAll, describe, expect, test } from "bun:test"; +import * as dockerCompose from "../../docker/index.ts"; + +// Tests for `prepare: false` (unnamed prepared statements). +// These verify that parameterized queries work correctly when using unnamed +// prepared statements, which is critical for PgBouncer compatibility. + +describe("PostgreSQL prepare: false", async () => { + let container: { port: number; host: string }; + + try { + const info = await dockerCompose.ensure("postgres_plain"); + container = { port: info.ports[5432], host: info.host }; + } catch (e) { + test.skip(`Docker not available: ${e}`); + return; + } + + const options = { + db: "bun_sql_test", + username: "bun_sql_test", + host: container.host, + port: container.port, + max: 1, + prepare: false, + }; + + afterAll(async () => { + if (!process.env.BUN_KEEP_DOCKER) { + await dockerCompose.down(); + } + }); + + test("basic parameterized query", async () => { + await using db = new SQL(options); + const [{ x }] = await db`SELECT ${42}::int AS x`; + expect(x).toBe(42); + }); + + test("multiple parameterized queries sequentially", async () => { + await using db = new SQL(options); + + const [{ a }] = await db`SELECT ${1}::int AS a`; + expect(a).toBe(1); + + const [{ b }] = await db`SELECT ${"hello"}::text AS b`; + expect(b).toBe("hello"); + + const [{ c }] = await db`SELECT ${3.14}::float8 AS c`; + expect(c).toBeCloseTo(3.14); + }); + + test("same query repeated with different params", async () => { + await using db = new SQL(options); + for (let i = 0; i < 10; i++) { + const [{ x }] = await db`SELECT ${i}::int AS x`; + expect(x).toBe(i); + } + }); + + test("concurrent queries with different tables return correct results", async () => { + // This test simulates the scenario where concurrent unnamed prepared + // statements could interfere with each other via PgBouncer. + await using db = new SQL({ ...options, max: 4 }); + + // Create real tables (not temp, so they're visible across connections) + await db`CREATE TABLE IF NOT EXISTS prepare_false_test_a (id int, val text)`; + await db`CREATE TABLE IF NOT EXISTS prepare_false_test_b (id int, val text)`; + await db`DELETE FROM prepare_false_test_a`; + await db`DELETE FROM prepare_false_test_b`; + await db`INSERT INTO prepare_false_test_a VALUES (1, 'from_a')`; + await db`INSERT INTO prepare_false_test_b VALUES (1, 'from_b')`; + + // Run concurrent parameterized queries against different tables + const results = await Promise.all([ + db`SELECT val FROM prepare_false_test_a WHERE id = ${1}`, + db`SELECT val FROM prepare_false_test_b WHERE id = ${1}`, + db`SELECT val FROM prepare_false_test_a WHERE id = ${1}`, + db`SELECT val FROM prepare_false_test_b WHERE id = ${1}`, + ]); + + expect(results[0][0].val).toBe("from_a"); + expect(results[1][0].val).toBe("from_b"); + expect(results[2][0].val).toBe("from_a"); + expect(results[3][0].val).toBe("from_b"); + + // Cleanup + await db`DROP TABLE IF EXISTS prepare_false_test_a`; + await db`DROP TABLE IF EXISTS prepare_false_test_b`; + }); + + test("parameterized query with multiple params", async () => { + await using db = new SQL(options); + const [{ sum }] = await db`SELECT (${10}::int + ${20}::int) AS sum`; + expect(sum).toBe(30); + }); + + test("query without params still works", async () => { + await using db = new SQL(options); + const [{ x }] = await db`SELECT 1 AS x`; + expect(x).toBe(1); + }); + + test("transactions with parameterized queries", async () => { + await using db = new SQL(options); + + await db`CREATE TEMP TABLE IF NOT EXISTS tx_test (id int, val text)`; + + await db.begin(async tx => { + await tx`INSERT INTO tx_test VALUES (${1}, ${"hello"})`; + await tx`INSERT INTO tx_test VALUES (${2}, ${"world"})`; + }); + + const rows = await db`SELECT * FROM tx_test ORDER BY id`; + expect(rows.length).toBe(2); + expect(rows[0].val).toBe("hello"); + expect(rows[1].val).toBe("world"); + }); + + test("concurrent parameterized queries with high concurrency", async () => { + await using db = new SQL({ ...options, max: 8 }); + + // Fire many concurrent queries to stress-test unnamed statement handling + const promises = []; + for (let i = 0; i < 50; i++) { + promises.push(db`SELECT ${i}::int AS x`.then(r => ({ expected: i, actual: r[0].x }))); + } + + const results = await Promise.all(promises); + for (const { expected, actual } of results) { + expect(actual).toBe(expected); + } + }); +});