Let a pipeline's commands notice when the next one is gone

If a command in a pipeline exits before reading all of its input, the
command feeding it keeps running and PipeCommands never returns.

The parent holds on to the read end of every pipe it wires between two
commands. A pipe with a reader is a pipe worth writing to, so the
command writing into it is never told that nobody is listening.

Close the parent's ends once the commands are running, since each of
them holds its own by then. The write ends have to go as well, or the
command reading a link never reaches the end of its input.

No caller reaches this today. The one chain lazygit pipes is
`git stash show -p` into `git apply -R`, and apply reads its whole input
before it does anything with it, so it never leaves the show writing to
nobody. The chain about to be added for diff renderers is a different
matter: a renderer can fail at any point of its input, and a render that
is no longer wanted is stopped part way through by design.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

pkg/commands/oscommands/os.go

@@ -203,7 +203,7 @@ func (c *OSCommand) FileExists(path string) (bool, error) {
 func (c *OSCommand) PipeCommands(cmdObjs ...*CmdObj) error {
 	c.LogCommand(pipelineString(cmdObjs), true)
 
-	cmds, err := wirePipeline(cmdObjs)
+	cmds, parentEnds, err := wirePipeline(cmdObjs)
 	if err != nil {
 		return err
 	}
@@ -213,7 +213,7 @@ func (c *OSCommand) PipeCommands(cmdObjs ...*CmdObj) error {
 		cmds[i].Stderr = &stderrs[i]
 	}
 
-	started, startErr := startPipeline(cmds)
+	started, startErr := startPipeline(cmds, parentEnds)
 
 	finalErrors := []string{}
 

pkg/commands/oscommands/pipeline.go

@@ -1,6 +1,8 @@
 package oscommands
 
 import (
+	"io"
+	"os"
 	"os/exec"
 	"strings"
 
@@ -18,22 +20,30 @@ func pipelineString(cmdObjs []*CmdObj) string {
 }
 
 // wirePipeline connects each command's output to the next one's input, like
-// A | B | C. The last command's output is left for the caller to direct.
-func wirePipeline(cmdObjs []*CmdObj) ([]*exec.Cmd, error) {
+// A | B | C, and returns the commands along with the parent's ends of those
+// pipes. The last command's output is left for the caller to direct.
+//
+// The parent's ends have to be closed once the commands are running.
+// startPipeline does that; see there for why it matters.
+func wirePipeline(cmdObjs []*CmdObj) ([]*exec.Cmd, []io.Closer, error) {
 	cmds := lo.Map(cmdObjs, func(cmdObj *CmdObj, _ int) *exec.Cmd {
 		return cmdObj.GetCmd()
 	})
 
+	parentEnds := []io.Closer{}
 	for i := range len(cmds) - 1 {
-		stdout, err := cmds[i].StdoutPipe()
+		reader, writer, err := os.Pipe()
 		if err != nil {
-			return nil, err
+			closeAll(parentEnds)
+			return nil, nil, err
 		}
 
-		cmds[i+1].Stdin = stdout
+		cmds[i].Stdout = writer
+		cmds[i+1].Stdin = reader
+		parentEnds = append(parentEnds, reader, writer)
 	}
 
-	return cmds, nil
+	return cmds, parentEnds, nil
 }
 
 // startPipeline starts every command and reports how many it got going. Every
@@ -41,11 +51,20 @@ func wirePipeline(cmdObjs []*CmdObj) ([]*exec.Cmd, error) {
 // the pipe that feeds the next command, and one that hasn't been started by
 // then would inherit a closed stdin.
 //
+// Once they are all running, each of them holds its own ends of the pipes it
+// reads and writes, and the parent lets go of its copies. Both directions
+// matter. While the parent holds the read end of a link, a command writing
+// into it never learns that the command meant to read it is gone, and keeps
+// running after the pipeline has been brought down. While the parent holds the
+// write end, the command reading it never reaches the end of its input.
+//
 // When a command fails to start, the ones already running are killed, since
 // without the rest of the pipeline to drain them they could block forever
 // writing to a full pipe. They still have to be reaped, so the count covers
 // them too.
-func startPipeline(cmds []*exec.Cmd) (int, error) {
+func startPipeline(cmds []*exec.Cmd, parentEnds []io.Closer) (int, error) {
+	defer closeAll(parentEnds)
+
 	for i, cmd := range cmds {
 		if err := cmd.Start(); err != nil {
 			for _, started := range cmds[:i] {
@@ -58,3 +77,9 @@ func startPipeline(cmds []*exec.Cmd) (int, error) {
 
 	return len(cmds), nil
 }
+
+func closeAll(closers []io.Closer) {
+	for _, closer := range closers {
+		_ = closer.Close()
+	}
+}

pkg/commands/oscommands/pipeline_test.go

@@ -0,0 +1,74 @@
+package oscommands
+
+import (
+	"fmt"
+	"io"
+	"os"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+// The pipeline tests need programs to run, and the test binary is the one
+// program every platform we test on is sure to have. pipelineMember builds a
+// command that re-runs this binary in the role a member of the pipeline is to
+// play; the roles are in TestPipelineMember.
+const pipelineRoleEnvVar = "LAZYGIT_TEST_PIPELINE_ROLE"
+
+func pipelineMember(role string) *CmdObj {
+	return NewDummyOSCommand().Cmd.
+		New([]string{os.Args[0], "-test.run=^TestPipelineMember$"}).
+		AddEnvVars(pipelineRoleEnvVar + "=" + role)
+}
+
+// TestPipelineMember is the program the pipeline tests run, not a test of its
+// own. It exits before the testing package reports anything, so that its output
+// is what the role wrote and nothing else.
+func TestPipelineMember(t *testing.T) {
+	switch os.Getenv(pipelineRoleEnvVar) {
+	case "":
+		t.Skip("not a test; the pipeline tests run this binary in a role")
+	case "count":
+		for i := 1; i <= 3; i++ {
+			fmt.Printf("line %d\n", i)
+		}
+	case "upcase":
+		input, _ := io.ReadAll(os.Stdin)
+		fmt.Print(strings.ToUpper(string(input)))
+	case "copy":
+		_, _ = io.Copy(os.Stdout, os.Stdin)
+	case "complain":
+		fmt.Fprintln(os.Stderr, "something went wrong")
+		os.Exit(3)
+	case "flood":
+		// A failed write means the reader is gone, and there is no point
+		// writing to nobody. On platforms that raise a signal for it instead,
+		// this process is already dead by the time the write returns.
+		for i := 1; ; i++ {
+			if _, err := fmt.Printf("line %d\n", i); err != nil {
+				break
+			}
+		}
+	}
+
+	os.Exit(0)
+}
+
+func TestPipeCommandsReturnsWhenALaterCommandDiesEarly(t *testing.T) {
+	done := make(chan error, 1)
+	go func() {
+		done <- NewDummyOSCommand().PipeCommands(
+			pipelineMember("flood"),
+			pipelineMember("complain"),
+		)
+	}()
+
+	select {
+	case err := <-done:
+		assert.ErrorContains(t, err, "something went wrong")
+	case <-time.After(10 * time.Second):
+		t.Fatal("PipeCommands was still waiting for a command whose output nothing reads")
+	}
+}