ストリーミング API は、正常な経路が単純に見える。外部処理を開始し、結果の断片を送信し、最後に実行状態を succeeded に変更すればよい。
しかし、ユーザーが応答の途中でタブを閉じると話が変わる。HTTP 接続は終了したのにモデル呼び出しが継続したり、ジェネレーターはクローズされたのに実行記録が running のまま残ったりする可能性がある。完了イベントを受信した画面と、データベース上の最終状態が食い違うこともあり得る。
この問題を解決するには、接続、実行、永続記録のライフサイクルを個別に定義し、それらの間の終了責任を明確に紐付ける必要がある。 本記事は、Python の非同期ジェネレーターにおける最小限の再現から出発し、キャンセル処理とトランザクション境界、複数ワーカーにおける実行所有権へと拡大する。
例は Python 3.12.13 で検証した。AnyIO と SQLAlchemy に関する説明は、それぞれ 4.x 版および 2.0 版のドキュメントを基準とする。後半のジョブリソースやイベント保存構造は設計例であり、前半の再現実験とは区別する。
1. まず 4 つのライフサイクルを分離する
| 対象 | 開始 | 終了 | 終了が意味すること |
|---|---|---|---|
| 画面サブスクリプション | ユーザーが結果画面に進入 | 画面移動・タブ終了 | ユーザーがもうこの接続経由で結果を見ていない |
| HTTP ストリーム | リクエスト受理 | 接続切断・応答終了 | 該当する送信経路が終わった |
| 実行タスク | 実行者がジョブを引き受けた | 成功・失敗・キャンセル | その実行者による演算が終わった |
| 実行記録 | 実行リクエストの永続保存 | 終止状態のコミット | 他のリクエストでも最終結果を照会できる |
4 つのライフサイクルを常に一致させる必要はない。代わりに、プロダクトの挙動に応じて紐付けルールを決める。
- リクエストに依存する実行: 接続切断をキャンセル信号として扱う。後始末の責任は、リクエストを処理している実行者にある。
- 継続させるべきジョブ: 接続切断はサブスクリプション解除として扱う。実行責任は、リクエストから独立した実行者が持つ。
ここで判断が曖昧だと、同一プログラム内に矛盾したコードが生じる。ある層は接続終了時にタスクをキャンセルし、別の層は完了通知を待つ。どちらも個別には自然なコードだが、統合されたシステムとしての契約は成立しない。
2. async for から抜け出しても、上流のジェネレーターは即座に閉じるとは限らない
入れ子構造のストリーミングは、通常以下のように構成される。
외부 응답 생성기 → 실행 이벤트 생성기 → HTTP 응답 생성기このとき、各層が単に async for で読み、yield だけを呼び出すと、外側ジェネレーターの終了が内側ジェネレーターの明示的な終了に紐付かない可能性がある。特に、外側が値を受け取った直後に停止しているケースが重要だ。
以下の例は、上流リソースの後始末タイミングを検証する。
import asyncio
from contextlib import aclosing
async def upstream(closed):
try:
yield "first chunk"
await asyncio.Event().wait()
finally:
closed.append("upstream closed")
async def relay(source):
async with aclosing(source) as chunks:
async for chunk in chunks:
yield chunk
async def check():
closed = []
response = relay(upstream(closed))
assert await anext(response) == "first chunk"
await response.aclose()
# 이벤트 루프 종료나 GC를 기다리기 전에 정리가 끝나야 한다.
assert closed == ["upstream closed"]
asyncio.run(check())aclosing() を除去した別の再現では、上流ジェネレーターの参照を保ったまま外側ジェネレーターだけを閉じたとき、後始末リストが空であった。aclosing() を適用すると、response.aclose() が返ってくる前に上流の後始末が完了していた。消費タスクを cancel() した場合でも、後始末とキャンセル伝播を併せて確認した。
contextlib.aclosing() は、ブロックから離れる際に非同期ジェネレーターの aclose() を呼び出す。後始末タイミングをループなどの実行コンテキストに紐付けてくれる点が重要である。Python contextlib 公式ドキュメント
このルールはストリーミングアダプターごとに適用する必要がある。自作のジェネレーターだけでなく、SDK が返したストリームも、誰が閉じるか決めておく。HTTP フレームワークが最も外側の応答を閉じてくれるという事実だけでは、すべての上流リソースのライフサイクルは保証されない。
3. エラーとキャンセルは異なる制御フローである
実行記録を次のように処理するコードを考えてみる。
try:
result = await execute()
except Exception:
record_failure()
else:
record_success(result)Python 3.12 では、asyncio.CancelledError は BaseException のサブタイプである。したがって、上記の except Exception はキャンセルを処理しない。非同期ジェネレーターを aclose() すると、停止した位置には GeneratorExit が入る。キャンセルと明示的なジェネレータークローズを個別に検証する必要があるのはこのためである。asyncio タスクキャンセルに関するドキュメント、非同期ジェネレーターメソッドに関するドキュメント
実行状態の記録には、少なくとも次の不変条件が必要である。
- 正常に後始末された実行には終止状態が存在する。
- 成功として確定した実行を、遅延接続終了が
stoppedに上書きすることはない。 - 部分応答がある場合は、停止記録とともに保存する。
- キャンセル信号は、後始末を完了した後に上位層へ伝播させる。
単一の実行者内では、finalized のようなローカル変数が、重複クローズを防ぐのに役立つ。成功または失敗をコミットした後にこれを true に変更し、最後の後始末経路でまだクローズされていない実行を処理する方式である。
ただし、finally に到達したからといって、すべてのケースをキャンセルに分類してはならない。終了記録自体の DB エラー、プロセス中断、外部ジョブの結果不明確状態は、個別に扱う必要がある。ローカル変数 1 つで解決できる範囲は、同じ実行者内部の制御フローに限られる。
4. finally 内の await もキャンセルされ得る
後始末コードを finally に移すと、次の問題が見えてくる。
finally:
await save_terminal_state()この await は、どのようなキャンセルポリシーのもとで実行されるのか？
AnyIO のキャンセルスコープ内では、キャンセル状態が維持されている間、待機ポイントごとにキャンセル例外が再発する可能性がある。したがって、非同期終了記録には制限付きの後始末区間が必要になることがある。move_on_after(..., shield=True) のように、外部スコープのキャンセルを一時的に遮断しつつ、独自に制限時間を設ける方式である。この保護は AnyIO スコープの仕様に基づくものであり、任意の asyncio.Task.cancel() まで無条件に遮断できると仮定してはならない。AnyIO キャンセル・後始末に関するドキュメント
後始末区間では、以下を区別するとよい。
- リソース後始末: ソケット・応答ストリーム・一時ファイルを閉じる。
- 状態クローズ: 実行結果を永続保存する。
- 復旧対象マーク: 制限時間や DB 障害でクローズできなかった場合、復旧経路がそれを発見できるようにする。
ジョブ全体をキャンセルから保護すると、ユーザーの停止リクエストが無効になる。逆に保護区間を無制限にすると、終了処理が完了しない。保護範囲は短く保ち、クローズ失敗は観測可能なエラーとして残すべきである。
同期 DB 呼び出しを使用する場合、Python レベルの await キャンセルが割り込まなくても、クエリがイベントループを長時間ブロックする可能性がある。この場合でも、DB 接続・クエリの制限時間とプロセス終了ポリシーが必要である。
5. 完了を送信した事実と完了を保存した事実
次の順序は失敗区間を生む。
완료 이벤트 전송 → DB 커밋イベント送信直後にプロセスが終了すると、ユーザーは完了を見たのに、リフレッシュした画面では実行中と表示される可能性がある。
終止結果を再照会できる必要があるなら、次の順序のほうが扱いやすい。
결과·종결 상태 저장 → 커밋 → 완료 이벤트 전송この順序でも、コミット直後に接続が切れる可能性がある。しかし、すでに保存した結果は再照会できる。送信失敗を復旧できる方向へ、失敗区間を移したのである。
SQLAlchemy を使用する際は、flush() と commit() も区別する必要がある。flush() は、変更 SQL を現在のトランザクションに反映させる段階であり、当該変更が最終コミットされたという意味ではない。同じセッション内で更新済みオブジェクトが見えることだけで保存検証を終わらせてはならない。SQLAlchemy セッション・トランザクションに関するドキュメント
検証は、別の接続での新規トランザクションで実施する。プロデューサーセッションがまだ開いていても、終止状態、終了時刻、部分応答が照会できなければならない。長時間開いた読み取りトランザクションは、分離レベルによっては以前のスナップショットを表示し得るため、テストの読み取り境界も明確にする必要がある。
長時間の外部呼び出しを 1 つの DB トランザクションでラップする設計にも注意が必要である。実行開始、進捗チェックポイント、終了を短いトランザクションで記録すれば、接続・ロックの保持時間を削減でき、プロセス中断後の復旧基準を作りやすい。
6. 複数ワーカーでは終了権限も保存する必要がある
実行者を 1 つ増やすと、ローカルの finalized 変数だけでは不十分になる。
워커 A: 작업 인수 → 긴 지연
복구기: A의 리스 만료 판단
워커 B: 같은 작업 재인수 → 완료 저장
워커 A: 뒤늦게 복귀 → 오래된 결과 저장 시도この問題は「誰が最後に書き込んだか」では解決できない。現在の実行権限を持つ試行のみ保存可能でなければならない。
1 つの設計として、実行記録に以下の値を持たせる方法がある。
| フィールド | 意味 |
|---|---|
run_id | ユーザーがリクエストした論理実行 |
execution_epoch | ジョブを引き受けるたびに増える世代番号 |
lease_until | 現在の実行者が権限を維持している時刻 |
state | queued / running / succeeded / failed / stopped など |
ジョブの再引き受けは原子的に実行し、世代番号を増やす。実行者は自身が受け取った番号を使って保存する。例えば、終止更新は以下の条件を持つ。SQL はバインド変数の表記を用いた概念例である。
UPDATE execution_run
SET state = :terminal_state,
finished_at = CURRENT_TIMESTAMP,
result_ref = :result_ref
WHERE run_id = :run_id
AND execution_epoch = :my_epoch
AND state = 'running'
AND lease_until > CURRENT_TIMESTAMP;更新件数が 0 なら、終止権限を失ったか、すでに状態が変わったということである。その実行者は、終止通知や後続結果ポインタを新たに記録してはならない。進行イベントやチェックポイントにも同じ権限ルールが必要である。ファイルを先に保存した場合、試行別パスを使用し、最終結果ポインタは権限検証に成功したトランザクションで紐付けるほうが安全である。
リース更新と有効期限判定では、可能な限り DB の時刻を一貫して使用する。プロセス間の時計ずれによって実行所有権が揺らぐのを減らせる。世代番号は、古い実行者による書き込みを防ぐフェンシングトークンの役割を果たす。
しかし、この SQL は外部副作用を一度だけ実行させるとは限らない。A が外部リクエストを成功させた後、結果保存前に停止した場合、B のリトライが同じ副作用を引き起こす可能性がある。外部 API の冪等性キー、副作用実行前後の段階記録、結果照会または補償ジョブを別途設計する必要がある。冪等性キーも、リトライごとに変わる世代番号とは分離し、同じ論理副作用には同じキーを使用する。
7. 実行を継続させるジョブはサブスクリプション接続から独立させる
タブを閉じても実行を継続させる必要があるなら、以下の 3 つの経路を分離できる。
명령: 실행 요청 → 요청 식별자 중복 검사 → 작업 저장
실행: 작업 인수 → 단계 실행 → 상태·결과·이벤트 저장
관측: 저장된 진행 내역 조회 + 실시간 이벤트 구독ここで create_task() 呼び出しを追加するだけでは、プロセス内部の並列実行は作れる。しかし、プロセス再起動後のジョブ復旧、実行引き受け、リトライ基準などが自動的に生まれるわけではない。
サブスクリプションを開き直したときも、最後に受信したイベント以降を復元できる必要がある。SSE には、id フィールドと再接続時に Last-Event-ID を送信する標準仕様がある。ただし、サーバ側のイベント保管と再生は、アプリケーションが実装しなければならない。また、ブラウザのデフォルトの EventSource ではなく fetch ベースのストリームクライアントでは、再接続とカーソル伝達を独自に実装する必要がある。WHATWG Server-sent events 標準
永続イベントを設計するなら、(run_id, sequence) を基準に重複を排除し、同一実行内の順序を復元できる。試行番号も併せて保存すれば、リトライ前後の記録を区別しやすい。ただし、増分する数値を割り当てたとしても、コミット順序まで保証されるとは限らない。複数の書き手がいる場合、実行別直列化やコミット順序を考慮したカーソルルールが必要である。
すべてのトークンを永続イベントとして残すことも必須ではない。段階開始・終了・結果は保存し、トークンは一定サイズのチャンクまたは定期的なスナップショットとして集約する選択肢もある。必要な再生精度と書き込みコストを併せて決定すればよい。
8. 完了通知は保存された状態から派生させる
完了状態の保存と通知送信の間にも失敗区間が存在する。
- 状態保存後、通知送信前の障害: 完了しているのに通知がない。
- 通知送信後、配信確認保存前の障害: リトライで同じ通知が繰り返される。
1 つの手法として、終止状態を変更するトランザクション内で、通知送信用レコードも併せて保存する方法がある。別の配信者がこのレコードを読み取り、通知を送信して配信結果を記録する。終止更新が前述の所有権条件を通過した場合にのみ、通知レコードを生成すべきである。
送信自体は繰り返され得るため、通知には安定した event_id を設定し、受信側で重複を排除する。同一実行の終止通知を一意にする制約も役立つ。この方式なら、状態と通知意図を併せて保持できるが、外部通知チャネルに対して正確に一度の配信を保証するわけではない。
画面の通知文言もこの契約を反映すべきである。「完了」通知は保存済み結果を開ける状態でなければならず、「中止」は外部システムですべての変更が取り消されたという意味で使うべきではない。ジョブ進行中にすでに発生した副作用と、応答生成の中断は異なる事実である。
9. 正常応答よりも終了境界をテストする
検証単位を HTTP 200 にすると、これらの問題を見落としやすい。以下のように、失敗を注入する位置と、その後に成立すべき状態を一組で記載すると効果的である。
| 失敗注入位置 | 確認すべき不変条件 |
|---|---|
最初のチャンク直後に外側ジェネレーターを aclose() | 上流リソースの後始末が返還前に完了している |
| 上流応答を待っている最中のタスクキャンセル | キャンセルが伝播し、クローズ記録が残っている |
| 終止コミット後の完了イベント送信失敗 | 再照会すると確定済み結果が表示される |
| 正常完了後の応答オブジェクト後始末 | 成功を中止状態で上書きしていない |
| 終了記録の保存失敗 | 成功を偽装せず、復旧対象として残っている |
| リリース再引き受け後の旧ワーカー復帰 | 古い世代の状態・イベント書き込みが拒否されている |
| 通知配信後の配信確認保存失敗 | 再送されても同一通知として識別可能 |
| プロセス強制終了 | 期限切れジョブを検出し、ポリシーに基づき復旧・中断処理を行う |
本記事の最小再現では、最初のチャンク直後のクローズ、aclosing() による即時後始末、待機中のタスクキャンセル伝播を直接確認した。DB を使用した別の統合検証では、ストリームを閉じた直後に独立した接続で、中止状態と部分応答、終了時刻がコミットされているかを確認した。リース再引き受けや通知再送の項目は、提案した設計を実際に実装する際に追加すべき失敗注入シナリオである。
運用指標も、平均応答時間だけでは不十分である。リース期限切れの実行数、終止記録失敗数、終止状態と通知配信の時間差、古い実行世代の書き込み拒否回数などの値が、復旧経路の健全性を示す。
まとめ
ストリーミングシステムにおける終了は、複数の段階からなる。消費を停止し、上流リソースを後始末し、実行状態を確定し、その事実をユーザーに伝える。
設計レビューでは、以下の 3 つの質問をまず確認する価値がある。
- このジョブを終了させる責任は現在誰にあるか？
- その実行者がいなくなっても、次の実行者が判断するための記録が残っているか？
- ユーザーに送信した最終状態を再照会しても同じ結果が得られるか？
これら 3 点が明確であれば、キャンセルは例外処理 1 行に紛れ込まない。正常完了と同水準で検証可能な実行プロトコルとなる。
