Pipelines under cancellation: guarded sends, closing discipline, and the canonical goroutine leak

Stages and channels — the shape

A pipeline is goroutines connected by channels: each stage receives values upstream, transforms them, and sends downstream. The 2014 blog version used a per-pipeline done channel; the modern form is context:

func parse(ctx context.Context, in <-chan []byte) <-chan Record {
	out := make(chan Record)
	go func() {
		defer close(out) // this stage owns out — only the sender closes
		for raw := range in {
			rec, err := decode(raw)
			if err != nil {
				continue // or: report on an error path, below
			}
			select {
			case out <- rec:
			case <-ctx.Done():
				return // abandon the send; defer closes out
			}
		}
	}()
	return out
}

Two structural facts make the pattern composable. The stage owns its output channel: it creates it, it alone sends, and defer close(out) signals “no more values” downstream — which is what lets the next stage use a plain for range. And the function returns the receive-only channel immediately; the pipeline is wired before any data flows.

The canonical leak: the unguarded send

Strip the select and the stage reads cleaner — out <- rec — and is wrong. A channel send blocks until a receiver is ready. The moment downstream stops reading — it crashed, returned early on bad input like the Hook, or simply decided it had enough — that send blocks forever. The goroutine is parked on chan send, the GC cannot touch it (a parked goroutine is a root, pinning its stack and every record it references), and no error surfaces anywhere. This is the single most common real-world goroutine leak, and the rule that prevents it is mechanical: every send into a pipeline channel sits inside a select with a ctx.Done() case. Receives in a for range are safe only because the sender’s defer close is guaranteed — if a stage can exit without closing its output, downstream ranges leak the same way.

Closing discipline and fan-in

When you first wire a fan-in, the question “who closes the merged channel?” is easy to get wrong — every sender feels like a candidate, but any one of them closing early triggers a panic on the others. The closing rule has no exceptions: only the sender closes a channel, because a send on a closed channel panics and a double close panics — the receiver can never know it is safe. Single-sender stages get it for free with defer close(out). Fan-in — N workers sending into one merged channel — is where discipline gets a shape: nobody may close until all senders are done, so you count them:

func merge(ctx context.Context, ins ...<-chan Record) <-chan Record {
	out := make(chan Record)
	var wg sync.WaitGroup
	for _, in := range ins {
		wg.Add(1)
		go func() { // one forwarder per input
			defer wg.Done()
			for rec := range in {
				select {
				case out <- rec:
				case <-ctx.Done():
					return
				}
			}
		}()
	}
	go func() { wg.Wait(); close(out) }() // WaitGroup-then-close, in its own goroutine
	return out
}

The wg.Wait(); close(out) pair runs in a separate goroutine so merge returns immediately — the close happens exactly once, exactly after the last forwarder exits, whether that exit was “input drained” or “ctx cancelled mid-send”.

Drain or abandon — who unblocks whom on cancel

Cancellation tears the pipeline down from both ends, and you should know which mechanism you are relying on. Abandoning is the default above: every sender guards its own send with ctx.Done, so when ctx is cancelled each stage unblocks itself and returns; values in flight are dropped. Draining is the alternative: the consumer keeps for range-ing until the channel closes, so upstream finishes naturally and nothing is dropped — at the cost of processing (or at least receiving) everything already produced. Draining without guarded sends is how people convince themselves the guards are optional; it holds only while every consumer is disciplined, which is exactly what the Hook’s one early return broke. The robust posture: guard every send anyway, and choose draining only when the data must not be lost (then the drain loop itself needs a bound — count or deadline — or a slow producer turns shutdown into a hang).

Buffers: what they buy, honestly

Making stage channels buffered — make(chan Record, 1024) — feels like a throughput fix. Be precise about what it does. A buffer absorbs jitter: short bursts, GC pauses, a momentarily slow consumer; producers keep moving instead of handing off in lockstep, and unbuffered handoff latency (~hundreds of nanoseconds per send-receive rendezvous) stops dominating tiny work items. What a buffer does not fix is imbalance: if a stage produces at 12k/s and the next consumes at 9k/s, a 1024-slot buffer is full after ~340 ms and the pipeline runs at exactly 9k/s — same as unbuffered, plus 1024 records of extra memory and staleness. Steady-state throughput is the slowest stage, full stop. Buffers also delay leak symptoms: an unguarded send into a big buffer fails only when the buffer fills, often minutes after the consumer died — your 02:13 bug surfaces at 02:47 with a stack that no longer points at the cause.

Errors and observability

A stage that can fail needs an error path. Two workable designs. Per-stage error channel: the stage sends errors into errc and the orchestrator selects over it — explicit, but every consumer must drain errc or you have built a second leak. errgroup-wrapped stages — usually the better default: each stage runs as g.Go, returns its error, and the group ctx cancels every other stage’s guarded sends; g.Wait gives you the first failure after the whole pipeline has unwound. For visibility, instrument the channels themselves: a periodic gauge of len(ch) against cap(ch) per stage shows where the queue sits — a buffer pinned at capacity marks the bottleneck stage, a buffer pinned at zero marks starvation upstream, and runtime.NumGoroutine() trending up while traffic is flat is the leak alarm from the previous lesson. Cheap numbers, and they turn “the batch is slow” into “stage 3 is the wall”.