Pipeline: suppress pipe errors in some circumstances

Consider a pipeline

    p := pipe.New()
    p.Add(s1, s2, s3, s4)

Suppose that stage `s3` sometimes intentionally exits early without
reading all of its `stdin`. When this happens, it is common for the
previous stage `s2` to fail with a pipe error. This is because `s2`
often continues trying to write to its stdout, but the other end of
its stdout is `s3`'s stdin, which got closed when `s3` exited. The
result is that `s2` fails with `SIGPIPE`, `syscall.EPIPE`, or
`io.ErrClosedPipe`.

But if it is expected that `s3` exits early, then a pipe error from
`s2` is uninteresting. The current way of dealing with this situation
is to explicitly wrap `s2` with something like `pipe.IgnoreError(s2,
IsPipeError)`, which causes a pipe error from that stage to be
ignored. But often that's not the end of the story. If `s2` exits due
to a pipe error, then it is often the case that `s1` will _also_ exit
due to a pipe error trying to write to _its_ stdout. So to handle this
situation correctly, the pipe would have to be created like

    p := pipe.New()
    p.Add(
        pipe.IgnoreError(s1, IsPipeError),
        pipe.IgnoreError(s2, IsPipeError),
        s3,
        s4,
    )

This is verbose, and experience shows that it is easy to forget. Let's
make this simpler.

If `s3` intentionally exits without reading all of its input, it
should return the special `FinishEarly` error, thereby telling the
enclosing pipeline that its early exit was intentional. In this case,
change `Pipeline` to ignore any pipe error from the preceding stage,
`s2`. Moreover, if `s2` exits with a pipe error, ignore any pipe error
from `s1`, and so on. However, if a stage exits without an error, then
again consider a pipe error from the previous stage to be interesting.

There is some chance that this could suppress actual, interesting
errors. For example, `s2` might exit due to a `SIGPIPE` that it got
when trying to write to a file descriptor other than stdout, for
example when writing to a subprocess or to a socket, and not handle
that `EPIPE` internally. But I think that these situations will be
vanishingly rare compared to the more common case described above,
which has caused real-life irritation.

Also add tests of the new behavior.

Author: mhagger

internal/pipe/pipeline.go

@@ -190,35 +190,69 @@ func (p *Pipeline) Wait() error {
 	var earliestStageErr error
 	var earliestFailedStage Stage
 
+	finishedEarly := false
 	for i := len(p.stages) - 1; i >= 0; i-- {
 		s := p.stages[i]
 		err := s.Wait()
 
-		// Error handling:
-
-		if err == nil {
-			// No error to handle.
+		// Handle errors:
+		switch {
+		case err == nil:
+			// No error to handle. But unset the `finishedEarly` flag,
+			// because earlier stages shouldn't be affected by the
+			// later stage that finished early.
+			finishedEarly = false
 			continue
-		}
 
-		if err == FinishEarly {
+		case err == FinishEarly:
 			// We ignore `FinishEarly` errors because that is how a
 			// stage informs us that it intentionally finished early.
+			// Moreover, if we see a `FinishEarly` error, ignore any
+			// pipe error from the immediately preceding stage,
+			// because it probably came from trying to write to this
+			// stage after this stage closed its stdin.
+			finishedEarly = true
 			continue
-		}
 
-		// If we reach this point, then the stage exited with a
-		// non-ignorable error. But multiple stages might report
-		// errors, and we want to report the one that is most
-		// informative. We take that to be the error from the earliest
-		// pipeline stage that failed from a non-pipe error. If that
-		// didn't happen, take the error from the last pipeline stage
-		// that failed due to a pipe error.
-		if earliestStageErr == nil || !IsPipeError(err) {
-			// Overwrite any existing values here so that we end up
-			// retaining the last error that we see; i.e., the error
-			// that happened earliest in the pipeline.
+		case IsPipeError(err):
+			switch {
+			case finishedEarly:
+				// A successor stage finished early. It is common for
+				// this to cause earlier stages to fail with pipe
+				// errors. Such errors are uninteresting, so ignore
+				// them. Leave the `finishedEarly` flag set, because
+				// the preceding stage might get a pipe error from
+				// trying to write to this one.
+			case earliestStageErr != nil:
+				// A later stage has already reported an error. This
+				// means that we don't want to report the error from
+				// this stage:
+				//
+				// * If the later error was also a pipe error: we want
+				//   to report the _last_ pipe error seen, which would
+				//   be the one already recorded.
+				//
+				// * If the later error was not a pipe error: non-pipe
+				//   errors are always considered more important than
+				//   pipe errors, so again we would want to keep the
+				//   error that is already recorded.
+			default:
+				// In this case, the pipe error from this stage is the
+				// most important error that we have seen so far, so
+				// remember it:
+				earliestFailedStage, earliestStageErr = s, err
+			}
+
+		default:
+			// This stage exited with a non-pipe error. If multiple
+			// stages exited with such errors, we want to report the
+			// one that is most informative. We take that to be the
+			// error from the earliest failing stage. Since we are
+			// iterating through stages in reverse order, overwrite
+			// any existing remembered errors (which would have come
+			// from a later stage):
 			earliestFailedStage, earliestStageErr = s, err
+			finishedEarly = false
 		}
 	}
 

internal/pipe/pipeline_test.go

@@ -514,14 +514,12 @@ func TestScannerFinishEarly(t *testing.T) {
 	var length int64
 
 	p := pipe.New()
-	// Print the numbers from 1 to 20 (generated from scratch):
 	p.Add(
-		pipe.IgnoreError(
-			seqFunction(20),
-			pipe.IsPipeError,
-		),
-		// Pass the numbers through up to 7, then exit with an
-		// ignored error:
+		// Print the numbers from 1 to 20 (generated from scratch):
+		seqFunction(20),
+
+		// Pass the numbers through up to 7, then exit with an ignored
+		// error:
 		pipe.LinewiseFunction(
 			"finish-after-7",
 			func(_ context.Context, _ pipe.Env, line []byte, w *bufio.Writer) error {
@@ -532,6 +530,7 @@ func TestScannerFinishEarly(t *testing.T) {
 				return nil
 			},
 		),
+
 		// Read the numbers and add them into the sum:
 		pipe.Function(
 			"compute-length",
@@ -574,6 +573,189 @@ func TestPrintf(t *testing.T) {
 	}
 }
 
+func TestErrors(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+
+	err1 := errors.New("error1")
+	err2 := errors.New("error2")
+
+	for _, tc := range []struct {
+		name        string
+		stages      []pipe.Stage
+		expectedErr error
+	}{
+		{
+			name: "no-error",
+			stages: []pipe.Stage{
+				pipe.Function("noop1", genErr(nil)),
+				pipe.Function("noop2", genErr(nil)),
+				pipe.Function("noop3", genErr(nil)),
+			},
+			expectedErr: nil,
+		},
+		{
+			name: "lonely-error",
+			stages: []pipe.Stage{
+				pipe.Function("err1", genErr(err1)),
+			},
+			expectedErr: err1,
+		},
+		{
+			name: "error",
+			stages: []pipe.Stage{
+				pipe.Function("noop1", genErr(nil)),
+				pipe.Function("err1", genErr(err1)),
+				pipe.Function("noop2", genErr(nil)),
+			},
+			expectedErr: err1,
+		},
+		{
+			name: "two-consecutive-errors",
+			stages: []pipe.Stage{
+				pipe.Function("noop1", genErr(nil)),
+				pipe.Function("err1", genErr(err1)),
+				pipe.Function("err2", genErr(err2)),
+				pipe.Function("noop2", genErr(nil)),
+			},
+			expectedErr: err1,
+		},
+		{
+			name: "pipe-then-error",
+			stages: []pipe.Stage{
+				pipe.Function("noop1", genErr(nil)),
+				pipe.Function("pipe-error", genErr(io.ErrClosedPipe)),
+				pipe.Function("err1", genErr(err1)),
+				pipe.Function("noop2", genErr(nil)),
+			},
+			expectedErr: err1,
+		},
+		{
+			name: "error-then-pipe",
+			stages: []pipe.Stage{
+				pipe.Function("noop1", genErr(nil)),
+				pipe.Function("err1", genErr(err1)),
+				pipe.Function("pipe-error", genErr(io.ErrClosedPipe)),
+				pipe.Function("noop2", genErr(nil)),
+			},
+			expectedErr: err1,
+		},
+		{
+			name: "two-spaced-errors",
+			stages: []pipe.Stage{
+				pipe.Function("noop1", genErr(nil)),
+				pipe.Function("err1", genErr(err1)),
+				pipe.Function("noop2", genErr(nil)),
+				pipe.Function("err2", genErr(err2)),
+				pipe.Function("noop3", genErr(nil)),
+			},
+			expectedErr: err1,
+		},
+		{
+			name: "finish-early-ignored",
+			stages: []pipe.Stage{
+				pipe.Function("noop1", genErr(nil)),
+				pipe.Function("finish-early1", genErr(pipe.FinishEarly)),
+				pipe.Function("noop2", genErr(nil)),
+				pipe.Function("finish-early2", genErr(pipe.FinishEarly)),
+				pipe.Function("noop3", genErr(nil)),
+			},
+			expectedErr: nil,
+		},
+		{
+			name: "error-before-finish-early",
+			stages: []pipe.Stage{
+				pipe.Function("err1", genErr(err1)),
+				pipe.Function("finish-early", genErr(pipe.FinishEarly)),
+			},
+			expectedErr: err1,
+		},
+		{
+			name: "error-after-finish-early",
+			stages: []pipe.Stage{
+				pipe.Function("finish-early", genErr(pipe.FinishEarly)),
+				pipe.Function("err1", genErr(err1)),
+			},
+			expectedErr: err1,
+		},
+		{
+			name: "pipe-then-finish-early",
+			stages: []pipe.Stage{
+				pipe.Function("pipe-error", genErr(io.ErrClosedPipe)),
+				pipe.Function("finish-early", genErr(pipe.FinishEarly)),
+			},
+			expectedErr: nil,
+		},
+		{
+			name: "pipe-then-two-finish-early",
+			stages: []pipe.Stage{
+				pipe.Function("pipe-error", genErr(io.ErrClosedPipe)),
+				pipe.Function("finish-early1", genErr(pipe.FinishEarly)),
+				pipe.Function("finish-early2", genErr(pipe.FinishEarly)),
+			},
+			expectedErr: nil,
+		},
+		{
+			name: "two-pipe-then-finish-early",
+			stages: []pipe.Stage{
+				pipe.Function("pipe-error1", genErr(io.ErrClosedPipe)),
+				pipe.Function("pipe-error2", genErr(io.ErrClosedPipe)),
+				pipe.Function("finish-early", genErr(pipe.FinishEarly)),
+			},
+			expectedErr: nil,
+		},
+		{
+			name: "pipe-then-finish-early-with-gap",
+			stages: []pipe.Stage{
+				pipe.Function("pipe-error", genErr(io.ErrClosedPipe)),
+				pipe.Function("noop", genErr(nil)),
+				pipe.Function("finish-early1", genErr(pipe.FinishEarly)),
+			},
+			expectedErr: io.ErrClosedPipe,
+		},
+		{
+			name: "finish-early-then-pipe",
+			stages: []pipe.Stage{
+				pipe.Function("finish-early", genErr(pipe.FinishEarly)),
+				pipe.Function("pipe-error", genErr(io.ErrClosedPipe)),
+			},
+			expectedErr: io.ErrClosedPipe,
+		},
+		{
+			name: "error-then-pipe-then-finish-early",
+			stages: []pipe.Stage{
+				pipe.Function("err1", genErr(err1)),
+				pipe.Function("pipe-error", genErr(io.ErrClosedPipe)),
+				pipe.Function("finish-early", genErr(pipe.FinishEarly)),
+			},
+			expectedErr: err1,
+		},
+		{
+			name: "pipe-then-error-then-finish-early",
+			stages: []pipe.Stage{
+				pipe.Function("pipe-error", genErr(io.ErrClosedPipe)),
+				pipe.Function("err1", genErr(err1)),
+				pipe.Function("finish-early", genErr(pipe.FinishEarly)),
+			},
+			expectedErr: err1,
+		},
+	} {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			p := pipe.New()
+			p.Add(tc.stages...)
+			err := p.Run(ctx)
+			if tc.expectedErr == nil {
+				assert.NoError(t, err)
+			} else {
+				assert.ErrorIs(t, err, tc.expectedErr)
+			}
+		})
+	}
+}
+
 func BenchmarkSingleProgram(b *testing.B) {
 	ctx := context.Background()
 
@@ -662,3 +844,9 @@ func catFn(_ context.Context, _ pipe.Env, stdin io.Reader, stdout io.Writer) err
 	_, err := io.Copy(stdout, stdin)
 	return err
 }
+
+func genErr(err error) pipe.StageFunc {
+	return func(_ context.Context, _ pipe.Env, _ io.Reader, _ io.Writer) error {
+		return err
+	}
+}