From 3cffd675185afbe0faf17552d10007cd898828e7 Mon Sep 17 00:00:00 2001 From: nils2525 Date: Mon, 20 Jul 2026 09:24:01 +0200 Subject: [PATCH] Fixed socket individual subscription target calculation (#282) * Fixed socket individual subscription target calculation * Fixed socket selection when the least-loaded connection has reached its individual subscription limit --- .../ClientTests/SocketClientTests.cs | 62 +++++++++++++++++++ .../Implementations/TestSocketApiClient.cs | 8 ++- CryptoExchange.Net/Clients/SocketApiClient.cs | 58 ++++++++--------- 3 files changed, 95 insertions(+), 33 deletions(-) diff --git a/CryptoExchange.Net.UnitTests/ClientTests/SocketClientTests.cs b/CryptoExchange.Net.UnitTests/ClientTests/SocketClientTests.cs index 5424e9bf..aad91a07 100644 --- a/CryptoExchange.Net.UnitTests/ClientTests/SocketClientTests.cs +++ b/CryptoExchange.Net.UnitTests/ClientTests/SocketClientTests.cs @@ -138,6 +138,68 @@ namespace CryptoExchange.Net.UnitTests.ClientTests Assert.That(socket2.Connected == false); } + [TestCase()] + public async Task BatchedSubscription_Should_NotExceedIndividualCombineTarget() + { + // arrange + var client = new TestSocketClient(options => + { + options.SocketSubscriptionsCombineTarget = 10; + options.SocketIndividualSubscriptionCombineTarget = 10; + }); + TestHelpers.ConfigureSocketClient(client, "wss://localhost"); + + // act + await client.ApiClient1.SubscribeToUpdatesAsync(x => { }, false, default, individualSubscriptionCount: 6); + TestHelpers.ConfigureSocketClient(client, "wss://localhost"); + await client.ApiClient1.SubscribeToUpdatesAsync(x => { }, false, default, individualSubscriptionCount: 6); + + // assert + Assert.That(client.ApiClient1._socketConnections.Count == 2); + Assert.That(client.ApiClient1._socketConnections.Values.All(connection => connection.Subscriptions.Sum(subscription => subscription.IndividualSubscriptionCount) <= 10)); + } + + [TestCase()] + public async Task BatchedSubscription_FullIndividualConnection_Should_NotPreventEligibleConnectionReuse() + { + // arrange + var client = new TestSocketClient(options => + { + options.SocketSubscriptionsCombineTarget = 5; + options.SocketIndividualSubscriptionCombineTarget = 10; + }); + TestHelpers.ConfigureSocketClient(client, "wss://localhost"); + await client.ApiClient1.SubscribeToUpdatesAsync(x => { }, false, default); + await client.ApiClient1.SubscribeToUpdatesAsync(x => { }, false, default); + + TestHelpers.ConfigureSocketClient(client, "wss://localhost"); + await client.ApiClient1.SubscribeToUpdatesAsync(x => { }, false, default, individualSubscriptionCount: 10); + + TestHelpers.ConfigureSocketClient(client, "wss://localhost"); + + // act + await client.ApiClient1.SubscribeToUpdatesAsync(x => { }, false, default); + + // assert + Assert.That( + client.ApiClient1._socketConnections.Count, + Is.EqualTo(2), + "The eligible connection should be reused instead of opening a new connection after selecting a full individual-subscription connection"); + + var fullConnection = client.ApiClient1._socketConnections.Values + .Single(connection => connection.Subscriptions.Sum(subscription => subscription.IndividualSubscriptionCount) == 10); + Assert.That( + fullConnection.UserSubscriptionCount, + Is.EqualTo(1), + "The full connection should not receive the normal subscription"); + + var eligibleConnection = client.ApiClient1._socketConnections.Values.Single(connection => connection != fullConnection); + Assert.That( + eligibleConnection.UserSubscriptionCount, + Is.EqualTo(3), + "The existing eligible connection should receive the normal subscription"); + } + [TestCase()] public async Task ErrorResponse_ShouldNot_ConfirmSubscription() { diff --git a/CryptoExchange.Net.UnitTests/Implementations/TestSocketApiClient.cs b/CryptoExchange.Net.UnitTests/Implementations/TestSocketApiClient.cs index 4dece6ab..eec63c13 100644 --- a/CryptoExchange.Net.UnitTests/Implementations/TestSocketApiClient.cs +++ b/CryptoExchange.Net.UnitTests/Implementations/TestSocketApiClient.cs @@ -36,9 +36,13 @@ namespace CryptoExchange.Net.UnitTests.Implementations protected override TestAuthenticationProvider CreateAuthenticationProvider(TestCredentials credentials) => new TestAuthenticationProvider(credentials); - public async Task> SubscribeToUpdatesAsync(Action> handler, bool subQuery, CancellationToken ct) + public async Task> SubscribeToUpdatesAsync(Action> handler, bool subQuery, CancellationToken ct, int individualSubscriptionCount = 1) { - return await base.SubscribeAsync(new TestSubscription(_logger, handler, subQuery, false), ct); + var subscription = new TestSubscription(_logger, handler, subQuery, false) + { + IndividualSubscriptionCount = individualSubscriptionCount + }; + return await base.SubscribeAsync(subscription, ct); } } } diff --git a/CryptoExchange.Net/Clients/SocketApiClient.cs b/CryptoExchange.Net/Clients/SocketApiClient.cs index d80def1d..aebfc00e 100644 --- a/CryptoExchange.Net/Clients/SocketApiClient.cs +++ b/CryptoExchange.Net/Clients/SocketApiClient.cs @@ -679,45 +679,26 @@ namespace CryptoExchange.Net.Clients && (s.Authenticated == authenticated || !authenticated) && s.Connected).ToList(); - SocketConnection? connection; - if (!dedicatedRequestConnection) - { - connection = socketQuery - .Where(s => !s.DedicatedRequestConnection.IsDedicatedRequestConnection) - .OrderBy(s => s.UserSubscriptionCount) - .FirstOrDefault(); - } - else + bool maxConnectionsReached = _socketConnections.Count >= (ApiOptions.MaxSocketConnections ?? ClientOptions.MaxSocketConnections); + SocketConnection? connection = null; + if (dedicatedRequestConnection) { connection = socketQuery.Where(s => s.DedicatedRequestConnection.IsDedicatedRequestConnection).FirstOrDefault(); if (connection != null && !connection.DedicatedRequestConnection.Authenticated) // Mark dedicated request connection as authenticated if the request is authenticated connection.DedicatedRequestConnection.Authenticated = authenticated; - - if (connection == null) - // Fall back to an existing connection if there is no dedicated request connection available - connection = socketQuery.OrderBy(s => s.UserSubscriptionCount).FirstOrDefault(); } - bool maxConnectionsReached = _socketConnections.Count >= (ApiOptions.MaxSocketConnections ?? ClientOptions.MaxSocketConnections); + if (connection == null) + // Use an eligible non-dedicated connection for subscriptions, or as fallback when no dedicated request connection is available + connection = socketQuery + .Where(s => !s.DedicatedRequestConnection.IsDedicatedRequestConnection) + .Where(s => IsConnectionEligible(s, individualSubscriptionCount, maxConnectionsReached)) + .OrderBy(s => s.UserSubscriptionCount) + .FirstOrDefault(); + if (connection != null) - { - bool lessThanBatchSubCombineTarget = connection.UserSubscriptionCount < ClientOptions.SocketSubscriptionsCombineTarget; - bool lessThanIndividualSubCombineTarget = connection.Subscriptions.Sum(x => x.IndividualSubscriptionCount) < ClientOptions.SocketIndividualSubscriptionCombineTarget; - - if ((lessThanBatchSubCombineTarget && lessThanIndividualSubCombineTarget) - || maxConnectionsReached) - { - // Use existing socket if it has less than target connections OR it has the least connections and we can't make new - // If there is a max subscriptions per connection limit also only use existing if the new subscription doesn't go over the limit - if (MaxIndividualSubscriptionsPerConnection == null) - return CallResult.Ok(connection); - - var currentCount = connection.Subscriptions.Sum(x => x.IndividualSubscriptionCount); - if (currentCount + individualSubscriptionCount <= MaxIndividualSubscriptionsPerConnection) - return CallResult.Ok(connection); - } - } + return CallResult.Ok(connection); if (maxConnectionsReached) return CallResult.Fail(new InvalidOperationError("Max amount of socket connections reached")); @@ -784,6 +765,21 @@ namespace CryptoExchange.Net.Clients return CallResult.Ok(socketConnection); } + private bool IsConnectionEligible(SocketConnection socketConnection, int individualSubscriptionCount, bool maxConnectionsReached) + { + var currentIndividualSubscriptionCount = socketConnection.Subscriptions.Sum(x => x.IndividualSubscriptionCount); + bool lessThanBatchSubCombineTarget = socketConnection.UserSubscriptionCount < ClientOptions.SocketSubscriptionsCombineTarget; + // Include the incoming batch so batched subscriptions cannot overshoot the configured socket target. + bool lessThanIndividualSubCombineTarget = currentIndividualSubscriptionCount + individualSubscriptionCount <= ClientOptions.SocketIndividualSubscriptionCombineTarget; + + if ((!lessThanBatchSubCombineTarget || !lessThanIndividualSubCombineTarget) + && !maxConnectionsReached) + return false; + + return MaxIndividualSubscriptionsPerConnection == null + || currentIndividualSubscriptionCount + individualSubscriptionCount <= MaxIndividualSubscriptionsPerConnection; + } + /// /// Process an unhandled message ///