Eat cancellation exceptions from WaitAsync in Parallel.ForEachAsync (#66712)

We can safely ignore cancellation exceptions due to WaitAsync being canceled.  They do not represent work that was started and interrupted, and we know exactly the source of the exception (as compared to if the exception emerged from either MoveNextAsync or the loop body).

src/libraries/System.Threading.Tasks.Parallel/src/System/Threading/Tasks/Parallel.ForEachAsync.cs

@@ -229,7 +229,17 @@ private static Task ForEachAsync<TSource>(IAsyncEnumerable<TSource> source, int
                     {
                         // Get the next element from the enumerator.  This requires asynchronously locking around MoveNextAsync/Current.
                         TSource element;
-                        await state.Lock.WaitAsync(state.Cancellation.Token);
+                        try
+                        {
+                            // TODO https://github.com/dotnet/runtime/issues/22144:
+                            // Use a no-throwing await if/when one is available built-in.
+                            await state.Lock.WaitAsync(state.Cancellation.Token);
+                        }
+                        catch (OperationCanceledException)
+                        {
+                            break;
+                        }
+
                         try
                         {
                             if (!await state.Enumerator.MoveNextAsync())

src/libraries/System.Threading.Tasks.Parallel/tests/ParallelForEachAsyncTests.cs

@@ -912,6 +912,36 @@ static async IAsyncEnumerable<int> Iterate()
             }));
         }
 
+        [ConditionalFact(typeof(PlatformDetection), nameof(PlatformDetection.IsThreadingSupported))]
+        public void Exception_LockWaitAsyncCancellationDoesntPropagate()
+        {
+            static async IAsyncEnumerable<int> Iterate(Task signal)
+            {
+                for (int i = 0; ; i++)
+                {
+                    if (i != 0)
+                    {
+                        await signal;
+                    }
+                    yield return i;
+                }
+            }
+
+            var signal = new TaskCompletionSource(TaskContinuationOptions.RunContinuationsAsynchronously);
+            AggregateException ae = Assert.Throws<AggregateException>(() => Parallel.ForEachAsync(Iterate(signal.Task), new ParallelOptions { MaxDegreeOfParallelism = 3 }, async (item, cancellationToken) =>
+            {
+                if (item == 0)
+                {
+                    signal.SetResult();
+                    throw new FormatException();
+                }
+                await Task.CompletedTask;
+            }).Wait());
+
+            Assert.Equal(1, ae.InnerExceptions.Count);
+            Assert.IsType<FormatException>(ae.InnerException);
+        }
+
         [ConditionalTheory(typeof(PlatformDetection), nameof(PlatformDetection.IsThreadingSupported))]
         [InlineData(false)]
         [InlineData(true)]