diff --git a/pkg/commands/oscommands/os.go b/pkg/commands/oscommands/os.go index 216ff7b5b..1a6921c55 100644 --- a/pkg/commands/oscommands/os.go +++ b/pkg/commands/oscommands/os.go @@ -201,7 +201,7 @@ func (c *OSCommand) FileExists(path string) (bool, error) { // PipeCommands runs a heap of commands and pipes their inputs/outputs together like A | B | C func (c *OSCommand) PipeCommands(cmdObjs ...*CmdObj) error { - c.LogCommand(pipelineString(cmdObjs), true) + c.logPipeline(cmdObjs) cmds, parentEnds, err := wirePipeline(cmdObjs) if err != nil { diff --git a/pkg/commands/oscommands/pipeline.go b/pkg/commands/oscommands/pipeline.go index 223b5efa7..668f24486 100644 --- a/pkg/commands/oscommands/pipeline.go +++ b/pkg/commands/oscommands/pipeline.go @@ -1,6 +1,7 @@ package oscommands import ( + "fmt" "io" "os" "os/exec" @@ -9,6 +10,100 @@ import ( "github.com/samber/lo" ) +// Pipeline is a chain of running commands, each one's output feeding the next +// one's input, with the last one's output going somewhere the caller reads. Its +// method set is the one a render task expects of a command (see tasks.Cmd), so +// a pipeline can render a view just as a single command can. +type Pipeline struct { + cmds []*exec.Cmd + cmdStr string +} + +// StartPipeline starts the given commands wired A | B | C and returns the +// pipeline together with the reader for its output. +// +// Every command's stderr goes to that same output, so whatever a command +// complains about is part of what the caller reads. A diff renderer's error +// message belongs on screen with the diff it failed to render. +// +// Closing the reader is how a pipeline is brought down. The last command's next +// write fails, so it exits, and the failure travels back up the chain as each +// command in turn writes into a pipe whose reader is gone. +func (c *OSCommand) StartPipeline(cmdObjs ...*CmdObj) (*Pipeline, io.ReadCloser, error) { + c.logPipeline(cmdObjs) + + cmds, parentEnds, err := wirePipeline(cmdObjs) + if err != nil { + return nil, nil, err + } + + reader, writer, err := os.Pipe() + if err != nil { + closeAll(parentEnds) + return nil, nil, err + } + for _, cmd := range cmds { + cmd.Stderr = writer + } + cmds[len(cmds)-1].Stdout = writer + parentEnds = append(parentEnds, writer) + + started, err := startPipeline(cmds, parentEnds) + if err != nil { + for _, cmd := range cmds[:started] { + _ = cmd.Wait() + } + _ = reader.Close() + return nil, nil, err + } + + return &Pipeline{cmds: cmds, cmdStr: pipelineString(cmdObjs)}, reader, nil +} + +func (self *Pipeline) String() string { + return self.cmdStr +} + +// Wait waits for every command to exit and reports the failure nearest the end +// of the pipeline. A command that fails leaves the ones before it writing into +// a pipe nobody reads, so their own broken-pipe failures are consequences of it +// rather than the cause worth reporting, while a command that fails early +// leaves the ones after it with nothing to read and no reason to fail at all. +func (self *Pipeline) Wait() error { + var lastErr error + for _, cmd := range self.cmds { + if err := cmd.Wait(); err != nil { + lastErr = fmt.Errorf("%s: %w", cmd.String(), err) + } + } + + return lastErr +} + +// Terminate asks every command to stop, without waiting for any of them. On +// platforms where that does nothing, the pipeline comes down when its output +// reader is closed; see StartPipeline. +func (self *Pipeline) Terminate() error { + var firstErr error + for _, cmd := range self.cmds { + if err := TerminateProcessGracefully(cmd.Process); err != nil && firstErr == nil { + firstErr = err + } + } + + return firstErr +} + +// logPipeline enters a chain of commands into the command log, unless the first +// command was marked not to be logged; it speaks for the pipeline. A render runs +// its pipeline again on every selection change, so a caller has to be able to +// keep it out of the log. +func (c *OSCommand) logPipeline(cmdObjs []*CmdObj) { + if cmdObjs[0].ShouldLog() { + c.LogCommand(pipelineString(cmdObjs), true) + } +} + // pipelineString names a chain of commands the way a shell would write it. func pipelineString(cmdObjs []*CmdObj) string { return strings.Join( diff --git a/pkg/commands/oscommands/pipeline_test.go b/pkg/commands/oscommands/pipeline_test.go index 8bcd08fce..e49aa3441 100644 --- a/pkg/commands/oscommands/pipeline_test.go +++ b/pkg/commands/oscommands/pipeline_test.go @@ -56,6 +56,75 @@ func TestPipelineMember(t *testing.T) { os.Exit(0) } +func TestStartPipelineStreamsTheOutputOfTheLastCommand(t *testing.T) { + pipeline, reader, err := NewDummyOSCommand().StartPipeline( + pipelineMember("count"), + pipelineMember("upcase"), + ) + assert.NoError(t, err) + + output, err := io.ReadAll(reader) + assert.NoError(t, err) + assert.Equal(t, "LINE 1\nLINE 2\nLINE 3\n", string(output)) + + assert.NoError(t, pipeline.Wait()) + assert.NoError(t, reader.Close()) +} + +func TestStartPipelineReadsWhatTheCommandsComplainAbout(t *testing.T) { + pipeline, reader, err := NewDummyOSCommand().StartPipeline( + pipelineMember("count"), + pipelineMember("complain"), + ) + assert.NoError(t, err) + + output, err := io.ReadAll(reader) + assert.NoError(t, err) + assert.Equal(t, "something went wrong\n", string(output)) + + // The failure of the command nearest the output is the one reported, even + // though the one feeding it was left writing into a pipe nobody reads. + assert.ErrorContains(t, pipeline.Wait(), "exit status 3") + + assert.NoError(t, reader.Close()) +} + +func TestClosingAPipelinesOutputBringsItDown(t *testing.T) { + pipeline, reader, err := NewDummyOSCommand().StartPipeline( + pipelineMember("flood"), + pipelineMember("copy"), + ) + assert.NoError(t, err) + + // Read some output first, so that both commands are past their startup and + // really running when the reader goes. + buf := make([]byte, len("line 1\n")) + _, err = io.ReadFull(reader, buf) + assert.NoError(t, err) + assert.Equal(t, "line 1\n", string(buf)) + + assert.NoError(t, reader.Close()) + + done := make(chan error, 1) + go func() { done <- pipeline.Wait() }() + + select { + case <-done: + case <-time.After(10 * time.Second): + t.Fatal("the pipeline was still running long after its output was closed") + } +} + +func TestStartPipelineReportsACommandItCannotStart(t *testing.T) { + osCommand := NewDummyOSCommand() + + _, _, err := osCommand.StartPipeline( + pipelineMember("count"), + osCommand.Cmd.New([]string{"lazygit-no-such-command"}), + ) + assert.Error(t, err) +} + func TestPipeCommandsReturnsWhenALaterCommandDiesEarly(t *testing.T) { done := make(chan error, 1) go func() {