Open.ChannelExtensions管道异常处理：异常未正确传播问题

问题现象 #

使用Open.ChannelExtensions构建管道时，发现异常无法正确传播到调用方，尤其是加入.Batch()步骤后，部分场景下异常被吞：

• Scenario1：无.Batch()，第一个元素处理时立即抛异常 → 异常能被捕获
• Scenario2：有.Batch()，第一个元素处理时立即抛异常 → 异常被吞
• Scenario3：无.Batch()，处理到元素20时抛异常 → 异常能被捕获
• Scenario4：有.Batch()，处理到元素20时抛异常 → 异常仅偶尔被捕获（约25%概率）

原因分析 #

问题根源在于.Batch()方法的实现逻辑：当上游管道因异常完成时，.Batch()内部的处理流程仅会发送已攒够的批次（若有），然后完成下游管道，但未将上游的异常传递到下游管道的完成状态中。

具体来说，.Batch()的ProcessAsync方法仅捕获自身执行过程中的异常，不会监听上游管道的Completion异常。当上游因异常终止时，WaitToReadAsync()会返回false，方法会处理剩余元素（若有）后直接调用writer.TryComplete()，导致下游管道认为是正常完成，不会抛出上游的异常。

解决方案 #

针对这个问题，有两种可行的处理方式：

方案1：手动监听上游管道的异常 #

在等待最终处理任务的同时，监听.Batch()上游管道的Completion状态，确保异常能被捕获。例如修改Scenario2的代码：

public async Task Scenario2()
{
    var channel = Channel.CreateBounded<int>(10000);

    for (int i = 0; i < 100; i++)
    {
        await channel.Writer.WriteAsync(i);
    }

    // 保留上游管道的引用，用于监听异常
    var upstreamReader = channel.Reader
        .Pipe(1, (element) =>
        {
            throw new Exception();
            Console.WriteLine(element);
            return 1;
        })
        .Pipe(2, (evt) =>
        {
            Console.WriteLine("	" + evt);
            return evt * 2;
        });

    var task = upstreamReader
        .Batch(20)
        .PipeAsync(1, async (evt) =>
        {
            Console.WriteLine("		" + evt);
            return Task.FromResult(evt);

        })
        .ReadAll(task =>
        {
        });

    channel.Writer.TryComplete();
    // 同时等待处理任务和上游管道的Completion
    await Task.WhenAll(task, upstreamReader.Completion);
}

方案2：替换为自定义的异常感知Batch方法 #

如果需要长期解决这个问题，可以自己实现一个能正确传播上游异常的Batch方法：

public static ChannelReader<T[]> BatchWithErrorPropagation<T>(this ChannelReader<T> source, int batchSize, int? maxBufferSize = null)
{
    if (source == null) throw new ArgumentNullException(nameof(source));
    if (batchSize < 1) throw new ArgumentOutOfRangeException(nameof(batchSize), batchSize, "必须至少为1");

    var channel = maxBufferSize.HasValue
        ? Channel.CreateBounded<T[]>(maxBufferSize.Value)
        : Channel.CreateUnbounded<T[]>();

    _ = ProcessAsync(source, channel.Writer, batchSize);
    return channel.Reader;
}

private static async Task ProcessAsync<T>(ChannelReader<T> source, ChannelWriter<T[]> writer, int batchSize)
{
    using var memory = new Memory<T>(new T[batchSize]);
    int count = 0;
    try
    {
        while (await source.WaitToReadAsync().ConfigureAwait(false))
        {
            while (source.TryRead(out var item))
            {
                memory.Span[count++] = item;
                if (count == batchSize)
                {
                    await writer.WriteAsync(memory.ToArray()).ConfigureAwait(false);
                    count = 0;
                }
            }
        }

        // 检查上游是否有异常
        if (!source.Completion.IsCompletedSuccessfully)
        {
            var ex = await source.Completion.ConfigureAwait(false);
            writer.TryComplete(ex);
            return;
        }

        if (count > 0)
            await writer.WriteAsync(memory.Slice(0, count).ToArray()).ConfigureAwait(false);
    }
    catch (Exception ex)
    {
        writer.TryComplete(ex);
        throw;
    }
    finally
    {
        // 仅当未因异常完成时才正常完成
        _ = writer.TryComplete();
    }
}

使用这个自定义方法替代原有的.Batch()，即可让上游异常自动传播到下游。

验证说明 #

• 对于Scenario2，使用上述任意方案后，异常会被正确捕获；
• 对于Scenario4，自定义Batch方法会在上游异常时立即传递异常，确保100%能捕获到异常，而不是偶尔触发。

内容的提问来源于stack exchange，提问作者Marduk