Serialize RateLimits.NextRate to stop token-bucket runaway (abuzuhri#940)

A single RateLimits instance is shared across all concurrent requests of a
given RateLimitType. With no synchronization, parallel calls race on the
token-bucket state: when the bucket is full, each thread advances LastRequest
by ratePeriodMs, so the wait target keeps receding and the limiter never
recovers until the process restarts (the reported "stuck until restart"
symptom). The logic is correct single-threaded; the bug is purely concurrency.

Gate NextRate behind a per-instance SemaphoreSlim (logic moved to NextRateCore).
Each RateLimitType owns its own instance/gate, so distinct endpoints do not
block each other.

Also repair RateLimitsTests, which referenced the rolled-back
WaitForPermittedRequest API and no longer compiled, and add a concurrency
regression test that drives 10 parallel requests and asserts they drain within
a bounded window.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: abuzuhri

Source/FikaAmazonAPI/Utils/RateLimits.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Runtime.CompilerServices;
+using System.Threading;
 using System.Threading.Tasks;
 
 [assembly: InternalsVisibleTo("Tests")]
@@ -12,6 +13,15 @@ internal class RateLimits
         internal DateTime LastRequest { get; set; }
         internal int RequestsSent { get; set; }
 
+        /// <summary>
+        /// A single shared <see cref="RateLimits"/> instance is reused for every concurrent request of the
+        /// same <see cref="RateLimitType"/>. Without serialization, parallel calls race on <see cref="RequestsSent"/>
+        /// and <see cref="LastRequest"/> — each thread that sees a full bucket advances <see cref="LastRequest"/>
+        /// further into the future, so the wait target keeps receding and the limiter never recovers until the
+        /// process is restarted. The gate forces one request through the token-bucket logic at a time.
+        /// </summary>
+        private readonly SemaphoreSlim _gate = new SemaphoreSlim(1, 1);
+
         /// <summary>
         /// Constructor for rate limits configuration object
         /// </summary>
@@ -36,6 +46,22 @@ internal RateLimits(decimal Rate, int Burst)
         /// <param name="cancellationToken">The cancellation token</param>
         /// <returns></returns>
         public async Task<RateLimits> NextRate(RateLimitType rateLimitType)
+        {
+            // Serialize access so concurrent requests for the same rate limit type cannot corrupt the
+            // shared token-bucket state (see _gate). Each rate limit type owns its own instance/gate, so
+            // requests to different resources are unaffected.
+            await _gate.WaitAsync();
+            try
+            {
+                return await NextRateCore(rateLimitType);
+            }
+            finally
+            {
+                _gate.Release();
+            }
+        }
+
+        private async Task<RateLimits> NextRateCore(RateLimitType rateLimitType)
         {
             if (RequestsSent < 0)
                 RequestsSent = 0;

Source/Tests/RateLimitsTests.cs

@@ -18,16 +18,15 @@ public class RateLimitsTests
         public async Task WaitForPermittedRequest_WaitsExpectedLengthOfTime(decimal rate, int burst, int numberRequests, int expectedWaitMilliseconds)
         {
             // Arrange
-            var rateLimit = new RateLimits(rate, burst, RateLimitType.UNSET);
+            var rateLimit = new RateLimits(rate, burst);
 
             var stopwatch = new Stopwatch();
-            var cancellationToken = new CancellationToken();
 
             // Act
             stopwatch.Start();
             for (int i = 0; i < numberRequests; i++)
             {
-                await rateLimit.WaitForPermittedRequest(cancellationToken, debugMode: true);
+                await rateLimit.NextRate(RateLimitType.UNSET);
             }
             stopwatch.Stop();
 
@@ -36,5 +35,28 @@ public async Task WaitForPermittedRequest_WaitsExpectedLengthOfTime(decimal rate
             // allow a second for additional time taken by the test to run
             Assert.That(stopwatch.ElapsedMilliseconds, Is.LessThanOrEqualTo(expectedWaitMilliseconds + 1000));
         }
+
+        // Regression test for issue #940: when many requests for the same rate limit type run in parallel
+        // against the shared RateLimits instance, the token bucket must drain within the expected window.
+        // Before NextRate was serialized, racing threads each pushed LastRequest further into the future,
+        // so the limiter never recovered and effectively hung until the process was restarted.
+        [Test]
+        public async Task ConcurrentRequests_DrainWithinExpectedWindow()
+        {
+            // rate 1/s, burst 5 => 10 concurrent requests = 5 immediate + 5 spaced ~1s apart ~= 5s.
+            var rateLimit = new RateLimits(1.0M, 5);
+
+            var stopwatch = Stopwatch.StartNew();
+
+            var tasks = Enumerable.Range(0, 10)
+                .Select(_ => rateLimit.NextRate(RateLimitType.UNSET))
+                .ToArray();
+            await Task.WhenAll(tasks);
+
+            stopwatch.Stop();
+
+            Assert.That(stopwatch.ElapsedMilliseconds, Is.GreaterThanOrEqualTo(4000));
+            Assert.That(stopwatch.ElapsedMilliseconds, Is.LessThanOrEqualTo(8000));
+        }
     }
 }