Go の並行処理は実務でどこに効くのか — channel を選ぶ判断基準と本番パターン

go func() と書けば goroutine が起動し、ch <- v で送って <-ch で受け取る。select で複数の channel を待てる。ここまでの文法は 30 分もあれば覚えられ、教科書の例も理解できます。

それなのに、いざ業務のコードを書くと、どこにも channel を使う場所が見つからない。あるいは逆に、無理に使ってみたら以前より読みにくく壊れやすいコードになった。

原因は、教科書が並行処理の「機構」を教えても「動機」を教えないことにあります。「1 から 100 までを 3 つの goroutine で足す」というサンプルは機構の説明としては正しいものの、実務で並行処理を導入する動機とは関係がありません。実務で必要なのは「この関数を並行にすべきか」「ここは channel か mutex か」「バッファーサイズをいくつにするか」という判断です。

この記事では、その判断の側から Go の並行処理を説明し直します。

先に解いておく 3 つの誤解

よくある 3 つの誤解があると、以降のパターンが脈絡のない「テクニック集」に見えてしまうので、先に解いておきます。

実務のサーバーアプリケーションで並行処理を入れる主な理由は、CPU を使い切ることではなく、待ち時間を重ねることとライフサイクルを制御することです。Web アプリケーションのプロファイルを取ると、時間の大半は DB・外部 API・ストレージの応答待ちで消えています。CPU 並列化が主役になるのは画像処理や暗号処理のような CPU バウンドな処理に限られます。

net/http サーバーは受け付けた接続ごとに goroutine を起動します（net/http/server.go の Server.Serve が go c.serve(connCtx) を呼びます。HTTP/1.1 の keep-alive では同一接続上のリクエストはその goroutine 内で逐次処理され、HTTP/2 ではストリームごとに goroutine が起動します）。いずれにせよ、異なるクライアントのハンドラーは同時に走ります。Go で Web アプリケーションを書いた時点で、それはもう並行プログラムです。ハンドラーの中で触るグローバル変数、共有キャッシュ、コネクションプールはすべて並行アクセスされます。並行処理を使うかどうかはすでに決まっていて、実務で問うべきは、並行に動いているプログラムのどこに構造を入れるかです。

これは間違いではありませんが、実務での比重を見誤らせます。実際のプロダクションコードで channel が担っているのは、データそのものよりも、タイミング（いつ進んでよいか）と所有権（今どの goroutine がこの値に触ってよいか）です。<-ctx.Done() はデータを運びませんが、実務で最も頻繁に書かれる channel 操作です。

以降は次の順で進めます。

1. 判断フレーム（第1章） — そもそも並行にすべきか、channel と mutex のどちらを選ぶか
2. 本番パターン 6 種（第2〜7章） — fan-out / 有界並行と背圧 / ライフサイクル管理 / 所有権の直列化 / バッチング / 購読
3. 事故カタログ（第8章） — 本番で実際に踏むバグと、Go 1.27 時点での検出手段
4. 道具箱（第9章） — 標準・準標準ライブラリーの使い分け
5. ケーススタディー（第10章） — Webhook 配信ワーカーを段階的に組み立てる
6. テストと観測（第11章）

この記事のコードはすべて Go 1.27 で検証しています（go vet と go test -race を通過）。扱うのは並行処理を「使う側」の話だけです。ランタイムの内部実装（G・M・P モデル、work stealing、netpoller）は別記事「Go スケジューラーの仕組み」で解説しています。

第1章 判断フレーム — 並行にすべきか、channel を使うべきか

1.1 並行処理を導入する 4 つの動機

実務のコードベースを観察すると、並行処理が入る理由はほぼこの 4 つに収まります。今どれを求めているのかを言葉にしてから、コードを書き始めてください。動機が違えば適切な構造も違うからです。

CPU バウンドな計算の並列化は 5 番目の動機ですが、典型的な Web バックエンドでは稀です。該当する場合も、「GOMAXPROCS 個のワーカーで分割する」という単純な話で済むことが多く、第3章・第4章で扱う背圧やライフサイクルの議論はあまり効いてきません。

①と②は別の目標です。①（レイテンシー）は「1 件の仕事を細かく割って同時に投げる」ことで達成され、②（スループット）は「多数の仕事を詰まらせずに流し続ける」ことで達成されます。①のつもりで無制限に goroutine を撒くと、②が崩れます（第3章）。

1.2 channel と mutex のどちらを選ぶか

Go には有名な標語があります。

Don't communicate by sharing memory; share memory by communicating.

この標語は Go の設計思想を表していますが、「常に channel を使え」という意味ではありません。Go 公式の Go Wiki: Use a sync.Mutex or a channel? は、この標語を引いたうえで、どちらを使うべきかに次のように答えています。

Use whichever is most expressive and/or most simple. （最も表現力が高い、あるいは最も単純な方を使え）

同ページは「channel が向く場合」として「データの所有権の受け渡し」「作業単位の分配」「非同期の結果通知」を、「mutex が向く場合」として「キャッシュ」「状態」を挙げています。さらに、こうも明記しています。

A common Go newbie mistake is to over-use channels and goroutines just because it's possible, and/or because it's fun. （よくある初心者の誤りは、できるから／楽しいからという理由だけで channel と goroutine を使いすぎることだ）

実務での判断基準は次のとおりです。

• 守りたいのがデータ構造そのもの（マップ、カウンター、設定値）
• クリティカルセクションが短く、その中で I/O をしない
• 操作が「読む」「書く」に還元でき、順序に意味がない

• 値の所有権を別の goroutine に渡す（渡したら元の goroutine はもう触らない）
• タイミングを伝えたい（完了した、キャンセルされた、時間が来た）
• 流量を制御したい（送り手を待たせたい ＝ 背圧）
• 複数の待機イベントを select で合流させたい

最後の点は channel にしかできません。select は「複数のイベント源のうち、最初に来たものに反応する」処理を安全に書ける唯一の構文です。「結果が来る」「タイムアウトした」「キャンセルされた」の 3 つを同時に待つ処理は、mutex では書けません。

1.3 コストを測る — channel は安くない

判断材料として実測値を示します。同じ「カウンターを増やす」処理を 3 通りで実装しました。

type mutexCounter struct {
    mu sync.Mutex
    n  int64
}
 
func (c *mutexCounter) Inc() {
    c.mu.Lock()
    c.n++
    c.mu.Unlock()
}
 
type atomicCounter struct{ n atomic.Int64 }
 
func (c *atomicCounter) Inc() { c.n.Add(1) }
 
// 「状態を 1 つの goroutine が所有する」アクター方式
type chanCounter struct {
    inc  chan struct{}
    stop chan struct{}
    n    int64
}
 
func newChanCounter() *chanCounter {
    c := &chanCounter{inc: make(chan struct{}), stop: make(chan struct{})}
    go func() {
        for {
            select {
            case <-c.inc:
                c.n++
            case <-c.stop:
                return
            }
        }
    }()
    return c
}
 
func (c *chanCounter) Inc() { c.inc <- struct{}{} }
 
// 注: このベンチ用ハーネスは stop を閉じていません（計測プロセスの寿命と
// 同じなので意図的です）。実務コードでは 4.2 の原則 2 に従ってください。

b.RunParallel で競合させた結果です。

goos: windows, goarch: amd64, go1.27
cpu: 12th Gen Intel(R) Core(TM) i7-12700K
GOMAXPROCS=8, -benchtime=2s -count=5（中央値）
 
BenchmarkCounterAtomic-8      14.3 ns/op
BenchmarkCounterMutex-8       50.2 ns/op
BenchmarkCounterChannel-8    441.9 ns/op

絶対値は環境で大きく変わりますし、この測定自体もノイズを含みます（同一環境の 5 回試行でも ±30% 程度は動きます）。読むべきは桁の関係です。単純な状態更新において、channel によるアクター方式は mutex よりおおよそ 1 桁遅く、atomic とは 30 倍程度の差があります。バッファーなし channel の送受信は goroutine の park / unpark（スケジューラーへの制御移譲）を伴うため、このコストは構造上避けられません。

だからといって「channel は遅いから使うな」という結論にはなりません。channel は 1 回あたり数百ナノ秒を払う価値のある構造を表すのに使い、単なる排他制御には使わないでください。

1 リクエストで 1 回だけ channel を使うなら、441ns は無視できます（ネットワーク I/O はミリ秒単位で、3〜4 桁上のオーダーです）。逆に、ホットループの中で 1 秒あたり数百万回カウンターを更新するなら、mutex ではなく atomic を選んでください。

1.4 判断のフローチャート

図の下 2 段は、goroutine を起動するパターンなら必ず通す確認です。ctx を全経路に通したか、すべての goroutine に終了経路があるかを確かめます。8.1 で挙げる goroutine リークの 4 大原因は、どれもこの 2 つのどちらかを欠いたコードで起きます。

第2章 パターン① fan-out / fan-in — 待ち時間を重ねる

2.1 実務での出現場所

待ち時間の重ね合わせ（動機①）は、実務で並行処理を入れる理由として最も多く、最も分かりやすいものです。典型例は BFF（Backend for Frontend）や集約 API です。

1 つの画面を描くために、ユーザー情報・注文履歴・レコメンド・バナーの 4 つを別々のサービスから取ってくるとします。それぞれ 100ms かかるので、直列に呼べば 400ms、並行に呼べば 100ms +α です。

// 直列版：400ms
func (c *Clients) BuildPageSerial(ctx context.Context, uid string) (*Page, error) {
    u, err := c.User.Get(ctx, uid)
    if err != nil {
        return nil, err
    }
    o, err := c.Order.List(ctx, uid)
    if err != nil {
        return nil, err
    }
    r, err := c.Reco.For(ctx, uid)
    if err != nil {
        return nil, err
    }
    b, err := c.Banner.Fetch(ctx)
    if err != nil {
        return nil, err
    }
    return &Page{User: u, Orders: o, Recos: r, Banners: b}, nil
}

このコードを並行にできるのは、4 つの呼び出しの間にデータ依存がないからです。もし o を取るのに u.TenantID が必要なら、その 2 つは直列にせざるを得ません（依存グラフの深さがレイテンシーの下限になります）。

2.2 素朴な並行化と、その 4 つの穴

多くの人が最初に書くのはこれです。

// 問題のある実装
func (c *Clients) BuildPageNaive(ctx context.Context, uid string) (*Page, error) {
    var (
        wg   sync.WaitGroup
        page Page
        mu   sync.Mutex
        errs []error
    )
    wg.Go(func() {
        u, err := c.User.Get(ctx, uid)
        mu.Lock()
        defer mu.Unlock()
        if err != nil {
            errs = append(errs, err)
            return
        }
        page.User = u
    })
    wg.Go(func() {
        o, err := c.Order.List(ctx, uid)
        mu.Lock()
        defer mu.Unlock()
        if err != nil {
            errs = append(errs, err)
            return
        }
        page.Orders = o
    })
    wg.Wait()
    return &page, errors.Join(errs...)
}

（wg.Go は Go 1.25 で追加されたメソッドです。wg.Add(1) と defer wg.Done() の対応漏れという古典的なバグを構造的に排除できるため、新規コードでは常にこちらを使うべきです。）

これでも動きますが、本番に出すには次の 4 つの穴があります。

1. キャンセルが伝播しない — User の取得が 500ms で失敗しても、Order は最後まで走り切ります。ユーザーはもう結果を受け取らないのに、下流サービスの負荷だけがかかり続けます。
2. エラーの意味が失われる — errors.Join で全部束ねると、呼び出し側は必須データの失敗と任意データの失敗を区別できません。
3. panic がプロセスを殺す — goroutine 内の panic は、その goroutine の中で recover されない限りプロセス全体を落とします。呼び出し元の defer recover() では拾えません。
4. ロックの用途を取り違えている — page.User と page.Orders は別々のメモリー位置なので、ページのフィールドを守るためのロックは要りません。この mu が実際に守っているのは errs への append の方です（詳細は 2.4）。

2.3 errgroup による実装

golang.org/x/sync/errgroup はこの用途のための準標準ライブラリーです。API は次の 5 つです。

func WithContext(ctx context.Context) (*Group, context.Context)
func (g *Group) Go(f func() error)
func (g *Group) TryGo(f func() error) bool
func (g *Group) SetLimit(n int)
func (g *Group) Wait() error

errgroup.WithContext が返す ctx は、Go に渡した関数のいずれかが非 nil のエラーを返した時点、または Wait が返った時点のどちらか早い方でキャンセルされます。前者の条件があるので、素朴版と違って 1 つの失敗が残りのタスクのキャンセルにつながります。後者の条件のため、g.Wait() の後にこの ctx を使い回すと、全タスクが成功していても context.Canceled になります。派生 ctx の寿命はグループの寿命と一致します。

func (c *Clients) BuildPage(ctx context.Context, uid string) (*Page, error) {
    g, ctx := errgroup.WithContext(ctx) // 以降はこの ctx を使う
    var page Page
 
    // 必須項目：1 つでも落ちたらページを出せない
    g.Go(func() error {
        u, err := c.User.Get(ctx, uid)
        if err != nil {
            return err
        }
        page.User = u
        return nil
    })
    g.Go(func() error {
        o, err := c.Order.List(ctx, uid)
        if err != nil {
            return err
        }
        page.Orders = o
        return nil
    })
 
    // 任意項目：失敗しても全体を落とさない（nil を返す）
    g.Go(func() error {
        rctx, cancel := context.WithTimeout(ctx, 80*time.Millisecond)
        defer cancel()
        r, err := c.Reco.For(rctx, uid)
        if err != nil {
            slog.WarnContext(ctx, "recommendation degraded", "err", err)
            return nil // ここが設計判断
        }
        page.Recos = r
        return nil
    })
 
    if err := g.Wait(); err != nil {
        return nil, err
    }
    return &page, nil
}

元の ctx を上書きすることで、以降で誤って「キャンセルされない方の ctx」を使う事故を防ぎます。逆に「グループのキャンセルに巻き込まれたくない処理」がある場合は、意図的に別名で元の ctx を保持してください。

errgroup は「1 つでも失敗したら全体を諦める」というセマンティクスを持ちます。これは必須項目には正しく、任意項目には強すぎます。レコメンドが取れなくてもページは出したい、という要件はよくあります。その場合は goroutine 内でエラーを握り潰し（ログは残す）、nil を返します。

errgroup に返す error が表すのは、エラーが起きたかどうかではなく、そのサブタスクの失敗がグループ全体を中止する理由になるかどうかです。ここを取り違えると、任意項目の一時的な失敗でページ全体が 500 になります。

必須項目より短いタイムアウトを与えることで、「あってもなくてもよいデータのために全体が待たされる」事態を防ぎます。上の例では 80ms を超えたらレコメンドを諦めます。

2.2 で挙げた 4 つの穴のうち、穴 3（panic）は errgroup でも埋まりません。errgroup は goroutine 内の panic を意図的に伝播させないため、g.Go に渡した関数が panic すればプロセスが落ちます。理由と対策は 3.2 と 8.7 で扱います。

2.4 なぜ page にロックが要らないのか

page.User と page.Orders に別々の goroutine から書き込んでいますが、これは race ではありません。Go のメモリーモデルにおいて、構造体の異なるフィールドは異なるメモリー位置であり、異なるメモリー位置への並行書き込みは競合しません。読み出し側は g.Wait() の後にのみ page を読むため、Wait が張る happens-before 関係によって全書き込みが可視になります。

ただし、この推論が成り立つのは次の条件をすべて満たす場合だけです。

• 各 goroutine が互いに素なフィールド集合にのみ書く
• フィールドがスライスやマップを共有していない（page.Orders と page.Recos が同じ backing array を指していない）
• 読み出しは必ず Wait() の後

逆に、同じスライスに複数 goroutine が append するのは明確な race です（append は長さと backing array の両方を読み書きするため）。結果を集める必要があるときは、インデックス固定のスライス（results[i] = v）を使うか、channel で集約します。

// OK: 各 goroutine が自分のインデックスだけに書く
results := make([]Result, len(inputs))
g, ctx := errgroup.WithContext(ctx)
for i, in := range inputs {
    g.Go(func() error {
        r, err := fetch(ctx, in)
        if err != nil {
            return err
        }
        results[i] = r // 異なるインデックス = 異なるメモリ位置
        return nil
    })
}

（このループが安全に書けるのは Go 1.22 でループ変数がイテレーションごとのスコープになったためです。Go 1.21 以前は i, in := i, in の再宣言が必須で、go.mod の go ディレクティブが 1.21 以前のモジュールでは今も必要です（8.6）。）

2.5 fan-out は tail レイテンシーを増幅する

fan-out の効果を「400ms → 100ms」と説明しましたが、本番ではもう少し悲観的に見積もる必要があります。合流点のレイテンシーは 4 つのうち最も遅いもので決まるため、各下流の p99 が 500ms なら、4 つ並行に呼んだときの全体の p99 は 500ms より悪くなります。遅い下流が 1 つでもあると、並行化の恩恵はそこで頭打ちになります。

4 つの独立な呼び出しがそれぞれ 1% の確率で遅い場合、少なくとも 1 つが遅い確率は 1 - 0.99^4 ≒ 3.9% です。fan-out は個々の遅さを増幅します。2.3 の「任意項目には個別タイムアウト」が効くのはこのためです。

対策の選択肢は、優先度順に次のとおりです。

1. 必須項目を減らす — 本当に全部そろわないとレスポンスを返せないのか
2. 任意項目に短いタイムアウトを切る — 上限を構造で保証する
3. キャッシュで下流呼び出し自体を消す — 最も効くが実装コストが高い
4. ヘッジリクエストを送る（一定時間経過後に 2 本目を投げる）— 効くが下流負荷が増えるので、必ず割合を制限する

2.6 errgroup と sync.WaitGroup の使い分け

Go 1.25 で WaitGroup.Go が入り、両者の見た目は近づきましたが、判断基準は単純です。エラーを返すタスクなら errgroup、返さないなら WaitGroup を使います。

「依存を増やしたくない」という理由で WaitGroup にエラー集約を手書きするのは、たいてい割に合いません（golang.org/x/sync は Go チームが管理する準標準ライブラリーです）。

第3章 パターン② 有界並行と背圧 — 「全部 goroutine にする」が事故になる理由

3.1 最も多い並行処理の事故

第2章の fan-out は「4 つ」という固定の小さな数だから安全でした。ここで扱うのは、個数が入力データで決まるケースです。

// 本番を落とすコード
func ProcessAllUnbounded(ctx context.Context, items []Item) {
    var wg sync.WaitGroup
    for _, it := range items {
        wg.Go(func() { _ = process(ctx, it) })
    }
    wg.Wait()
}

items が 10 件なら問題ありません。10 万件になった瞬間に本番が落ちます。goroutine 自体は軽量です。初期スタックは 2KB で、Go 1.19 以降はランタイムが観測した平均スタック使用量をもとに、より大きなスタックで goroutine を開始することもあります。問題になるのは goroutine が掴むリソースです。

• DB コネクションプール枯渇 — database/sql の SetMaxOpenConns(20) に対して 10 万本が殺到すると、99,980 本がプール待ちでブロックします。待ち時間はやがてアプリ側のタイムアウトを超え、全リクエストがエラーになります
• 下流サービスへの意図しない DoS — 自分のサービスは耐えても、呼び出し先が 10 万本の同時接続で死にます。そして障害が連鎖します
• メモリー — 各 goroutine が 1MB のバッファーを持てば 100GB です。goroutine が軽くてもペイロードは軽くありません
• ファイルディスクリプター枯渇 — too many open files でプロセス全体が機能停止します

厄介なのは、このコードが開発環境とステージングでは決して落ちないことです。テストデータが 10 件しかないので、落ちるのは本番だけです。

3.2 並行数を制限する 3 つの手段

func ProcessAllErrgroup(ctx context.Context, items []Item, limit int) error {
    g, ctx := errgroup.WithContext(ctx)
    g.SetLimit(limit)
    for _, it := range items {
        // 実行中が limit 本に達していると、ここでブロックする
        g.Go(func() error { return process(ctx, it) })
    }
    return g.Wait()
}

SetLimit(n) 以降、g.Go は実行中の goroutine が n 本に達しているとブロックします。エラー集約・キャンセル・有界化がこれだけで揃うので、大半のケースはこの手段で足ります。

注意点があります。実装を見ると g.Go の待機は素の channel 送信です（errgroup.go）。

func (g *Group) Go(f func() error) {
    if g.sem != nil {
        g.sem <- token{} // select ではない = ctx を見ていない
    }
    ...
}

そのため、g.Go がブロックしている間に ctx がキャンセルされても、g.Go は枠が空くまで戻りません。長時間タスクを大量に投入するループでキャンセル応答性が要るなら、TryGo（枠がなければ即 false を返す）を使うか、次の手段 2 のように select で待ち先を明示します。

また、2.3 で触れたとおり、errgroup は goroutine 内の panic を伝播しません。そう設計した理由は、ソースコードのコメントに書かれています（panic を遅延させるとバグの発見が遅れる、クラッシュ監視ツールからスタックが隠れる、など）。

SetLimit には前提条件もあります。グループ内に実行中の goroutine がある間は上限を変更してはいけません（既存の上限を変えると panic します）。必ず最初の Go より前に設定してください。

func ProcessAllSemaphoreChan(ctx context.Context, items []Item, limit int) error {
    sem := make(chan struct{}, limit)
    var (
        wg   sync.WaitGroup
        mu   sync.Mutex
        errs []error
    )
    for _, it := range items {
        select {
        case sem <- struct{}{}: // 枠を取る（満杯ならブロック）
        case <-ctx.Done():
            wg.Wait()
            return ctx.Err()
        }
        wg.Go(func() {
            defer func() { <-sem }() // 枠を返す
            if err := process(ctx, it); err != nil {
                mu.Lock()
                errs = append(errs, err)
                mu.Unlock()
            }
        })
    }
    wg.Wait()
    return errors.Join(errs...)
}

make(chan struct{}, limit) に入れられる値の数が、同時実行数の上限になります。手段 1 との違いは、select を使うことで「枠を待つ」と「キャンセルされる」を同時に待てる点です。

このイディオムは、channel がデータではなくタイミングを運ぶ例（誤解 3）でもあります。ここで送っている struct{}{} にはデータとしての意味がありません。運んでいるのは「進んでよい」という許可だけです。

func ProcessWeighted(ctx context.Context, items []Item, budgetMB int64) error {
    sem := semaphore.NewWeighted(budgetMB)
    g, ctx := errgroup.WithContext(ctx)
    for _, it := range items {
        // 重みが budgetMB を超えると Acquire は永久に待つ（後述）ので必ず丸める
        cost := min(estimateMB(it), budgetMB)
        if err := sem.Acquire(ctx, cost); err != nil {
            break // ctx キャンセル
        }
        g.Go(func() error {
            defer sem.Release(cost)
            return process(ctx, it)
        })
    }
    return g.Wait()
}

「同時 10 本」ではなく「同時に使うメモリー 2GB まで」を制限したい場合に使います。動画変換やバッチ集計のように、1 タスクのコストが桁で違うワークロードで効きます。

min で丸めているのは必須の防御です。semaphore.go の実装を読むと、Acquire(ctx, n) は n がセマフォの総容量を超える場合、エラーを返さずに <-ctx.Done() でブロックします（セマフォがすべて空いていても決して満たせないため）。3GB の項目を 2GB の予算に投げると、キャンセルされるまでループが停止します。呼び出し側で丸めるか、事前に弾いてください。

3.3 並行数をどう決めるか — 勘で決めない

limit の値を「とりあえず 10」「CPU 数」で決めているコードは多いのですが、根拠を持って決める方法があります。

runtime.GOMAXPROCS(0) を上限にします。それ以上増やしてもコンテキストスイッチが増えるだけです。Go 1.25 以降、GOMAXPROCS の既定値は Linux の cgroup CPU 制限を考慮するようになったため、コンテナー環境でもこの値は信頼できます（詳細は「Go スケジューラーの仕組み」を参照）。

実務ではこちらのケースがはるかに多くなります。並行数の上限を決めるのは自分の CPU ではなく、下流が受け入れられる量です。必要な並行数は、リトルの法則で計算できます。

L = λ × W
 
L : 系の中に同時に存在する仕事の数（＝必要な並行数）
λ : スループット（件/秒）
W : 1 件あたりの滞在時間（秒）

例えば、下流 API が「200 req/s まで」と契約で決まっていて、1 リクエストの平均応答時間が 50ms だとします。

L = 200 × 0.05 = 10

並行数 10 で、ちょうど 200 req/s が出ます。これを 100 にしても下流のレート制限に弾かれるだけで、スループットは増えずエラーとリトライだけが増えます。逆に 5 にすると 100 req/s しか出ません。

（厳密には、この換算は W が一定であることを前提にしています。下流が速くなったときに契約レートを超えてしまう問題は 3.3.1 で扱います。）

この計算を使うと、「並行数」という設定値を「守るべき契約」から導出できます。下流の許容量が変わったら、並行数もそれに合わせて変えるべきだと分かります。

もうひとつよくある制約が DB コネクションプールです。

アプリ 1 インスタンスの MaxOpenConns = 20
1 タスクが同時に掴むコネクション = 1
→ DB を触るワーカーの並行数上限は 20 未満にすべき
（HTTP ハンドラ側も同じプールを使うことを忘れずに）

バッチワーカーとオンラインリクエストが同じプールを共有していることに気づかず、バッチを流した瞬間に API が全滅する、というのは定番の障害です。

3.3.1 並行数制限とレート制限は別物

リトルの法則の式で言えば、並行数制限とレート制限は縛る変数が違います。多くの実装がここを混同しています。

L = λ × W

• セマフォ（SetLimit、semaphore.Weighted）が縛るのは L — 同時に「進行中」の仕事の数
• レートリミッターが縛るのは λ — 単位時間あたりに「開始する」仕事の数

セマフォは速度を保証しないので、セマフォだけでは λ を守れません。3.3 の計算（並行数 10 ＝ 200 req/s）は W が 50ms のままなら成り立ちますが、下流が速くなって W が 5ms になった瞬間、同じ並行数 10 で 2000 req/s に達し、レート制限を超えます。

// golang.org/x/time/rate
 
// 平常時 200 req/s、瞬間的なバーストは 20 まで許す
limiter := rate.NewLimiter(rate.Limit(200), 20)
 
g, ctx := errgroup.WithContext(ctx)
g.SetLimit(10) // L: 同時実行数（プールやメモリの制約）
 
for _, it := range items {
    g.Go(func() error {
        // λ: 開始速度。トークンが貯まるまでブロックする（ctx で中断可能）
        if err := limiter.Wait(ctx); err != nil {
            return err
        }
        return process(ctx, it)
    })
}

rate.Limiter の 3 つのメソッドは、そのまま 3.5 の 3 戦略に対応します。

Reserve() は呼んだ時点でトークンを消費します。返ってきた *Reservation の Delay() を見て「待てないからやめる」と判断したら、必ず r.Cancel() でトークンを返してください。返さないと実効レートが静かに下がります（ReserveN(t, n) で n がバースト値を超える場合は OK() == false の予約が返ります）。

第 2 引数のバースト値（上の 20）は、3.4 のバッファーサイズと同じ性質の設計判断です。「瞬間的に何件までなら一気に投げてよいか」を宣言しています。

実務では両方必要になることがほとんどです。レートリミッターだけでは、下流が詰まったときに進行中の仕事が無限に積み上がります（λ は守れても L が発散する）。セマフォだけでは、下流が速いときに契約速度を超えます。片方しかない実装を見たら、もう片方の制約が本当にないのかを確認してください。

3.4 背圧（backpressure）とは何か

channel を「非同期キュー」として捉えていると、背圧の役割は見えてきません。

バッファーなし channel は「完全な背圧」を意味します。

ch := make(chan Item) // バッファなし
ch <- item            // 受信側が受け取るまでブロックする

送信側は受信側が受け取るまで進めません。この待ちが背圧として働き、生産者は消費者のペースに合わせて進むことになります。そのため生産が消費より速くても、キューが無限に伸びてメモリーを食い潰すことは構造的に起こりません。

バッファー付き channel は「N 個分のバーストを吸収してよい」という宣言です。

ch := make(chan Item, 100) // 100 個までは送信側がブロックしない

「バッファーを増やせば速くなる」はよくある誤解です。バッファーで実際に出る処理速度が上がることはありますが、上限は上がりません。上限は消費側の処理能力 μ で決まります。

バッファーで得られるのは次の 2 つです。

1. 一時的なバーストの吸収 — μ を超えた瞬間的な流入を、後で消化する約束で受け取る
2. 生産者と消費者のスケジューリングの分離 — ランデブーのたびに払う goroutine の park / unpark（1.3 で測った数百 ns）を償却し、両者を重ねて走らせる

2 つ目の効果は、パイプラインのように両側が実際の処理をする構成では、計測で確認できるほど大きくなります。ただし生産が消費を恒常的に上回っているなら、バッファーは遅かれ早かれ満杯になり、そこから先の挙動はバッファーなしと同じです。

したがってバッファーサイズは性能のつまみではなく、どれだけの滞留を許容するかという設計判断です。

バッファーサイズは「何秒分の滞留を許すか」から逆算して決めます。1 件 10ms で処理でき、最大 1 秒の遅延を許容するなら、バッファーは 100 です。同じ 100 でも、「100 くらいでいいか」で決めた値には根拠がありません。

3.5 満杯のときどうするか — 3 つの戦略

背圧を設計するとは、キューが満杯になったときの振る舞いを明示的に決めることです。決めていないコードは、暗黙のうちに「ブロックする」を選んでいます。

// 戦略A: ブロックする（背圧を上流に伝える）
select {
case queue <- item:
case <-ctx.Done():
    return ctx.Err()
}
 
// 戦略B: 捨てる（負荷制限 / load shedding）
select {
case queue <- item:
default:
    metrics.Dropped.Inc()
    return ErrOverloaded
}
 
// 戦略C: 期限つきで待ち、駄目なら諦める
timer := time.NewTimer(50 * time.Millisecond)
defer timer.Stop()
select {
case queue <- item:
case <-timer.C:
    metrics.Dropped.Inc()
    return ErrOverloaded
case <-ctx.Done():
    return ctx.Err()
}

どの戦略を選ぶかは、データの性質で決まります。

戦略 B（捨てる）は諦めに見えますが、システム全体を守るために意図して選ぶ戦略です。満杯時に無条件でブロックすると、上流のリクエストハンドラーが次々と滞留し、goroutine 数とメモリーが増え続け、最終的にサービス全体が応答不能になります（典型的なカスケード障害です）。早期に 503 を返して負荷を落とす方が、システム全体としては健全です。

なお、戦略 B の default 節は「即座に諦める」という意味なので、for ループの中で使うとビジーループになります。リトライしたいなら必ず戦略 C のように待ち先を持たせてください。

3.6 キューを観測可能にする

背圧の設計が正しかったかどうかは、動かしてみないと分かりません。len(ch) と cap(ch) をメトリクスとして出すだけで、判断材料が大きく増えます（以下の metrics は Prometheus クライアントなど、アプリ側のメトリクスパッケージを指します）。

// 定期的にキュー滞留を計測する
func (d *Dispatcher) observeQueues(ctx context.Context, interval time.Duration) {
    t := time.NewTicker(interval)
    defer t.Stop()
    for {
        select {
        case <-ctx.Done():
            return
        case <-t.C:
            for i, lane := range d.lanes {
                metrics.QueueDepth.WithLabelValues(strconv.Itoa(i)).
                    Set(float64(len(lane)))
            }
        }
    }
}

読み方は次のとおりです。

「常に満杯付近」のときにバッファーを増やしても、問題を隠してレイテンシーを悪化させるだけです。滞留したアイテムは、それだけ古い（＝価値の下がった）データでもあります。

Go 1.26 では goroutine の状態別内訳が runtime/metrics に追加され、ランタイム側の飽和も直接読み取れるようになりました（こちらの metrics は標準ライブラリーの runtime/metrics です）。

// runtime/metrics
 
samples := []metrics.Sample{
    {Name: "/sched/goroutines:goroutines"},          // 総数（Go 1.16〜）
    {Name: "/sched/goroutines/runnable:goroutines"}, // 実行可能だが実行されていない（Go 1.26〜）
    {Name: "/sched/goroutines/running:goroutines"},  // 実行中（Go 1.26〜）
    {Name: "/sched/goroutines/waiting:goroutines"},  // I/O や同期プリミティブ待ち（Go 1.26〜）
    {Name: "/sched/latencies:seconds"},              // スケジューラ待ち時間の分布（Go 1.17〜）
}
metrics.Read(samples)

読み分けの目安は次のとおりです。

• runnable が慢性的に高い → CPU が足りていない。並行数を増やしても悪化するだけ
• waiting が高いまま安定していて runnable が低く CPU に余裕がある → I/O 待ちが支配的。並行数を増やす余地がある
• waiting が単調増加している → リーク（8.1）か下流の停止。並行数を増やすと悪化する。goroutine プロファイルでスタックを確認する
• /sched/latencies:seconds の裾が伸びている → goroutine がスケジューラーで待たされている

これらの指標があれば、「並行数を増やすべきか」に勘ではなく数字で答えられます。

第4章 パターン③ ライフサイクル管理 — 実務で最も多い channel の用途

4.1 <-ctx.Done() こそが実務の channel

冒頭で「どこにも channel を使う場所が見つからない」と書きましたが、実際には多くの人がすでに毎日 channel を使っています。context.Context の Done() がそれです。

type Context interface {
    Deadline() (deadline time.Time, ok bool)
    Done() <-chan struct{}   // これが channel
    Err() error
    Value(key any) any
}

context の設計は、channel の 2 つの性質を利用しています。

これが「1 対多のブロードキャスト」を実現します。cancel() が呼ばれると close(done) が実行され、<-ctx.Done() で待っていたすべての goroutine が同時に解放されます。値を送る方式では受信者 1 人にしか届かないので、この用途には close を使うしかありません。

標準ライブラリーの実装（context/context.go）もこのとおりです。

// cancelCtx.cancel の抜粋
func (c *cancelCtx) cancel(removeFromParent bool, err, cause error) {
    // ...
    c.mu.Lock()
    // ...
    d, _ := c.done.Load().(chan struct{})
    if d == nil {
        c.done.Store(closedchan)
    } else {
        close(d)          // ここで全待機者を一斉に起こす
    }
    for child := range c.children {
        child.cancel(false, err, cause) // 子へ再帰的に伝播
    }
    // ...
}

context.Background() の Done() は nil を返します。select で nil channel を待つ case は決して選ばれません。そのため「キャンセルされないコンテキスト」を特別扱いなしに表現できます。この性質は 8.4 で実務パターンとして再登場します。

4.2 キャンセル伝播の原則

実務で守るべき原則は 3 つです。

ctx を受け取っておきながら下流の呼び出しに渡していないコードは、キャンセルの鎖を断ち切っています。contextcheck linter で機械的に検出できます（go vet の lostcancel は後述の原則 3 専用で、伝播漏れは見ません）。

goroutine を起動する関数は、その goroutine が確実に終了する経路を用意する責任を負います。「いつ終わるか説明できない go 文」はバグです。

この原則には専用の道具があります。次のような、「ctx がキャンセルされたら後始末をする」だけの見張り goroutine を考えます。

// よく見るが、リークしやすい書き方
go func() {
    select {
    case <-ctx.Done():
        conn.Close()
    case <-done: // これを閉じ忘れると goroutine が残る
    }
}()

これは Go 1.21 の context.AfterFunc で置き換えられます。

stop := context.AfterFunc(ctx, func() { conn.Close() })
defer stop() // 登録解除

利点は、ctx の完了を待つ goroutine を常駐させないことです。f は親の children に登録されるだけで、キャンセル時になって初めて専用の goroutine で実行されます（context.go の afterFuncCtx.cancel が a.once.Do(func() { go a.f() }) を呼びます）。待機しているだけの goroutine が存在しないので、それがリークすることもありません。

stop() の戻り値から分かることは限られています。ドキュメントは「f の実行を止められたなら true」「false は『ctx が完了して f が起動済み』か『すでに停止済み』のどちらか」と定めており、さらに stop は f の完了を待ちません。true なら「後始末はまだ走っていない」と分かりますが、false から「後始末が完了した」は導けません。完了を知りたければ f 側と明示的に同期してください。

なお stop() を呼ばないと、f とそのクロージャが掴んでいるオブジェクトは親コンテキストが生きている限り解放されません。原則 3 の defer cancel() と同じ種類の落とし穴です。

ctx, cancel := context.WithTimeout(parent, 3*time.Second)
defer cancel() // 早く終わっても必ず呼ぶ

cancel() を呼ばないと、親コンテキストの children マップから自分が外れず、親が生きている限りメモリーが解放されません。タイムアウトが来れば自動で解放されますが、それまでの間リークします。この呼び忘れは go vet の lostcancel で検出できます。

4.3 タイムアウトの階層設計

実務で見落とされがちなのが、タイムアウトが入れ子になっていることです。

// HTTP ハンドラ全体で 3 秒
ctx, cancel := context.WithTimeout(r.Context(), 3*time.Second)
defer cancel()
 
// その中で DB は 1 秒
dbCtx, dbCancel := context.WithTimeout(ctx, 1*time.Second)
defer dbCancel()

context.WithTimeout は親のデッドラインを延ばせません。親が 3 秒なら、子に 10 秒を指定しても実質 3 秒です。この性質のおかげで「全体の上限」が構造的に保証されます。

Go 1.21 で追加された WithTimeoutCause を使うと、タイムアウトに理由を添えられます。

var ErrDBSlow = errors.New("db call exceeded budget")
 
dbCtx, dbCancel := context.WithTimeoutCause(ctx, time.Second, ErrDBSlow)
defer dbCancel()
 
if err := db.QueryRowContext(dbCtx, q).Scan(&v); err != nil {
    // ctx.Err() は常に context.DeadlineExceeded だが、
    // context.Cause(dbCtx) は ErrDBSlow を返す
    if errors.Is(context.Cause(dbCtx), ErrDBSlow) {
        // 「DB が遅くて切った」と「全体の期限が来た」を区別できる
    }
}

ctx.Err() は context.DeadlineExceeded しか返さないため、どの階層のタイムアウトで切れたのかが分かりません。Cause を使えば区別でき、障害調査でどの階層が原因かをすぐ特定できます。

4.3.1 context.Canceled / DeadlineExceeded には 4 通りの意味がある

実務では、キャンセルを伝えることより、受け取ったキャンセルをどう解釈するかの方が難しくなります。ハンドラーの中で context.Canceled や context.DeadlineExceeded を観測したとき、原因は少なくとも 4 通りあり、正しい反応はそれぞれ違います。

④を①より先に判定します。4.5 の構成では BaseContext に workerCtx を渡しているため、stopWorkers() は進行中の全ハンドラーの r.Context() もキャンセルします。そのため r.Context().Err() != nil は①と④の両方で真になります。①を先に判定すると、停止中のリクエストを「クライアントが切った」と誤分類し、返すべき 503 を返さないことになります。

①を②③と混同するのが、実務で最も多い間違いです。ユーザーがブラウザーを閉じただけの事象が「サーバーエラー」として計上され、デプロイのたびにエラー率グラフが跳ねているように見えます。「graceful shutdown を入れたのに 5xx が減らない」という相談の一定数は、これが原因です。クライアント都合の切断は、そもそも自分の失敗ではありません。

②の見分け方には、実装上の裏付けがあります。errgroup.go の WithContext は context.WithCancelCause でコンテキストを作り、失敗時に g.cancel(g.err) を呼びます。したがって兄弟の goroutine は、context.Cause(ctx) でグループを止めたエラーを取得できます。

g, ctx := errgroup.WithContext(context.Background())
 
g.Go(func() error {
    return errBoom // 誰かが失敗する
})
g.Go(func() error {
    <-ctx.Done()
    // ctx.Err()          → context.Canceled（情報がない）
    // context.Cause(ctx) → errBoom（本当の原因）
    return fmt.Errorf("aborted: %w", context.Cause(ctx))
})

手元の Go 1.27 ＋ x/sync v0.23.0 で実行すると、ctx.Err()=context canceled に対して context.Cause(ctx)=boom が得られます。2.3 の fan-out で「なぜページが出せなかったのか」を下位のタスク側でログに残したいとき、この 1 行が効きます。

自分でキャンセルを送る側なら、context.WithCancelCause で理由を添えられます。

ctx, cancel := context.WithCancelCause(parent)
defer cancel(nil)
 
// ... 何か問題が起きたら
cancel(fmt.Errorf("upstream quota exhausted"))
// 下流では context.Cause(ctx) でこの理由が読める

4.3 の WithTimeoutCause が「期限切れの理由」を伝えるのに対し、WithCancelCause は「明示的なキャンセルの理由」を伝えます。両方を使えば、context.Canceled / DeadlineExceeded の 2 値だけでは分からない原因を下流に伝えられます。

4.4 「リクエストが終わっても続けたい処理」

fire-and-forget な処理を次のように書くのは、よくあるバグです。

// バグ: ハンドラが return した瞬間に ctx がキャンセルされる
func handler(w http.ResponseWriter, r *http.Request) {
    doMainWork(r.Context())
    go auditLog(r.Context(), "action performed") // ほぼ確実に context canceled
    w.WriteHeader(http.StatusOK)
}

r.Context() はハンドラーが return した時点でキャンセルされるため、監査ログの書き込みは高確率で context canceled で失敗します。Go 1.21 の context.WithoutCancel がこの用途のためにあります。

func handler(w http.ResponseWriter, r *http.Request) {
    doMainWork(r.Context())
 
    // 値（trace ID など）は引き継ぎ、キャンセルだけ切り離す
    bg := context.WithoutCancel(r.Context())
    go func() {
        ctx, cancel := context.WithTimeout(bg, 5*time.Second) // 期限は必ず付ける
        defer cancel()
        if err := auditLog(ctx, "action performed"); err != nil {
            slog.ErrorContext(ctx, "audit log failed", "err", err)
        }
    }()
 
    w.WriteHeader(http.StatusOK)
}

WithoutCancel は trace ID などの Value を保持したまま、キャンセルとデッドラインだけを切り離します。ただし必ず新しいタイムアウトを設定してください。設定しないと期限のない goroutine を撒くことになり、第8章で扱うリークの原因になります。

なお、この go func() はプロセス停止時に取りこぼされます。本当に失われては困るデータなら、goroutine ではなく永続キュー（あるいは 4.5 で束ねるワーカー）に載せるべきです。fire-and-forget で書く処理は、失っても構わない処理に限ってください。

4.5 Graceful shutdown — 停止順序と時間予算

Kubernetes 環境では、Pod 停止時に SIGTERM が送られ、terminationGracePeriodSeconds（既定 30 秒）後に SIGKILL されます。この間にやるべきことは決まっており、何よりも順序が重要です。

順序を逆にすると障害になります。先にワーカーを止めると、まだ受け付けている HTTP リクエストが処理できずエラーになります。先に DB を閉じると、処理中のリクエストが全滅します。

func RunServer(handler http.Handler, workers []*Worker) error {
    // SIGTERM / SIGINT を受けてキャンセルされる ctx
    rootCtx, stop := signal.NotifyContext(
        context.Background(), os.Interrupt, syscall.SIGTERM)
    defer stop()
 
    // ワーカー用の ctx は「入口が閉じた後」に閉じたいので、
    // rootCtx とは別系統にする
    workerCtx, stopWorkers := context.WithCancel(context.Background())
    defer stopWorkers()
 
    srv := &http.Server{
        Addr:              ":8080",
        Handler:           handler,
        ReadHeaderTimeout: 5 * time.Second,
        BaseContext:       func(net.Listener) context.Context { return workerCtx },
    }
 
    g, gctx := errgroup.WithContext(rootCtx)
 
    g.Go(func() error {
        if err := srv.ListenAndServe(); !errors.Is(err, http.ErrServerClosed) {
            return err
        }
        return nil
    })
 
    for _, w := range workers {
        g.Go(func() error { return w.Run(workerCtx) })
    }
 
    // シャットダウン担当の goroutine
    g.Go(func() error {
        <-gctx.Done() // シグナル、またはいずれかのタスクの失敗
 
        // ステップ 1-2: 入口を閉じ、処理中の完了を待つ
        // タイムアウトは「停止シーケンス全体の予算」から配分する（判断 2）
        shutCtx, cancel := context.WithTimeout(context.Background(), 7*time.Second)
        defer cancel()
        if err := srv.Shutdown(shutCtx); err != nil {
            slog.Error("graceful shutdown timed out", "err", err)
            _ = srv.Close() // 諦めて強制切断
        }
 
        // ステップ 3: 入口が閉じた後にワーカーを止める
        stopWorkers()
        return nil
    })
 
    if err := g.Wait(); err != nil && !errors.Is(err, context.Canceled) {
        return err
    }
    return nil
}

signal.Notify と channel を自前で扱うより短く、ctx として下流に渡せます。

SIGKILL が来る前に自分で終わるには、grace period を Shutdown 単体ではなく停止シーケンス全体に配分する必要があります。Shutdown に 25 秒使い、そのあとワーカーのドレインに 15 秒使えば合計 40 秒で、30 秒の grace period を超えて SIGKILL されます。ドレインが救うはずだったデータを、まさにその瞬間に失います。

terminationGracePeriodSeconds = 30s
  LB 離脱待ち                    5s   （4.6）
+ srv.Shutdown                  7s   （4.5）
+ ワーカーのドレイン              10s   （10.2 の drain）
+ 予算超過後の後始末              3s   （10.2 の abandon）
+ 余裕                          5s
= 30s

（10.2 のディスパッチャーは各レーンを並行にドレインするので、10 秒と 3 秒は「レーン数ぶんの合計」ではなく「レーン間の最大値」です。）

各段のタイムアウトは、この配分表から決めてください。上のコードでは shutCtx を 7 秒にしています。Shutdown がタイムアウトしたら Close() で強制切断し、少なくとも「終わったこと」を確定させます。

誤解されやすい点ですが、Shutdown は「新規受付を止めて、アイドル接続を閉じ、処理中が終わるのを待つ」だけで、処理中のハンドラーの r.Context() はキャンセルしません（net/http/server.go の Shutdown はリスナーとアイドル接続を閉じたあと、接続が静止するまでポーリングするだけです）。したがって、無限ループするハンドラーがあると Shutdown 自身は待ち続け、上のコードでは shutCtx の 7 秒タイムアウトでしか抜けられません。

そのため BaseContext で workerCtx を渡しています。RunServer のステップ 3 で呼ぶ stopWorkers() は、バックグラウンドワーカーだけでなく BaseContext 経由で進行中のハンドラーにも停止を伝えるので、Shutdown が時間内に返らなかった場合の最終手段になります。

もしワーカーに rootCtx を渡すと、SIGTERM の瞬間にワーカーが止まります。しかしその時点ではまだ HTTP リクエストが処理中で、そのハンドラーがワーカーにジョブを投入するかもしれません。入口が閉じきってからワーカーを止める必要があるため、別系統の ctx にして手動で stopWorkers() を呼びます。

正常なシグナル停止でも gctx はキャンセルされるため、これをエラーとして扱うと終了コードが 1 になり、Kubernetes が異常終了として記録します。

4.6 ロードバランサーからの離脱を忘れない

SIGTERM を受けて即座にリスナーを閉じると、ロードバランサーがまだこの Pod にトラフィックを送っている可能性があります（Endpoints の更新はプロセス停止と非同期です）。結果として一部のリクエストが接続拒否になります。

実務での定番は、SIGTERM 受信後にまず readiness probe を落とし、数秒待ってからドレインを始めることです。4.5 の RunServer のシャットダウン担当 goroutine を、次のように差し替えます。

// hardStop は「2 回目の SIGTERM / Ctrl-C」で閉じる channel。
// signal.NotifyContext は stop() を呼ぶまで 2 回目以降のシグナルを
// 飲み込んでしまうため、即時停止したいときは自前で受ける必要がある。
// この goroutine はプロセスの寿命と運命を共にする（意図的に止めない）
hardStop := make(chan struct{})
go func() {
    sig := make(chan os.Signal, 2)
    signal.Notify(sig, os.Interrupt, syscall.SIGTERM)
    defer signal.Stop(sig)
    <-sig // 1 回目（NotifyContext と共有）
    <-sig // 2 回目 = 「待たずに落とせ」
    close(hardStop)
}()
 
g.Go(func() error {
    <-gctx.Done()
 
    // 1. まず readiness を落とす（LB から外れる）
    ready.Store(false)
 
    // 2. LB が気づくまで待つ（probe 間隔 + 猶予）
    select {
    case <-time.After(5 * time.Second):
    case <-hardStop:
    }
 
    // 3. ここからドレイン（判断 2 の予算配分に従う）
    shutCtx, cancel := context.WithTimeout(context.Background(), 7*time.Second)
    defer cancel()
    if err := srv.Shutdown(shutCtx); err != nil {
        slog.Error("graceful shutdown timed out", "err", err)
        _ = srv.Close()
    }
 
    stopWorkers()
    return nil // 判断 5 と同じく、正常停止を異常終了にしない
})

「graceful shutdown を実装したのにデプロイのたびに 5xx が出る」という相談の大半は、この待ち時間がないことが原因です。

第5章 パターン④ 所有権の直列化 — mutex では破綻する状態管理

5.1 mutex が向かなくなる 3 つの兆候

1.2 では「状態を守るだけなら mutex」としました。実務で mutex から channel に切り替えるべきなのは、次の兆候が出たときです。

// 危険なコード
func (c *Cache) Refresh(ctx context.Context, key string) error {
    c.mu.Lock()
    defer c.mu.Unlock()
 
    v, err := c.origin.Fetch(ctx, key) // ロックを持ったまま数百ms のネットワーク I/O
    if err != nil {
        return err
    }
    c.m[key] = v
    return nil
}

これは動きますが、オリジンが遅くなった瞬間に全リクエストが直列化します。ロックを持ったまま外部を呼ぶのは、実務で最も高くつくアンチパターンのひとつです。

func (c *Cache) Refresh(ctx context.Context, key string) error {
    v, err := c.origin.Fetch(ctx, key) // ロックの外で I/O する
    if err != nil {
        return err
    }
    c.mu.Lock()
    c.m[key] = v // ロックは代入の一瞬だけ
    c.mu.Unlock()
    return nil
}

これで直列化は解消します（同じキーへの重複フェッチが気になるなら 9.1 の singleflight を重ねます）。アクターに切り替えても、所有者ループの中で I/O をすれば同じように詰まります（5.4）。アクターが効くのは、I/O を外に出した後もなお状態遷移の順序を守る必要が残る場合で、これは後述の兆候 3 にあたります。

デッドロックの温床です。ロック順序の規約をドキュメントで守る運用は、コードベースが育つと必ず破綻します。

「接続は connecting → connected → draining → closed としか遷移しない」「クローズ後の書き込みは無視する」といった規則を mutex で表現すると、あらゆるメソッドの先頭に状態チェックが散らばります。

兆候 2 か兆候 3 が出たら（あるいは兆候 1 が、I/O をロックの外に出してもなお残るなら）、状態を 1 つの goroutine が所有し、他は channel 経由でしか触れない構造に切り替えます。いわゆるアクターモデルです。

5.2 コマンド channel と reply channel

実装の骨格は次のとおりです。

type entry struct {
    val       string
    expiresAt time.Time
}
 
// コマンドを型で表現する
type cacheCmd interface{ isCacheCmd() }
 
type getCmd struct {
    key   string
    reply chan getResult // 返信専用 channel
}
 
func (getCmd) isCacheCmd() {}
 
type getResult struct {
    val string
    ok  bool
}
 
type setCmd struct {
    key string
    val string
    ttl time.Duration
}
 
func (setCmd) isCacheCmd() {}
 
type Cache struct {
    cmds chan cacheCmd
    done chan struct{} // 所有者 goroutine の終了を通知する
}
 
func NewCache(ctx context.Context) *Cache {
    c := &Cache{
        cmds: make(chan cacheCmd),
        done: make(chan struct{}),
    }
    go c.loop(ctx)
    return c
}
 
// Done は「所有者が停止しきったこと」を待つための公開手段。
// 11.1 で見るように、テスト可能性のために必要になる
func (c *Cache) Done() <-chan struct{} { return c.done }

中核は 1 本の for-select ループです。このループの中で m に触るのは自分だけなので、ロックが 1 つも要りません。

func (c *Cache) loop(ctx context.Context) {
    defer close(c.done) // 終了を全待機者にブロードキャスト
 
    m := make(map[string]entry)
    gc := time.NewTicker(time.Minute)
    defer gc.Stop()
 
    for {
        select {
        case <-ctx.Done():
            return
 
        case cmd := <-c.cmds:
            switch cmd := cmd.(type) {
            case setCmd:
                m[cmd.key] = entry{cmd.val, time.Now().Add(cmd.ttl)}
            case getCmd:
                e, ok := m[cmd.key]
                if ok && time.Now().After(e.expiresAt) {
                    delete(m, cmd.key)
                    ok = false
                }
                if !ok {
                    e.val = "" // 期限切れの値を返さない
                }
                cmd.reply <- getResult{e.val, ok}
            }
 
        case now := <-gc.C:
            // 期限切れの一括削除。mutex 版ならここで全体をロックすることになる
            for k, e := range m {
                if now.After(e.expiresAt) {
                    delete(m, k)
                }
            }
        }
    }
}

最後の case now := <-gc.C: では、「定期的な掃除」という時間駆動の処理を、状態アクセスと同じ場所に置けています。mutex 版なら別 goroutine から全体をロックして走査する必要があり、その間 API 全体が止まります。時間と入力が合流する処理は、channel でなければ自然に書けません。

5.3 呼び出し側 — 3 つの待ち先を必ず持つ

公開 API 側では、すべての channel 操作を select で囲み、必ず 3 つの待ち先を用意します。

var ErrCacheClosed = errors.New("cache: closed")
 
func (c *Cache) Get(ctx context.Context, key string) (string, bool, error) {
    reply := make(chan getResult, 1) // バッファ 1 が重要（後述）
 
    // ① コマンドを送る
    select {
    case c.cmds <- getCmd{key: key, reply: reply}:
    case <-c.done:
        return "", false, ErrCacheClosed
    case <-ctx.Done():
        return "", false, ctx.Err()
    }
 
    // ② 返信を待つ。すでに届いている返信は、同時に起きた停止より優先する
    select {
    case r := <-reply:
        return r.val, r.ok, nil
    default:
    }
 
    select {
    case r := <-reply:
        return r.val, r.ok, nil
    case <-c.done:
        return "", false, ErrCacheClosed
    case <-ctx.Done():
        return "", false, ctx.Err()
    }
}

これがアクターパターン最大の落とし穴です。所有者 goroutine が終了した後に c.cmds <- cmd を実行すると、受信者がいないので永久にブロックします。ctx にタイムアウトがなければ、その goroutine は二度と戻りません。所有者の生死を伝える channel（ここでは done）は必須です。

これを省くと、正しく計算された結果を捨てて ErrCacheClosed を返すことがあります。所有者が cmd.reply <- ...（バッファー 1 なので即完了）を実行した直後にループが <-ctx.Done() に入って終了し、defer close(c.done) が走る、という並びは普通に起こります。この瞬間、呼び出し側の select からは reply と c.done の両方が準備できて見え、8.3 のとおり一様ランダムに選ばれるので、結果は 50% の確率で捨てられます。

手元で 400 回試行したところ、ちょうど 200 回が偽の ErrCacheClosed になりました。前段のノンブロッキング select を入れると 0 回でした。これは 8.3 の優先度問題をアクターの返信経路に適用したもので、すでに配送済みの返信より done の case が優先されてはいけません。

make(chan getResult, 1) の 1 には明確な意味があります。仮にバッファーなしにすると、呼び出し側が ctx.Done() で②を離脱した直後に所有者が cmd.reply <- ... を実行したとき、受信者がいないので所有者ループが永久にブロックします。1 個のアクターが止まれば全 API が止まる、致命的な障害です。バッファー 1 なら所有者は必ず送信を完了でき、値は誰にも読まれないまま GC されます。

3.4 で述べたとおり、バッファーサイズは設計判断です。この 1 は性能チューニングではなく、所有者を絶対にブロックさせないという不変条件を表しています。

5.4 アクターにすべきか、mutex のままか

読み込みが支配的なキャッシュなら、読み込みを並行に処理できる sync.RWMutex の方が向いています。アクターが向いているのは、状態遷移そのものが仕様の中心にある場合です。接続の状態機械、レート制限のトークンバケット、順序保証が要るジョブキュー、といった対象です。

アクター内の処理が遅いと、それが全体のボトルネックになります。所有者ループの中では状態の更新だけを行い、I/O は別 goroutine に投げるのが基本です。

第6章 パターン⑤ バッチング — 時間と入力を合流させる

6.1 実務での出現場所

「N 件たまるか、T 秒経つか、どちらか早い方でまとめて送る」処理は、実務でよく登場します。

• メトリクス / 監査ログの送信（1 件ずつ HTTP を叩くと下流が死ぬ）
• DB への一括 INSERT（1 件ずつだとラウンドトリップが支配的）
• 外部 API の bulk エンドポイント（1 リクエスト 500 件まで、など）
• 検索インデックスの更新

これは channel でしか自然に書けない処理の代表例です。「入力の到着」と「時間の経過」という 2 つの独立したイベント源を待ち合わせる必要があり、select 以外に自然な表現がありません。

6.2 実装

type Batcher struct {
    in       chan Event
    done     chan struct{}
    sink     Sink
    maxSize  int
    maxDelay time.Duration
}
 
func NewBatcher(sink Sink, queue, maxSize int, maxDelay time.Duration) *Batcher {
    return &Batcher{
        in:       make(chan Event, queue), // queue = 許容する滞留量（3.4）
        done:     make(chan struct{}),
        sink:     sink,
        maxSize:  maxSize,
        maxDelay: maxDelay,
    }
}
 
func (b *Batcher) Run(ctx context.Context) error {
    defer close(b.done)
 
    batch := make([]Event, 0, b.maxSize)
    timer := time.NewTimer(b.maxDelay)
    defer timer.Stop()
 
    flush := func(reason string) {
        if len(batch) == 0 {
            return
        }
        // シャットダウン中でも書き切れるよう、キャンセルを切り離す
        fctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
        defer cancel()
        if err := b.sink.WriteBatch(fctx, batch); err != nil {
            slog.Error("batch write failed", "reason", reason, "n", len(batch), "err", err)
        }
        batch = batch[:0]
    }
 
    for {
        select {
        case <-ctx.Done():
            // 停止時：キューに残っている分を吸い出してから flush
            for {
                select {
                case e := <-b.in:
                    batch = append(batch, e)
                    if len(batch) >= b.maxSize {
                        flush("drain")
                    }
                    continue
                default:
                }
                break
            }
            flush("shutdown")
            return nil
 
        case e := <-b.in:
            batch = append(batch, e)
            if len(batch) == 1 {
                // 「最初の 1 件が入った時刻」から最大遅延を測る。
                // Go 1.23+ では Stop() 後にチャネルへ古い値が残らないので
                // ドレイン（<-timer.C）を書いてはいけない（6.4 参照）
                timer.Stop()
                timer.Reset(b.maxDelay)
            }
            if len(batch) >= b.maxSize {
                timer.Stop()
                flush("size")
            }
 
        case <-timer.C:
            flush("time")
        }
    }
}

6.3 6.2 の実装に含まれる判断

len(batch) == 1 の条件がそれです。素朴に「常に 100ms ごとに flush」とすると、99ms 目に届いたイベントは 1ms 後に単独で送信されてしまいます。最初の 1 件から測っても、どのイベントも到着から最大 T 秒以内に送られ、各バッチには T 秒いっぱいの集約時間を確保できます。

case <-ctx.Done(): の中で残りを吸い出しています。これがないと、デプロイのたびに、手持ちのバッチ（最大 maxSize - 1 件）に加えて b.in に残っている分（最大 cap(b.in) 件）が消えます。「たまにログが欠ける」という不具合の典型的な原因です。

このドレインループは「入口がすでに閉じている」ことを前提にしています。default: で抜けるので、b.in が一瞬でも空になれば終わります。しかし生産者が Add を呼び続けている間はキューが供給され続け、maxSize を跨ぐたびに flush が最大 5 秒かかるため、停止時間に上限がなくなります。4.5 の停止順序（入口 → 処理中 → ワーカー）を守っていればこの状況は起きませんが、そうでない配置なら 10.2 の drain のように明示的な予算を設けてください。

context.WithoutCancel(ctx) がそれです。停止時の flush で ctx をそのまま使うと、すでにキャンセル済みなので送信が即座に失敗します。ただし WithTimeout は必ず付けます（永久に終わらない停止処理は SIGKILL されるだけです）。

var ErrBatcherClosed = errors.New("batcher: closed")
 
func (b *Batcher) Add(ctx context.Context, e Event) error {
    select {
    case b.in <- e:
        return nil
    case <-b.done: // 5.3 のポイント 1：所有者の生死を必ず待ち先に入れる
        return ErrBatcherClosed
    case <-ctx.Done():
        return ctx.Err()
    }
}

b.in のバッファーサイズが、そのまま「許容する滞留量」になります（3.4 参照）。メトリクスのように落としてよいデータなら、ここを default: にして捨てる設計もありえます。

<-b.done の case を忘れると、Run が終わった後の Add が誰も読まないバッファーに黙って値を捨てる（cap(b.in) 件まで）か、呼び出し側の ctx が切れるまでブロックします。5.3 でアクターに課したのと同じ規約が、ここにも同じ理由で必要です。

time.NewTimer(b.maxDelay) は作った瞬間に動き始めるため、起動後 maxDelay の間 1 件もイベントが来ないと、空バッチに対する case <-timer.C: が 1 度だけ発火します。flush は len(batch) == 0 で早期 return するので実害はありません。気になるなら生成直後に timer.Stop() を呼んでください。

6.4 time.Timer の作法 — Go 1.23 で変わったこと

6.2 の実装では timer.Stop() を呼ぶだけで、発火済みの値を <-timer.C で取り除く処理を書いていません。既存コードの多くには if !t.Stop() { <-t.C } という古典的なイディオムが入っていますが、Go 1.23 で time.Timer の実装が変わり、この作法の前提が崩れました。

Go 1.23 リリースノート の変更点は 2 つです。

1. 参照されなくなった Timer / Ticker は、Stop を呼ばなくても即座に GC 対象になる。以前は、未 Stop の Timer は発火するまで回収されず、Ticker に至っては永久に回収されませんでした
2. タイマーの channel がバッファーなし（容量 0）になった。Reset / Stop の呼び出し後に、それ以前に準備された古い値が送受信されないことが保証されるようになりました

実務への影響は次の 3 つです。

• time.After をループ内で使ってもメモリーリークにならなくなった（Go 1.23 以降）。それ以前は、for ループ内の select で time.After(time.Hour) を使うと、1 時間分のタイマーがすべて生き続けました
• ただし毎回タイマーを確保するコストは残るので、ホットループでは time.Timer の再利用か context.WithTimeout が依然として有利です
• Stop の後に <-timer.C でドレインする処理は、Go 1.23 以降は書いてはいけません

最後の点は、古い記事のコードをそのまま使っている場合ほど危険です。Go 1.23 以降の Stop() は、発火済みの値が channel に残っていればそれを取り除いたうえで true を返します。そのため Stop() が false を返すのは、channel が空であることが保証される場合に限られます。したがって次の古典的なイディオムは、Stop() が false を返したときに確実にデッドロックします（time/sleep.go が「Stop が返った後の t.C からの受信はブロックすることが保証される」と明記しています）。

if !t.Stop() {
    <-t.C // Go 1.23 以降、ここに到達したら必ずデッドロックする
}

timerActive のようなガードを足しても正しくなりません。ガードが false になる条件と、ドレインが安全な条件が一致しないからです。正しい書き方は単純で、timer.Stop() だけを呼び、ドレインを書かないことです。

手元の Go 1.27 で確認した挙動です。

tm := time.NewTimer(1 * time.Millisecond)
time.Sleep(20 * time.Millisecond) // 確実に発火させる
 
tm.Stop() // → true（発火済みの値を取り除いたため）
// この後 <-tm.C は永久にブロックする

なお、asynctimerchan GODEBUG（旧挙動に戻すための設定）は Go 1.27 で完全に削除されました。現行の Go では、タイマー channel は常にバッファーなしです。

第7章 パターン⑥ 購読とブロードキャスト — 1 対 N の通知

7.1 channel は 1 対 1 である

見落とされがちですが、1 つの channel に送った 1 つの値は、同じ channel から受信する goroutine が何本あっても、1 つの受信者にしか届きません。（ランタイムは待機者を recvq という FIFO キューで管理しており、実装上は最も長く待っている goroutine が起こされます。言語仕様は順序を規定していないので、依存すべきではありません。）作業の分配としては正しい挙動です。

したがって「全員に同じ通知を届ける」には別の手段が要ります。選択肢は 2 つです。

7.2 close によるブロードキャスト

「一度だけ起きる出来事」を全員に知らせるなら、これが最も安く確実です。

type Signal struct {
    once sync.Once
    ch   chan struct{}
}
 
func NewSignal() *Signal { return &Signal{ch: make(chan struct{})} }
 
func (s *Signal) Done() <-chan struct{} { return s.ch }
 
// 何回呼んでも安全（二重 close による panic を防ぐ）
func (s *Signal) Fire() { s.once.Do(func() { close(s.ch) }) }

閉じた channel を再度 close すると panic するので、sync.Once で包みます。複数の経路から停止が起こりうる実務コードでは、Once なしの close はいずれ必ず落ちます。

context も内部で同じことをしているので、停止を伝えるために自前で書く場面は多くありません。自前で書くのは、初期化完了やリーダー選出成功など、「停止」以外の一度きりのイベントを知らせるときです。

7.3 購読ハブ — SSE / WebSocket の実装

繰り返し発生するイベントを N 人に配る場合、購読者ごとに channel を持ちます。

type Hub struct {
    mu   sync.RWMutex
    subs map[*subscriber]struct{}
 
    // RLock は共有ロックなので、複数の Publish が同時に走る。
    // ここを素の int64 にすると -race が競合を報告する
    dropped atomic.Int64
}
 
type subscriber struct {
    ch    chan Message
    unsub func()
}
 
func NewHub() *Hub {
    return &Hub{subs: make(map[*subscriber]struct{})}
}
 
func (h *Hub) Subscribe(buffer int) (<-chan Message, func()) {
    if buffer < 1 {
        // バッファ 0 だと、受信者がすでに待機中の瞬間しか送信が成功しない。
        // Publish はノンブロッキング送信なので、ほぼ全部落ちてしまう
        buffer = 1
    }
    s := &subscriber{ch: make(chan Message, buffer)}
 
    var once sync.Once
    s.unsub = func() {
        once.Do(func() {
            h.mu.Lock()
            delete(h.subs, s)
            h.mu.Unlock()
            close(s.ch) // 送信側だけが閉じる
        })
    }
 
    h.mu.Lock()
    h.subs[s] = struct{}{}
    h.mu.Unlock()
 
    return s.ch, s.unsub
}
 
// Publish は決してブロックしない。遅い購読者には落とす。
func (h *Hub) Publish(m Message) {
    h.mu.RLock()
    defer h.mu.RUnlock()
    for s := range h.subs {
        select {
        case s.ch <- m:
        default:
            h.dropped.Add(1) // 観測できるようにする
        }
    }
}
 
// 落とした件数は必ず外から読めるようにする（7.4）
func (h *Hub) Dropped() int64 { return h.dropped.Load() }

このハブは channel と mutex を併用しています。1.2 の基準どおり、状態である購読者の集合（マップ）は mutex で守り、所有権の受け渡しにあたる各購読者へのメッセージ配送は channel で行います。片方に寄せる必要はありません。

dropped を atomic.Int64 にしているのは、RLock が共有ロックで、複数の Publish が同時にこのループを実行するからです。素の h.dropped++ は read-modify-write であり、共有ロック下では保護されません。「ロックを取っているから安全」という直感が外れる典型例で、go test -race が即座に報告します。ロックの種類（排他か共有か）と、その下で行う操作が対応しているかを常に確認してください。

Subscribe の返り値は <-chan Message（受信専用）です。購読者は受信しかできず、close もできません。「channel を閉じるのは常に送信側」という規約を、型で強制しています。

7.4 遅い購読者（slow consumer）問題

Publish の default: 節は、このハブで最も重い設計判断です。購読者の 1 人が遅い（ネットワークが詰まっている、処理が重い）ときの扱いには、次の 4 つの戦略があります。

「切断する」には落とし穴があります。Publish の default: 節の中で直接 unsub() を呼ぶと、確実に自己デッドロックします。Publish は h.mu.RLock() を保持しており、unsub は同じ RWMutex の Lock() を取りにいくためです。Go の sync.RWMutex は再入可能でもアップグレード可能でもありません。切断対象を集めておき、RUnlock の後で呼びます。

func (h *Hub) PublishDisconnectSlow(m Message) {
    var doomed []func()
 
    h.mu.RLock()
    for s := range h.subs {
        select {
        case s.ch <- m:
        default:
            doomed = append(doomed, s.unsub) // ここでは呼ばない。集めるだけ
        }
    }
    h.mu.RUnlock()
 
    for _, unsub := range doomed {
        unsub() // 排他ロックが要るので必ず RUnlock の後で
    }
}

実務上の第一選択は「落とす」か「切断する」です。ブロックを選ぶと、Publish の呼び出し元（多くはビジネスロジックの中心）が、最も遅い購読者のペースに引きずられます。1 人の遅いクライアントが全システムを止める構造は、外部に公開する API では特に危険です。

Publish を素の s.ch <- m に書き換えると、遅くなるだけでなくデッドロックも起こります。ブロック版の Publish は h.mu.RLock() を保持したまま park します。その購読者の読み手が「もう要らない」と unsub() を呼ぶと、unsub は h.mu.Lock() を待ってブロックし、同時に s.ch を読むのをやめます。誰も s.ch を読まないので Publish は永久に返らず、RUnlock されないので unsub も永久に返りません。これが相互デッドロックです。さらに Go の RWMutex は writer が待機し始めると新規 reader を通さないので、後続の Publish と Subscribe もすべて巻き込まれます。

ブロック戦略では、「遅い購読者を切り離す」という自己修復の経路そのものが動かなくなります。

「ロックの外で送る」で回避しようとすると、今度は panic します。RLock を保持したまま送っていることが、unsub の close(s.ch) との相互排他を担保していたからです。購読者集合のスナップショットを取ってロックの外で送ると、その排他が消えます。古い *subscriber を掴んだ送信者が、unsub が閉じた直後の s.ch に送って send on closed channel で落ちます。しかも並行する Publish があれば s.ch は多重送信者の channel なので、8.5 の規約 2 が「閉じるな」と言っている状況そのものです。

スナップショット方式を採るなら、close(s.ch) をやめる必要があります。購読終了は購読者ごとの done channel と select で表現し、channel 自体は閉じずに GC に任せます。そこまでやる価値がないなら、1 件も落とせないデータは Hub に載せず、永続キューに載せてください。

「古い方を捨てる」の実装は次のとおりです。

select {
case s.ch <- m:
default:
    select {
    case <-s.ch: // 最古の 1 件を捨てる
    default:
    }
    select {
    case s.ch <- m: // 空いた枠に最新を入れる
    default:
    }
}

ただし複数の Publish が並行すると厳密な「最新 N 件」にはなりません。厳密さが要るなら、購読者ごとにリングバッファーを持ち mutex で守る方が素直です。

いずれの戦略でも、落とした件数を必ずメトリクスに出してください。「気づかないうちにデータが消えている」のが、この構造で最も怖い失敗です。

第8章 事故カタログ — 本番で実際に踏むバグ

8.1 goroutine リークの 4 大原因

goroutine リークは、Go におけるメモリーリークの主要な形態です。goroutine が終わらない限り、そのスタックとスタックから参照される全オブジェクトが回収されません。

最も多いパターンです。

// リークする実装
func LeakySearch(ctx context.Context, q string) (string, error) {
    backends := []string{"a", "b", "c"}
    results := make(chan string) // バッファなし
 
    for _, backend := range backends {
        go func() {
            results <- search(backend, q) // 最初の 1 本以外は永久にブロック
        }()
    }
 
    select {
    case r := <-results:
        return r, nil // ← ここで返ると、残り 2 本が永久にリーク
    case <-ctx.Done():
        return "", ctx.Err() // ← ここなら 3 本全部リーク
    }
}

「最速の 1 件だけ使う」は正しい設計ですが、残りの送信側を解放していません。修正は 1 箇所です。

results := make(chan string, len(backends)) // 送信側が絶対にブロックしない

これも、3.4 で述べたとおりバッファーサイズが設計判断である例です。この len(backends) は「全送信者が受信者不在でも送信を完了できる」という不変条件を表しており、性能とは無関係です。

for v := range ch { // ch が誰にも close されないと永久にブロック
    process(v)
}

close する責任者が決まっていない、あるいは close する経路がエラーで飛ばされると起きます。送信側は defer close(ch) で閉じるのが原則です。

// 停止できない goroutine
go func() {
    for {
        time.Sleep(time.Minute)
        refreshCache()
    }
}()

プロセスが生きている限り止まりません。テストで起動すると、テストプロセスに残り続けます。

// 正しい実装
go func() {
    t := time.NewTicker(time.Minute)
    defer t.Stop()
    for {
        select {
        case <-ctx.Done():
            return
        case <-t.C:
            refreshCache(ctx)
        }
    }
}()

for v := range ch {
    if v.IsBad() {
        return errors.New("bad value") // 送信側は残りを送ろうとしてブロック
    }
}

早期 return する可能性があるなら、送信側にも ctx による脱出口が必要です。

8.2 Go 1.27 の goroutine リークプロファイル

ここまで挙げた 4 パターンは、これまでランタイムでは検出できず、見つけるには主にコードレビューに頼るしかありませんでした。Go 1.27 では、ランタイムがリークを検出できるようになります。

A new profile type that reports leaked goroutines, previously available as an experiment in Go 1.26, is now generally available. The new profile type, named goroutineleak, is supported in the runtime/pprof package. It is also available as the net/http/pprof endpoint /debug/pprof/goroutineleak.

検出には GC の到達可能性解析を流用しています。goroutine G が同期プリミティブ P でブロックしているとき、P が「実行可能な goroutine」からも「それらが起こしうる goroutine」からも到達不能なら、P は二度とアンブロックされません。このとき G は永久にリークしている、と判定します。

次のプログラムは goroutine を 3 本リークさせます。

func leak() {
    ch := make(chan int) // ローカル変数。leak() が返ると誰も参照しない
    go func() {
        ch <- 1 // 永久にブロック
    }()
}
 
func main() {
    for range 3 {
        leak()
    }
    time.Sleep(100 * time.Millisecond)
 
    p := pprof.Lookup("goroutineleak")
    fmt.Printf("goroutines alive: %d\n", runtime.NumGoroutine())
    fmt.Printf("Count() before write: %d\n", p.Count())
    _ = p.WriteTo(os.Stdout, 1)
    fmt.Printf("Count() after write: %d\n", p.Count())
}

Go 1.27 での実行結果です。

goroutines alive: 4
Count() before write: 0
---- profile (debug=1) ----
goroutineleak profile: total 3
3 @ 0x7ff77731c56a 0x7ff7772b0f5c 0x7ff7772b0b57 0x7ff77738829e 0x7ff777322661
#	0x7ff77738829d	main.leak.func1+0x1d	.../leakdemo/main.go:15
 
Count() after write: 3

リークした 3 本と、その発生箇所の行番号が正確に出ます。従来の goroutine プロファイルは「今ブロックしている goroutine」を全部並べるだけで、正常な待機とリークの区別がつきませんでした。goroutineleak では、その区別が自動でつきます。

実務での使い方は次の 2 つです。

// 1. HTTP エンドポイントとして公開（net/http/pprof を import するだけ）
//    go tool pprof http://localhost:6060/debug/pprof/goroutineleak
 
// 2. テストの最後に検査する
func TestNoLeaks(t *testing.T) {
    // ... テスト本体 ...
 
    p := pprof.Lookup("goroutineleak")
    var buf bytes.Buffer
    _ = p.WriteTo(&buf, 1) // 検出はここで走る。これより前の Count() は 0
    if n := p.Count(); n > 0 {
        t.Errorf("leaked %d goroutines:\n%s", n, buf.String())
    }
}

使ってみて分かった注意点が 2 つあります。

• Count() は最初の WriteTo を呼ぶまで 0 を返します（上の出力のとおり）。リーク検出は書き出し時に走るためで、Count() だけで判定するコードは常に 0 になります。テストで使うなら WriteTo を先に呼んでください
• リリースノートが明記するとおり、グローバル変数や実行中 goroutine のローカル変数から到達できるプリミティブでブロックしているリークは検出できません。到達可能性ベースの手法の原理的な限界です

そのため goroutineleak だけに頼らず、11.1 の goleak や goroutine 数の監視と併用するのが現実的です。

8.3 select のランダム性と優先度

複数の case が同時に準備できている場合、select は一様ランダムに 1 つを選びます（Go 言語仕様）。これは飢餓を防ぐための設計ですが、「優先度をつけたい」ときには障害になります。

// 意図どおりに動かないコード
select {
case <-ctx.Done():   // 「キャンセルを最優先したい」つもり
    return ctx.Err()
case job := <-jobs:
    process(job)
}

ctx がキャンセル済みで、かつ jobs にもデータがあるとき、50% の確率で job が処理されます。「キャンセル後は絶対に新しい仕事を始めない」という要件があるなら、二段構えにします。

// キャンセルを厳密に優先する
select {
case <-ctx.Done():
    return ctx.Err()
default:
}
 
select {
case <-ctx.Done():
    return ctx.Err()
case job := <-jobs:
    process(job)
}

シャットダウン時に残りを処理しきりたい場合は、優先度が逆になります（6.2 のドレインループがこの例です）。どちらが要件なのかを明示的に決めてください。

この二段構えで保証されるのは「キャンセルを観測した後は新しい仕事を始めない」ことまでです。2 つの select のあいだにキャンセルが届いた場合、後段は依然として五分五分です。

この窓は狭いものの 0 ではありません。1 件の取りこぼしも許されない処理では、仕事を始めた側でも先頭で ctx.Err() を確認し、キャンセル済みなら「処理する」のではなく「ドレイン用の予算に載せ替える」か「未処理として差し戻す」必要があります。10.2 の deliver でも同じ問題を取り上げます。

8.4 nil channel — 実務で効くイディオム

nil channel への送受信は永久にブロックします。一見すると単なるバグの元ですが、select の case を動的に有効・無効にする手段として役に立ちます。

func Pump(ctx context.Context, in <-chan Result, out chan<- Result) error {
    var pending *Result
    for {
        // 送るものが無い間は、送信腕を nil にして無効化する
        var send chan<- Result
        var val Result
        if pending != nil {
            send = out
            val = *pending
        }
 
        // 手持ちがある間は、受信腕を nil にして無効化する
        var recv <-chan Result
        if pending == nil {
            recv = in
        }
 
        select {
        case <-ctx.Done():
            return ctx.Err()
        case v, ok := <-recv:
            if !ok {
                return nil
            }
            pending = &v
        case send <- val:
            pending = nil
        }
    }
}

これを nil channel なしで書こうとすると、「手持ちがある場合」と「ない場合」で select 文を 2 つ書き分けることになり、ctx.Done() の処理が重複します。

Pump はパイプラインの 1 段にあたり、実務では 生成 → 変換 → 出力 のように段をつなぎます。上のコードが out を close していないのは、Pump が out を所有していないためです。パイプラインでは、各段が自分の出力 channel を所有し、defer close(out) で閉じるのが規約です（8.5 の規約 1）。

// 段 1: 生成。出力 channel を所有するので自分で閉じる
func gen(ctx context.Context, items []Item) <-chan Item {
    out := make(chan Item)
    go func() {
        defer close(out) // 所有者が閉じる
        for _, it := range items {
            select {
            case out <- it:
            case <-ctx.Done():
                return // 下流が消えても抜けられる
            }
        }
    }()
    return out
}
 
// 段 2: 変換
func transform(ctx context.Context, in <-chan Item) <-chan Result {
    out := make(chan Result)
    go func() {
        defer close(out)
        for it := range in { // 上流が閉じればループを抜ける
            v := convert(it) // select の外で評価する（後述）
            select {
            case out <- v:
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}
 
// 段 3: 出力。ここが受信しきることで、上流全体が終わる
func sink(ctx context.Context, in <-chan Result) error {
    for r := range in {
        if err := write(ctx, r); err != nil {
            return err
        }
    }
    // 上流が「中断された」のか「正常に閉じた」のかを区別するために返す
    return ctx.Err()
}
 
// 組み立て。ここで ctx を所有するのが要点
func RunPipeline(parent context.Context, items []Item) error {
    ctx, cancel := context.WithCancel(parent)
    defer cancel() // sink が早期 return したら、段 1・段 2 を解放する
 
    return sink(ctx, transform(ctx, gen(ctx, items)))
}

convert(it) を select の外で評価しているのは、Go 言語仕様が「送信文のチャネルと右辺の式は、select に入った時点でちょうど一度だけ評価される」と定めているからです。case out <- convert(it): と書くと、ctx.Done() の case が選ばれる場合でも convert は実行されます。純粋関数なら無害ですが、重い処理や副作用があるなら事前に評価してください。

パイプラインでリークするのは、受信側ではなく送信側です。受信を for v := range in と書くと安心しがちですが、下流の段が死ぬと out <- v は誰にも受け取られず永久にブロックします。そのため、すべての送信を select で囲み、ctx.Done() の case を付ける必要があります。8.1 の原因 1 が、パイプラインの形で現れたものです。

sink が早期 return する場合（上の write が失敗したケース）も、上流の段は in を読む goroutine がいなくなって詰まります。そのため ctx を所有するのは、組み立てを行う RunPipeline 側です。defer cancel() が走って初めて、段 1 と段 2 が <-ctx.Done() で抜けられます。sink 自身は渡された ctx をキャンセルできないので、この責任は必ず呼び出し側にあります。

これは実際に検証できます。上の RunPipeline から defer cancel() を外して sink を 3 件目で失敗させると、goleak が gen の goroutine を「まだ生きている」と報告します。付ければ報告されません。

パイプライン全体のスループットは最も遅い段の μ で決まります（3.4）。段を増やしても、律速段より速くはなりません。

同じ発想は「close された channel を select から外す」場合にも使えます。

for a != nil || b != nil {
    select {
    case v, ok := <-a:
        if !ok {
            a = nil // 閉じた channel は常に受信可能になるので、外す
            continue
        }
        handle(v)
    case v, ok := <-b:
        if !ok {
            b = nil
            continue
        }
        handle(v)
    }
}

閉じた channel は「常に受信可能」になります。nil に差し替えないと、その case が延々と選ばれ続けてビジーループになります。fan-in を自前で書くときの必須テクニックです。

8.5 close の規約

規約は次の 3 つです。

1. 送信側が閉じる。受信側は決して閉じない
2. 送信者が複数いる channel は閉じない。どうしても閉じる必要があるなら、done channel（または ctx）で全送信者に停止を伝え、wg.Wait() で全員が抜けたことを確認してから閉じます。sync.Once で包んでも解決しません。Once が防ぐのは二重 close の panic だけで、「閉じた channel への送信」は防げません。複数送信者で実際に問題になるのは後者です（ch <- v でパークしている送信者がいる最中に close すれば、Once の有無に関係なく panic します）
3. 閉じた channel への送信は panic、二重 close も panic。閉じた channel からの受信は、バッファーに残っている値を返し切ってから、ゼロ値と ok == false を返す

3 番目の但し書きは、7.3 で unsub が close(s.ch) する場面に当てはまります。このときバッファーにメッセージが残っていても失われず、受信側が読み切ってから ok == false になります。

（6.3 の Add との違いに注意してください。Add が cap(b.in) 件まで受け付けられるのは、単にバッファーに空きがあるからです。b.in は誰も閉じないので、この規約 3 は適用されません。そのため、誰も読まないバッファーに値を捨てないよう <-b.done の case が要ります。）

7.3 で Subscribe が <-chan Message を返していたのは、規約 1 を型で強制するためです。API の境界では方向付き channel 型（chan<- / <-chan）を使ってください。ドキュメントで規約を書くより確実です。

8.6 データ競合とメモリーモデル

データ競合への対策として、CI で必ず -race を回すことに勝るものはありません。競合検出器は「実際に発生した」競合しか報告しませんが、偽陽性はありません。報告されたら必ずバグです。

go test -race ./...

Data Race Detector によると、典型的なプログラムでは実行時間が 2〜20 倍、メモリー使用量が 5〜10 倍に増えることがあるので、本番バイナリーには入れません。CI と、可能ならステージング環境のカナリアで動かします。

Go メモリーモデルが channel について保証しているのは次の 4 点です（Go 1.19 の改訂で、メモリーモデルは synchronized before を基礎に整理されました。happens before も引き続き、synchronized before とプログラム順序から定義されています）。

1. channel への送信は、対応する受信の完了より前に synchronized される
2. channel の close は、close によりゼロ値を返す受信より前に synchronized される
3. バッファーなし channel からの受信は、対応する送信の完了より前に synchronized される
4. 容量 C の channel からの k 番目の受信は、k+C 番目の送信の完了より前に synchronized される

3 番目が「バッファーなし channel はランデブーである」ことの形式的な表現です。値の可視性を保証するのは 1 番目のほうで、送信側が送信より前に行った書き込みは、値を組み立てる書き込みも含めてすべて受信側から見えます。

4 番目は一見難解ですが、メモリーモデル自身が用途を明記しています。

This rule generalizes the previous rule to buffered channels. It allows a counting semaphore to be modeled by a buffered channel: the number of items in the channel corresponds to the number of active uses, the capacity of the channel corresponds to the maximum number of simultaneous uses, sending an item acquires the semaphore, and receiving an item releases the semaphore.

3.2 の「buffered channel をセマフォとして使う」パターンは、メモリーモデルが公式に保証している用法です。

Go 1.21 以前は、for ループの変数が全イテレーションで共有されていました。

// Go 1.21 以前: 全 goroutine が同じ i を見る（多くの場合すべて最後の値になる）
for i := 0; i < 3; i++ {
    go func() { fmt.Println(i) }()
}
 
// Go 1.21 以前で必要だった回避策
for i := 0; i < 3; i++ {
    i := i // 再宣言
    go func() { fmt.Println(i) }()
}

Go 1.22 でループ変数はイテレーションごとのスコープになり、この問題は解消しました。ただし、この挙動は go.mod の go ディレクティブで決まります。古い go 1.21 のままのモジュールでは旧挙動のままです。ネット上の古い記事からコピペする際は、i := i の有無ではなく、自分の go.mod のバージョンを確認してください。

8.7 goroutine 内の panic はプロセスを落とす

recover は goroutine ごとです。呼び出し元の defer recover() では、別 goroutine の panic を拾えません。

func handler(w http.ResponseWriter, r *http.Request) {
    defer func() {
        if v := recover(); v != nil { // この recover は届かない
            w.WriteHeader(500)
        }
    }()
    go doWork() // ここで panic するとプロセス全体が死ぬ
}

net/http はハンドラー自体の panic を捕捉して接続だけを閉じますが、ハンドラーが起動した goroutine の panic は捕捉できません。3.2 で見たとおり errgroup も意図的に panic を伝播しません。

対策は、ワーカー起動をラップすることです。

func Go(ctx context.Context, name string, f func()) {
    go func() {
        defer func() {
            if v := recover(); v != nil {
                slog.ErrorContext(ctx, "goroutine panicked",
                    "name", name, "panic", v,
                    "stack", string(debug.Stack()))
                metrics.PanicTotal.WithLabelValues(name).Inc()
            }
        }()
        f()
    }()
}

ただし、状態が壊れている可能性がある場合は、recover して生き延びるより落ちて再起動する方が安全です。判断基準は「その goroutine の失敗が、他のリクエストの正しさに影響するか」です。1 リクエストを処理するだけのワーカーなら recover して続行、共有状態を更新する途中なら落とす、が目安になります。

第9章 道具箱 — 何を使うべきか

自分で channel を組む前に、次の既製品で済まないかを確認してください。

9.1 singleflight — 実務価値が最も高い準標準ライブラリー

キャッシュが切れた瞬間に同じキーへのリクエストが 1000 件同時に来ると、そのすべてがオリジン（DB や外部 API）を叩きにいきます。キャッシュスタンピード（thundering herd）と呼ばれる、実務で頻発する障害です。

golang.org/x/sync/singleflight は「同じキーで同時に走る呼び出しを 1 本にまとめ、結果を全員に配る」だけのライブラリーです。1000 件のリクエストに対してオリジンへの呼び出しは 1 回になります。

func (l *Loader) Get(ctx context.Context, key string) (string, error) {
    if v, ok, err := l.cache.Get(ctx, key); err == nil && ok {
        return v, nil
    }
 
    ch := l.sf.DoChan(key, func() (any, error) {
        // 呼び出し元の ctx をそのまま使わない（後述）
        octx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 3*time.Second)
        defer cancel()
 
        v, err := l.origin(octx, key)
        if err != nil {
            return nil, err
        }
        _ = l.cache.Set(octx, key, v, time.Minute)
        return v, nil
    })
 
    select {
    case r := <-ch:
        if r.Err != nil {
            return "", fmt.Errorf("load %q: %w", key, r.Err)
        }
        return r.Val.(string), nil
    case <-ctx.Done():
        return "", ctx.Err() // 自分だけ離脱。他の待機者には影響しない
    }
}

このコードには判断が 2 つ含まれています。

Do はブロックするため、呼び出し元の ctx がキャンセルされても待ち続けます。DoChan は channel を返すので、select で「結果」と「自分のキャンセル」を同時に待てます。相乗りしている他の呼び出し元に影響を与えずに、自分だけ離脱できます。

離脱しても安全であることは、実装から確認できます。singleflight.go の DoChan は待機者ごとに make(chan Result, 1)（バッファー 1）を作ります。そのため、呼び出し元が ctx.Done() で離脱しても、結果を送る側はブロックせずに送信を完了できます。8.1 の「受信者が去った後の送信」でリークしないのは、このバッファー 1 のおかげです（ドキュメントにも「返された channel は close されない」と明記されています）。

これは実装を読まないと気づきにくい点です。Do は g.doCall(...) を呼び出し元の goroutine 上で同期実行し、DoChan は go g.doCall(...) と新しい goroutine で実行します。実行される場所は違いますが、どちらの場合も実行されるのは「そのキーで最初に呼んだ呼び出し元」が作ったクロージャです。相乗りした 2 番目以降の呼び出し元が渡した関数は、一度も実行されません。

そのため、その関数が呼び出し元の ctx をキャプチャしていると、最初の呼び出し元がタイムアウトした瞬間にオリジンへの呼び出しが中断され、相乗りしていた全員が巻き添えでエラーになります。しかも「最初の呼び出し元」は実行のたびに変わるため、再現性のない散発的なエラーとして現れます。context.WithoutCancel で切り離し、独立したタイムアウトを与えるのが安全です。

singleflight には、ほかにも既知の注意点が 3 つあります。

1. エラーも全員に配られます。オリジンが失敗すると「1 件の一時的な失敗が 1000 件のエラー」になります。リトライやフォールバック（古いキャッシュを返すなど）と組み合わせるのが実務的です。Forget(key) で進行中の呼び出しを共有対象から外せます
2. DoChan 経由では、実行関数内の panic を回復できません。doChan の待機者がいる場合、doCall は捕捉した panic を切り離した goroutine から go panic(e) で再送出します。3.2 の errgroup と同じ設計思想です（Do を待機者なしで呼んだ場合は呼び出し元の goroutine 上で再 panic するため、そこでは recover できます）。8.7 も参照してください
3. 重複排除はプロセス内に閉じています。レプリカが N 台あれば、オリジンへの呼び出しは「1000 → 1」ではなく「1000 → N」になります。10.4 で channel について述べるのと同じ境界が、ここにも当てはまります

また、context.WithoutCancel で切り離した ctx は、最初の呼び出し元のリクエストスコープの値をそのまま持ち込みます。trace ID もそのひとつなので、1000 人分の共有呼び出しが「たまたま最初だった 1 リクエスト」に紐づいて記録されます。観測性を気にするなら、値を落とすか、待機者全員にリンクした新しいスパンを張ってください。

9.2 atomic.Pointer によるコピーオンライト

設定やルーティングテーブルのように「頻繁に読み込まれ、まれにしか更新されない」データには、mutex より atomic.Pointer が向いています。

type Config struct {
    Timeout  time.Duration
    Features map[string]bool
}
 
type Store struct {
    cfg atomic.Pointer[Config]
}
 
// 読み込み: ロックなし、数 ns
func (s *Store) Get() *Config { return s.cfg.Load() }
 
// 更新: 新しいインスタンスを作って差し替える
func (s *Store) Update(c *Config) { s.cfg.Store(c) }

読み込み側はロックを一切取りません。その代わり、格納した *Config の中身は絶対に書き換えてはいけません。更新は必ず新しい Config を作って Store します。返された *Config を受け取った側が Features マップに書き込むと、そこで race が発生します。

この規約を守れないチームでは、素直に RWMutex を使う方が安全です。

9.3 sync.Map を使うべきでない場合

sync.Map は「あらゆる場面で速い並行マップ」ではありません。ドキュメントが明示する適用場面は 2 つだけです。

1. あるキーが一度書かれた後は読まれるだけ（追記のみのキャッシュ）
2. 複数の goroutine が互いに素なキー集合を読み書きする

Go 1.24 で sync.Map の実装が HashTrieMap ベースに入れ替わり、書き込みや削除の多いワークロード、競合の激しいワークロードの性能は以前より大きく改善しています。「書き込みが多いなら map + RWMutex の方が速い」という古い経験則は、Go 1.24 以降そのままでは通りません。

それでも手順は変わりません。まず map + RWMutex で書き、プロファイルでロック競合が見えてから切り替えを検討してください。ドキュメントが挙げる 2 つの用途に当てはまらないなら、性能上の理由で sync.Map を選ぶ前に、自分のワークロードで必ずベンチマークを取ってください。

第10章 ケーススタディー — Webhook 配信ワーカー

ここまでのパターンを 1 つのシステムに統合します。要件は実際のプロダクトによくあるものです。

1. イベントを外部の URL へ HTTP POST で配信する
2. 同一宛先への配信順序を保証する（イベントの順序が意味を持つ）
3. 失敗したら指数バックオフでリトライし、上限を超えたら dead letter へ
4. 下流を DoS しないよう並行数を制限する
5. Graceful shutdown 時に取りこぼさない
6. 滞留と失敗を観測できる

10.1 設計判断: 順序保証と並行性の両立

要件 2 と要件 4 は一見矛盾します。順序を保証するには逐次実行が必要で、それではスループットが出ません。

解決は宛先によるシャーディングです。宛先のハッシュでレーンを決め、レーンごとに 1 本のワーカーを動かします。レーン内は逐次なので同一宛先の順序が保たれ、レーン間は並行なので全体のスループットが出ます。

レーン数がそのまま最大並行数になります。3.3 のリトルの法則で決めた値を使います。

10.2 実装

type Delivery struct {
    Endpoint string // 宛先。同一宛先内では順序を保つ
    Payload  []byte
    Attempt  int
}
 
type DispatcherConfig struct {
    Shards       int           // レーン数 = 実質の最大並行度
    QueuePerLane int           // レーンあたりのバッファ = 許容する滞留量
    MaxAttempts  int
    BaseBackoff  time.Duration
    MaxBackoff   time.Duration
    SendTimeout  time.Duration
}
 
type Dispatcher struct {
    cfg    DispatcherConfig
    sender Sender
    lanes  []chan Delivery
    dead   func(context.Context, Delivery, error)
    done   chan struct{} // 5.3 ポイント1：所有者の生死を伝える
 
    enqueued atomic.Int64
    rejected atomic.Int64
    sent     atomic.Int64
    failed   atomic.Int64
}
 
func NewDispatcher(cfg DispatcherConfig, sender Sender,
    dead func(context.Context, Delivery, error)) *Dispatcher {
 
    if cfg.Shards < 1 {
        cfg.Shards = 1 // 0 だと lane() の剰余がゼロ除算で panic する
    }
    lanes := make([]chan Delivery, cfg.Shards)
    for i := range lanes {
        lanes[i] = make(chan Delivery, cfg.QueuePerLane)
    }
    return &Dispatcher{
        cfg: cfg, sender: sender, lanes: lanes, dead: dead,
        done: make(chan struct{}), // 忘れると 10.2 の生存確認が無効化される
    }
}
 
// 観測用のアクセサ（7.4 / 10.3 のとおり、必ずメトリクスに出す）
func (d *Dispatcher) Stats() (enqueued, rejected, sent, failed int64) {
    return d.enqueued.Load(), d.rejected.Load(), d.sent.Load(), d.failed.Load()
}
 
func (d *Dispatcher) lane(endpoint string) chan Delivery {
    h := fnv.New32a()
    _, _ = h.Write([]byte(endpoint))
    // 剰余は uint32 のまま取る。int(h.Sum32()) は 32bit 環境で負になりうる
    return d.lanes[h.Sum32()%uint32(len(d.lanes))]
}

投入側は満杯なら即座に諦めます（3.5 の戦略 B）。

var ErrQueueFull = errors.New("dispatcher: queue full")
var ErrDispatcherClosed = errors.New("dispatcher: closed")
 
func (d *Dispatcher) Enqueue(ctx context.Context, del Delivery) error {
    // 8.3 の二段構え。default 付き select は決してブロックしないため、
    // ctx.Done() を腕に混ぜると「キャンセル済みかつ空きあり」で
    // 50% の確率で投入されてしまう
    if err := ctx.Err(); err != nil {
        return err
    }
 
    // 5.3 ポイント1 と同じ規約。ワーカーが死んだ後に受け付けない
    select {
    case <-d.done:
        return ErrDispatcherClosed
    default:
    }
 
    select {
    case d.lane(del.Endpoint) <- del:
        d.enqueued.Add(1)
        return nil
    default:
        d.rejected.Add(1)
        return fmt.Errorf("%w: %s", ErrQueueFull, del.Endpoint)
    }
}

Enqueue には 8.3 の優先度問題がそのまま現れます。default を持つ select は決してブロックしないので、case <-ctx.Done(): を加えても意味がないどころか有害です。キャンセル済みで空きもある場合、2 つの case が同時に準備できているため 50% の確率でキャンセルが無視されて投入されてしまいます。ノンブロッキング投入では、ctx のチェックを select の前に置いてください。

<-d.done の事前確認も必須です。Run が返った後（Shutdown がタイムアウトして stopWorkers() が走った直後など）、レーンにはまだ空きがあります。この確認がないと Enqueue は成功を返し、enqueued を増やし、そしてその配信は誰にも処理されず、dead letter にも載りません。10.4 で警告する「たまにイベントが消える」を、自分のコードで作り込むことになります。5.3 のアクターと 6.3 の Batcher に課したのと同じ規約が、ここにも必要です。

ただし、この確認は検査と送信のあいだの窓を消しません。default を抜けた直後に close(d.done) が走れば、その 1 件は依然として行方不明になります。8.3 と同じで、二段構えで取りこぼしは桁違いに減りますが、0 にはなりません。1 件も失えないなら、10.4 のとおり永続キューが必要です。

エラーは番兵（sentinel）にします。fmt.Errorf だけで返すと呼び出し側が errors.Is で判別できず、3.5 の戦略 B が求める「満杯なら 503 を返す」を実装できません。%w でラップして ErrQueueFull を伝播させます。

ワーカーはレーンごとに 1 本です。

func (d *Dispatcher) Run(ctx context.Context) error {
    defer close(d.done) // 停止したことを Enqueue 側に伝える
 
    g := &errgroup.Group{}
    for i := range d.lanes {
        g.Go(func() error {
            d.runLane(ctx, d.lanes[i])
            return nil
        })
    }
    return g.Wait()
}
 
func (d *Dispatcher) runLane(ctx context.Context, lane chan Delivery) {
    for {
        // 8.3 の二段構え：停止が来ていたら新しい配信を始めない
        select {
        case <-ctx.Done():
            d.drain(lane)
            return
        default:
        }
 
        select {
        case <-ctx.Done():
            d.drain(lane)
            return
        case del := <-lane:
            d.deliver(ctx, del)
        }
    }
}
 
func (d *Dispatcher) drain(lane chan Delivery) {
    // シャットダウン用に有限の猶予を切る（4.5 の判断 2 と同じ理由）
    ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
    defer cancel()
    for {
        select {
        case <-ctx.Done():
            d.abandon(lane) // 予算超過。送信は諦めて記録だけ残す
            return
        default:
        }
 
        select {
        case del := <-lane:
            d.deliver(ctx, del)
        default:
            return
        }
    }
}
 
// abandon は予算超過後の後始末。送信を試みず、dead letter 書き込み全体にも
// 上限を与えることで、停止時間が cap(lane) に比例して伸びるのを防ぐ
func (d *Dispatcher) abandon(lane chan Delivery) {
    ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
    defer cancel()
    for {
        select {
        case del := <-lane:
            d.failed.Add(1)
            d.dead(ctx, del, context.DeadlineExceeded)
        default:
            return
        }
    }
}

runLane の二段構えは必須です。素直に 1 段の select で書くと、8.3 の優先度問題が最悪の形で現れます。停止が来た時点でレーンに残りがあると 2 つの case が同時に準備でき、50% の確率で <-lane が選ばれます。すると deliver はキャンセル済みの ctx から sctx を作るため送信が即座に失敗し、リトライのバックオフ待ちで <-ctx.Done() に落ちて、実際には一度も送信できないまま dead letter 行きになります。デプロイのたびにレーンあたり平均 1 件が理由もなく失敗する、厄介なバグです。

（前段のノンブロッキング select を入れても、2 つの select のあいだにキャンセルが届く窓は残ります。厳密さが要るなら 8.3 のとおり deliver の先頭でも ctx.Err() を確認し、その場合はドレイン用の予算で処理してください。）

drain が予算を超えると abandon に切り替わるのは、停止時間を抑えるためです。素朴に「予算が切れても deliver を呼び続ける」実装にすると、停止時間が cap(lane) に比例して伸びます。ctx が死んでいるので送信は即失敗しますが、1 件ごとに Send 試行・ログ出力・dead letter 書き込みが走るからです。d.dead の呼び出しは 1 件ずつなら 2 秒で打ち切られますが、全体をまとめて縛るものがありません。QueuePerLane = 1000 で 1 件 20ms なら、10 秒の予算のあとにさらに 20 秒かかります。4.5 の判断 2 で警告した「ドレインが救うはずだったデータを、まさにその瞬間に失う」状況そのものです。

そこで、予算を超えたら送信を諦め、dead letter 書き込み全体にも別途 3 秒の上限を与えます。

abandon の 3 秒という保証には注意点が 2 つあります。ひとつは、d.dead の実装が渡された ctx の期限を尊重しなければならないことです（無視する実装なら予算は意味を失います）。もうひとつは、3 秒を使い切った時点でレーンに残っている分は、送信も記録もされず失われることです。

配信本体にリトライとバックオフを入れます。

func (d *Dispatcher) deliver(ctx context.Context, del Delivery) {
    for {
        sctx, cancel := context.WithTimeout(ctx, d.cfg.SendTimeout)
        err := d.sender.Send(sctx, del)
        cancel() // defer にしない（ループなので毎回解放する）
 
        if err == nil {
            d.sent.Add(1)
            return
        }
 
        del.Attempt++
        permanent := errors.Is(err, ErrPermanent) // 4xx はリトライしても無駄
        exhausted := del.Attempt >= d.cfg.MaxAttempts
 
        if permanent || exhausted {
            d.failed.Add(1)
            // 停止中でも dead letter は書き切る。ただし無期限にはしない
            dctx, dcancel := context.WithTimeout(context.WithoutCancel(ctx), 2*time.Second)
            d.dead(dctx, del, err)
            dcancel()
            return
        }
 
        wait := backoff(d.cfg.BaseBackoff, d.cfg.MaxBackoff, del.Attempt)
        slog.WarnContext(ctx, "webhook retry",
            "endpoint", del.Endpoint, "attempt", del.Attempt,
            "wait", wait, "err", err)
 
        select {
        case <-time.After(wait):
        case <-ctx.Done():
            // 停止指示が来たら、未送信のまま dead letter へ逃がす
            d.failed.Add(1)
            dctx, dcancel := context.WithTimeout(context.WithoutCancel(ctx), 2*time.Second)
            d.dead(dctx, del, err)
            dcancel()
            return
        }
    }
}
 
// backoff は完全ジッタ付き指数バックオフ。戻り値の範囲は [1, maxWait]。
// rand は math/rand/v2（v1 に Int64N はありません）。
func backoff(base, maxWait time.Duration, attempt int) time.Duration {
    if maxWait <= 0 {
        return 1 // rand.Int64N(0) は panic する
    }
    shift := attempt - 1
    if shift < 0 {
        shift = 0 // 負のシフト量は実行時 panic になる
    }
    d := base << min(shift, 16) // シフト量を抑えてオーバーフローを防ぐ
    if d > maxWait || d <= 0 {
        d = maxWait
    }
    return time.Duration(rand.Int64N(int64(d)) + 1)
}

10.3 埋め込まれた判断の一覧

このコードは短いですが、ここまでの章の判断がほぼすべて入っています。

deliver で defer cancel() を使っていないのは、deliver が for ループだからです。defer にすると、関数が返るまで全リトライ分の cancel が積み上がります。リトライ 10 回なら 10 個の未解放コンテキストが残ります。ループ内で context.WithTimeout を使うときは、defer ではなくその場で cancel() を呼んでください。このリークは go vet の lostcancel では検出されないので、見つけにくい問題です。

d.dead に期限付きの context.WithoutCancel を渡すのは、dead letter への書き込みが「配信を諦めた」という最も失いたくない記録だからです。停止中（ctx がキャンセル済み）に到達する経路があるので、両方の呼び出し箇所でキャンセルを切り離します。ただし context.WithoutCancel が返すコンテキストにはデッドラインが一切ありません（Deadline() は ok == false を返します）。そのまま渡すと停止時間の上限が消えるので、必ず WithTimeout を重ねてください。

backoff は完全ジッターを使っています。rand.Int64N(int64(d)) + 1 は 1 以上 d 以下の値を返します。ジッターのないバックオフだと、障害から復旧した瞬間に全リトライが同時に殺到し、下流を再び落とします。ジッターを入れることでリトライが時間軸に散らばります。

10.4 残る課題 — プロセス内キューの限界

このディスパッチャーには、次の限界があります。

• プロセスが SIGKILL されるとキューの中身は消えます。drain は graceful な停止しか救いません
• 水平スケールすると順序保証が壊れます。同じ宛先へのイベントが別インスタンスに振り分けられうるためです。同様に、Shards の値を変えると hash % N の割り当てが変わるので、レーン数の変更自体が順序保証を壊します
• 1 つの死んだ宛先が、同じレーンに載った健全な宛先を巻き添えにします。deliver はリトライをレーン内でインラインに待つため、MaxAttempts × MaxBackoff（例えば 5 回 × 30 秒）のあいだレーン全体が停止します。やがてレーンが満杯になり、何も悪くない宛先の Enqueue が弾かれ始めます。設計に起因するカスケード障害です。対策は、宛先ごとのサーキットブレーカーで即座に dead letter へ落とすか、リトライ対象を遅延キュー（別レーン）へ退避させてレーンを解放することです

「1 件も失ってはいけない」「厳密な順序保証が要る」場合、プロセス内 channel はキューとして不適格です。永続キュー（PostgreSQL の SELECT ... FOR UPDATE SKIP LOCKED、Redis Streams、Kafka など）を使い、channel はそこから取り出したジョブをプロセス内のワーカーに配る役割に限定してください。

channel は優れたプロセス内の並行制御機構ですが、永続キューでも分散キューでもありません。プロセスが死ねば中身は消え、プロセスをまたげば共有できません。この境界を理解していないと、「たまにイベントが消える」という再現困難な障害を抱え込むことになります。実務では、この境界の見極めが最も重要な判断かもしれません。

第11章 テストと観測

11.1 -race は必須、goleak は強く推奨

go test -race ./...

go.uber.org/goleak は、テスト終了時にまだ生きている goroutine を検出します。8.2 の goroutineleak プロファイルが検出するのは「原理的に到達不能なもの」なので、両者は補完し合います。

func TestCacheNoLeak(t *testing.T) {
    defer goleak.VerifyNone(t)
 
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel() // t.Fatal で抜けても必ず止める
    c := NewCache(ctx)
    // ... テスト本体 ...
 
    cancel()
    <-c.Done() // 所有者 goroutine の終了を待ってから検査する
}

<-c.Done() は省略しないでください。goleak.VerifyNone 自体はバックオフしながらリトライする（options.go の既定では最大 20 回）ので、cancel() 直後の一瞬の残存で即失敗するわけではありません。しかし <-c.Done() を書かないと、テストの成否が goleak のリトライ猶予に間に合うかどうかに依存することになります。停止処理にドレインや最終 flush が入れば猶予を超え、フレーキーなテストになります。

そもそも「停止を指示する」と「停止を待つ」は別の操作です。テスト可能なコンポーネントは後者の手段を公開 API として提供すべきで、5.2 の Cache に Done() を用意したのはこのためです（c.done を直接触るテストは同一パッケージ内でしか書けません）。

パッケージ全体に適用するなら TestMain を使います。

func TestMain(m *testing.M) {
    goleak.VerifyTestMain(m)
}

11.2 testing/synctest — 時間に依存するテストを決定論的にする

Go 1.24 で実験的に導入され、Go 1.25 で正式版になったパッケージです。並行コードのテストで最大の悩みの種だった「time.Sleep を挟んだフレーキーなテスト」を根本から解消します。

Go 1.25 で入ったのは Test と Wait の 2 つだけで、本節の例で使う Sleep は Go 1.27 で追加されました（api/go1.25.txt に Test / Wait、api/go1.27.txt に Sleep があります）。Go 1.25 / 1.26 では synctest.Sleep(d) の代わりに time.Sleep(d) の後に synctest.Wait() を書いてください（Sleep はその 2 つと厳密に等価です）。