diff --git a/docs/Configuration.md b/docs/Configuration.md index 1df31a2ae..02f7d09ef 100644 --- a/docs/Configuration.md +++ b/docs/Configuration.md @@ -46,7 +46,13 @@ connection to the primary is lost, and - because the announcement is delivered o notices when the node last named as primary is still connected but now reports itself as a replica (a failover where nothing disconnected, such as a primary demoted in place). In that last case the client re-asks sentinel for the current primary and follows its answer, even to a node at an address it has never seen; while sentinel still names the demoted node (sentinel itself has not yet failed over), the client -keeps asking without tearing down or reconfiguring the healthy connections. +keeps asking without tearing down or reconfiguring the healthy connections. A command answered with `READONLY` is the same signal: the node +is marked as a replica at once, rather than at the next role check (`ConfigCheckSeconds`), and the client asks sentinel on its next heartbeat. +Sentinel is queried asynchronously: waiting on a slow or unreachable sentinel holds no thread-pool thread, and a sentinel query waits up to +`AsyncTimeout`, also when the connection is made with the synchronous `Connect`. A sentinel whose `SENTINEL MASTER` reply holds the primary +down (`s_down`/`o_down`) is passed over for one that does not. A sentinel without a usable `SENTINEL MASTER` reply - the subcommand is +unknown or refused, such as to an ACL user allowed only `get-master-addr-by-name`, or the reply has no flags - is asked with +`get-master-addr-by-name` instead. An overview of mapping between the `string` and `ConfigurationOptions` representation is shown below, but you can switch between them trivially: diff --git a/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs b/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs index 31f5296ce..af4c42e66 100644 --- a/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs +++ b/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs @@ -19,6 +19,11 @@ public partial class ConnectionMultiplexer internal ConnectionMultiplexer? sentinelConnection; private int _sentinelInitialized; private int _demotedPrimaryRecoveryActive; + private int _sentinelSwitchActive; + private int _sentinelAddressUpdateActive; + private readonly HashSet _sentinelAddressUpdatesPending = new(StringComparer.Ordinal); + private int _sentinelSwitchPending; + 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. @@ -50,12 +55,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 +77,7 @@ internal void InitializeSentinel(ILogger? log) if (switchTarget is not null) { - SwitchPrimary(switchBlame, switchTarget); + SwitchPrimaryAsync(switchBlame, switchTarget).RedisFireAndForget(); } }, CommandFlags.FireAndForget); @@ -86,7 +87,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 +97,10 @@ internal void InitializeSentinel(ILogger? log) (_, message) => { string[] messageParts = ((string)message!).Split(new[] { ' ' }, StringSplitOptions.RemoveEmptyEntries); - UpdateSentinelAddressList(messageParts[0]); + // " @ ": the service follows the "@" + int separator = Array.IndexOf(messageParts, "@"); + if (separator >= 0 && separator + 1 < messageParts.Length) + UpdateSentinelAddressListAsync(messageParts[separator + 1]).RedisFireAndForget(); }, CommandFlags.FireAndForget); } @@ -189,7 +193,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 +201,7 @@ private static async Task SentinelPrimaryConnectAsync(Con } catch { - try { sentinelConnection.Dispose(); } catch { } + try { await sentinelConnection.DisposeAsync().ForAwait(); } catch { } throw; } } @@ -208,6 +212,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 +254,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 +268,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,11 +301,20 @@ 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 + // 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 @@ -306,7 +323,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 +339,7 @@ public ConnectionMultiplexer GetSentinelMasterConnection(ConfigurationOptions co } } - Thread.Sleep(100); + await Task.Delay(100).ForAwait(); } while (sw.ElapsedMilliseconds < config.ConnectTimeout); @@ -337,28 +354,45 @@ public ConnectionMultiplexer GetSentinelMasterConnection(ConfigurationOptions co throw failure; } - connection ??= SeedConnection(failure); + connection ??= await SeedConnectionAsync(failure).ForAwait(); connection.LastException = failure; connection.Logger?.LogErrorSyncConnectTimeout(failure, failure.Message); } + connection.IsSentinelManaged = true; + // Attach to reconnect event to ensure proper connection to the new primary connection.ConnectionRestored += OnManagedConnectionRestored; // 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 +400,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 +418,7 @@ public ConnectionMultiplexer GetSentinelMasterConnection(ConfigurationOptions co sentinelConnectionChildren.Remove(serviceName); } } - try { connection.Dispose(); } catch { } + try { await connection.DisposeAsync().ForAwait(); } catch { } } throw; } @@ -393,127 +427,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 +524,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 +575,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 +609,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,136 +672,252 @@ 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 => + /// + /// 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. + /// + /// A sentinel that does not give a usable SENTINEL MASTER answer - any error (the subcommand unknown, or + /// denied by an ACL that allows only what discovery needed before), or a reply without the master flag or an + /// address - is asked get-master-addr-by-name instead, exactly as every sentinel was before. Only a sentinel + /// that explicitly holds the node down is passed over. + /// + /// + internal Task GetConfiguredPrimaryForServiceAsync(string serviceName) => + QuerySentinelsAsync(async server => + { + EndPoint? primary; + try + { + primary = ParseHealthyPrimary(await server.SentinelMasterAsync(serviceName).ForAwait(), out bool heldDown); + if (heldDown) return null; + } + catch (RedisServerException) { - 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 => + primary = null; + } + return primary ?? await server.SentinelGetMasterAddressByNameAsync(serviceName).ForAwait(); + }); + + // Every replica sentinel lists is added, whatever its flags: one that is down is left to its bridge to reconnect, + // so a replica down at connect time (a rolling restart) joins once it is back at the same address. + 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 + { + // sentinel-clients Step 1 tries the sentinels one by one; every route is asked at once instead, as the parallel + // query did - the requests are asynchronous and hold no worker, so a dead route delays nothing. + var sentinels = _serverSnapshot.Where(static s => s.ServerType == ServerType.Sentinel && !s.IsDisposed); + return FirstSentinelResultAsync(sentinels.Select(QueryAsync).ToArray(), AsyncTimeoutMilliseconds); + + async Task QueryAsync(ServerEndPoint server) + { + try { return await query(GetServer(server.EndPoint)).ForAwait(); } + catch (Exception) { return null; } + } + } + + // 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"); + + /// + /// The address a SENTINEL MASTER reply names, when it describes a healthy master; otherwise null. + /// + /// The reply. + /// Whether sentinel holds the node down (s_down/o_down): the one null + /// that is an answer. Any other null is a reply that cannot be judged. + internal static EndPoint? ParseHealthyPrimary(KeyValuePair[] row, out bool heldDown) + { + var fields = row.ToDictionary(static p => p.Key, static p => p.Value); + fields.TryGetValue("flags", out var flags); + heldDown = IsSentinelPrimaryDown(flags); + if (heldDown || flags is null || !flags.Split(',').Contains("master")) return null; + 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. + 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; + } - UpdateSentinelAddressList(serviceName); + // Get new primary - try twice (unless the caller already holds sentinel's answer) + EndPoint 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; + + // The endpoint set is rebuilt only around a primary this connection has never seen (e.g. a recreated + // node that came back on a brand-new address); a known primary keeps the set it has. Ask for the + // replicas before the claim, so the gate is not held across a sentinel round trip. + bool unknownPrimary = !connection.servers.Contains(newPrimaryEndPoint); + EndPoint[]? replicas = unknownPrimary + ? await GetReplicasForServiceAsync(serviceName).ForAwait() ?? await GetReplicasForServiceAsync(serviceName).ForAwait() + : null; + + // 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; - if (!connection.servers.Contains(newPrimaryEndPoint)) + bool stale = IsStalePrimaryView(connection, newPrimaryEndPoint); + if (unknownPrimary) { - // 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) + if (replicas is not null) { - foreach (var replicaEndPoint in replicaEndPoints) + connection.EndPoints.Clear(); + connection.EndPoints.TryAdd(newPrimaryEndPoint); + foreach (var replica in replicas) { - connection.EndPoints.TryAdd(replicaEndPoint); + connection.EndPoints.TryAdd(replica); } - } - // 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) + // 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(); + } + else { - RetireServersAbsentFromEndPoints(connection, logger); + // Keep the old nodes until discovery works. + connection.EndPoints.TryAdd(newPrimaryEndPoint); } - - TriggerReconfigure(reconfigureAll: false); } - else if (IsStalePrimaryView(connection, newPrimaryEndPoint)) + + bool switched = unknownPrimary || stale; + refreshAddresses |= switched; + if (switched) { - // 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. - TriggerReconfigure(reconfigureAll: true); + // 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: !unknownPrimary, + 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 +938,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 +963,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,33 +990,61 @@ 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); - - // 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 => + (EndPoint[]?)await server.SentinelGetSentinelAddressesAsync(serviceName).ForAwait()).ForAwait(); + if (peers is null) continue; + + 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) + { + hasNew |= EndPoints.TryAdd(peer); + } + 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 - ReconfigureAsync(first: false, reconfigureAll: true, Logger, EndPoints[0], "Updating Sentinel List", false).Wait(); + 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/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/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 0768dbcb7..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; @@ -1174,6 +1180,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..08c46b955 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,88 @@ 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 string? ReplicaError { get; set; } + + /// Answer get-master-addr-by-name with nil, as a sentinel that does not (yet) know the primary does. + public bool LegacyAnswersNothing { 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; + + /// Answer SENTINEL MASTER without its flags field, as some sentinel-compatible servers do. + public bool OmitPrimaryFlags { get; set; } + + private void CountPrimaryQuery() + { + Interlocked.Increment(ref _primaryQueries); + OnPrimaryQuery?.Invoke(); + } + + [RedisCommand(3, nameof(RedisCommand.SENTINEL), "master", LockFree = true)] + private TypedRedisValue SentinelMaster(RedisClient client, in RedisRequest request) + { + if (!SupportsSentinelMaster) return TypedRedisValue.Error("ERR Unknown sentinel subcommand 'master'"); + var reply = PrimaryError is string error ? TypedRedisValue.Error(error) + : IsService(request) ? Describe(Primary, OmitPrimaryFlags ? null : PrimaryFlags) : TypedRedisValue.Error("ERR No such master with that name"); + CountPrimaryQuery(); + return reply; + } + [RedisCommand(3, nameof(RedisCommand.SENTINEL), "get-master-addr-by-name", LockFree = true)] private TypedRedisValue SentinelGetMasterAddrByName(RedisClient client, in RedisRequest request) { - Interlocked.Increment(ref _primaryQueries); - if (!IsService(request)) return TypedRedisValue.NullArray(RespPrefix.Array); + Interlocked.Increment(ref _legacyQueries); + if (!IsService(request) || LegacyAnswersNothing) return TypedRedisValue.NullArray(RespPrefix.Array); var host = GetHost(Primary, out var port); var reply = TypedRedisValue.Rent(2, out var span, RespPrefix.Array); span[0] = TypedRedisValue.BulkString(host); span[1] = TypedRedisValue.BulkString(port.ToString()); + CountPrimaryQuery(); return reply; } @@ -72,16 +141,22 @@ private TypedRedisValue SentinelGetMasterAddrByName(RedisClient client, in Redis [RedisCommand(3, nameof(RedisCommand.SENTINEL), "slaves", LockFree = true)] private TypedRedisValue SentinelSlaves(RedisClient client, in RedisRequest request) => Replicas(request); - // only this sentinel exists [RedisCommand(3, nameof(RedisCommand.SENTINEL), "sentinels", LockFree = true)] private TypedRedisValue SentinelSentinels(RedisClient client, in RedisRequest request) - => IsService(request) ? TypedRedisValue.EmptyArray(RespPrefix.Array) : TypedRedisValue.Error("ERR No such master with that name"); + { + Interlocked.Increment(ref _peerQueries); + if (!IsService(request)) return TypedRedisValue.Error("ERR No such master with that name"); + var peers = PeerFlags.ToArray(); + var reply = TypedRedisValue.Rent(peers.Length, out var span, RespPrefix.Array); + for (int i = 0; i < peers.Length; i++) span[i] = Describe(peers[i].Key, "sentinel," + peers[i].Value); + return reply; + } [RedisCommand(2, nameof(RedisCommand.SENTINEL), "masters", LockFree = true)] private TypedRedisValue SentinelMasters(RedisClient client, in RedisRequest request) { var reply = TypedRedisValue.Rent(1, out var span, RespPrefix.Array); - span[0] = Describe(Primary, "master"); + span[0] = Describe(Primary, PrimaryFlags); return reply; } @@ -91,6 +166,8 @@ private TypedRedisValue Replicas(in RedisRequest request) { if (!IsService(request)) return TypedRedisValue.Error("ERR No such master with that name"); + Interlocked.Increment(ref _replicaQueries); + if (ReplicaError is string error) return TypedRedisValue.Error(error); // sentinel learns the replicas from the primary, so report whatever replicates it right now var replicas = new List(); if (_monitored.TryGetNode(Primary, out var primaryNode)) @@ -104,26 +181,37 @@ private TypedRedisValue Replicas(in RedisRequest request) } } + if (ReplicaAddressesOverride is { } addresses) + { + replicas.Clear(); + replicas.AddRange(addresses); + } + // Real sentinel keeps reporting failed replicas with down/disconnected flags until reset. + foreach (var endpoint in ReplicaFlags.Keys) + if (!replicas.Contains(endpoint)) replicas.Add(endpoint); var reply = TypedRedisValue.Rent(replicas.Count, out var span, RespPrefix.Array); for (int i = 0; i < replicas.Count; i++) { - span[i] = Describe(replicas[i], "slave"); + span[i] = Describe(replicas[i], ReplicaFlags.TryGetValue(replicas[i], out var flags) ? "slave," + flags : "slave"); } return reply; } - private TypedRedisValue Describe(EndPoint endpoint, string flags) + private TypedRedisValue Describe(EndPoint endpoint, string? flags) { var host = GetHost(endpoint, out var port); - var item = TypedRedisValue.Rent(8, out var span, RespPrefix.Array); + var item = TypedRedisValue.Rent(flags is null ? 6 : 8, out var span, RespPrefix.Array); span[0] = TypedRedisValue.BulkString("name"); - span[1] = TypedRedisValue.BulkString(flags == "master" ? ServiceName : $"{host}:{port}"); + span[1] = TypedRedisValue.BulkString(flags is null || flags.StartsWith("master", StringComparison.Ordinal) ? ServiceName : $"{host}:{port}"); span[2] = TypedRedisValue.BulkString("ip"); span[3] = TypedRedisValue.BulkString(host); span[4] = TypedRedisValue.BulkString("port"); span[5] = TypedRedisValue.BulkString(port.ToString()); - span[6] = TypedRedisValue.BulkString("flags"); - span[7] = TypedRedisValue.BulkString(flags); + if (flags is not null) + { + span[6] = TypedRedisValue.BulkString("flags"); + span[7] = TypedRedisValue.BulkString(flags); + } return item; } @@ -139,9 +227,9 @@ private sealed class RoutingTunnel(InProcessTestServer[] servers) : Tunnel } public override ValueTask 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..e3522b67f --- /dev/null +++ b/tests/StackExchange.Redis.Tests/SentinelDiscoveryUnitTests.cs @@ -0,0 +1,95 @@ +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)); + } + + [Theory] + [InlineData("master", true, false)] + [InlineData("master,disconnected", true, false)] + [InlineData("master,s_down", false, true)] + [InlineData("master,o_down,s_down", false, true)] + [InlineData("slave", false, false)] + [InlineData(null, false, false)] + public void OnlyAHealthyMasterIsAValidPrimaryAnswer(string? flags, bool valid, bool heldDown) + { + // heldDown is the only null that passes the sentinel over; any other null asks get-master-addr-by-name + 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.ParseHealthyPrimary(row.ToArray(), out var down); + if (valid) Assert.Equal(new IPEndPoint(IPAddress.Parse("192.0.2.1"), 6379), primary); + else Assert.Null(primary); + Assert.Equal(heldDown, down); + } + + [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.ParseHealthyPrimary(row, out _)); + } + + [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) + { + // an answer with no usable address cannot be judged: the sentinel is asked get-master-addr-by-name instead + KeyValuePair[] row = [new("name", "mymaster"), new("ip", ip), new("port", port), new("flags", "master")]; + Assert.Null(ConnectionMultiplexer.ParseHealthyPrimary(row, out var heldDown)); + Assert.False(heldDown); + } + + [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.ParseHealthyPrimary(row, out _)); + 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)); +} diff --git a/tests/StackExchange.Redis.Tests/SentinelPrimaryDemotionUnitTests.cs b/tests/StackExchange.Redis.Tests/SentinelPrimaryDemotionUnitTests.cs index 2a6a879c0..08e34ad80 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,748 @@ 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 KnownHealthyPrimaryKeepsItsEndPointsAndStopsRecovery() + { + 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()); + + // the replica fails and sentinel no longer lists it; the primary is unchanged + data.AddReplicaNode(NewNodeC, PrimaryA); + RemoveAndDisconnect(data, ReplicaB); + await UntilAsync(() => !conn.GetServer(ReplicaB).IsConnected, "removed replica stayed connected"); + var owner = conn.sentinelConnection!; + owner.OnManagedConnectionFailed(conn, Failure(ReplicaB, conn)); + await UntilAsync(() => conn.sentinelPrimaryReconnectTimer is null, "dead replica kept recovery alive"); + + // the endpoint set is rebuilt only around a primary never seen: the replica keeps its bridge to reconnect + Assert.Equal([PrimaryA], ConnectedPrimaries(conn)); + Assert.Contains(ReplicaB, conn.GetEndPoints()); + Assert.DoesNotContain(NewNodeC, conn.GetEndPoints()); + } + + [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 ReplicaDownAtConnectJoinsWhenItComesBack() + { + using var data = new InProcessTestServer(log, PrimaryA); + using var sentinel = new InProcessSentinelServer(data, ServiceName, PrimaryA, SentinelEndPoint, log); + // a rolling restart: sentinel lists the replica, holding it down, and nothing answers at its address + sentinel.ReplicaFlags[ReplicaB] = "s_down,disconnected"; + await using var conn = await ConnectionMultiplexer.ConnectAsync(CreateConfig(data, sentinel)); + Assert.Contains(ReplicaB, conn.GetEndPoints()); + Assert.False(conn.GetServer(ReplicaB).IsConnected); + await UntilAsync(() => conn.sentinelPrimaryReconnectTimer is null, "connect left a timer"); + + // the replica comes back at the same address; nothing else happens to the connection + data.AddReplicaNode(ReplicaB, PrimaryA); + Assert.True(sentinel.ReplicaFlags.TryRemove(ReplicaB, out _)); + await UntilAsync(() => conn.GetServer(ReplicaB) is { IsConnected: true, IsReplica: true }, "returned replica did not join"); + Assert.Equal([PrimaryA], ConnectedPrimaries(conn)); + } + + [Fact] + public async Task ReplicaDownAtConnectThatNeverReturnsDoesNotKeepRecoveryAlive() + { + using var data = new InProcessTestServer(log, PrimaryA); + using var sentinel = new InProcessSentinelServer(data, ServiceName, PrimaryA, SentinelEndPoint, log); + sentinel.ReplicaFlags[ReplicaB] = "s_down,disconnected"; + await using var conn = await ConnectionMultiplexer.ConnectAsync(CreateConfig(data, sentinel)); + Assert.Contains(ReplicaB, conn.GetEndPoints()); + await UntilAsync(() => conn.sentinelPrimaryReconnectTimer is null, "dead listed replica kept the timer"); + + // its bridge keeps reconnecting on its own; sentinel is not asked again for it + int queries = sentinel.PrimaryQueries; + await Task.Delay(2500); + Assert.Equal(queries, sentinel.PrimaryQueries); + Assert.Null(conn.sentinelPrimaryReconnectTimer); + Assert.False(conn.GetServer(ReplicaB).IsConnected); + await conn.GetDatabase().StringSetAsync("primary-still-works", "yes"); + } + + [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"); + } + + [Theory] + [InlineData("unsupported")] + [InlineData("noperm")] + [InlineData("flagless")] + public async Task SentinelWithoutAUsableSentinelMasterResolvesThePrimaryWithGetMasterAddrByName(string sentinelMaster) + { + using var data = new InProcessTestServer(log, PrimaryA); + data.AddReplicaNode(ReplicaB, PrimaryA); + // the only sentinel: there is no other answer to fall back on + using var sentinel = new InProcessSentinelServer(data, ServiceName, PrimaryA, SentinelEndPoint, log); + switch (sentinelMaster) + { + case "unsupported": sentinel.SupportsSentinelMaster = false; break; + // an ACL user allowed what discovery needed before SENTINEL MASTER was used: get-master-addr-by-name, replicas, sentinels + case "noperm": sentinel.PrimaryError = "NOPERM User client has no permissions to run the 'sentinel|master' command"; break; + case "flagless": sentinel.OmitPrimaryFlags = true; break; + } + await using var conn = await ConnectionMultiplexer.ConnectAsync(CreateConfig(data, sentinel)); + Assert.Equal([PrimaryA], ConnectedPrimaries(conn)); + Assert.Contains(ReplicaB, conn.GetEndPoints()); + Assert.True(sentinel.LegacyQueries > 0); + + FailOverToNewNode(data); + sentinel.Primary = NewNodeC; + await AssertPrimaryBecomesAsync(conn, NewNodeC); + await UntilAsync(() => conn.sentinelPrimaryReconnectTimer is null, "recovery did not settle"); + } + + [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() + { + 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 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() + { + 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 FailedReplicaQueryForANewPrimaryKeepsTheOldNodes() + { + 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)); + await UntilAsync(() => conn.sentinelPrimaryReconnectTimer is null, "connect left a timer"); + var held = conn.TryResolveServerEndPoint(ReplicaB); + + // sentinel names a primary never seen, but the replica query fails: a failed query is not an empty + // topology, so nothing is retired - the new primary is only added + sentinel.ReplicaError = "NOPERM User client has no permissions to run the 'sentinel|replicas' command"; + FailOverToNewNode(data); + sentinel.Primary = NewNodeC; + Assert.True(sentinel.ForAllClients(client => client.Publish( + RedisChannel.Literal("+switch-master"), "mymaster 192.0.2.1 6379 192.0.2.3 6379")) > 0, "no subscriber for +switch-master"); + + await AssertPrimaryBecomesAsync(conn, NewNodeC); + Assert.True(sentinel.ReplicaQueries > 0); + Assert.Contains(PrimaryA, conn.GetEndPoints()); + Assert.Contains(ReplicaB, conn.GetEndPoints()); + Assert.Same(held, conn.TryResolveServerEndPoint(ReplicaB)); + Assert.True(conn.GetServer(ReplicaB).IsConnected); + } + + [Fact] + public async Task ReplicasComeFromAnySentinelThatAnswers() + { + using var data = new InProcessTestServer(log, PrimaryA); + data.AddReplicaNode(ReplicaB, PrimaryA); + // the only sentinel whose primary answer is taken refuses the replica query + using var accepted = new InProcessSentinelServer(data, ServiceName, PrimaryA, SentinelEndPoint, log) + { + ReplicaError = "NOPERM User client has no permissions to run the 'sentinel|replicas' command", + }; + using var other = new InProcessSentinelServer(data, ServiceName, PrimaryA, SecondSentinelEndPoint, log) { PrimaryFlags = "master,s_down" }; + var config = CreateConfig(data, accepted, other); + config.EndPoints.Add(SecondSentinelEndPoint); + await using var conn = await ConnectionMultiplexer.ConnectAsync(config); + Assert.Equal([PrimaryA], ConnectedPrimaries(conn)); + Assert.Contains(ReplicaB, conn.GetEndPoints()); + await UntilAsync(() => conn.GetServer(ReplicaB) is { IsConnected: true, IsReplica: true }, "replica did not join"); + } + + [Fact] + public async Task OldPrimaryStillAnsweringMasterIsRetiredWhenSentinelNamesANewOne() + { + 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)); + await UntilAsync(() => conn.sentinelPrimaryReconnectTimer is null, "connect left a timer"); + + // C, never seen by the client, is promoted and B follows it. A is left out of the service but stays + // reachable and keeps answering role:master; sentinel no longer lists it. + data.AddEmptyNode(NewNodeC); + data.SetReplicaOf(ReplicaB, NewNodeC); + sentinel.Primary = NewNodeC; + Assert.True(sentinel.ForAllClients(client => client.Publish( + RedisChannel.Literal("+switch-master"), "mymaster 192.0.2.1 6379 192.0.2.3 6379")) > 0, "no subscriber for +switch-master"); + + await AssertPrimaryBecomesAsync(conn, NewNodeC); + await UntilAsync(() => conn.sentinelPrimaryReconnectTimer is null, "a second primary kept recovery alive"); + int queries = sentinel.PrimaryQueries; + await Task.Delay(2500); + Assert.Equal(queries, sentinel.PrimaryQueries); + Assert.Null(conn.sentinelPrimaryReconnectTimer); + Assert.DoesNotContain(PrimaryA, conn.GetServerSnapshot().ToArray().Select(x => x.EndPoint)); + } + + [Fact] + public async Task StaleAnswerFromAnErroringSentinelIsCorrectedByTheRoleCheck() + { + using var data = new InProcessTestServer(log, PrimaryA); + data.AddReplicaNode(ReplicaB, PrimaryA); + FailOverToNewNode(data); + // get-master-addr-by-name at the sentinel that refuses SENTINEL MASTER still names A, now a replica of C + using var erroring = new InProcessSentinelServer(data, ServiceName, PrimaryA, SentinelEndPoint, log) + { + PrimaryError = "NOPERM User client has no permissions to run the 'sentinel|master' command", + }; + using var current = new InProcessSentinelServer(data, ServiceName, NewNodeC, SecondSentinelEndPoint, log); + // hold back the current sentinel's first answer, so the stale one is taken first and the role check has to reject it + int replies = 0; + current.BeforePrimaryReplyAsync = () => Interlocked.Increment(ref replies) == 1 ? Task.Delay(500) : Task.CompletedTask; + var config = CreateConfig(data, erroring, current); + config.EndPoints.Add(SecondSentinelEndPoint); + await using var conn = await ConnectionMultiplexer.ConnectAsync(config); + Assert.True(erroring.LegacyQueries > 0, "the erroring sentinel was not asked get-master-addr-by-name"); + Assert.True(Volatile.Read(ref replies) > 1, "the current sentinel was asked only once: the stale answer was never rejected"); + Assert.Equal([NewNodeC], ConnectedPrimaries(conn)); + await UntilAsync(() => conn.sentinelPrimaryReconnectTimer is null, "connect left a timer"); + Assert.Equal([NewNodeC], ConnectedPrimaries(conn)); + } + + [Theory] + [InlineData("ERR No such master with that name")] + [InlineData("LOADING Redis is loading the dataset in memory")] + public async Task SentinelThatAnswersNeitherQueryIsPassedOver(string error) + { + using var data = new InProcessTestServer(log, PrimaryA); + data.AddReplicaNode(ReplicaB, PrimaryA); + FailOverToNewNode(data); + // a real sentinel answers both SENTINEL MASTER and get-master-addr-by-name the same way here + using var erroring = new InProcessSentinelServer(data, ServiceName, PrimaryA, SentinelEndPoint, log) + { + PrimaryError = error, + LegacyAnswersNothing = true, + }; + using var current = new InProcessSentinelServer(data, ServiceName, NewNodeC, SecondSentinelEndPoint, log); + var config = CreateConfig(data, erroring, current); + config.EndPoints.Add(SecondSentinelEndPoint); + await using var conn = await ConnectionMultiplexer.ConnectAsync(config); + await UntilAsync(() => erroring.LegacyQueries > 0, "the erroring sentinel was not asked get-master-addr-by-name"); + Assert.Equal([NewNodeC], ConnectedPrimaries(conn)); + await UntilAsync(() => conn.sentinelPrimaryReconnectTimer is null, "connect left a timer"); + } + + [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 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.TryResolveServerEndPoint(SecondSentinelEndPoint)?.IsConnected == true, "pending reconfigure did not connect the peer"); + } + + [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); + } + } + + [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() + { + 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); + // only the winner's connections are left: the interactive one, and a dedicated subscription one if it was opened + await UntilAsync(() => data.ClientCount == ConnectionsTo(child, PrimaryA), "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); + + // the dedicated subscription connection is opened on demand, so it may not exist at all + private static int ConnectionsTo(ConnectionMultiplexer conn, EndPoint endpoint) + => conn.TryResolveServerEndPoint(endpoint) is { } server + ? (server.IsConnected ? 1 : 0) + (!server.SharesSubscriptionConnection() && server.IsSubscriberConnected ? 1 : 0) + : 0; + + 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..bd9f78eb1 100644 --- a/tests/StackExchange.Redis.Tests/SentinelTests.cs +++ b/tests/StackExchange.Redis.Tests/SentinelTests.cs @@ -500,7 +500,7 @@ 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 + // endpoint this connection has never seen, which is what drives SwitchPrimaryAsync'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", @@ -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))); @@ -525,6 +525,32 @@ await UntilConditionAsync( await managed.GetDatabase().PingAsync(); } + [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() {