node-postgresのトランザクションを一貫して実行するには、pool.connect()でクライアントを1つ取得し、すべてのステートメントをそのクライアント経由で実行して、コミットまたはロールバック後に解放します。ヘルパーにも同じクライアントを渡してください。公式のTransactionsガイドでは、このパターンが必須とされています。

このチュートリアルは、MacのTerminalを使って、複数の記事行に対する変更をまとめて成功または失敗させる開発者向けです。合成記事行を使った完全なプログラム、意図的にロールバックを発生させるテスト、ヘルパー関数内に隠れたクエリを診断する手順が含まれています。

pool.query()を使うとトランザクションの範囲が崩れるのはなぜですか？

プールは接続を管理しますが、すべてのクエリを1つのトランザクションにまとめるものではありません。pool.query()を呼び出すたびに、内部で利用可能なクライアントが取得されます。そのため、連続する呼び出しが同じ接続を使う保証はありません。node-postgres Pool APIでは、この方法をトランザクションに使わないよう明示的に警告しています。

たとえば、BEGINをプール経由で送信した後、挿入をプール経由で送っても、その挿入が開始したトランザクションに含まれるとは限りません。PostgreSQLでは通常、明示的なトランザクションブロックの外で実行されたステートメントは、それぞれ独立したトランザクションとして実行されます。別の接続で後からロールバックしても、すでに個別にコミットされた変更は取り消せません。詳しくはPostgreSQLのBEGINドキュメントを参照してください。

重要な境界となるのは、チェックアウトしたクライアントオブジェクトです。ヘルパーの名前や、そのヘルパーを含むモジュール、呼び出し元がその処理をawaitしていることだけでは、トランザクションへの参加は決まりません。トランザクションを開始したクライアントを使ってクエリを実行して初めて、ヘルパーはそのトランザクションに参加します。これはTransactionsガイドの同一クライアント要件から直接導かれます。

Macでコミットとロールバックの完全な例を実行する

Node.jsとnpm、起動中のPostgreSQLサーバー、既存のローカル開発用データベースが必要です。この手順では、データベース名をarticle_demoとし、通常のローカル接続設定がすでに機能していることを前提としています。データベースのインストールは、このトランザクションの作業範囲外です。

Terminalで新しい作業ディレクトリを作成し、ドライバーをインストールします。

mkdir pg-transaction-demo
cd pg-transaction-demo
npm init -y
npm install pg

次の完全なプログラムをtransaction-demo.mjsとして保存します。一時テーブルには合成の記事IDとタイトルだけが入ります。PostgreSQLの一時テーブルは通常、コミット後も残り、セッション終了時に削除されます。この性質を利用して、2つ目のバッチがロールバックされた後に、成功したバッチの結果を確認できます。詳しくはCREATE TABLEを参照してください。

以下のトランザクションヘルパーは、ドキュメントに記載されたBEGIN、COMMIT、ROLLBACKの順序に従います。また、ロールバックも失敗した場合の処理を明示的に追加し、元のエラーを置き換えずに両方のエラーを保持します。クエリ値にはQueriesに記載されたテキストと値のAPIを使います。

import pg from 'pg';
import assert from 'node:assert/strict';

const { Pool } = pg;

async function inTransaction(client, work) {
  await client.query('BEGIN ISOLATION LEVEL READ COMMITTED');

  try {
    const result = await work(client);
    await client.query('COMMIT');
    return result;
  } catch (error) {
    try {
      await client.query('ROLLBACK');
    } catch (rollbackError) {
      throw new AggregateError(
        [error, rollbackError],
        'Transaction failed; rollback also failed'
      );
    }
    throw error;
  }
}

async function insertArticle(client, article) {
  const result = await client.query(
    `INSERT INTO pg_temp.transaction_demo_articles (id, title)
     VALUES ($1, $2)
     RETURNING id, title`,
    [article.id, article.title]
  );
  return result.rows[0];
}

async function insertBatch(client, articles) {
  const inserted = [];
  for (const article of articles) {
    inserted.push(await insertArticle(client, article));
  }
  return inserted;
}

async function main() {
  const pool = new Pool({ connectionTimeoutMillis: 5000 });
  let client;
  let discardClient = false;

  pool.on('error', (error) => {
    console.error('Idle client error:', error);
    process.exitCode = 1;
  });

  try {
    client = await pool.connect();

    await client.query(`
      CREATE TEMP TABLE transaction_demo_articles (
        id text PRIMARY KEY,
        title text NOT NULL
      ) ON COMMIT PRESERVE ROWS
    `);

    const saved = await inTransaction(client, (tx) =>
      insertBatch(tx, [
        { id: 'article-001', title: 'A guide to paper shapes' },
        { id: 'article-002', title: 'Choosing a notebook layout' }
      ])
    );
    assert.equal(saved.length, 2);

    await assert.rejects(
      () => inTransaction(client, (tx) =>
        insertBatch(tx, [
          { id: 'article-003', title: 'Arranging a reading desk' },
          { id: 'article-001', title: 'Deliberate duplicate ID' }
        ])
      ),
      (error) => error.code === '23505'
    );

    const remaining = await client.query(`
      SELECT id, title
      FROM pg_temp.transaction_demo_articles
      ORDER BY id
    `);

    assert.deepEqual(
      remaining.rows.map((row) => row.id),
      ['article-001', 'article-002']
    );

    console.log('Commit and rollback checks passed.');
    console.table(remaining.rows);
  } catch (error) {
    discardClient = true;
    throw error;
  } finally {
    if (client) client.release(discardClient);
    await pool.end();
  }
}

main().catch((error) => {
  console.error(error);
  process.exitCode = 1;
});

同じディレクトリから実行します。

PGDATABASE=article_demo node transaction-demo.mjs

プールはPostgreSQLの接続環境変数を読み取ります。ローカル環境で別の接続設定が必要な場合は、PGDATABASEとともに適切なPGHOSTおよびPGPORTを指定してください。公式のConnectingガイドに、これらの変数とデフォルト値が説明されています。

正常に実行されると、Commit and rollback checks passed.に続いて、元の2つの記事行が表示されます。2つ目のバッチではarticle-003を挿入した後、article-001を再利用しようとします。この重複によってSQLSTATE 23505が発生します。これはSerialization Failure Handlingでunique_violationとして説明されています。バッチをロールバックすると、先に行った挿入も取り消されます。アサーションではarticle-003が存在しないことを確認します。

トランザクションヘルパーの成功時、処理エラー時、ロールバックエラー時の各制御経路は、決定論的なモッククライアントで確認しました。ここではpgパッケージとPostgreSQLのテスト接続が環境になかったため、データベースの例は実行していません。上記の出力は期待される結果です。

ヘルパー間でトランザクションクライアントを共有する方法

呼び出しチェーン全体で、クライアントを明示的な引数として渡します。このプログラムでは、main()がクライアントを取得し、inTransaction()が同じオブジェクトをコールバックに渡し、insertBatch()がそれをinsertArticle()に転送します。どちらの挿入ヘルパーも、内部で接続をチェックアウトしません。

この実装では、各関数の役割を明確にします。

関数 役割
main() クライアントを取得し、デモの準備を行い、クライアントを解放して、スクリプトのプールを閉じる。
inTransaction() 渡されたクライアントで1つのトランザクションを開始し、終了する。
insertBatch() 各記事操作をawaitし、渡されたクライアントを転送する。
insertArticle() 渡されたクライアント経由で挿入を実行する。

ヘルパーの契約は意図的に限定されています。クライアントにアクティブなトランザクションがない場合に限りinTransaction()を呼び出し、コールバック内の失敗はそのまま伝播させます。既存のPostgreSQLトランザクション内でBEGINを呼び出しても、独立したネストトランザクションは作成されません。PostgreSQLでは、ネストした処理の仕組みとしてセーブポイントを説明しています。詳しくはBEGINを参照してください。

トランザクションに属するすべての操作を、コールバックから戻る前にawaitしてください。この例では逐次ループを使い、完了する時点を明確にしています。非同期挿入を開始しても、そのPromiseを返すかawaitしなければ、コールバックはすべての結果を確認する前に終了してしまいます。

クエリまたはロールバックに失敗した場合の動作

通常のクエリ失敗はトランザクションヘルパーのcatchブロックに届きます。そこで同じクライアントを使ってロールバックを試み、元のエラーを再スローします。意図的に発生させた重複エラーは、ロールバックが完了した場合にのみ許容されます。ロールバックに失敗すると、ヘルパーはAggregateErrorをスローします。重複エラーのチェックではそれを許容しないため、外側の失敗ハンドラーに処理が進みます。

この例では、予期しない失敗が発生した場合、client.release(true)を使ってクライアントを破棄する安全側の処理を行います。通常の経路では、client.release(false)でクライアントを返却します。破棄用の引数についてはPool APIに記載されています。

取得に成功したクライアントは必ず解放してください。解放しないと、後続の呼び出しが利用できない接続を待ち続けることになります。この単独スクリプトでは、処理終了後にプールを閉じます。アプリケーションでは、プールの終了処理はアプリケーションの終了時に行います。Poolingガイドでは、クライアントを返却する必要性と、チェックアウト中のクライアントが返却されるまでpool.end()が待機する仕組みを説明しています。

クライアントを1つにすると、スナップショットは安定しますか？

クライアントの継続性と分離レベルは、異なる問いに答えるものです。クライアントは、どのトランザクションでステートメントが実行されるかを決めます。分離レベルは、他のトランザクションが実行されている間に、ステートメントから何が見えるかを決めます。PostgreSQLのTransaction Isolationドキュメントでは、次のような制限を定義しています。

分離レベル 関連する制限
READ COMMITTED デフォルトです。連続する通常の読み取りで、同時にコミットされた異なるデータが見えることがあります。各ステートメントで新しいスナップショットが使われます。
REPEATABLE READ 最初のトランザクション制御以外のステートメントで確立したスナップショットを使います。PostgreSQLではファントムリードを防ぎますが、直列化異常は起こり得ます。
SERIALIZABLE コミットされた直列化可能トランザクションは、何らかの逐次実行と整合する結果になります。トランザクションが失敗し、再試行が必要になる場合があります。
READ UNCOMMITTED PostgreSQLではREAD COMMITTEDと同じように動作します。

読み取りでは、自分のトランザクション内でそれ以前に行った変更も確認できます。そのため、自分が挿入したデータを確認できても、後続のすべての読み取りで他のトランザクションの変更が固定された状態で見えるとは限りません。詳しくはTransaction Isolationを参照してください。

この例では、最初のステートメントでREAD COMMITTEDを明示的に選択します。別の分離レベルを使うには、トランザクションを実行する前にそのステートメントを変更します。PostgreSQLではBEGINでトランザクション特性を指定できます。代わりにSET TRANSACTIONを使う場合、分離レベルは最初のクエリまたはデータ変更ステートメントより前に設定する必要があります。BEGINとSET TRANSACTIONを参照してください。

REPEATABLE READまたはSERIALIZABLEを選ぶ場合は、SQLSTATE 40001に備えてください。失敗したクエリだけを再実行するのではなく、ステートメントと値を選ぶロジックも含め、トランザクション全体を再試行します。この例では再試行処理を実装せず、そのようなエラーをそのまま伝播させます。PostgreSQLのSerialization Failure Handlingでは、この要件を説明し、再試行しても再び失敗する可能性があると述べています。

トランザクションから外れたクエリを見つける方法

意図したバッチを、接続のチェックアウトから最終的なコミットまたはロールバックまで追跡します。コードを確認する際は、次の手順を使います。

  1. pool.connect()の呼び出しを見つけ、返されるクライアントを特定します。
  2. BEGINからトランザクション完了までに呼び出される各ヘルパーを追います。それぞれがそのクライアントを受け取り、使用していることを確認します。
  3. 共通のクエリラッパーを調べ、内部に隠れたpool.query()の呼び出しや追加の接続チェックアウトがないことを確認します。
  4. すべての操作がawaitされ、ヘルパーが失敗を伝播させることを確認します。
  5. 合成データを使った重複行テストをバッチに合わせて変更し、後続のステートメントが失敗したときに、先行する挿入が取り消されることを確認します。

ロールバックテストに合格すれば、そのテストで実行した経路について有用な根拠が得られますが、アプリケーションのすべての分岐で正しいクライアントが使われていることまでは証明できません。条件付きのヘルパーも確認してください。クエリが1つのクライアントにとどまっているのに読み取り結果が変わる場合は、接続処理を変更する前に、選択した分離レベルと照らし合わせてください。

接続の取得と解放を見える形にし、クライアントを明示的に渡し、意図したすべての操作が終わってからトランザクションが完了するようにします。こうすれば、具体的な構造を確認でき、記事のワークフローが拡大してもロールバックテストを繰り返せます。

出典