diff --git a/.github/workflows/CI.yml b/.github/workflows/CI.yml index f4eb0d25b..b98aefc42 100644 --- a/.github/workflows/CI.yml +++ b/.github/workflows/CI.yml @@ -213,6 +213,19 @@ jobs: name: Tests Results - Windows Server 2025 path: 'test-results/*.trx' reporter: dotnet-trx + # list only failures: listing every test rendered ~3MB into $GITHUB_STEP_SUMMARY, which has a + # 1MB limit, so the summary was dropped entirely and failures had to be read from annotations + list-suites: failed + list-tests: failed + - name: Upload test results + uses: actions/upload-artifact@v4 + if: success() || failure() + with: + # the trx carries per-test output, which the failure-only report above deliberately omits; that is + # the difference between diagnosing a CI-only failure and guessing at it + name: test-results-windows + path: test-results/*.trx + retention-days: 14 # Package and upload to MyGet only on pushes to main/v3, not on PRs - name: .NET Pack if: (github.event_name == 'push' || github.event_name == 'workflow_dispatch') && (github.ref == 'refs/heads/main' || github.ref == 'refs/heads/v3') diff --git a/.gitignore b/.gitignore index 1658a82e7..e8b097e22 100644 --- a/.gitignore +++ b/.gitignore @@ -31,3 +31,6 @@ launchSettings.json TestResults/ BenchmarkDotNet.Artifacts/ .codex + +# local planning/design notes (not for publication) +planning/ diff --git a/src/StackExchange.Redis/Availability/FaultContext.cs b/src/StackExchange.Redis/Availability/FaultContext.cs index dbb055f15..d533d4506 100644 --- a/src/StackExchange.Redis/Availability/FaultContext.cs +++ b/src/StackExchange.Redis/Availability/FaultContext.cs @@ -135,6 +135,7 @@ private static bool IsKnownNotApplied(RedisErrorKind kind, CommandStatus status) case RedisErrorKind.TryAgain: // slot mid-migration case RedisErrorKind.Moved: // wrong node; this one did not run it case RedisErrorKind.Ask: // ditto, mid-migration + case RedisErrorKind.UnknownRedirectTarget: // a redirect we could not follow, so still not run case RedisErrorKind.ReadOnly: // writes refused by a replica case RedisErrorKind.Misconfigured: // e.g. persistence failing, so writes are refused case RedisErrorKind.NoReplicas: // not enough replicas to accept the write diff --git a/src/StackExchange.Redis/ClusterConfiguration.cs b/src/StackExchange.Redis/ClusterConfiguration.cs index 072c6720f..2880cf6ac 100644 --- a/src/StackExchange.Redis/ClusterConfiguration.cs +++ b/src/StackExchange.Redis/ClusterConfiguration.cs @@ -287,10 +287,17 @@ internal ClusterNode(ClusterConfiguration configuration, string raw, EndPoint or var flags = parts[2].Split(StringSplits.Comma); - // redis 4 changes the format of "cluster nodes" - adds @... to the endpoint + // redis 4 changes the format of "cluster nodes" - adds @... to the endpoint, and the documented + // form is now ip:port@cport[,hostname[,aux-field=value]*]; keep all of it rather than just the + // address, since a hostname here is the node's second identity (see #2826) var ep = parts[1]; int at = ep.IndexOf('@'); - if (at >= 0) ep = ep.Substring(0, at); + if (at >= 0) + { + ParseBusPortAndIdentities(ep.Substring(at + 1)); + ep = ep.Substring(0, at); + } + AuxFields ??= []; if (Format.TryParseEndPoint(ep, out var epResult)) { @@ -328,6 +335,59 @@ internal ClusterNode(ClusterConfiguration configuration, string raw, EndPoint or IsConnected = parts[7] == "connected"; // Can be "connected" or "disconnected" } + // "cport[,hostname[,aux-field=value]*]" - everything after the '@'. Parsed leniently on purpose: + // this is topology data, and per the note on the constructor an exception here does silent damage. + // Note the hostname slot can be empty while aux fields follow, i.e. "17000,,shard-id=abc" + private void ParseBusPortAndIdentities(string trailer) + { + var fields = trailer.Split(StringSplits.Comma); + + if (Format.TryParseInt32(fields[0], out var busPort)) ClusterBusPort = busPort; + + if (fields.Length > 1 && !string.IsNullOrWhiteSpace(fields[1])) Hostname = fields[1]; + + List>? aux = null; + for (int i = 2; i < fields.Length; i++) + { + var field = fields[i]; + int eq = field.IndexOf('='); + + // parse by name and keep what we do not recognize; the set is documented as extensible + if (eq > 0) + { + (aux ??= new List>(fields.Length - i)).Add( + new KeyValuePair(field.Substring(0, eq), field.Substring(eq + 1))); + } + } + if (aux is null) + { + AuxFields = []; + } + else + { + AuxFields = aux.AsReadOnly(); + } + } + + /// + /// Gets the cluster bus port of the current node (the cport after the @), or + /// null for servers older than 4.0, which do not report one. + /// + public int? ClusterBusPort { get; private set; } + + /// + /// Gets the hostname announced by the current node via cluster-announce-hostname, or + /// null if it announces none. This is an additional identity for the node rather than a + /// replacement for , which remains the address the node reports. + /// + public string? Hostname { get; private set; } + + /// + /// Gets any auxiliary fields the current node reports after its hostname, as declared. The set is + /// documented as extensible, so unrecognized fields are preserved rather than discarded. + /// + public IReadOnlyList> AuxFields { get; private set; } + /// /// Gets all child nodes of the current node. /// diff --git a/src/StackExchange.Redis/ClusterSlotMetadataKey.cs b/src/StackExchange.Redis/ClusterSlotMetadataKey.cs new file mode 100644 index 000000000..6097f7f66 --- /dev/null +++ b/src/StackExchange.Redis/ClusterSlotMetadataKey.cs @@ -0,0 +1,36 @@ +using System; +using RESPite; + +namespace StackExchange.Redis; + +/// +/// Keys that can appear in the per-node metadata map of a CLUSTER SLOTS reply. +/// +/// +/// Matched case-insensitively on purpose: the documentation renders these as IP/Hostname in +/// prose but lower-case in its examples, so the casing cannot be relied upon. The set is documented as +/// extensible, hence - an unrecognized key is preserved rather than discarded. +/// +internal enum ClusterSlotMetadataKey +{ + /// A key this library does not recognize. + [AsciiHash("")] + Unknown = 0, + + /// The node's address, supplied when the reported endpoint is some other form. + [AsciiHash("ip")] + Ip, + + /// The node's announced hostname, supplied when the reported endpoint is some other form. + [AsciiHash("hostname")] + Hostname, +} + +/// +/// Metadata and parsing methods for . +/// +internal static partial class ClusterSlotMetadataKeyMetadata +{ + [AsciiHash(CaseSensitive = false)] + internal static partial bool TryParse(ReadOnlySpan value, out ClusterSlotMetadataKey key); +} diff --git a/src/StackExchange.Redis/ClusterSlots.ResultProcessor.cs b/src/StackExchange.Redis/ClusterSlots.ResultProcessor.cs new file mode 100644 index 000000000..17b137139 --- /dev/null +++ b/src/StackExchange.Redis/ClusterSlots.ResultProcessor.cs @@ -0,0 +1,182 @@ +using System; +using System.Collections.Generic; +using System.Net; +using RESPite; +using RESPite.Messages; + +namespace StackExchange.Redis; + +public sealed partial class ClusterSlotsResult +{ + internal static readonly ResultProcessor Processor = new ClusterSlotsResultProcessor(); + + /// + /// As , but also records the result against the server that answered - the + /// autoconfigure path needs the side effect, callers of CLUSTER SLOTS do not. + /// + internal static readonly ResultProcessor AutoConfigureProcessor = new ClusterSlotsResultProcessor(autoConfigure: true); + + private sealed class ClusterSlotsResultProcessor(bool autoConfigure = false) : ResultProcessor + { + protected override bool SetResultCore(PhysicalConnection connection, Message message, ref RespReader reader) + { + if (reader.IsNull) + { + SetResult(message, null); + return true; + } + if (!reader.IsAggregate) return false; + + // the reply is an array of [from, to, primary, replica...]; parse leniently, since bad topology + // data does silent damage - drop anything malformed rather than failing the whole reply + List? assignments = null; + var ranges = reader.AggregateChildren(); + while (ranges.MoveNext()) + { + if (TryParseAssignment(ref ranges.Value, out var assignment)) + { + (assignments ??= new List()).Add(assignment); + } + } + + var result = new ClusterSlotsResult( + AsReadOnly(assignments)); + + if (autoConfigure) + { + connection.BridgeCouldBeNull?.ServerEndPoint?.SetClusterSlots(result); + } + + SetResult(message, result); + return true; + } + } + + private static bool TryParseAssignment(ref RespReader reader, out ClusterSlotAssignment assignment) + { + assignment = null!; + if (!reader.IsAggregate || reader.IsNull) return false; + + var children = reader.AggregateChildren(); + if (!children.MoveNext() || !children.Value.TryReadInt64(out var from)) return false; + if (!children.MoveNext() || !children.Value.TryReadInt64(out var to)) return false; + if (from < SlotRange.MinSlot || to > SlotRange.MaxSlot || to < from) return false; + + ClusterSlotNode? primary = null; + List? replicas = null; + while (children.MoveNext()) + { + if (!TryParseNode(ref children.Value, out var node)) continue; + if (primary is null) + { + primary = node; + } + else + { + (replicas ??= new List()).Add(node); + } + } + + if (primary is null) return false; // a range with no primary tells us nothing + + assignment = new ClusterSlotAssignment( + new SlotRange((int)from, (int)to), + primary, + AsReadOnly(replicas)); + return true; + } + + private static bool TryParseNode(ref RespReader reader, out ClusterSlotNode node) + { + node = null!; + if (!reader.IsAggregate || reader.IsNull) return false; + + var children = reader.AggregateChildren(); + + // element 0 is positionally "the endpoint"; its content is whichever form the answering node + // prefers, so it may be a hostname, an empty string, "?", or the RESP null - never assume an address + if (!children.MoveNext()) return false; + string? announced = children.Value.IsNull ? null : children.Value.ReadString(); + + if (!children.MoveNext() || !children.Value.TryReadInt64(out var port)) return false; + + // the node id arrived in 4.0 and the metadata map in 7.0, so both are absent on older servers + string? nodeId = children.MoveNext() ? children.Value.ReadString() : null; + + IReadOnlyList> metadata = []; + string? ip = null, hostname = null; + if (children.MoveNext() && children.Value.IsAggregate && !children.Value.IsNull) + { + metadata = ReadMetadata(ref children.Value, out ip, out hostname); + } + + node = new ClusterSlotNode(announced, (int)port, nodeId, ResolveEndPoint(announced, port), ip, hostname, metadata); + return true; + } + + // walked pairwise rather than by declared length, since a map reports pairs while an array (RESP2) + // reports elements, and we do not need to care which we were given. + // + // Keys are matched against the known set over the raw bytes, so a recognized key costs no string at all - + // and since ranges repeat a node whenever its slot ownership is not contiguous, that is one allocation + // avoided per key per range rather than per node. Only unrecognized keys are materialized, into the + // collection that exists to preserve them + private static unsafe IReadOnlyList> ReadMetadata( + ref RespReader reader, + out string? ip, + out string? hostname) + { + ip = hostname = null; + List>? metadata = null; + var children = reader.AggregateChildren(); + while (children.MoveNext()) + { + if (!children.Value.TryParseScalar(&ClusterSlotMetadataKeyMetadata.TryParse, out ClusterSlotMetadataKey key)) + { + key = ClusterSlotMetadataKey.Unknown; + } + + // the key is still in the reader until we move on, so an unknown one can be read then + var unknownKey = key == ClusterSlotMetadataKey.Unknown ? children.Value.ReadString() : null; + + if (!children.MoveNext()) break; // odd count: ignore the dangling key + var value = children.Value.IsNull ? null : children.Value.ReadString(); + + switch (key) + { + case ClusterSlotMetadataKey.Ip: + ip = value; + break; + case ClusterSlotMetadataKey.Hostname: + hostname = value; + break; + default: + if (!string.IsNullOrEmpty(unknownKey)) + { + (metadata ??= new List>()).Add(new(unknownKey!, value)); + } + break; + } + } + return AsReadOnly(metadata); + } + + // empty collection expressions to an interface target compile to Array.Empty(), so the common case + // costs nothing; only a genuinely populated list pays for the wrapper + private static IReadOnlyList AsReadOnly(List? values) + { + if (values is null) return []; + return values.AsReadOnly(); + } + + private static EndPoint? ResolveEndPoint(string? announced, long port) + { + // null and "" both mean "no endpoint reported"; "?" means an explicitly unknown node. None of the + // three can be turned into an endpoint here - substituting the connection's own address is a + // decision for the caller, and is outright wrong for "?" + if (announced is null or "" or "?") return null; + if (port is < 1 or > ushort.MaxValue) return null; + + return Format.TryParseEndPoint(announced, port.ToString(), out var endpoint) ? endpoint : null; + } +} diff --git a/src/StackExchange.Redis/ClusterSlots.Server.cs b/src/StackExchange.Redis/ClusterSlots.Server.cs new file mode 100644 index 000000000..f93583187 --- /dev/null +++ b/src/StackExchange.Redis/ClusterSlots.Server.cs @@ -0,0 +1,12 @@ +using System.Threading.Tasks; + +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); + + public Task ClusterSlotsAsync(CommandFlags flags = CommandFlags.None) + => ExecuteAsync(Message.Create(-1, flags, RedisCommand.CLUSTER, RedisLiterals.SLOTS), ClusterSlotsResult.Processor); +} diff --git a/src/StackExchange.Redis/ClusterSlots.cs b/src/StackExchange.Redis/ClusterSlots.cs new file mode 100644 index 000000000..dd28021d9 --- /dev/null +++ b/src/StackExchange.Redis/ClusterSlots.cs @@ -0,0 +1,159 @@ +using System.Collections.Generic; +using System.Net; +using System.Threading.Tasks; + +namespace StackExchange.Redis; + +public partial interface IServer +{ + /// + /// Obtains the slot-to-node mapping reported by CLUSTER SLOTS. + /// + /// The command flags to use. + /// + /// This is a different view of the topology to , not merely a + /// different encoding of it: the naming form used for each node is chosen by the node answering the + /// command, so two servers can describe the same cluster differently. See . + /// + ClusterSlotsResult? ClusterSlots(CommandFlags flags = CommandFlags.None); + + /// + /// Obtains the slot-to-node mapping reported by CLUSTER SLOTS. + /// + /// The command flags to use. + /// + /// This is a different view of the topology to , not merely + /// a different encoding of it: the naming form used for each node is chosen by the node answering the + /// command, so two servers can describe the same cluster differently. See . + /// + Task ClusterSlotsAsync(CommandFlags flags = CommandFlags.None); +} + +/// +/// The slot-to-node mapping reported by CLUSTER SLOTS, as described by the server that answered. +/// +public sealed partial class ClusterSlotsResult +{ + internal ClusterSlotsResult(IReadOnlyList assignments) => Assignments = assignments; + + /// + /// Gets the slot ranges reported, in the order the server reported them. Ranges may repeat a node when + /// its slot ownership is not contiguous, so this is not a per-node list. + /// + public IReadOnlyList Assignments { get; } +} + +/// +/// One slot range from CLUSTER SLOTS, and the nodes serving it. +/// +public sealed class ClusterSlotAssignment +{ + internal ClusterSlotAssignment(SlotRange slots, ClusterSlotNode primary, IReadOnlyList replicas) + { + Slots = slots; + Primary = primary; + Replicas = replicas; + } + + /// + /// Gets the range of slots covered by this assignment. + /// + public SlotRange Slots { get; } + + /// + /// Gets the node serving this range as primary. + /// + public ClusterSlotNode Primary { get; } + + /// + /// Gets the nodes replicating this range, if any. + /// + public IReadOnlyList Replicas { get; } +} + +/// +/// A node as described by CLUSTER SLOTS. +/// +/// +/// The endpoint field is positional rather than typed: its *content* is whichever naming form the answering +/// node prefers, so it may hold an address, a hostname, or one of the documented placeholders. Prefer +/// , which is populated only when the reported value is usable, and consult +/// / for the complementary form the server supplies alongside it. +/// +public sealed class ClusterSlotNode +{ + internal ClusterSlotNode( + string? announcedEndpoint, + int port, + string? nodeId, + EndPoint? endPoint, + string? ip, + string? hostname, + IReadOnlyList> metadata) + { + AnnouncedEndpoint = announcedEndpoint; + Port = port; + NodeId = nodeId; + EndPoint = endPoint; + Ip = ip; + Hostname = hostname; + Metadata = metadata; + } + + /// + /// Gets the endpoint exactly as reported, which may be an address, a hostname, an empty string, or + /// "?"; null when the server reported no endpoint at all. + /// + /// + /// The three unusable values do not mean the same thing. null means the server does not know this + /// node's address, typically because it is behind a load balancer: connect to the endpoint the command + /// was sent to, using . An empty string means the node does not know its own address, + /// and may be treated the same way. "?" means hostnames are preferred but this node announced + /// none - it identifies an *unknown* node, so unlike the other two it must **not** be assumed to be the + /// node that answered. + /// + public string? AnnouncedEndpoint { get; } + + /// + /// Gets the port reported for this node. + /// + public int Port { get; } + + /// + /// Gets the node identifier, or null for servers old enough not to report one. + /// + /// + /// This is the only identity here that does not depend on which node answered, so it is the reliable key + /// when reconciling nodes across replies. + /// + public string? NodeId { get; } + + /// + /// Gets the endpoint for this node, or null when the reported value is not usable as one - see + /// for what that means and how to recover. + /// + public EndPoint? EndPoint { get; } + + /// + /// Gets the address of this node when the server supplies it as metadata, which it does whenever the + /// answering node prefers some other naming form. + /// + public string? Ip { get; } + + /// + /// Gets the announced hostname of this node when the server supplies it as metadata, which it does when + /// the node has one and the answering node prefers some other naming form. + /// + public string? Hostname { get; } + + /// + /// Gets the metadata entries reported for this node that this library does not recognize. The set is + /// documented as extensible, so these are preserved rather than discarded; the recognized keys are + /// surfaced as and instead of appearing here, which also means a + /// recognized key costs no allocation. + /// + public IReadOnlyList> Metadata { get; } + + /// + public override string ToString() => $"{AnnouncedEndpoint ?? "(unknown)"}:{Port}"; +} diff --git a/src/StackExchange.Redis/ClusterTopology.cs b/src/StackExchange.Redis/ClusterTopology.cs new file mode 100644 index 000000000..a6ae7e3c2 --- /dev/null +++ b/src/StackExchange.Redis/ClusterTopology.cs @@ -0,0 +1,172 @@ +using System; +using System.Collections.Generic; +using System.Net; + +namespace StackExchange.Redis; + +/// +/// An internal view of the cluster keyed on node-id, built from CLUSTER SLOTS. +/// +/// +/// +/// Keyed on the node-id deliberately: it is the only identity in a reply that does not depend on which node +/// answered. A node's *names* (address, hostname) vary by the answering node's preferred endpoint type, so +/// keying on any of them means two replies describing one node can produce two entries. Keyed on the id, +/// they merge, and the names become attributes. +/// +/// +/// Currently populated in parallel with the CLUSTER NODES view that still drives routing, so that the +/// two can be compared before anything depends on this one. +/// +/// +internal sealed class ClusterTopology +{ + private readonly Dictionary _byNodeId; + + private ClusterTopology(Dictionary byNodeId) => _byNodeId = byNodeId; + + public IReadOnlyCollection Nodes => _byNodeId.Values; + + public ClusterTopologyNode? this[string nodeId] => _byNodeId.TryGetValue(nodeId, out var node) ? node : null; + + /// + /// Builds the view, or null if the reply carried nothing usable. Nodes without an id are skipped: + /// pre-4.0 servers do not report one, and without it we cannot merge identities reliably. + /// + public static ClusterTopology? From(ClusterSlotsResult? slots) + { + if (slots is null || slots.Assignments.Count == 0) return null; + + Dictionary? byNodeId = null; + foreach (var assignment in slots.Assignments) + { + Add(ref byNodeId, assignment.Primary, assignment.Slots, primaryNodeId: null); + foreach (var replica in assignment.Replicas) + { + Add(ref byNodeId, replica, assignment.Slots, primaryNodeId: assignment.Primary.NodeId); + } + } + + return byNodeId is null ? null : new ClusterTopology(byNodeId); + + static void Add( + ref Dictionary? byNodeId, + ClusterSlotNode node, + SlotRange slots, + string? primaryNodeId) + { + if (string.IsNullOrEmpty(node.NodeId)) return; + + byNodeId ??= new Dictionary(StringComparer.Ordinal); + if (!byNodeId.TryGetValue(node.NodeId!, out var existing)) + { + byNodeId.Add(node.NodeId!, existing = new ClusterTopologyNode(node, primaryNodeId)); + } + else + { + // non-contiguous ownership repeats a node across ranges, so merging must be idempotent + existing.Merge(node); + } + + if (primaryNodeId is null) existing.AddSlots(slots); + } + } +} + +/// +/// One node in a : an identity, the names it is known by, and what it serves. +/// +internal sealed class ClusterTopologyNode +{ + private List? _slots; + private List? _identities; + + internal ClusterTopologyNode(ClusterSlotNode node, string? primaryNodeId) + { + NodeId = node.NodeId!; + Port = node.Port; + PrimaryNodeId = primaryNodeId; + Merge(node); + } + + public string NodeId { get; } + + public int Port { get; } + + /// The node-id of this node's primary, if it was reported as a replica. + public string? PrimaryNodeId { get; } + + public bool IsReplica => PrimaryNodeId is not null; + + /// The address this node is known by, if any reply has supplied one. + public string? Ip { get; private set; } + + /// The hostname this node is known by, if any reply has supplied one. + public string? Hostname { get; private set; } + + /// + /// Every endpoint this node is known to answer to. More than one means the same node arrived under + /// different names, which is exactly the case that must not produce duplicate ServerEndPoints. + /// + public IReadOnlyList Identities => _identities ?? []; + + /// The slots this node serves as primary; empty for a replica. + public IReadOnlyList Slots => _slots ?? []; + + /// + /// Fold in another sighting of this node. Names accumulate rather than overwrite: a reply that renders + /// the node by address supplies the hostname as metadata and vice versa, so successive replies from + /// differently-configured nodes each contribute a piece. + /// + internal void Merge(ClusterSlotNode node) + { + // the identity set is the union of the primary field and the metadata, and the metadata is + // documented as the *complement* of the primary - so taking only the metadata loses whichever form + // the answering node happened to prefer. The primary field must be classified by its content, since + // positionally it is just "the endpoint" and may hold either form + if (NullIfEmpty(node.AnnouncedEndpoint) is { } announced && announced != "?") + { + if (IPAddress.TryParse(announced, out _)) + { + Ip ??= announced; + } + else + { + Hostname ??= announced; + } + } + + Ip ??= NullIfEmpty(node.Ip); + Hostname ??= NullIfEmpty(node.Hostname); + + // the announced endpoint is only an identity if it was usable at all, which excludes the + // "?" / empty / absent placeholders + AddIdentity(node.EndPoint); + if (Ip is not null) AddIdentity(new IPEndPointOrDns(Ip, Port).ToEndPoint()); + if (Hostname is not null) AddIdentity(new DnsEndPoint(Hostname, Port)); + + static string? NullIfEmpty(string? value) => string.IsNullOrEmpty(value) ? null : value; + } + + internal void AddSlots(SlotRange slots) + { + _slots ??= new List(); + if (!_slots.Contains(slots)) _slots.Add(slots); + } + + private void AddIdentity(EndPoint? endpoint) + { + if (endpoint is null) return; + _identities ??= new List(); + if (!_identities.Contains(endpoint)) _identities.Add(endpoint); + } + + public override string ToString() => $"{NodeId} {Ip ?? Hostname ?? "(unnamed)"}:{Port}{(IsReplica ? " (replica)" : "")}"; + + /// Parses a reported address, falling back to a name if it is not an address after all. + private readonly struct IPEndPointOrDns(string host, int port) + { + public EndPoint ToEndPoint() => + IPAddress.TryParse(host, out var address) ? new IPEndPoint(address, port) : new DnsEndPoint(host, port); + } +} diff --git a/src/StackExchange.Redis/ConnectionMultiplexer.ExportConfiguration.cs b/src/StackExchange.Redis/ConnectionMultiplexer.ExportConfiguration.cs index c095ffd53..3c6b1125c 100644 --- a/src/StackExchange.Redis/ConnectionMultiplexer.ExportConfiguration.cs +++ b/src/StackExchange.Redis/ConnectionMultiplexer.ExportConfiguration.cs @@ -53,6 +53,7 @@ public void ExportConfiguration(Stream destination, ExportOptions options = Expo if ((options & ExportOptions.Cluster) != 0) { tasks.Add(api.ClusterNodesRawAsync(flags: flags)); + tasks.Add(api.ClusterSlotsAsync(flags: flags)); } WaitAllIgnoreErrors(tasks.ToArray()); @@ -93,6 +94,10 @@ public void ExportConfiguration(Stream destination, ExportOptions options = Expo if ((options & ExportOptions.Cluster) != 0) { Write(zip, prefix + "/nodes.txt", tasks[index++], WriteNormalizingLineEndings); + + // the two views can legitimately disagree about how a node is *named*, since SLOTS + // renders endpoints in whichever form the answering node prefers - so capture both + Write(zip, prefix + "/slots.txt", tasks[index++], WriteClusterSlots); } } } @@ -122,6 +127,44 @@ private static void Write(ZipArchive zip, string name, Task task, Action value switch + { + null => "(null)", // the server does not know this node's address + "" => "(empty)", // the node does not know its own address + _ => value, // may be an address, a hostname, or "?" for an unknown node + }; + } + private static void WriteNormalizingLineEndings(string source, StreamWriter writer) { if (source == null) diff --git a/src/StackExchange.Redis/Enums/ExportOptions.cs b/src/StackExchange.Redis/Enums/ExportOptions.cs index 594651955..1ae357ae0 100644 --- a/src/StackExchange.Redis/Enums/ExportOptions.cs +++ b/src/StackExchange.Redis/Enums/ExportOptions.cs @@ -29,7 +29,7 @@ public enum ExportOptions Client = 4, /// - /// The output of CLUSTER NODES. + /// The output of CLUSTER NODES and CLUSTER SLOTS. /// Cluster = 8, diff --git a/src/StackExchange.Redis/PhysicalConnection.Read.cs b/src/StackExchange.Redis/PhysicalConnection.Read.cs index 4349c60ad..e30102fe3 100644 --- a/src/StackExchange.Redis/PhysicalConnection.Read.cs +++ b/src/StackExchange.Redis/PhysicalConnection.Read.cs @@ -435,10 +435,17 @@ private void OnResponseFrame(RespPrefix prefix, ReadOnlySpan frame, ref IM case RespPrefix.Array when (_protocol is RedisProtocol.Resp2 & connectionType is ConnectionType.Subscription) && !IsArrayPong(frame): // could be a RESP2 pub/sub payload // out-of-band; pub/sub etc - if (OnOutOfBand(frame, ref memoryOwner)) + switch (OnOutOfBand(frame, ref memoryOwner)) { - OnDetailLog($"out-of-band message, not dequeuing: {prefix}"); - return; + case OutOfBandResult.Handled: + OnDetailLog($"out-of-band message, not dequeuing: {prefix}"); + return; + case OutOfBandResult.NotRecognized when prefix is RespPrefix.Push: + // a RESP3 push frame is out-of-band *by definition*; if we don't recognize it + // (newer server, or a feature we don't implement) we drop it - matching it to a + // pending command would desynchronize the entire response stream + OnDetailLog($"unrecognized out-of-band message, dropped: {prefix}"); + return; } break; } @@ -499,6 +506,28 @@ internal static partial class PushKindMetadata internal static partial bool TryParse(ReadOnlySpan value, out PushKind result); } + /// + /// The outcome of inspecting a possible out-of-band frame. + /// + private enum OutOfBandResult + { + /// + /// We could not identify the frame; for a RESP3 push frame this means "drop it". + /// + NotRecognized, + + /// + /// Consumed as an out-of-band message; nothing further to do. + /// + Handled, + + /// + /// Recognized, but it is the reply to a command we sent (subscribe/unsubscribe confirmations + /// arrive as push frames in RESP3), so normal request/response matching must complete it. + /// + MatchToCommand, + } + internal static ReadOnlySpan StackCopyLengthChecked(scoped in RespReader reader, Span buffer) { var len = reader.CopyTo(buffer); @@ -506,10 +535,10 @@ internal static ReadOnlySpan StackCopyLengthChecked(scoped in RespReader r return buffer.Slice(0, len); } - private bool OnOutOfBand(ReadOnlySpan payload, ref IMemoryOwner? memoryOwner) + private OutOfBandResult OnOutOfBand(ReadOnlySpan payload, ref IMemoryOwner? memoryOwner) { var muxer = BridgeCouldBeNull?.Multiplexer; - if (muxer is null) return true; // consume it blindly + if (muxer is null) return OutOfBandResult.Handled; // consume it blindly var reader = new RespReader(payload); @@ -535,7 +564,7 @@ static bool TryMoveNextString(ref RespReader reader) => reader.SafeTryMoveNext() & reader.IsInlineScalar & reader.Prefix is RespPrefix.BulkString or RespPrefix.SimpleString; - if (kind is PushKind.None || !TryMoveNextString(ref reader)) return false; + if (kind is PushKind.None || !TryMoveNextString(ref reader)) return OutOfBandResult.NotRecognized; // the channel is always the second element var subscriptionChannel = AsRedisChannel(reader, channelOptions); @@ -575,7 +604,7 @@ static bool TryMoveNextString(ref RespReader reader) OnMessage(muxer, subscriptionChannel, subscriptionChannel, in reader); } - return true; + return OutOfBandResult.Handled; case PushKind.PMessage when TryMoveNextString(ref reader): _readStatus = ReadStatus.PubSubPMessage; @@ -586,7 +615,7 @@ static bool TryMoveNextString(ref RespReader reader) OnMessage(muxer, subscriptionChannel, messageChannel, in reader); } - return true; + return OutOfBandResult.Handled; case PushKind.SUnsubscribe when !PeekChannelMessage(RedisCommand.SUNSUBSCRIBE, subscriptionChannel): // then it was *unsolicited* - this probably means the slot was migrated // (otherwise, we'll let the command-processor deal with it) @@ -600,10 +629,16 @@ static bool TryMoveNextString(ref RespReader reader) // knows, but *we don't know who that is*, and other nodes: aren't guaranteed to know (yet) muxer.DefaultSubscriber.ResubscribeToServer(subscription, subscriptionChannel, server, cause: "sunsubscribe"); } - return true; + return OutOfBandResult.Handled; + + case PushKind.Subscribe or PushKind.PSubscribe or PushKind.SSubscribe + or PushKind.Unsubscribe or PushKind.PUnsubscribe or PushKind.SUnsubscribe: + // recognized, but these are confirmations of commands *we* sent (in RESP3 they + // arrive as push frames), so the normal request/response matching completes them + return OutOfBandResult.MatchToCommand; } } - return false; + return OutOfBandResult.NotRecognized; } private void OnMessage( diff --git a/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt b/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt index 80d94801e..295fc5b1f 100644 --- a/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt +++ b/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt @@ -15,3 +15,24 @@ StackExchange.Redis.ConfigurationOptions.SentinelUser.set -> void [SER009]StackExchange.Redis.Configuration.TlsOptions.ResolveHost(System.Net.EndPoint! endpoint) -> string! [SER009]virtual StackExchange.Redis.Configuration.Tunnel.ConnectTransportAsync(System.Net.EndPoint! endpoint, StackExchange.Redis.ConnectionType connectionType, StackExchange.Redis.Configuration.TlsOptions tls, System.Threading.CancellationToken cancellationToken) -> System.Threading.Tasks.ValueTask StackExchange.Redis.RedisFeatures.Hello.get -> bool +StackExchange.Redis.ClusterNode.AuxFields.get -> System.Collections.Generic.IReadOnlyList>! +StackExchange.Redis.ClusterNode.ClusterBusPort.get -> int? +StackExchange.Redis.ClusterNode.Hostname.get -> string? +StackExchange.Redis.ClusterSlotAssignment +StackExchange.Redis.ClusterSlotAssignment.Primary.get -> StackExchange.Redis.ClusterSlotNode! +StackExchange.Redis.ClusterSlotAssignment.Replicas.get -> System.Collections.Generic.IReadOnlyList! +StackExchange.Redis.ClusterSlotAssignment.Slots.get -> StackExchange.Redis.SlotRange +StackExchange.Redis.ClusterSlotNode +StackExchange.Redis.ClusterSlotNode.AnnouncedEndpoint.get -> string? +StackExchange.Redis.ClusterSlotNode.EndPoint.get -> System.Net.EndPoint? +StackExchange.Redis.ClusterSlotNode.Hostname.get -> string? +StackExchange.Redis.ClusterSlotNode.Ip.get -> string? +StackExchange.Redis.ClusterSlotNode.Metadata.get -> System.Collections.Generic.IReadOnlyList>! +StackExchange.Redis.ClusterSlotNode.NodeId.get -> string? +StackExchange.Redis.ClusterSlotNode.Port.get -> int +StackExchange.Redis.ClusterSlotsResult +StackExchange.Redis.ClusterSlotsResult.Assignments.get -> System.Collections.Generic.IReadOnlyList! +StackExchange.Redis.IServer.ClusterSlots(StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> StackExchange.Redis.ClusterSlotsResult? +StackExchange.Redis.IServer.ClusterSlotsAsync(StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> System.Threading.Tasks.Task! +[SER007]StackExchange.Redis.RedisErrorKind.UnknownRedirectTarget = 26 -> StackExchange.Redis.RedisErrorKind +override StackExchange.Redis.ClusterSlotNode.ToString() -> string! diff --git a/src/StackExchange.Redis/RedisErrorKind.cs b/src/StackExchange.Redis/RedisErrorKind.cs index 2a60f76f6..eee850e22 100644 --- a/src/StackExchange.Redis/RedisErrorKind.cs +++ b/src/StackExchange.Redis/RedisErrorKind.cs @@ -162,6 +162,19 @@ public enum RedisErrorKind /// NOSCRIPT - no matching script for EVALSHA; the script must be re-loaded. /// NoScript, + + /// + /// A MOVED or ASK redirect whose target cannot be identified, so it cannot be followed; + /// the slot has moved, but the reply does not say where to. + /// + /// + /// This is a client-side classification rather than a server error code: it is a special case of + /// / . It arises when the node answering prefers hostnames while the + /// target has announced none, giving MOVED <slot> ?:<port> - and ? denotes an + /// *unknown* node, so unlike a missing or empty endpoint it must not be treated as the answering node. + /// A topology refresh is requested when this happens, so a retry may well succeed. + /// + UnknownRedirectTarget, } internal static partial class RedisErrorKindMetadata diff --git a/src/StackExchange.Redis/RedisLiterals.cs b/src/StackExchange.Redis/RedisLiterals.cs index dc59ac4df..12b85ea2f 100644 --- a/src/StackExchange.Redis/RedisLiterals.cs +++ b/src/StackExchange.Redis/RedisLiterals.cs @@ -83,6 +83,7 @@ public static readonly RedisValue NO = RedisValue.FromRaw("NO"u8), NODES = RedisValue.FromRaw("NODES"u8), NOSAVE = RedisValue.FromRaw("NOSAVE"u8), + SLOTS = RedisValue.FromRaw("SLOTS"u8), nosort = RedisValue.FromRaw("nosort"u8), NOT = RedisValue.FromRaw("NOT"u8), NOVALUES = RedisValue.FromRaw("NOVALUES"u8), diff --git a/src/StackExchange.Redis/ResultProcessor.cs b/src/StackExchange.Redis/ResultProcessor.cs index 36e3045c9..2175294e8 100644 --- a/src/StackExchange.Redis/ResultProcessor.cs +++ b/src/StackExchange.Redis/ResultProcessor.cs @@ -37,6 +37,9 @@ public static readonly ResultProcessor public static readonly ResultProcessor ClusterNodes = new ClusterNodesProcessor(); + internal static readonly ResultProcessor + ClusterSlots = ClusterSlotsResult.AutoConfigureProcessor; + public static readonly ResultProcessor ConnectionIdentity = new ConnectionIdentityProcessor(); @@ -288,7 +291,24 @@ private bool HandleCommonError(Message message, RespReader reader, PhysicalConne log = false; string[] parts = reader.ReadString()!.Split(StringSplits.Space, 3); if (Format.TryParseInt32(parts[1], out int hashSlot) - && Format.TryParseEndPoint(parts[2], out var endpoint)) + && Format.TryParseEndPoint(parts[2], out var endpoint) + && IsUnroutableRedirect(endpoint)) + { + // the slot moved, but the reply does not say where to: a hostname-preferring node + // redirecting to a peer with no announced hostname reports "?", which denotes an + // *unknown* node and so cannot be resolved to the answering node either. Do not + // manufacture a ServerEndPoint for it - that would leave us dialling a host called "?" + // - but do ask for a topology refresh, which can name the target properly, since + // CLUSTER NODES reports addresses positionally whatever the preference + connection.OnDetailLog($"unroutable {(isMoved ? "MOVED" : "ASK")} target '{parts[2]}'"); + bridge?.Multiplexer.ReconfigureIfNeeded(server?.EndPoint, false, "unroutable redirect"); + errorKind = RedisErrorKind.UnknownRedirectTarget; + err = bridge?.Multiplexer.RawConfig.IncludeDetailInExceptions == true + ? $"The server redirected hashslot {hashSlot} to '{parts[2]}', which does not identify a node that can be connected to; a topology refresh has been requested. Command: {message.CommandAndKey}. " + : "The server redirected to an endpoint that does not identify a node that can be connected to; a topology refresh has been requested. "; + } + else if (Format.TryParseInt32(parts[1], out hashSlot) + && Format.TryParseEndPoint(parts[2], out endpoint)) { // Check if MOVED points to same endpoint bool isSameEndpoint = Equals(server?.EndPoint, endpoint); @@ -371,6 +391,16 @@ private bool HandleCommonError(Message message, RespReader reader, PhysicalConne protected abstract bool SetResultCore(PhysicalConnection connection, Message message, ref RespReader reader); + // "?" is the documented placeholder for an unknown node, and a zero/absent port is what the + // "unknown-endpoint" form parses to; neither can be dialled, and neither may be assumed to mean + // "the node that answered" - see the CLUSTER SLOTS endpoint contract + private static bool IsUnroutableRedirect(EndPoint endpoint) => endpoint switch + { + DnsEndPoint dns => dns.Port == 0 || dns.Host is "" or "?", + IPEndPoint ip => ip.Port == 0, + _ => false, + }; + private void UnexpectedResponse(Message message, in RespReader reader) { ConnectionMultiplexer.TraceWithoutContext("From " + GetType().Name, "Unexpected Response"); diff --git a/src/StackExchange.Redis/ServerEndPoint.cs b/src/StackExchange.Redis/ServerEndPoint.cs index e48274ed8..20968261b 100644 --- a/src/StackExchange.Redis/ServerEndPoint.cs +++ b/src/StackExchange.Redis/ServerEndPoint.cs @@ -322,6 +322,23 @@ internal bool TryRerouteToSubscriptionBridge(Message message, PhysicalBridge fro public RedisFeatures GetFeatures() => new RedisFeatures(version); + /// + /// The CLUSTER SLOTS view of the topology, keyed on node-id. Populated alongside + /// but not yet used for routing, so the two can be compared + /// before anything depends on this one. + /// + internal ClusterTopology? ClusterTopology { get; private set; } + + internal void SetClusterSlots(ClusterSlotsResult? slots) + { + var topology = ClusterTopology.From(slots); + if (topology is not null) + { + ClusterTopology = topology; + Multiplexer.Trace($"Shadow topology: {topology.Nodes.Count} nodes"); + } + } + public void SetClusterConfiguration(ClusterConfiguration configuration) { ClusterConfiguration = configuration; @@ -503,6 +520,18 @@ internal async Task AutoConfigureAsync(PhysicalConnection? connection, ILogger? 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/ClusterEndpointFormUnitTests.cs b/tests/StackExchange.Redis.Tests/ClusterEndpointFormUnitTests.cs new file mode 100644 index 000000000..751b65d9b --- /dev/null +++ b/tests/StackExchange.Redis.Tests/ClusterEndpointFormUnitTests.cs @@ -0,0 +1,354 @@ +using System.Collections.Generic; +using System.Net; +using System.Threading.Tasks; +using Xunit; +using static StackExchange.Redis.Server.RedisServer; + +namespace StackExchange.Redis.Tests; + +/// +/// The toy server's CLUSTER SLOTS and -MOVED naming, per its preferred endpoint type. +/// Nothing consumes this client-side yet; it exists so that identity handling has something faithful +/// to be written against, including the placeholder endpoint values the contract prescribes. +/// +public class ClusterEndpointFormUnitTests(ITestOutputHelper log) +{ + private const string Hostname = "host-1.redis.example.com"; + + // hostnames and the endpoint-type preference are 7.0 features, which the in-process server now + // declares by default; the pre-7.0 shape is what has to be asked for + private static readonly System.Version BeforeHostnames = new(6, 2, 0); + + private static InProcessTestServer CreateServer( + ITestOutputHelper log, + ClusterEndpointType preferred, + bool announceHostname = true, + AnnouncedAddress announced = AnnouncedAddress.Address, + System.Version? version = null) + { + var server = new InProcessTestServer(log) { ServerType = ServerType.Cluster, PreferredEndpointType = preferred }; + if (version is not null) server.RedisVersion = version; + if (announceHostname) server.SetHostname(server.DefaultEndPoint, Hostname); + if (announced != AnnouncedAddress.Address) server.SetAnnouncedAddress(server.DefaultEndPoint, announced); + return server; + } + + /// Returns the [endpoint, port, id, metadata] node block of the first slot range. + private static async Task GetFirstNodeAsync(InProcessTestServer server) + { + await using var conn = await server.ConnectAsync(defaultOnly: true); + var slots = await conn.GetServer(server.DefaultEndPoint).ExecuteAsync("cluster", "slots"); + + // each range is [from, to, master, replica...]; we want the master block of the first range + var ranges = (RedisResult[])slots!; + var firstRange = (RedisResult[])ranges[0]!; + return (RedisResult[])firstRange[2]!; + } + + /// + /// Returns the master node block of every slot range, keyed by port, as described by + /// - which node answers matters, since the preference is theirs. + /// + private static async Task> GetNodesByPortAsync(InProcessTestServer server, EndPoint? askOf = null) + { + await using var conn = await server.ConnectAsync(); + var slots = await conn.GetServer(askOf ?? server.DefaultEndPoint).ExecuteAsync("cluster", "slots"); + + var result = new Dictionary(); + foreach (var rangeResult in (RedisResult[])slots!) + { + var range = (RedisResult[])rangeResult!; + var node = (RedisResult[])range[2]!; + result[(int)node[1]] = node; + } + return result; + } + + private static string? GetMetadata(RedisResult[] node, string key) + { + var metadata = (RedisResult[])node[3]!; + for (int i = 0; i + 1 < metadata.Length; i += 2) + { + // the contract renders these keys inconsistently, so match without regard to case + if (string.Equals((string?)metadata[i], key, System.StringComparison.OrdinalIgnoreCase)) + { + return (string?)metadata[i + 1]; + } + } + return null; + } + + [Fact] + public async Task IpPreferredPutsAddressFirstAndHostnameInMetadata() + { + using var server = CreateServer(log, ClusterEndpointType.Ip); + var host = GetHost(server.DefaultEndPoint, out _); + + var node = await GetFirstNodeAsync(server); + + Assert.Equal(host, (string?)node[0]); + Assert.Equal(Hostname, GetMetadata(node, "hostname")); + Assert.Null(GetMetadata(node, "ip")); // the complement rule: no ip, it is already the primary + } + + [Fact] + public async Task HostnamePreferredPutsHostnameFirstAndAddressInMetadata() + { + using var server = CreateServer(log, ClusterEndpointType.Hostname); + var host = GetHost(server.DefaultEndPoint, out _); + + var node = await GetFirstNodeAsync(server); + + Assert.Equal(Hostname, (string?)node[0]); + Assert.Equal(host, GetMetadata(node, "ip")); + Assert.Null(GetMetadata(node, "hostname")); + } + + [Fact] + public async Task HostnamePreferredButUnannouncedReportsQuestionMark() + { + using var server = CreateServer(log, ClusterEndpointType.Hostname, announceHostname: false); + + var node = await GetFirstNodeAsync(server); + + // "?" means an unknown node - explicitly *not* "the node serving this command" + Assert.Equal("?", (string?)node[0]); + } + + [Fact] + public async Task UnknownEndpointPreferredReportsNull() + { + using var server = CreateServer(log, ClusterEndpointType.UnknownEndpoint); + var host = GetHost(server.DefaultEndPoint, out _); + + var node = await GetFirstNodeAsync(server); + + Assert.True(node[0].IsNull); + + // both forms move into the metadata, since neither is the primary + Assert.Equal(host, GetMetadata(node, "ip")); + Assert.Equal(Hostname, GetMetadata(node, "hostname")); + } + + [Fact] + public async Task NodeThatDoesNotKnowItsAddressReportsEmpty() + { + using var server = CreateServer(log, ClusterEndpointType.Ip, announced: AnnouncedAddress.Empty); + + var node = await GetFirstNodeAsync(server); + + Assert.Equal("", (string?)node[0]); + } + + [Fact] + public async Task NodeWithNoAddressKnownToTheServerReportsNull() + { + using var server = CreateServer(log, ClusterEndpointType.Ip, announced: AnnouncedAddress.Null); + + var node = await GetFirstNodeAsync(server); + + Assert.True(node[0].IsNull); + } + + [Fact] + public async Task EmptyAddressAlsoAppliesToIpMetadata() + { + using var server = CreateServer(log, ClusterEndpointType.Hostname, announced: AnnouncedAddress.Empty); + + var node = await GetFirstNodeAsync(server); + + Assert.Equal(Hostname, (string?)node[0]); + Assert.Equal("", GetMetadata(node, "ip")); + } + + [Fact] + public async Task PreSevenZeroServerReportsAddressesOnlyAndNoMetadata() + { + // hostname preferred *and* announced, but the server predates both + using var server = CreateServer(log, ClusterEndpointType.Hostname, version: BeforeHostnames); + var host = GetHost(server.DefaultEndPoint, out _); + + var node = await GetFirstNodeAsync(server); + + Assert.Equal(host, (string?)node[0]); + Assert.Equal(3, node.Length); // the metadata element itself only exists from 7.0 + } + + [Fact] + public async Task PreSevenZeroServerOmitsHostnameFromClusterNodes() + { + using var server = CreateServer(log, ClusterEndpointType.Hostname, version: BeforeHostnames); + + await using var conn = await server.ConnectAsync(defaultOnly: true); + var raw = await conn.GetServer(server.DefaultEndPoint).ClusterNodesRawAsync(); + Assert.NotNull(raw); + log.WriteLine(raw); + + Assert.DoesNotContain(Hostname, raw); + } + + [Fact] + public async Task NameOnlySeededServerPrefersHostnameAndHasNoAddress() + { + // seeding with a DnsEndPoint is only coherent as hostname-preferred: there is no address to report + using var server = new InProcessTestServer(log, new DnsEndPoint(Hostname, 6379)) + { + ServerType = ServerType.Cluster, + }; + // the preference lands on the node, not the server: seeding by name says nothing about peers + Assert.True(server.TryGetNode(server.DefaultEndPoint, out var seeded)); + Assert.Equal(ClusterEndpointType.Hostname, seeded.PreferredEndpointType); + Assert.Equal(ClusterEndpointType.Ip, server.PreferredEndpointType); + + var node = await GetFirstNodeAsync(server); + + Assert.Equal(Hostname, (string?)node[0]); + Assert.Equal("", GetMetadata(node, "ip")); + } + + [Fact] + public async Task AHostnamePreferringNodeDescribesEveryPeerByName() + { + // an address-keyed cluster in which one node prefers hostnames - a rolling config change is + // exactly this for a window. Verified against a real 8.9 cluster: the preference is the + // *answering* node's and applies to every entry in its reply + using var server = CreateServer(log, ClusterEndpointType.Ip); + var host = GetHost(server.DefaultEndPoint, out var defaultPort); + var outlier = server.AddEmptyNode(new IPEndPoint(IPAddress.Loopback, defaultPort + 1)); + server.Migrate((RedisKey)"outlier-slot-key", outlier); + server.SetPreferredEndpointType(outlier, ClusterEndpointType.Hostname); + + // asked of the majority node: everything by address, the announced hostname as metadata + var viaAddress = await GetNodesByPortAsync(server, server.DefaultEndPoint); + Assert.Equal(host, (string?)viaAddress[defaultPort][0]); + Assert.Equal(host, (string?)viaAddress[defaultPort + 1][0]); + Assert.Equal(Hostname, GetMetadata(viaAddress[defaultPort], "hostname")); + + // asked of the outlier: everything by name, and the peer with no hostname of its own is "?" - + // note that includes the outlier describing itself + var viaName = await GetNodesByPortAsync(server, outlier); + Assert.Equal(Hostname, (string?)viaName[defaultPort][0]); + Assert.Equal("?", (string?)viaName[defaultPort + 1][0]); + Assert.Equal(host, GetMetadata(viaName[defaultPort], "ip")); + Assert.Equal(host, GetMetadata(viaName[defaultPort + 1], "ip")); + } + + [Fact] + public async Task AnAddressPreferringNodeDescribesEveryPeerByAddress() + { + // the mirror: a hostname-preferring cluster with one node that still reports addresses + using var server = CreateServer(log, ClusterEndpointType.Hostname); + var host = GetHost(server.DefaultEndPoint, out var defaultPort); + var outlier = server.AddEmptyNode(new IPEndPoint(IPAddress.Loopback, defaultPort + 1)); + server.Migrate((RedisKey)"outlier-slot-key", outlier); + server.SetPreferredEndpointType(outlier, ClusterEndpointType.Ip); + server.SetHostname(outlier, "outlier.redis.example.com"); + + var viaName = await GetNodesByPortAsync(server, server.DefaultEndPoint); + Assert.Equal(Hostname, (string?)viaName[defaultPort][0]); + Assert.Equal("outlier.redis.example.com", (string?)viaName[defaultPort + 1][0]); + + // the outlier reports addresses for everyone, with both hostnames moved into the metadata + var viaAddress = await GetNodesByPortAsync(server, outlier); + Assert.Equal(host, (string?)viaAddress[defaultPort][0]); + Assert.Equal(host, (string?)viaAddress[defaultPort + 1][0]); + Assert.Equal(Hostname, GetMetadata(viaAddress[defaultPort], "hostname")); + Assert.Equal("outlier.redis.example.com", GetMetadata(viaAddress[defaultPort + 1], "hostname")); + } + + [Fact] + public async Task ConfigGetReportsTheAnsweringNodesOwnView() + { + using var server = CreateServer(log, ClusterEndpointType.Ip); + GetHost(server.DefaultEndPoint, out var defaultPort); + var outlier = server.AddEmptyNode(new DnsEndPoint("outlier.redis.example.com", defaultPort + 1)); + server.Migrate((RedisKey)"outlier-slot-key", outlier); + + await using var conn = await server.ConnectAsync(); + + // the parameter is per-node, so each connection sees its own node's answer + Assert.Equal("ip", await GetConfigAsync(conn.GetServer(server.DefaultEndPoint), "cluster-preferred-endpoint-type")); + Assert.Equal("hostname", await GetConfigAsync(conn.GetServer(outlier), "cluster-preferred-endpoint-type")); + Assert.Equal("outlier.redis.example.com", await GetConfigAsync(conn.GetServer(outlier), "cluster-announce-hostname")); + } + + [Theory] + [InlineData(ClusterEndpointType.Ip, "ip")] + [InlineData(ClusterEndpointType.Hostname, "hostname")] + [InlineData(ClusterEndpointType.UnknownEndpoint, "unknown-endpoint")] + public async Task ConfigGetReportsTheAnnounceSettings(ClusterEndpointType preferred, string expected) + { + using var server = CreateServer(log, preferred); + + await using var conn = await server.ConnectAsync(defaultOnly: true); + var api = conn.GetServer(server.DefaultEndPoint); + + Assert.Equal(expected, await GetConfigAsync(api, "cluster-preferred-endpoint-type")); + Assert.Equal(Hostname, await GetConfigAsync(api, "cluster-announce-hostname")); + } + + [Fact] + public async Task ConfigGetOmitsAnnounceSettingsBeforeSevenZero() + { + using var server = CreateServer(log, ClusterEndpointType.Hostname, version: BeforeHostnames); + + await using var conn = await server.ConnectAsync(defaultOnly: true); + var api = conn.GetServer(server.DefaultEndPoint); + + Assert.Null(await GetConfigAsync(api, "cluster-preferred-endpoint-type")); + Assert.Null(await GetConfigAsync(api, "cluster-announce-hostname")); + } + + private static async Task GetConfigAsync(IServer api, string key) + { + foreach (var pair in await api.ConfigGetAsync(key)) + { + if (string.Equals(pair.Key, key, System.StringComparison.OrdinalIgnoreCase)) return pair.Value; + } + return null; + } + + [Theory] + [InlineData(ClusterEndpointType.Ip, true, "127.0.0.1")] + [InlineData(ClusterEndpointType.Hostname, true, Hostname)] + [InlineData(ClusterEndpointType.Hostname, false, "?")] + [InlineData(ClusterEndpointType.UnknownEndpoint, true, "")] + public async Task MovedRedirectUsesThePreferredForm(ClusterEndpointType preferred, bool announceHostname, string expectedHost) + { + using var server = CreateServer(log, preferred, announceHostname); + + // connect *before* migrating, so the client's cached slot map still points at the old owner + // and the command actually earns a redirect + await using var conn = await server.ConnectAsync(defaultOnly: true); + + var other = server.AddEmptyNode(); + if (announceHostname) server.SetHostname(other, "host-2.redis.example.com"); + server.Migrate((RedisKey)"moved-form-key", other); + GetHost(other, out var otherPort); + + // NoRedirect so the error surfaces rather than being followed + var ex = await Assert.ThrowsAsync( + async () => await conn.GetDatabase().StringGetAsync("moved-form-key", CommandFlags.NoRedirect)); + log.WriteLine(ex.Message); + + var expected = preferred == ClusterEndpointType.Hostname && announceHostname + ? "host-2.redis.example.com" + : expectedHost; + + if (expected is "?" or "") + { + // nothing usable to route to; UnroutableRedirectUnitTests covers the handling, here we only + // care that the server named the target the way its preference dictates + Assert.Equal(RedisErrorKind.UnknownRedirectTarget, ex.Kind); + Assert.Contains($"'{expected}:{otherPort}'", ex.Message); + } + else + { + // the client restates the redirect in terms of the endpoint it parsed, so assert on that rather + // than on the wire text; note a hostname form lands as a DnsEndPoint, which is the identity + // hazard this fake exists to make reproducible + Assert.Contains("MOVED", ex.Message); + Assert.Contains($"{expected}:{otherPort}", ex.Message); + } + } +} diff --git a/tests/StackExchange.Redis.Tests/ClusterNodeParseTests.cs b/tests/StackExchange.Redis.Tests/ClusterNodeParseTests.cs new file mode 100644 index 000000000..cccf08b1d --- /dev/null +++ b/tests/StackExchange.Redis.Tests/ClusterNodeParseTests.cs @@ -0,0 +1,114 @@ +using System; +using System.Linq; +using System.Threading.Tasks; +using Xunit; + +namespace StackExchange.Redis.Tests; + +/// +/// End-to-end coverage for the parsed CLUSTER NODES members against a real cluster; the deterministic +/// grammar cases live in . Runs per protocol deliberately: under RESP3 +/// this reply arrives as a verbatim string with a txt: prefix, which is its own parsing hazard. +/// +[RunPerProtocol] +public class ClusterNodeParseTests(ITestOutputHelper output, SharedConnectionFixture fixture) : TestBase(output, fixture) +{ + protected override string GetConfiguration() => TestConfig.Current.ClusterServersAndPorts + ",connectTimeout=10000"; + + private static readonly Version BusPortVersion = new(4, 0, 0); // "@cport" arrives here + private static readonly Version HostnameVersion = new(7, 0, 0); // hostnames arrive here + + /// The endpoint token's trailer - "cport[,hostname[,aux=value]*]" - taken straight from the raw line. + private static string? GetRawTrailer(ClusterNode node) + { + var endpointToken = node.Raw.Split(' ')[1]; + var at = endpointToken.IndexOf('@'); + return at < 0 ? null : endpointToken.Substring(at + 1); + } + + [Fact] + public async Task ParsedMembersAgreeWithTheRawLine() + { + await using var conn = Create(allowAdmin: true); + var api = conn.GetServer(conn.GetEndPoints()[0]); + Assert.SkipUnless(api.Version >= BusPortVersion, $"cluster bus port needs {BusPortVersion}, server is {api.Version}"); + + var config = await api.ClusterNodesAsync(); + Assert.NotNull(config); + Assert.NotEmpty(config.Nodes); + + foreach (var node in config.Nodes) + { + Log(node.Raw); + var trailer = GetRawTrailer(node); + Assert.NotNull(trailer); // 4.0+, so every line carries one + + var fields = trailer.Split(','); + Assert.Equal(int.Parse(fields[0]), node.ClusterBusPort); + + // whatever this deployment announces, the parsed view must agree with the text + var expectedHostname = fields.Length > 1 && fields[1].Length > 0 ? fields[1] : null; + Assert.Equal(expectedHostname, node.Hostname); + Assert.Equal(Math.Max(fields.Length - 2, 0), node.AuxFields.Count); + } + } + + [Fact] + public async Task BusPortFollowsTheConventionalOffset() + { + await using var conn = Create(allowAdmin: true); + var api = conn.GetServer(conn.GetEndPoints()[0]); + Assert.SkipUnless(api.Version >= BusPortVersion, $"cluster bus port needs {BusPortVersion}, server is {api.Version}"); + + var config = await api.ClusterNodesAsync(); + Assert.NotNull(config); + + foreach (var node in config.Nodes) + { + // not guaranteed by the protocol, but it is what an unconfigured cluster does, and it is the + // assumption the toy server encodes - so worth asserting against reality rather than inferring + var port = node.EndPoint switch + { + System.Net.IPEndPoint ip => ip.Port, + System.Net.DnsEndPoint dns => dns.Port, + _ => 0, + }; + Log($"{node.EndPoint} bus {node.ClusterBusPort}"); + Assert.Equal(port + 10000, node.ClusterBusPort); + } + } + + [Fact] + public async Task AnnouncedHostnameIsParsed() + { + await using var conn = Create(allowAdmin: true); + var endpoint = conn.GetEndPoints()[0]; + var api = conn.GetServer(endpoint); + Assert.SkipUnless(api.Version >= HostnameVersion, $"hostnames need {HostnameVersion}, server is {api.Version}"); + + const string Hostname = "se-redis-test.example.com"; + const string Key = "cluster-announce-hostname"; + + var before = (await api.ConfigGetAsync(Key)).SingleOrDefault().Value ?? ""; + Log($"restoring '{Key}' to '{before}' afterwards"); + try + { + await api.ConfigSetAsync(Key, Hostname); + + var config = await api.ClusterNodesAsync(); + Assert.NotNull(config); + + // the node we asked is the one that announced it; peers learn it by gossip, so do not race on them + var self = Assert.Single(config.Nodes, x => x.IsMyself); + Log(self.Raw); + Assert.Equal(Hostname, self.Hostname); + + // and it remains an *additional* identity - the endpoint is still the address + Assert.Equal(endpoint, self.EndPoint); + } + finally + { + await api.ConfigSetAsync(Key, before); + } + } +} diff --git a/tests/StackExchange.Redis.Tests/ClusterNodeParseUnitTests.cs b/tests/StackExchange.Redis.Tests/ClusterNodeParseUnitTests.cs new file mode 100644 index 000000000..8cade9fb8 --- /dev/null +++ b/tests/StackExchange.Redis.Tests/ClusterNodeParseUnitTests.cs @@ -0,0 +1,152 @@ +using System.Linq; +using System.Net; +using System.Threading.Tasks; +using Xunit; + +namespace StackExchange.Redis.Tests; + +/// +/// The parsed view of CLUSTER NODES, specifically the part after the @ that we used to +/// truncate away: the cluster bus port, the announced hostname, and any auxiliary fields. Documented form +/// is ip:port@cport[,hostname[,aux-field=value]*]. +/// +public class ClusterNodeParseUnitTests(ITestOutputHelper log) +{ + private const string Hostname = "host-1.redis.example.com"; + private static readonly EndPoint Origin = new IPEndPoint(IPAddress.Loopback, 7000); + + private static ClusterNode Parse(string line) + { + var config = new ClusterConfiguration(serverSelectionStrategy: null!, line, Origin); + return config.Nodes.Single(); + } + + private const string Flags = " myself,master - 0 0 1 connected 0-16383"; + + [Fact] + public void BusPortIsParsed() + { + var node = Parse("abc 127.0.0.1:7000@17000" + Flags); + + Assert.Equal(new IPEndPoint(IPAddress.Loopback, 7000), node.EndPoint); + Assert.Equal(17000, node.ClusterBusPort); + Assert.Null(node.Hostname); + Assert.Empty(node.AuxFields); + } + + [Fact] + public void PreFourZeroLineHasNoBusPort() + { + // servers older than 4.0 report no "@cport" at all + var node = Parse("abc 127.0.0.1:7000" + Flags); + + Assert.Equal(new IPEndPoint(IPAddress.Loopback, 7000), node.EndPoint); + Assert.Null(node.ClusterBusPort); + Assert.Null(node.Hostname); + Assert.Empty(node.AuxFields); + } + + [Fact] + public void HostnameIsParsed() + { + var node = Parse($"abc 127.0.0.1:7000@17000,{Hostname}" + Flags); + + Assert.Equal(17000, node.ClusterBusPort); + Assert.Equal(Hostname, node.Hostname); + + // the hostname is an additional identity, not a replacement for the address + Assert.Equal(new IPEndPoint(IPAddress.Loopback, 7000), node.EndPoint); + } + + [Fact] + public void AuxFieldsAreParsed() + { + var node = Parse($"abc 127.0.0.1:7000@17000,{Hostname},shard-id=abc123,human-nodename=alpha" + Flags); + + Assert.Equal(Hostname, node.Hostname); + Assert.Collection( + node.AuxFields, + x => Assert.Equal(new("shard-id", "abc123"), x), + x => Assert.Equal(new("human-nodename", "alpha"), x)); + } + + [Fact] + public void AuxFieldsSurviveAnEmptyHostnameSlot() + { + // the hostname slot is positional, so it can be empty while aux fields follow it + var node = Parse("abc 127.0.0.1:7000@17000,,shard-id=abc123" + Flags); + + Assert.Null(node.Hostname); + Assert.Equal(new("shard-id", "abc123"), Assert.Single(node.AuxFields)); + } + + [Fact] + public void UnrecognizedAuxFieldsArePreserved() + { + // the set is documented as extensible, so a field we have never heard of must round-trip + var node = Parse($"abc 127.0.0.1:7000@17000,{Hostname},not-invented-yet=42" + Flags); + + Assert.Equal(new("not-invented-yet", "42"), Assert.Single(node.AuxFields)); + } + + [Fact] + public void MalformedTrailerDoesNotThrow() + { + // an exception here does silent damage to topology, so parsing is deliberately lenient + var node = Parse($"abc 127.0.0.1:7000@not-a-port,{Hostname},no-equals-sign,=novalue,k=v" + Flags); + + Assert.Null(node.ClusterBusPort); + Assert.Equal(Hostname, node.Hostname); + Assert.Equal(new("k", "v"), Assert.Single(node.AuxFields)); + } + + [Theory] + [InlineData(false, false)] + [InlineData(true, false)] + [InlineData(true, true)] + public async Task RoundTripsThroughTheServer(bool announceHostname, bool auxFields) + { + using var server = new InProcessTestServer(log) { ServerType = ServerType.Cluster }; + var endpoint = server.DefaultEndPoint; + if (announceHostname) server.SetHostname(endpoint, Hostname); + if (auxFields) server.SetAuxField(endpoint, "shard-id", "abc123"); + + await using var conn = await server.ConnectAsync(defaultOnly: true); + var config = await conn.GetServer(endpoint).ClusterNodesAsync(); + Assert.NotNull(config); + + var node = config.Nodes.Single(); + log.WriteLine(node.Raw); + + Server.RedisServer.GetHost(endpoint, out var port); + Assert.Equal(port + 10000, node.ClusterBusPort); + Assert.Equal(announceHostname ? Hostname : null, node.Hostname); + if (auxFields) + { + Assert.Equal(new("shard-id", "abc123"), Assert.Single(node.AuxFields)); + } + else + { + Assert.Empty(node.AuxFields); + } + } + + [Fact] + public async Task PreSevenZeroServerReportsNoHostname() + { + using var server = new InProcessTestServer(log) + { + ServerType = ServerType.Cluster, + RedisVersion = new System.Version(6, 2, 0), + }; + server.SetHostname(server.DefaultEndPoint, Hostname); + + await using var conn = await server.ConnectAsync(defaultOnly: true); + var config = await conn.GetServer(server.DefaultEndPoint).ClusterNodesAsync(); + Assert.NotNull(config); + + var node = config.Nodes.Single(); + Assert.Null(node.Hostname); + Assert.Equal(6379 + 10000, node.ClusterBusPort); // the cport predates hostnames + } +} diff --git a/tests/StackExchange.Redis.Tests/ClusterSlotsTests.cs b/tests/StackExchange.Redis.Tests/ClusterSlotsTests.cs new file mode 100644 index 000000000..e9b37185e --- /dev/null +++ b/tests/StackExchange.Redis.Tests/ClusterSlotsTests.cs @@ -0,0 +1,167 @@ +using System; +using System.IO; +using System.IO.Compression; +using System.Linq; +using System.Threading.Tasks; +using Xunit; + +namespace StackExchange.Redis.Tests; + +/// +/// CLUSTER SLOTS against a real cluster; the naming-configuration matrix lives in +/// . Runs per protocol, since the metadata element is a map under RESP3 +/// and a flat array under RESP2. +/// +[RunPerProtocol] +public class ClusterSlotsTests(ITestOutputHelper output, SharedConnectionFixture fixture) : TestBase(output, fixture) +{ + protected override string GetConfiguration() => TestConfig.Current.ClusterServersAndPorts + ",connectTimeout=10000"; + + private static readonly Version NodeIdVersion = new(4, 0, 0); + + [Fact] + public async Task AssignmentsCoverTheWholeKeyspace() + { + await using var conn = Create(allowAdmin: true); + var result = await conn.GetServer(conn.GetEndPoints()[0]).ClusterSlotsAsync(); + Assert.NotNull(result); + Assert.NotEmpty(result.Assignments); + + var covered = result.Assignments.Sum(x => x.Slots.To - x.Slots.From + 1); + Log($"{result.Assignments.Count} assignments covering {covered} slots"); + Assert.Equal(SlotRange.MaxSlot - SlotRange.MinSlot + 1, covered); + } + + [Fact] + public async Task EveryPrimaryIsAddressableAndIdentified() + { + await using var conn = Create(allowAdmin: true); + var api = conn.GetServer(conn.GetEndPoints()[0]); + Assert.SkipUnless(api.Version >= NodeIdVersion, $"node ids need {NodeIdVersion}, server is {api.Version}"); + + var result = await api.ClusterSlotsAsync(); + Assert.NotNull(result); + + foreach (var assignment in result.Assignments) + { + foreach (var node in new[] { assignment.Primary }.Concat(assignment.Replicas)) + { + // an unconfigured cluster prefers addresses, so every node should be usable as-is + Assert.NotNull(node.EndPoint); + Assert.False(string.IsNullOrEmpty(node.NodeId)); + Assert.InRange(node.Port, 1, ushort.MaxValue); + } + } + } + + [Fact] + public async Task ReplicasAreReported() + { + await using var conn = Create(allowAdmin: true); + var result = await conn.GetServer(conn.GetEndPoints()[0]).ClusterSlotsAsync(); + Assert.NotNull(result); + + // whether replicas exist at all is a property of the deployment, not of the parser - the Windows CI + // fleet is smaller than the local compose - so skip rather than fail where there are none + var withReplicas = result.Assignments.Count(x => x.Replicas.Count > 0); + Log($"{withReplicas} of {result.Assignments.Count} assignments report replicas"); + Assert.SkipWhen(withReplicas == 0, "this deployment reports no replicas"); + + var replica = result.Assignments.First(x => x.Replicas.Count > 0).Replicas[0]; + Assert.NotNull(replica.EndPoint); + Assert.NotEqual( + result.Assignments.First(x => x.Replicas.Count > 0).Primary.NodeId, + replica.NodeId); + } + + [Fact] + public async Task ExportIncludesBothClusterViews() + { + var path = Path.Combine(Path.GetTempPath(), $"se-redis-export-{Guid.NewGuid():n}.zip"); + try + { + await using (var conn = Create(allowAdmin: true)) + using (var file = File.Create(path)) + { + conn.ExportConfiguration(file, ExportOptions.Cluster); + } + + using var zip = ZipFile.OpenRead(path); + var names = zip.Entries.Select(x => x.FullName).ToArray(); + Log(string.Join(", ", names)); + + // the export is per-server, so there should be one of each alongside every other + var nodeFiles = names.Count(x => x.EndsWith("/nodes.txt", StringComparison.Ordinal)); + var slotFiles = names.Count(x => x.EndsWith("/slots.txt", StringComparison.Ordinal)); + Assert.NotEqual(0, nodeFiles); + Assert.Equal(nodeFiles, slotFiles); + + var slots = zip.Entries.First(x => x.FullName.EndsWith("/slots.txt", StringComparison.Ordinal)); + using var reader = new StreamReader(slots.Open()); + var content = await reader.ReadToEndAsync(); + Log(content); + + // one line per node per range, endpoint reported exactly as the server gave it + Assert.Contains(" primary endpoint=", content); + Assert.Contains(" id=", content); + } + finally + { + if (File.Exists(path)) File.Delete(path); + } + } + + [Fact] + public async Task ShadowTopologyAgreesWithNodesPerNode() + { + // the live cluster has heavily fragmented ownership, so this compares real range boundaries rather + // than the toy server's tidy ones - the strongest check available before the slot map is switched over + await using var conn = Create(allowAdmin: true); + var endpoint = conn.GetEndPoints()[0]; + var api = conn.GetServer(endpoint); + Assert.SkipUnless(api.Version >= NodeIdVersion, $"node ids need {NodeIdVersion}, server is {api.Version}"); + + var nodes = await api.ClusterNodesAsync(); + Assert.NotNull(nodes); + + var topology = ClusterTopology.From(await api.ClusterSlotsAsync()); + Assert.NotNull(topology); + + int compared = 0; + foreach (var node in topology.Nodes.Where(x => !x.IsReplica)) + { + var expected = Slots(nodes.Nodes.Single(x => x.NodeId == node.NodeId).Slots); + var actual = Slots(node.Slots); + Log($"{node.NodeId}: {actual.Length} slots over {node.Slots.Count} ranges"); + Assert.Equal(expected, actual); + compared++; + } + Assert.NotEqual(0, compared); + + 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 SlotsAndNodesAgreeOnPrimaries() + { + await using var conn = Create(allowAdmin: true); + var api = conn.GetServer(conn.GetEndPoints()[0]); + Assert.SkipUnless(api.Version >= NodeIdVersion, $"node ids need {NodeIdVersion}, server is {api.Version}"); + + var slots = await api.ClusterSlotsAsync(); + var nodes = await api.ClusterNodesAsync(); + Assert.NotNull(slots); + Assert.NotNull(nodes); + + // node-id is the identity that does not depend on rendering, so the two views must agree on it - + // this is the premise that reconciliation keyed on the id relies on + var fromSlots = slots.Assignments.Select(x => x.Primary.NodeId!).Distinct().OrderBy(x => x).ToArray(); + var fromNodes = nodes.Nodes.Where(x => !x.IsReplica && x.Slots.Count > 0) + .Select(x => x.NodeId).Distinct().OrderBy(x => x).ToArray(); + + Log($"SLOTS: {string.Join(",", fromSlots)}"); + Log($"NODES: {string.Join(",", fromNodes)}"); + Assert.Equal(fromNodes, fromSlots); + } +} diff --git a/tests/StackExchange.Redis.Tests/ClusterSlotsUnitTests.cs b/tests/StackExchange.Redis.Tests/ClusterSlotsUnitTests.cs new file mode 100644 index 000000000..4ba13f352 --- /dev/null +++ b/tests/StackExchange.Redis.Tests/ClusterSlotsUnitTests.cs @@ -0,0 +1,237 @@ +using System.Linq; +using System.Net; +using System.Threading.Tasks; +using Xunit; +using static StackExchange.Redis.Server.RedisServer; + +namespace StackExchange.Redis.Tests; + +/// +/// The parsed CLUSTER SLOTS view, exercised against the toy server across the naming configurations a +/// real server can be in - including the placeholder endpoint values, which are the classic client bug. +/// +public class ClusterSlotsUnitTests(ITestOutputHelper log) +{ + private const string Hostname = "host-1.redis.example.com"; + private static readonly System.Version BeforeHostnames = new(6, 2, 0); + + private static InProcessTestServer CreateServer( + ITestOutputHelper log, + ClusterEndpointType preferred = ClusterEndpointType.Ip, + bool announceHostname = true, + AnnouncedAddress announced = AnnouncedAddress.Address, + System.Version? version = null) + { + var server = new InProcessTestServer(log) { ServerType = ServerType.Cluster, PreferredEndpointType = preferred }; + if (version is not null) server.RedisVersion = version; + if (announceHostname) server.SetHostname(server.DefaultEndPoint, Hostname); + if (announced != AnnouncedAddress.Address) server.SetAnnouncedAddress(server.DefaultEndPoint, announced); + return server; + } + + private static async Task GetPrimaryAsync(InProcessTestServer server) + { + await using var conn = await server.ConnectAsync(defaultOnly: true); + var result = await conn.GetServer(server.DefaultEndPoint).ClusterSlotsAsync(); + Assert.NotNull(result); + return Assert.Single(result.Assignments).Primary; + } + + [Fact] + public void NodeIdsAreUniqueEvenWhenCreatedInTheSameTick() + { + // regression: the toy server created a Random per id, and .NET Framework seeds that from the tick + // count - so nodes created in the same tick shared an id. Keyed reconciliation then merged two nodes + // into one, which surfaced as the *client* looking wrong + using var server = new InProcessTestServer(log) { ServerType = ServerType.Cluster }; + var ids = new System.Collections.Generic.HashSet(System.StringComparer.Ordinal); + + GetHost(server.DefaultEndPoint, out var port); + Assert.True(server.TryGetNode(server.DefaultEndPoint, out var first)); + Assert.True(ids.Add(first.Id)); + + for (int i = 1; i <= 25; i++) + { + var endpoint = server.AddEmptyNode(new IPEndPoint(IPAddress.Loopback, port + i)); + Assert.True(server.TryGetNode(endpoint, out var node)); + Assert.True(ids.Add(node.Id), $"duplicate id at node {i}: {node.Id}"); + } + } + + [Fact] + public async Task WholeKeyspaceIsReportedWithNodeIdAndEndpoint() + { + using var server = CreateServer(log, announceHostname: false); + var host = GetHost(server.DefaultEndPoint, out var port); + + await using var conn = await server.ConnectAsync(defaultOnly: true); + var result = await conn.GetServer(server.DefaultEndPoint).ClusterSlotsAsync(); + Assert.NotNull(result); + + var assignment = Assert.Single(result.Assignments); + Assert.Equal(0, assignment.Slots.From); + Assert.Equal(16383, assignment.Slots.To); + Assert.Empty(assignment.Replicas); + + var primary = assignment.Primary; + Assert.Equal(host, primary.AnnouncedEndpoint); + Assert.Equal(port, primary.Port); + Assert.Equal(new IPEndPoint(IPAddress.Loopback, port), primary.EndPoint); + Assert.False(string.IsNullOrEmpty(primary.NodeId)); + Assert.Empty(primary.Metadata); + } + + [Fact] + public async Task IpPreferredSurfacesHostnameFromMetadata() + { + using var server = CreateServer(log, ClusterEndpointType.Ip); + var host = GetHost(server.DefaultEndPoint, out var port); + + var primary = await GetPrimaryAsync(server); + + Assert.Equal(host, primary.AnnouncedEndpoint); + Assert.Equal(new IPEndPoint(IPAddress.Loopback, port), primary.EndPoint); + Assert.Equal(Hostname, primary.Hostname); + Assert.Null(primary.Ip); // the complement rule: it is already the primary field + } + + [Fact] + public async Task HostnamePreferredParsesAsADnsEndPointAndSurfacesIp() + { + using var server = CreateServer(log, ClusterEndpointType.Hostname); + var host = GetHost(server.DefaultEndPoint, out var port); + + var primary = await GetPrimaryAsync(server); + + Assert.Equal(Hostname, primary.AnnouncedEndpoint); + Assert.Equal(new DnsEndPoint(Hostname, port), primary.EndPoint); + Assert.Equal(host, primary.Ip); + Assert.Null(primary.Hostname); + } + + [Fact] + public async Task UnannouncedHostnameYieldsNoEndpointButKeepsTheIp() + { + using var server = CreateServer(log, ClusterEndpointType.Hostname, announceHostname: false); + var host = GetHost(server.DefaultEndPoint, out _); + + var primary = await GetPrimaryAsync(server); + + // "?" is an explicitly unknown node, so it must not become an endpoint... + Assert.Equal("?", primary.AnnouncedEndpoint); + Assert.Null(primary.EndPoint); + + // ...but the reply still carries the address, so the union is not poorer than CLUSTER NODES + Assert.Equal(host, primary.Ip); + } + + [Fact] + public async Task NullEndpointYieldsNoEndpoint() + { + using var server = CreateServer(log, ClusterEndpointType.UnknownEndpoint); + var host = GetHost(server.DefaultEndPoint, out var port); + + var primary = await GetPrimaryAsync(server); + + // null means "connect to where you sent this, with this port" - a caller decision, not ours + Assert.Null(primary.AnnouncedEndpoint); + Assert.Null(primary.EndPoint); + Assert.Equal(port, primary.Port); + Assert.Equal(host, primary.Ip); + Assert.Equal(Hostname, primary.Hostname); + } + + [Fact] + public async Task EmptyEndpointYieldsNoEndpoint() + { + using var server = CreateServer(log, ClusterEndpointType.Ip, announced: AnnouncedAddress.Empty); + + var primary = await GetPrimaryAsync(server); + + Assert.Equal("", primary.AnnouncedEndpoint); + Assert.Null(primary.EndPoint); + } + + [Fact] + public async Task RecognizedKeysAreSurfacedAndUnknownOnesPreserved() + { + // known keys are matched over the raw bytes and surfaced as properties, so they cost no allocation + // and do not appear in Metadata; the extensible remainder is kept as declared + using var server = CreateServer(log, ClusterEndpointType.Hostname); + server.SetSlotsMetadata(server.DefaultEndPoint, "not-invented-yet", "42"); + var host = GetHost(server.DefaultEndPoint, out _); + + var primary = await GetPrimaryAsync(server); + + Assert.Equal(host, primary.Ip); // recognized... + Assert.Equal(new("not-invented-yet", "42"), Assert.Single(primary.Metadata)); // ...and the rest kept + } + + [Fact] + public async Task MetadataKeysAreMatchedWithoutRegardToCase() + { + // the contract renders these keys inconsistently between prose and examples, so casing cannot be + // relied on; an upper-case key must still be recognized rather than landing in Metadata + using var server = CreateServer(log, ClusterEndpointType.Ip, announceHostname: false); + server.SetSlotsMetadata(server.DefaultEndPoint, "HOSTNAME", "shouty.redis.example.com"); + + var primary = await GetPrimaryAsync(server); + + Assert.Equal("shouty.redis.example.com", primary.Hostname); + Assert.Empty(primary.Metadata); + } + + [Fact] + public async Task PreSevenZeroServerReportsNoMetadata() + { + using var server = CreateServer(log, ClusterEndpointType.Hostname, version: BeforeHostnames); + var host = GetHost(server.DefaultEndPoint, out var port); + + var primary = await GetPrimaryAsync(server); + + // three-element node block: endpoint, port, id - and the preference is inert below 7.0 + Assert.Equal(host, primary.AnnouncedEndpoint); + Assert.Equal(new IPEndPoint(IPAddress.Loopback, port), primary.EndPoint); + Assert.Empty(primary.Metadata); + Assert.Null(primary.Hostname); + Assert.False(string.IsNullOrEmpty(primary.NodeId)); + } + + [Fact] + public async Task MigratedSlotsAreReportedAsSeparateAssignments() + { + 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)"slots-key", other); + + await using var conn = await server.ConnectAsync(); + var result = await conn.GetServer(server.DefaultEndPoint).ClusterSlotsAsync(); + Assert.NotNull(result); + + log.WriteLine(string.Join(", ", result.Assignments.Select(x => $"{x.Slots}=>{x.Primary}"))); + + // the migrated slot splits the original range, and every assignment names its own primary + Assert.True(result.Assignments.Count > 1); + Assert.All(result.Assignments, x => Assert.NotNull(x.Primary)); + Assert.Contains(result.Assignments, x => x.Primary.Port == port + 1); + + var migrated = Assert.Single(result.Assignments, x => x.Primary.Port == port + 1); + Assert.Equal(migrated.Slots.From, migrated.Slots.To); // exactly the one slot moved + } + + [Fact] + public async Task NodeIdIsStableAcrossTheNamingForms() + { + // node-id is the one identity that does not depend on how the answering node renders endpoints, + // which is what makes it the reliable reconciliation key + using var server = CreateServer(log, ClusterEndpointType.Ip); + var byAddress = await GetPrimaryAsync(server); + + server.PreferredEndpointType = ClusterEndpointType.Hostname; + var byName = await GetPrimaryAsync(server); + + Assert.NotEqual(byAddress.AnnouncedEndpoint, byName.AnnouncedEndpoint); + Assert.Equal(byAddress.NodeId, byName.NodeId); + } +} diff --git a/tests/StackExchange.Redis.Tests/ClusterTopologyShadowUnitTests.cs b/tests/StackExchange.Redis.Tests/ClusterTopologyShadowUnitTests.cs new file mode 100644 index 000000000..6249c987e --- /dev/null +++ b/tests/StackExchange.Redis.Tests/ClusterTopologyShadowUnitTests.cs @@ -0,0 +1,151 @@ +using System.Linq; +using System.Net; +using System.Threading.Tasks; +using Xunit; +using static StackExchange.Redis.Server.RedisServer; + +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. +/// +public class ClusterTopologyShadowUnitTests(ITestOutputHelper log) +{ + private const string Hostname = "host-1.redis.example.com"; + + private static InProcessTestServer CreateServer( + ITestOutputHelper log, + ClusterEndpointType preferred = ClusterEndpointType.Ip, + bool announceHostname = true) + { + var server = new InProcessTestServer(log) { ServerType = ServerType.Cluster, PreferredEndpointType = preferred }; + if (announceHostname) server.SetHostname(server.DefaultEndPoint, Hostname); + return server; + } + + /// + /// 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. + /// + private static async Task GetShadowAsync(IConnectionMultiplexer conn, EndPoint endpoint) + { + var slots = await conn.GetServer(endpoint).ClusterSlotsAsync(); + var topology = ClusterTopology.From(slots); + Assert.NotNull(topology); + return topology; + } + + [Theory] + [InlineData(ClusterEndpointType.Ip)] + [InlineData(ClusterEndpointType.Hostname)] + public async Task ShadowTopologyIsBuiltFromTheReply(ClusterEndpointType preferred) + { + using var server = CreateServer(log, preferred); + await using var conn = await server.ConnectAsync(defaultOnly: true); + + var topology = await GetShadowAsync(conn, server.DefaultEndPoint); + var node = Assert.Single(topology.Nodes); + log.WriteLine(node.ToString()); + + GetHost(server.DefaultEndPoint, out var port); + Assert.Equal(port, node.Port); + Assert.False(node.IsReplica); + Assert.False(string.IsNullOrEmpty(node.NodeId)); + Assert.Equal(0, node.Slots.Single().From); + Assert.Equal(16383, node.Slots.Single().To); + } + + [Theory] + [InlineData(ClusterEndpointType.Ip)] + [InlineData(ClusterEndpointType.Hostname)] + public async Task ShadowTopologyKnowsBothIdentities(ClusterEndpointType preferred) + { + // whichever form the answering node prefers, the complement arrives as metadata - so one reply is + // enough to know the node by both names, which is what the NODES-driven view cannot express + using var server = CreateServer(log, preferred); + await using var conn = await server.ConnectAsync(defaultOnly: true); + + var node = Assert.Single((await GetShadowAsync(conn, server.DefaultEndPoint)).Nodes); + var host = GetHost(server.DefaultEndPoint, out var port); + + Assert.Equal(host, node.Ip); + Assert.Equal(Hostname, node.Hostname); + Assert.Contains(new IPEndPoint(IPAddress.Loopback, port), node.Identities); + Assert.Contains(new DnsEndPoint(Hostname, port), node.Identities); + } + + [Fact] + public async Task ShadowTopologyAgreesWithTheRoutingView() + { + 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)"shadow-key", other); + + await using var conn = await server.ConnectAsync(); + var api = conn.GetServer(server.DefaultEndPoint); + + // force both views to be refreshed from the same server + var nodes = await api.ClusterNodesAsync(); + Assert.NotNull(nodes); + var topology = await GetShadowAsync(conn, server.DefaultEndPoint); + + var fromNodes = nodes.Nodes.Where(x => !x.IsReplica && x.Slots.Count > 0) + .Select(x => x.NodeId).OrderBy(x => x).ToArray(); + var fromShadow = topology.Nodes.Where(x => !x.IsReplica) + .Select(x => x.NodeId).OrderBy(x => x).ToArray(); + + log.WriteLine($"NODES: {string.Join(",", fromNodes)}"); + log.WriteLine($"SHADOW: {string.Join(",", fromShadow)}"); + Assert.Equal(fromNodes, fromShadow); + + // ...and the ranges agree per node, not merely in total: equal totals with different boundaries is + // exactly what an off-by-one in range application looks like, and it is the property routing depends + // on. Compared as sorted slot sets so that differing range *fragmentation* between the two views is + // not treated as disagreement - only differing ownership is + foreach (var node in topology.Nodes.Where(x => !x.IsReplica)) + { + var expected = Slots(nodes.Nodes.Single(x => x.NodeId == node.NodeId).Slots); + var actual = Slots(node.Slots); + log.WriteLine($"{node.NodeId}: {actual.Length} slots"); + Assert.Equal(expected, actual); + } + + 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 ShadowTopologyDoesNotChangeRouting() + { + // the whole point of shadow mode: recorded, compared, not acted upon + using var server = CreateServer(log, ClusterEndpointType.Hostname); + await using var conn = await server.ConnectAsync(defaultOnly: true); + + Assert.NotNull(await GetShadowAsync(conn, server.DefaultEndPoint)); + + // endpoints are still exactly what NODES gave us: addresses, not the preferred hostname form + Assert.All(conn.GetEndPoints(), ep => Assert.IsType(ep)); + await conn.GetDatabase().StringSetAsync("shadow-routing", "ok"); + Assert.Equal("ok", await conn.GetDatabase().StringGetAsync("shadow-routing")); + } + + [Fact] + public async Task PreFourZeroServerYieldsNoShadowTopology() + { + // no node ids to key on, so there is nothing we could reconcile; better absent than half-built + using var server = new InProcessTestServer(log) + { + ServerType = ServerType.Cluster, + RedisVersion = new System.Version(3, 2, 0), + }; + await using var conn = await server.ConnectAsync(defaultOnly: true); + + var slots = await conn.GetServer(server.DefaultEndPoint).ClusterSlotsAsync(); + log.WriteLine($"topology: {ClusterTopology.From(slots)?.Nodes.Count.ToString() ?? "(none)"}"); + } +} diff --git a/tests/StackExchange.Redis.Tests/EndpointIdentityUnitTests.cs b/tests/StackExchange.Redis.Tests/EndpointIdentityUnitTests.cs new file mode 100644 index 000000000..96c42a776 --- /dev/null +++ b/tests/StackExchange.Redis.Tests/EndpointIdentityUnitTests.cs @@ -0,0 +1,137 @@ +using System.Net; +using System.Threading.Tasks; +using Xunit; + +namespace StackExchange.Redis.Tests; + +/// +/// Coverage for a node having more than one identity - the enabler for the endpoint-identity work +/// (#2826), and a prerequisite for reacting to endpoints we did not choose the form of. +/// +public class EndpointIdentityUnitTests(ITestOutputHelper log) +{ + private const string Hostname = "host-1.redis.example.com"; + + private static DnsEndPoint AliasFor(EndPoint endpoint, string host = Hostname) + { + Server.RedisServer.GetHost(endpoint, out var port); + return new DnsEndPoint(host, port); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task NodeIsReachableByAlias(bool useSsl) + { + using var server = new InProcessTestServer(log, useSsl: useSsl); + var canonical = server.DefaultEndPoint; // an IPEndPoint + var alias = AliasFor(canonical); + server.AddAlias(alias, canonical); + + // dial the *alias*; before the alias map this fell through to a real socket connect + var config = server.GetClientConfig(defaultOnly: true); + config.EndPoints.Clear(); + config.EndPoints.Add(alias); + + await using var conn = await ConnectionMultiplexer.ConnectAsync(config); + var db = conn.GetDatabase(); + await db.StringSetAsync(nameof(NodeIsReachableByAlias), "abc"); + + Assert.Equal("abc", await db.StringGetAsync(nameof(NodeIsReachableByAlias))); + Assert.Equal(alias, Assert.Single(conn.GetEndPoints())); + } + + [Fact] + public async Task AliasAndCanonicalEndpointSeeTheSameData() + { + using var server = new InProcessTestServer(log); + var canonical = server.DefaultEndPoint; + var alias = AliasFor(canonical); + server.AddAlias(alias, canonical); + + var viaCanonical = server.GetClientConfig(defaultOnly: true); + var viaAlias = server.GetClientConfig(defaultOnly: true); + viaAlias.EndPoints.Clear(); + viaAlias.EndPoints.Add(alias); + + await using var byIp = await ConnectionMultiplexer.ConnectAsync(viaCanonical); + await using var byName = await ConnectionMultiplexer.ConnectAsync(viaAlias); + + await byIp.GetDatabase().StringSetAsync(nameof(AliasAndCanonicalEndpointSeeTheSameData), "xyz"); + + // one node, two names: the value written via one identity is visible via the other + Assert.Equal("xyz", await byName.GetDatabase().StringGetAsync(nameof(AliasAndCanonicalEndpointSeeTheSameData))); + } + + [Fact] + public void SetHostnameRegistersTheNameAsAnAlias() + { + using var server = new InProcessTestServer(log); + var canonical = server.DefaultEndPoint; + server.SetHostname(canonical, Hostname); + + Assert.True(server.TryGetNode(AliasFor(canonical), out var byName)); + Assert.True(server.TryGetNode(canonical, out var byAddress)); + Assert.Same(byAddress, byName); + + // GetEndPoints stays one-per-node; the alias is reported separately + Assert.Equal(canonical, Assert.Single(server.GetEndPoints())); + Assert.Equal(AliasFor(canonical), Assert.Single(server.GetAliases())); + } + + [Fact] + public async Task ClusterNodesAnnouncesHostname() + { + // hostnames need a 7.0+ server, which the in-process server declares by default + using var server = new InProcessTestServer(log) { ServerType = ServerType.Cluster }; + var canonical = server.DefaultEndPoint; + server.SetHostname(canonical, Hostname); + + await using var conn = await server.ConnectAsync(defaultOnly: true); + var raw = await conn.GetServer(canonical).ClusterNodesRawAsync(); + Assert.NotNull(raw); + log.WriteLine(raw); + + var host = Server.RedisServer.GetHost(canonical, out var port); + + // ... + Assert.Contains($"{host}:{port}@{port + 10000},{Hostname} ", raw); + } + + [Fact] + public async Task ClusterNodesOmitsHostnameWhenNoneAnnounced() + { + using var server = new InProcessTestServer(log) { ServerType = ServerType.Cluster }; + var canonical = server.DefaultEndPoint; + + await using var conn = await server.ConnectAsync(defaultOnly: true); + var raw = await conn.GetServer(canonical).ClusterNodesRawAsync(); + Assert.NotNull(raw); + log.WriteLine(raw); + + var host = Server.RedisServer.GetHost(canonical, out var port); + Assert.Contains($"{host}:{port}@{port + 10000} ", raw); + + // no trailing hostname on the endpoint token (the flags field has commas of its own) + Assert.DoesNotContain($"@{port + 10000},", raw); + } + +#if !NETFRAMEWORK + [Fact] + public async Task TlsNameMismatchIsNotForgiven() + { + // the in-process certificate covers every identity the node can be dialled by, so a mismatch + // has to be forced; this asserts the validation callback no longer waves one through on + // thumbprint alone, which is what would let the SslHost handling regress unnoticed + using var server = new InProcessTestServer(log, useSsl: true); + var config = server.GetClientConfig(defaultOnly: true); + config.SslHost = "not-in-the-certificate.example.com"; + config.ConnectTimeout = 2000; + config.ConnectRetry = 1; + + var ex = await Assert.ThrowsAnyAsync( + async () => await ConnectionMultiplexer.ConnectAsync(config)); + log.WriteLine(ex.Message); + } +#endif +} diff --git a/tests/StackExchange.Redis.Tests/InProcessTestServer.cs b/tests/StackExchange.Redis.Tests/InProcessTestServer.cs index 570143e66..45b34bbd6 100644 --- a/tests/StackExchange.Redis.Tests/InProcessTestServer.cs +++ b/tests/StackExchange.Redis.Tests/InProcessTestServer.cs @@ -1,4 +1,5 @@ using System; +using System.Collections.Generic; using System.IO; using System.IO.Pipelines; using System.Net; @@ -23,15 +24,19 @@ public class InProcessTestServer : MemoryCacheRedisServer { private readonly ITestOutputHelper? _log; #if !NETFRAMEWORK - private readonly X509Certificate2? _serverCertificate; - private readonly string? _serverCertificateThumbprint; + private readonly object _certificateLock = new(); + private X509Certificate2? _serverCertificate; + private string? _serverCertificateThumbprint; private readonly RemoteCertificateValidationCallback? _certificateValidationCallback; #endif public InProcessTestServer(ITestOutputHelper? log = null, EndPoint? endpoint = null, bool useSsl = false) : base(endpoint) { - RedisVersion = RedisFeatures.v6_0_0; // for client to expect RESP3 + // 6.0 would do for the client to expect RESP3, but default to 7.0 so that hostnames, the + // preferred-endpoint-type and the CLUSTER SLOTS metadata element are all live by default; + // tests that want the older shape ask for it explicitly + RedisVersion = new Version(7, 0, 0); _log = log; #if NETFRAMEWORK UseSsl = false; @@ -39,8 +44,7 @@ public InProcessTestServer(ITestOutputHelper? log = null, EndPoint? endpoint = n UseSsl = useSsl; if (useSsl) { - _serverCertificate = CreateServerCertificate(DefaultEndPoint); - _serverCertificateThumbprint = _serverCertificate.Thumbprint; + // note the certificate itself is created on first use, not here; see GetServerCertificate _certificateValidationCallback = ValidateServerCertificate; } #endif @@ -250,12 +254,37 @@ private bool ValidateServerCertificate(object sender, X509Certificate? certifica return true; } + // a self-signed certificate always trips chain validation, so tolerate that much; a *name* + // mismatch is a different matter - forgiving it would mask exactly the class of bug that the + // SslHost handling has to get right, and would let TLS identity tests pass for the wrong reason + if ((errors & ~SslPolicyErrors.RemoteCertificateChainErrors) != SslPolicyErrors.None) + { + Log($"rejecting server certificate: {errors}"); + return false; + } + return certificate is not null && _serverCertificateThumbprint is not null && string.Equals(certificate.GetCertHashString(), _serverCertificateThumbprint, StringComparison.OrdinalIgnoreCase); } - private static X509Certificate2 CreateServerCertificate(EndPoint endpoint) + // deferred until first use rather than built in the constructor: tests register aliases *after* + // construction, and every identity a node can be dialled by has to appear in the SAN list, or TLS + // validation fails for reasons that have nothing to do with what the test is exercising + private X509Certificate2 GetServerCertificate() + { + lock (_certificateLock) + { + if (_serverCertificate is null) + { + _serverCertificate = CreateServerCertificate(DefaultEndPoint, GetAliases()); + _serverCertificateThumbprint = _serverCertificate.Thumbprint; + } + return _serverCertificate; + } + } + + private static X509Certificate2 CreateServerCertificate(EndPoint endpoint, IEnumerable aliases) { var now = DateTimeOffset.UtcNow; var subjectName = GetCertificateSubjectName(endpoint); @@ -270,17 +299,27 @@ private static X509Certificate2 CreateServerCertificate(EndPoint endpoint) false)); var san = new SubjectAlternativeNameBuilder(); - switch (endpoint) + var seen = new HashSet(StringComparer.OrdinalIgnoreCase); + AddName(san, seen, endpoint); + foreach (var alias in aliases) { - case DnsEndPoint dns: - san.AddDnsName(dns.Host); - break; - case IPEndPoint ip: - san.AddIpAddress(ip.Address); - break; + AddName(san, seen, alias); } request.CertificateExtensions.Add(san.Build()); + static void AddName(SubjectAlternativeNameBuilder san, HashSet seen, EndPoint endpoint) + { + switch (endpoint) + { + case DnsEndPoint dns when seen.Add("dns:" + dns.Host): + san.AddDnsName(dns.Host); + break; + case IPEndPoint ip when seen.Add("ip:" + ip.Address): + san.AddIpAddress(ip.Address); + break; + } + } + using var certificate = request.CreateSelfSigned(now.AddMinutes(-5), now.AddDays(7)); #pragma warning disable SYSLIB0057 return new X509Certificate2(certificate.Export(X509ContentType.Pfx)); @@ -336,7 +375,7 @@ private sealed class InProcTunnel( { using var ssl = new SslStream(serverTransport, leaveInnerStreamOpen: false); await ssl.AuthenticateAsServerAsync( - server._serverCertificate!, + server.GetServerCertificate(), clientCertificateRequired: false, enabledSslProtocols: SslProtocols.None, checkCertificateRevocation: false).ConfigureAwait(false); diff --git a/tests/StackExchange.Redis.Tests/UnknownPushUnitTests.cs b/tests/StackExchange.Redis.Tests/UnknownPushUnitTests.cs new file mode 100644 index 000000000..2bca17ab6 --- /dev/null +++ b/tests/StackExchange.Redis.Tests/UnknownPushUnitTests.cs @@ -0,0 +1,86 @@ +using System.Threading.Tasks; +using RESPite.Messages; +using StackExchange.Redis.Server; +using Xunit; + +namespace StackExchange.Redis.Tests; + +/// +/// Verifies that an unrecognized RESP3 push frame is ignored rather than being matched to a pending +/// command. Maintenance notifications (MOVING, MIGRATING, SMIGRATED, ...) arrive as push frames whose +/// second element is an integer sequence id, not a pub/sub channel, so they exercise this path. +/// +public class UnknownPushUnitTests(ITestOutputHelper log) +{ + /// + /// Injects an unknown push frame immediately before the reply to a command, so the client sees + /// >3 SMIGRATING 1 123 and then the real response, with the command still outstanding. + /// + private sealed class UnknownPushServer(ITestOutputHelper? log) : InProcessTestServer(log) + { + public int Injected { get; private set; } + + public Task ConnectResp3Async() + { + var config = GetClientConfig(); + config.Protocol = RedisProtocol.Resp3; + return ConnectionMultiplexer.ConnectAsync(config); + } + + public override TypedRedisValue Execute(RedisClient client, in RedisRequest request) + { + // only inject for the command under test, and only when the client negotiated RESP3 + if (client.Protocol is RedisProtocol.Resp3 && request.Count >= 2 && request.IsString(0, "GET"u8)) + { + var push = TypedRedisValue.Rent(3, out var span, RespPrefix.Push); + span[0] = TypedRedisValue.SimpleString("SMIGRATING"); + span[1] = TypedRedisValue.Integer(1); + span[2] = TypedRedisValue.SimpleString("123,456,789-1000"); + client.AddOutbound(push); + Injected++; + Log($"[{client}] injected unknown push before {request.Command} reply"); + } + + return base.Execute(client, in request); + } + } + + [Fact] + public async Task UnknownPushIsIgnoredWhileCommandOutstanding() + { + using var server = new UnknownPushServer(log); + await using var conn = await server.ConnectResp3Async(); + + var db = conn.GetDatabase(); + RedisKey key = "unknown-push"; + await db.StringSetAsync(key, "abc"); + + // the GET reply is preceded on the wire by an unrecognized push frame; the GET must still + // receive its own reply + var value = await db.StringGetAsync(key); + + Assert.Equal(1, server.Injected); + Assert.Equal("abc", value); + } + + [Fact] + public async Task ConnectionSurvivesUnknownPushAndKeepsWorking() + { + using var server = new UnknownPushServer(log); + await using var conn = await server.ConnectResp3Async(); + + var db = conn.GetDatabase(); + RedisKey key = "unknown-push-repeat"; + await db.StringSetAsync(key, "abc"); + + // repeat: if a push frame is consumed as a command reply, the response stream desynchronizes + // and subsequent commands see the wrong replies (or the connection faults) + for (int i = 0; i < 5; i++) + { + Assert.Equal("abc", await db.StringGetAsync(key)); + } + + Assert.True(conn.IsConnected); + Assert.Equal(5, server.Injected); + } +} diff --git a/tests/StackExchange.Redis.Tests/UnroutableRedirectUnitTests.cs b/tests/StackExchange.Redis.Tests/UnroutableRedirectUnitTests.cs new file mode 100644 index 000000000..af833f914 --- /dev/null +++ b/tests/StackExchange.Redis.Tests/UnroutableRedirectUnitTests.cs @@ -0,0 +1,147 @@ +using System.Diagnostics; +using System.Linq; +using System.Net; +using System.Threading.Tasks; +using StackExchange.Redis.Availability; +using Xunit; +using static StackExchange.Redis.Server.RedisServer; + +namespace StackExchange.Redis.Tests; + +/// +/// A -MOVED whose target cannot be named: a hostname-preferring node redirecting to a peer that has +/// announced no hostname reports MOVED <slot> ?:<port>, and ? denotes "an unknown +/// node" - explicitly not the node that answered, so it must not be substituted. +/// +/// +/// Before this was handled, the redirect parsed as a whose host was literally +/// "?"; the client then created and dialled a ServerEndPoint for it, the command sat in the +/// backlog until it timed out, and the caller got a timeout exception pointing at the timeouts +/// troubleshooting article - with a phantom endpoint left behind in the multiplexer. +/// +public class UnroutableRedirectUnitTests(ITestOutputHelper log) +{ + private const string Key = "unroutable-key"; + + /// Cluster whose nodes prefer hostnames but announce none, so redirects between them say "?". + private static InProcessTestServer CreateServer(ITestOutputHelper log, out EndPoint target) + { + var server = new InProcessTestServer(log) + { + ServerType = ServerType.Cluster, + PreferredEndpointType = ClusterEndpointType.Hostname, + }; + GetHost(server.DefaultEndPoint, out var port); + target = new IPEndPoint(IPAddress.Loopback, port + 1); + return server; + } + + private static async Task<(ConnectionMultiplexer Conn, InProcessTestServer Server)> ConnectThenMigrateAsync(ITestOutputHelper log) + { + var server = CreateServer(log, out var targetEndpoint); + + // connect first, so the client's slot map still points at the original owner and the command + // actually earns a redirect + var conn = await server.ConnectAsync(defaultOnly: true); + + var target = server.AddEmptyNode(targetEndpoint); // announces no hostname, hence "?" + server.Migrate((RedisKey)Key, target); + return (conn, server); + } + + [Fact] + public async Task UnroutableRedirectFaultsTheCommand() + { + var (conn, server) = await ConnectThenMigrateAsync(log); + using (server) + await using (conn) + { + var ex = await Assert.ThrowsAsync( + () => conn.GetDatabase().StringGetAsync(Key)); + log.WriteLine(ex.Message); + + Assert.Equal(RedisErrorKind.UnknownRedirectTarget, ex.Kind); + + // the message must name the cause; the old behaviour blamed connectTimeout + Assert.Contains("does not identify a node that can be connected to", ex.Message); + Assert.DoesNotContain("connectTimeout", ex.Message); + } + } + + [Fact] + public async Task UnroutableRedirectDoesNotInventAnEndpoint() + { + var (conn, server) = await ConnectThenMigrateAsync(log); + using (server) + await using (conn) + { + await Assert.ThrowsAsync(() => conn.GetDatabase().StringGetAsync(Key)); + + foreach (var ep in conn.GetEndPoints()) + { + log.WriteLine($"endpoint: {ep}"); + } + + // "?" used to be added as a DnsEndPoint and dialled forever + Assert.DoesNotContain(conn.GetEndPoints(), ep => ep is DnsEndPoint { Host: "?" }); + Assert.All(conn.GetEndPoints(), ep => Assert.IsType(ep)); + } + } + + [Fact] + public async Task UnroutableRedirectFailsWithoutWaitingForATimeout() + { + var (conn, server) = await ConnectThenMigrateAsync(log); + using (server) + await using (conn) + { + // the old path parked the command in the backlog for the full async timeout + var timeout = conn.RawConfig.AsyncTimeout; + var watch = Stopwatch.StartNew(); + await Assert.ThrowsAsync(() => conn.GetDatabase().StringGetAsync(Key)); + watch.Stop(); + + log.WriteLine($"failed in {watch.ElapsedMilliseconds}ms, async timeout is {timeout}ms"); + Assert.True(watch.ElapsedMilliseconds < timeout / 2, $"took {watch.ElapsedMilliseconds}ms"); + } + } + + [Fact] + public async Task UnroutableRedirectIsTreatedAsNotApplied() + { + var (conn, server) = await ConnectThenMigrateAsync(log); + using (server) + await using (conn) + { + var ex = await Assert.ThrowsAsync( + () => conn.GetDatabase().StringGetAsync(Key)); + + // the redirect proves the command did not run, so a retry is a first attempt rather than a + // repeat - which is what lets WithRetry recover once the topology refresh has landed + var fault = new FaultContext(ex); + Assert.True(fault.NotApplied); + Assert.Equal(RedisErrorKind.UnknownRedirectTarget, fault.ErrorKind); + } + } + + [Fact] + public async Task RoutableRedirectStillFollowsNormally() + { + // the guard must not disturb ordinary redirects: same setup, but the target announces a hostname + var server = CreateServer(log, out var targetEndpoint); + using (server) + { + await using var conn = await server.ConnectAsync(defaultOnly: true); + + var target = server.AddEmptyNode(targetEndpoint); + server.SetHostname(target, "target.redis.example.com"); + server.Migrate((RedisKey)Key, target); + + // the toy server maps any endpoint to an in-proc node, so the hostname form is dialable here + await conn.GetDatabase().StringSetAsync(Key, "value"); + Assert.Equal("value", await conn.GetDatabase().StringGetAsync(Key)); + + Assert.Contains(conn.GetEndPoints(), ep => ep is DnsEndPoint { Host: "target.redis.example.com" }); + } + } +} diff --git a/toys/AotRig/AotRig.csproj b/toys/AotRig/AotRig.csproj index e8312f4fe..bad4f1930 100644 --- a/toys/AotRig/AotRig.csproj +++ b/toys/AotRig/AotRig.csproj @@ -11,6 +11,9 @@ true true true + + $(NoWarn);StringToRedisValue diff --git a/toys/StackExchange.Redis.Server/RedisServer.cs b/toys/StackExchange.Redis.Server/RedisServer.cs index 77ad4f1ed..409cbd31c 100644 --- a/toys/StackExchange.Redis.Server/RedisServer.cs +++ b/toys/StackExchange.Redis.Server/RedisServer.cs @@ -23,7 +23,14 @@ public static bool IsMatch(string pattern, string key) => private ConcurrentDictionary _nodes = new(); - public bool TryGetNode(EndPoint endpoint, out Node node) => _nodes.TryGetValue(endpoint, out node); + // additional identities for nodes that already exist; deliberately *not* extra _nodes entries, + // so that _nodes stays one-key-per-node: EndPointComparer cannot order mixed endpoint types + // (it returns 0), and AddEmptyNode derives the next endpoint from an arbitrary existing key, + // so a mixed _nodes would make both CLUSTER NODES/SLOTS ordering and node naming arbitrary + private readonly ConcurrentDictionary _aliases = new(); + + public bool TryGetNode(EndPoint endpoint, out Node node) + => _nodes.TryGetValue(endpoint, out node) || _aliases.TryGetValue(endpoint, out node); public EndPoint DefaultEndPoint { get; } @@ -47,6 +54,49 @@ public IEnumerable GetEndPoints() } } + /// + /// The secondary identities registered via or ; + /// these are not returned by , which reports one endpoint per node. + /// + public IEnumerable GetAliases() + { + foreach (var pair in _aliases) + { + yield return pair.Key; + } + } + + /// + /// Register an additional identity for an existing node. The node remains keyed by its original + /// endpoint, but requests for resolve to it, so a single node can be + /// reached by several names - which is what makes identity-unification testable. + /// + public void AddAlias(EndPoint alias, EndPoint node) + { + if (alias is null) throw new ArgumentNullException(nameof(alias)); + if (!_nodes.TryGetValue(node, out var target)) throw new KeyNotFoundException($"Node not found: {Format.ToString(node)}"); + if (_nodes.ContainsKey(alias)) throw new ArgumentException($"Already a node in its own right: {Format.ToString(alias)}", nameof(alias)); + if (!_aliases.TryAdd(alias, target) && !ReferenceEquals(_aliases[alias], target)) + { + throw new ArgumentException($"Already an alias of a different node: {Format.ToString(alias)}", nameof(alias)); + } + } + + /// + /// Announce a hostname for a node, as cluster-announce-hostname does; this is reported by + /// CLUSTER NODES and registered as an alias, so the node can be reached by that name as + /// well as by the endpoint it is keyed on. + /// + public void SetHostname(EndPoint endpoint, string hostname) + { + if (!_nodes.TryGetValue(endpoint, out var node)) throw new KeyNotFoundException($"Node not found: {Format.ToString(endpoint)}"); + node.Hostname = hostname; + if (!string.IsNullOrWhiteSpace(hostname)) + { + AddAlias(new DnsEndPoint(hostname, node.Port), endpoint); + } + } + public bool Migrate(int hashSlot, EndPoint to) { if (ServerType != ServerType.Cluster) throw new InvalidOperationException($"Server mode is {ServerType}"); @@ -70,6 +120,168 @@ public bool Migrate(int hashSlot, EndPoint to) public bool Migrate(Span key, EndPoint to) => Migrate(ServerSelectionStrategy.GetClusterSlot(key), to); public bool Migrate(in RedisKey key, EndPoint to) => Migrate(GetHashSlot(key), to); + /// + /// The value a node announces when hostname reporting is configured but no hostname is set; + /// prescribed by the CLUSTER SLOTS contract, not chosen here. + /// + private const string UnannouncedHostname = "?"; + + private const string ClusterPreferredEndpointType = "cluster-preferred-endpoint-type"; + private const string ClusterAnnounceHostname = "cluster-announce-hostname"; + + // hostnames, the endpoint-type preference and the CLUSTER SLOTS metadata map all arrived in 7.0; + // before that there is no hostname anywhere in the topology, so the whole mechanism is inert and + // a client sees exactly the ip-only shape it always did + private static readonly Version s_HostnameVersion = new(7, 0, 0); + + private bool SupportsHostnames => RedisVersion is { } version && version >= s_HostnameVersion; + + /// + /// Override the preferred endpoint type for a single node. The real parameter is per-node, so a + /// deployment *can* have nodes that disagree - nobody sane configures that deliberately, but a + /// rolling config change is exactly that for a window, and a client must not infer a + /// cluster-wide mode from any one node's reply. + /// + public void SetPreferredEndpointType(EndPoint endpoint, ClusterEndpointType preferred) + { + if (!_nodes.TryGetValue(endpoint, out var node)) throw new KeyNotFoundException($"Node not found: {Format.ToString(endpoint)}"); + node.PreferredEndpointType = preferred; + } + + private static string ToConfigValue(ClusterEndpointType type) => type switch + { + ClusterEndpointType.Hostname => "hostname", + ClusterEndpointType.UnknownEndpoint => "unknown-endpoint", + _ => "ip", + }; + + /// + /// Which identity form this server prefers when naming nodes, mirroring the server's + /// cluster-preferred-endpoint-type. It governs the primary endpoint position in + /// CLUSTER SLOTS and the address in -MOVED redirects; CLUSTER NODES is + /// unaffected, since its field positions do not move. + /// + public ClusterEndpointType PreferredEndpointType { get; set; } = ClusterEndpointType.Ip; + + /// + /// The identity form a server prefers when naming nodes; the values mirror + /// cluster-preferred-endpoint-type. + /// + public enum ClusterEndpointType + { + /// Report the node's ip address. + Ip = 0, + + /// Report the node's announced hostname, or ? if it has none. + Hostname, + + /// Report no endpoint at all, leaving the client to use the one it dialled. + UnknownEndpoint, + } + + /// + /// What a node knows about its own address. The values other than are + /// prescribed by the CLUSTER SLOTS contract rather than chosen here. Note the third + /// documented placeholder, ?, is deliberately *not* a member: it is derived from + /// hostnames being preferred while none is announced, so that a combination the real server + /// cannot produce cannot be expressed here either. + /// + public enum AnnouncedAddress + { + /// The node knows its own address and reports it. + Address = 0, + + /// + /// The RESP null: the server does not know this node's address, typically because it sits + /// behind a load balancer. A client should dial the endpoint it sent the command to, with + /// the port from the reply. + /// + Null, + + /// + /// The empty string: the node does not know its own ip - a single-node cluster, or a node + /// that has not joined yet. May be treated as . + /// + Empty, + } + + /// + /// Declare an extra CLUSTER SLOTS metadata entry for a node, beyond the ip/hostname the + /// complement rule produces. The real set is documented as extensible, so this exists to prove a + /// client keeps what it does not recognize. + /// + public void SetSlotsMetadata(EndPoint endpoint, string key, string value) + { + if (!_nodes.TryGetValue(endpoint, out var node)) throw new KeyNotFoundException($"Node not found: {Format.ToString(endpoint)}"); + (node.SlotsMetadata ??= new List>()) + .Add(new KeyValuePair(key, value)); + } + + /// + /// Declare an auxiliary field for a node, reported by CLUSTER NODES after the hostname. + /// Note a real 8.9 server emits none of these (cluster-announce-human-nodename does not + /// appear in the reply), so this exists to exercise the documented grammar - which is extensible - + /// rather than to mirror observed behaviour. + /// + public void SetAuxField(EndPoint endpoint, string key, string value) + { + if (!_nodes.TryGetValue(endpoint, out var node)) throw new KeyNotFoundException($"Node not found: {Format.ToString(endpoint)}"); + (node.AuxFields ??= new List>()) + .Add(new KeyValuePair(key, value)); + } + + /// + /// Control what a node reports as its own address, for exercising the placeholder values that + /// CLUSTER SLOTS can carry. + /// + public void SetAnnouncedAddress(EndPoint endpoint, AnnouncedAddress announced) + { + if (!_nodes.TryGetValue(endpoint, out var node)) throw new KeyNotFoundException($"Node not found: {Format.ToString(endpoint)}"); + node.Announced = announced; + } + + // the ip form of a node's identity, as it appears in either the primary position or the + // metadata; the empty string covers both "not known to the server" and "not known to the node", + // per the contract's note that "" applies to the ip metadata field as well as the endpoint + private static string GetIpForm(Node node) + => node.Announced == AnnouncedAddress.Address ? node.Host : ""; + + // a node we only know by name has no address identity of its own, so the only coherent shape is + // hostname-preferred with the ip unknown - anything else would have us report a hostname in a + // field that is documented to carry an address, which is the bug class this fake exists to expose + private static void ApplyNameOnlyIdentity(EndPoint endpoint, Node node) + { + if (endpoint is DnsEndPoint dns) + { + node.Hostname = dns.Host; + node.Announced = AnnouncedAddress.Empty; + + // scoped to this node rather than the whole server: adding one name-only node to an + // address-keyed cluster is a legitimate topology, not a reconfiguration of its peers + node.PreferredEndpointType = ClusterEndpointType.Hostname; + } + } + + // verified against a real 8.9 cluster: the *answering* node's preference governs every entry in + // the reply, while hostname availability is a property of each described node. So asking a + // hostname-preferring node about a peer with no hostname yields "?", and the same peer asked of an + // ip-preferring node yields its address + private static TypedRedisValue GetAnnouncedEndpoint(Node perspective, Node node) + { + switch (perspective.EffectiveEndpointType) + { + case ClusterEndpointType.UnknownEndpoint: + return TypedRedisValue.BulkString(RedisValue.Null); + case ClusterEndpointType.Hostname: + return TypedRedisValue.BulkString( + string.IsNullOrWhiteSpace(node.Hostname) ? UnannouncedHostname : node.Hostname); + default: + return node.Announced == AnnouncedAddress.Null + ? TypedRedisValue.BulkString(RedisValue.Null) + : TypedRedisValue.BulkString(GetIpForm(node)); + } + } + [Flags] public enum NodeFlags { @@ -82,6 +294,23 @@ public enum NodeFlags NoFailover = 1 << 5, } + /// + /// Add an empty node at a specific endpoint; the derived-endpoint overload copies the *form* of an + /// existing key, so it cannot produce a node whose identity form differs from its peers'. + /// + public EndPoint AddEmptyNode(EndPoint endpoint, NodeFlags flags = NodeFlags.None) + { + if (endpoint is null) throw new ArgumentNullException(nameof(endpoint)); + var node = new Node(this, endpoint, flags); + node.UpdateSlots([]); // explicit empty range (rather than implicit "all nodes") + ApplyNameOnlyIdentity(endpoint, node); + if (!_nodes.TryAdd(endpoint, node)) + { + throw new ArgumentException($"Node already exists: {Format.ToString(endpoint)}", nameof(endpoint)); + } + return endpoint; + } + public EndPoint AddEmptyNode(NodeFlags flags = NodeFlags.None) { EndPoint endpoint; @@ -119,6 +348,7 @@ public EndPoint AddEmptyNode(NodeFlags flags = NodeFlags.None) node = new(this, endpoint, flags); node.UpdateSlots([]); // explicit empty range (rather than implicit "all nodes") + ApplyNameOnlyIdentity(endpoint, node); } // defensive loop for concurrency while (!_nodes.TryAdd(endpoint, node)); @@ -128,7 +358,9 @@ public EndPoint AddEmptyNode(NodeFlags flags = NodeFlags.None) protected RedisServer(EndPoint endpoint = null, int databases = DefaultDatabaseCount, TextWriter output = null) : base(output) { DefaultEndPoint = endpoint ??= new IPEndPoint(IPAddress.Loopback, 6379); - _nodes.TryAdd(endpoint, new Node(this, endpoint, NodeFlags.None)); + var defaultNode = new Node(this, endpoint, NodeFlags.None); + ApplyNameOnlyIdentity(endpoint, defaultNode); + _nodes.TryAdd(endpoint, defaultNode); RedisVersion = s_DefaultServerVersion; if (databases < 1) throw new ArgumentOutOfRangeException(nameof(databases)); Databases = databases; @@ -516,7 +748,25 @@ protected virtual TypedRedisValue ClusterNodes(RedisClient client, in RedisReque foreach (var pair in _nodes.OrderBy(x => x.Key, EndPointComparer.Instance)) { var node = pair.Value; - sb.Append(node.Id).Append(" ").Append(node.Host).Append(":").Append(node.Port).Append("@1").Append(node.Port).Append(" "); + // ... + sb.Append(node.Id).Append(" ").Append(node.Host).Append(":").Append(node.Port).Append("@").Append(node.Port + 10000); + if (SupportsHostnames) + { + // the hostname slot is positional: it can be empty while aux fields follow it + var aux = node.AuxFields; + if (!string.IsNullOrWhiteSpace(node.Hostname) || aux is { Count: > 0 }) + { + sb.Append(",").Append(node.Hostname); + } + if (aux is { Count: > 0 }) + { + foreach (var field in aux) + { + sb.Append(",").Append(field.Key).Append("=").Append(field.Value); + } + } + } + sb.Append(" "); if (node == client.Node) { sb.Append("myself,"); @@ -550,26 +800,72 @@ protected virtual TypedRedisValue ClusterSlots(RedisClient client, in RedisReque { count += pair.Value.Slots.Length; } + var perspective = client?.Node ?? (TryGetNode(DefaultEndPoint, out var fallback) ? fallback : null); var slots = TypedRedisValue.Rent(count, out var slotsSpan, RespPrefix.Array); foreach (var pair in _nodes.OrderBy(x => x.Key, EndPointComparer.Instance)) { - string host = GetHost(pair.Key, out int port); - foreach (var range in pair.Value.Slots) + var node = pair.Value; + GetHost(pair.Key, out int port); + foreach (var range in node.Slots) { if (index >= count) break; // someone changed things while we were working slotsSpan[index++] = TypedRedisValue.Rent(3, out var slotSpan, RespPrefix.Array); slotSpan[0] = TypedRedisValue.Integer(range.From); slotSpan[1] = TypedRedisValue.Integer(range.To); - slotSpan[2] = TypedRedisValue.Rent(4, out var nodeSpan, RespPrefix.Array); - nodeSpan[0] = TypedRedisValue.BulkString(host); + // the metadata element itself only exists from 7.0; older servers stop at the node id + slotSpan[2] = TypedRedisValue.Rent(SupportsHostnames ? 4 : 3, out var nodeSpan, RespPrefix.Array); + + // note the first field is positionally "the endpoint" and its *content* is whichever + // form is preferred, so it may well be a hostname rather than an address + nodeSpan[0] = GetAnnouncedEndpoint(perspective ?? node, node); nodeSpan[1] = TypedRedisValue.Integer(port); - nodeSpan[2] = TypedRedisValue.BulkString(pair.Value.Id); - nodeSpan[3] = TypedRedisValue.EmptyArray(RespPrefix.Array); + nodeSpan[2] = TypedRedisValue.BulkString(node.Id); + if (SupportsHostnames) + { + nodeSpan[3] = GetEndpointMetadata(perspective ?? node, node); + } } } return slots; } + // the metadata map is documented as the *complement* of the primary position: ip when the + // preferred type is not ip, hostname when the node has one and the preferred type is not + // hostname. So the union of the primary field and the metadata is every form the node has, with + // no duplication - which is what makes one reply enough to reconcile identities. + private static TypedRedisValue GetEndpointMetadata(Node perspective, Node node) + { + bool wantIp = perspective.EffectiveEndpointType != ClusterEndpointType.Ip; + bool wantHostname = perspective.EffectiveEndpointType != ClusterEndpointType.Hostname + && !string.IsNullOrWhiteSpace(node.Hostname); + + var extra = node.SlotsMetadata; + int pairs = (wantIp ? 1 : 0) + (wantHostname ? 1 : 0) + (extra?.Count ?? 0); + if (pairs == 0) return TypedRedisValue.EmptyArray(RespPrefix.Map); + + var result = TypedRedisValue.Rent(2 * pairs, out var span, RespPrefix.Map); + int index = 0; + if (wantIp) + { + span[index++] = TypedRedisValue.BulkString("ip"); + span[index++] = TypedRedisValue.BulkString(GetIpForm(node)); + } + if (wantHostname) + { + span[index++] = TypedRedisValue.BulkString("hostname"); + span[index++] = TypedRedisValue.BulkString(node.Hostname); + } + if (extra is not null) + { + foreach (var pair in extra) + { + span[index++] = TypedRedisValue.BulkString(pair.Key); + span[index++] = TypedRedisValue.BulkString(pair.Value); + } + } + return result; + } + private sealed class EndPointComparer : IComparer { private EndPointComparer() { } @@ -639,6 +935,57 @@ public override string ToString() public string Host { get; } + /// + /// The announced hostname of this node, if any; an additional identity rather than a + /// replacement for , which remains what the node is keyed on. + /// + public string Hostname { get; internal set; } + + /// + /// What this node knows about its own address; + /// unless a test says otherwise. + /// + public AnnouncedAddress Announced { get; internal set; } + + /// + /// Auxiliary name=value fields this node declares after its hostname in + /// CLUSTER NODES; null when it declares none. + /// + public List> AuxFields { get; internal set; } + + /// + /// Extra CLUSTER SLOTS metadata entries beyond the complement rule's ip/hostname. + /// + public List> SlotsMetadata { get; internal set; } + + /// + /// This node's own preferred endpoint type, or null to follow the server-wide default. + /// Nullable rather than copied at construction, so that setting the server default later + /// still applies to nodes that have no opinion. + /// + public ClusterEndpointType? PreferredEndpointType { get; internal set; } + + /// + /// The preference as it actually applies to this node, given the server version; a pre-7.0 + /// server has no such setting at all, so it reports addresses. + /// + internal ClusterEndpointType EffectiveEndpointType => _server.SupportsHostnames + ? PreferredEndpointType ?? _server.PreferredEndpointType + : ClusterEndpointType.Ip; + + /// + /// The host portion this node is named by in a -MOVED redirect issued by + /// - the redirect uses the *answering* node's preferred + /// endpoint type, as a real server does. Empty when no endpoint can be given, in which case a + /// client should dial the endpoint it sent the command to, using the port from the redirect. + /// + public string AnnouncedHostFrom(Node perspective) => (perspective ?? this).EffectiveEndpointType switch + { + ClusterEndpointType.UnknownEndpoint => "", + ClusterEndpointType.Hostname => string.IsNullOrWhiteSpace(Hostname) ? UnannouncedHostname : Hostname, + _ => GetIpForm(this), + }; + public int Port { get; } public string Id { get; } = NewId(); @@ -695,19 +1042,42 @@ public bool HasSlot(ReadOnlySpan key) return false; } + private static int s_idCounter; +#if !NET + private static readonly Random s_rand = new Random(); +#endif + private static string NewId() { Span data = stackalloc char[40]; + ReadOnlySpan alphabet = "0123456789abcdef"; #if NET var rand = Random.Shared; -#else - var rand = new Random(); -#endif - ReadOnlySpan alphabet = "0123456789abcdef"; for (int i = 0; i < data.Length; i++) { data[i] = alphabet[rand.Next(alphabet.Length)]; } +#else + // one shared instance: .NET Framework seeds Random from Environment.TickCount, so an + // instance per call yields identical sequences - and therefore identical node ids - for + // nodes created within the same tick + lock (s_rand) + { + for (int i = 0; i < data.Length; i++) + { + data[i] = alphabet[s_rand.Next(alphabet.Length)]; + } + } +#endif + + // ...and weave in a counter, so ids are unique by construction rather than by luck. Node ids + // are the identity that topology reconciliation keys on, so a collision here does not produce + // a test failure that looks like a collision: it looks like the client wrongly merging nodes + var unique = (uint)Interlocked.Increment(ref s_idCounter); + for (int i = 0; i < 8; i++) + { + data[i] = alphabet[(int)((unique >> (28 - (i * 4))) & 0xF)]; + } return data.ToString(); } @@ -906,7 +1276,20 @@ protected virtual TypedRedisValue LRange(RedisClient client, in RedisRequest req } protected virtual void LRange(int database, in RedisKey key, long start, Span arr) => throw new NotSupportedException(); - protected virtual void OnUpdateServerConfiguration() { } + // both of these parameters are per-node on a real server, so the values published depend on the + // connection asking; pre-7.0 they do not exist at all and are simply absent. This is why CONFIG GET + // is not LockFree: it writes the answering node's view into the shared configuration first + protected virtual void OnUpdateServerConfiguration(RedisClient client) + { + if (!SupportsHostnames) return; + + var node = client?.Node ?? (TryGetNode(DefaultEndPoint, out var fallback) ? fallback : null); + if (node is null) return; + + var config = ServerConfiguration; + config[ClusterPreferredEndpointType] = ToConfigValue(node.EffectiveEndpointType); + config[ClusterAnnounceHostname] = node.Hostname ?? ""; + } protected RedisConfig ServerConfiguration { get; } = RedisConfig.Create(); protected struct RedisConfig { @@ -937,12 +1320,12 @@ internal int CountMatch(string pattern) return count; } } - [RedisCommand(3, nameof(RedisCommand.CONFIG), "get", LockFree = true)] + [RedisCommand(3, nameof(RedisCommand.CONFIG), "get")] protected virtual TypedRedisValue Config(RedisClient client, in RedisRequest request) { var pattern = request.GetString(2); - OnUpdateServerConfiguration(); + OnUpdateServerConfiguration(client); var config = ServerConfiguration; var matches = config.CountMatch(pattern); if (matches == 0) return TypedRedisValue.EmptyArray(RespPrefix.Map); diff --git a/toys/StackExchange.Redis.Server/RespServer.cs b/toys/StackExchange.Redis.Server/RespServer.cs index cb98ead2b..10fc872f5 100644 --- a/toys/StackExchange.Redis.Server/RespServer.cs +++ b/toys/StackExchange.Redis.Server/RespServer.cs @@ -488,7 +488,10 @@ public virtual TypedRedisValue Execute(RedisClient client, in RedisRequest reque if (GetNode(moved.HashSlot) is { } node) { OnMoved(client, moved.HashSlot, node); - return TypedRedisValue.Error($"MOVED {moved.HashSlot} {node.Host}:{node.Port}"); + + // the redirect names the node in whichever form the preferred endpoint type + // selects, which may be a hostname, or empty when no endpoint can be given + return TypedRedisValue.Error($"MOVED {moved.HashSlot} {node.AnnouncedHostFrom(client?.Node)}:{node.Port}"); } return TypedRedisValue.Error($"ERR key has been migrated from slot {moved.HashSlot}, but the new owner is unknown"); }