無人バッチの「黙った空振り」を発見して止める — 冪等取込の設計と検証

定期実行のバッチ（フォーム回答・受信メール・外部APIレコードの取込）が、 エラーを1件も出さないまま、毎回同じ仕事をやり直していることがある。 ログは毎回 ok: true で、監視も通知も鳴らない。気付かれないまま CI/CD の課金・外部APIのクォータ・LLM の従量課金だけが積み上がる。

この指示書は「その状態を見つける方法」「典型的な真因」「壊さずに直す設計」「効果の実測」をまとめたもの。 AI に読ませればそのまま診断〜修正まで実行できる粒度で書いてある。

---

1. まず疑う: 「複数 run の出力が完全に同一かどうか」

単発のログは正常に見える。連続する実行の出力を並べるのが唯一の確実な検出法。

# 例: CI の直近8回の実行から、バッチの結果 JSON だけを抜いて並べる
for id in $(gh run list --workflow <workflow>.yml --event schedule \
              --status success --limit 8 --json databaseId -q '.[].databaseId'); do
  echo "$id  $(gh run view "$id" --log 2>/dev/null | grep -oE '\{"ok":true[^}]*\}' | head -1)"
done

判定基準:

出力	意味
件数が毎回減る／増える	正常（進捗している）
8回連続で完全に同じ数値	進捗ゼロ。空振りが確定
inserted: 0 なのに updated: N が毎回同じ	同じレコードを毎回書き戻している

errors: [] は「正常」の証拠にならない。握り潰されたエラーは配列に入らない。

裏取り（推測で直さないための必須手順）

ソース件数と「取込済みとして記録されている件数」を突き合わせる。

# 例: PostgREST で「取込済みID を持つ実体」の件数を数える
curl -sS -I "$API/rest/v1/<entity>?select=id&<ingest_id_col>=not.is.null" \
  -H "apikey: $KEY" -H "Authorization: Bearer $KEY" \
  -H "Prefer: count=exact" -H "Range: 0-0" | grep -i content-range
# => Content-Range: 0-258/259

ソース総数 − 記録済み件数 = 毎回の滞留件数 がピタリ一致したら、真因は次章のパターン。

---

2. 典型的な真因: 「取込済み集合」を実体テーブルの単一カラムで持っている

よくある実装:

alter table entities add column source_record_id text;
create unique index entities_source_uidx on entities (source_record_id)
  where source_record_id is not null;   -- 「冪等のため」に張られがち

// 取込済み集合を、実体テーブルの単一カラムから作る
const { data } = await db.from("entities").select("source_record_id").not("source_record_id","is",null);
const done = new Set(data.map(r => r.source_record_id));
const pending = sourceRecords.filter(r => !done.has(r.id));

なぜ壊れるか

取込処理はたいてい名寄せ（dedup）を伴う。「会社名＋メールが一致したら既存実体に統合」など。 すると N 件のソースレコードが 1 実体に集約される。 しかしカラムは1実体あたり1値しか持てない。だから

• 記録できるのは最後の1件だけ
• 残り N−1 件は次回also「未取込」に戻る
• 既存更新パスが毎回 source_record_id を上書きすると、2件が互いを追い出し合う無限フリップフロップになる
• UNIQUE 違反が起きても update の戻り値を見ていなければ黙って消える

結果、滞留分は永久に、毎回フルコストで再処理される（画像OCR・LLM呼び出し・外部APIダウンロードを含む）。

原則: 「どのソースレコードを消化したか」は、実体とは別の主キー空間で持たなければならない。 実体側のIDカラムは「この実体はこの経路由来か」の表示用として残してよいが、冪等判定には使わない。

---

3. 直し方: 取込台帳（ledger）

3-1. 台帳テーブル

create table if not exists source_ingested (
  source_record_id text primary key,      -- ソース側のIDが主キー。1実体1件の制約から切り離す
  entity_id        uuid references entities(id) on delete set null,
  outcome          text not null,
  attempts         integer not null default 1,
  processed_at     timestamptz not null default now()
);

create index if not exists source_ingested_processed_at_idx
  on source_ingested (processed_at desc);

-- 【必須】既存カラムからバックフィル。忘れると移行直後に全件を再処理する
insert into source_ingested (source_record_id, entity_id, outcome)
select source_record_id, id, 'backfill' from entities
where source_record_id is not null
on conflict (source_record_id) do nothing;

3-2. 「終局」だけを記録する

記録してよいのは恒久的に決まった結果だけ。

outcome	記録する？	理由
inserted / updated	✅	成功。完了
excluded_*（業務ルールによる恒久除外）	✅	何度評価しても同じ結論
unusable（入力に必要項目が無い）	✅	入力が変わらない限り同じ
deferred（今回のバッチ上限に達した繰り越し）	❌	次回必ず処理させる
一時失敗（HTTP 5xx / レート制限 / ダウンロード失敗）	❌ ただし打ち切りあり	次回成功しうる
error（DB書き込み失敗など）	❌	次回再試行

一時失敗を終局として記録してはいけない。それをやると、たまたま1回失敗しただけのレコードが 二度と取り込まれず、業務データが静かに欠損する（顧客リード・注文などが消える）。

3-3. 一時失敗には必ず打ち切りを付ける

一時失敗を無条件に再試行すると、恒久的に壊れた入力（削除された添付ファイル・非対応フォーマット）で 今度はそこが永久ループになる。回数上限で打ち切る。

const MAX_ATTEMPTS = 3;   // 恒久的に壊れた入力で永久ループしないための打ち切り

// 台帳から「もう触らなくてよい集合」を作る（純粋関数にしてテストする）
export function ingestedIds(
  rows: ReadonlyArray<{ source_record_id: string; outcome: string; attempts: number }>,
  maxAttempts: number,
): Set<string> {
  return new Set(
    rows.filter(r => r.outcome !== "transient_failed" || r.attempts >= maxAttempts)
        .map(r => r.source_record_id)
  );
}

失敗時は attempts を +1 して upsert し、上限に達したらログの文言を変えて可視化する （... failed 3 times, giving up）。打ち切った件数は必ず表に出す。黙って捨てない。

3-4. 台帳への書き込みは「1件処理するたびに即時」

末尾で一括 insert すると、実行時間上限（サーバーレス関数の maxDuration 等）で途中終了した瞬間に 全部失われ、再処理が再発する。1件ずつ書く。台帳の書き込み自体が失敗したら、 エラー配列に積んで処理は続行する（throw でバッチ全体を止めない）。

---

4. 踏みやすい落とし穴

4-1. upsert(..., { ignoreDuplicates: true }) は既存行を更新しない

PostgREST の resolution=ignore-duplicates（= ON CONFLICT DO NOTHING）は既存行に一切触らない。 台帳に「途中状態の行」（transient_failed, attempts=1）を書いたあとで処理が成功しても、 終局の書き込みが ignoreDuplicates だと行は失敗のまま残り続ける。 → そのレコードは打ち切り上限まで無駄に再処理される。

ignoreDuplicates は「一度書いたら二度と変わらない行」専用。 途中状態を持つ台帳の終局書き込みは必ず merge（既存行を上書き）にする。

4-2. merge にすると payload の null が既存値を消す

上書き型 upsert では、値が分からないキーは payload に入れない（null を入れない）。 entity_id: null を送ると、以前記録した紐付けが消える。

4-3. エラーハンドリングの強度変更はスコープ外の仕様変更

「握り潰し ⇄ throw」の変更はバッチの挙動そのものを変える。 不正データ1件で全体を止める実装にすると、以後その1件が直るまで全レコードが取り込まれない。 握り潰しを直すときは「エラー配列に積んで continue」にとどめ、throw でバッチを止めない。

4-4. 一覧取得のページング上限

台帳は増え続ける。API の既定行数上限（PostgREST は既定 1000 行）に当たると 黙って途中までしか読めず、読めなかった分が「未取込」になって再処理が復活する。 必ずページングして全件読む。

const done = new Set<string>();
for (let from = 0; ; from += 1000) {
  const { data, error } = await db.from("source_ingested")
    .select("source_record_id, outcome, attempts").range(from, from + 999);
  if (error) { errors.push(`ledger read failed: ${error.message}`); break; }
  for (const id of ingestedIds(data ?? [], MAX_ATTEMPTS)) done.add(id);
  if ((data ?? []).length < 1000) break;
}

台帳が読めないときのフォールバックは「従来カラムとの和集合」にしておくと、 migration 未適用の状態でデプロイしても壊れない（効果が出ないだけ）。 ただしその失敗は必ずエラー配列に出す。黙って落ちると全件再処理に戻ったことに気付けない。

---

5. テストは純粋関数に切り出して固定する

外部認証を含む本体は分解しない。判定ロジックだけを抜き出してテストする。

export function selectPending<T extends { id?: string|null; createdAt?: string|null }>(
  records: T[], ingested: ReadonlySet<string>): T[];
export function isTerminal(outcome: string): boolean;
export function ingestedIds(rows, maxAttempts): Set<string>;

最低限のケース:

1. 取込済みIDを除外する
2. 処理順（新しい順など既存の順序）を保つ ← 既存挙動の固定
3. IDが無いレコードを落とす
4. isTerminal が一時失敗と繰り越しにだけ false を返す
5. 回帰: 同一実体に dedup される2件が両方台帳に入れば、2回目は空になる
6. attempts が上限未満なら未処理／上限到達で処理済みになる

---

6. 効果の実測（見積りで完了と言わない）

6-1. 収束を実物で見る

修正版をデプロイしたら、本番のエンドポイントを続けて叩き、数値が0へ落ちるのを確認する。

run1: {"inserted":0,"updated":15,"skipped":34,"deferred":0}  50s   ← 滞留を消化中
run2: {"inserted":0,"updated":13,"skipped":0, "deferred":0}  44s
run3: {"inserted":0,"updated":13,"skipped":0, "deferred":0}  36s   ← 一時失敗分が残る
run4: {"inserted":0,"updated":0, "skipped":0, "deferred":0}   3s   ← 打ち切り到達・空振りゼロ

skipped が消える＝恒久除外が台帳に載った証拠。所要時間が桁で落ちたら成功。

6-2. CI の課金は「分単位切り上げ」で効く

GitHub Actions はジョブごとに分単位で切り上げ課金する。だから 数秒〜数十秒のジョブを高頻度で回すのが一番損をする。

gh run list --workflow <wf>.yml --event schedule --limit 20 \
  --json createdAt,updatedAt,conclusion -q '.[] | "\(.conclusion) \(.createdAt) \(.updatedAt)"' \
| while read -r c s e; do
    d=$(( $(date -u -d "$e" +%s) - $(date -u -d "$s" +%s) ))
    echo "$c ${d}s -> billed $(( (d+59)/60 ))min"
  done

120秒の境界に張り付いているジョブは狙い目。数十秒削るだけで課金が 3分 → 1分 になる。

あわせて有効なのが同一 cron 式のワークフローのステップ統合。起動オーバーヘッドが1回に集約され、 実行回数を変えずに丸ごと削れる。ただし2点注意:

• cron 式が完全一致するワークフローにだけ寄せる。*/15 に hourly を「分ゲート」で混ぜるのは不可 （スケジューラは高頻度 cron を大量に間引き、発火時刻の分がランダムになるため実行回数が激減する）
• 同一ジョブに集約すると、既定では前のステップが失敗した時点で後続が skip される。 独立していた頃の耐障害性を保つため、各ステップに if: ${{ !cancelled() }} を付ける （ジョブ全体は赤いままにして、失敗は隠さない）

6-3. 判定は「定期実行の成功」で行う

手動トリガ（workflow_dispatch）が通っただけで直ったと言わない。 --event=schedule の成功と、そのステップが実際に走ったログで判定する。

---

7. デプロイ先を間違えない

「どのブランチが本番を配信しているか」を先に確認する。 CLI デプロイ運用だと、既定ブランチにマージした変更が本番に出ないことがある。

• pull_request の CI ワークフローは base ブランチのファイルが使われる。 CI を既定ブランチにだけ置くと、本番を配信する別ブランチが完全に無検証になる
• 定期実行（schedule）は既定ブランチの定義しか発火しない。 他ブランチに同名の cron 定義が残っていても動かない（「二重発火している」という誤診断の原因）
• 「PR が Merged」を完了判定にしない。git show origin/<branch>:<path> で実物が入ったかを見る

---

8. 実行手順まとめ

1. 連続する run の出力を並べ、数値が動かないことを確認する
2. ソース総数 − 記録済み件数 = 滞留件数 の一致で真因を確定する（推測で直さない）
3. 台帳テーブルを作り、既存カラムからバックフィルする
4. 判定ロジックを純粋関数へ切り出し、回帰テストを書く
5. 終局のみ即時記録／一時失敗は再試行＋回数打ち切り／ignoreDuplicates を使い分ける
6. migration を先に適用 → コードをデプロイ（逆順でも壊れないフォールバックを入れておく）
7. 本番エンドポイントを連打して空振りゼロへの収束を確認する
8. 定期実行の課金分が落ちたことを実測する。金額は次の請求サイクルで確定させる