diff --git a/docs/Configuration.md b/docs/Configuration.md index 1df31a2ae..6efc1d523 100644 --- a/docs/Configuration.md +++ b/docs/Configuration.md @@ -46,7 +46,11 @@ connection to the primary is lost, and - because the announcement is delivered o notices when the node last named as primary is still connected but now reports itself as a replica (a failover where nothing disconnected, such as a primary demoted in place). In that last case the client re-asks sentinel for the current primary and follows its answer, even to a node at an address it has never seen; while sentinel still names the demoted node (sentinel itself has not yet failed over), the client -keeps asking without tearing down or reconfiguring the healthy connections. +keeps asking without tearing down or reconfiguring the healthy connections. A command answered with `READONLY` is the same signal: the node +is marked as a replica at once, rather than at the next role check (`ConfigCheckSeconds`), and the client asks sentinel on its next heartbeat. +Sentinel is queried asynchronously - waiting on a slow or unreachable sentinel holds no thread-pool thread - and the client only follows a +sentinel that holds the node it names as a healthy master (`SENTINEL MASTER` flags without `s_down`/`o_down`; sentinels without +`SENTINEL MASTER` are asked with `get-master-addr-by-name`). An overview of mapping between the `string` and `ConfigurationOptions` representation is shown below, but you can switch between them trivially: diff --git a/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs b/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs index 31f5296ce..6901dfa27 100644 --- a/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs +++ b/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs @@ -19,6 +19,12 @@ 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; + private HashSet _sentinelDownRoutes = new(); + internal bool IsSentinelManaged { get; private set; } /// /// Initializes the connection as a Sentinel connection and adds the necessary event handlers to track changes to the managed primaries. @@ -50,12 +56,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 +78,7 @@ internal void InitializeSentinel(ILogger? log) if (switchTarget is not null) { - SwitchPrimary(switchBlame, switchTarget); + SwitchPrimaryAsync(switchBlame, switchTarget).RedisFireAndForget(); } }, CommandFlags.FireAndForget); @@ -86,7 +88,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 +98,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 +194,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 +202,7 @@ private static async Task SentinelPrimaryConnectAsync(Con } catch { - try { sentinelConnection.Dispose(); } catch { } + try { await sentinelConnection.DisposeAsync().ForAwait(); } catch { } throw; } } @@ -208,6 +213,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)) @@ -243,13 +252,13 @@ public ConnectionMultiplexer GetSentinelMasterConnection(ConfigurationOptions co // Sentinel has some fun race behavior internally - give things a few shots for a quicker overall connect. const int queryAttempts = 2; - EndPoint? newPrimaryEndPoint = null; - for (int i = 0; i < queryAttempts && newPrimaryEndPoint is null; i++) + SentinelPrimary? primary = null; + for (int i = 0; i < queryAttempts && primary is null; i++) { - newPrimaryEndPoint = GetConfiguredPrimaryForService(serviceName); + primary = await GetConfiguredPrimaryForServiceAsync(serviceName).ForAwait(); } - if (newPrimaryEndPoint is null) + if (primary is null) { failure = new RedisConnectionException( ConnectionFailureType.UnableToConnect, @@ -260,15 +269,16 @@ 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; + var newPrimaryEndPoint = primary.EndPoint; + SentinelReplica[]? replicaEndPoints = null; for (int i = 0; i < queryAttempts && replicaEndPoints is null; i++) { - replicaEndPoints = GetReplicasForService(serviceName); + replicaEndPoints = await GetReplicasForServiceAsync(serviceName, primary).ForAwait(); } endpoints = config.EndPoints.Clone(); @@ -287,17 +297,27 @@ public ConnectionMultiplexer GetSentinelMasterConnection(ConfigurationOptions co if (replicaEndPoints is not null) { - foreach (var replicaEndPoint in replicaEndPoints) + // a replica sentinel reports down is not connected to: it is added once its flags clear + foreach (var replica in replicaEndPoints) { - endpoints.TryAdd(replicaEndPoint); + if (!replica.Down) endpoints.TryAdd(replica.EndPoint); } } - 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 +326,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 +342,7 @@ public ConnectionMultiplexer GetSentinelMasterConnection(ConfigurationOptions co } } - Thread.Sleep(100); + await Task.Delay(100).ForAwait(); } while (sw.ElapsedMilliseconds < config.ConnectTimeout); @@ -337,28 +357,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 +403,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 +421,7 @@ public ConnectionMultiplexer GetSentinelMasterConnection(ConfigurationOptions co sentinelConnectionChildren.Remove(serviceName); } } - try { connection.Dispose(); } catch { } + try { await connection.DisposeAsync().ForAwait(); } catch { } } throw; } @@ -393,127 +430,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 +527,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 +578,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 +612,31 @@ 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 SentinelPrimary primary) return; + var reported = primary.EndPoint; 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: primary).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 +676,301 @@ 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 => + internal sealed class SentinelPrimary(IServer sentinel, EndPoint endPoint) + { + internal IServer Sentinel { get; } = sentinel; + internal EndPoint EndPoint { get; } = endPoint; + } + + /// + /// The primary sentinel holds for : the first valid answer from any sentinel route. + /// + /// + /// https://redis.io/docs/latest/develop/reference/sentinel-clients/ Step 2 asks the sentinels for the address; + /// an answer is taken only from a sentinel that holds that node as a healthy master (flags master, no + /// s_down/o_down), which SENTINEL MASTER reports and get-master-addr-by-name does not - + /// a sentinel still repeating a failed primary is passed over for one that already knows the new one. A stale + /// but healthy-looking answer is still caught by the role check (Step 3) and asked again. + /// + internal Task GetConfiguredPrimaryForServiceAsync(string serviceName) => + QuerySentinelsAsync(async server => + { + EndPoint? primary; + try + { + primary = ParseValidPrimary(await server.SentinelMasterAsync(serviceName).ForAwait()); + } + catch (RedisServerException ex) when (IsSentinelMasterUnsupported(ex)) { - 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 = await server.SentinelGetMasterAddressByNameAsync(serviceName).ForAwait(); + } + return primary is null ? null : new SentinelPrimary(server, primary); + }); + + internal static bool IsSentinelMasterUnsupported(RedisServerException ex) => + ex.Message.IndexOf("unknown command", StringComparison.OrdinalIgnoreCase) >= 0 + || ex.Message.IndexOf("unknown subcommand", StringComparison.OrdinalIgnoreCase) >= 0 + || ex.Message.IndexOf("unknown sentinel subcommand", StringComparison.OrdinalIgnoreCase) >= 0 + || ex.Message.IndexOf("unrecognized sentinel command", StringComparison.OrdinalIgnoreCase) >= 0; + + private static async Task GetReplicasForServiceAsync(string serviceName, SentinelPrimary primary) + { + try + { + // Use the same sentinel route whose primary answer won; a second race could combine unrelated views. + var rows = await primary.Sentinel.SentinelReplicasAsync(serviceName).ForAwait(); + var replicas = new List(rows.Length); + foreach (var row in rows) + { + if (ParseSentinelAddress(row, out var down) is EndPoint address) replicas.Add(new SentinelReplica(address, down)); + } + return replicas.ToArray(); + } + catch (Exception) { return null; } + } + + /// + /// Asks every sentinel at once and takes the first answer, as the parallel query this replaces did - but through + /// the asynchronous API: a request waiting on a slow or dead sentinel holds no thread-pool worker. + /// + private Task QuerySentinelsAsync(Func> query) where T : class + { + return FirstSentinelResultAsync(GetSentinelsForQuery().Select(QueryAsync).ToArray(), AsyncTimeoutMilliseconds); + + async Task QueryAsync(ServerEndPoint server) + { + try { return await query(GetServer(server.EndPoint)).ForAwait(); } + catch (Exception) { return null; } + } + } + + // sentinel-clients Step 1 tries the sentinels one by one, moving the one that answered to the front. Every route + // is asked at once instead - the requests are asynchronous and hold no worker, and a Service address and the + // advertised pod address of the same sentinel need not be told apart - connected routes first and routes + // sentinel itself reports down last, so a dead route can delay nothing and the first valid answer wins. + private ServerEndPoint[] GetSentinelsForQuery() + { + var down = Volatile.Read(ref _sentinelDownRoutes); + return OrderSentinelQueryTargets( + _serverSnapshot.Where(static s => s.ServerType == ServerType.Sentinel && !s.IsDisposed).ToArray(), + static s => s.IsConnected, + s => down.Contains(s.EndPoint)); + } + + internal static T[] OrderSentinelQueryTargets(T[] targets, Func connected, Func reportedDown) + => targets.OrderBy(t => connected(t) ? 0 : reportedDown(t) ? 2 : 1).ToArray(); // OrderBy is stable + + internal static bool IsSentinelNodeDown(string? flags) => flags is not null + && flags.Split(',').Any(static f => f is "s_down" or "o_down" or "disconnected"); + + // a primary is unusable only when sentinel holds it down; "disconnected" is a link sentinel is re-establishing + internal static bool IsSentinelPrimaryDown(string? flags) => flags is not null + && flags.Split(',').Any(static f => f is "s_down" or "o_down"); + + /// + /// A replica as sentinel reports it. Sentinel never forgets a replica (sentinel docs: "Sentinels never forget + /// about replicas of a given master" until SENTINEL RESET), so a failed one stays listed with down flags. + /// + internal readonly struct SentinelReplica(EndPoint endPoint, bool down) + { + public EndPoint EndPoint { get; } = endPoint; + public bool Down { get; } = down; + } + + internal static EndPoint? ParseValidPrimary(KeyValuePair[] row) + { + var fields = row.ToDictionary(static p => p.Key, static p => p.Value); + if (!fields.TryGetValue("flags", out var flags) || !flags.Split(',').Contains("master") || IsSentinelPrimaryDown(flags)) return null; + return ParseSentinelAddress(fields); + } + + private static EndPoint? ParseSentinelAddress(KeyValuePair[] row, out bool down) + { + var fields = row.ToDictionary(static p => p.Key, static p => p.Value); + down = fields.TryGetValue("flags", out var flags) && IsSentinelNodeDown(flags); + return ParseSentinelAddress(fields); + } + + private static EndPoint? ParseSentinelAddress(Dictionary fields) + { + if (fields.TryGetValue("ip", out var host) && fields.TryGetValue("port", out var port) + && Format.TryParseEndPoint(host, port, out var address)) return address; + return null; + } + + // Every request observes its own failure, including the requests still outstanding when a faster + // sentinel answers. One deadline covers the whole race, rather than a timeout per losing request. + // All requests use the asynchronous API: waiting for a slow/dead sentinel consumes no worker. + 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, SentinelPrimary? 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(); - - UpdateSentinelAddressList(serviceName); + logger?.LogInformationServiceNameNotDefined(); + return false; } - if (!connection.servers.Contains(newPrimaryEndPoint)) + // Get new primary - try twice (unless the caller already holds sentinel's answer) + var primary = reportedPrimary + ?? await GetConfiguredPrimaryForServiceAsync(serviceName).ForAwait() + ?? await GetConfiguredPrimaryForServiceAsync(serviceName).ForAwait() + ?? throw new RedisConnectionException(ConnectionFailureType.UnableToConnect, CommandFlags.None, $"Sentinel: Failed connecting to switch primary for service: {serviceName}"); + var newPrimaryEndPoint = primary.EndPoint; + connection.currentSentinelPrimaryEndPoint = confirmedPrimary = newPrimaryEndPoint; + + // Refresh replicas even when the primary is already known. Otherwise an unreachable old replica + // (or an old primary) stays in the snapshot forever after a node returns at a different address. + var replicas = await GetReplicasForServiceAsync(serviceName, primary).ForAwait(); + + // Claim the reconfigure gate BEFORE reshaping the connection. ReconfigureAsync refuses to run while + // another reconfigure holds this gate, so rewriting the endpoint list and retiring servers first would + // 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)) { - // 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); + Trace($"Primary switch for {serviceName} skipped; reconfiguration already in progress"); + return false; + } + claimed = true; - connection.EndPoints.Clear(); - connection.EndPoints.TryAdd(newPrimaryEndPoint); - if (replicaEndPoints is not null) + bool unknownPrimary = !connection.servers.Contains(newPrimaryEndPoint); + bool stale = IsStalePrimaryView(connection, newPrimaryEndPoint); + bool changed = false; + if (replicas is not null) + { + // Down flags (s_down/o_down/disconnected) decide only whether a node not yet held is added. A node + // this connection already holds stays whatever one sentinel's momentary view says - sentinel reports + // the demoted primary disconnected while it reconnects to it, and a live connection outranks that. + // Only a disconnected node missing from a nonempty reply by the same sentinel may leave; + // an incomplete or empty view must not retire live nodes. + var desired = new EndPointCollection { newPrimaryEndPoint }; + foreach (var held in connection.EndPoints) { - foreach (var replicaEndPoint in replicaEndPoints) - { - connection.EndPoints.TryAdd(replicaEndPoint); - } + if (replicas.Length == 0 || connection.TryResolveServerEndPoint(held) is { IsConnected: true }) desired.TryAdd(held); } - - // 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) + foreach (var replica in replicas) { - RetireServersAbsentFromEndPoints(connection, logger); + if (!replica.Down || connection.EndPoints.Contains(replica.EndPoint)) desired.TryAdd(replica.EndPoint); + } + changed = !new HashSet(connection.EndPoints).SetEquals(desired); + if (changed) + { + connection.EndPoints.Clear(); + foreach (var endpoint in desired) connection.EndPoints.TryAdd(endpoint); + await RetireServersAbsentFromEndPointsAsync(connection, logger).ForAwait(); } - - TriggerReconfigure(reconfigureAll: false); } - else if (IsStalePrimaryView(connection, newPrimaryEndPoint)) + else if (unknownPrimary) { - // The sentinel-reported primary IS an endpoint we already know about, but our - // cached view of the topology is stale (failover between known nodes: the old - // primary was demoted to a replica in-place and a former replica was promoted). - // A plain reconfigure with reconfigureAll:false would NOT re-read the role of an - // already-connected node, so the primary election would re-pick the stale demoted - // node. Force a full reconfigure so every connected node's role is refreshed before - // the primary is re-elected. - TriggerReconfigure(reconfigureAll: true); + // A failed replica query is not an empty topology. Keep the old nodes until discovery works. + connection.EndPoints.TryAdd(newPrimaryEndPoint); + changed = true; } + bool switched = changed || stale; + refreshAddresses |= switched; + if (switched) + { + // Step 3: a new node reads its role during handshake. For a known primary, reconfigureAll forces + // INFO on connected nodes too; an unknown primary can leave cached roles on old nodes until the + // next recovery attempt. Any remaining replica/second-primary view keeps recovery running. + Trace($"Switching primary to {newPrimaryEndPoint}"); + handedOver = true; // from here the claim is ReconfigureAsync's to release, however it exits + await connection.ReconfigureAsync( + first: false, + reconfigureAll: !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 +991,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 +1016,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 +1043,65 @@ 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 => + (KeyValuePair[][]?)await server.SentinelSentinelsAsync(serviceName).ForAwait()).ForAwait(); + if (peers is null) continue; + + var down = new HashSet(); + bool hasNew = false; + // https://valkey.io/topics/sentinel-clients/: append unknown ip:port pairs, retaining all old routes. + // Add before claiming reconfigure, so a busy claim cannot lose the discovery result. + foreach (var peer in peers) + { + if (ParseSentinelAddress(peer, out var isDown) is not EndPoint address) continue; + hasNew |= EndPoints.TryAdd(address); + if (isDown) down.Add(address); + } + Volatile.Write(ref _sentinelDownRoutes, down); + if (!hasNew) continue; - if (hasNew) + const string cause = "Updating Sentinel List"; + while (!IsDisposed) + { + if (TryClaimReconfigure(cause)) + { + await ReconfigureAsync(first: false, reconfigureAll: true, Logger, EndPoints[0], cause, false, claimed: true).ForAwait(); + break; + } + // Coalesce later announcements while waiting for the owner of the claim; no worker is held. + await Task.Delay(1000).ForAwait(); + } + } + } + finally { - // Reconfigure the sentinel multiplexer if we added new endpoints - 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..e33618e5e 100644 --- a/tests/StackExchange.Redis.Tests/InProcessSentinelServer.cs +++ b/tests/StackExchange.Redis.Tests/InProcessSentinelServer.cs @@ -1,4 +1,5 @@ using System; +using System.Collections.Concurrent; using System.Collections.Generic; using System.IO; using System.Net; @@ -49,20 +50,81 @@ public EndPoint Primary /// public static Tunnel CreateTunnel(params InProcessTestServer[] servers) => new RoutingTunnel(servers); - private int _primaryQueries; + private int _primaryQueries, _peerQueries, _replicaQueries, _legacyQueries; + private string _primaryFlags = "master"; /// How many times a client has asked this sentinel for the primary. public int PrimaryQueries => Volatile.Read(ref _primaryQueries); + public Action? OnPrimaryQuery { get; set; } + public int PeerQueries => Volatile.Read(ref _peerQueries); + public int ReplicaQueries => Volatile.Read(ref _replicaQueries); + public int LegacyQueries => Volatile.Read(ref _legacyQueries); + public string? PrimaryError { get; set; } + public EndPoint[]? ReplicaAddressesOverride { get; set; } + public Func? BeforePrimaryReplyAsync { get; set; } + public Func? BeforePeerReplyAsync { get; set; } + + protected override ValueTask ClientPauseAsync(RedisClient client, in RedisRequest request) + { + var pause = base.ClientPauseAsync(client, request); + var hook = request.Count > 1 && request.KnownCommand == RedisCommand.SENTINEL + ? request.GetString(1).ToLowerInvariant() switch + { + "master" or "get-master-addr-by-name" => BeforePrimaryReplyAsync, + "sentinels" => BeforePeerReplyAsync, + _ => null, + } : null; + return hook is null ? pause : new ValueTask(AwaitPausedReplyAsync(hook, pause)); + + static async Task AwaitPausedReplyAsync(Func hook, ValueTask pause) + { + // Execute has already captured the response; the barrier controls delivery, not the view being read. + await hook(); + await pause; + } + } + + /// Replicas to report with flags, including nodes that are no longer reachable. + public ConcurrentDictionary ReplicaFlags { get; } = new(); + + /// Peer routes to advertise; deliberately no MYID/runid support is needed for address discovery. + public ConcurrentDictionary PeerFlags { get; } = new(); + + public string PrimaryFlags + { + get => Volatile.Read(ref _primaryFlags); + set => Volatile.Write(ref _primaryFlags, value); + } + + public bool SupportsSentinelMaster { get; set; } = true; + + private void CountPrimaryQuery() + { + Interlocked.Increment(ref _primaryQueries); + OnPrimaryQuery?.Invoke(); + } + + [RedisCommand(3, nameof(RedisCommand.SENTINEL), "master", LockFree = true)] + private TypedRedisValue SentinelMaster(RedisClient client, in RedisRequest request) + { + if (!SupportsSentinelMaster) return TypedRedisValue.Error("ERR Unknown sentinel subcommand 'master'"); + var reply = PrimaryError is string error ? TypedRedisValue.Error(error) + : IsService(request) ? Describe(Primary, PrimaryFlags) : TypedRedisValue.Error("ERR No such master with that name"); + CountPrimaryQuery(); + return reply; + } + [RedisCommand(3, nameof(RedisCommand.SENTINEL), "get-master-addr-by-name", LockFree = true)] private TypedRedisValue SentinelGetMasterAddrByName(RedisClient client, in RedisRequest request) { - Interlocked.Increment(ref _primaryQueries); + Interlocked.Increment(ref _legacyQueries); if (!IsService(request)) return TypedRedisValue.NullArray(RespPrefix.Array); var host = GetHost(Primary, out var port); var reply = TypedRedisValue.Rent(2, out var span, RespPrefix.Array); span[0] = TypedRedisValue.BulkString(host); span[1] = TypedRedisValue.BulkString(port.ToString()); + CountPrimaryQuery(); return reply; } @@ -72,16 +134,22 @@ private TypedRedisValue SentinelGetMasterAddrByName(RedisClient client, in Redis [RedisCommand(3, nameof(RedisCommand.SENTINEL), "slaves", LockFree = true)] private TypedRedisValue SentinelSlaves(RedisClient client, in RedisRequest request) => Replicas(request); - // only this sentinel exists [RedisCommand(3, nameof(RedisCommand.SENTINEL), "sentinels", LockFree = true)] private TypedRedisValue SentinelSentinels(RedisClient client, in RedisRequest request) - => IsService(request) ? TypedRedisValue.EmptyArray(RespPrefix.Array) : TypedRedisValue.Error("ERR No such master with that name"); + { + Interlocked.Increment(ref _peerQueries); + if (!IsService(request)) return TypedRedisValue.Error("ERR No such master with that name"); + var peers = PeerFlags.ToArray(); + var reply = TypedRedisValue.Rent(peers.Length, out var span, RespPrefix.Array); + for (int i = 0; i < peers.Length; i++) span[i] = Describe(peers[i].Key, "sentinel," + peers[i].Value); + return reply; + } [RedisCommand(2, nameof(RedisCommand.SENTINEL), "masters", LockFree = true)] private TypedRedisValue SentinelMasters(RedisClient client, in RedisRequest request) { var reply = TypedRedisValue.Rent(1, out var span, RespPrefix.Array); - span[0] = Describe(Primary, "master"); + span[0] = Describe(Primary, PrimaryFlags); return reply; } @@ -91,6 +159,7 @@ private TypedRedisValue Replicas(in RedisRequest request) { if (!IsService(request)) return TypedRedisValue.Error("ERR No such master with that name"); + Interlocked.Increment(ref _replicaQueries); // sentinel learns the replicas from the primary, so report whatever replicates it right now var replicas = new List(); if (_monitored.TryGetNode(Primary, out var primaryNode)) @@ -104,10 +173,18 @@ private TypedRedisValue Replicas(in RedisRequest request) } } + if (ReplicaAddressesOverride is { } addresses) + { + replicas.Clear(); + replicas.AddRange(addresses); + } + // Real sentinel keeps reporting failed replicas with down/disconnected flags until reset. + foreach (var endpoint in ReplicaFlags.Keys) + if (!replicas.Contains(endpoint)) replicas.Add(endpoint); var reply = TypedRedisValue.Rent(replicas.Count, out var span, RespPrefix.Array); for (int i = 0; i < replicas.Count; i++) { - span[i] = Describe(replicas[i], "slave"); + span[i] = Describe(replicas[i], ReplicaFlags.TryGetValue(replicas[i], out var flags) ? "slave," + flags : "slave"); } return reply; } @@ -117,7 +194,7 @@ private TypedRedisValue Describe(EndPoint endpoint, string flags) var host = GetHost(endpoint, out var port); var item = TypedRedisValue.Rent(8, out var span, RespPrefix.Array); span[0] = TypedRedisValue.BulkString("name"); - span[1] = TypedRedisValue.BulkString(flags == "master" ? ServiceName : $"{host}:{port}"); + span[1] = TypedRedisValue.BulkString(flags.StartsWith("master", StringComparison.Ordinal) ? ServiceName : $"{host}:{port}"); span[2] = TypedRedisValue.BulkString("ip"); span[3] = TypedRedisValue.BulkString(host); span[4] = TypedRedisValue.BulkString("port"); @@ -139,9 +216,9 @@ private sealed class RoutingTunnel(InProcessTestServer[] servers) : Tunnel } public override ValueTask GetSocketConnectEndpointAsync(EndPoint endpoint, CancellationToken cancellationToken) - => Route(endpoint) is { } tunnel ? tunnel.GetSocketConnectEndpointAsync(endpoint, cancellationToken) : base.GetSocketConnectEndpointAsync(endpoint, cancellationToken); + => Route(endpoint) is { } tunnel ? tunnel.GetSocketConnectEndpointAsync(endpoint, cancellationToken) : throw new IOException("No in-process route for " + endpoint); public override ValueTask BeforeAuthenticateAsync(EndPoint endpoint, ConnectionType connectionType, Socket? socket, CancellationToken cancellationToken) - => Route(endpoint) is { } tunnel ? tunnel.BeforeAuthenticateAsync(endpoint, connectionType, socket, cancellationToken) : base.BeforeAuthenticateAsync(endpoint, connectionType, socket, cancellationToken); + => Route(endpoint) is { } tunnel ? tunnel.BeforeAuthenticateAsync(endpoint, connectionType, socket, cancellationToken) : throw new IOException("No in-process route for " + endpoint); } } diff --git a/tests/StackExchange.Redis.Tests/SentinelDiscoveryUnitTests.cs b/tests/StackExchange.Redis.Tests/SentinelDiscoveryUnitTests.cs new file mode 100644 index 000000000..a097ef51b --- /dev/null +++ b/tests/StackExchange.Redis.Tests/SentinelDiscoveryUnitTests.cs @@ -0,0 +1,111 @@ +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)); + } + + [Fact] + public void EveryRouteIsQueriedConnectedFirstAndReportedDownLast() + { + // sentinel-clients Step 1: no route is left out, a dead route only moves back in the launch order + string[] routes = ["dead-a", "service-a", "down-b", "service-b", "pod-c", "service-c"]; + var ordered = ConnectionMultiplexer.OrderSentinelQueryTargets(routes, s => s.StartsWith("service-"), s => s.StartsWith("down-")); + Assert.Equal(["service-a", "service-b", "service-c", "dead-a", "pod-c", "down-b"], ordered); + Assert.Empty(ConnectionMultiplexer.OrderSentinelQueryTargets(Array.Empty(), _ => true, _ => false)); + } + + [Theory] + [InlineData("master", true)] + [InlineData("master,disconnected", true)] + [InlineData("master,s_down", false)] + [InlineData("master,o_down,s_down", false)] + [InlineData("slave", false)] + [InlineData(null, false)] + public void OnlyAHealthyMasterIsAValidPrimaryAnswer(string? flags, bool valid) + { + var row = new List> + { + new("name", "mymaster"), + new("ip", "192.0.2.1"), + new("port", "6379"), + }; + if (flags is not null) row.Add(new("flags", flags)); + var primary = ConnectionMultiplexer.ParseValidPrimary(row.ToArray()); + if (valid) Assert.Equal(new IPEndPoint(IPAddress.Parse("192.0.2.1"), 6379), primary); + else Assert.Null(primary); + } + + [Fact] + public void IPv6PrimaryAddressIsParsed() + { + KeyValuePair[] row = [new("name", "m"), new("ip", "2001:db8::1"), new("port", "6379"), new("flags", "master")]; + Assert.Equal(new IPEndPoint(IPAddress.Parse("2001:db8::1"), 6379), ConnectionMultiplexer.ParseValidPrimary(row)); + } + + [Theory] + [InlineData("192.0.2.1", "1:2")] + [InlineData("192.0.2.1", "port")] + [InlineData("192.0.2.1", "")] + [InlineData("", "6379")] + public void MalformedAddressIsNotAValidPrimaryAnswer(string ip, string port) + { + // a sentinel whose answer has no usable address is passed over, like one that does not know the service + KeyValuePair[] row = [new("name", "mymaster"), new("ip", ip), new("port", port), new("flags", "master")]; + Assert.Null(ConnectionMultiplexer.ParseValidPrimary(row)); + } + + [Fact] + public void AddressIsReadAsAHostNameNeverAsALocalSocket() + { + // "!path" is the configuration spelling of a Unix socket; an address sentinel reports is never one + KeyValuePair[] row = [new("name", "m"), new("ip", "!/tmp/redis.sock"), new("port", "6379"), new("flags", "master")]; + var primary = Assert.IsType(ConnectionMultiplexer.ParseValidPrimary(row)); + Assert.Equal("!/tmp/redis.sock", primary.Host); + Assert.Equal(6379, primary.Port); + } + + [Theory] + [InlineData("master,s_down", true)] + [InlineData("master,o_down", true)] + [InlineData("master,disconnected", false)] + [InlineData("master", false)] + public void OnlyDownFlagsDisqualifyAPrimary(string flags, bool down) + => Assert.Equal(down, ConnectionMultiplexer.IsSentinelPrimaryDown(flags)); + + [Theory] + [InlineData("slave,s_down", true)] + [InlineData("sentinel,o_down", true)] + [InlineData("slave,disconnected", true)] + [InlineData("slave", false)] + [InlineData(null, false)] + public void DownFlagsAreExcludedFromManagedDiscovery(string? flags, bool down) + => Assert.Equal(down, ConnectionMultiplexer.IsSentinelNodeDown(flags)); +} diff --git a/tests/StackExchange.Redis.Tests/SentinelPrimaryDemotionUnitTests.cs b/tests/StackExchange.Redis.Tests/SentinelPrimaryDemotionUnitTests.cs index 2a6a879c0..8f61f11bd 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,649 @@ 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 KnownHealthyPrimaryRetiresAReplicaNoLongerAdvertisedAndStopsRecovery() + { + using var data = new InProcessTestServer(log, PrimaryA); + data.AddReplicaNode(ReplicaB, PrimaryA); + using var sentinel = new InProcessSentinelServer(data, ServiceName, PrimaryA, SentinelEndPoint, log); + await using var conn = await ConnectionMultiplexer.ConnectAsync(CreateConfig(data, sentinel)); + Assert.Contains(ReplicaB, conn.GetEndPoints()); + + data.AddReplicaNode(NewNodeC, PrimaryA); + RemoveAndDisconnect(data, ReplicaB); + await UntilAsync(() => !conn.GetServer(ReplicaB).IsConnected, "removed replica stayed connected"); + var owner = conn.sentinelConnection!; + var args = new ConnectionFailedEventArgs(null, conn, ReplicaB, ConnectionType.Interactive, ConnectionFailureType.SocketFailure, null, null); + owner.OnManagedConnectionFailed(conn, args); + await UntilAsync(() => !conn.GetEndPoints().Contains(ReplicaB) && conn.sentinelPrimaryReconnectTimer is null, "dead replica kept recovery alive"); + Assert.Equal([PrimaryA], ConnectedPrimaries(conn)); + Assert.DoesNotContain(ReplicaB, conn.GetServerSnapshot().ToArray().Select(x => x.EndPoint)); + } + + [Fact] + public async Task HeldReplicaReportedDownIsNotRetired() + { + using var data = new InProcessTestServer(log, PrimaryA); + data.AddReplicaNode(ReplicaB, PrimaryA); + using var sentinel = new InProcessSentinelServer(data, ServiceName, PrimaryA, SentinelEndPoint, log); + await using var conn = await ConnectionMultiplexer.ConnectAsync(CreateConfig(data, sentinel)); + Assert.Contains(ReplicaB, conn.GetEndPoints()); + + // sentinel's momentary view (it is reconnecting to the node) against our live connection: the connection wins + sentinel.ReplicaFlags[ReplicaB] = "s_down,disconnected"; + var owner = conn.sentinelConnection!; + int changes = 0; + conn.ConfigurationChanged += (_, _) => Interlocked.Increment(ref changes); + Assert.False(await owner.SwitchPrimaryAsync(ReplicaB, conn)); + Assert.Contains(ReplicaB, conn.GetEndPoints()); + Assert.Contains(ReplicaB, conn.GetServerSnapshot().ToArray().Select(x => x.EndPoint)); + Assert.True(conn.GetServer(ReplicaB).IsConnected); + Assert.Equal(0, Volatile.Read(ref changes)); + Assert.Null(conn.sentinelPrimaryReconnectTimer); + Assert.Equal([PrimaryA], ConnectedPrimaries(conn)); + } + + [Fact] + public async Task ReplicaReportedDownJoinsOnRecoveryOnceItsFlagsClear() + { + using var data = new InProcessTestServer(log, PrimaryA); + using var sentinel = new InProcessSentinelServer(data, ServiceName, PrimaryA, SentinelEndPoint, log); + await using var conn = await ConnectionMultiplexer.ConnectAsync(CreateConfig(data, sentinel)); + var owner = conn.sentinelConnection!; + Assert.Equal([PrimaryA], conn.GetEndPoints()); + + // a replica appears, but sentinel still holds it down: not a node to connect to yet + data.AddReplicaNode(ReplicaB, PrimaryA); + sentinel.ReplicaFlags[ReplicaB] = "s_down"; + Assert.False(await owner.SwitchPrimaryAsync(null, conn)); + Assert.Equal([PrimaryA], conn.GetEndPoints()); + Assert.Null(conn.sentinelPrimaryReconnectTimer); + + // The flags clear; the next recovery refresh adds the replica without retiring a held node. + Assert.True(sentinel.ReplicaFlags.TryRemove(ReplicaB, out _)); + Assert.True(await owner.SwitchPrimaryAsync(null, conn)); + await UntilAsync(() => conn.GetEndPoints().Contains(ReplicaB) && conn.GetServer(ReplicaB).IsConnected, "cleared replica did not join"); + Assert.Equal([PrimaryA], ConnectedPrimaries(conn)); + Assert.Contains(PrimaryA, conn.GetServerSnapshot().ToArray().Select(x => x.EndPoint)); + await UntilAsync(() => conn.sentinelPrimaryReconnectTimer is null, "recovery left a timer"); + } + + [Fact] + public async Task StaleSentinelNamingADownPrimaryIsPassedOver() + { + using var data = new InProcessTestServer(log, PrimaryA); + data.AddReplicaNode(ReplicaB, PrimaryA); + FailOverToNewNode(data); + // the stale sentinel still names A, but holds it down; the other already knows C + using var stale = new InProcessSentinelServer(data, ServiceName, PrimaryA, SentinelEndPoint, log) { PrimaryFlags = "master,s_down" }; + using var current = new InProcessSentinelServer(data, ServiceName, NewNodeC, SecondSentinelEndPoint, log); + var config = CreateConfig(data, stale, current); + config.EndPoints.Add(SecondSentinelEndPoint); + await using var conn = await ConnectionMultiplexer.ConnectAsync(config); + + Assert.Equal([NewNodeC], ConnectedPrimaries(conn)); + Assert.DoesNotContain(PrimaryA, conn.GetServerSnapshot().ToArray().Where(x => x.IsConnected && !x.IsReplica).Select(x => x.EndPoint)); + Assert.True(stale.PrimaryQueries > 0, "the stale sentinel was never asked"); + await UntilAsync(() => conn.sentinelPrimaryReconnectTimer is null, "connect left a timer"); + } + + [Fact] + public async Task LegacySentinelWithoutSentinelMasterStillResolvesThePrimary() + { + using var data = new InProcessTestServer(log, PrimaryA); + data.AddReplicaNode(ReplicaB, PrimaryA); + using var sentinel = new InProcessSentinelServer(data, ServiceName, PrimaryA, SentinelEndPoint, log) { SupportsSentinelMaster = false }; + await using var conn = await ConnectionMultiplexer.ConnectAsync(CreateConfig(data, sentinel)); + Assert.Equal([PrimaryA], ConnectedPrimaries(conn)); + Assert.Contains(ReplicaB, conn.GetEndPoints()); + + FailOverToNewNode(data); + sentinel.Primary = NewNodeC; + await AssertPrimaryBecomesAsync(conn, NewNodeC); + } + + [Fact] + public async Task 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"); + } + + [Theory] + [InlineData(true)] + [InlineData(false)] + public async Task IncompleteReplicaViewDoesNotRetireConnectedNodes(bool empty) + { + using var data = new InProcessTestServer(log, PrimaryA); + data.AddReplicaNode(ReplicaB, PrimaryA); + data.AddReplicaNode(NewNodeC, PrimaryA); + using var sentinel = new InProcessSentinelServer(data, ServiceName, PrimaryA, SentinelEndPoint, log); + await using var conn = await ConnectionMultiplexer.ConnectAsync(CreateConfig(data, sentinel)); + var held = conn.TryResolveServerEndPoint(ReplicaB); + sentinel.ReplicaAddressesOverride = empty ? [] : [NewNodeC]; + await conn.sentinelConnection!.SwitchPrimaryAsync(null, conn); + Assert.Contains(ReplicaB, conn.GetEndPoints()); + Assert.Same(held, conn.TryResolveServerEndPoint(ReplicaB)); + Assert.True(conn.GetServer(ReplicaB).IsConnected); + Assert.Null(conn.sentinelPrimaryReconnectTimer); + } + + [Fact] + public async Task ReplicasComeFromTheSentinelWhosePrimaryWasAccepted() + { + using var data = new InProcessTestServer(log, PrimaryA); + data.AddReplicaNode(ReplicaB, PrimaryA); + using var accepted = new InProcessSentinelServer(data, ServiceName, PrimaryA, SentinelEndPoint, log); + using var other = new InProcessSentinelServer(data, ServiceName, PrimaryA, SecondSentinelEndPoint, log) + { + PrimaryError = "ERR No such master with that name", + ReplicaAddressesOverride = [], + }; + var config = CreateConfig(data, accepted, other); + config.EndPoints.Add(SecondSentinelEndPoint); + await using var conn = await ConnectionMultiplexer.ConnectAsync(config); + Assert.Contains(ReplicaB, conn.GetEndPoints()); + Assert.True(accepted.ReplicaQueries > 0); + Assert.Equal(0, other.ReplicaQueries); + Assert.Equal(0, other.LegacyQueries); + } + + [Theory] + [InlineData("LOADING Redis is loading the dataset in memory")] + [InlineData("ERR No such master with that name")] + public async Task SentinelErrorsDoNotUseTheFlaglessFallback(string error) + { + using var data = new InProcessTestServer(log, PrimaryA); + data.AddReplicaNode(ReplicaB, PrimaryA); + FailOverToNewNode(data); + using var invalid = new InProcessSentinelServer(data, ServiceName, PrimaryA, SentinelEndPoint, log) { PrimaryError = error }; + using var current = new InProcessSentinelServer(data, ServiceName, NewNodeC, SecondSentinelEndPoint, log); + var config = CreateConfig(data, invalid, current); + config.EndPoints.Add(SecondSentinelEndPoint); + await using var conn = await ConnectionMultiplexer.ConnectAsync(config); + Assert.Equal([NewNodeC], ConnectedPrimaries(conn)); + Assert.Equal(0, invalid.LegacyQueries); + } + + [Fact] + public async Task DisconnectedHeldReplicaDoesNotKeepPrimaryRecoveryAlive() + { + using var data = new InProcessTestServer(log, PrimaryA); + data.AddReplicaNode(ReplicaB, PrimaryA); + using var sentinel = new InProcessSentinelServer(data, ServiceName, PrimaryA, SentinelEndPoint, log); + await using var conn = await ConnectionMultiplexer.ConnectAsync(CreateConfig(data, sentinel)); + var held = conn.TryResolveServerEndPoint(ReplicaB); + sentinel.ReplicaFlags[ReplicaB] = "s_down,disconnected"; + RemoveAndDisconnect(data, ReplicaB); + await UntilAsync(() => !conn.GetServer(ReplicaB).IsConnected, "replica did not disconnect"); + var owner = conn.sentinelConnection!; + owner.OnManagedConnectionFailed(conn, Failure(ReplicaB, conn)); + await UntilAsync(() => conn.sentinelPrimaryReconnectTimer is null, "dead held replica kept the timer"); + Assert.Contains(ReplicaB, conn.GetEndPoints()); + Assert.Same(held, conn.TryResolveServerEndPoint(ReplicaB)); + Assert.False(conn.GetServer(ReplicaB).IsConnected); + int queries = sentinel.PrimaryQueries; + await Task.Delay(2500); + Assert.Equal(queries, sentinel.PrimaryQueries); + Assert.Null(conn.sentinelPrimaryReconnectTimer); + await conn.GetDatabase().StringSetAsync("primary-still-works", "yes"); + } + + [Fact] + public async Task SentinelAnnouncementUsesTheServiceAfterTheSeparator() + { + using var data = new InProcessTestServer(log, PrimaryA); + using var sentinel = new InProcessSentinelServer(data, ServiceName, PrimaryA, SentinelEndPoint, log); + await using var conn = await ConnectionMultiplexer.ConnectAsync(CreateConfig(data, sentinel)); + sentinel.AddEmptyNode(SecondSentinelEndPoint); + sentinel.PeerFlags[SecondSentinelEndPoint] = ""; + Assert.True(sentinel.ForAllClients(client => client.Publish(RedisChannel.Literal("+sentinel"), + "sentinel peer 192.0.2.11 26379 @ mymaster 192.0.2.1 6379")) > 0); + var owner = conn.sentinelConnection!; + await UntilAsync(() => owner.GetEndPoints().Contains(SecondSentinelEndPoint), "announcement did not discover the peer"); + } + + [Fact] + public async Task SentinelAddressRequestDuringAnOlderReplyIsReplayed() + { + using var data = new InProcessTestServer(log, PrimaryA); + using var sentinel = new InProcessSentinelServer(data, ServiceName, PrimaryA, SentinelEndPoint, log); + await using var conn = await ConnectionMultiplexer.ConnectAsync(CreateConfig(data, sentinel)); + var owner = conn.sentinelConnection!; + var captured = NewSignal(); + var release = NewSignal(); + int replies = 0; + sentinel.BeforePeerReplyAsync = () => + { + if (Interlocked.Increment(ref replies) != 1) return Task.CompletedTask; + captured.TrySetResult(true); + return release.Task; + }; + var update = owner.UpdateSentinelAddressListAsync(ServiceName); + try + { + await AwaitSignalAsync(captured.Task); + sentinel.AddEmptyNode(SecondSentinelEndPoint); + sentinel.PeerFlags[SecondSentinelEndPoint] = ""; + await owner.UpdateSentinelAddressListAsync(ServiceName); + } + finally { release.TrySetResult(true); } + await update; + await UntilAsync(() => owner.GetEndPoints().Contains(SecondSentinelEndPoint), "pending peer refresh was lost"); + Assert.True(Volatile.Read(ref replies) >= 2); + } + + [Fact] + public async Task SentinelAddressIsRetainedWhileReconfigureIsBusy() + { + using var data = new InProcessTestServer(log, PrimaryA); + using var sentinel = new InProcessSentinelServer(data, ServiceName, PrimaryA, SentinelEndPoint, log); + await using var conn = await ConnectionMultiplexer.ConnectAsync(CreateConfig(data, sentinel)); + var owner = conn.sentinelConnection!; + sentinel.AddEmptyNode(SecondSentinelEndPoint); + sentinel.PeerFlags[SecondSentinelEndPoint] = ""; + await UntilAsync(() => owner.TryClaimReconfigure("unit holds the claim"), "could not claim reconfigure"); + Task update; + try + { + update = owner.UpdateSentinelAddressListAsync(ServiceName); + await UntilAsync(() => owner.EndPoints.Contains(SecondSentinelEndPoint), "busy claim lost the address"); + } + finally { owner.ReleaseReconfigureClaim(); } + await update; + await UntilAsync(() => owner.GetServer(SecondSentinelEndPoint).IsConnected, "pending reconfigure did not connect the peer"); + } + + [Fact] + public async Task MissingServiceNameDoesNotRearmAnEmptyTimer() + { + 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); + await UntilAsync(() => data.ClientCount == 2, "losing managed connection was not disposed"); + } + + [Fact] + public async Task SynchronousSoftConnectReturnsBeforeAsyncRoleTimeoutAndRecovers() + { + using var data = new InProcessTestServer(log, PrimaryA); + using var sentinel = new InProcessSentinelServer(data, ServiceName, PrimaryA, SentinelEndPoint, log); + var release = NewSignal(); + var config = CreateConfig(data, sentinel); + config.AbortOnConnectFail = false; + config.ConnectTimeout = 200; + config.ConnectRetry = 1; + config.SyncTimeout = 500; + config.AsyncTimeout = 5000; + config.Tunnel = new BlockedDataTunnel(config.Tunnel!, PrimaryA, release.Task); + var connecting = Task.Factory.StartNew(() => ConnectionMultiplexer.Connect(config), CancellationToken.None, + TaskCreationOptions.LongRunning, TaskScheduler.Default); + ConnectionMultiplexer conn; + try + { + Assert.Same(connecting, await Task.WhenAny(connecting, Task.Delay(2500))); + conn = await connecting; + Assert.False(conn.IsConnected); + } + finally { release.TrySetResult(true); } + await using (conn) + { + await AssertPrimaryBecomesAsync(conn, PrimaryA); + await UntilAsync(() => conn.sentinelPrimaryReconnectTimer is null, "soft connect recovery did not settle"); + } + } + + private sealed class BlockedDataTunnel(Tunnel inner, EndPoint endpoint, Task release, Action? connecting = null) : Tunnel + { + public override ValueTask GetSocketConnectEndpointAsync(EndPoint target, CancellationToken cancellationToken) + => inner.GetSocketConnectEndpointAsync(target, cancellationToken); + + public override ValueTask BeforeAuthenticateAsync(EndPoint target, ConnectionType connectionType, Socket? socket, CancellationToken cancellationToken) + => Equals(target, endpoint) ? new(AcceptAsync(target, connectionType, socket, cancellationToken)) + : inner.BeforeAuthenticateAsync(target, connectionType, socket, cancellationToken); + + private async Task AcceptAsync(EndPoint target, ConnectionType connectionType, Socket? socket, CancellationToken cancellationToken) + { + if (connectionType == ConnectionType.Interactive) connecting?.Invoke(); + await release; + return await inner.BeforeAuthenticateAsync(target, connectionType, socket, cancellationToken); + } + } + + private static void RemoveAndDisconnect(InProcessTestServer server, EndPoint endpoint) + { + Assert.True(server.TryGetNode(endpoint, out var node)); + node.UpdateSlots([]); + Assert.True(server.RemoveNode(endpoint)); + server.ForAllClients(client => + { + if (ReferenceEquals(client.Node, node)) client.Dispose(); + }); + } + + private static TaskCompletionSource NewSignal() => new(TaskCreationOptions.RunContinuationsAsynchronously); + + private static async Task AwaitSignalAsync(Task signal) + => Assert.Same(signal, await Task.WhenAny(signal, Task.Delay(10000))); + + private static ConnectionFailedEventArgs Failure(EndPoint endpoint, ConnectionMultiplexer conn) + => new(null, conn, endpoint, ConnectionType.Interactive, ConnectionFailureType.SocketFailure, null, null); + + private static ConfigurationOptions CreateConfig(InProcessTestServer data, InProcessSentinelServer sentinel, params InProcessTestServer[] more) { var config = data.GetClientConfig(); config.EndPoints.Clear(); config.EndPoints.Add(SentinelEndPoint); config.ServiceName = ServiceName; - config.Tunnel = InProcessSentinelServer.CreateTunnel(data, sentinel); + config.Tunnel = InProcessSentinelServer.CreateTunnel([data, sentinel, .. more]); config.ConfigCheckSeconds = 1; // how soon the demoted node's role is re-read return config; } diff --git a/tests/StackExchange.Redis.Tests/SentinelTests.cs b/tests/StackExchange.Redis.Tests/SentinelTests.cs index ffe6d3705..22e544c39 100644 --- a/tests/StackExchange.Redis.Tests/SentinelTests.cs +++ b/tests/StackExchange.Redis.Tests/SentinelTests.cs @@ -499,30 +499,65 @@ public async Task SwitchPrimaryRebuildRetiresEndpointsSentinelNoLongerNames() { SkipOnWindowsRelease(); - // A managed connection pointed somewhere sentinel does not name: the primary it reports is an - // endpoint this connection has never seen, which is what drives SwitchPrimary's rebuild branch - // (the real-world trigger is a node that came back on a brand-new address). - await using var managed = await ConnectionMultiplexer.ConnectAsync( - $"{TestConfig.Current.PrimaryServer}:{TestConfig.Current.PrimaryPort},allowAdmin=true", - Writer); - managed.RawConfig.ServiceName = ServiceName; - - var strayEndPoint = managed.GetEndPoints().Single(); - Assert.Contains(strayEndPoint, managed.GetServerSnapshot().ToArray().Select(x => x.EndPoint)); - - Conn.SwitchPrimary(null, managed, Writer); - await UntilConditionAsync( - TimeSpan.FromSeconds(10), - () => !managed.GetServerSnapshot().ToArray().Any(x => Equals(x.EndPoint, strayEndPoint))); - - // Retired, not merely dropped from the endpoint list: the snapshot is what server selection reads - // and what GetEndPoints reports, so a server left there stays connected and selectable. - Assert.DoesNotContain(strayEndPoint, managed.GetEndPoints()); - Assert.DoesNotContain(strayEndPoint, managed.GetServerSnapshot().ToArray().Select(x => x.EndPoint)); + // An actually disconnected stray address may be retired from a nonempty, consistent sentinel view. + // A connected node must survive an incomplete view (covered by the in-process unit tests). + var port = TcpForwarder.GetFreePort(); + TcpForwarder? forwarder = new(port, TestConfig.Current.PrimaryPort); + try + { + await using var managed = await ConnectionMultiplexer.ConnectAsync( + $"127.0.0.1:{port},allowAdmin=true", + Writer); + managed.RawConfig.ServiceName = ServiceName; + + var strayEndPoint = managed.GetEndPoints().Single(); + var stray = managed.GetServerSnapshot().ToArray().Single(x => Equals(x.EndPoint, strayEndPoint)); + forwarder.Dispose(); + forwarder = null; + await UntilConditionAsync(TimeSpan.FromSeconds(10), () => !stray.IsConnected); + Assert.False(stray.IsConnected); + + await Conn.SwitchPrimaryAsync(null, managed, Writer); + await UntilConditionAsync( + TimeSpan.FromSeconds(10), + () => !managed.GetServerSnapshot().ToArray().Any(x => Equals(x.EndPoint, strayEndPoint))); + + // Retirement removes the endpoint from both configuration and the server-selection snapshot. + Assert.DoesNotContain(strayEndPoint, managed.GetEndPoints()); + Assert.DoesNotContain(strayEndPoint, managed.GetServerSnapshot().ToArray().Select(x => x.EndPoint)); + Assert.NotEmpty(managed.GetEndPoints()); + await managed.GetDatabase().PingAsync(); + } + finally + { + forwarder?.Dispose(); + } + } - // ...and the rebuild still did its actual job - Assert.NotEmpty(managed.GetEndPoints()); - 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]