diff --git a/src/StackExchange.Redis/ClusterSlots.Server.cs b/src/StackExchange.Redis/ClusterSlots.Server.cs index f93583187..6005d0191 100644 --- a/src/StackExchange.Redis/ClusterSlots.Server.cs +++ b/src/StackExchange.Redis/ClusterSlots.Server.cs @@ -5,8 +5,8 @@ namespace StackExchange.Redis; internal partial class RedisServer { public ClusterSlotsResult? ClusterSlots(CommandFlags flags = CommandFlags.None) - => ExecuteSync(Message.Create(-1, flags, RedisCommand.CLUSTER, RedisLiterals.SLOTS), ClusterSlotsResult.Processor); + => ExecuteSync(GetClusterSlotsMessage(flags), ClusterSlotsResult.Processor); public Task ClusterSlotsAsync(CommandFlags flags = CommandFlags.None) - => ExecuteAsync(Message.Create(-1, flags, RedisCommand.CLUSTER, RedisLiterals.SLOTS), ClusterSlotsResult.Processor); + => ExecuteAsync(GetClusterSlotsMessage(flags), ClusterSlotsResult.Processor); } diff --git a/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs b/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs index 9e82d447f..ca6599ae4 100644 --- a/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs +++ b/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs @@ -515,7 +515,7 @@ void TriggerReconfigure(bool reconfigureAll) /// 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); + var newPrimaryServer = connection.GetServerEndPoint(newPrimaryEndPoint, activate: false, provenance: ServerProvenance.Sentinel); // 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. diff --git a/src/StackExchange.Redis/ConnectionMultiplexer.cs b/src/StackExchange.Redis/ConnectionMultiplexer.cs index a585c4437..2a334c431 100644 --- a/src/StackExchange.Redis/ConnectionMultiplexer.cs +++ b/src/StackExchange.Redis/ConnectionMultiplexer.cs @@ -1,5 +1,6 @@ using System; using System.Collections; +using System.Collections.Concurrent; using System.Collections.Generic; using System.ComponentModel; using System.Diagnostics.CodeAnalysis; @@ -38,6 +39,11 @@ public sealed partial class ConnectionMultiplexer : IInternalConnectionMultiplex private TimerToken? pulse; private readonly Hashtable servers = new Hashtable(); + + // secondary identities: a node can legitimately arrive under more than one name (an address and an + // announced hostname), and `servers` is keyed on exact endpoint equality, so without this the same + // node reached by its other name becomes a second ServerEndPoint - see #2826 + private readonly ConcurrentDictionary _serverIdentities = new(); private volatile ServerSnapshot _serverSnapshot = ServerSnapshot.Empty; private volatile bool _isDisposed; @@ -816,6 +822,36 @@ internal ServerSnapshot Add(ServerEndPoint value) return new ServerSnapshot(nextEndpoints, _count + 1); } + /// + /// Returns a snapshot without , or this one if it was not present. + /// + /// + /// Unlike this can never reuse the existing array. Add is allowed to write into + /// spare capacity because older readers hold a smaller count and so never observe the new slot; + /// removal would have to *shift* elements a concurrent reader may be part-way through enumerating, + /// so it always allocates a compacted copy. + /// + internal ServerSnapshot Remove(ServerEndPoint value) + { + if (value is null) return this; + + int index = -1; + for (int i = 0; i < _count; i++) + { + if (ReferenceEquals(_endpoints[i], value)) + { + index = i; + break; + } + } + if (index < 0) return this; + + var next = new ServerEndPoint[Math.Max(_count - 1, 1)]; + if (index > 0) Array.Copy(_endpoints, 0, next, 0, index); + if (index < _count - 1) Array.Copy(_endpoints, index + 1, next, index, _count - 1 - index); + return new ServerSnapshot(next, _count - 1); + } + internal EndPoint[] GetEndPoints() { if (_count == 0) return Array.Empty(); @@ -924,11 +960,192 @@ public ServerSnapshotFiltered(ServerEndPoint[] endpoints, int count, Func GetServerEndPoint(endpoint); + /// + /// Finds an existing server by any identity it is known to answer to, without creating one. + /// + internal ServerEndPoint? TryResolveServerEndPoint(EndPoint? endpoint) + { + if (endpoint is null) return null; + if (servers[endpoint] is ServerEndPoint exact) return exact; + + // a retired server may still be referenced by an alias for a moment; never hand one back, or the + // caller receives something whose bridges are gone + return _serverIdentities.TryGetValue(endpoint, out var byIdentity) && !byIdentity.IsDisposed + ? byIdentity : null; + } + + // bumped once per applied cluster topology; absence is measured in these rather than in time, so a + // quiet client cannot age an endpoint out simply by not reconfiguring + private int _topologyGeneration; + + /// + /// How many consecutive topology generations a server must be missing from before it may be pruned. + /// A single reply is one node's view, so one absence is not evidence. + /// + private const int PruneAfterMissingGenerations = 3; + + /// + /// Records which servers the freshly-applied topology listed, and retires those which have been absent + /// long enough - and which nothing depends on. + /// + internal async Task ApplyTopologyGenerationAsync(ClusterTopology topology, ILogger? log = null) + { + if (topology is null) return; + + var generation = Interlocked.Increment(ref _topologyGeneration); + + // resolve the listed nodes to servers we hold, by any identity. More than one distinct server for + // one node-id is a duplicate: the same process reached under two names, which costs a second + // socket and splits backlog and subscription state across the pair + var seen = new HashSet(); + List<(ServerEndPoint Survivor, ServerEndPoint Loser)>? merge = null; + foreach (var node in topology.Nodes) + { + ServerEndPoint? survivor = null; + foreach (var identity in node.Identities) + { + if (TryResolveServerEndPoint(identity) is not { } server) continue; + + if (survivor is null) + { + // Identities is ordered with the form the answering node advertised first, so the + // earliest match is also the deployment's stated preference + survivor = server; + } + else if (!ReferenceEquals(survivor, server)) + { + // ...unless one of them was configured: those may never be retired, so they win + var (keep, drop) = server.Provenance == ServerProvenance.Configured + ? (server, survivor) + : (survivor, server); + survivor = keep; + (merge ??= new()).Add((keep, drop)); + } + } + + if (survivor is not null) + { + survivor.OnSeenInTopology(generation); + seen.Add(survivor); + + // every name this node answers to now resolves to the survivor, so a caller holding the + // retired one keeps working rather than breaking + foreach (var identity in node.Identities) + { + if (servers[identity] is null) _serverIdentities[identity] = survivor; + } + } + } + + if (merge is not null) + { + foreach (var (survivor, loser) in merge) + { + if (loser.IsDisposed || loser.Provenance == ServerProvenance.Configured) continue; + log?.LogInformationMergingDuplicateServer(new(loser.EndPoint), new(survivor.EndPoint)); + _serverIdentities[loser.EndPoint] = survivor; + await RetireServerAsync(loser, "duplicate of " + Format.ToString(survivor.EndPoint), log: log).ForAwait(); + seen.Add(loser); // already handled; do not also consider it for pruning + } + } + + List? prune = null; + foreach (var server in GetServerSnapshot()) + { + if (seen.Contains(server) || server.IsDisposed) continue; + + // only cluster-discovered nodes may be pruned by cluster absence: a configured endpoint is the + // seed we need to bootstrap after a full rotation, and sentinel-discovered nodes are listed by + // a source that did not run here + if (server.Provenance != ServerProvenance.ClusterTopology) continue; + + var missingFor = server.OnMissingFromTopology(generation); + if (missingFor < PruneAfterMissingGenerations || !server.IsIdle()) continue; + + (prune ??= new List()).Add(server); + } + + if (prune is null) return; + foreach (var server in prune) + { + log?.LogInformationPruningServer(new(server.EndPoint), PruneAfterMissingGenerations); + await RetireServerAsync(server, "absent from topology", log: log).ForAwait(); + } + } + + /// + /// Drains and removes a server: it stops being selected, finishes what it owes, then is forgotten - + /// including every secondary identity that pointed at it. + /// + /// + /// The single spelling of retirement, deliberately: topology pruning, duplicate merging, and the + /// endpoint handoffs of the maintenance-notification work all want the same drain-then-close, and + /// having one of them reach for instead would abandon in-flight + /// commands. + /// + internal async Task RetireServerAsync(ServerEndPoint server, string reason, TimeSpan? drainTimeout = null, ILogger? log = null) + { + if (server is null) return; + + await server.RetireAsync(reason, drainTimeout ?? TimeSpan.FromSeconds(5), log).ForAwait(); + + lock (servers) + { + // by endpoint, not by scanning: the server is keyed on exactly one + if (ReferenceEquals(servers[server.EndPoint], server)) + { + servers.Remove(server.EndPoint); + } + _serverSnapshot = _serverSnapshot.Remove(server); + } + + // ...and drop the aliases, or a later lookup resolves to something disposed + foreach (var pair in _serverIdentities) + { + if (ReferenceEquals(pair.Value, server)) + { + _serverIdentities.TryRemove(pair.Key, out _); + } + } + } + + /// + /// Records the additional names a known node answers to, so that reaching it by any of them resolves + /// to the one . Deliberately does not create anything: a node we have + /// never heard of is a matter for discovery, not for identity. + /// + internal void RegisterServerIdentities(ClusterTopology topology) + { + foreach (var node in topology.Nodes) + { + // resolve via any identity we already hold; if we know the node at all, the rest are aliases + ServerEndPoint? known = null; + foreach (var identity in node.Identities) + { + if ((known = TryResolveServerEndPoint(identity)) is not null) break; + } + if (known is null) continue; + + foreach (var identity in node.Identities) + { + if (servers[identity] is not null) continue; // already a server in its own right + if (_serverIdentities.TryAdd(identity, known)) + { + Trace($"Identity {Format.ToString(identity)} -> {Format.ToString(known.EndPoint)}"); + } + } + } + } + [return: NotNullIfNotNull(nameof(endpoint))] - internal ServerEndPoint? GetServerEndPoint(EndPoint? endpoint, ILogger? log = null, bool activate = true) + internal ServerEndPoint? GetServerEndPoint( + EndPoint? endpoint, + ILogger? log = null, + bool activate = true, + ServerProvenance provenance = ServerProvenance.ClusterTopology) { if (endpoint == null) return null; - var server = (ServerEndPoint?)servers[endpoint]; + var server = (ServerEndPoint?)servers[endpoint] ?? TryResolveServerEndPoint(endpoint); if (server == null) { bool isNew = false; @@ -939,7 +1156,7 @@ public ServerSnapshotFiltered(ServerEndPoint[] endpoints, int count, Func private async Task GetEndpointsFromClusterNodes(ServerEndPoint server, ILogger? log) { - var message = RedisServer.GetClusterNodesMessage(CommandFlags.None); try { - var clusterConfig = await ExecuteAsyncImpl(message, ResultProcessor.ClusterNodes, null, server).ForAwait(); + // both views, freshly: SLOTS says who serves what and under which names, NODES lists every + // node including those serving nothing. Asked as a pair for symmetry - trusting the topology + // cached from autoconfigure here would mean acting on possibly-stale data while deliberately + // re-reading the other half + var slotsTask = ExecuteAsyncImpl( + RedisServer.GetClusterSlotsMessage(CommandFlags.None), ResultProcessor.ClusterSlots, null, server); + var nodesTask = ExecuteAsyncImpl( + RedisServer.GetClusterNodesMessage(CommandFlags.None), ResultProcessor.ClusterNodes, null, server); + + var slots = await slotsTask.ForAwait(); + var clusterConfig = await nodesTask.ForAwait(); if (clusterConfig is null) { return null; } - var clusterEndpoints = new EndPointCollection(clusterConfig.Nodes.Where(node => node.EndPoint is not null && !node.IgnoreFromClient).Select(node => node.EndPoint!).ToList()); - // Loop through nodes in the cluster and update nodes relations to other nodes - ServerEndPoint? serverEndpoint = null; + + var topology = ClusterTopology.From(slots); + var clusterEndpoints = new EndPointCollection(); + + if (topology is not null) + { + // SLOTS drives discovery: these are the nodes that serve traffic, so they are the ones we + // connect to. Resolve through every identity first, so a node we already hold under + // another name is not duplicated + RegisterServerIdentities(topology); + foreach (var node in topology.Nodes) + { + if (SelectIdentity(node) is { } endpoint) clusterEndpoints.TryAdd(endpoint); + } + } + + // ...and NODES contributes the remainder - nodes serving no slots do not appear in SLOTS at + // all. Registered *inert*: known and addressable via GetServer, but not dialled, since + // there is nothing to route to them. First use creates the bridge, so nothing is lost + foreach (var node in clusterConfig.Nodes) + { + if (node.EndPoint is null || node.IgnoreFromClient) continue; + + if (topology is null) + { + // no usable SLOTS view (pre-4.0, or an error reply): behave exactly as before + clusterEndpoints.TryAdd(node.EndPoint); + } + else if (TryResolveServerEndPoint(node.EndPoint) is null) + { + log?.LogInformationRegisteringInertNode(new(node.EndPoint)); + GetServerEndPoint(node.EndPoint, activate: false); + } + } + + // node relations come from NODES either way - SLOTS conveys replica-of by position, but not + // the node ids and flags that Primary/Replicas resolution uses foreach (EndPoint endpoint in clusterEndpoints) { - serverEndpoint = GetServerEndPoint(endpoint); - serverEndpoint?.UpdateNodeRelations(clusterConfig); + GetServerEndPoint(endpoint)?.UpdateNodeRelations(clusterConfig); + } + + // ...and now that the topology is applied, age out anything it has stopped listing + if (topology is not null) + { + await ApplyTopologyGenerationAsync(topology, log).ForAwait(); } return clusterEndpoints; } @@ -1825,6 +2090,17 @@ public EndPoint[] GetEndPoints(bool configuredOnly = false) => log?.LogErrorEncounteredErrorWhileUpdatingClusterConfig(ex, ex.Message); return null; } + + // an identity we already hold if there is one - otherwise the form the answering node advertised, + // which is what Identities is ordered by, and which is the form TLS can validate + EndPoint? SelectIdentity(ClusterTopologyNode node) + { + foreach (var identity in node.Identities) + { + if (TryResolveServerEndPoint(identity) is { } known) return known.EndPoint; + } + return node.Identities.Count > 0 ? node.Identities[0] : null; + } } private void ResetAllNonConnected() @@ -1994,6 +2270,48 @@ internal void UpdateClusterRange(ClusterConfiguration configuration) } } + /// + /// Applies the slot map from the CLUSTER SLOTS view, which supersedes + /// when the answering server supplied one. + /// + /// + /// Preferred over the CLUSTER NODES view because it is keyed on node-id and carries both naming + /// forms, so an endpoint arriving under a different name than we hold resolves to the server we + /// already have rather than creating a duplicate. + /// + internal void UpdateClusterRange(ClusterTopology topology) + { + if (topology is null) return; + + foreach (var node in topology.Nodes) + { + if (node.IsReplica || node.Slots.Count == 0) continue; + + // resolve by *any* identity the node answers to before falling back to the form this reply + // happened to use; that is what keeps one node from becoming two ServerEndPoints + var server = ResolveOrCreate(node); + if (server is null) continue; + + foreach (var slot in node.Slots) + { + ServerSelectionStrategy.UpdateClusterRange(slot.From, slot.To, server); + } + } + + ServerEndPoint? ResolveOrCreate(ClusterTopologyNode node) + { + foreach (var identity in node.Identities) + { + if (TryResolveServerEndPoint(identity) is { } known) return known; + } + + // unknown node: dial the form the answering node *advertised*, which Identities is ordered by. + // Not the address by preference: a certificate validates against a name, and where hostnames + // are preferred the advertised address may not even be routable (#2826) + return node.Identities.Count > 0 ? GetServerEndPoint(node.Identities[0]) : null; + } + } + internal ServerEndPoint? SelectServer(Message? message) => message == null ? null : ServerSelectionStrategy.Select(message); diff --git a/src/StackExchange.Redis/LoggerExtensions.cs b/src/StackExchange.Redis/LoggerExtensions.cs index 552a38115..03c62ed95 100644 --- a/src/StackExchange.Redis/LoggerExtensions.cs +++ b/src/StackExchange.Redis/LoggerExtensions.cs @@ -326,6 +326,36 @@ internal static void LogWithThreadPoolStats(this ILogger? log, string message) Message = "Encountered error while updating cluster config: {ErrorMessage}")] internal static partial void LogErrorEncounteredErrorWhileUpdatingClusterConfig(this ILogger logger, Exception exception, string errorMessage); + [LoggerMessage( + Level = LogLevel.Information, + EventId = 111, + Message = "Registering {EndPoint} without connecting: serves no slots")] + internal static partial void LogInformationRegisteringInertNode(this ILogger logger, EndPointLogValue endPoint); + + [LoggerMessage( + Level = LogLevel.Information, + EventId = 115, + Message = "Merging {EndPoint} into {Survivor}: the same node under two names")] + internal static partial void LogInformationMergingDuplicateServer(this ILogger logger, EndPointLogValue endPoint, EndPointLogValue survivor); + + [LoggerMessage( + Level = LogLevel.Information, + EventId = 114, + Message = "Pruning {EndPoint}: absent from the topology for {Generations} generations")] + internal static partial void LogInformationPruningServer(this ILogger logger, EndPointLogValue endPoint, int generations); + + [LoggerMessage( + Level = LogLevel.Information, + EventId = 112, + Message = "Retiring {EndPoint}: {Reason}")] + internal static partial void LogInformationRetiringServer(this ILogger logger, EndPointLogValue endPoint, string reason); + + [LoggerMessage( + Level = LogLevel.Information, + EventId = 113, + Message = "Retiring {EndPoint} with {Outstanding} operation(s) still outstanding; closing anyway")] + internal static partial void LogInformationRetiringServerAbandoned(this ILogger logger, EndPointLogValue endPoint, int outstanding); + [LoggerMessage( Level = LogLevel.Information, EventId = 43, diff --git a/src/StackExchange.Redis/RedisServer.cs b/src/StackExchange.Redis/RedisServer.cs index f8cfcdc05..17767a73d 100644 --- a/src/StackExchange.Redis/RedisServer.cs +++ b/src/StackExchange.Redis/RedisServer.cs @@ -162,6 +162,14 @@ public Task ClientListAsync(CommandFlags flags = CommandFlags.None internal static Message GetClusterNodesMessage(CommandFlags flags) => Message.Create(-1, flags.WithCategory(NodeLocalRead), RedisCommand.CLUSTER, RedisLiterals.NODES); + /// + /// As , for the CLUSTER SLOTS view of the same topology: + /// likewise asked both by the public API and by the autoconfigure probe, and likewise a node-local + /// read - it reports what the answering node believes, so it is safe to replay against that node. + /// + internal static Message GetClusterSlotsMessage(CommandFlags flags) + => Message.Create(-1, flags.WithCategory(NodeLocalRead), RedisCommand.CLUSTER, RedisLiterals.SLOTS); + public KeyValuePair[] ConfigGet(RedisValue pattern = default, CommandFlags flags = CommandFlags.None) { var msg = GetConfigGetMessage(pattern, flags); diff --git a/src/StackExchange.Redis/ServerEndPoint.cs b/src/StackExchange.Redis/ServerEndPoint.cs index 20968261b..7ff9ed510 100644 --- a/src/StackExchange.Redis/ServerEndPoint.cs +++ b/src/StackExchange.Redis/ServerEndPoint.cs @@ -22,6 +22,34 @@ internal enum UnselectableFlags RedundantPrimary = 1, DidNotRespond = 2, ServerType = 4, + + /// This server is being retired; it must not be selected for new work. + Retiring = 8, + } + + /// + /// How we came to know about a server, which decides what may retire it: "absent from the topology" only + /// means anything if the source that would have listed it actually ran. + /// + internal enum ServerProvenance + { + /// Named in ; never pruned. + Configured = 0, + + /// Discovered from cluster topology; prunable when the topology stops listing it. + ClusterTopology, + + /// + /// Discovered from sentinel. Cluster absence must never count against it - in a sentinel deployment + /// there is no cluster topology at all, so a single rule would prune the entire thing. + /// + Sentinel, + + /// + /// Learned from a redirect, so it is legitimately ahead of the topology; initial absence is expected + /// rather than evidence. + /// + Redirect, } internal sealed partial class ServerEndPoint : IDisposable @@ -46,10 +74,17 @@ internal void ResetNonConnected() subscription?.ResetNonConnected(); } - public ServerEndPoint(ConnectionMultiplexer multiplexer, EndPoint endpoint) + public ServerEndPoint(ConnectionMultiplexer multiplexer, EndPoint endpoint, ServerProvenance provenance = ServerProvenance.Configured) { Multiplexer = multiplexer; EndPoint = endpoint; + // both collections, deliberately: ResolveDns rewrites the multiplexer's working set at startup, + // replacing configured DnsEndPoints with the addresses they resolved to, while RawConfig keeps the + // original names. Testing only RawConfig would classify a *configured* endpoint as discovered - and + // therefore prunable - whenever ResolveDns is enabled + Provenance = multiplexer.RawConfig.EndPoints.Contains(endpoint) || multiplexer.EndPoints.Contains(endpoint) + ? ServerProvenance.Configured + : provenance; var config = multiplexer.RawConfig; version = config.DefaultVersion; replicaReadOnly = true; @@ -232,6 +267,116 @@ public int WriteEverySeconds internal ConnectionMultiplexer Multiplexer { get; } + /// + /// Whether this server has been disposed; a retired server must not be handed out again. + /// + internal bool IsDisposed => isDisposed; + + /// How we learned of this server; see . + internal ServerProvenance Provenance { get; private set; } + + /// + /// The topology generation in which this server was last listed, or -1 if it never has been. + /// + internal int LastSeenGeneration { get; private set; } = -1; + + /// + /// The generation in which this server first went missing from the topology, or -1 if present. + /// + internal int AbsentSinceGeneration { get; private set; } = -1; + + /// + /// Note that the topology still lists this server, clearing any accrued absence. + /// + internal void OnSeenInTopology(int generation) + { + LastSeenGeneration = generation; + AbsentSinceGeneration = -1; + + // a node first learned from a redirect is confirmed by the topology, so it stops being a special + // case; configured endpoints keep their provenance, since nothing may prune them + if (Provenance == ServerProvenance.Redirect) Provenance = ServerProvenance.ClusterTopology; + } + + /// + /// Note that the topology did not list this server. Returns the number of consecutive generations it + /// has now been missing for, counting this one. + /// + /// + /// The design notes proposed also resetting this whenever the server had been *used* since the last + /// absence, on the grounds that "recently useful" is stronger evidence than "not listed". That is not + /// implementable as stated and turns out to be unnecessary: the only usage counter available + /// (PhysicalBridge.IncrementOpCount) is incremented by our own heartbeat pings as well as by + /// callers, so an idle-but-connected server never looks unused - and every case it was meant to + /// protect is already covered by , since a server actually carrying traffic owns + /// slots in the map. What remains uncovered is a server used only via GetServer by hand while + /// absent from the topology, and pruning that is consistent with the endpoint collection being a + /// snapshot. + /// + internal int OnMissingFromTopology(int generation) + { + if (AbsentSinceGeneration < 0) + { + AbsentSinceGeneration = generation; + return 1; + } + return generation - AbsentSinceGeneration + 1; + } + + /// + /// Whether retiring this server would abandon anything: slots it owns, subscriptions it carries, or + /// work it still owes. + /// + internal bool IsIdle() + => !Multiplexer.ServerSelectionStrategy.OwnsAnySlot(this) + && (subscription?.SubscriptionCount ?? 0) == 0 + && (interactive?.SubscriptionCount ?? 0) == 0 + && GetOutstandingCount() == 0; + + /// + /// Work this server still owes an answer on: written-and-awaiting-response, plus anything queued in + /// the backlog. Zero means a retirement can complete without abandoning anyone. + /// + internal int GetOutstandingCount() + { + var counters = GetCounters(); + return counters.Interactive.SentItemsAwaitingResponse + counters.Interactive.PendingUnsentItems + + counters.Subscription.SentItemsAwaitingResponse + counters.Subscription.PendingUnsentItems; + } + + /// + /// Retire this server gracefully: stop accepting new work, let what has already been written complete, + /// then tear the connections down. Distinct from , which is the abrupt path and + /// abandons anything outstanding. + /// + /// Why this server is being retired; for logging. + /// How long to allow the drain before closing regardless. + /// Optional logger. + internal async Task RetireAsync(string reason, TimeSpan drainTimeout, ILogger? log = null) + { + if (isDisposed) return; + + // stop being selected *first*, so the drain is bounded: nothing new arrives while we wait + SetUnselectable(UnselectableFlags.Retiring); + log?.LogInformationRetiringServer(new(EndPoint), reason); + + // Stopwatch rather than TickCount64: the latter does not exist on the down-level targets + var watch = ValueStopwatch.StartNew(); + int outstanding; + while ((outstanding = GetOutstandingCount()) > 0 && watch.ElapsedMilliseconds < drainTimeout.TotalMilliseconds) + { + await Task.Delay(TimeSpan.FromMilliseconds(20)).ForAwait(); + } + + if (outstanding > 0) + { + // deliberately reported: an abandoned command is exactly what a caller will be asking about + log?.LogInformationRetiringServerAbandoned(new(EndPoint), outstanding); + } + + Dispose(); + } + public void Dispose() { isDisposed = true; @@ -336,6 +481,10 @@ internal void SetClusterSlots(ClusterSlotsResult? slots) { ClusterTopology = topology; Multiplexer.Trace($"Shadow topology: {topology.Nodes.Count} nodes"); + + // not routing on this yet, but the identities are useful immediately: they let a node reached + // by its other name resolve to the server we already have, rather than becoming a duplicate + Multiplexer.RegisterServerIdentities(topology); } } @@ -346,7 +495,17 @@ public void SetClusterConfiguration(ClusterConfiguration configuration) if (configuration != null) { Multiplexer.Trace("Updating cluster ranges..."); - Multiplexer.UpdateClusterRange(configuration); + + // the SLOTS view drives the slot map when this server supplied one; NODES remains the source + // for node relations below, and for anything SLOTS does not report + if (ClusterTopology is { } topology) + { + Multiplexer.UpdateClusterRange(topology); + } + else + { + Multiplexer.UpdateClusterRange(configuration); + } Multiplexer.Trace("Resolving genealogy..."); UpdateNodeRelations(configuration); Multiplexer.Trace("Cluster configured"); @@ -517,21 +676,18 @@ internal async Task AutoConfigureAsync(PhysicalConnection? connection, ILogger? } if (commandMap.IsAvailable(RedisCommand.CLUSTER)) { + // SLOTS first, deliberately: replies arrive in request order, so this is processed before + // the NODES reply below, and the identities it carries are therefore known before NODES + // starts creating servers by address. Without that ordering, a node created from a redirect + // under its announced hostname is duplicated under its address moments later by its own + // autoconfigure. Costs nothing: the burst stays a single pipeline with no round-trip stall + msg = RedisServer.GetClusterSlotsMessage(flags); + msg.SetInternalCall(); + await WriteDirectOrQueueFireAndForgetAsync(connection, msg, ResultProcessor.ClusterSlots).ForAwait(); + msg = RedisServer.GetClusterNodesMessage(flags); msg.SetInternalCall(); await WriteDirectOrQueueFireAndForgetAsync(connection, msg, ResultProcessor.ClusterNodes).ForAwait(); - - // CLUSTER SLOTS would go here - it is the view that conveys naming preference and node ids, - // and ClusterTopology/SetClusterSlots exist ready for it. Deliberately *not* invoked yet: - // this PR is scoped to work that cannot destabilise a connection, and asking every server for - // an extra command on every autoconfigure is a new failure surface on the connect path (an - // unexpected error reply to an internal call, a proxy that mangles the command) for no - // user-visible benefit until routing actually consumes it. Enabled in the follow-up, where - // ordering matters too: it must precede CLUSTER NODES so identities are known before NODES - // creates servers by address. - //// msg = Message.Create(-1, flags, RedisCommand.CLUSTER, RedisLiterals.SLOTS); - //// msg.SetInternalCall(); - //// await WriteDirectOrQueueFireAndForgetAsync(connection, msg, ResultProcessor.ClusterSlots).ForAwait(); } // 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 diff --git a/src/StackExchange.Redis/ServerSelectionStrategy.cs b/src/StackExchange.Redis/ServerSelectionStrategy.cs index 80c9b9efe..16a7a2334 100644 --- a/src/StackExchange.Redis/ServerSelectionStrategy.cs +++ b/src/StackExchange.Redis/ServerSelectionStrategy.cs @@ -204,7 +204,9 @@ public bool TryResend(int hashSlot, Message message, EndPoint endpoint, bool isM { if ((ServerType == ServerType.Standalone && !isSelf) || hashSlot < 0 || hashSlot >= RedisClusterSlotCount) return false; - ServerEndPoint? server = multiplexer?.GetServerEndPoint(endpoint); + // a redirect target is legitimately ahead of the topology, so mark it as such: it must not be + // pruned merely for being absent from a topology reply that predates it + ServerEndPoint? server = multiplexer?.GetServerEndPoint(endpoint, provenance: ServerProvenance.Redirect); if (server != null) { bool retry = false; @@ -294,6 +296,21 @@ internal int CombineSlot(int oldSlot, RedisKey[] keys) return oldSlot; } + /// + /// Whether currently owns any slot in the map, i.e. whether retiring it + /// would leave part of the keyspace unroutable. + /// + internal bool OwnsAnySlot(ServerEndPoint server) + { + var arr = map; + if (arr is null) return false; + for (int i = 0; i < arr.Length; i++) + { + if (ReferenceEquals(arr[i], server)) return true; + } + return false; + } + internal int CountCoveredSlots() { var arr = map; diff --git a/tests/StackExchange.Redis.Tests/ClusterTopologyShadowUnitTests.cs b/tests/StackExchange.Redis.Tests/ClusterTopologyUnitTests.cs similarity index 51% rename from tests/StackExchange.Redis.Tests/ClusterTopologyShadowUnitTests.cs rename to tests/StackExchange.Redis.Tests/ClusterTopologyUnitTests.cs index 6249c987e..dae5641e9 100644 --- a/tests/StackExchange.Redis.Tests/ClusterTopologyShadowUnitTests.cs +++ b/tests/StackExchange.Redis.Tests/ClusterTopologyUnitTests.cs @@ -7,12 +7,11 @@ namespace StackExchange.Redis.Tests; /// -/// The id-keyed CLUSTER SLOTS topology is currently populated *alongside* the CLUSTER NODES -/// view that drives routing, so that the two can be compared before anything depends on the new one. These -/// are the comparison: they assert the shadow view agrees with what routing actually uses, and that it -/// unifies identities where the old view cannot. +/// The id-keyed CLUSTER SLOTS topology, which now drives the slot map. The agreement tests against the +/// CLUSTER NODES view are retained deliberately: NODES is no longer what routes, but it remains +/// the public admin surface, and the two disagreeing would mean one of them is wrong. /// -public class ClusterTopologyShadowUnitTests(ITestOutputHelper log) +public class ClusterTopologyUnitTests(ITestOutputHelper log) { private const string Hostname = "host-1.redis.example.com"; @@ -27,9 +26,9 @@ private static InProcessTestServer CreateServer( } /// - /// Builds the view from an explicit CLUSTER SLOTS call. Autoconfigure does not ask for it yet - see - /// the comment in ServerEndPoint.AutoConfigureAsync - so these exercise the model and the parser - /// rather than the wiring; the wiring is covered where it is enabled. + /// Builds the view from an explicit CLUSTER SLOTS call, so that most of these exercise the model + /// and the parser independently of the autoconfigure wiring; + /// covers the wiring itself. /// private static async Task GetShadowAsync(IConnectionMultiplexer conn, EndPoint endpoint) { @@ -39,6 +38,30 @@ private static async Task GetShadowAsync(IConnectionMultiplexer return topology; } + [Theory] + [InlineData(ClusterEndpointType.Ip)] + [InlineData(ClusterEndpointType.Hostname)] + public async Task AutoConfigurePopulatesTheTopology(ClusterEndpointType preferred) + { + // the wiring, as opposed to the model: connecting is enough, because autoconfigure asks for + // CLUSTER SLOTS as part of its pipelined burst + using var server = CreateServer(log, preferred); + await using var conn = await server.ConnectAsync(defaultOnly: true); + + var endpoint = ((IInternalConnectionMultiplexer)conn).GetServerEndPoint(server.DefaultEndPoint); + var topology = endpoint.ClusterTopology; + Assert.NotNull(topology); + + var node = Assert.Single(topology.Nodes); + log.WriteLine(node.ToString()); + Assert.False(string.IsNullOrEmpty(node.NodeId)); + + // and both identities are known, which is what routing will later rely on + GetHost(server.DefaultEndPoint, out var port); + Assert.Contains(new IPEndPoint(IPAddress.Loopback, port), node.Identities); + Assert.Contains(new DnsEndPoint(Hostname, port), node.Identities); + } + [Theory] [InlineData(ClusterEndpointType.Ip)] [InlineData(ClusterEndpointType.Hostname)] @@ -99,6 +122,8 @@ public async Task ShadowTopologyAgreesWithTheRoutingView() var fromShadow = topology.Nodes.Where(x => !x.IsReplica) .Select(x => x.NodeId).OrderBy(x => x).ToArray(); + EndpointResolutionUnitTests.AssertOneEndpointPerNode(conn, log); + log.WriteLine($"NODES: {string.Join(",", fromNodes)}"); log.WriteLine($"SHADOW: {string.Join(",", fromShadow)}"); Assert.Equal(fromNodes, fromShadow); @@ -119,6 +144,98 @@ static int[] Slots(System.Collections.Generic.IEnumerable ranges) => ranges.SelectMany(r => Enumerable.Range(r.From, r.To - r.From + 1)).OrderBy(x => x).ToArray(); } + [Fact] + public async Task SlotMapIsDrivenByTheSlotsView() + { + // proves the flip took effect rather than the two views merely agreeing: the toy server reports a + // slot as migrated in SLOTS *only*, so routing can only be correct if the SLOTS view is what feeds + // ServerSelectionStrategy + using var server = CreateServer(log, announceHostname: false); + GetHost(server.DefaultEndPoint, out var port); + var other = server.AddEmptyNode(new IPEndPoint(IPAddress.Loopback, port + 1)); + server.Migrate((RedisKey)"slot-map-key", other); + + await using var conn = await server.ConnectAsync(); + + // NoRedirect: if the slot map is right, this lands on the owner first time and needs no redirect + var db = conn.GetDatabase(); + await db.StringSetAsync("slot-map-key", "value", flags: CommandFlags.NoRedirect); + Assert.Equal("value", await db.StringGetAsync("slot-map-key", CommandFlags.NoRedirect)); + + // ...and the command went to the node that owns the slot + var owner = conn.GetServer(new IPEndPoint(IPAddress.Loopback, port + 1)); + log.WriteLine($"owner: {owner.EndPoint}"); + Assert.Equal(new IPEndPoint(IPAddress.Loopback, port + 1), owner.EndPoint); + + EndpointResolutionUnitTests.AssertOneEndpointPerNode(conn, log); + } + + [Fact] + public async Task HostnamePreferredClusterRoutesWithoutDuplicatingEndpoints() + { + // the case the flip exists for: SLOTS names every node by hostname while NODES names them by + // address, so a slot map fed from SLOTS must still resolve to the servers we already hold + using var server = CreateServer(log, ClusterEndpointType.Hostname); + GetHost(server.DefaultEndPoint, out var port); + var other = server.AddEmptyNode(new IPEndPoint(IPAddress.Loopback, port + 1)); + server.SetHostname(other, "host-2.redis.example.com"); + server.Migrate((RedisKey)"hostname-routed-key", other); + + await using var conn = await server.ConnectAsync(); + var db = conn.GetDatabase(); + await db.StringSetAsync("hostname-routed-key", "value"); + Assert.Equal("value", await db.StringGetAsync("hostname-routed-key")); + + foreach (var ep in conn.GetEndPoints()) + { + log.WriteLine($"endpoint: {ep}"); + } + EndpointResolutionUnitTests.AssertOneEndpointPerNode(conn, log); + + // two nodes, two endpoints - not four + Assert.Equal(2, conn.GetEndPoints().Length); + } + + [Fact] + public async Task SlotLessNodesAreKnownButNotConnected() + { + // CLUSTER SLOTS does not list a node that serves nothing, so NODES contributes it - registered but + // not dialled, since there is nothing to route to it. It stays addressable, and first use connects it + using var server = CreateServer(log, announceHostname: false); + var idle = server.AddEmptyNode(); // no slots + + await using var conn = await server.ConnectAsync(defaultOnly: true); + await conn.GetServer(server.DefaultEndPoint).PingAsync(); // force a reconfigure pass + + foreach (var ep in conn.GetEndPoints()) + { + log.WriteLine($"endpoint: {ep}"); + } + Assert.Contains(idle, conn.GetEndPoints()); + + var api = conn.GetServer(idle); + Assert.False(api.IsConnected); // known, but no bridge was created for it + + // ...and using it activates it, so nothing is lost by not dialling eagerly + await api.PingAsync(); + Assert.True(api.IsConnected); + } + + [Fact] + public async Task SlotServingNodesAreConnectedEagerly() + { + // the counterpart: a node that owns slots is in the SLOTS view and so is connected as before + using var server = CreateServer(log, announceHostname: false); + GetHost(server.DefaultEndPoint, out var port); + var owner = server.AddEmptyNode(new IPEndPoint(IPAddress.Loopback, port + 1)); + server.Migrate((RedisKey)"eager-key", owner); + + await using var conn = await server.ConnectAsync(); + await conn.GetServer(server.DefaultEndPoint).PingAsync(); + + Assert.True(conn.GetServer(owner).IsConnected); + } + [Fact] public async Task ShadowTopologyDoesNotChangeRouting() { diff --git a/tests/StackExchange.Redis.Tests/EndpointPruningUnitTests.cs b/tests/StackExchange.Redis.Tests/EndpointPruningUnitTests.cs new file mode 100644 index 000000000..0d558a995 --- /dev/null +++ b/tests/StackExchange.Redis.Tests/EndpointPruningUnitTests.cs @@ -0,0 +1,222 @@ +using System.Linq; +using System.Net; +using System.Threading.Tasks; +using Xunit; +using static StackExchange.Redis.Server.RedisServer; + +namespace StackExchange.Redis.Tests; + +/// +/// Ageing endpoints out of the topology, and merging a node that arrived under two names. Both are policies +/// over the retirement primitive covered by . +/// +public class EndpointPruningUnitTests(ITestOutputHelper log) +{ + private const string Hostname = "host-1.redis.example.com"; + + private static InProcessTestServer CreateServer(ITestOutputHelper log, ClusterEndpointType preferred = ClusterEndpointType.Ip) + => new(log) { ServerType = ServerType.Cluster, PreferredEndpointType = preferred }; + + /// + /// Applies the topology generations that pruning is measured in - slot map *and* generation, as the + /// production path does. Applying only the generation would leave the slot map pointing at a node the + /// topology no longer lists, so it would never look idle and could never be pruned. + /// + private static async Task ApplyGenerationsAsync(IConnectionMultiplexer conn, EndPoint askOf, int count) + { + var mux = (ConnectionMultiplexer)conn; + for (int i = 0; i < count; i++) + { + var slots = await conn.GetServer(askOf).ClusterSlotsAsync(); + var topology = ClusterTopology.From(slots); + Assert.NotNull(topology); + mux.UpdateClusterRange(topology); + await mux.ApplyTopologyGenerationAsync(topology); + } + } + + [Fact] + public async Task NodeAbsentFromTopologyIsPrunedAfterThreeGenerations() + { + using var server = CreateServer(log); + GetHost(server.DefaultEndPoint, out var port); + var doomed = server.AddEmptyNode(new IPEndPoint(IPAddress.Loopback, port + 1)); + server.Migrate((RedisKey)"prune-key", doomed); + + // defaultOnly, so the doomed node is *discovered* rather than configured - a configured endpoint is + // exempt by design, and connecting to every toy node would make this test vacuous + await using var conn = await server.ConnectAsync(defaultOnly: true); + Assert.Contains(doomed, conn.GetEndPoints()); + + // hand its slot back, so the topology stops listing it - and it owns nothing, so it is prunable + server.Migrate((RedisKey)"prune-key", server.DefaultEndPoint); + + await ApplyGenerationsAsync(conn, server.DefaultEndPoint, 2); + Assert.Contains(doomed, conn.GetEndPoints()); // two absences is not yet evidence + + await ApplyGenerationsAsync(conn, server.DefaultEndPoint, 1); + log.WriteLine(string.Join(", ", conn.GetEndPoints().Select(x => x.ToString()))); + Assert.DoesNotContain(doomed, conn.GetEndPoints()); + } + + [Fact] + public async Task ConfiguredEndpointIsNeverPruned() + { + // the seed we need to bootstrap after a full rotation; absence must never remove it + using var server = CreateServer(log); + GetHost(server.DefaultEndPoint, out var port); + var second = server.AddEmptyNode(new IPEndPoint(IPAddress.Loopback, port + 1)); + + var config = server.GetClientConfig(); + Assert.Contains(second, config.EndPoints); // explicitly configured, and serves no slots + + await using var conn = await ConnectionMultiplexer.ConnectAsync(config); + await ApplyGenerationsAsync(conn, server.DefaultEndPoint, 5); + + Assert.Contains(second, conn.GetEndPoints()); + } + + [Fact] + public async Task NodeCarryingSubscriptionsIsNotPruned() + { + // "not idle" is what protects a server that is still doing something for someone - see the remarks on + // OnMissingFromTopology for why this is the test rather than a use-recency one + using var server = CreateServer(log); + GetHost(server.DefaultEndPoint, out var port); + var subscribed = server.AddEmptyNode(new IPEndPoint(IPAddress.Loopback, port + 1)); + server.Migrate((RedisKey)"sub-slot-key", subscribed); + + await using var conn = await server.ConnectAsync(); + var sub = conn.GetSubscriber(); + await sub.SubscribeAsync(RedisChannel.Literal(nameof(NodeCarryingSubscriptionsIsNotPruned)), (_, _) => { }); + + server.Migrate((RedisKey)"sub-slot-key", server.DefaultEndPoint); // absent from the topology now + await ApplyGenerationsAsync(conn, server.DefaultEndPoint, 5); + + // whichever server carries the subscription must survive; the other may legitimately go + var survivors = conn.GetEndPoints(); + log.WriteLine(string.Join(", ", survivors.Select(x => x.ToString()))); + Assert.Contains(server.DefaultEndPoint, survivors); + } + + [Fact] + public async Task NodeStillOwningSlotsIsNotPruned() + { + // belt and braces: even if it somehow went missing from a reply, retiring a slot owner would leave + // part of the keyspace unroutable + using var server = CreateServer(log); + GetHost(server.DefaultEndPoint, out var port); + var owner = server.AddEmptyNode(new IPEndPoint(IPAddress.Loopback, port + 1)); + server.Migrate((RedisKey)"owned-key", owner); + + await using var conn = await server.ConnectAsync(); + await ApplyGenerationsAsync(conn, server.DefaultEndPoint, 5); + + Assert.Contains(owner, conn.GetEndPoints()); + Assert.Equal("value", await Set(conn)); + + static async Task Set(IConnectionMultiplexer conn) + { + var db = conn.GetDatabase(); + await db.StringSetAsync("owned-key", "value"); + return await db.StringGetAsync("owned-key"); + } + } + + [Fact] + public async Task SentinelDiscoveredServerIsNotPrunedByClusterAbsence() + { + // the catastrophic case a single rule would produce: in a sentinel deployment no cluster topology runs + // at all, so everything looks absent. Provenance is what prevents it + using var server = CreateServer(log); + GetHost(server.DefaultEndPoint, out var port); + + // deliberately *not* a node of this cluster: sentinel knows of it, the cluster topology does not, + // which is precisely the shape that a single prune rule would destroy + var viaSentinel = new IPEndPoint(IPAddress.Loopback, port + 500); + + await using var conn = await server.ConnectAsync(defaultOnly: true); + var mux = (ConnectionMultiplexer)conn; + + // register it as sentinel-discovered, exactly as the sentinel path does + var sentinelServer = mux.GetServerEndPoint(viaSentinel, activate: false, provenance: ServerProvenance.Sentinel); + Assert.Equal(ServerProvenance.Sentinel, sentinelServer.Provenance); + + await ApplyGenerationsAsync(conn, server.DefaultEndPoint, 5); + Assert.Contains(viaSentinel, conn.GetEndPoints()); + } + + [Fact] + public async Task ConfiguredEndpointSurvivesResolveDns() + { + // regression: ResolveDns rewrites the working set at startup, replacing configured names with the + // addresses they resolved to, while the configuration keeps the names. Testing only the configuration + // for "was this configured?" classifies a configured endpoint as discovered - and so prunable + using var server = CreateServer(log); + var config = server.GetClientConfig(defaultOnly: true); + config.ResolveDns = true; + + await using var conn = await ConnectionMultiplexer.ConnectAsync(config); + var endpoint = conn.GetEndPoints().Single(); + + var sep = ((IInternalConnectionMultiplexer)conn).GetServerEndPoint(endpoint); + log.WriteLine($"{endpoint} provenance={sep.Provenance}"); + Assert.Equal(ServerProvenance.Configured, sep.Provenance); + + await ApplyGenerationsAsync(conn, endpoint, 5); + Assert.Contains(endpoint, conn.GetEndPoints()); + } + + [Fact] + public async Task NewNodeIsDialledByTheAdvertisedForm() + { + // a certificate validates against a name, and where hostnames are preferred the advertised address may + // not be routable at all - so a node we have never seen is dialled by the form the answering node + // advertised, not by an address we happen to have been given alongside it + using var server = CreateServer(log, ClusterEndpointType.Hostname); + GetHost(server.DefaultEndPoint, out var port); + var newcomer = server.AddEmptyNode(new IPEndPoint(IPAddress.Loopback, port + 1)); + server.SetHostname(newcomer, "host-2.redis.example.com"); + server.Migrate((RedisKey)"advertised-key", newcomer); + + await using var conn = await server.ConnectAsync(defaultOnly: true); + await ApplyGenerationsAsync(conn, server.DefaultEndPoint, 1); + + foreach (var ep in conn.GetEndPoints()) + { + log.WriteLine($"endpoint: {ep}"); + } + + // reached by name, since that is what this cluster advertises + Assert.Contains(conn.GetEndPoints(), ep => ep is DnsEndPoint { Host: "host-2.redis.example.com" }); + Assert.DoesNotContain(conn.GetEndPoints(), ep => ep is IPEndPoint { Port: var p } && p == port + 1); + } + + [Fact] + public async Task DuplicateUnderTwoNamesIsMergedIntoOne() + { + // one node, two ServerEndPoints - what happens when something creates by a name we had not yet linked. + // The merge must retire one and leave the retired name resolving to the survivor + using var server = CreateServer(log, ClusterEndpointType.Ip); + GetHost(server.DefaultEndPoint, out var port); + + // connect *before* the hostname is announced, so no alias for it is registered yet - which is the + // state in which something can create a second server for a node we already hold + await using var conn = await server.ConnectAsync(defaultOnly: true); + var mux = (ConnectionMultiplexer)conn; + + server.SetHostname(server.DefaultEndPoint, Hostname); + var byName = new DnsEndPoint(Hostname, port); + mux.GetServerEndPoint(byName, activate: false); + Assert.Equal(2, conn.GetEndPoints().Length); // one node, two endpoints + + await ApplyGenerationsAsync(conn, server.DefaultEndPoint, 1); + + log.WriteLine(string.Join(", ", conn.GetEndPoints().Select(x => x.ToString()))); + Assert.Single(conn.GetEndPoints()); + + // ...and the retired name still resolves, so a caller holding it is not broken + Assert.NotNull(conn.GetServer(byName)); + Assert.Equal(server.DefaultEndPoint, conn.GetServer(byName).EndPoint); + } +} diff --git a/tests/StackExchange.Redis.Tests/EndpointResolutionUnitTests.cs b/tests/StackExchange.Redis.Tests/EndpointResolutionUnitTests.cs new file mode 100644 index 000000000..c2f38700a --- /dev/null +++ b/tests/StackExchange.Redis.Tests/EndpointResolutionUnitTests.cs @@ -0,0 +1,155 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Net; +using System.Threading.Tasks; +using Xunit; +using static StackExchange.Redis.Server.RedisServer; + +namespace StackExchange.Redis.Tests; + +/// +/// Reaching one node by any of the names it answers to. servers is keyed on exact endpoint equality, +/// so before this an address-keyed node was simply unreachable by its announced hostname - and a redirect +/// naming it that way became a second for the same node (#2826). +/// +public class EndpointResolutionUnitTests(ITestOutputHelper log) +{ + private const string Hostname = "host-1.redis.example.com"; + + private static InProcessTestServer CreateServer(ITestOutputHelper log, ClusterEndpointType preferred = ClusterEndpointType.Ip) + { + var server = new InProcessTestServer(log) { ServerType = ServerType.Cluster, PreferredEndpointType = preferred }; + server.SetHostname(server.DefaultEndPoint, Hostname); + return server; + } + + [Theory] + [InlineData(ClusterEndpointType.Ip)] + [InlineData(ClusterEndpointType.Hostname)] + public async Task ServerResolvesByEitherIdentity(ClusterEndpointType preferred) + { + using var server = CreateServer(log, preferred); + await using var conn = await server.ConnectAsync(defaultOnly: true); + + var canonical = server.DefaultEndPoint; + GetHost(canonical, out var port); + var byName = new DnsEndPoint(Hostname, port); + + // the multiplexer is keyed on the address it was configured with... + Assert.Equal(canonical, Assert.Single(conn.GetEndPoints())); + + // ...but the node also answers to its announced hostname, and that now resolves to the same server + var viaAddress = conn.GetServer(canonical); + var viaName = conn.GetServer(byName); + Assert.Equal(viaAddress.EndPoint, viaName.EndPoint); + + // and resolving does not invent an endpoint + Assert.Equal(canonical, Assert.Single(conn.GetEndPoints())); + } + + /// + /// The invariant behind all of this: one node, one endpoint. Asserted from the shadow topology rather than + /// from a scenario, so it holds regardless of the order things were learned in - which is what makes it a + /// useful guard while two sources can still create servers (autoconfigure, and the independent + /// CLUSTER NODES read in ReconfigureAsync). + /// + internal static void AssertOneEndpointPerNode(IConnectionMultiplexer conn, ITestOutputHelper log) + { + var mux = (IInternalConnectionMultiplexer)conn; + var seen = new Dictionary(); + foreach (var endpoint in conn.GetEndPoints()) + { + var topology = mux.GetServerEndPoint(endpoint).ClusterTopology; + if (topology is null) continue; + + foreach (var node in topology.Nodes) + { + // does this endpoint identify this node? + if (!node.Identities.Contains(endpoint)) continue; + + if (seen.TryGetValue(node.NodeId, out var already) && !Equals(already, endpoint)) + { + Assert.Fail($"node {node.NodeId} is held under two endpoints: {already} and {endpoint}"); + } + seen[node.NodeId] = endpoint; + log.WriteLine($"{node.NodeId} <- {endpoint}"); + } + } + } + + [Fact] + public async Task UnknownIdentityStillThrows() + { + using var server = CreateServer(log); + await using var conn = await server.ConnectAsync(defaultOnly: true); + + var ex = Assert.Throws(() => conn.GetServer(new DnsEndPoint("not-this-node.example.com", 6379))); + log.WriteLine(ex.Message); + Assert.Contains("not defined", ex.Message); + } + + [Fact] + public async Task RedirectToANewNodeDoesNotDuplicateIt() + { + // The hazard in full: a hostname-preferring cluster redirects to a node, which then autoconfigures and + // reports itself by *address* via CLUSTER NODES. Left to itself that produces two ServerEndPoints for + // one node - doubled connections, with backlog and subscription state split across the pair. + // + // What prevents it is that autoconfigure asks for CLUSTER SLOTS before CLUSTER NODES: replies arrive + // in request order, so the identities are registered before NODES can create anything by address. + // That ordering is load-bearing - see the comment in ServerEndPoint.AutoConfigureAsync - and this is + // the test that fails if someone reorders the burst. + using var server = CreateServer(log, ClusterEndpointType.Hostname); + GetHost(server.DefaultEndPoint, out var port); + + await using var conn = await server.ConnectAsync(defaultOnly: true); + var before = conn.GetEndPoints().Length; + + var other = server.AddEmptyNode(new IPEndPoint(IPAddress.Loopback, port + 1)); + server.SetHostname(other, "host-2.redis.example.com"); + server.Migrate((RedisKey)"resolution-key", other); + + // follow the redirect; the target is named by hostname, since that is what this cluster prefers + await conn.GetDatabase().StringSetAsync("resolution-key", "value"); + Assert.Equal("value", await conn.GetDatabase().StringGetAsync("resolution-key")); + + foreach (var ep in conn.GetEndPoints()) + { + log.WriteLine($"endpoint: {ep}"); + } + + AssertOneEndpointPerNode(conn, log); + + // one new node, one endpoint for it + var added = conn.GetEndPoints().Length - before; + log.WriteLine($"added {added} endpoint(s) for one node"); + Assert.Equal(1, added); + Assert.Equal(1, conn.GetEndPoints().Count(ep => PortOf(ep) == port + 1)); + + // both names reach a server, which is the part resolution does deliver + Assert.NotNull(conn.GetServer(new IPEndPoint(IPAddress.Loopback, port + 1))); + Assert.NotNull(conn.GetServer(new DnsEndPoint("host-2.redis.example.com", port + 1))); + + static int PortOf(EndPoint ep) => ep switch + { + IPEndPoint ip => ip.Port, + DnsEndPoint dns => dns.Port, + _ => 0, + }; + } + + [Fact] + public async Task ResolutionSurvivesReconnect() + { + // identities are learned from topology, so they must be re-registered when it is re-read + using var server = CreateServer(log, ClusterEndpointType.Hostname); + GetHost(server.DefaultEndPoint, out var port); + + await using var conn = await server.ConnectAsync(defaultOnly: true); + Assert.NotNull(conn.GetServer(new DnsEndPoint(Hostname, port))); + + await conn.GetServer(server.DefaultEndPoint).PingAsync(); + Assert.NotNull(conn.GetServer(new DnsEndPoint(Hostname, port))); + } +} diff --git a/tests/StackExchange.Redis.Tests/ServerRetirementUnitTests.cs b/tests/StackExchange.Redis.Tests/ServerRetirementUnitTests.cs new file mode 100644 index 000000000..c7b334d00 --- /dev/null +++ b/tests/StackExchange.Redis.Tests/ServerRetirementUnitTests.cs @@ -0,0 +1,152 @@ +using System; +using System.Linq; +using System.Net; +using System.Threading.Tasks; +using Xunit; +using static StackExchange.Redis.Server.RedisServer; + +namespace StackExchange.Redis.Tests; + +/// +/// The retirement primitive: drain, then close, then forget. Driven directly through internals here - the +/// policies that decide *when* to retire (topology pruning, duplicate merging, and later the maintenance +/// handoffs) all call the same operation, so it is worth pinning on its own. +/// +public class ServerRetirementUnitTests(ITestOutputHelper log) +{ + private const string Hostname = "host-1.redis.example.com"; + + private static InProcessTestServer CreateServer(ITestOutputHelper log) + => new(log) { ServerType = ServerType.Cluster }; + + [Fact] + public async Task RetiredServerIsForgotten() + { + using var server = CreateServer(log); + GetHost(server.DefaultEndPoint, out var port); + var other = server.AddEmptyNode(new IPEndPoint(IPAddress.Loopback, port + 1)); + server.Migrate((RedisKey)"retire-key", other); + + await using var conn = await server.ConnectAsync(); + var mux = (ConnectionMultiplexer)conn; + Assert.Contains(other, conn.GetEndPoints()); + + var target = ((IInternalConnectionMultiplexer)conn).GetServerEndPoint(other); + await mux.RetireServerAsync(target, "test"); + + // gone from the collection, and no longer resolvable + Assert.DoesNotContain(other, conn.GetEndPoints()); + Assert.Throws(() => conn.GetServer(other)); + } + + [Fact] + public async Task RetirementDropsSecondaryIdentities() + { + // the trap: an alias outliving its server would resolve to something disposed + using var server = CreateServer(log); + server.SetHostname(server.DefaultEndPoint, Hostname); + GetHost(server.DefaultEndPoint, out var port); + + await using var conn = await server.ConnectAsync(defaultOnly: true); + var byName = new DnsEndPoint(Hostname, port); + Assert.NotNull(conn.GetServer(byName)); // resolvable via the alias + + var mux = (ConnectionMultiplexer)conn; + var target = ((IInternalConnectionMultiplexer)conn).GetServerEndPoint(server.DefaultEndPoint); + await mux.RetireServerAsync(target, "test"); + + Assert.Throws(() => conn.GetServer(byName)); + Assert.Throws(() => conn.GetServer(server.DefaultEndPoint)); + } + + [Fact] + public async Task RetirementCompletesInFlightWork() + { + // the point of draining rather than disposing: work already written must still get its answer + using var server = CreateServer(log); + GetHost(server.DefaultEndPoint, out var port); + var other = server.AddEmptyNode(new IPEndPoint(IPAddress.Loopback, port + 1)); + server.Migrate((RedisKey)"retire-inflight", other); + + await using var conn = await server.ConnectAsync(); + var db = conn.GetDatabase(); + await db.StringSetAsync("retire-inflight", "value"); + + // issue without awaiting, then retire underneath it + var pending = Enumerable.Range(0, 50) + .Select(_ => db.StringGetAsync("retire-inflight")) + .ToArray(); + + var mux = (ConnectionMultiplexer)conn; + var target = ((IInternalConnectionMultiplexer)conn).GetServerEndPoint(other); + await mux.RetireServerAsync(target, "test"); + + var results = await Task.WhenAll(pending); + Assert.All(results, x => Assert.Equal("value", x)); + log.WriteLine($"{results.Length} operations completed across the retirement"); + } + + [Fact] + public async Task RetiredServerIsNotSelectedForNewWork() + { + using var server = CreateServer(log); + GetHost(server.DefaultEndPoint, out var port); + var other = server.AddEmptyNode(new IPEndPoint(IPAddress.Loopback, port + 1)); + server.Migrate((RedisKey)"retire-select", other); + + await using var conn = await server.ConnectAsync(); + var mux = (ConnectionMultiplexer)conn; + var target = ((IInternalConnectionMultiplexer)conn).GetServerEndPoint(other); + + await mux.RetireServerAsync(target, "test"); + + // the slot it served has no owner now, so this must fail rather than reach a dead server + var ex = await Record.ExceptionAsync(() => conn.GetDatabase().StringGetAsync("retire-select", CommandFlags.NoRedirect)); + log.WriteLine($"{ex?.GetType().Name}: {ex?.Message}"); + Assert.NotNull(ex); + } + + [Fact] + public async Task RetiringTwiceIsHarmless() + { + using var server = CreateServer(log); + GetHost(server.DefaultEndPoint, out var port); + var other = server.AddEmptyNode(new IPEndPoint(IPAddress.Loopback, port + 1)); + server.Migrate((RedisKey)"retire-twice", other); + + await using var conn = await server.ConnectAsync(); + var mux = (ConnectionMultiplexer)conn; + var target = ((IInternalConnectionMultiplexer)conn).GetServerEndPoint(other); + + await mux.RetireServerAsync(target, "first"); + await mux.RetireServerAsync(target, "second"); // must not throw, must not resurrect + Assert.DoesNotContain(other, conn.GetEndPoints()); + } + + [Fact] + public async Task SnapshotRemovalLeavesOtherServersIntact() + { + // ServerSnapshot.Remove has to copy rather than compact in place; this is the shape that would break + using var server = CreateServer(log); + GetHost(server.DefaultEndPoint, out var port); + var second = server.AddEmptyNode(new IPEndPoint(IPAddress.Loopback, port + 1)); + var third = server.AddEmptyNode(new IPEndPoint(IPAddress.Loopback, port + 2)); + server.Migrate((RedisKey)"snap-a", second); + server.Migrate((RedisKey)"snap-b", third); + + await using var conn = await server.ConnectAsync(); + Assert.Equal(3, conn.GetEndPoints().Length); + + var mux = (ConnectionMultiplexer)conn; + await mux.RetireServerAsync(((IInternalConnectionMultiplexer)conn).GetServerEndPoint(second), "test"); + + var remaining = conn.GetEndPoints(); + log.WriteLine(string.Join(", ", remaining.Select(x => x.ToString()))); + Assert.Equal(2, remaining.Length); + Assert.Contains(server.DefaultEndPoint, remaining); + Assert.Contains(third, remaining); + + // and the survivors still work + Assert.True(conn.GetServer(third).IsConnected); + } +}