選択した行を ready から processing に変更するクエリ内で FOR UPDATE SKIP LOCKED を使用します。これにより、並行して動作するワーカーは異なる合成記事の行を取得でき、保存された状態によってロック終了後に別のワーカーが同じ行を取得するのを防げます。

この手順は、PostgreSQLへの接続が確立されているMacでTerminalを使う開発者向けです。合成記事を3件作成し、2つのデータベースセッションで並行取得を確認して、node-postgresの完全なサンプルを実行します。

2つのワーカーが同じ記事を選択するのはなぜですか？

単純な SELECT では、その結果は予約されません。PostgreSQLのデフォルトであるRead Committed分離レベルでは、どちらかが変更する前に、2つのワーカーが同じコミット済みの ready 行を読み取ることがあります。単純な読み取りと後続の更新をトランザクションで囲んでも、選択が排他的になるわけではありません。PostgreSQLのスナップショット動作については、トランザクション分離に記載されています。

取得には2つの保護策が必要です。記事を選択する間の行ロックと、そのロックを解放する前の状態変更です。PostgreSQLはトランザクション終了時に行ロックを解放します。そのため、ロック付き読み取りだけを行ってから処理すると、その読み取りのトランザクションが終了した時点で、行は再び取得可能になります。明示的ロックを参照してください。

すべてのワーカーで同じ取得プロトコルを使います。ready の記事を読み取り、取得に成功せずに処理を始めるアプリケーションの経路があると、作業が重複する可能性があります。

小さな合成記事キューを作成する

前提条件は、稼働中のPostgreSQLデータベース、Terminalで使える psql コマンド、既存の接続設定です。JavaScriptの節では、現在サポートされているNode.jsのリリースとnpmもインストールしてください。この記事では、データベースへの接続が確立されているところから始めます。インストール手順は、現在のMacの環境によって異なります。

SKIP LOCKED 機能がPostgreSQLに導入されたのは9.5です。これは構文上の最低要件であり、サポート対象外のリリースを使うことを推奨するものではありません。サポート対象のPostgreSQLバージョンを使用してください。取得した最新のドキュメントでは、PostgreSQL 18が現行版とされています。9.5のリリースノートと最新のSELECTドキュメントを参照してください。

Terminalを開き、通常の接続設定を使って psql を実行します。これにより、psqlのドキュメントに記載されているとおり、SQLを実行して結果を確認できる対話型プロンプトが表示されます。以下のセットアップは、article_claim_demo がまだ存在しないスクラッチデータベースで一度だけ実行してください。既存のテーブルを置き換えることはありません。

SHOW server_version;
SHOW transaction_isolation;

CREATE TABLE article_claim_demo (
    id integer PRIMARY KEY,
    title text NOT NULL,
    state text NOT NULL DEFAULT 'ready'
        CHECK (state IN ('ready', 'processing', 'done')),
    rendered_title text
);

INSERT INTO article_claim_demo (id, title) VALUES
    (1, 'Paper shapes'),
    (2, 'Desk arrangements'),
    (3, 'Color combinations');

以下の例では、Read Committed分離レベルと、追加のトリガーがない単純なテーブルを前提としています。2セッションのテストが終わるまで、サンプルの3行は変更しないでください。

1つのSQL文で記事を1件取得する

ワーカーが記事を必要とするたびに、この文を実行します。候補クエリは取得可能な行をロックし、更新によって取得状態を記録して入力フィールドを返します。PostgreSQLのUPDATEドキュメントには、このCTEとUPDATEのパターンが説明されており、RETURNING が実際に更新された行を返すことも確認できます。

WITH candidate AS (
    SELECT id
    FROM article_claim_demo
    WHERE state = 'ready'
    ORDER BY id
    LIMIT 1
    FOR UPDATE SKIP LOCKED
)
UPDATE article_claim_demo AS a
SET state = 'processing'
FROM candidate AS c
WHERE a.id = c.id
  AND a.state = 'ready'
RETURNING a.id, a.title;

ロックは候補CTEの中に置きます。SKIP LOCKED により、別のワーカーはすぐにロックできない候補を飛ばして進めます。LIMIT 1 は取得件数を1件に制限し、主キーとの結合によって更新対象を1つに特定します。PostgreSQLのSELECT: ロック句には、WITH クエリ内でのロックと、キューのような用途でロック中の行をスキップする方法が明記されています。

明示的な BEGIN がなければ、PostgreSQLはこの文を独自のトランザクション内で実行します。文が成功すると、後続の処理が始まる前に、選択に伴う更新と状態変更の両方がコミットされます。コミット後に行ロックはなくなりますが、state = 'ready' によってその記事は対象から除外されます。トランザクションを参照してください。

取得文が返した記事だけを処理してください。結果が空なら、この試行では記事を取得できなかったという意味です。すべての作業が完了したという意味ではありません。他のワーカーが取得可能な行を現在ロックしている場合があります。また、SKIP LOCKED は行ロック待ちを回避しますが、テーブルレベルのロックでは待機が発生することがあります。こうした制約はSELECTに記載されています。

2つのTerminalセッションで取得動作を確認する

psql を使って同じデータベースに接続したTerminalウィンドウを2つ用意します。元の ready の3行から始めます。以下の手順では各トランザクションを開いたままにし、重複する取得を直接確認できます。

  1. セッションAで BEGIN ISOLATION LEVEL READ COMMITTED; を実行し、続けて上記の取得文全体を貼り付けます。トランザクションは開いたままにします。サンプルデータが変更されていなければ、返されるIDは 1 のはずです。
  2. セッションBで BEGIN ISOLATION LEVEL READ COMMITTED; を実行し、同じ取得文を貼り付けます。Aが 1 の行をロックしている間に、ID 2 が返されるはずです。
  3. セッションBで ROLLBACK; を実行し、続けてセッションAでも ROLLBACK; を実行します。
  4. どちらかのセッションで、以下の確認クエリを実行します。両方の取得がロールバックされたため、3件すべての記事は再び ready になっているはずです。
SELECT id, title, state, rendered_title
FROM article_claim_demo
ORDER BY id;

これは、実行したデータベーステストの報告ではなく、ドキュメントに記載されたロックとロールバックの動作から導かれる想定結果です。ロールバックすると各状態更新が取り消され、ロックが解放されます。代わりに COMMIT を使うと、記事は processing のままになり、後続の取得対象にはなりません。明示的ロックとトランザクションを参照してください。

このテストでは、2つの有用な動作を確認できます。セッションBが現在ロックされている記事を飛ばして次に進むことと、コミット済みの状態変更によって記事が今後の取得対象から外れることです。既存のワーカーに適用するときは、両方の動作をテストしてください。

node-postgresで3つのワーカーを実行する

両方のテストトランザクションをロールバックした後、Terminalで新しいローカルプロジェクトディレクトリを作成します。ドキュメントにあるnode-postgresのインストールコマンドでパッケージをインストールします。

mkdir article-claim-demo
cd article-claim-demo
npm init -y
npm install pg

以下の完全なプログラムを workers.mjs として保存してください。psql と同じデータベース接続環境変数を使います。node-postgresの接続方法には、共通の接続環境変数が記載されています。各ワーカーは1件の取得を試み、合成タイトルを大文字に変換して結果を保存します。

import pg from 'pg';

const { Pool } = pg;
const pool = new Pool({ max: 3 });

const claimSQL = `
WITH candidate AS (
  SELECT id
  FROM article_claim_demo
  WHERE state = 'ready'
  ORDER BY id
  LIMIT 1
  FOR UPDATE SKIP LOCKED
)
UPDATE article_claim_demo AS a
SET state = 'processing'
FROM candidate AS c
WHERE a.id = c.id AND a.state = 'ready'
RETURNING a.id, a.title;
`;

async function worker(label) {
  const claimed = await pool.query(claimSQL);

  if (claimed.rows.length === 0) {
    console.log(`${label}: no row available now`);
    return;
  }

  const article = claimed.rows[0];
  const renderedTitle = article.title.toUpperCase();

  const completed = await pool.query(
    `UPDATE article_claim_demo
     SET rendered_title = $2, state = 'done'
     WHERE id = $1 AND state = 'processing'
     RETURNING id`,
    [article.id, renderedTitle]
  );

  if (completed.rowCount !== 1) {
    throw new Error(`${label}: completion did not update one row`);
  }

  console.log(`${label}: completed article ${article.id}`);
}

try {
  const results = await Promise.allSettled([
    worker('A'),
    worker('B'),
    worker('C')
  ]);

  for (const result of results) {
    if (result.status === 'rejected') {
      console.error(result.reason);
      process.exitCode = 1;
    }
  }
} finally {
  await pool.end();
}

node workers.mjs を実行し、続けて psql で確認クエリをもう一度実行します。取得可能な未変更の記事が3件あり、他の処理がなければ、3件すべての状態が done になり、rendered_title の値は大文字になっているはずです。ワーカーのラベルやコンソールへの出力順は、記事の順序と一致しない場合があります。

各取得が1つの完全な文で、各完了処理が別の文なので、ここでは pool.query を使うのが適切です。プールは接続の貸し出しと返却を管理し、試行がすべて完了した後に pool.end() がプールを閉じます。プーリングを参照してください。

完了クエリでは、クエリに従い、$1 と $2 を使って値を別々に渡します。プログラムは、取得結果が空かどうかを返された rows 配列で確認し、pg.Resultに沿って、更新された完了行の数を rowCount で確認します。

JavaScriptは準備中に構文チェックを通過しました。調査環境ではPostgreSQLの実行ファイルを利用できなかったため、SQLとデータベースの並行動作はドキュメントに照らして確認しましたが、ここでは実行していません。

適切なトランザクション境界を選び、未完了の作業に対処する

この例では、短い取得処理をコミットしてからタイトルを変換します。これによりロックの保持時間を短くできますが、意図的に復旧境界が生じます。取得後、完了を記録する前にワーカーが停止すると、記事は processing のままになります。プログラムを再実行しても、その記事は再取得されません。

このデスクトップでの実演では、すべてのワーカーを停止し、processing の記事を確認してから、意図的に再取得可能にしてください。以前のワーカーが処理を完了する可能性がある間は、自動で状態をリセットしないでください。この例には、以前の試行と後の試行を見分ける取得世代フィールドがありません。重複取得の防止だけでは、自動復旧や厳密に1回だけの処理は実現できません。

例を応用するときは、以下を判断の目安にしてください。

作業内容 トランザクションの方法 結果
記事を取得してから、時間のかかるアプリケーション処理を行う 単一文の取得を先にコミットする processing の状態を明示して維持し、復旧方法は別途設計する
小さな変換をすべてPostgreSQL内で行う 取得と完了処理を1つのトランザクション内で行う データベースの変更をまとめてコミットまたはロールバックできる
Node.jsで複数のSQL文を1つのトランザクションとして実行する 1つのクライアントを確保して使い続ける BEGIN、各文、COMMIT または ROLLBACK の実行中、同じ接続を使い続ける

最後の方法では、トランザクション内のすべての文に、同じ確保済みクライアントを使う必要があります。個別の pool.query 呼び出しは異なる接続を使う場合があるため、それらをまとめて1つのトランザクションにすることはできません。完了またはロールバック後にクライアントを解放してください。トランザクションとプーリングを参照してください。

ここで示した分離レベルの前提を明確にしてください。Repeatable ReadとSerializableでは、トランザクション開始後にロック対象の行が変更されている場合、エラーが発生することがあります。こうした分離レベルを使うアプリケーションでは、トランザクションの再試行処理が必要です。PostgreSQLの明示的ロックにこの制約が記載されています。

この合成記事の処理で重要な手順は明確です。取得可能な行を選択してロックし、同じ文でその状態を変更し、取得に成功した後に返された記事だけを処理します。2つのセッションで重複する取得を確認したら、ワーカーが完了を記録しない記事をアプリケーションでどう扱うかを決めてください。

出典