API Additions/Deprecations: MakePrimaryAsync/ReplicaOfAsync

.github/workflows/CI.yml

@@ -1,4 +1,4 @@
-name: CI Builds
+name: CI
 
 on:
   pull_request:
@@ -16,11 +16,10 @@ jobs:
     steps:
     - name: Checkout code
       uses: actions/checkout@v1
-    - name: Setup .NET Core
+    - name: Install .NET SDK
       uses: actions/setup-dotnet@v1
       with:
         dotnet-version: | 
-          3.1.x
           6.0.x
     - name: .NET Build
       run: dotnet build Build.csproj -c Release /p:CI=true
@@ -33,24 +32,25 @@ jobs:
       continue-on-error: true
       if: success() || failure()
       with:
-        name: StackExchange.Redis.Tests (Ubuntu) - Results
+        name: Test Results - Ubuntu
         path: 'test-results/*.trx'
         reporter: dotnet-trx
     - name: .NET Lib Pack
       run: dotnet pack src/StackExchange.Redis/StackExchange.Redis.csproj --no-build -c Release /p:Packing=true /p:PackageOutputPath=%CD%\.nupkgs /p:CI=true
 
   windows:
-    name: StackExchange.Redis (Windows Server 2019)
-    runs-on: windows-2019
+    name: StackExchange.Redis (Windows Server 2022)
+    runs-on: windows-2022
+    env:
+      NUGET_CERT_REVOCATION_MODE: offline # Disabling signing because of massive perf hit, see https://github.com/NuGet/Home/issues/11548
     steps:
     - name: Checkout code
       uses: actions/checkout@v1
-    - name: Setup .NET Core 3.x
-      uses: actions/setup-dotnet@v1
-      with:
-        dotnet-version: | 
-          3.1.x
-          6.0.x
+    # - name: Install .NET SDK
+    #   uses: actions/setup-dotnet@v1
+    #   with:
+    #     dotnet-version: | 
+    #       6.0.x
     - name: .NET Build
       run: dotnet build Build.csproj -c Release /p:CI=true
     - name: Start Redis Services (v3.0.503)
@@ -79,6 +79,6 @@ jobs:
       continue-on-error: true
       if: success() || failure()
       with:
-        name: StackExchange.Redis.Tests (Windows Server 2019) - Results
+        name: Tests Results - Windows Server 2022
         path: 'test-results/*.trx'
-        reporter: dotnet-trx      
+        reporter: dotnet-trx

appveyor.yml

@@ -6,8 +6,6 @@ init:
 
 install:
 - cmd: >-
-    choco install dotnet-sdk --version 5.0.404
-
     choco install dotnet-sdk --version 6.0.101
 
     cd tests\RedisConfigs\3.0.503

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 sealed 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

@@ -384,7 +384,7 @@ public void ExportConfiguration(Stream destination, ExportOptions options = Expo
             }
         }
 
-        internal void MakeMaster(ServerEndPoint server, ReplicationChangeOptions options, LogProxy log)
+        internal async Task MakePrimaryAsync(ServerEndPoint server, ReplicationChangeOptions options, LogProxy log)
         {
             var cmd = server.GetFeatures().ReplicaCommands ? RedisCommand.REPLICAOF : RedisCommand.SLAVEOF;
             CommandMap.AssertAvailable(cmd);
@@ -395,23 +395,21 @@ internal void MakeMaster(ServerEndPoint server, ReplicationChangeOptions options
             var srv = new RedisServer(this, server, null);
             if (!srv.IsConnected) throw ExceptionFactory.NoConnectionAvailable(this, null, server, GetServerSnapshot(), command: cmd);
 
-#pragma warning disable CS0618
-            const CommandFlags flags = CommandFlags.NoRedirect | CommandFlags.HighPriority;
-#pragma warning restore CS0618
+            const CommandFlags flags = CommandFlags.NoRedirect;
             Message msg;
 
             log?.WriteLine($"Checking {Format.ToString(srv.EndPoint)} is available...");
             try
             {
-                srv.Ping(flags); // if it isn't happy, we're not happy
+                await srv.PingAsync(flags); // if it isn't happy, we're not happy
             }
             catch (Exception ex)
             {
                 log?.WriteLine($"Operation failed on {Format.ToString(srv.EndPoint)}, aborting: {ex.Message}");
                 throw;
             }
 
-            var nodes = GetServerSnapshot();
+            var nodes = GetServerSnapshot().ToArray(); // Have to array because async/await
             RedisValue newMaster = Format.ToString(server.EndPoint);
 
             RedisKey tieBreakerKey = default(RedisKey);
@@ -423,20 +421,22 @@ internal void MakeMaster(ServerEndPoint server, ReplicationChangeOptions options
 
                 foreach (var node in nodes)
                 {
-                    if (!node.IsConnected) continue;
+                    if (!node.IsConnected || node.IsReplica) continue;
                     log?.WriteLine($"Attempting to set tie-breaker on {Format.ToString(node.EndPoint)}...");
-                    msg = Message.Create(0, flags, RedisCommand.SET, tieBreakerKey, newMaster);
-#pragma warning disable CS0618
-                    node.WriteDirectFireAndForgetSync(msg, ResultProcessor.DemandOK);
-#pragma warning restore CS0618
+                    msg = Message.Create(0, flags | CommandFlags.FireAndForget, RedisCommand.SET, tieBreakerKey, newMaster);
+                    try
+                    {
+                        await node.WriteDirectAsync(msg, ResultProcessor.DemandOK);
+                    }
+                    catch { }
                 }
             }
 
             // stop replicating, promote to a standalone primary
             log?.WriteLine($"Making {Format.ToString(srv.EndPoint)} a master...");
             try
             {
-                srv.ReplicaOf(null, flags);
+                await srv.ReplicaOfAsync(null, flags);
             }
             catch (Exception ex)
             {
@@ -445,13 +445,15 @@ internal void MakeMaster(ServerEndPoint server, ReplicationChangeOptions options
             }
 
             // also, in case it was a replica a moment ago, and hasn't got the tie-breaker yet, we re-send the tie-breaker to this one
-            if (!tieBreakerKey.IsNull)
+            if (!tieBreakerKey.IsNull && !server.IsReplica)
             {
                 log?.WriteLine($"Resending tie-breaker to {Format.ToString(server.EndPoint)}...");
-                msg = Message.Create(0, flags, RedisCommand.SET, tieBreakerKey, newMaster);
-#pragma warning disable CS0618
-                server.WriteDirectFireAndForgetSync(msg, ResultProcessor.DemandOK);
-#pragma warning restore CS0618
+                msg = Message.Create(0, flags | CommandFlags.FireAndForget, RedisCommand.SET, tieBreakerKey, newMaster);
+                try
+                {
+                    await server.WriteDirectAsync(msg, ResultProcessor.DemandOK);
+                }
+                catch { }
             }
 
             // There's an inherent race here in zero-latency environments (e.g. when Redis is on localhost) when a broadcast is specified
@@ -465,7 +467,7 @@ internal void MakeMaster(ServerEndPoint server, ReplicationChangeOptions options
             // We want everyone possible to pick it up.
             // We broadcast before *and after* the change to remote members, so that they don't go without detecting a change happened.
             // This eliminates the race of pub/sub *then* re-slaving happening, since a method both precedes and follows.
-            void Broadcast(ReadOnlySpan<ServerEndPoint> serverNodes)
+            async Task BroadcastAsync(ServerEndPoint[] serverNodes)
             {
                 if ((options & ReplicationChangeOptions.Broadcast) != 0 && ConfigurationChangedChannel != null
                     && CommandMap.IsAvailable(RedisCommand.PUBLISH))
@@ -475,16 +477,14 @@ void Broadcast(ReadOnlySpan<ServerEndPoint> serverNodes)
                     {
                         if (!node.IsConnected) continue;
                         log?.WriteLine($"Broadcasting via {Format.ToString(node.EndPoint)}...");
-                        msg = Message.Create(-1, flags, RedisCommand.PUBLISH, channel, newMaster);
-#pragma warning disable CS0618
-                        node.WriteDirectFireAndForgetSync(msg, ResultProcessor.Int64);
-#pragma warning restore CS0618
+                        msg = Message.Create(-1, flags | CommandFlags.FireAndForget, RedisCommand.PUBLISH, channel, newMaster);
+                        await node.WriteDirectAsync(msg, ResultProcessor.Int64);
                     }
                 }
             }
 
             // Send a message before it happens - because afterwards a new replica may be unresponsive
-            Broadcast(nodes);
+            await BroadcastAsync(nodes);
 
             if ((options & ReplicationChangeOptions.ReplicateToOtherEndpoints) != 0)
             {
@@ -494,16 +494,14 @@ void Broadcast(ReadOnlySpan<ServerEndPoint> serverNodes)
 
                     log?.WriteLine($"Replicating to {Format.ToString(node.EndPoint)}...");
                     msg = RedisServer.CreateReplicaOfMessage(node, server.EndPoint, flags);
-#pragma warning disable CS0618
-                    node.WriteDirectFireAndForgetSync(msg, ResultProcessor.DemandOK);
-#pragma warning restore CS0618
+                    await node.WriteDirectAsync(msg, ResultProcessor.DemandOK);
                 }
             }
 
             // ...and send one after it happens - because the first broadcast may have landed on a secondary client
             // and it can reconfigure before any topology change actually happened. This is most likely to happen
             // in low-latency environments.
-            Broadcast(nodes);
+            await BroadcastAsync(nodes);
 
             // and reconfigure the muxer
             log?.WriteLine("Reconfiguring all endpoints...");
@@ -513,7 +511,7 @@ void Broadcast(ReadOnlySpan<ServerEndPoint> serverNodes)
             {
                 Interlocked.Exchange(ref activeConfigCause, null);
             }
-            if (!ReconfigureAsync(first: false, reconfigureAll: true, log, srv.EndPoint, "make master").ObserveErrors().Wait(5000))
+            if (!await ReconfigureAsync(first: false, reconfigureAll: true, log, srv.EndPoint, "make master"))
             {
                 log?.WriteLine("Verifying the configuration was incomplete; please verify");
             }
@@ -804,15 +802,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)
                     {
@@ -2169,6 +2167,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?
             {
@@ -2186,7 +2190,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;

src/StackExchange.Redis/Enums/CommandFlags.cs

@@ -18,7 +18,7 @@ public enum CommandFlags
         /// <summary>
         /// From 2.0, this flag is not used
         /// </summary>
-        [Obsolete("From 2.0, this flag is not used", false)]
+        [Obsolete("From 2.0, this flag is not used, this will be removed in 3.0.", false)]
         HighPriority = 1,
         /// <summary>
         /// The caller is not interested in the result; the caller will immediately receive a default-value
@@ -81,7 +81,7 @@ public enum CommandFlags
         /// </summary>
         NoScriptCache = 512,
 
-        // 1024: used for timed-out; never user-specified, so not visible on the public API
+        // 1024: Removed - was used for async timeout checks; 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
     }

src/StackExchange.Redis/Interfaces/IServer.cs

@@ -418,12 +418,20 @@ public partial interface IServer : IRedis
         Task<DateTime> LastSaveAsync(CommandFlags flags = CommandFlags.None);
 
         /// <summary>
-        /// Promote the selected node to be master
+        /// Promote the selected node to be primary.
         /// </summary>
         /// <param name="options">The options to use for this topology change.</param>
         /// <param name="log">The log to write output to.</param>
+        [Obsolete("Please use " + nameof(MakePrimaryAsync) + ", this will be removed in 3.0.")]
         void MakeMaster(ReplicationChangeOptions options, TextWriter log = null);
 
+        /// <summary>
+        /// Promote the selected node to be primary.
+        /// </summary>
+        /// <param name="options">The options to use for this topology change.</param>
+        /// <param name="log">The log to write output to.</param>
+        Task MakePrimaryAsync(ReplicationChangeOptions options, TextWriter log = null);
+
         /// <summary>
         /// Returns the role info for the current server.
         /// </summary>
@@ -540,7 +548,7 @@ public partial interface IServer : IRedis
         /// <param name="master">Endpoint of the new master to replicate from.</param>
         /// <param name="flags">The command flags to use.</param>
         /// <remarks>https://redis.io/commands/replicaof</remarks>
-        [Obsolete("Starting with Redis version 5, Redis has moved to 'replica' terminology. Please use " + nameof(ReplicaOf) + " instead.")]
+        [Obsolete("Starting with Redis version 5, Redis has moved to 'replica' terminology. Please use " + nameof(ReplicaOfAsync) + " instead, this will be removed in 3.0.")]
         [Browsable(false), EditorBrowsable(EditorBrowsableState.Never)]
         void SlaveOf(EndPoint master, CommandFlags flags = CommandFlags.None);
 
@@ -550,6 +558,7 @@ public partial interface IServer : IRedis
         /// <param name="master">Endpoint of the new master to replicate from.</param>
         /// <param name="flags">The command flags to use.</param>
         /// <remarks>https://redis.io/commands/replicaof</remarks>
+        [Obsolete("Please use " + nameof(ReplicaOfAsync) + ", this will be removed in 3.0.")]
         void ReplicaOf(EndPoint master, CommandFlags flags = CommandFlags.None);
 
         /// <summary>
@@ -558,7 +567,7 @@ public partial interface IServer : IRedis
         /// <param name="master">Endpoint of the new master to replicate from.</param>
         /// <param name="flags">The command flags to use.</param>
         /// <remarks>https://redis.io/commands/replicaof</remarks>
-        [Obsolete("Starting with Redis version 5, Redis has moved to 'replica' terminology. Please use " + nameof(ReplicaOfAsync) + " instead.")]
+        [Obsolete("Starting with Redis version 5, Redis has moved to 'replica' terminology. Please use " + nameof(ReplicaOfAsync) + " instead, this will be removed in 3.0.")]
         [Browsable(false), EditorBrowsable(EditorBrowsableState.Never)]
         Task SlaveOfAsync(EndPoint master, CommandFlags flags = CommandFlags.None);