diff --git a/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs b/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs index 40d513a3f..ad7d73abb 100644 --- a/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs +++ b/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs @@ -150,19 +150,12 @@ private static ConnectionMultiplexer SentinelPrimaryConnect(ConfigurationOptions sentinelConfig.Password = configuration.SentinelPassword; var sentinelConnection = SentinelConnect(sentinelConfig, log); - try - { - var muxer = sentinelConnection.GetSentinelMasterConnection(configuration, log); - // Set reference to sentinel connection so that we can dispose it - muxer.sentinelConnection = sentinelConnection; - return muxer; - } - catch - { - try { sentinelConnection.Dispose(); } catch { } - throw; - } + var muxer = sentinelConnection.GetSentinelMasterConnection(configuration, log); + // Set reference to sentinel connection so that we can dispose it + muxer.sentinelConnection = sentinelConnection; + + return muxer; } /// @@ -179,19 +172,12 @@ private static async Task SentinelPrimaryConnectAsync(Con sentinelConfig.Password = configuration.SentinelPassword; var sentinelConnection = await SentinelConnectAsync(sentinelConfig, writer).ForAwait(); - try - { - var muxer = sentinelConnection.GetSentinelMasterConnection(configuration, writer); - // Set reference to sentinel connection so that we can dispose it - muxer.sentinelConnection = sentinelConnection; - return muxer; - } - catch - { - try { sentinelConnection.Dispose(); } catch { } - throw; - } + var muxer = sentinelConnection.GetSentinelMasterConnection(configuration, writer); + // Set reference to sentinel connection so that we can dispose it + muxer.sentinelConnection = sentinelConnection; + + return muxer; } /// @@ -201,8 +187,7 @@ private static async Task SentinelPrimaryConnectAsync(Con /// The writer to log to, if any. public ConnectionMultiplexer GetSentinelMasterConnection(ConfigurationOptions config, TextWriter? log = null) { - // A soft-failed SentinelConnect cannot detect a server type, but _isSentinel still records its intent. - if (ServerSelectionStrategy.ServerType != ServerType.Sentinel && (config.AbortOnConnectFail || !_isSentinel)) + if (ServerSelectionStrategy.ServerType != ServerType.Sentinel) { throw new RedisConnectionException( ConnectionFailureType.UnableToConnect, @@ -225,166 +210,110 @@ public ConnectionMultiplexer GetSentinelMasterConnection(ConfigurationOptions co bool success = false; ConnectionMultiplexer? connection = null; EndPointCollection? endpoints = null; - RedisConnectionException? failure = null; - try + var sw = ValueStopwatch.StartNew(); + do { - var sw = ValueStopwatch.StartNew(); - do - { - // Sentinel has some fun race behavior internally - give things a few shots for a quicker overall connect. - const int queryAttempts = 2; + // 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++) - { - newPrimaryEndPoint = GetConfiguredPrimaryForService(serviceName); - } - - if (newPrimaryEndPoint is null) - { - failure = new RedisConnectionException( - ConnectionFailureType.UnableToConnect, - CommandFlags.None, - $"Sentinel: Failed connecting to configured primary for service: {config.ServiceName}"); - if (config.AbortOnConnectFail) - { - throw failure; - } + EndPoint? newPrimaryEndPoint = null; + for (int i = 0; i < queryAttempts && newPrimaryEndPoint is null; i++) + { + newPrimaryEndPoint = GetConfiguredPrimaryForService(serviceName); + } - if (connection is not null) try { connection.Dispose(); } catch { } - endpoints = config.EndPoints.Clone(); - connection = ConnectImpl(config, log, endpoints: endpoints); - break; - } + if (newPrimaryEndPoint is null) + { + throw new RedisConnectionException( + ConnectionFailureType.UnableToConnect, + CommandFlags.None, + $"Sentinel: Failed connecting to configured primary for service: {config.ServiceName}"); + } - EndPoint[]? replicaEndPoints = null; - for (int i = 0; i < queryAttempts && replicaEndPoints is null; i++) - { - replicaEndPoints = GetReplicasForService(serviceName); - } + EndPoint[]? replicaEndPoints = null; + for (int i = 0; i < queryAttempts && replicaEndPoints is null; i++) + { + replicaEndPoints = GetReplicasForService(serviceName); + } - endpoints = config.EndPoints.Clone(); + endpoints = config.EndPoints.Clone(); - // Replace the primary endpoint, if we found another one - // If not, assume the last state is the best we have and minimize the race - if (endpoints.Count == 1) - { - endpoints[0] = newPrimaryEndPoint; - } - else - { - endpoints.Clear(); - endpoints.TryAdd(newPrimaryEndPoint); - } + // Replace the primary endpoint, if we found another one + // If not, assume the last state is the best we have and minimize the race + if (endpoints.Count == 1) + { + endpoints[0] = newPrimaryEndPoint; + } + else + { + endpoints.Clear(); + endpoints.TryAdd(newPrimaryEndPoint); + } - if (replicaEndPoints is not null) + if (replicaEndPoints is not null) + { + foreach (var replicaEndPoint in replicaEndPoints) { - foreach (var replicaEndPoint in replicaEndPoints) - { - endpoints.TryAdd(replicaEndPoint); - } + endpoints.TryAdd(replicaEndPoint); } + } - if (connection is not null) try { connection.Dispose(); } catch { } - connection = ConnectImpl(config, log, endpoints: endpoints); + connection = ConnectImpl(config, log, endpoints: endpoints); - // verify role is primary according to: - // https://redis.io/topics/sentinel-clients - bool isPrimary; - var server = connection.GetServer(newPrimaryEndPoint); - // ReSharper disable once ConditionIsAlwaysTrueOrFalseAccordingToNullableAPIContract - if (server is { }) + // verify role is primary according to: + // https://redis.io/topics/sentinel-clients + bool isPrimary; + var server = connection.GetServer(newPrimaryEndPoint); + // ReSharper disable once ConditionIsAlwaysTrueOrFalseAccordingToNullableAPIContract + if (server is { }) + { + try { - try - { - isPrimary = connection.CommandMap.IsAvailable(RedisCommand.ROLE) - ? server.Role()?.Value == Role.LabelForMaster - : !server.IsReplica; - } - catch - { - // fallback if ROLE unavailable but not declared; see #3064 - isPrimary = !server.IsReplica; - } - - if (isPrimary) - { - success = true; - break; - } + isPrimary = connection.CommandMap.IsAvailable(RedisCommand.ROLE) + ? server.Role()?.Value == Role.LabelForMaster + : !server.IsReplica; } - - Thread.Sleep(100); - } - while (sw.ElapsedMilliseconds < config.ConnectTimeout); - - if (!success) - { - failure ??= new RedisConnectionException( - ConnectionFailureType.UnableToConnect, - CommandFlags.None, - $"Sentinel: Failed connecting to configured primary for service: {config.ServiceName}"); - if (config.AbortOnConnectFail) + catch { - throw failure; + // fallback if ROLE unavailable but not declared; see #3064 + isPrimary = !server.IsReplica; } - if (connection is null) + if (isPrimary) { - endpoints = config.EndPoints.Clone(); - connection = ConnectImpl(config, log, endpoints: endpoints); + success = true; + break; } - - connection.LastException = failure; - connection.Logger?.LogErrorSyncConnectTimeout(failure, failure.Message); } - // Attach to reconnect event to ensure proper connection to the new primary - connection.ConnectionRestored += OnManagedConnectionRestored; + Thread.Sleep(100); + } + while (sw.ElapsedMilliseconds < config.ConnectTimeout); - // If we lost the connection, run a switch to a least try and get updated info about the primary - connection.ConnectionFailed += OnManagedConnectionFailed; + if (!success) + { + throw new RedisConnectionException( + ConnectionFailureType.UnableToConnect, + CommandFlags.None, + $"Sentinel: Failed connecting to configured primary for service: {config.ServiceName}"); + } - lock (sentinelConnectionChildren) - { - sentinelConnectionChildren[serviceName] = connection; - } + // Attach to reconnect event to ensure proper connection to the new primary + connection.ConnectionRestored += OnManagedConnectionRestored; - // Perform the initial switchover - var switchBlame = endpoints is { Count: > 0 } ? endpoints[0] : null; - try - { - SwitchPrimary(switchBlame, connection, log); - } - catch when (!config.AbortOnConnectFail) - { - // AbortOnConnectFail=false means "hand me the multiplexer and keep trying" - which has to - // 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); - } + // If we lost the connection, run a switch to a least try and get updated info about the primary + connection.ConnectionFailed += OnManagedConnectionFailed; - return connection; - } - catch + lock (sentinelConnectionChildren) { - if (connection is not null) - { - connection.ConnectionRestored -= OnManagedConnectionRestored; - connection.ConnectionFailed -= OnManagedConnectionFailed; - lock (sentinelConnectionChildren) - { - if (sentinelConnectionChildren.TryGetValue(serviceName, out var child) && ReferenceEquals(child, connection)) - { - sentinelConnectionChildren.Remove(serviceName); - } - } - try { connection.Dispose(); } catch { } - } - throw; + sentinelConnectionChildren[serviceName] = connection; } + + // Perform the initial switchover + SwitchPrimary(endpoints[0], connection, log); + + return connection; } [System.Diagnostics.CodeAnalysis.SuppressMessage("Roslynator", "RCS1075:Avoid empty catch clause that catches System.Exception.", Justification = "We don't care.")] @@ -431,6 +360,7 @@ internal void OnManagedConnectionRestored(object? sender, ConnectionFailedEventA } } + [System.Diagnostics.CodeAnalysis.SuppressMessage("Roslynator", "RCS1075:Avoid empty catch clause that catches System.Exception.", Justification = "We don't care.")] internal void OnManagedConnectionFailed(object? sender, ConnectionFailedEventArgs e) { if (sender is not ConnectionMultiplexer connection) @@ -438,59 +368,38 @@ internal void OnManagedConnectionFailed(object? sender, ConnectionFailedEventArg return; // This should never happen - called from non-nullable ConnectionFailedEventArgs } - ScheduleSentinelPrimaryReconnect(connection, e.EndPoint); - } - - [System.Diagnostics.CodeAnalysis.SuppressMessage("Roslynator", "RCS1075:Avoid empty catch clause that catches System.Exception.", Justification = "We don't care.")] - 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 + if (connection.sentinelPrimaryReconnectTimer == null) + { + connection.sentinelPrimaryReconnectTimer = new Timer( + _ => { try { - connection.sentinelPrimaryReconnectTimer?.Change(TimeSpan.FromSeconds(1), Timeout.InfiniteTimeSpan); + // Attempt, but do not fail here + SwitchPrimary(e.EndPoint, connection); } - catch (ObjectDisposedException) + catch (Exception) { - // 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 (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); - } - else - { - timer.Dispose(); + 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, + TimeSpan.Zero, + Timeout.InfiniteTimeSpan); } } diff --git a/tests/StackExchange.Redis.Tests/Helpers/TcpForwarder.cs b/tests/StackExchange.Redis.Tests/Helpers/TcpForwarder.cs deleted file mode 100644 index ea2b92665..000000000 --- a/tests/StackExchange.Redis.Tests/Helpers/TcpForwarder.cs +++ /dev/null @@ -1,89 +0,0 @@ -using System; -using System.Net; -using System.Net.Sockets; -using System.Threading; -using System.Threading.Tasks; - -namespace StackExchange.Redis.Tests; - -/// -/// Forwards a loopback port to another one, so a test can make a server reachable (or not) at an address of -/// its choosing without touching the shared topology. -/// -/// -/// The alternative - stopping a real server - is not available to a suite that runs against a shared set of -/// them, and "the address was dead and then it was not" is the only way to exercise a connect that has to -/// survive its dependencies not being up yet. -/// -public sealed class TcpForwarder : IDisposable -{ - private readonly TcpListener _listener; - private readonly int _targetPort; - private readonly CancellationTokenSource _cts = new(); - - /// A port with nothing listening on it, for a test that needs a dead address first. - public static int GetFreePort() - { - var probe = new TcpListener(IPAddress.Loopback, 0); - probe.Start(); - var port = ((IPEndPoint)probe.LocalEndpoint).Port; - probe.Stop(); - return port; - } - - public TcpForwarder(int listenPort, int targetPort) - { - _targetPort = targetPort; - _listener = new TcpListener(IPAddress.Loopback, listenPort); - _listener.Start(); - _ = Task.Run(AcceptLoopAsync); - } - - private async Task AcceptLoopAsync() - { - while (!_cts.IsCancellationRequested) - { - TcpClient client; - try - { - client = await _listener.AcceptTcpClientAsync().ForAwait(); - } - catch - { - return; // listener stopped - } - _ = Task.Run(() => PumpAsync(client)); - } - } - - private async Task PumpAsync(TcpClient client) - { - using (client) - { - try - { - using var upstream = new TcpClient(); - await upstream.ConnectAsync(IPAddress.Loopback, _targetPort).ForAwait(); - var near = client.GetStream(); - var far = upstream.GetStream(); - // the buffer size is spelled out because the two-argument (stream, token) overload does not - // exist on .NET Framework, which the test project still targets - const int BufferSize = 4096; - await Task.WhenAny( - near.CopyToAsync(far, BufferSize, _cts.Token), - far.CopyToAsync(near, BufferSize, _cts.Token)).ForAwait(); - } - catch - { - // a forwarded connection going away is normal; the test asserts on the client's behaviour - } - } - } - - public void Dispose() - { - _cts.Cancel(); - try { _listener.Stop(); } catch { } - _cts.Dispose(); - } -} diff --git a/tests/StackExchange.Redis.Tests/SentinelConfigTests.cs b/tests/StackExchange.Redis.Tests/SentinelConfigTests.cs index 814f99719..2e659eea6 100644 --- a/tests/StackExchange.Redis.Tests/SentinelConfigTests.cs +++ b/tests/StackExchange.Redis.Tests/SentinelConfigTests.cs @@ -1,6 +1,4 @@ using System; -using System.Net; -using System.Threading.Tasks; using Xunit; namespace StackExchange.Redis.Tests; @@ -46,63 +44,4 @@ public void Clone_Preserves_SentinelCredentials() Assert.Equal(options.SentinelUser, clone.SentinelUser); Assert.Equal(options.SentinelPassword, clone.SentinelPassword); } - - [Fact] - public void Connect_UnreachableSentinel_AbortDisabled_ReturnsDisconnectedMultiplexer() - { - var options = GetUnreachableSentinelOptions(abortOnConnectFail: false); - - using var connection = ConnectionMultiplexer.Connect(options); - - Assert.False(connection.IsConnected); - var exception = Assert.IsType(connection.LastException); - Assert.Equal(ConnectionFailureType.UnableToConnect, exception.FailureType); - Assert.Equal("Sentinel: Failed connecting to configured primary for service: unreachable-primary", exception.Message); - Assert.NotNull(connection.sentinelConnection); - } - - [Fact] - public async Task ConnectAsync_UnreachableSentinel_AbortDisabled_ReturnsDisconnectedMultiplexer() - { - var options = GetUnreachableSentinelOptions(abortOnConnectFail: false); - - await using var connection = await ConnectionMultiplexer.ConnectAsync(options); - - Assert.False(connection.IsConnected); - var exception = Assert.IsType(connection.LastException); - Assert.Equal(ConnectionFailureType.UnableToConnect, exception.FailureType); - Assert.Equal("Sentinel: Failed connecting to configured primary for service: unreachable-primary", exception.Message); - Assert.NotNull(connection.sentinelConnection); - } - - [Fact] - public void SentinelConnect_UnreachableSentinel_AbortDisabled_ReturnsDisconnectedMultiplexer() - { - var options = GetUnreachableSentinelOptions(abortOnConnectFail: false); - - using var connection = ConnectionMultiplexer.SentinelConnect(options); - - Assert.False(connection.IsConnected); - } - - [Fact] - public void Connect_UnreachableSentinel_AbortEnabled_Throws() - { - var options = GetUnreachableSentinelOptions(abortOnConnectFail: true); - - Assert.Throws(() => ConnectionMultiplexer.Connect(options)); - } - - private static ConfigurationOptions GetUnreachableSentinelOptions(bool abortOnConnectFail) - { - var options = new ConfigurationOptions - { - AbortOnConnectFail = abortOnConnectFail, - ConnectRetry = 0, - ConnectTimeout = 100, - ServiceName = "unreachable-primary", - }; - options.EndPoints.Add(IPAddress.Loopback, 1); - return options; - } } diff --git a/tests/StackExchange.Redis.Tests/SentinelTests.cs b/tests/StackExchange.Redis.Tests/SentinelTests.cs index d99e47db7..f4612c170 100644 --- a/tests/StackExchange.Redis.Tests/SentinelTests.cs +++ b/tests/StackExchange.Redis.Tests/SentinelTests.cs @@ -524,58 +524,4 @@ await UntilConditionAsync( Assert.NotEmpty(managed.GetEndPoints()); await managed.GetDatabase().PingAsync(); } - - [Fact] - public async Task AbortOnConnectFailFalseRecoversOnceSentinelsAppear() - { - SkipOnWindowsRelease(); - - // Returning a multiplexer instead of throwing is only worth anything if it goes on to connect, so - // drive the whole arc: start with nothing listening, then make a sentinel reachable at that address. - // A forwarder rather than stopping the real sentinel, so this cannot disturb the shared topology. - var port = TcpForwarder.GetFreePort(); - var options = new ConfigurationOptions - { - AbortOnConnectFail = false, - ServiceName = ServiceName, - ConnectTimeout = 2000, - AllowAdmin = true, - }; - options.EndPoints.Add(IPAddress.Loopback, port); - - await using var conn = await ConnectionMultiplexer.ConnectAsync(options, Writer); - - // the point of the fix: a multiplexer, not an exception - Assert.False(conn.IsConnected); - var failure = Assert.IsType(conn.LastException); - Assert.Equal(ConnectionFailureType.UnableToConnect, failure.FailureType); - var seedEndPoint = Assert.Single(conn.GetEndPoints()); - - using var forwarder = new TcpForwarder(port, TestConfig.Current.SentinelPortA); - Log($"forwarding {port} -> {TestConfig.Current.SentinelPortA}"); - - var db = conn.GetDatabase(); - await UntilConditionAsync(TimeSpan.FromSeconds(30), () => TryPing(db)); - Assert.True(TryPing(db), "did not recover once the sentinel became reachable"); - - // and the endpoint it was seeded with - a sentinel address, never a data node - does not linger as a - // server in the multiplexer once the real topology is known - await UntilConditionAsync( - TimeSpan.FromSeconds(10), - () => !conn.GetServerSnapshot().ToArray().Any(x => Equals(x.EndPoint, seedEndPoint))); - Assert.DoesNotContain(seedEndPoint, conn.GetEndPoints()); - - static bool TryPing(IDatabase db) - { - try - { - db.Ping(); - return true; - } - catch - { - return false; - } - } - } }