Sentinel: discover and recover without blocking thread-pool workers

docs/Configuration.md

@@ -46,7 +46,13 @@ connection to the primary is lost, and - because the announcement is delivered o
 notices when the node last named as primary is still connected but now reports itself as a replica (a failover where nothing disconnected,
 such as a primary demoted in place). In that last case the client re-asks sentinel for the current primary and follows its answer, even to
 a node at an address it has never seen; while sentinel still names the demoted node (sentinel itself has not yet failed over), the client
-keeps asking without tearing down or reconfiguring the healthy connections.
+keeps asking without tearing down or reconfiguring the healthy connections. A command answered with `READONLY` is the same signal: the node
+is marked as a replica at once, rather than at the next role check (`ConfigCheckSeconds`), and the client asks sentinel on its next heartbeat.
+Sentinel is queried asynchronously: waiting on a slow or unreachable sentinel holds no thread-pool thread, and a sentinel query waits up to
+`AsyncTimeout`, also when the connection is made with the synchronous `Connect`. A sentinel whose `SENTINEL MASTER` reply holds the primary
+down (`s_down`/`o_down`) is passed over for one that does not. A sentinel without a usable `SENTINEL MASTER` reply - the subcommand is
+unknown or refused, such as to an ACL user allowed only `get-master-addr-by-name`, or the reply has no flags - is asked with
+`get-master-addr-by-name` instead.
 
 An overview of mapping between the `string` and `ConfigurationOptions` representation is shown below, but you can switch between them trivially:
 

src/StackExchange.Redis/ConnectionMultiplexer.cs

@@ -642,23 +642,34 @@ static async Task<ConnectionMultiplexer> ApplyAfterConnectAsync(Task<ConnectionM
             }
         }
 
-        private static async Task<ConnectionMultiplexer> ConnectImplAsync(ConfigurationOptions configuration, TextWriter? writer = null, ServerType? serverType = null, Availability.CircuitBreaker? groupCircuitBreaker = null, bool isGroupMember = false)
+        private static async Task<ConnectionMultiplexer> ConnectImplAsync(ConfigurationOptions configuration, TextWriter? writer = null, ServerType? serverType = null, Availability.CircuitBreaker? groupCircuitBreaker = null, bool isGroupMember = false, EndPointCollection? endpoints = null, bool synchronousConnect = false)
         {
-            IDisposable? killMe = null;
+            ConnectionMultiplexer? killMe = null;
             EventHandler<ConnectionFailedEventArgs>? connectHandler = null;
             ConnectionMultiplexer? muxer = null;
             var configLogger = configuration.LoggerFactory?.CreateLogger<ConnectionMultiplexer>();
             var log = configLogger.With(writer);
             try
             {
                 var sw = ValueStopwatch.StartNew();
-                log?.LogInformationConnectingAsync(RuntimeInformation.FrameworkDescription, Utils.GetLibVersion());
+                if (synchronousConnect) log?.LogInformationConnectingSync(RuntimeInformation.FrameworkDescription, Utils.GetLibVersion());
+                else log?.LogInformationConnectingAsync(RuntimeInformation.FrameworkDescription, Utils.GetLibVersion());
 
-                muxer = CreateMultiplexer(configuration, log, serverType, out connectHandler, groupCircuitBreaker: groupCircuitBreaker, isGroupMember: isGroupMember);
+                muxer = CreateMultiplexer(configuration, log, serverType, out connectHandler, endpoints: endpoints, groupCircuitBreaker: groupCircuitBreaker, isGroupMember: isGroupMember);
                 killMe = muxer;
                 Interlocked.Increment(ref muxer._connectAttemptCount);
-                bool configured = await muxer.ReconfigureAsync(first: true, reconfigureAll: false, log, null, "connect").ObserveErrors().ForAwait();
-                if (!configured)
+                var configure = muxer.ReconfigureAsync(first: true, reconfigureAll: false, log, null, "connect").ObserveErrors();
+                if (synchronousConnect && !configure.IsCompleted
+                    && !await configure.TimeoutAfter(muxer.SyncConnectTimeout(true)).ForAwait())
+                {
+                    // The public sync Sentinel boundary retains ConnectImpl's soft timeout, without blocking a
+                    // worker inside the shared async core. Reconfigure keeps running on the returned multiplexer.
+                    var ex = ExceptionFactory.UnableToConnect(muxer, "ConnectTimeout");
+                    if (configuration.AbortOnConnectFail) throw ex;
+                    muxer.LastException = ex;
+                    muxer.Logger?.LogErrorSyncConnectTimeout(ex, ex.Message);
+                }
+                else if (!await configure.ForAwait())
                 {
                     throw ExceptionFactory.UnableToConnect(muxer, muxer.failureMessage);
                 }
@@ -671,7 +682,9 @@ private static async Task<ConnectionMultiplexer> ConnectImplAsync(ConfigurationO
                     muxer.InitializeSentinel(log);
                 }
 
-                await configuration.AfterConnectAsync(muxer, s => log?.LogInformationAfterConnect(s)).ForAwait();
+                var afterConnect = configuration.AfterConnectAsync(muxer, s => log?.LogInformationAfterConnect(s)).ObserveErrors();
+                if (synchronousConnect) await afterConnect.TimeoutAfter(muxer.SyncConnectTimeout(true)).ForAwait();
+                else await afterConnect.ForAwait();
 
                 log?.LogInformationTotalConnectTime(sw.ElapsedMilliseconds);
 
@@ -680,7 +693,7 @@ private static async Task<ConnectionMultiplexer> ConnectImplAsync(ConfigurationO
             finally
             {
                 if (connectHandler != null && muxer != null) muxer.ConnectionFailed -= connectHandler;
-                if (killMe != null) try { killMe.Dispose(); } catch { }
+                if (killMe != null) try { await killMe.DisposeAsync().ForAwait(); } catch { }
                 if (log is TextWriterLogger twLogger) twLogger.Release();
             }
         }
@@ -1765,7 +1778,7 @@ internal static void ActivateServer(ServerEndPoint server, ILogger? log)
 
         /// <summary>
         /// Claims the reconfigure gate without running a reconfigure, so that work that must not interleave with
-        /// one (such as <see cref="SwitchPrimary"/> rewriting the endpoint list) can make its pre-checks and its
+        /// one (such as <see cref="SwitchPrimaryAsync"/> rewriting the endpoint list) can make its pre-checks and its
         /// side effects atomic with respect to competing reconfigures. A successful claim MUST be handed to
         /// <see cref="ReconfigureAsync(bool, bool, ILogger, EndPoint, string, bool, CommandFlags, bool)"/> with
         /// <c>claimed: true</c> (which releases it however it exits), or released via

src/StackExchange.Redis/ResultProcessor.cs

@@ -308,6 +308,15 @@ private bool HandleCommonError(Message message, RespReader reader, PhysicalConne
                 case RedisErrorKind.WrongPass:
                     bridge?.Multiplexer.SetAuthSuspect(new RedisServerException(errorKind, message.Flags, reader.GetOverview()));
                     break;
+                case RedisErrorKind.ReadOnly:
+                    // Sentinel-managed writes can reveal a demotion before INFO. Only mark the local role here;
+                    // the sentinel owner's heartbeat resolves the primary (sentinel-clients: Handle reconnections).
+                    // Standalone/cluster keep their existing error behavior, and no reconfigure starts on this IO path.
+                    if (bridge?.Multiplexer.IsSentinelManaged == true && bridge.ServerEndPoint is { IsReplica: false } demoted)
+                    {
+                        demoted.MarkAsReplicaWithoutReconfigure();
+                    }
+                    break;
             }
 
             var server = bridge?.ServerEndPoint;

src/StackExchange.Redis/ServerEndPoint.cs

@@ -241,6 +241,12 @@ public bool IsReplica
             set => SetConfig(ref isReplica, value);
         }
 
+        internal void MarkAsReplicaWithoutReconfigure()
+        {
+            Volatile.Write(ref isReplica, true);
+            ClearMemoized();
+        }
+
         public bool ReplicaReadOnly
         {
             get => replicaReadOnly;
@@ -1174,6 +1180,20 @@ internal bool CheckInfoReplication()
             return false;
         }
 
+        internal async ValueTask<bool> CheckInfoReplicationAsync()
+        {
+            if (!version.IsAtLeast(RedisFeatures.v2_8_0) || !Multiplexer.CommandMap.IsAvailable(RedisCommand.INFO)
+                || GetBridge(ConnectionType.Interactive, false) is not PhysicalBridge bridge) return false;
+
+            var msg = Message.Create(-1, CommandFlags.FireAndForget | CommandFlags.NoRedirect, RedisCommand.INFO, RedisLiterals.replication);
+            msg.SetInternalCall();
+            msg.SetSource(ResultProcessor.AutoConfigure, null);
+            if (await bridge.TryWriteAsync(msg, isReplica, bypassBacklog: true).ForAwait() != WriteResult.Success) return false;
+            lastInfoReplicationCheckTicks = Environment.TickCount;
+            ResetExponentiallyReplicationCheck();
+            return true;
+        }
+
         private int lastInfoReplicationCheckTicks;
         internal volatile int ConfigCheckSeconds;
         [ThreadStatic]

tests/StackExchange.Redis.Tests/InProcessSentinelServer.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.IO;
 using System.Net;
@@ -49,20 +50,88 @@ public EndPoint Primary
     /// </summary>
     public static Tunnel CreateTunnel(params InProcessTestServer[] servers) => new RoutingTunnel(servers);
 
-    private int _primaryQueries;
+    private int _primaryQueries, _peerQueries, _replicaQueries, _legacyQueries;
+    private string _primaryFlags = "master";
 
     /// <summary>How many times a client has asked this sentinel for the primary.</summary>
     public int PrimaryQueries => Volatile.Read(ref _primaryQueries);
 
+    public Action? OnPrimaryQuery { get; set; }
+    public int PeerQueries => Volatile.Read(ref _peerQueries);
+    public int ReplicaQueries => Volatile.Read(ref _replicaQueries);
+    public int LegacyQueries => Volatile.Read(ref _legacyQueries);
+    public string? PrimaryError { get; set; }
+    public string? ReplicaError { get; set; }
+
+    /// <summary>Answer <c>get-master-addr-by-name</c> with nil, as a sentinel that does not (yet) know the primary does.</summary>
+    public bool LegacyAnswersNothing { get; set; }
+    public EndPoint[]? ReplicaAddressesOverride { get; set; }
+    public Func<Task>? BeforePrimaryReplyAsync { get; set; }
+    public Func<Task>? BeforePeerReplyAsync { get; set; }
+
+    protected override ValueTask ClientPauseAsync(RedisClient client, in RedisRequest request)
+    {
+        var pause = base.ClientPauseAsync(client, request);
+        var hook = request.Count > 1 && request.KnownCommand == RedisCommand.SENTINEL
+            ? request.GetString(1).ToLowerInvariant() switch
+            {
+                "master" or "get-master-addr-by-name" => BeforePrimaryReplyAsync,
+                "sentinels" => BeforePeerReplyAsync,
+                _ => null,
+            } : null;
+        return hook is null ? pause : new ValueTask(AwaitPausedReplyAsync(hook, pause));
+
+        static async Task AwaitPausedReplyAsync(Func<Task> hook, ValueTask pause)
+        {
+            // Execute has already captured the response; the barrier controls delivery, not the view being read.
+            await hook();
+            await pause;
+        }
+    }
+
+    /// <summary>Replicas to report with flags, including nodes that are no longer reachable.</summary>
+    public ConcurrentDictionary<EndPoint, string> ReplicaFlags { get; } = new();
+
+    /// <summary>Peer routes to advertise; deliberately no MYID/runid support is needed for address discovery.</summary>
+    public ConcurrentDictionary<EndPoint, string> PeerFlags { get; } = new();
+
+    public string PrimaryFlags
+    {
+        get => Volatile.Read(ref _primaryFlags);
+        set => Volatile.Write(ref _primaryFlags, value);
+    }
+
+    public bool SupportsSentinelMaster { get; set; } = true;
+
+    /// <summary>Answer <c>SENTINEL MASTER</c> without its <c>flags</c> field, as some sentinel-compatible servers do.</summary>
+    public bool OmitPrimaryFlags { get; set; }
+
+    private void CountPrimaryQuery()
+    {
+        Interlocked.Increment(ref _primaryQueries);
+        OnPrimaryQuery?.Invoke();
+    }
+
+    [RedisCommand(3, nameof(RedisCommand.SENTINEL), "master", LockFree = true)]
+    private TypedRedisValue SentinelMaster(RedisClient client, in RedisRequest request)
+    {
+        if (!SupportsSentinelMaster) return TypedRedisValue.Error("ERR Unknown sentinel subcommand 'master'");
+        var reply = PrimaryError is string error ? TypedRedisValue.Error(error)
+            : IsService(request) ? Describe(Primary, OmitPrimaryFlags ? null : PrimaryFlags) : TypedRedisValue.Error("ERR No such master with that name");
+        CountPrimaryQuery();
+        return reply;
+    }
+
     [RedisCommand(3, nameof(RedisCommand.SENTINEL), "get-master-addr-by-name", LockFree = true)]
     private TypedRedisValue SentinelGetMasterAddrByName(RedisClient client, in RedisRequest request)
     {
-        Interlocked.Increment(ref _primaryQueries);
-        if (!IsService(request)) return TypedRedisValue.NullArray(RespPrefix.Array);
+        Interlocked.Increment(ref _legacyQueries);
+        if (!IsService(request) || LegacyAnswersNothing) return TypedRedisValue.NullArray(RespPrefix.Array);
         var host = GetHost(Primary, out var port);
         var reply = TypedRedisValue.Rent(2, out var span, RespPrefix.Array);
         span[0] = TypedRedisValue.BulkString(host);
         span[1] = TypedRedisValue.BulkString(port.ToString());
+        CountPrimaryQuery();
         return reply;
     }
 
@@ -72,16 +141,22 @@ private TypedRedisValue SentinelGetMasterAddrByName(RedisClient client, in Redis
     [RedisCommand(3, nameof(RedisCommand.SENTINEL), "slaves", LockFree = true)]
     private TypedRedisValue SentinelSlaves(RedisClient client, in RedisRequest request) => Replicas(request);
 
-    // only this sentinel exists
     [RedisCommand(3, nameof(RedisCommand.SENTINEL), "sentinels", LockFree = true)]
     private TypedRedisValue SentinelSentinels(RedisClient client, in RedisRequest request)
-        => IsService(request) ? TypedRedisValue.EmptyArray(RespPrefix.Array) : TypedRedisValue.Error("ERR No such master with that name");
+    {
+        Interlocked.Increment(ref _peerQueries);
+        if (!IsService(request)) return TypedRedisValue.Error("ERR No such master with that name");
+        var peers = PeerFlags.ToArray();
+        var reply = TypedRedisValue.Rent(peers.Length, out var span, RespPrefix.Array);
+        for (int i = 0; i < peers.Length; i++) span[i] = Describe(peers[i].Key, "sentinel," + peers[i].Value);
+        return reply;
+    }
 
     [RedisCommand(2, nameof(RedisCommand.SENTINEL), "masters", LockFree = true)]
     private TypedRedisValue SentinelMasters(RedisClient client, in RedisRequest request)
     {
         var reply = TypedRedisValue.Rent(1, out var span, RespPrefix.Array);
-        span[0] = Describe(Primary, "master");
+        span[0] = Describe(Primary, PrimaryFlags);
         return reply;
     }
 
@@ -91,6 +166,8 @@ private TypedRedisValue Replicas(in RedisRequest request)
     {
         if (!IsService(request)) return TypedRedisValue.Error("ERR No such master with that name");
 
+        Interlocked.Increment(ref _replicaQueries);
+        if (ReplicaError is string error) return TypedRedisValue.Error(error);
         // sentinel learns the replicas from the primary, so report whatever replicates it right now
         var replicas = new List<EndPoint>();
         if (_monitored.TryGetNode(Primary, out var primaryNode))
@@ -104,26 +181,37 @@ private TypedRedisValue Replicas(in RedisRequest request)
             }
         }
 
+        if (ReplicaAddressesOverride is { } addresses)
+        {
+            replicas.Clear();
+            replicas.AddRange(addresses);
+        }
+        // Real sentinel keeps reporting failed replicas with down/disconnected flags until reset.
+        foreach (var endpoint in ReplicaFlags.Keys)
+            if (!replicas.Contains(endpoint)) replicas.Add(endpoint);
         var reply = TypedRedisValue.Rent(replicas.Count, out var span, RespPrefix.Array);
         for (int i = 0; i < replicas.Count; i++)
         {
-            span[i] = Describe(replicas[i], "slave");
+            span[i] = Describe(replicas[i], ReplicaFlags.TryGetValue(replicas[i], out var flags) ? "slave," + flags : "slave");
         }
         return reply;
     }
 
-    private TypedRedisValue Describe(EndPoint endpoint, string flags)
+    private TypedRedisValue Describe(EndPoint endpoint, string? flags)
     {
         var host = GetHost(endpoint, out var port);
-        var item = TypedRedisValue.Rent(8, out var span, RespPrefix.Array);
+        var item = TypedRedisValue.Rent(flags is null ? 6 : 8, out var span, RespPrefix.Array);
         span[0] = TypedRedisValue.BulkString("name");
-        span[1] = TypedRedisValue.BulkString(flags == "master" ? ServiceName : $"{host}:{port}");
+        span[1] = TypedRedisValue.BulkString(flags is null || flags.StartsWith("master", StringComparison.Ordinal) ? ServiceName : $"{host}:{port}");
         span[2] = TypedRedisValue.BulkString("ip");
         span[3] = TypedRedisValue.BulkString(host);
         span[4] = TypedRedisValue.BulkString("port");
         span[5] = TypedRedisValue.BulkString(port.ToString());
-        span[6] = TypedRedisValue.BulkString("flags");
-        span[7] = TypedRedisValue.BulkString(flags);
+        if (flags is not null)
+        {
+            span[6] = TypedRedisValue.BulkString("flags");
+            span[7] = TypedRedisValue.BulkString(flags);
+        }
         return item;
     }
 
@@ -139,9 +227,9 @@ private sealed class RoutingTunnel(InProcessTestServer[] servers) : Tunnel
         }
 
         public override ValueTask<EndPoint?> GetSocketConnectEndpointAsync(EndPoint endpoint, CancellationToken cancellationToken)
-            => Route(endpoint) is { } tunnel ? tunnel.GetSocketConnectEndpointAsync(endpoint, cancellationToken) : base.GetSocketConnectEndpointAsync(endpoint, cancellationToken);
+            => Route(endpoint) is { } tunnel ? tunnel.GetSocketConnectEndpointAsync(endpoint, cancellationToken) : throw new IOException("No in-process route for " + endpoint);
 
         public override ValueTask<Stream?> BeforeAuthenticateAsync(EndPoint endpoint, ConnectionType connectionType, Socket? socket, CancellationToken cancellationToken)
-            => Route(endpoint) is { } tunnel ? tunnel.BeforeAuthenticateAsync(endpoint, connectionType, socket, cancellationToken) : base.BeforeAuthenticateAsync(endpoint, connectionType, socket, cancellationToken);
+            => Route(endpoint) is { } tunnel ? tunnel.BeforeAuthenticateAsync(endpoint, connectionType, socket, cancellationToken) : throw new IOException("No in-process route for " + endpoint);
     }
 }