From c1c45820bf68b2eb68848c1d2656295a6cfa6e92 Mon Sep 17 00:00:00 2001 From: Marc Gravell Date: Fri, 14 Aug 2026 13:38:41 +0100 Subject: [PATCH 1/6] Resolve servers by any identity a node answers to servers is keyed on exact endpoint equality, so a node held under its address was unreachable by its announced hostname - and a redirect naming it that way became a second ServerEndPoint for one node, doubling connections and splitting backlog and subscription state across the pair (#2826). - _serverIdentities maps secondary names to the one ServerEndPoint, kept separate from the servers table so exact keying is untouched - TryResolveServerEndPoint: exact hit, then identity. Used by the public GetServer, which previously threw for a node known by its other name, and by the internal GetServerEndPoint, which resolves before creating - RegisterServerIdentities records the other names of nodes we already know, and deliberately creates nothing: an unheard-of node is discovery's business, not identity's Autoconfigure now asks for CLUSTER SLOTS *before* CLUSTER NODES. Replies arrive in request order, so the identities are registered before NODES can create anything by address; without it, a node created from a redirect under its hostname was duplicated under its address moments later by its own autoconfigure, which is exactly what the redirect test demonstrated. Costs nothing - the burst remains a single pipeline with no round-trip stall - but the ordering is now load-bearing, and commented as such in both places. Also adds AssertOneEndpointPerNode: an invariant check over the shadow topology - no node-id may be held under two endpoints - asserted from state rather than from a scenario, so it catches duplication however it arose. That matters because two creators still exist structurally: ReconfigureAsync reads CLUSTER NODES independently of autoconfigure, and that window is not deterministically reproducible. --- .../ClusterSlots.Server.cs | 4 +- .../ConnectionMultiplexer.cs | 48 +++++- src/StackExchange.Redis/RedisServer.cs | 8 + src/StackExchange.Redis/ServerEndPoint.cs | 25 +-- .../ClusterTopologyShadowUnitTests.cs | 2 + .../EndpointResolutionUnitTests.cs | 155 ++++++++++++++++++ 6 files changed, 226 insertions(+), 16 deletions(-) create mode 100644 tests/StackExchange.Redis.Tests/EndpointResolutionUnitTests.cs 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.cs b/src/StackExchange.Redis/ConnectionMultiplexer.cs index a585c4437..4fded1bc9 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; @@ -924,11 +930,49 @@ 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; + return _serverIdentities.TryGetValue(endpoint, out var byIdentity) ? byIdentity : null; + } + + /// + /// 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) { if (endpoint == null) return null; - var server = (ServerEndPoint?)servers[endpoint]; + var server = (ServerEndPoint?)servers[endpoint] ?? TryResolveServerEndPoint(endpoint); if (server == null) { bool isNew = false; @@ -1277,7 +1321,7 @@ public IServer GetServer(EndPoint? endpoint, object? asyncState = null) { throw new NotSupportedException($"The server API is not available via {RawConfig.Proxy}"); } - var server = servers[endpoint] as ServerEndPoint ?? throw new ArgumentException("The specified endpoint is not defined", nameof(endpoint)); + var server = TryResolveServerEndPoint(endpoint) ?? throw new ArgumentException("The specified endpoint is not defined", nameof(endpoint)); return server.GetRedisServer(asyncState); } 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..114a110a1 100644 --- a/src/StackExchange.Redis/ServerEndPoint.cs +++ b/src/StackExchange.Redis/ServerEndPoint.cs @@ -336,6 +336,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); } } @@ -517,21 +521,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/tests/StackExchange.Redis.Tests/ClusterTopologyShadowUnitTests.cs b/tests/StackExchange.Redis.Tests/ClusterTopologyShadowUnitTests.cs index 6249c987e..a131f53df 100644 --- a/tests/StackExchange.Redis.Tests/ClusterTopologyShadowUnitTests.cs +++ b/tests/StackExchange.Redis.Tests/ClusterTopologyShadowUnitTests.cs @@ -99,6 +99,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); 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))); + } +} From 2e5502843e8f5659b0eed443a42783ad4c467429 Mon Sep 17 00:00:00 2001 From: Marc Gravell Date: Mon, 17 Aug 2026 14:27:04 +0100 Subject: [PATCH 2/6] Drive the slot map from CLUSTER SLOTS Autoconfigure asks for CLUSTER SLOTS (ahead of CLUSTER NODES, per the previous commit), and the resulting id-keyed topology now feeds ServerSelectionStrategy where the answering server supplied one; the NODES-derived path remains as the fallback and still drives node relations. The reason this is worth doing rather than merely equivalent: the SLOTS view carries the node-id and both naming forms, so applying it resolves each node through every identity it answers to before considering the form this particular reply happened to use. The NODES path keyed on one endpoint and so created a second ServerEndPoint whenever a node arrived under its other name. Where a node is genuinely new, an address is preferred over a hostname - the address is dialable as-is, whereas a hostname is only usable if it resolves. Tests prove the flip took effect rather than the two views merely agreeing: a slot is migrated such that routing can only be correct if SLOTS is what feeds the map, and a hostname-preferring cluster (where SLOTS names every node by hostname while NODES names them by address) now routes with two endpoints for two nodes rather than four. Also adds the autoconfigure wiring test, since the existing coverage deliberately sourced the reply explicitly. ClusterTopologyShadowUnitTests renamed: it is no longer shadowing anything. Its agreement-with-NODES assertions are kept on purpose - NODES is no longer what routes, but it is still the public admin surface, and disagreement would mean one of the two is wrong. --- .../ConnectionMultiplexer.cs | 45 +++++++++ src/StackExchange.Redis/ServerEndPoint.cs | 12 ++- ...itTests.cs => ClusterTopologyUnitTests.cs} | 91 +++++++++++++++++-- 3 files changed, 139 insertions(+), 9 deletions(-) rename tests/StackExchange.Redis.Tests/{ClusterTopologyShadowUnitTests.cs => ClusterTopologyUnitTests.cs} (60%) diff --git a/src/StackExchange.Redis/ConnectionMultiplexer.cs b/src/StackExchange.Redis/ConnectionMultiplexer.cs index 4fded1bc9..1ecb15246 100644 --- a/src/StackExchange.Redis/ConnectionMultiplexer.cs +++ b/src/StackExchange.Redis/ConnectionMultiplexer.cs @@ -2038,6 +2038,51 @@ 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: create it under the form this reply used, preferring an address if we were + // given one - a hostname is only usable if it resolves, whereas the address is dialable as-is + foreach (var identity in node.Identities) + { + if (identity is IPEndPoint) return GetServerEndPoint(identity); + } + 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/ServerEndPoint.cs b/src/StackExchange.Redis/ServerEndPoint.cs index 114a110a1..4bd9c32ef 100644 --- a/src/StackExchange.Redis/ServerEndPoint.cs +++ b/src/StackExchange.Redis/ServerEndPoint.cs @@ -350,7 +350,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"); diff --git a/tests/StackExchange.Redis.Tests/ClusterTopologyShadowUnitTests.cs b/tests/StackExchange.Redis.Tests/ClusterTopologyUnitTests.cs similarity index 60% rename from tests/StackExchange.Redis.Tests/ClusterTopologyShadowUnitTests.cs rename to tests/StackExchange.Redis.Tests/ClusterTopologyUnitTests.cs index a131f53df..38194b682 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)] @@ -121,6 +144,58 @@ 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 ShadowTopologyDoesNotChangeRouting() { From d2b2c6487482b5bc8561e28d56b031a63c06ae8d Mon Sep 17 00:00:00 2001 From: Marc Gravell Date: Mon, 17 Aug 2026 14:43:44 +0100 Subject: [PATCH 3/6] Discover cluster endpoints from CLUSTER SLOTS, registering the rest inert Reconfigure now asks both topology commands of the connected server, in one pipelined pair rather than trusting the view cached by autoconfigure: re-reading one half while acting on possibly-stale data for the other would be worse than either choice on its own. CLUSTER SLOTS drives discovery, so the nodes that serve traffic are the ones we connect to, resolved through every identity they answer to first so that a node already held under another name is not duplicated. CLUSTER NODES then contributes what SLOTS cannot: a node serving no slots does not appear in the SLOTS reply at all, so those are registered with activate:false - known, in GetEndPoints(), addressable via GetServer, but not dialled, since there is nothing to route to them. Nothing is lost by that: Activate is only GetBridge(create:true) and GetBridge(Message) creates unconditionally, so the first command sent to such a node connects it then. Sentinel already used activate:false for a new primary, so the pattern is established rather than novel. If SLOTS yields nothing usable - a pre-4.0 server, or an error reply - discovery falls back to exactly the previous behaviour, with every NODES endpoint active. Node relations continue to come from NODES regardless: SLOTS conveys replica-of by position, but not the ids and flags that Primary/Replicas resolution reads. ClusterHandshakeNodesAreIgnored passes unchanged, which was the point of the dual source - an empty node stays reachable, it simply is not dialled. New tests cover both directions: a slot-less node is known but reports IsConnected false until used, and a node that owns slots is still connected eagerly. --- .../ConnectionMultiplexer.cs | 71 +++++++++++++++++-- src/StackExchange.Redis/LoggerExtensions.cs | 6 ++ .../ClusterTopologyUnitTests.cs | 40 +++++++++++ 3 files changed, 110 insertions(+), 7 deletions(-) diff --git a/src/StackExchange.Redis/ConnectionMultiplexer.cs b/src/StackExchange.Redis/ConnectionMultiplexer.cs index 1ecb15246..75673fe1f 100644 --- a/src/StackExchange.Redis/ConnectionMultiplexer.cs +++ b/src/StackExchange.Redis/ConnectionMultiplexer.cs @@ -1846,21 +1846,63 @@ public EndPoint[] GetEndPoints(bool configuredOnly = false) => 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); } return clusterEndpoints; } @@ -1869,6 +1911,21 @@ public EndPoint[] GetEndPoints(bool configuredOnly = false) => log?.LogErrorEncounteredErrorWhileUpdatingClusterConfig(ex, ex.Message); return null; } + + // prefer an identity we already know, then an address, then whatever we were given: an address is + // dialable as-is, whereas a hostname is only usable if it resolves + EndPoint? SelectIdentity(ClusterTopologyNode node) + { + foreach (var identity in node.Identities) + { + if (TryResolveServerEndPoint(identity) is { } known) return known.EndPoint; + } + foreach (var identity in node.Identities) + { + if (identity is IPEndPoint) return identity; + } + return node.Identities.Count > 0 ? node.Identities[0] : null; + } } private void ResetAllNonConnected() diff --git a/src/StackExchange.Redis/LoggerExtensions.cs b/src/StackExchange.Redis/LoggerExtensions.cs index 552a38115..875135adb 100644 --- a/src/StackExchange.Redis/LoggerExtensions.cs +++ b/src/StackExchange.Redis/LoggerExtensions.cs @@ -326,6 +326,12 @@ 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 = 43, diff --git a/tests/StackExchange.Redis.Tests/ClusterTopologyUnitTests.cs b/tests/StackExchange.Redis.Tests/ClusterTopologyUnitTests.cs index 38194b682..dae5641e9 100644 --- a/tests/StackExchange.Redis.Tests/ClusterTopologyUnitTests.cs +++ b/tests/StackExchange.Redis.Tests/ClusterTopologyUnitTests.cs @@ -196,6 +196,46 @@ public async Task HostnamePreferredClusterRoutesWithoutDuplicatingEndpoints() 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() { From eccdcd0930805d06a74e337f69b7e52a9e60b1e4 Mon Sep 17 00:00:00 2001 From: Marc Gravell Date: Mon, 17 Aug 2026 16:06:34 +0100 Subject: [PATCH 4/6] Add a graceful server-retirement primitive Nothing could previously remove a ServerEndPoint: ServerSnapshot had Add and no counterpart, nothing called servers.Remove, and Dispose tears the bridges down immediately - abandoning anything in flight. Four separate requirements want the same drain-then-close (topology pruning, duplicate merging, and both endpoint handoffs in the maintenance-notification work), so it is built once here, with no policy attached and no caller yet. ServerEndPoint.RetireAsync marks itself unselectable *first*, which bounds the drain by ensuring nothing new arrives, then waits for written-and-awaiting plus backlogged work to clear before disposing. Exceeding the drain timeout is logged with the count abandoned, since that is precisely what someone will be asking about later. ConnectionMultiplexer.RetireServerAsync then forgets it, including every secondary identity that pointed at it - without that, an alias outlives its server and TryResolveServerEndPoint hands back something whose bridges are gone. That method now also refuses disposed servers, restoring a check dropped during the cherry-pick because IsDisposed was not exposed. ServerSnapshot.Remove always allocates a compacted copy and never reuses the array. Add is allowed to write into spare capacity because older readers hold a smaller count and never observe the new slot; removal shifts elements a concurrent reader may be enumerating, so the same trick is unsafe. Tests drive retirement directly: in-flight work completes across it, the server is forgotten along with its aliases, it stops being selected, retiring twice is harmless, and removing the middle of three leaves the others working. --- .../ConnectionMultiplexer.cs | 72 ++++++++- src/StackExchange.Redis/LoggerExtensions.cs | 12 ++ src/StackExchange.Redis/ServerEndPoint.cs | 52 ++++++ .../ServerRetirementUnitTests.cs | 152 ++++++++++++++++++ 4 files changed, 287 insertions(+), 1 deletion(-) create mode 100644 tests/StackExchange.Redis.Tests/ServerRetirementUnitTests.cs diff --git a/src/StackExchange.Redis/ConnectionMultiplexer.cs b/src/StackExchange.Redis/ConnectionMultiplexer.cs index 75673fe1f..f8df35fa5 100644 --- a/src/StackExchange.Redis/ConnectionMultiplexer.cs +++ b/src/StackExchange.Redis/ConnectionMultiplexer.cs @@ -822,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(); @@ -937,7 +967,47 @@ public ServerSnapshotFiltered(ServerEndPoint[] endpoints, int count, Func + /// 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 _); + } + } } /// diff --git a/src/StackExchange.Redis/LoggerExtensions.cs b/src/StackExchange.Redis/LoggerExtensions.cs index 875135adb..4f9d26b87 100644 --- a/src/StackExchange.Redis/LoggerExtensions.cs +++ b/src/StackExchange.Redis/LoggerExtensions.cs @@ -332,6 +332,18 @@ internal static void LogWithThreadPoolStats(this ILogger? log, string message) Message = "Registering {EndPoint} without connecting: serves no slots")] internal static partial void LogInformationRegisteringInertNode(this ILogger logger, EndPointLogValue endPoint); + [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/ServerEndPoint.cs b/src/StackExchange.Redis/ServerEndPoint.cs index 4bd9c32ef..b183f2c43 100644 --- a/src/StackExchange.Redis/ServerEndPoint.cs +++ b/src/StackExchange.Redis/ServerEndPoint.cs @@ -22,6 +22,9 @@ internal enum UnselectableFlags RedundantPrimary = 1, DidNotRespond = 2, ServerType = 4, + + /// This server is being retired; it must not be selected for new work. + Retiring = 8, } internal sealed partial class ServerEndPoint : IDisposable @@ -232,6 +235,55 @@ 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; + + /// + /// 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; 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); + } +} From 73a7ff6a7ce478abd29944c1e820cf9787da6c9e Mon Sep 17 00:00:00 2001 From: Marc Gravell Date: Mon, 17 Aug 2026 16:19:24 +0100 Subject: [PATCH 5/6] Prune rotated-out endpoints, and merge duplicates, from the topology pass Provenance and generation tracking on ServerEndPoint, then two policies over the retirement primitive. Provenance records how we learned of a server, because "absent from the topology" only means anything if the source that would have listed it actually ran. A configured endpoint is never pruned - it is the seed needed to bootstrap after a full rotation. A sentinel-discovered one is never pruned by *cluster* absence: in a sentinel deployment no cluster topology runs at all, so a single rule would retire the entire deployment. A redirect target is legitimately ahead of the topology, so it is exempt until a topology confirms it, at which point it becomes an ordinary cluster node. Absence is counted in topology generations rather than time, so a quiet client cannot age endpoints out simply by not reconfiguring; three consecutive misses are required, since a single reply is only one node's view. A server is only retired if it is also idle - owns no slots, carries no subscriptions, owes no responses. Duplicate merge falls out of the same pass: resolving each node's identities and finding two distinct servers means one process reached under two names. The survivor is the configured one if either is, else the one matching the form the answering node advertised (Identities is ordered accordingly); the loser's name becomes an alias of the survivor, so a caller still holding it keeps resolving rather than breaking. One deviation from the design notes, recorded in the remarks on OnMissingFromTopology: they proposed also resetting the absence count whenever the server had been *used* since. That is not implementable - the only usage counter is incremented by our own heartbeat pings as well as by callers, so an idle-but-connected server never looks unused, and a first attempt at it silently prevented all pruning. It is also unnecessary, because the cases it was meant to protect are exactly the ones IsIdle already covers. --- .../ConnectionMultiplexer.Sentinel.cs | 2 +- .../ConnectionMultiplexer.cs | 113 ++++++++++- src/StackExchange.Redis/LoggerExtensions.cs | 12 ++ src/StackExchange.Redis/ServerEndPoint.cs | 89 ++++++++- .../ServerSelectionStrategy.cs | 19 +- .../EndpointPruningUnitTests.cs | 176 ++++++++++++++++++ 6 files changed, 406 insertions(+), 5 deletions(-) create mode 100644 tests/StackExchange.Redis.Tests/EndpointPruningUnitTests.cs 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 f8df35fa5..bef9b42c4 100644 --- a/src/StackExchange.Redis/ConnectionMultiplexer.cs +++ b/src/StackExchange.Redis/ConnectionMultiplexer.cs @@ -974,6 +974,105 @@ public ServerSnapshotFiltered(ServerEndPoint[] endpoints, int count, Func + /// 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. @@ -1039,7 +1138,11 @@ internal void RegisterServerIdentities(ClusterTopology topology) } [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] ?? TryResolveServerEndPoint(endpoint); @@ -1053,7 +1156,7 @@ internal void RegisterServerIdentities(ClusterTopology topology) { if (_isDisposed) throw new ObjectDisposedException(ToString()); - server = new ServerEndPoint(this, endpoint); + server = new ServerEndPoint(this, endpoint, provenance); servers.Add(endpoint, server); isNew = true; _serverSnapshot = _serverSnapshot.Add(server); @@ -1974,6 +2077,12 @@ public EndPoint[] GetEndPoints(bool configuredOnly = false) => { 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; } catch (Exception ex) diff --git a/src/StackExchange.Redis/LoggerExtensions.cs b/src/StackExchange.Redis/LoggerExtensions.cs index 4f9d26b87..03c62ed95 100644 --- a/src/StackExchange.Redis/LoggerExtensions.cs +++ b/src/StackExchange.Redis/LoggerExtensions.cs @@ -332,6 +332,18 @@ internal static void LogWithThreadPoolStats(this ILogger? log, string message) 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, diff --git a/src/StackExchange.Redis/ServerEndPoint.cs b/src/StackExchange.Redis/ServerEndPoint.cs index b183f2c43..494c286d0 100644 --- a/src/StackExchange.Redis/ServerEndPoint.cs +++ b/src/StackExchange.Redis/ServerEndPoint.cs @@ -27,6 +27,31 @@ internal enum UnselectableFlags 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 { internal volatile ServerEndPoint? Primary; @@ -49,10 +74,11 @@ 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; + Provenance = multiplexer.RawConfig.EndPoints.Contains(endpoint) ? ServerProvenance.Configured : provenance; var config = multiplexer.RawConfig; version = config.DefaultVersion; replicaReadOnly = true; @@ -240,6 +266,67 @@ public int WriteEverySeconds /// 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. 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/EndpointPruningUnitTests.cs b/tests/StackExchange.Redis.Tests/EndpointPruningUnitTests.cs new file mode 100644 index 000000000..dfadbd532 --- /dev/null +++ b/tests/StackExchange.Redis.Tests/EndpointPruningUnitTests.cs @@ -0,0 +1,176 @@ +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 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); + } +} From d63b5d9103a191c49d1265d95159fc1802e9fa9b Mon Sep 17 00:00:00 2001 From: Marc Gravell Date: Mon, 17 Aug 2026 16:40:05 +0100 Subject: [PATCH 6/6] Dial the advertised identity, and keep configured provenance under ResolveDns Two corrections to the previous commits in this branch. A new node is now dialled by the form the answering node *advertised* rather than by an address in preference. The earlier choice reasoned that an address is dialable as-is - true, but backwards for the deployments this work targets: a certificate validates against a name, and where hostnames are preferred the advertised address may not be routable at all, which is #2826's premise. Identities is already ordered with the advertised form first, so this is a deletion rather than an addition. And a bug found while checking the TLS story: ResolveDns rewrites the multiplexer's working set at startup, replacing configured DnsEndPoints with the addresses they resolved to, while RawConfig keeps the original names. Provenance tested only RawConfig, so with ResolveDns enabled a *configured* endpoint was classified as discovered - and therefore prunable. It now tests both collections. Both cases have regression tests. Also relevant, having checked it rather than assumed: the SslHost derivation hazard from the design notes does not apply here. ConfigurationOptions.SslHost falls back to a value derived from ConfigurationOptions.EndPoints, and nothing in this branch mutates that collection - the multiplexer works on a clone, discovery adds to the server table, and even sentinel's Clear/TryAdd hit the clone. So the derived TLS host cannot flap as endpoints come and go. --- .../ConnectionMultiplexer.cs | 17 ++----- src/StackExchange.Redis/ServerEndPoint.cs | 8 +++- .../EndpointPruningUnitTests.cs | 46 +++++++++++++++++++ 3 files changed, 58 insertions(+), 13 deletions(-) diff --git a/src/StackExchange.Redis/ConnectionMultiplexer.cs b/src/StackExchange.Redis/ConnectionMultiplexer.cs index bef9b42c4..2a334c431 100644 --- a/src/StackExchange.Redis/ConnectionMultiplexer.cs +++ b/src/StackExchange.Redis/ConnectionMultiplexer.cs @@ -2091,18 +2091,14 @@ public EndPoint[] GetEndPoints(bool configuredOnly = false) => return null; } - // prefer an identity we already know, then an address, then whatever we were given: an address is - // dialable as-is, whereas a hostname is only usable if it resolves + // 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; } - foreach (var identity in node.Identities) - { - if (identity is IPEndPoint) return identity; - } return node.Identities.Count > 0 ? node.Identities[0] : null; } } @@ -2309,12 +2305,9 @@ internal void UpdateClusterRange(ClusterTopology topology) if (TryResolveServerEndPoint(identity) is { } known) return known; } - // unknown node: create it under the form this reply used, preferring an address if we were - // given one - a hostname is only usable if it resolves, whereas the address is dialable as-is - foreach (var identity in node.Identities) - { - if (identity is IPEndPoint) return GetServerEndPoint(identity); - } + // 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; } } diff --git a/src/StackExchange.Redis/ServerEndPoint.cs b/src/StackExchange.Redis/ServerEndPoint.cs index 494c286d0..7ff9ed510 100644 --- a/src/StackExchange.Redis/ServerEndPoint.cs +++ b/src/StackExchange.Redis/ServerEndPoint.cs @@ -78,7 +78,13 @@ public ServerEndPoint(ConnectionMultiplexer multiplexer, EndPoint endpoint, Serv { Multiplexer = multiplexer; EndPoint = endpoint; - Provenance = multiplexer.RawConfig.EndPoints.Contains(endpoint) ? ServerProvenance.Configured : provenance; + // 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; diff --git a/tests/StackExchange.Redis.Tests/EndpointPruningUnitTests.cs b/tests/StackExchange.Redis.Tests/EndpointPruningUnitTests.cs index dfadbd532..0d558a995 100644 --- a/tests/StackExchange.Redis.Tests/EndpointPruningUnitTests.cs +++ b/tests/StackExchange.Redis.Tests/EndpointPruningUnitTests.cs @@ -146,6 +146,52 @@ public async Task SentinelDiscoveredServerIsNotPrunedByClusterAbsence() 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() {