ForEachAsync continues after cancellation token is cancelled

[da1rren]

Bug

Which library version?

System.Reactive.Async 6.0.0-alpha.3
System.Reactive.Linq 5.0.0"

What are the platform(s), environment(s) and related component version(s)?

N/A

What is the use case or problem?

Cancelling ForEachAsync should stop the enumeration.

What is the expected outcome?

Enumeration stopped after token cancelled

What is the actual outcome?

Enumeration continues forever

Do you have a code snippet or project that reproduces the problem?

using System.Reactive.Linq;

var observable = AsyncObservable.Interval(TimeSpan.FromSeconds(1));

var cancellationTokenSource = new CancellationTokenSource();
cancellationTokenSource.CancelAfter(TimeSpan.FromSeconds(5));

await observable.ForEachAsync(seconds =>
{
    Console.WriteLine($"Seconds: {seconds}, Is Cancelled: {cancellationTokenSource.Token.IsCancellationRequested}");
}, cancellationTokenSource.Token);

I might be just completely misunderstanding but the behaviour seems odd.

[idg10]

Having reproduced this and traced through the behaviour, I think we've found the problem. It's here:

reactive/AsyncRx.NET/System.Reactive.Async/Linq/Operators/ForEachAsync.cs

Lines 127 to 129 in 95d9ea9

	}
	await tcs.Task.ConfigureAwait(false);

The call to await tcs.Task.ConfigureAwait(false); occurs after a closing } that corresponds to this using statement further up the file:

reactive/AsyncRx.NET/System.Reactive.Async/Linq/Operators/ForEachAsync.cs

Lines 61 to 73 in 95d9ea9

	using (token.Register(() =>
	{
	tcs.TrySetCanceled(token);
	subscription.DisposeAsync().AsTask().ContinueWith(t =>
	{
	if (t.Exception != null)
	{
	// TODO: Trace?
	}
	});
	}))
	{

Once we reach that closing }, we Dispose the CancellationTokenRegistration returned by CancellationToken.Register, which means we're telling the CancellationToken that we are no longer interested in being notified when cancellation occurs.

But that's too early. We should do this only after the ForEachAsync has finished. So the call to await tcs.Task.ConfigureAwait(false); should move inside that code block to ensure that we don't unregister for cancellation until after the for each logic has reached an end.

[idg10]

So, although this isn't going to be terribly complex to fix, the larger problem here is that because Async.Rx was always an experimental feature, it doesn't have any tests.

#1900

Normally when fixing a bug I like to ensure that tests exist that would detect the problem so we can avoid regressions, and possibly detect other problems of a similar nature.

However, I'm not sure how soon we'll be able to address the lack of tests. As you can see from that other issue, we are hoping to work out some way of sharing test logic across normal and async Rx, so that we don't have to maintain two parallel, near-identical sets of tests. But that's a fairly large undertaking, and not something we expect to complete imminently. So we could just fix this without a test as a stopgap.

Is this problem causing you problems right now, or was it just a curiosity that you noticed? If this is blocking you, I'll fix it without waiting until we have a proper answer for how to test AsyncRx.NET.

[da1rren]

It was for something experimental, so please don't feel you have to fix it for myself. Feel free to close if you'd rather track the issue in another way.