Merge branch 'slow-network-tolerance-ConnectionTimeouts' into 'master'

After commit 555b28f we encountered a lot of...

See merge request vostok-libraries/clusterclient.core!4

Vostok.ClusterClient.Core.Tests/Sending/RequestSender_Tests.cs

@@ -92,7 +92,7 @@ public void Should_send_request_with_transport_when_request_conversion_succeeds(
 
             Send(tokenSource.Token);
 
-            transport.Received(1).SendAsync(absoluteRequest, connectionTimeout, Arg.Any<TimeSpan>(), tokenSource.Token);
+            transport.Received(1).SendAsync(absoluteRequest, Arg.Any<TimeSpan>(), Arg.Any<TimeSpan>(), tokenSource.Token);
         }
 
         [Test]
@@ -105,14 +105,18 @@ public void Should_pass_connection_timeout()
             transport.Received(1).SendAsync(absoluteRequest, connectionTimeout, Arg.Any<TimeSpan>(), Arg.Any<CancellationToken>());
         }
 
-        [Test]
-        public void Should_pass_null_connection_timeout_when_it_is_greater_than_full_timeout()
+        [TestCase(5, TestName = "ConnectionTimeout less than ClusterClientConstants.LastAttemptConnectionTimeBudget")]
+        [TestCase(9, TestName = "ConnectionTimeout greater than ClusterClientConstants.LastAttemptConnectionTimeBudget")]
+        public void Should_correctly_select_connection_timeout_when_it_is_greater_than_full_timeout(int secondsConnectionTimeout)
         {
-            connectionTimeout = 10.Seconds();
+            connectionTimeout = secondsConnectionTimeout.Seconds();
 
             Send();
 
-            transport.Received(1).SendAsync(absoluteRequest, null, Arg.Any<TimeSpan>(), Arg.Any<CancellationToken>());
+            var expectedConnectionTimeout = connectionTimeout > ClusterClientConstants.LastAttemptConnectionTimeBudget
+                ? connectionTimeout
+                : ClusterClientConstants.LastAttemptConnectionTimeBudget;
+            transport.Received(1).SendAsync(absoluteRequest, expectedConnectionTimeout, Arg.Any<TimeSpan>(), Arg.Any<CancellationToken>());
         }
 
         [Test]

Vostok.ClusterClient.Core.Tests/Strategies/ForkingRequestStrategy_Tests.cs

@@ -4,11 +4,11 @@
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
-using Castle.Core.Logging;
 using FluentAssertions;
 using FluentAssertions.Extensions;
 using NSubstitute;
 using NUnit.Framework;
+using Vostok.Clusterclient.Core.Misc;
 using Vostok.Clusterclient.Core.Model;
 using Vostok.Clusterclient.Core.Sending;
 using Vostok.Clusterclient.Core.Strategies;
@@ -217,15 +217,36 @@ public void Should_launch_all_requests_with_configured_connection_timeout()
 
             strategy = new ForkingRequestStrategy(delaysProvider, delaysPlanner, replicas.Length);
 
-            strategy.SendAsync(request, parameters, sender, Budget.WithRemaining(5.Seconds()), replicas, replicas.Length, token);
+            var remainingBudget = 5.Seconds();
+            strategy.SendAsync(request, parameters, sender, Budget.WithRemaining(remainingBudget), replicas, replicas.Length, token);
+
+            for (var i = 0; i < replicas.Length; ++i)
+                CompleteForkingDelay();
+
+            for (var i = 0; i < replicas.Length - 1; ++i)
+                sender.Received(1).SendToReplicaAsync(replicas[i], Arg.Any<Request>(), parameters.ConnectionTimeout, Arg.Any<TimeSpan>(), Arg.Any<CancellationToken>());
+
+            sender.Received(1).SendToReplicaAsync(replicas.Last(), Arg.Any<Request>(), ClusterClientConstants.LastAttemptConnectionTimeBudget, remainingBudget, Arg.Any<CancellationToken>());
+        }
+
+        [Test]
+        public void Should_select_connection_timeout_on_last_attempt_as_max_between_it_and_ClusterClientConstantsLastAttemptConnectionTimeBudget()
+        {
+            sender.ClearReceivedCalls();
+
+            strategy = new ForkingRequestStrategy(delaysProvider, delaysPlanner, replicas.Length);
+
+            var parameters = RequestParameters.Empty.WithConnectionTimeout(100.Seconds());
+            var remainingBudget = 5.Seconds();
+            strategy.SendAsync(request, parameters, sender, Budget.WithRemaining(remainingBudget), replicas, replicas.Length, token);
 
             for (var i = 0; i < replicas.Length; ++i)
                 CompleteForkingDelay();
 
             for (var i = 0; i < replicas.Length - 1; ++i)
                 sender.Received(1).SendToReplicaAsync(replicas[i], Arg.Any<Request>(), parameters.ConnectionTimeout, Arg.Any<TimeSpan>(), Arg.Any<CancellationToken>());
 
-            sender.Received(1).SendToReplicaAsync(replicas.Last(), Arg.Any<Request>(), parameters.ConnectionTimeout, Arg.Any<TimeSpan>(), Arg.Any<CancellationToken>());
+            sender.Received(1).SendToReplicaAsync(replicas.Last(), Arg.Any<Request>(), parameters.ConnectionTimeout, remainingBudget, Arg.Any<CancellationToken>());
         }
 
         [TestCase(0)]

Vostok.ClusterClient.Core.Tests/Strategies/ParallelRequestStrategy_Tests.cs

@@ -8,6 +8,7 @@
 using FluentAssertions.Extensions;
 using NSubstitute;
 using NUnit.Framework;
+using Vostok.Clusterclient.Core.Misc;
 using Vostok.Clusterclient.Core.Model;
 using Vostok.Clusterclient.Core.Sending;
 using Vostok.Clusterclient.Core.Strategies;
@@ -146,7 +147,7 @@ public void Should_fire_initial_requests_to_all_replicas_if_parallelism_level_is
         }
 
         [Test]
-        public void Should_use_configured_connection_timeout_for_all_requests()
+        public void Should_use_ClusterClientConstantsLastAttemptConnectionTimeBudget_connection_timeout_for_all_requests()
         {
             strategy = new ParallelRequestStrategy(int.MaxValue);
 
@@ -158,7 +159,7 @@ public void Should_use_configured_connection_timeout_for_all_requests()
 
             foreach (var replica in replicas)
             {
-                sender.Received(1).SendToReplicaAsync(replica, Arg.Any<Request>(), parameters.ConnectionTimeout, Arg.Any<TimeSpan>(), Arg.Any<CancellationToken>());
+                sender.Received(1).SendToReplicaAsync(replica, Arg.Any<Request>(), ClusterClientConstants.LastAttemptConnectionTimeBudget, Arg.Any<TimeSpan>(), Arg.Any<CancellationToken>());
             }
         }
 

Vostok.ClusterClient.Core.Tests/Strategies/SequentialRequestStrategy_Tests.cs

@@ -5,6 +5,7 @@
 using FluentAssertions.Extensions;
 using NSubstitute;
 using NUnit.Framework;
+using Vostok.Clusterclient.Core.Misc;
 using Vostok.Clusterclient.Core.Model;
 using Vostok.Clusterclient.Core.Sending;
 using Vostok.Clusterclient.Core.Strategies;
@@ -145,10 +146,22 @@ public void Should_limit_request_timeouts_by_remaining_time_budget()
         }
 
         [Test]
-        public void Should_use_configured_connection_timeout()
+        public void Should_use_configured_connection_timeout_and_ClusterClientConstantsLastAttemptConnectionTimeBudget()
         {
             Send(Budget.WithRemaining(1500.Milliseconds()));
 
+            sender.ReceivedCalls().Should().HaveCount(3);
+            sender.Received(1).SendToReplicaAsync(replica1, request, parameters.ConnectionTimeout, Arg.Any<TimeSpan>(), token);
+            sender.Received(1).SendToReplicaAsync(replica2, request, parameters.ConnectionTimeout, Arg.Any<TimeSpan>(), token);
+            sender.Received(1).SendToReplicaAsync(replica3, request, ClusterClientConstants.LastAttemptConnectionTimeBudget, Arg.Any<TimeSpan>(), token);
+        }
+
+        [Test]
+        public void Should_use_configured_connection_timeout_if_it_greater_than_ClusterClientConstantsLastAttemptConnectionTimeBudget()
+        {
+            var parameters = RequestParameters.Empty.WithConnectionTimeout(ClusterClientConstants.LastAttemptConnectionTimeBudget + 100.Milliseconds());
+            strategy.SendAsync(request, parameters, sender, Budget.WithRemaining(1500.Milliseconds()), replicas, replicas.Length, token).GetAwaiter().GetResult();
+
             sender.ReceivedCalls().Should().HaveCount(3);
             sender.Received(1).SendToReplicaAsync(replica1, request, parameters.ConnectionTimeout, Arg.Any<TimeSpan>(), token);
             sender.Received(1).SendToReplicaAsync(replica2, request, parameters.ConnectionTimeout, Arg.Any<TimeSpan>(), token);

Vostok.ClusterClient.Core.Tests/Strategies/SingleReplicaRequestStrategy_Tests.cs

@@ -4,6 +4,7 @@
 using FluentAssertions.Extensions;
 using NSubstitute;
 using NUnit.Framework;
+using Vostok.Clusterclient.Core.Misc;
 using Vostok.Clusterclient.Core.Model;
 using Vostok.Clusterclient.Core.Sending;
 using Vostok.Clusterclient.Core.Strategies;
@@ -49,7 +50,7 @@ public void Should_send_request_to_first_replica_with_correct_parameters()
 
             strategy.SendAsync(request, parameters, sender, budget, replicas, replicas.Length, token).Wait();
 
-            sender.Received().SendToReplicaAsync(replicas[0], request, parameters.ConnectionTimeout, budget.Remaining, token);
+            sender.Received().SendToReplicaAsync(replicas[0], request, ClusterClientConstants.LastAttemptConnectionTimeBudget, budget.Remaining, token);
         }
 
 

Vostok.ClusterClient.Core/Misc/ClusterClientConstants.cs

@@ -0,0 +1,13 @@
+using System;
+
+namespace Vostok.Clusterclient.Core.Misc;
+
+internal static class ClusterClientConstants
+{
+    //(deniaa): We can't use "null" as connection time budget because of a bug in Net6.
+    // Also we can't use TimeBudget.Remaining as connection time budget for the last attempt in strategies
+    // because of HttpMessageHandler cache in Vostok.Clusterclient.Transport.Sockets.SocketsHandlerProvider.
+    // Connection timeout is a key in this cache and we don't want to create a new HttpMessageHandler for each request.
+    // So we want to have only one constant "Infinite" for the whole ClusterClient.
+    public static TimeSpan LastAttemptConnectionTimeBudget = TimeSpan.FromSeconds(7);
+}

Vostok.ClusterClient.Core/Misc/TimeSpanExtensions.cs

@@ -0,0 +1,14 @@
+using System;
+using Vostok.Commons.Time;
+
+namespace Vostok.Clusterclient.Core.Misc;
+
+internal static class TimeSpanExtensions
+{
+    public static TimeSpan? SelectConnectionTimeoutForLastAttempt(TimeSpan lastAttemptConnectionTimeBudget, TimeSpan? connectionTimeoutFromParameters)
+    {
+        return connectionTimeoutFromParameters == null 
+            ? lastAttemptConnectionTimeBudget 
+            : TimeSpanArithmetics.Max(lastAttemptConnectionTimeBudget, connectionTimeoutFromParameters.Value);
+    }
+}

Vostok.ClusterClient.Core/PublicAPI.Unshipped.txt

@@ -0,0 +1,4 @@
+static Vostok.Clusterclient.Core.Misc.ClusterClientConstants.LastAttemptConnectionTimeBudget -> System.TimeSpan
+static Vostok.Clusterclient.Core.Misc.TimeSpanExtensions.Max(System.TimeSpan lastAttemptConnectionTimeBudget, System.TimeSpan? connectionTimeoutFromParameters) -> System.TimeSpan?
+Vostok.Clusterclient.Core.Misc.ClusterClientConstants
+Vostok.Clusterclient.Core.Misc.TimeSpanExtensions

Vostok.ClusterClient.Core/Strategies/ForkingRequestStrategy.cs

@@ -4,6 +4,7 @@
 using System.Threading;
 using System.Threading.Tasks;
 using JetBrains.Annotations;
+using Vostok.Clusterclient.Core.Misc;
 using Vostok.Clusterclient.Core.Model;
 using Vostok.Clusterclient.Core.Sending;
 using Vostok.Clusterclient.Core.Strategies.DelayProviders;
@@ -83,7 +84,11 @@ public async Task SendAsync(Request request, RequestParameters parameters, IRequ
                             break;
                         }
 
-                        LaunchRequest(currentTasks, request, budget, sender, replicasEnumerator, parameters.ConnectionTimeout, linkedCancellationToken);
+                        var connectionAttemptTimeout = i == replicasCount - 1 
+                            ? TimeSpanExtensions.SelectConnectionTimeoutForLastAttempt(ClusterClientConstants.LastAttemptConnectionTimeBudget, parameters.ConnectionTimeout)
+                            : parameters.ConnectionTimeout;
+
+                        LaunchRequest(currentTasks, request, budget, sender, replicasEnumerator, connectionAttemptTimeout, linkedCancellationToken);
 
                         ScheduleForkIfNeeded(currentTasks, request, budget, i, replicasCount, linkedCancellationToken);
 

Vostok.ClusterClient.Core/Strategies/ParallelRequestStrategy.cs

@@ -3,6 +3,7 @@
 using System.Threading;
 using System.Threading.Tasks;
 using JetBrains.Annotations;
+using Vostok.Clusterclient.Core.Misc;
 using Vostok.Clusterclient.Core.Model;
 using Vostok.Clusterclient.Core.Sending;
 
@@ -59,7 +60,8 @@ public async Task SendAsync(Request request, RequestParameters parameters, IRequ
                         if (!replicasEnumerator.MoveNext())
                             throw new InvalidOperationException("Replicas enumerator ended prematurely. This is definitely a bug in code.");
 
-                        currentTasks.Add(sender.SendToReplicaAsync(replicasEnumerator.Current, request, parameters.ConnectionTimeout, budget.Remaining, linkedCancellationToken));
+                        var connectionTimeout = TimeSpanExtensions.SelectConnectionTimeoutForLastAttempt(ClusterClientConstants.LastAttemptConnectionTimeBudget, parameters.ConnectionTimeout);
+                        currentTasks.Add(sender.SendToReplicaAsync(replicasEnumerator.Current, request, connectionTimeout, budget.Remaining, linkedCancellationToken));
                     }
 
                     while (currentTasks.Count > 0)

Vostok.ClusterClient.Core/Strategies/SequentialRequestStrategy.cs

@@ -3,11 +3,11 @@
 using System.Threading;
 using System.Threading.Tasks;
 using JetBrains.Annotations;
+using Vostok.Clusterclient.Core.Misc;
 using Vostok.Clusterclient.Core.Model;
 using Vostok.Clusterclient.Core.Sending;
 using Vostok.Clusterclient.Core.Strategies.TimeoutProviders;
 using Vostok.Commons.Time;
-using Vostok.Logging.Abstractions;
 
 namespace Vostok.Clusterclient.Core.Strategies
 {
@@ -53,7 +53,11 @@ public async Task SendAsync(Request request, RequestParameters parameters, IRequ
 
                 var timeout = TimeSpanArithmetics.Min(timeoutsProvider.GetTimeout(request, budget, currentReplicaIndex++, replicasCount), budget.Remaining);
 
-                var result = await sender.SendToReplicaAsync(replica, request, parameters.ConnectionTimeout, timeout, cancellationToken).ConfigureAwait(false);
+                var connectionAttemptTimeout = currentReplicaIndex == replicasCount
+                    ? TimeSpanExtensions.SelectConnectionTimeoutForLastAttempt(ClusterClientConstants.LastAttemptConnectionTimeBudget, parameters.ConnectionTimeout)
+                    : parameters.ConnectionTimeout;
+
+                var result = await sender.SendToReplicaAsync(replica, request, connectionAttemptTimeout, timeout, cancellationToken).ConfigureAwait(false);
                 if (result.Verdict == ResponseVerdict.Accept)
                     break;
 

Vostok.ClusterClient.Core/Strategies/SingleReplicaRequestStrategy.cs

@@ -4,6 +4,7 @@
 using System.Threading;
 using System.Threading.Tasks;
 using JetBrains.Annotations;
+using Vostok.Clusterclient.Core.Misc;
 using Vostok.Clusterclient.Core.Model;
 using Vostok.Clusterclient.Core.Sending;
 
@@ -23,7 +24,7 @@ public class SingleReplicaRequestStrategy : IRequestStrategy
     {
         /// <inheritdoc />
         public Task SendAsync(Request request, RequestParameters parameters, IRequestSender sender, IRequestTimeBudget budget, IEnumerable<Uri> replicas, int replicasCount, CancellationToken cancellationToken) =>
-            sender.SendToReplicaAsync(replicas.First(), request, parameters.ConnectionTimeout, budget.Remaining, cancellationToken);
+            sender.SendToReplicaAsync(replicas.First(), request, ClusterClientConstants.LastAttemptConnectionTimeBudget, budget.Remaining, cancellationToken);
 
         /// <inheritdoc />
         public override string ToString() => "SingleReplica";

Vostok.ClusterClient.Core/Transport/ConnectionAttemptsTransport.cs

@@ -2,6 +2,7 @@
 using System.Threading;
 using System.Threading.Tasks;
 using JetBrains.Annotations;
+using Vostok.Clusterclient.Core.Misc;
 using Vostok.Clusterclient.Core.Model;
 using Vostok.Commons.Time;
 
@@ -33,7 +34,7 @@ public async Task<Response> SendAsync(Request request, TimeSpan? connectionTimeo
             for (var attempt = 1; attempt <= connectionAttempts; ++attempt)
             {
                 var connectionAttemptTimeout = connectionTimeout == null || timeBudget.Remaining < connectionTimeout
-                    ? (TimeSpan?)null
+                    ? TimeSpanExtensions.SelectConnectionTimeoutForLastAttempt(ClusterClientConstants.LastAttemptConnectionTimeBudget, connectionTimeout)
                     : connectionTimeout.Value;
 
                 var response = await transport.SendAsync(request, connectionAttemptTimeout, timeBudget.Remaining, cancellationToken).ConfigureAwait(false);