hybrid-cloud-outboxes

混合云发件箱

Sentry 使用事务性发件箱模式来处理最终一致性的操作。当模型发生变更时，会在同一数据库事务内写入一行发件箱记录。事务提交后，发件箱被排空——触发一个信号，进而触发副作用，例如 RPC 调用、墓碑传播或审计日志记录。

最常见的用例是跨 silo 数据复制：在 Cell silo 中保存的模型会产生一个 CellOutbox，处理该记录时会将数据复制到 Control silo（或通过 ControlOutbox 反向复制）。但该模式是通用的——对于任何需要在事务提交后可靠执行的操作，即使在单个 silo 内，发件箱都适用。

对应两个数据流向，有两种发件箱类型：

• CellOutbox — 在 Cell silo 中写入，在 Cell silo 中处理，将数据推送至 Control（通过信号接收器中的 RPC 调用）。
• ControlOutbox — 在 Control silo 中写入，在 Control silo 中处理，将数据推送至一个或多个 Cell silo。每个 ControlOutbox 行都针对特定的 cell_name。

关键约束

发件箱必须与数据变更在同一事务中写入。 Mixin 类（ReplicatedCellModel、ReplicatedControlModel）通过 prepare_outboxes() 自动强制执行此约束。如果你手动写入发件箱，请始终使用 outbox_context(transaction.atomic(...))。

处理器必须幂等。 发件箱在失败时可以重试，并且会被合并——处理器可能只会收到变更的最新版本，或者针对同一变更被多次调用。

drain_shard() 不得在事务内运行。 它会获取 SELECT FOR UPDATE 锁，并逐条处理消息。在事务内调用它会导致死锁或长时间持有锁。

合并后只有最新的负载会保留。 针对同一 (scope, shard_identifier, category, object_identifier) 的多次发件箱写入会被合并——只有 ID 最大的行会被处理。切勿依赖每个中间负载都会被投递。

每个 OutboxCategory 必须且只能注册到一个 OutboxScope。 导入时会通过断言强制执行此约束。注册到零个或多个 scope 的 category 会导致导入时崩溃。

批量操作必须使用产生式管理器。 使用来自 CellOutboxProducingManager 或 ControlOutboxProducingManager 的 MyModel.objects.bulk_create() / bulk_update() / bulk_delete()。原始 queryset 会绕过发件箱的创建。

Snowflake ID 模型不能使用 bulk_create。 产生式管理器通过 SELECT nextval(...) 预分配 ID，这与 snowflake ID 生成机制冲突。请改用逐个 save() 调用。

第 1 步：确定你需要什么

意图	前往
为新的模型添加发件箱复制	第 2 步
添加新的 OutboxCategory（不绑定到被复制的模型）	第 3 步
编写手动信号接收器（不使用模型 mixin）	第 4 步
将现有模型迁移为使用发件箱	第 5 步，然后第 6 步
为现有数据设置回填	第 6 步
测试基于发件箱的复制	第 7 步
调试卡住或未处理的发件箱	第 8 步

第 2 步：为新的模型添加发件箱复制

2.1 选择 Mixin

数据位于...	复制至...	Mixin	发件箱类型
Cell silo	Control silo	ReplicatedCellModel	CellOutbox
Control silo	Cell silo	ReplicatedControlModel	ControlOutbox

2.2 ReplicatedCellModel 模板

当 Cell 模型需要将数据复制到 Control silo 时，使用此模板。

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 silo 时，使用此模板。关键区别在于：Control 的 outbox 会扇出到一个或多个 cell，因此模型必须声明要面向哪些 cell。

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 接好 Category 连接

mixin 类会通过 OutboxCategory.connect_cell_model_updates()（或 connect_control_model_updates()）自动连接信号接收器。这会在类定义时、设置 category 类变量时发生。该连接会自动将事件分派到你的 handle_async_replication 和 handle_async_deletion 方法。

对于可复制模型，无需手动编写信号接收器——mixin 会处理。只有那些没有映射到可复制模型的 category 才需要手动接收器（见步骤 4）。

如果你的 OutboxCategory 尚不存在，请先创建它（步骤 3）。

步骤 3：添加新的 OutboxCategory

每种 outbox 消息类型都需要一个 OutboxCategory 枚举值，并注册到恰好一个 OutboxScope。

快速步骤：

1. 在 src/sentry/hybridcloud/outbox/category.py 的 OutboxCategory 枚举中添加一个新值
2. 将其注册到适当的 OutboxScope 下（决定分片键）
3. 如果使用模型 mixin，在模型上设置 category = OutboxCategory.MY_CATEGORY

加载 references/category-and-scope.md 以获取完整的 scope 到 category 映射、如何选择 scope 以及注册机制。

步骤 4：编写手动信号接收器

当 outbox category 未绑定到 ReplicatedCellModel 或 ReplicatedControlModel 时，使用手动接收器。常见情况：

• 仅在 payload 中携带全部数据的操作（审计日志、IP 事件）
• 由模型变更触发但并不直接复制该模型的操作
• 跨 silo 信号转发（SEND_SIGNAL、RESET_IDP_FLAGS）
• 需要自定义分派逻辑的复杂多步骤操作

加载 references/signal-receivers.md 以获取可直接复制的接收器模板、maybe_process_tombstone 模式以及放置规则。

步骤 5：将已有模型迁移为使用 Outbox

当为已经在生产环境中有数据的模型添加 outbox 复制时：

5.1 代码变更（非破坏性）

1. 将模型的基类改为 ReplicatedCellModel 或 ReplicatedControlModel
2. 添加 category 类变量
3. 添加生产管理器（CellOutboxProducingManager / ControlOutboxProducingManager）
4. 实现 handle_async_replication 和 handle_async_deletion
5. 如有需要，添加 payload_for_update() 以提供删除恢复数据
6. 如果 OutboxCategory 不存在，创建它（步骤 3）

这些变更是非破坏性的：新的模型保存会创建 outbox，但已有行尚未有 outbox。

5.2 回填已有数据

已有行需要追溯创建 outbox。在模型类上设置 replication_version = 2（或更高），并配置回填系统——见步骤 6。

步骤 6：设置回填

回填系统会为早于 outbox 集成的已有模型行创建 outbox。它按批次处理行，并通过 Postgres 中的 watermark 行进行跟踪。

加载 references/backfill.md 以了解 replication_version 机制、选项键格式、watermark 状态跟踪以及 SaaS 与可自托管的发布流程。

步骤 7：测试基于 outbox 的复制

有关详细的 outbox 测试模板和可直接复制的模式，请调用 hybrid-cloud-test-gen 技能。 以下指南涵盖要测试什么；hybrid-cloud-test-gen 涵盖如何生成测试代码。

7.1 核心测试工具

outbox_runner() —— 主要的测试工具。上下文管理器，在包裹的代码成功执行后同步清空所有待处理的 outbox：

from sentry.testutils.outbox import outbox_runner

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

它最多运行 10 次清空迭代（如果超出则抛出 OutboxRecursionLimitError）。可与 TestCase 配合使用——标准 outbox 测试无需使用 TransactionTestCase。

outbox_context(flush=False) —— 创建 outbox 记录但不处理它们。用于独立于处理过程验证 outbox 的创建：

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 —— 在测试中切换 silo 上下文以查询跨 silo 的模型：

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 应测试的内容

Outbox 创建 —— 验证保存/删除模型是否创建了具有正确 scope、category 和标识符的 outbox 行：

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

复制传播 —— 验证完整的往返流程：保存模型 -> 清空 outbox -> 跨 silo 效果：

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 并清理跨 silo 资源：

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()

幂等性 —— 验证两次清空同一 shard 不会产生重复或错误：

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 Silo 测试装饰器

• 对于聚焦于 CellOutbox 创建的测试，使用 @cell_silo_test
• 对于聚焦于 ControlOutbox 创建的测试，使用 @control_silo_test
• 对于同时运行两个 silo 的端到端复制测试，使用 @all_silo_test
• 仅在多线程/并发测试（例如 threading.Barrier）中使用 TransactionTestCase，不要用于标准的 outbox 清空测试

7.4 常见陷阱

• 工厂调用（self.create_organization() 等）绝不能包裹在 assume_test_silo_mode 中。工厂会在内部处理 silo 模式。
• outbox_runner() 在退出时会清空 outbox。如果需要检查 outbox 状态，请改用 outbox_context(flush=False)。
• 如果某个 outbox 处理器创建了更多 outbox（级联），outbox_runner 会自动处理这种情况（最多 10 次迭代）。

第 8 步：调试卡住的 Outbox

症状	可能原因	排查
数据未复制到其他 silo	处理器错误，outbox 处于退避状态	检查卡住的 outbox 上的 scheduled_for
测试中出现 OutboxFlushError	信号接收器抛出异常	阅读错误消息中包裹的异常
Outbox 行不断累积	清空任务未运行或失败	检查 Celery 任务日志中的 enqueue_outbox_jobs
Shard 清空缓慢	合并批次过大或处理器超时	检查 outbox.coalesced_net_processing_time 指标
导入崩溃：scope/category 断言	Category 注册到了错误或多个 scope	检查 category.py 中的 OutboxScope 注册

加载 references/debugging.md 以获取完整的处理管道演练、shard 检查方法、退避时间表、kill switch 以及有用的 SQL/指标查询。

第 9 步：验证（提交前检查清单）

在提交 PR 之前，请验证：

• 模型继承自 ReplicatedCellModel 或 ReplicatedControlModel（或使用手动 receiver）
• category 类变量已设置为正确的 OutboxCategory
• OutboxCategory 已注册到恰好一个 OutboxScope
• 所选的 OutboxScope 与模型的分片键（organization_id、user_id 等）匹配
• handle_async_replication 是幂等的（可安全地多次调用）
• handle_async_deletion 是幂等的，并能处理该行已被删除的情况
• payload_for_update() 仅包含删除恢复所需的数据（不包含快速变化的字段）
• 已在模型上设置生产 manager（CellOutboxProducingManager / ControlOutboxProducingManager）
• 批量操作通过生产 manager 进行，而不是直接使用原始 queryset
• ReplicatedControlModel 具有正确的 outbox_cell_names() 实现
• 测试验证 outbox 创建（scope、category、标识符）
• 测试验证端到端复制（保存 -> 排空 -> 跨 silo 效果）
• 测试验证删除传播（删除 -> 排空 -> 清理）
• 测试验证幂等性（排空两次 -> 无重复）
• 如果迁移现有模型，已提升 replication_version 并配置了回填