fix(Bun.SQL) fix MySQL execution on windows (#22696)

### What does this PR do?
Fixes https://github.com/oven-sh/bun/issues/22695
Fixes https://github.com/oven-sh/bun/issues/22654

### How did you verify your code works?
Added mysql:9 + run mysql tests on windows

<img width="1035" height="708"
alt="489727987-3cca2da4-0ff8-4b4a-b5be-9fbdd1c9862d"
src="https://github.com/user-attachments/assets/02c6880d-547e-43b5-8af8-0b7c895c6166"
/>

src/sql/mysql/MySQLConnection.zig

@@ -71,7 +71,7 @@ pub fn canExecuteQuery(this: *@This()) bool {
     return this.queue.canExecuteQuery(this.getJSConnection());
 }
 
-pub inline fn isAbleToWrite(this: *@This()) bool {
+pub inline fn isAbleToWrite(this: *const @This()) bool {
     return this.status == .connected and
         !this.#flags.has_backpressure and
         this.#write_buffer.len() < MAX_PIPELINE_SIZE;
@@ -80,7 +80,7 @@ pub inline fn isAbleToWrite(this: *@This()) bool {
 pub inline fn isProcessingData(this: *@This()) bool {
     return this.#flags.is_processing_data;
 }
-pub inline fn hasBackpressure(this: *@This()) bool {
+pub inline fn hasBackpressure(this: *const @This()) bool {
     return this.#flags.has_backpressure;
 }
 pub inline fn resetBackpressure(this: *@This()) void {
@@ -99,19 +99,19 @@ pub const AuthState = union(enum) {
     };
 };
 
-pub fn canFlush(this: *@This()) bool {
+pub inline fn canFlush(this: *const @This()) bool {
     return !this.#flags.has_backpressure and // if has backpressure we need to wait for onWritable event
         this.status == .connected and //and we need to be connected
         // we need data to send
         (this.#write_buffer.len() > 0 or
-            if (this.queue.current()) |request| request.isPending() else false);
+            if (this.queue.current()) |request| request.isPending() and !request.isBeingPrepared() else false);
 }
 
-pub fn isIdle(this: *@This()) bool {
+pub inline fn isIdle(this: *const @This()) bool {
     return this.queue.current() == null and this.#write_buffer.len() == 0;
 }
 
-pub fn enqueueRequest(this: *@This(), request: *JSMySQLQuery) void {
+pub inline fn enqueueRequest(this: *@This(), request: *JSMySQLQuery) void {
     this.queue.add(request);
 }
 

src/sql/mysql/MySQLRequestQueue.zig

@@ -8,13 +8,13 @@ pub const MySQLRequestQueue = @This();
 #waiting_to_prepare: bool = false,
 #is_ready_for_query: bool = true,
 
-pub inline fn canExecuteQuery(this: *@This(), connection: *MySQLConnection) bool {
+pub inline fn canExecuteQuery(this: *const @This(), connection: *const MySQLConnection) bool {
     return connection.isAbleToWrite() and
         this.#is_ready_for_query and
         this.#nonpipelinable_requests == 0 and
         this.#pipelined_requests == 0;
 }
-pub inline fn canPrepareQuery(this: *@This(), connection: *MySQLConnection) bool {
+pub inline fn canPrepareQuery(this: *const @This(), connection: *const MySQLConnection) bool {
     return connection.isAbleToWrite() and
         this.#is_ready_for_query and
         !this.#waiting_to_prepare and
@@ -169,7 +169,7 @@ pub fn add(this: *@This(), request: *JSMySQLQuery) void {
     bun.handleOom(this.#requests.writeItem(request));
 }
 
-pub fn current(this: *@This()) ?*JSMySQLQuery {
+pub inline fn current(this: *const @This()) ?*JSMySQLQuery {
     if (this.#requests.readableLength() == 0) {
         return null;
     }

src/sql/mysql/js/JSMySQLConnection.zig

@@ -304,19 +304,19 @@ fn SocketHandler(comptime ssl: bool) type {
 }
 
 fn updateReferenceType(this: *@This()) void {
-    if (this.#js_value.isNotEmpty()) {
-        if (this.#connection.isActive()) {
-            if (this.#js_value == .weak) {
-                this.#js_value.upgrade(this.#globalObject);
-                this.#poll_ref.ref(this.#vm);
-            }
-            return;
-        }
-        if (this.#js_value == .strong) {
-            this.#js_value.downgrade();
-            this.#poll_ref.unref(this.#vm);
-            return;
+    if (this.#connection.isActive()) {
+        debug("connection is active", .{});
+        if (this.#js_value.isNotEmpty() and this.#js_value == .weak) {
+            debug("strong ref", .{});
+            this.#js_value.upgrade(this.#globalObject);
         }
+        this.#poll_ref.ref(this.#vm);
+        return;
+    }
+    debug("connection is not active", .{});
+    if (this.#js_value.isNotEmpty() and this.#js_value == .strong) {
+        debug("week ref", .{});
+        this.#js_value.downgrade();
     }
     this.#poll_ref.unref(this.#vm);
 }
@@ -589,10 +589,10 @@ pub fn getQueriesArray(this: *@This()) JSValue {
     return .js_undefined;
 }
 
-pub inline fn isAbleToWrite(this: *@This()) bool {
+pub inline fn isAbleToWrite(this: *const @This()) bool {
     return this.#connection.isAbleToWrite();
 }
-pub inline fn isConnected(this: *@This()) bool {
+pub inline fn isConnected(this: *const @This()) bool {
     return this.#connection.status == .connected;
 }
 pub inline fn canPipeline(this: *@This()) bool {
@@ -662,10 +662,7 @@ pub fn onConnectionEstabilished(this: *@This()) void {
     on_connect.ensureStillAlive();
     var js_value = this.#js_value.tryGet() orelse .js_undefined;
     js_value.ensureStillAlive();
-    // this.#globalObject.queueMicrotask(on_connect, &[_]JSValue{ JSValue.jsNull(), js_value });
-    const loop = this.#vm.eventLoop();
-    loop.runCallback(on_connect, this.#globalObject, .js_undefined, &[_]JSValue{ JSValue.jsNull(), js_value });
-    this.#poll_ref.unref(this.#vm);
+    this.#globalObject.queueMicrotask(on_connect, &[_]JSValue{ JSValue.jsNull(), js_value });
 }
 pub fn onQueryResult(this: *@This(), request: *JSMySQLQuery, result: MySQLQueryResult) void {
     request.resolve(this.getQueriesArray(), result);

test/harness.ts

@@ -863,7 +863,7 @@ export function isDockerEnabled(): boolean {
   }
 
   try {
-    const info = execSync(`${dockerCLI} info`, { stdio: ["ignore", "pipe", "inherit"] });
+    const info = execSync(`"${dockerCLI}" info`, { stdio: ["ignore", "pipe", "inherit"] });
     return info.toString().indexOf("Server Version:") !== -1;
   } catch {
     return false;
@@ -919,7 +919,7 @@ export async function describeWithContainer(
       return;
     }
     const { arch, platform } = process;
-    if ((archs && !archs?.includes(arch)) || platform === "win32") {
+    if (archs && !archs?.includes(arch)) {
       test.skip(`docker image is not supported on ${platform}/${arch}, skipped: ${image}`, () => {});
       return false;
     }

test/js/sql/sql-mysql.test.ts

@@ -34,6 +34,13 @@ if (docker) {
         MYSQL_ROOT_PASSWORD: "bun",
       },
     },
+    {
+      name: "MySQL 9",
+      image: "mysql:9",
+      env: {
+        MYSQL_ROOT_PASSWORD: "bun",
+      },
+    },
   ];
 
   for (const image of images) {