メインコンテンツまでスキップ

CQRS と Event Sourcing — 読み書きを分ける設計と、履歴を正本にする設計

第 12 章でドメインイベントを扱ったとき、イベントソーシングとの違いに触れました。ドメインイベントは通知の手段で、状態はエンティティが持ちます。イベントソーシングではイベントの列が正本で、状態はそこから組み立てます。

本章は後者を扱います。定義と、得られるもの、3 つの代償、採否の分かれ目は第 12 章が正本なので繰り返しません。ここで見るのは、実際に組むとテーブルとコードがどうなるかです。あわせて、イベントソーシングとほぼセットで語られる CQRS を、第 15 章の CQS の先として扱います。

状態を保存する形と、履歴を保存する形​

注文が確定するまでに何が起きるかを考えます。注文が作られ、明細が足され、確定する。

状態を保存する形では、orders テーブルの 1 行が上書きされていきます。

[orders テーブル]

1. 注文を作る
   id=1  status=draft      total_amount=0

2. 明細を足す
   id=1  status=draft      total_amount=3000   ← 同じ行を上書き

3. 確定する
   id=1  status=confirmed  total_amount=3000   ← 同じ行を上書き

履歴を保存する形では、行は上書きされません。起きたことが 1 行ずつ足されていきます。

[order_events テーブル]

stream_version=1  OrderPlaced      payload={"customer_name":"山田太郎","shipping_prefecture":"東京都", ...}
stream_version=2  OrderLineAdded   payload={"order_line_id":1,"product_id":7,"quantity":2,"unit_price":1500,"line_count":1}
stream_version=3  OrderConfirmed   payload={"order_line_ids":[1],"total_amount":3000}

現在の状態は、この 3 行を順に適用して組み立てます。orders に相当するテーブルは書き込み側に存在せず、正本はイベントの列だけです。

組み立ては、イベントの種類ごとに「この 1 行で状態がどう変わるか」を決めておいて、順に当てていく処理です。

[再生]

空の Order                     status=draft   total_amount=0
  ← OrderPlaced を適用          status=draft   total_amount=0
  ← OrderLineAdded を適用       status=draft   total_amount=3000   (1500 × 2 を足す)
  ← OrderConfirmed を適用       status=confirmed  total_amount=3000

3 行目まで当てた結果が、状態を保存する形の orders の 1 行と同じになります。

違いが効くのは、過去を聞かれたときです。状態を保存する形で「いつ確定したか」「確定前に明細が何回変わったか」に答えるには、その情報を別に記録しておく必要があります。記録し忘れた過去は、後から取り出せません。履歴を保存する形では、答えがデータ構造そのものにあります。

CQRS — 読みと書きでモデルを分ける​

第 15 章は CQS を扱いました。1 つのメソッドが状態を変えるか値を返すかのどちらかにする、という原則です。分離の単位はメソッドでした。

CQRS が分けるのはモデルです。書き込みには集約を使い、読み取りには別の形を使います。

なぜモデルを分けるのか​

集約は整合性を守るための形で、読み取りに都合がよいとは限りません。

「注文一覧に顧客名と商品名を出す」という要求を集約で満たすと、注文を引き、そこに載る識別子から顧客と商品をそれぞれ引いて組み立てることになります。第 10 章で見たとおり、集約は外の集約を識別子でしか持たないので、名前を出すには別の集約を引くしかありません。

読み取り専用のテーブルを 1 つ持ち、そこに必要な列を並べておけば、1 回のクエリで済みます。

// 読みモデル。正規化せず、画面が必要とする形のまま持つ
Schema::create('order_summaries', function (Blueprint $table) {
    $table->unsignedBigInteger('order_id')->primary();
    $table->string('status');
    $table->unsignedInteger('total_amount');
    // 書き込み側では別の集約にある値を、読み取り用に写している
    $table->string('customer_name');
    $table->unsignedInteger('line_count');
    // 注文した時刻。一覧の既定の並び順に使うので、後から動かさない
    $table->timestamp('placed_at');
    $table->timestamp('confirmed_at')->nullable();

    $table->index(['status', 'placed_at']);
});

このテーブルは正規化されていません。顧客名を持つので、顧客が改名すれば古い値が残ります。それでよいかは要求によります。一覧に出すのが「注文した時点の顧客名」でよいなら、むしろ写しておくほうが正確です。

2 つのモデルを揃える​

分けた瞬間に、2 つのモデルを揃える仕事が生まれます。書き込み側が変われば、読み取り側も更新しなければなりません。

その経路がドメインイベントです (第 12 章)。

ここに遅れが入ります。書き込みが終わった直後に読み取ると、まだ古い値が見えることがあります。これが結果整合性で、第 10 章が集約間の連携で説明したものと同じ性質です。

遅れを許せない画面では、書き込み側から直接読むか、同じトランザクションで読みモデルも更新します。後者は分離の利点を半分捨てることになるので、どの画面で遅れが許されるかを先に決めます。

読みモデルを引く側は、第 15 章の Query サービスがそのまま使えます。同章はインターフェースをアプリケーション層に、実装をインフラ層に置くと定めているので、参照先を orders から order_summaries に変えるだけで形は変わりません。

採らない判断のほうが多い​

読み取りが重くないなら、CQRS は要りません。集約から読んで API Resource で整形すれば足ります (第 16 章)。

分けるのが効くのは、読み取りと書き込みで要求が食い違うときです。

分けない分ける
読み取りが集約 1 件で足りる一覧に複数の集約の値を並べる
読み取りの頻度が書き込みと同程度読み取りが桁違いに多い
遅れを許せない画面が中心数秒の遅れを許せる

イベントソーシングを採るなら、CQRS はほぼ必ず要ります。イベントの列に対して「確定済みの注文を一覧する」を直接は書けないためです。event_type = 'OrderConfirmed' で絞れるのは「確定したことがある注文」で、そのあとキャンセルされたかどうかは、続くイベントまで畳み込まないと分かりません。現在の状態で絞るには、畳み込んだ結果を持つテーブルが要ります。

逆向きは成り立ちません。CQRS を採るからイベントソーシングが要る、ということはありません。

イベントストアのテーブル設計​

イベントを保存するテーブルを設計します。

Schema::create('order_events', function (Blueprint $table) {
    // 全ストリームを貫く通し番号。プロジェクションが読む順序をこれで決める
    $table->bigIncrements('id');
    // 集約 1 件分のイベントの並び = ストリーム。OrderId と同じ型にする
    $table->unsignedBigInteger('stream_id');
    // そのストリームの中での順番。1 から始まる
    $table->unsignedBigInteger('stream_version');
    $table->string('event_type');
    // 何が起きたか。イベントごとに形が違うので列に展開しない
    $table->json('payload');
    // 誰が・どの操作で起こしたか。業務の意味を持たない情報はこちらへ
    $table->json('metadata');
    $table->timestamp('occurred_at');

    // 並行制御の本体。同じストリームの同じ版は 2 つ存在できない
    $table->unique(['stream_id', 'stream_version']);
    $table->index('event_type');
});

1 テーブルにするか、集約の種類ごとに分けるか​

本章は注文だけを題材にしているので order_events という名前にしましたが、実運用では全種類の集約のイベントを 1 つのテーブルに入れるほうが多くなります。

分けたくなる理由は分かります。テーブルが小さくなり、注文のイベントだけを消すのも楽です。それでも 1 つにするのは、複数の集約にまたがるプロジェクションが順序を決められなくなるためです。在庫と注文の両方を見る集計を作ると、2 つのテーブルの通し番号は互いに独立しているので、どちらのイベントが先に起きたかをその番号からは判定できません。1 本の並びに載っていれば、コミット済みのイベントどうしの順序は id で決まります。

1 つにまとめるなら、列が 1 本増えます。本ガイドの識別子は集約ごとに 1 から始まる整数なので (第 8 章)、注文の 1 番と在庫の 1 番を区別する列がないと、両者が同じストリームに混ざります。

// どの種類の集約のストリームか。'order' / 'inventory' などが入る
$table->string('stream_type');

// 一意制約もこの 3 列になる。stream_type が無いと、無関係な集約どうしで
// 「同じ版が既にある」という衝突が起きる
$table->unique(['stream_type', 'stream_id', 'stream_version']);

この構成にすると、読む側も stream_type を伴うことになります。後述するインデックスの左端一致が効くのは WHERE stream_type = ? AND stream_id = ? ORDER BY stream_version の形で、stream_id だけで絞る書き方は使えません。他の集約の行まで返ってしまうためです。

識別子を UUID にすれば値が全域で一意になるので、この列は要りません。後述する spatie のテーブルが aggregate_uuid だけで足りているのはそのためです。

集約の種類ごとに分けるのは、片方が桁違いに大きくなって物理的に分けたくなったときです。そのときはパーティショニングを先に検討します。

複合一意制約が並行制御そのもの​

2 つのリクエストが同じ注文を同時に更新しようとすると、どちらも同じ stream_version で INSERT しようとします。先に入ったほうが成功し、後から来たほうが一意制約違反になります。

第 19 章の楽観的ロックと考え方は同じですが、競合を検出する場所が違います。

競合の検出
第 19 章の楽観的ロックUPDATE ... WHERE version = ? の影響行数が 0 なら競合
イベントストアINSERT が一意制約に弾かれたら競合

どちらもロックを取らずに競合を検出し、負けたほうに読み直させます。イベントストアの側は、テーブルが追記専用なので UPDATE が出てこないぶん形が単純になります。

通し番号とストリーム内の版は別の役割を持つ​

id は全ストリームを貫く通し番号で、stream_version はストリームの中での順番です。役割が違います。

  • stream_version — 集約 1 件の整合性を守る。並行制御が見るのはこちら
  • id — すべてのイベントを 1 本の並びに載せる。プロジェクションが読む順序を決める

プロジェクションは複数の集約のイベントをまとめて処理するので、ストリームごとの版では順序を表せません。

ペイロードとメタデータを分ける​

payload は業務上何が起きたかを持ちます。metadata が持つのは業務の意味を持たない情報です。誰が実行したか、どのリクエストから来たか、どの相関 ID に属するかといったものが入ります。

分けておくと、イベントの形を変えるときに payload だけを見ればよくなります。運用のために足した情報が業務の履歴に混ざらないという効果もあります。

インデックスは読み方から決める​

stream_id と stream_version の複合一意制約が、そのまま「1 つの集約のイベントを順に読む」クエリのインデックスになります。WHERE stream_id = ? ORDER BY stream_version は左端から使えます (第 18 章)。

event_type の単独インデックスは、「特定の種類のイベントだけを流し直す」用途に効きます。使わないなら張りません。

イベントをストアに載せる契約​

第 12 章のイベントは、リスナーに渡す前提で作られています。OrderConfirmed は orderId / orderLineIds / totalAmount / confirmedAt を public なプロパティとして持つだけで、保存のための形は持ちません。時刻のプロパティ名もイベントごとに違います (confirmedAt / cancelledAt / shippedAt)。

イベントストアに載せるには、保存側が要求する契約を足します。

なお本章は、第 12 章が実装を持つ 3 つ (OrderConfirmed / OrderCancelled / OrderShipped) に加えて、同章が名前だけ挙げた OrderPlaced と、明細を足す OrderLineAdded を使います。状態を保存する形では Order::create() と明細の追加メソッドが担っていた部分で、イベントソーシングではこれらもイベントになります。

OrderPlaced には配送先と、注文した時点の顧客名を載せます。配送先は Order の再生に要り (第 9 章の Order が readonly で要求するため)、顧客名は読みモデルが使います。生成イベントは、その集約を空から組み立てるのに要る値をすべて持つ必要があります。

OrderLineAdded も同じ理屈で、明細の識別子・商品・数量・単価に加えて追加後の明細数を載せます。識別子を載せるのは、再生のたびに振り直すと OrderConfirmed が持つ order_line_ids と対応しなくなるためです。明細数を載せる理由は「プロジェクション」で扱います。

// app/Domain/Shared/StorableEvent.php
// 本章で新しく導入するインターフェース。第 12 章のイベントは保存を前提にしていないので、
// ストアが必要とする 2 つだけをここで要求する
interface StorableEvent
{
    /** @return array<string, mixed> */
    public function payload(): array;

    public function occurredAt(): DateTimeImmutable;
}

第 12 章のイベントにこれを実装します。既存のプロパティは変えず、ストア向けの読み出し口を足すだけです。

// app/Domain/Order/Event/OrderConfirmed.php
final class OrderConfirmed implements StorableEvent
{
    public function __construct(
        public readonly OrderId $orderId,
        public readonly array $orderLineIds,
        public readonly Money $totalAmount,
        public readonly DateTimeImmutable $confirmedAt,
    ) {}

    public function payload(): array
    {
        // 再生でイベントを組み立て直すのに要る値は、すべてここに入れる。
        // 入れ忘れると、そのイベントは二度と復元できない
        return [
            'order_line_ids' => array_map(fn (OrderLineId $id): int => $id->value(), $this->orderLineIds),
            'total_amount' => $this->totalAmount->amount(),
        ];
    }

    // イベントごとに違う時刻のプロパティ名を、ストアから見た 1 つの名前に揃える
    public function occurredAt(): DateTimeImmutable
    {
        return $this->confirmedAt;
    }
}

orderId を payload に入れていないのは、stream_id 列が持つためです。同じ値を 2 か所に置くと、片方だけ直す事故が起きます。

集約を再構成する​

ストリームのイベントを順に適用して集約を作ります。

第 9 章の Order は create() と reconstruct() を持ちますが、どちらもイベントからは組み立てません。イベントソーシングを採るなら、ドメイン層にも足すものがあります。

// app/Domain/Order/Order.php — 本章で足すぶん
public static function replay(OrderId $id, iterable $events): self
{
    $order = null;

    foreach ($events as $event) {
        // 生成イベントだけがコンストラクタを呼ぶ。第 9 章の Order は配送先と作成日時を
        // readonly で要求するので、この 2 つは OrderPlaced が運ぶ必要がある
        if ($event instanceof OrderPlaced) {
            // 別の版に 2 つ目の生成イベントが載ったストリームは壊れている。同じ版なら
            // 一意制約が止めるが、別の版は素通りする。弾かないと、積み上げた状態を
            // 黙って捨てて作り直す
            if ($order !== null) {
                throw new InvalidArgumentException('ストリームに OrderPlaced が 2 つあります');
            }

            $order = new self($id, OrderStatus::DRAFT, $event->shippingAddress, [], $event->occurredAt());
            continue;
        }

        // 生成イベントより前に何か来ていたら、ストリームが壊れている
        if ($order === null) {
            throw new InvalidArgumentException('ストリームの先頭に OrderPlaced がありません');
        }

        $order->apply($event);
    }

    return $order ?? throw new InvalidArgumentException('ストリームが空です');
}

private function apply(StorableEvent $event): void
{
    // 各イベントが「この 1 行で状態がどう変わるか」を決める
    match (true) {
        // addLineFromEvent は payload から OrderLine を組み立てて追加する private メソッド。
        // 第 9 章の addItem() から状態の検査を外した形になる。再生は既に起きたことの
        // 再現なので、そこで不変条件を問い直すと過去のイベントで例外が飛ぶ
        $event instanceof OrderLineAdded => $this->addLineFromEvent($event),
        $event instanceof OrderConfirmed => $this->status = OrderStatus::CONFIRMED,
        $event instanceof OrderShipped => $this->status = OrderStatus::SHIPPED,
        $event instanceof OrderCancelled => $this->status = OrderStatus::CANCELLED,
        // 知らないイベントを黙って読み飛ばすと、再生した集約が古い状態のまま返る
        default => throw new LogicException('再生に未対応のイベント: ' . $event::class),
    };
}

自前で組む場合でも、ドメイン層はイベントソーシングを知ることになります。replay() と apply() は保存方式に紐づいたメソッドで、状態を保存する形では要りません。第 9 章の Order が持つ version プロパティも使わなくなります。版を持つのはストリームの側 (stream_version) で、集約は自分の版を知らなくてよくなるためです。

ストアから読む​

読み込み側は、集約とその時点の版をまとめて返します。版は次の書き込みで一意制約を効かせるために要ります。

// app/Domain/Order/RetrievedOrder.php
final class RetrievedOrder
{
    public function __construct(
        public readonly Order $order,
        public readonly int $version,
    ) {}
}
// app/Infrastructure/EventStore/EloquentOrderEventStore.php
// インターフェース OrderEventStoreInterface は app/Domain/Order/ に置く。
// 集約を出し入れする操作はドメインが必要とするものなので、第 17 章のリポジトリと同じ配置になる
public function retrieve(OrderId $id): ?RetrievedOrder
{
    $rows = DB::table('order_events')
        ->where('stream_id', $id->value())
        ->orderBy('stream_version')
        ->get();

    // 1 件も無いストリームは「まだ存在しない注文」。第 17 章の findById と同じく null を返す
    if ($rows->isEmpty()) {
        return null;
    }

    return new RetrievedOrder(
        Order::replay($id, $rows->map(fn ($row) => $this->deserializer->fromRow($row))),
        $rows->last()->stream_version,
    );
}

EventDeserializer は event_type の文字列からクラスを解決し、payload を渡してイベントを組み立てます。ストアと同じ app/Infrastructure/EventStore/ に置きます。ストアのメソッドにせず独立したクラスにするのは、後で出てくるプロジェクターが同じ復元を必要とするためです。クラス名を列に持つ以上、クラスをリネームすると過去のイベントが読めなくなります。イベントクラスの名前は、テーブルに書かれた値だと考えて扱います。

書き込む​

public function append(OrderId $streamId, array $events, int $expectedVersion): void
{
    try {
        DB::table('order_events')->insert(
            array_map(
                fn (int $offset, StorableEvent $event): array => [
                    'stream_id' => $streamId->value(),
                    'stream_version' => $expectedVersion + $offset + 1,
                    'event_type' => $event::class,
                    'payload' => json_encode($event->payload()),
                    'metadata' => json_encode(['user_id' => Auth::id()]),
                    'occurred_at' => $event->occurredAt(),
                ],
                array_keys($events),
                $events,
            ),
        );
    } catch (UniqueConstraintViolationException) {
        // 読み込んでから append するまでのあいだに、別の処理が同じ版を書いた
        throw new OptimisticLockException('他のユーザーによって更新されました。再度読み込んでください。');
    }
}

$expectedVersion は retrieve() が返した version です。新規の注文は retrieve() が null を返すので 0 を渡し、最初のイベントが stream_version = 1 になります。

UniqueConstraintViolationException は Laravel 10 で追加された QueryException の子クラスです。

イベントの永続化と読みモデルの更新を 1 つのトランザクションに入れるかは、別の判断です。分けると片方だけ成功する状態が生まれます。本ガイドは第 19 章で DB::afterCommit を使う形を採っており、イベントストアでも同じ道具が使えます。

スナップショットで再生を短絡する​

イベントが増えるほど、再生に時間がかかります。1 件の注文に数千のイベントが積もるなら、読むたびに数千行を処理することになります。

スナップショットは、再生を途中から始めるための短絡です。

Schema::create('order_snapshots', function (Blueprint $table) {
    $table->id();
    $table->unsignedBigInteger('stream_id');
    // この版までを適用した状態、という意味
    $table->unsignedBigInteger('stream_version');
    $table->json('state');
    $table->timestamps();

    // 同じ版のスナップショットが 2 つ入らないようにする
    $table->unique(['stream_id', 'stream_version']);
});

読み込みは、最新のスナップショットを引いてから、その版より後のイベントだけを適用する形になります。

$snapshot = DB::table('order_snapshots')
    ->where('stream_id', $id->value())
    ->orderByDesc('stream_version')
    ->first();

$rows = DB::table('order_events')
    ->where('stream_id', $id->value())
    ->where('stream_version', '>', $snapshot?->stream_version ?? 0)
    ->orderBy('stream_version')
    ->get();

短絡を入れると、retrieve() も変わります。イベントが 0 件でもスナップショットがあれば注文は存在するので、「1 件も無ければ null」という判定はスナップショットを引いた後に移ります。state の JSON から Order を組み立てる経路も要るので、ドメイン層に足すものは replay() と apply() だけでは済みません。

第 10 章にもスナップショットという語が出てきますが、指しているものが違います。第 10 章のは、注文時点の配送先住所のようにある時点の値を写して持ち続ける設計です。本章のは再生を短絡するための保存で、捨てて作り直しても業務的な意味は変わりません。

スナップショットは最適化なので、遅くなってから入れます。最初から作ると、状態の形を変えるたびに保存済みのスナップショットも作り直すことになります。

プロジェクション​

読みモデルをイベントから作り続ける処理をプロジェクションと呼びます。

読み取りは第 15 章が書いたとおりアプリケーションの関心事なので、Query サービスと同じく、インターフェースをアプリケーション層に、実装をインフラ層に置きます。

本章にはイベントストアとプロジェクションという 2 つのインターフェースが出てきて、置き場が違います。分かれ目は引数の型ではありません。whenOrderPlaced(OrderPlaced $event) はドメインのイベントを受け取りますが、アプリケーション層に置きます。その契約が誰の関心事かで決まります。集約の出し入れはドメインが必要とする操作なのでドメイン層 (第 17 章)、読みモデルの更新は読み取り側の都合なのでアプリケーション層です。どちらも実装だけがインフラ層に降りる点は第 14 章の依存性逆転のとおりです。

// インターフェース OrderSummaryProjectionInterface は app/Application/Projection/ に置く
// app/Infrastructure/Projection/EloquentOrderSummaryProjection.php
final class EloquentOrderSummaryProjection implements OrderSummaryProjectionInterface
{
    // 行を作るのは生成イベントの担当。以降のイベントは更新しかしない
    public function whenOrderPlaced(OrderPlaced $event): void
    {
        DB::table('order_summaries')->upsert(
            [
                'order_id' => $event->orderId->value(),
                'status' => 'draft',
                'total_amount' => 0,
                // 注文した時点の顧客名。OrderPlaced が載せている値で、後から顧客が
                // 改名しても動かない (この写しの是非は「なぜモデルを分けるのか」で扱った)
                'customer_name' => $event->customerName,
                'line_count' => 0,
                'placed_at' => $event->occurredAt(),
            ],
            'order_id',
            // 更新する列を OrderPlaced が持つぶんだけに絞る。既定の全列更新にすると、
            // このイベントが再配信されたときに status・total_amount・line_count が初期値へ戻る
            ['customer_name', 'placed_at'],
        );
    }

    public function whenOrderLineAdded(OrderLineAdded $event): void
    {
        DB::table('order_summaries')
            ->where('order_id', $event->orderId->value())
            // increment() にしない。同じイベントが 2 回届くと値が 1 多くなり、
            // しかも例外もログも出ない。イベントに追加後の数を持たせて確定値を書く
            ->update(['line_count' => $event->lineCount]);
    }

    public function whenOrderConfirmed(OrderConfirmed $event): void
    {
        DB::table('order_summaries')
            ->where('order_id', $event->orderId->value())
            ->update([
                'status' => 'confirmed',
                'total_amount' => $event->totalAmount->amount(),
                // placed_at は動かさない。一覧の並び順が確定のたびに入れ替わる
                'confirmed_at' => $event->occurredAt(),
            ]);
    }

    // 状態が変わるだけのイベント。status を持つ読みモデルでこの 2 つを取りこぼすと、
    // キャンセル済みの注文が confirmed のまま一覧に残る
    public function whenOrderCancelled(OrderCancelled $event): void
    {
        $this->updateStatus($event->orderId, 'cancelled');
    }

    public function whenOrderShipped(OrderShipped $event): void
    {
        $this->updateStatus($event->orderId, 'shipped');
    }

    private function updateStatus(OrderId $orderId, string $status): void
    {
        DB::table('order_summaries')
            ->where('order_id', $orderId->value())
            ->update(['status' => $status]);
    }
}

読みモデルはイベントから何度でも作り直せます。表示する項目が増えたら、テーブルと読んだ位置の記録を作り直し、イベントを最初から流し直せば埋まります。この作り直しやすさが、読みモデルを正規化しない判断を支えています。

同じイベントが 2 回届いても結果が変わらない形にしておきます。冪等性の作り方は第 12 章がリスナーの側で扱っており、一般的な考え方はウェブエンジニア基礎知識ガイドにあります。

読んだ位置を記録する — 通し番号には穴があく​

プロジェクションの動かし方は 2 通りあります。書き込み側からイベントを渡す (push) 形と、テーブルを定期的に読みに行く (polling) 形です。読みモデルを後から作り直せる状態に保つには polling が要るので、ここからは polling を前提にします。

when* を呼ぶのは、まだ読んでいないイベントを順に取り出す処理です。どの id まで読んだかを記録して、次はその続きから読みます。

// app/Infrastructure/Projection/ProjectOrderSummaries.php (Artisan コマンド)
// order_events を直接読むのでインフラ層に置き、アプリケーション層の
// OrderSummaryProjectionInterface を呼ぶ
public function handle(EventDeserializer $deserializer, OrderSummaryProjectionInterface $projection): void
{
    // projection_positions は name と last_event_id の 2 列を持つテーブル。
    // 読みモデルごとに 1 行
    $position = DB::table('projection_positions')
        ->where('name', 'order_summaries')
        ->value('last_event_id') ?? 0;

    $rows = DB::table('order_events')
        ->where('id', '>', $position)
        ->orderBy('id')
        ->limit(1000)
        ->get();

    foreach ($rows as $row) {
        $event = $deserializer->fromRow($row);

        match (true) {
            $event instanceof OrderPlaced => $projection->whenOrderPlaced($event),
            $event instanceof OrderLineAdded => $projection->whenOrderLineAdded($event),
            $event instanceof OrderConfirmed => $projection->whenOrderConfirmed($event),
            $event instanceof OrderCancelled => $projection->whenOrderCancelled($event),
            $event instanceof OrderShipped => $projection->whenOrderShipped($event),
            // この読みモデルが扱わないイベントは読み飛ばす。読み飛ばしてよいのは
            // どの列にも影響しないイベントだけで、状態を持つ列があるなら取りこぼせない
            default => null,
        };

        // 行がまだ無い初回も進むようにする。update だけだと位置が一度も動かず、
        // when* が冪等なぶん読みモデルは正しく見えたまま同じ範囲を読み続ける
        DB::table('projection_positions')->updateOrInsert(
            ['name' => 'order_summaries'],
            ['last_event_id' => $row->id],
        );
    }
}

これを schedule で定期的に叩くか、常駐させて回します。素朴に書くとこうなります。そして、この形は壊れます。

AUTO_INCREMENT の採番順と、コミットの順序は一致しません。1

時刻 →

TX1: id=100 を採番 ────────────── commit
TX2:      id=101 を採番 ── commit

プロジェクションが TX2 の commit 直後に読むと、101 は見えるが 100 はまだ見えない。
「101 まで読んだ」を記録すると、あとで commit された 100 は二度と読まれない。

Ecotone のドキュメントはこの結果を「エラーも、ログも、例外も出ない。読みモデルが恒久的に間違ったままになる」と書いています2。開発環境では並行実行が起きないので表に出ず、本番で負荷がかかったときだけ壊れます。

対処は 3 つあります。

方法内容代償
穴が埋まるまで止める未確定の id があるあいだ、それより先へ進まない長いトランザクション 1 本で全体が止まる
穴を記録して進む「どこまで進んだか」に加えて「欠けている位置」を持ち、後から埋める位置の持ち方が複雑になる
通し番号を自前で採るAUTO_INCREMENT に任せず、追記と同じトランザクションで採番用のテーブルを進める追記が直列化する

手を入れる側が違います。前の 2 つは読む側で閉じ、3 つ目は追記の側を変えます。上のコマンドをそのまま動かせるのは 3 つ目を採ったときだけで、前の 2 つを採るなら読む側の処理を作り直すことになります。プロジェクションを自前で書くなら、ここが最初に設計すべき点です。

イベントの形を変えるとき​

イベントは起きた事実の記録なので、後から書き換えられません。仕様変更で持たせたい情報が増えたときは、古い形を読み続ける仕組みが要ります。

道具は 2 つあります。

  • イベントごとの版数を持つ — イベントの種類とは別に、その形が何版目かを記録します。読み込み時に版数で分岐し、古い形を新しい形へ変換します
  • 新しいイベントの種類を足す — OrderConfirmed を変えるのでなく OrderConfirmedV2 を足し、集約が両方を適用できるようにします

どちらも、古い形を理解するコードを消せなくなります。運用が長いほどこの分岐が積もるので、イベントの形は最初から慎重に決めます。

Laravel で採るなら​

自前で組むか、パッケージを使うかの分岐があります。

spatie/laravel-event-sourcing は、集約・プロジェクター・リアクターの組み立てを引き受けるパッケージです。v7 の stored_events テーブルは本章で設計したものとほぼ同じ形です。aggregate_uuid と aggregate_version の複合一意制約を持ち、イベントクラスを event_class に入れ、ペイロードとメタデータを event_properties と meta_data に分けています。スナップショットも snapshots テーブルとして別に用意されています。

違いは識別子の型で、パッケージは UUID を前提にします。本ガイドの OrderId は正の整数なので (第 8 章)、乗せるなら識別子の設計から見直すことになります。

パッケージを使う判断は、集約の書き方をパッケージの基底クラスに合わせる覚悟とセットです。一部の集約だけをイベントソーシングにするなら自前のほうが既存の設計を保て、システム全体を移すならパッケージが引き受ける範囲のほうが大きくなります。

専用のフレームワークが置いている境界​

Java には Axon Framework という、DDD・CQRS・イベントソーシングのために作られたフレームワークがあります。集約、コマンドの処理、イベントの配信とプロセッサ、Saga、クエリの処理までを部品として持ちます。PHP には同じ規模のものがないので直接は使えませんが、この規模のフレームワークがどこに線を引いているかは、自前で組むときの参考になります。

とくにはっきりしているのが、イベントストアとメッセージブローカーの分け方です。Axon には Kafka 拡張がありますが、その公式ドキュメントはこう書いています。

Note that Kafka is a perfectly fine event distribution mechanism, but it is not an event store.

同じページは、この拡張では集約をイベントソーシングできないとしています ("the extension cannot be used to event source aggregates, as this requires an event store implementation")。使うべきものとして挙げているのは、専用のイベントストアか RDBMS ベースの実装です3。

役割が分かれると、処理の流れは次の形になります。

④ と ⑤ が別の段であることが、この節の要点です。④ は自分のデータベースへの追記で、ここが正本です。⑤ は外へ知らせる配信で、途中で失敗しても ④ は残ります。この 2 つは 1 つのトランザクションに入りません。ブローカーへの送信はデータベースの外の操作だからです。代わりに「④ が入ったなら送信予定も必ず残っている」状態を作り、送信は別のプロセスに任せます。その組み方は第 24 章の「イベントソーシングを採っているなら」が扱います。

⑦ の置き場は、読みモデルを誰が作るかで分かれます。同じサービスの中で作るなら、⑤⑥ を通らず ④ のイベントストアを直接読みます。本章の「読んだ位置を記録する」で組んだ ProjectOrderSummaries がその形で、order_events をポーリングします。上の図の ⑤⑥⑦ は、読みモデルを別のサービスに切り出した場合の経路です。そのサービスは自分のイベントストアを持たず、届いたイベントから自分用のテーブルを組み立てます。正本を持つサービスと、写しを持つサービスが分かれます。

本章が order_events をリレーショナルデータベースに置いたのは、この線と同じ判断です。集約を再構成するには、あるストリームのイベントを版の順に全件読め、追記のたびに一意制約で並行を弾ける必要があります。Kafka が消える条件は消費ではなく保持期間なので過去を読み直せますが、それは「保持期間のあいだは残る」ということであって、正本として無期限に持つことの保証ではありません。ブローカーが引き受けるのは届けることで、第 24 章がその範囲を扱います。

採否を実装コストで測り直す​

第 12 章は採否の分かれ目を置きました。本章で組んだものを並べると、それに加えて払うコストが見えます。

必要になるもの状態を保存する形なら
イベントストアのテーブルと追記処理Eloquent の save() で足りる
ドメイン層の replay() と apply()不要
集約の再構成とスナップショット1 行の SELECT で足りる
読みモデルとプロジェクション不要 (集約から読める)
通し番号の穴への対処不要
古いイベント形式の変換マイグレーションで列を足せば済む

最初の 1 つを除いて、どれも継続的に保守するコードです。とくに通し番号の穴は、動くものを作った後で気付く種類の問題で、そこから直すのは高くつきます。

第 12 章の基準に照らして採ると決めたなら、この表が見積もりの下敷きになります。

まとめ​

ポイント内容
CQRS分けるのはメソッドでなくモデル。読み取りの要求が書き込みと食い違うときに効く
2 つのモデルの同期ドメインイベント経由。遅れが入るので、どの画面で許されるかを先に決める
イベントストアstream_id と stream_version の複合一意制約が並行制御そのもの
ドメイン層への影響replay() と apply() が増える。イベントソーシングはドメイン層まで届く
通し番号の穴採番順とコミット順は一致しない。素朴に「続きから読む」と読みモデルが恒久的に壊れる
スナップショット再生の短絡。第 10 章の「時点の値を写す」スナップショットとは別物

参考資料​

次章では、プロセスをまたいでイベントを届けるメッセージングと pub/sub を扱います。

Footnotes​

  1. 番号が決まるのは INSERT を実行した時点で、その値はトランザクションを戻しても再利用されません (MySQL は「AUTO_INCREMENT Handling in InnoDB」、PostgreSQL はシーケンス関数の注記と CREATE SEQUENCE の注記)。一方で、その行が他のセッションから見えるようになるのはコミットの後です。番号を採る順と見えるようになる順が別々に決まるので、2 つはずれます。 ↩

  2. Ecotone「Gap Detection and Consistency」。同ドキュメントは位置を "15:10,12,14" (15 まで進んだが 10・12・14 は欠けている) の形で持ち、欠けた位置が後から埋まれば処理して一覧から消す方式を採っています。 ↩

  3. Axon Framework「Kafka Extension Reference」。Kafka が配信の機構であってイベントストアではないこと、集約の再構成には別のイベントストア実装が要ることを述べている。 ↩