データベースの並行処理で厄介なのが、レースコンディション（競合状態）です。テストで再現しにくく、本番環境で突然顔を出すという性質があるんですよね。PostgreSQLでは同期バリア（synchronization barriers）を使うことで、こうした並行処理の問題を確実に再現・テストできる手法が注目されています。

レースコンディションとは何か

レースコンディションは、複数のトランザクションが同じデータに同時にアクセスしたとき、実行順序によって結果が変わってしまう現象です。たとえば、在庫が1つしかない商品に対して2人のユーザーが同時に購入ボタンを押した場合、適切なロック処理がないと在庫がマイナスになるケースがあります。

問題なのは、通常のテストではこれが再現できないことです。テストは基本的にシーケンシャルに実行されるので、「たまたまタイミングが重なった」という状況を作り出せません。手動でスリープを入れたりする方法もありますが、不安定で信頼性に欠けます。

同期バリアによるテストアプローチ

同期バリアとは、複数のプロセスやスレッドを特定のポイントで「待ち合わせ」させる仕組みです。PostgreSQLでは、アドバイザリーロック（pg_advisory_lock）やイベントトリガーを活用して、トランザクションの実行タイミングを正確にコントロールできます。

具体的な手法として、以下のような流れでテストを構成します。

  • トランザクションAを開始し、特定の操作の直前でアドバイザリーロックを使って一時停止させる
  • トランザクションBを開始し、同じリソースへのアクセスを試みる
  • ロックを解放して、両トランザクションが同時に競合する状況を意図的に作る
  • 結果が期待通りか（デッドロック検出、適切なエラー、正しいシリアライゼーション）を検証する

PostgreSQLのアドバイザリーロックを使った実装例

実際にどう書くのか、簡単な例を見てみましょう。2つのセッションで同じ行を更新する場面を想定します。

-- セッション1: ロックを取得して待機
BEGIN;
SELECT pg_advisory_lock(1);
UPDATE accounts SET balance = balance - 100 WHERE id = 1;
-- ここでセッション2の準備が整うまで待つ
SELECT pg_advisory_lock(2);  -- セッション2がロック2を解放するまでブロック
COMMIT;

-- セッション2: 別ターミナルで実行
BEGIN;
SELECT pg_advisory_lock(2);  -- まず自分のロックを取得
UPDATE accounts SET balance = balance - 100 WHERE id = 1;
SELECT pg_advisory_unlock(2);  -- セッション1に「準備OK」を通知
COMMIT;

この方法を使えば、2つのトランザクションが確実に同じタイミングで同じ行を更新しようとする状況を作れるんですよね。PostgreSQL公式のロック機構ドキュメントにも詳しい説明があります。

テストフレームワークとの統合

PythonやGoなどのテストフレームワークと組み合わせる場合は、スレッドやasyncioを使って複数のデータベースセッションを並行実行します。各スレッドがアドバイザリーロックのタイミングで同期し合うことで、確実に競合状態を再現できます。

Pythonの場合、threading.Barrierとpsycopg2を組み合わせるのが比較的シンプルです。サーバーレスOCRの構築のように、少ないコードでも強力な仕組みは作れます。

import threading
import psycopg2

barrier = threading.Barrier(2)

def transaction_a():
    conn = psycopg2.connect("dbname=test")
    cur = conn.cursor()
    cur.execute("BEGIN")
    cur.execute("SELECT * FROM accounts WHERE id = 1 FOR UPDATE")
    barrier.wait()  # Bと同期
    cur.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
    conn.commit()

def transaction_b():
    conn = psycopg2.connect("dbname=test")
    cur = conn.cursor()
    cur.execute("BEGIN")
    barrier.wait()  # Aと同期
    cur.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
    conn.commit()

t1 = threading.Thread(target=transaction_a)
t2 = threading.Thread(target=transaction_b)
t1.start(); t2.start()
t1.join(); t2.join()

SSI（直列化可能スナップショット分離）での検証

PostgreSQLのSERIALIZABLE分離レベルでは、SSI（Serializable Snapshot Isolation）という仕組みが使われています。同期バリアを使ったテストは、このSSIが正しくシリアライゼーション異常を検出するかどうかの検証にも有効です。

たとえば、書き込みスキュー（write skew）という典型的な異常パターンがあります。2つのトランザクションがそれぞれ異なる行を読んで、相手の行に依存する更新を行うケースですね。PostgreSQLのトランザクション分離レベルのドキュメントで詳しく解説されています。

SQLiteビルドの実験でも触れましたが、データベースの内部動作を理解することで、より堅牢なアプリケーションが書けるようになります。

実務で役立つテストパターン

同期バリアを活用したテストで、特に検証しておきたいパターンをまとめてみました。

  • Lost Update（更新の消失）: 2つのトランザクションが同じ行を読んで更新するケース
  • Phantom Read（ファントムリード）: 範囲クエリの結果がトランザクション中に変わるケース
  • Write Skew（書き込みスキュー）: 読み取った値に基づいて異なる行を更新するケース
  • Deadlock（デッドロック）: 複数のトランザクションが互いのロックを待つケース

これらのパターンをCI/CDパイプラインに組み込んでおけば、スキーマ変更やクエリの修正が並行処理に影響を与えないことを継続的に検証できます。

まとめ

レースコンディションのテストは「運に頼る」ものではありません。同期バリアを使えば、確実に並行処理の問題を再現し、修正を検証できるようになります。PostgreSQLのアドバイザリーロックやプログラミング言語のバリア機構を組み合わせることで、本番環境で起きる前にバグを潰せるのは大きなメリットですね。

データベースの並行処理に不安を感じている方は、まず小さなテストケースから始めてみてはいかがでしょうか。一度仕組みを作ってしまえば、あとはパターンを増やしていくだけです。