fix: ensure process output pipe is drained by caller

src/process.tests/NullPipeConcurrencyTests.cs

@@ -0,0 +1,59 @@
+// Copyright (c) The Geekeey Authors
+// SPDX-License-Identifier: EUPL-1.2
+
+namespace Geekeey.Process.Tests;
+
+internal sealed class NullPipeConcurrencyTests
+{
+	// A process writing enough output to fill its pipe can receive SIGPIPE if the read end is closed too early.
+	// Run many chatty processes at once to make premature pipe closure observable and to exercise pipe lifetime
+	// while processes are being started and completed concurrently.
+	// Test both the default targets and explicitly configured null targets, since both must keep consuming
+	// process output until the process exits.
+
+	[Test]
+	public async Task Chatty_commands_can_run_concurrently_with_default_null_output_pipes()
+	{
+		var commands = new List<Command>();
+
+		for (var i = 0; i < 64; i++)
+		{
+			commands.Add(new Command(Testing.Fixture.Program.FilePath)
+				.WithArguments(["generate", "clob", "--length", "100000", "--lines", "2"]));
+		}
+
+		var results = await Task.WhenAll(commands.Select(async command => await command.ExecuteAsync()));
+
+		using (Assert.Multiple())
+		{
+			foreach (var result in results)
+			{
+				await Assert.That(result.ExitCode).IsZero();
+			}
+		}
+	}
+
+	[Test]
+	public async Task Chatty_commands_can_run_concurrently_with_explicit_null_output_pipes()
+	{
+		var commands = new List<Command>();
+
+		for (var i = 0; i < 64; i++)
+		{
+			commands.Add(new Command(Testing.Fixture.Program.FilePath)
+				.WithArguments(["generate", "clob", "--length", "100000", "--lines", "2"])
+				.WithStandardOutputPipe(PipeTarget.Null)
+				.WithStandardErrorPipe(PipeTarget.Null));
+		}
+
+		var results = await Task.WhenAll(commands.Select(async command => await command.ExecuteAsync()));
+
+		using (Assert.Multiple())
+		{
+			foreach (var result in results)
+			{
+				await Assert.That(result.ExitCode).IsZero();
+			}
+		}
+	}
+}

src/process/Command.Execute.cs

@@ -11,8 +11,6 @@ public sealed partial class Command
 		return process.MainModule?.FileName;
 	});
 
-	private static readonly TimeSpan CancelWaitTimeout = TimeSpan.FromSeconds(5);
-
 	private static string? ProcessPath => ProcessPathLazy.Value;
 
 	/// <summary>
@@ -135,60 +133,90 @@ public sealed partial class Command
 	{
 		using var _ = process;
 
-		// timeout is triggered when the cancel timeout expires after we tried to stop the process
+		// Used to trigger forceful cancellation also if an exception is thrown within this method,
-		// -> release wait for exit and pumping tasks after that timeout
+		// for example by the consumer-provided pipes. This ensures that the underlying process
-		using var timeout = new CancellationTokenSource();
+		// never outlives the execution of this method.
+		using var stop = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
 
-		var stdout = PipeStdOutAsync(process, timeout.Token);
+		// Used to trigger cancellation of the stdin pipe also when the process exits. The process may
-		var stderr = PipeStdErrAsync(process, timeout.Token);
+		// exit without fully consuming all the stdin data, so we want to avoid waiting for the
+		// pipe to finish if there's nothing reading from it anymore.
+		using var abort = CancellationTokenSource.CreateLinkedTokenSource(stop.Token);
 
-		var stdin = PipeStdInAsync(process, timeout.Token);
+		// Kill the process when forceful termination (via cancellation or panic) is requested
+		await using var watch = stop.Token.Register(process.Kill);
 
-		var pump = Task.WhenAll(stdout, stderr, stdin);
+		// Start piping streams in the background. In the event that any of the tasks fail,
+		// the corresponding standard stream(s) will be closed, so there is no risk of a deadlock.
+		// Output and error streams may legally outlive the process, so we don't cancel them on exit.
+		var stdin = PipeStdInAsync(process, abort.Token);
+		var stdout = PipeStdOutAsync(process, stop.Token);
+		var stderr = PipeStdErrAsync(process, stop.Token);
 
-		await using var registration = cancellationToken.Register(Stop, (process, timeout));
+		// Wait for the process to exit normally or get killed
-
+		// We deliberately don't time out here because we don't want to leave a detached running process.
-		// wait for the process to exit or cancellation to be requested when cancellation is requested,
+		// If the user wants to end the execution early, they can already do so by triggering cancellation
-		// we try to stop the process and then wait for it to exit with a timeout.
+		// and killing the process. All this timeout would help us with is handle a rare edge case where
-		// When the timeout expires, we cancel the pumping tasks as well as the wait for the process exit.
+		// the kill signal is sent but the process doesn't terminate for some reason. However, we don't
+		// really have anything we can do in that situation anyway.
+		var exe = process.WaitForExitAsync(CancellationToken.None);
 
 		try
 		{
-			await process.WaitForExitAsync(CancellationToken.None).WaitAsync(timeout.Token);
+			await foreach (var done in Task.WhenEach(stdin, stdout, stderr, exe).WithCancellation(CancellationToken.None))
-		}
-		catch (OperationCanceledException)
-		{
-		}
-
-		// we still wait for the pumping to complete but ignore cancellation here
-		try
-		{
-			await pump;
-		}
-		catch (OperationCanceledException)
-		{
-		}
-
-		// if cancellation was requested, throw after the process was tried to stop
-		cancellationToken.ThrowIfCancellationRequested();
-
-		if (process.ExitCode is 0 || !Validation.HasFlag(ValidationMode.ZeroExitCode))
-		{
-			return new CommandResult(process.ExitCode, process.StartTime, process.ExitTime);
-		}
-
-		var message = $"Command execution failed because the underlying process ({process.FileName}#{process.Id}) " +
-					  $"returned a non-zero exit code ({process.ExitCode}).";
-		throw new CommandExecutionException(this, process.ExitCode, message);
-
-		static void Stop(object? state)
-		{
-			if (state is (Process process, CancellationTokenSource timeout))
 			{
-				timeout.CancelAfter(CancelWaitTimeout);
+				// If the process task has finished, trigger the corresponding signal to cancel the
-				process.Kill();
+				// stdin pipe task since the process can't read any more data from it.
+				if (done == exe)
+				{
+					await abort.CancelAsync().ConfigureAwait(false);
+				}
+				// If a piping task failed while the process is still running, proactively terminate the process.
+				// It may continue running for a while even with the corresponding standard stream(s) closed, so
+				// we want to cut the wait short since we're going to throw an exception down the line anyway.
+				else if (done == stdin || done == stdout || done == stderr)
+				{
+					if (!done.IsCompletedSuccessfully && !exe.IsCompleted)
+					{
+						await stop.CancelAsync().ConfigureAwait(false);
+					}
+				}
 			}
+
+			// Join all tasks and propagate exceptions. If any of the tasks faulted, this will throw an aggregation
+			// of their exceptions. If some tasks failed while others were canceled, then the failures take
+			// precedence and the cancellation exceptions will be suppressed. If none of the tasks failed but
+			// some or all were canceled, then a single cancellation exception will be thrown.
+			await Task.WhenAll(stdin, stdout, stderr, exe).ConfigureAwait(false);
 		}
+		catch (OperationCanceledException exception)
+			when (exception.CancellationToken == stop.Token || exception.CancellationToken == abort.Token)
+		{
+			// Cancellation was either requested by the consumer or triggered internally. Consumer-initiated
+			// cancellations will be reported separately later, while the internal ones shouldn't be reported.
+		}
+		finally
+		{
+			// The process must never outlive the execution of this method
+			await exe.ConfigureAwait(false);
+		}
+
+		// Report forceful cancellation
+		if (cancellationToken.IsCancellationRequested)
+		{
+			var message = $"Command execution canceled. Underlying process ({process.FileName}#{process.Id}) was forcefully terminated.";
+			throw new OperationCanceledException(message, cancellationToken);
+		}
+
+		// Validate the exit code if required
+		if (process.ExitCode is not 0 && Validation.HasFlag(ValidationMode.ZeroExitCode))
+		{
+			var message = $"Command execution failed because the underlying process ({process.FileName}#{process.Id}) " +
+						  $"returned a non-zero exit code ({process.ExitCode}).";
+			throw new CommandExecutionException(this, process.ExitCode, message);
+		}
+
+		return new CommandResult(process.ExitCode, process.StartTime, process.ExitTime);
 	}
 
 	private async Task PipeStdOutAsync(Process process, CancellationToken cancellationToken = default)
@@ -211,12 +239,22 @@ public sealed partial class Command
 	{
 		await using (process.StandardInput)
 		{
+			var copyTask = StandardInputPipe.CopyToAsync(process.StandardInput, cancellationToken);
+
 			try
 			{
 				// Some streams do not support cancellation, so we add a fallback that drops the task and returns early.
 				// This is important with stdin because the process might finish before the pipe has been fully
 				// exhausted, and we don't want to wait for it.
-				await StandardInputPipe.CopyToAsync(process.StandardInput, cancellationToken).WaitAsync(cancellationToken);
+				await copyTask.WaitAsync(cancellationToken).ConfigureAwait(false);
+			}
+			catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+			{
+				// We tried to cancel the copy task, but it may have not cooperated and may still be
+				// running in the background. To make sure its exception doesn't bubble up to the scheduler,
+				// we explicitly observe it.
+				_ = Catch(copyTask);
+				throw;
 			}
 			// Expect IOException: "The pipe has been ended" (Windows) or "Broken pipe" (Unix). This may happen if the
 			// process is terminated before the pipe has been exhausted. It's not an exceptional situation because the
@@ -226,6 +264,18 @@ public sealed partial class Command
 			{
 				// Don't catch derived exceptions, such as FileNotFoundException, to avoid false positives.
 			}
+
+			static async Task Catch(Task t)
+			{
+				try
+				{
+					await t;
+				}
+				catch
+				{
+					// ignored exception
+				}
+			}
 		}
 	}
 }

src/process/PipeTarget.cs

@@ -119,14 +119,10 @@ public partial class PipeTarget
 	/// Pipe target that discards all data. Functionally equivalent to a null device.
 	/// </summary>
 	/// <remarks>
-	/// Using this target results in the corresponding stream (standard output or standard error) not being opened for
+	/// The corresponding stream is consumed for the lifetime of the underlying process so that the process never writes
-	/// the underlying process at all. In the vast majority of cases, this behavior should be functionally equivalent to
+	/// to a pipe whose read end has already been closed.
-	/// piping to a null stream, but without the performance overhead of consuming and discarding unneeded data. This
-	/// may be undesirable in certain situations, in which case it's recommended to pipe to a null stream explicitly
-	/// using <see cref="ToStream(Stream)" /> with <see cref="Stream.Null" />.
 	/// </remarks>
-	public static PipeTarget Null { get; } = Create((_, cancellationToken) =>
+	public static PipeTarget Null { get; } = ToStream(Stream.Null);
-		!cancellationToken.IsCancellationRequested ? Task.CompletedTask : Task.FromCanceled(cancellationToken));
 
 	/// <summary>
 	/// Creates an anonymous pipe target with the <see cref="CopyFromAsync(Stream, CancellationToken)" /> method