go-workgroup

go-workgroup is the gomatic ecosystem’s concurrent worker-group primitive for Go. It distributes work from a single producer across N consumer goroutines with type-safe generics, structured slog logging, and configurable error handling. A Source[T] feeds items into an internal channel, a pool of Worker[T] goroutines consumes them, and Run blocks until the work is done or the context is cancelled — returning the first error (fail-fast) or every error joined (CollectAll).

• Source: gomatic/go-workgroup
• API reference: pkg.go.dev/github.com/gomatic/go-workgroup

Install

go get github.com/gomatic/go-workgroup

Requires Go 1.26+.

Why a worker group

Spinning up goroutines by hand for a producer/consumer workload means re-deriving the same plumbing every time: a work channel, a sync.WaitGroup per stage, context cancellation propagated to every worker, and a thread-safe place to collect errors. Getting any one of these wrong leaks goroutines or deadlocks on a blocked send. go-workgroup owns that plumbing once. You supply only the two functions that are actually domain-specific — a Source[T] that produces items and a Worker[T] that consumes one — and the package runs them concurrently, cancels cleanly, and aggregates the result.

The two function types are the whole contract:

import workgroup "github.com/gomatic/go-workgroup"

// A Source sends work items to the channel until it is done or ctx is cancelled.
type Source[T any] func(context.Context, chan<- T) error

// A Worker processes one item; id is the 0-based worker index.
type Worker[T any] func(context.Context, int, T) error

Usage

Fan-out: distribute work across workers

FanOut runs the source against n concurrent workers:

package main

import (
	"context"
	"fmt"

	workgroup "github.com/gomatic/go-workgroup"
)

func main() {
	source := workgroup.Source[int](func(ctx context.Context, out chan<- int) error {
		for i := range 100 {
			select {
			case out <- i:
			case <-ctx.Done():
				return ctx.Err()
			}
		}
		return nil
	})

	worker := workgroup.Worker[int](func(ctx context.Context, id int, item int) error {
		fmt.Printf("worker %d processed item %d\n", id, item)
		return nil
	})

	if err := workgroup.FanOut(context.Background(), 8, source, worker); err != nil {
		panic(err)
	}
}

Fan-in: a single worker

FanIn is Run with exactly one worker — every item is processed serially by one goroutine, useful for aggregating a fan-out stage’s output into shared state without a mutex:

err := workgroup.FanIn(ctx, source, worker)

Run: explicit options

Run is the underlying entry point; FanOut and FanIn are thin wrappers that preset Workers. Pass any combination of options:

err := workgroup.Run(ctx, source, worker,
	workgroup.Workers(16),
	workgroup.Name("processor"),
	workgroup.Log{Logger: slog.Default()},
	workgroup.CollectAll,
)

With no options, Run defaults to runtime.NumCPU() workers, slog.Default(), and fail-fast error handling.

Pipe: chain stages

Pipe turns a Transformer[In, Out] into a new Source[Out], so a fan-out transform feeds a downstream stage. The returned source runs the upstream source with n workers applying the transform, emitting results for the next Run:

doubled := workgroup.Pipe(4, source,
	func(ctx context.Context, id int, item int) (int, error) {
		return item * 2, nil
	},
)

err := workgroup.FanIn(ctx, doubled, aggregator)

Error handling

The default is fail-fast: the first worker error cancels the shared context, every other worker stops, and that error is returned.

// Fail-fast (default): the first error cancels all workers.
err := workgroup.Run(ctx, source, riskyWorker)

Pass CollectAll to keep processing every item and aggregate the failures — the returned error joins them with errors.Join, so each is recoverable with errors.Is:

// Collect-all: continue processing, join all errors at the end.
err := workgroup.Run(ctx, source, riskyWorker, workgroup.CollectAll)
// err contains every worker error via errors.Join.

A nil source or nil worker is rejected before any goroutine starts, with the sentinels ErrNilSource and ErrNilWorker, both matchable with errors.Is.

API summary

Design

• Generic, value-typed contract. Source[T], Worker[T], and Transformer[In, Out] are plain function types — no interfaces to implement, no work-item boxing into any.
• Options are typed values. Each option (Workers, Name, Log, FailFast/CollectAll) is a small named type satisfying the Optional interface, so configuration is self-describing and nil options are skipped.
• Clean cancellation. The group runs under a context.WithCancelCause; a source or worker failure cancels it with the causing error. After every worker returns, the package drains any item still pending on the channel, so even a non-cooperative source that ignores ctx and blocks on a raw send is unblocked rather than deadlocking the group.
• Deterministic outcome. The final error is resolved in priority order: worker errors first, then a real (non-context) source error, then any context cancellation — otherwise the run completed successfully and nil is returned.
• Structured logging built in. Start, success, and error transitions are logged via slog.LogAttrs with the worker count and optional name as attributes.
• Sentinel errors. The package’s own failures are constants of an Error string newtype whose .With(cause, args...) wraps with %w, mirroring the gomatic/go-error mechanism so error construction stays uniform across the ecosystem.
• Dependency-light — the package depends only on the standard library (context, errors, log/slog, runtime, sync).

Who uses it

go-workgroup is the shared concurrency primitive for gomatic Go projects that fan work across goroutines, alongside the rest of the gomatic/go-* libraries.