From 84cf74971cf70001f7efb549c101ba679d7b147e Mon Sep 17 00:00:00 2001 From: robobun Date: Tue, 9 Jun 2026 11:50:32 +0000 Subject: [PATCH 1/4] mysql: route query writes through advance() so wire order matches queue order JSMySQLQuery::do_run wrote a query's packets optimistically before enqueueing it, while responses are matched to requests in FIFO queue order. A write from the enqueue path could jump ahead of an earlier queued-but-unwritten request, making that request consume the new query's response packets and desyncing the connection. do_run now only enqueues; the queue's advance() walk (pumped by the auto-flusher and the response handlers) performs every write in queue order, same as the postgres fix in #32006. This also fixes a hang: when run() failed for an already-queued query (e.g. queued behind an in-flight COM_STMT_PREPARE of the same statement whose prepare then failed), run()'s error guard pre-marked the query as failed, so reject_with_js_value's settle-once gate dropped the rejection and the promise never settled. run() no longer marks the query failed; the reject path owns that transition. Fixes #32005 --- src/sql_jsc/mysql/JSMySQLQuery.rs | 40 ++-- .../sql/sql-mysql-queue-write-order.test.ts | 187 ++++++++++++++++++ 2 files changed, 208 insertions(+), 19 deletions(-) create mode 100644 test/js/sql/sql-mysql-queue-write-order.test.ts diff --git a/src/sql_jsc/mysql/JSMySQLQuery.rs b/src/sql_jsc/mysql/JSMySQLQuery.rs index efd430326452..c9fb21fe0e9d 100644 --- a/src/sql_jsc/mysql/JSMySQLQuery.rs +++ b/src/sql_jsc/mysql/JSMySQLQuery.rs @@ -192,16 +192,22 @@ impl JSMySQLQuery { return Err(global_object.throw_invalid_argument_type("run", "query", "Query")); } this.set_target(target); - if let Err(err) = this.run(connection) { - if !global_object.has_exception() { - return Err(global_object.throw_value(mysql_error_to_js( - global_object, - "failed to execute query", - err, - ))); - } - return Err(jsc::JsError::Thrown); - } + // Keep the JS wrapper alive while the request sits in the queue: the + // cached target/binding/columns properties live on it. advance() also + // upgrades when it first touches the request, but that can be after a + // GC-observable gap (e.g. while an earlier statement is preparing). + if this.is_pending() { + this.this_value.with_mut(|v| v.upgrade(global_object)); + } + // Do not write the query here. Responses are matched to requests in + // FIFO queue order, so every write must go through advance(), which + // walks the queue in that order and enforces the ordering barriers. + // Writing from the enqueue path could put this query's packets on the + // wire ahead of an earlier queued-but-unwritten request, making that + // request consume this query's response packets and desyncing the + // connection. enqueue_request() registers the auto-flusher, which + // pumps the queue through advance() on an idle connection; otherwise + // the response handlers advance it. connection.enqueue_request(this.as_ctx_ptr()); Ok(JSValue::UNDEFINED) } @@ -435,13 +441,6 @@ impl JSMySQLQuery { } let global_object: &JSGlobalObject = self.global_object(); self.this_value.with_mut(|v| v.upgrade(global_object)); - // R-2: errdefer rollback — `&Self` is `Copy`; the guard captures it by - // value, mutation is `JsCell`-backed, and `into_inner` disarms on the - // success path below. - let errguard = scopeguard::guard(self, |s| { - s.this_value.with_mut(|v| v.downgrade()); - let _ = s.query.with_mut(|q| q.fail()); - }); let columns_value = self.get_columns().unwrap_or(JSValue::UNDEFINED); let binding_value = self.get_binding().unwrap_or(JSValue::UNDEFINED); @@ -467,10 +466,13 @@ impl JSMySQLQuery { err, )); } + // Do not mark the query failed here: the caller (advance) routes + // the error to on_error → reject, whose fail() gate must still be + // open or the rejection is silently dropped and the promise never + // settles. reject/mark_as_failed own the Fail transition and the + // this_value downgrade. return Err(AnyMySQLError::Error::JSError); } - // disarm errdefer on success - scopeguard::ScopeGuard::into_inner(errguard); Ok(()) } diff --git a/test/js/sql/sql-mysql-queue-write-order.test.ts b/test/js/sql/sql-mysql-queue-write-order.test.ts new file mode 100644 index 000000000000..9022a6712109 --- /dev/null +++ b/test/js/sql/sql-mysql-queue-write-order.test.ts @@ -0,0 +1,187 @@ +// https://github.com/oven-sh/bun/issues/32005 +// +// The MySQL native request queue matches server responses to requests in FIFO +// queue order, so query packets must reach the wire in that same order. +// JSMySQLQuery::do_run used to write optimistically before enqueueing, which +// could put a query's packets ahead of an earlier queued-but-unwritten +// request; all writes now go through the queue's advance() walk. +// +// The optimistic write path also had a concrete user-visible bug: when run() +// failed for a query that was already queued (e.g. it was queued behind an +// in-flight COM_STMT_PREPARE of the same statement and that prepare failed), +// run() pre-marked the query as failed, so the reject path's settle-once gate +// concluded the query was already settled and the promise never rejected. +// +// Uses a minimal mock MySQL server so it can run without Docker. + +import { SQL } from "bun"; +import { expect, test } from "bun:test"; +import { once } from "events"; +import net from "net"; + +function u16le(n: number) { + return Buffer.from([n & 0xff, (n >> 8) & 0xff]); +} +function u24le(n: number) { + return Buffer.from([n & 0xff, (n >> 8) & 0xff, (n >> 16) & 0xff]); +} +function u32le(n: number) { + return Buffer.from([n & 0xff, (n >> 8) & 0xff, (n >> 16) & 0xff, (n >>> 24) & 0xff]); +} +function packet(seq: number, payload: Buffer) { + return Buffer.concat([u24le(payload.length), Buffer.from([seq]), payload]); +} + +// Server capability flags (subset sufficient for these paths). +const CLIENT_PROTOCOL_41 = 1 << 9; +const CLIENT_SECURE_CONNECTION = 1 << 15; +const CLIENT_PLUGIN_AUTH = 1 << 19; +const CLIENT_PLUGIN_AUTH_LENENC_CLIENT_DATA = 1 << 21; +const CLIENT_DEPRECATE_EOF = 1 << 24; +const SERVER_CAPS = + CLIENT_PROTOCOL_41 | + CLIENT_SECURE_CONNECTION | + CLIENT_PLUGIN_AUTH | + CLIENT_PLUGIN_AUTH_LENENC_CLIENT_DATA | + CLIENT_DEPRECATE_EOF; + +function handshakeV10() { + const authData1 = Buffer.alloc(8, 0x61); + const authData2 = Buffer.alloc(13, 0x62); // includes trailing NUL as part of 13 bytes + authData2[12] = 0; + const payload = Buffer.concat([ + Buffer.from([10]), // protocol version + Buffer.from("mock-5.7.0\0"), // server version NUL-terminated + u32le(1), // connection id + authData1, // auth-plugin-data-part-1 (8) + Buffer.from([0]), // filler + u16le(SERVER_CAPS & 0xffff), // capability flags lower + Buffer.from([0x2d]), // character set (utf8mb4_general_ci) + u16le(0x0002), // status flags (SERVER_STATUS_AUTOCOMMIT) + u16le((SERVER_CAPS >>> 16) & 0xffff), // capability flags upper + Buffer.from([21]), // length of auth-plugin-data + Buffer.alloc(10, 0), // reserved + authData2, // auth-plugin-data-part-2 (13 bytes) + Buffer.from("mysql_native_password\0"), + ]); + return packet(0, payload); +} + +function okPacket(seq: number) { + // header, affected_rows (lenenc 0), last_insert_id (lenenc 0), status flags, warnings + return packet(seq, Buffer.from([0x00, 0x00, 0x00, 0x02, 0x00, 0x00, 0x00])); +} + +function errorPacket(seq: number, errno: number, message: string) { + const payload = Buffer.concat([Buffer.from([0xff]), u16le(errno), Buffer.from("#42000"), Buffer.from(message)]); + return packet(seq, payload); +} + +const COM_QUERY = 0x03; +const COM_STMT_PREPARE = 0x16; + +// Mock server: OK to every COM_QUERY, ERROR to every COM_STMT_PREPARE, and a +// wire-order log of the commands it received. +function mockServer() { + const wireLog: string[] = []; + const server = net.createServer(socket => { + let buffered = Buffer.alloc(0); + let authed = false; + + socket.write(handshakeV10()); + + socket.on("data", chunk => { + buffered = Buffer.concat([buffered, chunk]); + while (buffered.length >= 4) { + const len = buffered[0] | (buffered[1] << 8) | (buffered[2] << 16); + if (buffered.length < 4 + len) break; + const seq = buffered[3]; + const payload = buffered.subarray(4, 4 + len); + buffered = buffered.subarray(4 + len); + + if (!authed) { + // HandshakeResponse41 from client -> accept unconditionally. + authed = true; + socket.write(okPacket(seq + 1)); + continue; + } + + const cmd = payload[0]; + if (cmd === COM_STMT_PREPARE) { + wireLog.push(`prepare:${payload.subarray(1).toString()}`); + socket.write(errorPacket(seq + 1, 1064, "mock prepare failure")); + } else if (cmd === COM_QUERY) { + wireLog.push(`query:${payload.subarray(1).toString()}`); + socket.write(okPacket(seq + 1)); + } else { + // COM_QUIT or anything else -> close. + socket.end(); + } + } + }); + }); + return { server, wireLog }; +} + +test("query queued behind a failing prepare of the same statement rejects instead of hanging", async () => { + const { server, wireLog } = mockServer(); + server.listen(0, "127.0.0.1"); + await once(server, "listening"); + const { port } = server.address() as net.AddressInfo; + + try { + await using sql = new SQL({ url: `mysql://root@127.0.0.1:${port}/db`, max: 1 }); + + // Same query text and parameter shape -> same statement signature. q2 is + // issued while q1's COM_STMT_PREPARE is still in flight, so it joins q1's + // statement and waits in the queue. When the prepare fails, q1 rejects + // and q2 must reject too (with the cached statement error); before the + // fix q2's promise never settled. + const q1 = sql`wat ${1}`; + const q2 = sql`wat ${1}`; + (q1 as any).execute(); + (q2 as any).execute(); + + const settle = (q: Promise) => + q.then( + () => ({ ok: true }) as const, + (err: any) => ({ ok: false, errno: err?.errno, message: String(err?.message ?? err) }) as const, + ); + const [r1, r2] = await Promise.all([settle(q1), settle(q2)]); + + expect(r1).toEqual({ ok: false, errno: 1064, message: "mock prepare failure" }); + expect(r2).toEqual({ ok: false, errno: 1064, message: "mock prepare failure" }); + + // Only q1's prepare may reach the wire; q2 hits the cached failed + // statement without writing anything. + expect(wireLog).toEqual(["prepare:wat ? "]); + + // The connection must still be usable afterwards. + await sql.unsafe("do 1"); + expect(wireLog).toEqual(["prepare:wat ? ", "query:do 1"]); + } finally { + await new Promise(r => server.close(() => r())); + } +}); + +test("queries reach the wire in issuance order and all settle", async () => { + const { server, wireLog } = mockServer(); + server.listen(0, "127.0.0.1"); + await once(server, "listening"); + const { port } = server.address() as net.AddressInfo; + + try { + await using sql = new SQL({ url: `mysql://root@127.0.0.1:${port}/db`, max: 1 }); + + // Simple-protocol queries (no params) issued in one tick: responses are + // matched to requests in FIFO queue order, so the server must receive + // them in exactly the order they were issued. + const queries = Array.from({ length: 4 }, (_, i) => sql.unsafe(`do ${i}`)); + for (const q of queries) (q as any).execute(); + await Promise.all(queries); + + expect(wireLog).toEqual(["query:do 0", "query:do 1", "query:do 2", "query:do 3"]); + } finally { + await new Promise(r => server.close(() => r())); + } +}); From 69ca8ac53b87976ac9c39a83ae75b72b2c73a86d Mon Sep 17 00:00:00 2001 From: robobun Date: Tue, 9 Jun 2026 12:03:19 +0000 Subject: [PATCH 2/4] test: park queries behind a held in-flight prepare and assert FIFO drain The mock can now hold the first COM_STMT_PREPARE response until the test releases it, so queries issued during that window park in the native queue behind an in-flight request. The wire log must show them only after the release marker, in issuance order. --- .../sql/sql-mysql-queue-write-order.test.ts | 57 ++++++++++++++++++- 1 file changed, 54 insertions(+), 3 deletions(-) diff --git a/test/js/sql/sql-mysql-queue-write-order.test.ts b/test/js/sql/sql-mysql-queue-write-order.test.ts index 9022a6712109..8dee76e6583d 100644 --- a/test/js/sql/sql-mysql-queue-write-order.test.ts +++ b/test/js/sql/sql-mysql-queue-write-order.test.ts @@ -81,9 +81,14 @@ const COM_QUERY = 0x03; const COM_STMT_PREPARE = 0x16; // Mock server: OK to every COM_QUERY, ERROR to every COM_STMT_PREPARE, and a -// wire-order log of the commands it received. -function mockServer() { +// wire-order log of the commands it received. With `holdPrepare`, the first +// prepare's ERROR response is held until the test calls the release function +// resolved through `heldPrepare`, keeping that request in flight on the wire. +function mockServer(opts: { holdPrepare?: boolean } = {}) { const wireLog: string[] = []; + let onPrepareHeld: (release: () => void) => void; + const heldPrepare = new Promise<() => void>(resolve => (onPrepareHeld = resolve)); + let held = false; const server = net.createServer(socket => { let buffered = Buffer.alloc(0); let authed = false; @@ -109,6 +114,15 @@ function mockServer() { const cmd = payload[0]; if (cmd === COM_STMT_PREPARE) { wireLog.push(`prepare:${payload.subarray(1).toString()}`); + if (opts.holdPrepare && !held) { + held = true; + const respond = () => { + wireLog.push("release"); + socket.write(errorPacket(seq + 1, 1064, "mock prepare failure")); + }; + onPrepareHeld(respond); + continue; + } socket.write(errorPacket(seq + 1, 1064, "mock prepare failure")); } else if (cmd === COM_QUERY) { wireLog.push(`query:${payload.subarray(1).toString()}`); @@ -120,7 +134,7 @@ function mockServer() { } }); }); - return { server, wireLog }; + return { server, wireLog, heldPrepare }; } test("query queued behind a failing prepare of the same statement rejects instead of hanging", async () => { @@ -185,3 +199,40 @@ test("queries reach the wire in issuance order and all settle", async () => { await new Promise(r => server.close(() => r())); } }); + +test("queries issued while a request is in flight stay parked behind it and drain in FIFO order", async () => { + const { server, wireLog, heldPrepare } = mockServer({ holdPrepare: true }); + server.listen(0, "127.0.0.1"); + await once(server, "listening"); + const { port } = server.address() as net.AddressInfo; + + try { + await using sql = new SQL({ url: `mysql://root@127.0.0.1:${port}/db`, max: 1 }); + + // Prepared query whose COM_STMT_PREPARE the server holds in flight. + const q0 = sql`hold ${1}`; + (q0 as any).execute(); + const release = await heldPrepare; // the server has received the prepare + + // Issued while that prepare is in flight: these park in the native queue + // behind it. None of them may reach the wire before the server answers + // the prepare (the "release" marker); afterwards they drain in issuance + // order. The client only processes the release bytes on a later I/O + // event, after all pending microtasks (including these dispatches) ran. + const queries = Array.from({ length: 3 }, (_, i) => sql.unsafe(`do ${i}`)); + for (const q of queries) (q as any).execute(); + + release(); + + const q0errno = await q0.then( + () => null, + (err: any) => err?.errno, + ); + await Promise.all(queries); + + expect(q0errno).toBe(1064); + expect(wireLog).toEqual(["prepare:hold ? ", "release", "query:do 0", "query:do 1", "query:do 2"]); + } finally { + await new Promise(r => server.close(() => r())); + } +}); From e14fba981aa53c3585c8342fb988e590a00a5d38 Mon Sep 17 00:00:00 2001 From: robobun Date: Tue, 9 Jun 2026 12:10:18 +0000 Subject: [PATCH 3/4] test: reject heldPrepare when the mock connection dies early If the handshake or the held COM_STMT_PREPARE never arrives, the awaiting test now fails immediately with a message instead of burning the per-test timeout. --- test/js/sql/sql-mysql-queue-write-order.test.ts | 16 ++++++++++++++-- 1 file changed, 14 insertions(+), 2 deletions(-) diff --git a/test/js/sql/sql-mysql-queue-write-order.test.ts b/test/js/sql/sql-mysql-queue-write-order.test.ts index 8dee76e6583d..253a0de02496 100644 --- a/test/js/sql/sql-mysql-queue-write-order.test.ts +++ b/test/js/sql/sql-mysql-queue-write-order.test.ts @@ -86,10 +86,22 @@ const COM_STMT_PREPARE = 0x16; // resolved through `heldPrepare`, keeping that request in flight on the wire. function mockServer(opts: { holdPrepare?: boolean } = {}) { const wireLog: string[] = []; - let onPrepareHeld: (release: () => void) => void; - const heldPrepare = new Promise<() => void>(resolve => (onPrepareHeld = resolve)); + const { + promise: heldPrepare, + resolve: onPrepareHeld, + reject: rejectHeldPrepare, + } = Promise.withResolvers<() => void>(); let held = false; const server = net.createServer(socket => { + if (opts.holdPrepare) { + // Fail the awaiting test fast with a message if the connection dies + // before the prepare is held; no-ops once heldPrepare has resolved. + socket.once("error", rejectHeldPrepare); + socket.once("close", () => { + rejectHeldPrepare(new Error("mock connection closed before COM_STMT_PREPARE was observed")); + }); + } + let buffered = Buffer.alloc(0); let authed = false; From 3d07e55742b2c7848a7834b925d57d0188c2a844 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Wed, 10 Jun 2026 03:14:54 +0000 Subject: [PATCH 4/4] ci: retrigger