Sentry Outbox 反向回填机制深度解析：replication_version 版本门控与 Redis 游标增量复制指南

本文是 Sentry（sentry/hybridcloud）事务性 Outbox 跨域复制体系中的工程实操指南，围绕"模型迁移到 Outbox 之后如何为历史存量行补建 outbox、使其被增量复制到对端 silo"这一核心问题展开。读者将掌握 replication_version 版本门控的运作原理、Redis 游标的状态推进规则、SaaS 灰度与自托管同步两种回填上线方式，以及基于真实源码与测试用例的监控与验证手段。

在 Sentry 混合云架构中，当某个模型被改造为基于 Outbox 的复制模型（继承 ReplicatedCellModel 或 ReplicatedControlModel，参见 .agents/skills/hybrid-cloud-outboxes/SKILL.md），或它的复制逻辑发生变化时，数据库中已经存在的存量行并不会自动获得 outbox——只有新写入的数据才会触发 outbox 生成。backfill_outboxes.py 实现的回填系统正是为了解决这一问题：它按批次增量地处理存量行，用 Redis 记录游标进度，并通过 Sentry options 体系做版本门控。以下内容全部以当前仓库源码为据。

1. 回填系统总览

回填系统核心实现位于 src/sentry/hybridcloud/tasks/backfill_outboxes.py，其工作方式可概括为：

1. 当一个模型的 replication_version 类变量被上调时，回填任务就会把"补建 outbox"当成后台工作增量执行；
2. 每批处理以 Redis 中记录的 (lower_bound_id, current_version) 游标为起点，通过主键 ID 区间扫描模型表；
3. 处理进度由 Sentry options 系统中的版本门控控制，决定回填是否已经"被允许"推进到代码中的最新复制版本。

换句话说，Sentry options 决定"是否该跑这次回填"，Redis 游标决定"从哪里继续跑"。两者相互配合，构成了一个可暂停、可恢复、可灰度、可回滚的存量复制机制。

2. replication_version 版本机制

每个 outbox 复制模型的基类上都声明了默认版本号。在 src/sentry/hybridcloud/outbox/base.py#L42-L43（Cell 侧基类 CellOutboxProducingModel）与 src/sentry/hybridcloud/outbox/base.py#L227-L228（Control 侧基类 ControlOutboxProducingModel）中：

class CellOutboxProducingModel(Model):
    default_flush: bool | None = None
    replication_version: int = 1   # 默认版本

class ControlOutboxProducingModel(Model):
    default_flush: bool | None = None
    replication_version: int = 1

实际业务模型一般继承 ReplicatedCellModel（Cell silo → Control silo，见 base.py#L152）或 ReplicatedControlModel（Control silo → 各 Cell silo，见 base.py#L336），它们再分别继承上述两个产生者基类。一旦模型类上的 replication_version 上调，所有该表的存量行就需要重新生成 outbox 完成一次"全量复制"，这正是回填的触发条件：

class MyModel(ReplicatedCellModel):
    replication_version = 2  # 原来是 1；上调即触发新一轮回填

2.1 版本解析：options 门控

真正决定"当前生效的复制目标版本"的不是代码里写死的数值，而是由 find_replication_version() 函数计算出的 有效版本，实现见 backfill_outboxes.py#L80-L97：

def find_replication_version(
    model: type[ControlOutboxProducingModel] | type[CellOutboxProducingModel] | type[User],
    force_synchronous: bool = False,
) -> int:
    coded_version = model.replication_version
    if force_synchronous:
        return coded_version

    model_key = f"outbox_replication.{model._meta.db_table}.replication_version"
    return min(options.get(model_key), coded_version)

其语义非常关键——有效版本永远取 min(options 中配置的版本, 代码里的版本)：

• option 未设置或设置值低于代码版本：有效版本停留在旧值，回填不会推进到新版本（代码先上线但回填不触发）；
• option 设置值等于或高于代码版本：有效版本被代码版本封顶，回填得以推进到最新代码版本；
• force_synchronous=True（自托管场景）：完全绕过 options，直接使用 model.replication_version。

option 的键名格式为 outbox_replication.{模型db_table}.replication_version，例如 OrganizationMember 模型（db_table 为 sentry_organizationmember）对应：

"outbox_replication.sentry_organizationmember.replication_version"

这种"代码版本 + options 版本取较小值"的设计，是两阶段（先发布代码、后打开开关）灰度上线机制能够成立的根本原因。

2.2 游标跟踪：Redis 状态

Redis 按模型表记录回填进度，key 与值格式在 backfill_outboxes.py#L50-L51 定义：

def get_backfill_key(table_name: str) -> str:
    return f"outbox_backfill.{table_name}"

值为 JSON 编码的二元组 (lower_bound_id, current_version)。读写该状态的函数为 get_processing_state()（L54-L67）与 set_processing_state()（L70-L77）。注意 get_processing_state 在 key 不存在时会自动写入初始状态 (0, 1) 并返回，即首条游标记录是惰性创建的：

def get_processing_state(table_name: str) -> tuple[int, int]:
    ...
    if v is None:
        result = (0, 1)
        client.set(key, json.dumps(result))
    ...
    return result

写入游标的同时还会打点 backfill_outboxes.low_bound gauge，带 table_name 与 version 两个 tag。

2.3 版本比对与游标重置

批次划分函数 _chunk_processing_batch()（backfill_outboxes.py#L100-L127）会对 Redis 游标中的 version 与 options 解析出的 target_version 做三方比对：

• version > target_version：本轮回填已经完成（游标版本已越过目标），直接跳过，返回 None；
• version < target_version：检测到新版本，将 lower 重置为 0、version 提升到 target_version，从头开始整表回填；
• version == target_version：从上次离开的位置继续。

随后它通过 Min/Max 聚合把 lower 钳制在表内真实最小 ID 之上，并向后取 batch_size + 1 行计算本批上界 up，最终返回一个 BackfillBatch(low, up, version, has_more) 数据类（定义见 L32-L47，其中 has_more = upper > lower 表示是否还有更多行）。

3. Redis 游标状态机

综合源码实现（backfill_outboxes.py#L100-L181），游标状态可归纳为四种：

1. 初始：(0, 1) —— 尚未运行过任何回填（首次调用 get_processing_state 时创建）；
2. 进行中：(last_processed_id + 1, target_version) —— 正在逐批处理存量行，批与批之间游标单调前进；
3. 完成：(0, replication_version + 1) —— 所有行都已处理，版本越过目标，表示该版本的回填已收尾；
4. 检测到新版本：游标重置为 (0, new_target_version) —— 从头重新开始处理。

"完成状态版本号 = 代码版本 + 1"这个约定非常实用：它是判断回填是否完成、以及监控指标是否可读的黄金判据（详见第 7 节）。

4. 批次处理与执行细节

4.1 单批处理逻辑

每一批由 process_outbox_backfill_batch()（backfill_outboxes.py#L130-L181）执行，完整流程如下：

1. 调用 _chunk_processing_batch 确定本批的 ID 范围 (low, up)；
2. 对 model.objects.filter(id__gte=low, id__lte=up) 中的每个实例：
   • Cell 模型：在 outbox_context(flush=False) 内执行 inst.outbox_for_update().save()；
   • Control 模型 / User：遍历 inst.outboxes_for_update() 并逐个 save()（同样包在 outbox_context(flush=False) 内）。
3. 若 has_more 为假（本批已到表尾）：将游标置为 (0, model.replication_version + 1)，标记完成；
4. 否则将游标推进到 (up + 1, version)。

Control 侧还支持 按目标 cell 过滤回填：源码会尝试读取 option outbox_replication.{db_table}.backfill.target_cells（一个 cell 名称列表，如 ["us"]），若命中则只对 cell_name 落在列表内的 outbox 执行 save()，其余跳过。这在多 cell 环境（如同时存在 us、de 两个 cell）下，把控制面存量 token 等数据定向复制到指定 cell 时非常有用。

注意：process_outbox_backfill_batch 内部校验了模型类型——只有 CellOutboxProducingModel、ControlOutboxProducingModel 或 User 的子类才会被处理，其他模型直接返回 None（L133-L138）。

4.2 全模型调度与速率限制

回填的顶层入口是 backfill_outboxes_for()（backfill_outboxes.py#L187-L234），默认速率上限定义在 L184：

OUTBOX_BACKFILLS_PER_MINUTE = 10_000

函数的核心逻辑：

def backfill_outboxes_for(
    silo_mode: SiloMode,
    scheduled_count: int = 0,
    max_batch_rate: int = OUTBOX_BACKFILLS_PER_MINUTE,
    force_synchronous: bool = False,
) -> bool:
    # 用常规调度任务已安排的行数，从期望速率中扣除，以维持 outbox 处理的稳态
    remaining_to_backfill = max_batch_rate - scheduled_count
    ...
    if remaining_to_backfill > 0:
        for app, app_models in apps.all_models.items():
            for model in app_models.values():
                if not hasattr(model._meta, "silo_limit"):
                    continue
                # 只处理当前运行模式下本地的模型
                if silo_mode is not SiloMode.MONOLITH and silo_mode not in model._meta.silo_limit.modes:
                    continue
                batch = process_outbox_backfill_batch(
                    model, batch_size=remaining_to_backfill, force_synchronous=force_synchronous
                )
                if batch is None:
                    continue
                remaining_to_backfill -= batch.count
                backfilled += batch.count
                if remaining_to_backfill <= 0:
                    break
    metrics.incr("backfill_outboxes.backfilled", amount=backfilled, ...)
    return backfilled > 0

两个值得注意的工程细节：

• 速率自我抑制：remaining_to_backfill = max_batch_rate - scheduled_count。也就是说，如果这一轮调度周期内常规 outbox 任务已经安排了大量工作，回填会主动让路、削减本轮的批大小，从而在"补存量"与"处理新增量"之间维持平衡，避免回填挤占正常复制吞吐；
• Silo 本地性过滤：借助 model._meta.silo_limit.modes 判断模型是否属于当前运行模式（MONOLITH 模式下所有模型都可处理），确保 Cell 侧任务只回填 Cell 本地模型、Control 侧任务只回填 Control 本地模型。

函数以 backfilled > 0 作为返回值，调用方可以用 while 循环反复驱动直到返回 False（即无剩余可回填工作）。

4.3 与调度器的对接

回填并不是一个独立常驻的守护进程，而是挂在常规 outbox 调度任务 enqueue_outbox_jobs（对应 sentry.tasks.enqueue_outbox_jobs）的尾部。在 src/sentry/hybridcloud/tasks/deliver_from_outbox.py#L49-L57 中：

@instrumented_task(
    name="sentry.tasks.enqueue_outbox_jobs",
    namespace=hybridcloud_tasks,
    silo_mode=SiloMode.CELL,
    processing_deadline_duration=30,
)
def enqueue_outbox_jobs(
    concurrency: int | None = None, process_outbox_backfills: bool = True, **kwargs: Any
) -> None:
    schedule_batch(
        silo_mode=SiloMode.CELL,
        drain_task=drain_outbox_shards,
        concurrency=concurrency,
        process_outbox_backfills=process_outbox_backfills,
    )

而 deliver_from_outbox.py#L87-L88 的 schedule_batch 在调度完本轮 outbox 分片后，会把已调度数量传入回填函数：

if process_outbox_backfills:
    backfill_outboxes_for(silo_mode, scheduled_count)

process_outbox_backfills=False 可以临时关闭回填路径——测试中正是用它来隔离"普通 outbox 排水"与"回填"两类行为（参见 tests/sentry/hybridcloud/models/test_outbox.py）。

5. SaaS 与自托管两种上线路径

5.1 SaaS：通过 options 渐进灰度

SaaS 场景的 option 键格式：

f"outbox_replication.{model._meta.db_table}.replication_version"

# 以 OrganizationMember 为例：
"outbox_replication.sentry_organizationmember.replication_version"

标准灰度流程：

1. 合并带有上调 replication_version 的代码改动。此时由于 min(options 版本, 代码版本) 仍取旧值，回填还不会运行；
2. 在 Sentry options 系统中，把该模型对应的 option 设置到新版本值；
3. 此时 min(options 版本, 代码版本) 返回新版本，下一个 enqueue_outbox_jobs 周期开始回填；
4. 通过 Redis 游标状态与任务指标持续监控（见第 6、7 节）。

这种"先发代码、后开开关"的两步法带来两个关键收益：一是可以与数据库迁移、RPC 服务端上线等其他变更协调时序；二是回滚极快——只需把 option 调回旧值，min() 立即回到旧版本，未完成的新版本回填即告停止（已生成但未消费的 outbox 仍会被幂等地处理掉，符合 outbox handler 必须幂等的约束）。相关选项门控行为在测试 tests/sentry/hybridcloud/tasks/test_backfill_outboxes.py#L47-L69 的 test_processing_awaits_options 中得到验证：未覆盖 option 时 backfill_outboxes_for 返回 False（不执行回填），用 override_options 把 outbox_replication.sentry_authprovider.replication_version 顶到代码版本后，回填立即生效。

5.2 自托管：sentry upgrade 同步回填

自托管实例没有 options 灰度环境，回填改为在 sentry upgrade 时同步执行。信号处理函数 run_outbox_replications_for_self_hosted 定义在 src/sentry/hybridcloud/outbox/base.py#L436-L450，并通过 @receiver(post_upgrade) 挂到升级流程的 post_upgrade 信号上：

@receiver(post_upgrade)
def run_outbox_replications_for_self_hosted(*args: Any, **kwds: Any) -> None:
    from django.conf import settings
    from sentry.hybridcloud.models.outbox import OutboxBase
    from sentry.hybridcloud.tasks.backfill_outboxes import backfill_outboxes_for

    if not settings.SENTRY_SELF_HOSTED:
        return

    logger.info("Executing outbox replication backfill")
    while backfill_outboxes_for(
        SiloMode.get_current_mode(), max_batch_rate=1000, force_synchronous=True
    ):
        pass

该函数的执行要点：

1. 调用 backfill_outboxes_for(force_synchronous=True) —— 绕过 options，直接以 model.replication_version 为目标；
2. 每轮限速 1000 条（max_batch_rate=1000），用 while 循环反复执行，直至所有注册模型全部回填完毕；
3. 排空所有待处理的 outbox 分片，确保每次升级后实例都完全追平复制状态；
4. 仅当 settings.SENTRY_SELF_HOSTED 为真时才生效（SaaS 环境直接 return）。

对应的端到端测试见 test_backfill_outboxes.py#L222-L238：构造存量 Organization 与 AuthProvider 数据、清空 outbox 表后，在 override_settings(SENTRY_SELF_HOSTED=True) 下调用该函数，即可验证 OrganizationMapping、AuthProviderReplica 等对端副本被补齐。

6. 触发一次回填的最小改动

综合第 2 节，触发回填本质上是修改模型类上的版本标记：

class MyModel(ReplicatedCellModel):
    replication_version = 2  # 原来是 1；上调触发本轮回填

• 自托管：随版本发布，sentry upgrade 的 post_upgrade 钩子会自动同步回填；
• SaaS：代码合入后仍需把 outbox_replication.sentry_mymodel.replication_version 这个 option 设为目标值，回填才会在下个 enqueue_outbox_jobs 周期启动。

7. 监控一次回填

7.1 查看 Redis 游标状态

from sentry.hybridcloud.tasks.backfill_outboxes import get_processing_state

lower_bound, version = get_processing_state("sentry_mymodel")
# lower_bound > 0 表示回填仍在进行中
# version == model.replication_version + 1 表示回填已完成

测试代码中正是用这个判据断言回填完成：例如 test_control_processing_auth 在跑完 run_for_model(AuthIdentity) 后断言 get_processing_state(AuthIdentity._meta.db_table)[1] == AuthIdentity.replication_version + 1（test_backfill_outboxes.py#L118-L120）。

7.2 查看 option 门控值

from sentry import options

# 查看当前 option 把版本门控到了多少：
options.get("outbox_replication.sentry_mymodel.replication_version")

7.3 查看 outbox 队列深度

回填会向 outbox 表写入大量待处理行，可用 SQL 观察排水压力：

-- 特定 category 的 Cell outbox 数量
SELECT count(*) FROM sentry_regionoutbox
WHERE category = <category_value>;

-- 按分片深度排序，定位"热点分片"
SELECT shard_scope, shard_identifier, count(*) as depth
FROM sentry_regionoutbox
GROUP BY shard_scope, shard_identifier
ORDER BY depth DESC
LIMIT 10;

如果某个大组织（组织 ID 是回填分片键）恰好拥有海量存量行，回填可能在单个分片上瞬间堆积大量 outbox——这正对应 .agents/skills/hybrid-cloud-outboxes/references/category-and-scope.md 中描述的 hot shard 成因之一（"a backfill that generates thousands of outboxes for a single shard"），监控分片深度有助于提前发现这类瓶颈。

7.4 相关指标

回填与 outbox 流水线相关的指标（以实际打点位置为准，见 backfill_outboxes.py#L73-L77、L120-L125、L227-L233）：

指标	类型	含义
backfill_outboxes.low_bound	gauge	每个表当前游标位置（带 table_name/version tag）
backfill_outboxes.backfilled	counter	每个周期回填的行数（带 silo_mode/force_synchronous tag）
outbox.saved	counter	每次 outbox 落库计数
outbox.processed	counter	每个合并后 outbox 被处理计数
outbox.processing_lag	histogram	从 outbox 创建到被处理的时延分布

8. 版本提升与幂等性：来自测试的证据

回填系统的正确性由 tests/sentry/hybridcloud/tasks/test_backfill_outboxes.py 全面覆盖，几个代表性场景可直接作为理解运行语义的佐证：

• test_cell_processing：创建 5 个组织并清空 outbox 后，以 force_synchronous=True 循环驱动回填，断言重新生成 5 条 CellOutbox，排空后 Control 侧 OrganizationMapping 恰好为 5 条（L72-L86）；
• test_control_processing_auth：对 AuthIdentity、AuthProvider 按批大小 1 逐批处理，验证游标最终置为 replication_version + 1，并验证对端 replica 补齐；随后不提升版本再跑一次回填，不会产生任何新的 outbox——已完成的版本不会重复执行；而用 patch 把 AuthIdentity.replication_version 临时提到 10000 后，所有存量记录（两个 org 共 10 条）重新生成 outbox 并被复制，游标版本停在 10001（L89-L173）；
• test_control_processing_target_cells：在 us/de 双 cell 环境下，配置 {"outbox_replication.sentry_apitoken.backfill.target_cells": ["us"]}，回填产生的 2 条 ControlOutbox 全部落在 us cell，de cell 为 0（L176-L219）；
• test_run_outbox_replications_for_self_hosted：验证自托管 post_upgrade 路径（L222-L238）。

这些测试共同确认了三条安全语义：options 未开则回填不动、同版本只回填一次、新版本触发后从零全量重跑——三者组合起来，使回填可以放心地在生产环境反复编排。

9. 实战核对清单

在把存量模型迁入 outbox 复制体系（Step 5）并为其配置回填（Step 6）时，可对照 .agents/skills/hybrid-cloud-outboxes/SKILL.md 的检查表逐项确认，其中与回填强相关的条目包括：

• 模型已改为继承 ReplicatedCellModel 或 ReplicatedControlModel，且 category 正确注册到唯一 OutboxScope；
• 存量数据场景下已将 replication_version 上调并完成回填配置；
• 批量写操作必须走 CellOutboxProducingManager / ControlOutboxProducingManager（回填自身的批量路径同样经由这些抽象，见 outbox/base.py 中 manager 的 bulk_create/bulk_update/bulk_delete 实现），裸 queryset 会绕过 outbox 产生；
• handle_async_replication / handle_async_deletion 具备幂等性，因为回填生成的 outbox 与正常复制 outbox 一样可能被重复处理或合并；
• payload_for_update() 只放删除恢复所需的最小数据（payload 会因 coalescing 只保留最新一份）。

10. 总结

Sentry 的 outbox 回填系统（src/sentry/hybridcloud/tasks/backfill_outboxes.py）用两个轻量原语解决了一个典型的分布式系统难题：Sentry options 做版本门控（控制回填开关与灰度节奏），Redis 游标做断点续传（记录 (lower_bound_id, current_version) 进度）。代码先上线不触发任何复制行为，option 到位后回填在 enqueue_outbox_jobs 调度周期内按 10 000 条/分钟的速率上限增量推进；自托管实例则在 sentry upgrade 的 post_upgrade 钩子中同步、强制、全量地完成回填。理解 replication_version + 1 的完成约定与游标四种状态，配合第 7 节的监控手段，即可在 SaaS 灰度与自托管升级两种形态下安全地把存量数据纳入 outbox 复制体系。