Goroutines with an owner: concurrency patterns for Go agents

Go makes starting concurrent work cheap: go f() is all it takes. An agent uses that everywhere, for tool calls in parallel, one run per batch of work, a server answering several requests at once, a timer that watches a stream. The hard part is never starting a goroutine; it is knowing who stops it, and making sure it can stop.

The rule this page follows comes from the companion code’s style guide: every goroutine has an owner and a way to stop. The owner is the code that waits for it to finish, and the way to stop is almost always a context. The examples are from the companion repository, where they are tested with the race detector.

Start and wait: sync.WaitGroup.Go

sync.WaitGroup is how an owner waits for the goroutines it started. Since Go 1.25 it has a Go method, which starts a function in a goroutine and counts it in one call, wg.Go(func() { ... }). It replaces the older three-step form, wg.Add(1), go func() { defer wg.Done(); ... }(), and the mistakes that came with it: an Add placed inside the goroutine, where Wait can run before it, or a Done forgotten on one return path.

A closure started in a loop can use the loop’s variables directly. Since Go 1.22, each iteration of a for loop has its own copy of its variables, so the old habit of copying them first, name := name, is no longer needed.

Merging several runs into one stream

Running several agents at once, one per group of crash reports, is a fan-out; reading all their events as one stream is a fan-in. Here is the fan-in, which starts one goroutine per run and sends every event through one channel:

// Merge runs several event streams at once, such as one agent's run per
// fingerprint group, and yields their events as they happen, each tagged with
// the name of the run it came from. Each run's StoppedEvent comes
// through like any other event, even when ctx is cancelled; Merge ends
// when every run has stopped. Breaking out of the range loop cancels
// every run and waits for their goroutines to finish, so none is left
// blocked on a send.
func Merge(ctx context.Context,
	runs map[string]func(context.Context) iter.Seq[Event]) iter.Seq[Event] {
	return func(yield func(Event) bool) {
		ctx, cancel := context.WithCancel(ctx)
		defer cancel()
		events := make(chan Event)
		quit := make(chan struct{}) // closed when the caller stops reading
		var wg sync.WaitGroup
		for name, run := range runs {
			wg.Go(func() {
				for ev := range run(ctx) {
					ev.Agent = name
					select {
					case events <- ev:
					case <-quit:
						return
					}
				}
			})
		}
		go func() { wg.Wait(); close(events) }()

Each goroutine forwards its run’s events, tagged with the run’s name. A separate goroutine waits for all of them and then closes the channel, so the reader knows when every run has finished; closing a channel is the sender’s job, and here the senders’ owner does it once, after the last of them is done.

The reading side is where the leaks are avoided:

		for ev := range events {
			if !yield(ev) {
				close(quit)
				cancel()
				for range events {
				}
				return
			}
		}
	}
}

When the caller stops reading, by breaking out of its range loop, three things happen in order. quit is closed, so a goroutine waiting to send gives up and returns, which also ends its run. The context is cancelled, which stops the runs that were busy rather than sending. And the reader drains the channel until it closes, which happens only once every goroutine has returned, so when Merge returns, nothing it started is still running.

Look at what the send doesn’t wait on: the context. After Ctrl-C, each run still has a last event to report, the one that says how it stopped, and the caller is still reading. A send that also gave up on ctx.Done() would sometimes drop it, because when several cases of a select are ready, Go picks one at random; TestMergeCancel checks that every run’s last event arrives.

Without both protections, the quit case and the drain, a goroutine blocked on a send that nobody will ever receive stays blocked for the life of the process: a goroutine leak. In a service that runs agents all day, each abandoned run leaves one behind.

Proving there’s no leak

A leak doesn’t fail anything; it just accumulates. The companion code tests for one with the goroutine leak profile in runtime/pprof, which reports goroutines blocked forever on something nothing else can reach:

// TestMergeLeak reads one event from each fan-in and stops. The leak
// profile finds leakyMerge's blocked senders, and none from Merge.
func TestMergeLeak(t *testing.T) {
	ctx := t.Context()
	for range leakyMerge([]iter.Seq[Event]{counter(9)(ctx), counter(9)(ctx)}) {
		break
	}
	for range Merge(ctx, map[string]func(context.Context) iter.Seq[Event]{
		"a": counter(9), "b": counter(9)}) {
		break
	}
	var b bytes.Buffer
	if err := pprof.Lookup("goroutineleak").WriteTo(&b, 1); err != nil {
		t.Fatal(err)
	}
	t.Log(b.String())
	if !strings.Contains(b.String(), "leakyMerge") {
		t.Error("the leak profile didn't find leakyMerge's senders")
	}
	if strings.Contains(b.String(), "agent.Merge") {
		t.Error("Merge left a goroutine blocked")
	}
}

The test runs a deliberately broken version, leakyMerge, beside the real one, so it checks both that the profile finds a leak when there is one and that Merge leaves none.

The other tool is the race detector, go test -race, which reports two goroutines touching the same memory without synchronization while the tests run. The companion code’s checks run every test with it.

Requests in flight: a map behind a mutex

A server that answers several requests at once needs to know which are running, to cancel one when the client asks. The MCP server keeps each request’s cancel function in a map, and starts each request in its own goroutine:

// inflight holds the requests that are running, by ID, with the cancel
// function of each, so a notifications/cancelled can stop it.
type inflight struct {
	c  *conn
	wg sync.WaitGroup

	mu     sync.Mutex
	cancel map[string]context.CancelCauseFunc
}

// start runs a request in a goroutine of its own. A request that was
// cancelled gets no response: the client has stopped waiting for it.
func (s *Server) start(ctx context.Context, m message,
	calls *inflight) {
	id := canonicalID(m.ID)
	ctx, cancel := context.WithCancelCause(ctx)
	if !calls.add(id, cancel) {
		s.reply(calls.c, m.ID, nil, &Error{Code: CodeInvalidRequest,
			Message: "request ID " + id + " is already in use"})
		return
	}
	calls.wg.Go(func() {
		defer calls.remove(id)
		result, rpcErr := s.handle(ctx, m)
		if ctx.Err() == nil {
			s.reply(calls.c, m.ID, result, rpcErr)
		}
	})
}

The map is shared by every request goroutine and the reading loop, so it sits behind a sync.Mutex, and every access locks it:

func (f *inflight) add(id string, cancel context.CancelCauseFunc) bool {
	f.mu.Lock()
	defer f.mu.Unlock()
	if _, ok := f.cancel[id]; ok {
		cancel(nil)
		return false
	}
	f.cancel[id] = cancel
	return true
}

func (f *inflight) remove(id string) {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.cancel[id](nil)
	delete(f.cancel, id)
}

add refuses a request ID already in use and cancels the context it was given, so nothing leaks from a request that never starts. remove runs when the request’s goroutine ends, and cancels its context too, which releases what context.WithCancelCause holds.

Shutting down in order

The server’s main loop owns every request goroutine, and its deferred calls put the shutdown in order:

// Serve answers requests from r on w until r ends, each request in a
// goroutine of its own, and waits for them all before it returns.
func (s *Server) Serve(ctx context.Context, r io.Reader,
	w io.Writer) error {
	ctx, stop := context.WithCancelCause(ctx)
	calls := &inflight{c: newConn(r, w),
		cancel: map[string]context.CancelCauseFunc{}}
	defer calls.wg.Wait()
	defer stop(errServerStopped) // runs first: cancels what is running
	for {
		m, err := calls.c.receive()
		if errors.Is(err, io.EOF) { // the client closed our input
			calls.finish(drainGrace) // then the defers cancel the rest
			return nil
		}
		if rpcErr, ok := errors.AsType[*Error](err); ok {
			s.reply(calls.c, jsontext.Value("null"), nil, rpcErr)
			continue
		}
		if err != nil {
			return fmt.Errorf("read request: %w", err)
		}
		s.dispatch(ctx, m, calls)
	}
}

Deferred calls run in reverse order, so stop runs before calls.wg.Wait(): first every running request is cancelled, then the loop waits for each goroutine to return. The other order would wait for requests that nobody has told to stop. When the client closes its end, finish first gives the requests a short grace period to answer:

// finish waits up to d for the running requests to end.
func (f *inflight) finish(d time.Duration) {
	done := make(chan struct{})
	go func() { // ends once every request does, cancelled or not
		f.wg.Wait()
		close(done)
	}()
	select {
	case <-done:
	case <-time.After(d):
	}
}

The goroutine inside finish can outlive the call, when the grace period runs out first; it ends when the last request does, and the cancellation that the deferred calls run next makes sure that is soon.

Limiting how much runs at once

Starting a goroutine per task is fine for a handful; for hundreds of tool calls or crash reports, something has to cap how many run at once, both to stay within a provider’s rate limit and to bound memory. The companion code uses a buffered channel as a counting semaphore, with an Acquire that gives up when the context ends; the context page shows it, beside a rate limiter that spaces requests out in time. The parallel tool calls guide shows the runner that decides which of a reply’s tool calls may run together, and what running them in parallel saved.

Mistakes to avoid

• A goroutine with no way to stop. Every blocking operation inside it needs a ctx.Done() case, or a channel its owner closes.
• A send nobody will receive. If the reader can leave early, the sender needs another way out, like Merge’s quit, or the reader has to drain what is left.
• Closing a channel from the receiving side, or more than once. Close it from the sender’s side, once, after the last send; with several senders, their owner closes it after they finish.
• wg.Add inside the goroutine. Use wg.Go, which counts before it starts.
• Waiting before cancelling. At shutdown, cancel first, then wait; deferred calls run last-in, first-out, so defer the wait before the cancel.
• Shared maps without a lock. A map read and written from several goroutines needs a mutex; the race detector finds the ones the tests reach.

Try it

From the code directory of a clone of the companion repository:

$ go test -race ./ch15/agent -run 'TestMerge' -v

1. In TestMergeLeak, read the leak profile the test logs, and find leakyMerge’s blocked senders in it.
2. In Merge, remove both the case <-quit branch and the drain loop, run TestMergeLeak again, and see the profile name Merge’s blocked sends.
3. Put the quit case back and add a case <-ctx.Done(): return beside it, then run TestMergeCancel: some runs’ last events go missing.