fix: unify streaming exception fidelity and prevent cancellation hang (#360)

Streaming results via GetResultsAsyncEnumerable used task.Result on the
net9+ Task.WhenEach path, wrapping failures in AggregateException while
the net8 fallback rethrew the original exception. Both paths now await
the completed task so every TFM surfaces identical exceptions.

In the bounded async-enumerable result pipeline, a worker observing
cancellation between an item being written and claimed abandoned that
item's completion source, hanging the consumer forever. The producer now
awaits pending results with WaitAsync(pipelineToken), workers cancel any
unclaimed items they strand on exit, and abandoned faulted results are
observed so they cannot raise UnobservedTaskException.

Author: thomhurst

EnumerableAsyncProcessor.UnitTests/CancellationRegressionTests.cs

@@ -0,0 +1,70 @@
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using EnumerableAsyncProcessor.Extensions;
+
+namespace EnumerableAsyncProcessor.UnitTests;
+
+/// <summary>
+/// Guards against cancellation hanging the bounded async-enumerable result pipeline:
+/// a worker that observed cancellation between an item being written to the channel
+/// and that item being claimed used to abandon the item's completion source, leaving
+/// the consumer awaiting a task that never completes.
+/// </summary>
+public class CancellationRegressionTests
+{
+    [Test, Timeout(120_000)]
+    public async Task Cancelling_Bounded_Result_Streaming_Never_Hangs(CancellationToken cancellationToken)
+    {
+        for (var iteration = 0; iteration < 100; iteration++)
+        {
+            using var cts = new CancellationTokenSource();
+            var cancelAfter = 1 + iteration % 10;
+            var consumed = 0;
+
+            var stream = InfiniteSource()
+                .SelectAsync(async item =>
+                {
+                    await Task.Yield();
+                    return item;
+                }, cts.Token)
+                .ProcessInParallel(maxConcurrency: 2);
+
+            async Task ConsumeAsync()
+            {
+                await foreach (var _ in stream.ExecuteAsync())
+                {
+                    if (Interlocked.Increment(ref consumed) == cancelAfter)
+                    {
+                        cts.Cancel();
+                    }
+                }
+            }
+
+            Exception? caught = null;
+            try
+            {
+                await ConsumeAsync().WaitAsync(TimeSpan.FromSeconds(10), cancellationToken);
+            }
+            catch (Exception exception)
+            {
+                caught = exception;
+            }
+
+            // A TimeoutException here means the pipeline hung instead of observing cancellation.
+            await Assert.That(caught is OperationCanceledException).IsTrue();
+        }
+    }
+
+    // Deliberately ignores cancellation so the pipeline's own guards are what stop it.
+    private static async IAsyncEnumerable<int> InfiniteSource()
+    {
+        var i = 0;
+        while (true)
+        {
+            yield return i++;
+            await Task.Yield();
+        }
+    }
+}

EnumerableAsyncProcessor.UnitTests/ExceptionFidelityTests.cs

@@ -100,6 +100,57 @@ public async Task Failed_Items_Do_Not_Prevent_Remaining_Items_From_Processing()
         await Assert.That(processor.GetEnumerableTasks().Count(x => x.IsFaulted)).IsEqualTo(10);
     }
 
+    [Test]
+    public async Task Streaming_Results_Surface_The_Original_Exception_Unwrapped()
+    {
+        await using var processor = Enumerable.Range(0, 5).ToList()
+            .SelectAsync(i => i == 2
+                ? Task.FromException<int>(new InvalidOperationException("item 2 failed"))
+                : Task.FromResult(i))
+            .ProcessInParallel(2);
+
+        Exception? caught = null;
+        try
+        {
+            await foreach (var _ in processor.GetResultsAsyncEnumerable())
+            {
+            }
+        }
+        catch (Exception exception)
+        {
+            caught = exception;
+        }
+
+        await Assert.That(caught).IsNotNull();
+        await Assert.That(caught!.GetType()).IsEqualTo(typeof(InvalidOperationException));
+        await Assert.That(caught.Message).IsEqualTo("item 2 failed");
+    }
+
+    [Test]
+    public async Task Streaming_Results_Surface_Cancellation_As_OperationCanceledException_Not_AggregateException()
+    {
+        using var itemCancellation = new CancellationTokenSource();
+        itemCancellation.Cancel();
+
+        await using var processor = new[] { 1 }
+            .SelectAsync(_ => Task.FromCanceled<int>(itemCancellation.Token))
+            .ProcessInParallel(2);
+
+        Exception? caught = null;
+        try
+        {
+            await foreach (var _ in processor.GetResultsAsyncEnumerable())
+            {
+            }
+        }
+        catch (Exception exception)
+        {
+            caught = exception;
+        }
+
+        await Assert.That(caught is OperationCanceledException).IsTrue();
+    }
+
     [Test]
     public async Task Item_Task_Continuations_Are_Not_Forced_Inline()
     {

EnumerableAsyncProcessor/AsyncEnumerableWorkerPool.cs

@@ -79,15 +79,23 @@ internal static async IAsyncEnumerable<TOutput> ProcessResultsAsync<TInput, TOut
 
                 if (pendingResults.Count == workerCount)
                 {
-                    yield return await pendingResults.Dequeue().ConfigureAwait(false);
+                    // WaitAsync guards against a worker observing cancellation and exiting
+                    // between this item being written and it being claimed - the item's
+                    // completion source would otherwise never complete and this await
+                    // would hang forever.
+                    var value = await pendingResults.Peek().WaitAsync(pipelineToken).ConfigureAwait(false);
+                    pendingResults.Dequeue();
+                    yield return value;
                 }
             }
 
             channel.Writer.TryComplete();
 
-            while (pendingResults.TryDequeue(out var resultTask))
+            while (pendingResults.Count > 0)
             {
-                yield return await resultTask.ConfigureAwait(false);
+                var value = await pendingResults.Peek().WaitAsync(pipelineToken).ConfigureAwait(false);
+                pendingResults.Dequeue();
+                yield return value;
             }
 
             await Task.WhenAll(workers).ConfigureAwait(false);
@@ -105,6 +113,17 @@ internal static async IAsyncEnumerable<TOutput> ProcessResultsAsync<TInput, TOut
             {
                 // Expected when enumeration is canceled or the consumer stops early.
             }
+
+            // Results abandoned by cancellation or an earlier failure may still fault;
+            // observe them so they cannot surface as UnobservedTaskException.
+            while (pendingResults.TryDequeue(out var abandonedResult))
+            {
+                _ = abandonedResult.ContinueWith(
+                    static t => _ = t.Exception,
+                    CancellationToken.None,
+                    TaskContinuationOptions.ExecuteSynchronously,
+                    TaskScheduler.Default);
+            }
         }
     }
 
@@ -169,19 +188,33 @@ private static Task[] StartResultWorkers<TInput, TOutput>(
         {
             workers[i] = Task.Run(async () =>
             {
-                await foreach (var workItem in reader.ReadAllAsync(cancellationToken).ConfigureAwait(false))
+                try
                 {
-                    Task<TOutput>? task = null;
-
-                    try
+                    await foreach (var workItem in reader.ReadAllAsync(cancellationToken).ConfigureAwait(false))
                     {
-                        task = taskSelector(workItem.Input);
-                        workItem.CompletionSource.TrySetResult(await task.ConfigureAwait(false));
+                        Task<TOutput>? task = null;
+
+                        try
+                        {
+                            task = taskSelector(workItem.Input);
+                            workItem.CompletionSource.TrySetResult(await task.ConfigureAwait(false));
+                        }
+                        catch (Exception exception)
+                        {
+                            workItem.CompletionSource.TrySetFromFault(task, exception, cancellationToken);
+                        }
                     }
-                    catch (Exception exception)
+                }
+                catch (OperationCanceledException)
+                {
+                    // Cancellation can strand items that were written but never claimed;
+                    // complete them so nothing awaiting their results hangs.
+                    while (reader.TryRead(out var abandonedItem))
                     {
-                        workItem.CompletionSource.TrySetFromFault(task, exception, cancellationToken);
+                        abandonedItem.CompletionSource.TrySetCanceled(cancellationToken);
                     }
+
+                    throw;
                 }
             }, cancellationToken);
         }

EnumerableAsyncProcessor/Extensions/EnumerableExtensions.cs

@@ -155,7 +155,10 @@ internal static async IAsyncEnumerable<T> ToIAsyncEnumerable<T>(this IEnumerable
 #if NET9_0_OR_GREATER
         await foreach (var task in Task.WhenEach(tasks).ConfigureAwait(false).WithCancellation(cancellationToken))
         {
-            yield return task.Result;
+            // Await rather than .Result: the task is already complete, but .Result wraps
+            // failures in AggregateException while the net8 fallback path rethrows the
+            // original exception. Both paths must surface identical exceptions.
+            yield return await task.ConfigureAwait(false);
         }
 #else
         // Interleaving via completion-order buckets: each task's continuation claims the next