fix: unify streaming exception fidelity and prevent cancellation hang

[thomhurst]

Two release blockers found in a pre-v4.0 audit.

1. Streaming exception fidelity differs per TFM

GetResultsAsyncEnumerable() routes through ToIAsyncEnumerable, whose net9+/net10 path used task.Result inside the Task.WhenEach loop. A faulted item surfaced as AggregateException (and a canceled item as AggregateException(TaskCanceledException)), while the net8 completion-order-bucket fallback rethrew the original exception unwrapped. Same code, different exception type per TFM — and the net9+ streaming path also disagreed with GetResultsAsync() in the same TFM.

Fix: await the (already completed) task instead of touching .Result.

2. Cancellation could hang the bounded async-enumerable result pipeline forever

In AsyncEnumerableWorkerPool.ProcessResultsAsync, the producer awaited per-item TaskCompletionSource tasks with no cancellation linkage. A worker that observed cancellation between an item being written to the channel and that item being claimed (channel WaitToReadAsync checks the token before checking for data) exited without completing the item's TCS — leaving the consumer's await foreach hanging forever, precisely when the user cancelled to regain control.

Fix (belt and braces):

• Producer awaits pending results via WaitAsync(pipelineToken) — airtight no-hang guarantee.
• Workers drain unclaimed channel items on cancellation exit and complete their TCSs as canceled.
• Results abandoned by cancellation or an earlier failure get their exceptions observed in the finally, so they can never surface as UnobservedTaskException.

Tests

• ExceptionFidelityTests: two new streaming tests asserting the original exception type (not AggregateException) surfaces from GetResultsAsyncEnumerable() for faulted and canceled items. Verified they fail on net10 without the fix (exactly 2 failures) and pass with it.
• CancellationRegressionTests (new): 100-iteration stress test cancelling bounded result streaming at varying points against a cancellation-ignoring infinite source; each iteration must observe OperationCanceledException within a bounded wait instead of hanging.

Full suite: 1665 passing across net8.0/net9.0/net10.0.

Note (not addressed here): TUnit --treenode-filter discovery currently fails on this repo with a MissingMethodException referencing the removed v3 ProcessInParallel(Int32) overload — pre-existing on main, unrelated to this change; filter-free runs work.

[greptile-apps]

Greptile Summary

This PR aligns streaming exception behavior and prevents bounded result streams from hanging during cancellation. The main changes are:

• Await completed tasks to preserve original exception types across target frameworks.
• Make pending result waits cancellation-aware.
• Cancel unclaimed channel items and observe abandoned task failures.
• Add tests for exception fidelity and cancellation hangs.

Confidence Score: 5/5

This looks safe to merge.

• No blocking issues found in the changed code.
• Cancellation now terminates pending waits and completes unclaimed results.
• Awaiting completed tasks matches the existing fallback exception behavior.

Important Files Changed

Filename	Overview
EnumerableAsyncProcessor/AsyncEnumerableWorkerPool.cs	Adds cancellation-aware result waits, cleanup for unclaimed work, and observation of abandoned failures.
EnumerableAsyncProcessor/Extensions/EnumerableExtensions.cs	Awaits completed tasks so streamed failures retain their original exception types.
EnumerableAsyncProcessor.UnitTests/CancellationRegressionTests.cs	Adds repeated coverage for cancellation of bounded result streams.
EnumerableAsyncProcessor.UnitTests/ExceptionFidelityTests.cs	Adds streamed fault and cancellation exception-fidelity coverage.

Sequence Diagram

%%{init: {'theme': 'neutral'}}%%
sequenceDiagram
    participant Producer
    participant Channel
    participant Worker
    participant Consumer

    Producer->>Channel: Write item and completion source
    Producer->>Producer: Queue pending result
    Channel->>Worker: Claim item
    Worker-->>Producer: Complete result source
    Producer-->>Consumer: Await and yield result

    alt Pipeline cancellation
        Consumer->>Producer: Cancel pipeline token
        Producer->>Producer: WaitAsync exits
        Worker->>Channel: Drain unclaimed items
        Worker->>Worker: Cancel completion sources
        Producer->>Producer: Observe abandoned failures
    end

Loading

%%{init: {'theme': 'base', 'themeVariables': {"darkMode": true, "background": "#0d1117", "primaryColor": "#21262d", "primaryTextColor": "#e6edf3", "primaryBorderColor": "#8b949e", "lineColor": "#8b949e", "textColor": "#e6edf3", "edgeLabelBackground": "#161b22", "actorBkg": "#21262d", "actorBorder": "#8b949e", "actorTextColor": "#e6edf3", "actorLineColor": "#8b949e", "signalColor": "#8b949e", "signalTextColor": "#e6edf3", "noteBkgColor": "#373320", "noteBorderColor": "#d4a72c", "noteTextColor": "#f0e6c0", "labelBoxBkgColor": "#21262d", "labelBoxBorderColor": "#8b949e", "labelTextColor": "#e6edf3", "loopTextColor": "#e6edf3", "activationBkgColor": "#30363d", "activationBorderColor": "#8b949e"}}}%%
sequenceDiagram
    participant Producer
    participant Channel
    participant Worker
    participant Consumer

    Producer->>Channel: Write item and completion source
    Producer->>Producer: Queue pending result
    Channel->>Worker: Claim item
    Worker-->>Producer: Complete result source
    Producer-->>Consumer: Await and yield result

    alt Pipeline cancellation
        Consumer->>Producer: Cancel pipeline token
        Producer->>Producer: WaitAsync exits
        Worker->>Channel: Drain unclaimed items
        Worker->>Worker: Cancel completion sources
        Producer->>Producer: Observe abandoned failures
    end

Loading

_{Reviews (1): Last reviewed commit: "fix: unify streaming exception fidelity ..." | Re-trigger Greptile}