Skip to content

Fix hostname-preferred cluster reconnects #3346

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
meerasridharan-28 wants to merge 3 commits into from
Closed

Fix hostname-preferred cluster reconnects #3346

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
63a3d58
fixreview
meerasridharan-28 Nov 3, 2026
2e6f2ee
Fix hostname-preferred cluster reconnects
meerasridharan-28 Nov 11, 2026
7102d53
Avoid duplicate xUnit test method names
meerasridharan-28 Nov 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions src/StackExchange.Redis/ClusterTopology.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -110,6 +110,12 @@ internal ClusterTopologyNode(ClusterSlotNode node, string? primaryNodeId)
/// </summary>
public IReadOnlyList<EndPoint> Identities => _identities ?? [];

/// <summary>
/// The announced endpoint form to use for dialing. Companion identities such as an IP address must not be
/// registered as separate dialable servers when this preferred endpoint is a hostname.
/// </summary>
public EndPoint? PreferredDialEndPoint => Identities.Count > 0 ? Identities[0] : null;

/// <summary>The slots this node serves as primary; empty for a replica.</summary>
public IReadOnlyList<SlotRange> Slots => _slots ?? [];

Expand Down
143 changes: 98 additions & 45 deletions src/StackExchange.Redis/ConnectionMultiplexer.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,7 @@ public sealed partial class ConnectionMultiplexer : IInternalConnectionMultiplex
// 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<EndPoint, ServerEndPoint> _serverIdentities = new();
private readonly ConcurrentDictionary<ServerEndPoint, EndPoint> _preferredDialEndpoints = new();
private volatile ServerSnapshot _serverSnapshot = ServerSnapshot.Empty;

private volatile bool _isDisposed;
Expand Down Expand Up @@ -1006,6 +1007,9 @@ ServerEndPoint IInternalConnectionMultiplexer.GetServerEndPoint(EndPoint endpoin
: null;
}

internal EndPoint GetDialEndPoint(ServerEndPoint server)
=> _preferredDialEndpoints.TryGetValue(server, out var endpoint) ? endpoint : server.EndPoint;

// 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;
Expand Down Expand Up @@ -1033,30 +1037,28 @@ internal async Task ApplyTopologyGenerationAsync(ClusterTopology topology, ILogg
List<DuplicateServers>? merge = null;
foreach (var node in topology.Nodes)
{
ServerEndPoint? survivor = null;
var directServers = new List<ServerEndPoint>();
foreach (var identity in node.Identities)
{
if (TryResolveServerEndPoint(identity) is not { } server) continue;
if (Equals(server.EndPoint, identity) && !directServers.Contains(server)) directServers.Add(server);
}

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))
var survivor = ResolveKnownServer(node);
if (survivor is not null)
{
foreach (var server in directServers)
{
if (ReferenceEquals(survivor, server)) continue;
// ...unless one of them was configured: those may never be retired, so they win
var pair = server.Provenance == ServerProvenance.Configured
? new DuplicateServers(server, survivor)
: new DuplicateServers(survivor, server);
survivor = pair.Survivor;
(merge ??= new()).Add(pair);
}
}

if (survivor is not null)
{
SetPreferredDialEndPoint(node, survivor);
survivor.OnSeenInTopology(generation);
seen.Add(survivor);

Expand Down Expand Up @@ -1141,6 +1143,7 @@ internal async Task RetireServerAsync(ServerEndPoint server, string reason, Time
_serverIdentities.TryRemove(pair.Key, out _);
}
}
_preferredDialEndpoints.TryRemove(server, out _);
}

/// <summary>
Expand All @@ -1152,23 +1155,59 @@ 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)
RegisterServerIdentities(node);
}
}

private void RegisterServerIdentities(ClusterTopologyNode node)
{
if (ResolveKnownServer(node) is not { } known) return;
SetPreferredDialEndPoint(node, known);

foreach (var identity in node.Identities)
{
if (servers[identity] is not null) continue; // already a server in its own right
var previous = _serverIdentities.AddOrUpdate(identity, known, (_, _) => known);
if (!ReferenceEquals(previous, known))
{
if ((known = TryResolveServerEndPoint(identity)) is not null) break;
Trace($"Identity {Format.ToString(identity)} -> {Format.ToString(known.EndPoint)}");
}
if (known is null) continue;
}
}

foreach (var identity in node.Identities)
private void SetPreferredDialEndPoint(ClusterTopologyNode node, ServerEndPoint server)
{
if (node.PreferredDialEndPoint is DnsEndPoint preferred)
{
_preferredDialEndpoints[server] = preferred;
}
else
{
_preferredDialEndpoints.TryRemove(server, out _);
}
}

private ServerEndPoint? ResolveKnownServer(ClusterTopologyNode node)
{
var preferred = node.PreferredDialEndPoint;
ServerEndPoint? preferredKnown = null, dnsKnown = null, anyDirectKnown = null, aliasKnown = null;

foreach (var identity in node.Identities)
{
if (TryResolveServerEndPoint(identity) is not { } known) continue;
if (!Equals(known.EndPoint, identity))
{
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)}");
}
aliasKnown ??= known;
continue;
}

anyDirectKnown ??= known;
if (preferred is not null && Equals(known.EndPoint, preferred)) preferredKnown = known;
if (known.EndPoint is DnsEndPoint) dnsKnown ??= known;
}

return preferredKnown
?? (preferred is DnsEndPoint ? dnsKnown ?? anyDirectKnown ?? aliasKnown : anyDirectKnown ?? aliasKnown);
}

/// <summary>
Expand Down Expand Up @@ -2249,6 +2288,8 @@ public EndPoint[] GetEndPoints(bool configuredOnly = false) =>
var topology = ClusterTopology.From(slots);
var clusterEndpoints = new EndPointCollection();
HashSet<EndPoint>? listedInSlots = null;
Dictionary<EndPoint, EndPoint>? preferredByIdentity = null;
Dictionary<EndPoint, ClusterTopologyNode>? nodeByIdentity = null;

if (topology is not null)
{
Expand All @@ -2257,13 +2298,23 @@ public EndPoint[] GetEndPoints(bool configuredOnly = false) =>
// another name is not duplicated
RegisterServerIdentities(topology);
listedInSlots = new HashSet<EndPoint>();
preferredByIdentity = new Dictionary<EndPoint, EndPoint>();
nodeByIdentity = new Dictionary<EndPoint, ClusterTopologyNode>();
foreach (var node in topology.Nodes)
{
if (SelectIdentity(node) is { } endpoint) clusterEndpoints.TryAdd(endpoint);

// every name the node answers to, not just the one chosen above: NODES may well
// report it under a different form than SLOTS did
foreach (var identity in node.Identities) listedInSlots.Add(identity);
foreach (var identity in node.Identities)
{
listedInSlots.Add(identity);
if (!nodeByIdentity.ContainsKey(identity)) nodeByIdentity.Add(identity, node);
if (node.PreferredDialEndPoint is { } preferred && !preferredByIdentity.ContainsKey(identity))
{
preferredByIdentity.Add(identity, preferred);
}
}
}
}

Expand All @@ -2286,12 +2337,17 @@ public EndPoint[] GetEndPoints(bool configuredOnly = false) =>
else if (listedInSlots!.Contains(node.EndPoint))
{
// SLOTS lists it, so it is in clusterEndpoints and we are about to connect to it and
// wait for it. Only the *existence* of a server was ever checked here, which a node
// in this state fails - a replica of a shard we have not reached yet is created by
// its primary's genealogy pass, and that has not run. Registering it inert then left
// it waited on but undialled, for the whole ConnectTimeout (#3232)
log?.LogInformationRegisteringSlotMapNode(new(node.EndPoint));
GetServerEndPoint(node.EndPoint, ServerProvenance.ClusterTopology, log);
// wait for it. Prefer the SLOTS hostname over its companion IP from NODES: that IP
// may be an unreachable pod address using the ingress client port.
var dialEndPoint = preferredByIdentity!.TryGetValue(node.EndPoint, out var preferred)
? preferred
: node.EndPoint;
log?.LogInformationRegisteringSlotMapNode(new(dialEndPoint));
GetServerEndPoint(dialEndPoint, ServerProvenance.ClusterTopology, log);
if (!Equals(dialEndPoint, node.EndPoint) && nodeByIdentity!.TryGetValue(node.EndPoint, out var slotsNode))
{
RegisterServerIdentities(slotsNode);
}
}
else
{
Expand Down Expand Up @@ -2320,15 +2376,12 @@ public EndPoint[] GetEndPoints(bool configuredOnly = false) =>
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;
var known = ResolveKnownServer(node);
if (known is null) return node.PreferredDialEndPoint;
SetPreferredDialEndPoint(node, known);
return known.EndPoint;
}
}

Expand Down Expand Up @@ -2552,8 +2605,6 @@ internal void UpdateClusterRange(ClusterTopology topology)
{
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;

Expand All @@ -2565,15 +2616,17 @@ internal void UpdateClusterRange(ClusterTopology topology)

ServerEndPoint? ResolveOrCreate(ClusterTopologyNode node)
{
foreach (var identity in node.Identities)
var server = ResolveKnownServer(node);
if (server is null && node.PreferredDialEndPoint is { } preferred)
{
if (TryResolveServerEndPoint(identity) is { } known) return known;
server = GetServerEndPoint(preferred, ServerProvenance.ClusterTopology);
}

// 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], ServerProvenance.ClusterTopology) : null;
if (server is not null)
{
SetPreferredDialEndPoint(node, server);
RegisterServerIdentities(node);
}
return server;
}
}

Expand Down
13 changes: 7 additions & 6 deletions src/StackExchange.Redis/PhysicalConnection.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -296,7 +296,9 @@ private static void DiscardTimeout(ref CancellationTokenSource? source)
internal async Task BeginConnectAsync(ILogger? log)
{
var bridge = BridgeCouldBeNull;
var endpoint = bridge?.ServerEndPoint?.EndPoint;
var endpoint = bridge is { } currentBridge && currentBridge.ServerEndPoint is { } serverEndPoint
? currentBridge.Multiplexer.GetDialEndPoint(serverEndPoint)
: null;
if (bridge == null || endpoint == null)
{
log?.LogErrorNoEndpoint(new ArgumentNullException(nameof(endpoint)));
Expand All @@ -309,9 +311,8 @@ internal async Task BeginConnectAsync(ILogger? log)
var connectTo = endpoint;

// A MOVING may name where to go next; prefer it over resolving the endpoint again, since not
// waiting for DNS is the point. Note this changes only the *socket target*: the endpoint itself is
// untouched, so identity, server selection and - importantly - the TLS host and SNI (derived from
// ServerEndPoint.EndPoint below) all stay exactly as configured.
// waiting for DNS is the point. It changes only the socket target; the preferred dial endpoint
// remains the source for TLS host and SNI.
if (bridge.ServerEndPoint?.HandoffTarget is { } handoffTarget)
{
Trace($"handoff: connecting to {Format.ToString(handoffTarget)} in place of {Format.ToString(endpoint)}");
Expand Down Expand Up @@ -1308,7 +1309,7 @@ internal async ValueTask<bool> ConnectedAsync(Socket? socket, ILogger? log)
if (tunnel is not null)
{
_attemptStage = ConnectionAttemptStage.Tunnel;
stream = await tunnel.BeforeAuthenticateAsync(bridge.ServerEndPoint.EndPoint, bridge.ConnectionType, socket, CancellationToken.None).ForAwait();
stream = await tunnel.BeforeAuthenticateAsync(bridge.Multiplexer.GetDialEndPoint(bridge.ServerEndPoint), bridge.ConnectionType, socket, CancellationToken.None).ForAwait();
}

static Stream DemandSocketStream(Socket? socket)
Expand All @@ -1318,7 +1319,7 @@ static Stream DemandSocketStream(Socket? socket)
{
_attemptStage = ConnectionAttemptStage.Tls;
log?.LogInformationConfiguringTLS();
var host = _tlsHostName = config.ResolveTlsHostName(bridge.ServerEndPoint.EndPoint);
var host = _tlsHostName = config.ResolveTlsHostName(bridge.Multiplexer.GetDialEndPoint(bridge.ServerEndPoint));

stream ??= DemandSocketStream(socket);
var certificateValidation = config.CertificateValidationCallback ?? GetAmbientIssuerCertificateCallback();
Expand Down
11 changes: 5 additions & 6 deletions src/StackExchange.Redis/ServerEndPoint.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -532,9 +532,8 @@ internal bool TryRerouteToSubscriptionBridge(Message message, PhysicalBridge fro
public RedisFeatures GetFeatures() => new RedisFeatures(version);

/// <summary>
/// The <c>CLUSTER SLOTS</c> view of the topology, keyed on node-id. Populated alongside
/// <see cref="ClusterConfiguration"/> but not yet used for routing, so the two can be compared
/// before anything depends on this one.
/// The <c>CLUSTER SLOTS</c> view of the topology, keyed on node-id. It drives slot routing when
/// available; <see cref="ClusterConfiguration"/> continues to provide node relations.
/// </summary>
internal ClusterTopology? ClusterTopology { get; private set; }

Expand All @@ -544,10 +543,10 @@ internal void SetClusterSlots(ClusterSlotsResult? slots)
if (topology is not null)
{
ClusterTopology = topology;
Multiplexer.Trace($"Shadow topology: {topology.Nodes.Count} nodes");
Multiplexer.Trace($"Cluster 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
// Register identities immediately, so a node reached by its other name resolves to the
// existing server rather than becoming a duplicate.
Multiplexer.RegisterServerIdentities(topology);
}
}
Expand Down
Loading
Loading