# 無人バッチの「黙った空振り」を発見して止める — 冪等取込の設計と検証 定期実行のバッチ（フォーム回答・受信メール・外部APIレコードの取込）が、 **エラーを1件も出さないまま、毎回同じ仕事をやり直している**ことがある。 ログは毎回 `ok: true` で、監視も通知も鳴らない。気付かれないまま CI/CD の課金・外部APIのクォータ・LLM の従量課金だけが積み上がる。 この指示書は「その状態を見つける方法」「典型的な真因」「壊さずに直す設計」「効果の実測」をまとめたもの。 AI に読ませればそのまま診断〜修正まで実行できる粒度で書いてある。 --- ## 1. まず疑う: 「複数 run の出力が完全に同一かどうか」 単発のログは正常に見える。**連続する実行の出力を並べる**のが唯一の確実な検出法。 ```bash # 例: CI の直近8回の実行から、バッチの結果 JSON だけを抜いて並べる for id in $(gh run list --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: []` は「正常」の証拠にならない。**握り潰されたエラーは配列に入らない**。 ### 裏取り（推測で直さないための必須手順） ソース件数と「取込済みとして記録されている件数」を突き合わせる。 ```bash # 例: PostgREST で「取込済みID を持つ実体」の件数を数える curl -sS -I "$API/rest/v1/?select=id&=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. 典型的な真因: 「取込済み集合」を実体テーブルの単一カラムで持っている よくある実装: ```sql 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; -- 「冪等のため」に張られがち ``` ```ts // 取込済み集合を、実体テーブルの単一カラムから作る 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. 台帳テーブル ```sql 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. 一時失敗には必ず打ち切りを付ける 一時失敗を無条件に再試行すると、**恒久的に壊れた入力**（削除された添付ファイル・非対応フォーマット）で 今度はそこが永久ループになる。**回数上限で打ち切る**。 ```ts const MAX_ATTEMPTS = 3; // 恒久的に壊れた入力で永久ループしないための打ち切り // 台帳から「もう触らなくてよい集合」を作る（純粋関数にしてテストする） export function ingestedIds( rows: ReadonlyArray<{ source_record_id: string; outcome: string; attempts: number }>, maxAttempts: number, ): Set { 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 行）に当たると **黙って途中までしか読めず、読めなかった分が「未取込」になって再処理が復活する**。 必ずページングして全件読む。 ```ts const done = new Set(); 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. テストは純粋関数に切り出して固定する 外部認証を含む本体は分解しない。**判定ロジックだけ**を抜き出してテストする。 ```ts export function selectPending( records: T[], ingested: ReadonlySet): T[]; export function isTerminal(outcome: string): boolean; export function ingestedIds(rows, maxAttempts): Set; ``` 最低限のケース: 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 は**ジョブごとに分単位で切り上げ**課金する。だから **数秒〜数十秒のジョブを高頻度で回すのが一番損**をする。 ```bash gh run list --workflow .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/:` で**実物が入ったか**を見る --- ## 8. 実行手順まとめ 1. 連続する run の出力を並べ、**数値が動かないこと**を確認する 2. ソース総数 − 記録済み件数 = 滞留件数 の一致で真因を確定する（推測で直さない） 3. 台帳テーブルを作り、**既存カラムからバックフィル**する 4. 判定ロジックを純粋関数へ切り出し、回帰テストを書く 5. 終局のみ即時記録／一時失敗は再試行＋回数打ち切り／`ignoreDuplicates` を使い分ける 6. migration を先に適用 → コードをデプロイ（逆順でも壊れないフォールバックを入れておく） 7. 本番エンドポイントを連打して**空振りゼロへの収束**を確認する 8. 定期実行の**課金分**が落ちたことを実測する。金額は次の請求サイクルで確定させる ---