Webhookの冪等性をTransactional Inboxで実装する：イベントIDの重複排除だけでは防げない『永久消失』と『二重処理』

# ❌ 壊れた例：受信記録を先に別トランザクションでコミットしている
def receive_mark_first_broken(conn, *, endpoint, event):
    with conn.transaction():
        inserted = conn.execute(
            "INSERT INTO webhook_inbox (provider, endpoint, event_id, event_type, payload) "
            "VALUES ('stripe', %s, %s, %s, %s) ON CONFLICT DO NOTHING RETURNING event_id",
            (endpoint, event.id, event.type, Jsonb(event.payload)),
        ).fetchone()
    if inserted is None:
        return Outcome.DUPLICATE        # 記録済み＝処理済みとみなして 200
    with conn.transaction():
        apply_event(conn, event)        # ← ここで失敗すると…
    return Outcome.PROCESSED

1回目の配信:  受信記録 COMMIT → 業務処理で DB エラー → ROLLBACK → 500 を返す
2回目の配信:  受信記録の INSERT が衝突 → 「重複」と判断 → 200 を返す
→ 入金は一度も計上されず、Stripe はもう再送しない

# ❌ 壊れた例：業務処理のコミットと受信記録が別トランザクション
def receive_mark_after_broken(conn, *, endpoint, event):
    seen = conn.execute(
        "SELECT 1 FROM webhook_inbox WHERE provider='stripe' AND endpoint=%s AND event_id=%s",
        (endpoint, event.id),
    ).fetchone()
    if seen:
        return Outcome.DUPLICATE
    with conn.transaction():
        apply_event(conn, event)        # 業務処理を COMMIT
    # ← ここでプロセスが落ちる（デプロイ、OOM、タイムアウト）
    with conn.transaction():
        conn.execute("INSERT INTO webhook_inbox ...")
    return Outcome.PROCESSED

1回目の配信:  業務処理 COMMIT（入金を計上）→ 受信記録の前にクラッシュ → 2xx は返らない
2回目の配信:  受信記録が無い → 未処理と判断 → 業務処理をもう一度 COMMIT
→ 入金が2回計上される

BEGIN
  INSERT INTO webhook_inbox (event_id ...) ON CONFLICT DO NOTHING RETURNING ...
    → 行が返らなければ重複。何もせず COMMIT して 200
  UPDATE invoices / INSERT ledger_entries（業務更新）
  INSERT INTO outbox（下流への通知が要るなら）
  UPDATE webhook_inbox SET processed_at = now()
COMMIT → 200

-- 受信台帳。主キーが「同じ配信は1回だけ受け付ける」を DB に裁かせる
CREATE TABLE webhook_inbox (
    provider     text        NOT NULL,              -- 'stripe' など。プロバイダ間の ID 衝突を分離
    endpoint     text        NOT NULL,              -- 同じイベントを受ける別エンドポイントを分離
    event_id     text        NOT NULL,              -- Stripe なら evt_...
    event_type   text        NOT NULL,
    payload      jsonb       NOT NULL,              -- 署名検証済みの生 JSON
    received_at  timestamptz NOT NULL DEFAULT now(),
    available_at timestamptz NOT NULL DEFAULT now(), -- 非同期処理の再試行時刻
    attempts     integer     NOT NULL DEFAULT 0,
    last_error   text,
    processed_at timestamptz,                       -- NULL = 未処理
    PRIMARY KEY (provider, endpoint, event_id)
);
CREATE INDEX webhook_inbox_unprocessed
    ON webhook_inbox (available_at) WHERE processed_at IS NULL;

CREATE TABLE invoices (
    id          text        PRIMARY KEY,           -- in_...
    customer_id text        NOT NULL,
    status      text        NOT NULL CHECK (status IN ('draft', 'open', 'paid', 'void', 'uncollectible')),
    amount_paid bigint      NOT NULL DEFAULT 0,
    paid_at     timestamptz,
    updated_at  timestamptz NOT NULL DEFAULT now()
);

-- 下流への通知（構造は Transactional Outbox の記事と同じ）
CREATE TABLE outbox (
    id             bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    aggregate_type text        NOT NULL,
    aggregate_id   text        NOT NULL,
    event_type     text        NOT NULL,
    payload        jsonb       NOT NULL,
    created_at     timestamptz NOT NULL DEFAULT now(),
    published_at   timestamptz
);

-- 業務レベルの冪等性：同じ請求の入金を2回計上しない（別 Event オブジェクトで届いても）
CREATE TABLE ledger_entries (
    id         bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    invoice_id text   NOT NULL,
    kind       text   NOT NULL,
    amount     bigint NOT NULL,
    UNIQUE (invoice_id, kind)
);

import json
from dataclasses import dataclass
from typing import Any

import stripe


@dataclass(frozen=True)
class VerifiedEvent:
    """署名検証を通過したイベント。これ以外の型は業務ロジックへ入れない。"""
    id: str
    type: str
    payload: dict[str, Any]


class InvalidWebhook(Exception):
    pass


def verify_stripe_event(raw_body: bytes, sig_header: str | None, secret: str) -> VerifiedEvent:
    """フレームワークがボディを dict に変換する前の bytes で署名を検証する。"""
    try:
        stripe.Webhook.construct_event(raw_body, sig_header, secret)
    except (ValueError, stripe.SignatureVerificationError) as exc:
        raise InvalidWebhook(str(exc)) from exc
    payload = json.loads(raw_body)
    return VerifiedEvent(id=payload["id"], type=payload["type"], payload=payload)

def _apply_invoice_created(conn: psycopg.Connection, inv: dict[str, Any]) -> None:
    # 作成は「無ければ入れる」だけ。paid の後に遅れて届いても状態を巻き戻さない
    conn.execute(
        """
        INSERT INTO invoices (id, customer_id, status)
        VALUES (%(id)s, %(customer)s, 'open')
        ON CONFLICT (id) DO NOTHING
        """,
        {"id": inv["id"], "customer": inv["customer"]},
    )


def _apply_invoice_paid(conn: psycopg.Connection, inv: dict[str, Any]) -> None:
    # paid は created より先に届くことがある → upsert。void 済みの請求は上書きしない
    conn.execute(
        """
        INSERT INTO invoices (id, customer_id, status, amount_paid, paid_at)
        VALUES (%(id)s, %(customer)s, 'paid', %(amount_paid)s, now())
        ON CONFLICT (id) DO UPDATE
           SET status      = 'paid',
               amount_paid = EXCLUDED.amount_paid,
               paid_at     = COALESCE(invoices.paid_at, EXCLUDED.paid_at),
               updated_at  = now()
         WHERE invoices.status NOT IN ('paid', 'void')
        """,
        {"id": inv["id"], "customer": inv["customer"], "amount_paid": inv["amount_paid"]},
    )
    row = conn.execute("SELECT status FROM invoices WHERE id = %s", (inv["id"],)).fetchone()
    if row is None or row[0] != "paid":
        # void 済みの請求への入金は異常系。計上せず、Reconciliation で人の確認に回す
        return
    booked = conn.execute(
        """
        INSERT INTO ledger_entries (invoice_id, kind, amount)
        VALUES (%(id)s, 'payment', %(amount_paid)s)
        ON CONFLICT (invoice_id, kind) DO NOTHING
        RETURNING id
        """,
        {"id": inv["id"], "amount_paid": inv["amount_paid"]},
    ).fetchone()
    if booked is not None:
        # 下流（領収書メール等）への通知は Outbox へ。業務更新と同じトランザクションで書く
        conn.execute(
            """
            INSERT INTO outbox (aggregate_type, aggregate_id, event_type, payload)
            VALUES ('Invoice', %(id)s, 'InvoicePaymentBooked', %(payload)s)
            """,
            {"id": inv["id"], "payload": Jsonb({"invoice_id": inv["id"], "amount": inv["amount_paid"]})},
        )


HANDLERS = {
    "invoice.created": _apply_invoice_created,
    "invoice.paid": _apply_invoice_paid,
}

from enum import Enum


class Outcome(Enum):
    PROCESSED = "processed"
    DUPLICATE = "duplicate"
    IGNORED = "ignored"


def apply_event(conn: psycopg.Connection, event: VerifiedEvent) -> bool:
    handler = HANDLERS.get(event.type)
    if handler is None:
        return False
    handler(conn, event.payload["data"]["object"])
    return True


def receive_sync(
    conn: psycopg.Connection, *, provider: str, endpoint: str, event: VerifiedEvent
) -> Outcome:
    """受信記録と業務更新を1つのトランザクションでコミットする。"""
    with conn.transaction():
        inserted = conn.execute(
            """
            INSERT INTO webhook_inbox (provider, endpoint, event_id, event_type, payload)
            VALUES (%s, %s, %s, %s, %s)
            ON CONFLICT (provider, endpoint, event_id) DO NOTHING
            RETURNING event_id
            """,
            (provider, endpoint, event.id, event.type, Jsonb(event.payload)),
        ).fetchone()
        if inserted is None:
            return Outcome.DUPLICATE

        handled = apply_event(conn, event)
        conn.execute(
            """
            UPDATE webhook_inbox SET processed_at = now()
             WHERE provider = %s AND endpoint = %s AND event_id = %s
            """,
            (provider, endpoint, event.id),
        )
    return Outcome.PROCESSED if handled else Outcome.IGNORED

時刻  配信 1（接続1）                              配信 2（接続2）
t1    BEGIN; INSERT inbox(evt_1) → 成功（未コミット）
t2                                               BEGIN; INSERT inbox(evt_1) → 一意インデックスで待機
t3    業務更新
t4    COMMIT
t5                                               → 衝突が確定。ON CONFLICT DO NOTHING → 0行
t6                                               COMMIT → DUPLICATE（200）

import os

from fastapi import FastAPI, Request, Response
from psycopg_pool import ConnectionPool
from starlette.concurrency import run_in_threadpool

app = FastAPI()
pool = ConnectionPool(os.environ["DATABASE_URL"], kwargs={"autocommit": True}, open=True)
WEBHOOK_SECRET = os.environ["STRIPE_WEBHOOK_SECRET"]  # ハードコードしない


def _handle(event: VerifiedEvent) -> Outcome:
    with pool.connection() as conn:
        return receive_sync(conn, provider="stripe", endpoint="billing", event=event)


@app.post("/webhooks/stripe")
async def stripe_webhook(request: Request) -> Response:
    raw_body = await request.body()  # パース前の生のバイト列
    try:
        event = verify_stripe_event(raw_body, request.headers.get("stripe-signature"), WEBHOOK_SECRET)
    except InvalidWebhook:
        return Response(status_code=400)
    await run_in_threadpool(_handle, event)  # 例外は 500 になり、Stripe が再送する
    return Response(status_code=200)

def receive_async(
    conn: psycopg.Connection, *, provider: str, endpoint: str, event: VerifiedEvent
) -> Outcome:
    """受信記録だけをコミットして 2xx を返す。以降の再試行責任はプロバイダからこちらへ移る。"""
    with conn.transaction():
        inserted = conn.execute(
            """
            INSERT INTO webhook_inbox (provider, endpoint, event_id, event_type, payload)
            VALUES (%s, %s, %s, %s, %s)
            ON CONFLICT (provider, endpoint, event_id) DO NOTHING
            RETURNING event_id
            """,
            (provider, endpoint, event.id, event.type, Jsonb(event.payload)),
        ).fetchone()
    return Outcome.PROCESSED if inserted else Outcome.DUPLICATE

from datetime import timedelta

InboxKey = tuple[str, str, str]  # (provider, endpoint, event_id)

CLAIM_INBOX_SQL = """
UPDATE webhook_inbox AS w
   SET attempts     = w.attempts + 1,
       available_at = now() + %(lease)s       -- lease: 処理中は他のワーカーから見えなくする
  FROM (SELECT provider, endpoint, event_id
          FROM webhook_inbox
         WHERE processed_at IS NULL
           AND available_at <= now()
           AND attempts < %(max_attempts)s
         ORDER BY available_at
         LIMIT 1
           FOR UPDATE SKIP LOCKED) AS picked
 WHERE (w.provider, w.endpoint, w.event_id) = (picked.provider, picked.endpoint, picked.event_id)
RETURNING w.provider, w.endpoint, w.event_id
"""


def claim_next(
    conn: psycopg.Connection, *, max_attempts: int = 10, lease: timedelta = timedelta(minutes=2)
) -> InboxKey | None:
    """未処理を1件選び、試行回数を数えてからコミットする（処理より前に数えるので、
    処理中にプロセスごと落ちても回数は残り、上限で止まる）。"""
    with conn.transaction():
        row = conn.execute(CLAIM_INBOX_SQL, {"lease": lease, "max_attempts": max_attempts}).fetchone()
    return (row[0], row[1], row[2]) if row else None


def process_claimed(conn: psycopg.Connection, key: InboxKey, *, timeout: str = "30s") -> bool:
    """取り出した1件を処理する。行をロックして processed_at を確かめ直し、
    業務更新と processed_at を同じトランザクションで確定する。

    lease が切れて別のワーカーが同じ行を取っても、ロックと再確認があるので業務更新は1回だけ。
    （業務の効果が「処理済み」の確認と同じ DB トランザクションにあるので fencing token は要らない。
    外部 API を呼ぶ処理では成り立たないので、Outbox Dispatcher の lease + fencing を使う）
    """
    try:
        with conn.transaction():
            # 1件の処理時間に上限を付ける。lease より十分短くする
            conn.execute("SELECT set_config('statement_timeout', %s, true)", (timeout,))
            row = conn.execute(
                """
                SELECT event_type, payload FROM webhook_inbox
                 WHERE provider = %s AND endpoint = %s AND event_id = %s
                   AND processed_at IS NULL
                   FOR UPDATE
                """,
                key,
            ).fetchone()
            # SKIP LOCKED にしない: 他ワーカーの claim 文が候補として一瞬ロックしただけでも
            # 飛ばしてしまい、lease が切れるまで処理されなくなる。待つ時間は statement_timeout が上限
            if row is None:
                return False  # 待っている間に別のワーカーが処理済みにした
            event_type, payload = row
            apply_event(conn, VerifiedEvent(id=key[2], type=event_type, payload=payload))
            conn.execute(
                """
                UPDATE webhook_inbox SET processed_at = now(), last_error = NULL
                 WHERE provider = %s AND endpoint = %s AND event_id = %s
                """,
                key,
            )
        return True
    except Exception as exc:  # ワーカー境界: 1件の失敗でループ全体を止めない
        # 業務更新はロールバック済み。失敗を記録し、lease の残りを待たずにバックオフ後へ回す
        # （試行回数は claim_next で数え済み）
        with conn.transaction():
            conn.execute(
                """
                UPDATE webhook_inbox
                   SET last_error   = %s,
                       available_at = now() + make_interval(secs => least(300, 2 ^ attempts) * random())
                 WHERE provider = %s AND endpoint = %s AND event_id = %s
                   AND processed_at IS NULL   -- 他のワーカーが処理済みにした行は触らない
                """,
                (type(exc).__name__, *key),
            )
        return False


def process_next(conn: psycopg.Connection, *, max_attempts: int = 10) -> bool:
    """1件取り出して処理する。取り出せる行が無ければ False（ワーカーは少し待ってから再試行）。"""
    key = claim_next(conn, max_attempts=max_attempts)
    if key is None:
        return False
    process_claimed(conn, key)
    return True

idempotency_key = f"webhook:{event.id}:update-customer-metadata"

-- 未処理の受信と、最も古い未処理の待ち時間
SELECT count(*)                        AS unprocessed,
       max(now() - received_at)        AS oldest_age,
       count(*) FILTER (WHERE attempts >= 10) AS needs_review
  FROM webhook_inbox
 WHERE processed_at IS NULL;

-- 直近1時間の受信と重複率（重複は DUPLICATE を返した数をアプリ側で数える）
SELECT event_type, count(*)
  FROM webhook_inbox
 WHERE received_at > now() - interval '1 hour'
 GROUP BY event_type
 ORDER BY 2 DESC;

DELETE FROM webhook_inbox
 WHERE processed_at IS NOT NULL
   AND processed_at < now() - interval '30 days';

def test_concurrent_same_event(dsn, conn):
    e = ev("evt_1", "invoice.paid")
    barrier = threading.Barrier(8)

    def worker():
        with psycopg.connect(dsn) as c:
            barrier.wait()
            return receive_sync(c, provider="stripe", endpoint="billing", event=e)

    with ThreadPoolExecutor(8) as ex:
        results = [f.result() for f in [ex.submit(worker) for _ in range(8)]]

    assert results.count(Outcome.PROCESSED) == 1
    assert results.count(Outcome.DUPLICATE) == 7
    assert conn.execute("SELECT count(*) FROM ledger_entries").fetchone()[0] == 1