A+B 複合 — 仕様化テストで守りながら Pub/Sub ワーカーを冪等にし、GKE の Pod 終了で二重発行しない（第15章 リファクタリング × PID 1 と SIGTERM × at-least-once）

📋 コピー"""本番の実装（Cloud SQL for PostgreSQL・push API・Pub/Sub）と起動口。"""
from __future__ import annotations

import logging
import os

import psycopg
import requests
from psycopg_pool import ConnectionPool

from coupon_issuer import (CALLBACK_THREADS, DB_POOL_TIMEOUT_S, DB_STATEMENT_TIMEOUT_S,
                           MAX_OUTSTANDING_MESSAGES, PUSH_CONNECT_TIMEOUT_S, PUSH_READ_TIMEOUT_S,
                           CouponDecision, Dependencies, IssuedCoupon, MemberRank, TransientError, run)

_INSERT_ONCE = """
INSERT INTO issued_coupons (member_id, campaign_id, amount)
VALUES (%(member_id)s, %(campaign_id)s, %(amount)s)
ON CONFLICT (member_id, campaign_id) WHERE voided_at IS NULL DO NOTHING
RETURNING coupon_id, amount, notified_at IS NOT NULL
"""
# INSERT と SELECT を1文（CTE）にしない: 同時発行で相手が後からコミットした行は、
# 同じ文の snapshot から見えず0行になる（実測: Postgres 17.11 で [] が返った）
_SELECT_EXISTING = """
SELECT coupon_id, amount, notified_at IS NOT NULL FROM issued_coupons
WHERE member_id = %(member_id)s AND campaign_id = %(campaign_id)s AND voided_at IS NULL
"""


def make_pool(dsn: str) -> ConnectionPool:
    """コールバックのスレッド数と同じ本数の接続。1本を共有しない（トランザクションが混ざる）。"""
    return ConnectionPool(
        dsn, min_size=2, max_size=CALLBACK_THREADS, timeout=DB_POOL_TIMEOUT_S,
        kwargs={"options": f"-c statement_timeout={int(DB_STATEMENT_TIMEOUT_S * 1000)}"},
    )


class PgMemberRepository:
    def __init__(self, pool: ConnectionPool) -> None:
        self._pool = pool

    def find_rank(self, member_id: int) -> MemberRank | None:
        try:
            with self._pool.connection() as conn:   # 抜けるとき commit / 例外なら rollback
                row = conn.execute("SELECT rank FROM members WHERE id = %s", (member_id,)).fetchone()
        except (psycopg.OperationalError, psycopg.errors.QueryCanceled) as e:
            raise TransientError(str(e)) from e
        if row is None:
            return None
        return MemberRank(row[0]) if row[0] in MemberRank else MemberRank.REGULAR


class PgCouponRepository:
    def __init__(self, pool: ConnectionPool) -> None:
        self._pool = pool

    def issue_once(self, d: CouponDecision) -> IssuedCoupon:
        params = {"member_id": d.member_id, "campaign_id": d.campaign_id, "amount": d.amount_yen}
        try:
            with self._pool.connection() as conn:
                row = conn.execute(_INSERT_ONCE, params).fetchone()
                if row is None:                        # 既に発行済み（再配信・同時発行）
                    row = conn.execute(_SELECT_EXISTING, params).fetchone()
        except (psycopg.OperationalError, psycopg.errors.QueryCanceled) as e:
            raise TransientError(str(e)) from e
        return IssuedCoupon(coupon_id=row[0], amount_yen=row[1], notified=row[2])

    def mark_notified(self, coupon_id: int) -> None:
        try:
            with self._pool.connection() as conn:
                conn.execute("UPDATE issued_coupons SET notified_at = now() "
                             "WHERE coupon_id = %s AND notified_at IS NULL", (coupon_id,))
        except (psycopg.OperationalError, psycopg.errors.QueryCanceled) as e:
            raise TransientError(str(e)) from e


class HttpPushSender:
    def __init__(self, url: str, session: requests.Session | None = None) -> None:
        self._url, self._session = url, session or requests.Session()

    def send(self, member_id: int, title: str, *, idempotency_key: str) -> None:
        try:
            resp = self._session.post(
                self._url, json={"member_id": member_id, "title": title},
                headers={"Idempotency-Key": idempotency_key},
                timeout=(PUSH_CONNECT_TIMEOUT_S, PUSH_READ_TIMEOUT_S),  # 無期限に待たない
            )
        except requests.RequestException as e:
            raise TransientError(str(e)) from e
        if resp.status_code >= 500 or resp.status_code == 429:
            raise TransientError(f"push {resp.status_code}")
        resp.raise_for_status()   # 4xx はバグ → 想定外の例外として nack、5回で dead letter


def main() -> None:
    from google.cloud import pubsub_v1  # 起動時だけ import（テストは Pub/Sub 無しで走る）

    logging.basicConfig(level=logging.INFO)
    pool = make_pool(os.environ["DB_DSN"])
    deps = Dependencies(PgMemberRepository(pool), PgCouponRepository(pool),
                        HttpPushSender(os.environ["PUSH_URL"]))
    code = run(pubsub_v1.SubscriberClient(), os.environ["SUBSCRIPTION"], deps,
               flow_control=pubsub_v1.types.FlowControl(max_messages=MAX_OUTSTANDING_MESSAGES))
    pool.close()
    logging.shutdown()
    os._exit(code)  # 排出が時間切れのとき、実行中スレッドの終了を待たずに抜ける


if __name__ == "__main__":
    main()

📋 コピー"""handle / make_callback の単体テスト。legacy の仕様化テストのうち、意図して変えた振る舞いを明示する。"""
from __future__ import annotations

import json

import pytest

from coupon_issuer import (CALLBACK_WORST_CASE_S, DRAIN_TIMEOUT_S, CouponDecision, Dependencies,
                           IssuedCoupon, MemberRank, Outcome, TransientError, handle, make_callback)


class FakeMembers:
    def __init__(self, ranks: dict[int, MemberRank]) -> None:
        self.ranks = ranks

    def find_rank(self, member_id: int) -> MemberRank | None:
        return self.ranks.get(member_id)


class FakeCoupons:
    """(member_id, campaign_id) 一意の発行台帳。本番は部分ユニーク索引 + ON CONFLICT。"""
    def __init__(self) -> None:
        self.rows: dict[tuple[int, str], IssuedCoupon] = {}

    def issue_once(self, d: CouponDecision) -> IssuedCoupon:
        key = (d.member_id, d.campaign_id)
        if key not in self.rows:
            self.rows[key] = IssuedCoupon(len(self.rows) + 1, d.amount_yen, notified=False)
        return self.rows[key]

    def mark_notified(self, coupon_id: int) -> None:
        for k, c in self.rows.items():
            if c.coupon_id == coupon_id:
                self.rows[k] = IssuedCoupon(c.coupon_id, c.amount_yen, notified=True)


class FakePush:
    def __init__(self, fail_times: int = 0) -> None:
        self.sent: list[str] = []
        self.fail_times = fail_times

    def send(self, member_id: int, title: str, *, idempotency_key: str) -> None:
        if self.fail_times:
            self.fail_times -= 1
            raise TransientError("push 503")
        self.sent.append(idempotency_key)


def deps(push: FakePush | None = None) -> Dependencies:
    return Dependencies(FakeMembers({42: MemberRank.GOLD, 7: MemberRank.SILVER}), FakeCoupons(), push or FakePush())


def ev(member_id=42, month="2026-10") -> bytes:
    return json.dumps({"member_id": member_id, "campaign_month": month}).encode()


def test_amount_table_unchanged_from_legacy():  # 仕様化テスト test_amount_by_rank と同じ表
    d = deps()
    handle(ev(42), d)
    handle(ev(7), d)
    assert [c.amount_yen for c in d.coupons.rows.values()] == [1_000, 500]


def test_campaign_month_comes_from_event_not_clock():  # 意図した変更 1（legacy: 処理時刻の UTC 月）
    d = deps()
    handle(ev(month="2026-10"), d)
    assert (42, "birthday-2026-10") in d.coupons.rows


def test_redelivery_issues_once_and_notifies_once():  # 意図した変更 2（legacy: 2回発行・2回通知）
    d = deps()
    assert handle(ev(), d) is Outcome.ISSUED
    assert handle(ev(), d) is Outcome.ALREADY_DONE
    assert len(d.coupons.rows) == 1 and d.push.sent == ["coupon-1"]


def test_push_failure_retries_without_second_coupon():  # 意図した変更 3（legacy: 5xx を成功扱いで ack）
    d = deps(FakePush(fail_times=1))
    assert handle(ev(), d) is Outcome.RETRY
    assert handle(ev(), d) is Outcome.ISSUED        # 再配信で通知だけやり直す
    assert len(d.coupons.rows) == 1 and d.push.sent == ["coupon-1"]


@pytest.mark.parametrize("raw", [b"{not json", b'{"member_id": 42}', b'{"member_id": -1, "campaign_month": "2026-10"}'])
def test_invalid_event_is_rejected(raw):
    assert handle(raw, deps()) is Outcome.REJECT


def test_unknown_member_is_rejected():
    assert handle(ev(member_id=404), deps()) is Outcome.REJECT


class Msg:
    def __init__(self, data: bytes) -> None:
        self.data, self.calls = data, []

    def ack(self) -> None:
        self.calls.append("ack")

    def nack(self) -> None:
        self.calls.append("nack")


@pytest.mark.parametrize(("raw", "expected"), [(ev(), ["ack"]), (b"{not json", ["nack"])])
def test_callback_maps_outcome_to_ack_or_nack(raw, expected):  # 意図した変更 4（legacy: 失敗も ack）
    m = Msg(raw)
    make_callback(deps())(m)
    assert m.calls == expected


def test_unexpected_exception_is_nacked():
    class Boom(FakeMembers):
        def find_rank(self, member_id):
            raise KeyError("bug")
    m = Msg(ev())
    make_callback(Dependencies(Boom({}), FakeCoupons(), FakePush()))(m)
    assert m.calls == ["nack"]


def test_drain_budget_fits_in_grace_period():
    TERMINATION_GRACE_PERIOD_S = 30  # deploy/coupon-issuer.yaml と同じ値
    assert CALLBACK_WORST_CASE_S == 18.0
    assert CALLBACK_WORST_CASE_S < DRAIN_TIMEOUT_S < TERMINATION_GRACE_PERIOD_S

📋 コピーapiVersion: apps/v1
kind: Deployment
metadata: { name: coupon-issuer, namespace: mops-promo }
spec:
  replicas: 4
  strategy:
    type: RollingUpdate
    rollingUpdate: { maxSurge: 1, maxUnavailable: 0 }
  selector: { matchLabels: { app: coupon-issuer } }
  template:
    metadata: { labels: { app: coupon-issuer } }
    spec:
      serviceAccountName: coupon-issuer          # Workload Identity で Pub/Sub・Cloud SQL へ
      # 排出予算 20 秒（コールバック最悪 18 秒 + ストリーム停止 2 秒）+ 余裕 10 秒。
      # 既定値と同じ 30 秒だが、根拠を coupon_issuer.py の定数と対にして明示する
      terminationGracePeriodSeconds: 30
      containers:
        - name: worker
          image: asia-northeast1-docker.pkg.dev/mops-prod/promo/coupon-issuer:1.4.0
          # exec 形式: python が PID 1 になり、SIGTERM を自分のハンドラで直接受ける
          command: ["python", "-m", "adapters"]
          env:
            - { name: SUBSCRIPTION, value: projects/mops-prod/subscriptions/birthday-coupon-issuer }
          resources:
            requests: { cpu: 250m, memory: 256Mi }
            limits:   { memory: 256Mi }
          # preStop は付けない: pull 型で Service の endpoints が無く、外すべき受信経路が無い
---
apiVersion: policy/v1
kind: PodDisruptionBudget
metadata: { name: coupon-issuer, namespace: mops-promo }
spec:
  maxUnavailable: 1                              # ノードアップグレードで4台同時に排出しない
  selector: { matchLabels: { app: coupon-issuer } }
---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata: { name: coupon-issuer, namespace: mops-promo }
spec:
  scaleTargetRef: { apiVersion: apps/v1, kind: Deployment, name: coupon-issuer }
  minReplicas: 2
  maxReplicas: 12
  metrics:
    - type: Resource
      resource: { name: cpu, target: { type: Utilization, averageUtilization: 60 } }
  behavior:
    scaleDown:
      stabilizationWindowSeconds: 600            # 既定 300 秒。縮小131回/月の揺れを減らす
      policies: [{ type: Pods, value: 1, periodSeconds: 120 }]

📋 コピーresource "google_pubsub_subscription" "birthday_coupon_issuer" {
  name                 = "birthday-coupon-issuer"
  topic                = google_pubsub_topic.member_birthday.id
  ack_deadline_seconds = 30  # 初期値。実際の延長はクライアントのリース管理が行う

  retry_policy {             # 無いと nack は即時再配信（公式: 既定は immediate redelivery）
    minimum_backoff = "10s"
    maximum_backoff = "600s"
  }
  dead_letter_policy {
    dead_letter_topic     = google_pubsub_topic.birthday_coupon_dlq.id
    max_delivery_attempts = 5  # 5〜100。REJECT も5回で DLQ に落ちる
  }
}

# Pub/Sub のサービスエージェントに、DLQ への publish と元サブスクリプションの subscribe を許可する
locals { pubsub_agent = "serviceAccount:service-${data.google_project.this.number}@gcp-sa-pubsub.iam.gserviceaccount.com" }

resource "google_pubsub_topic_iam_member" "dlq_publisher" {
  topic  = google_pubsub_topic.birthday_coupon_dlq.id
  role   = "roles/pubsub.publisher"
  member = local.pubsub_agent
}
resource "google_pubsub_subscription_iam_member" "dlq_subscriber" {
  subscription = google_pubsub_subscription.birthday_coupon_issuer.id
  role         = "roles/pubsub.subscriber"
  member       = local.pubsub_agent
}