Fix issues with runaway tasks and broken iterators

CliWrap.Tests/CancellationSpecs.cs

@@ -1,5 +1,4 @@
 using System;
-using System.Collections.Concurrent;
 using System.Diagnostics;
 using System.Reactive.Linq;
 using System.Text;
@@ -13,6 +12,7 @@
 
 namespace CliWrap.Tests;
 
+[Collection(nameof(NonParallelCollection))]
 public class CancellationSpecs
 {
     [Fact(Timeout = 15000)]
@@ -185,55 +185,54 @@ public async Task I_can_execute_a_command_as_a_pull_based_event_stream_and_cance
     }
 
     [Fact(Timeout = 15000)]
-    public async Task I_can_execute_a_command_as_a_pull_based_event_stream_with_no_unobserved_exception()
+    public async Task I_can_execute_a_command_as_a_pull_based_event_stream_and_cancel_it_without_unobserved_exception()
     {
+        // https://github.com/Tyrrrz/CliWrap/issues/336
+
         // Arrange
-        var unobservedExceptions = new ConcurrentBag<Exception>();
+        var exception = default(Exception?);
 
-        void OnUnobservedException(object? sender, UnobservedTaskExceptionEventArgs e)
+        void OnUnobservedException(object? sender, UnobservedTaskExceptionEventArgs args)
         {
-            unobservedExceptions.Add(e.Exception);
-            e.SetObserved();
+            Interlocked.CompareExchange(ref exception, args.Exception, null);
+            args.SetObserved();
         }
 
         TaskScheduler.UnobservedTaskException += OnUnobservedException;
 
-        var cmd = Cli.Wrap("dotnet").WithArguments(["--version"]);
+        var cmd = Cli.Wrap(Dummy.Program.FilePath).WithArguments(["sleep", "00:00:20"]);
 
         // Act
-        // Listening to a pull event stream followed by cancelling the listening, should no trigger any UnobservedTaskException
-        // Since the issue is a race condition, run the operation multiple times and concurently to maximize the chances of triggering it
         try
         {
+            // This is not a deterministic test, so run it multiple times to increase exposure to the race condition
             for (var i = 0; i < 50; i++)
             {
                 using var cts = new CancellationTokenSource();
 
-                await Task.Run(async () =>
+                try
                 {
-                    try
+                    await foreach (var _ in cmd.ListenAsync(cts.Token))
                     {
-                        await foreach (var _ in cmd.ListenAsync(cts.Token))
-                        {
-                            cts.Cancel();
-                        }
+                        await cts.CancelAsync();
                     }
-                    catch (OperationCanceledException) { }
-                });
+                }
+                catch (OperationCanceledException) { }
             }
 
+            await Task.Delay(500);
+
             GC.Collect();
             GC.WaitForPendingFinalizers();
             GC.Collect();
-            await Task.Delay(500);
         }
         finally
         {
             TaskScheduler.UnobservedTaskException -= OnUnobservedException;
         }
 
         // Assert
-        unobservedExceptions.Should().BeEmpty();
+        Volatile.Read(ref exception).Should().BeNull();
     }
 
     [Fact(Timeout = 15000)]

CliWrap.Tests/TestCollections.cs

@@ -0,0 +1,7 @@
+using Xunit;
+
+namespace CliWrap.Tests;
+
+// Don't run these tests in parallel because they rely on observing global state
+[CollectionDefinition(nameof(NonParallelCollection), DisableParallelization = true)]
+public class NonParallelCollection;

CliWrap/Buffered/BufferedCommandExtensions.cs

@@ -41,17 +41,16 @@ CancellationToken gracefulCancellationToken
                 PipeTarget.ToStringBuilder(stdErrBuffer, standardErrorEncoding)
             );
 
-            var commandWithPipes = command
+            // Execute the command with the pipes extended to capture the output and error streams into buffers
+            return command
                 .WithStandardOutputPipe(stdOutPipe)
-                .WithStandardErrorPipe(stdErrPipe);
-
-            return commandWithPipes
+                .WithStandardErrorPipe(stdErrPipe)
                 .ExecuteAsync(forcefulCancellationToken, gracefulCancellationToken)
                 .Bind(async task =>
                 {
                     try
                     {
-                        var result = await task;
+                        var result = await task.ConfigureAwait(false);
 
                         return new BufferedCommandResult(
                             result.ExitCode,
@@ -90,15 +89,13 @@ public CommandTask<BufferedCommandResult> ExecuteBufferedAsync(
             Encoding standardOutputEncoding,
             Encoding standardErrorEncoding,
             CancellationToken cancellationToken = default
-        )
-        {
-            return command.ExecuteBufferedAsync(
+        ) =>
+            command.ExecuteBufferedAsync(
                 standardOutputEncoding,
                 standardErrorEncoding,
                 cancellationToken,
                 CancellationToken.None
             );
-        }
 
         /// <summary>
         /// Executes the command asynchronously with buffering.
@@ -111,10 +108,7 @@ public CommandTask<BufferedCommandResult> ExecuteBufferedAsync(
         public CommandTask<BufferedCommandResult> ExecuteBufferedAsync(
             Encoding encoding,
             CancellationToken cancellationToken = default
-        )
-        {
-            return command.ExecuteBufferedAsync(encoding, encoding, cancellationToken);
-        }
+        ) => command.ExecuteBufferedAsync(encoding, encoding, cancellationToken);
 
         /// <summary>
         /// Executes the command asynchronously with buffering.
@@ -127,9 +121,6 @@ public CommandTask<BufferedCommandResult> ExecuteBufferedAsync(
         /// </remarks>
         public CommandTask<BufferedCommandResult> ExecuteBufferedAsync(
             CancellationToken cancellationToken = default
-        )
-        {
-            return command.ExecuteBufferedAsync(Encoding.Default, cancellationToken);
-        }
+        ) => command.ExecuteBufferedAsync(Encoding.Default, cancellationToken);
     }
 }

CliWrap/Command.Execution.cs

@@ -8,6 +8,7 @@
 using System.Threading.Tasks;
 using CliWrap.Exceptions;
 using CliWrap.Utils;
+using CliWrap.Utils.Extensions;
 using PowerKit.Extensions;
 
 namespace CliWrap;
@@ -144,10 +145,11 @@ private async Task PipeStandardInputAsync(
     {
         await using (process.StandardInput.ToAsyncDisposable())
         {
+            var copyTask = StandardInputPipe.CopyToAsync(process.StandardInput, cancellationToken);
+
             try
             {
-                await StandardInputPipe
-                    .CopyToAsync(process.StandardInput, cancellationToken)
+                await copyTask
                     // The input pipe may never respond to cancellation, so we add a fallback
                     // that drops the task and returns early when cancellation is requested.
                     // This prevents hanging when the process exits before consuming all stdin data.
@@ -159,13 +161,24 @@ await StandardInputPipe
                     .WaitAsync(cancellationToken)
                     .ConfigureAwait(false);
             }
-            // Expect IOException: "The pipe has been ended" (Windows) or "Broken pipe" (Unix).
-            // This may happen if the process terminated before the pipe has been exhausted.
-            // It's not an exceptional situation because the process may not need the entire
-            // stdin to complete successfully.
-            // Don't catch derived exceptions, such as FileNotFoundException, to avoid false positives.
-            // We also can't rely on process.HasExited here because of potential race conditions.
-            catch (IOException ex) when (ex.GetType() == typeof(IOException)) { }
+            catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+            {
+                // CopyToAsync() may not honor cancellation and can fault after WaitAsync() returns.
+                // Observe that fault to avoid unobserved task exceptions.
+                _ = copyTask.ObserveException();
+
+                throw;
+            }
+            catch (IOException ex)
+                // Don't catch derived exceptions, such as FileNotFoundException, to avoid false positives.
+                // We also can't rely on process.HasExited here because of potential race conditions.
+                when (ex.GetType() == typeof(IOException))
+            {
+                // Expect IOException: "The pipe has been ended" (Windows) or "Broken pipe" (Unix).
+                // This may happen if the process terminated before the pipe has been exhausted.
+                // It's not an exceptional situation because the process may not need the entire
+                // stdin to complete successfully.
+            }
         }
     }
 
@@ -250,39 +263,37 @@ private async Task<CommandResult> ExecuteAsync(
         catch (OperationCanceledException ex) when (ex.CancellationToken == waitTimeoutCts.Token)
         {
             // We tried to kill the process, but it didn't exit within the allotted timeout, meaning
-            // that the termination attempt failed. This should never happen, but inform the user if it does.
+            // that the termination attempt failed. This should never happen, but it's not impossible.
             throw new TimeoutException(
                 $"Failed to terminate the underlying process ({process.Name}#{process.Id}) within the allotted timeout.",
                 ex
             );
         }
-        catch (OperationCanceledException ex) when (ex.CancellationToken == stdInCts.Token)
-        {
-            // This clause will be hit both when stdin piping is canceled due to process exit
-            // and when it aborts due to an actual cancellation request (because of the link).
-            // Swallow this exception because it was triggered by an internal cancellation,
-            // we will throw a more meaningful one later if needed.
-        }
         catch (OperationCanceledException ex)
-            when (ex.CancellationToken == forcefulCancellationToken
-                || ex.CancellationToken == gracefulCancellationToken
+            when (ex.CancellationToken == stdInCts.Token
+                && !forcefulCancellationToken.IsCancellationRequested
             )
         {
-            // This clause should never hit due to the registrations above, but just in case it does,
-            // swallow the exception here to throw a more meaningful one later.
+            // The process has exited on its own, but the stdin pipe was still trying to write data to it.
+            // This is an internal cancellation that is not meant to be surfaced to the user.
         }
-
-        // Check if the process exited after forceful cancellation
-        if (forcefulCancellationToken.IsCancellationRequested)
+        catch (OperationCanceledException ex)
+            // The exception's own token may be the stdin cancellation token, because it's linked,
+            // so we need to check the forceful cancellation token directly.
+            when (forcefulCancellationToken.IsCancellationRequested)
         {
+            // We tried to kill the process and it exited. Rethrow a more meaningful exception.
             throw new OperationCanceledException(
                 "Command execution canceled. "
                     + $"Underlying process ({process.Name}#{process.Id}) was forcefully terminated.",
+                ex,
                 forcefulCancellationToken
             );
         }
 
-        // Check if the process exited after graceful cancellation
+        // The process has exited on its own, but it might have done so because we requested a graceful cancellation.
+        // Check the token manually because we don't pass it to any of the other methods and won't get the exception
+        // propagated automatically.
         if (gracefulCancellationToken.IsCancellationRequested)
         {
             throw new OperationCanceledException(

CliWrap/CommandResult.cs

@@ -13,7 +13,7 @@ public partial class CommandResult(int exitCode, DateTimeOffset startTime, DateT
     public int ExitCode { get; } = exitCode;
 
     /// <summary>
-    /// Whether the command execution was successful (i.e. exit code is zero).
+    /// Whether the command execution was successful (i.e., exit code is zero).
     /// </summary>
     public bool IsSuccess => ExitCode == 0;