なぜWebSocketワーカーは常に問題を起こすのか
リアルタイム文字起こしサービスが同時に500件の会議を処理しているとします。各会議には専用のWebSocket接続が必要で、1つのワーカーが100以上の接続を抱えています。ワーカーが1つ落ちるだけで100接続が切断され、運用担当が手動で再起動するまで2〜3分間データが失われます。
HTTPリクエストとWebSocketは根本的に異なります。HTTPはリクエスト-レスポンスでサーバーが状態を持ちませんが、WebSocketは持続的な双方向チャネルであり、ワーカーはTCP接続を維持し、keep-alive pingに応答し続ける必要があります。このステートフル性が以下の運用課題を生みます。
- ワーカー障害: プロセスが落ちると接続が放置され、上流が一時的にバッファしても再割当の仕組みがなければデータ損失
- ローリングデプロイ: ECSが旧タスクを停止すると接続が切れ、所有権調整がなければ「オーナー不在の接続」が発生
- 水平スケーリング: 増やすのは簡単だが減らすのが難しい。退出するワーカーの接続をドレインし、他ワーカーが引き継いだことを確認する必要がある
- 二重取得(Double-claiming): 2つのワーカーが同じ接続を所有していると誤認すると、上流に重複接続してデータ重複処理やプロトコルエラーを引き起こす
ワーカーはクライアントとして外部へ出ていく(outbound)接続を張るため、ネットワーク層ではなくアプリケーション層での調整メカニズムが不可欠です。本記事ではAWS Architecture Blogで共有されたDynamoDBリースパターンを実務視点で解説します。(根拠資料: AWS Architecture Blog)
核心アイデア: DynamoDBの条件付き書き込み = 分散ロック
別途のコーディネーションサービス(ZooKeeper、etcd)なしに、DynamoDBの**条件付き書き込み(Conditional Write)だけで分散ロックを実装します。各接続にリース(lease)**を付与し、ワーカーは定期的に更新(heartbeat)する必要があり、更新が途切れるとリースが失効して他のワーカーが引き継げる仕組みです。
テーブルスキーマ
| 属性 | 型 | 説明 |
|---|---|---|
Pk | String (PK) | 接続ID (例: CONN#meeting-123) |
desired_state | String | STARTED または STOPPED |
ws_url | String | 上流WebSocket URL |
lease_owner | String | 現在の所有ワーカーID |
lease_expires_at_ms | Number | 失効時刻 (epoch ms) |
last_seq | Number | 最終処理シーケンス (再開用) |
desired_state(PK) + lease_expires_at_ms(SK)にGSIを張れば、「アクティブであるべきなのにオーナー不在の接続」を一括で照会できます。
リース取得 — 条件付き書き込みの真髄
# worker.py — リース取得の核心ロジック
async def try_acquire_lease(table, conn_id: str, worker_id: str, now_ms: int) -> bool:
try:
await table.update_item(
Key={"Pk": f"CONN#{conn_id}"},
UpdateExpression=(
"SET lease_owner = :w, lease_expires_at_ms = :exp"
),
# 核心: リースが存在しない(attribute_not_exists)か、既に失効している場合のみ成功
ConditionExpression=(
"attribute_not_exists(lease_owner) "
"OR lease_expires_at_ms < :now"
),
ExpressionAttributeValues={
":w": worker_id,
":exp": now_ms + 20_000, # 20秒リース
":now": now_ms,
},
)
return True
except table.meta.client.exceptions.ConditionalCheckFailedException:
# 他ワーカーが既に取得済み → 優雅に撤退
return False
2つのワーカーが同時に同じ接続を狙っても、DynamoDBがアトミックに条件を評価するため、正確に1つのワーカーだけが成功します。残りはConditionalCheckFailedExceptionを受け取りバックオフすれば完了です。
3つの並行タスクで接続を守る
リースを取得したワーカーはasyncio.gatherで3つのタスクを同時に走らせます。1つでも終了すれば接続全体がクリーンアップされる構造です。
async def manage_connection(conn_id, ws_url, table, worker_id):
try:
await asyncio.gather(
heartbeat_loop(conn_id, table, worker_id), # 5秒ごとにリース更新
receive_loop(conn_id, ws_url, table, worker_id), # メッセージ受信/保存
check_desired_state(conn_id, table), # 10秒ごとにSTOP確認
)
finally:
# どんな理由で終わっても必ずリースを返却
await release_lease(table, conn_id, worker_id)
- heartbeat_loop: 5秒ごとに
renew_lease。失敗すれば所有権喪失 → 即座に終了 - receive_loop: 上流メッセージをDynamoDB messagesテーブルに保存
- check_desired_state: 外部からSTOPイベントが来ていないか確認
Graceful Shutdown — デプロイ中の無停止の秘訣
ECSがSIGTERMを送ると、ワーカーは3段階でクリーンアップします。
- シグナル伝播: 共有
shutdown_eventをset → すべてのループがwhile not shutdown_event.is_set()から即座に脱出 - 並列クリーンアップ:
asyncio.gatherで全WebSocketを同時に閉じ、リースを同時に返却 (接続数に関係なく終了時間はほぼ一定) - 即時返却:
lease_expires_at_ms = 0に設定 → 他ワーカーのreconciliationクエリに即座に失効したように見える → 20秒待たずに次サイクルで引き継ぎ
Orphan Reconciliation — 最後の安全網
ワーカーがクラッシュしてSIGTERMハンドラを実行できなかった場合は? 60秒ごとに走るreconciliationループがGSIを照会し、desired_state=STARTED AND lease_expires_at_ms < nowの孤児接続を見つけて再取得します。複数ワーカーが同時に走っても条件付き書き込みのおかげで正確に1つだけが勝ちます。最悪の復旧時間はリース20秒 + reconciliation 60秒 = 約80秒です。
実務で必ず押さえるべき落とし穴
1. 時計精度 — このパターンのアキレス腱
リース失効の判定はワーカーのローカル時計で生成したepoch msを基準にします。DynamoDBサーバー側の時計ではなく、呼び出したワーカーが渡した:now値を使用します。つまり、全ワーカーの時計が同期している必要があります。
- AWS FargateはAmazon Time Sync Serviceによりリージョン内のskewを数ms以内に維持 → 20秒リース + 5秒heartbeatの安全マージンで十分
- Fargate外(オンプレミス、他クラウド)で動かすなら必ずNTPを設定し、clock driftを監視してください。 時計精度を保証できない環境では、想定される最大skew分だけリース期間を延ばす必要があります。
2. コスト — heartbeat書き込みが支配的
各接続がheartbeatごとにupdate_item 1回 = 1 WCU。5秒周期の場合:
| アクティブ接続 | WCU/秒 | オンデマンド月額 | プロビジョニング月額 |
|---|---|---|---|
| 100 | 20 | ~$65 | ~$10 |
| 500 | 100 | ~$325 | ~$47 |
| 2,000 | 400 | ~$1,300 | ~$190 |
持続的に高い接続数を維持するプロダクションなら、オンデマンドよりプロビジョニング + Auto Scalingが圧倒的に安価です。heartbeat書き込みは予測可能で一定のため、プロビジョニング容量と相性が良いです。
3. スケール時の注意
check_desired_state()を接続ごとに個別GetItemで回すと、10秒ごとにN回の読み取りが発生します。大規模フリートでは**BatchGetItemで中央ループ1つ**に統合し、N回 → 1回のバッチに削減してください。
4. チューニングガイド
| 設定 | デフォルト | 根拠 |
|---|---|---|
| リース期間 | 20秒 | 短いネットワークhiccupは耐え、failoverは高速に |
| Heartbeat間隔 | 5秒 | 失効前に4回更新試行 (4:1安全マージン) |
| Reconciliation間隔 | 60秒 | 復旧速度 vs DynamoDB読み取りコストのバランス |
| Max接続/タスク | 700 | 接続あたり2〜5MBのメモリプロファイリングに基づく |
| Scale-outクールダウン | 2分 | トラフィックスパイク時のthrashing防止 |
| Scale-inクールダウン | 15分 | 接続安定化後に容量縮小 |
5. このパターンが不向きなケース
- 短命接続(数秒以内): リースのオーバーヘッドが利得を上回ります
- 強い一貫性が必須な金融トランザクション: GSIクエリはデフォルトで結果整合性 → RCUを半分節約する代わりに若干の遅延を許容
- 単一ワーカーで十分な小規模: 素直にSQS + 単一コンシューマーで十分です
まとめ: このパターンが本当に与えるもの
このアーキテクチャの本質は、**「別途のコーディネーションサービスなしにDynamoDB1つで分散ロック + ドメイン状態 + フェイルオーバーを同時に解決する」**という点です。ZooKeeperやetcdを新たに立てる代わりに、既に使っているかもしれないDynamoDBテーブル1つで完結します。
核心的なTakeaway 3行まとめ:
- 条件付き書き込み = アトミックな分散ロック。
attribute_not_exists OR lease_expires_at_ms < :now一つで、正確に1つのワーカーだけが接続を所有します。 - Heartbeat + Reconciliationが障害の全スペクトルをカバー。 予期せぬ終了はリース失効で、ローリングデプロイはgraceful shutdownで処理。新規ワーカーは取得し、退出ワーカーは返却するのでスケーリングが安全です。
- 適用範囲が広い。 リアルタイム文字起こし、IoTデータ収集、金融フィード処理、ライブイベントストリーミングなど、長命なWebSocket接続を扱うあらゆるシステムにそのまま移植可能。
日本の開発エコシステムにおける適用文脈
日本のSI/スタートアップ環境でこのパターンが特に有用なのは、「運用人数が薄いチーム」です。ワーカーが落ちるたびに深夜に叩き起こされて手動再起動する構造から脱するには、この程度の自動復旧レイヤーは必須と言えます。ただし、日本で多いオンプレミスIDC + ハイブリッドクラウド構成の場合、時計同期(NTP)を必ず事前点検してください。閉域網でchrony設定が不十分だと、リースが誤って失効し、接続が延々と再割当される地獄を見ることになります。
次のステップ学習方向
- 分散トレーシング: AWS X-Rayでワーカー間のend-to-end可視性を確保
- 再接続ロジック: 上流replayまたはoffsetベースの再開で、ワーカー障害〜復旧間のデータギャップを埋める
- IaC: TerraformでDynamoDBテーブル + SQS + ECSクラスターまで一括プロビジョニング (原文GitHubリポジトリにテンプレート同梱)
併せて読みたい記事
- 「GallupがAmazon Bedrockで数千人のリーダーをリアルタイムコーチングした方法 (アーキテクチャ解剖)」 — AWS上で大規模リアルタイムワークロードをどう設計するか、実例で学ぶ
- 「工場アラーム数百件、AIエージェントが5秒で処理する構造 (NVIDIA Nemotron実践分析)」 — 大量イベントをリアルタイム処理するもう一つのアプローチ