Pub/Sub fixes for subscribe/re-subscribe

.github/workflows/CI.yml

@@ -50,7 +50,6 @@ jobs:
       with:
         dotnet-version: | 
           3.1.x
-          5.0.x
           6.0.x
     - name: .NET Build
       run: dotnet build Build.csproj -c Release /p:CI=true

docs/ReleaseNotes.md

@@ -9,6 +9,11 @@
 - Moved tiebreaker fetching in connections into the handshake phase (streamline + simplification) (#1931 via NickCraver)
 - Fixed potential disposed object usage around Arenas (pulling in [Piplines.Sockets.Unofficial#63](https://github.com/mgravell/Pipelines.Sockets.Unofficial/pull/63) by MarcGravell)
 - Adds thread pool work item stats to exception messages to help diagnose contention (#1964 via NickCraver)
+- Overhauls pub/sub implementation for correctness (#1947 via NickCraver)
+  - Fixes a race in subscribing right after connected
+  - Fixes a race in subscribing immediately before a publish
+  - Fixes subscription routing on clusters (spreading instead of choosing 1 node)
+  - More correctly reconnects subscriptions on connection failures, including to other endpoints
 
 ## 2.2.88
 

src/StackExchange.Redis/ConnectionMultiplexer.cs

@@ -1898,14 +1898,15 @@ internal async Task<bool> ReconfigureAsync(bool first, bool reconfigureAll, LogP
                     }
                     if (!first)
                     {
-                        long subscriptionChanges = ValidateSubscriptions();
+                        // Calling the sync path here because it's all fire and forget
+                        long subscriptionChanges = EnsureSubscriptions(CommandFlags.FireAndForget);
                         if (subscriptionChanges == 0)
                         {
                             log?.WriteLine("No subscription changes necessary");
                         }
                         else
                         {
-                            log?.WriteLine($"Subscriptions reconfigured: {subscriptionChanges}");
+                            log?.WriteLine($"Subscriptions attempting reconnect: {subscriptionChanges}");
                         }
                     }
                     if (showStats)
@@ -2325,7 +2326,7 @@ internal void InitializeSentinel(LogProxy logProxy)
                             }
                         }
                     }
-                });
+                }, CommandFlags.FireAndForget);
             }
 
             // If we lose connection to a sentinel server,
@@ -2344,7 +2345,7 @@ internal void InitializeSentinel(LogProxy logProxy)
                 {
                     string[] messageParts = ((string)message).Split(new[] { ' ' }, StringSplitOptions.RemoveEmptyEntries);
                     UpdateSentinelAddressList(messageParts[0]);
-                });
+                }, CommandFlags.FireAndForget);
             }
         }
 

src/StackExchange.Redis/Interfaces/IDatabase.cs

@@ -851,7 +851,10 @@ public interface IDatabase : IRedis, IDatabaseAsync
         /// <param name="channel">The channel to publish to.</param>
         /// <param name="message">The message to send.</param>
         /// <param name="flags">The flags to use for this operation.</param>
-        /// <returns>The number of clients that received the message.</returns>
+        /// <returns>
+        /// The number of clients that received the message *on the destination server*,
+        /// note that this doesn't mean much in a cluster as clients can get the message through other nodes.
+        /// </returns>
         /// <remarks>https://redis.io/commands/publish</remarks>
         long Publish(RedisChannel channel, RedisValue message, CommandFlags flags = CommandFlags.None);
 

src/StackExchange.Redis/Interfaces/IDatabaseAsync.cs

@@ -828,7 +828,10 @@ public interface IDatabaseAsync : IRedisAsync
         /// <param name="channel">The channel to publish to.</param>
         /// <param name="message">The message to send.</param>
         /// <param name="flags">The flags to use for this operation.</param>
-        /// <returns>The number of clients that received the message.</returns>
+        /// <returns>
+        /// The number of clients that received the message *on the destination server*,
+        /// note that this doesn't mean much in a cluster as clients can get the message through other nodes.
+        /// </returns>
         /// <remarks>https://redis.io/commands/publish</remarks>
         Task<long> PublishAsync(RedisChannel channel, RedisValue message, CommandFlags flags = CommandFlags.None);
 

src/StackExchange.Redis/Interfaces/ISubscriber.cs

@@ -38,7 +38,10 @@ public interface ISubscriber : IRedis
         /// <param name="channel">The channel to publish to.</param>
         /// <param name="message">The message to publish.</param>
         /// <param name="flags">The command flags to use.</param>
-        /// <returns>the number of clients that received the message.</returns>
+        /// <returns>
+        /// The number of clients that received the message *on the destination server*,
+        /// note that this doesn't mean much in a cluster as clients can get the message through other nodes.
+        /// </returns>
         /// <remarks>https://redis.io/commands/publish</remarks>
         long Publish(RedisChannel channel, RedisValue message, CommandFlags flags = CommandFlags.None);
 
@@ -48,7 +51,10 @@ public interface ISubscriber : IRedis
         /// <param name="channel">The channel to publish to.</param>
         /// <param name="message">The message to publish.</param>
         /// <param name="flags">The command flags to use.</param>
-        /// <returns>the number of clients that received the message.</returns>
+        /// <returns>
+        /// The number of clients that received the message *on the destination server*,
+        /// note that this doesn't mean much in a cluster as clients can get the message through other nodes.
+        /// </returns>
         /// <remarks>https://redis.io/commands/publish</remarks>
         Task<long> PublishAsync(RedisChannel channel, RedisValue message, CommandFlags flags = CommandFlags.None);
 

src/StackExchange.Redis/Message.cs

@@ -589,7 +589,7 @@ internal bool TrySetResult<T>(T value)
         internal void SetEnqueued(PhysicalConnection connection)
         {
             SetWriteTime();
-            performance?.SetEnqueued();
+            performance?.SetEnqueued(connection?.BridgeCouldBeNull?.ConnectionType);
             _enqueuedTo = connection;
             if (connection == null)
             {
@@ -735,6 +735,8 @@ protected CommandChannelBase(int db, CommandFlags flags, RedisCommand command, i
             }
 
             public override string CommandAndKey => Command + " " + Channel;
+
+            public override int GetHashSlot(ServerSelectionStrategy serverSelectionStrategy) => serverSelectionStrategy.HashSlot(Channel);
         }
 
         internal abstract class CommandKeyBase : Message

src/StackExchange.Redis/PhysicalBridge.cs

@@ -8,7 +8,6 @@
 using System.Threading.Channels;
 using System.Threading.Tasks;
 using static StackExchange.Redis.ConnectionMultiplexer;
-using PendingSubscriptionState = global::StackExchange.Redis.ConnectionMultiplexer.Subscription.PendingSubscriptionState;
 #if !NETCOREAPP
 using Pipelines.Sockets.Unofficial.Threading;
 using static Pipelines.Sockets.Unofficial.Threading.MutexSlim;
@@ -102,7 +101,6 @@ public enum State : byte
         public void Dispose()
         {
             isDisposed = true;
-            ShutdownSubscriptionQueue();
             using (var tmp = physical)
             {
                 physical = null;
@@ -220,71 +218,6 @@ internal void GetCounters(ConnectionCounters counters)
             physical?.GetCounters(counters);
         }
 
-        private Channel<PendingSubscriptionState> _subscriptionBackgroundQueue;
-        private static readonly UnboundedChannelOptions s_subscriptionQueueOptions = new UnboundedChannelOptions
-        {
-             AllowSynchronousContinuations = false, // we do *not* want the async work to end up on the caller's thread
-             SingleReader = true, // only one reader will be started per channel
-             SingleWriter = true, // writes will be synchronized, because order matters
-        };
-
-        private Channel<PendingSubscriptionState> GetSubscriptionQueue()
-        {
-            var queue = _subscriptionBackgroundQueue;
-            if (queue == null)
-            {
-                queue = Channel.CreateUnbounded<PendingSubscriptionState>(s_subscriptionQueueOptions);
-                var existing = Interlocked.CompareExchange(ref _subscriptionBackgroundQueue, queue, null);
-
-                if (existing != null) return existing; // we didn't win, but that's fine 
-
-                // we won (_subqueue is now queue)
-                // this means we have a new channel without a reader; let's fix that!
-                Task.Run(() => ExecuteSubscriptionLoop());
-            }
-            return queue;
-        }
-
-        private void ShutdownSubscriptionQueue()
-        {
-            try
-            {
-                Interlocked.CompareExchange(ref _subscriptionBackgroundQueue, null, null)?.Writer.TryComplete();
-            }
-            catch { }
-        }
-
-        private async Task ExecuteSubscriptionLoop() // pushes items that have been enqueued over the bridge
-        {
-            // note: this will execute on the default pool rather than our dedicated pool; I'm... OK with this
-            var queue = _subscriptionBackgroundQueue ?? Interlocked.CompareExchange(ref _subscriptionBackgroundQueue, null, null); // just to be sure we can read it!
-            try
-            {
-                while (await queue.Reader.WaitToReadAsync().ForAwait() && queue.Reader.TryRead(out var next))
-                {
-                    try
-                    {
-                        // Treat these commands as background/handshake and do not allow queueing to backlog
-                        if ((await TryWriteAsync(next.Message, next.IsReplica).ForAwait()) != WriteResult.Success)
-                        {
-                            next.Abort();
-                        }
-                    }
-                    catch (Exception ex)
-                    {
-                        next.Fail(ex);
-                    }
-                }
-            }
-            catch (Exception ex)
-            {
-                Multiplexer.OnInternalError(ex, ServerEndPoint?.EndPoint, ConnectionType);
-            }
-        }
-
-        internal bool TryEnqueueBackgroundSubscriptionWrite(in PendingSubscriptionState state)
-            => !isDisposed && (_subscriptionBackgroundQueue ?? GetSubscriptionQueue()).Writer.TryWrite(state);
-
         internal readonly struct BridgeStatus
         {
             /// <summary>

src/StackExchange.Redis/Profiling/ProfiledCommand.cs

@@ -55,6 +55,7 @@ private static TimeSpan GetElapsedTime(long timestampDelta)
         private long RequestSentTimeStamp;
         private long ResponseReceivedTimeStamp;
         private long CompletedTimeStamp;
+        private ConnectionType? ConnectionType;
 
         private readonly ProfilingSession PushToWhenFinished;
 
@@ -86,7 +87,11 @@ public void SetMessage(Message msg)
             MessageCreatedTimeStamp = msg.CreatedTimestamp;
         }
 
-        public void SetEnqueued() => SetTimestamp(ref EnqueuedTimeStamp);
+        public void SetEnqueued(ConnectionType? connType)
+        {
+            SetTimestamp(ref EnqueuedTimeStamp);
+            ConnectionType = connType;
+        }
 
         public void SetRequestSent() => SetTimestamp(ref RequestSentTimeStamp);
 
@@ -117,16 +122,10 @@ public void SetCompleted()
         }
 
         public override string ToString() =>
-$@"EndPoint = {EndPoint}
-Db = {Db}
-Command = {Command}
-CommandCreated = {CommandCreated:u}
-CreationToEnqueued = {CreationToEnqueued}
-EnqueuedToSending = {EnqueuedToSending}
-SentToResponse = {SentToResponse}
-ResponseToCompletion = {ResponseToCompletion}
-ElapsedTime = {ElapsedTime}
-Flags = {Flags}
-RetransmissionOf = ({RetransmissionOf?.ToString() ?? "nothing"})";
+$@"{Command} (DB: {Db}, Flags: {Flags})
+     EndPoint = {EndPoint} ({ConnectionType})
+     Created = {CommandCreated:HH:mm:ss.ffff}
+     ElapsedTime = {ElapsedTime.TotalMilliseconds} ms (CreationToEnqueued: {CreationToEnqueued.TotalMilliseconds} ms, EnqueuedToSending: {EnqueuedToSending.TotalMilliseconds} ms, SentToResponse: {SentToResponse.TotalMilliseconds} ms, ResponseToCompletion = {ResponseToCompletion.TotalMilliseconds} ms){(RetransmissionOf != null ? @"
+     RetransmissionOf = " + RetransmissionOf : "")}";
     }
 }