Cancellation Propagation — Middle Level¶

Table of Contents¶

1. Introduction
2. Prerequisites
3. Recap of Junior-Level Mechanics
4. Multi-Stage Pipelines: Threading the Context
5. Cancel-on-Error with errgroup
6. Deadline Propagation
7. Upstream and Downstream Cancellation in Detail
8. Fan-Out with Coordinated Cancellation
9. Fan-In with Drain Semantics
10. Context Trees in Real Applications
11. context.AfterFunc and WithCancelCause
12. Cancellable I/O
13. Cancellation in Worker Pools
14. Pipeline Lifecycle Diagram
15. Patterns and Anti-Patterns
16. Testing Cancellation
17. Performance Considerations
18. Common Mistakes at the Middle Level
19. Tricky Questions
20. Cheat Sheet
21. Summary
22. Further Reading

Introduction¶

At the junior level you learned how a single stage uses select and ctx.Done() to exit cleanly. Middle-level pipelines have multiple stages, fan-out workers, error propagation that triggers cancellation, deadlines that travel through the system, and shutdown protocols that drain in-flight work before terminating. The mechanics are the same — close a channel, propagate the close — but the wiring is more intricate.

This file covers:

• How to thread one context.Context through five or fifty stages without losing the cancellation path.
• How errgroup combines wait-for-all with cancel-on-error in a single primitive.
• How deadlines flow from the request boundary all the way to the deepest goroutine, including when child stages narrow the deadline further.
• The difference between upstream cancellation (source decides to stop) and downstream cancellation (sink decides to stop), and how the same primitive handles both.
• Fan-out cancellation: how N parallel workers all exit when one of them, or the controller, decides to stop.
• Fan-in cancellation: how a merge stage drains its inputs to allow upstream stages to exit.
• How to design a cancellation-aware worker pool that bounds concurrency, handles errors, and shuts down cleanly.
• Practical testing patterns to verify that no goroutine leaks under cancellation.

By the end of this file you should be able to design a pipeline whose every goroutine has a documented exit path, every block point has a select on cancellation, and every shutdown completes within a known time bound.

Prerequisites¶

• The junior-level material on context.Context, done channels, and the cancellable stage template.
• Comfort with select, for v := range ch, defer close(out), and sync.WaitGroup.
• Familiarity with closures and how loop variables interact with goroutine captures.
• Awareness that channels can be closed once and that closing broadcasts to all receivers.

A working knowledge of golang.org/x/sync/errgroup is helpful but not required — it is introduced here.

Recap of Junior-Level Mechanics¶

Before diving in, the foundations in one box:

• context.Context carries a Done() channel that closes on cancellation, an Err() that reports why, and a Deadline() for time-based bounds.
• Every stage is a goroutine that owns its output channel; it closes the output in defer when it returns.
• Every blocking channel operation is wrapped in a select whose other case is <-ctx.Done().
• defer cancel() is mandatory wherever WithCancel/WithTimeout/WithDeadline is called.
• A range over a channel exits when the channel closes. A stage that sees ctx.Done() returns, which fires defer close(out), which makes the next stage's range exit.

These rules do not change at the middle level. What changes is the scale and the number of moving parts.

Reframing: cancellation as a first-class part of pipeline design¶

At junior level cancellation feels like an extra concern bolted onto a pipeline. At middle level it should be the first thing you design. The right mental shift: a pipeline is a graph of stages connected by channels and by a shared cancellation signal. The channels carry data; the cancellation signal carries authority. Both flow through the system; without both, the pipeline is incomplete.

When you sketch a new pipeline on a whiteboard, draw two arrows between every pair of stages: a thick arrow for data and a thin arrow for ctx.Done(). The thin arrow does not literally connect stages — it is a shared channel that all stages select on — but visualising it makes the design explicit.

Three questions to ask while designing:

1. Where does cancellation originate? A signal handler? A deadline? An error from a stage? A user action? Often more than one.
2. Who has authority to call cancel? The orchestrator? Each stage? An external watcher? Authority should be narrow and explicit.
3. What is the worst-case latency from cancel to fully stopped? The sum over all stages of their per-item work plus their select latency.

If you can answer these three before writing code, the implementation falls out naturally. If you cannot, the pipeline will have at least one cancellation bug.

Multi-Stage Pipelines: Threading the Context¶

A five-stage pipeline looks deceptively simple in the type signature:

out5 := stage5(ctx, stage4(ctx, stage3(ctx, stage2(ctx, stage1(ctx)))))

The single ctx flows into every stage. Every stage runs as a goroutine. All five goroutines select on the same ctx.Done(). When cancel fires, every one of them exits on its next iteration. The chain unwinds naturally: stage1 closes its output, stage2's range ends, stage2 closes its output, and so on.

A concrete example¶

package main

import (
    "context"
    "fmt"
    "strings"
    "time"
)

func gen(ctx context.Context, lines []string) <-chan string {
    out := make(chan string)
    go func() {
        defer close(out)
        for _, l := range lines {
            select {
            case out <- l:
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

func split(ctx context.Context, in <-chan string) <-chan string {
    out := make(chan string)
    go func() {
        defer close(out)
        for line := range in {
            for _, w := range strings.Fields(line) {
                select {
                case out <- w:
                case <-ctx.Done():
                    return
                }
            }
        }
    }()
    return out
}

func lower(ctx context.Context, in <-chan string) <-chan string {
    out := make(chan string)
    go func() {
        defer close(out)
        for w := range in {
            select {
            case out <- strings.ToLower(w):
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

func unique(ctx context.Context, in <-chan string) <-chan string {
    out := make(chan string)
    go func() {
        defer close(out)
        seen := map[string]struct{}{}
        for w := range in {
            if _, ok := seen[w]; ok {
                continue
            }
            seen[w] = struct{}{}
            select {
            case out <- w:
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

func print(ctx context.Context, in <-chan string) {
    for w := range in {
        select {
        case <-ctx.Done():
            return
        default:
        }
        fmt.Println(w)
    }
}

func main() {
    ctx, cancel := context.WithTimeout(context.Background(), 250*time.Millisecond)
    defer cancel()

    lines := []string{
        "The quick brown fox",
        "jumps over the lazy dog",
        "the quick brown fox",
    }

    print(ctx, unique(ctx, lower(ctx, split(ctx, gen(ctx, lines)))))
    fmt.Println("done:", ctx.Err())
}

Five stages: gen, split, lower, unique, print. The same ctx flows through all of them. If the 250 ms deadline fires before the pipeline completes, every stage exits on its next select; the pipeline drains and main prints the deadline error.

Why the same context for every stage works¶

ctx.Done() is a single channel. When cancel fires, it closes once. Every goroutine that selects on it gets the same notification at the same time. There is no need to "fork" or "split" the context per stage; the broadcast is already broadcast. The context is the wire; the stages are the listeners.

If you do want to give a stage a tighter deadline than the rest, derive a child:

func splitWithLimit(parent context.Context, in <-chan string) <-chan string {
    ctx, cancel := context.WithTimeout(parent, 50*time.Millisecond)
    defer cancel() // wait, this won't work — see below
    // ...
}

The trouble: the stage's goroutine outlives the function call. defer cancel() fires when the function returns, not when the goroutine returns. To make the per-stage deadline work, the cancel must live inside the goroutine and fire when the goroutine returns:

func splitWithLimit(parent context.Context, in <-chan string) <-chan string {
    out := make(chan string)
    go func() {
        ctx, cancel := context.WithTimeout(parent, 50*time.Millisecond)
        defer cancel()
        defer close(out)
        // ... loop with ctx ...
    }()
    return out
}

The pattern: context creation lives where it can be cancelled at the right time. For pipeline-level cancellation, that is the orchestrator. For per-stage deadlines, it is inside each stage's goroutine.

The "context per scope" design rule¶

When designing a multi-stage pipeline, choose the scope of each context deliberately. The rule of thumb:

• One root context at the top of the request or job. This is the source of authority for the entire operation.
• One pipeline context derived from the root, owned by the orchestrator. This is what every stage in the pipeline observes.
• One per-stage context only when a stage needs its own deadline or its own cancel reason. Otherwise, every stage shares the pipeline context.
• One per-operation context for each external call (DB, HTTP, RPC) with a tight timeout, derived from the pipeline context.

The picture:

   request context (deadline: 5s)
     └── pipeline context (errgroup)
           ├── stage 1
           ├── stage 2
           │     └── db query context (timeout: 200ms)
           └── stage 3
                 └── http call context (timeout: 500ms)

Every node is a context. Cancellation at any node cancels everything below. Deadlines are inherited and narrowed but not extended.

The trap: nesting more than necessary. Each derived context allocates memory and may add a goroutine watcher. A pipeline with 50 stages each deriving their own context for no reason is wasteful. Derive only when you need.

The other trap: nesting too little. A stage with a 5-minute work step inside a 5-second request context will be cancelled mid-step — that is fine, that is the contract. But if the stage's external call uses a separate background context, the call continues past the request deadline, defeating the purpose. Always derive call contexts from the pipeline context.

Cancel-on-Error with errgroup¶

The single biggest middle-level upgrade: golang.org/x/sync/errgroup. It bundles three things:

1. A WaitGroup-like join.
2. A context that cancels when any goroutine returns a non-nil error.
3. A return of the first error.

Basic usage¶

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

g, ctx := errgroup.WithContext(parentCtx)
g.Go(func() error {
    return stageA(ctx)
})
g.Go(func() error {
    return stageB(ctx)
})
if err := g.Wait(); err != nil {
    return err
}

What this does:

1. WithContext creates a child context and stores a cancel.
2. Each g.Go(f) spawns a goroutine running f.
3. The first f to return a non-nil error triggers cancel, which closes ctx.Done().
4. Every other goroutine running with ctx sees the cancel and exits.
5. g.Wait() blocks until every spawned goroutine returns, then returns the first error.

This is the answer to "how do I cancel my siblings when I fail?" and "how do I wait for all of them?" in one. It is the workhorse of every real Go pipeline.

Replacing manual WaitGroup + cancel with errgroup¶

Before:

ctx, cancel := context.WithCancel(parent)
defer cancel()
var wg sync.WaitGroup
var firstErr error
var errOnce sync.Once
wg.Add(2)
go func() {
    defer wg.Done()
    if err := stageA(ctx); err != nil {
        errOnce.Do(func() {
            firstErr = err
            cancel()
        })
    }
}()
go func() {
    defer wg.Done()
    if err := stageB(ctx); err != nil {
        errOnce.Do(func() {
            firstErr = err
            cancel()
        })
    }
}()
wg.Wait()
return firstErr

After:

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

The same semantics, an order of magnitude less code, and harder to misuse.

Combining errgroup with a worker pool¶

A common pattern: bounded concurrency over a stream of items, with cancellation on the first error.

g, ctx := errgroup.WithContext(parent)
sem := make(chan struct{}, 10) // limit to 10 concurrent
for _, item := range items {
    item := item
    select {
    case sem <- struct{}{}:
    case <-ctx.Done():
        break
    }
    g.Go(func() error {
        defer func() { <-sem }()
        return process(ctx, item)
    })
}
if err := g.Wait(); err != nil {
    return err
}

A semaphore (buffered channel) caps concurrency; errgroup cancels and joins.

errgroup.SetLimit (Go 1.20+)¶

The standard library now offers a built-in concurrency limit:

g, ctx := errgroup.WithContext(parent)
g.SetLimit(10)
for _, item := range items {
    item := item
    g.Go(func() error {
        return process(ctx, item)
    })
}
return g.Wait()

SetLimit(n) makes g.Go block until fewer than n goroutines are running. No manual semaphore needed.

errgroup and panics¶

If a function passed to g.Go panics, the panic propagates up through the goroutine and crashes the process by default. errgroup does not recover for you. If your work can panic, wrap it:

g.Go(func() (err error) {
    defer func() {
        if r := recover(); r != nil {
            err = fmt.Errorf("panic: %v", r)
        }
    }()
    return risky(ctx)
})

The recovered panic becomes the returned error; the rest of the group cancels via the normal cancel-on-error mechanism.

errgroup internals worth knowing¶

A short look at how errgroup.Group works. The relevant fields:

type Group struct {
    cancel  func(error)
    wg      sync.WaitGroup
    sem     chan token
    errOnce sync.Once
    err     error
}

• cancel is the cancel function from WithContext.
• wg is the wait group for the spawned goroutines.
• sem is the optional semaphore from SetLimit.
• errOnce ensures only the first error is stored and cancel is called once.
• err is the captured first error.

When you call g.Go(f):

1. If sem != nil, the call acquires a token (possibly blocking).
2. g.wg.Add(1) increments the wait group.
3. A goroutine is spawned that runs f().
4. On return, if f returned an error, errOnce.Do captures it and calls g.cancel.
5. The semaphore token is released; g.wg.Done() decrements.

When you call g.Wait():

1. g.wg.Wait() blocks until every spawned goroutine has returned.
2. The captured cancel is called with the error (so any external observers see cancellation).
3. The stored error is returned.

Key properties:

• The cancel is called as soon as the first error is captured, not at Wait. Siblings get the cancellation signal immediately.
• Wait always cancels, even if no error occurred (covered by Wait's call to g.cancel(g.err) — passing nil is fine).
• Only the first error is reported. Subsequent errors are discarded.

The implication for design: if you need all errors, do not rely on errgroup. Collect them in a slice via a channel:

errCh := make(chan error, len(items))
var wg sync.WaitGroup
for _, item := range items {
    item := item
    wg.Add(1)
    go func() {
        defer wg.Done()
        if err := process(ctx, item); err != nil {
            errCh <- err
        }
    }()
}
wg.Wait()
close(errCh)
var errs []error
for e := range errCh {
    errs = append(errs, e)
}
return errors.Join(errs...) // Go 1.20+

Verbose, but you get every error. errors.Join aggregates them into a single multi-error.

Deadline Propagation¶

A request arrives with a deadline. That deadline must reach every operation the pipeline performs — every database query, every HTTP call, every internal stage. Deadline propagation is the discipline of carrying the deadline forward without losing it.

The natural propagation path¶

func handler(w http.ResponseWriter, r *http.Request) {
    ctx := r.Context() // may already have a deadline from upstream
    if err := work(ctx); err != nil { ... }
}

func work(ctx context.Context) error {
    return db.QueryContext(ctx, "SELECT ...")
}

The deadline travels through ctx automatically. Each function passes ctx to its callees; the callees check ctx.Deadline() and ctx.Done() as needed.

Narrowing a deadline¶

A child operation should not exceed a slice of the parent deadline. To enforce:

func work(parent context.Context) error {
    ctx, cancel := context.WithTimeout(parent, 100*time.Millisecond)
    defer cancel()
    return innerCall(ctx)
}

If parent has 500 ms remaining and this child has 100 ms, the child cancels at 100 ms — whichever deadline fires first. WithTimeout does not extend the parent; it cannot. The parent always wins.

Why narrowing matters¶

Suppose a request has 1 second total. The pipeline does three sequential steps. Each step should not take more than, say, 300 ms; otherwise the request budget is blown by step 1 alone. Narrowing the per-step deadline enforces the budget:

for _, step := range steps {
    stepCtx, cancel := context.WithTimeout(ctx, 300*time.Millisecond)
    if err := step(stepCtx); err != nil {
        cancel()
        return err
    }
    cancel()
}

Or simpler with errgroup:

g, ctx := errgroup.WithContext(parent)
for _, step := range steps {
    step := step
    g.Go(func() error {
        c, cancel := context.WithTimeout(ctx, 300*time.Millisecond)
        defer cancel()
        return step(c)
    })
}
return g.Wait()

Detecting "no deadline" in a context¶

ctx.Deadline() returns (time.Time, bool). If the second value is false, there is no deadline. You can choose to enforce one:

deadline, ok := ctx.Deadline()
if !ok {
    // no upstream deadline; install our own
    var cancel context.CancelFunc
    ctx, cancel = context.WithTimeout(ctx, 30*time.Second)
    defer cancel()
} else if time.Until(deadline) < 100*time.Millisecond {
    // not enough budget to do useful work; fail fast
    return errors.New("not enough time budget")
}

The second branch is the defensive read: if you arrive with less than 100 ms left, do not even start. This prevents partial work that will be cancelled mid-way.

Deadlines that survive RPC boundaries¶

When the pipeline calls out to a remote service over gRPC, the gRPC client serialises the deadline in metadata. The remote server reconstructs a context with the same deadline. Cancellation also propagates: cancelling the client context closes the stream, the server sees it and cancels its own context.

HTTP/1.1 has no native deadline header, but the client's http.Request with a context will close the TCP connection on cancel, which the server's r.Context() sees as cancellation. This works for the most common case: a client that gives up on a request before the server is done.

For HTTP/2 the same TCP-level signalling applies, plus the framing supports stream RST that the server can detect explicitly.

Deadline budgets and the "deadline accounting" pattern¶

A request that arrives with a 1-second deadline cannot afford to spend the full second on one step if it has three steps. Deadline accounting is the practice of dividing the budget among the steps and enforcing the per-step limit.

Static allocation¶

The simplest scheme: assign a fixed budget to each step.

func handle(parent context.Context) error {
    if err := stepA(withTimeout(parent, 200*time.Millisecond)); err != nil {
        return err
    }
    if err := stepB(withTimeout(parent, 300*time.Millisecond)); err != nil {
        return err
    }
    return stepC(withTimeout(parent, 500*time.Millisecond))
}

func withTimeout(parent context.Context, d time.Duration) context.Context {
    ctx, _ := context.WithTimeout(parent, d)
    // (Cancel deliberately discarded for brevity in this snippet —
    //  real code uses `ctx, cancel := ...; defer cancel()`.)
    return ctx
}

Total budget: up to 1 second. Each step has its own ceiling. If step A is slow, it cancels at 200 ms regardless of how much parent budget is left.

Dynamic allocation¶

Sometimes you want to share unused budget. After step A finishes early, give the remainder to step B and C.

func handle(parent context.Context) error {
    parentDeadline, ok := parent.Deadline()
    if !ok {
        return errors.New("no deadline")
    }
    if err := step(parent, "A"); err != nil {
        return err
    }
    if time.Now().After(parentDeadline) {
        return context.DeadlineExceeded
    }
    if err := step(parent, "B"); err != nil {
        return err
    }
    return step(parent, "C")
}

Here each step uses the parent's full remaining budget. The early-check after each step prevents wasted work. This is the "best-effort" style; steps that finish early benefit the next steps.

Hybrid¶

A common production pattern: cap each step at a fraction of the parent budget.

func budget(parent context.Context, fraction float64) (context.Context, context.CancelFunc) {
    deadline, ok := parent.Deadline()
    if !ok {
        return parent, func() {}
    }
    remaining := time.Until(deadline)
    return context.WithTimeout(parent, time.Duration(float64(remaining)*fraction))
}

budget(parent, 0.5) gives the next step half of the remaining parent budget. After three steps each taking 0.5 of the remainder, you've used 7/8 of the original. This protects against any single step monopolising the budget.

When to skip deadlines entirely¶

Internal background tasks that should run "as long as it takes" often have no deadline. The cancellation primitive is still useful — they should stop on shutdown — but no timer. Use context.WithCancel(parent) without a timeout.

Upstream and Downstream Cancellation in Detail¶

Junior-level material introduced the two directions. The mechanism is the same channel close, but the design implications differ.

Downstream cancellation¶

The consumer at the end of the pipeline decides to stop. Examples:

• The HTTP client disconnected; r.Context() is cancelled.
• The user pressed Ctrl-C; signal.NotifyContext fires.
• A separate timer hit a deadline.

The cancel is at the bottom of the chain; the signal travels back up the wire. Upstream stages see ctx.Done() and exit, freeing the resources they held.

Downstream cancellation is the "normal" case. It is the model the standard library is built around: the caller cancels, the callee sees it.

Upstream cancellation¶

A stage at the top of the pipeline decides to stop. Examples:

• The source database connection broke; the source stage cancels.
• The fan-out controller saw N errors and decides to bail.
• A configuration reload triggered the pipeline owner to cancel.

The cancel is at the top; downstream stages see ctx.Done() and exit. The remaining values in the channels are either drained by downstream or dropped on cancel.

Upstream cancellation also flows through ctx. The difference is policy: who has the authority to call cancel? In simple pipelines, only the orchestrator holds the cancel function. In richer pipelines, stages may hold a cancel and trigger it on error.

Diagonal: middle stage cancels¶

A middle stage encounters an error and wants to cancel the rest:

func middle(ctx context.Context, cancel context.CancelFunc, in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for v := range in {
            if v < 0 {
                cancel() // upstream and downstream both stop
                return
            }
            select {
            case out <- v * 2:
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

By calling cancel(), the middle stage triggers cancellation that flows to both producers (upstream) and consumers (downstream). The single ctx.Done() is observed by all of them.

This is why a context model is more flexible than a one-directional done channel: any participant can trigger cancellation, and the same wire delivers the message in either direction.

Authority and trust: who is allowed to cancel?¶

In small pipelines, the orchestrator (the function that built the pipeline) is the only one who calls cancel. In larger pipelines, multiple participants may have authority. Three patterns:

Orchestrator-only authority¶

ctx, cancel := context.WithCancel(parent)
defer cancel()
go stage(ctx, ...) // stage does not see cancel

Stages do not receive the cancel function; they only observe ctx.Done(). This is the strictest model: cancellation is centralised. If a stage encounters an error, it must signal up through its return value or an error channel; the orchestrator decides whether to cancel.

Shared authority¶

ctx, cancel := context.WithCancel(parent)
defer cancel()
go stage(ctx, cancel, ...) // stage can trigger cancel

Stages receive the cancel and may trigger it on internal errors. This is the errgroup model implicitly: every g.Go function effectively holds the cancel because returning a non-nil error invokes it.

Shared authority is convenient but it expands the trust surface. Any stage can shut down the whole pipeline; bugs in any stage can lead to spurious cancellations. Document carefully who is allowed to cancel and on what conditions.

Hierarchical authority¶

ctx, cancel := context.WithCancel(parent)
defer cancel()

subCtx, subCancel := context.WithCancel(ctx)
defer subCancel()
go subStage(subCtx, subCancel, ...) // can only cancel the sub-tree

The sub-stage cancels its own sub-context but cannot affect siblings or the parent. This is the structured-concurrency pattern; explored in depth at senior level.

The choice of authority model is an architectural decision. For most middle-level pipelines, errgroup's implicit shared authority (every goroutine can return an error, which cancels) is the right balance.

Fan-Out with Coordinated Cancellation¶

Fan-out: one producer feeds N workers, each processing items in parallel. The output may merge back (fan-in) or remain N separate streams. Cancellation must reach every worker.

Bounded fan-out with errgroup and SetLimit¶

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

Eight workers maximum. Each runs process(ctx, item). If any returns an error, errgroup cancels ctx; remaining workers see ctx.Done() and exit (assuming process respects context, which it must).

Fan-out from a channel source¶

func fanOut(ctx context.Context, in <-chan Item, workers int) error {
    g, ctx := errgroup.WithContext(ctx)
    for i := 0; i < workers; i++ {
        g.Go(func() error {
            for item := range in {
                if err := process(ctx, item); err != nil {
                    return err
                }
                if ctx.Err() != nil {
                    return ctx.Err()
                }
            }
            return nil
        })
    }
    return g.Wait()
}

The shared input channel in feeds all workers. Each worker ranges over in; when in closes (upstream is done), each worker's range exits and the worker returns. On error or cancel, the ctx.Err() check inside the loop catches the cancellation.

Note: for item := range in does not natively check ctx.Done(). If in never closes and we cancel, workers may block on the next receive. The cure is to make the receive itself a select:

for {
    select {
    case <-ctx.Done():
        return ctx.Err()
    case item, ok := <-in:
        if !ok {
            return nil
        }
        if err := process(ctx, item); err != nil {
            return err
        }
    }
}

This is the canonical worker loop. Always.

What happens to the source on cancel¶

If the source is a separate goroutine producing into in, it must also respect ctx. The same template:

func source(ctx context.Context, items []Item) <-chan Item {
    out := make(chan Item)
    go func() {
        defer close(out)
        for _, item := range items {
            select {
            case out <- item:
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

On cancel, the source returns, closes out, and the worker ranges exit.

Fan-out with results¶

When fan-out workers produce results, you need a single result channel and a coordinated closer.

func fanOutWithResults(ctx context.Context, items []Item, workers int) ([]Result, error) {
    g, ctx := errgroup.WithContext(ctx)
    in := make(chan Item)
    out := make(chan Result)

    // feeder
    g.Go(func() error {
        defer close(in)
        for _, item := range items {
            select {
            case in <- item:
            case <-ctx.Done():
                return ctx.Err()
            }
        }
        return nil
    })

    // workers
    var wg sync.WaitGroup
    wg.Add(workers)
    for i := 0; i < workers; i++ {
        g.Go(func() error {
            defer wg.Done()
            for item := range in {
                r, err := process(ctx, item)
                if err != nil {
                    return err
                }
                select {
                case out <- r:
                case <-ctx.Done():
                    return ctx.Err()
                }
            }
            return nil
        })
    }

    // closer
    go func() {
        wg.Wait()
        close(out)
    }()

    // collector
    var results []Result
    for r := range out {
        results = append(results, r)
    }
    if err := g.Wait(); err != nil {
        return nil, err
    }
    return results, nil
}

func process(ctx context.Context, i Item) (Result, error) { return Result{}, nil }

type Item struct{}
type Result struct{}

Trace:

1. The feeder pushes items into in until it is done or cancelled, then closes in.
2. Workers range in, process, push results into out.
3. The closer waits for all workers (via wg) and closes out.
4. The collector ranges over out and accumulates results.
5. After the range exits, g.Wait returns the first error (or nil).

On cancellation, every stage's select picks <-ctx.Done() and returns. The feeder closes in; workers' ranges end; the closer closes out; the collector's range ends. Clean cascade.

The two wait groups (g.wg from errgroup, plus the manual wg for closing out) coexist. The manual wg is only for ordering the close; the errgroup g.wg is for the overall error join.

Fan-In with Drain Semantics¶

Fan-in: N producers feed into a single output channel. The standard pattern:

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

Each input has a forwarder goroutine. They all write into out. A separate closer waits for all of them and closes out.

Why the closer is separate¶

Multiple goroutines cannot safely close the same channel — the second close panics. The convention is: any goroutine that writes to a shared channel does not close it; one designated closer (or a WaitGroup-coordinated closer) does.

What does cancel mean for fan-in?¶

When cancel fires:

1. Every forwarder's select may pick <-ctx.Done() instead of the send. The forwarder returns.
2. defer wg.Done() runs for each forwarder.
3. The closer's wg.Wait() unblocks.
4. close(out) runs.
5. Downstream ranges over out exit.

The forwarders may have values in their input that are not delivered. That is the price of cancellation — drop the in-flight work. If you need to deliver every item, you need a different protocol (covered at senior level).

A fan-in that drops cancelled inputs cleanly¶

func mergeCancellable(ctx context.Context, ins ...<-chan int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup
    wg.Add(len(ins))
    for _, in := range ins {
        go func(in <-chan int) {
            defer wg.Done()
            for {
                select {
                case <-ctx.Done():
                    return
                case v, ok := <-in:
                    if !ok {
                        return
                    }
                    select {
                    case out <- v:
                    case <-ctx.Done():
                        return
                    }
                }
            }
        }(in)
    }
    go func() {
        wg.Wait()
        close(out)
    }()
    return out
}

The full template, with both the receive and the send guarded. This is what production fan-in code looks like.

Context Trees in Real Applications¶

A real application has a context tree. The root is context.Background() at the top of main. Below it:

• A signal-cancellable context for the whole process.
• A per-request context for each HTTP/gRPC handler.
• Per-stage contexts within each pipeline.
• Per-operation contexts for database queries with their own timeouts.

Background
  └── signal-cancel (SIGTERM, SIGINT)
       └── http-request-1 (deadline from header)
       │     ├── pipeline (errgroup)
       │     │     ├── stage A
       │     │     └── stage B
       │     │           └── db.QueryContext (timeout 100ms)
       │     └── http-call-to-backend
       └── http-request-2
             └── ...

Cancelling at any node cancels everything below. A SIGTERM cancels every request. A request deadline cancels its pipeline. A stage error cancels its siblings.

Owning the cancel function¶

The function that creates a node also owns the cancel function. By convention, the owner calls defer cancel(). Passing the cancel to a callee is allowed but should be explicit and documented — it transfers ownership of the cancel.

Long-lived background contexts¶

Some pipelines outlive any single request — background workers, reconciliation loops, scheduled tasks. They use a long-lived context, often the signal-cancel context, so they stop on shutdown:

func runWorker(ctx context.Context) {
    for {
        select {
        case <-ctx.Done():
            return
        case <-time.After(time.Minute):
            doScheduledWork(ctx)
        }
    }
}

The worker exits cleanly on SIGTERM because ctx is the signal-cancellable context from main.

context.AfterFunc and WithCancelCause¶

Go 1.21 added two useful helpers.

context.AfterFunc¶

Registers a callback that runs when a context is cancelled.

stop := context.AfterFunc(ctx, func() {
    metrics.IncCancellations()
    log.Println("pipeline cancelled")
})
defer stop()

AfterFunc runs the callback in a goroutine when ctx is cancelled. stop() unregisters the callback (and returns true if it was unregistered before it fired). This is cleaner than spawning a watcher goroutine yourself.

Use it for side effects on cancellation: metrics, logs, releasing external resources. Do not use it for the main shutdown logic — that should be in the stages that use ctx.

context.WithCancelCause¶

Cancels with a custom error reason.

ctx, cancel := context.WithCancelCause(parent)
defer cancel(nil)

// when an error occurs:
cancel(fmt.Errorf("upstream failed: %w", err))

// elsewhere:
if err := context.Cause(ctx); err != nil {
    return fmt.Errorf("pipeline aborted: %w", err)
}

ctx.Err() still returns context.Canceled, but context.Cause(ctx) returns the original error. This is the cleanest way to surface "why we cancelled" through a pipeline.

errgroup does not yet integrate with WithCancelCause (as of Go 1.22), but you can compose the two:

ctx, cancelCause := context.WithCancelCause(parent)
defer cancelCause(nil)

g, gctx := errgroup.WithContext(ctx)
g.Go(func() error {
    if err := work(gctx); err != nil {
        cancelCause(err) // remember why
        return err
    }
    return nil
})
err := g.Wait()
if err != nil {
    return fmt.Errorf("cause: %w", context.Cause(ctx))
}
return nil

Verbose, but the error reason survives the cancellation cascade.

Cancellation in pipelines that read from streams¶

A common shape: the source reads from an io.Reader, a network connection, or a database cursor. Cancellation of the read is harder than cancellation of a select.

Reading from io.Reader with cancellation¶

The naive code:

buf := make([]byte, 4096)
for {
    n, err := r.Read(buf)
    if err != nil {
        return err
    }
    process(buf[:n])
}

r.Read blocks until data arrives or the underlying source errors. It does not check ctx.Done(). If the source is a network connection, you can call conn.SetReadDeadline(time.Now()) from a watcher goroutine to interrupt the read.

go func() {
    <-ctx.Done()
    conn.SetReadDeadline(time.Now())
}()
for {
    n, err := conn.Read(buf)
    if err != nil {
        if errors.Is(err, os.ErrDeadlineExceeded) {
            return ctx.Err()
        }
        return err
    }
    process(buf[:n])
}

The watcher goroutine sets the deadline to the past, causing the next Read to return os.ErrDeadlineExceeded. The caller maps that to the context error and returns.

For pipes and files that do not support deadlines, you cannot interrupt a blocked Read. The least-bad option is to close the underlying file from the watcher:

go func() {
    <-ctx.Done()
    f.Close()
}()

f.Close makes the next read return an error. The pipeline sees the error and exits. The downside: the file is closed even if you wanted to reuse it. Treat this as a forceful cancellation, not graceful.

Reading from a database cursor¶

sql.Rows is cancellable if the parent query used QueryContext. Once the context cancels, the driver aborts the query and the next rows.Next() returns false; rows.Err() returns the cancellation reason.

rows, err := db.QueryContext(ctx, query)
if err != nil {
    return err
}
defer rows.Close()
for rows.Next() {
    var v int
    if err := rows.Scan(&v); err != nil {
        return err
    }
    process(v)
}
return rows.Err()

rows.Err() returns nil on clean completion, context.Canceled or context.DeadlineExceeded on cancellation, or a driver error.

Reading from an HTTP response body¶

req, _ := http.NewRequestWithContext(ctx, "GET", url, nil)
resp, err := http.DefaultClient.Do(req)
if err != nil {
    return err
}
defer resp.Body.Close()
buf := make([]byte, 4096)
for {
    n, err := resp.Body.Read(buf)
    if err != nil {
        if err == io.EOF {
            return nil
        }
        return err
    }
    process(buf[:n])
}

If ctx cancels mid-stream, resp.Body.Read returns context.Canceled (or wraps it). The transport closes the underlying connection.

The pattern: always create the request with NewRequestWithContext, never with the bare NewRequest. The latter has no cancellation hook.

Reading from a streaming gRPC call¶

gRPC server streams are inherently cancellable. The context passed in is closed when the client cancels or the deadline fires. Inside the server handler:

func (s *server) Stream(req *pb.Req, stream pb.Service_StreamServer) error {
    ctx := stream.Context()
    for {
        select {
        case <-ctx.Done():
            return ctx.Err()
        case event := <-source:
            if err := stream.Send(event); err != nil {
                return err
            }
        }
    }
}

The Send itself returns an error if the client cancelled; the explicit <-ctx.Done() is an extra safety net.

Cancellable I/O¶

A pipeline that calls out to I/O — database, HTTP, file, gRPC — must use the context-aware variant of every call. Otherwise cancellation only stops the pipeline's control flow; the I/O calls themselves continue.

Database¶

rows, err := db.QueryContext(ctx, query, args...)

If ctx cancels mid-query, the driver issues a cancel to the database (Postgres: PQcancel; MySQL: kill query) and returns context.Canceled. Always use QueryContext, ExecContext, BeginTx (which accepts a context).

HTTP¶

req, _ := http.NewRequestWithContext(ctx, "GET", url, nil)
resp, err := http.DefaultClient.Do(req)

If ctx cancels mid-request, the transport closes the connection. The response body's Read returns context.Canceled.

gRPC¶

resp, err := client.Method(ctx, req)

If ctx cancels, the framework sends a RST_STREAM and the server's ctx also cancels. Symmetric cancellation across the wire.

File I/O¶

os.File does not accept a context directly. To make file reads cancellable, you have two options:

1. Use io.Reader with a select-wrapped Read (as shown in the junior file). The underlying read still completes; only the calling goroutine sees the cancel. The OS may eventually close the FD when the parent process exits or when you call f.Close().
2. Use os/exec.CommandContext (for child processes) which kills the child on cancel.

In general, file I/O is the weak link in cancellable pipelines. Be aware of it.

Network connections¶

net.Dialer.DialContext accepts a context; net.Conn.SetDeadline accepts a time. The latter is useful for retro-fitting cancellation: a watcher goroutine watches ctx.Done() and calls conn.SetReadDeadline(time.Now()) to force an I/O error.

go func() {
    <-ctx.Done()
    conn.SetReadDeadline(time.Now())
}()

The next conn.Read returns immediately with a timeout error. The watcher exits; the read function returns; the caller sees cancellation.

Cancellation in Worker Pools¶

A worker pool: N long-lived workers pull from a job channel, process, and may emit results.

Template¶

type Pool struct {
    jobs    chan Job
    results chan Result
    ctx     context.Context
    cancel  context.CancelFunc
    wg      sync.WaitGroup
}

func NewPool(parent context.Context, workers int, buf int) *Pool {
    ctx, cancel := context.WithCancel(parent)
    p := &Pool{
        jobs:    make(chan Job, buf),
        results: make(chan Result, buf),
        ctx:     ctx,
        cancel:  cancel,
    }
    p.wg.Add(workers)
    for i := 0; i < workers; i++ {
        go p.worker()
    }
    go func() {
        p.wg.Wait()
        close(p.results)
    }()
    return p
}

func (p *Pool) worker() {
    defer p.wg.Done()
    for {
        select {
        case <-p.ctx.Done():
            return
        case j, ok := <-p.jobs:
            if !ok {
                return
            }
            r := process(p.ctx, j)
            select {
            case p.results <- r:
            case <-p.ctx.Done():
                return
            }
        }
    }
}

func (p *Pool) Submit(j Job) error {
    select {
    case p.jobs <- j:
        return nil
    case <-p.ctx.Done():
        return p.ctx.Err()
    }
}

func (p *Pool) Close() {
    close(p.jobs)
    p.cancel()
}

func (p *Pool) Results() <-chan Result {
    return p.results
}

type Job struct{}
type Result struct{}

func process(ctx context.Context, j Job) Result { return Result{} }

Walking through the cancellation paths:

• Submit is cancellable: if the pool is shutting down, it returns an error rather than blocking on the jobs channel.
• Each worker selects on ctx.Done() and on the jobs channel; on cancel, it returns.
• The producer (whoever calls Submit) is responsible for not submitting after Close.
• Close cancels the context (so workers stop accepting new jobs) and closes the jobs channel (so workers exit their loop).
• The closer goroutine waits for all workers and closes results.

This is the minimal pattern. Real pools add metrics, retries, timeouts, and back-pressure.

Graceful close vs forceful close¶

Close as written stops the pool. But what about in-flight jobs? The workers may be in process(p.ctx, j) when cancel fires. If process respects context, it returns mid-work. If not, it runs to completion and then the worker sees the cancel.

For graceful shutdown — finish current jobs but accept no new ones — split the two signals:

func (p *Pool) Drain() {
    close(p.jobs) // stop accepting new
    p.wg.Wait()   // wait for workers to finish in-flight
    p.cancel()    // tidy up
}

func (p *Pool) Stop() {
    p.cancel()    // cancel everything immediately
    close(p.jobs) // close jobs so workers exit
    p.wg.Wait()
}

Drain is graceful; Stop is forceful. The choice depends on the SLA. Most servers expose both.

Graceful shutdown protocols¶

A long-running server has multiple kinds of work in flight when shutdown begins:

• HTTP handlers serving requests.
• Background workers processing queues.
• Periodic tasks (metrics, reconciliation, cleanup).
• Outbound connections to dependencies.

A graceful shutdown stops accepting new work, finishes (or politely cancels) in-flight work, releases resources, and exits within a budget. The components:

Stage 1: stop accepting new work¶

Close listening sockets, stop queue consumers, set a "draining" flag on the HTTP server. The server stops accepting new connections; existing connections continue.

srv := &http.Server{...}
go srv.ListenAndServe()
// ... on shutdown:
shutdownCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
srv.Shutdown(shutdownCtx)

http.Server.Shutdown does exactly this: stops the listener, waits for in-flight handlers to return, optionally cancels them on shutdownCtx.Done().

Stage 2: cancel in-flight work that respects context¶

The root context cancels; every handler that uses r.Context() exits. Pipeline stages see ctx.Done() and exit. Workers see ctx.Done() and exit.

rootCtx, rootCancel := signal.NotifyContext(context.Background(), os.Interrupt)
defer rootCancel()
// ... pipeline uses rootCtx ...

The single cancellation cascades to every consumer of the root context.

Stage 3: wait with a deadline¶

Some work may not respect cancellation (legacy code, third-party libraries). For these, you cannot do anything except wait with a deadline.

done := make(chan struct{})
go func() {
    cleanup()
    close(done)
}()
select {
case <-done:
case <-time.After(10 * time.Second):
    log.Println("cleanup timed out")
}

After the deadline, abandon the work and proceed to exit. Better to lose some work than to hang the shutdown.

Stage 4: process exit¶

After all the cancellations and joins, main returns. The OS reaps the process.

Example: a server with a worker pool¶

func main() {
    rootCtx, rootCancel := signal.NotifyContext(context.Background(),
        os.Interrupt, syscall.SIGTERM)
    defer rootCancel()

    pool := NewPool(rootCtx, 16, 256)

    srv := &http.Server{
        Addr:    ":8080",
        Handler: handlerWith(pool),
        BaseContext: func(net.Listener) context.Context {
            return rootCtx
        },
    }

    go func() {
        if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
            log.Println("server:", err)
        }
    }()

    <-rootCtx.Done()
    log.Println("shutdown initiated")

    shutdownCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
    defer cancel()

    // Stop accepting new HTTP requests
    if err := srv.Shutdown(shutdownCtx); err != nil {
        log.Println("server shutdown:", err)
    }

    // Drain the worker pool
    pool.Drain()

    log.Println("shutdown complete")
}

func handlerWith(p *Pool) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}) }

The flow:

1. SIGTERM arrives; rootCtx cancels.
2. main proceeds to shutdown.
3. srv.Shutdown stops the listener and waits for in-flight handlers (which also see rootCtx via BaseContext).
4. pool.Drain waits for in-flight jobs to finish, then cancels remaining workers.
5. main returns.

If any of these steps takes too long, the 30-second shutdown deadline forces an exit. The process dies, and Kubernetes or systemd kills it harder if it does not exit by then.

This shape — signal-cancellable root, BaseContext propagation, Shutdown with deadline, then pool drain — is the standard production server in Go.

Pipeline Lifecycle Diagram¶

            +----------------+
            | parent context | (from main, request, scheduler)
            +-------+--------+
                    |
                    v
             +---------------+
             | pipeline ctx  |  (WithCancel or errgroup)
             +------+--------+
                    |
       +------------+------------+
       |            |            |
       v            v            v
   stage 1      stage 2      stage 3
   (goroutine)  (goroutine)  (goroutine)
       |            |            |
       +--- ctx.Done() --- shared close ---
                    |
              cancel triggered
              (deadline, error,
               signal, manual)