Misc connection/subscription fixes (#2001)

In investigating an issue in #1989, I found a few gaps. Overall:
1. Twemproxy has an out of date CommandMap, which propagated to Envoy.
2. We were expecting both interactive and subscription connections to complete to complete the async handler...but we shouldn't because subscriptions may be disabled.
3. RedisSubscriber changes on the sync path weren't validating the message (asserting the command map has it enabled).

This fixes all of the above and adds another test considering all 3.

Author: NickCraver

docs/ReleaseNotes.md

@@ -2,6 +2,7 @@
 
 ## Unreleased
 - Adds bounds checking for `ExponentialRetry` backoff policy (#1921 via gliljas) 
+- When `SUBSCRIBE` is disabled, give proper errors and connect faster (#2001 via NickCraver)
 
 ## 2.5.27 (prerelease)
 

src/StackExchange.Redis/CommandMap.cs

@@ -26,25 +26,19 @@ public sealed class CommandMap
         {
             // see https://github.com/twitter/twemproxy/blob/master/notes/redis.md
             RedisCommand.KEYS, RedisCommand.MIGRATE, RedisCommand.MOVE, RedisCommand.OBJECT, RedisCommand.RANDOMKEY,
-            RedisCommand.RENAME, RedisCommand.RENAMENX, RedisCommand.SORT, RedisCommand.SCAN,
+            RedisCommand.RENAME, RedisCommand.RENAMENX, RedisCommand.SCAN,
 
-            RedisCommand.BITOP, RedisCommand.MSET, RedisCommand.MSETNX,
-
-            RedisCommand.HSCAN,
+            RedisCommand.BITOP, RedisCommand.MSETNX,
 
             RedisCommand.BLPOP, RedisCommand.BRPOP, RedisCommand.BRPOPLPUSH, // yeah, me neither!
 
-            RedisCommand.SSCAN,
-
-            RedisCommand.ZSCAN,
-
             RedisCommand.PSUBSCRIBE, RedisCommand.PUBLISH, RedisCommand.PUNSUBSCRIBE, RedisCommand.SUBSCRIBE, RedisCommand.UNSUBSCRIBE,
 
             RedisCommand.DISCARD, RedisCommand.EXEC, RedisCommand.MULTI, RedisCommand.UNWATCH, RedisCommand.WATCH,
 
             RedisCommand.SCRIPT,
 
-            RedisCommand.ECHO, RedisCommand.PING, RedisCommand.QUIT, RedisCommand.SELECT,
+            RedisCommand.ECHO, RedisCommand.PING, RedisCommand.SELECT,
 
             RedisCommand.BGREWRITEAOF, RedisCommand.BGSAVE, RedisCommand.CLIENT, RedisCommand.CLUSTER, RedisCommand.CONFIG, RedisCommand.DBSIZE,
             RedisCommand.DEBUG, RedisCommand.FLUSHALL, RedisCommand.FLUSHDB, RedisCommand.INFO, RedisCommand.LASTSAVE, RedisCommand.MONITOR, RedisCommand.REPLICAOF,

src/StackExchange.Redis/ConnectionMultiplexer.cs

@@ -1654,7 +1654,7 @@ private void ActivateAllServers(LogProxy log)
             foreach (var server in GetServerSnapshot())
             {
                 server.Activate(ConnectionType.Interactive, log);
-                if (CommandMap.IsAvailable(RedisCommand.SUBSCRIBE))
+                if (server.SupportsSubscriptions)
                 {
                     // Intentionally not logging the sub connection
                     server.Activate(ConnectionType.Subscription, null);

src/StackExchange.Redis/RedisSubscriber.cs

@@ -378,7 +378,7 @@ internal bool EnsureSubscribedToServer(Subscription sub, RedisChannel channel, C
             sub.SetCurrentServer(null); // we're not appropriately connected, so blank it out for eligible reconnection
             var message = sub.GetMessage(channel, SubscriptionAction.Subscribe, flags, internalCall);
             var selected = multiplexer.SelectServer(message);
-            return multiplexer.ExecuteSyncImpl(message, sub.Processor, selected);
+            return ExecuteSync(message, sub.Processor, selected);
         }
 
         Task ISubscriber.SubscribeAsync(RedisChannel channel, Action<RedisChannel, RedisValue> handler, CommandFlags flags)

src/StackExchange.Redis/ServerEndPoint.cs

@@ -75,6 +75,8 @@ public ServerEndPoint(ConnectionMultiplexer multiplexer, EndPoint endpoint)
         public bool IsConnected => interactive?.IsConnected == true;
         public bool IsSubscriberConnected => subscription?.IsConnected == true;
 
+        public bool SupportsSubscriptions => Multiplexer.CommandMap.IsAvailable(RedisCommand.SUBSCRIBE);
+
         public bool IsConnecting => interactive?.IsConnecting == true;
 
         private readonly List<TaskCompletionSource<string>> _pendingConnectionMonitors = new List<TaskCompletionSource<string>>();
@@ -629,9 +631,10 @@ internal void OnFullyEstablished(PhysicalConnection connection, string source)
                         // Since we're issuing commands inside a SetResult path in a message, we'd create a deadlock by waiting.
                         Multiplexer.EnsureSubscriptions(CommandFlags.FireAndForget);
                     }
-                    if (IsConnected && IsSubscriberConnected)
+                    if (IsConnected && (IsSubscriberConnected || !SupportsSubscriptions))
                     {
                         // Only connect on the second leg - we can accomplish this by checking both
+                        // Or the first leg, if we're only making 1 connection because subscriptions aren't supported
                         CompletePendingConnectionMonitors(source);
                     }
 

tests/StackExchange.Redis.Tests/Config.cs

@@ -263,6 +263,21 @@ public void ReadConfigWithConfigDisabled()
             }
         }
 
+        [Fact]
+        public void ConnectWithSubscribeDisabled()
+        {
+            using (var muxer = Create(allowAdmin: true, disabledCommands: new[] { "subscribe" }))
+            {
+                Assert.True(muxer.IsConnected);
+                var servers = muxer.GetServerSnapshot();
+                Assert.True(servers[0].IsConnected);
+                Assert.False(servers[0].IsSubscriberConnected);
+
+                var ex = Assert.Throws<RedisCommandException>(() => muxer.GetSubscriber().Subscribe(Me(), (_, _) => GC.KeepAlive(this)));
+                Assert.Equal("This operation has been disabled in the command-map and cannot be used: SUBSCRIBE", ex.Message);
+            }
+        }
+
         [Fact]
         public void ReadConfig()
         {