メッセージセマンティクス

メッセージセマンティクスは、システムが再試行、接続切断、プロセスクラッシュが発生した後に、どのような保証を依然として行えるかを記述するものです。「正確に1回（Exactly-once）」を議論する前に、数え上げる対象を明確にする必要があります。それは、ネットワーク上の送信回数、コンシューマーによる実行回数、それともデータベース内で最終的に有効となるビジネス操作の回数でしょうか。これらは必ずしも一致するわけではありません。

本稿では、まず確認応答（ACK）の喪失という事象を用いて問題点を解説し、その後、3つのセマンティクスと実装メカニズムについて議論します。Kafkaのログ構造については Kafkaアーキテクチャ を、パーティションの譲渡とオフセット管理については コンシューマーグループと協調動作 を参照してください。

問題の発生源：ACKを受信できないことは、成功していないことを意味しない

コンシューマーがイベント E42 を処理し、在庫を1つ減らした後、メッセージシステムに対して確認応答を行います。

メッセージシステムが見ているのは「ACKを受信していない」状態だけです。コンシューマーは処理を行っていなかった可能性もあれば、処理は完了していたがACKが消失した可能性もあります。前者の場合は再試行が必要ですが、後者の場合、再試行すると在庫が2回減らされることになります。タイムアウトを長く設定するだけでは、この不確実なウィンドウを解消できません。

同様の問題は、プロデューサーからブローカーへのパスにも存在します。プロデューサーの確認応答は、メッセージがブローカー側で何らかの受領条件を満たしたことを示します。コンシューマーの確認応答は、処理の進行状況を前進させられることを示します。これら2つの確認応答は互いに代替できません。RabbitMQ はこれらを明確に Publisher Confirms と Consumer Acknowledgements に分けています。

3つのセマンティクスの正確な意味

「少なくとも1回」は、メッセージの保持、永続化、回復能力、再試行戦略に依存します。キューの有効期限切れによる削除、再試行回数の枯渇、またはすべてのレプリカの消失は、無条件の保証対象外です。「最大1回」も、単にどこかの再試行を無効にすれば完了するわけではありません。アプリケーション層で同じビジネスリクエストを再提出すると、異なる2つのメッセージとして生成される可能性があります。

「正確に1回」は、処理関数が物理的に1回だけ実行されることを要求するものではありません。1回目の実行でロールバックされ、2回目の実行でコミットされることもあり、外部からは1回分の結果しか見えません。FLP不可能性定理から「正確に1回の効果は不可能」と直接導き出すことはできません。FLPは特定の非同期故障モデルにおける決定性コンセンサスの停止性について議論するものであり、ここではトランザクションの境界と可視化された結果の分析に取って代わるものではありません。

確認応答の順序が故障ウィンドウをどのように決定するか

ビジネス効果を E、確認応答または進行状況のコミットを C とします。これらが独立している場合、順序は以下の2通りしかありません。

したがって、単に順序を入れ替えるだけでは、漏れと重複の両方を同時に解消することはできません。システムはさらに、E と C をアトミックに発生させるか、E の重複を許容しつつ、その重複を副作用のない操作に変換する必要があります。これがトランザクションと冪等性の役割です。

冪等性：重複リクエストが同じビジネス結果を得る方法

数学的に、操作 f が冪等であるとは、f(f(x)) = f(x) が成り立つことを意味します。状態を CONFIRMED に設定する場合、そのフィールドへの代入のみを考慮すれば冪等ですが、数量を1つ加算する場合は冪等ではありません。しかし、代入の横にメール送信、課金、監査イベントの書き出しなどの処理があれば、処理関数全体は冪等であるとは限りません。

ビジネスレベルでのデジュープリケーションには、安定したイベントIDが必要です。同じイベントの再送信ではIDを再利用しなければなりません。再試行のたびに新しいUUIDを生成すると、デジュープリケーションは無効になります。一般的なキーは (consumer_name, event_id) です。これにより、同じイベントを異なるビジネスコンシューマーがそれぞれ1回ずつ処理できます。

以下は、PostgreSQL におけるアトミックな消費の例です。前提として、在庫テーブルで product_id が一意であり、処理の成功は在庫の増加として定義されます。メッセージ形式のバリデーションはトランザクションの前に行われます。

CREATE TABLE inventory (
    product_id bigint PRIMARY KEY,
    count bigint NOT NULL
);

CREATE TABLE processed_events (
    consumer_name text NOT NULL,
    event_id text NOT NULL,
    PRIMARY KEY (consumer_name, event_id)
);

-- 引数例：consumer_name='inventory', event_id='E42'
BEGIN;
WITH first_delivery AS (
    INSERT INTO processed_events (consumer_name, event_id)
    VALUES ('inventory', 'E42')
    ON CONFLICT DO NOTHING
    RETURNING event_id
)
INSERT INTO inventory (product_id, count)
SELECT 42, 1 FROM first_delivery
ON CONFLICT (product_id)
DO UPDATE SET count = inventory.count + EXCLUDED.count;
COMMIT;
-- COMMIT の成功後にのみメッセージを確認応答する

1回目のデジュープリケーションキーの挿入が成功すると在庫が増加し、2回目の first_delivery は空となり、在庫は変更されません。在庫の更新に失敗した場合、トランザクション全体がロールバックされ、デジュープリケーションキーも残りません。これは PostgreSQL の ON CONFLICT と RETURNING のセマンティクス に依存しています。

この例は2つの境界条件も示しています。デジュープリケーション記録と在庫が2つの独立したデータベースにある場合、上記のアトミック性は成立しません。また、E42 のデジュープリケーション記録を事前に削除すると、古いメッセージの再配信時に再び有効化される可能性があります。デジュープリケーションの保持期間は、ビジネスが許容する再配信範囲をカバーしていなければなりません。

バージョン条件による更新も上書きを防ぎますが、version mismatch は状態が変化しただけを示すものであり、「このイベントがすでに処理された」ことを証明するには不十分です。イベントの同一性と並行バージョンの解決は、2つの異なる問題を扱っています。

演習：トランザクションコミット前後の失敗がどのような状態を残すか

ローカルの SQLite ファイルを使用して在庫コンシューマーをシミュレートします。初期在庫は 100 で、E42 は 3 を減らします。ここではイベント内容を整数の delta として簡略化しています。デジュープリケーション記録には delta も同時に保存し、同じ ID が異なる操作に対応している場合に静かに無視されるのを防いでいます。実際のイベントでは、商品、テナント、ビジネスパラメータを比較するか、正規化されたコンテンツのダイジェストを比較する必要があります。

import sqlite3
from pathlib import Path
from tempfile import TemporaryDirectory

class Interrupted(Exception):
    pass

def consume(path, event_id, delta, fail_at=None):
    db = sqlite3.connect(path, isolation_level=None)
    try:
        db.execute("BEGIN IMMEDIATE")
        inserted = db.execute(
            "INSERT INTO done VALUES (?, ?) ON CONFLICT(id) DO NOTHING",
            (event_id, delta),
        ).rowcount
        if not inserted:
            old_delta = db.execute(
                "SELECT delta FROM done WHERE id = ?", (event_id,)
            ).fetchone()[0]
            if old_delta != delta:
                raise ValueError("同じイベント ID が異なる内容に対応している")
        else:
            if fail_at == "after_claim":
                raise Interrupted()
            changed = db.execute(
                "UPDATE stock SET n = n + ? WHERE n + ? >= 0",
                (delta, delta),
            ).rowcount
            if changed != 1:
                raise ValueError("在庫不足")
            if fail_at == "after_effect":
                raise Interrupted()
        db.execute("COMMIT")
        if fail_at == "after_commit":
            raise Interrupted()  # 結果はコミット済みだが、呼び出し側は成功応答を得ていない
    except Exception:
        if db.in_transaction:
            db.execute("ROLLBACK")
        raise
    finally:
        db.close()

def state(path):
    with sqlite3.connect(path) as db:
        return (db.execute("SELECT n FROM stock").fetchone()[0],
                db.execute("SELECT count(*) FROM done").fetchone()[0])

with TemporaryDirectory() as directory:
    for cut in ("after_claim", "after_effect", "after_commit"):
        path = Path(directory) / f"{cut}.sqlite"
        with sqlite3.connect(path) as db:
            db.executescript("""
                CREATE TABLE stock (n INTEGER NOT NULL CHECK(n >= 0));
                INSERT INTO stock VALUES (100);
                CREATE TABLE done (id TEXT PRIMARY KEY, delta INTEGER NOT NULL);
            """)
        try:
            consume(path, "E42", -3, cut)
        except Interrupted:
            pass
        assert state(path) == ((97, 1) if cut == "after_commit" else (100, 0))
        consume(path, "E42", -3)  # 呼び出し側は前回の結果を知らないため再試行
        consume(path, "E42", -3)  # メッセージが再配信される
        assert state(path) == (97, 1)
        for event_id, delta in (("E42", -4), ("E43", -200)):
            try:
                consume(path, event_id, delta)
            except ValueError:
                pass
            else:
                raise AssertionError("内容の競合または在庫不足を拒否すべき")
        assert state(path) == (97, 1)

各呼び出しでデータベースを再接続し、コミット後に読み取り可能な状態を観察します。ここでは Python の例外を注入して明示的にロールバックしており、トランザクションの境界を示しています。これは停電、ディスク障害、またはメッセージブローカーをシミュレートしたものではありません。SQLite は1つの書き込みトランザクションのみを許可しており、BEGIN IMMEDIATE は他の書き込み者が存在する場合にビジー状態を返す可能性があります。本番環境のコンシューマーでは、このような再試行可能な失敗を処理する必要があります。具体的なトランザクションの動作については SQLite トランザクションドキュメント を参照してください。

3行目が、冪等記録が真に解消する不確実なウィンドウです。もしデジュープリケーションキーが独立したトランザクションで先にコミットされていた場合、1行目は「処理済みだが在庫は変更されていない」状態になります。もしビジネス処理が先に独立してコミットされていた場合、2行目は「在庫は変更されたがデジュープリケーションの証拠がない」状態になります。2つのテーブルを同じデータベースに置くだけでは不十分で、同じトランザクションによってコミットされなければなりません。

在庫不足はここでは実行拒否として定義されているため、成功したデジュープリケーション記録は残りません。ビジネスシステムはこれを照会可能な拒否結果として記録できますが、この結果がその後の再試行を許容するかどうかを明確にする必要があります。もし再試行時に最初に生成された注文番号、領収書、または拒否理由を返す必要がある場合、その結果をトランザクション内で保存しなければなりません。「存在した」ID を記録するだけでは、応答を再構築することはできません。

応答が消失した場合、再配信で再び在庫が減らされるか？

同じ E42 に対する「処理試行回数」と「永続的な減算回数」をそれぞれカウントすることで、正確に1回の効果を正確に議論できます。コミット前にクラッシュする場合もあれば、コミット後、ACK 前にクラッシュする場合もあります。前者は一時データを破棄し、後者はデータベースの結果を保持しますが、ブローカーは再配信を継続できます。戦略を切り替えると、この一連の履歴がリセットされ、デジュープリケーション記録が効果とアトミックな境界を共有しているかどうかを比較することになります。

Kafka の冪等生産とトランザクション

Kafka の冪等生産は、プロデューサーの識別子、エポック、パーティション内のシーケンス番号をプロトコルに組み込み、クライアントが再試行によって生成した重複バッチを識別します。これはビジネスコンテンツによるデジュープリケーションとは異なります。アプリケーションが同じ注文を新しいレコードとして再送信した場合、正常にアペンドされる可能性があります。関連する制約については Producer Configs を参照してください。

トランザクションは、複数のパーティションへの出力と消費の進行状況を1回のコミットに統合します。Kafka から Kafka への変換は以下のように記述できます。

クラッシュ時、トランザクションはコミットされていない場合もあれば、コミット済みだが応答が消失している場合もあります。回復はトランザクションプロトコルによって処理されます。ダウンストリームでは read_committed を使用して、コミットされていないまたは中止されたトランザクションのレコードを除外します。「結果の可視性」と「回復時にどこから継続するか」を、同じアトミックな決定として捉えるべきです。メカニズムと制限については Kafka メッセージ配信セマンティクス / トランザクション を参照してください。

Kafka トランザクションは、外部の SQL 更新、HTTP リクエスト、またはメール送信を自動的に含まないことに注意してください。データベース側では独自のトランザクションとデジュープリケーションを使用する必要があります。外部インターフェースには安定した冪等キー、または不確実な結果を解消するための照会可能なビジネス状態が必要です。

Outbox：データベースコミット後、イベントをどのように確実に送信するか

「先にデータベースに書き込み、その後メッセージを送信する」アプローチにも、2つの独立したコミットポイントがあります。データベースでの成功後にプロセスがクラッシュすると、メッセージは永遠に送信されない可能性があります。逆に、先にメッセージを送信すると、最終的にコミットされないビジネス事実が公開される可能性があります。

Outbox パターンでは、ビジネス変更と公開待ちイベントを同じデータベーストランザクションに書き込み、独立したフォワーダーがイベントを読み取って配信します。

フォワーダーが「送信済み、未マーク」の状態でクラッシュしても再送信されるため、Outbox は主に送信漏れを解決し、コンシューマーのデジュープリケーションが重複を処理します。この2つが連携して初めて、完全なチェーンが形成されます。より一般的なトランザクションのトレードオフについては 分散トランザクション を参照してください。

保証境界を素早く判断する

メッセージチェーンを確認する際は、以下の順で決定します。誰が安定したイベントIDを生成するか、どのステップで再試行が可能か、ACK は実際に何を確定させるのか、ビジネス効果と進行状況を同じトランザクションで行えるか、外部の副作用をどのようにデジュープリケートするか、デジュープリケーション記録と元のメッセージはどの程度保持されるか。

これらの問いに明確な答えがない場合、「exactly-once を有効にする」と記述しただけでは、システムが実際にどのような保証を提供しているかは説明できません。

参考文献

• RabbitMQ: 確認メカニズム
• Kafka: デザインとメッセージセマンティクス
• Kafka: プロデューサー設定
• SQLite: トランザクションと明示的なロールバック
• PostgreSQL: INSERT

キーワード：メッセージセマンティクス、確認応答、再試行、冪等性、デジュープリケーション、exactly-once、トランザクション境界、Outbox