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

2026-09-27 日曜(Day 136) 日曜弱点補強 ★★★★★ PostgreSQL 14.11 で実測 Python 3.13 / psycopg 3.3

概要

🧮

分離レベルは「何を検出するか」の選択

REPEATABLE READ は同じ行の競合(lost update)は 40001 で止めるが、違う行を書く write skew は通す。実測: 上限 2,000 円の会員が 3,000 円に。

🎯

判定と更新を1文に

UPDATE … WHERE x + :a <= 上限 RETURNING。READ COMMITTED でも先行のコミット後の行で WHERE を評価し直す。書籍の「尋ねるな、命じろ」の DB 版。

🧱

無い行はロックできない

FOR UPDATE は 0 行なら何もロックしない。会員×月の行を先に作り、衝突の相手を実体化する(materializing conflicts)。

📏

正しさは SQL で数える

1,800 リクエスト × 4 実装 × 2 並列度 × 3 回。Good は違反 0・再試行 0。SERIALIZABLE も違反 0 だが 32 並列で 3.3〜4.4% を断った。

問題

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

決まり中身出どころ
R1 予算上限キャンペーンごとの発行額は稟議で承認された予算以内販促予算の稟議
R2 1会員1枚同じキャンペーンは1会員1枚キャンペーン規約
R3 会員の月間上限1会員が1か月に受け取る値引きは合計 2,000 円まで(3本のうち2本)粗利を守る社内規程(転売目的の大量取得対策を兼ねる)

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

観測値出どころ
SW-A の実発行額3,412,000 円(予算 300万円、412枚超過)issued_coupons の合計
同時点の campaigns.issued_yen2,987,000 円(予算内に見えていた)campaigns 表
同じ会員・同じキャンペーンに2枚以上1,036 件集計
月間上限を超えた会員388 人集計
使用済みの予算外値引き81.6 万円経理
ピーク(9/19 10:00 の push 直後)420 req/s、p99 1.9 秒DataDog APM

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

後輩A「トランザクションを REPEATABLE READ にすれば、途中で読んだ値が変わらなくなるので直ります。1行足すだけです」
後輩B「いっそ全部 SERIALIZABLE にしましょう。PostgreSQL の SSI なら直列実行と同じ結果が保証されます」

設問

  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)

モックの単体テスト4本は全部緑。それでも 7つの問題 がある。
bad_issue.py(psycopg 3、autocommit 接続+conn.transaction())
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) の突き合わせなし/一意制約なし/テストはモックの逐次実行のみ
問題点サマリー(7点)
1カウンタの書き戻し — 読んだ値+1,000 を書くので加算が消える(lost update)。カウンタ 298.7万 対 実額 341.2万
2予算を読んでから書く — 判定と書き込みの隙間で超える
3重複を SELECT で確認 — まだ無い行(phantom)は誰も止めない。一意制約なし
4月間上限: 既存行だけ FOR UPDATE — 0 行ならロックなし。別キャンペーンは別の行を書く(write skew)
5通知がトランザクション内 — ロックを持って外部 API を待ち、巻き戻ると「通知だけ届く」
6何でも即時3回やり直し — 40001 も接続断も区別なし、待たずに再衝突
7正本が2つ・制約もテストも無い — ずれに誰も気づけない

ヒント(段階的開示)

ヒント1 — 方向性
「読んで、Python で判定して、書く」の3手に分かれている箇所を全部探す。読んでから書くまでに別のトランザクションが同じ判定を通っていないか。そして、2つのトランザクションが書く行は同じ行か、違う行か。同じ行なら PostgreSQL は気づけるが、違う行だと REPEATABLE READ では気づけない。
ヒント2 — アプローチ
  • 判定と更新を1文に: UPDATE … SET x = x + :a WHERE … AND x + :a <= :上限 RETURNING …。READ COMMITTED でも先行がコミットしたら新しい行で WHERE を評価し直す
  • まだ存在しない行はロックできない。R3 を守るにはロックを取る相手の行(会員×月)を先に作る(materializing conflicts)
  • R2 は一意制約に判定させる(ON CONFLICT DO NOTHING)
  • 再試行は SQLSTATE 40001・40P01・55P03 だけ。副作用がトランザクション内にあると再試行が副作用を増やす
ヒント3 — 骨格
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
測るときは会員ごとに「3キャンペーン × 二重送信」の6本を同時に投げ、終わった後に不変条件の違反件数を SQL で数える。

問題点分析(7点)

#箇所何が起きるか異常の名前破る決まり
1SET issued_yen=%s(読んだ値+d)同時に読んだ2本の片方の加算が消え、カウンタが実額より小さくなり予算判定が通り続けるlost updateR1
2予算を SELECT で判定 → INSERT判定と書き込みの隙間で他の発行が入るcheck-then-act(TOCTOU)R1
3SELECT 1 で重複確認、一意制約なし二重タップの2本が両方「未発行」と判定phantomR2
4既存行を FOR UPDATE して Python で合計0 行ならロックなし。別キャンペーンの2本は別の新しい行を書くので互いに気づかないwrite skewR3
5通知がトランザクション内(INSERT と UPDATE の間)ロックを持って外部 API を待つ。巻き戻ると通知だけ届く副作用の非原子性顧客体験
6except Exception: continue再試行の分類もバックオフもない。#5 と組むと再試行のたびに通知が増える―可用性
7正本が2つ、突き合わせ・制約・同時実行テストなし「カウンタ 298.7万・実額 341.2万」に誰も気づけない不変条件の置き場所がない検知

実測 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回の範囲。絶対値でなく違反件数と相対差を読む。

実装並列R1: SW-A 実額(予算10万)R1: カウンタR2: 重複R3: 超過会員巻き戻った発行への通知断った正当な要求再試行req/s
Bad(READ COMMITTED)829.9〜30.5万10.0万735〜744299〜300000346〜378
Bad(READ COMMITTED)3259.3〜59.8万5.9〜6.7万877〜892300000485〜1,024
後輩A: +REPEATABLE READ810.0万10.0万074〜821,424〜1,481137〜1431,424〜1,481209〜234
後輩A: +REPEATABLE READ3210.0万10.0万020〜233,057〜3,140748〜7833,057〜3,140412〜447
後輩B: SERIALIZABLE+40001 再試行810.0万10.0万0000714〜7401,427〜1,546
後輩B: SERIALIZABLE+40001 再試行3210.0万10.0万00059〜791,511〜1,954830〜1,002
Good: 条件付き更新+実体化810.0万10.0万000001,669〜1,746
Good: 条件付き更新+実体化3210.0万10.0万00000606〜1,015

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

後輩A(REPEATABLE READ): 採らない。 R1・R2 が直ったのは「読んだ値が変わらないから」ではなく、Bad が全員同じ campaigns 行を UPDATE していたので 40001(concurrent update)が偶然すべての競合を拾ったから。R3 の2本は違う行を書くので直らない(超過 20〜82 人)。さらに 32 並列で約42%が SYSTEM_ERROR、巻き戻った発行への通知が最大 3,140 通と、現状に無かった壊れ方を新しく生む。
後輩B(SERIALIZABLE): 正しさは本物。既定にはしないが、条件つきで次の候補に残す。 R1〜R3 すべて違反 0。ただし 8 並列で約4割がやり直し、32 並列では再試行を使い切った 59〜79 件(3.3〜4.4%)で 600 枚配るべきところ 572〜579 枚。全書き込み経路を SERIALIZABLE に揃える必要があり、ホットスタンバイでは使えず、索引が無いと述語ロックが粗くなって偽の衝突が増える。

write skew の図解(Before / 後輩A・B / Good)

会員 m0001(今月すでに SW-A の 1,000 円を保有、上限 2,000 円)に SW-B と SW-C を同時発行 後輩A: REPEATABLE READ Tx A(SW-B) Tx B(SW-C) 読む: 今月の合計 1,000 読む: 今月の合計 1,000 INSERT 行X(SW-B, 1,000) INSERT 行Y(SW-C, 1,000) COMMIT ✓ COMMIT ✓ 合計 3,000 円 書いた行が X と Y で違う → 同じ行の競合が無いので検出されない 実測: 2セッション再現で両方 COMMIT → total used = 3000 / 1,800 リクエスト負荷で上限超過会員 20〜82 人 後輩B: SERIALIZABLE 同じ流れで、B の COMMIT が 40001(read/write dependencies)で失敗 → 合計 2,000。正しいが、再試行が常態化(32 並列で 3.3〜4.4% が再試行を使い切る) Good: 会員×月の行を実体化して条件付き UPDATE(READ COMMITTED) Tx A(SW-B) Tx B(SW-C) INSERT 発行行(一意) INSERT 発行行(一意) UPDATE 会員×月 used 1,000→2,000(行ロック) 同じ行の UPDATE → A の確定を待つ COMMIT ✓ 新しい行で WHERE を再評価 2,000 + 1,000 ≤ 2,000 は偽 → 0 行 → MONTHLY_CAP → 発行行ごと ROLLBACK 衝突させる相手を「会員×月の1行」に決めたので、違う行を書く2本が必ず同じ行でぶつかる(materializing conflicts) 実測: 1,800 リクエスト × 8/32 並列 × 3 回で、R1〜R3 の違反 0 件・再試行 0 回・発行ちょうど 600 枚 見分け方: 「SELECT の結果で分岐した後に書く行は、SELECT が読んだ行と同じか?」 違うなら write skew の候補

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

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

決まり正本ロックの実体判定
R1 予算campaign_budgets.issued_yen+CHECKキャンペーン行条件付き加算
R2 1会員1枚一意制約 (campaign_id, member_id)一意索引のキーON CONFLICT DO NOTHING RETURNING
R3 月間上限member_monthly_usage.used_yen会員×月の行(無ければ先に作る)条件付き加算
通知notification_outbox(コミットと同時に確定)relay が coupon_id を冪等キーに送る
1回の発行(READ COMMITTED の1トランザクション)で、ロックを取る順番 発行要求 IssueRequest (1) 一意索引のキー (campaign, member) R2 1会員1枚 0 行 → ALREADY_ISSUED (2) 会員×月の行 無ければ先に作る R3 月間上限 0 行 → MONTHLY_CAP (3) キャンペーン行 いちばん混む → 最後 R1 予算上限 0 行 → BUDGET_EXCEEDED (4) outbox 行 → COMMIT 送信は relay が後で ロックを持つ時間(右端=COMMIT) 一意キー: 同じ会員×キャンペーンの二重送信だけが待つ 会員×月: 同じ会員の別キャンペーンだけが待つ キャンペーン行: 全員が待つ → 最短に 順番を全員で固定するのでデッドロックは起きない(実測 40P01 = 0 回)。業務ルールで断るときは例外で抜け、(1)〜(3) をまとめて巻き戻す。

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

模範解答(Bad → Good)

Before — 読む・Python で判定・書く
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回
After — 判定しながら書く(1文ずつ)
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)
  • _Rejected を with conn.transaction() の中で投げて外で受ける。psycopg は例外で抜けると巻き戻すので、(1) の発行行も (2) の会員×月の行も、断ったときには残らない
  • R3 を R1 より先に置いたのはロック保持時間のため。予算切れと月間上限の両方に当たる会員には MONTHLY_CAP が返る(「次に何をすべきか」が分かりやすい方を優先)
  • relay_outbox は送信とマークの間で落ちると同じ通知をもう一度送る(少なくとも1回)。二重は push 事業者側の冪等キーで消す(9/26 と同じ考え方)
-- 不変条件ごとに「ロックを取る行」を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;

月間上限 2,000 円は CHECK に書かずアプリの定数にした。規程改定で変わる値で、CHECK にすると過去月の行まで新しい上限で縛るため。予算は行ごとに値を持つので CHECK に書ける。

パターン使った場所『良いコード/悪いコードで学ぶ設計入門』
条件付き更新(compare-and-set)R1・R3 の UPDATE … WHERE … RETURNINGメソッドの章「尋ねるな、命じろ」
衝突の実体化(materializing conflicts)member_monthly_usage―(DDIA 第7章)
一意制約による冪等化R29/26 の続き
完全コンストラクタ・値オブジェクトIssueRequestクラス設計の章
列挙型で結果を表すIssueResult(StrEnum)条件分岐の章
transactional outboxnotification_outbox―(9/26 の続き)
exponential backoff with full jitter再試行―

テストと実行例

同時実行テストの要点: ① 実 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 はその事実を消さずに、これ以上増やさないための選択。旧表と新表への二重書き込み期間は作らない(その間に再びずれるため)。

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

案採らなかった理由前提が変わったら
全体を SERIALIZABLE(後輩B)正しいが 32 並列で 3.3〜4.4% の正当な要求を断った。全書き込み経路を揃える運用コスト、リードレプリカで使えないロックすべき行を人が特定しにくい不変条件が増えたら切り替える
先頭で SELECT … FOR UPDATE(悲観ロック)正しいがロック保持がトランザクション全体に伸びる。条件付き UPDATE の方が短い判定が WHERE に書けないほど複雑になったら
pg_advisory_xact_lock(hashtext(…))ロックがスキーマに現れず、別経路がロックし忘れると破られる。ハッシュ衝突で無関係な会員同士が待つ他チーム管理の DB で表を増やせないときの次善策
Redis の DECRBYDB と Redis の二重書き込みで、今回の「正本が2つでずれた」を作り直すピーク 1 万 req/s でキャンペーン行が律速になったら事前割り当てと比較
予算を N 個の小口行に分割集中は N 分の1になるが、小口ごとの残りによる取りこぼしと寄せ直しが要る。今の負荷では1行で足りる行ロック待ちが p99 の半分を超えたら
排他制約・トリガで合計を検査排他制約は「重なり」しか表現できない。トリガの中の SUM も READ COMMITTED では同時の INSERT が見えず write skew が残る―
発行後に夜間バッチで超過分を取り消す一度見せたクーポンの取り消しは信頼の損失と問い合わせが大きい超過が許される業務(抽選の後追い調整など)なら補償の方が安い

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

指標正常こうなったら止める/切り替える
毎日 03:00 の突き合わせ(カウンタ 対 実額)0 件差が出たら、アプリ以外の経路が正本を迂回している。当日中に特定
R3 超過会員0 人(移行前の過去月を除く)1人でも出たら前提が崩れている。発行を止めて調べる
RETRY_EXHAUSTED の割合0%(実測 0 回)0.1% 超でキャンペーン行の混雑 → カウンタ分割を検討
行ロック待ち(wait_event_type = 'Lock')p99 の 1 割未満p99 の半分を超えたらカウンタ分割。逆に、実体化の行が3種類を超えたら SERIALIZABLE を同じ負荷試験で測り、RETRY_EXHAUSTED < 0.1% なら切り替え
55P03(lock_timeout)0長いトランザクション(手作業の BEGIN 放置など)を xact_start で探す

ロールバック: 旧コードは 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
  • 仙塲大也『良いコード/悪いコードで学ぶ設計入門』改訂新版 —「尋ねるな、命じろ」、完全コンストラクタ

今日のまとめ

「読んで・判定して・書く」を別々の文で行うと、分離レベルを上げても違う行を書く競合(write skew)は残る。不変条件ごとに「どの1行で衝突させるか」を決め、その行を先に作って UPDATE … WHERE 上限内 RETURNING の1文で判定と更新を同時に行えば、READ COMMITTED のまま再試行 0 回で守れる(実測: 1,800 リクエストで違反 0 件)。

次のステップ

  • 発展問題: ブラックフライデーで先着クーポンを 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

自己評価(あとで記入)

自分の回答

気づき・メモ