Fix CPU spin in BatchingChannelReader when source completes (#53)

* Initial plan

* Add repro test for batching hang issue

Co-authored-by: electricessence <5899455+electricessence@users.noreply.github.com>

* Fix CPU hang in BatchingChannelReader.WaitToReadAsyncCore

The issue was that after Task.WhenAny completed, the code did not await
the source task before looping. This caused a tight CPU-consuming loop
when the source channel was closed but the buffer was empty, as the
closed source would immediately return a completed ValueTask on each
WaitToReadAsync call.

The fix properly awaits the source task after WhenAny to check its
completion status, preventing the tight loop. This matches the pattern
used in the base BufferingChannelReader class.

Co-authored-by: electricessence <5899455+electricessence@users.noreply.github.com>

* Address code review feedback

- Remove unused System.Text.Json import from test
- Use consistent attribute style in test
- Add clarifying comment explaining the buffer completion check

Co-authored-by: electricessence <5899455+electricessence@users.noreply.github.com>

* Bump version to 9.1.2

Updated Open.ChannelExtensions.csproj to increase the project version from 9.1.1 to 9.1.2. No other modifications were made.

---------

Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
Co-authored-by: electricessence <5899455+electricessence@users.noreply.github.com>

Author: Copilot

Open.ChannelExtensions.Tests/HangReproTest.cs

@@ -0,0 +1,58 @@
+using System.Runtime.CompilerServices;
+
+namespace Open.ChannelExtensions.Tests;
+
+public static class HangReproTest
+{
+	/// <summary>
+	/// Reproduction test for the hanging issue described in GitHub.
+	/// This test should fail if the bug exists (by timing out).
+	/// </summary>
+	[Fact]
+	public static async Task MultipleBatchReadersDoNotHang()
+	{
+		var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)); // 30 second timeout
+		var token = cts.Token;
+
+		var counts = new int[5]; // Using 5 tasks instead of 10 for faster testing
+		var tasks = Enumerable.Range(0, 5)
+			.Select(x => Task.Run(async () =>
+			{
+				for (int i = 0; i < 50; i++) // Run 50 iterations instead of infinite
+				{
+					if (token.IsCancellationRequested)
+						break;
+
+					await GetSource()
+						.ToChannel(singleReader: true, cancellationToken: token)
+						.Batch(Random.Shared.Next(15, 30), singleReader: true)
+						.ReadAllAsync(async batch =>
+						{
+							await Task.Delay(Random.Shared.Next(2, 15), token);
+						}, token);
+
+					counts[x] += 1;
+				}
+			}, token)).ToArray();
+
+		// Wait for all tasks to complete or timeout
+		await Task.WhenAll(tasks);
+
+		// Verify all tasks completed at least some iterations
+		Assert.All(counts, count => Assert.True(count >= 40, $"Count was {count}, expected at least 40"));
+	}
+
+	private static async IAsyncEnumerable<int> GetSource([EnumeratorCancellation] CancellationToken cancellationToken = default)
+	{
+		foreach (var value in Enumerable.Range(0, Random.Shared.Next(80, 120)))
+		{
+			if (cancellationToken.IsCancellationRequested)
+				yield break;
+
+			yield return value;
+
+			if (value % Random.Shared.Next(15, 25) == 0)
+				await Task.Delay(Random.Shared.Next(2, 15), cancellationToken);
+		}
+	}
+}

Open.ChannelExtensions/Open.ChannelExtensions.csproj

@@ -22,7 +22,7 @@
 		<RepositoryType>git</RepositoryType>
 		<GeneratePackageOnBuild>true</GeneratePackageOnBuild>
 		<GenerateDocumentationFile>true</GenerateDocumentationFile>
-		<Version>9.1.1</Version>
+		<Version>9.1.2</Version>
 		<PackageReleaseNotes>Ensure the current task scheduler is used and that it can be configured.</PackageReleaseNotes>
 		<PackageLicenseExpression>MIT</PackageLicenseExpression>
 		<PublishRepositoryUrl>true</PublishRepositoryUrl>

Open.ChannelExtensions/Readers/BatchingChannelReader.cs

@@ -245,13 +245,22 @@ protected override async ValueTask<bool> WaitToReadAsyncCore(
 		}
 
 		// WhenAny will not throw when a task is cancelled.
-		await Task.WhenAny(s.AsTask(), b).ConfigureAwait(false);
+		Task<bool> sTask = s.AsTask();
+		await Task.WhenAny(sTask, b).ConfigureAwait(false);
 		if (b.IsCompleted) // Assuming it was bufferWait that completed.
 		{
 			await tokenSource.CancelAsync().ConfigureAwait(false);
 			return await b.ConfigureAwait(false);
 		}
 
+		// Await the source task to check if it completed and get its result
+		// This prevents a tight loop when the source is closed
+		await sTask.ConfigureAwait(false);
+		// Check if buffer completed while we were waiting (e.g., timeout forced a batch)
+		// If so, return immediately without trying to pipe more items
+		if (b.IsCompleted)
+			return await b.ConfigureAwait(false);
+
 		TryPipeItems(false);
 
 		if (b.IsCompleted || token.IsCancellationRequested)