この記事のテーマは PostgreSQLの FOR UPDATE SKIP LOCKED です。URLに残っている github-actions-concurrency は公開時の命名ミスですが、公開済みリンクを維持するためURLは変更していません。
DBテーブルを簡易ジョブキューとして使い、複数ワーカーが同時に処理対象を取りに行くと、同じ行を複数ワーカーが拾う競合が起きます。
PostgreSQLではFOR UPDATE SKIP LOCKEDを使うと、他トランザクションがすでにロックしている行を待たずに飛ばせます。
基本例
例えば次のようなジョブテーブルを考えます。
CREATE TABLE jobs (
id bigserial PRIMARY KEY,
status text NOT NULL,
created_at timestamptz NOT NULL DEFAULT now()
);
ワーカーはトランザクション内で1件取得します。
BEGIN;
SELECT id
FROM jobs
WHERE status = 'pending'
ORDER BY created_at
FOR UPDATE SKIP LOCKED
LIMIT 1;
選ばれた行はそのトランザクションからロックされます。
別ワーカーが同じSQLを実行しても、ロック済みの行を待たずに次の候補へ進みます。
取得と状態更新を同じトランザクションにする
SELECTした後、同じトランザクション内でprocessingへ変更します。
UPDATE jobs
SET status = 'processing'
WHERE id = :id;
COMMIT;
取得と更新を別トランザクションにすると、その間に別ワーカーから見える状態になり、競合を作りやすくなります。
CTEで1クエリにまとめる
取得と更新をまとめる方法もあります。
WITH picked AS (
SELECT id
FROM jobs
WHERE status = 'pending'
ORDER BY created_at
FOR UPDATE SKIP LOCKED
LIMIT 1
)
UPDATE jobs
SET status = 'processing'
FROM picked
WHERE jobs.id = picked.id
RETURNING jobs.*;
ワーカー側は返ってきた行を処理対象にできます。
SKIP LOCKEDが向く場面
SKIP LOCKEDは、厳密な順序より「空いている仕事を複数ワーカーで並列処理したい」ケースに向いています。
一方、ロック中の行を飛ばすため、完全なFIFOを保証したい処理には注意が必要です。
長時間トランザクションにしない
実際の重い処理までDBトランザクション内で行うと、ロックを長時間保持します。
通常は、
- 短いトランザクションでジョブを確保して
processingへ変更 - COMMIT
- 外部API呼び出しなどの本処理
- 成功なら
done、失敗なら再試行可能な状態へ更新
という形にします。
ワーカー異常終了への対策
processingへ変更した直後にワーカーが落ちると、そのジョブが残ります。
そのため、started_atやlease期限を持たせ、一定時間以上処理中のジョブを再取得できるようにすると復旧しやすくなります。
まとめ
PostgreSQLを小規模なジョブキューとして使う場合、FOR UPDATE SKIP LOCKEDを使うと複数ワーカーが同じ行を取り合う競合を抑えられます。
ただし、ロックはジョブ確保の短い区間だけにし、異常終了したジョブを回収する仕組みも合わせて設計するのが重要です。
参考
- PostgreSQL: SELECT / locking clause
- PostgreSQL: Explicit Locking