A+B 複合 — 意味不明な命名 × ネスト × データクラス問題（クーポン適用ロジック Bad→Good Ch1-2）× coupon-service Cloud SQL コネクション枯渇対策（PgBouncer transaction pooling × リードレプリカ読み書き分離 × Workload Identity Federation × statement_timeout × 原子的なUPDATE...RETURNING × コネクション監視アラート）

📋 コピー# tests/test_coupon.py
import pytest
from datetime import datetime
from decimal import Decimal
from coupon import Coupon, DiscountType, apply_coupons


class TestCouponIsApplicable:
    def test_expired_coupon_is_not_applicable(self):
        coupon = Coupon(
            code="EXPIRED", discount_type=DiscountType.FIXED, value=Decimal("100"),
            expires_at=datetime(2026, 1, 1),
        )
        assert coupon.is_applicable(Decimal("5000"), datetime(2026, 8, 15)) is False

    def test_below_minimum_purchase_is_not_applicable(self):
        coupon = Coupon(
            code="MIN3000", discount_type=DiscountType.FIXED, value=Decimal("100"),
            minimum_purchase_amount=Decimal("3000"),
        )
        assert coupon.is_applicable(Decimal("2000"), datetime(2026, 8, 15)) is False


class TestCouponCalculateDiscount:
    def test_rate_discount_is_clipped_by_maximum(self):
        coupon = Coupon(
            code="RATE10", discount_type=DiscountType.RATE, value=Decimal("10"),
            maximum_discount_amount=Decimal("500"),
        )
        assert coupon.calculate_discount(Decimal("10000")) == Decimal("500")


class TestApplyCoupons:
    def test_non_stackable_coupon_applied_once_only(self):
        now = datetime(2026, 8, 15)
        coupons = [
            Coupon(code="A", discount_type=DiscountType.FIXED, value=Decimal("100"), is_stackable=False),
            Coupon(code="B", discount_type=DiscountType.FIXED, value=Decimal("200"), is_stackable=False),
        ]
        assert apply_coupons(Decimal("5000"), coupons, now) == Decimal("100")

    def test_remaining_uses_is_decremented(self):
        now = datetime(2026, 8, 15)
        coupon = Coupon(code="C", discount_type=DiscountType.FIXED, value=Decimal("100"), remaining_uses=1)
        apply_coupons(Decimal("5000"), [coupon], now)
        assert coupon.remaining_uses == 0

📋 コピーapiVersion: apps/v1
kind: Deployment
metadata:
  name: coupon-service
  namespace: mops
spec:
  replicas: 2  # HPAで最大30まで
  selector:
    matchLabels: { app: coupon-service }
  template:
    metadata:
      labels: { app: coupon-service }
    spec:
      serviceAccountName: coupon-service  # 修正3: Workload Identity Federation
      containers:
        - name: app
          image: asia-northeast1-docker.pkg.dev/PROJECT_ID/mops/coupon-service@sha256:abc123
          env:
            - name: DB_WRITE_DSN
              value: "postgresql://coupon_app@127.0.0.1:6432/coupon"  # 修正1/2: PgBouncer経由(primary=書き込み)
            - name: DB_READ_DSN
              value: "postgresql://coupon_app@127.0.0.1:6433/coupon"  # 修正2: PgBouncer経由(replica=読み取り)
          resources:
            requests: { cpu: "250m", memory: "512Mi" }
            limits:   { cpu: "250m", memory: "512Mi" }

        # 修正1: primary用PgBouncer(transaction pooling)
        - name: pgbouncer-primary
          image: edoburu/pgbouncer:1.22.1
          env:
            - name: DATABASE_URL
              value: "postgresql://coupon_app@127.0.0.1:5432/coupon"  # cloud-sql-proxy(primary)経由
            - name: POOL_MODE
              value: "transaction"  # トランザクション単位でコネクションを使い回す
            - name: MAX_CLIENT_CONN
              value: "1000"
            - name: DEFAULT_POOL_SIZE
              value: "5"  # 30レプリカ×5=150 < max_connections 200(replicaと合わせても余裕を残す)
          ports: [{ containerPort: 6432 }]

        # 修正2: replica用PgBouncer(読み取り専用トラフィック)
        - name: pgbouncer-replica
          image: edoburu/pgbouncer:1.22.1
          env:
            - name: DATABASE_URL
              value: "postgresql://coupon_app@127.0.0.1:5433/coupon"
            - name: POOL_MODE
              value: "transaction"
            - name: DEFAULT_POOL_SIZE
              value: "5"
          ports: [{ containerPort: 6433 }]

        # 修正3: Cloud SQL Auth Proxy(primary) — Workload IdentityによるIAM認証、SAキー不要
        - name: cloud-sql-proxy-primary
          image: gcr.io/cloud-sql-connectors/cloud-sql-proxy:2.14.0
          args:
            - "--structured-logs"
            - "--port=5432"
            - "PROJECT_ID:asia-northeast1:coupon-db-primary"
          resources:
            requests: { cpu: "100m", memory: "128Mi" }

        # 修正3: Cloud SQL Auth Proxy(replica)
        - name: cloud-sql-proxy-replica
          image: gcr.io/cloud-sql-connectors/cloud-sql-proxy:2.14.0
          args:
            - "--structured-logs"
            - "--port=5433"
            - "PROJECT_ID:asia-northeast1:coupon-db-replica"
          resources:
            requests: { cpu: "100m", memory: "128Mi" }

📋 コピー"""coupon_repository.py — 読み書き分離(修正2) × statement_timeout(修正4) × 原子的な残数減算(修正5)"""
from __future__ import annotations

from dataclasses import dataclass
from typing import Final

import psycopg

# 名前付き定数: 遅いクエリでコネクションを握り続けさせない上限(修正4)
_STATEMENT_TIMEOUT_MS: Final[int] = 3000


@dataclass(frozen=True, slots=True)
class CouponRecord:
    code: str
    remaining_uses: int | None


class CouponRepository:
    """クーポンの検索(読み取りレプリカ)と残数消費(primary)を分離するリポジトリ。

    Args:
        write_dsn: 書き込み用(primary経由PgBouncer)のDSN。
        read_dsn: 読み取り用(replica経由PgBouncer)のDSN。
    """

    def __init__(self, write_dsn: str, read_dsn: str) -> None:
        self._write_dsn = write_dsn
        self._read_dsn = read_dsn

    def find_by_code(self, code: str) -> CouponRecord | None:
        """クーポンコードで検索する(修正2: 読み取りレプリカへルーティング)。"""
        with psycopg.connect(
            self._read_dsn, options=f"-c statement_timeout={_STATEMENT_TIMEOUT_MS}"  # 修正4
        ) as conn:
            row = conn.execute(
                "SELECT code, remaining_uses FROM coupons WHERE code = %s",
                (code,),
            ).fetchone()
            return CouponRecord(code=row[0], remaining_uses=row[1]) if row else None

    def consume_one_use(self, code: str) -> bool:
        """残数を原子的に1減算する(修正5: SELECT→アプリ判定→UPDATEの2段階操作をやめ、
        1つのUPDATE文でLost Updateを防ぐ)。

        Args:
            code: 消費対象のクーポンコード。

        Returns:
            減算に成功した場合True。残数が既に0で減算できなかった場合False。
        """
        with psycopg.connect(
            self._write_dsn, options=f"-c statement_timeout={_STATEMENT_TIMEOUT_MS}"  # 修正4
        ) as conn:
            row = conn.execute(
                """
                UPDATE coupons
                SET remaining_uses = remaining_uses - 1
                WHERE code = %s AND remaining_uses > 0
                RETURNING remaining_uses
                """,
                (code,),
            ).fetchone()
            conn.commit()
            return row is not None

Bad vs Good 設計比較

確認コマンド

📋 コピー# 1. PgBouncerがtransaction poolingモードで起動していることを確認
kubectl exec -n mops deploy/coupon-service -c pgbouncer-primary -- cat /etc/pgbouncer/pgbouncer.ini | grep pool_mode
# Expected: pool_mode = transaction

# 2. Cloud SQL replicaへの接続確認(recovery mode=trueならレプリカ)
kubectl exec -n mops deploy/coupon-service -c app -- psql "$DB_READ_DSN" -c "SELECT pg_is_in_recovery();"
# Expected: t

# 3. アクティブコネクション数の監視値を確認
gcloud monitoring time-series list \
  --filter='metric.type="cloudsql.googleapis.com/database/postgresql/num_backends"' \
  --format="table(points[0].value.int64Value)"
# Expected: max_connections(200)を大きく下回る値で安定

# 4. 残数消費の競合安全性を確認(remaining_uses=1のクーポンに同時2リクエスト)
for i in 1 2; do
  kubectl exec -n mops deploy/coupon-service -c app -- \
    python -c "from coupon_repository import CouponRepository; \
    repo = CouponRepository('$DB_WRITE_DSN', '$DB_READ_DSN'); \
    print(repo.consume_one_use('LOWSTOCK1'))" &
done
wait
# Expected: 片方はTrue、もう片方はFalse(remaining_usesがマイナスにならない)

# 5. Secret ManagerからDB認証情報が取得できていることを確認(K8s Secretに平文パスワードがないこと)
kubectl get secret -n mops coupon-db-credentials 2>&1
# Expected: Error from server (NotFound) — Secret Manager + Workload Identityへ完全移行済み