hybrid-cloud-outboxes

ハイブリッドクラウドアウトボックス

Sentry は、結果整合性のある操作のために トランザクショナルアウトボックスパターン を使用しています。モデルが変更されると、同じデータベーストランザクション内でアウトボックス行が書き込まれます。トランザクションがコミットされた後、アウトボックスがドレインされ、RPC 呼び出し、トゥームストーンの伝播、監査ログなどの副作用をトリガーするシグナルが発火します。

最も一般的なユースケースは クロスサイロデータレプリケーション です。Cell サイロに保存されたモデルは CellOutbox を生成し、それが処理されるとデータが Control サイロにレプリケートされます（または ControlOutbox を介してその逆も同様です）。しかし、このパターンは汎用的です。アウトボックスは、単一のサイロ内であっても、トランザクションのコミット後に確実に実行されるべきあらゆる操作に使用できます。

データの流れる方向に対応する 2 種類のアウトボックスがあります：

• CellOutbox — Cell サイロで書き込まれ、Cell サイロで処理されて、データを Control に向けてプッシュします（シグナルレシーバー内の RPC 呼び出しを介して）。
• ControlOutbox — Control サイロで書き込まれ、Control サイロで処理されて、データを 1 つ以上の Cell サイロに向けてプッシュします。各 ControlOutbox 行は特定の cell_name をターゲットにします。

重要な制約

アウトボックスは、データ変更と同じトランザクション内で書き込まなければなりません。 ミックスインクラス（ReplicatedCellModel、ReplicatedControlModel）は、prepare_outboxes() を介してこれを自動的に強制します。アウトボックスを手動で書き込む場合は、常に outbox_context(transaction.atomic(...)) を使用してください。

ハンドラーは冪等でなければなりません。 アウトボックスは失敗時に再試行される可能性があり、また合流（coalesce）されます。ハンドラーは変更の最新バージョンのみを受け取る場合や、同じ変更に対して複数回呼び出される場合があります。

drain_shard() はトランザクション内で実行してはなりません。 これは SELECT FOR UPDATE ロックを取得し、メッセージを 1 つずつ処理します。トランザクション内で呼び出すと、デッドロックが発生したり、ロックを長時間保持したりします。

合流後は最新のペイロードのみが残ります。 同じ (scope, shard_identifier, category, object_identifier) に対する複数のアウトボックス書き込みは合流され、最も大きい ID を持つ行のみが処理されます。すべての中間ペイロードが配信されることを前提にしてはなりません。

すべての OutboxCategory は、ちょうど 1 つの OutboxScope に登録しなければなりません。 インポート時にアサーションによってこれが強制されます。ゼロまたは複数のスコープに登録されたカテゴリは、インポートクラッシュを引き起こします。

一括操作では、producing マネージャーを使用しなければなりません。 CellOutboxProducingManager または ControlOutboxProducingManager の MyModel.objects.bulk_create() / bulk_update() / bulk_delete() を使用してください。生のクエリセットはアウトボックスの作成をバイパスします。

Snowflake ID モデルでは bulk_create を使用できません。 producing マネージャーは SELECT nextval(...) を介して ID を事前割り当てしますが、これは snowflake ID の生成と競合します。代わりに個別の save() 呼び出しを使用してください。

ステップ 1：必要なものを判断する

目的	参照先
新しいモデルにアウトボックスレプリケーションを追加する	ステップ 2
新しい OutboxCategory を追加する（レプリケートされたモデルに紐づかない）	ステップ 3
手動のシグナルレシーバーを書く（モデルミックスインを使用しない）	ステップ 4
既存のモデルをアウトボックスを使用するように移行する	ステップ 5、次にステップ 6
既存データのバックフィルを設定する	ステップ 6
アウトボックスベースのレプリケーションをテストする	ステップ 7
スタックした、または未処理のアウトボックスをデバッグする	ステップ 8

ステップ 2：新しいモデルにアウトボックスレプリケーションを追加する

2.1 ミックスインを選択する

データの保存場所	レプリケート先	ミックスイン	アウトボックスの種類
Cell サイロ	Control サイロ	ReplicatedCellModel	CellOutbox
Control サイロ	Cell サイロ	ReplicatedControlModel	ControlOutbox

2.2 ReplicatedCellModel テンプレート

Cell モデルが Control シロにデータを複製する必要がある場合にこれを使用します。

from sentry.backup.scopes import RelocationScope
from sentry.db.models import (
    FlexibleForeignKey,
    Model,
    cell_silo_model,
    sane_repr,
)
from sentry.db.models.manager.base_query_set import BaseQuerySet
from sentry.hybridcloud.outbox.base import ReplicatedCellModel, CellOutboxProducingManager
from sentry.hybridcloud.outbox.category import OutboxCategory


class MyModelManager(CellOutboxProducingManager["MyModel"]):
    """Manager that ensures bulk operations create outboxes."""
    pass


@cell_silo_model
class MyModel(ReplicatedCellModel):
    __relocation_scope__ = RelocationScope.Organization

    # Required: the OutboxCategory for this model (must already be registered)
    category = OutboxCategory.MY_MODEL_UPDATE

    # Use the producing manager for bulk operation support
    objects: ClassVar[MyModelManager] = MyModelManager()

    # Model fields...
    organization = FlexibleForeignKey("sentry.Organization")
    name = models.CharField(max_length=128)

    class Meta:
        app_label = "sentry"
        db_table = "sentry_mymodel"

    def payload_for_update(self) -> dict[str, Any] | None:
        """
        Optional: include data needed by the deletion handler.
        Keep payloads minimal — only data that cannot be recovered
        after the row is deleted. Payloads are coalesced (only the
        latest survives).
        """
        return None  # Override if needed

    @classmethod
    def handle_async_deletion(
        cls,
        identifier: int,
        shard_identifier: int,
        payload: Mapping[str, Any] | None,
    ) -> None:
        """
        Called when this object has been deleted (row no longer exists).
        Clean up cross-silo resources. Must be idempotent.
        """
        my_mapping_service.delete(
            my_model_id=identifier,
            organization_id=shard_identifier,
        )

    def handle_async_replication(self, shard_identifier: int) -> None:
        """
        Called when this object has been created or updated.
        Replicate to the control silo via RPC. Must be idempotent.
        """
        my_mapping_service.upsert(
            my_model_id=self.id,
            organization_id=shard_identifier,
            mapping=RpcMyModelMapping.from_orm(self),
        )

2.3 ReplicatedControlModel テンプレート

Control モデルが Cell シロにデータを複製する必要がある場合にこれを使用します。重要な違いは、Control のアウトボックスは 1 つ以上のセルにファンアウトするため、モデルは対象とするセルを宣言する必要があることです。

from sentry.db.models import control_silo_model
from sentry.hybridcloud.outbox.base import ReplicatedControlModel, ControlOutboxProducingManager
from sentry.hybridcloud.outbox.category import OutboxCategory


class MyControlModelManager(ControlOutboxProducingManager["MyControlModel"]):
    pass


@control_silo_model
class MyControlModel(ReplicatedControlModel):
    __relocation_scope__ = RelocationScope.Global

    category = OutboxCategory.MY_CONTROL_MODEL_UPDATE

    objects: ClassVar[MyControlModelManager] = MyControlModelManager()

    # Model fields...
    organization = FlexibleForeignKey("sentry.Organization")
    user = FlexibleForeignKey("sentry.User")

    class Meta:
        app_label = "sentry"
        db_table = "sentry_mycontrolmodel"

    def outbox_cell_names(self) -> Collection[str]:
        """
        Which cells should receive outboxes for this change.
        Default implementation checks organization_id then user_id.
        Override for custom logic (e.g., all cells, specific cells).
        """
        # Default: auto-detects from organization_id or user_id attributes.
        # Override only if the default doesn't work for your model.
        return super().outbox_cell_names()

    @classmethod
    def handle_async_deletion(
        cls,
        identifier: int,
        cell_name: str,
        shard_identifier: int,
        payload: Mapping[str, Any] | None,
    ) -> None:
        """Note: receives cell_name — one call per target cell."""
        pass

    def handle_async_replication(self, cell_name: str, shard_identifier: int) -> None:
        """Note: receives cell_name — one call per target cell."""
        pass

2.4 カテゴリ接続の配線

ミックスインクラスは、OutboxCategory.connect_cell_model_updates()（または connect_control_model_updates()）を介してシグナルレシーバーを自動接続します。これは、category クラス変数が設定されている場合、クラス定義時に実行されます。この接続は、handle_async_replication メソッドと handle_async_deletion メソッドに自動的にディスパッチします。

複製モデルでは手動のシグナルレシーバーは不要です。ミックスインが処理します。手動レシーバーが必要なのは、複製モデルにマッピングされないカテゴリのみです（ステップ 4 を参照）。

OutboxCategory がまだ存在しない場合は、まずそれを作成してください（ステップ 3）。

ステップ 3: 新しい OutboxCategory の追加

すべてのアウトボックスメッセージタイプには、ちょうど 1 つの OutboxScope に登録された OutboxCategory 列挙値が必要です。

簡単な手順:

1. src/sentry/hybridcloud/outbox/category.py の OutboxCategory 列挙に新しい値を追加します
2. 適切な OutboxScope（シャードキーを決定します）の下に登録します
3. モデルミックスインを使用する場合は、モデルに category = OutboxCategory.MY_CATEGORY を設定します

スコープからカテゴリへの完全なマッピング、スコープの選び方、登録の仕組みについては、references/category-and-scope.md をロードしてください。

ステップ 4: 手動シグナルレシーバーの作成

アウトボックスカテゴリが ReplicatedCellModel または ReplicatedControlModel に紐付いていない場合は、手動レシーバーを使用します。一般的なケース:

• すべてのデータをペイロードに含むペイロードのみの操作（監査ログ、IP イベント）
• モデルの変更によってトリガーされるが、そのモデルを直接複製しないアクション
• シロ間のシグナル転送（SEND_SIGNAL、RESET_IDP_FLAGS）
• カスタムディスパッチロジックを必要とする複雑な多段階操作

コピーして使えるレシーバーテンプレート、maybe_process_tombstone パターン、配置ルールについては、references/signal-receivers.md をロードしてください。

ステップ 5: 既存モデルのアウトボックス使用への移行

本番環境にすでにデータがあるモデルにアウトボックス複製を追加する場合:

5.1 コード変更（非破壊的）

1. モデルの基底クラスを ReplicatedCellModel または ReplicatedControlModel に変更します
2. category クラス変数を追加します
3. プロデューシングマネージャー（CellOutboxProducingManager / ControlOutboxProducingManager）を追加します
4. handle_async_replication と handle_async_deletion を実装します
5. 必要に応じて、削除復旧データ用に payload_for_update() を追加します
6. OutboxCategory が存在しない場合は作成します（ステップ 3）

これらの変更は非破壊的です。新しいモデルの保存ではアウトボックスが作成されますが、既存の行にはまだアウトボックスがありません。

5.2 既存データのバックフィル

既存の行には、遡ってアウトボックスを作成する必要があります。モデルクラスに replication_version = 2（またはそれ以上）を設定し、バックフィルシステムを構成します — ステップ 6 を参照してください。

ステップ 6: バックフィルの設定

バックフィルシステムは、アウトボックス統合より前に存在する既存のモデル行に対してアウトボックスを作成します。行をバッチ単位で処理し、Postgres のウォーターマーク行で追跡します。

replication_version の仕組み、オプションキーの形式、ウォーターマークの状態追跡、SaaS とセルフホストでのロールアウト手順については、references/backfill.md をロードしてください。

ステップ 7: アウトベースのレプリケーションのテスト

詳細なアウトボックステストテンプレートとコピーして使えるパターンについては、hybrid-cloud-test-gen スキルを呼び出してください。 以下のガイダンスは何をテストするかを説明しており、hybrid-cloud-test-gen はテストコードの生成方法を説明しています。

7.1 コアテストユーティリティ

outbox_runner() — 主要なテストツールです。ラップされたコードが成功した後、保留中のすべてのアウトボックスを同期的に処理するコンテキストマネージャーです:

from sentry.testutils.outbox import outbox_runner

with outbox_runner():
    my_model.save()
# All outboxes drained — cross-silo effects have happened

最大 10 回のドレイン反復を実行します（超過すると OutboxRecursionLimitError を発生させます）。TestCase で動作します — 標準的なアウトボックステストに TransactionTestCase は不要です。

outbox_context(flush=False) — アウトボックスレコードを処理せずに作成します。処理とは独立してアウトボックスの作成を検証するために使用します:

from sentry.hybridcloud.models.outbox import outbox_context

with outbox_context(flush=False):
    MyModel(id=10).outbox_for_update().save()

assert CellOutbox.objects.count() == 1

assume_test_silo_mode / assume_test_silo_mode_of — テスト内でサイロコンテキストを切り替え、クロスサイロモデルをクエリします:

from sentry.testutils.silo import assume_test_silo_mode_of

with assume_test_silo_mode_of(MyMapping):
    assert MyMapping.objects.filter(my_model_id=obj.id).exists()

7.2 テストすべき内容

アウトボックスの作成 — モデルの保存/削除が、正しいスコープ、カテゴリ、識別子を持つアウトボックス行を作成することを検証します:

def test_outbox_created_on_save(self):
    with outbox_context(flush=False):
        obj = MyModel(id=10, organization_id=1)
        obj.outbox_for_update().save()

    outbox = CellOutbox.objects.first()
    assert outbox.category == OutboxCategory.MY_MODEL_UPDATE.value
    assert outbox.shard_scope == OutboxScope.ORGANIZATION_SCOPE.value
    assert outbox.shard_identifier == 1

レプリケーションの伝播 — 完全なラウンドトリップを検証します: モデルの保存 -> アウトボックスのドレイン -> クロスサイロ効果:

def test_replication_creates_mapping(self):
    org = self.create_organization()
    with outbox_runner():
        obj = MyModel.objects.create(organization=org, name="test")

    with assume_test_silo_mode_of(MyMapping):
        mapping = MyMapping.objects.get(my_model_id=obj.id)
        assert mapping.name == "test"

削除と tombstone — モデルの削除が handle_async_deletion をトリガーし、クロスサイロリソースをクリーンアップすることを検証します:

def test_delete_cleans_up_mapping(self):
    org = self.create_organization()
    with outbox_runner():
        obj = MyModel.objects.create(organization=org, name="test")

    with outbox_runner():
        obj.delete()

    with assume_test_silo_mode_of(MyMapping):
        assert not MyMapping.objects.filter(my_model_id=obj.id).exists()

冪等性 — 同じシャードを 2 回ドレインしても重複やエラーが発生しないことを検証します:

def test_idempotent_replication(self):
    with outbox_runner():
        obj = MyModel.objects.create(organization=org, name="test")

    with assume_test_silo_mode_of(MyMapping):
        count_after_first = MyMapping.objects.count()

    with outbox_runner():
        pass  # Drain again — should be a no-op

    with assume_test_silo_mode_of(MyMapping):
        assert MyMapping.objects.count() == count_after_first

7.3 サイロテストデコレーター

• CellOutbox の作成に焦点を当てたテストには @cell_silo_test を使用します
• ControlOutbox の作成に焦点を当てたテストには @control_silo_test を使用します
• 両方のサイロを実行するエンドツーエンドのレプリケーションテストには @all_silo_test を使用します
• TransactionTestCase はスレッド/並行性テスト（例: threading.Barrier）にのみ使用し、標準的なアウトボックスドレインテストには使用しないでください

7.4 よくある落とし穴

• ファクトリー呼び出し（self.create_organization() など）は決して assume_test_silo_mode でラップしてはいけません。ファクトリーはサイロモードを内部的に処理します。
• outbox_runner() は終了時にアウトボックスをクリアします。アウトボックスの状態を検査する必要がある場合は、代わりに outbox_context(flush=False) を使用してください。
• アウトボックスハンドラーがさらにアウトボックスを作成する場合（カスケード）、outbox_runner はこれを自動的に処理します（最大 10 回の反復）。

Step 8: スタックしたアウトボックスのデバッグ

症状	考えられる原因	調査
データが他のサイロにレプリケートされない	ハンドラーエラー、バックオフ中のアウトボックス	スタックしたアウトボックスの scheduled_for を確認する
テストで OutboxFlushError が発生する	シグナルレシーバーが例外を発生させる	エラーメッセージ内のラップされた例外を読む
アウトボックス行が蓄積される	ドレインタスクが実行されていないか失敗している	enqueue_outbox_jobs の Celery タスクログを確認する
シャードのドレインが遅い	大きな合体バッチまたはハンドラーのタイムアウト	outbox.coalesced_net_processing_time メトリクスを確認する
インポートクラッシュ: スコープ/カテゴリのアサーション	カテゴリが誤った、または複数のスコープに登録されている	category.py の OutboxScope 登録を確認する

完全な処理パイプラインのウォークスルー、シャード検査方法、バックオフスケジュール、キルスイッチ、および有用な SQL/メトリクスクエリについては、references/debugging.md を読み込んでください。

ステップ 9: 検証 (事前チェックリスト)

PR を送信する前に、以下を検証してください:

• モデルが ReplicatedCellModel または ReplicatedControlModel を継承している (または手動レシーバーを使用している)
• category クラス変数が正しい OutboxCategory に設定されている
• OutboxCategory がちょうど 1 つの OutboxScope に登録されている
• 選択した OutboxScope がモデルのシャードキー (organization_id、user_id など) と一致している
• handle_async_replication が冪等である (複数回呼び出しても安全である)
• handle_async_deletion が冪等であり、行がすでに存在しない場合を処理している
• payload_for_update() に削除復旧に必要なデータのみが含まれている (頻繁に変化するフィールドは含まない)
• 生成マネージャー (CellOutboxProducingManager / ControlOutboxProducingManager) がモデルに設定されている
• 一括操作が生のクエリセットではなく、生成マネージャーを経由している
• ReplicatedControlModel が正しい outbox_cell_names() の実装を持っている
• テストがアウトボックスの作成 (スコープ、カテゴリ、識別子) を検証している
• テストがエンドツーエンドのレプリケーション (保存 -> ドレイン -> クロスサイロ効果) を検証している
• テストが削除の伝播 (削除 -> ドレイン -> クリーンアップ) を検証している
• テストが冪等性 (2 回ドレイン -> 重複なし) を検証している
• 既存のモデルを移行する場合、replication_version が更新され、バックフィルが設定されている