conker

README

conker: concurrency with the string attached

conker is a structured concurrency library for Go 1.25+, built on the lessons of sourcegraph/conc with a modernised stdlib-first approach.

go get github.com/purpleclay/conker

At a glance

• Use pool.Pool if you want bounded, panic-safe concurrent task execution with recursive submission and error aggregation.
• Use pool.ResultPool[T] if tasks produce typed results and you want them back in submission order.
• Use stream.Stream if you want concurrent producers with callbacks that execute strictly in submission order, each blocking until its slot's turn arrives.
• Use iter.MapSeq / iter.MapSeq2 / iter.MapMap if you want to concurrently transform a sequence, iter.Seq2, or map, with results in order by default.
• Use iter.ForEachSeq / iter.ForEachMap if you want concurrent side effects over a sequence or map, bounded by WithMaxGoroutines.
• Use iter.MapSeqErr / iter.ForEachSeqErr if those operations can fail — errors are joined via errors.Join, and WithCancelOnError cancels in-flight work on the first error.
• Use conker.WaitGroup if you want a panic-safe replacement for sync.WaitGroup.
• Use panics.Catcher if you want to catch panics in goroutines you manage yourself.
• Use panics.ErrPanic with errors.Is to detect recovered panics without a type assertion.
• Use daemon.Pool if you want long-lived background workers with coordinated cancellation and graceful, bounded shutdown.

Goals

Typed panics, not bare any

Go's recover() returns any. Every library that catches goroutine panics has to decide what to do with that opaque value, and the stdlib gives you nothing beyond the raw interface.

conker makes recovered panics first-class errors. *panics.Recovered implements error and integrates with errors.Is and errors.As — the same tools you already use for every other error in your program.

stdlib	conker
done := make(chan any, 1) go func() { defer func() { done <- recover() }() doSomethingThatMightPanic() }() val := <-done if val != nil { // val is any — type-switch required, // no errors.Is/As, stack lost panic(val) }	var c panics.Catcher c.Try(doSomethingThatMightPanic) if r := c.Recovered(); r != nil { // r is *panics.Recovered — a typed error errors.Is(r, panics.ErrPanic) // true errors.Is(r, ErrTimeout) // true if panic value wrapped it log.Print(r.Stack) // stack at panic site, not here }

Panic-safe goroutines without boilerplate

Spawning goroutines safely with sync.WaitGroup requires a lot of ceremony: Add, Done, a deferred recover, and a decision about what to do with the panic value. conker.WaitGroup collapses all of that to one line.

Under the hood it delegates directly to sync.WaitGroup.Go — introduced in Go 1.25 — so there is no reimplementation of what the stdlib already does.

stdlib	conker
var wg sync.WaitGroup for i := 0; i < 10; i++ { wg.Add(1) go func() { defer wg.Done() defer func() { if v := recover(); v != nil { // swallowed, or re-panic loses stack } }() doSomething() }() } wg.Wait()	var wg conker.WaitGroup for i := 0; i < 10; i++ { wg.Go(doSomething) } // re-panics as *panics.Recovered with // the original stack trace preserved wg.Wait()

When you want to handle a panic rather than propagate it, use WaitAndRecover:

var wg conker.WaitGroup
wg.Go(riskyOperation)

if r := wg.WaitAndRecover(); r != nil {
    log.Printf("goroutine panicked: %v\n%s", r.Value, r.Stack)
}

Bounded concurrent tasks, not cancel-on-first-error

Running bounded concurrent tasks in Go typically means errgroup — and when used with errgroup.WithContext, it cancels remaining work on first error and still returns only one error from Wait.

pool.Pool collects every error from every task via errors.Join, recovers panics as typed *panics.Recovered errors, and passes each task its own context for per-task timeouts and backpressure via GoCtx.

errgroup	conker
g, ctx := errgroup.WithContext(context.Background()) g.SetLimit(10) for _, item := range items { g.Go(func() error { return process(ctx, item) }) } // cancels remaining tasks on first error; // Wait returns only that one error; // panics crash the process if err := g.Wait(); err != nil { ... }	p := pool.New().WithMaxGoroutines(10) for _, item := range items { p.Go(func(ctx context.Context) error { return process(ctx, item) }) } // collects all errors via errors.Join; // panics surface as *panics.Recovered if err := p.Wait(); err != nil { ... }

Concurrent processing with ordered callbacks

Processing items concurrently but emitting results in the original order typically requires buffering everything, sorting by index, and iterating only after all work is done.

stream.Stream separates the two concerns: producers run concurrently, but the dispatcher advances strictly in submission order — blocking on each slot until its producer finishes before calling the callback and moving to the next. A fast producer whose slot comes after a slow one waits; nothing fires out of sequence.

stdlib	conker
results := make([]string, len(items)) var wg sync.WaitGroup for i, item := range items { wg.Add(1) go func(i int, item Item) { defer wg.Done() results[i] = process(item) }(i, item) } wg.Wait() // ordered, but only after all work is done for _, r := range results { emit(r) }	s := stream.New() for _, item := range items { s.Go(func(ctx context.Context) stream.Callback { result := process(ctx, item) // concurrent return func() { emit(result) // ordered, fires when this slot's turn arrives } }) } s.Wait()

All results collected, not silently dropped

sourcegraph/conc's error-variant pools (ResultErrorPool, ResultContextPool) drop results from failed tasks by default. The WithCollectErrored() option exists specifically to opt back in — meaning silent data loss is the default behaviour.

pool.ResultPool[T] always returns every result alongside every error. Nothing is silently discarded because a task failed or panicked.

sourcegraph/conc	conker
p := pool.NewWithResults[string]().WithErrors() for _, id := range ids { id := id p.Go(func() (string, error) { return fetch(id) }) } results, err := p.Wait() // results from failed tasks are silently // dropped unless WithCollectErrored() is set	p := pool.NewWithResults[string]() for _, id := range ids { p.Go(func(_ context.Context) (string, error) { return fetch(id) }) } results, err := p.Wait() // all results returned alongside errors; // nothing is silently dropped

Concurrent slice and map iteration, ordered by default

Mapping over a slice concurrently while preserving order is a common pattern, but doing it correctly means hand-rolling an index-based result slice, a semaphore for bounded concurrency, and a WaitGroup — repeated in every codebase that needs it.

iter.MapSeq (and iter.MapSeq2 / iter.MapMap for iter.Seq2 and Go maps) does this once: bounded concurrency via WithMaxGoroutines, results in original order, and iter.MapSeqErr / iter.ForEachSeqErr for error-returning, context-aware variants with errors.Join aggregation.

stdlib	conker
results := make([]string, len(urls)) sem := make(chan struct{}, 8) var wg sync.WaitGroup for i, url := range urls { wg.Add(1) sem <- struct{}{} go func(i int, url string) { defer wg.Done() defer func() { <-sem }() results[i] = fetch(url) }(i, url) } wg.Wait()	results := slices.Collect(iter.MapSeq( slices.Values(urls), fetch, iter.WithMaxGoroutines(8), ))

Examples

Each example is a self-contained, runnable program. External clients are stubbed so there are no runtime dependencies beyond the Go standard library and conker itself.

Example	Feature	Use case
etl-pipeline	pool.ResultPool[T]	ETL pipeline over an object store — download, transform, upload — with bounded concurrency, per-task timeouts, and submission-order results.
log-enrichment	stream.Stream	Enrich a web server log stream concurrently — resolving each client IP to a country — while preserving the original chronological order.
fleet-healthcheck	iter.MapSeqErr	Concurrently health-check a fixed fleet of hosts, returning results in original fleet order with error aggregation and fail-fast cancellation.
background-services	daemon.Pool	Run long-lived background workers for a service, with coordinated cancellation, panic isolation, and bounded graceful shutdown.

Documentation

Index

• func OnceErr[T any](fn func() (T, error)) func() (T, error)
• type WaitGroup
  • func (w *WaitGroup) Go(f func())
  • func (w *WaitGroup) Wait()
  • func (w *WaitGroup) WaitAndRecover() *panics.Recovered

Functions

func OnceErr

func OnceErr[T any](fn func() (T, error)) func() (T, error)

OnceErr returns a function that invokes fn at most once, no matter how many goroutines call the returned function, and how often. Every caller receives the same (T, error) pair produced by that single call.

It is the error-returning equivalent of sync.OnceValue.

Example:

load := conker.OnceErr(func() (*Config, error) {
    return loadConfig("config.yaml")
})

cfg, err := load() // loads config.yaml
cfg, err = load()  // returns the cached result; fn is not called again

Types

type WaitGroup

type WaitGroup struct {
	// contains filtered or unexported fields
}

WaitGroup is a panic-safe wrapper around sync.WaitGroup that propagates panics from goroutines spawned with WaitGroup.Go to the caller of WaitGroup.Wait or WaitGroup.WaitAndRecover.

The zero value is ready to use. WaitGroup is reusable across multiple Wait cycles when no panic occurs. If a goroutine panics, create a new WaitGroup for subsequent work rather than reusing the same instance.

Example — basic usage:

var wg conker.WaitGroup
wg.Go(func() { doWork() })
wg.Go(func() { doMoreWork() })
wg.Wait() // re-panics if any goroutine panicked

Example — recovering from a panic without propagating:

var wg conker.WaitGroup
wg.Go(func() { riskyWork() })
if r := wg.WaitAndRecover(); r != nil {
    log.Printf("goroutine panicked at:\n%s", r.Stack)
}

func (*WaitGroup) Go

func (w *WaitGroup) Go(f func())

Go spawns f in a new goroutine. Any panic raised by f is captured and surfaced when WaitGroup.Wait or WaitGroup.WaitAndRecover is called, with the stack trace preserved at the original panic site.

func (*WaitGroup) Wait

func (w *WaitGroup) Wait()

Wait blocks until all goroutines launched with WaitGroup.Go have returned, then re-panics with the first captured panic as a *panics.Recovered. It is a no-op if no goroutine panicked.

func (*WaitGroup) WaitAndRecover

func (w *WaitGroup) WaitAndRecover() *panics.Recovered

WaitAndRecover blocks until all goroutines launched with WaitGroup.Go have returned and returns the first captured panic, or nil if none occurred.