Go / Advanced: Safety, Performance, and Service Design / Lesson 14

Cancellation and Bounded Pipelines

Construct worker pipelines whose queues, goroutines, cancellation, and closing responsibilities all have explicit bounds.

About 5 min readContent reviewed 2026-10-25

On this page

Lesson goals

Assign stopping and joining responsibility to each stage.
Make sends and receives cancellation-aware.
Bound worker count and preserve error/commit policy.

Each stage owns a lifetime

A pipeline connects stages that transform or transport values. Define who starts each stage, who closes each output, what happens after downstream rejection, and who waits for every goroutine. A stage that can block forever after its consumer exits is incomplete even when its normal-path output is correct.

Bound work admission as well as worker count. A fixed pool limits active workers, but an unbounded pending queue can still exhaust memory. Decide whether admission blocks with cancellation, rejects, or uses a documented drop policy. Resource limits are part of the API’s behavior, not merely a tuning parameter.

An abandoned send needs an exit path

The first program starts an unbuffered producer under a context already cancelled. main waits for completion before trying to receive, so the send is never ready and the cancellation branch must end the worker. The worker closes its output and then signals done. No timing claim or random selection between two ready cases is needed.

This controlled setup demonstrates the missing-receiver case safely. In a live pipeline, either a send or cancellation may become ready, so the contract should tolerate the allowed final send. Do not make an absolute priority promise from the order of select cases. Keep the returned channel read-only to clarify consumer permissions.

go A cancelled producer with an explicit join

package main

import (
	"context"
	"fmt"
)

func produce(ctx context.Context) (<-chan int, <-chan struct{}) {
	output := make(chan int)
	done := make(chan struct{})
	go func() {
		defer close(done)
		defer close(output)
		select {
		case output <- 7:
		case <-ctx.Done():
			return
		}
	}()
	return output, done
}

func main() {
	ctx, cancel := context.WithCancel(context.Background())
	cancel()
	output, done := produce(ctx)
	<-done
	_, ok := <-output
	fmt.Println("closed without value", !ok)
}

Expected output

closed without value true

No receiver is available before done, so cancellation ends the producer and closes its output.

Coordinate closing after every sender

The next program has two workers, an unbuffered job channel, and a results channel. The producer owns job closure; a coordinator closes results only after both workers finish. The collector drains results and sums squares, so worker scheduling changes result order but not the fixed aggregate. Context-aware sends prevent an abandoned downstream receiver from stranding workers when cancellation is used.

All goroutines have a path to completion. WaitGroup additions occur before launching workers, and each worker defers Done. A production operation must also wait for the producer and coordinator according to its ownership contract on early-return paths; the successful complete drain here establishes their required progress without sleeping.

go A bounded two-worker pipeline

package main

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

func main() {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	jobs := make(chan int)
	results := make(chan int)
	producerDone := make(chan struct{})
	go func() {
		defer close(producerDone)
		defer close(jobs)
		for n := 1; n <= 4; n++ {
			select {
			case jobs <- n:
			case <-ctx.Done():
				return
			}
		}
	}()
	var workers sync.WaitGroup
	workers.Add(2)
	for i := 0; i < 2; i++ {
		go func() {
			defer workers.Done()
			for n := range jobs {
				select {
				case results <- n * n:
				case <-ctx.Done():
					return
				}
			}
		}()
	}
	coordinatorDone := make(chan struct{})
	go func() { defer close(coordinatorDone); workers.Wait(); close(results) }()
	total := 0
	for result := range results {
		total += result
	}
	<-producerDone
	<-coordinatorDone
	fmt.Println("squares", total)
}

Expected output

squares 30

The four squares sum to thirty; all producer, worker, and coordinator lifetimes are observed.

Define first error and partial success

A concurrent operation needs a policy for errors: stop on the first reported failure, collect several, or continue independent work. Cancellation should stop unneeded work, while the owner still joins every started goroutine. If effects are already committed, returning an error must not falsely imply nothing happened. Document partial results or transactional boundaries.

Retries need separate idempotency reasoning. Repeating a timed-out action may duplicate an effect even when no response arrived. Carry stable operation identifiers where the protocol supports them, and keep retry budgets bounded by context and application policy. A worker pool is not a transaction manager.

Test shutdown as a normal operation

Test cancellation before admission, while the queue is full, while a worker sends, and after downstream failure. Use completion channels, barriers, and bounded test deadlines to observe progress. Goroutine counts alone are a noisy signal because runtime and test infrastructure may create goroutines independently. Inspect goroutine profiles when an owned worker fails to finish.

Do not close a shared work channel from arbitrary cancellation handlers while senders remain active. Let its designated owner close after sending ends, or use context signaling without closing that channel. Buffers and recover calls do not replace this ownership protocol. Review cleanup order and retained references as part of every early-return path.

Common errors and fixes

Downstream returns early and upstream never ends.

Why it happens: A send has no cancellation path or no owner waits.
How to fix it: Add cancellation-aware communication and join all stages.
Prevent it next time: Test early consumer rejection explicitly.

Cancelling causes send-on-closed panics.

Why it happens: A cancellation handler closed a channel with active senders.
How to fix it: Keep closing with the designated sender/coordinator.
Prevent it next time: Separate cancellation signaling from data-channel ownership.

A fixed worker count still consumes unbounded memory.

Why it happens: Admission or result buffering is unbounded.
How to fix it: Bound queue size and define overload behavior.
Prevent it next time: Measure pending work as well as active workers.

Check your understanding

1. A caller needs only the first successful result. What must it do before returning from an operation that owns all workers?

Show worked solution

Signal cancellation to stop remaining work, ensure every blocking stage observes it, and wait for all owned goroutines before releasing shared resources or returning. Preserve the chosen result according to its ownership contract. Do not merely stop reading results; that can block other workers forever. A bounded join failure should be diagnosed as a lifecycle defect, not hidden by leaking workers.

2. Why does a results buffer of one thousand not prove that a pipeline cannot leak?