feat: add token-bucket rate limiting (#357)

Timed processors limit operation starts per window. New overload separates permit count from maximum in-flight concurrency.

Closes #335

Author: thomhurst

EnumerableAsyncProcessor.UnitTests/ValidationRegressionTests.cs

@@ -50,6 +50,9 @@ await AssertThrows<ArgumentOutOfRangeException>(() =>
         // The result variant previously skipped this validation entirely
         await AssertThrows<ArgumentOutOfRangeException>(() =>
             new[] { 1 }.SelectAsync(i => Task.FromResult(i)).ProcessInParallel(0, TimeSpan.FromSeconds(1)));
+
+        await AssertThrows<ArgumentOutOfRangeException>(() =>
+            new[] { 1 }.SelectAsync(i => Task.FromResult(i)).ProcessInParallel(1, TimeSpan.FromSeconds(1), 0));
     }
 
     [Test]

EnumerableAsyncProcessor.UnitTests/WorkerPoolBehaviourTests.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Concurrent;
+using System.Diagnostics;
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
@@ -133,6 +134,71 @@ public async Task Timed_RateLimited_Processor_Cancels_Unprocessed_Items_Promptly
         await processor.DisposeAsync();
     }
 
+    [Test, Timeout(10_000)]
+    public async Task Timed_Rate_Limit_Allows_Concurrency_Independent_Of_Permit_Count(CancellationToken cancellationToken)
+    {
+        const int itemCount = 6;
+
+        var startedCount = 0;
+        var allStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        await using var processor = Enumerable.Range(0, itemCount).ToList()
+            .ForEachAsync(async _ =>
+            {
+                if (Interlocked.Increment(ref startedCount) == itemCount)
+                {
+                    allStarted.TrySetResult();
+                }
+
+                await release.Task;
+            }, cancellationToken)
+            .ProcessInParallel(
+                permitsPerWindow: 2,
+                window: TimeSpan.FromMilliseconds(100),
+                maxConcurrency: itemCount);
+
+        try
+        {
+            await allStarted.Task.WaitAsync(TimeSpan.FromSeconds(3), cancellationToken);
+        }
+        finally
+        {
+            release.TrySetResult();
+        }
+
+        await processor.WaitAsync();
+
+        await Assert.That(startedCount).IsEqualTo(itemCount);
+    }
+
+    [Test, Retry(3), Timeout(10_000)]
+    public async Task Timed_Rate_Limit_Does_Not_Exceed_Permits_At_Replenishment(CancellationToken cancellationToken)
+    {
+        const int permitsPerWindow = 3;
+        var window = TimeSpan.FromMilliseconds(200);
+        var startedAt = new ConcurrentBag<TimeSpan>();
+        var stopwatch = Stopwatch.StartNew();
+
+        await using var processor = Enumerable.Range(0, 9).ToList()
+            .ForEachAsync(_ =>
+            {
+                startedAt.Add(stopwatch.Elapsed);
+                return Task.CompletedTask;
+            }, cancellationToken)
+            .ProcessInParallel(permitsPerWindow, window, maxConcurrency: 9);
+
+        await processor.WaitAsync();
+
+        var orderedStarts = startedAt.OrderBy(x => x).ToArray();
+
+        for (var i = permitsPerWindow; i < orderedStarts.Length; i++)
+        {
+            await Assert.That(orderedStarts[i] - orderedStarts[i - permitsPerWindow])
+                .IsGreaterThan(window / 2);
+        }
+    }
+
     [Test]
     public async Task Result_Order_Is_Preserved_Regardless_Of_Completion_Order()
     {

EnumerableAsyncProcessor/Builders/ActionAsyncProcessorBuilder.cs

@@ -31,7 +31,16 @@ public IAsyncProcessor ProcessInBatches(int batchSize)
 
     public IAsyncProcessor ProcessInParallel(int maxConcurrency, TimeSpan timeSpan)
     {
-        return new TimedRateLimitedParallelAsyncProcessor(_count, _taskSelector, maxConcurrency, timeSpan, _cancellationTokenSource).StartProcessing();
+        return ProcessInParallel(maxConcurrency, timeSpan, maxConcurrency);
+    }
+
+    /// <summary>Processes tasks with independent start-rate and concurrency limits.</summary>
+    /// <param name="permitsPerWindow">Maximum operations that may start in each window.</param>
+    /// <param name="window">Rate-limit replenishment window.</param>
+    /// <param name="maxConcurrency">Maximum operations that may remain in flight.</param>
+    public IAsyncProcessor ProcessInParallel(int permitsPerWindow, TimeSpan window, int maxConcurrency)
+    {
+        return new TimedRateLimitedParallelAsyncProcessor(_count, _taskSelector, permitsPerWindow, window, maxConcurrency, _cancellationTokenSource).StartProcessing();
     }
 
     /// <summary>

EnumerableAsyncProcessor/Builders/ActionAsyncProcessorBuilder_1.cs

@@ -31,7 +31,16 @@ public IAsyncProcessor<TOutput> ProcessInBatches(int batchSize)
 
     public IAsyncProcessor<TOutput> ProcessInParallel(int maxConcurrency, TimeSpan timeSpan)
     {
-        return new ResultTimedRateLimitedParallelAsyncProcessor<TOutput>(_count, _taskSelector, maxConcurrency, timeSpan, _cancellationTokenSource).StartProcessing();
+        return ProcessInParallel(maxConcurrency, timeSpan, maxConcurrency);
+    }
+
+    /// <summary>Processes tasks with independent start-rate and concurrency limits.</summary>
+    /// <param name="permitsPerWindow">Maximum operations that may start in each window.</param>
+    /// <param name="window">Rate-limit replenishment window.</param>
+    /// <param name="maxConcurrency">Maximum operations that may remain in flight.</param>
+    public IAsyncProcessor<TOutput> ProcessInParallel(int permitsPerWindow, TimeSpan window, int maxConcurrency)
+    {
+        return new ResultTimedRateLimitedParallelAsyncProcessor<TOutput>(_count, _taskSelector, permitsPerWindow, window, maxConcurrency, _cancellationTokenSource).StartProcessing();
     }
 
     /// <summary>

EnumerableAsyncProcessor/Builders/ItemActionAsyncProcessorBuilder_1.cs

@@ -32,7 +32,16 @@ public IAsyncProcessor ProcessInBatches(int batchSize)
 
     public IAsyncProcessor ProcessInParallel(int maxConcurrency, TimeSpan timeSpan)
     {
-        return new TimedRateLimitedParallelAsyncProcessor<TInput>(_items, _taskSelector, maxConcurrency, timeSpan, _cancellationTokenSource)
+        return ProcessInParallel(maxConcurrency, timeSpan, maxConcurrency);
+    }
+
+    /// <summary>Processes items with independent start-rate and concurrency limits.</summary>
+    /// <param name="permitsPerWindow">Maximum operations that may start in each window.</param>
+    /// <param name="window">Rate-limit replenishment window.</param>
+    /// <param name="maxConcurrency">Maximum operations that may remain in flight.</param>
+    public IAsyncProcessor ProcessInParallel(int permitsPerWindow, TimeSpan window, int maxConcurrency)
+    {
+        return new TimedRateLimitedParallelAsyncProcessor<TInput>(_items, _taskSelector, permitsPerWindow, window, maxConcurrency, _cancellationTokenSource)
             .StartProcessing();
     }
 

EnumerableAsyncProcessor/Builders/ItemActionAsyncProcessorBuilder_2.cs

@@ -52,7 +52,19 @@ public IAsyncProcessor<TOutput> ProcessInBatches(int batchSize)
     /// </remarks>
     public IAsyncProcessor<TOutput> ProcessInParallel(int maxConcurrency, TimeSpan timeSpan)
     {
-        return new ResultTimedRateLimitedParallelAsyncProcessor<TInput, TOutput>(_items, _taskSelector, maxConcurrency, timeSpan, _cancellationTokenSource).StartProcessing();
+        return ProcessInParallel(maxConcurrency, timeSpan, maxConcurrency);
+    }
+
+    /// <summary>
+    /// Processes items with independent start-rate and concurrency limits.
+    /// </summary>
+    /// <param name="permitsPerWindow">Maximum operations that may start in each window.</param>
+    /// <param name="window">Rate-limit replenishment window.</param>
+    /// <param name="maxConcurrency">Maximum operations that may remain in flight.</param>
+    /// <returns>An async processor that implements IDisposable and IAsyncDisposable.</returns>
+    public IAsyncProcessor<TOutput> ProcessInParallel(int permitsPerWindow, TimeSpan window, int maxConcurrency)
+    {
+        return new ResultTimedRateLimitedParallelAsyncProcessor<TInput, TOutput>(_items, _taskSelector, permitsPerWindow, window, maxConcurrency, _cancellationTokenSource).StartProcessing();
     }
 
     /// <summary>

EnumerableAsyncProcessor/EnumerableAsyncProcessor.csproj

@@ -29,6 +29,7 @@
         <None Include="$(MSBuildThisFileDirectory)..\README.md" Pack="true" PackagePath="\" />
 
         <PackageReference Include="Microsoft.SourceLink.GitHub" Version="10.0.301" PrivateAssets="All"/>
+        <PackageReference Include="System.Threading.RateLimiting" Version="10.0.10" />
 
     </ItemGroup>
 

EnumerableAsyncProcessor/RunnableProcessors/ParallelAsyncProcessor.cs

@@ -46,6 +46,6 @@ await Task.WhenAll(TaskWrappers.Select(taskWrapper =>
 
         // Throttled processing runs on a fixed worker pool: P worker tasks instead of
         // one queued task, closure and semaphore wait per item
-        await WorkerPool.ProcessAsync(TaskWrappers, _maxConcurrency.Value, minimumIterationTime: null, CancellationToken).ConfigureAwait(false);
+        await WorkerPool.ProcessAsync(TaskWrappers, _maxConcurrency.Value, rateLimiter: null, CancellationToken).ConfigureAwait(false);
     }
 }

EnumerableAsyncProcessor/RunnableProcessors/ParallelAsyncProcessor_1.cs

@@ -46,6 +46,6 @@ await Task.WhenAll(TaskWrappers.Select(taskWrapper =>
 
         // Throttled processing runs on a fixed worker pool: P worker tasks instead of
         // one queued task, closure and semaphore wait per item
-        await WorkerPool.ProcessAsync(TaskWrappers, _maxConcurrency.Value, minimumIterationTime: null, CancellationToken).ConfigureAwait(false);
+        await WorkerPool.ProcessAsync(TaskWrappers, _maxConcurrency.Value, rateLimiter: null, CancellationToken).ConfigureAwait(false);
     }
 }

EnumerableAsyncProcessor/RunnableProcessors/ResultProcessors/ResultParallelAsyncProcessor_1.cs

@@ -46,6 +46,6 @@ await Task.WhenAll(TaskWrappers.Select(taskWrapper =>
 
         // Throttled processing runs on a fixed worker pool: P worker tasks instead of
         // one queued task, closure and semaphore wait per item
-        await WorkerPool.ProcessAsync(TaskWrappers, _maxConcurrency.Value, minimumIterationTime: null, CancellationToken).ConfigureAwait(false);
+        await WorkerPool.ProcessAsync(TaskWrappers, _maxConcurrency.Value, rateLimiter: null, CancellationToken).ConfigureAwait(false);
     }
 }