Fix handler race condition and MySQL caching_sha2_password auth

- Fix PgHandler and MyHandler race: schedule TryDeliver on the event
  loop thread when called from SetPendingResponse (fixes buffer
  corruption at high row counts with localhost latency)
- Fix MySQL caching_sha2_password fast-auth algorithm (H1 XOR H3 where
  H3 = SHA256(H2 || nonce), was incorrectly including password bytes)
- Implement MySQL caching_sha2_password full-auth (RSA OAEP-SHA1 path
  for cache-miss first connection, including 0x01 0x03 fast-auth-OK
  and 0x01 0x04 full-auth-required handling)
- Fix MySQL 8.4 compatibility: mysql_native_password removed, now use
  caching_sha2_password with proper auth switch handling
- Fix QueryJsonStreamAsync for MySQL: JSON_OBJECT() returns Blob type,
  decode as UTF-8 string instead of calling AsString() (which base64s)
- Fix JSON_OBJECT() SQL syntax: use comma syntax ('k', v) not VALUE kw
- Postgres: 6/6 tests pass; MySQL: 5/5 tests pass

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: vkuttyp

src/SqlDotnetty.MySql/MySqlConnection.cs

@@ -1,4 +1,5 @@
 using System.Net;
+using System.Security.Cryptography;
 using System.Text;
 using DotNetty.Buffers;
 using DotNetty.Transport.Bootstrapping;
@@ -63,10 +64,21 @@ internal sealed class MyHandler : ChannelHandlerAdapter
 {
     private readonly List<byte> _buffer = new();
     private TaskCompletionSource<byte[]>? _pendingTcs;
+    private IChannelHandlerContext? _ctx;
 
-    public void SetPendingResponse(TaskCompletionSource<byte[]> tcs)
-        => _pendingTcs = tcs;
+    public override void ChannelActive(IChannelHandlerContext ctx) { _ctx = ctx; base.ChannelActive(ctx); }
 
+    public void SetPendingResponse(TaskCompletionSource<byte[]> tcs)
+    {
+        _pendingTcs = tcs;
+        // Schedule TryDeliver on the event loop to avoid a race on _buffer.
+        var ctx = _ctx;
+        if (ctx != null)
+        {
+            if (ctx.Executor.InEventLoop) TryDeliver();
+            else ctx.Executor.Execute(TryDeliver);
+        }
+    }
     public override void ChannelRead(IChannelHandlerContext ctx, object msg)
     {
         if (msg is not IByteBuffer incoming)
@@ -189,6 +201,106 @@ public Task<SqlDataTable> QueryTableAsync(
             .ContinueWith(t => SqlDataTable.From(sql, t.Result),
                 TaskContinuationOptions.OnlyOnRanToCompletion);
 
+    // ── Streaming (IAsyncEnumerable) ──────────────────────────────────────────
+
+    /// <summary>
+    /// Execute a query and stream rows one at a time as they arrive from the server,
+    /// without buffering the full result set in memory.
+    /// </summary>
+    public async IAsyncEnumerable<SqlRow> QueryStreamAsync(
+        string sql,
+        IReadOnlyList<SqlParameter>? parameters = null,
+        [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken ct = default)
+    {
+        // Parameterized path: use prepared statement (buffers internally — MySQL binary protocol
+        // doesn't expose individual row packets to us here without a larger refactor).
+        if (parameters is { Count: > 0 })
+        {
+            var buffered = await QueryAsync(sql, parameters, ct).ConfigureAwait(false);
+            foreach (var row in buffered) yield return row;
+            yield break;
+        }
+
+        await _lock.WaitAsync(ct).ConfigureAwait(false);
+        try
+        {
+            _seqId = 0;
+            await WritePacketAsync(MyQueryMessage.Build(sql)).ConfigureAwait(false);
+
+            // Read result-set header (column count + column definitions).
+            var firstPayload = await ReceivePayloadAsync(ct).ConfigureAwait(false);
+            if (firstPayload[0] == 0x00) yield break;    // OK (DDL)
+            if (firstPayload[0] == 0xFF)
+            {
+                var e = MyDecoder.ParseErr(firstPayload);
+                throw SqlException.Query($"MySQL error [{e.ErrorCode}]: {e.ErrorMessage}");
+            }
+
+            int tmp = 0;
+            long colCount = MyDecoder.ReadLenEnc(firstPayload, out tmp);
+            var columns   = new List<MyColumnDef>((int)colCount);
+            for (int i = 0; i < colCount; i++)
+            {
+                var colPayload = await ReceivePayloadAsync(ct).ConfigureAwait(false);
+                columns.Add(MyDecoder.ParseColumnDef(colPayload));
+            }
+            var sqlColumns = columns.Select(c => new SqlColumn(c.Name, c.FieldType.ToString())).ToList();
+
+            // Stream rows.
+            while (true)
+            {
+                var rowPayload = await ReceivePayloadAsync(ct).ConfigureAwait(false);
+                if (rowPayload[0] == 0xFE && rowPayload.Length < 9) yield break;  // EOF
+                if (rowPayload[0] == 0xFF)
+                {
+                    var e = MyDecoder.ParseErr(rowPayload);
+                    throw SqlException.Query($"MySQL error [{e.ErrorCode}]: {e.ErrorMessage}");
+                }
+                if (rowPayload[0] == 0x00 && rowPayload.Length <= 7) yield break; // OK (deprecate EOF)
+
+                var values = ParseTextRow(rowPayload, columns);
+                yield return new SqlRow(sqlColumns, values);
+            }
+        }
+        finally { _lock.Release(); }
+    }
+
+    /// <summary>
+    /// Execute a query where each row contains a complete JSON object in
+    /// <paramref name="jsonColumnIndex"/> (e.g. <c>SELECT JSON_OBJECT(…) FROM …</c>)
+    /// and stream one <see cref="System.Text.Json.JsonElement"/> per row.
+    /// </summary>
+    /// <remarks>
+    /// Unlike SQL Server's <c>FOR JSON PATH</c>, MySQL's <c>JSON_OBJECT()</c> returns
+    /// one <b>complete</b> JSON object per row — no chunk reassembly is needed.  Each row
+    /// is parsed independently, so memory usage is bounded to the largest single object.
+    /// </remarks>
+    /// <example>
+    /// <code>
+    /// await foreach (var elem in conn.QueryJsonStreamAsync(
+    ///     "SELECT JSON_OBJECT('Id', Id, 'Name', Name) FROM Products"))
+    /// {
+    ///     var id = elem.GetProperty("Id").GetInt32();
+    /// }
+    /// </code>
+    /// </example>
+    public async IAsyncEnumerable<System.Text.Json.JsonElement> QueryJsonStreamAsync(
+        string sql,
+        IReadOnlyList<SqlParameter>? parameters = null,
+        int jsonColumnIndex = 0,
+        [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken ct = default)
+    {
+        await foreach (var row in QueryStreamAsync(sql, parameters, ct).ConfigureAwait(false))
+        {
+            string? json = row[jsonColumnIndex] is SqlValue.Bytes bytes
+                ? System.Text.Encoding.UTF8.GetString(bytes.Value)
+                : row[jsonColumnIndex].AsString();
+            if (string.IsNullOrEmpty(json)) continue;
+            using var doc = System.Text.Json.JsonDocument.Parse(json);
+            yield return doc.RootElement.Clone();
+        }
+    }
+
     public async Task BeginTransactionAsync(CancellationToken ct = default)
     {
         await _lock.WaitAsync(ct).ConfigureAwait(false);
@@ -310,7 +422,74 @@ private async Task ConnectAsync(CancellationToken ct)
             var err = MyDecoder.ParseErr(responseRaw);
             throw SqlException.Auth($"MySQL authentication error [{err.ErrorCode}]: {err.ErrorMessage}");
         }
-        if (responseRaw[0] != 0x00)
+        if (responseRaw[0] == 0xFE)
+        {
+            // AuthSwitchRequest: server wants us to use a different auth plugin
+            // Packet: 0xFE | plugin_name\0 | new_auth_data
+            int nameEnd = Array.IndexOf(responseRaw, (byte)0, 1);
+            if (nameEnd < 0) nameEnd = responseRaw.Length;
+            var switchPlugin = Encoding.UTF8.GetString(responseRaw, 1, nameEnd - 1);
+            var switchData   = nameEnd + 1 < responseRaw.Length
+                ? responseRaw[(nameEnd + 1)..]
+                : Array.Empty<byte>();
+
+            byte[] switchResponse = switchPlugin switch
+            {
+                "caching_sha2_password" => MyHandshakeResponse41.ComputeCachingSha2Auth(_config.Password, switchData),
+                _                       => MyHandshakeResponse41.ComputeNativePasswordAuth(_config.Password, switchData),
+            };
+
+            // seqId continues from the packet sequence (3rd packet sent = seqId 3)
+            await WritePacketAsync(switchResponse, seqId: 3).ConfigureAwait(false);
+
+            // Read response: may be fast-auth OK (0x01 0x03), full-auth required (0x01 0x04), or ERR
+            var finalRaw = await ReceivePayloadAsync(ct).ConfigureAwait(false);
+            if (finalRaw[0] == 0xFF)
+            {
+                var err = MyDecoder.ParseErr(finalRaw);
+                throw SqlException.Auth($"MySQL auth switch error [{err.ErrorCode}]: {err.ErrorMessage}");
+            }
+            // 0x01 0x03 = fast auth success — read following OK
+            if (finalRaw[0] == 0x01 && finalRaw.Length > 1 && finalRaw[1] == 0x03)
+            {
+                finalRaw = await ReceivePayloadAsync(ct).ConfigureAwait(false);
+                if (finalRaw[0] == 0xFF)
+                {
+                    var err2 = MyDecoder.ParseErr(finalRaw);
+                    throw SqlException.Auth($"MySQL auth fast-auth OK error [{err2.ErrorCode}]: {err2.ErrorMessage}");
+                }
+            }
+            // 0x01 0x04 = full auth required — exchange RSA public key and encrypt password
+            else if (finalRaw[0] == 0x01 && finalRaw.Length > 1 && finalRaw[1] == 0x04)
+            {
+                // Request server's RSA public key
+                await WritePacketAsync(new byte[] { 0x02 }, seqId: 5).ConfigureAwait(false);
+                var pkPacket = await ReceivePayloadAsync(ct).ConfigureAwait(false);
+                // pkPacket[0] == 0x01, rest is PEM public key
+                int pemStart = pkPacket[0] == 0x01 ? 1 : 0;
+                var pem = Encoding.UTF8.GetString(pkPacket, pemStart, pkPacket.Length - pemStart).Trim();
+                // XOR password bytes (null-terminated) with nonce (cyclic)
+                var pwBytes  = Encoding.UTF8.GetBytes(_config.Password + "\0");
+                var nonce    = switchData;
+                var obfuscated = new byte[pwBytes.Length];
+                for (int i = 0; i < pwBytes.Length; i++)
+                    obfuscated[i] = (byte)(pwBytes[i] ^ nonce[i % nonce.Length]);
+                // RSA-OAEP-SHA1 encrypt
+                using var rsa = RSA.Create();
+                rsa.ImportFromPem(pem);
+                var encrypted = rsa.Encrypt(obfuscated, RSAEncryptionPadding.OaepSHA1);
+                await WritePacketAsync(encrypted, seqId: 7).ConfigureAwait(false);
+                finalRaw = await ReceivePayloadAsync(ct).ConfigureAwait(false);
+                if (finalRaw[0] == 0xFF)
+                {
+                    var err3 = MyDecoder.ParseErr(finalRaw);
+                    throw SqlException.Auth($"MySQL RSA auth error [{err3.ErrorCode}]: {err3.ErrorMessage}");
+                }
+            }
+            if (finalRaw[0] != 0x00)
+                throw SqlException.Auth($"MySQL: unexpected post-switch response byte 0x{finalRaw[0]:X2}.");
+        }
+        else if (responseRaw[0] != 0x00)
         {
             // Could be auth switch request or other — for simplicity, fail
             throw SqlException.Auth($"MySQL: unexpected authentication response byte 0x{responseRaw[0]:X2}.");

src/SqlDotnetty.MySql/MySqlConnectionPool.cs

@@ -122,6 +122,35 @@ public async Task RollbackAsync(CancellationToken ct = default)
         finally { await ReleaseAsync(conn).ConfigureAwait(false); }
     }
 
+    public async IAsyncEnumerable<SqlRow> QueryStreamAsync(
+        string sql,
+        IReadOnlyList<SqlParameter>? parameters = null,
+        [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken ct = default)
+    {
+        var conn = await AcquireAsync(ct).ConfigureAwait(false);
+        try
+        {
+            await foreach (var row in conn.QueryStreamAsync(sql, parameters, ct).ConfigureAwait(false))
+                yield return row;
+        }
+        finally { await ReleaseAsync(conn).ConfigureAwait(false); }
+    }
+
+    public async IAsyncEnumerable<System.Text.Json.JsonElement> QueryJsonStreamAsync(
+        string sql,
+        IReadOnlyList<SqlParameter>? parameters = null,
+        int jsonColumnIndex = 0,
+        [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken ct = default)
+    {
+        var conn = await AcquireAsync(ct).ConfigureAwait(false);
+        try
+        {
+            await foreach (var elem in conn.QueryJsonStreamAsync(sql, parameters, jsonColumnIndex, ct).ConfigureAwait(false))
+                yield return elem;
+        }
+        finally { await ReleaseAsync(conn).ConfigureAwait(false); }
+    }
+
     public Task CloseAsync() => DisposeAsync().AsTask();
 
     public async ValueTask DisposeAsync()

src/SqlDotnetty.MySql/Proto/MyMessage.cs

@@ -124,7 +124,7 @@ public static byte[] Build(
         return ms.ToArray();
     }
 
-    private static byte[] ComputeNativePasswordAuth(string password, byte[] scramble)
+    internal static byte[] ComputeNativePasswordAuth(string password, byte[] scramble)
     {
         if (string.IsNullOrEmpty(password)) return Array.Empty<byte>();
 
@@ -144,25 +144,24 @@ private static byte[] ComputeNativePasswordAuth(string password, byte[] scramble
         return result;
     }
 
-    private static byte[] ComputeCachingSha2Auth(string password, byte[] scramble)
+    internal static byte[] ComputeCachingSha2Auth(string password, byte[] scramble)
     {
         if (string.IsNullOrEmpty(password)) return Array.Empty<byte>();
 
         var pwBytes = Encoding.UTF8.GetBytes(password);
-        // SHA256(password)
-        var sha256pw  = SHA256.HashData(pwBytes);
-        // SHA256(SHA256(password))
-        var sha256pw2 = SHA256.HashData(sha256pw);
-        // SHA256(password + scramble + SHA256(SHA256(password)))
-        var combined = new byte[pwBytes.Length + scramble.Length + sha256pw2.Length];
-        pwBytes.CopyTo(combined, 0);
-        scramble.CopyTo(combined, pwBytes.Length);
-        sha256pw2.CopyTo(combined, pwBytes.Length + scramble.Length);
-        var hash = SHA256.HashData(combined);
-        // XOR with SHA256(password)
-        var result = new byte[hash.Length];
-        for (int i = 0; i < hash.Length; i++)
-            result[i] = (byte)(hash[i] ^ sha256pw[i]);
+        // H1 = SHA256(password)
+        var h1 = SHA256.HashData(pwBytes);
+        // H2 = SHA256(SHA256(password))
+        var h2 = SHA256.HashData(h1);
+        // H3 = SHA256(H2 || nonce)
+        var combined = new byte[h2.Length + scramble.Length];
+        h2.CopyTo(combined, 0);
+        scramble.CopyTo(combined, h2.Length);
+        var h3 = SHA256.HashData(combined);
+        // token = H1 XOR H3
+        var result = new byte[h1.Length];
+        for (int i = 0; i < h1.Length; i++)
+            result[i] = (byte)(h1[i] ^ h3[i]);
         return result;
     }
 }