Some tasks hang and consume CPU when reading from a batched channel

[arkolka]

Description

When reading from a batched channel using Batch and ReadAllAsync, some tasks occasionally hang indefinitely and consume CPU. This happens even with a simple repro using multiple tasks reading from an IAsyncEnumerable

Repro

public static class Program
{
    public static async Task Main()
    {
        var counts = new int[10];
        var tasks = Enumerable.Range(0, 10)
            .Select(x => Task.Run(async () =>
            {
                while (true)
                {
                    await GetSource()
                        .ToChannel(singleReader: true, cancellationToken: CancellationToken.None)
                        .Batch(Random.Shared.Next(15, 30), singleReader: true)
                        .ReadAllAsync(async batch =>
                        {
                            await Task.Delay(Random.Shared.Next(2, 15));
                        }, CancellationToken.None);

                    counts[x] += 1;
                    if (counts[x] % 10 == 0)
                    {
                        Console.WriteLine(counts.ToJson());
                    }
                }
            })).ToArray();

        await Task.WhenAll(tasks);
    }

    private static async IAsyncEnumerable<int> GetSource()
    {
        foreach (var value in Enumerable.Range(0, Random.Shared.Next(80, 120)))
        {
            yield return value;

            if (value % Random.Shared.Next(15, 25) == 0)
                await Task.Delay(Random.Shared.Next(2, 15));
        }
    }
}

Result

The program works correctly for a while. Then some tasks stop incrementing their counts (for example, task with index 2). They appear to hang in WaitToReadAsyncCore indefinitely, consuming CPU. Sample output of counts every 10 iterations

[1094,1096,642,1101,1098,1088,1093,1101,1076,1090]
[1096,1097,642,1102,1099,1090,1095,1103,1078,1092]
[1096,1097,642,1102,1100,1090,1095,1103,1078,1092]
[1098,1099,642,1104,1101,1092,1097,1105,1080,1094]
[1098,1100,642,1104,1101,1092,1097,1105,1080,1094]
[1100,1102,642,1107,1102,1093,1098,1107,1081,1096]
[1101,1103,642,1108,1103,1094,1100,1109,1083,1097]
[1101,1103,642,1108,1103,1095,1100,1110,1083,1097]

[arkolka]

fyi seems like an issue in Channels: dotnet/runtime#123544