Fan-In / Fan-Out Inside a Pipeline — Middle Level¶

Table of Contents¶

1. Introduction
2. Prerequisites
3. Glossary
4. Core Concepts
5. Production-Grade Merges
6. Ordered Merge
7. Backpressure in Practice
8. Error Handling at Scale
9. Real-World Analogies
10. Mental Models
11. Pros & Cons
12. Use Cases
13. Code Examples
14. Coding Patterns
15. Clean Code
16. Product Use / Feature
17. Performance Tips
18. Best Practices
19. Edge Cases & Pitfalls
20. Common Mistakes
21. Common Misconceptions
22. Tricky Points
23. Test
24. Tricky Questions
25. Cheat Sheet
26. Self-Assessment Checklist
27. Summary
28. What You Can Build
29. Further Reading
30. Related Topics
31. Diagrams & Visual Aids

Introduction¶

Focus: "How do real production fan-out / fan-in pipelines look? How do I preserve order? How do I propagate errors? When does fan-out actually help?"

At the junior level you learned the mechanics: workers, merges, closers, and the rules about who closes what. At the middle level you build pipelines that ship — that survive partial failure, slow downstreams, ordering requirements, and the inevitable moment when someone wants to add metrics.

The middle level is about three things:

1. Production-grade structure. Real merges are wrapped in stages that own their own goroutines, expose <-chan T ports, and accept context.Context. The shape gets formal: every stage is a function func(ctx, in) (out, err) or func(ctx, in) out for infallible stages.
2. Ordering. Many real applications need the output in the same order as the input. The plain merge does not give you that. You need a sequence-number tag, a small reorder buffer, or a per-worker output that the consumer round-robins through.
3. Behaviour under failure. What happens when one worker errors? When the consumer dies? When the context is cancelled mid-flight? The patterns are well-known, and middle-level Go is where they become reflexes.

We will also tighten the discussion of backpressure — when to buffer, how much, and what buffering costs. And we will introduce a few new merge variants: ordered merge by sequence number, weighted merge by source priority, and a merge with a per-input timeout that drops slow sources.

After this file you should be able to:

• Build a multi-stage pipeline with fan-out at every stage that requires parallelism.
• Implement an ordered merge with a small reorder buffer.
• Use errgroup.Group for fan-out with first-error cancellation.
• Tune buffer sizes based on the consumer's rate.
• Detect goroutine leaks in CI using goleak or a manual runtime.NumGoroutine check.
• Pick the right fan-out factor based on whether work is CPU- or I/O-bound.

We will not yet cover reflect.Select for dynamic merges or the runtime internals of select. Those are senior and professional concerns.

Prerequisites¶

• Required: The junior file's content — canonical merge, worker pools, ownership rules, the closer pattern.
• Required: Solid context.Context use. You can WithCancel, WithTimeout, WithDeadline, and you select on ctx.Done() in every blocking operation.
• Required: You have written tests that use go test -race and you know how to interpret the output.
• Required: Comfort with errgroup.Group from golang.org/x/sync/errgroup. If not, read the docs first; it is a 50-line file you can read in five minutes.
• Helpful: Some exposure to container/heap. Ordered merge can be built on a min-heap.
• Helpful: You have profiled at least one Go program with pprof and seen the goroutine profile.

Glossary¶

Core Concepts¶

A stage is a typed function¶

A production pipeline is composed of named stages, each shaped like:

func stage(ctx context.Context, in <-chan In) <-chan Out

or for stages that can fail:

func stage(ctx context.Context, in <-chan In) (<-chan Out, <-chan error)

or, more commonly, the stage wraps each value with an error:

type Result[T any] struct {
    Value T
    Err   error
}
func stage(ctx context.Context, in <-chan In) <-chan Result[Out]

Each stage's contract is: it spawns its own goroutines, closes its own output channel(s), and respects ctx.Done() for cancellation.

A pipeline is just a sequence of stage calls:

in := source(ctx)
parsed := parse(ctx, in)
hashed := hash(ctx, parsed)
out := mergeResults(ctx, fanOut(ctx, hashed, 8)...)
for r := range out {
    consume(r)
}

Fan-out is one of those stages; fan-in is another. They compose like any other transformer.

Order is a separate concern from parallelism¶

A frequent design mistake: "I want this pipeline to be parallel and in-order, so I will use a single-worker pipeline." But that is sequential, not in-order parallel. The right structure is:

1. Tag every input with a sequence number.
2. Fan out the work in parallel (order lost).
3. Reorder by sequence number at the end.

The reorder buffer is a small heap keyed on sequence number. Out-of-order arrivals are held; in-order arrivals (those matching the expected next sequence) are emitted immediately. The buffer typically stays small if the workers' completion times do not vary wildly.

This is the canonical pattern for "parallel and ordered" pipelines.

Bounded fan-out is the default¶

Junior-level fan-out is often "spawn one worker per input." That is unbounded fan-out. It works for small inputs and breaks for large ones — memory bloats, scheduler thrashes, downstream services get rate-limited.

Production fan-out is bounded: a fixed number of workers (8, 64, or NumCPU depending on the work) sharing a single input channel. The workers compete for items; whoever is free takes the next one.

For CPU-bound work, bound to NumCPU or slightly less.

For I/O-bound work, bound to whatever the downstream can sustain: connection pool size, rate limit budget, etc.

Almost never spawn len(input) goroutines.

errgroup.Group is the production primitive¶

For fan-out with error handling, errgroup.Group from golang.org/x/sync/errgroup is the workhorse. The pattern:

g, ctx := errgroup.WithContext(ctx)
for i := 0; i < N; i++ {
    g.Go(func() error {
        return doWork(ctx)
    })
}
if err := g.Wait(); err != nil {
    return fmt.Errorf("pipeline: %w", err)
}

Properties:

• First worker error cancels the group's context.
• Other workers see ctx.Done() and exit (if they are well-written).
• g.Wait() returns the first error or nil.
• Built-in bounded variant: g.SetLimit(N) caps concurrent Go calls.

For fan-out / fan-in with errors, errgroup is almost always the right starting point.

Backpressure works on a rate, not on a count¶

People think of buffers as "how many items can sit in the channel." The more useful view: a buffer is a time window — the producer can run for buffer-size / rate seconds before backpressure kicks in.

If your producer runs at 1000 items/sec and you buffer 100, the producer can run for 0.1 seconds without consumer feedback. If your producer's bursty maximum is 5000 items/sec for 50 ms, a buffer of 250 absorbs the burst.

Sizing buffers is an exercise in matching producer burstiness to consumer mean throughput. Default to unbuffered (no smoothing) and add buffer only when you see consumer-side variability that you want to absorb.

Cancellation paths must converge¶

In a multi-stage pipeline, cancellation can come from:

• The caller (cancel() on the parent context).
• An error from one worker (via errgroup).
• A timeout (WithTimeout).
• A panic recovered locally.

Whichever happens first, the cancellation needs to reach every stage. The way: pass the same context (or a derived one) to every stage; have every stage select on <-ctx.Done(). If any stage forgets, that stage leaks.

A test that runs the pipeline, cancels, and then checks runtime.NumGoroutine is the easiest leak-detection tool. Even better: goleak.VerifyNone(t) at the end of every test.

Production-Grade Merges¶

A merge that respects context.Context¶

The starting point for everything in this file:

func merge[T any](ctx context.Context, cs ...<-chan T) <-chan T {
    out := make(chan T)
    var wg sync.WaitGroup
    wg.Add(len(cs))
    for _, c := range cs {
        c := c
        go func() {
            defer wg.Done()
            for v := range c {
                select {
                case out <- v:
                case <-ctx.Done():
                    return
                }
            }
        }()
    }
    go func() {
        wg.Wait()
        close(out)
    }()
    return out
}

This is the standard. Memorise it. The senior variant uses reflect.Select for dynamic input lists; the professional variant studies its scheduling cost. The middle-level version is this 18 lines.

A merge that respects buffer pressure on the input side¶

Sometimes the inputs themselves are slow. A forwarder that selects on both <-c and <-ctx.Done() simultaneously:

for {
    select {
    case v, ok := <-c:
        if !ok {
            return
        }
        select {
        case out <- v:
        case <-ctx.Done():
            return
        }
    case <-ctx.Done():
        return
    }
}

Outer select waits for either a value from the input or cancellation. Inner select forwards the value or notices cancellation mid-send. This is slightly more code than for v := range c { ... }, but it cancels faster when both the input and the output are slow.

A merge with per-input timeouts¶

If one source is very slow, you may want to drop it after a timeout:

func mergeTimeout[T any](ctx context.Context, perInputTimeout time.Duration, cs ...<-chan T) <-chan T {
    out := make(chan T)
    var wg sync.WaitGroup
    wg.Add(len(cs))
    for _, c := range cs {
        c := c
        go func() {
            defer wg.Done()
            t := time.NewTimer(perInputTimeout)
            defer t.Stop()
            for {
                if !t.Stop() {
                    select { case <-t.C: default: }
                }
                t.Reset(perInputTimeout)
                select {
                case v, ok := <-c:
                    if !ok {
                        return
                    }
                    select {
                    case out <- v:
                    case <-ctx.Done():
                        return
                    }
                case <-t.C:
                    return // slow input dropped
                case <-ctx.Done():
                    return
                }
            }
        }()
    }
    go func() {
        wg.Wait()
        close(out)
    }()
    return out
}

The forwarder resets a timer between values. If a value does not arrive within perInputTimeout, the forwarder returns, effectively dropping that input. Useful for hedged requests where one of N replicas is allowed to be slow.

A merge that preserves a count of remaining inputs¶

Sometimes you want to know how many inputs are still live (for metrics). Maintain a counter using atomic:

func mergeWithLive[T any](ctx context.Context, live *atomic.Int64, cs ...<-chan T) <-chan T {
    out := make(chan T)
    var wg sync.WaitGroup
    wg.Add(len(cs))
    live.Store(int64(len(cs)))
    for _, c := range cs {
        c := c
        go func() {
            defer wg.Done()
            defer live.Add(-1)
            for v := range c {
                select {
                case out <- v:
                case <-ctx.Done():
                    return
                }
            }
        }()
    }
    go func() {
        wg.Wait()
        close(out)
    }()
    return out
}

The metric live.Load() tells you how many inputs are still producing. Useful for dashboards.

Ordered Merge¶

This is the section that often surprises middle-level engineers: ordered merge is not provided by the standard library, and the patterns are subtle.

The problem¶

You have a fan-out of workers. The input is ordered (rows 1, 2, 3, ...). The output should be in the same order, but the workers process at different speeds and finish out of order.

Naive merge produces 1, 3, 2, 5, 4, .... You need 1, 2, 3, 4, 5, ....

Solution 1: Tag-and-reorder¶

Tag each input with its sequence number. The workers preserve the tag. A reorder stage holds out-of-order results until the next expected sequence arrives.

type Tagged[T any] struct {
    Seq int
    Val T
}

func tag[T any](ctx context.Context, in <-chan T) <-chan Tagged[T] {
    out := make(chan Tagged[T])
    go func() {
        defer close(out)
        n := 0
        for v := range in {
            select {
            case out <- Tagged[T]{Seq: n, Val: v}:
                n++
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

func reorder[T any](ctx context.Context, in <-chan Tagged[T]) <-chan T {
    out := make(chan T)
    go func() {
        defer close(out)
        next := 0
        pending := make(map[int]T)
        for r := range in {
            pending[r.Seq] = r.Val
            for {
                v, ok := pending[next]
                if !ok {
                    break
                }
                select {
                case out <- v:
                case <-ctx.Done():
                    return
                }
                delete(pending, next)
                next++
            }
        }
    }()
    return out
}

Pipeline:

tagged := tag(ctx, source)
results := pool(ctx, tagged, N, work)
ordered := reorder(ctx, results)

The reorder stage uses a map indexed by sequence number. When result with seq = next arrives, it is emitted; if subsequent sequences are already in the map, they are emitted too. Out-of-order arrivals sit in the map.

Solution 2: Heap-based reorder¶

If sequence numbers may be sparse or unbounded, use a min-heap instead of a map:

import "container/heap"

type item[T any] struct {
    Seq int
    Val T
}

type minHeap[T any] []item[T]

func (h minHeap[T]) Len() int            { return len(h) }
func (h minHeap[T]) Less(i, j int) bool  { return h[i].Seq < h[j].Seq }
func (h minHeap[T]) Swap(i, j int)       { h[i], h[j] = h[j], h[i] }
func (h *minHeap[T]) Push(x any)         { *h = append(*h, x.(item[T])) }
func (h *minHeap[T]) Pop() any           { old := *h; n := len(old); x := old[n-1]; *h = old[:n-1]; return x }

func reorderHeap[T any](ctx context.Context, in <-chan Tagged[T]) <-chan T {
    out := make(chan T)
    go func() {
        defer close(out)
        next := 0
        h := &minHeap[T]{}
        for r := range in {
            heap.Push(h, item[T]{Seq: r.Seq, Val: r.Val})
            for h.Len() > 0 && (*h)[0].Seq == next {
                top := heap.Pop(h).(item[T])
                select {
                case out <- top.Val:
                case <-ctx.Done():
                    return
                }
                next++
            }
        }
    }()
    return out
}

Same idea, different data structure. Heap is more efficient when the queue grows large or the sequence numbers are sparse.

Solution 3: Per-worker output, round-robin merge¶

Pre-assign each input to a specific worker by index. Each worker writes to its own output channel. The consumer round-robins through the worker outputs in input order.

func roundRobinMerge[T any](ctx context.Context, cs []<-chan T) <-chan T {
    out := make(chan T)
    go func() {
        defer close(out)
        n := len(cs)
        for i := 0; ; i = (i + 1) % n {
            v, ok := <-cs[i]
            if !ok {
                return // one stream done; in practice you handle more carefully
            }
            select {
            case out <- v:
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

The catch: round-robin requires each worker to receive items in input order. If worker 0 gets inputs 0, 3, 6, 9, ..., and worker 1 gets 1, 4, 7, 10, ..., and worker 2 gets 2, 5, 8, 11, ..., then round-robin reads in order. The dispatcher must do the assignment, not the runtime.

This works well when work per input is uniform. When it is skewed (some inputs take much longer), round-robin starves: a slow worker holds up the merge.

Solution 4: Strict in-order pipeline (no parallelism)¶

Sometimes the cleanest answer is: do not parallelise. A single-goroutine transformer preserves order automatically. If the per-item work is small or the parallelism gain is marginal, sequential is simpler and provably correct.

A useful question: how much speedup do you need? If 2x is enough, two workers + a tiny reorder buffer is the right answer. If 10x, full fan-out with sequence tagging. If 1.1x, do not bother.

Choosing among the solutions¶

Most production pipelines use tag-and-reorder with a heap. It is the most flexible.

Backpressure in Practice¶

Buffers are time windows¶

Repeating the core idea: a buffered channel of size n lets the producer run for n / consumerRate seconds without consumer feedback.

For a producer at 1000/sec and consumer at 100/sec, buffer = 1000 means the producer can run for 10 seconds before stalling. After that, you accumulate at the consumer's pace.

Choosing buffer:

• Buffer = 0 (unbuffered): producer is exactly matched to consumer. Maximum backpressure. Minimum latency from send to receive.
• Buffer = small (1-10): absorbs scheduler jitter. Tiny latency improvement.
• Buffer = medium (100-1000): absorbs producer bursts. Adds latency.
• Buffer = huge (10 000+): masks the consumer slowness, possibly catastrophically. Avoid.

In benchmarks, a buffer of 1 often gives the best throughput per latency. A buffer of 0 gives the best latency. Larger buffers help only when producer is genuinely bursty.

Backpressure under fan-out¶

In a fan-out, "producer" is the source channel and "consumer" is the worker pool. If the workers can drain the source at rate N * w (where N is worker count and w is per-worker rate), the source can be unbuffered.

If workers vary in speed, a small buffer (size 1 or 2) absorbs the variance. Workers compete for items; a fast worker takes more than a slow one.

A common shape:

source := producer(ctx)             // unbuffered
buffered := buffer(ctx, source, 4)  // small smoothing
workers := pool(ctx, buffered, 8, fn)

The buffer stage is just a one-goroutine forwarder with a buffered output channel. It smooths the source so the workers do not idle waiting for the producer's next batch.

Backpressure under fan-in¶

In a fan-in, "producer" is each worker and "consumer" is the merged channel's reader. If the reader is fast, workers do not block; their items flow through the merge with minimal delay.

If the reader is slow, all workers block on out <- v inside the forwarder. They stop reading their input channels. The fan-out source backs up. Everything cascades.

Buffering the merged channel slightly (size = N workers, perhaps) lets each worker push one in-flight value into the buffer without blocking. This is a common micro-optimisation; larger buffers do not help and can hide problems.

When to use a select-with-default drop¶

If you have an attempt-once-then-drop semantics (e.g., metrics, sampling), use select with default:

select {
case metrics <- v:
default:
    // dropped (metrics buffer full)
}

This is not backpressure. It is loss. It is fine for non-critical telemetry; it is bad for primary data flow.

When to use a bounded queue with a size limit¶

If you must drop new items when overloaded (so old items get processed), drop-on-full is right. If you must drop old items so new ones flow (real-time data), use a ring buffer that overwrites. Channels do neither natively; you build them.

Most production pipelines do not drop. They block. Backpressure to the source is preferable to lossy buffers in 90% of cases. The exceptions: real-time analytics, real-time gaming, and high-rate telemetry.

Error Handling at Scale¶

The fan-out / errgroup pattern¶

The standard recipe:

func process(ctx context.Context, items []Item) error {
    g, ctx := errgroup.WithContext(ctx)
    g.SetLimit(8) // bounded fan-out
    for _, it := range items {
        it := it
        g.Go(func() error {
            return processOne(ctx, it)
        })
    }
    return g.Wait()
}

Properties:

• 8 workers max in flight.
• First error cancels the group context.
• Other workers see ctx.Done() and exit (each processOne must respect ctx).
• g.Wait() returns the first error.

The trade-off: you do not see other errors after the first. If you want all errors, accumulate them yourself with a mutex.

Continue-on-error pattern¶

If you want to keep processing despite individual failures:

type Result struct {
    Item Item
    Err  error
}

func processAll(ctx context.Context, items []Item) []Result {
    in := make(chan Item)
    out := make(chan Result)
    var wg sync.WaitGroup
    for i := 0; i < 8; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for it := range in {
                err := processOne(ctx, it)
                select {
                case out <- Result{Item: it, Err: err}:
                case <-ctx.Done():
                    return
                }
            }
        }()
    }
    go func() {
        defer close(in)
        for _, it := range items {
            select {
            case in <- it:
            case <-ctx.Done():
                return
            }
        }
    }()
    go func() {
        wg.Wait()
        close(out)
    }()
    var results []Result
    for r := range out {
        results = append(results, r)
    }
    return results
}

Every item gets a Result with either success or failure. Caller examines.

Error from the merge itself¶

The merge function does not usually fail — it just forwards. If a forwarder gets a value that contains an error (e.g., the Result[T] type), it forwards it like any other value. The consumer is responsible for inspecting.

Some teams pre-check: at the merge, drop errored values into a separate channel, forward only successes. This complicates the API and is usually not worth it.

Panic isolation¶

A panicking worker kills the program unless caught. Wrap each worker's body in recover:

go func() {
    defer wg.Done()
    defer func() {
        if r := recover(); r != nil {
            log.Printf("worker panic: %v\n%s", r, debug.Stack())
            // optionally send a sentinel error
        }
    }()
    for v := range in {
        out <- process(v)
    }
}()

For production fan-out, panic recovery is non-negotiable. One malformed input should not bring down the service.

Real-World Analogies¶

Ordered merge is an ER triage system¶

Patients arrive and are tagged with arrival time. Doctors process them in parallel but at different speeds — some cases are quick, others take an hour. Discharges happen in arrival order (insurance reasons), so the receptionist holds completed paperwork until the next-arrived patient's paperwork is ready. The "next" pointer advances as patients are discharged. That is tag-and-reorder.

Backpressure is a thermostat¶

A thermostat regulates the heating system based on temperature. If the room is cold, it turns the heat up; if hot, down. The room temperature feeds back to the controller, which adjusts the source rate. A Go channel is exactly that: the consumer's reading rate feeds back to the producer's sending rate via the channel's fill state. No central controller — the negotiation is implicit.

errgroup is a fire drill¶

Each worker is a person. When one person spots a fire and shouts (returns an error), everyone evacuates (context cancels). The fire team (the g.Wait caller) is told what happened.

Bounded fan-out is a security checkpoint¶

The airport has a fixed number of metal detectors. Travellers (input items) line up; whichever detector is free serves them. Throughput is bounded by the number of detectors. Adding detectors helps until you saturate the conveyor belt of bags (the bottleneck).

Mental Models¶

Model 1: "Pipelines are functions composed over channels"¶

Each stage is a function from <-chan A to <-chan B. Composition is left-to-right:

source >>= stage1 >>= stage2 >>= sink

You can refactor by inserting stages without changing the rest. A metrics stage, a logging stage, a filter — they all fit the same shape.

Model 2: "Order is a property added by ceremony"¶

By default, fan-out destroys order. To get it back, you must do something — usually tag-and-reorder. Order is not free; it costs a reorder buffer and a small amount of latency.

Model 3: "Errors flow like data — but cancellation flows in reverse"¶

Data flows downstream: producer to consumer. Errors flow with the data (as part of the value or via a parallel channel). Cancellation flows upstream: the consumer or supervisor cancels ctx, and every stage observes.

Picture a river: water flows down; the alarm to stop the source goes up.

Model 4: "Throughput is the slowest stage, latency is the sum of stages"¶

A pipeline's throughput equals the throughput of its slowest stage. A pipeline's per-item latency equals the time to traverse every stage. Fan-out at the slowest stage increases throughput. Fan-out at a fast stage adds latency for no gain.

Model 5: "Cancellation is a deadline, not a kill switch"¶

ctx.Done() is a request to stop. It is the stages' responsibility to honour it. If a stage holds a lock or is mid-syscall, cancellation has to wait until the next select. Production pipelines minimise the maximum delay between cancel() and full shutdown.

Pros & Cons¶

Pros¶

• Compositional. Stages are functions; pipelines are sequences of function calls. Adding logging, metrics, or buffering is a one-line insert.
• Type-safe. With generics, each stage's input and output types are checked at compile time.
• Backpressure for free. Channels self-regulate.
• Cancellation is uniform. One context.Context for the whole pipeline.
• Test-friendly. Each stage is a function; you can test it with a fake input channel and inspect the output channel.

Cons¶

• Reorder buffers add latency. Tag-and-reorder costs a small but real delay; the slowest worker pins the next-emit pointer.
• Buffering hides problems. Large buffers turn fast feedback into slow feedback. Backpressure becomes invisible.
• errgroup returns only the first error. If you need all errors, you must collect them separately.
• Stages must agree on context. A stage that ignores ctx.Done() leaks. Easy to forget on a quick refactor.
• Multi-stage pipelines have many goroutines. A six-stage pipeline with fan-out 8 has roughly 60 goroutines. Debugging requires tooling.

Use Cases¶

Code Examples¶

Example 1: A complete pipeline with metrics¶

package main

import (
    "context"
    "fmt"
    "sync"
    "sync/atomic"
    "time"
)

type stats struct {
    produced, processed, emitted atomic.Int64
}

func produce(ctx context.Context, n int, s *stats) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for i := 0; i < n; i++ {
            select {
            case out <- i:
                s.produced.Add(1)
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

func worker(ctx context.Context, in <-chan int, s *stats) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for v := range in {
            time.Sleep(time.Millisecond) // simulate work
            s.processed.Add(1)
            select {
            case out <- v * v:
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

func merge(ctx context.Context, s *stats, cs ...<-chan int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup
    wg.Add(len(cs))
    for _, c := range cs {
        c := c
        go func() {
            defer wg.Done()
            for v := range c {
                select {
                case out <- v:
                    s.emitted.Add(1)
                case <-ctx.Done():
                    return
                }
            }
        }()
    }
    go func() { wg.Wait(); close(out) }()
    return out
}

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()
    s := &stats{}
    in := produce(ctx, 100, s)
    const N = 4
    outs := make([]<-chan int, N)
    for i := 0; i < N; i++ {
        outs[i] = worker(ctx, in, s)
    }
    merged := merge(ctx, s, outs...)
    sum := 0
    for v := range merged {
        sum += v
    }
    fmt.Printf("produced=%d processed=%d emitted=%d sum=%d\n",
        s.produced.Load(), s.processed.Load(), s.emitted.Load(), sum)
}

The stats struct is shared via atomic counters. Every stage increments its relevant counter. End of run, the totals should match: produced == processed == emitted == 100. If any differ, you have a leak or a drop.

Example 2: Ordered merge with tag-and-reorder¶

package main

import (
    "context"
    "fmt"
    "sync"
    "time"
)

type tagged struct {
    seq int
    val int
}

func source(ctx context.Context, n int) <-chan tagged {
    out := make(chan tagged)
    go func() {
        defer close(out)
        for i := 0; i < n; i++ {
            select {
            case out <- tagged{seq: i, val: i}:
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

func work(ctx context.Context, in <-chan tagged) <-chan tagged {
    out := make(chan tagged)
    go func() {
        defer close(out)
        for t := range in {
            // simulate variable work
            time.Sleep(time.Duration((t.seq*13)%7) * time.Millisecond)
            select {
            case out <- tagged{seq: t.seq, val: t.val * 10}:
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

func merge(ctx context.Context, cs ...<-chan tagged) <-chan tagged {
    out := make(chan tagged)
    var wg sync.WaitGroup
    wg.Add(len(cs))
    for _, c := range cs {
        c := c
        go func() {
            defer wg.Done()
            for v := range c {
                select {
                case out <- v:
                case <-ctx.Done():
                    return
                }
            }
        }()
    }
    go func() { wg.Wait(); close(out) }()
    return out
}

func reorder(ctx context.Context, in <-chan tagged) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        next := 0
        pending := map[int]int{}
        for t := range in {
            pending[t.seq] = t.val
            for {
                v, ok := pending[next]
                if !ok {
                    break
                }
                select {
                case out <- v:
                case <-ctx.Done():
                    return
                }
                delete(pending, next)
                next++
            }
        }
    }()
    return out
}

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()
    in := source(ctx, 20)
    const N = 4
    outs := make([]<-chan tagged, N)
    for i := 0; i < N; i++ {
        outs[i] = work(ctx, in)
    }
    ordered := reorder(ctx, merge(ctx, outs...))
    for v := range ordered {
        fmt.Println(v)
    }
}

Output is 0 10 20 30 ... 190, strictly in order, even though the work stage finishes items out of order.

Example 3: errgroup fan-out¶

package main

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

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

func processItem(ctx context.Context, i int) error {
    select {
    case <-time.After(time.Duration(i*10) * time.Millisecond):
    case <-ctx.Done():
        return ctx.Err()
    }
    if i == 7 {
        return fmt.Errorf("item %d: simulated failure", i)
    }
    return nil
}

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()
    g, gctx := errgroup.WithContext(ctx)
    g.SetLimit(4)
    for i := 0; i < 20; i++ {
        i := i
        g.Go(func() error {
            return processItem(gctx, i)
        })
    }
    err := g.Wait()
    if errors.Is(err, context.Canceled) {
        fmt.Println("cancelled")
        return
    }
    if err != nil {
        fmt.Println("failed:", err)
        return
    }
    fmt.Println("all done")
}

When item 7 returns its error, g's context is cancelled; in-flight items see gctx.Done() and exit; g.Wait() returns item 7's error.

Example 4: A pipeline with goleak¶

package mypipe_test

import (
    "context"
    "testing"

    "go.uber.org/goleak"
)

func TestPipelineNoLeak(t *testing.T) {
    defer goleak.VerifyNone(t)
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()
    runPipeline(ctx)
    cancel()
}

goleak snapshots goroutines at the start and end of the test. If any user goroutine survives, the test fails. The single most useful CI check for pipeline code.

Example 5: Bounded fan-out using errgroup.SetLimit¶

g, ctx := errgroup.WithContext(ctx)
g.SetLimit(8)
for _, url := range urls {
    url := url
    g.Go(func() error {
        return fetch(ctx, url)
    })
}
return g.Wait()

SetLimit(8) ensures at most 8 Go calls are in flight. The 9th call blocks until one finishes. Built-in bounded fan-out.

Example 6: A merge with a small buffer for smoothing¶

func smoothMerge[T any](ctx context.Context, buf int, cs ...<-chan T) <-chan T {
    out := make(chan T, buf)
    var wg sync.WaitGroup
    wg.Add(len(cs))
    for _, c := range cs {
        c := c
        go func() {
            defer wg.Done()
            for v := range c {
                select {
                case out <- v:
                case <-ctx.Done():
                    return
                }
            }
        }()
    }
    go func() { wg.Wait(); close(out) }()
    return out
}

Buffer of 1 or 2 absorbs scheduler jitter. Buffer of 100 hides backpressure. Always justify the value in code comments.

Example 7: Round-robin pre-assignment¶

func roundRobinFanOut[T any](ctx context.Context, in <-chan T, n int) []<-chan T {
    outs := make([]chan T, n)
    for i := range outs {
        outs[i] = make(chan T)
    }
    go func() {
        defer func() {
            for _, o := range outs {
                close(o)
            }
        }()
        i := 0
        for v := range in {
            select {
            case outs[i] <- v:
            case <-ctx.Done():
                return
            }
            i = (i + 1) % n
        }
    }()
    result := make([]<-chan T, n)
    for i, o := range outs {
        result[i] = o
    }
    return result
}

Combined with round-robin merge, this preserves order. Risk: skewed work distributions cause one worker to become a straggler.

Example 8: A pipeline that drops on full instead of blocking¶

func bestEffort[T any](in <-chan T, out chan<- T) {
    for v := range in {
        select {
        case out <- v:
        default:
            // dropped
        }
    }
}

For non-critical paths only. Telemetry, sampling, optional features.

Example 9: A merge that returns the first non-error¶

func firstNonError[T any](ctx context.Context, cs ...<-chan Result[T]) (T, error) {
    var zero T
    for {
        for _, c := range cs {
            select {
            case r, ok := <-c:
                if !ok {
                    continue
                }
                if r.Err == nil {
                    return r.Value, nil
                }
            case <-ctx.Done():
                return zero, ctx.Err()
            }
        }
        if allClosed(cs) {
            return zero, errors.New("all failed")
        }
    }
}

Useful for hedged requests: race N backends, take the first that succeeds.

Example 10: A bounded fan-out worker pool with errgroup and panic recovery¶

func panicSafeWorker(ctx context.Context, item Item) (err error) {
    defer func() {
        if r := recover(); r != nil {
            err = fmt.Errorf("panic: %v", r)
        }
    }()
    return doWork(ctx, item)
}

func processAll(ctx context.Context, items []Item) error {
    g, gctx := errgroup.WithContext(ctx)
    g.SetLimit(8)
    for _, it := range items {
        it := it
        g.Go(func() error {
            return panicSafeWorker(gctx, it)
        })
    }
    return g.Wait()
}

The named return err is set by the deferred recover. A panic becomes an error, which propagates through errgroup.

Coding Patterns¶

Pattern 1: The "stage wrapper" idiom¶

Every stage in your pipeline conforms to:

type Stage[I, O any] func(ctx context.Context, in <-chan I) <-chan O

Combinator:

func pipe[A, B, C any](s1 Stage[A, B], s2 Stage[B, C]) Stage[A, C] {
    return func(ctx context.Context, in <-chan A) <-chan C {
        return s2(ctx, s1(ctx, in))
    }
}

Now you can write pipeline := pipe(parse, hash) and call it. With generics, this is type-safe.

Pattern 2: The "fan-out factor" parameter¶

Always expose the fan-out factor as a parameter:

func processor(ctx context.Context, in <-chan Item, workers int) <-chan Result {
    ...
}

Default in tests is 1 (serial). Default in production is configurable via env or config. Hardcoding 8 or NumCPU is a code smell.

Pattern 3: Per-stage instrumentation¶

Wrap stages with a measuring wrapper:

func measured[I, O any](name string, s Stage[I, O]) Stage[I, O] {
    return func(ctx context.Context, in <-chan I) <-chan O {
        out := s(ctx, in)
        // optional: wrap out with a count-and-time stage
        return out
    }
}

Pluggable, removable, never affects core logic.

Pattern 4: The "drain on cancel" idiom¶

After cancellation, the pipeline may have in-flight values. To complete shutdown:

cancel()
for range out {
    // drain
}

With every blocking send guarded by <-ctx.Done(), this drain runs quickly. The pattern guarantees no leaked goroutines.

Pattern 5: The "shutdown hook" idiom¶

Wrap the whole pipeline in a function that owns the context:

func runPipeline(ctx context.Context, in []Item) (results []Result, err error) {
    ctx, cancel := context.WithCancel(ctx)
    defer cancel()
    ...
}

Caller passes a parent context; the function creates a child it controls. If the caller cancels, the child cancels. If the function returns, the child is cancelled.

Clean Code¶

• Each stage is a function. Not a struct, not a method on a service, not embedded in a larger orchestrator. A free function. Stages compose.
• Stages take context.Context first. Always. Even if the stage does not use it today, take it.
• Stages return <-chan T. Receive-only. Hide the bidirectional channel.
• No nested closures longer than 20 lines. Extract a function.
• Channel names describe the data, not the shape. out, in, jobs, results, errors — not ch1, ch2.
• Buffer sizes are documented. make(chan T, 32 /* one batch of records */).
• Each goroutine has a comment on entry: who owns it, who exits it. Especially important for pipelines.

Product Use / Feature¶

Performance Tips¶

Match fan-out to the bottleneck¶

Profile with pprof. If CPU is saturated, more workers do nothing (CPU is the bottleneck). If CPU is idle and latency is from I/O, more workers help.

A rule of thumb:

• CPU-bound: workers = NumCPU (or NumCPU - 1 to leave space for the merge).
• I/O-bound (HTTP, DB): workers = NumCPU * (1 + waitTime / cpuTime). For HTTP with 100 ms wait and 1 ms CPU per request, workers = NumCPU * 101. Practically: 32-256.
• Mixed: profile and tune.

Avoid one-channel-op per tiny item¶

If your per-item work takes < 1 microsecond, the channel ops (~100-200 ns each) become significant. Batch into slices.

Use sync.Pool for per-worker buffers¶

If each worker needs a temporary buffer (e.g., decoding into a byte slice), use sync.Pool to reuse:

var bufPool = sync.Pool{New: func() any { return make([]byte, 0, 4096) }}

func worker(in <-chan task) <-chan result {
    out := make(chan result)
    go func() {
        defer close(out)
        for t := range in {
            buf := bufPool.Get().([]byte)
            // use buf
            bufPool.Put(buf[:0])
            out <- result{...}
        }
    }()
    return out
}

Cuts allocations dramatically.

Watch for false sharing¶

Per-worker atomic counters in adjacent memory cache lines cause false sharing. Pad them or use sync.Map for sparse counters.

Profile the merge¶

The merge is often invisible until it becomes the bottleneck. If the merged channel is full most of the time, the consumer is slow. If the worker outputs are full most of the time, the merge is slow (rare but possible at very high item rates).

Best Practices¶

1. Stages take context.Context; every blocking op selects on <-ctx.Done().
2. Bound fan-out with a fixed worker count or errgroup.SetLimit.
3. Tag-and-reorder when order matters. Do not single-thread for ordering.
4. Use goleak.VerifyNone(t) in tests for pipelines.
5. Recover panics inside each worker.
6. Document buffer sizes with a comment explaining the chosen value.
7. Expose fan-out factor as a parameter, not a constant.
8. Use errgroup for fan-out with first-error semantics.
9. Use a Result[T] channel for continue-on-error semantics.
10. Test with -race and goleak.

Edge Cases & Pitfalls¶

A worker panics and the others continue without it¶

Without per-worker recover, the panic kills the program. With recover, the worker dies and the merge sees its output channel close. The pipeline continues with fewer workers — but no one notices. Solution: log the panic loudly, increment a metric.

errgroup.Wait deadlocks because a worker is stuck¶

A worker that does not select on ctx.Done() is stuck even after the group cancels. Wait() waits forever. Solution: every worker must respect ctx.

Reorder buffer grows unboundedly¶

If the slow worker keeps getting low sequence numbers and the fast workers keep producing high sequence numbers, the reorder buffer can grow. Bound it with a max size; over the bound, log a warning and stall.

Order matters but fan-out is round-robin and inputs vary in size¶

Round-robin sends item i to worker i mod N. If items vary wildly in size, one worker becomes a straggler. The merge waits for that straggler. Solution: switch to tag-and-reorder, where any worker can handle any item.

Merge gets zero inputs¶

merge() with no arguments. The closer goroutine's wg.Wait() returns immediately (counter is 0). close(out) runs. The caller sees a closed empty stream. Correct.

Cancellation does not propagate to a goroutine spawned inside a worker¶

If a worker spawns its own goroutines, those need the context too. Pass ctx recursively.

A buffered channel hides a stuck consumer¶

Producer pushes 1000 items into a buffered channel; buffer holds them; consumer is dead. The producer never blocks. The buffer's items are leaked. Solution: small buffers + monitoring of consumer health.

Worker reads from a closed input channel¶

for v := range in { ... } handles this — the loop exits when in is closed. But if the worker uses <-in directly, check ok:

v, ok := <-in
if !ok { return }

Common Mistakes¶

Common Misconceptions¶

"errgroup is just sync.WaitGroup with errors." — Partly. It also has a context.Context that cancels on first error, which is the main feature.

"Tag-and-reorder is too expensive." — Usually negligible. A small map or heap per pipeline. The bottleneck is almost always the workers, not the reorder buffer.

"More buffer means more throughput." — No. Throughput is bounded by the slowest stage. Buffer absorbs jitter, not slowness.

"context.WithTimeout will stop everything when the timer fires." — It closes Done(). Stages must select on it. If a stage holds a long-running syscall, the cancellation waits.

"Fan-out always speeds things up." — Only when the bottleneck is at that stage. Fan-out at a fast stage adds overhead.

"Round-robin merge preserves order." — Only if the dispatcher pre-assigned items in input order. The runtime's automatic load balancing breaks ordering.

Tricky Points¶

errgroup.Wait returns the first error, not all errors¶

If three workers fail, Wait returns one of their errors (the first to return). The other two are dropped. To collect all, build your own:

type multiErr struct{ errs []error }
var mu sync.Mutex
var errs []error
...
g.Go(func() error {
    if err := work(); err != nil {
        mu.Lock()
        errs = append(errs, err)
        mu.Unlock()
        return err // still triggers cancellation
    }
    return nil
})

g.SetLimit(0) is invalid¶

It panics. Use a positive number. To disable the limit, do not call SetLimit at all.

g.SetLimit(-1) disables the limit¶

A negative argument removes any previous limit. Documented in the source but easy to miss.

errgroup.WithContext returns the derived context¶

g, ctx := errgroup.WithContext(parent)