A+B 複合 — CronJob の配信バッチを「何度起動されても・重なっても・途中で落ちても」安全にする（第10・13・4章 × CronJob × SKIP LOCKED）

概要

テーマ: Kubernetes CronJob の販促メール配信バッチを「何度起動されても・重なっても・途中で落ちても」安全にする（「良いコード・悪いコードで学ぶ設計入門」第10章 例外の握り潰し × 第13章 コマンド・クエリ分離 × 第4章 副作用の局所化 ／ CronJob の concurrencyPolicy・backoffLimit ／ PostgreSQL の FOR UPDATE SKIP LOCKED ／ 特定電子メール法第3条第3項）

問題

MOps 販促システムチームの spring-campaign-sender は、春キャンペーンの案内メールを会員 120万人に配信するバッチである。GKE Standard 上の CronJob が毎日 02:00（JST）に Pod を1つ起動し、Cloud SQL for PostgreSQL 17 の campaign_targets（配信対象表）を読み、メール配信事業者の HTTP API に1通ずつ送り、送れたら sent_at を書く。Pod のメモリ上限は 256Mi。

9月の棚卸しで、次のことが分かった（集計期間 2026-09-01〜09-30。本問の前提データ。実データではなく本問の設定値）。

後輩から「メモリ上限を 1Gi に上げて、Job の backoffLimit を 0 にすれば、再実行が無くなるので二重送信は止まるはずです」と提案が来た。あなたはシニアとして、バッチのコードと CronJob の両方を直す設計を出す。

制約・前提条件

• Python 3.12+（検証は 3.13.12）、psycopg 3、requests、Cloud SQL for PostgreSQL 17（検証は 17 の公式コンテナ）、Kubernetes 1.30+（API の受理確認は 1.37 の kind）
• 配信事業者の API は Idempotency-Key ヘッダを受け付け、同じキーの再送を24時間は配信せずに 200 で返す（仮定。事業者の仕様書で確認する前提）。レート上限は 500通/秒（仮定）
• 配信対象は 120万人で固定の一覧（campaign_targets に事前に投入済み）。会員DBでの配信停止は campaign_targets.opted_out_at に反映される（前提。反映は別システムが行う）
• 一時障害（5xx・タイムアウト）は4%、恒久エラー（宛先不正の 4xx）は0.1%程度（仮定。本問のフェイクで再現する値）
• 読み手はチームのエンジニア4名とテックリード

期待する回答形式: (1) 問題点の列挙（7点、番号付き。コードと CronJob の両方） (2) 改善後コード（Google スタイル docstring・名前付き定数・インラインコメント）と、DB 移行 SQL・CronJob manifest (3) 性質テスト（pytest） (4) 実行例（input → output。壊した場合の数も） (5) 後輩の提案への回答と、迷った案・採らなかった理由 (6) 直した後に何を見て、どうなったら止めるか (7) 適用した設計パターン名と書籍の対応章

Bad → Good（対比）

cur.execute("SELECT member_id, email FROM campaign_targets WHERE campaign_id = %s", ...)
rows = cur.fetchall()
for member_id, email in rows:
    r = requests.post(MAIL_URL, json={...})
    if r.status_code != 200:
        print("failed", member_id)
        continue
    cur.execute("UPDATE ... SET sent_at = now() WHERE ...")
    conn.commit()

# CronJob
concurrencyPolicy: （既定 Allow）
backoffLimit: （既定 6）  timeZone: （未指定）

WITH c AS (SELECT member_id FROM campaign_targets
    WHERE campaign_id = %(cid)s AND opted_out_at IS NULL  -- #6 取る瞬間に見る
      AND ((state='pending' AND (not_before IS NULL OR not_before <= now()))
           OR (state='sending' AND lease_until < now()))
    ORDER BY member_id LIMIT %(n)s FOR UPDATE SKIP LOCKED)
UPDATE ... SET state='sending', claim_id=%(claim)s, lease_until=now()+...

headers={"Idempotency-Key": f"{cid}:{member_id}"},
timeout=(MAIL_CONNECT_TIMEOUT_S, MAIL_READ_TIMEOUT_S)

RETRY → pending（待ち時間つき）／5回で failed
PERMANENT → failed ／ exit_code: 0 / 1 / 2

... WHERE t.state='sending' AND t.claim_id = %(claim)s

# CronJob
concurrencyPolicy: Forbid   backoffLimit: 2
activeDeadlineSeconds: 4200  timeZone: Asia/Tokyo

注釈の ◀ #N は、問題点の番号に対応する。

Bad の全文

現状のコード（Bad）

bad_send.py

📋 コピーimport os

import psycopg
import requests

MAIL_URL = os.environ.get("MAIL_URL", "https://mail.example.internal/v1/send")


def main():
    conn = psycopg.connect(os.environ["DB_DSN"])
    cur = conn.cursor()
    cur.execute(
        "SELECT member_id, email FROM campaign_targets WHERE campaign_id = %s",
        (os.environ["CAMPAIGN_ID"],),
    )
    rows = cur.fetchall()
    for member_id, email in rows:
        r = requests.post(MAIL_URL, json={"to": email, "member_id": member_id, "template": "spring"})
        if r.status_code != 200:
            print("failed", member_id)
            continue
        cur.execute("UPDATE campaign_targets SET sent_at = now() WHERE campaign_id = %s AND member_id = %s",
                    (os.environ["CAMPAIGN_ID"], member_id))
        conn.commit()


if __name__ == "__main__":
    main()

CronJob（抜粋）

📋 コピーspec:
  schedule: "0 17 * * *"         # UTC 17:00 = JST 02:00 のつもり。timeZone は未指定
  jobTemplate:
    spec:
      template:
        spec:
          restartPolicy: OnFailure
          containers:
            - name: sender
              image: asia-northeast1-docker.pkg.dev/mops-prod/promo/campaign-sender:1.8.0
              resources:
                limits: { memory: 256Mi }
# concurrencyPolicy, startingDeadlineSeconds, backoffLimit, activeDeadlineSeconds は未指定

設計の図（SVG）

パターン名: work queue on PostgreSQL（claim + lease）、fencing token（札）、idempotency key。書籍対応: 第13章（コマンド・クエリ分離。claim は変更して値を返す CQS の例外）、第10章（例外の握り潰し）、第4章（副作用の局所化: send_batch は DB に触らない）。

ヒント（段階的開示）

📋 コピーdef run(conn, campaign_id, mail, *, deadline_s):
    claim = uuid4()                                   # このワーカーの札
    while time_left():
        batch = claim_batch(conn, campaign_id, claim)   # SKIP LOCKED で 200 件取り、state='sending'、lease 90 秒
        if not batch:
            wait = seconds_until_work(conn, campaign_id)  # 再試行待ち・他の期限待ち。None なら終わり
            ...
        outcomes = send_batch(batch, mail)               # DB に触らない。予算超過分は RELEASE（手放す）
        finish_batch(conn, campaign_id, claim, batch, outcomes)  # 札が一致する行だけ 1 文で更新
    return report   # exit_code: 0 / 1（時間切れ。起動し直せば続き）/ 2（恒久失敗が多すぎる）

問題点分析（7点）

後輩の提案が #1〜#7 のどれを直すかは (5) で答える。

改善後コード（Good）

構造は第13章（コマンド・クエリ分離）・第4章（副作用の局所化）に沿って、取る（DB を変える）／送る（外に副作用。DB に触らない）／記録する（札つき） の3つに分けた。claim_batch は「変更して値を返す」ので CQS の例外（pop と同じ種類）で、これを意識して1か所に閉じた。

campaign_sender.py

📋 コピー"""販促メール配信バッチ（K8s CronJob から毎日起動）。何度起動されても、重なっても、途中で落ちても安全に続きから配る。

構造（第13章 コマンド・クエリ分離／第4章 副作用の局所化）:
    claim_batch()  … 「次に配る相手」を取る（DB を変える。取った札を返す）
    send_batch()   … 取った相手に送る（外部への副作用。DB は触らない）
    finish_batch() … 結果を札つきで記録する（札が違えば何もしない）
    run()          … 上の3つを回し、状態を Report で返す
"""
from __future__ import annotations

import logging
import time
import uuid
from collections.abc import Callable
from dataclasses import dataclass, field
from enum import StrEnum
from typing import Final, Protocol

import psycopg
import requests

logger = logging.getLogger("campaign_sender")

# ── 数値の根拠は一箇所にまとめる（manifest の activeDeadlineSeconds と対にする） ──
BATCH_SIZE: Final[int] = 200            # 1回に取る件数。落ちたとき再送の恐れがある最大件数でもある
BATCH_BUDGET_S: Final[float] = 60.0     # 1バッチに使う時間の上限。超えたら残りは手放す
MAIL_CONNECT_TIMEOUT_S: Final[float] = 2.0
MAIL_READ_TIMEOUT_S: Final[float] = 5.0
LEASE_S: Final[int] = 90                # = BATCH_BUDGET_S 60 + 1通の最悪 7 + 余裕 23
MAX_ATTEMPTS: Final[int] = 5
RETRY_BACKOFF_BASE_S: Final[float] = 30.0
RETRY_BACKOFF_CAP_S: Final[float] = 600.0
FAILED_RATIO_LIMIT: Final[float] = 0.01  # 恒久失敗がこれを超えたら非0で終える（宛先データの異常を疑う）


class Outcome(StrEnum):
    SENT = "sent"
    RETRY = "retry"          # 待てば成功しうる（5xx・429・タイムアウト）
    PERMANENT = "permanent"  # 何度やっても失敗（4xx。宛先不正など）
    RELEASE = "release"      # 時間切れで送らずに手放した（試行回数に数えない）


@dataclass(frozen=True, slots=True)
class Target:
    campaign_id: str
    member_id: int
    email: str
    attempts: int


class MailClient(Protocol):
    def send(self, target: Target) -> Outcome: ...


class HttpMailClient:
    """配信事業者 API。同じ相手への再送は同じ Idempotency-Key になる（事業者が24時間重複排除する前提）。"""

    def __init__(self, url: str, session: requests.Session | None = None) -> None:
        self._url, self._session = url, session or requests.Session()  # 接続を使い回す

    def send(self, target: Target) -> Outcome:
        try:
            resp = self._session.post(
                self._url,
                json={"to": target.email, "member_id": target.member_id, "template": target.campaign_id},
                headers={"Idempotency-Key": f"{target.campaign_id}:{target.member_id}"},  # 試行回数を混ぜない
                timeout=(MAIL_CONNECT_TIMEOUT_S, MAIL_READ_TIMEOUT_S),
            )
        except requests.RequestException:
            return Outcome.RETRY
        if resp.status_code == 200:
            return Outcome.SENT
        if resp.status_code == 429 or resp.status_code >= 500:
            return Outcome.RETRY
        return Outcome.PERMANENT


_CLAIM = """
WITH c AS (
    SELECT member_id FROM campaign_targets
    WHERE campaign_id = %(cid)s AND opted_out_at IS NULL            -- 取る瞬間の最新の配信停止を見る
      AND ((state = 'pending' AND (not_before IS NULL OR not_before <= now()))
           OR (state = 'sending' AND lease_until < now()))          -- 落ちた作業の取り直し
    ORDER BY member_id LIMIT %(n)s
    FOR UPDATE SKIP LOCKED                                          -- 重なって動いても同じ行を取らない
)
UPDATE campaign_targets t
SET state = 'sending', attempts = t.attempts + 1, claim_id = %(claim)s,
    lease_until = now() + make_interval(secs => %(lease)s)
FROM c WHERE t.campaign_id = %(cid)s AND t.member_id = c.member_id
RETURNING t.member_id, t.email, t.attempts
"""

_FINISH = """
UPDATE campaign_targets t
SET state = v.state, sent_at = CASE WHEN v.state = 'sent' THEN now() END,
    attempts = t.attempts - v.release, lease_until = NULL,
    not_before = CASE WHEN v.state = 'pending' AND v.backoff > 0
                      THEN now() + make_interval(secs => v.backoff) END
FROM unnest(%(ids)s::bigint[], %(states)s::text[], %(release)s::int[], %(backoff)s::float8[])
     AS v(member_id, state, release, backoff)
WHERE t.campaign_id = %(cid)s AND t.member_id = v.member_id
  AND t.state = 'sending' AND t.claim_id = %(claim)s                -- 札が違う（取り直された）なら書かない
"""


def claim_batch(conn: psycopg.Connection, campaign_id: str, claim: uuid.UUID,
                *, size: int = BATCH_SIZE, lease_s: float = LEASE_S) -> list[Target]:
    """次に配る相手を最大 size 件取り、'sending' にして返す。"""
    with conn.transaction():
        rows = conn.execute(_CLAIM, {"cid": campaign_id, "n": size, "claim": claim, "lease": lease_s}).fetchall()
    return [Target(campaign_id, m, e, a) for m, e, a in sorted(rows)]


def send_batch(batch: list[Target], mail: MailClient, *, clock: Callable[[], float] = time.monotonic,
               budget_s: float = BATCH_BUDGET_S, min_interval_s: float = 0.0,
               sleep: Callable[[float], None] = time.sleep) -> list[Outcome]:
    """1バッチを順に送る。予算を超えたら残りは RELEASE（送らない）。DB には触らない。"""
    start, outcomes = clock(), []
    for target in batch:
        if clock() - start > budget_s:
            outcomes.append(Outcome.RELEASE)
            continue
        outcomes.append(mail.send(target))
        if min_interval_s:
            sleep(min_interval_s)          # 事業者のレート上限の内側に収める
    return outcomes


def finish_batch(conn: psycopg.Connection, campaign_id: str, claim: uuid.UUID,
                 batch: list[Target], outcomes: list[Outcome]) -> int:
    """結果を1文で記録する。戻り値は実際に書けた件数（札が違えば少なくなる）。"""
    ids, states, release, backoff = [], [], [], []
    for t, o in zip(batch, outcomes, strict=True):
        ids.append(t.member_id)
        release.append(1 if o is Outcome.RELEASE else 0)
        if o is Outcome.SENT:
            states.append("sent"); backoff.append(0.0)
        elif o is Outcome.PERMANENT or (o is Outcome.RETRY and t.attempts >= MAX_ATTEMPTS):
            states.append("failed"); backoff.append(0.0)
        else:  # RETRY（回数に余裕あり）か RELEASE → pending に戻す
            states.append("pending")
            backoff.append(0.0 if o is Outcome.RELEASE else
                           min(RETRY_BACKOFF_CAP_S, RETRY_BACKOFF_BASE_S * 2 ** (t.attempts - 1)))
    with conn.transaction():
        cur = conn.execute(_FINISH, {"cid": campaign_id, "claim": claim, "ids": ids, "states": states,
                                     "release": release, "backoff": backoff})
    return cur.rowcount


@dataclass(slots=True)
class Report:
    batches: int = 0
    sent: int = 0
    stale_writes: int = 0
    remaining: int = 0       # pending / sending のまま残った件数（配信停止を除く）
    failed: int = 0
    total: int = 0
    notes: list[str] = field(default_factory=list)

    @property
    def exit_code(self) -> int:
        if self.remaining:
            return 1          # 時間切れ。起動し直せば続きから（安全）
        if self.total and self.failed / self.total > FAILED_RATIO_LIMIT:
            return 2          # 宛先データの異常の疑い。人が見る
        return 0


def _count(conn: psycopg.Connection, campaign_id: str) -> tuple[int, int, int]:
    return conn.execute(
        "SELECT count(*), count(*) FILTER (WHERE state IN ('pending','sending') AND opted_out_at IS NULL), "
        "count(*) FILTER (WHERE state = 'failed') FROM campaign_targets WHERE campaign_id = %s",
        (campaign_id,)).fetchone()


def _seconds_until_work(conn: psycopg.Connection, campaign_id: str) -> float | None:
    row = conn.execute(
        "SELECT min(extract(epoch FROM (CASE WHEN state = 'pending' THEN coalesce(not_before, now()) "
        "ELSE lease_until END) - now())) FROM campaign_targets "
        "WHERE campaign_id = %s AND state IN ('pending','sending') AND opted_out_at IS NULL",
        (campaign_id,)).fetchone()
    return None if row[0] is None else max(0.0, float(row[0]))


def run(conn: psycopg.Connection, campaign_id: str, mail: MailClient, *, deadline_s: float,
        min_interval_s: float = 0.0, clock: Callable[[], float] = time.monotonic,
        sleep: Callable[[float], None] = time.sleep) -> Report:
    """終わるか deadline_s に達するまで配る。何度呼んでも・同時に呼んでも二重には配らない。"""
    report, start, claim = Report(), clock(), uuid.uuid4()
    while clock() - start < deadline_s:
        batch = claim_batch(conn, campaign_id, claim)
        if not batch:
            wait = _seconds_until_work(conn, campaign_id)
            if wait is None:
                break                                  # 配るものが残っていない
            sleep(min(wait + 0.05, 5.0))               # 再試行の待ち・他の作業の期限を待つ
            continue
        outcomes = send_batch(batch, mail, min_interval_s=min_interval_s)
        written = finish_batch(conn, campaign_id, claim, batch, outcomes)
        report.batches += 1
        report.sent += sum(o is Outcome.SENT for o in outcomes)
        report.stale_writes += len(batch) - written
    report.total, report.remaining, report.failed = _count(conn, campaign_id)
    return report


def main() -> None:
    import os
    import sys

    logging.basicConfig(level=logging.INFO)
    with psycopg.connect(os.environ["DB_DSN"], autocommit=True,
                         options="-c statement_timeout=10000") as conn:
        report = run(conn, os.environ["CAMPAIGN_ID"], HttpMailClient(os.environ["MAIL_URL"]),
                     deadline_s=float(os.environ.get("DEADLINE_S", "3000")),
                     min_interval_s=float(os.environ.get("MIN_INTERVAL_S", "0")))
    logging.info("report: %s", report)
    sys.exit(report.exit_code)


if __name__ == "__main__":
    main()

DB 移行（schema.sql。既存の campaign_targets に足す）

📋 コピー-- 移行（既存の campaign_targets に足す。既存行は state='sent' か 'pending' に寄せる: 下の UPDATE）
ALTER TABLE campaign_targets
  ADD COLUMN IF NOT EXISTS opted_out_at timestamptz,            -- 配信停止の通知を受けた時刻（会員DB側が更新）
  ADD COLUMN IF NOT EXISTS state        text NOT NULL DEFAULT 'pending'
        CHECK (state IN ('pending','sending','sent','failed')),
  ADD COLUMN IF NOT EXISTS attempts     int  NOT NULL DEFAULT 0,
  ADD COLUMN IF NOT EXISTS not_before   timestamptz,            -- 一時障害の再試行を待たせる時刻
  ADD COLUMN IF NOT EXISTS lease_until  timestamptz,            -- 取った作業の期限。切れたら他が取り直せる
  ADD COLUMN IF NOT EXISTS claim_id     uuid;                   -- 取った作業の持ち主の札
UPDATE campaign_targets SET state = 'sent' WHERE sent_at IS NOT NULL AND state = 'pending';
-- 「次に配る相手」だけを引く部分索引（sent が大多数になっても小さいまま）
CREATE INDEX IF NOT EXISTS ix_targets_work ON campaign_targets (campaign_id, member_id)
  WHERE state IN ('pending','sending');

本番では、CHECK は NOT VALID で足して後から VALIDATE CONSTRAINT、索引は CREATE INDEX CONCURRENTLY（トランザクション外）にする。未測定 本問の検証は1万行の表への適用までで、120万行の表でのロック時間は測っていない。

CronJob（cronjob.yaml。kubectl apply --dry-run=server で受理を確認）

📋 コピーapiVersion: batch/v1
kind: CronJob
metadata: { name: spring-campaign-sender, namespace: default }
spec:
  schedule: "0 2 * * *"
  timeZone: "Asia/Tokyo"
  concurrencyPolicy: Forbid          # 前日分が終わっていなければ今日の起動は飛ばす（重なりを減らす。安全性は DB 側の取り方が持つ）
  startingDeadlineSeconds: 1800      # 02:30 までに起動できなければその日は見送る（遅れて朝に配らない）
  successfulJobsHistoryLimit: 3
  failedJobsHistoryLimit: 5
  jobTemplate:
    spec:
      backoffLimit: 2                # 既定は 6。OOM やノード障害の再起動は2回まで（続きから動くので何回でも安全だが、無限に粘らない）
      activeDeadlineSeconds: 4200    # 70 分。アプリ自身は DEADLINE_S=3000（50 分）で止まる。これはハング時の最後の砦
      ttlSecondsAfterFinished: 604800
      template:
        spec:
          restartPolicy: Never       # 失敗した Pod は新しい Pod に置き換える（同じ Pod の再起動で状態を引きずらない）
          containers:
            - name: sender
              image: asia-northeast1-docker.pkg.dev/mops-prod/promo/campaign-sender:2.0.0
              command: ["python", "-m", "campaign_sender"]
              env:
                - { name: CAMPAIGN_ID, value: spring-2026 }
                - { name: DEADLINE_S, value: "3000" }
                - { name: MIN_INTERVAL_S, value: "0.002" }   # 事業者の上限 500 通/秒（仮定）に収める
              resources:
                requests: { cpu: 250m, memory: 128Mi }
                limits: { memory: 256Mi }       # 取る件数が一定なので、会員数が増えても 256Mi で足りる

時間の予算: 120万通 ÷ 500通/秒 = 2,400秒（導出）。DEADLINE_S 3,000秒（50分）の内側に収まる。アプリは最後のバッチを終えたところで止まるので、最大でも 3,000 + 60 + 7 秒。activeDeadlineSeconds 4,200秒はハングしたときだけ効く。事業者の上限が 100通/秒だと 12,000秒かかり、1日では終わらない。その場合は「翌日の起動が続きから配る」（下の(5)・(6)）。

性質テスト

実 PostgreSQL に対して、「何が起きても守られる性質」を10本で固定した（test_campaign_sender.py、全文は 2026-10-03-AB-coding-infra-code/）。主なもの:

📋 コピーdef test_two_workers_claim_disjoint_rows(conn, conn2):          # 同時に取っても被らない
def test_claim_does_not_wait_for_rows_locked_by_another_worker  # 別ワーカーが取得中の行で待たされない（SKIP LOCKED）
def test_opted_out_member_is_never_claimed_or_sent(conn):       # 配信停止は取らない・「残り」にも数えない
def test_crashed_claim_is_retaken_only_after_lease_expires(conn)  # 落ちた作業は期限後にだけ取り直せる
def test_stale_worker_cannot_overwrite_after_reclaim(conn):     # 取り直された後の古い作業は書けない（札）
def test_retry_backs_off_then_fails_after_max_attempts(conn):   # 待ち時間つき再試行 → 5回で failed
def test_budget_exceeded_releases_unsent_without_counting_attempt(conn)  # 時間切れの残りは手放す（回数に数えない）
def test_idempotency_key_is_same_on_every_attempt():            # 冪等キーは試行回数に依存しない

📋 コピー$ uv run --with 'psycopg[binary]' --with requests --with pytest pytest -q test_campaign_sender.py
..........                                                               [100%]
10 passed in 0.47s

実測 テストが効いているかの確認（変異テスト）: 本体を壊して赤になるかを4か所で試した。SKIP LOCKED を外す → test_claim_does_not_wait_... が赤。claim_id の照合を外す → test_stale_worker_... が赤。opted_out_at IS NULL を外す → test_opted_out_... が赤。Idempotency-Key に試行回数を混ぜる → test_idempotency_key_... が赤。いずれも1本だけ赤で、他の9本は緑のまま。 最初は SKIP LOCKED を外しても全部緑だった（test_two_workers_... は自動コミットで取るため、行ロックが重ならない）。重ならない読み方では、SKIP LOCKED がなくても被りは起きない（状態の述語で弾かれる）。SKIP LOCKED が守るのは「待たされないこと」で、それを見るテストを足して初めて赤になった。

実行例（input → output。実測）

検証環境: M1 Mac、Python 3.13.12、Docker（Colima）上の postgres:17-alpine、自作のフェイク配信事業者（fake_mail.py。Idempotency-Key で重複排除。member_id % 25 == 0 は初回だけ 503（4%）、% 997 == 0 は 422（恒久エラー）。本物の事業者・本物の Kubernetes の Pod 削除は使っていない）。1万人、harness.py・harness_out.txt を同梱。

① 6,000通目で kill -9 → 起動し直す 実測

Good の「13」は、落ちた時点でバッチ（200件）のうち送信済み・記録前だった人。上限は BATCH_SIZE = 200 × 落ちた回数で、1回の落下あたり最大200通（導出）。別の回では 22 だった。再開の前に、リースが切れるまで待つ代わりに lease_until を手で過去にしている（実際は最大 90 秒待つ）。

② 2つの Pod が重なって動く 実測

Bad は両方のプロセスが異常終了（exit 1）した。stderr を捨てていたため原因は未確認（接続を1通ごとに作り直すので、一時ポートの枯渇を疑う）。同じ条件の別の回では両方 exit 0 で、9,590人に2通だった（出力は残していない）。どちらの回でも Good は 0。

③ 配信中に500人が配信停止し、事業者が一時障害を返す 実測

④ メモリ（mem.py。件数ごとに別プロセスで測った最大 RSS）実測

Good は 120万件を 77 秒で処理した（DB の取得・更新だけの時間。送信は含まない）。1件あたり約 64 μs で、メール1通（2ミリ秒、500通/秒）の 3% 程度。

⑤ 効果の見積もり 導出

後輩の提案への回答と、迷った案

後輩の提案「メモリ 1Gi、backoffLimit: 0」は採らない。

• メモリ 1Gi は OOM を先送りするだけ。傾き 219 B/件、切片 39 MB から、1Gi は約450万件（導出）。会員が4倍弱になればまた落ちる。メモリを件数に比例させない（取る方式）ほうが、メモリ上限を決める必要自体が無くなる
• backoffLimit: 0 は #4（重なり）を直さない。重なりは Pod の再起動とは別の経路（前日の Job が長引く、CronJob が複数 Job を作る）。さらに、Pod が消えたとき（ノードのアップグレード・Spot 退去）に新しい Pod が作られず、その日の残りが配られない。今の設計なら、backoffLimit を何回にしても安全なので、続きから動く回数を残すほうが得
• 後輩の案が実際に直すのは #1〜#7 のうち #2 の一部だけ。ただし「メモリ上限と再実行回数を意識して決める」という発想は正しいので、根拠のある数字（256Mi・backoffLimit: 2）として残した

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

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

ロールバック: コード（campaign-sender:2.0.0 → 1.8.0）は戻せるが、DB の列と部分索引は残す（古いコードは新しい列を読まないだけ）。ただし 1.8.0 に戻すと、問題#1〜#6 がすべて戻る。戻す前に CronJob を suspend: true にして、手で配信するほうがよい。

ポイント解説

1. 「再実行されない」を目指さず、「何度動いても同じ」にする: CronJob は重なる・二重に作られる・Pod が作り直される。backoffLimit: 0 は一部の経路を塞ぐだけ。取る→送る→記録するを、取った札つきの行で回し、事業者には冪等キーを付けると、起動のされ方を問わなくなる
2. 全件を読まない: fetchall はメモリが件数に比例する（実測 219 B/件）。少しずつ取る方式は、メモリが一定（約 53 MB）で、配信停止のような最新の状態を取る瞬間に見られる副産物も得る
3. 失敗は2種類に分け、数え、終了コードに出す: 握り潰した print（第10章）は、Job を毎日 Succeeded にしたまま 4% を永久に失った。待てば直る失敗には間隔つき再試行、恒久失敗は failed として数え、割合が大きければ非0で終える
4. テストは壊して確かめる: SKIP LOCKED を外しても最初の9本が全部緑だった。「何を守るためのコードか」（待たされないこと）を見るテストを足して、初めて赤になった。テストが緑なのは、コードが正しいからとは限らない
5. Kubernetes の期限は2層: アプリ自身が止まる時刻（DEADLINE_S）と、ハングしたときの最後の砦（activeDeadlineSeconds）を分け、前者を後者の内側に置く。止まった後の続きは、次の起動が引き受ける

実務への応用

• 自チームの他のバッチにも同じ3点検査を当てる: ① SELECT に「未処理」の条件があるか（再起動すると頭からにならないか）、② fetchall や巨大なリストを作っていないか、③ 失敗を print して終了コード 0 で終えていないか。クーポン発行・ポイント付与・請求の締め処理が同じ形をしていないか見る
• Argo Workflows のステップでも同じ: ステップの再実行（retryStrategy）は、副作用のあるステップでは必ず冪等キーとセットにする
• メール以外の外部 API（push 通知・決済・在庫引当）にも Idempotency-Key を付ける。IETF の draft（Idempotency-Key ヘッダ）は、この慣行を標準にしようとしている
• 配信停止は「反映の遅れ」も指標にする: 本バッチは取る瞬間に見るが、会員DBから campaign_targets.opted_out_at への反映が遅ければ意味が薄い。反映の遅延（P95）をダッシュボードに置く
• dbt 側の検査: 配信ログを BigQuery に取り込んでいるなら、(campaign_id, member_id) の unique テスト（配信回数=1）を翌朝のバッチで走らせると、事業者側の重複排除が効いていない日に気づける

扱わなかった範囲・一次資料

• 本物の Kubernetes での Pod 削除試験・CronJob の実行 — manifest は kubectl apply --dry-run=server（Kubernetes 1.37 の kind）で受理されることまでしか確認していない。重なりや Pod の再作成は、Kubernetes 公式文書の記述（参照日 2026-10-03）に依拠した。本番前に、ステージングで Pod を kill して配信数を数える
• 本物の配信事業者 — フェイクで再現した。Idempotency-Key の対応・保持期間（24時間）・レート上限は仮定で、事業者の仕様書で確認する
• 配信停止の反映遅延の法的な許容範囲 — 本バッチは「取る瞬間」の最新を見るところまで。通知を受けてから何分以内に止めれば足りるかは、法務に確認する（条文は「意思に反して送信してはならない」）
• 同意の記録の保存（同法第3条第2項） — 本バッチの範囲外
• 120万行の UPDATE による表の肥大（dead tuple）・VACUUM — 1回の配信で全行が更新される。<測っていない>。本番では autovacuum の設定と、配信後の表の大きさを確認する
• BigQuery から campaign_targets への取り込み — 配信対象は投入済みとした
• メール本文の生成・件名のA/Bテスト・開封率の計測 — 範囲外
• 世界の実物との対応: 本解答の取り方（claim + lease + SKIP LOCKED）は、PostgreSQL をキューにする実装（Que、GoodJob、Oban、River など）の中核と同じ。各実装が持つ中核の概念（claim・lease・再試行の待ち・失敗状態・優先度・スケジュール・ユニーク制約・可視化）のうち、本解答が扱うのは claim・lease・再試行の待ち・失敗状態の4つ。優先度・スケジュール・ユニーク制約・可視化は扱っていない。割合は各実装の文書を読んで数えたものではなく、概念を私が列挙しただけなので、「8割」のような数字は付けない

一次資料（参照日 2026-10-03）

• Kubernetes「CronJob」 https://kubernetes.io/docs/concepts/workloads/controllers/cron-jobs/ — 引用: "a single CronJob can create multiple concurrent Jobs"。concurrencyPolicy は Allow が既定。startingDeadlineSeconds を指定しないと期限なし
• Kubernetes「Jobs」 https://kubernetes.io/docs/concepts/workloads/controllers/job/ — Pod が失敗したり削除されたりしたとき、Job は新しい Pod を始める。backoffLimit の既定は 6（kubectl explain job.spec.backoffLimit で確認、v1.37）
• PostgreSQL 17「SELECT — The Locking Clause」 https://www.postgresql.org/docs/17/sql-select.html — 原文: "can be used to avoid lock contention with multiple consumers accessing a queue-like table"。「一貫しない見え方を与える」ので汎用の処理には向かない、との注意付き
• e-Gov 法令検索「特定電子メールの送信の適正化等に関する法律」第3条第3項 https://laws.e-gov.go.jp/law/414AC0100000026 — 送信しないように求める旨の通知を受けたときは、その通知に示された意思に反して特定電子メールの送信をしてはならない（API で条文を取得して確認）
• IETF「The Idempotency-Key HTTP Header Field」（draft-ietf-httpapi-idempotency-key-header）— 冪等キーの慣行
• 仙塲大也『良いコード／悪いコードで学ぶ設計入門』改訂新版 第4章・第10章・第13章

今日のまとめ

CronJob のバッチは「再実行を防ぐ」ではなく「何度動いても同じ」にする。少しずつ取って（SKIP LOCKED＋リース＋札）、冪等キーつきで送り、失敗を2種類に分けて数えれば、重なりも・途中の死も・配信停止の到着も、数字で守れる。

次のステップ

• 発展問題: 事業者の上限が 100通/秒になり、120万通が1日で終わらなくなった。配信を2〜3日にまたがらせる運用（優先度の高い会員から・配信窓を深夜に限る）、DEADLINE_S と CronJob の頻度（毎晩 → 6時間おき）の設計、事業者の suppress list との二重管理を、重複と取りこぼしの数で比較せよ
• 参考: idempotency key、lease（リース）、fencing token（札）、work queue on PostgreSQL、FOR UPDATE SKIP LOCKED、CronJob concurrencyPolicy、特定電子メール法のオプトアウト

自己評価（あとで記入）