SUNSUBSCRIBE handling; if possible, use the active connection to find where we should be subscribing

Author: mgravell

src/StackExchange.Redis/PhysicalConnection.cs

@@ -1870,11 +1870,11 @@ private void MatchResult(in RawResult result)
                         var server = BridgeCouldBeNull?.ServerEndPoint;
                         if (server is not null && muxer.TryGetSubscription(subscriptionChannel, out var subscription))
                         {
-                            if (subscription.GetCurrentServer() == server)
-                            {
-                                subscription.SetCurrentServer(null); // wipe
-                                muxer.ReconfigureIfNeeded(server.EndPoint, fromBroadcast: true, PushSUnsubscribe.Text);
-                            }
+                            // wipe and reconnect; but: to where?
+                            // counter-intuitively, the only server we *know* already knows the new route is:
+                            // the outgoing server, since it had to change to MIGRATING etc; the new INCOMING server
+                            // knows, but *we don't know who that is*, and other nodes: aren't guaranteed to know (yet)
+                            muxer.DefaultSubscriber.ResubscribeToServer(subscription, subscriptionChannel, server, cause: PushSUnsubscribe.Text);
                         }
                         return; // and STOP PROCESSING; unsolicited
                 }

src/StackExchange.Redis/RedisSubscriber.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Collections.Concurrent;
 using System.Diagnostics.CodeAnalysis;
+using System.Diagnostics.SymbolStore;
 using System.Net;
 using System.Threading;
 using System.Threading.Tasks;
@@ -13,7 +14,7 @@ namespace StackExchange.Redis
     public partial class ConnectionMultiplexer
     {
         private RedisSubscriber? _defaultSubscriber;
-        private RedisSubscriber DefaultSubscriber => _defaultSubscriber ??= new RedisSubscriber(this, null);
+        internal RedisSubscriber DefaultSubscriber => _defaultSubscriber ??= new RedisSubscriber(this, null);
 
         private readonly ConcurrentDictionary<RedisChannel, Subscription> subscriptions = new();
 
@@ -282,6 +283,17 @@ internal void GetSubscriberCounts(out int handlers, out int queues)
 
             internal ServerEndPoint? GetCurrentServer() => Volatile.Read(ref CurrentServer);
             internal void SetCurrentServer(ServerEndPoint? server) => CurrentServer = server;
+            // conditional clear
+            internal bool ClearCurrentServer(ServerEndPoint expected)
+            {
+                if (CurrentServer == expected)
+                {
+                    CurrentServer = null;
+                    return true;
+                }
+
+                return false;
+            }
 
             /// <summary>
             /// Evaluates state and if we're not currently connected, clears the server reference.
@@ -425,6 +437,27 @@ internal bool EnsureSubscribedToServer(Subscription sub, RedisChannel channel, C
             return ExecuteSync(message, sub.Processor, selected);
         }
 
+        internal void ResubscribeToServer(Subscription sub, RedisChannel channel, ServerEndPoint serverEndPoint, string cause)
+        {
+            // conditional: only if that's the server we were connected to, or "none"; we don't want to end up duplicated
+            if (sub.ClearCurrentServer(serverEndPoint) || !sub.IsConnected)
+            {
+                if (serverEndPoint.IsSubscriberConnected)
+                {
+                    // we'll *try* for a simple resubscribe, following any -MOVED etc, but if that fails: fall back
+                    // to full reconfigure; importantly, note that we've already recorded the disconnect
+                    var message = sub.GetMessage(channel, SubscriptionAction.Subscribe, CommandFlags.None, false);
+                    _ = ExecuteAsync(message, sub.Processor, serverEndPoint).ContinueWith(
+                        t => multiplexer.ReconfigureIfNeeded(serverEndPoint.EndPoint, false, cause: cause),
+                        TaskContinuationOptions.OnlyOnFaulted);
+                }
+                else
+                {
+                    multiplexer.ReconfigureIfNeeded(serverEndPoint.EndPoint, false, cause: cause);
+                }
+            }
+        }
+
         Task ISubscriber.SubscribeAsync(RedisChannel channel, Action<RedisChannel, RedisValue> handler, CommandFlags flags)
             => SubscribeAsync(channel, handler, null, flags);
 

tests/StackExchange.Redis.Tests/ClusterShardedTests.cs

@@ -339,6 +339,8 @@ void WriteLog(string caption, RedisResult result)
             var received = await queue.ReadAsync(timeout.Token);
             Log($"Message received: {received.Message}");
             Assert.Equal(msg, (string)received.Message!);
+            ep = subscriber.SubscribedEndpoint(channel);
+            Log($"Endpoint after receiving message: {Format.ToString(ep)}");
         }
 
         Log("Unsubscribing...");