From a8cd56ba9b9155a997c018088c669cc47098e225 Mon Sep 17 00:00:00 2001 From: Stefan Haller Date: Tue, 15 Sep 2026 16:57:01 +0200 Subject: [PATCH] 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) --- pkg/commands/oscommands/os.go | 4 +- pkg/commands/oscommands/pipeline.go | 39 ++++++++++--- pkg/commands/oscommands/pipeline_test.go | 74 ++++++++++++++++++++++++ 3 files changed, 108 insertions(+), 9 deletions(-) create mode 100644 pkg/commands/oscommands/pipeline_test.go diff --git a/pkg/commands/oscommands/os.go b/pkg/commands/oscommands/os.go index 0d8ae64dd..216ff7b5b 100644 --- a/pkg/commands/oscommands/os.go +++ b/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{} diff --git a/pkg/commands/oscommands/pipeline.go b/pkg/commands/oscommands/pipeline.go index 1fcab36ae..223b5efa7 100644 --- a/pkg/commands/oscommands/pipeline.go +++ b/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() + } +} diff --git a/pkg/commands/oscommands/pipeline_test.go b/pkg/commands/oscommands/pipeline_test.go new file mode 100644 index 000000000..8bcd08fce --- /dev/null +++ b/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") + } +}