Fix duplicate Pub/Sub subscription restoration after Sentinel failover

docs/PubSubOrder.md

@@ -23,3 +23,11 @@ multiplexer.GetSubscriber().Subscribe("messages", (channel, message) => {
     Console.WriteLine((string)message);
 });
 ```
+Reconnection
+---
+
+Single-node subscription restoration is coordinated until Redis acknowledges the subscription. Concurrent restoration triggers cannot establish independent routes for the same logical subscription, and late acknowledgements do not replace a newer confirmed connection. This applies to callback and queue subscriptions, including pattern channels; the processing-order choices above remain unchanged.
+
+A previously confirmed, still-live connection remains the delivery route while its replacement awaits acknowledgement. Ownership changes on the read thread when the replacement is acknowledged; later deliveries from the previous connection are rejected. Connection-state checks describe that confirmed delivery route, not a pending restore. Messages already admitted to a handler may finish after the handover. This does not make the handover atomic with publication across Redis nodes.
+
+Redis Pub/Sub does not replay messages missed during a disconnect or failover. Work already queued for a handler may still complete after unsubscribe; applications requiring durable delivery need a replay-capable messaging design.

src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs

@@ -19,6 +19,7 @@ public partial class ConnectionMultiplexer
     internal ConnectionMultiplexer? sentinelConnection;
     private int _sentinelInitialized;
     private int _demotedPrimaryRecoveryActive;
+    private int _sentinelSwitchInProgress;
 
     /// <summary>
     /// Initializes the connection as a Sentinel connection and adds the necessary event handlers to track changes to the managed primaries.
@@ -711,6 +712,15 @@ private static bool IsSentinelPrimaryDemoted(ConnectionMultiplexer connection)
     /// the view was already correct, or another reconfigure holds the gate - the callers that retry (the reconnect
     /// timer, the demotion heartbeat) get a later attempt either way.</returns>
     internal bool SwitchPrimary(EndPoint? switchBlame, ConnectionMultiplexer connection, TextWriter? writer = null, EndPoint? reportedPrimary = null)
+    {
+        // Keep a notification even if another switch or unrelated reconfigure wins the race.
+        if (reportedPrimary is not null) connection.currentSentinelPrimaryEndPoint = reportedPrimary;
+        if (Interlocked.CompareExchange(ref connection._sentinelSwitchInProgress, 1, 0) != 0) return false;
+        try { return SwitchPrimaryCore(switchBlame, connection, writer, reportedPrimary); }
+        finally { Volatile.Write(ref connection._sentinelSwitchInProgress, 0); }
+    }
+
+    private bool SwitchPrimaryCore(EndPoint? switchBlame, ConnectionMultiplexer connection, TextWriter? writer, EndPoint? reportedPrimary)
     {
         var logger = Logger.With(writer);
         if (connection.RawConfig.ServiceName is not string serviceName)
@@ -719,18 +729,15 @@ internal bool SwitchPrimary(EndPoint? switchBlame, ConnectionMultiplexer connect
             return false;
         }
 
-        // Get new primary - try twice (unless the caller already holds sentinel's answer)
+        // Discovery and recording Sentinel's answer do not reshape the connection. Keep blocking
+        // queries outside the claim, and retain the reported primary even if another reconfigure wins.
         EndPoint newPrimaryEndPoint = reportedPrimary
                                     ?? GetConfiguredPrimaryForService(serviceName)
                                     ?? GetConfiguredPrimaryForService(serviceName)
                                     ?? throw new RedisConnectionException(ConnectionFailureType.UnableToConnect, CommandFlags.None, $"Sentinel: Failed connecting to switch primary for service: {serviceName}");
-
         connection.currentSentinelPrimaryEndPoint = newPrimaryEndPoint;
 
-        // Claim the reconfigure gate BEFORE reshaping the connection. ReconfigureAsync refuses to run while
-        // another reconfigure holds this gate, so rewriting the endpoint list and retiring servers first would
-        // leave the connection reshaped but unconfigured whenever we lose that race - and two concurrent
-        // switches could interleave their endpoint rewrites. Claim first and a lost race touches nothing.
+        // Claim before rewriting endpoints: a lost claim must not leave a half-reconfigured topology.
         string cause = $"Primary switch {serviceName}";
         if (!connection.TryClaimReconfigure(cause))
         {

src/StackExchange.Redis/PhysicalConnection.Read.cs

@@ -694,13 +694,13 @@ private void OnMessage(
         {
             case RespPrefix.BulkString:
             case RespPrefix.SimpleString:
-                muxer.OnMessage(subscriptionChannel, messageChannel, reader.ReadRedisValue());
+                muxer.OnMessage(subscriptionChannel, messageChannel, reader.ReadRedisValue(), this);
                 break;
             case RespPrefix.Array:
                 var iter = reader.AggregateChildren();
                 while (iter.MoveNext())
                 {
-                    muxer.OnMessage(subscriptionChannel, messageChannel, iter.Value.ReadRedisValue());
+                    muxer.OnMessage(subscriptionChannel, messageChannel, iter.Value.ReadRedisValue(), this);
                 }
 
                 break;

src/StackExchange.Redis/RedisSubscriber.cs

@@ -76,11 +76,15 @@ internal bool GetSubscriberCounts(in RedisChannel channel, out int handlers, out
         /// Handler that executes whenever a message comes in, this doles out messages to any registered handlers.
         /// </summary>
         internal void OnMessage(in RedisChannel subscription, in RedisChannel channel, in RedisValue payload)
+            => OnMessage(subscription, channel, payload, null);
+
+        internal void OnMessage(in RedisChannel subscription, in RedisChannel channel, in RedisValue payload, PhysicalConnection? source)
         {
             ICompletable? completable = null;
             ChannelMessageQueue? queues = null;
             if (subscriptions.TryGetValue(subscription, out Subscription? sub))
             {
+                if (source is not null && !sub.Accepts(source)) return;
                 completable = sub.ForInvoke(channel, payload, out queues);
             }
             if (queues != null)
@@ -262,8 +266,10 @@ internal void ResubscribeToServer(Subscription sub, in RedisChannel channel, Ser
                 {
                     // 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.GetSubscriptionMessage(channel, SubscriptionAction.Subscribe, CommandFlags.None, false);
-                    _ = ExecuteAsync(message, sub.Processor, serverEndPoint).ContinueWith(
+                    var restore = sub is SingleNodeSubscription
+                        ? (Task)sub.EnsureSubscribedToServerAsync(this, channel, CommandFlags.None, false, serverEndPoint)
+                        : ExecuteAsync(sub.GetSubscriptionMessage(channel, SubscriptionAction.Subscribe, CommandFlags.None, false), sub.Processor, serverEndPoint);
+                    _ = restore.ContinueWith(
                         t => multiplexer.ReconfigureIfNeeded(serverEndPoint.EndPoint, false, cause: cause),
                         TaskContinuationOptions.OnlyOnFaulted);
                 }

src/StackExchange.Redis/ResultProcessor.cs

@@ -537,7 +537,10 @@ protected override void WriteImpl(in MessageWriter writer)
         public sealed class TrackSubscriptionsProcessor : ResultProcessor<bool>
         {
             private ConnectionMultiplexer.Subscription? Subscription { get; }
+            private readonly Action<PhysicalConnection>? _onSubscribed;
             public TrackSubscriptionsProcessor(ConnectionMultiplexer.Subscription? sub) => Subscription = sub;
+            internal TrackSubscriptionsProcessor(ConnectionMultiplexer.Subscription sub, Action<PhysicalConnection> onSubscribed)
+                : this(sub) => _onSubscribed = onSubscribed;
 
             protected override bool SetResultCore(PhysicalConnection connection, Message message, ref RespReader reader)
             {
@@ -555,23 +558,27 @@ protected override bool SetResultCore(PhysicalConnection connection, Message mes
                         if (iter.Value.TryReadInt64(out long count))
                         {
                             connection.SubscriptionCount = count;
-                            SetResult(message, true);
-
-                            var ep = connection.BridgeCouldBeNull?.ServerEndPoint;
-                            if (ep is not null)
+                            try
                             {
-                                switch (message.Command)
+                                var ep = connection.BridgeCouldBeNull?.ServerEndPoint;
+                                if (ep is not null)
                                 {
-                                    case RedisCommand.SUBSCRIBE:
-                                    case RedisCommand.SSUBSCRIBE:
-                                    case RedisCommand.PSUBSCRIBE:
-                                        Subscription?.AddEndpoint(ep);
-                                        break;
-                                    default:
-                                        Subscription?.TryRemoveEndpoint(ep);
-                                        break;
+                                    switch (message.Command)
+                                    {
+                                        case RedisCommand.SUBSCRIBE:
+                                        case RedisCommand.SSUBSCRIBE:
+                                        case RedisCommand.PSUBSCRIBE:
+                                            if (_onSubscribed is not null) _onSubscribed(connection);
+                                            else Subscription?.AddEndpoint(ep);
+                                            break;
+                                        default:
+                                            Subscription?.TryRemoveEndpoint(ep);
+                                            break;
+                                    }
                                 }
                             }
+                            catch (Exception ex) { connection.BridgeCouldBeNull?.Multiplexer.OnInternalError(ex); }
+                            finally { SetResult(message, true); }
                             return true;
                         }
                     }