More powerful virtual test server for unit tests (#3022)

* wip

* Multi-node test infrastructure

* respect CROSSSLOT in the toy server

* extensively use "in" in the toy server; lots of stack values

* implement decrby (for desting from CI)

* implement basic TTL to support basic-ops test in primary tests

* basic resp3

* - simplify connecting to the test server
- include all commands and endpoints when connecting to the test server
- better RESP3 aggregate support

* run basic tests on RESP2+3

* Fix remaining RESP3 handshake snafus

* test fixture for in-proc; run with basic tests

* implement everything in multi/exec except the serialize/deserialize (d'oh!)

* also force attributes to be pairs (same semantics as maps)

* don't buffer any of the transaction commands

Author: mgravell

src/StackExchange.Redis/ClusterConfiguration.cs

@@ -45,6 +45,8 @@ private SlotRange(short from, short to)
         /// </summary>
         public int To => to;
 
+        internal bool IsSingleSlot => From == To;
+
         internal const int MinSlot = 0, MaxSlot = 16383;
 
         private static SlotRange[]? s_SharedAllSlots;

src/StackExchange.Redis/ConnectionMultiplexer.cs

@@ -356,7 +356,7 @@ internal void CheckMessage(Message message)
             }
         }
 
-        internal bool TryResend(int hashSlot, Message message, EndPoint endpoint, bool isMoved)
+        internal bool TryResend(int hashSlot, Message message, EndPoint endpoint, bool isMoved, bool isSelf)
         {
             // If we're being told to re-send something because the hash slot moved, that means our topology is out of date
             // ...and we should re-evaluate what's what.
@@ -367,7 +367,7 @@ internal bool TryResend(int hashSlot, Message message, EndPoint endpoint, bool i
                 ReconfigureIfNeeded(endpoint, false, "MOVED encountered");
             }
 
-            return ServerSelectionStrategy.TryResend(hashSlot, message, endpoint, isMoved);
+            return ServerSelectionStrategy.TryResend(hashSlot, message, endpoint, isMoved, isSelf);
         }
 
         /// <summary>

src/StackExchange.Redis/PhysicalConnection.cs

@@ -925,6 +925,30 @@ internal static void WriteMultiBulkHeader(PipeWriter output, long count)
             output.Advance(offset);
         }
 
+        internal static void WriteMultiBulkHeader(PipeWriter output, long count, ResultType type)
+        {
+            // *{count}\r\n         = 3 + MaxInt32TextLen
+            var span = output.GetSpan(3 + Format.MaxInt32TextLen);
+            span[0] = type switch
+            {
+                ResultType.Push => (byte)'>',
+                ResultType.Attribute => (byte)'|',
+                ResultType.Map => (byte)'%',
+                ResultType.Set => (byte)'~',
+                 _ => (byte)'*',
+            };
+            if ((type is ResultType.Map or ResultType.Attribute) & count > 0)
+            {
+                if ((count & 1) != 0) Throw(type, count);
+                count >>= 1;
+                static void Throw(ResultType type, long count) => throw new ArgumentOutOfRangeException(
+                    paramName: nameof(count),
+                    message: $"{type} data must be in pairs; got {count}");
+            }
+            int offset = WriteRaw(span, count, offset: 1);
+            output.Advance(offset);
+        }
+
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
         internal static int WriteCrlf(Span<byte> span, int offset)
         {

src/StackExchange.Redis/ResultProcessor.cs

@@ -274,7 +274,7 @@ public virtual bool SetResult(PhysicalConnection connection, Message message, in
                         {
                             // already toast
                         }
-                        else if (bridge.Multiplexer.TryResend(hashSlot, message, endpoint, isMoved))
+                        else if (bridge.Multiplexer.TryResend(hashSlot, message, endpoint, isMoved, isSameEndpoint))
                         {
                             bridge.Multiplexer.Trace(message.Command + " re-issued to " + endpoint, isMoved ? "MOVED" : "ASK");
                             return false;

src/StackExchange.Redis/ServerSelectionStrategy.cs

@@ -46,7 +46,7 @@ internal sealed class ServerSelectionStrategy
             0x6e17, 0x7e36, 0x4e55, 0x5e74, 0x2e93, 0x3eb2, 0x0ed1, 0x1ef0,
         };
 
-        private readonly ConnectionMultiplexer multiplexer;
+        private readonly ConnectionMultiplexer? multiplexer;
         private int anyStartOffset = SharedRandom.Next(); // initialize to a random value so routing isn't uniform
 
         #if NET6_0_OR_GREATER
@@ -57,7 +57,7 @@ internal sealed class ServerSelectionStrategy
 
         private ServerEndPoint[]? map;
 
-        public ServerSelectionStrategy(ConnectionMultiplexer multiplexer) => this.multiplexer = multiplexer;
+        public ServerSelectionStrategy(ConnectionMultiplexer? multiplexer) => this.multiplexer = multiplexer;
 
         public ServerType ServerType { get; set; } = ServerType.Standalone;
         internal static int TotalSlots => RedisClusterSlotCount;
@@ -96,6 +96,8 @@ public int HashSlot(in RedisKey key)
             }
         }
 
+        private byte[] ChannelPrefix => multiplexer?.ChannelPrefix ?? [];
+
         /// <summary>
         /// Computes the hash-slot that would be used by the given channel.
         /// </summary>
@@ -106,7 +108,7 @@ public int HashSlot(in RedisChannel channel)
 
             ReadOnlySpan<byte> routingSpan = channel.RoutingSpan;
             byte[] prefix;
-            return channel.IgnoreChannelPrefix || (prefix = multiplexer.ChannelPrefix).Length == 0
+            return channel.IgnoreChannelPrefix || (prefix = ChannelPrefix).Length == 0
                 ? GetClusterSlot(routingSpan) : GetClusterSlotWithPrefix(prefix, routingSpan);
 
             static int GetClusterSlotWithPrefix(byte[] prefixRaw, ReadOnlySpan<byte> routingSpan)
@@ -133,15 +135,15 @@ static int GetClusterSlotWithPrefix(byte[] prefixRaw, ReadOnlySpan<byte> routing
         /// <remarks>
         /// HASH_SLOT = CRC16(key) mod 16384.
         /// </remarks>
-        private static unsafe int GetClusterSlot(ReadOnlySpan<byte> blob)
+        internal static unsafe int GetClusterSlot(ReadOnlySpan<byte> key)
         {
             unchecked
             {
-                fixed (byte* ptr = blob)
+                fixed (byte* ptr = key)
                 {
                     fixed (ushort* crc16tab = ServerSelectionStrategy.Crc16tab)
                     {
-                        int offset = 0, count = blob.Length, start, end;
+                        int offset = 0, count = key.Length, start, end;
                         if ((start = IndexOf(ptr, (byte)'{', 0, count - 1)) >= 0
                             && (end = IndexOf(ptr, (byte)'}', start + 1, count)) >= 0
                             && --end != start)
@@ -169,7 +171,7 @@ private static unsafe int GetClusterSlot(ReadOnlySpan<byte> blob)
                 // the same, so this does a pretty good job of spotting illegal commands before sending them
                 case ServerType.Twemproxy:
                     slot = message.GetHashSlot(this);
-                    if (slot == MultipleSlots) throw ExceptionFactory.MultiSlot(multiplexer.RawConfig.IncludeDetailInExceptions, message);
+                    if (slot == MultipleSlots) throw ExceptionFactory.MultiSlot(multiplexer?.RawConfig?.IncludeDetailInExceptions ?? false, message);
                     break;
                 /* just shown for completeness
                 case ServerType.Standalone: // don't use sharding
@@ -193,13 +195,13 @@ private static unsafe int GetClusterSlot(ReadOnlySpan<byte> blob)
             return Select(slot, command, flags, allowDisconnected);
         }
 
-        public bool TryResend(int hashSlot, Message message, EndPoint endpoint, bool isMoved)
+        public bool TryResend(int hashSlot, Message message, EndPoint endpoint, bool isMoved, bool isSelf)
         {
             try
             {
-                if (ServerType == ServerType.Standalone || hashSlot < 0 || hashSlot >= RedisClusterSlotCount) return false;
+                if ((ServerType == ServerType.Standalone && !isSelf) || hashSlot < 0 || hashSlot >= RedisClusterSlotCount) return false;
 
-                ServerEndPoint server = multiplexer.GetServerEndPoint(endpoint);
+                ServerEndPoint? server = multiplexer?.GetServerEndPoint(endpoint);
                 if (server != null)
                 {
                     bool retry = false;
@@ -230,7 +232,7 @@ public bool TryResend(int hashSlot, Message message, EndPoint endpoint, bool isM
                         }
                         if (resendVia == null)
                         {
-                            multiplexer.Trace("Unable to resend to " + endpoint);
+                            multiplexer?.Trace("Unable to resend to " + endpoint);
                         }
                         else
                         {
@@ -248,7 +250,7 @@ public bool TryResend(int hashSlot, Message message, EndPoint endpoint, bool isM
                         arr[hashSlot] = server;
                         if (oldServer != server)
                         {
-                            multiplexer.OnHashSlotMoved(hashSlot, oldServer?.EndPoint, endpoint);
+                            multiplexer?.OnHashSlotMoved(hashSlot, oldServer?.EndPoint, endpoint);
                         }
                     }
 
@@ -305,7 +307,7 @@ private static unsafe int IndexOf(byte* ptr, byte value, int start, int end)
         }
 
         private ServerEndPoint? Any(RedisCommand command, CommandFlags flags, bool allowDisconnected) =>
-            multiplexer.AnyServer(ServerType, (uint)Interlocked.Increment(ref anyStartOffset), command, flags, allowDisconnected);
+            multiplexer?.AnyServer(ServerType, (uint)Interlocked.Increment(ref anyStartOffset), command, flags, allowDisconnected);
 
         private static ServerEndPoint? FindPrimary(ServerEndPoint endpoint, RedisCommand command)
         {

tests/StackExchange.Redis.Tests/BasicOpTests.cs

@@ -6,7 +6,21 @@
 
 namespace StackExchange.Redis.Tests;
 
-public class BasicOpsTests(ITestOutputHelper output, SharedConnectionFixture fixture) : TestBase(output, fixture)
+[RunPerProtocol]
+public class BasicOpsTests(ITestOutputHelper output, SharedConnectionFixture fixture)
+    : BasicOpsTestsBase(output, fixture, null)
+{
+}
+
+[RunPerProtocol]
+public class InProcBasicOpsTests(ITestOutputHelper output, InProcServerFixture fixture)
+    : BasicOpsTestsBase(output, null, fixture)
+{
+}
+
+[RunPerProtocol]
+public abstract class BasicOpsTestsBase(ITestOutputHelper output, SharedConnectionFixture? connection, InProcServerFixture? server)
+    : TestBase(output, connection, server)
 {
     [Fact]
     public async Task PingOnce()
@@ -471,6 +485,7 @@ public async Task WrappedDatabasePrefixIntegration()
     public async Task TransactionSync()
     {
         await using var conn = Create();
+        Assert.SkipUnless(conn.RawConfig.CommandMap.IsAvailable(RedisCommand.MULTI), "MULTI is not available");
         var db = conn.GetDatabase();
 
         RedisKey key = Me();
@@ -490,6 +505,8 @@ public async Task TransactionSync()
     public async Task TransactionAsync()
     {
         await using var conn = Create();
+        Assert.SkipUnless(conn.RawConfig.CommandMap.IsAvailable(RedisCommand.MULTI), "MULTI is not available");
+
         var db = conn.GetDatabase();
 
         RedisKey key = Me();

tests/StackExchange.Redis.Tests/Helpers/InProcServerFixture.cs

@@ -0,0 +1,27 @@
+using System;
+using StackExchange.Redis.Configuration;
+using Xunit;
+
+[assembly: AssemblyFixture(typeof(StackExchange.Redis.Tests.InProcServerFixture))]
+
+// ReSharper disable once CheckNamespace
+namespace StackExchange.Redis.Tests;
+
+public class InProcServerFixture : IDisposable
+{
+    private readonly InProcessTestServer _server = new();
+    private readonly ConfigurationOptions _config;
+    public InProcServerFixture()
+    {
+        _config = _server.GetClientConfig();
+        Configuration = _config.ToString();
+    }
+
+    public ConfigurationOptions Config => _config;
+
+    public string Configuration { get; }
+
+    public Tunnel? Tunnel => _server.Tunnel;
+
+    public void Dispose() => _server.Dispose();
+}

tests/StackExchange.Redis.Tests/HighIntegrityBasicOpsTests.cs

@@ -6,3 +6,8 @@ public class HighIntegrityBasicOpsTests(ITestOutputHelper output, SharedConnecti
 {
     internal override bool HighIntegrity => true;
 }
+
+public class InProcHighIntegrityBasicOpsTests(ITestOutputHelper output, InProcServerFixture fixture) : InProcBasicOpsTests(output, fixture)
+{
+    internal override bool HighIntegrity => true;
+}

tests/StackExchange.Redis.Tests/InProcessTestServer.cs

@@ -3,6 +3,7 @@
 using System.IO.Pipelines;
 using System.Net;
 using System.Net.Sockets;
+using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
 using Pipelines.Sockets.Unofficial;
@@ -14,17 +15,79 @@ namespace StackExchange.Redis.Tests;
 
 public class InProcessTestServer : MemoryCacheRedisServer
 {
-    public Tunnel Tunnel { get; }
-
     private readonly ITestOutputHelper? _log;
     public InProcessTestServer(ITestOutputHelper? log = null)
     {
+        RedisVersion = RedisFeatures.v6_0_0; // for client to expect RESP3
         _log = log;
         // ReSharper disable once VirtualMemberCallInConstructor
         _log?.WriteLine($"Creating in-process server: {ToString()}");
         Tunnel = new InProcTunnel(this);
     }
 
+    public Task<ConnectionMultiplexer> ConnectAsync(bool withPubSub = false, TextWriter? log = null)
+        => ConnectionMultiplexer.ConnectAsync(GetClientConfig(withPubSub), log);
+
+    public ConfigurationOptions GetClientConfig(bool withPubSub = false)
+    {
+        var commands = GetCommands();
+        if (!withPubSub)
+        {
+            commands.Remove(nameof(RedisCommand.SUBSCRIBE));
+            commands.Remove(nameof(RedisCommand.PSUBSCRIBE));
+            commands.Remove(nameof(RedisCommand.SSUBSCRIBE));
+            commands.Remove(nameof(RedisCommand.UNSUBSCRIBE));
+            commands.Remove(nameof(RedisCommand.PUNSUBSCRIBE));
+            commands.Remove(nameof(RedisCommand.SUNSUBSCRIBE));
+            commands.Remove(nameof(RedisCommand.PUBLISH));
+            commands.Remove(nameof(RedisCommand.SPUBLISH));
+        }
+        // transactions don't work yet
+        commands.Remove(nameof(RedisCommand.MULTI));
+        commands.Remove(nameof(RedisCommand.EXEC));
+        commands.Remove(nameof(RedisCommand.DISCARD));
+        commands.Remove(nameof(RedisCommand.WATCH));
+        commands.Remove(nameof(RedisCommand.UNWATCH));
+
+        var config = new ConfigurationOptions
+        {
+            CommandMap = CommandMap.Create(commands),
+            ConfigurationChannel = "",
+            TieBreaker = "",
+            DefaultVersion = RedisVersion,
+            ConnectTimeout = 10000,
+            SyncTimeout = 5000,
+            AsyncTimeout = 5000,
+            AllowAdmin = true,
+            Tunnel = Tunnel,
+        };
+        foreach (var endpoint in GetEndPoints())
+        {
+            config.EndPoints.Add(endpoint);
+        }
+        return config;
+    }
+
+    public Tunnel Tunnel { get; }
+
+    public override void Log(string message)
+    {
+        _log?.WriteLine(message);
+        base.Log(message);
+    }
+
+    protected override void OnMoved(RedisClient client, int hashSlot, Node node)
+    {
+        _log?.WriteLine($"Client {client.Id} being redirected: {hashSlot} to {node}");
+        base.OnMoved(client, hashSlot, node);
+    }
+
+    public override TypedRedisValue OnUnknownCommand(in RedisClient client, in RedisRequest request, ReadOnlySpan<byte> command)
+    {
+        _log?.WriteLine($"[{client.Id}] unknown command: {Encoding.ASCII.GetString(command)}");
+        return base.OnUnknownCommand(in client, in request, command);
+    }
+
     private sealed class InProcTunnel(
         InProcessTestServer server,
         PipeOptions? pipeOptions = null) : Tunnel
@@ -33,8 +96,12 @@ private sealed class InProcTunnel(
             EndPoint endpoint,
             CancellationToken cancellationToken)
         {
-            // server._log?.WriteLine($"Disabling client creation, requested endpoint: {Format.ToString(endpoint)}");
-            return default;
+            if (server.TryGetNode(endpoint, out _))
+            {
+                // server._log?.WriteLine($"Disabling client creation, requested endpoint: {Format.ToString(endpoint)}");
+                return default;
+            }
+            return base.GetSocketConnectEndpointAsync(endpoint, cancellationToken);
         }
 
         public override ValueTask<Stream?> BeforeAuthenticateAsync(
@@ -43,13 +110,18 @@ private sealed class InProcTunnel(
             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);
+            if (server.TryGetNode(endpoint, out var node))
+            {
+                server._log?.WriteLine(
+                    $"Client intercepted, endpoint {Format.ToString(endpoint)} ({connectionType}) mapped to {server.ServerType} node {node}");
+                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, node: node), cancellationToken);
+                var clientSide = StreamConnection.GetDuplex(serverToClient.Reader, clientToServer.Writer);
+                return new(clientSide);
+            }
+            return base.BeforeAuthenticateAsync(endpoint, connectionType, socket, cancellationToken);
         }
 
         private sealed class Duplex(PipeReader input, PipeWriter output) : IDuplexPipe
@@ -65,6 +137,7 @@ public ValueTask Dispose()
             }
         }
     }
+
     /*
 
     private readonly RespServer _server;