add test infrastructure for rich in-proc dummy servers (#3021)

* - add test infrastructure for rich in-proc dummy servers, for simulating connection state
- migrate the -MOVED/self test to the new infrastructure, and use better server/cluster/etc impl

* log client id

Author: mgravell

tests/StackExchange.Redis.Tests/InProcessTestServer.cs

@@ -0,0 +1,87 @@
+using System;
+using System.IO;
+using System.IO.Pipelines;
+using System.Net;
+using System.Net.Sockets;
+using System.Threading;
+using System.Threading.Tasks;
+using Pipelines.Sockets.Unofficial;
+using StackExchange.Redis.Configuration;
+using StackExchange.Redis.Server;
+using Xunit;
+
+namespace StackExchange.Redis.Tests;
+
+public class InProcessTestServer : MemoryCacheRedisServer
+{
+    public Tunnel Tunnel { get; }
+
+    private readonly ITestOutputHelper? _log;
+    public InProcessTestServer(ITestOutputHelper? log = null)
+    {
+        _log = log;
+        // ReSharper disable once VirtualMemberCallInConstructor
+        _log?.WriteLine($"Creating in-process server: {ToString()}");
+        Tunnel = new InProcTunnel(this);
+    }
+
+    private sealed class InProcTunnel(
+        InProcessTestServer server,
+        PipeOptions? pipeOptions = null) : Tunnel
+    {
+        public override ValueTask<EndPoint?> GetSocketConnectEndpointAsync(
+            EndPoint endpoint,
+            CancellationToken cancellationToken)
+        {
+            // server._log?.WriteLine($"Disabling client creation, requested endpoint: {Format.ToString(endpoint)}");
+            return default;
+        }
+
+        public override ValueTask<Stream?> BeforeAuthenticateAsync(
+            EndPoint endpoint,
+            ConnectionType connectionType,
+            Socket? socket,
+            CancellationToken cancellationToken)
+        {
+            server._log?.WriteLine($"Client intercepted, requested endpoint: {Format.ToString(endpoint)} for {connectionType} usage");
+            var clientToServer = new Pipe(pipeOptions ?? PipeOptions.Default);
+            var serverToClient = new Pipe(pipeOptions ?? PipeOptions.Default);
+            var serverSide = new Duplex(clientToServer.Reader, serverToClient.Writer);
+            _ = Task.Run(async () => await server.RunClientAsync(serverSide), cancellationToken);
+            var clientSide = StreamConnection.GetDuplex(serverToClient.Reader, clientToServer.Writer);
+            return new(clientSide);
+        }
+
+        private sealed class Duplex(PipeReader input, PipeWriter output) : IDuplexPipe
+        {
+            public PipeReader Input => input;
+            public PipeWriter Output => output;
+
+            public ValueTask Dispose()
+            {
+                input.Complete();
+                output.Complete();
+                return default;
+            }
+        }
+    }
+    /*
+
+    private readonly RespServer _server;
+    public RespSocketServer(RespServer server)
+    {
+        _server = server ?? throw new ArgumentNullException(nameof(server));
+        server.Shutdown.ContinueWith((_, o) => ((SocketServer)o).Dispose(), this);
+    }
+    protected override void OnStarted(EndPoint endPoint)
+        => _server.Log("Server is listening on " + endPoint);
+
+    protected override Task OnClientConnectedAsync(in ClientConnection client)
+        => _server.RunClientAsync(client.Transport);
+
+    protected override void Dispose(bool disposing)
+    {
+        if (disposing) _server.Dispose();
+    }
+    */
+}

tests/StackExchange.Redis.Tests/MovedTestServer.cs

@@ -2,9 +2,12 @@
 using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Net;
+using System.Net.Sockets;
+using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
 using StackExchange.Redis.Server;
+using Xunit;
 
 namespace StackExchange.Redis.Tests;
 
@@ -14,7 +17,7 @@ namespace StackExchange.Redis.Tests;
 /// When a MOVED error points to the same endpoint, it signals the client to reconnect before retrying the command,
 /// allowing the DNS record/proxy/load balancer to route the connection to a different underlying server host.
 /// </summary>
-public class MovedTestServer : MemoryCacheRedisServer
+public class MovedTestServer : InProcessTestServer
 {
     /// <summary>
     /// Represents the simulated server host state behind a proxy/load balancer.
@@ -34,19 +37,26 @@ private enum SimulatedHost
 
     private int _setCmdCount = 0;
     private int _movedResponseCount = 0;
-    private int _connectionCount = 0;
+
     private SimulatedHost _currentServerHost = SimulatedHost.OldServer;
-    private readonly ConcurrentDictionary<RedisClient, SimulatedHost> _clientHostAssignments = new();
+
     private readonly Func<string> _getEndpoint;
     private readonly string _triggerKey;
     private readonly int _hashSlot;
     private EndPoint? _actualEndpoint;
 
-    public MovedTestServer(Func<string> getEndpoint, string triggerKey = "testkey", int hashSlot = 12345)
+    public MovedTestServer(Func<string> getEndpoint, string triggerKey = "testkey", int hashSlot = 12345, ITestOutputHelper? log = null) : base(log)
     {
         _getEndpoint = getEndpoint;
         _triggerKey = triggerKey;
         _hashSlot = hashSlot;
+        ServerType = ServerType.Cluster;
+        RedisVersion = RedisFeatures.v7_2_0_rc1;
+    }
+
+    private sealed class MovedTestClient(SimulatedHost assignedHost) : RedisClient
+    {
+        public SimulatedHost AssignedHost => assignedHost;
     }
 
     /// <summary>
@@ -55,32 +65,11 @@ public MovedTestServer(Func<string> getEndpoint, string triggerKey = "testkey",
     /// </summary>
     public override RedisClient CreateClient()
     {
-        var client = base.CreateClient();
-        var assignedHost = _currentServerHost;
-        _clientHostAssignments[client] = assignedHost;
-        Interlocked.Increment(ref _connectionCount);
-        Log($"New client connection established (assigned to {assignedHost}, total connections: {_connectionCount}), endpoint: {_actualEndpoint}");
+        var client = new MovedTestClient(_currentServerHost);
+        Log($"New client connection established (assigned to {client.AssignedHost}, total connections: {TotalClientCount}), endpoint: {_actualEndpoint}");
         return client;
     }
 
-    /// <summary>
-    /// Handles the INFO command, reporting cluster mode as enabled.
-    /// </summary>
-    protected override TypedRedisValue Info(RedisClient client, RedisRequest request)
-    {
-        // Override INFO to report cluster mode enabled
-        var section = request.Count >= 2 ? request.GetString(1) : null;
-
-        // Return cluster-enabled info
-        var infoResponse = section?.Equals("CLUSTER", StringComparison.OrdinalIgnoreCase) == true
-            ? "# Cluster\r\ncluster_enabled:1\r\n"
-            : "# Server\r\nredis_version:7.0.0\r\nredis_mode:cluster\r\n# Cluster\r\ncluster_enabled:1\r\n";
-
-        Log($"Returning INFO response (cluster_enabled:1), endpoint: {_actualEndpoint}");
-
-        return TypedRedisValue.BulkString(infoResponse);
-    }
-
     /// <summary>
     /// Handles CLUSTER commands, supporting SLOTS and NODES subcommands for cluster mode simulation.
     /// </summary>
@@ -122,10 +111,11 @@ protected override TypedRedisValue Set(RedisClient client, RedisRequest request)
         Interlocked.Increment(ref _setCmdCount);
 
         // Get the client's assigned server host
-        if (!_clientHostAssignments.TryGetValue(client, out var clientHost))
+        if (client is not MovedTestClient movedClient)
         {
-            throw new InvalidOperationException("Client host assignment not found - this indicates a test infrastructure error");
+            throw new InvalidOperationException($"Client is not a {nameof(MovedTestClient)}");
         }
+        var clientHost = movedClient.AssignedHost;
 
         // Check if this is the trigger key from an old server client
         if (key == _triggerKey && clientHost == SimulatedHost.OldServer)
@@ -213,11 +203,6 @@ private TypedRedisValue GetClusterNodesResponse()
     /// </summary>
     public int MovedResponseCount => _movedResponseCount;
 
-    /// <summary>
-    /// Gets the number of client connections established.
-    /// </summary>
-    public int ConnectionCount => _connectionCount;
-
     /// <summary>
     /// Gets the actual endpoint the server is listening on.
     /// </summary>
@@ -236,10 +221,10 @@ public void SetActualEndpoint(EndPoint endPoint)
     /// <summary>
     /// Resets all counters for test reusability.
     /// </summary>
-    public void ResetCounters()
+    public override void ResetCounters()
     {
         Interlocked.Exchange(ref _setCmdCount, 0);
         Interlocked.Exchange(ref _movedResponseCount, 0);
-        Interlocked.Exchange(ref _connectionCount, 0);
+        base.ResetCounters();
     }
 }

tests/StackExchange.Redis.Tests/MovedToSameEndpointTests.cs

@@ -1,10 +1,6 @@
-using System;
 using System.Net;
-using System.Net.Sockets;
 using System.Threading.Tasks;
-using StackExchange.Redis.Server;
 using Xunit;
-using Xunit.Sdk;
 
 namespace StackExchange.Redis.Tests;
 
@@ -15,18 +11,6 @@ namespace StackExchange.Redis.Tests;
 /// </summary>
 public class MovedToSameEndpointTests(ITestOutputHelper log)
 {
-    /// <summary>
-    /// Gets a free port by temporarily binding to port 0 and retrieving the OS-assigned port.
-    /// </summary>
-    private static int GetFreePort()
-    {
-        var listener = new TcpListener(IPAddress.Loopback, 0);
-        listener.Start();
-        var port = ((IPEndPoint)listener.LocalEndpoint).Port;
-        listener.Stop();
-        return port;
-    }
-
     /// <summary>
     /// Integration test: Verifies that when a MOVED error points to the same endpoint,
     /// the client reconnects and successfully retries the operation.
@@ -49,73 +33,67 @@ private static int GetFreePort()
     public async Task MovedToSameEndpoint_TriggersReconnectAndRetry_CommandSucceeds()
     {
         var keyName = "MovedToSameEndpoint_TriggersReconnectAndRetry_CommandSucceeds";
-        // Arrange: Get a free port to avoid conflicts when tests run in parallel
-        var port = GetFreePort();
-        var listenEndpoint = new IPEndPoint(IPAddress.Loopback, port);
 
-        var testServer = new MovedTestServer(
+        var listenEndpoint = new IPEndPoint(IPAddress.Loopback, 6382);
+        using var testServer = new MovedTestServer(
             getEndpoint: () => Format.ToString(listenEndpoint),
-            triggerKey: keyName);
+            triggerKey: keyName,
+            log: log);
 
-        var socketServer = new RespSocketServer(testServer);
+        testServer.SetActualEndpoint(listenEndpoint);
 
-        try
-        {
-            // Start listening on the free port
-            socketServer.Listen(listenEndpoint);
-            testServer.SetActualEndpoint(listenEndpoint);
+        // Wait a moment for the server to fully start
+        await Task.Delay(100);
 
-            // Wait a moment for the server to fully start
-            await Task.Delay(100);
+        // Act: Connect to the test server
+        var config = new ConfigurationOptions
+        {
+            EndPoints = { listenEndpoint },
+            ConnectTimeout = 10000,
+            SyncTimeout = 5000,
+            AsyncTimeout = 5000,
+            AllowAdmin = true,
+            Tunnel = testServer.Tunnel,
+        };
 
-            // Act: Connect to the test server
-            var config = new ConfigurationOptions
-            {
-                EndPoints = { listenEndpoint },
-                ConnectTimeout = 10000,
-                SyncTimeout = 5000,
-                AsyncTimeout = 5000,
-                AllowAdmin = true,
-            };
+        await using var conn = await ConnectionMultiplexer.ConnectAsync(config);
+        // Ping the server to ensure it's responsive
+        var server = conn.GetServer(listenEndpoint);
+        log?.WriteLine((await server.InfoRawAsync()) ?? "");
+        var id = await server.ExecuteAsync("client", "id");
+        log?.WriteLine($"client id: {id}");
 
-            await using var conn = await ConnectionMultiplexer.ConnectAsync(config);
-            // Ping the server to ensure it's responsive
-            var server = conn.GetServer(listenEndpoint);
-            log?.WriteLine($"info: {await server.InfoRawAsync()}");
-            await server.PingAsync();
-            // Verify server is detected as cluster mode
-            Assert.Equal(ServerType.Cluster, server.ServerType);
-            var db = conn.GetDatabase();
+        await server.PingAsync();
+        // Verify server is detected as cluster mode
+        Assert.Equal(ServerType.Cluster, server.ServerType);
+        var db = conn.GetDatabase();
 
-            // Record baseline counters after initial connection
-            var initialSetCmdCount = testServer.SetCmdCount;
-            var initialMovedResponseCount = testServer.MovedResponseCount;
-            var initialConnectionCount = testServer.ConnectionCount;
-            // Execute SET command: This should receive MOVED → reconnect → retry → succeed
-            var setResult = await db.StringSetAsync(keyName, "testvalue");
+        // Record baseline counters after initial connection
+        var initialSetCmdCount = testServer.SetCmdCount;
+        var initialMovedResponseCount = testServer.MovedResponseCount;
+        var initialConnectionCount = testServer.TotalClientCount;
+        // Execute SET command: This should receive MOVED → reconnect → retry → succeed
+        var setResult = await db.StringSetAsync(keyName, "testvalue");
 
-            // Assert: Verify SET command succeeded
-            Assert.True(setResult, "SET command should return true (OK)");
+        // Assert: Verify SET command succeeded
+        Assert.True(setResult, "SET command should return true (OK)");
 
-            // Verify the value was actually stored (proving retry succeeded)
-            var retrievedValue = await db.StringGetAsync(keyName);
-            Assert.Equal("testvalue", (string?)retrievedValue);
+        // Verify the value was actually stored (proving retry succeeded)
+        var retrievedValue = await db.StringGetAsync(keyName);
+        Assert.Equal("testvalue", (string?)retrievedValue);
 
-            // Verify SET command was executed twice: once with MOVED response, once successfully
-            var expectedSetCmdCount = initialSetCmdCount + 2;
-            Assert.Equal(expectedSetCmdCount, testServer.SetCmdCount);
+        // Verify SET command was executed twice: once with MOVED response, once successfully
+        var expectedSetCmdCount = initialSetCmdCount + 2;
+        Assert.Equal(expectedSetCmdCount, testServer.SetCmdCount);
 
-            // Verify MOVED response was returned exactly once
-            var expectedMovedResponseCount = initialMovedResponseCount + 1;
-            Assert.Equal(expectedMovedResponseCount, testServer.MovedResponseCount);
+        // Verify MOVED response was returned exactly once
+        var expectedMovedResponseCount = initialMovedResponseCount + 1;
+        Assert.Equal(expectedMovedResponseCount, testServer.MovedResponseCount);
 
-            // Verify reconnection occurred: connection count should have increased by 1
-            var expectedConnectionCount = initialConnectionCount + 1;
-            Assert.Equal(expectedConnectionCount, testServer.ConnectionCount);
-        }
-        finally
-        {
-            socketServer?.Dispose();
-        }
+        // Verify reconnection occurred: connection count should have increased by 1
+        var expectedConnectionCount = initialConnectionCount + 1;
+        Assert.Equal(expectedConnectionCount, testServer.TotalClientCount);
+        id = await server.ExecuteAsync("client", "id");
+        log?.WriteLine($"client id: {id}");
     }
 }

toys/StackExchange.Redis.Server/RedisClient.cs

@@ -4,7 +4,7 @@
 
 namespace StackExchange.Redis.Server
 {
-    public sealed class RedisClient : IDisposable
+    public class RedisClient : IDisposable
     {
         internal int SkipReplies { get; set; }
         internal bool ShouldSkipResponse()