diff --git a/docs/Configuration.md b/docs/Configuration.md index 2519c0e2e..ac044e4dc 100644 --- a/docs/Configuration.md +++ b/docs/Configuration.md @@ -41,6 +41,13 @@ throws if the sentinels cannot be reached or do not know the service. With `abor it is not connected, operations fail with a connection exception that names the problem, and it connects itself to the primary once the sentinels become reachable and name one. If you want startup to fail when the sentinels are unavailable, keep `abortConnect=true`. +A sentinel-managed connection tracks failovers through several signals: the `+switch-master` announcement, a reconnect poll whenever the +connection to the primary is lost, and - because the announcement is delivered over pub/sub and can be missed - a periodic check that +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. + An overview of mapping between the `string` and `ConfigurationOptions` representation is shown below, but you can switch between them trivially: ```csharp diff --git a/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs b/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs index a93f8bd95..31f5296ce 100644 --- a/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs +++ b/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs @@ -11,11 +11,14 @@ namespace StackExchange.Redis; public partial class ConnectionMultiplexer { - internal EndPoint? currentSentinelPrimaryEndPoint; + // volatile: written by SwitchPrimary on worker/handler threads and read by the heartbeat + // (IsSentinelPrimaryDemoted / RecoverDemotedPrimary) - the demotion check must see the latest switch + internal volatile EndPoint? currentSentinelPrimaryEndPoint; internal Timer? sentinelPrimaryReconnectTimer; internal readonly Dictionary sentinelConnectionChildren = new(); internal ConnectionMultiplexer? sentinelConnection; private int _sentinelInitialized; + private int _demotedPrimaryRecoveryActive; /// /// Initializes the connection as a Sentinel connection and adds the necessary event handlers to track changes to the managed primaries. @@ -62,9 +65,7 @@ internal void InitializeSentinel(ILogger? log) // Is the connection still valid? if (child.IsDisposed) { - child.ConnectionFailed -= OnManagedConnectionFailed; - child.ConnectionRestored -= OnManagedConnectionRestored; - sentinelConnectionChildren.Remove(messageParts[0]); + ForgetDisposedChild(messageParts[0], child); } else { @@ -515,6 +516,167 @@ private void ScheduleSentinelPrimaryReconnect(ConnectionMultiplexer connection, } } + /// + /// Unhooks and forgets a disposed managed connection. Must be called under the + /// lock. + /// + private void ForgetDisposedChild(string serviceName, ConnectionMultiplexer child) + { + child.ConnectionFailed -= OnManagedConnectionFailed; + child.ConnectionRestored -= OnManagedConnectionRestored; + sentinelConnectionChildren.Remove(serviceName); + } + + /// + /// Re-resolves the primary through sentinel for every managed connection whose primary has been demoted + /// while its connection stayed up. + /// + /// + /// + /// Every other recovery path needs a signal that this case does not produce: +switch-master can be + /// missed (pub/sub delivery is not guaranteed, and the sentinel-clients guidance says not to rely on it), + /// and the reconnect timer and the restore handler only run once a connection fails. A primary that is + /// demoted in place, with nothing announced and nothing disconnected, is still noticed by the managed + /// connection - its periodic INFO replication sees the role change - but then only re-elected + /// among the nodes it already knows. When sentinel promoted a node at an address it has never seen, + /// there is no primary among them, and nothing asked sentinel again. + /// + /// + /// Runs on this sentinel connection's heartbeat. The check reads only local state, so sentinel is asked + /// only when the managed connection itself reports the demotion, and at most one recovery per managed + /// connection is in flight. A recovery that does not take - sentinel not yet updated, or a reconfigure + /// already running - is simply retried on a later heartbeat, and the check clears itself once the + /// sentinel-reported primary is connected as a primary. + /// + /// + private void CheckManagedPrimariesDemoted() + { + List? demoted = null; + List? disposed = null; + lock (sentinelConnectionChildren) + { + foreach (var pair in sentinelConnectionChildren) + { + if (pair.Value.IsDisposed) + { + (disposed ??= new List()).Add(pair.Key); + } + else if (IsSentinelPrimaryDemoted(pair.Value)) + { + (demoted ??= new List()).Add(pair.Value); + } + } + + if (disposed is not null) + { + foreach (var serviceName in disposed) + { + ForgetDisposedChild(serviceName, sentinelConnectionChildren[serviceName]); + } + } + } + + if (demoted is null) return; + foreach (var child in demoted) + { + if (Interlocked.CompareExchange(ref child._demotedPrimaryRecoveryActive, 1, 0) != 0) continue; + + // asking sentinel and SwitchPrimary both block; keep that off the heartbeat + ThreadPool.QueueUserWorkItem(_ => + { + 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); + } + }); + } + } + + /// + /// One recovery attempt for a managed connection whose sentinel-reported primary was seen demoted. + /// + /// + /// Sentinel is asked first, because what to do depends on its answer, and a repeat of an unchanged answer + /// must stay cheap - the attempt runs on every heartbeat until the view clears: + /// + /// no answer: nothing to act on; the next heartbeat asks again. + /// a node we hold as a connected replica: either sentinel has not failed over yet, or our + /// view of that node's role is stale. Re-reading its role settles both, where a switch would reconfigure every + /// 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 + /// evaluated, rather than asking again and possibly acting on a different one. + /// + /// 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) + { + 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 + if (Volatile.Read(ref connection.activeConfigCause) is not null) return; + + if (GetConfiguredPrimaryForService(serviceName) is not EndPoint reported) return; + if (connection.TryResolveServerEndPoint(reported) is { IsConnected: true, IsReplica: true } replica + && replica.CheckInfoReplication()) + { + return; + } + + // the view may have been put right while sentinel was answering + if (!IsSentinelPrimaryDemoted(connection)) return; + + if (SwitchPrimary(connection.currentSentinelPrimaryEndPoint, connection, reportedPrimary: reported) + && connection.TryResolveServerEndPoint(reported) is { IsConnected: true, IsReplica: false }) + { + // As OnManagedConnectionRestored does: we are connected to a sentinel-approved primary again, so a + // soft-failed connect's discovery failure should stop being folded into every timeout message. + connection.LastException = null; + } + } + + /// + /// Whether the primary sentinel last reported for is connected and reporting + /// itself as a replica, or was never added (a switch to it whose reconfigure did not run). A disconnected + /// primary is not this case: that is the reconnect timer's job. + /// + private static bool IsSentinelPrimaryDemoted(ConnectionMultiplexer connection) + { + if (connection.IsDisposed + || Volatile.Read(ref connection.sentinelPrimaryReconnectTimer) is not null + || connection.currentSentinelPrimaryEndPoint is not EndPoint primary) + { + return false; + } + + // a lookup, not GetServerEndPoint: that would create a server for an endpoint the connection does not have + if (connection.TryResolveServerEndPoint(primary) is { } server) + { + return server.IsConnected && server.IsReplica; + } + + // never added: a switch to it that was refused (another reconfigure held the gate) or threw. A missing server + // alone is not enough evidence - the endpoint sentinel gave may simply be spelled differently from the + // server's own (a DNS name against an address) - so only while nothing else is serving as the primary, + // which keeps a healthy connection from switching on every heartbeat. + foreach (var other in connection.GetServerSnapshot()) + { + if (other.ServerType != ServerType.Sentinel && other.IsConnected && !other.IsReplica) return false; + } + return true; + } + internal EndPoint? GetConfiguredPrimaryForService(string serviceName) => _serverSnapshot // same as GetServerSnapshot, but without forcing span .Where(static s => s.ServerType == ServerType.Sentinel) @@ -543,78 +705,108 @@ private void ScheduleSentinelPrimaryReconnect(ConnectionMultiplexer connection, /// 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. - internal void SwitchPrimary(EndPoint? switchBlame, ConnectionMultiplexer connection, TextWriter? writer = null) + /// 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) { var logger = Logger.With(writer); if (connection.RawConfig.ServiceName is not string serviceName) { logger?.LogInformationServiceNameNotDefined(); - return; + return false; } - // Get new primary - try twice - EndPoint newPrimaryEndPoint = GetConfiguredPrimaryForService(serviceName) + // 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; - // 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) + // Claim the reconfigure gate BEFORE reshaping the connection. ReconfigureAsync refuses to run while + // another reconfigure holds this gate, so rewriting the endpoint list and retiring servers first would + // leave the connection reshaped but unconfigured whenever we lose that race - and two concurrent + // switches could interleave their endpoint rewrites. Claim first and a lost race touches nothing. + string cause = $"Primary switch {serviceName}"; + if (!connection.TryClaimReconfigure(cause)) { - Trace($"Switching primary to {newPrimaryEndPoint}"); - connection.ReconfigureAsync( - first: false, - reconfigureAll: reconfigureAll, - log: logger, - blame: switchBlame, - cause: $"Primary switch {serviceName}", - publishReconfigure: false, - publishReconfigureFlags: CommandFlags.PreferMaster).Wait(); - - UpdateSentinelAddressList(serviceName); + Trace($"Primary switch for {serviceName} skipped; reconfiguration already in progress"); + return false; } - if (!connection.servers.Contains(newPrimaryEndPoint)) + bool switched = false; + try { - // 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); + // 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) + { + 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); + } - connection.EndPoints.Clear(); - connection.EndPoints.TryAdd(newPrimaryEndPoint); - if (replicaEndPoints is not null) + if (!connection.servers.Contains(newPrimaryEndPoint)) { - foreach (var replicaEndPoint in replicaEndPoints) + // The sentinel-reported primary is an endpoint we have never seen (e.g. a + // recreated node that came back on a brand-new IP). Rebuild the endpoint set + // from scratch around it and reconfigure. + EndPoint[]? replicaEndPoints = GetReplicasForService(serviceName) + ?? GetReplicasForService(serviceName); + + connection.EndPoints.Clear(); + connection.EndPoints.TryAdd(newPrimaryEndPoint); + if (replicaEndPoints is not null) { - connection.EndPoints.TryAdd(replicaEndPoint); + foreach (var replicaEndPoint in replicaEndPoints) + { + connection.EndPoints.TryAdd(replicaEndPoint); + } } - } - // Only once sentinel gave us the whole picture: a null replica list is a failed query, not a - // deployment with no replicas, and retiring on one would drop healthy nodes over a hiccup. - if (replicaEndPoints is not null) + // Only once sentinel gave us the whole picture: a null replica list is a failed query, not a + // deployment with no replicas, and retiring on one would drop healthy nodes over a hiccup. + if (replicaEndPoints is not null) + { + RetireServersAbsentFromEndPoints(connection, logger); + } + + TriggerReconfigure(reconfigureAll: false); + } + else if (IsStalePrimaryView(connection, newPrimaryEndPoint)) { - RetireServersAbsentFromEndPoints(connection, logger); + // 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); } - TriggerReconfigure(reconfigureAll: false); + return switched; } - else if (IsStalePrimaryView(connection, newPrimaryEndPoint)) + finally { - // 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); + // not handed over (view already correct, or we threw before the reconfigure): release it ourselves + if (!switched) connection.ReleaseReconfigureClaim(); } } diff --git a/src/StackExchange.Redis/ConnectionMultiplexer.cs b/src/StackExchange.Redis/ConnectionMultiplexer.cs index ee4105444..a764e574d 100644 --- a/src/StackExchange.Redis/ConnectionMultiplexer.cs +++ b/src/StackExchange.Redis/ConnectionMultiplexer.cs @@ -1401,6 +1401,7 @@ internal void OnHeartbeat() Trace("heartbeat"); CheckTopologyRefreshDue(now); + if (_isSentinel) CheckManagedPrimariesDemoted(); var tmp = GetServerSnapshot(); int token = 0; @@ -1760,6 +1761,20 @@ 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 + /// 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 + /// - a leaked claim refuses every later reconfigure. + /// + internal bool TryClaimReconfigure(string cause) => + Interlocked.CompareExchange(ref activeConfigCause, cause, null) == null; + + /// Releases a claim taken by that was never handed to a reconfigure. + internal void ReleaseReconfigureClaim() => Interlocked.Exchange(ref activeConfigCause, null); + internal bool ReconfigureIfNeeded(EndPoint? blame, bool fromBroadcast, string cause, bool publishReconfigure = false, CommandFlags flags = CommandFlags.None) { if (fromBroadcast) @@ -1788,16 +1803,23 @@ internal bool ReconfigureIfNeeded(EndPoint? blame, bool fromBroadcast, string ca public Task ReconfigureAsync(string reason) => ReconfigureAsync(first: false, reconfigureAll: false, log: Logger, blame: null, cause: reason); - internal async Task ReconfigureAsync(bool first, bool reconfigureAll, ILogger? log, EndPoint? blame, string cause, bool publishReconfigure = false, CommandFlags publishReconfigureFlags = CommandFlags.None) + internal async Task ReconfigureAsync(bool first, bool reconfigureAll, ILogger? log, EndPoint? blame, string cause, bool publishReconfigure = false, CommandFlags publishReconfigureFlags = CommandFlags.None, bool claimed = false) { - if (_isDisposed) throw new ObjectDisposedException(ToString()); + if (_isDisposed) + { + // a caller-held claim must not leak: nothing else would ever release it + if (claimed) ReleaseReconfigureClaim(); + throw new ObjectDisposedException(ToString()); + } bool showStats = log is not null; bool ranThisCall = false; try { - // Note that we *always* exchange the reason (first one counts) to prevent duplicate runs - ranThisCall = Interlocked.CompareExchange(ref activeConfigCause, cause, null) == null; + // Note that we *always* exchange the reason (first one counts) to prevent duplicate runs; + // claimed means the caller already owns the gate (TryClaimReconfigure) and we run under - and + // release - that claim + ranThisCall = claimed || Interlocked.CompareExchange(ref activeConfigCause, cause, null) == null; if (!ranThisCall) { diff --git a/tests/StackExchange.Redis.Tests/InProcessSentinelServer.cs b/tests/StackExchange.Redis.Tests/InProcessSentinelServer.cs new file mode 100644 index 000000000..8fecb98e6 --- /dev/null +++ b/tests/StackExchange.Redis.Tests/InProcessSentinelServer.cs @@ -0,0 +1,147 @@ +using System; +using System.Collections.Generic; +using System.IO; +using System.Net; +using System.Net.Sockets; +using System.Threading; +using System.Threading.Tasks; +using RESPite.Messages; +using StackExchange.Redis.Configuration; +using StackExchange.Redis.Server; +using Xunit; + +namespace StackExchange.Redis.Tests; + +/// +/// An in-process sentinel that monitors one service on an : it answers the +/// SENTINEL queries a client makes from the monitored server's topology, and does nothing on its own - +/// no failover, no +switch-master. Which node is the primary is whatever the test says it is, which is +/// what lets a test put sentinel and the client's cached view out of step. +/// +public sealed class InProcessSentinelServer : InProcessTestServer +{ + private readonly InProcessTestServer _monitored; + private EndPoint _primary; + + public InProcessSentinelServer(InProcessTestServer monitored, string serviceName, EndPoint primary, EndPoint endpoint, ITestOutputHelper? log = null) + : base(log, endpoint) + { + _monitored = monitored ?? throw new ArgumentNullException(nameof(monitored)); + ServiceName = serviceName ?? throw new ArgumentNullException(nameof(serviceName)); + _primary = primary ?? throw new ArgumentNullException(nameof(primary)); + ServerType = ServerType.Sentinel; + } + + public string ServiceName { get; } + + /// + /// The primary this sentinel reports for ; changing it announces nothing. + /// + public EndPoint Primary + { + get => Volatile.Read(ref _primary); + set => Volatile.Write(ref _primary, value ?? throw new ArgumentNullException(nameof(value))); + } + + /// + /// A tunnel that routes each endpoint to whichever of these servers owns it, so that one client configuration + /// can reach the sentinel and the nodes it monitors. + /// + public static Tunnel CreateTunnel(params InProcessTestServer[] servers) => new RoutingTunnel(servers); + + private int _primaryQueries; + + /// How many times a client has asked this sentinel for the primary. + public int PrimaryQueries => Volatile.Read(ref _primaryQueries); + + [RedisCommand(3, nameof(RedisCommand.SENTINEL), "get-master-addr-by-name", LockFree = true)] + private TypedRedisValue SentinelGetMasterAddrByName(RedisClient client, in RedisRequest request) + { + Interlocked.Increment(ref _primaryQueries); + if (!IsService(request)) return TypedRedisValue.NullArray(RespPrefix.Array); + 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()); + return reply; + } + + [RedisCommand(3, nameof(RedisCommand.SENTINEL), "replicas", LockFree = true)] + private TypedRedisValue SentinelReplicas(RedisClient client, in RedisRequest request) => Replicas(request); + + [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"); + + [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"); + return reply; + } + + private bool IsService(in RedisRequest request) => string.Equals(request.GetString(2), ServiceName, StringComparison.Ordinal); + + private TypedRedisValue Replicas(in RedisRequest request) + { + if (!IsService(request)) return TypedRedisValue.Error("ERR No such master with that name"); + + // 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)) + { + foreach (var endpoint in _monitored.GetEndPoints()) + { + if (_monitored.TryGetNode(endpoint, out var node) && node.PrimaryId == primaryNode.Id) + { + 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"); + } + return reply; + } + + 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[2] = TypedRedisValue.BulkString("ip"); + span[3] = TypedRedisValue.BulkString(host); + span[4] = TypedRedisValue.BulkString("port"); + span[5] = TypedRedisValue.BulkString(port.ToString()); + span[6] = TypedRedisValue.BulkString("flags"); + span[7] = TypedRedisValue.BulkString(flags); + return item; + } + + private sealed class RoutingTunnel(InProcessTestServer[] servers) : Tunnel + { + private Tunnel? Route(EndPoint endpoint) + { + foreach (var server in servers) + { + if (server.TryGetNode(endpoint, out _)) return server.Tunnel; + } + return null; + } + + public override ValueTask GetSocketConnectEndpointAsync(EndPoint endpoint, CancellationToken cancellationToken) + => Route(endpoint) is { } tunnel ? tunnel.GetSocketConnectEndpointAsync(endpoint, cancellationToken) : base.GetSocketConnectEndpointAsync(endpoint, cancellationToken); + + 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); + } +} diff --git a/tests/StackExchange.Redis.Tests/SentinelPrimaryDemotionUnitTests.cs b/tests/StackExchange.Redis.Tests/SentinelPrimaryDemotionUnitTests.cs new file mode 100644 index 000000000..2a6a879c0 --- /dev/null +++ b/tests/StackExchange.Redis.Tests/SentinelPrimaryDemotionUnitTests.cs @@ -0,0 +1,127 @@ +using System; +using System.Diagnostics; +using System.Linq; +using System.Net; +using System.Threading; +using System.Threading.Tasks; +using Xunit; + +namespace StackExchange.Redis.Tests; + +/// +/// A sentinel-managed connection whose primary is demoted while every connection stays up. Sentinel's own +/// recovery levers - +switch-master, and killing the demoted node's clients - are exactly what is +/// missing here, so the only signal the client gets is the demoted node reporting itself as a replica. +/// +public class SentinelPrimaryDemotionUnitTests(ITestOutputHelper log) +{ + private const string ServiceName = "mymaster"; + + // documentation-range addresses (RFC 5737): a node that is not in-process can never be reached by accident + private static readonly EndPoint PrimaryA = new IPEndPoint(IPAddress.Parse("192.0.2.1"), 6379); + 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); + + [Fact] + public async Task DemotedPrimaryFollowsSentinelToPrimaryAtUnknownAddress() + { + 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.Equal([PrimaryA], ConnectedPrimaries(conn)); + + // the primary moves to a node the client has never been told about - a pod that came back on a new + // address - and the client misses the announcement; no connection is dropped + FailOverToNewNode(data); + sentinel.Primary = NewNodeC; + + await AssertPrimaryBecomesAsync(conn, NewNodeC); + } + + [Fact] + public async Task DemotedPrimaryDoesNotReconfigureWhileSentinelStillNamesIt() + { + 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)); + int configurationChanges = 0; + conn.ConfigurationChanged += (_, _) => Interlocked.Increment(ref configurationChanges); + + // sentinel has not caught up: it still names the node that is now a replica + FailOverToNewNode(data); + await UntilAsync(() => ConnectedPrimaries(conn).Length == 0, "the demotion was not noticed"); + + // The property under test: the client keeps asking sentinel, but does not reconfigure every node to + // elect a primary that does not exist. Anchor on the sentinel polls themselves rather than wall-clock + // quiet: require a stretch of several polls with zero configuration changes. A straggler reconfigure + // from the demotion itself re-baselines the stretch; the per-heartbeat reconfigure storm this guards + // against would raise a change on every poll, so no clean stretch would ever complete. + int changesBaseline = Volatile.Read(ref configurationChanges); + int queriesBaseline = sentinel.PrimaryQueries; + await UntilAsync( + () => + { + int changesNow = Volatile.Read(ref configurationChanges); + if (changesNow != changesBaseline) + { + changesBaseline = changesNow; + queriesBaseline = sentinel.PrimaryQueries; + return false; + } + return sentinel.PrimaryQueries >= queriesBaseline + 3; + }, + "no stretch of sentinel polls without a reconfigure"); + log.WriteLine($"clean stretch: {sentinel.PrimaryQueries - queriesBaseline} sentinel polls, configuration changes still {changesBaseline}"); + + // and once sentinel does catch up, the connection follows it + sentinel.Primary = NewNodeC; + await AssertPrimaryBecomesAsync(conn, NewNodeC); + } + + private static ConfigurationOptions CreateConfig(InProcessTestServer data, InProcessSentinelServer sentinel) + { + var config = data.GetClientConfig(); + config.EndPoints.Clear(); + config.EndPoints.Add(SentinelEndPoint); + config.ServiceName = ServiceName; + config.Tunnel = InProcessSentinelServer.CreateTunnel(data, sentinel); + config.ConfigCheckSeconds = 1; // how soon the demoted node's role is re-read + return config; + } + + // C is promoted, and A and B follow it; nothing is announced and no connection drops + private static void FailOverToNewNode(InProcessTestServer data) + { + data.AddEmptyNode(NewNodeC); + data.SetReplicaOf(PrimaryA, NewNodeC); + data.SetReplicaOf(ReplicaB, NewNodeC); + } + + private async Task AssertPrimaryBecomesAsync(IConnectionMultiplexer conn, EndPoint expected) + { + var timer = Stopwatch.StartNew(); + await UntilAsync(() => ConnectedPrimaries(conn).SequenceEqual([expected]), $"primary did not become {Format.ToString(expected)}"); + log.WriteLine($"primary is {Format.ToString(expected)} after {timer.ElapsedMilliseconds}ms"); + + Assert.Equal([expected], ConnectedPrimaries(conn)); + Assert.Equal(expected, conn.GetDatabase().IdentifyEndpoint("demotion-key", CommandFlags.DemandMaster)); + } + + private static async Task UntilAsync(Func condition, string failure) + { + var timer = Stopwatch.StartNew(); + while (!condition()) + { + if (timer.Elapsed > TimeSpan.FromSeconds(20)) Assert.Fail($"{failure} within {timer.Elapsed.TotalSeconds:0}s"); + await Task.Delay(100); + } + } + + private static EndPoint[] ConnectedPrimaries(IConnectionMultiplexer conn) + => conn.GetServers().Where(static s => s.IsConnected && !s.IsReplica).Select(static s => s.EndPoint).ToArray(); +} diff --git a/toys/StackExchange.Redis.Server/RedisServer.cs b/toys/StackExchange.Redis.Server/RedisServer.cs index beb4aea65..a0862ff86 100644 --- a/toys/StackExchange.Redis.Server/RedisServer.cs +++ b/toys/StackExchange.Redis.Server/RedisServer.cs @@ -149,6 +149,39 @@ public void Failover(EndPoint replica) Log($"failover: {Format.ToString(replica)} promoted, {demoted.Host}:{demoted.Port} demoted"); } + /// + /// Makes replicate - or a primary again, when + /// is null - as REPLICAOF does on a standalone server. Nothing is announced and no connection is dropped; + /// a client finds out from ROLE or INFO replication, or not at all. Both nodes are left serving + /// the whole keyspace, as every standalone node does - including one added by , + /// which otherwise owns no slots and would redirect every key. + /// + public void SetReplicaOf(EndPoint node, EndPoint primary) + { + if (ServerType == ServerType.Cluster) throw new InvalidOperationException($"Server mode is {ServerType}; use {nameof(Failover)}"); + if (!TryGetNode(node ?? throw new ArgumentNullException(nameof(node)), out var target)) throw new KeyNotFoundException($"Node not found: {Format.ToString(node)}"); + Node primaryNode = null; + if (primary is not null) + { + if (!TryGetNode(primary, out primaryNode)) throw new KeyNotFoundException($"Node not found: {Format.ToString(primary)}"); + if (ReferenceEquals(primaryNode, target)) throw new ArgumentException("A node cannot replicate itself", nameof(primary)); + } + + target.UpdateSlots(null); + if (primaryNode is null) + { + target.PrimaryId = null; + target.Flags &= ~NodeFlags.Replica; + } + else + { + primaryNode.UpdateSlots(null); + target.PrimaryId = primaryNode.Id; + target.Flags |= NodeFlags.Replica; + } + Log($"replicaof: {Format.ToString(node)} now replicates {(primary is null ? "no one" : Format.ToString(primary))}"); + } + public bool Migrate(Span key, EndPoint to) => Migrate(ServerSelectionStrategy.GetClusterSlot(key), to); public bool Migrate(in RedisKey key, EndPoint to) => Migrate(GetHashSlot(key), to);