AWSで実践するイベント駆動アーキテクチャ 第3回:難しい基礎

もう一度、Outboxパターンを振り返ってみましょう。このパターンが解決するのは、「ビジネストランザクションがすでにコミットされた後にメッセージを発行しようとすると、そもそもイベントが発行されないことがある」というよくある問題です。このパターンを導入することで、すべてのイベントが 少なくとも1回(at-least-once) は発行されることが保証されます。この「少なくとも1回」という点について補足しましょう。

図1: Outboxパターン

Publishing Relayは運用データベースからイベントを取得し、発行が完了したらレコードを削除する、あるいは発行済みとしてマークすることで確認応答を行います。では、イベントを発行した直後、発行済みとマークする前にプロセスが落ちたらどうなるでしょうか。次の試行で同じイベントが再度発行され、サブスクライバーは同じイベントを複数回受け取ることになります。これは大きな問題でしょうか。いいえ、まったく問題ありません。

サービスが購読しているイベントの処理にAWS SQSを利用しているなら(利用すべきです)、SQSがそもそも「at-least-once delivery」を保証していることを押さえておきましょう。

「Amazon SQSは、冗長性と高可用性のため、メッセージのコピーを複数のサーバーに保存します。まれに、メッセージを受信または削除するときに、メッセージのコピーを保存しているサーバーの1つが利用できないことがあります……このような場合、メッセージを受信した際に同じメッセージのコピーをもう一度受け取る可能性があります。」

これはAWS SQSに限った話ではなく、(誠実な)分散メッセージバス全般に当てはまる事実です。ここで、こう疑問に思うかもしれません。SQS FIFOキューはどうなのか、と。ドキュメントによれば、SQS FIFOキューはまさにこの問題を解決するためのものとされています。

「標準キューとは異なり、FIFOキューは重複メッセージを発生させません。FIFOキューは、キューへの重複送信を回避するのに役立ちます。5分間の重複排除期間内にSendMessageアクションをリトライしても、Amazon SQSはキューに重複を生じさせません。」

しかし、これは問題の一部しか解決しません。確かに、Outboxが同じメッセージを(5分以内に)複数回発行した場合、FIFOキューは重複を検知して無視してくれます。ただし、ここではコンシューマー側に目を向けてみましょう。分散キューからメッセージを処理する流れは、次の3ステップに分けられます。

1. 次に利用可能なメッセージを取得する。
2. メッセージを処理する。
3. メッセージを処理済みとしてマークするか、SQSの場合はキューから削除して確認応答を行う。

ここで徹底した悲観論者になりきって、前回の記事でも投げかけた次の3つの問いを考えてみてください。

1. ここで何が起こり得るか?
2. その結果、どのような影響があるか?
3. このメッセージ処理フローは信頼できると言えるか?

先を読み進める前に、ぜひ一度立ち止まって考えてみてください。

では、答え合わせをしましょう。

もしメッセージを処理した後、確認応答を行う前に何かが失敗すれば、次回の実行で同じメッセージが再び取得され、複数回処理されてしまいます。FIFOキューは「exactly-once processing」を謳っていますが、その保証はSQS側でのメッセージ処理に対するものであり、あなたのアプリケーション内での処理に対するものではありません。あなたにとってどちらが重要でしょうか。もちろん後者です。したがって、信頼性の高い分散システムを設計するには、「あらゆるメッセージはサブスクライバーに複数回配信され得る」と前提を置く必要があります。

重複配信の可能性を認めているSQSのドキュメントに戻ると、その対処法も示されています。

アプリケーションを冪等(idempotent)に設計しましょう(同じメッセージを複数回処理しても悪影響が出ないようにします)。

これは的確なアドバイスです。同じイベントやコマンドをアプリケーションロジックに何度渡しても結果が変わらないのであれば、メッセージが重複してしまうエッジケースに頭を悩ませる必要はありません。

では、信頼性が高く、堅牢な冪等メッセージ処理ロジックを実装するのは簡単でしょうか。まったくそんなことはありません。ただし、その実装方法を語る前に、まず「やってはいけない方法」について少し触れておきます。

冪等な処理を実現する手段として、多くの場合Idempotent Consumerパターンが採用されます。パターンの根底にある発想自体に問題はないのですが、実装の大半は、そもそもなぜこのパターンが必要なのかという肝心な理由を取り違えています。一般的な実装は次のようなものです。

まず、アプリケーションが処理するコマンド/イベントには、それぞれ一意の識別子を割り当てます。次に「冪等性ストア」を用意します。これはリクエストIDとその結果を対応づけるキー/バリュー型のデータベースです(例:Redis、DynamoDB、インメモリマップなど)。

この2つの要件が揃ったら、受信リクエストの処理は次のシンプルなロジックに従います。

受信リクエストが届いたら、まずそのIDが冪等性ストアにすでに存在するかをチェックします。存在すれば、保存されている結果を返すだけで、処理は不要です。存在しなければ、次のように進めます。

1. リクエストを実行する。
2. IDと結果を冪等性ストアに保存する。

では、もう一度悲観論者の帽子をかぶって、ここで何が起こり得るか考えてみてください。

もちろん、何らかの理由で処理が完了した直後、結果を冪等性ストアに保存する前にプロセスが落ちれば、操作は複数回実行されることになります。

「こんな素朴な解決策が本当に使われているのか」と疑問に思うかもしれません。それなら、AWS Lambda Powertoolsに関するこのブログを読んでみてください。

このアルゴリズムには、もう少し高度なバージョンもあります。リクエストをまず「処理中」とマークし、一定時間が経過したらタイムアウトさせる方式です。しかし弱点は同じで、処理完了後、冪等性ストアが更新される前にプロセスが落ちれば、リクエストは再度処理されてしまいます。

一般に、操作対象のデータベースが冪等性ストアと同じトランザクションに参加していない限り、リクエストが複数回処理される可能性は残ります。

ここで「そんなレアなエッジケースに、わざわざ気を配る必要があるのか?」と感じるかもしれません。しかし、それは私たちが本来解決しようとしている問題そのものについても同じことが言えます。具体的なシステム次第では許容できる場合もあるでしょう。もしそうなら、わざわざシステムに余計な構成要素を持ち込む必要はありません。いずれにせよ、重複処理を許すか許さないかは、意識的に下すべき設計判断です。

では、信頼性が高く堅牢な冪等メッセージ処理ロジックは、どう実装すればよいのでしょうか。

Outboxパターンと同様に、Idempotent Consumerパターンを信頼性高く実装するには、ビジネスロジックと処理済みイベントの追跡の両方をまたぐアトミックなトランザクションが必要です。また、Outboxの場合と同じく、実装の具体的な方法は技術スタックによって変わります。

運用データベースが複数レコードのトランザクションをサポートしているなら、冪等性キー(=受信リクエストのID)を保存するテーブルを作成します(私はこのテーブルを「inbox」と呼ぶのが気に入っています)。リクエストの実行が終わったら、運用データの更新と冪等性キーの挿入を1つのアトミックトランザクションでまとめて行います。ただし、キーの挿入には条件を付ける必要があります。同じ値がテーブルにまだ存在しない場合のみ成功するようにし、すでに存在すれば失敗させるのです。失敗した場合、その受信リクエストはすでに処理済みだとわかります。

...

response = dynamodb.transact_write_items(
    TransactItems=[\
        {\
            'Put': {\
                'TableName': 'operational_data',\
                'Item': {\
                    ...\
                }\
            }\
        },\
        {\
            'Put': {\
                'TableName': 'inbox',\
                'Item': {\
                    'request_id': {'S': event_id},\
                },\
                'ConditionExpression': 'attribute_not_exists(request_id)'\
            }\
        }\
    ]
)

...

追加の最適化として、処理ロジックを実行する前に冪等性キーの有無を確認しておくこともできます。

使っているデータベースが複数レコードトランザクションをサポートしていない場合でも、楽観的並行性制御(Optimistic Concurrency Control)を活用すれば、信頼性の高い冪等処理を実装できます。

1. 各操作はテーブル内の単一レコード(例:JSONドキュメント)の更新で完結する。
2. 各レコードには並行性例外を制御するためのバージョンフィールドを持たせる。
3. レコードを更新するたびにバージョン値をインクリメントする。
4. 更新時にデータベースは、上書き対象レコードのバージョンが最初に読み取った時点のバージョンと一致することを保証しなければならない。

楽観的並行性制御が整ったら、管理対象レコードに processed_events(配列)というフィールドを追加で持たせられます。リクエストを実行する前に、そのリクエストのIDが processed_events 配列に含まれているかをチェックし、含まれていなければイベントの処理を進めて、最後にそのIDを配列に追加します。

1. Event-Driven Architecture on AWS, Part I: The Basics
2. Event-Driven Architecture on AWS, Part II: The Advanced Basics
3. Event-Driven Architecture on AWS, Part III: The Hard Basics(本記事)

原文初出: https://vladikk.com。