現在、 AWS Graviton5 プロセッサを搭載した  Amazon EC2 の R9g インスタンスと R9gd インスタンスが一般的に利用できるようになっています。R9g インスタンスはメモリ最適化されており、AWS がこれまでに構築した中で最もエネルギー効率の高いプロセッサを搭載した Graviton4 ベースの R8g インスタンスと比較して、コンピューティングパフォーマンスが最大 25% 向上します。 R9g インスタンスは、データベース、インメモリキャッシュ (Valkey、Redis、MemCached)、リアルタイムのビッグデータ分析、コンテナ化されたマイクロサービスベースのアプリケーション (Kubernetes、Docker、EKS、ECS など) を含む Linux ベースのワークロード、C/C++、Rust、Go、Java、Python、.NET Core、Node.js、Ruby、PHP などの一般的なプログラミング言語で記述されたアプリケーションなど、メモリを大量に消費するワークロードに最適です。 R9gd インスタンスには、ローカルの NVMe ベースの SSD ブロックレベルストレージが含まれており、オープンソースデータベース、分散型リアルタイムビッグデータ分析、大規模なインメモリデータベース、大規模なキャッシュワークロードなど、高速で低レイテンシーのローカルストレージを必要とするメモリ集約型のワークロードに最適です。 現在 R8g インスタンスでワークロードを実行している場合、R9g を使用すると、消費電力を抑えながら、より高速なメモリ、より高いネットワークと Amazon EBS の帯域幅、大きな L3 キャッシュにより、vCPU あたりのパフォーマンスが向上します。 R9g は何が違うのか Graviton5 プロセッサは、Graviton4 に比べていくつかのハードウェア改善をもたらしています。 vCPU あたりのコンピューティングパフォーマンスが最大 25% 向上 DDR5 8800 MT/s メモリ（Graviton4 の 5600 MT/s から増加）、クラウドで利用可能な最速のメモリ L3 キャッシュが 5 倍大きく、データの局所性が向上 最大のインスタンスサイズでは最大 2 倍のネットワークと EBS 帯域幅 (最大 100 Gbps のネットワーク、48 xlarge では最大 72 Gbps の EBS) 最大 3 倍のパケット処理パフォーマンス R9g インスタンスと R9gd インスタンスは、Amazon EBS と Amazon VPC ネットワーキング間の帯域幅割り当てを 25% 調整できるインスタンス帯域幅設定 (IBC) をサポートしています。これにより、データベースやキャッシュなど、特定の帯域幅要件を持つワークロードのパフォーマンスを最適化できます。 すべての R9g および R9gd インスタンスは AWS Nitro System 上で動作し、仮想化、ストレージ、ネットワーキングを専用ハードウェアにオフロードします。これにより、インスタンス間の強固なセキュリティ分離を維持しながら、アプリケーションはベアメタルに近いパフォーマンスを実現できます。 R9g インスタンスと R9gd インスタンスには、今年初めに C9g と M9g インスタンスで導入された Nitro System と同じ拡張機能である Nitro Isolation Engine (NIE) が搭載されています。Nitro Isolation Engine (NIE) は、インスタンスの分離を強制し、正式な検証を利用して数学的な精度で分離を保証します。Nitro Isolation Engine は、仮想マシン間の分離を強制する役割を担う専用コンポーネントです。その役割には、最小限の一連の API を通じて、仮想マシンのメモリ、CPU レジスタ状態、I/O デバイスに対するあらゆるアクセスを仲介することが含まれます。Nitro Isolation Engine は形式検証を活用しています。形式検証とは、ハードウェアまたはソフトウェアが、特定のテストケースにおいてだけでなく、意図されたとおりに動作することを数学的に証明する手法です。この高度な検証手法により、Nitro は形式的に検証された初のクラウドハイパーバイザーとなっており、数学的に証明されたクラウドセキュリティの新たな標準を打ち立てています。Nitro Isolation Engine の詳細については、 ブログ投稿 をご覧ください。対象範囲や前提条件を含む形式検証の結果の詳細については、 テクニカルホワイトペーパー を参照してください。 EC2 R9g および R9gd インスタンスの仕様 R9gとR9gdのインスタンスはそれぞれ、ミディアムからメタル 48XL まで、11種類のサイズでご利用いただけます。次の表は、各サイズの完全な仕様を示しています。 インスタンスサイズ vCPU メモリ (GiB) インスタンスストレージ ネットワーク帯域幅 (Gbps) EBS 帯域幅 (Gbps) r9g.medium 1 8 EBS のみ 最大 15 最大 12 r9g.large 2 16 EBS のみ 最大 15 最大 12 r9g.xlarge 4 32 EBS のみ 最大 15 最大 12 r9g.2xlarge 8 64 EBS のみ 最大 17 最大 12 r9g.4xlarge 16 128 EBS のみ 最大 17 最大 12 r9g.8xlarge 32 256 EBS のみ 17 12 r9g.12xlarge 48 384 EBS のみ 25 18 r9g.16xlarge 64 512 EBS のみ 34 24 r9g.24xlarge 96 768 EBS のみ 50 36 r9g.48xlarge 192 1536 EBS のみ 100 72 r9g.metal‑48xl 192 1536 EBS のみ 100 72 R9gd インスタンスは、高速で低レイテンシーのスクラッチスペースまたは一時キャッシュを必要とするワークロード向けに、ローカル NVMe ベースの SSD ストレージを追加することで、R9g と同じコンピューティング性能とネットワークパフォーマンスを提供します。 インスタンスサイズ vCPU メモリ (GiB) インスタンスストレージ (NVMe SSD) ネットワーク帯域幅 (Gbps) EBS 帯域幅 (Gbps) r9gd.medium 1 8 1 x 59 GB 最大 15 最大 12 r9gd.large 2 16 1 x 118 GB 最大 15 最大 12 r9gd.xlarge 4 32 1 x 237 GB 最大 15 最大 12 r9gd.2xlarg 8 64 1 x 474 GB 最大 17 最大 12 r9gd.4xlarge 16 128 1 x 950 GB 最大 17 最大 12 r9gd.8xlarge 32 256 1 x 1900 GB 17 12 r9gd.12xlarge 48 384 3 x 950 GB 25 18 r9gd.16xlarge 64 512 1 x 3800 GB 34 24 r9gd.24xlarge 96 768 3 x 1900 GB 50 36 r9gd.48xlarge 192 1536 3 x 3800 GB 100 72 r9gd.metal‑48xl 192 1536 3 x 3800 GB 100 72 使用の開始 サポートされている任意の ARM ベースの AMI を使用して、 Amazon EC2 コンソールから R9g インスタンスと R9gd インスタンスを起動できます。R9g インスタンスは、Amazon Linux 2023、Amazon Linux 2、Ubuntu 22.04+、RHEL 8.4+、SUSE Linux エンタープライズサーバー 15 SP3+、Debian 12+、およびその他の主要な Linux ディストリビューションをサポートしています。 R8g から移行する場合、ほとんどのアプリケーションではコードを変更する必要はありません。同等の R9g インスタンスサイズを選択すると、アプリケーションのパフォーマンスが向上します。コンテナ化されたワークロードの場合、R9g は Amazon EKS、 Amazon   ECS 、および標準の Kubernetes デプロイメントと連携します。Arm64 用に構築されたマルチアーキテクチャコンテナイメージは変更なしで動作します。 始めるのに役立つリソースはいくつかあります。 AWS Graviton 入門ガイドでは 、Graviton ベースのインスタンスでワークロードを構築、実行、最適化する方法について説明しています。 Graviton節約ダッシュボードは 、コスト削減を追跡するのに役立ちます。 AWS Transform は Java アプリケーションを x86 から Graviton に移行するためのコード変換を自動化します。詳細については、 AWS Graviton プロセッサをご覧になるか、AWS Graviton でコンピューティングをレベルアップしてください 。 料金と利用可能なリージョン Amazon EC2 R9g および R9gd インスタンスは、米国東部 (バージニア北部、オハイオ)、米国西部 (オレゴン)、および欧州 (フランクフルト) リージョンで利用可能です。 R9g および R9gd インスタンスは、Savings Plans、オンデマンド、スポットインスタンス、ハードウェア専有インスタンス、または専有ホストを通じて購入できます。詳細な料金については、「 Amazon EC2 の料金 」ページにアクセスしてください。 始める準備はできましたか? Amazon EC2 コンソール から R9g インスタンスを起動してください。詳細については、 Amazon EC2 R9g インスタンスページを参照してください 。 API を呼び出したり、ドキュメントを検索したり、リージョンごとの提供状況を確認したり、この新機能に関するトラブルシューティングを確認したりする場合は、お好みの AI ツールで AWS MCP Server と プラグイン を使用してみてください。 Amazon EC2 用 AWS re:Post でフィードバックを共有するか、通常の AWS サポートの連絡先から連絡してください 。 – Daniel Abib 原文は こちら です。

はじめに 前提：エージェントの構成 課題：画面を離れると回答が消える 設計方針：実行の作成と購読を分離する DB スキーマ：残す履歴と実行中の状態を分ける 保存単位は「AG-UI の Message 1 件 = 1 行」 DynamoDB ではなく Aurora MySQL を選んだ理由 会話履歴はサーバーが組み立てる 書き込み設計：イベントは保存せず「生成途中の回答」を上書きする 再合流：会話全体のスナップショットで追いつく 同時実行制御：実行中ロックを NULL 可のユニーク列で作る ロックの解放漏れに備える 停止：切断とキャンセルを区別する Redis は要るか まとめ はじめに 開発本部 開発1部の いくまる です。 私たちのチームでは、Web アプリの新機能として、チャット形式でデータを分析できる AI エージェントを開発中です。開発を進める中で、「回答の生成中に画面を離れると、その回答を受け取れなくなり、会話も残らない」という課題に向き合うことになりました。 本記事では、この課題を解決するために行った「会話履歴の永続化」と「バックグラウンド実行」の設計と実装を紹介します。DB スキーマ・実装コード・検討して捨てた案まで含めて書きます。 前提：エージェントの構成 このエージェントは次の構成で動いています。 ブラウザ（チャット UI） │ AG-UI イベント（SSE） ▼ Next.js（API Route ── ブラウザと AgentCore の中継役） │ InvokeAgentRuntime（SSE） ▼ Amazon Bedrock AgentCore Runtime（Strands Agents 製エージェント） │ MCP ▼ MCP サーバー（自社データの検索・集計ツール群） Amazon Bedrock AgentCore : AI エージェントの実行基盤となる AWS のサービスです。セッションごとに microVM 単位で実行環境が分離されます。 Strands Agents : AWS が公開しているオープンソースの AI エージェント SDK です。 AG-UI : エージェントとフロントエンド間のイベントストリーミングのプロトコルです。 RUN_STARTED ・ TEXT_MESSAGE_CONTENT ・ TOOL_CALL_* ・ RUN_FINISHED などのイベント型を定めています。転送方法は SSE に限定されませんが、このアプリでは SSE で流しています。 ユーザーが質問を送ると、エージェントが MCP ツールでデータを取得・分析し、回答をストリーミングで返します。ツールを繰り返し呼ぶため、1 回の回答に数十秒かかることがあります。 課題：画面を離れると回答が消える AI チャットで広く使われるのは「POST + ストリーミング応答」の構成です。ブラウザが質問を POST し、サーバーが生成イベントを流し、画面に逐次表示します。画面離脱を想定しなければ、これで十分に機能します。私たちの初期実装もこの構成でした。 私たちの場合は回答に数十秒かかるため、「生成中に画面を離れても実行は完走してほしい」という要件が加わりました。この要件を満たそうとすると、3 箇所が問題になります。 改修前の構造。画面遷移した瞬間に、以降のイベントを受け取る手段がなくなる 1 つ目はフロントエンドです。 この構成では、チャット画面の hook が unmount 時に実行を中断（ abortRun() ）する作りになりがちです。画面遷移がそのまま実行中断になります。 // チャット画面の hook（unmount 時に実行を中断する作り） useEffect( () => () => { stopRequestedRef. current = true ; agentRef. current ?.abortRun(); } , [] , ); 2 つ目は中継役の API Route です。 ブラウザと AgentCore の間で SSE を中継する Next.js の API Route は、クライアントの切断を上流の AgentCore への読み取りキャンセルとして伝播します。 ReadableStream の cancel() は「クライアントがもう読まない」ときに呼ばれるコールバックで、そこで上流の読み取りも止めると、切断とキャンセルの区別がなくなります。 // 中継処理（クライアントが切れると上流の読み取りも止まる作り） const readable = new ReadableStream( { async start ( controller ) { // 上流（AgentCore）の SSE を読み、そのままクライアントへ中継する } , cancel () { reader.cancel(); } , } ); 3 つ目は保存先です。 イベントは中継されるだけで、どこにも保存されません。仮に 1 つ目と 2 つ目を直して実行が完走するようにしても、戻ってきた画面に表示するデータがありません。 「実行状態を React のグローバルストアに持てば、画面遷移に耐えられるのでは」という案も検討しました。しかしこのアプリでは、チャット画面から他の画面への遷移が window.location.href によるページ全体の再読み込みで実装されています。再読み込み後のページは JavaScript の実行環境ごと新しく作られるため、React の state やグローバルストアに入れた値は引き継がれません。 設計方針：実行の作成と購読を分離する 大きく変えたのは次の 3 つです。 作成と購読の分離 : POST /runs は実行を開始して 202 { runId } を即座に返します。表示は GET /runs/{runId}/events の SSE で受け取ります。この「SSE を受信し続けること」を、本記事では「購読」と呼びます。購読はいつ切れてもよく、何度でも再開できます。 実行ワーカーの独立 : AgentCore の SSE を最後まで読み切って記録する処理（実行ワーカー）を、HTTP レスポンスから独立した非同期タスクにしました。ブラウザが切断しても実行は完走します。 二層の保存 : 実行中は「生成途中の回答」を DB に上書き保存し続け、完了したら完成したメッセージを DB の履歴テーブルに保存します。テーブル構成は次の節で説明します。 なお、実行ワーカーは Next.js と同じプロセス内で動かしているため、リクエストごとに実行環境が終了するサーバーレス環境ではこの形は取れません。現在は検証段階のためこの構成にしていますが、Next.js のデプロイに走行中の実行が巻き込まれないようにするため、本来は実行ワーカーを独立したプロセスに切り出す方が望ましいと考えています。現状、プロセスがデプロイなどで止まる場合の後始末は、同時実行制御の節で説明する回収の仕組みが担います。 改修後の構造。実行は接続と無関係に完走し、購読は何度でも再入場できる ブラウザとサーバーの間の API は次の 5 本です。 API 役割 POST /runs 実行を作成して 202 { runId, conversationId } を即返す GET /runs/{id}/events SSE 購読。切断・再入場が自由 POST /runs/{id}/cancel 明示的なキャンセル GET /conversations 会話一覧（履歴サイドバー用） GET /conversations/{id} 会話の全メッセージ + 実行中の run（あれば） 最後の API がポイントです。リロードや別タブで会話を開いた直後、クライアントは会話 ID しか知らず、実行中の run があるかどうかも分かりません。そこで GET /conversations/{id} は、会話のメッセージに加えて「実行中の run の ID」を返します。クライアントはその ID で GET /runs/{id}/events を購読し、生成途中から表示を再開します。 DB スキーマ：残す履歴と実行中の状態を分ける このアプリでは以前から、本体機能のデータを Aurora MySQL 8.0 + Prisma で管理しています。エージェントの履歴も同じ DB に、3 つのテーブルで持つことにしました。ずっと残す「履歴」と、実行中だけ使う「実行状態」でテーブルを分けています。 区分 テーブル 役割 行の扱い 履歴 conversations 会話スレッド 1 件のメタ情報 ずっと残す 履歴 conversation_messages メッセージ 1 件 = 1 行。完成した発話を保存 ずっと残す 実行状態 runs 実行 1 回の状態 + 生成途中の回答 + ロック 行は実行 1 回ごとに増え、終了後も記録として残る。生成途中の回答やロックは実行中だけ使う ER 図 Prisma スキーマは次の通りです。実際に採用したものから、タイムスタンプ列・リレーション定義・enum 定義（RunStatus / MessageStatus）を省いています。 model Conversation { id String @id @default(cuid()) companyId Int @map("company_id") userId String @map("user_id") threadId String @map("thread_id") @db.Char(36) title String @db.VarChar(255) deletedAt DateTime? @map("deleted_at") @@unique([companyId, userId, threadId]) @@map("conversations") } model ConversationMessage { id String @id @default(cuid()) conversationId String @map("conversation_id") runId String? @map("run_id") sequence Int role String @db.VarChar(16) parts Json status MessageStatus @default(complete) @@unique([conversationId, sequence]) @@map("conversation_messages") } model Run { id String @id @default(cuid()) conversationId String @map("conversation_id") clientTurnId String @map("client_turn_id") @db.VarChar(64) status RunStatus @default(queued) lockKey String? @unique @map("lock_key") ownerInstanceId String? @map("owner_instance_id") @db.VarChar(64) errorCode String? @map("error_code") @db.VarChar(64) partialState Json? @map("partial_state") heartbeatAt DateTime @default(now()) @map("heartbeat_at") @@unique([conversationId, clientTurnId]) @@index([status, heartbeatAt]) @@map("runs") } lockKey の UNIQUE、 clientTurnId の複合ユニーク、 [status, heartbeatAt] のインデックスがそれぞれ何のためにあるかは、後の節で順に説明します。 また、本記事には 4 種類の ID が登場します。ここで整理しておきます。 ID 何を指すか conversationId DB 上の会話。API で会話を指すときに使う threadId AG-UI 上の会話 ID。DB 上の会話（ conversationId ）と 1 対 1 で対応 runId 質問 1 回ぶんの実行 runtimeSessionId AgentCore の実行環境を束ねる ID。 t{companyId}-u{userId}-{threadId} の形式で、会話ごとに固定 保存単位は「AG-UI の Message 1 件 = 1 行」 conversation_messages は追記専用で、AG-UI の Message をそのまま parts （JSON）に格納します。テキストだけのターンは user / assistant の 2 行、ツールを使うターンは assistant（ツール呼び出し）と tool（結果）の行が挟まって 4 行以上になります。 1 会話のメッセージ行の例。ツールを使うターンは user・assistant（ツール呼び出し）・tool・assistant の 4 行、使わないターンは 2 行になる この保存単位を選んだ理由は、フロントの表示ロジックの作りにあります。ライブ表示は「AG-UI の Message 配列を受け取り、ターンの区切りやツールの実行ステップ表示を組み立てる純粋関数」として自前で実装しています。保存した Message 列をそのままこの関数に渡せば、画面を離れなかった場合と同一の表示が再現されます。保存時に表示用の形へ加工してしまうと、同じ表示を再現できなくなります。 DynamoDB ではなく Aurora MySQL を選んだ理由 会話履歴のアクセスパターン（会話 ID + 連番で順に全件取得、追記専用、JSON 主体）は DynamoDB の得意領域で、実際に移行案も検討しました。それでも Aurora MySQL 一本にしています。 まず、トランザクション要件が構成の選択肢を絞ります。質問の受付時には「会話 + user メッセージ + 実行（ロック）の INSERT」を、完了時には「assistant メッセージの INSERT + 実行の完了 + ロック解放」を、それぞれ単一トランザクションで行う必要があります。「履歴は DynamoDB、実行状態は Aurora」のように 2 つのストアに分けると、この原子性を保証できません。原子性が無いと、たとえば次のような壊れ方をします。 回答は残ったのに次の質問ができない : 完了処理の「回答を保存」と「ロック解放」の間でプロセスが落ちると、画面には回答が出ているのに DB はロックを握ったままになり、次の質問が「実行中です」と拒否され続けます。 答えのない質問が履歴に残る : 送信の二度押しで 2 本目が「質問を保存 → ロックで弾かれる」の順に進むと、誰も回答しない質問だけが履歴に残ります。1 トランザクションならロック取得の失敗と同時に質問の保存も取り消され、エラー応答だけを返せます。 したがって選択肢は「全部 Aurora」か「全部 DynamoDB」に絞られます。後者も技術的には成立します（DynamoDB でも TransactWriteItems で複数の項目をまとめて原子的に書けます）。 それでも Aurora にしたのは、既存の運用との一貫性のためです。このアプリの他のデータはすべて Aurora + Prisma で管理していて、マイグレーションの手順やレビューの観点といったチームの運用もそこで揃っています。データストアを 2 つにすると、この運用も 2 系統になります。規模の面でも、DB への書き込みはピークでも毎秒数十回の見積もりで、Aurora で十分に受けられます。DynamoDB のスケール性能が必要になる水準ではありません。 会話履歴はサーバーが組み立てる エージェントは毎回の呼び出しで会話の全履歴を受け取り、状態をゼロから組み立て直す作りにしています。この全履歴を誰が用意するかには 2 つの形があります。クライアントが手元の Message 配列を毎回送るか、 サーバーが DB から組み立てる かです。私たちは後者にしました。クライアントが送るのは新しいメッセージ 1 件だけです。 前者を避けた理由は、バックグラウンド実行と相性が悪いからです。実行を放置して別のタブで完走させると、元のタブが持っている履歴は古いままになります。その古いタブから次の質問を履歴ごと送ると、完走したはずの回答がモデルへの入力から抜け落ち、会話のつじつまが合わなくなります。後述するロックは実行中しか効かないため、この事故は防げません。最新の会話を常に持っているのは DB だけです。 なお、AgentCore 側に会話の状態を持たせる案も 2 つ検討し、見送りました。 実行環境（microVM）のメモリに持つ : セッション ID は会話ごとに固定なので、同じ会話の呼び出しは同じ実行環境に届き、メモリに状態を残すこと自体はできます。ただしこの環境は無操作 15 分（デフォルト）などで終了し、メモリごと消えます。時間を空けて続く会話の置き場にはできません Memory サービスに持つ : AgentCore には会話を保存する Memory というサービスもあります。ただし履歴はどのみち表示のために自前の DB へ保存するので、足すと同じ役割の保存先が 2 つになります POST /runs のボディは { conversationId, message, clientTurnId } だけです。モデルへ渡す履歴は、実行ワーカーが Aurora から組み立てます。 // 実行ワーカーの一部：DB から会話履歴を読み出し、モデル入力用に整える export async function buildModelMessages ( conversationId : string ): Promise < ModelMessagesResult > { const rows = await prisma.conversationMessage.findMany( { where : { conversationId , status : "complete" } , orderBy : { sequence : "asc" } , select : { parts : true } , } ); return normalizeHistory(rows); } // 実行ワーカーの一部：組み立てた履歴をエージェントに渡して生成を開始する const history = await buildModelMessages(claimed.conversationId); const { companyId , userId , threadId } = claimed.conversation; const agent = new AgentCoreAgent( { threadId , initialMessages : history .messages, agentArn , runtimeUserId : `c ${ companyId } :u ${ userId } ` , runtimeSessionId : `t ${ companyId } -u ${ userId } - ${ threadId } ` , } ); こうすると、会話の内容はサーバー（DB）だけが持つ構造になります。AgentCore の microVM がタイムアウトで終了しても、クライアントがリロードで状態を失っても、会話は Aurora から再構成できます。 書き込み設計：イベントは保存せず「生成途中の回答」を上書きする 1 回の回答で AG-UI イベントは数十〜数百個流れます。本文の断片（デルタ）1 つ 1 つやツール呼び出しがそれぞれイベントになるため、回答が長いほど増えます。これを 1 行ずつ INSERT すると、1 回答ごとに大量の行が永久に積み上がります。採用したのは次の形です。 時点 DB 操作 内容 質問送信 INSERT conversations に会話（新規会話のときのみ）、 conversation_messages に質問、 runs に実行レコード（ロック取得を兼ねる）の最大 3 行 生成中 runs.partial_state を上書き UPDATE AG-UI のイベントでは本文が細切れ（デルタ）で届く。実行ワーカーはそれをつなぎ合わせた「その時点のメッセージ配列」を保持しており、これを数秒ごとに同じ 1 行へ上書き保存。行は増えない 完了 conversation_messages に INSERT そのターンで生まれたメッセージ（回答、ツールを使った場合はその呼び出しと結果も）を保存。 runs.partial_state を空にし、ロックを解放 質問送信時のトランザクションは以下のようになっています。ロック（ lockKey ）の取得と質問の保存が、同時に成立するか同時に失敗するかのどちらかになります。 // API Route の一部：質問受付時の書き込み return prisma.$transaction( async ( tx ) => { const conversationId = existingId ?? ( await tx.conversation.create( { /* 省略 */ } )).id; // ロック取得に失敗したら、下で保存する質問ごと取り消される（同一トランザクションのため） const run = await tx.run.create( { data : { conversationId , clientTurnId , lockKey : conversationId } , select : { id : true } , } ); // aggregate（集計クエリ）で会話内の最大 sequence を取り、次の連番を振る const highest = await tx.conversationMessage.aggregate( { where : { conversationId } , _max : { sequence : true } , } ); await tx.conversationMessage.create( { data : { conversationId , runId : run. id , sequence : (highest._max.sequence ?? 0 ) + 1 , role : "user" , parts : { id : randomUUID(), role : "user" , content : message } , } , select : { id : true } , } ); return { runId : run. id , conversationId } ; } ); 生成中の partial_state は 3 秒間隔で間引いて書きます。間引きに加えて「前の UPDATE が完了するまで、次の UPDATE を発行しない」という制御も入れています。UPDATE を発行した順と DB に反映される順は一致するとは限らないため、古い内容の UPDATE が新しい内容の後に適用されると、保存済みの「生成途中の回答」が巻き戻ってしまうからです。 // 実行ワーカーの一部：生成途中の回答を数秒ごとに DB へ上書き保存する const PARTIAL_STATE_INTERVAL_MS = 3_000 ; return { schedule() { // 前の書き込みが完了するまで次をスケジュールしない（古い内容への巻き戻りを防ぐ） if (timer || writing || truncated) return ; timer = setTimeout (() => { timer = null ; writing = true ; void writePartialState(runId, ownerInstanceId, produced()) . then (( state ) => { truncated = state.truncated; } ) . catch (( error : unknown ) => console .error( "partial_state の更新に失敗しました" , { runId , error } ), ) . finally (() => { writing = false ; } ); } , PARTIAL_STATE_INTERVAL_MS); } , } ; 数秒おきに UPDATE を発行し続けて DB の負荷は大丈夫なのか、という点は検討しました。結論としては、同時に走る生成が多くても数十本という規模では問題になりません。更新は各実行が自分の 1 行だけを主キー指定で行い、実行間のロック競合はありません。さらに、接続中のユーザーの画面へは実行ワーカーがメモリ上のイベントを直接流すため、 partial_state の用途は後述する再合流だけです。毎秒書く必要も、イベントを 1 個ずつ書く必要もありません。 完了時は「回答の確定保存」「実行ステータスの完了への更新」「ロック解放」「生成途中の回答（ partial_state ）の削除」を 1 トランザクションで行います。 // 実行ワーカーの一部：完了時の書き込み。lock_key を外し損ねるとその会話が永久に 409 になる return prisma.$transaction( async ( tx ) => { const claimed = await tx.run.updateMany( { where : terminableWhere(runId, ownerInstanceId), data : { status , errorCode , finishedAt : new Date (), lockKey : null , partialState : Prisma.DbNull, } , } ); // 別の経路（キャンセルや、後述する異常終了時の回収処理）が先にこの実行を // 終わらせていたら、生成物は保存しない if (claimed. count === 0 ) return false ; await tx.conversationMessage.createMany( { data : messages. map (( message , index ) => ( { conversationId : targetId, runId , sequence : base + index + 1 , role : message.role, parts : message as Prisma.InputJsonValue , status : messageStatusAt( status , message. id , openMessageIds), } )), } ); return true ; } ); メッセージの status は通常 complete で保存します。キャンセルやエラーで実行が正常に終わらなかった場合は、そこまでに生成できていた分を partial（部分的、の意味）として保存し、画面に残せるようにしています。 再合流：会話全体のスナップショットで追いつく 実行中の会話に購読者が入ってくると、サーバーはまず RUN_STARTED （「実行が進行中です」の合図）と MESSAGES_SNAPSHOT を送ります。どちらも上流から届いたイベントの中継ではなく、この購読のためにサーバーが新しく作って送るものです。 MESSAGES_SNAPSHOT を受け取ったクライアントは、手元のメッセージ一覧を捨てて、スナップショットの内容で丸ごと置き換えます。そのため、スナップショットに生成途中の 1 件だけを入れると、過去のメッセージが画面からすべて消えてしまいます。必ず会話の全メッセージを入れて送ります。 その後の配信は 2 つのモードに分かれます。分かれ目は「購読がいつ始まったか」です。 モード いつ使われるか 配信内容 live 質問の送信直後から購読している場合（生成イベントがまだ 1 件も流れていないうちに購読が始まったとき） 実行ワーカーが受け取る生成イベント（ TEXT_MESSAGE_CONTENT など）を、メモリからそのまま逐次中継 poll それ以外すべて（リロード・別タブ・離脱して戻ってきた場合） 1 秒間隔で DB を読み、確定済み履歴と partial_state の生成途中回答をマージした会話全体の MESSAGES_SNAPSHOT を、内容が変わったときだけ送り直す。実行が終わったら RUN_FINISHED （または RUN_ERROR ）で締める つまり、途中から戻ってきた購読者が受け取るのは live 配信のイベント列ではなく、「会話全体のスナップショットの送り直し」です。 // 配信モードの選択。イベントが 1 件でも流れた後に始まった購読は poll に回す export function attach ( runId : string , signal : AbortSignal ): LiveSubscription | null { const fanout = fanouts. get (runId); if (!fanout) return pollInstead(runId, "fanout_absent" ); if (fanout. closed ) return pollInstead(runId, "fanout_closed" ); // 1 件でも中継済みなら列の途中からになるので、履歴を出せる poll に任せる if (fanout.relayed > 0 ) return pollInstead(runId, "already_relayed" ); // 省略（購読者を登録し、生成イベントを流す AsyncGenerator を返す） } // poll 配信のループ。会話全体のスナップショットを、内容が変わったときだけ送り直す while ( true ) { const event = snapshot(stored, readPartialMessages(progress.partialState)); const serialized = JSON . stringify (event); if (serialized !== previous) { previous = serialized; yield event; } if (isTerminal(progress. status )) break ; await sleep(POLL_INTERVAL_MS, signal); progress = await readProgress(run. id ); if (isTerminal(progress. status )) stored = await loadConversationMessages(run.conversationId); } yield terminalEvent(run, progress); 途中合流の購読者を live のイベント列に合流させず poll に回すのは、正しさを優先したためです。デルタの続きから流すには、「スナップショットに含めた分」と「これから流すデルタ」の境界を厳密に合わせる必要があります。境界がずれると、 content += delta の積み上げで本文が二重に連結されます。会話全体のスナップショットを送り直す形なら、毎回が丸ごとの置き換えなので、この事故が原理的に起きません。その代わり、poll 配信の画面は live 配信のようなストリーミング表示にはならず、数秒おきに文章がまとまって進む表示になります。途中合流でもストリーミング表示にすることは、後続の課題にしています。 同時実行制御：実行中ロックを NULL 可のユニーク列で作る 「同一会話に実行中の run は 1 本だけ」を DB で強制します。PostgreSQL なら 部分インデックス （partial index。 CREATE UNIQUE INDEX ... WHERE status IN ('queued','running') のように、条件を満たす行だけへ一意制約をかけられます）で書けますが、MySQL 8.0 には相当する構文が用意されていません。 代わりに runs.lock_key （NULL 可・UNIQUE）を使いました。実行中は lock_key = conversationId 、終了時に NULL へ戻します。MySQL のユニークインデックスは NULL を重複として扱わない ため、終了済みの run は何本でも共存でき、実行中は会話ごとに 1 本に絞られます。 同一会話への 2 本目の POST /runs は、INSERT 時にユニーク制約違反のエラーとして原子的に弾かれます。重複には 2 種類あります。1 つは「同じ送信の二度押し」です。クライアントは送信 1 回ごとに ID（ client_turn_id ）を発行し、リトライでも同じ ID を送るため、この列の重複で検出できます。送信ボタンの連打はフロントでも抑止できますが、ネットワーク不調時の自動再送などフロントの制御では防げない経路が残るため、DB でも守ります。もう 1 つは「別の質問の並行送信」（ lock_key の重複）で、同じ会話を複数のタブで開いているときに起きます。どちらだったかを引き直して、応答を分岐します。 // API Route の一部：重複キーエラーの解釈 } catch (error) { if (!isUniqueViolation(error)) throw error; // 二度押しなら、先行の run をそのまま返す（同じ送信は 1 回として扱う） const raced = await findRunByClientTurnId(params); if (raced) return { ok : true , runId : raced. id , conversationId : raced.conversationId } ; // 並行送信なら、実行中の run を添えて拒否へ const activeRunId = await findActiveRunIn(conversationId); if (activeRunId) return { ok : false , reason : "active_run" , activeRunId } ; } // API Route の一部：並行送信への応答 if (result.reason === "active_run" ) { return Response .json( { errors : "この会話はいま実行中です" , activeRunId : result.activeRunId } , { status : 409 } , ); } 409 のレスポンスに activeRunId を含めているのは、UI がそれを使って「拒否」ではなく「実行中の run への購読切り替え」に変換できるようにするためです。 ロックの解放漏れに備える ロックには解放漏れへの備えも必要です。サーバーのプロセスが突然落ちると、running のままロックを握った run が残ります。そうなると、誰も実行していないのにその会話への質問が「実行中です」と拒否され続けます。備えは 2 つの仕組みの組み合わせです。 実行ワーカーは、実行中の run の heartbeat_at を定期的に現在時刻へ更新します（処理が続いていることの記録です） それとは別の掃除処理が、 heartbeat_at の更新が一定時間止まっている queued / running の run を「担当プロセスが異常終了した」とみなして failed にし、ロックを解放します。 heartbeat_at は INSERT 時に現在時刻が入るため、202 を返した直後・実行が始まる前にプロセスが落ちて queued のまま残った run も、この経路で回収されます（スキーマの @@index([status, heartbeatAt]) はこの検索用です） 「サーバー起動時に、残っている running を全部 failed にする」というより単純な方法は採れませんでした。デプロイ中は新旧のサーバーがしばらく同時に動いており、旧サーバーがまだ実行している最中の run を、新サーバーの起動処理が誤って failed にしてしまうためです。run に owner_instance_id （どのサーバーがその実行を担当しているか）を持たせているのも同じ理由です。掃除処理は、自分のサーバーがいま実行している run を誤って回収しないよう、この ID とメモリ上の実行一覧を突き合わせて判定します。 この回収の仕組みは、デプロイやスケールインでプロセスごと止められた場合の後始末も兼ねています。止まったプロセスが抱えていた実行は途中から再開できませんが、heartbeat が途絶えるため数分以内に failed になり、そこまでの生成分は partial として履歴に残り、会話のロックも解放されます。ユーザーは失敗を確認して、すぐ次の質問に進めます。 停止：切断とキャンセルを区別する この設計では、タブを閉じる・画面を遷移するのは「切断」であり、実行は継続します。明示的に止めたいときは POST /runs/{id}/cancel を呼びます。実行ワーカーにキャンセル要求の印を立てて上流への購読を切り離し、実行を「キャンセル」として記録します。記録後に遅れて届いた生成物は、前述の完了時トランザクションの「別の経路が先に終わらせていたら保存しない」分岐で破棄されます。 // API Route の一部：キャンセル処理 if (getActiveRun(runId) || run.ownerInstanceId === OWNER_INSTANCE_ID) { // 終了の記録は実行ワーカーに任せる。ここで書くと、ワーカーが「先に終了済み」と判定して生成物を捨てる const active = registerRun(runId); active.cancelRequested = true ; await active.agent?.detachActiveRun(); return { ok : true } ; } // 別のサーバーが担当している run。会話のロックを解放するために記録だけ書く await finalizeRun( { runId , status : "cancelled" , finalizedBy : `cancel: ${ OWNER_INSTANCE_ID } ` } ); 実行中の会話への追加送信は、前述のロックにより 409 で拒否されます。ただし 409 を返すだけだと、「戻ってきたら画面が止まって見える → もう一度送る → エラー」という流れになりやすいため、途中経過の可視化（再合流）を初回リリースの範囲に含めています。実行中であることが見えていれば追加送信は起きにくく、方向を変えたい場合も「停止してから送る」導線に誘導できます。 Redis は要るか 同種の設計では、実行中イベントの共有に Redis（Redis Streams）を使う構成がよく知られています。ただし Redis が必要になるのは「タスクを 2 つ以上に増やし、かつストリーミング表示を保ちたい」場合です。今回はどちらにも当てはまらないため、入れていません。 現在は ECS 1 タスクで動かしており、通常時は再接続のリクエストが実行ワーカーと同じプロセスに届きます。イベントはプロセス内のメモリで手渡せるため、Redis なしで live 配信が成立します。 タスクを 2 つ以上に増やすと、購読のリクエストが実行ワーカーのいない方のタスクへ届くことがあります。live 配信はワーカーと同じプロセスのメモリを介して成り立っているため、別のタスクに届いた購読では使えません。ただし DB はどのタスクからも読めるので、poll 配信はそのまま動きます。つまり live 配信できたはずの購読が poll 配信になり、ストリーミング表示が数秒おきの更新になるだけで、履歴も途中経過も見られます。設計方針の節で触れた「実行ワーカーを独立したプロセスに切り出す」場合も、ワーカーと購読者が必ず別プロセスになるため、同じく中継が必要になります。当面は 1 タスクで足りる規模のため、現時点ではこの構成にしています。 まとめ 実行の作成と購読を分離し、実行ワーカーを HTTP 接続から独立させることで、画面を離れても実行が完走する構造にしました。 履歴は「メッセージ 1 件 = 1 行」でずっと残し、生成途中の回答は上書き更新の 1 行に分けました。イベントの逐次保存はせず、モデルへ渡す履歴もサーバーが DB から組み立てます。 同時実行制御は MySQL の「NULL 可ユニーク列」によるロックで実現しました。キャンセルは購読の切り離しと、実行を「キャンセル」として記録することで実現し、切断とは明確に区別しています。 AI チャットの「履歴」と「バックグラウンド実行」は別々の機能に見えますが、作ってみると、どちらも「会話の状態はサーバー側で持つ」という同じ設計に行き着きました。AI チャットの実行基盤を作る際に共通して現れる論点だと思うので、同じものを作る方の参考になれば幸いです。

はじめに こんにちは、販促基盤ブロックの村井です。普段は販促基盤の開発を担当しています。 販促基盤は、ZOZOTOWNにおけるキャンペーン・割引・友だち紹介機能など販売促進の施策を横断的に支える共通基盤です。2025年に構築が始まり、初期スコープとして友だち紹介機能のリプレイスを行いました。友だち紹介機能は、既存会員が非会員を招待し、条件を満たすと双方にポイントを付与する機能です。本記事では、基盤の立ち上げ時に行ったO/Rマッパーの選定と、Auroraの読み書き分離でつまずいた話をご紹介します。 目次 はじめに 目次 背景・課題 O/RマッパーにDoma 3を選んだ理由 Domaでの実装パターン インフラ層のクラス構成 ドメインクラスで値オブジェクトを使用 Entityをテーブルの表現に徹させる Entityとドメインオブジェクトの変換 RepositoryImplでDBの例外を業務の例外へ変える 条件分岐をSQLテンプレートで書く 読み書きの振り分け 読み書きを分けた理由 1つ目の構成：判定の順序でつまずく 2つ目の構成：TransactionManagerを2つに分ける 調査：コネクションの取得回数を数える 3つ目の構成：DataSourceの参照先を揃える ThreadLocalを使う上での注意 別解：LazyConnectionDataSourceProxy まとめ 背景・課題 販促基盤には、エントリーや抽選といった複数種類の施策を1つの基盤で扱う構想を置いており、初期フェーズでは友だち紹介機能から着手しました。 友だち紹介機能を最初の対象に選んだ理由は2つあります。 1つは、既存の実装を流用できない状態だったことです。既存の友だち紹介機能は数年前に作った機能を暫定対応として運用しており、そのまま引き継げるソースコードがありませんでした。もう1つは機能不足です。ビジネス側が実施したい施策を、既存機能では実現できませんでした。そのため既存のソースコードとデータを引き継がず、新規開発として作り直す判断をしました。 この基盤では、立ち上げ時にJava 25とSpring Boot 3.5を土台にオニオンアーキテクチャを採用しました。ドメイン層を中心に置き、その外側にアプリケーション層、インフラ層、プレゼンテーション層を並べる構成です。層をまたぐ依存の向きはArchUnitのテストで検査しています。ドメイン層からインフラ層を参照するコードを書くと、このテストが失敗してビルドが通りません。 この方針を決めたあと、論点になったのがO/Rマッパーの選定です。求めたのは次の2点でした。 ドメイン層に永続化の都合を持ち込まない 発行されるSQLを自分たちで制御できる 1点目は、オニオンアーキテクチャを採用した時点で譲れない条件です。2点目は基盤の構想から来ています。扱う施策の種類が増えるほど、検索の条件は複雑になります。発行されるSQLの中身が見えないと、性能の問題が起きたときに原因を追えなくなります。 以下では、選定の過程と、採用したあとの実装の形を順に書きます。最後は、この構成で運用を始める前につまずいた話です。 O/RマッパーにDoma 3を選んだ理由 候補はSpring Data JPA、MyBatis、Domaの3つです。選定時に3候補それぞれの利点と欠点を洗い出しました。 候補 利点 欠点 Spring Data JPA Spring Bootとの親和性が高い。メソッド名からクエリを導出できる SQLを自分で書かないため、性能の問題が起きたときに原因を特定しにくい。関連データの取得方法を誤ると過取得やN+1を招く MyBatis SQLを自分で書ける。関連データのマッピングもできる XMLで書くと記述が冗長になる。XMLとアノテーションのどちらでも書けるため、書き方が割れると管理が複雑になる Doma SQLをファイルとして管理でき、そのままDBで実行して確認できる。SQLファイルの不足やDoma独自のコメント記法の誤りをビルド時に検出できる JPAのような暗黙的な関連取得や遅延ロードはない 最終的にDoma 3を採用しました。 決め手はSQLの扱いでした。DomaにはSQLテンプレートという仕組みがあります。公式ドキュメントで「two-way SQL」と呼ばれているもので、条件分岐やバインド変数をSQLのコメントとして書きます。SQLファイルをそのままDBのクライアントへ貼り付けて実行でき、書いたSQLとログに出るSQLもほぼ一致します。遅いクエリの原因を追いやすくなります。 また、クエリの自動生成に寄せたくない意図もありました。運用に入ったあと、どのようなSQLが出るかを把握できる状態を保ちたかったためです。Domaはこの点で扱いやすいと判断しました。 DomainConverterという仕組みによって値オブジェクトを永続化層まで持ち込める点も利点です。 Domaでの実装パターン 全体の層構成は次のとおりです。 層 役割 主な中身 ドメイン層 業務の言葉とルール 集約、値オブジェクト、Repositoryのインタフェース アプリケーション層 ユースケースの手続きとトランザクション境界 UseCaseのインタフェースとHandler インフラ層 技術による実現 Domaを使った永続化、DataSource、外部サービスとの連携 プレゼンテーション層 HTTPの受け口 Controller、リクエストとレスポンスのスキーマ 依存は内側へ向かう一方向です。ドメイン層は他のどの層も参照しません。Repositoryはインタフェースだけをドメイン層に置き、実装はインフラ層が持ちます。 ドメイン層には、紹介や報酬といった業務概念ごとにパッケージを分けて置いています。 本記事で扱うのは、このうちインフラ層です。 以降のコードには友だち紹介機能の概念が出てきます。各用語は以下のような意味です。 紹介者：招待した側 被紹介者：招待された側 紹介成立：紹介者と被紹介者が結びついた記録 報酬：付与するインセンティブ インフラ層のクラス構成 インフラ層は次のクラスで構成されます。 クラス 責務 DAO SQLファイルとの対応づけと、DB操作の宣言 Entity テーブル1行に対応するオブジェクト DomainConverter 値オブジェクトとDBの型の相互変換 RepositoryImpl Entityとドメインオブジェクトを組み立て、ドメインの操作として公開 DAO、Entity、DomainConverterはDomaが用意した仕組みです。RepositoryImplはこの基盤で用意したクラスです。 Entityはテーブルの構造をそのまま写した型です。Entityとドメインモデルを直接つなぐと、テーブルの変更がドメインモデルにも及びます。そこで両者の変換はRepositoryImplが行います。 ドメインクラスで値オブジェクトを使用 ドメインクラス は、カラムの値をJavaのオブジェクトとして扱うDomaの仕組みです。作り方は2つあります。 内部ドメインクラス：型そのものに @Domain を付け、対応するDBの型を valueType で指定する 外部ドメインクラス：型には手を入れず、 DomainConverter を実装したコンバータクラスに @ExternalDomain を付ける @Domain を選ぶと、ドメイン層の型にDomaのアノテーションが載ります。この基盤ではそれを避けたいので、外部ドメインクラスを選びました。コンバータクラスはインフラ層に置いています。次は紹介成立のIDを表す ReferralId のものです。 @ExternalDomain public class ReferralIdConverter implements DomainConverter<ReferralId, Long> { @Override @Nullable public Long fromDomainToValue( @Nullable ReferralId domain) { if (domain == null ) { return null ; } return domain.value(); } @Override @Nullable public ReferralId fromValueToDomain( @Nullable Long value) { if (value == null ) { return null ; } return ReferralId.of(value); // ReferralIdのファクトリメソッド } } 型引数がドメイン側とDB側の型の対応です。これを置くと、DAOの引数に値オブジェクトをそのまま使えます。 @Dao @ConfigAutowireable public interface ReferralDao { @Select List<ReferralEntity> selectReferrals( // 中略 @Nullable ReferrerId referrerId, @Nullable ReferredId referredId, SelectOptions options); @Insert (exclude = { "createdAt" , "updatedAt" }) Result<ReferralEntity> insert(ReferralEntity referral); @Select boolean existsById(ReferralId id); } DAOもDomaの仕組みです。 @Dao を付けたインタフェースの実装は、コンパイル時にアノテーションプロセッサが生成します。 @Select のメソッドは同名のSQLファイルと対応づき、 @Insert のようにSQLファイルを持たないものはDomaがSQLを組み立てます。 SelectOptions はページング、 Result は登録や更新の結果を受け取る型です。 外部ドメインクラスの方式では、値オブジェクト1つにつきコンバータクラスが1つ必要です。 Entityをテーブルの表現に徹させる Entityはテーブルまたはクエリの結果セットに対応するDomaの仕組みです。この基盤ではすべてrecordで書いています。以下は紹介成立を保持する referrals テーブルのものです。 @Entity (immutable = true , naming = NamingType.SNAKE_LOWER_CASE) @Table (name = "referrals" ) public record ReferralEntity( @Id @GeneratedValue (strategy = GenerationType.IDENTITY) @Nullable ReferralId id, // 中略 ReferrerId referrerId, ReferredId referredId, @Nullable LocalDateTime createdAt, @Nullable LocalDateTime updatedAt) {} referrerId や referredId などはDB上では BIGINT ですが、コード上は値オブジェクトとして読み書きできます。 id に付けた @GeneratedValue は、IDの採番をDB側の AUTO_INCREMENT に任せる指定です。登録するまでIDが決まらないため、 id だけ @Nullable にしています。 Entityとドメインオブジェクトの変換 ここからはDomaの仕組みではなく、この基盤の独自の話です。 ドメイン層の型は、IDの採番前と採番後で分けています。採番前は NewReferral のように、これから登録する内容だけを持ち、IDがありません。採番後は Referral でIDを持ちます。RepositoryImplは登録のときに採番前の型からEntityを組み立て、取得のときにEntityからドメインオブジェクトを復元します。 変換の中身は値の移し替えだけではありません。例えば報酬の実績を持つ rewards テーブルは、発行・付与・取消の日時を issued_at 、 granted_at 、 revoked_at の3列で持ちます。一方でドメイン側の Reward は RewardStatus という状態を持ちます。この状態に対応する列はなく、3つの日時から計算します。 private RewardStatus getStatus(RewardEntity entity) { if (entity.revokedAt() != null ) { return RewardStatus.REVOKED; } if (entity.grantedAt() != null ) { return RewardStatus.GRANTED; } return RewardStatus.ISSUED; } 状態の導出をドメイン層に置くと、テーブルの列構成がドメインへ漏れます。インフラ層に置くことで、それを防いでいます。 RepositoryImplでDBの例外を業務の例外へ変える referrals テーブルには複数の列の組に一意制約があり、各列に外部キー制約もあります。登録が失敗する理由はこの2つで、意味が違います。 なおDoma自体は UniqueConstraintException を投げます。これがSpringの DuplicateKeyException へ変わるのは、doma-spring-bootが例外を変換しているためです。 次は紹介成立を登録するRepositoryImplのメソッドです。 @Override public Referral create(NewReferral newReferral) throws ReferralAlreadyExistsException, ReferencedEntityNotExistsException { ReferralEntity entity = ... ; // NewReferral から Entity へ変換 try { Result<ReferralEntity> result = referralDao.insert(entity); return ... ; // Entity から Referral へ復元 } catch (DuplicateKeyException e) { throw new ReferralAlreadyExistsException( "Referral already exists: ..." , e); } catch (DataIntegrityViolationException e) { throw new ReferencedEntityNotExistsException( "Referenced entity does not exist: ..." , e); } } アプリケーション層はこの2つを別の例外へ変え、コントローラが409と400に振り分けます。 ただし DataIntegrityViolationException は DuplicateKeyException の親クラスで、整合性制約の違反を広く拾います。この基盤では外部キー制約の違反として扱っているため、他の制約に違反した場合も同じ例外になります。 条件分岐をSQLテンプレートで書く 紹介成立の検索は、紹介者や被紹介者を任意の組み合わせで絞り込めます。この分岐はSQLファイルに書きます。 SELECT id, -- 中略 referrer_id, referred_id, created_at, updated_at FROM referrals WHERE 1 = 1 -- 中略 /*%if referrerId != null */ AND referrer_id = /* referrerId */ 1 /*%end */ /*%if referredId != null */ AND referred_id = /* referredId */ 1 /*%end */ ORDER BY id コメントで書かれた命令をDomaはディレクティブと呼びます。 /*%if*/ と /*%end*/ が条件ディレクティブ、 /* referrerId */ がバインド変数ディレクティブです。 1 はコメントの外に置いたテスト用の値で、実行時にプレースホルダへ置き換わります。 この基盤では INSERT と UPDATE のSQLファイルを書かず、 @Insert と @Update の自動生成に任せています。 exclude で createdAt と updatedAt を外しており、登録時はMySQL側のデフォルト値で埋まります。SQLを手で書くのは、条件分岐や結合が必要な参照だけに絞る形です。 読み書きの振り分け ここからは実装後に判明した不備の話です。読み書きの振り分けを実装したつもりで、DataSourceのレベルでは切り替わっていませんでした。ユニットテストと結合テストは通っており、気付いたのは障害試験のときでした。 読み書きを分けた理由 データベースにはAurora MySQLを使っています。Auroraのクラスターは、書き込みを受け付けるWriterインスタンスと、読み取り専用のReaderインスタンスで構成されます。アプリケーションからはそれぞれのエンドポイントへ別に接続するため、接続プールも2つ用意しています。プールにはSpring Bootがデフォルトで使うHikariCPをそのまま採用しています。 参照をReaderへ寄せようとしたのは、以下の2つの傾向からでした。 1つはアクセスの傾向です。ユーザーの入り口が施策のLPであることが多く、公開期間の短いLPにはプッシュ通知やSNS配信による一時的なアクセス増が見込まれました。もう1つはデータの形です。キャンペーンやポイントのマスタを持つテーブルは読み取りの比率が高く、状態を追跡するテーブルでも読み取りのほうが多くなります。 この見込みに合わせて接続プールの上限を決め、Readerのほうを厚く取っています。 振り分けの入口はアプリケーション層に置いた TransactionService です。参照系のハンドラは readOnly 、更新系は required を呼びます。 @Override public SearchRewardsResult execute(SearchRewardsQuery query) throws SearchRewardsException { return transactionService.readOnly(() -> handle(query)); } 呼び出し側に経路の指定は出てきません。 readOnly と required のどちらを呼ぶかだけで、向かう先が決まる形を目指しました。 1つ目の構成：判定の順序でつまずく 最初は AbstractRoutingDataSource を1つ使いました。接続先を動的に選ぶためのSpringのクラスです。これを継承して determineCurrentLookupKey メソッドを実装すると、Springはコネクションを要求されるたびにそれを呼び、戻り値をキーとして接続先を引き当てます。キーとReaderやWriterの対応は、あらかじめ登録する仕組みです。 そこで、実行中のトランザクションが読み取り専用かどうかでキーを決めることにしました。この判定はSpringの TransactionSynchronizationManager.isCurrentTransactionReadOnly() で取得できます。 final var routingDataSource = new AbstractRoutingDataSource() { @Override protected Object determineCurrentLookupKey() { boolean isReadOnly = TransactionSynchronizationManager.isCurrentTransactionReadOnly(); return isReadOnly ? RouteFor.READER : RouteFor.WRITER; } }; しかし参照もWriterへ流れました。Springがトランザクションを開始する処理が2段になっているためです。前半でコネクションを取得し、後半で読み取り専用フラグをスレッドへ登録します 1 。接続先が決まるのは前半、フラグが立つのは後半です。キーを判定する時点ではフラグがまだ立っておらず、常にWriterが選ばれていました。 そこで判定のタイミングに依存しない構成としました。 2つ目の構成：TransactionManagerを2つに分ける 次に AbstractRoutingDataSource をやめ、ReaderとWriterそれぞれに TransactionManager を用意しました。 SpringTransactionService のコンストラクタで @Qualifier を使い、経路ごとのテンプレートを組み立てる形です。 public SpringTransactionService( @Qualifier ( "readerTransactionManager" ) PlatformTransactionManager readerTransactionManager, @Qualifier ( "writerTransactionManager" ) PlatformTransactionManager writerTransactionManager) { this .readOnlyTemplate = createTemplate(readerTransactionManager, template -> template.setReadOnly( true )); this .requiredTemplate = createTemplate(writerTransactionManager, _ -> {}); // 以下略 } 判定のタイミングの問題は消えました。ユニットテストと結合テストも通り、この構成でリリースの準備まで進みました。 この構成の不備は、障害試験中に判明しました。例外は出ておらず、ログにも警告はありません。SQLは正しく実行され、レスポンスも正常です。 不備の中身は2つありました。 1つは、行き先を切り替える仕組みが TransactionManager の選択しかなかったことです。 TransactionManager は生成時に1つのDataSourceと結びつきます。この構成ではReader用とWriter用を2つ用意し、 readOnly ならReader用、更新系ならWriter用を使う形にしていました。行き先が決まるのはこの使い分けだけで、DataSource自身は切り替える仕組みを持ちません。そのため TransactionManager を通らずにコネクションを取りに行く経路があると、そこは振り分けの対象外です。 もう1つは、参照系のトランザクションでコネクションを余分に取得していたことです。 調査：コネクションの取得回数を数える 使ったのはSpringのログです。 DataSourceUtils は、トランザクションに紐付いたコネクションを見つけられなかったときに Fetching JDBC Connection from DataSource を出力します。この行がトランザクション内で何回出るかを数えれば、余分な取得が起きているか分かります。 MySQLとRedisを起動し、アプリケーションのログをファイルへ出力しながら、全エンドポイントを順に呼び出します。呼び出しは調査用に書いた使い捨てのスクリプトで行いました。 docker compose up -d mysql redis ./gradlew bootRun > /tmp/bootrun_test_output.log 2 >&1 & 参照系のログは以下のようになっていました。 DataSourceTransactionManager - Creating new transaction with name [null]: PROPAGATION_REQUIRED,ISOLATION_DEFAULT,readOnly DataSourceTransactionManager - Acquired Connection [HikariProxyConnection@425938879 ...] for JDBC transaction DataSourceTransactionManager - Switching JDBC Connection [...] to manual commit DataSourceUtils - Fetching JDBC Connection from DataSource DataSourceUtils - Fetching JDBC Connection from DataSource DataSourceTransactionManager - Initiating transaction commit DataSourceTransactionManager - Releasing JDBC Connection [HikariProxyConnection@425938879 ...] after transaction Fetching JDBC Connection from DataSource が2行並びます。更新系では1行でした。22のエンドポイントを実行したところ、参照系の10件が2行、更新系の11件が1行でした。残る1件はマスタの検索で、Redisのキャッシュに当たってDBへ行かないため0行です。 原因はDomaが受け取るDataSourceでした。このときReaderとWriterのHikariDataSourceを、それぞれ個別に TransactionAwareDataSourceProxy で包んでいました。トランザクションに紐付いたコネクションを返すプロキシで、これが2つある状態です。そしてDomaへ渡っていたのは、 @Primary が付いたWriter側のプロキシだけでした。 TransactionManager がコネクションを紐付ける相手と、Domaの問い合わせ先が食い違っていました。 3つ目の構成：DataSourceの参照先を揃える TransactionAwareDataSourceProxy を1つに統合し、その下に AbstractRoutingDataSource を置きました。 TransactionManager とDomaがどちらも同じ routingDataSource を起点にするため、コネクションを紐付ける側と参照する側が一致します。 @Bean public DataSource routingDataSource() { ReadWriteRoutingDataSource routingDS = new ReadWriteRoutingDataSource(); routingDS.setTargetDataSources( Map.of( ReadWriteRoutingDataSource.Route.READER, readerHikariDataSource(), ReadWriteRoutingDataSource.Route.WRITER, writerHikariDataSource())); routingDS.setDefaultTargetDataSource(writerHikariDataSource()); return routingDS; } @Bean @Primary public DataSource dataSource() { return new TransactionAwareDataSourceProxy(routingDataSource()); } @Bean public PlatformTransactionManager transactionManager() { return new DataSourceTransactionManager(routingDataSource()); } TransactionManager も1つに戻しました。経路はThreadLocalで持ちます。 public class ReadWriteRoutingDataSource extends AbstractRoutingDataSource { public enum Route { READER, WRITER } private static final ThreadLocal<Route> currentRoute = new ThreadLocal<>(); public static void setRoute(Route route) { currentRoute.set(route); } public static void clear() { currentRoute.remove(); } @Override protected Object determineCurrentLookupKey() { Route route = currentRoute.get(); return (route != null ) ? route : Route.WRITER; } } キーを設定するのは SpringTransactionService です。トランザクションを開始する前にThreadLocalへ書き込み、終わったら必ず消します。 private <T> T executeWithRoute( ReadWriteRoutingDataSource.Route route, TransactionTemplate template, Supplier<T> operation) { ReadWriteRoutingDataSource.setRoute(route); try { T result = template.execute(_ -> operation.get()); if (result == null ) { throw new IllegalStateException( "Transaction operation returned null" ); } return result; } finally { ReadWriteRoutingDataSource.clear(); } } template.execute より前にキーが決まるため、順序の問題も起きません。 修正後に同じスクリプトを実行すると、参照系の Fetching JDBC Connection from DataSource は2行から1行になりました。 ThreadLocalを使う上での注意 経路をThreadLocalに置いているため、別のスレッドへは伝わりません。現在この基盤は非同期処理を使っていないため問題になっていませんが、導入するときは経路の受け渡しを考える必要があります。 ThreadLocalが空のときはWriterへ倒す実装にしています。経路の設定を忘れた場合でも、読み取り専用の接続で更新を試みる事故は避けられます。 executeWithRoute は処理の終わりにThreadLocalを空にします。そのため readOnly から required を呼ぶような入れ子にすると、内側を抜けた時点で外側の経路まで消えます。この基盤ではこのような呼び出しをしていないため、問題にはなっていません。許す場合は、内側へ入る前の経路を退避し、抜けるときに書き戻す必要があります。 別解：LazyConnectionDataSourceProxy 判定の順序については、ルーティングDataSourceを LazyConnectionDataSourceProxy で包む方法もあります。物理的なコネクションの取得を最初のSQL実行まで遅らせるクラスで、キーの判定が読み取り専用フラグの設定より後になります 2 。 TransactionAwareDataSourceProxy のJavadocにも、中間プロキシとしてこのクラスへ委譲できると書かれています 3 。 この場合、読み取り専用フラグをそのままキーに使えます。ThreadLocalを自分で管理する必要がなくなるため、前節で挙げた注意も不要になります。現在の構成で問題なく動いているため移行はしていませんが、同じものをこれから組むなら検討の余地があると考えます。 Spring Framework 6.1.2以降は setReadOnlyDataSource も使えます。読み取り専用のDataSourceを直接指定でき、 AbstractRoutingDataSource を書かずに済みます。 いずれの構成でも、 TransactionAwareDataSourceProxy はプロキシの最も外側に置きます。そのうえで TransactionManager とDomaが同じDataSourceを見るようにします。 まとめ 販促基盤のO/Rマッパーには、発行されるSQLを自分たちで追えることを重視してDoma 3を選びました。外部ドメインクラスを使ったので、ドメイン層をDomaから切り離したまま値オブジェクトを永続化層で用いることができました。RepositoryImplではDBの制約違反をドメインの例外へ変えており、一意制約の違反と外部キー制約の違反でAPIのエラーを振り分けられます。 読み書きの振り分けでつまずいた原因は、 TransactionManager とDomaが別のDataSourceを見ていたことでした。この状態では、トランザクションに紐付いたコネクションを再利用できずに取り直します。両者が同じDataSourceを起点にする形へ直して解決しました。動作確認には Fetching JDBC Connection from DataSource のログ行数を数えました。これがいちばん手軽でした。 JavaでDDDを実践する際のO/Rマッパー選定の参考になれば幸いです。 ZOZOでは、一緒にサービスを作り上げてくれる方を募集中です。ご興味のある方は、以下のリンクからぜひご応募ください。 corp.zozo.com 前半は DataSourceTransactionManager の doBegin メソッドです。後半は AbstractPlatformTransactionManager の prepareSynchronization メソッドです。この中で setCurrentTransactionReadOnly メソッドが呼ばれます。 ↩ Document LazyConnectionDataSourceProxy setup for routing datasource to act on transaction definition read-only flag ↩ TransactionAwareDataSourceProxy (Spring Framework API) 。「should be the outermost DataSource of a chain of DataSource proxies/adapters」と記載があります。 ↩