Merge pull request #169 from GetStream/feature/improve-tests-concurrent-stability

Feature/improve tests concurrent stability

Author: sierpinskid

Assets/Plugins/StreamChat/Core/LowLevelClient/API/Internal/InternalApiClientBase.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Linq;
 using System.Text;
 using System.Threading.Tasks;
 using StreamChat.Core.Exceptions;
@@ -83,7 +84,7 @@ private object TrySerializeRequestBodyContent(object content, out string seriali
         }
 
         private async Task<TResponse> HttpRequest<TResponse>(HttpMethodType httpMethod, string endpoint,
-            object requestBody = default, QueryParameters queryParameters = null)
+            object requestBody = default, QueryParameters queryParameters = null, int attempt = 0)
         {
             //StreamTodo: perhaps remove this requirement, sometimes we send empty body without any properties
             if (requestBody == null && IsRequestBodyRequiredByHttpMethod(httpMethod))
@@ -101,7 +102,14 @@ private async Task<TResponse> HttpRequest<TResponse>(HttpMethodType httpMethod,
                     queryParameters = QueryParameters.Default;
                 }
 
-                queryParameters.Append("payload", serializedContent);
+                if (queryParameters.ContainsKey("paload"))
+                {
+                    queryParameters["payload"] = serializedContent;
+                }
+                else
+                {
+                    queryParameters.Append("payload", serializedContent);
+                }
             }
 
             var uri = _requestUriFactory.CreateEndpointUri(endpoint, queryParameters);
@@ -129,7 +137,19 @@ private async Task<TResponse> HttpRequest<TResponse>(HttpMethodType httpMethod,
                             Message = responseContent,
                             Code = 504,
                         };
+
+#if !STREAM_TESTS_ENABLED
                         throw new StreamApiException(apiError);
+#else
+                        if (attempt >= 20)
+                        {
+                            throw new StreamApiException(apiError);
+                        }
+
+                        _logs.Warning($"API CLIENT, TESTS MODE, Upstream Request Timeout - Make another attempt");
+                        return await HttpRequest<TResponse>(httpMethod, endpoint,
+                            requestBody, queryParameters, ++attempt);
+#endif
                     }
 
                     LogRestCall(uri, endpoint, httpMethod, responseContent, success: false, logContent);
@@ -146,6 +166,14 @@ private async Task<TResponse> HttpRequest<TResponse>(HttpMethodType httpMethod,
                     throw new StreamDeserializationException(responseContent, typeof(TResponse), e);
                 }
 
+#if STREAM_TESTS_ENABLED
+                if (apiError.StatusCode == StreamApiException.RateLimitErrorHttpStatusCode && attempt < 50)
+                {
+                    return await HandleRateLimit<TResponse>(httpMethod, endpoint, requestBody, queryParameters, attempt,
+                        httpResponse);
+                }
+#endif
+
                 if (apiError.Code != InvalidAuthTokenErrorCode)
                 {
                     LogRestCall(uri, endpoint, httpMethod, responseContent, success: false, logContent);
@@ -273,5 +301,45 @@ private void LogRestCall(Uri uri, string endpoint, HttpMethodType httpMethod, st
 
             _logs.Info(_sb.ToString());
         }
+
+        private async Task<TResponse> HandleRateLimit<TResponse>(HttpMethodType httpMethod, string endpoint,
+            object requestBody, QueryParameters queryParameters, int attempt, HttpResponse httpResponse)
+        {
+            if (attempt >= 50)
+            {
+                throw new StreamApiException(new APIErrorInternalDTO
+                    { Code = StreamApiException.RateLimitErrorHttpStatusCode });
+            }
+
+            var delaySeconds = GetBackoffDelay(attempt, httpResponse, out var resetHeaderTimestamp);
+            var now = (int)new DateTimeOffset(DateTime.UtcNow).ToUnixTimeSeconds();
+            _logs.Warning($"API CLIENT, TESTS MODE, Rate Limit API Error - Wait for {delaySeconds} seconds. " +
+                          $"Timestamp reset header: {resetHeaderTimestamp}, Current timestamp: {now}, Dif: {resetHeaderTimestamp - now}");
+            await Task.Delay(delaySeconds * 1000);
+            return await HttpRequest<TResponse>(httpMethod, endpoint, requestBody, queryParameters, ++attempt);
+        }
+
+        private int GetBackoffDelay(int attempt, HttpResponse httpResponse, out int resetHeaderTimestamp)
+        {
+            resetHeaderTimestamp = -1;
+            // StreamTodo: Backoff based on the header doesn't seem to work. Perhaps concurrency is conflicting with this approach
+            if (httpResponse.TryGetHeader("x-ratelimit-reset", out var values))
+            {
+                var resetTimestamp = values.FirstOrDefault();
+
+                if (int.TryParse(resetTimestamp, out var rateLimitTimestamp))
+                {
+                    resetHeaderTimestamp = rateLimitTimestamp;
+                    var now = (int)new DateTimeOffset(DateTime.UtcNow).ToUnixTimeSeconds();
+                    var secondsLeft = rateLimitTimestamp - now;
+                    // if (secondsLeft > 0)
+                    // {
+                    //     return secondsLeft + 5;
+                    // }
+                }
+            }
+
+            return 61 + attempt * 20;
+        }
     }
 }