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

Table of Contents¶

1. Introduction
2. Prerequisites
3. Glossary
4. Architecture
5. Dynamic Channel Sets with reflect.Select
6. Weighted and Priority Fan-Out
7. Partial Failure and Supervisor Patterns
8. Backpressure Engineering
9. Leak Prevention at the Architectural Level
10. Real-World Analogies
11. Mental Models
12. Pros & Cons
13. Use Cases
14. Code Examples
15. Coding Patterns
16. Clean Code
17. Product Use / Feature
18. Performance Tips
19. Best Practices
20. Edge Cases & Pitfalls
21. Common Mistakes
22. Common Misconceptions
23. Tricky Points
24. Test
25. Tricky Questions
26. Cheat Sheet
27. Self-Assessment Checklist
28. Summary
29. What You Can Build
30. Further Reading
31. Related Topics
32. Diagrams & Visual Aids

Introduction¶

Focus: "When and how do I redesign fan-out / fan-in for architecture-scale problems — dynamic stages, weighted streams, partial failure, supervisor restart, and the corners of reflect.Select?"

Senior-level work on fan-out / fan-in moves from "I can write a merge" to "I am responsible for a multi-team pipeline that processes a billion items per day." The mechanics from junior and middle remain; what changes is the scope. You make decisions about:

• Stage boundaries. Where does one team's pipeline end and another's begin? What is the contract on those boundaries?
• Failure modes. A single bad item should not bring down a billion-item pipeline. How is bad-item isolation built?
• Resource governance. Memory, CPU, network bandwidth, downstream rate limits — all need explicit caps.
• Observability. With dozens of stages and hundreds of goroutines, debugging is impossible without metrics, traces, and structured logs.
• Restartability. Pipelines must be safely restartable after a crash. Idempotency and checkpointing matter.
• Dynamic shapes. Sometimes the channel set is not known at compile time. reflect.Select is the tool, but it has costs.

We also visit weighted fan-out (some sources should send more than others) and priority merging (one input must drain before another). And we look hard at supervisor patterns: how do you restart a failed worker without taking down the whole pipeline?

By the end of this file you should be able to:

• Use reflect.Select when channel sets are dynamic, and explain its cost.
• Design fan-out / fan-in pipelines that survive partial worker failures with restart policies.
• Implement weighted and priority merges.
• Lay out a multi-stage pipeline with clear ownership and resource caps.
• Detect and diagnose pipeline-level performance problems (head-of-line blocking, fan-out skew, merge bottleneck).
• Write supervisor goroutines that restart failed workers.

We will not yet open the runtime hood; that is the professional level.

Prerequisites¶

• Required: Junior and middle file content fully internalised.
• Required: Production experience building or maintaining at least one fan-out / fan-in pipeline.
• Required: Comfort with reflect package, especially reflect.Value, reflect.Type, and reflect.SelectCase.
• Required: Knowledge of supervisor / actor patterns from any language (Erlang, Akka, or simply "restart on fault" libraries).
• Helpful: Exposure to streaming systems (Kafka Streams, Flink, Beam) — the patterns translate.
• Helpful: Experience with pprof, trace, and runtime/metrics.

Glossary¶

Architecture¶

The stage as the unit of design¶

At senior level, you do not think in terms of individual goroutines — you think in terms of stages. A stage is a self-contained unit with:

• A typed input channel.
• A typed output channel.
• A start function that takes context.Context and returns the output.
• A close behaviour: on context cancel, drain quickly and close output.
• An error channel or Result[T] wrapper for typed errors.
• An observability contract: metrics for items in, items out, errors, latency.

Stages compose linearly. Composing two stages gives you a stage. Composing a stage with itself in parallel (with merge) gives you a fan-out stage. The shape is fractal.

A multi-team pipeline often has stage boundaries that map to team boundaries. The contract between teams is the stage interface. Each team can rebuild the inside of its stage independently.

Pipeline-as-graph¶

A non-trivial pipeline is a directed acyclic graph (DAG) of stages. Some examples:

• Linear: A → B → C → D.
• Diamond (fan-out / fan-in): A → [B1, B2, B3] → C.
• Split: A → splitter → [path1: B → C, path2: D → E] → merger → F.
• Tree: source → [parser, validator, classifier] in parallel, each emitting its own output.

The DAG is built at startup. Each edge is a channel. Each node is a stage (one or more goroutines). The runtime is mostly inert: you start the leaves (sources), data flows downstream, and shutdown propagates upstream via context.

The supervisor¶

A supervisor goroutine runs alongside the pipeline. Its job is to monitor worker health and react to failures. The contract:

1. Worker reports its lifecycle: started, running, dying, dead.
2. Supervisor decides: restart, escalate, or terminate the whole pipeline.
3. Restart involves: cleanup of the dead worker's resources, spawn a fresh worker, hook it back into the pipeline.

In Go, this looks like:

type supervisor struct {
    workers int
    spawn   func(ctx context.Context, id int) error
}

func (s *supervisor) run(ctx context.Context) {
    var wg sync.WaitGroup
    for i := 0; i < s.workers; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            for {
                err := s.spawn(ctx, id)
                if err == nil || errors.Is(err, context.Canceled) {
                    return
                }
                log.Printf("worker %d died: %v; restarting", id, err)
                select {
                case <-time.After(backoff()):
                case <-ctx.Done():
                    return
                }
            }
        }(i)
    }
    wg.Wait()
}

The supervisor keeps the worker count stable. If a worker dies, the supervisor spawns a replacement after a backoff. The pipeline as a whole continues.

Restart strategies¶

Borrowed from Erlang/OTP:

• One-for-one: A dead worker is restarted; other workers continue. Default.
• One-for-all: A dead worker triggers restart of all workers. Use when workers share critical state.
• Rest-for-one: Workers spawned after the dead one (in start order) are also restarted. Useful for dependency chains.

Most fan-out workers are independent — one-for-one is correct. Sharded workers (where each worker handles a partition) might benefit from rest-for-one if the partition mapping changes on restart.

Backoff¶

Exponential backoff prevents restart storms:

func backoff(attempt int) time.Duration {
    d := time.Duration(1<<attempt) * 100 * time.Millisecond
    if d > 30*time.Second {
        d = 30 * time.Second
    }
    return d + time.Duration(rand.Intn(int(d/2)))
}

The random jitter avoids synchronised retries. After a few failed restarts, the supervisor may escalate to the parent (tear down the stage, propagate to the pipeline supervisor).

Idempotency and checkpointing¶

For pipelines that consume from external sources (Kafka, S3, files), restart is only safe if processing is idempotent or if the pipeline remembers its position.

Idempotency: the same input processed twice produces the same downstream effect. Achievable via deterministic transformations and unique identifiers on side-effect operations.

Checkpointing: the pipeline periodically persists its progress (e.g., "we've processed up to Kafka offset 42 783"). On restart, it resumes from that offset.

For purely in-memory pipelines, neither applies — restart starts fresh.

Dynamic Channel Sets with reflect.Select¶

When you need it¶

The canonical merge(cs ...<-chan T) requires you to know all channels at compile time. The variadic argument is fixed once the call happens.

Real systems sometimes need to:

• Add a new input at runtime (subscriber joins).
• Remove an input at runtime (subscriber leaves).
• Merge channels of different types.

For these, you cannot use a fixed select statement. You need reflect.Select.

The API¶

package reflect

type SelectCase struct {
    Dir  SelectDir   // direction of case: SelectSend, SelectRecv, SelectDefault
    Chan Value       // the channel (reflect.Value)
    Send Value       // for SelectSend: the value to send
}

func Select(cases []SelectCase) (chosen int, recv Value, recvOK bool)

You build a slice of SelectCase describing the current set of channels and operations. Call reflect.Select. It returns the index of the case that fired, the received value (if any), and a boolean indicating whether the channel was open.

A dynamic merge¶

func dynamicMerge(ctx context.Context, cs []<-chan int) <-chan int {
    out := make(chan int)
    cases := make([]reflect.SelectCase, len(cs)+1)
    for i, c := range cs {
        cases[i] = reflect.SelectCase{
            Dir:  reflect.SelectRecv,
            Chan: reflect.ValueOf(c),
        }
    }
    cases[len(cs)] = reflect.SelectCase{
        Dir:  reflect.SelectRecv,
        Chan: reflect.ValueOf(ctx.Done()),
    }
    go func() {
        defer close(out)
        remaining := len(cs)
        for remaining > 0 {
            chosen, recv, ok := reflect.Select(cases)
            if chosen == len(cs) {
                return // context done
            }
            if !ok {
                // input channel closed: disable case
                cases[chosen].Chan = reflect.ValueOf((chan int)(nil))
                remaining--
                continue
            }
            v := int(recv.Int())
            select {
            case out <- v:
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

The "disable case" trick: setting Chan to a nil channel makes the case never fire (a receive from a nil channel blocks forever). This is the standard way to disable a case in reflect.Select.

Adding a channel at runtime¶

type DynMerger[T any] struct {
    add chan <-chan T
    out chan T
    ctx context.Context
}

func newDynMerger[T any](ctx context.Context) *DynMerger[T] {
    d := &DynMerger[T]{
        add: make(chan <-chan T),
        out: make(chan T),
        ctx: ctx,
    }
    go d.loop()
    return d
}

func (d *DynMerger[T]) Add(c <-chan T) {
    select {
    case d.add <- c:
    case <-d.ctx.Done():
    }
}

func (d *DynMerger[T]) Out() <-chan T {
    return d.out
}

func (d *DynMerger[T]) loop() {
    defer close(d.out)
    var inputs []<-chan T
    for {
        cases := make([]reflect.SelectCase, 0, len(inputs)+2)
        // case 0: ctx.Done
        cases = append(cases, reflect.SelectCase{
            Dir:  reflect.SelectRecv,
            Chan: reflect.ValueOf(d.ctx.Done()),
        })
        // case 1: add channel
        cases = append(cases, reflect.SelectCase{
            Dir:  reflect.SelectRecv,
            Chan: reflect.ValueOf(d.add),
        })
        // cases 2..: input channels
        for _, c := range inputs {
            cases = append(cases, reflect.SelectCase{
                Dir:  reflect.SelectRecv,
                Chan: reflect.ValueOf(c),
            })
        }
        chosen, recv, ok := reflect.Select(cases)
        switch chosen {
        case 0:
            return
        case 1:
            if !ok {
                return
            }
            inputs = append(inputs, recv.Interface().(<-chan T))
        default:
            inputIdx := chosen - 2
            if !ok {
                // close: remove from inputs
                inputs = append(inputs[:inputIdx], inputs[inputIdx+1:]...)
                continue
            }
            select {
            case d.out <- recv.Interface().(T):
            case <-d.ctx.Done():
                return
            }
        }
    }
}

The merger maintains a slice of input channels. On each iteration, it builds a fresh []SelectCase and calls reflect.Select. New channels can be added via Add. Closed channels are removed from the slice.

This is the canonical "runtime-extensible merge." It is more code than the static merge, and it pays a performance cost (allocations per iteration, reflection overhead).

The cost of reflect.Select¶

A reflect.Select call costs roughly 10-100x a static select for the same number of cases. The overhead comes from:

• Building the []SelectCase slice (allocation per call).
• The runtime walking the slice in reflect.rselect.
• Reflection's boxing of channel values and received values.

For pipelines with hundreds of static channels, prefer to break them into groups of 8-16 and use cascaded static merges. For genuinely dynamic sets, reflect.Select is the right tool, but use it sparingly — once per stage, not once per item.

Tricks with reflect.Select¶

Disable a case: Set Chan to a nil channel (reflect.ValueOf((chan T)(nil))). Receives from nil channels block forever, so the case never fires.

Force a default: Add a SelectDefault case if you want non-blocking behaviour.

Mix send and receive: A SelectCase can have Dir: SelectSend with a Send value. The case fires when the channel is ready to receive your value.

Type erasure: All cases must work with reflect.Value. The merge's caller boxes values into reflect.Value; the merge unboxes via recv.Interface().(T).

Weighted and Priority Fan-Out¶

Weighted fan-out¶

Suppose three workers, with capacities 1x, 2x, 4x respectively (perhaps because they run on machines of different sizes). You want to distribute work proportionally.

The runtime's automatic load balancing already approximates this. A faster worker handles items faster, so it reads more from the shared input channel. Weighted fan-out is rarely necessary in Go; competition for input is self-balancing.

When explicit weighting matters:

• The work is not uniform per item, but you can predict its weight in advance.
• You want to enforce a quota even if a worker becomes free.

Then a dispatcher routes by hand:

type weightedWorker struct {
    in     <-chan Item
    weight int
}

func dispatch(ctx context.Context, source <-chan Item, workers []weightedWorker) {
    totalWeight := 0
    for _, w := range workers {
        totalWeight += w.weight
    }
    pos := 0
    for item := range source {
        // pick worker by weighted round-robin
        target := pickWeighted(workers, pos%totalWeight)
        pos++
        select {
        case workers[target].in <- item:
        case <-ctx.Done():
            return
        }
    }
}

pickWeighted returns the worker index whose cumulative weight covers pos. With weights [1, 2, 4], the sequence of picks is [0, 1, 1, 2, 2, 2, 2, 0, 1, 1, ...].

Priority merge¶

Three input channels, ranked: critical, normal, low. Drain critical first, then normal, then low.

Approach 1: Try critical with non-blocking select, fall back if empty.

func priorityMerge3[T any](ctx context.Context, crit, norm, low <-chan T) <-chan T {
    out := make(chan T)
    go func() {
        defer close(out)
        for {
            // try crit
            select {
            case v, ok := <-crit:
                if !ok {
                    crit = nil // disable
                } else {
                    if !forward(ctx, out, v) {
                        return
                    }
                    continue
                }
            default:
            }
            // try norm
            select {
            case v, ok := <-crit:
                if !ok {
                    crit = nil
                } else {
                    if !forward(ctx, out, v) {
                        return
                    }
                    continue
                }
            case v, ok := <-norm:
                if !ok {
                    norm = nil
                } else {
                    if !forward(ctx, out, v) {
                        return
                    }
                    continue
                }
            default:
            }
            // try all
            select {
            case v, ok := <-crit:
                if !ok {
                    crit = nil
                } else if !forward(ctx, out, v) {
                    return
                }
            case v, ok := <-norm:
                if !ok {
                    norm = nil
                } else if !forward(ctx, out, v) {
                    return
                }
            case v, ok := <-low:
                if !ok {
                    low = nil
                } else if !forward(ctx, out, v) {
                    return
                }
            case <-ctx.Done():
                return
            }
            if crit == nil && norm == nil && low == nil {
                return
            }
        }
    }()
    return out
}

func forward[T any](ctx context.Context, out chan<- T, v T) bool {
    select {
    case out <- v:
        return true
    case <-ctx.Done():
        return false
    }
}

The three-tier select: first try critical only, then critical or normal, then all three. If critical has something, take it; only if it does not do we even consider normal.

This is starvation-prone: if critical is non-empty forever, low never gets drained. For starvation-free priority, allocate a quota to each level (e.g., process 10 critical, then 5 normal, then 1 low, repeat).

Hedged merge (variant of priority)¶

Hedged: the first reply wins, the others are cancelled. Different from "priority" in that all inputs race; whichever arrives first is the answer.

Already covered in middle. Senior-level: combine hedging with retries, exponential backoff, and cross-region preference. The code grows quickly; isolate it in its own package.

Partial Failure and Supervisor Patterns¶

The hierarchy¶

A pipeline at senior level has a hierarchy of supervisors:

Pipeline supervisor
    +-- Stage A supervisor
    |       +-- Worker 1
    |       +-- Worker 2
    |       +-- Worker 3
    +-- Stage B supervisor
    |       +-- Worker 1
    |       +-- Worker 2
    +-- Stage C (single goroutine — no supervisor needed)

The stage supervisor restarts dead workers. The pipeline supervisor restarts whole stages if a stage's supervisor fails repeatedly.

Worker contract¶

A worker is a function func(ctx context.Context) error. Contracts:

• Returns nil on graceful exit (input drained, ctx cancelled).
• Returns non-nil error on failure (panic recovered into error, sentinel error, etc.).
• Respects ctx cancellation promptly.
• Releases all resources on exit (defer close, defer unlock, defer cleanup).

The supervisor invokes the worker, catches the return value, and decides what to do.

Restart loop¶

func supervise(ctx context.Context, name string, fn func(ctx context.Context) error) error {
    attempts := 0
    for {
        err := func() (innerErr error) {
            defer func() {
                if r := recover(); r != nil {
                    innerErr = fmt.Errorf("panic: %v", r)
                }
            }()
            return fn(ctx)
        }()
        if err == nil || errors.Is(err, context.Canceled) {
            return err
        }
        attempts++
        log.Printf("supervisor %s: attempt %d failed: %v", name, attempts, err)
        if attempts > 10 {
            return fmt.Errorf("supervisor %s: too many failures: %w", name, err)
        }
        select {
        case <-time.After(backoff(attempts)):
        case <-ctx.Done():
            return ctx.Err()
        }
    }
}

Recovers panics. Backs off between attempts. Escalates after too many failures.

Decoupling restart from input¶

A subtle issue: if the worker reads from a shared channel, the in-flight item when it crashed is lost. Decouple by buffering input per worker:

type supervised struct {
    name    string
    in      <-chan Item
    inFlight chan Item   // buffer of 1
    fn      func(ctx context.Context, it Item) error
}

Each worker reads first from inFlight, then from in. On crash, the supervisor sees inFlight still has the item and feeds it to the next worker.

Simpler alternative: idempotent processing. If the operation can safely be retried, the lost item is just retried.

Bulkheading¶

Bulkhead: prevent one failure from cascading. In a fan-out, if one worker is failing repeatedly, the supervisor should isolate it (stop restarting after N attempts) so the rest of the pipeline can continue.

g, ctx := errgroup.WithContext(ctx)
for i := 0; i < N; i++ {
    i := i
    g.Go(func() error {
        return supervise(ctx, fmt.Sprintf("worker-%d", i), func(ctx context.Context) error {
            return workerLoop(ctx, i)
        })
    })
}
return g.Wait()

If supervise returns a fatal error, the whole group cancels. The bulkhead is the per-worker supervise call: it absorbs transient failures.

Backpressure Engineering¶

At senior level, backpressure becomes engineering rather than reflex.

End-to-end backpressure¶

A pipeline has end-to-end backpressure when the slowest stage's rate determines every upstream stage's rate. Default Go channels give you this if you use unbuffered or small-buffered channels everywhere.

When you violate it (large buffer, drop-on-full, async write), you lose the end-to-end property. The pipeline can outrun itself; memory grows; eventually some hard limit breaks (OOM, downstream rate limiter triggers, etc.).

Senior-level work: identify where backpressure breaks and decide if it should be restored.

Diagnosing backpressure loss¶

Symptoms:

• Memory grows over time, but throughput does not improve.
• A channel sits at full capacity for extended periods (its len ≈ cap).
• One stage's worker count is configured high, but actual concurrency is low (others are blocked).

Fixes:

• Shrink buffers.
• Add explicit feedback (e.g., the consumer publishes its rate; the producer reads it and slows down).
• Switch to pull-based for the affected stage.

Reactive load-shedding¶

If you cannot block (because the input is real-time and shedding is preferred over backing up), drop:

select {
case out <- v:
default:
    sheddedCounter.Inc()
}

But this is loss. Make it explicit, measure it, alert on it.

Rate matching¶

A pipeline that runs continuously should match its source's rate. If the source produces at 1000 RPS but the pipeline processes 800, items accumulate forever (somewhere in some buffer). The pipeline must either:

• Match the source's rate (scale up workers).
• Apply backpressure to the source (often impossible — the source is "the world").
• Shed.

Most production pipelines run at 2-3x headroom over their peak source rate, with shedding only in emergencies.

Auto-scaling the fan-out factor¶

Some pipelines auto-scale the fan-out factor based on observed lag:

go func() {
    t := time.NewTicker(5 * time.Second)
    for range t.C {
        if observedLag > threshold {
            addWorker()
        } else if observedLag < threshold/10 {
            removeWorker()
        }
    }
}()

addWorker spawns a new worker reading from the same input. removeWorker cancels one. Requires careful handling of in-flight items.

Leak Prevention at the Architectural Level¶

Goroutine accounting¶

Maintain a counter per stage:

type stageMetrics struct {
    spawned, exited atomic.Int64
}

func (s *stageMetrics) Live() int64 {
    return s.spawned.Load() - s.exited.Load()
}

If Live() grows over time, that stage is leaking. Wire to Prometheus.

Mandatory goleak in CI¶

Every pipeline test runs goleak.VerifyNone. New tests cannot be merged if they leak. This catches regressions before they ship.

Bounded buffers¶

Every buffered channel has a cap. The cap is documented. The total memory usage of all buffers is bounded.

out := make(chan T, 32) // smooths 32-item bursts, see ADR-014

Owners marked in comments¶

Each goroutine, channel, and resource has an owner. A comment near the spawn says who owns it and when it exits:

// merge spawns N+1 goroutines.
// N forwarders exit when their input closes or ctx cancels.
// The closer exits when all forwarders have exited.
go func() { ... }()

Routine pprof in production¶

A long-running service should expose /debug/pprof/goroutine and have an alert on goroutine count exceeding expected. The alert fires before the pipeline OOMs.

Real-World Analogies¶

Pipeline supervisor as restaurant manager¶

A restaurant has a manager (pipeline supervisor) and section heads (stage supervisors). Section heads are responsible for their waiters. If a waiter quits mid-shift, the section head finds a replacement (one-for-one). If the entire kitchen catches fire, the manager closes the restaurant for the day (escalation).

Weighted fan-out as a court with sized rooms¶

A courthouse has rooms of different sizes. The big room handles big trials; small rooms handle small cases. The dispatcher routes cases by size, not by which room is free. That is weighted fan-out by capacity, not by availability.

Priority merge as a hospital ER¶

Emergency arrivals (critical) go first. Walk-ins (low priority) wait. The triage nurse runs a priority merge across multiple queues. Starvation is a real issue: long-waiting walk-ins eventually get seen even if the ER never stops getting emergencies (the policy includes a "max wait" guarantee).

Bulkheading as ship compartments¶

A ship has watertight bulkheads so a hole in one compartment does not sink the whole ship. A pipeline with bulkheads isolates failures: one worker dying does not stop the others; one stage failing does not stop downstream stages from finishing what they have already received.

Mental Models¶

Model 1: "Pipelines are graphs; goroutines are edges"¶

A pipeline is a graph of stages; the channels are the edges; the data flows along the edges; the goroutines are the agents that move data along the edges. Some edges have multiple agents (fan-out); some edges merge (fan-in).

Model 2: "Cancellation is gravity, errors are wind"¶

Cancellation flows downward through ctx.Done() propagation. Once cancelled, every stage observes and shuts down. Errors flow with data, side-channeled, until a supervisor decides to escalate. Cancellation is universal; errors are local.

Model 3: "Stages are like microservices"¶

Each stage is like a microservice: typed input, typed output, internal state, observability, error contract. The difference is that stages share a process and communicate via channels, not over a network. The discipline is similar.

Model 4: "Backpressure is the heartbeat"¶

A healthy pipeline has subtle backpressure pulses: the consumer briefly pauses, the producer briefly pauses, then both resume. Watch for two pathologies: no backpressure (consumer dies but producer keeps going) and constant backpressure (consumer is always slow).

Model 5: "The supervisor is the immune system"¶

Workers die from time to time — bad inputs, transient errors, panics. The supervisor's job is to detect and replace, not to prevent. Like an immune system: it does not stop the disease, it contains it. Healthy pipelines die many small deaths.

Pros & Cons¶

Pros¶

• Resilience. Supervised pipelines survive worker failures.
• Dynamic shapes. reflect.Select enables runtime extensibility.
• Scalability. Bounded fan-out factors plus auto-scaling adapt to load.
• Observability. Per-stage metrics make pipelines debuggable.
• Modularity. Stage boundaries align with team boundaries.

Cons¶

• Complexity. Supervisors, dynamic merges, and weighted routing add code.
• Cost. reflect.Select is 10-100x slower than static select.
• Failure modes are many. Each restart strategy has corner cases.
• Testing is harder. Restart logic needs fault injection.
• Ownership ambiguity. "Whose stage is this?" can become a real organisational question.

Use Cases¶

Code Examples¶

Example 1: Supervised fan-out with errgroup¶

package main

import (
    "context"
    "errors"
    "fmt"
    "log"
    "math/rand"
    "time"

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

func worker(ctx context.Context, id int, in <-chan int, out chan<- int) error {
    for {
        select {
        case v, ok := <-in:
            if !ok {
                return nil
            }
            if rand.Float64() < 0.05 {
                return fmt.Errorf("worker %d: random failure", id)
            }
            select {
            case out <- v * v:
            case <-ctx.Done():
                return ctx.Err()
            }
        case <-ctx.Done():
            return ctx.Err()
        }
    }
}

func supervise(ctx context.Context, fn func(ctx context.Context) error, maxAttempts int) error {
    for attempt := 0; ; attempt++ {
        err := fn(ctx)
        if err == nil || errors.Is(err, context.Canceled) {
            return err
        }
        if attempt >= maxAttempts {
            return fmt.Errorf("max attempts reached: %w", err)
        }
        log.Printf("attempt %d failed: %v; restarting", attempt, err)
        select {
        case <-time.After(time.Duration(100*(1<<attempt)) * time.Millisecond):
        case <-ctx.Done():
            return ctx.Err()
        }
    }
}

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()
    in := make(chan int)
    out := make(chan int)
    g, gctx := errgroup.WithContext(ctx)
    const workers = 4
    for i := 0; i < workers; i++ {
        i := i
        g.Go(func() error {
            return supervise(gctx, func(ctx context.Context) error {
                return worker(ctx, i, in, out)
            }, 5)
        })
    }
    g.Go(func() error {
        defer close(in)
        for v := 0; v < 50; v++ {
            select {
            case in <- v:
            case <-gctx.Done():
                return gctx.Err()
            }
        }
        return nil
    })
    g.Go(func() error {
        defer close(out)
        // wait until workers + producer all return
        return nil // dummy; the merge close pattern needs more
    })
    for v := range out {
        fmt.Println(v)
    }
    if err := g.Wait(); err != nil {
        log.Fatal(err)
    }
}

(Note: simplified; production version would carefully coordinate the close of out.)

Example 2: Dynamic merge with reflect.Select¶

Already shown above in the deep-dive. Use when subscribers add and remove channels at runtime.

Example 3: Weighted fan-out dispatcher¶

type weighted struct {
    ch     chan<- int
    weight int
}

func dispatch(ctx context.Context, in <-chan int, ws []weighted) {
    total := 0
    for _, w := range ws {
        total += w.weight
    }
    counts := make([]int, len(ws))
    for v := range in {
        target := pick(ws, counts, total)
        counts[target]++
        select {
        case ws[target].ch <- v:
        case <-ctx.Done():
            return
        }
    }
    for _, w := range ws {
        close(w.ch)
    }
}

func pick(ws []weighted, counts []int, total int) int {
    var best int
    var bestRatio float64 = 2.0
    for i, w := range ws {
        if w.weight == 0 {
            continue
        }
        ratio := float64(counts[i]) / float64(w.weight)
        if ratio < bestRatio {
            bestRatio = ratio
            best = i
        }
    }
    return best
}

This is "deficit round-robin": pick the worker whose counts are most behind its weight.

Example 4: Priority merge with quotas¶

func quotaMerge[T any](ctx context.Context, crit, low <-chan T, ratio int) <-chan T {
    out := make(chan T)
    go func() {
        defer close(out)
        for {
            // process up to `ratio` from crit
            for i := 0; i < ratio; i++ {
                select {
                case v, ok := <-crit:
                    if !ok {
                        // drain low
                        for v := range low {
                            select {
                            case out <- v:
                            case <-ctx.Done():
                                return
                            }
                        }
                        return
                    }
                    select {
                    case out <- v:
                    case <-ctx.Done():
                        return
                    }
                default:
                    goto tryLow
                }
            }
        tryLow:
            select {
            case v, ok := <-low:
                if !ok {
                    low = nil
                    continue
                }
                select {
                case out <- v:
                case <-ctx.Done():
                    return
                }
            case v, ok := <-crit:
                if !ok {
                    // drain low
                    for v := range low {
                        select {
                        case out <- v:
                        case <-ctx.Done():
                            return
                        }
                    }
                    return
                }
                select {
                case out <- v:
                case <-ctx.Done():
                    return
                }
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

For every ratio items from crit, one item from low is allowed. Starvation-free.

Example 5: Auto-scaling fan-out¶

type autoScaler struct {
    minWorkers, maxWorkers int
    target                 *atomic.Int64
    workers                []context.CancelFunc
    in                     <-chan int
    out                    chan<- int
    spawn                  func(ctx context.Context, in <-chan int, out chan<- int)
}

func (a *autoScaler) loop(ctx context.Context) {
    a.target.Store(int64(a.minWorkers))
    a.adjust(ctx)
    t := time.NewTicker(5 * time.Second)
    defer t.Stop()
    for {
        select {
        case <-t.C:
            // measure lag, adjust target
            if measureLag() > thresholdHigh {
                a.target.Add(1)
            } else if measureLag() < thresholdLow {
                a.target.Add(-1)
            }
            if a.target.Load() < int64(a.minWorkers) {
                a.target.Store(int64(a.minWorkers))
            }
            if a.target.Load() > int64(a.maxWorkers) {
                a.target.Store(int64(a.maxWorkers))
            }
            a.adjust(ctx)
        case <-ctx.Done():
            return
        }
    }
}

func (a *autoScaler) adjust(ctx context.Context) {
    for int64(len(a.workers)) < a.target.Load() {
        wctx, cancel := context.WithCancel(ctx)
        a.workers = append(a.workers, cancel)
        go a.spawn(wctx, a.in, a.out)
    }
    for int64(len(a.workers)) > a.target.Load() {
        a.workers[len(a.workers)-1]()
        a.workers = a.workers[:len(a.workers)-1]
    }
}

func measureLag() float64 { return 0.0 }

const (
    thresholdHigh = 0.8
    thresholdLow  = 0.2
)

Skeleton; real version measures via the pipeline's metrics and tunes carefully to avoid oscillation.

Example 6: Bulkheaded pipeline¶

func bulkheaded(ctx context.Context, in <-chan Item) error {
    g, gctx := errgroup.WithContext(ctx)
    g.SetLimit(8)
    for item := range in {
        item := item
        g.Go(func() error {
            defer func() {
                if r := recover(); r != nil {
                    log.Printf("item %v: panic %v", item, r)
                }
            }()
            err := process(gctx, item)
            if err != nil {
                log.Printf("item %v: error %v", item, err)
                // do NOT return err; we don't want one bad item to cancel the group
                return nil
            }
            return nil
        })
    }
    return g.Wait()
}

The trick: workers swallow errors locally so the errgroup does not cancel everything. The bulkheading is the swallow — errors stay local.

Example 7: A pipeline with idempotent restart¶

type pipeline struct {
    checkpointer Checkpointer
}

func (p *pipeline) run(ctx context.Context) error {
    last, err := p.checkpointer.Load(ctx)
    if err != nil {
        return err
    }
    in := readFrom(ctx, last)
    for v := range in {
        if err := processIdempotent(ctx, v); err != nil {
            return err
        }
        if err := p.checkpointer.Save(ctx, v.Position); err != nil {
            return err
        }
    }
    return nil
}

On restart, Load() returns the last saved position; readFrom resumes from there. processIdempotent is safe to retry. Checkpoint after each item (slow but safe) or every N items (fast but may reprocess).

Example 8: Health-checked fan-out¶

type healthCheckedWorker struct {
    id     int
    in     <-chan Item
    out    chan<- Result
    healthy atomic.Bool
}

func (w *healthCheckedWorker) loop(ctx context.Context) {
    w.healthy.Store(true)
    defer w.healthy.Store(false)
    for item := range w.in {
        result, err := process(ctx, item)
        if err != nil {
            w.healthy.Store(false)
            return
        }
        select {
        case w.out <- result:
        case <-ctx.Done():
            return
        }
    }
}

A health endpoint exposes the healthy state. Load balancers can route traffic away from unhealthy workers.

Example 9: Drain on graceful shutdown¶

func shutdownGracefully(ctx context.Context, p *Pipeline, drainTimeout time.Duration) error {
    p.stopAccepting()
    drainCtx, cancel := context.WithTimeout(ctx, drainTimeout)
    defer cancel()
    return p.waitForDrain(drainCtx)
}

Stop accepting new inputs; wait up to drainTimeout for in-flight to finish; force cancel if not done. Useful for SIGTERM handling.

Example 10: Cross-stage metric aggregator¶

type metrics struct {
    stages map[string]*stageStats
    mu     sync.RWMutex
}

func (m *metrics) Snapshot() map[string]stageStats {
    m.mu.RLock()
    defer m.mu.RUnlock()
    out := make(map[string]stageStats, len(m.stages))
    for k, v := range m.stages {
        out[k] = *v
    }
    return out
}

A central registry collects per-stage metrics. Exposed via HTTP or Prometheus.

Coding Patterns¶

Pattern 1: The "actor" stage¶

A stage is implemented as an "actor" — a struct with private state, an input channel, an output channel, and a Run(ctx) method.

type Actor[I, O any] struct {
    in  <-chan I
    out chan O
    fn  func(I) O
}

func (a *Actor[I, O]) Run(ctx context.Context) {
    defer close(a.out)
    for v := range a.in {
        select {
        case a.out <- a.fn(v):
        case <-ctx.Done():
            return
        }
    }
}

Pure function inside. State, if any, is captured. Easy to test in isolation.

Pattern 2: The "DAG builder"¶

A small DSL for declaring pipelines:

pipeline := Builder().
    Source("read", readFile).
    FanOut("parse", 8, parseRow).
    Stage("validate", validate).
    FanOut("hash", 4, hashRow).
    Sink("write", writeCSV).
    Build()

Each method appends a stage. Build() returns a runner. Internally, the builder wires channels between stages.

Pattern 3: The "circuit breaker"¶

A stage that monitors error rate and trips open when errors exceed a threshold:

type breaker struct {
    open atomic.Bool
}

func (b *breaker) Maybe(fn func() error) error {
    if b.open.Load() {
        return errBreakerOpen
    }
    err := fn()
    if err != nil {
        b.maybeTrip()
    }
    return err
}

When open, the stage short-circuits failing operations. Combine with retry and back-off.

Pattern 4: The "outbox"¶

For pipelines that produce side effects (DB writes, API calls), use an outbox table:

1. Worker writes the result + the side-effect description to a single transactional outbox.
2. A separate worker reads the outbox and performs the side effects.

Guarantees at-least-once delivery even if the pipeline crashes between produce and side-effect.

Pattern 5: The "drain on flush"¶

Periodically flush in-flight items to a side store for resumability:

type flushable struct {
    pending []Item
    mu      sync.Mutex
}

func (f *flushable) flushLoop(ctx context.Context, interval time.Duration) {
    t := time.NewTicker(interval)
    defer t.Stop()
    for {
        select {
        case <-t.C:
            f.flush()
        case <-ctx.Done():
            f.flush()
            return
        }
    }
}

Tradeoff: flush adds latency but enables recovery. Tune the interval.

Clean Code¶

• One stage, one file. Or one package. Easier to test, easier to navigate.
• Stage interfaces, not concrete types. Allow swapping implementations (real, fake, mock).
• Bound everything. Fan-out factor, buffer size, retry attempts, restart attempts. No infinities in production code.
• Document the failure modes. What happens when a worker dies? When the input closes? When ctx is cancelled? Write it down.
• Reject API misuse at compile time. Use receive-only channels, typed errors, sealed unions where possible.

Product Use / Feature¶

Performance Tips¶

Profile, don't guess¶

Use pprof:

• CPU profile: where is time spent?
• Goroutine profile: how many goroutines, where are they?
• Block profile: where are goroutines blocked? (Channel operations show up here.)
• Mutex profile: where is contention?

Open every pipeline-related profile when investigating performance.

Avoid reflect.Select in hot paths¶

10-100x slower than static select. Use static select with cascaded merges for high-throughput pipelines.

Pre-allocate slices¶

results := make([]Result, 0, expectedCount)

Avoids reallocation during fill. Especially useful in tag-and-reorder buffers.

Sync.Pool for short-lived buffers¶

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

Reuses buffers across goroutines. Cuts GC pressure significantly.

Batched channel sends¶

For very high throughput, send slices over the channel instead of individual items:

out := make(chan []T)
batch := make([]T, 0, 100)
for v := range in {
    batch = append(batch, v)
    if len(batch) == cap(batch) {
        out <- batch
        batch = make([]T, 0, 100)
    }
}

Cuts channel ops by 100x.

Avoid atomic.Map for hot counters¶

sync.Map has overhead. For per-shard counters, []atomic.Int64 is faster.

Best Practices¶

1. Every stage has a supervisor or is robust by construction.
2. reflect.Select is used only when channel sets are truly dynamic.
3. Fan-out factor is configured per stage, not global.
4. Order-preserving stages use tag-and-reorder, not single-threading.
5. Backpressure is end-to-end; buffers are bounded and documented.
6. Metrics on each stage: in, out, errors, latency, in-flight.
7. Tests include: leak (goleak), cancellation, error injection, restart.
8. Idempotent workers where restart safety matters.
9. Checkpointing for long-running pipelines.
10. Graceful shutdown with drain timeout.

Edge Cases & Pitfalls¶

Restart storm¶

A worker fails repeatedly. The supervisor restarts. The new worker hits the same failing input and dies. Repeat at high rate. Fixes:

• Backoff between restarts.
• Cap on restart attempts; escalate after.
• Quarantine the failing input (move to DLQ).

Supervisor loop never exits¶

If the supervisor catches every error including context.Canceled, it loops forever. Always check for cancellation:

if errors.Is(err, context.Canceled) {
    return err
}

reflect.Select with sends that can fail¶

A send case Dir: SelectSend fires when the channel is ready. If the channel is closed, the send case panics. Handle by removing closed channels from the case list.

Mixed-type dynamic merge¶

If your dynamic merge handles channels of different types, you lose type safety at the merge. Either:

• Use a single common type (e.g., chan any) and accept the type assertions.
• Use a generic merge but only over one type at a time.

Backpressure from a stage that doesn't apply backpressure¶

If a stage drops (select with default), backpressure does not propagate through it. The pipeline runs at the source's rate, regardless of downstream. Mark these stages clearly.

Hot fan-in path¶

In a pipeline with thousands of inputs, the merge can become the bottleneck. Use hierarchical merging (cascade of 8-input merges) instead of one giant reflect.Select.

Auto-scaling oscillation¶

If the auto-scaler adjusts every few seconds, it can oscillate: add a worker, lag drops, remove a worker, lag rises, add a worker. Use longer measurement windows and hysteresis.

Common Mistakes¶

Common Misconceptions¶

"reflect.Select is free." — 10-100x slower than static. Use when needed, not by default.

"Supervisor pattern requires Erlang-style processes." — Goroutines + errgroup + restart loops give 90% of OTP's features.

"Auto-scaling always helps." — Often it adds latency from oscillation and complicates debugging. Static fan-out is often better.

"Idempotency means I can ignore restart logic." — Idempotency makes restart safe; it does not make it free.

"Bulkheading isolates everything." — Only what you explicitly isolate. A leaked goroutine still consumes shared memory.

"Priority merging is fair." — Priority is the opposite of fairness. Without quotas, low priority can starve.

Tricky Points¶

reflect.Select order is fair (per call)¶

When multiple cases are ready, reflect.Select picks pseudo-randomly, same as static select. Fair across many calls; unpredictable per call.

reflect.SelectCase{Dir: SelectDefault} is the default case¶

Add at the end. Fires if no other case is ready. Use to make reflect.Select non-blocking.

Closing a channel inside reflect.Select¶

reflect.Select does not close anything. You close channels manually as usual.

Supervisor must not block¶

If the supervisor itself blocks (e.g., on a logging call), it cannot react to new failures. Keep supervisor loops fast; offload work elsewhere.

Bulkhead means errors don't propagate, not that they're ignored¶

A swallowed error must still be logged, counted, and exposed via metrics. Otherwise it is silent failure.

Restart loses in-flight items¶

When a worker dies mid-process, the item it was handling is lost. To not lose: per-worker input buffer + re-feed on restart, or idempotent processing + retry from upstream.

errgroup.SetLimit(N) after Go calls is undefined¶

Set the limit before any Go call. Once Go is called, the limit is fixed for the remainder.

Test¶

package pipeline_test

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

    "go.uber.org/goleak"
)

func TestSupervisorRestartsWorker(t *testing.T) {
    defer goleak.VerifyNone(t)
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()
    attempts := 0
    err := supervise(ctx, "test", func(ctx context.Context) error {
        attempts++
        if attempts < 3 {
            return errors.New("transient")
        }
        return nil
    }, 5)
    if err != nil {
        t.Fatalf("expected nil after recovery, got %v", err)
    }
    if attempts != 3 {
        t.Fatalf("expected 3 attempts, got %d", attempts)
    }
}

func TestSupervisorRespectsCancellation(t *testing.T) {
    defer goleak.VerifyNone(t)
    ctx, cancel := context.WithCancel(context.Background())
    go func() {
        time.Sleep(50 * time.Millisecond)
        cancel()
    }()
    err := supervise(ctx, "test", func(ctx context.Context) error {
        return errors.New("always fails")
    }, 100)
    if !errors.Is(err, context.Canceled) {
        t.Fatalf("expected canceled, got %v", err)
    }
}

func TestDynamicMergeAddRemove(t *testing.T) {
    defer goleak.VerifyNone(t)
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()
    m := newDynMerger[int](ctx)
    a := make(chan int)
    b := make(chan int)
    m.Add(a)
    m.Add(b)
    go func() { a <- 1; close(a) }()
    go func() { b <- 2; close(b) }()
    got := 0
    for v := range m.Out() {
        got += v
        if got == 3 {
            cancel()
        }
    }
    if got != 3 {
        t.Fatalf("expected 3, got %d", got)
    }
}