diff --git a/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs b/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs index 9e82d447f..64917a872 100644 --- a/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs +++ b/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs @@ -150,12 +150,19 @@ 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; - var muxer = sentinelConnection.GetSentinelMasterConnection(configuration, log); - // Set reference to sentinel connection so that we can dispose it - muxer.sentinelConnection = sentinelConnection; - - return muxer; + return muxer; + } + catch + { + try { sentinelConnection.Dispose(); } catch { } + throw; + } } /// @@ -172,12 +179,19 @@ 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; - var muxer = sentinelConnection.GetSentinelMasterConnection(configuration, writer); - // Set reference to sentinel connection so that we can dispose it - muxer.sentinelConnection = sentinelConnection; - - return muxer; + return muxer; + } + catch + { + try { sentinelConnection.Dispose(); } catch { } + throw; + } } /// @@ -187,7 +201,8 @@ private static async Task SentinelPrimaryConnectAsync(Con /// The writer to log to, if any. public ConnectionMultiplexer GetSentinelMasterConnection(ConfigurationOptions config, TextWriter? log = null) { - if (ServerSelectionStrategy.ServerType != ServerType.Sentinel) + // A soft-failed SentinelConnect cannot detect a server type, but _isSentinel still records its intent. + if (ServerSelectionStrategy.ServerType != ServerType.Sentinel && (config.AbortOnConnectFail || !_isSentinel)) { throw new RedisConnectionException( ConnectionFailureType.UnableToConnect, @@ -210,110 +225,162 @@ public ConnectionMultiplexer GetSentinelMasterConnection(ConfigurationOptions co bool success = false; ConnectionMultiplexer? connection = null; EndPointCollection? endpoints = null; + RedisConnectionException? failure = null; - var sw = ValueStopwatch.StartNew(); - do + try { - // 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++) + var sw = ValueStopwatch.StartNew(); + do { - newPrimaryEndPoint = GetConfiguredPrimaryForService(serviceName); - } + // Sentinel has some fun race behavior internally - give things a few shots for a quicker overall connect. + const int queryAttempts = 2; - if (newPrimaryEndPoint is null) - { - throw new RedisConnectionException( - ConnectionFailureType.UnableToConnect, - CommandFlags.None, - $"Sentinel: Failed connecting to configured primary for service: {config.ServiceName}"); - } + EndPoint? newPrimaryEndPoint = null; + for (int i = 0; i < queryAttempts && newPrimaryEndPoint is null; i++) + { + newPrimaryEndPoint = GetConfiguredPrimaryForService(serviceName); + } - EndPoint[]? replicaEndPoints = null; - for (int i = 0; i < queryAttempts && replicaEndPoints is null; i++) - { - replicaEndPoints = GetReplicasForService(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; + } - endpoints = config.EndPoints.Clone(); + if (connection is not null) try { connection.Dispose(); } catch { } + endpoints = config.EndPoints.Clone(); + connection = ConnectImpl(config, log, endpoints: endpoints); + break; + } - // 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); - } + EndPoint[]? replicaEndPoints = null; + for (int i = 0; i < queryAttempts && replicaEndPoints is null; i++) + { + replicaEndPoints = GetReplicasForService(serviceName); + } - if (replicaEndPoints is not null) - { - foreach (var replicaEndPoint in replicaEndPoints) + 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.TryAdd(replicaEndPoint); + endpoints[0] = newPrimaryEndPoint; + } + else + { + endpoints.Clear(); + endpoints.TryAdd(newPrimaryEndPoint); } - } - connection = ConnectImpl(config, log, endpoints: endpoints); + if (replicaEndPoints is not null) + { + foreach (var replicaEndPoint in replicaEndPoints) + { + endpoints.TryAdd(replicaEndPoint); + } + } - // 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 + if (connection is not null) try { connection.Dispose(); } catch { } + 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 { }) { - isPrimary = connection.CommandMap.IsAvailable(RedisCommand.ROLE) - ? server.Role()?.Value == Role.LabelForMaster - : !server.IsReplica; + 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; + } } - catch + + 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) { - // fallback if ROLE unavailable but not declared; see #3064 - isPrimary = !server.IsReplica; + throw failure; } - if (isPrimary) + if (connection is null) { - success = true; - break; + endpoints = config.EndPoints.Clone(); + connection = ConnectImpl(config, log, endpoints: endpoints); } + + connection.LastException = failure; + connection.Logger?.LogErrorSyncConnectTimeout(failure, failure.Message); } - Thread.Sleep(100); - } - while (sw.ElapsedMilliseconds < config.ConnectTimeout); + // Attach to reconnect event to ensure proper connection to the new primary + connection.ConnectionRestored += OnManagedConnectionRestored; - if (!success) - { - throw new RedisConnectionException( - ConnectionFailureType.UnableToConnect, - CommandFlags.None, - $"Sentinel: Failed connecting to configured primary for service: {config.ServiceName}"); - } + // If we lost the connection, run a switch to a least try and get updated info about the primary + connection.ConnectionFailed += OnManagedConnectionFailed; - // Attach to reconnect event to ensure proper connection to the new primary - connection.ConnectionRestored += OnManagedConnectionRestored; + lock (sentinelConnectionChildren) + { + sentinelConnectionChildren[serviceName] = connection; + } - // If we lost the connection, run a switch to a least try and get updated info about the primary - connection.ConnectionFailed += OnManagedConnectionFailed; + // Perform the initial switchover + var switchBlame = endpoints is { Count: > 0 } ? endpoints[0] : null; + try + { + SwitchPrimary(switchBlame, connection, log); + } + catch when (!success && !config.AbortOnConnectFail) + { + ScheduleSentinelPrimaryReconnect(connection, switchBlame); + } - lock (sentinelConnectionChildren) + return connection; + } + catch { - sentinelConnectionChildren[serviceName] = connection; + 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; } - - // 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.")] @@ -360,7 +427,6 @@ 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) @@ -368,6 +434,12 @@ 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. @@ -379,7 +451,7 @@ internal void OnManagedConnectionFailed(object? sender, ConnectionFailedEventArg try { // Attempt, but do not fail here - SwitchPrimary(e.EndPoint, connection); + SwitchPrimary(endPoint, connection); } catch (Exception) { diff --git a/src/StackExchange.Redis/ServerEndPoint.cs b/src/StackExchange.Redis/ServerEndPoint.cs index 20968261b..278d98193 100644 --- a/src/StackExchange.Redis/ServerEndPoint.cs +++ b/src/StackExchange.Redis/ServerEndPoint.cs @@ -355,7 +355,7 @@ public void SetClusterConfiguration(ClusterConfiguration configuration) public void UpdateNodeRelations(ClusterConfiguration configuration) { - var thisNode = configuration.Nodes.FirstOrDefault(x => x.EndPoint?.Equals(EndPoint) == true); + var thisNode = GetClusterNode(configuration); if (thisNode != null) { Multiplexer.Trace($"Updating node relations for {Format.ToString(thisNode.EndPoint)}..."); @@ -379,6 +379,9 @@ public void UpdateNodeRelations(ClusterConfiguration configuration) } } + private ClusterNode? GetClusterNode(ClusterConfiguration? configuration) => + configuration?.Nodes.FirstOrDefault(x => x.EndPoint?.Equals(EndPoint) == true); + public void SetUnselectable(UnselectableFlags flags) { if (flags != 0) @@ -502,7 +505,9 @@ internal async Task AutoConfigureAsync(PhysicalConnection? connection, ILogger? await WriteDirectOrQueueFireAndForgetAsync(connection, msg, autoConfigProcessor).ForAwait(); } } - else if (commandMap.IsAvailable(RedisCommand.SET) && !(helloPending || RoleKnownFromHello)) + else if (commandMap.IsAvailable(RedisCommand.SET) + && !(helloPending || RoleKnownFromHello) + && !(ServerType == ServerType.Cluster && GetClusterNode(ClusterConfiguration) is not null)) { // This is a nasty way to find if we are a replica, and it will only work on up-level servers, but... // (note we only get here when HELLO isn't going to tell us: the HELLO reply carries "role", and @@ -535,7 +540,9 @@ internal async Task AutoConfigureAsync(PhysicalConnection? connection, ILogger? } // If we are going to fetch a tie breaker, do so last and we'll get it in before the tracer fires completing the connection // But if GETs are disabled on this, do not fail the connection - we just don't get tiebreaker benefits - if (Multiplexer.RawConfig.TryGetTieBreaker(out var tieBreakerKey) && Multiplexer.CommandMap.IsAvailable(RedisCommand.GET)) + if (ServerType != ServerType.Cluster + && Multiplexer.RawConfig.TryGetTieBreaker(out var tieBreakerKey) + && Multiplexer.CommandMap.IsAvailable(RedisCommand.GET)) { log?.LogInformationRequestingTieBreak(new(EndPoint), tieBreakerKey); msg = Message.Create(0, flags, RedisCommand.GET, tieBreakerKey); @@ -680,12 +687,24 @@ internal Message GetTracerMessage(bool checkResponse) else { map.AssertAvailable(RedisCommand.EXISTS); - msg = Message.Create(0, flags, RedisCommand.EXISTS, (RedisValue)Multiplexer.UniqueId); + msg = Message.Create(0, flags, RedisCommand.EXISTS, GetTracerKey()); } msg.SetInternalCall(); return msg; } + internal RedisKey GetTracerKey() + { + RedisKey key = Multiplexer.UniqueId; + if (ServerType == ServerType.Cluster + && GetClusterNode(ClusterConfiguration) is { } node + && node.Slots.Count > 0) + { + key = key.Prepend(ServerSelectionStrategy.GetHashTagPrefix(node.Slots[0].From)); + } + return key; + } + internal UnselectableFlags GetUnselectableFlags() => unselectableReasons; internal bool IsSelectable(RedisCommand command, bool allowDisconnected = false) diff --git a/src/StackExchange.Redis/ServerSelectionStrategy.HashTags.cs b/src/StackExchange.Redis/ServerSelectionStrategy.HashTags.cs index 4cbd4538c..a1751f50a 100644 --- a/src/StackExchange.Redis/ServerSelectionStrategy.HashTags.cs +++ b/src/StackExchange.Redis/ServerSelectionStrategy.HashTags.cs @@ -1,6 +1,7 @@ -using System; +using System; using System.Diagnostics; using System.Text; +using System.Threading; namespace StackExchange.Redis; @@ -10,9 +11,25 @@ internal sealed partial class ServerSelectionStrategy private static class HashTags { private static readonly string[] Cache = Populate(); + private static readonly byte[]?[] PrefixCache = new byte[TotalSlots][]; + private static readonly object PrefixCacheLock = new(); + public static ReadOnlySpan Tags => Cache; public static string Get(int slot) => Cache[slot]; + public static byte[] GetPrefix(int slot) + { + var prefix = Volatile.Read(ref PrefixCache[slot]); + if (prefix is null) + { + lock (PrefixCacheLock) + { + prefix = PrefixCache[slot] ??= Encoding.ASCII.GetBytes("{" + Get(slot) + "}"); + } + } + return prefix; + } + private static string[] Populate() { // Via testing, we know that 3 characters is sufficient to populate all slots diff --git a/src/StackExchange.Redis/ServerSelectionStrategy.cs b/src/StackExchange.Redis/ServerSelectionStrategy.cs index 80c9b9efe..8c0623e67 100644 --- a/src/StackExchange.Redis/ServerSelectionStrategy.cs +++ b/src/StackExchange.Redis/ServerSelectionStrategy.cs @@ -442,5 +442,7 @@ internal string GetHashTag(ServerEndPoint endpoint) /// Gets a string that can be used as a hash-tag to reference a specific slot. /// internal static string GetHashTag(int slot) => slot < 0 ? "" : HashTags.Get(slot); + + internal static RedisKey GetHashTagPrefix(int slot) => HashTags.GetPrefix(slot); } } diff --git a/tests/StackExchange.Redis.Tests/HashTagUnitTests.cs b/tests/StackExchange.Redis.Tests/HashTagUnitTests.cs index 91b63d980..6709da826 100644 --- a/tests/StackExchange.Redis.Tests/HashTagUnitTests.cs +++ b/tests/StackExchange.Redis.Tests/HashTagUnitTests.cs @@ -1,4 +1,4 @@ -using System; +using System; using System.Collections.Generic; using System.Text; using Xunit; @@ -26,4 +26,18 @@ public void TestHashTagCoverage() } Assert.Equal(ServerSelectionStrategy.TotalSlots, uniques.Count); } + + [Theory] + [InlineData(0)] + [InlineData(1)] + [InlineData(8191)] + [InlineData(16383)] + public void TestHashTagPrefixTargetsSlot(int slot) + { + var prefix = ServerSelectionStrategy.GetHashTagPrefix(slot); + RedisKey key = ((RedisKey)"probe-id").Prepend(prefix); + + Assert.Equal(slot, ServerSelectionStrategy.GetHashSlot(key)); + Assert.Same((byte[]?)prefix, (byte[]?)ServerSelectionStrategy.GetHashTagPrefix(slot)); + } } diff --git a/tests/StackExchange.Redis.Tests/SentinelConfigTests.cs b/tests/StackExchange.Redis.Tests/SentinelConfigTests.cs index 2e659eea6..814f99719 100644 --- a/tests/StackExchange.Redis.Tests/SentinelConfigTests.cs +++ b/tests/StackExchange.Redis.Tests/SentinelConfigTests.cs @@ -1,4 +1,6 @@ using System; +using System.Net; +using System.Threading.Tasks; using Xunit; namespace StackExchange.Redis.Tests; @@ -44,4 +46,63 @@ 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/ServerEndPointClusterProbeUnitTests.cs b/tests/StackExchange.Redis.Tests/ServerEndPointClusterProbeUnitTests.cs new file mode 100644 index 000000000..3ff088ceb --- /dev/null +++ b/tests/StackExchange.Redis.Tests/ServerEndPointClusterProbeUnitTests.cs @@ -0,0 +1,66 @@ +using System.Linq; +using System.Net; +using System.Threading.Tasks; +using Xunit; + +namespace StackExchange.Redis.Tests; + +public class ServerEndPointClusterProbeUnitTests +{ + [Fact] + public async Task ExistsTracerUsesOwnedClusterSlot() + { + using var server = new InProcessTestServer { ServerType = ServerType.Cluster }; + var config = server.GetClientConfig(defaultOnly: true); + var commands = server.GetCommands(); + commands.Remove(nameof(RedisCommand.ECHO)); + commands.Remove(nameof(RedisCommand.PING)); + commands.Remove(nameof(RedisCommand.TIME)); + config.CommandMap = CommandMap.Create(commands); + + await using var connection = await ConnectionMultiplexer.ConnectAsync(config); + var endpoint = connection.GetServerEndPoint(server.DefaultEndPoint); + var node = endpoint.ClusterConfiguration?.Nodes.Single(x => x.EndPoint?.Equals(endpoint.EndPoint) == true); + Assert.NotNull(node); + var targetSlot = node.Slots[0].From; + + var message = endpoint.GetTracerMessage(checkResponse: true); + + Assert.Equal(RedisCommand.EXISTS, message.Command); + Assert.Equal(targetSlot, message.GetHashSlot(connection.ServerSelectionStrategy)); + var key = endpoint.GetTracerKey(); + var keyBytes = (byte[]?)key; + var prefixBytes = (byte[]?)ServerSelectionStrategy.GetHashTagPrefix(targetSlot); + Assert.NotNull(keyBytes); + Assert.NotNull(prefixBytes); + Assert.True(keyBytes.Take(prefixBytes.Length).SequenceEqual(prefixBytes)); + Assert.True(keyBytes.Skip(keyBytes.Length - connection.UniqueId.Length).SequenceEqual(connection.UniqueId)); + } + + [Theory] + [InlineData(ServerType.Standalone)] + [InlineData(ServerType.Cluster)] + public async Task ExistsTracerUsesPlainKeyWithoutKnownOwnedSlots(ServerType serverType) + { + using var server = new InProcessTestServer(); + var config = server.GetClientConfig(defaultOnly: true); + var commands = server.GetCommands(); + commands.Remove(nameof(RedisCommand.ECHO)); + commands.Remove(nameof(RedisCommand.PING)); + commands.Remove(nameof(RedisCommand.TIME)); + config.CommandMap = CommandMap.Create(commands); + + await using var connection = await ConnectionMultiplexer.ConnectAsync(config); + var endpoint = new ServerEndPoint(connection, new IPEndPoint(IPAddress.Loopback, 12345)) + { + ServerType = serverType, + }; + + var message = endpoint.GetTracerMessage(checkResponse: true); + var clusterStrategy = new ServerSelectionStrategy(null) { ServerType = ServerType.Cluster }; + + Assert.Equal(RedisCommand.EXISTS, message.Command); + Assert.Equal(ServerSelectionStrategy.GetHashSlot((RedisKey)connection.UniqueId), message.GetHashSlot(clusterStrategy)); + Assert.Equal(connection.UniqueId, (byte[]?)endpoint.GetTracerKey()); + } +}