Berkan Çetinkaya
Go 10 min read

Building ch.Merge: Fan-In Coordination in Go

Published on May 15, 2026 By Berkan Çetinkaya
Series Building ch: Small Channel Helpers From First Principles Part 5 of 6 See list Previous Next

Sometimes a feature starts with multiple independent producers.

For example, a model comparison flow may call multiple generation providers at the same time.

Each provider finishes whenever it finishes.

One may return quickly.

Another may be slower.

From the caller’s point of view, we do not want to manually wait on different result streams.

We want one stream of results.

That is the fan-in problem.

many producers
↓
one consumer-facing stream

Start Without an Abstraction

First, define the business value moving through the system:

type GenerationResult struct {
	Model string
	URL   string
	Err   error
}

Now imagine two provider result streams:

imagenResults := make(chan GenerationResult)
geminiResults := make(chan GenerationResult)

They can produce independently:

go func() {
	defer close(imagenResults)

	imagenResults <- GenerationResult{
		Model: "imagen",
		URL:   "https://example.com/imagen.png",
	}
}()

go func() {
	defer close(geminiResults)

	geminiResults <- GenerationResult{
		Model: "gemini",
		URL:   "https://example.com/gemini.png",
	}
}()

At this point, there is no Merge.

There are only two channels.

Sequential Receive Is Fragile

The simplest thing we can do is receive from both:

imagen := <-imagenResults
gemini := <-geminiResults

fmt.Println(imagen.Model, imagen.URL)
fmt.Println(gemini.Model, gemini.URL)

This works only if we are fine waiting for Imagen first.

But if Gemini finishes earlier, we still block on Imagen.

That is not fan-in.

That is sequential waiting.

select Gives Us Manual Fan-In

A better version listens to both channels:

select {
case result := <-imagenResults:
	fmt.Println(result.Model, result.URL)

case result := <-geminiResults:
	fmt.Println(result.Model, result.URL)
}

Now whichever provider produces first wins.

But this only receives one value.

So we loop:

for received := 0; received < 2; received++ {
	select {
	case result := <-imagenResults:
		fmt.Println(result.Model, result.URL)

	case result := <-geminiResults:
		fmt.Println(result.Model, result.URL)
	}
}

This is the first real fan-in shape:

two independent producers
↓
one consumer path

But it is still only an intermediate step.

This loop assumes there will be exactly two successful values.

That is fine for a tiny demo.

It is not a good general mental model for fan-in.

Closed Channels Change the Problem

The loop above ignores an important case.

A provider channel may close.

Receiving from a closed channel does not panic.

It returns the zero value and ok=false.

So this is not enough:

case result := <-imagenResults:

We need:

case result, ok := <-imagenResults:
	if !ok {
		// this input stream is finished
	}

And if we keep the received < 2 loop, another bug appears.

If we write this:

for received := 0; received < 2; received++ {
	select {
	case result, ok := <-imagenResults:
		if !ok {
			imagenResults = nil
			continue
		}

		fmt.Println(result.Model, result.URL)

	case result, ok := <-geminiResults:
		if !ok {
			geminiResults = nil
			continue
		}

		fmt.Println(result.Model, result.URL)
	}
}

we are still tying loop progress to an expected value count instead of input-stream lifecycle.

That matters because received < 2 is only safe if both producers are guaranteed to emit exactly one real value.

If one producer closes early without producing a value, received may never reach 2.

But there is one more trap.

A closed channel is always ready.

If a closed channel stays inside a select, that case can keep winning immediately.

So when a channel closes, we set it to nil:

if !ok {
	imagenResults = nil
	continue
}

Why?

closed channel = always ready
nil channel    = never ready

Inside a select, a nil channel effectively disables that case.

This is the critical transition in the teaching progression:

we stop counting expected values and start tracking whether input streams are still alive.

Manual Fan-In With Closed Channel Handling

Now the more correct manual version becomes:

for imagenResults != nil || geminiResults != nil {
	select {
	case result, ok := <-imagenResults:
		if !ok {
			imagenResults = nil
			continue
		}

		fmt.Println(result.Model, result.URL)

	case result, ok := <-geminiResults:
		if !ok {
			geminiResults = nil
			continue
		}

		fmt.Println(result.Model, result.URL)
	}
}

This is more correct.

Why is it better?

Because the loop condition now matches the real termination rule:

keep going until all input streams are finished.

But the feature code is now carrying a lot of coordination detail:

  • receive from multiple channels
  • detect closed channels
  • disable closed select cases
  • keep looping until all inputs finish
  • eventually add cancellation
  • eventually forward results somewhere else

That is too much infrastructure inside business logic.

Adding Cancellation

Now add context cancellation:

for imagenResults != nil || geminiResults != nil {
	select {
	case <-ctx.Done():
		return

	case result, ok := <-imagenResults:
		if !ok {
			imagenResults = nil
			continue
		}

		fmt.Println(result.Model, result.URL)

	case result, ok := <-geminiResults:
		if !ok {
			geminiResults = nil
			continue
		}

		fmt.Println(result.Model, result.URL)
	}
}

The behavior is better.

But the coordination code keeps growing.

At this point, extracting a business-specific helper is justified.

Not generic yet.

Just business-specific.

A Business-Specific Merge

func MergeGenerationResults(
	ctx context.Context,
	imagenResults <-chan GenerationResult,
	geminiResults <-chan GenerationResult,
) <-chan GenerationResult {
	out := make(chan GenerationResult)

	go func() {
		defer close(out)

		for imagenResults != nil || geminiResults != nil {
			select {
			case <-ctx.Done():
				return

			case result, ok := <-imagenResults:
				if !ok {
					imagenResults = nil
					continue
				}

				select {
				case <-ctx.Done():
					return
				case out <- result:
				}

			case result, ok := <-geminiResults:
				if !ok {
					geminiResults = nil
					continue
				}

				select {
				case <-ctx.Done():
					return
				case out <- result:
				}
			}
		}
	}()

	return out
}

Now the caller becomes simpler:

func CompareModels(ctx context.Context) <-chan GenerationResult {
	imagenResults := GenerateWithImagen(ctx)
	geminiResults := GenerateWithGemini(ctx)

	return MergeGenerationResults(ctx, imagenResults, geminiResults)
}

And the consumer only deals with one stream:

for result := range CompareModels(ctx) {
	if result.Err != nil {
		fmt.Println(result.Model, "failed:", result.Err)
		continue
	}

	fmt.Println(result.Model, "->", result.URL)
}

This is the key shift.

The caller no longer coordinates providers manually.

It consumes a single stream.

One Channel Can Produce Many Values

So far, each provider sent one result.

But a channel is a stream, not a single value.

A provider may send multiple results before closing:

func GenerateWithImagen(ctx context.Context) <-chan GenerationResult {
	out := make(chan GenerationResult)

	go func() {
		defer close(out)

		for i := 0; i < 3; i++ {
			out <- GenerationResult{
				Model: "imagen",
				URL:   fmt.Sprintf("https://example.com/imagen-%d.png", i),
			}
		}
	}()

	return out
}

Merge should not read only one value from this channel.

It should keep reading until the channel is closed.

This becomes important in the next version.

The Provider Count Grows

The business-specific helper works for two providers.

But now imagine the feature grows:

  • Imagen
  • Gemini
  • OpenAI
  • Fal
  • Replicate

The function signature starts getting worse:

func MergeGenerationResults(
	ctx context.Context,
	imagen <-chan GenerationResult,
	gemini <-chan GenerationResult,
	openai <-chan GenerationResult,
	fal <-chan GenerationResult,
	replicate <-chan GenerationResult,
) <-chan GenerationResult

That is the next pressure point.

The abstraction needs to support any number of input streams.

Variadic Business Merge

func MergeGenerationResults(
	ctx context.Context,
	inputs ...<-chan GenerationResult,
) <-chan GenerationResult {
	out := make(chan GenerationResult)

	var wg sync.WaitGroup

	for _, in := range inputs {
		wg.Add(1)

		go func(in <-chan GenerationResult) {
			defer wg.Done()

			for {
				select {
				case <-ctx.Done():
					return

				case result, ok := <-in:
					if !ok {
						return
					}

					select {
					case <-ctx.Done():
						return
					case out <- result:
					}
				}
			}
		}(in)
	}

	go func() {
		wg.Wait()
		close(out)
	}()

	return out
}

This version uses two loops.

The outer loop is about input channels:

for _, in := range inputs {
	// start one forwarding goroutine per input channel
}

The inner loop is about messages inside each channel:

for {
	// keep reading from this channel until it closes
}

Mental model:

  • outer loop: how many channels do we have?
  • inner loop: how many messages does each channel produce?

Example:

  • imagen channel -> 3 results
  • gemini channel -> 2 results
  • openai channel -> 5 results

MergeGenerationResults forwards all 10 results.

It does not stop after one value per channel.

It stops reading from an input only when that input channel closes or the context is cancelled.

Where ch.Read Fits, and Where It Does Not

In previous steps, we already built three small helpers:

func Send[T any](ctx context.Context, out chan<- T, val T) error
func Receive[T any](ctx context.Context, in <-chan T) (T, bool, error)
func Read[T any](ctx context.Context, in <-chan T) <-chan T

At first glance, Merge may seem like a good place to reuse Read.

Instead of spelling out repeated receive logic for each input goroutine, it can range over a context-aware view of each input channel:

for result := range Read(ctx, in) {
	if err := Send(ctx, out, result); err != nil {
		return
	}
}

That is compositionally clean.

But it is not always the best trade-off.

Read creates one more output channel and one more forwarding goroutine.

Inside a helper like Merge, we are already inside a forwarding goroutine per input.

So this:

for val := range Read(ctx, in) {
	if err := Send(ctx, out, val); err != nil {
		return
	}
}

is often less direct than this:

for {
	val, ok, err := Receive(ctx, in)
	if err != nil || !ok {
		return
	}

	if err := Send(ctx, out, val); err != nil {
		return
	}
}

The second version avoids creating another channel just to immediately forward from it again.

So while Read is a useful helper, Merge is one of the places where staying with Receive can be the more practical choice.

With that trade-off in mind, the business-specific merge becomes:

func MergeGenerationResults(
	ctx context.Context,
	inputs ...<-chan GenerationResult,
) <-chan GenerationResult {
	out := make(chan GenerationResult)

	var wg sync.WaitGroup

	for _, in := range inputs {
		wg.Add(1)

		go func(in <-chan GenerationResult) {
			defer wg.Done()

			for {
				result, ok, err := Receive(ctx, in)
				if err != nil || !ok {
					return
				}

				if err := Send(ctx, out, result); err != nil {
					return
				}
			}
		}(in)
	}

	go func() {
		wg.Wait()
		close(out)
	}()

	return out
}

At this point, there is almost nothing generation-specific left.

Only the type remains.

That is when a generic helper becomes reasonable.

Not before.

ch.Merge

func Merge[T any](
	ctx context.Context,
	buffer int,
	inputs ...<-chan T,
) <-chan T {
	out := make(chan T, buffer)

	var wg sync.WaitGroup

	for _, in := range inputs {
		wg.Add(1)

		go func(in <-chan T) {
			defer wg.Done()

			for {
				val, ok, err := Receive(ctx, in)
				if err != nil || !ok {
					return
				}

				if err := Send(ctx, out, val); err != nil {
					return
				}
			}
		}(in)
	}

	go func() {
		wg.Wait()
		close(out)
	}()

	return out
}

Now the model comparison flow can use the generic helper:

func CompareModels(ctx context.Context) <-chan GenerationResult {
	return ch.Merge(
		ctx,
		0,
		GenerateWithImagen(ctx),
		GenerateWithGemini(ctx),
	)
}

The abstraction was not invented upfront.

It emerged from repeated fan-in coordination code.

Input and Output Ownership

Merge accepts input channels:

inputs ...<-chan T

These channels are created and closed by their producers.

Merge only receives from them.

So Merge must never close input channels.

The output channel is different:

out := make(chan T, buffer)

Merge creates this channel.

That means Merge owns it.

And because Merge owns the output channel, it is responsible for closing it after all input forwarding goroutines finish:

go func() {
	wg.Wait()
	close(out)
}()

This ownership rule is the real API contract:

  • producers close inputs
  • merge closes output

Ordering Guarantees

Merge does not preserve global ordering.

If one provider finishes faster, its result appears first.

That is the point.

Merge coordinates independent producers.

It does not serialize them.

Buffering

The output channel can be buffered:

out := make(chan T, buffer)

With:

buffer = 0

Each forwarded value requires the downstream consumer to be ready.

With:

buffer > 0

Merge can absorb short bursts before blocking.

But buffering does not remove backpressure.

It only delays it.

If the consumer stays slow, the buffer eventually fills.

Final Thought

Merge is not about combining data structures.

It is about coordinating independent producers into one controlled stream.

The business problem came first.

Manual fan-in came next.

The generic helper came last.

That order matters.