Fix incorrect routing of some sorted-set and stream commands (#3080)

* 1. test and fix incorrect routing of some "sorted set" and "stream" commands
2. fix via new custom efficient message cores
3. add reusable helpers for common null checks on key/value arrays
4. prep work for likely ZDIFFCARD/ZUNIONCARD

* rationalize

* prefer async tests

* add release note

Author: mgravell

docs/ReleaseNotes.md

@@ -11,6 +11,7 @@ Current package versions:
 - Add experimental Redis 8.8 array support, including array APIs on `IDatabase`/`IDatabaseAsync`,
   array helper types, `RedisType.Array`, and array delete keyspace notification event types. ([#3076 by @mgravell](https://github.com/StackExchange/StackExchange.Redis/pull/3076))
 - Enable TCP keep-alives ([#3078 by @mgravell](https://github.com/StackExchange/StackExchange.Redis/pull/3078))
+- Fix incorrect routing of some sorted-set and stream commands ([#3080 by @mgravell](https://github.com/StackExchange/StackExchange.Redis/pull/3080))
 - `ConfigurationOptions` : don't persist `Protocol` when it comes from the defaults-provider. ([#3082 by @mgravell](https://github.com/StackExchange/StackExchange.Redis/pull/3082))
 
 ## 2.13.1

src/StackExchange.Redis/Enums/SetOperation.cs

@@ -25,15 +25,39 @@ public enum SetOperation
 
     internal static class SetOperationExtensions
     {
-        internal static RedisCommand ToCommand(this SetOperation operation, bool store) => operation switch
+        internal static RedisCommand ToSetCommand(this SetOperation operation) => operation switch
+        {
+            SetOperation.Union => RedisCommand.SUNION,
+            SetOperation.Intersect => RedisCommand.SINTER,
+            SetOperation.Difference => RedisCommand.SDIFF,
+            _ => OutOfRange(operation),
+        };
+
+        internal static RedisCommand ToSetStoreCommand(this SetOperation operation) => operation switch
+        {
+            SetOperation.Union => RedisCommand.SUNIONSTORE,
+            SetOperation.Intersect => RedisCommand.SINTERSTORE,
+            SetOperation.Difference => RedisCommand.SDIFFSTORE,
+            _ => OutOfRange(operation),
+        };
+
+        internal static RedisCommand ToSortedSetCommand(this SetOperation operation) => operation switch
         {
-            SetOperation.Intersect when store => RedisCommand.ZINTERSTORE,
-            SetOperation.Intersect => RedisCommand.ZINTER,
-            SetOperation.Union when store => RedisCommand.ZUNIONSTORE,
             SetOperation.Union => RedisCommand.ZUNION,
-            SetOperation.Difference when store => RedisCommand.ZDIFFSTORE,
+            SetOperation.Intersect => RedisCommand.ZINTER,
             SetOperation.Difference => RedisCommand.ZDIFF,
-            _ => throw new ArgumentOutOfRangeException(nameof(operation)),
+            _ => OutOfRange(operation),
+        };
+
+        internal static RedisCommand ToSortedSetStoreCommand(this SetOperation operation) => operation switch
+        {
+            SetOperation.Union => RedisCommand.ZUNIONSTORE,
+            SetOperation.Intersect => RedisCommand.ZINTERSTORE,
+            SetOperation.Difference => RedisCommand.ZDIFFSTORE,
+            _ => OutOfRange(operation),
         };
+
+        private static RedisCommand OutOfRange(SetOperation operation) =>
+            throw new ArgumentOutOfRangeException(nameof(operation), operation, null);
     }
 }

src/StackExchange.Redis/ExtensionMethods.Internal.cs

@@ -1,4 +1,5 @@
-using System.Collections.Generic;
+using System;
+using System.Collections.Generic;
 using System.Diagnostics.CodeAnalysis;
 
 namespace StackExchange.Redis
@@ -11,6 +12,26 @@ internal static bool IsNullOrEmpty([NotNullWhen(false)] this string? s) =>
         internal static bool IsNullOrWhiteSpace([NotNullWhen(false)] this string? s) =>
             string.IsNullOrWhiteSpace(s);
 
+        internal static RedisKey[] AssertAllNonNull(this RedisKey[] keys)
+        {
+            if (keys is null) throw new ArgumentNullException(nameof(keys));
+            for (var i = 0; i < keys.Length; i++)
+            {
+                keys[i].AssertNotNull();
+            }
+            return keys;
+        }
+
+        internal static RedisValue[] AssertAllNonNull(this RedisValue[] values)
+        {
+            if (values is null) throw new ArgumentNullException(nameof(values));
+            for (var i = 0; i < values.Length; i++)
+            {
+                values[i].AssertNotNull();
+            }
+            return values;
+        }
+
 #if !NET
         internal static bool TryDequeue<T>(this Queue<T> queue, [NotNullWhen(true)] out T? result)
         {

src/StackExchange.Redis/Message.cs

@@ -393,6 +393,11 @@ public static Message Create(
         public static Message CreateInSlot(int db, int slot, CommandFlags flags, RedisCommand command, RedisValue[] values) =>
             new CommandSlotValuesMessage(db, slot, flags, command, values);
 
+        // The key here is used only to route the message in cluster mode; it is not written as an argument.
+        // Use this for command shapes where the key appears in a non-standard position in the values payload.
+        public static Message CreateInKeySlot(int db, in RedisKey key, CommandFlags flags, RedisCommand command, RedisValue[] values) =>
+            new CommandKeySlotValuesMessage(db, flags, command, key, values);
+
         public static Message Create(int db, CommandFlags flags, RedisCommand command, KeyValuePair<RedisKey, RedisValue>[] values, Expiration expiry, When when)
             => new MultiSetMessage(db, flags, command, values, expiry, when);
 
@@ -984,21 +989,13 @@ private sealed class CommandKeyKeysMessage : CommandKeyBase
             private readonly RedisKey[] keys;
             public CommandKeyKeysMessage(int db, CommandFlags flags, RedisCommand command, in RedisKey key, RedisKey[] keys) : base(db, flags, command, key)
             {
-                for (int i = 0; i < keys.Length; i++)
-                {
-                    keys[i].AssertNotNull();
-                }
-                this.keys = keys;
+                this.keys = keys.AssertAllNonNull();
             }
 
             public override int GetHashSlot(ServerSelectionStrategy serverSelectionStrategy)
             {
                 var slot = serverSelectionStrategy.HashSlot(Key);
-                for (int i = 0; i < keys.Length; i++)
-                {
-                    slot = serverSelectionStrategy.CombineSlot(slot, keys[i]);
-                }
-                return slot;
+                return serverSelectionStrategy.CombineSlot(slot, keys);
             }
 
             protected override void WriteImpl(PhysicalConnection physical)
@@ -1050,11 +1047,7 @@ private sealed class CommandValuesMessage : Message
             private readonly RedisValue[] values;
             public CommandValuesMessage(int db, CommandFlags flags, RedisCommand command, RedisValue[] values) : base(db, flags, command)
             {
-                for (int i = 0; i < values.Length; i++)
-                {
-                    values[i].AssertNotNull();
-                }
-                this.values = values;
+                this.values = values.AssertAllNonNull();
             }
 
             protected override void WriteImpl(PhysicalConnection physical)
@@ -1073,22 +1066,10 @@ private sealed class CommandKeysMessage : Message
             private readonly RedisKey[] keys;
             public CommandKeysMessage(int db, CommandFlags flags, RedisCommand command, RedisKey[] keys) : base(db, flags, command)
             {
-                for (int i = 0; i < keys.Length; i++)
-                {
-                    keys[i].AssertNotNull();
-                }
-                this.keys = keys;
+                this.keys = keys.AssertAllNonNull();
             }
 
-            public override int GetHashSlot(ServerSelectionStrategy serverSelectionStrategy)
-            {
-                int slot = ServerSelectionStrategy.NoSlot;
-                for (int i = 0; i < keys.Length; i++)
-                {
-                    slot = serverSelectionStrategy.CombineSlot(slot, keys[i]);
-                }
-                return slot;
-            }
+            public override int GetHashSlot(ServerSelectionStrategy serverSelectionStrategy) => serverSelectionStrategy.HashSlot(keys);
 
             protected override void WriteImpl(PhysicalConnection physical)
             {
@@ -1125,11 +1106,7 @@ private sealed class CommandKeyValuesKeyMessage : CommandKeyBase
             private readonly RedisValue[] values;
             public CommandKeyValuesKeyMessage(int db, CommandFlags flags, RedisCommand command, in RedisKey key0, RedisValue[] values, in RedisKey key1) : base(db, flags, command, key0)
             {
-                for (int i = 0; i < values.Length; i++)
-                {
-                    values[i].AssertNotNull();
-                }
-                this.values = values;
+                this.values = values.AssertAllNonNull();
                 key1.AssertNotNull();
                 this.key1 = key1;
             }
@@ -1155,11 +1132,7 @@ private sealed class CommandKeyValuesMessage : CommandKeyBase
             private readonly RedisValue[] values;
             public CommandKeyValuesMessage(int db, CommandFlags flags, RedisCommand command, in RedisKey key, RedisValue[] values) : base(db, flags, command, key)
             {
-                for (int i = 0; i < values.Length; i++)
-                {
-                    values[i].AssertNotNull();
-                }
-                this.values = values;
+                this.values = values.AssertAllNonNull();
             }
 
             protected override void WriteImpl(PhysicalConnection physical)
@@ -1177,14 +1150,9 @@ private sealed class CommandKeyKeyValuesMessage : CommandKeyBase
             private readonly RedisValue[] values;
             public CommandKeyKeyValuesMessage(int db, CommandFlags flags, RedisCommand command, in RedisKey key, in RedisKey key1, RedisValue[] values) : base(db, flags, command, key)
             {
-                for (int i = 0; i < values.Length; i++)
-                {
-                    values[i].AssertNotNull();
-                }
-
                 key1.AssertNotNull();
                 this.key1 = key1;
-                this.values = values;
+                this.values = values.AssertAllNonNull();
             }
 
             protected override void WriteImpl(PhysicalConnection physical)
@@ -1204,16 +1172,11 @@ private sealed class CommandKeyValueValueValuesMessage : CommandKeyBase
             private readonly RedisValue[] values;
             public CommandKeyValueValueValuesMessage(int db, CommandFlags flags, RedisCommand command, in RedisKey key, in RedisValue value0, in RedisValue value1, RedisValue[] values) : base(db, flags, command, key)
             {
-                for (int i = 0; i < values.Length; i++)
-                {
-                    values[i].AssertNotNull();
-                }
-
                 value0.AssertNotNull();
                 value1.AssertNotNull();
                 this.value0 = value0;
                 this.value1 = value1;
-                this.values = values;
+                this.values = values.AssertAllNonNull();
             }
 
             protected override void WriteImpl(PhysicalConnection physical)
@@ -1683,14 +1646,32 @@ public CommandSlotValuesMessage(int db, int slot, CommandFlags flags, RedisComma
                 : base(db, flags, command)
             {
                 this.slot = slot;
+                this.values = values.AssertAllNonNull();
+            }
+
+            public override int GetHashSlot(ServerSelectionStrategy serverSelectionStrategy) => slot;
+
+            protected override void WriteImpl(PhysicalConnection physical)
+            {
+                physical.WriteHeader(command, values.Length);
                 for (int i = 0; i < values.Length; i++)
                 {
-                    values[i].AssertNotNull();
+                    physical.WriteBulkString(values[i]);
                 }
-                this.values = values;
             }
+            public override int ArgCount => values.Length;
+        }
 
-            public override int GetHashSlot(ServerSelectionStrategy serverSelectionStrategy) => slot;
+        private sealed class CommandKeySlotValuesMessage : CommandKeyBase
+        {
+            private readonly RedisValue[] values;
+
+            public CommandKeySlotValuesMessage(int db, CommandFlags flags, RedisCommand command, in RedisKey key, RedisValue[] values)
+                : base(db, flags, command, key)
+            {
+                // Key is captured by CommandKeyBase for routing only; values are the complete serialized arguments.
+                this.values = values.AssertAllNonNull();
+            }
 
             protected override void WriteImpl(PhysicalConnection physical)
             {