fix: preserve async stream exception type (#358)

Author: thomhurst

EnumerableAsyncProcessor.UnitTests/AsyncEnumerableProcessorTests.cs

@@ -278,6 +278,34 @@ public async Task ForEachAsync_WithException_PropagatesException()
         var exception = await Assert.ThrowsAsync<InvalidOperationException>(async () => await task);
         await Assert.That(exception!.Message).IsEqualTo("Test exception");
     }
+
+    [Test, Timeout(10_000)]
+    public async Task ForEachAsync_WithConcurrentExceptions_PropagatesOriginalException(
+        CancellationToken cancellationToken)
+    {
+        var startedCount = 0;
+        var workersStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var releaseWorkers = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        var task = GenerateAsyncEnumerable(2)
+            .ForEachAsync(async _ =>
+            {
+                if (Interlocked.Increment(ref startedCount) == 2)
+                {
+                    workersStarted.TrySetResult();
+                }
+
+                await releaseWorkers.Task.WaitAsync(cancellationToken);
+                throw new InvalidOperationException("Concurrent failure");
+            })
+            .ProcessInParallel(2)
+            .ExecuteAsync();
+
+        await workersStarted.Task.WaitAsync(cancellationToken);
+        releaseWorkers.TrySetResult();
+
+        await Assert.ThrowsAsync<InvalidOperationException>(() => task);
+    }
 
     [Test]
     public async Task ForEachAsync_ProcessInParallel_UnboundedConcurrency_ProcessesAllItems()

EnumerableAsyncProcessor/AsyncEnumerableWorkerPool.cs

@@ -214,14 +214,7 @@ private static void ThrowIfFailed(
     {
         if (exceptions.TryDequeue(out var firstException))
         {
-            if (exceptions.IsEmpty)
-            {
-                ExceptionDispatchInfo.Capture(firstException).Throw();
-            }
-
-            var allExceptions = new List<Exception> { firstException };
-            allExceptions.AddRange(exceptions);
-            throw new AggregateException(allExceptions);
+            ExceptionDispatchInfo.Capture(firstException).Throw();
         }
 
         if (wasCanceled != 0)