Errgroup Parallel API Fanout SetLimit

package main

import (
	"context"
	"errors"
	"fmt"
	"sync"

	"golang.org/x/sync/errgroup"
)

type ShardResult struct {
	ShardID int
	Rows    []string
}

func fetchShard(ctx context.Context, shard int) ([]string, error) {
	if shard == 3 {
		return nil, errors.New("shard 3 overloaded")
	}
	return []string{
		fmt.Sprintf("shard-%d-row-a", shard),
		fmt.Sprintf("shard-%d-row-b", shard),
	}, nil
}

func fanout(ctx context.Context, shards, concurrency int) ([]ShardResult, error) {
	g, ctx := errgroup.WithContext(ctx)
	g.SetLimit(concurrency)

	var mu sync.Mutex
	results := make([]ShardResult, 0, shards)
	var failures []error

	for shard := 0; shard < shards; shard++ {
		g.Go(func() error {
			rows, err := fetchShard(ctx, shard)
			if err != nil {
				mu.Lock()
				failures = append(failures, fmt.Errorf("shard %d: %w", shard, err))
				mu.Unlock()
				return nil // 单分片失败不取消全局，继续收其余结果
			}
			mu.Lock()
			results = append(results, ShardResult{ShardID: shard, Rows: rows})
			mu.Unlock()
			return nil
		})
	}

	_ = g.Wait()
	if len(failures) == shards {
		return nil, fmt.Errorf("all shards failed: %v", failures)
	}
	return results, nil
}

func main() {
	results, err := fanout(context.Background(), 6, 3)
	fmt.Println("shards succeeded:", len(results), "err:", err)
}