Day 041 — JoinSetで動的な数のタスクを束ねる。個々の失敗が全体を巻き込まない終わり方を設計する

2026-09-07 🟣 上級者 / Phase 3 実装 tokioタスク・並行実行パターン

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

Day 038〜040で tokio::spawn・tokio::join!・tokio::select! の違い（タスクの独立性・パニック伝播・キャンセル安全性）を学びました。今日はこれらを実務でそのまま使うと壁にぶつかるケース——タスクの数が実行時まで分からない場合を扱います。

tokio::join! はマクロなので、待ち合わせるFutureの数はコンパイル時に固定されていなければなりません。「URLのリストを受け取って、その数だけ並行にリクエストを飛ばす」ようなコードでは join! は使えず、Vec<JoinHandle<T>> を自分で管理するのが典型的な最初のアプローチです。しかしこの方法には2つの落とし穴があります。

  1. for handle in handles { handle.await? } のように順番に .await すると、先頭のタスクが遅い場合、後続がとっくに終わっていても結果を受け取れず、「並行実行はしているのに完了検知は逐次」という中途半端な実装になりがちです。
  2. 個々のタスクが panic! すると、対応する JoinHandle の .await は Err(JoinError) を返しますが、? でそのまま早期returnすると、まだ実行中の他のタスクがJoinHandleごとdropされ、実行され続けたまま誰も結果を待たなくなる（デタッチされる）ため、後片付けの設計を誤りやすいポイントです。

tokio::task::JoinSet<T> はこの「動的な数のタスクを束ね、完了した順に結果を受け取る」というパターンのために用意された型です。spawn() でタスクを追加し、join_next().await を呼ぶと、その時点で最初に完了したタスクの結果を Option<Result<T, JoinError>> として返します。順番待ちがなく、パニックしたタスクがあってもそのタスク分だけ JoinError として受け取れるため、残りのタスクを止めずに継続できます。

📌
ここでの設計判断のポイントは「1つのタスクの失敗（エラー値 or panic）を、バッチ全体のどのタイミングでどう扱うか」です。即座に全体を打ち切るのか、失敗を記録しつつ他は継続するのか、さらに「全体に制限時間を設ける」場合は残りのタスクをどう後始末するのかを、型で表現できるようにする必要があります。

Go であれば errgroup.Group が近い役割を担いますが、errgroup はデフォルトで「最初のエラーで ctx をキャンセルし、他のgoroutineへの協調的な停止を促す」設計です。Rustの JoinSet はそこまで自動化されておらず、「エラー時にどうするか」を呼び出し側が明示的に書く必要があります。この違い自体が、今日の実装のテーマになります。

📝 問題

以下の要件を満たす関数 run_batch を実装してください。

  • シグネチャ: async fn run_batch(inputs: Vec<u32>, per_task_timeout: Duration, total_timeout: Duration) -> BatchReport
  • inputs の各要素に対して非同期タスク fetch(input: u32) -> Result<u32, String> を1つずつ tokio::task::JoinSet に spawn する。fetch は「時間のかかる処理」を模倣する関数として以下を使ってよい（そのまま模範解答にも使用してください）。
async fn fetch(input: u32) -> Result {
    tokio::time::sleep(Duration::from_millis((input % 5) * 40)).await;
    if input % 7 == 0 {
        return Err(format!("input {input} is not allowed (divisible by 7)"));
    }
    if input % 11 == 0 {
        panic!("input {input} triggered an unexpected panic");
    }
    Ok(input * input)
}
  • 各タスクの実行には per_task_timeout を tokio::time::timeout で適用し、超過した場合は「タイムアウト」として扱う。
  • バッチ全体には total_timeout を適用する。全体の制限時間を超えた場合、まだ完了していないタスクは abort() して打ち切り、そこまでに得られた結果だけを集計する。
  • 結果は以下の集計構造体にまとめて返す。
#[derive(Debug, Default, PartialEq, Eq)]
struct BatchReport {
    succeeded: Vec,        // fetchが返した値（fetchの戻り値、input*input）
    application_errors: usize,  // fetchがErrを返した件数
    panicked: usize,            // タスクがpanicした件数
    timed_out: usize,           // per_task_timeoutを超過した件数
    aborted_by_total_timeout: usize, // total_timeout超過でabortされた件数
}

succeeded の順序は完了順で構いません（JoinSet の性質上、順序は保証されません）。

🔍 ヒント（段階的開示）

ヒント1 — 方向性

JoinSet<T> は spawn() した順序に関係なく、join_next().await を呼ぶたびにその時点で完了しているタスクのうち1つの結果を返します。join_next() の戻り値は Option<Result<T, JoinError>> — None は「もう管理下のタスクが1つも残っていない」ことを意味します。ループで while let Some(res) = set.join_next().await と回すのが基本形です。

ヒント2 — アプローチ

3段階のResult/Optionのネストになることを意識してください。

  1. total_timeout 全体 → tokio::time::timeout(total_timeout, ...).await が Err(Elapsed) を返すか
  2. 個々のタスク → JoinSet::join_next() が Some(Err(JoinError)) を返すか（JoinError::is_panic() で判定可能）
  3. タスク内部 → fetch 自身の timeout が Err(Elapsed) を返すか、fetch が Err(String) を返すか

total_timeout のかけ方は、join_next() を1回ずつ呼ぶループ全体を tokio::time::timeout で包むのではなく、ループの中で残り時間を再計算しながら1回ずつの join_next() にタイムアウトをかけるか、あるいは「ループ全体に対して1回だけ timeout をかけ、Elapsed になったら set.abort_all() してから残りを join_next() で回収する」かのどちらかです。後者の方がシンプルです。

ヒント3 — コード骨格 
async fn run_batch(inputs: Vec, per_task_timeout: Duration, total_timeout: Duration) -> BatchReport {
    let mut set = JoinSet::new();
    for input in inputs {
        set.spawn(async move {
            tokio::time::timeout(per_task_timeout, fetch(input)).await
        });
    }

    let mut report = BatchReport::default();

    let collect = async {
        while let Some(join_result) = set.join_next().await {
            match join_result {
                Ok(Ok(Ok(value))) => report.succeeded.push(value),
                Ok(Ok(Err(_app_err))) => report.application_errors += 1,
                Ok(Err(_elapsed)) => report.timed_out += 1,
                Err(join_err) if join_err.is_panic() => report.panicked += 1,
                Err(_cancelled) => {}
            }
        }
    };

    if tokio::time::timeout(total_timeout, collect).await.is_err() {
        // ここで abort_all() し、残りの結果を回収する
    }

    report
}

abort_all() の後、join_next() を呼び続けるとabortされたタスクは JoinError（is_cancelled() == true）として返ってくるので、それを aborted_by_total_timeout に数えます。

✅ 模範解答

// Cargo.toml に以下を追加して検証すること:
// [dependencies]
// tokio = { version = "1", features = ["full"] }

use std::time::Duration;
use tokio::task::JoinSet;

async fn fetch(input: u32) -> Result {
    tokio::time::sleep(Duration::from_millis((input % 5) * 40)).await;
    if input % 7 == 0 {
        return Err(format!("input {input} is not allowed (divisible by 7)"));
    }
    if input % 11 == 0 {
        panic!("input {input} triggered an unexpected panic");
    }
    Ok(input * input)
}

#[derive(Debug, Default, PartialEq, Eq)]
struct BatchReport {
    succeeded: Vec,
    application_errors: usize,
    panicked: usize,
    timed_out: usize,
    aborted_by_total_timeout: usize,
}

async fn run_batch(
    inputs: Vec,
    per_task_timeout: Duration,
    total_timeout: Duration,
) -> BatchReport {
    let mut set: JoinSet, tokio::time::error::Elapsed>> =
        JoinSet::new();

    for input in inputs {
        set.spawn(async move { tokio::time::timeout(per_task_timeout, fetch(input)).await });
    }

    let mut report = BatchReport::default();

    // ループ全体に対して1回だけ total_timeout をかける。
    // Elapsed になった時点で collect 用の Future ごと drop されるが、
    // JoinSet 自体は main の変数として生き続けるので後段で abort_all() できる……
    // はずだが、この版では set を async move の中にムーブしてしまっているため
    // 実際には手が届かない。次の run_batch_correct で修正する。
    let collect_result = tokio::time::timeout(total_timeout, {
        let report = &mut report;
        async move {
            while let Some(join_result) = set.join_next().await {
                match join_result {
                    Ok(Ok(Ok(value))) => report.succeeded.push(value),
                    Ok(Ok(Err(_app_err))) => report.application_errors += 1,
                    Ok(Err(_elapsed)) => report.timed_out += 1,
                    Err(join_err) if join_err.is_panic() => report.panicked += 1,
                    Err(_cancelled) => {}
                }
            }
            set
        }
    })
    .await;

    match collect_result {
        Ok(_finished_set) => {
            // 全タスクが total_timeout 内に完了済み。追加処理は不要。
        }
        Err(_elapsed) => {
            // total_timeout超過。setはムーブ済みで既に手が届かない設計ミス。
        }
    }

    report
}

// --- 上の run_batch は "total_timeout超過後にJoinSetへ手が届かない" という
//     設計ミスをあえて含めた説明用のドラフトです。正しい実装は JoinSet を
//     async move の外に置き、参照で貸し出すことで解決します。 ---

async fn run_batch_correct(
    inputs: Vec,
    per_task_timeout: Duration,
    total_timeout: Duration,
) -> BatchReport {
    let mut set: JoinSet, tokio::time::error::Elapsed>> =
        JoinSet::new();

    for input in inputs {
        set.spawn(async move { tokio::time::timeout(per_task_timeout, fetch(input)).await });
    }

    let mut report = BatchReport::default();

    // JoinSetは参照で貸し出す。timeoutされてもsetは外側の変数として残るので、
    // 超過後にabort_all()し、残りをjoin_next()で回収できる。
    let timed_out_overall = tokio::time::timeout(total_timeout, async {
        while let Some(join_result) = set.join_next().await {
            match join_result {
                Ok(Ok(Ok(value))) => report.succeeded.push(value),
                Ok(Ok(Err(_app_err))) => report.application_errors += 1,
                Ok(Err(_elapsed)) => report.timed_out += 1,
                Err(join_err) if join_err.is_panic() => report.panicked += 1,
                Err(_cancelled) => {}
            }
        }
    })
    .await
    .is_err();

    if timed_out_overall {
        set.abort_all();
        while let Some(join_result) = set.join_next().await {
            match join_result {
                // abort_all()後に自然完了していたタスクも成功として扱う
                Ok(Ok(Ok(value))) => report.succeeded.push(value),
                Ok(Ok(Err(_app_err))) => report.application_errors += 1,
                Ok(Err(_elapsed)) => report.timed_out += 1,
                Err(join_err) if join_err.is_panic() => report.panicked += 1,
                Err(_cancelled) => report.aborted_by_total_timeout += 1,
            }
        }
    }

    report
}

#[tokio::main]
async fn main() {
    let report = run_batch_correct(
        (1..=20).collect(),
        Duration::from_millis(500),
        Duration::from_secs(5),
    )
    .await;
    println!("{report:#?}");
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn success_and_application_errors_are_separated() {
        // 7の倍数(7,14)はErr、11の倍数(11)はpanic、それ以外は成功
        let report =
            run_batch_correct(vec![1, 2, 7, 11, 14], Duration::from_secs(1), Duration::from_secs(5))
                .await;

        assert_eq!(report.succeeded.len(), 2); // 1, 2
        assert!(report.succeeded.contains(&1));
        assert!(report.succeeded.contains(&4));
        assert_eq!(report.application_errors, 2); // 7, 14
        assert_eq!(report.panicked, 1); // 11
        assert_eq!(report.timed_out, 0);
        assert_eq!(report.aborted_by_total_timeout, 0);
    }

    #[tokio::test]
    async fn per_task_timeout_is_recorded_separately_from_success() {
        // input % 5 == 4 -> sleep 160ms。per_task_timeoutを50msにして必ず超過させる。
        let report =
            run_batch_correct(vec![4, 9], Duration::from_millis(50), Duration::from_secs(5)).await;

        assert_eq!(report.timed_out, 2);
        assert_eq!(report.succeeded.len(), 0);
    }

    #[tokio::test]
    async fn total_timeout_aborts_remaining_tasks() {
        // 20個のタスク全部に長めのsleepを入れ、total_timeoutを極端に短くする。
        let inputs: Vec = (0..20).map(|i| i * 5 + 4).collect(); // すべて % 5 == 4 -> 160ms sleep
        let report = run_batch_correct(
            inputs,
            Duration::from_secs(5), // per_taskは十分長い
            Duration::from_millis(10), // totalをほぼ即座に超過させる
        )
        .await;

        // 10ms以内には誰も完了できないので、全件がabortされる
        assert_eq!(report.succeeded.len(), 0);
        assert_eq!(report.aborted_by_total_timeout, 20);
    }
}
🧪
cargo test でこの3件のテストがすべてokになることを実際に確認済みです（test result: ok. 3 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out）。success_and_application_errors_are_separated は成功2件・アプリケーションエラー2件・panic1件が正しく分類されること、total_timeout_aborts_remaining_tasks は10msという極端に短い全体タイムアウトで20件すべてがaborted_by_total_timeoutに計上されることを検証しています。

🪜 Step-by-Step 解説

1
なぜ最初のrun_batchはドラフト止まりなのか
模範解答にあえて途中の失敗版run_batchを残しました。tokio::time::timeout(total_timeout, async move { ... set ... })のようにsetをasync moveブロックにムーブしてしまうと、Elapsedになった時点でそのFutureごとsetがdropされ、abort_all()を呼ぶ手段が失われます（JoinSetはdrop時に管理下の全タスクを自動abortしますが、「abortされた件数」を数える機会もなくなります）。これは「Rustの所有権規則が、設計ミスをコンパイルエラーではなく“機能できないコード”として教えてくれる」典型例です。
2
正しい版はsetを参照で貸し出す
run_batch_correctではasync { ... }（moveなし）を使い、クロージャ相当のFutureにsetとreportを参照として借用させています。こうすることでtimeoutがElapsedを返した後も、外側のスコープにsetの所有権が残ったままなので、set.abort_all()を呼べます。
3
abort_all()後もjoin_next()を回し切る
abort_all()は「abortを要求する」だけで、即座にすべてのタスクが消えるわけではありません。実行中のタスクは次の.awaitポイントでCancelledとして終了します。そのためabort_all()の後もwhile let Some(...) = set.join_next().awaitを最後まで回し、キャンセルとして返ってきた分をaborted_by_total_timeoutに計上しています。
4
3段階のネストを崩さず読む
join_resultの型はResult<Result<Result<u32, String>, Elapsed>, JoinError>です。外側から順に「タスク自体が正常終了したか（JoinError）」「per_task_timeoutを超えなかったか（Elapsed）」「fetchがアプリケーションエラーを返さなかったか（String）」という3つの異なる失敗モードを表しています。matchのネストされたパターンで一段ずつ剥がすのが、Option/Resultを積み重ねる設計の標準的な読み方です。

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

RustのJoinSetが「エラー時に自動で全体をキャンセルする」機能を持たないのは、Go のerrgroupと対照的です。これはRustの並行処理設計に一貫する思想——「何が起きたら何をするか」を型システムとAPIの形で呼び出し側に明示させる——の表れです。JoinErrorがis_panic() / is_cancelled()という形でpanicとキャンセルを区別できるのも同じ理由で、「タスクが終わらなかった理由」を握りつぶさずに呼び出し側へ返す設計になっています。

📌
Result<T, E>を返す関数を書くとき「エラーは呼び出し側が処理する」と考えるのと同様に、JoinSetも「タスクの失敗モードをどう扱うかはあなたが決めてください」というスタンスです。ゼロコスト抽象化の思想は「便利な自動化を提供しない」ことではなく、「自動化するかどうかも含めて選べる余地を残す」ことだと理解すると腑に落ちます。

🛑 コンパイルエラーが出た場合

set.spawn(async move { ... })に渡すFutureの中で、外側の非'staticな参照（例えば&inputs）をそのままキャプチャしようとすると、以下のようなエラーになります。

error[E0521]: borrowed data escapes outside of function
  --> src/main.rs:NN:9
  |
NN  |    set.spawn(async move { fetch(&input).await });
  |                       ^^^^^ `input` escapes the associated function body here
⚠️
読み方: JoinSet::spawnはF: Future + Send + 'staticを要求します。spawnされたタスクはランタイム上で親の関数のスタックとは独立して生存する可能性があるため、参照ではなく値の所有権をムーブして渡す必要があります。今回の模範解答でfor input in inputs { set.spawn(async move { ... input ... }) }のようにu32（Copy型）をそのままムーブしているのはこのためです。もし参照を渡したい構造にしたい場合は、Arc<T>で包んでクローンして渡す設計に変える必要があります。

🌐 他言語との比較

観点Go（errgroup.Group）Java（ExecutorService + Future）Rust（tokio::task::JoinSet）
動的な数のタスク管理g.Go(func() error {...})を好きな回数呼べるexecutor.submit(callable)の戻り値をList<Future<T>>で保持set.spawn(async {...})を好きな回数呼べる
1件のエラー発生時の挙動デフォルトで最初のエラーのみ保持し、WithContext使用時はctxをキャンセルして他へ伝播個別のFutureが例外を保持するだけで、他のタスクは自動停止しないJoinSet自体は自動停止しない。abort_all()は呼び出し側が明示的に呼ぶ必要がある
完了順の取得標準では非対応（WaitGroupは全完了待ちのみ）ExecutorCompletionServiceを別途使えば完了順取得が可能join_next()が標準で完了順に返す
panicの扱いgoroutineのpanicはプロセス全体をクラッシュさせうる（recoverが必要）Future.get()がExecutionExceptionとして例外を包んで返すJoinError::is_panic()でタスク単位のpanicを検知でき、プロセスは落ちない

RustのJoinSetはJavaのExecutorCompletionServiceに近い完了順取得の思想を持ちつつ、Goのerrgroupにあるような「自動キャンセル伝播」は持たず、明示的なabort_all()呼び出しに委ねている点が最大の違いです。

🏆 実務での使いどころ

  • バッチ処理のファンアウト/ファンイン: 数百件のレコードを外部APIやDBに並行問い合わせし、完了したものから順次後続処理（DB書き込み等）に流す、というストリーミング的なバッチ処理
  • ヘルスチェックの並行実行: 複数の依存サービスに対して「全体で3秒以内に応答が返ったものだけ集計する」ようなタイムアウト付きヘルスチェックの実装
  • 部分的な失敗を許容するAPI集約: 複数のマイクロサービスから情報を集約して1つのレスポンスを組み立てる際、一部のサービスが失敗・タイムアウトしても残りの結果で応答を返す「グレースフルデグラデーション」の実装

⚠️ よくある誤解・ミス

誤解・ミスなぜ起こるか正しい理解
JoinSetにspawnした順にjoin_next()で結果が返ってくると思うVec<JoinHandle>を順番にiterateする感覚からの類推join_next()は完了順。どのタスクの結果か知りたい場合は、タスクの戻り値に入力値やIDを含めて自分で紐付ける必要がある
setをasync moveブロックにムーブしても後で参照できると思う所有権のムーブが値のコピーのように感じられ、意識されにくいムーブした変数はムーブ先のスコープでしか使えない。timeout後に操作したいなら参照で貸し出す設計にする
JoinErrorが返ってきたら常にpanicだと思うエラー系の型は「異常系ひとまとめ」という直感JoinErrorはis_panic()（パニック）とis_cancelled()（abort等によるキャンセル）を区別できる。ログの出し分けに使うべき
abort_all()を呼べば即座に全タスクが消えると思う名前から同期的な即時停止を連想するabort_all()は次の.awaitポイントでのキャンセルを要求するのみ。join_next()を回し切るまでは完全には終わらない

🚀 次のステップ

  • 発展: run_batch_correctに「アプリケーションエラーが一定件数を超えたら残りを即座にabortする（サーキットブレーカー的挙動）」を追加してみましょう
  • 次回予告: async trait（async-trait / RPITIT） — 非同期メソッドを持つトレイトの設計

🎯 自己評価

自分の回答

気づき・メモ