Skip to content

Commit 36d38b0

Browse files
thomhurstthomhurst
committed
test: regression coverage for correctness and perf fixes
30 new tests across five suites, each pinned to a previously shipped bug: - InputEnumerationRegressionTests: input enumerated exactly once across process/await/CancelAll/dispose; one-shot enumerables work; iterator exceptions throw at build time instead of hanging awaiters. - DisposalRegressionTests: DisposeAsync actually cancels pending tasks and completes promptly; synchronous Dispose no longer blocks on in-flight work; CancelAll on result processors returns immediately (previously blocked up to 30s via the Dispose cancellation callback); disposal is idempotent in any order. - ExceptionFidelityTests: multi-fault tasks preserve every inner exception (GetBaseException previously kept one); one failing item cannot stop the rest; completion sources run continuations asynchronously. - ValidationRegressionTests: zero/negative parallelism, maxConcurrency, batch size and negative timespans throw at build time on void AND result variants (result variants previously skipped validation; maxConcurrency 0 previously failed mid-processing); the removed 10k caps stay removed; already-cancelled tokens fail cleanly, never NRE. - WorkerPoolBehaviourTests: concurrency limit holds on the worker-pool path, every item processed exactly once, oversized limits fine, cancellation drains promptly, result order preserved regardless of completion order. - StreamingResultsTests: results stream in completion order, exactly once, and honour cancellation. Test project now targets net9.0 and net8.0 so the pre-net9 completion-order bucket implementation of ToIAsyncEnumerable is exercised in CI rather than only compiled.
1 parent d494f49 commit 36d38b0

7 files changed

Lines changed: 756 additions & 1 deletion

‎EnumerableAsyncProcessor.UnitTests/DisposalRegressionTests.cs‎

Lines changed: 130 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,130 @@
1+
using System;
2+
using System.Diagnostics;
3+
using System.Linq;
4+
using System.Threading.Tasks;
5+
using EnumerableAsyncProcessor.Extensions;
6+
7+
namespace EnumerableAsyncProcessor.UnitTests;
8+
9+
/// <summary>
10+
/// Guards the disposal and cancellation semantics:
11+
/// - disposal must actually cancel pending work (a guard-ordering bug previously made it a no-op),
12+
/// - synchronous Dispose must not block (it previously ran sync-over-async with a 30s wait),
13+
/// - CancelAll on a result processor must not block the calling thread (it previously invoked
14+
/// blocking Dispose via the cancellation callback).
15+
/// </summary>
16+
public class DisposalRegressionTests
17+
{
18+
[Test]
19+
public async Task DisposeAsync_Cancels_Unstarted_Tasks_And_Completes_Promptly()
20+
{
21+
var blocker = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
22+
var firstItemStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
23+
24+
var processor = Enumerable.Range(0, 10).ToList()
25+
.ForEachAsync(async _ =>
26+
{
27+
firstItemStarted.TrySetResult();
28+
await blocker.Task;
29+
})
30+
.ProcessInParallel(1);
31+
32+
await firstItemStarted.Task;
33+
34+
var stopwatch = Stopwatch.StartNew();
35+
var disposeTask = processor.DisposeAsync();
36+
37+
// Cancellation happens synchronously inside DisposeAsync, before it waits for in-flight work.
38+
// Previously nothing was cancelled and disposal just waited for tasks to finish naturally.
39+
await Assert.That(processor.GetEnumerableTasks().Count(x => x.IsCanceled)).IsEqualTo(10);
40+
41+
blocker.TrySetResult();
42+
await disposeTask;
43+
stopwatch.Stop();
44+
45+
await Assert.That(stopwatch.Elapsed).IsLessThan(TimeSpan.FromSeconds(5));
46+
}
47+
48+
[Test]
49+
public async Task Synchronous_Dispose_Returns_Without_Blocking_On_InFlight_Work()
50+
{
51+
var blocker = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
52+
var firstItemStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
53+
54+
var processor = Enumerable.Range(0, 10).ToList()
55+
.ForEachAsync(async _ =>
56+
{
57+
firstItemStarted.TrySetResult();
58+
await blocker.Task;
59+
})
60+
.ProcessInParallel(1);
61+
62+
await firstItemStarted.Task;
63+
64+
try
65+
{
66+
var stopwatch = Stopwatch.StartNew();
67+
processor.Dispose();
68+
stopwatch.Stop();
69+
70+
await Assert.That(stopwatch.Elapsed).IsLessThan(TimeSpan.FromSeconds(2));
71+
await Assert.That(processor.GetEnumerableTasks().Count(x => x.IsCanceled)).IsEqualTo(10);
72+
}
73+
finally
74+
{
75+
blocker.TrySetResult();
76+
}
77+
}
78+
79+
[Test]
80+
public async Task CancelAll_On_Result_Processor_Does_Not_Block_And_Cancels_Pending_Tasks()
81+
{
82+
var blocker = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
83+
var firstItemStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
84+
85+
var processor = Enumerable.Range(0, 10).ToList()
86+
.SelectAsync(async i =>
87+
{
88+
firstItemStarted.TrySetResult();
89+
await blocker.Task;
90+
return i;
91+
})
92+
.ProcessInParallel(1);
93+
94+
await firstItemStarted.Task;
95+
96+
try
97+
{
98+
var stopwatch = Stopwatch.StartNew();
99+
processor.CancelAll();
100+
stopwatch.Stop();
101+
102+
await Assert.That(stopwatch.Elapsed).IsLessThan(TimeSpan.FromSeconds(2));
103+
await Assert.That(processor.GetEnumerableTasks().Count(x => x.IsCanceled)).IsEqualTo(10);
104+
await Assert.ThrowsAsync<TaskCanceledException>(() => processor.GetResultsAsync());
105+
}
106+
finally
107+
{
108+
blocker.TrySetResult();
109+
}
110+
111+
await processor.DisposeAsync();
112+
}
113+
114+
[Test]
115+
public async Task Disposal_Is_Idempotent_And_Safe_In_Any_Order()
116+
{
117+
var processor = Enumerable.Range(0, 5).ToList()
118+
.ForEachAsync(_ => Task.CompletedTask)
119+
.ProcessInParallel();
120+
121+
await processor.WaitAsync();
122+
123+
await processor.DisposeAsync();
124+
await processor.DisposeAsync();
125+
processor.Dispose();
126+
processor.CancelAll();
127+
128+
await Assert.That(processor.GetEnumerableTasks().Count(x => x.IsCompletedSuccessfully)).IsEqualTo(5);
129+
}
130+
}

‎EnumerableAsyncProcessor.UnitTests/EnumerableAsyncProcessor.UnitTests.csproj‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,8 @@
11
<Project Sdk="Microsoft.NET.Sdk">
22

33
<PropertyGroup>
4-
<TargetFramework>net9.0</TargetFramework>
4+
<!-- net8.0 also exercises the pre-net9 completion-order streaming path in ToIAsyncEnumerable -->
5+
<TargetFrameworks>net9.0;net8.0</TargetFrameworks>
56
<Nullable>enable</Nullable>
67
<IsPackable>false</IsPackable>
78
<!-- Do not warn about non-strong named dependencies, as we target .NET (not .NET Framework).

‎EnumerableAsyncProcessor.UnitTests/ExceptionFidelityTests.cs‎

Lines changed: 115 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,115 @@
1+
using System;
2+
using System.Linq;
3+
using System.Threading;
4+
using System.Threading.Tasks;
5+
using EnumerableAsyncProcessor.Extensions;
6+
7+
namespace EnumerableAsyncProcessor.UnitTests;
8+
9+
/// <summary>
10+
/// Guards exception propagation fidelity:
11+
/// - a multi-fault task must surface every inner exception (GetBaseException previously kept only one),
12+
/// - one failing item must not stop the remaining items from processing,
13+
/// - completion sources must run continuations asynchronously so user continuations
14+
/// cannot run inline on the completing thread.
15+
/// </summary>
16+
public class ExceptionFidelityTests
17+
{
18+
[Test]
19+
public async Task All_Exceptions_From_A_MultiFault_Task_Are_Preserved()
20+
{
21+
await using var processor = new[] { 1 }
22+
.ForEachAsync(_ => Task.WhenAll(
23+
Task.FromException(new InvalidOperationException("first failure")),
24+
Task.FromException(new InvalidOperationException("second failure"))))
25+
.ProcessOneAtATime();
26+
27+
Exception? caught = null;
28+
try
29+
{
30+
await processor.WaitAsync();
31+
}
32+
catch (Exception exception)
33+
{
34+
caught = exception;
35+
}
36+
37+
await Assert.That(caught).IsNotNull();
38+
39+
var itemTask = processor.GetEnumerableTasks().Single();
40+
var messages = itemTask.Exception!.InnerExceptions.Select(e => e.Message).ToList();
41+
42+
await Assert.That(messages.Count).IsEqualTo(2);
43+
await Assert.That(messages.Contains("first failure")).IsTrue();
44+
await Assert.That(messages.Contains("second failure")).IsTrue();
45+
}
46+
47+
[Test]
48+
public async Task Synchronous_Selector_Throw_Faults_Only_That_Item()
49+
{
50+
await using var processor = Enumerable.Range(0, 10).ToList()
51+
.ForEachAsync(i => i == 5
52+
? throw new InvalidOperationException("item 5 failed")
53+
: Task.CompletedTask)
54+
.ProcessInParallel(2);
55+
56+
Exception? caught = null;
57+
try
58+
{
59+
await processor.WaitAsync();
60+
}
61+
catch (Exception exception)
62+
{
63+
caught = exception;
64+
}
65+
66+
await Assert.That(caught).IsNotNull();
67+
await Assert.That(processor.GetEnumerableTasks().Count(x => x.IsCompletedSuccessfully)).IsEqualTo(9);
68+
await Assert.That(processor.GetEnumerableTasks().Count(x => x.IsFaulted)).IsEqualTo(1);
69+
}
70+
71+
[Test]
72+
public async Task Failed_Items_Do_Not_Prevent_Remaining_Items_From_Processing()
73+
{
74+
var processedCount = 0;
75+
76+
await using var processor = Enumerable.Range(0, 30).ToList()
77+
.ForEachAsync(async i =>
78+
{
79+
Interlocked.Increment(ref processedCount);
80+
await Task.Yield();
81+
82+
if (i % 3 == 0)
83+
{
84+
throw new InvalidOperationException($"item {i} failed");
85+
}
86+
})
87+
.ProcessInParallel(maxConcurrency: 3);
88+
89+
try
90+
{
91+
await processor.WaitAsync();
92+
}
93+
catch (Exception)
94+
{
95+
// Expected - a third of the items fail
96+
}
97+
98+
await Assert.That(processedCount).IsEqualTo(30);
99+
await Assert.That(processor.GetEnumerableTasks().Count(x => x.IsCompletedSuccessfully)).IsEqualTo(20);
100+
await Assert.That(processor.GetEnumerableTasks().Count(x => x.IsFaulted)).IsEqualTo(10);
101+
}
102+
103+
[Test]
104+
public async Task Item_Task_Continuations_Are_Not_Forced_Inline()
105+
{
106+
await using var processor = new[] { 1 }
107+
.ForEachAsync(_ => Task.CompletedTask)
108+
.ProcessOneAtATime();
109+
110+
await processor.WaitAsync();
111+
112+
var itemTask = processor.GetEnumerableTasks().Single();
113+
await Assert.That(itemTask.CreationOptions.HasFlag(TaskCreationOptions.RunContinuationsAsynchronously)).IsTrue();
114+
}
115+
}

‎EnumerableAsyncProcessor.UnitTests/InputEnumerationRegressionTests.cs‎

Lines changed: 136 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,136 @@
1+
using System;
2+
using System.Collections;
3+
using System.Collections.Generic;
4+
using System.Linq;
5+
using System.Threading.Tasks;
6+
using EnumerableAsyncProcessor.Extensions;
7+
8+
namespace EnumerableAsyncProcessor.UnitTests;
9+
10+
/// <summary>
11+
/// Guards against the input enumerable being lazily re-enumerated by the processor internals.
12+
/// Historically the input was enumerated up to five times (validation, processing, awaiting,
13+
/// cancellation and disposal), which broke one-shot sources and re-ran side effects.
14+
/// </summary>
15+
public class InputEnumerationRegressionTests
16+
{
17+
[Test]
18+
public async Task Input_Enumerable_Is_Enumerated_Exactly_Once()
19+
{
20+
var source = new TrackingEnumerable(20);
21+
22+
await using (var processor = source.ForEachAsync(_ => Task.CompletedTask).ProcessInParallel(4))
23+
{
24+
await processor.WaitAsync();
25+
26+
// These all re-enumerated the input in previous versions
27+
_ = processor.GetEnumerableTasks().ToList();
28+
_ = processor.GetEnumerableTasks().ToList();
29+
processor.CancelAll();
30+
}
31+
32+
await Assert.That(source.EnumerationCount).IsEqualTo(1);
33+
await Assert.That(source.ItemsYielded).IsEqualTo(20);
34+
}
35+
36+
[Test]
37+
public async Task OneShot_Enumerable_Processes_All_Items()
38+
{
39+
var source = new OneShotEnumerable<int>(Enumerable.Range(0, 10));
40+
41+
await using var processor = source.SelectAsync(i => Task.FromResult(i * 2)).ProcessInParallel(2);
42+
var results = await processor.GetResultsAsync();
43+
44+
await Assert.That(results.OrderBy(x => x).SequenceEqual(Enumerable.Range(0, 10).Select(i => i * 2))).IsTrue();
45+
}
46+
47+
[Test]
48+
public async Task Result_Processor_Enumerates_Input_Exactly_Once()
49+
{
50+
var source = new TrackingEnumerable(10);
51+
52+
await using (var processor = source.SelectAsync(i => Task.FromResult(i)).ProcessInParallel())
53+
{
54+
_ = await processor.GetResultsAsync();
55+
_ = processor.GetEnumerableTasks().ToList();
56+
processor.CancelAll();
57+
}
58+
59+
await Assert.That(source.EnumerationCount).IsEqualTo(1);
60+
}
61+
62+
[Test]
63+
public async Task Iterator_Exception_Surfaces_At_Build_Time_Instead_Of_Hanging_Awaiters()
64+
{
65+
InvalidOperationException? caught = null;
66+
67+
try
68+
{
69+
_ = FaultyItems().ForEachAsync(_ => Task.CompletedTask).ProcessInParallel();
70+
}
71+
catch (InvalidOperationException exception)
72+
{
73+
caught = exception;
74+
}
75+
76+
await Assert.That(caught).IsNotNull();
77+
await Assert.That(caught!.Message).IsEqualTo("enumeration failed");
78+
}
79+
80+
private static IEnumerable<int> FaultyItems()
81+
{
82+
yield return 1;
83+
yield return 2;
84+
throw new InvalidOperationException("enumeration failed");
85+
}
86+
87+
private sealed class TrackingEnumerable : IEnumerable<int>
88+
{
89+
private readonly int _count;
90+
91+
public int EnumerationCount { get; private set; }
92+
public int ItemsYielded { get; private set; }
93+
94+
public TrackingEnumerable(int count)
95+
{
96+
_count = count;
97+
}
98+
99+
public IEnumerator<int> GetEnumerator()
100+
{
101+
EnumerationCount++;
102+
103+
for (var i = 0; i < _count; i++)
104+
{
105+
ItemsYielded++;
106+
yield return i;
107+
}
108+
}
109+
110+
IEnumerator IEnumerable.GetEnumerator() => GetEnumerator();
111+
}
112+
113+
private sealed class OneShotEnumerable<T> : IEnumerable<T>
114+
{
115+
private readonly IEnumerable<T> _items;
116+
private bool _enumerated;
117+
118+
public OneShotEnumerable(IEnumerable<T> items)
119+
{
120+
_items = items;
121+
}
122+
123+
public IEnumerator<T> GetEnumerator()
124+
{
125+
if (_enumerated)
126+
{
127+
throw new InvalidOperationException("This enumerable can only be enumerated once.");
128+
}
129+
130+
_enumerated = true;
131+
return _items.GetEnumerator();
132+
}
133+
134+
IEnumerator IEnumerable.GetEnumerator() => GetEnumerator();
135+
}
136+
}

0 commit comments

Comments
 (0)