From 984af1811ea9581e00e16c439fa6a4d394ef6004 Mon Sep 17 00:00:00 2001 From: rebaseandpanic Date: Mon, 26 Oct 2026 10:26:53 +0000 Subject: [PATCH 1/4] fix(sentinel): discover and recover without blocking thread-pool workers A sentinel-managed connection that loses its primary arms a reconnect timer that fires every second until OnManagedConnectionRestored stops it. Each tick ran SwitchPrimary, which asked every sentinel through a PLINQ query of synchronous commands: one blocked pool worker per sentinel endpoint, waiting on socket completions that are themselves pool work items. With a handful of sentinel endpoints this takes the whole pool, and the process stalls until the pool injects threads - every second, for as long as the timer runs. The timer can also run forever: a node that never returns at its old address, or a Restored delivered before its Failed, leaves nothing to stop it. - Sentinel queries use the async API, asking every sentinel at once with one deadline for the whole query; the first answer wins and the rest are only observed. No worker waits on a sentinel. - SwitchPrimary becomes SwitchPrimaryAsync, with a per-connection gate around discovery and reconfiguration. A switch requested while one is running is recorded and replayed instead of overlapping it. - The reconnect timer is one-shot: the next attempt is armed one second after the previous one finished. It stops on state - the sentinel-reported primary is connected, is not a replica, and no other node is a connected primary - rather than on the order Failed/Restored arrive in. - The +switch-master, +sentinel, lost-sentinel, restore and demotion paths start async work instead of blocking their caller; demotion recovery sends INFO through TryWriteAsync, and a write that did not go out is not taken as a role check. - GetSentinelMasterConnection and the async connect share one async core. The public synchronous Connect keeps its soft ConnectTimeout behaviour with AbortOnConnectFail=false, and is the only remaining synchronous wait. - Two concurrent GetSentinelMasterConnection calls for one service return the same child; the losing connection is disposed. - Restored no longer runs a switch unconditionally: a connection already on the sentinel-reported primary, with no stale view and no recovery running, is left alone; anything else schedules the same recovery as Failed. - The sentinel address list is refreshed after the first switch of a managed connection as well, not only after a switch that reconfigured it. Which sentinel answer the client accepts, and how it reconciles replicas, is unchanged here. Spec: https://redis.io/docs/latest/develop/reference/sentinel-clients/#step-1-connect-to-the-first-sentinel https://redis.io/docs/latest/develop/reference/sentinel-clients/#step-2-ask-for-the-master-address https://redis.io/docs/latest/develop/reference/sentinel-clients/#handle-reconnections https://valkey.io/topics/sentinel-clients/#step-1-connect-to-the-first-sentinel https://valkey.io/topics/sentinel-clients/#step-2-ask-for-primary-address https://valkey.io/topics/sentinel-clients/#handling-reconnections --- .../ConnectionMultiplexer.Sentinel.cs | 516 ++++++++++-------- .../ConnectionMultiplexer.cs | 31 +- src/StackExchange.Redis/ServerEndPoint.cs | 14 + .../InProcessSentinelServer.cs | 95 +++- .../SentinelDiscoveryUnitTests.cs | 34 ++ .../SentinelPrimaryDemotionUnitTests.cs | 306 ++++++++++- .../SentinelTests.cs | 2 +- 7 files changed, 739 insertions(+), 259 deletions(-) create mode 100644 tests/StackExchange.Redis.Tests/SentinelDiscoveryUnitTests.cs diff --git a/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs b/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs index 31f5296ce..174a02f9a 100644 --- a/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs +++ b/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs @@ -19,6 +19,8 @@ public partial class ConnectionMultiplexer internal ConnectionMultiplexer? sentinelConnection; private int _sentinelInitialized; private int _demotedPrimaryRecoveryActive; + private int _sentinelSwitchActive; + private int _sentinelSwitchPending; /// /// Initializes the connection as a Sentinel connection and adds the necessary event handlers to track changes to the managed primaries. @@ -50,12 +52,8 @@ internal void InitializeSentinel(ILogger? log) // We don't care about the result of this - we're just trying _ = Format.TryParseEndPoint(string.Format("{0}:{1}", messageParts[1], messageParts[2]), out var switchBlame); - // Resolve the child under the lock, but run the (potentially multi-second, - // blocking) SwitchPrimary reconfigure OUTSIDE it: otherwise a failover for one - // service would block +switch-master handling and GetSentinelMasterConnection - // for every other service sharing this sentinel connection. SwitchPrimary is - // already invoked without this lock from the reconnect timer and the restore - // handlers, so holding it here is not a required invariant. + // Resolve the child under the lock, then start discovery/reconfiguration outside it: + // services sharing a sentinel connection must not hold up one another's failover. ConnectionMultiplexer? switchTarget = null; lock (sentinelConnectionChildren) { @@ -76,7 +74,7 @@ internal void InitializeSentinel(ILogger? log) if (switchTarget is not null) { - SwitchPrimary(switchBlame, switchTarget); + SwitchPrimaryAsync(switchBlame, switchTarget).RedisFireAndForget(); } }, CommandFlags.FireAndForget); @@ -86,7 +84,7 @@ internal void InitializeSentinel(ILogger? log) // we need to reconfigure to make sure we still have a subscription to the +switch-master channel ConnectionFailed += (sender, e) => // Reconfigure to get subscriptions back online - ReconfigureAsync(first: false, reconfigureAll: true, log, e.EndPoint, "Lost sentinel connection", false).Wait(); + ReconfigureAsync(first: false, reconfigureAll: true, log, e.EndPoint, "Lost sentinel connection", false).RedisFireAndForget(); // Subscribe to new sentinels being added if (sub.SubscribedEndpoint(RedisChannel.Literal("+sentinel")) == null) @@ -96,7 +94,7 @@ internal void InitializeSentinel(ILogger? log) (_, message) => { string[] messageParts = ((string)message!).Split(new[] { ' ' }, StringSplitOptions.RemoveEmptyEntries); - UpdateSentinelAddressList(messageParts[0]); + UpdateSentinelAddressListAsync(messageParts[0]).RedisFireAndForget(); }, CommandFlags.FireAndForget); } @@ -189,7 +187,7 @@ private static async Task SentinelPrimaryConnectAsync(Con var sentinelConnection = await SentinelConnectAsync(sentinelConfig, writer).ForAwait(); try { - var muxer = sentinelConnection.GetSentinelMasterConnection(configuration, writer); + var muxer = await sentinelConnection.GetSentinelMasterConnectionAsync(configuration, writer).ForAwait(); // Set reference to sentinel connection so that we can dispose it muxer.sentinelConnection = sentinelConnection; @@ -197,7 +195,7 @@ private static async Task SentinelPrimaryConnectAsync(Con } catch { - try { sentinelConnection.Dispose(); } catch { } + try { await sentinelConnection.DisposeAsync().ForAwait(); } catch { } throw; } } @@ -208,6 +206,10 @@ private static async Task SentinelPrimaryConnectAsync(Con /// The configuration to be used when connecting to the primary. /// The writer to log to, if any. public ConnectionMultiplexer GetSentinelMasterConnection(ConfigurationOptions config, TextWriter? log = null) + // The public synchronous connect boundary only. Background recovery and ConnectAsync call the async core. + => GetSentinelMasterConnectionAsync(config, log, synchronousConnect: true).GetAwaiter().GetResult(); + + private async Task GetSentinelMasterConnectionAsync(ConfigurationOptions config, TextWriter? log, bool synchronousConnect = false) { // A soft-failed SentinelConnect cannot detect a server type, but _isSentinel still records its intent. if (ServerSelectionStrategy.ServerType != ServerType.Sentinel && (config.AbortOnConnectFail || !_isSentinel)) @@ -246,7 +248,7 @@ public ConnectionMultiplexer GetSentinelMasterConnection(ConfigurationOptions co EndPoint? newPrimaryEndPoint = null; for (int i = 0; i < queryAttempts && newPrimaryEndPoint is null; i++) { - newPrimaryEndPoint = GetConfiguredPrimaryForService(serviceName); + newPrimaryEndPoint = await GetConfiguredPrimaryForServiceAsync(serviceName).ForAwait(); } if (newPrimaryEndPoint is null) @@ -260,15 +262,15 @@ public ConnectionMultiplexer GetSentinelMasterConnection(ConfigurationOptions co throw failure; } - if (connection is not null) try { connection.Dispose(); } catch { } - connection = SeedConnection(failure); + if (connection is not null) try { await connection.DisposeAsync().ForAwait(); } catch { } + connection = await SeedConnectionAsync(failure).ForAwait(); break; } EndPoint[]? replicaEndPoints = null; for (int i = 0; i < queryAttempts && replicaEndPoints is null; i++) { - replicaEndPoints = GetReplicasForService(serviceName); + replicaEndPoints = await GetReplicasForServiceAsync(serviceName).ForAwait(); } endpoints = config.EndPoints.Clone(); @@ -293,8 +295,16 @@ public ConnectionMultiplexer GetSentinelMasterConnection(ConfigurationOptions co } } - if (connection is not null) try { connection.Dispose(); } catch { } - connection = ConnectImpl(config, log, endpoints: endpoints); + if (connection is not null) try { await connection.DisposeAsync().ForAwait(); } catch { } + connection = await ConnectImplAsync(config, log, endpoints: endpoints, synchronousConnect: synchronousConnect).ForAwait(); + + // Preserve the synchronous Connect soft timeout: a disconnected multiplexer is returned while its + // initial reconfigure continues; ROLE and the initial switch are recovery work in this case. + if (synchronousConnect && !config.AbortOnConnectFail && !connection.IsConnected) + { + failure = connection.LastException as RedisConnectionException; + break; + } // verify role is primary according to: // https://redis.io/topics/sentinel-clients @@ -306,7 +316,7 @@ public ConnectionMultiplexer GetSentinelMasterConnection(ConfigurationOptions co try { isPrimary = connection.CommandMap.IsAvailable(RedisCommand.ROLE) - ? server.Role()?.Value == Role.LabelForMaster + ? (await server.RoleAsync().ForAwait())?.Value == Role.LabelForMaster : !server.IsReplica; } catch @@ -322,7 +332,7 @@ public ConnectionMultiplexer GetSentinelMasterConnection(ConfigurationOptions co } } - Thread.Sleep(100); + await Task.Delay(100).ForAwait(); } while (sw.ElapsedMilliseconds < config.ConnectTimeout); @@ -337,7 +347,7 @@ public ConnectionMultiplexer GetSentinelMasterConnection(ConfigurationOptions co throw failure; } - connection ??= SeedConnection(failure); + connection ??= await SeedConnectionAsync(failure).ForAwait(); connection.LastException = failure; connection.Logger?.LogErrorSyncConnectTimeout(failure, failure.Message); @@ -349,16 +359,31 @@ public ConnectionMultiplexer GetSentinelMasterConnection(ConfigurationOptions co // If we lost the connection, run a switch to a least try and get updated info about the primary connection.ConnectionFailed += OnManagedConnectionFailed; + ConnectionMultiplexer? existing = null; lock (sentinelConnectionChildren) { - sentinelConnectionChildren[serviceName] = connection; + if (sentinelConnectionChildren.TryGetValue(serviceName, out var child) && !child.IsDisposed) + existing = child; + else + sentinelConnectionChildren[serviceName] = connection; + } + if (existing is not null) + { + connection.ConnectionRestored -= OnManagedConnectionRestored; + connection.ConnectionFailed -= OnManagedConnectionFailed; + try { await connection.DisposeAsync().ForAwait(); } catch { } + connection = null; + return existing; } // Perform the initial switchover var switchBlame = endpoints is { Count: > 0 } ? endpoints[0] : null; try { - SwitchPrimary(switchBlame, connection, log); + if (synchronousConnect && !config.AbortOnConnectFail && !connection.IsConnected) + ScheduleSentinelPrimaryReconnect(connection, switchBlame); + else + await SwitchPrimaryAsync(switchBlame, connection, log).ForAwait(); } catch when (!config.AbortOnConnectFail) { @@ -366,7 +391,7 @@ public ConnectionMultiplexer GetSentinelMasterConnection(ConfigurationOptions co // hold however far we got. Gating this on !success would tear down a connection that had // *successfully* reached its primary just because the follow-up switchover threw, and throw // from a call the caller asked never to throw. - ScheduleSentinelPrimaryReconnect(connection, switchBlame); + // SwitchPrimaryAsync's finally has already scheduled recovery. } return connection; @@ -384,7 +409,7 @@ public ConnectionMultiplexer GetSentinelMasterConnection(ConfigurationOptions co sentinelConnectionChildren.Remove(serviceName); } } - try { connection.Dispose(); } catch { } + try { await connection.DisposeAsync().ForAwait(); } catch { } } throw; } @@ -393,127 +418,89 @@ public ConnectionMultiplexer GetSentinelMasterConnection(ConfigurationOptions co // the caller gave us (the sentinel addresses, via Connect), for SwitchPrimary to rebuild once discovery works. // A config with no endpoints of its own has nothing to seed with, and that is the discovery failure the caller // should see - not the "No nodes to consider" that ConnectImpl would throw at an empty endpoint list. - ConnectionMultiplexer SeedConnection(RedisConnectionException discoveryFailure) + Task SeedConnectionAsync(RedisConnectionException discoveryFailure) { if (config.EndPoints.Count == 0) { throw discoveryFailure; } endpoints = config.EndPoints.Clone(); - return ConnectImpl(config, log, endpoints: endpoints); + return ConnectImplAsync(config, log, endpoints: endpoints, synchronousConnect: synchronousConnect); } } - [System.Diagnostics.CodeAnalysis.SuppressMessage("Roslynator", "RCS1075:Avoid empty catch clause that catches System.Exception.", Justification = "We don't care.")] internal void OnManagedConnectionRestored(object? sender, ConnectionFailedEventArgs e) { - if (sender is not ConnectionMultiplexer connection) + if (sender is ConnectionMultiplexer connection) { - return; // This should never happen - called from non-nullable ConnectionFailedEventArgs - } - - var oldTimer = Interlocked.Exchange(ref connection.sentinelPrimaryReconnectTimer, null); - oldTimer?.Dispose(); - - try - { - // Run a switch to make sure we have update-to-date - // information about which primary we should connect to - SwitchPrimary(e.EndPoint, connection); - - try - { - // Verify that the reconnected endpoint is a primary, - // and the correct one otherwise we should reconnect - if (connection.GetServer(e.EndPoint).IsReplica || e.EndPoint != connection.currentSentinelPrimaryEndPoint) - { - // This isn't a primary, so try connecting again - SwitchPrimary(e.EndPoint, connection); - } - } - catch (Exception) - { - // If we get here it means that we tried to reconnect to a server that is no longer - // considered a primary by Sentinel and was removed from the list of endpoints. - - // If we caught an exception, we may have gotten a stale endpoint - // we are not aware of, so retry - SwitchPrimary(e.EndPoint, connection); - } - - // A soft-failed connect (AbortOnConnectFail=false) records the discovery failure as LastException, and - // the exception factory folds that into every "no connection available" message. We are connected to a - // sentinel-approved primary again, so stop reporting it. - connection.LastException = null; - } - catch (Exception) - { - // Log, but don't throw in an event handler - // TODO: Log via new event handler? a la ConnectionFailed? + // Failed/Restored are delivered on workers and may arrive out of order. Both ask for a fresh + // sentinel view; only that view plus the current connection state can stop recovery. + if (Volatile.Read(ref connection.sentinelPrimaryReconnectTimer) is null + && Volatile.Read(ref connection._sentinelSwitchActive) == 0 + && connection.currentSentinelPrimaryEndPoint is EndPoint primary + && !IsStalePrimaryView(connection, primary)) return; + ScheduleSentinelPrimaryReconnect(connection, e.EndPoint); } } internal void OnManagedConnectionFailed(object? sender, ConnectionFailedEventArgs e) { - if (sender is not ConnectionMultiplexer connection) + if (sender is ConnectionMultiplexer connection) { - return; // This should never happen - called from non-nullable ConnectionFailedEventArgs + ScheduleSentinelPrimaryReconnect(connection, e.EndPoint); } - - ScheduleSentinelPrimaryReconnect(connection, e.EndPoint); } - [System.Diagnostics.CodeAnalysis.SuppressMessage("Roslynator", "RCS1075:Avoid empty catch clause that catches System.Exception.", Justification = "We don't care.")] + private static void StopSentinelPrimaryReconnect(ConnectionMultiplexer connection) + => Interlocked.Exchange(ref connection.sentinelPrimaryReconnectTimer, null)?.Dispose(); + private void ScheduleSentinelPrimaryReconnect(ConnectionMultiplexer connection, EndPoint? endPoint) { - // Periodically check to see if we can reconnect to the proper primary. - // This is here in case we lost our subscription to a good sentinel instance - // or if we miss the published primary change. - // - // Built before it is published, and published with a compare-exchange: this is now reached from the - // connect path as well as from ConnectionFailed, and the handler is wired up a few lines before the - // connect path calls it - on a connection that is already failing in the background. A plain null - // check would let both threads create a timer, and only one of them could ever be reached by - // OnManagedConnectionRestored to be disposed; the loser would keep firing SwitchPrimary every second - // for the life of the process, with no handle left to stop it. - var timer = new Timer( - _ => - { - try - { - // Attempt, but do not fail here - SwitchPrimary(endPoint, connection); - } - catch (Exception) - { - } - finally - { - try - { - connection.sentinelPrimaryReconnectTimer?.Change(TimeSpan.FromSeconds(1), Timeout.InfiniteTimeSpan); - } - catch (ObjectDisposedException) - { - // If we get here the managed connection was restored and the timer was - // disposed by another thread, so there's no need to run the timer again. - } - } - }, - null, - Timeout.InfiniteTimeSpan, - Timeout.InfiniteTimeSpan); + if (IsDisposed || connection.IsDisposed || connection.RawConfig.ServiceName is null) + { + StopSentinelPrimaryReconnect(connection); + return; + } + if (Volatile.Read(ref connection.sentinelPrimaryReconnectTimer) is not null) return; + // A one-shot timer, rearmed only after the asynchronous attempt finishes. Capture its own identity: + // a finishing callback must never rearm a replacement timer installed by a concurrent failure. + Timer? timer = null; + timer = new Timer(_ => ReconnectAsync().RedisFireAndForget(), null, Timeout.InfiniteTimeSpan, Timeout.InfiniteTimeSpan); if (Interlocked.CompareExchange(ref connection.sentinelPrimaryReconnectTimer, timer, null) is null) { - // we won: start it. Deliberately not started at construction - a timer that fires before it is - // published cannot be found by the callback's own Change call, nor disposed on restore. - timer.Change(TimeSpan.Zero, Timeout.InfiniteTimeSpan); + try { timer.Change(TimeSpan.Zero, Timeout.InfiniteTimeSpan); } + catch (ObjectDisposedException) { } // disposed/restored between publication and Change } else { timer.Dispose(); } + + async Task ReconnectAsync() + { + try + { + if (IsDisposed || connection.IsDisposed) + { + StopSentinelPrimaryReconnect(connection); + return; + } + await SwitchPrimaryAsync(endPoint, connection).ForAwait(); + } + catch (Exception ex) + { + Trace("Sentinel primary reconnect failed: " + ex.Message, "Sentinel"); + } + finally + { + if (ReferenceEquals(Volatile.Read(ref connection.sentinelPrimaryReconnectTimer), timer)) + { + try { timer!.Change(TimeSpan.FromSeconds(1), Timeout.InfiniteTimeSpan); } + catch (ObjectDisposedException) { } + } + } + } } /// @@ -528,8 +515,7 @@ private void ForgetDisposedChild(string serviceName, ConnectionMultiplexer child } /// - /// Re-resolves the primary through sentinel for every managed connection whose primary has been demoted - /// while its connection stayed up. + /// Re-resolves a managed primary that was seen demoted. /// /// /// @@ -580,24 +566,24 @@ private void CheckManagedPrimariesDemoted() foreach (var child in demoted) { if (Interlocked.CompareExchange(ref child._demotedPrimaryRecoveryActive, 1, 0) != 0) continue; + RecoverAsync(child).RedisFireAndForget(); + } - // asking sentinel and SwitchPrimary both block; keep that off the heartbeat - ThreadPool.QueueUserWorkItem(_ => + async Task RecoverAsync(ConnectionMultiplexer child) + { + try { - try - { - RecoverDemotedPrimary(child); - } - catch (Exception ex) - { - // sentinel unreachable, or either connection disposed meanwhile: a later heartbeat looks again - Trace("Demoted primary recovery failed: " + ex.Message, "Sentinel"); - } - finally - { - Volatile.Write(ref child._demotedPrimaryRecoveryActive, 0); - } - }); + await RecoverDemotedPrimaryAsync(child).ForAwait(); + } + catch (Exception ex) + { + // sentinel unreachable, or either connection disposed meanwhile: a later heartbeat looks again + Trace("Demoted primary recovery failed: " + ex.Message, "Sentinel"); + } + finally + { + Volatile.Write(ref child._demotedPrimaryRecoveryActive, 0); + } } } @@ -614,30 +600,30 @@ private void CheckManagedPrimariesDemoted() /// node to elect a primary that may not exist - once per second for as long as sentinel lags. Where that read /// cannot be sent (INFO unavailable), fall through to the switch rather than ask sentinel forever. /// anything else: switch, exactly as a +switch-master announcement would have. - /// Sentinel's answer is handed to so the switch acts on the answer these guards + /// Sentinel's answer is handed to so the switch acts on the answer these guards /// evaluated, rather than asking again and possibly acting on a different one. /// - /// A reconfigure already running is left alone: claims the reconfigure gate + /// A reconfigure already running is left alone: claims the reconfigure gate /// before touching anything, so a lost race leaves no trace and a later heartbeat retries. /// - private void RecoverDemotedPrimary(ConnectionMultiplexer connection) + private async Task RecoverDemotedPrimaryAsync(ConnectionMultiplexer connection) { if (IsDisposed || connection.IsDisposed || connection.RawConfig.ServiceName is not string serviceName) return; - // cheap early out before the blocking sentinel query; SwitchPrimary's claim is the authoritative check + // cheap early out before touching the topology; SwitchPrimaryAsync's claim is the authoritative check if (Volatile.Read(ref connection.activeConfigCause) is not null) return; - if (GetConfiguredPrimaryForService(serviceName) is not EndPoint reported) return; + if (await GetConfiguredPrimaryForServiceAsync(serviceName).ForAwait() is not EndPoint reported) return; if (connection.TryResolveServerEndPoint(reported) is { IsConnected: true, IsReplica: true } replica - && replica.CheckInfoReplication()) + && await replica.CheckInfoReplicationAsync().ForAwait()) { return; } - // the view may have been put right while sentinel was answering + // The view may have been put right while sentinel was answering. if (!IsSentinelPrimaryDemoted(connection)) return; - if (SwitchPrimary(connection.currentSentinelPrimaryEndPoint, connection, reportedPrimary: reported) + if (await SwitchPrimaryAsync(connection.currentSentinelPrimaryEndPoint, connection, reportedPrimary: reported).ForAwait() && connection.TryResolveServerEndPoint(reported) is { IsConnected: true, IsReplica: false }) { // As OnManagedConnectionRestored does: we are connected to a sentinel-approved primary again, so a @@ -677,99 +663,139 @@ private static bool IsSentinelPrimaryDemoted(ConnectionMultiplexer connection) return true; } - internal EndPoint? GetConfiguredPrimaryForService(string serviceName) => - _serverSnapshot // same as GetServerSnapshot, but without forcing span - .Where(static s => s.ServerType == ServerType.Sentinel) - .AsParallel() - .Select(s => - { - try { return GetServer(s.EndPoint).SentinelGetMasterAddressByName(serviceName); } - catch { return null; } - }) - .FirstOrDefault(r => r != null); - - internal EndPoint[]? GetReplicasForService(string serviceName) => - _serverSnapshot // same as GetServerSnapshot, but without forcing span - .Where(static s => s.ServerType == ServerType.Sentinel) - .AsParallel() - .Select(s => + internal Task GetConfiguredPrimaryForServiceAsync(string serviceName) => + QuerySentinelsAsync(server => server.SentinelGetMasterAddressByNameAsync(serviceName)); + + internal Task GetReplicasForServiceAsync(string serviceName) => + QuerySentinelsAsync(async server => (EndPoint[]?)await server.SentinelGetReplicaAddressesAsync(serviceName).ForAwait()); + + /// + /// Asks every sentinel at once and takes the first answer, as the parallel query this replaces did - but through + /// the asynchronous API: a request waiting on a slow or dead sentinel holds no thread-pool worker. + /// + private Task QuerySentinelsAsync(Func> query) where T : class + { + return FirstSentinelResultAsync(GetSentinelsForQuery().Select(QueryAsync).ToArray(), AsyncTimeoutMilliseconds); + + async Task QueryAsync(ServerEndPoint server) + { + try { return await query(GetServer(server.EndPoint)).ForAwait(); } + catch (Exception) { return null; } + } + } + + private ServerEndPoint[] GetSentinelsForQuery() + => _serverSnapshot.Where(static s => s.ServerType == ServerType.Sentinel && !s.IsDisposed).ToArray(); + + // Every request observes its own failure, including the requests still outstanding when a faster + // sentinel answers. One deadline covers the whole race, rather than a timeout per losing request. + // All requests use the asynchronous API: waiting for a slow/dead sentinel consumes no worker. + internal static async Task FirstSentinelResultAsync(Task[] requests, int timeoutMilliseconds) where T : class + { + using var timeout = new CancellationTokenSource(); + var deadline = Task.Delay(timeoutMilliseconds, timeout.Token); + var pending = new List(requests.Length + 1) { deadline }; + foreach (var request in requests) pending.Add(request.ObserveErrors()); + try + { + while (pending.Count > 1) { - try { return GetServer(s.EndPoint).SentinelGetReplicaAddresses(serviceName); } - catch { return null; } - }) - .FirstOrDefault(r => r != null); + var completed = await Task.WhenAny(pending).ForAwait(); + if (ReferenceEquals(completed, deadline)) break; + pending.Remove(completed); + try + { + if (await ((Task)completed).ForAwait() is T result) return result; + } + catch (Exception) { } // another sentinel may still have an answer + } + return null; + } + finally + { + timeout.Cancel(); + } + } /// - /// Switches the SentinelMasterConnection over to a new primary. + /// Switches the SentinelMasterConnection over to a new primary, without blocking a worker. /// + /// + /// The reconnect timer, the demotion heartbeat, the restore handler and +switch-master all call this. A + /// per-connection gate covers discovery as well as reconfiguration, so they cannot overlap; a switch requested while + /// one is running is not dropped, but recorded and replayed once it finishes. The reconnect timer is stopped here, + /// and only here, once the connection is on the sentinel-reported primary with no stale view left. + /// /// The endpoint responsible for the switch. /// The connection that should be switched over to a new primary endpoint. /// The writer to log to, if any. /// The primary sentinel already reported, when the caller has just asked it; /// otherwise sentinel is queried here. /// true when a switch ran (the endpoint set and/or roles were reconfigured); false when - /// 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. - internal bool SwitchPrimary(EndPoint? switchBlame, ConnectionMultiplexer connection, TextWriter? writer = null, EndPoint? reportedPrimary = null) + /// the view was already correct, another switch is running, or another reconfigure holds the gate - the callers + /// that retry (the reconnect timer, the demotion heartbeat) get a later attempt either way. + internal async Task SwitchPrimaryAsync(EndPoint? switchBlame, ConnectionMultiplexer connection, TextWriter? writer = null, EndPoint? reportedPrimary = null) { - var logger = Logger.With(writer); - if (connection.RawConfig.ServiceName is not string serviceName) + if (IsDisposed || connection.IsDisposed) return false; + if (connection.RawConfig.ServiceName is null) { - logger?.LogInformationServiceNameNotDefined(); + StopSentinelPrimaryReconnect(connection); return false; } - - // Get new primary - try twice (unless the caller already holds sentinel's answer) - 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. - string cause = $"Primary switch {serviceName}"; - if (!connection.TryClaimReconfigure(cause)) + if (Interlocked.CompareExchange(ref connection._sentinelSwitchActive, 1, 0) != 0) { - Trace($"Primary switch for {serviceName} skipped; reconfiguration already in progress"); + Interlocked.Exchange(ref connection._sentinelSwitchPending, 1); + ScheduleSentinelPrimaryReconnect(connection, switchBlame); return false; } - bool switched = false; + var serviceName = connection.RawConfig.ServiceName; + bool refreshAddresses = connection.currentSentinelPrimaryEndPoint is null; + var logger = Logger.With(writer); + bool healthy = false, claimed = false, handedOver = false; + EndPoint? confirmedPrimary = null; try { - // Trigger a blocking reconfigure for the switch, then refresh the sentinel address list. - // reconfigureAll:true forces every already-connected node to re-read its role before the - // primary is re-elected (see IsStalePrimaryView); reconfigureAll:false matches the legacy - // behavior for the "unknown endpoint" rebuild path. - void TriggerReconfigure(bool reconfigureAll) + if (serviceName is null) { - Trace($"Switching primary to {newPrimaryEndPoint}"); - switched = true; // from here the claim is ReconfigureAsync's to release, however it exits - connection.ReconfigureAsync( - first: false, - reconfigureAll: reconfigureAll, - log: logger, - blame: switchBlame, - cause: cause, - publishReconfigure: false, - publishReconfigureFlags: CommandFlags.PreferMaster, - claimed: true).Wait(); + logger?.LogInformationServiceNameNotDefined(); + return false; + } + + // Get new primary - try twice (unless the caller already holds sentinel's answer) + var newPrimaryEndPoint = reportedPrimary + ?? await GetConfiguredPrimaryForServiceAsync(serviceName).ForAwait() + ?? await GetConfiguredPrimaryForServiceAsync(serviceName).ForAwait() + ?? throw new RedisConnectionException(ConnectionFailureType.UnableToConnect, CommandFlags.None, $"Sentinel: Failed connecting to switch primary for service: {serviceName}"); + connection.currentSentinelPrimaryEndPoint = confirmedPrimary = newPrimaryEndPoint; - UpdateSentinelAddressList(serviceName); + // A primary we have never seen rebuilds the endpoint set around it. Ask for its replicas before claiming + // the reconfigure gate below, so that no other reconfigure waits on a sentinel round trip. + EndPoint[]? replicaEndPoints = null; + if (!connection.servers.Contains(newPrimaryEndPoint)) + { + replicaEndPoints = await GetReplicasForServiceAsync(serviceName).ForAwait() + ?? await GetReplicasForServiceAsync(serviceName).ForAwait(); + } + + // 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. + string cause = $"Primary switch {serviceName}"; + if (!connection.TryClaimReconfigure(cause)) + { + Trace($"Primary switch for {serviceName} skipped; reconfiguration already in progress"); + return false; } + claimed = true; + bool switched = false, reconfigureAll = false; if (!connection.servers.Contains(newPrimaryEndPoint)) { // The sentinel-reported primary is an endpoint we have never seen (e.g. a // recreated node that came back on a brand-new IP). Rebuild the endpoint set // from scratch around it and reconfigure. - EndPoint[]? replicaEndPoints = GetReplicasForService(serviceName) - ?? GetReplicasForService(serviceName); - connection.EndPoints.Clear(); connection.EndPoints.TryAdd(newPrimaryEndPoint); if (replicaEndPoints is not null) @@ -778,16 +804,14 @@ void TriggerReconfigure(bool reconfigureAll) { connection.EndPoints.TryAdd(replicaEndPoint); } - } - // Only once sentinel gave us the whole picture: a null replica list is a failed query, not a - // deployment with no replicas, and retiring on one would drop healthy nodes over a hiccup. - if (replicaEndPoints is not null) - { - RetireServersAbsentFromEndPoints(connection, logger); + // Only once sentinel gave us the whole picture: a null replica list is a failed query, not a + // deployment with no replicas, and retiring on one would drop healthy nodes over a hiccup. + await RetireServersAbsentFromEndPointsAsync(connection, logger).ForAwait(); } - TriggerReconfigure(reconfigureAll: false); + // reconfigureAll:false matches the legacy behavior for the "unknown endpoint" rebuild path + switched = true; } else if (IsStalePrimaryView(connection, newPrimaryEndPoint)) { @@ -798,15 +822,45 @@ void TriggerReconfigure(bool reconfigureAll) // already-connected node, so the primary election would re-pick the stale demoted // node. Force a full reconfigure so every connected node's role is refreshed before // the primary is re-elected. - TriggerReconfigure(reconfigureAll: true); + switched = reconfigureAll = true; } + refreshAddresses |= switched; + if (switched) + { + Trace($"Switching primary to {newPrimaryEndPoint}"); + handedOver = true; // from here the claim is ReconfigureAsync's to release, however it exits + await connection.ReconfigureAsync( + first: false, + reconfigureAll: reconfigureAll, + log: logger, + blame: switchBlame, + cause: cause, + publishReconfigure: false, + publishReconfigureFlags: CommandFlags.PreferMaster, + claimed: true).ForAwait(); + } + + healthy = !IsStalePrimaryView(connection, newPrimaryEndPoint); + if (healthy) connection.LastException = null; return switched; } finally { // not handed over (view already correct, or we threw before the reconfigure): release it ourselves - if (!switched) connection.ReleaseReconfigureClaim(); + if (claimed && !handedOver) connection.ReleaseReconfigureClaim(); + if (healthy && Volatile.Read(ref connection._sentinelSwitchPending) == 0) + { + StopSentinelPrimaryReconnect(connection); + // A failure delivered while the old timer was still published may not have installed a + // replacement. Recheck after removing it so a concurrent disconnect cannot lose recovery. + healthy = confirmedPrimary is not null && !IsStalePrimaryView(connection, confirmedPrimary); + } + Volatile.Write(ref connection._sentinelSwitchActive, 0); + if (Interlocked.Exchange(ref connection._sentinelSwitchPending, 0) != 0 || !healthy) + ScheduleSentinelPrimaryReconnect(connection, switchBlame); + // Address discovery runs on its own; it must not widen the switch gate. + if (refreshAddresses && serviceName is not null) UpdateSentinelAddressListAsync(serviceName).RedisFireAndForget(); } } @@ -827,21 +881,13 @@ void TriggerReconfigure(bool reconfigureAll) /// well: a dropped node finishes what it owes rather than having its in-flight commands abandoned. /// /// - private static void RetireServersAbsentFromEndPoints(ConnectionMultiplexer connection, ILogger? log) + private static async Task RetireServersAbsentFromEndPointsAsync(ConnectionMultiplexer connection, ILogger? log) { - List? drop = null; - foreach (var server in connection.GetServerSnapshot()) - { - if (server.IsDisposed || connection.EndPoints.Contains(server.EndPoint)) continue; - (drop ??= new List()).Add(server); - } - - if (drop is null) return; + var drop = connection._serverSnapshot.Where(s => !s.IsDisposed && !connection.EndPoints.Contains(s.EndPoint)).ToArray(); foreach (var server in drop) { - // short drain: this runs on the +switch-master handler and the reconnect timer, and the - // reconfigure that follows is the part that matters for getting traffic moving again - connection.RetireServerAsync(server, "no longer named by sentinel", TimeSpan.FromSeconds(1), log).Wait(); + // short drain: the reconfigure that follows is the part that matters for getting traffic moving again + await connection.RetireServerAsync(server, "no longer named by sentinel", TimeSpan.FromSeconds(1), log).ForAwait(); } } @@ -860,10 +906,11 @@ private static void RetireServersAbsentFromEndPoints(ConnectionMultiplexer conne /// The primary endpoint reported by sentinel (already known to the connection). private static bool IsStalePrimaryView(ConnectionMultiplexer connection, EndPoint newPrimaryEndPoint) { - var newPrimaryServer = connection.GetServerEndPoint(newPrimaryEndPoint, activate: false, provenance: ServerProvenance.Sentinel); + var newPrimaryServer = connection.TryResolveServerEndPoint(newPrimaryEndPoint); // We do not know this endpoint yet, or we still think the sentinel-reported primary is a - // replica, or we are not actually connected to it: our view is stale. + // replica, or its interactive connection is down: our view is stale. Subscription recovery is + // handled by the normal bridges; it must not keep primary discovery polling a healthy data node. if (newPrimaryServer is null || newPrimaryServer.IsReplica || !newPrimaryServer.IsConnected) { return true; @@ -886,17 +933,10 @@ private static bool IsStalePrimaryView(ConnectionMultiplexer connection, EndPoin return false; } - internal void UpdateSentinelAddressList(string serviceName) + internal async Task UpdateSentinelAddressListAsync(string serviceName) { - var firstCompleteRequest = _serverSnapshot // same as GetServerSnapshot, but without forcing span - .Where(static s => s.ServerType == ServerType.Sentinel) - .AsParallel() - .Select(s => - { - try { return GetServer(s.EndPoint).SentinelGetSentinelAddresses(serviceName); } - catch { return null; } - }) - .FirstOrDefault(r => r != null); + var firstCompleteRequest = await QuerySentinelsAsync(async server => + (EndPoint[]?)await server.SentinelGetSentinelAddressesAsync(serviceName).ForAwait()).ForAwait(); // Ignore errors, as having an updated sentinel list is not essential if (firstCompleteRequest == null) @@ -912,7 +952,7 @@ internal void UpdateSentinelAddressList(string serviceName) if (hasNew) { // Reconfigure the sentinel multiplexer if we added new endpoints - ReconfigureAsync(first: false, reconfigureAll: true, Logger, EndPoints[0], "Updating Sentinel List", false).Wait(); + await ReconfigureAsync(first: false, reconfigureAll: true, Logger, EndPoints[0], "Updating Sentinel List", false).ForAwait(); } } } diff --git a/src/StackExchange.Redis/ConnectionMultiplexer.cs b/src/StackExchange.Redis/ConnectionMultiplexer.cs index 50c57162a..4f54d9268 100644 --- a/src/StackExchange.Redis/ConnectionMultiplexer.cs +++ b/src/StackExchange.Redis/ConnectionMultiplexer.cs @@ -642,9 +642,9 @@ static async Task ApplyAfterConnectAsync(Task ConnectImplAsync(ConfigurationOptions configuration, TextWriter? writer = null, ServerType? serverType = null, Availability.CircuitBreaker? groupCircuitBreaker = null, bool isGroupMember = false) + private static async Task 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? connectHandler = null; ConnectionMultiplexer? muxer = null; var configLogger = configuration.LoggerFactory?.CreateLogger(); @@ -652,13 +652,24 @@ private static async Task ConnectImplAsync(ConfigurationO 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 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 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) /// /// Claims the reconfigure gate without running a reconfigure, so that work that must not interleave with - /// one (such as rewriting the endpoint list) can make its pre-checks and its + /// one (such as 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 /// with /// claimed: true (which releases it however it exits), or released via diff --git a/src/StackExchange.Redis/ServerEndPoint.cs b/src/StackExchange.Redis/ServerEndPoint.cs index 0768dbcb7..b21fdd654 100644 --- a/src/StackExchange.Redis/ServerEndPoint.cs +++ b/src/StackExchange.Redis/ServerEndPoint.cs @@ -1174,6 +1174,20 @@ internal bool CheckInfoReplication() return false; } + internal async ValueTask 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] diff --git a/tests/StackExchange.Redis.Tests/InProcessSentinelServer.cs b/tests/StackExchange.Redis.Tests/InProcessSentinelServer.cs index 8fecb98e6..e33618e5e 100644 --- a/tests/StackExchange.Redis.Tests/InProcessSentinelServer.cs +++ b/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,81 @@ public EndPoint Primary /// public static Tunnel CreateTunnel(params InProcessTestServer[] servers) => new RoutingTunnel(servers); - private int _primaryQueries; + private int _primaryQueries, _peerQueries, _replicaQueries, _legacyQueries; + private string _primaryFlags = "master"; /// How many times a client has asked this sentinel for the primary. 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 EndPoint[]? ReplicaAddressesOverride { get; set; } + public Func? BeforePrimaryReplyAsync { get; set; } + public Func? 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 hook, ValueTask pause) + { + // Execute has already captured the response; the barrier controls delivery, not the view being read. + await hook(); + await pause; + } + } + + /// Replicas to report with flags, including nodes that are no longer reachable. + public ConcurrentDictionary ReplicaFlags { get; } = new(); + + /// Peer routes to advertise; deliberately no MYID/runid support is needed for address discovery. + public ConcurrentDictionary PeerFlags { get; } = new(); + + public string PrimaryFlags + { + get => Volatile.Read(ref _primaryFlags); + set => Volatile.Write(ref _primaryFlags, value); + } + + public bool SupportsSentinelMaster { get; set; } = true; + + 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, 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); + Interlocked.Increment(ref _legacyQueries); if (!IsService(request)) 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 +134,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 +159,7 @@ private TypedRedisValue Replicas(in RedisRequest request) { if (!IsService(request)) return TypedRedisValue.Error("ERR No such master with that name"); + Interlocked.Increment(ref _replicaQueries); // sentinel learns the replicas from the primary, so report whatever replicates it right now var replicas = new List(); if (_monitored.TryGetNode(Primary, out var primaryNode)) @@ -104,10 +173,18 @@ 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; } @@ -117,7 +194,7 @@ private TypedRedisValue Describe(EndPoint endpoint, string flags) var host = GetHost(endpoint, out var port); var item = TypedRedisValue.Rent(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.StartsWith("master", StringComparison.Ordinal) ? ServiceName : $"{host}:{port}"); span[2] = TypedRedisValue.BulkString("ip"); span[3] = TypedRedisValue.BulkString(host); span[4] = TypedRedisValue.BulkString("port"); @@ -139,9 +216,9 @@ private sealed class RoutingTunnel(InProcessTestServer[] servers) : Tunnel } public override ValueTask 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 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); } } diff --git a/tests/StackExchange.Redis.Tests/SentinelDiscoveryUnitTests.cs b/tests/StackExchange.Redis.Tests/SentinelDiscoveryUnitTests.cs new file mode 100644 index 000000000..e8a291b4f --- /dev/null +++ b/tests/StackExchange.Redis.Tests/SentinelDiscoveryUnitTests.cs @@ -0,0 +1,34 @@ +using System; +using System.Collections.Generic; +using System.Net; +using System.Threading.Tasks; +using Xunit; + +namespace StackExchange.Redis.Tests; + +public class SentinelDiscoveryUnitTests +{ + [Fact] + public async Task FirstAnswerDoesNotWaitForASlowSentinel() + { + var never = new TaskCompletionSource(); + var answer = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var query = ConnectionMultiplexer.FirstSentinelResultAsync( + [never.Task, Task.FromResult(null), Task.FromException(new InvalidOperationException()), answer.Task], 10000); + Assert.False(query.IsCompleted); + answer.SetResult("primary"); + Assert.Same(query, await Task.WhenAny(query, Task.Delay(5000))); + Assert.Equal("primary", await query); + Assert.False(never.Task.IsCompleted); + } + + [Fact] + public async Task NoAnswerHasOneDeadlineForTheWholeQuery() + { + var never = new TaskCompletionSource(); + var query = ConnectionMultiplexer.FirstSentinelResultAsync([never.Task, never.Task], 20); + Assert.Same(query, await Task.WhenAny(query, Task.Delay(5000))); + Assert.Null(await query); + Assert.Null(await ConnectionMultiplexer.FirstSentinelResultAsync(Array.Empty>(), 20)); + } +} diff --git a/tests/StackExchange.Redis.Tests/SentinelPrimaryDemotionUnitTests.cs b/tests/StackExchange.Redis.Tests/SentinelPrimaryDemotionUnitTests.cs index 2a6a879c0..0c662c7d8 100644 --- a/tests/StackExchange.Redis.Tests/SentinelPrimaryDemotionUnitTests.cs +++ b/tests/StackExchange.Redis.Tests/SentinelPrimaryDemotionUnitTests.cs @@ -1,9 +1,14 @@ using System; using System.Diagnostics; +using System.IO; using System.Linq; using System.Net; +using System.Net.Sockets; using System.Threading; using System.Threading.Tasks; +using RESPite.Messages; +using StackExchange.Redis.Server; +using StackExchange.Redis.Configuration; using Xunit; namespace StackExchange.Redis.Tests; @@ -22,6 +27,7 @@ public class SentinelPrimaryDemotionUnitTests(ITestOutputHelper log) private static readonly EndPoint ReplicaB = new IPEndPoint(IPAddress.Parse("192.0.2.2"), 6379); private static readonly EndPoint NewNodeC = new IPEndPoint(IPAddress.Parse("192.0.2.3"), 6379); private static readonly EndPoint SentinelEndPoint = new IPEndPoint(IPAddress.Parse("192.0.2.10"), 26379); + private static readonly EndPoint SecondSentinelEndPoint = new IPEndPoint(IPAddress.Parse("192.0.2.11"), 26379); [Fact] public async Task DemotedPrimaryFollowsSentinelToPrimaryAtUnknownAddress() @@ -83,13 +89,309 @@ await UntilAsync( await AssertPrimaryBecomesAsync(conn, NewNodeC); } - private static ConfigurationOptions CreateConfig(InProcessTestServer data, InProcessSentinelServer sentinel) + [Fact] + public async Task RestoredBeforeFailedStillStopsTheReconnectTimer() + { + using var data = new InProcessTestServer(log, PrimaryA); + using var sentinel = new InProcessSentinelServer(data, ServiceName, PrimaryA, SentinelEndPoint, log); + await using var conn = await ConnectionMultiplexer.ConnectAsync(CreateConfig(data, sentinel)); + var owner = conn.sentinelConnection!; + var args = new ConnectionFailedEventArgs(null, conn, PrimaryA, ConnectionType.Interactive, ConnectionFailureType.SocketFailure, null, null); + + owner.OnManagedConnectionRestored(conn, args); + await UntilAsync(() => conn.sentinelPrimaryReconnectTimer is null, "restore did not settle"); + int baseline = sentinel.PrimaryQueries; + owner.OnManagedConnectionFailed(conn, args); // delayed worker delivery, socket is already healthy + await UntilAsync(() => sentinel.PrimaryQueries > baseline && conn.sentinelPrimaryReconnectTimer is null, "late failure left a timer"); + Assert.Equal([PrimaryA], ConnectedPrimaries(conn)); + } + + [Fact] + public async Task DemotionRecoveryUsesTheManagedServiceName() + { + using var data = new InProcessTestServer(log, PrimaryA); + using var sentinel = new InProcessSentinelServer(data, ServiceName, PrimaryA, SentinelEndPoint, log); + var otherPrimary = new IPEndPoint(IPAddress.Parse("192.0.2.4"), 6379); + var otherNewPrimary = new IPEndPoint(IPAddress.Parse("192.0.2.5"), 6379); + using var otherData = new InProcessTestServer(log, otherPrimary); + using var otherSentinel = new InProcessSentinelServer(otherData, "othermaster", otherPrimary, SecondSentinelEndPoint, log); + var config = CreateConfig(data, sentinel, otherData, otherSentinel); + config.EndPoints.Add(SecondSentinelEndPoint); + await using var conn = await ConnectionMultiplexer.ConnectAsync(config); + var owner = conn.sentinelConnection!; + var otherConfig = config.Clone(); + otherConfig.ServiceName = "othermaster"; + await using var otherConn = owner.GetSentinelMasterConnection(otherConfig); + Assert.Equal([otherPrimary], ConnectedPrimaries(otherConn)); + + // The first sentinel does not know this service. Its null answer must not hide the second service's + // recovery, and the healthy first service must stay on its own primary. + otherData.AddEmptyNode(otherNewPrimary); + otherData.SetReplicaOf(otherPrimary, otherNewPrimary); + otherSentinel.Primary = otherNewPrimary; + await AssertPrimaryBecomesAsync(otherConn, otherNewPrimary); + Assert.Equal([PrimaryA], ConnectedPrimaries(conn)); + await UntilAsync(() => otherConn.sentinelPrimaryReconnectTimer is null, "service recovery did not settle"); + } + + [Fact] + public async Task SwitchRequestedDuringDiscoveryIsReplayed() + { + using var data = new InProcessTestServer(log, PrimaryA); + data.AddReplicaNode(ReplicaB, PrimaryA); + using var sentinel = new InProcessSentinelServer(data, ServiceName, PrimaryA, SentinelEndPoint, log); + var config = CreateConfig(data, sentinel); + config.ConfigCheckSeconds = 3600; // role polling must not rescue a lost announcement + await using var conn = await ConnectionMultiplexer.ConnectAsync(config); + var owner = conn.sentinelConnection!; + var captured = NewSignal(); + var release = NewSignal(); + int replies = 0; + sentinel.BeforePrimaryReplyAsync = () => + { + if (Interlocked.Increment(ref replies) != 1) return Task.CompletedTask; + captured.TrySetResult(true); + return release.Task; + }; + var oldView = owner.SwitchPrimaryAsync(null, conn); + try + { + await AwaitSignalAsync(captured.Task); + FailOverToNewNode(data); + sentinel.Primary = NewNodeC; + Assert.False(await owner.SwitchPrimaryAsync(PrimaryA, conn)); + } + finally { release.TrySetResult(true); } + await oldView; + await AssertPrimaryBecomesAsync(conn, NewNodeC); + await UntilAsync(() => conn.sentinelPrimaryReconnectTimer is null, "replayed switch did not settle"); + Assert.True(Volatile.Read(ref replies) >= 2); + } + + [Fact] + public async Task ReconnectWaitsOneSecondAfterTheAsyncAttemptCompletes() + { + using var data = new InProcessTestServer(log, PrimaryA); + data.AddReplicaNode(ReplicaB, PrimaryA); + using var sentinel = new InProcessSentinelServer(data, ServiceName, PrimaryA, SentinelEndPoint, log); + await using var conn = await ConnectionMultiplexer.ConnectAsync(CreateConfig(data, sentinel)); + sentinel.Primary = ReplicaB; // still a replica, so recovery must retry + var first = NewSignal(); + var release = NewSignal(); + var second = NewSignal(); + long secondAt = 0; + int replies = 0; + sentinel.BeforePrimaryReplyAsync = () => + { + int number = Interlocked.Increment(ref replies); + if (number == 1) + { + first.TrySetResult(true); + return release.Task; + } + if (number == 2) + { + Interlocked.Exchange(ref secondAt, Stopwatch.GetTimestamp()); + second.TrySetResult(true); + } + return Task.CompletedTask; + }; + var owner = conn.sentinelConnection!; + owner.OnManagedConnectionFailed(conn, Failure(PrimaryA, conn)); + long releasedAt; + try + { + await AwaitSignalAsync(first.Task); + await Task.Delay(1200); + Assert.Equal(1, Volatile.Read(ref replies)); // no second callback while the first attempt is blocked + releasedAt = Stopwatch.GetTimestamp(); + } + finally { release.TrySetResult(true); } + await AwaitSignalAsync(second.Task); + // Allow scheduling/clock granularity margin instead of adding nominal Task.Delay durations exactly. + Assert.True((Interlocked.Read(ref secondAt) - releasedAt) / (double)Stopwatch.Frequency >= 0.8); + sentinel.Primary = PrimaryA; + await UntilAsync(() => conn.sentinelPrimaryReconnectTimer is null, "healthy primary did not stop recovery"); + } + + [Fact] + public async Task MissingServiceNameDoesNotRearmAnEmptyTimer() + { + using var data = new InProcessTestServer(log, PrimaryA); + using var sentinel = new InProcessSentinelServer(data, ServiceName, PrimaryA, SentinelEndPoint, log); + var config = CreateConfig(data, sentinel); + await using var conn = await ConnectionMultiplexer.ConnectAsync(config); + config.ServiceName = null; + var owner = conn.sentinelConnection!; + int queries = sentinel.PrimaryQueries; + owner.OnManagedConnectionFailed(conn, Failure(PrimaryA, conn)); + Assert.False(await owner.SwitchPrimaryAsync(null, conn)); + Assert.Null(conn.sentinelPrimaryReconnectTimer); + Assert.Equal(queries, sentinel.PrimaryQueries); + } + + [Fact] + public async Task AllSentinelsUnavailableKeepRecoveryUntilOneReturns() + { + using var data = new InProcessTestServer(log, PrimaryA); + using var sentinel = new InProcessSentinelServer(data, ServiceName, PrimaryA, SentinelEndPoint, log); + var config = CreateConfig(data, sentinel); + config.AsyncTimeout = 250; + await using var conn = await ConnectionMultiplexer.ConnectAsync(config); + var owner = conn.sentinelConnection!; + RemoveAndDisconnect(sentinel, SentinelEndPoint); + owner.OnManagedConnectionFailed(conn, Failure(PrimaryA, conn)); + await Task.Delay(2500); + Assert.NotNull(conn.sentinelPrimaryReconnectTimer); + sentinel.AddEmptyNode(SentinelEndPoint); + await UntilAsync(() => conn.sentinelPrimaryReconnectTimer is null && owner.IsConnected, "sentinel return did not end recovery"); + } + + [Fact] + public async Task DisposeDuringATimerAttemptDoesNotRestartRecovery() + { + using var data = new InProcessTestServer(log, PrimaryA); + using var sentinel = new InProcessSentinelServer(data, ServiceName, PrimaryA, SentinelEndPoint, log); + var conn = await ConnectionMultiplexer.ConnectAsync(CreateConfig(data, sentinel)); + await using (conn) + { + var captured = NewSignal(); + var release = NewSignal(); + sentinel.BeforePrimaryReplyAsync = () => + { + captured.TrySetResult(true); + return release.Task; + }; + conn.sentinelConnection!.OnManagedConnectionFailed(conn, Failure(PrimaryA, conn)); + Task dispose; + try + { + await AwaitSignalAsync(captured.Task); + dispose = conn.DisposeAsync().AsTask(); + } + finally { release.TrySetResult(true); } + await dispose; + await Task.Delay(1200); + Assert.True(conn.IsDisposed); + Assert.Null(conn.sentinelPrimaryReconnectTimer); + } + } + + [Fact] + public async Task FailedInfoWriteDoesNotCountAsAReplicationCheck() + { + using var data = new InProcessTestServer(log, PrimaryA); + var config = data.GetClientConfig(); + config.Tunnel = InProcessSentinelServer.CreateTunnel(data); + config.BacklogPolicy = BacklogPolicy.FailFast; + config.ConfigCheckSeconds = 1; + await using var conn = await ConnectionMultiplexer.ConnectAsync(config); + var server = conn.TryResolveServerEndPoint(PrimaryA)!; + RemoveAndDisconnect(data, PrimaryA); + await UntilAsync(() => !server.IsConnected, "server did not disconnect"); + Assert.False(await server.CheckInfoReplicationAsync()); + } + + [Fact] + public async Task ConcurrentManagedConnectReturnsOneChildAndDisposesTheOther() + { + using var data = new InProcessTestServer(log, PrimaryA); + using var sentinel = new InProcessSentinelServer(data, ServiceName, PrimaryA, SentinelEndPoint, log); + var ready = NewSignal(); + var release = NewSignal(); + var config = CreateConfig(data, sentinel); + config.Protocol = RedisProtocol.Resp2; + int connecting = 0; + config.Tunnel = new BlockedDataTunnel(config.Tunnel!, PrimaryA, release.Task, () => + { + if (Interlocked.Increment(ref connecting) == 2) ready.TrySetResult(true); + }); + await using var owner = await ConnectionMultiplexer.SentinelConnectAsync(config); + Task Start() => Task.Factory.StartNew(() => owner.GetSentinelMasterConnection(config), + CancellationToken.None, TaskCreationOptions.LongRunning, TaskScheduler.Default); + var first = Start(); + var second = Start(); + try { await AwaitSignalAsync(ready.Task); } + finally { release.TrySetResult(true); } + await using var child = await first; + Assert.Same(child, await second); + Assert.Single(owner.sentinelConnectionChildren); + await UntilAsync(() => data.ClientCount == 2, "losing managed connection was not disposed"); + } + + [Fact] + public async Task SynchronousSoftConnectReturnsBeforeAsyncRoleTimeoutAndRecovers() + { + using var data = new InProcessTestServer(log, PrimaryA); + using var sentinel = new InProcessSentinelServer(data, ServiceName, PrimaryA, SentinelEndPoint, log); + var release = NewSignal(); + var config = CreateConfig(data, sentinel); + config.AbortOnConnectFail = false; + config.ConnectTimeout = 200; + config.ConnectRetry = 1; + config.SyncTimeout = 500; + config.AsyncTimeout = 5000; + config.Tunnel = new BlockedDataTunnel(config.Tunnel!, PrimaryA, release.Task); + var connecting = Task.Factory.StartNew(() => ConnectionMultiplexer.Connect(config), CancellationToken.None, + TaskCreationOptions.LongRunning, TaskScheduler.Default); + ConnectionMultiplexer conn; + try + { + Assert.Same(connecting, await Task.WhenAny(connecting, Task.Delay(2500))); + conn = await connecting; + Assert.False(conn.IsConnected); + } + finally { release.TrySetResult(true); } + await using (conn) + { + await AssertPrimaryBecomesAsync(conn, PrimaryA); + await UntilAsync(() => conn.sentinelPrimaryReconnectTimer is null, "soft connect recovery did not settle"); + } + } + + private sealed class BlockedDataTunnel(Tunnel inner, EndPoint endpoint, Task release, Action? connecting = null) : Tunnel + { + public override ValueTask GetSocketConnectEndpointAsync(EndPoint target, CancellationToken cancellationToken) + => inner.GetSocketConnectEndpointAsync(target, cancellationToken); + + public override ValueTask BeforeAuthenticateAsync(EndPoint target, ConnectionType connectionType, Socket? socket, CancellationToken cancellationToken) + => Equals(target, endpoint) ? new(AcceptAsync(target, connectionType, socket, cancellationToken)) + : inner.BeforeAuthenticateAsync(target, connectionType, socket, cancellationToken); + + private async Task AcceptAsync(EndPoint target, ConnectionType connectionType, Socket? socket, CancellationToken cancellationToken) + { + if (connectionType == ConnectionType.Interactive) connecting?.Invoke(); + await release; + return await inner.BeforeAuthenticateAsync(target, connectionType, socket, cancellationToken); + } + } + + private static void RemoveAndDisconnect(InProcessTestServer server, EndPoint endpoint) + { + Assert.True(server.TryGetNode(endpoint, out var node)); + node.UpdateSlots([]); + Assert.True(server.RemoveNode(endpoint)); + server.ForAllClients(client => + { + if (ReferenceEquals(client.Node, node)) client.Dispose(); + }); + } + + private static TaskCompletionSource NewSignal() => new(TaskCreationOptions.RunContinuationsAsynchronously); + + private static async Task AwaitSignalAsync(Task signal) + => Assert.Same(signal, await Task.WhenAny(signal, Task.Delay(10000))); + + private static ConnectionFailedEventArgs Failure(EndPoint endpoint, ConnectionMultiplexer conn) + => new(null, conn, endpoint, ConnectionType.Interactive, ConnectionFailureType.SocketFailure, null, null); + + private static ConfigurationOptions CreateConfig(InProcessTestServer data, InProcessSentinelServer sentinel, params InProcessTestServer[] more) { var config = data.GetClientConfig(); config.EndPoints.Clear(); config.EndPoints.Add(SentinelEndPoint); config.ServiceName = ServiceName; - config.Tunnel = InProcessSentinelServer.CreateTunnel(data, sentinel); + config.Tunnel = InProcessSentinelServer.CreateTunnel([data, sentinel, .. more]); config.ConfigCheckSeconds = 1; // how soon the demoted node's role is re-read return config; } diff --git a/tests/StackExchange.Redis.Tests/SentinelTests.cs b/tests/StackExchange.Redis.Tests/SentinelTests.cs index ffe6d3705..05bef8c8c 100644 --- a/tests/StackExchange.Redis.Tests/SentinelTests.cs +++ b/tests/StackExchange.Redis.Tests/SentinelTests.cs @@ -510,7 +510,7 @@ public async Task SwitchPrimaryRebuildRetiresEndpointsSentinelNoLongerNames() var strayEndPoint = managed.GetEndPoints().Single(); Assert.Contains(strayEndPoint, managed.GetServerSnapshot().ToArray().Select(x => x.EndPoint)); - Conn.SwitchPrimary(null, managed, Writer); + await Conn.SwitchPrimaryAsync(null, managed, Writer); await UntilConditionAsync( TimeSpan.FromSeconds(10), () => !managed.GetServerSnapshot().ToArray().Any(x => Equals(x.EndPoint, strayEndPoint))); From 90db1124ff5ccd96002eb039837d1aff8d55a88e Mon Sep 17 00:00:00 2001 From: rebaseandpanic Date: Mon, 26 Oct 2026 10:29:24 +0000 Subject: [PATCH 2/4] fix(sentinel): take only a healthy primary answer, and keep the replica view consistent With every sentinel asked at once, the first answer wins - so which answer is taken matters more than when the sentinels were asked one by one. - The primary is read with SENTINEL MASTER and accepted only from a sentinel that holds it as a healthy master (flags contain master, no s_down/o_down). A sentinel still repeating a failed primary is passed over for one that already knows the new one. get-master-addr-by-name, which carries no flags, is the fallback only when SENTINEL MASTER itself is not supported; any other error is not a reason to trust a flagless answer. - Replicas are read from the sentinel whose primary answer was accepted, not from a second race that could combine two unrelated views. - Every switch refreshes the replica set, also for a primary the connection already knows; before, only an unknown primary did, so a replica that came back at a new address left the old one in the snapshot for good. - Sentinel keeps reporting failed replicas (it never forgets them until SENTINEL RESET). Down flags only decide whether a node not yet held is added; a node this connection holds and is connected to is kept whatever one sentinel's momentary view says. Only a disconnected node missing from a nonempty reply is retired, and an empty or failed replica query retires nothing. Spec: https://redis.io/docs/latest/develop/reference/sentinel-clients/#step-2-ask-for-the-master-address https://redis.io/docs/latest/develop/reference/sentinel-clients/#connect-to-replicas https://redis.io/docs/latest/develop/reference/sentinel-clients/#sentinel-failover-disconnection https://valkey.io/topics/sentinel-clients/#step-2-ask-for-primary-address https://valkey.io/topics/sentinel-clients/#connecting-to-replicas https://valkey.io/topics/sentinel-clients/#sentinel-failover-disconnection --- .../ConnectionMultiplexer.Sentinel.cs | 194 +++++++++++++----- .../SentinelDiscoveryUnitTests.cs | 67 ++++++ .../SentinelPrimaryDemotionUnitTests.cs | 182 ++++++++++++++++ .../SentinelTests.cs | 57 ++--- 4 files changed, 423 insertions(+), 77 deletions(-) diff --git a/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs b/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs index 174a02f9a..4a653bf9c 100644 --- a/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs +++ b/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs @@ -245,13 +245,13 @@ private async Task GetSentinelMasterConnectionAsync(Confi // Sentinel has some fun race behavior internally - give things a few shots for a quicker overall connect. const int queryAttempts = 2; - EndPoint? newPrimaryEndPoint = null; - for (int i = 0; i < queryAttempts && newPrimaryEndPoint is null; i++) + SentinelPrimary? primary = null; + for (int i = 0; i < queryAttempts && primary is null; i++) { - newPrimaryEndPoint = await GetConfiguredPrimaryForServiceAsync(serviceName).ForAwait(); + primary = await GetConfiguredPrimaryForServiceAsync(serviceName).ForAwait(); } - if (newPrimaryEndPoint is null) + if (primary is null) { failure = new RedisConnectionException( ConnectionFailureType.UnableToConnect, @@ -267,10 +267,11 @@ private async Task GetSentinelMasterConnectionAsync(Confi break; } - EndPoint[]? replicaEndPoints = null; + var newPrimaryEndPoint = primary.EndPoint; + SentinelReplica[]? replicaEndPoints = null; for (int i = 0; i < queryAttempts && replicaEndPoints is null; i++) { - replicaEndPoints = await GetReplicasForServiceAsync(serviceName).ForAwait(); + replicaEndPoints = await GetReplicasForServiceAsync(serviceName, primary).ForAwait(); } endpoints = config.EndPoints.Clone(); @@ -289,9 +290,10 @@ private async Task GetSentinelMasterConnectionAsync(Confi if (replicaEndPoints is not null) { - foreach (var replicaEndPoint in replicaEndPoints) + // a replica sentinel reports down is not connected to: it is added once its flags clear + foreach (var replica in replicaEndPoints) { - endpoints.TryAdd(replicaEndPoint); + if (!replica.Down) endpoints.TryAdd(replica.EndPoint); } } @@ -307,7 +309,8 @@ private async Task GetSentinelMasterConnectionAsync(Confi } // verify role is primary according to: - // https://redis.io/topics/sentinel-clients + // https://redis.io/docs/latest/develop/reference/sentinel-clients/ Step 3: ROLE (or INFO) on the node + // sentinel named; a non-primary answer means a stale sentinel view, so the loop asks again. bool isPrimary; var server = connection.GetServer(newPrimaryEndPoint); // ReSharper disable once ConditionIsAlwaysTrueOrFalseAccordingToNullableAPIContract @@ -613,7 +616,8 @@ private async Task RecoverDemotedPrimaryAsync(ConnectionMultiplexer connection) // cheap early out before touching the topology; SwitchPrimaryAsync's claim is the authoritative check if (Volatile.Read(ref connection.activeConfigCause) is not null) return; - if (await GetConfiguredPrimaryForServiceAsync(serviceName).ForAwait() is not EndPoint reported) return; + if (await GetConfiguredPrimaryForServiceAsync(serviceName).ForAwait() is not SentinelPrimary primary) return; + var reported = primary.EndPoint; if (connection.TryResolveServerEndPoint(reported) is { IsConnected: true, IsReplica: true } replica && await replica.CheckInfoReplicationAsync().ForAwait()) { @@ -623,7 +627,7 @@ private async Task RecoverDemotedPrimaryAsync(ConnectionMultiplexer connection) // The view may have been put right while sentinel was answering. if (!IsSentinelPrimaryDemoted(connection)) return; - if (await SwitchPrimaryAsync(connection.currentSentinelPrimaryEndPoint, connection, reportedPrimary: reported).ForAwait() + if (await SwitchPrimaryAsync(connection.currentSentinelPrimaryEndPoint, connection, reportedPrimary: primary).ForAwait() && connection.TryResolveServerEndPoint(reported) is { IsConnected: true, IsReplica: false }) { // As OnManagedConnectionRestored does: we are connected to a sentinel-approved primary again, so a @@ -663,11 +667,58 @@ private static bool IsSentinelPrimaryDemoted(ConnectionMultiplexer connection) return true; } - internal Task GetConfiguredPrimaryForServiceAsync(string serviceName) => - QuerySentinelsAsync(server => server.SentinelGetMasterAddressByNameAsync(serviceName)); + internal sealed class SentinelPrimary(IServer sentinel, EndPoint endPoint) + { + internal IServer Sentinel { get; } = sentinel; + internal EndPoint EndPoint { get; } = endPoint; + } + + /// + /// The primary sentinel holds for : the first valid answer from any sentinel route. + /// + /// + /// https://redis.io/docs/latest/develop/reference/sentinel-clients/ Step 2 asks the sentinels for the address; + /// an answer is taken only from a sentinel that holds that node as a healthy master (flags master, no + /// s_down/o_down), which SENTINEL MASTER reports and get-master-addr-by-name does not - + /// a sentinel still repeating a failed primary is passed over for one that already knows the new one. A stale + /// but healthy-looking answer is still caught by the role check (Step 3) and asked again. + /// + internal Task GetConfiguredPrimaryForServiceAsync(string serviceName) => + QuerySentinelsAsync(async server => + { + EndPoint? primary; + try + { + primary = ParseValidPrimary(await server.SentinelMasterAsync(serviceName).ForAwait()); + } + catch (RedisServerException ex) when (IsSentinelMasterUnsupported(ex)) + { + primary = await server.SentinelGetMasterAddressByNameAsync(serviceName).ForAwait(); + } + return primary is null ? null : new SentinelPrimary(server, primary); + }); - internal Task GetReplicasForServiceAsync(string serviceName) => - QuerySentinelsAsync(async server => (EndPoint[]?)await server.SentinelGetReplicaAddressesAsync(serviceName).ForAwait()); + internal static bool IsSentinelMasterUnsupported(RedisServerException ex) => + ex.Message.IndexOf("unknown command", StringComparison.OrdinalIgnoreCase) >= 0 + || ex.Message.IndexOf("unknown subcommand", StringComparison.OrdinalIgnoreCase) >= 0 + || ex.Message.IndexOf("unknown sentinel subcommand", StringComparison.OrdinalIgnoreCase) >= 0 + || ex.Message.IndexOf("unrecognized sentinel command", StringComparison.OrdinalIgnoreCase) >= 0; + + private static async Task GetReplicasForServiceAsync(string serviceName, SentinelPrimary primary) + { + try + { + // Use the same sentinel route whose primary answer won; a second race could combine unrelated views. + var rows = await primary.Sentinel.SentinelReplicasAsync(serviceName).ForAwait(); + var replicas = new List(rows.Length); + foreach (var row in rows) + { + if (ParseSentinelAddress(row, out var down) is EndPoint address) replicas.Add(new SentinelReplica(address, down)); + } + return replicas.ToArray(); + } + catch (Exception) { return null; } + } /// /// Asks every sentinel at once and takes the first answer, as the parallel query this replaces did - but through @@ -687,6 +738,44 @@ private static bool IsSentinelPrimaryDemoted(ConnectionMultiplexer connection) private ServerEndPoint[] GetSentinelsForQuery() => _serverSnapshot.Where(static s => s.ServerType == ServerType.Sentinel && !s.IsDisposed).ToArray(); + internal static bool IsSentinelNodeDown(string? flags) => flags is not null + && flags.Split(',').Any(static f => f is "s_down" or "o_down" or "disconnected"); + + // a primary is unusable only when sentinel holds it down; "disconnected" is a link sentinel is re-establishing + internal static bool IsSentinelPrimaryDown(string? flags) => flags is not null + && flags.Split(',').Any(static f => f is "s_down" or "o_down"); + + /// + /// A replica as sentinel reports it. Sentinel never forgets a replica (sentinel docs: "Sentinels never forget + /// about replicas of a given master" until SENTINEL RESET), so a failed one stays listed with down flags. + /// + internal readonly struct SentinelReplica(EndPoint endPoint, bool down) + { + public EndPoint EndPoint { get; } = endPoint; + public bool Down { get; } = down; + } + + internal static EndPoint? ParseValidPrimary(KeyValuePair[] row) + { + var fields = row.ToDictionary(static p => p.Key, static p => p.Value); + if (!fields.TryGetValue("flags", out var flags) || !flags.Split(',').Contains("master") || IsSentinelPrimaryDown(flags)) return null; + return ParseSentinelAddress(fields); + } + + private static EndPoint? ParseSentinelAddress(KeyValuePair[] row, out bool down) + { + var fields = row.ToDictionary(static p => p.Key, static p => p.Value); + down = fields.TryGetValue("flags", out var flags) && IsSentinelNodeDown(flags); + return ParseSentinelAddress(fields); + } + + private static EndPoint? ParseSentinelAddress(Dictionary fields) + { + if (fields.TryGetValue("ip", out var host) && fields.TryGetValue("port", out var port) + && Format.TryParseEndPoint(host, port, out var address)) return address; + return null; + } + // Every request observes its own failure, including the requests still outstanding when a faster // sentinel answers. One deadline covers the whole race, rather than a timeout per losing request. // All requests use the asynchronous API: waiting for a slow/dead sentinel consumes no worker. @@ -734,7 +823,7 @@ private ServerEndPoint[] GetSentinelsForQuery() /// true when a switch ran (the endpoint set and/or roles were reconfigured); false when /// the view was already correct, another switch is running, or another reconfigure holds the gate - the callers /// that retry (the reconnect timer, the demotion heartbeat) get a later attempt either way. - internal async Task SwitchPrimaryAsync(EndPoint? switchBlame, ConnectionMultiplexer connection, TextWriter? writer = null, EndPoint? reportedPrimary = null) + internal async Task SwitchPrimaryAsync(EndPoint? switchBlame, ConnectionMultiplexer connection, TextWriter? writer = null, SentinelPrimary? reportedPrimary = null) { if (IsDisposed || connection.IsDisposed) return false; if (connection.RawConfig.ServiceName is null) @@ -763,20 +852,16 @@ internal async Task SwitchPrimaryAsync(EndPoint? switchBlame, ConnectionMu } // Get new primary - try twice (unless the caller already holds sentinel's answer) - var newPrimaryEndPoint = reportedPrimary + var primary = reportedPrimary ?? await GetConfiguredPrimaryForServiceAsync(serviceName).ForAwait() ?? await GetConfiguredPrimaryForServiceAsync(serviceName).ForAwait() ?? throw new RedisConnectionException(ConnectionFailureType.UnableToConnect, CommandFlags.None, $"Sentinel: Failed connecting to switch primary for service: {serviceName}"); + var newPrimaryEndPoint = primary.EndPoint; connection.currentSentinelPrimaryEndPoint = confirmedPrimary = newPrimaryEndPoint; - // A primary we have never seen rebuilds the endpoint set around it. Ask for its replicas before claiming - // the reconfigure gate below, so that no other reconfigure waits on a sentinel round trip. - EndPoint[]? replicaEndPoints = null; - if (!connection.servers.Contains(newPrimaryEndPoint)) - { - replicaEndPoints = await GetReplicasForServiceAsync(serviceName).ForAwait() - ?? await GetReplicasForServiceAsync(serviceName).ForAwait(); - } + // Refresh replicas even when the primary is already known. Otherwise an unreachable old replica + // (or an old primary) stays in the snapshot forever after a node returns at a different address. + var replicas = await GetReplicasForServiceAsync(serviceName, primary).ForAwait(); // 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 @@ -790,49 +875,52 @@ internal async Task SwitchPrimaryAsync(EndPoint? switchBlame, ConnectionMu } claimed = true; - bool switched = false, reconfigureAll = false; - if (!connection.servers.Contains(newPrimaryEndPoint)) + bool unknownPrimary = !connection.servers.Contains(newPrimaryEndPoint); + bool stale = IsStalePrimaryView(connection, newPrimaryEndPoint); + bool changed = false; + if (replicas is not null) { - // The sentinel-reported primary is an endpoint we have never seen (e.g. a - // recreated node that came back on a brand-new IP). Rebuild the endpoint set - // from scratch around it and reconfigure. - connection.EndPoints.Clear(); - connection.EndPoints.TryAdd(newPrimaryEndPoint); - if (replicaEndPoints is not null) + // Down flags (s_down/o_down/disconnected) decide only whether a node not yet held is added. A node + // this connection already holds stays whatever one sentinel's momentary view says - sentinel reports + // the demoted primary disconnected while it reconnects to it, and a live connection outranks that. + // Only a disconnected node missing from a nonempty reply by the same sentinel may leave; + // an incomplete or empty view must not retire live nodes. + var desired = new EndPointCollection { newPrimaryEndPoint }; + foreach (var held in connection.EndPoints) { - foreach (var replicaEndPoint in replicaEndPoints) - { - connection.EndPoints.TryAdd(replicaEndPoint); - } - - // Only once sentinel gave us the whole picture: a null replica list is a failed query, not a - // deployment with no replicas, and retiring on one would drop healthy nodes over a hiccup. + if (replicas.Length == 0 || connection.TryResolveServerEndPoint(held) is { IsConnected: true }) desired.TryAdd(held); + } + foreach (var replica in replicas) + { + if (!replica.Down || connection.EndPoints.Contains(replica.EndPoint)) desired.TryAdd(replica.EndPoint); + } + changed = !new HashSet(connection.EndPoints).SetEquals(desired); + if (changed) + { + connection.EndPoints.Clear(); + foreach (var endpoint in desired) connection.EndPoints.TryAdd(endpoint); await RetireServersAbsentFromEndPointsAsync(connection, logger).ForAwait(); } - - // reconfigureAll:false matches the legacy behavior for the "unknown endpoint" rebuild path - switched = true; } - else if (IsStalePrimaryView(connection, newPrimaryEndPoint)) + else if (unknownPrimary) { - // The sentinel-reported primary IS an endpoint we already know about, but our - // cached view of the topology is stale (failover between known nodes: the old - // primary was demoted to a replica in-place and a former replica was promoted). - // A plain reconfigure with reconfigureAll:false would NOT re-read the role of an - // already-connected node, so the primary election would re-pick the stale demoted - // node. Force a full reconfigure so every connected node's role is refreshed before - // the primary is re-elected. - switched = reconfigureAll = true; + // A failed replica query is not an empty topology. Keep the old nodes until discovery works. + connection.EndPoints.TryAdd(newPrimaryEndPoint); + changed = true; } + bool switched = changed || stale; refreshAddresses |= switched; if (switched) { + // Step 3: a new node reads its role during handshake. For a known primary, reconfigureAll forces + // INFO on connected nodes too; an unknown primary can leave cached roles on old nodes until the + // next recovery attempt. Any remaining replica/second-primary view keeps recovery running. Trace($"Switching primary to {newPrimaryEndPoint}"); handedOver = true; // from here the claim is ReconfigureAsync's to release, however it exits await connection.ReconfigureAsync( first: false, - reconfigureAll: reconfigureAll, + reconfigureAll: !unknownPrimary, log: logger, blame: switchBlame, cause: cause, diff --git a/tests/StackExchange.Redis.Tests/SentinelDiscoveryUnitTests.cs b/tests/StackExchange.Redis.Tests/SentinelDiscoveryUnitTests.cs index e8a291b4f..13c94650e 100644 --- a/tests/StackExchange.Redis.Tests/SentinelDiscoveryUnitTests.cs +++ b/tests/StackExchange.Redis.Tests/SentinelDiscoveryUnitTests.cs @@ -31,4 +31,71 @@ public async Task NoAnswerHasOneDeadlineForTheWholeQuery() Assert.Null(await query); Assert.Null(await ConnectionMultiplexer.FirstSentinelResultAsync(Array.Empty>(), 20)); } + + [Theory] + [InlineData("master", true)] + [InlineData("master,disconnected", true)] + [InlineData("master,s_down", false)] + [InlineData("master,o_down,s_down", false)] + [InlineData("slave", false)] + [InlineData(null, false)] + public void OnlyAHealthyMasterIsAValidPrimaryAnswer(string? flags, bool valid) + { + var row = new List> + { + new("name", "mymaster"), + new("ip", "192.0.2.1"), + new("port", "6379"), + }; + if (flags is not null) row.Add(new("flags", flags)); + var primary = ConnectionMultiplexer.ParseValidPrimary(row.ToArray()); + if (valid) Assert.Equal(new IPEndPoint(IPAddress.Parse("192.0.2.1"), 6379), primary); + else Assert.Null(primary); + } + + [Fact] + public void IPv6PrimaryAddressIsParsed() + { + KeyValuePair[] row = [new("name", "m"), new("ip", "2001:db8::1"), new("port", "6379"), new("flags", "master")]; + Assert.Equal(new IPEndPoint(IPAddress.Parse("2001:db8::1"), 6379), ConnectionMultiplexer.ParseValidPrimary(row)); + } + + [Theory] + [InlineData("192.0.2.1", "1:2")] + [InlineData("192.0.2.1", "port")] + [InlineData("192.0.2.1", "")] + [InlineData("", "6379")] + public void MalformedAddressIsNotAValidPrimaryAnswer(string ip, string port) + { + // a sentinel whose answer has no usable address is passed over, like one that does not know the service + KeyValuePair[] row = [new("name", "mymaster"), new("ip", ip), new("port", port), new("flags", "master")]; + Assert.Null(ConnectionMultiplexer.ParseValidPrimary(row)); + } + + [Fact] + public void AddressIsReadAsAHostNameNeverAsALocalSocket() + { + // "!path" is the configuration spelling of a Unix socket; an address sentinel reports is never one + KeyValuePair[] row = [new("name", "m"), new("ip", "!/tmp/redis.sock"), new("port", "6379"), new("flags", "master")]; + var primary = Assert.IsType(ConnectionMultiplexer.ParseValidPrimary(row)); + Assert.Equal("!/tmp/redis.sock", primary.Host); + Assert.Equal(6379, primary.Port); + } + + [Theory] + [InlineData("master,s_down", true)] + [InlineData("master,o_down", true)] + [InlineData("master,disconnected", false)] + [InlineData("master", false)] + public void OnlyDownFlagsDisqualifyAPrimary(string flags, bool down) + => Assert.Equal(down, ConnectionMultiplexer.IsSentinelPrimaryDown(flags)); + + [Theory] + [InlineData("slave,s_down", true)] + [InlineData("sentinel,o_down", true)] + [InlineData("slave,disconnected", true)] + [InlineData("slave", false)] + [InlineData(null, false)] + public void DownFlagsAreExcludedFromManagedDiscovery(string? flags, bool down) + => Assert.Equal(down, ConnectionMultiplexer.IsSentinelNodeDown(flags)); } diff --git a/tests/StackExchange.Redis.Tests/SentinelPrimaryDemotionUnitTests.cs b/tests/StackExchange.Redis.Tests/SentinelPrimaryDemotionUnitTests.cs index 0c662c7d8..a8bbb90f1 100644 --- a/tests/StackExchange.Redis.Tests/SentinelPrimaryDemotionUnitTests.cs +++ b/tests/StackExchange.Redis.Tests/SentinelPrimaryDemotionUnitTests.cs @@ -106,6 +106,108 @@ public async Task RestoredBeforeFailedStillStopsTheReconnectTimer() Assert.Equal([PrimaryA], ConnectedPrimaries(conn)); } + [Fact] + public async Task KnownHealthyPrimaryRetiresAReplicaNoLongerAdvertisedAndStopsRecovery() + { + using var data = new InProcessTestServer(log, PrimaryA); + data.AddReplicaNode(ReplicaB, PrimaryA); + using var sentinel = new InProcessSentinelServer(data, ServiceName, PrimaryA, SentinelEndPoint, log); + await using var conn = await ConnectionMultiplexer.ConnectAsync(CreateConfig(data, sentinel)); + Assert.Contains(ReplicaB, conn.GetEndPoints()); + + data.AddReplicaNode(NewNodeC, PrimaryA); + RemoveAndDisconnect(data, ReplicaB); + await UntilAsync(() => !conn.GetServer(ReplicaB).IsConnected, "removed replica stayed connected"); + var owner = conn.sentinelConnection!; + var args = new ConnectionFailedEventArgs(null, conn, ReplicaB, ConnectionType.Interactive, ConnectionFailureType.SocketFailure, null, null); + owner.OnManagedConnectionFailed(conn, args); + await UntilAsync(() => !conn.GetEndPoints().Contains(ReplicaB) && conn.sentinelPrimaryReconnectTimer is null, "dead replica kept recovery alive"); + Assert.Equal([PrimaryA], ConnectedPrimaries(conn)); + Assert.DoesNotContain(ReplicaB, conn.GetServerSnapshot().ToArray().Select(x => x.EndPoint)); + } + + [Fact] + public async Task HeldReplicaReportedDownIsNotRetired() + { + using var data = new InProcessTestServer(log, PrimaryA); + data.AddReplicaNode(ReplicaB, PrimaryA); + using var sentinel = new InProcessSentinelServer(data, ServiceName, PrimaryA, SentinelEndPoint, log); + await using var conn = await ConnectionMultiplexer.ConnectAsync(CreateConfig(data, sentinel)); + Assert.Contains(ReplicaB, conn.GetEndPoints()); + + // sentinel's momentary view (it is reconnecting to the node) against our live connection: the connection wins + sentinel.ReplicaFlags[ReplicaB] = "s_down,disconnected"; + var owner = conn.sentinelConnection!; + int changes = 0; + conn.ConfigurationChanged += (_, _) => Interlocked.Increment(ref changes); + Assert.False(await owner.SwitchPrimaryAsync(ReplicaB, conn)); + Assert.Contains(ReplicaB, conn.GetEndPoints()); + Assert.Contains(ReplicaB, conn.GetServerSnapshot().ToArray().Select(x => x.EndPoint)); + Assert.True(conn.GetServer(ReplicaB).IsConnected); + Assert.Equal(0, Volatile.Read(ref changes)); + Assert.Null(conn.sentinelPrimaryReconnectTimer); + Assert.Equal([PrimaryA], ConnectedPrimaries(conn)); + } + + [Fact] + public async Task ReplicaReportedDownJoinsOnRecoveryOnceItsFlagsClear() + { + using var data = new InProcessTestServer(log, PrimaryA); + using var sentinel = new InProcessSentinelServer(data, ServiceName, PrimaryA, SentinelEndPoint, log); + await using var conn = await ConnectionMultiplexer.ConnectAsync(CreateConfig(data, sentinel)); + var owner = conn.sentinelConnection!; + Assert.Equal([PrimaryA], conn.GetEndPoints()); + + // a replica appears, but sentinel still holds it down: not a node to connect to yet + data.AddReplicaNode(ReplicaB, PrimaryA); + sentinel.ReplicaFlags[ReplicaB] = "s_down"; + Assert.False(await owner.SwitchPrimaryAsync(null, conn)); + Assert.Equal([PrimaryA], conn.GetEndPoints()); + Assert.Null(conn.sentinelPrimaryReconnectTimer); + + // The flags clear; the next recovery refresh adds the replica without retiring a held node. + Assert.True(sentinel.ReplicaFlags.TryRemove(ReplicaB, out _)); + Assert.True(await owner.SwitchPrimaryAsync(null, conn)); + await UntilAsync(() => conn.GetEndPoints().Contains(ReplicaB) && conn.GetServer(ReplicaB).IsConnected, "cleared replica did not join"); + Assert.Equal([PrimaryA], ConnectedPrimaries(conn)); + Assert.Contains(PrimaryA, conn.GetServerSnapshot().ToArray().Select(x => x.EndPoint)); + await UntilAsync(() => conn.sentinelPrimaryReconnectTimer is null, "recovery left a timer"); + } + + [Fact] + public async Task StaleSentinelNamingADownPrimaryIsPassedOver() + { + using var data = new InProcessTestServer(log, PrimaryA); + data.AddReplicaNode(ReplicaB, PrimaryA); + FailOverToNewNode(data); + // the stale sentinel still names A, but holds it down; the other already knows C + using var stale = new InProcessSentinelServer(data, ServiceName, PrimaryA, SentinelEndPoint, log) { PrimaryFlags = "master,s_down" }; + using var current = new InProcessSentinelServer(data, ServiceName, NewNodeC, SecondSentinelEndPoint, log); + var config = CreateConfig(data, stale, current); + config.EndPoints.Add(SecondSentinelEndPoint); + await using var conn = await ConnectionMultiplexer.ConnectAsync(config); + + Assert.Equal([NewNodeC], ConnectedPrimaries(conn)); + Assert.DoesNotContain(PrimaryA, conn.GetServerSnapshot().ToArray().Where(x => x.IsConnected && !x.IsReplica).Select(x => x.EndPoint)); + Assert.True(stale.PrimaryQueries > 0, "the stale sentinel was never asked"); + await UntilAsync(() => conn.sentinelPrimaryReconnectTimer is null, "connect left a timer"); + } + + [Fact] + public async Task LegacySentinelWithoutSentinelMasterStillResolvesThePrimary() + { + using var data = new InProcessTestServer(log, PrimaryA); + data.AddReplicaNode(ReplicaB, PrimaryA); + using var sentinel = new InProcessSentinelServer(data, ServiceName, PrimaryA, SentinelEndPoint, log) { SupportsSentinelMaster = false }; + await using var conn = await ConnectionMultiplexer.ConnectAsync(CreateConfig(data, sentinel)); + Assert.Equal([PrimaryA], ConnectedPrimaries(conn)); + Assert.Contains(ReplicaB, conn.GetEndPoints()); + + FailOverToNewNode(data); + sentinel.Primary = NewNodeC; + await AssertPrimaryBecomesAsync(conn, NewNodeC); + } + [Fact] public async Task DemotionRecoveryUsesTheManagedServiceName() { @@ -214,6 +316,86 @@ public async Task ReconnectWaitsOneSecondAfterTheAsyncAttemptCompletes() await UntilAsync(() => conn.sentinelPrimaryReconnectTimer is null, "healthy primary did not stop recovery"); } + [Theory] + [InlineData(true)] + [InlineData(false)] + public async Task IncompleteReplicaViewDoesNotRetireConnectedNodes(bool empty) + { + using var data = new InProcessTestServer(log, PrimaryA); + data.AddReplicaNode(ReplicaB, PrimaryA); + data.AddReplicaNode(NewNodeC, PrimaryA); + using var sentinel = new InProcessSentinelServer(data, ServiceName, PrimaryA, SentinelEndPoint, log); + await using var conn = await ConnectionMultiplexer.ConnectAsync(CreateConfig(data, sentinel)); + var held = conn.TryResolveServerEndPoint(ReplicaB); + sentinel.ReplicaAddressesOverride = empty ? [] : [NewNodeC]; + await conn.sentinelConnection!.SwitchPrimaryAsync(null, conn); + Assert.Contains(ReplicaB, conn.GetEndPoints()); + Assert.Same(held, conn.TryResolveServerEndPoint(ReplicaB)); + Assert.True(conn.GetServer(ReplicaB).IsConnected); + Assert.Null(conn.sentinelPrimaryReconnectTimer); + } + + [Fact] + public async Task ReplicasComeFromTheSentinelWhosePrimaryWasAccepted() + { + using var data = new InProcessTestServer(log, PrimaryA); + data.AddReplicaNode(ReplicaB, PrimaryA); + using var accepted = new InProcessSentinelServer(data, ServiceName, PrimaryA, SentinelEndPoint, log); + using var other = new InProcessSentinelServer(data, ServiceName, PrimaryA, SecondSentinelEndPoint, log) + { + PrimaryError = "ERR No such master with that name", + ReplicaAddressesOverride = [], + }; + var config = CreateConfig(data, accepted, other); + config.EndPoints.Add(SecondSentinelEndPoint); + await using var conn = await ConnectionMultiplexer.ConnectAsync(config); + Assert.Contains(ReplicaB, conn.GetEndPoints()); + Assert.True(accepted.ReplicaQueries > 0); + Assert.Equal(0, other.ReplicaQueries); + Assert.Equal(0, other.LegacyQueries); + } + + [Theory] + [InlineData("LOADING Redis is loading the dataset in memory")] + [InlineData("ERR No such master with that name")] + public async Task SentinelErrorsDoNotUseTheFlaglessFallback(string error) + { + using var data = new InProcessTestServer(log, PrimaryA); + data.AddReplicaNode(ReplicaB, PrimaryA); + FailOverToNewNode(data); + using var invalid = new InProcessSentinelServer(data, ServiceName, PrimaryA, SentinelEndPoint, log) { PrimaryError = error }; + using var current = new InProcessSentinelServer(data, ServiceName, NewNodeC, SecondSentinelEndPoint, log); + var config = CreateConfig(data, invalid, current); + config.EndPoints.Add(SecondSentinelEndPoint); + await using var conn = await ConnectionMultiplexer.ConnectAsync(config); + Assert.Equal([NewNodeC], ConnectedPrimaries(conn)); + Assert.Equal(0, invalid.LegacyQueries); + } + + [Fact] + public async Task DisconnectedHeldReplicaDoesNotKeepPrimaryRecoveryAlive() + { + using var data = new InProcessTestServer(log, PrimaryA); + data.AddReplicaNode(ReplicaB, PrimaryA); + using var sentinel = new InProcessSentinelServer(data, ServiceName, PrimaryA, SentinelEndPoint, log); + await using var conn = await ConnectionMultiplexer.ConnectAsync(CreateConfig(data, sentinel)); + var held = conn.TryResolveServerEndPoint(ReplicaB); + sentinel.ReplicaFlags[ReplicaB] = "s_down,disconnected"; + RemoveAndDisconnect(data, ReplicaB); + await UntilAsync(() => !conn.GetServer(ReplicaB).IsConnected, "replica did not disconnect"); + var owner = conn.sentinelConnection!; + owner.OnManagedConnectionFailed(conn, Failure(ReplicaB, conn)); + await UntilAsync(() => conn.sentinelPrimaryReconnectTimer is null, "dead held replica kept the timer"); + Assert.Contains(ReplicaB, conn.GetEndPoints()); + Assert.Same(held, conn.TryResolveServerEndPoint(ReplicaB)); + Assert.False(conn.GetServer(ReplicaB).IsConnected); + int queries = sentinel.PrimaryQueries; + await Task.Delay(2500); + Assert.Equal(queries, sentinel.PrimaryQueries); + Assert.Null(conn.sentinelPrimaryReconnectTimer); + await conn.GetDatabase().StringSetAsync("primary-still-works", "yes"); + } + [Fact] public async Task MissingServiceNameDoesNotRearmAnEmptyTimer() { diff --git a/tests/StackExchange.Redis.Tests/SentinelTests.cs b/tests/StackExchange.Redis.Tests/SentinelTests.cs index 05bef8c8c..c93d80051 100644 --- a/tests/StackExchange.Redis.Tests/SentinelTests.cs +++ b/tests/StackExchange.Redis.Tests/SentinelTests.cs @@ -499,30 +499,39 @@ public async Task SwitchPrimaryRebuildRetiresEndpointsSentinelNoLongerNames() { SkipOnWindowsRelease(); - // A managed connection pointed somewhere sentinel does not name: the primary it reports is an - // endpoint this connection has never seen, which is what drives SwitchPrimary's rebuild branch - // (the real-world trigger is a node that came back on a brand-new address). - await using var managed = await ConnectionMultiplexer.ConnectAsync( - $"{TestConfig.Current.PrimaryServer}:{TestConfig.Current.PrimaryPort},allowAdmin=true", - Writer); - managed.RawConfig.ServiceName = ServiceName; - - var strayEndPoint = managed.GetEndPoints().Single(); - Assert.Contains(strayEndPoint, managed.GetServerSnapshot().ToArray().Select(x => x.EndPoint)); - - await Conn.SwitchPrimaryAsync(null, managed, Writer); - await UntilConditionAsync( - TimeSpan.FromSeconds(10), - () => !managed.GetServerSnapshot().ToArray().Any(x => Equals(x.EndPoint, strayEndPoint))); - - // Retired, not merely dropped from the endpoint list: the snapshot is what server selection reads - // and what GetEndPoints reports, so a server left there stays connected and selectable. - Assert.DoesNotContain(strayEndPoint, managed.GetEndPoints()); - Assert.DoesNotContain(strayEndPoint, managed.GetServerSnapshot().ToArray().Select(x => x.EndPoint)); - - // ...and the rebuild still did its actual job - Assert.NotEmpty(managed.GetEndPoints()); - await managed.GetDatabase().PingAsync(); + // An actually disconnected stray address may be retired from a nonempty, consistent sentinel view. + // A connected node must survive an incomplete view (covered by the in-process unit tests). + var port = TcpForwarder.GetFreePort(); + TcpForwarder? forwarder = new(port, TestConfig.Current.PrimaryPort); + try + { + await using var managed = await ConnectionMultiplexer.ConnectAsync( + $"127.0.0.1:{port},allowAdmin=true", + Writer); + managed.RawConfig.ServiceName = ServiceName; + + var strayEndPoint = managed.GetEndPoints().Single(); + var stray = managed.GetServerSnapshot().ToArray().Single(x => Equals(x.EndPoint, strayEndPoint)); + forwarder.Dispose(); + forwarder = null; + await UntilConditionAsync(TimeSpan.FromSeconds(10), () => !stray.IsConnected); + Assert.False(stray.IsConnected); + + await Conn.SwitchPrimaryAsync(null, managed, Writer); + await UntilConditionAsync( + TimeSpan.FromSeconds(10), + () => !managed.GetServerSnapshot().ToArray().Any(x => Equals(x.EndPoint, strayEndPoint))); + + // Retirement removes the endpoint from both configuration and the server-selection snapshot. + Assert.DoesNotContain(strayEndPoint, managed.GetEndPoints()); + Assert.DoesNotContain(strayEndPoint, managed.GetServerSnapshot().ToArray().Select(x => x.EndPoint)); + Assert.NotEmpty(managed.GetEndPoints()); + await managed.GetDatabase().PingAsync(); + } + finally + { + forwarder?.Dispose(); + } } [Fact] From 7d331da3d29709e858061c95e13359754619d586 Mon Sep 17 00:00:00 2001 From: rebaseandpanic Date: Mon, 26 Oct 2026 10:34:36 +0000 Subject: [PATCH 3/4] fix(sentinel): keep every sentinel route, parse +sentinel, ask reported-down routes last - The +sentinel handler passed messageParts[0] - the literal instance type "sentinel" - as the service name, so the address refresh it triggers never matched a service and the announcement was never acted on. The service is read after the "@" separator. - The sentinel list stays append-only, as before: seeds and routes found earlier are kept, and a Service address and the pod address the same sentinel advertises are both kept without being matched to each other. What changes is how a refresh runs: new routes are added before the reconfigure is claimed, so a busy reconfigure cannot drop what was discovered, and a refresh requested while one is running is replayed rather than lost. - Every route is still asked at once, but connected routes are started first and routes sentinel itself reports down last. Spec: https://redis.io/docs/latest/develop/reference/sentinel-clients/#step-1-connect-to-the-first-sentinel https://redis.io/docs/latest/develop/reference/sentinel-clients/#sentinels-list-automatic-refresh https://redis.io/docs/latest/develop/reference/sentinel-clients/#subscribe-to-sentinel-events-to-improve-responsiveness https://valkey.io/topics/sentinel-clients/#step-1-connect-to-the-first-sentinel https://valkey.io/topics/sentinel-clients/#sentinels-list-automatic-refresh https://valkey.io/topics/sentinel-clients/#subscribe-to-sentinel-events-to-improve-responsiveness --- .../ConnectionMultiplexer.Sentinel.cs | 90 ++++++++++++++--- .../SentinelDiscoveryUnitTests.cs | 10 ++ .../SentinelPrimaryDemotionUnitTests.cs | 98 +++++++++++++++++++ .../SentinelTests.cs | 26 +++++ 4 files changed, 208 insertions(+), 16 deletions(-) diff --git a/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs b/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs index 4a653bf9c..e116d5510 100644 --- a/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs +++ b/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs @@ -20,7 +20,10 @@ public partial class ConnectionMultiplexer private int _sentinelInitialized; private int _demotedPrimaryRecoveryActive; private int _sentinelSwitchActive; + private int _sentinelAddressUpdateActive; + private readonly HashSet _sentinelAddressUpdatesPending = new(StringComparer.Ordinal); private int _sentinelSwitchPending; + private HashSet _sentinelDownRoutes = new(); /// /// Initializes the connection as a Sentinel connection and adds the necessary event handlers to track changes to the managed primaries. @@ -94,7 +97,10 @@ internal void InitializeSentinel(ILogger? log) (_, message) => { string[] messageParts = ((string)message!).Split(new[] { ' ' }, StringSplitOptions.RemoveEmptyEntries); - UpdateSentinelAddressListAsync(messageParts[0]).RedisFireAndForget(); + // " @ ": the service follows the "@" + int separator = Array.IndexOf(messageParts, "@"); + if (separator >= 0 && separator + 1 < messageParts.Length) + UpdateSentinelAddressListAsync(messageParts[separator + 1]).RedisFireAndForget(); }, CommandFlags.FireAndForget); } @@ -735,8 +741,21 @@ internal static bool IsSentinelMasterUnsupported(RedisServerException ex) => } } + // sentinel-clients Step 1 tries the sentinels one by one, moving the one that answered to the front. Every route + // is asked at once instead - the requests are asynchronous and hold no worker, and a Service address and the + // advertised pod address of the same sentinel need not be told apart - connected routes first and routes + // sentinel itself reports down last, so a dead route can delay nothing and the first valid answer wins. private ServerEndPoint[] GetSentinelsForQuery() - => _serverSnapshot.Where(static s => s.ServerType == ServerType.Sentinel && !s.IsDisposed).ToArray(); + { + var down = Volatile.Read(ref _sentinelDownRoutes); + return OrderSentinelQueryTargets( + _serverSnapshot.Where(static s => s.ServerType == ServerType.Sentinel && !s.IsDisposed).ToArray(), + static s => s.IsConnected, + s => down.Contains(s.EndPoint)); + } + + internal static T[] OrderSentinelQueryTargets(T[] targets, Func connected, Func reportedDown) + => targets.OrderBy(t => connected(t) ? 0 : reportedDown(t) ? 2 : 1).ToArray(); // OrderBy is stable internal static bool IsSentinelNodeDown(string? flags) => flags is not null && flags.Split(',').Any(static f => f is "s_down" or "o_down" or "disconnected"); @@ -1023,24 +1042,63 @@ private static bool IsStalePrimaryView(ConnectionMultiplexer connection, EndPoin internal async Task UpdateSentinelAddressListAsync(string serviceName) { - var firstCompleteRequest = await QuerySentinelsAsync(async server => - (EndPoint[]?)await server.SentinelGetSentinelAddressesAsync(serviceName).ForAwait()).ForAwait(); - - // Ignore errors, as having an updated sentinel list is not essential - if (firstCompleteRequest == null) - return; - - bool hasNew = false; - foreach (EndPoint newSentinel in firstCompleteRequest.Where(x => !EndPoints.Contains(x))) + lock (_sentinelAddressUpdatesPending) { - hasNew = true; - EndPoints.TryAdd(newSentinel); + if (IsDisposed) return; + _sentinelAddressUpdatesPending.Add(serviceName); + if (_sentinelAddressUpdateActive != 0) return; + _sentinelAddressUpdateActive = 1; } + try + { + while (!IsDisposed) + { + lock (_sentinelAddressUpdatesPending) + { + if (_sentinelAddressUpdatesPending.Count == 0) break; + serviceName = _sentinelAddressUpdatesPending.First(); + _sentinelAddressUpdatesPending.Remove(serviceName); + } + var peers = await QuerySentinelsAsync(async server => + (KeyValuePair[][]?)await server.SentinelSentinelsAsync(serviceName).ForAwait()).ForAwait(); + if (peers is null) continue; + + var down = new HashSet(); + bool hasNew = false; + // https://valkey.io/topics/sentinel-clients/: append unknown ip:port pairs, retaining all old routes. + // Add before claiming reconfigure, so a busy claim cannot lose the discovery result. + foreach (var peer in peers) + { + if (ParseSentinelAddress(peer, out var isDown) is not EndPoint address) continue; + hasNew |= EndPoints.TryAdd(address); + if (isDown) down.Add(address); + } + Volatile.Write(ref _sentinelDownRoutes, down); + if (!hasNew) continue; - if (hasNew) + const string cause = "Updating Sentinel List"; + while (!IsDisposed) + { + if (TryClaimReconfigure(cause)) + { + await ReconfigureAsync(first: false, reconfigureAll: true, Logger, EndPoints[0], cause, false, claimed: true).ForAwait(); + break; + } + // Coalesce later announcements while waiting for the owner of the claim; no worker is held. + await Task.Delay(1000).ForAwait(); + } + } + } + finally { - // Reconfigure the sentinel multiplexer if we added new endpoints - await ReconfigureAsync(first: false, reconfigureAll: true, Logger, EndPoints[0], "Updating Sentinel List", false).ForAwait(); + string? pending; + lock (_sentinelAddressUpdatesPending) + { + _sentinelAddressUpdateActive = 0; + pending = !IsDisposed ? _sentinelAddressUpdatesPending.FirstOrDefault() : null; + if (IsDisposed) _sentinelAddressUpdatesPending.Clear(); + } + if (pending is not null) UpdateSentinelAddressListAsync(pending).RedisFireAndForget(); } } } diff --git a/tests/StackExchange.Redis.Tests/SentinelDiscoveryUnitTests.cs b/tests/StackExchange.Redis.Tests/SentinelDiscoveryUnitTests.cs index 13c94650e..a097ef51b 100644 --- a/tests/StackExchange.Redis.Tests/SentinelDiscoveryUnitTests.cs +++ b/tests/StackExchange.Redis.Tests/SentinelDiscoveryUnitTests.cs @@ -32,6 +32,16 @@ public async Task NoAnswerHasOneDeadlineForTheWholeQuery() Assert.Null(await ConnectionMultiplexer.FirstSentinelResultAsync(Array.Empty>(), 20)); } + [Fact] + public void EveryRouteIsQueriedConnectedFirstAndReportedDownLast() + { + // sentinel-clients Step 1: no route is left out, a dead route only moves back in the launch order + string[] routes = ["dead-a", "service-a", "down-b", "service-b", "pod-c", "service-c"]; + var ordered = ConnectionMultiplexer.OrderSentinelQueryTargets(routes, s => s.StartsWith("service-"), s => s.StartsWith("down-")); + Assert.Equal(["service-a", "service-b", "service-c", "dead-a", "pod-c", "down-b"], ordered); + Assert.Empty(ConnectionMultiplexer.OrderSentinelQueryTargets(Array.Empty(), _ => true, _ => false)); + } + [Theory] [InlineData("master", true)] [InlineData("master,disconnected", true)] diff --git a/tests/StackExchange.Redis.Tests/SentinelPrimaryDemotionUnitTests.cs b/tests/StackExchange.Redis.Tests/SentinelPrimaryDemotionUnitTests.cs index a8bbb90f1..3260d9b08 100644 --- a/tests/StackExchange.Redis.Tests/SentinelPrimaryDemotionUnitTests.cs +++ b/tests/StackExchange.Redis.Tests/SentinelPrimaryDemotionUnitTests.cs @@ -208,6 +208,39 @@ public async Task LegacySentinelWithoutSentinelMasterStillResolvesThePrimary() await AssertPrimaryBecomesAsync(conn, NewNodeC); } + [Fact] + public async Task SentinelRoutesAreOnlyAddedAndDoNotNeedIdentity() + { + using var data = new InProcessTestServer(log, PrimaryA); + var seed = new DnsEndPoint("sentinel-service", 26379); + using var sentinel = new InProcessSentinelServer(data, ServiceName, PrimaryA, seed, log); + sentinel.AddEmptyNode(SentinelEndPoint); + sentinel.PeerFlags[SentinelEndPoint] = "s_down,disconnected"; + var config = CreateConfig(data, sentinel); + config.EndPoints.Clear(); + config.EndPoints.Add(seed); + await using var conn = await ConnectionMultiplexer.ConnectAsync(config); + var owner = conn.sentinelConnection!; + await UntilAsync(() => owner.GetEndPoints().Contains(SentinelEndPoint) && owner.GetServer(SentinelEndPoint).IsConnected, "down peer route was not added"); + Assert.Contains(seed, owner.GetEndPoints()); + + // A later response no longer names the advertised IP. Both that route and the DNS seed must remain, + // with one server per address. + sentinel.PeerFlags.Clear(); + int baseline = sentinel.PeerQueries; + var elapsed = Stopwatch.StartNew(); + while (sentinel.PeerQueries == baseline && elapsed.Elapsed < TimeSpan.FromSeconds(10)) + { + await owner.UpdateSentinelAddressListAsync(ServiceName); + await Task.Delay(50); + } + Assert.True(sentinel.PeerQueries > baseline, "empty peer response was not queried"); + Assert.Equal(2, owner.GetEndPoints().Length); + Assert.Contains(seed, owner.GetEndPoints()); + Assert.Contains(SentinelEndPoint, owner.GetEndPoints()); + Assert.Single(owner.GetServerSnapshot().ToArray(), x => Equals(x.EndPoint, SentinelEndPoint)); + } + [Fact] public async Task DemotionRecoveryUsesTheManagedServiceName() { @@ -396,6 +429,71 @@ public async Task DisconnectedHeldReplicaDoesNotKeepPrimaryRecoveryAlive() await conn.GetDatabase().StringSetAsync("primary-still-works", "yes"); } + [Fact] + public async Task SentinelAnnouncementUsesTheServiceAfterTheSeparator() + { + using var data = new InProcessTestServer(log, PrimaryA); + using var sentinel = new InProcessSentinelServer(data, ServiceName, PrimaryA, SentinelEndPoint, log); + await using var conn = await ConnectionMultiplexer.ConnectAsync(CreateConfig(data, sentinel)); + sentinel.AddEmptyNode(SecondSentinelEndPoint); + sentinel.PeerFlags[SecondSentinelEndPoint] = ""; + Assert.True(sentinel.ForAllClients(client => client.Publish(RedisChannel.Literal("+sentinel"), + "sentinel peer 192.0.2.11 26379 @ mymaster 192.0.2.1 6379")) > 0); + var owner = conn.sentinelConnection!; + await UntilAsync(() => owner.GetEndPoints().Contains(SecondSentinelEndPoint), "announcement did not discover the peer"); + } + + [Fact] + public async Task SentinelAddressRequestDuringAnOlderReplyIsReplayed() + { + using var data = new InProcessTestServer(log, PrimaryA); + using var sentinel = new InProcessSentinelServer(data, ServiceName, PrimaryA, SentinelEndPoint, log); + await using var conn = await ConnectionMultiplexer.ConnectAsync(CreateConfig(data, sentinel)); + var owner = conn.sentinelConnection!; + var captured = NewSignal(); + var release = NewSignal(); + int replies = 0; + sentinel.BeforePeerReplyAsync = () => + { + if (Interlocked.Increment(ref replies) != 1) return Task.CompletedTask; + captured.TrySetResult(true); + return release.Task; + }; + var update = owner.UpdateSentinelAddressListAsync(ServiceName); + try + { + await AwaitSignalAsync(captured.Task); + sentinel.AddEmptyNode(SecondSentinelEndPoint); + sentinel.PeerFlags[SecondSentinelEndPoint] = ""; + await owner.UpdateSentinelAddressListAsync(ServiceName); + } + finally { release.TrySetResult(true); } + await update; + await UntilAsync(() => owner.GetEndPoints().Contains(SecondSentinelEndPoint), "pending peer refresh was lost"); + Assert.True(Volatile.Read(ref replies) >= 2); + } + + [Fact] + public async Task SentinelAddressIsRetainedWhileReconfigureIsBusy() + { + using var data = new InProcessTestServer(log, PrimaryA); + using var sentinel = new InProcessSentinelServer(data, ServiceName, PrimaryA, SentinelEndPoint, log); + await using var conn = await ConnectionMultiplexer.ConnectAsync(CreateConfig(data, sentinel)); + var owner = conn.sentinelConnection!; + sentinel.AddEmptyNode(SecondSentinelEndPoint); + sentinel.PeerFlags[SecondSentinelEndPoint] = ""; + await UntilAsync(() => owner.TryClaimReconfigure("unit holds the claim"), "could not claim reconfigure"); + Task update; + try + { + update = owner.UpdateSentinelAddressListAsync(ServiceName); + await UntilAsync(() => owner.EndPoints.Contains(SecondSentinelEndPoint), "busy claim lost the address"); + } + finally { owner.ReleaseReconfigureClaim(); } + await update; + await UntilAsync(() => owner.GetServer(SecondSentinelEndPoint).IsConnected, "pending reconfigure did not connect the peer"); + } + [Fact] public async Task MissingServiceNameDoesNotRearmAnEmptyTimer() { diff --git a/tests/StackExchange.Redis.Tests/SentinelTests.cs b/tests/StackExchange.Redis.Tests/SentinelTests.cs index c93d80051..22e544c39 100644 --- a/tests/StackExchange.Redis.Tests/SentinelTests.cs +++ b/tests/StackExchange.Redis.Tests/SentinelTests.cs @@ -534,6 +534,32 @@ await UntilConditionAsync( } } + [Fact] + public async Task SentinelDiscoveryKeepsConfiguredNamesAndPreviouslyKnownRoutes() + { + var options = ServiceOptions.Clone(); + var seed = new DnsEndPoint("localhost", TestConfig.Current.SentinelPortA); + options.EndPoints.Add(seed); + await using var connection = await ConnectionMultiplexer.SentinelConnectAsync(options); + var missing = new IPEndPoint(IPAddress.Loopback, TcpForwarder.GetFreePort()); + connection.EndPoints.TryAdd(missing); + connection.GetServerEndPoint(missing, ServerProvenance.Sentinel, activate: false); + + await connection.UpdateSentinelAddressListAsync(ServiceName); + Assert.Contains(seed, connection.EndPoints); + Assert.Contains(seed, connection.GetEndPoints()); + Assert.Contains(missing, connection.EndPoints); + Assert.Contains(missing, connection.GetEndPoints()); + // Repeated replies can omit a route (including the responding sentinel itself); discovery only appends. + // Which sentinel answers first is a race, and a reply from another one can still add the first + // responder's own address - so a later refresh may grow the list, but must not shrink it. + await connection.UpdateSentinelAddressListAsync(ServiceName); + var endpoints = new HashSet(connection.EndPoints); + await connection.UpdateSentinelAddressListAsync(ServiceName); + Assert.Superset(endpoints, new HashSet(connection.EndPoints)); + Assert.True(connection.IsConnected); + } + [Fact] public async Task AbortOnConnectFailFalseRecoversOnceSentinelsAppear() { From ca307f3a4cbd869f18eafd3d567443d413b04655 Mon Sep 17 00:00:00 2001 From: rebaseandpanic Date: Mon, 26 Oct 2026 10:36:44 +0000 Subject: [PATCH 4/4] fix(sentinel): a READONLY reply marks a managed node as a replica A primary demoted in place answers writes with READONLY long before the next INFO replication poll (ConfigCheckSeconds, 60s by default) notices the role change, and until then commands keep failing against it. On a sentinel-managed connection a READONLY reply now marks the node as a replica right away - locally, without starting a reconfigure on the IO path - and the demotion heartbeat then asks sentinel and switches, as it does for a demotion seen by INFO. Standalone and cluster connections keep their existing READONLY behaviour. Spec: https://redis.io/docs/latest/develop/reference/sentinel-clients/#step-3-call-the-role-command-in-the-target-instance https://redis.io/docs/latest/develop/reference/sentinel-clients/#sentinel-failover-disconnection https://valkey.io/topics/sentinel-clients/#step-3-call-the-role-command-in-the-target-instance https://valkey.io/topics/sentinel-clients/#sentinel-failover-disconnection --- docs/Configuration.md | 6 +- .../ConnectionMultiplexer.Sentinel.cs | 3 + src/StackExchange.Redis/ResultProcessor.cs | 9 +++ src/StackExchange.Redis/ServerEndPoint.cs | 6 ++ .../SentinelPrimaryDemotionUnitTests.cs | 60 +++++++++++++++++++ 5 files changed, 83 insertions(+), 1 deletion(-) diff --git a/docs/Configuration.md b/docs/Configuration.md index 1df31a2ae..6efc1d523 100644 --- a/docs/Configuration.md +++ b/docs/Configuration.md @@ -46,7 +46,11 @@ 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 the client only follows a +sentinel that holds the node it names as a healthy master (`SENTINEL MASTER` flags without `s_down`/`o_down`; sentinels without +`SENTINEL MASTER` are asked with `get-master-addr-by-name`). An overview of mapping between the `string` and `ConfigurationOptions` representation is shown below, but you can switch between them trivially: diff --git a/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs b/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs index e116d5510..6901dfa27 100644 --- a/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs +++ b/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs @@ -24,6 +24,7 @@ public partial class ConnectionMultiplexer private readonly HashSet _sentinelAddressUpdatesPending = new(StringComparer.Ordinal); private int _sentinelSwitchPending; private HashSet _sentinelDownRoutes = new(); + internal bool IsSentinelManaged { get; private set; } /// /// Initializes the connection as a Sentinel connection and adds the necessary event handlers to track changes to the managed primaries. @@ -362,6 +363,8 @@ private async Task GetSentinelMasterConnectionAsync(Confi connection.Logger?.LogErrorSyncConnectTimeout(failure, failure.Message); } + connection.IsSentinelManaged = true; + // Attach to reconnect event to ensure proper connection to the new primary connection.ConnectionRestored += OnManagedConnectionRestored; diff --git a/src/StackExchange.Redis/ResultProcessor.cs b/src/StackExchange.Redis/ResultProcessor.cs index 9b2180301..568be65ea 100644 --- a/src/StackExchange.Redis/ResultProcessor.cs +++ b/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; diff --git a/src/StackExchange.Redis/ServerEndPoint.cs b/src/StackExchange.Redis/ServerEndPoint.cs index b21fdd654..20f015735 100644 --- a/src/StackExchange.Redis/ServerEndPoint.cs +++ b/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; diff --git a/tests/StackExchange.Redis.Tests/SentinelPrimaryDemotionUnitTests.cs b/tests/StackExchange.Redis.Tests/SentinelPrimaryDemotionUnitTests.cs index 3260d9b08..8f61f11bd 100644 --- a/tests/StackExchange.Redis.Tests/SentinelPrimaryDemotionUnitTests.cs +++ b/tests/StackExchange.Redis.Tests/SentinelPrimaryDemotionUnitTests.cs @@ -269,6 +269,49 @@ public async Task DemotionRecoveryUsesTheManagedServiceName() await UntilAsync(() => otherConn.sentinelPrimaryReconnectTimer is null, "service recovery did not settle"); } + [Fact] + public async Task ReadOnlyReplyMarksTheNodeAsReplicaAndRecoveryFollowsSentinel() + { + using var data = new ReadOnlyReplicaTestServer(log, PrimaryA); + data.AddReplicaNode(ReplicaB, PrimaryA); + using var sentinel = new InProcessSentinelServer(data, ServiceName, PrimaryA, SentinelEndPoint, log); + var config = CreateConfig(data, sentinel); + config.ConfigCheckSeconds = 3600; // role polling must not be what notices the demotion + await using var conn = await ConnectionMultiplexer.ConnectAsync(config); + await UntilAsync(() => conn.sentinelPrimaryReconnectTimer is null, "connect left a timer"); + var db = conn.GetDatabase(); + await db.StringSetAsync("readonly-key", "before"); + + // Demoted in place: nothing announced or disconnected, and role polling is an hour away. + FailOverToNewNode(data); + sentinel.Primary = NewNodeC; + Assert.False(conn.GetServer(PrimaryA).IsReplica); + + var ex = await Assert.ThrowsAsync(() => db.StringSetAsync("readonly-key", "after")); + Assert.StartsWith("READONLY", ex.Message); + Assert.True(conn.GetServer(PrimaryA).IsReplica, "READONLY did not mark the node as a replica"); + + // the heartbeat sees a connected primary that reports replica, asks sentinel and switches + await AssertPrimaryBecomesAsync(conn, NewNodeC); + await db.StringSetAsync("readonly-key", "after"); + } + + /// + /// The in-process server accepts writes on any node; a real replica answers READONLY, which is the signal under test. + /// + private sealed class ReadOnlyReplicaTestServer(ITestOutputHelper log, EndPoint endpoint) : InProcessTestServer(log, endpoint) + { + public override TypedRedisValue Execute(RedisClient client, in RedisRequest request) + { + if (request.Count > 0 && string.Equals(request.GetString(0), "SET", StringComparison.OrdinalIgnoreCase) + && client.Node is { PrimaryId: { Length: > 0 } }) + { + return TypedRedisValue.Error("READONLY You can't write against a read only replica."); + } + return base.Execute(client, in request); + } + } + [Fact] public async Task SwitchRequestedDuringDiscoveryIsReplayed() { @@ -557,6 +600,23 @@ public async Task DisposeDuringATimerAttemptDoesNotRestartRecovery() } } + [Theory] + [InlineData(ServerType.Standalone)] + [InlineData(ServerType.Cluster)] + public async Task ReadOnlyDoesNotChangeNonSentinelTopologies(ServerType topology) + { + using var data = new ReadOnlyReplicaTestServer(log, PrimaryA) { ServerType = topology }; + data.AddReplicaNode(ReplicaB, PrimaryA); + var config = data.GetClientConfig(); + config.ConfigCheckSeconds = 3600; + await using var conn = await ConnectionMultiplexer.ConnectAsync(config); + if (topology == ServerType.Cluster) data.Failover(ReplicaB); + else FailOverToNewNode(data); + Assert.False(conn.GetServer(PrimaryA).IsReplica); + await Assert.ThrowsAsync(() => conn.GetServer(PrimaryA).ExecuteAsync("SET", "key", "value")); + Assert.False(conn.GetServer(PrimaryA).IsReplica); + } + [Fact] public async Task FailedInfoWriteDoesNotCountAsAReplicationCheck() {