refactor!: freeze a clean v4 public API surface

CLAUDE.md

@@ -29,6 +29,8 @@ dotnet run --project EnumerableAsyncProcessor.UnitTests -f net10.0 -- --treenode
 
 TUnit test projects compile to executables; VSTest-style `dotnet test --filter` does not work. See the `tunit-testing` skill for full filter syntax.
 
+**TUnit itself depends on this library.** `TUnit.Engine` is compiled against EnumerableAsyncProcessor v3, and the locally built assembly shadows the copy TUnit shipped with (same assembly identity), so removing or changing a public member that TUnit.Engine binds to crashes test discovery with `MissingMethodException` before any test runs. The exact signatures TUnit needs are pinned by `V3BinaryCompatibilityTests` and marked as binary-compat members in the source — do not remove them until TUnit ships a build compiled against v4.
+
 CI (`.github/workflows/dotnet.yml`) runs the `EnumerableAsyncProcessor.Pipeline` project (a ModularPipelines app, `dotnet run -c Release` from that directory), which builds, tests, packs, and — on `main` — publishes to NuGet. Versioning comes from GitVersion (`GitVersion.yml` pins `next-version: 4.0.0`; keep that file present, its absence makes ModularPipelines generate a Mainline config that crashes on GitHub PR merge commits).
 
 ## Architecture
@@ -45,17 +47,17 @@ Processor classes vary along three axes, reflected in naming:
 
 - **Input**: with items (`<TInput>`) vs. execution-count only (non-generic).
 - **Output**: `Result*`-prefixed classes (in `RunnableProcessors/ResultProcessors/`) return values via `IAsyncProcessor<TOutput>` (`GetResultsAsync()`, `GetResultsAsyncEnumerable()`, `GetEnumerableTasks()`); unprefixed classes are fire-and-await (`WaitAsync()`).
-- **Strategy**: `OneAtATime`, `Batch`, `Parallel`, `RateLimitedParallel`, `TimedRateLimitedParallel`.
+- **Strategy**: `OneAtATime`, `Batch`, `Parallel`, `TimedRateLimitedParallel`.
 
-`RunnableProcessors/AsyncEnumerable/` holds parallel variants for `IAsyncEnumerable<T>` sources. File-name suffixes `_1`/`_2` distinguish generic arity (e.g. `BatchAsyncProcessor_1.cs` is `BatchAsyncProcessor<TInput>`).
+`RunnableProcessors/AsyncEnumerable/` holds the `IAsyncEnumerable<T>`-source variants (Parallel, OneAtATime, Batch). File-name suffixes `_1`/`_2` distinguish generic arity (e.g. `BatchAsyncProcessor_1.cs` is `BatchAsyncProcessor<TInput>`).
 
 ### Core mechanics (read these before changing behavior)
 
 - **`ProcessorLifecycle.cs`**: owns start/cancel/dispose shared by both base-class hierarchies. `AbstractAsyncProcessorBase` (void) and `ResultAbstractAsyncProcessorBase` (results) cannot share an ancestor because they fan out to differently typed `TaskCompletionSource` lists, so both delegate to this class. Cancellation is registered in `Start`, not the constructor, so a pre-cancelled token can never fire on a partially built instance. `DisposeAsync` waits up to 30 seconds for in-flight tasks; sync `Dispose` cancels without blocking.
 - **TCS-per-item**: each item gets a `TaskCompletionSource`; `TaskWrapper.Process` never throws — it completes the item's TCS with the failure/cancellation instead, so one failed item cannot kill the run or leave awaiters hanging.
-- **`WorkerPool.cs`**: rate-limited processors run a fixed pool of worker loops claiming items via `Interlocked.Increment` (P `Task.Run` tasks total, not N throttled tasks + semaphore). `minimumIterationTime` is how timed rate limiting is implemented: each worker holds its slot for at least that duration per item.
+- **`WorkerPool.cs`**: rate-limited processors run a fixed pool of worker loops claiming items via `Interlocked.Increment` (P `Task.Run` tasks total, not N throttled tasks + semaphore). Timed rate limiting is a shared `TokenBucketRateLimiter` (`System.Threading.RateLimiting`): workers acquire a permit before starting each item, so `permitsPerWindow`/`window` bound the start rate independently of `maxConcurrency`.
 - **Multi-targeting**: `EnumerableExtensions.ToIAsyncEnumerable` uses `Task.WhenEach` on `NET9_0_OR_GREATER` and a completion-order-bucket fallback otherwise. The test project targets `net8.0` specifically to exercise the fallback path — don't drop that TFM.
 
 ### Disposal contract
 
-All processors implement `IDisposable`/`IAsyncDisposable`; the README documents the patterns users rely on (`await using`, safe double/early disposal). The builder extension shortcuts on `IAsyncEnumerable<T>` dispose internally; processors returned from the builder pattern are the caller's responsibility. Preserve these semantics — there are dedicated regression tests (`DisposalRegressionTests`, `ExceptionFidelityTests`, `InputEnumerationRegressionTests`).
+All processors implement `IDisposable`/`IAsyncDisposable`; the README documents the patterns users rely on (`await using`, safe double/early disposal). `IAsyncEnumerableProcessor` implementations are single-use and additionally dispose their internal linked `CancellationTokenSource` when `ExecuteAsync` completes; `IAsyncProcessor` objects returned from the builder pattern are the caller's responsibility. Preserve these semantics — there are dedicated regression tests (`DisposalRegressionTests`, `ExceptionFidelityTests`, `InputEnumerationRegressionTests`).

EnumerableAsyncProcessor.Example/ProcessInParallelExample.cs

@@ -15,10 +15,10 @@ public static async Task RunExample()
         Console.WriteLine("ProcessInParallel Extension Examples");
         Console.WriteLine("====================================\n");
 
-        // Example 1: Simple parallel processing without transformation
-        Console.WriteLine("Example 1: Simple parallel processing (no transformation needed!)");
+        // Example 1: Simple parallel processing with an identity transformation
+        Console.WriteLine("Example 1: Simple parallel processing");
         var asyncEnumerable1 = GenerateAsyncEnumerable(5);
-        IEnumerable<int> results1 = await asyncEnumerable1.ProcessInParallel();  // <-- This is the simple extension!
+        IEnumerable<int> results1 = await asyncEnumerable1.ProcessInParallel(item => Task.FromResult(item));
         Console.WriteLine($"Results: {string.Join(", ", results1)}");
 
         // Example 2: Parallel processing with transformation

EnumerableAsyncProcessor.UnitTests/AsyncEnumerableParallelExtensionsTests.cs

@@ -32,24 +32,13 @@ private static async IAsyncEnumerable<int> GenerateDelayedAsyncEnumerable(int co
         }
     }
 
-    [Test]
-    public async Task ProcessInParallel_WithoutTransformation_ReturnsAllItems()
-    {
-        var asyncEnumerable = GenerateAsyncEnumerable(10);
-
-        var results = await asyncEnumerable.ProcessInParallel();
-
-        await Assert.That(results.Count()).IsEqualTo(10);
-        await Assert.That(results.OrderBy(x => x)).IsEquivalentTo(Enumerable.Range(1, 10));
-    }
-
     [Test]
     public async Task ProcessInParallel_WithMaxConcurrency_ReturnsAllItems()
     {
         var asyncEnumerable = GenerateAsyncEnumerable(20);
-        
-        var results = await asyncEnumerable.ProcessInParallel(5);
-        
+
+        var results = await asyncEnumerable.ProcessInParallel(item => Task.FromResult(item), 5);
+
         await Assert.That(results.Count()).IsEqualTo(20);
         await Assert.That(results.OrderBy(x => x)).IsEquivalentTo(Enumerable.Range(1, 20));
     }
@@ -94,7 +83,7 @@ public async Task ProcessInParallel_WithCancellation_ThrowsOperationCanceledExce
         using var cts = new CancellationTokenSource();
         var asyncEnumerable = GenerateDelayedAsyncEnumerable(100, 50);
 
-        var task = asyncEnumerable.ProcessInParallel(cancellationToken: cts.Token);
+        var task = asyncEnumerable.ProcessInParallel(item => Task.FromResult(item), cancellationToken: cts.Token);
 
         // Cancel after a short delay
         cts.CancelAfter(100);
@@ -160,23 +149,10 @@ public async Task ProcessInParallel_WithMaxConcurrency_LimitsConcurrency()
     public async Task ProcessInParallel_EmptyEnumerable_ReturnsEmptyResult()
     {
         var asyncEnumerable = GenerateAsyncEnumerable(0);
-
-        var results = await asyncEnumerable.ProcessInParallel();
-
-        await Assert.That(results.Count()).IsEqualTo(0);
-    }
 
-    [Test]
-    public async Task ProcessInParallel_WithScheduleOnThreadPool_ProcessesAllItems()
-    {
-        var asyncEnumerable = GenerateAsyncEnumerable(10);
-
-        var results = await asyncEnumerable.ProcessInParallel(
-            maxConcurrency: null,
-            scheduleOnThreadPool: true);
-
-        await Assert.That(results.Count()).IsEqualTo(10);
-        await Assert.That(results.OrderBy(x => x)).IsEquivalentTo(Enumerable.Range(1, 10));
+        var results = await asyncEnumerable.ProcessInParallel(item => Task.FromResult(item));
+
+        await Assert.That(results.Count()).IsEqualTo(0);
     }
 
     [Test]

EnumerableAsyncProcessor.UnitTests/DisposalRegressionTests.cs

@@ -1,6 +1,8 @@
 using System;
+using System.Collections.Generic;
 using System.Diagnostics;
 using System.Linq;
+using System.Threading;
 using System.Threading.Tasks;
 using EnumerableAsyncProcessor.Extensions;
 
@@ -127,4 +129,132 @@ public async Task Disposal_Is_Idempotent_And_Safe_In_Any_Order()
 
         await Assert.That(processor.GetEnumerableTasks().Count(x => x.IsCompletedSuccessfully)).IsEqualTo(5);
     }
+
+    [Test]
+    public async Task AsyncEnumerable_Processor_Disposal_Is_Idempotent_After_Execution()
+    {
+        var processedCount = 0;
+
+        var processor = GenerateAsyncEnumerable(5)
+            .ForEachAsync(_ =>
+            {
+                Interlocked.Increment(ref processedCount);
+                return Task.CompletedTask;
+            })
+            .ProcessInParallel(maxConcurrency: 2);
+
+        await processor.ExecuteAsync();
+
+        // ExecuteAsync disposes internal resources on completion; explicit disposal stays safe.
+        await processor.DisposeAsync();
+        processor.Dispose();
+
+        await Assert.That(processedCount).IsEqualTo(5);
+    }
+
+    [Test]
+    public async Task AsyncEnumerable_Result_Processor_Supports_Await_Using_Without_Execution()
+    {
+        await using (GenerateAsyncEnumerable(3).SelectAsync(i => Task.FromResult(i)).ProcessInParallel(2))
+        {
+            // Never executed - disposal alone must not throw.
+        }
+    }
+
+    [Test, Timeout(30_000)]
+    public async Task AsyncEnumerable_Processor_Dispose_During_Execution_Cancels_Processing(CancellationToken cancellationToken)
+    {
+        var firstItemStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        var processor = InfiniteAsyncEnumerable()
+            .ForEachAsync(async _ =>
+            {
+                firstItemStarted.TrySetResult();
+                await Task.Yield();
+            })
+            .ProcessInParallel(maxConcurrency: 1);
+
+        var executeTask = processor.ExecuteAsync();
+        await firstItemStarted.Task;
+
+        processor.Dispose();
+
+        Exception? caught = null;
+        try
+        {
+            await executeTask.WaitAsync(TimeSpan.FromSeconds(10), cancellationToken);
+        }
+        catch (Exception exception)
+        {
+            caught = exception;
+        }
+
+        // A TimeoutException here means disposal did not cancel the in-flight run.
+        await Assert.That(caught is OperationCanceledException).IsTrue();
+    }
+
+    [Test, Timeout(30_000)]
+    public async Task AsyncEnumerable_Processor_DisposeAsync_Waits_For_InFlight_Selector(CancellationToken cancellationToken)
+    {
+        var firstItemStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var releaseItem = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var itemsCompleted = 0;
+
+        var processor = InfiniteAsyncEnumerable()
+            .ForEachAsync(async _ =>
+            {
+                firstItemStarted.TrySetResult();
+                await releaseItem.Task;
+                Interlocked.Increment(ref itemsCompleted);
+            })
+            .ProcessInParallel(maxConcurrency: 1);
+
+        var executeTask = processor.ExecuteAsync();
+        await firstItemStarted.Task;
+
+        var disposeTask = processor.DisposeAsync().AsTask();
+
+        // The selector ignores cancellation and is still blocked, so disposal must still be waiting.
+        await Assert.That(disposeTask.IsCompleted).IsFalse();
+
+        releaseItem.TrySetResult();
+        await disposeTask.WaitAsync(TimeSpan.FromSeconds(10), cancellationToken);
+
+        // At least the in-flight item finished before disposal returned; the worker may also
+        // drain one already-buffered item before it observes cancellation.
+        await Assert.That(itemsCompleted).IsGreaterThanOrEqualTo(1);
+
+        Exception? caught = null;
+        try
+        {
+            await executeTask.WaitAsync(TimeSpan.FromSeconds(10), cancellationToken);
+        }
+        catch (Exception exception)
+        {
+            caught = exception;
+        }
+
+        await Assert.That(caught is OperationCanceledException).IsTrue();
+    }
+
+    private static async IAsyncEnumerable<int> InfiniteAsyncEnumerable(
+        [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)
+    {
+        var i = 0;
+        while (true)
+        {
+            cancellationToken.ThrowIfCancellationRequested();
+            yield return i++;
+            await Task.Yield();
+        }
+    }
+
+    private static async IAsyncEnumerable<int> GenerateAsyncEnumerable(int count)
+    {
+        for (var i = 0; i < count; i++)
+        {
+            await Task.Yield();
+            yield return i;
+        }
+    }
 }

EnumerableAsyncProcessor.UnitTests/V3BinaryCompatibilityTests.cs

@@ -0,0 +1,47 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using EnumerableAsyncProcessor.Builders;
+using EnumerableAsyncProcessor.Extensions;
+
+namespace EnumerableAsyncProcessor.UnitTests;
+
+/// <summary>
+/// TUnit.Engine (the framework running this suite) is itself compiled against
+/// EnumerableAsyncProcessor, and the locally built assembly shadows the version TUnit shipped
+/// with because the assembly identity matches. If a member TUnit.Engine binds to disappears,
+/// test DISCOVERY crashes with MissingMethodException before any test runs. These are the
+/// exact signatures TUnit.Engine references - keep them until TUnit rebuilds against v4.
+/// </summary>
+public class V3BinaryCompatibilityTests
+{
+    [Test]
+    public async Task Members_Bound_By_TUnit_Engine_Exist_With_Exact_Signatures()
+    {
+        // ItemActionAsyncProcessorBuilder<TInput, TOutput>.ProcessInParallel(int)
+        var builderMethod = typeof(ItemActionAsyncProcessorBuilder<,>).GetMethods()
+            .SingleOrDefault(m => m.Name == "ProcessInParallel"
+                                  && m.GetParameters().Length == 1
+                                  && m.GetParameters()[0].ParameterType == typeof(int));
+        await Assert.That(builderMethod).IsNotNull();
+
+        // AsyncEnumerableExtensions.ProcessInParallel<T>(IAsyncEnumerable<T>, CancellationToken)
+        var extensionMethod = typeof(AsyncEnumerableExtensions).GetMethods()
+            .SingleOrDefault(m => m.Name == "ProcessInParallel"
+                                  && m.GetGenericArguments().Length == 1
+                                  && m.GetParameters().Length == 2
+                                  && m.GetParameters()[1].ParameterType == typeof(CancellationToken));
+        await Assert.That(extensionMethod).IsNotNull();
+
+        // TUnit.Engine also binds EnumerableExtensions.SelectAsync / SelectManyAsync and
+        // IAsyncProcessor<T>.GetAwaiter; these exact-signature method groups stop compiling if they drift.
+        Func<IEnumerable<int>, Func<int, Task<int>>, CancellationToken, ItemActionAsyncProcessorBuilder<int, int>> selectAsync =
+            EnumerableExtensions.SelectAsync;
+        Func<IEnumerable<int>, Func<int, IAsyncEnumerable<int>>, CancellationToken, IAsyncEnumerable<int>> selectManyAsync =
+            EnumerableExtensions.SelectManyAsync;
+        await Assert.That(selectAsync).IsNotNull();
+        await Assert.That(selectManyAsync).IsNotNull();
+    }
+}

EnumerableAsyncProcessor/Builders/ActionAsyncProcessorBuilder.cs

@@ -4,7 +4,7 @@
 
 namespace EnumerableAsyncProcessor.Builders;
 
-public class ActionAsyncProcessorBuilder
+public sealed class ActionAsyncProcessorBuilder
 {
     private readonly int _count;
     private readonly Func<Task> _taskSelector;
@@ -53,6 +53,16 @@ public IAsyncProcessor ProcessInParallel(int? maxConcurrency = null, bool schedu
     {
         return new ParallelAsyncProcessor(_count, _taskSelector, _cancellationTokenSource, maxConcurrency, scheduleOnThreadPool).StartProcessing();
     }
+
+    /// <summary>
+    /// Processes items in parallel with bounded concurrency. Binary-compatible with assemblies
+    /// compiled against v3 (equivalent to <c>ProcessInParallel(maxConcurrency: n)</c>).
+    /// </summary>
+    [System.ComponentModel.EditorBrowsable(System.ComponentModel.EditorBrowsableState.Never)]
+    public IAsyncProcessor ProcessInParallel(int maxConcurrency)
+    {
+        return ProcessInParallel((int?)maxConcurrency);
+    }
 
     public IAsyncProcessor ProcessOneAtATime()
     {

EnumerableAsyncProcessor/Builders/ActionAsyncProcessorBuilder_1.cs

@@ -4,7 +4,7 @@
 
 namespace EnumerableAsyncProcessor.Builders;
 
-public class ActionAsyncProcessorBuilder<TOutput>
+public sealed class ActionAsyncProcessorBuilder<TOutput>
 {
     private readonly int _count;
     private readonly Func<Task<TOutput>> _taskSelector;
@@ -53,6 +53,16 @@ public IAsyncProcessor<TOutput> ProcessInParallel(int? maxConcurrency = null, bo
     {
         return new ResultParallelAsyncProcessor<TOutput>(_count, _taskSelector, _cancellationTokenSource, maxConcurrency, scheduleOnThreadPool).StartProcessing();
     }
+
+    /// <summary>
+    /// Processes items in parallel with bounded concurrency. Binary-compatible with assemblies
+    /// compiled against v3 (equivalent to <c>ProcessInParallel(maxConcurrency: n)</c>).
+    /// </summary>
+    [System.ComponentModel.EditorBrowsable(System.ComponentModel.EditorBrowsableState.Never)]
+    public IAsyncProcessor<TOutput> ProcessInParallel(int maxConcurrency)
+    {
+        return ProcessInParallel((int?)maxConcurrency);
+    }
 
     public IAsyncProcessor<TOutput> ProcessOneAtATime()
     {

EnumerableAsyncProcessor/Builders/AsyncEnumerableActionAsyncProcessorBuilder_1.cs

@@ -1,9 +1,9 @@
-using EnumerableAsyncProcessor.Extensions;
+using EnumerableAsyncProcessor.Interfaces;
 using EnumerableAsyncProcessor.RunnableProcessors.AsyncEnumerable;
 
 namespace EnumerableAsyncProcessor.Builders;
 
-public class AsyncEnumerableActionAsyncProcessorBuilder<TInput>
+public sealed class AsyncEnumerableActionAsyncProcessorBuilder<TInput>
 {
     private readonly IAsyncEnumerable<TInput> _items;
     private readonly Func<TInput, Task> _taskSelector;
@@ -26,7 +26,9 @@ public AsyncEnumerableActionAsyncProcessorBuilder(
     {
         _items = items;
         _cancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
-        _taskSelector = item => taskSelector(item, _cancellationTokenSource.Token);
+        // Capture the token now: the source may be disposed by the time a late item runs the selector.
+        var processorToken = _cancellationTokenSource.Token;
+        _taskSelector = item => taskSelector(item, processorToken);
     }
 
     /// <summary>

EnumerableAsyncProcessor/Builders/AsyncEnumerableActionAsyncProcessorBuilder_2.cs

@@ -1,9 +1,9 @@
-using EnumerableAsyncProcessor.Extensions;
+using EnumerableAsyncProcessor.Interfaces;
 using EnumerableAsyncProcessor.RunnableProcessors.AsyncEnumerable.ResultProcessors;
 
 namespace EnumerableAsyncProcessor.Builders;
 
-public class AsyncEnumerableActionAsyncProcessorBuilder<TInput, TOutput>
+public sealed class AsyncEnumerableActionAsyncProcessorBuilder<TInput, TOutput>
 {
     private readonly IAsyncEnumerable<TInput> _items;
     private readonly Func<TInput, Task<TOutput>> _taskSelector;
@@ -26,7 +26,9 @@ public AsyncEnumerableActionAsyncProcessorBuilder(
     {
         _items = items;
         _cancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
-        _taskSelector = item => taskSelector(item, _cancellationTokenSource.Token);
+        // Capture the token now: the source may be disposed by the time a late item runs the selector.
+        var processorToken = _cancellationTokenSource.Token;
+        _taskSelector = item => taskSelector(item, processorToken);
     }
 
     /// <summary>