B システム設計/インフラ — coupon-issuance-service Pub/Sub非同期クーポン発行処理の信頼性設計（重複配信×順序保証×DLQ×冪等性）（MOps 販促システムチーム クーポン二重発行インシデント Bad→Good）

概要

問題

ECサイト MOps 販促システムチームが運用するクーポン発行サービス（coupon-issuance-service）は、キャンペーン管理基盤（campaign-orchestrator）がPub/Subトピックcampaign-eventsへpublishするcoupon.issue_requested（発行）・coupon.revoke_requested（注文キャンセル時の取り消し）イベントを、Cloud Runのpushサブスクリプションで受信し、Firestoreにクーポンを発行・失効させている。

先週起きたインシデント: 大型セール中、ダウンストリームのBigQueryログ書き込みが平常時200msからセール中最大3秒まで遅延した。Pub/Subのack期限はデフォルトの10秒のままだったため、処理が終わる前にack期限が切れてPub/Sub側が「未処理」と判断し同じメッセージを再送した。pushサブスクリプションはat-least-once配信であるにもかかわらず受信側に冪等性の仕組みが一切なく、同一の発行リクエストに対してクーポンが二重発行される事象が発生し、約1,200人の顧客に重複クーポンが付与された（想定損失 約360万円）。

問題点を全て洗い出し、Pub/Sub `enable_exactly_once_delivery` × アプリ側冪等性チェック（Firestoreトランザクション） × `ack_deadline_seconds`適正化 × Dead Letter Topic(DLQ) × メッセージ順序保証（ordering key） × サブスクリプションfilter × リトライポリシー（バックオフ） × バックログ監視アラートを考慮したBad→Goodリファクタリングを行ってください。

制約・前提条件

• GCP Pub/Sub、Terraform 1.8+（googleプロバイダー5.x）、リージョンasia-northeast1、受信側はCloud Run v2（pushサブスクリプション経由）
• campaign-eventsトピックには、クーポン発行・取り消し以外にも在庫連携・A/Bテスト割当などの無関係なイベントタイプが同じトピックに混在してpublishされている
• ダウンストリームのBigQueryログ書き込みは平常時200ms、セール中ピークで最大3秒（この前提でack期限を設計すること）
• coupon.issue_requestedとcoupon.revoke_requestedは同一order_idに対して発生し得る（例: クーポン発行直後に顧客が注文をキャンセルした場合）。この2つのイベントは同一注文内で発生順に処理される必要がある
• 新しいキャンペーンタイプ追加時にバリデーションが漏れた不正payloadが将来的にも発生し得る前提で設計すること（今回は実際に発生し、無限リトライでCloud Run同時実行枠の約40%を2時間占有した）
• 監視基盤: Cloud Monitoring + DataDog Agent v7 併用

悪い構成 (Before)

📋 コピー# coupon-issuance-service/app/handler.py
from fastapi import FastAPI, Request

app = FastAPI()

@app.post("/pubsub/push")
async def handle_push(request: Request):
    envelope = await request.json()
    message = envelope["message"]
    data = json.loads(base64.b64decode(message["data"]))

    # 問題①: event_idによる冪等性チェックが一切ない
    # 問題⑤: coupon以外のイベントも全部届き、if文でフィルタしている
    event_type = data.get("event_type")
    if event_type == "coupon.issue_requested":
        await issue_coupon(data["order_id"], data["customer_id"], data["discount_rate"])
    elif event_type == "coupon.revoke_requested":
        await revoke_coupon(data["order_id"])
    else:
        pass  # 在庫連携・A/Bテストイベント等は無視するだけ

    return {"status": "ok"}


async def issue_coupon(order_id, customer_id, discount_rate):
    # 問題①: 同じorder_idで再度呼ばれても無条件で新規発行してしまう
    coupon = Coupon(order_id=order_id, customer_id=customer_id, discount_rate=discount_rate)
    await firestore_client.collection("coupons").add(coupon.to_dict())
    await bq_client.insert_rows_json("coupon_issuance_log", [coupon.to_dict()])  # 最大3秒


# terraform/pubsub.tf（抜粋）
resource "google_pubsub_topic" "campaign_events" {
  name = "campaign-events"
  # 問題④: message_orderingが未設定、順序が保証されない
}

resource "google_pubsub_subscription" "coupon_issuance_push" {
  name  = "coupon-issuance-push-sub"
  topic = google_pubsub_topic.campaign_events.id
  push_config {
    push_endpoint = "https://coupon-issuance-service-xxxx.a.run.app/pubsub/push"
  }
  # 問題②: ack_deadline_secondsが未設定(デフォルト10秒のまま)
  # 問題③: dead_letter_policyが未設定。不正payloadが無限に再送され続ける
  # 問題⑤: filterが未設定。coupon関連以外の全イベントを受信してしまう
  # 問題⑥: retry_policyが未設定(デフォルトの即時・最小バックオフ)
}
# 問題⑦: バックログ・DLQ滞留を監視するアラートが存在しない

ヒント（段階的開示）

# Pub/Sub: 順序保証 × DLQ × filter × retry policyのスケルトン
resource "google_pubsub_subscription" "coupon_issuance_push" {
  name                     = "coupon-issuance-push-sub"
  topic                    = google_pubsub_topic.campaign_events.id
  ack_deadline_seconds     = 60
  enable_message_ordering  = true
  filter                   = "attributes.event_type = \"coupon.issue_requested\" OR attributes.event_type = \"coupon.revoke_requested\""

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

# アプリ側冪等性チェックのスケルトン
async def issue_coupon_idempotent(event_id, order_id, customer_id, discount_rate):
    doc_ref = firestore_client.collection("coupon_issuance_log").document(event_id)

    @firestore.transactional
    def _tx(transaction):
        snapshot = doc_ref.get(transaction=transaction)
        if snapshot.exists:
            return False  # 既に処理済み = 再送とみなしスキップ
        transaction.set(doc_ref, {"order_id": order_id, "processed_at": ...})
        return True

    if _tx(firestore_client.transaction()):
        ...  # 初回のみクーポンを実際に発行する

問題点分析（7点）

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

模範解答

📋 コピー# terraform/pubsub.tf
resource "google_pubsub_topic" "campaign_events" {
  name = "campaign-events"
}

# 修正③: DLQ専用トピック。不正payloadをここへ隔離し正常メッセージの処理能力を守る
resource "google_pubsub_topic" "coupon_events_dlq" {
  name = "coupon-events-dlq"
}

resource "google_pubsub_subscription" "coupon_issuance_push" {
  name  = "coupon-issuance-push-sub"
  topic = google_pubsub_topic.campaign_events.id

  # 修正②: ダウンストリーム実測(セール中最大3秒)に安全マージンを載せて60秒に設定
  ack_deadline_seconds = 60

  # 修正④: order_id単位での順序保証を有効化(Publisher側もordering_key=order_idで送信)
  enable_message_ordering = true

  # 修正⑤: coupon関連イベントのみを配信し、無関係なイベントによる無駄な起動を防ぐ
  filter = "attributes.event_type = \"coupon.issue_requested\" OR attributes.event_type = \"coupon.revoke_requested\""

  # 修正③: 5回失敗したメッセージはDLQへ隔離。無限リトライによる同時実行枠の占有を防ぐ
  dead_letter_policy {
    dead_letter_topic     = google_pubsub_topic.coupon_events_dlq.id
    max_delivery_attempts = 5
  }

  # 修正⑥: ダウンストリーム障害時の再送を指数バックオフで分散し、サンダリングハードを防ぐ
  retry_policy {
    minimum_backoff = "10s"
    maximum_backoff = "600s"
  }

  push_config {
    push_endpoint = "https://coupon-issuance-service-xxxx.a.run.app/pubsub/push"
    oidc_token {
      service_account_email = google_service_account.coupon_pubsub_invoker.email
    }
  }
}

# DLQへのpublish権限をPub/Subサービスエージェントに付与(dead_letter_policy利用の前提条件)
resource "google_pubsub_topic_iam_member" "dlq_publisher" {
  topic  = google_pubsub_topic.coupon_events_dlq.id
  role   = "roles/pubsub.publisher"
  member = "serviceAccount:service-${data.google_project.current.number}@gcp-sa-pubsub.iam.gserviceaccount.com"
}

📋 コピー"""coupon-issuance-service/app/handler.py

Pub/Sub は at-least-once 配信のため、同一メッセージが複数回届くことを
前提に、受信側で冪等性を担保する(修正①)。
"""
from __future__ import annotations

import base64
import json
import logging
from typing import Final

from fastapi import FastAPI, Request
from google.cloud import firestore

logger = logging.getLogger(__name__)
app = FastAPI()

_ISSUANCE_LOG_COLLECTION: Final[str] = "coupon_issuance_log"  # event_idをdocument IDにする冪等性台帳


@app.post("/pubsub/push")
async def handle_push(request: Request) -> dict:
    """Pub/Sub push サブスクリプションからのリクエストを処理する。

    Args:
        request: Pub/Sub push エンベロープを含むHTTPリクエスト。

    Returns:
        処理結果を示すdict。200を返すとPub/Subはメッセージをackしたとみなす。
    """
    envelope = await request.json()
    message = envelope["message"]
    event_id: str = message["messageId"]  # 修正①: Pub/Subが払い出す一意なmessageIdを冪等性キーに使う
    data = json.loads(base64.b64decode(message["data"]))

    event_type = data["event_type"]  # filter設定済みなのでcoupon関連イベントのみ届く(修正⑤)
    if event_type == "coupon.issue_requested":
        await issue_coupon_idempotent(
            event_id=event_id,
            order_id=data["order_id"],
            customer_id=data["customer_id"],
            discount_rate=data["discount_rate"],
        )
    elif event_type == "coupon.revoke_requested":
        await revoke_coupon_idempotent(event_id=event_id, order_id=data["order_id"])

    return {"status": "ok"}


async def issue_coupon_idempotent(
    event_id: str, order_id: str, customer_id: str, discount_rate: float
) -> None:
    """クーポンを冪等に発行する。同一event_idの再送では何もしない。

    Args:
        event_id: Pub/Subメッセージの一意ID(冪等性キー)。
        order_id: 対象注文ID。
        customer_id: 対象顧客ID。
        discount_rate: 割引率。

    Note:
        Firestoreトランザクションで「未処理なら記録して発行」を1操作にまとめ、
        再送によるレースコンディションでも二重発行が起きないようにする(修正①)。
    """
    doc_ref = firestore_client.collection(_ISSUANCE_LOG_COLLECTION).document(event_id)

    @firestore.transactional
    def _issue_if_new(transaction: firestore.Transaction) -> bool:
        snapshot = doc_ref.get(transaction=transaction)
        if snapshot.exists:
            return False  # 既に処理済み = Pub/Subによる再送とみなしスキップ
        transaction.set(doc_ref, {"order_id": order_id, "event_type": "issue", "processed_at": firestore.SERVER_TIMESTAMP})
        return True

    is_new = _issue_if_new(firestore_client.transaction())
    if not is_new:
        logger.info("coupon_issuance_skipped_duplicate", extra={"event_id": event_id, "order_id": order_id})
        return

    coupon = Coupon(order_id=order_id, customer_id=customer_id, discount_rate=discount_rate)
    await firestore_client.collection("coupons").add(coupon.to_dict())
    await bq_client.insert_rows_json("coupon_issuance_log", [coupon.to_dict()])


async def revoke_coupon_idempotent(event_id: str, order_id: str) -> None:
    """クーポンを冪等に失効させる。issue/revokeの処理順序はordering keyで保証される(修正④)。"""
    doc_ref = firestore_client.collection(_ISSUANCE_LOG_COLLECTION).document(event_id)

    @firestore.transactional
    def _revoke_if_new(transaction: firestore.Transaction) -> bool:
        snapshot = doc_ref.get(transaction=transaction)
        if snapshot.exists:
            return False
        transaction.set(doc_ref, {"order_id": order_id, "event_type": "revoke", "processed_at": firestore.SERVER_TIMESTAMP})
        return True

    if _revoke_if_new(firestore_client.transaction()):
        await mark_coupons_revoked(order_id)

📋 コピーresource "google_monitoring_alert_policy" "coupon_subscription_backlog" {
  display_name = "coupon-issuance-push-sub バックログ滞留"
  conditions {
    display_name = "oldest_unacked_message_age > 300s"
    condition_threshold {
      filter = <<-EOT
        resource.type="pubsub_subscription"
        AND resource.label.subscription_id="coupon-issuance-push-sub"
        AND metric.type="pubsub.googleapis.com/subscription/oldest_unacked_message_age"
      EOT
      comparison      = "COMPARISON_GT"
      threshold_value = 300
      duration        = "60s"
    }
  }
  combiner               = "OR"
  notification_channels  = [google_monitoring_notification_channel.slack.id]
}

resource "google_monitoring_alert_policy" "coupon_dlq_messages" {
  display_name = "coupon-events-dlq メッセージ滞留"
  conditions {
    display_name = "dead_letter_message_count > 0"
    condition_threshold {
      filter = <<-EOT
        resource.type="pubsub_subscription"
        AND resource.label.subscription_id="coupon-issuance-push-sub"
        AND metric.type="pubsub.googleapis.com/subscription/dead_letter_message_count"
      EOT
      comparison      = "COMPARISON_GT"
      threshold_value = 0
      duration        = "0s"
    }
  }
  combiner               = "OR"
  notification_channels  = [google_monitoring_notification_channel.pagerduty.id]
}

ポイント解説

実務への応用

• MOps販促システムチームでは、Argo Workflowsのバッチ完了通知やA/Bテスト割当イベントなど、Pub/Subを介した非同期メッセージングが今後さらに増える見込みである: 今回設計した「event_idベースの冪等性台帳(Firestoreトランザクション) × DLQ × ordering key × filter」というパターンは、決済確定イベントや在庫引当イベントのような「重複実行が金銭的損失に直結する」処理全般に横展開できる
• 金銭が絡む非同期処理では、Pub/Sub側の配信保証機能を過信せず、必ずアプリ側の冪等性チェックとセットで設計する習慣を持つことが重要である

今日のまとめ

次のステップ

• 発展問題: DLQへ滞留したメッセージを人手で確認した後、修正済みのpayloadとして元のトピックへ再publishする「再処理(replay)」の運用フローと、そのためのCloud Run管理エンドポイント（認可・監査ログ含む）を設計せよ
• 参考: Cloud Pub/Sub Dead Letter Topics / Ordering Keys / Exactly-once Delivery（公式ドキュメント）、Firestore Transactions、Cloud Monitoring pubsub.googleapis.com/subscription/* メトリクス

自己評価（あとで記入）