释放IAsyncEnumerator未触发底层IAsyncEnumerable取消令牌的解决方案咨询

经测试发现，释放IAsyncEnumerable生成的IAsyncEnumerator时，底层IAsyncEnumerable的CancellationToken并未被触发。具体表现为以下单元测试的断言失败：

[Test]
public async Task test()
{
    await foreach (var value in TestData())
        break;
}

private static async IAsyncEnumerable<int> TestData([EnumeratorCancellation] CancellationToken ct = default)
{
    try
    {
        yield return 1;
        await Task.CompletedTask;
        yield return 2;
    }
    finally
    {
        Assert.That(ct.IsCancellationRequested, Is.True);
    }
}

我对该行为感到意外，想知道是否有标准方法能在此场景下让TestData的CancellationToken被取消？我可以实现自定义操作符，但这过于繁琐，希望有更优方案。以下是我自行实现的可行方案，但不够理想：

[Test]
public async Task this_test_passes()
{
    await foreach (var value in CancelWhenUnsubscribed(TestData()))
        break;
}

public static async IAsyncEnumerable<T> CancelWhenUnsubscribed<T>(IAsyncEnumerable<T> source, [EnumeratorCancellation] CancellationToken ct = default)
{
    using var cts = CancellationTokenSource.CreateLinkedTokenSource(ct);
    await using var en = source.GetAsyncEnumerator(cts.Token);
    using var _ = Disposable.Create(() => cts.Cancel());
    while (await en.MoveNextAsync())
        yield return en.Current;
}

背景说明 #

我使用System.Interactive.Async合并多个IAsyncEnumerable序列。其实现逻辑为：当合并序列被释放时，会先等待所有进行中的MoveNext任务完成，再释放并终止合并序列。在我的场景中，若合并序列终止时未取消底层IAsyncEnumerable，会导致准死锁，如下测试案例会陷入停滞：

[Test]
public async Task this_test_gets_stuck()
{
    await foreach (var value in AsyncEnumerableEx.Merge([TestData(), TestData()]))
        break;
}

private static async IAsyncEnumerable<int> TestData([EnumeratorCancellation] CancellationToken ct = default)
{
    yield return 1;
    await Task.Delay(TimeSpan.FromHours(100), ct);
    yield return 2;
}

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