B システム設計/インフラ — campaign-notification-dispatcher 大量会員一斉通知配信の信頼性設計（同期ループ×冪等性欠如×fan-out×SLO）（MOps 販促システムチーム 大型セール通知未達・重複送信インシデント Bad→Good）

概要

問題

EC企業 MOps 販促システムチームが運用するキャンペーン通知配信サービス（campaign-notification-dispatcher）は、大型キャンペーン開始時に対象会員（最大800万人）へプッシュ通知・メール通知を一斉送信する。現行実装は、単一のCloud Runサービスが/dispatchリクエスト内でBigQueryから対象会員を抽出し、そのまま同期ループで通知プロバイダAPIを1件ずつ呼び出している。

先月起きたインシデント: 大型セール開始時、このジョブはCloud Runの60分タイムアウトに達して強制終了し、対象800万人のうち約210万人に通知が届かないまま処理が打ち切られた。担当者が気づいたのは、SNSでの「セール開始通知が来ない」という顧客の声がきっかけだった。慌てて/dispatchを再実行したところ、冪等性の仕組みが一切ないため、既に送信済みの約590万人にも通知が重複送信され、「同じ通知が3回来た」というクレームにつながった。

問題点を全て洗い出し、抽出/送信の分離 × Pub/Subバッチ単位fan-out × member単位idempotency_key(冪等性台帳) × Dead Letter Topic(DLQ) × concurrency/Semaphoreによるレート制限遵守 × Secret Manager × Terraformモジュール化 × SLI/SLO監視を考慮したBad→Goodリファクタリングを行ってください。

制約・前提条件

• GCP、Terraform 1.8+（googleプロバイダー5.x）、リージョンasia-northeast1
• 通知プロバイダAPIは1件あたり約150ms、レート制限は秒間500リクエストまで（超過で429エラー＋一定時間ブロック）
• 対象会員数は最大800万人、channelはpush/emailのいずれか
• 通知プロバイダのAPI鍵はCloud Runの環境変数に平文設定されており、Secret Manager未使用
• スケーリング設定はmin-instances=1, max-instances=1に意図的に固定されている（担当者いわく「並列化すると重複送信が怖いから」）
• 監視はCloud Runの標準ログのみで、送信成功率・失敗理由別件数・処理速度（件/秒）は可視化されていない
• Terraformコードは存在せず、Cloud Runの設定はGCPコンソールから手動変更されている
• 目標SLO: キャンペーン開始から30分以内に対象会員の99.5%へ送信完了（エラーバジェット0.5%）
• 監視基盤: Cloud Monitoring + DataDog Agent v7 併用
• GKE（Autopilot含む）を使う場合は、Cloud Runに対する優位性を具体的な理由とともに説明すること

悪い構成 (Before)

📋 コピー# campaign-notification-dispatcher/app/main.py
from fastapi import FastAPI

app = FastAPI()

@app.post("/dispatch")
async def dispatch_campaign(campaign_id: str):
    # 問題③: BigQuery抽出とAPI送信が同一リクエスト内。コールドスタート込みで起動に3分以上
    members = bq_client.query(
        f"SELECT member_id, channel FROM `mart.campaign_targets` WHERE campaign_id = '{campaign_id}'"
    ).result()

    sent, failed = 0, 0
    # 問題①: 800万件を1リクエスト内で同期ループ。Cloud Runの60分タイムアウトに全体が縛られる
    for member in members:
        try:
            notification_provider.send(member.member_id, member.channel, campaign_id)  # 1件あたり約150ms
            sent += 1
        except ProviderError:
            failed += 1
            # 問題②: リトライなし。例外を握り潰して次の会員へ進むだけで失敗が記録に残らない
            continue

    return {"sent": sent, "failed": failed}

# 問題④: 冪等性の仕組みが一切ない。再実行すると送信済み会員にも無条件で再送する
# 問題⑤: min-instances=1, max-instances=1に固定(「並列化すると重複送信が怖い」という理由)
# 問題⑥: 通知プロバイダAPI鍵がCloud Run環境変数に平文設定。Secret Manager未使用
# 問題⑦: Terraformコードが存在せず、Cloud Run設定はコンソールから手動変更されている

ヒント（段階的開示）

# Pub/Sub: バッチ単位fan-out × DLQ × retry policyのスケルトン
resource "google_pubsub_subscription" "notification_worker_push" {
  name                 = "notification-worker-push-sub"
  topic                = google_pubsub_topic.notification_tasks.id
  ack_deadline_seconds = 60

  dead_letter_policy {
    dead_letter_topic     = google_pubsub_topic.notification_tasks_dlq.id
    max_delivery_attempts = 5
  }
  retry_policy {
    minimum_backoff = "10s"
    maximum_backoff = "300s"
  }
}

# アプリ側冪等性チェックのスケルトン
async def send_idempotent(campaign_id: str, member_id: str, channel: str) -> bool:
    idempotency_key = hashlib.sha256(f"{campaign_id}:{member_id}".encode()).hexdigest()
    if await already_sent(idempotency_key):  # 送信済みログを確認
        return False  # 再送・再実行によるスキップ
    async with semaphore:  # プロバイダのレート制限(秒間500req)を超えないよう並行数を制御
        await notification_provider.send(member_id, channel, campaign_id)
    await mark_sent(idempotency_key)
    return True

問題点分析（7点）

障害伝播フロー図 — Bad vs Good の変換

模範解答

改善後アーキテクチャ

[Cloud Scheduler / 手動トリガー]
        │ campaign_id を publish
        ▼
[Pub/Sub topic: campaign-dispatch-requests]
        │ push
        ▼
[Cloud Run job: target-extractor]              ← 抽出専用。送信とタイムアウト制約を分離(修正1,3)
  - BigQueryから対象会員を抽出(ページング、10万件ずつ)
  - 500件単位でバッチ化し batch_id = hash(campaign_id + offset) を付与
        │ publish (16,000 batches / 800万件)
        ▼
[Pub/Sub topic: notification-tasks]
  - ack_deadline_seconds = 60
  - retry_policy: min10s / max300s（指数バックオフ）
  - dead_letter_policy: max_delivery_attempts = 5
        │ push
        ▼
[Cloud Run service: notification-worker]        ← concurrency/max-instancesで全体スループットを制御(修正4,5)
  - concurrency = 20, max-instances = 25
  - バッチ内はasyncio.Semaphore(20)で並行送信
  - 送信前にidempotency_keyの送信済みログをチェック(修正4)
  - Secret Managerからプロバイダ鍵を取得(修正6)
        │ 5回失敗
        ▼
[Pub/Sub topic: notification-tasks-dlq]
        │
        ▼
[Cloud Run job: dlq-reviewer]（手動トリガー） ← DLQ滞留メッセージを人手で確認し再publish

[Firestore: notification_sent_log]               冪等性台帳(idempotency_key = document ID)
[Secret Manager: notification-provider-api-key]   修正6: 平文環境変数の廃止
[Cloud Monitoring / DataDog]                       修正7: 送信速度・成功率・DLQ滞留の可視化

冪等性・リトライ・DLQ設計

📋 コピー"""notification-worker/app/handler.py

Pub/Sub push配信はat-least-once配信であり、かつ手動再実行によっても
同一バッチが複数回処理され得る。「同じ処理を何度実行しても結果が変わらない」
冪等性を持たせることで、min/max-instances固定を解除した並列化(修正4,5)
を安全に行える。
"""
from __future__ import annotations

import asyncio
import hashlib
import logging
from typing import Final

logger = logging.getLogger(__name__)

_PROVIDER_CONCURRENCY: Final[int] = 20  # プロバイダのレート制限(秒間500req)を守るためのバッチ内並行数上限
_MAX_RETRY_ATTEMPTS: Final[int] = 5     # Pub/Sub側のdead_letter_policyと合わせる


async def handle_notification_batch(batch: NotificationBatch) -> BatchResult:
    """通知バッチ(会員500件)を冪等に送信する。

    Args:
        batch: campaign_id・batch_id・会員リスト(member_id, channel)を含むバッチ。

    Returns:
        バッチ内の送信成功/スキップ/失敗件数をまとめた結果。

    Note:
        1件ずつFirestoreへ問い合わせるとレイテンシが積み上がるため、
        バッチ内の全idempotency_keyを1回のバッチ読み取りでまとめて
        確認してから送信ループに入る。
    """
    semaphore = asyncio.Semaphore(_PROVIDER_CONCURRENCY)
    keys = {m.member_id: _idempotency_key(batch.campaign_id, m.member_id) for m in batch.members}
    already_sent = await _bulk_check_sent(keys.values())  # 修正4: 冪等性チェックをバッチでまとめて実行

    async def _send_one(member: Member) -> SendOutcome:
        key = keys[member.member_id]
        if key in already_sent:
            return SendOutcome.SKIPPED_DUPLICATE  # 再送・再実行によるスキップ
        async with semaphore:  # 修正5: 並列化してもレート制限は守られる
            try:
                await notification_provider.send(member.member_id, member.channel, batch.campaign_id)
            except ProviderTransientError:
                return SendOutcome.FAILED_RETRYABLE  # 修正2: 例外を握り潰さず結果として返す
        await _mark_sent(key)
        return SendOutcome.SENT

    outcomes = await asyncio.gather(*(_send_one(m) for m in batch.members))

    if any(o == SendOutcome.FAILED_RETRYABLE for o in outcomes):
        raise BatchPartiallyFailedError(batch.batch_id)  # 非200を返しPub/Subに再送させる(DLQへ最終隔離)

    return BatchResult.from_outcomes(outcomes)


def _idempotency_key(campaign_id: str, member_id: str) -> str:
    """campaign_id×member_id単位で一意な冪等性キーを生成する。"""
    return hashlib.sha256(f"{campaign_id}:{member_id}".encode()).hexdigest()

📋 コピー# terraform/cloud_run_service.tf（抜粋）
resource "google_cloud_run_v2_service" "notification_worker" {
  name     = "notification-worker"
  location = "asia-northeast1"

  template {
    # 修正5: min/max-instances固定(1,1)を撤廃し、冪等性を前提に安全に並列化する
    scaling {
      min_instance_count = 0
      max_instance_count = 25
    }
    containers {
      image = var.notification_worker_image
      resources {
        limits = { cpu = "1", memory = "512Mi" }
      }
      env {
        name = "NOTIFICATION_PROVIDER_API_KEY"
        value_source {
          secret_key_ref {
            secret  = google_secret_manager_secret.notification_provider_api_key.secret_id  # 修正6
            version = "latest"
          }
        }
      }
    }
    max_instance_request_concurrency = 20  # プロバイダのレート制限を守るための1インスタンスあたり並行数上限
    timeout                           = "60s"
  }
}

# terraform/pubsub.tf（抜粋）
resource "google_pubsub_subscription" "notification_worker_push" {
  name                 = "notification-worker-push-sub"
  topic                = google_pubsub_topic.notification_tasks.id
  ack_deadline_seconds = 60

  # 修正2: 5回失敗したメッセージはDLQへ隔離。正常なメッセージの処理能力を守る
  dead_letter_policy {
    dead_letter_topic     = google_pubsub_topic.notification_tasks_dlq.id
    max_delivery_attempts = 5
  }
  retry_policy {
    minimum_backoff = "10s"
    maximum_backoff = "300s"
  }

  push_config {
    push_endpoint = "https://notification-worker-xxxx.a.run.app/pubsub/push"
    oidc_token {
      service_account_email = google_service_account.notification_pubsub_invoker.email
    }
  }
}

ポイント解説

実務への応用

• MOps販促システムチームでは、キャンペーン通知の他にも「クーポン発行通知」「カート放棄リマインドメール」「在庫復活通知」など、対象会員数が数百万〜数千万規模になり得る一括配信処理が今後も発生しうる: 今回設計した「抽出/送信の分離 × バッチ単位fan-out × member単位冪等性台帳 × DLQ × SLI/SLO監視」というパターンはそのまま横展開できる
• 「並列化・自動化すると事故が怖いので手動運用・逐次処理のままにする」という判断は一見安全に見えるが、実際には障害時の被害（今回のタイムアウト+重複送信）をむしろ拡大させることがある: 安全策として選ぶべきは「並列化しない」ではなく「並列化しても安全な構造（冪等性・DLQ）を先に作る」ことだという判断軸は、他の一括処理の設計判断にも応用できる

今日のまとめ

次のステップ

• 発展問題: キャンペーンが複数同時に開始される場合（例: 複数事業部が同日に別々のキャンペーンを実行）、notification-workerのmax-instancesをキャンペーン間でどう配分するか、Pub/Subのトピック設計（単一トピック+filter vs キャンペーン別トピック）を含めて検討せよ
• 発展問題: Firestoreの冪等性台帳（notification_sent_log）は無制限に増え続ける。TTLポリシーやBigQueryへの定期エクスポートを含めたライフサイクル管理を設計せよ
• 参考: Cloud Run v2 scaling / concurrency設定、Cloud Pub/Sub Dead Letter Topics、Cloud Run jobsとservicesの使い分け、Cloud Monitoring SLOモニタリング機能

自己評価（あとで記入）