Bounded Channel Backpressure

package main

import (
	"context"
	"errors"
	"fmt"
	"sync"
	"time"
)

func produce(ctx context.Context, out chan<- int, total int) error {
	for i := 1; i <= total; i++ {
		select {
		case out <- i:
		case <-ctx.Done():
			return ctx.Err()
		}
	}
	return nil
}

func consume(ctx context.Context, in <-chan int, delay time.Duration) ([]int, error) {
	var seen []int
	for {
		select {
		case value, ok := <-in:
			if !ok {
				return seen, nil
			}
			time.Sleep(delay)
			seen = append(seen, value)
		case <-ctx.Done():
			return seen, ctx.Err()
		}
	}
}

func runPipeline(total, buffer int, delay, budget time.Duration) ([]int, error) {
	ctx, cancel := context.WithTimeout(context.Background(), budget)
	defer cancel()

	queue := make(chan int, buffer)
	var wg sync.WaitGroup
	wg.Add(1)
	go func() {
		defer wg.Done()
		defer close(queue)
		if err := produce(ctx, queue, total); err != nil && !errors.Is(err, context.DeadlineExceeded) {
			fmt.Println("producer stopped:", err)
		}
	}()

	seen, err := consume(ctx, queue, delay)
	wg.Wait()
	return seen, err
}

func main() {
	start := time.Now()
	seen, err := runPipeline(10, 2, 20*time.Millisecond, 5*time.Second)
	fmt.Printf("consumed %d items in %s (err=%v)\n", len(seen), time.Since(start).Round(10*time.Millisecond), err)

	short, err := runPipeline(10, 2, 20*time.Millisecond, 60*time.Millisecond)
	fmt.Printf("budget run consumed %d items, deadline=%v\n", len(short), errors.Is(err, context.DeadlineExceeded))
}