Day 043 — mpsc Senderをforループでcloneし忘れると2周目で移動エラー。cloneしても余分なクローンを持ち続けると受信側は永遠に終わらない

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

Day 042 で run_batch_bounded を実装し、「タスクの数は動的」「同時実行数には上限がある」「全体にも個別にもタイムアウトがある」の3つを満たす設計を組みました。今日はこれに実務でほぼ必ず追加要求として出てくる機能——バッチの進捗をリアルタイムに外へ通知する——を足そうとして、チームメンバーが書いたコードをレビューします。

要求は単純です。「各タスクが完了するたびに、進捗イベントを tokio::sync::mpsc チャンネル経由で外部（CLIの進捗バー・WebSocket配信・ログ集約など、何でもよい）に送りたい」。バッチ処理関数の外にチャンネルの受信側 Receiver を渡し、送信側 Sender をバッチ処理関数に渡す設計です。

チームメンバーは以下のようなコードを書き、プルリクエストを出してきました。

📋 コピーasync fn run_batch_with_progress(
    inputs: Vec,
    max_concurrency: usize,
    per_task_timeout: Duration,
    total_timeout: Duration,
    progress_tx: mpsc::UnboundedSender,
) -> BatchReport {
    let semaphore = Arc::new(Semaphore::new(max_concurrency));
    let mut set: JoinSet, tokio::time::error::Elapsed>> =
        JoinSet::new();

    for input in inputs {
        let semaphore = Arc::clone(&semaphore);
        set.spawn(async move {
            let _permit = semaphore.acquire_owned().await.expect("semaphore is never closed");
            let result = tokio::time::timeout(per_task_timeout, fetch(input)).await;
            if let Ok(Ok(output)) = &result {
                let _ = progress_tx.send(ProgressEvent::Completed { input, output: *output });
            }
            result
        });
    }

    // ... 以降の集計ループはDay 042のrun_batch_boundedと同じ
    todo!()
}

cargo build を叩いた瞬間にコンパイルが通りません。Arc<Semaphore> は Arc::clone して各タスクに渡しているのに、progress_tx はそのまま for ループの中で使い回そうとしています。これは Day 042 で見た「借用がスコープを逃げようとする」エラーとは別の種類のエラーです。

さらに、このコンパイルエラーを素直に直した後——つまり progress_tx を Arc<Semaphore> と同じように「ループの中でクローンしてから move」しても、もう1つ別の落とし穴が本番相当のコードに潜んでいます。こちらはコンパイルエラーにはならず、統合テストが（あるいは本番のリクエストが）いつまで待っても完了しないという形で発覚します。

今日はこの2段階の不具合——①コンパイルエラー（Sender のムーブ）と②実行時のハング（Sender のクローンの閉じ忘れ）——を順番に発見・修正します。tokio::sync::mpsc の「チャンネルが閉じる条件」を正確に理解しているかが問われる問題です。

📝 問題

以下の2つのバグを、この順番で発見・修正してください。fetch 関数・BatchReport 構造体は Day 042 と同一のものを使用します。

バグ①（コンパイルエラー）

上記のコード（progress_tx を for ループ内で move せずに使い回そうとしている版）をコンパイルすると何が起きるか、実際に出るエラーの種類を説明し、修正してください。

バグ②（実行時のハング）

バグ①を修正して run_batch_with_progress が正しくコンパイル・動作するようになったとします。これを次のように「バッチ完了時に追加でサマリーイベントを送る」ために progress_tx をもう1つクローンして保持するコードから呼び出すと、統合テストが完了しません。なぜハングするのかをSenderのクローン数とrecv()の関係から説明し、ハングしないコードに修正してください。

📋 コピーasync fn integrate_and_wait_broken(inputs: Vec) -> (BatchReport, Vec) {
    let (tx, mut rx) = mpsc::unbounded_channel();
    let done_notifier = tx.clone(); // バッチ完了後に「完了サマリー」を送るために保持
    let batch_fut =
        run_batch_with_progress(inputs, 4, Duration::from_secs(2), Duration::from_secs(5), tx);
    let collect_fut = async {
        let mut events = Vec::new();
        while let Some(event) = rx.recv().await {
            events.push(event);
        }
        events
    };
    let (report, events) = tokio::join!(batch_fut, collect_fut);
    let _ = done_notifier.send(ProgressEvent::Completed { input: 0, output: 0 });
    (report, events)
}

• ProgressEvent は以下を使用してください。

📋 コピー#[derive(Debug, Clone, PartialEq, Eq)]
enum ProgressEvent {
    Completed { input: u32, output: u32 },
}

• 両方のバグについて、修正前の状態をテストで再現し（バグ②はハングそのものではなく tokio::time::timeout で「完了しないこと」を検知する形でよい）、修正後にテストが通ることを確認してください。

🔍 ヒント（段階的開示）

📋 コピーfor input in inputs {
    let semaphore = Arc::clone(&semaphore);
    let progress_tx = progress_tx.clone(); // 追加: ループのたびに複製する
    set.spawn(async move {
        // ...
    });
}
drop(progress_tx); // 追加: 関数自身が持っている元のハンドルも手放す

✅ 模範解答

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

use std::sync::Arc;
use std::time::Duration;
use tokio::sync::{mpsc, Semaphore};
use tokio::task::JoinSet;

async fn fetch(input: u32) -> Result {
    tokio::time::sleep(Duration::from_millis(((input % 5) * 40) as u64)).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,
}

#[derive(Debug, Clone, PartialEq, Eq)]
enum ProgressEvent {
    Completed { input: u32, output: u32 },
}

/// バグ①（moveの二重使用）とバグ②（Senderクローンの閉じ忘れ）を両方修正した版。
///
/// 修正①: progress_tx は Arc と同様、ループの各周でclone()して
///        spawnに持ち込む。所有権を持つのは各タスクの中だけになる。
/// 修正②: 関数自身が最初に受け取った progress_tx（=もう使わない元のハンドル）を
///        ループの後で明示的にdropする。こうしないと、この関数の呼び出し元が
///        「Senderが全部dropされるのを待つ」設計（後述のrun_batch_with_progressの
///        呼び出し側）を組んだ場合に、この関数自身が最後の1個を握ったままになる。
async fn run_batch_with_progress(
    inputs: Vec,
    max_concurrency: usize,
    per_task_timeout: Duration,
    total_timeout: Duration,
    progress_tx: mpsc::UnboundedSender,
) -> BatchReport {
    let semaphore = Arc::new(Semaphore::new(max_concurrency));
    let mut set: JoinSet, tokio::time::error::Elapsed>> =
        JoinSet::new();

    for input in inputs {
        let semaphore = Arc::clone(&semaphore);
        let progress_tx = progress_tx.clone(); // 修正①: 周ごとに複製する
        set.spawn(async move {
            let _permit = semaphore
                .acquire_owned()
                .await
                .expect("semaphore is never closed");
            let result = tokio::time::timeout(per_task_timeout, fetch(input)).await;
            if let Ok(Ok(output)) = &result {
                let _ = progress_tx.send(ProgressEvent::Completed { input, output: *output });
            }
            result
            // progress_tx（このタスク用のクローン）はここでdropされる
        });
    }
    drop(progress_tx); // 修正②: 関数自身が持つ元のハンドルも手放す

    let mut report = BatchReport::default();

    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 {
                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
}

/// バグ②を再現する版: 「完了サマリーを送るため」に追加でcloneした
/// `done_notifier` を join! の完了後まで保持してしまっている。
/// この`done_notifier`が生きている間、rx.recv()は絶対にNoneを返さない。
async fn integrate_and_wait_broken(inputs: Vec) -> (BatchReport, Vec) {
    let (tx, mut rx) = mpsc::unbounded_channel();
    let done_notifier = tx.clone(); // <- このクローンがjoin!の間ずっと生き続ける
    let batch_fut =
        run_batch_with_progress(inputs, 4, Duration::from_secs(2), Duration::from_secs(5), tx);
    let collect_fut = async {
        let mut events = Vec::new();
        while let Some(event) = rx.recv().await {
            events.push(event);
        }
        events
    };
    let (report, events) = tokio::join!(batch_fut, collect_fut); // <- ここで永遠に止まる
    let _ = done_notifier.send(ProgressEvent::Completed { input: 0, output: 0 });
    (report, events)
}

/// 修正版: 「完了サマリー」を追加で送りたい要求自体は取り下げず、
/// done_notifierをjoin!より前に使い切ってdropしてから待ちに入る。
/// （今回はサマリー送信機能自体をシンプルに削除し、余計なcloneを作らない設計にした）
async fn integrate_and_wait_fixed(inputs: Vec) -> (BatchReport, Vec) {
    let (tx, mut rx) = mpsc::unbounded_channel();
    let batch_fut =
        run_batch_with_progress(inputs, 4, Duration::from_secs(2), Duration::from_secs(5), tx);
    let collect_fut = async {
        let mut events = Vec::new();
        while let Some(event) = rx.recv().await {
            events.push(event);
        }
        events
    };
    tokio::join!(batch_fut, collect_fut)
    // txの元ハンドルはrun_batch_with_progress内部でdrop済み(修正②)、
    // 各タスク用クローンもタスク終了時にdrop済みなので、
    // 全タスク完了時点でSenderのクローンは0個になり、rx.recv()はNoneを返す。
}

#[tokio::main]
async fn main() {
    let (report, events) = integrate_and_wait_fixed((1..=20).collect()).await;
    println!("{report:#?}");
    println!("received {} progress events", events.len());
}

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

    #[tokio::test]
    async fn broken_version_hangs_because_extra_sender_clone_outlives_the_join() {
        // 7や11の倍数を避け、全件成功させてprogressイベントの数を単純にする
        let inputs = vec![1, 2, 3, 4, 5];
        let result = tokio::time::timeout(
            Duration::from_millis(500),
            integrate_and_wait_broken(inputs),
        )
        .await;
        assert!(
            result.is_err(),
            "done_notifierがjoin!の間生きている限り、rx.recv()はNoneを返さずハングするはず"
        );
    }

    #[tokio::test]
    async fn fixed_version_completes_and_collects_all_progress_events() {
        let inputs = vec![1, 2, 3, 4, 5];
        let result =
            tokio::time::timeout(Duration::from_secs(3), integrate_and_wait_fixed(inputs)).await;

        let (report, events) = result.expect("余計なSenderクローンがなければ必ず完了するはず");
        assert_eq!(report.succeeded.len(), 5);
        assert_eq!(events.len(), 5); // 成功した5件分だけprogressイベントが届く
    }

    #[tokio::test]
    async fn application_errors_and_panics_do_not_emit_progress_events() {
        // 7の倍数はErr、11の倍数はpanicなのでprogressイベントは送られない
        let inputs = vec![7, 11, 14];
        let (report, events) = integrate_and_wait_fixed(inputs).await;

        assert_eq!(report.succeeded.len(), 0);
        assert_eq!(report.application_errors, 2); // 7, 14
        assert_eq!(report.panicked, 1); // 11
        assert_eq!(events.len(), 0); // 誰も成功していないのでイベントは0件
    }
}

🪜 Step-by-Step 解説

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

tokio::sync::mpsc が「明示的なclose()」を持たず、代わりに「全Senderのdrop」でチャンネルを閉じる設計にしているのは、Rustの所有権システム全体に一貫する思想——「リソースの生死をスコープと所有権で表現し、専用の解放APIを人間が呼び忘れる余地をなくす」——の延長です。Day 042のSemaphore permitがDropで自動解放されたのと同じ発想が、ここでは「チャンネルのクローズ」という別のリソース管理に適用されています。

ただし今回のバグが示す通り、この設計は「クローンをどこかに残さない」という責任を完全には肩代わりしてくれません。ArcやSenderを安易に複製して長く生かしてしまうコードは、コンパイラのチェックをすり抜けたまま「誰も待っていない・何も終わらない」不具合を生みます。所有権システムは「解放し忘れ」を防ぐ一方で、「複製しすぎ・持ちすぎ」は依然として設計者の責任である、という限界を理解しておく必要があります。

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

バグ①（progress_tx を .clone() せずにループで使い回す）を実際にビルドすると、以下のようなエラーになります。

📋 コピー// アンチパターン: progress_txをcloneせずループで使い回そうとする
for input in inputs {
    let semaphore = Arc::clone(&semaphore);
    set.spawn(async move {
        let _permit = semaphore.acquire_owned().await.unwrap();
        let result = tokio::time::timeout(per_task_timeout, fetch(input)).await;
        if let Ok(Ok(output)) = &result {
            let _ = progress_tx.send(ProgressEvent::Completed { input, output: *output }); // ここでエラー
        }
        result
    });
}

🌐 他言語との比較

Go・Javaは「誰が・いつクローズ（相当の合図）を送るか」を人間が設計・実装する必要があるのに対し、Rustは「所有権の生死」という言語機構がクローズ条件を代行します。ただし今回のバグが示す通り、代行してくれるのはあくまで「全クローンがdropされた」という事実の検知だけであり、「クローンをどこにも余分に残さない」設計自体は依然として書き手の責任です。

🏆 実務での使いどころ

• CLIツールの進捗バー: 長時間実行するバッチ処理の各タスク完了をmpscチャンネル経由でメインスレッドに送り、indicatif等のプログレスバーを更新する
• WebSocketでのリアルタイム進捗配信: サーバーサイドでバッチ処理を実行しつつ、SenderをWebSocketのブロードキャストタスクに渡してクライアントへ配信する
• ジョブキューのライフサイクル管理: ワーカープールでジョブを処理し、全ワーカーが終了（＝全Senderがdrop）したことを検知してリソースを解放する

⚠️ よくある誤解・ミス

🚀 次のステップ

• 発展: run_batch_with_progressに「進捗イベントの送信先が受信を止めてしまった（Receiverがdropされた）場合、send()がErrを返す」ケースへの対処を追加する（ログだけ出して処理を継続するか、バッチ自体を打ち切るかの設計判断）
• 次回予告: async trait（async-trait / RPITIT） — 非同期メソッドを持つトレイトの設計

🎯 自己評価

自分の回答

気づき・メモ