v2.5 work: BacklogPolicy

.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

src/StackExchange.Redis/BacklogPolicy.cs

@@ -0,0 +1,43 @@
+namespace StackExchange.Redis
+{
+    /// <summary>
+    /// The backlog policy to use for commands. This policy comes into effect when a connection is unhealthy or unavailable.
+    /// The policy can choose to backlog commands and wait to try them (within their timeout) against a connection when it comes up,
+    /// or it could choose to fail fast and throw ASAP. Different apps desire different behaviors with backpressure and how to handle
+    /// large amounts of load, so this is configurable to optimize the happy path but avoid spiral-of-death queue scenarios for others.
+    /// </summary>
+    public class BacklogPolicy
+    {
+        /// <summary>
+        /// Backlog behavior matching StackExchange.Redis's 2.x line, failing fast and not attempting to queue
+        /// and retry when a connection is available again.
+        /// </summary>
+        public static BacklogPolicy FailFast { get; } = new()
+        {
+            QueueWhileDisconnected = false,
+            AbortPendingOnConnectionFailure = true,
+        };
+
+        /// <summary>
+        /// Default backlog policy which will allow commands to be issues against an endpoint and queue up.
+        /// Commands are still subject to their async timeout (which serves as a queue size check).
+        /// </summary>
+        public static BacklogPolicy Default { get; } = new()
+        {
+            QueueWhileDisconnected = true,
+            AbortPendingOnConnectionFailure = false,
+        };
+
+        /// <summary>
+        /// Whether to queue commands while disconnected.
+        /// True means queue for attempts up until their timeout.
+        /// False means to fail ASAP and queue nothing.
+        /// </summary>
+        public bool QueueWhileDisconnected { get; init; }
+
+        /// <summary>
+        /// Whether to immediately abandon (with an exception) all pending commands when a connection goes unhealthy.
+        /// </summary>
+        public bool AbortPendingOnConnectionFailure { get; init; }
+    }
+}

src/StackExchange.Redis/ConfigurationOptions.cs

@@ -146,6 +146,8 @@ public static string TryNormalize(string value)
 
         private IReconnectRetryPolicy reconnectRetryPolicy;
 
+        private BacklogPolicy backlogPolicy;
+
         /// <summary>
         /// A LocalCertificateSelectionCallback delegate responsible for selecting the certificate used for authentication; note
         /// that this cannot be specified in the configuration-string.
@@ -372,6 +374,15 @@ public IReconnectRetryPolicy ReconnectRetryPolicy
             set => reconnectRetryPolicy = value;
         }
 
+        /// <summary>
+        /// The backlog policy to be used for commands when a connection is unhealthy.
+        /// </summary>
+        public BacklogPolicy BacklogPolicy
+        {
+            get => backlogPolicy ?? BacklogPolicy.Default;
+            set => backlogPolicy = value;
+        }
+
         /// <summary>
         /// Indicates whether endpoints should be resolved via DNS before connecting.
         /// If enabled the ConnectionMultiplexer will not re-resolve DNS
@@ -541,6 +552,7 @@ public ConfigurationOptions Clone()
                 responseTimeout = responseTimeout,
                 DefaultDatabase = DefaultDatabase,
                 ReconnectRetryPolicy = reconnectRetryPolicy,
+                BacklogPolicy = backlogPolicy,
                 SslProtocols = SslProtocols,
                 checkCertificateRevocation = checkCertificateRevocation,
             };

src/StackExchange.Redis/ConnectionMultiplexer.cs

@@ -804,15 +804,15 @@ internal void OnHashSlotMoved(int hashSlot, EndPoint old, EndPoint @new)
         /// <param name="key">The key to get a hash slot ID for.</param>
         public int HashSlot(RedisKey key) => ServerSelectionStrategy.HashSlot(key);
 
-        internal ServerEndPoint AnyConnected(ServerType serverType, uint startOffset, RedisCommand command, CommandFlags flags)
+        internal ServerEndPoint AnyServer(ServerType serverType, uint startOffset, RedisCommand command, CommandFlags flags, bool allowDisconnected)
         {
             var tmp = GetServerSnapshot();
             int len = tmp.Length;
             ServerEndPoint fallback = null;
             for (int i = 0; i < len; i++)
             {
                 var server = tmp[(int)(((uint)i + startOffset) % len)];
-                if (server != null && server.ServerType == serverType && server.IsSelectable(command))
+                if (server != null && server.ServerType == serverType && server.IsSelectable(command, allowDisconnected))
                 {
                     if (server.IsReplica)
                     {
@@ -1898,7 +1898,7 @@ internal async Task<bool> ReconfigureAsync(bool first, bool reconfigureAll, LogP
                     }
                     if (!first)
                     {
-                        long subscriptionChanges = ValidateSubscriptions();
+                        long subscriptionChanges = await EnsureSubscriptionsAsync();
                         if (subscriptionChanges == 0)
                         {
                             log?.WriteLine("No subscription changes necessary");
@@ -2168,6 +2168,12 @@ private bool PrepareToPushMessageToBridge<T>(Message message, ResultProcessor<T>
             {
                 // Infer a server automatically
                 server = SelectServer(message);
+
+                // If we didn't find one successfully, and we're allowed, queue for any viable server
+                if (server == null && message != null && RawConfig.BacklogPolicy.QueueWhileDisconnected)
+                {
+                    server = ServerSelectionStrategy.Select(message, allowDisconnected: true);
+                }
             }
             else // a server was specified; do we trust their choice, though?
             {
@@ -2185,7 +2191,9 @@ private bool PrepareToPushMessageToBridge<T>(Message message, ResultProcessor<T>
                         }
                         break;
                 }
-                if (!server.IsConnected)
+
+                // If we're not allowed to queue while disconnected, we'll bomb out below.
+                if (!server.IsConnected && !RawConfig.BacklogPolicy.QueueWhileDisconnected)
                 {
                     // well, that's no use!
                     server = null;
@@ -2325,7 +2333,7 @@ internal void InitializeSentinel(LogProxy logProxy)
                             }
                         }
                     }
-                });
+                }, CommandFlags.FireAndForget);
             }
 
             // If we lose connection to a sentinel server,
@@ -2344,7 +2352,7 @@ internal void InitializeSentinel(LogProxy logProxy)
                 {
                     string[] messageParts = ((string)message).Split(new[] { ' ' }, StringSplitOptions.RemoveEmptyEntries);
                     UpdateSentinelAddressList(messageParts[0]);
-                });
+                }, CommandFlags.FireAndForget);
             }
         }
 

src/StackExchange.Redis/Enums/CommandFlags.cs

@@ -81,8 +81,6 @@ public enum CommandFlags
         /// </summary>
         NoScriptCache = 512,
 
-        // 1024: used for timed-out; never user-specified, so not visible on the public API
-
-        // 2048: Use subscription connection type; never user-specified, so not visible on the public API
+        // 1024: Use subscription connection type; never user-specified, so not visible on the public API
     }
 }

src/StackExchange.Redis/Message.cs

@@ -1,12 +1,10 @@
 using System;
 using System.Collections.Generic;
 using System.Diagnostics;
-using System.IO;
 using System.Linq;
 using System.Runtime.CompilerServices;
 using System.Text;
 using System.Threading;
-using System.Threading.Tasks;
 using StackExchange.Redis.Profiling;
 using static StackExchange.Redis.ConnectionMultiplexer;
 
@@ -60,8 +58,7 @@ internal abstract class Message : ICompletable
 
         private const CommandFlags AskingFlag = (CommandFlags)32,
                                    ScriptUnavailableFlag = (CommandFlags)256,
-                                   NeedsAsyncTimeoutCheckFlag = (CommandFlags)1024,
-                                   DemandSubscriptionConnection = (CommandFlags)2048;
+                                   DemandSubscriptionConnection = (CommandFlags)1024;
 
         private const CommandFlags MaskMasterServerPreference = CommandFlags.DemandMaster
                                                               | CommandFlags.DemandReplica
@@ -589,7 +586,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)
             {
@@ -645,10 +642,7 @@ internal void SetRequestSent()
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
         internal void SetWriteTime()
         {
-            if ((Flags & NeedsAsyncTimeoutCheckFlag) != 0)
-            {
-                _writeTickCount = Environment.TickCount; // note this might be reset if we resend a message, cluster-moved etc; I'm OK with that
-            }
+            _writeTickCount = Environment.TickCount; // note this might be reset if we resend a message, cluster-moved etc; I'm OK with that
         }
         private int _writeTickCount;
         public int GetWriteTime() => Volatile.Read(ref _writeTickCount);
@@ -662,21 +656,17 @@ internal void SetWriteTime()
         /// </summary>
         internal void SetForSubscriptionBridge() => Flags |= DemandSubscriptionConnection;
 
-        private void SetNeedsTimeoutCheck() => Flags |= NeedsAsyncTimeoutCheckFlag;
-        internal bool HasAsyncTimedOut(int now, int timeoutMilliseconds, out int millisecondsTaken)
+        /// <summary>
+        /// Checks if this message has violated the provided timeout.
+        /// Whether it's a sync operation in a .Wait() or in the backlog queue or written/pending asynchronously, we need to timeout everything.
+        /// ...or we get indefinite Task hangs for completions.
+        /// </summary>
+        internal bool HasTimedOut(int now, int timeoutMilliseconds, out int millisecondsTaken)
         {
-            if ((Flags & NeedsAsyncTimeoutCheckFlag) != 0)
+            millisecondsTaken = unchecked(now - _writeTickCount); // note: we can't just check "if sent < cutoff" because of wrap-aro
+            if (millisecondsTaken >= timeoutMilliseconds)
             {
-                millisecondsTaken = unchecked(now - _writeTickCount); // note: we can't just check "if sent < cutoff" because of wrap-aro
-                if (millisecondsTaken >= timeoutMilliseconds)
-                {
-                    Flags &= ~NeedsAsyncTimeoutCheckFlag; // note: we don't remove it from the queue - still might need to marry it up; but: it is toast
-                    return true;
-                }
-            }
-            else
-            {
-                millisecondsTaken = default;
+                return true;
             }
             return false;
         }
@@ -695,16 +685,17 @@ internal void SetPreferMaster() =>
         internal void SetPreferReplica() =>
             Flags = (Flags & ~MaskMasterServerPreference) | CommandFlags.PreferReplica;
 
+        /// <remarks>
+        /// Note order here reversed to prevent overload resolution errors
+        /// </remarks>
         internal void SetSource(ResultProcessor resultProcessor, IResultBox resultBox)
-        { // note order here reversed to prevent overload resolution errors
-            if (resultBox != null && resultBox.IsAsync) SetNeedsTimeoutCheck();
+        {
             this.resultBox = resultBox;
             this.resultProcessor = resultProcessor;
         }
 
         internal void SetSource<T>(IResultBox<T> resultBox, ResultProcessor<T> resultProcessor)
         {
-            if (resultBox != null && resultBox.IsAsync) SetNeedsTimeoutCheck();
             this.resultBox = resultBox;
             this.resultProcessor = resultProcessor;
         }
@@ -735,6 +726,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