中断したバッチを最後まで進める

一言でいうと

チェックポイントは、画面に表示するパーセントではなく、承認された一覧のどの部分までを、業務上の変更と監査も含めて確定したかを示す約束です。

なぜ必要なのか

エイリアン・デザート・フェスティバルが、豪雨で中止になりました。顧客は、未決済の注文40件をキャンセルしてほしいと言っています。前のレッスンでは、すべて成功するか、すべてロールバックする変更を作りました。今回は、担当者が「10件ずつ確定してよい。後ろで問題が起きても、すでに終わった10件は維持してほしい」と承認しました。この一言で、トランザクションの境界が変わります。性能のために開発者が勝手に分けたわけではありません。

一度に長くロックすると、ほかの担当者が注文を処理しにくくなります。逆に1行ごとにコミットすると、チャンクの5行目で問題が起きても、前の4行は元に戻せません。適切なサイズは、DBの速度だけでは決まりません。部分的な完了を顧客が理解できるか、どの地点で止めても一貫した説明ができるか、というところから問います。このラボの10件は教育用の選択であり、すべての運用システムへの推奨値ではありません。

どう動くのか

1. 承認一覧と実行結果を分ける

登録するときに、job_id・tenant・各注文のid・revision・qty・chunk_sizeを保存します。承認一覧はID順に正規化し、実行中は変更しません。同じジョブIDで、一覧の順序だけを入れ替えて再登録した場合は、同じリクエストです。同じIDに別の顧客・数量・チャンクサイズを付けたら、衝突です。黙って上書きすると、以前の進捗が新しい一覧を指すことになります。

実行中に、現在のpending注文を検索し直して一覧を作ったら、どうなるでしょうか。最初のチャンクが終わったあとに新しい注文が入ると、最初はなかった注文がキャンセル対象に混ざるおそれがあります。注文が削除されると、OFFSETの意味も変わります。そこでここでは、WHEREの結果の何番目の行かではなく、凍結した承認配列のnext_indexを使います。データが40件から39件に変わっても、承認そのものを39件に書き換えません。

2. ジョブの行をロックして、次のチャンクを選ぶ

2つのワーカーが同時にnext_index=10を読むと、どちらも2つ目のチャンクを取れてしまいます。jobsの該当の行をSELECT FOR UPDATEで読み、トランザクションが終わるまで所有すれば、同じジョブの次の区間の選択を直列化できます。PostgreSQLの行ロックはトランザクションの終了時に解除され、通常の参照すべてを止める全体のロックではありません。公式の行ロックのドキュメントを参照してください。

この設計では、2つのワーカーが同じジョブの別々のチャンクを同時に処理することはありません。2つのワーカーは、障害後の引き継ぎと、重複した選択の防止を確認するためのものです。さらに大きな並列処理のスループットが必要なら、チャンクごとのリース・期限切れ・所有権の世代などを、別に設計する必要があります。現在のコードにそうした保証があるとは説明しないでください。

3. 1つのチャンクの中で、3つの記録をまとめる

チャンクが11–20番目の注文なら、次の変化が1つのトランザクションです。

保存先	確定する内容	別々にコミットすると起きる問題
orders	承認バージョンのpendingをcancelledに変え、バージョンを増やす	注文だけが変わり、再開位置はそのままになる
job_audit	順番・ID・前後のバージョン・数量	監査だけがあり、実際の注文は未変更になる
jobs.next_index	次の区間の開始である20	実行していない注文を完了と誤認する

UPDATEの条件には、顧客・ID・バージョン・数量・状態をすべて入れます。RETURNINGは、実際に変更された行の新しい値を返します。一致する行がないときにSQL自体がエラーを出すわけではないので、プログラムがConflictに変えて、チャンク全体をロールバックする必要があります。UPDATEの公式の説明の戻り値と、影響を受けた行数を、あわせて読んでください。

Pythonの低レベルの関数もトランザクションを使いますが、外側のrun_chunkの中から呼ばれたときに、全体を早期にコミットしてはいけません。psycopgの入れ子のtransactionコンテキストは、SAVEPOINTとして動作します。このレッスンは、autocommit=Trueの接続で、外側のトランザクションの境界を明示的に開く方式を使います。psycopgのトランザクションの案内と、自分のコードで最も外側のコンテキストがどこかを、照らし合わせてください。

4. 再開の前に進捗を疑う

next_indexが20だからといって、すべてを信じてはいけません。監査の順番とID・前後のバージョン・数量が、承認配列の先頭20個と正確に一致している必要があります。20行あるという事実だけでは、別のIDが紛れ込んでいるかどうかはわかりません。このラボは、監査のない進捗、チャンクの途中を指す進捗、一覧の長さを超えた進捗を拒否します。壊れた記録を自動で「修復」して、注文をさらに変更することはしません。

最初のチャンクが確定したあと、2つ目のチャンクの15番目の注文が別の担当者に変更されたら、2つ目のチャンクの11–14番目の変更も、まとめてロールバックします。最初のチャンクの記録と、15番目のその後の変更は維持します。新しいrevisionを読んで、その場で承認を作り直すのは、再試行ではなく新しい業務上の判断です。中断して、残りの範囲を説明したうえで、新しい承認を受ける必要があります。

現場での姿

応答を失ったときにもう一度呼ぶと、何が返るのか

チャンクのコミットは成功したのに、クライアントが結果を受け取る前に終了したとします。同じジョブを再開すると、いま終わったチャンクの答えを再生する代わりに、次のチャンクを処理します。このAPIの契約が、「チャンクNの実行リクエスト」ではなく、「このジョブの次の未完了のチャンクを進める」ものだからです。したがって、返されたprocessedだけを合算して、ジョブ全体の完了の証拠にしてはいけません。全体の履歴は、DBの監査とチェックポイントから読みます。

max_chunksは、1回のdrain呼び出しが試行するチャンクの呼び出し回数です。各呼び出しのSQLの時間の上限や、全体の経過時間とは同じではありません。ほかのワーカーが大部分を終えていれば、自分の呼び出しは、処理したIDが空の完了応答を受け取ることもあります。正常に完了していればすぐに止まり、衝突や接続エラーを、無条件にもう一度試すことはしません。前のレッスンで学んだエラーの分類を忘れないでください。

完了と現在の状態を区別して、顧客に説明する

過去に完了した注文が、あとで再び変更されたり削除されたりしても、そのときに完了したという事実は消えません。報告書は、承認ID・確定ID・未処理IDを分け、確定IDの現在の状態をmatching・drifted・missingに分けます。同じ数量の別の注文が見つかったからといって、missingを消してはいけません。進捗と現在の行を1回のSELECTで読めば、Read Committedの文単位のスナップショットで照合できます。分離レベルの公式ドキュメントを参考にしてください。

FDEは、顧客の業務条件を実装の不変条件に変え、失敗したときにどこまで確定したのかを説明する必要があります。確認したPalantir FDEの求人は、顧客の課題の理解と、実際の解決策の実装を強調しています。この架空の事例は、その能力を練習するための著者の設計であり、この求人がPostgreSQLやこの実装を必須としているという意味ではありません。

次のラボですること

40件の承認と10件のチャンクから始めて、別の顧客・不連続なID・最後の短いチャンクでも動くようにします。実際のクライアントを、注文の変更のあと・監査のあと・チェックポイントのあと・コミットのあとの4か所で終了させ、新しい接続で記録を確認します。2つのプロセスが同じジョブを引き継いで進め、承認した正確なIDが1回ずつだけ処理されたかを、独立したSQLで照合します。

前提も明確にしてください。注文の通常の書き込みはrevisionを増やし、確定した承認と監査は変更しません。ラボのDBアカウントへの悪意のある直接の変更まで防ぐ権限設計は、別です。サーバーは生かしたままクライアントだけを終了するので、サーバーの電源障害への耐久性・外部決済の返金・無制限のスループットまで検証したわけではありません。