Postgres をジョブキューにする：FOR UPDATE SKIP LOCKED の正しい使い方

バックグラウンドジョブを動かすのに、Redis もメッセージブローカーも必要ありません。Postgres のテーブルと FOR UPDATE SKIP LOCKED があれば、各ジョブを一度に 1 つのワーカーだけに渡し、クラッシュにも耐え、ジョブをそれが属するデータと同じトランザクションに含めることができます。ただし、最初に思いつく素直な実装は間違っています。私たちのテストでは、単純に「いちばん古いジョブを選んでから更新する」処理を 8 つのワーカーで実行したところ、同じ行が最大 8 回取得され、10,000 件のジョブのうち完了したのは約 2,100 件でした。この記事では、このバグを再現し、それを修正するキュー（取得、完了、リトライ、回収）を組み立て、あとから問題になる部分を取り上げます。作業中に開いたままのトランザクション、LISTEN/NOTIFY、VACUUM、そしてアイドル時にサスペンドするコンピュートに対してポーリングするワーカーが与える影響です。以下のステートメントと出力はすべて、Relyt One が採用しているメジャーバージョンであるローカルの PostgreSQL 17.11 で実行したものです。

テーブルは 1 つで、1 行が 1 ジョブです。部分インデックスはワーカーが探す行だけを対象にしているので、テーブルに完了済みジョブの長い履歴があっても小さいままです。

CREATE TABLE jobs (
  id           bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  queue        text        NOT NULL DEFAULT 'default',
  payload      jsonb       NOT NULL,
  status       text        NOT NULL DEFAULT 'queued'
               CHECK (status IN ('queued', 'running', 'done', 'failed')),
  run_at       timestamptz NOT NULL DEFAULT now(),
  attempts     int         NOT NULL DEFAULT 0,
  max_attempts int         NOT NULL DEFAULT 5,
  locked_until timestamptz,
  last_error   text,
  created_at   timestamptz NOT NULL DEFAULT now()
);

CREATE INDEX jobs_ready_idx ON jobs (queue, run_at) WHERE status = 'queued';

INSERT INTO jobs (payload)
SELECT jsonb_build_object('email_id', g) FROM generate_series(1, 10) AS g;

run_at はリトライのスケジュールも兼ねています。ジョブが queued で、かつ run_at <= now() のとき、そのジョブは実行可能です。locked_until はリースで、これについては後で説明します。

多くの人が最初に書くのがこのバージョンです。1 つのステートメントなので、アトミックに見えます。

UPDATE jobs
SET status = 'running', attempts = attempts + 1
WHERE id = (
  SELECT id FROM jobs
  WHERE status = 'queued'
  ORDER BY run_at
  LIMIT 1
)
RETURNING id, attempts;

2 つのセッションで同時に実行してみます。セッション A はトランザクションの中でこれを実行し、まだ作業中のところで、セッション B が同じステートメントを実行します。

-- session A (inside BEGIN, not committed yet)
 id | attempts
----+----------
  1 |        1

-- session B, returns after A commits
 id | attempts
----+----------
  1 |        2

両方のセッションがジョブ 1 を取得し、ジョブ 2 はまだ queued のままです。理由はこうです。サブクエリは一度だけ実行され、ジョブ 1 をいちばん古い queued のジョブとして見つけ、外側の UPDATE に id = 1 という値を渡します。次に B は A の行ロックを待たなければなりません。READ COMMITTED では、A がコミットすると、Postgres は行の新しいバージョンに対して WHERE 句を再評価します。残っている条件は id = 1 だけで、これはまだ一致するので、B は A がすでに取得したジョブを更新してしまいます。

実験環境の 2 セッションだけでは何とも言えないので、pgbench で 8 つのワーカーを 10,000 件の待機中ジョブに対して実行し、ワーカーごとに 1,250 回取得させました。

pgbench -n -c 8 -j 8 -t 1250 -f naive.sql postgres

naive.sql は上のステートメントを status = 'done' にしたもので、取得するたびに完了にもなります。実行後の結果は次のとおりです。

 status | count | max_attempts
--------+-------+--------------
 done   |  2100 |            8
 queued |  7900 |            0

10,000 回の取得で完了したジョブは約 2,100 件でした（2 台のマシンで 8 回実行し、2,068 件から 2,137 件の間）。また、8 つのワーカーすべてに取得されたジョブもありました。実際のシステムなら、同じメールが 8 通送られることになります。

選んだ行をロックすれば、二重取得は直ります。選択を FOR UPDATE 付きの CTE に入れ、それを通して更新します。

WITH next AS (
  SELECT id FROM jobs
  WHERE queue = 'default' AND status = 'queued' AND run_at <= now()
  ORDER BY run_at
  LIMIT 1
  FOR UPDATE
)
UPDATE jobs j
SET status = 'running', attempts = j.attempts + 1
FROM next
WHERE j.id = next.id
RETURNING j.id;

今度は、セッション B はジョブ 1 でブロックされます。A がコミットすると、Postgres はジョブ 1 を再確認し、もう status = 'queued' に一致しないことを確認してそれを除外し、ジョブ 2 に進みます。これは正しい動作ですが、B にかかった時間を見てください。

 id
----
  2
(1 row)

UPDATE 1
Time: 1858.126 ms (00:01.858)

B は、A がトランザクションを開いていた間ずっと待っていました。ワーカーが増えると、全員が同じ先頭の行の後ろに並び、1 つずつ順番に動くことになります。

2 語を追加します。SELECT のリファレンスにはこうあります。"With SKIP LOCKED, any selected rows that cannot be immediately locked are skipped."（SKIP LOCKED を指定すると、すぐにロックできない選択行はスキップされる）。ドキュメントは、これが何のためのものかも述べています。データの一貫しないビューを返すので汎用の処理には向かない "but can be used to avoid lock contention with multiple consumers accessing a queue-like table."（ただし、キューのようなテーブルに複数のコンシューマーがアクセスする場合のロック競合を避けるために使える）。

A がまだジョブ 1 を保持している状態でも、B はすぐにジョブ 2 を取得します。

 id
----
  2
(1 row)

UPDATE 1
Time: 2.512 ms

素朴なステートメントの代わりに SKIP LOCKED を使って、同じ pgbench を実行した結果です。

 status | count | max_attempts
--------+-------+--------------
 done   | 10000 |            1

すべてのジョブが完了し、どのジョブもちょうど 1 回ずつ実行されました。

これ以降で使う取得処理は次のとおりです。1 回の往復で最大 10 件のジョブを取得し（各ジョブが短時間で終わる場合に効いてきます）、リースを書き込みます。

WITH next AS (
  SELECT id FROM jobs
  WHERE queue = 'default' AND status = 'queued' AND run_at <= now()
  ORDER BY run_at
  LIMIT 10
  FOR UPDATE SKIP LOCKED
)
UPDATE jobs j
SET status       = 'running',
    attempts     = j.attempts + 1,
    locked_until = now() + interval '5 minutes'
FROM next
WHERE j.id = next.id
RETURNING j.id, j.payload, j.attempts;

数千行のテーブルでは、EXPLAIN (COSTS OFF) は望ましい形を示します。jobs_ready_idx のインデックススキャンの上に LockRows、その上に Limit があるので、ワーカーは実行可能な行だけに触れ、10 件で止まります。

 Update on jobs j
   CTE next
     ->  Limit
           ->  LockRows
                 ->  Index Scan using jobs_ready_idx on jobs
                       Index Cond: ((queue = 'default'::text) AND (run_at <= now()))
                       Filter: (status = 'queued'::text)
   ->  Nested Loop
         ->  CTE Scan on next
         ->  Index Scan using jobs_pkey on jobs j
               Index Cond: (id = next.id)

この取得処理の使い方は 2 通りあります。

1. 作業中ずっとロックする。 BEGIN、取得、作業、完了にする、COMMIT の順に行います。クラッシュするとロールバックされ、ジョブはまた空きになるので、魅力的に見えます。
2. リース。 status = 'running' と locked_until を付けて取得し、すぐに COMMIT します。作業はどのトランザクションの外でも行います。完了の記録は、2 つ目の短いトランザクションで行います。

1 つ目は、数ミリ秒で終わるジョブなら問題ありません。API を呼ぶ、メールを送る、ファイルをレンダリングするといった処理には、リースを使ってください。何かを書き込んだ（取得処理は書き込みます）あとで待機しているトランザクションは idle in transaction になり、それが開いている間、VACUUM は、そのトランザクションの開始後にデッドになった行バージョンを、そのデータベースのどのテーブルからも削除できません。開いたままの REPEATABLE READ トランザクションは、読み取りしかしていなくても同じことを引き起こします。

そのような形で取得を 1 つ開いたままにし、ほかのワーカーに 2,000 件のジョブを処理させてから、VACUUM (VERBOSE) jobs を実行しました。

tuples: 0 removed, 11992 remain, 2000 are dead but not yet removable

開いていたトランザクションをコミットしたあとで、同じコマンドを実行した結果です。

tuples: 2001 removed, 8542 remain, 0 are dead but not yet removable

キューのテーブルは更新と削除ばかりなので、VACUUM が止められると真っ先に肥大化するテーブルです。開いたトランザクションはサーバー接続も占有し、トランザクションモードのプーラーの背後にいる場合、その接続はコミットするまでプールから外れたままになります。短いトランザクションなら、どちらも避けられます。

リースを使うと、ジョブのライフサイクル全体は 4 つの短いステートメントになります。取得後にワーカーが実行するステートメントは、status = 'running' と、取得処理が返した値である attempts を確認するので、ワーカーはまだ自分が所有しているジョブしか変更できません。

完了。 ワーカーは、取得時に受け取った ID と試行回数を渡します。

UPDATE jobs
SET status = 'done', locked_until = NULL
WHERE id = 1 AND status = 'running' AND attempts = 1;

失敗とバックオフ付きリトライ。 1 回目の失敗の 20 秒後、2 回目の失敗の 40 秒後と倍々に増やして上限を 1 時間とし、max_attempts に達したら最終的に failed にします。デフォルトの 5 回の試行では最長の待ち時間は 160 秒で、上限が意味を持つのは max_attempts を引き上げた場合だけです。

UPDATE jobs
SET status       = CASE WHEN attempts >= max_attempts THEN 'failed' ELSE 'queued' END,
    run_at       = now() + least(interval '1 hour', interval '10 seconds' * 2 ^ attempts),
    locked_until = NULL,
    last_error   = 'SMTP 451: try again later'
WHERE id = 2 AND status = 'running' AND attempts = 1
RETURNING id, status, attempts, run_at - now() AS retry_in;

 id | status | attempts | retry_in
----+--------+----------+----------
  2 | queued |        1 | 00:00:20

期限切れのリースを回収する。 ワーカーがジョブの途中で落ちると、そのリースは期限切れになり、任意のワーカー（または cron ジョブ）がジョブを戻します。毎回ワーカーを落とすジョブは、そのままだと永遠に戻ってくるので、回収処理も max_attempts を守ります。

UPDATE jobs
SET status       = CASE WHEN attempts >= max_attempts THEN 'failed' ELSE 'queued' END,
    locked_until = NULL,
    last_error   = coalesce(last_error, 'lease expired')
WHERE status = 'running' AND locked_until < now()
RETURNING id, status, attempts;

試行回数の確認が重要な理由。 ジョブ 3 を担当していたワーカーが落ちたのではなく、単に遅かっただけだとします。そのリースは期限切れになり、回収処理がジョブを戻し、別のワーカーがそれを取得したので、attempts は 2 になっています。遅いワーカーがようやく作業を終え、attempts = 1 で完了処理を実行しても、何も起きません。

UPDATE 0

ジョブは現在の所有者のもとに残ります。この確認がなければ、2 つ目のワーカーがまだ実行している間に、遅いワーカーがジョブを完了にしてしまいます。遅いワーカーの副作用はすでに起きているので、作業自体はやはり冪等にしておくべきです。この確認は、ジョブの状態を正しく保つだけのものです。

リースは、通常のジョブのうち最も遅いものより長くするか、長いジョブには実行しながら自分のリースを延長させてください。延長でどの行も更新されなかった場合は、そのジョブはもう自分のものではないので、処理をやめます。

UPDATE jobs
SET locked_until = now() + interval '5 minutes'
WHERE id = $1 AND status = 'running' AND attempts = $2;

ワーカーは、仕事があることを知る必要があります。いちばん簡単なのは、取得処理をループで実行し、何も返らなければスリープする方法です。もう 1 つの方法は、Postgres に知らせてもらうことです。

CREATE FUNCTION jobs_notify() RETURNS trigger
LANGUAGE plpgsql AS $$
BEGIN
  PERFORM pg_notify('jobs', NEW.queue);
  RETURN NULL;
END;
$$;

CREATE TRIGGER jobs_notify
AFTER INSERT ON jobs
FOR EACH ROW EXECUTE FUNCTION jobs_notify();

ワーカーは LISTEN jobs を実行し、通知が届いたら取得します。NOTIFY のリファレンスにある 3 つの性質が、使い方を左右します。

• 配信はコミット時に行われる。 トランザクションの中では、通知は "are not delivered until and unless the transaction is committed."（トランザクションがコミットされるまで、またコミットされない限り配信されない）。私たちのテストでは、プロデューサーが 13:08:30 にジョブを挿入し、13:08:33 にコミットしました。リスナーの 13:08:31 の確認では何も見えず、通知はその次の確認で現れました。これは望ましい動作です。早く起こされたワーカーは、まだ見えないジョブを見つけることになるからです。
• 通知はキューではない。 通知を受け取れるのは、その時点で LISTEN しているセッションだけで、1 つのトランザクション内の同一の通知は 1 つにまとめられます。通知は「見に行け」という合図として扱い、ジョブそのものとして扱わないでください。また、再接続中に取りこぼしたものに備えて、より間隔の長いポーリングをフォールバックとして残しておきます。
• ペイロードは小さい。 デフォルトの設定では、ペイロードは 8000 バイト未満でなければなりません。ジョブそのものではなく、キュー名や ID を送ってください。

もう 1 点あります。LISTEN はサーバー接続に属するので、トランザクションモードのプーラーを経由した接続ではなく、ダイレクト接続が必要です。理由は PgBouncer のトランザクションプーリングで壊れるものと、それぞれの直し方で説明しました。プロデューサーと取得クエリは、プール経由の接続を使ってかまいません。

Relyt One では、約 10 分間何もすることがなかったコンピュートはサスペンドし、次の接続で再開します。キューにとって重要なのは、何が「すること」に数えられるかです。コンピュートのアクティビティモニターは、0.5 秒ごとにクライアントセッションを確認します。

• idle でないセッション（クエリを実行中、または idle in transaction）は、その時点のアクティビティとして数えられます。
• idle のセッションは、最後に状態が変わった時点、つまり最後のクエリが終わった時点のアクティビティとして数えられます。
• 実行中の autovacuum ワーカーや、アクティブな論理レプリケーション接続も、コンピュートを起こしたままにします。

そのため、1 秒ごとにポーリングするワーカーも、5 分ごとにポーリングするワーカーも、たとえ一晩中キューが空でも、コンピュートを 24 時間起こしたままにします。作業中にトランザクションを開いたままにするワーカーは、ジョブの間ずっとコンピュートを起こしたままにします。アイドル接続それ自体はそうなりません。最後の状態変化がどんどん古くなっていくからです。ここから 3 つのことが言えます。サスペンドは LISTEN の接続も含めてすべての接続を閉じ、すぐに再接続するワーカーはコンピュートを再開させます。ドライバーやプールからのヘルスチェッククエリはアクティビティとして数えられます。そして、10 分より短い間隔で実行されるフォールバックのポーリングは、コンピュートを起動したままにします。

したがって、現実的な選択肢は 3 つです。

1. ジョブを数秒以内に開始しなければならない。 ワーカーが動いている間は、常時稼働のコンピュートを受け入れます。これは、どのプラットフォームでも低レイテンシの代償です。許容できる間隔でポーリングするか、フォールバックのポーリングと組み合わせて LISTEN を使います。
2. ジョブはしばらく待てる。 ワーカーを 24 時間動かし続けないでください。アイドルタイムアウトより長い空き時間ができるスケジュールでワーカーを起動し、バッチ取得でキューを空にしたら終了させます。実行と実行の間にコンピュートはサスペンドでき、次の実行の最初の接続でコンピュートが再開します。
3. ジョブは自分のアプリから来る。 ジョブをキューに入れるリクエストで、ワーカーも起動します（サーバーレス関数や、プラットフォームのジョブランナーのタスクなど）。状態、リトライ、リースの信頼できる情報源はデータベースのままで、何もポーリングする必要がありません。

ここではブランチも役に立ちます。ブランチごとに独自のコンピュートがあるので、本番データのコピーを持つブランチにテスト用のワーカー群を向け、取得クエリに負荷をかけても、本番のコンピュートには影響しません。

完了したジョブを、頻繁にアクセスされるテーブルに残しておく必要はありません。スケジュールを決めて、バッチで削除するか、アーカイブテーブルに移します。

DELETE FROM jobs
WHERE id IN (
  SELECT id FROM jobs
  WHERE status = 'done' AND created_at < now() - interval '7 days'
  LIMIT 10000
);

取得、完了、リトライのたびに新しい行バージョンが書き込まれるので、このテーブルは絶えずデッドタプルを生みます。pg_stat_user_tables で jobs の n_dead_tup と last_autovacuum を監視し、VACUUM が見つけたものを実際に削除できるよう、トランザクションを短く保ってください。

ジョブが同じデータベースの行に関係していて、スループットがジョブごとに UPDATE 1 回で収まる範囲で、キューへの投入をデータ変更と同じトランザクションに含めたい場合、ジョブテーブルはよく合います。最後の点は、外部のブローカーでは追加の仕組みなしには実現できません。一方、多数の独立したコンシューマーへのファンアウト、あとで再生するためのストリームの保持、更新の多さそのものがボトルネックになるような量には向きません。そうした用途には、ブローカーやログを使ってください。

• CTE と FOR UPDATE SKIP LOCKED で取得し、ロックなしの WHERE id = (SELECT ... LIMIT 1) は決して使わない
• 実行可能の条件に部分インデックスを使う
• 取得したらすぐにコミットし、locked_until を記録し、作業はトランザクションの外で行う
• 完了、失敗、延長は WHERE id = $1 AND status = 'running' AND attempts = $2 で行い、行数を確認する
• 期限切れのリースを回収する処理を動かし、max_attempts でリトライ回数に上限を設ける
• LISTEN はダイレクト接続で使い、通知はヒントとして扱い、フォールバックのポーリングを残す
• 完了したジョブを削除またはアーカイブし、n_dead_tup を監視する
• サーバーレスのコンピュートを起こしたままにするものを把握したうえで、ワーカーの待ち方を決める

この記事のステートメントはすべて、拡張機能を使わない素の PostgreSQL 17 の SQL です。無料の Relyt One プロジェクトを作成し、ブランチを開き、そこに 2 つの psql セッションを向けて、SKIP LOCKED がそれぞれに別の行を渡す様子を確かめてください。

無料で始める