Cancellation not propagated by various operators

[glenringer]

If an operand's MoveNextAsync call returns a canceled task, various operators do not propagate an OperationCanceledException but never return due to unhandled internal exceptions.

Tested with at least Prefetch and SwitchMap, but any operator relying on the following pattern (e.g., Amb, CombineLatest, ConcatMapEager, Debounce, FlatMap, Merge, Sample, TakeUntil, WithLatestFrom) might be affected as well:

if (t.IsFaulted) // t is typically of type Task<bool>.
{
    // ...
}    
else if (t.Result) // Throws if t.IsCanceled.
{
    // ...
}

Example unit test with Prefetch:

[Fact]
public async Task ThrowsCancellationAsync()
{
    await Assert.ThrowsAnyAsync<OperationCanceledException>(async () =>
        await AsyncEnumerable
            .FromTask(Task.FromCanceled<int>(new CancellationToken(true)))
            .Prefetch(1) // Comment line to fix.
            .FirstAsync());
}

Unhandled internal (not propagated) exception:

System.AggregateException
  HResult=0x80131500
  Message=One or more errors occurred.
  Source=mscorlib
  StackTrace:
   at System.Threading.Tasks.Task.ThrowIfExceptional(Boolean includeTaskCanceledExceptions)

Inner Exception 1:
TaskCanceledException: A task was canceled.

Stack trace:

mscorlib.dll!System.Threading.Tasks.Task.ThrowIfExceptional(bool includeTaskCanceledExceptions)	Unknown
mscorlib.dll!System.Threading.Tasks.Task<bool>.GetResultCore(bool waitCompletionNotification)	Unknown
async-enumerable-dotnet.dll!async_enumerable_dotnet.impl.Prefetch<int>.PrefetchEnumerator.SourceHandler(System.Threading.Tasks.Task<bool> t = Id = 7, Status = Canceled, Method = "{null}", Result = "{Not yet computed}")	Unknown
async-enumerable-dotnet.dll!async_enumerable_dotnet.impl.Prefetch<int>.PrefetchEnumerator..cctor.AnonymousMethod__22_0(System.Threading.Tasks.Task<bool> t = Id = 7, Status = Canceled, Method = "{null}", Result = "{Not yet computed}", object state = {async_enumerable_dotnet.impl.Prefetch<int>.PrefetchEnumerator})	Unknown
mscorlib.dll!System.Threading.Tasks.Task.Execute()	Unknown
mscorlib.dll!System.Threading.ExecutionContext.RunInternal(System.Threading.ExecutionContext executionContext, System.Threading.ContextCallback callback, object state, bool preserveSyncCtx)	Unknown
mscorlib.dll!System.Threading.ExecutionContext.Run(System.Threading.ExecutionContext executionContext, System.Threading.ContextCallback callback, object state, bool preserveSyncCtx)	Unknown
mscorlib.dll!System.Threading.Tasks.Task.ExecuteWithThreadLocal(ref System.Threading.Tasks.Task currentTaskSlot = Id = 8, Status = Running, Method = "Void <.cctor>b__22_0(System.Threading.Tasks.Task`1[System.Boolean], System.Object)")	Unknown
mscorlib.dll!System.Threading.Tasks.Task.ExecuteEntry(bool bPreventDoubleExecution)	Unknown
mscorlib.dll!System.Threading.ThreadPoolWorkQueue.Dispatch()	Unknown
[Native to Managed Transition]

[akarnokd]

Thanks for reporting. My knowledge about C# async/await and Task cancellation is limited. If you could suggest a fix for one of the operators, that would be great!

[glenringer]

For Prefetch:

                if (t.IsFaulted)
                {
                    _error = ExceptionHelper.Extract(t.Exception);
                    _done = true;
                }
+               else if (t.IsCanceled)
+               {
+                   _error = new OperationCanceledException();
+                   _done = true;
+               }
                else if (t.Result)

[akarnokd]

Would you like to post a PR?

[akarnokd]

I've updated the code to handle the IsCanceled cases.