AIが書いた並行処理をテストする｜同時実行で崩れる条件を確かめる

並行処理の不具合は、通常のテストが通った後にも残ります。処理Aと処理Bを起動するだけでは、実行のたびに停止位置が変わり、失敗を再現できません。AIへ「並行処理をテストして」とだけ頼んだ場合も、何を競合させ、どの結果を合格とするかが実装側の推測に委ねられます。コードが生成されたという事実だけでは、テスト条件を説明したことになりません。

コードより先に条件台帳を作ります。形式はテストケースのコメントでも、チケットの表でも構いません。実行手順と受入条件を人が読み返せる形にし、AIには台帳どおりに停止と解放を制御するテストを依頼します。

台帳には、競合前のレコード・業務キー・入力という初期状態を置きます。停止点は読み取り後、ロック取得後、書き込み直前、外部I/O待機中など、テストから到達を観測できる地点です。解放順は同時解放か、Bの完了後にAを再開するかまで指定します。注入する事象も、同じ対象への更新、先行処理のコミットまたはロールバック、対象タスクのキャンセルへ分けます。観測対象は各処理の結果、例外、再試行、最終状態、待機者や未完了タスクの残留です。許す最終状態、拒否する上書き、二重反映の有無、タイムアウト時の終了方法と一対一で結び付けます。空欄が残るなら、AIへ渡す前に受入条件の決定が必要です。

条件台帳では「製品がどう動くか」と「業務として何を正しいとするか」を同じ欄に書きません。2026年9月21日時点の[PostgreSQL公式ドキュメント](https://www.postgresql.org/docs/current/transaction-iso.html)では、Read Committedで同じ行が先に更新されている場合、後続の更新は先行処理のcommitまたはrollbackを待ちます。先行処理がcommitすると、後続処理は更新後の行に対してWHERE条件を再評価し、条件を満たせば処理します。これはDBの挙動です。後から来た更新を採用するのか、古い読み取りに基づく更新を拒否するのかは、業務側の規則です。

同じ公式資料によれば、Read Committedでは同一トランザクション内の連続するSELECTが異なるデータを見る場合があります。Repeatable Readでは競合更新によってトランザクションがロールバックされ得るため、アプリケーションはトランザクション全体の再試行に備える必要があります。Serializableは、コミットできた並行トランザクション群が、いずれかの直列実行順と同じ効果になるよう監視し、異常条件でserialization failureを発生させます。分離レベルを上げても、業務規則までは決まりません。

台帳の左列には、利用中のDB、分離レベル、制約、実際に確認した例外や待機を置きます。右列には、重複排除、更新の優先順位、再試行の可否、利用者へ返す結果を置きます。AIには左列を公式仕様に合わせて実装させ、右列をassertionとして表現させます。これなら、DBを変えたときに見直す条件と、業務判断として維持する条件を切り分けられます。

重複更新では、同じ業務キーを扱う二つの処理を更新直前で確実にそろえます。2026年9月21日時点の[Python asyncio同期プリミティブの公式文書](https://docs.python.org/3/library/asyncio-sync.html)では、asyncio.Barrierは指定数のタスクがwait()へ到達するまで待機させ、全タスクが到達すると待機中のタスクを同時に解放します。双方が狙った地点に来た事実を、短い待ち時間の代わりに解放条件として使えます。

AIへの依頼は、たとえば次の内容です。「同じ業務キーを扱う処理AとBを用意する。両方を更新直前のバリアで待機させ、そろってから解放する。各処理の成功、競合、再試行の結果と最終状態を記録する。同じ業務イベントが二重反映されていないことを検証する。期待結果は、現在のDB、分離レベル、制約、アプリケーションの重複排除方針から勝手に補わず、未定なら失敗するテストとして示す」。

この依頼で判定したいのは、両方の処理が成功するかだけではありません。成功、競合、再試行のどれを許容するかは構成によって変わります。同じ業務イベントの二重反映を拒む方針なら、最終状態でその方針を検証します。タイムアウトは失敗とし、どのタスクがどの停止点に残ったかを出力させます。待機を終了させる条件も台帳へ加えることで、偶然通過したケースを合格から外せます。

順序逆転は解放の順番を固定して作ります。処理Aに対象を読み取らせた後で停止させ、処理Bを更新・コミットまで進め、それからAを再開します。Aが古い読み取り結果を持ったまま書き込みへ進む条件を狙うためです。実行速度や負荷による順序の変化を待つ必要はありません。

依頼文は「Aを読み取り後のチェックポイントで停止する。Bを更新・コミットさせ、その完了を観測してからAを再開する。Aの書き込みについて、古い値による上書きを許す、競合として拒否する、処理全体を再試行する、のいずれかを台帳の受入条件から取得する。DBが返した結果と業務上の判定を別々に記録する」とします。受入条件が空欄なら、一般論で補完させず、仕様決定が必要なテストとして残します。

「新しい更新を勝たせる」という表現には迷いが残ります。更新時刻を比べるのか、版を比べるのか、特定の状態遷移だけを許すのかでテストは変わるからです。DBが後続のUPDATEを待機させてWHERE条件を再評価しても、望む業務結果が必ず残るとは限りません。判定可能な条件に直せない表現は、実装前の未決事項として責任者へ返します。

中断テストでは、キャンセルを投げる時点を観測可能にします。候補はロック取得後、外部I/O待機中、書き込み直前です。各地点を別のケースにし、対象タスクがチェックポイントへ到達した通知を受けてからキャンセルします。ランダムな時刻に止めるだけでは、クリーンアップ前なのか後なのかを説明できません。

2026年9月21日時点の[Python asyncioタスク公式文書](https://docs.python.org/3/library/asyncio-task.html)では、タスクをキャンセルすると次の機会にCancelledErrorがタスク内で発生します。クリーンアップにはtry/finallyを使い、CancelledErrorを明示的に捕捉した場合は、クリーンアップ後に原則として再送出することが推奨されています。また、asyncio.Barrierの待機中タスクがキャンセルされると、そのタスクはバリアから抜け、filling状態なら待機数が減ります。abort()は待機中と将来のwait()をBrokenBarrierErrorにし、無期限の待機を避ける用途が示されています。

依頼は「指定したチェックポイントへの到達後に対象タスクをキャンセルする。finallyによるクリーンアップ、呼び出し元へのキャンセル伝播、未完了タスクと待機者が残らないこと、部分的な業務更新が残らないことを検証する。バリアが成立しなくなった場合はabort()等で待機側を終了させ、タイムアウトを合格条件にしない」と具体化します。キャンセル例外を握りつぶして処理継続する実装や、相手側だけが待ち続けるテストは失敗として観測します。

実行へ進める条件は三つです。対象処理の停止点をテストから観測できる。許容する最終状態を責任者が決めている。DB設定と業務規則を分けて記録している。sleepの長さだけに依存するテスト、再試行の上限や終了条件がない依頼、タイムアウトを成功とみなすケース、期待結果をAIに推測させる台帳は、コード生成の前に差し戻します。

条件が決まったテストは、通常の単体テストを置き換えるものではなく、競合する処理の動作確認として管理します。AI生成コードを通すレビューやCIはAI駆動開発の開発フロー｜レビュー・テスト・CIで人の判断を残すで整理できます。AIエージェントへテスト実行を任せる際の読み取り・実行範囲はClaude Codeのセキュリティ｜法人導入前に決める権限とデータ管理と合わせて決めます。

重複更新は更新直前にそろえ、順序逆転はAの読み取り後にBをコミットさせ、中断は観測可能な地点でキャンセルして後始末と伝播まで確認します。三つを曖昧な「同時実行テスト」にまとめず、停止点、解放順、残してよい結果を個別に確定する。それが自社で実行できる条件です。AIは条件を再現するコードを書けても、結果を受け入れる責任までは引き受けません。条件台帳を自社のコードや運用へ落とし込む判断が難しい場合は、AI駆動開発の研修・導入支援・技術顧問の初回60分無料相談で、対象範囲を整理できます。