AsyncEnumerableEx.Merge doesn't support aborting enumeration

[fschmied]

I think I've found a bug or at least missing feature in System.Interactive.Async, version 6.0.3, tested on .NET 9.0.9 (on Windows).

When using AsyncEnumerableEx.Merge, it seems that the async enumeration of the resulting IAsyncEnumerable sequence cannot be aborted as long as one of the underlying sequence(s) are still active.

The following is a complete repro sample. It is based on an infinite sequence that produces a single item, then waits indefinitely until canceled. The sequence is enumerated by an await foreach loop, producing a derived sequence that optionally throws an exception or yields the item. The outermost code again loops through the resulting sequence, breaking the loop after the first item or exception.

When the infinite sequence is wrapped by Merge (shouldMerge = true), the program will hang, i.e., breaking the loop doesn't work and the program will continue waiting for the infinite sequence. The exception, if thrown, isn't observed at all.

When Merge is not used (shouldMerge = false), the program will run correctly and terminate, either after the first item or the exception.

using System.Runtime.CompilerServices;

internal static class Program
{
    public static async Task Main(string[] args)
    {
        const bool shouldMerge = true;
        const bool shouldThrow = true;

        var inner = Enumerate(shouldMerge, shouldThrow, CancellationToken.None);

        try
        {
            await foreach (var message in inner)
            {
                Console.WriteLine(message);
                break;
            }
        }
        catch (Exception ex)
        {
            Console.WriteLine(ex);
        }

        Console.WriteLine("After");
    }

    private static async IAsyncEnumerable<string> Enumerate(bool shouldMerge, bool shouldThrow, [EnumeratorCancellation] CancellationToken cancellationToken)
    {
        var infinite = shouldMerge
            ? AsyncEnumerableEx.Merge(Infinite(cancellationToken))
            : Infinite(cancellationToken);

        await foreach (var item in infinite.WithCancellation(cancellationToken))
        {
            if (shouldThrow)
                throw new Exception("Test");

            yield return item;
        }
    }

    private static async IAsyncEnumerable<string> Infinite([EnumeratorCancellation] CancellationToken cancellationToken)
    {
        yield return "first item";

        while (true)
        {
            await Task.Delay(1000, cancellationToken);
        }
    }
}

A zipped solution can be downloaded here:
2025-11-01 AsyncEnumerableCatch.zip

As I'm not an expert with working with IAsyncEnumerable or Ix.NET, it might well be that I'm missing something. If so, please tell me so. :)

[fschmied]

I originally thought this was related to cancellation, but it's actually related to the disposing of the IAsyncEnumerator created by Merge. I copied over Merge's source code and found that the cause is in the finally block here:

try
{
    if (moveNextTask != null && moveNextTask != TaskExt.Never)
    {
        _ = await moveNextTask.ConfigureAwait(false);
    }
}
finally
{
    if (enumerator != null)
    {
        await enumerator.DisposeAsync().ConfigureAwait(false);
    }
}

So, the finally block waits for the moveNextTasks to complete, which - in my case - they might never do.

[fschmied]

For now, I've switched to Observable.Merge instead:

    private static IAsyncEnumerable<T> MyMerge<T>(params IAsyncEnumerable<T>[] sources)
    {
        // Use Observable.Merge because AsyncEnumerableEx.Merge has issues with disposal, i.e., when "break"-ing the enumerating await foreach loop
        // or when an exception is thrown.
        // See https://github.com/dotnet/reactive/issues/2264.
        return sources
            .Select(source => source.ToObservable())
            .Merge()
            .ToAsyncEnumerable();
    }

I'd be very interested in alternative workarounds.

[NetherGranite]

Also running into this issue. From what I understand, disposing IAsyncEnumerator is supposed to not wait for MoveNextAsync to complete since it creates the very confusion you described, so I would argue this is a bug. Concurrent merges in other libraries do not behave like this.

With this issue and the nightmare that is the AsyncEnumeratorEx.Merge overloads, this could be an opportunity to deprecate all of them and create the previously discussed ConcurrentMerge implementations.