Collect Errors From Concurrent Pipelines in Go Safely

A pipeline is a chain of stages joined by channels: one stage produces values, the next transforms them, a last one stores or sends them. It is one of the best-known concurrency shapes in Go, and the version in the textbooks is all happy path. Real stages fail. A line cannot be parsed, a lookup times out, a write is rejected. Once stages run in separate goroutines, the question stops being "how do I return the error" and becomes "how does the rest of the pipeline find out, and stop?"

This article answers it for two policies, stop on the first error and keep going, shows the leak that the careless version has, and gives a test. It builds on returning errors from goroutines, the pipeline pattern and errgroup, and belongs to the error handling guide. Everything ran on Go 1.25.6, with golang.org/x/sync v0.19.0 for errgroup.

The shape of a stage

Each stage is a function that takes an input channel and returns an output channel, or in the fail-fast version, is run by an errgroup. Two rules from the pipeline pattern matter for errors. The goroutine that sends on a channel closes it, usually with defer close(out), so the next stage's range ends. And every send and receive must be able to give up, because when a stage stops early, its neighbors must not be left waiting.

That second rule is the one that errors break. In a happy-path pipeline a stage only stops when its input is exhausted. With errors, a stage may stop in the middle, and the producer upstream is then trying to send to a channel nobody reads.

Policy 1: stop everything on the first error

Use this when the items belong together and a failure makes the rest pointless: loading a file where line 40 is corrupt, processing a batch that must be all-or-nothing. The tools are an errgroup with a context:

package main

import (
	"context"
	"fmt"
	"strconv"

	"golang.org/x/sync/errgroup"
)

func main() {
	lines := []string{"10", "20", "oops", "40", "50"}

	g, ctx := errgroup.WithContext(context.Background())
	raw := make(chan string)
	parsed := make(chan int)

	// Stage 1: produce lines.
	g.Go(func() error {
		defer close(raw)
		for _, l := range lines {
			select {
			case raw <- l:
			case <-ctx.Done():
				return ctx.Err()
			}
		}
		return nil
	})

	// Stage 2: parse, failing on bad input.
	g.Go(func() error {
		defer close(parsed)
		for l := range raw {
			n, err := strconv.Atoi(l)
			if err != nil {
				return fmt.Errorf("parse stage: %w", err)
			}
			select {
			case parsed <- n:
			case <-ctx.Done():
				return ctx.Err()
			}
		}
		return nil
	})

	// Stage 3: sum.
	var sum int
	g.Go(func() error {
		for n := range parsed {
			sum += n
		}
		return nil
	})

	err := g.Wait()
	fmt.Println("sum so far:", sum)
	fmt.Println("error:", err)
}

sum so far: 30
error: parse stage: strconv.Atoi: parsing "oops": invalid syntax

Walk through the failure. Stage 2 hits oops and returns an error. The errgroup records the first error, cancels the context, and defer close(parsed) ends stage 3's loop. Stage 1 may be blocked trying to send the next line on raw; its select sees ctx.Done() fire and returns, so it exits too. Wait returns the first error: the parse failure, and not the context canceled that stage 1 reported, because the errgroup keeps whichever came first. The sum printed is the work done before the failure, which is 30 here (the lines 10 and 20). Nothing is left running.

Three details make this correct, and each is a place to check in review.

Every channel operation that can block selects on ctx.Done(). Stage 2's receive from raw is a range, which is safe only because stage 1 closes raw on every path through defer. If it were possible for stage 1 to stop without closing, stage 2 would wait forever.

Each producer closes its own output with defer. The defer close(parsed) in stage 2 runs on both the success and the error return, which is what lets stage 3 end.

The errgroup's context is the only cancellation signal, shared by every stage. Don't create a second one.

What the careless version leaks

The most common mistake is a stage that returns early on an error without a way for its neighbors to know. Compare the producer that does not select on the context:

package main

import (
	"errors"
	"fmt"
	"runtime"
	"time"
)

func run(careful bool) int {
	before := runtime.NumGoroutine()
	done := make(chan struct{})
	out := make(chan int)

	go func() { // producer
		defer close(out)
		for i := 0; i < 100; i++ {
			if careful {
				select {
				case out <- i:
				case <-done:
					return
				}
			} else {
				out <- i
			}
		}
	}()

	// consumer fails on the third item and stops reading
	var err error
	for v := range out {
		if v == 2 {
			err = errors.New("bad item")
			break
		}
	}
	close(done)
	_ = err

	time.Sleep(50 * time.Millisecond)
	return runtime.NumGoroutine() - before
}

func main() {
	fmt.Println("careless leaks:", run(false))
	fmt.Println("careful leaks: ", run(true))
}

careless leaks: 1
careful leaks:  0

The careless producer is blocked on out <- 3 with no receiver for the rest of the process's life. The careful one sees done closed and returns. The number printed is the difference in goroutine count, 1 for the careless version and 0 for the careful one. One leaked goroutine per failed run is invisible in a test and fatal in a long-running service that processes a thousand batches an hour. Goroutine leaks and detecting them in tests cover the diagnosis.

Policy 2: keep going and report every error

When the items are independent, a bad one should not stop the good ones. The error becomes part of the data flowing through the stages: each message carries its value, its position and its error, and a stage that sees an error passes it along unchanged.

package main

import (
	"errors"
	"fmt"
	"strconv"
	"sync"
)

type item struct {
	index int
	raw   string
	value int
	err   error
}

func produce(lines []string) <-chan item {
	out := make(chan item)
	go func() {
		defer close(out)
		for i, l := range lines {
			out <- item{index: i, raw: l}
		}
	}()
	return out
}

func parse(in <-chan item, workers int) <-chan item {
	out := make(chan item)
	var wg sync.WaitGroup
	for range workers {
		wg.Go(func() {
			for it := range in {
				if it.err == nil {
					n, err := strconv.Atoi(it.raw)
					if err != nil {
						it.err = fmt.Errorf("line %d: %w", it.index+1, err)
					}
					it.value = n
				}
				out <- it
			}
		})
	}
	go func() {
		wg.Wait()
		close(out)
	}()
	return out
}

func main() {
	lines := []string{"10", "20", "oops", "40", "5x", "60"}

	sum := 0
	var errs []error
	seen := 0
	for it := range parse(produce(lines), 3) {
		seen++
		if it.err != nil {
			errs = append(errs, it.err)
			continue
		}
		sum += it.value
	}
	// Workers finish in any order; sort the errors by line for stable output.
	for i := 0; i < len(errs); i++ {
		for j := i + 1; j < len(errs); j++ {
			if errs[j].Error() < errs[i].Error() {
				errs[i], errs[j] = errs[j], errs[i]
			}
		}
	}
	fmt.Println("items:", seen, "sum:", sum)
	fmt.Println(errors.Join(errs...))
}

items: 6 sum: 130
line 3: strconv.Atoi: parsing "oops": invalid syntax
line 5: strconv.Atoi: parsing "5x": invalid syntax

Every line is processed, three workers run the parse in parallel, and the final report has the sum of the good lines and an error for each bad one with its line number. No stage stops, so no cancellation machinery is needed. The WaitGroup plus a closer goroutine is the standard way to close a channel that several workers write to: no worker can close it, because the others may still be sending (fan-out and fan-in). The hand-written sort is only there to make the output stable; in real code use slices.SortFunc, and sort by the index you already carry.

The price of this style is that every stage must pass errored items through, and a stage that forgets turns a bad item into a hole. Keep that rule in one place by having a small helper wrap each stage's function so errored items skip the work.

A deadline for the whole pipeline

Errors are not the only reason to stop. A pipeline over a slow source needs a time limit too, and the context is the right carrier. Wrap the errgroup's context with a timeout and the same select statements that handle a failure handle the deadline, with no new code in the stages:

package main

import (
	"context"
	"errors"
	"fmt"
	"time"

	"golang.org/x/sync/errgroup"
)

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Millisecond)
	defer cancel()
	g, ctx := errgroup.WithContext(ctx)

	ticks := make(chan int)
	g.Go(func() error {
		defer close(ticks)
		for i := 0; ; i++ {
			select {
			case ticks <- i:
				time.Sleep(20 * time.Millisecond)
			case <-ctx.Done():
				return ctx.Err()
			}
		}
	})

	count := 0
	g.Go(func() error {
		for range ticks {
			count++
		}
		return nil
	})

	err := g.Wait()
	fmt.Println(errors.Is(err, context.DeadlineExceeded), count >= 2)
}

true true

The producer would run forever, and the deadline ends it after three or so ticks. Wait returns an error that matches context.DeadlineExceeded, which a caller can map to a timeout response (context with timeout and deadline). Notice that the first error to arrive is the context's own. If a stage fails at the same moment, either may win, so treat both as "the pipeline stopped" and do not depend on which one is reported.

Keeping the order

Parallel workers return results in whatever order they finish. If the output must match the input, the cheapest honest method is to carry the position with each item, as the collecting example does with index, and put the result into a pre-sized slice at that position. Reordering through a priority queue or a resequencing stage is possible and rarely worth the complexity. When the order matters only for the error report, sorting the errors by index at the end, as above, is enough. When the downstream consumer needs ordered values as they arrive, a single worker per stage preserves order for free, at the price of parallelism; that is often the right trade for a stage that is cheap relative to the I/O around it.

Backpressure is a feature

An unbuffered channel makes a fast producer wait for a slow consumer, which is the behavior you want: memory use stays flat, and a stalled consumer stalls the pipeline instead of filling a queue. Adding a large buffer to "speed things up" removes that property and hides errors: the producer races ahead, reads a million records, and only then does the consumer fail on the first. The work was wasted and the memory spent. If you buffer, make the buffer small, sized to smooth out jitter between stages and not to hold the data set, and remember that after a failure everything in the buffers is discarded or must be drained.

Bounding parallelism

A stage that starts one goroutine per item is a bug on a large input. Use a fixed number of workers reading from the shared channel, as the collecting example does, or errgroup.SetLimit(n) to cap concurrent goroutines started with g.Go (the SetLimit article). With a limit, g.Go blocks when the group is full, which is another form of backpressure and another place a cancelled context must be checked: a producer blocked in g.Go while the pipeline has already failed will start more work than needed unless it tests ctx.Err() first.

Panics inside a stage

A panic in any stage's goroutine ends the process, since nothing upstream can recover it. If the stages run code you do not fully control, or a library with a history of panics on odd input, add a deferred recover in each stage that converts the panic into the stage's error, so it flows through the same machinery as any other failure (goroutine panics). Keep the conversion at the stage boundary; a pipeline whose inner functions recover on their own leaves state half-updated and hides the bug.

Knowing which stage failed

When a pipeline has five stages, "context deadline exceeded" is not enough. Wrap each stage's error with its name, as the parse stage does above, and the final message says where the failure began. For a collecting pipeline, record the stage on the item's error as it passes, so the report reads line 3: parse: invalid syntax or line 9: enrich: lookup timed out. A per-stage counter of items in, items out and errors gives you the shape of a run at a glance: a stage whose output count is far below its input count is where records are disappearing, and one with zero output and many errors is the one that is broken. These counters cost almost nothing and are the first thing an operator asks for when a nightly job produces half the usual data.

errgroup or hand-rolled

The errgroup version is shorter and handles the first-error and cancellation bookkeeping, so it should be the default for fail-fast work. The hand-rolled collecting version is not an errgroup case at all, since errgroup returns one error and the whole point there is to keep every one. Don't force one into the other's shape. A hybrid that works well in practice runs the stages in an errgroup for lifecycle and cancellation, while the items carry their own errors for the per-record outcomes. Stage-level failures, such as the database going away, return from the stage function and cancel everything; record-level failures, such as one malformed line, travel as data. Drawing that line explicitly in the design is what keeps the error handling understandable: infrastructure failure stops the run, and data failure is reported at the end.

A checklist for review

When you review a pipeline, a short list finds most defects.

1. Who closes each channel, and does that close happen on every path out of the goroutine?
2. Can every send and receive give up when the context is cancelled?
3. What happens to items already in flight when a stage fails?
4. Is parallelism bounded, and where?
5. Does the first error returned say which stage and which item?
6. Is there a test where each stage fails, with a timeout and a goroutine count?

If you can answer all six from reading the code, the pipeline is probably sound. If one answer is "I'm not sure", that is where the bug will be.

Choosing a policy

Ask what a failure means for the other items. If the whole job is invalid when one piece is, fail fast. If each item stands alone, collect all. There is a middle path that is often right in practice: collect, but stop when the error count passes a threshold, using the same cancellation as fail-fast. A file with one bad line in ten million is data; a file in which every line fails is a wrong file, and there is no point reading it to the end (collecting multiple errors shows the cap).

Test the failure in each stage

A pipeline test that only feeds good data tests the part that never breaks. For each stage, inject a failure and check three things: the pipeline returns the expected error, it returns promptly instead of hanging, and no goroutines are left. A leak check does not need a library:

package main

import (
	"context"
	"errors"
	"runtime"
	"testing"
	"time"

	"golang.org/x/sync/errgroup"
)

func runPipeline(failAt int) error {
	g, ctx := errgroup.WithContext(context.Background())
	nums := make(chan int)

	g.Go(func() error {
		defer close(nums)
		for i := 0; i < 1000; i++ {
			select {
			case nums <- i:
			case <-ctx.Done():
				return ctx.Err()
			}
		}
		return nil
	})
	g.Go(func() error {
		for n := range nums {
			if n == failAt {
				return errors.New("stage failed")
			}
		}
		return nil
	})
	return g.Wait()
}

func TestFailureStopsPipelineWithoutLeaks(t *testing.T) {
	before := runtime.NumGoroutine()

	done := make(chan error, 1)
	go func() { done <- runPipeline(5) }()

	select {
	case err := <-done:
		if err == nil || err.Error() != "stage failed" {
			t.Fatalf("err = %v, want stage failed", err)
		}
	case <-time.After(2 * time.Second):
		t.Fatal("pipeline did not stop after a stage failed")
	}

	time.Sleep(50 * time.Millisecond)
	if after := runtime.NumGoroutine(); after > before {
		t.Errorf("goroutines: %d before, %d after", before, after)
	}
}

=== RUN   TestFailureStopsPipelineWithoutLeaks
--- PASS: TestFailureStopsPipelineWithoutLeaks (0.05s)
PASS
ok  	example.test/lab	1.298s

The timeout turns a hang into a clear failure. The goroutine count catches the leak. If the consumer's early return did not cancel the producer, the second check would fail, and the message would say how many goroutines were left. For larger suites, a leak detector gives better diagnostics, but this is enough to hold the line. The behaviors worth testing are listed in testing concurrent code. The Go blog's Pipelines and cancellation is the original treatment of the pattern, and the errgroup documentation describes how the first error and the context interact.