DynamoDBリースで実現するWebSocketワーカーフリート管理：フェイルオーバーから無停止デプロイまで

WebSocketワーカー管理が難しい理由

HTTPリクエストは処理して応答を返せばサーバー側に状態は残りません。一方、WebSocketは持続的な双方向チャネルであり、ワーカーはTCP接続を開いたままにし、上流から届くメッセージを処理し、keep-alive pingにも応答する必要があります。

このステートフル性が運用上のいくつかの課題を生みます。

• ワーカー障害: コンテナが停止すると接続が切れ、上流が一時的にバッファしても再割り当ての仕組みがなければデータは失われます。
• ローリングデプロイ: ECSが旧タスクを終了し新タスクを起動する間、接続の所有者が不在となるウィンドウが発生します。
• 水平スケーリング（スケールイン）: 増やすのは簡単ですが、減らす際は接続をドレインし、他のワーカーが引き継いだことを確認する必要があります。
• 二重クレーム: 2つのワーカーが同じ接続を所有していると誤認すると、上流へ重複接続しプロトコルエラーや重複処理を引き起こします。

ワーカーは上流へアウトバウンド接続を張るクライアントであるため、ネットワーク層ではなくアプリケーション層での調整が必要です。本記事では、Amazon DynamoDBの条件付き書き込みを分散ロックのように活用してこの課題を解決するパターンを共有します。

核心：リース（Lease）パターンと条件付き書き込み

リースとは「誰がこの接続を所有し、いつ所有権が失効するか」を保持するDynamoDBアイテムです。スキーマは以下の通りです。

desired_state（PK）とlease_expires_at_ms（SK）にGSIを張ることで、「アクティブであるべきなのに所有者がいない接続」を効率的に検索できます。

取得（Acquire）：アトミックな所有権確保

ワーカーはConditionExpressionにより、**「リースが存在しない（attribute_not_exists）、または既に失効している（lease_expires_at_ms < :now）」**場合のみ書き込みが成功するよう条件を付けます。2つのワーカーが同時に試行してもDynamoDBがアトミックに評価するため、正確に1つだけが成功し、他はConditionalCheckFailedExceptionを受け取ってバックオフします。

import time
from botocore.exceptions import ClientError

async def try_acquire_lease(table, conn_id: str, worker_id: str, lease_seconds: int = 20):
    now_ms = int(time.time() * 1000)
    try:
        table.update_item(
            Key={"Pk": f"CONN#{conn_id}"},
            UpdateExpression=(
                "SET lease_owner = :w, lease_expires_at_ms = :exp"
            ),
            ConditionExpression=(
                "attribute_not_exists(lease_owner) "
                "OR lease_expires_at_ms < :now"
            ),
            ExpressionAttributeValues={
                ":w": worker_id,
                ":exp": now_ms + lease_seconds * 1000,
                ":now": now_ms,
            },
        )
        return True
    except ClientError as e:
        if e.response["Error"]["Code"] == "ConditionalCheckFailedException":
            return False  # 他のワーカーが既に所有中
        raise

更新（Renew）：ハートビート

所有ワーカーは数秒ごとにリースを更新します。更新が失敗した場合、所有権を失ったことを意味するためループを終了します。

async def renew_lease(table, conn_id: str, worker_id: str, lease_seconds: int = 20) -> bool:
    now_ms = int(time.time() * 1000)
    try:
        table.update_item(
            Key={"Pk": f"CONN#{conn_id}"},
            UpdateExpression="SET lease_expires_at_ms = :exp",
            ConditionExpression="lease_owner = :w",
            ExpressionAttributeValues={
                ":w": worker_id,
                ":exp": now_ms + lease_seconds * 1000,
            },
        )
        return True
    except ClientError as e:
        if e.response["Error"]["Code"] == "ConditionalCheckFailedException":
            return False
        raise

解放（Release）と失効（Expired）

正常終了時はlease_expires_at_ms = 0に設定します。0であれば調整ループから即座に失効として見えるため、他のワーカーが待機なしで引き継げます。逆にワーカーがクラッシュすると更新が途絶え、リースは自然失効します。このときdesired_state = STARTEDのまま所有者不在となった接続が**オーファン（orphan）**です。

⚠️ 注意：時計同期 失効判定はワーカーが送信したnow値を基準とするため、全ワーカーの時計がある程度同期している必要があります。FargateはAmazon Time Syncにより数ms以内の誤差に収まりますが、他のインフラにデプロイする場合はNTPを必ず設定し、時計精度が保証されない環境ではリース期間を最大想定スキュー分だけ延ばして誤失効を防いでください。

調整ループとグレースフルシャットダウン

オーファン調整（Reconciliation）

全ワーカーが定期的（デフォルト60秒）にGSIを検索し、desired_state = STARTEDかつlease_expires_at_ms < nowのレコードを見つけます。各オーファンに対してtry_acquire_leaseを呼びますが、条件付き書き込みのため複数ワーカーが同時に調整を回しても正確に1つだけが成功します。

最悪の復旧時間は概ねリース失効（20秒）+ 調整間隔（60秒）= 約80秒です。グレースフルシャットダウンであればlease_expires_at_ms = 0により即時失効扱いとなり、次の調整サイクルで引き継がれます。

グレースフルシャットダウンの3ステップ

ECSがローリングデプロイやスケールインでSIGTERMを送信した場合：

1. シグナル伝播: signal_handlerが共有shutdown_eventをセットすると、全接続のハートビート・状態チェッカー・調整ループがwhile not shutdown_event.is_set()を抜けます。個別の終了ロジックを書く必要はありません。
2. 並列クリーンアップ: 接続を順次閉じると接続数に比例して遅くなるため、asyncio.gatherで同時にWebSocketを閉じリースを解放します。
3. 即時解放: lease_expires_at_ms = 0に設定し、他ワーカーが次の調整サイクルで即座に引き継げるようにします。

import asyncio, signal

shutdown_event = asyncio.Event()

def signal_handler():
    shutdown_event.set()

async def graceful_shutdown(active_connections: dict):
    # 全接続を並列に閉じ、リースを解放
    await asyncio.gather(*[
        conn.close_and_release(lease_expires_at_ms=0)
        for conn in active_connections.values()
    ], return_exceptions=True)

async def main():
    loop = asyncio.get_running_loop()
    loop.add_signal_handler(signal.SIGTERM, signal_handler)
    # ... ワーカー実行 ...
    await shutdown_event.wait()
    await graceful_shutdown(active_connections)

実務適用時の注意点とコスト

日本の開発現場での適用文脈 日本ではリアルタイム文字起こしやライブコマースのチャットなどで、当初は単一ワーカーで運用し、接続数が数百を超えたあたりでこのパターンが必要になります。DynamoDBは東京リージョンでもオンデマンド／プロビジョニングが選択可能で、ハートビート書き込みがコストの大部分を占めるため、プロビジョニング + Auto Scalingを初期から設定するのが安全です。オンデマンドのまま放置すると接続数に比例してWCUコストが線形に増加します。

批判的視点：このパターンの限界

• 時計依存性: 結局ワーカーの時計に依存します。時計が大きくずれる環境（例：オンプレミス混在）では安全マージンを大きく取る必要があり、その分フェイルオーバーが遅くなります。
• DynamoDB依存: 条件付き書き込みの原子性に依存する構造のため、他のストレージへの移行が困難です。マルチクラウドが必要なら別途etcdやConsulなどの調整サービスを検討すべきです。
• ホットパーティションのリスク: 接続ごとにハートビートで書き込みが発生するため、接続数が増えるとパーティション分散が重要になります。接続IDをキーにするのは良いですが、調整GSIの検索が集中しないようページサイズと間隔を調整してください。
• per-connection get_item: 小規模なら問題ありませんが、数百を超える場合はBatchGetItemでまとめてRCUを削減するのが望ましいです。

次のステップ学習の方向性

• AWS X-Rayでワーカー間の分散トレーシングを導入し、フェイルオーバー経路を可視化しましょう。
• 上流のリプレイやオフセットベースの再開（last_seq活用）を実装し、ワーカー障害から復旧までのデータギャップを埋めるロジックを追加すると完成度が上がります。
• 同様の分散ロックパターンを学ぶには、DynamoDBの条件式ドキュメントとGSI設計ガイドを併せて読むことをおすすめします。

まとめ：所有権を明示的に管理する

長寿命のWebSocket接続を大規模に運用するには、明示的な所有権追跡、自動フェイルオーバー、フリート全体での調整が不可欠です。本記事で紹介したパターンの要点は次の3つです。

1. DynamoDBの条件付き書き込み = 外部ロックサービス不要の分散調整。update_itemのConditionExpressionにより、ある時点で1つのワーカーだけが接続を所有することを保証します。
2. ハートビート + 調整ループ = 障害の全スペクトルをカバー。予期しない終了はリース失効で、ローリングデプロイはグレースフルシャットダウンで処理されます。
3. スケールアウト／インが安全になる。新規ワーカーはリースを取得し、退出するワーカーは解放するため、スケーリングが自然に安全になります。

リアルタイム文字起こし、IoTデータ取り込み、金融フィード、ライブイベントストリーミングなど、長寿命接続を扱うシステムであればこのパターンはそのまま適用可能です。完全な実装（ワーカーアプリ、Lambdaイベントルーター、Terraformテンプレート）は原文リポジトリで確認できます。詳細なアーキテクチャ解説は根拠資料を参照してください。

あわせて読みたい記事

• CSS offset-path 完全ガイド：SVGパスに沿って動くアニメーションの作り方
• Cloudflareで垂直型マイクロフロントエンド（VMFE）を構築する：チーム独立性と統合UXの解決策