fix RESP3 handshakes

Author: mgravell

src/StackExchange.Redis/ResultProcessor.cs

@@ -2892,12 +2892,20 @@ public override bool SetResult(PhysicalConnection connection, Message message, i
                     }
                 }
 
+                // We need to SetResult before OnFullyEstablished so that the
+                // protocol is reliably known *before* we do next-steps.
                 if (connection.Protocol is null)
                 {
                     // if we didn't get a valid response from HELLO, then we have to assume RESP2 at some point
                     connection.SetProtocol(RedisProtocol.Resp2);
                 }
 
+                if (final & establishConnection)
+                {
+                    // This is what ultimately brings us to complete a connection, by advancing the state forward from a successful tracer after connection.
+                    connection.BridgeCouldBeNull?.OnFullyEstablished(connection, $"From command: {message.Command}");
+                }
+
                 return final;
             }
 
@@ -2939,11 +2947,6 @@ protected override bool SetResultCore(PhysicalConnection connection, Message mes
                 }
                 if (happy)
                 {
-                    if (establishConnection)
-                    {
-                        // This is what ultimately brings us to complete a connection, by advancing the state forward from a successful tracer after connection.
-                        connection.BridgeCouldBeNull?.OnFullyEstablished(connection, $"From command: {message.Command}");
-                    }
                     SetResult(message, happy);
                     return true;
                 }

src/StackExchange.Redis/ServerEndPoint.cs

@@ -695,14 +695,19 @@ internal void OnFullyEstablished(PhysicalConnection connection, string source)
                     // Clear the unselectable flag ASAP since we are open for business
                     ClearUnselectable(UnselectableFlags.DidNotRespond);
 
-                    bool isResp3 = KnowOrAssumeResp3();
+                    bool isResp3 = connection?.Protocol is >= RedisProtocol.Resp3;
                     if (bridge == subscription || isResp3)
                     {
                         // Note: this MUST be fire and forget, because we might be in the middle of a Sync processing
                         // TracerProcessor which is executing this line inside a SetResultCore().
                         // Since we're issuing commands inside a SetResult path in a message, we'd create a deadlock by waiting.
                         Multiplexer.EnsureSubscriptions(CommandFlags.FireAndForget);
                     }
+                    else if (SupportsSubscriptions && Multiplexer.RawConfig.Protocol > RedisProtocol.Resp2)
+                    {
+                        // interactive, and we wanted RESP3+, but we didn't get it; spin up pub/sub
+                        Activate(ConnectionType.Subscription, null);
+                    }
                     if (IsConnected && (IsSubscriberConnected || !SupportsSubscriptions || isResp3))
                     {
                         // Only connect on the second leg - we can accomplish this by checking both

tests/StackExchange.Redis.Tests/InProcessTestServer.cs

@@ -17,7 +17,8 @@ namespace StackExchange.Redis.Tests;
 public class InProcessTestServer : MemoryCacheRedisServer
 {
     private readonly ITestOutputHelper? _log;
-    public InProcessTestServer(ITestOutputHelper? log = null)
+    public InProcessTestServer(ITestOutputHelper? log = null, EndPoint? endpoint = null)
+        : base(endpoint)
     {
         RedisVersion = RedisFeatures.v6_0_0; // for client to expect RESP3
         _log = log;

tests/StackExchange.Redis.Tests/Resp3HandshakeTests.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Collections.Concurrent;
 using System.Collections.Generic;
+using System.Net;
 using System.Threading.Tasks;
 using StackExchange.Redis.Server;
 using Xunit;
@@ -99,8 +100,9 @@ public async Task Handshake(RedisProtocol client, ServerResponse server, Handsha
         }
     }
 
+    private static readonly EndPoint EP = new DnsEndPoint("home", 8000);
     private sealed class HandshakeServer(ServerResponse response, ITestOutputHelper log)
-        : InProcessTestServer(log)
+        : InProcessTestServer(log, EP)
     {
         protected override RedisProtocol MaxProtocol => response switch
         {