Error Propagation in Pipelines — Senior Level¶

Table of Contents¶

1. Introduction
2. Prerequisites
3. Error Aggregation Strategies
4. errors.Join Deep Dive
5. Multi-Stage Failure Rollback
6. Compensating Actions
7. Saga Patterns in Go Pipelines
8. Panic Recovery in Stages
9. Memory Model Details
10. Lock-Free Aggregation
11. Structured Concurrency in Depth
12. Error Routing Architectures
13. First Error vs All Errors Trade-Off
14. Dead-Letter Queues
15. Bulkheads
16. Circuit Breakers in Pipelines
17. Observability and Tracing
18. Senior-Level Code Examples
19. Architecture Patterns
20. Clean Code at Scale
21. Product Considerations
22. Best Practices
23. Edge Cases
24. Common Mistakes
25. Tricky Points
26. Test Strategies
27. Tricky Questions
28. Cheat Sheet
29. Self-Assessment
30. Summary
31. Further Reading

Introduction¶

Focus: "I need to design error handling for a system, not a function. Multiple stages with side effects. Partial failure recovery. Aggregated errors. Tracing across stages."

At junior level you learned mechanics. At middle level, design. At senior level, architecture. A pipeline embedded in a production service has additional concerns: partial-failure recovery (some stages succeeded, the next must not orphan the work), aggregated reporting (every failed item, not just one), bulkheads (failure in one part shouldn't cascade), and observability (you must be able to debug it in production at 3 AM).

This file covers:

• Aggregating errors with errors.Join and beyond
• Multi-stage rollback when downstream fails after upstream succeeded
• Compensating actions and the saga pattern
• Panic recovery as a deliberate part of pipeline design
• The Go memory model as it applies to errgroup
• Lock-free aggregation patterns
• Structured concurrency as a design philosophy
• Bulkheads, circuit breakers, dead-letter queues
• Distributed tracing across pipeline stages

This is the territory where you stop using libraries and start designing libraries.

Prerequisites¶

• Junior and middle files in this series.
• Comfort with sync/atomic, the Go memory model, and sync.Mutex/sync.RWMutex.
• Understanding of context.Context plumbing and cancellation semantics.
• Familiarity with database transactions and ACID properties.
• Exposure to distributed system concepts: idempotency, sagas, dead-letter queues.
• Experience reading production goroutine dumps.

Error Aggregation Strategies¶

First-error-wins is the default for errgroup. But many scenarios demand aggregation.

When to aggregate¶

• Batch processing: report every failed item.
• Validation: tell the user every field that's invalid, not just one.
• Multi-source fetch: report every source that failed.
• Schema migrations: list every check that failed.
• Audit logs: every failure recorded for postmortem.

Three aggregation patterns¶

Pattern A: errors.Join (Go 1.20+)¶

func processAll(ctx context.Context, items []Item) error {
    var (
        errs []error
        mu   sync.Mutex
    )
    g, ctx := errgroup.WithContext(ctx)
    g.SetLimit(16)

    for _, it := range items {
        it := it
        g.Go(func() error {
            if err := process(ctx, it); err != nil {
                mu.Lock()
                errs = append(errs, fmt.Errorf("%s: %w", it.ID, err))
                mu.Unlock()
            }
            return nil // do NOT propagate; we collect locally
        })
    }
    _ = g.Wait()
    return errors.Join(errs...)
}

The goroutine returns nil so errgroup doesn't cancel siblings. Errors are collected via a mutex. errors.Join produces a single error that wraps all of them. Callers can iterate via errors.Unwrap or check membership via errors.Is.

Pattern B: Result slice with per-item error¶

type Result struct {
    Item Item
    Err  error
}

func processAll(ctx context.Context, items []Item) []Result {
    results := make([]Result, len(items))
    var g errgroup.Group
    g.SetLimit(16)
    for i, it := range items {
        i, it := i, it
        g.Go(func() error {
            err := process(ctx, it)
            results[i] = Result{Item: it, Err: err}
            return nil
        })
    }
    _ = g.Wait()
    return results
}

No combined error. Caller iterates results, decides per-item action. Useful when callers want fine-grained control.

Pattern C: Multi-error struct¶

type MultiError struct {
    PerItem map[string]error
}

func (m *MultiError) Error() string {
    var parts []string
    for k, v := range m.PerItem {
        parts = append(parts, fmt.Sprintf("%s: %v", k, v))
    }
    return strings.Join(parts, "; ")
}

func (m *MultiError) Unwrap() []error {
    out := make([]error, 0, len(m.PerItem))
    for _, v := range m.PerItem {
        out = append(out, v)
    }
    return out
}

Custom typed multi-error. Callers can errors.As to a *MultiError, then inspect PerItem map. Most expressive but most boilerplate.

Choosing between them¶

• errors.Join: simplest, idiomatic in 1.20+. Default for new code.
• Result slice: when item identity matters and callers process per-item.
• Custom multi-error: when callers need structured access and aggregation logic.

Don't mix in one pipeline. Pick one, document it.

errors.Join Deep Dive¶

errors.Join was added in Go 1.20. It creates an error that wraps multiple errors.

Signature¶

func Join(errs ...error) error

• Returns nil if all errs are nil.
• Otherwise returns an error whose Error() is the concatenation of each error's message (newline-separated) and whose Unwrap() returns []error{errs...} (excluding nils).

Example¶

e1 := errors.New("one")
e2 := errors.New("two")
err := errors.Join(e1, e2)

fmt.Println(err)
// Output:
// one
// two

errors.Is(err, e1) // true
errors.Is(err, e2) // true

Implementation sketch¶

type joinError struct{ errs []error }

func (e *joinError) Error() string {
    var b strings.Builder
    for i, err := range e.errs {
        if i > 0 { b.WriteByte('\n') }
        b.WriteString(err.Error())
    }
    return b.String()
}

func (e *joinError) Unwrap() []error { return e.errs }

func Join(errs ...error) error {
    n := 0
    for _, e := range errs { if e != nil { n++ } }
    if n == 0 { return nil }
    je := &joinError{errs: make([]error, 0, n)}
    for _, e := range errs { if e != nil { je.errs = append(je.errs, e) } }
    return je
}

About 20 lines. Worth understanding so you know what you're getting.

Custom format vs Join's format¶

errors.Join's message is just newline-separated. Often you want richer formatting:

type Aggregate struct {
    Stage string
    Errs  []error
}

func (a *Aggregate) Error() string {
    return fmt.Sprintf("%s: %d errors: %s",
        a.Stage, len(a.Errs), formatList(a.Errs))
}
func (a *Aggregate) Unwrap() []error { return a.Errs }

This gives a structured prefix and a count. errors.Is/errors.As still walk the children.

errors.Is with multi-unwrap¶

errors.Is (in Go 1.20+) handles multi-unwrap correctly. For an error returned by errors.Join(e1, e2):

• errors.Is(joined, e1) → walks to e1, returns true.
• errors.Is(joined, e2) → walks to e2, returns true.

errors.As similarly checks each branch.

For nested multi-unwraps, the walk is a depth-first traversal. Cycles are not handled — if you construct a cycle, errors.Is loops forever. Don't construct cycles.

Performance¶

Joining N errors allocates one wrapper plus the slice. Error() walks all N. Is/As walks until match or exhaustion. All O(N).

If N is large (1000+), the formatted message is huge. Truncate at the boundary:

const maxErrs = 20
if len(errs) > maxErrs {
    errs = append(errs[:maxErrs], fmt.Errorf("(%d more)", len(errs)-maxErrs))
}
return errors.Join(errs...)

Multi-Stage Failure Rollback¶

Pipelines often have side effects: writing to a DB, calling an API, sending a message. If a downstream stage fails after upstream stages succeeded, those side effects must be undone.

The problem¶

A: insert row in users
B: call external API to provision account
C: send welcome email

If C fails, A's row should be deleted and B's account deprovisioned. Otherwise you have a half-created user.

Strategy 1: Two-phase¶

Stage 1 stages all work without committing. Stage 2 commits. If stage 1 fails on any item, no commits happen.

type Tx interface {
    Stage(ctx context.Context, item Item) error
    Commit(ctx context.Context) error
    Rollback() error
}

func process(ctx context.Context, items []Item, tx Tx) error {
    defer tx.Rollback() // no-op if commit succeeded

    g, ctx := errgroup.WithContext(ctx)
    for _, it := range items {
        it := it
        g.Go(func() error {
            return tx.Stage(ctx, it)
        })
    }
    if err := g.Wait(); err != nil { return err }
    return tx.Commit(ctx)
}

Works for DB transactions but not for external APIs that lack two-phase.

Strategy 2: Compensating actions¶

Each forward step has a corresponding rollback step. On failure, run rollbacks in reverse order:

type Step struct {
    Forward    func(ctx context.Context) error
    Compensate func(ctx context.Context) error
}

func runSteps(ctx context.Context, steps []Step) error {
    var completed []Step
    for _, s := range steps {
        if err := s.Forward(ctx); err != nil {
            // rollback in reverse
            rollback(ctx, completed)
            return fmt.Errorf("step failed: %w", err)
        }
        completed = append(completed, s)
    }
    return nil
}

func rollback(ctx context.Context, steps []Step) {
    for i := len(steps) - 1; i >= 0; i-- {
        if err := steps[i].Compensate(ctx); err != nil {
            log.Error("rollback step failed", "err", err)
            // continue trying others
        }
    }
}

Strategy 3: Saga with persistent state¶

For long-running multi-step processes, persist state. If the process crashes mid-rollback, restart it from saved state.

We expand on sagas next.

Compensating Actions¶

A compensating action undoes a previously successful step. Designing them well is half the battle.

Properties of a good compensator¶

• Idempotent: running it twice has the same effect as once.
• Robust against partial state: if forward step partially succeeded (e.g., row inserted but trigger not run), compensator handles both cases.
• Independent of context: doesn't require state from the forward call that might be lost on crash.

Example: account provisioning¶

Forward: provisionAccount(userID) calls API, returns accountID. Compensator: deprovisionAccount(accountID).

If forward succeeded, accountID is known. Compensator just needs accountID — no other state.

If forward partially succeeded (API call timed out, account may or may not exist), compensator must handle both "account exists" and "account does not exist." Idempotent: if exists, delete; else, no-op.

Example: email send¶

Forward: sendEmail(to, body). Compensator: ... ?

You can't unsend an email. Some compensators are impossible. Options:

• Send a follow-up "ignore the previous message" email.
• Mark the email as "test" in your own DB, so the user knows.
• Accept that this step is non-compensatable and design accordingly (e.g., make it the last step so nothing can fail after it).

Compensation ordering¶

Reverse order: last forward step undone first.

forward: A B C D
compensate: D C B A

Why reverse? Because dependencies are forward. D depends on C's output; D's compensator must run before C's, or C's resources are gone.

Compensation under concurrent failure¶

If two steps fail simultaneously, do you run both compensators? Or only the failing step's? Best practice: only run compensators for succeeded steps. The failing step didn't produce side effects (assuming it returned without success), so it has nothing to compensate.

If the failing step partially succeeded (network timeout after API call), you do need its compensator. Detect this via "best-effort cleanup":

func (s *Step) ForwardWithCleanup(ctx context.Context) (didStart bool, err error) {
    didStart = false
    // before doing anything that has side effects
    err = s.precheck(ctx)
    if err != nil { return false, err }
    didStart = true // we're about to do work that may partially succeed
    err = s.do(ctx)
    return didStart, err
}

Caller:

didStart, err := step.ForwardWithCleanup(ctx)
if err != nil {
    if didStart {
        // run compensator
    }
    return err
}

Saga Patterns in Go Pipelines¶

A saga is a long-running transaction implemented as a sequence of forward steps with corresponding compensating actions.

Two flavors¶

Orchestration¶

A central coordinator drives each step:

type Orchestrator struct {
    steps []Step
}

func (o *Orchestrator) Run(ctx context.Context) error {
    var completed []Step
    defer func() {
        for i := len(completed) - 1; i >= 0; i-- {
            completed[i].Compensate(ctx)
        }
    }()

    for _, s := range o.steps {
        if err := s.Forward(ctx); err != nil {
            return err
        }
        completed = append(completed, s)
    }
    completed = nil // success; defer is a no-op
    return nil
}

Simple. Easy to debug. Single point of control. Doesn't scale to long-running (days, weeks) sagas without persistence.

Choreography¶

Each step emits an event; the next step listens and runs. Compensations are also event-driven. More flexible but harder to reason about.

In a Go pipeline, orchestration is the norm. Choreography is for cross-service workflows (often using a message broker).

Persistent saga state¶

If your saga must survive a crash, persist state at each step:

type SagaState struct {
    ID            string
    Step          int
    Completed     []string
    LastForward   string
    LastError     string
}

func (s *SagaState) Save(ctx context.Context, db *sql.DB) error {
    _, err := db.ExecContext(ctx, "INSERT INTO sagas ... ON CONFLICT (id) DO UPDATE ...", s)
    return err
}

On restart, load all incomplete sagas and resume:

sagas, _ := loadIncompleteSagas(ctx, db)
for _, s := range sagas {
    go resumeSaga(ctx, s)
}

The resume logic: if we're rolling back, continue rollback. If forward, continue forward. Each step is idempotent so re-running is safe.

This is the architecture for systems like AWS Step Functions, Cadence, Temporal. You can implement a lightweight version in your own service.

Panic Recovery in Stages¶

A panic in a g.Go function crashes the program. In production this is usually catastrophic. Recovery is a deliberate design choice.

When to recover¶

• Stages that process untrusted input (user data, external API responses).
• Stages that use third-party libraries with unknown panic behaviour.
• Long-running workers where uptime matters.

When NOT to recover¶

• Genuine programmer errors (nil dereference, index out of range). Recovery hides bugs.
• Critical invariant violations. Better to crash and restart than continue in a corrupt state.

Recovery pattern¶

func safeStage(ctx context.Context, fn func(context.Context) error) (err error) {
    defer func() {
        if r := recover(); r != nil {
            buf := make([]byte, 1<<16)
            n := runtime.Stack(buf, false)
            err = fmt.Errorf("panic in stage: %v\n%s", r, buf[:n])
        }
    }()
    return fn(ctx)
}

g.Go(func() error { return safeStage(ctx, parseStage) })

Notes:

• Named return so deferred recover can set it.
• Capture stack with runtime.Stack for debugging.
• Wrap in error so the pipeline's error flow continues normally.

Distinguishing panics from errors¶

Sometimes useful at the caller:

type PanicError struct {
    Value any
    Stack []byte
}

func (e *PanicError) Error() string { return fmt.Sprintf("panic: %v", e.Value) }

func safeStage(ctx context.Context, fn func(context.Context) error) (err error) {
    defer func() {
        if r := recover(); r != nil {
            buf := make([]byte, 1<<16)
            n := runtime.Stack(buf, false)
            err = &PanicError{Value: r, Stack: buf[:n]}
        }
    }()
    return fn(ctx)
}

// caller
var pe *PanicError
if errors.As(err, &pe) {
    log.Error("panic recovered", "stack", string(pe.Stack))
}

The *PanicError lets callers distinguish recovered panics from regular errors and act differently (alert, page on-call, etc.).

Recovery and partial state¶

If a panic happened mid-write, your data may be inconsistent. Recovery converts the panic to an error but doesn't undo the partial write. Use compensating actions (sagas) for that.

Don't recover at every level¶

Recover at the goroutine boundary, not inside every helper. Otherwise:

func helper() {
    defer recover() // useless? hides real bugs?
}

The deeply nested recover swallows the panic. The function above thinks everything is fine. The bug is hidden. Recover only at well-defined boundaries.

Memory Model Details¶

The Go memory model defines when one goroutine's reads see another's writes. Errgroup interacts with it precisely.

Happens-before relationships¶

The memory model says: in g.Go(f), the call to Go happens before f starts. So state set up before Go is visible inside f.

After g.Wait() returns, every operation inside every spawned f has completed and happened-before the return. So state written inside any f is visible to the caller.

This is why:

results := make([]Result, n)
for i := 0; i < n; i++ {
    i := i
    g.Go(func() error {
        results[i] = compute(i)
        return nil
    })
}
g.Wait()
// results is fully populated; reads are safe.

Works without explicit synchronisation. The Wait provides the necessary happens-before.

What about reads inside g.Go?¶

shared := 0
g.Go(func() error { shared = 1; return nil })
g.Go(func() error { fmt.Println(shared); return nil })
g.Wait()

The two goroutines race. No happens-before between them. Race detector flags this. Use atomics or a mutex.

sync.Once internals¶

errgroup uses sync.Once to capture the first error. sync.Once.Do(f) is implemented via an atomic counter and a mutex. Its guarantee: writes inside f happen-before any subsequent return from Do for any caller.

So in errgroup:

g.errOnce.Do(func() {
    g.err = err
    if g.cancel != nil { g.cancel() }
})

After this returns to any caller, g.err and g.cancel are observably set. Wait reads g.err after wg.Wait(), which synchronises with all Done calls, which include the errOnce.Do execution by the failing goroutine.

The chain: failing goroutine writes err -> Once synchronises -> Done -> wg.Wait -> read err. All correct.

Why mutex around aggregation works¶

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 nil
})
g.Wait()
// Read errs here; safe due to wg.Wait synchronising with every Done,
// and mu.Unlock inside Done's window.

The mutex ensures writes don't race. g.Wait() ensures the final state is visible.

When you DON'T need a mutex¶

If each goroutine writes to a unique slot (slice index), no synchronisation between goroutines is needed. They write to disjoint memory. g.Wait provides the final visibility.

results := make([]int, n)
for i := 0; i < n; i++ {
    i := i
    g.Go(func() error { results[i] = compute(i); return nil })
}
g.Wait()
// safe to read all of results

This is the "result slot" pattern. Fast and lock-free.

Lock-Free Aggregation¶

When you want to aggregate without mutex contention.

Atomic counters¶

For numerical aggregation:

var total atomic.Int64
g.Go(func() error {
    n, err := work()
    if err != nil { return err }
    total.Add(n)
    return nil
})

atomic.Int64 is faster than mutex for simple updates. Use for counters, sums, etc.

Per-goroutine buffers¶

Each goroutine writes to its own buffer; aggregate after.

buffers := make([][]Result, n)
for i := 0; i < n; i++ {
    i := i
    g.Go(func() error {
        buf := buffers[i]
        for it := range items[i] {
            buf = append(buf, compute(it))
        }
        buffers[i] = buf
        return nil
    })
}
g.Wait()

// Merge after
var all []Result
for _, buf := range buffers {
    all = append(all, buf...)
}

No locking during the hot path. Merge is sequential but fast.

Channels for streaming aggregation¶

results := make(chan Result, runtime.NumCPU())
go func() {
    g.Wait()
    close(results)
}()

for _, item := range items {
    item := item
    g.Go(func() error {
        r, err := compute(item)
        if err != nil { return err }
        results <- r
        return nil
    })
}

var all []Result
for r := range results {
    all = append(all, r)
}

Channels serialize through the channel's internal lock; for very high throughput, prefer slot patterns.

sync.Map for keyed aggregation¶

For per-key aggregation:

var counts sync.Map

g.Go(func() error {
    for v := range in {
        key := compute(v)
        val, _ := counts.LoadOrStore(key, &atomic.Int64{})
        val.(*atomic.Int64).Add(1)
    }
    return nil
})

sync.Map is optimised for "write once, read many" workloads. For "write many, read once," a plain map with a mutex is often faster.

Structured Concurrency in Depth¶

The principle: every goroutine has a defined lifetime bounded by its parent's lifetime. Tony Hoare proposed it in CSP; modern languages (Kotlin, Swift) implement it natively. Go doesn't, but you can implement it with discipline.

Three rules¶

1. Every goroutine is spawned via a tracked mechanism (errgroup, WaitGroup, channel-based).
2. The parent does not return until all spawned children have returned.
3. Cancellation propagates downward.

errgroup.WithContext implements all three.

Anti-pattern: orphan goroutines¶

go updateMetrics()

updateMetrics runs forever. The function that started it has returned. Who cancels it? Often nobody. This is unstructured.

Refactor to structured¶

func service(ctx context.Context) error {
    g, ctx := errgroup.WithContext(ctx)
    g.Go(func() error { return updateMetrics(ctx) })
    g.Go(func() error { return handleRequests(ctx) })
    return g.Wait()
}

Both goroutines are now bounded by service's lifetime. When ctx is cancelled, both stop. service returns when both stop.

Nested groups¶

g.Go(func() error {
    inner, ctx := errgroup.WithContext(ctx)
    inner.Go(...)
    inner.Go(...)
    return inner.Wait()
})

Inner's children are bounded by the inner goroutine's lifetime, which is bounded by outer. Trees of bounded lifetimes.

Benefits¶

• No leaks.
• Clear ownership.
• Cancellation is well-defined.
• Easy to reason about: lifetime trees are straightforward.

Cost¶

• More boilerplate.
• "Just go func()" is sometimes more concise.
• Worth it for any non-trivial service.

Error Routing Architectures¶

In a complex service, errors flow through multiple layers. Designing this routing is its own art.

Layer 1: stage-internal handling¶

Sentinel errors that the stage knows how to handle:

for v := range in {
    if err := tryProcess(v); err != nil {
        switch {
        case errors.Is(err, ErrSkip):
            metrics.SkipCount.Inc()
            continue
        case errors.Is(err, ErrPoison):
            deadLetter(v)
            continue
        default:
            return fmt.Errorf("process: %w", err)
        }
    }
}

Local errors don't bubble. The stage decides.

Layer 2: stage-boundary wrapping¶

Errors that escape the stage are wrapped:

return fmt.Errorf("parse-stage: %w", err)

Or attributed via a typed error:

return &StageError{Stage: "parse", Err: err}

Layer 3: pipeline-level routing¶

The pipeline runner classifies and routes:

err := pipeline.Run(ctx)
switch {
case err == nil:
    return Success()
case errors.Is(err, context.Canceled):
    return Cancelled()
case errors.Is(err, context.DeadlineExceeded):
    return Timeout()
case errors.Is(err, ErrTransient):
    return Retry()
default:
    return Failure(err)
}

Layer 4: service-level reporting¶

The service surfaces errors to callers:

• HTTP handler: map to status code.
• gRPC: map to status code with details.
• CLI: print user-friendly message; log full chain.
• Background worker: increment metrics, log, requeue.

Each layer has its job. Errors flow up; metadata accumulates; the final form is appropriate for the audience.

Don't skip layers¶

Tempting to handle errors "at the top" and skip stage-internal handling. Don't:

• Loses context.
• Forces top-level to know about every stage.
• Couples layers.

Each layer should handle what it knows; pass up what it doesn't.

First Error vs All Errors Trade-Off¶

When designing a pipeline's error policy, decide explicitly:

Use first-error-wins when¶

• The caller is a human user expecting one clear message.
• Subsequent failures are likely consequences of the first.
• Latency matters: fail fast saves work.
• Atomicity matters: any failure invalidates the whole.

Use aggregation when¶

• Each item is independent.
• Callers want a full report.
• Latency is less important than completeness.
• Partial results have value.

Hybrid¶

A pipeline might:

1. Fail fast on structural errors (DB unavailable, config wrong).
2. Aggregate on per-item errors (this row invalid, that one valid).

In code:

func run(ctx context.Context, items []Item) error {
    if err := preflight(); err != nil {
        return fmt.Errorf("preflight: %w", err)
    }
    var perItem []error
    var mu sync.Mutex

    g, ctx := errgroup.WithContext(ctx)
    g.SetLimit(16)
    for _, it := range items {
        it := it
        g.Go(func() error {
            err := process(ctx, it)
            if err != nil {
                if errors.Is(err, ErrCritical) {
                    return err // first-error path
                }
                mu.Lock()
                perItem = append(perItem, err)
                mu.Unlock()
            }
            return nil
        })
    }
    if err := g.Wait(); err != nil {
        return err
    }
    if len(perItem) > 0 {
        return errors.Join(perItem...)
    }
    return nil
}

Critical errors bubble up immediately; per-item errors aggregate. Best of both.

Dead-Letter Queues¶

A dead-letter queue (DLQ) holds items that repeatedly fail. They're not lost but not blocking the pipeline.

Pattern¶

type DLQ interface {
    Put(ctx context.Context, item Item, reason error) error
}

func processWithDLQ(ctx context.Context, item Item, dlq DLQ, maxAttempts int) error {
    var lastErr error
    for attempt := 0; attempt < maxAttempts; attempt++ {
        err := process(ctx, item)
        if err == nil { return nil }
        if errors.Is(err, ErrPoison) {
            return dlq.Put(ctx, item, err)
        }
        lastErr = err
        time.Sleep(backoff(attempt))
    }
    return dlq.Put(ctx, item, fmt.Errorf("max attempts: %w", lastErr))
}

After max attempts, item moves to DLQ. Pipeline continues. Operators inspect DLQ later.

Implementation¶

• Database table: dead_letter_queue (id, payload, error, attempts, last_attempt).
• Message queue (Kafka, RabbitMQ) with a separate DLQ topic.
• Cloud-native (AWS SQS, GCP Pub/Sub) — built-in DLQ support.

When to use¶

• Async pipelines where items can be retried later.
• Async workers processing user-submitted content.
• Background ETL where some items are bad.

Not appropriate for synchronous request-response.

Bulkheads¶

A bulkhead isolates failures to a portion of the system. Like watertight compartments on a ship.

Pattern: per-tenant isolation¶

type TenantPipeline struct {
    tenantID string
    limiter  *rate.Limiter
    workers  *errgroup.Group
}

func (tp *TenantPipeline) Submit(ctx context.Context, item Item) error {
    if !tp.limiter.Allow() {
        return ErrTenantThrottled
    }
    tp.workers.Go(func() error { return tp.process(ctx, item) })
    return nil
}

Each tenant gets its own limiter and worker pool. If tenant A's pipeline backs up, tenant B is unaffected.

Pattern: per-resource semaphore¶

dbSem := semaphore.NewWeighted(20)
apiSem := semaphore.NewWeighted(5)

Resources have bounded contention. If the API is slow, only 5 goroutines are blocked on it, not all.

Pattern: separate goroutine pools¶

fastPool, _ := errgroup.WithContext(ctx)
fastPool.SetLimit(100)
slowPool, _ := errgroup.WithContext(ctx)
slowPool.SetLimit(10)

if isFast(item) {
    fastPool.Go(...)
} else {
    slowPool.Go(...)
}

Slow items don't starve fast ones.

Circuit Breakers in Pipelines¶

A circuit breaker stops requests to a failing dependency, letting it recover.

States¶

• Closed: normal operation; requests flow.
• Open: failures exceeded threshold; requests fail fast.
• Half-open: after cooldown, try one request to see if recovered.

Implementation¶

Use github.com/sony/gobreaker or roll your own:

type Breaker struct {
    mu        sync.Mutex
    failures  int
    threshold int
    state     int
    openUntil time.Time
}

func (b *Breaker) Call(ctx context.Context, fn func() error) error {
    b.mu.Lock()
    if b.state == StateOpen && time.Now().Before(b.openUntil) {
        b.mu.Unlock()
        return ErrCircuitOpen
    }
    b.mu.Unlock()

    err := fn()
    b.mu.Lock()
    defer b.mu.Unlock()
    if err != nil {
        b.failures++
        if b.failures >= b.threshold {
            b.state = StateOpen
            b.openUntil = time.Now().Add(30 * time.Second)
        }
    } else {
        b.failures = 0
        b.state = StateClosed
    }
    return err
}

In a pipeline:

g.Go(func() error {
    return breaker.Call(ctx, func() error { return apiCall(ctx, item) })
})

When the API is down, the breaker opens; subsequent goroutines fail fast without hammering the API. After 30 seconds, one is admitted to test recovery.

Observability and Tracing¶

Production pipelines need to be debuggable.

Structured logging¶

log.Error("stage failed",
    "stage", "parse",
    "item", item.ID,
    "err", err)

Every error log includes the stage, item ID, and full error. Searchable, filterable.

Metrics¶

• pipeline.duration (histogram, labelled by status).
• pipeline.errors.count (counter, labelled by stage, error type).
• stage.duration (histogram, labelled by stage).
• pipeline.in_flight (gauge).
• pipeline.retries.count (counter).

Tracing¶

Distributed tracing (OpenTelemetry) propagates a trace ID through context. Each stage spans:

func parseStage(ctx context.Context, in <-chan []byte, out chan<- Record) error {
    for b := range in {
        ctx, span := tracer.Start(ctx, "parse")
        r, err := parse(b)
        if err != nil { span.RecordError(err) }
        span.End()
        if err != nil { return err }
        out <- r
    }
    return nil
}

In a tracing UI you see the pipeline run as a tree: total time, per-stage time, per-item span. Indispensable for diagnosing slow stages or contention.

Don't log inside hot loops¶

log.Info per item, at 100k items/sec, generates 100k log lines. Sample:

if rand.Intn(1000) == 0 {
    log.Info("processed", "item", item.ID)
}

Or log only on error:

if err != nil { log.Error("failed", "err", err) }

Senior-Level Code Examples¶

Example: ETL with aggregation and rollback¶

type Importer struct {
    db        *sql.DB
    workers   int
    dlq       DLQ
}

func (im *Importer) Import(ctx context.Context, items []Item) error {
    var (
        succeeded []string
        succMu    sync.Mutex
        errs      []error
        errMu     sync.Mutex
    )

    g, ctx := errgroup.WithContext(ctx)
    g.SetLimit(im.workers)

    for _, it := range items {
        it := it
        g.Go(func() error {
            err := im.importOne(ctx, it)
            if err == nil {
                succMu.Lock()
                succeeded = append(succeeded, it.ID)
                succMu.Unlock()
                return nil
            }
            if errors.Is(err, ErrCritical) {
                return err
            }
            if errors.Is(err, ErrPoison) {
                _ = im.dlq.Put(ctx, it, err)
                return nil
            }
            errMu.Lock()
            errs = append(errs, fmt.Errorf("%s: %w", it.ID, err))
            errMu.Unlock()
            return nil
        })
    }

    if err := g.Wait(); err != nil {
        // Critical failure: rollback succeeded items.
        im.rollback(context.Background(), succeeded)
        return err
    }

    if len(errs) > 0 {
        return errors.Join(errs...)
    }
    return nil
}

func (im *Importer) rollback(ctx context.Context, ids []string) {
    g, ctx := errgroup.WithContext(ctx)
    g.SetLimit(im.workers)
    for _, id := range ids {
        id := id
        g.Go(func() error {
            _, err := im.db.ExecContext(ctx, "DELETE FROM imports WHERE id = $1", id)
            if err != nil {
                log.Error("rollback failed", "id", id, "err", err)
            }
            return nil
        })
    }
    _ = g.Wait()
}

Three error categories:

1. Critical (DB down, etc.): fail-fast, rollback succeeded items.
2. Poison (bad data): DLQ, continue.
3. Per-item (transient, retryable later): aggregate, return at end.

This is the shape of a production importer.

Example: Saga with persistence¶

type Saga struct {
    ID    string
    Steps []Step
    State *SagaState
    DB    *sql.DB
}

func (s *Saga) Run(ctx context.Context) error {
    for i := s.State.NextStep; i < len(s.Steps); i++ {
        s.State.NextStep = i
        s.persist(ctx)

        if err := s.Steps[i].Forward(ctx); err != nil {
            s.State.Error = err.Error()
            s.persist(ctx)
            return s.rollback(ctx, i)
        }
    }
    s.State.NextStep = len(s.Steps)
    s.persist(ctx)
    return nil
}

func (s *Saga) rollback(ctx context.Context, failedStep int) error {
    for i := failedStep - 1; i >= 0; i-- {
        s.State.RollbackStep = i
        s.persist(ctx)
        if err := s.Steps[i].Compensate(ctx); err != nil {
            log.Error("compensation failed", "step", i, "err", err)
            // continue; can't help it
        }
    }
    return errors.New("saga rolled back: " + s.State.Error)
}

func (s *Saga) persist(ctx context.Context) {
    _, _ = s.DB.ExecContext(ctx,
        "INSERT INTO sagas (id, state) VALUES ($1, $2) ON CONFLICT (id) DO UPDATE SET state = $2",
        s.ID, s.State)
}

State persisted between steps. On crash, reload state, resume.

Example: panic-safe pipeline¶

func runSafely(ctx context.Context, fn func(context.Context) error) (err error) {
    defer func() {
        if r := recover(); r != nil {
            buf := make([]byte, 1<<16)
            n := runtime.Stack(buf, false)
            err = &PanicError{Value: r, Stack: buf[:n]}
            log.Error("panic", "value", r, "stack", string(buf[:n]))
        }
    }()
    return fn(ctx)
}

func runPipeline(ctx context.Context) error {
    g, ctx := errgroup.WithContext(ctx)
    g.Go(func() error { return runSafely(ctx, stage1) })
    g.Go(func() error { return runSafely(ctx, stage2) })
    g.Go(func() error { return runSafely(ctx, stage3) })
    return g.Wait()
}

Every stage is wrapped. A panic becomes an error. The pipeline behaves predictably.

Architecture Patterns¶

Microservice pipeline¶

Each stage is a separate service connected by a message broker. Errors propagate via DLQ topics. Sagas via event sourcing.

Pros: scalability, isolation. Cons: complexity, latency.

Modular monolith pipeline¶

Stages are Go packages in one binary. Errgroup-based wiring. Errors via the standard pipeline patterns.

Pros: simplicity, low latency. Cons: scaling per-stage requires whole-binary scaling.

Stream processing¶

Apache Kafka + a Go consumer. Each consumer processes a stream of events. Errors via DLQ topic. Compensations via event sourcing.

Pros: durability, replay. Cons: operational overhead.

Choosing¶

For most internal pipelines, the modular monolith with errgroup is sufficient. Reach for distributed architectures when scale or durability requires it.

Clean Code at Scale¶

At scale, "clean" means:

1. One package per pipeline. All stages, errors, types, tests in one package.
2. Public surface is minimal. Most types unexported. Just Run(ctx, ...), errors, and config.
3. Configuration through structs, not function arguments.
4. Tests exhaustively cover error paths. Not just happy.
5. Metrics and logs in every stage, via cross-cutting wrappers.
6. No mutable global state. Even loggers are injected.
7. Documentation per stage explains its contract.

A senior-quality pipeline package looks like a library: stable API, comprehensive tests, clear docs.

Product Considerations¶

• Cost of failure: a failed import affects N customers; design accordingly.
• Recovery time: if cancelled, how long to resume? Persistent sagas help.
• Observability budget: tracing every item is expensive; sample 1%.
• Tenancy: shared infrastructure needs bulkheads.
• Compliance: error logs may contain PII; redact before logging.

Best Practices¶

1. Default to first-error; consider aggregation when partial results matter.
2. Use errors.Join for aggregation (1.20+).
3. Design compensating actions for every step with side effects.
4. Persist saga state for long-running workflows.
5. Recover panics at goroutine boundaries.
6. Lock-free aggregation where possible (slot pattern).
7. Bulkheads per tenant/resource.
8. Circuit breakers for external dependencies.
9. DLQ for poisoned items.
10. Structured logs + metrics + tracing for every pipeline.

Edge Cases¶

Cancellation during rollback¶

If the parent context is cancelled mid-rollback, rollback may not complete. Workaround: use a separate, longer-lived context for rollback:

func (s *Saga) rollback(parentCtx context.Context, ...) error {
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
    defer cancel()
    // use ctx for rollback steps
}

Rollback persists; parent's cancellation doesn't abort cleanup.

Compensator failure during rollback¶

What if a compensator fails? Best-effort: log and continue. Pipeline ends with rollback partially complete. Operators must finish manually.

For critical systems, persist rollback state. On restart, resume rollback.

Errors.Join with 1000 items¶

Memory: 1000 errors + a slice. Manageable. But Error() produces a 1000-line string. Truncate before serialising.

Multi-error errors.Is performance¶

errors.Is walks both single and multi unwraps. For deeply nested or wide multi-errors, it's O(N) total. In practice, fast.

Common Mistakes¶

1. Aggregating with errors.Join and then returning g.Wait()'s error too — double-report.
2. Recovering panics without logging the stack trace.
3. Compensating in forward order instead of reverse.
4. Forgetting to persist saga state after each step.
5. Using shared errgroup across requests — single-use violations.
6. Circuit breaker without timeout — open forever.
7. DLQ without consumer — items accumulate forever.

Tricky Points¶

• errors.Join(nil) returns nil. Useful: harmless to call with all nil.
• errors.Is against a joined error walks all branches.
• recover only catches panics in the same goroutine.
• runtime.Stack is expensive; avoid in tight loops.
• sync.Once cannot be reset; for re-runs, use a new one.
• errgroup is single-use; for re-runs, create a new group.

Test Strategies¶

Property-based: cancellation always terminates¶

func TestPipelineAlwaysTerminates(t *testing.T) {
    f := func(seed int64) bool {
        ctx, cancel := context.WithCancel(context.Background())
        rng := rand.New(rand.NewSource(seed))
        go func() {
            time.Sleep(time.Duration(rng.Intn(100)) * time.Millisecond)
            cancel()
        }()
        done := make(chan struct{})
        go func() {
            _ = Run(ctx, randomItems(rng))
            close(done)
        }()
        select {
        case <-done:
            return true
        case <-time.After(time.Second):
            return false
        }
    }
    if err := quick.Check(f, nil); err != nil { t.Fatal(err) }
}

Fault injection¶

type FaultyDB struct {
    real    *sql.DB
    failOn  string
}

func (f *FaultyDB) Insert(ctx context.Context, r Record) error {
    if r.ID == f.failOn {
        return fmt.Errorf("simulated: %w", ErrTransient)
    }
    return f.real.Insert(ctx, r)
}