Part 12 - errgroup: When One Task Fails, Cancel the Rest

Part of the series: Production-Grade Concurrent AI Systems in Go

→ Full code for this post: github.com/madmmas/go-concurrent-ai-systems/tree/part-12 → Diff from Part 11: compare/part-11...part-12 → Run it: go run ./cmd/news-processor -articles=3 -workers=2 inside arc-2-production/part-12-errgroup

Part 10 introduced fan-out: three goroutines launch simultaneously for each article, and the fan-in collects whatever each one returns. That design works well when tasks are independent and partial results are acceptable — if keyword extraction fails, you might still want the summary and sentiment.

But not all pipelines work that way. If any one of the three tasks fails, many pipelines want to stop the others immediately. There is no point spending another 600ms on a sentiment call if summarisation just returned a server error and the whole article result is going to be discarded. The cost is wasted tokens, wasted time, and unnecessary load on a provider that may already be struggling.

The Part 10 approach has no mechanism for this. Once goroutines are launched, they run to completion regardless of what their siblings return. Cancelling siblings on first failure requires a shared cancellable context — and managing that manually means writing code you will almost certainly get wrong the first time.

errgroup solves this cleanly.

What errgroup Does

errgroup.Group is a thin wrapper around sync.WaitGroup that adds two things:

1. A shared context that gets cancelled the moment any goroutine returns a non-nil error
2. First-error collection — Wait() returns the first error that occurred, not all of them

The API is three methods:

g, ctx := errgroup.WithContext(parentCtx)

g.Go(func() error {
    // work using ctx — if another goroutine fails, ctx is cancelled here
    return someWork(ctx)
})

g.Go(func() error {
    return otherWork(ctx)
})

if err := g.Wait(); err != nil {
    // first error from any goroutine
}

g.Wait() blocks until all goroutines have returned, then gives you the first error. If all succeed, it returns nil. If one fails, the shared ctx is cancelled mid-flight — other goroutines see ctx.Done() close and should return promptly.

That last part is the caller's responsibility. errgroup cancels the context; it cannot forcibly stop goroutines. Your task functions must respect the context by passing it to every blocking call. LLM calls, database queries, HTTP requests — anything that takes time needs to receive the context and return when it is cancelled. The simulator's Call method does this correctly, which is why the cancellation happens immediately in the examples below.

Under the Hood

Rather than treat errgroup as a black box, the repo includes a self-contained implementation using only the standard library. It is 40 lines:

type Group struct {
    cancel   func()
    wg       sync.WaitGroup
    mu       sync.Mutex
    firstErr error
}

func WithContext(ctx context.Context) (*Group, context.Context) {
    ctx, cancel := context.WithCancel(ctx)
    return &Group{cancel: cancel}, ctx
}

func (g *Group) Go(fn func() error) {
    g.wg.Add(1)
    go func() {
        defer g.wg.Done()
        if err := fn(); err != nil {
            g.mu.Lock()
            if g.firstErr == nil {
                g.firstErr = err
                g.cancel() // cancel the shared context
            }
            g.mu.Unlock()
        }
    }()
}

func (g *Group) Wait() error {
    g.wg.Wait()
    if g.cancel != nil {
        g.cancel() // release resources even on success
    }
    return g.firstErr
}

The mutex protects firstErr — multiple goroutines can fail simultaneously, and only the first error is kept. Once cancel() fires, the shared context transitions to done, and every goroutine checking ctx.Done() or passing the context to a blocking call will see it.

g.cancel() is also called inside Wait() on success. This is mandatory — context.WithCancel allocates a goroutine internally that leaks if cancel is never called. The defer cancel() pattern from Part 8 applies here too, and errgroup handles it for you so you cannot forget.

In production, use golang.org/x/sync/errgroup directly. This implementation is included so you can see the mechanism rather than import a package and trust it.

Applying errgroup to Fan-Out

In Part 10, processArticleFanOut used a manual taskResult channel and WaitGroup. Part 12 replaces that with errgroup:

func (p *ErrgroupPool) processWithErrgroup(ctx context.Context, article model.Article) model.AIResult {
    result := model.AIResult{ArticleID: article.ID}

    // WithContext creates the group and a derived context.
    // If any g.Go function returns an error, gctx is cancelled.
    g, gctx := WithContext(ctx)

    var (
        mu        sync.Mutex
        summary   string
        sentiment string
        keywords  string
    )

    g.Go(func() error {
        if err := p.llm.Call(gctx, "Summarisation", article.ID); err != nil {
            return err  // cancels gctx immediately
        }
        mu.Lock(); summary = "AI-generated summary"; mu.Unlock()
        return nil
    })

    g.Go(func() error {
        if err := p.llm.Call(gctx, "Sentiment Analysis", article.ID); err != nil {
            return err
        }
        mu.Lock(); sentiment = "Positive"; mu.Unlock()
        return nil
    })

    g.Go(func() error {
        if err := p.llm.Call(gctx, "Keyword Extraction", article.ID); err != nil {
            return err
        }
        mu.Lock(); keywords = "AI,Go,Concurrency"; mu.Unlock()
        return nil
    })

    if err := g.Wait(); err != nil {
        result.Err = err
        return result
    }

    result.Summary   = summary
    result.Sentiment = sentiment
    result.Keywords  = []string{keywords}
    return result
}

The mutex on the result fields is still necessary — three goroutines write to the same variables, and even though errgroup serialises error collection, successful writes can still race. This is one of the easy mistakes to make when adopting errgroup: assuming the group handles all synchronisation. It handles error collection and context cancellation, nothing else.

What the Output Shows

Happy path — all three tasks succeed:

[article 1] starting errgroup fan-out
  [1] Keyword Extraction started (656ms)
  [1] Summarisation started (447ms)
  [1] Sentiment Analysis started (378ms)
  [1] Sentiment Analysis completed
  [1] Summarisation completed
  [1] Keyword Extraction completed
[article 1] errgroup: all tasks complete

Three tasks in parallel, fan-in via g.Wait(), result assembled. Identical behaviour to Part 10 on the happy path.

Failure path — 100% server error rate:

[article 1] starting errgroup fan-out
  [1] Keyword Extraction → 503 server error
  [1] Summarisation → 503 server error
  [1] Sentiment Analysis → 503 server error
[article 1] errgroup: task failed: llm: server error (503)

Succeeded: 0 | Failed: 3 | Duration: 1ms

All three tasks fail immediately. g.Wait() returns the first error, the article result carries it, and the pipeline moves on to the next article. Total duration is 1ms — the tasks failed fast and nothing waited for timeouts that would never fire.

This is the key difference from Part 10. If the failing task had been the summariser, and summarisation took 600ms before returning an error, Part 10 would have kept the sentiment and keyword goroutines running for their full 600ms even though their results were going to be discarded. With errgroup, the failed task's error fires immediately and cancels gctx — sentiment and keyword calls see the context cancelled and return within microseconds.

The Comparison That Makes the Case

Part 10's manual fan-out needed:

• a taskResult type to carry name + value + error through a channel
• a buffered channel with capacity 3
• a WaitGroup to close the channel after all goroutines finished
• a separate closer goroutine to avoid deadlock
• a switch on tr.name to route results to the right field
• manual iteration over the channel to collect everything

No single piece is hard, but the combination has several places to get wrong — the buffer size, the close ordering, the name routing. And after all that, you still have no mechanism for sibling cancellation.

Part 12's errgroup version needs:

• g, gctx := WithContext(ctx)
• three g.Go(func() error { ... }) calls
• if err := g.Wait(); err != nil { ... }

Sibling cancellation is implicit — built into the mechanism, not added on top.

The tradeoff is that errgroup gives you only the first error. If you need all errors from all failing tasks, you still need a manual channel or a custom accumulator. For most AI pipelines, the first error is enough — it tells you why the article failed, and retrying it is Part 13's job.

errgroup vs Part 10's Fan-Out — When to Use Which

For our news pipeline, errgroup is the right fit. A result with a missing summary and present sentiment score is not useful — we cannot half-index an article. Either all three fields are populated or the article goes back for retry. Part 10's approach of collecting whatever succeeded and setting Err on partial failure still has its place in pipelines where partial results have value, but structured concurrency with errgroup is the production default for cases where you need all or nothing.

What's Next

errgroup handles the case where tasks fail cleanly and the article needs to be retried. But what does retry actually look like? How long do you wait before trying again? Do you wait the same amount each time? What happens when you have tried three times and still failed?

Part 13 adds retries with exponential backoff and jitter — the standard approach for any pipeline that calls an external API under real conditions.

See you in Part 13.

This is Part 12 of the series "Production-Grade Concurrent AI Systems in Go." Read Part 11 — Multi-Stage Pipeline or continue to Part 13 — Retries and Exponential Backoff: Handling Failure Gracefully.