Process jobs concurrently with a bounded worker pool

Approach

Start a fixed number of workers reading from a channel, cancel everyone on the first error with a cancellable context, and wait with sync.WaitGroup.Go (Go 1.25+).

Solution

package main

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

func processAll(ctx context.Context, ids []int, workers int, process func(context.Context, int) error) error {
	ctx, cancel := context.WithCancelCause(ctx)
	defer cancel(nil)

	jobs := make(chan int)
	var wg sync.WaitGroup
	for range workers {
		wg.Go(func() {
			for id := range jobs {
				if err := process(ctx, id); err != nil {
					cancel(fmt.Errorf("item %d: %w", id, err)) // first error wins
					return
				}
			}
		})
	}

feed:
	for _, id := range ids {
		select {
		case jobs <- id:
		case <-ctx.Done():
			break feed
		}
	}
	close(jobs)
	wg.Wait()
	return context.Cause(ctx) // nil when all items succeeded
}

func main() {
	ids := make([]int, 20)
	for i := range ids {
		ids[i] = i + 1
	}
	err := processAll(context.Background(), ids, 4, func(ctx context.Context, id int) error {
		select {
		case <-time.After(10 * time.Millisecond):
			return nil
		case <-ctx.Done():
			return ctx.Err()
		}
	})
	fmt.Println("error:", err)
}

Pitfalls

• Unbounded goroutines (for _, id := range ids { go ... }) for thousands of items.
• Workers that ignore ctx.Done(), so cancellation never stops them.
• Closing the jobs channel from a worker or closing it twice.

Variations

• golang.org/x/sync/errgroup with SetLimit(n) expresses the same pattern in fewer lines.
• Use a semaphore (chan struct{} of size n) when each item is launched independently.

Further reading: sync.WaitGroup