diff --git a/rust/Cargo.lock b/rust/Cargo.lock index e3c73c31..cdfe806c 100644 --- a/rust/Cargo.lock +++ b/rust/Cargo.lock @@ -701,6 +701,19 @@ dependencies = [ "cmov", ] +[[package]] +name = "dashmap" +version = "5.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "978747c1d849a7d2ee5e8adc0159961c48fb7e5db2f06af6723b80123bb53856" +dependencies = [ + "cfg-if", + "hashbrown 0.14.5", + "lock_api", + "once_cell", + "parking_lot_core", +] + [[package]] name = "dashmap" version = "6.2.1" @@ -1014,6 +1027,7 @@ dependencies = [ "aws-credential-types", "aws-sigv4", "bytes", + "dashmap 5.5.3", "futures", "futures-intrusive", "http 1.4.2", @@ -2080,7 +2094,7 @@ dependencies = [ "bytes", "combine", "crc16", - "dashmap", + "dashmap 6.2.1", "dispose", "futures", "futures-util", diff --git a/rust/src/ffi.rs b/rust/src/ffi.rs index 8db1f810..68fe31d9 100644 --- a/rust/src/ffi.rs +++ b/rust/src/ffi.rs @@ -406,6 +406,7 @@ pub(crate) unsafe fn create_connection_request( tcp_nodelay: false, // TODO #490: Expose TCP_NODELAY. periodic_checks: None, // TODO #485: Expose cluster periodic checks. inflight_requests_limit: None, // TODO #484: Expose request limiting. + recovery_requests_queue_size: None, }) } diff --git a/sources/Valkey.Glide/Abstract/ConnectionMultiplexer.cs b/sources/Valkey.Glide/Abstract/ConnectionMultiplexer.cs index cd8e3fc4..f4e11891 100644 --- a/sources/Valkey.Glide/Abstract/ConnectionMultiplexer.cs +++ b/sources/Valkey.Glide/Abstract/ConnectionMultiplexer.cs @@ -127,8 +127,19 @@ public IServer[] GetServers() // run INFO on all nodes, but disregard the node responses, we need node addresses only if (_db!.IsCluster) { - Dictionary info = _db.Command(Request.Info([]).ToMultiNodeValue(), Route.AllNodes).GetAwaiter().GetResult(); - return [.. info.Keys.Select(addr => new ValkeyServer(_db, IPEndPoint.Parse(addr)))]; + var servers = new List(); + + foreach (var addr in _db.Command(Request.Info([]).ToMultiNodeValue(), Route.AllNodes).GetAwaiter().GetResult().Keys) + { + if (!Format.TryParseEndPoint(addr, out EndPoint? ep)) + { + throw new FormatException($"Could not parse endpoint address: '{addr}'"); + } + + servers.Add(new ValkeyServer(_db, ep)); + } + + return [.. servers]; } else { diff --git a/tests/Valkey.Glide.IntegrationTests/ClusterClientTests.cs b/tests/Valkey.Glide.IntegrationTests/ClusterClientTests.cs index 53c38c1b..ddb541e8 100644 --- a/tests/Valkey.Glide.IntegrationTests/ClusterClientTests.cs +++ b/tests/Valkey.Glide.IntegrationTests/ClusterClientTests.cs @@ -363,7 +363,7 @@ public async Task TestClientId_WithRoute(GlideClusterClient client) // Test CLIENT ID with all nodes routing var allNodesResult = await client.ClientIdAsync(AllNodes); Assert.True(allNodesResult.HasMultiData); - Assert.True(allNodesResult.MultiValue.Count > 0); + Assert.NotEmpty(allNodesResult.MultiValue); foreach (var kvp in allNodesResult.MultiValue) { @@ -417,7 +417,7 @@ public async Task TestClientGetName_WithRoute(GlideClusterClient client) // Test CLIENT GETNAME with all nodes routing var allNodesResult = await client.ClientGetNameAsync(AllNodes); Assert.True(allNodesResult.HasMultiData); - Assert.True(allNodesResult.MultiValue.Count > 0); + Assert.NotEmpty(allNodesResult.MultiValue); foreach (var kvp in allNodesResult.MultiValue) { diff --git a/tests/Valkey.Glide.IntegrationTests/ConnectionManagementCommandTests.cs b/tests/Valkey.Glide.IntegrationTests/ConnectionManagementCommandTests.cs index 232b7e1a..d6b0965c 100644 --- a/tests/Valkey.Glide.IntegrationTests/ConnectionManagementCommandTests.cs +++ b/tests/Valkey.Glide.IntegrationTests/ConnectionManagementCommandTests.cs @@ -168,6 +168,8 @@ public async Task TestClientPause_ReadsPausedUntilExpires(bool clusterMode) // Verify that read commands are blocked until the pause expires. _ = await client.GetAsync(key); Assert.True(sw.Elapsed >= pauseFor); + + await client.ClientUnpauseAsync(); } [Theory] @@ -187,6 +189,8 @@ public async Task TestClientPause_WritesPausedUntilExpires(bool clusterMode) // Verify that write commands are blocked until the pause expires. await client.SetAsync(key, "after"); Assert.True(sw.Elapsed >= pauseFor); + + await client.ClientUnpauseAsync(); } [Theory] diff --git a/tests/Valkey.Glide.IntegrationTests/DnsTests.cs b/tests/Valkey.Glide.IntegrationTests/DnsTests.cs index fe0ed9e3..c1e826df 100644 --- a/tests/Valkey.Glide.IntegrationTests/DnsTests.cs +++ b/tests/Valkey.Glide.IntegrationTests/DnsTests.cs @@ -1,8 +1,9 @@ // Copyright Valkey GLIDE Project Contributors - SPDX Identifier: Apache-2.0 +using System.Net; + using Valkey.Glide.TestUtils; -using static Valkey.Glide.ConnectionConfiguration; using static Valkey.Glide.Errors; using static Valkey.Glide.TestUtils.Client; using static Valkey.Glide.TestUtils.Constants; @@ -22,100 +23,131 @@ public class DnsTests(DnsTestsFixture fixture) : IClassFixture /// Environment variable for enabling DNS tests. /// See DEVELOPER.md for more details. /// - private const string DnsEnabledEnvVar = "VALKEY_GLIDE_DNS_TESTS_ENABLED"; + internal const string DnsEnabledEnvVar = "VALKEY_GLIDE_DNS_TESTS_ENABLED"; #endregion - #region Tests + #region Test Data - [Theory] - [MemberData(nameof(ClusterMode), MemberType = typeof(Data))] - public async Task ConnectWithValidHostname_Succeeds(bool useCluster) + public static TheoryData TlsMode => [true, false]; + + public static TheoryData ClusterAndTlsMode => new() { - SkipIfDnsTestsNotEnabled(); - await using var client = await BuildClient(useCluster, useTls: false, HostnameNoTls); - await AssertConnected(client); - } + { true, true }, + { true, false }, + { false, true }, + { false, false }, + }; - [Theory] - [MemberData(nameof(ClusterMode), MemberType = typeof(Data))] - public async Task ConnectWithInvalidHostname_Fails(bool useCluster) + #endregion + #region Constructor + + static DnsTests() { - SkipIfDnsTestsNotEnabled(); - _ = await Assert.ThrowsAsync(async () - => await BuildClient(useCluster, useTls: false, "NONEXISTENT.INVALID")); + Assert.SkipWhen( + string.IsNullOrEmpty(Environment.GetEnvironmentVariable(DnsEnabledEnvVar)), + "DNS tests are disabled. See DEVELOPER.md for setup instructions."); } + #endregion + #region Tests + [Theory] - [MemberData(nameof(ClusterMode), MemberType = typeof(Data))] - public async Task Tls_WithHostnameInCertificate_Succeeds(bool useCluster) + [MemberData(nameof(ClusterAndTlsMode))] + public async Task Connect_WithValidHostname_Succeeds(bool useCluster, bool useTls) { - SkipIfDnsTestsNotEnabled(); - await using var client = await BuildClient(useCluster, useTls: true, HostnameTls); + var server = GetServer(useCluster, useTls); + var host = useTls ? HostnameTls : HostnameNoTls; + await using var client = await server.CreateClientAsync(host); + await AssertConnected(client); } [Theory] - [MemberData(nameof(ClusterMode), MemberType = typeof(Data))] - public async Task Tls_WithHostnameNotInCertificate_Fails(bool useCluster) - { - SkipIfDnsTestsNotEnabled(); - _ = await Assert.ThrowsAsync(async () - => await BuildClient(useCluster, useTls: true, HostnameNoTls)); - } + [MemberData(nameof(ClusterAndTlsMode))] + public async Task Connect_WithInvalidHostname_Fails(bool useCluster, bool useTls) + => _ = await Assert.ThrowsAsync(async () + => await GetServer(useCluster, useTls).CreateClientAsync("NONEXISTENT.INVALID")); - #endregion - #region Helpers + [Theory] + [MemberData(nameof(ClusterMode), MemberType = typeof(Data))] + public async Task Connect_WithHostnameNotInCertificate_Fails(bool useCluster) + => _ = await Assert.ThrowsAsync(async () + => await GetServer(useCluster, useTls: true).CreateClientAsync(HostnameNoTls)); /// - /// Returns true if DNS tests are enabled. + /// Verifies that returns servers with + /// instances when the cluster topology reports DNS hostnames (via cluster-announce-hostname). /// - public static bool IsDnsTestsEnabled() - => !string.IsNullOrEmpty(Environment.GetEnvironmentVariable(DnsEnabledEnvVar)); + /// #419 + [Theory] + [MemberData(nameof(TlsMode))] + public void GetServers_WithDnsHostname_ReturnsDnsEndPoints(bool useTls) + { + var server = GetServer(useCluster: true, useTls); + var config = new ConfigurationOptions { Ssl = useTls }; + config.EndPoints.Add(server.Address.Host, server.Address.Port); - /// - /// Skips the current test if DNS tests are not enabled. - /// - private static void SkipIfDnsTestsNotEnabled() - => Assert.SkipWhen( - !IsDnsTestsEnabled(), - $"DNS tests are disabled. See DEVELOPER.md for setup instructions."); + if (useTls) + { + config.TrustIssuer(ServerManager.ServerCertificatePath); + } + + using var conn = ConnectionMultiplexer.Connect(config); + + var servers = conn.GetServers(); + Assert.NotEmpty(servers); + + foreach (var s in servers) + { + _ = Assert.IsType(s.EndPoint); + Assert.Contains(HostnameTls, s.EndPoint.ToString()); + } + } /// - /// Builds and returns a client configured with the specified parameters. + /// Verifies that returns + /// instances when the cluster topology reports DNS hostnames (via cluster-announce-hostname). /// - private async Task BuildClient(bool useCluster, bool useTls, string host) + /// #419 + [Theory] + [MemberData(nameof(TlsMode))] + public void GetEndPoints_WithDnsHostname_ReturnsDnsEndPoints(bool useTls) { - if (useCluster) + var server = GetServer(useCluster: true, useTls); + var config = new ConfigurationOptions { Ssl = useTls }; + config.EndPoints.Add(server.Address.Host, server.Address.Port); + + if (useTls) { - var server = useTls ? fixture.TlsClusterServer! : fixture.ClusterServer!; - var builder = new ClusterClientConfigurationBuilder() - .WithAddress(host, server.Address.Port); + config.TrustIssuer(ServerManager.ServerCertificatePath); + } - if (useTls) - { - _ = builder.WithTls(); - _ = builder.WithTrustedCertificate(server.CertificateData!); - } + using var conn = ConnectionMultiplexer.Connect(config); - return await GlideClusterClient.CreateClient(builder.Build()); - } + var endpoints = conn.GetEndPoints(false); + Assert.NotEmpty(endpoints); - else + foreach (var endpoint in endpoints) { - var server = useTls ? fixture.TlsStandaloneServer! : fixture.StandaloneServer!; - var builder = new StandaloneClientConfigurationBuilder() - .WithAddress(host, server.Address.Port); + _ = Assert.IsType(endpoint); + Assert.Contains(HostnameTls, endpoint.ToString()); - if (useTls) - { - _ = builder.WithTls(); - _ = builder.WithTrustedCertificate(server.CertificateData!); - } - - return await GlideClient.CreateClient(builder.Build()); + var s = conn.GetServer(endpoint); + Assert.Equal(endpoint, s.EndPoint); } } + #endregion + #region Helpers + + private Server GetServer(bool useCluster, bool useTls) => (useCluster, useTls) switch + { + (true, true) => fixture.TlsClusterServer!, + (true, false) => fixture.ClusterServer!, + (false, true) => fixture.TlsStandaloneServer!, + _ => fixture.StandaloneServer!, + }; + #endregion } @@ -131,14 +163,10 @@ public class DnsTestsFixture : IAsyncLifetime public ValueTask InitializeAsync() { - // Only start the servers if DNS tests are enabled. - if (DnsTests.IsDnsTestsEnabled()) - { - ClusterServer = new(useTls: false); - StandaloneServer = new(useTls: false); - TlsClusterServer = new(useTls: true); - TlsStandaloneServer = new(useTls: true); - } + ClusterServer = new(host: HostnameTls); + StandaloneServer = new(host: HostnameTls); + TlsClusterServer = new(useTls: true, host: HostnameTls); + TlsStandaloneServer = new(useTls: true, host: HostnameTls); return ValueTask.CompletedTask; } diff --git a/tests/Valkey.Glide.IntegrationTests/ScriptingCommandTests.cs b/tests/Valkey.Glide.IntegrationTests/ScriptingCommandTests.cs index 891040ae..76385811 100644 --- a/tests/Valkey.Glide.IntegrationTests/ScriptingCommandTests.cs +++ b/tests/Valkey.Glide.IntegrationTests/ScriptingCommandTests.cs @@ -1046,7 +1046,7 @@ public async Task FunctionStatsAsync_ReturnsStatistics(GlideClient client) Assert.NotNull(stats); Assert.NotNull(stats.Engines); - Assert.True(stats.Engines.Count > 0); + Assert.NotEmpty(stats.Engines); // Check LUA engine stats Assert.True(stats.Engines.ContainsKey("LUA")); @@ -1122,7 +1122,7 @@ public async Task FunctionDumpAsync_CreatesBackup(GlideClient client) byte[] backup = await client.FunctionDumpAsync(CancellationToken); Assert.NotNull(backup); - Assert.True(backup.Length > 0); + Assert.NotEmpty(backup); } [Theory(DisableDiscoveryEnumeration = true)] @@ -1249,7 +1249,7 @@ public async Task FCallAsync_WithAllPrimariesRouting_ExecutesOnAllPrimaries(Glid // Verify load succeeded (may be single or multi-value depending on cluster configuration) if (loadResult.HasMultiData) { - Assert.True(loadResult.MultiValue.Count > 0); + Assert.NotEmpty(loadResult.MultiValue); Assert.All(loadResult.MultiValue.Values, name => Assert.Equal(libName, name)); } else @@ -1263,7 +1263,7 @@ public async Task FCallAsync_WithAllPrimariesRouting_ExecutesOnAllPrimaries(Glid // Verify execution (may be single or multi-value depending on cluster configuration) if (result.HasMultiData) { - Assert.True(result.MultiValue.Count > 0); + Assert.NotEmpty(result.MultiValue); Assert.All(result.MultiValue.Values, r => Assert.Equal("Hello from primary", r.ToString())); } else @@ -1297,7 +1297,7 @@ public async Task FCallAsync_WithAllNodesRouting_ExecutesOnAllNodes(GlideCluster // Verify load succeeded (may be single or multi-value depending on cluster configuration) if (loadResult.HasMultiData) { - Assert.True(loadResult.MultiValue.Count > 0); + Assert.NotEmpty(loadResult.MultiValue); } else { @@ -1310,7 +1310,7 @@ public async Task FCallAsync_WithAllNodesRouting_ExecutesOnAllNodes(GlideCluster // Verify execution (may be single or multi-value depending on cluster configuration) if (result.HasMultiData) { - Assert.True(result.MultiValue.Count > 0); + Assert.NotEmpty(result.MultiValue); Assert.All(result.MultiValue.Values, r => Assert.Equal("Hello from node", r.ToString())); } else @@ -1428,7 +1428,7 @@ public async Task FunctionListAsync_WithRouting_ReturnsLibrariesFromSpecifiedNod // Verify list returned (may be single or multi-value depending on cluster configuration) if (result.HasMultiData) { - Assert.True(result.MultiValue.Count > 0); + Assert.NotEmpty(result.MultiValue); // Verify each node has the library foreach (var (node, libraries) in result.MultiValue) { @@ -1467,7 +1467,7 @@ public async Task FunctionStatsAsync_WithRouting_ReturnsPerNodeStats(GlideCluste // Verify stats returned (may be single or multi-value depending on cluster configuration) if (result.HasMultiData) { - Assert.True(result.MultiValue.Count > 0); + Assert.NotEmpty(result.MultiValue); // Verify each node has stats foreach (var (_, stats) in result.MultiValue) { @@ -1614,7 +1614,7 @@ public async Task ClusterValue_MultiNodeResults_HandlesCorrectly(GlideClusterCli { Assert.False(loadResult.HasSingleData); Assert.NotNull(loadResult.MultiValue); - Assert.True(loadResult.MultiValue.Count > 0); + Assert.NotEmpty(loadResult.MultiValue); // Verify each node address is a key in the dictionary foreach (var (nodeAddress, libraryName) in loadResult.MultiValue) diff --git a/tests/Valkey.Glide.TestUtils/Server.cs b/tests/Valkey.Glide.TestUtils/Server.cs index b7c99560..200f5480 100644 --- a/tests/Valkey.Glide.TestUtils/Server.cs +++ b/tests/Valkey.Glide.TestUtils/Server.cs @@ -85,7 +85,8 @@ public abstract class Server : IDisposable protected Server( bool useClusterMode, bool useTls = false, - int? replicaCount = null) + int? replicaCount = null, + string? host = null) { UseTls = useTls; @@ -93,7 +94,8 @@ protected Server( name: _name, useClusterMode: useClusterMode, useTls: UseTls, - replicaCount: replicaCount).First(); + replicaCount: replicaCount, + host: host).First(); if (UseTls) { @@ -128,7 +130,8 @@ public void Dispose() /// /// Builds and returns a client for this server. /// - public abstract Task CreateClientAsync(); + /// Optional hostname. + public abstract Task CreateClientAsync(string? host = null); /// /// Updates the password for the server to the given value. @@ -186,7 +189,7 @@ protected static async Task CreateClientAsync(Func> factory) /// /// Valkey cluster server. /// -public sealed class ClusterServer(bool useTls = false) : Server(useClusterMode: true, useTls: useTls) +public sealed class ClusterServer(bool useTls = false, string? host = null) : Server(useClusterMode: true, useTls: useTls, host: host) { #region Public Methods @@ -200,18 +203,32 @@ public ClusterClientConfigurationBuilder CreateConfigBuilder() trustedCertificate: UseTls ? CertificateData : null, password: _password); - /// - public override async Task CreateClientAsync() - => await CreateClusterClientAsync(); + /// + public override async Task CreateClientAsync(string? host = null) + => await CreateClusterClientAsync(host); /// /// Builds and returns a cluster client for this server. /// - public async Task CreateClusterClientAsync() + /// Optional hostname. + public async Task CreateClusterClientAsync(string? host = null) { - var config = CreateConfigBuilder().Build(); - Task factory() => GlideClusterClient.CreateClient(config); - return await CreateClientAsync(factory); + var builder = new ClusterClientConfigurationBuilder() + .WithAddress(host ?? Address.Host, Address.Port); + + if (UseTls) + { + _ = builder.WithTls(); + _ = builder.WithTrustedCertificate(CertificateData!); + } + + if (_password != null) + { + _ = builder.WithAuthentication(password: _password); + } + + return await CreateClientAsync(() + => GlideClusterClient.CreateClient(builder.Build())); } public override async Task SetPasswordAsync(string password) @@ -242,7 +259,8 @@ public override async Task KillClientsAsync() /// public sealed class StandaloneServer( bool useTls = false, - int? replicaCount = null) : Server(useClusterMode: false, useTls: useTls, replicaCount: replicaCount) + int? replicaCount = null, + string? host = null) : Server(useClusterMode: false, useTls: useTls, replicaCount: replicaCount, host: host) { #region Public Methods @@ -256,18 +274,32 @@ public StandaloneClientConfigurationBuilder CreateConfigBuilder() trustedCertificate: UseTls ? CertificateData : null, password: _password); - /// - public override async Task CreateClientAsync() - => await CreateStandaloneClientAsync(); + /// + public override async Task CreateClientAsync(string? host = null) + => await CreateStandaloneClientAsync(host); /// /// Builds and returns a standalone client for this server. /// - public async Task CreateStandaloneClientAsync() + /// Optional hostname. + public async Task CreateStandaloneClientAsync(string? host = null) { - var config = CreateConfigBuilder().Build(); - Task factory() => GlideClient.CreateClient(config); - return await CreateClientAsync(factory); + var builder = new StandaloneClientConfigurationBuilder() + .WithAddress(host ?? Address.Host, Address.Port); + + if (UseTls) + { + _ = builder.WithTls(); + _ = builder.WithTrustedCertificate(CertificateData!); + } + + if (_password != null) + { + _ = builder.WithAuthentication(password: _password); + } + + return await CreateClientAsync(() + => GlideClient.CreateClient(builder.Build())); } public override async Task SetPasswordAsync(string password) diff --git a/tests/Valkey.Glide.TestUtils/ServerManager.cs b/tests/Valkey.Glide.TestUtils/ServerManager.cs index 1995256a..9145caa7 100644 --- a/tests/Valkey.Glide.TestUtils/ServerManager.cs +++ b/tests/Valkey.Glide.TestUtils/ServerManager.cs @@ -67,7 +67,8 @@ public static IList
StartServer( string name, bool useClusterMode = false, bool useTls = false, - int? replicaCount = null) + int? replicaCount = null, + string? host = null) { // Build command arguments. List args = []; @@ -77,6 +78,11 @@ public static IList
StartServer( args.Add("--tls"); } + if (host != null) + { + args.AddRange(["--host", host]); + } + args.Add("start"); args.AddRange(["--prefix", name]); args.AddRange(["-r", (replicaCount ?? DefaultReplicaCount).ToString()]); diff --git a/valkey-glide b/valkey-glide index 6d9fae25..cf3d56f0 160000 --- a/valkey-glide +++ b/valkey-glide @@ -1 +1 @@ -Subproject commit 6d9fae25aef473cbfbdc253c0475d5499c959d1b +Subproject commit cf3d56f08075dec7f6bab38395a6d85050b493bb