弱点補強 — 「読んで・判定して・書く」クーポン発行を、分離レベルに頼らず守る（lost update × write skew × REPEATABLE READ / SERIALIZABLE の実測）

概要

問題

MOps 販促システムチームの coupon-api は、アプリと LINE ミニアプリの「クーポンを受け取る」ボタンから呼ばれる発行 API。シルバーウィーク（2026-09-19〜09-23 の5連休）に 1,000 円クーポンを3本（SW-A 早割・SW-B 家電・SW-C 日用品）配った。

連休明けの棚卸し（本問の前提データ・架空）

二重タップとアプリの3秒タイムアウトによる再送で、同じ会員から同じキャンペーンへの要求が数十ミリ秒以内に2本届くのが普通だった。

設問

1. Bad の問題点を7つ以上。どの決まりを、どの異常（lost update／write skew／phantom）で破るか
2. 後輩A・Bの提案で、何が直り・何が直らず・何が新たに壊れるか。測って確かめる方法
3. READ COMMITTED のまま R1〜R3 を守る表定義とコード（docstring・コメント・名前付き定数）
4. Bad に流すと赤になる同時実行テスト（実 DB）
5. 既に壊れたデータ（予算超過・重複・カウンタのずれ）を抱えたままの移行手順
6. 直した後に見る指標と、別方式に切り替える条件

制約: PostgreSQL 14+（本番 Cloud SQL 15）、psycopg 3、Redis などの別ストアは増やさない、通知の outbox は 9/26 の続きとして使ってよい、DB 待ちはアプリの3秒タイムアウトの内側。

悪いコード (Before)

📋 コピーMONTHLY_CAP_YEN = 2_000

def issue_coupon(conn, member_id, campaign_id, month) -> str:
    for _ in range(3):
        try:
            with conn.transaction():
                discount, budget, issued = conn.execute(
                    "SELECT discount_yen, budget_yen, issued_yen FROM campaigns ...").fetchone()
                if issued + discount > budget:
                    return "BUDGET_EXCEEDED"
                if conn.execute("SELECT 1 FROM issued_coupons WHERE campaign_id=%s AND member_id=%s",
                                (campaign_id, member_id)).fetchone():
                    return "ALREADY_ISSUED"
                rows = conn.execute(
                    "SELECT amount_yen FROM issued_coupons WHERE member_id=%s AND month=%s FOR UPDATE",
                    (member_id, month)).fetchall()
                if sum(r[0] for r in rows) + discount > MONTHLY_CAP_YEN:
                    return "MONTHLY_CAP"
                coupon_id = str(uuid.uuid4())
                conn.execute("INSERT INTO issued_coupons VALUES (%s,%s,%s,%s,%s)", ...)
                notify_push(member_id, coupon_id)   # 平均 20 ms、タイムアウト未設定
                conn.execute("UPDATE campaigns SET issued_yen=%s WHERE campaign_id=%s",
                             (issued + discount, campaign_id))
            return "ISSUED"
        except Exception:
            continue
    return "SYSTEM_ERROR"
-- campaigns.issued_yen と SUM(issued_coupons) の突き合わせなし／一意制約なし／テストはモックの逐次実行のみ

ヒント（段階的開示）

with conn.transaction():                       # READ COMMITTED のまま
    # (1) R2: INSERT … ON CONFLICT (campaign_id, member_id) DO NOTHING RETURNING amount_yen
    # (2) R3: INSERT INTO member_monthly_usage … ON CONFLICT DO NOTHING  ← ロックの実体
    #     UPDATE member_monthly_usage SET used_yen = used_yen + :a WHERE … AND used_yen + :a <= 2000
    # (3) R1: UPDATE campaign_budgets SET issued_yen = issued_yen + :a WHERE … <= budget_yen
    #     （いちばん混む行を最後に）
    # (4) INSERT INTO notification_outbox …
# except (SerializationFailure, DeadlockDetected, LockNotAvailable): full jitter

問題点分析（7点）

実測 SELECT SUM(amount_yen) … FOR UPDATE → 0A000 FOR UPDATE is not allowed with aggregate functions。Bad の作者はこれを避けて「行をロックして Python で合計」に逃げたが、#4 が残る。

後輩A・Bの提案（実測）

環境: PostgreSQL 14.11（Homebrew x86_64 ビルドを Rosetta 2 で実行）、Apple M1・メモリ 8GB、Python 3.13.12、psycopg 3.3.6。負荷: 会員 300 × 3キャンペーン × 二重送信 = 1,800 リクエスト、正解は発行ちょうど 600 枚。各条件3回の範囲。絶対値でなく違反件数と相対差を読む。

Bad の行は通知の模擬（20 ms）をトランザクション内に含む。後輩B と Good は通知を outbox に出した後の形。

write skew の図解（Before / 後輩A・B / Good）

上段が write skew。2本は同じ事実（今月 1,000 円）を読み、違う行を書くので、同じ行の競合しか見ない REPEATABLE READ は気づかない。下段は、衝突の相手を「会員×月の1行」に決めて、違う行を書く2本を同じ行でぶつける。

設計: 不変条件ごとに「ロックを取る行」を1つ決める

カウンタ（issued_yen・used_yen）は発行行と同じトランザクションで書くので食い違わない。役割は「数える」より「ロックを取る相手になる」こと。念のため毎日突き合わせる。

模範解答（Bad → Good）

📋 コピーSELECT issued_yen ...           → Python: issued + d > budget ?
UPDATE campaigns SET issued_yen = %s   -- 読んだ値＋d を書き戻す

SELECT 1 FROM issued_coupons WHERE campaign_id, member_id
INSERT INTO issued_coupons ...         -- 一意制約なし

SELECT amount_yen ... FOR UPDATE      -- 0 行ならロックなし
Python: sum(rows) + d > 2000 ?

notify_push(...)                      -- トランザクションの中

except Exception: continue            -- 待たずに3回

📋 コピーINSERT INTO issued_coupons ...
  ON CONFLICT ON CONSTRAINT one_coupon_per_member DO NOTHING
  RETURNING amount_yen                 -- 0 行なら ALREADY_ISSUED

INSERT INTO member_monthly_usage ... ON CONFLICT DO NOTHING
UPDATE member_monthly_usage SET used_yen = used_yen + :a
 WHERE ... AND used_yen + :a <= 2000 RETURNING used_yen

UPDATE campaign_budgets SET issued_yen = issued_yen + :a
 WHERE ... AND issued_yen + :a <= budget_yen RETURNING issued_yen

INSERT INTO notification_outbox ...   -- 送るのは別プロセス

except (40001, 40P01, 55P03): full jitter で最大5回
CHECK (issued_yen <= budget_yen) ＋ 毎日の突き合わせ

全文

📋 コピー"""販促クーポンの発行（予算上限・会員の月間上限・1会員1枚を同時実行下でも守る）。"""
from __future__ import annotations

import random
import re
import time
import uuid
from collections.abc import Callable
from dataclasses import dataclass
from enum import StrEnum

import psycopg
from psycopg import errors

# ── 業務ルール（出どころ: 販促予算の稟議と、粗利を守る会員別上限の社内規程） ──
MONTHLY_CAP_YEN = 2_000

# ── 再試行（再試行してよいのは「DB が安全に巻き戻した」と分かっている失敗だけ） ──
RETRYABLE_ERRORS = (
    errors.SerializationFailure,  # 40001
    errors.DeadlockDetected,      # 40P01
    errors.LockNotAvailable,      # 55P03（lock_timeout 超過）
)
MAX_ATTEMPTS = 5
BACKOFF_BASE_SEC = 0.01
BACKOFF_CAP_SEC = 0.2

# ── 1回の発行で DB を待つ上限（API の 3 秒タイムアウトより内側に収める） ──
LOCK_TIMEOUT = "2s"
STATEMENT_TIMEOUT = "3s"

_MONTH_PATTERN = re.compile(r"^\d{4}-(0[1-9]|1[0-2])$")


class IssueResult(StrEnum):
    """発行 API の結果。呼び出し側はこの値でだけ分岐する。"""

    ISSUED = "ISSUED"
    ALREADY_ISSUED = "ALREADY_ISSUED"    # 同じキャンペーンで発行済み（二重送信・再送を含む）
    MONTHLY_CAP = "MONTHLY_CAP"          # 会員の月間上限に達した
    BUDGET_EXCEEDED = "BUDGET_EXCEEDED"  # キャンペーン予算を使い切った
    RETRY_EXHAUSTED = "RETRY_EXHAUSTED"  # 競合が続いた。クライアントは時間をおいて再送してよい


@dataclass(frozen=True, slots=True)
class IssueRequest:
    """発行要求（完全コンストラクタ: 不正な月は作れない）。"""

    member_id: str
    campaign_id: str
    month: str  # JST の暦月 "YYYY-MM"。業務日付の決め方は呼び出し側の責務

    def __post_init__(self) -> None:
        if not self.member_id or not self.campaign_id:
            raise ValueError("member_id と campaign_id は必須")
        if not _MONTH_PATTERN.match(self.month):
            raise ValueError(f"month は YYYY-MM: {self.month!r}")


class _Rejected(Exception):
    """業務ルールで断る。with ブロックの外へ投げてトランザクションごと巻き戻す。"""

    def __init__(self, result: IssueResult) -> None:
        super().__init__(result)
        self.result = result


def issue_coupon(
    conn: psycopg.Connection,
    req: IssueRequest,
    *,
    sleep: Callable[[float], None] = time.sleep,
    jitter: Callable[[float, float], float] = random.uniform,
) -> IssueResult:
    """クーポンを1枚発行する。READ COMMITTED のまま、判定と更新を1文ずつにまとめる。

    ロックを取る順番は常に「会員×キャンペーンの一意索引 → 会員×月の行 → キャンペーン行」。
    いちばん混むキャンペーン行を最後に取り、保持時間をコミットまでの数ミリ秒に抑える。

    Args:
        conn: autocommit=True の接続（トランザクションはこの関数が張る）。
        req: 発行要求。
        sleep: 再試行の待ち（テストで差し替える）。
        jitter: 待ち時間の乱数（full jitter）。

    Returns:
        IssueResult。ISSUED のときだけ発行行と通知 outbox 行がコミット済み。
    """
    for attempt in range(MAX_ATTEMPTS):
        try:
            with conn.transaction():
                conn.execute(f"SET LOCAL lock_timeout = '{LOCK_TIMEOUT}'")
                conn.execute(f"SET LOCAL statement_timeout = '{STATEMENT_TIMEOUT}'")
                coupon_id = uuid.uuid4()

                # (1) 1会員1枚: 一意制約が判定する。同時に来た2本目は1本目の確定を待ち、
                #     1本目がコミットなら 0 行（発行済み）、巻き戻しなら自分が入る
                row = conn.execute(
                    """INSERT INTO issued_coupons (coupon_id, campaign_id, member_id, month, amount_yen)
                       SELECT %s, campaign_id, %s, %s, discount_yen
                         FROM campaign_budgets WHERE campaign_id = %s
                       ON CONFLICT ON CONSTRAINT one_coupon_per_member DO NOTHING
                       RETURNING amount_yen""",
                    (coupon_id, req.member_id, req.month, req.campaign_id),
                ).fetchone()
                if row is None:
                    raise _Rejected(IssueResult.ALREADY_ISSUED)
                (amount_yen,) = row

                # (2) 会員の月間上限: まだ無い「会員×月」の行を先に作り、ロックできる実体にする
                conn.execute(
                    """INSERT INTO member_monthly_usage (member_id, month) VALUES (%s, %s)
                       ON CONFLICT DO NOTHING""",
                    (req.member_id, req.month),
                )
                # 判定と加算を1文で。先行トランザクションがコミットしたら、
                # PostgreSQL は新しい行で WHERE を評価し直す（READ COMMITTED の再評価）
                if conn.execute(
                    """UPDATE member_monthly_usage SET used_yen = used_yen + %(amount)s
                        WHERE member_id = %(member)s AND month = %(month)s
                          AND used_yen + %(amount)s <= %(cap)s
                       RETURNING used_yen""",
                    {"amount": amount_yen, "member": req.member_id,
                     "month": req.month, "cap": MONTHLY_CAP_YEN},
                ).fetchone() is None:
                    raise _Rejected(IssueResult.MONTHLY_CAP)

                # (3) キャンペーン予算: 同じく条件付き加算。いちばん混む行なので最後に取る
                if conn.execute(
                    """UPDATE campaign_budgets SET issued_yen = issued_yen + %(amount)s
                        WHERE campaign_id = %(campaign)s
                          AND issued_yen + %(amount)s <= budget_yen
                       RETURNING issued_yen""",
                    {"amount": amount_yen, "campaign": req.campaign_id},
                ).fetchone() is None:
                    raise _Rejected(IssueResult.BUDGET_EXCEEDED)

                # (4) 通知は送らずに「送る予定」を同じトランザクションで書く（outbox）
                conn.execute(
                    "INSERT INTO notification_outbox (coupon_id, member_id) VALUES (%s, %s)",
                    (coupon_id, req.member_id),
                )
            return IssueResult.ISSUED
        except _Rejected as rejected:
            return rejected.result
        except RETRYABLE_ERRORS:
            # full jitter: 0〜min(上限, 基準×2^n) の一様乱数。同時に失敗した同士が再衝突しにくい
            sleep(jitter(0, min(BACKOFF_CAP_SEC, BACKOFF_BASE_SEC * 2**attempt)))
    return IssueResult.RETRY_EXHAUSTED


OUTBOX_BATCH = 100


def relay_outbox(conn: psycopg.Connection, send: Callable[[str, str], None]) -> int:
    """未送信の通知を送る（別プロセスで常駐）。少なくとも1回送る。二重は送信先の冪等キーで消す。

    Args:
        conn: autocommit=True の接続。
        send: send(member_id, idempotency_key)。失敗時は例外を投げる。

    Returns:
        送信済みにした件数。
    """
    with conn.transaction():
        rows = conn.execute(
            """SELECT coupon_id, member_id FROM notification_outbox
                WHERE sent_at IS NULL ORDER BY created_at
                LIMIT %s FOR UPDATE SKIP LOCKED""",  # 複数 relay でも同じ行を取り合わない
            (OUTBOX_BATCH,),
        ).fetchall()
        for coupon_id, member_id in rows:
            send(member_id, str(coupon_id))  # Idempotency-Key = coupon_id
            conn.execute("UPDATE notification_outbox SET sent_at = now() WHERE coupon_id = %s",
                         (coupon_id,))
    return len(rows)

📋 コピー-- 不変条件ごとに「ロックを取る行」を1つずつ持つ（衝突の実体化）
CREATE TABLE campaign_budgets (
  campaign_id  text PRIMARY KEY,
  discount_yen integer NOT NULL CHECK (discount_yen > 0),
  budget_yen   integer NOT NULL CHECK (budget_yen > 0),
  issued_yen   integer NOT NULL DEFAULT 0,
  CONSTRAINT issued_within_budget CHECK (issued_yen BETWEEN 0 AND budget_yen)
);
CREATE TABLE member_monthly_usage (
  member_id text NOT NULL,
  month     text NOT NULL CHECK (month ~ '^\d{4}-(0[1-9]|1[0-2])$'),
  used_yen  integer NOT NULL DEFAULT 0 CHECK (used_yen >= 0),
  PRIMARY KEY (member_id, month)
);
CREATE TABLE issued_coupons (
  coupon_id   uuid PRIMARY KEY,
  campaign_id text NOT NULL REFERENCES campaign_budgets,
  member_id   text NOT NULL,
  month       text NOT NULL,
  amount_yen  integer NOT NULL CHECK (amount_yen > 0),
  issued_at   timestamptz NOT NULL DEFAULT now(),
  CONSTRAINT one_coupon_per_member UNIQUE (campaign_id, member_id)
);
CREATE TABLE notification_outbox (
  coupon_id  uuid PRIMARY KEY REFERENCES issued_coupons,
  member_id  text NOT NULL,
  created_at timestamptz NOT NULL DEFAULT now(),
  sent_at    timestamptz
);
CREATE INDEX notification_outbox_unsent ON notification_outbox (created_at) WHERE sent_at IS NULL;

テストと実行例

同時実行テストの要点: ① 実 DB（モックに行ロックも一意索引も無い）② Barrier で全スレッドの接続がそろってから一斉に始める ③ 応答の成功率でなく終わった後の表を SQL で数える。

📋 コピー"""実 PostgreSQL に対するテスト。モックでは同時実行の不具合は再現しない。"""
from __future__ import annotations

import os
import random
import threading
from collections import Counter
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

import psycopg
import pytest

from coupon_issuance import MONTHLY_CAP_YEN, IssueRequest, IssueResult, issue_coupon, relay_outbox

DSN = os.environ.get("TEST_PG_DSN")
pytestmark = pytest.mark.skipif(not DSN, reason="TEST_PG_DSN が無い")
MONTH = "2026-09"
BUDGETS = {"SW-A": 100_000, "SW-B": 300_000, "SW-C": 300_000}


@pytest.fixture
def conn():
    with psycopg.connect(DSN, autocommit=True) as c:
        c.execute("DROP TABLE IF EXISTS notification_outbox, issued_coupons, "
                  "member_monthly_usage, campaign_budgets")
        c.execute(Path(__file__).with_name("schema.sql").read_text())
        for cid, budget in BUDGETS.items():
            c.execute("INSERT INTO campaign_budgets (campaign_id, discount_yen, budget_yen) "
                      "VALUES (%s, 1000, %s)", (cid, budget))
        yield c


def test_sequential_rules(conn):
    req = lambda cid: IssueRequest("m0001", cid, MONTH)  # noqa: E731
    assert issue_coupon(conn, req("SW-A")) is IssueResult.ISSUED
    assert issue_coupon(conn, req("SW-A")) is IssueResult.ALREADY_ISSUED
    assert issue_coupon(conn, req("SW-B")) is IssueResult.ISSUED
    assert issue_coupon(conn, req("SW-C")) is IssueResult.MONTHLY_CAP
    # 断った発行は何も残さない（発行行も outbox 行も、枠の消費も）
    assert conn.execute("SELECT count(*) FROM issued_coupons").fetchone()[0] == 2
    assert conn.execute("SELECT count(*) FROM notification_outbox").fetchone()[0] == 2


def test_budget_exhausted(conn):
    conn.execute("UPDATE campaign_budgets SET issued_yen = budget_yen WHERE campaign_id = 'SW-A'")
    assert issue_coupon(conn, IssueRequest("m0002", "SW-A", MONTH)) is IssueResult.BUDGET_EXCEEDED
    assert conn.execute("SELECT count(*) FROM issued_coupons").fetchone()[0] == 0


def test_invalid_month_is_unrepresentable():
    with pytest.raises(ValueError):
        IssueRequest("m0001", "SW-A", "2026-9")


@pytest.mark.parametrize("workers", [8, 32])
def test_invariants_hold_under_concurrency(conn, workers):
    """会員ごとに 3 キャンペーン × 二重送信の 6 本を同時に投げ、不変条件を数える。"""
    rng = random.Random(workers)
    reqs = []
    for i in range(300):
        burst = [IssueRequest(f"m{i:04d}", cid, MONTH) for cid in BUDGETS for _ in range(2)]
        rng.shuffle(burst)
        reqs.extend(burst)
    start = threading.Barrier(workers)
    local = threading.local()

    def run(req):
        if not hasattr(local, "c"):
            local.c = psycopg.connect(DSN, autocommit=True)
            start.wait()  # 全スレッドの接続がそろってから一斉に始める
        return issue_coupon(local.c, req)

    with ThreadPoolExecutor(workers) as ex:
        outcomes = Counter(ex.map(run, reqs))

    q = lambda sql: conn.execute(sql).fetchall()  # noqa: E731
    # 予算: 実際の発行額 == カウンタ <= 予算
    for cid, budget, counter, actual in q(
        """SELECT b.campaign_id, b.budget_yen, b.issued_yen,
                  COALESCE(SUM(i.amount_yen), 0)
             FROM campaign_budgets b LEFT JOIN issued_coupons i USING (campaign_id)
            GROUP BY 1, 2, 3"""):
        assert actual == counter <= budget, cid
    # 会員の月間上限
    assert q(f"""SELECT member_id FROM issued_coupons GROUP BY member_id
                 HAVING SUM(amount_yen) > {MONTHLY_CAP_YEN}""") == []
    # 通知: 発行した分だけ、ちょうど1行ずつ
    assert q("SELECT count(*) FROM notification_outbox")[0][0] == outcomes[IssueResult.ISSUED]
    # この負荷では全員が上限の2枚まで取れる（予算で断られた人は別キャンペーンで取る）
    assert outcomes[IssueResult.ISSUED] == 600
    assert outcomes[IssueResult.RETRY_EXHAUSTED] == 0


def test_relay_marks_sent_and_uses_coupon_id_as_key(conn):
    issue_coupon(conn, IssueRequest("m0001", "SW-A", MONTH))
    sent = []
    assert relay_outbox(conn, lambda m, key: sent.append((m, key))) == 1
    assert relay_outbox(conn, lambda m, key: sent.append((m, key))) == 0  # 2回目は送らない
    coupon_id = conn.execute("SELECT coupon_id::text FROM issued_coupons").fetchone()[0]
    assert sent == [("m0001", coupon_id)]

実行結果 実測

$ TEST_PG_DSN="host=127.0.0.1 port=55432 user=app dbname=postgres" \
    uv run --no-project --with 'psycopg[binary]' --with pytest python -m pytest -v
test_coupon_issuance.py::test_sequential_rules PASSED                    [ 16%]
test_coupon_issuance.py::test_budget_exhausted PASSED                    [ 33%]
test_coupon_issuance.py::test_invalid_month_is_unrepresentable PASSED    [ 50%]
test_coupon_issuance.py::test_invariants_hold_under_concurrency[8] PASSED [ 66%]
test_coupon_issuance.py::test_invariants_hold_under_concurrency[32] PASSED [ 83%]
test_coupon_issuance.py::test_relay_marks_sent_and_uses_coupon_id_as_key PASSED [100%]
============================== 6 passed in 2.97s ===============================

Bad を同じ負荷（付録の sim.py）に掛けると、R1 の actual == counter <= budget（実額 29.9〜59.8万 対 予算10万）と ISSUED == 600（1,487〜1,791 枚）で落ちる。REPEATABLE READ 版は R3（超過 20〜82 人）で落ちる。ruff check --select E,F,B,UP,SIM は指摘 0 件。

実行例（input → output、SW-A の予算を 2,000 円にした場合）実測

issue_coupon(m0001, SW-A) -> ISSUED
issue_coupon(m0001, SW-A) -> ALREADY_ISSUED      # 二重タップ
issue_coupon(m0001, SW-B) -> ISSUED
issue_coupon(m0001, SW-C) -> MONTHLY_CAP         # 3本目は月間上限
issue_coupon(m0002, SW-A) -> ISSUED
issue_coupon(m0003, SW-A) -> BUDGET_EXCEEDED     # 予算を使い切った
campaign_budgets:     [('SW-A', 2000, 2000), ('SW-B', 1000, 300000), ('SW-C', 0, 300000)]
member_monthly_usage: [('m0001', '2026-09', 2000), ('m0002', '2026-09', 1000)]
                      # m0003 の行が無い = 断ったとき (2) で作った行も巻き戻った

既に壊れているデータを抱えたままの移行

既に予算を超えたキャンペーンがあるので CHECK をそのまま付けると失敗する。重複の一部は使用済みで、注文と結び付いているので消せない（顧客に落ち度もない）。

📋 コピー-- 0) 現状（Bad）の表: campaigns(issued_yen はずれている) と issued_coupons（重複あり、redeemed_at あり）
-- 1) 重複の後始末。未使用の 2 枚目以降は無効化、使用済みは「過去の重複」として残す（顧客に落ち度はない）
ALTER TABLE issued_coupons
  ADD COLUMN voided_at timestamptz,
  ADD COLUMN legacy_duplicate boolean NOT NULL DEFAULT false;
WITH ranked AS (
  SELECT coupon_id, redeemed_at,
         row_number() OVER (PARTITION BY campaign_id, member_id
                            ORDER BY redeemed_at IS NULL, issued_at, coupon_id) AS rn
    FROM issued_coupons)
UPDATE issued_coupons i
   SET voided_at        = CASE WHEN r.redeemed_at IS NULL THEN now() END,
       legacy_duplicate = true
  FROM ranked r
 WHERE i.coupon_id = r.coupon_id AND r.rn > 1;
-- 2) 1会員1枚（過去の重複は対象外）。本番は CONCURRENTLY で別に流す（トランザクション外）
CREATE UNIQUE INDEX one_coupon_per_member ON issued_coupons (campaign_id, member_id)
  WHERE NOT legacy_duplicate;
-- 3) 予算カウンタを実額から作り直す（ずれたカウンタは信じない）
CREATE TABLE campaign_budgets AS
SELECT c.campaign_id, c.discount_yen, c.budget_yen,
       COALESCE(SUM(i.amount_yen) FILTER (WHERE i.voided_at IS NULL), 0)::int AS issued_yen
  FROM campaigns c LEFT JOIN issued_coupons i USING (campaign_id)
 GROUP BY 1, 2, 3;
ALTER TABLE campaign_budgets ADD PRIMARY KEY (campaign_id);
-- 既に予算を超えたキャンペーンがあるので、検査は「これからの書き込み」にだけ掛ける
ALTER TABLE campaign_budgets
  ADD CONSTRAINT issued_within_budget CHECK (issued_yen BETWEEN 0 AND budget_yen) NOT VALID;
-- 4) 会員×月の使用額も実額から
CREATE TABLE member_monthly_usage AS
SELECT member_id, month, SUM(amount_yen)::int AS used_yen
  FROM issued_coupons WHERE voided_at IS NULL GROUP BY 1, 2;
ALTER TABLE member_monthly_usage ADD PRIMARY KEY (member_id, month);

既存表の一意性は部分一意索引なので、アプリの ON CONFLICT の対象は (campaign_id, member_id) WHERE NOT legacy_duplicate に変える（述語から部分索引を推論させる）。

汚れたデータで流した結果 実測

('m1', redeemed=False, voided=True,  legacy_duplicate=True)   # 未使用の重複は無効化
('m1', redeemed=True,  voided=False, legacy_duplicate=False)  # 使用済みを正として残す
('m3', redeemed=True,  voided=False, legacy_duplicate=False)
('m3', redeemed=False, voided=True,  legacy_duplicate=True)
campaign_budgets: [('SW-A', 1000, 2000, 3000)]                  # 予算超過のまま移行できた
新しい書き込みで超過 → 23514 violates check constraint "issued_within_budget"
m1 への再発行        → None（0 行）

超過した予算（SW-A の 41.2 万円）は制約では直せない事実なので、経理と販促に予算超過の実績として報告する。NOT VALID はその事実を消さずに、これ以上増やさないための選択。旧表と新表への二重書き込み期間は作らない（その間に再びずれるため）。

迷った案と採らなかった理由

直した後に見るもの・止め方

ロールバック: 旧コードは campaigns.issued_yen を読むので、戻すなら campaign_budgets から書き戻してから。一意索引と NOT VALID の CHECK は旧コードでも害がないので残す。

ポイント解説

1. 分離レベルは「どの異常を DB が検出するか」の選択で、正しさの保証ではない: REPEATABLE READ は同じ行の lost update を 40001 で止めるが、違う行への write skew は通す。後輩A の案で R1・R2 が直ったのは偶然
2. 「読んで・判定して・書く」を「判定しながら書く」1文に: UPDATE … WHERE x + :a <= 上限 RETURNING は READ COMMITTED でも先行のコミットを待って新しい行で再評価する。「尋ねるな、命じろ」の DB 版
3. まだ無い行はロックできないので、ロックの相手を先に作る: 不変条件ごとに「どの1行で衝突させるか」を設計表に書くのが同時実行制御の設計の中心
4. SERIALIZABLE は正しいが、再試行が「たまに」から「いつも」に変わる: 8 並列で約4割がやり直し、32 並列で 3.3〜4.4% が使い切り。どちらを使うかは同じ負荷試験で RETRY_EXHAUSTED を数えて決める
5. 再試行は DB が巻き戻したと分かっている失敗だけ、副作用は外へ: 通知がトランザクション内だと、再試行のたびに「クーポンの無い通知」が増える（最大 3,140 通／1,800 リクエスト）

実務への応用

• 「残数・上限」を全部棚卸しする: クーポン残数・キャンペーン予算・会員別上限・在庫引当・ポイント付与上限。「どの1行で衝突させているか」が書けないものが write skew の候補
• レビューの1行: 「この SELECT の結果で分岐した後に書く行は、SELECT が読んだ行と同じか？」 #2・#3・#4 はこの1問で出る
• 同時実行テストを CI に常設: 実 PostgreSQL のサービスコンテナで3秒。モックの単体テストでは7点のどれも捕まらない
• dbt で翌朝に気づく: 「会員×月の合計 ≤ 2,000」「キャンペーン合計 ≤ 予算」を singular test に（BigQuery 同期前提）
• 証券会社の経験との接続: 買付余力を読んで足りれば発注、も同じ構造。余力を1行に実体化して条件付きで拘束するか、口座単位で直列化する。クーポン予算は「販促部門の買付余力」

扱わなかった範囲（と理由）

• Cloud SQL・AlloyDB での数値 — 違反件数は環境に依存しないが、req/s と p99 は変わる。Good の 32 並列の p99（118〜557 ms）のばらつきはステージングで再測定する
• 接続プール（pgbouncer のトランザクションプーリング） — SET LOCAL は安全だが、プールの大きさは別問題として外した
• 分散 DB（Spanner など） — 単一 PostgreSQL に絞った
• 複数アカウントによる不正 — R3 は会員 ID 単位。本人確認・端末指紋の問題として外した
• クーポン利用（redeem）側の同時実行 — 同じ構造（redeemed_at IS NULL を条件に更新）だがカート側のシステム
• 景品表示法 — 自社取引で使える値引きクーポンは一般に景品規制の対象外として扱われるため、上限は社内規程として扱った

一次資料（参照日 2026-09-27。挙動は手元の PostgreSQL 14.11 で実測して突き合わせ）

• PostgreSQL 14「13.2 Transaction Isolation」 https://www.postgresql.org/docs/14/transaction-iso.html
• PostgreSQL 14「13.3 Explicit Locking」 https://www.postgresql.org/docs/14/explicit-locking.html ／「SELECT — The Locking Clause」 https://www.postgresql.org/docs/14/sql-select.html#SQL-FOR-UPDATE-SHARE
• PostgreSQL 14「INSERT — ON CONFLICT」 https://www.postgresql.org/docs/14/sql-insert.html ／「ALTER TABLE — NOT VALID」 https://www.postgresql.org/docs/14/sql-altertable.html ／「Error Codes」 https://www.postgresql.org/docs/current/errcodes-appendix.html
• D. Ports, K. Grittner「Serializable Snapshot Isolation in PostgreSQL」VLDB 2012 https://arxiv.org/abs/1208.4179
• H. Berenson ほか「A Critique of ANSI SQL Isolation Levels」SIGMOD 1995 https://arxiv.org/abs/cs/0701157
• M. Kleppmann『Designing Data-Intensive Applications』第7章 ／ M. Brooker「Exponential Backoff And Jitter」 https://aws.amazon.com/blogs/architecture/exponential-backoff-and-jitter/ ／ C. Richardson「Transactional outbox」 https://microservices.io/patterns/data/transactional-outbox.html
• 仙塲大也『良いコード／悪いコードで学ぶ設計入門』改訂新版 —「尋ねるな、命じろ」、完全コンストラクタ

今日のまとめ

次のステップ

• 発展問題: ブラックフライデーで先着クーポンを 1 万 req/s で配る。(a) 予算を 16 個の小口行に分割、(b) Pod ごとに予算を事前割り当て、(c) SERIALIZABLE、を取りこぼし・超過可能性・実装量で比べ、sim.py を拡張して測って決めよ
• 参考: lost update、write skew、phantom、snapshot isolation、SSI、EvalPlanQual、materializing conflicts、compare-and-set、FOR UPDATE SKIP LOCKED、NOT VALID constraint、partial unique index、full jitter

自己評価（あとで記入）

自分の回答

気づき・メモ