🛠️ Project: A Bounded Concurrent Fan-Out Worker with errgroup

Project Brief: Fan-Out That Does Not Melt the Backend

Picture a handler that receives 500 IDs, and each ID needs a call to an upstream service or a database query. Doing that serially is a latency disaster. Doing it with 500 bare go statements is a different disaster: you open 500 connections at once, the pool chokes, and one slow dependency takes down the whole request.

[!NOTE] Warm-up recall from the last lesson: goroutines are cheap but every one of them needs an exit path, and channels express handoff while context carries the deadline. This project exists because "cheap" is not "free": ==your concurrency limit is a capacity decision, not a style choice==.

The deliverable is a small reusable package: a Run function that fans out over N tasks, runs at most limit of them at a time, cancels everything on the first error, returns results in input order, and stays clean under go test -race.

Why it matters: this is the shape of most real fan-out work in a Go service. Search hydration, batch enrichment, parallel cache reads, fan-out to a gRPC backend. Once you can write this function correctly, you can write it correctly anywhere.

flowchart LR
    In["📥 tasks N"] --> Sched["🎚️ bounded scheduler (limit)"]
    Sched --> W1["⚙️ worker"]
    Sched --> W2["⚙️ worker"]
    Sched --> W3["⚙️ worker"]
    W1 --> Res["📊 results[i]"]
    W2 --> Res
    W3 --> Res
    W1 -.->|first error| Cancel["🛑 cancel group ctx"]
    W2 -.-> Cancel
    W3 -.-> Cancel

Requirements

  1. Exported API: func Run(ctx context.Context, tasks []Task, limit int, fn func(context.Context, Task) (Result, error)) ([]Result, error)
  2. Bounded. Never more than limit calls to fn execute at once. limit < 1 returns an error, not a panic.
  3. Fail fast. The first non-nil error cancels sibling work. Run returns that error, and errors.Is must find the original through it.
  4. Context-respecting. If the caller's ctx is canceled or its deadline passes, Run returns promptly with ctx.Err(); no worker keeps running.
  5. Ordered results. results[i] corresponds to tasks[i], and on error you return nil results.
  6. Race-free. go test -race ./... passes. No goroutine outlives Run.

Starter shape to grow from:

package fanout

import (
	"context"
	"fmt"

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

func Run(ctx context.Context, tasks []Task, limit int,
	fn func(context.Context, Task) (Result, error)) ([]Result, error) {

	if limit < 1 {
		return nil, fmt.Errorf("fanout: limit must be >= 1, got %d", limit)
	}

	results := make([]Result, len(tasks)) // pre-sized: one goroutine per index

	g, gctx := errgroup.WithContext(ctx)  // gctx is canceled on first error
	g.SetLimit(limit)                     // g.Go blocks once limit slots are busy

	for i, task := range tasks {
		i, task := i, task // explicit copy; unneeded on Go 1.22+, harmless
		g.Go(func() error {
			if err := gctx.Err(); err != nil {
				return err // sibling already failed; skip the work
			}
			res, err := fn(gctx, task)
			if err != nil {
				return fmt.Errorf("task %s: %w", task.ID, err)
			}
			results[i] = res
			return nil
		})
	}

	if err := g.Wait(); err != nil {
		return nil, err
	}
	return results, nil
}

Milestones

1. Write the pure function: `limit` validation, `errgroup.WithContext`, `SetLimit`, launch loop, `g.Wait()`. Compile it with fake `fn` implementations.
2. Make `fn` take a `context.Context` and pass `gctx`, not the parent `ctx`. Add the `gctx.Err()` guard at the top of each task.
3. Add a caller deadline test: `context.WithTimeout(ctx, 20*time.Millisecond)` against a `fn` that sleeps 200ms, and assert `errors.Is(err, context.DeadlineExceeded)`.
4. Add the first-error test: one task returns a sentinel error; assert the sentinel survives `errors.Is`.
5. Add the concurrency-cap test: 30 tasks, limit 4, track current and peak in-flight counts with `atomic.Int32`, assert peak <= 4.
6. Add the leak check: assert `runtime.NumGoroutine()` settles back to the starting value after `Wait` (allowing a brief settle).
7. Run everything under `go test -race -count=5 ./...`.

Acceptance Criteria

  • go vet ./... and go test -race -count=3 ./... are clean.
  • Peak concurrency never exceeds limit across runs, not just on average.
  • The first error wins: a sentinel from task b is retrievable via errors.Is even when a and c also fail.
  • A canceled caller context produces ctx.Err() in under the timeout window, with no goroutines left behind.
  • limit = 0 and limit = -1 return an error and do not panic.
  • Results are index-aligned: results[i].TaskID == tasks[i].ID.

Hints and Common Pitfalls

  • SetLimit blocks the caller. That is the feature. It also means a nested g.Go from inside a task can deadlock when the limit is saturated. Keep the fan-out flat.
  • g.Wait() returns only one error. Document it. If you need the full set, collect your own []error behind a mutex, and accept that you are no longer fail-fast.
  • The group context dies with Wait. Do not pass gctx into anything that outlives Run, like a background refresh or a queued job.
  • Shared slice, distinct indexes is fine. Appending to results from tasks is a data race; writing results[i] is not. This is the cheapest way to avoid a mutex.
  • -race is a detector, not a proof. It only reports races on paths you actually executed. Loop-count your tests when hunting flakiness.
  • Do not swallow ctx.Err() behind a generic error. Wrap with %w or return the context error directly, otherwise your caller cannot distinguish "canceled" from "bad input".
  • The channel-semaphore alternative: sem := make(chan struct{}, limit), sem <- struct{}{} before the work, defer func() { <-sem }(). Reach for it when you are not using errgroup at all, or when you need select non-blocking scheduling via g.TryGo.
  • Never close a channel you do not own. If you add a fan-in stage, the producers close, the consumer ranges.

A quick self-check with cloze:

  • errgroup.WithContext gives you a group and a context that is {{c1::canceled when the first goroutine returns a non-nil error}}.
  • g.SetLimit(n) bounds the fan-out because g.Go {{c2::blocks until a slot is free}}.
  • Writing results[i] from the goroutine that owns index i needs {{c3::no mutex at all}}.
  • You catch ordering bugs and races with go test {{c4::-race}}.
  • A task should return the context error, not a generic one, so callers can match it with {{c5::errors.Is}}.

Summary and Next Steps

You built Run: bounded, fail-fast, context-aware, index-stable, and race-detector clean. The three ideas that carry it are errgroup.WithContext for cancellation, SetLimit for backpressure, and per-index ownership to avoid locking. Keep Task, Result, and fn free of transport details so the package stays testable with plain fakes.

Next up is the Concurrency Foundations checkpoint. Expect questions on comparing errgroup, sync.WaitGroup, and a plain channel semaphore, on what Wait does and does not return, and on explaining precisely when fan-out is the wrong choice because the dependency is already serialized, rate-limited, or cheaper to batch in one query.

[
  { "front": "What two values does errgroup.WithContext return?", "back": "A *errgroup.Group and a derived context that is canceled when the first goroutine returns a non-nil error." },
  { "front": "What does g.Wait() return when several tasks fail?", "back": "Only the first non-nil error. Later errors are dropped." },
  { "front": "How do you cap concurrency with errgroup?", "back": "g.SetLimit(n). g.Go then blocks once n slots are in use." },
  { "front": "Channel-semaphore idiom as an alternative to SetLimit?", "back": "sem := make(chan struct{}, limit); send before the work, receive (defer) when done." },
  { "front": "Why write results[i] instead of appending to a shared slice?", "back": "Each index is owned by one goroutine, so no mutex and no data race." },
  { "front": "Why check gctx.Err() at the top of each task?", "back": "To skip work that is already doomed after a sibling failed or the caller canceled." },
  { "front": "Why wrap task errors with fmt.Errorf(\"...: %w\", err)?", "back": "It preserves the chain so errors.Is and errors.As work through the returned error." },
  { "front": "What happens to the group context once Wait returns?", "back": "It is canceled, even if every task succeeded. Do not reuse it afterwards." },
  { "front": "Which flag runs the race detector in tests?", "back": "go test -race. It reports races only on paths that actually execute." },
  { "front": "What should Run return when the caller's deadline expires?", "back": "ctx.Err() (typically context.DeadlineExceeded), reachable via errors.Is." },
  { "front": "Why is unbounded fan-out dangerous even though goroutines are cheap?", "back": "Connections, file handles and memory are not cheap, and one slow dependency can pile up thousands of waiters." },
  { "front": "What is g.TryGo useful for?", "back": "A non-blocking variant of g.Go that returns false when the limit has no free slot." },
  { "front": "Deadlock risk when combining SetLimit with nested g.Go calls?", "back": "A task holding a slot that waits on g.Go can block forever when the limit is saturated." },
  { "front": "How do you validate the concurrency cap in a test?", "back": "Track current and peak in-flight counts with atomic.Int32 and assert peak <= limit." }
]