Day 032 — errgroup パッケージ

📚 背景知識（読んでから問題へ）

複数のgoroutineを起動し、そのうち1つでもエラーになったら全体を失敗として扱いたい、というケースは実務で非常に多いです（複数APIへの並列問い合わせ、複数ファイルの並列処理、複数DBクエリの並列実行など）。これをDay 012で学んだsync.WaitGroupだけで素朴に実装しようとすると、「エラーをどの変数に集約するか」「複数goroutineが同時にその変数へ書き込んだらどうするか」「1つが失敗した時点で残りのgoroutineを打ち切れないか」という3つの問題に必ずぶつかります。

golang.org/x/sync/errgroupは、Goチーム自身がメンテナンスする準標準パッケージで、この3つの問題をまとめて解決します。中心となるのは2つのAPIだけです。

• g, ctx := errgroup.WithContext(parent): *errgroup.Groupと、parentから派生したcontext.Contextを返す。このctxは、グループ内のどれか1つのgoroutineがエラーを返した瞬間に自動的にキャンセルされる
• g.Go(func() error { ... }): 新しいgoroutineを起動する。複数のgoroutineが同時にエラーを返しても、内部で最初の1つだけが安全に保持される（呼び出し側がmutexを書く必要はない）
• g.Wait(): sync.WaitGroup.Wait()と同様に全goroutineの終了を待つが、戻り値として保持された最初のエラー（またはnil）を返す

sync.WaitGroupが「待ち合わせ」だけを担当するのに対し、errgroup.Groupは「待ち合わせ + エラー集約 + キャンセル伝播」の3点セットを1つの型に凝縮しています。ただし、ctxがキャンセルされても、各goroutine自身がselect { case <-ctx.Done(): ... }のように能動的にチェックしなければ実際には止まらない、という点はDay 031のPipelineパターンと共通する設計です。

📝 問題

以下は、複数の商品IDについて価格を並列取得するfetchAll関数です。一見動きそうに見えますが、複数の欠陥を含んでいます。

📋 コピーpackage main

import (
	"fmt"
	"sync"
	"time"
)

type fetchResult struct {
	name  string
	price int
}

// fetchPrice は外部APIをシミュレートする。idが3の場合だけ失敗させる。
// 取得にかかる時間はidに比例するものとする（id=5が最も遅い）。
func fetchPrice(id int) (fetchResult, error) {
	time.Sleep(time.Duration(id) * 100 * time.Millisecond)
	if id == 3 {
		return fetchResult{}, fmt.Errorf("item %d: fetch failed", id)
	}
	return fetchResult{name: fmt.Sprintf("item-%d", id), price: id * 100}, nil
}

func fetchAll(ids []int) ([]fetchResult, error) {
	var wg sync.WaitGroup
	var results []fetchResult
	var firstErr error

	for _, id := range ids {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			r, err := fetchPrice(id)
			if err != nil {
				firstErr = err
				return
			}
			results = append(results, r)
		}(id)
	}
	wg.Wait()

	if firstErr != nil {
		return nil, firstErr
	}
	return results, nil
}

func main() {
	ids := []int{1, 2, 3, 4, 5}
	results, err := fetchAll(ids)
	if err != nil {
		fmt.Println("error:", err)
		return
	}
	fmt.Println(results)
}

設問

1. このコードをgo run -race main.goで実行するとWARNING: DATA RACEが検出されます。競合している変数を2つ挙げ、それぞれなぜ競合が起きるのか説明してください
2. データ競合を無視したとしても、このコードには「早期打ち切りの欠如」という設計上の無駄があります。ids := []int{1, 2, 3, 4, 5}の場合に何が無駄になるか、具体的な時間の目安とともに説明してください
3. golang.org/x/sync/errgroupを使ってfetchAllを書き直し、上記2つの欠陥（データ競合・早期打ち切りの欠如）を両方解消してください。結果の順序は保持しなくてよいですが、どのIDの結果かは分かるようにすること

要件

• fetchPriceはcontext.Contextを受け取れるように書き換え、キャンセル時には即座に処理を打ち切ってctx.Err()を返すこと（time.Sleepではなくselect { case <-time.After(d): case <-ctx.Done(): ... }を使う）
• go.modにgolang.org/x/syncへの依存を追加し、go run .でそのまま実行できる完全なコードとして提示すること
• 修正後のコードをgo run -race .で実行してもデータ競合が検出されないこと

🔍 ヒント（段階的開示）

📋 コピーg, ctx := errgroup.WithContext(context.Background())
results := make([]fetchResult, len(ids)) // インデックス書き込み用に事前確保

for i, id := range ids {
	i, id := i, id // ループ変数をgoroutineごとに固定
	g.Go(func() error {
		r, err := fetchPrice(ctx, id)
		if err != nil {
			return err
		}
		results[i] = r
		return nil
	})
}

if err := g.Wait(); err != nil {
	return nil, err
}
return results, nil

✅ 模範解答

ファイル構成: day032/go.mod + day032/main.go

📋 コピー// go.mod
module example.com/day032

go 1.22

require golang.org/x/sync v0.8.0

📋 コピーpackage main

import (
	"context"
	"fmt"
	"time"

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

type fetchResult struct {
	name  string
	price int
}

// fetchPrice は外部APIをシミュレートする。idが3の場合だけ失敗させる。
// time.Sleepの代わりにselectを使い、ctxのキャンセルで即座に打ち切れるようにしている。
func fetchPrice(ctx context.Context, id int) (fetchResult, error) {
	delay := time.Duration(id) * 100 * time.Millisecond
	select {
	case <-time.After(delay):
	case <-ctx.Done():
		return fetchResult{}, ctx.Err()
	}
	if id == 3 {
		return fetchResult{}, fmt.Errorf("item %d: fetch failed", id)
	}
	return fetchResult{name: fmt.Sprintf("item-%d", id), price: id * 100}, nil
}

// fetchAll はerrgroupを使い、データ競合なくエラー集約と早期打ち切りを両立する。
func fetchAll(ids []int) ([]fetchResult, error) {
	g, ctx := errgroup.WithContext(context.Background())
	results := make([]fetchResult, len(ids)) // 各goroutineが担当indexだけ書くのでmutex不要

	for i, id := range ids {
		i, id := i, id // ループ変数をgoroutineローカルに固定
		g.Go(func() error {
			r, err := fetchPrice(ctx, id)
			if err != nil {
				return fmt.Errorf("id=%d: %w", id, err)
			}
			results[i] = r
			return nil
		})
	}

	if err := g.Wait(); err != nil {
		return nil, err
	}
	return results, nil
}

func main() {
	start := time.Now()
	ids := []int{1, 2, 3, 4, 5}
	results, err := fetchAll(ids)
	elapsed := time.Since(start)

	if err != nil {
		fmt.Printf("error: %v (elapsed=%s)\n", err, elapsed)
		return
	}
	fmt.Printf("results=%v (elapsed=%s)\n", results, elapsed)
}

mkdir day032 && cd day032
go mod init example.com/day032
go get golang.org/x/sync@v0.8.0
# main.go を上記の内容で保存
go run .
go run -race .   # データ競合が検出されないことを確認

go.modにgolang.org/x/syncへの依存を追加し、go run .でそのまま実行できる完全なコードとして提示済みです。

🪜 Step-by-Step 解説

💡 設計思想・なぜこう書くのか

🌐 他言語との比較

🏆 実務での使いどころ

• 複数外部API/DBへの並列問い合わせ: ユーザープロフィールAPI・注文履歴API・レコメンドAPIを並列取得し、1つでも失敗したら残りを打ち切って早くエラーレスポンスを返す
• バッチ処理の並列ファイル処理: 複数ファイルを並列で読み込み・変換する際、1ファイルの破損を検知したら残りの処理を打ち切ってリソースを節約する
• マイグレーション・一括更新処理: 複数テーブル/シャードへの並列書き込みで、1つが失敗した時点で残りをキャンセルし、部分的な不整合を最小化する
• ヘルスチェックの並列実行: 複数の依存サービスへ並列にヘルスチェックを飛ばし、最初の異常検知で残りをキャンセルしてレイテンシを抑える

⚠️ よくある誤解・ミス

🚀 次のステップ

• 発展: g.SetLimit(n)を使い、idsが数百件ある場合でも同時実行数を制限しながらerrgroupでエラー集約・早期打ち切りを両立する版を実装してみましょう
• 次回予告: インターフェース設計の原則（小さいインターフェース）— 「Accept interfaces, return structs」の実践を、これまで学んだ並行処理コードのテスト容易性向上と絡めて学びます

🎯 自己評価

自分の回答

気づき・メモ